Fixes

src/core/ext/transport/chttp2/transport/flow_control.cc

@@ -358,10 +358,15 @@ FlowControlAction TransportFlowControl::UpdateForBdp(grpc_exec_ctx* exec_ctx,
           frame_size);
     }
   }
+  return action;
 }
 
 FlowControlAction TransportFlowControl::MakeAction(grpc_exec_ctx* exec_ctx) {
   FlowControlAction action;
+  if (announced_window_ < target_window() / 2) {
+    action.set_send_transport_update(
+        FlowControlAction::Urgency::UPDATE_IMMEDIATELY);
+  }
   action = UpdateForBdp(exec_ctx, action);
   return action;
 }

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

@@ -123,8 +123,6 @@ class FlowControlTrace {
   int64_t remote_window_delta_;
   int64_t local_window_delta_;
   int64_t announced_window_delta_;
-  uint32_t local_init_window_;
-  uint32_t local_max_frame_;
 };
 
 class TransportFlowControl {
@@ -137,7 +135,8 @@ class TransportFlowControl {
   }
 
   // toggle bdp probing
-  void SetBdpProbe(bool enable);
+  // TODO(ctiller): make this safe to dynamically toggle
+  void SetBdpProbe(bool enable) { enable_bdp_probe_ = enable; }
 
   // returns an announce if we should send a transport update to our peer,
   // else returns zero; writing_anyway indicates if a write would happen

src/core/ext/transport/chttp2/transport/frame_settings.cc

@@ -202,13 +202,13 @@ grpc_error *grpc_chttp2_settings_parser_parse(grpc_exec_ctx *exec_ctx, void *p,
           }
           if (id == GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE &&
               parser->incoming_settings[id] != parser->value) {
-            t->flow_control.initial_window_update +=
+            t->initial_window_update +=
                 (int64_t)parser->value - parser->incoming_settings[id];
             if (GRPC_TRACER_ON(grpc_http_trace) ||
                 GRPC_TRACER_ON(grpc_flowctl_trace)) {
               gpr_log(GPR_DEBUG, "%p[%s] adding %d for initial_window change",
                       t, t->is_client ? "cli" : "svr",
-                      (int)t->flow_control.initial_window_update);
+                      (int)t->initial_window_update);
             }
           }
           parser->incoming_settings[id] = parser->value;

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

