1 files changed, 81 insertions, 11 deletions

diff --git a/src/core/lib/channel/connected_channel.c b/src/core/lib/channel/connected_channel.c
index af06ca802e..8285226fc4 100644
--- a/src/core/lib/channel/connected_channel.c
+++ b/src/core/lib/channel/connected_channel.c
@@ -36,7 +36,57 @@ typedef struct connected_channel_channel_data {
   grpc_transport *transport;
 } channel_data;
 
-typedef struct connected_channel_call_data { void *unused; } call_data;
+typedef struct {
+  grpc_closure closure;
+  grpc_closure *original_closure;
+  grpc_call_combiner *call_combiner;
+  const char *reason;
+} callback_state;
+
+typedef struct connected_channel_call_data {
+  grpc_call_combiner *call_combiner;
+  // Closures used for returning results on the call combiner.
+  callback_state on_complete[6];  // Max number of pending batches.
+  callback_state recv_initial_metadata_ready;
+  callback_state recv_message_ready;
+} call_data;
+
+static void run_in_call_combiner(grpc_exec_ctx *exec_ctx, void *arg,
+                                 grpc_error *error) {
+  callback_state *state = (callback_state *)arg;
+  GRPC_CALL_COMBINER_START(exec_ctx, state->call_combiner,
+                           state->original_closure, GRPC_ERROR_REF(error),
+                           state->reason);
+}
+
+static void run_cancel_in_call_combiner(grpc_exec_ctx *exec_ctx, void *arg,
+                                        grpc_error *error) {
+  run_in_call_combiner(exec_ctx, arg, error);
+  gpr_free(arg);
+}
+
+static void intercept_callback(call_data *calld, callback_state *state,
+                               bool free_when_done, const char *reason,
+                               grpc_closure **original_closure) {
+  state->original_closure = *original_closure;
+  state->call_combiner = calld->call_combiner;
+  state->reason = reason;
+  *original_closure = GRPC_CLOSURE_INIT(
+      &state->closure,
+      free_when_done ? run_cancel_in_call_combiner : run_in_call_combiner,
+      state, grpc_schedule_on_exec_ctx);
+}
+
+static callback_state *get_state_for_batch(
+    call_data *calld, grpc_transport_stream_op_batch *batch) {
+  if (batch->send_initial_metadata) return &calld->on_complete[0];
+  if (batch->send_message) return &calld->on_complete[1];
+  if (batch->send_trailing_metadata) return &calld->on_complete[2];
+  if (batch->recv_initial_metadata) return &calld->on_complete[3];
+  if (batch->recv_message) return &calld->on_complete[4];
+  if (batch->recv_trailing_metadata) return &calld->on_complete[5];
+  GPR_UNREACHABLE_CODE(return NULL);
+}
 
 /* We perform a small hack to locate transport data alongside the connected
    channel data in call allocations, to allow everything to be pulled in minimal
@@ -49,13 +99,38 @@ typedef struct connected_channel_call_data { void *unused; } call_data;
    into transport stream operations */
 static void con_start_transport_stream_op_batch(
     grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
-    grpc_transport_stream_op_batch *op) {
+    grpc_transport_stream_op_batch *batch) {
   call_data *calld = elem->call_data;
   channel_data *chand = elem->channel_data;
-  GRPC_CALL_LOG_OP(GPR_INFO, elem, op);
-
+  if (batch->recv_initial_metadata) {
+    callback_state *state = &calld->recv_initial_metadata_ready;
+    intercept_callback(
+        calld, state, false, "recv_initial_metadata_ready",
+        &batch->payload->recv_initial_metadata.recv_initial_metadata_ready);
+  }
+  if (batch->recv_message) {
+    callback_state *state = &calld->recv_message_ready;
+    intercept_callback(calld, state, false, "recv_message_ready",
+                       &batch->payload->recv_message.recv_message_ready);
+  }
+  if (batch->cancel_stream) {
+    // There can be more than one cancellation batch in flight at any
+    // given time, so we can't just pick out a fixed index into
+    // calld->on_complete like we can for the other ops.  However,
+    // cancellation isn't in the fast path, so we just allocate a new
+    // closure for each one.
+    callback_state *state = (callback_state *)gpr_malloc(sizeof(*state));
+    intercept_callback(calld, state, true, "on_complete (cancel_stream)",
+                       &batch->on_complete);
+  } else {
+    callback_state *state = get_state_for_batch(calld, batch);
+    intercept_callback(calld, state, false, "on_complete", &batch->on_complete);
+  }
   grpc_transport_perform_stream_op(exec_ctx, chand->transport,
-                                   TRANSPORT_STREAM_FROM_CALL_DATA(calld), op);
+                                   TRANSPORT_STREAM_FROM_CALL_DATA(calld),
+                                   batch);
+  GRPC_CALL_COMBINER_STOP(exec_ctx, calld->call_combiner,
+                          "passed batch to transport");
 }
 
 static void con_start_transport_op(grpc_exec_ctx *exec_ctx,
@@ -71,6 +146,7 @@ static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx,
                                   const grpc_call_element_args *args) {
   call_data *calld = elem->call_data;
   channel_data *chand = elem->channel_data;
+  calld->call_combiner = args->call_combiner;
   int r = grpc_transport_init_stream(
       exec_ctx, chand->transport, TRANSPORT_STREAM_FROM_CALL_DATA(calld),
       &args->call_stack->refcount, args->server_transport_data, args->arena);
@@ -118,11 +194,6 @@ static void destroy_channel_elem(grpc_exec_ctx *exec_ctx,
   }
 }
 
-static char *con_get_peer(grpc_exec_ctx *exec_ctx, grpc_call_element *elem) {
-  channel_data *chand = elem->channel_data;
-  return grpc_transport_get_peer(exec_ctx, chand->transport);
-}
-
 /* No-op. */
 static void con_get_channel_info(grpc_exec_ctx *exec_ctx,
                                  grpc_channel_element *elem,
@@ -138,7 +209,6 @@ const grpc_channel_filter grpc_connected_filter = {
     sizeof(channel_data),
     init_channel_elem,
     destroy_channel_elem,
-    con_get_peer,
     con_get_channel_info,
     "connected",
 };