|  | @@ -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",
 | 
	
		
			
				|  |  |  };
 |