From 650ae16bf7911bad69707c27ae3fd88e375771e1 Mon Sep 17 00:00:00 2001 From: Egor Suvorov Date: Fri, 26 Aug 2016 16:26:23 -0400 Subject: [PATCH 1/7] Add IWYU export pragmas to grpc++/grpc++.h. IWYU is a C++ tool which ensures that end users explicitly #include every header they need. Without pragmas IWYU have no way of distinguishing between 'implementation-specific' headers and headers which are shortcuts for groups of other headers, like grpc++.h. --- include/grpc++/grpc++.h | 2 ++ 1 file changed, 2 insertions(+) diff --git a/include/grpc++/grpc++.h b/include/grpc++/grpc++.h index ef07e199759..e4b0478d770 100644 --- a/include/grpc++/grpc++.h +++ b/include/grpc++/grpc++.h @@ -51,6 +51,7 @@ #ifndef GRPCXX_GRPCXX_H #define GRPCXX_GRPCXX_H +// IWYU pragma: begin_exports #include #include @@ -62,5 +63,6 @@ #include #include #include +// IWYU pragma: end_exports #endif // GRPCXX_GRPCXX_H From 93901a7a8696d0496fa336e0b74ed821da9e340e Mon Sep 17 00:00:00 2001 From: Egor Suvorov Date: Fri, 26 Aug 2016 16:48:38 -0400 Subject: [PATCH 2/7] Add clarifying comment about IWYU in grpc++.h --- include/grpc++/grpc++.h | 2 ++ 1 file changed, 2 insertions(+) diff --git a/include/grpc++/grpc++.h b/include/grpc++/grpc++.h index e4b0478d770..afb1c555bb6 100644 --- a/include/grpc++/grpc++.h +++ b/include/grpc++/grpc++.h @@ -51,6 +51,8 @@ #ifndef GRPCXX_GRPCXX_H #define GRPCXX_GRPCXX_H +// Pragma for http://include-what-you-use.org/ tool, tells that following +// headers are not private for grpc++.h and are part of its interface. // IWYU pragma: begin_exports #include From 7176a828c5418cfb1084c7b776be65a0ea7b138e Mon Sep 17 00:00:00 2001 From: Yuchen Zeng Date: Mon, 29 Aug 2016 17:04:50 -0700 Subject: [PATCH 3/7] Use --protofiles --- test/cpp/util/grpc_tool.cc | 22 +++++++++++----------- test/cpp/util/grpc_tool.h | 2 +- 2 files changed, 12 insertions(+), 12 deletions(-) diff --git a/test/cpp/util/grpc_tool.cc b/test/cpp/util/grpc_tool.cc index f06053ca23e..4af00cdc966 100644 --- a/test/cpp/util/grpc_tool.cc +++ b/test/cpp/util/grpc_tool.cc @@ -57,7 +57,7 @@ DEFINE_bool(remotedb, true, "Use server types to parse and format messages"); DEFINE_string(metadata, "", "Metadata to send to server, in the form of key1:val1:key2:val2"); DEFINE_string(proto_path, ".", "Path to look for the proto file."); -DEFINE_string(proto_file, "", "Name of the proto file."); +DEFINE_string(protofiles, "", "Name of the proto file."); DEFINE_bool(binary_input, false, "Input in binary format"); DEFINE_bool(binary_output, false, "Output in binary format"); DEFINE_string(infile, "", "Input file (default is stdin)"); @@ -71,9 +71,9 @@ class GrpcTool { explicit GrpcTool(); virtual ~GrpcTool() {} - bool Help(int argc, const char** argv, CliCredentials cred, + bool Help(int argc, const char** argv, const CliCredentials& cred, GrpcToolOutputCallback callback); - bool CallMethod(int argc, const char** argv, CliCredentials cred, + bool CallMethod(int argc, const char** argv, const CliCredentials& cred, GrpcToolOutputCallback callback); // TODO(zyc): implement the following methods // bool ListServices(int argc, const char** argv, GrpcToolOutputCallback @@ -101,7 +101,7 @@ class GrpcTool { }; template -std::function BindWith5Args(T&& func) { return std::bind(std::forward(func), std::placeholders::_1, @@ -156,7 +156,7 @@ void PrintMetadata(const T& m, const grpc::string& message) { struct Command { const char* command; - std::function function; int min_args; @@ -201,7 +201,7 @@ const Command* FindCommand(const grpc::string& name) { } } // namespace -int GrpcToolMainLib(int argc, const char** argv, const CliCredentials cred, +int GrpcToolMainLib(int argc, const char** argv, const CliCredentials& cred, GrpcToolOutputCallback callback) { if (argc < 2) { Usage("No command specified"); @@ -238,7 +238,7 @@ void GrpcTool::CommandUsage(const grpc::string& usage) const { } } -bool GrpcTool::Help(int argc, const char** argv, const CliCredentials cred, +bool GrpcTool::Help(int argc, const char** argv, const CliCredentials& cred, GrpcToolOutputCallback callback) { CommandUsage( "Print help\n" @@ -258,7 +258,7 @@ bool GrpcTool::Help(int argc, const char** argv, const CliCredentials cred, } bool GrpcTool::CallMethod(int argc, const char** argv, - const CliCredentials cred, + const CliCredentials& cred, GrpcToolOutputCallback callback) { CommandUsage( "Call method\n" @@ -267,10 +267,10 @@ bool GrpcTool::CallMethod(int argc, const char** argv, " ; Exported service name\n" " ; Method name\n" " ; Text protobuffer (overrides infile)\n" - " --proto_file ; Comma separated proto files used as a" + " --protofiles ; Comma separated proto files used as a" " fallback when parsing request/response\n" " --proto_path ; The search path of proto files, valid" - " only when --proto_file is given\n" + " only when --protofiles is given\n" " --metadata ; The metadata to be sent to the server\n" " --infile ; Input filename (defaults to stdin)\n" " --outfile ; Output filename (defaults to stdout)\n" @@ -310,7 +310,7 @@ bool GrpcTool::CallMethod(int argc, const char** argv, if (!FLAGS_binary_input || !FLAGS_binary_output) { parser.reset( new grpc::testing::ProtoFileParser(FLAGS_remotedb ? channel : nullptr, - FLAGS_proto_path, FLAGS_proto_file)); + FLAGS_proto_path, FLAGS_protofiles)); if (parser->HasError()) { return false; } diff --git a/test/cpp/util/grpc_tool.h b/test/cpp/util/grpc_tool.h index 3da86937c2d..df3680258b8 100644 --- a/test/cpp/util/grpc_tool.h +++ b/test/cpp/util/grpc_tool.h @@ -45,7 +45,7 @@ namespace testing { typedef std::function GrpcToolOutputCallback; -int GrpcToolMainLib(int argc, const char **argv, CliCredentials cred, +int GrpcToolMainLib(int argc, const char **argv, const CliCredentials &cred, GrpcToolOutputCallback callback); } // namespace testing From 11d3c8f3d4463a63facba3d53f7e6e7f9008aad0 Mon Sep 17 00:00:00 2001 From: Yuchen Zeng Date: Mon, 29 Aug 2016 18:18:41 -0700 Subject: [PATCH 4/7] Add ServerBuilderPlugin::UpdateChannelArguments --- include/grpc++/impl/server_builder_plugin.h | 5 +++++ src/cpp/server/server_builder.cc | 12 +++++------- 2 files changed, 10 insertions(+), 7 deletions(-) diff --git a/include/grpc++/impl/server_builder_plugin.h b/include/grpc++/impl/server_builder_plugin.h index 1e157efa111..61632e32fa4 100644 --- a/include/grpc++/impl/server_builder_plugin.h +++ b/include/grpc++/impl/server_builder_plugin.h @@ -41,6 +41,7 @@ namespace grpc { class ServerInitializer; +class ChannelArguments; class ServerBuilderPlugin { public: @@ -58,6 +59,10 @@ class ServerBuilderPlugin { // ServerBuilderOption::UpdatePlugins virtual void ChangeArguments(const grpc::string& name, void* value) = 0; + // UpdateChannelArguments will be called in ServerBuilder::BuildAndStart(), + // before the Server instance is created. + virtual void UpdateChannelArguments(ChannelArguments* args) {} + virtual bool has_sync_methods() const { return false; } virtual bool has_async_methods() const { return false; } }; diff --git a/src/cpp/server/server_builder.cc b/src/cpp/server/server_builder.cc index 45bb858e2ed..669af9da783 100644 --- a/src/cpp/server/server_builder.cc +++ b/src/cpp/server/server_builder.cc @@ -154,14 +154,12 @@ std::unique_ptr ServerBuilder::BuildAndStart() { (*option)->UpdateArguments(&args); (*option)->UpdatePlugins(&plugins_); } - if (!thread_pool) { - for (auto plugin = plugins_.begin(); plugin != plugins_.end(); plugin++) { - if ((*plugin)->has_sync_methods()) { - thread_pool.reset(CreateDefaultThreadPool()); - has_sync_methods = true; - break; - } + for (auto plugin = plugins_.begin(); plugin != plugins_.end(); plugin++) { + if (!thread_pool && (*plugin)->has_sync_methods()) { + thread_pool.reset(CreateDefaultThreadPool()); + has_sync_methods = true; } + (*plugin)->UpdateChannelArguments(&args); } if (max_message_size_ > 0) { args.SetInt(GRPC_ARG_MAX_MESSAGE_LENGTH, max_message_size_); From 4c0fe49f76e0e3d7e7ceed5722cc43010af61ab3 Mon Sep 17 00:00:00 2001 From: "Mark D. Roth" Date: Wed, 31 Aug 2016 13:51:55 -0700 Subject: [PATCH 5/7] Move subchannel_call_holder code into client_channel module. --- BUILD | 8 - CMakeLists.txt | 3 - Makefile | 3 - binding.gyp | 1 - build.yaml | 2 - config.m4 | 1 - gRPC-Core.podspec | 3 - grpc.gemspec | 2 - package.xml | 2 - src/core/ext/client_config/client_channel.c | 322 ++++++++++++++++-- .../client_config/subchannel_call_holder.c | 292 ---------------- .../client_config/subchannel_call_holder.h | 99 ------ src/python/grpcio/grpc_core_dependencies.py | 1 - tools/doxygen/Doxyfile.core.internal | 2 - tools/run_tests/sources_and_headers.json | 3 - vsprojects/vcxproj/grpc/grpc.vcxproj | 3 - vsprojects/vcxproj/grpc/grpc.vcxproj.filters | 6 - .../grpc_unsecure/grpc_unsecure.vcxproj | 3 - .../grpc_unsecure.vcxproj.filters | 6 - 19 files changed, 290 insertions(+), 472 deletions(-) delete mode 100644 src/core/ext/client_config/subchannel_call_holder.c delete mode 100644 src/core/ext/client_config/subchannel_call_holder.h diff --git a/BUILD b/BUILD index 7dc5b6ca398..927b6e532a5 100644 --- a/BUILD +++ b/BUILD @@ -299,7 +299,6 @@ cc_library( "src/core/ext/client_config/resolver_registry.h", "src/core/ext/client_config/resolver_result.h", "src/core/ext/client_config/subchannel.h", - "src/core/ext/client_config/subchannel_call_holder.h", "src/core/ext/client_config/subchannel_index.h", "src/core/ext/client_config/uri_parser.h", "src/core/ext/lb_policy/grpclb/grpclb.h", @@ -474,7 +473,6 @@ cc_library( "src/core/ext/client_config/resolver_registry.c", "src/core/ext/client_config/resolver_result.c", "src/core/ext/client_config/subchannel.c", - "src/core/ext/client_config/subchannel_call_holder.c", "src/core/ext/client_config/subchannel_index.c", "src/core/ext/client_config/uri_parser.c", "src/core/ext/transport/chttp2/server/insecure/server_chttp2.c", @@ -671,7 +669,6 @@ cc_library( "src/core/ext/client_config/resolver_registry.h", "src/core/ext/client_config/resolver_result.h", "src/core/ext/client_config/subchannel.h", - "src/core/ext/client_config/subchannel_call_holder.h", "src/core/ext/client_config/subchannel_index.h", "src/core/ext/client_config/uri_parser.h", "src/core/lib/security/context/security_context.h", @@ -830,7 +827,6 @@ cc_library( "src/core/ext/client_config/resolver_registry.c", "src/core/ext/client_config/resolver_result.c", "src/core/ext/client_config/subchannel.c", - "src/core/ext/client_config/subchannel_call_holder.c", "src/core/ext/client_config/subchannel_index.c", "src/core/ext/client_config/uri_parser.c", "src/core/lib/http/httpcli_security_connector.c", @@ -1024,7 +1020,6 @@ cc_library( "src/core/ext/client_config/resolver_registry.h", "src/core/ext/client_config/resolver_result.h", "src/core/ext/client_config/subchannel.h", - "src/core/ext/client_config/subchannel_call_holder.h", "src/core/ext/client_config/subchannel_index.h", "src/core/ext/client_config/uri_parser.h", "src/core/ext/load_reporting/load_reporting.h", @@ -1174,7 +1169,6 @@ cc_library( "src/core/ext/client_config/resolver_registry.c", "src/core/ext/client_config/resolver_result.c", "src/core/ext/client_config/subchannel.c", - "src/core/ext/client_config/subchannel_call_holder.c", "src/core/ext/client_config/subchannel_index.c", "src/core/ext/client_config/uri_parser.c", "src/core/ext/resolver/dns/native/dns_resolver.c", @@ -2320,7 +2314,6 @@ objc_library( "src/core/ext/client_config/resolver_registry.c", "src/core/ext/client_config/resolver_result.c", "src/core/ext/client_config/subchannel.c", - "src/core/ext/client_config/subchannel_call_holder.c", "src/core/ext/client_config/subchannel_index.c", "src/core/ext/client_config/uri_parser.c", "src/core/ext/transport/chttp2/server/insecure/server_chttp2.c", @@ -2519,7 +2512,6 @@ objc_library( "src/core/ext/client_config/resolver_registry.h", "src/core/ext/client_config/resolver_result.h", "src/core/ext/client_config/subchannel.h", - "src/core/ext/client_config/subchannel_call_holder.h", "src/core/ext/client_config/subchannel_index.h", "src/core/ext/client_config/uri_parser.h", "src/core/ext/lb_policy/grpclb/grpclb.h", diff --git a/CMakeLists.txt b/CMakeLists.txt index b8194652519..7b4ec92c40d 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -448,7 +448,6 @@ add_library(grpc src/core/ext/client_config/resolver_registry.c src/core/ext/client_config/resolver_result.c src/core/ext/client_config/subchannel.c - src/core/ext/client_config/subchannel_call_holder.c src/core/ext/client_config/subchannel_index.c src/core/ext/client_config/uri_parser.c src/core/ext/transport/chttp2/server/insecure/server_chttp2.c @@ -680,7 +679,6 @@ add_library(grpc_cronet src/core/ext/client_config/resolver_registry.c src/core/ext/client_config/resolver_result.c src/core/ext/client_config/subchannel.c - src/core/ext/client_config/subchannel_call_holder.c src/core/ext/client_config/subchannel_index.c src/core/ext/client_config/uri_parser.c src/core/lib/http/httpcli_security_connector.c @@ -912,7 +910,6 @@ add_library(grpc_unsecure src/core/ext/client_config/resolver_registry.c src/core/ext/client_config/resolver_result.c src/core/ext/client_config/subchannel.c - src/core/ext/client_config/subchannel_call_holder.c src/core/ext/client_config/subchannel_index.c src/core/ext/client_config/uri_parser.c src/core/ext/resolver/dns/native/dns_resolver.c diff --git a/Makefile b/Makefile index 4bd4c306c17..0f8cfdf294d 100644 --- a/Makefile +++ b/Makefile @@ -2664,7 +2664,6 @@ LIBGRPC_SRC = \ src/core/ext/client_config/resolver_registry.c \ src/core/ext/client_config/resolver_result.c \ src/core/ext/client_config/subchannel.c \ - src/core/ext/client_config/subchannel_call_holder.c \ src/core/ext/client_config/subchannel_index.c \ src/core/ext/client_config/uri_parser.c \ src/core/ext/transport/chttp2/server/insecure/server_chttp2.c \ @@ -2914,7 +2913,6 @@ LIBGRPC_CRONET_SRC = \ src/core/ext/client_config/resolver_registry.c \ src/core/ext/client_config/resolver_result.c \ src/core/ext/client_config/subchannel.c \ - src/core/ext/client_config/subchannel_call_holder.c \ src/core/ext/client_config/subchannel_index.c \ src/core/ext/client_config/uri_parser.c \ src/core/lib/http/httpcli_security_connector.c \ @@ -3373,7 +3371,6 @@ LIBGRPC_UNSECURE_SRC = \ src/core/ext/client_config/resolver_registry.c \ src/core/ext/client_config/resolver_result.c \ src/core/ext/client_config/subchannel.c \ - src/core/ext/client_config/subchannel_call_holder.c \ src/core/ext/client_config/subchannel_index.c \ src/core/ext/client_config/uri_parser.c \ src/core/ext/resolver/dns/native/dns_resolver.c \ diff --git a/binding.gyp b/binding.gyp index a29cfda6fc5..76276475df8 100644 --- a/binding.gyp +++ b/binding.gyp @@ -720,7 +720,6 @@ 'src/core/ext/client_config/resolver_registry.c', 'src/core/ext/client_config/resolver_result.c', 'src/core/ext/client_config/subchannel.c', - 'src/core/ext/client_config/subchannel_call_holder.c', 'src/core/ext/client_config/subchannel_index.c', 'src/core/ext/client_config/uri_parser.c', 'src/core/ext/transport/chttp2/server/insecure/server_chttp2.c', diff --git a/build.yaml b/build.yaml index 9b182d2ffcf..1f47a41858a 100644 --- a/build.yaml +++ b/build.yaml @@ -350,7 +350,6 @@ filegroups: - src/core/ext/client_config/resolver_registry.h - src/core/ext/client_config/resolver_result.h - src/core/ext/client_config/subchannel.h - - src/core/ext/client_config/subchannel_call_holder.h - src/core/ext/client_config/subchannel_index.h - src/core/ext/client_config/uri_parser.h src: @@ -370,7 +369,6 @@ filegroups: - src/core/ext/client_config/resolver_registry.c - src/core/ext/client_config/resolver_result.c - src/core/ext/client_config/subchannel.c - - src/core/ext/client_config/subchannel_call_holder.c - src/core/ext/client_config/subchannel_index.c - src/core/ext/client_config/uri_parser.c plugin: grpc_client_config diff --git a/config.m4 b/config.m4 index 5b47074bcd7..038af4569a8 100644 --- a/config.m4 +++ b/config.m4 @@ -239,7 +239,6 @@ if test "$PHP_GRPC" != "no"; then src/core/ext/client_config/resolver_registry.c \ src/core/ext/client_config/resolver_result.c \ src/core/ext/client_config/subchannel.c \ - src/core/ext/client_config/subchannel_call_holder.c \ src/core/ext/client_config/subchannel_index.c \ src/core/ext/client_config/uri_parser.c \ src/core/ext/transport/chttp2/server/insecure/server_chttp2.c \ diff --git a/gRPC-Core.podspec b/gRPC-Core.podspec index 41ba1b1058b..1e58d328018 100644 --- a/gRPC-Core.podspec +++ b/gRPC-Core.podspec @@ -390,7 +390,6 @@ Pod::Spec.new do |s| 'src/core/ext/client_config/resolver_registry.h', 'src/core/ext/client_config/resolver_result.h', 'src/core/ext/client_config/subchannel.h', - 'src/core/ext/client_config/subchannel_call_holder.h', 'src/core/ext/client_config/subchannel_index.h', 'src/core/ext/client_config/uri_parser.h', 'src/core/ext/lb_policy/grpclb/grpclb.h', @@ -569,7 +568,6 @@ Pod::Spec.new do |s| 'src/core/ext/client_config/resolver_registry.c', 'src/core/ext/client_config/resolver_result.c', 'src/core/ext/client_config/subchannel.c', - 'src/core/ext/client_config/subchannel_call_holder.c', 'src/core/ext/client_config/subchannel_index.c', 'src/core/ext/client_config/uri_parser.c', 'src/core/ext/transport/chttp2/server/insecure/server_chttp2.c', @@ -751,7 +749,6 @@ Pod::Spec.new do |s| 'src/core/ext/client_config/resolver_registry.h', 'src/core/ext/client_config/resolver_result.h', 'src/core/ext/client_config/subchannel.h', - 'src/core/ext/client_config/subchannel_call_holder.h', 'src/core/ext/client_config/subchannel_index.h', 'src/core/ext/client_config/uri_parser.h', 'src/core/ext/lb_policy/grpclb/grpclb.h', diff --git a/grpc.gemspec b/grpc.gemspec index 95eb28bd1d7..31a99e9c126 100755 --- a/grpc.gemspec +++ b/grpc.gemspec @@ -309,7 +309,6 @@ Gem::Specification.new do |s| s.files += %w( src/core/ext/client_config/resolver_registry.h ) s.files += %w( src/core/ext/client_config/resolver_result.h ) s.files += %w( src/core/ext/client_config/subchannel.h ) - s.files += %w( src/core/ext/client_config/subchannel_call_holder.h ) s.files += %w( src/core/ext/client_config/subchannel_index.h ) s.files += %w( src/core/ext/client_config/uri_parser.h ) s.files += %w( src/core/ext/lb_policy/grpclb/grpclb.h ) @@ -488,7 +487,6 @@ Gem::Specification.new do |s| s.files += %w( src/core/ext/client_config/resolver_registry.c ) s.files += %w( src/core/ext/client_config/resolver_result.c ) s.files += %w( src/core/ext/client_config/subchannel.c ) - s.files += %w( src/core/ext/client_config/subchannel_call_holder.c ) s.files += %w( src/core/ext/client_config/subchannel_index.c ) s.files += %w( src/core/ext/client_config/uri_parser.c ) s.files += %w( src/core/ext/transport/chttp2/server/insecure/server_chttp2.c ) diff --git a/package.xml b/package.xml index 1df4940530d..39284b7a5ad 100644 --- a/package.xml +++ b/package.xml @@ -317,7 +317,6 @@ - @@ -496,7 +495,6 @@ - diff --git a/src/core/ext/client_config/client_channel.c b/src/core/ext/client_config/client_channel.c index 566d3d5ce49..e4d7527ce8d 100644 --- a/src/core/ext/client_config/client_channel.c +++ b/src/core/ext/client_config/client_channel.c @@ -33,6 +33,7 @@ #include "src/core/ext/client_config/client_channel.h" +#include #include #include @@ -41,10 +42,11 @@ #include #include -#include "src/core/ext/client_config/subchannel_call_holder.h" +#include "src/core/ext/client_config/subchannel.h" #include "src/core/lib/channel/channel_args.h" #include "src/core/lib/channel/connected_channel.h" #include "src/core/lib/iomgr/iomgr.h" +#include "src/core/lib/iomgr/polling_entity.h" #include "src/core/lib/profiling/timers.h" #include "src/core/lib/support/string.h" #include "src/core/lib/surface/channel.h" @@ -52,13 +54,60 @@ /* Client channel implementation */ -typedef grpc_subchannel_call_holder call_data; +#define GET_CALL(call_data) \ + ((grpc_subchannel_call *)(gpr_atm_acq_load(&(call_data)->subchannel_call))) + +#define CANCELLED_CALL ((grpc_subchannel_call *)1) + +/** Picks a subchannel. + Returns true if subchannel is available immediately (in which case on_ready + should not be called), or false otherwise (in which case on_ready should be + called when the subchannel is available) */ +typedef bool (*pick_subchannel_cb)( + grpc_exec_ctx *exec_ctx, void *arg, grpc_metadata_batch *initial_metadata, + uint32_t initial_metadata_flags, + grpc_connected_subchannel **connected_subchannel, grpc_closure *on_ready); + +typedef enum { + GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING, + GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL +} subchannel_creation_phase; + +/** Call data. Holds a pointer to grpc_subchannel_call and the + associated machinery to create such a pointer. + Handles queueing of stream ops until a call object is ready, waiting + for initial metadata before trying to create a call object, + and handling cancellation gracefully. */ +typedef struct client_channel_call_data { + /** either 0 for no call, 1 for cancelled, or a pointer to a + grpc_subchannel_call */ + gpr_atm subchannel_call; + /** Helper function to choose the subchannel on which to create + the call object. Channel filter delegates to the load + balancing policy (once it's ready). */ + pick_subchannel_cb pick_subchannel; + void *pick_subchannel_arg; + + gpr_mu mu; + + subchannel_creation_phase creation_phase; + grpc_connected_subchannel *connected_subchannel; + grpc_polling_entity *pollent; + + grpc_transport_stream_op *waiting_ops; + size_t waiting_ops_count; + size_t waiting_ops_capacity; + + grpc_closure next_step; + + grpc_call_stack *owning_call; +} call_data; typedef struct client_channel_channel_data { /** resolver for this channel */ grpc_resolver *resolver; /** have we started resolving this channel */ - int started_resolving; + bool started_resolving; /** mutex protecting client configuration, including all variables below in this data structure */ @@ -74,7 +123,7 @@ typedef struct client_channel_channel_data { /** connectivity state being tracked */ grpc_connectivity_state_tracker state_tracker; /** when an lb_policy arrives, should we try to exit idle */ - int exit_idle_when_lb_policy_arrives; + bool exit_idle_when_lb_policy_arrives; /** owning stack */ grpc_channel_stack *owning_stack; /** interested parties (owned) */ @@ -82,10 +131,8 @@ typedef struct client_channel_channel_data { } channel_data; /** We create one watcher for each new lb_policy that is returned from a - resolver, - to watch for state changes from the lb_policy. When a state change is seen, - we - update the channel, and create a new watcher */ + resolver, to watch for state changes from the lb_policy. When a state + change is seen, we update the channel, and create a new watcher. */ typedef struct { channel_data *chand; grpc_closure on_changed; @@ -98,15 +145,208 @@ typedef struct { grpc_call_element *elem; } waiting_call; +static void add_waiting_locked(call_data *calld, grpc_transport_stream_op *op) { + GPR_TIMER_BEGIN("add_waiting_locked", 0); + if (calld->waiting_ops_count == calld->waiting_ops_capacity) { + calld->waiting_ops_capacity = GPR_MAX(3, 2 * calld->waiting_ops_capacity); + calld->waiting_ops = + gpr_realloc(calld->waiting_ops, + calld->waiting_ops_capacity * sizeof(*calld->waiting_ops)); + } + calld->waiting_ops[calld->waiting_ops_count++] = *op; + GPR_TIMER_END("add_waiting_locked", 0); +} + +static void fail_locked(grpc_exec_ctx *exec_ctx, call_data *calld, + grpc_error *error) { + size_t i; + for (i = 0; i < calld->waiting_ops_count; i++) { + grpc_transport_stream_op_finish_with_failure( + exec_ctx, &calld->waiting_ops[i], GRPC_ERROR_REF(error)); + } + calld->waiting_ops_count = 0; + GRPC_ERROR_UNREF(error); +} + +typedef struct { + grpc_transport_stream_op *ops; + size_t nops; + grpc_subchannel_call *call; +} retry_ops_args; + +static void retry_ops(grpc_exec_ctx *exec_ctx, void *args, grpc_error *error) { + retry_ops_args *a = args; + size_t i; + for (i = 0; i < a->nops; i++) { + grpc_subchannel_call_process_op(exec_ctx, a->call, &a->ops[i]); + } + GRPC_SUBCHANNEL_CALL_UNREF(exec_ctx, a->call, "retry_ops"); + gpr_free(a->ops); + gpr_free(a); +} + +static void retry_waiting_locked(grpc_exec_ctx *exec_ctx, call_data *calld) { + retry_ops_args *a = gpr_malloc(sizeof(*a)); + a->ops = calld->waiting_ops; + a->nops = calld->waiting_ops_count; + a->call = GET_CALL(calld); + if (a->call == CANCELLED_CALL) { + gpr_free(a); + fail_locked(exec_ctx, calld, GRPC_ERROR_CANCELLED); + return; + } + calld->waiting_ops = NULL; + calld->waiting_ops_count = 0; + calld->waiting_ops_capacity = 0; + GRPC_SUBCHANNEL_CALL_REF(a->call, "retry_ops"); + grpc_exec_ctx_sched(exec_ctx, grpc_closure_create(retry_ops, a), + GRPC_ERROR_NONE, NULL); +} + +static void subchannel_ready(grpc_exec_ctx *exec_ctx, void *arg, + grpc_error *error) { + call_data *calld = arg; + gpr_mu_lock(&calld->mu); + GPR_ASSERT(calld->creation_phase == + GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL); + calld->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING; + if (calld->connected_subchannel == NULL) { + gpr_atm_no_barrier_store(&calld->subchannel_call, 1); + fail_locked(exec_ctx, calld, GRPC_ERROR_CREATE_REFERENCING( + "Failed to create subchannel", &error, 1)); + } else if (1 == gpr_atm_acq_load(&calld->subchannel_call)) { + /* already cancelled before subchannel became ready */ + fail_locked(exec_ctx, calld, + GRPC_ERROR_CREATE_REFERENCING( + "Cancelled before creating subchannel", &error, 1)); + } else { + grpc_subchannel_call *subchannel_call = NULL; + grpc_error *new_error = grpc_connected_subchannel_create_call( + exec_ctx, calld->connected_subchannel, calld->pollent, + &subchannel_call); + if (new_error != GRPC_ERROR_NONE) { + new_error = grpc_error_add_child(new_error, error); + subchannel_call = CANCELLED_CALL; + fail_locked(exec_ctx, calld, new_error); + } + gpr_atm_rel_store(&calld->subchannel_call, + (gpr_atm)(uintptr_t)subchannel_call); + retry_waiting_locked(exec_ctx, calld); + } + gpr_mu_unlock(&calld->mu); + GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call, "pick_subchannel"); +} + static char *cc_get_peer(grpc_exec_ctx *exec_ctx, grpc_call_element *elem) { - return grpc_subchannel_call_holder_get_peer(exec_ctx, elem->call_data); + call_data *calld = elem->call_data; + grpc_subchannel_call *subchannel_call = GET_CALL(calld); + if (subchannel_call == NULL || subchannel_call == CANCELLED_CALL) { + return NULL; + } else { + return grpc_subchannel_call_get_peer(exec_ctx, subchannel_call); + } } +// The logic here is fairly complicated, due to (a) the fact that we +// need to handle the case where we receive the send op before the +// initial metadata op, and (b) the need for efficiency, especially in +// the streaming case. +// TODO(ctiller): Explain this more thoroughly. static void cc_start_transport_stream_op(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, grpc_transport_stream_op *op) { + call_data *calld = elem->call_data; GRPC_CALL_LOG_OP(GPR_INFO, elem, op); - grpc_subchannel_call_holder_perform_op(exec_ctx, elem->call_data, op); + /* try to (atomically) get the call */ + grpc_subchannel_call *call = GET_CALL(calld); + GPR_TIMER_BEGIN("cc_start_transport_stream_op", 0); + if (call == CANCELLED_CALL) { + grpc_transport_stream_op_finish_with_failure(exec_ctx, op, + GRPC_ERROR_CANCELLED); + GPR_TIMER_END("cc_start_transport_stream_op", 0); + return; + } + if (call != NULL) { + grpc_subchannel_call_process_op(exec_ctx, call, op); + GPR_TIMER_END("cc_start_transport_stream_op", 0); + return; + } + /* we failed; lock and figure out what to do */ + gpr_mu_lock(&calld->mu); +retry: + /* need to recheck that another thread hasn't set the call */ + call = GET_CALL(calld); + if (call == CANCELLED_CALL) { + gpr_mu_unlock(&calld->mu); + grpc_transport_stream_op_finish_with_failure(exec_ctx, op, + GRPC_ERROR_CANCELLED); + GPR_TIMER_END("cc_start_transport_stream_op", 0); + return; + } + if (call != NULL) { + gpr_mu_unlock(&calld->mu); + grpc_subchannel_call_process_op(exec_ctx, call, op); + GPR_TIMER_END("cc_start_transport_stream_op", 0); + return; + } + /* if this is a cancellation, then we can raise our cancelled flag */ + if (op->cancel_error != GRPC_ERROR_NONE) { + if (!gpr_atm_rel_cas(&calld->subchannel_call, 0, + (gpr_atm)(uintptr_t)CANCELLED_CALL)) { + goto retry; + } else { + switch (calld->creation_phase) { + case GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING: + fail_locked(exec_ctx, calld, GRPC_ERROR_REF(op->cancel_error)); + break; + case GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL: + calld->pick_subchannel(exec_ctx, calld->pick_subchannel_arg, NULL, 0, + &calld->connected_subchannel, NULL); + break; + } + gpr_mu_unlock(&calld->mu); + grpc_transport_stream_op_finish_with_failure(exec_ctx, op, + GRPC_ERROR_CANCELLED); + GPR_TIMER_END("cc_start_transport_stream_op", 0); + return; + } + } + /* if we don't have a subchannel, try to get one */ + if (calld->creation_phase == GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING && + calld->connected_subchannel == NULL && + op->send_initial_metadata != NULL) { + calld->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL; + grpc_closure_init(&calld->next_step, subchannel_ready, calld); + GRPC_CALL_STACK_REF(calld->owning_call, "pick_subchannel"); + if (calld->pick_subchannel( + exec_ctx, calld->pick_subchannel_arg, op->send_initial_metadata, + op->send_initial_metadata_flags, &calld->connected_subchannel, + &calld->next_step)) { + calld->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING; + GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call, "pick_subchannel"); + } + } + /* if we've got a subchannel, then let's ask it to create a call */ + if (calld->creation_phase == GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING && + calld->connected_subchannel != NULL) { + grpc_subchannel_call *subchannel_call = NULL; + grpc_error *error = grpc_connected_subchannel_create_call( + exec_ctx, calld->connected_subchannel, calld->pollent, + &subchannel_call); + if (error != GRPC_ERROR_NONE) { + subchannel_call = CANCELLED_CALL; + fail_locked(exec_ctx, calld, GRPC_ERROR_REF(error)); + grpc_transport_stream_op_finish_with_failure(exec_ctx, op, error); + } + gpr_atm_rel_store(&calld->subchannel_call, + (gpr_atm)(uintptr_t)subchannel_call); + retry_waiting_locked(exec_ctx, calld); + goto retry; + } + /* nothing to be done but wait */ + add_waiting_locked(calld, op); + gpr_mu_unlock(&calld->mu); + GPR_TIMER_END("cc_start_transport_stream_op", 0); } static void watch_lb_policy(grpc_exec_ctx *exec_ctx, channel_data *chand, @@ -183,7 +423,7 @@ static void cc_on_resolver_result_changed(grpc_exec_ctx *exec_ctx, void *arg, grpc_lb_policy *lb_policy = NULL; grpc_lb_policy *old_lb_policy; grpc_connectivity_state state = GRPC_CHANNEL_TRANSIENT_FAILURE; - int exit_idle = 0; + bool exit_idle = false; grpc_error *state_error = GRPC_ERROR_CREATE("No load balancing policy"); if (chand->resolver_result != NULL) { @@ -221,8 +461,8 @@ static void cc_on_resolver_result_changed(grpc_exec_ctx *exec_ctx, void *arg, } if (lb_policy != NULL && chand->exit_idle_when_lb_policy_arrives) { GRPC_LB_POLICY_REF(lb_policy, "exit_idle"); - exit_idle = 1; - chand->exit_idle_when_lb_policy_arrives = 0; + exit_idle = true; + chand->exit_idle_when_lb_policy_arrives = false; } if (error == GRPC_ERROR_NONE && chand->resolver) { @@ -339,11 +579,11 @@ typedef struct { grpc_closure closure; } continue_picking_args; -static int cc_pick_subchannel(grpc_exec_ctx *exec_ctx, void *arg, - grpc_metadata_batch *initial_metadata, - uint32_t initial_metadata_flags, - grpc_connected_subchannel **connected_subchannel, - grpc_closure *on_ready); +static bool cc_pick_subchannel(grpc_exec_ctx *exec_ctx, void *arg, + grpc_metadata_batch *initial_metadata, + uint32_t initial_metadata_flags, + grpc_connected_subchannel **connected_subchannel, + grpc_closure *on_ready); static void continue_picking(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { @@ -360,11 +600,11 @@ static void continue_picking(grpc_exec_ctx *exec_ctx, void *arg, gpr_free(cpa); } -static int cc_pick_subchannel(grpc_exec_ctx *exec_ctx, void *elemp, - grpc_metadata_batch *initial_metadata, - uint32_t initial_metadata_flags, - grpc_connected_subchannel **connected_subchannel, - grpc_closure *on_ready) { +static bool cc_pick_subchannel(grpc_exec_ctx *exec_ctx, void *elemp, + grpc_metadata_batch *initial_metadata, + uint32_t initial_metadata_flags, + grpc_connected_subchannel **connected_subchannel, + grpc_closure *on_ready) { GPR_TIMER_BEGIN("cc_pick_subchannel", 0); grpc_call_element *elem = elemp; @@ -392,7 +632,7 @@ static int cc_pick_subchannel(grpc_exec_ctx *exec_ctx, void *elemp, } gpr_mu_unlock(&chand->mu); GPR_TIMER_END("cc_pick_subchannel", 0); - return 1; + return true; } if (chand->lb_policy != NULL) { grpc_lb_policy *lb_policy = chand->lb_policy; @@ -407,7 +647,7 @@ static int cc_pick_subchannel(grpc_exec_ctx *exec_ctx, void *elemp, return r; } if (chand->resolver != NULL && !chand->started_resolving) { - chand->started_resolving = 1; + chand->started_resolving = true; GRPC_CHANNEL_STACK_REF(chand->owning_stack, "resolver"); grpc_resolver_next(exec_ctx, chand->resolver, &chand->resolver_result, &chand->on_resolver_result_changed); @@ -429,15 +669,25 @@ static int cc_pick_subchannel(grpc_exec_ctx *exec_ctx, void *elemp, gpr_mu_unlock(&chand->mu); GPR_TIMER_END("cc_pick_subchannel", 0); - return 0; + return false; } /* Constructor for call_data */ static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, grpc_call_element_args *args) { - grpc_subchannel_call_holder_init(elem->call_data, cc_pick_subchannel, elem, - args->call_stack); + call_data *calld = elem->call_data; + gpr_atm_rel_store(&calld->subchannel_call, 0); + calld->pick_subchannel = cc_pick_subchannel; + calld->pick_subchannel_arg = elem; + gpr_mu_init(&calld->mu); + calld->connected_subchannel = NULL; + calld->waiting_ops = NULL; + calld->waiting_ops_count = 0; + calld->waiting_ops_capacity = 0; + calld->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING; + calld->owning_call = args->call_stack; + calld->pollent = NULL; return GRPC_ERROR_NONE; } @@ -445,7 +695,15 @@ static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx, static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, const grpc_call_final_info *final_info, void *and_free_memory) { - grpc_subchannel_call_holder_destroy(exec_ctx, elem->call_data); + call_data *calld = elem->call_data; + grpc_subchannel_call *call = GET_CALL(calld); + if (call != NULL && call != CANCELLED_CALL) { + GRPC_SUBCHANNEL_CALL_UNREF(exec_ctx, call, "client_channel_destroy_call"); + } + GPR_ASSERT(calld->creation_phase == GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING); + gpr_mu_destroy(&calld->mu); + GPR_ASSERT(calld->waiting_ops_count == 0); + gpr_free(calld->waiting_ops); gpr_free(and_free_memory); } @@ -523,7 +781,7 @@ void grpc_client_channel_set_resolver(grpc_exec_ctx *exec_ctx, GRPC_RESOLVER_REF(resolver, "channel"); if (!grpc_closure_list_empty(chand->waiting_for_config_closures) || chand->exit_idle_when_lb_policy_arrives) { - chand->started_resolving = 1; + chand->started_resolving = true; GRPC_CHANNEL_STACK_REF(chand->owning_stack, "resolver"); grpc_resolver_next(exec_ctx, resolver, &chand->resolver_result, &chand->on_resolver_result_changed); @@ -541,10 +799,10 @@ grpc_connectivity_state grpc_client_channel_check_connectivity_state( if (chand->lb_policy != NULL) { grpc_lb_policy_exit_idle(exec_ctx, chand->lb_policy); } else { - chand->exit_idle_when_lb_policy_arrives = 1; + chand->exit_idle_when_lb_policy_arrives = true; if (!chand->started_resolving && chand->resolver != NULL) { GRPC_CHANNEL_STACK_REF(chand->owning_stack, "resolver"); - chand->started_resolving = 1; + chand->started_resolving = true; grpc_resolver_next(exec_ctx, chand->resolver, &chand->resolver_result, &chand->on_resolver_result_changed); } diff --git a/src/core/ext/client_config/subchannel_call_holder.c b/src/core/ext/client_config/subchannel_call_holder.c deleted file mode 100644 index be6d054af4e..00000000000 --- a/src/core/ext/client_config/subchannel_call_holder.c +++ /dev/null @@ -1,292 +0,0 @@ -/* - * - * Copyright 2015, Google Inc. - * All rights reserved. - * - * Redistribution and use in source and binary forms, with or without - * modification, are permitted provided that the following conditions are - * met: - * - * * Redistributions of source code must retain the above copyright - * notice, this list of conditions and the following disclaimer. - * * Redistributions in binary form must reproduce the above - * copyright notice, this list of conditions and the following disclaimer - * in the documentation and/or other materials provided with the - * distribution. - * * Neither the name of Google Inc. nor the names of its - * contributors may be used to endorse or promote products derived from - * this software without specific prior written permission. - * - * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS - * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT - * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR - * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT - * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, - * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT - * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, - * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY - * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT - * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE - * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. - * - */ - -#include "src/core/ext/client_config/subchannel_call_holder.h" - -#include - -#include "src/core/lib/profiling/timers.h" - -#define GET_CALL(holder) \ - ((grpc_subchannel_call *)(gpr_atm_acq_load(&(holder)->subchannel_call))) - -#define CANCELLED_CALL ((grpc_subchannel_call *)1) - -static void subchannel_ready(grpc_exec_ctx *exec_ctx, void *holder, - grpc_error *error); -static void retry_ops(grpc_exec_ctx *exec_ctx, void *retry_ops_args, - grpc_error *error); - -static void add_waiting_locked(grpc_subchannel_call_holder *holder, - grpc_transport_stream_op *op); -static void fail_locked(grpc_exec_ctx *exec_ctx, - grpc_subchannel_call_holder *holder, grpc_error *error); -static void retry_waiting_locked(grpc_exec_ctx *exec_ctx, - grpc_subchannel_call_holder *holder); - -void grpc_subchannel_call_holder_init( - grpc_subchannel_call_holder *holder, - grpc_subchannel_call_holder_pick_subchannel pick_subchannel, - void *pick_subchannel_arg, grpc_call_stack *owning_call) { - gpr_atm_rel_store(&holder->subchannel_call, 0); - holder->pick_subchannel = pick_subchannel; - holder->pick_subchannel_arg = pick_subchannel_arg; - gpr_mu_init(&holder->mu); - holder->connected_subchannel = NULL; - holder->waiting_ops = NULL; - holder->waiting_ops_count = 0; - holder->waiting_ops_capacity = 0; - holder->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING; - holder->owning_call = owning_call; - holder->pollent = NULL; -} - -void grpc_subchannel_call_holder_destroy(grpc_exec_ctx *exec_ctx, - grpc_subchannel_call_holder *holder) { - grpc_subchannel_call *call = GET_CALL(holder); - if (call != NULL && call != CANCELLED_CALL) { - GRPC_SUBCHANNEL_CALL_UNREF(exec_ctx, call, "holder"); - } - GPR_ASSERT(holder->creation_phase == - GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING); - gpr_mu_destroy(&holder->mu); - GPR_ASSERT(holder->waiting_ops_count == 0); - gpr_free(holder->waiting_ops); -} - -// The logic here is fairly complicated, due to (a) the fact that we -// need to handle the case where we receive the send op before the -// initial metadata op, and (b) the need for efficiency, especially in -// the streaming case. -// TODO(ctiller): Explain this more thoroughly. -void grpc_subchannel_call_holder_perform_op(grpc_exec_ctx *exec_ctx, - grpc_subchannel_call_holder *holder, - grpc_transport_stream_op *op) { - /* try to (atomically) get the call */ - grpc_subchannel_call *call = GET_CALL(holder); - GPR_TIMER_BEGIN("grpc_subchannel_call_holder_perform_op", 0); - if (call == CANCELLED_CALL) { - grpc_transport_stream_op_finish_with_failure(exec_ctx, op, - GRPC_ERROR_CANCELLED); - GPR_TIMER_END("grpc_subchannel_call_holder_perform_op", 0); - return; - } - if (call != NULL) { - grpc_subchannel_call_process_op(exec_ctx, call, op); - GPR_TIMER_END("grpc_subchannel_call_holder_perform_op", 0); - return; - } - /* we failed; lock and figure out what to do */ - gpr_mu_lock(&holder->mu); -retry: - /* need to recheck that another thread hasn't set the call */ - call = GET_CALL(holder); - if (call == CANCELLED_CALL) { - gpr_mu_unlock(&holder->mu); - grpc_transport_stream_op_finish_with_failure(exec_ctx, op, - GRPC_ERROR_CANCELLED); - GPR_TIMER_END("grpc_subchannel_call_holder_perform_op", 0); - return; - } - if (call != NULL) { - gpr_mu_unlock(&holder->mu); - grpc_subchannel_call_process_op(exec_ctx, call, op); - GPR_TIMER_END("grpc_subchannel_call_holder_perform_op", 0); - return; - } - /* if this is a cancellation, then we can raise our cancelled flag */ - if (op->cancel_error != GRPC_ERROR_NONE) { - if (!gpr_atm_rel_cas(&holder->subchannel_call, 0, - (gpr_atm)(uintptr_t)CANCELLED_CALL)) { - goto retry; - } else { - switch (holder->creation_phase) { - case GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING: - fail_locked(exec_ctx, holder, GRPC_ERROR_REF(op->cancel_error)); - break; - case GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL: - holder->pick_subchannel(exec_ctx, holder->pick_subchannel_arg, NULL, - 0, &holder->connected_subchannel, NULL); - break; - } - gpr_mu_unlock(&holder->mu); - grpc_transport_stream_op_finish_with_failure(exec_ctx, op, - GRPC_ERROR_CANCELLED); - GPR_TIMER_END("grpc_subchannel_call_holder_perform_op", 0); - return; - } - } - /* if we don't have a subchannel, try to get one */ - if (holder->creation_phase == GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING && - holder->connected_subchannel == NULL && - op->send_initial_metadata != NULL) { - holder->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL; - grpc_closure_init(&holder->next_step, subchannel_ready, holder); - GRPC_CALL_STACK_REF(holder->owning_call, "pick_subchannel"); - if (holder->pick_subchannel( - exec_ctx, holder->pick_subchannel_arg, op->send_initial_metadata, - op->send_initial_metadata_flags, &holder->connected_subchannel, - &holder->next_step)) { - holder->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING; - GRPC_CALL_STACK_UNREF(exec_ctx, holder->owning_call, "pick_subchannel"); - } - } - /* if we've got a subchannel, then let's ask it to create a call */ - if (holder->creation_phase == GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING && - holder->connected_subchannel != NULL) { - grpc_subchannel_call *subchannel_call = NULL; - grpc_error *error = grpc_connected_subchannel_create_call( - exec_ctx, holder->connected_subchannel, holder->pollent, - &subchannel_call); - if (error != GRPC_ERROR_NONE) { - subchannel_call = CANCELLED_CALL; - fail_locked(exec_ctx, holder, GRPC_ERROR_REF(error)); - grpc_transport_stream_op_finish_with_failure(exec_ctx, op, error); - } - gpr_atm_rel_store(&holder->subchannel_call, - (gpr_atm)(uintptr_t)subchannel_call); - retry_waiting_locked(exec_ctx, holder); - goto retry; - } - /* nothing to be done but wait */ - add_waiting_locked(holder, op); - gpr_mu_unlock(&holder->mu); - GPR_TIMER_END("grpc_subchannel_call_holder_perform_op", 0); -} - -static void subchannel_ready(grpc_exec_ctx *exec_ctx, void *arg, - grpc_error *error) { - grpc_subchannel_call_holder *holder = arg; - gpr_mu_lock(&holder->mu); - GPR_ASSERT(holder->creation_phase == - GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL); - holder->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING; - if (holder->connected_subchannel == NULL) { - gpr_atm_no_barrier_store(&holder->subchannel_call, 1); - fail_locked(exec_ctx, holder, - GRPC_ERROR_CREATE_REFERENCING("Failed to create subchannel", - &error, 1)); - } else if (1 == gpr_atm_acq_load(&holder->subchannel_call)) { - /* already cancelled before subchannel became ready */ - fail_locked(exec_ctx, holder, - GRPC_ERROR_CREATE_REFERENCING( - "Cancelled before creating subchannel", &error, 1)); - } else { - grpc_subchannel_call *subchannel_call = NULL; - grpc_error *new_error = grpc_connected_subchannel_create_call( - exec_ctx, holder->connected_subchannel, holder->pollent, - &subchannel_call); - if (new_error != GRPC_ERROR_NONE) { - new_error = grpc_error_add_child(new_error, error); - subchannel_call = CANCELLED_CALL; - fail_locked(exec_ctx, holder, new_error); - } - gpr_atm_rel_store(&holder->subchannel_call, - (gpr_atm)(uintptr_t)subchannel_call); - retry_waiting_locked(exec_ctx, holder); - } - gpr_mu_unlock(&holder->mu); - GRPC_CALL_STACK_UNREF(exec_ctx, holder->owning_call, "pick_subchannel"); -} - -typedef struct { - grpc_transport_stream_op *ops; - size_t nops; - grpc_subchannel_call *call; -} retry_ops_args; - -static void retry_waiting_locked(grpc_exec_ctx *exec_ctx, - grpc_subchannel_call_holder *holder) { - retry_ops_args *a = gpr_malloc(sizeof(*a)); - a->ops = holder->waiting_ops; - a->nops = holder->waiting_ops_count; - a->call = GET_CALL(holder); - if (a->call == CANCELLED_CALL) { - gpr_free(a); - fail_locked(exec_ctx, holder, GRPC_ERROR_CANCELLED); - return; - } - holder->waiting_ops = NULL; - holder->waiting_ops_count = 0; - holder->waiting_ops_capacity = 0; - GRPC_SUBCHANNEL_CALL_REF(a->call, "retry_ops"); - grpc_exec_ctx_sched(exec_ctx, grpc_closure_create(retry_ops, a), - GRPC_ERROR_NONE, NULL); -} - -static void retry_ops(grpc_exec_ctx *exec_ctx, void *args, grpc_error *error) { - retry_ops_args *a = args; - size_t i; - for (i = 0; i < a->nops; i++) { - grpc_subchannel_call_process_op(exec_ctx, a->call, &a->ops[i]); - } - GRPC_SUBCHANNEL_CALL_UNREF(exec_ctx, a->call, "retry_ops"); - gpr_free(a->ops); - gpr_free(a); -} - -static void add_waiting_locked(grpc_subchannel_call_holder *holder, - grpc_transport_stream_op *op) { - GPR_TIMER_BEGIN("add_waiting_locked", 0); - if (holder->waiting_ops_count == holder->waiting_ops_capacity) { - holder->waiting_ops_capacity = GPR_MAX(3, 2 * holder->waiting_ops_capacity); - holder->waiting_ops = - gpr_realloc(holder->waiting_ops, holder->waiting_ops_capacity * - sizeof(*holder->waiting_ops)); - } - holder->waiting_ops[holder->waiting_ops_count++] = *op; - GPR_TIMER_END("add_waiting_locked", 0); -} - -static void fail_locked(grpc_exec_ctx *exec_ctx, - grpc_subchannel_call_holder *holder, - grpc_error *error) { - size_t i; - for (i = 0; i < holder->waiting_ops_count; i++) { - grpc_transport_stream_op_finish_with_failure( - exec_ctx, &holder->waiting_ops[i], GRPC_ERROR_REF(error)); - } - holder->waiting_ops_count = 0; - GRPC_ERROR_UNREF(error); -} - -char *grpc_subchannel_call_holder_get_peer( - grpc_exec_ctx *exec_ctx, grpc_subchannel_call_holder *holder) { - grpc_subchannel_call *subchannel_call = GET_CALL(holder); - - if (subchannel_call == NULL || subchannel_call == CANCELLED_CALL) { - return NULL; - } else { - return grpc_subchannel_call_get_peer(exec_ctx, subchannel_call); - } -} diff --git a/src/core/ext/client_config/subchannel_call_holder.h b/src/core/ext/client_config/subchannel_call_holder.h deleted file mode 100644 index 8d2deb02f38..00000000000 --- a/src/core/ext/client_config/subchannel_call_holder.h +++ /dev/null @@ -1,99 +0,0 @@ -/* - * - * Copyright 2015, Google Inc. - * All rights reserved. - * - * Redistribution and use in source and binary forms, with or without - * modification, are permitted provided that the following conditions are - * met: - * - * * Redistributions of source code must retain the above copyright - * notice, this list of conditions and the following disclaimer. - * * Redistributions in binary form must reproduce the above - * copyright notice, this list of conditions and the following disclaimer - * in the documentation and/or other materials provided with the - * distribution. - * * Neither the name of Google Inc. nor the names of its - * contributors may be used to endorse or promote products derived from - * this software without specific prior written permission. - * - * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS - * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT - * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR - * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT - * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, - * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT - * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, - * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY - * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT - * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE - * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. - * - */ - -#ifndef GRPC_CORE_EXT_CLIENT_CONFIG_SUBCHANNEL_CALL_HOLDER_H -#define GRPC_CORE_EXT_CLIENT_CONFIG_SUBCHANNEL_CALL_HOLDER_H - -#include "src/core/ext/client_config/subchannel.h" -#include "src/core/lib/iomgr/polling_entity.h" - -/** Pick a subchannel for grpc_subchannel_call_holder; - Return 1 if subchannel is available immediately (in which case on_ready - should not be called), or 0 otherwise (in which case on_ready should be - called when the subchannel is available) */ -typedef int (*grpc_subchannel_call_holder_pick_subchannel)( - grpc_exec_ctx *exec_ctx, void *arg, grpc_metadata_batch *initial_metadata, - uint32_t initial_metadata_flags, - grpc_connected_subchannel **connected_subchannel, grpc_closure *on_ready); - -typedef enum { - GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING, - GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL -} grpc_subchannel_call_holder_creation_phase; - -/** Wrapper for holding a pointer to grpc_subchannel_call, and the - associated machinery to create such a pointer. - Handles queueing of stream ops until a call object is ready, waiting - for initial metadata before trying to create a call object, - and handling cancellation gracefully. - - The channel filter uses this as their call_data. */ -typedef struct grpc_subchannel_call_holder { - /** either 0 for no call, 1 for cancelled, or a pointer to a - grpc_subchannel_call */ - gpr_atm subchannel_call; - /** Helper function to choose the subchannel on which to create - the call object. Channel filter delegates to the load - balancing policy (once it's ready). */ - grpc_subchannel_call_holder_pick_subchannel pick_subchannel; - void *pick_subchannel_arg; - - gpr_mu mu; - - grpc_subchannel_call_holder_creation_phase creation_phase; - grpc_connected_subchannel *connected_subchannel; - grpc_polling_entity *pollent; - - grpc_transport_stream_op *waiting_ops; - size_t waiting_ops_count; - size_t waiting_ops_capacity; - - grpc_closure next_step; - - grpc_call_stack *owning_call; -} grpc_subchannel_call_holder; - -void grpc_subchannel_call_holder_init( - grpc_subchannel_call_holder *holder, - grpc_subchannel_call_holder_pick_subchannel pick_subchannel, - void *pick_subchannel_arg, grpc_call_stack *owning_call); -void grpc_subchannel_call_holder_destroy(grpc_exec_ctx *exec_ctx, - grpc_subchannel_call_holder *holder); - -void grpc_subchannel_call_holder_perform_op(grpc_exec_ctx *exec_ctx, - grpc_subchannel_call_holder *holder, - grpc_transport_stream_op *op); -char *grpc_subchannel_call_holder_get_peer(grpc_exec_ctx *exec_ctx, - grpc_subchannel_call_holder *holder); - -#endif /* GRPC_CORE_EXT_CLIENT_CONFIG_SUBCHANNEL_CALL_HOLDER_H */ diff --git a/src/python/grpcio/grpc_core_dependencies.py b/src/python/grpcio/grpc_core_dependencies.py index 660e34d7420..a3ff66224a2 100644 --- a/src/python/grpcio/grpc_core_dependencies.py +++ b/src/python/grpcio/grpc_core_dependencies.py @@ -233,7 +233,6 @@ CORE_SOURCE_FILES = [ 'src/core/ext/client_config/resolver_registry.c', 'src/core/ext/client_config/resolver_result.c', 'src/core/ext/client_config/subchannel.c', - 'src/core/ext/client_config/subchannel_call_holder.c', 'src/core/ext/client_config/subchannel_index.c', 'src/core/ext/client_config/uri_parser.c', 'src/core/ext/transport/chttp2/server/insecure/server_chttp2.c', diff --git a/tools/doxygen/Doxyfile.core.internal b/tools/doxygen/Doxyfile.core.internal index 77f951af50e..94ade7787e9 100644 --- a/tools/doxygen/Doxyfile.core.internal +++ b/tools/doxygen/Doxyfile.core.internal @@ -928,7 +928,6 @@ src/core/ext/client_config/resolver_factory.h \ src/core/ext/client_config/resolver_registry.h \ src/core/ext/client_config/resolver_result.h \ src/core/ext/client_config/subchannel.h \ -src/core/ext/client_config/subchannel_call_holder.h \ src/core/ext/client_config/subchannel_index.h \ src/core/ext/client_config/uri_parser.h \ src/core/ext/lb_policy/grpclb/grpclb.h \ @@ -1107,7 +1106,6 @@ src/core/ext/client_config/resolver_factory.c \ src/core/ext/client_config/resolver_registry.c \ src/core/ext/client_config/resolver_result.c \ src/core/ext/client_config/subchannel.c \ -src/core/ext/client_config/subchannel_call_holder.c \ src/core/ext/client_config/subchannel_index.c \ src/core/ext/client_config/uri_parser.c \ src/core/ext/transport/chttp2/server/insecure/server_chttp2.c \ diff --git a/tools/run_tests/sources_and_headers.json b/tools/run_tests/sources_and_headers.json index ce94c5d5e89..b6b220f449f 100644 --- a/tools/run_tests/sources_and_headers.json +++ b/tools/run_tests/sources_and_headers.json @@ -6100,7 +6100,6 @@ "src/core/ext/client_config/resolver_registry.h", "src/core/ext/client_config/resolver_result.h", "src/core/ext/client_config/subchannel.h", - "src/core/ext/client_config/subchannel_call_holder.h", "src/core/ext/client_config/subchannel_index.h", "src/core/ext/client_config/uri_parser.h" ], @@ -6136,8 +6135,6 @@ "src/core/ext/client_config/resolver_result.h", "src/core/ext/client_config/subchannel.c", "src/core/ext/client_config/subchannel.h", - "src/core/ext/client_config/subchannel_call_holder.c", - "src/core/ext/client_config/subchannel_call_holder.h", "src/core/ext/client_config/subchannel_index.c", "src/core/ext/client_config/subchannel_index.h", "src/core/ext/client_config/uri_parser.c", diff --git a/vsprojects/vcxproj/grpc/grpc.vcxproj b/vsprojects/vcxproj/grpc/grpc.vcxproj index d805fdfdee5..9dcb76c48bc 100644 --- a/vsprojects/vcxproj/grpc/grpc.vcxproj +++ b/vsprojects/vcxproj/grpc/grpc.vcxproj @@ -437,7 +437,6 @@ - @@ -776,8 +775,6 @@ - - diff --git a/vsprojects/vcxproj/grpc/grpc.vcxproj.filters b/vsprojects/vcxproj/grpc/grpc.vcxproj.filters index 991bbec9e25..637dade8e1f 100644 --- a/vsprojects/vcxproj/grpc/grpc.vcxproj.filters +++ b/vsprojects/vcxproj/grpc/grpc.vcxproj.filters @@ -475,9 +475,6 @@ src\core\ext\client_config - - src\core\ext\client_config - src\core\ext\client_config @@ -1079,9 +1076,6 @@ src\core\ext\client_config - - src\core\ext\client_config - src\core\ext\client_config diff --git a/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj b/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj index 66ec41d2c6a..e77ea8dd720 100644 --- a/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj +++ b/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj @@ -403,7 +403,6 @@ - @@ -692,8 +691,6 @@ - - diff --git a/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj.filters b/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj.filters index 6f2dc3571e8..d97f8248c91 100644 --- a/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj.filters +++ b/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj.filters @@ -400,9 +400,6 @@ src\core\ext\client_config - - src\core\ext\client_config - src\core\ext\client_config @@ -917,9 +914,6 @@ src\core\ext\client_config - - src\core\ext\client_config - src\core\ext\client_config From 2a5959f2aeca2e86cc752288d890bc10b24c55d0 Mon Sep 17 00:00:00 2001 From: "Mark D. Roth" Date: Thu, 1 Sep 2016 08:20:27 -0700 Subject: [PATCH 6/7] A bit more cleanup. --- src/core/ext/client_config/client_channel.c | 679 ++++++++++---------- 1 file changed, 335 insertions(+), 344 deletions(-) diff --git a/src/core/ext/client_config/client_channel.c b/src/core/ext/client_config/client_channel.c index e4d7527ce8d..fff80abb802 100644 --- a/src/core/ext/client_config/client_channel.c +++ b/src/core/ext/client_config/client_channel.c @@ -54,54 +54,9 @@ /* Client channel implementation */ -#define GET_CALL(call_data) \ - ((grpc_subchannel_call *)(gpr_atm_acq_load(&(call_data)->subchannel_call))) - -#define CANCELLED_CALL ((grpc_subchannel_call *)1) - -/** Picks a subchannel. - Returns true if subchannel is available immediately (in which case on_ready - should not be called), or false otherwise (in which case on_ready should be - called when the subchannel is available) */ -typedef bool (*pick_subchannel_cb)( - grpc_exec_ctx *exec_ctx, void *arg, grpc_metadata_batch *initial_metadata, - uint32_t initial_metadata_flags, - grpc_connected_subchannel **connected_subchannel, grpc_closure *on_ready); - -typedef enum { - GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING, - GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL -} subchannel_creation_phase; - -/** Call data. Holds a pointer to grpc_subchannel_call and the - associated machinery to create such a pointer. - Handles queueing of stream ops until a call object is ready, waiting - for initial metadata before trying to create a call object, - and handling cancellation gracefully. */ -typedef struct client_channel_call_data { - /** either 0 for no call, 1 for cancelled, or a pointer to a - grpc_subchannel_call */ - gpr_atm subchannel_call; - /** Helper function to choose the subchannel on which to create - the call object. Channel filter delegates to the load - balancing policy (once it's ready). */ - pick_subchannel_cb pick_subchannel; - void *pick_subchannel_arg; - - gpr_mu mu; - - subchannel_creation_phase creation_phase; - grpc_connected_subchannel *connected_subchannel; - grpc_polling_entity *pollent; - - grpc_transport_stream_op *waiting_ops; - size_t waiting_ops_count; - size_t waiting_ops_capacity; - - grpc_closure next_step; - - grpc_call_stack *owning_call; -} call_data; +/************************************************************************* + * CHANNEL-WIDE FUNCTIONS + */ typedef struct client_channel_channel_data { /** resolver for this channel */ @@ -140,215 +95,6 @@ typedef struct { grpc_lb_policy *lb_policy; } lb_policy_connectivity_watcher; -typedef struct { - grpc_closure closure; - grpc_call_element *elem; -} waiting_call; - -static void add_waiting_locked(call_data *calld, grpc_transport_stream_op *op) { - GPR_TIMER_BEGIN("add_waiting_locked", 0); - if (calld->waiting_ops_count == calld->waiting_ops_capacity) { - calld->waiting_ops_capacity = GPR_MAX(3, 2 * calld->waiting_ops_capacity); - calld->waiting_ops = - gpr_realloc(calld->waiting_ops, - calld->waiting_ops_capacity * sizeof(*calld->waiting_ops)); - } - calld->waiting_ops[calld->waiting_ops_count++] = *op; - GPR_TIMER_END("add_waiting_locked", 0); -} - -static void fail_locked(grpc_exec_ctx *exec_ctx, call_data *calld, - grpc_error *error) { - size_t i; - for (i = 0; i < calld->waiting_ops_count; i++) { - grpc_transport_stream_op_finish_with_failure( - exec_ctx, &calld->waiting_ops[i], GRPC_ERROR_REF(error)); - } - calld->waiting_ops_count = 0; - GRPC_ERROR_UNREF(error); -} - -typedef struct { - grpc_transport_stream_op *ops; - size_t nops; - grpc_subchannel_call *call; -} retry_ops_args; - -static void retry_ops(grpc_exec_ctx *exec_ctx, void *args, grpc_error *error) { - retry_ops_args *a = args; - size_t i; - for (i = 0; i < a->nops; i++) { - grpc_subchannel_call_process_op(exec_ctx, a->call, &a->ops[i]); - } - GRPC_SUBCHANNEL_CALL_UNREF(exec_ctx, a->call, "retry_ops"); - gpr_free(a->ops); - gpr_free(a); -} - -static void retry_waiting_locked(grpc_exec_ctx *exec_ctx, call_data *calld) { - retry_ops_args *a = gpr_malloc(sizeof(*a)); - a->ops = calld->waiting_ops; - a->nops = calld->waiting_ops_count; - a->call = GET_CALL(calld); - if (a->call == CANCELLED_CALL) { - gpr_free(a); - fail_locked(exec_ctx, calld, GRPC_ERROR_CANCELLED); - return; - } - calld->waiting_ops = NULL; - calld->waiting_ops_count = 0; - calld->waiting_ops_capacity = 0; - GRPC_SUBCHANNEL_CALL_REF(a->call, "retry_ops"); - grpc_exec_ctx_sched(exec_ctx, grpc_closure_create(retry_ops, a), - GRPC_ERROR_NONE, NULL); -} - -static void subchannel_ready(grpc_exec_ctx *exec_ctx, void *arg, - grpc_error *error) { - call_data *calld = arg; - gpr_mu_lock(&calld->mu); - GPR_ASSERT(calld->creation_phase == - GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL); - calld->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING; - if (calld->connected_subchannel == NULL) { - gpr_atm_no_barrier_store(&calld->subchannel_call, 1); - fail_locked(exec_ctx, calld, GRPC_ERROR_CREATE_REFERENCING( - "Failed to create subchannel", &error, 1)); - } else if (1 == gpr_atm_acq_load(&calld->subchannel_call)) { - /* already cancelled before subchannel became ready */ - fail_locked(exec_ctx, calld, - GRPC_ERROR_CREATE_REFERENCING( - "Cancelled before creating subchannel", &error, 1)); - } else { - grpc_subchannel_call *subchannel_call = NULL; - grpc_error *new_error = grpc_connected_subchannel_create_call( - exec_ctx, calld->connected_subchannel, calld->pollent, - &subchannel_call); - if (new_error != GRPC_ERROR_NONE) { - new_error = grpc_error_add_child(new_error, error); - subchannel_call = CANCELLED_CALL; - fail_locked(exec_ctx, calld, new_error); - } - gpr_atm_rel_store(&calld->subchannel_call, - (gpr_atm)(uintptr_t)subchannel_call); - retry_waiting_locked(exec_ctx, calld); - } - gpr_mu_unlock(&calld->mu); - GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call, "pick_subchannel"); -} - -static char *cc_get_peer(grpc_exec_ctx *exec_ctx, grpc_call_element *elem) { - call_data *calld = elem->call_data; - grpc_subchannel_call *subchannel_call = GET_CALL(calld); - if (subchannel_call == NULL || subchannel_call == CANCELLED_CALL) { - return NULL; - } else { - return grpc_subchannel_call_get_peer(exec_ctx, subchannel_call); - } -} - -// The logic here is fairly complicated, due to (a) the fact that we -// need to handle the case where we receive the send op before the -// initial metadata op, and (b) the need for efficiency, especially in -// the streaming case. -// TODO(ctiller): Explain this more thoroughly. -static void cc_start_transport_stream_op(grpc_exec_ctx *exec_ctx, - grpc_call_element *elem, - grpc_transport_stream_op *op) { - call_data *calld = elem->call_data; - GRPC_CALL_LOG_OP(GPR_INFO, elem, op); - /* try to (atomically) get the call */ - grpc_subchannel_call *call = GET_CALL(calld); - GPR_TIMER_BEGIN("cc_start_transport_stream_op", 0); - if (call == CANCELLED_CALL) { - grpc_transport_stream_op_finish_with_failure(exec_ctx, op, - GRPC_ERROR_CANCELLED); - GPR_TIMER_END("cc_start_transport_stream_op", 0); - return; - } - if (call != NULL) { - grpc_subchannel_call_process_op(exec_ctx, call, op); - GPR_TIMER_END("cc_start_transport_stream_op", 0); - return; - } - /* we failed; lock and figure out what to do */ - gpr_mu_lock(&calld->mu); -retry: - /* need to recheck that another thread hasn't set the call */ - call = GET_CALL(calld); - if (call == CANCELLED_CALL) { - gpr_mu_unlock(&calld->mu); - grpc_transport_stream_op_finish_with_failure(exec_ctx, op, - GRPC_ERROR_CANCELLED); - GPR_TIMER_END("cc_start_transport_stream_op", 0); - return; - } - if (call != NULL) { - gpr_mu_unlock(&calld->mu); - grpc_subchannel_call_process_op(exec_ctx, call, op); - GPR_TIMER_END("cc_start_transport_stream_op", 0); - return; - } - /* if this is a cancellation, then we can raise our cancelled flag */ - if (op->cancel_error != GRPC_ERROR_NONE) { - if (!gpr_atm_rel_cas(&calld->subchannel_call, 0, - (gpr_atm)(uintptr_t)CANCELLED_CALL)) { - goto retry; - } else { - switch (calld->creation_phase) { - case GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING: - fail_locked(exec_ctx, calld, GRPC_ERROR_REF(op->cancel_error)); - break; - case GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL: - calld->pick_subchannel(exec_ctx, calld->pick_subchannel_arg, NULL, 0, - &calld->connected_subchannel, NULL); - break; - } - gpr_mu_unlock(&calld->mu); - grpc_transport_stream_op_finish_with_failure(exec_ctx, op, - GRPC_ERROR_CANCELLED); - GPR_TIMER_END("cc_start_transport_stream_op", 0); - return; - } - } - /* if we don't have a subchannel, try to get one */ - if (calld->creation_phase == GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING && - calld->connected_subchannel == NULL && - op->send_initial_metadata != NULL) { - calld->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL; - grpc_closure_init(&calld->next_step, subchannel_ready, calld); - GRPC_CALL_STACK_REF(calld->owning_call, "pick_subchannel"); - if (calld->pick_subchannel( - exec_ctx, calld->pick_subchannel_arg, op->send_initial_metadata, - op->send_initial_metadata_flags, &calld->connected_subchannel, - &calld->next_step)) { - calld->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING; - GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call, "pick_subchannel"); - } - } - /* if we've got a subchannel, then let's ask it to create a call */ - if (calld->creation_phase == GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING && - calld->connected_subchannel != NULL) { - grpc_subchannel_call *subchannel_call = NULL; - grpc_error *error = grpc_connected_subchannel_create_call( - exec_ctx, calld->connected_subchannel, calld->pollent, - &subchannel_call); - if (error != GRPC_ERROR_NONE) { - subchannel_call = CANCELLED_CALL; - fail_locked(exec_ctx, calld, GRPC_ERROR_REF(error)); - grpc_transport_stream_op_finish_with_failure(exec_ctx, op, error); - } - gpr_atm_rel_store(&calld->subchannel_call, - (gpr_atm)(uintptr_t)subchannel_call); - retry_waiting_locked(exec_ctx, calld); - goto retry; - } - /* nothing to be done but wait */ - add_waiting_locked(calld, op); - gpr_mu_unlock(&calld->mu); - GPR_TIMER_END("cc_start_transport_stream_op", 0); -} - static void watch_lb_policy(grpc_exec_ctx *exec_ctx, channel_data *chand, grpc_lb_policy *lb_policy, grpc_connectivity_state current_state); @@ -417,8 +163,8 @@ static void watch_lb_policy(grpc_exec_ctx *exec_ctx, channel_data *chand, &w->on_changed); } -static void cc_on_resolver_result_changed(grpc_exec_ctx *exec_ctx, void *arg, - grpc_error *error) { +static void on_resolver_result_changed(grpc_exec_ctx *exec_ctx, void *arg, + grpc_error *error) { channel_data *chand = arg; grpc_lb_policy *lb_policy = NULL; grpc_lb_policy *old_lb_policy; @@ -570,44 +316,225 @@ static void cc_start_transport_op(grpc_exec_ctx *exec_ctx, gpr_mu_unlock(&chand->mu); } -typedef struct { - grpc_metadata_batch *initial_metadata; - uint32_t initial_metadata_flags; - grpc_connected_subchannel **connected_subchannel; - grpc_closure *on_ready; - grpc_call_element *elem; - grpc_closure closure; -} continue_picking_args; +/* Constructor for channel_data */ +static void cc_init_channel_elem(grpc_exec_ctx *exec_ctx, + grpc_channel_element *elem, + grpc_channel_element_args *args) { + channel_data *chand = elem->channel_data; -static bool cc_pick_subchannel(grpc_exec_ctx *exec_ctx, void *arg, - grpc_metadata_batch *initial_metadata, - uint32_t initial_metadata_flags, - grpc_connected_subchannel **connected_subchannel, - grpc_closure *on_ready); + memset(chand, 0, sizeof(*chand)); -static void continue_picking(grpc_exec_ctx *exec_ctx, void *arg, - grpc_error *error) { - continue_picking_args *cpa = arg; - if (cpa->connected_subchannel == NULL) { + GPR_ASSERT(args->is_last); + GPR_ASSERT(elem->filter == &grpc_client_channel_filter); + + gpr_mu_init(&chand->mu); + grpc_closure_init(&chand->on_resolver_result_changed, + on_resolver_result_changed, chand); + chand->owning_stack = args->channel_stack; + + grpc_connectivity_state_init(&chand->state_tracker, GRPC_CHANNEL_IDLE, + "client_channel"); + chand->interested_parties = grpc_pollset_set_create(); +} + +/* Destructor for channel_data */ +static void cc_destroy_channel_elem(grpc_exec_ctx *exec_ctx, + grpc_channel_element *elem) { + channel_data *chand = elem->channel_data; + + if (chand->resolver != NULL) { + grpc_resolver_shutdown(exec_ctx, chand->resolver); + GRPC_RESOLVER_UNREF(exec_ctx, chand->resolver, "channel"); + } + if (chand->lb_policy != NULL) { + grpc_pollset_set_del_pollset_set(exec_ctx, + chand->lb_policy->interested_parties, + chand->interested_parties); + GRPC_LB_POLICY_UNREF(exec_ctx, chand->lb_policy, "channel"); + } + grpc_connectivity_state_destroy(exec_ctx, &chand->state_tracker); + grpc_pollset_set_destroy(chand->interested_parties); + gpr_mu_destroy(&chand->mu); +} + +/************************************************************************* + * PER-CALL FUNCTIONS + */ + +#define GET_CALL(call_data) \ + ((grpc_subchannel_call *)(gpr_atm_acq_load(&(call_data)->subchannel_call))) + +#define CANCELLED_CALL ((grpc_subchannel_call *)1) + +typedef enum { + GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING, + GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL +} subchannel_creation_phase; + +/** Call data. Holds a pointer to grpc_subchannel_call and the + associated machinery to create such a pointer. + Handles queueing of stream ops until a call object is ready, waiting + for initial metadata before trying to create a call object, + and handling cancellation gracefully. */ +typedef struct client_channel_call_data { + /** either 0 for no call, 1 for cancelled, or a pointer to a + grpc_subchannel_call */ + gpr_atm subchannel_call; + + gpr_mu mu; + + subchannel_creation_phase creation_phase; + grpc_connected_subchannel *connected_subchannel; + grpc_polling_entity *pollent; + + grpc_transport_stream_op *waiting_ops; + size_t waiting_ops_count; + size_t waiting_ops_capacity; + + grpc_closure next_step; + + grpc_call_stack *owning_call; +} call_data; + +static void add_waiting_locked(call_data *calld, grpc_transport_stream_op *op) { + GPR_TIMER_BEGIN("add_waiting_locked", 0); + if (calld->waiting_ops_count == calld->waiting_ops_capacity) { + calld->waiting_ops_capacity = GPR_MAX(3, 2 * calld->waiting_ops_capacity); + calld->waiting_ops = + gpr_realloc(calld->waiting_ops, + calld->waiting_ops_capacity * sizeof(*calld->waiting_ops)); + } + calld->waiting_ops[calld->waiting_ops_count++] = *op; + GPR_TIMER_END("add_waiting_locked", 0); +} + +static void fail_locked(grpc_exec_ctx *exec_ctx, call_data *calld, + grpc_error *error) { + size_t i; + for (i = 0; i < calld->waiting_ops_count; i++) { + grpc_transport_stream_op_finish_with_failure( + exec_ctx, &calld->waiting_ops[i], GRPC_ERROR_REF(error)); + } + calld->waiting_ops_count = 0; + GRPC_ERROR_UNREF(error); +} + +typedef struct { + grpc_transport_stream_op *ops; + size_t nops; + grpc_subchannel_call *call; +} retry_ops_args; + +static void retry_ops(grpc_exec_ctx *exec_ctx, void *args, grpc_error *error) { + retry_ops_args *a = args; + size_t i; + for (i = 0; i < a->nops; i++) { + grpc_subchannel_call_process_op(exec_ctx, a->call, &a->ops[i]); + } + GRPC_SUBCHANNEL_CALL_UNREF(exec_ctx, a->call, "retry_ops"); + gpr_free(a->ops); + gpr_free(a); +} + +static void retry_waiting_locked(grpc_exec_ctx *exec_ctx, call_data *calld) { + retry_ops_args *a = gpr_malloc(sizeof(*a)); + a->ops = calld->waiting_ops; + a->nops = calld->waiting_ops_count; + a->call = GET_CALL(calld); + if (a->call == CANCELLED_CALL) { + gpr_free(a); + fail_locked(exec_ctx, calld, GRPC_ERROR_CANCELLED); + return; + } + calld->waiting_ops = NULL; + calld->waiting_ops_count = 0; + calld->waiting_ops_capacity = 0; + GRPC_SUBCHANNEL_CALL_REF(a->call, "retry_ops"); + grpc_exec_ctx_sched(exec_ctx, grpc_closure_create(retry_ops, a), + GRPC_ERROR_NONE, NULL); +} + +static void subchannel_ready(grpc_exec_ctx *exec_ctx, void *arg, + grpc_error *error) { + call_data *calld = arg; + gpr_mu_lock(&calld->mu); + GPR_ASSERT(calld->creation_phase == + GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL); + calld->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING; + if (calld->connected_subchannel == NULL) { + gpr_atm_no_barrier_store(&calld->subchannel_call, 1); + fail_locked(exec_ctx, calld, GRPC_ERROR_CREATE_REFERENCING( + "Failed to create subchannel", &error, 1)); + } else if (1 == gpr_atm_acq_load(&calld->subchannel_call)) { + /* already cancelled before subchannel became ready */ + fail_locked(exec_ctx, calld, + GRPC_ERROR_CREATE_REFERENCING( + "Cancelled before creating subchannel", &error, 1)); + } else { + grpc_subchannel_call *subchannel_call = NULL; + grpc_error *new_error = grpc_connected_subchannel_create_call( + exec_ctx, calld->connected_subchannel, calld->pollent, + &subchannel_call); + if (new_error != GRPC_ERROR_NONE) { + new_error = grpc_error_add_child(new_error, error); + subchannel_call = CANCELLED_CALL; + fail_locked(exec_ctx, calld, new_error); + } + gpr_atm_rel_store(&calld->subchannel_call, + (gpr_atm)(uintptr_t)subchannel_call); + retry_waiting_locked(exec_ctx, calld); + } + gpr_mu_unlock(&calld->mu); + GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call, "pick_subchannel"); +} + +static char *cc_get_peer(grpc_exec_ctx *exec_ctx, grpc_call_element *elem) { + call_data *calld = elem->call_data; + grpc_subchannel_call *subchannel_call = GET_CALL(calld); + if (subchannel_call == NULL || subchannel_call == CANCELLED_CALL) { + return NULL; + } else { + return grpc_subchannel_call_get_peer(exec_ctx, subchannel_call); + } +} + +typedef struct { + grpc_metadata_batch *initial_metadata; + uint32_t initial_metadata_flags; + grpc_connected_subchannel **connected_subchannel; + grpc_closure *on_ready; + grpc_call_element *elem; + grpc_closure closure; +} continue_picking_args; + +static bool pick_subchannel(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, + grpc_metadata_batch *initial_metadata, + uint32_t initial_metadata_flags, + grpc_connected_subchannel **connected_subchannel, + grpc_closure *on_ready); + +static void continue_picking(grpc_exec_ctx *exec_ctx, void *arg, + grpc_error *error) { + continue_picking_args *cpa = arg; + if (cpa->connected_subchannel == NULL) { /* cancelled, do nothing */ } else if (error != GRPC_ERROR_NONE) { grpc_exec_ctx_sched(exec_ctx, cpa->on_ready, GRPC_ERROR_REF(error), NULL); - } else if (cc_pick_subchannel(exec_ctx, cpa->elem, cpa->initial_metadata, - cpa->initial_metadata_flags, - cpa->connected_subchannel, cpa->on_ready)) { + } else if (pick_subchannel(exec_ctx, cpa->elem, cpa->initial_metadata, + cpa->initial_metadata_flags, + cpa->connected_subchannel, cpa->on_ready)) { grpc_exec_ctx_sched(exec_ctx, cpa->on_ready, GRPC_ERROR_NONE, NULL); } gpr_free(cpa); } -static bool cc_pick_subchannel(grpc_exec_ctx *exec_ctx, void *elemp, - grpc_metadata_batch *initial_metadata, - uint32_t initial_metadata_flags, - grpc_connected_subchannel **connected_subchannel, - grpc_closure *on_ready) { - GPR_TIMER_BEGIN("cc_pick_subchannel", 0); +static bool pick_subchannel(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, + grpc_metadata_batch *initial_metadata, + uint32_t initial_metadata_flags, + grpc_connected_subchannel **connected_subchannel, + grpc_closure *on_ready) { + GPR_TIMER_BEGIN("pick_subchannel", 0); - grpc_call_element *elem = elemp; channel_data *chand = elem->channel_data; call_data *calld = elem->call_data; continue_picking_args *cpa; @@ -631,19 +558,19 @@ static bool cc_pick_subchannel(grpc_exec_ctx *exec_ctx, void *elemp, } } gpr_mu_unlock(&chand->mu); - GPR_TIMER_END("cc_pick_subchannel", 0); + GPR_TIMER_END("pick_subchannel", 0); return true; } if (chand->lb_policy != NULL) { grpc_lb_policy *lb_policy = chand->lb_policy; int r; - GRPC_LB_POLICY_REF(lb_policy, "cc_pick_subchannel"); + GRPC_LB_POLICY_REF(lb_policy, "pick_subchannel"); gpr_mu_unlock(&chand->mu); r = grpc_lb_policy_pick(exec_ctx, lb_policy, calld->pollent, initial_metadata, initial_metadata_flags, connected_subchannel, on_ready); - GRPC_LB_POLICY_UNREF(exec_ctx, lb_policy, "cc_pick_subchannel"); - GPR_TIMER_END("cc_pick_subchannel", 0); + GRPC_LB_POLICY_UNREF(exec_ctx, lb_policy, "pick_subchannel"); + GPR_TIMER_END("pick_subchannel", 0); return r; } if (chand->resolver != NULL && !chand->started_resolving) { @@ -668,18 +595,118 @@ static bool cc_pick_subchannel(grpc_exec_ctx *exec_ctx, void *elemp, } gpr_mu_unlock(&chand->mu); - GPR_TIMER_END("cc_pick_subchannel", 0); + GPR_TIMER_END("pick_subchannel", 0); return false; } +// The logic here is fairly complicated, due to (a) the fact that we +// need to handle the case where we receive the send op before the +// initial metadata op, and (b) the need for efficiency, especially in +// the streaming case. +// TODO(ctiller): Explain this more thoroughly. +static void cc_start_transport_stream_op(grpc_exec_ctx *exec_ctx, + grpc_call_element *elem, + grpc_transport_stream_op *op) { + call_data *calld = elem->call_data; + GRPC_CALL_LOG_OP(GPR_INFO, elem, op); + /* try to (atomically) get the call */ + grpc_subchannel_call *call = GET_CALL(calld); + GPR_TIMER_BEGIN("cc_start_transport_stream_op", 0); + if (call == CANCELLED_CALL) { + grpc_transport_stream_op_finish_with_failure(exec_ctx, op, + GRPC_ERROR_CANCELLED); + GPR_TIMER_END("cc_start_transport_stream_op", 0); + return; + } + if (call != NULL) { + grpc_subchannel_call_process_op(exec_ctx, call, op); + GPR_TIMER_END("cc_start_transport_stream_op", 0); + return; + } + /* we failed; lock and figure out what to do */ + gpr_mu_lock(&calld->mu); +retry: + /* need to recheck that another thread hasn't set the call */ + call = GET_CALL(calld); + if (call == CANCELLED_CALL) { + gpr_mu_unlock(&calld->mu); + grpc_transport_stream_op_finish_with_failure(exec_ctx, op, + GRPC_ERROR_CANCELLED); + GPR_TIMER_END("cc_start_transport_stream_op", 0); + return; + } + if (call != NULL) { + gpr_mu_unlock(&calld->mu); + grpc_subchannel_call_process_op(exec_ctx, call, op); + GPR_TIMER_END("cc_start_transport_stream_op", 0); + return; + } + /* if this is a cancellation, then we can raise our cancelled flag */ + if (op->cancel_error != GRPC_ERROR_NONE) { + if (!gpr_atm_rel_cas(&calld->subchannel_call, 0, + (gpr_atm)(uintptr_t)CANCELLED_CALL)) { + goto retry; + } else { + switch (calld->creation_phase) { + case GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING: + fail_locked(exec_ctx, calld, GRPC_ERROR_REF(op->cancel_error)); + break; + case GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL: + pick_subchannel(exec_ctx, elem, NULL, 0, + &calld->connected_subchannel, NULL); + break; + } + gpr_mu_unlock(&calld->mu); + grpc_transport_stream_op_finish_with_failure(exec_ctx, op, + GRPC_ERROR_CANCELLED); + GPR_TIMER_END("cc_start_transport_stream_op", 0); + return; + } + } + /* if we don't have a subchannel, try to get one */ + if (calld->creation_phase == GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING && + calld->connected_subchannel == NULL && + op->send_initial_metadata != NULL) { + calld->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL; + grpc_closure_init(&calld->next_step, subchannel_ready, calld); + GRPC_CALL_STACK_REF(calld->owning_call, "pick_subchannel"); + if (pick_subchannel( + exec_ctx, elem, op->send_initial_metadata, + op->send_initial_metadata_flags, &calld->connected_subchannel, + &calld->next_step)) { + calld->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING; + GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call, "pick_subchannel"); + } + } + /* if we've got a subchannel, then let's ask it to create a call */ + if (calld->creation_phase == GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING && + calld->connected_subchannel != NULL) { + grpc_subchannel_call *subchannel_call = NULL; + grpc_error *error = grpc_connected_subchannel_create_call( + exec_ctx, calld->connected_subchannel, calld->pollent, + &subchannel_call); + if (error != GRPC_ERROR_NONE) { + subchannel_call = CANCELLED_CALL; + fail_locked(exec_ctx, calld, GRPC_ERROR_REF(error)); + grpc_transport_stream_op_finish_with_failure(exec_ctx, op, error); + } + gpr_atm_rel_store(&calld->subchannel_call, + (gpr_atm)(uintptr_t)subchannel_call); + retry_waiting_locked(exec_ctx, calld); + goto retry; + } + /* nothing to be done but wait */ + add_waiting_locked(calld, op); + gpr_mu_unlock(&calld->mu); + GPR_TIMER_END("cc_start_transport_stream_op", 0); +} + /* Constructor for call_data */ -static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx, - grpc_call_element *elem, - grpc_call_element_args *args) { +static grpc_error *cc_init_call_elem(grpc_exec_ctx *exec_ctx, + grpc_call_element *elem, + grpc_call_element_args *args) { call_data *calld = elem->call_data; gpr_atm_rel_store(&calld->subchannel_call, 0); - calld->pick_subchannel = cc_pick_subchannel; - calld->pick_subchannel_arg = elem; gpr_mu_init(&calld->mu); calld->connected_subchannel = NULL; calld->waiting_ops = NULL; @@ -692,9 +719,10 @@ static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx, } /* Destructor for call_data */ -static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, - const grpc_call_final_info *final_info, - void *and_free_memory) { +static void cc_destroy_call_elem(grpc_exec_ctx *exec_ctx, + grpc_call_element *elem, + const grpc_call_final_info *final_info, + void *and_free_memory) { call_data *calld = elem->call_data; grpc_subchannel_call *call = GET_CALL(calld); if (call != NULL && call != CANCELLED_CALL) { @@ -707,47 +735,6 @@ static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, gpr_free(and_free_memory); } -/* Constructor for channel_data */ -static void init_channel_elem(grpc_exec_ctx *exec_ctx, - grpc_channel_element *elem, - grpc_channel_element_args *args) { - channel_data *chand = elem->channel_data; - - memset(chand, 0, sizeof(*chand)); - - GPR_ASSERT(args->is_last); - GPR_ASSERT(elem->filter == &grpc_client_channel_filter); - - gpr_mu_init(&chand->mu); - grpc_closure_init(&chand->on_resolver_result_changed, - cc_on_resolver_result_changed, chand); - chand->owning_stack = args->channel_stack; - - grpc_connectivity_state_init(&chand->state_tracker, GRPC_CHANNEL_IDLE, - "client_channel"); - chand->interested_parties = grpc_pollset_set_create(); -} - -/* Destructor for channel_data */ -static void destroy_channel_elem(grpc_exec_ctx *exec_ctx, - grpc_channel_element *elem) { - channel_data *chand = elem->channel_data; - - if (chand->resolver != NULL) { - grpc_resolver_shutdown(exec_ctx, chand->resolver); - GRPC_RESOLVER_UNREF(exec_ctx, chand->resolver, "channel"); - } - if (chand->lb_policy != NULL) { - grpc_pollset_set_del_pollset_set(exec_ctx, - chand->lb_policy->interested_parties, - chand->interested_parties); - GRPC_LB_POLICY_UNREF(exec_ctx, chand->lb_policy, "channel"); - } - grpc_connectivity_state_destroy(exec_ctx, &chand->state_tracker); - grpc_pollset_set_destroy(chand->interested_parties); - gpr_mu_destroy(&chand->mu); -} - static void cc_set_pollset_or_pollset_set(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, grpc_polling_entity *pollent) { @@ -755,16 +742,20 @@ static void cc_set_pollset_or_pollset_set(grpc_exec_ctx *exec_ctx, calld->pollent = pollent; } +/************************************************************************* + * EXPORTED SYMBOLS + */ + const grpc_channel_filter grpc_client_channel_filter = { cc_start_transport_stream_op, cc_start_transport_op, sizeof(call_data), - init_call_elem, + cc_init_call_elem, cc_set_pollset_or_pollset_set, - destroy_call_elem, + cc_destroy_call_elem, sizeof(channel_data), - init_channel_elem, - destroy_channel_elem, + cc_init_channel_elem, + cc_destroy_channel_elem, cc_get_peer, "client-channel", }; From d4c0f550b0662d2ec2bc0deb7985aa1357c195a7 Mon Sep 17 00:00:00 2001 From: "Mark D. Roth" Date: Thu, 1 Sep 2016 09:25:32 -0700 Subject: [PATCH 7/7] clang-format --- src/core/ext/client_config/client_channel.c | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) diff --git a/src/core/ext/client_config/client_channel.c b/src/core/ext/client_config/client_channel.c index fff80abb802..61e012578e0 100644 --- a/src/core/ext/client_config/client_channel.c +++ b/src/core/ext/client_config/client_channel.c @@ -652,8 +652,8 @@ retry: fail_locked(exec_ctx, calld, GRPC_ERROR_REF(op->cancel_error)); break; case GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL: - pick_subchannel(exec_ctx, elem, NULL, 0, - &calld->connected_subchannel, NULL); + pick_subchannel(exec_ctx, elem, NULL, 0, &calld->connected_subchannel, + NULL); break; } gpr_mu_unlock(&calld->mu); @@ -670,10 +670,9 @@ retry: calld->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL; grpc_closure_init(&calld->next_step, subchannel_ready, calld); GRPC_CALL_STACK_REF(calld->owning_call, "pick_subchannel"); - if (pick_subchannel( - exec_ctx, elem, op->send_initial_metadata, - op->send_initial_metadata_flags, &calld->connected_subchannel, - &calld->next_step)) { + if (pick_subchannel(exec_ctx, elem, op->send_initial_metadata, + op->send_initial_metadata_flags, + &calld->connected_subchannel, &calld->next_step)) { calld->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING; GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call, "pick_subchannel"); }