Fix HTTP/2 Buffering Issues

include/proxy/http2/Http2CommonSession.h

@@ -36,15 +36,17 @@
 // HTTP2_SESSION_EVENT_RECV   Http2Frame *          Received a frame
 // HTTP2_SESSION_EVENT_PRIO   Http2Frame *          Send this priority frame
 // HTTP2_SESSION_EVENT_DATA   Http2Frame *          Send the data frames in the stream
+// HTTP2_SESSION_EVENT_XMIT   Http2CommonSession *  Try retransmitting frames.
 
 #define HTTP2_SESSION_EVENT_INIT          (HTTP2_SESSION_EVENTS_START + 1)
 #define HTTP2_SESSION_EVENT_FINI          (HTTP2_SESSION_EVENTS_START + 2)
 #define HTTP2_SESSION_EVENT_RECV          (HTTP2_SESSION_EVENTS_START + 3)
-#define HTTP2_SESSION_EVENT_XMIT          (HTTP2_SESSION_EVENTS_START + 4)
+#define HTTP2_SESSION_EVENT_PRIO          (HTTP2_SESSION_EVENTS_START + 4)
 #define HTTP2_SESSION_EVENT_DATA          (HTTP2_SESSION_EVENTS_START + 5)
-#define HTTP2_SESSION_EVENT_SHUTDOWN_INIT (HTTP2_SESSION_EVENTS_START + 6)
-#define HTTP2_SESSION_EVENT_SHUTDOWN_CONT (HTTP2_SESSION_EVENTS_START + 7)
-#define HTTP2_SESSION_EVENT_REENABLE      (HTTP2_SESSION_EVENTS_START + 8)
+#define HTTP2_SESSION_EVENT_XMIT          (HTTP2_SESSION_EVENTS_START + 6)
+#define HTTP2_SESSION_EVENT_SHUTDOWN_INIT (HTTP2_SESSION_EVENTS_START + 7)
+#define HTTP2_SESSION_EVENT_SHUTDOWN_CONT (HTTP2_SESSION_EVENTS_START + 8)
+#define HTTP2_SESSION_EVENT_REENABLE      (HTTP2_SESSION_EVENTS_START + 9)
 
 enum class Http2SessionCod : int {
   NOT_PROVIDED,

include/proxy/http2/Http2ConnectionState.h

@@ -158,6 +158,8 @@ class Http2ConnectionState : public Continuation
   void schedule_stream_to_send_priority_frames(Http2Stream *stream);
   void send_data_frames_depends_on_priority();
   void schedule_stream_to_send_data_frames(Http2Stream *stream);
+  void schedule_retransmit(ink_hrtime t);
+  void cancel_retransmit();
   void send_data_frames(Http2Stream *stream);
   Http2SendDataFrameResult send_a_data_frame(Http2Stream *stream, size_t &payload_length);
   void send_headers_frame(Http2Stream *stream);
@@ -398,6 +400,7 @@ class Http2ConnectionState : public Continuation
   Event *shutdown_cont_event        = nullptr;
   Event *fini_event                 = nullptr;
   Event *zombie_event               = nullptr;
+  Event *retransmit_event           = nullptr;
 
   uint32_t configured_max_settings_frames_per_minute   = 0;
   uint32_t configured_max_ping_frames_per_minute       = 0;

src/proxy/http2/Http2ClientSession.cc

@@ -210,7 +210,7 @@ Http2ClientSession::main_event_handler(int event, void *edata)
     retval = 0;
     break;
 
-  case HTTP2_SESSION_EVENT_XMIT:
+  case HTTP2_SESSION_EVENT_PRIO:
   default:
     Http2SsnDebug("unexpected event=%d edata=%p", event, edata);
     ink_release_assert(0);

src/proxy/http2/Http2CommonSession.cc

@@ -158,6 +158,11 @@ Http2CommonSession::xmit(const Http2TxFrame &frame, bool flush)
     // A frame size can be 16MB at maximum so blocks can be added, but that's fine.
     if (this->_pending_sending_data_size >= this->_write_size_threshold) {
       flush = true;
+    } else {
+      Note("Calling schedule_transmit because write threshold is not exceeded.");
+      // Observe that schedule_transmit will only schedule the first time we
+      // don't flush because the threshold is not met.
+      this->connection_state.schedule_retransmit(HRTIME_MSECONDS(Http2::write_time_threshold));
     }
   }
   if (flush) {
@@ -170,6 +175,7 @@ Http2CommonSession::xmit(const Http2TxFrame &frame, bool flush)
 void
 Http2CommonSession::flush()
 {
+  this->connection_state.cancel_retransmit();
   if (this->_pending_sending_data_size > 0) {
     this->_pending_sending_data_size = 0;
     this->_write_buffer_last_flush   = ink_get_hrtime();

src/proxy/http2/Http2ConnectionState.cc

@@ -1354,6 +1354,9 @@ Http2ConnectionState::destroy()
   if (zombie_event) {
     zombie_event->cancel();
   }
+  if (retransmit_event) {
+    retransmit_event->cancel();
+  }
   // release the mutex after the events are cancelled and sessions are destroyed.
   mutex = nullptr; // magic happens - assigning to nullptr frees the ProxyMutex
 }
@@ -1430,6 +1433,8 @@ Http2ConnectionState::main_event_handler(int event, void *edata)
     ink_release_assert(zombie_event == nullptr);
   } else if (edata == fini_event) {
     fini_event = nullptr;
+  } else if (edata == retransmit_event) {
+    retransmit_event = nullptr;
   }
   ++recursion;
   switch (event) {
@@ -1445,7 +1450,7 @@ Http2ConnectionState::main_event_handler(int event, void *edata)
     SET_HANDLER(&Http2ConnectionState::state_closed);
   } break;
 
-  case HTTP2_SESSION_EVENT_XMIT: {
+  case HTTP2_SESSION_EVENT_PRIO: {
     REMEMBER(event, this->recursion);
     SCOPED_MUTEX_LOCK(lock, this->mutex, this_ethread());
     send_data_frames_depends_on_priority();
@@ -1459,6 +1464,13 @@ Http2ConnectionState::main_event_handler(int event, void *edata)
     _data_scheduled = false;
   } break;
 
+  case HTTP2_SESSION_EVENT_XMIT: {
+    REMEMBER(event, this->recursion);
+    SCOPED_MUTEX_LOCK(lock, this->mutex, this_ethread());
+    Note("Flushing due to XMIT event");
+    this->session->flush();
+  } break;
+
   // Initiate a graceful shutdown
   case HTTP2_SESSION_EVENT_SHUTDOWN_INIT: {
     REMEMBER(event, this->recursion);
@@ -1522,6 +1534,8 @@ Http2ConnectionState::state_closed(int event, void *edata)
     fini_event = nullptr;
   } else if (edata == shutdown_cont_event) {
     shutdown_cont_event = nullptr;
+  } else if (edata == retransmit_event) {
+    retransmit_event = nullptr;
   }
   return 0;
 }
@@ -2037,7 +2051,7 @@ Http2ConnectionState::schedule_stream_to_send_priority_frames(Http2Stream *strea
     _priority_scheduled = true;
 
     SET_HANDLER(&Http2ConnectionState::main_event_handler);
-    this_ethread()->schedule_imm_local((Continuation *)this, HTTP2_SESSION_EVENT_XMIT);
+    this_ethread()->schedule_imm_local((Continuation *)this, HTTP2_SESSION_EVENT_PRIO);
   }
 }
 
