diff --git a/include/proxy/http2/HTTP2.h b/include/proxy/http2/HTTP2.h index 370bbbef929..3612aab2692 100644 --- a/include/proxy/http2/HTTP2.h +++ b/include/proxy/http2/HTTP2.h @@ -31,6 +31,8 @@ #include "tsutil/Metrics.h" +#include + 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 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 flow_control_policy_out; static float stream_error_rate_threshold; static uint32_t stream_error_sampling_threshold; diff --git a/src/proxy/http2/HTTP2.cc b/src/proxy/http2/HTTP2.cc index e0e0d62d099..aee44d2d9a9 100644 --- a/src/proxy/http2/HTTP2.cc +++ b/src/proxy/http2/HTTP2.cc @@ -45,6 +45,28 @@ struct Http2HeaderName { static VersionConverter hvc; +void +establish_flow_control_policy(const char *name, std::atomic &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 *>(cookie)->store(static_cast(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 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 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(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(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"); diff --git a/src/proxy/http2/Http2ConnectionState.cc b/src/proxy/http2/Http2ConnectionState.cc index 446dfc4dbdd..5eccce014b6 100644 --- a/src/proxy/http2/Http2ConnectionState.cc +++ b/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); } } diff --git a/tests/gold_tests/h2/http2_flow_control_reload.py b/tests/gold_tests/h2/http2_flow_control_reload.py new file mode 100644 index 00000000000..69ed9f58454 --- /dev/null +++ b/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() diff --git a/tests/gold_tests/h2/http2_flow_control_reload.test.py b/tests/gold_tests/h2/http2_flow_control_reload.test.py new file mode 100644 index 00000000000..7f6f19f4b72 --- /dev/null +++ b/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()