Chiebot-Mirror
/
grpc
mirror of https://github.com/grpc/grpc.git
8
0
Fork 0
Code Issues Projects Releases Wiki Activity

Initial pass to remove mu_lock from client_channel: trickier cases remain

Browse Source
reviewable/pr9660/r1
Craig Tiller 8 years ago
parent 7bd2a627f6
commit befafe64f9
1 changed files with 150 additions and 141 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 291
      src/core/ext/client_channel/client_channel.c

291
src/core/ext/client_channel/client_channel.c
Unescape View File

@ -51,6 +51,7 @@
#include "src/core/lib/channel/channel_args.h" #include "src/core/lib/channel/channel_args.h"
#include "src/core/lib/channel/connected_channel.h" #include "src/core/lib/channel/connected_channel.h"
#include "src/core/lib/channel/deadline_filter.h" #include "src/core/lib/channel/deadline_filter.h"
#include "src/core/lib/iomgr/combiner.h"
#include "src/core/lib/iomgr/iomgr.h" #include "src/core/lib/iomgr/iomgr.h"
#include "src/core/lib/iomgr/polling_entity.h" #include "src/core/lib/iomgr/polling_entity.h"
#include "src/core/lib/profiling/timers.h" #include "src/core/lib/profiling/timers.h"
@ -160,8 +161,8 @@ typedef struct client_channel_channel_data {
/** client channel factory */ /** client channel factory */
grpc_client_channel_factory *client_channel_factory; grpc_client_channel_factory *client_channel_factory;
/** mutex protecting all variables below in this data structure */ /** combiner protecting all variables below in this data structure */
gpr_mu mu; grpc_combiner *combiner;
/** currently active load balancer */ /** currently active load balancer */
char *lb_policy_name; char *lb_policy_name;
grpc_lb_policy *lb_policy; grpc_lb_policy *lb_policy;
@ -218,8 +219,8 @@ static void set_channel_connectivity_state_locked(grpc_exec_ctx *exec_ctx,
} }
static void on_lb_policy_state_changed_locked(grpc_exec_ctx *exec_ctx, static void on_lb_policy_state_changed_locked(grpc_exec_ctx *exec_ctx,
lb_policy_connectivity_watcher *w, void *arg, grpc_error *error) {
grpc_error *error) { lb_policy_connectivity_watcher *w = arg;
grpc_connectivity_state publish_state = w->state; grpc_connectivity_state publish_state = w->state;
/* check if the notification is for a stale policy */ /* check if the notification is for a stale policy */
if (w->lb_policy != w->chand->lb_policy) return; if (w->lb_policy != w->chand->lb_policy) return;
@ -235,15 +236,6 @@ static void on_lb_policy_state_changed_locked(grpc_exec_ctx *exec_ctx,
if (w->state != GRPC_CHANNEL_SHUTDOWN) { if (w->state != GRPC_CHANNEL_SHUTDOWN) {
watch_lb_policy(exec_ctx, w->chand, w->lb_policy, w->state); watch_lb_policy(exec_ctx, w->chand, w->lb_policy, w->state);
} }
}
static void on_lb_policy_state_changed(grpc_exec_ctx *exec_ctx, void *arg,
grpc_error *error) {
lb_policy_connectivity_watcher *w = arg;
gpr_mu_lock(&w->chand->mu);
on_lb_policy_state_changed_locked(exec_ctx, w, error);
gpr_mu_unlock(&w->chand->mu);
GRPC_CHANNEL_STACK_UNREF(exec_ctx, w->chand->owning_stack, "watch_lb_policy"); GRPC_CHANNEL_STACK_UNREF(exec_ctx, w->chand->owning_stack, "watch_lb_policy");
gpr_free(w); gpr_free(w);
@ -256,16 +248,16 @@ static void watch_lb_policy(grpc_exec_ctx *exec_ctx, channel_data *chand,
GRPC_CHANNEL_STACK_REF(chand->owning_stack, "watch_lb_policy"); GRPC_CHANNEL_STACK_REF(chand->owning_stack, "watch_lb_policy");
w->chand = chand; w->chand = chand;
grpc_closure_init(&w->on_changed, on_lb_policy_state_changed, w, grpc_closure_init(&w->on_changed, on_lb_policy_state_changed_locked, w,
grpc_schedule_on_exec_ctx); grpc_combiner_scheduler(chand->combiner, false));
w->state = current_state; w->state = current_state;
w->lb_policy = lb_policy; w->lb_policy = lb_policy;
grpc_lb_policy_notify_on_state_change(exec_ctx, lb_policy, &w->state, grpc_lb_policy_notify_on_state_change(exec_ctx, lb_policy, &w->state,
&w->on_changed); &w->on_changed);
} }
static void on_resolver_result_changed(grpc_exec_ctx *exec_ctx, void *arg, static void on_resolver_result_changed_locked(grpc_exec_ctx *exec_ctx,
grpc_error *error) { void *arg, grpc_error *error) {
channel_data *chand = arg; channel_data *chand = arg;
char *lb_policy_name = NULL; char *lb_policy_name = NULL;
grpc_lb_policy *lb_policy = NULL; grpc_lb_policy *lb_policy = NULL;
@ -353,7 +345,6 @@ static void on_resolver_result_changed(grpc_exec_ctx *exec_ctx, void *arg,
chand->interested_parties); chand->interested_parties);
} }
gpr_mu_lock(&chand->mu);
if (lb_policy_name != NULL) { if (lb_policy_name != NULL) {
gpr_free(chand->lb_policy_name); gpr_free(chand->lb_policy_name);
chand->lb_policy_name = lb_policy_name; chand->lb_policy_name = lb_policy_name;
@ -391,7 +382,6 @@ static void on_resolver_result_changed(grpc_exec_ctx *exec_ctx, void *arg,
GRPC_CHANNEL_STACK_REF(chand->owning_stack, "resolver"); GRPC_CHANNEL_STACK_REF(chand->owning_stack, "resolver");
grpc_resolver_next(exec_ctx, chand->resolver, &chand->resolver_result, grpc_resolver_next(exec_ctx, chand->resolver, &chand->resolver_result,
&chand->on_resolver_result_changed); &chand->on_resolver_result_changed);
gpr_mu_unlock(&chand->mu);
} else { } else {
if (chand->resolver != NULL) { if (chand->resolver != NULL) {
grpc_resolver_shutdown(exec_ctx, chand->resolver); grpc_resolver_shutdown(exec_ctx, chand->resolver);
@ -404,7 +394,6 @@ static void on_resolver_result_changed(grpc_exec_ctx *exec_ctx, void *arg,
GRPC_ERROR_CREATE_REFERENCING("Got config after disconnection", refs, GRPC_ERROR_CREATE_REFERENCING("Got config after disconnection", refs,
GPR_ARRAY_SIZE(refs)), GPR_ARRAY_SIZE(refs)),
"resolver_gone"); "resolver_gone");
gpr_mu_unlock(&chand->mu);
} }
if (exit_idle) { if (exit_idle) {
@ -426,20 +415,12 @@ static void on_resolver_result_changed(grpc_exec_ctx *exec_ctx, void *arg,
GRPC_ERROR_UNREF(state_error); GRPC_ERROR_UNREF(state_error);
} }
static void cc_start_transport_op(grpc_exec_ctx *exec_ctx, static void cc_start_transport_op_locked(grpc_exec_ctx *exec_ctx, void *arg,
grpc_channel_element *elem, grpc_error *error_ignored) {
grpc_transport_op *op) { grpc_transport_op *op = arg;
grpc_channel_element *elem = op->transport_private.args[0];
channel_data *chand = elem->channel_data; channel_data *chand = elem->channel_data;
grpc_closure_sched(exec_ctx, op->on_consumed, GRPC_ERROR_NONE);
GPR_ASSERT(op->set_accept_stream == false);
if (op->bind_pollset != NULL) {
grpc_pollset_set_add_pollset(exec_ctx, chand->interested_parties,
op->bind_pollset);
}
gpr_mu_lock(&chand->mu);
if (op->on_connectivity_state_change != NULL) { if (op->on_connectivity_state_change != NULL) {
grpc_connectivity_state_notify_on_state_change( grpc_connectivity_state_notify_on_state_change(
exec_ctx, &chand->state_tracker, op->connectivity_state, exec_ctx, &chand->state_tracker, op->connectivity_state,
@ -482,7 +463,28 @@ static void cc_start_transport_op(grpc_exec_ctx *exec_ctx,
} }
GRPC_ERROR_UNREF(op->disconnect_with_error); GRPC_ERROR_UNREF(op->disconnect_with_error);
} }
gpr_mu_unlock(&chand->mu); }
static void cc_start_transport_op(grpc_exec_ctx *exec_ctx,
grpc_channel_element *elem,
grpc_transport_op *op) {
channel_data *chand = elem->channel_data;
grpc_closure_sched(exec_ctx, op->on_consumed, GRPC_ERROR_NONE);
GPR_ASSERT(op->set_accept_stream == false);
if (op->bind_pollset != NULL) {
grpc_pollset_set_add_pollset(exec_ctx, chand->interested_parties,
op->bind_pollset);
}
op->transport_private.args[0] = elem;
grpc_closure_sched(
exec_ctx,
grpc_closure_init(&op->transport_private.closure,
cc_start_transport_op_locked, op,
grpc_combiner_scheduler(chand->combiner, false)),
GRPC_ERROR_NONE);
} }
static void cc_get_channel_info(grpc_exec_ctx *exec_ctx, static void cc_get_channel_info(grpc_exec_ctx *exec_ctx,
@ -512,11 +514,11 @@ static grpc_error *cc_init_channel_elem(grpc_exec_ctx *exec_ctx,
GPR_ASSERT(args->is_last); GPR_ASSERT(args->is_last);
GPR_ASSERT(elem->filter == &grpc_client_channel_filter); GPR_ASSERT(elem->filter == &grpc_client_channel_filter);
// Initialize data members. // Initialize data members.
gpr_mu_init(&chand->mu); chand->combiner = grpc_combiner_create(NULL);
chand->owning_stack = args->channel_stack; chand->owning_stack = args->channel_stack;
grpc_closure_init(&chand->on_resolver_result_changed, grpc_closure_init(&chand->on_resolver_result_changed,
on_resolver_result_changed, chand, on_resolver_result_changed_locked, chand,
grpc_schedule_on_exec_ctx); grpc_combiner_scheduler(chand->combiner, false));
chand->interested_parties = grpc_pollset_set_create(); chand->interested_parties = grpc_pollset_set_create();
grpc_connectivity_state_init(&chand->state_tracker, GRPC_CHANNEL_IDLE, grpc_connectivity_state_init(&chand->state_tracker, GRPC_CHANNEL_IDLE,
"client_channel"); "client_channel");
@ -572,7 +574,7 @@ static void cc_destroy_channel_elem(grpc_exec_ctx *exec_ctx,
} }
grpc_connectivity_state_destroy(exec_ctx, &chand->state_tracker); grpc_connectivity_state_destroy(exec_ctx, &chand->state_tracker);
grpc_pollset_set_destroy(chand->interested_parties); grpc_pollset_set_destroy(chand->interested_parties);
gpr_mu_destroy(&chand->mu); grpc_combiner_destroy(exec_ctx, chand->combiner);
} }
/************************************************************************* /*************************************************************************
@ -615,8 +617,6 @@ typedef struct client_channel_call_data {
grpc_subchannel_call */ grpc_subchannel_call */
gpr_atm subchannel_call; gpr_atm subchannel_call;
gpr_mu mu;
subchannel_creation_phase creation_phase; subchannel_creation_phase creation_phase;
grpc_connected_subchannel *connected_subchannel; grpc_connected_subchannel *connected_subchannel;
grpc_polling_entity *pollent; grpc_polling_entity *pollent;
@ -701,12 +701,11 @@ static void retry_waiting_locked(grpc_exec_ctx *exec_ctx, call_data *calld) {
GRPC_ERROR_NONE); GRPC_ERROR_NONE);
} }
static void subchannel_ready(grpc_exec_ctx *exec_ctx, void *arg, static void subchannel_ready_locked(grpc_exec_ctx *exec_ctx, void *arg,
grpc_error *error) { grpc_error *error) {
grpc_call_element *elem = arg; grpc_call_element *elem = arg;
call_data *calld = elem->call_data; call_data *calld = elem->call_data;
channel_data *chand = elem->channel_data; channel_data *chand = elem->channel_data;
gpr_mu_lock(&calld->mu);
GPR_ASSERT(calld->creation_phase == GPR_ASSERT(calld->creation_phase ==
GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL); GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL);
grpc_polling_entity_del_from_pollset_set(exec_ctx, calld->pollent, grpc_polling_entity_del_from_pollset_set(exec_ctx, calld->pollent,
@ -742,7 +741,6 @@ static void subchannel_ready(grpc_exec_ctx *exec_ctx, void *arg,
(gpr_atm)(uintptr_t)subchannel_call); (gpr_atm)(uintptr_t)subchannel_call);
retry_waiting_locked(exec_ctx, calld); retry_waiting_locked(exec_ctx, calld);
} }
gpr_mu_unlock(&calld->mu);
GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call, "pick_subchannel"); GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call, "pick_subchannel");
} }
@ -768,37 +766,35 @@ typedef struct {
/** Return true if subchannel is available immediately (in which case on_ready /** Return true if subchannel is available immediately (in which case on_ready
should not be called), or false otherwise (in which case on_ready should be should not be called), or false otherwise (in which case on_ready should be
called when the subchannel is available). */ called when the subchannel is available). */
static bool pick_subchannel(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, static bool pick_subchannel_locked(
grpc_metadata_batch *initial_metadata, grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
uint32_t initial_metadata_flags, grpc_metadata_batch *initial_metadata, uint32_t initial_metadata_flags,
grpc_connected_subchannel **connected_subchannel, grpc_connected_subchannel **connected_subchannel, grpc_closure *on_ready,
grpc_closure *on_ready, grpc_error *error); grpc_error *error);
static void continue_picking(grpc_exec_ctx *exec_ctx, void *arg, static void continue_picking_locked(grpc_exec_ctx *exec_ctx, void *arg,
grpc_error *error) { grpc_error *error) {
continue_picking_args *cpa = arg; continue_picking_args *cpa = arg;
if (cpa->connected_subchannel == NULL) { if (cpa->connected_subchannel == NULL) {
/* cancelled, do nothing */ /* cancelled, do nothing */
} else if (error != GRPC_ERROR_NONE) { } else if (error != GRPC_ERROR_NONE) {
grpc_closure_sched(exec_ctx, cpa->on_ready, GRPC_ERROR_REF(error)); grpc_closure_sched(exec_ctx, cpa->on_ready, GRPC_ERROR_REF(error));
} else { } else {
call_data *calld = cpa->elem->call_data; if (pick_subchannel_locked(exec_ctx, cpa->elem, cpa->initial_metadata,
gpr_mu_lock(&calld->mu); cpa->initial_metadata_flags,
if (pick_subchannel(exec_ctx, cpa->elem, cpa->initial_metadata, cpa->connected_subchannel, cpa->on_ready,
cpa->initial_metadata_flags, cpa->connected_subchannel, GRPC_ERROR_NONE)) {
cpa->on_ready, GRPC_ERROR_NONE)) {
grpc_closure_sched(exec_ctx, cpa->on_ready, GRPC_ERROR_NONE); grpc_closure_sched(exec_ctx, cpa->on_ready, GRPC_ERROR_NONE);
} }
gpr_mu_unlock(&calld->mu);
} }
gpr_free(cpa); gpr_free(cpa);
} }
static bool pick_subchannel(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, static bool pick_subchannel_locked(
grpc_metadata_batch *initial_metadata, grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
uint32_t initial_metadata_flags, grpc_metadata_batch *initial_metadata, uint32_t initial_metadata_flags,
grpc_connected_subchannel **connected_subchannel, grpc_connected_subchannel **connected_subchannel, grpc_closure *on_ready,
grpc_closure *on_ready, grpc_error *error) { grpc_error *error) {
GPR_TIMER_BEGIN("pick_subchannel", 0); GPR_TIMER_BEGIN("pick_subchannel", 0);
channel_data *chand = elem->channel_data; channel_data *chand = elem->channel_data;
@ -808,7 +804,6 @@ static bool pick_subchannel(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
GPR_ASSERT(connected_subchannel); GPR_ASSERT(connected_subchannel);
gpr_mu_lock(&chand->mu);
if (initial_metadata == NULL) { if (initial_metadata == NULL) {
if (chand->lb_policy != NULL) { if (chand->lb_policy != NULL) {
grpc_lb_policy_cancel_pick(exec_ctx, chand->lb_policy, grpc_lb_policy_cancel_pick(exec_ctx, chand->lb_policy,
@ -824,7 +819,6 @@ static bool pick_subchannel(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
GRPC_ERROR_CREATE_REFERENCING("Pick cancelled", &error, 1)); GRPC_ERROR_CREATE_REFERENCING("Pick cancelled", &error, 1));
} }
} }
gpr_mu_unlock(&chand->mu);
GPR_TIMER_END("pick_subchannel", 0); GPR_TIMER_END("pick_subchannel", 0);
GRPC_ERROR_UNREF(error); GRPC_ERROR_UNREF(error);
return true; return true;
@ -833,7 +827,6 @@ static bool pick_subchannel(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
if (chand->lb_policy != NULL) { if (chand->lb_policy != NULL) {
grpc_lb_policy *lb_policy = chand->lb_policy; grpc_lb_policy *lb_policy = chand->lb_policy;
GRPC_LB_POLICY_REF(lb_policy, "pick_subchannel"); GRPC_LB_POLICY_REF(lb_policy, "pick_subchannel");
gpr_mu_unlock(&chand->mu);
// If the application explicitly set wait_for_ready, use that. // If the application explicitly set wait_for_ready, use that.
// Otherwise, if the service config specified a value for this // Otherwise, if the service config specified a value for this
// method, use that. // method, use that.
@ -872,59 +865,37 @@ static bool pick_subchannel(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
cpa->connected_subchannel = connected_subchannel; cpa->connected_subchannel = connected_subchannel;
cpa->on_ready = on_ready; cpa->on_ready = on_ready;
cpa->elem = elem; cpa->elem = elem;
grpc_closure_init(&cpa->closure, continue_picking, cpa, grpc_closure_init(&cpa->closure, continue_picking_locked, cpa,
grpc_schedule_on_exec_ctx); grpc_combiner_scheduler(chand->combiner, true));
grpc_closure_list_append(&chand->waiting_for_config_closures, &cpa->closure, grpc_closure_list_append(&chand->waiting_for_config_closures, &cpa->closure,
GRPC_ERROR_NONE); GRPC_ERROR_NONE);
} else { } else {
grpc_closure_sched(exec_ctx, on_ready, GRPC_ERROR_CREATE("Disconnected")); grpc_closure_sched(exec_ctx, on_ready, GRPC_ERROR_CREATE("Disconnected"));
} }
gpr_mu_unlock(&chand->mu);
GPR_TIMER_END("pick_subchannel", 0); GPR_TIMER_END("pick_subchannel", 0);
return false; return false;
} }
// The logic here is fairly complicated, due to (a) the fact that we static void cc_start_transport_stream_op_locked(grpc_exec_ctx *exec_ctx,
// need to handle the case where we receive the send op before the void *arg,
// initial metadata op, and (b) the need for efficiency, especially in grpc_error *error_ignored) {
// the streaming case. grpc_transport_stream_op *op = arg;
// TODO(ctiller): Explain this more thoroughly. grpc_call_element *elem = op->transport_private.args[0];
static void cc_start_transport_stream_op(grpc_exec_ctx *exec_ctx,
grpc_call_element *elem,
grpc_transport_stream_op *op) {
call_data *calld = elem->call_data; call_data *calld = elem->call_data;
channel_data *chand = elem->channel_data; channel_data *chand = elem->channel_data;
GRPC_CALL_LOG_OP(GPR_INFO, elem, op); grpc_subchannel_call *call;
grpc_deadline_state_client_start_transport_stream_op(exec_ctx, elem, op);
/* try to (atomically) get the call */
grpc_subchannel_call *call = GET_CALL(calld);
GPR_TIMER_BEGIN("cc_start_transport_stream_op", 0);
if (call == CANCELLED_CALL) {
grpc_transport_stream_op_finish_with_failure(
exec_ctx, op, GRPC_ERROR_REF(calld->cancel_error));
GPR_TIMER_END("cc_start_transport_stream_op", 0);
return;
}
if (call != NULL) {
grpc_subchannel_call_process_op(exec_ctx, call, op);
GPR_TIMER_END("cc_start_transport_stream_op", 0);
return;
}
/* we failed; lock and figure out what to do */
gpr_mu_lock(&calld->mu);
retry: retry:
/* need to recheck that another thread hasn't set the call */ /* need to recheck that another thread hasn't set the call */
call = GET_CALL(calld); call = GET_CALL(calld);
if (call == CANCELLED_CALL) { if (call == CANCELLED_CALL) {
gpr_mu_unlock(&calld->mu);
grpc_transport_stream_op_finish_with_failure( grpc_transport_stream_op_finish_with_failure(
exec_ctx, op, GRPC_ERROR_REF(calld->cancel_error)); exec_ctx, op, GRPC_ERROR_REF(calld->cancel_error));
GPR_TIMER_END("cc_start_transport_stream_op", 0); GPR_TIMER_END("cc_start_transport_stream_op", 0);
return; return;
} }
if (call != NULL) { if (call != NULL) {
gpr_mu_unlock(&calld->mu);
grpc_subchannel_call_process_op(exec_ctx, call, op); grpc_subchannel_call_process_op(exec_ctx, call, op);
GPR_TIMER_END("cc_start_transport_stream_op", 0); GPR_TIMER_END("cc_start_transport_stream_op", 0);
return; return;
@ -946,11 +917,11 @@ retry:
fail_locked(exec_ctx, calld, GRPC_ERROR_REF(op->cancel_error)); fail_locked(exec_ctx, calld, GRPC_ERROR_REF(op->cancel_error));
break; break;
case GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL: case GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL:
pick_subchannel(exec_ctx, elem, NULL, 0, &calld->connected_subchannel, pick_subchannel_locked(exec_ctx, elem, NULL, 0,
NULL, GRPC_ERROR_REF(op->cancel_error)); &calld->connected_subchannel, NULL,
GRPC_ERROR_REF(op->cancel_error));
break; break;
} }
gpr_mu_unlock(&calld->mu);
grpc_transport_stream_op_finish_with_failure( grpc_transport_stream_op_finish_with_failure(
exec_ctx, op, GRPC_ERROR_REF(op->cancel_error)); exec_ctx, op, GRPC_ERROR_REF(op->cancel_error));
GPR_TIMER_END("cc_start_transport_stream_op", 0); GPR_TIMER_END("cc_start_transport_stream_op", 0);
@ -962,16 +933,16 @@ retry:
calld->connected_subchannel == NULL && calld->connected_subchannel == NULL &&
op->send_initial_metadata != NULL) { op->send_initial_metadata != NULL) {
calld->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL; calld->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL;
grpc_closure_init(&calld->next_step, subchannel_ready, elem, grpc_closure_init(&calld->next_step, subchannel_ready_locked, elem,
grpc_schedule_on_exec_ctx); grpc_combiner_scheduler(chand->combiner, true));
GRPC_CALL_STACK_REF(calld->owning_call, "pick_subchannel"); GRPC_CALL_STACK_REF(calld->owning_call, "pick_subchannel");
/* If a subchannel is not available immediately, the polling entity from /* If a subchannel is not available immediately, the polling entity from
call_data should be provided to channel_data's interested_parties, so call_data should be provided to channel_data's interested_parties, so
that IO of the lb_policy and resolver could be done under it. */ that IO of the lb_policy and resolver could be done under it. */
if (pick_subchannel(exec_ctx, elem, op->send_initial_metadata, if (pick_subchannel_locked(exec_ctx, elem, op->send_initial_metadata,
op->send_initial_metadata_flags, op->send_initial_metadata_flags,
&calld->connected_subchannel, &calld->next_step, &calld->connected_subchannel, &calld->next_step,
GRPC_ERROR_NONE)) { GRPC_ERROR_NONE)) {
calld->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING; calld->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING;
GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call, "pick_subchannel"); GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call, "pick_subchannel");
} else { } else {
@ -998,27 +969,60 @@ retry:
} }
/* nothing to be done but wait */ /* nothing to be done but wait */
add_waiting_locked(calld, op); add_waiting_locked(calld, op);
gpr_mu_unlock(&calld->mu);
GPR_TIMER_END("cc_start_transport_stream_op", 0); GPR_TIMER_END("cc_start_transport_stream_op", 0);
} }
// The logic here is fairly complicated, due to (a) the fact that we
// need to handle the case where we receive the send op before the
// initial metadata op, and (b) the need for efficiency, especially in
// the streaming case.
// TODO(ctiller): Explain this more thoroughly.
static void cc_start_transport_stream_op(grpc_exec_ctx *exec_ctx,
grpc_call_element *elem,
grpc_transport_stream_op *op) {
call_data *calld = elem->call_data;
channel_data *chand = elem->channel_data;
GRPC_CALL_LOG_OP(GPR_INFO, elem, op);
grpc_deadline_state_client_start_transport_stream_op(exec_ctx, elem, op);
/* try to (atomically) get the call */
grpc_subchannel_call *call = GET_CALL(calld);
GPR_TIMER_BEGIN("cc_start_transport_stream_op", 0);
if (call == CANCELLED_CALL) {
grpc_transport_stream_op_finish_with_failure(
exec_ctx, op, GRPC_ERROR_REF(calld->cancel_error));
GPR_TIMER_END("cc_start_transport_stream_op", 0);
return;
}
if (call != NULL) {
grpc_subchannel_call_process_op(exec_ctx, call, op);
GPR_TIMER_END("cc_start_transport_stream_op", 0);
return;
}
/* we failed; lock and figure out what to do */
op->transport_private.args[0] = elem;
grpc_closure_sched(
exec_ctx,
grpc_closure_init(&op->transport_private.closure,
cc_start_transport_stream_op_locked, op,
grpc_combiner_scheduler(chand->combiner, false)),
GRPC_ERROR_NONE);
}
// Gets data from the service config. Invoked when the resolver returns // Gets data from the service config. Invoked when the resolver returns
// its initial result. // its initial result.
static void read_service_config(grpc_exec_ctx *exec_ctx, void *arg, static void read_service_config_locked(grpc_exec_ctx *exec_ctx, void *arg,
grpc_error *error) { grpc_error *error) {
grpc_call_element *elem = arg; grpc_call_element *elem = arg;
channel_data *chand = elem->channel_data; channel_data *chand = elem->channel_data;
call_data *calld = elem->call_data; call_data *calld = elem->call_data;
// If this is an error, there's no point in looking at the service config. // If this is an error, there's no point in looking at the service config.
if (error == GRPC_ERROR_NONE) { if (error == GRPC_ERROR_NONE) {
// Get the method config table from channel data. // Get the method config table from channel data.
gpr_mu_lock(&chand->mu);
grpc_slice_hash_table *method_params_table = NULL; grpc_slice_hash_table *method_params_table = NULL;
if (chand->method_params_table != NULL) { if (chand->method_params_table != NULL) {
method_params_table = method_params_table =
grpc_slice_hash_table_ref(chand->method_params_table); grpc_slice_hash_table_ref(chand->method_params_table);
} }
gpr_mu_unlock(&chand->mu);
// If the method config table was present, use it. // If the method config table was present, use it.
if (method_params_table != NULL) { if (method_params_table != NULL) {
const method_parameters *method_params = grpc_method_config_table_get( const method_parameters *method_params = grpc_method_config_table_get(
@ -1028,7 +1032,6 @@ static void read_service_config(grpc_exec_ctx *exec_ctx, void *arg,
gpr_time_cmp(method_params->timeout, gpr_time_0(GPR_TIMESPAN)) != 0; gpr_time_cmp(method_params->timeout, gpr_time_0(GPR_TIMESPAN)) != 0;
if (have_method_timeout || if (have_method_timeout ||
method_params->wait_for_ready != WAIT_FOR_READY_UNSET) { method_params->wait_for_ready != WAIT_FOR_READY_UNSET) {
gpr_mu_lock(&calld->mu);
if (have_method_timeout) { if (have_method_timeout) {
const gpr_timespec per_method_deadline = const gpr_timespec per_method_deadline =
gpr_time_add(calld->call_start_time, method_params->timeout); gpr_time_add(calld->call_start_time, method_params->timeout);
@ -1042,7 +1045,6 @@ static void read_service_config(grpc_exec_ctx *exec_ctx, void *arg,
calld->wait_for_ready_from_service_config = calld->wait_for_ready_from_service_config =
method_params->wait_for_ready; method_params->wait_for_ready;
} }
gpr_mu_unlock(&calld->mu);
} }
} }
grpc_slice_hash_table_unref(exec_ctx, method_params_table); grpc_slice_hash_table_unref(exec_ctx, method_params_table);
@ -1051,43 +1053,25 @@ static void read_service_config(grpc_exec_ctx *exec_ctx, void *arg,
GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call, "read_service_config"); GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call, "read_service_config");
} }
/* Constructor for call_data */ static void initial_read_service_config_locked(grpc_exec_ctx *exec_ctx,
static grpc_error *cc_init_call_elem(grpc_exec_ctx *exec_ctx, void *arg,
grpc_call_element *elem, grpc_error *error_ignored) {
grpc_call_element_args *args) { grpc_call_element *elem = arg;
channel_data *chand = elem->channel_data; channel_data *chand = elem->channel_data;
call_data *calld = elem->call_data; call_data *calld = elem->call_data;
// Initialize data members.
grpc_deadline_state_init(exec_ctx, elem, args->call_stack);
calld->path = grpc_slice_ref_internal(args->path);
calld->call_start_time = args->start_time;
calld->deadline = gpr_convert_clock_type(args->deadline, GPR_CLOCK_MONOTONIC);
calld->wait_for_ready_from_service_config = WAIT_FOR_READY_UNSET;
calld->cancel_error = GRPC_ERROR_NONE;
gpr_atm_rel_store(&calld->subchannel_call, 0);
gpr_mu_init(&calld->mu);
calld->connected_subchannel = NULL;
calld->waiting_ops = NULL;
calld->waiting_ops_count = 0;
calld->waiting_ops_capacity = 0;
calld->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING;
calld->owning_call = args->call_stack;
calld->pollent = NULL;
// If the resolver has already returned results, then we can access // If the resolver has already returned results, then we can access
// the service config parameters immediately. Otherwise, we need to // the service config parameters immediately. Otherwise, we need to
// defer that work until the resolver returns an initial result. // defer that work until the resolver returns an initial result.
// TODO(roth): This code is almost but not quite identical to the code // TODO(roth): This code is almost but not quite identical to the code
// in read_service_config() above. It would be nice to find a way to // in read_service_config() above. It would be nice to find a way to
// combine them, to avoid having to maintain it twice. // combine them, to avoid having to maintain it twice.
gpr_mu_lock(&chand->mu);
if (chand->lb_policy != NULL) { if (chand->lb_policy != NULL) {
// We already have a resolver result, so check for service config. // We already have a resolver result, so check for service config.
if (chand->method_params_table != NULL) { if (chand->method_params_table != NULL) {
grpc_slice_hash_table *method_params_table = grpc_slice_hash_table *method_params_table =
grpc_slice_hash_table_ref(chand->method_params_table); grpc_slice_hash_table_ref(chand->method_params_table);
gpr_mu_unlock(&chand->mu);
method_parameters *method_params = grpc_method_config_table_get( method_parameters *method_params = grpc_method_config_table_get(
exec_ctx, method_params_table, args->path); exec_ctx, method_params_table, calld->path);
if (method_params != NULL) { if (method_params != NULL) {
if (gpr_time_cmp(method_params->timeout, if (gpr_time_cmp(method_params->timeout,
gpr_time_0(GPR_CLOCK_MONOTONIC)) != 0) { gpr_time_0(GPR_CLOCK_MONOTONIC)) != 0) {
@ -1101,24 +1085,50 @@ static grpc_error *cc_init_call_elem(grpc_exec_ctx *exec_ctx,
} }
} }
grpc_slice_hash_table_unref(exec_ctx, method_params_table); grpc_slice_hash_table_unref(exec_ctx, method_params_table);
} else {
gpr_mu_unlock(&chand->mu);
} }
} else { } else {
// We don't yet have a resolver result, so register a callback to // We don't yet have a resolver result, so register a callback to
// get the service config data once the resolver returns. // get the service config data once the resolver returns.
// Take a reference to the call stack to be owned by the callback. // Take a reference to the call stack to be owned by the callback.
GRPC_CALL_STACK_REF(calld->owning_call, "read_service_config"); GRPC_CALL_STACK_REF(calld->owning_call, "read_service_config");
grpc_closure_init(&calld->read_service_config, read_service_config, elem, grpc_closure_init(&calld->read_service_config, read_service_config_locked,
grpc_schedule_on_exec_ctx); elem, grpc_combiner_scheduler(chand->combiner, false));
grpc_closure_list_append(&chand->waiting_for_config_closures, grpc_closure_list_append(&chand->waiting_for_config_closures,
&calld->read_service_config, GRPC_ERROR_NONE); &calld->read_service_config, GRPC_ERROR_NONE);
gpr_mu_unlock(&chand->mu);
} }
// Start the deadline timer with the current deadline value. If we // Start the deadline timer with the current deadline value. If we
// do not yet have service config data, then the timer may be reset // do not yet have service config data, then the timer may be reset
// later. // later.
grpc_deadline_state_start(exec_ctx, elem, calld->deadline); grpc_deadline_state_start(exec_ctx, elem, calld->deadline);
}
/* Constructor for call_data */
static grpc_error *cc_init_call_elem(grpc_exec_ctx *exec_ctx,
grpc_call_element *elem,
grpc_call_element_args *args) {
channel_data *chand = elem->channel_data;
call_data *calld = elem->call_data;
// Initialize data members.
grpc_deadline_state_init(exec_ctx, elem, args->call_stack);
calld->path = grpc_slice_ref_internal(args->path);
calld->call_start_time = args->start_time;
calld->deadline = gpr_convert_clock_type(args->deadline, GPR_CLOCK_MONOTONIC);
calld->wait_for_ready_from_service_config = WAIT_FOR_READY_UNSET;
calld->cancel_error = GRPC_ERROR_NONE;
gpr_atm_rel_store(&calld->subchannel_call, 0);
calld->connected_subchannel = NULL;
calld->waiting_ops = NULL;
calld->waiting_ops_count = 0;
calld->waiting_ops_capacity = 0;
calld->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING;
calld->owning_call = args->call_stack;
calld->pollent = NULL;
grpc_closure_sched(
exec_ctx,
grpc_closure_init(&calld->read_service_config,
initial_read_service_config_locked, elem,
grpc_combiner_scheduler(chand->combiner, false)),
GRPC_ERROR_NONE);
return GRPC_ERROR_NONE; return GRPC_ERROR_NONE;
} }
@ -1136,7 +1146,6 @@ static void cc_destroy_call_elem(grpc_exec_ctx *exec_ctx,
GRPC_SUBCHANNEL_CALL_UNREF(exec_ctx, call, "client_channel_destroy_call"); GRPC_SUBCHANNEL_CALL_UNREF(exec_ctx, call, "client_channel_destroy_call");
} }
GPR_ASSERT(calld->creation_phase == GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING); GPR_ASSERT(calld->creation_phase == GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING);
gpr_mu_destroy(&calld->mu);
GPR_ASSERT(calld->waiting_ops_count == 0); GPR_ASSERT(calld->waiting_ops_count == 0);
if (calld->connected_subchannel != NULL) { if (calld->connected_subchannel != NULL) {
GRPC_CONNECTED_SUBCHANNEL_UNREF(exec_ctx, calld->connected_subchannel, GRPC_CONNECTED_SUBCHANNEL_UNREF(exec_ctx, calld->connected_subchannel,

Write Preview
Loading…
Cancel
Save