@@ -2056,6 +2070,31 @@ Http2ConnectionState::schedule_stream_to_send_data_frames(Http2Stream *stream)
   }
 }
 
+void
+Http2ConnectionState::schedule_retransmit(ink_hrtime t)
+{
+  Http2StreamDebug(session, 0, "Scheduling retransmitting data frames");
+  SCOPED_MUTEX_LOCK(lock, this->mutex, this_ethread());
+
+  if (retransmit_event == nullptr) {
+    Note("Scheduling retransmit in %" PRId64 "ms", t / HRTIME_MSECOND);
+
+    SET_HANDLER(&Http2ConnectionState::main_event_handler);
+    retransmit_event = this_ethread()->schedule_in((Continuation *)this, t, HTTP2_SESSION_EVENT_XMIT);
+  }
+}
+
+void
+Http2ConnectionState::cancel_retransmit()
+{
+  Http2StreamDebug(session, 0, "Scheduling retransmitting data frames");
+  SCOPED_MUTEX_LOCK(lock, this->mutex, this_ethread());
+  if (retransmit_event != nullptr) {
+    retransmit_event->cancel();
+    retransmit_event = nullptr;
+  }
+}
+
 void
 Http2ConnectionState::send_data_frames_depends_on_priority()
 {
@@ -2098,7 +2137,7 @@ Http2ConnectionState::send_data_frames_depends_on_priority()
     break;
   }
 
