1 files changed, 194 insertions, 50 deletions

diff --git a/src/core/ext/transport/chttp2/transport/flow_control.h b/src/core/ext/transport/chttp2/transport/flow_control.h
index 8306047dbc..2ee1345260 100644
--- a/src/core/ext/transport/chttp2/transport/flow_control.h
+++ b/src/core/ext/transport/chttp2/transport/flow_control.h
@@ -24,7 +24,8 @@
 
 #include <grpc/support/useful.h>
 #include "src/core/ext/transport/chttp2/transport/http2_settings.h"
-#include "src/core/lib/support/manual_constructor.h"
+#include "src/core/lib/gprpp/abstract.h"
+#include "src/core/lib/gprpp/manual_constructor.h"
 #include "src/core/lib/transport/bdp_estimator.h"
 #include "src/core/lib/transport/pid_controller.h"
 
@@ -43,10 +44,16 @@ namespace grpc_core {
 namespace chttp2 {
 
 static constexpr uint32_t kDefaultWindow = 65535;
+static constexpr int64_t kMaxWindow = (int64_t)((1u << 31) - 1);
+// TODO(ncteisen): Tune this
+static constexpr uint32_t kFrameSize = 1024 * 1024;
 
 class TransportFlowControl;
 class StreamFlowControl;
 
+// Encapsulates a collections of actions the transport needs to take with
+// regard to flow control. Each action comes with urgencies that tell the
+// transport how quickly the action must take place.
 class FlowControlAction {
  public:
   enum class Urgency : uint8_t {
@@ -132,36 +139,122 @@ class FlowControlTrace {
   int64_t announced_window_delta_;
 };
 
-class TransportFlowControl {
+// Fat interface with all methods a flow control implementation needs to
+// support. gRPC C Core does not support pure virtual functions, so instead
+// we abort in any methods which require implementation in the base class.
+class TransportFlowControlBase {
+ public:
+  TransportFlowControlBase() {}
+  virtual ~TransportFlowControlBase() {}
+
+  // Is flow control enabled? This is needed in other codepaths like the checks
+  // in parsing and in writing.
+  virtual bool flow_control_enabled() const { abort(); }
+
+  // Called to check if the transport needs to send a WINDOW_UPDATE frame
+  virtual uint32_t MaybeSendUpdate(bool writing_anyway) { abort(); }
+
+  // Using the protected members, returns and Action to be taken by the
+  // tranport.
+  virtual FlowControlAction MakeAction() { abort(); }
+
+  // Using the protected members, returns and Action to be taken by the
+  // tranport. Also checks for updates to our BDP estimate and acts
+  // accordingly.
+  virtual FlowControlAction PeriodicUpdate() { abort(); }
+
+  // Called to do bookkeeping when a stream owned by this transport sends
+  // data on the wire
+  virtual void StreamSentData(int64_t size) { abort(); }
+
+  // Called to do bookkeeping when a stream owned by this transport receives
+  // data from the wire. Also does error checking for frame size.
+  virtual grpc_error* RecvData(int64_t incoming_frame_size) { abort(); }
+
+  // Called to do bookkeeping when we receive a WINDOW_UPDATE frame.
+  virtual void RecvUpdate(uint32_t size) { abort(); }
+
+  // Returns the BdpEstimator held by this object. Caller is responsible for
+  // checking for nullptr. TODO(ncteisen): consider fully encapsulating all
+  // bdp estimator actions inside TransportFlowControl
+  virtual BdpEstimator* bdp_estimator() { return nullptr; }
+
+  // Getters
+  int64_t remote_window() const { return remote_window_; }
+  virtual int64_t target_window() const { return target_initial_window_size_; }
+  int64_t announced_window() const { return announced_window_; }
+
+  // Used in certain benchmarks in which we don't want FlowControl to be a
+  // factor
+  virtual void TestOnlyForceHugeWindow() {}
+
+  GRPC_ABSTRACT_BASE_CLASS
+
+ protected:
+  friend class ::grpc::testing::TrickledCHTTP2;
+  int64_t remote_window_ = kDefaultWindow;
+  int64_t target_initial_window_size_ = kDefaultWindow;
+  int64_t announced_window_ = kDefaultWindow;
+};
+
+// Implementation of flow control that does NOTHING. Always returns maximum
+// values, never initiates writes, and assumes that the remote peer is doing
+// the same. To be used to narrow down on flow control as the cause of negative
+// performance.
+class TransportFlowControlDisabled final : public TransportFlowControlBase {
+ public:
+  // Maxes out all values
+  TransportFlowControlDisabled(grpc_chttp2_transport* t);
+
+  bool flow_control_enabled() const override { return false; }
+
+  // Never do anything.
+  uint32_t MaybeSendUpdate(bool writing_anyway) override { return 0; }
+  FlowControlAction MakeAction() override { return FlowControlAction(); }
+  FlowControlAction PeriodicUpdate() override { return FlowControlAction(); }
+  void StreamSentData(int64_t size) override {}
+  grpc_error* RecvData(int64_t incoming_frame_size) override {
+    return GRPC_ERROR_NONE;
+  }
+  void RecvUpdate(uint32_t size) override {}
+};
+
+// Implementation of flow control that abides to HTTP/2 spec and attempts
+// to be as performant as possible.
+class TransportFlowControl final : public TransportFlowControlBase {
  public:
   TransportFlowControl(const grpc_chttp2_transport* t, bool enable_bdp_probe);
   ~TransportFlowControl() {}
 
+  bool flow_control_enabled() const override { return true; }
+
   bool bdp_probe() const { return enable_bdp_probe_; }
 
   // returns an announce if we should send a transport update to our peer,
   // else returns zero; writing_anyway indicates if a write would happen
   // regardless of the send - if it is false and this function returns non-zero,
   // this announce will cause a write to occur
-  uint32_t MaybeSendUpdate(bool writing_anyway);
+  uint32_t MaybeSendUpdate(bool writing_anyway) override;
 
   // Reads the flow control data and returns and actionable struct that will
   // tell chttp2 exactly what it needs to do
-  FlowControlAction MakeAction() { return UpdateAction(FlowControlAction()); }
+  FlowControlAction MakeAction() override {
+    return UpdateAction(FlowControlAction());
+  }
 
   // Call periodically (at a low-ish rate, 100ms - 10s makes sense)
   // to perform more complex flow control calculations and return an action
   // to let chttp2 change its parameters
-  FlowControlAction PeriodicUpdate();
+  FlowControlAction PeriodicUpdate() override;
 
-  void StreamSentData(int64_t size) { remote_window_ -= size; }
+  void StreamSentData(int64_t size) override { remote_window_ -= size; }
 
   grpc_error* ValidateRecvData(int64_t incoming_frame_size);
   void CommitRecvData(int64_t incoming_frame_size) {
     announced_window_ -= incoming_frame_size;
   }
 
-  grpc_error* RecvData(int64_t incoming_frame_size) {
+  grpc_error* RecvData(int64_t incoming_frame_size) override {
     FlowControlTrace trace("  data recv", this, nullptr);
     grpc_error* error = ValidateRecvData(incoming_frame_size);
     if (error != GRPC_ERROR_NONE) return error;
@@ -170,18 +263,18 @@ class TransportFlowControl {
   }
 
   // we have received a WINDOW_UPDATE frame for a transport
-  void RecvUpdate(uint32_t size) {
+  void RecvUpdate(uint32_t size) override {
     FlowControlTrace trace("t updt recv", this, nullptr);
     remote_window_ += size;
   }
 
-  int64_t remote_window() const { return remote_window_; }
-  int64_t target_window() const {
+  // See comment above announced_stream_total_over_incoming_window_ for the
+  // logic behind this decision.
+  int64_t target_window() const override {
     return (uint32_t)GPR_MIN((int64_t)((1u << 31) - 1),
                              announced_stream_total_over_incoming_window_ +
                                  target_initial_window_size_);
   }
-  int64_t announced_window() const { return announced_window_; }
 
   const grpc_chttp2_transport* transport() const { return t_; }
 
@@ -201,18 +294,17 @@ class TransportFlowControl {
     }
   }
 
-  BdpEstimator* bdp_estimator() { return &bdp_estimator_; }
+  BdpEstimator* bdp_estimator() override { return &bdp_estimator_; }
 
-  void TestOnlyForceHugeWindow() {
+  void TestOnlyForceHugeWindow() override {
     announced_window_ = 1024 * 1024 * 1024;
     remote_window_ = 1024 * 1024 * 1024;
   }
 
  private:
-  friend class ::grpc::testing::TrickledCHTTP2;
   double TargetLogBdp();
   double SmoothLogBdp(double value);
-  FlowControlAction::Urgency DeltaUrgency(int32_t value,
+  FlowControlAction::Urgency DeltaUrgency(int64_t value,
                                           grpc_chttp2_setting_id setting_id);
 
   FlowControlAction UpdateAction(FlowControlAction action) {
@@ -225,9 +317,6 @@ class TransportFlowControl {
 
   const grpc_chttp2_transport* const t_;
 
-  /** Our bookkeeping for the remote peer's available window */
-  int64_t remote_window_ = kDefaultWindow;
-
   /** calculating what we should give for local window:
       we track the total amount of flow control over initial window size
       across all streams: this is data that we want to receive right now (it
@@ -239,13 +328,6 @@ class TransportFlowControl {
   int64_t announced_stream_total_over_incoming_window_ = 0;
   int64_t announced_stream_total_under_incoming_window_ = 0;
 
-  /** This is out window according to what we have sent to our remote peer. The
-   * difference between this and target window is what we use to decide when
-   * to send WINDOW_UPDATE frames. */
-  int64_t announced_window_ = kDefaultWindow;
-
-  int32_t target_initial_window_size_ = kDefaultWindow;
-
   /** should we probe bdp? */
   const bool enable_bdp_probe_;
 
@@ -257,39 +339,117 @@ class TransportFlowControl {
   grpc_millis last_pid_update_ = 0;
 };
 
-class StreamFlowControl {
+// Fat interface with all methods a stream flow control implementation needs
+// to support. gRPC C Core does not support pure virtual functions, so instead
+// we abort in any methods which require implementation in the base class.
+class StreamFlowControlBase {
+ public:
+  StreamFlowControlBase() {}
+  virtual ~StreamFlowControlBase() {}
+
+  // Updates an action using the protected members.
+  virtual FlowControlAction UpdateAction(FlowControlAction action) { abort(); }
+
+  // Using the protected members, returns an Action for this stream to be
+  // taken by the tranport.
+  virtual FlowControlAction MakeAction() { abort(); }
+
+  // Bookkeeping for when data is sent on this stream.
+  virtual void SentData(int64_t outgoing_frame_size) { abort(); }
+
+  // Bookkeeping and error checking for when data is received by this stream.
+  virtual grpc_error* RecvData(int64_t incoming_frame_size) { abort(); }
+
+  // Called to check if this stream needs to send a WINDOW_UPDATE frame.
+  virtual uint32_t MaybeSendUpdate() { abort(); }
+
+  // Bookkeeping for receiving a WINDOW_UPDATE from for this stream.
+  virtual void RecvUpdate(uint32_t size) { abort(); }
+
+  // Bookkeeping for when a call pulls bytes out of the transport. At this
+  // point we consider the data 'used' and can thus let out peer know we are
+  // ready for more data.
+  virtual void IncomingByteStreamUpdate(size_t max_size_hint,
+                                        size_t have_already) {
+    abort();
+  }
+
+  // Used in certain benchmarks in which we don't want FlowControl to be a
+  // factor
+  virtual void TestOnlyForceHugeWindow() {}
+
+  // Getters
+  int64_t remote_window_delta() { return remote_window_delta_; }
+  int64_t local_window_delta() { return local_window_delta_; }
+  int64_t announced_window_delta() { return announced_window_delta_; }
+
+  GRPC_ABSTRACT_BASE_CLASS
+
+ protected:
+  friend class ::grpc::testing::TrickledCHTTP2;
+  int64_t remote_window_delta_ = 0;
+  int64_t local_window_delta_ = 0;
+  int64_t announced_window_delta_ = 0;
+};
+
+// Implementation of flow control that does NOTHING. Always returns maximum
+// values, never initiates writes, and assumes that the remote peer is doing
+// the same. To be used to narrow down on flow control as the cause of negative
+// performance.
+class StreamFlowControlDisabled : public StreamFlowControlBase {
+ public:
+  FlowControlAction UpdateAction(FlowControlAction action) override {
+    return action;
+  }
+  FlowControlAction MakeAction() override { return FlowControlAction(); }
+  void SentData(int64_t outgoing_frame_size) override {}
+  grpc_error* RecvData(int64_t incoming_frame_size) override {
+    return GRPC_ERROR_NONE;
+  }
+  uint32_t MaybeSendUpdate() override { return 0; }
+  void RecvUpdate(uint32_t size) override {}
+  void IncomingByteStreamUpdate(size_t max_size_hint,
+                                size_t have_already) override {}
+};
+
+// Implementation of flow control that abides to HTTP/2 spec and attempts
+// to be as performant as possible.
+class StreamFlowControl final : public StreamFlowControlBase {
  public:
   StreamFlowControl(TransportFlowControl* tfc, const grpc_chttp2_stream* s);
   ~StreamFlowControl() {
     tfc_->PreUpdateAnnouncedWindowOverIncomingWindow(announced_window_delta_);
   }
 
-  FlowControlAction UpdateAction(FlowControlAction action);
-  FlowControlAction MakeAction() { return UpdateAction(tfc_->MakeAction()); }
+  FlowControlAction UpdateAction(FlowControlAction action) override;
+  FlowControlAction MakeAction() override {
+    return UpdateAction(tfc_->MakeAction());
+  }
 
   // we have sent data on the wire, we must track this in our bookkeeping for
   // the remote peer's flow control.
-  void SentData(int64_t outgoing_frame_size) {
+  void SentData(int64_t outgoing_frame_size) override {
     FlowControlTrace tracer("  data sent", tfc_, this);
     tfc_->StreamSentData(outgoing_frame_size);
     remote_window_delta_ -= outgoing_frame_size;
   }
 
   // we have received data from the wire
-  grpc_error* RecvData(int64_t incoming_frame_size);
+  grpc_error* RecvData(int64_t incoming_frame_size) override;
 
   // returns an announce if we should send a stream update to our peer, else
   // returns zero
-  uint32_t MaybeSendUpdate();
+  uint32_t MaybeSendUpdate() override;
 
   // we have received a WINDOW_UPDATE frame for a stream
-  void RecvUpdate(uint32_t size) {
+  void RecvUpdate(uint32_t size) override {
     FlowControlTrace trace("s updt recv", tfc_, this);
     remote_window_delta_ += size;
   }
 
   // the application is asking for a certain amount of bytes
-  void IncomingByteStreamUpdate(size_t max_size_hint, size_t have_already);
+  void IncomingByteStreamUpdate(size_t max_size_hint,
+                                size_t have_already) override;
 
   int64_t remote_window_delta() const { return remote_window_delta_; }
   int64_t local_window_delta() const { return local_window_delta_; }
@@ -297,14 +457,13 @@ class StreamFlowControl {
 
   const grpc_chttp2_stream* stream() const { return s_; }
 
-  void TestOnlyForceHugeWindow() {
+  void TestOnlyForceHugeWindow() override {
     announced_window_delta_ = 1024 * 1024 * 1024;
     local_window_delta_ = 1024 * 1024 * 1024;
     remote_window_delta_ = 1024 * 1024 * 1024;
   }
 
  private:
-  friend class ::grpc::testing::TrickledCHTTP2;
   TransportFlowControl* const tfc_;
   const grpc_chttp2_stream* const s_;
 
@@ -313,21 +472,6 @@ class StreamFlowControl {
     announced_window_delta_ += change;
     tfc->PostUpdateAnnouncedWindowOverIncomingWindow(announced_window_delta_);
   }
-
-  /** window available for us to send to peer, over or under the initial
-   * window
-   * size of the transport... ie:
-   * remote_window = remote_window_delta + transport.initial_window_size */
-  int64_t remote_window_delta_ = 0;
-
-  /** window available for peer to send to us (as a delta on
-   * transport.initial_window_size)
-   * local_window = local_window_delta + transport.initial_window_size */
-  int64_t local_window_delta_ = 0;
-
-  /** window available for peer to send to us over this stream that we have
-   * announced to the peer */
-  int64_t announced_window_delta_ = 0;
 };
 
 }  // namespace chttp2