|
|
|
|
@ -796,7 +796,8 @@ static void cc_destroy_channel_elem(grpc_exec_ctx *exec_ctx, |
|
|
|
|
// send_message
|
|
|
|
|
// recv_trailing_metadata
|
|
|
|
|
// send_trailing_metadata
|
|
|
|
|
#define MAX_WAITING_BATCHES 6 |
|
|
|
|
// We also add room for a single cancel_stream batch.
|
|
|
|
|
#define MAX_WAITING_BATCHES 7 |
|
|
|
|
|
|
|
|
|
/** Call data. Holds a pointer to grpc_subchannel_call and the
|
|
|
|
|
associated machinery to create such a pointer. |
|
|
|
|
@ -808,23 +809,25 @@ typedef struct client_channel_call_data { |
|
|
|
|
// The code in deadline_filter.c requires this to be the first field.
|
|
|
|
|
// TODO(roth): This is slightly sub-optimal in that grpc_deadline_state
|
|
|
|
|
// and this struct both independently store a pointer to the call
|
|
|
|
|
// stack and each has its own mutex. If/when we have time, find a way
|
|
|
|
|
// to avoid this without breaking the grpc_deadline_state abstraction.
|
|
|
|
|
// combiner. If/when we have time, find a way to avoid this without
|
|
|
|
|
// breaking the grpc_deadline_state abstraction.
|
|
|
|
|
grpc_deadline_state deadline_state; |
|
|
|
|
|
|
|
|
|
grpc_slice path; // Request path.
|
|
|
|
|
gpr_timespec call_start_time; |
|
|
|
|
gpr_timespec deadline; |
|
|
|
|
gpr_arena *arena; |
|
|
|
|
grpc_call_combiner *call_combiner; |
|
|
|
|
|
|
|
|
|
grpc_server_retry_throttle_data *retry_throttle_data; |
|
|
|
|
method_parameters *method_params; |
|
|
|
|
|
|
|
|
|
/** either 0 for no call, a pointer to a grpc_subchannel_call (if the lowest
|
|
|
|
|
bit is 0), or a pointer to an error (if the lowest bit is 1) */ |
|
|
|
|
gpr_atm subchannel_call_or_error; |
|
|
|
|
gpr_arena *arena; |
|
|
|
|
grpc_subchannel_call *subchannel_call; |
|
|
|
|
grpc_error *error; |
|
|
|
|
|
|
|
|
|
grpc_lb_policy *lb_policy; // Holds ref while LB pick is pending.
|
|
|
|
|
grpc_closure lb_pick_closure; |
|
|
|
|
grpc_closure cancel_closure; |
|
|
|
|
|
|
|
|
|
grpc_connected_subchannel *connected_subchannel; |
|
|
|
|
grpc_call_context_element subchannel_call_context[GRPC_CONTEXT_COUNT]; |
|
|
|
|
@ -832,10 +835,9 @@ typedef struct client_channel_call_data { |
|
|
|
|
|
|
|
|
|
grpc_transport_stream_op_batch *waiting_for_pick_batches[MAX_WAITING_BATCHES]; |
|
|
|
|
size_t waiting_for_pick_batches_count; |
|
|
|
|
grpc_closure handle_pending_batch_in_call_combiner[MAX_WAITING_BATCHES]; |
|
|
|
|
|
|
|
|
|
grpc_transport_stream_op_batch_payload *initial_metadata_payload; |
|
|
|
|
|
|
|
|
|
grpc_call_stack *owning_call; |
|
|
|
|
grpc_transport_stream_op_batch *initial_metadata_batch; |
|
|
|
|
|
|
|
|
|
grpc_linked_mdelem lb_token_mdelem; |
|
|
|
|
|
|
|
|
|
@ -843,55 +845,42 @@ typedef struct client_channel_call_data { |
|
|
|
|
grpc_closure *original_on_complete; |
|
|
|
|
} call_data; |
|
|
|
|
|
|
|
|
|
typedef struct { |
|
|
|
|
grpc_subchannel_call *subchannel_call; |
|
|
|
|
grpc_error *error; |
|
|
|
|
} call_or_error; |
|
|
|
|
|
|
|
|
|
static call_or_error get_call_or_error(call_data *p) { |
|
|
|
|
gpr_atm c = gpr_atm_acq_load(&p->subchannel_call_or_error); |
|
|
|
|
if (c == 0) |
|
|
|
|
return (call_or_error){NULL, NULL}; |
|
|
|
|
else if (c & 1) |
|
|
|
|
return (call_or_error){NULL, (grpc_error *)((c) & ~(gpr_atm)1)}; |
|
|
|
|
else |
|
|
|
|
return (call_or_error){(grpc_subchannel_call *)c, NULL}; |
|
|
|
|
grpc_subchannel_call *grpc_client_channel_get_subchannel_call( |
|
|
|
|
grpc_call_element *elem) { |
|
|
|
|
call_data *calld = elem->call_data; |
|
|
|
|
return calld->subchannel_call; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static bool set_call_or_error(call_data *p, call_or_error coe) { |
|
|
|
|
// this should always be under a lock
|
|
|
|
|
call_or_error existing = get_call_or_error(p); |
|
|
|
|
if (existing.error != GRPC_ERROR_NONE) { |
|
|
|
|
GRPC_ERROR_UNREF(coe.error); |
|
|
|
|
return false; |
|
|
|
|
} |
|
|
|
|
GPR_ASSERT(existing.subchannel_call == NULL); |
|
|
|
|
if (coe.error != GRPC_ERROR_NONE) { |
|
|
|
|
GPR_ASSERT(coe.subchannel_call == NULL); |
|
|
|
|
gpr_atm_rel_store(&p->subchannel_call_or_error, 1 | (gpr_atm)coe.error); |
|
|
|
|
// This is called via the call combiner, so access to calld is synchronized.
|
|
|
|
|
static void waiting_for_pick_batches_add( |
|
|
|
|
call_data *calld, grpc_transport_stream_op_batch *batch) { |
|
|
|
|
if (batch->send_initial_metadata) { |
|
|
|
|
GPR_ASSERT(calld->initial_metadata_batch == NULL); |
|
|
|
|
calld->initial_metadata_batch = batch; |
|
|
|
|
} else { |
|
|
|
|
GPR_ASSERT(coe.subchannel_call != NULL); |
|
|
|
|
gpr_atm_rel_store(&p->subchannel_call_or_error, |
|
|
|
|
(gpr_atm)coe.subchannel_call); |
|
|
|
|
GPR_ASSERT(calld->waiting_for_pick_batches_count < MAX_WAITING_BATCHES); |
|
|
|
|
calld->waiting_for_pick_batches[calld->waiting_for_pick_batches_count++] = |
|
|
|
|
batch; |
|
|
|
|
} |
|
|
|
|
return true; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
grpc_subchannel_call *grpc_client_channel_get_subchannel_call( |
|
|
|
|
grpc_call_element *call_elem) { |
|
|
|
|
return get_call_or_error(call_elem->call_data).subchannel_call; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void waiting_for_pick_batches_add_locked( |
|
|
|
|
call_data *calld, grpc_transport_stream_op_batch *batch) { |
|
|
|
|
GPR_ASSERT(calld->waiting_for_pick_batches_count < MAX_WAITING_BATCHES); |
|
|
|
|
calld->waiting_for_pick_batches[calld->waiting_for_pick_batches_count++] = |
|
|
|
|
batch; |
|
|
|
|
// This is called via the call combiner, so access to calld is synchronized.
|
|
|
|
|
static void fail_pending_batch_in_call_combiner(grpc_exec_ctx *exec_ctx, |
|
|
|
|
void *arg, grpc_error *error) { |
|
|
|
|
call_data *calld = arg; |
|
|
|
|
if (calld->waiting_for_pick_batches_count > 0) { |
|
|
|
|
--calld->waiting_for_pick_batches_count; |
|
|
|
|
grpc_transport_stream_op_batch_finish_with_failure( |
|
|
|
|
exec_ctx, |
|
|
|
|
calld->waiting_for_pick_batches[calld->waiting_for_pick_batches_count], |
|
|
|
|
GRPC_ERROR_REF(error), calld->call_combiner); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void waiting_for_pick_batches_fail_locked(grpc_exec_ctx *exec_ctx, |
|
|
|
|
grpc_call_element *elem, |
|
|
|
|
grpc_error *error) { |
|
|
|
|
// This is called via the call combiner, so access to calld is synchronized.
|
|
|
|
|
static void waiting_for_pick_batches_fail(grpc_exec_ctx *exec_ctx, |
|
|
|
|
grpc_call_element *elem, |
|
|
|
|
grpc_error *error) { |
|
|
|
|
call_data *calld = elem->call_data; |
|
|
|
|
if (GRPC_TRACER_ON(grpc_client_channel_trace)) { |
|
|
|
|
gpr_log(GPR_DEBUG, |
|
|
|
|
@ -900,34 +889,60 @@ static void waiting_for_pick_batches_fail_locked(grpc_exec_ctx *exec_ctx, |
|
|
|
|
grpc_error_string(error)); |
|
|
|
|
} |
|
|
|
|
for (size_t i = 0; i < calld->waiting_for_pick_batches_count; ++i) { |
|
|
|
|
GRPC_CLOSURE_INIT(&calld->handle_pending_batch_in_call_combiner[i], |
|
|
|
|
fail_pending_batch_in_call_combiner, calld, |
|
|
|
|
grpc_schedule_on_exec_ctx); |
|
|
|
|
GRPC_CALL_COMBINER_START(exec_ctx, calld->call_combiner, |
|
|
|
|
&calld->handle_pending_batch_in_call_combiner[i], |
|
|
|
|
GRPC_ERROR_REF(error), |
|
|
|
|
"waiting_for_pick_batches_fail"); |
|
|
|
|
} |
|
|
|
|
if (calld->initial_metadata_batch != NULL) { |
|
|
|
|
grpc_transport_stream_op_batch_finish_with_failure( |
|
|
|
|
exec_ctx, calld->waiting_for_pick_batches[i], GRPC_ERROR_REF(error)); |
|
|
|
|
exec_ctx, calld->initial_metadata_batch, GRPC_ERROR_REF(error), |
|
|
|
|
calld->call_combiner); |
|
|
|
|
} else { |
|
|
|
|
GRPC_CALL_COMBINER_STOP(exec_ctx, calld->call_combiner, |
|
|
|
|
"waiting_for_pick_batches_fail"); |
|
|
|
|
} |
|
|
|
|
calld->waiting_for_pick_batches_count = 0; |
|
|
|
|
GRPC_ERROR_UNREF(error); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void waiting_for_pick_batches_resume_locked(grpc_exec_ctx *exec_ctx, |
|
|
|
|
grpc_call_element *elem) { |
|
|
|
|
call_data *calld = elem->call_data; |
|
|
|
|
if (calld->waiting_for_pick_batches_count == 0) return; |
|
|
|
|
call_or_error coe = get_call_or_error(calld); |
|
|
|
|
if (coe.error != GRPC_ERROR_NONE) { |
|
|
|
|
waiting_for_pick_batches_fail_locked(exec_ctx, elem, |
|
|
|
|
GRPC_ERROR_REF(coe.error)); |
|
|
|
|
return; |
|
|
|
|
// This is called via the call combiner, so access to calld is synchronized.
|
|
|
|
|
static void run_pending_batch_in_call_combiner(grpc_exec_ctx *exec_ctx, |
|
|
|
|
void *arg, grpc_error *ignored) { |
|
|
|
|
call_data *calld = arg; |
|
|
|
|
if (calld->waiting_for_pick_batches_count > 0) { |
|
|
|
|
--calld->waiting_for_pick_batches_count; |
|
|
|
|
grpc_subchannel_call_process_op( |
|
|
|
|
exec_ctx, calld->subchannel_call, |
|
|
|
|
calld->waiting_for_pick_batches[calld->waiting_for_pick_batches_count]); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// This is called via the call combiner, so access to calld is synchronized.
|
|
|
|
|
static void waiting_for_pick_batches_resume(grpc_exec_ctx *exec_ctx, |
|
|
|
|
grpc_call_element *elem) { |
|
|
|
|
channel_data *chand = elem->channel_data; |
|
|
|
|
call_data *calld = elem->call_data; |
|
|
|
|
if (GRPC_TRACER_ON(grpc_client_channel_trace)) { |
|
|
|
|
gpr_log(GPR_DEBUG, "chand=%p calld=%p: sending %" PRIdPTR |
|
|
|
|
" pending batches to subchannel_call=%p", |
|
|
|
|
elem->channel_data, calld, calld->waiting_for_pick_batches_count, |
|
|
|
|
coe.subchannel_call); |
|
|
|
|
chand, calld, calld->waiting_for_pick_batches_count, |
|
|
|
|
calld->subchannel_call); |
|
|
|
|
} |
|
|
|
|
for (size_t i = 0; i < calld->waiting_for_pick_batches_count; ++i) { |
|
|
|
|
grpc_subchannel_call_process_op(exec_ctx, coe.subchannel_call, |
|
|
|
|
calld->waiting_for_pick_batches[i]); |
|
|
|
|
} |
|
|
|
|
calld->waiting_for_pick_batches_count = 0; |
|
|
|
|
GRPC_CLOSURE_INIT(&calld->handle_pending_batch_in_call_combiner[i], |
|
|
|
|
run_pending_batch_in_call_combiner, calld, |
|
|
|
|
grpc_schedule_on_exec_ctx); |
|
|
|
|
GRPC_CALL_COMBINER_START(exec_ctx, calld->call_combiner, |
|
|
|
|
&calld->handle_pending_batch_in_call_combiner[i], |
|
|
|
|
GRPC_ERROR_NONE, |
|
|
|
|
"waiting_for_pick_batches_resume"); |
|
|
|
|
} |
|
|
|
|
GPR_ASSERT(calld->initial_metadata_batch != NULL); |
|
|
|
|
grpc_subchannel_call_process_op(exec_ctx, calld->subchannel_call, |
|
|
|
|
calld->initial_metadata_batch); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// Applies service config to the call. Must be invoked once we know
|
|
|
|
|
@ -968,29 +983,28 @@ static void apply_service_config_to_call_locked(grpc_exec_ctx *exec_ctx, |
|
|
|
|
static void create_subchannel_call_locked(grpc_exec_ctx *exec_ctx, |
|
|
|
|
grpc_call_element *elem, |
|
|
|
|
grpc_error *error) { |
|
|
|
|
channel_data *chand = elem->channel_data; |
|
|
|
|
call_data *calld = elem->call_data; |
|
|
|
|
grpc_subchannel_call *subchannel_call = NULL; |
|
|
|
|
const grpc_connected_subchannel_call_args call_args = { |
|
|
|
|
.pollent = calld->pollent, |
|
|
|
|
.path = calld->path, |
|
|
|
|
.start_time = calld->call_start_time, |
|
|
|
|
.deadline = calld->deadline, |
|
|
|
|
.arena = calld->arena, |
|
|
|
|
.context = calld->subchannel_call_context}; |
|
|
|
|
.context = calld->subchannel_call_context, |
|
|
|
|
.call_combiner = calld->call_combiner}; |
|
|
|
|
grpc_error *new_error = grpc_connected_subchannel_create_call( |
|
|
|
|
exec_ctx, calld->connected_subchannel, &call_args, &subchannel_call); |
|
|
|
|
exec_ctx, calld->connected_subchannel, &call_args, |
|
|
|
|
&calld->subchannel_call); |
|
|
|
|
if (GRPC_TRACER_ON(grpc_client_channel_trace)) { |
|
|
|
|
gpr_log(GPR_DEBUG, "chand=%p calld=%p: create subchannel_call=%p: error=%s", |
|
|
|
|
elem->channel_data, calld, subchannel_call, |
|
|
|
|
grpc_error_string(new_error)); |
|
|
|
|
chand, calld, calld->subchannel_call, grpc_error_string(new_error)); |
|
|
|
|
} |
|
|
|
|
GPR_ASSERT(set_call_or_error( |
|
|
|
|
calld, (call_or_error){.subchannel_call = subchannel_call})); |
|
|
|
|
if (new_error != GRPC_ERROR_NONE) { |
|
|
|
|
new_error = grpc_error_add_child(new_error, error); |
|
|
|
|
waiting_for_pick_batches_fail_locked(exec_ctx, elem, new_error); |
|
|
|
|
waiting_for_pick_batches_fail(exec_ctx, elem, new_error); |
|
|
|
|
} else { |
|
|
|
|
waiting_for_pick_batches_resume_locked(exec_ctx, elem); |
|
|
|
|
waiting_for_pick_batches_resume(exec_ctx, elem); |
|
|
|
|
} |
|
|
|
|
GRPC_ERROR_UNREF(error); |
|
|
|
|
} |
|
|
|
|
@ -1002,60 +1016,27 @@ static void subchannel_ready_locked(grpc_exec_ctx *exec_ctx, |
|
|
|
|
channel_data *chand = elem->channel_data; |
|
|
|
|
grpc_polling_entity_del_from_pollset_set(exec_ctx, calld->pollent, |
|
|
|
|
chand->interested_parties); |
|
|
|
|
call_or_error coe = get_call_or_error(calld); |
|
|
|
|
if (calld->connected_subchannel == NULL) { |
|
|
|
|
// Failed to create subchannel.
|
|
|
|
|
grpc_error *failure = |
|
|
|
|
error == GRPC_ERROR_NONE |
|
|
|
|
? GRPC_ERROR_CREATE_FROM_STATIC_STRING( |
|
|
|
|
"Call dropped by load balancing policy") |
|
|
|
|
: GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING( |
|
|
|
|
"Failed to create subchannel", &error, 1); |
|
|
|
|
GRPC_ERROR_UNREF(calld->error); |
|
|
|
|
calld->error = error == GRPC_ERROR_NONE |
|
|
|
|
? GRPC_ERROR_CREATE_FROM_STATIC_STRING( |
|
|
|
|
"Call dropped by load balancing policy") |
|
|
|
|
: GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING( |
|
|
|
|
"Failed to create subchannel", &error, 1); |
|
|
|
|
if (GRPC_TRACER_ON(grpc_client_channel_trace)) { |
|
|
|
|
gpr_log(GPR_DEBUG, |
|
|
|
|
"chand=%p calld=%p: failed to create subchannel: error=%s", chand, |
|
|
|
|
calld, grpc_error_string(failure)); |
|
|
|
|
} |
|
|
|
|
set_call_or_error(calld, (call_or_error){.error = GRPC_ERROR_REF(failure)}); |
|
|
|
|
waiting_for_pick_batches_fail_locked(exec_ctx, elem, failure); |
|
|
|
|
} else if (coe.error != GRPC_ERROR_NONE) { |
|
|
|
|
/* already cancelled before subchannel became ready */ |
|
|
|
|
grpc_error *child_errors[] = {error, coe.error}; |
|
|
|
|
grpc_error *cancellation_error = |
|
|
|
|
GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING( |
|
|
|
|
"Cancelled before creating subchannel", child_errors, |
|
|
|
|
GPR_ARRAY_SIZE(child_errors)); |
|
|
|
|
/* if due to deadline, attach the deadline exceeded status to the error */ |
|
|
|
|
if (gpr_time_cmp(calld->deadline, gpr_now(GPR_CLOCK_MONOTONIC)) < 0) { |
|
|
|
|
cancellation_error = |
|
|
|
|
grpc_error_set_int(cancellation_error, GRPC_ERROR_INT_GRPC_STATUS, |
|
|
|
|
GRPC_STATUS_DEADLINE_EXCEEDED); |
|
|
|
|
calld, grpc_error_string(calld->error)); |
|
|
|
|
} |
|
|
|
|
if (GRPC_TRACER_ON(grpc_client_channel_trace)) { |
|
|
|
|
gpr_log(GPR_DEBUG, |
|
|
|
|
"chand=%p calld=%p: cancelled before subchannel became ready: %s", |
|
|
|
|
chand, calld, grpc_error_string(cancellation_error)); |
|
|
|
|
} |
|
|
|
|
waiting_for_pick_batches_fail_locked(exec_ctx, elem, cancellation_error); |
|
|
|
|
waiting_for_pick_batches_fail(exec_ctx, elem, GRPC_ERROR_REF(calld->error)); |
|
|
|
|
} else { |
|
|
|
|
/* Create call on subchannel. */ |
|
|
|
|
create_subchannel_call_locked(exec_ctx, elem, GRPC_ERROR_REF(error)); |
|
|
|
|
} |
|
|
|
|
GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call, "pick_subchannel"); |
|
|
|
|
GRPC_ERROR_UNREF(error); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static char *cc_get_peer(grpc_exec_ctx *exec_ctx, grpc_call_element *elem) { |
|
|
|
|
call_data *calld = elem->call_data; |
|
|
|
|
grpc_subchannel_call *subchannel_call = |
|
|
|
|
get_call_or_error(calld).subchannel_call; |
|
|
|
|
if (subchannel_call == NULL) { |
|
|
|
|
return NULL; |
|
|
|
|
} else { |
|
|
|
|
return grpc_subchannel_call_get_peer(exec_ctx, subchannel_call); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
/** 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 |
|
|
|
|
@ -1069,6 +1050,44 @@ typedef struct { |
|
|
|
|
grpc_closure closure; |
|
|
|
|
} pick_after_resolver_result_args; |
|
|
|
|
|
|
|
|
|
// Note: This runs under the client_channel combiner, but will NOT be
|
|
|
|
|
// holding the call combiner.
|
|
|
|
|
static void pick_after_resolver_result_cancel_locked(grpc_exec_ctx *exec_ctx, |
|
|
|
|
void *arg, |
|
|
|
|
grpc_error *error) { |
|
|
|
|
grpc_call_element *elem = arg; |
|
|
|
|
channel_data *chand = elem->channel_data; |
|
|
|
|
call_data *calld = 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
|
|
|
|
|
// cancelled, 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.
|
|
|
|
|
for (grpc_closure *closure = chand->waiting_for_resolver_result_closures.head; |
|
|
|
|
closure != NULL; closure = closure->next_data.next) { |
|
|
|
|
pick_after_resolver_result_args *args = closure->cb_arg; |
|
|
|
|
if (!args->cancelled && args->elem == elem) { |
|
|
|
|
if (GRPC_TRACER_ON(grpc_client_channel_trace)) { |
|
|
|
|
gpr_log(GPR_DEBUG, |
|
|
|
|
"chand=%p calld=%p: " |
|
|
|
|
"cancelling pick waiting for resolver result", |
|
|
|
|
chand, calld); |
|
|
|
|
} |
|
|
|
|
args->cancelled = true; |
|
|
|
|
// 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
|
|
|
|
|
// 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)); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void pick_after_resolver_result_done_locked(grpc_exec_ctx *exec_ctx, |
|
|
|
|
void *arg, |
|
|
|
|
grpc_error *error) { |
|
|
|
|
@ -1079,21 +1098,24 @@ static void pick_after_resolver_result_done_locked(grpc_exec_ctx *exec_ctx, |
|
|
|
|
gpr_log(GPR_DEBUG, "call cancelled before resolver result"); |
|
|
|
|
} |
|
|
|
|
} else { |
|
|
|
|
channel_data *chand = args->elem->channel_data; |
|
|
|
|
call_data *calld = args->elem->call_data; |
|
|
|
|
grpc_call_element *elem = args->elem; |
|
|
|
|
channel_data *chand = elem->channel_data; |
|
|
|
|
call_data *calld = elem->call_data; |
|
|
|
|
grpc_call_combiner_set_notify_on_cancel(exec_ctx, calld->call_combiner, |
|
|
|
|
NULL); |
|
|
|
|
if (error != GRPC_ERROR_NONE) { |
|
|
|
|
if (GRPC_TRACER_ON(grpc_client_channel_trace)) { |
|
|
|
|
gpr_log(GPR_DEBUG, "chand=%p calld=%p: resolver failed to return data", |
|
|
|
|
chand, calld); |
|
|
|
|
} |
|
|
|
|
subchannel_ready_locked(exec_ctx, args->elem, GRPC_ERROR_REF(error)); |
|
|
|
|
subchannel_ready_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, args->elem)) { |
|
|
|
|
subchannel_ready_locked(exec_ctx, args->elem, GRPC_ERROR_NONE); |
|
|
|
|
if (pick_subchannel_locked(exec_ctx, elem)) { |
|
|
|
|
subchannel_ready_locked(exec_ctx, elem, GRPC_ERROR_NONE); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
@ -1116,41 +1138,33 @@ static void pick_after_resolver_result_start_locked(grpc_exec_ctx *exec_ctx, |
|
|
|
|
args, grpc_combiner_scheduler(chand->combiner)); |
|
|
|
|
grpc_closure_list_append(&chand->waiting_for_resolver_result_closures, |
|
|
|
|
&args->closure, GRPC_ERROR_NONE); |
|
|
|
|
grpc_call_combiner_set_notify_on_cancel( |
|
|
|
|
exec_ctx, calld->call_combiner, |
|
|
|
|
GRPC_CLOSURE_INIT(&calld->cancel_closure, |
|
|
|
|
pick_after_resolver_result_cancel_locked, elem, |
|
|
|
|
grpc_combiner_scheduler(chand->combiner))); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void pick_after_resolver_result_cancel_locked(grpc_exec_ctx *exec_ctx, |
|
|
|
|
grpc_call_element *elem, |
|
|
|
|
grpc_error *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 = arg; |
|
|
|
|
channel_data *chand = elem->channel_data; |
|
|
|
|
call_data *calld = 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
|
|
|
|
|
// cancelled, 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.
|
|
|
|
|
for (grpc_closure *closure = chand->waiting_for_resolver_result_closures.head; |
|
|
|
|
closure != NULL; closure = closure->next_data.next) { |
|
|
|
|
pick_after_resolver_result_args *args = closure->cb_arg; |
|
|
|
|
if (!args->cancelled && args->elem == elem) { |
|
|
|
|
if (GRPC_TRACER_ON(grpc_client_channel_trace)) { |
|
|
|
|
gpr_log(GPR_DEBUG, |
|
|
|
|
"chand=%p calld=%p: " |
|
|
|
|
"cancelling pick waiting for resolver result", |
|
|
|
|
chand, calld); |
|
|
|
|
} |
|
|
|
|
args->cancelled = true; |
|
|
|
|
subchannel_ready_locked(exec_ctx, elem, |
|
|
|
|
GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING( |
|
|
|
|
"Pick cancelled", &error, 1)); |
|
|
|
|
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_ERROR_UNREF(error); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// Callback invoked by grpc_lb_policy_pick_locked() for async picks.
|
|
|
|
|
// Unrefs the LB policy after invoking subchannel_ready_locked().
|
|
|
|
|
// 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) { |
|
|
|
|
grpc_call_element *elem = arg; |
|
|
|
|
@ -1160,6 +1174,7 @@ static void pick_callback_done_locked(grpc_exec_ctx *exec_ctx, void *arg, |
|
|
|
|
gpr_log(GPR_DEBUG, "chand=%p calld=%p: pick completed asynchronously", |
|
|
|
|
chand, calld); |
|
|
|
|
} |
|
|
|
|
grpc_call_combiner_set_notify_on_cancel(exec_ctx, calld->call_combiner, NULL); |
|
|
|
|
GPR_ASSERT(calld->lb_policy != NULL); |
|
|
|
|
GRPC_LB_POLICY_UNREF(exec_ctx, calld->lb_policy, "pick_subchannel"); |
|
|
|
|
calld->lb_policy = NULL; |
|
|
|
|
@ -1194,24 +1209,15 @@ static bool pick_callback_start_locked(grpc_exec_ctx *exec_ctx, |
|
|
|
|
} |
|
|
|
|
GRPC_LB_POLICY_UNREF(exec_ctx, calld->lb_policy, "pick_subchannel"); |
|
|
|
|
calld->lb_policy = NULL; |
|
|
|
|
} else { |
|
|
|
|
grpc_call_combiner_set_notify_on_cancel( |
|
|
|
|
exec_ctx, calld->call_combiner, |
|
|
|
|
GRPC_CLOSURE_INIT(&calld->cancel_closure, pick_callback_cancel_locked, |
|
|
|
|
elem, grpc_combiner_scheduler(chand->combiner))); |
|
|
|
|
} |
|
|
|
|
return pick_done; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void pick_callback_cancel_locked(grpc_exec_ctx *exec_ctx, |
|
|
|
|
grpc_call_element *elem, |
|
|
|
|
grpc_error *error) { |
|
|
|
|
channel_data *chand = elem->channel_data; |
|
|
|
|
call_data *calld = elem->call_data; |
|
|
|
|
GPR_ASSERT(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, error); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static bool pick_subchannel_locked(grpc_exec_ctx *exec_ctx, |
|
|
|
|
grpc_call_element *elem) { |
|
|
|
|
GPR_TIMER_BEGIN("pick_subchannel", 0); |
|
|
|
|
@ -1224,7 +1230,7 @@ static bool pick_subchannel_locked(grpc_exec_ctx *exec_ctx, |
|
|
|
|
// Otherwise, if the service config specified a value for this
|
|
|
|
|
// method, use that.
|
|
|
|
|
uint32_t initial_metadata_flags = |
|
|
|
|
calld->initial_metadata_payload->send_initial_metadata |
|
|
|
|
calld->initial_metadata_batch->payload->send_initial_metadata |
|
|
|
|
.send_initial_metadata_flags; |
|
|
|
|
const bool wait_for_ready_set_from_api = |
|
|
|
|
initial_metadata_flags & |
|
|
|
|
@ -1241,7 +1247,7 @@ static bool pick_subchannel_locked(grpc_exec_ctx *exec_ctx, |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
const grpc_lb_policy_pick_args inputs = { |
|
|
|
|
calld->initial_metadata_payload->send_initial_metadata |
|
|
|
|
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); |
|
|
|
|
@ -1258,91 +1264,33 @@ static bool pick_subchannel_locked(grpc_exec_ctx *exec_ctx, |
|
|
|
|
return pick_done; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void start_transport_stream_op_batch_locked(grpc_exec_ctx *exec_ctx, |
|
|
|
|
void *arg, |
|
|
|
|
grpc_error *error_ignored) { |
|
|
|
|
GPR_TIMER_BEGIN("start_transport_stream_op_batch_locked", 0); |
|
|
|
|
grpc_transport_stream_op_batch *batch = arg; |
|
|
|
|
grpc_call_element *elem = batch->handler_private.extra_arg; |
|
|
|
|
call_data *calld = elem->call_data; |
|
|
|
|
channel_data *chand = elem->channel_data; |
|
|
|
|
/* need to recheck that another thread hasn't set the call */ |
|
|
|
|
call_or_error coe = get_call_or_error(calld); |
|
|
|
|
if (coe.error != GRPC_ERROR_NONE) { |
|
|
|
|
if (GRPC_TRACER_ON(grpc_client_channel_trace)) { |
|
|
|
|
gpr_log(GPR_DEBUG, "chand=%p calld=%p: failing batch with error: %s", |
|
|
|
|
chand, calld, grpc_error_string(coe.error)); |
|
|
|
|
} |
|
|
|
|
grpc_transport_stream_op_batch_finish_with_failure( |
|
|
|
|
exec_ctx, batch, GRPC_ERROR_REF(coe.error)); |
|
|
|
|
goto done; |
|
|
|
|
} |
|
|
|
|
if (coe.subchannel_call != NULL) { |
|
|
|
|
if (GRPC_TRACER_ON(grpc_client_channel_trace)) { |
|
|
|
|
gpr_log(GPR_DEBUG, |
|
|
|
|
"chand=%p calld=%p: sending batch to subchannel_call=%p", chand, |
|
|
|
|
calld, coe.subchannel_call); |
|
|
|
|
} |
|
|
|
|
grpc_subchannel_call_process_op(exec_ctx, coe.subchannel_call, batch); |
|
|
|
|
goto done; |
|
|
|
|
} |
|
|
|
|
// Add to waiting-for-pick list. If we succeed in getting a
|
|
|
|
|
// subchannel call below, we'll handle this batch (along with any
|
|
|
|
|
// other waiting batches) in waiting_for_pick_batches_resume_locked().
|
|
|
|
|
waiting_for_pick_batches_add_locked(calld, batch); |
|
|
|
|
// If this is a cancellation, cancel the pending pick (if any) and
|
|
|
|
|
// fail any pending batches.
|
|
|
|
|
if (batch->cancel_stream) { |
|
|
|
|
grpc_error *error = batch->payload->cancel_stream.cancel_error; |
|
|
|
|
if (GRPC_TRACER_ON(grpc_client_channel_trace)) { |
|
|
|
|
gpr_log(GPR_DEBUG, "chand=%p calld=%p: recording cancel_error=%s", chand, |
|
|
|
|
calld, grpc_error_string(error)); |
|
|
|
|
} |
|
|
|
|
/* Stash a copy of cancel_error in our call data, so that we can use
|
|
|
|
|
it for subsequent operations. This ensures that if the call is |
|
|
|
|
cancelled before any batches are passed down (e.g., if the deadline |
|
|
|
|
is in the past when the call starts), we can return the right |
|
|
|
|
error to the caller when the first batch does get passed down. */ |
|
|
|
|
set_call_or_error(calld, (call_or_error){.error = GRPC_ERROR_REF(error)}); |
|
|
|
|
if (calld->lb_policy != NULL) { |
|
|
|
|
pick_callback_cancel_locked(exec_ctx, elem, GRPC_ERROR_REF(error)); |
|
|
|
|
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 { |
|
|
|
|
pick_after_resolver_result_cancel_locked(exec_ctx, elem, |
|
|
|
|
GRPC_ERROR_REF(error)); |
|
|
|
|
} |
|
|
|
|
waiting_for_pick_batches_fail_locked(exec_ctx, elem, GRPC_ERROR_REF(error)); |
|
|
|
|
goto done; |
|
|
|
|
} |
|
|
|
|
/* if we don't have a subchannel, try to get one */ |
|
|
|
|
if (batch->send_initial_metadata) { |
|
|
|
|
GPR_ASSERT(calld->connected_subchannel == NULL); |
|
|
|
|
calld->initial_metadata_payload = batch->payload; |
|
|
|
|
GRPC_CALL_STACK_REF(calld->owning_call, "pick_subchannel"); |
|
|
|
|
/* If a subchannel is not available immediately, the polling entity from
|
|
|
|
|
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. */ |
|
|
|
|
if (pick_subchannel_locked(exec_ctx, elem)) { |
|
|
|
|
// Pick was returned synchronously.
|
|
|
|
|
GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call, "pick_subchannel"); |
|
|
|
|
if (calld->connected_subchannel == NULL) { |
|
|
|
|
grpc_error *error = GRPC_ERROR_CREATE_FROM_STATIC_STRING( |
|
|
|
|
"Call dropped by load balancing policy"); |
|
|
|
|
set_call_or_error(calld, |
|
|
|
|
(call_or_error){.error = GRPC_ERROR_REF(error)}); |
|
|
|
|
waiting_for_pick_batches_fail_locked(exec_ctx, elem, error); |
|
|
|
|
} else { |
|
|
|
|
// Create subchannel call.
|
|
|
|
|
create_subchannel_call_locked(exec_ctx, elem, GRPC_ERROR_NONE); |
|
|
|
|
} |
|
|
|
|
} else { |
|
|
|
|
grpc_polling_entity_add_to_pollset_set(exec_ctx, calld->pollent, |
|
|
|
|
chand->interested_parties); |
|
|
|
|
// 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); |
|
|
|
|
} |
|
|
|
|
done: |
|
|
|
|
GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call, |
|
|
|
|
"start_transport_stream_op_batch"); |
|
|
|
|
GPR_TIMER_END("start_transport_stream_op_batch_locked", 0); |
|
|
|
|
GPR_TIMER_END("start_pick_locked", 0); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void on_complete(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { |
|
|
|
|
@ -1365,27 +1313,49 @@ static void on_complete(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { |
|
|
|
|
GRPC_ERROR_REF(error)); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
/* 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. |
|
|
|
|
|
|
|
|
|
We use double-checked locking to initially see if initialization has been |
|
|
|
|
performed. If it has not, we acquire the combiner and perform initialization. |
|
|
|
|
If it has, we proceed on the fast path. */ |
|
|
|
|
static void cc_start_transport_stream_op_batch( |
|
|
|
|
grpc_exec_ctx *exec_ctx, grpc_call_element *elem, |
|
|
|
|
grpc_transport_stream_op_batch *batch) { |
|
|
|
|
call_data *calld = elem->call_data; |
|
|
|
|
channel_data *chand = elem->channel_data; |
|
|
|
|
if (GRPC_TRACER_ON(grpc_client_channel_trace) || |
|
|
|
|
GRPC_TRACER_ON(grpc_trace_channel)) { |
|
|
|
|
grpc_call_log_op(GPR_INFO, elem, batch); |
|
|
|
|
} |
|
|
|
|
if (chand->deadline_checking_enabled) { |
|
|
|
|
grpc_deadline_state_client_start_transport_stream_op_batch(exec_ctx, elem, |
|
|
|
|
batch); |
|
|
|
|
} |
|
|
|
|
GPR_TIMER_BEGIN("cc_start_transport_stream_op_batch", 0); |
|
|
|
|
// If we've previously been cancelled, immediately fail any new batches.
|
|
|
|
|
if (calld->error != GRPC_ERROR_NONE) { |
|
|
|
|
if (GRPC_TRACER_ON(grpc_client_channel_trace)) { |
|
|
|
|
gpr_log(GPR_DEBUG, "chand=%p calld=%p: failing batch with error: %s", |
|
|
|
|
chand, calld, grpc_error_string(calld->error)); |
|
|
|
|
} |
|
|
|
|
grpc_transport_stream_op_batch_finish_with_failure( |
|
|
|
|
exec_ctx, batch, GRPC_ERROR_REF(calld->error), calld->call_combiner); |
|
|
|
|
goto done; |
|
|
|
|
} |
|
|
|
|
if (batch->cancel_stream) { |
|
|
|
|
// Stash a copy of cancel_error in our call data, so that we can use
|
|
|
|
|
// it for subsequent operations. This ensures that if the call is
|
|
|
|
|
// cancelled before any batches are passed down (e.g., if the deadline
|
|
|
|
|
// is in the past when the call starts), we can return the right
|
|
|
|
|
// error to the caller when the first batch does get passed down.
|
|
|
|
|
GRPC_ERROR_UNREF(calld->error); |
|
|
|
|
calld->error = GRPC_ERROR_REF(batch->payload->cancel_stream.cancel_error); |
|
|
|
|
if (GRPC_TRACER_ON(grpc_client_channel_trace)) { |
|
|
|
|
gpr_log(GPR_DEBUG, "chand=%p calld=%p: recording cancel_error=%s", chand, |
|
|
|
|
calld, grpc_error_string(calld->error)); |
|
|
|
|
} |
|
|
|
|
// If we have a subchannel call, send the cancellation batch down.
|
|
|
|
|
// Otherwise, fail all pending batches.
|
|
|
|
|
if (calld->subchannel_call != NULL) { |
|
|
|
|
grpc_subchannel_call_process_op(exec_ctx, calld->subchannel_call, batch); |
|
|
|
|
} else { |
|
|
|
|
waiting_for_pick_batches_add(calld, batch); |
|
|
|
|
waiting_for_pick_batches_fail(exec_ctx, elem, |
|
|
|
|
GRPC_ERROR_REF(calld->error)); |
|
|
|
|
} |
|
|
|
|
goto done; |
|
|
|
|
} |
|
|
|
|
// Intercept on_complete for recv_trailing_metadata so that we can
|
|
|
|
|
// check retry throttle status.
|
|
|
|
|
if (batch->recv_trailing_metadata) { |
|
|
|
|
@ -1395,38 +1365,43 @@ static void cc_start_transport_stream_op_batch( |
|
|
|
|
grpc_schedule_on_exec_ctx); |
|
|
|
|
batch->on_complete = &calld->on_complete; |
|
|
|
|
} |
|
|
|
|
/* try to (atomically) get the call */ |
|
|
|
|
call_or_error coe = get_call_or_error(calld); |
|
|
|
|
GPR_TIMER_BEGIN("cc_start_transport_stream_op_batch", 0); |
|
|
|
|
if (coe.error != GRPC_ERROR_NONE) { |
|
|
|
|
// Check if we've already gotten a subchannel call.
|
|
|
|
|
// Note that once we have completed the pick, we do not need to enter
|
|
|
|
|
// the channel combiner, which is more efficient (especially for
|
|
|
|
|
// streaming calls).
|
|
|
|
|
if (calld->subchannel_call != NULL) { |
|
|
|
|
if (GRPC_TRACER_ON(grpc_client_channel_trace)) { |
|
|
|
|
gpr_log(GPR_DEBUG, "chand=%p calld=%p: failing batch with error: %s", |
|
|
|
|
chand, calld, grpc_error_string(coe.error)); |
|
|
|
|
gpr_log(GPR_DEBUG, |
|
|
|
|
"chand=%p calld=%p: sending batch to subchannel_call=%p", chand, |
|
|
|
|
calld, calld->subchannel_call); |
|
|
|
|
} |
|
|
|
|
grpc_transport_stream_op_batch_finish_with_failure( |
|
|
|
|
exec_ctx, batch, GRPC_ERROR_REF(coe.error)); |
|
|
|
|
grpc_subchannel_call_process_op(exec_ctx, calld->subchannel_call, batch); |
|
|
|
|
goto done; |
|
|
|
|
} |
|
|
|
|
if (coe.subchannel_call != NULL) { |
|
|
|
|
// We do not yet have a subchannel call.
|
|
|
|
|
// Add the batch to the waiting-for-pick list.
|
|
|
|
|
waiting_for_pick_batches_add(calld, batch); |
|
|
|
|
// For batches containing a send_initial_metadata op, enter the channel
|
|
|
|
|
// 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); |
|
|
|
|
} |
|
|
|
|
GRPC_CLOSURE_SCHED( |
|
|
|
|
exec_ctx, |
|
|
|
|
GRPC_CLOSURE_INIT(&batch->handler_private.closure, start_pick_locked, |
|
|
|
|
elem, grpc_combiner_scheduler(chand->combiner)), |
|
|
|
|
GRPC_ERROR_NONE); |
|
|
|
|
} else { |
|
|
|
|
// For all other batches, release the call combiner.
|
|
|
|
|
if (GRPC_TRACER_ON(grpc_client_channel_trace)) { |
|
|
|
|
gpr_log(GPR_DEBUG, |
|
|
|
|
"chand=%p calld=%p: sending batch to subchannel_call=%p", chand, |
|
|
|
|
calld, coe.subchannel_call); |
|
|
|
|
"chand=%p calld=%p: saved batch, yeilding call combiner", chand, |
|
|
|
|
calld); |
|
|
|
|
} |
|
|
|
|
grpc_subchannel_call_process_op(exec_ctx, coe.subchannel_call, batch); |
|
|
|
|
goto done; |
|
|
|
|
} |
|
|
|
|
/* we failed; lock and figure out what to do */ |
|
|
|
|
if (GRPC_TRACER_ON(grpc_client_channel_trace)) { |
|
|
|
|
gpr_log(GPR_DEBUG, "chand=%p calld=%p: entering combiner", chand, calld); |
|
|
|
|
GRPC_CALL_COMBINER_STOP(exec_ctx, calld->call_combiner, |
|
|
|
|
"batch does not include send_initial_metadata"); |
|
|
|
|
} |
|
|
|
|
GRPC_CALL_STACK_REF(calld->owning_call, "start_transport_stream_op_batch"); |
|
|
|
|
batch->handler_private.extra_arg = elem; |
|
|
|
|
GRPC_CLOSURE_SCHED( |
|
|
|
|
exec_ctx, GRPC_CLOSURE_INIT(&batch->handler_private.closure, |
|
|
|
|
start_transport_stream_op_batch_locked, batch, |
|
|
|
|
grpc_combiner_scheduler(chand->combiner)), |
|
|
|
|
GRPC_ERROR_NONE); |
|
|
|
|
done: |
|
|
|
|
GPR_TIMER_END("cc_start_transport_stream_op_batch", 0); |
|
|
|
|
} |
|
|
|
|
@ -1441,10 +1416,11 @@ static grpc_error *cc_init_call_elem(grpc_exec_ctx *exec_ctx, |
|
|
|
|
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->owning_call = args->call_stack; |
|
|
|
|
calld->arena = args->arena; |
|
|
|
|
calld->call_combiner = args->call_combiner; |
|
|
|
|
if (chand->deadline_checking_enabled) { |
|
|
|
|
grpc_deadline_state_init(exec_ctx, elem, args->call_stack, calld->deadline); |
|
|
|
|
grpc_deadline_state_init(exec_ctx, elem, args->call_stack, |
|
|
|
|
args->call_combiner, calld->deadline); |
|
|
|
|
} |
|
|
|
|
return GRPC_ERROR_NONE; |
|
|
|
|
} |
|
|
|
|
@ -1463,13 +1439,12 @@ static void cc_destroy_call_elem(grpc_exec_ctx *exec_ctx, |
|
|
|
|
if (calld->method_params != NULL) { |
|
|
|
|
method_parameters_unref(calld->method_params); |
|
|
|
|
} |
|
|
|
|
call_or_error coe = get_call_or_error(calld); |
|
|
|
|
GRPC_ERROR_UNREF(coe.error); |
|
|
|
|
if (coe.subchannel_call != NULL) { |
|
|
|
|
grpc_subchannel_call_set_cleanup_closure(coe.subchannel_call, |
|
|
|
|
GRPC_ERROR_UNREF(calld->error); |
|
|
|
|
if (calld->subchannel_call != NULL) { |
|
|
|
|
grpc_subchannel_call_set_cleanup_closure(calld->subchannel_call, |
|
|
|
|
then_schedule_closure); |
|
|
|
|
then_schedule_closure = NULL; |
|
|
|
|
GRPC_SUBCHANNEL_CALL_UNREF(exec_ctx, coe.subchannel_call, |
|
|
|
|
GRPC_SUBCHANNEL_CALL_UNREF(exec_ctx, calld->subchannel_call, |
|
|
|
|
"client_channel_destroy_call"); |
|
|
|
|
} |
|
|
|
|
GPR_ASSERT(calld->lb_policy == NULL); |
|
|
|
|
@ -1508,7 +1483,6 @@ const grpc_channel_filter grpc_client_channel_filter = { |
|
|
|
|
sizeof(channel_data), |
|
|
|
|
cc_init_channel_elem, |
|
|
|
|
cc_destroy_channel_elem, |
|
|
|
|
cc_get_peer, |
|
|
|
|
cc_get_channel_info, |
|
|
|
|
"client-channel", |
|
|
|
|
}; |
|
|
|
|
|
Mark D. Roth
7 years ago