@@ -355,14 +355,15 @@ static grpc_error *init_data_frame_parser(grpc_exec_ctx *exec_ctx,
   grpc_chttp2_stream *s =
       grpc_chttp2_parsing_lookup_stream(t, t->incoming_stream_id);
   grpc_error *err = GRPC_ERROR_NONE;
-  err = grpc_chttp2_flowctl_recv_data(&t->flow_control,
-                                      s == NULL ? NULL : &s->flow_control,
-                                      t->incoming_frame_size);
-  grpc_chttp2_act_on_flowctl_action(
-      exec_ctx,
-      grpc_chttp2_flowctl_get_action(exec_ctx, &t->flow_control,
-                                     s == NULL ? NULL : &s->flow_control),
-      t, s);
+  grpc_core::chttp2::FlowControlAction action;
+  if (s == nullptr) {
+    err = t->flow_control->RecvData(t->incoming_frame_size);
+    action = t->flow_control->MakeAction(exec_ctx);
+  } else {
+    err = s->flow_control->RecvData(t->incoming_frame_size);
+    action = s->flow_control->MakeAction(exec_ctx);
+  }
+  grpc_chttp2_act_on_flowctl_action(exec_ctx, action, t, s);
   if (err != GRPC_ERROR_NONE) {
     goto error_handler;
   }

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

@@ -146,13 +146,13 @@ static void report_stall(grpc_chttp2_transport *t, grpc_chttp2_stream *s,
       s->flow_controlled_bytes_flowed,
       t->settings[GRPC_ACKED_SETTINGS]
                  [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE],
-      t->flow_control.remote_window,
+      t->flow_control->remote_window(),
       (uint32_t)GPR_MAX(
           0,
-          s->flow_control.remote_window_delta +
+          s->flow_control->remote_window_delta() +
               (int64_t)t->settings[GRPC_PEER_SETTINGS]
                                   [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE]),
-      s->flow_control.remote_window_delta);
+      s->flow_control->remote_window_delta());
 }
 
 static bool stream_ref_if_not_destroyed(gpr_refcount *r) {
@@ -216,8 +216,7 @@ class WriteContext {
 
   void FlushWindowUpdates(grpc_exec_ctx *exec_ctx) {
     uint32_t transport_announce =
-        grpc_chttp2_flowctl_maybe_send_transport_update(&t_->flow_control,
-                                                        t_->outbuf.count > 0);
+        t_->flow_control->MaybeSendUpdate(t_->outbuf.count > 0);
     if (transport_announce) {
       grpc_transport_one_way_stats throwaway_stats;
       grpc_slice_buffer_add(
@@ -311,7 +310,7 @@ class DataSendContext {
 
   uint32_t stream_remote_window() const {
     return (uint32_t)GPR_MAX(
-        0, s_->flow_control.remote_window_delta +
+        0, s_->flow_control->remote_window_delta() +
                (int64_t)t_->settings[GRPC_PEER_SETTINGS]
                                     [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE]);
   }
@@ -319,7 +318,7 @@ class DataSendContext {
   uint32_t max_outgoing() const {
     return (uint32_t)GPR_MIN(
         t_->settings[GRPC_PEER_SETTINGS][GRPC_CHTTP2_SETTINGS_MAX_FRAME_SIZE],
-        GPR_MIN(stream_remote_window(), t_->flow_control.remote_window));
+        GPR_MIN(stream_remote_window(), t_->flow_control->remote_window()));
   }
 
   bool AnyOutgoing() const { return max_outgoing() != 0; }
@@ -351,8 +350,7 @@ class DataSendContext {
                      grpc_metadata_batch_is_empty(s_->send_trailing_metadata);
     grpc_chttp2_encode_data(s_->id, &s_->compressed_data_buffer, send_bytes,
                             is_last_frame_, &s_->stats.outgoing, &t_->outbuf);
-    grpc_chttp2_flowctl_sent_data(&t_->flow_control, &s_->flow_control,
-                                  send_bytes);
+    s_->flow_control->SentData(send_bytes);
     if (s_->compressed_data_buffer.length == 0) {
       s_->sending_bytes += s_->uncompressed_data_size;
     }
@@ -399,8 +397,8 @@ class StreamWriteContext {
         gpr_log(GPR_DEBUG, "W:%p %s[%d] im-(sent,send)=(%d,%d) announce=%d", t_,
                 t_->is_client ? "CLIENT" : "SERVER", s->id,
                 s->sent_initial_metadata, s->send_initial_metadata != NULL,
-                (int)(s->flow_control.local_window_delta -
-                      s->flow_control.announced_window_delta)));
+                (int)(s->flow_control->local_window_delta() -
+                      s->flow_control->announced_window_delta())));
   }
 
   void FlushInitialMetadata(grpc_exec_ctx *exec_ctx) {
@@ -446,8 +444,7 @@ class StreamWriteContext {
 
   void FlushWindowUpdates(grpc_exec_ctx *exec_ctx) {
     /* send any window updates */
-    uint32_t stream_announce = grpc_chttp2_flowctl_maybe_send_stream_update(
-        &t_->flow_control, &s_->flow_control);
+    const uint32_t stream_announce = s_->flow_control->MaybeSendUpdate();
     if (stream_announce == 0) return;
 
     grpc_slice_buffer_add(
@@ -468,10 +465,10 @@ class StreamWriteContext {
     DataSendContext data_send_context(write_context_, t_, s_);
 
     if (!data_send_context.AnyOutgoing()) {
-      if (t_->flow_control.remote_window == 0) {
+      if (t_->flow_control->remote_window() <= 0) {
         report_stall(t_, s_, "transport");
         grpc_chttp2_list_add_stalled_by_transport(t_, s_);
-      } else if (data_send_context.stream_remote_window() == 0) {
+      } else if (data_send_context.stream_remote_window() <= 0) {
         report_stall(t_, s_, "stream");
         grpc_chttp2_list_add_stalled_by_stream(t_, s_);
       }
@@ -587,7 +584,7 @@ grpc_chttp2_begin_write_result grpc_chttp2_begin_write(
   ctx.FlushQueuedBuffers(exec_ctx);
   ctx.EnactHpackSettings(exec_ctx);
 
-  if (t->flow_control.remote_window > 0) {
+  if (t->flow_control->remote_window() > 0) {
     ctx.UpdateStreamsNoLongerStalled();
   }