Merge branch 'stats' into stats_histo

1 files changed, 373 insertions, 90 deletions

diff --git a/src/core/lib/surface/call.c b/src/core/lib/surface/call.c
index 2f42726c7f..03eaaf99ac 100644
--- a/src/core/lib/surface/call.c
+++ b/src/core/lib/surface/call.c
@@ -122,6 +122,7 @@ typedef struct batch_control {
       bool is_closure;
     } notify_tag;
   } completion_data;
+  grpc_closure start_batch;
   grpc_closure finish_batch;
   gpr_refcount steps_to_complete;
 
@@ -145,9 +146,13 @@ typedef struct {
   grpc_call *sibling_prev;
 } child_call;
 
+#define RECV_NONE ((gpr_atm)0)
+#define RECV_INITIAL_METADATA_FIRST ((gpr_atm)1)
+
 struct grpc_call {
   gpr_refcount ext_ref;
   gpr_arena *arena;
+  grpc_call_combiner call_combiner;
   grpc_completion_queue *cq;
   grpc_polling_entity pollent;
   grpc_channel *channel;
@@ -171,9 +176,6 @@ struct grpc_call {
   gpr_atm any_ops_sent_atm;
   gpr_atm received_final_op_atm;
 
-  /* have we received initial metadata */
-  bool has_initial_md_been_received;
-
   batch_control *active_batches[MAX_CONCURRENT_BATCHES];
   grpc_transport_stream_op_batch_payload stream_op_payload;
 
@@ -184,6 +186,11 @@ struct grpc_call {
      Element 0 is initial metadata, element 1 is trailing metadata. */
   grpc_metadata_array *buffered_metadata[2];
 
+  grpc_metadata compression_md;
+
+  // A char* indicating the peer name.
+  gpr_atm peer_string;
+
   /* Packed received call statuses from various sources */
   gpr_atm status[STATUS_SOURCE_COUNT];
 
@@ -193,8 +200,12 @@ struct grpc_call {
 
   /* Compression algorithm for *incoming* data */
   grpc_compression_algorithm incoming_compression_algorithm;
+  /* Stream compression algorithm for *incoming* data */
+  grpc_stream_compression_algorithm incoming_stream_compression_algorithm;
   /* Supported encodings (compression algorithms), a bitset */
   uint32_t encodings_accepted_by_peer;
+  /* Supported stream encodings (stream compression algorithms), a bitset */
+  uint32_t stream_encodings_accepted_by_peer;
 
   /* Contexts for various subsystems (security, tracing, ...). */
   grpc_call_context_element context[GRPC_CONTEXT_COUNT];
@@ -227,7 +238,23 @@ struct grpc_call {
     } server;
   } final_op;
 
-  void *saved_receiving_stream_ready_bctlp;
+  /* recv_state can contain one of the following values:
+     RECV_NONE :                 :  no initial metadata and messages received
+     RECV_INITIAL_METADATA_FIRST :  received initial metadata first
+     a batch_control*            :  received messages first
+
+                 +------1------RECV_NONE------3-----+
+                 |                                  |
+                 |                                  |
+                 v                                  v
+     RECV_INITIAL_METADATA_FIRST        receiving_stream_ready_bctlp
+           |           ^                      |           ^
+           |           |                      |           |
+           +-----2-----+                      +-----4-----+
+
+    For 1, 4: See receiving_initial_metadata_ready() function
+    For 2, 3: See receiving_stream_ready() function */
+  gpr_atm recv_state;
 };
 
 grpc_tracer_flag grpc_call_error_trace =
@@ -242,8 +269,9 @@ grpc_tracer_flag grpc_compression_trace =
 #define CALL_FROM_TOP_ELEM(top_elem) \
   CALL_FROM_CALL_STACK(grpc_call_stack_from_top_element(top_elem))
 
-static void execute_op(grpc_exec_ctx *exec_ctx, grpc_call *call,
-                       grpc_transport_stream_op_batch *op);
+static void execute_batch(grpc_exec_ctx *exec_ctx, grpc_call *call,
+                          grpc_transport_stream_op_batch *op,
+                          grpc_closure *start_batch_closure);
 static void cancel_with_status(grpc_exec_ctx *exec_ctx, grpc_call *c,
                                status_source source, grpc_status_code status,
                                const char *description);
@@ -308,6 +336,7 @@ grpc_error *grpc_call_create(grpc_exec_ctx *exec_ctx,
                          sizeof(grpc_call) + channel_stack->call_stack_size);
   gpr_ref_init(&call->ext_ref, 1);
   call->arena = arena;
+  grpc_call_combiner_init(&call->call_combiner);
   *out_call = call;
   call->channel = args->channel;
   call->cq = args->cq;
@@ -416,7 +445,8 @@ grpc_error *grpc_call_create(grpc_exec_ctx *exec_ctx,
       .path = path,
       .start_time = call->start_time,
       .deadline = send_deadline,
-      .arena = call->arena};
+      .arena = call->arena,
+      .call_combiner = &call->call_combiner};
   add_init_error(&error, grpc_call_stack_init(exec_ctx, channel_stack, 1,
                                               destroy_call, call, &call_args));
   if (error != GRPC_ERROR_NONE) {
@@ -483,6 +513,8 @@ static void release_call(grpc_exec_ctx *exec_ctx, void *call,
                          grpc_error *error) {
   grpc_call *c = call;
   grpc_channel *channel = c->channel;
+  grpc_call_combiner_destroy(&c->call_combiner);
+  gpr_free((char *)c->peer_string);
   grpc_channel_update_call_size_estimate(channel, gpr_arena_destroy(c->arena));
   GRPC_CHANNEL_INTERNAL_UNREF(exec_ctx, channel, "call");
 }
@@ -582,30 +614,37 @@ grpc_call_error grpc_call_cancel(grpc_call *call, void *reserved) {
   return GRPC_CALL_OK;
 }
 
-static void execute_op(grpc_exec_ctx *exec_ctx, grpc_call *call,
-                       grpc_transport_stream_op_batch *op) {
-  grpc_call_element *elem;
-
-  GPR_TIMER_BEGIN("execute_op", 0);
-  elem = CALL_ELEM_FROM_CALL(call, 0);
-  elem->filter->start_transport_stream_op_batch(exec_ctx, elem, op);
-  GPR_TIMER_END("execute_op", 0);
+// This is called via the call combiner to start sending a batch down
+// the filter stack.
+static void execute_batch_in_call_combiner(grpc_exec_ctx *exec_ctx, void *arg,
+                                           grpc_error *ignored) {
+  grpc_transport_stream_op_batch *batch = arg;
+  grpc_call *call = batch->handler_private.extra_arg;
+  GPR_TIMER_BEGIN("execute_batch", 0);
+  grpc_call_element *elem = CALL_ELEM_FROM_CALL(call, 0);
+  GRPC_CALL_LOG_OP(GPR_INFO, elem, batch);
+  elem->filter->start_transport_stream_op_batch(exec_ctx, elem, batch);
+  GPR_TIMER_END("execute_batch", 0);
+}
+
+// start_batch_closure points to a caller-allocated closure to be used
+// for entering the call combiner.
+static void execute_batch(grpc_exec_ctx *exec_ctx, grpc_call *call,
+                          grpc_transport_stream_op_batch *batch,
+                          grpc_closure *start_batch_closure) {
+  batch->handler_private.extra_arg = call;
+  GRPC_CLOSURE_INIT(start_batch_closure, execute_batch_in_call_combiner, batch,
+                    grpc_schedule_on_exec_ctx);
+  GRPC_CALL_COMBINER_START(exec_ctx, &call->call_combiner, start_batch_closure,
+                           GRPC_ERROR_NONE, "executing batch");
 }
 
 char *grpc_call_get_peer(grpc_call *call) {
-  grpc_call_element *elem = CALL_ELEM_FROM_CALL(call, 0);
-  grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
-  char *result;
-  GRPC_API_TRACE("grpc_call_get_peer(%p)", 1, (call));
-  result = elem->filter->get_peer(&exec_ctx, elem);
-  if (result == NULL) {
-    result = grpc_channel_get_target(call->channel);
-  }
-  if (result == NULL) {
-    result = gpr_strdup("unknown");
-  }
-  grpc_exec_ctx_finish(&exec_ctx);
-  return result;
+  char *peer_string = (char *)gpr_atm_acq_load(&call->peer_string);
+  if (peer_string != NULL) return gpr_strdup(peer_string);
+  peer_string = grpc_channel_get_target(call->channel);
+  if (peer_string != NULL) return peer_string;
+  return gpr_strdup("unknown");
 }
 
 grpc_call *grpc_call_from_top_element(grpc_call_element *elem) {
@@ -632,24 +671,47 @@ grpc_call_error grpc_call_cancel_with_status(grpc_call *c,
   return GRPC_CALL_OK;
 }
 
-static void done_termination(grpc_exec_ctx *exec_ctx, void *call,
+typedef struct {
+  grpc_call *call;
+  grpc_closure start_batch;
+  grpc_closure finish_batch;
+} cancel_state;
+
+// The on_complete callback used when sending a cancel_stream batch down
+// the filter stack.  Yields the call combiner when the batch is done.
+static void done_termination(grpc_exec_ctx *exec_ctx, void *arg,
                              grpc_error *error) {
-  GRPC_CALL_INTERNAL_UNREF(exec_ctx, call, "termination");
+  cancel_state *state = (cancel_state *)arg;
+  GRPC_CALL_COMBINER_STOP(exec_ctx, &state->call->call_combiner,
+                          "on_complete for cancel_stream op");
+  GRPC_CALL_INTERNAL_UNREF(exec_ctx, state->call, "termination");
+  gpr_free(state);
 }
 
 static void cancel_with_error(grpc_exec_ctx *exec_ctx, grpc_call *c,
                               status_source source, grpc_error *error) {
   GRPC_CALL_INTERNAL_REF(c, "termination");
+  // Inform the call combiner of the cancellation, so that it can cancel
+  // any in-flight asynchronous actions that may be holding the call
+  // combiner.  This ensures that the cancel_stream batch can be sent
+  // down the filter stack in a timely manner.
+  grpc_call_combiner_cancel(exec_ctx, &c->call_combiner, GRPC_ERROR_REF(error));
   set_status_from_error(exec_ctx, c, source, GRPC_ERROR_REF(error));
-  grpc_transport_stream_op_batch *op = grpc_make_transport_stream_op(
-      GRPC_CLOSURE_CREATE(done_termination, c, grpc_schedule_on_exec_ctx));
+  cancel_state *state = (cancel_state *)gpr_malloc(sizeof(*state));
+  state->call = c;
+  GRPC_CLOSURE_INIT(&state->finish_batch, done_termination, state,
+                    grpc_schedule_on_exec_ctx);
+  grpc_transport_stream_op_batch *op =
+      grpc_make_transport_stream_op(&state->finish_batch);
   op->cancel_stream = true;
   op->payload->cancel_stream.cancel_error = error;
-  execute_op(exec_ctx, c, op);
+  execute_batch(exec_ctx, c, op, &state->start_batch);
 }
 
 static grpc_error *error_from_status(grpc_status_code status,
                                      const char *description) {
+  // copying 'description' is needed to ensure the grpc_call_cancel_with_status
+  // guarantee that can be short-lived.
   return grpc_error_set_int(
       grpc_error_set_str(GRPC_ERROR_CREATE_FROM_COPIED_STRING(description),
                          GRPC_ERROR_STR_GRPC_MESSAGE,
@@ -756,6 +818,12 @@ static void set_incoming_compression_algorithm(
   call->incoming_compression_algorithm = algo;
 }
 
+static void set_incoming_stream_compression_algorithm(
+    grpc_call *call, grpc_stream_compression_algorithm algo) {
+  GPR_ASSERT(algo < GRPC_STREAM_COMPRESS_ALGORITHMS_COUNT);
+  call->incoming_stream_compression_algorithm = algo;
+}
+
 grpc_compression_algorithm grpc_call_test_only_get_compression_algorithm(
     grpc_call *call) {
   grpc_compression_algorithm algorithm;
@@ -769,6 +837,13 @@ static grpc_compression_algorithm compression_algorithm_for_level_locked(
                                               call->encodings_accepted_by_peer);
 }
 
+static grpc_stream_compression_algorithm
+stream_compression_algorithm_for_level_locked(
+    grpc_call *call, grpc_stream_compression_level level) {
+  return grpc_stream_compression_algorithm_for_level(
+      level, call->stream_encodings_accepted_by_peer);
+}
+
 uint32_t grpc_call_test_only_get_message_flags(grpc_call *call) {
   uint32_t flags;
   flags = call->test_only_last_message_flags;
@@ -823,13 +898,71 @@ static void set_encodings_accepted_by_peer(grpc_exec_ctx *exec_ctx,
       (void *)(((uintptr_t)call->encodings_accepted_by_peer) + 1));
 }
 
+static void set_stream_encodings_accepted_by_peer(grpc_exec_ctx *exec_ctx,
+                                                  grpc_call *call,
+                                                  grpc_mdelem mdel) {
+  size_t i;
+  grpc_stream_compression_algorithm algorithm;
+  grpc_slice_buffer accept_encoding_parts;
+  grpc_slice accept_encoding_slice;
+  void *accepted_user_data;
+
+  accepted_user_data =
+      grpc_mdelem_get_user_data(mdel, destroy_encodings_accepted_by_peer);
+  if (accepted_user_data != NULL) {
+    call->stream_encodings_accepted_by_peer =
+        (uint32_t)(((uintptr_t)accepted_user_data) - 1);
+    return;
+  }
+
+  accept_encoding_slice = GRPC_MDVALUE(mdel);
+  grpc_slice_buffer_init(&accept_encoding_parts);
+  grpc_slice_split(accept_encoding_slice, ",", &accept_encoding_parts);
+
+  /* Always support no compression */
+  GPR_BITSET(&call->stream_encodings_accepted_by_peer,
+             GRPC_STREAM_COMPRESS_NONE);
+  for (i = 0; i < accept_encoding_parts.count; i++) {
+    grpc_slice accept_encoding_entry_slice = accept_encoding_parts.slices[i];
+    if (grpc_stream_compression_algorithm_parse(accept_encoding_entry_slice,
+                                                &algorithm)) {
+      GPR_BITSET(&call->stream_encodings_accepted_by_peer, algorithm);
+    } else {
+      char *accept_encoding_entry_str =
+          grpc_slice_to_c_string(accept_encoding_entry_slice);
+      gpr_log(GPR_ERROR,
+              "Invalid entry in accept encoding metadata: '%s'. Ignoring.",
+              accept_encoding_entry_str);
+      gpr_free(accept_encoding_entry_str);
+    }
+  }
+
+  grpc_slice_buffer_destroy_internal(exec_ctx, &accept_encoding_parts);
+
+  grpc_mdelem_set_user_data(
+      mdel, destroy_encodings_accepted_by_peer,
+      (void *)(((uintptr_t)call->stream_encodings_accepted_by_peer) + 1));
+}
+
 uint32_t grpc_call_test_only_get_encodings_accepted_by_peer(grpc_call *call) {
   uint32_t encodings_accepted_by_peer;
   encodings_accepted_by_peer = call->encodings_accepted_by_peer;
   return encodings_accepted_by_peer;
 }
 
-static grpc_linked_mdelem *linked_from_md(grpc_metadata *md) {
+uint32_t grpc_call_test_only_get_stream_encodings_accepted_by_peer(
+    grpc_call *call) {
+  uint32_t stream_encodings_accepted_by_peer;
+  stream_encodings_accepted_by_peer = call->stream_encodings_accepted_by_peer;
+  return stream_encodings_accepted_by_peer;
+}
+
+grpc_stream_compression_algorithm
+grpc_call_test_only_get_incoming_stream_encodings(grpc_call *call) {
+  return call->incoming_stream_compression_algorithm;
+}
+
+static grpc_linked_mdelem *linked_from_md(const grpc_metadata *md) {
   return (grpc_linked_mdelem *)&md->internal_data;
 }
 
@@ -853,7 +986,7 @@ static int prepare_application_metadata(
   for (i = 0; i < total_count; i++) {
     const grpc_metadata *md =
         get_md_elem(metadata, additional_metadata, i, count);
-    grpc_linked_mdelem *l = (grpc_linked_mdelem *)&md->internal_data;
+    grpc_linked_mdelem *l = linked_from_md(md);
     GPR_ASSERT(sizeof(grpc_linked_mdelem) == sizeof(md->internal_data));
     if (!GRPC_LOG_IF_ERROR("validate_metadata",
                            grpc_validate_header_key_is_legal(md->key))) {
@@ -870,7 +1003,7 @@ static int prepare_application_metadata(
     for (int j = 0; j < i; j++) {
       const grpc_metadata *md =
           get_md_elem(metadata, additional_metadata, j, count);
-      grpc_linked_mdelem *l = (grpc_linked_mdelem *)&md->internal_data;
+      grpc_linked_mdelem *l = linked_from_md(md);
       GRPC_MDELEM_UNREF(exec_ctx, l->md);
     }
     return 0;
@@ -888,9 +1021,12 @@ static int prepare_application_metadata(
   }
   for (i = 0; i < total_count; i++) {
     grpc_metadata *md = get_md_elem(metadata, additional_metadata, i, count);
-    GRPC_LOG_IF_ERROR(
-        "prepare_application_metadata",
-        grpc_metadata_batch_link_tail(exec_ctx, batch, linked_from_md(md)));
+    grpc_linked_mdelem *l = linked_from_md(md);
+    grpc_error *error = grpc_metadata_batch_link_tail(exec_ctx, batch, l);
+    if (error != GRPC_ERROR_NONE) {
+      GRPC_MDELEM_UNREF(exec_ctx, l->md);
+    }
+    GRPC_LOG_IF_ERROR("prepare_application_metadata", error);
   }
   call->send_extra_metadata_count = 0;
 
@@ -937,6 +1073,22 @@ static grpc_compression_algorithm decode_compression(grpc_mdelem md) {
   return algorithm;
 }
 
+static grpc_stream_compression_algorithm decode_stream_compression(
+    grpc_mdelem md) {
+  grpc_stream_compression_algorithm algorithm =
+      grpc_stream_compression_algorithm_from_slice(GRPC_MDVALUE(md));
+  if (algorithm == GRPC_STREAM_COMPRESS_ALGORITHMS_COUNT) {
+    char *md_c_str = grpc_slice_to_c_string(GRPC_MDVALUE(md));
+    gpr_log(GPR_ERROR,
+            "Invalid incoming stream compression algorithm: '%s'. Interpreting "
+            "incoming data as uncompressed.",
+            md_c_str);
+    gpr_free(md_c_str);
+    return GRPC_STREAM_COMPRESS_NONE;
+  }
+  return algorithm;
+}
+
 static void publish_app_metadata(grpc_call *call, grpc_metadata_batch *b,
                                  int is_trailing) {
   if (b->list.count == 0) return;
@@ -961,7 +1113,19 @@ static void publish_app_metadata(grpc_call *call, grpc_metadata_batch *b,
 
 static void recv_initial_filter(grpc_exec_ctx *exec_ctx, grpc_call *call,
                                 grpc_metadata_batch *b) {
-  if (b->idx.named.grpc_encoding != NULL) {
+  if (b->idx.named.content_encoding != NULL) {
+    if (b->idx.named.grpc_encoding != NULL) {
+      gpr_log(GPR_ERROR,
+              "Received both content-encoding and grpc-encoding header. "
+              "Ignoring grpc-encoding.");
+      grpc_metadata_batch_remove(exec_ctx, b, b->idx.named.grpc_encoding);
+    }
+    GPR_TIMER_BEGIN("incoming_stream_compression_algorithm", 0);
+    set_incoming_stream_compression_algorithm(
+        call, decode_stream_compression(b->idx.named.content_encoding->md));
+    GPR_TIMER_END("incoming_stream_compression_algorithm", 0);
+    grpc_metadata_batch_remove(exec_ctx, b, b->idx.named.content_encoding);
+  } else if (b->idx.named.grpc_encoding != NULL) {
     GPR_TIMER_BEGIN("incoming_compression_algorithm", 0);
     set_incoming_compression_algorithm(
         call, decode_compression(b->idx.named.grpc_encoding->md));
@@ -975,6 +1139,13 @@ static void recv_initial_filter(grpc_exec_ctx *exec_ctx, grpc_call *call,
     grpc_metadata_batch_remove(exec_ctx, b, b->idx.named.grpc_accept_encoding);
     GPR_TIMER_END("encodings_accepted_by_peer", 0);
   }
+  if (b->idx.named.accept_encoding != NULL) {
+    GPR_TIMER_BEGIN("stream_encodings_accepted_by_peer", 0);
+    set_stream_encodings_accepted_by_peer(exec_ctx, call,
+                                          b->idx.named.accept_encoding->md);
+    grpc_metadata_batch_remove(exec_ctx, b, b->idx.named.accept_encoding);
+    GPR_TIMER_END("stream_encodings_accepted_by_peer", 0);
+  }
   publish_app_metadata(call, b, false);
 }
 
@@ -1291,19 +1462,73 @@ static void receiving_stream_ready(grpc_exec_ctx *exec_ctx, void *bctlp,
     cancel_with_error(exec_ctx, call, STATUS_FROM_SURFACE,
                       GRPC_ERROR_REF(error));
   }
-  if (call->has_initial_md_been_received || error != GRPC_ERROR_NONE ||
-      call->receiving_stream == NULL) {
+  /* If recv_state is RECV_NONE, we will save the batch_control
+   * object with rel_cas, and will not use it after the cas. Its corresponding
+   * acq_load is in receiving_initial_metadata_ready() */
+  if (error != GRPC_ERROR_NONE || call->receiving_stream == NULL ||
+      !gpr_atm_rel_cas(&call->recv_state, RECV_NONE, (gpr_atm)bctlp)) {
     process_data_after_md(exec_ctx, bctlp);
-  } else {
-    call->saved_receiving_stream_ready_bctlp = bctlp;
   }
 }
 
+// The recv_message_ready callback used when sending a batch containing
+// a recv_message op down the filter stack.  Yields the call combiner
+// before processing the received message.
+static void receiving_stream_ready_in_call_combiner(grpc_exec_ctx *exec_ctx,
+                                                    void *bctlp,
+                                                    grpc_error *error) {
+  batch_control *bctl = bctlp;
+  grpc_call *call = bctl->call;
+  GRPC_CALL_COMBINER_STOP(exec_ctx, &call->call_combiner, "recv_message_ready");
+  receiving_stream_ready(exec_ctx, bctlp, error);
+}
+
 static void validate_filtered_metadata(grpc_exec_ctx *exec_ctx,
                                        batch_control *bctl) {
   grpc_call *call = bctl->call;
-  /* validate call->incoming_compression_algorithm */
-  if (call->incoming_compression_algorithm != GRPC_COMPRESS_NONE) {
+  /* validate compression algorithms */
+  if (call->incoming_stream_compression_algorithm !=
+      GRPC_STREAM_COMPRESS_NONE) {
+    const grpc_stream_compression_algorithm algo =
+        call->incoming_stream_compression_algorithm;
+    char *error_msg = NULL;
+    const grpc_compression_options compression_options =
+        grpc_channel_compression_options(call->channel);
+    if (algo >= GRPC_STREAM_COMPRESS_ALGORITHMS_COUNT) {
+      gpr_asprintf(&error_msg,
+                   "Invalid stream compression algorithm value '%d'.", algo);
+      gpr_log(GPR_ERROR, "%s", error_msg);
+      cancel_with_status(exec_ctx, call, STATUS_FROM_SURFACE,
+                         GRPC_STATUS_UNIMPLEMENTED, error_msg);
+    } else if (grpc_compression_options_is_stream_compression_algorithm_enabled(
+                   &compression_options, algo) == 0) {
+      /* check if algorithm is supported by current channel config */
+      char *algo_name = NULL;
+      grpc_stream_compression_algorithm_name(algo, &algo_name);
+      gpr_asprintf(&error_msg, "Stream compression algorithm '%s' is disabled.",
+                   algo_name);
+      gpr_log(GPR_ERROR, "%s", error_msg);
+      cancel_with_status(exec_ctx, call, STATUS_FROM_SURFACE,
+                         GRPC_STATUS_UNIMPLEMENTED, error_msg);
+    }
+    gpr_free(error_msg);
+
+    GPR_ASSERT(call->stream_encodings_accepted_by_peer != 0);
+    if (!GPR_BITGET(call->stream_encodings_accepted_by_peer,
+                    call->incoming_stream_compression_algorithm)) {
+      if (GRPC_TRACER_ON(grpc_compression_trace)) {
+        char *algo_name = NULL;
+        grpc_stream_compression_algorithm_name(
+            call->incoming_stream_compression_algorithm, &algo_name);
+        gpr_log(
+            GPR_ERROR,
+            "Stream compression algorithm (content-encoding = '%s') not "
+            "present in the bitset of accepted encodings (accept-encodings: "
+            "'0x%x')",
+            algo_name, call->stream_encodings_accepted_by_peer);
+      }
+    }
+  } else if (call->incoming_compression_algorithm != GRPC_COMPRESS_NONE) {
     const grpc_compression_algorithm algo =
         call->incoming_compression_algorithm;
     char *error_msg = NULL;
@@ -1330,22 +1555,20 @@ static void validate_filtered_metadata(grpc_exec_ctx *exec_ctx,
       call->incoming_compression_algorithm = algo;
     }
     gpr_free(error_msg);
-  }
 
-  /* make sure the received grpc-encoding is amongst the ones listed in
-   * grpc-accept-encoding */
-  GPR_ASSERT(call->encodings_accepted_by_peer != 0);
-  if (!GPR_BITGET(call->encodings_accepted_by_peer,
-                  call->incoming_compression_algorithm)) {
-    if (GRPC_TRACER_ON(grpc_compression_trace)) {
-      char *algo_name = NULL;
-      grpc_compression_algorithm_name(call->incoming_compression_algorithm,
-                                      &algo_name);
-      gpr_log(GPR_ERROR,
-              "Compression algorithm (grpc-encoding = '%s') not present in "
-              "the bitset of accepted encodings (grpc-accept-encodings: "
-              "'0x%x')",
-              algo_name, call->encodings_accepted_by_peer);
+    GPR_ASSERT(call->encodings_accepted_by_peer != 0);
+    if (!GPR_BITGET(call->encodings_accepted_by_peer,
+                    call->incoming_compression_algorithm)) {
+      if (GRPC_TRACER_ON(grpc_compression_trace)) {
+        char *algo_name = NULL;
+        grpc_compression_algorithm_name(call->incoming_compression_algorithm,
+                                        &algo_name);
+        gpr_log(GPR_ERROR,
+                "Compression algorithm (grpc-encoding = '%s') not present in "
+                "the bitset of accepted encodings (grpc-accept-encodings: "
+                "'0x%x')",
+                algo_name, call->encodings_accepted_by_peer);
+      }
     }
   }
 }
@@ -1366,6 +1589,9 @@ static void receiving_initial_metadata_ready(grpc_exec_ctx *exec_ctx,
   batch_control *bctl = bctlp;
   grpc_call *call = bctl->call;
 
+  GRPC_CALL_COMBINER_STOP(exec_ctx, &call->call_combiner,
+                          "recv_initial_metadata_ready");
+
   add_batch_error(exec_ctx, bctl, GRPC_ERROR_REF(error), false);
   if (error == GRPC_ERROR_NONE) {
     grpc_metadata_batch *md =
@@ -1385,12 +1611,31 @@ static void receiving_initial_metadata_ready(grpc_exec_ctx *exec_ctx,
     }
   }
 
-  call->has_initial_md_been_received = true;
-  if (call->saved_receiving_stream_ready_bctlp != NULL) {
-    grpc_closure *saved_rsr_closure = GRPC_CLOSURE_CREATE(
-        receiving_stream_ready, call->saved_receiving_stream_ready_bctlp,
-        grpc_schedule_on_exec_ctx);
-    call->saved_receiving_stream_ready_bctlp = NULL;
+  grpc_closure *saved_rsr_closure = NULL;
+  while (true) {
+    gpr_atm rsr_bctlp = gpr_atm_acq_load(&call->recv_state);
+    /* Should only receive initial metadata once */
+    GPR_ASSERT(rsr_bctlp != 1);
+    if (rsr_bctlp == 0) {
+      /* We haven't seen initial metadata and messages before, thus initial
+       * metadata is received first.
+       * no_barrier_cas is used, as this function won't access the batch_control
+       * object saved by receiving_stream_ready() if the initial metadata is
+       * received first. */
+      if (gpr_atm_no_barrier_cas(&call->recv_state, RECV_NONE,
+                                 RECV_INITIAL_METADATA_FIRST)) {
+        break;
+      }
+    } else {
+      /* Already received messages */
+      saved_rsr_closure = GRPC_CLOSURE_CREATE(receiving_stream_ready,
+                                              (batch_control *)rsr_bctlp,
+                                              grpc_schedule_on_exec_ctx);
+      /* No need to modify recv_state */
+      break;
+    }
+  }
+  if (saved_rsr_closure != NULL) {
     GRPC_CLOSURE_RUN(exec_ctx, saved_rsr_closure, GRPC_ERROR_REF(error));
   }
 
@@ -1400,7 +1645,8 @@ static void receiving_initial_metadata_ready(grpc_exec_ctx *exec_ctx,
 static void finish_batch(grpc_exec_ctx *exec_ctx, void *bctlp,
                          grpc_error *error) {
   batch_control *bctl = bctlp;
-
+  grpc_call *call = bctl->call;
+  GRPC_CALL_COMBINER_STOP(exec_ctx, &call->call_combiner, "on_complete");
   add_batch_error(exec_ctx, bctl, GRPC_ERROR_REF(error), false);
   finish_batch_step(exec_ctx, bctl);
 }
@@ -1420,15 +1666,12 @@ static grpc_call_error call_start_batch(grpc_exec_ctx *exec_ctx,
   int num_completion_callbacks_needed = 1;
   grpc_call_error error = GRPC_CALL_OK;
 
-  // sent_initial_metadata guards against variable reuse.
-  grpc_metadata compression_md;
-
   GPR_TIMER_BEGIN("grpc_call_start_batch", 0);
   GRPC_CALL_LOG_BATCH(GPR_INFO, call, ops, nops, notify_tag);
 
   if (nops == 0) {
     if (!is_notify_tag_closure) {
-      grpc_cq_begin_op(call->cq, notify_tag);
+      GPR_ASSERT(grpc_cq_begin_op(call->cq, notify_tag));
       grpc_cq_end_op(exec_ctx, call->cq, notify_tag, GRPC_ERROR_NONE,
                      free_no_op_completion, NULL,
                      gpr_malloc(sizeof(grpc_cq_completion)));
@@ -1470,31 +1713,60 @@ static grpc_call_error call_start_batch(grpc_exec_ctx *exec_ctx,
           goto done_with_error;
         }
         /* process compression level */
-        memset(&compression_md, 0, sizeof(compression_md));
+        memset(&call->compression_md, 0, sizeof(call->compression_md));
         size_t additional_metadata_count = 0;
-        grpc_compression_level effective_compression_level;
+        grpc_compression_level effective_compression_level =
+            GRPC_COMPRESS_LEVEL_NONE;
+        grpc_stream_compression_level effective_stream_compression_level =
+            GRPC_STREAM_COMPRESS_LEVEL_NONE;
         bool level_set = false;
-        if (op->data.send_initial_metadata.maybe_compression_level.is_set) {
+        bool stream_compression = false;
+        if (op->data.send_initial_metadata.maybe_stream_compression_level
+                .is_set) {
+          effective_stream_compression_level =
+              op->data.send_initial_metadata.maybe_stream_compression_level
+                  .level;
+          level_set = true;
+          stream_compression = true;
+        } else if (op->data.send_initial_metadata.maybe_compression_level
+                       .is_set) {
           effective_compression_level =
               op->data.send_initial_metadata.maybe_compression_level.level;
           level_set = true;
         } else {
           const grpc_compression_options copts =
               grpc_channel_compression_options(call->channel);
-          level_set = copts.default_level.is_set;
-          if (level_set) {
+          if (copts.default_stream_compression_level.is_set) {
+            level_set = true;
+            effective_stream_compression_level =
+                copts.default_stream_compression_level.level;
+            stream_compression = true;
+          } else if (copts.default_level.is_set) {
+            level_set = true;
             effective_compression_level = copts.default_level.level;
           }
         }
         if (level_set && !call->is_client) {
-          const grpc_compression_algorithm calgo =
-              compression_algorithm_for_level_locked(
-                  call, effective_compression_level);
-          // the following will be picked up by the compress filter and used as
-          // the call's compression algorithm.
-          compression_md.key = GRPC_MDSTR_GRPC_INTERNAL_ENCODING_REQUEST;
-          compression_md.value = grpc_compression_algorithm_slice(calgo);
-          additional_metadata_count++;
+          if (stream_compression) {
+            const grpc_stream_compression_algorithm calgo =
+                stream_compression_algorithm_for_level_locked(
+                    call, effective_stream_compression_level);
+            call->compression_md.key =
+                GRPC_MDSTR_GRPC_INTERNAL_STREAM_ENCODING_REQUEST;
+            call->compression_md.value =
+                grpc_stream_compression_algorithm_slice(calgo);
+          } else {
+            const grpc_compression_algorithm calgo =
+                compression_algorithm_for_level_locked(
+                    call, effective_compression_level);
+            /* the following will be picked up by the compress filter and used
+             * as the call's compression algorithm. */
+            call->compression_md.key =
+                GRPC_MDSTR_GRPC_INTERNAL_ENCODING_REQUEST;
+            call->compression_md.value =
+                grpc_compression_algorithm_slice(calgo);
+            additional_metadata_count++;
+          }
         }
 
         if (op->data.send_initial_metadata.count + additional_metadata_count >
@@ -1507,7 +1779,7 @@ static grpc_call_error call_start_batch(grpc_exec_ctx *exec_ctx,
         if (!prepare_application_metadata(
                 exec_ctx, call, (int)op->data.send_initial_metadata.count,
                 op->data.send_initial_metadata.metadata, 0, call->is_client,
-                &compression_md, (int)additional_metadata_count)) {
+                &call->compression_md, (int)additional_metadata_count)) {
           error = GRPC_CALL_ERROR_INVALID_METADATA;
           goto done_with_error;
         }
@@ -1519,6 +1791,10 @@ static grpc_call_error call_start_batch(grpc_exec_ctx *exec_ctx,
             &call->metadata_batch[0 /* is_receiving */][0 /* is_trailing */];
         stream_op_payload->send_initial_metadata.send_initial_metadata_flags =
             op->flags;
+        if (call->is_client) {
+          stream_op_payload->send_initial_metadata.peer_string =
+              &call->peer_string;
+        }
         break;
       case GRPC_OP_SEND_MESSAGE:
         if (!are_write_flags_valid(op->flags)) {
@@ -1651,6 +1927,10 @@ static grpc_call_error call_start_batch(grpc_exec_ctx *exec_ctx,
             &call->metadata_batch[1 /* is_receiving */][0 /* is_trailing */];
         stream_op_payload->recv_initial_metadata.recv_initial_metadata_ready =
             &call->receiving_initial_metadata_ready;
+        if (!call->is_client) {
+          stream_op_payload->recv_initial_metadata.peer_string =
+              &call->peer_string;
+        }
         num_completion_callbacks_needed++;
         break;
       case GRPC_OP_RECV_MESSAGE:
@@ -1667,8 +1947,9 @@ static grpc_call_error call_start_batch(grpc_exec_ctx *exec_ctx,
         stream_op->recv_message = true;
         call->receiving_buffer = op->data.recv_message.recv_message;
         stream_op_payload->recv_message.recv_message = &call->receiving_stream;
-        GRPC_CLOSURE_INIT(&call->receiving_stream_ready, receiving_stream_ready,
-                          bctl, grpc_schedule_on_exec_ctx);
+        GRPC_CLOSURE_INIT(&call->receiving_stream_ready,
+                          receiving_stream_ready_in_call_combiner, bctl,
+                          grpc_schedule_on_exec_ctx);
         stream_op_payload->recv_message.recv_message_ready =
             &call->receiving_stream_ready;
         num_completion_callbacks_needed++;
@@ -1729,7 +2010,7 @@ static grpc_call_error call_start_batch(grpc_exec_ctx *exec_ctx,
 
   GRPC_CALL_INTERNAL_REF(call, "completion");
   if (!is_notify_tag_closure) {
-    grpc_cq_begin_op(call->cq, notify_tag);
+    GPR_ASSERT(grpc_cq_begin_op(call->cq, notify_tag));
   }
   gpr_ref_init(&bctl->steps_to_complete, num_completion_callbacks_needed);
 
@@ -1738,7 +2019,7 @@ static grpc_call_error call_start_batch(grpc_exec_ctx *exec_ctx,
   stream_op->on_complete = &bctl->finish_batch;
   gpr_atm_rel_store(&call->any_ops_sent_atm, 1);
 
-  execute_op(exec_ctx, call, stream_op);
+  execute_batch(exec_ctx, call, stream_op, &bctl->start_batch);
 
 done:
   GPR_TIMER_END("grpc_call_start_batch", 0);
@@ -1850,6 +2131,8 @@ const char *grpc_call_error_to_string(grpc_call_error error) {
       return "GRPC_CALL_ERROR_PAYLOAD_TYPE_MISMATCH";
     case GRPC_CALL_ERROR_TOO_MANY_OPERATIONS:
       return "GRPC_CALL_ERROR_TOO_MANY_OPERATIONS";
+    case GRPC_CALL_ERROR_COMPLETION_QUEUE_SHUTDOWN:
+      return "GRPC_CALL_ERROR_COMPLETION_QUEUE_SHUTDOWN";
     case GRPC_CALL_OK:
       return "GRPC_CALL_OK";
   }