Merge pull request #11758 from ctiller/write_completion

Write completion changes

doc/environment_variables.md

@@ -48,6 +48,7 @@ some configuration as environment variables that can be set.
   - compression - traces compression operations
   - connectivity_state - traces connectivity state changes to channels
   - channel_stack_builder - traces information about channel stacks being built
+  - executor - traces grpc's internal thread pool ('the executor')
   - http - traces state in the http2 transport engine
   - http1 - traces HTTP/1.x operations performed by gRPC
   - inproc - traces the in-process transport

include/grpc++/impl/codegen/call.h

@@ -169,6 +169,15 @@ class WriteOptions {
     return *this;
   }
 
+  /// Guarantee that all bytes have been written to the wire before completing
+  /// this write (usually writes are completed when they pass flow control)
+  inline WriteOptions& set_write_through() {
+    SetBit(GRPC_WRITE_THROUGH);
+    return *this;
+  }
+
+  inline bool is_write_through() const { return GetBit(GRPC_WRITE_THROUGH); }
+
   /// Get value for the flag indicating that this is the last message, and
   /// should be coalesced with trailing metadata.
   ///

include/grpc/impl/codegen/grpc_types.h

@@ -355,8 +355,11 @@ typedef enum grpc_call_error {
 /** Force compression to be disabled for a particular write
     (start_write/add_metadata). Illegal on invoke/accept. */
 #define GRPC_WRITE_NO_COMPRESS (0x00000002u)
+/** Force this message to be written to the socket before completing it */
+#define GRPC_WRITE_THROUGH (0x00000004u)
 /** Mask of all valid flags. */
-#define GRPC_WRITE_USED_MASK (GRPC_WRITE_BUFFER_HINT | GRPC_WRITE_NO_COMPRESS)
+#define GRPC_WRITE_USED_MASK \
+  (GRPC_WRITE_BUFFER_HINT | GRPC_WRITE_NO_COMPRESS | GRPC_WRITE_THROUGH)
 
 /** Initial metadata flags */
 /** Signal that the call is idempotent */
@@ -377,7 +380,7 @@ typedef enum grpc_call_error {
    GRPC_INITIAL_METADATA_WAIT_FOR_READY |                \
    GRPC_INITIAL_METADATA_CACHEABLE_REQUEST |             \
    GRPC_INITIAL_METADATA_WAIT_FOR_READY_EXPLICITLY_SET | \
-   GRPC_INITIAL_METADATA_CORKED)
+   GRPC_INITIAL_METADATA_CORKED | GRPC_WRITE_THROUGH)
 
 /** A single metadata element */
 typedef struct grpc_metadata {

include/grpc/impl/codegen/port_platform.h

@@ -409,4 +409,15 @@ typedef unsigned __int64 uint64_t;
 #define CENSUSAPI GRPCAPI
 #endif
 
+#ifndef GPR_ATTRIBUTE_NO_TSAN /* (1) */
+#if defined(__has_feature)
+#if __has_feature(thread_sanitizer)
+#define GPR_ATTRIBUTE_NO_TSAN __attribute__((no_sanitize("thread")))
+#endif                        /* __has_feature(thread_sanitizer) */
+#endif                        /* defined(__has_feature) */
+#ifndef GPR_ATTRIBUTE_NO_TSAN /* (2) */
+#define GPR_ATTRIBUTE_NO_TSAN
+#endif /* GPR_ATTRIBUTE_NO_TSAN (2) */
+#endif /* GPR_ATTRIBUTE_NO_TSAN (1) */
+
 #endif /* GRPC_IMPL_CODEGEN_PORT_PLATFORM_H */

src/core/ext/filters/client_channel/http_connect_handshaker.c

@@ -309,7 +309,7 @@ static void http_connect_handshaker_do_handshake(
   grpc_httpcli_request request;
   memset(&request, 0, sizeof(request));
   request.host = server_name;
-  request.http.method = "CONNECT";
+  request.http.method = (char*)"CONNECT";
   request.http.path = server_name;
   request.http.hdrs = headers;
   request.http.hdr_count = num_headers;

src/core/ext/transport/chttp2/transport/chttp2_transport.c

@@ -84,8 +84,6 @@ grpc_tracer_flag grpc_trace_chttp2_refcount =
     GRPC_TRACER_INITIALIZER(false, "chttp2_refcount");
 #endif
 
-static const grpc_transport_vtable vtable;
-
 /* forward declarations of various callbacks that we'll build closures around */
 static void write_action_begin_locked(grpc_exec_ctx *exec_ctx, void *t,
                                       grpc_error *error);
@@ -248,6 +246,8 @@ void grpc_chttp2_unref_transport(grpc_exec_ctx *exec_ctx,
 void grpc_chttp2_ref_transport(grpc_chttp2_transport *t) { gpr_ref(&t->refs); }
 #endif
 
+static const grpc_transport_vtable *get_vtable(void);
+
 static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
                            const grpc_channel_args *channel_args,
                            grpc_endpoint *ep, bool is_client) {
@@ -257,7 +257,7 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
   GPR_ASSERT(strlen(GRPC_CHTTP2_CLIENT_CONNECT_STRING) ==
              GRPC_CHTTP2_CLIENT_CONNECT_STRLEN);
 
-  t->base.vtable = &vtable;
+  t->base.vtable = get_vtable();
   t->ep = ep;
   /* one ref is for destroy */
   gpr_ref_init(&t->refs, 1);
@@ -557,11 +557,6 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
     }
   }
 
-  GRPC_CLOSURE_INIT(&t->write_action, write_action, t,
-                    t->opt_target == GRPC_CHTTP2_OPTIMIZE_FOR_THROUGHPUT
-                        ? grpc_executor_scheduler
-                        : grpc_schedule_on_exec_ctx);
-
   t->ping_state.pings_before_data_required =
       t->ping_policy.max_pings_without_data;
   t->ping_state.is_delayed_ping_timer_set = false;
@@ -799,7 +794,7 @@ static void destroy_stream(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
 
 grpc_chttp2_stream *grpc_chttp2_parsing_lookup_stream(grpc_chttp2_transport *t,
                                                       uint32_t id) {
-  return grpc_chttp2_stream_map_find(&t->stream_map, id);
+  return (grpc_chttp2_stream *)grpc_chttp2_stream_map_find(&t->stream_map, id);
 }
 
 grpc_chttp2_stream *grpc_chttp2_parsing_accept_stream(grpc_exec_ctx *exec_ctx,
@@ -858,6 +853,7 @@ void grpc_chttp2_initiate_write(grpc_exec_ctx *exec_ctx,
   switch (t->write_state) {
     case GRPC_CHTTP2_WRITE_STATE_IDLE:
       set_write_state(exec_ctx, t, GRPC_CHTTP2_WRITE_STATE_WRITING, reason);
+      t->is_first_write_in_batch = true;
       GRPC_CHTTP2_REF_TRANSPORT(t, "writing");
       GRPC_CLOSURE_SCHED(
           exec_ctx,
@@ -876,46 +872,94 @@ void grpc_chttp2_initiate_write(grpc_exec_ctx *exec_ctx,
   GPR_TIMER_END("grpc_chttp2_initiate_write", 0);
 }
 
-void grpc_chttp2_become_writable(
+void grpc_chttp2_become_writable(grpc_exec_ctx *exec_ctx,
-    grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, grpc_chttp2_stream *s,
+                                 grpc_chttp2_transport *t,
-    grpc_chttp2_stream_write_type stream_write_type, const char *reason) {
+                                 grpc_chttp2_stream *s,
+                                 bool also_initiate_write, const char *reason) {
   if (!t->closed && grpc_chttp2_list_add_writable_stream(t, s)) {
     GRPC_CHTTP2_STREAM_REF(s, "chttp2_writing:become");
   }
-  switch (stream_write_type) {
+  if (also_initiate_write) {
-    case GRPC_CHTTP2_STREAM_WRITE_PIGGYBACK:
+    grpc_chttp2_initiate_write(exec_ctx, t, reason);
-      break;
-    case GRPC_CHTTP2_STREAM_WRITE_INITIATE_COVERED:
-      grpc_chttp2_initiate_write(exec_ctx, t, reason);
-      break;
-    case GRPC_CHTTP2_STREAM_WRITE_INITIATE_UNCOVERED:
-      grpc_chttp2_initiate_write(exec_ctx, t, reason);
-      break;
   }
 }
 
+static grpc_closure_scheduler *write_scheduler(grpc_chttp2_transport *t,
+                                               bool early_results_scheduled,
+                                               bool partial_write) {
+  /* if it's not the first write in a batch, always offload to the executor:
+     we'll probably end up queuing against the kernel anyway, so we'll likely
+     get better latency overall if we switch writing work elsewhere and continue
+     with application work above */
+  if (!t->is_first_write_in_batch) {
+    return grpc_executor_scheduler(GRPC_EXECUTOR_SHORT);
+  }
+  /* equivalently, if it's a partial write, we *know* we're going to be taking a
+     thread jump to write it because of the above, may as well do so
+     immediately */
+  if (partial_write) {
+    return grpc_executor_scheduler(GRPC_EXECUTOR_SHORT);
+  }
+  switch (t->opt_target) {
+    case GRPC_CHTTP2_OPTIMIZE_FOR_THROUGHPUT:
+      /* executor gives us the largest probability of being able to batch a
+       * write with others on this transport */
+      return grpc_executor_scheduler(GRPC_EXECUTOR_SHORT);
+    case GRPC_CHTTP2_OPTIMIZE_FOR_LATENCY:
+      return grpc_schedule_on_exec_ctx;
+  }
+  GPR_UNREACHABLE_CODE(return NULL);
+}
+
+#define WRITE_STATE_TUPLE_TO_INT(p, i) (2 * (int)(p) + (int)(i))
+static const char *begin_writing_desc(bool partial, bool inlined) {
+  switch (WRITE_STATE_TUPLE_TO_INT(partial, inlined)) {
+    case WRITE_STATE_TUPLE_TO_INT(false, false):
+      return "begin write in background";
+    case WRITE_STATE_TUPLE_TO_INT(false, true):
+      return "begin write in current thread";
+    case WRITE_STATE_TUPLE_TO_INT(true, false):
+      return "begin partial write in background";
+    case WRITE_STATE_TUPLE_TO_INT(true, true):
+      return "begin partial write in current thread";
+  }
+  GPR_UNREACHABLE_CODE(return "bad state tuple");
+}
+
 static void write_action_begin_locked(grpc_exec_ctx *exec_ctx, void *gt,
                                       grpc_error *error_ignored) {
   GPR_TIMER_BEGIN("write_action_begin_locked", 0);
   grpc_chttp2_transport *t = (grpc_chttp2_transport *)gt;
   GPR_ASSERT(t->write_state != GRPC_CHTTP2_WRITE_STATE_IDLE);
-  switch (t->closed ? GRPC_CHTTP2_NOTHING_TO_WRITE
+  grpc_chttp2_begin_write_result r;
-                    : grpc_chttp2_begin_write(exec_ctx, t)) {
+  if (t->closed) {
-    case GRPC_CHTTP2_NOTHING_TO_WRITE:
+    r.writing = false;
-      set_write_state(exec_ctx, t, GRPC_CHTTP2_WRITE_STATE_IDLE,
+  } else {
-                      "begin writing nothing");
+    r = grpc_chttp2_begin_write(exec_ctx, t);
-      GRPC_CHTTP2_UNREF_TRANSPORT(exec_ctx, t, "writing");
+  }
-      break;
+  if (r.writing) {
-    case GRPC_CHTTP2_PARTIAL_WRITE:
+    if (r.partial) {
-      set_write_state(exec_ctx, t, GRPC_CHTTP2_WRITE_STATE_WRITING_WITH_MORE,
+      GRPC_STATS_INC_HTTP2_PARTIAL_WRITES(exec_ctx);
-                      "begin writing partial");
+    }
-      GRPC_CLOSURE_SCHED(exec_ctx, &t->write_action, GRPC_ERROR_NONE);
+    if (!t->is_first_write_in_batch) {
-      break;
+      GRPC_STATS_INC_HTTP2_WRITES_CONTINUED(exec_ctx);
-    case GRPC_CHTTP2_FULL_WRITE:
+    }
-      set_write_state(exec_ctx, t, GRPC_CHTTP2_WRITE_STATE_WRITING,
+    grpc_closure_scheduler *scheduler =
-                      "begin writing");
+        write_scheduler(t, r.early_results_scheduled, r.partial);
-      GRPC_CLOSURE_SCHED(exec_ctx, &t->write_action, GRPC_ERROR_NONE);
+    if (scheduler != grpc_schedule_on_exec_ctx) {
-      break;
+      GRPC_STATS_INC_HTTP2_WRITES_OFFLOADED(exec_ctx);
+    }
+    set_write_state(
+        exec_ctx, t, r.partial ? GRPC_CHTTP2_WRITE_STATE_WRITING_WITH_MORE
+                               : GRPC_CHTTP2_WRITE_STATE_WRITING,
+        begin_writing_desc(r.partial, scheduler == grpc_schedule_on_exec_ctx));
+    GRPC_CLOSURE_SCHED(exec_ctx, GRPC_CLOSURE_INIT(&t->write_action,
+                                                   write_action, t, scheduler),
+                       GRPC_ERROR_NONE);
+  } else {
+    set_write_state(exec_ctx, t, GRPC_CHTTP2_WRITE_STATE_IDLE,
+                    "begin writing nothing");
+    GRPC_CHTTP2_UNREF_TRANSPORT(exec_ctx, t, "writing");
   }
   GPR_TIMER_END("write_action_begin_locked", 0);
 }
@@ -958,7 +1002,8 @@ static void write_action_end_locked(grpc_exec_ctx *exec_ctx, void *tp,
     case GRPC_CHTTP2_WRITE_STATE_WRITING_WITH_MORE:
       GPR_TIMER_MARK("state=writing_stale_no_poller", 0);
       set_write_state(exec_ctx, t, GRPC_CHTTP2_WRITE_STATE_WRITING,
-                      "continue writing [!covered]");
+                      "continue writing");
+      t->is_first_write_in_batch = false;
       GRPC_CHTTP2_REF_TRANSPORT(t, "writing");
       GRPC_CLOSURE_RUN(
           exec_ctx,
@@ -1060,9 +1105,7 @@ static void maybe_start_some_streams(grpc_exec_ctx *exec_ctx,
 
     grpc_chttp2_stream_map_add(&t->stream_map, s->id, s);
     post_destructive_reclaimer(exec_ctx, t);
-    grpc_chttp2_become_writable(exec_ctx, t, s,
+    grpc_chttp2_become_writable(exec_ctx, t, s, true, "new_stream");
-                                GRPC_CHTTP2_STREAM_WRITE_INITIATE_COVERED,
-                                "new_stream");
   }
   /* cancel out streams that will never be started */
   while (t->next_stream_id >= MAX_CLIENT_STREAM_ID &&
@@ -1111,12 +1154,14 @@ void grpc_chttp2_complete_closure_step(grpc_exec_ctx *exec_ctx,
   closure->next_data.scratch -= CLOSURE_BARRIER_FIRST_REF_BIT;
   if (GRPC_TRACER_ON(grpc_http_trace)) {
     const char *errstr = grpc_error_string(error);
-    gpr_log(GPR_DEBUG,
+    gpr_log(
-            "complete_closure_step: %p refs=%d flags=0x%04x desc=%s err=%s",
+        GPR_DEBUG,
-            closure,
+        "complete_closure_step: t=%p %p refs=%d flags=0x%04x desc=%s err=%s "
-            (int)(closure->next_data.scratch / CLOSURE_BARRIER_FIRST_REF_BIT),
+        "write_state=%s",
-            (int)(closure->next_data.scratch % CLOSURE_BARRIER_FIRST_REF_BIT),
+        t, closure,
-            desc, errstr);
+        (int)(closure->next_data.scratch / CLOSURE_BARRIER_FIRST_REF_BIT),
+        (int)(closure->next_data.scratch % CLOSURE_BARRIER_FIRST_REF_BIT), desc,
+        errstr, write_state_name(t->write_state));
   }
   if (error != GRPC_ERROR_NONE) {
     if (closure->error_data.error == GRPC_ERROR_NONE) {
@@ -1157,9 +1202,7 @@ static void maybe_become_writable_due_to_send_msg(grpc_exec_ctx *exec_ctx,
                                                   grpc_chttp2_stream *s) {
   if (s->id != 0 && (!s->write_buffering ||
                      s->flow_controlled_buffer.length > t->write_buffer_size)) {
-    grpc_chttp2_become_writable(exec_ctx, t, s,
+    grpc_chttp2_become_writable(exec_ctx, t, s, true, "op.send_message");
-                                GRPC_CHTTP2_STREAM_WRITE_INITIATE_COVERED,
-                                "op.send_message");
   }
 }
 
@@ -1198,8 +1241,12 @@ static void continue_fetching_send_locked(grpc_exec_ctx *exec_ctx,
         cb->call_at_byte = notify_offset;
         cb->closure = s->fetching_send_message_finished;
         s->fetching_send_message_finished = NULL;
-        cb->next = s->on_write_finished_cbs;
+        grpc_chttp2_write_cb **list =
-        s->on_write_finished_cbs = cb;
+            s->fetching_send_message->flags & GRPC_WRITE_THROUGH
+                ? &s->on_write_finished_cbs
+                : &s->on_flow_controlled_cbs;
+        cb->next = *list;
+        *list = cb;
       }
       s->fetching_send_message = NULL;
       return; /* early out */
@@ -1357,14 +1404,13 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, void *stream_op,
           }
         } else {
           GPR_ASSERT(s->id != 0);
-          grpc_chttp2_stream_write_type write_type =
+          bool initiate_write = true;
-              GRPC_CHTTP2_STREAM_WRITE_INITIATE_COVERED;
           if (op->send_message &&
               (op->payload->send_message.send_message->flags &
                GRPC_WRITE_BUFFER_HINT)) {
-            write_type = GRPC_CHTTP2_STREAM_WRITE_PIGGYBACK;
+            initiate_write = false;
           }
-          grpc_chttp2_become_writable(exec_ctx, t, s, write_type,
+          grpc_chttp2_become_writable(exec_ctx, t, s, initiate_write,
                                       "op.send_initial_metadata");
         }
       } else {
@@ -1473,8 +1519,7 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, void *stream_op,
       } else if (s->id != 0) {
         /* TODO(ctiller): check if there's flow control for any outstanding
            bytes before going writable */
-        grpc_chttp2_become_writable(exec_ctx, t, s,
+        grpc_chttp2_become_writable(exec_ctx, t, s, true,
-                                    GRPC_CHTTP2_STREAM_WRITE_INITIATE_COVERED,
                                     "op.send_trailing_metadata");
       }
     }
@@ -1999,6 +2044,21 @@ static grpc_error *removal_error(grpc_error *extra_error, grpc_chttp2_stream *s,
   return error;
 }
 
+static void flush_write_list(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
+                             grpc_chttp2_stream *s, grpc_chttp2_write_cb **list,
+                             grpc_error *error) {
+  while (*list) {
+    grpc_chttp2_write_cb *cb = *list;
+    *list = cb->next;
+    grpc_chttp2_complete_closure_step(exec_ctx, t, s, &cb->closure,
+                                      GRPC_ERROR_REF(error),
+                                      "on_write_finished_cb");
+    cb->next = t->write_cb_pool;
+    t->write_cb_pool = cb;
+  }
+  GRPC_ERROR_UNREF(error);
+}
+
 void grpc_chttp2_fail_pending_writes(grpc_exec_ctx *exec_ctx,
                                      grpc_chttp2_transport *t,
                                      grpc_chttp2_stream *s, grpc_error *error) {
@@ -2018,16 +2078,9 @@ void grpc_chttp2_fail_pending_writes(grpc_exec_ctx *exec_ctx,
   grpc_chttp2_complete_closure_step(
       exec_ctx, t, s, &s->fetching_send_message_finished, GRPC_ERROR_REF(error),
       "fetching_send_message_finished");
-  while (s->on_write_finished_cbs) {
+  flush_write_list(exec_ctx, t, s, &s->on_write_finished_cbs,
-    grpc_chttp2_write_cb *cb = s->on_write_finished_cbs;
+                   GRPC_ERROR_REF(error));
-    s->on_write_finished_cbs = cb->next;
+  flush_write_list(exec_ctx, t, s, &s->on_flow_controlled_cbs, error);
-    grpc_chttp2_complete_closure_step(exec_ctx, t, s, &cb->closure,
-                                      GRPC_ERROR_REF(error),
-                                      "on_write_finished_cb");
-    cb->next = t->write_cb_pool;
-    t->write_cb_pool = cb;
-  }
-  GRPC_ERROR_UNREF(error);
 }
 
 void grpc_chttp2_mark_stream_closed(grpc_exec_ctx *exec_ctx,
@@ -2271,13 +2324,11 @@ void grpc_chttp2_act_on_flowctl_action(grpc_exec_ctx *exec_ctx,
     case GRPC_CHTTP2_FLOWCTL_NO_ACTION_NEEDED:
       break;
     case GRPC_CHTTP2_FLOWCTL_UPDATE_IMMEDIATELY:
-      grpc_chttp2_become_writable(exec_ctx, t, s,
+      grpc_chttp2_become_writable(exec_ctx, t, s, true,
-                                  GRPC_CHTTP2_STREAM_WRITE_INITIATE_COVERED,
                                   "immediate stream flowctl");
       break;
     case GRPC_CHTTP2_FLOWCTL_QUEUE_UPDATE:
-      grpc_chttp2_become_writable(exec_ctx, t, s,
+      grpc_chttp2_become_writable(exec_ctx, t, s, false,
-                                  GRPC_CHTTP2_STREAM_WRITE_PIGGYBACK,
                                   "queue stream flowctl");
       break;
   }
@@ -2390,9 +2441,7 @@ static void read_action_locked(grpc_exec_ctx *exec_ctx, void *tp,
       if (t->flow_control.initial_window_update > 0) {
         grpc_chttp2_stream *s;
         while (grpc_chttp2_list_pop_stalled_by_stream(t, &s)) {
-          grpc_chttp2_become_writable(
+          grpc_chttp2_become_writable(exec_ctx, t, s, true, "unstalled");
-              exec_ctx, t, s, GRPC_CHTTP2_STREAM_WRITE_INITIATE_UNCOVERED,
-              "unstalled");
         }
       }
       t->flow_control.initial_window_update = 0;
@@ -2983,6 +3032,8 @@ static const grpc_transport_vtable vtable = {sizeof(grpc_chttp2_stream),
                                              destroy_transport,
                                              chttp2_get_endpoint};
 
+static const grpc_transport_vtable *get_vtable(void) { return &vtable; }
+
 grpc_transport *grpc_create_chttp2_transport(
     grpc_exec_ctx *exec_ctx, const grpc_channel_args *channel_args,
     grpc_endpoint *ep, int is_client) {

src/core/ext/transport/chttp2/transport/frame_window_update.c

@@ -99,9 +99,8 @@ grpc_error *grpc_chttp2_window_update_parser_parse(
         grpc_chttp2_flowctl_recv_stream_update(
             &t->flow_control, &s->flow_control, received_update);
         if (grpc_chttp2_list_remove_stalled_by_stream(t, s)) {
-          grpc_chttp2_become_writable(
+          grpc_chttp2_become_writable(exec_ctx, t, s, true,
-              exec_ctx, t, s, GRPC_CHTTP2_STREAM_WRITE_INITIATE_UNCOVERED,
+                                      "stream.read_flow_control");
-              "stream.read_flow_control");
         }
       }
     } else {

src/core/ext/transport/chttp2/transport/internal.h

@@ -262,6 +262,10 @@ struct grpc_chttp2_transport {
 
   /** write execution state of the transport */
   grpc_chttp2_write_state write_state;
+  /** is this the first write in a series of writes?
+      set when we initiate writing from idle, cleared when we
+      initiate writing from writing+more */
+  bool is_first_write_in_batch;
 
   /** is the transport destroying itself? */
   uint8_t destroying;
@@ -483,6 +487,7 @@ struct grpc_chttp2_stream {
   grpc_slice fetching_slice;
   int64_t next_message_end_offset;
   int64_t flow_controlled_bytes_written;
+  int64_t flow_controlled_bytes_flowed;
   grpc_closure complete_fetch_locked;
   grpc_closure *fetching_send_message_finished;
 
@@ -555,6 +560,7 @@ struct grpc_chttp2_stream {
 
   grpc_slice_buffer flow_controlled_buffer;
 
+  grpc_chttp2_write_cb *on_flow_controlled_cbs;
   grpc_chttp2_write_cb *on_write_finished_cbs;
   grpc_chttp2_write_cb *finish_after_write;
   size_t sending_bytes;
@@ -595,10 +601,13 @@ struct grpc_chttp2_stream {
 void grpc_chttp2_initiate_write(grpc_exec_ctx *exec_ctx,
                                 grpc_chttp2_transport *t, const char *reason);
 
-typedef enum {
+typedef struct {
-  GRPC_CHTTP2_NOTHING_TO_WRITE,
+  /** are we writing? */
-  GRPC_CHTTP2_PARTIAL_WRITE,
+  bool writing;
-  GRPC_CHTTP2_FULL_WRITE,
+  /** if writing: was it a complete flush (false) or a partial flush (true) */
+  bool partial;
+  /** did we queue any completions as part of beginning the write */
+  bool early_results_scheduled;
 } grpc_chttp2_begin_write_result;
 
 grpc_chttp2_begin_write_result grpc_chttp2_begin_write(
@@ -840,22 +849,12 @@ void grpc_chttp2_ack_ping(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
 void grpc_chttp2_add_ping_strike(grpc_exec_ctx *exec_ctx,
                                  grpc_chttp2_transport *t);
 
-typedef enum {
-  /* don't initiate a transport write, but piggyback on the next one */
-  GRPC_CHTTP2_STREAM_WRITE_PIGGYBACK,
-  /* initiate a covered write */
-  GRPC_CHTTP2_STREAM_WRITE_INITIATE_COVERED,
-  /* initiate an uncovered write */
-  GRPC_CHTTP2_STREAM_WRITE_INITIATE_UNCOVERED
-} grpc_chttp2_stream_write_type;
-
 /** add a ref to the stream and add it to the writable list;
     ref will be dropped in writing.c */
 void grpc_chttp2_become_writable(grpc_exec_ctx *exec_ctx,
                                  grpc_chttp2_transport *t,
                                  grpc_chttp2_stream *s,
-                                 grpc_chttp2_stream_write_type type,
+                                 bool also_initiate_write, const char *reason);
-                                 const char *reason);
 
 void grpc_chttp2_cancel_stream(grpc_exec_ctx *exec_ctx,
                                grpc_chttp2_transport *t, grpc_chttp2_stream *s,

src/core/ext/transport/chttp2/transport/parsing.c

@@ -106,7 +106,8 @@ grpc_error *grpc_chttp2_perform_read(grpc_exec_ctx *exec_ctx,
           return err;
         }
         ++cur;
-        ++t->deframe_state;
+        t->deframe_state =
+            (grpc_chttp2_deframe_transport_state)(1 + (int)t->deframe_state);
       }
       if (cur == end) {
         return GRPC_ERROR_NONE;

src/core/ext/transport/chttp2/transport/writing.c

@@ -123,15 +123,18 @@ static void maybe_initiate_ping(grpc_exec_ctx *exec_ctx,
       (t->ping_state.pings_before_data_required != 0);
 }
 
-static void update_list(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
+static bool update_list(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
                         grpc_chttp2_stream *s, int64_t send_bytes,
-                        grpc_chttp2_write_cb **list, grpc_error *error) {
+                        grpc_chttp2_write_cb **list, int64_t *ctr,
+                        grpc_error *error) {
+  bool sched_any = false;
   grpc_chttp2_write_cb *cb = *list;
   *list = NULL;
-  s->flow_controlled_bytes_written += send_bytes;
+  *ctr += send_bytes;
   while (cb) {
     grpc_chttp2_write_cb *next = cb->next;
-    if (cb->call_at_byte <= s->flow_controlled_bytes_written) {
+    if (cb->call_at_byte <= *ctr) {
+      sched_any = true;
       finish_write_cb(exec_ctx, t, s, cb, GRPC_ERROR_REF(error));
     } else {
       add_to_write_list(list, cb);
@@ -139,6 +142,7 @@ static void update_list(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
     cb = next;
   }
   GRPC_ERROR_UNREF(error);
+  return sched_any;
 }
 
 static bool stream_ref_if_not_destroyed(gpr_refcount *r) {
@@ -164,6 +168,13 @@ grpc_chttp2_begin_write_result grpc_chttp2_begin_write(
     grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t) {
   grpc_chttp2_stream *s;
 
+  /* stats histogram counters: we increment these throughout this function,
+     and at the end publish to the central stats histograms */
+  int flow_control_writes = 0;
+  int initial_metadata_writes = 0;
+  int trailing_metadata_writes = 0;
+  int message_writes = 0;
+
   GRPC_STATS_INC_HTTP2_WRITES_BEGUN(exec_ctx);
 
   GPR_TIMER_BEGIN("grpc_chttp2_begin_write", 0);
@@ -177,6 +188,7 @@ grpc_chttp2_begin_write_result grpc_chttp2_begin_write(
     t->force_send_settings = 0;
     t->dirtied_local_settings = 0;
     t->sent_local_settings = 1;
+    GRPC_STATS_INC_HTTP2_SETTINGS_WRITES(exec_ctx);
   }
 
   /* simple writes are queued to qbuf, and flushed here */
@@ -196,13 +208,13 @@ grpc_chttp2_begin_write_result grpc_chttp2_begin_write(
     }
   }
 
-  bool partial_write = false;
+  grpc_chttp2_begin_write_result result = {false, false, false};
 
   /* for each grpc_chttp2_stream that's become writable, frame it's data
      (according to available window sizes) and add to the output buffer */
   while (true) {
     if (t->outbuf.length > target_write_size(t)) {
-      partial_write = true;
+      result.partial = true;
       break;
     }
 
@@ -246,7 +258,6 @@ grpc_chttp2_begin_write_result grpc_chttp2_begin_write(
             .stats = &s->stats.outgoing};
         grpc_chttp2_encode_header(exec_ctx, &t->hpack_compressor, NULL, 0,
                                   s->send_initial_metadata, &hopt, &t->outbuf);
-        now_writing = true;
         t->ping_state.pings_before_data_required =
             t->ping_policy.max_pings_without_data;
         if (!t->is_client) {
@@ -254,6 +265,7 @@ grpc_chttp2_begin_write_result grpc_chttp2_begin_write(
               gpr_inf_past(GPR_CLOCK_MONOTONIC);
           t->ping_recv_state.ping_strikes = 0;
         }
+        initial_metadata_writes++;
       } else {
         GRPC_CHTTP2_IF_TRACING(
             gpr_log(GPR_INFO, "not sending initial_metadata (Trailers-Only)"));
@@ -269,10 +281,15 @@ grpc_chttp2_begin_write_result grpc_chttp2_begin_write(
               [num_extra_headers_for_trailing_metadata++] =
                   &s->send_initial_metadata->idx.named.content_type->md;
         }
+        trailing_metadata_writes++;
       }
       s->send_initial_metadata = NULL;
       s->sent_initial_metadata = true;
       sent_initial_metadata = true;
+      result.early_results_scheduled = true;
+      grpc_chttp2_complete_closure_step(
+          exec_ctx, t, s, &s->send_initial_metadata_finished, GRPC_ERROR_NONE,
+          "send_initial_metadata_finished");
     }
     /* send any window updates */
     uint32_t stream_announce = grpc_chttp2_flowctl_maybe_send_stream_update(
@@ -288,6 +305,7 @@ grpc_chttp2_begin_write_result grpc_chttp2_begin_write(
             gpr_inf_past(GPR_CLOCK_MONOTONIC);
         t->ping_recv_state.ping_strikes = 0;
       }
+      flow_control_writes++;
     }
     if (sent_initial_metadata) {
       /* send any body bytes, if allowed by flow control */
@@ -306,6 +324,7 @@ grpc_chttp2_begin_write_result grpc_chttp2_begin_write(
         if (max_outgoing > 0) {
           bool is_last_data_frame = false;
           bool is_last_frame = false;
+          size_t sending_bytes_before = s->sending_bytes;
           if (s->stream_compression_send_enabled) {
             while ((s->flow_controlled_buffer.length > 0 ||
                     s->compressed_data_buffer->length > 0) &&
@@ -373,6 +392,11 @@ grpc_chttp2_begin_write_result grpc_chttp2_begin_write(
                                                     &s->stats.outgoing));
             }
           }
+          result.early_results_scheduled |=
+              update_list(exec_ctx, t, s,
+                          (int64_t)(s->sending_bytes - sending_bytes_before),
+                          &s->on_flow_controlled_cbs,
+                          &s->flow_controlled_bytes_flowed, GRPC_ERROR_NONE);
           now_writing = true;
           if (s->flow_controlled_buffer.length > 0 ||
               (s->stream_compression_send_enabled &&
@@ -380,6 +404,7 @@ grpc_chttp2_begin_write_result grpc_chttp2_begin_write(
             GRPC_CHTTP2_STREAM_REF(s, "chttp2_writing:fork");
             grpc_chttp2_list_add_writable_stream(t, s);
           }
+          message_writes++;
         } else if (t->flow_control.remote_window == 0) {
           grpc_chttp2_list_add_stalled_by_transport(t, s);
           now_writing = true;
@@ -415,6 +440,7 @@ grpc_chttp2_begin_write_result grpc_chttp2_begin_write(
                                     num_extra_headers_for_trailing_metadata,
                                     s->send_trailing_metadata, &hopt,
                                     &t->outbuf);
+          trailing_metadata_writes++;
         }
         s->send_trailing_metadata = NULL;
         s->sent_trailing_metadata = true;
@@ -424,10 +450,22 @@ grpc_chttp2_begin_write_result grpc_chttp2_begin_write(
                               s->id, GRPC_HTTP2_NO_ERROR, &s->stats.outgoing));
         }
         now_writing = true;
+        result.early_results_scheduled = true;
+        grpc_chttp2_complete_closure_step(
+            exec_ctx, t, s, &s->send_trailing_metadata_finished,
+            GRPC_ERROR_NONE, "send_trailing_metadata_finished");
       }
     }
 
     if (now_writing) {
+      GRPC_STATS_INC_HTTP2_SEND_INITIAL_METADATA_PER_WRITE(
+          exec_ctx, initial_metadata_writes);
+      GRPC_STATS_INC_HTTP2_SEND_MESSAGE_PER_WRITE(exec_ctx, message_writes);
+      GRPC_STATS_INC_HTTP2_SEND_TRAILING_METADATA_PER_WRITE(
+          exec_ctx, trailing_metadata_writes);
+      GRPC_STATS_INC_HTTP2_SEND_FLOWCTL_PER_WRITE(exec_ctx,
+                                                  flow_control_writes);
+
       if (!grpc_chttp2_list_add_writing_stream(t, s)) {
         /* already in writing list: drop ref */
         GRPC_CHTTP2_STREAM_UNREF(exec_ctx, s, "chttp2_writing:already_writing");
@@ -465,9 +503,8 @@ grpc_chttp2_begin_write_result grpc_chttp2_begin_write(
 
   GPR_TIMER_END("grpc_chttp2_begin_write", 0);
 
-  return t->outbuf.count > 0 ? (partial_write ? GRPC_CHTTP2_PARTIAL_WRITE
+  result.writing = t->outbuf.count > 0;
-                                              : GRPC_CHTTP2_FULL_WRITE)
+  return result;
-                             : GRPC_CHTTP2_NOTHING_TO_WRITE;
 }
 
 void grpc_chttp2_end_write(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
@@ -476,20 +513,13 @@ void grpc_chttp2_end_write(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
   grpc_chttp2_stream *s;
 
   while (grpc_chttp2_list_pop_writing_stream(t, &s)) {
-    if (s->sent_initial_metadata) {
-      grpc_chttp2_complete_closure_step(
-          exec_ctx, t, s, &s->send_initial_metadata_finished,
-          GRPC_ERROR_REF(error), "send_initial_metadata_finished");
-    }
     if (s->sending_bytes != 0) {
       update_list(exec_ctx, t, s, (int64_t)s->sending_bytes,
-                  &s->on_write_finished_cbs, GRPC_ERROR_REF(error));
+                  &s->on_write_finished_cbs, &s->flow_controlled_bytes_written,
+                  GRPC_ERROR_REF(error));
       s->sending_bytes = 0;
     }
     if (s->sent_trailing_metadata) {
-      grpc_chttp2_complete_closure_step(
-          exec_ctx, t, s, &s->send_trailing_metadata_finished,
-          GRPC_ERROR_REF(error), "send_trailing_metadata_finished");
       grpc_chttp2_mark_stream_closed(exec_ctx, t, s, !t->is_client, 1,
                                      GRPC_ERROR_REF(error));
     }

src/core/lib/debug/stats_data.c

@@ -30,6 +30,8 @@ const char *grpc_stats_counter_name[GRPC_STATS_COUNTER_COUNT] = {
     "histogram_slow_lookups",
     "syscall_write",
     "syscall_read",
+    "tcp_backup_pollers_created",
+    "tcp_backup_poller_polls",
     "http2_op_batches",
     "http2_op_cancel",
     "http2_op_send_initial_metadata",
@@ -38,16 +40,22 @@ const char *grpc_stats_counter_name[GRPC_STATS_COUNTER_COUNT] = {
     "http2_op_recv_initial_metadata",
     "http2_op_recv_message",
     "http2_op_recv_trailing_metadata",
+    "http2_settings_writes",
     "http2_pings_sent",
     "http2_writes_begun",
+    "http2_writes_offloaded",
+    "http2_writes_continued",
+    "http2_partial_writes",
     "combiner_locks_initiated",
     "combiner_locks_scheduled_items",
     "combiner_locks_scheduled_final_items",
     "combiner_locks_offloaded",
-    "executor_scheduled_items",
+    "executor_scheduled_short_items",
+    "executor_scheduled_long_items",
     "executor_scheduled_to_self",
     "executor_wakeup_initiated",
     "executor_queue_drained",
+    "executor_push_retries",
 };
 const char *grpc_stats_counter_doc[GRPC_STATS_COUNTER_COUNT] = {
     "Number of client side calls created by this process",
@@ -59,6 +67,8 @@ const char *grpc_stats_counter_doc[GRPC_STATS_COUNTER_COUNT] = {
     "Number of write syscalls (or equivalent - eg sendmsg) made by this "
     "process",
     "Number of read syscalls (or equivalent - eg recvmsg) made by this process",
+    "Number of times a backup poller has been created (this can be expensive)",
+    "Number of polls performed on the backup poller",
     "Number of batches received by HTTP2 transport",
     "Number of cancelations received by HTTP2 transport",
     "Number of batches containing send initial metadata",
@@ -67,20 +77,39 @@ const char *grpc_stats_counter_doc[GRPC_STATS_COUNTER_COUNT] = {
     "Number of batches containing receive initial metadata",
     "Number of batches containing receive message",
     "Number of batches containing receive trailing metadata",
-    "Number of HTTP2 pings sent by process", "Number of HTTP2 writes initiated",
+    "Number of settings frames sent", "Number of HTTP2 pings sent by process",
+    "Number of HTTP2 writes initiated",
+    "Number of HTTP2 writes offloaded to the executor from application threads",
+    "Number of HTTP2 writes that finished seeing more data needed to be "
+    "written",
+    "Number of HTTP2 writes that were made knowing there was still more data "
+    "to be written (we cap maximum write size to syscall_write)",
     "Number of combiner lock entries by process (first items queued to a "
     "combiner)",
     "Number of items scheduled against combiner locks",
     "Number of final items scheduled against combiner locks",
     "Number of combiner locks offloaded to different threads",
-    "Number of closures scheduled against the executor (gRPC thread pool)",
+    "Number of finite runtime closures scheduled against the executor (gRPC "
+    "thread pool)",
+    "Number of potentially infinite runtime closures scheduled against the "
+    "executor (gRPC thread pool)",
     "Number of closures scheduled by the executor to the executor",
     "Number of thread wakeups initiated within the executor",
     "Number of times an executor queue was drained",
+    "Number of times we raced and were forced to retry pushing a closure to "
+    "the executor",
 };
 const char *grpc_stats_histogram_name[GRPC_STATS_HISTOGRAM_COUNT] = {
-    "tcp_write_size", "tcp_write_iov_size",      "tcp_read_size",
+    "tcp_write_size",
-    "tcp_read_offer", "tcp_read_offer_iov_size", "http2_send_message_size",
+    "tcp_write_iov_size",
+    "tcp_read_size",
+    "tcp_read_offer",
+    "tcp_read_offer_iov_size",
+    "http2_send_message_size",
+    "http2_send_initial_metadata_per_write",
+    "http2_send_message_per_write",
+    "http2_send_trailing_metadata_per_write",
+    "http2_send_flowctl_per_write",
 };
 const char *grpc_stats_histogram_doc[GRPC_STATS_HISTOGRAM_COUNT] = {
     "Number of bytes offered to each syscall_write",
@@ -89,6 +118,10 @@ const char *grpc_stats_histogram_doc[GRPC_STATS_HISTOGRAM_COUNT] = {
     "Number of bytes offered to each syscall_read",
     "Number of byte segments offered to each syscall_read",
     "Size of messages received by HTTP2 transport",
+    "Number of streams initiated written per TCP write",
+    "Number of streams whose payload was written per TCP write",
+    "Number of streams terminated per TCP write",
+    "Number of flow control updates written per TCP write",
 };
 const int grpc_stats_table_0[65] = {
     0,       1,       2,       3,       4,       6,       8,        11,
@@ -273,15 +306,135 @@ void grpc_stats_inc_http2_send_message_size(grpc_exec_ctx *exec_ctx,
                            grpc_stats_histo_find_bucket_slow(
                                (exec_ctx), value, grpc_stats_table_0, 64));
 }
-const int grpc_stats_histo_buckets[6] = {64, 64, 64, 64, 64, 64};
+void grpc_stats_inc_http2_send_initial_metadata_per_write(
-const int grpc_stats_histo_start[6] = {0, 64, 128, 192, 256, 320};
+    grpc_exec_ctx *exec_ctx, int value) {
-const int *const grpc_stats_histo_bucket_boundaries[6] = {
+  value = GPR_CLAMP(value, 0, 1024);
+  if (value < 13) {
+    GRPC_STATS_INC_HISTOGRAM(
+        (exec_ctx), GRPC_STATS_HISTOGRAM_HTTP2_SEND_INITIAL_METADATA_PER_WRITE,
+        value);
+    return;
+  }
+  union {
+    double dbl;
+    uint64_t uint;
+  } _val, _bkt;
+  _val.dbl = value;
+  if (_val.uint < 4637863191261478912ull) {
+    int bucket =
+        grpc_stats_table_3[((_val.uint - 4623507967449235456ull) >> 48)] + 13;
+    _bkt.dbl = grpc_stats_table_2[bucket];
+    bucket -= (_val.uint < _bkt.uint);
+    GRPC_STATS_INC_HISTOGRAM(
+        (exec_ctx), GRPC_STATS_HISTOGRAM_HTTP2_SEND_INITIAL_METADATA_PER_WRITE,
+        bucket);
+    return;
+  }
+  GRPC_STATS_INC_HISTOGRAM(
+      (exec_ctx), GRPC_STATS_HISTOGRAM_HTTP2_SEND_INITIAL_METADATA_PER_WRITE,
+      grpc_stats_histo_find_bucket_slow((exec_ctx), value, grpc_stats_table_2,
+                                        64));
+}
+void grpc_stats_inc_http2_send_message_per_write(grpc_exec_ctx *exec_ctx,
+                                                 int value) {
+  value = GPR_CLAMP(value, 0, 1024);
+  if (value < 13) {
+    GRPC_STATS_INC_HISTOGRAM(
+        (exec_ctx), GRPC_STATS_HISTOGRAM_HTTP2_SEND_MESSAGE_PER_WRITE, value);
+    return;
+  }
+  union {
+    double dbl;
+    uint64_t uint;
+  } _val, _bkt;
+  _val.dbl = value;
+  if (_val.uint < 4637863191261478912ull) {
+    int bucket =
+        grpc_stats_table_3[((_val.uint - 4623507967449235456ull) >> 48)] + 13;
+    _bkt.dbl = grpc_stats_table_2[bucket];
+    bucket -= (_val.uint < _bkt.uint);
+    GRPC_STATS_INC_HISTOGRAM(
+        (exec_ctx), GRPC_STATS_HISTOGRAM_HTTP2_SEND_MESSAGE_PER_WRITE, bucket);
+    return;
+  }
+  GRPC_STATS_INC_HISTOGRAM((exec_ctx),
+                           GRPC_STATS_HISTOGRAM_HTTP2_SEND_MESSAGE_PER_WRITE,
+                           grpc_stats_histo_find_bucket_slow(
+                               (exec_ctx), value, grpc_stats_table_2, 64));
+}
+void grpc_stats_inc_http2_send_trailing_metadata_per_write(
+    grpc_exec_ctx *exec_ctx, int value) {
+  value = GPR_CLAMP(value, 0, 1024);
+  if (value < 13) {
+    GRPC_STATS_INC_HISTOGRAM(
+        (exec_ctx), GRPC_STATS_HISTOGRAM_HTTP2_SEND_TRAILING_METADATA_PER_WRITE,
+        value);
+    return;
+  }
+  union {
+    double dbl;
+    uint64_t uint;
+  } _val, _bkt;
+  _val.dbl = value;
+  if (_val.uint < 4637863191261478912ull) {
+    int bucket =
+        grpc_stats_table_3[((_val.uint - 4623507967449235456ull) >> 48)] + 13;
+    _bkt.dbl = grpc_stats_table_2[bucket];
+    bucket -= (_val.uint < _bkt.uint);
+    GRPC_STATS_INC_HISTOGRAM(
+        (exec_ctx), GRPC_STATS_HISTOGRAM_HTTP2_SEND_TRAILING_METADATA_PER_WRITE,
+        bucket);
+    return;
+  }
+  GRPC_STATS_INC_HISTOGRAM(
+      (exec_ctx), GRPC_STATS_HISTOGRAM_HTTP2_SEND_TRAILING_METADATA_PER_WRITE,
+      grpc_stats_histo_find_bucket_slow((exec_ctx), value, grpc_stats_table_2,
+                                        64));
+}
+void grpc_stats_inc_http2_send_flowctl_per_write(grpc_exec_ctx *exec_ctx,
+                                                 int value) {
+  value = GPR_CLAMP(value, 0, 1024);
+  if (value < 13) {
+    GRPC_STATS_INC_HISTOGRAM(
+        (exec_ctx), GRPC_STATS_HISTOGRAM_HTTP2_SEND_FLOWCTL_PER_WRITE, value);
+    return;
+  }
+  union {
+    double dbl;
+    uint64_t uint;
+  } _val, _bkt;
+  _val.dbl = value;
+  if (_val.uint < 4637863191261478912ull) {
+    int bucket =
+        grpc_stats_table_3[((_val.uint - 4623507967449235456ull) >> 48)] + 13;
+    _bkt.dbl = grpc_stats_table_2[bucket];
+    bucket -= (_val.uint < _bkt.uint);
+    GRPC_STATS_INC_HISTOGRAM(
+        (exec_ctx), GRPC_STATS_HISTOGRAM_HTTP2_SEND_FLOWCTL_PER_WRITE, bucket);
+    return;
+  }
+  GRPC_STATS_INC_HISTOGRAM((exec_ctx),
+                           GRPC_STATS_HISTOGRAM_HTTP2_SEND_FLOWCTL_PER_WRITE,
+                           grpc_stats_histo_find_bucket_slow(
+                               (exec_ctx), value, grpc_stats_table_2, 64));
+}
+const int grpc_stats_histo_buckets[10] = {64, 64, 64, 64, 64,
+                                          64, 64, 64, 64, 64};
+const int grpc_stats_histo_start[10] = {0,   64,  128, 192, 256,
+                                        320, 384, 448, 512, 576};
+const int *const grpc_stats_histo_bucket_boundaries[10] = {
+    grpc_stats_table_0, grpc_stats_table_2, grpc_stats_table_0,
     grpc_stats_table_0, grpc_stats_table_2, grpc_stats_table_0,
-    grpc_stats_table_0, grpc_stats_table_2, grpc_stats_table_0};
+    grpc_stats_table_2, grpc_stats_table_2, grpc_stats_table_2,
-void (*const grpc_stats_inc_histogram[6])(grpc_exec_ctx *exec_ctx, int x) = {
+    grpc_stats_table_2};
+void (*const grpc_stats_inc_histogram[10])(grpc_exec_ctx *exec_ctx, int x) = {
     grpc_stats_inc_tcp_write_size,
     grpc_stats_inc_tcp_write_iov_size,
     grpc_stats_inc_tcp_read_size,
     grpc_stats_inc_tcp_read_offer,
     grpc_stats_inc_tcp_read_offer_iov_size,
-    grpc_stats_inc_http2_send_message_size};
+    grpc_stats_inc_http2_send_message_size,
+    grpc_stats_inc_http2_send_initial_metadata_per_write,
+    grpc_stats_inc_http2_send_message_per_write,
+    grpc_stats_inc_http2_send_trailing_metadata_per_write,
+    grpc_stats_inc_http2_send_flowctl_per_write};

src/core/lib/debug/stats_data.h

@@ -32,6 +32,8 @@ typedef enum {
   GRPC_STATS_COUNTER_HISTOGRAM_SLOW_LOOKUPS,
   GRPC_STATS_COUNTER_SYSCALL_WRITE,
   GRPC_STATS_COUNTER_SYSCALL_READ,
+  GRPC_STATS_COUNTER_TCP_BACKUP_POLLERS_CREATED,
+  GRPC_STATS_COUNTER_TCP_BACKUP_POLLER_POLLS,
   GRPC_STATS_COUNTER_HTTP2_OP_BATCHES,
   GRPC_STATS_COUNTER_HTTP2_OP_CANCEL,
   GRPC_STATS_COUNTER_HTTP2_OP_SEND_INITIAL_METADATA,
@@ -40,16 +42,22 @@ typedef enum {
   GRPC_STATS_COUNTER_HTTP2_OP_RECV_INITIAL_METADATA,
   GRPC_STATS_COUNTER_HTTP2_OP_RECV_MESSAGE,
   GRPC_STATS_COUNTER_HTTP2_OP_RECV_TRAILING_METADATA,
+  GRPC_STATS_COUNTER_HTTP2_SETTINGS_WRITES,
   GRPC_STATS_COUNTER_HTTP2_PINGS_SENT,
   GRPC_STATS_COUNTER_HTTP2_WRITES_BEGUN,
+  GRPC_STATS_COUNTER_HTTP2_WRITES_OFFLOADED,
+  GRPC_STATS_COUNTER_HTTP2_WRITES_CONTINUED,
+  GRPC_STATS_COUNTER_HTTP2_PARTIAL_WRITES,
   GRPC_STATS_COUNTER_COMBINER_LOCKS_INITIATED,
   GRPC_STATS_COUNTER_COMBINER_LOCKS_SCHEDULED_ITEMS,
   GRPC_STATS_COUNTER_COMBINER_LOCKS_SCHEDULED_FINAL_ITEMS,
   GRPC_STATS_COUNTER_COMBINER_LOCKS_OFFLOADED,
-  GRPC_STATS_COUNTER_EXECUTOR_SCHEDULED_ITEMS,
+  GRPC_STATS_COUNTER_EXECUTOR_SCHEDULED_SHORT_ITEMS,
+  GRPC_STATS_COUNTER_EXECUTOR_SCHEDULED_LONG_ITEMS,
   GRPC_STATS_COUNTER_EXECUTOR_SCHEDULED_TO_SELF,
   GRPC_STATS_COUNTER_EXECUTOR_WAKEUP_INITIATED,
   GRPC_STATS_COUNTER_EXECUTOR_QUEUE_DRAINED,
+  GRPC_STATS_COUNTER_EXECUTOR_PUSH_RETRIES,
   GRPC_STATS_COUNTER_COUNT
 } grpc_stats_counters;
 extern const char *grpc_stats_counter_name[GRPC_STATS_COUNTER_COUNT];
@@ -61,6 +69,10 @@ typedef enum {
   GRPC_STATS_HISTOGRAM_TCP_READ_OFFER,
   GRPC_STATS_HISTOGRAM_TCP_READ_OFFER_IOV_SIZE,
   GRPC_STATS_HISTOGRAM_HTTP2_SEND_MESSAGE_SIZE,
+  GRPC_STATS_HISTOGRAM_HTTP2_SEND_INITIAL_METADATA_PER_WRITE,
+  GRPC_STATS_HISTOGRAM_HTTP2_SEND_MESSAGE_PER_WRITE,
+  GRPC_STATS_HISTOGRAM_HTTP2_SEND_TRAILING_METADATA_PER_WRITE,
+  GRPC_STATS_HISTOGRAM_HTTP2_SEND_FLOWCTL_PER_WRITE,
   GRPC_STATS_HISTOGRAM_COUNT
 } grpc_stats_histograms;
 extern const char *grpc_stats_histogram_name[GRPC_STATS_HISTOGRAM_COUNT];
@@ -78,7 +90,15 @@ typedef enum {
   GRPC_STATS_HISTOGRAM_TCP_READ_OFFER_IOV_SIZE_BUCKETS = 64,
   GRPC_STATS_HISTOGRAM_HTTP2_SEND_MESSAGE_SIZE_FIRST_SLOT = 320,
   GRPC_STATS_HISTOGRAM_HTTP2_SEND_MESSAGE_SIZE_BUCKETS = 64,
-  GRPC_STATS_HISTOGRAM_BUCKETS = 384
+  GRPC_STATS_HISTOGRAM_HTTP2_SEND_INITIAL_METADATA_PER_WRITE_FIRST_SLOT = 384,
+  GRPC_STATS_HISTOGRAM_HTTP2_SEND_INITIAL_METADATA_PER_WRITE_BUCKETS = 64,
+  GRPC_STATS_HISTOGRAM_HTTP2_SEND_MESSAGE_PER_WRITE_FIRST_SLOT = 448,
+  GRPC_STATS_HISTOGRAM_HTTP2_SEND_MESSAGE_PER_WRITE_BUCKETS = 64,
+  GRPC_STATS_HISTOGRAM_HTTP2_SEND_TRAILING_METADATA_PER_WRITE_FIRST_SLOT = 512,
+  GRPC_STATS_HISTOGRAM_HTTP2_SEND_TRAILING_METADATA_PER_WRITE_BUCKETS = 64,
+  GRPC_STATS_HISTOGRAM_HTTP2_SEND_FLOWCTL_PER_WRITE_FIRST_SLOT = 576,
+  GRPC_STATS_HISTOGRAM_HTTP2_SEND_FLOWCTL_PER_WRITE_BUCKETS = 64,
+  GRPC_STATS_HISTOGRAM_BUCKETS = 640
 } grpc_stats_histogram_constants;
 #define GRPC_STATS_INC_CLIENT_CALLS_CREATED(exec_ctx) \
   GRPC_STATS_INC_COUNTER((exec_ctx), GRPC_STATS_COUNTER_CLIENT_CALLS_CREATED)
@@ -94,6 +114,11 @@ typedef enum {
   GRPC_STATS_INC_COUNTER((exec_ctx), GRPC_STATS_COUNTER_SYSCALL_WRITE)
 #define GRPC_STATS_INC_SYSCALL_READ(exec_ctx) \
   GRPC_STATS_INC_COUNTER((exec_ctx), GRPC_STATS_COUNTER_SYSCALL_READ)
+#define GRPC_STATS_INC_TCP_BACKUP_POLLERS_CREATED(exec_ctx) \
+  GRPC_STATS_INC_COUNTER((exec_ctx),                        \
+                         GRPC_STATS_COUNTER_TCP_BACKUP_POLLERS_CREATED)
+#define GRPC_STATS_INC_TCP_BACKUP_POLLER_POLLS(exec_ctx) \
+  GRPC_STATS_INC_COUNTER((exec_ctx), GRPC_STATS_COUNTER_TCP_BACKUP_POLLER_POLLS)
 #define GRPC_STATS_INC_HTTP2_OP_BATCHES(exec_ctx) \
   GRPC_STATS_INC_COUNTER((exec_ctx), GRPC_STATS_COUNTER_HTTP2_OP_BATCHES)
 #define GRPC_STATS_INC_HTTP2_OP_CANCEL(exec_ctx) \
@@ -114,10 +139,18 @@ typedef enum {
 #define GRPC_STATS_INC_HTTP2_OP_RECV_TRAILING_METADATA(exec_ctx) \
   GRPC_STATS_INC_COUNTER((exec_ctx),                             \
                          GRPC_STATS_COUNTER_HTTP2_OP_RECV_TRAILING_METADATA)
+#define GRPC_STATS_INC_HTTP2_SETTINGS_WRITES(exec_ctx) \
+  GRPC_STATS_INC_COUNTER((exec_ctx), GRPC_STATS_COUNTER_HTTP2_SETTINGS_WRITES)
 #define GRPC_STATS_INC_HTTP2_PINGS_SENT(exec_ctx) \
   GRPC_STATS_INC_COUNTER((exec_ctx), GRPC_STATS_COUNTER_HTTP2_PINGS_SENT)
 #define GRPC_STATS_INC_HTTP2_WRITES_BEGUN(exec_ctx) \
   GRPC_STATS_INC_COUNTER((exec_ctx), GRPC_STATS_COUNTER_HTTP2_WRITES_BEGUN)
+#define GRPC_STATS_INC_HTTP2_WRITES_OFFLOADED(exec_ctx) \
+  GRPC_STATS_INC_COUNTER((exec_ctx), GRPC_STATS_COUNTER_HTTP2_WRITES_OFFLOADED)
+#define GRPC_STATS_INC_HTTP2_WRITES_CONTINUED(exec_ctx) \
+  GRPC_STATS_INC_COUNTER((exec_ctx), GRPC_STATS_COUNTER_HTTP2_WRITES_CONTINUED)
+#define GRPC_STATS_INC_HTTP2_PARTIAL_WRITES(exec_ctx) \
+  GRPC_STATS_INC_COUNTER((exec_ctx), GRPC_STATS_COUNTER_HTTP2_PARTIAL_WRITES)
 #define GRPC_STATS_INC_COMBINER_LOCKS_INITIATED(exec_ctx) \
   GRPC_STATS_INC_COUNTER((exec_ctx),                      \
                          GRPC_STATS_COUNTER_COMBINER_LOCKS_INITIATED)
@@ -130,9 +163,12 @@ typedef enum {
 #define GRPC_STATS_INC_COMBINER_LOCKS_OFFLOADED(exec_ctx) \
   GRPC_STATS_INC_COUNTER((exec_ctx),                      \
                          GRPC_STATS_COUNTER_COMBINER_LOCKS_OFFLOADED)
-#define GRPC_STATS_INC_EXECUTOR_SCHEDULED_ITEMS(exec_ctx) \
+#define GRPC_STATS_INC_EXECUTOR_SCHEDULED_SHORT_ITEMS(exec_ctx) \
-  GRPC_STATS_INC_COUNTER((exec_ctx),                      \
+  GRPC_STATS_INC_COUNTER((exec_ctx),                            \
-                         GRPC_STATS_COUNTER_EXECUTOR_SCHEDULED_ITEMS)
+                         GRPC_STATS_COUNTER_EXECUTOR_SCHEDULED_SHORT_ITEMS)
+#define GRPC_STATS_INC_EXECUTOR_SCHEDULED_LONG_ITEMS(exec_ctx) \
+  GRPC_STATS_INC_COUNTER((exec_ctx),                           \
+                         GRPC_STATS_COUNTER_EXECUTOR_SCHEDULED_LONG_ITEMS)
 #define GRPC_STATS_INC_EXECUTOR_SCHEDULED_TO_SELF(exec_ctx) \
   GRPC_STATS_INC_COUNTER((exec_ctx),                        \
                          GRPC_STATS_COUNTER_EXECUTOR_SCHEDULED_TO_SELF)
@@ -141,6 +177,8 @@ typedef enum {
                          GRPC_STATS_COUNTER_EXECUTOR_WAKEUP_INITIATED)
 #define GRPC_STATS_INC_EXECUTOR_QUEUE_DRAINED(exec_ctx) \
   GRPC_STATS_INC_COUNTER((exec_ctx), GRPC_STATS_COUNTER_EXECUTOR_QUEUE_DRAINED)
+#define GRPC_STATS_INC_EXECUTOR_PUSH_RETRIES(exec_ctx) \
+  GRPC_STATS_INC_COUNTER((exec_ctx), GRPC_STATS_COUNTER_EXECUTOR_PUSH_RETRIES)
 #define GRPC_STATS_INC_TCP_WRITE_SIZE(exec_ctx, value) \
   grpc_stats_inc_tcp_write_size((exec_ctx), (int)(value))
 void grpc_stats_inc_tcp_write_size(grpc_exec_ctx *exec_ctx, int x);
@@ -159,10 +197,27 @@ void grpc_stats_inc_tcp_read_offer_iov_size(grpc_exec_ctx *exec_ctx, int x);
 #define GRPC_STATS_INC_HTTP2_SEND_MESSAGE_SIZE(exec_ctx, value) \
   grpc_stats_inc_http2_send_message_size((exec_ctx), (int)(value))
 void grpc_stats_inc_http2_send_message_size(grpc_exec_ctx *exec_ctx, int x);
-extern const int grpc_stats_histo_buckets[6];
+#define GRPC_STATS_INC_HTTP2_SEND_INITIAL_METADATA_PER_WRITE(exec_ctx, value) \
-extern const int grpc_stats_histo_start[6];
+  grpc_stats_inc_http2_send_initial_metadata_per_write((exec_ctx), (int)(value))
-extern const int *const grpc_stats_histo_bucket_boundaries[6];
+void grpc_stats_inc_http2_send_initial_metadata_per_write(
-extern void (*const grpc_stats_inc_histogram[6])(grpc_exec_ctx *exec_ctx,
+    grpc_exec_ctx *exec_ctx, int x);
+#define GRPC_STATS_INC_HTTP2_SEND_MESSAGE_PER_WRITE(exec_ctx, value) \
+  grpc_stats_inc_http2_send_message_per_write((exec_ctx), (int)(value))
+void grpc_stats_inc_http2_send_message_per_write(grpc_exec_ctx *exec_ctx,
+                                                 int x);
+#define GRPC_STATS_INC_HTTP2_SEND_TRAILING_METADATA_PER_WRITE(exec_ctx, value) \
+  grpc_stats_inc_http2_send_trailing_metadata_per_write((exec_ctx),            \
+                                                        (int)(value))
+void grpc_stats_inc_http2_send_trailing_metadata_per_write(
+    grpc_exec_ctx *exec_ctx, int x);
+#define GRPC_STATS_INC_HTTP2_SEND_FLOWCTL_PER_WRITE(exec_ctx, value) \
+  grpc_stats_inc_http2_send_flowctl_per_write((exec_ctx), (int)(value))
+void grpc_stats_inc_http2_send_flowctl_per_write(grpc_exec_ctx *exec_ctx,
                                                  int x);
+extern const int grpc_stats_histo_buckets[10];
+extern const int grpc_stats_histo_start[10];
+extern const int *const grpc_stats_histo_bucket_boundaries[10];
+extern void (*const grpc_stats_inc_histogram[10])(grpc_exec_ctx *exec_ctx,
+                                                  int x);
 
 #endif /* GRPC_CORE_LIB_DEBUG_STATS_DATA_H */

src/core/lib/debug/stats_data.yaml

@@ -54,6 +54,10 @@
   max: 1024
   buckets: 64
   doc: Number of byte segments offered to each syscall_read
+- counter: tcp_backup_pollers_created
+  doc: Number of times a backup poller has been created (this can be expensive)
+- counter: tcp_backup_poller_polls
+  doc: Number of polls performed on the backup poller
 # chttp2
 - counter: http2_op_batches
   doc: Number of batches received by HTTP2 transport
@@ -75,10 +79,36 @@
   max: 16777216
   buckets: 64
   doc: Size of messages received by HTTP2 transport
+- histogram: http2_send_initial_metadata_per_write
+  max: 1024
+  buckets: 64
+  doc: Number of streams initiated written per TCP write
+- histogram: http2_send_message_per_write
+  max: 1024
+  buckets: 64
+  doc: Number of streams whose payload was written per TCP write
+- histogram: http2_send_trailing_metadata_per_write
+  max: 1024
+  buckets: 64
+  doc: Number of streams terminated per TCP write
+- histogram: http2_send_flowctl_per_write
+  max: 1024
+  buckets: 64
+  doc: Number of flow control updates written per TCP write
+- counter: http2_settings_writes
+  doc: Number of settings frames sent
 - counter: http2_pings_sent
   doc: Number of HTTP2 pings sent by process
 - counter: http2_writes_begun
   doc: Number of HTTP2 writes initiated
+- counter: http2_writes_offloaded
+  doc: Number of HTTP2 writes offloaded to the executor from application threads
+- counter: http2_writes_continued
+  doc: Number of HTTP2 writes that finished seeing more data needed to be
+       written
+- counter: http2_partial_writes
+  doc: Number of HTTP2 writes that were made knowing there was still more data
+       to be written (we cap maximum write size to syscall_write)
 # combiner locks
 - counter: combiner_locks_initiated
   doc: Number of combiner lock entries by process
@@ -90,11 +120,18 @@
 - counter: combiner_locks_offloaded
   doc: Number of combiner locks offloaded to different threads
 # executor
-- counter: executor_scheduled_items
+- counter: executor_scheduled_short_items
-  doc: Number of closures scheduled against the executor (gRPC thread pool)
+  doc: Number of finite runtime closures scheduled against the executor
+       (gRPC thread pool)
+- counter: executor_scheduled_long_items
+  doc: Number of potentially infinite runtime closures scheduled against the
+       executor (gRPC thread pool)
 - counter: executor_scheduled_to_self
   doc: Number of closures scheduled by the executor to the executor
 - counter: executor_wakeup_initiated
   doc: Number of thread wakeups initiated within the executor
 - counter: executor_queue_drained
   doc: Number of times an executor queue was drained
+- counter: executor_push_retries
+  doc: Number of times we raced and were forced to retry pushing a closure to
+       the executor

src/core/lib/iomgr/combiner.c

@@ -81,7 +81,8 @@ grpc_combiner *grpc_combiner_create(void) {
   gpr_atm_no_barrier_store(&lock->state, STATE_UNORPHANED);
   gpr_mpscq_init(&lock->queue);
   grpc_closure_list_init(&lock->final_list);
-  GRPC_CLOSURE_INIT(&lock->offload, offload, lock, grpc_executor_scheduler);
+  GRPC_CLOSURE_INIT(&lock->offload, offload, lock,
+                    grpc_executor_scheduler(GRPC_EXECUTOR_SHORT));
   GRPC_COMBINER_TRACE(gpr_log(GPR_DEBUG, "C:%p create", lock));
   return lock;
 }

src/core/lib/iomgr/ev_poll_posix.c

@@ -989,6 +989,10 @@ static grpc_error *pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
       r = grpc_poll_function(pfds, pfd_count, timeout);
       GRPC_SCHEDULING_END_BLOCKING_REGION;
 
+      if (GRPC_TRACER_ON(grpc_polling_trace)) {
+        gpr_log(GPR_DEBUG, "%p poll=%d", pollset, r);
+      }
+
       if (r < 0) {
         if (errno != EINTR) {
           work_combine_error(&error, GRPC_OS_ERROR(errno, "poll"));
@@ -1009,6 +1013,9 @@ static grpc_error *pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
         }
       } else {
         if (pfds[0].revents & POLLIN_CHECK) {
+          if (GRPC_TRACER_ON(grpc_polling_trace)) {
+            gpr_log(GPR_DEBUG, "%p: got_wakeup", pollset);
+          }
           work_combine_error(
               &error, grpc_wakeup_fd_consume_wakeup(&worker.wakeup_fd->fd));
         }
@@ -1016,6 +1023,11 @@ static grpc_error *pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
           if (watchers[i].fd == NULL) {
             fd_end_poll(exec_ctx, &watchers[i], 0, 0, NULL);
           } else {
+            if (GRPC_TRACER_ON(grpc_polling_trace)) {
+              gpr_log(GPR_DEBUG, "%p got_event: %d r:%d w:%d [%d]", pollset,
+                      pfds[i].fd, (pfds[i].revents & POLLIN_CHECK) != 0,
+                      (pfds[i].revents & POLLOUT_CHECK) != 0, pfds[i].revents);
+            }
             fd_end_poll(exec_ctx, &watchers[i], pfds[i].revents & POLLIN_CHECK,
                         pfds[i].revents & POLLOUT_CHECK, pollset);
           }

src/core/lib/iomgr/executor.c

@@ -40,6 +40,7 @@ typedef struct {
   grpc_closure_list elems;
   size_t depth;
   bool shutdown;
+  bool queued_long_job;
   gpr_thd_id id;
 } thread_state;
 
@@ -50,6 +51,9 @@ static gpr_spinlock g_adding_thread_lock = GPR_SPINLOCK_STATIC_INITIALIZER;
 
 GPR_TLS_DECL(g_this_thread_state);
 
+static grpc_tracer_flag executor_trace =
+    GRPC_TRACER_INITIALIZER(false, "executor");
+
 static void executor_thread(void *arg);
 
 static size_t run_closures(grpc_exec_ctx *exec_ctx, grpc_closure_list list) {
@@ -59,6 +63,14 @@ static size_t run_closures(grpc_exec_ctx *exec_ctx, grpc_closure_list list) {
   while (c != NULL) {
     grpc_closure *next = c->next_data.next;
     grpc_error *error = c->error_data.error;
+    if (GRPC_TRACER_ON(executor_trace)) {
+#ifndef NDEBUG
+      gpr_log(GPR_DEBUG, "EXECUTOR: run %p [created by %s:%d]", c,
+              c->file_created, c->line_created);
+#else
+      gpr_log(GPR_DEBUG, "EXECUTOR: run %p", c);
+#endif
+    }
 #ifndef NDEBUG
     c->scheduled = false;
 #endif
@@ -66,6 +78,7 @@ static size_t run_closures(grpc_exec_ctx *exec_ctx, grpc_closure_list list) {
     GRPC_ERROR_UNREF(error);
     c = next;
     n++;
+    grpc_exec_ctx_flush(exec_ctx);
   }
 
   return n;
@@ -121,6 +134,7 @@ void grpc_executor_set_threading(grpc_exec_ctx *exec_ctx, bool threading) {
 }
 
 void grpc_executor_init(grpc_exec_ctx *exec_ctx) {
+  grpc_register_tracer(&executor_trace);
   gpr_atm_no_barrier_store(&g_cur_threads, 0);
   grpc_executor_set_threading(exec_ctx, true);
 }
@@ -138,12 +152,21 @@ static void executor_thread(void *arg) {
 
   size_t subtract_depth = 0;
   for (;;) {
+    if (GRPC_TRACER_ON(executor_trace)) {
+      gpr_log(GPR_DEBUG, "EXECUTOR[%d]: step (sub_depth=%" PRIdPTR ")",
+              (int)(ts - g_thread_state), subtract_depth);
+    }
     gpr_mu_lock(&ts->mu);
     ts->depth -= subtract_depth;
     while (grpc_closure_list_empty(ts->elems) && !ts->shutdown) {
+      ts->queued_long_job = false;
       gpr_cv_wait(&ts->cv, &ts->mu, gpr_inf_future(GPR_CLOCK_REALTIME));
     }
     if (ts->shutdown) {
+      if (GRPC_TRACER_ON(executor_trace)) {
+        gpr_log(GPR_DEBUG, "EXECUTOR[%d]: shutdown",
+                (int)(ts - g_thread_state));
+      }
       gpr_mu_unlock(&ts->mu);
       break;
     }
@@ -151,52 +174,128 @@ static void executor_thread(void *arg) {
     grpc_closure_list exec = ts->elems;
     ts->elems = (grpc_closure_list)GRPC_CLOSURE_LIST_INIT;
     gpr_mu_unlock(&ts->mu);
+    if (GRPC_TRACER_ON(executor_trace)) {
+      gpr_log(GPR_DEBUG, "EXECUTOR[%d]: execute", (int)(ts - g_thread_state));
+    }
 
     subtract_depth = run_closures(&exec_ctx, exec);
-    grpc_exec_ctx_flush(&exec_ctx);
   }
   grpc_exec_ctx_finish(&exec_ctx);
 }
 
 static void executor_push(grpc_exec_ctx *exec_ctx, grpc_closure *closure,
-                          grpc_error *error) {
+                          grpc_error *error, bool is_short) {
-  size_t cur_thread_count = (size_t)gpr_atm_no_barrier_load(&g_cur_threads);
+  bool retry_push;
-  GRPC_STATS_INC_EXECUTOR_SCHEDULED_ITEMS(exec_ctx);
+  if (is_short) {
-  if (cur_thread_count == 0) {
+    GRPC_STATS_INC_EXECUTOR_SCHEDULED_SHORT_ITEMS(exec_ctx);
-    grpc_closure_list_append(&exec_ctx->closure_list, closure, error);
-    return;
-  }
-  thread_state *ts = (thread_state *)gpr_tls_get(&g_this_thread_state);
-  if (ts == NULL) {
-    ts = &g_thread_state[GPR_HASH_POINTER(exec_ctx, cur_thread_count)];
   } else {
-    GRPC_STATS_INC_EXECUTOR_SCHEDULED_TO_SELF(exec_ctx);
+    GRPC_STATS_INC_EXECUTOR_SCHEDULED_LONG_ITEMS(exec_ctx);
-  }
-  gpr_mu_lock(&ts->mu);
-  if (grpc_closure_list_empty(ts->elems)) {
-    GRPC_STATS_INC_EXECUTOR_WAKEUP_INITIATED(exec_ctx);
-    gpr_cv_signal(&ts->cv);
   }
-  grpc_closure_list_append(&ts->elems, closure, error);
+  do {
-  ts->depth++;
+    retry_push = false;
-  bool try_new_thread = ts->depth > MAX_DEPTH &&
+    size_t cur_thread_count = (size_t)gpr_atm_no_barrier_load(&g_cur_threads);
-                        cur_thread_count < g_max_threads && !ts->shutdown;
+    if (cur_thread_count == 0) {
-  gpr_mu_unlock(&ts->mu);
+      if (GRPC_TRACER_ON(executor_trace)) {
-  if (try_new_thread && gpr_spinlock_trylock(&g_adding_thread_lock)) {
+#ifndef NDEBUG
-    cur_thread_count = (size_t)gpr_atm_no_barrier_load(&g_cur_threads);
+        gpr_log(GPR_DEBUG, "EXECUTOR: schedule %p (created %s:%d) inline",
-    if (cur_thread_count < g_max_threads) {
+                closure, closure->file_created, closure->line_created);
-      gpr_atm_no_barrier_store(&g_cur_threads, cur_thread_count + 1);
+#else
-
+        gpr_log(GPR_DEBUG, "EXECUTOR: schedule %p inline", closure);
-      gpr_thd_options opt = gpr_thd_options_default();
+#endif
-      gpr_thd_options_set_joinable(&opt);
+      }
-      gpr_thd_new(&g_thread_state[cur_thread_count].id, executor_thread,
+      grpc_closure_list_append(&exec_ctx->closure_list, closure, error);
-                  &g_thread_state[cur_thread_count], &opt);
+      return;
     }
-    gpr_spinlock_unlock(&g_adding_thread_lock);
+    thread_state *ts = (thread_state *)gpr_tls_get(&g_this_thread_state);
-  }
+    if (ts == NULL) {
+      ts = &g_thread_state[GPR_HASH_POINTER(exec_ctx, cur_thread_count)];
+    } else {
+      GRPC_STATS_INC_EXECUTOR_SCHEDULED_TO_SELF(exec_ctx);
+    }
+    thread_state *orig_ts = ts;
+
+    bool try_new_thread;
+    for (;;) {
+      if (GRPC_TRACER_ON(executor_trace)) {
+#ifndef NDEBUG
+        gpr_log(
+            GPR_DEBUG,
+            "EXECUTOR: try to schedule %p (%s) (created %s:%d) to thread %d",
+            closure, is_short ? "short" : "long", closure->file_created,
+            closure->line_created, (int)(ts - g_thread_state));
+#else
+        gpr_log(GPR_DEBUG, "EXECUTOR: try to schedule %p (%s) to thread %d",
+                closure, is_short ? "short" : "long",
+                (int)(ts - g_thread_state));
+#endif
+      }
+      gpr_mu_lock(&ts->mu);
+      if (ts->queued_long_job) {
+        // if there's a long job queued, we never queue anything else to this
+        // queue (since long jobs can take 'infinite' time and we need to
+        // guarantee no starvation)
+        // ... spin through queues and try again
+        gpr_mu_unlock(&ts->mu);
+        size_t idx = (size_t)(ts - g_thread_state);
+        ts = &g_thread_state[(idx + 1) % cur_thread_count];
+        if (ts == orig_ts) {
+          retry_push = true;
+          try_new_thread = true;
+          break;
+        }
+        continue;
+      }
+      if (grpc_closure_list_empty(ts->elems)) {
+        GRPC_STATS_INC_EXECUTOR_WAKEUP_INITIATED(exec_ctx);
+        gpr_cv_signal(&ts->cv);
+      }
+      grpc_closure_list_append(&ts->elems, closure, error);
+      ts->depth++;
+      try_new_thread = ts->depth > MAX_DEPTH &&
+                       cur_thread_count < g_max_threads && !ts->shutdown;
+      if (!is_short) ts->queued_long_job = true;
+      gpr_mu_unlock(&ts->mu);
+      break;
+    }
+    if (try_new_thread && gpr_spinlock_trylock(&g_adding_thread_lock)) {
+      cur_thread_count = (size_t)gpr_atm_no_barrier_load(&g_cur_threads);
+      if (cur_thread_count < g_max_threads) {
+        gpr_atm_no_barrier_store(&g_cur_threads, cur_thread_count + 1);
+
+        gpr_thd_options opt = gpr_thd_options_default();
+        gpr_thd_options_set_joinable(&opt);
+        gpr_thd_new(&g_thread_state[cur_thread_count].id, executor_thread,
+                    &g_thread_state[cur_thread_count], &opt);
+      }
+      gpr_spinlock_unlock(&g_adding_thread_lock);
+    }
+    if (retry_push) {
+      GRPC_STATS_INC_EXECUTOR_PUSH_RETRIES(exec_ctx);
+    }
+  } while (retry_push);
 }
 
-static const grpc_closure_scheduler_vtable executor_vtable = {
+static void executor_push_short(grpc_exec_ctx *exec_ctx, grpc_closure *closure,
-    executor_push, executor_push, "executor"};
+                                grpc_error *error) {
-static grpc_closure_scheduler executor_scheduler = {&executor_vtable};
+  executor_push(exec_ctx, closure, error, true);
-grpc_closure_scheduler *grpc_executor_scheduler = &executor_scheduler;
+}
+
+static void executor_push_long(grpc_exec_ctx *exec_ctx, grpc_closure *closure,
+                               grpc_error *error) {
+  executor_push(exec_ctx, closure, error, false);
+}
+
+static const grpc_closure_scheduler_vtable executor_vtable_short = {
+    executor_push_short, executor_push_short, "executor"};
+static grpc_closure_scheduler executor_scheduler_short = {
+    &executor_vtable_short};
+
+static const grpc_closure_scheduler_vtable executor_vtable_long = {
+    executor_push_long, executor_push_long, "executor"};
+static grpc_closure_scheduler executor_scheduler_long = {&executor_vtable_long};
+
+grpc_closure_scheduler *grpc_executor_scheduler(
+    grpc_executor_job_length length) {
+  return length == GRPC_EXECUTOR_SHORT ? &executor_scheduler_short
+                                       : &executor_scheduler_long;
+}

src/core/lib/iomgr/executor.h

@@ -21,6 +21,11 @@
 
 #include "src/core/lib/iomgr/closure.h"
 
+typedef enum {
+  GRPC_EXECUTOR_SHORT,
+  GRPC_EXECUTOR_LONG
+} grpc_executor_job_length;
+
 /** Initialize the global executor.
  *
  * This mechanism is meant to outsource work (grpc_closure instances) to a
@@ -28,7 +33,7 @@
  * non-blocking solution available. */
 void grpc_executor_init(grpc_exec_ctx *exec_ctx);
 
-extern grpc_closure_scheduler *grpc_executor_scheduler;
+grpc_closure_scheduler *grpc_executor_scheduler(grpc_executor_job_length);
 
 /** Shutdown the executor, running all pending work as part of the call */
 void grpc_executor_shutdown(grpc_exec_ctx *exec_ctx);

src/core/lib/iomgr/resolve_address_posix.c

@@ -177,7 +177,7 @@ static void resolve_address_impl(grpc_exec_ctx *exec_ctx, const char *name,
                                  grpc_resolved_addresses **addrs) {
   request *r = (request *)gpr_malloc(sizeof(request));
   GRPC_CLOSURE_INIT(&r->request_closure, do_request_thread, r,
-                    grpc_executor_scheduler);
+                    grpc_executor_scheduler(GRPC_EXECUTOR_SHORT));
   r->name = gpr_strdup(name);
   r->default_port = gpr_strdup(default_port);
   r->on_done = on_done;

src/core/lib/iomgr/resolve_address_windows.c

@@ -159,7 +159,7 @@ static void resolve_address_impl(grpc_exec_ctx *exec_ctx, const char *name,
                                  grpc_resolved_addresses **addresses) {
   request *r = gpr_malloc(sizeof(request));
   GRPC_CLOSURE_INIT(&r->request_closure, do_request_thread, r,
-                    grpc_executor_scheduler);
+                    grpc_executor_scheduler(GRPC_EXECUTOR_SHORT));
   r->name = gpr_strdup(name);
   r->default_port = gpr_strdup(default_port);
   r->on_done = on_done;

src/core/lib/iomgr/tcp_posix.c

@@ -43,6 +43,7 @@
 #include "src/core/lib/debug/stats.h"
 #include "src/core/lib/debug/trace.h"
 #include "src/core/lib/iomgr/ev_posix.h"
+#include "src/core/lib/iomgr/executor.h"
 #include "src/core/lib/profiling/timers.h"
 #include "src/core/lib/slice/slice_internal.h"
 #include "src/core/lib/slice/slice_string_helpers.h"
@@ -90,8 +91,8 @@ typedef struct {
   grpc_closure *release_fd_cb;
   int *release_fd;
 
-  grpc_closure read_closure;
+  grpc_closure read_done_closure;
-  grpc_closure write_closure;
+  grpc_closure write_done_closure;
 
   char *peer_string;
 
@@ -99,6 +100,148 @@ typedef struct {
   grpc_resource_user_slice_allocator slice_allocator;
 } grpc_tcp;
 
+typedef struct backup_poller {
+  gpr_mu *pollset_mu;
+  grpc_closure run_poller;
+} backup_poller;
+
+#define BACKUP_POLLER_POLLSET(b) ((grpc_pollset *)((b) + 1))
+
+static gpr_atm g_uncovered_notifications_pending;
+static gpr_atm g_backup_poller; /* backup_poller* */
+
+static void tcp_handle_read(grpc_exec_ctx *exec_ctx, void *arg /* grpc_tcp */,
+                            grpc_error *error);
+static void tcp_handle_write(grpc_exec_ctx *exec_ctx, void *arg /* grpc_tcp */,
+                             grpc_error *error);
+static void tcp_drop_uncovered_then_handle_write(grpc_exec_ctx *exec_ctx,
+                                                 void *arg /* grpc_tcp */,
+                                                 grpc_error *error);
+
+static void done_poller(grpc_exec_ctx *exec_ctx, void *bp,
+                        grpc_error *error_ignored) {
+  backup_poller *p = (backup_poller *)bp;
+  if (GRPC_TRACER_ON(grpc_tcp_trace)) {
+    gpr_log(GPR_DEBUG, "BACKUP_POLLER:%p destroy", p);
+  }
+  grpc_pollset_destroy(exec_ctx, BACKUP_POLLER_POLLSET(p));
+  gpr_free(p);
+}
+
+static void run_poller(grpc_exec_ctx *exec_ctx, void *bp,
+                       grpc_error *error_ignored) {
+  backup_poller *p = (backup_poller *)bp;
+  if (GRPC_TRACER_ON(grpc_tcp_trace)) {
+    gpr_log(GPR_DEBUG, "BACKUP_POLLER:%p run", p);
+  }
+  gpr_mu_lock(p->pollset_mu);
+  gpr_timespec now = gpr_now(GPR_CLOCK_MONOTONIC);
+  gpr_timespec deadline =
+      gpr_time_add(now, gpr_time_from_seconds(10, GPR_TIMESPAN));
+  GRPC_STATS_INC_TCP_BACKUP_POLLER_POLLS(exec_ctx);
+  GRPC_LOG_IF_ERROR("backup_poller:pollset_work",
+                    grpc_pollset_work(exec_ctx, BACKUP_POLLER_POLLSET(p), NULL,
+                                      now, deadline));
+  gpr_mu_unlock(p->pollset_mu);
+  /* last "uncovered" notification is the ref that keeps us polling, if we get
+   * there try a cas to release it */
+  if (gpr_atm_no_barrier_load(&g_uncovered_notifications_pending) == 1 &&
+      gpr_atm_full_cas(&g_uncovered_notifications_pending, 1, 0)) {
+    gpr_mu_lock(p->pollset_mu);
+    bool cas_ok = gpr_atm_full_cas(&g_backup_poller, (gpr_atm)p, 0);
+    if (GRPC_TRACER_ON(grpc_tcp_trace)) {
+      gpr_log(GPR_DEBUG, "BACKUP_POLLER:%p done cas_ok=%d", p, cas_ok);
+    }
+    gpr_mu_unlock(p->pollset_mu);
+    if (GRPC_TRACER_ON(grpc_tcp_trace)) {
+      gpr_log(GPR_DEBUG, "BACKUP_POLLER:%p shutdown", p);
+    }
+    grpc_pollset_shutdown(exec_ctx, BACKUP_POLLER_POLLSET(p),
+                          GRPC_CLOSURE_INIT(&p->run_poller, done_poller, p,
+                                            grpc_schedule_on_exec_ctx));
+  } else {
+    if (GRPC_TRACER_ON(grpc_tcp_trace)) {
+      gpr_log(GPR_DEBUG, "BACKUP_POLLER:%p reschedule", p);
+    }
+    GRPC_CLOSURE_SCHED(exec_ctx, &p->run_poller, GRPC_ERROR_NONE);
+  }
+}
+
+static void drop_uncovered(grpc_exec_ctx *exec_ctx, grpc_tcp *tcp) {
+  backup_poller *p = (backup_poller *)gpr_atm_acq_load(&g_backup_poller);
+  gpr_atm old_count =
+      gpr_atm_no_barrier_fetch_add(&g_uncovered_notifications_pending, -1);
+  if (GRPC_TRACER_ON(grpc_tcp_trace)) {
+    gpr_log(GPR_DEBUG, "BACKUP_POLLER:%p uncover cnt %d->%d", p, (int)old_count,
+            (int)old_count - 1);
+  }
+  GPR_ASSERT(old_count != 1);
+}
+
+static void cover_self(grpc_exec_ctx *exec_ctx, grpc_tcp *tcp) {
+  backup_poller *p;
+  gpr_atm old_count =
+      gpr_atm_no_barrier_fetch_add(&g_uncovered_notifications_pending, 2);
+  if (GRPC_TRACER_ON(grpc_tcp_trace)) {
+    gpr_log(GPR_DEBUG, "BACKUP_POLLER: cover cnt %d->%d", (int)old_count,
+            2 + (int)old_count);
+  }
+  if (old_count == 0) {
+    GRPC_STATS_INC_TCP_BACKUP_POLLERS_CREATED(exec_ctx);
+    p = (backup_poller *)gpr_malloc(sizeof(*p) + grpc_pollset_size());
+    if (GRPC_TRACER_ON(grpc_tcp_trace)) {
+      gpr_log(GPR_DEBUG, "BACKUP_POLLER:%p create", p);
+    }
+    grpc_pollset_init(BACKUP_POLLER_POLLSET(p), &p->pollset_mu);
+    gpr_atm_rel_store(&g_backup_poller, (gpr_atm)p);
+    GRPC_CLOSURE_SCHED(
+        exec_ctx,
+        GRPC_CLOSURE_INIT(&p->run_poller, run_poller, p,
+                          grpc_executor_scheduler(GRPC_EXECUTOR_LONG)),
+        GRPC_ERROR_NONE);
+  } else {
+    while ((p = (backup_poller *)gpr_atm_acq_load(&g_backup_poller)) == NULL) {
+      // spin waiting for backup poller
+    }
+  }
+  if (GRPC_TRACER_ON(grpc_tcp_trace)) {
+    gpr_log(GPR_DEBUG, "BACKUP_POLLER:%p add %p", p, tcp);
+  }
+  grpc_pollset_add_fd(exec_ctx, BACKUP_POLLER_POLLSET(p), tcp->em_fd);
+  if (old_count != 0) {
+    drop_uncovered(exec_ctx, tcp);
+  }
+}
+
+static void notify_on_read(grpc_exec_ctx *exec_ctx, grpc_tcp *tcp) {
+  if (GRPC_TRACER_ON(grpc_tcp_trace)) {
+    gpr_log(GPR_DEBUG, "TCP:%p notify_on_read", tcp);
+  }
+  GRPC_CLOSURE_INIT(&tcp->read_done_closure, tcp_handle_read, tcp,
+                    grpc_schedule_on_exec_ctx);
+  grpc_fd_notify_on_read(exec_ctx, tcp->em_fd, &tcp->read_done_closure);
+}
+
+static void notify_on_write(grpc_exec_ctx *exec_ctx, grpc_tcp *tcp) {
+  if (GRPC_TRACER_ON(grpc_tcp_trace)) {
+    gpr_log(GPR_DEBUG, "TCP:%p notify_on_write", tcp);
+  }
+  cover_self(exec_ctx, tcp);
+  GRPC_CLOSURE_INIT(&tcp->write_done_closure,
+                    tcp_drop_uncovered_then_handle_write, tcp,
+                    grpc_schedule_on_exec_ctx);
+  grpc_fd_notify_on_write(exec_ctx, tcp->em_fd, &tcp->write_done_closure);
+}
+
+static void tcp_drop_uncovered_then_handle_write(grpc_exec_ctx *exec_ctx,
+                                                 void *arg, grpc_error *error) {
+  if (GRPC_TRACER_ON(grpc_tcp_trace)) {
+    gpr_log(GPR_DEBUG, "TCP:%p got_write: %s", arg, grpc_error_string(error));
+  }
+  drop_uncovered(exec_ctx, (grpc_tcp *)arg);
+  tcp_handle_write(exec_ctx, arg, error);
+}
+
 static void add_to_estimate(grpc_tcp *tcp, size_t bytes) {
   tcp->bytes_read_this_round += (double)bytes;
 }
@@ -214,6 +357,7 @@ static void call_read_cb(grpc_exec_ctx *exec_ctx, grpc_tcp *tcp,
   grpc_closure *cb = tcp->read_cb;
 
   if (GRPC_TRACER_ON(grpc_tcp_trace)) {
+    gpr_log(GPR_DEBUG, "TCP:%p call_cb %p %p:%p", tcp, cb, cb->cb, cb->cb_arg);
     size_t i;
     const char *str = grpc_error_string(error);
     gpr_log(GPR_DEBUG, "read: error=%s", str);
@@ -271,7 +415,7 @@ static void tcp_do_read(grpc_exec_ctx *exec_ctx, grpc_tcp *tcp) {
     if (errno == EAGAIN) {
       finish_estimate(tcp);
       /* We've consumed the edge, request a new one */
-      grpc_fd_notify_on_read(exec_ctx, tcp->em_fd, &tcp->read_closure);
+      notify_on_read(exec_ctx, tcp);
     } else {
       grpc_slice_buffer_reset_and_unref_internal(exec_ctx,
                                                  tcp->incoming_buffer);
@@ -308,6 +452,10 @@ static void tcp_do_read(grpc_exec_ctx *exec_ctx, grpc_tcp *tcp) {
 static void tcp_read_allocation_done(grpc_exec_ctx *exec_ctx, void *tcpp,
                                      grpc_error *error) {
   grpc_tcp *tcp = (grpc_tcp *)tcpp;
+  if (GRPC_TRACER_ON(grpc_tcp_trace)) {
+    gpr_log(GPR_DEBUG, "TCP:%p read_allocation_done: %s", tcp,
+            grpc_error_string(error));
+  }
   if (error != GRPC_ERROR_NONE) {
     grpc_slice_buffer_reset_and_unref_internal(exec_ctx, tcp->incoming_buffer);
     grpc_slice_buffer_reset_and_unref_internal(exec_ctx,
@@ -323,9 +471,15 @@ static void tcp_continue_read(grpc_exec_ctx *exec_ctx, grpc_tcp *tcp) {
   size_t target_read_size = get_target_read_size(tcp);
   if (tcp->incoming_buffer->length < target_read_size &&
       tcp->incoming_buffer->count < MAX_READ_IOVEC) {
+    if (GRPC_TRACER_ON(grpc_tcp_trace)) {
+      gpr_log(GPR_DEBUG, "TCP:%p alloc_slices", tcp);
+    }
     grpc_resource_user_alloc_slices(exec_ctx, &tcp->slice_allocator,
                                     target_read_size, 1, tcp->incoming_buffer);
   } else {
+    if (GRPC_TRACER_ON(grpc_tcp_trace)) {
+      gpr_log(GPR_DEBUG, "TCP:%p do_read", tcp);
+    }
     tcp_do_read(exec_ctx, tcp);
   }
 }
@@ -334,6 +488,9 @@ static void tcp_handle_read(grpc_exec_ctx *exec_ctx, void *arg /* grpc_tcp */,
                             grpc_error *error) {
   grpc_tcp *tcp = (grpc_tcp *)arg;
   GPR_ASSERT(!tcp->finished_edge);
+  if (GRPC_TRACER_ON(grpc_tcp_trace)) {
+    gpr_log(GPR_DEBUG, "TCP:%p got_read: %s", tcp, grpc_error_string(error));
+  }
 
   if (error != GRPC_ERROR_NONE) {
     grpc_slice_buffer_reset_and_unref_internal(exec_ctx, tcp->incoming_buffer);
@@ -357,9 +514,9 @@ static void tcp_read(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
   TCP_REF(tcp, "read");
   if (tcp->finished_edge) {
     tcp->finished_edge = false;
-    grpc_fd_notify_on_read(exec_ctx, tcp->em_fd, &tcp->read_closure);
+    notify_on_read(exec_ctx, tcp);
   } else {
-    GRPC_CLOSURE_SCHED(exec_ctx, &tcp->read_closure, GRPC_ERROR_NONE);
+    GRPC_CLOSURE_SCHED(exec_ctx, &tcp->read_done_closure, GRPC_ERROR_NONE);
   }
 }
 
@@ -472,7 +629,7 @@ static void tcp_handle_write(grpc_exec_ctx *exec_ctx, void *arg /* grpc_tcp */,
     if (GRPC_TRACER_ON(grpc_tcp_trace)) {
       gpr_log(GPR_DEBUG, "write: delayed");
     }
-    grpc_fd_notify_on_write(exec_ctx, tcp->em_fd, &tcp->write_closure);
+    notify_on_write(exec_ctx, tcp);
   } else {
     cb = tcp->write_cb;
     tcp->write_cb = NULL;
@@ -525,7 +682,7 @@ static void tcp_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
     if (GRPC_TRACER_ON(grpc_tcp_trace)) {
       gpr_log(GPR_DEBUG, "write: delayed");
     }
-    grpc_fd_notify_on_write(exec_ctx, tcp->em_fd, &tcp->write_closure);
+    notify_on_write(exec_ctx, tcp);
   } else {
     if (GRPC_TRACER_ON(grpc_tcp_trace)) {
       const char *str = grpc_error_string(error);
@@ -602,7 +759,7 @@ grpc_endpoint *grpc_tcp_create(grpc_exec_ctx *exec_ctx, grpc_fd *em_fd,
                  strcmp(channel_args->args[i].key, GRPC_ARG_RESOURCE_QUOTA)) {
         grpc_resource_quota_unref_internal(exec_ctx, resource_quota);
         resource_quota = grpc_resource_quota_ref_internal(
-            channel_args->args[i].value.pointer.p);
+            (grpc_resource_quota *)channel_args->args[i].value.pointer.p);
       }
     }
   }
@@ -631,10 +788,6 @@ grpc_endpoint *grpc_tcp_create(grpc_exec_ctx *exec_ctx, grpc_fd *em_fd,
   gpr_ref_init(&tcp->refcount, 1);
   gpr_atm_no_barrier_store(&tcp->shutdown_count, 0);
   tcp->em_fd = em_fd;
-  GRPC_CLOSURE_INIT(&tcp->read_closure, tcp_handle_read, tcp,
-                    grpc_schedule_on_exec_ctx);
-  GRPC_CLOSURE_INIT(&tcp->write_closure, tcp_handle_write, tcp,
-                    grpc_schedule_on_exec_ctx);
   grpc_slice_buffer_init(&tcp->last_read_buffer);
   tcp->resource_user = grpc_resource_user_create(resource_quota, peer_string);
   grpc_resource_user_slice_allocator_init(

src/core/lib/security/transport/security_handshaker.c

@@ -128,13 +128,11 @@ static void security_handshake_failed_locked(grpc_exec_ctx *exec_ctx,
   GRPC_CLOSURE_SCHED(exec_ctx, h->on_handshake_done, error);
 }
 
-static void on_peer_checked(grpc_exec_ctx *exec_ctx, void *arg,
+static void on_peer_checked_inner(grpc_exec_ctx *exec_ctx,
-                            grpc_error *error) {
+                                  security_handshaker *h, grpc_error *error) {
-  security_handshaker *h = arg;
-  gpr_mu_lock(&h->mu);
   if (error != GRPC_ERROR_NONE || h->shutdown) {
     security_handshake_failed_locked(exec_ctx, h, GRPC_ERROR_REF(error));
-    goto done;
+    return;
   }
   // Create zero-copy frame protector, if implemented.
   tsi_zero_copy_grpc_protector *zero_copy_protector = NULL;
@@ -146,7 +144,7 @@ static void on_peer_checked(grpc_exec_ctx *exec_ctx, void *arg,
             "Zero-copy frame protector creation failed"),
         result);
     security_handshake_failed_locked(exec_ctx, h, error);
-    goto done;
+    return;
   }
   // Create frame protector if zero-copy frame protector is NULL.
   tsi_frame_protector *protector = NULL;
@@ -158,7 +156,7 @@ static void on_peer_checked(grpc_exec_ctx *exec_ctx, void *arg,
                                             "Frame protector creation failed"),
                                         result);
       security_handshake_failed_locked(exec_ctx, h, error);
-      goto done;
+      return;
     }
   }
   // Get unused bytes.
@@ -192,7 +190,13 @@ static void on_peer_checked(grpc_exec_ctx *exec_ctx, void *arg,
   // Set shutdown to true so that subsequent calls to
   // security_handshaker_shutdown() do nothing.
   h->shutdown = true;
-done:
+}
+
+static void on_peer_checked(grpc_exec_ctx *exec_ctx, void *arg,
+                            grpc_error *error) {
+  security_handshaker *h = (security_handshaker *)arg;
+  gpr_mu_lock(&h->mu);
+  on_peer_checked_inner(exec_ctx, h, error);
   gpr_mu_unlock(&h->mu);
   security_handshaker_unref(exec_ctx, h);
 }
@@ -254,7 +258,7 @@ static grpc_error *on_handshake_next_done_locked(
 static void on_handshake_next_done_grpc_wrapper(
     tsi_result result, void *user_data, const unsigned char *bytes_to_send,
     size_t bytes_to_send_size, tsi_handshaker_result *handshaker_result) {
-  security_handshaker *h = user_data;
+  security_handshaker *h = (security_handshaker *)user_data;
   // This callback will be invoked by TSI in a non-grpc thread, so it's
   // safe to create our own exec_ctx here.
   grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
@@ -296,7 +300,7 @@ static grpc_error *do_handshaker_next_locked(
 
 static void on_handshake_data_received_from_peer(grpc_exec_ctx *exec_ctx,
                                                  void *arg, grpc_error *error) {
-  security_handshaker *h = arg;
+  security_handshaker *h = (security_handshaker *)arg;
   gpr_mu_lock(&h->mu);
   if (error != GRPC_ERROR_NONE || h->shutdown) {
     security_handshake_failed_locked(
@@ -313,7 +317,8 @@ static void on_handshake_data_received_from_peer(grpc_exec_ctx *exec_ctx,
     bytes_received_size += GRPC_SLICE_LENGTH(h->args->read_buffer->slices[i]);
   }
   if (bytes_received_size > h->handshake_buffer_size) {
-    h->handshake_buffer = gpr_realloc(h->handshake_buffer, bytes_received_size);
+    h->handshake_buffer =
+        (uint8_t *)gpr_realloc(h->handshake_buffer, bytes_received_size);
     h->handshake_buffer_size = bytes_received_size;
   }
   size_t offset = 0;
@@ -338,7 +343,7 @@ static void on_handshake_data_received_from_peer(grpc_exec_ctx *exec_ctx,
 
 static void on_handshake_data_sent_to_peer(grpc_exec_ctx *exec_ctx, void *arg,
                                            grpc_error *error) {
-  security_handshaker *h = arg;
+  security_handshaker *h = (security_handshaker *)arg;
   gpr_mu_lock(&h->mu);
   if (error != GRPC_ERROR_NONE || h->shutdown) {
     security_handshake_failed_locked(
@@ -415,14 +420,15 @@ static const grpc_handshaker_vtable security_handshaker_vtable = {
 static grpc_handshaker *security_handshaker_create(
     grpc_exec_ctx *exec_ctx, tsi_handshaker *handshaker,
     grpc_security_connector *connector) {
-  security_handshaker *h = gpr_zalloc(sizeof(security_handshaker));
+  security_handshaker *h =
+      (security_handshaker *)gpr_zalloc(sizeof(security_handshaker));
   grpc_handshaker_init(&security_handshaker_vtable, &h->base);
   h->handshaker = handshaker;
   h->connector = GRPC_SECURITY_CONNECTOR_REF(connector, "handshake");
   gpr_mu_init(&h->mu);
   gpr_ref_init(&h->refs, 1);
   h->handshake_buffer_size = GRPC_INITIAL_HANDSHAKE_BUFFER_SIZE;
-  h->handshake_buffer = gpr_malloc(h->handshake_buffer_size);
+  h->handshake_buffer = (uint8_t *)gpr_malloc(h->handshake_buffer_size);
   GRPC_CLOSURE_INIT(&h->on_handshake_data_sent_to_peer,
                     on_handshake_data_sent_to_peer, h,
                     grpc_schedule_on_exec_ctx);
@@ -465,7 +471,7 @@ static const grpc_handshaker_vtable fail_handshaker_vtable = {
     fail_handshaker_do_handshake};
 
 static grpc_handshaker *fail_handshaker_create() {
-  grpc_handshaker *h = gpr_malloc(sizeof(*h));
+  grpc_handshaker *h = (grpc_handshaker *)gpr_malloc(sizeof(*h));
   grpc_handshaker_init(&fail_handshaker_vtable, h);
   return h;
 }

src/core/lib/surface/server.c

@@ -1117,9 +1117,11 @@ void grpc_server_start(grpc_server *server) {
 
   server_ref(server);
   server->starting = true;
-  GRPC_CLOSURE_SCHED(&exec_ctx, GRPC_CLOSURE_CREATE(start_listeners, server,
+  GRPC_CLOSURE_SCHED(
-                                                    grpc_executor_scheduler),
+      &exec_ctx,
-                     GRPC_ERROR_NONE);
+      GRPC_CLOSURE_CREATE(start_listeners, server,
+                          grpc_executor_scheduler(GRPC_EXECUTOR_SHORT)),
+      GRPC_ERROR_NONE);
 
   grpc_exec_ctx_finish(&exec_ctx);
 }

src/core/lib/transport/transport.c

@@ -72,7 +72,8 @@ void grpc_stream_unref(grpc_exec_ctx *exec_ctx,
          cope with.
          Throw this over to the executor (on a core-owned thread) and process it
          there. */
-      refcount->destroy.scheduler = grpc_executor_scheduler;
+      refcount->destroy.scheduler =
+          grpc_executor_scheduler(GRPC_EXECUTOR_SHORT);
     }
     GRPC_CLOSURE_SCHED(exec_ctx, &refcount->destroy, GRPC_ERROR_NONE);
   }

test/core/bad_client/bad_client.c

@@ -45,18 +45,18 @@ typedef struct {
 } thd_args;
 
 static void thd_func(void *arg) {
-  thd_args *a = arg;
+  thd_args *a = (thd_args *)arg;
   a->validator(a->server, a->cq, a->registered_method);
   gpr_event_set(&a->done_thd, (void *)1);
 }
 
 static void done_write(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) {
-  thd_args *a = arg;
+  thd_args *a = (thd_args *)arg;
   gpr_event_set(&a->done_write, (void *)1);
 }
 
 static void server_setup_transport(void *ts, grpc_transport *transport) {
-  thd_args *a = ts;
+  thd_args *a = (thd_args *)ts;
   grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
   grpc_server_setup_transport(&exec_ctx, a->server, transport, NULL,
                               grpc_server_get_channel_args(a->server));
@@ -64,7 +64,7 @@ static void server_setup_transport(void *ts, grpc_transport *transport) {
 }
 
 static void read_done(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) {
-  gpr_event *read_done = arg;
+  gpr_event *read_done = (gpr_event *)arg;
   gpr_event_set(read_done, (void *)1);
 }
 

test/core/end2end/bad_server_response_test.c

@@ -136,7 +136,7 @@ static void on_connect(grpc_exec_ctx *exec_ctx, void *arg, grpc_endpoint *tcp,
                        grpc_pollset *accepting_pollset,
                        grpc_tcp_server_acceptor *acceptor) {
   gpr_free(acceptor);
-  test_tcp_server *server = arg;
+  test_tcp_server *server = (test_tcp_server *)arg;
   GRPC_CLOSURE_INIT(&on_read, handle_read, NULL, grpc_schedule_on_exec_ctx);
   GRPC_CLOSURE_INIT(&on_write, done_write, NULL, grpc_schedule_on_exec_ctx);
   grpc_slice_buffer_init(&state.temp_incoming_buffer);
@@ -237,7 +237,7 @@ typedef struct {
 } poll_args;
 
 static void actually_poll_server(void *arg) {
-  poll_args *pa = arg;
+  poll_args *pa = (poll_args *)arg;
   gpr_timespec deadline = n_sec_deadline(10);
   while (true) {
     bool done = gpr_atm_acq_load(&state.done_atm) != 0;
@@ -259,7 +259,7 @@ static void poll_server_until_read_done(test_tcp_server *server,
   gpr_atm_rel_store(&state.done_atm, 0);
   state.write_done = 0;
   gpr_thd_id id;
-  poll_args *pa = gpr_malloc(sizeof(*pa));
+  poll_args *pa = (poll_args *)gpr_malloc(sizeof(*pa));
   pa->server = server;
   pa->signal_when_done = signal_when_done;
   gpr_thd_new(&id, actually_poll_server, pa, NULL);

test/core/end2end/tests/resource_quota_server.c

@@ -111,10 +111,10 @@ void resource_quota_server(grpc_end2end_test_config config) {
   grpc_resource_quota_resize(resource_quota, 5 * 1024 * 1024);
 
 #define NUM_CALLS 100
-#define CLIENT_BASE_TAG 1000
+#define CLIENT_BASE_TAG 0x1000
-#define SERVER_START_BASE_TAG 2000
+#define SERVER_START_BASE_TAG 0x2000
-#define SERVER_RECV_BASE_TAG 3000
+#define SERVER_RECV_BASE_TAG 0x3000
-#define SERVER_END_BASE_TAG 4000
+#define SERVER_END_BASE_TAG 0x4000
 
   grpc_arg arg;
   arg.key = GRPC_ARG_RESOURCE_QUOTA;

test/core/iomgr/tcp_posix_test.c

@@ -89,7 +89,7 @@ static ssize_t fill_socket(int fd) {
 static size_t fill_socket_partial(int fd, size_t bytes) {
   ssize_t write_bytes;
   size_t total_bytes = 0;
-  unsigned char *buf = gpr_malloc(bytes);
+  unsigned char *buf = (unsigned char *)gpr_malloc(bytes);
   unsigned i;
   for (i = 0; i < bytes; ++i) {
     buf[i] = (uint8_t)(i % 256);
@@ -267,7 +267,7 @@ struct write_socket_state {
 static grpc_slice *allocate_blocks(size_t num_bytes, size_t slice_size,
                                    size_t *num_blocks, uint8_t *current_data) {
   size_t nslices = num_bytes / slice_size + (num_bytes % slice_size ? 1u : 0u);
-  grpc_slice *slices = gpr_malloc(sizeof(grpc_slice) * nslices);
+  grpc_slice *slices = (grpc_slice *)gpr_malloc(sizeof(grpc_slice) * nslices);
   size_t num_bytes_left = num_bytes;
   unsigned i, j;
   unsigned char *buf;
@@ -301,7 +301,7 @@ static void write_done(grpc_exec_ctx *exec_ctx,
 }
 
 void drain_socket_blocking(int fd, size_t num_bytes, size_t read_size) {
-  unsigned char *buf = gpr_malloc(read_size);
+  unsigned char *buf = (unsigned char *)gpr_malloc(read_size);
   ssize_t bytes_read;
   size_t bytes_left = num_bytes;
   int flags;
@@ -404,7 +404,7 @@ static void write_test(size_t num_bytes, size_t slice_size) {
 }
 
 void on_fd_released(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *errors) {
-  int *done = arg;
+  int *done = (int *)arg;
   *done = 1;
   GPR_ASSERT(
       GRPC_LOG_IF_ERROR("pollset_kick", grpc_pollset_kick(g_pollset, NULL)));
@@ -548,7 +548,7 @@ static grpc_endpoint_test_config configs[] = {
 
 static void destroy_pollset(grpc_exec_ctx *exec_ctx, void *p,
                             grpc_error *error) {
-  grpc_pollset_destroy(exec_ctx, p);
+  grpc_pollset_destroy(exec_ctx, (grpc_pollset *)p);
 }
 
 int main(int argc, char **argv) {
@@ -556,7 +556,7 @@ int main(int argc, char **argv) {
   grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
   grpc_test_init(argc, argv);
   grpc_init();
-  g_pollset = gpr_zalloc(grpc_pollset_size());
+  g_pollset = (grpc_pollset *)gpr_zalloc(grpc_pollset_size());
   grpc_pollset_init(g_pollset, &g_mu);
   grpc_endpoint_tests(configs[0], g_pollset, g_mu);
   run_tests();

test/core/security/secure_endpoint_test.c

@@ -70,7 +70,7 @@ static grpc_endpoint_test_fixture secure_endpoint_create_fixture_tcp_socketpair(
     size_t still_pending_size;
     size_t total_buffer_size = 8192;
     size_t buffer_size = total_buffer_size;
-    uint8_t *encrypted_buffer = gpr_malloc(buffer_size);
+    uint8_t *encrypted_buffer = (uint8_t *)gpr_malloc(buffer_size);
     uint8_t *cur = encrypted_buffer;
     grpc_slice encrypted_leftover;
     for (i = 0; i < leftover_nslices; i++) {
@@ -202,7 +202,7 @@ static void test_leftover(grpc_endpoint_test_config config, size_t slice_size) {
 
 static void destroy_pollset(grpc_exec_ctx *exec_ctx, void *p,
                             grpc_error *error) {
-  grpc_pollset_destroy(exec_ctx, p);
+  grpc_pollset_destroy(exec_ctx, (grpc_pollset *)p);
 }
 
 int main(int argc, char **argv) {
@@ -211,7 +211,7 @@ int main(int argc, char **argv) {
   grpc_test_init(argc, argv);
 
   grpc_init();
-  g_pollset = gpr_zalloc(grpc_pollset_size());
+  g_pollset = (grpc_pollset *)gpr_zalloc(grpc_pollset_size());
   grpc_pollset_init(g_pollset, &g_mu);
   grpc_endpoint_tests(configs[0], g_pollset, g_mu);
   grpc_endpoint_tests(configs[1], g_pollset, g_mu);

test/cpp/end2end/async_end2end_test.cc

@@ -266,6 +266,7 @@ class AsyncEnd2endTest : public ::testing::TestWithParam<TestScenario> {
   }
 
   void TearDown() override {
+    gpr_tls_set(&g_is_async_end2end_test, 0);
     server_->Shutdown();
     void* ignored_tag;
     bool ignored_ok;
@@ -274,7 +275,6 @@ class AsyncEnd2endTest : public ::testing::TestWithParam<TestScenario> {
       ;
     stub_.reset();
     poll_overrider_.reset();
-    gpr_tls_set(&g_is_async_end2end_test, 0);
     grpc_recycle_unused_port(port_);
   }
 
@@ -396,6 +396,7 @@ TEST_P(AsyncEnd2endTest, WaitAndShutdownTest) {
   ResetStub();
   SendRpc(1);
   EXPECT_EQ(0, notify);
+  gpr_tls_set(&g_is_async_end2end_test, 0);
   server_->Shutdown();
   wait_thread.join();
   EXPECT_EQ(1, notify);
@@ -404,8 +405,9 @@ TEST_P(AsyncEnd2endTest, WaitAndShutdownTest) {
 TEST_P(AsyncEnd2endTest, ShutdownThenWait) {
   ResetStub();
   SendRpc(1);
-  server_->Shutdown();
+  std::thread t([this]() { server_->Shutdown(); });
   server_->Wait();
+  t.join();
 }
 
 // Test a simple RPC using the async version of Next

test/cpp/end2end/end2end_test.cc

@@ -757,6 +757,22 @@ TEST_P(End2endTest, RequestStreamTwoRequests) {
   EXPECT_TRUE(s.ok());
 }
 
+TEST_P(End2endTest, RequestStreamTwoRequestsWithWriteThrough) {
+  ResetStub();
+  EchoRequest request;
+  EchoResponse response;
+  ClientContext context;
+
+  auto stream = stub_->RequestStream(&context, &response);
+  request.set_message("hello");
+  EXPECT_TRUE(stream->Write(request, WriteOptions().set_write_through()));
+  EXPECT_TRUE(stream->Write(request, WriteOptions().set_write_through()));
+  stream->WritesDone();
+  Status s = stream->Finish();
+  EXPECT_EQ(response.message(), "hellohello");
+  EXPECT_TRUE(s.ok());
+}
+
 TEST_P(End2endTest, RequestStreamTwoRequestsWithCoalescingApi) {
   ResetStub();
   EchoRequest request;

test/cpp/microbenchmarks/bm_chttp2_transport.cc

@@ -29,6 +29,7 @@
 extern "C" {
 #include "src/core/ext/transport/chttp2/transport/chttp2_transport.h"
 #include "src/core/ext/transport/chttp2/transport/internal.h"
+#include "src/core/lib/iomgr/closure.h"
 #include "src/core/lib/iomgr/resource_quota.h"
 #include "src/core/lib/slice/slice_internal.h"
 #include "src/core/lib/transport/static_metadata.h"
@@ -154,23 +155,59 @@ class Fixture {
   grpc_transport *t_;
 };
 
-static void DoNothing(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) {}
+class Closure : public grpc_closure {
+ public:
+  virtual ~Closure() {}
+};
+
+template <class F>
+std::unique_ptr<Closure> MakeClosure(
+    F f, grpc_closure_scheduler *sched = grpc_schedule_on_exec_ctx) {
+  struct C : public Closure {
+    C(const F &f, grpc_closure_scheduler *sched) : f_(f) {
+      GRPC_CLOSURE_INIT(this, Execute, this, sched);
+    }
+    F f_;
+    static void Execute(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) {
+      static_cast<C *>(arg)->f_(exec_ctx, error);
+    }
+  };
+  return std::unique_ptr<Closure>(new C(f, sched));
+}
+
+template <class F>
+grpc_closure *MakeOnceClosure(
+    F f, grpc_closure_scheduler *sched = grpc_schedule_on_exec_ctx) {
+  struct C : public grpc_closure {
+    C(const F &f) : f_(f) {}
+    F f_;
+    static void Execute(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) {
+      static_cast<C *>(arg)->f_(exec_ctx, error);
+      delete static_cast<C *>(arg);
+    }
+  };
+  auto *c = new C{f};
+  return GRPC_CLOSURE_INIT(c, C::Execute, c, sched);
+}
 
 class Stream {
  public:
   Stream(Fixture *f) : f_(f) {
-    GRPC_STREAM_REF_INIT(&refcount_, 1, DoNothing, nullptr, "test_stream");
     stream_size_ = grpc_transport_stream_size(f->transport());
     stream_ = gpr_malloc(stream_size_);
     arena_ = gpr_arena_create(4096);
   }
 
   ~Stream() {
+    gpr_event_wait(&done_, gpr_inf_future(GPR_CLOCK_REALTIME));
     gpr_free(stream_);
     gpr_arena_destroy(arena_);
   }
 
   void Init(benchmark::State &state) {
+    GRPC_STREAM_REF_INIT(&refcount_, 1, &Stream::FinishDestroy, this,
+                         "test_stream");
+    gpr_event_init(&done_);
     memset(stream_, 0, stream_size_);
     if ((state.iterations() & 0xffff) == 0) {
       gpr_arena_destroy(arena_);
@@ -181,13 +218,17 @@ class Stream {
                                NULL, arena_);
   }
 
-  void DestroyThen(grpc_closure *closure) {
+  void DestroyThen(grpc_exec_ctx *exec_ctx, grpc_closure *closure) {
-    grpc_transport_destroy_stream(f_->exec_ctx(), f_->transport(),
+    destroy_closure_ = closure;
-                                  static_cast<grpc_stream *>(stream_), closure);
+#ifndef NDEBUG
+    grpc_stream_unref(exec_ctx, &refcount_, "DestroyThen");
+#else
+    grpc_stream_unref(exec_ctx, &refcount_);
+#endif
   }
 
-  void Op(grpc_transport_stream_op_batch *op) {
+  void Op(grpc_exec_ctx *exec_ctx, grpc_transport_stream_op_batch *op) {
-    grpc_transport_perform_stream_op(f_->exec_ctx(), f_->transport(),
+    grpc_transport_perform_stream_op(exec_ctx, f_->transport(),
                                      static_cast<grpc_stream *>(stream_), op);
   }
 
@@ -196,48 +237,24 @@ class Stream {
   }
 
  private:
+  static void FinishDestroy(grpc_exec_ctx *exec_ctx, void *arg,
+                            grpc_error *error) {
+    auto stream = static_cast<Stream *>(arg);
+    grpc_transport_destroy_stream(exec_ctx, stream->f_->transport(),
+                                  static_cast<grpc_stream *>(stream->stream_),
+                                  stream->destroy_closure_);
+    gpr_event_set(&stream->done_, (void *)1);
+  }
+
   Fixture *f_;
   grpc_stream_refcount refcount_;
   gpr_arena *arena_;
   size_t stream_size_;
   void *stream_;
+  grpc_closure *destroy_closure_ = nullptr;
+  gpr_event done_;
 };
 
-class Closure : public grpc_closure {
- public:
-  virtual ~Closure() {}
-};
-
-template <class F>
-std::unique_ptr<Closure> MakeClosure(
-    F f, grpc_closure_scheduler *sched = grpc_schedule_on_exec_ctx) {
-  struct C : public Closure {
-    C(const F &f, grpc_closure_scheduler *sched) : f_(f) {
-      GRPC_CLOSURE_INIT(this, Execute, this, sched);
-    }
-    F f_;
-    static void Execute(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) {
-      static_cast<C *>(arg)->f_(exec_ctx, error);
-    }
-  };
-  return std::unique_ptr<Closure>(new C(f, sched));
-}
-
-template <class F>
-grpc_closure *MakeOnceClosure(
-    F f, grpc_closure_scheduler *sched = grpc_schedule_on_exec_ctx) {
-  struct C : public grpc_closure {
-    C(const F &f) : f_(f) {}
-    F f_;
-    static void Execute(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) {
-      static_cast<C *>(arg)->f_(exec_ctx, error);
-      delete static_cast<C *>(arg);
-    }
-  };
-  auto *c = new C{f};
-  return GRPC_CLOSURE_INIT(c, C::Execute, c, sched);
-}
-
 ////////////////////////////////////////////////////////////////////////////////
 // Benchmarks
 //
@@ -246,11 +263,18 @@ static void BM_StreamCreateDestroy(benchmark::State &state) {
   TrackCounters track_counters;
   Fixture f(grpc::ChannelArguments(), true);
   Stream s(&f);
+  grpc_transport_stream_op_batch op;
+  grpc_transport_stream_op_batch_payload op_payload;
+  memset(&op, 0, sizeof(op));
+  op.cancel_stream = true;
+  op.payload = &op_payload;
+  op_payload.cancel_stream.cancel_error = GRPC_ERROR_CANCELLED;
   std::unique_ptr<Closure> next =
       MakeClosure([&](grpc_exec_ctx *exec_ctx, grpc_error *error) {
         if (!state.KeepRunning()) return;
         s.Init(state);
-        s.DestroyThen(next.get());
+        s.Op(exec_ctx, &op);
+        s.DestroyThen(exec_ctx, next.get());
       });
   GRPC_CLOSURE_RUN(f.exec_ctx(), next.get(), GRPC_ERROR_NONE);
   f.FlushExecCtx();
@@ -314,14 +338,14 @@ static void BM_StreamCreateSendInitialMetadataDestroy(benchmark::State &state) {
     op.on_complete = done.get();
     op.send_initial_metadata = true;
     op.payload->send_initial_metadata.send_initial_metadata = &b;
-    s.Op(&op);
+    s.Op(exec_ctx, &op);
   });
   done = MakeClosure([&](grpc_exec_ctx *exec_ctx, grpc_error *error) {
     reset_op();
     op.cancel_stream = true;
     op.payload->cancel_stream.cancel_error = GRPC_ERROR_CANCELLED;
-    s.Op(&op);
+    s.Op(exec_ctx, &op);
-    s.DestroyThen(start.get());
+    s.DestroyThen(exec_ctx, start.get());
   });
   GRPC_CLOSURE_SCHED(f.exec_ctx(), start.get(), GRPC_ERROR_NONE);
   f.FlushExecCtx();
@@ -348,22 +372,28 @@ static void BM_TransportEmptyOp(benchmark::State &state) {
         if (!state.KeepRunning()) return;
         reset_op();
         op.on_complete = c.get();
-        s.Op(&op);
+        s.Op(exec_ctx, &op);
       });
   GRPC_CLOSURE_SCHED(f.exec_ctx(), c.get(), GRPC_ERROR_NONE);
   f.FlushExecCtx();
-  s.DestroyThen(
+  reset_op();
-      MakeOnceClosure([](grpc_exec_ctx *exec_ctx, grpc_error *error) {}));
+  op.cancel_stream = true;
+  op_payload.cancel_stream.cancel_error = GRPC_ERROR_CANCELLED;
+  s.Op(f.exec_ctx(), &op);
+  s.DestroyThen(f.exec_ctx(), MakeOnceClosure([](grpc_exec_ctx *exec_ctx,
+                                                 grpc_error *error) {}));
   f.FlushExecCtx();
   track_counters.Finish(state);
 }
 BENCHMARK(BM_TransportEmptyOp);
 
+std::vector<std::unique_ptr<gpr_event>> done_events;
+
 static void BM_TransportStreamSend(benchmark::State &state) {
   TrackCounters track_counters;
   Fixture f(grpc::ChannelArguments(), true);
-  Stream s(&f);
+  auto s = std::unique_ptr<Stream>(new Stream(&f));
-  s.Init(state);
+  s->Init(state);
   grpc_transport_stream_op_batch op;
   grpc_transport_stream_op_batch_payload op_payload;
   memset(&op_payload, 0, sizeof(op_payload));
@@ -390,11 +420,17 @@ static void BM_TransportStreamSend(benchmark::State &state) {
         grpc_metadata_batch_add_tail(f.exec_ctx(), &b, &storage[i], elems[i])));
   }
 
+  gpr_event *bm_done = new gpr_event;
+  gpr_event_init(bm_done);
+
   std::unique_ptr<Closure> c =
       MakeClosure([&](grpc_exec_ctx *exec_ctx, grpc_error *error) {
-        if (!state.KeepRunning()) return;
+        if (!state.KeepRunning()) {
+          gpr_event_set(bm_done, (void *)1);
+          return;
+        }
         // force outgoing window to be yuge
-        s.chttp2_stream()->flow_control.remote_window_delta =
+        s->chttp2_stream()->flow_control.remote_window_delta =
             1024 * 1024 * 1024;
         f.chttp2_transport()->flow_control.remote_window = 1024 * 1024 * 1024;
         grpc_slice_buffer_stream_init(&send_stream, &send_buffer, 0);
@@ -402,23 +438,27 @@ static void BM_TransportStreamSend(benchmark::State &state) {
         op.on_complete = c.get();
         op.send_message = true;
         op.payload->send_message.send_message = &send_stream.base;
-        s.Op(&op);
+        s->Op(exec_ctx, &op);
       });
 
   reset_op();
   op.send_initial_metadata = true;
   op.payload->send_initial_metadata.send_initial_metadata = &b;
   op.on_complete = c.get();
-  s.Op(&op);
+  s->Op(f.exec_ctx(), &op);
 
   f.FlushExecCtx();
+  gpr_event_wait(bm_done, gpr_inf_future(GPR_CLOCK_REALTIME));
+  done_events.emplace_back(bm_done);
+
   reset_op();
   op.cancel_stream = true;
   op.payload->cancel_stream.cancel_error = GRPC_ERROR_CANCELLED;
-  s.Op(&op);
+  s->Op(f.exec_ctx(), &op);
-  s.DestroyThen(
+  s->DestroyThen(f.exec_ctx(), MakeOnceClosure([](grpc_exec_ctx *exec_ctx,
-      MakeOnceClosure([](grpc_exec_ctx *exec_ctx, grpc_error *error) {}));
+                                                  grpc_error *error) {}));
   f.FlushExecCtx();
+  s.reset();
   track_counters.Finish(state);
   grpc_metadata_batch_destroy(f.exec_ctx(), &b);
   grpc_slice_buffer_destroy(&send_buffer);
@@ -535,7 +575,7 @@ static void BM_TransportStreamRecv(benchmark::State &state) {
     op.recv_message = true;
     op.payload->recv_message.recv_message = &recv_stream;
     op.payload->recv_message.recv_message_ready = drain_start.get();
-    s.Op(&op);
+    s.Op(exec_ctx, &op);
     f.PushInput(grpc_slice_ref(incoming_data));
   });
 
@@ -578,7 +618,7 @@ static void BM_TransportStreamRecv(benchmark::State &state) {
   op.payload->recv_initial_metadata.recv_initial_metadata_ready =
       do_nothing.get();
   op.on_complete = c.get();
-  s.Op(&op);
+  s.Op(f.exec_ctx(), &op);
   f.PushInput(SLICE_FROM_BUFFER(
       "\x00\x00\x00\x04\x00\x00\x00\x00\x00"
       // Generated using:
@@ -596,9 +636,9 @@ static void BM_TransportStreamRecv(benchmark::State &state) {
   reset_op();
   op.cancel_stream = true;
   op.payload->cancel_stream.cancel_error = GRPC_ERROR_CANCELLED;
-  s.Op(&op);
+  s.Op(f.exec_ctx(), &op);
-  s.DestroyThen(
+  s.DestroyThen(f.exec_ctx(), MakeOnceClosure([](grpc_exec_ctx *exec_ctx,
-      MakeOnceClosure([](grpc_exec_ctx *exec_ctx, grpc_error *error) {}));
+                                                 grpc_error *error) {}));
   f.FlushExecCtx();
   track_counters.Finish(state);
   grpc_metadata_batch_destroy(f.exec_ctx(), &b);

test/cpp/microbenchmarks/bm_fullstack_trickle.cc

@@ -105,7 +105,7 @@ class TrickledCHTTP2 : public EndpointPairFixture {
             (double)state.iterations());
   }
 
-  void Log(int64_t iteration) {
+  void Log(int64_t iteration) GPR_ATTRIBUTE_NO_TSAN {
     auto now = gpr_time_sub(gpr_now(GPR_CLOCK_MONOTONIC), start_);
     grpc_chttp2_transport* client =
         reinterpret_cast<grpc_chttp2_transport*>(client_transport_);
@@ -193,7 +193,8 @@ class TrickledCHTTP2 : public EndpointPairFixture {
     return p;
   }
 
-  void UpdateStats(grpc_chttp2_transport* t, Stats* s, size_t backlog) {
+  void UpdateStats(grpc_chttp2_transport* t, Stats* s,
+                   size_t backlog) GPR_ATTRIBUTE_NO_TSAN {
     if (backlog == 0) {
       if (t->lists[GRPC_CHTTP2_LIST_STALLED_BY_STREAM].head != NULL) {
         s->streams_stalled_due_to_stream_flow_control++;

tools/run_tests/performance/scenario_config.py

@@ -156,7 +156,7 @@ def _ping_pong_scenario(name, rpc_type,
 
   # Optimization target of 'throughput' does not work well with epoll1 polling
   # engine. Use the default value of 'blend'
-  optimization_target = 'blend'
+  optimization_target = 'throughput'
 
   if unconstrained_client:
     outstanding_calls = outstanding if outstanding is not None else OUTSTANDING_REQUESTS[unconstrained_client]
@@ -175,6 +175,7 @@ def _ping_pong_scenario(name, rpc_type,
     scenario['client_config']['outstanding_rpcs_per_channel'] = 1
     scenario['client_config']['client_channels'] = 1
     scenario['client_config']['async_client_threads'] = 1
+    optimization_target = 'latency'
 
   optimization_channel_arg = {
     'name': 'grpc.optimization_target',