-  this_ethread()->schedule_imm_local((Continuation *)this, HTTP2_SESSION_EVENT_XMIT);
+  this_ethread()->schedule_imm_local((Continuation *)this, HTTP2_SESSION_EVENT_PRIO);
   return;
 }
 

src/proxy/http2/Http2ServerSession.cc

@@ -22,6 +22,7 @@
  */
 
 #include "proxy/http2/Http2ServerSession.h"
+#include "iocore/net/TLSSNISupport.h"
 #include "proxy/http/HttpDebugNames.h"
 #include "tscore/ink_base64.h"
 #include "proxy/http2/Http2CommonSessionInternal.h"
@@ -127,6 +128,14 @@ Http2ServerSession::new_connection(NetVConnection *new_vc, MIOBuffer *iobuf, IOB
   this->_write_buffer_reader   = this->write_buffer->alloc_reader();
   this->_write_size_threshold  = index_to_buffer_size(buffer_block_size_index) * Http2::write_size_threshold;
 
+  uint32_t buffer_water_mark;
+  if (auto snis = this->_vc->get_service<TLSSNISupport>(); snis && snis->hints_from_sni.http2_buffer_water_mark.has_value()) {
+    buffer_water_mark = snis->hints_from_sni.http2_buffer_water_mark.value();
+  } else {
+    buffer_water_mark = Http2::buffer_water_mark;
+  }
+  this->write_buffer->water_mark = buffer_water_mark;
+
   this->_handle_if_ssl(new_vc);
 
   do_api_callout(TS_HTTP_SSN_START_HOOK);
@@ -206,7 +215,7 @@ Http2ServerSession::main_event_handler(int event, void *edata)
     retval = 0;
     break;
 
-  case HTTP2_SESSION_EVENT_XMIT:
+  case HTTP2_SESSION_EVENT_PRIO:
   default:
     Http2SsnDebug("unexpected event=%d edata=%p", event, edata);
     ink_release_assert(0);

tests/Pipfile

@@ -53,5 +53,8 @@ pyyaml ="*"
 grpcio = "*"
 grpcio-tools = "*"
 
+pyOpenSSL = "*"
+eventlet = "*"
+
 [requires]
 python_version = "3"

tests/gold_tests/h2/http2_write_threshold.test.py

@@ -0,0 +1,137 @@
+"""Test proxy.config.http2.write_size_threshold."""
+
+#  Licensed to the Apache Software Foundation (ASF) under one
+#  or more contributor license agreements.  See the NOTICE file
+#  distributed with this work for additional information
+#  regarding copyright ownership.  The ASF licenses this file
+#  to you under the Apache License, Version 2.0 (the
+#  "License"); you may not use this file except in compliance
+#  with the License.  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+import os
+from ports import get_port
+import sys
+
+
+class TestGrpc():
+    """Test proxy.config.http2.write_size_threshold and its associated timeout."""
+
+    def __init__(self, description: str, write_threshold: int, write_timeout: int) -> None:
+        """Configure a TestRun for gRPC traffic.
+
+        :param description: The description for the test runs.
+        """
+        self._description = description
+        tr = Test.AddTestRun(self._description)
+        dns = self._configure_dns(tr)
+        server = self._configure_h2_server(tr, write_timeout)
+        ts = self._configure_traffic_server(tr, dns.Variables.Port, server.Variables.port, write_threshold, write_timeout)
+
+        ts.StartBefore(dns)
+        ts.StartBefore(server)
+        tr.Processes.Default.StartBefore(ts)
+
+        tr.TimeOut = 10
+
+        self._configure_h2_client(tr, ts.Variables.ssl_port, write_timeout)
+
+    def _configure_dns(self, tr: 'TestRun') -> 'Process':
+        """Configure a locally running MicroDNS server.
+
+        :param tr: The TestRun with which to associate the MicroDNS server.
+        :return: The MicroDNS server process.
+        """
+        self._dns = tr.MakeDNServer("dns", default=['127.0.0.1'])
+        return self._dns
+
+    def _configure_h2_server(self, tr: 'TestRun', write_timeout: int) -> 'Process':
+        """Set up the go HTTP/2 server.
+
+        :param tr: The TestRun with which to associate the server.
+        :param write_timeout: The expected maximum amount of time frames should be delivered.
+        :return: The server process.
+        """
+        tr.Setup.Copy('trickle_server.py')
+        self._server = tr.Processes.Process('server')
+
+        server_pem = os.path.join(Test.Variables.AtsTestToolsDir, "ssl", "server.pem")
+        server_key = os.path.join(Test.Variables.AtsTestToolsDir, "ssl", "server.key")
+        self._server.Setup.Copy(server_pem)
+        self._server.Setup.Copy(server_key)
+
+        port = get_port(self._server, 'port')
+        command = (f'{sys.executable} {tr.RunDirectory}/trickle_server.py {port} '
+                   f'server.pem server.key {write_timeout}')
+        self._server.Command = command
+        self._server.ReturnCode = 0
+        self._server.Ready = When.PortOpen(port)
+        return self._server
+
+    def _configure_traffic_server(
+            self, tr: 'TestRun', dns_port: int, server_port: int, write_threshold: int, write_timeout: int) -> 'Process':
+        """Configure the traffic server process.
+
+        :param tr: The TestRun with which to associate the traffic server.
+        :param dns_port: The MicroDNS server port that traffic server should connect to.
+        :param server_port: The server port that traffic server should connect to.
+        :param write_threshold: The value to set for proxy.config.http2.write_size_threshold.
+        :param write_timeout: The value to set for proxy.config.http2.write_time_threshold.
+        :return: The traffic server process.
+        """
+        self._ts = tr.MakeATSProcess("ts", enable_tls=True, enable_cache=False)
+
+        self._ts.addDefaultSSLFiles()
+        self._ts.Disk.ssl_multicert_config.AddLine("dest_ip=* ssl_cert_name=server.pem ssl_key_name=server.key")
+
+        self._ts.Disk.remap_config.AddLine(f"map / https://example.com:{server_port}/")
+
+        self._ts.Disk.records_config.update(
+            {
+                "proxy.config.ssl.server.cert.path": self._ts.Variables.SSLDir,
+                "proxy.config.ssl.server.private_key.path": self._ts.Variables.SSLDir,
+                'proxy.config.ssl.client.alpn_protocols': 'h2,http/1.1',
+                'proxy.config.http.server_session_sharing.pool': 'thread',
+                'proxy.config.ssl.client.verify.server.policy': 'PERMISSIVE',
+                'proxy.config.dns.nameservers': f"127.0.0.1:{dns_port}",
+                'proxy.config.dns.resolv_conf': "NULL",
+                'proxy.config.http2.write_size_threshold': write_threshold,
+                'proxy.config.http2.write_time_threshold': write_timeout,
+
+                # Only enable debug logging during manual exectution. All the
+                # DATA frames get multiple logs and it makes the traffic.out too
+                # unwieldy.
+                "proxy.config.diags.debug.enabled": 0,
+                "proxy.config.diags.debug.tags": "http",
+            })
+        return self._ts
+
+    def _configure_h2_client(self, tr: 'TestRun', proxy_port: int, write_timeout: int) -> None:
+        """Start the HTTP/2 client.
+
+        :param tr: The TestRun with which to associate the client.
+        :param proxy_port: The proxy_port to which to connect.
+        """
+        tr.Setup.Copy('trickle_client.py')
+        ca = os.path.join(Test.Variables.AtsTestToolsDir, "ssl", "server.pem")
+        key = os.path.join(Test.Variables.AtsTestToolsDir, "ssl", "server.key")
+
+        self._server.Setup.Copy(ca)
+        self._server.Setup.Copy(key)
+        # The cert is for example.com, so we must use that domain.
+        hostname = 'example.com'
+        command = (f'{sys.executable} {tr.RunDirectory}/trickle_client.py '
+                   f'{hostname} {proxy_port} server.pem {write_timeout}')
+        p = tr.Processes.Default
+        p.Command = command
+        p.ReturnCode = 0
+
+
+test = TestGrpc("Test proxy.config.http2.write_size_threshold", 0.5, 10)