Handle shrinking stream window sizes.

Author: bneradt

proxy/http2/Http2ConnectionState.cc

@@ -1360,14 +1360,36 @@ Http2ConnectionState::create_stream(Http2StreamId new_id, Http2Error &error)
     }
   }
 
-  uint32_t stream_window = 0;
+  uint32_t initial_stream_window        = 0;
+  uint32_t initial_stream_window_target = 0;
   if (client_streamid && this->_has_dynamic_stream_window()) {
-    stream_window = this->_get_configured_receive_session_window_size_in() / (client_streams_in_count.load() + 1);
+    // If we use a flow control mechanism with dynamic stream sizes, then we
+    // may send SETTINGS frames which shrink the stream window. In reality, if we
+    // shrink the window sizes like this, we cannot be sure that the peer will
+    // receive these before sending data. The peer may start up many streams in
+    // quick succession, with each new stream resulting in us sending new
+    // SETTINGS frames to it, thus shrinking the stream window. But the peer
+    // may not receive any of these before sending data on any of these
+    // streams. Therefore we have to initialize all our window sizes at the
+    // maximal value so we don't incorrectly reply with a FLOW_CONTROL_ERROR
+    // RST_STREAM frame in these situations.
+    //
+    // Effectively what we do here is:
+    // 1. initial_stream_window: Initialize our HttpStreams with a generous
+    //    stream window size. This is what is used to detect flow control
+    //    errors.
+    // 2. initial_stream_window_target: Communicate our intended, smaller
+    //    initial stream window size to the peer via a SETTINGS frames.
+    //
+    // This situation is described in RFC 9113 section 6.9.3.
+    initial_stream_window        = this->_get_configured_receive_session_window_size_in();
+    initial_stream_window_target = initial_stream_window / (client_streams_in_count.load() + 1);
   } else {
-    stream_window = this->local_settings.get(HTTP2_SETTINGS_INITIAL_WINDOW_SIZE);
+    initial_stream_window        = this->local_settings.get(HTTP2_SETTINGS_INITIAL_WINDOW_SIZE);
+    initial_stream_window_target = initial_stream_window;
   }
   Http2Stream *new_stream = THREAD_ALLOC_INIT(http2StreamAllocator, this_ethread(), session->get_proxy_session(), new_id,
-                                              peer_settings.get(HTTP2_SETTINGS_INITIAL_WINDOW_SIZE), stream_window);
+                                              peer_settings.get(HTTP2_SETTINGS_INITIAL_WINDOW_SIZE), initial_stream_window);
 
   ink_assert(nullptr != new_stream);
   ink_assert(!stream_list.in(new_stream));
@@ -1382,11 +1404,8 @@ Http2ConnectionState::create_stream(Http2StreamId new_id, Http2Error &error)
     ++client_streams_in_count;
     if (this->_has_dynamic_stream_window()) {
       Http2ConnectionSettings new_settings = local_settings;
-      new_settings.set(HTTP2_SETTINGS_INITIAL_WINDOW_SIZE, stream_window);
+      new_settings.set(HTTP2_SETTINGS_INITIAL_WINDOW_SIZE, initial_stream_window_target);
       send_settings_frame(new_settings);
-
-      // Update all the streams for the new window.
-      this->update_initial_server_rwnd(stream_window);
     }
   } else {
     latest_streamid_out = new_id;
@@ -1477,22 +1496,42 @@ Http2ConnectionState::restart_receiving(Http2Stream *stream)
     return;
   }
 
+  uint32_t initial_stream_window = this->local_settings.get(HTTP2_SETTINGS_INITIAL_WINDOW_SIZE);
+  if (this->_has_dynamic_stream_window()) {
+    // If our stream window size is dynamic, then we have to allow for this situation:
+    //
+    // 1. ATS initializes the stream window size to the configured value via a
+    // SETTINGS frame, say 10 bytes.
+    //
+    // 2. The peer sends a data frame of that size, again 10 bytes.
+    //
+    // 3. At the same time that the peer sends the data frame, we shrink the
+    // stream window size via a SETTINGS frame to a smaller value, say 5 bytes.
+    //
+    // 4. The peer receives the SETTINGS frame and calculates its send window
+    // to be -5 bytes.
+    //
+    // 5. ATS cannot simply send a WINDOW_UPDATE frame of 5 bytes to bring the
+    // window to 5 bytes, the current initial window size for the stream.
+    // Instead, we have to free the peer up to send us data per the maximum value
+    // we initialized our streams with.
+    initial_stream_window = this->_get_configured_receive_session_window_size_in();
+  }
   // If read_vio is buffering data, do not fully update window
