diff --git a/include/iocore/net/quic/QUICStream.h b/include/iocore/net/quic/QUICStream.h index e287daa5c90..e22f4d92550 100644 --- a/include/iocore/net/quic/QUICStream.h +++ b/include/iocore/net/quic/QUICStream.h @@ -59,6 +59,13 @@ class QUICStream public: using ErrorCode = uint64_t; //!< recv/send stream application error codes. + // Guaranteed per-stream send budget for one write event when many streams are + // contending for the connection's write path this round. + static constexpr size_t MIN_STREAM_SEND_BYTES_PER_EVENT = 16 * 1024; + // Total send budget for one connection's write event, divided across its writable + // streams by compute_fair_send_budget(). A lone writable stream gets it all. + static constexpr size_t MAX_CONNECTION_SEND_BYTES_PER_EVENT = 256 * 1024; + QUICStream() {} QUICStream(QUICConnectionInfoProvider *cinfo, QUICStreamId sid); virtual ~QUICStream(); @@ -76,7 +83,13 @@ class QUICStream void reset(QUICStreamErrorUPtr error); void receive_data(QUICStreamIO &stream_io); - int64_t send_data(QUICStreamIO &stream_io); + int64_t send_data(QUICStreamIO &stream_io, size_t max_bytes_this_event); + + // Computes the per-stream send budget for one write event given how many streams + // are writable this event on this connection. Scales down toward + // MIN_STREAM_SEND_BYTES_PER_EVENT under contention, up toward + // MAX_CONNECTION_SEND_BYTES_PER_EVENT when a stream has the write path to itself. + static size_t compute_fair_send_budget(size_t num_writable_streams); /* * QUICApplication need to call one of these functions when it process VC_EVENT_* diff --git a/include/proxy/http3/Http3Frame.h b/include/proxy/http3/Http3Frame.h index a0255496385..8c8705729ee 100644 --- a/include/proxy/http3/Http3Frame.h +++ b/include/proxy/http3/Http3Frame.h @@ -63,19 +63,6 @@ class Http3Frame bool _is_ready = false; }; -class Http3UnknownFrame : public Http3Frame -{ -public: - Http3UnknownFrame() : Http3Frame() {} - Http3UnknownFrame(IOBufferReader &reader); - - Ptr to_io_buffer_block() const override; - -protected: - const uint8_t *_buf = nullptr; - size_t _buf_len = 0; -}; - // // DATA Frame // @@ -112,7 +99,11 @@ class Http3HeadersFrame : public Http3Frame public: Http3HeadersFrame() : Http3Frame() {} Http3HeadersFrame(IOBufferReader &reader); - Http3HeadersFrame(ats_unique_buf header_block, size_t header_block_len); + // Shares the caller's buffer via a cloned reader instead of copying header_block_len bytes. + // Safe as long as the source MIOBuffer outlives this frame, which holds for the qmux/quic + // write path: the frame is created, serialized via to_io_buffer_block(), and destroyed, all + // synchronously, well within the lifetime of the Http3HeaderFramer that owns the source buffer. + Http3HeadersFrame(IOBufferReader &header_block_reader, size_t header_block_len); ~Http3HeadersFrame(); Ptr to_io_buffer_block() const override; @@ -125,9 +116,9 @@ class Http3HeadersFrame : public Http3Frame bool _parse() override; private: - uint8_t *_header_block = nullptr; - ats_unique_buf _header_block_uptr = {nullptr}; - size_t _header_block_len = 0; + uint8_t *_header_block = nullptr; + size_t _header_block_len = 0; + IOBufferReader *_header_block_reader = nullptr; }; // @@ -247,7 +238,6 @@ class Http3FrameFactory /* * Creates a HEADERS frame. */ - static Http3HeadersFrameUPtr create_headers_frame(const uint8_t *header_block, size_t header_block_len); static Http3HeadersFrameUPtr create_headers_frame(IOBufferReader *header_block_reader, size_t header_block_len); /* diff --git a/include/proxy/http3/Http3FrameCollector.h b/include/proxy/http3/Http3FrameCollector.h index 0845cbc57c5..1e1e7841927 100644 --- a/include/proxy/http3/Http3FrameCollector.h +++ b/include/proxy/http3/Http3FrameCollector.h @@ -33,6 +33,10 @@ class QUICStreamVCAdapter; class Http3FrameCollector { public: + // Http3Transaction always adds exactly 2 generators (header framer, data framer) per + // transaction; reserving avoids the growth-triggered reallocation on the second add_generator(). + Http3FrameCollector() { _generators.reserve(2); } + Http3ErrorUPtr on_write_ready(QUICStreamId stream_id, MIOBuffer &writer, size_t &nread, bool &all_done); void add_generator(Http3FrameGenerator *generator); diff --git a/include/proxy/http3/Http3FrameCounter.h b/include/proxy/http3/Http3FrameCounter.h index ead94eb28f9..83d0270a20f 100644 --- a/include/proxy/http3/Http3FrameCounter.h +++ b/include/proxy/http3/Http3FrameCounter.h @@ -32,7 +32,7 @@ class Http3FrameCounter : public Http3FrameHandler Http3FrameCounter(){}; // Http3FrameHandler - std::vector interests() override; + std::vector const &interests() override; Http3ErrorUPtr handle_frame(std::shared_ptr frame, Http3StreamType s_type = Http3StreamType::UNKNOWN) override; uint64_t get_count(uint64_t type) const; diff --git a/include/proxy/http3/Http3FrameDispatcher.h b/include/proxy/http3/Http3FrameDispatcher.h index d904956c1d1..9b0a229b472 100644 --- a/include/proxy/http3/Http3FrameDispatcher.h +++ b/include/proxy/http3/Http3FrameDispatcher.h @@ -26,7 +26,8 @@ #include "iocore/net/quic/QUICApplication.h" #include "proxy/http3/Http3Frame.h" #include "proxy/http3/Http3FrameHandler.h" -#include +#include +#include class QUICStreamVCAdapter; @@ -38,17 +39,25 @@ class Http3FrameDispatcher void add_handler(Http3FrameHandler *handler); private: + // At most a handful of handlers ever register interest in the same frame type (currently + // up to 3: the frame counter, the protocol enforcer, and one of the header/data handlers). + // Inline storage avoids a heap allocation per handler registration, which otherwise runs + // once per HTTP/3 request since this dispatcher is a per-transaction object. + static constexpr size_t MAX_HANDLERS_PER_TYPE = 4; + enum READING_STATE { READING_TYPE_LEN, READING_LENGTH_LEN, READING_PAYLOAD_LEN, READING_PAYLOAD, } _reading_state = READING_TYPE_LEN; - int64_t _reading_frame_type_len; - int64_t _reading_frame_length_len; - uint64_t _reading_frame_payload_len; - uint64_t _bytes_to_skip; - Http3FrameFactory _frame_factory; - std::shared_ptr _current_frame = nullptr; - std::vector _handlers[256]; + int64_t _reading_frame_type_len; + int64_t _reading_frame_length_len; + uint64_t _reading_frame_payload_len; + uint64_t _bytes_to_skip; + Http3FrameFactory _frame_factory; + std::shared_ptr _current_frame = nullptr; + + std::array _handlers[256] = {}; + uint8_t _handler_count[256] = {}; }; diff --git a/include/proxy/http3/Http3FrameHandler.h b/include/proxy/http3/Http3FrameHandler.h index 2a8bb3ff4b1..d73cf33a3e4 100644 --- a/include/proxy/http3/Http3FrameHandler.h +++ b/include/proxy/http3/Http3FrameHandler.h @@ -31,7 +31,7 @@ class Http3FrameHandler { public: virtual ~Http3FrameHandler(){}; - virtual std::vector interests() = 0; - virtual Http3ErrorUPtr handle_frame(std::shared_ptr frame, - Http3StreamType s_type = Http3StreamType::UNKNOWN) = 0; + virtual std::vector const &interests() = 0; + virtual Http3ErrorUPtr handle_frame(std::shared_ptr frame, + Http3StreamType s_type = Http3StreamType::UNKNOWN) = 0; }; diff --git a/include/proxy/http3/Http3HeaderVIOAdaptor.h b/include/proxy/http3/Http3HeaderVIOAdaptor.h index 13b30f2314d..23be0c852ec 100644 --- a/include/proxy/http3/Http3HeaderVIOAdaptor.h +++ b/include/proxy/http3/Http3HeaderVIOAdaptor.h @@ -35,7 +35,7 @@ class Http3HeaderVIOAdaptor : public Continuation, public Http3FrameHandler ~Http3HeaderVIOAdaptor(); // Http3FrameHandler - std::vector interests() override; + std::vector const &interests() override; Http3ErrorUPtr handle_frame(std::shared_ptr frame, Http3StreamType s_type = Http3StreamType::UNKNOWN) override; bool is_complete(); diff --git a/include/proxy/http3/Http3ProtocolEnforcer.h b/include/proxy/http3/Http3ProtocolEnforcer.h index e9801f6802b..2170ea27778 100644 --- a/include/proxy/http3/Http3ProtocolEnforcer.h +++ b/include/proxy/http3/Http3ProtocolEnforcer.h @@ -32,7 +32,7 @@ class Http3ProtocolEnforcer : public Http3FrameHandler Http3ProtocolEnforcer(){}; // Http3FrameHandler - std::vector interests() override; + std::vector const &interests() override; Http3ErrorUPtr handle_frame(std::shared_ptr frame, Http3StreamType s_type = Http3StreamType::UNKNOWN) override; private: diff --git a/include/proxy/http3/Http3SettingsHandler.h b/include/proxy/http3/Http3SettingsHandler.h index 51545ee07f6..8469b9d9eb8 100644 --- a/include/proxy/http3/Http3SettingsHandler.h +++ b/include/proxy/http3/Http3SettingsHandler.h @@ -32,7 +32,7 @@ class Http3SettingsHandler : public Http3FrameHandler Http3SettingsHandler(Http3Session *session) : _session(session){}; // Http3FrameHandler - std::vector interests() override; + std::vector const &interests() override; Http3ErrorUPtr handle_frame(std::shared_ptr frame, Http3StreamType s_type = Http3StreamType::UNKNOWN) override; private: diff --git a/include/proxy/http3/Http3StreamDataVIOAdaptor.h b/include/proxy/http3/Http3StreamDataVIOAdaptor.h index 8968ab65cc2..ba1e17e41b5 100644 --- a/include/proxy/http3/Http3StreamDataVIOAdaptor.h +++ b/include/proxy/http3/Http3StreamDataVIOAdaptor.h @@ -34,12 +34,12 @@ class Http3StreamDataVIOAdaptor : public Http3FrameHandler virtual ~Http3StreamDataVIOAdaptor(); // Http3FrameHandler - std::vector interests() override; + std::vector const &interests() override; Http3ErrorUPtr handle_frame(std::shared_ptr frame, Http3StreamType s_type = Http3StreamType::UNKNOWN) override; // Http3StreamDataVIOAdaptor void finalize(); - bool has_data(); + bool has_data() const; private: VIO *_sink_vio = nullptr; diff --git a/include/proxy/http3/Http3Transaction.h b/include/proxy/http3/Http3Transaction.h index 5adb9bda173..abb7dfd6359 100644 --- a/include/proxy/http3/Http3Transaction.h +++ b/include/proxy/http3/Http3Transaction.h @@ -28,6 +28,11 @@ #include "iocore/net/quic/QUICStreamVCAdapter.h" #include "proxy/http3/Http3FrameDispatcher.h" #include "proxy/http3/Http3FrameCollector.h" +#include "proxy/http3/Http3HeaderFramer.h" +#include "proxy/http3/Http3DataFramer.h" +#include "proxy/http3/Http3ProtocolEnforcer.h" +#include "proxy/http3/Http3HeaderVIOAdaptor.h" +#include "proxy/http3/Http3StreamDataVIOAdaptor.h" #include @@ -35,11 +40,6 @@ class QUICStreamIO; class HQSession; class Http09Session; class Http3Session; -class Http3HeaderFramer; -class Http3DataFramer; -class Http3HeaderVIOAdaptor; -class Http3ProtocolEnforcer; -class Http3StreamDataVIOAdaptor; class HQTransaction : public ProxyTransaction { @@ -151,13 +151,13 @@ class Http3Transaction : public HQTransaction void _handle_error(const Http3Error &error); // These are for HTTP/3 - Http3FrameDispatcher _frame_dispatcher; - Http3FrameCollector _frame_collector; - Http3ProtocolEnforcer *_protocol_enforcer = nullptr; - Http3HeaderFramer *_header_framer = nullptr; - Http3DataFramer *_data_framer = nullptr; - Http3HeaderVIOAdaptor *_header_handler = nullptr; - Http3StreamDataVIOAdaptor *_data_handler = nullptr; + Http3FrameDispatcher _frame_dispatcher; + Http3FrameCollector _frame_collector; + Http3ProtocolEnforcer _protocol_enforcer; + Http3HeaderFramer _header_framer; + Http3DataFramer _data_framer; + Http3HeaderVIOAdaptor _header_handler; + Http3StreamDataVIOAdaptor _data_handler; }; /** diff --git a/src/iocore/net/CMakeLists.txt b/src/iocore/net/CMakeLists.txt index b317e26b3ea..8e15101f9af 100644 --- a/src/iocore/net/CMakeLists.txt +++ b/src/iocore/net/CMakeLists.txt @@ -159,6 +159,9 @@ if(BUILD_TESTING) if(TS_USE_QUIC) target_sources(test_net PRIVATE unit_tests/test_QUICTokenKeyConfig.cc) endif() + if(TS_USE_QUIC OR TS_USE_QMUX) + target_sources(test_net PRIVATE unit_tests/test_QUICStream.cc) + endif() # Use link groups to solve circular dependency set(LINK_GROUP_LIBS ts::logging diff --git a/src/iocore/net/OpenSSLQUICNetVConnection.cc b/src/iocore/net/OpenSSLQUICNetVConnection.cc index 9a06fb94d26..0218b468315 100644 --- a/src/iocore/net/OpenSSLQUICNetVConnection.cc +++ b/src/iocore/net/OpenSSLQUICNetVConnection.cc @@ -749,9 +749,9 @@ QUICNetVConnection::_process_openssl_streams() } if ((stream_type & SSL_STREAM_TYPE_WRITE) != 0 || stream->has_data_to_send()) { if (stream->has_data_to_send()) { - while (stream->has_data_to_send() && stream->send_data(*this) > 0) {} + while (stream->has_data_to_send() && stream->send_data(*this, QUICStream::MIN_STREAM_SEND_BYTES_PER_EVENT) > 0) {} } else { - stream->send_data(*this); + stream->send_data(*this, QUICStream::MIN_STREAM_SEND_BYTES_PER_EVENT); } } } diff --git a/src/iocore/net/QUICNetProcessor.cc b/src/iocore/net/QUICNetProcessor.cc index d7f7012606a..5dd56847c88 100644 --- a/src/iocore/net/QUICNetProcessor.cc +++ b/src/iocore/net/QUICNetProcessor.cc @@ -84,7 +84,11 @@ QUICNetProcessor::start(int, size_t /* stacksize ATS_UNUSED */) QUICCertConfig::startup(); QUICConfig::scoped_config params; - if (dbg_ctl_vv_quiche.tag_on()) { + // tag_on() only checks the tag pattern, not whether debug output is globally + // enabled -- use on() so a tag that happens to match "vv_quiche" as a substring + // (e.g. "vv_quic") doesn't permanently install quiche's trace-level Rust logger + // regardless of proxy.config.diags.debug.enabled. + if (dbg_ctl_vv_quiche.on()) { quiche_enable_debug_logging(debug_log, NULL); } this->_quiche_config = quiche_config_new(QUICHE_PROTOCOL_VERSION); diff --git a/src/iocore/net/QUICNetVConnection.cc b/src/iocore/net/QUICNetVConnection.cc index 451b1fc1243..c08712d8242 100644 --- a/src/iocore/net/QUICNetVConnection.cc +++ b/src/iocore/net/QUICNetVConnection.cc @@ -688,6 +688,23 @@ void QUICNetVConnection::_handle_write_ready() { if (quiche_conn_is_established(this->_quiche_con)) { + // Count real contention for THIS event before deciding its budget, rather than + // sizing it from a previous event's count -- a stale count can be wrong in either + // direction whenever contention swings between events, not just on the first + // event. writable() is a pure, side-effect-free snapshot (verified against + // quiche's source), so draining it twice costs one extra O(n) collect and n extra + // FFI calls, n bounded by this connection's stream limit -- cheap next to the + // per-stream work that follows. + quiche_stream_iter *probe = quiche_conn_writable(this->_quiche_con); + uint64_t probe_id = 0; + size_t writable_count = 0; + while (quiche_stream_iter_next(probe, &probe_id)) { + ++writable_count; + } + quiche_stream_iter_free(probe); + + const size_t budget = QUICStream::compute_fair_send_budget(writable_count); + quiche_stream_iter *writable = quiche_conn_writable(this->_quiche_con); uint64_t s = 0; while (quiche_stream_iter_next(writable, &s)) { @@ -696,7 +713,7 @@ QUICNetVConnection::_handle_write_ready() [[maybe_unused]] QUICConnectionError err; stream = this->_stream_manager->create_stream(s, err); } - stream->send_data(*this); + stream->send_data(*this, budget); } quiche_stream_iter_free(writable); } diff --git a/src/iocore/net/qmux/QMuxConnection.cc b/src/iocore/net/qmux/QMuxConnection.cc index 3b73b16c5bc..588c7a3789e 100644 --- a/src/iocore/net/qmux/QMuxConnection.cc +++ b/src/iocore/net/qmux/QMuxConnection.cc @@ -355,13 +355,30 @@ QMuxConnection::_handle_write_streams() return; } + // Count real contention for THIS event before deciding its budget, rather than + // sizing it from a previous event's count -- a stale count can be wrong in either + // direction whenever contention swings between events, not just on the first + // event. writable() is a pure, side-effect-free snapshot (verified against quiche's + // source), so draining it twice costs one extra O(n) collect and n extra FFI calls, + // n bounded by this connection's stream limit -- cheap next to the per-stream work + // that follows. + quiche_stream_iter *probe = quiche_conn_writable(_quiche_con); + uint64_t probe_id; + size_t writable_count = 0; + while (quiche_stream_iter_next(probe, &probe_id)) { + ++writable_count; + } + quiche_stream_iter_free(probe); + + const size_t budget = QUICStream::compute_fair_send_budget(writable_count); + quiche_stream_iter *writable = quiche_conn_writable(_quiche_con); uint64_t stream_id; while (quiche_stream_iter_next(writable, &stream_id)) { QUICStream *stream = _stream_manager->find_stream(stream_id); if (stream != nullptr) { - stream->send_data(*this); + stream->send_data(*this, budget); } } quiche_stream_iter_free(writable); diff --git a/src/iocore/net/quic/QUICStream.cc b/src/iocore/net/quic/QUICStream.cc index 9f8de91fa0a..e035516dc48 100644 --- a/src/iocore/net/quic/QUICStream.cc +++ b/src/iocore/net/quic/QUICStream.cc @@ -24,13 +24,24 @@ #include "iocore/net/quic/QUICStream.h" #include "iocore/net/quic/QUICStreamAdapter.h" -constexpr uint32_t MAX_STREAM_FRAME_OVERHEAD = 24; -constexpr size_t MAX_STREAM_SEND_BYTES_PER_EVENT = 16 * 1024; +#include + +constexpr uint32_t MAX_STREAM_FRAME_OVERHEAD = 24; QUICStream::QUICStream(QUICConnectionInfoProvider *cinfo, QUICStreamId sid) : _connection_info(cinfo), _id(sid) {} QUICStream::~QUICStream() {} +size_t +QUICStream::compute_fair_send_budget(size_t num_writable_streams) +{ + if (num_writable_streams <= 1) { + return MAX_CONNECTION_SEND_BYTES_PER_EVENT; + } + return std::clamp(MAX_CONNECTION_SEND_BYTES_PER_EVENT / num_writable_streams, MIN_STREAM_SEND_BYTES_PER_EVENT, + MAX_CONNECTION_SEND_BYTES_PER_EVENT); +} + QUICStreamId QUICStream::id() const { @@ -144,24 +155,28 @@ QUICStream::receive_data(QUICStreamIO &stream_io) } int64_t -QUICStream::send_data(QUICStreamIO &stream_io) +QUICStream::send_data(QUICStreamIO &stream_io, size_t max_bytes_this_event) { bool fin = false; ssize_t len = 0; [[maybe_unused]] ErrorCode error_code{0}; size_t written_this_event = 0; + // _write_vio.nbytes is set once when the VIO is armed and doesn't change over the + // course of this call, so query it once instead of re-locking the adapter's mutex + // for the same value on every loop iteration below. + const uint64_t total_len = this->_adapter->total_len(); - while (written_this_event < MAX_STREAM_SEND_BYTES_PER_EVENT) { + while (written_this_event < max_bytes_this_event) { len = stream_io.stream_write_capacity(this->_id); if (len <= 0) { return written_this_event; } if (!this->_pending_send_block) { - size_t read_len = std::min(static_cast(len), MAX_STREAM_SEND_BYTES_PER_EVENT - written_this_event); + size_t read_len = std::min(static_cast(len), max_bytes_this_event - written_this_event); this->_pending_send_block = this->_adapter->read(read_len); if (!this->_pending_send_block) { - if (!this->_sent_fin && this->_adapter->is_eos() && this->_adapter->total_len() == this->_sent_bytes) { + if (!this->_sent_fin && this->_adapter->is_eos() && total_len == this->_sent_bytes) { static constexpr uint8_t empty_data = 0; ssize_t written_len = stream_io.write_stream(this->_id, &empty_data, 0, true, error_code); if (written_len >= 0) { @@ -172,11 +187,17 @@ QUICStream::send_data(QUICStreamIO &stream_io) this->_adapter->encourge_write(); return written_this_event; } - this->_pending_send_fin = this->_adapter->total_len() == this->_sent_bytes + this->_pending_send_block->size(); + this->_pending_send_fin = total_len == this->_sent_bytes + this->_pending_send_block->size(); } Ptr block = this->_pending_send_block; - fin = this->_pending_send_fin; + // A pending block may have been carried over from a previous event, sized under + // that event's (possibly larger) budget. Cap what we submit here to what's left + // of this event's budget, and only claim fin if the whole remaining block is + // included in this submission. + size_t remaining_budget = max_bytes_this_event - written_this_event; + size_t to_write = std::min(static_cast(block->size()), remaining_budget); + fin = this->_pending_send_fin && to_write == static_cast(block->size()); if (block->size() == 0 && !fin) { this->_pending_send_block = nullptr; this->_pending_send_fin = false; @@ -184,19 +205,19 @@ QUICStream::send_data(QUICStreamIO &stream_io) continue; } - if (block->size() > 0 || fin) { + if (to_write > 0 || fin) { ssize_t written_len = - stream_io.write_stream(this->_id, reinterpret_cast(block->start()), block->size(), fin, error_code); + stream_io.write_stream(this->_id, reinterpret_cast(block->start()), to_write, fin, error_code); if (written_len >= 0) { this->_adapter->consume(written_len); this->_sent_bytes += written_len; written_this_event += static_cast(written_len); - if (written_len >= block->size()) { + block->consume(written_len); + if (block->size() == 0) { this->_pending_send_block = nullptr; this->_pending_send_fin = false; this->_sent_fin = fin; } else { - block->consume(written_len); return written_this_event; } if (!this->has_data_to_send()) { diff --git a/src/iocore/net/unit_tests/test_QUICStream.cc b/src/iocore/net/unit_tests/test_QUICStream.cc new file mode 100644 index 00000000000..455db8b1ede --- /dev/null +++ b/src/iocore/net/unit_tests/test_QUICStream.cc @@ -0,0 +1,193 @@ +/** @file + + Catch based unit tests for QUICStream + + @section license License + + 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. + */ + +#include "iocore/net/quic/QUICStream.h" +#include "iocore/net/quic/QUICStreamAdapter.h" + +#include +#include +#include + +namespace +{ + +// Hands out a single fixed-size block of data as one contiguous run, tracking how much +// of it remains -- just enough surface for QUICStream::send_data() to exercise its +// pending-block accounting across multiple calls. +class BudgetTestAdapter : public QUICStreamAdapter +{ +public: + BudgetTestAdapter(QUICStream &stream, size_t total_len) : QUICStreamAdapter(stream), _total_len(total_len), _remaining(total_len) + { + } + + int64_t + write(QUICOffset, const uint8_t *, uint64_t, bool) override + { + return 0; + } + bool + is_eos() override + { + return true; + } + uint64_t + unread_len() override + { + return _remaining; + } + uint64_t + read_len() override + { + return 0; + } + uint64_t + total_len() override + { + return _total_len; + } + void + encourge_read() override + { + } + void + encourge_write() override + { + } + void + notify_eos() override + { + } + +protected: + Ptr + _read(size_t len) override + { + len = std::min(len, _remaining); + Ptr block = make_ptr(new_IOBufferBlock()); + block->alloc(iobuffer_size_to_index(std::max(len, 1), BUFFER_SIZE_INDEX_128)); + block->fill(len); + return block; + } + + void + _consume(size_t len) override + { + _remaining -= std::min(len, _remaining); + } + +private: + size_t _total_len; + size_t _remaining; +}; + +// Records what QUICStream::send_data() actually submits to the wire, and lets a test +// simulate connection-level flow control by capping how much of a write is "accepted". +class BudgetTestStreamIO : public QUICStreamIO +{ +public: + int64_t + read_stream(QUICStreamId, uint8_t *, size_t, bool &, ErrorCode &) override + { + return 0; + } + bool + stream_read_finished(QUICStreamId) override + { + return false; + } + int64_t + stream_write_capacity(QUICStreamId) override + { + return write_capacity; + } + int64_t + write_stream(QUICStreamId, uint8_t const *, size_t len, bool fin, ErrorCode &) override + { + last_requested_len = len; + last_fin = fin; + size_t accepted = std::min(len, accept_up_to); + return static_cast(accepted); + } + + int64_t write_capacity = std::numeric_limits::max(); + size_t accept_up_to = std::numeric_limits::max(); + size_t last_requested_len = 0; + bool last_fin = false; +}; + +} // namespace + +TEST_CASE("QUICStream::send_data caps a carried-over pending block to the current event's budget") +{ + QUICStream stream(nullptr, 0); + BudgetTestAdapter adapter(stream, 200 * 1024); + BudgetTestStreamIO io; + + stream.set_io_adapter(&adapter); + + // Event 1: generous budget, but the connection only accepts part of the write -- + // leaves a pending block that was sized under this (large) budget. + io.accept_up_to = 100 * 1024; + int64_t written1 = stream.send_data(io, QUICStream::MAX_CONNECTION_SEND_BYTES_PER_EVENT); + REQUIRE(written1 == 100 * 1024); + + // Event 2: contention spikes and the budget drops to the floor. The connection no + // longer constrains writes, so if send_data() submitted the whole carried-over + // pending block (100KB) instead of capping to the new budget, this event would + // blow past MIN_STREAM_SEND_BYTES_PER_EVENT. + io.accept_up_to = std::numeric_limits::max(); + int64_t written2 = stream.send_data(io, QUICStream::MIN_STREAM_SEND_BYTES_PER_EVENT); + + CHECK(written2 == static_cast(QUICStream::MIN_STREAM_SEND_BYTES_PER_EVENT)); + CHECK(io.last_requested_len == QUICStream::MIN_STREAM_SEND_BYTES_PER_EVENT); + CHECK(io.last_fin == false); +} + +TEST_CASE("QUICStream::compute_fair_send_budget") +{ + SECTION("No contention (0 or 1 writable streams) returns the max budget") + { + CHECK(QUICStream::compute_fair_send_budget(0) == QUICStream::MAX_CONNECTION_SEND_BYTES_PER_EVENT); + CHECK(QUICStream::compute_fair_send_budget(1) == QUICStream::MAX_CONNECTION_SEND_BYTES_PER_EVENT); + } + + SECTION("Heavy contention clamps to the min budget") + { + CHECK(QUICStream::compute_fair_send_budget(100) == QUICStream::MIN_STREAM_SEND_BYTES_PER_EVENT); + } + + SECTION("Mid-range contention divides the max budget evenly") + { + CHECK(QUICStream::compute_fair_send_budget(8) == QUICStream::MAX_CONNECTION_SEND_BYTES_PER_EVENT / 8); + } + + SECTION("Floor-transition boundary") + { + // MAX / MIN is the exact stream count at which the division result equals the floor. + const size_t boundary = QUICStream::MAX_CONNECTION_SEND_BYTES_PER_EVENT / QUICStream::MIN_STREAM_SEND_BYTES_PER_EVENT; + + CHECK(QUICStream::compute_fair_send_budget(boundary) == QUICStream::MIN_STREAM_SEND_BYTES_PER_EVENT); + CHECK(QUICStream::compute_fair_send_budget(boundary + 1) == QUICStream::MIN_STREAM_SEND_BYTES_PER_EVENT); + } +} diff --git a/src/proxy/http3/Http3Frame.cc b/src/proxy/http3/Http3Frame.cc index 2c15ee03ef3..86e94d517ae 100644 --- a/src/proxy/http3/Http3Frame.cc +++ b/src/proxy/http3/Http3Frame.cc @@ -193,27 +193,6 @@ Http3Frame::reset(IOBufferReader &reader) new (this) Http3Frame(reader); } -// -// UNKNOWN Frame -// -Http3UnknownFrame::Http3UnknownFrame(IOBufferReader &reader) : Http3Frame(reader) {} - -Ptr -Http3UnknownFrame::to_io_buffer_block() const -{ - Ptr block; - size_t n = 0; - - block = make_ptr(new_IOBufferBlock()); - block->alloc(iobuffer_size_to_index(HEADER_OVERHEAD + this->length(), BUFFER_SIZE_INDEX_32K)); - uint8_t *block_start = reinterpret_cast(block->start()); - memcpy(block_start, this->_buf, this->_buf_len); - n += this->_buf_len; - - block->fill(n); - return block; -} - // // DATA Frame // @@ -283,16 +262,17 @@ Http3DataFrame::data() const // Http3HeadersFrame::Http3HeadersFrame(IOBufferReader &reader) : Http3Frame(reader) {} -Http3HeadersFrame::Http3HeadersFrame(ats_unique_buf header_block, size_t header_block_len) - : Http3Frame(Http3FrameType::HEADERS), _header_block_uptr(std::move(header_block)), _header_block_len(header_block_len) +Http3HeadersFrame::Http3HeadersFrame(IOBufferReader &header_block_reader, size_t header_block_len) + : Http3Frame(Http3FrameType::HEADERS), _header_block_len(header_block_len), _header_block_reader(header_block_reader.clone()) { - this->_length = header_block_len; - this->_header_block = this->_header_block_uptr.get(); + this->_length = header_block_len; } Http3HeadersFrame::~Http3HeadersFrame() { - if (this->_header_block_uptr == nullptr) { + if (this->_header_block_reader != nullptr) { + this->_header_block_reader->dealloc(); + } else { ats_free(this->_header_block); } } @@ -312,7 +292,11 @@ Http3HeadersFrame::to_io_buffer_block() const written += n; QUICVariableInt::encode(block_start + written, UINT64_MAX, n, this->_length); written += n; - memcpy(block_start + written, this->_header_block, this->_header_block_len); + if (this->_header_block_reader != nullptr) { + this->_header_block_reader->memcpy(block_start + written, this->_header_block_len); + } else { + memcpy(block_start + written, this->_header_block, this->_header_block_len); + } written += this->_header_block_len; block->fill(written); @@ -583,29 +567,15 @@ Http3FrameFactory::fast_create(IOBufferReader &reader) return frame; } -Http3HeadersFrameUPtr -Http3FrameFactory::create_headers_frame(const uint8_t *header_block, size_t header_block_len) -{ - ats_unique_buf buf = ats_unique_malloc(header_block_len); - memcpy(buf.get(), header_block, header_block_len); - - Http3HeadersFrame *frame = http3HeadersFrameAllocator.alloc(); - new (frame) Http3HeadersFrame(std::move(buf), header_block_len); - return Http3HeadersFrameUPtr(frame, &Http3FrameDeleter::delete_headers_frame); -} - Http3HeadersFrameUPtr Http3FrameFactory::create_headers_frame(IOBufferReader *header_block_reader, size_t header_block_len) { - ats_unique_buf buf = ats_unique_malloc(header_block_len); - - int64_t nread; - while ((nread = header_block_reader->read(buf.get(), header_block_len)) > 0) { - ; - } - Http3HeadersFrame *frame = http3HeadersFrameAllocator.alloc(); - new (frame) Http3HeadersFrame(std::move(buf), header_block_len); + new (frame) Http3HeadersFrame(*header_block_reader, header_block_len); + // The frame clones the reader before this, so consuming here only advances the caller's + // reader (for chunking header blocks larger than one generate_frame() call), not the frame's + // own clone. + header_block_reader->consume(header_block_len); return Http3HeadersFrameUPtr(frame, &Http3FrameDeleter::delete_headers_frame); } diff --git a/src/proxy/http3/Http3FrameCounter.cc b/src/proxy/http3/Http3FrameCounter.cc index d9696735a8f..f0701ef8b98 100644 --- a/src/proxy/http3/Http3FrameCounter.cc +++ b/src/proxy/http3/Http3FrameCounter.cc @@ -25,12 +25,14 @@ #include "proxy/http3/Http3.h" #include "proxy/http3/Http3FrameCounter.h" -std::vector +std::vector const & Http3FrameCounter::interests() { - return {Http3FrameType::DATA, Http3FrameType::HEADERS, Http3FrameType::X_RESERVED_1, Http3FrameType::CANCEL_PUSH, - Http3FrameType::SETTINGS, Http3FrameType::PUSH_PROMISE, Http3FrameType::X_RESERVED_2, Http3FrameType::GOAWAY, - Http3FrameType::X_RESERVED_3, Http3FrameType::X_RESERVED_4, Http3FrameType::MAX_PUSH_ID, Http3FrameType::UNKNOWN}; + static std::vector const types = { + Http3FrameType::DATA, Http3FrameType::HEADERS, Http3FrameType::X_RESERVED_1, Http3FrameType::CANCEL_PUSH, + Http3FrameType::SETTINGS, Http3FrameType::PUSH_PROMISE, Http3FrameType::X_RESERVED_2, Http3FrameType::GOAWAY, + Http3FrameType::X_RESERVED_3, Http3FrameType::X_RESERVED_4, Http3FrameType::MAX_PUSH_ID, Http3FrameType::UNKNOWN}; + return types; } Http3ErrorUPtr diff --git a/src/proxy/http3/Http3FrameDispatcher.cc b/src/proxy/http3/Http3FrameDispatcher.cc index f697cb4eadf..11c10cd5cee 100644 --- a/src/proxy/http3/Http3FrameDispatcher.cc +++ b/src/proxy/http3/Http3FrameDispatcher.cc @@ -24,6 +24,7 @@ #include "proxy/http3/Http3FrameDispatcher.h" #include "tscore/Diags.h" +#include "tscore/ink_assert.h" #include "iocore/net/quic/QUICIntUtil.h" #include "proxy/http3/Http3DebugNames.h" @@ -47,8 +48,9 @@ Http3FrameDispatcher::add_handler(Http3FrameHandler *handler) for (Http3FrameType t : handler->interests()) { auto const type = static_cast(t); if (!registered[type]) { - this->_handlers[type].push_back(handler); registered[type] = true; + ink_release_assert(this->_handler_count[type] < MAX_HANDLERS_PER_TYPE); + this->_handlers[type][this->_handler_count[type]++] = handler; } } } @@ -127,9 +129,9 @@ Http3FrameDispatcher::on_read_ready(QUICStreamId stream_id, Http3StreamType stre Http3FrameType type = this->_current_frame->type(); Dbg(dbg_ctl_http3, "[RX] [%" PRIu64 "] | %s size=%" PRIu64 "/%" PRIu64, stream_id, Http3DebugNames::frame_type(type), this->_current_frame->total_length() - _bytes_to_skip, this->_current_frame->total_length()); - std::vector handlers = this->_handlers[static_cast(type)]; - for (auto h : handlers) { - error = h->handle_frame(this->_current_frame, stream_type); + uint8_t const type_idx = static_cast(type); + for (uint8_t i = 0; i < this->_handler_count[type_idx]; ++i) { + error = this->_handlers[type_idx][i]->handle_frame(this->_current_frame, stream_type); if (error && error->cls != Http3ErrorClass::UNDEFINED) { return error; } diff --git a/src/proxy/http3/Http3HeaderVIOAdaptor.cc b/src/proxy/http3/Http3HeaderVIOAdaptor.cc index 7f489d4fa28..e53c758a3a9 100644 --- a/src/proxy/http3/Http3HeaderVIOAdaptor.cc +++ b/src/proxy/http3/Http3HeaderVIOAdaptor.cc @@ -48,10 +48,11 @@ Http3HeaderVIOAdaptor::~Http3HeaderVIOAdaptor() this->_header.destroy(); } -std::vector +std::vector const & Http3HeaderVIOAdaptor::interests() { - return {Http3FrameType::HEADERS}; + static std::vector const types = {Http3FrameType::HEADERS}; + return types; } Http3ErrorUPtr diff --git a/src/proxy/http3/Http3ProtocolEnforcer.cc b/src/proxy/http3/Http3ProtocolEnforcer.cc index 06121d3aa99..defc1ff3c0c 100644 --- a/src/proxy/http3/Http3ProtocolEnforcer.cc +++ b/src/proxy/http3/Http3ProtocolEnforcer.cc @@ -24,13 +24,15 @@ #include "proxy/http3/Http3ProtocolEnforcer.h" #include "proxy/http3/Http3DebugNames.h" -std::vector +std::vector const & Http3ProtocolEnforcer::interests() { - return {Http3FrameType::DATA, Http3FrameType::HEADERS, Http3FrameType::X_RESERVED_1, Http3FrameType::CANCEL_PUSH, - Http3FrameType::SETTINGS, Http3FrameType::PUSH_PROMISE, Http3FrameType::X_RESERVED_2, Http3FrameType::GOAWAY, - Http3FrameType::X_RESERVED_3, Http3FrameType::X_RESERVED_4, Http3FrameType::MAX_PUSH_ID, Http3FrameType::RESERVED, - Http3FrameType::UNKNOWN}; + static std::vector const types = { + Http3FrameType::DATA, Http3FrameType::HEADERS, Http3FrameType::X_RESERVED_1, Http3FrameType::CANCEL_PUSH, + Http3FrameType::SETTINGS, Http3FrameType::PUSH_PROMISE, Http3FrameType::X_RESERVED_2, Http3FrameType::GOAWAY, + Http3FrameType::X_RESERVED_3, Http3FrameType::X_RESERVED_4, Http3FrameType::MAX_PUSH_ID, Http3FrameType::RESERVED, + Http3FrameType::UNKNOWN}; + return types; } Http3ErrorUPtr diff --git a/src/proxy/http3/Http3SettingsHandler.cc b/src/proxy/http3/Http3SettingsHandler.cc index 7bc4c7ede9a..ccf95220382 100644 --- a/src/proxy/http3/Http3SettingsHandler.cc +++ b/src/proxy/http3/Http3SettingsHandler.cc @@ -32,10 +32,11 @@ DbgCtl dbg_ctl_http3{"http3"}; // // SETTINGS frame handler // -std::vector +std::vector const & Http3SettingsHandler::interests() { - return {Http3FrameType::SETTINGS}; + static std::vector const types = {Http3FrameType::SETTINGS}; + return types; } Http3ErrorUPtr diff --git a/src/proxy/http3/Http3StreamDataVIOAdaptor.cc b/src/proxy/http3/Http3StreamDataVIOAdaptor.cc index 296763972b8..cdfd1ee3744 100644 --- a/src/proxy/http3/Http3StreamDataVIOAdaptor.cc +++ b/src/proxy/http3/Http3StreamDataVIOAdaptor.cc @@ -35,10 +35,11 @@ Http3StreamDataVIOAdaptor::~Http3StreamDataVIOAdaptor() free_MIOBuffer(this->_buffer); } -std::vector +std::vector const & Http3StreamDataVIOAdaptor::interests() { - return {Http3FrameType::DATA}; + static std::vector const types = {Http3FrameType::DATA}; + return types; } Http3ErrorUPtr @@ -71,7 +72,7 @@ Http3StreamDataVIOAdaptor::finalize() } bool -Http3StreamDataVIOAdaptor::has_data() +Http3StreamDataVIOAdaptor::has_data() const { return this->_total_data_length > 0; } diff --git a/src/proxy/http3/Http3Transaction.cc b/src/proxy/http3/Http3Transaction.cc index 084ea0d566f..44b2c828ef7 100644 --- a/src/proxy/http3/Http3Transaction.cc +++ b/src/proxy/http3/Http3Transaction.cc @@ -493,30 +493,22 @@ HQTransaction::_delete_if_possible() // // Http3Transaction // -Http3Transaction::Http3Transaction(Http3Session *session, QUICStreamVCAdapter::IOInfo &info) : super(session, info) +Http3Transaction::Http3Transaction(Http3Session *session, QUICStreamVCAdapter::IOInfo &info) + : super(session, info), + _header_framer(this, &this->_write_vio, session->local_qpack(), this->_stream_id), + _data_framer(this, &this->_write_vio), + _header_handler(&this->_read_vio, this->direction() == NET_VCONNECTION_OUT ? HTTPType::RESPONSE : HTTPType::REQUEST, + session->remote_qpack(), this->_stream_id, this), + _data_handler(&this->_read_vio) { - QUICStreamId stream_id = this->_info.adapter.stream().id(); - - this->_header_framer = new Http3HeaderFramer(this, &this->_write_vio, session->local_qpack(), stream_id); - this->_data_framer = new Http3DataFramer(this, &this->_write_vio); - this->_frame_collector.add_generator(this->_header_framer); - this->_frame_collector.add_generator(this->_data_framer); + this->_frame_collector.add_generator(&this->_header_framer); + this->_frame_collector.add_generator(&this->_data_framer); // this->_frame_collector.add_generator(this->_push_controller); - HTTPType http_type = HTTPType::UNKNOWN; - if (this->direction() == NET_VCONNECTION_OUT) { - http_type = HTTPType::RESPONSE; - } else { - http_type = HTTPType::REQUEST; - } - this->_protocol_enforcer = new Http3ProtocolEnforcer(); - this->_header_handler = new Http3HeaderVIOAdaptor(&this->_read_vio, http_type, session->remote_qpack(), stream_id, this); - this->_data_handler = new Http3StreamDataVIOAdaptor(&this->_read_vio); - this->_frame_dispatcher.add_handler(session->get_received_frame_counter()); - this->_frame_dispatcher.add_handler(this->_protocol_enforcer); - this->_frame_dispatcher.add_handler(this->_header_handler); - this->_frame_dispatcher.add_handler(this->_data_handler); + this->_frame_dispatcher.add_handler(&this->_protocol_enforcer); + this->_frame_dispatcher.add_handler(&this->_header_handler); + this->_frame_dispatcher.add_handler(&this->_data_handler); SET_HANDLER(&Http3Transaction::state_stream_open); } @@ -527,24 +519,13 @@ Http3Transaction::~Http3Transaction() // This should have already been called but call it here just incase. do_io_close(); - - delete this->_header_framer; - this->_header_framer = nullptr; - delete this->_data_framer; - this->_data_framer = nullptr; - delete this->_protocol_enforcer; - this->_protocol_enforcer = nullptr; - delete this->_header_handler; - this->_header_handler = nullptr; - delete this->_data_handler; - this->_data_handler = nullptr; } VIO * Http3Transaction::do_io_write(Continuation *c, int64_t nbytes, IOBufferReader *buf, bool owner) { if (c != nullptr && nbytes > 0 && buf != nullptr) { - this->_header_framer->reset(); + this->_header_framer.reset(); } return super::do_io_write(c, nbytes, buf, owner); @@ -574,14 +555,14 @@ Http3Transaction::state_stream_open(int event, Event *edata) Http3TransVDebug("%s (%d)", get_vc_event_name(event), event); this->_close_read_complete_event(edata); int64_t nread = this->_process_read_vio(); - if (!this->_header_handler->is_complete()) { + if (!this->_header_handler.is_complete()) { if (nread > 0) { // Delay processing READ_COMPLETE until the header block can be fully decoded. this->_schedule_read_complete_event(); } break; } - this->_data_handler->finalize(); + this->_data_handler.finalize(); // always signal regardless of progress this->_signal_read_event(); if (!this->_is_closed()) { @@ -686,13 +667,13 @@ Http3Transaction::on_header_decode_complete() bool Http3Transaction::is_response_header_sent() const { - return this->_header_framer->is_final_header_sent(); + return this->_header_framer.is_final_header_sent(); } bool Http3Transaction::is_response_body_sent() const { - return this->_data_framer->is_done(); + return this->_data_framer.is_done(); } void @@ -773,7 +754,7 @@ Http3Transaction::has_request_body(int64_t content_length, bool /* is_chunked_se } // Has body if there is DATA frame received (In case Content-Length is omitted) - if (this->_data_handler->has_data()) { + if (this->_data_handler.has_data()) { return true; } diff --git a/src/proxy/http3/test/Mock.h b/src/proxy/http3/test/Mock.h index 64340ad1e9f..acdcf9f2d6b 100644 --- a/src/proxy/http3/test/Mock.h +++ b/src/proxy/http3/test/Mock.h @@ -32,10 +32,11 @@ class Http3MockFrameHandler : public Http3FrameHandler // Http3FrameHandler - std::vector + std::vector const & interests() override { - return {Http3FrameType::DATA, Http3FrameType::SETTINGS}; + static std::vector const types = {Http3FrameType::DATA, Http3FrameType::SETTINGS}; + return types; } Http3ErrorUPtr diff --git a/src/proxy/http3/test/test_Http3Frame.cc b/src/proxy/http3/test/test_Http3Frame.cc index 003c32d79e6..80461da31ee 100644 --- a/src/proxy/http3/test/test_Http3Frame.cc +++ b/src/proxy/http3/test/test_Http3Frame.cc @@ -95,7 +95,7 @@ TEST_CASE("Store DATA Frame", "[http3]") TEST_CASE("Store HEADERS Frame", "[http3]") { - SECTION("Normal") + SECTION("From reader, via factory") { uint8_t buf[32] = {0}; size_t len; @@ -105,20 +105,31 @@ TEST_CASE("Store HEADERS Frame", "[http3]") 0x11, 0x22, 0x33, 0x44, // Payload }; - uint8_t raw1[] = "\x11\x22\x33\x44"; - ats_unique_buf header_block = ats_unique_malloc(4); - memcpy(header_block.get(), raw1, 4); + uint8_t raw1[] = "\x11\x22\x33\x44"; + MIOBuffer *header_block = new_MIOBuffer(BUFFER_SIZE_INDEX_8K); + header_block->set(raw1, 4); + IOBufferReader *header_block_reader = header_block->alloc_reader(); - Http3HeadersFrame hdrs_frame(std::move(header_block), 4); - CHECK(hdrs_frame.length() == 4); + Http3HeadersFrameUPtr frame = Http3FrameFactory::create_headers_frame(header_block_reader, 4); + CHECK(frame->length() == 4); + // The factory must consume the caller's reader (needed so a header block sent across + // multiple generate_frame() calls advances instead of re-reading the same bytes). + CHECK(header_block_reader->read_avail() == 0); - auto ibb = hdrs_frame.to_io_buffer_block(); + // The frame must still serialize correctly even though the original reader was already + // consumed above -- it holds its own independent clone of the reader. + auto ibb = frame->to_io_buffer_block(); IOBufferReader reader; reader.block = ibb.get(); len = reader.read_avail(); reader.read(buf, sizeof(buf)); CHECK(len == 6); CHECK(memcmp(buf, expected1, len) == 0); + + // The frame holds a reader cloned from header_block; it must be torn down before the + // source MIOBuffer is freed, or its destructor dealloc()s a reader into freed memory. + frame.reset(); + free_MIOBuffer(header_block); } } diff --git a/src/proxy/http3/test/test_Http3FrameDispatcher.cc b/src/proxy/http3/test/test_Http3FrameDispatcher.cc index dec273aff4a..e4466ea1756 100644 --- a/src/proxy/http3/test/test_Http3FrameDispatcher.cc +++ b/src/proxy/http3/test/test_Http3FrameDispatcher.cc @@ -32,10 +32,11 @@ namespace class AliasedInterestsFrameHandler : public Http3FrameHandler { public: - std::vector + std::vector const & interests() override { - return {Http3FrameType::MAX_PUSH_ID, Http3FrameType::X_MAX_DEFINED}; + static std::vector const types = {Http3FrameType::MAX_PUSH_ID, Http3FrameType::X_MAX_DEFINED}; + return types; } Http3ErrorUPtr