Got chttp2 transport compiling again

4 files changed, 48 insertions, 18 deletions

diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.c b/src/core/ext/transport/chttp2/transport/chttp2_transport.c
index a3020236cf..194264c8f9 100644
--- a/src/core/ext/transport/chttp2/transport/chttp2_transport.c
+++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.c
@@ -98,6 +98,8 @@ static void post_parse_locked(grpc_exec_ctx *exec_ctx, void *arg,
                               grpc_error *error);
 static void initiate_writing_locked(grpc_exec_ctx *exec_ctx, void *t,
                                     grpc_error *error);
+static void initiate_read_flush_locked(grpc_exec_ctx *exec_ctx, void *t,
+                                       grpc_error *error);
 static void terminate_writing_with_lock(grpc_exec_ctx *exec_ctx, void *t,
                                         grpc_error *error);
 
@@ -273,6 +275,8 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
   grpc_closure_init(&t->post_parse_locked, post_parse_locked, t);
   grpc_closure_init(&t->initiate_writing, initiate_writing_locked, t);
   grpc_closure_init(&t->terminate_writing, terminate_writing_with_lock, t);
+  grpc_closure_init(&t->initiate_read_flush_locked, initiate_read_flush_locked,
+                    t);
   grpc_closure_init(&t->writing.done_cb, grpc_chttp2_terminate_writing,
                     &t->writing);
 
