diff options
Diffstat (limited to 'src/core/ext/filters/client_channel')
20 files changed, 424 insertions, 352 deletions
diff --git a/src/core/ext/filters/client_channel/channel_connectivity.c b/src/core/ext/filters/client_channel/channel_connectivity.c index e5f6fa76ae..3844b98021 100644 --- a/src/core/ext/filters/client_channel/channel_connectivity.c +++ b/src/core/ext/filters/client_channel/channel_connectivity.c @@ -86,7 +86,7 @@ static void delete_state_watcher(grpc_exec_ctx *exec_ctx, state_watcher *w) { static void finished_completion(grpc_exec_ctx *exec_ctx, void *pw, grpc_cq_completion *ignored) { - int delete = 0; + bool should_delete = false; state_watcher *w = (state_watcher *)pw; gpr_mu_lock(&w->mu); switch (w->phase) { @@ -94,12 +94,12 @@ static void finished_completion(grpc_exec_ctx *exec_ctx, void *pw, case READY_TO_CALL_BACK: GPR_UNREACHABLE_CODE(return ); case CALLING_BACK_AND_FINISHED: - delete = 1; + should_delete = true; break; } gpr_mu_unlock(&w->mu); - if (delete) { + if (should_delete) { delete_state_watcher(exec_ctx, w); } } @@ -161,12 +161,12 @@ static void partly_done(grpc_exec_ctx *exec_ctx, state_watcher *w, static void watch_complete(grpc_exec_ctx *exec_ctx, void *pw, grpc_error *error) { - partly_done(exec_ctx, pw, true, GRPC_ERROR_REF(error)); + partly_done(exec_ctx, (state_watcher *)pw, true, GRPC_ERROR_REF(error)); } static void timeout_complete(grpc_exec_ctx *exec_ctx, void *pw, grpc_error *error) { - partly_done(exec_ctx, pw, false, GRPC_ERROR_REF(error)); + partly_done(exec_ctx, (state_watcher *)pw, false, GRPC_ERROR_REF(error)); } int grpc_channel_num_external_connectivity_watchers(grpc_channel *channel) { diff --git a/src/core/ext/filters/client_channel/client_channel.c b/src/core/ext/filters/client_channel/client_channel.c index dad5c4fce5..016199b1f4 100644 --- a/src/core/ext/filters/client_channel/client_channel.c +++ b/src/core/ext/filters/client_channel/client_channel.c @@ -85,7 +85,7 @@ static void method_parameters_unref(method_parameters *method_params) { } static void method_parameters_free(grpc_exec_ctx *exec_ctx, void *value) { - method_parameters_unref(value); + method_parameters_unref((method_parameters *)value); } static bool parse_wait_for_ready(grpc_json *field, @@ -375,7 +375,7 @@ static void on_resolver_result_changed_locked(grpc_exec_ctx *exec_ctx, } // Extract the following fields from the resolver result, if non-NULL. bool lb_policy_updated = false; - char *lb_policy_name = NULL; + char *lb_policy_name_dup = NULL; bool lb_policy_name_changed = false; grpc_lb_policy *new_lb_policy = NULL; char *service_config_json = NULL; @@ -383,6 +383,7 @@ static void on_resolver_result_changed_locked(grpc_exec_ctx *exec_ctx, grpc_slice_hash_table *method_params_table = NULL; if (chand->resolver_result != NULL) { // Find LB policy name. + const char *lb_policy_name = NULL; const grpc_arg *channel_arg = grpc_channel_args_find(chand->resolver_result, GRPC_ARG_LB_POLICY_NAME); if (channel_arg != NULL) { @@ -473,7 +474,7 @@ static void on_resolver_result_changed_locked(grpc_exec_ctx *exec_ctx, // Before we clean up, save a copy of lb_policy_name, since it might // be pointing to data inside chand->resolver_result. // The copy will be saved in chand->lb_policy_name below. - lb_policy_name = gpr_strdup(lb_policy_name); + lb_policy_name_dup = gpr_strdup(lb_policy_name); grpc_channel_args_destroy(exec_ctx, chand->resolver_result); chand->resolver_result = NULL; } @@ -481,8 +482,8 @@ static void on_resolver_result_changed_locked(grpc_exec_ctx *exec_ctx, gpr_log(GPR_DEBUG, "chand=%p: resolver result: lb_policy_name=\"%s\"%s, " "service_config=\"%s\"", - chand, lb_policy_name, lb_policy_name_changed ? " (changed)" : "", - service_config_json); + chand, lb_policy_name_dup, + lb_policy_name_changed ? " (changed)" : "", service_config_json); } // Now swap out fields in chand. Note that the new values may still // be NULL if (e.g.) the resolver failed to return results or the @@ -490,9 +491,9 @@ static void on_resolver_result_changed_locked(grpc_exec_ctx *exec_ctx, // // First, swap out the data used by cc_get_channel_info(). gpr_mu_lock(&chand->info_mu); - if (lb_policy_name != NULL) { + if (lb_policy_name_dup != NULL) { gpr_free(chand->info_lb_policy_name); - chand->info_lb_policy_name = lb_policy_name; + chand->info_lb_policy_name = lb_policy_name_dup; } if (service_config_json != NULL) { gpr_free(chand->info_service_config_json); @@ -717,7 +718,8 @@ static grpc_error *cc_init_channel_elem(grpc_exec_ctx *exec_ctx, return GRPC_ERROR_CREATE_FROM_STATIC_STRING( "client channel factory arg must be a pointer"); } - grpc_client_channel_factory_ref(arg->value.pointer.p); + grpc_client_channel_factory_ref( + (grpc_client_channel_factory *)arg->value.pointer.p); chand->client_channel_factory = (grpc_client_channel_factory *)arg->value.pointer.p; // Get server name to resolve, using proxy mapper if needed. @@ -1016,13 +1018,11 @@ static void create_subchannel_call_locked(grpc_exec_ctx *exec_ctx, GRPC_ERROR_UNREF(error); } -static void subchannel_ready_locked(grpc_exec_ctx *exec_ctx, - grpc_call_element *elem, - grpc_error *error) { +// Invoked when a pick is completed, on both success or failure. +static void pick_done_locked(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, + grpc_error *error) { call_data *calld = (call_data *)elem->call_data; channel_data *chand = (channel_data *)elem->channel_data; - grpc_polling_entity_del_from_pollset_set(exec_ctx, calld->pollent, - chand->interested_parties); if (calld->connected_subchannel == NULL) { // Failed to create subchannel. GRPC_ERROR_UNREF(calld->error); @@ -1044,12 +1044,116 @@ static void subchannel_ready_locked(grpc_exec_ctx *exec_ctx, GRPC_ERROR_UNREF(error); } -/** Return true if subchannel is available immediately (in which case - subchannel_ready_locked() should not be called), or false otherwise (in - which case subchannel_ready_locked() should be called when the subchannel - is available). */ -static bool pick_subchannel_locked(grpc_exec_ctx *exec_ctx, - grpc_call_element *elem); +// A wrapper around pick_done_locked() that is used in cases where +// either (a) the pick was deferred pending a resolver result or (b) the +// pick was done asynchronously. Removes the call's polling entity from +// chand->interested_parties before invoking pick_done_locked(). +static void async_pick_done_locked(grpc_exec_ctx *exec_ctx, + grpc_call_element *elem, grpc_error *error) { + channel_data *chand = (channel_data *)elem->channel_data; + call_data *calld = (call_data *)elem->call_data; + grpc_polling_entity_del_from_pollset_set(exec_ctx, calld->pollent, + chand->interested_parties); + pick_done_locked(exec_ctx, elem, error); +} + +// Note: This runs under the client_channel combiner, but will NOT be +// holding the call combiner. +static void pick_callback_cancel_locked(grpc_exec_ctx *exec_ctx, void *arg, + grpc_error *error) { + grpc_call_element *elem = (grpc_call_element *)arg; + channel_data *chand = (channel_data *)elem->channel_data; + call_data *calld = (call_data *)elem->call_data; + if (calld->lb_policy != NULL) { + if (GRPC_TRACER_ON(grpc_client_channel_trace)) { + gpr_log(GPR_DEBUG, "chand=%p calld=%p: cancelling pick from LB policy %p", + chand, calld, calld->lb_policy); + } + grpc_lb_policy_cancel_pick_locked(exec_ctx, calld->lb_policy, + &calld->connected_subchannel, + GRPC_ERROR_REF(error)); + } + GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call, "pick_callback_cancel"); +} + +// Callback invoked by grpc_lb_policy_pick_locked() for async picks. +// Unrefs the LB policy and invokes async_pick_done_locked(). +static void pick_callback_done_locked(grpc_exec_ctx *exec_ctx, void *arg, + grpc_error *error) { + grpc_call_element *elem = (grpc_call_element *)arg; + channel_data *chand = (channel_data *)elem->channel_data; + call_data *calld = (call_data *)elem->call_data; + if (GRPC_TRACER_ON(grpc_client_channel_trace)) { + gpr_log(GPR_DEBUG, "chand=%p calld=%p: pick completed asynchronously", + chand, calld); + } + GPR_ASSERT(calld->lb_policy != NULL); + GRPC_LB_POLICY_UNREF(exec_ctx, calld->lb_policy, "pick_subchannel"); + calld->lb_policy = NULL; + async_pick_done_locked(exec_ctx, elem, GRPC_ERROR_REF(error)); +} + +// Takes a ref to chand->lb_policy and calls grpc_lb_policy_pick_locked(). +// If the pick was completed synchronously, unrefs the LB policy and +// returns true. +static bool pick_callback_start_locked(grpc_exec_ctx *exec_ctx, + grpc_call_element *elem) { + channel_data *chand = (channel_data *)elem->channel_data; + call_data *calld = (call_data *)elem->call_data; + if (GRPC_TRACER_ON(grpc_client_channel_trace)) { + gpr_log(GPR_DEBUG, "chand=%p calld=%p: starting pick on lb_policy=%p", + chand, calld, chand->lb_policy); + } + apply_service_config_to_call_locked(exec_ctx, elem); + // If the application explicitly set wait_for_ready, use that. + // Otherwise, if the service config specified a value for this + // method, use that. + uint32_t initial_metadata_flags = + calld->initial_metadata_batch->payload->send_initial_metadata + .send_initial_metadata_flags; + const bool wait_for_ready_set_from_api = + initial_metadata_flags & + GRPC_INITIAL_METADATA_WAIT_FOR_READY_EXPLICITLY_SET; + const bool wait_for_ready_set_from_service_config = + calld->method_params != NULL && + calld->method_params->wait_for_ready != WAIT_FOR_READY_UNSET; + if (!wait_for_ready_set_from_api && wait_for_ready_set_from_service_config) { + if (calld->method_params->wait_for_ready == WAIT_FOR_READY_TRUE) { + initial_metadata_flags |= GRPC_INITIAL_METADATA_WAIT_FOR_READY; + } else { + initial_metadata_flags &= ~GRPC_INITIAL_METADATA_WAIT_FOR_READY; + } + } + const grpc_lb_policy_pick_args inputs = { + calld->initial_metadata_batch->payload->send_initial_metadata + .send_initial_metadata, + initial_metadata_flags, &calld->lb_token_mdelem}; + // Keep a ref to the LB policy in calld while the pick is pending. + GRPC_LB_POLICY_REF(chand->lb_policy, "pick_subchannel"); + calld->lb_policy = chand->lb_policy; + GRPC_CLOSURE_INIT(&calld->lb_pick_closure, pick_callback_done_locked, elem, + grpc_combiner_scheduler(chand->combiner)); + const bool pick_done = grpc_lb_policy_pick_locked( + exec_ctx, chand->lb_policy, &inputs, &calld->connected_subchannel, + calld->subchannel_call_context, NULL, &calld->lb_pick_closure); + if (pick_done) { + /* synchronous grpc_lb_policy_pick call. Unref the LB policy. */ + if (GRPC_TRACER_ON(grpc_client_channel_trace)) { + gpr_log(GPR_DEBUG, "chand=%p calld=%p: pick completed synchronously", + chand, calld); + } + GRPC_LB_POLICY_UNREF(exec_ctx, calld->lb_policy, "pick_subchannel"); + calld->lb_policy = NULL; + } else { + GRPC_CALL_STACK_REF(calld->owning_call, "pick_callback_cancel"); + grpc_call_combiner_set_notify_on_cancel( + exec_ctx, calld->call_combiner, + GRPC_CLOSURE_INIT(&calld->lb_pick_cancel_closure, + pick_callback_cancel_locked, elem, + grpc_combiner_scheduler(chand->combiner))); + } + return pick_done; +} typedef struct { grpc_call_element *elem; @@ -1069,17 +1173,17 @@ static void pick_after_resolver_result_cancel_locked(grpc_exec_ctx *exec_ctx, gpr_free(args); return; } - args->finished = true; - grpc_call_element *elem = args->elem; - channel_data *chand = (channel_data *)elem->channel_data; - call_data *calld = (call_data *)elem->call_data; // If we don't yet have a resolver result, then a closure for // pick_after_resolver_result_done_locked() will have been added to // chand->waiting_for_resolver_result_closures, and it may not be invoked // until after this call has been destroyed. We mark the operation as // finished, so that when pick_after_resolver_result_done_locked() // is called, it will be a no-op. We also immediately invoke - // subchannel_ready_locked() to propagate the error back to the caller. + // async_pick_done_locked() to propagate the error back to the caller. + args->finished = true; + grpc_call_element *elem = args->elem; + channel_data *chand = (channel_data *)elem->channel_data; + call_data *calld = (call_data *)elem->call_data; if (GRPC_TRACER_ON(grpc_client_channel_trace)) { gpr_log(GPR_DEBUG, "chand=%p calld=%p: cancelling pick waiting for resolver result", @@ -1087,12 +1191,12 @@ static void pick_after_resolver_result_cancel_locked(grpc_exec_ctx *exec_ctx, } // Note: Although we are not in the call combiner here, we are // basically stealing the call combiner from the pending pick, so - // it's safe to call subchannel_ready_locked() here -- we are + // it's safe to call async_pick_done_locked() here -- we are // essentially calling it here instead of calling it in // pick_after_resolver_result_done_locked(). - subchannel_ready_locked(exec_ctx, elem, - GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING( - "Pick cancelled", &error, 1)); + async_pick_done_locked(exec_ctx, elem, + GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING( + "Pick cancelled", &error, 1)); } static void pick_after_resolver_result_done_locked(grpc_exec_ctx *exec_ctx, @@ -1117,14 +1221,19 @@ static void pick_after_resolver_result_done_locked(grpc_exec_ctx *exec_ctx, gpr_log(GPR_DEBUG, "chand=%p calld=%p: resolver failed to return data", chand, calld); } - subchannel_ready_locked(exec_ctx, elem, GRPC_ERROR_REF(error)); + async_pick_done_locked(exec_ctx, elem, GRPC_ERROR_REF(error)); } else { if (GRPC_TRACER_ON(grpc_client_channel_trace)) { gpr_log(GPR_DEBUG, "chand=%p calld=%p: resolver returned, doing pick", chand, calld); } - if (pick_subchannel_locked(exec_ctx, elem)) { - subchannel_ready_locked(exec_ctx, elem, GRPC_ERROR_NONE); + if (pick_callback_start_locked(exec_ctx, elem)) { + // Even if the LB policy returns a result synchronously, we have + // already added our polling entity to chand->interested_parties + // in order to wait for the resolver result, so we need to + // remove it here. Therefore, we call async_pick_done_locked() + // instead of pick_done_locked(). + async_pick_done_locked(exec_ctx, elem, GRPC_ERROR_NONE); } } } @@ -1152,154 +1261,38 @@ static void pick_after_resolver_result_start_locked(grpc_exec_ctx *exec_ctx, grpc_combiner_scheduler(chand->combiner))); } -// Note: This runs under the client_channel combiner, but will NOT be -// holding the call combiner. -static void pick_callback_cancel_locked(grpc_exec_ctx *exec_ctx, void *arg, - grpc_error *error) { - grpc_call_element *elem = (grpc_call_element *)arg; - channel_data *chand = (channel_data *)elem->channel_data; - call_data *calld = (call_data *)elem->call_data; - if (error != GRPC_ERROR_NONE && calld->lb_policy != NULL) { - if (GRPC_TRACER_ON(grpc_client_channel_trace)) { - gpr_log(GPR_DEBUG, "chand=%p calld=%p: cancelling pick from LB policy %p", - chand, calld, calld->lb_policy); - } - grpc_lb_policy_cancel_pick_locked(exec_ctx, calld->lb_policy, - &calld->connected_subchannel, - GRPC_ERROR_REF(error)); - } - GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call, "pick_callback_cancel"); -} - -// Callback invoked by grpc_lb_policy_pick_locked() for async picks. -// Unrefs the LB policy and invokes subchannel_ready_locked(). -static void pick_callback_done_locked(grpc_exec_ctx *exec_ctx, void *arg, - grpc_error *error) { +static void start_pick_locked(grpc_exec_ctx *exec_ctx, void *arg, + grpc_error *ignored) { grpc_call_element *elem = (grpc_call_element *)arg; - channel_data *chand = (channel_data *)elem->channel_data; call_data *calld = (call_data *)elem->call_data; - if (GRPC_TRACER_ON(grpc_client_channel_trace)) { - gpr_log(GPR_DEBUG, "chand=%p calld=%p: pick completed asynchronously", - chand, calld); - } - GPR_ASSERT(calld->lb_policy != NULL); - GRPC_LB_POLICY_UNREF(exec_ctx, calld->lb_policy, "pick_subchannel"); - calld->lb_policy = NULL; - subchannel_ready_locked(exec_ctx, elem, GRPC_ERROR_REF(error)); -} - -// Takes a ref to chand->lb_policy and calls grpc_lb_policy_pick_locked(). -// If the pick was completed synchronously, unrefs the LB policy and -// returns true. -static bool pick_callback_start_locked(grpc_exec_ctx *exec_ctx, - grpc_call_element *elem, - const grpc_lb_policy_pick_args *inputs) { channel_data *chand = (channel_data *)elem->channel_data; - call_data *calld = (call_data *)elem->call_data; - if (GRPC_TRACER_ON(grpc_client_channel_trace)) { - gpr_log(GPR_DEBUG, "chand=%p calld=%p: starting pick on lb_policy=%p", - chand, calld, chand->lb_policy); - } - // Keep a ref to the LB policy in calld while the pick is pending. - GRPC_LB_POLICY_REF(chand->lb_policy, "pick_subchannel"); - calld->lb_policy = chand->lb_policy; - GRPC_CLOSURE_INIT(&calld->lb_pick_closure, pick_callback_done_locked, elem, - grpc_combiner_scheduler(chand->combiner)); - const bool pick_done = grpc_lb_policy_pick_locked( - exec_ctx, chand->lb_policy, inputs, &calld->connected_subchannel, - calld->subchannel_call_context, NULL, &calld->lb_pick_closure); - if (pick_done) { - /* synchronous grpc_lb_policy_pick call. Unref the LB policy. */ - if (GRPC_TRACER_ON(grpc_client_channel_trace)) { - gpr_log(GPR_DEBUG, "chand=%p calld=%p: pick completed synchronously", - chand, calld); + GPR_ASSERT(calld->connected_subchannel == NULL); + if (chand->lb_policy != NULL) { + // We already have an LB policy, so ask it for a pick. + if (pick_callback_start_locked(exec_ctx, elem)) { + // Pick completed synchronously. + pick_done_locked(exec_ctx, elem, GRPC_ERROR_NONE); + return; } - GRPC_LB_POLICY_UNREF(exec_ctx, calld->lb_policy, "pick_subchannel"); - calld->lb_policy = NULL; } else { - GRPC_CALL_STACK_REF(calld->owning_call, "pick_callback_cancel"); - grpc_call_combiner_set_notify_on_cancel( - exec_ctx, calld->call_combiner, - GRPC_CLOSURE_INIT(&calld->lb_pick_cancel_closure, - pick_callback_cancel_locked, elem, - grpc_combiner_scheduler(chand->combiner))); - } - return pick_done; -} - -static bool pick_subchannel_locked(grpc_exec_ctx *exec_ctx, - grpc_call_element *elem) { - GPR_TIMER_BEGIN("pick_subchannel", 0); - channel_data *chand = (channel_data *)elem->channel_data; - call_data *calld = (call_data *)elem->call_data; - bool pick_done = false; - if (chand->lb_policy != NULL) { - apply_service_config_to_call_locked(exec_ctx, elem); - // If the application explicitly set wait_for_ready, use that. - // Otherwise, if the service config specified a value for this - // method, use that. - uint32_t initial_metadata_flags = - calld->initial_metadata_batch->payload->send_initial_metadata - .send_initial_metadata_flags; - const bool wait_for_ready_set_from_api = - initial_metadata_flags & - GRPC_INITIAL_METADATA_WAIT_FOR_READY_EXPLICITLY_SET; - const bool wait_for_ready_set_from_service_config = - calld->method_params != NULL && - calld->method_params->wait_for_ready != WAIT_FOR_READY_UNSET; - if (!wait_for_ready_set_from_api && - wait_for_ready_set_from_service_config) { - if (calld->method_params->wait_for_ready == WAIT_FOR_READY_TRUE) { - initial_metadata_flags |= GRPC_INITIAL_METADATA_WAIT_FOR_READY; - } else { - initial_metadata_flags &= ~GRPC_INITIAL_METADATA_WAIT_FOR_READY; - } + // We do not yet have an LB policy, so wait for a resolver result. + if (chand->resolver == NULL) { + pick_done_locked(exec_ctx, elem, + GRPC_ERROR_CREATE_FROM_STATIC_STRING("Disconnected")); + return; } - const grpc_lb_policy_pick_args inputs = { - calld->initial_metadata_batch->payload->send_initial_metadata - .send_initial_metadata, - initial_metadata_flags, &calld->lb_token_mdelem}; - pick_done = pick_callback_start_locked(exec_ctx, elem, &inputs); - } else if (chand->resolver != NULL) { if (!chand->started_resolving) { start_resolving_locked(exec_ctx, chand); } pick_after_resolver_result_start_locked(exec_ctx, elem); - } else { - subchannel_ready_locked( - exec_ctx, elem, GRPC_ERROR_CREATE_FROM_STATIC_STRING("Disconnected")); - } - GPR_TIMER_END("pick_subchannel", 0); - return pick_done; -} - -static void start_pick_locked(grpc_exec_ctx *exec_ctx, void *arg, - grpc_error *error_ignored) { - GPR_TIMER_BEGIN("start_pick_locked", 0); - grpc_call_element *elem = (grpc_call_element *)arg; - call_data *calld = (call_data *)elem->call_data; - channel_data *chand = (channel_data *)elem->channel_data; - GPR_ASSERT(calld->connected_subchannel == NULL); - if (pick_subchannel_locked(exec_ctx, elem)) { - // Pick was returned synchronously. - if (calld->connected_subchannel == NULL) { - GRPC_ERROR_UNREF(calld->error); - calld->error = GRPC_ERROR_CREATE_FROM_STATIC_STRING( - "Call dropped by load balancing policy"); - waiting_for_pick_batches_fail(exec_ctx, elem, - GRPC_ERROR_REF(calld->error)); - } else { - // Create subchannel call. - create_subchannel_call_locked(exec_ctx, elem, GRPC_ERROR_NONE); - } - } else { - // Pick will be done asynchronously. Add the call's polling entity to - // the channel's interested_parties, so that I/O for the resolver - // and LB policy can be done under it. - grpc_polling_entity_add_to_pollset_set(exec_ctx, calld->pollent, - chand->interested_parties); } - GPR_TIMER_END("start_pick_locked", 0); + // We need to wait for either a resolver result or for an async result + // from the LB policy. Add the polling entity from call_data to the + // channel_data's interested_parties, so that the I/O of the LB policy + // and resolver can be done under it. The polling entity will be + // removed in async_pick_done_locked(). + grpc_polling_entity_add_to_pollset_set(exec_ctx, calld->pollent, + chand->interested_parties); } static void on_complete(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { @@ -1394,7 +1387,8 @@ static void cc_start_transport_stream_op_batch( // combiner to start a pick. if (batch->send_initial_metadata) { if (GRPC_TRACER_ON(grpc_client_channel_trace)) { - gpr_log(GPR_DEBUG, "chand=%p calld=%p: entering combiner", chand, calld); + gpr_log(GPR_DEBUG, "chand=%p calld=%p: entering client_channel combiner", + chand, calld); } GRPC_CLOSURE_SCHED( exec_ctx, diff --git a/src/core/ext/filters/client_channel/client_channel_factory.c b/src/core/ext/filters/client_channel/client_channel_factory.c index 7220a8639e..57eac8f875 100644 --- a/src/core/ext/filters/client_channel/client_channel_factory.c +++ b/src/core/ext/filters/client_channel/client_channel_factory.c @@ -43,14 +43,13 @@ grpc_channel* grpc_client_channel_factory_create_channel( } static void* factory_arg_copy(void* factory) { - grpc_client_channel_factory_ref(factory); + grpc_client_channel_factory_ref((grpc_client_channel_factory*)factory); return factory; } static void factory_arg_destroy(grpc_exec_ctx* exec_ctx, void* factory) { - // TODO(roth): Remove local exec_ctx when - // https://github.com/grpc/grpc/pull/8705 is merged. - grpc_client_channel_factory_unref(exec_ctx, factory); + grpc_client_channel_factory_unref(exec_ctx, + (grpc_client_channel_factory*)factory); } static int factory_arg_cmp(void* factory1, void* factory2) { @@ -64,6 +63,6 @@ static const grpc_arg_pointer_vtable factory_arg_vtable = { grpc_arg grpc_client_channel_factory_create_channel_arg( grpc_client_channel_factory* factory) { - return grpc_channel_arg_pointer_create(GRPC_ARG_CLIENT_CHANNEL_FACTORY, + return grpc_channel_arg_pointer_create((char*)GRPC_ARG_CLIENT_CHANNEL_FACTORY, factory, &factory_arg_vtable); } diff --git a/src/core/ext/filters/client_channel/client_channel_plugin.c b/src/core/ext/filters/client_channel/client_channel_plugin.c index c32e83d012..1f71c5a7f9 100644 --- a/src/core/ext/filters/client_channel/client_channel_plugin.c +++ b/src/core/ext/filters/client_channel/client_channel_plugin.c @@ -54,8 +54,8 @@ static bool set_default_host_if_unset(grpc_exec_ctx *exec_ctx, char *default_authority = grpc_get_default_authority( exec_ctx, grpc_channel_stack_builder_get_target(builder)); if (default_authority != NULL) { - grpc_arg arg = grpc_channel_arg_string_create(GRPC_ARG_DEFAULT_AUTHORITY, - default_authority); + grpc_arg arg = grpc_channel_arg_string_create( + (char *)GRPC_ARG_DEFAULT_AUTHORITY, default_authority); grpc_channel_args *new_args = grpc_channel_args_copy_and_add(args, &arg, 1); grpc_channel_stack_builder_set_channel_arguments(exec_ctx, builder, new_args); diff --git a/src/core/ext/filters/client_channel/http_connect_handshaker.c b/src/core/ext/filters/client_channel/http_connect_handshaker.c index d4c3bc0ad3..418bb41ef6 100644 --- a/src/core/ext/filters/client_channel/http_connect_handshaker.c +++ b/src/core/ext/filters/client_channel/http_connect_handshaker.c @@ -309,7 +309,7 @@ static void http_connect_handshaker_do_handshake( grpc_httpcli_request request; memset(&request, 0, sizeof(request)); request.host = server_name; - request.http.method = "CONNECT"; + request.http.method = (char*)"CONNECT"; request.http.path = server_name; request.http.hdrs = headers; request.http.hdr_count = num_headers; diff --git a/src/core/ext/filters/client_channel/http_proxy.c b/src/core/ext/filters/client_channel/http_proxy.c index ef3512ed83..c507a2750e 100644 --- a/src/core/ext/filters/client_channel/http_proxy.c +++ b/src/core/ext/filters/client_channel/http_proxy.c @@ -44,6 +44,8 @@ static char* get_http_proxy_server(grpc_exec_ctx* exec_ctx, char** user_cred) { GPR_ASSERT(user_cred != NULL); char* proxy_name = NULL; char* uri_str = gpr_getenv("http_proxy"); + char** authority_strs = NULL; + size_t authority_nstrs; if (uri_str == NULL) return NULL; grpc_uri* uri = grpc_uri_parse(exec_ctx, uri_str, false /* suppress_errors */); @@ -56,8 +58,6 @@ static char* get_http_proxy_server(grpc_exec_ctx* exec_ctx, char** user_cred) { goto done; } /* Split on '@' to separate user credentials from host */ - char** authority_strs = NULL; - size_t authority_nstrs; gpr_string_split(uri->authority, "@", &authority_strs, &authority_nstrs); GPR_ASSERT(authority_nstrs != 0); /* should have at least 1 string */ if (authority_nstrs == 1) { @@ -157,7 +157,7 @@ static bool proxy_mapper_map_name(grpc_exec_ctx* exec_ctx, } grpc_arg args_to_add[2]; args_to_add[0] = grpc_channel_arg_string_create( - GRPC_ARG_HTTP_CONNECT_SERVER, + (char*)GRPC_ARG_HTTP_CONNECT_SERVER, uri->path[0] == '/' ? uri->path + 1 : uri->path); if (user_cred != NULL) { /* Use base64 encoding for user credentials as stated in RFC 7617 */ @@ -166,8 +166,8 @@ static bool proxy_mapper_map_name(grpc_exec_ctx* exec_ctx, char* header; gpr_asprintf(&header, "Proxy-Authorization:Basic %s", encoded_user_cred); gpr_free(encoded_user_cred); - args_to_add[1] = - grpc_channel_arg_string_create(GRPC_ARG_HTTP_CONNECT_HEADERS, header); + args_to_add[1] = grpc_channel_arg_string_create( + (char*)GRPC_ARG_HTTP_CONNECT_HEADERS, header); *new_args = grpc_channel_args_copy_and_add(args, args_to_add, 2); gpr_free(header); } else { diff --git a/src/core/ext/filters/client_channel/lb_policy/grpclb/client_load_reporting_filter.c b/src/core/ext/filters/client_channel/lb_policy/grpclb/client_load_reporting_filter.c index bd290464c8..7ad322902b 100644 --- a/src/core/ext/filters/client_channel/lb_policy/grpclb/client_load_reporting_filter.c +++ b/src/core/ext/filters/client_channel/lb_policy/grpclb/client_load_reporting_filter.c @@ -75,7 +75,8 @@ static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx, GPR_ASSERT(args->context != NULL); GPR_ASSERT(args->context[GRPC_GRPCLB_CLIENT_STATS].value != NULL); calld->client_stats = grpc_grpclb_client_stats_ref( - args->context[GRPC_GRPCLB_CLIENT_STATS].value); + (grpc_grpclb_client_stats *)args->context[GRPC_GRPCLB_CLIENT_STATS] + .value); // Record call started. grpc_grpclb_client_stats_add_call_started(calld->client_stats); return GRPC_ERROR_NONE; diff --git a/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.c b/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.c index 5aafed1374..85ef7894ea 100644 --- a/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.c +++ b/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.c @@ -101,6 +101,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/subchannel_index.h" #include "src/core/lib/channel/channel_args.h" #include "src/core/lib/channel/channel_stack.h" #include "src/core/lib/iomgr/combiner.h" @@ -137,7 +138,7 @@ static grpc_error *initial_metadata_add_lb_token( } static void destroy_client_stats(void *arg) { - grpc_grpclb_client_stats_unref(arg); + grpc_grpclb_client_stats_unref((grpc_grpclb_client_stats *)arg); } typedef struct wrapped_rr_closure_arg { @@ -285,7 +286,7 @@ static void add_pending_ping(pending_ping **root, grpc_closure *notify) { * glb_lb_policy */ typedef struct rr_connectivity_data rr_connectivity_data; -static const grpc_lb_policy_vtable glb_lb_policy_vtable; + typedef struct glb_lb_policy { /** base policy: must be first */ grpc_lb_policy base; @@ -727,7 +728,7 @@ static void create_rr_locked(grpc_exec_ctx *exec_ctx, glb_lb_policy *glb_policy, /* Allocate the data for the tracking of the new RR policy's connectivity. * It'll be deallocated in glb_rr_connectivity_changed() */ rr_connectivity_data *rr_connectivity = - gpr_zalloc(sizeof(rr_connectivity_data)); + (rr_connectivity_data *)gpr_zalloc(sizeof(rr_connectivity_data)); GRPC_CLOSURE_INIT(&rr_connectivity->on_change, glb_rr_connectivity_changed_locked, rr_connectivity, grpc_combiner_scheduler(glb_policy->base.combiner)); @@ -869,7 +870,8 @@ static grpc_channel_args *build_lb_channel_args( grpc_lb_addresses *lb_addresses = grpc_lb_addresses_create(num_grpclb_addrs, NULL); grpc_slice_hash_table_entry *targets_info_entries = - gpr_zalloc(sizeof(*targets_info_entries) * num_grpclb_addrs); + (grpc_slice_hash_table_entry *)gpr_zalloc(sizeof(*targets_info_entries) * + num_grpclb_addrs); size_t lb_addresses_idx = 0; for (size_t i = 0; i < addresses->num_addresses; ++i) { @@ -911,92 +913,6 @@ static grpc_channel_args *build_lb_channel_args( return result; } -static void glb_lb_channel_on_connectivity_changed_cb(grpc_exec_ctx *exec_ctx, - void *arg, - grpc_error *error); -static grpc_lb_policy *glb_create(grpc_exec_ctx *exec_ctx, - grpc_lb_policy_factory *factory, - grpc_lb_policy_args *args) { - /* Count the number of gRPC-LB addresses. There must be at least one. - * TODO(roth): For now, we ignore non-balancer addresses, but in the - * future, we may change the behavior such that we fall back to using - * the non-balancer addresses if we cannot reach any balancers. In the - * fallback case, we should use the LB policy indicated by - * GRPC_ARG_LB_POLICY_NAME (although if that specifies grpclb or is - * unset, we should default to pick_first). */ - const grpc_arg *arg = - grpc_channel_args_find(args->args, GRPC_ARG_LB_ADDRESSES); - if (arg == NULL || arg->type != GRPC_ARG_POINTER) { - return NULL; - } - grpc_lb_addresses *addresses = (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; - } - if (num_grpclb_addrs == 0) return NULL; - - glb_lb_policy *glb_policy = (glb_lb_policy *)gpr_zalloc(sizeof(*glb_policy)); - - /* Get server name. */ - arg = grpc_channel_args_find(args->args, GRPC_ARG_SERVER_URI); - GPR_ASSERT(arg != NULL); - GPR_ASSERT(arg->type == GRPC_ARG_STRING); - grpc_uri *uri = grpc_uri_parse(exec_ctx, arg->value.string, true); - GPR_ASSERT(uri->path[0] != '\0'); - glb_policy->server_name = - gpr_strdup(uri->path[0] == '/' ? uri->path + 1 : uri->path); - if (GRPC_TRACER_ON(grpc_lb_glb_trace)) { - gpr_log(GPR_INFO, "Will use '%s' as the server name for LB request.", - glb_policy->server_name); - } - grpc_uri_destroy(uri); - - glb_policy->cc_factory = args->client_channel_factory; - GPR_ASSERT(glb_policy->cc_factory != NULL); - - arg = grpc_channel_args_find(args->args, GRPC_ARG_GRPCLB_CALL_TIMEOUT_MS); - glb_policy->lb_call_timeout_ms = - grpc_channel_arg_get_integer(arg, (grpc_integer_options){0, 0, INT_MAX}); - - // Make sure that GRPC_ARG_LB_POLICY_NAME is set in channel args, - // since we use this to trigger the client_load_reporting filter. - grpc_arg new_arg = - grpc_channel_arg_string_create(GRPC_ARG_LB_POLICY_NAME, "grpclb"); - static const char *args_to_remove[] = {GRPC_ARG_LB_POLICY_NAME}; - glb_policy->args = grpc_channel_args_copy_and_add_and_remove( - args->args, args_to_remove, GPR_ARRAY_SIZE(args_to_remove), &new_arg, 1); - - /* Create a client channel over them to communicate with a LB service */ - glb_policy->response_generator = - grpc_fake_resolver_response_generator_create(); - grpc_channel_args *lb_channel_args = build_lb_channel_args( - exec_ctx, addresses, glb_policy->response_generator, args->args); - char *uri_str; - gpr_asprintf(&uri_str, "fake:///%s", glb_policy->server_name); - glb_policy->lb_channel = grpc_lb_policy_grpclb_create_lb_channel( - exec_ctx, uri_str, args->client_channel_factory, lb_channel_args); - - /* Propagate initial resolution */ - grpc_fake_resolver_response_generator_set_response( - exec_ctx, glb_policy->response_generator, lb_channel_args); - grpc_channel_args_destroy(exec_ctx, lb_channel_args); - gpr_free(uri_str); - if (glb_policy->lb_channel == NULL) { - gpr_free((void *)glb_policy->server_name); - grpc_channel_args_destroy(exec_ctx, glb_policy->args); - gpr_free(glb_policy); - return NULL; - } - GRPC_CLOSURE_INIT(&glb_policy->lb_channel_on_connectivity_changed, - glb_lb_channel_on_connectivity_changed_cb, glb_policy, - grpc_combiner_scheduler(args->combiner)); - grpc_lb_policy_init(&glb_policy->base, &glb_lb_policy_vtable, args->combiner); - grpc_connectivity_state_init(&glb_policy->state_tracker, GRPC_CHANNEL_IDLE, - "grpclb"); - return &glb_policy->base; -} - static void glb_destroy(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) { glb_lb_policy *glb_policy = (glb_lb_policy *)pol; GPR_ASSERT(glb_policy->pending_picks == NULL); @@ -1011,6 +927,7 @@ static void glb_destroy(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) { grpc_grpclb_destroy_serverlist(glb_policy->serverlist); } grpc_fake_resolver_response_generator_unref(glb_policy->response_generator); + grpc_subchannel_index_unref(); if (glb_policy->pending_update_args != NULL) { grpc_channel_args_destroy(exec_ctx, glb_policy->pending_update_args->args); gpr_free(glb_policy->pending_update_args); @@ -1303,7 +1220,8 @@ static void do_send_client_load_report_locked(grpc_exec_ctx *exec_ctx, static bool load_report_counters_are_zero(grpc_grpclb_request *request) { grpc_grpclb_dropped_call_counts *drop_entries = - request->client_stats.calls_finished_with_drop.arg; + (grpc_grpclb_dropped_call_counts *) + request->client_stats.calls_finished_with_drop.arg; return request->client_stats.num_calls_started == 0 && request->client_stats.num_calls_finished == 0 && request->client_stats.num_calls_finished_with_client_failed_to_send == @@ -1642,6 +1560,9 @@ static void lb_on_response_received_locked(grpc_exec_ctx *exec_ctx, void *arg, exec_ctx, glb_policy->lb_call, ops, (size_t)(op - ops), &glb_policy->lb_on_response_received); /* loop */ GPR_ASSERT(GRPC_CALL_OK == call_error); + } else { + GRPC_LB_POLICY_WEAK_UNREF(exec_ctx, &glb_policy->base, + "lb_on_response_received_locked_shutdown"); } } else { /* empty payload: call cancelled. */ /* dispose of the "lb_on_response_received_locked" weak ref taken in @@ -1865,6 +1786,90 @@ static const grpc_lb_policy_vtable glb_lb_policy_vtable = { glb_notify_on_state_change_locked, glb_update_locked}; +static grpc_lb_policy *glb_create(grpc_exec_ctx *exec_ctx, + grpc_lb_policy_factory *factory, + grpc_lb_policy_args *args) { + /* Count the number of gRPC-LB addresses. There must be at least one. + * TODO(roth): For now, we ignore non-balancer addresses, but in the + * future, we may change the behavior such that we fall back to using + * the non-balancer addresses if we cannot reach any balancers. In the + * fallback case, we should use the LB policy indicated by + * GRPC_ARG_LB_POLICY_NAME (although if that specifies grpclb or is + * unset, we should default to pick_first). */ + const grpc_arg *arg = + grpc_channel_args_find(args->args, GRPC_ARG_LB_ADDRESSES); + if (arg == NULL || arg->type != GRPC_ARG_POINTER) { + return NULL; + } + grpc_lb_addresses *addresses = (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; + } + if (num_grpclb_addrs == 0) return NULL; + + glb_lb_policy *glb_policy = (glb_lb_policy *)gpr_zalloc(sizeof(*glb_policy)); + + /* Get server name. */ + arg = grpc_channel_args_find(args->args, GRPC_ARG_SERVER_URI); + GPR_ASSERT(arg != NULL); + GPR_ASSERT(arg->type == GRPC_ARG_STRING); + grpc_uri *uri = grpc_uri_parse(exec_ctx, arg->value.string, true); + GPR_ASSERT(uri->path[0] != '\0'); + glb_policy->server_name = + gpr_strdup(uri->path[0] == '/' ? uri->path + 1 : uri->path); + if (GRPC_TRACER_ON(grpc_lb_glb_trace)) { + gpr_log(GPR_INFO, "Will use '%s' as the server name for LB request.", + glb_policy->server_name); + } + grpc_uri_destroy(uri); + + glb_policy->cc_factory = args->client_channel_factory; + GPR_ASSERT(glb_policy->cc_factory != NULL); + + arg = grpc_channel_args_find(args->args, GRPC_ARG_GRPCLB_CALL_TIMEOUT_MS); + glb_policy->lb_call_timeout_ms = + grpc_channel_arg_get_integer(arg, (grpc_integer_options){0, 0, INT_MAX}); + + // Make sure that GRPC_ARG_LB_POLICY_NAME is set in channel args, + // since we use this to trigger the client_load_reporting filter. + grpc_arg new_arg = grpc_channel_arg_string_create( + (char *)GRPC_ARG_LB_POLICY_NAME, (char *)"grpclb"); + static const char *args_to_remove[] = {GRPC_ARG_LB_POLICY_NAME}; + glb_policy->args = grpc_channel_args_copy_and_add_and_remove( + args->args, args_to_remove, GPR_ARRAY_SIZE(args_to_remove), &new_arg, 1); + + /* Create a client channel over them to communicate with a LB service */ + glb_policy->response_generator = + grpc_fake_resolver_response_generator_create(); + grpc_channel_args *lb_channel_args = build_lb_channel_args( + exec_ctx, addresses, glb_policy->response_generator, args->args); + char *uri_str; + gpr_asprintf(&uri_str, "fake:///%s", glb_policy->server_name); + glb_policy->lb_channel = grpc_lb_policy_grpclb_create_lb_channel( + exec_ctx, uri_str, args->client_channel_factory, lb_channel_args); + + /* Propagate initial resolution */ + grpc_fake_resolver_response_generator_set_response( + exec_ctx, glb_policy->response_generator, lb_channel_args); + grpc_channel_args_destroy(exec_ctx, lb_channel_args); + gpr_free(uri_str); + if (glb_policy->lb_channel == NULL) { + gpr_free((void *)glb_policy->server_name); + grpc_channel_args_destroy(exec_ctx, glb_policy->args); + gpr_free(glb_policy); + return NULL; + } + grpc_subchannel_index_ref(); + GRPC_CLOSURE_INIT(&glb_policy->lb_channel_on_connectivity_changed, + glb_lb_channel_on_connectivity_changed_cb, glb_policy, + grpc_combiner_scheduler(args->combiner)); + grpc_lb_policy_init(&glb_policy->base, &glb_lb_policy_vtable, args->combiner); + grpc_connectivity_state_init(&glb_policy->state_tracker, GRPC_CHANNEL_IDLE, + "grpclb"); + return &glb_policy->base; +} + static void glb_factory_ref(grpc_lb_policy_factory *factory) {} static void glb_factory_unref(grpc_lb_policy_factory *factory) {} diff --git a/src/core/ext/filters/client_channel/lb_policy/grpclb/load_balancer_api.c b/src/core/ext/filters/client_channel/lb_policy/grpclb/load_balancer_api.c index 407bd18adb..8ef6dfc6f4 100644 --- a/src/core/ext/filters/client_channel/lb_policy/grpclb/load_balancer_api.c +++ b/src/core/ext/filters/client_channel/lb_policy/grpclb/load_balancer_api.c @@ -148,7 +148,8 @@ grpc_slice grpc_grpclb_request_encode(const grpc_grpclb_request *request) { void grpc_grpclb_request_destroy(grpc_grpclb_request *request) { if (request->has_client_stats) { grpc_grpclb_dropped_call_counts *drop_entries = - request->client_stats.calls_finished_with_drop.arg; + (grpc_grpclb_dropped_call_counts *) + request->client_stats.calls_finished_with_drop.arg; grpc_grpclb_dropped_call_counts_destroy(drop_entries); } gpr_free(request); @@ -170,7 +171,8 @@ grpc_grpclb_initial_response *grpc_grpclb_initial_response_parse( if (!res.has_initial_response) return NULL; grpc_grpclb_initial_response *initial_res = - gpr_malloc(sizeof(grpc_grpclb_initial_response)); + (grpc_grpclb_initial_response *)gpr_malloc( + sizeof(grpc_grpclb_initial_response)); memcpy(initial_res, &res.initial_response, sizeof(grpc_grpclb_initial_response)); diff --git a/src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.c b/src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.c index fab3073eb9..d20cbb8388 100644 --- a/src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.c +++ b/src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.c @@ -89,6 +89,7 @@ static void pf_destroy(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) { "picked_first_destroy"); } grpc_connectivity_state_destroy(exec_ctx, &p->state_tracker); + grpc_subchannel_index_unref(); if (p->pending_update_args != NULL) { grpc_channel_args_destroy(exec_ctx, p->pending_update_args->args); gpr_free(p->pending_update_args); @@ -330,8 +331,8 @@ static void pf_update_locked(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, gpr_log(GPR_INFO, "Pick First %p received update with %lu addresses", (void *)p, (unsigned long)addresses->num_addresses); } - grpc_subchannel_args *sc_args = - gpr_zalloc(sizeof(*sc_args) * addresses->num_addresses); + grpc_subchannel_args *sc_args = (grpc_subchannel_args *)gpr_zalloc( + sizeof(*sc_args) * addresses->num_addresses); /* We remove the following keys in order for subchannel keys belonging to * subchannels point to the same address to match. */ static const char *keys_to_remove[] = {GRPC_ARG_SUBCHANNEL_ADDRESS, @@ -403,7 +404,7 @@ static void pf_update_locked(grpc_exec_ctx *exec_ctx, grpc_lb_policy *policy, } /* Create the subchannels for the new subchannel args/addresses. */ grpc_subchannel **new_subchannels = - gpr_zalloc(sizeof(*new_subchannels) * sc_args_count); + (grpc_subchannel **)gpr_zalloc(sizeof(*new_subchannels) * sc_args_count); size_t num_new_subchannels = 0; for (size_t i = 0; i < sc_args_count; i++) { grpc_subchannel *subchannel = grpc_client_channel_factory_create_subchannel( @@ -686,6 +687,7 @@ static grpc_lb_policy *create_pick_first(grpc_exec_ctx *exec_ctx, } pf_update_locked(exec_ctx, &p->base, args); grpc_lb_policy_init(&p->base, &pick_first_lb_policy_vtable, args->combiner); + grpc_subchannel_index_ref(); GRPC_CLOSURE_INIT(&p->connectivity_changed, pf_connectivity_changed_locked, p, grpc_combiner_scheduler(args->combiner)); return &p->base; diff --git a/src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.c b/src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.c index be91d3d651..a3a62e9f3c 100644 --- a/src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.c +++ b/src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.c @@ -30,6 +30,7 @@ #include "src/core/ext/filters/client_channel/lb_policy_registry.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" #include "src/core/lib/debug/trace.h" #include "src/core/lib/iomgr/combiner.h" @@ -310,6 +311,7 @@ static void rr_destroy(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) { (void *)pol, (void *)pol); } grpc_connectivity_state_destroy(exec_ctx, &p->state_tracker); + grpc_subchannel_index_unref(); gpr_free(p); } @@ -587,7 +589,7 @@ static void rr_connectivity_changed_locked(grpc_exec_ctx *exec_ctx, void *arg, // Dispose of outdated subchannel lists. if (sd->subchannel_list != p->subchannel_list && sd->subchannel_list != p->latest_pending_subchannel_list) { - char *reason = NULL; + const char *reason = NULL; if (sd->subchannel_list->shutting_down) { reason = "sl_outdated_straggler"; rr_subchannel_list_unref(exec_ctx, sd->subchannel_list, reason); @@ -890,6 +892,7 @@ static grpc_lb_policy *round_robin_create(grpc_exec_ctx *exec_ctx, GPR_ASSERT(args->client_channel_factory != NULL); round_robin_lb_policy *p = (round_robin_lb_policy *)gpr_zalloc(sizeof(*p)); grpc_lb_policy_init(&p->base, &round_robin_lb_policy_vtable, args->combiner); + grpc_subchannel_index_ref(); grpc_connectivity_state_init(&p->state_tracker, GRPC_CHANNEL_IDLE, "round_robin"); rr_update_locked(exec_ctx, &p->base, args); diff --git a/src/core/ext/filters/client_channel/lb_policy_factory.c b/src/core/ext/filters/client_channel/lb_policy_factory.c index cdcaf17544..4d1405454c 100644 --- a/src/core/ext/filters/client_channel/lb_policy_factory.c +++ b/src/core/ext/filters/client_channel/lb_policy_factory.c @@ -126,13 +126,14 @@ void grpc_lb_addresses_destroy(grpc_exec_ctx* exec_ctx, } static void* lb_addresses_copy(void* addresses) { - return grpc_lb_addresses_copy(addresses); + return grpc_lb_addresses_copy((grpc_lb_addresses*)addresses); } static void lb_addresses_destroy(grpc_exec_ctx* exec_ctx, void* addresses) { - grpc_lb_addresses_destroy(exec_ctx, addresses); + grpc_lb_addresses_destroy(exec_ctx, (grpc_lb_addresses*)addresses); } static int lb_addresses_cmp(void* addresses1, void* addresses2) { - return grpc_lb_addresses_cmp(addresses1, addresses2); + return grpc_lb_addresses_cmp((grpc_lb_addresses*)addresses1, + (grpc_lb_addresses*)addresses2); } static const grpc_arg_pointer_vtable lb_addresses_arg_vtable = { lb_addresses_copy, lb_addresses_destroy, lb_addresses_cmp}; @@ -140,7 +141,7 @@ static const grpc_arg_pointer_vtable lb_addresses_arg_vtable = { grpc_arg grpc_lb_addresses_create_channel_arg( const grpc_lb_addresses* addresses) { return grpc_channel_arg_pointer_create( - GRPC_ARG_LB_ADDRESSES, (void*)addresses, &lb_addresses_arg_vtable); + (char*)GRPC_ARG_LB_ADDRESSES, (void*)addresses, &lb_addresses_arg_vtable); } grpc_lb_addresses* grpc_lb_addresses_find_channel_arg( @@ -149,7 +150,7 @@ grpc_lb_addresses* grpc_lb_addresses_find_channel_arg( grpc_channel_args_find(channel_args, GRPC_ARG_LB_ADDRESSES); if (lb_addresses_arg == NULL || lb_addresses_arg->type != GRPC_ARG_POINTER) return NULL; - return lb_addresses_arg->value.pointer.p; + return (grpc_lb_addresses*)lb_addresses_arg->value.pointer.p; } void grpc_lb_policy_factory_ref(grpc_lb_policy_factory* factory) { diff --git a/src/core/ext/filters/client_channel/resolver/dns/c_ares/dns_resolver_ares.c b/src/core/ext/filters/client_channel/resolver/dns/c_ares/dns_resolver_ares.c index b87a3b7082..9bb229ad95 100644 --- a/src/core/ext/filters/client_channel/resolver/dns/c_ares/dns_resolver_ares.c +++ b/src/core/ext/filters/client_channel/resolver/dns/c_ares/dns_resolver_ares.c @@ -204,7 +204,7 @@ static char *choose_service_config(char *service_config_choice_json) { int random_pct = rand() % 100; int percentage; if (sscanf(field->value, "%d", &percentage) != 1 || - random_pct > percentage) { + random_pct > percentage || percentage == 0) { service_config_json = NULL; break; } @@ -249,7 +249,7 @@ static void dns_ares_on_resolved_locked(grpc_exec_ctx *exec_ctx, void *arg, 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( - GRPC_ARG_SERVICE_CONFIG, service_config_string); + (char *)GRPC_ARG_SERVICE_CONFIG, service_config_string); service_config = grpc_service_config_create(service_config_string); if (service_config != NULL) { const char *lb_policy_name = @@ -257,7 +257,7 @@ static void dns_ares_on_resolved_locked(grpc_exec_ctx *exec_ctx, void *arg, if (lb_policy_name != NULL) { args_to_remove[num_args_to_remove++] = GRPC_ARG_LB_POLICY_NAME; new_args[num_args_to_add++] = grpc_channel_arg_string_create( - GRPC_ARG_LB_POLICY_NAME, (char *)lb_policy_name); + (char *)GRPC_ARG_LB_POLICY_NAME, (char *)lb_policy_name); } } } diff --git a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver_posix.c b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver_posix.c index 9747d39a16..c30cc93b6f 100644 --- a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver_posix.c +++ b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver_posix.c @@ -20,6 +20,7 @@ #if GRPC_ARES == 1 && defined(GRPC_POSIX_SOCKET) #include <ares.h> +#include <sys/ioctl.h> #include "src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver.h" @@ -37,8 +38,6 @@ typedef struct fd_node { /** the owner of this fd node */ grpc_ares_ev_driver *ev_driver; - /** the grpc_fd owned by this fd node */ - grpc_fd *grpc_fd; /** a closure wrapping on_readable_cb, which should be invoked when the grpc_fd in this node becomes readable. */ grpc_closure read_closure; @@ -50,10 +49,14 @@ typedef struct fd_node { /** mutex guarding the rest of the state */ gpr_mu mu; + /** the grpc_fd owned by this fd node */ + grpc_fd *fd; /** if the readable closure has been registered */ bool readable_registered; /** if the writable closure has been registered */ bool writable_registered; + /** if the fd is being shut down */ + bool shutting_down; } fd_node; struct grpc_ares_ev_driver { @@ -96,19 +99,31 @@ static void grpc_ares_ev_driver_unref(grpc_ares_ev_driver *ev_driver) { } static void fd_node_destroy(grpc_exec_ctx *exec_ctx, fd_node *fdn) { - gpr_log(GPR_DEBUG, "delete fd: %d", grpc_fd_wrapped_fd(fdn->grpc_fd)); + gpr_log(GPR_DEBUG, "delete fd: %d", grpc_fd_wrapped_fd(fdn->fd)); GPR_ASSERT(!fdn->readable_registered); GPR_ASSERT(!fdn->writable_registered); gpr_mu_destroy(&fdn->mu); - grpc_pollset_set_del_fd(exec_ctx, fdn->ev_driver->pollset_set, fdn->grpc_fd); /* c-ares library has closed the fd inside grpc_fd. This fd may be picked up immediately by another thread, and should not be closed by the following grpc_fd_orphan. */ - grpc_fd_orphan(exec_ctx, fdn->grpc_fd, NULL, NULL, true /* already_closed */, + grpc_fd_orphan(exec_ctx, fdn->fd, NULL, NULL, true /* already_closed */, "c-ares query finished"); gpr_free(fdn); } +static void fd_node_shutdown(grpc_exec_ctx *exec_ctx, fd_node *fdn) { + gpr_mu_lock(&fdn->mu); + fdn->shutting_down = true; + if (!fdn->readable_registered && !fdn->writable_registered) { + gpr_mu_unlock(&fdn->mu); + fd_node_destroy(exec_ctx, fdn); + } else { + grpc_fd_shutdown(exec_ctx, fdn->fd, GRPC_ERROR_CREATE_FROM_STATIC_STRING( + "c-ares fd shutdown")); + gpr_mu_unlock(&fdn->mu); + } +} + grpc_error *grpc_ares_ev_driver_create(grpc_ares_ev_driver **ev_driver, grpc_pollset_set *pollset_set) { *ev_driver = (grpc_ares_ev_driver *)gpr_malloc(sizeof(grpc_ares_ev_driver)); @@ -150,9 +165,8 @@ void grpc_ares_ev_driver_shutdown(grpc_exec_ctx *exec_ctx, ev_driver->shutting_down = true; fd_node *fn = ev_driver->fds; while (fn != NULL) { - grpc_fd_shutdown( - exec_ctx, fn->grpc_fd, - GRPC_ERROR_CREATE_FROM_STATIC_STRING("grpc_ares_ev_driver_shutdown")); + grpc_fd_shutdown(exec_ctx, fn->fd, GRPC_ERROR_CREATE_FROM_STATIC_STRING( + "grpc_ares_ev_driver_shutdown")); fn = fn->next; } gpr_mu_unlock(&ev_driver->mu); @@ -165,7 +179,7 @@ static fd_node *pop_fd_node(fd_node **head, int fd) { dummy_head.next = *head; fd_node *node = &dummy_head; while (node->next != NULL) { - if (grpc_fd_wrapped_fd(node->next->grpc_fd) == fd) { + if (grpc_fd_wrapped_fd(node->next->fd) == fd) { fd_node *ret = node->next; node->next = node->next->next; *head = dummy_head.next; @@ -176,18 +190,33 @@ static fd_node *pop_fd_node(fd_node **head, int fd) { return NULL; } +/* Check if \a fd is still readable */ +static bool grpc_ares_is_fd_still_readable(grpc_ares_ev_driver *ev_driver, + int fd) { + size_t bytes_available = 0; + return ioctl(fd, FIONREAD, &bytes_available) == 0 && bytes_available > 0; +} + static void on_readable_cb(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { fd_node *fdn = (fd_node *)arg; grpc_ares_ev_driver *ev_driver = fdn->ev_driver; gpr_mu_lock(&fdn->mu); + const int fd = grpc_fd_wrapped_fd(fdn->fd); fdn->readable_registered = false; + if (fdn->shutting_down && !fdn->writable_registered) { + gpr_mu_unlock(&fdn->mu); + fd_node_destroy(exec_ctx, fdn); + grpc_ares_ev_driver_unref(ev_driver); + return; + } gpr_mu_unlock(&fdn->mu); - gpr_log(GPR_DEBUG, "readable on %d", grpc_fd_wrapped_fd(fdn->grpc_fd)); + gpr_log(GPR_DEBUG, "readable on %d", fd); if (error == GRPC_ERROR_NONE) { - ares_process_fd(ev_driver->channel, grpc_fd_wrapped_fd(fdn->grpc_fd), - ARES_SOCKET_BAD); + do { + ares_process_fd(ev_driver->channel, fd, ARES_SOCKET_BAD); + } while (grpc_ares_is_fd_still_readable(ev_driver, fd)); } else { // If error is not GRPC_ERROR_NONE, it means the fd has been shutdown or // timed out. The pending lookups made on this ev_driver will be cancelled @@ -208,13 +237,19 @@ static void on_writable_cb(grpc_exec_ctx *exec_ctx, void *arg, fd_node *fdn = (fd_node *)arg; grpc_ares_ev_driver *ev_driver = fdn->ev_driver; gpr_mu_lock(&fdn->mu); + const int fd = grpc_fd_wrapped_fd(fdn->fd); fdn->writable_registered = false; + if (fdn->shutting_down && !fdn->readable_registered) { + gpr_mu_unlock(&fdn->mu); + fd_node_destroy(exec_ctx, fdn); + grpc_ares_ev_driver_unref(ev_driver); + return; + } gpr_mu_unlock(&fdn->mu); - gpr_log(GPR_DEBUG, "writable on %d", grpc_fd_wrapped_fd(fdn->grpc_fd)); + gpr_log(GPR_DEBUG, "writable on %d", fd); if (error == GRPC_ERROR_NONE) { - ares_process_fd(ev_driver->channel, ARES_SOCKET_BAD, - grpc_fd_wrapped_fd(fdn->grpc_fd)); + ares_process_fd(ev_driver->channel, ARES_SOCKET_BAD, fd); } else { // If error is not GRPC_ERROR_NONE, it means the fd has been shutdown or // timed out. The pending lookups made on this ev_driver will be cancelled @@ -253,17 +288,17 @@ static void grpc_ares_notify_on_event_locked(grpc_exec_ctx *exec_ctx, gpr_asprintf(&fd_name, "ares_ev_driver-%" PRIuPTR, i); fdn = (fd_node *)gpr_malloc(sizeof(fd_node)); gpr_log(GPR_DEBUG, "new fd: %d", socks[i]); - fdn->grpc_fd = grpc_fd_create(socks[i], fd_name); + fdn->fd = grpc_fd_create(socks[i], fd_name); fdn->ev_driver = ev_driver; fdn->readable_registered = false; fdn->writable_registered = false; + fdn->shutting_down = false; gpr_mu_init(&fdn->mu); GRPC_CLOSURE_INIT(&fdn->read_closure, on_readable_cb, fdn, grpc_schedule_on_exec_ctx); GRPC_CLOSURE_INIT(&fdn->write_closure, on_writable_cb, fdn, grpc_schedule_on_exec_ctx); - grpc_pollset_set_add_fd(exec_ctx, ev_driver->pollset_set, - fdn->grpc_fd); + grpc_pollset_set_add_fd(exec_ctx, ev_driver->pollset_set, fdn->fd); gpr_free(fd_name); } fdn->next = new_list; @@ -274,9 +309,8 @@ static void grpc_ares_notify_on_event_locked(grpc_exec_ctx *exec_ctx, if (ARES_GETSOCK_READABLE(socks_bitmask, i) && !fdn->readable_registered) { grpc_ares_ev_driver_ref(ev_driver); - gpr_log(GPR_DEBUG, "notify read on: %d", - grpc_fd_wrapped_fd(fdn->grpc_fd)); - grpc_fd_notify_on_read(exec_ctx, fdn->grpc_fd, &fdn->read_closure); + gpr_log(GPR_DEBUG, "notify read on: %d", grpc_fd_wrapped_fd(fdn->fd)); + grpc_fd_notify_on_read(exec_ctx, fdn->fd, &fdn->read_closure); fdn->readable_registered = true; } // Register write_closure if the socket is writable and write_closure @@ -284,9 +318,9 @@ static void grpc_ares_notify_on_event_locked(grpc_exec_ctx *exec_ctx, if (ARES_GETSOCK_WRITABLE(socks_bitmask, i) && !fdn->writable_registered) { gpr_log(GPR_DEBUG, "notify write on: %d", - grpc_fd_wrapped_fd(fdn->grpc_fd)); + grpc_fd_wrapped_fd(fdn->fd)); grpc_ares_ev_driver_ref(ev_driver); - grpc_fd_notify_on_write(exec_ctx, fdn->grpc_fd, &fdn->write_closure); + grpc_fd_notify_on_write(exec_ctx, fdn->fd, &fdn->write_closure); fdn->writable_registered = true; } gpr_mu_unlock(&fdn->mu); @@ -299,7 +333,7 @@ static void grpc_ares_notify_on_event_locked(grpc_exec_ctx *exec_ctx, while (ev_driver->fds != NULL) { fd_node *cur = ev_driver->fds; ev_driver->fds = ev_driver->fds->next; - fd_node_destroy(exec_ctx, cur); + fd_node_shutdown(exec_ctx, cur); } ev_driver->fds = new_list; // If the ev driver has no working fd, all the tasks are done. diff --git a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.c b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.c index 0d71f3560e..04379975e1 100644 --- a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.c +++ b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.c @@ -123,8 +123,8 @@ static void grpc_ares_request_unref(grpc_exec_ctx *exec_ctx, static grpc_ares_hostbyname_request *create_hostbyname_request( grpc_ares_request *parent_request, char *host, uint16_t port, bool is_balancer) { - grpc_ares_hostbyname_request *hr = - gpr_zalloc(sizeof(grpc_ares_hostbyname_request)); + grpc_ares_hostbyname_request *hr = (grpc_ares_hostbyname_request *)gpr_zalloc( + sizeof(grpc_ares_hostbyname_request)); hr->parent_request = parent_request; hr->host = gpr_strdup(host); hr->port = port; @@ -174,7 +174,7 @@ static void on_hostbyname_done_cb(void *arg, int status, int timeouts, grpc_lb_addresses_set_address( *lb_addresses, i, &addr, addr_len, hr->is_balancer /* is_balancer */, - hr->is_balancer ? strdup(hr->host) : NULL /* balancer_name */, + hr->is_balancer ? hr->host : NULL /* balancer_name */, NULL /* user_data */); char output[INET6_ADDRSTRLEN]; ares_inet_ntop(AF_INET6, &addr.sin6_addr, output, INET6_ADDRSTRLEN); @@ -195,7 +195,7 @@ static void on_hostbyname_done_cb(void *arg, int status, int timeouts, grpc_lb_addresses_set_address( *lb_addresses, i, &addr, addr_len, hr->is_balancer /* is_balancer */, - hr->is_balancer ? strdup(hr->host) : NULL /* balancer_name */, + hr->is_balancer ? hr->host : NULL /* balancer_name */, NULL /* user_data */); char output[INET_ADDRSTRLEN]; ares_inet_ntop(AF_INET, &addr.sin_addr, output, INET_ADDRSTRLEN); @@ -275,14 +275,15 @@ static void on_txt_done_cb(void *arg, int status, int timeouts, gpr_log(GPR_DEBUG, "on_txt_done_cb"); char *error_msg; grpc_ares_request *r = (grpc_ares_request *)arg; + const size_t prefix_len = sizeof(g_service_config_attribute_prefix) - 1; + struct ares_txt_ext *result = NULL; + struct ares_txt_ext *reply = NULL; + grpc_error *error = GRPC_ERROR_NONE; gpr_mu_lock(&r->mu); if (status != ARES_SUCCESS) goto fail; - struct ares_txt_ext *reply = NULL; status = ares_parse_txt_reply_ext(buf, len, &reply); if (status != ARES_SUCCESS) goto fail; // Find service config in TXT record. - const size_t prefix_len = sizeof(g_service_config_attribute_prefix) - 1; - struct ares_txt_ext *result; for (result = reply; result != NULL; result = result->next) { if (result->record_start && memcmp(result->txt, g_service_config_attribute_prefix, prefix_len) == @@ -313,7 +314,7 @@ static void on_txt_done_cb(void *arg, int status, int timeouts, fail: gpr_asprintf(&error_msg, "C-ares TXT lookup status is not ARES_SUCCESS: %s", ares_strerror(status)); - grpc_error *error = GRPC_ERROR_CREATE_FROM_COPIED_STRING(error_msg); + error = GRPC_ERROR_CREATE_FROM_COPIED_STRING(error_msg); gpr_free(error_msg); if (r->error == GRPC_ERROR_NONE) { r->error = error; @@ -331,6 +332,9 @@ static grpc_ares_request *grpc_dns_lookup_ares_impl( grpc_closure *on_done, grpc_lb_addresses **addrs, bool check_grpclb, char **service_config_json) { grpc_error *error = GRPC_ERROR_NONE; + grpc_ares_hostbyname_request *hr = NULL; + grpc_ares_request *r = NULL; + ares_channel *channel = NULL; /* 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", @@ -360,8 +364,7 @@ static grpc_ares_request *grpc_dns_lookup_ares_impl( error = grpc_ares_ev_driver_create(&ev_driver, interested_parties); if (error != GRPC_ERROR_NONE) goto error_cleanup; - grpc_ares_request *r = - (grpc_ares_request *)gpr_zalloc(sizeof(grpc_ares_request)); + r = (grpc_ares_request *)gpr_zalloc(sizeof(grpc_ares_request)); gpr_mu_init(&r->mu); r->ev_driver = ev_driver; r->on_done = on_done; @@ -369,7 +372,7 @@ static grpc_ares_request *grpc_dns_lookup_ares_impl( r->service_config_json_out = service_config_json; r->success = false; r->error = GRPC_ERROR_NONE; - ares_channel *channel = grpc_ares_ev_driver_get_channel(r->ev_driver); + channel = grpc_ares_ev_driver_get_channel(r->ev_driver); // If dns_server is specified, use it. if (dns_server != NULL) { @@ -410,12 +413,12 @@ static grpc_ares_request *grpc_dns_lookup_ares_impl( } gpr_ref_init(&r->pending_queries, 1); if (grpc_ipv6_loopback_available()) { - grpc_ares_hostbyname_request *hr = create_hostbyname_request( - r, host, strhtons(port), false /* is_balancer */); + hr = create_hostbyname_request(r, host, strhtons(port), + false /* is_balancer */); ares_gethostbyname(*channel, hr->host, AF_INET6, on_hostbyname_done_cb, hr); } - grpc_ares_hostbyname_request *hr = create_hostbyname_request( - r, host, strhtons(port), false /* is_balancer */); + hr = create_hostbyname_request(r, host, strhtons(port), + false /* is_balancer */); ares_gethostbyname(*channel, hr->host, AF_INET, on_hostbyname_done_cb, hr); if (check_grpclb) { /* Query the SRV record */ @@ -527,7 +530,8 @@ static void grpc_resolve_address_ares_impl(grpc_exec_ctx *exec_ctx, grpc_closure *on_done, grpc_resolved_addresses **addrs) { grpc_resolve_address_ares_request *r = - gpr_zalloc(sizeof(grpc_resolve_address_ares_request)); + (grpc_resolve_address_ares_request *)gpr_zalloc( + sizeof(grpc_resolve_address_ares_request)); r->addrs_out = addrs; r->on_resolve_address_done = on_done; GRPC_CLOSURE_INIT(&r->on_dns_lookup_done, on_dns_lookup_done_cb, r, diff --git a/src/core/ext/filters/client_channel/resolver/fake/fake_resolver.c b/src/core/ext/filters/client_channel/resolver/fake/fake_resolver.c index 3ff081a514..69ea440ae6 100644 --- a/src/core/ext/filters/client_channel/resolver/fake/fake_resolver.c +++ b/src/core/ext/filters/client_channel/resolver/fake/fake_resolver.c @@ -159,7 +159,7 @@ typedef struct set_response_closure_arg { static void set_response_closure_fn(grpc_exec_ctx* exec_ctx, void* arg, grpc_error* error) { - set_response_closure_arg* closure_arg = arg; + set_response_closure_arg* closure_arg = (set_response_closure_arg*)arg; grpc_fake_resolver_response_generator* generator = closure_arg->generator; fake_resolver* r = generator->resolver; if (r->next_results != NULL) { @@ -178,7 +178,8 @@ void grpc_fake_resolver_response_generator_set_response( grpc_exec_ctx* exec_ctx, grpc_fake_resolver_response_generator* generator, grpc_channel_args* next_response) { GPR_ASSERT(generator->resolver != NULL); - set_response_closure_arg* closure_arg = gpr_zalloc(sizeof(*closure_arg)); + set_response_closure_arg* closure_arg = + (set_response_closure_arg*)gpr_zalloc(sizeof(*closure_arg)); closure_arg->generator = generator; closure_arg->next_response = grpc_channel_args_copy(next_response); GRPC_CLOSURE_SCHED(exec_ctx, @@ -209,7 +210,7 @@ grpc_arg grpc_fake_resolver_response_generator_arg( grpc_fake_resolver_response_generator* generator) { grpc_arg arg; arg.type = GRPC_ARG_POINTER; - arg.key = GRPC_ARG_FAKE_RESOLVER_RESPONSE_GENERATOR; + arg.key = (char*)GRPC_ARG_FAKE_RESOLVER_RESPONSE_GENERATOR; arg.value.pointer.p = generator; arg.value.pointer.vtable = &response_generator_arg_vtable; return arg; diff --git a/src/core/ext/filters/client_channel/retry_throttle.c b/src/core/ext/filters/client_channel/retry_throttle.c index 6cd6654b6f..09dcade089 100644 --- a/src/core/ext/filters/client_channel/retry_throttle.c +++ b/src/core/ext/filters/client_channel/retry_throttle.c @@ -99,7 +99,7 @@ static grpc_server_retry_throttle_data* grpc_server_retry_throttle_data_create( int max_milli_tokens, int milli_token_ratio, grpc_server_retry_throttle_data* old_throttle_data) { grpc_server_retry_throttle_data* throttle_data = - gpr_malloc(sizeof(*throttle_data)); + (grpc_server_retry_throttle_data*)gpr_malloc(sizeof(*throttle_data)); memset(throttle_data, 0, sizeof(*throttle_data)); gpr_ref_init(&throttle_data->refs, 1); throttle_data->max_milli_tokens = max_milli_tokens; @@ -131,11 +131,11 @@ static grpc_server_retry_throttle_data* grpc_server_retry_throttle_data_create( // static void* copy_server_name(void* key, void* unused) { - return gpr_strdup(key); + return gpr_strdup((const char*)key); } static long compare_server_name(void* key1, void* key2, void* unused) { - return strcmp(key1, key2); + return strcmp((const char*)key1, (const char*)key2); } static void destroy_server_retry_throttle_data(void* value, void* unused) { @@ -177,7 +177,8 @@ grpc_server_retry_throttle_data* grpc_retry_throttle_map_get_data_for_server( const char* server_name, int max_milli_tokens, int milli_token_ratio) { gpr_mu_lock(&g_mu); grpc_server_retry_throttle_data* throttle_data = - gpr_avl_get(g_avl, (char*)server_name, NULL); + (grpc_server_retry_throttle_data*)gpr_avl_get(g_avl, (char*)server_name, + NULL); if (throttle_data == NULL) { // Entry not found. Create a new one. throttle_data = grpc_server_retry_throttle_data_create( diff --git a/src/core/ext/filters/client_channel/subchannel.c b/src/core/ext/filters/client_channel/subchannel.c index 05c55aaa89..40a51c72d6 100644 --- a/src/core/ext/filters/client_channel/subchannel.c +++ b/src/core/ext/filters/client_channel/subchannel.c @@ -32,6 +32,7 @@ #include "src/core/ext/filters/client_channel/uri_parser.h" #include "src/core/lib/channel/channel_args.h" #include "src/core/lib/channel/connected_channel.h" +#include "src/core/lib/debug/stats.h" #include "src/core/lib/iomgr/sockaddr_utils.h" #include "src/core/lib/iomgr/timer.h" #include "src/core/lib/profiling/timers.h" @@ -290,6 +291,7 @@ grpc_subchannel *grpc_subchannel_create(grpc_exec_ctx *exec_ctx, return c; } + GRPC_STATS_INC_CLIENT_SUBCHANNELS_CREATED(exec_ctx); c = (grpc_subchannel *)gpr_zalloc(sizeof(*c)); c->key = key; gpr_atm_no_barrier_store(&c->ref_pair, 1 << INTERNAL_REF_BITS); @@ -809,6 +811,6 @@ const char *grpc_get_subchannel_address_uri_arg(const grpc_channel_args *args) { grpc_arg grpc_create_subchannel_address_arg(const grpc_resolved_address *addr) { return grpc_channel_arg_string_create( - GRPC_ARG_SUBCHANNEL_ADDRESS, + (char *)GRPC_ARG_SUBCHANNEL_ADDRESS, addr->len > 0 ? grpc_sockaddr_to_uri(addr) : gpr_strdup("")); } diff --git a/src/core/ext/filters/client_channel/subchannel_index.c b/src/core/ext/filters/client_channel/subchannel_index.c index f319aff0df..d7a51f3899 100644 --- a/src/core/ext/filters/client_channel/subchannel_index.c +++ b/src/core/ext/filters/client_channel/subchannel_index.c @@ -34,6 +34,8 @@ static gpr_avl g_subchannel_index; static gpr_mu g_mu; +static gpr_refcount g_refcount; + struct grpc_subchannel_key { grpc_subchannel_args args; }; @@ -88,24 +90,26 @@ void grpc_subchannel_key_destroy(grpc_exec_ctx *exec_ctx, static void sck_avl_destroy(void *p, void *user_data) { grpc_exec_ctx *exec_ctx = (grpc_exec_ctx *)user_data; - grpc_subchannel_key_destroy(exec_ctx, p); + grpc_subchannel_key_destroy(exec_ctx, (grpc_subchannel_key *)p); } static void *sck_avl_copy(void *p, void *unused) { - return subchannel_key_copy(p); + return subchannel_key_copy((grpc_subchannel_key *)p); } static long sck_avl_compare(void *a, void *b, void *unused) { - return grpc_subchannel_key_compare(a, b); + return grpc_subchannel_key_compare((grpc_subchannel_key *)a, + (grpc_subchannel_key *)b); } static void scv_avl_destroy(void *p, void *user_data) { grpc_exec_ctx *exec_ctx = (grpc_exec_ctx *)user_data; - GRPC_SUBCHANNEL_WEAK_UNREF(exec_ctx, p, "subchannel_index"); + GRPC_SUBCHANNEL_WEAK_UNREF(exec_ctx, (grpc_subchannel *)p, + "subchannel_index"); } static void *scv_avl_copy(void *p, void *unused) { - GRPC_SUBCHANNEL_WEAK_REF(p, "subchannel_index"); + GRPC_SUBCHANNEL_WEAK_REF((grpc_subchannel *)p, "subchannel_index"); return p; } @@ -119,15 +123,27 @@ static const gpr_avl_vtable subchannel_avl_vtable = { void grpc_subchannel_index_init(void) { g_subchannel_index = gpr_avl_create(&subchannel_avl_vtable); gpr_mu_init(&g_mu); + gpr_ref_init(&g_refcount, 1); } void grpc_subchannel_index_shutdown(void) { - grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; - gpr_mu_destroy(&g_mu); - gpr_avl_unref(g_subchannel_index, &exec_ctx); - grpc_exec_ctx_finish(&exec_ctx); + // TODO(juanlishen): This refcounting mechanism may lead to memory leackage. + // To solve that, we should force polling to flush any pending callbacks, then + // shutdown safely. + grpc_subchannel_index_unref(); +} + +void grpc_subchannel_index_unref(void) { + if (gpr_unref(&g_refcount)) { + grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; + gpr_mu_destroy(&g_mu); + gpr_avl_unref(g_subchannel_index, &exec_ctx); + grpc_exec_ctx_finish(&exec_ctx); + } } +void grpc_subchannel_index_ref(void) { gpr_ref_non_zero(&g_refcount); } + grpc_subchannel *grpc_subchannel_index_find(grpc_exec_ctx *exec_ctx, grpc_subchannel_key *key) { // Lock, and take a reference to the subchannel index. @@ -136,8 +152,8 @@ grpc_subchannel *grpc_subchannel_index_find(grpc_exec_ctx *exec_ctx, gpr_avl index = gpr_avl_ref(g_subchannel_index, exec_ctx); gpr_mu_unlock(&g_mu); - grpc_subchannel *c = (grpc_subchannel *)GRPC_SUBCHANNEL_REF_FROM_WEAK_REF( - gpr_avl_get(index, key, exec_ctx), "index_find"); + grpc_subchannel *c = GRPC_SUBCHANNEL_REF_FROM_WEAK_REF( + (grpc_subchannel *)gpr_avl_get(index, key, exec_ctx), "index_find"); gpr_avl_unref(index, exec_ctx); return c; diff --git a/src/core/ext/filters/client_channel/subchannel_index.h b/src/core/ext/filters/client_channel/subchannel_index.h index 98d882a453..92e36d5283 100644 --- a/src/core/ext/filters/client_channel/subchannel_index.h +++ b/src/core/ext/filters/client_channel/subchannel_index.h @@ -59,6 +59,13 @@ void grpc_subchannel_index_init(void); /** Shutdown the subchannel index (global) */ void grpc_subchannel_index_shutdown(void); +/** Increment the refcount (non-zero) of subchannel index (global). */ +void grpc_subchannel_index_ref(void); + +/** Decrement the refcount of subchannel index (global). If the refcount drops + to zero, unref the subchannel index and destroy its mutex. */ +void grpc_subchannel_index_unref(void); + /** \em TEST ONLY. * If \a force_creation is true, all key comparisons will be false, resulting in * new subchannels always being created. Otherwise, the keys will be compared as |