-  uint32_t const stream_window = this->local_settings.get(HTTP2_SETTINGS_INITIAL_WINDOW_SIZE);
-  int64_t data_size            = stream->read_vio_read_avail();
-  if (data_size >= stream_window) {
+  int64_t data_size = stream->read_vio_read_avail();
+  if (data_size >= initial_stream_window) {
     return;
   }
 
   // Update the window size for the stream
-  Http2WindowSize diff_size = 0;
-  if (stream->server_rwnd() < 0 && data_size == 0) {
-    // Receive windows can be negative if we sent a SETTINGS frame that
-    // decreased the stream window size mid-stream.
-    diff_size = stream_window - stream->server_rwnd();
-  } else {
-    diff_size = stream_window - std::max(static_cast<int64_t>(stream->server_rwnd()), data_size);
-  }
+  Http2WindowSize const diff_size = initial_stream_window - std::max(static_cast<int64_t>(stream->server_rwnd()), data_size);
+
+  // If our flow control policy has dynamic stream window sizes, then it could
+  // be that we shrank the initial stream window size. In this case, our
+  // calculated window size might be in a state where we are shrinking it to
+  // the new value. If so, this calculation would come out negative and we will
+  // not want to send a WINDOW_UPDATE frame but will rather let the peer
+  // continue to send us data until the window shrinks to the appropriate size.
   if (diff_size > 0) {
     stream->increment_server_rwnd(diff_size);
     this->send_window_update_frame(stream->get_id(), diff_size);
@@ -1637,27 +1676,6 @@ Http2ConnectionState::update_initial_client_rwnd(Http2WindowSize new_size)
   }
 }
 
-void
-Http2ConnectionState::update_initial_server_rwnd(Http2WindowSize new_size)
-{
-  // Update stream level window sizes
-  for (Http2Stream *s = stream_list.head; s; s = static_cast<Http2Stream *>(s->link.next)) {
-    SCOPED_MUTEX_LOCK(lock, s->mutex, this_ethread());
-    // Set the new window size, but take into account the already adjusted
-    // window based on previously sent bytes.
-    //
-    // For example:
-    // 1. ATS initializes the stream window to 10K bytes.
-    // 2. ATS receives 3K bytes from the client. The stream window is now 7K bytes.
-    // 3. ATS sends a SETTINGS frame to the client to update the initial window size to 20K bytes.
-    // 4. The stream window should be updated to 17K bytes: 20K - (10K - 7K).
-    //
-    // Note that if ATS reduces the stream window, this may result in negative
-    // receive window values.
-    s->set_server_rwnd(new_size - (local_settings.get(HTTP2_SETTINGS_INITIAL_WINDOW_SIZE) - s->server_rwnd()));
-  }
-}
-
 void
 Http2ConnectionState::schedule_stream(Http2Stream *stream)
 {

proxy/http2/Http2ConnectionState.h

@@ -115,9 +115,6 @@ class Http2ConnectionState : public Continuation
   /** Update all streams for the peer's newly dictated stream window size. */
   void update_initial_client_rwnd(Http2WindowSize new_size);
 
-  /** Update all streams for our newly dictated stream window size. */
-  void update_initial_server_rwnd(Http2WindowSize new_size);
-
   Http2StreamId get_latest_stream_id_in() const;
   Http2StreamId get_latest_stream_id_out() const;
   int get_stream_requests() const;

proxy/http2/Http2Stream.h

@@ -126,7 +126,6 @@ class Http2Stream : public ProxyTransaction
   Http2StreamState get_state() const;
   bool change_state(uint8_t type, uint8_t flags);
   void set_client_rwnd(Http2WindowSize new_size);
-  void set_server_rwnd(Http2WindowSize new_size);
   bool has_trailing_header() const;
   void set_receive_headers(HTTPHdr &h2_headers);
   MIOBuffer *read_vio_writer() const;
@@ -265,12 +264,6 @@ Http2Stream::set_client_rwnd(Http2WindowSize new_size)
   this->_client_rwnd = new_size;
 }
 
