mirror of https://github.com/grpc/grpc.git
[channel-stack] Make ordering explicit (#32852)
Ditch the old priority scheme for ordering filters, instead explicitly mark up before/after constraints. --------- Co-authored-by: ctiller <ctiller@users.noreply.github.com>pull/34714/head
Craig Tiller
1 year ago
committed by
GitHub
parent
4db2625512
commit
975184f04b
67 changed files with 1384 additions and 726 deletions
@ -0,0 +1,19 @@ |
||||
// Copyright 2023 gRPC authors.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include "src/core/lib/channel/channel_stack_trace.h" |
||||
|
||||
grpc_core::TraceFlag grpc_trace_channel_stack(false, "channel_stack"); |
||||
@ -0,0 +1,24 @@ |
||||
// Copyright 2023 gRPC authors.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
#ifndef GRPC_SRC_CORE_LIB_CHANNEL_CHANNEL_STACK_TRACE_H |
||||
#define GRPC_SRC_CORE_LIB_CHANNEL_CHANNEL_STACK_TRACE_H |
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include "src/core/lib/debug/trace.h" |
||||
|
||||
extern grpc_core::TraceFlag grpc_trace_channel_stack; |
||||
|
||||
#endif // GRPC_SRC_CORE_LIB_CHANNEL_CHANNEL_STACK_TRACE_H
|
||||
@ -0,0 +1,87 @@ |
||||
//
|
||||
//
|
||||
// Copyright 2016 gRPC authors.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
//
|
||||
//
|
||||
|
||||
#include "src/cpp/common/channel_filter.h" |
||||
|
||||
#include <type_traits> |
||||
|
||||
#include "absl/strings/str_cat.h" |
||||
#include "absl/strings/string_view.h" |
||||
|
||||
#include <grpc/support/log.h> |
||||
|
||||
#include "src/core/lib/channel/channel_args.h" |
||||
#include "src/core/lib/channel/channel_stack.h" |
||||
#include "src/core/lib/config/core_configuration.h" |
||||
#include "src/core/lib/slice/slice.h" |
||||
|
||||
namespace grpc { |
||||
|
||||
// MetadataBatch
|
||||
|
||||
void MetadataBatch::AddMetadata(const string& key, const string& value) { |
||||
batch_->Append(key, grpc_core::Slice::FromCopiedString(value), |
||||
[&](absl::string_view error, const grpc_core::Slice&) { |
||||
gpr_log(GPR_INFO, "%s", |
||||
absl::StrCat("MetadataBatch::AddMetadata error:", |
||||
error, " key=", key, " value=", value) |
||||
.c_str()); |
||||
}); |
||||
} |
||||
|
||||
// ChannelData
|
||||
|
||||
void ChannelData::StartTransportOp(grpc_channel_element* elem, |
||||
TransportOp* op) { |
||||
grpc_channel_next_op(elem, op->op()); |
||||
} |
||||
|
||||
void ChannelData::GetInfo(grpc_channel_element* elem, |
||||
const grpc_channel_info* channel_info) { |
||||
grpc_channel_next_get_info(elem, channel_info); |
||||
} |
||||
|
||||
// CallData
|
||||
|
||||
void CallData::StartTransportStreamOpBatch(grpc_call_element* elem, |
||||
TransportStreamOpBatch* op) { |
||||
grpc_call_next_op(elem, op->op()); |
||||
} |
||||
|
||||
void CallData::SetPollsetOrPollsetSet(grpc_call_element* elem, |
||||
grpc_polling_entity* pollent) { |
||||
grpc_call_stack_ignore_set_pollset_or_pollset_set(elem, pollent); |
||||
} |
||||
|
||||
namespace internal { |
||||
|
||||
void RegisterChannelFilter( |
||||
grpc_channel_stack_type stack_type, int, |
||||
std::function<bool(const grpc_core::ChannelArgs&)> include_filter, |
||||
const grpc_channel_filter* filter) { |
||||
grpc_core::CoreConfiguration::RegisterBuilder( |
||||
[stack_type, filter, include_filter = std::move(include_filter)]( |
||||
grpc_core::CoreConfiguration::Builder* builder) { |
||||
auto& f = builder->channel_init()->RegisterFilter(stack_type, filter); |
||||
if (include_filter) f.If(include_filter); |
||||
}); |
||||
} |
||||
|
||||
} // namespace internal
|
||||
|
||||
} // namespace grpc
|
||||
@ -0,0 +1,212 @@ |
||||
// Copyright 2023 gRPC authors.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
#include "src/core/lib/surface/channel_init.h" |
||||
|
||||
#include <map> |
||||
#include <string> |
||||
|
||||
#include "absl/strings/string_view.h" |
||||
#include "gtest/gtest.h" |
||||
|
||||
#include "src/core/lib/channel/channel_stack.h" |
||||
#include "src/core/lib/channel/channel_stack_builder_impl.h" |
||||
#include "src/core/lib/surface/channel_stack_type.h" |
||||
#include "test/core/util/test_config.h" |
||||
|
||||
namespace grpc_core { |
||||
namespace { |
||||
|
||||
const grpc_channel_filter* FilterNamed(const char* name) { |
||||
static auto* filters = |
||||
new std::map<absl::string_view, const grpc_channel_filter*>; |
||||
auto it = filters->find(name); |
||||
if (it != filters->end()) return it->second; |
||||
return filters |
||||
->emplace(name, |
||||
new grpc_channel_filter{nullptr, nullptr, nullptr, 0, nullptr, |
||||
nullptr, nullptr, 0, nullptr, nullptr, |
||||
nullptr, nullptr, name}) |
||||
.first->second; |
||||
} |
||||
|
||||
std::vector<std::string> GetFilterNames(const ChannelInit& init, |
||||
grpc_channel_stack_type type, |
||||
const ChannelArgs& args) { |
||||
ChannelStackBuilderImpl b("test", type, args); |
||||
if (!init.CreateStack(&b)) return {}; |
||||
std::vector<std::string> names; |
||||
for (auto f : b.stack()) { |
||||
names.push_back(f->name); |
||||
} |
||||
EXPECT_NE(names, std::vector<std::string>()); |
||||
return names; |
||||
} |
||||
|
||||
TEST(ChannelInitTest, Empty) { |
||||
ChannelInit::Builder b; |
||||
b.RegisterFilter(GRPC_CLIENT_CHANNEL, FilterNamed("terminator")).Terminal(); |
||||
auto init = b.Build(); |
||||
EXPECT_EQ(GetFilterNames(init, GRPC_CLIENT_CHANNEL, ChannelArgs()), |
||||
std::vector<std::string>({"terminator"})); |
||||
} |
||||
|
||||
TEST(ChannelInitTest, OneClientFilter) { |
||||
ChannelInit::Builder b; |
||||
b.RegisterFilter(GRPC_CLIENT_CHANNEL, FilterNamed("foo")); |
||||
b.RegisterFilter(GRPC_CLIENT_CHANNEL, FilterNamed("terminator")).Terminal(); |
||||
b.RegisterFilter(GRPC_SERVER_CHANNEL, FilterNamed("terminator")).Terminal(); |
||||
auto init = b.Build(); |
||||
EXPECT_EQ(GetFilterNames(init, GRPC_CLIENT_CHANNEL, ChannelArgs()), |
||||
std::vector<std::string>({"foo", "terminator"})); |
||||
EXPECT_EQ(GetFilterNames(init, GRPC_SERVER_CHANNEL, ChannelArgs()), |
||||
std::vector<std::string>({"terminator"})); |
||||
} |
||||
|
||||
TEST(ChannelInitTest, DefaultLexicalOrdering) { |
||||
// ChannelInit defaults to lexical ordering in the absense of other
|
||||
// constraints, to ensure that a stable ordering is produced between builds.
|
||||
ChannelInit::Builder b; |
||||
b.RegisterFilter(GRPC_CLIENT_CHANNEL, FilterNamed("foo")); |
||||
b.RegisterFilter(GRPC_CLIENT_CHANNEL, FilterNamed("bar")); |
||||
b.RegisterFilter(GRPC_CLIENT_CHANNEL, FilterNamed("baz")); |
||||
b.RegisterFilter(GRPC_CLIENT_CHANNEL, FilterNamed("aaa")).Terminal(); |
||||
auto init = b.Build(); |
||||
EXPECT_EQ(GetFilterNames(init, GRPC_CLIENT_CHANNEL, ChannelArgs()), |
||||
std::vector<std::string>({"bar", "baz", "foo", "aaa"})); |
||||
} |
||||
|
||||
TEST(ChannelInitTest, AfterConstraintsApply) { |
||||
ChannelInit::Builder b; |
||||
b.RegisterFilter(GRPC_CLIENT_CHANNEL, FilterNamed("foo")); |
||||
b.RegisterFilter(GRPC_CLIENT_CHANNEL, FilterNamed("bar")) |
||||
.After({FilterNamed("foo")}); |
||||
b.RegisterFilter(GRPC_CLIENT_CHANNEL, FilterNamed("baz")); |
||||
b.RegisterFilter(GRPC_CLIENT_CHANNEL, FilterNamed("aaa")).Terminal(); |
||||
auto init = b.Build(); |
||||
EXPECT_EQ(GetFilterNames(init, GRPC_CLIENT_CHANNEL, ChannelArgs()), |
||||
std::vector<std::string>({"baz", "foo", "bar", "aaa"})); |
||||
} |
||||
|
||||
TEST(ChannelInitTest, BeforeConstraintsApply) { |
||||
ChannelInit::Builder b; |
||||
b.RegisterFilter(GRPC_CLIENT_CHANNEL, FilterNamed("foo")) |
||||
.Before({FilterNamed("bar")}); |
||||
b.RegisterFilter(GRPC_CLIENT_CHANNEL, FilterNamed("bar")); |
||||
b.RegisterFilter(GRPC_CLIENT_CHANNEL, FilterNamed("baz")); |
||||
b.RegisterFilter(GRPC_CLIENT_CHANNEL, FilterNamed("aaa")).Terminal(); |
||||
auto init = b.Build(); |
||||
EXPECT_EQ(GetFilterNames(init, GRPC_CLIENT_CHANNEL, ChannelArgs()), |
||||
std::vector<std::string>({"baz", "foo", "bar", "aaa"})); |
||||
} |
||||
|
||||
TEST(ChannelInitTest, PredicatesCanFilter) { |
||||
ChannelInit::Builder b; |
||||
b.RegisterFilter(GRPC_CLIENT_CHANNEL, FilterNamed("foo")) |
||||
.IfChannelArg("foo", true); |
||||
b.RegisterFilter(GRPC_CLIENT_CHANNEL, FilterNamed("bar")) |
||||
.IfChannelArg("bar", false); |
||||
b.RegisterFilter(GRPC_CLIENT_CHANNEL, FilterNamed("aaa")).Terminal(); |
||||
auto init = b.Build(); |
||||
EXPECT_EQ(GetFilterNames(init, GRPC_CLIENT_CHANNEL, ChannelArgs()), |
||||
std::vector<std::string>({"foo", "aaa"})); |
||||
EXPECT_EQ(GetFilterNames(init, GRPC_CLIENT_CHANNEL, |
||||
ChannelArgs().Set("foo", false)), |
||||
std::vector<std::string>({"aaa"})); |
||||
EXPECT_EQ( |
||||
GetFilterNames(init, GRPC_CLIENT_CHANNEL, ChannelArgs().Set("bar", true)), |
||||
std::vector<std::string>({"bar", "foo", "aaa"})); |
||||
EXPECT_EQ(GetFilterNames(init, GRPC_CLIENT_CHANNEL, |
||||
ChannelArgs().Set("bar", true).Set("foo", false)), |
||||
std::vector<std::string>({"bar", "aaa"})); |
||||
} |
||||
|
||||
TEST(ChannelInitTest, CanAddTerminalFilter) { |
||||
ChannelInit::Builder b; |
||||
b.RegisterFilter(GRPC_CLIENT_CHANNEL, FilterNamed("foo")); |
||||
b.RegisterFilter(GRPC_CLIENT_CHANNEL, FilterNamed("bar")).Terminal(); |
||||
auto init = b.Build(); |
||||
EXPECT_EQ(GetFilterNames(init, GRPC_CLIENT_CHANNEL, ChannelArgs()), |
||||
std::vector<std::string>({"foo", "bar"})); |
||||
} |
||||
|
||||
TEST(ChannelInitTest, CanAddMultipleTerminalFilters) { |
||||
ChannelInit::Builder b; |
||||
b.RegisterFilter(GRPC_CLIENT_CHANNEL, FilterNamed("foo")); |
||||
b.RegisterFilter(GRPC_CLIENT_CHANNEL, FilterNamed("bar")) |
||||
.Terminal() |
||||
.IfChannelArg("bar", false); |
||||
b.RegisterFilter(GRPC_CLIENT_CHANNEL, FilterNamed("baz")) |
||||
.Terminal() |
||||
.IfChannelArg("baz", false); |
||||
auto init = b.Build(); |
||||
EXPECT_EQ(GetFilterNames(init, GRPC_CLIENT_CHANNEL, ChannelArgs()), |
||||
std::vector<std::string>()); |
||||
EXPECT_EQ( |
||||
GetFilterNames(init, GRPC_CLIENT_CHANNEL, ChannelArgs().Set("bar", true)), |
||||
std::vector<std::string>({"foo", "bar"})); |
||||
EXPECT_EQ( |
||||
GetFilterNames(init, GRPC_CLIENT_CHANNEL, ChannelArgs().Set("baz", true)), |
||||
std::vector<std::string>({"foo", "baz"})); |
||||
EXPECT_EQ(GetFilterNames(init, GRPC_CLIENT_CHANNEL, |
||||
ChannelArgs().Set("bar", true).Set("baz", true)), |
||||
std::vector<std::string>()); |
||||
} |
||||
|
||||
TEST(ChannelInitTest, CanAddBeforeAllOnce) { |
||||
ChannelInit::Builder b; |
||||
b.RegisterFilter(GRPC_CLIENT_CHANNEL, FilterNamed("foo")).BeforeAll(); |
||||
b.RegisterFilter(GRPC_CLIENT_CHANNEL, FilterNamed("bar")); |
||||
b.RegisterFilter(GRPC_CLIENT_CHANNEL, FilterNamed("baz")); |
||||
b.RegisterFilter(GRPC_CLIENT_CHANNEL, FilterNamed("aaa")).Terminal(); |
||||
EXPECT_EQ(GetFilterNames(b.Build(), GRPC_CLIENT_CHANNEL, ChannelArgs()), |
||||
std::vector<std::string>({"foo", "bar", "baz", "aaa"})); |
||||
} |
||||
|
||||
TEST(ChannelInitTest, CanAddBeforeAllTwice) { |
||||
ChannelInit::Builder b; |
||||
b.RegisterFilter(GRPC_CLIENT_CHANNEL, FilterNamed("foo")).BeforeAll(); |
||||
b.RegisterFilter(GRPC_CLIENT_CHANNEL, FilterNamed("bar")).BeforeAll(); |
||||
b.RegisterFilter(GRPC_CLIENT_CHANNEL, FilterNamed("baz")); |
||||
b.RegisterFilter(GRPC_CLIENT_CHANNEL, FilterNamed("aaa")).Terminal(); |
||||
EXPECT_DEATH_IF_SUPPORTED(b.Build(), "Unresolvable graph of channel filters"); |
||||
} |
||||
|
||||
TEST(ChannelInitTest, CanPostProcessFilters) { |
||||
ChannelInit::Builder b; |
||||
b.RegisterFilter(GRPC_CLIENT_CHANNEL, FilterNamed("foo")); |
||||
b.RegisterFilter(GRPC_CLIENT_CHANNEL, FilterNamed("aaa")).Terminal(); |
||||
int called_post_processor = 0; |
||||
b.RegisterPostProcessor( |
||||
GRPC_CLIENT_CHANNEL, |
||||
ChannelInit::PostProcessorSlot::kXdsChannelStackModifier, |
||||
[&called_post_processor](ChannelStackBuilder& b) { |
||||
++called_post_processor; |
||||
b.mutable_stack()->push_back(FilterNamed("bar")); |
||||
}); |
||||
auto init = b.Build(); |
||||
EXPECT_EQ(called_post_processor, 0); |
||||
EXPECT_EQ(GetFilterNames(init, GRPC_CLIENT_CHANNEL, ChannelArgs()), |
||||
std::vector<std::string>({"foo", "aaa", "bar"})); |
||||
} |
||||
|
||||
} // namespace
|
||||
} // namespace grpc_core
|
||||
|
||||
int main(int argc, char** argv) { |
||||
grpc::testing::TestEnvironment env(&argc, argv); |
||||
::testing::InitGoogleTest(&argc, argv); |
||||
grpc::testing::TestGrpcScope grpc_scope; |
||||
return RUN_ALL_TESTS(); |
||||
} |
||||
Loading…
Reference in new issue