aboutsummaryrefslogtreecommitdiffhomepage
path: root/src
diff options
context:
space:
mode:
authorGravatar Mark D. Roth <roth@google.com>2017-05-02 08:27:08 -0700
committerGravatar Mark D. Roth <roth@google.com>2017-05-02 08:27:08 -0700
commit64a317cc3e249a1b81327807f3dd7b998e0d9b88 (patch)
tree25bfe0082ce40fbea1e1c174de80ab63cae60619 /src
parentd8696e10238a7e35e25f363e16f921c6c642069a (diff)
Code clean-up in client_channel.
Diffstat (limited to 'src')
-rw-r--r--src/core/ext/filters/client_channel/client_channel.c101
1 files changed, 43 insertions, 58 deletions
diff --git a/src/core/ext/filters/client_channel/client_channel.c b/src/core/ext/filters/client_channel/client_channel.c
index 0463b25412..652d789ec4 100644
--- a/src/core/ext/filters/client_channel/client_channel.c
+++ b/src/core/ext/filters/client_channel/client_channel.c
@@ -760,12 +760,6 @@ static void cc_destroy_channel_elem(grpc_exec_ctx *exec_ctx,
#define CANCELLED_CALL ((grpc_subchannel_call *)1)
-typedef enum {
- /* zero so that it can be default-initialized */
- GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING = 0,
- GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL
-} subchannel_creation_phase;
-
/** Call data. Holds a pointer to grpc_subchannel_call and the
associated machinery to create such a pointer.
Handles queueing of stream ops until a call object is ready, waiting
@@ -793,7 +787,7 @@ typedef struct client_channel_call_data {
gpr_atm subchannel_call;
gpr_arena *arena;
- subchannel_creation_phase creation_phase;
+ bool pick_pending;
grpc_connected_subchannel *connected_subchannel;
grpc_polling_entity *pollent;
@@ -914,11 +908,10 @@ static void subchannel_ready_locked(grpc_exec_ctx *exec_ctx, void *arg,
grpc_call_element *elem = arg;
call_data *calld = elem->call_data;
channel_data *chand = elem->channel_data;
- GPR_ASSERT(calld->creation_phase ==
- GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL);
+ GPR_ASSERT(calld->pick_pending);
+ calld->pick_pending = false;
grpc_polling_entity_del_from_pollset_set(exec_ctx, calld->pollent,
chand->interested_parties);
- calld->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING;
if (calld->connected_subchannel == NULL) {
gpr_atm_no_barrier_store(&calld->subchannel_call, 1);
fail_locked(exec_ctx, calld,
@@ -984,8 +977,7 @@ typedef struct {
static bool pick_subchannel_locked(
grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
grpc_metadata_batch *initial_metadata, uint32_t initial_metadata_flags,
- grpc_connected_subchannel **connected_subchannel, grpc_closure *on_ready,
- grpc_error *error);
+ grpc_connected_subchannel **connected_subchannel, grpc_closure *on_ready);
static void continue_picking_locked(grpc_exec_ctx *exec_ctx, void *arg,
grpc_error *error) {
@@ -997,49 +989,46 @@ static void continue_picking_locked(grpc_exec_ctx *exec_ctx, void *arg,
} else {
if (pick_subchannel_locked(exec_ctx, cpa->elem, cpa->initial_metadata,
cpa->initial_metadata_flags,
- cpa->connected_subchannel, cpa->on_ready,
- GRPC_ERROR_NONE)) {
+ cpa->connected_subchannel, cpa->on_ready)) {
grpc_closure_sched(exec_ctx, cpa->on_ready, GRPC_ERROR_NONE);
}
}
gpr_free(cpa);
}
+static void cancel_pick_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;
+ if (chand->lb_policy != NULL) {
+ grpc_lb_policy_cancel_pick_locked(exec_ctx, chand->lb_policy,
+ &calld->connected_subchannel,
+ GRPC_ERROR_REF(error));
+ }
+ for (grpc_closure *closure = chand->waiting_for_config_closures.head;
+ closure != NULL; closure = closure->next_data.next) {
+ continue_picking_args *cpa = closure->cb_arg;
+ if (cpa->connected_subchannel == &calld->connected_subchannel) {
+ cpa->connected_subchannel = NULL;
+ grpc_closure_sched(exec_ctx, cpa->on_ready,
+ GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING(
+ "Pick cancelled", &error, 1));
+ }
+ }
+ GRPC_ERROR_UNREF(error);
+}
+
static bool pick_subchannel_locked(
grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
grpc_metadata_batch *initial_metadata, uint32_t initial_metadata_flags,
- grpc_connected_subchannel **connected_subchannel, grpc_closure *on_ready,
- grpc_error *error) {
+ grpc_connected_subchannel **connected_subchannel, grpc_closure *on_ready) {
GPR_TIMER_BEGIN("pick_subchannel", 0);
channel_data *chand = elem->channel_data;
call_data *calld = elem->call_data;
- continue_picking_args *cpa;
- grpc_closure *closure;
GPR_ASSERT(connected_subchannel);
- if (initial_metadata == NULL) {
- if (chand->lb_policy != NULL) {
- grpc_lb_policy_cancel_pick_locked(exec_ctx, chand->lb_policy,
- connected_subchannel,
- GRPC_ERROR_REF(error));
- }
- for (closure = chand->waiting_for_config_closures.head; closure != NULL;
- closure = closure->next_data.next) {
- cpa = closure->cb_arg;
- if (cpa->connected_subchannel == connected_subchannel) {
- cpa->connected_subchannel = NULL;
- grpc_closure_sched(exec_ctx, cpa->on_ready,
- GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING(
- "Pick cancelled", &error, 1));
- }
- }
- GPR_TIMER_END("pick_subchannel", 0);
- GRPC_ERROR_UNREF(error);
- return true;
- }
- GPR_ASSERT(error == GRPC_ERROR_NONE);
if (chand->lb_policy != NULL) {
apply_final_configuration_locked(exec_ctx, elem);
grpc_lb_policy *lb_policy = chand->lb_policy;
@@ -1096,7 +1085,7 @@ static bool pick_subchannel_locked(
&chand->on_resolver_result_changed);
}
if (chand->resolver != NULL) {
- cpa = gpr_malloc(sizeof(*cpa));
+ continue_picking_args *cpa = gpr_malloc(sizeof(*cpa));
cpa->initial_metadata = initial_metadata;
cpa->initial_metadata_flags = initial_metadata_flags;
cpa->connected_subchannel = connected_subchannel;
@@ -1151,16 +1140,13 @@ static void start_transport_stream_op_batch_locked_inner(
error to the caller when the first op does get passed down. */
calld->cancel_error =
GRPC_ERROR_REF(op->payload->cancel_stream.cancel_error);
- switch (calld->creation_phase) {
- case GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING:
- fail_locked(exec_ctx, calld,
- GRPC_ERROR_REF(op->payload->cancel_stream.cancel_error));
- break;
- case GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL:
- pick_subchannel_locked(
- exec_ctx, elem, NULL, 0, &calld->connected_subchannel, NULL,
- GRPC_ERROR_REF(op->payload->cancel_stream.cancel_error));
- break;
+ if (calld->pick_pending) {
+ cancel_pick_locked(
+ exec_ctx, elem,
+ GRPC_ERROR_REF(op->payload->cancel_stream.cancel_error));
+ } else {
+ fail_locked(exec_ctx, calld,
+ GRPC_ERROR_REF(op->payload->cancel_stream.cancel_error));
}
grpc_transport_stream_op_batch_finish_with_failure(
exec_ctx, op,
@@ -1170,9 +1156,9 @@ static void start_transport_stream_op_batch_locked_inner(
}
}
/* if we don't have a subchannel, try to get one */
- if (calld->creation_phase == GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING &&
- calld->connected_subchannel == NULL && op->send_initial_metadata) {
- calld->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL;
+ if (!calld->pick_pending && calld->connected_subchannel == NULL &&
+ op->send_initial_metadata) {
+ calld->pick_pending = true;
grpc_closure_init(&calld->next_step, subchannel_ready_locked, elem,
grpc_combiner_scheduler(chand->combiner, true));
GRPC_CALL_STACK_REF(calld->owning_call, "pick_subchannel");
@@ -1183,8 +1169,8 @@ static void start_transport_stream_op_batch_locked_inner(
exec_ctx, elem,
op->payload->send_initial_metadata.send_initial_metadata,
op->payload->send_initial_metadata.send_initial_metadata_flags,
- &calld->connected_subchannel, &calld->next_step, GRPC_ERROR_NONE)) {
- calld->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING;
+ &calld->connected_subchannel, &calld->next_step)) {
+ calld->pick_pending = false;
GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call, "pick_subchannel");
} else {
grpc_polling_entity_add_to_pollset_set(exec_ctx, calld->pollent,
@@ -1192,8 +1178,7 @@ static void start_transport_stream_op_batch_locked_inner(
}
}
/* if we've got a subchannel, then let's ask it to create a call */
- if (calld->creation_phase == GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING &&
- calld->connected_subchannel != NULL) {
+ if (!calld->pick_pending && calld->connected_subchannel != NULL) {
grpc_subchannel_call *subchannel_call = NULL;
const grpc_connected_subchannel_call_args call_args = {
.pollent = calld->pollent,
@@ -1349,7 +1334,7 @@ static void cc_destroy_call_elem(grpc_exec_ctx *exec_ctx,
then_schedule_closure = NULL;
GRPC_SUBCHANNEL_CALL_UNREF(exec_ctx, call, "client_channel_destroy_call");
}
- GPR_ASSERT(calld->creation_phase == GRPC_SUBCHANNEL_CALL_HOLDER_NOT_CREATING);
+ GPR_ASSERT(!calld->pick_pending);
GPR_ASSERT(calld->waiting_ops_count == 0);
if (calld->connected_subchannel != NULL) {
GRPC_CONNECTED_SUBCHANNEL_UNREF(exec_ctx, calld->connected_subchannel,