Merge github.com:grpc/grpc into flow-like-lava-to-a-barnyard

1 files changed, 156 insertions, 192 deletions

diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c
index 0307cc3154..6db65405d0 100644
--- a/src/core/transport/chttp2_transport.c
+++ b/src/core/transport/chttp2_transport.c
@@ -37,18 +37,19 @@
 #include <stdio.h>
 #include <string.h>
 
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/slice_buffer.h>
+#include <grpc/support/string_util.h>
+#include <grpc/support/useful.h>
+
 #include "src/core/profiling/timers.h"
 #include "src/core/support/string.h"
 #include "src/core/transport/chttp2/http2_errors.h"
+#include "src/core/transport/chttp2/internal.h"
 #include "src/core/transport/chttp2/status_conversion.h"
 #include "src/core/transport/chttp2/timeout_encoding.h"
-#include "src/core/transport/chttp2/internal.h"
 #include "src/core/transport/transport_impl.h"
-#include <grpc/support/alloc.h>
-#include <grpc/support/log.h>
-#include <grpc/support/slice_buffer.h>
-#include <grpc/support/string_util.h>
-#include <grpc/support/useful.h>
 
 #define DEFAULT_WINDOW 65535
 #define DEFAULT_CONNECTION_WINDOW_TARGET (1024 * 1024)
@@ -79,13 +80,11 @@ static const grpc_transport_vtable vtable;
 static void lock(grpc_chttp2_transport *t);
 static void unlock(grpc_chttp2_transport *t);
 
-static void unlock_check_channel_callbacks(grpc_chttp2_transport *t);
 static void unlock_check_read_write_state(grpc_chttp2_transport *t);
 
 /* forward declarations of various callbacks that we'll build closures around */
 static void writing_action(void *t, int iomgr_success_ignored);
 static void reading_action(void *t, int iomgr_success_ignored);
