Fix flow control

- sending of window updates is now integrated with the primary write
  path, making this far more robust
- iomgr starts up after shutdown correctly again

5 files changed, 104 insertions, 91 deletions

diff --git a/src/core/transport/chttp2/internal.h b/src/core/transport/chttp2/internal.h
index e7901da510..f0eeb6de50 100644
--- a/src/core/transport/chttp2/internal.h
+++ b/src/core/transport/chttp2/internal.h
@@ -60,7 +60,6 @@ typedef enum {
   GRPC_CHTTP2_LIST_WRITABLE,
   GRPC_CHTTP2_LIST_WRITING,
   GRPC_CHTTP2_LIST_WRITTEN,
-  GRPC_CHTTP2_LIST_WRITABLE_WINDOW_UPDATE,
   GRPC_CHTTP2_LIST_PARSING_SEEN,
   GRPC_CHTTP2_LIST_CLOSED_WAITING_FOR_PARSING,
   GRPC_CHTTP2_LIST_CANCELLED_WAITING_FOR_WRITING,
@@ -383,6 +382,8 @@ typedef struct {
   gpr_uint8 published_cancelled;
   /** is this stream in the stream map? (boolean) */
   gpr_uint8 in_stream_map;
+  /** is this stream actively being written? */
+  gpr_uint8 writing_now;
 
   /** stream state already published to the upper layer */
   grpc_stream_state published_state;
@@ -475,11 +476,17 @@ void grpc_chttp2_publish_reads(grpc_chttp2_transport_global *global,
 void grpc_chttp2_list_add_writable_stream(
     grpc_chttp2_transport_global *transport_global,
     grpc_chttp2_stream_global *stream_global);
+void grpc_chttp2_list_add_first_writable_stream(
+    grpc_chttp2_transport_global *transport_global,
+    grpc_chttp2_stream_global *stream_global);
 int grpc_chttp2_list_pop_writable_stream(
     grpc_chttp2_transport_global *transport_global,
     grpc_chttp2_transport_writing *transport_writing,
     grpc_chttp2_stream_global **stream_global,
     grpc_chttp2_stream_writing **stream_writing);
+void grpc_chttp2_list_remove_writable_stream(
+    grpc_chttp2_transport_global *transport_global,
+    grpc_chttp2_stream_global *stream_global);
 
 void grpc_chttp2_list_add_incoming_window_updated(
     grpc_chttp2_transport_global *transport_global,
@@ -511,18 +518,6 @@ int grpc_chttp2_list_pop_written_stream(
     grpc_chttp2_stream_global **stream_global,
     grpc_chttp2_stream_writing **stream_writing);
 
-void grpc_chttp2_list_add_writable_window_update_stream(
-    grpc_chttp2_transport_global *transport_global,
-    grpc_chttp2_stream_global *stream_global);
-int grpc_chttp2_list_pop_writable_window_update_stream(
-    grpc_chttp2_transport_global *transport_global,
-    grpc_chttp2_transport_writing *transport_writing,
-    grpc_chttp2_stream_global **stream_global,
-    grpc_chttp2_stream_writing **stream_writing);
-void grpc_chttp2_list_remove_writable_window_update_stream(
-    grpc_chttp2_transport_global *transport_global,
-    grpc_chttp2_stream_global *stream_global);
-
 void grpc_chttp2_list_add_parsing_seen_stream(
     grpc_chttp2_transport_parsing *transport_parsing,
     grpc_chttp2_stream_parsing *stream_parsing);
diff --git a/src/core/transport/chttp2/parsing.c b/src/core/transport/chttp2/parsing.c
index 904b9afce7..9105746567 100644
--- a/src/core/transport/chttp2/parsing.c
+++ b/src/core/transport/chttp2/parsing.c
@@ -182,8 +182,7 @@ void grpc_chttp2_publish_reads(
       stream_global->max_recv_bytes -= 
           stream_parsing->incoming_window_delta;
       stream_parsing->incoming_window_delta = 0;
-      grpc_chttp2_list_add_writable_window_update_stream(transport_global,
-                                                         stream_global);
+      grpc_chttp2_list_add_writable_stream(transport_global, stream_global);
     }
 
     /* update outgoing flow control window */
diff --git a/src/core/transport/chttp2/stream_lists.c b/src/core/transport/chttp2/stream_lists.c
index 590f6abfbc..9e68c1e146 100644
--- a/src/core/transport/chttp2/stream_lists.c
+++ b/src/core/transport/chttp2/stream_lists.c
@@ -108,6 +108,23 @@ static void stream_list_maybe_remove(grpc_chttp2_transport *t,
   }
 }
 
+static void stream_list_add_head(grpc_chttp2_transport *t,
+                                 grpc_chttp2_stream *s,
+                                 grpc_chttp2_stream_list_id id) {
+  grpc_chttp2_stream *old_head;
+  GPR_ASSERT(!s->included[id]);
+  old_head = t->lists[id].head;
+  s->links[id].next = old_head;
+  s->links[id].prev = NULL;
+  if (old_head) {
+    old_head->links[id].prev = s;
+  } else {
+    t->lists[id].tail = s;
+  }
+  t->lists[id].head = s;
+  s->included[id] = 1;
+}
+
 static void stream_list_add_tail(grpc_chttp2_transport *t,
                                  grpc_chttp2_stream *s,
                                  grpc_chttp2_stream_list_id id) {
@@ -119,7 +136,6 @@ static void stream_list_add_tail(grpc_chttp2_transport *t,
   if (old_tail) {
     old_tail->links[id].next = s;
   } else {
-    s->links[id].prev = NULL;
     t->lists[id].head = s;
   }
   t->lists[id].tail = s;
@@ -144,6 +160,18 @@ void grpc_chttp2_list_add_writable_stream(
                   STREAM_FROM_GLOBAL(stream_global), GRPC_CHTTP2_LIST_WRITABLE);
 }
 
+void grpc_chttp2_list_add_first_writable_stream(
+    grpc_chttp2_transport_global *transport_global,
+    grpc_chttp2_stream_global *stream_global) {
+  GPR_ASSERT(stream_global->id != 0);
+  gpr_log(GPR_DEBUG, "add:%d:%d:%d:%d", stream_global->id,
+          stream_global->write_state, stream_global->in_stream_map,
+          stream_global->read_closed);
+  stream_list_add_head(TRANSPORT_FROM_GLOBAL(transport_global),
+                       STREAM_FROM_GLOBAL(stream_global),
+                       GRPC_CHTTP2_LIST_WRITABLE);
+}
+
 int grpc_chttp2_list_pop_writable_stream(
     grpc_chttp2_transport_global *transport_global,
     grpc_chttp2_transport_writing *transport_writing,
@@ -157,6 +185,14 @@ int grpc_chttp2_list_pop_writable_stream(
   return r;
 }
 
+void grpc_chttp2_list_remove_writable_stream(
+    grpc_chttp2_transport_global *transport_global,
+    grpc_chttp2_stream_global *stream_global) {
+  stream_list_maybe_remove(TRANSPORT_FROM_GLOBAL(transport_global),
+                           STREAM_FROM_GLOBAL(stream_global),
+                           GRPC_CHTTP2_LIST_WRITABLE);
+}
+
 void grpc_chttp2_list_add_writing_stream(
     grpc_chttp2_transport_writing *transport_writing,
     grpc_chttp2_stream_writing *stream_writing) {
@@ -202,36 +238,6 @@ int grpc_chttp2_list_pop_written_stream(
   return r;
 }
 
-void grpc_chttp2_list_add_writable_window_update_stream(
-    grpc_chttp2_transport_global *transport_global,
-    grpc_chttp2_stream_global *stream_global) {
-  GPR_ASSERT(stream_global->id != 0);
-  stream_list_add(TRANSPORT_FROM_GLOBAL(transport_global),
-                  STREAM_FROM_GLOBAL(stream_global),
-                  GRPC_CHTTP2_LIST_WRITABLE_WINDOW_UPDATE);
-}
-
-int grpc_chttp2_list_pop_writable_window_update_stream(
-    grpc_chttp2_transport_global *transport_global,
-    grpc_chttp2_transport_writing *transport_writing,
-    grpc_chttp2_stream_global **stream_global,
-    grpc_chttp2_stream_writing **stream_writing) {
-  grpc_chttp2_stream *stream;
-  int r = stream_list_pop(TRANSPORT_FROM_GLOBAL(transport_global), &stream,
-                          GRPC_CHTTP2_LIST_WRITABLE_WINDOW_UPDATE);
-  *stream_global = &stream->global;
-  *stream_writing = &stream->writing;
-  return r;
-}
-
-void grpc_chttp2_list_remove_writable_window_update_stream(
-    grpc_chttp2_transport_global *transport_global,
-    grpc_chttp2_stream_global *stream_global) {
-  stream_list_maybe_remove(TRANSPORT_FROM_GLOBAL(transport_global),
-                           STREAM_FROM_GLOBAL(stream_global),
-                           GRPC_CHTTP2_LIST_WRITABLE_WINDOW_UPDATE);
-}
-
 void grpc_chttp2_list_add_parsing_seen_stream(
     grpc_chttp2_transport_parsing *transport_parsing,
     grpc_chttp2_stream_parsing *stream_parsing) {
diff --git a/src/core/transport/chttp2/writing.c b/src/core/transport/chttp2/writing.c
index 041c4efd1f..54d38f2841 100644
--- a/src/core/transport/chttp2/writing.c
+++ b/src/core/transport/chttp2/writing.c
@@ -44,6 +44,7 @@ int grpc_chttp2_unlocking_check_writes(
     grpc_chttp2_transport_writing *transport_writing) {
   grpc_chttp2_stream_global *stream_global;
   grpc_chttp2_stream_writing *stream_writing;
+  grpc_chttp2_stream_global *first_reinserted_stream = NULL;
   gpr_uint32 window_delta;
 
   /* simple writes are queued to qbuf, and flushed here */
@@ -64,51 +65,53 @@ int grpc_chttp2_unlocking_check_writes(
   }
 
   /* 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 (transport_global->outgoing_window > 0 &&
-         grpc_chttp2_list_pop_writable_stream(transport_global,
-                                              transport_writing, &stream_global,
-                                              &stream_writing)) {
+     (according to available window sizes) and add to the output buffer */
+  while (grpc_chttp2_list_pop_writable_stream(
+      transport_global, transport_writing, &stream_global, &stream_writing)) {
+    if (stream_global == first_reinserted_stream) {
+      /* prevent infinite loop */
+      grpc_chttp2_list_add_first_writable_stream(transport_global,
+                                                 stream_global);
+      break;
+    }
+
     stream_writing->id = stream_global->id;
-    window_delta = grpc_chttp2_preencode(
-        stream_global->outgoing_sopb->ops, &stream_global->outgoing_sopb->nops,
-        GPR_MIN(transport_global->outgoing_window,
-                stream_global->outgoing_window),
-        &stream_writing->sopb);
-    GRPC_CHTTP2_FLOWCTL_TRACE_TRANSPORT(
-        "write", transport_global, outgoing_window, -(gpr_int64)window_delta);
-    GRPC_CHTTP2_FLOWCTL_TRACE_STREAM("write", transport_global, stream_global,
-                                     outgoing_window, -(gpr_int64)window_delta);
-    transport_global->outgoing_window -= window_delta;
-    stream_global->outgoing_window -= window_delta;
-
-    if (stream_global->write_state == GRPC_WRITE_STATE_QUEUED_CLOSE &&
-        stream_global->outgoing_sopb->nops == 0) {
-      if (!transport_global->is_client && !stream_global->read_closed) {
-        stream_writing->send_closed = GRPC_SEND_CLOSED_WITH_RST_STREAM;
-      } else {
-        stream_writing->send_closed = GRPC_SEND_CLOSED;
+    stream_writing->send_closed = GRPC_DONT_SEND_CLOSED;
+
+    if (stream_global->outgoing_sopb) {
+      window_delta =
+          grpc_chttp2_preencode(stream_global->outgoing_sopb->ops,
+                                &stream_global->outgoing_sopb->nops,
+                                GPR_MIN(transport_global->outgoing_window,
+                                        stream_global->outgoing_window),
+                                &stream_writing->sopb);
+      GRPC_CHTTP2_FLOWCTL_TRACE_TRANSPORT(
+          "write", transport_global, outgoing_window, -(gpr_int64)window_delta);
+      GRPC_CHTTP2_FLOWCTL_TRACE_STREAM("write", transport_global, stream_global,
+                                       outgoing_window,
+                                       -(gpr_int64)window_delta);
+      transport_global->outgoing_window -= window_delta;
+      stream_global->outgoing_window -= window_delta;
+
+      if (stream_global->write_state == GRPC_WRITE_STATE_QUEUED_CLOSE &&
+          stream_global->outgoing_sopb->nops == 0) {
+        if (!transport_global->is_client && !stream_global->read_closed) {
+          stream_writing->send_closed = GRPC_SEND_CLOSED_WITH_RST_STREAM;
+        } else {
+          stream_writing->send_closed = GRPC_SEND_CLOSED;
+        }
       }
-    }
-    if (stream_writing->sopb.nops > 0 ||
-        stream_writing->send_closed != GRPC_DONT_SEND_CLOSED) {
-      grpc_chttp2_list_add_writing_stream(transport_writing, stream_writing);
-    }
 
-    if (stream_global->outgoing_window > 0 &&
-        stream_global->outgoing_sopb->nops != 0) {
-      grpc_chttp2_list_add_writable_stream(transport_global, stream_global);
+      if (stream_global->outgoing_window > 0 &&
+          stream_global->outgoing_sopb->nops != 0) {
+        grpc_chttp2_list_add_writable_stream(transport_global, stream_global);
+        if (first_reinserted_stream == NULL &&
+            transport_global->outgoing_window == 0) {
+          first_reinserted_stream = stream_global;
+        }
+      }
     }
-  }
 
-  /* for each grpc_chttp2_stream that wants to update its window, add that
-   * window here */
-  while (grpc_chttp2_list_pop_writable_window_update_stream(transport_global,
-                                                            transport_writing,
-                                                            &stream_global,
-                                                            &stream_writing)) {
-    stream_writing->id = stream_global->id;
     if (!stream_global->read_closed && stream_global->unannounced_incoming_window > 0) {
       stream_writing->announce_window = stream_global->unannounced_incoming_window;
       GRPC_CHTTP2_FLOWCTL_TRACE_STREAM("write", transport_global, stream_global,
@@ -119,6 +122,11 @@ int grpc_chttp2_unlocking_check_writes(
       stream_global->unannounced_incoming_window = 0;
       grpc_chttp2_list_add_incoming_window_updated(transport_global,
                                                    stream_global);
+      stream_global->writing_now = 1;
+      grpc_chttp2_list_add_writing_stream(transport_writing, stream_writing);
+    } else if (stream_writing->sopb.nops > 0 ||
+               stream_writing->send_closed != GRPC_DONT_SEND_CLOSED) {
+      stream_global->writing_now = 1;
       grpc_chttp2_list_add_writing_stream(transport_writing, stream_writing);
     }
   }
@@ -206,6 +214,8 @@ void grpc_chttp2_cleanup_writing(
 
   while (grpc_chttp2_list_pop_written_stream(
       transport_global, transport_writing, &stream_global, &stream_writing)) {
+    GPR_ASSERT(stream_global->writing_now);
+    stream_global->writing_now = 0;
     if (stream_global->outgoing_sopb != NULL &&
         stream_global->outgoing_sopb->nops == 0) {
       stream_global->outgoing_sopb = NULL;
@@ -219,6 +229,9 @@ void grpc_chttp2_cleanup_writing(
       }
       grpc_chttp2_list_add_read_write_state_changed(transport_global,
                                                     stream_global);
+    } else if (stream_global->read_closed) {
+      grpc_chttp2_list_add_read_write_state_changed(transport_global,
+                                                    stream_global);
     }
   }
   transport_writing->outbuf.count = 0;
diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c
index eb435a2ee8..0540252546 100644
--- a/src/core/transport/chttp2_transport.c
+++ b/src/core/transport/chttp2_transport.c
@@ -395,7 +395,6 @@ static void destroy_stream(grpc_transport *gt, grpc_stream *gs) {
   }
 
   grpc_chttp2_list_remove_incoming_window_updated(&t->global, &s->global);
-  grpc_chttp2_list_remove_writable_window_update_stream(&t->global, &s->global);
 
   gpr_mu_unlock(&t->mu);
 
@@ -576,8 +575,6 @@ static void maybe_start_some_streams(
     grpc_chttp2_list_add_incoming_window_updated(transport_global,
                                                  stream_global);
     grpc_chttp2_list_add_writable_stream(transport_global, stream_global);
-    grpc_chttp2_list_add_writable_window_update_stream(transport_global,
-                                                       stream_global);
 
   }
   /* cancel out streams that will never be started */
@@ -643,8 +640,7 @@ static void perform_stream_op_locked(
     if (stream_global->id != 0) {
       grpc_chttp2_list_add_read_write_state_changed(transport_global,
                                                     stream_global);
-      grpc_chttp2_list_add_writable_window_update_stream(transport_global,
-                                                         stream_global);
+      grpc_chttp2_list_add_writable_stream(transport_global, stream_global);
     }
   }
 
@@ -752,6 +748,7 @@ static void remove_stream(grpc_chttp2_transport *t, gpr_uint32 id) {
   if (!s) {
     s = grpc_chttp2_stream_map_delete(&t->new_stream_map, id);
   }
+  grpc_chttp2_list_remove_writable_stream(&t->global, &s->global);
   GPR_ASSERT(s);
   s->global.in_stream_map = 0;
   if (t->parsing.incoming_stream == &s->parsing) {
@@ -833,6 +830,9 @@ static void unlock_check_read_write_state(grpc_chttp2_transport *t) {
     if (!stream_global->publish_sopb) {
       continue;
     }
+    if (stream_global->writing_now) {
+      continue;
+    }
     /* FIXME(ctiller): we include in_stream_map in our computation of
        whether the stream is write-closed. This is completely bogus,
        but has the effect of delaying stream-closed until the stream