diff options
Diffstat (limited to 'src/core/ext')
55 files changed, 1698 insertions, 1619 deletions
diff --git a/src/core/ext/filters/client_channel/client_channel.cc b/src/core/ext/filters/client_channel/client_channel.cc index 8e9ee889e1..70aac47231 100644 --- a/src/core/ext/filters/client_channel/client_channel.cc +++ b/src/core/ext/filters/client_channel/client_channel.cc @@ -34,10 +34,11 @@ #include "src/core/ext/filters/client_channel/backup_poller.h" #include "src/core/ext/filters/client_channel/http_connect_handshaker.h" #include "src/core/ext/filters/client_channel/lb_policy_registry.h" -#include "src/core/ext/filters/client_channel/method_params.h" #include "src/core/ext/filters/client_channel/proxy_mapper_registry.h" #include "src/core/ext/filters/client_channel/resolver_registry.h" +#include "src/core/ext/filters/client_channel/resolver_result_parsing.h" #include "src/core/ext/filters/client_channel/retry_throttle.h" +#include "src/core/ext/filters/client_channel/server_address.h" #include "src/core/ext/filters/client_channel/subchannel.h" #include "src/core/ext/filters/deadline/deadline_filter.h" #include "src/core/lib/backoff/backoff.h" @@ -62,7 +63,10 @@ #include "src/core/lib/transport/static_metadata.h" #include "src/core/lib/transport/status_metadata.h" +using grpc_core::ServerAddressList; using grpc_core::internal::ClientChannelMethodParams; +using grpc_core::internal::ClientChannelMethodParamsTable; +using grpc_core::internal::ProcessedResolverResult; using grpc_core::internal::ServerRetryThrottleData; /* Client channel implementation */ @@ -83,10 +87,6 @@ grpc_core::TraceFlag grpc_client_channel_trace(false, "client_channel"); struct external_connectivity_watcher; -typedef grpc_core::SliceHashTable< - grpc_core::RefCountedPtr<ClientChannelMethodParams>> - MethodParamsTable; - typedef struct client_channel_channel_data { grpc_core::OrphanablePtr<grpc_core::Resolver> resolver; bool started_resolving; @@ -102,7 +102,7 @@ typedef struct client_channel_channel_data { /** retry throttle data */ grpc_core::RefCountedPtr<ServerRetryThrottleData> retry_throttle_data; /** maps method names to method_parameters structs */ - grpc_core::RefCountedPtr<MethodParamsTable> method_params_table; + grpc_core::RefCountedPtr<ClientChannelMethodParamsTable> method_params_table; /** incoming resolver result - set by resolver.next() */ grpc_channel_args* resolver_result; /** a list of closures that are all waiting for resolver result to come in */ @@ -251,66 +251,6 @@ static void start_resolving_locked(channel_data* chand) { &chand->on_resolver_result_changed); } -typedef struct { - char* server_name; - grpc_core::RefCountedPtr<ServerRetryThrottleData> retry_throttle_data; -} service_config_parsing_state; - -static void parse_retry_throttle_params( - const grpc_json* field, service_config_parsing_state* parsing_state) { - if (strcmp(field->key, "retryThrottling") == 0) { - if (parsing_state->retry_throttle_data != nullptr) return; // Duplicate. - if (field->type != GRPC_JSON_OBJECT) return; - int max_milli_tokens = 0; - int milli_token_ratio = 0; - for (grpc_json* sub_field = field->child; sub_field != nullptr; - sub_field = sub_field->next) { - if (sub_field->key == nullptr) return; - if (strcmp(sub_field->key, "maxTokens") == 0) { - if (max_milli_tokens != 0) return; // Duplicate. - if (sub_field->type != GRPC_JSON_NUMBER) return; - max_milli_tokens = gpr_parse_nonnegative_int(sub_field->value); - if (max_milli_tokens == -1) return; - max_milli_tokens *= 1000; - } else if (strcmp(sub_field->key, "tokenRatio") == 0) { - if (milli_token_ratio != 0) return; // Duplicate. - if (sub_field->type != GRPC_JSON_NUMBER) return; - // We support up to 3 decimal digits. - size_t whole_len = strlen(sub_field->value); - uint32_t multiplier = 1; - uint32_t decimal_value = 0; - const char* decimal_point = strchr(sub_field->value, '.'); - if (decimal_point != nullptr) { - whole_len = static_cast<size_t>(decimal_point - sub_field->value); - multiplier = 1000; - size_t decimal_len = strlen(decimal_point + 1); - if (decimal_len > 3) decimal_len = 3; - if (!gpr_parse_bytes_to_uint32(decimal_point + 1, decimal_len, - &decimal_value)) { - return; - } - uint32_t decimal_multiplier = 1; - for (size_t i = 0; i < (3 - decimal_len); ++i) { - decimal_multiplier *= 10; - } - decimal_value *= decimal_multiplier; - } - uint32_t whole_value; - if (!gpr_parse_bytes_to_uint32(sub_field->value, whole_len, - &whole_value)) { - return; - } - milli_token_ratio = - static_cast<int>((whole_value * multiplier) + decimal_value); - if (milli_token_ratio <= 0) return; - } - } - parsing_state->retry_throttle_data = - grpc_core::internal::ServerRetryThrottleMap::GetDataForServer( - parsing_state->server_name, max_milli_tokens, milli_token_ratio); - } -} - // Invoked from the resolver NextLocked() callback when the resolver // is shutting down. static void on_resolver_shutdown_locked(channel_data* chand, @@ -352,37 +292,6 @@ static void on_resolver_shutdown_locked(channel_data* chand, GRPC_ERROR_UNREF(error); } -// Returns the LB policy name from the resolver result. -static grpc_core::UniquePtr<char> -get_lb_policy_name_from_resolver_result_locked(channel_data* chand) { - // Find LB policy name in channel args. - const grpc_arg* channel_arg = - grpc_channel_args_find(chand->resolver_result, GRPC_ARG_LB_POLICY_NAME); - const char* lb_policy_name = grpc_channel_arg_get_string(channel_arg); - // Special case: If at least one balancer address is present, we use - // the grpclb policy, regardless of what the resolver actually specified. - channel_arg = - grpc_channel_args_find(chand->resolver_result, GRPC_ARG_LB_ADDRESSES); - if (channel_arg != nullptr && channel_arg->type == GRPC_ARG_POINTER) { - grpc_lb_addresses* addresses = - static_cast<grpc_lb_addresses*>(channel_arg->value.pointer.p); - if (grpc_lb_addresses_contains_balancer_address(*addresses)) { - if (lb_policy_name != nullptr && - gpr_stricmp(lb_policy_name, "grpclb") != 0) { - gpr_log(GPR_INFO, - "resolver requested LB policy %s but provided at least one " - "balancer address -- forcing use of grpclb LB policy", - lb_policy_name); - } - lb_policy_name = "grpclb"; - } - } - // Use pick_first if nothing was specified and we didn't select grpclb - // above. - if (lb_policy_name == nullptr) lb_policy_name = "pick_first"; - return grpc_core::UniquePtr<char>(gpr_strdup(lb_policy_name)); -} - static void request_reresolution_locked(void* arg, grpc_error* error) { reresolution_request_args* args = static_cast<reresolution_request_args*>(arg); @@ -410,13 +319,14 @@ using TraceStringVector = grpc_core::InlinedVector<char*, 3>; // *connectivity_error to its initial connectivity state; otherwise, // leaves them unchanged. static void create_new_lb_policy_locked( - channel_data* chand, char* lb_policy_name, + channel_data* chand, char* lb_policy_name, grpc_json* lb_config, grpc_connectivity_state* connectivity_state, grpc_error** connectivity_error, TraceStringVector* trace_strings) { grpc_core::LoadBalancingPolicy::Args lb_policy_args; lb_policy_args.combiner = chand->combiner; lb_policy_args.client_channel_factory = chand->client_channel_factory; lb_policy_args.args = chand->resolver_result; + lb_policy_args.lb_config = lb_config; grpc_core::OrphanablePtr<grpc_core::LoadBalancingPolicy> new_lb_policy = grpc_core::LoadBalancingPolicyRegistry::CreateLoadBalancingPolicy( lb_policy_name, lb_policy_args); @@ -473,56 +383,12 @@ static void create_new_lb_policy_locked( } } -// Returns the service config (as a JSON string) from the resolver result. -// Also updates state in chand. -static grpc_core::UniquePtr<char> -get_service_config_from_resolver_result_locked(channel_data* chand) { - const grpc_arg* channel_arg = - grpc_channel_args_find(chand->resolver_result, GRPC_ARG_SERVICE_CONFIG); - const char* service_config_json = grpc_channel_arg_get_string(channel_arg); - if (service_config_json != nullptr) { - if (grpc_client_channel_trace.enabled()) { - gpr_log(GPR_INFO, "chand=%p: resolver returned service config: \"%s\"", - chand, service_config_json); - } - grpc_core::UniquePtr<grpc_core::ServiceConfig> service_config = - grpc_core::ServiceConfig::Create(service_config_json); - if (service_config != nullptr) { - if (chand->enable_retries) { - channel_arg = - grpc_channel_args_find(chand->resolver_result, GRPC_ARG_SERVER_URI); - const char* server_uri = grpc_channel_arg_get_string(channel_arg); - GPR_ASSERT(server_uri != nullptr); - grpc_uri* uri = grpc_uri_parse(server_uri, true); - GPR_ASSERT(uri->path[0] != '\0'); - service_config_parsing_state parsing_state; - parsing_state.server_name = - uri->path[0] == '/' ? uri->path + 1 : uri->path; - service_config->ParseGlobalParams(parse_retry_throttle_params, - &parsing_state); - grpc_uri_destroy(uri); - chand->retry_throttle_data = - std::move(parsing_state.retry_throttle_data); - } - chand->method_params_table = service_config->CreateMethodConfigTable( - ClientChannelMethodParams::CreateFromJson); - } - } - return grpc_core::UniquePtr<char>(gpr_strdup(service_config_json)); -} - static void maybe_add_trace_message_for_address_changes_locked( channel_data* chand, TraceStringVector* trace_strings) { - int resolution_contains_addresses = false; - const grpc_arg* channel_arg = - grpc_channel_args_find(chand->resolver_result, GRPC_ARG_LB_ADDRESSES); - if (channel_arg != nullptr && channel_arg->type == GRPC_ARG_POINTER) { - grpc_lb_addresses* addresses = - static_cast<grpc_lb_addresses*>(channel_arg->value.pointer.p); - if (addresses->num_addresses > 0) { - resolution_contains_addresses = true; - } - } + const ServerAddressList* addresses = + grpc_core::FindServerAddressListChannelArg(chand->resolver_result); + const bool resolution_contains_addresses = + addresses != nullptr && addresses->size() > 0; if (!resolution_contains_addresses && chand->previous_resolution_contained_addresses) { trace_strings->push_back(gpr_strdup("Address list became empty")); @@ -597,36 +463,47 @@ static void on_resolver_result_changed_locked(void* arg, grpc_error* error) { if (grpc_client_channel_trace.enabled()) { gpr_log(GPR_INFO, "chand=%p: resolver transient failure", chand); } + // Don't override connectivity state if we already have an LB policy. + if (chand->lb_policy != nullptr) set_connectivity_state = false; } else { + // Parse the resolver result. + ProcessedResolverResult resolver_result(chand->resolver_result, + chand->enable_retries); + chand->retry_throttle_data = resolver_result.retry_throttle_data(); + chand->method_params_table = resolver_result.method_params_table(); + grpc_core::UniquePtr<char> service_config_json = + resolver_result.service_config_json(); + if (service_config_json != nullptr && grpc_client_channel_trace.enabled()) { + gpr_log(GPR_INFO, "chand=%p: resolver returned service config: \"%s\"", + chand, service_config_json.get()); + } grpc_core::UniquePtr<char> lb_policy_name = - get_lb_policy_name_from_resolver_result_locked(chand); + resolver_result.lb_policy_name(); + grpc_json* lb_policy_config = resolver_result.lb_policy_config(); // Check to see if we're already using the right LB policy. // Note: It's safe to use chand->info_lb_policy_name here without // taking a lock on chand->info_mu, because this function is the // only thing that modifies its value, and it can only be invoked // once at any given time. - bool lb_policy_name_changed = chand->info_lb_policy_name == nullptr || - gpr_stricmp(chand->info_lb_policy_name.get(), - lb_policy_name.get()) != 0; + bool lb_policy_name_changed = + chand->info_lb_policy_name == nullptr || + strcmp(chand->info_lb_policy_name.get(), lb_policy_name.get()) != 0; if (chand->lb_policy != nullptr && !lb_policy_name_changed) { // Continue using the same LB policy. Update with new addresses. if (grpc_client_channel_trace.enabled()) { gpr_log(GPR_INFO, "chand=%p: updating existing LB policy \"%s\" (%p)", chand, lb_policy_name.get(), chand->lb_policy.get()); } - chand->lb_policy->UpdateLocked(*chand->resolver_result); + chand->lb_policy->UpdateLocked(*chand->resolver_result, lb_policy_config); // No need to set the channel's connectivity state; the existing // watch on the LB policy will take care of that. set_connectivity_state = false; } else { // Instantiate new LB policy. - create_new_lb_policy_locked(chand, lb_policy_name.get(), + create_new_lb_policy_locked(chand, lb_policy_name.get(), lb_policy_config, &connectivity_state, &connectivity_error, &trace_strings); } - // Find service config. - grpc_core::UniquePtr<char> service_config_json = - get_service_config_from_resolver_result_locked(chand); // Note: It's safe to use chand->info_service_config_json here without // taking a lock on chand->info_mu, because this function is the // only thing that modifies its value, and it can only be invoked @@ -689,12 +566,6 @@ static void start_transport_op_locked(void* arg, grpc_error* error_ignored) { } else { grpc_error* error = GRPC_ERROR_NONE; grpc_core::LoadBalancingPolicy::PickState pick_state; - pick_state.initial_metadata = nullptr; - pick_state.initial_metadata_flags = 0; - pick_state.on_complete = nullptr; - memset(&pick_state.subchannel_call_context, 0, - sizeof(pick_state.subchannel_call_context)); - pick_state.user_data = nullptr; // Pick must return synchronously, because pick_state.on_complete is null. GPR_ASSERT(chand->lb_policy->PickLocked(&pick_state, &error)); if (pick_state.connected_subchannel != nullptr) { diff --git a/src/core/ext/filters/client_channel/health/health_check_client.cc b/src/core/ext/filters/client_channel/health/health_check_client.cc index 587919596f..2232c57120 100644 --- a/src/core/ext/filters/client_channel/health/health_check_client.cc +++ b/src/core/ext/filters/client_channel/health/health_check_client.cc @@ -51,8 +51,7 @@ HealthCheckClient::HealthCheckClient( RefCountedPtr<ConnectedSubchannel> connected_subchannel, grpc_pollset_set* interested_parties, grpc_core::RefCountedPtr<grpc_core::channelz::SubchannelNode> channelz_node) - : InternallyRefCountedWithTracing<HealthCheckClient>( - &grpc_health_check_client_trace), + : InternallyRefCounted<HealthCheckClient>(&grpc_health_check_client_trace), service_name_(service_name), connected_subchannel_(std::move(connected_subchannel)), interested_parties_(interested_parties), @@ -281,8 +280,7 @@ bool DecodeResponse(grpc_slice_buffer* slice_buffer, grpc_error** error) { HealthCheckClient::CallState::CallState( RefCountedPtr<HealthCheckClient> health_check_client, grpc_pollset_set* interested_parties) - : InternallyRefCountedWithTracing<CallState>( - &grpc_health_check_client_trace), + : InternallyRefCounted<CallState>(&grpc_health_check_client_trace), health_check_client_(std::move(health_check_client)), pollent_(grpc_polling_entity_create_from_pollset_set(interested_parties)), arena_(gpr_arena_create(health_check_client_->connected_subchannel_ diff --git a/src/core/ext/filters/client_channel/health/health_check_client.h b/src/core/ext/filters/client_channel/health/health_check_client.h index f6babef7d6..2369b73fea 100644 --- a/src/core/ext/filters/client_channel/health/health_check_client.h +++ b/src/core/ext/filters/client_channel/health/health_check_client.h @@ -41,8 +41,7 @@ namespace grpc_core { -class HealthCheckClient - : public InternallyRefCountedWithTracing<HealthCheckClient> { +class HealthCheckClient : public InternallyRefCounted<HealthCheckClient> { public: HealthCheckClient(const char* service_name, RefCountedPtr<ConnectedSubchannel> connected_subchannel, @@ -61,7 +60,7 @@ class HealthCheckClient private: // Contains a call to the backend and all the data related to the call. - class CallState : public InternallyRefCountedWithTracing<CallState> { + class CallState : public InternallyRefCounted<CallState> { public: CallState(RefCountedPtr<HealthCheckClient> health_check_client, grpc_pollset_set* interested_parties_); diff --git a/src/core/ext/filters/client_channel/lb_policy.cc b/src/core/ext/filters/client_channel/lb_policy.cc index e065f45639..b4e803689e 100644 --- a/src/core/ext/filters/client_channel/lb_policy.cc +++ b/src/core/ext/filters/client_channel/lb_policy.cc @@ -27,7 +27,7 @@ grpc_core::DebugOnlyTraceFlag grpc_trace_lb_policy_refcount( namespace grpc_core { LoadBalancingPolicy::LoadBalancingPolicy(const Args& args) - : InternallyRefCountedWithTracing(&grpc_trace_lb_policy_refcount), + : InternallyRefCounted(&grpc_trace_lb_policy_refcount), combiner_(GRPC_COMBINER_REF(args.combiner, "lb_policy")), client_channel_factory_(args.client_channel_factory), interested_parties_(grpc_pollset_set_create()), diff --git a/src/core/ext/filters/client_channel/lb_policy.h b/src/core/ext/filters/client_channel/lb_policy.h index b0040457a6..6b76fe5d5d 100644 --- a/src/core/ext/filters/client_channel/lb_policy.h +++ b/src/core/ext/filters/client_channel/lb_policy.h @@ -42,8 +42,7 @@ namespace grpc_core { /// /// Any I/O done by the LB policy should be done under the pollset_set /// returned by \a interested_parties(). -class LoadBalancingPolicy - : public InternallyRefCountedWithTracing<LoadBalancingPolicy> { +class LoadBalancingPolicy : public InternallyRefCounted<LoadBalancingPolicy> { public: struct Args { /// The combiner under which all LB policy calls will be run. @@ -56,8 +55,10 @@ class LoadBalancingPolicy grpc_client_channel_factory* client_channel_factory = nullptr; /// Channel args from the resolver. /// Note that the LB policy gets the set of addresses from the - /// GRPC_ARG_LB_ADDRESSES channel arg. + /// GRPC_ARG_SERVER_ADDRESS_LIST channel arg. grpc_channel_args* args = nullptr; + /// Load balancing config from the resolver. + grpc_json* lb_config = nullptr; }; /// State used for an LB pick. @@ -79,11 +80,6 @@ class LoadBalancingPolicy /// Will be populated with context to pass to the subchannel call, if /// needed. grpc_call_context_element subchannel_call_context[GRPC_CONTEXT_COUNT] = {}; - /// Upon success, \a *user_data will be set to whatever opaque information - /// may need to be propagated from the LB policy, or nullptr if not needed. - // TODO(roth): As part of revamping our metadata APIs, try to find a - // way to clean this up and C++-ify it. - void** user_data = nullptr; /// Next pointer. For internal use by LB policy. PickState* next = nullptr; }; @@ -92,10 +88,11 @@ class LoadBalancingPolicy LoadBalancingPolicy(const LoadBalancingPolicy&) = delete; LoadBalancingPolicy& operator=(const LoadBalancingPolicy&) = delete; - /// Updates the policy with a new set of \a args from the resolver. - /// Note that the LB policy gets the set of addresses from the - /// GRPC_ARG_LB_ADDRESSES channel arg. - virtual void UpdateLocked(const grpc_channel_args& args) GRPC_ABSTRACT; + /// Updates the policy with a new set of \a args and a new \a lb_config from + /// the resolver. Note that the LB policy gets the set of addresses from the + /// GRPC_ARG_SERVER_ADDRESS_LIST channel arg. + virtual void UpdateLocked(const grpc_channel_args& args, + grpc_json* lb_config) GRPC_ABSTRACT; /// Finds an appropriate subchannel for a call, based on data in \a pick. /// \a pick must remain alive until the pick is complete. diff --git a/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.cc b/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.cc index dbb90b438c..a9a5965ed1 100644 --- a/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.cc +++ b/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.cc @@ -84,6 +84,7 @@ #include "src/core/ext/filters/client_channel/lb_policy_registry.h" #include "src/core/ext/filters/client_channel/parse_address.h" #include "src/core/ext/filters/client_channel/resolver/fake/fake_resolver.h" +#include "src/core/ext/filters/client_channel/server_address.h" #include "src/core/ext/filters/client_channel/subchannel_index.h" #include "src/core/lib/backoff/backoff.h" #include "src/core/lib/channel/channel_args.h" @@ -113,6 +114,8 @@ #define GRPC_GRPCLB_RECONNECT_JITTER 0.2 #define GRPC_GRPCLB_DEFAULT_FALLBACK_TIMEOUT_MS 10000 +#define GRPC_ARG_GRPCLB_ADDRESS_LB_TOKEN "grpc.grpclb_address_lb_token" + namespace grpc_core { TraceFlag grpc_lb_glb_trace(false, "glb"); @@ -121,9 +124,10 @@ namespace { class GrpcLb : public LoadBalancingPolicy { public: - GrpcLb(const grpc_lb_addresses* addresses, const Args& args); + explicit GrpcLb(const Args& args); - void UpdateLocked(const grpc_channel_args& args) override; + void UpdateLocked(const grpc_channel_args& args, + grpc_json* lb_config) override; bool PickLocked(PickState* pick, grpc_error** error) override; void CancelPickLocked(PickState* pick, grpc_error* error) override; void CancelMatchingPicksLocked(uint32_t initial_metadata_flags_mask, @@ -160,9 +164,6 @@ class GrpcLb : public LoadBalancingPolicy { // Our on_complete closure and the original one. grpc_closure on_complete; grpc_closure* original_on_complete; - // The LB token associated with the pick. This is set via user_data in - // the pick. - grpc_mdelem lb_token; // Stats for client-side load reporting. RefCountedPtr<GrpcLbClientStats> client_stats; // Next pending pick. @@ -170,8 +171,7 @@ class GrpcLb : public LoadBalancingPolicy { }; /// Contains a call to the LB server and all the data related to the call. - class BalancerCallState - : public InternallyRefCountedWithTracing<BalancerCallState> { + class BalancerCallState : public InternallyRefCounted<BalancerCallState> { public: explicit BalancerCallState( RefCountedPtr<LoadBalancingPolicy> parent_grpclb_policy); @@ -329,7 +329,7 @@ class GrpcLb : public LoadBalancingPolicy { // 0 means not using fallback. int lb_fallback_timeout_ms_ = 0; // The backend addresses from the resolver. - grpc_lb_addresses* fallback_backend_addresses_ = nullptr; + UniquePtr<ServerAddressList> fallback_backend_addresses_; // Fallback timer. bool fallback_timer_callback_pending_ = false; grpc_timer lb_fallback_timer_; @@ -349,7 +349,7 @@ class GrpcLb : public LoadBalancingPolicy { // serverlist parsing code // -// vtable for LB tokens in grpc_lb_addresses +// vtable for LB token channel arg. void* lb_token_copy(void* token) { return token == nullptr ? nullptr @@ -361,38 +361,11 @@ void lb_token_destroy(void* token) { } } int lb_token_cmp(void* token1, void* token2) { - if (token1 > token2) return 1; - if (token1 < token2) return -1; - return 0; + return GPR_ICMP(token1, token2); } -const grpc_lb_user_data_vtable lb_token_vtable = { +const grpc_arg_pointer_vtable lb_token_arg_vtable = { lb_token_copy, lb_token_destroy, lb_token_cmp}; -// Returns the backend addresses extracted from the given addresses. -grpc_lb_addresses* ExtractBackendAddresses(const grpc_lb_addresses* addresses) { - // First pass: count the number of backend addresses. - size_t num_backends = 0; - for (size_t i = 0; i < addresses->num_addresses; ++i) { - if (!addresses->addresses[i].is_balancer) { - ++num_backends; - } - } - // Second pass: actually populate the addresses and (empty) LB tokens. - grpc_lb_addresses* backend_addresses = - grpc_lb_addresses_create(num_backends, &lb_token_vtable); - size_t num_copied = 0; - for (size_t i = 0; i < addresses->num_addresses; ++i) { - if (addresses->addresses[i].is_balancer) continue; - const grpc_resolved_address* addr = &addresses->addresses[i].address; - grpc_lb_addresses_set_address(backend_addresses, num_copied, &addr->addr, - addr->len, false /* is_balancer */, - nullptr /* balancer_name */, - (void*)GRPC_MDELEM_LB_TOKEN_EMPTY.payload); - ++num_copied; - } - return backend_addresses; -} - bool IsServerValid(const grpc_grpclb_server* server, size_t idx, bool log) { if (server->drop) return false; const grpc_grpclb_ip_address* ip = &server->ip_address; @@ -440,30 +413,16 @@ void ParseServer(const grpc_grpclb_server* server, } // Returns addresses extracted from \a serverlist. -grpc_lb_addresses* ProcessServerlist(const grpc_grpclb_serverlist* serverlist) { - size_t num_valid = 0; - /* first pass: count how many are valid in order to allocate the necessary - * memory in a single block */ +ServerAddressList ProcessServerlist(const grpc_grpclb_serverlist* serverlist) { + ServerAddressList addresses; for (size_t i = 0; i < serverlist->num_servers; ++i) { - if (IsServerValid(serverlist->servers[i], i, true)) ++num_valid; - } - grpc_lb_addresses* lb_addresses = - grpc_lb_addresses_create(num_valid, &lb_token_vtable); - /* second pass: actually populate the addresses and LB tokens (aka user data - * to the outside world) to be read by the RR policy during its creation. - * Given that the validity tests are very cheap, they are performed again - * instead of marking the valid ones during the first pass, as this would - * incurr in an allocation due to the arbitrary number of server */ - size_t addr_idx = 0; - for (size_t sl_idx = 0; sl_idx < serverlist->num_servers; ++sl_idx) { - const grpc_grpclb_server* server = serverlist->servers[sl_idx]; - if (!IsServerValid(serverlist->servers[sl_idx], sl_idx, false)) continue; - GPR_ASSERT(addr_idx < num_valid); - /* address processing */ + const grpc_grpclb_server* server = serverlist->servers[i]; + if (!IsServerValid(serverlist->servers[i], i, false)) continue; + // Address processing. grpc_resolved_address addr; ParseServer(server, &addr); - /* lb token processing */ - void* user_data; + // LB token processing. + void* lb_token; if (server->has_load_balance_token) { const size_t lb_token_max_length = GPR_ARRAY_SIZE(server->load_balance_token); @@ -471,7 +430,7 @@ grpc_lb_addresses* ProcessServerlist(const grpc_grpclb_serverlist* serverlist) { strnlen(server->load_balance_token, lb_token_max_length); grpc_slice lb_token_mdstr = grpc_slice_from_copied_buffer( server->load_balance_token, lb_token_length); - user_data = + lb_token = (void*)grpc_mdelem_from_slices(GRPC_MDSTR_LB_TOKEN, lb_token_mdstr) .payload; } else { @@ -481,15 +440,16 @@ grpc_lb_addresses* ProcessServerlist(const grpc_grpclb_serverlist* serverlist) { "be used instead", uri); gpr_free(uri); - user_data = (void*)GRPC_MDELEM_LB_TOKEN_EMPTY.payload; + lb_token = (void*)GRPC_MDELEM_LB_TOKEN_EMPTY.payload; } - grpc_lb_addresses_set_address(lb_addresses, addr_idx, &addr.addr, addr.len, - false /* is_balancer */, - nullptr /* balancer_name */, user_data); - ++addr_idx; - } - GPR_ASSERT(addr_idx == num_valid); - return lb_addresses; + // Add address. + grpc_arg arg = grpc_channel_arg_pointer_create( + const_cast<char*>(GRPC_ARG_GRPCLB_ADDRESS_LB_TOKEN), lb_token, + &lb_token_arg_vtable); + grpc_channel_args* args = grpc_channel_args_copy_and_add(nullptr, &arg, 1); + addresses.emplace_back(addr, args); + } + return addresses; } // @@ -498,7 +458,7 @@ grpc_lb_addresses* ProcessServerlist(const grpc_grpclb_serverlist* serverlist) { GrpcLb::BalancerCallState::BalancerCallState( RefCountedPtr<LoadBalancingPolicy> parent_grpclb_policy) - : InternallyRefCountedWithTracing<BalancerCallState>(&grpc_lb_glb_trace), + : InternallyRefCounted<BalancerCallState>(&grpc_lb_glb_trace), grpclb_policy_(std::move(parent_grpclb_policy)) { GPR_ASSERT(grpclb_policy_ != nullptr); GPR_ASSERT(!grpclb_policy()->shutting_down_); @@ -748,7 +708,7 @@ void GrpcLb::BalancerCallState::OnBalancerMessageReceivedLocked( void* arg, grpc_error* error) { BalancerCallState* lb_calld = static_cast<BalancerCallState*>(arg); GrpcLb* grpclb_policy = lb_calld->grpclb_policy(); - // Empty payload means the LB call was cancelled. + // Null payload means the LB call was cancelled. if (lb_calld != grpclb_policy->lb_calld_.get() || lb_calld->recv_message_payload_ == nullptr) { lb_calld->Unref(DEBUG_LOCATION, "on_message_received"); @@ -802,54 +762,44 @@ void GrpcLb::BalancerCallState::OnBalancerMessageReceivedLocked( gpr_free(ipport); } } - /* update serverlist */ - if (serverlist->num_servers > 0) { - // Start sending client load report only after we start using the - // serverlist returned from the current LB call. - if (lb_calld->client_stats_report_interval_ > 0 && - lb_calld->client_stats_ == nullptr) { - lb_calld->client_stats_.reset(New<GrpcLbClientStats>()); - // TODO(roth): We currently track this ref manually. Once the - // ClosureRef API is ready, we should pass the RefCountedPtr<> along - // with the callback. - auto self = lb_calld->Ref(DEBUG_LOCATION, "client_load_report"); - self.release(); - lb_calld->ScheduleNextClientLoadReportLocked(); - } - if (grpc_grpclb_serverlist_equals(grpclb_policy->serverlist_, - serverlist)) { - if (grpc_lb_glb_trace.enabled()) { - gpr_log(GPR_INFO, - "[grpclb %p] Incoming server list identical to current, " - "ignoring.", - grpclb_policy); - } - grpc_grpclb_destroy_serverlist(serverlist); - } else { /* new serverlist */ - if (grpclb_policy->serverlist_ != nullptr) { - /* dispose of the old serverlist */ - grpc_grpclb_destroy_serverlist(grpclb_policy->serverlist_); - } else { - /* or dispose of the fallback */ - grpc_lb_addresses_destroy(grpclb_policy->fallback_backend_addresses_); - grpclb_policy->fallback_backend_addresses_ = nullptr; - if (grpclb_policy->fallback_timer_callback_pending_) { - grpc_timer_cancel(&grpclb_policy->lb_fallback_timer_); - } - } - // and update the copy in the GrpcLb instance. This - // serverlist instance will be destroyed either upon the next - // update or when the GrpcLb instance is destroyed. - grpclb_policy->serverlist_ = serverlist; - grpclb_policy->serverlist_index_ = 0; - grpclb_policy->CreateOrUpdateRoundRobinPolicyLocked(); - } - } else { + // Start sending client load report only after we start using the + // serverlist returned from the current LB call. + if (lb_calld->client_stats_report_interval_ > 0 && + lb_calld->client_stats_ == nullptr) { + lb_calld->client_stats_.reset(New<GrpcLbClientStats>()); + // TODO(roth): We currently track this ref manually. Once the + // ClosureRef API is ready, we should pass the RefCountedPtr<> along + // with the callback. + auto self = lb_calld->Ref(DEBUG_LOCATION, "client_load_report"); + self.release(); + lb_calld->ScheduleNextClientLoadReportLocked(); + } + // Check if the serverlist differs from the previous one. + if (grpc_grpclb_serverlist_equals(grpclb_policy->serverlist_, serverlist)) { if (grpc_lb_glb_trace.enabled()) { - gpr_log(GPR_INFO, "[grpclb %p] Received empty server list, ignoring.", + gpr_log(GPR_INFO, + "[grpclb %p] Incoming server list identical to current, " + "ignoring.", grpclb_policy); } grpc_grpclb_destroy_serverlist(serverlist); + } else { // New serverlist. + if (grpclb_policy->serverlist_ != nullptr) { + // Dispose of the old serverlist. + grpc_grpclb_destroy_serverlist(grpclb_policy->serverlist_); + } else { + // Dispose of the fallback. + grpclb_policy->fallback_backend_addresses_.reset(); + if (grpclb_policy->fallback_timer_callback_pending_) { + grpc_timer_cancel(&grpclb_policy->lb_fallback_timer_); + } + } + // Update the serverlist in the GrpcLb instance. This serverlist + // instance will be destroyed either upon the next update or when the + // GrpcLb instance is destroyed. + grpclb_policy->serverlist_ = serverlist; + grpclb_policy->serverlist_index_ = 0; + grpclb_policy->CreateOrUpdateRoundRobinPolicyLocked(); } } else { // No valid initial response or serverlist found. @@ -919,31 +869,25 @@ void GrpcLb::BalancerCallState::OnBalancerStatusReceivedLocked( // helper code for creating balancer channel // -grpc_lb_addresses* ExtractBalancerAddresses( - const grpc_lb_addresses* addresses) { - size_t num_grpclb_addrs = 0; - for (size_t i = 0; i < addresses->num_addresses; ++i) { - if (addresses->addresses[i].is_balancer) ++num_grpclb_addrs; - } - // There must be at least one balancer address, or else the - // client_channel would not have chosen this LB policy. - GPR_ASSERT(num_grpclb_addrs > 0); - grpc_lb_addresses* lb_addresses = - grpc_lb_addresses_create(num_grpclb_addrs, nullptr); - size_t lb_addresses_idx = 0; - for (size_t i = 0; i < addresses->num_addresses; ++i) { - if (!addresses->addresses[i].is_balancer) continue; - if (GPR_UNLIKELY(addresses->addresses[i].user_data != nullptr)) { - gpr_log(GPR_ERROR, - "This LB policy doesn't support user data. It will be ignored"); +ServerAddressList ExtractBalancerAddresses(const ServerAddressList& addresses) { + ServerAddressList balancer_addresses; + for (size_t i = 0; i < addresses.size(); ++i) { + if (addresses[i].IsBalancer()) { + // Strip out the is_balancer channel arg, since we don't want to + // recursively use the grpclb policy in the channel used to talk to + // the balancers. Note that we do NOT strip out the balancer_name + // channel arg, since we need that to set the authority correctly + // to talk to the balancers. + static const char* args_to_remove[] = { + GRPC_ARG_ADDRESS_IS_BALANCER, + }; + balancer_addresses.emplace_back( + addresses[i].address(), + grpc_channel_args_copy_and_remove(addresses[i].args(), args_to_remove, + GPR_ARRAY_SIZE(args_to_remove))); } - grpc_lb_addresses_set_address( - lb_addresses, lb_addresses_idx++, addresses->addresses[i].address.addr, - addresses->addresses[i].address.len, false /* is balancer */, - addresses->addresses[i].balancer_name, nullptr /* user data */); } - GPR_ASSERT(num_grpclb_addrs == lb_addresses_idx); - return lb_addresses; + return balancer_addresses; } /* Returns the channel args for the LB channel, used to create a bidirectional @@ -955,10 +899,10 @@ grpc_lb_addresses* ExtractBalancerAddresses( * above the grpclb policy. * - \a args: other args inherited from the grpclb policy. */ grpc_channel_args* BuildBalancerChannelArgs( - const grpc_lb_addresses* addresses, + const ServerAddressList& addresses, FakeResolverResponseGenerator* response_generator, const grpc_channel_args* args) { - grpc_lb_addresses* lb_addresses = ExtractBalancerAddresses(addresses); + ServerAddressList balancer_addresses = ExtractBalancerAddresses(addresses); // Channel args to remove. static const char* args_to_remove[] = { // LB policy name, since we want to use the default (pick_first) in @@ -976,7 +920,7 @@ grpc_channel_args* BuildBalancerChannelArgs( // is_balancer=true. We need the LB channel to return addresses with // is_balancer=false so that it does not wind up recursively using the // grpclb LB policy, as per the special case logic in client_channel.c. - GRPC_ARG_LB_ADDRESSES, + GRPC_ARG_SERVER_ADDRESS_LIST, // The fake resolver response generator, because we are replacing it // with the one from the grpclb policy, used to propagate updates to // the LB channel. @@ -992,10 +936,10 @@ grpc_channel_args* BuildBalancerChannelArgs( }; // Channel args to add. const grpc_arg args_to_add[] = { - // New LB addresses. + // New address list. // Note that we pass these in both when creating the LB channel // and via the fake resolver. The latter is what actually gets used. - grpc_lb_addresses_create_channel_arg(lb_addresses), + CreateServerAddressListChannelArg(&balancer_addresses), // The fake resolver response generator, which we use to inject // address updates into the LB channel. grpc_core::FakeResolverResponseGenerator::MakeChannelArg( @@ -1013,18 +957,14 @@ grpc_channel_args* BuildBalancerChannelArgs( args, args_to_remove, GPR_ARRAY_SIZE(args_to_remove), args_to_add, GPR_ARRAY_SIZE(args_to_add)); // Make any necessary modifications for security. - new_args = grpc_lb_policy_grpclb_modify_lb_channel_args(new_args); - // Clean up. - grpc_lb_addresses_destroy(lb_addresses); - return new_args; + return grpc_lb_policy_grpclb_modify_lb_channel_args(new_args); } // // ctor and dtor // -GrpcLb::GrpcLb(const grpc_lb_addresses* addresses, - const LoadBalancingPolicy::Args& args) +GrpcLb::GrpcLb(const LoadBalancingPolicy::Args& args) : LoadBalancingPolicy(args), response_generator_(MakeRefCounted<FakeResolverResponseGenerator>()), lb_call_backoff_( @@ -1081,9 +1021,6 @@ GrpcLb::~GrpcLb() { if (serverlist_ != nullptr) { grpc_grpclb_destroy_serverlist(serverlist_); } - if (fallback_backend_addresses_ != nullptr) { - grpc_lb_addresses_destroy(fallback_backend_addresses_); - } grpc_subchannel_index_unref(); } @@ -1131,7 +1068,6 @@ void GrpcLb::HandOffPendingPicksLocked(LoadBalancingPolicy* new_policy) { while ((pp = pending_picks_) != nullptr) { pending_picks_ = pp->next; pp->pick->on_complete = pp->original_on_complete; - pp->pick->user_data = nullptr; grpc_error* error = GRPC_ERROR_NONE; if (new_policy->PickLocked(pp->pick, &error)) { // Synchronous return; schedule closure. @@ -1285,9 +1221,27 @@ void GrpcLb::NotifyOnStateChangeLocked(grpc_connectivity_state* current, notify); } +// Returns the backend addresses extracted from the given addresses. +UniquePtr<ServerAddressList> ExtractBackendAddresses( + const ServerAddressList& addresses) { + void* lb_token = (void*)GRPC_MDELEM_LB_TOKEN_EMPTY.payload; + grpc_arg arg = grpc_channel_arg_pointer_create( + const_cast<char*>(GRPC_ARG_GRPCLB_ADDRESS_LB_TOKEN), lb_token, + &lb_token_arg_vtable); + auto backend_addresses = MakeUnique<ServerAddressList>(); + for (size_t i = 0; i < addresses.size(); ++i) { + if (!addresses[i].IsBalancer()) { + backend_addresses->emplace_back( + addresses[i].address(), + grpc_channel_args_copy_and_add(addresses[i].args(), &arg, 1)); + } + } + return backend_addresses; +} + void GrpcLb::ProcessChannelArgsLocked(const grpc_channel_args& args) { - const grpc_arg* arg = grpc_channel_args_find(&args, GRPC_ARG_LB_ADDRESSES); - if (GPR_UNLIKELY(arg == nullptr || arg->type != GRPC_ARG_POINTER)) { + const ServerAddressList* addresses = FindServerAddressListChannelArg(&args); + if (addresses == nullptr) { // Ignore this update. gpr_log( GPR_ERROR, @@ -1295,13 +1249,8 @@ void GrpcLb::ProcessChannelArgsLocked(const grpc_channel_args& args) { this); return; } - const grpc_lb_addresses* addresses = - static_cast<const grpc_lb_addresses*>(arg->value.pointer.p); // Update fallback address list. - if (fallback_backend_addresses_ != nullptr) { - grpc_lb_addresses_destroy(fallback_backend_addresses_); - } - fallback_backend_addresses_ = ExtractBackendAddresses(addresses); + fallback_backend_addresses_ = ExtractBackendAddresses(*addresses); // Make sure that GRPC_ARG_LB_POLICY_NAME is set in channel args, // since we use this to trigger the client_load_reporting filter. static const char* args_to_remove[] = {GRPC_ARG_LB_POLICY_NAME}; @@ -1312,7 +1261,7 @@ void GrpcLb::ProcessChannelArgsLocked(const grpc_channel_args& args) { &args, args_to_remove, GPR_ARRAY_SIZE(args_to_remove), &new_arg, 1); // Construct args for balancer channel. grpc_channel_args* lb_channel_args = - BuildBalancerChannelArgs(addresses, response_generator_.get(), &args); + BuildBalancerChannelArgs(*addresses, response_generator_.get(), &args); // Create balancer channel if needed. if (lb_channel_ == nullptr) { char* uri_str; @@ -1331,7 +1280,7 @@ void GrpcLb::ProcessChannelArgsLocked(const grpc_channel_args& args) { grpc_channel_args_destroy(lb_channel_args); } -void GrpcLb::UpdateLocked(const grpc_channel_args& args) { +void GrpcLb::UpdateLocked(const grpc_channel_args& args, grpc_json* lb_config) { ProcessChannelArgsLocked(args); // Update the existing RR policy. if (rr_policy_ != nullptr) CreateOrUpdateRoundRobinPolicyLocked(); @@ -1518,12 +1467,17 @@ void DestroyClientStats(void* arg) { } void GrpcLb::PendingPickSetMetadataAndContext(PendingPick* pp) { - /* if connected_subchannel is nullptr, no pick has been made by the RR - * policy (e.g., all addresses failed to connect). There won't be any - * user_data/token available */ + // If connected_subchannel is nullptr, no pick has been made by the RR + // policy (e.g., all addresses failed to connect). There won't be any + // LB token available. if (pp->pick->connected_subchannel != nullptr) { - if (GPR_LIKELY(!GRPC_MDISNULL(pp->lb_token))) { - AddLbTokenToInitialMetadata(GRPC_MDELEM_REF(pp->lb_token), + const grpc_arg* arg = + grpc_channel_args_find(pp->pick->connected_subchannel->args(), + GRPC_ARG_GRPCLB_ADDRESS_LB_TOKEN); + if (arg != nullptr) { + grpc_mdelem lb_token = { + reinterpret_cast<uintptr_t>(arg->value.pointer.p)}; + AddLbTokenToInitialMetadata(GRPC_MDELEM_REF(lb_token), &pp->pick->lb_token_mdelem_storage, pp->pick->initial_metadata); } else { @@ -1582,7 +1536,7 @@ void GrpcLb::AddPendingPick(PendingPick* pp) { bool GrpcLb::PickFromRoundRobinPolicyLocked(bool force_async, PendingPick* pp, grpc_error** error) { // Check for drops if we are not using fallback backend addresses. - if (serverlist_ != nullptr) { + if (serverlist_ != nullptr && serverlist_->num_servers > 0) { // Look at the index into the serverlist to see if we should drop this call. grpc_grpclb_server* server = serverlist_->servers[serverlist_index_++]; if (serverlist_index_ == serverlist_->num_servers) { @@ -1607,12 +1561,10 @@ bool GrpcLb::PickFromRoundRobinPolicyLocked(bool force_async, PendingPick* pp, return true; } } - // Set client_stats and user_data. + // Set client_stats. if (lb_calld_ != nullptr && lb_calld_->client_stats() != nullptr) { pp->client_stats = lb_calld_->client_stats()->Ref(); } - GPR_ASSERT(pp->pick->user_data == nullptr); - pp->pick->user_data = (void**)&pp->lb_token; // Pick via the RR policy. bool pick_done = rr_policy_->PickLocked(pp->pick, error); if (pick_done) { @@ -1677,11 +1629,11 @@ void GrpcLb::CreateRoundRobinPolicyLocked(const Args& args) { } grpc_channel_args* GrpcLb::CreateRoundRobinPolicyArgsLocked() { - grpc_lb_addresses* addresses; + ServerAddressList tmp_addresses; + ServerAddressList* addresses = &tmp_addresses; bool is_backend_from_grpclb_load_balancer = false; if (serverlist_ != nullptr) { - GPR_ASSERT(serverlist_->num_servers > 0); - addresses = ProcessServerlist(serverlist_); + tmp_addresses = ProcessServerlist(serverlist_); is_backend_from_grpclb_load_balancer = true; } else { // If CreateOrUpdateRoundRobinPolicyLocked() is invoked when we haven't @@ -1690,14 +1642,14 @@ grpc_channel_args* GrpcLb::CreateRoundRobinPolicyArgsLocked() { // empty, in which case the new round_robin policy will keep the requested // picks pending. GPR_ASSERT(fallback_backend_addresses_ != nullptr); - addresses = grpc_lb_addresses_copy(fallback_backend_addresses_); + addresses = fallback_backend_addresses_.get(); } GPR_ASSERT(addresses != nullptr); - // Replace the LB addresses in the channel args that we pass down to + // Replace the server address list in the channel args that we pass down to // the subchannel. - static const char* keys_to_remove[] = {GRPC_ARG_LB_ADDRESSES}; + static const char* keys_to_remove[] = {GRPC_ARG_SERVER_ADDRESS_LIST}; grpc_arg args_to_add[3] = { - grpc_lb_addresses_create_channel_arg(addresses), + CreateServerAddressListChannelArg(addresses), // A channel arg indicating if the target is a backend inferred from a // grpclb load balancer. grpc_channel_arg_integer_create( @@ -1714,7 +1666,6 @@ grpc_channel_args* GrpcLb::CreateRoundRobinPolicyArgsLocked() { grpc_channel_args* args = grpc_channel_args_copy_and_add_and_remove( args_, keys_to_remove, GPR_ARRAY_SIZE(keys_to_remove), args_to_add, num_args_to_add); - grpc_lb_addresses_destroy(addresses); return args; } @@ -1727,7 +1678,7 @@ void GrpcLb::CreateOrUpdateRoundRobinPolicyLocked() { gpr_log(GPR_INFO, "[grpclb %p] Updating RR policy %p", this, rr_policy_.get()); } - rr_policy_->UpdateLocked(*args); + rr_policy_->UpdateLocked(*args, nullptr); } else { LoadBalancingPolicy::Args lb_policy_args; lb_policy_args.combiner = combiner(); @@ -1847,19 +1798,18 @@ class GrpcLbFactory : public LoadBalancingPolicyFactory { OrphanablePtr<LoadBalancingPolicy> CreateLoadBalancingPolicy( const LoadBalancingPolicy::Args& args) const override { /* Count the number of gRPC-LB addresses. There must be at least one. */ - const grpc_arg* arg = - grpc_channel_args_find(args.args, GRPC_ARG_LB_ADDRESSES); - if (arg == nullptr || arg->type != GRPC_ARG_POINTER) { - return nullptr; - } - grpc_lb_addresses* addresses = - static_cast<grpc_lb_addresses*>(arg->value.pointer.p); - size_t num_grpclb_addrs = 0; - for (size_t i = 0; i < addresses->num_addresses; ++i) { - if (addresses->addresses[i].is_balancer) ++num_grpclb_addrs; + const ServerAddressList* addresses = + FindServerAddressListChannelArg(args.args); + if (addresses == nullptr) return nullptr; + bool found_balancer = false; + for (size_t i = 0; i < addresses->size(); ++i) { + if ((*addresses)[i].IsBalancer()) { + found_balancer = true; + break; + } } - if (num_grpclb_addrs == 0) return nullptr; - return OrphanablePtr<LoadBalancingPolicy>(New<GrpcLb>(addresses, args)); + if (!found_balancer) return nullptr; + return OrphanablePtr<LoadBalancingPolicy>(New<GrpcLb>(args)); } const char* name() const override { return "grpclb"; } diff --git a/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb_channel.h b/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb_channel.h index 825065a9c3..3b2dc370eb 100644 --- a/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb_channel.h +++ b/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb_channel.h @@ -21,7 +21,7 @@ #include <grpc/support/port_platform.h> -#include "src/core/ext/filters/client_channel/lb_policy_factory.h" +#include <grpc/impl/codegen/grpc_types.h> /// Makes any necessary modifications to \a args for use in the grpclb /// balancer channel. diff --git a/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb_channel_secure.cc b/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb_channel_secure.cc index 441efd5e23..657ff69312 100644 --- a/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb_channel_secure.cc +++ b/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb_channel_secure.cc @@ -26,6 +26,7 @@ #include <grpc/support/string_util.h> #include "src/core/ext/filters/client_channel/client_channel.h" +#include "src/core/ext/filters/client_channel/server_address.h" #include "src/core/lib/channel/channel_args.h" #include "src/core/lib/gpr/string.h" #include "src/core/lib/iomgr/sockaddr_utils.h" @@ -42,22 +43,23 @@ int BalancerNameCmp(const grpc_core::UniquePtr<char>& a, } RefCountedPtr<TargetAuthorityTable> CreateTargetAuthorityTable( - grpc_lb_addresses* addresses) { + const ServerAddressList& addresses) { TargetAuthorityTable::Entry* target_authority_entries = - static_cast<TargetAuthorityTable::Entry*>(gpr_zalloc( - sizeof(*target_authority_entries) * addresses->num_addresses)); - for (size_t i = 0; i < addresses->num_addresses; ++i) { + static_cast<TargetAuthorityTable::Entry*>( + gpr_zalloc(sizeof(*target_authority_entries) * addresses.size())); + for (size_t i = 0; i < addresses.size(); ++i) { char* addr_str; - GPR_ASSERT(grpc_sockaddr_to_string( - &addr_str, &addresses->addresses[i].address, true) > 0); + GPR_ASSERT( + grpc_sockaddr_to_string(&addr_str, &addresses[i].address(), true) > 0); target_authority_entries[i].key = grpc_slice_from_copied_string(addr_str); - target_authority_entries[i].value.reset( - gpr_strdup(addresses->addresses[i].balancer_name)); gpr_free(addr_str); + char* balancer_name = grpc_channel_arg_get_string(grpc_channel_args_find( + addresses[i].args(), GRPC_ARG_ADDRESS_BALANCER_NAME)); + target_authority_entries[i].value.reset(gpr_strdup(balancer_name)); } RefCountedPtr<TargetAuthorityTable> target_authority_table = - TargetAuthorityTable::Create(addresses->num_addresses, - target_authority_entries, BalancerNameCmp); + TargetAuthorityTable::Create(addresses.size(), target_authority_entries, + BalancerNameCmp); gpr_free(target_authority_entries); return target_authority_table; } @@ -72,13 +74,12 @@ grpc_channel_args* grpc_lb_policy_grpclb_modify_lb_channel_args( grpc_arg args_to_add[2]; size_t num_args_to_add = 0; // Add arg for targets info table. - const grpc_arg* arg = grpc_channel_args_find(args, GRPC_ARG_LB_ADDRESSES); - GPR_ASSERT(arg != nullptr); - GPR_ASSERT(arg->type == GRPC_ARG_POINTER); - grpc_lb_addresses* addresses = - static_cast<grpc_lb_addresses*>(arg->value.pointer.p); + grpc_core::ServerAddressList* addresses = + grpc_core::FindServerAddressListChannelArg(args); + GPR_ASSERT(addresses != nullptr); grpc_core::RefCountedPtr<grpc_core::TargetAuthorityTable> - target_authority_table = grpc_core::CreateTargetAuthorityTable(addresses); + target_authority_table = + grpc_core::CreateTargetAuthorityTable(*addresses); args_to_add[num_args_to_add++] = grpc_core::CreateTargetAuthorityTableChannelArg( target_authority_table.get()); @@ -87,22 +88,18 @@ grpc_channel_args* grpc_lb_policy_grpclb_modify_lb_channel_args( // bearer token credentials. grpc_channel_credentials* channel_credentials = grpc_channel_credentials_find_in_args(args); - grpc_channel_credentials* creds_sans_call_creds = nullptr; + grpc_core::RefCountedPtr<grpc_channel_credentials> creds_sans_call_creds; if (channel_credentials != nullptr) { creds_sans_call_creds = - grpc_channel_credentials_duplicate_without_call_credentials( - channel_credentials); + channel_credentials->duplicate_without_call_credentials(); GPR_ASSERT(creds_sans_call_creds != nullptr); args_to_remove[num_args_to_remove++] = GRPC_ARG_CHANNEL_CREDENTIALS; args_to_add[num_args_to_add++] = - grpc_channel_credentials_to_arg(creds_sans_call_creds); + grpc_channel_credentials_to_arg(creds_sans_call_creds.get()); } grpc_channel_args* result = grpc_channel_args_copy_and_add_and_remove( args, args_to_remove, num_args_to_remove, args_to_add, num_args_to_add); // Clean up. grpc_channel_args_destroy(args); - if (creds_sans_call_creds != nullptr) { - grpc_channel_credentials_unref(creds_sans_call_creds); - } return result; } diff --git a/src/core/ext/filters/client_channel/lb_policy/grpclb/load_balancer_api.h b/src/core/ext/filters/client_channel/lb_policy/grpclb/load_balancer_api.h index 9ca7b28d8e..71d371c880 100644 --- a/src/core/ext/filters/client_channel/lb_policy/grpclb/load_balancer_api.h +++ b/src/core/ext/filters/client_channel/lb_policy/grpclb/load_balancer_api.h @@ -25,7 +25,7 @@ #include "src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb_client_stats.h" #include "src/core/ext/filters/client_channel/lb_policy/grpclb/proto/grpc/lb/v1/load_balancer.pb.h" -#include "src/core/ext/filters/client_channel/lb_policy_factory.h" +#include "src/core/lib/iomgr/exec_ctx.h" #define GRPC_GRPCLB_SERVICE_NAME_MAX_LENGTH 128 diff --git a/src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.cc b/src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.cc index eb494486b9..74c17612a2 100644 --- a/src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.cc +++ b/src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.cc @@ -24,6 +24,7 @@ #include "src/core/ext/filters/client_channel/lb_policy/subchannel_list.h" #include "src/core/ext/filters/client_channel/lb_policy_registry.h" +#include "src/core/ext/filters/client_channel/server_address.h" #include "src/core/ext/filters/client_channel/subchannel.h" #include "src/core/ext/filters/client_channel/subchannel_index.h" #include "src/core/lib/channel/channel_args.h" @@ -46,7 +47,8 @@ class PickFirst : public LoadBalancingPolicy { public: explicit PickFirst(const Args& args); - void UpdateLocked(const grpc_channel_args& args) override; + void UpdateLocked(const grpc_channel_args& args, + grpc_json* lb_config) override; bool PickLocked(PickState* pick, grpc_error** error) override; void CancelPickLocked(PickState* pick, grpc_error* error) override; void CancelMatchingPicksLocked(uint32_t initial_metadata_flags_mask, @@ -74,11 +76,9 @@ class PickFirst : public LoadBalancingPolicy { PickFirstSubchannelData( SubchannelList<PickFirstSubchannelList, PickFirstSubchannelData>* subchannel_list, - const grpc_lb_user_data_vtable* user_data_vtable, - const grpc_lb_address& address, grpc_subchannel* subchannel, + const ServerAddress& address, grpc_subchannel* subchannel, grpc_combiner* combiner) - : SubchannelData(subchannel_list, user_data_vtable, address, subchannel, - combiner) {} + : SubchannelData(subchannel_list, address, subchannel, combiner) {} void ProcessConnectivityChangeLocked( grpc_connectivity_state connectivity_state, grpc_error* error) override; @@ -94,7 +94,7 @@ class PickFirst : public LoadBalancingPolicy { PickFirstSubchannelData> { public: PickFirstSubchannelList(PickFirst* policy, TraceFlag* tracer, - const grpc_lb_addresses* addresses, + const ServerAddressList& addresses, grpc_combiner* combiner, grpc_client_channel_factory* client_channel_factory, const grpc_channel_args& args) @@ -159,7 +159,7 @@ PickFirst::PickFirst(const Args& args) : LoadBalancingPolicy(args) { if (grpc_lb_pick_first_trace.enabled()) { gpr_log(GPR_INFO, "Pick First %p created.", this); } - UpdateLocked(*args.args); + UpdateLocked(*args.args, args.lb_config); grpc_subchannel_index_ref(); } @@ -333,10 +333,11 @@ void PickFirst::UpdateChildRefsLocked() { child_subchannels_ = std::move(cs); } -void PickFirst::UpdateLocked(const grpc_channel_args& args) { +void PickFirst::UpdateLocked(const grpc_channel_args& args, + grpc_json* lb_config) { AutoChildRefsUpdater guard(this); - const grpc_arg* arg = grpc_channel_args_find(&args, GRPC_ARG_LB_ADDRESSES); - if (arg == nullptr || arg->type != GRPC_ARG_POINTER) { + const ServerAddressList* addresses = FindServerAddressListChannelArg(&args); + if (addresses == nullptr) { if (subchannel_list_ == nullptr) { // If we don't have a current subchannel list, go into TRANSIENT FAILURE. grpc_connectivity_state_set( @@ -352,19 +353,17 @@ void PickFirst::UpdateLocked(const grpc_channel_args& args) { } return; } - const grpc_lb_addresses* addresses = - static_cast<const grpc_lb_addresses*>(arg->value.pointer.p); if (grpc_lb_pick_first_trace.enabled()) { gpr_log(GPR_INFO, "Pick First %p received update with %" PRIuPTR " addresses", this, - addresses->num_addresses); + addresses->size()); } grpc_arg new_arg = grpc_channel_arg_integer_create( const_cast<char*>(GRPC_ARG_INHIBIT_HEALTH_CHECKING), 1); grpc_channel_args* new_args = grpc_channel_args_copy_and_add(&args, &new_arg, 1); auto subchannel_list = MakeOrphanable<PickFirstSubchannelList>( - this, &grpc_lb_pick_first_trace, addresses, combiner(), + this, &grpc_lb_pick_first_trace, *addresses, combiner(), client_channel_factory(), *new_args); grpc_channel_args_destroy(new_args); if (subchannel_list->num_subchannels() == 0) { @@ -378,6 +377,31 @@ void PickFirst::UpdateLocked(const grpc_channel_args& args) { selected_ = nullptr; return; } + // If one of the subchannels in the new list is already in state + // READY, then select it immediately. This can happen when the + // currently selected subchannel is also present in the update. It + // can also happen if one of the subchannels in the update is already + // in the subchannel index because it's in use by another channel. + for (size_t i = 0; i < subchannel_list->num_subchannels(); ++i) { + PickFirstSubchannelData* sd = subchannel_list->subchannel(i); + grpc_error* error = GRPC_ERROR_NONE; + grpc_connectivity_state state = sd->CheckConnectivityStateLocked(&error); + GRPC_ERROR_UNREF(error); + if (state == GRPC_CHANNEL_READY) { + subchannel_list_ = std::move(subchannel_list); + sd->ProcessUnselectedReadyLocked(); + sd->StartConnectivityWatchLocked(); + // If there was a previously pending update (which may or may + // not have contained the currently selected subchannel), drop + // it, so that it doesn't override what we've done here. + latest_pending_subchannel_list_.reset(); + // Make sure that subsequent calls to ExitIdleLocked() don't cause + // us to start watching a subchannel other than the one we've + // selected. + started_picking_ = true; + return; + } + } if (selected_ == nullptr) { // We don't yet have a selected subchannel, so replace the current // subchannel list immediately. @@ -385,46 +409,14 @@ void PickFirst::UpdateLocked(const grpc_channel_args& args) { // If we've started picking, start trying to connect to the first // subchannel in the new list. if (started_picking_) { - subchannel_list_->subchannel(0) - ->CheckConnectivityStateAndStartWatchingLocked(); + // Note: No need to use CheckConnectivityStateAndStartWatchingLocked() + // here, since we've already checked the initial connectivity + // state of all subchannels above. + subchannel_list_->subchannel(0)->StartConnectivityWatchLocked(); } } else { - // We do have a selected subchannel. - // Check if it's present in the new list. If so, we're done. - for (size_t i = 0; i < subchannel_list->num_subchannels(); ++i) { - PickFirstSubchannelData* sd = subchannel_list->subchannel(i); - if (sd->subchannel() == selected_->subchannel()) { - // The currently selected subchannel is in the update: we are done. - if (grpc_lb_pick_first_trace.enabled()) { - gpr_log(GPR_INFO, - "Pick First %p found already selected subchannel %p " - "at update index %" PRIuPTR " of %" PRIuPTR "; update done", - this, selected_->subchannel(), i, - subchannel_list->num_subchannels()); - } - // Make sure it's in state READY. It might not be if we grabbed - // the combiner while a connectivity state notification - // informing us otherwise is pending. - // Note that CheckConnectivityStateLocked() also takes a ref to - // the connected subchannel. - grpc_error* error = GRPC_ERROR_NONE; - if (sd->CheckConnectivityStateLocked(&error) == GRPC_CHANNEL_READY) { - selected_ = sd; - subchannel_list_ = std::move(subchannel_list); - sd->StartConnectivityWatchLocked(); - // If there was a previously pending update (which may or may - // not have contained the currently selected subchannel), drop - // it, so that it doesn't override what we've done here. - latest_pending_subchannel_list_.reset(); - return; - } - GRPC_ERROR_UNREF(error); - } - } - // Not keeping the previous selected subchannel, so set the latest - // pending subchannel list to the new subchannel list. We will wait - // for it to report READY before swapping it into the current - // subchannel list. + // We do have a selected subchannel, so keep using it until one of + // the subchannels in the new list reports READY. if (latest_pending_subchannel_list_ != nullptr) { if (grpc_lb_pick_first_trace.enabled()) { gpr_log(GPR_INFO, @@ -438,8 +430,11 @@ void PickFirst::UpdateLocked(const grpc_channel_args& args) { // If we've started picking, start trying to connect to the first // subchannel in the new list. if (started_picking_) { + // Note: No need to use CheckConnectivityStateAndStartWatchingLocked() + // here, since we've already checked the initial connectivity + // state of all subchannels above. latest_pending_subchannel_list_->subchannel(0) - ->CheckConnectivityStateAndStartWatchingLocked(); + ->StartConnectivityWatchLocked(); } } } diff --git a/src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.cc b/src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.cc index e9ed85cf66..63089afbd7 100644 --- a/src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.cc +++ b/src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.cc @@ -57,7 +57,8 @@ class RoundRobin : public LoadBalancingPolicy { public: explicit RoundRobin(const Args& args); - void UpdateLocked(const grpc_channel_args& args) override; + void UpdateLocked(const grpc_channel_args& args, + grpc_json* lb_config) override; bool PickLocked(PickState* pick, grpc_error** error) override; void CancelPickLocked(PickState* pick, grpc_error* error) override; void CancelMatchingPicksLocked(uint32_t initial_metadata_flags_mask, @@ -81,8 +82,6 @@ class RoundRobin : public LoadBalancingPolicy { // Data for a particular subchannel in a subchannel list. // This subclass adds the following functionality: - // - Tracks user_data associated with each address, which will be - // returned along with picks that select the subchannel. // - Tracks the previous connectivity state of the subchannel, so that // we know how many subchannels are in each state. class RoundRobinSubchannelData @@ -92,26 +91,9 @@ class RoundRobin : public LoadBalancingPolicy { RoundRobinSubchannelData( SubchannelList<RoundRobinSubchannelList, RoundRobinSubchannelData>* subchannel_list, - const grpc_lb_user_data_vtable* user_data_vtable, - const grpc_lb_address& address, grpc_subchannel* subchannel, + const ServerAddress& address, grpc_subchannel* subchannel, grpc_combiner* combiner) - : SubchannelData(subchannel_list, user_data_vtable, address, subchannel, - combiner), - user_data_vtable_(user_data_vtable), - user_data_(user_data_vtable_ != nullptr - ? user_data_vtable_->copy(address.user_data) - : nullptr) {} - - void UnrefSubchannelLocked(const char* reason) override { - SubchannelData::UnrefSubchannelLocked(reason); - if (user_data_ != nullptr) { - GPR_ASSERT(user_data_vtable_ != nullptr); - user_data_vtable_->destroy(user_data_); - user_data_ = nullptr; - } - } - - void* user_data() const { return user_data_; } + : SubchannelData(subchannel_list, address, subchannel, combiner) {} grpc_connectivity_state connectivity_state() const { return last_connectivity_state_; @@ -124,8 +106,6 @@ class RoundRobin : public LoadBalancingPolicy { void ProcessConnectivityChangeLocked( grpc_connectivity_state connectivity_state, grpc_error* error) override; - const grpc_lb_user_data_vtable* user_data_vtable_; - void* user_data_ = nullptr; grpc_connectivity_state last_connectivity_state_ = GRPC_CHANNEL_IDLE; }; @@ -136,7 +116,7 @@ class RoundRobin : public LoadBalancingPolicy { public: RoundRobinSubchannelList( RoundRobin* policy, TraceFlag* tracer, - const grpc_lb_addresses* addresses, grpc_combiner* combiner, + const ServerAddressList& addresses, grpc_combiner* combiner, grpc_client_channel_factory* client_channel_factory, const grpc_channel_args& args) : SubchannelList(policy, tracer, addresses, combiner, @@ -232,7 +212,7 @@ RoundRobin::RoundRobin(const Args& args) : LoadBalancingPolicy(args) { gpr_mu_init(&child_refs_mu_); grpc_connectivity_state_init(&state_tracker_, GRPC_CHANNEL_IDLE, "round_robin"); - UpdateLocked(*args.args); + UpdateLocked(*args.args, args.lb_config); if (grpc_lb_round_robin_trace.enabled()) { gpr_log(GPR_INFO, "[RR %p] Created with %" PRIuPTR " subchannels", this, subchannel_list_->num_subchannels()); @@ -353,9 +333,6 @@ bool RoundRobin::DoPickLocked(PickState* pick) { subchannel_list_->subchannel(next_ready_index); GPR_ASSERT(sd->connected_subchannel() != nullptr); pick->connected_subchannel = sd->connected_subchannel()->Ref(); - if (pick->user_data != nullptr) { - *pick->user_data = sd->user_data(); - } if (grpc_lb_round_robin_trace.enabled()) { gpr_log(GPR_INFO, "[RR %p] Picked target <-- Subchannel %p (connected %p) (sl %p, " @@ -664,10 +641,11 @@ void RoundRobin::NotifyOnStateChangeLocked(grpc_connectivity_state* current, notify); } -void RoundRobin::UpdateLocked(const grpc_channel_args& args) { - const grpc_arg* arg = grpc_channel_args_find(&args, GRPC_ARG_LB_ADDRESSES); +void RoundRobin::UpdateLocked(const grpc_channel_args& args, + grpc_json* lb_config) { AutoChildRefsUpdater guard(this); - if (GPR_UNLIKELY(arg == nullptr || arg->type != GRPC_ARG_POINTER)) { + const ServerAddressList* addresses = FindServerAddressListChannelArg(&args); + if (addresses == nullptr) { gpr_log(GPR_ERROR, "[RR %p] update provided no addresses; ignoring", this); // If we don't have a current subchannel list, go into TRANSIENT_FAILURE. // Otherwise, keep using the current subchannel list (ignore this update). @@ -679,11 +657,9 @@ void RoundRobin::UpdateLocked(const grpc_channel_args& args) { } return; } - grpc_lb_addresses* addresses = - static_cast<grpc_lb_addresses*>(arg->value.pointer.p); if (grpc_lb_round_robin_trace.enabled()) { gpr_log(GPR_INFO, "[RR %p] received update with %" PRIuPTR " addresses", - this, addresses->num_addresses); + this, addresses->size()); } // Replace latest_pending_subchannel_list_. if (latest_pending_subchannel_list_ != nullptr) { @@ -694,7 +670,7 @@ void RoundRobin::UpdateLocked(const grpc_channel_args& args) { } } latest_pending_subchannel_list_ = MakeOrphanable<RoundRobinSubchannelList>( - this, &grpc_lb_round_robin_trace, addresses, combiner(), + this, &grpc_lb_round_robin_trace, *addresses, combiner(), client_channel_factory(), args); // If we haven't started picking yet or the new list is empty, // immediately promote the new list to the current list. diff --git a/src/core/ext/filters/client_channel/lb_policy/subchannel_list.h b/src/core/ext/filters/client_channel/lb_policy/subchannel_list.h index 4ec9e935ed..6f31a643c1 100644 --- a/src/core/ext/filters/client_channel/lb_policy/subchannel_list.h +++ b/src/core/ext/filters/client_channel/lb_policy/subchannel_list.h @@ -26,6 +26,7 @@ #include <grpc/support/alloc.h> #include "src/core/ext/filters/client_channel/lb_policy_registry.h" +#include "src/core/ext/filters/client_channel/server_address.h" #include "src/core/ext/filters/client_channel/subchannel.h" #include "src/core/lib/channel/channel_args.h" #include "src/core/lib/debug/trace.h" @@ -141,8 +142,7 @@ class SubchannelData { protected: SubchannelData( SubchannelList<SubchannelListType, SubchannelDataType>* subchannel_list, - const grpc_lb_user_data_vtable* user_data_vtable, - const grpc_lb_address& address, grpc_subchannel* subchannel, + const ServerAddress& address, grpc_subchannel* subchannel, grpc_combiner* combiner); virtual ~SubchannelData(); @@ -156,9 +156,8 @@ class SubchannelData { grpc_connectivity_state connectivity_state, grpc_error* error) GRPC_ABSTRACT; - // Unrefs the subchannel. May be overridden by subclasses that need - // to perform extra cleanup when unreffing the subchannel. - virtual void UnrefSubchannelLocked(const char* reason); + // Unrefs the subchannel. + void UnrefSubchannelLocked(const char* reason); private: // Updates connected_subchannel_ based on pending_connectivity_state_unsafe_. @@ -186,8 +185,7 @@ class SubchannelData { // A list of subchannels. template <typename SubchannelListType, typename SubchannelDataType> -class SubchannelList - : public InternallyRefCountedWithTracing<SubchannelListType> { +class SubchannelList : public InternallyRefCounted<SubchannelListType> { public: typedef InlinedVector<SubchannelDataType, 10> SubchannelVector; @@ -226,15 +224,14 @@ class SubchannelList // Note: Caller must ensure that this is invoked inside of the combiner. void Orphan() override { ShutdownLocked(); - InternallyRefCountedWithTracing<SubchannelListType>::Unref(DEBUG_LOCATION, - "shutdown"); + InternallyRefCounted<SubchannelListType>::Unref(DEBUG_LOCATION, "shutdown"); } GRPC_ABSTRACT_BASE_CLASS protected: SubchannelList(LoadBalancingPolicy* policy, TraceFlag* tracer, - const grpc_lb_addresses* addresses, grpc_combiner* combiner, + const ServerAddressList& addresses, grpc_combiner* combiner, grpc_client_channel_factory* client_channel_factory, const grpc_channel_args& args); @@ -279,8 +276,7 @@ class SubchannelList template <typename SubchannelListType, typename SubchannelDataType> SubchannelData<SubchannelListType, SubchannelDataType>::SubchannelData( SubchannelList<SubchannelListType, SubchannelDataType>* subchannel_list, - const grpc_lb_user_data_vtable* user_data_vtable, - const grpc_lb_address& address, grpc_subchannel* subchannel, + const ServerAddress& address, grpc_subchannel* subchannel, grpc_combiner* combiner) : subchannel_list_(subchannel_list), subchannel_(subchannel), @@ -490,19 +486,19 @@ void SubchannelData<SubchannelListType, SubchannelDataType>::ShutdownLocked() { template <typename SubchannelListType, typename SubchannelDataType> SubchannelList<SubchannelListType, SubchannelDataType>::SubchannelList( LoadBalancingPolicy* policy, TraceFlag* tracer, - const grpc_lb_addresses* addresses, grpc_combiner* combiner, + const ServerAddressList& addresses, grpc_combiner* combiner, grpc_client_channel_factory* client_channel_factory, const grpc_channel_args& args) - : InternallyRefCountedWithTracing<SubchannelListType>(tracer), + : InternallyRefCounted<SubchannelListType>(tracer), policy_(policy), tracer_(tracer), combiner_(GRPC_COMBINER_REF(combiner, "subchannel_list")) { if (tracer_->enabled()) { gpr_log(GPR_INFO, "[%s %p] Creating subchannel list %p for %" PRIuPTR " subchannels", - tracer_->name(), policy, this, addresses->num_addresses); + tracer_->name(), policy, this, addresses.size()); } - subchannels_.reserve(addresses->num_addresses); + subchannels_.reserve(addresses.size()); // We need to remove the LB addresses in order to be able to compare the // subchannel keys of subchannels from a different batch of addresses. // We also remove the inhibit-health-checking arg, since we are @@ -510,19 +506,27 @@ SubchannelList<SubchannelListType, SubchannelDataType>::SubchannelList( inhibit_health_checking_ = grpc_channel_arg_get_bool( grpc_channel_args_find(&args, GRPC_ARG_INHIBIT_HEALTH_CHECKING), false); static const char* keys_to_remove[] = {GRPC_ARG_SUBCHANNEL_ADDRESS, - GRPC_ARG_LB_ADDRESSES, + GRPC_ARG_SERVER_ADDRESS_LIST, GRPC_ARG_INHIBIT_HEALTH_CHECKING}; // Create a subchannel for each address. grpc_subchannel_args sc_args; - for (size_t i = 0; i < addresses->num_addresses; i++) { - // If there were any balancer, we would have chosen grpclb policy instead. - GPR_ASSERT(!addresses->addresses[i].is_balancer); + for (size_t i = 0; i < addresses.size(); i++) { + // If there were any balancer addresses, we would have chosen grpclb + // policy, which does not use a SubchannelList. + GPR_ASSERT(!addresses[i].IsBalancer()); memset(&sc_args, 0, sizeof(grpc_subchannel_args)); - grpc_arg addr_arg = - grpc_create_subchannel_address_arg(&addresses->addresses[i].address); + InlinedVector<grpc_arg, 4> args_to_add; + args_to_add.emplace_back( + grpc_create_subchannel_address_arg(&addresses[i].address())); + if (addresses[i].args() != nullptr) { + for (size_t j = 0; j < addresses[i].args()->num_args; ++j) { + args_to_add.emplace_back(addresses[i].args()->args[j]); + } + } grpc_channel_args* new_args = grpc_channel_args_copy_and_add_and_remove( - &args, keys_to_remove, GPR_ARRAY_SIZE(keys_to_remove), &addr_arg, 1); - gpr_free(addr_arg.value.string); + &args, keys_to_remove, GPR_ARRAY_SIZE(keys_to_remove), + args_to_add.data(), args_to_add.size()); + gpr_free(args_to_add[0].value.string); sc_args.args = new_args; grpc_subchannel* subchannel = grpc_client_channel_factory_create_subchannel( client_channel_factory, &sc_args); @@ -530,8 +534,7 @@ SubchannelList<SubchannelListType, SubchannelDataType>::SubchannelList( if (subchannel == nullptr) { // Subchannel could not be created. if (tracer_->enabled()) { - char* address_uri = - grpc_sockaddr_to_uri(&addresses->addresses[i].address); + char* address_uri = grpc_sockaddr_to_uri(&addresses[i].address()); gpr_log(GPR_INFO, "[%s %p] could not create subchannel for address uri %s, " "ignoring", @@ -541,8 +544,7 @@ SubchannelList<SubchannelListType, SubchannelDataType>::SubchannelList( continue; } if (tracer_->enabled()) { - char* address_uri = - grpc_sockaddr_to_uri(&addresses->addresses[i].address); + char* address_uri = grpc_sockaddr_to_uri(&addresses[i].address()); gpr_log(GPR_INFO, "[%s %p] subchannel list %p index %" PRIuPTR ": Created subchannel %p for address uri %s", @@ -550,8 +552,7 @@ SubchannelList<SubchannelListType, SubchannelDataType>::SubchannelList( address_uri); gpr_free(address_uri); } - subchannels_.emplace_back(this, addresses->user_data_vtable, - addresses->addresses[i], subchannel, combiner); + subchannels_.emplace_back(this, addresses[i], subchannel, combiner); } } diff --git a/src/core/ext/filters/client_channel/lb_policy/xds/xds.cc b/src/core/ext/filters/client_channel/lb_policy/xds/xds.cc index 59d57295d4..3c25de2386 100644 --- a/src/core/ext/filters/client_channel/lb_policy/xds/xds.cc +++ b/src/core/ext/filters/client_channel/lb_policy/xds/xds.cc @@ -79,6 +79,7 @@ #include "src/core/ext/filters/client_channel/lb_policy_registry.h" #include "src/core/ext/filters/client_channel/parse_address.h" #include "src/core/ext/filters/client_channel/resolver/fake/fake_resolver.h" +#include "src/core/ext/filters/client_channel/server_address.h" #include "src/core/ext/filters/client_channel/subchannel_index.h" #include "src/core/lib/backoff/backoff.h" #include "src/core/lib/channel/channel_args.h" @@ -116,9 +117,10 @@ namespace { class XdsLb : public LoadBalancingPolicy { public: - XdsLb(const grpc_lb_addresses* addresses, const Args& args); + explicit XdsLb(const Args& args); - void UpdateLocked(const grpc_channel_args& args) override; + void UpdateLocked(const grpc_channel_args& args, + grpc_json* lb_config) override; bool PickLocked(PickState* pick, grpc_error** error) override; void CancelPickLocked(PickState* pick, grpc_error* error) override; void CancelMatchingPicksLocked(uint32_t initial_metadata_flags_mask, @@ -155,9 +157,6 @@ class XdsLb : public LoadBalancingPolicy { // Our on_complete closure and the original one. grpc_closure on_complete; grpc_closure* original_on_complete; - // The LB token associated with the pick. This is set via user_data in - // the pick. - grpc_mdelem lb_token; // Stats for client-side load reporting. RefCountedPtr<XdsLbClientStats> client_stats; // Next pending pick. @@ -165,8 +164,7 @@ class XdsLb : public LoadBalancingPolicy { }; /// Contains a call to the LB server and all the data related to the call. - class BalancerCallState - : public InternallyRefCountedWithTracing<BalancerCallState> { + class BalancerCallState : public InternallyRefCounted<BalancerCallState> { public: explicit BalancerCallState( RefCountedPtr<LoadBalancingPolicy> parent_xdslb_policy); @@ -198,7 +196,6 @@ class XdsLb : public LoadBalancingPolicy { static bool LoadReportCountersAreZero(xds_grpclb_request* request); static void MaybeSendClientLoadReportLocked(void* arg, grpc_error* error); - static void ClientLoadReportDoneLocked(void* arg, grpc_error* error); static void OnInitialRequestSentLocked(void* arg, grpc_error* error); static void OnBalancerMessageReceivedLocked(void* arg, grpc_error* error); static void OnBalancerStatusReceivedLocked(void* arg, grpc_error* error); @@ -257,7 +254,7 @@ class XdsLb : public LoadBalancingPolicy { grpc_error* error); // Pending pick methods. - static void PendingPickSetMetadataAndContext(PendingPick* pp); + static void PendingPickCleanup(PendingPick* pp); PendingPick* PendingPickCreate(PickState* pick); void AddPendingPick(PendingPick* pp); static void OnPendingPickComplete(void* arg, grpc_error* error); @@ -320,7 +317,7 @@ class XdsLb : public LoadBalancingPolicy { // 0 means not using fallback. int lb_fallback_timeout_ms_ = 0; // The backend addresses from the resolver. - grpc_lb_addresses* fallback_backend_addresses_ = nullptr; + UniquePtr<ServerAddressList> fallback_backend_addresses_; // Fallback timer. bool fallback_timer_callback_pending_ = false; grpc_timer lb_fallback_timer_; @@ -340,47 +337,15 @@ class XdsLb : public LoadBalancingPolicy { // serverlist parsing code // -// vtable for LB tokens in grpc_lb_addresses -void* lb_token_copy(void* token) { - return token == nullptr - ? nullptr - : (void*)GRPC_MDELEM_REF(grpc_mdelem{(uintptr_t)token}).payload; -} -void lb_token_destroy(void* token) { - if (token != nullptr) { - GRPC_MDELEM_UNREF(grpc_mdelem{(uintptr_t)token}); - } -} -int lb_token_cmp(void* token1, void* token2) { - if (token1 > token2) return 1; - if (token1 < token2) return -1; - return 0; -} -const grpc_lb_user_data_vtable lb_token_vtable = { - lb_token_copy, lb_token_destroy, lb_token_cmp}; - // Returns the backend addresses extracted from the given addresses. -grpc_lb_addresses* ExtractBackendAddresses(const grpc_lb_addresses* addresses) { - // First pass: count the number of backend addresses. - size_t num_backends = 0; - for (size_t i = 0; i < addresses->num_addresses; ++i) { - if (!addresses->addresses[i].is_balancer) { - ++num_backends; +UniquePtr<ServerAddressList> ExtractBackendAddresses( + const ServerAddressList& addresses) { + auto backend_addresses = MakeUnique<ServerAddressList>(); + for (size_t i = 0; i < addresses.size(); ++i) { + if (!addresses[i].IsBalancer()) { + backend_addresses->emplace_back(addresses[i]); } } - // Second pass: actually populate the addresses and (empty) LB tokens. - grpc_lb_addresses* backend_addresses = - grpc_lb_addresses_create(num_backends, &lb_token_vtable); - size_t num_copied = 0; - for (size_t i = 0; i < addresses->num_addresses; ++i) { - if (addresses->addresses[i].is_balancer) continue; - const grpc_resolved_address* addr = &addresses->addresses[i].address; - grpc_lb_addresses_set_address(backend_addresses, num_copied, &addr->addr, - addr->len, false /* is_balancer */, - nullptr /* balancer_name */, - (void*)GRPC_MDELEM_LB_TOKEN_EMPTY.payload); - ++num_copied; - } return backend_addresses; } @@ -430,56 +395,17 @@ void ParseServer(const xds_grpclb_server* server, grpc_resolved_address* addr) { } // Returns addresses extracted from \a serverlist. -grpc_lb_addresses* ProcessServerlist(const xds_grpclb_serverlist* serverlist) { - size_t num_valid = 0; - /* first pass: count how many are valid in order to allocate the necessary - * memory in a single block */ +UniquePtr<ServerAddressList> ProcessServerlist( + const xds_grpclb_serverlist* serverlist) { + auto addresses = MakeUnique<ServerAddressList>(); for (size_t i = 0; i < serverlist->num_servers; ++i) { - if (IsServerValid(serverlist->servers[i], i, true)) ++num_valid; - } - grpc_lb_addresses* lb_addresses = - grpc_lb_addresses_create(num_valid, &lb_token_vtable); - /* second pass: actually populate the addresses and LB tokens (aka user data - * to the outside world) to be read by the child policy during its creation. - * Given that the validity tests are very cheap, they are performed again - * instead of marking the valid ones during the first pass, as this would - * incurr in an allocation due to the arbitrary number of server */ - size_t addr_idx = 0; - for (size_t sl_idx = 0; sl_idx < serverlist->num_servers; ++sl_idx) { - const xds_grpclb_server* server = serverlist->servers[sl_idx]; - if (!IsServerValid(serverlist->servers[sl_idx], sl_idx, false)) continue; - GPR_ASSERT(addr_idx < num_valid); - /* address processing */ + const xds_grpclb_server* server = serverlist->servers[i]; + if (!IsServerValid(serverlist->servers[i], i, false)) continue; grpc_resolved_address addr; ParseServer(server, &addr); - /* lb token processing */ - void* user_data; - if (server->has_load_balance_token) { - const size_t lb_token_max_length = - GPR_ARRAY_SIZE(server->load_balance_token); - const size_t lb_token_length = - strnlen(server->load_balance_token, lb_token_max_length); - grpc_slice lb_token_mdstr = grpc_slice_from_copied_buffer( - server->load_balance_token, lb_token_length); - user_data = - (void*)grpc_mdelem_from_slices(GRPC_MDSTR_LB_TOKEN, lb_token_mdstr) - .payload; - } else { - char* uri = grpc_sockaddr_to_uri(&addr); - gpr_log(GPR_INFO, - "Missing LB token for backend address '%s'. The empty token will " - "be used instead", - uri); - gpr_free(uri); - user_data = (void*)GRPC_MDELEM_LB_TOKEN_EMPTY.payload; - } - grpc_lb_addresses_set_address(lb_addresses, addr_idx, &addr.addr, addr.len, - false /* is_balancer */, - nullptr /* balancer_name */, user_data); - ++addr_idx; + addresses->emplace_back(addr, nullptr); } - GPR_ASSERT(addr_idx == num_valid); - return lb_addresses; + return addresses; } // @@ -488,7 +414,7 @@ grpc_lb_addresses* ProcessServerlist(const xds_grpclb_serverlist* serverlist) { XdsLb::BalancerCallState::BalancerCallState( RefCountedPtr<LoadBalancingPolicy> parent_xdslb_policy) - : InternallyRefCountedWithTracing<BalancerCallState>(&grpc_lb_xds_trace), + : InternallyRefCounted<BalancerCallState>(&grpc_lb_xds_trace), xdslb_policy_(std::move(parent_xdslb_policy)) { GPR_ASSERT(xdslb_policy_ != nullptr); GPR_ASSERT(!xdslb_policy()->shutting_down_); @@ -667,6 +593,7 @@ bool XdsLb::BalancerCallState::LoadReportCountersAreZero( (drop_entries == nullptr || drop_entries->empty()); } +// TODO(vpowar): Use LRS to send the client Load Report. void XdsLb::BalancerCallState::SendClientLoadReportLocked() { // Construct message payload. GPR_ASSERT(send_message_payload_ == nullptr); @@ -684,38 +611,8 @@ void XdsLb::BalancerCallState::SendClientLoadReportLocked() { } else { last_client_load_report_counters_were_zero_ = false; } - grpc_slice request_payload_slice = xds_grpclb_request_encode(request); - send_message_payload_ = - grpc_raw_byte_buffer_create(&request_payload_slice, 1); - grpc_slice_unref_internal(request_payload_slice); + // TODO(vpowar): Send the report on LRS stream. xds_grpclb_request_destroy(request); - // Send the report. - grpc_op op; - memset(&op, 0, sizeof(op)); - op.op = GRPC_OP_SEND_MESSAGE; - op.data.send_message.send_message = send_message_payload_; - GRPC_CLOSURE_INIT(&client_load_report_closure_, ClientLoadReportDoneLocked, - this, grpc_combiner_scheduler(xdslb_policy()->combiner())); - grpc_call_error call_error = grpc_call_start_batch_and_execute( - lb_call_, &op, 1, &client_load_report_closure_); - if (GPR_UNLIKELY(call_error != GRPC_CALL_OK)) { - gpr_log(GPR_ERROR, "[xdslb %p] call_error=%d", xdslb_policy_.get(), - call_error); - GPR_ASSERT(GRPC_CALL_OK == call_error); - } -} - -void XdsLb::BalancerCallState::ClientLoadReportDoneLocked(void* arg, - grpc_error* error) { - BalancerCallState* lb_calld = static_cast<BalancerCallState*>(arg); - XdsLb* xdslb_policy = lb_calld->xdslb_policy(); - grpc_byte_buffer_destroy(lb_calld->send_message_payload_); - lb_calld->send_message_payload_ = nullptr; - if (error != GRPC_ERROR_NONE || lb_calld != xdslb_policy->lb_calld_.get()) { - lb_calld->Unref(DEBUG_LOCATION, "client_load_report"); - return; - } - lb_calld->ScheduleNextClientLoadReportLocked(); } void XdsLb::BalancerCallState::OnInitialRequestSentLocked(void* arg, @@ -819,8 +716,7 @@ void XdsLb::BalancerCallState::OnBalancerMessageReceivedLocked( xds_grpclb_destroy_serverlist(xdslb_policy->serverlist_); } else { /* or dispose of the fallback */ - grpc_lb_addresses_destroy(xdslb_policy->fallback_backend_addresses_); - xdslb_policy->fallback_backend_addresses_ = nullptr; + xdslb_policy->fallback_backend_addresses_.reset(); if (xdslb_policy->fallback_timer_callback_pending_) { grpc_timer_cancel(&xdslb_policy->lb_fallback_timer_); } @@ -906,31 +802,15 @@ void XdsLb::BalancerCallState::OnBalancerStatusReceivedLocked( // helper code for creating balancer channel // -grpc_lb_addresses* ExtractBalancerAddresses( - const grpc_lb_addresses* addresses) { - size_t num_grpclb_addrs = 0; - for (size_t i = 0; i < addresses->num_addresses; ++i) { - if (addresses->addresses[i].is_balancer) ++num_grpclb_addrs; - } - // There must be at least one balancer address, or else the - // client_channel would not have chosen this LB policy. - GPR_ASSERT(num_grpclb_addrs > 0); - grpc_lb_addresses* lb_addresses = - grpc_lb_addresses_create(num_grpclb_addrs, nullptr); - size_t lb_addresses_idx = 0; - for (size_t i = 0; i < addresses->num_addresses; ++i) { - if (!addresses->addresses[i].is_balancer) continue; - if (GPR_UNLIKELY(addresses->addresses[i].user_data != nullptr)) { - gpr_log(GPR_ERROR, - "This LB policy doesn't support user data. It will be ignored"); +UniquePtr<ServerAddressList> ExtractBalancerAddresses( + const ServerAddressList& addresses) { + auto balancer_addresses = MakeUnique<ServerAddressList>(); + for (size_t i = 0; i < addresses.size(); ++i) { + if (addresses[i].IsBalancer()) { + balancer_addresses->emplace_back(addresses[i]); } - grpc_lb_addresses_set_address( - lb_addresses, lb_addresses_idx++, addresses->addresses[i].address.addr, - addresses->addresses[i].address.len, false /* is balancer */, - addresses->addresses[i].balancer_name, nullptr /* user data */); } - GPR_ASSERT(num_grpclb_addrs == lb_addresses_idx); - return lb_addresses; + return balancer_addresses; } /* Returns the channel args for the LB channel, used to create a bidirectional @@ -942,10 +822,11 @@ grpc_lb_addresses* ExtractBalancerAddresses( * above the grpclb policy. * - \a args: other args inherited from the xds policy. */ grpc_channel_args* BuildBalancerChannelArgs( - const grpc_lb_addresses* addresses, + const ServerAddressList& addresses, FakeResolverResponseGenerator* response_generator, const grpc_channel_args* args) { - grpc_lb_addresses* lb_addresses = ExtractBalancerAddresses(addresses); + UniquePtr<ServerAddressList> balancer_addresses = + ExtractBalancerAddresses(addresses); // Channel args to remove. static const char* args_to_remove[] = { // LB policy name, since we want to use the default (pick_first) in @@ -963,7 +844,7 @@ grpc_channel_args* BuildBalancerChannelArgs( // is_balancer=true. We need the LB channel to return addresses with // is_balancer=false so that it does not wind up recursively using the // xds LB policy, as per the special case logic in client_channel.c. - GRPC_ARG_LB_ADDRESSES, + GRPC_ARG_SERVER_ADDRESS_LIST, // The fake resolver response generator, because we are replacing it // with the one from the xds policy, used to propagate updates to // the LB channel. @@ -979,10 +860,10 @@ grpc_channel_args* BuildBalancerChannelArgs( }; // Channel args to add. const grpc_arg args_to_add[] = { - // New LB addresses. + // New server address list. // Note that we pass these in both when creating the LB channel // and via the fake resolver. The latter is what actually gets used. - grpc_lb_addresses_create_channel_arg(lb_addresses), + CreateServerAddressListChannelArg(balancer_addresses.get()), // The fake resolver response generator, which we use to inject // address updates into the LB channel. grpc_core::FakeResolverResponseGenerator::MakeChannelArg( @@ -1000,18 +881,15 @@ grpc_channel_args* BuildBalancerChannelArgs( args, args_to_remove, GPR_ARRAY_SIZE(args_to_remove), args_to_add, GPR_ARRAY_SIZE(args_to_add)); // Make any necessary modifications for security. - new_args = grpc_lb_policy_xds_modify_lb_channel_args(new_args); - // Clean up. - grpc_lb_addresses_destroy(lb_addresses); - return new_args; + return grpc_lb_policy_xds_modify_lb_channel_args(new_args); } // // ctor and dtor // -XdsLb::XdsLb(const grpc_lb_addresses* addresses, - const LoadBalancingPolicy::Args& args) +// TODO(vishalpowar): Use lb_config in args to configure LB policy. +XdsLb::XdsLb(const LoadBalancingPolicy::Args& args) : LoadBalancingPolicy(args), response_generator_(MakeRefCounted<FakeResolverResponseGenerator>()), lb_call_backoff_( @@ -1067,9 +945,6 @@ XdsLb::~XdsLb() { if (serverlist_ != nullptr) { xds_grpclb_destroy_serverlist(serverlist_); } - if (fallback_backend_addresses_ != nullptr) { - grpc_lb_addresses_destroy(fallback_backend_addresses_); - } grpc_subchannel_index_unref(); } @@ -1117,7 +992,6 @@ void XdsLb::HandOffPendingPicksLocked(LoadBalancingPolicy* new_policy) { while ((pp = pending_picks_) != nullptr) { pending_picks_ = pp->next; pp->pick->on_complete = pp->original_on_complete; - pp->pick->user_data = nullptr; grpc_error* error = GRPC_ERROR_NONE; if (new_policy->PickLocked(pp->pick, &error)) { // Synchronous return; schedule closure. @@ -1270,21 +1144,16 @@ void XdsLb::NotifyOnStateChangeLocked(grpc_connectivity_state* current, } void XdsLb::ProcessChannelArgsLocked(const grpc_channel_args& args) { - const grpc_arg* arg = grpc_channel_args_find(&args, GRPC_ARG_LB_ADDRESSES); - if (GPR_UNLIKELY(arg == nullptr || arg->type != GRPC_ARG_POINTER)) { + const ServerAddressList* addresses = FindServerAddressListChannelArg(&args); + if (addresses == nullptr) { // Ignore this update. gpr_log(GPR_ERROR, "[xdslb %p] No valid LB addresses channel arg in update, ignoring.", this); return; } - const grpc_lb_addresses* addresses = - static_cast<const grpc_lb_addresses*>(arg->value.pointer.p); // Update fallback address list. - if (fallback_backend_addresses_ != nullptr) { - grpc_lb_addresses_destroy(fallback_backend_addresses_); - } - fallback_backend_addresses_ = ExtractBackendAddresses(addresses); + fallback_backend_addresses_ = ExtractBackendAddresses(*addresses); // Make sure that GRPC_ARG_LB_POLICY_NAME is set in channel args, // since we use this to trigger the client_load_reporting filter. static const char* args_to_remove[] = {GRPC_ARG_LB_POLICY_NAME}; @@ -1295,7 +1164,7 @@ void XdsLb::ProcessChannelArgsLocked(const grpc_channel_args& args) { &args, args_to_remove, GPR_ARRAY_SIZE(args_to_remove), &new_arg, 1); // Construct args for balancer channel. grpc_channel_args* lb_channel_args = - BuildBalancerChannelArgs(addresses, response_generator_.get(), &args); + BuildBalancerChannelArgs(*addresses, response_generator_.get(), &args); // Create balancer channel if needed. if (lb_channel_ == nullptr) { char* uri_str; @@ -1314,7 +1183,8 @@ void XdsLb::ProcessChannelArgsLocked(const grpc_channel_args& args) { grpc_channel_args_destroy(lb_channel_args); } -void XdsLb::UpdateLocked(const grpc_channel_args& args) { +// TODO(vishalpowar): Use lb_config to configure LB policy. +void XdsLb::UpdateLocked(const grpc_channel_args& args, grpc_json* lb_config) { ProcessChannelArgsLocked(args); // Update the existing child policy. // Note: We have disabled fallback mode in the code, so this child policy must @@ -1485,37 +1355,15 @@ void XdsLb::OnBalancerChannelConnectivityChangedLocked(void* arg, // PendingPick // -// Adds lb_token of selected subchannel (address) to the call's initial -// metadata. -grpc_error* AddLbTokenToInitialMetadata( - grpc_mdelem lb_token, grpc_linked_mdelem* lb_token_mdelem_storage, - grpc_metadata_batch* initial_metadata) { - GPR_ASSERT(lb_token_mdelem_storage != nullptr); - GPR_ASSERT(!GRPC_MDISNULL(lb_token)); - return grpc_metadata_batch_add_tail(initial_metadata, lb_token_mdelem_storage, - lb_token); -} - // Destroy function used when embedding client stats in call context. void DestroyClientStats(void* arg) { static_cast<XdsLbClientStats*>(arg)->Unref(); } -void XdsLb::PendingPickSetMetadataAndContext(PendingPick* pp) { - /* if connected_subchannel is nullptr, no pick has been made by the - * child policy (e.g., all addresses failed to connect). There won't be any - * user_data/token available */ +void XdsLb::PendingPickCleanup(PendingPick* pp) { + // If connected_subchannel is nullptr, no pick has been made by the + // child policy (e.g., all addresses failed to connect). if (pp->pick->connected_subchannel != nullptr) { - if (GPR_LIKELY(!GRPC_MDISNULL(pp->lb_token))) { - AddLbTokenToInitialMetadata(GRPC_MDELEM_REF(pp->lb_token), - &pp->pick->lb_token_mdelem_storage, - pp->pick->initial_metadata); - } else { - gpr_log(GPR_ERROR, - "[xdslb %p] No LB token for connected subchannel pick %p", - pp->xdslb_policy, pp->pick); - abort(); - } // Pass on client stats via context. Passes ownership of the reference. if (pp->client_stats != nullptr) { pp->pick->subchannel_call_context[GRPC_GRPCLB_CLIENT_STATS].value = @@ -1533,7 +1381,7 @@ void XdsLb::PendingPickSetMetadataAndContext(PendingPick* pp) { * order to unref the child policy instance upon its invocation */ void XdsLb::OnPendingPickComplete(void* arg, grpc_error* error) { PendingPick* pp = static_cast<PendingPick*>(arg); - PendingPickSetMetadataAndContext(pp); + PendingPickCleanup(pp); GRPC_CLOSURE_SCHED(pp->original_on_complete, GRPC_ERROR_REF(error)); Delete(pp); } @@ -1565,16 +1413,14 @@ void XdsLb::AddPendingPick(PendingPick* pp) { // completion callback even if the pick is available immediately. bool XdsLb::PickFromChildPolicyLocked(bool force_async, PendingPick* pp, grpc_error** error) { - // Set client_stats and user_data. + // Set client_stats. if (lb_calld_ != nullptr && lb_calld_->client_stats() != nullptr) { pp->client_stats = lb_calld_->client_stats()->Ref(); } - GPR_ASSERT(pp->pick->user_data == nullptr); - pp->pick->user_data = (void**)&pp->lb_token; // Pick via the child policy. bool pick_done = child_policy_->PickLocked(pp->pick, error); if (pick_done) { - PendingPickSetMetadataAndContext(pp); + PendingPickCleanup(pp); if (force_async) { GRPC_CLOSURE_SCHED(pp->original_on_complete, *error); *error = GRPC_ERROR_NONE; @@ -1636,20 +1482,19 @@ void XdsLb::CreateChildPolicyLocked(const Args& args) { } grpc_channel_args* XdsLb::CreateChildPolicyArgsLocked() { - grpc_lb_addresses* addresses; bool is_backend_from_grpclb_load_balancer = false; // This should never be invoked if we do not have serverlist_, as fallback // mode is disabled for xDS plugin. GPR_ASSERT(serverlist_ != nullptr); GPR_ASSERT(serverlist_->num_servers > 0); - addresses = ProcessServerlist(serverlist_); - is_backend_from_grpclb_load_balancer = true; + UniquePtr<ServerAddressList> addresses = ProcessServerlist(serverlist_); GPR_ASSERT(addresses != nullptr); - // Replace the LB addresses in the channel args that we pass down to + is_backend_from_grpclb_load_balancer = true; + // Replace the server address list in the channel args that we pass down to // the subchannel. - static const char* keys_to_remove[] = {GRPC_ARG_LB_ADDRESSES}; + static const char* keys_to_remove[] = {GRPC_ARG_SERVER_ADDRESS_LIST}; const grpc_arg args_to_add[] = { - grpc_lb_addresses_create_channel_arg(addresses), + CreateServerAddressListChannelArg(addresses.get()), // A channel arg indicating if the target is a backend inferred from a // grpclb load balancer. grpc_channel_arg_integer_create( @@ -1659,7 +1504,6 @@ grpc_channel_args* XdsLb::CreateChildPolicyArgsLocked() { grpc_channel_args* args = grpc_channel_args_copy_and_add_and_remove( args_, keys_to_remove, GPR_ARRAY_SIZE(keys_to_remove), args_to_add, GPR_ARRAY_SIZE(args_to_add)); - grpc_lb_addresses_destroy(addresses); return args; } @@ -1672,7 +1516,8 @@ void XdsLb::CreateOrUpdateChildPolicyLocked() { gpr_log(GPR_INFO, "[xdslb %p] Updating the child policy %p", this, child_policy_.get()); } - child_policy_->UpdateLocked(*args); + // TODO(vishalpowar): Pass the correct LB config. + child_policy_->UpdateLocked(*args, nullptr); } else { LoadBalancingPolicy::Args lb_policy_args; lb_policy_args.combiner = combiner(); @@ -1792,22 +1637,21 @@ class XdsFactory : public LoadBalancingPolicyFactory { OrphanablePtr<LoadBalancingPolicy> CreateLoadBalancingPolicy( const LoadBalancingPolicy::Args& args) const override { /* Count the number of gRPC-LB addresses. There must be at least one. */ - const grpc_arg* arg = - grpc_channel_args_find(args.args, GRPC_ARG_LB_ADDRESSES); - if (arg == nullptr || arg->type != GRPC_ARG_POINTER) { - return nullptr; - } - grpc_lb_addresses* addresses = - static_cast<grpc_lb_addresses*>(arg->value.pointer.p); - size_t num_grpclb_addrs = 0; - for (size_t i = 0; i < addresses->num_addresses; ++i) { - if (addresses->addresses[i].is_balancer) ++num_grpclb_addrs; + const ServerAddressList* addresses = + FindServerAddressListChannelArg(args.args); + if (addresses == nullptr) return nullptr; + bool found_balancer_address = false; + for (size_t i = 0; i < addresses->size(); ++i) { + if ((*addresses)[i].IsBalancer()) { + found_balancer_address = true; + break; + } } - if (num_grpclb_addrs == 0) return nullptr; - return OrphanablePtr<LoadBalancingPolicy>(New<XdsLb>(addresses, args)); + if (!found_balancer_address) return nullptr; + return OrphanablePtr<LoadBalancingPolicy>(New<XdsLb>(args)); } - const char* name() const override { return "xds"; } + const char* name() const override { return "xds_experimental"; } }; } // namespace diff --git a/src/core/ext/filters/client_channel/lb_policy/xds/xds_channel.h b/src/core/ext/filters/client_channel/lb_policy/xds/xds_channel.h index 32c4acc8a3..f713b7f563 100644 --- a/src/core/ext/filters/client_channel/lb_policy/xds/xds_channel.h +++ b/src/core/ext/filters/client_channel/lb_policy/xds/xds_channel.h @@ -21,7 +21,7 @@ #include <grpc/support/port_platform.h> -#include "src/core/ext/filters/client_channel/lb_policy_factory.h" +#include <grpc/impl/codegen/grpc_types.h> /// Makes any necessary modifications to \a args for use in the xds /// balancer channel. diff --git a/src/core/ext/filters/client_channel/lb_policy/xds/xds_channel_secure.cc b/src/core/ext/filters/client_channel/lb_policy/xds/xds_channel_secure.cc index 5ab72efce4..55c646e6ee 100644 --- a/src/core/ext/filters/client_channel/lb_policy/xds/xds_channel_secure.cc +++ b/src/core/ext/filters/client_channel/lb_policy/xds/xds_channel_secure.cc @@ -25,6 +25,7 @@ #include <string.h> #include "src/core/ext/filters/client_channel/client_channel.h" +#include "src/core/ext/filters/client_channel/server_address.h" #include "src/core/lib/channel/channel_args.h" #include "src/core/lib/gpr/string.h" #include "src/core/lib/iomgr/sockaddr_utils.h" @@ -41,22 +42,23 @@ int BalancerNameCmp(const grpc_core::UniquePtr<char>& a, } RefCountedPtr<TargetAuthorityTable> CreateTargetAuthorityTable( - grpc_lb_addresses* addresses) { + const ServerAddressList& addresses) { TargetAuthorityTable::Entry* target_authority_entries = - static_cast<TargetAuthorityTable::Entry*>(gpr_zalloc( - sizeof(*target_authority_entries) * addresses->num_addresses)); - for (size_t i = 0; i < addresses->num_addresses; ++i) { + static_cast<TargetAuthorityTable::Entry*>( + gpr_zalloc(sizeof(*target_authority_entries) * addresses.size())); + for (size_t i = 0; i < addresses.size(); ++i) { char* addr_str; - GPR_ASSERT(grpc_sockaddr_to_string( - &addr_str, &addresses->addresses[i].address, true) > 0); + GPR_ASSERT( + grpc_sockaddr_to_string(&addr_str, &addresses[i].address(), true) > 0); target_authority_entries[i].key = grpc_slice_from_copied_string(addr_str); - target_authority_entries[i].value.reset( - gpr_strdup(addresses->addresses[i].balancer_name)); gpr_free(addr_str); + char* balancer_name = grpc_channel_arg_get_string(grpc_channel_args_find( + addresses[i].args(), GRPC_ARG_ADDRESS_BALANCER_NAME)); + target_authority_entries[i].value.reset(gpr_strdup(balancer_name)); } RefCountedPtr<TargetAuthorityTable> target_authority_table = - TargetAuthorityTable::Create(addresses->num_addresses, - target_authority_entries, BalancerNameCmp); + TargetAuthorityTable::Create(addresses.size(), target_authority_entries, + BalancerNameCmp); gpr_free(target_authority_entries); return target_authority_table; } @@ -71,13 +73,12 @@ grpc_channel_args* grpc_lb_policy_xds_modify_lb_channel_args( grpc_arg args_to_add[2]; size_t num_args_to_add = 0; // Add arg for targets info table. - const grpc_arg* arg = grpc_channel_args_find(args, GRPC_ARG_LB_ADDRESSES); - GPR_ASSERT(arg != nullptr); - GPR_ASSERT(arg->type == GRPC_ARG_POINTER); - grpc_lb_addresses* addresses = - static_cast<grpc_lb_addresses*>(arg->value.pointer.p); + grpc_core::ServerAddressList* addresses = + grpc_core::FindServerAddressListChannelArg(args); + GPR_ASSERT(addresses != nullptr); grpc_core::RefCountedPtr<grpc_core::TargetAuthorityTable> - target_authority_table = grpc_core::CreateTargetAuthorityTable(addresses); + target_authority_table = + grpc_core::CreateTargetAuthorityTable(*addresses); args_to_add[num_args_to_add++] = grpc_core::CreateTargetAuthorityTableChannelArg( target_authority_table.get()); @@ -86,22 +87,18 @@ grpc_channel_args* grpc_lb_policy_xds_modify_lb_channel_args( // bearer token credentials. grpc_channel_credentials* channel_credentials = grpc_channel_credentials_find_in_args(args); - grpc_channel_credentials* creds_sans_call_creds = nullptr; + grpc_core::RefCountedPtr<grpc_channel_credentials> creds_sans_call_creds; if (channel_credentials != nullptr) { creds_sans_call_creds = - grpc_channel_credentials_duplicate_without_call_credentials( - channel_credentials); + channel_credentials->duplicate_without_call_credentials(); GPR_ASSERT(creds_sans_call_creds != nullptr); args_to_remove[num_args_to_remove++] = GRPC_ARG_CHANNEL_CREDENTIALS; args_to_add[num_args_to_add++] = - grpc_channel_credentials_to_arg(creds_sans_call_creds); + grpc_channel_credentials_to_arg(creds_sans_call_creds.get()); } grpc_channel_args* result = grpc_channel_args_copy_and_add_and_remove( args, args_to_remove, num_args_to_remove, args_to_add, num_args_to_add); // Clean up. grpc_channel_args_destroy(args); - if (creds_sans_call_creds != nullptr) { - grpc_channel_credentials_unref(creds_sans_call_creds); - } return result; } diff --git a/src/core/ext/filters/client_channel/lb_policy/xds/xds_load_balancer_api.h b/src/core/ext/filters/client_channel/lb_policy/xds/xds_load_balancer_api.h index 9d08defa7e..6704995641 100644 --- a/src/core/ext/filters/client_channel/lb_policy/xds/xds_load_balancer_api.h +++ b/src/core/ext/filters/client_channel/lb_policy/xds/xds_load_balancer_api.h @@ -25,7 +25,7 @@ #include "src/core/ext/filters/client_channel/lb_policy/grpclb/proto/grpc/lb/v1/load_balancer.pb.h" #include "src/core/ext/filters/client_channel/lb_policy/xds/xds_client_stats.h" -#include "src/core/ext/filters/client_channel/lb_policy_factory.h" +#include "src/core/lib/iomgr/exec_ctx.h" #define XDS_SERVICE_NAME_MAX_LENGTH 128 diff --git a/src/core/ext/filters/client_channel/lb_policy_factory.cc b/src/core/ext/filters/client_channel/lb_policy_factory.cc deleted file mode 100644 index 5c6363d295..0000000000 --- a/src/core/ext/filters/client_channel/lb_policy_factory.cc +++ /dev/null @@ -1,163 +0,0 @@ -/* - * - * Copyright 2015 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 <string.h> - -#include <grpc/support/alloc.h> -#include <grpc/support/string_util.h> - -#include "src/core/lib/channel/channel_args.h" - -#include "src/core/ext/filters/client_channel/lb_policy_factory.h" -#include "src/core/ext/filters/client_channel/parse_address.h" - -grpc_lb_addresses* grpc_lb_addresses_create( - size_t num_addresses, const grpc_lb_user_data_vtable* user_data_vtable) { - grpc_lb_addresses* addresses = - static_cast<grpc_lb_addresses*>(gpr_zalloc(sizeof(grpc_lb_addresses))); - addresses->num_addresses = num_addresses; - addresses->user_data_vtable = user_data_vtable; - const size_t addresses_size = sizeof(grpc_lb_address) * num_addresses; - addresses->addresses = - static_cast<grpc_lb_address*>(gpr_zalloc(addresses_size)); - return addresses; -} - -grpc_lb_addresses* grpc_lb_addresses_copy(const grpc_lb_addresses* addresses) { - grpc_lb_addresses* new_addresses = grpc_lb_addresses_create( - addresses->num_addresses, addresses->user_data_vtable); - memcpy(new_addresses->addresses, addresses->addresses, - sizeof(grpc_lb_address) * addresses->num_addresses); - for (size_t i = 0; i < addresses->num_addresses; ++i) { - if (new_addresses->addresses[i].balancer_name != nullptr) { - new_addresses->addresses[i].balancer_name = - gpr_strdup(new_addresses->addresses[i].balancer_name); - } - if (new_addresses->addresses[i].user_data != nullptr) { - new_addresses->addresses[i].user_data = addresses->user_data_vtable->copy( - new_addresses->addresses[i].user_data); - } - } - return new_addresses; -} - -void grpc_lb_addresses_set_address(grpc_lb_addresses* addresses, size_t index, - const void* address, size_t address_len, - bool is_balancer, const char* balancer_name, - void* user_data) { - GPR_ASSERT(index < addresses->num_addresses); - if (user_data != nullptr) GPR_ASSERT(addresses->user_data_vtable != nullptr); - grpc_lb_address* target = &addresses->addresses[index]; - memcpy(target->address.addr, address, address_len); - target->address.len = static_cast<socklen_t>(address_len); - target->is_balancer = is_balancer; - target->balancer_name = gpr_strdup(balancer_name); - target->user_data = user_data; -} - -bool grpc_lb_addresses_set_address_from_uri(grpc_lb_addresses* addresses, - size_t index, const grpc_uri* uri, - bool is_balancer, - const char* balancer_name, - void* user_data) { - grpc_resolved_address address; - if (!grpc_parse_uri(uri, &address)) return false; - grpc_lb_addresses_set_address(addresses, index, address.addr, address.len, - is_balancer, balancer_name, user_data); - return true; -} - -int grpc_lb_addresses_cmp(const grpc_lb_addresses* addresses1, - const grpc_lb_addresses* addresses2) { - if (addresses1->num_addresses > addresses2->num_addresses) return 1; - if (addresses1->num_addresses < addresses2->num_addresses) return -1; - if (addresses1->user_data_vtable > addresses2->user_data_vtable) return 1; - if (addresses1->user_data_vtable < addresses2->user_data_vtable) return -1; - for (size_t i = 0; i < addresses1->num_addresses; ++i) { - const grpc_lb_address* target1 = &addresses1->addresses[i]; - const grpc_lb_address* target2 = &addresses2->addresses[i]; - if (target1->address.len > target2->address.len) return 1; - if (target1->address.len < target2->address.len) return -1; - int retval = memcmp(target1->address.addr, target2->address.addr, - target1->address.len); - if (retval != 0) return retval; - if (target1->is_balancer > target2->is_balancer) return 1; - if (target1->is_balancer < target2->is_balancer) return -1; - const char* balancer_name1 = - target1->balancer_name != nullptr ? target1->balancer_name : ""; - const char* balancer_name2 = - target2->balancer_name != nullptr ? target2->balancer_name : ""; - retval = strcmp(balancer_name1, balancer_name2); - if (retval != 0) return retval; - if (addresses1->user_data_vtable != nullptr) { - retval = addresses1->user_data_vtable->cmp(target1->user_data, - target2->user_data); - if (retval != 0) return retval; - } - } - return 0; -} - -void grpc_lb_addresses_destroy(grpc_lb_addresses* addresses) { - for (size_t i = 0; i < addresses->num_addresses; ++i) { - gpr_free(addresses->addresses[i].balancer_name); - if (addresses->addresses[i].user_data != nullptr) { - addresses->user_data_vtable->destroy(addresses->addresses[i].user_data); - } - } - gpr_free(addresses->addresses); - gpr_free(addresses); -} - -static void* lb_addresses_copy(void* addresses) { - return grpc_lb_addresses_copy(static_cast<grpc_lb_addresses*>(addresses)); -} -static void lb_addresses_destroy(void* addresses) { - grpc_lb_addresses_destroy(static_cast<grpc_lb_addresses*>(addresses)); -} -static int lb_addresses_cmp(void* addresses1, void* addresses2) { - return grpc_lb_addresses_cmp(static_cast<grpc_lb_addresses*>(addresses1), - static_cast<grpc_lb_addresses*>(addresses2)); -} -static const grpc_arg_pointer_vtable lb_addresses_arg_vtable = { - lb_addresses_copy, lb_addresses_destroy, lb_addresses_cmp}; - -grpc_arg grpc_lb_addresses_create_channel_arg( - const grpc_lb_addresses* addresses) { - return grpc_channel_arg_pointer_create( - (char*)GRPC_ARG_LB_ADDRESSES, (void*)addresses, &lb_addresses_arg_vtable); -} - -grpc_lb_addresses* grpc_lb_addresses_find_channel_arg( - const grpc_channel_args* channel_args) { - const grpc_arg* lb_addresses_arg = - grpc_channel_args_find(channel_args, GRPC_ARG_LB_ADDRESSES); - if (lb_addresses_arg == nullptr || lb_addresses_arg->type != GRPC_ARG_POINTER) - return nullptr; - return static_cast<grpc_lb_addresses*>(lb_addresses_arg->value.pointer.p); -} - -bool grpc_lb_addresses_contains_balancer_address( - const grpc_lb_addresses& addresses) { - for (size_t i = 0; i < addresses.num_addresses; ++i) { - if (addresses.addresses[i].is_balancer) return true; - } - return false; -} diff --git a/src/core/ext/filters/client_channel/lb_policy_factory.h b/src/core/ext/filters/client_channel/lb_policy_factory.h index a59deadb26..a165ebafab 100644 --- a/src/core/ext/filters/client_channel/lb_policy_factory.h +++ b/src/core/ext/filters/client_channel/lb_policy_factory.h @@ -21,91 +21,9 @@ #include <grpc/support/port_platform.h> -#include "src/core/lib/iomgr/resolve_address.h" - -#include "src/core/ext/filters/client_channel/client_channel_factory.h" #include "src/core/ext/filters/client_channel/lb_policy.h" -#include "src/core/lib/uri/uri_parser.h" - -// -// representation of an LB address -// - -// Channel arg key for grpc_lb_addresses. -#define GRPC_ARG_LB_ADDRESSES "grpc.lb_addresses" - -/** A resolved address alongside any LB related information associated with it. - * \a user_data, if not NULL, contains opaque data meant to be consumed by the - * gRPC LB policy. Note that no all LB policies support \a user_data as input. - * Those who don't will simply ignore it and will correspondingly return NULL in - * their namesake pick() output argument. */ -// TODO(roth): Once we figure out a better way of handling user_data in -// LB policies, convert these structs to C++ classes. -typedef struct grpc_lb_address { - grpc_resolved_address address; - bool is_balancer; - char* balancer_name; /* For secure naming. */ - void* user_data; -} grpc_lb_address; - -typedef struct grpc_lb_user_data_vtable { - void* (*copy)(void*); - void (*destroy)(void*); - int (*cmp)(void*, void*); -} grpc_lb_user_data_vtable; - -typedef struct grpc_lb_addresses { - size_t num_addresses; - grpc_lb_address* addresses; - const grpc_lb_user_data_vtable* user_data_vtable; -} grpc_lb_addresses; - -/** Returns a grpc_addresses struct with enough space for - \a num_addresses addresses. The \a user_data_vtable argument may be - NULL if no user data will be added. */ -grpc_lb_addresses* grpc_lb_addresses_create( - size_t num_addresses, const grpc_lb_user_data_vtable* user_data_vtable); - -/** Creates a copy of \a addresses. */ -grpc_lb_addresses* grpc_lb_addresses_copy(const grpc_lb_addresses* addresses); - -/** Sets the value of the address at index \a index of \a addresses. - * \a address is a socket address of length \a address_len. */ -void grpc_lb_addresses_set_address(grpc_lb_addresses* addresses, size_t index, - const void* address, size_t address_len, - bool is_balancer, const char* balancer_name, - void* user_data); - -/** Sets the value of the address at index \a index of \a addresses from \a uri. - * Returns true upon success, false otherwise. */ -bool grpc_lb_addresses_set_address_from_uri(grpc_lb_addresses* addresses, - size_t index, const grpc_uri* uri, - bool is_balancer, - const char* balancer_name, - void* user_data); - -/** Compares \a addresses1 and \a addresses2. */ -int grpc_lb_addresses_cmp(const grpc_lb_addresses* addresses1, - const grpc_lb_addresses* addresses2); - -/** Destroys \a addresses. */ -void grpc_lb_addresses_destroy(grpc_lb_addresses* addresses); - -/** Returns a channel arg containing \a addresses. */ -grpc_arg grpc_lb_addresses_create_channel_arg( - const grpc_lb_addresses* addresses); - -/** Returns the \a grpc_lb_addresses instance in \a channel_args or NULL */ -grpc_lb_addresses* grpc_lb_addresses_find_channel_arg( - const grpc_channel_args* channel_args); - -// Returns true if addresses contains at least one balancer address. -bool grpc_lb_addresses_contains_balancer_address( - const grpc_lb_addresses& addresses); - -// -// LB policy factory -// +#include "src/core/lib/gprpp/abstract.h" +#include "src/core/lib/gprpp/orphanable.h" namespace grpc_core { diff --git a/src/core/ext/filters/client_channel/lb_policy_registry.cc b/src/core/ext/filters/client_channel/lb_policy_registry.cc index d651b1120d..ad459c9c8c 100644 --- a/src/core/ext/filters/client_channel/lb_policy_registry.cc +++ b/src/core/ext/filters/client_channel/lb_policy_registry.cc @@ -94,4 +94,9 @@ LoadBalancingPolicyRegistry::CreateLoadBalancingPolicy( return factory->CreateLoadBalancingPolicy(args); } +bool LoadBalancingPolicyRegistry::LoadBalancingPolicyExists(const char* name) { + GPR_ASSERT(g_state != nullptr); + return g_state->GetLoadBalancingPolicyFactory(name) != nullptr; +} + } // namespace grpc_core diff --git a/src/core/ext/filters/client_channel/lb_policy_registry.h b/src/core/ext/filters/client_channel/lb_policy_registry.h index 2e9bb061ed..338f7c9f69 100644 --- a/src/core/ext/filters/client_channel/lb_policy_registry.h +++ b/src/core/ext/filters/client_channel/lb_policy_registry.h @@ -47,6 +47,10 @@ class LoadBalancingPolicyRegistry { /// Creates an LB policy of the type specified by \a name. static OrphanablePtr<LoadBalancingPolicy> CreateLoadBalancingPolicy( const char* name, const LoadBalancingPolicy::Args& args); + + /// Returns true if the LB policy factory specified by \a name exists in this + /// registry. + static bool LoadBalancingPolicyExists(const char* name); }; } // namespace grpc_core diff --git a/src/core/ext/filters/client_channel/method_params.cc b/src/core/ext/filters/client_channel/method_params.cc deleted file mode 100644 index 1f116bb67d..0000000000 --- a/src/core/ext/filters/client_channel/method_params.cc +++ /dev/null @@ -1,178 +0,0 @@ -/* - * - * Copyright 2015 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 <stdio.h> -#include <string.h> - -#include <grpc/support/alloc.h> -#include <grpc/support/log.h> -#include <grpc/support/string_util.h> - -#include "src/core/ext/filters/client_channel/method_params.h" -#include "src/core/lib/channel/status_util.h" -#include "src/core/lib/gpr/string.h" -#include "src/core/lib/gprpp/memory.h" - -// As per the retry design, we do not allow more than 5 retry attempts. -#define MAX_MAX_RETRY_ATTEMPTS 5 - -namespace grpc_core { -namespace internal { - -namespace { - -bool ParseWaitForReady( - grpc_json* field, ClientChannelMethodParams::WaitForReady* wait_for_ready) { - if (field->type != GRPC_JSON_TRUE && field->type != GRPC_JSON_FALSE) { - return false; - } - *wait_for_ready = field->type == GRPC_JSON_TRUE - ? ClientChannelMethodParams::WAIT_FOR_READY_TRUE - : ClientChannelMethodParams::WAIT_FOR_READY_FALSE; - return true; -} - -// Parses a JSON field of the form generated for a google.proto.Duration -// proto message, as per: -// https://developers.google.com/protocol-buffers/docs/proto3#json -bool ParseDuration(grpc_json* field, grpc_millis* duration) { - if (field->type != GRPC_JSON_STRING) return false; - size_t len = strlen(field->value); - if (field->value[len - 1] != 's') return false; - UniquePtr<char> buf(gpr_strdup(field->value)); - *(buf.get() + len - 1) = '\0'; // Remove trailing 's'. - char* decimal_point = strchr(buf.get(), '.'); - int nanos = 0; - if (decimal_point != nullptr) { - *decimal_point = '\0'; - nanos = gpr_parse_nonnegative_int(decimal_point + 1); - if (nanos == -1) { - return false; - } - int num_digits = static_cast<int>(strlen(decimal_point + 1)); - if (num_digits > 9) { // We don't accept greater precision than nanos. - return false; - } - for (int i = 0; i < (9 - num_digits); ++i) { - nanos *= 10; - } - } - int seconds = - decimal_point == buf.get() ? 0 : gpr_parse_nonnegative_int(buf.get()); - if (seconds == -1) return false; - *duration = seconds * GPR_MS_PER_SEC + nanos / GPR_NS_PER_MS; - return true; -} - -UniquePtr<ClientChannelMethodParams::RetryPolicy> ParseRetryPolicy( - grpc_json* field) { - auto retry_policy = MakeUnique<ClientChannelMethodParams::RetryPolicy>(); - if (field->type != GRPC_JSON_OBJECT) return nullptr; - for (grpc_json* sub_field = field->child; sub_field != nullptr; - sub_field = sub_field->next) { - if (sub_field->key == nullptr) return nullptr; - if (strcmp(sub_field->key, "maxAttempts") == 0) { - if (retry_policy->max_attempts != 0) return nullptr; // Duplicate. - if (sub_field->type != GRPC_JSON_NUMBER) return nullptr; - retry_policy->max_attempts = gpr_parse_nonnegative_int(sub_field->value); - if (retry_policy->max_attempts <= 1) return nullptr; - if (retry_policy->max_attempts > MAX_MAX_RETRY_ATTEMPTS) { - gpr_log(GPR_ERROR, - "service config: clamped retryPolicy.maxAttempts at %d", - MAX_MAX_RETRY_ATTEMPTS); - retry_policy->max_attempts = MAX_MAX_RETRY_ATTEMPTS; - } - } else if (strcmp(sub_field->key, "initialBackoff") == 0) { - if (retry_policy->initial_backoff > 0) return nullptr; // Duplicate. - if (!ParseDuration(sub_field, &retry_policy->initial_backoff)) { - return nullptr; - } - if (retry_policy->initial_backoff == 0) return nullptr; - } else if (strcmp(sub_field->key, "maxBackoff") == 0) { - if (retry_policy->max_backoff > 0) return nullptr; // Duplicate. - if (!ParseDuration(sub_field, &retry_policy->max_backoff)) { - return nullptr; - } - if (retry_policy->max_backoff == 0) return nullptr; - } else if (strcmp(sub_field->key, "backoffMultiplier") == 0) { - if (retry_policy->backoff_multiplier != 0) return nullptr; // Duplicate. - if (sub_field->type != GRPC_JSON_NUMBER) return nullptr; - if (sscanf(sub_field->value, "%f", &retry_policy->backoff_multiplier) != - 1) { - return nullptr; - } - if (retry_policy->backoff_multiplier <= 0) return nullptr; - } else if (strcmp(sub_field->key, "retryableStatusCodes") == 0) { - if (!retry_policy->retryable_status_codes.Empty()) { - return nullptr; // Duplicate. - } - if (sub_field->type != GRPC_JSON_ARRAY) return nullptr; - for (grpc_json* element = sub_field->child; element != nullptr; - element = element->next) { - if (element->type != GRPC_JSON_STRING) return nullptr; - grpc_status_code status; - if (!grpc_status_code_from_string(element->value, &status)) { - return nullptr; - } - retry_policy->retryable_status_codes.Add(status); - } - if (retry_policy->retryable_status_codes.Empty()) return nullptr; - } - } - // Make sure required fields are set. - if (retry_policy->max_attempts == 0 || retry_policy->initial_backoff == 0 || - retry_policy->max_backoff == 0 || retry_policy->backoff_multiplier == 0 || - retry_policy->retryable_status_codes.Empty()) { - return nullptr; - } - return retry_policy; -} - -} // namespace - -RefCountedPtr<ClientChannelMethodParams> -ClientChannelMethodParams::CreateFromJson(const grpc_json* json) { - RefCountedPtr<ClientChannelMethodParams> method_params = - MakeRefCounted<ClientChannelMethodParams>(); - for (grpc_json* field = json->child; field != nullptr; field = field->next) { - if (field->key == nullptr) continue; - if (strcmp(field->key, "waitForReady") == 0) { - if (method_params->wait_for_ready_ != WAIT_FOR_READY_UNSET) { - return nullptr; // Duplicate. - } - if (!ParseWaitForReady(field, &method_params->wait_for_ready_)) { - return nullptr; - } - } else if (strcmp(field->key, "timeout") == 0) { - if (method_params->timeout_ > 0) return nullptr; // Duplicate. - if (!ParseDuration(field, &method_params->timeout_)) return nullptr; - } else if (strcmp(field->key, "retryPolicy") == 0) { - if (method_params->retry_policy_ != nullptr) { - return nullptr; // Duplicate. - } - method_params->retry_policy_ = ParseRetryPolicy(field); - if (method_params->retry_policy_ == nullptr) return nullptr; - } - } - return method_params; -} - -} // namespace internal -} // namespace grpc_core diff --git a/src/core/ext/filters/client_channel/method_params.h b/src/core/ext/filters/client_channel/method_params.h deleted file mode 100644 index a31d360f17..0000000000 --- a/src/core/ext/filters/client_channel/method_params.h +++ /dev/null @@ -1,78 +0,0 @@ -/* - * - * Copyright 2015 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_CORE_EXT_FILTERS_CLIENT_CHANNEL_METHOD_PARAMS_H -#define GRPC_CORE_EXT_FILTERS_CLIENT_CHANNEL_METHOD_PARAMS_H - -#include <grpc/support/port_platform.h> - -#include "src/core/lib/channel/status_util.h" -#include "src/core/lib/gprpp/ref_counted.h" -#include "src/core/lib/gprpp/ref_counted_ptr.h" -#include "src/core/lib/iomgr/exec_ctx.h" // for grpc_millis -#include "src/core/lib/json/json.h" - -namespace grpc_core { -namespace internal { - -class ClientChannelMethodParams : public RefCounted<ClientChannelMethodParams> { - public: - enum WaitForReady { - WAIT_FOR_READY_UNSET = 0, - WAIT_FOR_READY_FALSE, - WAIT_FOR_READY_TRUE - }; - - struct RetryPolicy { - int max_attempts = 0; - grpc_millis initial_backoff = 0; - grpc_millis max_backoff = 0; - float backoff_multiplier = 0; - StatusCodeSet retryable_status_codes; - }; - - /// Creates a method_parameters object from \a json. - /// Intended for use with ServiceConfig::CreateMethodConfigTable(). - static RefCountedPtr<ClientChannelMethodParams> CreateFromJson( - const grpc_json* json); - - grpc_millis timeout() const { return timeout_; } - WaitForReady wait_for_ready() const { return wait_for_ready_; } - const RetryPolicy* retry_policy() const { return retry_policy_.get(); } - - private: - // So New() can call our private ctor. - template <typename T, typename... Args> - friend T* grpc_core::New(Args&&... args); - - // So Delete() can call our private dtor. - template <typename T> - friend void grpc_core::Delete(T*); - - ClientChannelMethodParams() {} - virtual ~ClientChannelMethodParams() {} - - grpc_millis timeout_ = 0; - WaitForReady wait_for_ready_ = WAIT_FOR_READY_UNSET; - UniquePtr<RetryPolicy> retry_policy_; -}; - -} // namespace internal -} // namespace grpc_core - -#endif /* GRPC_CORE_EXT_FILTERS_CLIENT_CHANNEL_METHOD_PARAMS_H */ diff --git a/src/core/ext/filters/client_channel/resolver.cc b/src/core/ext/filters/client_channel/resolver.cc index cd11eeb9e4..601b08be24 100644 --- a/src/core/ext/filters/client_channel/resolver.cc +++ b/src/core/ext/filters/client_channel/resolver.cc @@ -27,7 +27,7 @@ grpc_core::DebugOnlyTraceFlag grpc_trace_resolver_refcount(false, namespace grpc_core { Resolver::Resolver(grpc_combiner* combiner) - : InternallyRefCountedWithTracing(&grpc_trace_resolver_refcount), + : InternallyRefCounted(&grpc_trace_resolver_refcount), combiner_(GRPC_COMBINER_REF(combiner, "resolver")) {} Resolver::~Resolver() { GRPC_COMBINER_UNREF(combiner_, "resolver"); } diff --git a/src/core/ext/filters/client_channel/resolver.h b/src/core/ext/filters/client_channel/resolver.h index e9acbb7c41..9da849a101 100644 --- a/src/core/ext/filters/client_channel/resolver.h +++ b/src/core/ext/filters/client_channel/resolver.h @@ -44,7 +44,7 @@ namespace grpc_core { /// /// Note: All methods with a "Locked" suffix must be called from the /// combiner passed to the constructor. -class Resolver : public InternallyRefCountedWithTracing<Resolver> { +class Resolver : public InternallyRefCounted<Resolver> { public: // Not copyable nor movable. Resolver(const Resolver&) = delete; diff --git a/src/core/ext/filters/client_channel/resolver/dns/c_ares/dns_resolver_ares.cc b/src/core/ext/filters/client_channel/resolver/dns/c_ares/dns_resolver_ares.cc index 9562a3f893..abacd0c960 100644 --- a/src/core/ext/filters/client_channel/resolver/dns/c_ares/dns_resolver_ares.cc +++ b/src/core/ext/filters/client_channel/resolver/dns/c_ares/dns_resolver_ares.cc @@ -33,6 +33,7 @@ #include "src/core/ext/filters/client_channel/lb_policy_registry.h" #include "src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.h" #include "src/core/ext/filters/client_channel/resolver_registry.h" +#include "src/core/ext/filters/client_channel/server_address.h" #include "src/core/lib/backoff/backoff.h" #include "src/core/lib/channel/channel_args.h" #include "src/core/lib/gpr/env.h" @@ -117,11 +118,13 @@ class AresDnsResolver : public Resolver { /// retry backoff state BackOff backoff_; /// currently resolving addresses - grpc_lb_addresses* lb_addresses_ = nullptr; + UniquePtr<ServerAddressList> addresses_; /// currently resolving service config char* service_config_json_ = nullptr; // has shutdown been initiated bool shutdown_initiated_ = false; + // timeout in milliseconds for active DNS queries + int query_timeout_ms_; }; AresDnsResolver::AresDnsResolver(const ResolverArgs& args) @@ -159,10 +162,15 @@ AresDnsResolver::AresDnsResolver(const ResolverArgs& args) grpc_combiner_scheduler(combiner())); GRPC_CLOSURE_INIT(&on_resolved_, OnResolvedLocked, this, grpc_combiner_scheduler(combiner())); + const grpc_arg* query_timeout_ms_arg = + grpc_channel_args_find(channel_args_, GRPC_ARG_DNS_ARES_QUERY_TIMEOUT_MS); + query_timeout_ms_ = grpc_channel_arg_get_integer( + query_timeout_ms_arg, + {GRPC_DNS_ARES_DEFAULT_QUERY_TIMEOUT_MS, 0, INT_MAX}); } AresDnsResolver::~AresDnsResolver() { - gpr_log(GPR_DEBUG, "destroying AresDnsResolver"); + GRPC_CARES_TRACE_LOG("resolver:%p destroying AresDnsResolver", this); if (resolved_result_ != nullptr) { grpc_channel_args_destroy(resolved_result_); } @@ -174,7 +182,8 @@ AresDnsResolver::~AresDnsResolver() { void AresDnsResolver::NextLocked(grpc_channel_args** target_result, grpc_closure* on_complete) { - gpr_log(GPR_DEBUG, "AresDnsResolver::NextLocked() is called."); + GRPC_CARES_TRACE_LOG("resolver:%p AresDnsResolver::NextLocked() is called.", + this); GPR_ASSERT(next_completion_ == nullptr); next_completion_ = on_complete; target_result_ = target_result; @@ -217,12 +226,14 @@ void AresDnsResolver::ShutdownLocked() { void AresDnsResolver::OnNextResolutionLocked(void* arg, grpc_error* error) { AresDnsResolver* r = static_cast<AresDnsResolver*>(arg); GRPC_CARES_TRACE_LOG( - "%p re-resolution timer fired. error: %s. shutdown_initiated_: %d", r, - grpc_error_string(error), r->shutdown_initiated_); + "resolver:%p re-resolution timer fired. error: %s. shutdown_initiated_: " + "%d", + r, grpc_error_string(error), r->shutdown_initiated_); r->have_next_resolution_timer_ = false; if (error == GRPC_ERROR_NONE && !r->shutdown_initiated_) { if (!r->resolving_) { - GRPC_CARES_TRACE_LOG("%p start resolving due to re-resolution timer", r); + GRPC_CARES_TRACE_LOG( + "resolver:%p start resolving due to re-resolution timer", r); r->StartResolvingLocked(); } } @@ -307,53 +318,40 @@ void AresDnsResolver::OnResolvedLocked(void* arg, grpc_error* error) { r->resolving_ = false; gpr_free(r->pending_request_); r->pending_request_ = nullptr; - if (r->lb_addresses_ != nullptr) { - static const char* args_to_remove[2]; + if (r->addresses_ != nullptr) { + static const char* args_to_remove[1]; size_t num_args_to_remove = 0; - grpc_arg new_args[3]; + grpc_arg args_to_add[2]; size_t num_args_to_add = 0; - new_args[num_args_to_add++] = - grpc_lb_addresses_create_channel_arg(r->lb_addresses_); - grpc_core::UniquePtr<grpc_core::ServiceConfig> service_config; + args_to_add[num_args_to_add++] = + CreateServerAddressListChannelArg(r->addresses_.get()); char* service_config_string = nullptr; if (r->service_config_json_ != nullptr) { service_config_string = ChooseServiceConfig(r->service_config_json_); gpr_free(r->service_config_json_); if (service_config_string != nullptr) { - gpr_log(GPR_INFO, "selected service config choice: %s", - service_config_string); + GRPC_CARES_TRACE_LOG("resolver:%p selected service config choice: %s", + r, service_config_string); args_to_remove[num_args_to_remove++] = GRPC_ARG_SERVICE_CONFIG; - new_args[num_args_to_add++] = grpc_channel_arg_string_create( + args_to_add[num_args_to_add++] = grpc_channel_arg_string_create( (char*)GRPC_ARG_SERVICE_CONFIG, service_config_string); - service_config = - grpc_core::ServiceConfig::Create(service_config_string); - if (service_config != nullptr) { - const char* lb_policy_name = - service_config->GetLoadBalancingPolicyName(); - if (lb_policy_name != nullptr) { - args_to_remove[num_args_to_remove++] = GRPC_ARG_LB_POLICY_NAME; - new_args[num_args_to_add++] = grpc_channel_arg_string_create( - (char*)GRPC_ARG_LB_POLICY_NAME, - const_cast<char*>(lb_policy_name)); - } - } } } result = grpc_channel_args_copy_and_add_and_remove( - r->channel_args_, args_to_remove, num_args_to_remove, new_args, + r->channel_args_, args_to_remove, num_args_to_remove, args_to_add, num_args_to_add); gpr_free(service_config_string); - grpc_lb_addresses_destroy(r->lb_addresses_); + r->addresses_.reset(); // Reset backoff state so that we start from the beginning when the // next request gets triggered. r->backoff_.Reset(); } else if (!r->shutdown_initiated_) { const char* msg = grpc_error_string(error); - gpr_log(GPR_DEBUG, "dns resolution failed: %s", msg); + GRPC_CARES_TRACE_LOG("resolver:%p dns resolution failed: %s", r, msg); grpc_millis next_try = r->backoff_.NextAttemptTime(); grpc_millis timeout = next_try - ExecCtx::Get()->Now(); - gpr_log(GPR_INFO, "dns resolution failed (will retry): %s", - grpc_error_string(error)); + GRPC_CARES_TRACE_LOG("resolver:%p dns resolution failed (will retry): %s", + r, grpc_error_string(error)); GPR_ASSERT(!r->have_next_resolution_timer_); r->have_next_resolution_timer_ = true; // TODO(roth): We currently deal with this ref manually. Once the @@ -362,9 +360,10 @@ void AresDnsResolver::OnResolvedLocked(void* arg, grpc_error* error) { RefCountedPtr<Resolver> self = r->Ref(DEBUG_LOCATION, "retry-timer"); self.release(); if (timeout > 0) { - gpr_log(GPR_DEBUG, "retrying in %" PRId64 " milliseconds", timeout); + GRPC_CARES_TRACE_LOG("resolver:%p retrying in %" PRId64 " milliseconds", + r, timeout); } else { - gpr_log(GPR_DEBUG, "retrying immediately"); + GRPC_CARES_TRACE_LOG("resolver:%p retrying immediately", r); } grpc_timer_init(&r->next_resolution_timer_, next_try, &r->on_next_resolution_); @@ -390,10 +389,10 @@ void AresDnsResolver::MaybeStartResolvingLocked() { if (ms_until_next_resolution > 0) { const grpc_millis last_resolution_ago = grpc_core::ExecCtx::Get()->Now() - last_resolution_timestamp_; - gpr_log(GPR_DEBUG, - "In cooldown from last resolution (from %" PRId64 - " ms ago). Will resolve again in %" PRId64 " ms", - last_resolution_ago, ms_until_next_resolution); + GRPC_CARES_TRACE_LOG( + "resolver:%p In cooldown from last resolution (from %" PRId64 + " ms ago). Will resolve again in %" PRId64 " ms", + this, last_resolution_ago, ms_until_next_resolution); have_next_resolution_timer_ = true; // TODO(roth): We currently deal with this ref manually. Once the // new closure API is done, find a way to track this ref with the timer @@ -410,7 +409,6 @@ void AresDnsResolver::MaybeStartResolvingLocked() { } void AresDnsResolver::StartResolvingLocked() { - gpr_log(GPR_DEBUG, "Start resolving."); // TODO(roth): We currently deal with this ref manually. Once the // new closure API is done, find a way to track this ref with the timer // callback as part of the type system. @@ -418,13 +416,15 @@ void AresDnsResolver::StartResolvingLocked() { self.release(); GPR_ASSERT(!resolving_); resolving_ = true; - lb_addresses_ = nullptr; service_config_json_ = nullptr; pending_request_ = grpc_dns_lookup_ares_locked( dns_server_, name_to_resolve_, kDefaultPort, interested_parties_, - &on_resolved_, &lb_addresses_, true /* check_grpclb */, - request_service_config_ ? &service_config_json_ : nullptr, combiner()); + &on_resolved_, &addresses_, true /* check_grpclb */, + request_service_config_ ? &service_config_json_ : nullptr, + query_timeout_ms_, combiner()); last_resolution_timestamp_ = grpc_core::ExecCtx::Get()->Now(); + GRPC_CARES_TRACE_LOG("resolver:%p Started resolving. pending_request_:%p", + this, pending_request_); } void AresDnsResolver::MaybeFinishNextLocked() { @@ -432,7 +432,8 @@ void AresDnsResolver::MaybeFinishNextLocked() { *target_result_ = resolved_result_ == nullptr ? nullptr : grpc_channel_args_copy(resolved_result_); - gpr_log(GPR_DEBUG, "AresDnsResolver::MaybeFinishNextLocked()"); + GRPC_CARES_TRACE_LOG("resolver:%p AresDnsResolver::MaybeFinishNextLocked()", + this); GRPC_CLOSURE_SCHED(next_completion_, GRPC_ERROR_NONE); next_completion_ = nullptr; published_version_ = resolved_version_; @@ -470,11 +471,16 @@ static grpc_error* blocking_resolve_address_ares( static grpc_address_resolver_vtable ares_resolver = { grpc_resolve_address_ares, blocking_resolve_address_ares}; +static bool should_use_ares(const char* resolver_env) { + return resolver_env != nullptr && gpr_stricmp(resolver_env, "ares") == 0; +} + void grpc_resolver_dns_ares_init() { char* resolver_env = gpr_getenv("GRPC_DNS_RESOLVER"); /* TODO(zyc): Turn on c-ares based resolver by default after the address sorter and the CNAME support are added. */ - if (resolver_env != nullptr && gpr_stricmp(resolver_env, "ares") == 0) { + if (should_use_ares(resolver_env)) { + gpr_log(GPR_DEBUG, "Using ares dns resolver"); address_sorting_init(); grpc_error* error = grpc_ares_init(); if (error != GRPC_ERROR_NONE) { @@ -494,7 +500,7 @@ void grpc_resolver_dns_ares_init() { void grpc_resolver_dns_ares_shutdown() { char* resolver_env = gpr_getenv("GRPC_DNS_RESOLVER"); - if (resolver_env != nullptr && gpr_stricmp(resolver_env, "ares") == 0) { + if (should_use_ares(resolver_env)) { address_sorting_shutdown(); grpc_ares_cleanup(); } diff --git a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver.cc b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver.cc index fdbd07ebf5..d99c2e3004 100644 --- a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver.cc +++ b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver.cc @@ -31,8 +31,10 @@ #include <grpc/support/time.h> #include "src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.h" #include "src/core/lib/gpr/string.h" +#include "src/core/lib/iomgr/combiner.h" #include "src/core/lib/iomgr/iomgr_internal.h" #include "src/core/lib/iomgr/sockaddr_utils.h" +#include "src/core/lib/iomgr/timer.h" typedef struct fd_node { /** the owner of this fd node */ @@ -76,21 +78,30 @@ struct grpc_ares_ev_driver { grpc_ares_request* request; /** Owned by the ev_driver. Creates new GrpcPolledFd's */ grpc_core::UniquePtr<grpc_core::GrpcPolledFdFactory> polled_fd_factory; + /** query timeout in milliseconds */ + int query_timeout_ms; + /** alarm to cancel active queries */ + grpc_timer query_timeout; + /** cancels queries on a timeout */ + grpc_closure on_timeout_locked; }; static void grpc_ares_notify_on_event_locked(grpc_ares_ev_driver* ev_driver); static grpc_ares_ev_driver* grpc_ares_ev_driver_ref( grpc_ares_ev_driver* ev_driver) { - gpr_log(GPR_DEBUG, "Ref ev_driver %" PRIuPTR, (uintptr_t)ev_driver); + GRPC_CARES_TRACE_LOG("request:%p Ref ev_driver %p", ev_driver->request, + ev_driver); gpr_ref(&ev_driver->refs); return ev_driver; } static void grpc_ares_ev_driver_unref(grpc_ares_ev_driver* ev_driver) { - gpr_log(GPR_DEBUG, "Unref ev_driver %" PRIuPTR, (uintptr_t)ev_driver); + GRPC_CARES_TRACE_LOG("request:%p Unref ev_driver %p", ev_driver->request, + ev_driver); if (gpr_unref(&ev_driver->refs)) { - gpr_log(GPR_DEBUG, "destroy ev_driver %" PRIuPTR, (uintptr_t)ev_driver); + GRPC_CARES_TRACE_LOG("request:%p destroy ev_driver %p", ev_driver->request, + ev_driver); GPR_ASSERT(ev_driver->fds == nullptr); GRPC_COMBINER_UNREF(ev_driver->combiner, "free ares event driver"); ares_destroy(ev_driver->channel); @@ -100,7 +111,8 @@ static void grpc_ares_ev_driver_unref(grpc_ares_ev_driver* ev_driver) { } static void fd_node_destroy_locked(fd_node* fdn) { - gpr_log(GPR_DEBUG, "delete fd: %s", fdn->grpc_polled_fd->GetName()); + GRPC_CARES_TRACE_LOG("request:%p delete fd: %s", fdn->ev_driver->request, + fdn->grpc_polled_fd->GetName()); GPR_ASSERT(!fdn->readable_registered); GPR_ASSERT(!fdn->writable_registered); GPR_ASSERT(fdn->already_shutdown); @@ -116,8 +128,11 @@ static void fd_node_shutdown_locked(fd_node* fdn, const char* reason) { } } +static void on_timeout_locked(void* arg, grpc_error* error); + grpc_error* grpc_ares_ev_driver_create_locked(grpc_ares_ev_driver** ev_driver, grpc_pollset_set* pollset_set, + int query_timeout_ms, grpc_combiner* combiner, grpc_ares_request* request) { *ev_driver = grpc_core::New<grpc_ares_ev_driver>(); @@ -125,7 +140,7 @@ grpc_error* grpc_ares_ev_driver_create_locked(grpc_ares_ev_driver** ev_driver, memset(&opts, 0, sizeof(opts)); opts.flags |= ARES_FLAG_STAYOPEN; int status = ares_init_options(&(*ev_driver)->channel, &opts, ARES_OPT_FLAGS); - gpr_log(GPR_DEBUG, "grpc_ares_ev_driver_create_locked"); + GRPC_CARES_TRACE_LOG("request:%p grpc_ares_ev_driver_create_locked", request); if (status != ARES_SUCCESS) { char* err_msg; gpr_asprintf(&err_msg, "Failed to init ares channel. C-ares error: %s", @@ -146,6 +161,9 @@ grpc_error* grpc_ares_ev_driver_create_locked(grpc_ares_ev_driver** ev_driver, grpc_core::NewGrpcPolledFdFactory((*ev_driver)->combiner); (*ev_driver) ->polled_fd_factory->ConfigureAresChannelLocked((*ev_driver)->channel); + GRPC_CLOSURE_INIT(&(*ev_driver)->on_timeout_locked, on_timeout_locked, + *ev_driver, grpc_combiner_scheduler(combiner)); + (*ev_driver)->query_timeout_ms = query_timeout_ms; return GRPC_ERROR_NONE; } @@ -155,6 +173,7 @@ void grpc_ares_ev_driver_on_queries_complete_locked( // is working, grpc_ares_notify_on_event_locked will shut down the // fds; if it's not working, there are no fds to shut down. ev_driver->shutting_down = true; + grpc_timer_cancel(&ev_driver->query_timeout); grpc_ares_ev_driver_unref(ev_driver); } @@ -185,12 +204,25 @@ static fd_node* pop_fd_node_locked(fd_node** head, ares_socket_t as) { return nullptr; } +static void on_timeout_locked(void* arg, grpc_error* error) { + grpc_ares_ev_driver* driver = static_cast<grpc_ares_ev_driver*>(arg); + GRPC_CARES_TRACE_LOG( + "request:%p ev_driver=%p on_timeout_locked. driver->shutting_down=%d. " + "err=%s", + driver->request, driver, driver->shutting_down, grpc_error_string(error)); + if (!driver->shutting_down && error == GRPC_ERROR_NONE) { + grpc_ares_ev_driver_shutdown_locked(driver); + } + grpc_ares_ev_driver_unref(driver); +} + static void on_readable_locked(void* arg, grpc_error* error) { fd_node* fdn = static_cast<fd_node*>(arg); grpc_ares_ev_driver* ev_driver = fdn->ev_driver; const ares_socket_t as = fdn->grpc_polled_fd->GetWrappedAresSocketLocked(); fdn->readable_registered = false; - gpr_log(GPR_DEBUG, "readable on %s", fdn->grpc_polled_fd->GetName()); + GRPC_CARES_TRACE_LOG("request:%p readable on %s", fdn->ev_driver->request, + fdn->grpc_polled_fd->GetName()); if (error == GRPC_ERROR_NONE) { do { ares_process_fd(ev_driver->channel, as, ARES_SOCKET_BAD); @@ -213,7 +245,8 @@ static void on_writable_locked(void* arg, grpc_error* error) { grpc_ares_ev_driver* ev_driver = fdn->ev_driver; const ares_socket_t as = fdn->grpc_polled_fd->GetWrappedAresSocketLocked(); fdn->writable_registered = false; - gpr_log(GPR_DEBUG, "writable on %s", fdn->grpc_polled_fd->GetName()); + GRPC_CARES_TRACE_LOG("request:%p writable on %s", ev_driver->request, + fdn->grpc_polled_fd->GetName()); if (error == GRPC_ERROR_NONE) { ares_process_fd(ev_driver->channel, ARES_SOCKET_BAD, as); } else { @@ -252,7 +285,8 @@ static void grpc_ares_notify_on_event_locked(grpc_ares_ev_driver* ev_driver) { fdn->grpc_polled_fd = ev_driver->polled_fd_factory->NewGrpcPolledFdLocked( socks[i], ev_driver->pollset_set, ev_driver->combiner); - gpr_log(GPR_DEBUG, "new fd: %s", fdn->grpc_polled_fd->GetName()); + GRPC_CARES_TRACE_LOG("request:%p new fd: %s", ev_driver->request, + fdn->grpc_polled_fd->GetName()); fdn->ev_driver = ev_driver; fdn->readable_registered = false; fdn->writable_registered = false; @@ -269,8 +303,9 @@ static void grpc_ares_notify_on_event_locked(grpc_ares_ev_driver* ev_driver) { if (ARES_GETSOCK_READABLE(socks_bitmask, i) && !fdn->readable_registered) { grpc_ares_ev_driver_ref(ev_driver); - gpr_log(GPR_DEBUG, "notify read on: %s", - fdn->grpc_polled_fd->GetName()); + GRPC_CARES_TRACE_LOG("request:%p notify read on: %s", + ev_driver->request, + fdn->grpc_polled_fd->GetName()); fdn->grpc_polled_fd->RegisterForOnReadableLocked(&fdn->read_closure); fdn->readable_registered = true; } @@ -278,8 +313,9 @@ static void grpc_ares_notify_on_event_locked(grpc_ares_ev_driver* ev_driver) { // has not been registered with this socket. if (ARES_GETSOCK_WRITABLE(socks_bitmask, i) && !fdn->writable_registered) { - gpr_log(GPR_DEBUG, "notify write on: %s", - fdn->grpc_polled_fd->GetName()); + GRPC_CARES_TRACE_LOG("request:%p notify write on: %s", + ev_driver->request, + fdn->grpc_polled_fd->GetName()); grpc_ares_ev_driver_ref(ev_driver); fdn->grpc_polled_fd->RegisterForOnWriteableLocked( &fdn->write_closure); @@ -306,7 +342,8 @@ static void grpc_ares_notify_on_event_locked(grpc_ares_ev_driver* ev_driver) { // If the ev driver has no working fd, all the tasks are done. if (new_list == nullptr) { ev_driver->working = false; - gpr_log(GPR_DEBUG, "ev driver stop working"); + GRPC_CARES_TRACE_LOG("request:%p ev driver stop working", + ev_driver->request); } } @@ -314,6 +351,17 @@ void grpc_ares_ev_driver_start_locked(grpc_ares_ev_driver* ev_driver) { if (!ev_driver->working) { ev_driver->working = true; grpc_ares_notify_on_event_locked(ev_driver); + grpc_millis timeout = + ev_driver->query_timeout_ms == 0 + ? GRPC_MILLIS_INF_FUTURE + : ev_driver->query_timeout_ms + grpc_core::ExecCtx::Get()->Now(); + GRPC_CARES_TRACE_LOG( + "request:%p ev_driver=%p grpc_ares_ev_driver_start_locked. timeout in " + "%" PRId64 " ms", + ev_driver->request, ev_driver, timeout); + grpc_ares_ev_driver_ref(ev_driver); + grpc_timer_init(&ev_driver->query_timeout, timeout, + &ev_driver->on_timeout_locked); } } diff --git a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver.h b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver.h index 671c537fe7..b8cefd9470 100644 --- a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver.h +++ b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver.h @@ -43,6 +43,7 @@ ares_channel* grpc_ares_ev_driver_get_channel_locked( created successfully. */ grpc_error* grpc_ares_ev_driver_create_locked(grpc_ares_ev_driver** ev_driver, grpc_pollset_set* pollset_set, + int query_timeout_ms, grpc_combiner* combiner, grpc_ares_request* request); diff --git a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.cc b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.cc index 582e2203fc..1a7e5d0626 100644 --- a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.cc +++ b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.cc @@ -37,12 +37,16 @@ #include "src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver.h" #include "src/core/lib/gpr/host_port.h" #include "src/core/lib/gpr/string.h" +#include "src/core/lib/iomgr/combiner.h" #include "src/core/lib/iomgr/error.h" #include "src/core/lib/iomgr/executor.h" #include "src/core/lib/iomgr/iomgr_internal.h" #include "src/core/lib/iomgr/nameser.h" #include "src/core/lib/iomgr/sockaddr_utils.h" +using grpc_core::ServerAddress; +using grpc_core::ServerAddressList; + static gpr_once g_basic_init = GPR_ONCE_INIT; static gpr_mu g_init_mu; @@ -58,7 +62,7 @@ struct grpc_ares_request { /** closure to call when the request completes */ grpc_closure* on_done; /** the pointer to receive the resolved addresses */ - grpc_lb_addresses** lb_addrs_out; + grpc_core::UniquePtr<grpc_core::ServerAddressList>* addresses_out; /** the pointer to receive the service config in JSON */ char** service_config_json_out; /** the evernt driver used by this request */ @@ -87,46 +91,44 @@ typedef struct grpc_ares_hostbyname_request { static void do_basic_init(void) { gpr_mu_init(&g_init_mu); } -static void log_address_sorting_list(grpc_lb_addresses* lb_addrs, +static void log_address_sorting_list(const ServerAddressList& addresses, const char* input_output_str) { - for (size_t i = 0; i < lb_addrs->num_addresses; i++) { + for (size_t i = 0; i < addresses.size(); i++) { char* addr_str; - if (grpc_sockaddr_to_string(&addr_str, &lb_addrs->addresses[i].address, - true)) { - gpr_log(GPR_DEBUG, "c-ares address sorting: %s[%" PRIuPTR "]=%s", + if (grpc_sockaddr_to_string(&addr_str, &addresses[i].address(), true)) { + gpr_log(GPR_INFO, "c-ares address sorting: %s[%" PRIuPTR "]=%s", input_output_str, i, addr_str); gpr_free(addr_str); } else { - gpr_log(GPR_DEBUG, + gpr_log(GPR_INFO, "c-ares address sorting: %s[%" PRIuPTR "]=<unprintable>", input_output_str, i); } } } -void grpc_cares_wrapper_address_sorting_sort(grpc_lb_addresses* lb_addrs) { +void grpc_cares_wrapper_address_sorting_sort(ServerAddressList* addresses) { if (grpc_trace_cares_address_sorting.enabled()) { - log_address_sorting_list(lb_addrs, "input"); + log_address_sorting_list(*addresses, "input"); } address_sorting_sortable* sortables = (address_sorting_sortable*)gpr_zalloc( - sizeof(address_sorting_sortable) * lb_addrs->num_addresses); - for (size_t i = 0; i < lb_addrs->num_addresses; i++) { - sortables[i].user_data = &lb_addrs->addresses[i]; - memcpy(&sortables[i].dest_addr.addr, &lb_addrs->addresses[i].address.addr, - lb_addrs->addresses[i].address.len); - sortables[i].dest_addr.len = lb_addrs->addresses[i].address.len; + sizeof(address_sorting_sortable) * addresses->size()); + for (size_t i = 0; i < addresses->size(); ++i) { + sortables[i].user_data = &(*addresses)[i]; + memcpy(&sortables[i].dest_addr.addr, &(*addresses)[i].address().addr, + (*addresses)[i].address().len); + sortables[i].dest_addr.len = (*addresses)[i].address().len; } - address_sorting_rfc_6724_sort(sortables, lb_addrs->num_addresses); - grpc_lb_address* sorted_lb_addrs = (grpc_lb_address*)gpr_zalloc( - sizeof(grpc_lb_address) * lb_addrs->num_addresses); - for (size_t i = 0; i < lb_addrs->num_addresses; i++) { - sorted_lb_addrs[i] = *(grpc_lb_address*)sortables[i].user_data; + address_sorting_rfc_6724_sort(sortables, addresses->size()); + ServerAddressList sorted; + sorted.reserve(addresses->size()); + for (size_t i = 0; i < addresses->size(); ++i) { + sorted.emplace_back(*static_cast<ServerAddress*>(sortables[i].user_data)); } gpr_free(sortables); - gpr_free(lb_addrs->addresses); - lb_addrs->addresses = sorted_lb_addrs; + *addresses = std::move(sorted); if (grpc_trace_cares_address_sorting.enabled()) { - log_address_sorting_list(lb_addrs, "output"); + log_address_sorting_list(*addresses, "output"); } } @@ -145,9 +147,9 @@ void grpc_ares_complete_request_locked(grpc_ares_request* r) { /* Invoke on_done callback and destroy the request */ r->ev_driver = nullptr; - grpc_lb_addresses* lb_addrs = *(r->lb_addrs_out); - if (lb_addrs != nullptr) { - grpc_cares_wrapper_address_sorting_sort(lb_addrs); + ServerAddressList* addresses = r->addresses_out->get(); + if (addresses != nullptr) { + grpc_cares_wrapper_address_sorting_sort(addresses); } GRPC_CLOSURE_SCHED(r->on_done, r->error); } @@ -181,60 +183,53 @@ static void on_hostbyname_done_locked(void* arg, int status, int timeouts, GRPC_ERROR_UNREF(r->error); r->error = GRPC_ERROR_NONE; r->success = true; - grpc_lb_addresses** lb_addresses = r->lb_addrs_out; - if (*lb_addresses == nullptr) { - *lb_addresses = grpc_lb_addresses_create(0, nullptr); - } - size_t prev_naddr = (*lb_addresses)->num_addresses; - size_t i; - for (i = 0; hostent->h_addr_list[i] != nullptr; i++) { + if (*r->addresses_out == nullptr) { + *r->addresses_out = grpc_core::MakeUnique<ServerAddressList>(); } - (*lb_addresses)->num_addresses += i; - (*lb_addresses)->addresses = static_cast<grpc_lb_address*>( - gpr_realloc((*lb_addresses)->addresses, - sizeof(grpc_lb_address) * (*lb_addresses)->num_addresses)); - for (i = prev_naddr; i < (*lb_addresses)->num_addresses; i++) { + ServerAddressList& addresses = **r->addresses_out; + for (size_t i = 0; hostent->h_addr_list[i] != nullptr; ++i) { + grpc_core::InlinedVector<grpc_arg, 2> args_to_add; + if (hr->is_balancer) { + args_to_add.emplace_back(grpc_channel_arg_integer_create( + const_cast<char*>(GRPC_ARG_ADDRESS_IS_BALANCER), 1)); + args_to_add.emplace_back(grpc_channel_arg_string_create( + const_cast<char*>(GRPC_ARG_ADDRESS_BALANCER_NAME), hr->host)); + } + grpc_channel_args* args = grpc_channel_args_copy_and_add( + nullptr, args_to_add.data(), args_to_add.size()); switch (hostent->h_addrtype) { case AF_INET6: { size_t addr_len = sizeof(struct sockaddr_in6); struct sockaddr_in6 addr; memset(&addr, 0, addr_len); - memcpy(&addr.sin6_addr, hostent->h_addr_list[i - prev_naddr], + memcpy(&addr.sin6_addr, hostent->h_addr_list[i], sizeof(struct in6_addr)); addr.sin6_family = static_cast<unsigned char>(hostent->h_addrtype); addr.sin6_port = hr->port; - grpc_lb_addresses_set_address( - *lb_addresses, i, &addr, addr_len, - hr->is_balancer /* is_balancer */, - hr->is_balancer ? hr->host : nullptr /* balancer_name */, - nullptr /* user_data */); + addresses.emplace_back(&addr, addr_len, args); char output[INET6_ADDRSTRLEN]; ares_inet_ntop(AF_INET6, &addr.sin6_addr, output, INET6_ADDRSTRLEN); - gpr_log(GPR_DEBUG, - "c-ares resolver gets a AF_INET6 result: \n" - " addr: %s\n port: %d\n sin6_scope_id: %d\n", - output, ntohs(hr->port), addr.sin6_scope_id); + GRPC_CARES_TRACE_LOG( + "request:%p c-ares resolver gets a AF_INET6 result: \n" + " addr: %s\n port: %d\n sin6_scope_id: %d\n", + r, output, ntohs(hr->port), addr.sin6_scope_id); break; } case AF_INET: { size_t addr_len = sizeof(struct sockaddr_in); struct sockaddr_in addr; memset(&addr, 0, addr_len); - memcpy(&addr.sin_addr, hostent->h_addr_list[i - prev_naddr], + memcpy(&addr.sin_addr, hostent->h_addr_list[i], sizeof(struct in_addr)); addr.sin_family = static_cast<unsigned char>(hostent->h_addrtype); addr.sin_port = hr->port; - grpc_lb_addresses_set_address( - *lb_addresses, i, &addr, addr_len, - hr->is_balancer /* is_balancer */, - hr->is_balancer ? hr->host : nullptr /* balancer_name */, - nullptr /* user_data */); + addresses.emplace_back(&addr, addr_len, args); char output[INET_ADDRSTRLEN]; ares_inet_ntop(AF_INET, &addr.sin_addr, output, INET_ADDRSTRLEN); - gpr_log(GPR_DEBUG, - "c-ares resolver gets a AF_INET result: \n" - " addr: %s\n port: %d\n", - output, ntohs(hr->port)); + GRPC_CARES_TRACE_LOG( + "request:%p c-ares resolver gets a AF_INET result: \n" + " addr: %s\n port: %d\n", + r, output, ntohs(hr->port)); break; } } @@ -257,9 +252,9 @@ static void on_hostbyname_done_locked(void* arg, int status, int timeouts, static void on_srv_query_done_locked(void* arg, int status, int timeouts, unsigned char* abuf, int alen) { grpc_ares_request* r = static_cast<grpc_ares_request*>(arg); - gpr_log(GPR_DEBUG, "on_query_srv_done_locked"); + GRPC_CARES_TRACE_LOG("request:%p on_query_srv_done_locked", r); if (status == ARES_SUCCESS) { - gpr_log(GPR_DEBUG, "on_query_srv_done_locked ARES_SUCCESS"); + GRPC_CARES_TRACE_LOG("request:%p on_query_srv_done_locked ARES_SUCCESS", r); struct ares_srv_reply* reply; const int parse_status = ares_parse_srv_reply(abuf, alen, &reply); if (parse_status == ARES_SUCCESS) { @@ -302,9 +297,9 @@ static const char g_service_config_attribute_prefix[] = "grpc_config="; static void on_txt_done_locked(void* arg, int status, int timeouts, unsigned char* buf, int len) { - gpr_log(GPR_DEBUG, "on_txt_done_locked"); char* error_msg; grpc_ares_request* r = static_cast<grpc_ares_request*>(arg); + GRPC_CARES_TRACE_LOG("request:%p on_txt_done_locked", r); const size_t prefix_len = sizeof(g_service_config_attribute_prefix) - 1; struct ares_txt_ext* result = nullptr; struct ares_txt_ext* reply = nullptr; @@ -337,7 +332,8 @@ static void on_txt_done_locked(void* arg, int status, int timeouts, service_config_len += result->length; } (*r->service_config_json_out)[service_config_len] = '\0'; - gpr_log(GPR_INFO, "found service config: %s", *r->service_config_json_out); + GRPC_CARES_TRACE_LOG("request:%p found service config: %s", r, + *r->service_config_json_out); } // Clean up. ares_free_data(reply); @@ -359,16 +355,10 @@ done: void grpc_dns_lookup_ares_continue_after_check_localhost_and_ip_literals_locked( grpc_ares_request* r, const char* dns_server, const char* name, const char* default_port, grpc_pollset_set* interested_parties, - bool check_grpclb, grpc_combiner* combiner) { + bool check_grpclb, int query_timeout_ms, grpc_combiner* combiner) { grpc_error* error = GRPC_ERROR_NONE; grpc_ares_hostbyname_request* hr = nullptr; ares_channel* channel = nullptr; - /* TODO(zyc): Enable tracing after #9603 is checked in */ - /* if (grpc_dns_trace) { - gpr_log(GPR_DEBUG, "resolve_address (blocking): name=%s, default_port=%s", - name, default_port); - } */ - /* parse name, splitting it into host and port parts */ char* host; char* port; @@ -388,12 +378,12 @@ void grpc_dns_lookup_ares_continue_after_check_localhost_and_ip_literals_locked( port = gpr_strdup(default_port); } error = grpc_ares_ev_driver_create_locked(&r->ev_driver, interested_parties, - combiner, r); + query_timeout_ms, combiner, r); if (error != GRPC_ERROR_NONE) goto error_cleanup; channel = grpc_ares_ev_driver_get_channel_locked(r->ev_driver); // If dns_server is specified, use it. if (dns_server != nullptr) { - gpr_log(GPR_INFO, "Using DNS server %s", dns_server); + GRPC_CARES_TRACE_LOG("request:%p Using DNS server %s", r, dns_server); grpc_resolved_address addr; if (grpc_parse_ipv4_hostport(dns_server, &addr, false /* log_errors */)) { r->dns_server_addr.family = AF_INET; @@ -467,11 +457,10 @@ error_cleanup: gpr_free(port); } -static bool inner_resolve_as_ip_literal_locked(const char* name, - const char* default_port, - grpc_lb_addresses** addrs, - char** host, char** port, - char** hostport) { +static bool inner_resolve_as_ip_literal_locked( + const char* name, const char* default_port, + grpc_core::UniquePtr<grpc_core::ServerAddressList>* addrs, char** host, + char** port, char** hostport) { gpr_split_host_port(name, host, port); if (*host == nullptr) { gpr_log(GPR_ERROR, @@ -495,18 +484,16 @@ static bool inner_resolve_as_ip_literal_locked(const char* name, if (grpc_parse_ipv4_hostport(*hostport, &addr, false /* log errors */) || grpc_parse_ipv6_hostport(*hostport, &addr, false /* log errors */)) { GPR_ASSERT(*addrs == nullptr); - *addrs = grpc_lb_addresses_create(1, nullptr); - grpc_lb_addresses_set_address( - *addrs, 0, addr.addr, addr.len, false /* is_balancer */, - nullptr /* balancer_name */, nullptr /* user_data */); + *addrs = grpc_core::MakeUnique<ServerAddressList>(); + (*addrs)->emplace_back(addr.addr, addr.len, nullptr /* args */); return true; } return false; } -static bool resolve_as_ip_literal_locked(const char* name, - const char* default_port, - grpc_lb_addresses** addrs) { +static bool resolve_as_ip_literal_locked( + const char* name, const char* default_port, + grpc_core::UniquePtr<grpc_core::ServerAddressList>* addrs) { char* host = nullptr; char* port = nullptr; char* hostport = nullptr; @@ -518,20 +505,47 @@ static bool resolve_as_ip_literal_locked(const char* name, return out; } +static bool target_matches_localhost_inner(const char* name, char** host, + char** port) { + if (!gpr_split_host_port(name, host, port)) { + gpr_log(GPR_ERROR, "Unable to split host and port for name: %s", name); + return false; + } + if (gpr_stricmp(*host, "localhost") == 0) { + return true; + } else { + return false; + } +} + +static bool target_matches_localhost(const char* name) { + char* host = nullptr; + char* port = nullptr; + bool out = target_matches_localhost_inner(name, &host, &port); + gpr_free(host); + gpr_free(port); + return out; +} + static grpc_ares_request* grpc_dns_lookup_ares_locked_impl( const char* dns_server, const char* name, const char* default_port, grpc_pollset_set* interested_parties, grpc_closure* on_done, - grpc_lb_addresses** addrs, bool check_grpclb, char** service_config_json, + grpc_core::UniquePtr<grpc_core::ServerAddressList>* addrs, + bool check_grpclb, char** service_config_json, int query_timeout_ms, grpc_combiner* combiner) { grpc_ares_request* r = static_cast<grpc_ares_request*>(gpr_zalloc(sizeof(grpc_ares_request))); r->ev_driver = nullptr; r->on_done = on_done; - r->lb_addrs_out = addrs; + r->addresses_out = addrs; r->service_config_json_out = service_config_json; r->success = false; r->error = GRPC_ERROR_NONE; r->pending_queries = 0; + GRPC_CARES_TRACE_LOG( + "request:%p c-ares grpc_dns_lookup_ares_locked_impl name=%s, " + "default_port=%s", + r, name, default_port); // Early out if the target is an ipv4 or ipv6 literal. if (resolve_as_ip_literal_locked(name, default_port, addrs)) { GRPC_CLOSURE_SCHED(on_done, GRPC_ERROR_NONE); @@ -543,17 +557,25 @@ static grpc_ares_request* grpc_dns_lookup_ares_locked_impl( GRPC_CLOSURE_SCHED(on_done, GRPC_ERROR_NONE); return r; } + // Don't query for SRV and TXT records if the target is "localhost", so + // as to cut down on lookups over the network, especially in tests: + // https://github.com/grpc/proposal/pull/79 + if (target_matches_localhost(name)) { + check_grpclb = false; + r->service_config_json_out = nullptr; + } // Look up name using c-ares lib. grpc_dns_lookup_ares_continue_after_check_localhost_and_ip_literals_locked( r, dns_server, name, default_port, interested_parties, check_grpclb, - combiner); + query_timeout_ms, combiner); return r; } grpc_ares_request* (*grpc_dns_lookup_ares_locked)( const char* dns_server, const char* name, const char* default_port, grpc_pollset_set* interested_parties, grpc_closure* on_done, - grpc_lb_addresses** addrs, bool check_grpclb, char** service_config_json, + grpc_core::UniquePtr<grpc_core::ServerAddressList>* addrs, + bool check_grpclb, char** service_config_json, int query_timeout_ms, grpc_combiner* combiner) = grpc_dns_lookup_ares_locked_impl; static void grpc_cancel_ares_request_locked_impl(grpc_ares_request* r) { @@ -598,8 +620,8 @@ typedef struct grpc_resolve_address_ares_request { grpc_combiner* combiner; /** the pointer to receive the resolved addresses */ grpc_resolved_addresses** addrs_out; - /** currently resolving lb addresses */ - grpc_lb_addresses* lb_addrs; + /** currently resolving addresses */ + grpc_core::UniquePtr<ServerAddressList> addresses; /** closure to call when the resolve_address_ares request completes */ grpc_closure* on_resolve_address_done; /** a closure wrapping on_resolve_address_done, which should be invoked when @@ -612,7 +634,7 @@ typedef struct grpc_resolve_address_ares_request { /* pollset_set to be driven by */ grpc_pollset_set* interested_parties; /* underlying ares_request that the query is performed on */ - grpc_ares_request* ares_request; + grpc_ares_request* ares_request = nullptr; } grpc_resolve_address_ares_request; static void on_dns_lookup_done_locked(void* arg, grpc_error* error) { @@ -620,25 +642,24 @@ static void on_dns_lookup_done_locked(void* arg, grpc_error* error) { static_cast<grpc_resolve_address_ares_request*>(arg); gpr_free(r->ares_request); grpc_resolved_addresses** resolved_addresses = r->addrs_out; - if (r->lb_addrs == nullptr || r->lb_addrs->num_addresses == 0) { + if (r->addresses == nullptr || r->addresses->empty()) { *resolved_addresses = nullptr; } else { *resolved_addresses = static_cast<grpc_resolved_addresses*>( gpr_zalloc(sizeof(grpc_resolved_addresses))); - (*resolved_addresses)->naddrs = r->lb_addrs->num_addresses; + (*resolved_addresses)->naddrs = r->addresses->size(); (*resolved_addresses)->addrs = static_cast<grpc_resolved_address*>(gpr_zalloc( sizeof(grpc_resolved_address) * (*resolved_addresses)->naddrs)); - for (size_t i = 0; i < (*resolved_addresses)->naddrs; i++) { - GPR_ASSERT(!r->lb_addrs->addresses[i].is_balancer); - memcpy(&(*resolved_addresses)->addrs[i], - &r->lb_addrs->addresses[i].address, sizeof(grpc_resolved_address)); + for (size_t i = 0; i < (*resolved_addresses)->naddrs; ++i) { + GPR_ASSERT(!(*r->addresses)[i].IsBalancer()); + memcpy(&(*resolved_addresses)->addrs[i], &(*r->addresses)[i].address(), + sizeof(grpc_resolved_address)); } } GRPC_CLOSURE_SCHED(r->on_resolve_address_done, GRPC_ERROR_REF(error)); - if (r->lb_addrs != nullptr) grpc_lb_addresses_destroy(r->lb_addrs); GRPC_COMBINER_UNREF(r->combiner, "on_dns_lookup_done_cb"); - gpr_free(r); + grpc_core::Delete(r); } static void grpc_resolve_address_invoke_dns_lookup_ares_locked( @@ -647,8 +668,9 @@ static void grpc_resolve_address_invoke_dns_lookup_ares_locked( static_cast<grpc_resolve_address_ares_request*>(arg); r->ares_request = grpc_dns_lookup_ares_locked( nullptr /* dns_server */, r->name, r->default_port, r->interested_parties, - &r->on_dns_lookup_done_locked, &r->lb_addrs, false /* check_grpclb */, - nullptr /* service_config_json */, r->combiner); + &r->on_dns_lookup_done_locked, &r->addresses, false /* check_grpclb */, + nullptr /* service_config_json */, GRPC_DNS_ARES_DEFAULT_QUERY_TIMEOUT_MS, + r->combiner); } static void grpc_resolve_address_ares_impl(const char* name, @@ -657,8 +679,7 @@ static void grpc_resolve_address_ares_impl(const char* name, grpc_closure* on_done, grpc_resolved_addresses** addrs) { grpc_resolve_address_ares_request* r = - static_cast<grpc_resolve_address_ares_request*>( - gpr_zalloc(sizeof(grpc_resolve_address_ares_request))); + grpc_core::New<grpc_resolve_address_ares_request>(); r->combiner = grpc_combiner_create(); r->addrs_out = addrs; r->on_resolve_address_done = on_done; diff --git a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.h b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.h index a1231cc4e0..2808250456 100644 --- a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.h +++ b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.h @@ -21,11 +21,13 @@ #include <grpc/support/port_platform.h> -#include "src/core/ext/filters/client_channel/lb_policy_factory.h" +#include "src/core/ext/filters/client_channel/server_address.h" #include "src/core/lib/iomgr/iomgr.h" #include "src/core/lib/iomgr/polling_entity.h" #include "src/core/lib/iomgr/resolve_address.h" +#define GRPC_DNS_ARES_DEFAULT_QUERY_TIMEOUT_MS 10000 + extern grpc_core::TraceFlag grpc_trace_cares_address_sorting; extern grpc_core::TraceFlag grpc_trace_cares_resolver; @@ -59,8 +61,9 @@ extern void (*grpc_resolve_address_ares)(const char* name, extern grpc_ares_request* (*grpc_dns_lookup_ares_locked)( const char* dns_server, const char* name, const char* default_port, grpc_pollset_set* interested_parties, grpc_closure* on_done, - grpc_lb_addresses** addresses, bool check_grpclb, - char** service_config_json, grpc_combiner* combiner); + grpc_core::UniquePtr<grpc_core::ServerAddressList>* addresses, + bool check_grpclb, char** service_config_json, int query_timeout_ms, + grpc_combiner* combiner); /* Cancel the pending grpc_ares_request \a request */ extern void (*grpc_cancel_ares_request_locked)(grpc_ares_request* request); @@ -87,10 +90,12 @@ bool grpc_ares_query_ipv6(); * Returns a bool indicating whether or not such an action was performed. * See https://github.com/grpc/grpc/issues/15158. */ bool grpc_ares_maybe_resolve_localhost_manually_locked( - const char* name, const char* default_port, grpc_lb_addresses** addrs); + const char* name, const char* default_port, + grpc_core::UniquePtr<grpc_core::ServerAddressList>* addrs); /* Sorts destinations in lb_addrs according to RFC 6724. */ -void grpc_cares_wrapper_address_sorting_sort(grpc_lb_addresses* lb_addrs); +void grpc_cares_wrapper_address_sorting_sort( + grpc_core::ServerAddressList* addresses); #endif /* GRPC_CORE_EXT_FILTERS_CLIENT_CHANNEL_RESOLVER_DNS_C_ARES_GRPC_ARES_WRAPPER_H \ */ diff --git a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper_fallback.cc b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper_fallback.cc index 9f293c1ac0..1f4701c999 100644 --- a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper_fallback.cc +++ b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper_fallback.cc @@ -29,7 +29,8 @@ struct grpc_ares_request { static grpc_ares_request* grpc_dns_lookup_ares_locked_impl( const char* dns_server, const char* name, const char* default_port, grpc_pollset_set* interested_parties, grpc_closure* on_done, - grpc_lb_addresses** addrs, bool check_grpclb, char** service_config_json, + grpc_core::UniquePtr<grpc_core::ServerAddressList>* addrs, + bool check_grpclb, char** service_config_json, int query_timeout_ms, grpc_combiner* combiner) { return NULL; } @@ -37,7 +38,8 @@ static grpc_ares_request* grpc_dns_lookup_ares_locked_impl( grpc_ares_request* (*grpc_dns_lookup_ares_locked)( const char* dns_server, const char* name, const char* default_port, grpc_pollset_set* interested_parties, grpc_closure* on_done, - grpc_lb_addresses** addrs, bool check_grpclb, char** service_config_json, + grpc_core::UniquePtr<grpc_core::ServerAddressList>* addrs, + bool check_grpclb, char** service_config_json, int query_timeout_ms, grpc_combiner* combiner) = grpc_dns_lookup_ares_locked_impl; static void grpc_cancel_ares_request_locked_impl(grpc_ares_request* r) {} diff --git a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper_posix.cc b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper_posix.cc index 639eec2323..028d844216 100644 --- a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper_posix.cc +++ b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper_posix.cc @@ -27,7 +27,8 @@ bool grpc_ares_query_ipv6() { return grpc_ipv6_loopback_available(); } bool grpc_ares_maybe_resolve_localhost_manually_locked( - const char* name, const char* default_port, grpc_lb_addresses** addrs) { + const char* name, const char* default_port, + grpc_core::UniquePtr<grpc_core::ServerAddressList>* addrs) { return false; } diff --git a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper_windows.cc b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper_windows.cc index 7e34784691..202452f1b2 100644 --- a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper_windows.cc +++ b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper_windows.cc @@ -23,9 +23,9 @@ #include <grpc/support/string_util.h> -#include "src/core/ext/filters/client_channel/lb_policy_factory.h" #include "src/core/ext/filters/client_channel/parse_address.h" #include "src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.h" +#include "src/core/ext/filters/client_channel/server_address.h" #include "src/core/lib/gpr/host_port.h" #include "src/core/lib/gpr/string.h" #include "src/core/lib/iomgr/socket_windows.h" @@ -33,8 +33,9 @@ bool grpc_ares_query_ipv6() { return grpc_ipv6_loopback_available(); } static bool inner_maybe_resolve_localhost_manually_locked( - const char* name, const char* default_port, grpc_lb_addresses** addrs, - char** host, char** port) { + const char* name, const char* default_port, + grpc_core::UniquePtr<grpc_core::ServerAddressList>* addrs, char** host, + char** port) { gpr_split_host_port(name, host, port); if (*host == nullptr) { gpr_log(GPR_ERROR, @@ -55,7 +56,7 @@ static bool inner_maybe_resolve_localhost_manually_locked( } if (gpr_stricmp(*host, "localhost") == 0) { GPR_ASSERT(*addrs == nullptr); - *addrs = grpc_lb_addresses_create(2, nullptr); + *addrs = grpc_core::MakeUnique<grpc_core::ServerAddressList>(); uint16_t numeric_port = grpc_strhtons(*port); // Append the ipv6 loopback address. struct sockaddr_in6 ipv6_loopback_addr; @@ -63,10 +64,8 @@ static bool inner_maybe_resolve_localhost_manually_locked( ((char*)&ipv6_loopback_addr.sin6_addr)[15] = 1; ipv6_loopback_addr.sin6_family = AF_INET6; ipv6_loopback_addr.sin6_port = numeric_port; - grpc_lb_addresses_set_address( - *addrs, 0, &ipv6_loopback_addr, sizeof(ipv6_loopback_addr), - false /* is_balancer */, nullptr /* balancer_name */, - nullptr /* user_data */); + (*addrs)->emplace_back(&ipv6_loopback_addr, sizeof(ipv6_loopback_addr), + nullptr /* args */); // Append the ipv4 loopback address. struct sockaddr_in ipv4_loopback_addr; memset(&ipv4_loopback_addr, 0, sizeof(ipv4_loopback_addr)); @@ -74,19 +73,18 @@ static bool inner_maybe_resolve_localhost_manually_locked( ((char*)&ipv4_loopback_addr.sin_addr)[3] = 0x01; ipv4_loopback_addr.sin_family = AF_INET; ipv4_loopback_addr.sin_port = numeric_port; - grpc_lb_addresses_set_address( - *addrs, 1, &ipv4_loopback_addr, sizeof(ipv4_loopback_addr), - false /* is_balancer */, nullptr /* balancer_name */, - nullptr /* user_data */); + (*addrs)->emplace_back(&ipv4_loopback_addr, sizeof(ipv4_loopback_addr), + nullptr /* args */); // Let the address sorter figure out which one should be tried first. - grpc_cares_wrapper_address_sorting_sort(*addrs); + grpc_cares_wrapper_address_sorting_sort(addrs->get()); return true; } return false; } bool grpc_ares_maybe_resolve_localhost_manually_locked( - const char* name, const char* default_port, grpc_lb_addresses** addrs) { + const char* name, const char* default_port, + grpc_core::UniquePtr<grpc_core::ServerAddressList>* addrs) { char* host = nullptr; char* port = nullptr; bool out = inner_maybe_resolve_localhost_manually_locked(name, default_port, diff --git a/src/core/ext/filters/client_channel/resolver/dns/native/dns_resolver.cc b/src/core/ext/filters/client_channel/resolver/dns/native/dns_resolver.cc index 65ff1ec1a5..c365f1abfd 100644 --- a/src/core/ext/filters/client_channel/resolver/dns/native/dns_resolver.cc +++ b/src/core/ext/filters/client_channel/resolver/dns/native/dns_resolver.cc @@ -26,8 +26,8 @@ #include <grpc/support/string_util.h> #include <grpc/support/time.h> -#include "src/core/ext/filters/client_channel/lb_policy_registry.h" #include "src/core/ext/filters/client_channel/resolver_registry.h" +#include "src/core/ext/filters/client_channel/server_address.h" #include "src/core/lib/backoff/backoff.h" #include "src/core/lib/channel/channel_args.h" #include "src/core/lib/gpr/env.h" @@ -198,18 +198,14 @@ void NativeDnsResolver::OnResolvedLocked(void* arg, grpc_error* error) { grpc_error_set_str(error, GRPC_ERROR_STR_TARGET_ADDRESS, grpc_slice_from_copied_string(r->name_to_resolve_)); if (r->addresses_ != nullptr) { - grpc_lb_addresses* addresses = grpc_lb_addresses_create( - r->addresses_->naddrs, nullptr /* user_data_vtable */); + ServerAddressList addresses; for (size_t i = 0; i < r->addresses_->naddrs; ++i) { - grpc_lb_addresses_set_address( - addresses, i, &r->addresses_->addrs[i].addr, - r->addresses_->addrs[i].len, false /* is_balancer */, - nullptr /* balancer_name */, nullptr /* user_data */); + addresses.emplace_back(&r->addresses_->addrs[i].addr, + r->addresses_->addrs[i].len, nullptr /* args */); } - grpc_arg new_arg = grpc_lb_addresses_create_channel_arg(addresses); + grpc_arg new_arg = CreateServerAddressListChannelArg(&addresses); result = grpc_channel_args_copy_and_add(r->channel_args_, &new_arg, 1); grpc_resolved_addresses_destroy(r->addresses_); - grpc_lb_addresses_destroy(addresses); // Reset backoff state so that we start from the beginning when the // next request gets triggered. r->backoff_.Reset(); diff --git a/src/core/ext/filters/client_channel/resolver/fake/fake_resolver.cc b/src/core/ext/filters/client_channel/resolver/fake/fake_resolver.cc index 144ac24a56..258339491c 100644 --- a/src/core/ext/filters/client_channel/resolver/fake/fake_resolver.cc +++ b/src/core/ext/filters/client_channel/resolver/fake/fake_resolver.cc @@ -28,12 +28,13 @@ #include <grpc/support/alloc.h> #include <grpc/support/string_util.h> -#include "src/core/ext/filters/client_channel/lb_policy_factory.h" #include "src/core/ext/filters/client_channel/parse_address.h" #include "src/core/ext/filters/client_channel/resolver_registry.h" +#include "src/core/ext/filters/client_channel/server_address.h" #include "src/core/lib/channel/channel_args.h" #include "src/core/lib/gpr/host_port.h" #include "src/core/lib/gpr/string.h" +#include "src/core/lib/gpr/useful.h" #include "src/core/lib/iomgr/closure.h" #include "src/core/lib/iomgr/combiner.h" #include "src/core/lib/iomgr/resolve_address.h" @@ -103,7 +104,7 @@ void FakeResolver::NextLocked(grpc_channel_args** target_result, } void FakeResolver::RequestReresolutionLocked() { - if (reresolution_results_ != nullptr) { + if (reresolution_results_ != nullptr || return_failure_) { grpc_channel_args_destroy(next_results_); next_results_ = grpc_channel_args_copy(reresolution_results_); MaybeFinishNextLocked(); @@ -141,6 +142,7 @@ struct SetResponseClosureArg { grpc_closure set_response_closure; FakeResolverResponseGenerator* generator; grpc_channel_args* response; + bool immediate = true; }; void FakeResolverResponseGenerator::SetResponseLocked(void* arg, @@ -194,7 +196,7 @@ void FakeResolverResponseGenerator::SetFailureLocked(void* arg, SetResponseClosureArg* closure_arg = static_cast<SetResponseClosureArg*>(arg); FakeResolver* resolver = closure_arg->generator->resolver_; resolver->return_failure_ = true; - resolver->MaybeFinishNextLocked(); + if (closure_arg->immediate) resolver->MaybeFinishNextLocked(); Delete(closure_arg); } @@ -209,6 +211,18 @@ void FakeResolverResponseGenerator::SetFailure() { GRPC_ERROR_NONE); } +void FakeResolverResponseGenerator::SetFailureOnReresolution() { + GPR_ASSERT(resolver_ != nullptr); + SetResponseClosureArg* closure_arg = New<SetResponseClosureArg>(); + closure_arg->generator = this; + closure_arg->immediate = false; + GRPC_CLOSURE_SCHED( + GRPC_CLOSURE_INIT(&closure_arg->set_response_closure, SetFailureLocked, + closure_arg, + grpc_combiner_scheduler(resolver_->combiner())), + GRPC_ERROR_NONE); +} + namespace { static void* response_generator_arg_copy(void* p) { diff --git a/src/core/ext/filters/client_channel/resolver/fake/fake_resolver.h b/src/core/ext/filters/client_channel/resolver/fake/fake_resolver.h index 74a3062e7f..d86111c382 100644 --- a/src/core/ext/filters/client_channel/resolver/fake/fake_resolver.h +++ b/src/core/ext/filters/client_channel/resolver/fake/fake_resolver.h @@ -19,10 +19,9 @@ #include <grpc/support/port_platform.h> -#include "src/core/ext/filters/client_channel/lb_policy_factory.h" #include "src/core/lib/channel/channel_args.h" #include "src/core/lib/gprpp/ref_counted.h" -#include "src/core/lib/uri/uri_parser.h" +#include "src/core/lib/iomgr/error.h" #define GRPC_ARG_FAKE_RESOLVER_RESPONSE_GENERATOR \ "grpc.fake_resolver.response_generator" @@ -61,6 +60,10 @@ class FakeResolverResponseGenerator // returning a null result with no error). void SetFailure(); + // Same as SetFailure(), but instead of returning the error + // immediately, waits for the next call to RequestReresolutionLocked(). + void SetFailureOnReresolution(); + // Returns a channel arg containing \a generator. static grpc_arg MakeChannelArg(FakeResolverResponseGenerator* generator); diff --git a/src/core/ext/filters/client_channel/resolver/sockaddr/sockaddr_resolver.cc b/src/core/ext/filters/client_channel/resolver/sockaddr/sockaddr_resolver.cc index 801734764b..1654747a79 100644 --- a/src/core/ext/filters/client_channel/resolver/sockaddr/sockaddr_resolver.cc +++ b/src/core/ext/filters/client_channel/resolver/sockaddr/sockaddr_resolver.cc @@ -26,9 +26,9 @@ #include <grpc/support/alloc.h> #include <grpc/support/string_util.h> -#include "src/core/ext/filters/client_channel/lb_policy_factory.h" #include "src/core/ext/filters/client_channel/parse_address.h" #include "src/core/ext/filters/client_channel/resolver_registry.h" +#include "src/core/ext/filters/client_channel/server_address.h" #include "src/core/lib/channel/channel_args.h" #include "src/core/lib/gpr/host_port.h" #include "src/core/lib/gpr/string.h" @@ -45,7 +45,8 @@ namespace { class SockaddrResolver : public Resolver { public: /// Takes ownership of \a addresses. - SockaddrResolver(const ResolverArgs& args, grpc_lb_addresses* addresses); + SockaddrResolver(const ResolverArgs& args, + UniquePtr<ServerAddressList> addresses); void NextLocked(grpc_channel_args** result, grpc_closure* on_complete) override; @@ -58,7 +59,7 @@ class SockaddrResolver : public Resolver { void MaybeFinishNextLocked(); /// the addresses that we've "resolved" - grpc_lb_addresses* addresses_ = nullptr; + UniquePtr<ServerAddressList> addresses_; /// channel args grpc_channel_args* channel_args_ = nullptr; /// have we published? @@ -70,13 +71,12 @@ class SockaddrResolver : public Resolver { }; SockaddrResolver::SockaddrResolver(const ResolverArgs& args, - grpc_lb_addresses* addresses) + UniquePtr<ServerAddressList> addresses) : Resolver(args.combiner), - addresses_(addresses), + addresses_(std::move(addresses)), channel_args_(grpc_channel_args_copy(args.args)) {} SockaddrResolver::~SockaddrResolver() { - grpc_lb_addresses_destroy(addresses_); grpc_channel_args_destroy(channel_args_); } @@ -100,7 +100,7 @@ void SockaddrResolver::ShutdownLocked() { void SockaddrResolver::MaybeFinishNextLocked() { if (next_completion_ != nullptr && !published_) { published_ = true; - grpc_arg arg = grpc_lb_addresses_create_channel_arg(addresses_); + grpc_arg arg = CreateServerAddressListChannelArg(addresses_.get()); *target_result_ = grpc_channel_args_copy_and_add(channel_args_, &arg, 1); GRPC_CLOSURE_SCHED(next_completion_, GRPC_ERROR_NONE); next_completion_ = nullptr; @@ -127,27 +127,27 @@ OrphanablePtr<Resolver> CreateSockaddrResolver( grpc_slice_buffer path_parts; grpc_slice_buffer_init(&path_parts); grpc_slice_split(path_slice, ",", &path_parts); - grpc_lb_addresses* addresses = grpc_lb_addresses_create( - path_parts.count, nullptr /* user_data_vtable */); + auto addresses = MakeUnique<ServerAddressList>(); bool errors_found = false; - for (size_t i = 0; i < addresses->num_addresses; i++) { + for (size_t i = 0; i < path_parts.count; i++) { grpc_uri ith_uri = *args.uri; - char* part_str = grpc_slice_to_c_string(path_parts.slices[i]); - ith_uri.path = part_str; - if (!parse(&ith_uri, &addresses->addresses[i].address)) { + UniquePtr<char> part_str(grpc_slice_to_c_string(path_parts.slices[i])); + ith_uri.path = part_str.get(); + grpc_resolved_address addr; + if (!parse(&ith_uri, &addr)) { errors_found = true; /* GPR_TRUE */ + break; } - gpr_free(part_str); - if (errors_found) break; + addresses->emplace_back(addr, nullptr /* args */); } grpc_slice_buffer_destroy_internal(&path_parts); grpc_slice_unref_internal(path_slice); if (errors_found) { - grpc_lb_addresses_destroy(addresses); return OrphanablePtr<Resolver>(nullptr); } // Instantiate resolver. - return OrphanablePtr<Resolver>(New<SockaddrResolver>(args, addresses)); + return OrphanablePtr<Resolver>( + New<SockaddrResolver>(args, std::move(addresses))); } class IPv4ResolverFactory : public ResolverFactory { diff --git a/src/core/ext/filters/client_channel/resolver_result_parsing.cc b/src/core/ext/filters/client_channel/resolver_result_parsing.cc new file mode 100644 index 0000000000..22b06db45c --- /dev/null +++ b/src/core/ext/filters/client_channel/resolver_result_parsing.cc @@ -0,0 +1,377 @@ +/* + * + * Copyright 2018 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/ext/filters/client_channel/resolver_result_parsing.h" + +#include <ctype.h> +#include <stdio.h> +#include <string.h> + +#include <grpc/support/alloc.h> +#include <grpc/support/log.h> +#include <grpc/support/string_util.h> + +#include "src/core/ext/filters/client_channel/client_channel.h" +#include "src/core/ext/filters/client_channel/lb_policy_registry.h" +#include "src/core/ext/filters/client_channel/server_address.h" +#include "src/core/lib/channel/status_util.h" +#include "src/core/lib/gpr/string.h" +#include "src/core/lib/gprpp/memory.h" +#include "src/core/lib/uri/uri_parser.h" + +// As per the retry design, we do not allow more than 5 retry attempts. +#define MAX_MAX_RETRY_ATTEMPTS 5 + +namespace grpc_core { +namespace internal { + +ProcessedResolverResult::ProcessedResolverResult( + const grpc_channel_args* resolver_result, bool parse_retry) { + ProcessServiceConfig(resolver_result, parse_retry); + // If no LB config was found above, just find the LB policy name then. + if (lb_policy_name_ == nullptr) ProcessLbPolicyName(resolver_result); +} + +void ProcessedResolverResult::ProcessServiceConfig( + const grpc_channel_args* resolver_result, bool parse_retry) { + const grpc_arg* channel_arg = + grpc_channel_args_find(resolver_result, GRPC_ARG_SERVICE_CONFIG); + const char* service_config_json = grpc_channel_arg_get_string(channel_arg); + if (service_config_json != nullptr) { + service_config_json_.reset(gpr_strdup(service_config_json)); + service_config_ = grpc_core::ServiceConfig::Create(service_config_json); + if (service_config_ != nullptr) { + if (parse_retry) { + channel_arg = + grpc_channel_args_find(resolver_result, GRPC_ARG_SERVER_URI); + const char* server_uri = grpc_channel_arg_get_string(channel_arg); + GPR_ASSERT(server_uri != nullptr); + grpc_uri* uri = grpc_uri_parse(server_uri, true); + GPR_ASSERT(uri->path[0] != '\0'); + server_name_ = uri->path[0] == '/' ? uri->path + 1 : uri->path; + service_config_->ParseGlobalParams(ParseServiceConfig, this); + grpc_uri_destroy(uri); + } else { + service_config_->ParseGlobalParams(ParseServiceConfig, this); + } + method_params_table_ = service_config_->CreateMethodConfigTable( + ClientChannelMethodParams::CreateFromJson); + } + } +} + +void ProcessedResolverResult::ProcessLbPolicyName( + const grpc_channel_args* resolver_result) { + // Prefer the LB policy name found in the service config. Note that this is + // checking the deprecated loadBalancingPolicy field, rather than the new + // loadBalancingConfig field. + if (service_config_ != nullptr) { + lb_policy_name_.reset( + gpr_strdup(service_config_->GetLoadBalancingPolicyName())); + // Convert to lower-case. + if (lb_policy_name_ != nullptr) { + char* lb_policy_name = lb_policy_name_.get(); + for (size_t i = 0; i < strlen(lb_policy_name); ++i) { + lb_policy_name[i] = tolower(lb_policy_name[i]); + } + } + } + // Otherwise, find the LB policy name set by the client API. + if (lb_policy_name_ == nullptr) { + const grpc_arg* channel_arg = + grpc_channel_args_find(resolver_result, GRPC_ARG_LB_POLICY_NAME); + lb_policy_name_.reset(gpr_strdup(grpc_channel_arg_get_string(channel_arg))); + } + // Special case: If at least one balancer address is present, we use + // the grpclb policy, regardless of what the resolver has returned. + const ServerAddressList* addresses = + FindServerAddressListChannelArg(resolver_result); + if (addresses != nullptr) { + bool found_balancer_address = false; + for (size_t i = 0; i < addresses->size(); ++i) { + const ServerAddress& address = (*addresses)[i]; + if (address.IsBalancer()) { + found_balancer_address = true; + break; + } + } + if (found_balancer_address) { + if (lb_policy_name_ != nullptr && + strcmp(lb_policy_name_.get(), "grpclb") != 0) { + gpr_log(GPR_INFO, + "resolver requested LB policy %s but provided at least one " + "balancer address -- forcing use of grpclb LB policy", + lb_policy_name_.get()); + } + lb_policy_name_.reset(gpr_strdup("grpclb")); + } + } + // Use pick_first if nothing was specified and we didn't select grpclb + // above. + if (lb_policy_name_ == nullptr) { + lb_policy_name_.reset(gpr_strdup("pick_first")); + } +} + +void ProcessedResolverResult::ParseServiceConfig( + const grpc_json* field, ProcessedResolverResult* parsing_state) { + parsing_state->ParseLbConfigFromServiceConfig(field); + if (parsing_state->server_name_ != nullptr) { + parsing_state->ParseRetryThrottleParamsFromServiceConfig(field); + } +} + +void ProcessedResolverResult::ParseLbConfigFromServiceConfig( + const grpc_json* field) { + if (lb_policy_config_ != nullptr) return; // Already found. + // Find the LB config global parameter. + if (field->key == nullptr || strcmp(field->key, "loadBalancingConfig") != 0 || + field->type != GRPC_JSON_ARRAY) { + return; // Not valid lb config array. + } + // Find the first LB policy that this client supports. + for (grpc_json* lb_config = field->child; lb_config != nullptr; + lb_config = lb_config->next) { + if (lb_config->type != GRPC_JSON_OBJECT) return; + // Find the policy object. + grpc_json* policy = nullptr; + for (grpc_json* field = lb_config->child; field != nullptr; + field = field->next) { + if (field->key == nullptr || strcmp(field->key, "policy") != 0 || + field->type != GRPC_JSON_OBJECT) { + return; + } + if (policy != nullptr) return; // Duplicate. + policy = field; + } + // Find the specific policy content since the policy object is of type + // "oneof". + grpc_json* policy_content = nullptr; + for (grpc_json* field = policy->child; field != nullptr; + field = field->next) { + if (field->key == nullptr || field->type != GRPC_JSON_OBJECT) return; + if (policy_content != nullptr) return; // Violate "oneof" type. + policy_content = field; + } + // If we support this policy, then select it. + if (grpc_core::LoadBalancingPolicyRegistry::LoadBalancingPolicyExists( + policy_content->key)) { + lb_policy_name_.reset(gpr_strdup(policy_content->key)); + lb_policy_config_ = policy_content->child; + return; + } + } +} + +void ProcessedResolverResult::ParseRetryThrottleParamsFromServiceConfig( + const grpc_json* field) { + if (strcmp(field->key, "retryThrottling") == 0) { + if (retry_throttle_data_ != nullptr) return; // Duplicate. + if (field->type != GRPC_JSON_OBJECT) return; + int max_milli_tokens = 0; + int milli_token_ratio = 0; + for (grpc_json* sub_field = field->child; sub_field != nullptr; + sub_field = sub_field->next) { + if (sub_field->key == nullptr) return; + if (strcmp(sub_field->key, "maxTokens") == 0) { + if (max_milli_tokens != 0) return; // Duplicate. + if (sub_field->type != GRPC_JSON_NUMBER) return; + max_milli_tokens = gpr_parse_nonnegative_int(sub_field->value); + if (max_milli_tokens == -1) return; + max_milli_tokens *= 1000; + } else if (strcmp(sub_field->key, "tokenRatio") == 0) { + if (milli_token_ratio != 0) return; // Duplicate. + if (sub_field->type != GRPC_JSON_NUMBER) return; + // We support up to 3 decimal digits. + size_t whole_len = strlen(sub_field->value); + uint32_t multiplier = 1; + uint32_t decimal_value = 0; + const char* decimal_point = strchr(sub_field->value, '.'); + if (decimal_point != nullptr) { + whole_len = static_cast<size_t>(decimal_point - sub_field->value); + multiplier = 1000; + size_t decimal_len = strlen(decimal_point + 1); + if (decimal_len > 3) decimal_len = 3; + if (!gpr_parse_bytes_to_uint32(decimal_point + 1, decimal_len, + &decimal_value)) { + return; + } + uint32_t decimal_multiplier = 1; + for (size_t i = 0; i < (3 - decimal_len); ++i) { + decimal_multiplier *= 10; + } + decimal_value *= decimal_multiplier; + } + uint32_t whole_value; + if (!gpr_parse_bytes_to_uint32(sub_field->value, whole_len, + &whole_value)) { + return; + } + milli_token_ratio = + static_cast<int>((whole_value * multiplier) + decimal_value); + if (milli_token_ratio <= 0) return; + } + } + retry_throttle_data_ = + grpc_core::internal::ServerRetryThrottleMap::GetDataForServer( + server_name_, max_milli_tokens, milli_token_ratio); + } +} + +namespace { + +bool ParseWaitForReady( + grpc_json* field, ClientChannelMethodParams::WaitForReady* wait_for_ready) { + if (field->type != GRPC_JSON_TRUE && field->type != GRPC_JSON_FALSE) { + return false; + } + *wait_for_ready = field->type == GRPC_JSON_TRUE + ? ClientChannelMethodParams::WAIT_FOR_READY_TRUE + : ClientChannelMethodParams::WAIT_FOR_READY_FALSE; + return true; +} + +// Parses a JSON field of the form generated for a google.proto.Duration +// proto message, as per: +// https://developers.google.com/protocol-buffers/docs/proto3#json +bool ParseDuration(grpc_json* field, grpc_millis* duration) { + if (field->type != GRPC_JSON_STRING) return false; + size_t len = strlen(field->value); + if (field->value[len - 1] != 's') return false; + UniquePtr<char> buf(gpr_strdup(field->value)); + *(buf.get() + len - 1) = '\0'; // Remove trailing 's'. + char* decimal_point = strchr(buf.get(), '.'); + int nanos = 0; + if (decimal_point != nullptr) { + *decimal_point = '\0'; + nanos = gpr_parse_nonnegative_int(decimal_point + 1); + if (nanos == -1) { + return false; + } + int num_digits = static_cast<int>(strlen(decimal_point + 1)); + if (num_digits > 9) { // We don't accept greater precision than nanos. + return false; + } + for (int i = 0; i < (9 - num_digits); ++i) { + nanos *= 10; + } + } + int seconds = + decimal_point == buf.get() ? 0 : gpr_parse_nonnegative_int(buf.get()); + if (seconds == -1) return false; + *duration = seconds * GPR_MS_PER_SEC + nanos / GPR_NS_PER_MS; + return true; +} + +UniquePtr<ClientChannelMethodParams::RetryPolicy> ParseRetryPolicy( + grpc_json* field) { + auto retry_policy = MakeUnique<ClientChannelMethodParams::RetryPolicy>(); + if (field->type != GRPC_JSON_OBJECT) return nullptr; + for (grpc_json* sub_field = field->child; sub_field != nullptr; + sub_field = sub_field->next) { + if (sub_field->key == nullptr) return nullptr; + if (strcmp(sub_field->key, "maxAttempts") == 0) { + if (retry_policy->max_attempts != 0) return nullptr; // Duplicate. + if (sub_field->type != GRPC_JSON_NUMBER) return nullptr; + retry_policy->max_attempts = gpr_parse_nonnegative_int(sub_field->value); + if (retry_policy->max_attempts <= 1) return nullptr; + if (retry_policy->max_attempts > MAX_MAX_RETRY_ATTEMPTS) { + gpr_log(GPR_ERROR, + "service config: clamped retryPolicy.maxAttempts at %d", + MAX_MAX_RETRY_ATTEMPTS); + retry_policy->max_attempts = MAX_MAX_RETRY_ATTEMPTS; + } + } else if (strcmp(sub_field->key, "initialBackoff") == 0) { + if (retry_policy->initial_backoff > 0) return nullptr; // Duplicate. + if (!ParseDuration(sub_field, &retry_policy->initial_backoff)) { + return nullptr; + } + if (retry_policy->initial_backoff == 0) return nullptr; + } else if (strcmp(sub_field->key, "maxBackoff") == 0) { + if (retry_policy->max_backoff > 0) return nullptr; // Duplicate. + if (!ParseDuration(sub_field, &retry_policy->max_backoff)) { + return nullptr; + } + if (retry_policy->max_backoff == 0) return nullptr; + } else if (strcmp(sub_field->key, "backoffMultiplier") == 0) { + if (retry_policy->backoff_multiplier != 0) return nullptr; // Duplicate. + if (sub_field->type != GRPC_JSON_NUMBER) return nullptr; + if (sscanf(sub_field->value, "%f", &retry_policy->backoff_multiplier) != + 1) { + return nullptr; + } + if (retry_policy->backoff_multiplier <= 0) return nullptr; + } else if (strcmp(sub_field->key, "retryableStatusCodes") == 0) { + if (!retry_policy->retryable_status_codes.Empty()) { + return nullptr; // Duplicate. + } + if (sub_field->type != GRPC_JSON_ARRAY) return nullptr; + for (grpc_json* element = sub_field->child; element != nullptr; + element = element->next) { + if (element->type != GRPC_JSON_STRING) return nullptr; + grpc_status_code status; + if (!grpc_status_code_from_string(element->value, &status)) { + return nullptr; + } + retry_policy->retryable_status_codes.Add(status); + } + if (retry_policy->retryable_status_codes.Empty()) return nullptr; + } + } + // Make sure required fields are set. + if (retry_policy->max_attempts == 0 || retry_policy->initial_backoff == 0 || + retry_policy->max_backoff == 0 || retry_policy->backoff_multiplier == 0 || + retry_policy->retryable_status_codes.Empty()) { + return nullptr; + } + return retry_policy; +} + +} // namespace + +RefCountedPtr<ClientChannelMethodParams> +ClientChannelMethodParams::CreateFromJson(const grpc_json* json) { + RefCountedPtr<ClientChannelMethodParams> method_params = + MakeRefCounted<ClientChannelMethodParams>(); + for (grpc_json* field = json->child; field != nullptr; field = field->next) { + if (field->key == nullptr) continue; + if (strcmp(field->key, "waitForReady") == 0) { + if (method_params->wait_for_ready_ != WAIT_FOR_READY_UNSET) { + return nullptr; // Duplicate. + } + if (!ParseWaitForReady(field, &method_params->wait_for_ready_)) { + return nullptr; + } + } else if (strcmp(field->key, "timeout") == 0) { + if (method_params->timeout_ > 0) return nullptr; // Duplicate. + if (!ParseDuration(field, &method_params->timeout_)) return nullptr; + } else if (strcmp(field->key, "retryPolicy") == 0) { + if (method_params->retry_policy_ != nullptr) { + return nullptr; // Duplicate. + } + method_params->retry_policy_ = ParseRetryPolicy(field); + if (method_params->retry_policy_ == nullptr) return nullptr; + } + } + return method_params; +} + +} // namespace internal +} // namespace grpc_core diff --git a/src/core/ext/filters/client_channel/resolver_result_parsing.h b/src/core/ext/filters/client_channel/resolver_result_parsing.h new file mode 100644 index 0000000000..f1fb7406bc --- /dev/null +++ b/src/core/ext/filters/client_channel/resolver_result_parsing.h @@ -0,0 +1,146 @@ +/* + * + * Copyright 2018 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_CORE_EXT_FILTERS_CLIENT_CHANNEL_RESOLVER_RESULT_PARSING_H +#define GRPC_CORE_EXT_FILTERS_CLIENT_CHANNEL_RESOLVER_RESULT_PARSING_H + +#include <grpc/support/port_platform.h> + +#include "src/core/ext/filters/client_channel/retry_throttle.h" +#include "src/core/lib/channel/status_util.h" +#include "src/core/lib/gprpp/ref_counted.h" +#include "src/core/lib/gprpp/ref_counted_ptr.h" +#include "src/core/lib/iomgr/exec_ctx.h" // for grpc_millis +#include "src/core/lib/json/json.h" +#include "src/core/lib/slice/slice_hash_table.h" +#include "src/core/lib/transport/service_config.h" + +namespace grpc_core { +namespace internal { + +class ClientChannelMethodParams; + +// A table mapping from a method name to its method parameters. +typedef grpc_core::SliceHashTable< + grpc_core::RefCountedPtr<ClientChannelMethodParams>> + ClientChannelMethodParamsTable; + +// A container of processed fields from the resolver result. Simplifies the +// usage of resolver result. +class ProcessedResolverResult { + public: + // Processes the resolver result and populates the relative members + // for later consumption. Tries to parse retry parameters only if parse_retry + // is true. + ProcessedResolverResult(const grpc_channel_args* resolver_result, + bool parse_retry); + + // Getters. Any managed object's ownership is transferred. + grpc_core::UniquePtr<char> service_config_json() { + return std::move(service_config_json_); + } + grpc_core::RefCountedPtr<ServerRetryThrottleData> retry_throttle_data() { + return std::move(retry_throttle_data_); + } + grpc_core::RefCountedPtr<ClientChannelMethodParamsTable> + method_params_table() { + return std::move(method_params_table_); + } + grpc_core::UniquePtr<char> lb_policy_name() { + return std::move(lb_policy_name_); + } + grpc_json* lb_policy_config() { return lb_policy_config_; } + + private: + // Finds the service config; extracts LB config and (maybe) retry throttle + // params from it. + void ProcessServiceConfig(const grpc_channel_args* resolver_result, + bool parse_retry); + + // Finds the LB policy name (when no LB config was found). + void ProcessLbPolicyName(const grpc_channel_args* resolver_result); + + // Parses the service config. Intended to be used by + // ServiceConfig::ParseGlobalParams. + static void ParseServiceConfig(const grpc_json* field, + ProcessedResolverResult* parsing_state); + // Parses the LB config from service config. + void ParseLbConfigFromServiceConfig(const grpc_json* field); + // Parses the retry throttle parameters from service config. + void ParseRetryThrottleParamsFromServiceConfig(const grpc_json* field); + + // Service config. + grpc_core::UniquePtr<char> service_config_json_; + grpc_core::UniquePtr<grpc_core::ServiceConfig> service_config_; + // LB policy. + grpc_json* lb_policy_config_ = nullptr; + grpc_core::UniquePtr<char> lb_policy_name_; + // Retry throttle data. + char* server_name_ = nullptr; + grpc_core::RefCountedPtr<ServerRetryThrottleData> retry_throttle_data_; + // Method params table. + grpc_core::RefCountedPtr<ClientChannelMethodParamsTable> method_params_table_; +}; + +// The parameters of a method. +class ClientChannelMethodParams : public RefCounted<ClientChannelMethodParams> { + public: + enum WaitForReady { + WAIT_FOR_READY_UNSET = 0, + WAIT_FOR_READY_FALSE, + WAIT_FOR_READY_TRUE + }; + + struct RetryPolicy { + int max_attempts = 0; + grpc_millis initial_backoff = 0; + grpc_millis max_backoff = 0; + float backoff_multiplier = 0; + StatusCodeSet retryable_status_codes; + }; + + /// Creates a method_parameters object from \a json. + /// Intended for use with ServiceConfig::CreateMethodConfigTable(). + static RefCountedPtr<ClientChannelMethodParams> CreateFromJson( + const grpc_json* json); + + grpc_millis timeout() const { return timeout_; } + WaitForReady wait_for_ready() const { return wait_for_ready_; } + const RetryPolicy* retry_policy() const { return retry_policy_.get(); } + + private: + // So New() can call our private ctor. + template <typename T, typename... Args> + friend T* grpc_core::New(Args&&... args); + + // So Delete() can call our private dtor. + template <typename T> + friend void grpc_core::Delete(T*); + + ClientChannelMethodParams() {} + virtual ~ClientChannelMethodParams() {} + + grpc_millis timeout_ = 0; + WaitForReady wait_for_ready_ = WAIT_FOR_READY_UNSET; + UniquePtr<RetryPolicy> retry_policy_; +}; + +} // namespace internal +} // namespace grpc_core + +#endif /* GRPC_CORE_EXT_FILTERS_CLIENT_CHANNEL_RESOLVER_RESULT_PARSING_H */ diff --git a/src/core/ext/filters/client_channel/server_address.cc b/src/core/ext/filters/client_channel/server_address.cc new file mode 100644 index 0000000000..ec33cbbd95 --- /dev/null +++ b/src/core/ext/filters/client_channel/server_address.cc @@ -0,0 +1,103 @@ +/* + * + * Copyright 2018 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/ext/filters/client_channel/server_address.h" + +#include <string.h> + +namespace grpc_core { + +// +// ServerAddress +// + +ServerAddress::ServerAddress(const grpc_resolved_address& address, + grpc_channel_args* args) + : address_(address), args_(args) {} + +ServerAddress::ServerAddress(const void* address, size_t address_len, + grpc_channel_args* args) + : args_(args) { + memcpy(address_.addr, address, address_len); + address_.len = static_cast<socklen_t>(address_len); +} + +int ServerAddress::Cmp(const ServerAddress& other) const { + if (address_.len > other.address_.len) return 1; + if (address_.len < other.address_.len) return -1; + int retval = memcmp(address_.addr, other.address_.addr, address_.len); + if (retval != 0) return retval; + return grpc_channel_args_compare(args_, other.args_); +} + +bool ServerAddress::IsBalancer() const { + return grpc_channel_arg_get_bool( + grpc_channel_args_find(args_, GRPC_ARG_ADDRESS_IS_BALANCER), false); +} + +// +// ServerAddressList +// + +namespace { + +void* ServerAddressListCopy(void* addresses) { + ServerAddressList* a = static_cast<ServerAddressList*>(addresses); + return New<ServerAddressList>(*a); +} + +void ServerAddressListDestroy(void* addresses) { + ServerAddressList* a = static_cast<ServerAddressList*>(addresses); + Delete(a); +} + +int ServerAddressListCompare(void* addresses1, void* addresses2) { + ServerAddressList* a1 = static_cast<ServerAddressList*>(addresses1); + ServerAddressList* a2 = static_cast<ServerAddressList*>(addresses2); + if (a1->size() > a2->size()) return 1; + if (a1->size() < a2->size()) return -1; + for (size_t i = 0; i < a1->size(); ++i) { + int retval = (*a1)[i].Cmp((*a2)[i]); + if (retval != 0) return retval; + } + return 0; +} + +const grpc_arg_pointer_vtable server_addresses_arg_vtable = { + ServerAddressListCopy, ServerAddressListDestroy, ServerAddressListCompare}; + +} // namespace + +grpc_arg CreateServerAddressListChannelArg(const ServerAddressList* addresses) { + return grpc_channel_arg_pointer_create( + const_cast<char*>(GRPC_ARG_SERVER_ADDRESS_LIST), + const_cast<ServerAddressList*>(addresses), &server_addresses_arg_vtable); +} + +ServerAddressList* FindServerAddressListChannelArg( + const grpc_channel_args* channel_args) { + const grpc_arg* lb_addresses_arg = + grpc_channel_args_find(channel_args, GRPC_ARG_SERVER_ADDRESS_LIST); + if (lb_addresses_arg == nullptr || lb_addresses_arg->type != GRPC_ARG_POINTER) + return nullptr; + return static_cast<ServerAddressList*>(lb_addresses_arg->value.pointer.p); +} + +} // namespace grpc_core diff --git a/src/core/ext/filters/client_channel/server_address.h b/src/core/ext/filters/client_channel/server_address.h new file mode 100644 index 0000000000..3a1bf1df67 --- /dev/null +++ b/src/core/ext/filters/client_channel/server_address.h @@ -0,0 +1,108 @@ +/* + * + * Copyright 2018 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_CORE_EXT_FILTERS_CLIENT_CHANNEL_SERVER_ADDRESS_H +#define GRPC_CORE_EXT_FILTERS_CLIENT_CHANNEL_SERVER_ADDRESS_H + +#include <grpc/support/port_platform.h> + +#include "src/core/lib/channel/channel_args.h" +#include "src/core/lib/gprpp/inlined_vector.h" +#include "src/core/lib/iomgr/resolve_address.h" +#include "src/core/lib/uri/uri_parser.h" + +// Channel arg key for ServerAddressList. +#define GRPC_ARG_SERVER_ADDRESS_LIST "grpc.server_address_list" + +// Channel arg key for a bool indicating whether an address is a grpclb +// load balancer (as opposed to a backend). +#define GRPC_ARG_ADDRESS_IS_BALANCER "grpc.address_is_balancer" + +// Channel arg key for a string indicating an address's balancer name. +#define GRPC_ARG_ADDRESS_BALANCER_NAME "grpc.address_balancer_name" + +namespace grpc_core { + +// +// ServerAddress +// + +// A server address is a grpc_resolved_address with an associated set of +// channel args. Any args present here will be merged into the channel +// args when a subchannel is created for this address. +class ServerAddress { + public: + // Takes ownership of args. + ServerAddress(const grpc_resolved_address& address, grpc_channel_args* args); + ServerAddress(const void* address, size_t address_len, + grpc_channel_args* args); + + ~ServerAddress() { grpc_channel_args_destroy(args_); } + + // Copyable. + ServerAddress(const ServerAddress& other) + : address_(other.address_), args_(grpc_channel_args_copy(other.args_)) {} + ServerAddress& operator=(const ServerAddress& other) { + address_ = other.address_; + grpc_channel_args_destroy(args_); + args_ = grpc_channel_args_copy(other.args_); + return *this; + } + + // Movable. + ServerAddress(ServerAddress&& other) + : address_(other.address_), args_(other.args_) { + other.args_ = nullptr; + } + ServerAddress& operator=(ServerAddress&& other) { + address_ = other.address_; + args_ = other.args_; + other.args_ = nullptr; + return *this; + } + + bool operator==(const ServerAddress& other) const { return Cmp(other) == 0; } + + int Cmp(const ServerAddress& other) const; + + const grpc_resolved_address& address() const { return address_; } + const grpc_channel_args* args() const { return args_; } + + bool IsBalancer() const; + + private: + grpc_resolved_address address_; + grpc_channel_args* args_; +}; + +// +// ServerAddressList +// + +typedef InlinedVector<ServerAddress, 1> ServerAddressList; + +// Returns a channel arg containing \a addresses. +grpc_arg CreateServerAddressListChannelArg(const ServerAddressList* addresses); + +// Returns the ServerListAddress instance in channel_args or NULL. +ServerAddressList* FindServerAddressListChannelArg( + const grpc_channel_args* channel_args); + +} // namespace grpc_core + +#endif /* GRPC_CORE_EXT_FILTERS_CLIENT_CHANNEL_SERVER_ADDRESS_H */ diff --git a/src/core/ext/filters/client_channel/subchannel.cc b/src/core/ext/filters/client_channel/subchannel.cc index a56db0201b..9077aa9753 100644 --- a/src/core/ext/filters/client_channel/subchannel.cc +++ b/src/core/ext/filters/client_channel/subchannel.cc @@ -153,7 +153,7 @@ struct grpc_subchannel { /** have we started the backoff loop */ bool backoff_begun; // reset_backoff() was called while alarm was pending - bool deferred_reset_backoff; + bool retry_immediately; /** our alarm */ grpc_timer alarm; @@ -709,8 +709,8 @@ static void on_alarm(void* arg, grpc_error* error) { if (c->disconnected) { error = GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING("Disconnected", &error, 1); - } else if (c->deferred_reset_backoff) { - c->deferred_reset_backoff = false; + } else if (c->retry_immediately) { + c->retry_immediately = false; error = GRPC_ERROR_NONE; } else { GRPC_ERROR_REF(error); @@ -837,7 +837,7 @@ static bool publish_transport_locked(grpc_subchannel* c) { /* publish */ c->connected_subchannel.reset(grpc_core::New<grpc_core::ConnectedSubchannel>( - stk, c->channelz_subchannel, socket_uuid)); + stk, c->args, c->channelz_subchannel, socket_uuid)); gpr_log(GPR_INFO, "New connected subchannel at %p for subchannel %p", c->connected_subchannel.get(), c); @@ -887,12 +887,12 @@ static void on_subchannel_connected(void* arg, grpc_error* error) { void grpc_subchannel_reset_backoff(grpc_subchannel* subchannel) { gpr_mu_lock(&subchannel->mu); + subchannel->backoff->Reset(); if (subchannel->have_alarm) { - subchannel->deferred_reset_backoff = true; + subchannel->retry_immediately = true; grpc_timer_cancel(&subchannel->alarm); } else { subchannel->backoff_begun = false; - subchannel->backoff->Reset(); maybe_start_connecting_locked(subchannel); } gpr_mu_unlock(&subchannel->mu); @@ -1068,16 +1068,18 @@ grpc_arg grpc_create_subchannel_address_arg(const grpc_resolved_address* addr) { namespace grpc_core { ConnectedSubchannel::ConnectedSubchannel( - grpc_channel_stack* channel_stack, + grpc_channel_stack* channel_stack, const grpc_channel_args* args, grpc_core::RefCountedPtr<grpc_core::channelz::SubchannelNode> channelz_subchannel, intptr_t socket_uuid) - : RefCountedWithTracing<ConnectedSubchannel>(&grpc_trace_stream_refcount), + : RefCounted<ConnectedSubchannel>(&grpc_trace_stream_refcount), channel_stack_(channel_stack), + args_(grpc_channel_args_copy(args)), channelz_subchannel_(std::move(channelz_subchannel)), socket_uuid_(socket_uuid) {} ConnectedSubchannel::~ConnectedSubchannel() { + grpc_channel_args_destroy(args_); GRPC_CHANNEL_STACK_UNREF(channel_stack_, "connected_subchannel_dtor"); } diff --git a/src/core/ext/filters/client_channel/subchannel.h b/src/core/ext/filters/client_channel/subchannel.h index ec3b4d86e4..14f87f2c68 100644 --- a/src/core/ext/filters/client_channel/subchannel.h +++ b/src/core/ext/filters/client_channel/subchannel.h @@ -72,7 +72,7 @@ typedef struct grpc_subchannel_key grpc_subchannel_key; namespace grpc_core { -class ConnectedSubchannel : public RefCountedWithTracing<ConnectedSubchannel> { +class ConnectedSubchannel : public RefCounted<ConnectedSubchannel> { public: struct CallArgs { grpc_polling_entity* pollent; @@ -85,28 +85,31 @@ class ConnectedSubchannel : public RefCountedWithTracing<ConnectedSubchannel> { size_t parent_data_size; }; - explicit ConnectedSubchannel( - grpc_channel_stack* channel_stack, + ConnectedSubchannel( + grpc_channel_stack* channel_stack, const grpc_channel_args* args, grpc_core::RefCountedPtr<grpc_core::channelz::SubchannelNode> channelz_subchannel, intptr_t socket_uuid); ~ConnectedSubchannel(); - grpc_channel_stack* channel_stack() { return channel_stack_; } void NotifyOnStateChange(grpc_pollset_set* interested_parties, grpc_connectivity_state* state, grpc_closure* closure); void Ping(grpc_closure* on_initiate, grpc_closure* on_ack); grpc_error* CreateCall(const CallArgs& args, grpc_subchannel_call** call); - channelz::SubchannelNode* channelz_subchannel() { + + grpc_channel_stack* channel_stack() const { return channel_stack_; } + const grpc_channel_args* args() const { return args_; } + channelz::SubchannelNode* channelz_subchannel() const { return channelz_subchannel_.get(); } - intptr_t socket_uuid() { return socket_uuid_; } + intptr_t socket_uuid() const { return socket_uuid_; } size_t GetInitialCallSizeEstimate(size_t parent_data_size) const; private: grpc_channel_stack* channel_stack_; + grpc_channel_args* args_; // ref counted pointer to the channelz node in this connected subchannel's // owning subchannel. grpc_core::RefCountedPtr<grpc_core::channelz::SubchannelNode> diff --git a/src/core/ext/filters/client_channel/subchannel_index.cc b/src/core/ext/filters/client_channel/subchannel_index.cc index 1c23a6c4be..aa8441f17b 100644 --- a/src/core/ext/filters/client_channel/subchannel_index.cc +++ b/src/core/ext/filters/client_channel/subchannel_index.cc @@ -91,7 +91,7 @@ void grpc_subchannel_key_destroy(grpc_subchannel_key* k) { gpr_free(k); } -static void sck_avl_destroy(void* p, void* user_data) { +static void sck_avl_destroy(void* p, void* unused) { grpc_subchannel_key_destroy(static_cast<grpc_subchannel_key*>(p)); } @@ -104,7 +104,7 @@ static long sck_avl_compare(void* a, void* b, void* unused) { static_cast<grpc_subchannel_key*>(b)); } -static void scv_avl_destroy(void* p, void* user_data) { +static void scv_avl_destroy(void* p, void* unused) { GRPC_SUBCHANNEL_WEAK_UNREF((grpc_subchannel*)p, "subchannel_index"); } @@ -137,7 +137,7 @@ void grpc_subchannel_index_shutdown(void) { void grpc_subchannel_index_unref(void) { if (gpr_unref(&g_refcount)) { gpr_mu_destroy(&g_mu); - grpc_avl_unref(g_subchannel_index, grpc_core::ExecCtx::Get()); + grpc_avl_unref(g_subchannel_index, nullptr); } } @@ -147,13 +147,12 @@ grpc_subchannel* grpc_subchannel_index_find(grpc_subchannel_key* key) { // Lock, and take a reference to the subchannel index. // We don't need to do the search under a lock as avl's are immutable. gpr_mu_lock(&g_mu); - grpc_avl index = grpc_avl_ref(g_subchannel_index, grpc_core::ExecCtx::Get()); + grpc_avl index = grpc_avl_ref(g_subchannel_index, nullptr); gpr_mu_unlock(&g_mu); grpc_subchannel* c = GRPC_SUBCHANNEL_REF_FROM_WEAK_REF( - (grpc_subchannel*)grpc_avl_get(index, key, grpc_core::ExecCtx::Get()), - "index_find"); - grpc_avl_unref(index, grpc_core::ExecCtx::Get()); + (grpc_subchannel*)grpc_avl_get(index, key, nullptr), "index_find"); + grpc_avl_unref(index, nullptr); return c; } @@ -169,13 +168,11 @@ grpc_subchannel* grpc_subchannel_index_register(grpc_subchannel_key* key, // Compare and swap loop: // - take a reference to the current index gpr_mu_lock(&g_mu); - grpc_avl index = - grpc_avl_ref(g_subchannel_index, grpc_core::ExecCtx::Get()); + grpc_avl index = grpc_avl_ref(g_subchannel_index, nullptr); gpr_mu_unlock(&g_mu); // - Check to see if a subchannel already exists - c = static_cast<grpc_subchannel*>( - grpc_avl_get(index, key, grpc_core::ExecCtx::Get())); + c = static_cast<grpc_subchannel*>(grpc_avl_get(index, key, nullptr)); if (c != nullptr) { c = GRPC_SUBCHANNEL_REF_FROM_WEAK_REF(c, "index_register"); } @@ -184,11 +181,9 @@ grpc_subchannel* grpc_subchannel_index_register(grpc_subchannel_key* key, need_to_unref_constructed = true; } else { // no -> update the avl and compare/swap - grpc_avl updated = - grpc_avl_add(grpc_avl_ref(index, grpc_core::ExecCtx::Get()), - subchannel_key_copy(key), - GRPC_SUBCHANNEL_WEAK_REF(constructed, "index_register"), - grpc_core::ExecCtx::Get()); + grpc_avl updated = grpc_avl_add( + grpc_avl_ref(index, nullptr), subchannel_key_copy(key), + GRPC_SUBCHANNEL_WEAK_REF(constructed, "index_register"), nullptr); // it may happen (but it's expected to be unlikely) // that some other thread has changed the index: @@ -200,9 +195,9 @@ grpc_subchannel* grpc_subchannel_index_register(grpc_subchannel_key* key, } gpr_mu_unlock(&g_mu); - grpc_avl_unref(updated, grpc_core::ExecCtx::Get()); + grpc_avl_unref(updated, nullptr); } - grpc_avl_unref(index, grpc_core::ExecCtx::Get()); + grpc_avl_unref(index, nullptr); } if (need_to_unref_constructed) { @@ -219,24 +214,22 @@ void grpc_subchannel_index_unregister(grpc_subchannel_key* key, // Compare and swap loop: // - take a reference to the current index gpr_mu_lock(&g_mu); - grpc_avl index = - grpc_avl_ref(g_subchannel_index, grpc_core::ExecCtx::Get()); + grpc_avl index = grpc_avl_ref(g_subchannel_index, nullptr); gpr_mu_unlock(&g_mu); // Check to see if this key still refers to the previously // registered subchannel - grpc_subchannel* c = static_cast<grpc_subchannel*>( - grpc_avl_get(index, key, grpc_core::ExecCtx::Get())); + grpc_subchannel* c = + static_cast<grpc_subchannel*>(grpc_avl_get(index, key, nullptr)); if (c != constructed) { - grpc_avl_unref(index, grpc_core::ExecCtx::Get()); + grpc_avl_unref(index, nullptr); break; } // compare and swap the update (some other thread may have // mutated the index behind us) grpc_avl updated = - grpc_avl_remove(grpc_avl_ref(index, grpc_core::ExecCtx::Get()), key, - grpc_core::ExecCtx::Get()); + grpc_avl_remove(grpc_avl_ref(index, nullptr), key, nullptr); gpr_mu_lock(&g_mu); if (index.root == g_subchannel_index.root) { @@ -245,8 +238,8 @@ void grpc_subchannel_index_unregister(grpc_subchannel_key* key, } gpr_mu_unlock(&g_mu); - grpc_avl_unref(updated, grpc_core::ExecCtx::Get()); - grpc_avl_unref(index, grpc_core::ExecCtx::Get()); + grpc_avl_unref(updated, nullptr); + grpc_avl_unref(index, nullptr); } } diff --git a/src/core/ext/transport/chttp2/client/chttp2_connector.cc b/src/core/ext/transport/chttp2/client/chttp2_connector.cc index 60a32022f5..42a2e2e896 100644 --- a/src/core/ext/transport/chttp2/client/chttp2_connector.cc +++ b/src/core/ext/transport/chttp2/client/chttp2_connector.cc @@ -117,8 +117,9 @@ static void on_handshake_done(void* arg, grpc_error* error) { c->args.interested_parties); c->result->transport = grpc_create_chttp2_transport(args->args, args->endpoint, true); - c->result->socket_uuid = - grpc_chttp2_transport_get_socket_uuid(c->result->transport); + grpc_core::RefCountedPtr<grpc_core::channelz::SocketNode> socket_node = + grpc_chttp2_transport_get_socket_node(c->result->transport); + c->result->socket_uuid = socket_node == nullptr ? 0 : socket_node->uuid(); GPR_ASSERT(c->result->transport); // TODO(roth): We ideally want to wait until we receive HTTP/2 // settings from the server before we consider the connection diff --git a/src/core/ext/transport/chttp2/client/secure/secure_channel_create.cc b/src/core/ext/transport/chttp2/client/secure/secure_channel_create.cc index e73eee4353..9612698e96 100644 --- a/src/core/ext/transport/chttp2/client/secure/secure_channel_create.cc +++ b/src/core/ext/transport/chttp2/client/secure/secure_channel_create.cc @@ -110,14 +110,14 @@ static grpc_subchannel_args* get_secure_naming_subchannel_args( grpc_channel_args* args_with_authority = grpc_channel_args_copy_and_add(args->args, args_to_add, num_args_to_add); grpc_uri_destroy(server_uri); - grpc_channel_security_connector* subchannel_security_connector = nullptr; // Create the security connector using the credentials and target name. grpc_channel_args* new_args_from_connector = nullptr; - const grpc_security_status security_status = - grpc_channel_credentials_create_security_connector( - channel_credentials, authority.get(), args_with_authority, - &subchannel_security_connector, &new_args_from_connector); - if (security_status != GRPC_SECURITY_OK) { + grpc_core::RefCountedPtr<grpc_channel_security_connector> + subchannel_security_connector = + channel_credentials->create_security_connector( + /*call_creds=*/nullptr, authority.get(), args_with_authority, + &new_args_from_connector); + if (subchannel_security_connector == nullptr) { gpr_log(GPR_ERROR, "Failed to create secure subchannel for secure name '%s'", authority.get()); @@ -125,15 +125,14 @@ static grpc_subchannel_args* get_secure_naming_subchannel_args( return nullptr; } grpc_arg new_security_connector_arg = - grpc_security_connector_to_arg(&subchannel_security_connector->base); + grpc_security_connector_to_arg(subchannel_security_connector.get()); grpc_channel_args* new_args = grpc_channel_args_copy_and_add( new_args_from_connector != nullptr ? new_args_from_connector : args_with_authority, &new_security_connector_arg, 1); - GRPC_SECURITY_CONNECTOR_UNREF(&subchannel_security_connector->base, - "lb_channel_create"); + subchannel_security_connector.reset(DEBUG_LOCATION, "lb_channel_create"); if (new_args_from_connector != nullptr) { grpc_channel_args_destroy(new_args_from_connector); } diff --git a/src/core/ext/transport/chttp2/server/chttp2_server.cc b/src/core/ext/transport/chttp2/server/chttp2_server.cc index 33d2b22aa5..3d09187b9b 100644 --- a/src/core/ext/transport/chttp2/server/chttp2_server.cc +++ b/src/core/ext/transport/chttp2/server/chttp2_server.cc @@ -149,7 +149,7 @@ static void on_handshake_done(void* arg, grpc_error* error) { grpc_server_setup_transport( connection_state->svr_state->server, transport, connection_state->accepting_pollset, args->args, - grpc_chttp2_transport_get_socket_uuid(transport), resource_user); + grpc_chttp2_transport_get_socket_node(transport), resource_user); // Use notify_on_receive_settings callback to enforce the // handshake deadline. connection_state->transport = diff --git a/src/core/ext/transport/chttp2/server/insecure/server_chttp2_posix.cc b/src/core/ext/transport/chttp2/server/insecure/server_chttp2_posix.cc index b9024a87e2..c29c1e58cd 100644 --- a/src/core/ext/transport/chttp2/server/insecure/server_chttp2_posix.cc +++ b/src/core/ext/transport/chttp2/server/insecure/server_chttp2_posix.cc @@ -61,7 +61,7 @@ void grpc_server_add_insecure_channel_from_fd(grpc_server* server, grpc_endpoint_add_to_pollset(server_endpoint, pollsets[i]); } - grpc_server_setup_transport(server, transport, nullptr, server_args, 0); + grpc_server_setup_transport(server, transport, nullptr, server_args, nullptr); grpc_chttp2_transport_start_reading(transport, nullptr, nullptr); } diff --git a/src/core/ext/transport/chttp2/server/secure/server_secure_chttp2.cc b/src/core/ext/transport/chttp2/server/secure/server_secure_chttp2.cc index 6689a17da6..98fdb62070 100644 --- a/src/core/ext/transport/chttp2/server/secure/server_secure_chttp2.cc +++ b/src/core/ext/transport/chttp2/server/secure/server_secure_chttp2.cc @@ -31,6 +31,7 @@ #include "src/core/ext/transport/chttp2/transport/chttp2_transport.h" #include "src/core/lib/channel/channel_args.h" #include "src/core/lib/channel/handshaker.h" +#include "src/core/lib/gprpp/ref_counted_ptr.h" #include "src/core/lib/security/context/security_context.h" #include "src/core/lib/security/credentials/credentials.h" #include "src/core/lib/surface/api_trace.h" @@ -40,9 +41,8 @@ int grpc_server_add_secure_http2_port(grpc_server* server, const char* addr, grpc_server_credentials* creds) { grpc_core::ExecCtx exec_ctx; grpc_error* err = GRPC_ERROR_NONE; - grpc_server_security_connector* sc = nullptr; + grpc_core::RefCountedPtr<grpc_server_security_connector> sc; int port_num = 0; - grpc_security_status status; grpc_channel_args* args = nullptr; GRPC_API_TRACE( "grpc_server_add_secure_http2_port(" @@ -54,30 +54,27 @@ int grpc_server_add_secure_http2_port(grpc_server* server, const char* addr, "No credentials specified for secure server port (creds==NULL)"); goto done; } - status = grpc_server_credentials_create_security_connector(creds, &sc); - if (status != GRPC_SECURITY_OK) { + sc = creds->create_security_connector(); + if (sc == nullptr) { char* msg; gpr_asprintf(&msg, "Unable to create secure server with credentials of type %s.", - creds->type); - err = grpc_error_set_int(GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg), - GRPC_ERROR_INT_SECURITY_STATUS, status); + creds->type()); + err = GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg); gpr_free(msg); goto done; } // Create channel args. grpc_arg args_to_add[2]; args_to_add[0] = grpc_server_credentials_to_arg(creds); - args_to_add[1] = grpc_security_connector_to_arg(&sc->base); + args_to_add[1] = grpc_security_connector_to_arg(sc.get()); args = grpc_channel_args_copy_and_add(grpc_server_get_channel_args(server), args_to_add, GPR_ARRAY_SIZE(args_to_add)); // Add server port. err = grpc_chttp2_server_add_port(server, addr, args, &port_num); done: - if (sc != nullptr) { - GRPC_SECURITY_CONNECTOR_UNREF(&sc->base, "server"); - } + sc.reset(DEBUG_LOCATION, "server"); if (err != GRPC_ERROR_NONE) { const char* msg = grpc_error_string(err); diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.cc b/src/core/ext/transport/chttp2/transport/chttp2_transport.cc index 978ecd59e4..9b6574b612 100644 --- a/src/core/ext/transport/chttp2/transport/chttp2_transport.cc +++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.cc @@ -1,6 +1,6 @@ /* * - * Copyright 2015 gRPC authors. + * Copyright 2018 gRPC authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -31,6 +31,7 @@ #include <grpc/support/log.h> #include <grpc/support/string_util.h> +#include "src/core/ext/transport/chttp2/transport/context_list.h" #include "src/core/ext/transport/chttp2/transport/frame_data.h" #include "src/core/ext/transport/chttp2/transport/internal.h" #include "src/core/ext/transport/chttp2/transport/varint.h" @@ -154,6 +155,7 @@ bool g_flow_control_enabled = true; /******************************************************************************* * CONSTRUCTION/DESTRUCTION/REFCOUNTING */ + grpc_chttp2_transport::~grpc_chttp2_transport() { size_t i; @@ -168,6 +170,9 @@ grpc_chttp2_transport::~grpc_chttp2_transport() { grpc_slice_buffer_destroy_internal(&outbuf); grpc_chttp2_hpack_compressor_destroy(&hpack_compressor); + grpc_core::ContextList::Execute(cl, nullptr, GRPC_ERROR_NONE); + cl = nullptr; + grpc_slice_buffer_destroy_internal(&read_buffer); grpc_chttp2_hpack_parser_destroy(&hpack_parser); grpc_chttp2_goaway_parser_destroy(&goaway_parser); @@ -202,38 +207,6 @@ grpc_chttp2_transport::~grpc_chttp2_transport() { gpr_free(peer_string); } -#ifndef NDEBUG -void grpc_chttp2_unref_transport(grpc_chttp2_transport* t, const char* reason, - const char* file, int line) { - if (grpc_trace_chttp2_refcount.enabled()) { - gpr_atm val = gpr_atm_no_barrier_load(&t->refs.count); - gpr_log(GPR_DEBUG, "chttp2:unref:%p %" PRIdPTR "->%" PRIdPTR " %s [%s:%d]", - t, val, val - 1, reason, file, line); - } - if (!gpr_unref(&t->refs)) return; - t->~grpc_chttp2_transport(); - gpr_free(t); -} - -void grpc_chttp2_ref_transport(grpc_chttp2_transport* t, const char* reason, - const char* file, int line) { - if (grpc_trace_chttp2_refcount.enabled()) { - gpr_atm val = gpr_atm_no_barrier_load(&t->refs.count); - gpr_log(GPR_DEBUG, "chttp2: ref:%p %" PRIdPTR "->%" PRIdPTR " %s [%s:%d]", - t, val, val + 1, reason, file, line); - } - gpr_ref(&t->refs); -} -#else -void grpc_chttp2_unref_transport(grpc_chttp2_transport* t) { - if (!gpr_unref(&t->refs)) return; - t->~grpc_chttp2_transport(); - gpr_free(t); -} - -void grpc_chttp2_ref_transport(grpc_chttp2_transport* t) { gpr_ref(&t->refs); } -#endif - static const grpc_transport_vtable* get_vtable(void); /* Returns whether bdp is enabled */ @@ -485,7 +458,8 @@ static void init_keepalive_pings_if_enabled(grpc_chttp2_transport* t) { grpc_chttp2_transport::grpc_chttp2_transport( const grpc_channel_args* channel_args, grpc_endpoint* ep, bool is_client, grpc_resource_user* resource_user) - : ep(ep), + : refs(1, &grpc_trace_chttp2_refcount), + ep(ep), peer_string(grpc_endpoint_get_peer(ep)), resource_user(resource_user), combiner(grpc_combiner_create()), @@ -495,8 +469,6 @@ grpc_chttp2_transport::grpc_chttp2_transport( GPR_ASSERT(strlen(GRPC_CHTTP2_CLIENT_CONNECT_STRING) == GRPC_CHTTP2_CLIENT_CONNECT_STRLEN); base.vtable = get_vtable(); - /* one ref is for destroy */ - gpr_ref_init(&refs, 1); /* 8 is a random stab in the dark as to a good initial size: it's small enough that it shouldn't waste memory for infrequently used connections, yet large enough that the exponential growth should happen nicely when it's @@ -1065,11 +1037,13 @@ static void write_action_begin_locked(void* gt, grpc_error* error_ignored) { static void write_action(void* gt, grpc_error* error) { GPR_TIMER_SCOPE("write_action", 0); grpc_chttp2_transport* t = static_cast<grpc_chttp2_transport*>(gt); + void* cl = t->cl; + t->cl = nullptr; grpc_endpoint_write( t->ep, &t->outbuf, GRPC_CLOSURE_INIT(&t->write_action_end_locked, write_action_end_locked, t, grpc_combiner_scheduler(t->combiner)), - nullptr); + cl); } /* Callback from the grpc_endpoint after bytes have been written by calling @@ -1393,6 +1367,8 @@ static void perform_stream_op_locked(void* stream_op, GRPC_STATS_INC_HTTP2_OP_BATCHES(); + s->context = op->payload->context; + s->traced = op->is_traced; if (grpc_http_trace.enabled()) { char* str = grpc_transport_stream_op_batch_string(op); gpr_log(GPR_INFO, "perform_stream_op_locked: %s; on_complete = %p", str, @@ -2837,8 +2813,8 @@ Chttp2IncomingByteStream::Chttp2IncomingByteStream( : ByteStream(frame_size, flags), transport_(transport), stream_(stream), + refs_(2), remaining_bytes_(frame_size) { - gpr_ref_init(&refs_, 2); GRPC_ERROR_UNREF(stream->byte_stream_error); stream->byte_stream_error = GRPC_ERROR_NONE; } @@ -2863,14 +2839,6 @@ void Chttp2IncomingByteStream::Orphan() { GRPC_ERROR_NONE); } -void Chttp2IncomingByteStream::Unref() { - if (gpr_unref(&refs_)) { - Delete(this); - } -} - -void Chttp2IncomingByteStream::Ref() { gpr_ref(&refs_); } - void Chttp2IncomingByteStream::NextLocked(void* arg, grpc_error* error_ignored) { Chttp2IncomingByteStream* bs = static_cast<Chttp2IncomingByteStream*>(arg); @@ -3177,21 +3145,18 @@ static const grpc_transport_vtable vtable = {sizeof(grpc_chttp2_stream), static const grpc_transport_vtable* get_vtable(void) { return &vtable; } -intptr_t grpc_chttp2_transport_get_socket_uuid(grpc_transport* transport) { +grpc_core::RefCountedPtr<grpc_core::channelz::SocketNode> +grpc_chttp2_transport_get_socket_node(grpc_transport* transport) { grpc_chttp2_transport* t = reinterpret_cast<grpc_chttp2_transport*>(transport); - if (t->channelz_socket != nullptr) { - return t->channelz_socket->uuid(); - } else { - return 0; - } + return t->channelz_socket; } grpc_transport* grpc_create_chttp2_transport( const grpc_channel_args* channel_args, grpc_endpoint* ep, bool is_client, grpc_resource_user* resource_user) { - auto t = new (gpr_malloc(sizeof(grpc_chttp2_transport))) - grpc_chttp2_transport(channel_args, ep, is_client, resource_user); + auto t = grpc_core::New<grpc_chttp2_transport>(channel_args, ep, is_client, + resource_user); return &t->base; } diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.h b/src/core/ext/transport/chttp2/transport/chttp2_transport.h index b3fe1c082e..c22cfb0ad7 100644 --- a/src/core/ext/transport/chttp2/transport/chttp2_transport.h +++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.h @@ -21,6 +21,7 @@ #include <grpc/support/port_platform.h> +#include "src/core/lib/channel/channelz.h" #include "src/core/lib/debug/trace.h" #include "src/core/lib/iomgr/endpoint.h" #include "src/core/lib/transport/transport.h" @@ -35,7 +36,8 @@ grpc_transport* grpc_create_chttp2_transport( const grpc_channel_args* channel_args, grpc_endpoint* ep, bool is_client, grpc_resource_user* resource_user = nullptr); -intptr_t grpc_chttp2_transport_get_socket_uuid(grpc_transport* transport); +grpc_core::RefCountedPtr<grpc_core::channelz::SocketNode> +grpc_chttp2_transport_get_socket_node(grpc_transport* transport); /// Takes ownership of \a read_buffer, which (if non-NULL) contains /// leftover bytes previously read from the endpoint (e.g., by handshakers). diff --git a/src/core/ext/transport/chttp2/transport/context_list.cc b/src/core/ext/transport/chttp2/transport/context_list.cc new file mode 100644 index 0000000000..f30d41c332 --- /dev/null +++ b/src/core/ext/transport/chttp2/transport/context_list.cc @@ -0,0 +1,51 @@ +/* + * + * Copyright 2018 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/ext/transport/chttp2/transport/context_list.h" + +namespace { +void (*write_timestamps_callback_g)(void*, grpc_core::Timestamps*) = nullptr; +} + +namespace grpc_core { +void ContextList::Execute(void* arg, grpc_core::Timestamps* ts, + grpc_error* error) { + ContextList* head = static_cast<ContextList*>(arg); + ContextList* to_be_freed; + while (head != nullptr) { + if (error == GRPC_ERROR_NONE && ts != nullptr) { + if (write_timestamps_callback_g) { + ts->byte_offset = static_cast<uint32_t>(head->byte_offset_); + write_timestamps_callback_g(head->s_->context, ts); + } + } + GRPC_CHTTP2_STREAM_UNREF(static_cast<grpc_chttp2_stream*>(head->s_), + "timestamp"); + to_be_freed = head; + head = head->next_; + grpc_core::Delete(to_be_freed); + } +} + +void grpc_http2_set_write_timestamps_callback( + void (*fn)(void*, grpc_core::Timestamps*)) { + write_timestamps_callback_g = fn; +} +} /* namespace grpc_core */ diff --git a/src/core/ext/transport/chttp2/transport/context_list.h b/src/core/ext/transport/chttp2/transport/context_list.h new file mode 100644 index 0000000000..d870107749 --- /dev/null +++ b/src/core/ext/transport/chttp2/transport/context_list.h @@ -0,0 +1,72 @@ +/* + * + * Copyright 2018 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_CORE_EXT_TRANSPORT_CHTTP2_TRANSPORT_CONTEXT_LIST_H +#define GRPC_CORE_EXT_TRANSPORT_CHTTP2_TRANSPORT_CONTEXT_LIST_H + +#include <grpc/support/port_platform.h> + +#include "src/core/lib/iomgr/buffer_list.h" + +#include "src/core/ext/transport/chttp2/transport/internal.h" + +namespace grpc_core { +/** A list of RPC Contexts */ +class ContextList { + public: + /* Creates a new element with \a context as the value and appends it to the + * list. */ + static void Append(ContextList** head, grpc_chttp2_stream* s) { + /* Make sure context is not already present */ + GRPC_CHTTP2_STREAM_REF(s, "timestamp"); + +#ifndef NDEBUG + ContextList* ptr = *head; + while (ptr != nullptr) { + if (ptr->s_ == s) { + GPR_ASSERT( + false && + "Trying to append a stream that is already present in the list"); + } + ptr = ptr->next_; + } +#endif + + /* Create a new element in the list and add it at the front */ + ContextList* elem = grpc_core::New<ContextList>(); + elem->s_ = s; + elem->byte_offset_ = s->byte_counter; + elem->next_ = *head; + *head = elem; + } + + /* Executes a function \a fn with each context in the list and \a ts. It also + * frees up the entire list after this operation. */ + static void Execute(void* arg, grpc_core::Timestamps* ts, grpc_error* error); + + private: + grpc_chttp2_stream* s_ = nullptr; + ContextList* next_ = nullptr; + size_t byte_offset_ = 0; +}; + +void grpc_http2_set_write_timestamps_callback( + void (*fn)(void*, grpc_core::Timestamps*)); +} /* namespace grpc_core */ + +#endif /* GRPC_CORE_EXT_TRANSPORT_CHTTP2_TRANSPORT_CONTEXT_LIST_H */ diff --git a/src/core/ext/transport/chttp2/transport/internal.h b/src/core/ext/transport/chttp2/transport/internal.h index 3ee408c103..341f5b3977 100644 --- a/src/core/ext/transport/chttp2/transport/internal.h +++ b/src/core/ext/transport/chttp2/transport/internal.h @@ -45,6 +45,10 @@ #include "src/core/lib/transport/connectivity_state.h" #include "src/core/lib/transport/transport_impl.h" +namespace grpc_core { +class ContextList; +} + /* streams are kept in various linked lists depending on what things need to happen to them... this enum labels each list */ typedef enum { @@ -232,8 +236,12 @@ class Chttp2IncomingByteStream : public ByteStream { // alone for now. We can revisit this once we're able to link against // libc++, at which point we can eliminate New<> and Delete<> and // switch to std::shared_ptr<>. - void Ref(); - void Unref(); + void Ref() { refs_.Ref(); } + void Unref() { + if (refs_.Unref()) { + grpc_core::Delete(this); + } + } void PublishError(grpc_error* error); @@ -252,7 +260,7 @@ class Chttp2IncomingByteStream : public ByteStream { grpc_chttp2_transport* transport_; // Immutable. grpc_chttp2_stream* stream_; // Immutable. - gpr_refcount refs_; + grpc_core::RefCount refs_; /* Accessed only by transport thread when stream->pending_byte_stream == false * Accessed only by application thread when stream->pending_byte_stream == @@ -286,7 +294,7 @@ struct grpc_chttp2_transport { ~grpc_chttp2_transport(); grpc_transport base; /* must be first */ - gpr_refcount refs; + grpc_core::RefCount refs; grpc_endpoint* ep; char* peer_string; @@ -481,7 +489,7 @@ struct grpc_chttp2_transport { bool keepalive_permit_without_calls = false; /** keep-alive state machine state */ grpc_chttp2_keepalive_state keepalive_state; - + grpc_core::ContextList* cl = nullptr; grpc_core::RefCountedPtr<grpc_core::channelz::SocketNode> channelz_socket; uint32_t num_messages_in_next_write = 0; }; @@ -498,6 +506,7 @@ struct grpc_chttp2_stream { const void* server_data, gpr_arena* arena); ~grpc_chttp2_stream(); + void* context; grpc_chttp2_transport* t; grpc_stream_refcount* refcount; @@ -633,8 +642,12 @@ struct grpc_chttp2_stream { /** Whether bytes stored in unprocessed_incoming_byte_stream is decompressed */ bool unprocessed_incoming_frames_decompressed = false; + /** Whether the bytes needs to be traced using Fathom */ + bool traced = false; /** gRPC header bytes that are already decompressed */ size_t decompressed_header_bytes = 0; + /** Byte counter for number of bytes written */ + size_t byte_counter = 0; }; /** Transport writing call flow: @@ -779,15 +792,29 @@ void grpc_chttp2_stream_unref(grpc_chttp2_stream* s); grpc_chttp2_ref_transport(t, r, __FILE__, __LINE__) #define GRPC_CHTTP2_UNREF_TRANSPORT(t, r) \ grpc_chttp2_unref_transport(t, r, __FILE__, __LINE__) -void grpc_chttp2_unref_transport(grpc_chttp2_transport* t, const char* reason, - const char* file, int line); -void grpc_chttp2_ref_transport(grpc_chttp2_transport* t, const char* reason, - const char* file, int line); +inline void grpc_chttp2_unref_transport(grpc_chttp2_transport* t, + const char* reason, const char* file, + int line) { + if (t->refs.Unref(grpc_core::DebugLocation(file, line), reason)) { + grpc_core::Delete(t); + } +} +inline void grpc_chttp2_ref_transport(grpc_chttp2_transport* t, + const char* reason, const char* file, + int line) { + t->refs.Ref(grpc_core::DebugLocation(file, line), reason); +} #else #define GRPC_CHTTP2_REF_TRANSPORT(t, r) grpc_chttp2_ref_transport(t) #define GRPC_CHTTP2_UNREF_TRANSPORT(t, r) grpc_chttp2_unref_transport(t) -void grpc_chttp2_unref_transport(grpc_chttp2_transport* t); -void grpc_chttp2_ref_transport(grpc_chttp2_transport* t); +inline void grpc_chttp2_unref_transport(grpc_chttp2_transport* t) { + if (t->refs.Unref()) { + grpc_core::Delete(t); + } +} +inline void grpc_chttp2_ref_transport(grpc_chttp2_transport* t) { + t->refs.Ref(); +} #endif void grpc_chttp2_ack_ping(grpc_chttp2_transport* t, uint64_t id); diff --git a/src/core/ext/transport/chttp2/transport/writing.cc b/src/core/ext/transport/chttp2/transport/writing.cc index d533989444..265d3365d3 100644 --- a/src/core/ext/transport/chttp2/transport/writing.cc +++ b/src/core/ext/transport/chttp2/transport/writing.cc @@ -18,6 +18,7 @@ #include <grpc/support/port_platform.h> +#include "src/core/ext/transport/chttp2/transport/context_list.h" #include "src/core/ext/transport/chttp2/transport/internal.h" #include <limits.h> @@ -362,6 +363,7 @@ class DataSendContext { grpc_chttp2_encode_data(s_->id, &s_->compressed_data_buffer, send_bytes, is_last_frame_, &s_->stats.outgoing, &t_->outbuf); s_->flow_control->SentData(send_bytes); + s_->byte_counter += send_bytes; if (s_->compressed_data_buffer.length == 0) { s_->sending_bytes += s_->uncompressed_data_size; } @@ -496,6 +498,9 @@ class StreamWriteContext { data_send_context.CompressMoreBytes(); } } + if (s_->traced && grpc_endpoint_can_track_err(t_->ep)) { + grpc_core::ContextList::Append(&t_->cl, s_); + } write_context_->ResetPingClock(); if (data_send_context.is_last_frame()) { SentLastFrame(); diff --git a/src/core/ext/transport/inproc/inproc_transport.cc b/src/core/ext/transport/inproc/inproc_transport.cc index 61968de4d5..0b9bf5dd11 100644 --- a/src/core/ext/transport/inproc/inproc_transport.cc +++ b/src/core/ext/transport/inproc/inproc_transport.cc @@ -1236,7 +1236,7 @@ grpc_channel* grpc_inproc_channel_create(grpc_server* server, // TODO(ncteisen): design and support channelz GetSocket for inproc. grpc_server_setup_transport(server, server_transport, nullptr, server_args, - 0); + nullptr); grpc_channel* channel = grpc_channel_create( "inproc", client_args, GRPC_CLIENT_DIRECT_CHANNEL, client_transport); |