Fix HTTP flow control policy reload

include/proxy/http2/HTTP2.h

@@ -31,6 +31,8 @@
 
 #include "tsutil/Metrics.h"
 
+#include <atomic>
+
 using ts::Metrics;
 
 class HTTPHdr;
@@ -411,30 +413,30 @@ enum class Http2FlowControlPolicy {
 class Http2
 {
 public:
-  static uint32_t               max_concurrent_streams_in;
-  static uint32_t               min_concurrent_streams_in;
-  static uint32_t               max_active_streams_in;
-  static uint32_t               max_active_streams_policy_in;
-  static bool                   throttling;
-  static uint32_t               stream_priority_enabled;
-  static uint32_t               initial_window_size_in;
-  static Http2FlowControlPolicy flow_control_policy_in;
-  static uint32_t               max_frame_size;
-  static uint32_t               header_table_size;
-  static uint32_t               max_header_list_size;
-  static uint32_t               accept_no_activity_timeout;
-  static uint32_t               no_activity_timeout_in;
-  static uint32_t               active_timeout_in;
-  static uint32_t               incomplete_header_timeout_in;
-  static uint32_t               push_diary_size;
-  static uint32_t               zombie_timeout_in;
-
-  static uint32_t               max_concurrent_streams_out;
-  static uint32_t               min_concurrent_streams_out;
-  static uint32_t               max_active_streams_out;
-  static uint32_t               no_activity_timeout_out;
-  static uint32_t               initial_window_size_out;
-  static Http2FlowControlPolicy flow_control_policy_out;
+  static uint32_t                            max_concurrent_streams_in;
+  static uint32_t                            min_concurrent_streams_in;
+  static uint32_t                            max_active_streams_in;
+  static uint32_t                            max_active_streams_policy_in;
+  static bool                                throttling;
+  static uint32_t                            stream_priority_enabled;
+  static uint32_t                            initial_window_size_in;
+  static std::atomic<Http2FlowControlPolicy> flow_control_policy_in;
+  static uint32_t                            max_frame_size;
+  static uint32_t                            header_table_size;
+  static uint32_t                            max_header_list_size;
+  static uint32_t                            accept_no_activity_timeout;
+  static uint32_t                            no_activity_timeout_in;
+  static uint32_t                            active_timeout_in;
+  static uint32_t                            incomplete_header_timeout_in;
+  static uint32_t                            push_diary_size;
+  static uint32_t                            zombie_timeout_in;
+
+  static uint32_t                            max_concurrent_streams_out;
+  static uint32_t                            min_concurrent_streams_out;
+  static uint32_t                            max_active_streams_out;
+  static uint32_t                            no_activity_timeout_out;
+  static uint32_t                            initial_window_size_out;
+  static std::atomic<Http2FlowControlPolicy> flow_control_policy_out;
 
   static float    stream_error_rate_threshold;
   static uint32_t stream_error_sampling_threshold;

src/proxy/http2/HTTP2.cc

@@ -45,6 +45,28 @@ struct Http2HeaderName {
 
 static VersionConverter hvc;
 
+void
+establish_flow_control_policy(const char *name, std::atomic<Http2FlowControlPolicy> &policy)
+{
+  auto update = [](const char *name, RecDataT type, RecData data, void *cookie) -> int {
+    ink_assert(type == RECD_INT);
+    RecInt value = data.rec_int;
+
+    if (value < 0 || value > 2) {
+      Error("Invalid value for %s: %" PRId64, name, value);
+      value = 0;
+    }
+    static_cast<std::atomic<Http2FlowControlPolicy> *>(cookie)->store(static_cast<Http2FlowControlPolicy>(value),
+                                                                      std::memory_order_relaxed);
+    return REC_ERR_OKAY;
+  };
+  RecData data;
+
+  RecRegisterConfigUpdateCb(name, update, &policy);
+  data.rec_int = RecGetRecordInt(name).value_or(0);
+  update(name, RECD_INT, data, &policy);
+}
+
 } // namespace
 
 // Statistics
@@ -460,17 +482,17 @@ http2_decode_header_blocks(HTTPHdr *hdr, const uint8_t *buf_start, const uint32_
 }
 
 // Initialize this subsystem with librecords configs (for now)
-uint32_t               Http2::max_concurrent_streams_in    = 100;
-uint32_t               Http2::min_concurrent_streams_in    = 10;
-uint32_t               Http2::max_active_streams_in        = 200000;
-uint32_t               Http2::max_active_streams_policy_in = 0;
-bool                   Http2::throttling                   = false;
-uint32_t               Http2::stream_priority_enabled      = 0;
-uint32_t               Http2::initial_window_size_in       = 65535;
-Http2FlowControlPolicy Http2::flow_control_policy_in       = Http2FlowControlPolicy::STATIC_SESSION_AND_STATIC_STREAM;
-uint32_t               Http2::max_frame_size               = 16384;
-uint32_t               Http2::header_table_size            = 4096;
-uint32_t               Http2::max_header_list_size         = 4294967295;
+uint32_t                            Http2::max_concurrent_streams_in    = 100;
+uint32_t                            Http2::min_concurrent_streams_in    = 10;
+uint32_t                            Http2::max_active_streams_in        = 200000;
+uint32_t                            Http2::max_active_streams_policy_in = 0;
+bool                                Http2::throttling                   = false;
+uint32_t                            Http2::stream_priority_enabled      = 0;
+uint32_t                            Http2::initial_window_size_in       = 65535;
+std::atomic<Http2FlowControlPolicy> Http2::flow_control_policy_in{Http2FlowControlPolicy::STATIC_SESSION_AND_STATIC_STREAM};
+uint32_t                            Http2::max_frame_size       = 16384;
+uint32_t                            Http2::header_table_size    = 4096;
+uint32_t                            Http2::max_header_list_size = 4294967295;
 
 uint32_t Http2::accept_no_activity_timeout   = 120;
 uint32_t Http2::no_activity_timeout_in       = 120;
@@ -479,12 +501,12 @@ uint32_t Http2::incomplete_header_timeout_in = 10;
 uint32_t Http2::push_diary_size              = 256;
 uint32_t Http2::zombie_timeout_in            = 0;
 
-uint32_t               Http2::max_concurrent_streams_out = 100;
-uint32_t               Http2::min_concurrent_streams_out = 10;
-uint32_t               Http2::max_active_streams_out     = 0;
-uint32_t               Http2::initial_window_size_out    = 65535;
-Http2FlowControlPolicy Http2::flow_control_policy_out    = Http2FlowControlPolicy::STATIC_SESSION_AND_STATIC_STREAM;
-uint32_t               Http2::no_activity_timeout_out    = 120;
+uint32_t                            Http2::max_concurrent_streams_out = 100;
+uint32_t                            Http2::min_concurrent_streams_out = 10;
+uint32_t                            Http2::max_active_streams_out     = 0;
+uint32_t                            Http2::initial_window_size_out    = 65535;
+std::atomic<Http2FlowControlPolicy> Http2::flow_control_policy_out{Http2FlowControlPolicy::STATIC_SESSION_AND_STATIC_STREAM};
+uint32_t                            Http2::no_activity_timeout_out = 120;
 
 float    Http2::stream_error_rate_threshold        = 0.1;
 uint32_t Http2::stream_error_sampling_threshold    = 10;
@@ -519,22 +541,10 @@ Http2::init()
   RecEstablishStaticConfigUInt32(stream_priority_enabled, "proxy.config.http2.stream_priority_enabled");
 
   RecEstablishStaticConfigUInt32(initial_window_size_in, "proxy.config.http2.initial_window_size_in");
-  uint32_t flow_control_policy_in_int = 0;
-  RecEstablishStaticConfigUInt32(flow_control_policy_in_int, "proxy.config.http2.flow_control.policy_in");
-  if (flow_control_policy_in_int > 2) {
-    Error("Invalid value for proxy.config.http2.flow_control.policy_in: %d", flow_control_policy_in_int);
-    flow_control_policy_in_int = 0;
-  }
-  flow_control_policy_in = static_cast<Http2FlowControlPolicy>(flow_control_policy_in_int);
+  establish_flow_control_policy("proxy.config.http2.flow_control.policy_in", flow_control_policy_in);
 
   RecEstablishStaticConfigUInt32(initial_window_size_out, "proxy.config.http2.initial_window_size_out");
-  uint32_t flow_control_policy_out_int = 0;
-  RecEstablishStaticConfigUInt32(flow_control_policy_out_int, "proxy.config.http2.flow_control.policy_out");
-  if (flow_control_policy_out_int > 2) {
-    Error("Invalid value for proxy.config.http2.flow_control.policy_out: %d", flow_control_policy_out_int);
-    flow_control_policy_out_int = 0;
-  }
-  flow_control_policy_out = static_cast<Http2FlowControlPolicy>(flow_control_policy_out_int);
+  establish_flow_control_policy("proxy.config.http2.flow_control.policy_out", flow_control_policy_out);
 
   RecEstablishStaticConfigUInt32(max_frame_size, "proxy.config.http2.max_frame_size");
   RecEstablishStaticConfigUInt32(header_table_size, "proxy.config.http2.header_table_size");

src/proxy/http2/Http2ConnectionState.cc

@@ -1236,9 +1236,9 @@ Http2ConnectionState::_get_configured_flow_control_policy() const
 {
   ink_assert(this->session != nullptr);
   if (this->session->is_outbound()) {
-    return Http2::flow_control_policy_out;
+    return Http2::flow_control_policy_out.load(std::memory_order_relaxed);
   } else {
-    return Http2::flow_control_policy_in;
+    return Http2::flow_control_policy_in.load(std::memory_order_relaxed);
   }
 }
 

tests/gold_tests/h2/http2_flow_control_reload.py

@@ -0,0 +1,142 @@
+"""Probe HTTP/2 receive windows while reloading flow control policies."""
+
+#  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 argparse
+import concurrent.futures
+import http.client
+import socket
+import ssl
+import subprocess
+import time
+
+from h2.config import H2Configuration
+from h2.connection import H2Connection
+from h2.events import PingAckReceived, RequestReceived
+
+
+class WindowProbe:
+    """Measure ATS receive windows from each side of a fresh connection."""
+
+    def __init__(self, args: argparse.Namespace) -> None:
+        self._args = args
+        self._client_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+        self._client_context.check_hostname = False
+        self._client_context.verify_mode = ssl.CERT_NONE
+        self._client_context.set_alpn_protocols(['h2'])
+        self._server_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+        self._server_context.load_cert_chain(args.cert, args.key)
+        self._server_context.set_alpn_protocols(['h2'])
+
+    def _window(self, sock: ssl.SSLSocket, is_client: bool) -> int:
+        assert sock.selected_alpn_protocol() == 'h2', 'Expected HTTP/2'
+        connection = H2Connection(config=H2Configuration(client_side=is_client))
+        connection.initiate_connection()
+        connection.ping(b'reload!!')
+        sock.sendall(connection.data_to_send())
+        # The PING acknowledgement is a barrier: ATS has sent its initial
+        # SETTINGS and connection WINDOW_UPDATE before acknowledging our PING.
+        acknowledged = False
+        received_request = is_client
+        while not (acknowledged and received_request):
+            data = sock.recv(65536)
+            assert data, 'ATS closed the connection before the probe completed'
+            for event in connection.receive_data(data):
+                if isinstance(event, PingAckReceived):
+                    acknowledged = True
+                elif isinstance(event, RequestReceived):
+                    received_request = True
+                    connection.send_headers(event.stream_id, [(':status', '200'), ('content-length', '0')], end_stream=True)
+            sock.sendall(connection.data_to_send())
+        return connection.outbound_flow_control_window
+
+    def inbound(self) -> int:
+        with socket.create_connection(('127.0.0.1', self._args.https_port), timeout=5) as raw:
+            with self._client_context.wrap_socket(raw, server_hostname='localhost') as sock:
+                return self._window(sock, True)
+
+    def _request(self) -> None:
+        connection = http.client.HTTPConnection('127.0.0.1', self._args.http_port, timeout=5)
+        try:
+            connection.request('GET', '/', headers={'Connection': 'close'})
+            response = connection.getresponse()
+            assert response.status == 200, f'Unexpected origin response: {response.status}'
+            response.read()
+        finally:
+            connection.close()
+
+    def outbound(self) -> int:
+        with socket.socket() as listener:
+            listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+            listener.bind(('127.0.0.1', self._args.origin_port))
+            listener.listen(1)
+            listener.settimeout(5)
+            with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
+                request = executor.submit(self._request)
+                raw, _ = listener.accept()
+                with raw:
+                    raw.settimeout(5)
+                    with self._server_context.wrap_socket(raw, server_side=True) as sock:
+                        window = self._window(sock, False)
+                        request.result(timeout=5)
+                return window
+
+    def expect(self, inbound_policy: int, outbound_policy: int, wait: bool = False) -> None:
+        expected = tuple(65535 if policy == 0 else 6553500 for policy in (inbound_policy, outbound_policy))
+        deadline = time.monotonic() + (20 if wait else 0)
+        while True:
+            actual = (self.inbound(), self.outbound())
+            if actual == expected:
+                print(f'policies in={inbound_policy}, out={outbound_policy}: windows={actual}', flush=True)
+                return
+            if time.monotonic() >= deadline:
+                raise AssertionError(
+                    f'policies in={inbound_policy}, out={outbound_policy}: expected windows {expected}, got {actual}')
+            time.sleep(0.25)
+
+    def run(self) -> None:
+        self.expect(0, 0)
+        # Reset to policy 0 before policy 2 so the large window proves that
+        # both nonzero policy values actually take effect. Probe the opposite
+        # direction too, so crossed or shared callbacks cannot pass.
+        for direction in ('in', 'out'):
+            record = f'proxy.config.http2.flow_control.policy_{direction}'
+            for policy in (1, 0, 2, 0):
+                subprocess.run(['traffic_ctl', 'config', 'set', record, str(policy)], check=True, timeout=10)
+                self.expect(policy if direction == 'in' else 0, policy if direction == 'out' else 0, wait=True)
+        # Exercise the records.yaml reload path as well as live config set.
+        for policy in (1, 0):
+            for direction in ('in', 'out'):
+                record = f'proxy.config.http2.flow_control.policy_{direction}'
+                subprocess.run(['traffic_ctl', 'config', 'set', record, str(policy), '--cold'], check=True, timeout=10)
+            subprocess.run(['traffic_ctl', 'config', 'reload', '--monitor'], check=True, timeout=30)
+            self.expect(policy, policy, wait=True)
+        print('PASS: both flow control policies reloaded', flush=True)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--https-port', type=int, required=True)
+    parser.add_argument('--http-port', type=int, required=True)
+    parser.add_argument('--origin-port', type=int, required=True)
+    parser.add_argument('--cert', required=True)
+    parser.add_argument('--key', required=True)
+    WindowProbe(parser.parse_args()).run()
+
+
+if __name__ == '__main__':
+    main()

tests/gold_tests/h2/http2_flow_control_reload.test.py

@@ -0,0 +1,70 @@
+"""Verify HTTP/2 flow control policies change without restarting ATS."""
+
+#  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 sys
+from ports import get_port
+
+Test.Summary = __doc__
+
+
+class FlowControlReloadTest:
+    """Observe both connection receive windows across runtime policy changes."""
+
+    def __init__(self) -> None:
+        self._ts = self._configure_trafficserver()
+        self._configure_probe()
+
+    def _configure_trafficserver(self) -> 'Process':
+        ts = Test.MakeATSProcess('ts', enable_tls=True, enable_cache=False)
+        ts.addDefaultSSLFiles()
+        get_port(ts, 'origin_port')
+        ts.Disk.records_config.update(
+            {
+                'proxy.config.ssl.server.cert.path': ts.Variables.SSLDir,
+                'proxy.config.ssl.server.private_key.path': ts.Variables.SSLDir,
+                'proxy.config.ssl.client.verify.server.policy': 'PERMISSIVE',
+                'proxy.config.ssl.client.alpn_protocols': 'h2',
+                'proxy.config.http2.initial_window_size_in': 65535,
+                'proxy.config.http2.initial_window_size_out': 65535,
+                'proxy.config.http2.max_concurrent_streams_in': 100,
+                'proxy.config.http2.max_concurrent_streams_out': 100,
+                'proxy.config.http2.flow_control.policy_in': 0,
+                'proxy.config.http2.flow_control.policy_out': 0,
+            })
+        ts.Disk.ssl_multicert_yaml.AddLines(
+            ['ssl_multicert:', '  - dest_ip: "*"', '    ssl_cert_name: server.pem', '    ssl_key_name: server.key'])
+        ts.Disk.remap_config.AddLine(f'map / https://127.0.0.1:{ts.Variables.origin_port}')
+        return ts
+
+    def _configure_probe(self) -> None:
+        tr = Test.AddTestRun('Reload inbound and outbound flow control policies on the same ATS process')
+        tr.Processes.Default.StartBefore(self._ts)
+        tr.Processes.Default.Env = self._ts.Env
+        tr.Processes.Default.Command = (
+            f'{sys.executable} {Test.TestDirectory}/http2_flow_control_reload.py '
+            f'--https-port {self._ts.Variables.ssl_port} --http-port {self._ts.Variables.port} '
+            f'--origin-port {self._ts.Variables.origin_port} '
+            f'--cert {self._ts.Variables.SSLDir}/server.pem --key {self._ts.Variables.SSLDir}/server.key')
+        tr.Processes.Default.ReturnCode = 0
+        tr.Processes.Default.TimeOut = 180
+        tr.Processes.Default.Streams.stdout = Testers.ContainsExpression(
+            'PASS: both flow control policies reloaded', 'Both policies must affect new connections without a restart')
+        tr.StillRunningAfter = self._ts
+
+
+FlowControlReloadTest()