@@ -666,6 +670,7 @@ static void initiate_writing_locked(grpc_exec_ctx *exec_ctx, void *tp,
 static void initiate_read_flush_locked(grpc_exec_ctx *exec_ctx, void *tp,
                                        grpc_error *error) {
   grpc_chttp2_transport *t = tp;
+  t->executor.check_read_ops_scheduled = false;
   check_read_ops(exec_ctx, &t->global);
 }
 
@@ -1012,7 +1017,8 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, void *stream_op,
     } else {
       if (contains_non_ok_status(transport_global, op->send_initial_metadata)) {
         stream_global->seen_error = true;
-        grpc_chttp2_list_add_check_read_ops(transport_global, stream_global);
+        grpc_chttp2_list_add_check_read_ops(exec_ctx, transport_global,
+                                            stream_global);
       }
       if (!stream_global->write_closed) {
         if (transport_global->is_client) {
@@ -1079,7 +1085,8 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, void *stream_op,
       if (contains_non_ok_status(transport_global,
                                  op->send_trailing_metadata)) {
         stream_global->seen_error = true;
-        grpc_chttp2_list_add_check_read_ops(transport_global, stream_global);
+        grpc_chttp2_list_add_check_read_ops(exec_ctx, transport_global,
+                                            stream_global);
       }
       if (stream_global->write_closed) {
         stream_global->send_trailing_metadata = NULL;
@@ -1104,7 +1111,8 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, void *stream_op,
     stream_global->recv_initial_metadata_ready =
         op->recv_initial_metadata_ready;
     stream_global->recv_initial_metadata = op->recv_initial_metadata;
-    grpc_chttp2_list_add_check_read_ops(transport_global, stream_global);
+    grpc_chttp2_list_add_check_read_ops(exec_ctx, transport_global,
+                                        stream_global);
   }
 
   if (op->recv_message != NULL) {
@@ -1118,7 +1126,8 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, void *stream_op,
           exec_ctx, transport_global, stream_global,
           transport_global->stream_lookahead, 0);
     }
-    grpc_chttp2_list_add_check_read_ops(transport_global, stream_global);
+    grpc_chttp2_list_add_check_read_ops(exec_ctx, transport_global,
+                                        stream_global);
   }
 
   if (op->recv_trailing_metadata != NULL) {
@@ -1127,7 +1136,8 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, void *stream_op,
         add_closure_barrier(on_complete);
     stream_global->recv_trailing_metadata = op->recv_trailing_metadata;
     stream_global->final_metadata_requested = true;
-    grpc_chttp2_list_add_check_read_ops(transport_global, stream_global);
+    grpc_chttp2_list_add_check_read_ops(exec_ctx, transport_global,
+                                        stream_global);
   }
 
   grpc_chttp2_complete_closure_step(exec_ctx, transport_global, stream_global,
@@ -1357,7 +1367,8 @@ static void decrement_active_streams_locked(
     grpc_chttp2_stream_global *stream_global) {
   if ((stream_global->all_incoming_byte_streams_finished =
            gpr_unref(&stream_global->active_streams))) {
-    grpc_chttp2_list_add_check_read_ops(transport_global, stream_global);
+    grpc_chttp2_list_add_check_read_ops(exec_ctx, transport_global,
+                                        stream_global);
   }
 }
 
@@ -1460,7 +1471,8 @@ static void cancel_from_api(grpc_exec_ctx *exec_ctx,
   }
   if (due_to_error != GRPC_ERROR_NONE && !stream_global->seen_error) {
     stream_global->seen_error = true;
-    grpc_chttp2_list_add_check_read_ops(transport_global, stream_global);
+    grpc_chttp2_list_add_check_read_ops(exec_ctx, transport_global,
+                                        stream_global);
   }
   grpc_chttp2_mark_stream_closed(exec_ctx, transport_global, stream_global, 1,
                                  1, due_to_error);
@@ -1472,7 +1484,8 @@ void grpc_chttp2_fake_status(grpc_exec_ctx *exec_ctx,
                              grpc_status_code status, gpr_slice *slice) {
   if (status != GRPC_STATUS_OK) {
     stream_global->seen_error = true;
-    grpc_chttp2_list_add_check_read_ops(transport_global, stream_global);
+    grpc_chttp2_list_add_check_read_ops(exec_ctx, transport_global,
+                                        stream_global);
   }
   /* stream_global->recv_trailing_metadata_finished gives us a
      last chance replacement: we've received trailing metadata,
@@ -1496,7 +1509,8 @@ void grpc_chttp2_fake_status(grpc_exec_ctx *exec_ctx,
               grpc_mdstr_from_slice(gpr_slice_ref(*slice))));
     }
     stream_global->published_trailing_metadata = true;
-    grpc_chttp2_list_add_check_read_ops(transport_global, stream_global);
+    grpc_chttp2_list_add_check_read_ops(exec_ctx, transport_global,
+                                        stream_global);
   }
   if (slice) {
     gpr_slice_unref(*slice);
@@ -1555,7 +1569,8 @@ void grpc_chttp2_mark_stream_closed(
     GRPC_ERROR_UNREF(error);
     return;
   }
-  grpc_chttp2_list_add_check_read_ops(transport_global, stream_global);
+  grpc_chttp2_list_add_check_read_ops(exec_ctx, transport_global,
+                                      stream_global);
   if (close_reads && !stream_global->read_closed) {
     stream_global->read_closed_error = GRPC_ERROR_REF(error);
     stream_global->read_closed = true;
diff --git a/src/core/ext/transport/chttp2/transport/internal.h b/src/core/ext/transport/chttp2/transport/internal.h
index efac10a93c..d1395e42f8 100644
--- a/src/core/ext/transport/chttp2/transport/internal.h
+++ b/src/core/ext/transport/chttp2/transport/internal.h
@@ -337,6 +337,8 @@ struct grpc_chttp2_transport {
     bool parsing_active;
     /** write execution state of the transport */
     grpc_chttp2_write_state write_state;
+    /** has a check_read_ops been scheduled */
+    bool check_read_ops_scheduled;
   } executor;
 
   /** is the transport destroying itself? */
@@ -380,6 +382,8 @@ struct grpc_chttp2_transport {
   grpc_closure initiate_writing;
   /** closure to finish writing */
   grpc_closure terminate_writing;
+  /** closure to flush read state up the stack */
+  grpc_closure initiate_read_flush_locked;
 
   /** incoming read bytes */
   gpr_slice_buffer read_buffer;
@@ -622,7 +626,7 @@ int grpc_chttp2_list_pop_waiting_for_concurrency(
     grpc_chttp2_stream_global **stream_global);
 
 void grpc_chttp2_list_add_check_read_ops(
-    grpc_chttp2_transport_global *transport_global,
+    grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global,
     grpc_chttp2_stream_global *stream_global);
 bool grpc_chttp2_list_remove_check_read_ops(
     grpc_chttp2_transport_global *transport_global,
diff --git a/src/core/ext/transport/chttp2/transport/parsing.c b/src/core/ext/transport/chttp2/transport/parsing.c
index fbb44ec54a..957aec9253 100644
--- a/src/core/ext/transport/chttp2/transport/parsing.c
+++ b/src/core/ext/transport/chttp2/transport/parsing.c
@@ -87,8 +87,8 @@ void grpc_chttp2_prepare_to_read(
          transport_global->settings[GRPC_SENT_SETTINGS],
          sizeof(transport_parsing->last_sent_settings));
   transport_parsing->max_frame_size =
-      transport_global->settings[GRPC_ACKED_SETTINGS]
-                                [GRPC_CHTTP2_SETTINGS_MAX_FRAME_SIZE];
+      transport_global
+          ->settings[GRPC_ACKED_SETTINGS][GRPC_CHTTP2_SETTINGS_MAX_FRAME_SIZE];
 
   /* update the parsing view of incoming window */
   while (grpc_chttp2_list_pop_unannounced_incoming_window_available(
@@ -177,7 +177,8 @@ void grpc_chttp2_publish_reads(
       stream_global->seen_error = true;
       stream_global->exceeded_metadata_size =
           stream_parsing->exceeded_metadata_size;
-      grpc_chttp2_list_add_check_read_ops(transport_global, stream_global);
+      grpc_chttp2_list_add_check_read_ops(exec_ctx, transport_global,
+                                          stream_global);
     }
 
     /* flush stats to global stream state */
@@ -203,7 +204,8 @@ void grpc_chttp2_publish_reads(
       stream_global->incoming_frames.tail->is_tail = 0;
     }
     if (stream_parsing->data_parser.incoming_frames.head != NULL) {
-      grpc_chttp2_list_add_check_read_ops(transport_global, stream_global);
+      grpc_chttp2_list_add_check_read_ops(exec_ctx, transport_global,
+                                          stream_global);
     }
     grpc_chttp2_incoming_frame_queue_merge(
         &stream_global->incoming_frames,
@@ -219,7 +221,8 @@ void grpc_chttp2_publish_reads(
       GPR_SWAP(grpc_chttp2_incoming_metadata_buffer,
                stream_parsing->metadata_buffer[0],
                stream_global->received_initial_metadata);
-      grpc_chttp2_list_add_check_read_ops(transport_global, stream_global);
+      grpc_chttp2_list_add_check_read_ops(exec_ctx, transport_global,
+                                          stream_global);
     }
     if (!stream_global->published_trailing_metadata &&
         stream_parsing->got_metadata_on_parse[1]) {
@@ -228,7 +231,8 @@ void grpc_chttp2_publish_reads(
       GPR_SWAP(grpc_chttp2_incoming_metadata_buffer,
                stream_parsing->metadata_buffer[1],
                stream_global->received_trailing_metadata);
-      grpc_chttp2_list_add_check_read_ops(transport_global, stream_global);
+      grpc_chttp2_list_add_check_read_ops(exec_ctx, transport_global,
+                                          stream_global);
     }
 
     if (stream_parsing->forced_close_error != GRPC_ERROR_NONE) {
diff --git a/src/core/ext/transport/chttp2/transport/stream_lists.c b/src/core/ext/transport/chttp2/transport/stream_lists.c
index 2eb5f5f632..4dc4968248 100644
--- a/src/core/ext/transport/chttp2/transport/stream_lists.c
+++ b/src/core/ext/transport/chttp2/transport/stream_lists.c
@@ -298,8 +298,15 @@ int grpc_chttp2_list_pop_waiting_for_concurrency(
 }
 
 void grpc_chttp2_list_add_check_read_ops(
-    grpc_chttp2_transport_global *transport_global,
+    grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global,
     grpc_chttp2_stream_global *stream_global) {
+  grpc_chttp2_transport *t = TRANSPORT_FROM_GLOBAL(transport_global);
+  if (!t->executor.check_read_ops_scheduled) {
+    grpc_combiner_execute_finally(exec_ctx, t->executor.combiner,
+                                  &t->initiate_read_flush_locked,
+                                  GRPC_ERROR_NONE, false);
+    t->executor.check_read_ops_scheduled = true;
+  }
   stream_list_add(TRANSPORT_FROM_GLOBAL(transport_global),
                   STREAM_FROM_GLOBAL(stream_global),
                   GRPC_CHTTP2_LIST_CHECK_READ_OPS);