Merge branch 'minimal' into minimal_test
diff --git a/src/core/ext/filters/client_channel/client_channel.c b/src/core/ext/filters/client_channel/client_channel.c
index 4b6c532..9dda1a2 100644
--- a/src/core/ext/filters/client_channel/client_channel.c
+++ b/src/core/ext/filters/client_channel/client_channel.c
@@ -534,7 +534,7 @@
static void start_transport_op_locked(grpc_exec_ctx *exec_ctx, void *arg,
grpc_error *error_ignored) {
grpc_transport_op *op = arg;
- grpc_channel_element *elem = op->transport_private.args[0];
+ grpc_channel_element *elem = op->handler_private.extra_arg;
channel_data *chand = elem->channel_data;
if (op->on_connectivity_state_change != NULL) {
@@ -596,12 +596,12 @@
op->bind_pollset);
}
- op->transport_private.args[0] = elem;
+ op->handler_private.extra_arg = elem;
GRPC_CHANNEL_STACK_REF(chand->owning_stack, "start_transport_op");
grpc_closure_sched(
- exec_ctx, grpc_closure_init(
- &op->transport_private.closure, start_transport_op_locked,
- op, grpc_combiner_scheduler(chand->combiner, false)),
+ exec_ctx,
+ grpc_closure_init(&op->handler_private.closure, start_transport_op_locked,
+ op, grpc_combiner_scheduler(chand->combiner, false)),
GRPC_ERROR_NONE);
}
@@ -770,7 +770,7 @@
grpc_connected_subchannel *connected_subchannel;
grpc_polling_entity *pollent;
- grpc_transport_stream_op **waiting_ops;
+ grpc_transport_stream_op_batch **waiting_ops;
size_t waiting_ops_count;
size_t waiting_ops_capacity;
@@ -790,7 +790,8 @@
return scc == CANCELLED_CALL ? NULL : scc;
}
-static void add_waiting_locked(call_data *calld, grpc_transport_stream_op *op) {
+static void add_waiting_locked(call_data *calld,
+ grpc_transport_stream_op_batch *op) {
GPR_TIMER_BEGIN("add_waiting_locked", 0);
if (calld->waiting_ops_count == calld->waiting_ops_capacity) {
calld->waiting_ops_capacity = GPR_MAX(3, 2 * calld->waiting_ops_capacity);
@@ -806,7 +807,7 @@
grpc_error *error) {
size_t i;
for (i = 0; i < calld->waiting_ops_count; i++) {
- grpc_transport_stream_op_finish_with_failure(
+ grpc_transport_stream_op_batch_finish_with_failure(
exec_ctx, calld->waiting_ops[i], GRPC_ERROR_REF(error));
}
calld->waiting_ops_count = 0;
@@ -819,7 +820,7 @@
}
grpc_subchannel_call *call = GET_CALL(calld);
- grpc_transport_stream_op **ops = calld->waiting_ops;
+ grpc_transport_stream_op_batch **ops = calld->waiting_ops;
size_t nops = calld->waiting_ops_count;
if (call == CANCELLED_CALL) {
fail_locked(exec_ctx, calld, GRPC_ERROR_CANCELLED);
@@ -1069,9 +1070,9 @@
return false;
}
-static void start_transport_stream_op_locked_inner(grpc_exec_ctx *exec_ctx,
- grpc_transport_stream_op *op,
- grpc_call_element *elem) {
+static void start_transport_stream_op_batch_locked_inner(
+ grpc_exec_ctx *exec_ctx, grpc_transport_stream_op_batch *op,
+ grpc_call_element *elem) {
channel_data *chand = elem->channel_data;
call_data *calld = elem->call_data;
grpc_subchannel_call *call;
@@ -1079,7 +1080,7 @@
/* need to recheck that another thread hasn't set the call */
call = GET_CALL(calld);
if (call == CANCELLED_CALL) {
- grpc_transport_stream_op_finish_with_failure(
+ grpc_transport_stream_op_batch_finish_with_failure(
exec_ctx, op, GRPC_ERROR_REF(calld->cancel_error));
/* early out */
return;
@@ -1090,11 +1091,11 @@
return;
}
/* if this is a cancellation, then we can raise our cancelled flag */
- if (op->cancel_error != GRPC_ERROR_NONE) {
+ if (op->cancel_stream) {
if (!gpr_atm_rel_cas(&calld->subchannel_call, 0,
(gpr_atm)(uintptr_t)CANCELLED_CALL)) {
/* recurse to retry */
- start_transport_stream_op_locked_inner(exec_ctx, op, elem);
+ start_transport_stream_op_batch_locked_inner(exec_ctx, op, elem);
/* early out */
return;
} else {
@@ -1103,27 +1104,29 @@
cancelled before any ops 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 op does get passed down. */
- calld->cancel_error = GRPC_ERROR_REF(op->cancel_error);
+ 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->cancel_error));
+ 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->cancel_error));
+ pick_subchannel_locked(
+ exec_ctx, elem, NULL, 0, &calld->connected_subchannel, NULL,
+ GRPC_ERROR_REF(op->payload->cancel_stream.cancel_error));
break;
}
- grpc_transport_stream_op_finish_with_failure(
- exec_ctx, op, GRPC_ERROR_REF(op->cancel_error));
+ grpc_transport_stream_op_batch_finish_with_failure(
+ exec_ctx, op,
+ GRPC_ERROR_REF(op->payload->cancel_stream.cancel_error));
/* early out */
return;
}
}
/* 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 != NULL) {
+ calld->connected_subchannel == NULL && op->send_initial_metadata) {
calld->creation_phase = GRPC_SUBCHANNEL_CALL_HOLDER_PICKING_SUBCHANNEL;
grpc_closure_init(&calld->next_step, subchannel_ready_locked, elem,
grpc_combiner_scheduler(chand->combiner, true));
@@ -1131,10 +1134,11 @@
/* 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, op->send_initial_metadata,
- op->send_initial_metadata_flags,
- &calld->connected_subchannel, &calld->next_step,
- GRPC_ERROR_NONE)) {
+ if (pick_subchannel_locked(
+ 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;
GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call, "pick_subchannel");
} else {
@@ -1157,13 +1161,13 @@
if (error != GRPC_ERROR_NONE) {
subchannel_call = CANCELLED_CALL;
fail_locked(exec_ctx, calld, GRPC_ERROR_REF(error));
- grpc_transport_stream_op_finish_with_failure(exec_ctx, op, error);
+ grpc_transport_stream_op_batch_finish_with_failure(exec_ctx, op, error);
}
gpr_atm_rel_store(&calld->subchannel_call,
(gpr_atm)(uintptr_t)subchannel_call);
retry_waiting_locked(exec_ctx, calld);
/* recurse to retry */
- start_transport_stream_op_locked_inner(exec_ctx, op, elem);
+ start_transport_stream_op_batch_locked_inner(exec_ctx, op, elem);
/* early out */
return;
}
@@ -1191,15 +1195,16 @@
GRPC_ERROR_REF(error));
}
-static void start_transport_stream_op_locked(grpc_exec_ctx *exec_ctx, void *arg,
- grpc_error *error_ignored) {
- GPR_TIMER_BEGIN("start_transport_stream_op_locked", 0);
+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 *op = arg;
- grpc_call_element *elem = op->handler_private.args[0];
+ grpc_transport_stream_op_batch *op = arg;
+ grpc_call_element *elem = op->handler_private.extra_arg;
call_data *calld = elem->call_data;
- if (op->recv_trailing_metadata != NULL) {
+ if (op->recv_trailing_metadata) {
GPR_ASSERT(op->on_complete != NULL);
calld->original_on_complete = op->on_complete;
grpc_closure_init(&calld->on_complete, on_complete, elem,
@@ -1207,11 +1212,11 @@
op->on_complete = &calld->on_complete;
}
- start_transport_stream_op_locked_inner(exec_ctx, op, elem);
+ start_transport_stream_op_batch_locked_inner(exec_ctx, op, elem);
GRPC_CALL_STACK_UNREF(exec_ctx, calld->owning_call,
- "start_transport_stream_op");
- GPR_TIMER_END("start_transport_stream_op_locked", 0);
+ "start_transport_stream_op_batch");
+ GPR_TIMER_END("start_transport_stream_op_batch_locked", 0);
}
/* The logic here is fairly complicated, due to (a) the fact that we
@@ -1222,41 +1227,42 @@
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(grpc_exec_ctx *exec_ctx,
- grpc_call_element *elem,
- grpc_transport_stream_op *op) {
+static void cc_start_transport_stream_op_batch(
+ grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
+ grpc_transport_stream_op_batch *op) {
call_data *calld = elem->call_data;
channel_data *chand = elem->channel_data;
GRPC_CALL_LOG_OP(GPR_INFO, elem, op);
if (chand->deadline_checking_enabled) {
- grpc_deadline_state_client_start_transport_stream_op(exec_ctx, elem, op);
+ grpc_deadline_state_client_start_transport_stream_op_batch(exec_ctx, elem,
+ op);
}
/* try to (atomically) get the call */
grpc_subchannel_call *call = GET_CALL(calld);
- GPR_TIMER_BEGIN("cc_start_transport_stream_op", 0);
+ GPR_TIMER_BEGIN("cc_start_transport_stream_op_batch", 0);
if (call == CANCELLED_CALL) {
- grpc_transport_stream_op_finish_with_failure(
+ grpc_transport_stream_op_batch_finish_with_failure(
exec_ctx, op, GRPC_ERROR_REF(calld->cancel_error));
- GPR_TIMER_END("cc_start_transport_stream_op", 0);
+ GPR_TIMER_END("cc_start_transport_stream_op_batch", 0);
/* early out */
return;
}
if (call != NULL) {
grpc_subchannel_call_process_op(exec_ctx, call, op);
- GPR_TIMER_END("cc_start_transport_stream_op", 0);
+ GPR_TIMER_END("cc_start_transport_stream_op_batch", 0);
/* early out */
return;
}
/* we failed; lock and figure out what to do */
- GRPC_CALL_STACK_REF(calld->owning_call, "start_transport_stream_op");
- op->handler_private.args[0] = elem;
+ GRPC_CALL_STACK_REF(calld->owning_call, "start_transport_stream_op_batch");
+ op->handler_private.extra_arg = elem;
grpc_closure_sched(
exec_ctx,
grpc_closure_init(&op->handler_private.closure,
- start_transport_stream_op_locked, op,
+ start_transport_stream_op_batch_locked, op,
grpc_combiner_scheduler(chand->combiner, false)),
GRPC_ERROR_NONE);
- GPR_TIMER_END("cc_start_transport_stream_op", 0);
+ GPR_TIMER_END("cc_start_transport_stream_op_batch", 0);
}
/* Constructor for call_data */
@@ -1321,7 +1327,7 @@
*/
const grpc_channel_filter grpc_client_channel_filter = {
- cc_start_transport_stream_op,
+ cc_start_transport_stream_op_batch,
cc_start_transport_op,
sizeof(call_data),
cc_init_call_elem,