-inline void
-Http2Stream::set_server_rwnd(Http2WindowSize new_size)
-{
-  this->_server_rwnd = new_size;
-}
-
 inline bool
 Http2Stream::has_trailing_header() const
 {

tests/gold_tests/autest-site/trafficserver.test.ext

@@ -338,6 +338,14 @@ def MakeATSProcess(obj, name, command='traffic_server', select_ports=True,
     get_port(p, "manager_port")
     get_port(p, "admin_port")
 
+    if enable_tls:
+        fname = "tls_session_keys.txt"
+        tmpname = os.path.join(log_dir, fname)
+        p.Disk.File(tmpname, id='tls_session_keys')
+        p.Disk.records_config.update({
+            'proxy.config.ssl.keylog_file': tmpname,
+        })
+
     if enable_cache:
         # In records.config, the cache is enabled by default so there's nothing
         # we have to do here to functionally enable it. However, the tests that

tests/gold_tests/h2/http2_flow_control.test.py

@@ -172,6 +172,7 @@ def _configure_client(self, tr):
             # need to set up client expectations.
             return
 
+        # ATS currently always sends a MAX_CONCURRENT_STREAMS setting.
         tr.Processes.Default.Streams.stdout += Testers.ContainsExpression(
             f'MAX_CONCURRENT_STREAMS:{self._expected_max_concurrent_streams_in}',
             "Client should receive a MAX_CONCURRENT_STREAMS setting.")
@@ -181,21 +182,22 @@ def _configure_client(self, tr):
                 f'INITIAL_WINDOW_SIZE:{self._expected_initial_stream_window_size}',
                 "Client should receive an INITIAL_WINDOW_SIZE setting.")
 
-            if self._expected_flow_control_policy == 0:
-                update_window_size = (
-                    self._expected_initial_stream_window_size -
-                    self._default_initial_window_size)
-                if update_window_size > 0:
-                    tr.Processes.Default.Streams.stdout += Testers.ContainsExpression(
-                        f'WINDOW_UPDATE.*id 0: {update_window_size}',
-                        "Client should receive a session WINDOW_UPDATE.")
+        if self._expected_flow_control_policy == 0:
+            update_window_size = (
+                self._expected_initial_stream_window_size -
+                self._default_initial_window_size)
+            if update_window_size > 0:
+                tr.Processes.Default.Streams.stdout += Testers.ContainsExpression(
+                    f'WINDOW_UPDATE.*id 0: {update_window_size}',
+                    "Client should receive a session WINDOW_UPDATE.")
 
         if self._expected_flow_control_policy in (1, 2):
             # Verify the larger window size.
 
             session_window_size = (
                 self._expected_initial_stream_window_size *
                 self._expected_max_concurrent_streams_in)
+
             # ATS will send a WINDOW_UPDATE frame to the client to increase
             # the session window size to the configured value from the default
             # value.
@@ -207,6 +209,16 @@ def _configure_client(self, tr):
             if update_window_size > Http2FlowControlTest._default_initial_window_size:
                 tr.Processes.Default.Streams.stdout += Testers.ContainsExpression(
                     f'WINDOW_UPDATE.*id 0: {update_window_size}',
+                    "Client should receive an initial session WINDOW_UPDATE.")
+            else:
+                # Our test traffic is large enough that eventually we should
+                # send a session WINDOW_UPDATE frame for the smaller window.
+                # It's not clear what it will be in advance though. A 100 byte
+                # session window may not receive a 100 byte WINDOW_UPDATE frame
+                # if the client is sending DATA frames in 10 byte chunks due to
+                # a smaller stream window.
+                tr.Processes.Default.Streams.stdout += Testers.ContainsExpression(
+                    'WINDOW_UPDATE.*id 0: ',
                     "Client should receive a session WINDOW_UPDATE.")
 
             if self._expected_flow_control_policy == 2:
@@ -278,18 +290,18 @@ def run(self):
     flow_control_policy=23)
 test.run()
 test = Http2FlowControlTest(
-    description="Static flow control with a small initial_window_size_in",
-    initial_window_size=500,
+    description="Flow control policy 0 (default): small initial_window_size_in",
+    initial_window_size=500,  # The default is 65 KB.
     flow_control_policy=0)
 test.run()
 test = Http2FlowControlTest(
-    description="Static flow control with a large initial_window_size_in",
+    description="Flow control policy 1: 100 byte session, 10 byte stream windows",
     max_concurrent_streams_in=10,
     initial_window_size=10,
     flow_control_policy=1)
 test.run()
 test = Http2FlowControlTest(
-    description="Dynamic flow control with a small initial_window_size_in",
+    description="Flow control policy 2: 100 byte session, dynamic stream windows",
     max_concurrent_streams_in=10,
     initial_window_size=10,
     flow_control_policy=2)