From bb868d0d2f5fb644ca53ce833c0c34444ce4a4fa Mon Sep 17 00:00:00 2001 From: Peter Eisenlohr Date: Thu, 20 Aug 2026 04:11:40 +0000 Subject: [PATCH 01/20] remove non-awaitable request handler --- include/anyhttp/server.hpp | 4 +--- include/anyhttp/server_impl.hpp | 6 ------ src/beast_session.cpp | 2 +- src/nghttp2_stream.cpp | 2 +- src/server.cpp | 5 ----- src/server_impl_udp.cpp | 2 +- src/server_main.cpp | 2 +- test/test_server.cpp | 2 +- 8 files changed, 6 insertions(+), 19 deletions(-) diff --git a/include/anyhttp/server.hpp b/include/anyhttp/server.hpp index 5075b1f..c64c44b 100644 --- a/include/anyhttp/server.hpp +++ b/include/anyhttp/server.hpp @@ -135,8 +135,7 @@ class Response // ================================================================================================= -using RequestHandler = std::function; -using RequestHandlerCoro = std::function(Request, Response)>; +using RequestHandler = std::function(Request, Response)>; class Server { @@ -151,7 +150,6 @@ class Server executor_type get_executor() const noexcept; void setRequestHandler(RequestHandler&& handler); - void setRequestHandlerCoro(RequestHandlerCoro&& handler); asio::ip::tcp::endpoint local_endpoint() const; diff --git a/include/anyhttp/server_impl.hpp b/include/anyhttp/server_impl.hpp index 7744dd9..2063fb4 100644 --- a/include/anyhttp/server_impl.hpp +++ b/include/anyhttp/server_impl.hpp @@ -79,12 +79,7 @@ class Server::Impl : public std::enable_shared_from_this } void setRequestHandler(RequestHandler&& handler) { m_requestHandler = std::move(handler); } - void setRequestHandler(RequestHandlerCoro&& handler) - { - m_requestHandlerCoro = std::move(handler); - } const RequestHandler& requestHandler() const { return m_requestHandler; } - const RequestHandlerCoro& requestHandlerCoro() const { return m_requestHandlerCoro; } asio::awaitable udp_receive_loop(); int udp_on_read(Endpoint& ep); @@ -110,7 +105,6 @@ class Server::Impl : public std::enable_shared_from_this std::unordered_map> m_quic_handlers; RequestHandler m_requestHandler; - RequestHandlerCoro m_requestHandlerCoro; bool m_stopped = false; }; diff --git a/src/beast_session.cpp b/src/beast_session.cpp index 003f42c..608bfee 100644 --- a/src/beast_session.cpp +++ b/src/beast_session.cpp @@ -716,7 +716,7 @@ awaitable ServerSession::do_session(Buffer&& buffer) // server::Request request_wrapper(std::move(reader)); server::Response response_wrapper(std::move(writer)); - if (auto& handler = server().requestHandlerCoro()) + if (auto& handler = server().requestHandler()) { try { diff --git a/src/nghttp2_stream.cpp b/src/nghttp2_stream.cpp index 8ded962..9b15565 100644 --- a/src/nghttp2_stream.cpp +++ b/src/nghttp2_stream.cpp @@ -762,7 +762,7 @@ void NGHttp2Stream::on_request() server::Response response(std::make_unique>(*this)); auto& server = dynamic_cast(parent).server(); - if (auto& handler = server.requestHandlerCoro()) + if (auto& handler = server.requestHandler()) co_spawn(get_executor(), handler(std::move(request), std::move(response)), detached); else if (auto& handler = server.requestHandler()) server.requestHandler()(std::move(request), std::move(response)); diff --git a/src/server.cpp b/src/server.cpp index 3215597..40e7022 100644 --- a/src/server.cpp +++ b/src/server.cpp @@ -116,11 +116,6 @@ void Server::setRequestHandler(RequestHandler&& handler) impl->setRequestHandler(std::move(handler)); } -void Server::setRequestHandlerCoro(RequestHandlerCoro&& handler) -{ - impl->setRequestHandler(std::move(handler)); -} - asio::any_io_executor Server::get_executor() const noexcept { return impl->get_executor(); } asio::ip::tcp::endpoint Server::local_endpoint() const { return impl->local_endpoint(); } diff --git a/src/server_impl_udp.cpp b/src/server_impl_udp.cpp index fd0d07e..1727161 100644 --- a/src/server_impl_udp.cpp +++ b/src/server_impl_udp.cpp @@ -2090,7 +2090,7 @@ int Http3Session::h3_cb_end_headers(nghttp3_conn*, int64_t stream_id, int /*fin* server::Response response(std::make_unique>(*s)); auto& sv = self->server_; - if (auto& handler = sv.requestHandlerCoro()) + if (auto& handler = sv.requestHandler()) co_spawn(self->get_executor(), handler(std::move(request), std::move(response)), detached); else if (auto& handler = sv.requestHandler()) handler(std::move(request), std::move(response)); diff --git a/src/server_main.cpp b/src/server_main.cpp index 5de244e..83ece3b 100644 --- a/src/server_main.cpp +++ b/src/server_main.cpp @@ -97,7 +97,7 @@ int main(int argc, char* argv[]) server.reset(); }); - server->setRequestHandlerCoro( + server->setRequestHandler( [](server::Request request, server::Response response) -> awaitable { std::string path = request.url().path(); diff --git a/test/test_server.cpp b/test/test_server.cpp index 5a12afd..126082c 100644 --- a/test/test_server.cpp +++ b/test/test_server.cpp @@ -183,7 +183,7 @@ class Server : public testing::TestWithParam // strand is created after accepting a new connection. // server.emplace(context.get_executor(), config); - server->setRequestHandlerCoro( + server->setRequestHandler( [this](server::Request request, server::Response response) -> awaitable { logd("{} ({})", request.url().path(), request.url().buffer()); From cfe0e77278aaf4bfea1672a6ed8e8c90e04c3eea Mon Sep 17 00:00:00 2001 From: Peter Eisenlohr Date: Thu, 20 Aug 2026 12:42:43 +0000 Subject: [PATCH 02/20] remove redundant request handler invocation in NGHttp2Stream and Http3Session --- src/nghttp2_stream.cpp | 2 -- src/server_impl_udp.cpp | 2 -- 2 files changed, 4 deletions(-) diff --git a/src/nghttp2_stream.cpp b/src/nghttp2_stream.cpp index 9b15565..18ad37c 100644 --- a/src/nghttp2_stream.cpp +++ b/src/nghttp2_stream.cpp @@ -764,8 +764,6 @@ void NGHttp2Stream::on_request() auto& server = dynamic_cast(parent).server(); if (auto& handler = server.requestHandler()) co_spawn(get_executor(), handler(std::move(request), std::move(response)), detached); - else if (auto& handler = server.requestHandler()) - server.requestHandler()(std::move(request), std::move(response)); else { loge("[{}] on_request: no request handler!", logPrefix); diff --git a/src/server_impl_udp.cpp b/src/server_impl_udp.cpp index 1727161..d1bae3d 100644 --- a/src/server_impl_udp.cpp +++ b/src/server_impl_udp.cpp @@ -2092,8 +2092,6 @@ int Http3Session::h3_cb_end_headers(nghttp3_conn*, int64_t stream_id, int /*fin* auto& sv = self->server_; if (auto& handler = sv.requestHandler()) co_spawn(self->get_executor(), handler(std::move(request), std::move(response)), detached); - else if (auto& handler = sv.requestHandler()) - handler(std::move(request), std::move(response)); else { loge("[{}] no request handler set", s->log_prefix); From fe142ca2e2ea16956dbab8d370e4bfae9e2d468f Mon Sep 17 00:00:00 2001 From: Peter Eisenlohr Date: Thu, 20 Aug 2026 18:19:00 +0000 Subject: [PATCH 03/20] HTTP/3: resume a stream blocked mid-write A single async_write() larger than kWriteChunkSize (16K) stalled forever: data_reader() answers NGHTTP3_ERR_WOULDBLOCK while the current chunk is offered but not yet confirmed, which makes nghttp3 block the stream, and nothing ever resumed it -- start_write() was the only caller of nghttp3_conn_resume_stream(), and for a multi-chunk write there is no next start_write() to reach it. The response body ended up truncated after the first chunk, with no FIN, leaving the peer waiting forever. Resume the stream from on_write_consumed() once the current chunk is confirmed and there is more of write_source left to carve up. Writes of up to one chunk are unaffected, which is why nothing hit this so far: every existing sender chunks at 16K. The client side had the same latent defect for request bodies written in one call; fixed identically to keep the two implementations parallel. Co-Authored-By: Claude Opus 5 --- src/client_impl_udp.cpp | 20 +++++++++++++++++++- src/server_impl_udp.cpp | 21 +++++++++++++++++++-- test/test_server.cpp | 35 +++++++++++++++++++++++++++++++++++ 3 files changed, 73 insertions(+), 3 deletions(-) diff --git a/src/client_impl_udp.cpp b/src/client_impl_udp.cpp index 7eba893..16217ad 100644 --- a/src/client_impl_udp.cpp +++ b/src/client_impl_udp.cpp @@ -891,11 +891,29 @@ void Http3ClientStream::on_write_consumed(size_t n) n = std::min(n, write_chunk.size() - write_confirmed); write_confirmed += n; + if (write_confirmed < write_chunk.size()) + return; + // The write is fully done once its current chunk is confirmed and there is no more of // write_source left to carve into further chunks -- data_reader() advances write_chunk/ // write_source_copied otherwise, so this is the terminal state. - if (write_confirmed == write_chunk.size() && write_source_copied == asio::buffer_size(write_source)) + if (write_source_copied == asio::buffer_size(write_source)) + { finish_active_write(); + return; + } + + // + // There is more of write_source to carve into chunks, but nghttp3 may have asked for data + // while this chunk was offered and still unconfirmed, in which case data_reader() answered + // NGHTTP3_ERR_WOULDBLOCK -- and a blocked stream is never polled again until it is explicitly + // resumed. Now that the chunk is confirmed, there is something new to hand out, so unblock + // the stream. Without this, any single async_write() larger than kWriteChunkSize stalls here + // forever, with the response body truncated and no FIN. + // + if (auto h3 = session.h3()) + nghttp3_conn_resume_stream(h3, id); + session.wake_write(); } void Http3ClientStream::finish_active_write() diff --git a/src/server_impl_udp.cpp b/src/server_impl_udp.cpp index d1bae3d..89fdab3 100644 --- a/src/server_impl_udp.cpp +++ b/src/server_impl_udp.cpp @@ -1063,12 +1063,29 @@ void Http3Stream::on_write_consumed(size_t n) n = std::min(n, write_chunk.size() - write_confirmed); write_confirmed += n; + if (write_confirmed < write_chunk.size()) + return; + // The write is fully done once its current chunk is confirmed and there is no more of // write_source left to carve into further chunks -- data_reader() advances write_chunk/ // write_source_copied otherwise, so this is the terminal state. - if (write_confirmed == write_chunk.size() && - write_source_copied == asio::buffer_size(write_source)) + if (write_source_copied == asio::buffer_size(write_source)) + { finish_active_write(); + return; + } + + // + // There is more of write_source to carve into chunks, but nghttp3 may have asked for data + // while this chunk was offered and still unconfirmed, in which case data_reader() answered + // NGHTTP3_ERR_WOULDBLOCK -- and a blocked stream is never polled again until it is explicitly + // resumed. Now that the chunk is confirmed, there is something new to hand out, so unblock + // the stream. Without this, any single async_write() larger than kWriteChunkSize stalls here + // forever, with the response body truncated and no FIN. + // + if (auto h3 = session.h3()) + nghttp3_conn_resume_stream(h3, id); + session.wake_write(); } void Http3Stream::finish_active_write() diff --git a/test/test_server.cpp b/test/test_server.cpp index 126082c..518d3dd 100644 --- a/test/test_server.cpp +++ b/test/test_server.cpp @@ -958,6 +958,41 @@ TEST_P(ClientAsync, HelloWorld) // ------------------------------------------------------------------------------------------------- +// +// A single async_write() larger than what the transport hands to its peer in one go, i.e. the +// whole body in one call instead of chunk by chunk. HTTP/3 used to stall here: with its response +// chunk fully offered but not yet confirmed, nghttp3 got NGHTTP3_ERR_WOULDBLOCK and blocked the +// stream, which nothing resumed once the chunk was confirmed. +// +TEST_P(ClientAsync, WHEN_server_writes_large_buffer_at_once_THEN_receives_all) +{ + static const std::vector body = [] + { + std::vector data(256 * 1024); + std::ranges::generate(data, [i = uint8_t(0)]() mutable { return i++; }); + return data; + }(); + + custom = [this](server::Request request, server::Response response) -> awaitable + { + std::array buffer; + while (co_await request.async_read_some(asio::buffer(buffer)) > 0) + ; // drain the request -- HTTP/1.1 closes the connection on an unfinished parser + + co_await response.async_submit(200, fields({{"Content-Length", body.size()}})); + co_await response.async_write(asio::buffer(body)); + co_await response.async_write({}); + }; + test = [this](Session session) -> awaitable + { + auto request = co_await session.async_submit(url); + co_await request.async_write({}); + EXPECT_EQ(co_await read_response(request), body.size()); + }; +} + +// ------------------------------------------------------------------------------------------------- + TEST_P(ClientAsync, ServerYieldFirst) { custom = [this](server::Request request, server::Response response) -> awaitable From 04ddcdeef256150a5943b4aee1504bcde63b9381 Mon Sep 17 00:00:00 2001 From: Peter Eisenlohr Date: Thu, 20 Aug 2026 18:19:36 +0000 Subject: [PATCH 04/20] add a request handler serving files with mmap() serve_file() maps the requested file into memory for as long as the coroutine frame lives, so it is unmapped on normal exit, on exception and on cancellation alike. Empty files are served without a mapping at all -- mmap() rejects a zero length, and the empty buffer that would be written already means EOF. The URL path below the mount prefix is resolved against the docroot with weakly_canonical(), which folds "..", "." and symlinks before the result is compared against the canonical root, so nothing outside of it can be reached. The prefix has to match whole path segments. Errors map to 404 (missing, directory, escape), 403 (unreadable) or 500. Mounted on "/test" in the example server, serving the test/ directory. Tests cover content, subdirectories, empty and large (multi-chunk) files and every error condition, over HTTP/1.1, HTTP/2 and HTTP/3. To let a prefix-mounted handler see sub-paths at all, the test server now routes everything below /custom to the testcase handler. Co-Authored-By: Claude Opus 5 --- include/anyhttp/file_handler.hpp | 23 ++++ src/file_handler.cpp | 223 +++++++++++++++++++++++++++++++ src/server_main.cpp | 3 + test/test_server.cpp | 217 +++++++++++++++++++++++++++++- 4 files changed, 465 insertions(+), 1 deletion(-) create mode 100644 include/anyhttp/file_handler.hpp create mode 100644 src/file_handler.cpp diff --git a/include/anyhttp/file_handler.hpp b/include/anyhttp/file_handler.hpp new file mode 100644 index 0000000..ad87006 --- /dev/null +++ b/include/anyhttp/file_handler.hpp @@ -0,0 +1,23 @@ +#pragma once + +#include "anyhttp/server.hpp" + +#include +#include + +namespace anyhttp +{ + +// ================================================================================================= + +/** + * Serve a single file from below \p root, mapped into memory with mmap(). The part of the request + * path below \p prefix is taken as the path relative to \p root; anything that would escape + * \p root ("..", a symlink pointing outside) is rejected with 404. + */ +awaitable serve_file(server::Request request, server::Response response, + std::filesystem::path root, std::string prefix); + +// ================================================================================================= + +} // namespace anyhttp diff --git a/src/file_handler.cpp b/src/file_handler.cpp new file mode 100644 index 0000000..a61ba9d --- /dev/null +++ b/src/file_handler.cpp @@ -0,0 +1,223 @@ +#include "anyhttp/file_handler.hpp" +#include "anyhttp/formatter.hpp" // IWYU pragma: keep +#include "anyhttp/request_handlers.hpp" // for send() + +#include +#include +#include +#include + +#include + +using namespace std::string_view_literals; +using namespace anyhttp; +using boost::system::error_code; + +// ================================================================================================= + +namespace +{ + +namespace fs = std::filesystem; + +error_code from_errno(int error) { return {error, boost::system::system_category()}; } + +// +// A file mapped into memory for as long as this object lives. An empty file maps to an empty +// span: mmap() rejects a zero length, and there is nothing to send anyway. +// +class MappedFile +{ +public: + MappedFile() = default; + MappedFile(MappedFile&& other) noexcept + : m_data(std::exchange(other.m_data, nullptr)), m_size(std::exchange(other.m_size, 0)), + m_mtime(other.m_mtime) + { + } + MappedFile& operator=(MappedFile&& other) noexcept + { + std::swap(m_data, other.m_data); + std::swap(m_size, other.m_size); + std::swap(m_mtime, other.m_mtime); + return *this; + } + ~MappedFile() + { + if (m_data) + ::munmap(m_data, m_size); + } + + static expected open(const fs::path& path) + { + const int fd = ::open(path.c_str(), O_RDONLY | O_CLOEXEC); + if (fd < 0) + return std::unexpected(from_errno(errno)); + auto close = defer([fd] { ::close(fd); }); + + struct stat st{}; + if (::fstat(fd, &st) < 0) + return std::unexpected(from_errno(errno)); + + // + // A directory can be open()ed, but not mapped -- and we do not serve listings anyway. + // Anything else that is not a regular file (FIFO, device, socket) has no size to speak of. + // + if (!S_ISREG(st.st_mode)) + return std::unexpected(from_errno(S_ISDIR(st.st_mode) ? EISDIR : EINVAL)); + + MappedFile file; + file.m_size = static_cast(st.st_size); + file.m_mtime = std::chrono::system_clock::from_time_t(st.st_mtime); + if (file.m_size == 0) + return file; + + void* data = ::mmap(nullptr, file.m_size, PROT_READ, MAP_PRIVATE, fd, 0); + if (data == MAP_FAILED) + return std::unexpected(from_errno(errno)); + + ::posix_madvise(data, file.m_size, POSIX_MADV_SEQUENTIAL); + file.m_data = data; + return file; + } + + size_t size() const noexcept { return m_size; } + auto mtime() const noexcept { return m_mtime; } + std::span bytes() const noexcept + { + return {static_cast(m_data), m_size}; + } + +private: + void* m_data = nullptr; + size_t m_size = 0; + std::chrono::system_clock::time_point m_mtime; +}; + +// +// Map the part of the request path below 'prefix' onto 'root'. weakly_canonical() resolves ".." +// and symlinks, so comparing the result against the canonical root catches any attempt to reach +// outside of it. +// +expected resolve(std::string_view path, std::string_view prefix, const fs::path& root) +{ + const auto reject = std::unexpected(from_errno(ENOENT)); + + if (!path.starts_with(prefix)) + return reject; + path.remove_prefix(prefix.size()); + if (!path.empty() && !path.starts_with('/')) + return reject; // 'prefix' matched in the middle of a segment, e.g. "/testament" for "/test" + + while (path.starts_with('/')) + path.remove_prefix(1); + + std::error_code ec; + const auto base = fs::weakly_canonical(root, ec); + if (ec) + return std::unexpected(from_errno(ec.value())); + + const auto file = fs::weakly_canonical(base / fs::path(path), ec); + if (ec) + return std::unexpected(from_errno(ec.value())); + + const auto relative = file.lexically_relative(base); + if (relative.empty() || *relative.begin() == "..") + return reject; + + return file; +} + +std::string_view content_type(const fs::path& path) +{ + static constexpr std::pair types[] = { + {".css", "text/css"}, {".gif", "image/gif"}, {".htm", "text/html"}, + {".html", "text/html"}, {".jpeg", "image/jpeg"}, {".jpg", "image/jpeg"}, + {".js", "text/javascript"}, {".json", "application/json"}, {".pdf", "application/pdf"}, + {".png", "image/png"}, {".svg", "image/svg+xml"}, {".txt", "text/plain"}, + {".xml", "application/xml"}, {".zip", "application/zip"}}; + + auto extension = path.extension().string(); + std::ranges::transform(extension, extension.begin(), + [](unsigned char ch) { return std::tolower(ch); }); + + const auto it = + std::ranges::find(types, extension, &std::pair::first); + return it == std::ranges::end(types) ? "application/octet-stream"sv : it->second; +} + +unsigned status_for(const error_code& ec) +{ + switch (ec.value()) + { + case ENOENT: + case ENOTDIR: + case EISDIR: + case ENAMETOOLONG: + return 404; + case EACCES: + case EPERM: + return 403; + default: + return 500; + } +} + +awaitable respond(server::Response& response, unsigned status) +{ + co_await response.async_submit(status, fields({{"Content-Length", 0}})); + co_await response.async_write({}); +} + +} // namespace + +namespace anyhttp +{ + +awaitable serve_file(server::Request request, server::Response response, fs::path root, + std::string prefix) +{ + // + // Read the request body to EOF before responding. A GET normally carries none, but HTTP/1.1 + // has to close the connection when a handler leaves the request unparsed, which would + // truncate the response we are about to write. + // + std::array discard; + while (co_await request.async_read_some(asio::buffer(discard)) > 0) + ; + + const std::string path = request.url().path(); + const auto resolved = resolve(path, prefix, root); + if (!resolved) + { + logw("serve_file: {}: {}", path, resolved.error().message()); + co_await respond(response, status_for(resolved.error())); + co_return; + } + + const auto file = MappedFile::open(*resolved); + if (!file) + { + logw("serve_file: {}: {}", resolved->native(), file.error().message()); + co_await respond(response, status_for(file.error())); + co_return; + } + + logd("serve_file: {} ({} bytes)", resolved->native(), file->size()); + co_await response.async_submit(200, + fields({{"Content-Length", file->size()}, + {"Content-Type", content_type(*resolved)}, + {"Last-Modified", format_http_date(file->mtime())}})); + + // + // The mapping lives in the coroutine frame, so it stays valid across the write and is + // unmapped no matter how we leave -- normally, by exception or by cancellation. Note that + // touching a mapped page may block on disk I/O, which no amount of chunking would avoid. + // + if (file->size() > 0) + co_await send(response, file->bytes()); // an empty write already means EOF, don't send two + + co_await response.async_write({}); +} + +} // namespace anyhttp diff --git a/src/server_main.cpp b/src/server_main.cpp index 83ece3b..d4d084a 100644 --- a/src/server_main.cpp +++ b/src/server_main.cpp @@ -1,3 +1,4 @@ +#include "anyhttp/file_handler.hpp" #include "anyhttp/request_handlers.hpp" #include "anyhttp/server.hpp" #include "anyhttp/utils.hpp" @@ -111,6 +112,8 @@ int main(int argc, char* argv[]) co_await dump(std::move(request), std::move(response)); else if (path == "/discard") co_return; + else if (path == "/test" || path.starts_with("/test/")) + co_await serve_file(std::move(request), std::move(response), "test", "/test"); else if (path == "/eat_request") co_await eat_request(std::move(request), std::move(response)); else if (path == "/" || path == "/h2spec") diff --git a/test/test_server.cpp b/test/test_server.cpp index 518d3dd..228172d 100644 --- a/test/test_server.cpp +++ b/test/test_server.cpp @@ -1,4 +1,5 @@ #include "anyhttp/client.hpp" +#include "anyhttp/file_handler.hpp" #include "anyhttp/formatter.hpp" // IWYU pragma: keep #include "anyhttp/request_handlers.hpp" #include "anyhttp/server.hpp" @@ -42,6 +43,8 @@ #include +#include + #include #include @@ -218,7 +221,7 @@ class Server : public testing::TestWithParam co_await dump(std::move(request), std::move(response)); else if (request.url().path() == "/detach") co_await detach(std::move(request), std::move(response)); - else if (request.url().path() == "/custom") + else if (request.url().path().starts_with("/custom")) co_await custom(std::move(request), std::move(response)); else co_await not_found(std::move(request), std::move(response)); @@ -991,6 +994,218 @@ TEST_P(ClientAsync, WHEN_server_writes_large_buffer_at_once_THEN_receives_all) }; } +// ================================================================================================= + +// +// serve_file() mounted on "/custom", serving a directory tree created fresh for each testcase. +// +class FileHandler : public ClientAsync +{ +public: + void SetUp() override + { + base = std::filesystem::temp_directory_path() / + std::format("anyhttp-file-handler-{}", ::getpid()); + root = base / "docroot"; + std::filesystem::remove_all(base); + std::filesystem::create_directories(root / "sub"); + + write(root / "hello.txt", "Hello, File!"); + write(root / "empty.txt", ""); + write(root / "sub" / "nested.txt", "Nested!"); + write(root / "large.bin", std::string(256 * 1024, 'x')); + write(root / "secret.txt", "no peeking"); + write(root / "er.txt", "leaked"); // what "/customer.txt" resolves to without a segment check + std::filesystem::permissions(root / "secret.txt", std::filesystem::perms::none); + std::filesystem::create_symlink(base / "outside.txt", root / "escape.txt"); + + // just outside the root, reachable only by escaping it -- so a missing check shows up as + // content served instead of a 404 + write(base / "outside.txt", "outside"); + + ClientAsync::SetUp(); + + custom = [this](server::Request request, server::Response response) -> awaitable { + co_await serve_file(std::move(request), std::move(response), root, "/custom"); + }; + } + + void TearDown() override + { + ClientAsync::TearDown(); + std::filesystem::remove_all(base); + } + + static void write(const std::filesystem::path& path, std::string_view content) + { + std::ofstream(path, std::ios::binary).write(content.data(), content.size()); + } + + // + // Requests \p target and returns status code and body. The request is finished right away -- + // serve_file() ignores the request body, but still has to consume it. + // + awaitable> get(Session& session, boost::urls::url target) + { + auto request = co_await session.async_submit(target, {}); + co_await request.async_write({}); + auto response = co_await request.async_get_response(); + auto body = co_await read(response); + co_return std::make_tuple(response.status_code(), std::move(body)); + } + + awaitable> get(Session& session, std::string_view path) + { + co_return co_await get(session, boost::urls::url(url).set_path(path)); + } + + /// Target with a percent-encoded path, passed to the server as-is. + boost::urls::url encoded(std::string_view path) const + { + auto target = boost::urls::url(url); + target.set_encoded_path(path); + return target; + } + +protected: + std::filesystem::path base; ///< holds the docroot and the file just outside of it + std::filesystem::path root; ///< what serve_file() is mounted on +}; + +INSTANTIATE_TEST_SUITE_P(FileHandler, FileHandler, + ::testing::Values(anyhttp::Protocol::http11, anyhttp::Protocol::h2, + anyhttp::Protocol::h3), + NameGenerator); + +// ------------------------------------------------------------------------------------------------- + +TEST_P(FileHandler, WHEN_file_exists_THEN_serves_content) +{ + test = [this](Session session) -> awaitable + { + auto [status, body] = co_await get(session, "/custom/hello.txt"); + EXPECT_EQ(status, 200); + EXPECT_EQ(body, "Hello, File!"); + }; +} + +TEST_P(FileHandler, WHEN_file_is_in_subdirectory_THEN_serves_content) +{ + test = [this](Session session) -> awaitable + { + auto [status, body] = co_await get(session, "/custom/sub/nested.txt"); + EXPECT_EQ(status, 200); + EXPECT_EQ(body, "Nested!"); + }; +} + +// +// An empty file cannot be mmap()ed at all, and must not send the empty buffer that already means +// EOF twice. +// +TEST_P(FileHandler, WHEN_file_is_empty_THEN_serves_empty_body) +{ + test = [this](Session session) -> awaitable + { + auto [status, body] = co_await get(session, "/custom/empty.txt"); + EXPECT_EQ(status, 200); + EXPECT_EQ(body, ""); + }; +} + +// +// Larger than the 16K chunk the HTTP/3 write path carves a single async_write() into. +// +TEST_P(FileHandler, WHEN_file_is_large_THEN_serves_all_of_it) +{ + test = [this](Session session) -> awaitable + { + auto [status, body] = co_await get(session, "/custom/large.bin"); + EXPECT_EQ(status, 200); + EXPECT_EQ(body, std::string(256 * 1024, 'x')); + }; +} + +TEST_P(FileHandler, WHEN_file_does_not_exist_THEN_error_404) +{ + test = [this](Session session) -> awaitable + { + auto [status, body] = co_await get(session, "/custom/missing.txt"); + EXPECT_EQ(status, 404); + EXPECT_EQ(body, ""); + }; +} + +// +// A directory can be open()ed but not mapped, and we do not serve listings. +// +TEST_P(FileHandler, WHEN_path_is_a_directory_THEN_error_404) +{ + test = [this](Session session) -> awaitable + { + EXPECT_EQ(std::get<0>(co_await get(session, "/custom/sub")), 404); + EXPECT_EQ(std::get<0>(co_await get(session, "/custom/")), 404); + }; +} + +TEST_P(FileHandler, WHEN_path_escapes_the_root_THEN_error_404) +{ + test = [this](Session session) -> awaitable + { + EXPECT_EQ(std::get<0>(co_await get(session, "/custom/../outside.txt")), 404); + EXPECT_EQ(std::get<0>(co_await get(session, "/custom/sub/../../outside.txt")), 404); + EXPECT_EQ(std::get<0>(co_await get(session, encoded("/custom/%2e%2e/outside.txt"))), 404); + }; +} + +// +// weakly_canonical() resolves the link, so a link out of the root is caught like any other escape. +// +TEST_P(FileHandler, WHEN_symlink_points_outside_the_root_THEN_error_404) +{ + test = [this](Session session) -> awaitable + { + EXPECT_EQ(std::get<0>(co_await get(session, "/custom/escape.txt")), 404); + }; +} + +// +// The mount prefix must match whole path segments, not just any leading characters: stripping +// "/custom" off "/customer.txt" would otherwise serve "er.txt" out of the docroot. +// +TEST_P(FileHandler, WHEN_prefix_matches_mid_segment_THEN_error_404) +{ + test = [this](Session session) -> awaitable + { + auto [status, body] = co_await get(session, "/customer.txt"); + EXPECT_EQ(status, 404); + EXPECT_EQ(body, ""); + EXPECT_EQ(std::get<0>(co_await get(session, "/customer/hello.txt")), 404); + }; +} + +TEST_P(FileHandler, WHEN_file_is_not_readable_THEN_error_403) +{ + if (::geteuid() == 0) + GTEST_SKIP() << "running as root, permissions do not apply"; + + test = [this](Session session) -> awaitable + { + auto [status, body] = co_await get(session, "/custom/secret.txt"); + EXPECT_EQ(status, 403); + EXPECT_EQ(body, ""); + }; +} + +TEST_P(FileHandler, WHEN_same_file_is_requested_twice_THEN_serves_it_twice) +{ + test = [this](Session session) -> awaitable + { + EXPECT_EQ(std::get<1>(co_await get(session, "/custom/hello.txt")), "Hello, File!"); + EXPECT_EQ(std::get<1>(co_await get(session, "/custom/hello.txt")), "Hello, File!"); + }; +} + // ------------------------------------------------------------------------------------------------- TEST_P(ClientAsync, ServerYieldFirst) From 100773c0e44ff8f81bf1f999b01baba13ddf4b5b Mon Sep 17 00:00:00 2001 From: Peter Eisenlohr Date: Thu, 20 Aug 2026 18:20:08 +0000 Subject: [PATCH 05/20] HTTP/3: send a Date header with the response --- src/server_impl_udp.cpp | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/server_impl_udp.cpp b/src/server_impl_udp.cpp index 89fdab3..5c98cfc 100644 --- a/src/server_impl_udp.cpp +++ b/src/server_impl_udp.cpp @@ -842,7 +842,9 @@ void Http3Stream::submit_response() auto status_str = std::to_string(response_status); std::vector nva; nva.reserve(16); // small typical header count; vector will grow if needed + auto date_str = format_http_date(std::chrono::system_clock::now()); nva.push_back(make_nv(":status", status_str)); + nva.push_back(make_nv("date", date_str)); nva.push_back(make_nv("server", "anyhttp-quic/0.1")); if (response_content_length) From 2bfc883ef89a0993c511852920e772deda8b6573 Mon Sep 17 00:00:00 2001 From: Peter Eisenlohr Date: Thu, 20 Aug 2026 20:25:30 +0000 Subject: [PATCH 06/20] ignore out-of-tree build directories Separate build trees like build-asan sit next to build/ and should not show up as untracked. Co-Authored-By: Claude Opus 5 --- .gitignore | 1 + 1 file changed, 1 insertion(+) diff --git a/.gitignore b/.gitignore index a99041f..6c89aab 100644 --- a/.gitignore +++ b/.gitignore @@ -5,6 +5,7 @@ docs/_build gtest-parallel-logs report.xml build +build-* secrets Testing *.pcap From e4cc39f9c1078bfabd1bafb62847a1909036d556 Mon Sep 17 00:00:00 2001 From: Peter Eisenlohr Date: Thu, 20 Aug 2026 20:27:17 +0000 Subject: [PATCH 07/20] HTTP/3: pass the response body to nghttp3 by reference data_reader() used to copy the caller's buffer into a bounded write_chunk (kWriteChunkSize) and hand nghttp3 pointers into that copy, so a write could complete as soon as the copy was made. Point the nghttp3_vec straight into the caller's buffer instead: the body now travels from the page cache into QUIC packets without an intermediate byte, which for serve_file() means the mmap()ed file is never copied on its way down. What that costs is when the write completes. ngtcp2 keeps pointing into that memory for as long as the bytes may still be retransmitted, so the write handler -- which is what releases the caller's buffer -- has to wait for the data to be acknowledged, per nghttp3's documented retention contract for read_data. A single async_write() is therefore done roughly a round trip later rather than roughly a memcpy later, as it still is on HTTP/2. write_chunk, write_source_copied, write_confirmed and in_flight_writes give way to write_offered (handed to nghttp3) and write_acked (reported back through the new acked_stream_data callback); the write completes once write_acked catches up with the buffer. Cancelling mid-write can no longer just abandon the remainder, because the bytes already offered are still referenced. Reset the stream instead, which makes ngtcp2 drop the queued data and stop reclaiming in-flight bytes -- a body cut short is truncated either way, which is what delete_writer() already resets for. A write that offered nothing, or whose bytes are all acknowledged, leaves the stream unharmed. Co-Authored-By: Claude Opus 5 --- src/server_impl_udp.cpp | 240 ++++++++++++++++++++-------------------- test/test_server.cpp | 57 +++++++++- 2 files changed, 174 insertions(+), 123 deletions(-) diff --git a/src/server_impl_udp.cpp b/src/server_impl_udp.cpp index 5c98cfc..099c4b7 100644 --- a/src/server_impl_udp.cpp +++ b/src/server_impl_udp.cpp @@ -282,14 +282,6 @@ void ngtcp2_log_printf(void* /*user*/, const char* fmt, ...) noexcept class Http3Session; class Http3Stream; -// -// Bound on how much of the caller's async_write() buffer we copy into write_chunk at a time (see -// Http3Stream's write_* members) -- copying is paced by how much nghttp3/ngtcp2 actually drains, -// rather than copying a huge caller buffer in one synchronous allocation+memcpy, mirroring -// nghttp2's own per-call copy into its frame buffer. -// -inline constexpr size_t kWriteChunkSize = 16 * 1024; - class Http3Stream : public std::enable_shared_from_this { public: @@ -333,35 +325,34 @@ class Http3Stream : public std::enable_shared_from_this // this is flat per-stream state rather than a queue of pending writes. See the client-side // counterpart (Http3ClientStream in client_impl_udp.cpp) for the fuller rationale. // - // write_source is the caller's buffer, referenced (not copied) the way asio::async_write - // generally requires -- it must stay valid until write_handler fires (and no longer: nghttp3 - // only ever gets pointers into write_chunk, our own copy, so once the handler has fired -- - // including via cancellation -- the caller's buffer is no longer touched). - // - // write_chunk is a bounded (<= kWriteChunkSize) slice of write_source, lazily refilled by - // data_reader() as it's drained. write_offered and write_confirmed are tracked separately - // because nghttp3 may call data_reader() several times in a row for the same stream before ever - // reporting consumption back via on_write_consumed() -- e.g. to gather more vecs than fit in a - // single call. If data_reader() just kept re-handing out write_chunk[0, write_chunk.size()) - // unconditionally (tracking only write_confirmed), nghttp3 would treat each repeat offer as - // *additional*, distinct stream bytes and duplicate the content on the wire. write_offered - // marks how much has already been handed to nghttp3 (whether or not it has been placed in a - // packet yet) so a repeat call sees nothing new and gets NGHTTP3_ERR_WOULDBLOCK instead. + // write_source is the caller's buffer, passed on to nghttp3 by reference: data_reader() points + // the nghttp3_vec straight into it, so the response body is never copied on its way down to the + // nghttp3/ngtcp2 boundary, however large it is -- the mmap()ed file of serve_file() travels + // from the page cache into QUIC packets without an intermediate byte. + // + // What that costs is *when* the write completes. ngtcp2 keeps pointing into this memory for as + // long as the bytes may still have to be retransmitted (it only ever copies the nghttp3_vec + // descriptors, never the payload), and running the write handler is what releases the caller's + // buffer -- so the handler has to wait for the data to be acknowledged. This is the model + // nghttp3 documents for its read_data callback: "the application must retain data until they + // are safe to free; it is notified by nghttp3_acked_stream_data". HTTP/2 completes a write as + // soon as nghttp2 has copied it into its own frame buffer, so a single async_write() there is + // done roughly a memcpy later, and here roughly a round trip later. + // + // write_offered tracks how much of write_source has been handed to nghttp3, which may ask + // again before any of it goes out and would take a repeated offer as *additional*, distinct + // stream bytes -- duplicating the body on the wire -- so a repeat call gets + // NGHTTP3_ERR_WOULDBLOCK instead. write_acked tracks what came back through nghttp3's + // acked_stream_data callback; the write is complete once that has caught up with write_source. // bool write_active = false; asio::const_buffer write_source; - size_t write_source_copied = 0; - std::vector write_chunk; size_t write_offered = 0; - size_t write_confirmed = 0; + size_t write_acked = 0; bool write_is_eof = false; WriteHandler write_handler; uint64_t write_token = 0; uint64_t next_write_token = 1; - - std::vector> in_flight_writes; // kept alive for the stream's lifetime -- - // ngtcp2 may still need this memory for - // retransmission until acked bool eof_submitted = false; // user signalled EOF via empty write bool eof_sent_to_h3 = false; // NGHTTP3_DATA_FLAG_EOF returned @@ -385,11 +376,11 @@ class Http3Stream : public std::enable_shared_from_this void submit_response(); void start_write(WriteHandler&& handler, asio::const_buffer buffer); nghttp3_ssize data_reader(nghttp3_vec* vec, size_t veccnt, uint32_t* pflags); - void on_write_consumed(size_t n); + void on_write_acked(size_t n); private: void bind_write_cancellation(WriteHandler& handler, uint64_t token); // arms cancellation - void finish_active_write(); // completes the active write once fully handed to nghttp3 + void finish_active_write(); // completes the active write and releases the caller's buffer public: // Called from either reader or writer destructor. @@ -527,6 +518,8 @@ class Http3Session : public Session::Impl // // nghttp3 callback bridges // + static int h3_cb_acked_stream_data(nghttp3_conn*, int64_t stream_id, uint64_t datalen, + void* user, void*); static int h3_cb_stream_close(nghttp3_conn*, int64_t stream_id, uint64_t app_error_code, void* user, void*); static int h3_cb_recv_data(nghttp3_conn*, int64_t stream_id, const uint8_t* data, size_t datalen, @@ -928,10 +921,8 @@ void Http3Stream::start_write(WriteHandler&& handler, asio::const_buffer buffer) write_active = true; write_source = buffer; // referenced, not copied -- see class comment above write_active - write_source_copied = 0; - write_chunk.clear(); write_offered = 0; - write_confirmed = 0; + write_acked = 0; write_is_eof = is_eof; write_token = token; write_handler = std::move(handler); @@ -954,11 +945,8 @@ void Http3Stream::bind_write_cancellation(WriteHandler& handler, uint64_t token) cs.assign([this, token](asio::cancellation_type_t ct) { // - // Cancellation completes the write immediately: nghttp3/ngtcp2 only ever hold pointers into - // write_chunk (our own copy), never into the caller's buffer, so the un-copied remainder of - // write_source can simply be abandoned. Bytes already offered to nghttp3 still go out (they - // can't be un-offered), so write_chunk is retired to in_flight_writes to keep that memory - // alive. The caller may issue a fresh async_write() as soon as the handler fires. + // Cancellation completes the write immediately, without waiting for the acknowledgements + // it would normally complete on -- see below for what that costs. // if (write_token != token || !write_handler) return; // already completed naturally before the cancellation was delivered @@ -979,10 +967,28 @@ void Http3Stream::bind_write_cancellation(WriteHandler& handler, uint64_t token) return; } logd("[{}] async_write: \x1b[1;31mcancelled\x1b[0m ({})", log_prefix, ct); - if (!write_chunk.empty()) - in_flight_writes.emplace_back(std::move(write_chunk)); - write_chunk.clear(); // moved-from + + // + // The handler runs now, and the caller is free to destroy its buffer the moment it does -- + // but nghttp3/ngtcp2 point straight into that buffer (see the class comment above + // write_active), so whatever was offered and is not acknowledged yet has to stop being + // referenced first. Only a reset can guarantee that: RESET_STREAM makes ngtcp2 drop the + // stream's queued data and keeps it from reclaiming in-flight bytes for retransmission. + // That costs nothing in expressiveness -- a body cut short mid-write is truncated, which + // is exactly what delete_writer() resets the stream for as well. + // + // A write that never got to offer a byte, or whose bytes are all acknowledged already, + // leaves nothing behind and lets the stream carry on unharmed. + // + if (write_offered > write_acked && !closed) + { + logw("[{}] async_write: cancelled with {} bytes unacknowledged, resetting stream", + log_prefix, write_offered - write_acked); + session.reset_stream(id, NGHTTP3_H3_REQUEST_CANCELLED); + closed = true; + } write_active = false; + write_source = {}; // make sure to post this -- otherwise "MAIN COROUTINE DID NOT COMPLETE" happens asio::post(get_executor(), [handler = std::move(write_handler)]() mutable { std::move(handler)(errc::make_error_code(errc::operation_canceled)); }); @@ -997,97 +1003,62 @@ nghttp3_ssize Http3Stream::data_reader(nghttp3_vec* vec, size_t veccnt, uint32_t if (!write_active) return NGHTTP3_ERR_WOULDBLOCK; - if (write_offered < write_chunk.size()) - { - vec[0].base = write_chunk.data() + write_offered; - vec[0].len = write_chunk.size() - write_offered; - write_offered = write_chunk.size(); // don't re-offer these bytes on a repeat call -- see - // class comment above write_active - return 1; - } - - // - // Current chunk fully offered. If it hasn't been confirmed yet (on_write_consumed()), there's - // nothing new until that happens -- see the class comment above write_active on why we can't - // just carve off the next slice of write_source early. - // - if (write_confirmed < write_chunk.size()) - return NGHTTP3_ERR_WOULDBLOCK; - // - // The current chunk is fully drained; retire it (ngtcp2 may still need this exact memory for - // retransmission until acked) and pull the next bounded slice out of write_source, if any. + // Hand out what is left of the caller's buffer, by reference and in one go: a nghttp3_vec is + // just a pointer and a length, so there is nothing to be gained from slicing it up, and + // nghttp3 gets to frame the whole thing as a single DATA frame. It picks up the rest by + // itself as packets are filled -- see write_pkt(), which keeps feeding the same vec to + // ngtcp2_conn_writev_stream() and advances nghttp3 by whatever went into the packet. // - if (!write_chunk.empty()) - in_flight_writes.emplace_back(std::move(write_chunk)); - - const size_t remaining = asio::buffer_size(write_source) - write_source_copied; - if (remaining > 0) - { - const size_t take = std::min(remaining, kWriteChunkSize); - auto* src = static_cast(write_source.data()) + write_source_copied; - write_chunk.assign(src, src + take); - write_source_copied += take; - write_offered = write_chunk.size(); - write_confirmed = 0; - vec[0].base = write_chunk.data(); - vec[0].len = write_chunk.size(); + const size_t total = asio::buffer_size(write_source); + if (write_offered < total) + { + auto* base = static_cast(write_source.data()) + write_offered; + vec[0].base = const_cast(base); // nghttp3 reads through this, never writes + vec[0].len = total - write_offered; + write_offered = total; // don't offer these bytes twice -- see class comment above + // write_active return 1; } // - // Nothing left in write_source either. If this is the EOF marker (write_source is always - // empty), retire it now -- a FIN carries no stream bytes, so there is nothing for - // on_write_consumed() to report back. A non-EOF write with nothing left to offer is instead - // retired from on_write_consumed() once its last chunk is confirmed (see there). + // Everything has been offered. For a body write there is nothing new until the caller starts + // the next one (which resumes the stream), so block here rather than returning 0 bytes -- + // returning 0 without NGHTTP3_DATA_FLAG_EOF would tell nghttp3 the body ended. // if (!write_is_eof) return NGHTTP3_ERR_WOULDBLOCK; + // + // The EOF marker (write_source is always empty for it) completes as soon as nghttp3 has taken + // the FIN: unlike body data, a FIN carries no memory of the caller's that we would have to + // keep alive until it is acknowledged. + // *pflags |= NGHTTP3_DATA_FLAG_EOF; eof_sent_to_h3 = true; finish_active_write(); return 0; } -void Http3Stream::on_write_consumed(size_t n) +void Http3Stream::on_write_acked(size_t n) { // - // n is the number of bytes of *stream* data ngtcp2 just committed to a packet, which also - // includes the HTTP/3 HEADERS frame nghttp3 sends ahead of any body -- e.g. the very first - // write_pkt() call after submit_response() drains the headers before there is an active write - // yet. Only attribute bytes once there is an active, non-EOF write to charge them against; - // clamp defensively in case a single packet still straddles the header/body boundary. + // n counts *application* data acknowledged on this stream -- nghttp3 accounts for the HTTP/3 + // framing it puts around the body itself, so unlike ngtcp2's stream offsets these bytes are + // exactly the ones the caller handed us. All of them belong to the write currently active: a + // write only completes once every byte it offered is acknowledged, so nothing can still be + // outstanding from an earlier one. Clamp defensively anyway -- an accounting mismatch should + // complete the write early, not run write_acked past the end of the buffer. // if (n == 0 || !write_active || write_is_eof) return; - n = std::min(n, write_chunk.size() - write_confirmed); - write_confirmed += n; - - if (write_confirmed < write_chunk.size()) - return; + write_acked = std::min(write_acked + n, asio::buffer_size(write_source)); + logd("[{}] on_write_acked: {} bytes, {}/{} acknowledged", log_prefix, n, write_acked, + asio::buffer_size(write_source)); - // The write is fully done once its current chunk is confirmed and there is no more of - // write_source left to carve into further chunks -- data_reader() advances write_chunk/ - // write_source_copied otherwise, so this is the terminal state. - if (write_source_copied == asio::buffer_size(write_source)) - { + if (write_acked == asio::buffer_size(write_source)) finish_active_write(); - return; - } - - // - // There is more of write_source to carve into chunks, but nghttp3 may have asked for data - // while this chunk was offered and still unconfirmed, in which case data_reader() answered - // NGHTTP3_ERR_WOULDBLOCK -- and a blocked stream is never polled again until it is explicitly - // resumed. Now that the chunk is confirmed, there is something new to hand out, so unblock - // the stream. Without this, any single async_write() larger than kWriteChunkSize stalls here - // forever, with the response body truncated and no FIN. - // - if (auto h3 = session.h3()) - nghttp3_conn_resume_stream(h3, id); - session.wake_write(); } void Http3Stream::finish_active_write() @@ -1095,13 +1066,11 @@ void Http3Stream::finish_active_write() assert(write_active); // - // ngtcp2 may still need this memory for retransmission until the bytes are acked; rather than - // tracking acks precisely, keep every chunk alive for the life of the stream (in_flight_writes - // is freed on stream destruction). + // Invoking the handler hands the caller's buffer back to it, so this must only ever run when + // nothing points into it any more: every offered byte acknowledged (on_write_acked()), or no + // bytes offered at all (the EOF marker). // - if (!write_chunk.empty()) - in_flight_writes.emplace_back(std::move(write_chunk)); - write_chunk.clear(); // moved-from + write_source = {}; auto handler = std::move(write_handler); write_active = false; @@ -1562,8 +1531,6 @@ ngtcp2_ssize Http3Session::write_pkt(ngtcp2_path* path, ngtcp2_pkt_info* pi, uin loge("[{}] nghttp3_conn_add_write_offset: {}", log_prefix_, nghttp3_strerror(rv)); return NGTCP2_ERR_CALLBACK_FAILURE; } - if (auto s = find_stream(stream_id)) - s->on_write_consumed(static_cast(ndatalen)); } continue; default: @@ -1582,8 +1549,6 @@ ngtcp2_ssize Http3Session::write_pkt(ngtcp2_path* path, ngtcp2_pkt_info* pi, uin loge("[{}] nghttp3_conn_add_write_offset: {}", log_prefix_, nghttp3_strerror(rv)); return NGTCP2_ERR_CALLBACK_FAILURE; } - if (auto s = find_stream(stream_id)) - s->on_write_consumed(static_cast(ndatalen)); } return nwrite; @@ -1927,6 +1892,7 @@ int Http3Session::setup_http3() return 0; nghttp3_callbacks h3cb{}; + h3cb.acked_stream_data = &Http3Session::h3_cb_acked_stream_data; h3cb.stream_close = &Http3Session::h3_cb_stream_close; h3cb.recv_data = &Http3Session::h3_cb_recv_data; h3cb.deferred_consume = &Http3Session::h3_cb_deferred_consume; @@ -1987,6 +1953,29 @@ int Http3Session::setup_http3() // nghttp3 callbacks // ------------------------------------------------------------------------------------------------- +// +// The only notification that the peer is done with response body bytes we handed out by +// reference, and hence that the caller's buffer may be released -- see the comment above +// Http3Stream::write_active. +// +int Http3Session::h3_cb_acked_stream_data(nghttp3_conn*, int64_t stream_id, uint64_t datalen, + void* user, void*) +{ + auto self = static_cast(user); + auto stream = self->find_stream(stream_id); + if (!stream) + return 0; + + // + // Completing a write resumes the application, which may drop the last reference to this + // session -- while ngtcp2 is still in the middle of processing the ACK that got us here. + // weak_from_this(), not shared_from_this(): the ACK may well arrive during teardown. + // + auto session_guard = self->weak_from_this().lock(); + stream->on_write_acked(static_cast(datalen)); + return 0; +} + int Http3Session::h3_cb_stream_close(nghttp3_conn*, int64_t stream_id, uint64_t /*app_error*/, void* user, void*) { @@ -1998,6 +1987,19 @@ int Http3Session::h3_cb_stream_close(nghttp3_conn*, int64_t stream_id, uint64_t // Waiting readers/writers should see the close now. if (s->read_handler) swap_and_invoke(s->read_handler, boost::system::error_code{}, 0); + + // + // A write waiting for its data to be acknowledged will never see those acknowledgements + // now: ngtcp2 drops whatever of this stream is still in flight. It also stops touching the + // caller's buffer, which is all the wait was ever for, so complete the write -- as failed, + // because the body did not make it -- instead of leaving it pending forever. + // + if (s->write_active && s->write_handler) + { + s->write_active = false; + s->write_source = {}; + swap_and_invoke(s->write_handler, errc::make_error_code(errc::connection_reset)); + } s->maybe_close(); } if (ngtcp2_conn_is_server(self->conn_)) diff --git a/test/test_server.cpp b/test/test_server.cpp index 228172d..15c07a9 100644 --- a/test/test_server.cpp +++ b/test/test_server.cpp @@ -963,9 +963,9 @@ TEST_P(ClientAsync, HelloWorld) // // A single async_write() larger than what the transport hands to its peer in one go, i.e. the -// whole body in one call instead of chunk by chunk. HTTP/3 used to stall here: with its response -// chunk fully offered but not yet confirmed, nghttp3 got NGHTTP3_ERR_WOULDBLOCK and blocked the -// stream, which nothing resumed once the chunk was confirmed. +// whole body in one call instead of chunk by chunk. HTTP/3 has to keep offering the same buffer +// to nghttp3 across many packets here, and complete the write only once all of it is +// acknowledged. // TEST_P(ClientAsync, WHEN_server_writes_large_buffer_at_once_THEN_receives_all) { @@ -994,6 +994,54 @@ TEST_P(ClientAsync, WHEN_server_writes_large_buffer_at_once_THEN_receives_all) }; } +// +// Cancelling a response write mid-body. HTTP/3 hands the caller's buffer to nghttp3 by reference, +// so bytes already offered and not yet acknowledged cannot simply be abandoned -- the stream is +// reset instead, which is what the peer would see anyway for a body that stops short of its end. +// +TEST_P(ClientAsync, WHEN_server_cancels_write_THEN_client_sees_truncated_body) +{ + static const std::vector body(8 * 1024 * 1024, 'x'); + + custom = [this](server::Request request, server::Response response) -> awaitable + { + std::array buffer; + while (co_await request.async_read_some(asio::buffer(buffer)) > 0) + ; // drain the request -- HTTP/1.1 closes the connection on an unfinished parser + + co_await response.async_submit(200, {}); + + // + // Far more than the peer's receive window, and the client below doesn't read a byte until + // this is over, so the write is guaranteed to still be in progress when it is cancelled. + // + auto executor = co_await this_coro::executor; + auto [ep] = co_await co_spawn(executor, send(response, std::span(body)), + cancel_after(50ms, as_tuple)); + EXPECT_EQ(code(ep), boost::system::errc::operation_canceled); + }; + test = [this](Session session) -> awaitable + { + auto request = co_await session.async_submit(url); + co_await request.async_write({}); + auto response = co_await request.async_get_response(); + + // + // Leave the response body untouched for now: whatever the server manages to send fills up + // the receive window and stays there, so its write cannot run to completion before the + // cancellation above hits. + // + asio::steady_timer timer(co_await this_coro::executor, 150ms); + co_await timer.async_wait(deferred); + + boost::system::error_code ec; + auto received = co_await try_receive(response, ec); + std::println("received {} of {} bytes ({})", received, body.size(), ec.message()); + EXPECT_LT(received, body.size()); + EXPECT_EQ(ec, boost::beast::http::error::partial_message); + }; +} + // ================================================================================================= // @@ -1114,7 +1162,8 @@ TEST_P(FileHandler, WHEN_file_is_empty_THEN_serves_empty_body) } // -// Larger than the 16K chunk the HTTP/3 write path carves a single async_write() into. +// Large enough that a single async_write() spans many QUIC packets, so the HTTP/3 write path has +// to keep handing out the caller's buffer across several write_pkt() rounds. // TEST_P(FileHandler, WHEN_file_is_large_THEN_serves_all_of_it) { From bf34a2c9f3b7dc3bbac9b81c76d50a0cfe476601 Mon Sep 17 00:00:00 2001 From: Peter Eisenlohr Date: Thu, 20 Aug 2026 20:27:40 +0000 Subject: [PATCH 08/20] QUIC: prefer AES-128-GCM over AES-256-GCM Neither side set a TLS 1.3 ciphersuite list, so OpenSSL's default order applied and TLS_AES_256_GCM_SHA384 won -- decisively so on the server, which also sets SSL_OP_CIPHER_SERVER_PREFERENCE. Use the same order as ngtcp2's example server, which puts TLS_AES_128_GCM_SHA256 first. The extra rounds of AES-256 buy nothing here, but on hardware with AES-NI this is worth only a few percent of bulk throughput, not the factor it looks like. Co-Authored-By: Claude Opus 5 --- src/client_impl_udp.cpp | 9 +++++++++ src/server_impl_udp.cpp | 9 +++++++++ 2 files changed, 18 insertions(+) diff --git a/src/client_impl_udp.cpp b/src/client_impl_udp.cpp index 16217ad..a0f99a4 100644 --- a/src/client_impl_udp.cpp +++ b/src/client_impl_udp.cpp @@ -95,6 +95,15 @@ struct TlsClientContext static constexpr unsigned char alpn[] = "\x02h3"; SSL_CTX_set_alpn_protos(ctx, alpn, sizeof(alpn) - 1); + // + // Same order as the server, so AES-128 GCM is also picked against peers that leave the + // choice to the client. + // + if (SSL_CTX_set_ciphersuites(ctx, "TLS_AES_128_GCM_SHA256:TLS_AES_256_GCM_SHA384:" + "TLS_CHACHA20_POLY1305_SHA256") != 1) + throw std::runtime_error(std::string{"SSL_CTX_set_ciphersuites: "} + + ERR_error_string(ERR_get_error(), nullptr)); + // // TODO: verify the server certificate (e.g. against pki/out/root.pem) instead of accepting // anything. diff --git a/src/server_impl_udp.cpp b/src/server_impl_udp.cpp index 099c4b7..49f7d8d 100644 --- a/src/server_impl_udp.cpp +++ b/src/server_impl_udp.cpp @@ -110,6 +110,15 @@ struct TlsServerContext SSL_OP_NO_ANTI_REPLAY); SSL_CTX_set_mode(ctx, SSL_MODE_RELEASE_BUFFERS); + // + // Prefer AES-128 over AES-256 GCM, like ngtcp2's example server does. Combined with + // SSL_OP_CIPHER_SERVER_PREFERENCE above, this is what actually picks the bulk cipher. + // + if (SSL_CTX_set_ciphersuites(ctx, "TLS_AES_128_GCM_SHA256:TLS_AES_256_GCM_SHA384:" + "TLS_CHACHA20_POLY1305_SHA256") != 1) + throw std::runtime_error(std::string{"SSL_CTX_set_ciphersuites: "} + + ERR_error_string(ERR_get_error(), nullptr)); + SSL_CTX_set_alpn_select_cb(ctx, &TlsServerContext::alpn_select_cb, nullptr); if (SSL_CTX_use_PrivateKey_file(ctx, "pki/out/server-key.pem", SSL_FILETYPE_PEM) != 1) From cb7f1f86b57931e2c00598d0933c93b3de315a24 Mon Sep 17 00:00:00 2001 From: Peter Eisenlohr Date: Thu, 20 Aug 2026 20:28:02 +0000 Subject: [PATCH 09/20] QUIC: only install ngtcp2's log callback when tracing ngtcp2_log_printf() checked the log level and returned early, but by then the work was already done: ngtcp2 formats every frame of every packet into a string before invoking the callback, and only skips that when log_printf is NULL. So each packet paid for formatting that was then discarded. Install the callback only when trace logging is actually enabled. Measured with callgrind on a 64 KB file benchmark, this drops the server from 344M to 269M instructions (-22%), with ngtcp2_fmt_write_str, ngtcp2_encode_uint, log_fr and strlen leaving the profile entirely. The level is now sampled when the connection is created rather than per call, so raising it at runtime does not affect connections that already exist. Co-Authored-By: Claude Opus 5 --- src/client_impl_udp.cpp | 8 +++++++- src/server_impl_udp.cpp | 9 ++++++++- 2 files changed, 15 insertions(+), 2 deletions(-) diff --git a/src/client_impl_udp.cpp b/src/client_impl_udp.cpp index a0f99a4..031e1c9 100644 --- a/src/client_impl_udp.cpp +++ b/src/client_impl_udp.cpp @@ -1181,7 +1181,13 @@ int Http3ClientSession::init(asio::ip::udp::endpoint remote) ngtcp2_settings settings; ngtcp2_settings_default(&settings); settings.initial_ts = ngtcp2::util::timestamp(); - settings.log_printf = &ngtcp2_log_printf; + + // + // See the server-side counterpart: ngtcp2 does the full frame formatting before calling this, + // so only install it when trace logging is actually enabled. + // + if (spdlog::default_logger_raw()->should_log(spdlog::level::trace)) + settings.log_printf = &ngtcp2_log_printf; ngtcp2_transport_params params; ngtcp2_transport_params_default(¶ms); diff --git a/src/server_impl_udp.cpp b/src/server_impl_udp.cpp index 49f7d8d..13df32a 100644 --- a/src/server_impl_udp.cpp +++ b/src/server_impl_udp.cpp @@ -1349,7 +1349,14 @@ int Http3Session::init(const ngtcp2_cid& dcid, const ngtcp2_cid& scid, uint32_t ngtcp2_settings settings; ngtcp2_settings_default(&settings); settings.initial_ts = ngtcp2::util::timestamp(); - settings.log_printf = &ngtcp2_log_printf; + + // + // Only install the log callback when trace logging is actually enabled: ngtcp2 formats every + // frame of every packet into a string *before* invoking it, so a callback that discards its + // input still pays for the full formatting. A NULL log_printf makes ngtcp2 skip that work. + // + if (spdlog::default_logger_raw()->should_log(spdlog::level::trace)) + settings.log_printf = &ngtcp2_log_printf; ngtcp2_transport_params params; ngtcp2_transport_params_default(¶ms); From ded78ee8cb33c79439a85855d738d6e2e7bf0737 Mon Sep 17 00:00:00 2001 From: Peter Eisenlohr Date: Thu, 20 Aug 2026 20:28:17 +0000 Subject: [PATCH 10/20] serve_file: keep file mappings across requests serve_file() resolved the path and mmap()ed the file on every request, then unmapped it on the way out. For a file already in the page cache that was the single largest per-request cost -- larger than resolving the path, and larger than the QUIC send path itself: ~17 syscalls per request, of which the map and unmap pair dominated. Cache the mapping, keyed by request path, together with the response headers derived from it. A hit still stat()s the file and compares device, inode, size and mtime against what was mapped, so a file replaced or modified on disk is picked up on the next request -- one syscall instead of seventeen. MappedFile records that identity from the fstat() on the fd it mapped, leaving no window between checking and mapping. The cache is bounded (256 entries / 64 MiB, least recently used evicted first, keeping at least the entry just inserted) and guarded by a mutex, with the mapping itself built outside the lock so concurrent misses do not serialise. serve_file() holds a shared_ptr, so an in-flight response keeps its mapping valid even if the entry is evicted or replaced mid-write. A hit deliberately does not re-run resolve(), so re-pointing a symlink along the path is only noticed once the file it pointed at changes. That direction is safe: a stale entry can only keep serving a file that already passed the containment check, never a newly escaping one. Measured on `h2load --h3 -n 10000 -c 4 -m 3` against a 64 KB file: 7.9k -> 15.5k requests/s, with system time down from 350ms to 130ms per 10k requests. Co-Authored-By: Claude Opus 5 --- src/file_handler.cpp | 187 +++++++++++++++++++++++++++++++++++++------ 1 file changed, 164 insertions(+), 23 deletions(-) diff --git a/src/file_handler.cpp b/src/file_handler.cpp index a61ba9d..26562f0 100644 --- a/src/file_handler.cpp +++ b/src/file_handler.cpp @@ -7,7 +7,11 @@ #include #include +#include +#include +#include #include +#include using namespace std::string_view_literals; using namespace anyhttp; @@ -32,7 +36,7 @@ class MappedFile MappedFile() = default; MappedFile(MappedFile&& other) noexcept : m_data(std::exchange(other.m_data, nullptr)), m_size(std::exchange(other.m_size, 0)), - m_mtime(other.m_mtime) + m_mtime(other.m_mtime), m_id(other.m_id) { } MappedFile& operator=(MappedFile&& other) noexcept @@ -40,6 +44,7 @@ class MappedFile std::swap(m_data, other.m_data); std::swap(m_size, other.m_size); std::swap(m_mtime, other.m_mtime); + std::swap(m_id, other.m_id); return *this; } ~MappedFile() @@ -69,6 +74,7 @@ class MappedFile MappedFile file; file.m_size = static_cast(st.st_size); file.m_mtime = std::chrono::system_clock::from_time_t(st.st_mtime); + file.m_id = Identity{st}; if (file.m_size == 0) return file; @@ -81,8 +87,32 @@ class MappedFile return file; } + // + // What the mapping was made from, so a cached mapping can be checked against the file that is + // on disk now. st_ino/st_dev catch a replaced file (the usual atomic rename), the rest catches + // a file modified in place. + // + struct Identity + { + dev_t dev; + ino_t ino; + off_t size; + decltype(std::declval().st_mtim) mtim; + + explicit Identity(const struct stat& st = {}) + : dev(st.st_dev), ino(st.st_ino), size(st.st_size), mtim(st.st_mtim) + { + } + bool operator==(const Identity& other) const noexcept + { + return dev == other.dev && ino == other.ino && size == other.size && + mtim.tv_sec == other.mtim.tv_sec && mtim.tv_nsec == other.mtim.tv_nsec; + } + }; + size_t size() const noexcept { return m_size; } auto mtime() const noexcept { return m_mtime; } + const Identity& identity() const noexcept { return m_id; } std::span bytes() const noexcept { return {static_cast(m_data), m_size}; @@ -92,6 +122,7 @@ class MappedFile void* m_data = nullptr; size_t m_size = 0; std::chrono::system_clock::time_point m_mtime; + Identity m_id{}; }; // @@ -169,6 +200,123 @@ awaitable respond(server::Response& response, unsigned status) co_await response.async_write({}); } +// +// Everything serving one request needs, computed once per file instead of once per request: the +// mapping plus the response headers derived from it. +// +struct CachedFile +{ + fs::path resolved; + MappedFile file; + std::string last_modified; + std::string_view content_type; +}; + +// +// Mapping a file per request costs an mmap()/munmap() pair plus the faults to populate the +// mapping, which measured as the single largest per-request cost when serving a file that is +// already in the page cache -- larger than resolving the path, and larger than the QUIC send +// path itself. Keeping the mapping alive across requests removes all of that. +// +// A hit still stat()s the file, so a file replaced or modified on disk is picked up on the next +// request; that is one syscall instead of the ~17 a full resolve-and-map takes. What a hit does +// *not* redo is resolve(), so re-pointing a symlink along the path is only noticed once the file +// it used to point at changes. That direction is safe -- a stale entry can only keep serving a +// file that already passed the containment check -- but it is why this is a cache of resolved +// paths and not a cache of open files. +// +class FileCache +{ +public: + // Bounded so that a large tree cannot pin unbounded address space; least recently used first. + static constexpr size_t max_entries = 256; + static constexpr size_t max_bytes = 64u << 20; + + expected> get(const std::string& request_path, + std::string_view prefix, const fs::path& root) + { + if (auto hit = lookup(request_path)) + { + struct stat st{}; + if (::stat(hit->resolved.c_str(), &st) == 0 && + MappedFile::Identity{st} == hit->file.identity()) + return hit; + } + + // + // Miss, or the file changed underneath us. Build the entry outside the lock: mapping is the + // expensive part and two requests racing on the same path may as well both do it. + // + const auto resolved = resolve(request_path, prefix, root); + if (!resolved) + return std::unexpected(resolved.error()); + + auto mapped = MappedFile::open(*resolved); + if (!mapped) + return std::unexpected(mapped.error()); + + // Read mtime() before the move, rather than relying on argument evaluation order. + auto last_modified = format_http_date(mapped->mtime()); + auto entry = std::make_shared(*resolved, std::move(*mapped), + std::move(last_modified), + content_type(*resolved)); + insert(request_path, entry); + return entry; + } + +private: + std::shared_ptr lookup(const std::string& key) + { + const std::lock_guard lock{m_mutex}; + const auto it = m_entries.find(key); + if (it == m_entries.end()) + return nullptr; + m_lru.splice(m_lru.begin(), m_lru, it->second.lru); // most recently used first + return it->second.entry; + } + + void insert(const std::string& key, std::shared_ptr entry) + { + const std::lock_guard lock{m_mutex}; + + if (const auto it = m_entries.find(key); it != m_entries.end()) + { + m_bytes -= it->second.entry->file.size(); + m_lru.erase(it->second.lru); + m_entries.erase(it); + } + + m_bytes += entry->file.size(); + m_lru.push_front(key); + m_entries.emplace(key, Slot{std::move(entry), m_lru.begin()}); + + // + // Keep at least the entry just inserted, so a file larger than the byte budget still gets + // served from the cache rather than being evicted immediately every time. + // + while (m_entries.size() > 1 && (m_entries.size() > max_entries || m_bytes > max_bytes)) + { + const auto victim = m_entries.find(m_lru.back()); + m_bytes -= victim->second.entry->file.size(); + m_entries.erase(victim); + m_lru.pop_back(); + } + } + + struct Slot + { + std::shared_ptr entry; + std::list::iterator lru; + }; + + std::mutex m_mutex; + std::unordered_map m_entries; + std::list m_lru; + size_t m_bytes = 0; +}; + +FileCache g_cache; + } // namespace namespace anyhttp @@ -187,35 +335,28 @@ awaitable serve_file(server::Request request, server::Response response, f ; const std::string path = request.url().path(); - const auto resolved = resolve(path, prefix, root); - if (!resolved) - { - logw("serve_file: {}: {}", path, resolved.error().message()); - co_await respond(response, status_for(resolved.error())); - co_return; - } - - const auto file = MappedFile::open(*resolved); - if (!file) + const auto entry = g_cache.get(path, prefix, root); + if (!entry) { - logw("serve_file: {}: {}", resolved->native(), file.error().message()); - co_await respond(response, status_for(file.error())); + logw("serve_file: {}: {}", path, entry.error().message()); + co_await respond(response, status_for(entry.error())); co_return; } - logd("serve_file: {} ({} bytes)", resolved->native(), file->size()); - co_await response.async_submit(200, - fields({{"Content-Length", file->size()}, - {"Content-Type", content_type(*resolved)}, - {"Last-Modified", format_http_date(file->mtime())}})); + const auto& file = (*entry)->file; + logd("serve_file: {} ({} bytes)", (*entry)->resolved.native(), file.size()); + co_await response.async_submit(200, fields({{"Content-Length", file.size()}, + {"Content-Type", (*entry)->content_type}, + {"Last-Modified", (*entry)->last_modified}})); // - // The mapping lives in the coroutine frame, so it stays valid across the write and is - // unmapped no matter how we leave -- normally, by exception or by cancellation. Note that - // touching a mapped page may block on disk I/O, which no amount of chunking would avoid. + // The shared_ptr lives in the coroutine frame, so the mapping stays valid across the write no + // matter how we leave -- normally, by exception or by cancellation -- even if the entry is + // evicted or replaced meanwhile. Note that touching a mapped page may block on disk I/O, + // which no amount of chunking would avoid. // - if (file->size() > 0) - co_await send(response, file->bytes()); // an empty write already means EOF, don't send two + if (file.size() > 0) + co_await send(response, file.bytes()); // an empty write already means EOF, don't send two co_await response.async_write({}); } From 62021e96121ceecd0792fc38817a76e2e52c2b28 Mon Sep 17 00:00:00 2001 From: Peter Eisenlohr Date: Thu, 20 Aug 2026 20:29:21 +0000 Subject: [PATCH 11/20] HTTP/3: write once per receive batch, not once per datagram on_read() ran the whole write path -- aggregate, flush, re-arm the timer -- for every datagram, and udp_on_read() feeds it up to 32 of them per wakeup (more once a GRO-coalesced read is split). Each pass therefore saw only what ngtcp2 happened to have queued at that instant, so a response went out as several small GSO batches instead of one big one. Mark the session in on_read() instead and let udp_on_read() write once per session, after every datagram the socket had queued has been handed to ngtcp2. This is what ngtcp2's example server does with signal_write() and its writable watcher. handle_expiry() and wake_write() go through the same flush_write(), and the EAGAIN return became a break so a drained socket still reaches the write. Per request this removes all of the single-packet sendto() calls (1.45 -> 0), and takes sendmsg() from 1.96 to 1.67 and recvmsg() from 3.93 to 2.08 -- 9.85 syscalls per request down to 7.0. It does not move throughput: at this point the server is bound by user-space CPU, not syscalls, and the ~3us per request this saves does not show up against a ~57us budget. Committed because it is strictly less work for the same result. Co-Authored-By: Claude Opus 5 --- src/server_impl_udp.cpp | 82 +++++++++++++++++++++++++++++++++++------ 1 file changed, 70 insertions(+), 12 deletions(-) diff --git a/src/server_impl_udp.cpp b/src/server_impl_udp.cpp index 13df32a..cc83bd2 100644 --- a/src/server_impl_udp.cpp +++ b/src/server_impl_udp.cpp @@ -56,6 +56,7 @@ #include #include #include +#include #include #include "ngtcp2/shared.h" @@ -424,6 +425,16 @@ class Http3Session : public Session::Impl int on_read(const ngtcp2_pkt_info& pi, std::span data, const ngtcp2::Address& remote); int write_streams(); + + // + // on_read() does not write; it only marks the session here, and Server::Impl calls + // flush_write() once the whole receive batch has been fed to ngtcp2. Writing per datagram + // means every aggregate pass only sees what happened to be queued at that instant, so a + // response goes out as several small GSO batches instead of one big one. + // + void defer_write() noexcept { write_pending_ = true; } + int flush_write(); + ngtcp2_ssize write_pkt(ngtcp2_path* path, ngtcp2_pkt_info* pi, uint8_t* dest, size_t destlen, ngtcp2_tstamp ts); void update_timer(); @@ -572,6 +583,8 @@ class Http3Session : public Session::Impl std::string log_prefix_; + bool write_pending_ = false; // set by on_read(), acted on by flush_write() + std::vector conn_closebuf_; // buffered CONNECTION_CLOSE packet // Aggregated TX buffer: ngtcp2_conn_write_aggregate_pkt2() packs as many same-sized @@ -1304,8 +1317,7 @@ void Http3Session::wake_write() auto session = std::static_pointer_cast(self.lock()); if (!session || session->closed_) return; - if (session->write_streams() == 0) - session->update_timer(); + session->flush_write(); }); } @@ -1447,10 +1459,10 @@ int Http3Session::on_read(const ngtcp2_pkt_info& pi, std::span da return handle_error(rv); } - if (auto wrv = write_streams(); wrv != 0) - return wrv; - - update_timer(); + // + // Deliberately no write here -- see defer_write(). + // + defer_write(); return 0; } @@ -1608,6 +1620,22 @@ int Http3Session::write_streams() // ------------------------------------------------------------------------------------------------- +int Http3Session::flush_write() +{ + write_pending_ = false; + + if (closed_ || !conn_) + return 0; + + if (auto rv = write_streams(); rv != 0) + return rv; + + update_timer(); + return 0; +} + +// ------------------------------------------------------------------------------------------------- + void Http3Session::update_timer() { arm_timer_from_ngtcp2(); } void Http3Session::arm_timer_from_ngtcp2() @@ -1648,10 +1676,7 @@ int Http3Session::handle_expiry() ngtcp2_ccerr_set_liberr(&last_error_, rv, nullptr, 0); return handle_error(rv); } - if (auto rv = write_streams(); rv != 0) - return rv; - update_timer(); - return 0; + return flush_write(); } // ------------------------------------------------------------------------------------------------- @@ -2212,6 +2237,18 @@ int Server::Impl::udp_on_read(Endpoint& ep) msg_ctrl[CMSG_SPACE(sizeof(int)) + CMSG_SPACE(sizeof(in6_pktinfo)) + CMSG_SPACE(sizeof(int))]; msg.msg_control = msg_ctrl; + // + // Sessions that received something in this batch. They are written once, below, after every + // datagram the socket had queued has been handed to ngtcp2 -- so one aggregate pass can pack + // a whole response into a single GSO sendmsg() instead of dribbling it out per datagram. + // + std::vector> pending; + auto mark_pending = [&pending](const std::shared_ptr& session) + { + if (std::ranges::find(pending, session) == pending.end()) + pending.push_back(session); + }; + for (size_t pktcnt = 0; pktcnt < 32; ++pktcnt) { if (pktcnt) @@ -2225,7 +2262,7 @@ int Server::Impl::udp_on_read(Endpoint& ep) { if (errno != EAGAIN && errno != EWOULDBLOCK && errno != ENOTCONN) loge("recvmsg: {}", strerror(errno)); - return 0; + break; // socket drained (or broken): fall through to the write pass } if (nread < 22) @@ -2318,6 +2355,8 @@ int Server::Impl::udp_on_read(Endpoint& ep) auto lock = std::lock_guard(self->m_sessionMutex); self->m_sessions.erase(session); }); + + mark_pending(session); } else { @@ -2341,7 +2380,11 @@ int Server::Impl::udp_on_read(Endpoint& ep) continue; } - if (session->on_read(pi, data, *remote) != 0 && session->closed()) + if (session->on_read(pi, data, *remote) == 0) + { + mark_pending(session); + } + else if (session->closed()) { // // Only erase immediately when not in closing/draining period. @@ -2359,6 +2402,21 @@ int Server::Impl::udp_on_read(Endpoint& ep) } } } + + // + // One write pass per session, after the whole batch has been read. + // + for (const auto& session : pending) + { + if (session->flush_write() == 0 || !session->closed()) + continue; + + auto* conn = session->conn(); + if (!conn || (!ngtcp2_conn_in_closing_period(conn) && !ngtcp2_conn_in_draining_period(conn))) + std::erase_if(m_quic_handlers, + [&](const auto& kv) { return kv.second.get() == session.get(); }); + } + return 0; } From b0a4b68762ca9f761984ab95762d806bdc05c066 Mon Sep 17 00:00:00 2001 From: Peter Eisenlohr Date: Fri, 21 Aug 2026 06:30:18 +0000 Subject: [PATCH 12/20] HTTP/3: arm the write flush once per wake, not once per submission wake_write() posted a flush unconditionally, and a response submits its headers, its body and its EOF separately -- so three posts per response where one would do. The first pass wrote everything there was, and the other two walked the connection for nothing and re-armed the timer on the way out. Arm once and clear the flag when the flush runs, which is what ngtcp2's example server gets for free from ev_io_start() on an already-active watcher. Measured over 2000 requests this takes timerfd_settime from 3480 calls to 1075 (1.74 -> 0.54 per request). sendmsg and recvmsg are unchanged, confirming those extra passes never produced a packet. It is not a throughput win. Interleaved A/B against the parent commit, five runs each: 14408 vs 14459 req/s and 578ms vs 574ms of server CPU per 10k requests, both inside the run-to-run spread. It also draws ~23% more acknowledgements from the peer, for reasons not established. Committed because it is strictly less work per response, not because it made the benchmark faster. Co-Authored-By: Claude Opus 5 --- src/server_impl_udp.cpp | 17 ++++++++++++++++- 1 file changed, 16 insertions(+), 1 deletion(-) diff --git a/src/server_impl_udp.cpp b/src/server_impl_udp.cpp index cc83bd2..ebc739f 100644 --- a/src/server_impl_udp.cpp +++ b/src/server_impl_udp.cpp @@ -584,6 +584,7 @@ class Http3Session : public Session::Impl std::string log_prefix_; bool write_pending_ = false; // set by on_read(), acted on by flush_write() + bool write_posted_ = false; // a wake_write() flush is already on the way std::vector conn_closebuf_; // buffered CONNECTION_CLOSE packet @@ -1312,10 +1313,24 @@ void Http3Session::wake_write() // Reader/Writer destructor that runs as part of *this* session's own teardown (e.g. a // still-in-flight request/response destroyed by Server::Impl cancelling everything on // shutdown), at which point shared_from_this() would throw bad_weak_ptr. + // + // One flush per wake, not one per submission. A response submits its headers, its body and + // its EOF separately, and posting for each means the first pass writes everything and the + // rest walk the connection for nothing -- and still re-arm the timer on the way out. Arming + // once and clearing when the flush runs is what ngtcp2's example server gets for free from + // ev_io_start() on an already-active watcher. + // + if (write_posted_) + return; + write_posted_ = true; + asio::post(get_executor(), [self = weak_from_this()] { auto session = std::static_pointer_cast(self.lock()); - if (!session || session->closed_) + if (!session) + return; + session->write_posted_ = false; + if (session->closed_) return; session->flush_write(); }); From c4b45950a00bc4d87cba7aeaee1c1c970c7d8d40 Mon Sep 17 00:00:00 2001 From: Peter Eisenlohr Date: Fri, 21 Aug 2026 07:17:06 +0000 Subject: [PATCH 13/20] HTTP/3: fill a read from every queued chunk, not just the first async_read_some() copied out of pending_read.front() and stopped there, however much room the caller had left and however much was queued behind it. A chunk is what arrived in one QUIC packet, so a 64k request body was handed over in ~48 reads of ~1.4k each. For a handler that only consumes, that is 48 coroutine round trips instead of one or two -- wasteful but survivable. For a handler that answers each read with a write, it is fatal: a body write completes only once the peer acknowledges it (the response body is passed to nghttp3 by reference, so the caller's buffer cannot be released any earlier -- see the comment above write_active), so every one of those 48 iterations costs a full round trip. The server ends up with at most one packet of data to send at a time: strace shows one sendto() of 1444 bytes per received datagram, never a GSO burst, with the socket idle in between. h2load --h3 https://localhost:8080/echo -d test/data/64kminus1 -n 1000 -c 4 -m 3 before 306 / 308 / 272 / 451 / 382 req/s, 890-1130 ms server CPU / 1500 req after 7286 / 7379 / 6671 / 7292 / 7379 req/s, 130-160 ms server CPU / 1500 req Interleaved runs, same binary pair alternating. That is ~24x the throughput at ~7x less CPU per request: the lockstep was not just waiting, it was paying a wake, a flush, a timer re-arm and a one-packet send for every 1.4k of body. Reads per request drop from 48 to 3. /eat_request and the file handler are unchanged (13.0k and 15.5k req/s before and after): the former reads into a 1k buffer, so there is nothing to coalesce, and the latter has no request body at all. Co-Authored-By: Claude Opus 5 --- src/client_impl_udp.cpp | 29 ++++++++++++++++++++++------- src/server_impl_udp.cpp | 29 ++++++++++++++++++++++------- 2 files changed, 44 insertions(+), 14 deletions(-) diff --git a/src/client_impl_udp.cpp b/src/client_impl_udp.cpp index 031e1c9..dec254f 100644 --- a/src/client_impl_udp.cpp +++ b/src/client_impl_udp.cpp @@ -672,15 +672,30 @@ void Http3ClientStream::call_read_handler() { if (asio::buffer_size(read_head) > 0) { - auto copied = asio::buffer_copy(read_handler_buffer, read_head); - read_head += copied; - consumed += copied; - if (read_head.size() == 0) + // + // Fill the caller's buffer from as many queued chunks as it takes, rather than stopping + // at the end of the first one. Each chunk is what arrived in a single QUIC packet -- a + // little over a kilobyte -- so handing them out one per read turns a 64k body into ~48 + // reads, and a handler that answers every read with a write (an echo) pays a full + // round trip for each of them, because a body write only completes once the peer has + // acknowledged it (see the comment above write_active). + // + auto dest = read_handler_buffer; + size_t copied = 0; + while (dest.size() > 0 && read_head.size() > 0) { - pending_read.pop_front(); - read_head = - pending_read.empty() ? asio::const_buffer{} : asio::buffer(pending_read.front()); + auto n = asio::buffer_copy(dest, read_head); + dest += n; + read_head += n; + copied += n; + if (read_head.size() == 0) + { + pending_read.pop_front(); + read_head = + pending_read.empty() ? asio::const_buffer{} : asio::buffer(pending_read.front()); + } } + consumed += copied; swap_and_invoke(read_handler, boost::system::error_code{}, copied); continue; } diff --git a/src/server_impl_udp.cpp b/src/server_impl_udp.cpp index ebc739f..c68847a 100644 --- a/src/server_impl_udp.cpp +++ b/src/server_impl_udp.cpp @@ -796,15 +796,30 @@ void Http3Stream::call_read_handler() { if (asio::buffer_size(read_head) > 0) { - auto copied = asio::buffer_copy(read_handler_buffer, read_head); - read_head += copied; - consumed += copied; - if (read_head.size() == 0) + // + // Fill the caller's buffer from as many queued chunks as it takes, rather than stopping + // at the end of the first one. Each chunk is what arrived in a single QUIC packet -- a + // little over a kilobyte -- so handing them out one per read turns a 64k body into ~48 + // reads, and a handler that answers every read with a write (an echo) pays a full + // round trip for each of them, because a body write only completes once the peer has + // acknowledged it (see the comment above write_active). + // + auto dest = read_handler_buffer; + size_t copied = 0; + while (dest.size() > 0 && read_head.size() > 0) { - pending_read.pop_front(); - read_head = - pending_read.empty() ? asio::const_buffer{} : asio::buffer(pending_read.front()); + auto n = asio::buffer_copy(dest, read_head); + dest += n; + read_head += n; + copied += n; + if (read_head.size() == 0) + { + pending_read.pop_front(); + read_head = + pending_read.empty() ? asio::const_buffer{} : asio::buffer(pending_read.front()); + } } + consumed += copied; swap_and_invoke(read_handler, boost::system::error_code{}, copied); continue; } From ac2d8fe59929890c5ac401d94b0551c5dd9cd40d Mon Sep 17 00:00:00 2001 From: Peter Eisenlohr Date: Fri, 21 Aug 2026 07:17:20 +0000 Subject: [PATCH 14/20] HTTP/3: hand an arriving chunk to a waiting reader without parking it first nghttp3's recv_data callback points into the packet it is parsing, so on_data_chunk() copied the bytes into a vector of their own and pushed that onto pending_read -- a malloc, a copy in, a copy out and a free per QUIC packet, about fifty of each per 64k of request body. In the common case the copy out happened immediately afterwards, because a handler that keeps a read outstanding was already waiting for exactly those bytes. Offer the chunk to call_read_handler() where it lies instead, through a new `incoming` buffer that the fill loop drains after the queued chunks (which arrived earlier and must go first). Only what the reader could not take is parked. A callgrind profile of /eat_request showed the pending_read vectors accounting for roughly half of all allocations, 90 per request. h2load --h3 https://localhost:8080/eat_request -d test/data/64kminus1 -n ... Marginal instructions per request, differenced between n=100 and n=400 runs to cancel startup: before 575,236 after 543,105 (-5.6%) Not visible in wall clock at this operating point -- 4000 requests take ~220 ms of server CPU either way, and 5.6% of that is inside the 10 ms accounting granularity. It is strictly less work per packet, not a measured speedup. Co-Authored-By: Claude Opus 5 --- src/client_impl_udp.cpp | 40 ++++++++++++++++++++++++++++++++++++---- src/server_impl_udp.cpp | 40 ++++++++++++++++++++++++++++++++++++---- 2 files changed, 72 insertions(+), 8 deletions(-) diff --git a/src/client_impl_udp.cpp b/src/client_impl_udp.cpp index dec254f..3eccc67 100644 --- a/src/client_impl_udp.cpp +++ b/src/client_impl_udp.cpp @@ -241,6 +241,7 @@ class Http3ClientStream : public std::enable_shared_from_this // std::deque> pending_read; asio::const_buffer read_head; + asio::const_buffer incoming; // chunk on_data_chunk() is delivering, not yet taken bool eof_received = false; ReadSomeHandler read_handler; asio::mutable_buffer read_handler_buffer; @@ -632,10 +633,27 @@ void Http3ClientStream::on_data_chunk(const uint8_t* data, size_t len) { if (len == 0) return; - pending_read.emplace_back(data, data + len); - if (read_head.size() == 0) - read_head = asio::buffer(pending_read.front()); + + // + // nghttp3 hands us a view into the packet it is parsing, valid only until this callback + // returns. Offer it to a waiting reader as it stands before copying it anywhere: a handler + // that keeps a read outstanding -- the usual shape -- takes the bytes with a single copy, and + // the vector that would otherwise carry them (a malloc, a copy in, a copy out and a free, per + // QUIC packet, so about fifty of each per 64k of request body) is never created at all. Only + // what the reader could not take is parked for later. + // + auto self = shared_from_this(); // a resumed reader may drop the last reference to this stream + incoming = asio::const_buffer{data, len}; call_read_handler(); + + if (incoming.size() > 0) + { + auto* rest = static_cast(incoming.data()); + pending_read.emplace_back(rest, rest + incoming.size()); + if (read_head.size() == 0) + read_head = asio::buffer(pending_read.front()); + incoming = {}; + } } void Http3ClientStream::on_eof() @@ -670,7 +688,7 @@ void Http3ClientStream::call_read_handler() size_t consumed = 0; while (read_handler) { - if (asio::buffer_size(read_head) > 0) + if (read_head.size() > 0 || incoming.size() > 0) { // // Fill the caller's buffer from as many queued chunks as it takes, rather than stopping @@ -695,6 +713,19 @@ void Http3ClientStream::call_read_handler() pending_read.empty() ? asio::const_buffer{} : asio::buffer(pending_read.front()); } } + + // + // ... and last from the chunk being delivered right now, which on_data_chunk() offers + // through `incoming` instead of parking it in a vector of its own first. Queued chunks + // go first: they arrived earlier. + // + if (dest.size() > 0 && incoming.size() > 0) + { + auto n = asio::buffer_copy(dest, incoming); + incoming += n; + copied += n; + } + consumed += copied; swap_and_invoke(read_handler, boost::system::error_code{}, copied); continue; @@ -1043,6 +1074,7 @@ void Http3ClientStream::delete_reader() auto self = shared_from_this(); // see delete_writer() pending_read.clear(); read_head = {}; + incoming = {}; maybe_close(); } diff --git a/src/server_impl_udp.cpp b/src/server_impl_udp.cpp index c68847a..3d4555e 100644 --- a/src/server_impl_udp.cpp +++ b/src/server_impl_udp.cpp @@ -324,6 +324,7 @@ class Http3Stream : public std::enable_shared_from_this // std::deque> pending_read; asio::const_buffer read_head; // view of pending_read.front() not yet delivered + asio::const_buffer incoming; // chunk on_data_chunk() is delivering, not yet taken bool eof_received = false; ReadSomeHandler read_handler; asio::mutable_buffer read_handler_buffer; @@ -756,10 +757,27 @@ void Http3Stream::on_data_chunk(const uint8_t* data, size_t len) { if (len == 0) return; - pending_read.emplace_back(data, data + len); - if (read_head.size() == 0) - read_head = asio::buffer(pending_read.front()); + + // + // nghttp3 hands us a view into the packet it is parsing, valid only until this callback + // returns. Offer it to a waiting reader as it stands before copying it anywhere: a handler + // that keeps a read outstanding -- the usual shape -- takes the bytes with a single copy, and + // the vector that would otherwise carry them (a malloc, a copy in, a copy out and a free, per + // QUIC packet, so about fifty of each per 64k of request body) is never created at all. Only + // what the reader could not take is parked for later. + // + auto self = shared_from_this(); // a resumed reader may drop the last reference to this stream + incoming = asio::const_buffer{data, len}; call_read_handler(); + + if (incoming.size() > 0) + { + auto* rest = static_cast(incoming.data()); + pending_read.emplace_back(rest, rest + incoming.size()); + if (read_head.size() == 0) + read_head = asio::buffer(pending_read.front()); + incoming = {}; + } } void Http3Stream::on_eof() @@ -794,7 +812,7 @@ void Http3Stream::call_read_handler() size_t consumed = 0; while (read_handler) { - if (asio::buffer_size(read_head) > 0) + if (read_head.size() > 0 || incoming.size() > 0) { // // Fill the caller's buffer from as many queued chunks as it takes, rather than stopping @@ -819,6 +837,19 @@ void Http3Stream::call_read_handler() pending_read.empty() ? asio::const_buffer{} : asio::buffer(pending_read.front()); } } + + // + // ... and last from the chunk being delivered right now, which on_data_chunk() offers + // through `incoming` instead of parking it in a vector of its own first. Queued chunks + // go first: they arrived earlier. + // + if (dest.size() > 0 && incoming.size() > 0) + { + auto n = asio::buffer_copy(dest, incoming); + incoming += n; + copied += n; + } + consumed += copied; swap_and_invoke(read_handler, boost::system::error_code{}, copied); continue; @@ -1124,6 +1155,7 @@ void Http3Stream::delete_reader() auto self = shared_from_this(); // see delete_writer() pending_read.clear(); read_head = {}; + incoming = {}; // // The handler dropped the Request without reading the body to its end (e.g. not_found(), which From ae91d0c60d039d6ebdf202f6aa2c793f87b3a979 Mon Sep 17 00:00:00 2001 From: Peter Eisenlohr Date: Fri, 21 Aug 2026 07:23:36 +0000 Subject: [PATCH 15/20] log: change log level from info to debug in eat_request function --- src/request_handlers.cpp | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/request_handlers.cpp b/src/request_handlers.cpp index 0c62b12..672a2fb 100644 --- a/src/request_handlers.cpp +++ b/src/request_handlers.cpp @@ -109,7 +109,7 @@ awaitable not_found(server::Request, server::Response response) awaitable eat_request(server::Request request, server::Response response) { - logi("eat_request: going to eat {} bytes", request.content_length().value_or(-1)); + logd("eat_request: going to eat {} bytes", request.content_length().value_or(-1)); co_await response.async_submit(200, {}); co_await response.async_write({}); @@ -127,7 +127,7 @@ awaitable eat_request(server::Request request, server::Response response) logd("eat_request: ate {} bytes", n); bytes += n; } - logi("eat_request: ate {} bytes", bytes); + logd("eat_request: ate {} bytes", bytes); } catch (const boost::system::system_error& e) { From 3002243e49598fb70f25eaa563630b2792d543c2 Mon Sep 17 00:00:00 2001 From: Peter Eisenlohr Date: Fri, 21 Aug 2026 08:35:49 +0000 Subject: [PATCH 16/20] HTTP/3: let a re-issued EOF adopt a pending FIN without tripping the assert start_write() asserted !write_active before the block that handles a second async_write() of EOF -- and that block exists precisely for the case where write_active is still true: a cancelled FIN keeps its write alive (a FIN cannot be un-sent) and only detaches its handler, so the re-issued EOF adopts it. The assert and the branch it guards have contradicted each other since 843593f; what changed on this branch is the write-flush timing, which now leaves the FIN waiting for credit long enough for Backpressure/HTTP3 to hit it. Only Debug builds have asserts, so the RelWithDebInfo CI job stayed green while ASAN, TSAN and Coverage all aborted with SIGABRT at the same spot: client_impl_udp.cpp:771: Http3ClientStream::start_write(...): Assertion `!write_active' failed. Move the assert below the early return, where the invariant it states actually has to hold. The server side had the same shape, not yet reachable. Verified with the CI configuration (Debug, GITHUB_ACTIONS defined): ASAN 189 passed / 5 skipped, TSAN 192 / 2 with no ThreadSanitizer warnings, coverage target 194/194. No effect on release builds, where the assert is compiled out. Co-Authored-By: Claude Opus 5 --- src/client_impl_udp.cpp | 10 +++++++--- src/server_impl_udp.cpp | 10 +++++++--- 2 files changed, 14 insertions(+), 6 deletions(-) diff --git a/src/client_impl_udp.cpp b/src/client_impl_udp.cpp index 3eccc67..dbcbbdf 100644 --- a/src/client_impl_udp.cpp +++ b/src/client_impl_udp.cpp @@ -767,9 +767,6 @@ void Http3ClientStream::start_write(WriteHandler&& handler, asio::const_buffer b const bool is_eof = (n == 0); logd("[{}] start_write: n={} is_eof={}", log_prefix, n, is_eof); - // Only one async_write() may be active at a time -- see the class comment above write_active. - assert(!write_active); - // // Once accepted, the caller's intent to end the request body is final: this is what tells // delete_writer() the body ended where it was meant to, so it need not reset the stream. An @@ -799,6 +796,13 @@ void Http3ClientStream::start_write(WriteHandler&& handler, asio::const_buffer b return; } + // + // Only one async_write() may be active at a time -- see the class comment above write_active. + // The re-issued EOF handled above is not an exception to that: it adopts the FIN that is + // already in flight instead of starting a write of its own, and has returned by now. + // + assert(!write_active); + if (is_eof) eof_submitted = true; diff --git a/src/server_impl_udp.cpp b/src/server_impl_udp.cpp index 3d4555e..b276593 100644 --- a/src/server_impl_udp.cpp +++ b/src/server_impl_udp.cpp @@ -950,9 +950,6 @@ void Http3Stream::start_write(WriteHandler&& handler, asio::const_buffer buffer) const bool is_eof = (n == 0); logd("[{}] start_write: n={} is_eof={}", log_prefix, n, is_eof); - // Only one async_write() may be active at a time -- see the class comment above write_active. - assert(!write_active); - // // Once accepted, the caller's intent to end the response body is final: this is what tells // delete_writer() the body ended where it was meant to, so it need not reset the stream. An @@ -982,6 +979,13 @@ void Http3Stream::start_write(WriteHandler&& handler, asio::const_buffer buffer) return; } + // + // Only one async_write() may be active at a time -- see the class comment above write_active. + // The re-issued EOF handled above is not an exception to that: it adopts the FIN that is + // already in flight instead of starting a write of its own, and has returned by now. + // + assert(!write_active); + if (is_eof) eof_submitted = true; From ed04180c951f915a09df22080cf3bc2af311a627 Mon Sep 17 00:00:00 2001 From: Peter Eisenlohr Date: Fri, 21 Aug 2026 11:19:42 +0000 Subject: [PATCH 17/20] log: log submitted headers, and h3 received headers as one block Received headers were logged for every protocol, submitted ones for none except the h2 client request -- so a debug log showed what came in but never what went back out, which is the half you need when a response looks wrong on the wire. Log the full submitted header block at debug level in all three response paths (ResponseWriter::async_submit, NGHttp2Writer::async_submit, Http3Stream::submit_response) and in the h3 client's request path, which had been missing the dump its h2 counterpart already did. Same style as the received ones: bold-blue name, indented under the message's start line. On the h3 receive side, the recv_header callback logged one line per header as it arrived, so the block came out *above* the request/status line and interleaved with unrelated session logging. Buffer the raw name/value pairs on the stream and dump them in end_headers instead. Buffering is guarded by should_log(debug) and the storage is released as soon as it has been logged, so nothing is allocated at info level and above. The pairs are kept verbatim rather than re-rendered from the parsed request, which no longer has them: pseudo-headers are consumed into method/url/status_code/content_length and Fields::set() collapses duplicates. The status lines now read "200 OK" in all three, matching the "POST /echo" request line, rather than each protocol picking its own phrasing. [h3:[127.0.0.1]:50266.0] POST http://127.0.0.2/echo [h3:[127.0.0.1]:50266.0] :method: POST [h3:[127.0.0.1]:50266.0] :scheme: http [h3:[127.0.0.1]:50266.0] :path: /echo [h3:[127.0.0.1]:50266.0] :authority: 127.0.0.2 Trailers are unaffected: nghttp3 routes them through separate recv_trailer / end_trailers callbacks, which this code does not register. 192 passed / 2 skipped, unchanged. Co-Authored-By: Claude Opus 5 --- src/beast_session.cpp | 5 ++++ src/client_impl_udp.cpp | 29 +++++++++++++++++++- src/nghttp2_stream.cpp | 7 ++++- src/server_impl_udp.cpp | 61 +++++++++++++++++++++++++++++------------ 4 files changed, 83 insertions(+), 19 deletions(-) diff --git a/src/beast_session.cpp b/src/beast_session.cpp index 608bfee..0a0841e 100644 --- a/src/beast_session.cpp +++ b/src/beast_session.cpp @@ -395,6 +395,7 @@ class ResponseWriter WriterBase>; public: + using super::logPrefix; using super::message; using super::serializer; using super::stream; @@ -426,6 +427,10 @@ class ResponseWriter if (message.find(http::field::server) == message.end()) message.set(http::field::server, "anyhttp"); + mlogd("{} {}", message.result_int(), message.reason()); + for (const auto& header : message) + mlogd(" \x1b[1;34m{}\x1b[0m: {}", header.name_string(), header.value()); + // // TODO: For bundling writing the header and body, we should just post the writing here, // giving an async_write the chance to add a body to the message first. diff --git a/src/client_impl_udp.cpp b/src/client_impl_udp.cpp index dbcbbdf..1817399 100644 --- a/src/client_impl_udp.cpp +++ b/src/client_impl_udp.cpp @@ -142,6 +142,23 @@ nghttp3_nv make_nv(std::string_view name, std::string_view value) return nv; } +/// Logs a block of headers, one per line, in the same style as the received ones. +void log_headers(std::string_view log_prefix, const std::vector& nva) +{ + for (const auto& nv : nva) + logd("[{}] \x1b[1;34m{}\x1b[0m: {}", log_prefix, + std::string_view(reinterpret_cast(nv.name), nv.namelen), + std::string_view(reinterpret_cast(nv.value), nv.valuelen)); +} + +/// Same, for a header block buffered up by the recv_header callback. +void log_headers(std::string_view log_prefix, + const std::vector>& headers) +{ + for (const auto& [name, value] : headers) + logd("[{}] \x1b[1;34m{}\x1b[0m: {}", log_prefix, name, value); +} + void ngtcp2_log_printf(void* /*user*/, const char* fmt, ...) noexcept { if (!spdlog::default_logger()->should_log(spdlog::level::trace)) @@ -192,6 +209,13 @@ class Http3ClientStream : public std::enable_shared_from_this unsigned int status_code = 0; Fields response_fields; std::optional content_length; + + // + // The header block as it arrived, buffered so that h3_cb_end_headers() can log it in one go, + // below the status line, instead of one stray line per header as they come in. Only filled + // when debug logging is on, and dropped again as soon as it has been logged. + // + std::vector> received_headers; bool headers_received = false; bool response_delivered = false; client::Request::GetResponseHandler response_handler; @@ -1916,7 +1940,8 @@ int Http3ClientSession::h3_cb_recv_header(nghttp3_conn*, int64_t stream_id, int3 if (!s) return 0; - logd("[{}] \x1b[1;34m{}\x1b[0m: {}", s->log_prefix, name_view, value_view); + if (spdlog::default_logger_raw()->should_log(spdlog::level::debug)) + s->received_headers.emplace_back(name_view, value_view); try { @@ -1951,6 +1976,7 @@ int Http3ClientSession::h3_cb_end_headers(nghttp3_conn*, int64_t stream_id, int return 0; logd("[{}] response headers: status={}", s->log_prefix, s->status_code); + log_headers(s->log_prefix, std::exchange(s->received_headers, {})); s->headers_received = true; s->deliver_response(); return 0; @@ -2051,6 +2077,7 @@ void Http3ClientSession::async_submit(SubmitHandler&& handler, boost::urls::url } logd("[{}] async_submit: new stream ID: {}", stream->log_prefix, stream_id); + log_headers(stream->log_prefix, nva); wake_write(); post(get_executor(), [handler = std::move(handler), diff --git a/src/nghttp2_stream.cpp b/src/nghttp2_stream.cpp index 18ad37c..cb8a099 100644 --- a/src/nghttp2_stream.cpp +++ b/src/nghttp2_stream.cpp @@ -17,6 +17,7 @@ #include #include #include +#include #include #include @@ -202,7 +203,8 @@ void NGHttp2Writer::async_submit(StatusHandler&& handler, unsigned int sta return; } - logd("[{}] async_submit: {}", stream->logPrefix, status_code); + logd("[{}] {} {}", stream->logPrefix, status_code, + boost::beast::http::obsolete_reason(boost::beast::http::int_to_status(status_code))); auto nva = std::vector(); // nva.reserve(3 + headers.size()); @@ -227,6 +229,9 @@ void NGHttp2Writer::async_submit(StatusHandler&& handler, unsigned int sta nva.push_back(make_nv_ls("content-length", length_str)); } + for (auto nv : nva) + logd("[{0}] \x1b[1;34m{1:n}\x1b[0m: {1:v}", stream->logPrefix, nv); + // TODO: If we already know that there is no body, don't set a producer. nghttp2_data_provider2 prd; prd.source.ptr = stream; diff --git a/src/server_impl_udp.cpp b/src/server_impl_udp.cpp index b276593..d5a6be5 100644 --- a/src/server_impl_udp.cpp +++ b/src/server_impl_udp.cpp @@ -46,6 +46,7 @@ #include #include +#include #include #include #include @@ -56,7 +57,6 @@ #include #include #include -#include #include #include "ngtcp2/shared.h" @@ -271,6 +271,23 @@ nghttp3_nv make_nv(std::string_view name, std::string_view value) return nv; } +/// Logs a block of headers, one per line, in the same style as the received ones. +void log_headers(std::string_view log_prefix, const std::vector& nva) +{ + for (const auto& nv : nva) + logd("[{}] \x1b[1;34m{}\x1b[0m: {}", log_prefix, + std::string_view(reinterpret_cast(nv.name), nv.namelen), + std::string_view(reinterpret_cast(nv.value), nv.valuelen)); +} + +/// Same, for a header block buffered up by the recv_header callback. +void log_headers(std::string_view log_prefix, + const std::vector>& headers) +{ + for (const auto& [name, value] : headers) + logd("[{}] \x1b[1;34m{}\x1b[0m: {}", log_prefix, name, value); +} + void ngtcp2_log_printf(void* /*user*/, const char* fmt, ...) noexcept { if (!spdlog::default_logger()->should_log(spdlog::level::trace)) @@ -310,6 +327,13 @@ class Http3Stream : public std::enable_shared_from_this std::optional content_length; Fields request_fields; + // + // The header block as it arrived, buffered so that h3_cb_end_headers() can log it in one go, + // below the request line, instead of one stray line per header as they come in. Only filled + // when debug logging is on, and dropped again as soon as it has been logged. + // + std::vector> received_headers; + // // Response state (populated by user via Http3Writer). // @@ -324,7 +348,7 @@ class Http3Stream : public std::enable_shared_from_this // std::deque> pending_read; asio::const_buffer read_head; // view of pending_read.front() not yet delivered - asio::const_buffer incoming; // chunk on_data_chunk() is delivering, not yet taken + asio::const_buffer incoming; // chunk on_data_chunk() is delivering, not yet taken bool eof_received = false; ReadSomeHandler read_handler; asio::mutable_buffer read_handler_buffer; @@ -585,7 +609,7 @@ class Http3Session : public Session::Impl std::string log_prefix_; bool write_pending_ = false; // set by on_read(), acted on by flush_write() - bool write_posted_ = false; // a wake_write() flush is already on the way + bool write_posted_ = false; // a wake_write() flush is already on the way std::vector conn_closebuf_; // buffered CONNECTION_CLOSE packet @@ -941,7 +965,10 @@ void Http3Stream::submit_response() return; } response_submitted = true; - logd("[{}] response submitted (status={})", log_prefix, response_status); + + using namespace boost::beast::http; + logd("[{}] {} {}", log_prefix, response_status, obsolete_reason(int_to_status(response_status))); + log_headers(log_prefix, nva); } void Http3Stream::start_write(WriteHandler&& handler, asio::const_buffer buffer) @@ -1149,7 +1176,6 @@ void Http3Stream::finish_active_write() if (handler) swap_and_invoke(handler, boost::system::error_code{}); - } // ------------------------------------------------------------------------------------------------- @@ -1320,9 +1346,9 @@ void Http3Session::destroy() noexcept ngtcp2_pkt_info pi; ngtcp2_path_storage_zero(&ps); - auto nwrite = ngtcp2_conn_write_connection_close(conn_, &ps.path, &pi, closebuf.data(), - closebuf.size(), &last_error_, - ngtcp2::util::timestamp()); + auto nwrite = + ngtcp2_conn_write_connection_close(conn_, &ps.path, &pi, closebuf.data(), closebuf.size(), + &last_error_, ngtcp2::util::timestamp()); if (nwrite > 0) send_udp(ep_.fd, ps.path.remote.addr, ps.path.remote.addrlen, {closebuf.data(), static_cast(nwrite)}); @@ -1580,10 +1606,9 @@ ngtcp2_ssize Http3Session::write_pkt(ngtcp2_path* path, ngtcp2_pkt_info* pi, uin if (fin) flags |= NGTCP2_WRITE_STREAM_FLAG_FIN; - auto nwrite = - ngtcp2_conn_writev_stream(conn_, path, pi, dest, destlen, &ndatalen, flags, stream_id, - reinterpret_cast(vec.data()), - static_cast(sveccnt), ts); + auto nwrite = ngtcp2_conn_writev_stream( + conn_, path, pi, dest, destlen, &ndatalen, flags, stream_id, + reinterpret_cast(vec.data()), static_cast(sveccnt), ts); if (nwrite < 0) { @@ -1663,9 +1688,9 @@ int Http3Session::write_streams() ngtcp2_path_storage_zero(&ps); size_t gso_size = 0; - auto nwrite = ngtcp2_conn_write_aggregate_pkt2(conn_, &ps.path, &pi, tx_buf_.data(), - tx_buf_.size(), &gso_size, &write_pkt_cb, 0, - ngtcp2::util::timestamp()); + auto nwrite = + ngtcp2_conn_write_aggregate_pkt2(conn_, &ps.path, &pi, tx_buf_.data(), tx_buf_.size(), + &gso_size, &write_pkt_cb, 0, ngtcp2::util::timestamp()); if (nwrite < 0) { loge("[{}] ngtcp2_conn_write_aggregate_pkt2: {}", log_prefix_, @@ -2164,7 +2189,8 @@ int Http3Session::h3_cb_recv_header(nghttp3_conn*, int64_t stream_id, int32_t /* if (!s) return 0; - logd("[{}] \x1b[1;34m{}\x1b[0m: {}", s->log_prefix, name_view, value_view); + if (spdlog::default_logger_raw()->should_log(spdlog::level::debug)) + s->received_headers.emplace_back(name_view, value_view); try { @@ -2210,6 +2236,7 @@ int Http3Session::h3_cb_end_headers(nghttp3_conn*, int64_t stream_id, int /*fin* return 0; logd("[{}] {} {}", s->log_prefix, s->method, s->url.buffer()); + log_headers(s->log_prefix, std::exchange(s->received_headers, {})); // // Build user-facing Request/Response and dispatch through the shared handler. @@ -2318,7 +2345,7 @@ int Server::Impl::udp_on_read(Endpoint& ep) for (size_t pktcnt = 0; pktcnt < 32; ++pktcnt) { if (pktcnt) - logd("- - {} - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - ", pktcnt); + logd("- - {} - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - ", pktcnt); msg.msg_namelen = sizeof(su); msg.msg_controllen = sizeof(msg_ctrl); From bca74b51db4ef72e72836e85a585828acab4ce1b Mon Sep 17 00:00:00 2001 From: Peter Eisenlohr Date: Fri, 21 Aug 2026 13:30:14 +0000 Subject: [PATCH 18/20] log: add segment counter to udp_on_read for better debugging --- src/server_impl_udp.cpp | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/server_impl_udp.cpp b/src/server_impl_udp.cpp index d5a6be5..4b93cc0 100644 --- a/src/server_impl_udp.cpp +++ b/src/server_impl_udp.cpp @@ -2394,8 +2394,11 @@ int Server::Impl::udp_on_read(Endpoint& ep) auto all_data = std::span{buf.data(), static_cast(nread)}; const size_t seg_size = gro_size > 0 ? gro_size : all_data.size(); - while (!all_data.empty()) + for (size_t segcnt = 0; !all_data.empty(); ++segcnt) { + if (segcnt) + logd("- {} - - - - - - - - - - - - - - -", segcnt); + auto data = all_data.subspan(0, std::min(seg_size, all_data.size())); all_data = all_data.subspan(data.size()); From 89701fd8a72a043300d49b6f5515f0ec619aa57b Mon Sep 17 00:00:00 2001 From: Peter Eisenlohr Date: Fri, 21 Aug 2026 20:54:17 +0000 Subject: [PATCH 19/20] HTTP/3: drop the session when the QUIC idle timer fires A peer that vanishes without a CONNECTION_CLOSE -- h2load interrupted mid-run, a killed client -- is noticed only by the idle timer. But the IDLE_CLOSE branch of handle_error() just signalled done: with no packet sent there is no closing period, so the cleanup in udp_on_read() never ran for such a session, and it stayed in m_quic_handlers for the lifetime of the server, holding streams whose request handlers were still suspended on a peer that was long gone. Erase it from the demux map right there instead, and cancel its timer. The client side skips its CONNECTION_CLOSE in the same situation, and both sides log the idle timeout as an ordinary end of life rather than as a warning. Server::Config gains an idle_timeout knob (default 30s, as before) so the new testcase can freeze a client mid-request and watch the server give up on it half a second later. Co-Authored-By: Claude Opus 5 --- include/anyhttp/server.hpp | 8 +++ src/client_impl_udp.cpp | 22 +++++++- src/server_impl_udp.cpp | 27 ++++++++- test/test_server.cpp | 110 +++++++++++++++++++++++++++++++++++++ 4 files changed, 162 insertions(+), 5 deletions(-) diff --git a/include/anyhttp/server.hpp b/include/anyhttp/server.hpp index c64c44b..0e35b6e 100644 --- a/include/anyhttp/server.hpp +++ b/include/anyhttp/server.hpp @@ -21,6 +21,14 @@ struct Config std::string listen_address = "::"; uint16_t port = 8080; bool use_strand = false; + + // + // HTTP/3 only: how long a QUIC connection may go without a packet from its peer before it is + // dropped. This is the only way a peer that vanished without a CONNECTION_CLOSE -- a killed + // client, a machine that went to sleep -- is ever noticed, so it also bounds how long its + // session and streams stay around. 30s is what the ngtcp2 examples use. + // + std::chrono::nanoseconds idle_timeout = 30s; }; // ================================================================================================= diff --git a/src/client_impl_udp.cpp b/src/client_impl_udp.cpp index 1817399..00d501d 100644 --- a/src/client_impl_udp.cpp +++ b/src/client_impl_udp.cpp @@ -1380,7 +1380,16 @@ void Http3ClientSession::close() for (auto& stream : streams) stream->fail(errc::make_error_code(errc::connection_reset)); - if (conn_ && !ngtcp2_conn_in_closing_period(conn_) && !ngtcp2_conn_in_draining_period(conn_)) + // + // An idle-timed-out (or dropped) connection is discarded silently: RFC 9000 has no + // CONNECTION_CLOSE for it, and there is nobody left listening anyway -- writing one would + // just put a packet on a path whose peer has been gone for a full idle period. + // + const bool silent = last_error_.type == NGTCP2_CCERR_TYPE_IDLE_CLOSE || + last_error_.type == NGTCP2_CCERR_TYPE_DROP_CONN; + + if (conn_ && !silent && !ngtcp2_conn_in_closing_period(conn_) && + !ngtcp2_conn_in_draining_period(conn_)) { std::array closebuf; ngtcp2_path_storage ps; @@ -1631,7 +1640,16 @@ int Http3ClientSession::handle_expiry() auto now = ngtcp2::util::timestamp(); if (auto rv = ngtcp2_conn_handle_expiry(conn_, now); rv != 0) { - logw("[{}] ngtcp2_conn_handle_expiry: {}", log_prefix_, ngtcp2_strerror(rv)); + // + // NGTCP2_ERR_IDLE_CLOSE is how a connection whose peer stopped talking ends: a normal + // end of life, not a failure worth a warning. close() then discards it silently, see + // there. + // + if (rv == NGTCP2_ERR_IDLE_CLOSE) + logi("[{}] idle timeout, dropping connection", log_prefix_); + else + logw("[{}] ngtcp2_conn_handle_expiry: {}", log_prefix_, ngtcp2_strerror(rv)); + ngtcp2_ccerr_set_liberr(&last_error_, rv, nullptr, 0); return handle_error(rv); } diff --git a/src/server_impl_udp.cpp b/src/server_impl_udp.cpp index 4b93cc0..00f11d7 100644 --- a/src/server_impl_udp.cpp +++ b/src/server_impl_udp.cpp @@ -1470,7 +1470,7 @@ int Http3Session::init(const ngtcp2_cid& dcid, const ngtcp2_cid& scid, uint32_t params.initial_max_data = 1_m; params.initial_max_streams_bidi = 100; params.initial_max_streams_uni = 3; - params.max_idle_timeout = std::chrono::nanoseconds(30s).count(); + params.max_idle_timeout = server_.config().idle_timeout.count(); params.original_dcid = dcid; params.original_dcid_present = 1; @@ -1763,7 +1763,18 @@ int Http3Session::handle_expiry() auto now = ngtcp2::util::timestamp(); if (auto rv = ngtcp2_conn_handle_expiry(conn_, now); rv != 0) { - logw("[{}] ngtcp2_conn_handle_expiry: {}", log_prefix_, ngtcp2_strerror(rv)); + // + // NGTCP2_ERR_IDLE_CLOSE is how a connection whose peer simply stopped talking ends -- + // an interrupted client leaves one behind per connection it had open -- so it is a + // normal end of life, not a failure worth a warning. handle_error() takes it from here + // either way; what makes it special is that it discards the connection silently, see + // there. + // + if (rv == NGTCP2_ERR_IDLE_CLOSE) + logi("[{}] idle timeout, dropping connection", log_prefix_); + else + logw("[{}] ngtcp2_conn_handle_expiry: {}", log_prefix_, ngtcp2_strerror(rv)); + ngtcp2_ccerr_set_liberr(&last_error_, rv, nullptr, 0); return handle_error(rv); } @@ -1778,10 +1789,20 @@ int Http3Session::handle_error(int /*rv*/) return -1; closed_ = true; - // Idle timeout and drop-conn need no CONNECTION_CLOSE packet. + // + // Idle timeout and drop-conn need no CONNECTION_CLOSE packet -- and with no packet there is + // no closing period either, so none of the cleanup in Server::Impl::udp_on_read() can ever + // run for this session: it is reached from the expiry timer precisely because nothing is + // arriving any more. Drop the session from the demux map right here instead, or it would sit + // in m_quic_handlers for the lifetime of the server, holding streams whose request handlers + // are still waiting on a peer that went away. What is left of it then dies with do_session(). + // if (last_error_.type == NGTCP2_CCERR_TYPE_IDLE_CLOSE || last_error_.type == NGTCP2_CCERR_TYPE_DROP_CONN) { + auto self = weak_from_this().lock(); // erase_quic_session() may drop the last reference + timer_.cancel(); + server_.erase_quic_session(this); signal_done(); return -1; } diff --git a/test/test_server.cpp b/test/test_server.cpp index 15c07a9..60bb8fc 100644 --- a/test/test_server.cpp +++ b/test/test_server.cpp @@ -1777,3 +1777,113 @@ TEST_P(ClientAsync, DISABLED_SpawnAndForget) } // ================================================================================================= + +// +// A QUIC peer that goes away without a word -- a killed client, a machine that went to sleep in +// the middle of a request -- leaves the server nothing to react to: no CONNECTION_CLOSE arrives, +// and no further packet ever will. Only the idle timer can notice, and dropping the connection +// when it fires is what releases the session, its streams, and the request handlers suspended on +// them. +// +// Note that this is *not* what a client calling Session::reset() looks like: that one says +// goodbye, and the server cleans up right away by way of the draining period. +// +class Http3IdleTimeout : public testing::Test +{ +protected: + static constexpr auto IdleTimeout = 500ms; + + void SetUp() override + { + setupLogging(); + + server.emplace(context.get_executor(), server::Config{.listen_address = "127.0.0.2", + .port = 0, + .idle_timeout = IdleTimeout}); + server->setRequestHandler( + [this](server::Request request, server::Response response) -> awaitable + { + co_await response.async_submit(200, {}); + + // + // Wait for a request body that never comes: this first read is where the handler is + // suspended when the client freezes, and it must be resumed -- with an error -- once + // the server gives up on the connection. + // + std::array buffer; + auto [ec, n] = co_await request.async_read_some(asio::buffer(buffer), as_tuple); + handler_result.set_value(ec); + }); + + url.set_port_number(server->local_endpoint().port()); + } + + asio::io_context context; + std::optional server; + + std::promise handler_result; + boost::urls::url url{"http://127.0.0.2/echo"}; +}; + +// ------------------------------------------------------------------------------------------------- + +TEST_F(Http3IdleTimeout, WHEN_client_vanishes_in_flight_THEN_idle_timer_drops_the_session) +{ + auto result = handler_result.get_future(); + + // + // The server has to keep running while the client is frozen, so it gets a thread of its own. + // + std::jthread server_thread([this] { context.run(); }); + boost::scope::scope_exit stop_server([this] { context.stop(); }); + + // + // The client runs on its own io_context, which is what makes freezing it possible: stopping + // that context takes the client off the air mid-request without unwinding anything, so no + // CONNECTION_CLOSE is ever sent -- just like a client process that was killed. + // + asio::io_context client_context; + client::Client client(client_context.get_executor(), + client::Config{.url = url, .protocol = anyhttp::Protocol::h3}); + + // + // Session, request and response are kept out here rather than in the coroutine frame, which is + // destroyed as soon as the coroutine below returns: unwinding them would reset the stream, and + // that is a packet -- the one thing this client must not send. + // + std::optional session; + std::optional request; + std::optional response; + + bool responded = false; + co_spawn(client_context, [&]() -> awaitable + { + session = co_await client.async_connect(); + request = co_await session->async_submit(url, {}); + response = co_await request->async_get_response(); + responded = true; + }, detached); + + // + // Run the client just far enough to have the request open and answered, then stop running it: + // from here on it never touches its socket again. + // + while (client_context.run_one() && !responded); + ASSERT_TRUE(responded) << "client never received a response"; + std::println("=== freezing the client, request still in flight ==="); + + // + // From here on the server is on its own. Without the idle timer dropping the connection, the + // request handler stays suspended in async_read_some() forever, and the session it belongs to + // sits in the server's connection table for good. + // + ASSERT_EQ(result.wait_for(5s), std::future_status::ready) << "request handler never completed"; + EXPECT_EQ(result.get(), boost::system::errc::connection_reset); + + // + // Only now, on the way out, is the frozen client allowed to unwind: doing so earlier would + // have sent the CONNECTION_CLOSE that this test is all about not sending. + // +} + +// ================================================================================================= From 3c713356339a18f42fe7551c74acf5139d6c75e0 Mon Sep 17 00:00:00 2001 From: Peter Eisenlohr Date: Fri, 21 Aug 2026 21:08:41 +0000 Subject: [PATCH 20/20] HTTP/3: add random UDP packet dropping for testing Loss recovery -- retransmits, PTO, ACK handling -- was only ever exercised by whatever the loopback happened to do, which is nothing. Server::Config gains drop_rate_rx and drop_rate_tx, the probability (0.0 .. 1.0, default 0.0 = off) that an individual QUIC packet is thrown away instead of being processed or sent. The rates ride along on Endpoint, which every Http3Session copies, so there is no process-wide state to collide between servers in a test. RX dropping sits in the segment loop of udp_on_read(), after GRO splitting, so coalesced datagrams are dropped one packet at a time rather than 64k at once. TX dropping sits in send_udp(), the funnel all send paths share; send_udp_gso() takes its per-packet fallback path while drop_rate_tx is non-zero, because one GSO batch would otherwise be an all-or-nothing drop of up to N packets. A dropped packet reports success, so ngtcp2 simply retransmits it. The server binary exposes both as --drop-rx and --drop-tx. At 15% in both directions, an h2load run over h3 still completes every request, with the drops visible in the log. Co-Authored-By: Claude Opus 5 --- include/anyhttp/server.hpp | 10 +++++++ src/server_impl_udp.cpp | 61 ++++++++++++++++++++++++++++++-------- src/server_main.cpp | 14 +++++++++ 3 files changed, 73 insertions(+), 12 deletions(-) diff --git a/include/anyhttp/server.hpp b/include/anyhttp/server.hpp index 0e35b6e..5a9bb6e 100644 --- a/include/anyhttp/server.hpp +++ b/include/anyhttp/server.hpp @@ -29,6 +29,16 @@ struct Config // session and streams stay around. 30s is what the ngtcp2 examples use. // std::chrono::nanoseconds idle_timeout = 30s; + + // + // HTTP/3 only, testing aid: probability (0.0 ... 1.0) with which an individual QUIC datagram + // is thrown away instead of being processed (rx) or actually sent (tx). This exercises loss + // recovery -- retransmits, PTO, ACK handling -- without needing a lossy network. Dropping + // happens per QUIC packet, i.e. GRO-coalesced datagrams are dropped individually and TX + // GSO batching is bypassed while `drop_rate_tx` is non-zero. + // + double drop_rate_rx = 0.0; + double drop_rate_tx = 0.0; }; // ================================================================================================= diff --git a/src/server_impl_udp.cpp b/src/server_impl_udp.cpp index 00f11d7..5bb6573 100644 --- a/src/server_impl_udp.cpp +++ b/src/server_impl_udp.cpp @@ -53,6 +53,7 @@ #include #include #include +#include #include #include #include @@ -75,6 +76,10 @@ struct Endpoint { ngtcp2::Address addr; int fd; + + // Testing aid, see server::Config::drop_rate_rx/tx. + double drop_rate_rx = 0.0; + double drop_rate_tx = 0.0; }; // ================================================================================================= @@ -181,11 +186,34 @@ std::string cid_key(const uint8_t* data, size_t len) // ------------------------------------------------------------------------------------------------- -int send_udp(int fd, const sockaddr* sa, socklen_t salen, std::span data) +// +// Testing aid: rolls the dice for a single QUIC packet. `rate` is the probability of the packet +// being dropped, 0.0 (never, the default) to 1.0 (always). The generator is deliberately not +// seeded deterministically -- this is meant to shake out loss handling over many runs, not to +// reproduce one exact sequence. +// +bool drop_packet(double rate) +{ + if (rate <= 0.0) + return false; + + static thread_local std::mt19937 rng{std::random_device{}()}; + return std::uniform_real_distribution{0.0, 1.0}(rng) < rate; +} + +// ------------------------------------------------------------------------------------------------- + +int send_udp(const Endpoint& ep, const sockaddr* sa, socklen_t salen, std::span data) { + if (drop_packet(ep.drop_rate_tx)) + { + // logw("*** dropping outgoing packet ({} bytes) ***", data.size()); + return 0; // pretend it went out; ngtcp2 will retransmit + } + for (;;) { - auto n = ::sendto(fd, data.data(), data.size(), 0, sa, salen); + auto n = ::sendto(ep.fd, data.data(), data.size(), 0, sa, salen); if (n == -1) { if (errno == EINTR) @@ -205,15 +233,16 @@ int send_udp(int fd, const sockaddr* sa, socklen_t salen, std::span data, - size_t gso_size, bool& no_gso) +int send_udp_gso(const Endpoint& ep, const sockaddr* sa, socklen_t salen, + std::span data, size_t gso_size, bool& no_gso) { - if (no_gso || data.size() <= gso_size) + // With TX dropping enabled, go packet by packet so each one can be dropped individually. + if (no_gso || data.size() <= gso_size || ep.drop_rate_tx > 0.0) { for (; !data.empty();) { auto len = std::min(gso_size, data.size()); - if (send_udp(fd, sa, salen, data.first(len)) != 0) + if (send_udp(ep, sa, salen, data.first(len)) != 0) return -1; data = data.subspan(len); } @@ -239,7 +268,7 @@ int send_udp_gso(int fd, const sockaddr* sa, socklen_t salen, std::span 0) - send_udp(ep_.fd, ps.path.remote.addr, ps.path.remote.addrlen, + send_udp(ep_, ps.path.remote.addr, ps.path.remote.addrlen, {closebuf.data(), static_cast(nwrite)}); } @@ -1705,7 +1734,7 @@ int Http3Session::write_streams() if (nwrite == 0) return 0; - return send_udp_gso(ep_.fd, ps.path.remote.addr, ps.path.remote.addrlen, + return send_udp_gso(ep_, ps.path.remote.addr, ps.path.remote.addrlen, {tx_buf_.data(), static_cast(nwrite)}, gso_size, no_gso_); } @@ -1823,7 +1852,7 @@ int Http3Session::handle_error(int /*rv*/) { conn_closebuf_.resize(static_cast(nwrite)); logi("[{}] sending CONNECTION_CLOSE", log_prefix_); - send_udp(ep_.fd, ps.path.remote.addr, ps.path.remote.addrlen, + send_udp(ep_, ps.path.remote.addr, ps.path.remote.addrlen, {conn_closebuf_.data(), conn_closebuf_.size()}); } else @@ -1866,7 +1895,7 @@ void Http3Session::resend_conn_close() if (!path) return; logd("[{}] resending CONNECTION_CLOSE", log_prefix_); - send_udp(ep_.fd, path->remote.addr, path->remote.addrlen, + send_udp(ep_, path->remote.addr, path->remote.addrlen, {conn_closebuf_.data(), conn_closebuf_.size()}); } @@ -2423,6 +2452,12 @@ int Server::Impl::udp_on_read(Endpoint& ep) auto data = all_data.subspan(0, std::min(seg_size, all_data.size())); all_data = all_data.subspan(data.size()); + if (drop_packet(ep.drop_rate_rx)) + { + // logw("*** dropping received packet ({} bytes) ***", data.size()); + continue; + } + ngtcp2_version_cid vc; auto rv = ngtcp2_pkt_decode_version_cid(&vc, data.data(), data.size(), QUIC_SCIDLEN); if (rv != 0) @@ -2557,6 +2592,8 @@ awaitable Server::Impl::udp_receive_loop() Endpoint ep{}; ep.fd = m_udp_socket->native_handle(); + ep.drop_rate_rx = m_config.drop_rate_rx; + ep.drop_rate_tx = m_config.drop_rate_tx; auto local = m_udp_socket->local_endpoint(); auto data = local.data(); std::memcpy(&ep.addr.su, data, local.size()); diff --git a/src/server_main.cpp b/src/server_main.cpp index d4d084a..fc01311 100644 --- a/src/server_main.cpp +++ b/src/server_main.cpp @@ -45,6 +45,10 @@ std::expected parseConfig(int argc, char* argv[]) opts("threads,t", po::value(&config.threads)->default_value(1), "number of threads to run"); opts("port,p", po::value(&config.server.port)->default_value(config.server.port), "listening port"); + opts("drop-rx", po::value(&config.server.drop_rate_rx)->default_value(0.0), + "HTTP/3 testing: probability (0.0 .. 1.0) of dropping a received QUIC packet"); + opts("drop-tx", po::value(&config.server.drop_rate_tx)->default_value(0.0), + "HTTP/3 testing: probability (0.0 .. 1.0) of dropping a QUIC packet before sending it"); po::variables_map vm; try @@ -71,6 +75,16 @@ std::expected parseConfig(int argc, char* argv[]) return std::unexpected(1); } + for (auto [name, rate] : {std::pair{"drop-rx", config.server.drop_rate_rx}, + std::pair{"drop-tx", config.server.drop_rate_tx}}) + { + if (rate < 0.0 || rate > 1.0) + { + std::println(std::cerr, "--{} must be between 0.0 and 1.0", name); + return std::unexpected(1); + } + } + return {std::move(config)}; }