-static void notify_closed(void *t, int iomgr_success_ignored);
 
 /** Set a transport level setting, and push it to our peer */
 static void push_setting(grpc_chttp2_transport *t, grpc_chttp2_setting_id id,
@@ -99,9 +98,9 @@ static void recv_data(void *tp, gpr_slice *slices, size_t nslices,
 static void drop_connection(grpc_chttp2_transport *t);
 
 /** Perform a transport_op */
-static void perform_op_locked(grpc_chttp2_transport_global *transport_global,
-                              grpc_chttp2_stream_global *stream_global,
-                              grpc_transport_op *op);
+static void perform_stream_op_locked(
+    grpc_chttp2_transport_global *transport_global,
+    grpc_chttp2_stream_global *stream_global, grpc_transport_stream_op *op);
 
 /** Cancel a stream: coming from the transport API */
 static void cancel_from_api(grpc_chttp2_transport_global *transport_global,
@@ -116,6 +115,10 @@ static void add_to_pollset_locked(grpc_chttp2_transport *t,
 static void maybe_start_some_streams(
     grpc_chttp2_transport_global *transport_global);
 
+static void connectivity_state_set(
+    grpc_chttp2_transport_global *transport_global,
+    grpc_connectivity_state state);
+
 /*
  * CONSTRUCTION/DESTRUCTION/REFCOUNTING
  */
@@ -148,6 +151,7 @@ static void destruct_transport(grpc_chttp2_transport *t) {
 
   grpc_chttp2_stream_map_destroy(&t->parsing_stream_map);
   grpc_chttp2_stream_map_destroy(&t->new_stream_map);
+  grpc_connectivity_state_destroy(&t->channel_callback.state_tracker);
 
   gpr_mu_unlock(&t->mu);
   gpr_mu_destroy(&t->mu);
@@ -196,13 +200,11 @@ static void ref_transport(grpc_chttp2_transport *t) { gpr_ref(&t->refs); }
 #endif
 
 static void init_transport(grpc_chttp2_transport *t,
-                           grpc_transport_setup_callback setup, void *arg,
                            const grpc_channel_args *channel_args,
-                           grpc_endpoint *ep, gpr_slice *slices, size_t nslices,
-                           grpc_mdctx *mdctx, int is_client) {
+                           grpc_endpoint *ep, grpc_mdctx *mdctx,
+                           int is_client) {
   size_t i;
   int j;
-  grpc_transport_setup_result sr;
 
   GPR_ASSERT(strlen(GRPC_CHTTP2_CLIENT_CONNECT_STRING) ==
              GRPC_CHTTP2_CLIENT_CONNECT_STRLEN);
@@ -217,7 +219,6 @@ static void init_transport(grpc_chttp2_transport *t,
   grpc_mdctx_ref(mdctx);
   t->metadata_context = mdctx;
   t->endpoint_reading = 1;
-  t->global.error_state = GRPC_CHTTP2_ERROR_STATE_NONE;
   t->global.next_stream_id = is_client ? 1 : 2;
   t->global.is_client = is_client;
   t->global.outgoing_window = DEFAULT_WINDOW;
@@ -231,6 +232,8 @@ static void init_transport(grpc_chttp2_transport *t,
   t->parsing.deframe_state =
       is_client ? GRPC_DTS_FH_0 : GRPC_DTS_CLIENT_PREFIX_0;
   t->writing.is_client = is_client;
+  grpc_connectivity_state_init(&t->channel_callback.state_tracker,
+                               GRPC_CHANNEL_READY);
 
   gpr_slice_buffer_init(&t->global.qbuf);
 
@@ -243,7 +246,6 @@ static void init_transport(grpc_chttp2_transport *t,
   grpc_chttp2_goaway_parser_init(&t->parsing.goaway_parser);
   grpc_chttp2_hpack_parser_init(&t->parsing.hpack_parser, t->metadata_context);
 
-  grpc_iomgr_closure_init(&t->channel_callback.notify_closed, notify_closed, t);
   if (is_client) {
     gpr_slice_buffer_add(
         &t->global.qbuf,
@@ -309,24 +311,6 @@ static void init_transport(grpc_chttp2_transport *t,
       }
     }
   }
-
-  gpr_mu_lock(&t->mu);
-  t->channel_callback.executing = 1;
-  REF_TRANSPORT(t, "init"); /* matches unref at end of this function */
-  gpr_mu_unlock(&t->mu);
-
-  sr = setup(arg, &t->base, t->metadata_context);
-
-  lock(t);
-  t->channel_callback.cb = sr.callbacks;
-  t->channel_callback.cb_user_data = sr.user_data;
-  t->channel_callback.executing = 0;
-  unlock(t);
-
-  REF_TRANSPORT(t, "recv_data"); /* matches unref inside recv_data */
-  recv_data(t, slices, nslices, GRPC_ENDPOINT_CB_OK);
-
-  UNREF_TRANSPORT(t, "init");
 }
 
 static void destroy_transport(grpc_transport *gt) {
@@ -343,31 +327,16 @@ static void destroy_transport(grpc_transport *gt) {
 static void close_transport_locked(grpc_chttp2_transport *t) {
   if (!t->closed) {
     t->closed = 1;
+    connectivity_state_set(&t->global, GRPC_CHANNEL_FATAL_FAILURE);
     if (t->ep) {
       grpc_endpoint_shutdown(t->ep);
     }
   }
 }
 
-static void close_transport(grpc_transport *gt) {
-  grpc_chttp2_transport *t = (grpc_chttp2_transport *)gt;
-  gpr_mu_lock(&t->mu);
-  close_transport_locked(t);
-  gpr_mu_unlock(&t->mu);
-}
-
-static void goaway(grpc_transport *gt, grpc_status_code status,
-                   gpr_slice debug_data) {
-  grpc_chttp2_transport *t = (grpc_chttp2_transport *)gt;
-  lock(t);
-  grpc_chttp2_goaway_append(t->global.last_incoming_stream_id,
-                            grpc_chttp2_grpc_status_to_http2_error(status),
-                            debug_data, &t->global.qbuf);
-  unlock(t);
-}
-
 static int init_stream(grpc_transport *gt, grpc_stream *gs,
-                       const void *server_data, grpc_transport_op *initial_op) {
+                       const void *server_data,
+                       grpc_transport_stream_op *initial_op) {
   grpc_chttp2_transport *t = (grpc_chttp2_transport *)gt;
   grpc_chttp2_stream *s = (grpc_chttp2_stream *)gs;
 
@@ -399,7 +368,7 @@ static int init_stream(grpc_transport *gt, grpc_stream *gs,
     s->global.in_stream_map = 1;
   }
 
-  if (initial_op) perform_op_locked(&t->global, &s->global, initial_op);
+  if (initial_op) perform_stream_op_locked(&t->global, &s->global, initial_op);
   unlock(t);
 
   return 0;
@@ -457,8 +426,8 @@ grpc_chttp2_stream_parsing *grpc_chttp2_parsing_accept_stream(
   grpc_chttp2_transport *t = TRANSPORT_FROM_PARSING(transport_parsing);
   GPR_ASSERT(t->accepting_stream == NULL);
   t->accepting_stream = &accepting;
-  t->channel_callback.cb->accept_stream(t->channel_callback.cb_user_data,
-                                        &t->base, (void *)(gpr_uintptr)id);
+  t->channel_callback.accept_stream(t->channel_callback.accept_stream_user_data,
+                                    &t->base, (void *)(gpr_uintptr)id);
   t->accepting_stream = NULL;
   return &accepting->parsing;
 }
@@ -479,17 +448,16 @@ static void unlock(grpc_chttp2_transport *t) {
   grpc_iomgr_closure *run_closures;
 
   unlock_check_read_write_state(t);
-  if (!t->writing_active && t->global.error_state == GRPC_CHTTP2_ERROR_STATE_NONE &&
+  if (!t->writing_active && !t->closed &&
       grpc_chttp2_unlocking_check_writes(&t->global, &t->writing)) {
     t->writing_active = 1;
     REF_TRANSPORT(t, "writing");
     grpc_chttp2_schedule_closure(&t->global, &t->writing_action, 1);
   }
-  /* unlock_check_parser(t); */
-  unlock_check_channel_callbacks(t);
 
-  run_closures = t->global.pending_closures;
-  t->global.pending_closures = NULL;
+  run_closures = t->global.pending_closures_head;
+  t->global.pending_closures_head = NULL;
+  t->global.pending_closures_tail = NULL;
 
   gpr_mu_unlock(&t->mu);
 
@@ -555,15 +523,13 @@ static void writing_action(void *gt, int iomgr_success_ignored) {
 void grpc_chttp2_add_incoming_goaway(
     grpc_chttp2_transport_global *transport_global, gpr_uint32 goaway_error,
     gpr_slice goaway_text) {
-  char *msg = gpr_hexdump((char*)GPR_SLICE_START_PTR(goaway_text), GPR_SLICE_LENGTH(goaway_text), GPR_HEXDUMP_PLAINTEXT);
+  char *msg = gpr_hexdump((char *)GPR_SLICE_START_PTR(goaway_text),
+                          GPR_SLICE_LENGTH(goaway_text), GPR_HEXDUMP_PLAINTEXT);
+  gpr_log(GPR_DEBUG, "got goaway [%d]: %s", goaway_error, msg);
   gpr_free(msg);
-  if (transport_global->goaway_state == GRPC_CHTTP2_ERROR_STATE_NONE) {
-    transport_global->goaway_state = GRPC_CHTTP2_ERROR_STATE_SEEN;
-    transport_global->goaway_text = goaway_text;
-    transport_global->goaway_error = goaway_error;
-  } else {
-    gpr_slice_unref(goaway_text);
-  }
+  gpr_slice_unref(goaway_text);
+  transport_global->seen_goaway = 1;
+  connectivity_state_set(transport_global, GRPC_CHANNEL_FATAL_FAILURE);
 }
 
 static void maybe_start_some_streams(
@@ -588,9 +554,7 @@ static void maybe_start_some_streams(
     transport_global->next_stream_id += 2;
 
     if (transport_global->next_stream_id >= MAX_CLIENT_STREAM_ID) {
-      grpc_chttp2_add_incoming_goaway(
-          transport_global, GRPC_CHTTP2_NO_ERROR,
-          gpr_slice_from_copied_string("Exceeded sequence number limit"));
+      connectivity_state_set(transport_global, GRPC_CHANNEL_TRANSIENT_FAILURE);
     }
 
     stream_global->outgoing_window =
@@ -621,9 +585,9 @@ static void maybe_start_some_streams(
   }
 }
 
-static void perform_op_locked(grpc_chttp2_transport_global *transport_global,
-                              grpc_chttp2_stream_global *stream_global,
-                              grpc_transport_op *op) {
+static void perform_stream_op_locked(
+    grpc_chttp2_transport_global *transport_global,
+    grpc_chttp2_stream_global *stream_global, grpc_transport_stream_op *op) {
   if (op->cancel_with_status != GRPC_STATUS_OK) {
     cancel_from_api(transport_global, stream_global, op->cancel_with_status);
   }
@@ -688,21 +652,19 @@ static void perform_op_locked(grpc_chttp2_transport_global *transport_global,
   }
 }
 
-static void perform_op(grpc_transport *gt, grpc_stream *gs,
-                       grpc_transport_op *op) {
+static void perform_stream_op(grpc_transport *gt, grpc_stream *gs,
+                              grpc_transport_stream_op *op) {
   grpc_chttp2_transport *t = (grpc_chttp2_transport *)gt;
   grpc_chttp2_stream *s = (grpc_chttp2_stream *)gs;
 
   lock(t);
-  perform_op_locked(&t->global, &s->global, op);
+  perform_stream_op_locked(&t->global, &s->global, op);
   unlock(t);
 }
 
-static void send_ping(grpc_transport *gt, grpc_iomgr_closure *on_recv) {
-  grpc_chttp2_transport *t = (grpc_chttp2_transport *)gt;
+static void send_ping_locked(grpc_chttp2_transport *t,
+                             grpc_iomgr_closure *on_recv) {
   grpc_chttp2_outstanding_ping *p = gpr_malloc(sizeof(*p));
-
-  lock(t);
   p->next = &t->global.pings;
   p->prev = p->next->prev;
   p->prev->next = p->next->prev = p;
@@ -716,6 +678,48 @@ static void send_ping(grpc_transport *gt, grpc_iomgr_closure *on_recv) {
   p->id[7] = t->global.ping_counter & 0xff;
   p->on_recv = on_recv;
   gpr_slice_buffer_add(&t->global.qbuf, grpc_chttp2_ping_create(0, p->id));
+}
+
+static void perform_transport_op(grpc_transport *gt, grpc_transport_op *op) {
+  grpc_chttp2_transport *t = (grpc_chttp2_transport *)gt;
+
+  lock(t);
+
+  if (op->on_consumed) {
+    grpc_chttp2_schedule_closure(&t->global, op->on_consumed, 1);
+  }
+
+  if (op->on_connectivity_state_change) {
+    grpc_connectivity_state_notify_on_state_change(
+        &t->channel_callback.state_tracker, op->connectivity_state,
+        op->on_connectivity_state_change);
+  }
+
+  if (op->send_goaway) {
+    grpc_chttp2_goaway_append(
+        t->global.last_incoming_stream_id,
+        grpc_chttp2_grpc_status_to_http2_error(op->goaway_status),
+        gpr_slice_ref(*op->goaway_message), &t->global.qbuf);
+  }
+
+  if (op->set_accept_stream != NULL) {
+    t->channel_callback.accept_stream = op->set_accept_stream;
+    t->channel_callback.accept_stream_user_data =
+        op->set_accept_stream_user_data;
+  }
+
+  if (op->bind_pollset) {
+    add_to_pollset_locked(t, op->bind_pollset);
+  }
+
+  if (op->send_ping) {
+    send_ping_locked(t, op->send_ping);
+  }
+
+  if (op->disconnect) {
+    close_transport_locked(t);
+  }
+
   unlock(t);
 }
 
@@ -745,9 +749,8 @@ static void remove_stream(grpc_chttp2_transport *t, gpr_uint32 id) {
     grpc_chttp2_parsing_become_skip_parser(&t->parsing);
   }
 
-  new_stream_count =
-      grpc_chttp2_stream_map_size(&t->parsing_stream_map) +
-      grpc_chttp2_stream_map_size(&t->new_stream_map);
+  new_stream_count = grpc_chttp2_stream_map_size(&t->parsing_stream_map) +
+                     grpc_chttp2_stream_map_size(&t->new_stream_map);
   if (new_stream_count != t->global.concurrent_stream_count) {
     t->global.concurrent_stream_count = new_stream_count;
     maybe_start_some_streams(&t->global);
@@ -774,20 +777,35 @@ static void unlock_check_read_write_state(grpc_chttp2_transport *t) {
     }
   }
 
+  if (!t->writing_active) {
+    while (grpc_chttp2_list_pop_cancelled_waiting_for_writing(transport_global,
+                                                              &stream_global)) {
+      grpc_chttp2_list_add_read_write_state_changed(transport_global,
+                                                    stream_global);
+    }
+  }
+
   while (grpc_chttp2_list_pop_read_write_state_changed(transport_global,
                                                        &stream_global)) {
     if (stream_global->cancelled) {
-      stream_global->write_state = GRPC_WRITE_STATE_SENT_CLOSE;
-      stream_global->read_closed = 1;
-      if (!stream_global->published_cancelled) {
-        char buffer[GPR_LTOA_MIN_BUFSIZE];
-        gpr_ltoa(stream_global->cancelled_status, buffer);
-        grpc_chttp2_incoming_metadata_buffer_add(&stream_global->incoming_metadata,
-          grpc_mdelem_from_strings(t->metadata_context, "grpc-status", buffer));
-        grpc_chttp2_incoming_metadata_buffer_place_metadata_batch_into(
-          &stream_global->incoming_metadata,
-          &stream_global->incoming_sopb);
-        stream_global->published_cancelled = 1;
+      if (t->writing_active &&
+          stream_global->write_state != GRPC_WRITE_STATE_SENT_CLOSE) {
+        grpc_chttp2_list_add_cancelled_waiting_for_writing(transport_global,
+                                                           stream_global);
+      } else {
+        stream_global->write_state = GRPC_WRITE_STATE_SENT_CLOSE;
+        stream_global->read_closed = 1;
+        if (!stream_global->published_cancelled) {
+          char buffer[GPR_LTOA_MIN_BUFSIZE];
+          gpr_ltoa(stream_global->cancelled_status, buffer);
+          grpc_chttp2_incoming_metadata_buffer_add(
+              &stream_global->incoming_metadata,
+              grpc_mdelem_from_strings(t->metadata_context, "grpc-status",
+                                       buffer));
+          grpc_chttp2_incoming_metadata_buffer_place_metadata_batch_into(
+              &stream_global->incoming_metadata, &stream_global->incoming_sopb);
+          stream_global->published_cancelled = 1;
+        }
       }
     }
     if (stream_global->write_state == GRPC_WRITE_STATE_SENT_CLOSE &&
@@ -835,10 +853,10 @@ static void cancel_from_api(grpc_chttp2_transport_global *transport_global,
   stream_global->cancelled = 1;
   stream_global->cancelled_status = status;
   if (stream_global->id != 0) {
-    gpr_slice_buffer_add(&transport_global->qbuf,
-                         grpc_chttp2_rst_stream_create(
-                             stream_global->id,
-                             grpc_chttp2_grpc_status_to_http2_error(status)));
+    gpr_slice_buffer_add(
+        &transport_global->qbuf,
+        grpc_chttp2_rst_stream_create(
+            stream_global->id, grpc_chttp2_grpc_status_to_http2_error(status)));
   }
   grpc_chttp2_list_add_read_write_state_changed(transport_global,
                                                 stream_global);
@@ -855,9 +873,6 @@ static void end_all_the_calls(grpc_chttp2_transport *t) {
 }
 
 static void drop_connection(grpc_chttp2_transport *t) {
-  if (t->global.error_state == GRPC_CHTTP2_ERROR_STATE_NONE) {
-    t->global.error_state = GRPC_CHTTP2_ERROR_STATE_SEEN;
-  }
   close_transport_locked(t);
   end_all_the_calls(t);
 }
@@ -907,7 +922,7 @@ static void recv_data(void *tp, gpr_slice *slices, size_t nslices,
       lock(t);
       i = 0;
       GPR_ASSERT(!t->parsing_active);
-      if (t->global.error_state == GRPC_CHTTP2_ERROR_STATE_NONE) {
+      if (!t->closed) {
         t->parsing_active = 1;
         /* merge stream lists */
         grpc_chttp2_stream_map_move_into(&t->new_stream_map,
@@ -925,7 +940,8 @@ static void recv_data(void *tp, gpr_slice *slices, size_t nslices,
         /* merge stream lists */
         grpc_chttp2_stream_map_move_into(&t->new_stream_map,
                                          &t->parsing_stream_map);
-        t->global.concurrent_stream_count = grpc_chttp2_stream_map_size(&t->parsing_stream_map);
+        t->global.concurrent_stream_count =
+            grpc_chttp2_stream_map_size(&t->parsing_stream_map);
         if (t->parsing.initial_window_update != 0) {
           grpc_chttp2_stream_map_for_each(&t->parsing_stream_map,
                                           update_global_window, t);
@@ -959,76 +975,33 @@ static void reading_action(void *pt, int iomgr_success_ignored) {
  * CALLBACK LOOP
  */
 
-typedef struct {
-  grpc_chttp2_transport *t;
-  gpr_uint32 error;
-  gpr_slice text;
-  grpc_iomgr_closure closure;
-} notify_goaways_args;
-
-static void notify_goaways(void *p, int iomgr_success_ignored) {
-  notify_goaways_args *a = p;
-  grpc_chttp2_transport *t = a->t;
-
-  t->channel_callback.cb->goaway(t->channel_callback.cb_user_data, &t->base,
-                                 a->error, a->text);
-
-  gpr_free(a);
-
-  lock(t);
-  t->channel_callback.executing = 0;
-  unlock(t);
-
-  UNREF_TRANSPORT(t, "notify_goaways");
+static void schedule_closure_for_connectivity(void *a,
+                                              grpc_iomgr_closure *closure) {
+  grpc_chttp2_schedule_closure(a, closure, 1);
 }
 
-static void notify_closed(void *gt, int iomgr_success_ignored) {
-  grpc_chttp2_transport *t = gt;
-  t->channel_callback.cb->closed(t->channel_callback.cb_user_data, &t->base);
-
-  lock(t);
-  t->channel_callback.executing = 0;
-  unlock(t);
-
-  UNREF_TRANSPORT(t, "notify_closed");
-}
-
-static void unlock_check_channel_callbacks(grpc_chttp2_transport *t) {
-  if (t->channel_callback.executing) {
-    return;
-  }
-  if (t->global.goaway_state != GRPC_CHTTP2_ERROR_STATE_NONE) {
-    if (t->global.goaway_state == GRPC_CHTTP2_ERROR_STATE_SEEN &&
-        t->global.error_state != GRPC_CHTTP2_ERROR_STATE_NOTIFIED) {
-      notify_goaways_args *a = gpr_malloc(sizeof(*a));
-      a->t = t;
-      a->error = t->global.goaway_error;
-      a->text = t->global.goaway_text;
-      t->global.goaway_state = GRPC_CHTTP2_ERROR_STATE_NOTIFIED;
-      t->channel_callback.executing = 1;
-      grpc_iomgr_closure_init(&a->closure, notify_goaways, a);
-      REF_TRANSPORT(t, "notify_goaways");
-      grpc_chttp2_schedule_closure(&t->global, &a->closure, 1);
-      return;
-    } else if (t->global.goaway_state != GRPC_CHTTP2_ERROR_STATE_NOTIFIED) {
-      return;
-    }
-  }
-  if (t->global.error_state == GRPC_CHTTP2_ERROR_STATE_SEEN) {
-    t->global.error_state = GRPC_CHTTP2_ERROR_STATE_NOTIFIED;
-    t->channel_callback.executing = 1;
-    REF_TRANSPORT(t, "notify_closed");
-    grpc_chttp2_schedule_closure(&t->global, &t->channel_callback.notify_closed,
-                                 1);
-  }
+static void connectivity_state_set(
+    grpc_chttp2_transport_global *transport_global,
+    grpc_connectivity_state state) {
+  GRPC_CHTTP2_IF_TRACING(
+      gpr_log(GPR_DEBUG, "set connectivity_state=%d", state));
+  grpc_connectivity_state_set_with_scheduler(
+      &TRANSPORT_FROM_GLOBAL(transport_global)->channel_callback.state_tracker,
+      state, schedule_closure_for_connectivity, transport_global);
 }
 
 void grpc_chttp2_schedule_closure(
     grpc_chttp2_transport_global *transport_global, grpc_iomgr_closure *closure,
     int success) {
   closure->success = success;
-  closure->next = transport_global->pending_closures;
-  transport_global->pending_closures = closure;
+  if (transport_global->pending_closures_tail == NULL) {
+    transport_global->pending_closures_head =
+        transport_global->pending_closures_tail = closure;
+  } else {
+    transport_global->pending_closures_tail->next = closure;
+    transport_global->pending_closures_tail = closure;
+  }
+  closure->next = NULL;
 }
 
 /*
@@ -1042,13 +1015,6 @@ static void add_to_pollset_locked(grpc_chttp2_transport *t,
   }
 }
 
-static void add_to_pollset(grpc_transport *gt, grpc_pollset *pollset) {
-  grpc_chttp2_transport *t = (grpc_chttp2_transport *)gt;
-  lock(t);
-  add_to_pollset_locked(t, pollset);
-  unlock(t);
-}
-
 /*
  * TRACING
  */
@@ -1084,23 +1050,21 @@ void grpc_chttp2_flowctl_trace(const char *file, int line, const char *reason,
  * INTEGRATION GLUE
  */
 
-static const grpc_transport_vtable vtable = {sizeof(grpc_chttp2_stream),
-                                             init_stream,
-                                             perform_op,
-                                             add_to_pollset,
-                                             destroy_stream,
-                                             goaway,
-                                             close_transport,
-                                             send_ping,
-                                             destroy_transport};
-
-void grpc_create_chttp2_transport(grpc_transport_setup_callback setup,
-                                  void *arg,
-                                  const grpc_channel_args *channel_args,
-                                  grpc_endpoint *ep, gpr_slice *slices,
-                                  size_t nslices, grpc_mdctx *mdctx,
-                                  int is_client) {
+static const grpc_transport_vtable vtable = {
+    sizeof(grpc_chttp2_stream), init_stream,    perform_stream_op,
+    perform_transport_op,       destroy_stream, destroy_transport};
+
+grpc_transport *grpc_create_chttp2_transport(
+    const grpc_channel_args *channel_args, grpc_endpoint *ep, grpc_mdctx *mdctx,
+    int is_client) {
   grpc_chttp2_transport *t = gpr_malloc(sizeof(grpc_chttp2_transport));
-  init_transport(t, setup, arg, channel_args, ep, slices, nslices, mdctx,
-                 is_client);
+  init_transport(t, channel_args, ep, mdctx, is_client);
+  return &t->base;
+}
+
+void grpc_chttp2_transport_start_reading(grpc_transport *transport,
+                                         gpr_slice *slices, size_t nslices) {
+  grpc_chttp2_transport *t = (grpc_chttp2_transport *)transport;
+  REF_TRANSPORT(t, "recv_data"); /* matches unref inside recv_data */
+  recv_data(t, slices, nslices, GRPC_ENDPOINT_CB_OK);
 }