From 426a6328f23920f504408deeca8f089d5c6242d1 Mon Sep 17 00:00:00 2001 From: Peter Eisenlohr Date: Sat, 7 Nov 2026 20:05:16 +0000 Subject: [PATCH 1/6] chore: a build-coverage preset for the published report, without docs/ The Coverage job configured its tree by hand; it now uses the build-coverage preset, which reproduces the report locally. docs/snippets.cpp is linked into test_all to be run, not to be measured, so docs/ joins test/ among the paths the report leaves out. Co-Authored-By: Claude Opus 5.5 --- .github/workflows/release.yml | 11 ++++++----- CLAUDE.md | 5 +++++ CMakeLists.txt | 5 ++++- CMakePresets.json | 22 ++++++++++++++++++++++ 4 files changed, 37 insertions(+), 6 deletions(-) diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index edfc43a..c9ca4cb 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -183,19 +183,20 @@ jobs: - name: Configure CMake with Coverage run: | - cmake -B build -G Ninja -DCMAKE_BUILD_TYPE=Debug -DENABLE_COVERAGE=ON + cmake --preset build-coverage - - *build + - name: Build + run: cmake --build --preset build-coverage - name: Generate Coverage Report run: | - cmake --build build --target coverage + cmake --build --preset build-coverage --target coverage - name: Prepare Pages directory if: github.ref == 'refs/heads/master' && github.event_name == 'push' run: | mkdir -p pages/coverage - cp -r build/coverage_report/* pages/coverage/ + cp -r build-coverage/coverage_report/* pages/coverage/ - name: Get the documentation if: github.ref == 'refs/heads/master' && github.event_name == 'push' @@ -208,7 +209,7 @@ jobs: uses: actions/upload-artifact@v7 with: name: coverage-report - path: build/coverage_report/ + path: build-coverage/coverage_report/ retention-days: 30 - name: Upload coverage and documentation to GitHub Pages diff --git a/CLAUDE.md b/CLAUDE.md index e239f7a..3b21945 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -50,6 +50,11 @@ FetchContent subproject, in `build-package/`. Run it after changing anything in or what a public header includes: each tree here builds the library with the tree's own directory flags, so a usage requirement that is missing from the `anyhttp` target shows up only out there. +`build-coverage/` is the report CI publishes at https://pgit.github.io/anyhttp/coverage/: Debug, +clang with `ENABLE_COVERAGE`. `cmake --build build-coverage --target coverage` runs the suite and +writes `build-coverage/coverage_report/index.html`. It counts the library only: `test/` and `docs/` +(the snippets are linked into `test_all`) are left out. + Every tree is a preset of its own name in [CMakePresets.json](CMakePresets.json) -- that is where the options of each one live, and the options of a new tree belong there rather than in a command line. `cmake --list-presets` names them all. Do not configure a tree by hand: `cc` and `c++` are diff --git a/CMakeLists.txt b/CMakeLists.txt index 219050e..e188036 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -266,7 +266,10 @@ if (ENABLE_COVERAGE) set(COVERAGE_HTML_DIR "${PROJECT_BINARY_DIR}/coverage_report") set(COVERAGE_MERGE_SCRIPT "${PROJECT_SOURCE_DIR}/cmake/merge_profraw.cmake") set(COVERAGE_TEST_BIN "${PROJECT_BINARY_DIR}/test/test_all") - set(LLVM_COV_IGNORE_REGEX "/usr/|${PROJECT_BINARY_DIR}/|${PROJECT_SOURCE_DIR}/test/") + # The code the guide shows (docs/snippets.cpp) is linked into test_all, but it is there to be + # read, not tested, much like test/: a snippet may be compiled and never run. + set(LLVM_COV_IGNORE_REGEX + "/usr/|${PROJECT_BINARY_DIR}/|${PROJECT_SOURCE_DIR}/test/|${PROJECT_SOURCE_DIR}/docs/") add_custom_target(coverage COMMAND ${CMAKE_COMMAND} -E rm -rf ${COVERAGE_RAW_DIR} ${COVERAGE_HTML_DIR} diff --git a/CMakePresets.json b/CMakePresets.json index 5698a75..8670697 100644 --- a/CMakePresets.json +++ b/CMakePresets.json @@ -149,6 +149,19 @@ "base" ] }, + { + "name": "build-coverage", + "displayName": "Coverage", + "description": "What CI's Coverage job publishes: 'cmake --build build-coverage --target coverage' runs the tests and writes build-coverage/coverage_report/", + "inherits": [ + "clang", + "base" + ], + "cacheVariables": { + "CMAKE_BUILD_TYPE": "Debug", + "ENABLE_COVERAGE": "ON" + } + }, { "name": "build-openssl", "displayName": "OpenSSL", @@ -315,6 +328,10 @@ "name": "build-ubsan", "configurePreset": "build-ubsan" }, + { + "name": "build-coverage", + "configurePreset": "build-coverage" + }, { "name": "build-openssl", "configurePreset": "build-openssl" @@ -403,6 +420,11 @@ "configurePreset": "build-ubsan", "inherits": "base" }, + { + "name": "build-coverage", + "configurePreset": "build-coverage", + "inherits": "base" + }, { "name": "build-openssl", "configurePreset": "build-openssl", From ec387ffec7f3476159a9303d290849cd61eaa70e Mon Sep 17 00:00:00 2001 From: Peter Eisenlohr Date: Sat, 7 Nov 2026 20:05:16 +0000 Subject: [PATCH 2/6] test: Session::Impl's default shutdown() reports operation_not_supported No backend that hands out a Session leaves the default in place, so a session with nothing behind it stands in for one. Co-Authored-By: Claude Opus 5.5 --- test/test_session.cpp | 67 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 67 insertions(+) create mode 100644 test/test_session.cpp diff --git a/test/test_session.cpp b/test/test_session.cpp new file mode 100644 index 0000000..49cae96 --- /dev/null +++ b/test/test_session.cpp @@ -0,0 +1,67 @@ +// +// What Session::Impl does for a backend that does not do it itself. Only the client hands out a +// Session, and each client backend has an orderly end of its own, so these defaults are not +// reachable over a real connection: a session with nothing behind it stands in for one. +// +#include "test_fixtures.hpp" + +#include "anyhttp/session_impl.hpp" + +#include +#include +#include +#include + +using namespace testing; + +namespace +{ + +// ================================================================================================= + +/// A session that implements nothing beyond what it must, to see what the base class adds. +class BareSession : public Session::Impl +{ +public: + explicit BareSession(Executor executor) : executor_(std::move(executor)) {} + + Executor get_executor() const noexcept override { return executor_; } + Protocol protocol() const noexcept override { return Protocol::h1; } + void async_submit(SubmitHandler&& handler, std::string_view, boost::urls::url, + const Fields&) override + { + std::move(handler)(make_error_code(errc::operation_not_supported), client::Request{}); + } + Task do_session(Buffer&&) override { co_return; } + void destroy() noexcept override { destroyed = true; } + + bool destroyed = false; + +private: + Executor executor_; +}; + +Task shut_down(Session& session, error_code& result) +{ + std::tie(result) = co_await session.shutdown(); +} + +} // namespace + +// ================================================================================================= + +TEST(SessionImpl, WHEN_backend_has_no_orderly_end_THEN_shutdown_is_not_supported) +{ + IoContext context; + auto impl = std::make_shared(context.get_executor()); + Session session{impl}; + + error_code result = errors::canceled; + launch(context.get_executor(), shut_down(session, result)); + context.run(); + + EXPECT_EQ(result, errc::operation_not_supported); + EXPECT_FALSE(impl->destroyed) << "not supported is not a reason to end it the abrupt way"; +} + +// ================================================================================================= From 21ca831208361d480ff93edc59f086c6e0520603 Mon Sep 17 00:00:00 2001 From: Peter Eisenlohr Date: Sat, 7 Nov 2026 20:05:47 +0000 Subject: [PATCH 3/6] fix: complete immediately through the caller's immediate executor complete_immediately() handed asio::async_immediate() a lambda around the handler, and async_immediate() asks what it is given for its immediate executor: the lambda has none, so an immediate executor the caller bound was never fetched and every completion was posted. It now executes on the handler's own, falling back to a post as before, so the caller decides whether it is completed inline. So does the default async_shutdown_impl() of any_async_stream. The second half of the Recursion test had passed regardless, through the extra frames of the binder's path; it now reads twice and requires the second read to nest deeper (HTTP/2 and HTTP/3 only, HTTP/1.1 reads in a task of its own and never completes immediately). Each operation completes the same way on all of its paths. The client's submit and get_response called their handler inline when they failed, and post it now, as they do when they succeed. A server's Response::async_submit() over HTTP/2 and HTTP/3 completes inline on all paths: the responses of requests that arrived together go out in one send only if their handlers have all written before the write pass runs, and queueing the completion, ahead of the pass or behind it, cost 16-41% of throughput. docs/benchmark.md has the measurements. complete_later() is complete() now, being the default way to complete. Co-Authored-By: Claude Opus 5.5 --- docs/benchmark.md | 66 +++++++++++++++++++ docs/corosio-port-plan.md | 2 +- .../anyhttp/asio/any_async_stream_impl.hpp | 12 +--- include/anyhttp/asio/runtime.hpp | 18 +++-- include/anyhttp/corosio/runtime.hpp | 2 +- include/anyhttp/h3/stream.hpp | 7 +- include/anyhttp/quic/stream.hpp | 4 +- include/anyhttp/runtime.hpp | 26 ++++---- src/h2/session.cpp | 17 +++-- src/h2/stream.cpp | 21 ++++-- src/h2v2/session.cpp | 9 +-- src/h2v2/stream.cpp | 17 +++-- src/h3/client.cpp | 20 +++--- src/quic/client.cpp | 3 +- src/quic/stream.cpp | 6 +- src/quic/streams.cpp | 9 ++- test/test_client_async.cpp | 23 +++++-- test/test_corosio.cpp | 4 +- 18 files changed, 185 insertions(+), 81 deletions(-) diff --git a/docs/benchmark.md b/docs/benchmark.md index bd9415a..163e433 100644 --- a/docs/benchmark.md +++ b/docs/benchmark.md @@ -15,6 +15,7 @@ never against a number in this file. - [HTTP/2: nghttpd vs anyhttp](#http2-nghttpd-vs-anyhttp) - [Threads: one shared context or one context per thread](#threads-one-shared-context-or-one-context-per-thread) - [ASIO vs COROSIO](#asio-vs-corosio) +- [Completing an operation inline or posted](#completing-an-operation-inline-or-posted) - [Earlier findings](#earlier-findings) - [Bugs found while benchmarking](#bugs-found-while-benchmarking) @@ -374,6 +375,71 @@ posts everything. `server --inline-budget N` / `--unassisted-budget N` (COROSIO --------------------------------------------------------------------------------------------------- +## Completing an operation inline or posted + +*2026-11-06, on `b489baa` plus the uncommitted completion-token work.* What an operation that has +nothing left to wait for costs, depending on how it completes: by calling its handler from the +initiating function, or through `complete_immediately()` / `complete()`, which post. + +### What `complete_immediately()` does + +It completes through the handler's own immediate executor, so the caller decides: a handler that +has one bound (`bind_immediate_executor()`) may be completed inline, any other is posted to the +fallback with `blocking.never`. From `44851a7` until now it handed `asio::async_immediate()` a +lambda that wraps the handler instead, and `async_immediate()` asks what it is given for its +immediate executor: the lambda has none, so everything was posted and a caller's binding was +ignored. `experiments/immediate_binding.cpp` shows the difference; the comment in +`complete_immediately()` says why it does not use `async_immediate()`. The `Recursion` test in +`test_client_async.cpp` passed throughout, through the extra frames of the binder's path; it now +reads twice and requires the second read to nest deeper than the first. + +### Results + +`experiments/immediate_completion.cpp`, Release, one thread: a coroutine awaits a trivial +operation in a loop, as a reader does that drains a body which has already arrived. Three runs +each, ns per operation: + +| completion | back to back | with a post after each | +|---|---|---| +| handler called inline | 133-169 | 146-148 | +| `complete_immediately()` before the fix (`async_immediate()`) | 69-75 | 167-174 | +| `complete()` | 73-76 | 171-174 | +| `complete_immediately()` now, nothing bound | 81-82 | 190-192 | + +- **Inline nests.** An ASIO coroutine resumed from inside the initiating function runs on top of + it: about 416 bytes of stack per operation with `as_tuple(use_awaitable)`, 576 with an immediate + executor bound, 752 through the `#else` branch with one. Back to back, the stack is gone after + some 20,000 operations (8 MiB); the inline numbers in that column include faulting in fresh + stack. Every posted mode stays flat. +- **Where nothing nests, inline saves about 25 ns** per operation over a post (the second + column: each operation is followed by a real post, as a server's coroutine waits for the + network sooner or later). A request costs about 3 us of CPU, so that is under 1% even for an + operation that every request goes through. +- **Honouring the binding costs 10-20 ns** per operation over a plain post, for the + `any_completion_executor` in between. The server does not show it: five interleaved rounds of + `h2load -c 32` (HTTP/1.1 `-m 1`, HTTP/2 and HTTP/3 `-m 10`) on `/`, and three of a 64 KiB upload + to `/upload`, were within noise of the build that still used `async_immediate()` (2026-11-07). + +### What it means + +Calling a handler inline is safe where the caller cannot loop on it without waiting for something +real in between: once per response or per request, not once per read of a buffered body. What it +buys there is not the post, but ordering. `Response::async_submit()` on HTTP/2 and HTTP/3 completes +inline because the requests that arrive together on a connection start their handlers together, +and their responses go out in one send only if every handler has written before the write pass +runs. A queued completion, ahead of the write pass or behind it, lets the pass run in the middle: +with `h2load -c 32 -m 10`, sends per request went from 0.11 to 0.31, and throughput fell by 19-27% +(HTTP/2) and 16-41% (HTTP/3). With `-m 1`, one stream per connection, it made no difference +(1.01 sends per request either way). + +### Reproduce + +The command is in the header of `experiments/immediate_completion.cpp`. Arguments: the number of +operations, then `posted` (leave out the modes that nest, so that the count can be large) or +`unwind` (a post after each operation). + +--------------------------------------------------------------------------------------------------- + ## Earlier findings ASIO, single-threaded, before `bench.sh` existed: plain h2load runs against diff --git a/docs/corosio-port-plan.md b/docs/corosio-port-plan.md index 23840d0..e2bc2df 100644 --- a/docs/corosio-port-plan.md +++ b/docs/corosio-port-plan.md @@ -129,7 +129,7 @@ the h3 write path also run under ASAN. From step 5 on, COROSIO's own tests must use it. *2a–2c done 2026-09-30* (`fa8b4d5`..`63d0894`). `runtime.hpp` has `Task`, `Executor`, - `error_code`, `Completion`, `errc`/`errors`, `complete_immediately`/`complete_later`, + `error_code`, `Completion`, `errc`/`errors`, `complete_immediately`/`complete`, `on_cancel`, `run_later`, `dispatch_to`, `new_strand`, `launch`, `when_both`, `Event`, `Timer`, and `io::read_some`/`write`/`receive`/`wait_readable`. Two findings: - `Event::set()` posts in ASIO too, as capy's does, because h3 signals from inside ngtcp2. diff --git a/include/anyhttp/asio/any_async_stream_impl.hpp b/include/anyhttp/asio/any_async_stream_impl.hpp index 8a027a0..21063f8 100644 --- a/include/anyhttp/asio/any_async_stream_impl.hpp +++ b/include/anyhttp/asio/any_async_stream_impl.hpp @@ -6,6 +6,7 @@ // #include "anyhttp/asio/any_async_stream.hpp" +#include "anyhttp/asio/runtime.hpp" #include "anyhttp/net.hpp" #include @@ -37,16 +38,7 @@ class any_async_stream::Impl virtual void async_read_some(ReadWriteHandler handler, MutableBufferVector buffer) = 0; virtual void async_shutdown_impl(ShutdownHandler handler) { -#if 1 - boost::asio::async_immediate(get_executor(), [handler = std::move(handler)]() mutable { - std::move(handler)(boost::system::error_code()); - }); -#else - auto ex = boost::asio::get_associated_immediate_executor(handler, get_executor()); - ex.execute([handler = std::move(handler)]() mutable { // - std::move(handler)(boost::system::error_code()); - }); -#endif + complete_immediately(std::move(handler), get_executor(), boost::system::error_code{}); } }; diff --git a/include/anyhttp/asio/runtime.hpp b/include/anyhttp/asio/runtime.hpp index e145f56..8e6405d 100644 --- a/include/anyhttp/asio/runtime.hpp +++ b/include/anyhttp/asio/runtime.hpp @@ -215,9 +215,15 @@ inline void complete_immediately(Handler&& handler, const Executor& fallback, Ar if (!handler) return; -#if 1 - boost::asio::async_immediate(fallback, [handler = std::forward(handler), - ... args = std::forward(args)]() mutable { // + // + // Not async_immediate(): it asks the handler it is given for its immediate executor, and that is + // the lambda here, which has none -- the one bound to \p handler is never fetched, so the caller + // could not choose to be completed inline. Both post to \p fallback for a handler without one. + // See experiments/immediate_binding.cpp. + // +#if 0 + asio::async_immediate(fallback, [handler = std::forward(handler), + ... args = std::forward(args)]() mutable { // std::move(handler)(std::move(args)...); }); #else @@ -236,7 +242,7 @@ inline void complete_immediately(Handler&& handler, const Executor& fallback, Ar * application inside nghttp2 or ngtcp2 would let it call back into the engine from inside it. */ template -inline void complete_later(Handler&& handler, const Executor& executor, Args&&... args) +inline void complete(Handler&& handler, const Executor& executor, Args&&... args) { boost::asio::post(executor, [handler = std::forward(handler), ... args = std::forward(args)]() mutable { // @@ -247,7 +253,7 @@ inline void complete_later(Handler&& handler, const Executor& executor, Args&&.. /** * Calls \p on_cancel when the caller of the operation \p handler stands for cancels it. The * callback runs on the caller's side, which may be another thread with a strand, so all it may do - * is take the parked handler back and complete it with \ref complete_later. Does nothing if the + * is take the parked handler back and complete it with \ref complete. Does nothing if the * caller has no way of cancelling. */ template @@ -494,7 +500,7 @@ class Event } private: - void wake(error_code ec) { complete_later(std::move(waiter_), executor_, ec); } + void wake(error_code ec) { complete(std::move(waiter_), executor_, ec); } bool set_ = false; Completion waiter_; diff --git a/include/anyhttp/corosio/runtime.hpp b/include/anyhttp/corosio/runtime.hpp index bbfe331..38e9706 100644 --- a/include/anyhttp/corosio/runtime.hpp +++ b/include/anyhttp/corosio/runtime.hpp @@ -239,7 +239,7 @@ inline void complete_immediately(Handler&& handler, const Executor&, Args&&... a } template -inline void complete_later(Handler&& handler, const Executor&, Args&&... args) +inline void complete(Handler&& handler, const Executor&, Args&&... args) { if (handler) std::move(handler)(std::forward(args)...); diff --git a/include/anyhttp/h3/stream.hpp b/include/anyhttp/h3/stream.hpp index 4c69be6..624c80d 100644 --- a/include/anyhttp/h3/stream.hpp +++ b/include/anyhttp/h3/stream.hpp @@ -171,10 +171,15 @@ class Http3Writer : public quic::QuicWriter { if (!stream || stream->closed) { - std::move(handler)(make_error_code(errc::connection_reset)); + std::move(handler)(make_error_code(errc::connection_reset)); // inline like success return; } stream->submit_response(status_code, fields); + // + // Inline, against the rule of complete_immediately(), so that the responses of the requests + // that arrived together go out together, as with HTTP/2 (NGHttp2Writer::async_submit()). + // Queueing the completion cost 16-41% of throughput (h2load, 2026-11-06). + // std::move(handler)(error_code{}); } }; diff --git a/include/anyhttp/quic/stream.hpp b/include/anyhttp/quic/stream.hpp index 0012b41..dfe6d8a 100644 --- a/include/anyhttp/quic/stream.hpp +++ b/include/anyhttp/quic/stream.hpp @@ -282,8 +282,8 @@ class QuicReader : public Interface on_cancel(handler, [this] { if (stream && stream->read_handler) - complete_later(std::move(stream->read_handler), stream->get_executor(), - errors::canceled, size_t{0}); + complete(std::move(stream->read_handler), stream->get_executor(), errors::canceled, + size_t{0}); }); assert(!stream->read_handler); diff --git a/include/anyhttp/runtime.hpp b/include/anyhttp/runtime.hpp index b7d9c03..ab958d6 100644 --- a/include/anyhttp/runtime.hpp +++ b/include/anyhttp/runtime.hpp @@ -7,21 +7,19 @@ // // Apart from the runtime layer (this header, anyhttp/net.hpp and what they include), only the // public API front-ends and the test fixtures test ANYHTTP_ASIO or ANYHTTP_COROSIO: the protocol -// backends are written once, against these names. +// backends are written once, against these names. asio/runtime.hpp declares them in these +// sections and documents each, corosio/runtime.hpp follows it: // -// Task, IoTask, results, Executor, error_code, Completion, -// system_error -// IpAddress, TcpEndpoint, UdpEndpoint -- the runtime's own -// errc (portable error conditions), errors:: (the codes of the API contract), throw_error() -// last_error() -- errno, as an error_code -// complete_immediately(), complete_later(), on_cancel() -- parked operations -// initiate(), launch() -- the two directions between them -// yield_now(), run_later(), dispatch_to(), new_strand(), delay(), when_both(), when_either() -// reset_cancellation(), shielded() -// Event, Timer -// io::read_some(), io::write(), io::peek(), io::receive(), io::wait_readable() -// -// asio/runtime.hpp has the documentation of each. +// types Task, Executor, IpAddress, TcpEndpoint, UdpEndpoint, error_code, +// results, IoTask; asio:: for the buffers, which stay Asio's +// errors errc, errors::, system_error, throw_error(), last_error(), and ::code() of an +// exception_ptr; multithreaded_runtime +// completions Completion, complete_immediately(), complete(), on_cancel(), +// and to and from coroutines: initiate(), launch() +// scheduling yield_now(), run_later(), dispatch_to(), new_strand(), delay(), when_both(), +// when_either(), reset_cancellation(), shielded() +// engines Event, Timer +// I/O io::read_some(), io::write(), io::peek(), io::receive(), io::wait_readable() // #include "anyhttp/config.hpp" diff --git a/src/h2/session.cpp b/src/h2/session.cpp index 802b4b5..bb51a15 100644 --- a/src/h2/session.cpp +++ b/src/h2/session.cpp @@ -501,13 +501,15 @@ void NGHttp2Session::async_submit(SubmitHandler&& handler, std::string_view meth if (!session) { mloge("submit: session already gone!"); - std::move(handler)(errors::canceled, client::Request{nullptr}); + complete_immediately(std::move(handler), get_executor(), errors::canceled, + client::Request{nullptr}); return; } if (!nghttp2_session_check_request_allowed(session)) { mloge("submit: request not allowed!"); - std::move(handler)(errors::canceled, client::Request{nullptr}); + complete_immediately(std::move(handler), get_executor(), errors::canceled, + client::Request{nullptr}); return; } @@ -525,7 +527,8 @@ void NGHttp2Session::async_submit(SubmitHandler&& handler, std::string_view meth if (streams_.size() >= limit) { mlogd("submit: {} streams in progress, peer allows {}", streams_.size(), limit); - std::move(handler)(errors::would_block, client::Request{nullptr}); + complete_immediately(std::move(handler), get_executor(), errors::would_block, + client::Request{nullptr}); return; } @@ -588,7 +591,9 @@ void NGHttp2Session::async_submit(SubmitHandler&& handler, std::string_view meth { mloge("submit: nghttp2_submit_request: ERROR: {}", id); using namespace boost::system; - std::move(handler)(make_error_code(errc::invalid_argument), client::Request{nullptr}); + complete_immediately(std::move(handler), get_executor(), + make_error_code(errc::invalid_argument), client::Request{nullptr}); + return; } stream->id = id; @@ -596,8 +601,8 @@ void NGHttp2Session::async_submit(SubmitHandler&& handler, std::string_view meth logd("[{}] submit: new stream ID: {}", stream->log_prefix(), id); streams_.emplace(id, stream); - complete_later(std::move(handler), get_executor(), error_code{}, - client::Request{std::make_unique>(*stream)}); + complete(std::move(handler), get_executor(), error_code{}, + client::Request{std::make_unique>(*stream)}); start_write(); } diff --git a/src/h2/stream.cpp b/src/h2/stream.cpp index 8107346..a6143d6 100644 --- a/src/h2/stream.cpp +++ b/src/h2/stream.cpp @@ -126,7 +126,7 @@ void NGHttp2Reader::async_read_some(boost::asio::mutable_buffer buffer, on_cancel(handler, [this] { logd("[{}] async_read_some: \x1b[1;31m{}\x1b[0m", stream->log_prefix(), "cancelled"); if (stream->read_handler_) - complete_later(std::move(stream->read_handler_), get_executor(), errors::canceled, 0); + complete(std::move(stream->read_handler_), get_executor(), errors::canceled, 0); }); #endif @@ -254,7 +254,7 @@ void NGHttp2Writer::async_submit(StatusHandler&& handler, unsigned int sta if (!stream) { mlogw("async_submit: stream already gone"); - std::move(handler)(errors::connection_aborted); + std::move(handler)(errors::connection_aborted); // inline like success, see below return; } @@ -311,6 +311,14 @@ void NGHttp2Writer::async_submit(StatusHandler&& handler, unsigned int sta nghttp2_submit_response2(stream->parent.session, stream->id, nva.data(), nva.size(), &prd); stream->parent.start_write(); + // + // Inline, against the rule of complete_immediately(), which a plain callback relies on. The + // requests that arrive together on a connection start their handlers together, and what they + // write goes out in one send if all of them have written before the write pass runs. A queued + // completion, whether ahead of the write pass or behind it, lets the pass run in the middle of + // that: with ten streams per connection, three times the sends, and 19-27% less throughput + // (h2load, 2026-11-06). With one stream it makes no difference. + // std::move(handler)(error_code{}); } @@ -350,7 +358,8 @@ void NGHttp2Writer::async_get_response(client::Request::GetResponseHandler if (!stream) { mlogw("async_get_response: stream already gone"); - std::move(handler)(errors::connection_aborted, client::Response{nullptr}); + complete_immediately(std::move(handler), get_executor(), errors::connection_aborted, + client::Response{nullptr}); } else stream->async_get_response(std::move(handler)); @@ -696,7 +705,7 @@ void NGHttp2Stream::async_write(WriteHandler handler, boost::asio::const_buffer mlogd("async_write: \x1b[1;31m{}\x1b[0m", "cancelled"); // make sure to post this -- otherwise "MAIN COROUTINE DID NOT COMPLETE" happens if (write_handler) - complete_later(std::move(write_handler), get_executor(), errors::canceled); + complete(std::move(write_handler), get_executor(), errors::canceled); }); resume(); @@ -738,8 +747,8 @@ void NGHttp2Stream::async_get_response(client::Request::GetResponseHandler&& han on_cancel(handler, [this] { mlogd("async_get_response: \x1b[1;31m{}\x1b[0m", "cancelled"); if (response_handler) - complete_later(std::move(response_handler), get_executor(), errors::canceled, - client::Response{nullptr}); + complete(std::move(response_handler), get_executor(), errors::canceled, + client::Response{nullptr}); }); response_handler = std::move(handler); diff --git a/src/h2v2/session.cpp b/src/h2v2/session.cpp index acc76a4..63f2d9c 100644 --- a/src/h2v2/session.cpp +++ b/src/h2v2/session.cpp @@ -417,7 +417,8 @@ void NGHttp2Session::async_submit(SubmitHandler&& handler, std::string_view meth if (!conn || closing_) { mloge("submit: session already gone!"); - std::move(handler)(errors::canceled, client::Request{nullptr}); + complete_immediately(std::move(handler), get_executor(), errors::canceled, + client::Request{nullptr}); return; } @@ -489,7 +490,7 @@ void NGHttp2Session::async_submit(SubmitHandler&& handler, std::string_view meth // a request that only has to wait for a free stream is normal back pressure, not a problem logwd(ec != errors::would_block, "[{}] submit: nghttp2_conn_submit_request: {}", stream->log_prefix(), nghttp2_strerror(rv)); - std::move(handler)(ec, client::Request{nullptr}); + complete_immediately(std::move(handler), get_executor(), ec, client::Request{nullptr}); return; } @@ -498,8 +499,8 @@ void NGHttp2Session::async_submit(SubmitHandler&& handler, std::string_view meth logd("[{}] submit: new stream ID: {}", stream->log_prefix(), id); streams_.emplace(id, stream); - complete_later(std::move(handler), get_executor(), error_code{}, - client::Request{std::make_unique>(*stream)}); + complete(std::move(handler), get_executor(), error_code{}, + client::Request{std::make_unique>(*stream)}); start_write(); } diff --git a/src/h2v2/stream.cpp b/src/h2v2/stream.cpp index 7a9ed23..a36a035 100644 --- a/src/h2v2/stream.cpp +++ b/src/h2v2/stream.cpp @@ -122,7 +122,7 @@ void NGHttp2Reader::async_read_some(boost::asio::mutable_buffer buffer, on_cancel(handler, [this] { logd("[{}] async_read_some: \x1b[1;31m{}\x1b[0m", stream->log_prefix(), "cancelled"); if (stream->read_handler_) - complete_later(std::move(stream->read_handler_), get_executor(), errors::canceled, 0); + complete(std::move(stream->read_handler_), get_executor(), errors::canceled, 0); }); assert(!stream->read_handler_); @@ -268,7 +268,7 @@ void NGHttp2Writer::async_submit(StatusHandler&& handler, unsigned int sta if (!stream) { mlogw("async_submit: stream already gone"); - std::move(handler)(errors::connection_aborted); + std::move(handler)(errors::connection_aborted); // inline like success, see below return; } @@ -329,6 +329,10 @@ void NGHttp2Writer::async_submit(StatusHandler&& handler, unsigned int sta stream->parent.start_write(); + // + // Inline, against the rule of complete_immediately(), so that the responses of the requests + // that arrived together go out together, see version 1. + // std::move(handler)(error_code{}); } @@ -368,7 +372,8 @@ void NGHttp2Writer::async_get_response(client::Request::GetResponseHandler if (!stream) { mlogw("async_get_response: stream already gone"); - std::move(handler)(errors::connection_aborted, client::Response{nullptr}); + complete_immediately(std::move(handler), get_executor(), errors::connection_aborted, + client::Response{nullptr}); } else stream->async_get_response(std::move(handler)); @@ -693,7 +698,7 @@ void NGHttp2Stream::async_write(WriteHandler handler, boost::asio::const_buffer // make sure to post this -- otherwise "MAIN COROUTINE DID NOT COMPLETE" happens if (write_handler) - complete_later(std::move(write_handler), get_executor(), errors::canceled); + complete(std::move(write_handler), get_executor(), errors::canceled); }); resume(); @@ -735,8 +740,8 @@ void NGHttp2Stream::async_get_response(client::Request::GetResponseHandler&& han on_cancel(handler, [this] { mlogd("async_get_response: \x1b[1;31m{}\x1b[0m", "cancelled"); if (response_handler) - complete_later(std::move(response_handler), get_executor(), errors::canceled, - client::Response{nullptr}); + complete(std::move(response_handler), get_executor(), errors::canceled, + client::Response{nullptr}); }); response_handler = std::move(handler); diff --git a/src/h3/client.cpp b/src/h3/client.cpp index 566ddb5..af0ccc4 100644 --- a/src/h3/client.cpp +++ b/src/h3/client.cpp @@ -125,7 +125,8 @@ class Http3ClientWriter : public h3::Http3Writer { if (!stream) { - std::move(handler)(make_error_code(errc::connection_aborted), client::Response{nullptr}); + complete_immediately(std::move(handler), this->get_executor(), + make_error_code(errc::connection_aborted), client::Response{nullptr}); return; } static_cast(stream)->async_get_response(std::move(handler)); @@ -281,8 +282,8 @@ void Http3ClientStream::async_get_response(client::Request::GetResponseHandler&& on_cancel(handler, [this] { mlogd("async_get_response: cancelled"); if (response_handler) - complete_later(std::move(response_handler), get_executor(), errors::canceled, - client::Response{nullptr}); + complete(std::move(response_handler), get_executor(), errors::canceled, + client::Response{nullptr}); }); // @@ -339,12 +340,14 @@ void Http3ClientApplication::async_submit(SubmitHandler&& handler, std::string_v if (rv == NGTCP2_ERR_STREAM_ID_BLOCKED) { mlogd("async_submit: no stream available, peer's limit reached"); - std::move(handler)(errors::would_block, client::Request{nullptr}); + complete_immediately(std::move(handler), session_.get_executor(), errors::would_block, + client::Request{nullptr}); return; } mloge("async_submit: ngtcp2_conn_open_bidi_stream: {}", ngtcp2_strerror(rv)); - std::move(handler)(make_error_code(errc::invalid_argument), client::Request{nullptr}); + complete_immediately(std::move(handler), session_.get_executor(), + make_error_code(errc::invalid_argument), client::Request{nullptr}); return; } @@ -352,15 +355,16 @@ void Http3ClientApplication::async_submit(SubmitHandler&& handler, std::string_v if (!stream->submit_request(method, url, headers)) { session_.erase_stream(stream_id); - std::move(handler)(make_error_code(errc::invalid_argument), client::Request{nullptr}); + complete_immediately(std::move(handler), session_.get_executor(), + make_error_code(errc::invalid_argument), client::Request{nullptr}); return; } logd("[{}] async_submit: new stream ID: {}", stream->log_prefix_, stream_id); session_.wake_write(); - complete_later(std::move(handler), session_.get_executor(), error_code{}, - client::Request{std::make_unique(*stream)}); + complete(std::move(handler), session_.get_executor(), error_code{}, + client::Request{std::make_unique(*stream)}); } // ================================================================================================= diff --git a/src/quic/client.cpp b/src/quic/client.cpp index b7dff40..22438a7 100644 --- a/src/quic/client.cpp +++ b/src/quic/client.cpp @@ -421,7 +421,8 @@ void QuicClientSession::async_submit(SubmitHandler&& handler, std::string_view m if (closed() || !ready()) { mloge("async_submit: session not ready"); - std::move(handler)(errors::canceled, client::Request{nullptr}); + complete_immediately(std::move(handler), get_executor(), errors::canceled, + client::Request{nullptr}); return; } application_->async_submit(std::move(handler), method, std::move(url), headers); diff --git a/src/quic/stream.cpp b/src/quic/stream.cpp index 449d814..12e3155 100644 --- a/src/quic/stream.cpp +++ b/src/quic/stream.cpp @@ -318,7 +318,7 @@ void QuicStream::bind_write_cancellation(WriteHandler& handler, uint64_t token) // pointing into freed memory. // mlogd("async_write: \x1b[1;31mcancelled\x1b[0m, FIN still pending"); - complete_later(std::move(write_handler), get_executor(), errors::canceled); + complete(std::move(write_handler), get_executor(), errors::canceled); return; } mlogd("async_write: \x1b[1;31mcancelled\x1b[0m"); @@ -369,7 +369,7 @@ void QuicStream::bind_write_cancellation(WriteHandler& handler, uint64_t token) write_active = false; write_source = {}; // make sure to post this -- otherwise "MAIN COROUTINE DID NOT COMPLETE" happens - complete_later(std::move(write_handler), get_executor(), errors::canceled); + complete(std::move(write_handler), get_executor(), errors::canceled); }); } @@ -678,7 +678,7 @@ void QuicStream::abort_writing(error_code ec) write_active = false; write_source = {}; if (write_handler) - complete_later(std::move(write_handler), get_executor(), ec); + complete(std::move(write_handler), get_executor(), ec); } void QuicStream::maybe_close() diff --git a/src/quic/streams.cpp b/src/quic/streams.cpp index 86db5bc..58cb64f 100644 --- a/src/quic/streams.cpp +++ b/src/quic/streams.cpp @@ -129,7 +129,7 @@ void StreamApplication::on_closed(error_code ec) accept_queue_.clear(); write_queue_.clear(); if (accept_handler_) - complete_later(std::move(accept_handler_), session_.get_executor(), closed_ec_, Stream{}); + complete(std::move(accept_handler_), session_.get_executor(), closed_ec_, Stream{}); } // ------------------------------------------------------------------------------------------------- @@ -166,8 +166,8 @@ void StreamApplication::deliver_accepted() auto stream = std::move(accept_queue_.front()); accept_queue_.pop_front(); // posted: this is reached from inside ngtcp2 - complete_later(std::move(accept_handler_), session_.get_executor(), error_code{}, - make_handle(*stream)); + complete(std::move(accept_handler_), session_.get_executor(), error_code{}, + make_handle(*stream)); } void StreamApplication::async_accept_stream(AcceptStreamHandler&& handler) @@ -196,8 +196,7 @@ void StreamApplication::async_accept_stream(AcceptStreamHandler&& handler) on_cancel(handler, [this] { if (accept_handler_) - complete_later(std::move(accept_handler_), session_.get_executor(), errors::canceled, - Stream{}); + complete(std::move(accept_handler_), session_.get_executor(), errors::canceled, Stream{}); }); accept_handler_ = std::move(handler); } diff --git a/test/test_client_async.cpp b/test/test_client_async.cpp index b3a8e24..3656aaa 100644 --- a/test/test_client_async.cpp +++ b/test/test_client_async.cpp @@ -829,11 +829,24 @@ TEST_P(ClientAsync, Recursion) EXPECT_EQ(s0, s1); #if ANYHTTP_ASIO - // however, ASIO allows us to control this behavior using "immediate executors" - auto ex = co_await this_coro::executor; - co_await response.async_read_some(boost::asio::buffer(empty), bind_immediate_executor(ex)); - auto s2 = stack_remaining_bytes().value(); - EXPECT_GT(s1, s2); + // + // However, ASIO lets the caller choose to be completed inline, by binding an "immediate + // executor": then each read runs on top of the one before. The second read is what tells + // nesting from a path that merely has more frames, which the first would show as well. + // + // Not over HTTP/1.1, which reads in a task of its own (launch()) that starts with a post: + // its reads never complete immediately, so there is nothing for the binding to choose. + // + if (GetParam() != Protocol::h1) + { + auto ex = co_await this_coro::executor; + co_await response.async_read_some(buffer(empty), bind_immediate_executor(ex)); + auto s2 = stack_remaining_bytes().value(); + co_await response.async_read_some(buffer(empty), bind_immediate_executor(ex)); + auto s3 = stack_remaining_bytes().value(); + EXPECT_GT(s1, s2); + EXPECT_GT(s2, s3) << "a bound immediate executor does not complete inline"; + } #endif }; } diff --git a/test/test_corosio.cpp b/test/test_corosio.cpp index ce0adb6..98662d2 100644 --- a/test/test_corosio.cpp +++ b/test/test_corosio.cpp @@ -212,12 +212,12 @@ struct Parked on_cancel(handler, [this] { ++cancelled; if (handler) - complete_later(std::move(handler), Executor{}, errors::canceled, size_t{0}); + complete(std::move(handler), Executor{}, errors::canceled, size_t{0}); }); }); } - void finish(size_t n) { complete_later(std::move(handler), Executor{}, error_code{}, n); } + void finish(size_t n) { complete(std::move(handler), Executor{}, error_code{}, n); } }; Task finish_later(Parked& parked, size_t n) From 72001a5a8593483454803468d269c8784030f36b Mon Sep 17 00:00:00 2001 From: Peter Eisenlohr Date: Sat, 7 Nov 2026 20:05:47 +0000 Subject: [PATCH 4/6] test: the completion-token API, which only ASIO has test_asio.cpp runs async_submit(), async_shutdown() and the rest with as_tuple, cancel_after, the default token and plain callbacks, over all three protocols, and cancel_after on a shutdown that waits for a peer that never answers. test_quic.cpp adds cancel_after on async_accept_stream(). An empty handle has no executor, so a callback completes inside the call and cancel_after cannot start its timer: documented on Reader::get_executor(). Co-Authored-By: Claude Opus 5.5 --- include/anyhttp/quic.hpp | 6 +- include/anyhttp/reader.hpp | 11 +- include/anyhttp/writer.hpp | 3 +- test/test_asio.cpp | 379 +++++++++++++++++++++++++++++++++++++ test/test_quic.cpp | 29 +++ 5 files changed, 423 insertions(+), 5 deletions(-) create mode 100644 test/test_asio.cpp diff --git a/include/anyhttp/quic.hpp b/include/anyhttp/quic.hpp index 1fa59f2..57862ee 100644 --- a/include/anyhttp/quic.hpp +++ b/include/anyhttp/quic.hpp @@ -119,7 +119,8 @@ class Stream : public Reader, public Writer /// The executor type, for what looks for one. using executor_type = Executor; - /// \return The executor of the connection, or an empty one after \c reset(). + /// \return The executor of the connection, or an empty one after \c reset(), with what that + /// means for an operation started then, see \c Reader::get_executor(). executor_type get_executor() const noexcept; /// \return The [quic:address:port.stream] tag the library's log lines carry for this @@ -193,7 +194,8 @@ class Connection /// The executor type, for what looks for one. using executor_type = Executor; - /// \return The executor of the connection, which all of it is to be used on. + /// \return The executor of the connection, which all of it is to be used on, or an empty one + /// for a handle without a connection, see \c Reader::get_executor(). executor_type get_executor() const noexcept; /// \return The [quic:address:port] tag the library's log lines carry for this diff --git a/include/anyhttp/reader.hpp b/include/anyhttp/reader.hpp index c5e428a..3e01476 100644 --- a/include/anyhttp/reader.hpp +++ b/include/anyhttp/reader.hpp @@ -60,8 +60,15 @@ class Reader /// \return Whether this handle has an implementation: not after \c reset(), or being moved from. explicit constexpr operator bool() const noexcept { return static_cast(impl_); } - /// \return The executor of the session this message belongs to, or an empty one after - /// \c reset(). + /** + * \return The executor of the session this message belongs to, or an empty one after + * \c reset(). + * + * Without one, what is started on an empty handle with a completion token cannot complete the + * way ASIO promises: it fails with \c errors::bad_descriptor from inside the call, and a token + * that needs an executor of the operation, as \c cancel_after does for its timer, throws + * \c bad_executor instead. Awaited, it just fails. + */ executor_type get_executor() const noexcept; /// \return The [proto:address:port.stream] tag the library's log lines carry for this diff --git a/include/anyhttp/writer.hpp b/include/anyhttp/writer.hpp index ed3e657..2d7554e 100644 --- a/include/anyhttp/writer.hpp +++ b/include/anyhttp/writer.hpp @@ -55,7 +55,8 @@ class Writer explicit constexpr operator bool() const noexcept { return static_cast(impl_); } /// \return The executor of the session this message belongs to, or an empty one after - /// \c reset(). + /// \c reset(), with what that means for an operation started then, see + /// \c Reader::get_executor(). executor_type get_executor() const noexcept; /// \return The [proto:address:port.stream] tag the library's log lines carry for this diff --git a/test/test_asio.cpp b/test/test_asio.cpp new file mode 100644 index 0000000..7f2ce2a --- /dev/null +++ b/test/test_asio.cpp @@ -0,0 +1,379 @@ +// +// The completion-token spelling of the API, which only ASIO has: async_submit(), async_shutdown() +// and the rest. Underneath, each of them is the operation the coroutine spelling runs, which the +// rest of the suite tests; what is tested here is what the token layer adds -- that the initiations +// work with the tokens a caller would use, a plain callback included, and give cancel_after what it +// needs, an executor for its timer. Empty in a COROSIO build. +// +#include "anyhttp/config.hpp" + +#if ANYHTTP_ASIO + +#include "test_fixtures.hpp" + +#include "anyhttp/net.hpp" +#include "anyhttp/quic.hpp" + +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include + +using namespace testing; + +namespace +{ + +// ================================================================================================= + +/// A timeout for cancel_after() that no operation here comes near: it must never fire. +constexpr auto never = 10s; + +/// Rethrows what a coroutine spawned by a test threw, failing the test. +void rethrow(const std::exception_ptr& ep) +{ + if (ep) + std::rethrow_exception(ep); +} + +/// Reads \p reader to its end, with async_read_some() and \p token. +template +Task read_to_end(Reader& reader, Token token) +{ + std::string body; + std::array buffer; + for (;;) + { + auto [ec, n] = co_await reader.async_read_some(asio::buffer(buffer), token); + body.append(buffer.data(), n); + if (ec == errors::eof) + co_return body; + EXPECT_FALSE(ec) << ec.message(); + if (ec) + co_return body; + } +} + +/// The server's side of an exchange: echoes the request body, with \p token on every operation. +template +Task echo(server::Request request, server::Response response, Token token) +{ + auto body = co_await read_to_end(request, token); + auto [submit_ec] = co_await response.async_submit(200, {}, token); + EXPECT_FALSE(submit_ec) << submit_ec.message(); + auto [write_ec] = co_await response.async_write(asio::buffer(body), token); + EXPECT_FALSE(write_ec) << write_ec.message(); + // + // Not checked: over HTTP/3, the server's writes complete when the peer acknowledges them, and a + // client that has read all of the response may close the connection before it acknowledges the + // last of it. + // + std::ignore = co_await response.async_write_eof(asio::buffer("!", 1), token); +} + +/// The client's side of an exchange, with \p token on every operation, ending with a shutdown. +template +Task exchange(Session& session, boost::urls::url url, Token token) +{ + auto [submit_ec, request] = co_await session.async_submit(url, {}, token); + EXPECT_FALSE(submit_ec) << submit_ec.message(); + auto [write_ec] = co_await request.async_write(asio::buffer("Hello, ", 7), token); + EXPECT_FALSE(write_ec) << write_ec.message(); + auto [eof_ec] = co_await request.async_write_eof(asio::buffer("World", 5), token); + EXPECT_FALSE(eof_ec) << eof_ec.message(); + + auto [response_ec, response] = co_await request.async_get_response(token); + EXPECT_FALSE(response_ec) << response_ec.message(); + EXPECT_EQ(response.status_code(), 200); + auto body = co_await read_to_end(response, token); + EXPECT_EQ(body, "Hello, World!"); + + auto [shutdown_ec] = co_await session.async_shutdown(token); + EXPECT_EQ(shutdown_ec, error_code{}); +} + +} // namespace + +// ================================================================================================= + +class AsioTokens : public ClientAsync +{ +protected: + // + // An operation that has completed cancels its cancel_after() timer. One that completed without + // the timer hearing of it -- its handler dropped, say -- would leave the timer to keep the + // context running until it fires, and the caller waiting for good. + // + void TearDown() override + { + const auto start = std::chrono::steady_clock::now(); + ClientAsync::TearDown(); + EXPECT_LT(std::chrono::steady_clock::now() - start, never / 2); + } +}; + +INSTANTIATE_TEST_SUITE_P(AsioTokens, AsioTokens, ValuesIn(protocols()), NameGenerator); + +// ------------------------------------------------------------------------------------------------- + +TEST_P(AsioTokens, WHEN_every_operation_takes_as_tuple_THEN_both_sides_see_all_of_the_body) +{ + requestHandler = [](server::Request request, server::Response response) { + return echo(std::move(request), std::move(response), as_tuple); + }; + clientSession = [this](Session session) -> Task { + co_await exchange(session, url, as_tuple); + }; +} + +TEST_P(AsioTokens, WHEN_every_operation_takes_cancel_after_THEN_results_pass_through) +{ + requestHandler = [](server::Request request, server::Response response) { + return echo(std::move(request), std::move(response), cancel_after(never, as_tuple)); + }; + clientSession = [this](Session session) -> Task { + co_await exchange(session, url, cancel_after(never, as_tuple)); + }; +} + +TEST_P(AsioTokens, WHEN_connect_and_get_take_cancel_after_THEN_results_pass_through) +{ + clientSession = [this](Session) -> Task { + auto [connect_ec, session] = co_await client->async_connect(cancel_after(never, as_tuple)); + EXPECT_FALSE(connect_ec) << connect_ec.message(); + auto [get_ec, message] = + co_await session.async_get(url.set_path("missing"), {}, cancel_after(never, as_tuple)); + EXPECT_FALSE(get_ec) << get_ec.message(); + EXPECT_EQ(message.result_int(), 404); + }; +} + +// +// The default token gives the result itself and throws the error. A request with a method takes +// no body, so it is complete as soon as it is submitted. +// +TEST_P(AsioTokens, WHEN_default_token_is_used_THEN_errors_are_thrown) +{ + requestHandler = [](server::Request, server::Response response) -> Task { + co_await response.async_submit(200, {}); + std::ignore = co_await response.async_write_eof(as_tuple); + }; + clientSession = [this](Session session) -> Task { + auto request = co_await session.async_submit("GET", url, {}); + auto response = co_await request.async_get_response(); + EXPECT_EQ(response.status_code(), 200); + co_await session.async_shutdown(); + + error_code ec; + try + { + co_await session.async_submit("GET", url, {}); + } + catch (const boost::system::system_error& e) + { + ec = e.code(); + } + EXPECT_TRUE(ec) << "a session that has been shut down takes no more requests"; + }; +} + +// +// ASIO's guarantee: a handler is not called from inside the function that starts the operation, +// not even when there is nothing left to wait for. A caller that holds a lock around the call, or +// starts the next operation from the handler, relies on it. +// +TEST_P(AsioTokens, WHEN_callback_is_given_THEN_it_is_not_called_before_the_operation_returns) +{ + clientSession = [this](Session session) -> Task { + std::optional shutdown_result; + session.async_shutdown([&](error_code ec) { shutdown_result = ec; }); + EXPECT_FALSE(shutdown_result) << "completed inside async_shutdown()"; + while (!shutdown_result) + co_await yield(); + EXPECT_EQ(*shutdown_result, error_code{}); + + // + // A session that has been shut down fails a request right away: nothing to wait for. + // + std::optional submit_result; + session.async_submit(url, {}, [&](error_code ec, client::Request) { submit_result = ec; }); + EXPECT_FALSE(submit_result) << "completed inside async_submit()"; + while (!submit_result) + co_await yield(); + EXPECT_TRUE(*submit_result); + }; +} + +// +// Over HTTP/2 and HTTP/3, a response's submit completes inside the call on purpose, see +// NGHttp2Writer::async_submit(): a callback there cannot rely on ASIO's guarantee. +// +TEST_P(AsioTokens, WHEN_server_gives_a_callback_THEN_it_is_not_called_before_the_operation_returns) +{ + if (GetParam() != Protocol::h1) + GTEST_SKIP() << "completes inside async_submit(), which costs less than posting it"; + + requestHandler = [](server::Request, server::Response response) -> Task { + std::optional result; + response.async_submit(200, {}, [&](error_code ec) { result = ec; }); + EXPECT_FALSE(result) << "completed inside async_submit()"; + while (!result) + co_await yield(); + EXPECT_EQ(*result, error_code{}); + std::ignore = co_await response.async_write_eof(as_tuple); + }; + clientSession = [this](Session session) -> Task { + auto [ec, message] = co_await session.async_get(url, {}, as_tuple); + EXPECT_FALSE(ec) << ec.message(); + EXPECT_EQ(message.result_int(), 200); + }; +} + +// ================================================================================================= + +// +// A peer that takes what the client sends and never ends its side, so that a shutdown, after +// sending its own end, waits for the peer's in vain: for its "close_notify" over TLS, for its FIN +// in any case -- two seconds each, before the shutdown gives up on it. Over HTTP/2 it sends its +// SETTINGS first, which is all the client needs to see of a server. +// +class AsioTokensSilentPeer : public TestWithParam> +{ +protected: + void SetUp() override { setup_logging(); } + + Protocol protocol() const { return std::get<0>(GetParam()); } + bool tls() const { return std::get<1>(GetParam()); } + + Task serve() + { + auto socket = co_await acceptor.async_accept(use_awaitable); + if (tls()) + { + auto& stream = tls_peer.emplace(std::move(socket), tls_context); + co_await stream.async_handshake(asio::ssl::stream_base::server, use_awaitable); + co_await read_on(stream); + } + else + co_await read_on(peer.emplace(std::move(socket))); + } + + /// Reads until the client has ended its side, and leaves the connection as it is. + template + Task read_on(Stream& stream) + { + if (protocol() == Protocol::h2) + co_await asio::async_write(stream, asio::buffer("\0\0\0\x04\0\0\0\0\0"sv), use_awaitable); + + std::array buffer; + for (;;) + { + auto [ec, n] = co_await stream.async_read_some(asio::buffer(buffer), as_tuple); + if (ec) + break; + } + } + + boost::urls::url url() const + { + boost::urls::url url(tls() ? "https://127.0.0.2/" : "http://127.0.0.2/"); + url.set_port_number(acceptor.local_endpoint().port()); + return url; + } + + IoContext context; + tcp::acceptor acceptor{context, {asio::ip::make_address("127.0.0.2"), 0}}; + asio::ssl::context tls_context = make_server_tls_context(server::Config{}.tls_certificate_chain, + server::Config{}.tls_private_key); + std::optional peer; + std::optional> tls_peer; +}; + +INSTANTIATE_TEST_SUITE_P(AsioTokensSilentPeer, AsioTokensSilentPeer, + Combine(Values(Protocol::h1, Protocol::h2), Bool()), [](const auto& info) { + return to_string(std::get<0>(info.param)) + + (std::get<1>(info.param) ? "_TLS" : "_Cleartext"); + }); + +// +// The caller's cancellation reaches the wait, and the session takes no more requests afterwards. +// What the shutdown reports then, and whether the connection is closed by the time it does, is not +// settled yet: it differs between the protocols, and between cleartext and TLS. +// +TEST_P(AsioTokensSilentPeer, WHEN_shutdown_takes_cancel_after_THEN_it_ends_the_wait) +{ + co_spawn(context, serve(), detached); + co_spawn( + context, + [this]() -> Task { + client::Client client(context.get_executor(), + client::Config{.url = url(), + .protocol = protocol(), + .tls_ca_file = "pki/out/root.pem"}); + auto [connect_ec, session] = co_await client.async_connect(as_tuple); + EXPECT_FALSE(connect_ec) << connect_ec.message(); + + const auto start = std::chrono::steady_clock::now(); + std::ignore = co_await session.async_shutdown(cancel_after(100ms, as_tuple)); + EXPECT_LT(std::chrono::steady_clock::now() - start, 1s) + << "cancel_after did not reach the wait"; + + auto [submit_ec, request] = co_await session.async_submit(url(), {}, as_tuple); + EXPECT_TRUE(submit_ec) << "a session that has been shut down takes no more requests"; + + acceptor.close(); + }, + rethrow); + context.run(); +} + +// ================================================================================================= + +// +// A handle without an implementation -- default-constructed, moved from or reset -- fails what is +// started on it with bad_descriptor. It has no executor, though: a plain callback is called from +// inside the call, and a token that needs an executor from the operation, as cancel_after() does +// for its timer, throws bad_executor before anything has started. +// +class AsioTokensEmptyHandle : public Test +{ +protected: + void SetUp() override { setup_logging(); } + + IoContext context; + client::Request request; + client::Response response; + quic::Connection connection; + std::array buffer; +}; + +TEST_F(AsioTokensEmptyHandle, WHEN_awaited_THEN_operations_fail_with_bad_descriptor) +{ + co_spawn( + context, + [this]() -> Task { + auto [read_ec, n] = co_await response.async_read_some(asio::buffer(buffer), as_tuple); + EXPECT_EQ(read_ec, errors::bad_descriptor); + auto [write_ec] = co_await request.async_write(asio::buffer(buffer), as_tuple); + EXPECT_EQ(write_ec, errors::bad_descriptor); + auto [eof_ec] = co_await request.async_write_eof(as_tuple); + EXPECT_EQ(eof_ec, errors::bad_descriptor); + auto [response_ec, r] = co_await request.async_get_response(as_tuple); + EXPECT_EQ(response_ec, errors::bad_descriptor); + auto [accept_ec, stream] = co_await connection.async_accept_stream(as_tuple); + EXPECT_EQ(accept_ec, errors::bad_descriptor); + }, + rethrow); + context.run(); +} + +// ================================================================================================= + +#endif diff --git a/test/test_quic.cpp b/test/test_quic.cpp index 49c9a34..825cfe8 100644 --- a/test/test_quic.cpp +++ b/test/test_quic.cpp @@ -543,6 +543,35 @@ TEST_P(QuicStreams, WHEN_using_completion_tokens_THEN_the_same_works) }()); } +// +// An accept that nothing arrives for ends with cancel_after(), and the connection goes on +// accepting: the stream the client opens later is not lost. +// +TEST_P(QuicStreams, WHEN_accept_stream_takes_cancel_after_THEN_it_can_be_cancelled) +{ + serverConnection = [](quic::Connection connection) -> Task { + auto [ec, none] = co_await connection.async_accept_stream(cancel_after(100ms, as_tuple)); + EXPECT_EQ(ec, errc::operation_canceled); + + auto [accept_ec, stream] = + co_await connection.async_accept_stream(cancel_after(10s, as_tuple)); + EXPECT_FALSE(accept_ec) << accept_ec.message(); + auto [read_ec, body] = co_await read_all(stream); + co_await stream.async_write_eof(asio::buffer(body), use_awaitable); + std::ignore = co_await connection.async_accept_stream(as_tuple(use_awaitable)); + }; + + run([&]() -> Task { + auto connection = co_await quic::async_connect(context.get_executor(), "127.0.0.2", port(), + client_config(), use_awaitable); + co_await delay(200ms); // the server's first accept has given up by now + auto [ec, stream] = connection.open_stream(); + co_await stream.async_write_eof(asio::buffer("late", 4), use_awaitable); + auto [read_ec, body] = co_await read_all(stream); + EXPECT_EQ(body, "late"); + }()); +} + #endif // ================================================================================================= From 35814afd6650ccdf4d6b834be3abc55cc7dbbc26 Mon Sep 17 00:00:00 2001 From: Peter Eisenlohr Date: Sat, 7 Nov 2026 20:47:08 +0000 Subject: [PATCH 5/6] test: spell Boost's asio in full in the completion-token tests Since the namespaces are named after their directories, a bare asio:: in a test is anyhttp's inline runtime namespace, and test_asio.cpp and test_quic.cpp no longer compiled after the rebase onto master. Co-Authored-By: Claude Opus 5.5 --- test/test_asio.cpp | 30 ++++++++++++++++-------------- test/test_quic.cpp | 4 ++-- 2 files changed, 18 insertions(+), 16 deletions(-) diff --git a/test/test_asio.cpp b/test/test_asio.cpp index 7f2ce2a..c20d637 100644 --- a/test/test_asio.cpp +++ b/test/test_asio.cpp @@ -51,7 +51,7 @@ Task read_to_end(Reader& reader, Token token) std::array buffer; for (;;) { - auto [ec, n] = co_await reader.async_read_some(asio::buffer(buffer), token); + auto [ec, n] = co_await reader.async_read_some(boost::asio::buffer(buffer), token); body.append(buffer.data(), n); if (ec == errors::eof) co_return body; @@ -68,14 +68,14 @@ Task echo(server::Request request, server::Response response, Token token) auto body = co_await read_to_end(request, token); auto [submit_ec] = co_await response.async_submit(200, {}, token); EXPECT_FALSE(submit_ec) << submit_ec.message(); - auto [write_ec] = co_await response.async_write(asio::buffer(body), token); + auto [write_ec] = co_await response.async_write(boost::asio::buffer(body), token); EXPECT_FALSE(write_ec) << write_ec.message(); // // Not checked: over HTTP/3, the server's writes complete when the peer acknowledges them, and a // client that has read all of the response may close the connection before it acknowledges the // last of it. // - std::ignore = co_await response.async_write_eof(asio::buffer("!", 1), token); + std::ignore = co_await response.async_write_eof(boost::asio::buffer("!", 1), token); } /// The client's side of an exchange, with \p token on every operation, ending with a shutdown. @@ -84,9 +84,9 @@ Task exchange(Session& session, boost::urls::url url, Token token) { auto [submit_ec, request] = co_await session.async_submit(url, {}, token); EXPECT_FALSE(submit_ec) << submit_ec.message(); - auto [write_ec] = co_await request.async_write(asio::buffer("Hello, ", 7), token); + auto [write_ec] = co_await request.async_write(boost::asio::buffer("Hello, ", 7), token); EXPECT_FALSE(write_ec) << write_ec.message(); - auto [eof_ec] = co_await request.async_write_eof(asio::buffer("World", 5), token); + auto [eof_ec] = co_await request.async_write_eof(boost::asio::buffer("World", 5), token); EXPECT_FALSE(eof_ec) << eof_ec.message(); auto [response_ec, response] = co_await request.async_get_response(token); @@ -258,7 +258,7 @@ class AsioTokensSilentPeer : public TestWithParam> if (tls()) { auto& stream = tls_peer.emplace(std::move(socket), tls_context); - co_await stream.async_handshake(asio::ssl::stream_base::server, use_awaitable); + co_await stream.async_handshake(boost::asio::ssl::stream_base::server, use_awaitable); co_await read_on(stream); } else @@ -270,12 +270,13 @@ class AsioTokensSilentPeer : public TestWithParam> Task read_on(Stream& stream) { if (protocol() == Protocol::h2) - co_await asio::async_write(stream, asio::buffer("\0\0\0\x04\0\0\0\0\0"sv), use_awaitable); + co_await boost::asio::async_write(stream, boost::asio::buffer("\0\0\0\x04\0\0\0\0\0"sv), + use_awaitable); std::array buffer; for (;;) { - auto [ec, n] = co_await stream.async_read_some(asio::buffer(buffer), as_tuple); + auto [ec, n] = co_await stream.async_read_some(boost::asio::buffer(buffer), as_tuple); if (ec) break; } @@ -289,11 +290,11 @@ class AsioTokensSilentPeer : public TestWithParam> } IoContext context; - tcp::acceptor acceptor{context, {asio::ip::make_address("127.0.0.2"), 0}}; - asio::ssl::context tls_context = make_server_tls_context(server::Config{}.tls_certificate_chain, - server::Config{}.tls_private_key); + tcp::acceptor acceptor{context, {boost::asio::ip::make_address("127.0.0.2"), 0}}; + boost::asio::ssl::context tls_context = make_server_tls_context( + server::Config{}.tls_certificate_chain, server::Config{}.tls_private_key); std::optional peer; - std::optional> tls_peer; + std::optional> tls_peer; }; INSTANTIATE_TEST_SUITE_P(AsioTokensSilentPeer, AsioTokensSilentPeer, @@ -359,9 +360,10 @@ TEST_F(AsioTokensEmptyHandle, WHEN_awaited_THEN_operations_fail_with_bad_descrip co_spawn( context, [this]() -> Task { - auto [read_ec, n] = co_await response.async_read_some(asio::buffer(buffer), as_tuple); + auto [read_ec, n] = + co_await response.async_read_some(boost::asio::buffer(buffer), as_tuple); EXPECT_EQ(read_ec, errors::bad_descriptor); - auto [write_ec] = co_await request.async_write(asio::buffer(buffer), as_tuple); + auto [write_ec] = co_await request.async_write(boost::asio::buffer(buffer), as_tuple); EXPECT_EQ(write_ec, errors::bad_descriptor); auto [eof_ec] = co_await request.async_write_eof(as_tuple); EXPECT_EQ(eof_ec, errors::bad_descriptor); diff --git a/test/test_quic.cpp b/test/test_quic.cpp index 825cfe8..7636bc8 100644 --- a/test/test_quic.cpp +++ b/test/test_quic.cpp @@ -557,7 +557,7 @@ TEST_P(QuicStreams, WHEN_accept_stream_takes_cancel_after_THEN_it_can_be_cancell co_await connection.async_accept_stream(cancel_after(10s, as_tuple)); EXPECT_FALSE(accept_ec) << accept_ec.message(); auto [read_ec, body] = co_await read_all(stream); - co_await stream.async_write_eof(asio::buffer(body), use_awaitable); + co_await stream.async_write_eof(boost::asio::buffer(body), use_awaitable); std::ignore = co_await connection.async_accept_stream(as_tuple(use_awaitable)); }; @@ -566,7 +566,7 @@ TEST_P(QuicStreams, WHEN_accept_stream_takes_cancel_after_THEN_it_can_be_cancell client_config(), use_awaitable); co_await delay(200ms); // the server's first accept has given up by now auto [ec, stream] = connection.open_stream(); - co_await stream.async_write_eof(asio::buffer("late", 4), use_awaitable); + co_await stream.async_write_eof(boost::asio::buffer("late", 4), use_awaitable); auto [read_ec, body] = co_await read_all(stream); EXPECT_EQ(body, "late"); }()); From 480de6ec0b4e0695d51a0baff872ef925b1a736e Mon Sep 17 00:00:00 2001 From: Peter Eisenlohr Date: Sat, 7 Nov 2026 20:47:08 +0000 Subject: [PATCH 6/6] test: shut down a stuck write only once the peer's window has closed WHEN_write_is_stuck_THEN_shutdown_interrupts_it_and_says_so/HTTP2 failed about one run in four under gtest-parallel -w 100. It called shutdown() 50ms in, taking the write to be stuck by then, but a peer that stops reading keeps a small window open for some 200ms on loopback, and a client that got little CPU had not filled the pipe yet: its DATA frame completed, the GOAWAY fit in after it, and the session ended cleanly -- correctly reported as success. The peer now waits for its advertised receive window to be 0 on two samples 10ms apart before it lets the client shut down: that window stays closed, and the client has filled its send buffer behind it. A receive queue that has stopped growing is no proof of that, it stands still for tens of milliseconds at a time. tcpi_rcv_wnd is read in tcp_info.cpp, as clashes with the Asio includes, and glibc's struct tcp_info ends before that field. Co-Authored-By: Claude Opus 5.5 --- test/CMakeLists.txt | 2 ++ test/tcp_info.cpp | 28 ++++++++++++++++++++++++++++ test/test_client_tls.cpp | 33 ++++++++++++++++++++++++++++++++- 3 files changed, 62 insertions(+), 1 deletion(-) create mode 100644 test/tcp_info.cpp diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index 2496ba5..f3a5777 100644 --- a/test/CMakeLists.txt +++ b/test/CMakeLists.txt @@ -43,6 +43,8 @@ endif() add_executable(test_all ${SRC_FILES}) # not a test: lets TSAN see weak_ptr's synchronization, see there target_sources(test_all PRIVATE tsan_libcxx.cpp) +# not a test either: what a test needs of , which clashes with Asio's includes +target_sources(test_all PRIVATE tcp_info.cpp) target_link_libraries(test_all PRIVATE GTest::gtest_main GTest::gmock) target_link_libraries(test_all PRIVATE spdlog::spdlog_header_only) diff --git a/test/tcp_info.cpp b/test/tcp_info.cpp new file mode 100644 index 0000000..900a1ec --- /dev/null +++ b/test/tcp_info.cpp @@ -0,0 +1,28 @@ +// +// The receive window a TCP socket has advertised last. Linux reports it in TCP_INFO, but glibc's +// struct tcp_info ends before that field, and the kernel's, in , cannot share a +// translation unit with , which Asio includes. Hence a file of its own. +// +#include +#include +#include + +#include +#include +#include + +namespace anyhttp +{ + +/// The receive window that socket \p fd has advertised last, in bytes, if the kernel says. +std::optional advertised_receive_window(int fd) +{ + tcp_info info{}; + socklen_t size = sizeof(info); + if (::getsockopt(fd, IPPROTO_TCP, TCP_INFO, &info, &size) != 0 || + size < offsetof(tcp_info, tcpi_rcv_wnd) + sizeof(info.tcpi_rcv_wnd)) + return std::nullopt; + return info.tcpi_rcv_wnd; +} + +} // namespace anyhttp diff --git a/test/test_client_tls.cpp b/test/test_client_tls.cpp index ebf49e5..02a451d 100644 --- a/test/test_client_tls.cpp +++ b/test/test_client_tls.cpp @@ -30,6 +30,9 @@ namespace anyhttp // ================================================================================================= +/// The receive window that socket \p fd has advertised last, if the kernel says, see tcp_info.cpp. +std::optional advertised_receive_window(int fd); + namespace { @@ -312,6 +315,8 @@ class ClientTlsEnd : public TestWithParam // if (then == Then::stall) { + co_await until_full(stream); + stalled.set(); co_await delay(h2() ? 2200ms : 200ms); break; } @@ -323,6 +328,31 @@ class ClientTlsEnd : public TestWithParam peer_done.set(); } + /// + /// Returns once \p stream has closed its receive window to the client, on two samples in a row. + /// The window stays closed, as nothing is read any more, and the second sample gives the client + /// the time to fill its send buffer behind it: then its write is stuck for good. + /// + /// A fixed time is not enough for that: the client may get to run so little that its write is + /// still under way when shutdown() is called, and the GOAWAY fits in after it. Neither is a + /// receive queue that has stopped growing: it does so for tens of milliseconds at a time. + /// + Task until_full(TlsStream& stream) + { + const auto fd = get_socket(stream).native_handle(); + for (int closed = 0; closed < 2;) + { + co_await delay(10ms); + auto window = advertised_receive_window(fd); + if (!window) + { + ADD_FAILURE() << "the kernel does not report the receive window"; + break; + } + closed = *window == 0 ? closed + 1 : 0; + } + } + /// Whether a frame of \p type has arrived, of the HTTP/2 the peer has received. bool received_frame(uint8_t type) const { @@ -351,6 +381,7 @@ class ClientTlsEnd : public TestWithParam boost::urls::url target; std::string received; error_code end; + Event stalled; ///< set once the client's write is stuck, with Then::stall Event peer_done; }; @@ -434,7 +465,7 @@ TEST_P(ClientTlsEnd, WHEN_write_is_stuck_THEN_shutdown_interrupts_it_and_says_so std::tie(write_result) = co_await request.write(boost::asio::buffer(body)); }; auto end_it = [&]() -> Task { - co_await delay(50ms); + co_await stalled.wait(); std::tie(result) = co_await session.shutdown(); }; co_await when_both(write(), end_it());