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", 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.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/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/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/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/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/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/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_asio.cpp b/test/test_asio.cpp new file mode 100644 index 0000000..c20d637 --- /dev/null +++ b/test/test_asio.cpp @@ -0,0 +1,381 @@ +// +// 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(boost::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(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(boost::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(boost::asio::buffer("Hello, ", 7), token); + EXPECT_FALSE(write_ec) << write_ec.message(); + 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); + 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(boost::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 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(boost::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, {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; +}; + +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(boost::asio::buffer(buffer), as_tuple); + EXPECT_EQ(read_ec, errors::bad_descriptor); + 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); + 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_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_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()); 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) diff --git a/test/test_quic.cpp b/test/test_quic.cpp index 49c9a34..7636bc8 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(boost::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(boost::asio::buffer("late", 4), use_awaitable); + auto [read_ec, body] = co_await read_all(stream); + EXPECT_EQ(body, "late"); + }()); +} + #endif // ================================================================================================= 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"; +} + +// =================================================================================================