From 65febe7d49d56fe4b8f80e34e7b4c204d98c5068 Mon Sep 17 00:00:00 2001 From: Peter Eisenlohr Date: Sun, 8 Nov 2026 19:53:26 +0000 Subject: [PATCH 1/5] fix: only terminal cancellation cancels a parked operation on ASIO There is one kind of cancellation: it completes the operation right away, whatever that leaves behind, the same as a stop request on COROSIO. That is Asio's terminal type. A request for partial or total alone promises a state that a cancelled write cannot always keep, so on_cancel() now ignores it, which is what Asio asks of an operation that does not support a type, and what its own coroutines do by default. cancel_after sends terminal and || sends all three, so neither notices. Co-Authored-By: Claude Opus 5.5 --- include/anyhttp/asio/runtime.hpp | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/include/anyhttp/asio/runtime.hpp b/include/anyhttp/asio/runtime.hpp index 8e6405d..3bf8aea 100644 --- a/include/anyhttp/asio/runtime.hpp +++ b/include/anyhttp/asio/runtime.hpp @@ -255,16 +255,25 @@ inline void complete(Handler&& handler, const Executor& executor, Args&&... args * 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. Does nothing if the * caller has no way of cancelling. + * + * Only a request with the \c terminal bit counts: that is the one kind of cancellation there is, + * the same as a stop request on COROSIO, and it completes right away, whatever it leaves behind. A + * request for \c partial or \c total alone promises a state that is not always there to keep, so it + * is ignored, which is what Asio asks of an operation that does not support a type. Asio's own + * coroutines filter the same way by default, \c cancel_after sends \c terminal and \c || sends all + * three. */ template inline void on_cancel(Completion& handler, F&& on_cancel) { auto slot = boost::asio::get_associated_cancellation_slot(handler); if (slot.is_connected() && !slot.has_handler()) - slot.assign( - [on_cancel = std::forward(on_cancel)](boost::asio::cancellation_type_t) mutable { // + slot.assign([on_cancel = std::forward(on_cancel)]( + boost::asio::cancellation_type_t type) mutable { // + if ((type & boost::asio::cancellation_type::terminal) != + boost::asio::cancellation_type::none) on_cancel(); - }); + }); } // ------------------------------------------------------------------------------------------------- From f82a6d7cc9a8cf351155872d8e461765189b942d Mon Sep 17 00:00:00 2001 From: Peter Eisenlohr Date: Sun, 8 Nov 2026 19:54:41 +0000 Subject: [PATCH 2/5] fix: a write says how much of its buffer it has taken A cancelled write completes right away, and so far said nothing but that. Where the message is still usable afterwards, the caller cannot go on without knowing where the write stopped, so Write is void(error_code, size_t) now, and write() and write_eof() yield IoTask: all of the buffer on success, none when the entry ladder answers, and otherwise what the writer has taken -- copied by nghttp2 version 1, copied in Staged and framed or handed to the stack in ZeroCopy on HTTP/2 version 2 and HTTP/3, and over HTTP/1.1 the body bytes that reached the connection, which h1::write_some() counts among the header and chunk framing Beast hands out. On COROSIO, Writer::write_some() passes that on, where it reported 0 for any error. Cancelling write_eof() before nghttp2 has been told about the end now takes the end back on HTTP/2, as HTTP/3 already did, so that write_eof(buffer + n) finishes the body after any cancelled write, on every protocol. Until now a data write after it failed with broken_pipe there, and only a bare re-end was possible. The callers in the library, the programs, the tests and the snippets take the count; send() counts what was actually taken. Co-Authored-By: Claude Opus 5.5 --- docs/snippets.cpp | 14 ++++---- include/anyhttp/common.hpp | 6 ++-- include/anyhttp/h1/io.hpp | 41 +++++++++++++++++++---- include/anyhttp/h2/stream.hpp | 12 +++++-- include/anyhttp/h2v2/stream.hpp | 11 +++++-- include/anyhttp/quic/stream.hpp | 14 +++++++- include/anyhttp/request_handlers.hpp | 11 +++++-- include/anyhttp/writer.hpp | 22 ++++++------- include/anyhttp/writer_impl.hpp | 6 +++- src/apps/client_main.cpp | 2 +- src/h1/session.cpp | 17 +++++----- src/h2/session.cpp | 3 +- src/h2/stream.cpp | 33 +++++++++++-------- src/h2v2/session.cpp | 8 +++-- src/h2v2/stream.cpp | 43 ++++++++++++++----------- src/quic/stream.cpp | 24 ++++++++------ src/request_handlers.cpp | 5 +-- src/session.cpp | 2 +- src/writer.cpp | 8 ++--- test/test_asio.cpp | 13 +++++--- test/test_client_async.cpp | 24 +++++++------- test/test_client_async_cancellation.cpp | 4 +-- test/test_client_tls.cpp | 4 +-- test/test_coroutine_api.cpp | 8 ++--- test/test_get.cpp | 2 +- test/test_headers.cpp | 2 +- test/test_quic.cpp | 31 +++++++++--------- 27 files changed, 233 insertions(+), 137 deletions(-) diff --git a/docs/snippets.cpp b/docs/snippets.cpp index 8f157cb..1c03522 100644 --- a/docs/snippets.cpp +++ b/docs/snippets.cpp @@ -64,7 +64,8 @@ Task serve_echo(server::Request request, server::Response response) if (ec) co_return; - if (auto [write_ec] = co_await response.write(boost::asio::buffer(buffer, n)); write_ec) + if (auto [write_ec, written] = co_await response.write(boost::asio::buffer(buffer, n)); + write_ec) co_return; } @@ -133,7 +134,7 @@ IoTask post(Session& session, boost::urls::url url, std::string bod if (submit_ec) co_return std::tuple{submit_ec, std::string{}}; - if (auto [ec] = co_await request.write_eof(boost::asio::buffer(body)); ec) + if (auto [ec, written] = co_await request.write_eof(boost::asio::buffer(body)); ec) co_return std::tuple{ec, std::string{}}; auto [response_ec, response] = co_await request.get_response(); @@ -189,10 +190,10 @@ Task read_at_most(Session session, boost::urls::url url, size_t limit) Task upload(Writer& writer, std::string_view body) { for (; !body.empty(); body.remove_prefix(std::min(body.size(), size_t{16_k}))) - if (auto [ec] = co_await writer.write(boost::asio::buffer(body.substr(0, 16_k))); ec) + if (auto [ec, written] = co_await writer.write(boost::asio::buffer(body.substr(0, 16_k))); ec) throw_error(ec); - if (auto [ec] = co_await writer.write_eof(); ec) + if (auto [ec, written] = co_await writer.write_eof(); ec) throw_error(ec); } @@ -358,7 +359,8 @@ Task quic_echo(quic::Stream stream) if (ec) co_return; - if (auto [write_ec] = co_await stream.write(boost::asio::buffer(buffer, n)); write_ec) + if (auto [write_ec, written] = co_await stream.write(boost::asio::buffer(buffer, n)); + write_ec) co_return; } std::ignore = co_await stream.write_eof(); @@ -393,7 +395,7 @@ IoTask quic_hello(Executor executor, std::string host, std::string if (open_ec) co_return std::tuple{open_ec, std::string{}}; - if (auto [ec] = co_await stream.write_eof(boost::asio::buffer("hello", 5)); ec) + if (auto [ec, written] = co_await stream.write_eof(boost::asio::buffer("hello", 5)); ec) co_return std::tuple{ec, std::string{}}; co_return co_await read_all(stream); diff --git a/include/anyhttp/common.hpp b/include/anyhttp/common.hpp index fe4ce7d..84082b5 100644 --- a/include/anyhttp/common.hpp +++ b/include/anyhttp/common.hpp @@ -139,8 +139,10 @@ using ReadSomeHandler = Completion; using WriteSome = void(error_code, size_t); using WriteSomeHandler = Completion; -/// The completion signature of \c Writer::async_write and \ref Writer::async_write_eof. -using Write = void(error_code); +/// The completion signature of \c Writer::async_write and \ref Writer::async_write_eof, with the +/// error and how many bytes of the buffer the writer has taken -- all of them, unless it failed or +/// was cancelled. Those are the caller's no longer, so what is left to write starts after them. +using Write = void(error_code, size_t); using WriteHandler = Completion; /// The completion signature of an operation that has nothing to report but how it went. diff --git a/include/anyhttp/h1/io.hpp b/include/anyhttp/h1/io.hpp index 74191bf..a0d6dc2 100644 --- a/include/anyhttp/h1/io.hpp +++ b/include/anyhttp/h1/io.hpp @@ -156,15 +156,28 @@ IoTask read_message(Stream& stream, DynamicBuffer& buffer, // ------------------------------------------------------------------------------------------------- +/** + * Counts how many of the bytes written come from \c range, the caller's body buffer, which the + * serializer hands out by reference among the header and the chunk framing. That is how much of + * the body a write that stops short has taken. + */ +struct BodyWatch +{ + boost::asio::const_buffer range; + size_t written = 0; +}; + /** * Writes what \p serializer has to offer next, as Beast's \c http::async_write_some(). Yields the - * number of bytes written, which is not the size of the body that went with them. + * number of bytes written, which is not the size of the body that went with them: \p watch, if + * given, counts those. * * With a \c buffer_body, a serializer that has used up the body buffer it was given offers nothing * but \c http::error::need_buffer, which is passed on here. */ template -IoTask write_some(Stream& stream, http::serializer& serializer) +IoTask write_some(Stream& stream, http::serializer& serializer, + BodyWatch* watch = nullptr) { if (serializer.is_done()) co_return std::tuple{error_code{}, size_t{0}}; @@ -186,6 +199,21 @@ IoTask write_some(Stream& stream, http::serializer(watch->range.data()); + const auto* end = begin + watch->range.size(); + size_t left = n; + for (const auto& buffer : buffers) + { + const auto size = std::min(left, buffer.size()); + const auto* data = static_cast(buffer.data()); + if (size && data >= begin && data + buffer.size() <= end) + watch->written += size; + left -= size; + } + } co_return std::tuple{write_ec, n}; } @@ -194,12 +222,13 @@ IoTask write_some(Stream& stream, http::serializer -IoTask write_until(Stream& stream, Serializer& serializer, Predicate done) +IoTask write_until(Stream& stream, Serializer& serializer, Predicate done, + BodyWatch* watch = nullptr) { size_t written = 0; while (!done(serializer)) { - auto [ec, n] = co_await h1::write_some(stream, serializer); + auto [ec, n] = co_await h1::write_some(stream, serializer, watch); written += n; if (ec == to_error_code(http::error::need_buffer)) break; @@ -219,9 +248,9 @@ IoTask write_header(Stream& stream, Serializer& serializer) /// Writes all there is, as Beast's \c http::async_write() with a serializer. template -IoTask write(Stream& stream, Serializer& serializer) +IoTask write(Stream& stream, Serializer& serializer, BodyWatch* watch = nullptr) { - return h1::write_until(stream, serializer, [](auto& sr) { return sr.is_done(); }); + return h1::write_until(stream, serializer, [](auto& sr) { return sr.is_done(); }, watch); } /// Writes all of \p message, as Beast's \c http::async_write() with a message. diff --git a/include/anyhttp/h2/stream.hpp b/include/anyhttp/h2/stream.hpp index 97fb1b8..f84fbfd 100644 --- a/include/anyhttp/h2/stream.hpp +++ b/include/anyhttp/h2/stream.hpp @@ -141,7 +141,14 @@ class NGHttp2Stream : public std::enable_shared_from_this // async_write() // boost::asio::const_buffer write_buffer; // undefined unless write_handler is set + size_t write_size = 0; // of the write in progress, so it can say what it took WriteHandler write_handler; + /// How much of the write in progress nghttp2 has copied, which is what it completes with when + /// it stops short. + size_t write_taken() const noexcept + { + return write_handler ? write_size - write_buffer.size() : 0; + } bool is_deferred = false; // @@ -149,8 +156,9 @@ class NGHttp2Stream : public std::enable_shared_from_this // producer callback turns that into NGHTTP2_DATA_FLAG_EOF on the very DATA frame that carries // the write's last bytes -- so a body that ends with data needs no second, empty frame. // \c eof_requested says the user has ended the body (only a bare re-end is accepted after - // that), \c eof_submitted that nghttp2 has actually been told; between the two lies a - // cancelled async_write_eof(), whose re-issue delivers the still-owed EOF flag. + // that), \c eof_submitted that nghttp2 has actually been told. Cancelling an + // async_write_eof() before nghttp2 has been told takes the intent back, so the body stays open + // for the rest of the buffer, as it does over HTTP/3. // bool eof_requested = false; bool eof_submitted = false; diff --git a/include/anyhttp/h2v2/stream.hpp b/include/anyhttp/h2v2/stream.hpp index 6e7281c..2d6a48c 100644 --- a/include/anyhttp/h2v2/stream.hpp +++ b/include/anyhttp/h2v2/stream.hpp @@ -188,6 +188,12 @@ class NGHttp2Stream : public std::enable_shared_from_this std::vector write_chunk_; // Staged: the piece of the body nghttp2 is pointing into size_t offered_ = 0; size_t written_ = 0; + /// How much of the write in progress is taken, what it completes with when it stops short: + /// copied out of the caller's buffer in Staged, framed from it in ZeroCopy. + size_t write_taken() const noexcept + { + return write_handler ? (write_mode == WriteMode::Staged ? offered_ : written_) : 0; + } bool is_deferred = false; // @@ -202,8 +208,9 @@ class NGHttp2Stream : public std::enable_shared_from_this // data_reader() turns that into NGHTTP2_READ_DATA_FLAG_EOF on the very DATA frame that carries // the write's last bytes -- so a body that ends with data needs no second, empty frame. // \c eof_requested says the user has ended the body (only a bare re-end is accepted after - // that), \c eof_submitted that nghttp2 has actually been told; between the two lies a - // cancelled async_write_eof(), whose re-issue delivers the still-owed EOF flag. + // that), \c eof_submitted that nghttp2 has actually been told. Cancelling an + // async_write_eof() before nghttp2 has been told takes the intent back, so the body stays open + // for the rest of the buffer, as it does over HTTP/3. // bool eof_requested = false; bool eof_submitted = false; diff --git a/include/anyhttp/quic/stream.hpp b/include/anyhttp/quic/stream.hpp index dfe6d8a..256cf0c 100644 --- a/include/anyhttp/quic/stream.hpp +++ b/include/anyhttp/quic/stream.hpp @@ -125,6 +125,18 @@ class QuicStream : public std::enable_shared_from_this // uint64_t write_base = 0; // where write_source begins size_t write_copied = 0; // Staged: how much of write_source went into chunks + + /// How much of the active write is taken, what it completes with when it stops short: copied + /// into chunks in Staged, handed to the stack in ZeroCopy. + size_t write_taken() const noexcept + { + if (!write_active) + return 0; + if (write_mode == WriteMode::Staged) + return write_copied; + const uint64_t size = boost::asio::buffer_size(write_source); + return tx_offered > write_base ? size_t(std::min(tx_offered - write_base, size)) : 0; + } uint64_t tx_queued = 0; uint64_t tx_offered = 0; uint64_t tx_sent = 0; @@ -366,7 +378,7 @@ class QuicWriter : public Interface ec = empty ? error_code{} : make_error_code(errc::broken_pipe); else ec = make_error_code(errc::connection_reset); - complete_immediately(std::move(handler), executor, ec); + complete_immediately(std::move(handler), executor, ec, size_t{0}); } void detach() override diff --git a/include/anyhttp/request_handlers.hpp b/include/anyhttp/request_handlers.hpp index 71d05ba..5009443 100644 --- a/include/anyhttp/request_handlers.hpp +++ b/include/anyhttp/request_handlers.hpp @@ -91,8 +91,12 @@ template Task send(Writer& request, Range range) { logd("send: (contiguous range)..."); - if (auto [ec] = co_await request.write(boost::asio::buffer(range.data(), range.size())); ec) + if (auto [ec, n] = co_await request.write(boost::asio::buffer(range.data(), range.size())); ec) + { + logd("[{}] send: (contiguous range) {} after {} of {} bytes", request.log_prefix(), + ec.message(), n, range.size()); throw_error(ec); + } logd("send: (contiguous range)... done"); } @@ -113,7 +117,9 @@ Task send(Writer& request, Range range) for (; it != last && out != buffer.end(); ++it, ++out) *out = *it; const auto n = static_cast(out - buffer.begin()); - if (auto [ec] = co_await request.write(boost::asio::buffer(buffer.data(), n)); ec) + auto [ec, taken] = co_await request.write(boost::asio::buffer(buffer.data(), n)); + bytes += taken; + if (ec) { if (ec == errc::operation_canceled) logd("[{}] send: (range) {} after {} bytes", request.log_prefix(), ec.message(), bytes); @@ -122,7 +128,6 @@ Task send(Writer& request, Range range) ec.message(), bytes); throw_error(ec); } - bytes += n; } logd("send: (range) sent {} bytes", bytes); diff --git a/include/anyhttp/writer.hpp b/include/anyhttp/writer.hpp index 2d7554e..19e18b6 100644 --- a/include/anyhttp/writer.hpp +++ b/include/anyhttp/writer.hpp @@ -132,21 +132,22 @@ class Writer // with either runtime -- the library's own request handlers, the shared tests -- uses this. // - /// Writes \p buffer as part of the body: auto [ec] = co_await writer.write(buffer); + /// Writes \p buffer as part of the body: auto [ec, n] = co_await writer.write(buffer); /// \param buffer What to write. - /// \return The error. - IoTask<> write(boost::asio::const_buffer buffer); + /// \return The error, and how many bytes of \p buffer the writer has taken, see \c Write. + IoTask write(boost::asio::const_buffer buffer); - /// Writes \p buffer, if any, and ends the body: auto [ec] = co_await + /// Writes \p buffer, if any, and ends the body: auto [ec, n] = co_await /// writer.write_eof(); /// \param buffer What to write. - /// \return The error. - IoTask<> write_eof(boost::asio::const_buffer buffer = {}); + /// \return The error, and how many bytes of \p buffer the writer has taken, see \c Write. + IoTask write_eof(boost::asio::const_buffer buffer = {}); #if ANYHTTP_COROSIO /** - * Writes the first non-empty buffer of \p buffers as part of the body, and yields its size: - * this is what makes a Writer a \c capy::WriteStream. The body stays open, as with write(). + * Writes the first non-empty buffer of \p buffers as part of the body, and yields how much of + * it the writer has taken: this is what makes a Writer a \c capy::WriteStream. The body stays + * open, as with write(). */ template IoTask write_some(Buffers buffers) @@ -154,10 +155,7 @@ class Writer for (auto it = capy::buffer_sequence_begin(buffers); it != capy::buffer_sequence_end(buffers); ++it) if (capy::const_buffer buffer = *it; buffer.size() > 0) - { - auto [ec] = co_await write(boost::asio::const_buffer(buffer.data(), buffer.size())); - co_return std::tuple{ec, ec ? size_t{0} : buffer.size()}; - } + co_return co_await write(boost::asio::const_buffer(buffer.data(), buffer.size())); co_return std::tuple{error_code{}, size_t{0}}; } #endif diff --git a/include/anyhttp/writer_impl.hpp b/include/anyhttp/writer_impl.hpp index a951af5..53ba5df 100644 --- a/include/anyhttp/writer_impl.hpp +++ b/include/anyhttp/writer_impl.hpp @@ -43,10 +43,14 @@ class Writer::Impl : public std::enable_shared_from_this // while re-ending it with no data attached is an idempotent no-op. Only then do stream-level // failures (closed, cancelled) get their say. // + // The size it completes with is how much of \p buffer the writer has taken: all of it on + // success, none when the entry ladder answers, and what it got to otherwise -- copied, framed + // or on the wire, whichever means the caller must not hand it over again. + // virtual void async_write(WriteHandler&& handler, boost::asio::const_buffer buffer, bool eof) = 0; /// async_write() as a coroutine, see Reader::Impl::read_some(). - virtual IoTask<> write(boost::asio::const_buffer buffer, bool eof) + virtual IoTask write(boost::asio::const_buffer buffer, bool eof) { co_return co_await initiate([this, buffer, eof](WriteHandler handler) { // async_write(std::move(handler), buffer, eof); diff --git a/src/apps/client_main.cpp b/src/apps/client_main.cpp index a7e2e67..59432fd 100644 --- a/src/apps/client_main.cpp +++ b/src/apps/client_main.cpp @@ -27,7 +27,7 @@ using IoContext = boost::asio::io_context; static Task send(client::Request& request, std::string_view body) { logd("send: sending string of {} bytes...", body.size()); - auto [ec] = co_await request.write_eof(boost::asio::buffer(body)); + auto [ec, written] = co_await request.write_eof(boost::asio::buffer(body)); if (ec) throw_error(ec); logd("send: sending string of {} bytes... done", body.size()); diff --git a/src/h1/session.cpp b/src/h1/session.cpp index 230d5b0..c56b673 100644 --- a/src/h1/session.cpp +++ b/src/h1/session.cpp @@ -314,7 +314,7 @@ class WriterBase : public Parent launch(get_executor(), write(buffer, eof), std::move(handler)); } - IoTask<> write(boost::asio::const_buffer buffer, bool eof) override + IoTask write(boost::asio::const_buffer buffer, bool eof) override { const bool empty = buffer.size() == 0; @@ -326,25 +326,25 @@ class WriterBase : public Parent // Only then does it matter that the session, and with it the stream, may be gone. // if (empty && !eof) - co_return std::tuple{error_code{}}; + co_return std::tuple{error_code{}, size_t{0}}; if (eof_submitted) { if (!empty) mloge("async_write: body has already been ended"); - co_return std::tuple{empty ? error_code{} : make_error_code(errc::broken_pipe)}; + co_return std::tuple{empty ? error_code{} : make_error_code(errc::broken_pipe), size_t{0}}; } if (!session) { mlogw("async_write: session already gone"); - co_return std::tuple{errors::connection_aborted}; + co_return std::tuple{errors::connection_aborted, size_t{0}}; } if (cancelled) { mloge("async_write: already canceled"); - co_return std::tuple{errors::canceled}; + co_return std::tuple{errors::canceled, size_t{0}}; } assert(!writing); @@ -377,13 +377,14 @@ class WriterBase : public Parent auto self = Parent::shared_from_this(); const auto expected = buffer.size(); - // 'n' is the number of bytes written to the stream, not the number taken from the buffer + // 'n' is the number of bytes written to the stream, 'body' counts those taken from the buffer error_code ec; size_t n = 0; + h1::BodyWatch body{buffer}; try { const auto op = session->write_op(); - std::tie(ec, n) = co_await h1::write(stream, serializer); + std::tie(ec, n) = co_await h1::write(stream, serializer, &body); } catch (const system_error& e) // a cancelled coroutine throws from its next co_await { @@ -447,7 +448,7 @@ class WriterBase : public Parent // effect immediately, not only when this task is gone. // self.reset(); - co_return std::tuple{ec}; + co_return std::tuple{ec, ec ? body.written : expected}; } /// Called when a write has ended the body, before its handler is invoked. diff --git a/src/h2/session.cpp b/src/h2/session.cpp index bb51a15..50d3fa7 100644 --- a/src/h2/session.cpp +++ b/src/h2/session.cpp @@ -668,7 +668,8 @@ void NGHttp2Session::close_stream(int32_t stream_id) if (stream->write_handler) { // Use the same error code as reported by the underlying TCP connection used in HTTP/1.1. - swap_and_invoke(stream->write_handler, make_error_code(errc::connection_reset)); + const auto taken = stream->write_taken(); + swap_and_invoke(stream->write_handler, make_error_code(errc::connection_reset), taken); } // diff --git a/src/h2/stream.cpp b/src/h2/stream.cpp index a6143d6..26937d8 100644 --- a/src/h2/stream.cpp +++ b/src/h2/stream.cpp @@ -349,7 +349,7 @@ void NGHttp2Writer::async_write(WriteHandler&& handler, boost::asio::const mlogw("async_write: stream already gone"); ec = errors::connection_aborted; } - complete_immediately(std::move(handler), executor, ec); + complete_immediately(std::move(handler), executor, ec, size_t{0}); } template @@ -632,7 +632,7 @@ NGHttp2Stream::~NGHttp2Stream() if (write_handler) { mlogw("Stream: dtor... cancelling async_write()"); - swap_and_invoke(write_handler, errors::canceled); + swap_and_invoke(write_handler, errors::canceled, write_taken()); } mlogd("\x1b[33mStream: dtor... done\x1b[0m"); @@ -652,7 +652,7 @@ void NGHttp2Stream::async_write(WriteHandler handler, boost::asio::const_buffer // if (empty && !eof) { - complete_immediately(std::move(handler), get_executor(), error_code{}); + complete_immediately(std::move(handler), get_executor(), error_code{}, size_t{0}); return; } @@ -662,20 +662,20 @@ void NGHttp2Stream::async_write(WriteHandler handler, boost::asio::const_buffer { mloge("async_write: body has already been ended"); complete_immediately(std::move(handler), get_executor(), - make_error_code(errc::broken_pipe)); + make_error_code(errc::broken_pipe), size_t{0}); return; } // // A bare re-end. If nghttp2 already knows about the end, there is nothing left to do. If - // not -- a cancelled async_write_eof() leaves the intent standing but takes its handler - // away, so the flag never reached the producer callback -- fall through to the normal path - // below: the re-issued write picks up where the cancelled one left off, which is what lets - // an upload be ended once the peer's flow control window reopens. + // not, fall through to the normal path below, which delivers the flag. (A cancelled + // async_write_eof() no longer gets here: it takes its intent back, so re-ending after it is + // an ordinary first end, which is what lets an upload be ended once the peer's flow control + // window reopens.) // if (eof_submitted) { - complete_immediately(std::move(handler), get_executor(), error_code{}); + complete_immediately(std::move(handler), get_executor(), error_code{}, size_t{0}); return; } } @@ -689,7 +689,7 @@ void NGHttp2Stream::async_write(WriteHandler handler, boost::asio::const_buffer { mlogw("async_write: stream already closed"); complete_immediately(std::move(handler), get_executor(), - make_error_code(errc::connection_reset)); + make_error_code(errc::connection_reset), size_t{0}); return; } @@ -698,14 +698,21 @@ void NGHttp2Stream::async_write(WriteHandler handler, boost::asio::const_buffer mlogd("async_write: buffer={} eof={} is_deferred={}", buffer.size(), eof, is_deferred); write_buffer = buffer; + write_size = buffer.size(); eof_requested |= eof; write_handler = std::move(handler); on_cancel(write_handler, [this] { mlogd("async_write: \x1b[1;31m{}\x1b[0m", "cancelled"); + // + // What nghttp2 has copied stays taken, the rest is the caller's again. An end of the body + // that nghttp2 has not been told about is taken back with it, see eof_requested. + // + if (!eof_submitted) + eof_requested = false; // make sure to post this -- otherwise "MAIN COROUTINE DID NOT COMPLETE" happens if (write_handler) - complete(std::move(write_handler), get_executor(), errors::canceled); + complete(std::move(write_handler), get_executor(), errors::canceled, write_taken()); }); resume(); @@ -808,7 +815,7 @@ ssize_t NGHttp2Stream::producer_callback(uint8_t* buf, size_t length, uint32_t* } mlogd("write callback: running handler..."); - swap_and_invoke(write_handler, error_code{}); + swap_and_invoke(write_handler, error_code{}, write_size); if (write_handler) mlogd("write callback: running handler... done -- RESPAWNED"); else @@ -824,7 +831,7 @@ ssize_t NGHttp2Stream::producer_callback(uint8_t* buf, size_t length, uint32_t* mlogd("write callback: EOF"); eof_submitted = true; *data_flags |= NGHTTP2_DATA_FLAG_EOF; - swap_and_invoke(write_handler, error_code{}); + swap_and_invoke(write_handler, error_code{}, size_t{0}); } return copied; diff --git a/src/h2v2/session.cpp b/src/h2v2/session.cpp index 63f2d9c..32e647f 100644 --- a/src/h2v2/session.cpp +++ b/src/h2v2/session.cpp @@ -590,9 +590,11 @@ void NGHttp2Session::close_stream(int64_t stream_id) // if (stream->write_handler) { - swap_and_invoke(stream->write_handler, stream->write_cancelled_ - ? errors::canceled - : make_error_code(errc::connection_reset)); + const auto taken = stream->write_taken(); + swap_and_invoke(stream->write_handler, + stream->write_cancelled_ ? errors::canceled + : make_error_code(errc::connection_reset), + taken); } // diff --git a/src/h2v2/stream.cpp b/src/h2v2/stream.cpp index a36a035..71dc20e 100644 --- a/src/h2v2/stream.cpp +++ b/src/h2v2/stream.cpp @@ -363,7 +363,7 @@ void NGHttp2Writer::async_write(WriteHandler&& handler, boost::asio::const mlogw("async_write: stream already gone"); ec = errors::connection_aborted; } - complete_immediately(std::move(handler), executor, ec); + complete_immediately(std::move(handler), executor, ec, size_t{0}); } template @@ -600,7 +600,7 @@ NGHttp2Stream::~NGHttp2Stream() if (write_handler) { mlogw("Stream: dtor... cancelling async_write()"); - swap_and_invoke(write_handler, errors::canceled); + swap_and_invoke(write_handler, errors::canceled, write_taken()); } mlogd("\x1b[33mStream: dtor... done\x1b[0m"); @@ -620,7 +620,7 @@ void NGHttp2Stream::async_write(WriteHandler handler, boost::asio::const_buffer // if (empty && !eof) { - complete_immediately(std::move(handler), get_executor(), error_code{}); + complete_immediately(std::move(handler), get_executor(), error_code{}, size_t{0}); return; } @@ -630,20 +630,20 @@ void NGHttp2Stream::async_write(WriteHandler handler, boost::asio::const_buffer { mloge("async_write: body has already been ended"); complete_immediately(std::move(handler), get_executor(), - make_error_code(errc::broken_pipe)); + make_error_code(errc::broken_pipe), size_t{0}); return; } // // A bare re-end. If nghttp2 already knows about the end, there is nothing left to do. If - // not -- a cancelled async_write_eof() leaves the intent standing but takes its handler - // away, so the flag never reached the data reader -- fall through to the normal path below: - // the re-issued write picks up where the cancelled one left off, which is what lets an - // upload be ended once the peer's flow control window reopens. + // not, fall through to the normal path below, which delivers the flag. (A cancelled + // async_write_eof() no longer gets here: it takes its intent back, so re-ending after it is + // an ordinary first end, which is what lets an upload be ended once the peer's flow control + // window reopens.) // if (eof_submitted) { - complete_immediately(std::move(handler), get_executor(), error_code{}); + complete_immediately(std::move(handler), get_executor(), error_code{}, size_t{0}); return; } } @@ -657,7 +657,7 @@ void NGHttp2Stream::async_write(WriteHandler handler, boost::asio::const_buffer { mlogw("async_write: stream already closed"); complete_immediately(std::move(handler), get_executor(), - make_error_code(errc::connection_reset)); + make_error_code(errc::connection_reset), size_t{0}); return; } @@ -696,9 +696,17 @@ void NGHttp2Stream::async_write(WriteHandler handler, boost::asio::const_buffer return; } + // + // Otherwise the write stops where it is, and the stream goes on. What it has taken stays + // taken, the rest is the caller's again, and so is an end of the body nghttp2 has not been + // told about, see eof_requested. + // + if (!eof_submitted) + eof_requested = false; + // make sure to post this -- otherwise "MAIN COROUTINE DID NOT COMPLETE" happens if (write_handler) - complete(std::move(write_handler), get_executor(), errors::canceled); + complete(std::move(write_handler), get_executor(), errors::canceled, write_taken()); }); resume(); @@ -761,10 +769,9 @@ nghttp2_ssize NGHttp2Stream::data_reader(nghttp2_vec* vec, size_t veccnt, uint32 assert(!is_deferred); // - // No write in progress. Nothing may be handed over, and above all the body must not be ended - // here: a cancelled async_write_eof() leaves \c eof_requested standing without a handler, and it - // is the write that re-issues it which gets to end the body -- otherwise END_STREAM would go out - // for a write that reported itself cancelled. Version 1 deferred here for the same reason. + // No write in progress. Nothing may be handed over, and the body must not be ended here: only + // a write gets to end it -- otherwise END_STREAM could go out for a write that reported itself + // cancelled, which takes its intent back. Version 1 deferred here for the same reason. // // Returning no data without the EOF flag counts as a callback failure, hence WOULDBLOCK: the // stream waits until resume() wakes it. @@ -800,7 +807,7 @@ nghttp2_ssize NGHttp2Stream::data_reader(nghttp2_vec* vec, size_t veccnt, uint32 mlogd("write callback: EOF"); eof_submitted = true; *pflags |= NGHTTP2_READ_DATA_FLAG_EOF; - swap_and_invoke(write_handler, error_code{}); + swap_and_invoke(write_handler, error_code{}, size_t{0}); return 0; } @@ -851,7 +858,7 @@ nghttp2_ssize NGHttp2Stream::data_reader(nghttp2_vec* vec, size_t veccnt, uint32 if (last && write_mode == WriteMode::Staged) { mlogd("write callback: copied the last {} bytes, running handler...", offer); - swap_and_invoke(write_handler, error_code{}); + swap_and_invoke(write_handler, error_code{}, write_buffer.size()); } else mlogd("write callback: offered {} bytes, {} left", offer, write_buffer.size() - offered_); @@ -879,7 +886,7 @@ void NGHttp2Stream::on_write_offset(size_t len) return; mlogd("write offset: running handler..."); - swap_and_invoke(write_handler, error_code{}); + swap_and_invoke(write_handler, error_code{}, write_buffer.size()); if (write_handler) mlogd("write offset: running handler... done -- RESPAWNED"); else diff --git a/src/quic/stream.cpp b/src/quic/stream.cpp index 12e3155..4295109 100644 --- a/src/quic/stream.cpp +++ b/src/quic/stream.cpp @@ -47,7 +47,10 @@ void QuicStream::detach_all() if (read_handler) swap_and_invoke(read_handler, make_error_code(errc::connection_reset), 0); if (write_active && write_handler) - swap_and_invoke(write_handler, make_error_code(errc::connection_reset)); + { + const auto taken = write_taken(); + swap_and_invoke(write_handler, make_error_code(errc::connection_reset), taken); + } } Executor QuicStream::get_executor() const noexcept { return session.get_executor(); } @@ -198,7 +201,7 @@ void QuicStream::start_write(WriteHandler&& handler, boost::asio::const_buffer b mlogd("start_write: n={} eof={}", n, eof); auto complete_immediately = [&](error_code ec) { // - anyhttp::complete_immediately(std::move(handler), get_executor(), ec); + anyhttp::complete_immediately(std::move(handler), get_executor(), ec, size_t{0}); }; // @@ -318,10 +321,11 @@ 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(std::move(write_handler), get_executor(), errors::canceled); + complete(std::move(write_handler), get_executor(), errors::canceled, size_t{0}); return; } mlogd("async_write: \x1b[1;31mcancelled\x1b[0m"); + const auto taken = write_taken(); // before the bookkeeping below moves on if (write_mode == WriteMode::ZeroCopy) { @@ -369,7 +373,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(std::move(write_handler), get_executor(), errors::canceled); + complete(std::move(write_handler), get_executor(), errors::canceled, taken); }); } @@ -515,6 +519,7 @@ void QuicStream::finish_active_write() // nothing points into it any more: every byte acknowledged (ZeroCopy), copied out (Staged), or // no bytes at all (the EOF marker). // + const auto taken = boost::asio::buffer_size(write_source); write_source = {}; auto handler = std::move(write_handler); write_active = false; @@ -530,9 +535,8 @@ void QuicStream::finish_active_write() // write pass this is nested in and at worst trips ngtcp2's own "time must not go backwards" // assertion. Post instead -- one hop, on a path that is not latency critical. // - run_later(get_executor(), [self = shared_from_this(), handler = std::move(handler)]() mutable { - std::move(handler)(error_code{}); - }); + run_later(get_executor(), [self = shared_from_this(), handler = std::move(handler), + taken]() mutable { std::move(handler)(error_code{}, taken); }); } // ================================================================================================= @@ -573,10 +577,11 @@ void QuicStream::fail(error_code ec) // if (write_active) { + const auto taken = write_taken(); write_active = false; write_source = {}; if (write_handler) - swap_and_invoke(write_handler, ec ? ec : make_error_code(errc::connection_reset)); + swap_and_invoke(write_handler, ec ? ec : make_error_code(errc::connection_reset), taken); } maybe_close(); @@ -675,10 +680,11 @@ void QuicStream::abort_writing(error_code ec) // buffer is free -- in ZeroCopy as much as in Staged mode. Posted, as this is reached from // inside ngtcp2. // + const auto taken = write_taken(); write_active = false; write_source = {}; if (write_handler) - complete(std::move(write_handler), get_executor(), ec); + complete(std::move(write_handler), get_executor(), ec, taken); } void QuicStream::maybe_close() diff --git a/src/request_handlers.cpp b/src/request_handlers.cpp index fad1276..36729da 100644 --- a/src/request_handlers.cpp +++ b/src/request_handlers.cpp @@ -103,7 +103,8 @@ Task echo(server::Request request, server::Response response) if (ec) co_return; // a body cut short, see not ending ours below - if (auto [write_ec] = co_await response.write(boost::asio::buffer(buffer, n)); write_ec) + if (auto [write_ec, written] = co_await response.write(boost::asio::buffer(buffer, n)); + write_ec) co_return; } @@ -279,7 +280,7 @@ Task> try_read_response(client::Request& request) Task send_eof(Writer& writer) { - if (auto [ec] = co_await writer.write_eof(); ec) + if (auto [ec, n] = co_await writer.write_eof(); ec) throw_error(ec); } diff --git a/src/session.cpp b/src/session.cpp index 6b8944a..7aa3b37 100644 --- a/src/session.cpp +++ b/src/session.cpp @@ -114,7 +114,7 @@ IoTask get_message(std::shared_ptr session, boos if (ec) co_return failed(ec); - if (auto [ec] = co_await request.write_eof(); ec) + if (auto [ec, n] = co_await request.write_eof(); ec) co_return failed(ec); auto [response_ec, response] = co_await request.get_response(); diff --git a/src/writer.cpp b/src/writer.cpp index f8e5485..44f4d54 100644 --- a/src/writer.cpp +++ b/src/writer.cpp @@ -37,14 +37,14 @@ void Writer::content_length(std::optional content_length) return impl_ ? impl_->content_length(content_length) : void{}; } -static IoTask<> no_writer() { co_return std::tuple{errors::bad_descriptor}; } +static IoTask no_writer() { co_return std::tuple{errors::bad_descriptor, size_t{0}}; } -IoTask<> Writer::write(boost::asio::const_buffer buffer) +IoTask Writer::write(boost::asio::const_buffer buffer) { return impl_ ? impl_->write(buffer, false) : no_writer(); } -IoTask<> Writer::write_eof(boost::asio::const_buffer buffer) +IoTask Writer::write_eof(boost::asio::const_buffer buffer) { return impl_ ? impl_->write(buffer, true) : no_writer(); } @@ -55,7 +55,7 @@ void Writer::async_write_any(WriteHandler&& handler, boost::asio::const_buffer b if (impl_) impl_->async_write(std::move(handler), buffer, eof); else - std::move(handler)(errors::bad_descriptor); + std::move(handler)(errors::bad_descriptor, size_t{0}); } #endif diff --git a/test/test_asio.cpp b/test/test_asio.cpp index c20d637..747bf43 100644 --- a/test/test_asio.cpp +++ b/test/test_asio.cpp @@ -68,7 +68,7 @@ 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); + auto [write_ec, written] = 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 @@ -84,9 +84,11 @@ 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); + auto [write_ec, written] = + 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); + auto [eof_ec, eof_written] = + 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); @@ -363,9 +365,10 @@ TEST_F(AsioTokensEmptyHandle, WHEN_awaited_THEN_operations_fail_with_bad_descrip 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); + auto [write_ec, written] = + 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); + auto [eof_ec, eof_written] = 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); diff --git a/test/test_client_async.cpp b/test/test_client_async.cpp index 3656aaa..9edc350 100644 --- a/test/test_client_async.cpp +++ b/test/test_client_async.cpp @@ -165,10 +165,10 @@ TEST_P(ClientAsync, WHEN_session_is_gone_THEN_request_reports_error) auto request = check(co_await session.submit(url.set_path("echo"), {})); session.reset(); - auto [ec] = co_await request.write(boost::asio::buffer("Hello"sv)); + auto [ec, written] = co_await request.write(boost::asio::buffer("Hello"sv)); EXPECT_TRUE(is_connection_error(ec)) << what(ec); - std::tie(ec) = co_await request.write_eof(); + std::tie(ec, written) = co_await request.write_eof(); EXPECT_TRUE(is_connection_error(ec)) << what(ec); auto [ec2, response] = co_await request.get_response(); @@ -206,7 +206,7 @@ TEST_P(ClientAsync, WHEN_server_session_is_gone_THEN_response_reports_error) auto [ec] = co_await response.submit(200, {}); EXPECT_TRUE(is_connection_error(ec)) << what(ec); - std::tie(ec) = co_await response.write(boost::asio::buffer("Hello"sv)); + std::tie(ec, std::ignore) = co_await response.write(boost::asio::buffer("Hello"sv)); EXPECT_TRUE(is_connection_error(ec)) << what(ec); responded->notify(); @@ -311,7 +311,7 @@ TEST_P(ClientAsync, request1.reset(); session.reset(); - auto [ec] = co_await request2.write(boost::asio::buffer("Hello"sv)); + auto [ec, written] = co_await request2.write(boost::asio::buffer("Hello"sv)); EXPECT_TRUE(is_connection_error(ec)) << what(ec); }; } @@ -593,16 +593,16 @@ TEST_P(ClientAsync, WHEN_written_after_eof_THEN_reports_broken_pipe) check(co_await response.submit(200, fields({{"Content-Length", hello.size()}}))); check(co_await response.write_eof(boost::asio::buffer(hello))); - auto [ec] = co_await response.write_eof(); + auto [ec, written] = co_await response.write_eof(); EXPECT_FALSE(ec); - std::tie(ec) = co_await response.write({}); + std::tie(ec, written) = co_await response.write({}); EXPECT_FALSE(ec); - std::tie(ec) = co_await response.write(boost::asio::buffer(hello)); + std::tie(ec, written) = co_await response.write(boost::asio::buffer(hello)); EXPECT_EQ(ec, errc::broken_pipe); - std::tie(ec) = co_await response.write_eof(boost::asio::buffer(hello)); + std::tie(ec, written) = co_await response.write_eof(boost::asio::buffer(hello)); EXPECT_EQ(ec, errc::broken_pipe); }; clientSession = [this](Session session) -> Task { @@ -671,9 +671,9 @@ TEST_P(ClientAsync, WHEN_server_cancels_write_eof_THEN_client_sees_truncated_bod // this is over, so the write is guaranteed to still be in progress when it is cancelled. // #if ANYHTTP_COROSIO - auto [ec] = co_await stop_after(50ms, response.write_eof(boost::asio::buffer(body))); + auto [ec, written] = co_await stop_after(50ms, response.write_eof(boost::asio::buffer(body))); #else - auto [ec] = + auto [ec, written] = co_await response.async_write_eof(boost::asio::buffer(body), cancel_after(50ms, as_tuple)); #endif EXPECT_EQ(ec, errc::operation_canceled); @@ -1146,9 +1146,9 @@ TEST_P(ClientAsync, WHEN_request_has_no_body_THEN_it_is_complete_after_submit) auto request2 = check(co_await session.submit(url.set_path("echo"), {})); check(co_await request2.write_eof(boost::asio::buffer(body2))); - auto [ec] = co_await request1.write_eof(); + auto [ec, written] = co_await request1.write_eof(); EXPECT_FALSE(ec) << what(ec); - std::tie(ec) = co_await request1.write(boost::asio::buffer(body1)); + std::tie(ec, written) = co_await request1.write(boost::asio::buffer(body1)); EXPECT_EQ(ec, errc::broken_pipe) << what(ec); auto response1 = check(co_await request1.get_response()); diff --git a/test/test_client_async_cancellation.cpp b/test/test_client_async_cancellation.cpp index 2ce9f48..e594c60 100644 --- a/test/test_client_async_cancellation.cpp +++ b/test/test_client_async_cancellation.cpp @@ -353,9 +353,9 @@ TEST_P(SmallSocketBuffers, WHEN_client_write_with_cancel_after_THEN_cancelled) const std::vector body(8_m, 'x'); #if ANYHTTP_COROSIO - auto [ec] = co_await stop_after(100ms, request.write(boost::asio::buffer(body))); + auto [ec, written] = co_await stop_after(100ms, request.write(boost::asio::buffer(body))); #else - auto [ec] = + auto [ec, written] = co_await request.async_write(boost::asio::buffer(body), cancel_after(100ms, as_tuple)); #endif EXPECT_EQ(ec, errc::operation_canceled); diff --git a/test/test_client_tls.cpp b/test/test_client_tls.cpp index 02a451d..807e24f 100644 --- a/test/test_client_tls.cpp +++ b/test/test_client_tls.cpp @@ -462,7 +462,7 @@ TEST_P(ClientTlsEnd, WHEN_write_is_stuck_THEN_shutdown_interrupts_it_and_says_so auto request = check(co_await session.submit(target, {})); auto write = [&]() -> Task { - std::tie(write_result) = co_await request.write(boost::asio::buffer(body)); + std::tie(write_result, std::ignore) = co_await request.write(boost::asio::buffer(body)); }; auto end_it = [&]() -> Task { co_await stalled.wait(); @@ -531,7 +531,7 @@ TEST_P(ClientShutdown, WHEN_request_is_open_THEN_it_fails_right_away) }; co_await when_both(read(), end()); - auto [write_ec] = co_await request.write(boost::asio::const_buffer("x", 1)); + auto [write_ec, written] = co_await request.write(boost::asio::const_buffer("x", 1)); EXPECT_TRUE(write_ec); }; } diff --git a/test/test_coroutine_api.cpp b/test/test_coroutine_api.cpp index 86cc595..d5bb93d 100644 --- a/test/test_coroutine_api.cpp +++ b/test/test_coroutine_api.cpp @@ -40,7 +40,7 @@ TEST_P(CoroutineApi, WHEN_exchanging_a_body_THEN_both_sides_see_all_of_it) auto [submit_ec] = co_await response.submit(200, {}); EXPECT_FALSE(submit_ec) << submit_ec.message(); - auto [write_ec] = co_await response.write(boost::asio::buffer(body)); + auto [write_ec, written] = co_await response.write(boost::asio::buffer(body)); EXPECT_FALSE(write_ec) << write_ec.message(); // // Not checked: over HTTP/3, the server's writes complete when the peer acknowledges them, @@ -54,9 +54,9 @@ TEST_P(CoroutineApi, WHEN_exchanging_a_body_THEN_both_sides_see_all_of_it) auto [ec, request] = co_await session.submit(url); EXPECT_FALSE(ec) << ec.message(); - auto [write_ec] = co_await request.write(boost::asio::buffer("Hello, ", 7)); + auto [write_ec, written] = co_await request.write(boost::asio::buffer("Hello, ", 7)); EXPECT_FALSE(write_ec) << write_ec.message(); - auto [eof_ec] = co_await request.write_eof(boost::asio::buffer("World", 5)); + auto [eof_ec, eof_written] = co_await request.write_eof(boost::asio::buffer("World", 5)); EXPECT_FALSE(eof_ec) << eof_ec.message(); auto [response_ec, response] = co_await request.get_response(); @@ -92,7 +92,7 @@ TEST_P(CoroutineApi, WHEN_writing_after_the_end_THEN_error_is_reported_not_throw EXPECT_FALSE(ec) << ec.message(); EXPECT_FALSE(std::get<0>(co_await request.write_eof())); - auto [write_ec] = co_await request.write(boost::asio::buffer("late", 4)); + auto [write_ec, written] = co_await request.write(boost::asio::buffer("late", 4)); EXPECT_EQ(write_ec, errc::broken_pipe); auto [response_ec, response] = co_await request.get_response(); diff --git a/test/test_get.cpp b/test/test_get.cpp index 182bbe4..fe424ac 100644 --- a/test/test_get.cpp +++ b/test/test_get.cpp @@ -303,7 +303,7 @@ TEST_P(SubmitMethod, WHEN_get_stops_reading_at_a_limit_THEN_status_and_fields_ar std::vector chunk(16_k); for (;;) { - auto [ec] = co_await response.write(boost::asio::buffer(chunk)); + auto [ec, written] = co_await response.write(boost::asio::buffer(chunk)); if (ec) break; sent += chunk.size(); diff --git a/test/test_headers.cpp b/test/test_headers.cpp index 0a1667e..f45e685 100644 --- a/test/test_headers.cpp +++ b/test/test_headers.cpp @@ -240,7 +240,7 @@ class HeaderLimits : public ClientAsync fields.set("Content-Length", "0"); auto [ec] = co_await response.submit(200, fields); if (!ec) - std::tie(ec) = co_await response.write_eof(); + std::tie(ec, std::ignore) = co_await response.write_eof(); logi("server: {}", what(ec)); }; } diff --git a/test/test_quic.cpp b/test/test_quic.cpp index 7636bc8..9b74e48 100644 --- a/test/test_quic.cpp +++ b/test/test_quic.cpp @@ -44,7 +44,7 @@ Task echo_streams(quic::Connection connection) auto echo = [](quic::Stream stream) -> Task { auto [ec, body] = co_await read_all(stream); EXPECT_FALSE(ec) << ec.message(); - auto [write_ec] = co_await stream.write_eof(boost::asio::buffer(body)); + auto [write_ec, written] = co_await stream.write_eof(boost::asio::buffer(body)); std::ignore = write_ec; // the client may be gone before it acknowledges the end }; for (;;) @@ -139,9 +139,9 @@ TEST_P(QuicStreams, WHEN_stream_is_echoed_THEN_both_ends_see_the_whole_body) EXPECT_FALSE(ec) << ec.message(); EXPECT_EQ(stream.id(), 0); - auto [write_ec] = co_await stream.write(boost::asio::buffer("Hello, ", 7)); + auto [write_ec, written] = co_await stream.write(boost::asio::buffer("Hello, ", 7)); EXPECT_FALSE(write_ec) << write_ec.message(); - auto [eof_ec] = co_await stream.write_eof(boost::asio::buffer("World!", 6)); + auto [eof_ec, eof_written] = co_await stream.write_eof(boost::asio::buffer("World!", 6)); EXPECT_FALSE(eof_ec) << eof_ec.message(); auto [read_ec, body] = co_await read_all(stream); @@ -162,7 +162,7 @@ TEST_P(QuicStreams, WHEN_bodies_are_larger_than_the_windows_THEN_they_arrive_int auto connection = co_await connect(); auto exchange = [](quic::Stream stream, std::string body) -> Task { - auto [write_ec] = co_await stream.write_eof(boost::asio::buffer(body)); + auto [write_ec, written] = co_await stream.write_eof(boost::asio::buffer(body)); EXPECT_FALSE(write_ec) << write_ec.message(); auto [read_ec, echoed] = co_await read_all(stream); EXPECT_FALSE(read_ec) << read_ec.message(); @@ -191,7 +191,8 @@ TEST_P(QuicStreams, WHEN_one_side_ends_its_body_THEN_the_other_can_still_write) // the client's side is over, ours is not for (auto part : {"an", "s", "wer"}) { - auto [write_ec] = co_await stream.write(boost::asio::buffer(std::string_view(part))); + auto [write_ec, written] = + co_await stream.write(boost::asio::buffer(std::string_view(part))); EXPECT_FALSE(write_ec) << write_ec.message(); } std::ignore = co_await stream.write_eof(); @@ -201,7 +202,7 @@ TEST_P(QuicStreams, WHEN_one_side_ends_its_body_THEN_the_other_can_still_write) run([&]() -> Task { auto connection = co_await connect(); auto [ec, stream] = connection.open_stream(); - auto [eof_ec] = co_await stream.write_eof(boost::asio::buffer("question", 8)); + auto [eof_ec, eof_written] = co_await stream.write_eof(boost::asio::buffer("question", 8)); EXPECT_FALSE(eof_ec) << eof_ec.message(); auto [read_ec, response] = co_await read_all(stream); @@ -209,9 +210,9 @@ TEST_P(QuicStreams, WHEN_one_side_ends_its_body_THEN_the_other_can_still_write) EXPECT_EQ(response, "answer"); // the end is final: data after it is refused, a second end is a no-op - auto [late_ec] = co_await stream.write(boost::asio::buffer("x", 1)); + auto [late_ec, late_written] = co_await stream.write(boost::asio::buffer("x", 1)); EXPECT_EQ(late_ec, errc::broken_pipe); - auto [again_ec] = co_await stream.write_eof(); + auto [again_ec, again_written] = co_await stream.write_eof(); EXPECT_FALSE(again_ec) << again_ec.message(); }()); } @@ -235,12 +236,12 @@ TEST_P(QuicStreams, WHEN_peer_resets_its_side_THEN_reads_report_partial_message) run([&]() -> Task { auto connection = co_await connect(); auto [ec, stream] = connection.open_stream(); - auto [write_ec] = co_await stream.write(boost::asio::buffer("partial", 7)); + auto [write_ec, written] = co_await stream.write(boost::asio::buffer("partial", 7)); EXPECT_FALSE(write_ec) << write_ec.message(); co_await delay(50ms); // let it arrive, so that the server sees a stream to accept stream.reset_stream(42); - auto [late_ec] = co_await stream.write(boost::asio::buffer("more", 4)); + auto [late_ec, late_written] = co_await stream.write(boost::asio::buffer("more", 4)); EXPECT_EQ(late_ec, errc::connection_reset); auto [read_ec, body] = co_await read_all(stream); @@ -272,13 +273,13 @@ TEST_P(QuicStreams, WHEN_peer_stops_reading_THEN_writes_fail_with_connection_res error_code write_ec; for (int i = 0; i < 100 && !write_ec; ++i) { - std::tie(write_ec) = co_await stream.write(boost::asio::buffer("data", 4)); + std::tie(write_ec, std::ignore) = co_await stream.write(boost::asio::buffer("data", 4)); if (!write_ec) co_await delay(5ms); } EXPECT_EQ(write_ec, errc::connection_reset); - auto [late_ec] = co_await stream.write(boost::asio::buffer("more", 4)); + auto [late_ec, late_written] = co_await stream.write(boost::asio::buffer("more", 4)); EXPECT_EQ(late_ec, errc::connection_reset); }()); } @@ -296,7 +297,7 @@ TEST_P(QuicStreams, WHEN_writer_is_dropped_before_the_end_THEN_peer_sees_partial run([&]() -> Task { auto connection = co_await connect(); auto [ec, stream] = connection.open_stream(); - auto [write_ec] = co_await stream.write(boost::asio::buffer("unfinished", 10)); + auto [write_ec, written] = co_await stream.write(boost::asio::buffer("unfinished", 10)); EXPECT_FALSE(write_ec) << write_ec.message(); co_await delay(50ms); { @@ -340,7 +341,7 @@ TEST_P(QuicStreamsSmallWindow, WHEN_reader_is_slow_THEN_flow_control_holds_the_w auto connection = co_await connect(); auto [ec, stream] = connection.open_stream(); auto body = pattern(8 * window); - auto [write_ec] = co_await stream.write_eof(boost::asio::buffer(body)); + auto [write_ec, written] = co_await stream.write_eof(boost::asio::buffer(body)); EXPECT_FALSE(write_ec) << write_ec.message(); EXPECT_TRUE(reading) << "the write completed before the server started reading"; std::ignore = co_await read_all(stream); @@ -468,7 +469,7 @@ TEST_P(QuicStreams, WHEN_write_is_cancelled_THEN_the_stream_goes_on_with_a_short // stream being blocked by flow control, it usually is. Otherwise the stream had to be // reset. // - auto [write_ec] = co_await stream.write_eof(boost::asio::buffer("end", 3)); + auto [write_ec, written] = co_await stream.write_eof(boost::asio::buffer("end", 3)); if (GetParam() && write_ec) EXPECT_EQ(write_ec, errc::connection_reset); else From c9c53c46dc8cd8396304c7039adb06a30a9556fe Mon Sep 17 00:00:00 2001 From: Peter Eisenlohr Date: Sun, 8 Nov 2026 19:54:46 +0000 Subject: [PATCH 3/5] test: what a cancelled operation promises On ASIO, a request for partial or total cancellation alone leaves a read running, and terminal completes it. A cancelled read_some() can be followed by reading the rest of the body. A write cancelled while the server is not reading says how much of an 8 MiB body it has taken, and write_eof(buffer + n) then gets the whole body across intact, after write() as after write_eof(). HTTP/1.1 skips the last two, as its cancelled write ends the connection. Co-Authored-By: Claude Opus 5.5 --- test/test_client_async_cancellation.cpp | 158 ++++++++++++++++++++++++ 1 file changed, 158 insertions(+) diff --git a/test/test_client_async_cancellation.cpp b/test/test_client_async_cancellation.cpp index e594c60..2a86283 100644 --- a/test/test_client_async_cancellation.cpp +++ b/test/test_client_async_cancellation.cpp @@ -455,3 +455,161 @@ TEST_P(ClientAsyncCancellation, DISABLED_SpawnAndForget) } // ================================================================================================= + +// +// The contract of a cancelled operation, see docs/cancellation.rst: it completes right away, says +// how much of its buffer it has taken, and leaves the message usable or not, which the next +// operation tells. +// + +#if ANYHTTP_ASIO +// +// Only the terminal kind of cancellation counts. A request for partial or total alone promises a +// state that is not always there to keep, so it is ignored, as Asio asks of an operation that does +// not support it. +// +TEST_P(ClientAsyncCancellation, WHEN_only_partial_or_total_is_requested_THEN_read_some_ignores_it) +{ + clientSession = [this](Session session) -> Task { + auto request = check(co_await session.submit(url.set_path("echo"), {})); + auto response = check(co_await request.get_response()); + + std::array buffer; + std::optional> result; + boost::asio::cancellation_signal cancel; + response.async_read_some( + boost::asio::buffer(buffer), + boost::asio::bind_cancellation_slot(cancel.slot(), [&result](error_code ec, size_t n) { // + result.emplace(ec, n); + })); + + cancel.emit(boost::asio::cancellation_type::partial | boost::asio::cancellation_type::total); + co_await sleep(50ms); + EXPECT_FALSE(result) << "completed with " << std::get<0>(*result).message(); + + cancel.emit(boost::asio::cancellation_type::terminal); + for (int i = 0; !result && i < 1000; ++i) + co_await sleep(1ms); + if (!result) + { + ADD_FAILURE() << "terminal cancellation did not complete the read"; + co_return; + } + EXPECT_EQ(std::get<0>(*result), errc::operation_canceled); + EXPECT_EQ(std::get<1>(*result), 0u); + }; +} +#endif + +// +// A cancelled read leaves the body where it was: what arrives afterwards is read as usual. +// +TEST_P(ClientAsyncCancellation, WHEN_read_some_is_cancelled_THEN_the_body_can_still_be_read) +{ + clientSession = [this](Session session) -> Task { + auto request = check(co_await session.submit(url.set_path("echo"), {})); + auto response = check(co_await request.get_response()); + + std::array buffer; +#if ANYHTTP_COROSIO + auto [ec, n] = co_await stop_after(50ms, response.read_some(boost::asio::buffer(buffer))); +#else + auto [ec, n] = co_await response.async_read_some(boost::asio::buffer(buffer), + cancel_after(50ms, as_tuple)); +#endif + EXPECT_EQ(ec, errc::operation_canceled); + EXPECT_EQ(n, 0u); + + constexpr auto msg = "Hello, Client!"sv; + check(co_await request.write_eof(boost::asio::buffer(msg))); + EXPECT_EQ(co_await read(response), msg); + }; +} + +// ------------------------------------------------------------------------------------------------- + +// +// A write cancelled while the peer is not reading says how much of its buffer it has taken. Where +// that leaves the request usable, writing what is left after it and ending the body -- the drain +// after the cancel -- gets the whole body across, intact. That holds for write_eof() as much as for +// write(): an end of the body the protocol stack has not been given yet is taken back with it. +// +// HTTP/1.1 sends a body by reference, without a copy, so its cancelled write ends the connection +// instead, and there is no rest to follow. +// +class CancelledWrite : public ClientAsyncCancellation +{ +protected: + static const std::vector& body() + { + static const std::vector body = [] { + std::vector result(8_m); + for (size_t i = 0; i < result.size(); ++i) + result[i] = uint8_t(i * 7 + i / 251); + return result; + }(); + return body; + } + + void cancel_then_finish(bool eof) + { + // + // The server does not read until the write has been cancelled, so far more than the window + // is outstanding when it is. Then it reads the body and says whether it is the one sent. + // + requestHandler = [](server::Request request, server::Response response) -> Task { + co_await sleep(300ms); + const auto received = co_await read(request); + const bool intact = std::ranges::equal(received, body(), [](char a, uint8_t b) { // + return uint8_t(a) == b; + }); + const auto answer = std::format("{} {}", received.size(), intact ? "intact" : "broken"); + check(co_await response.submit(200, {})); + check(co_await response.write_eof(boost::asio::buffer(answer))); + }; + + clientSession = [this, eof](Session session) -> Task { + auto request = check(co_await session.submit(url, {})); + const auto buffer = boost::asio::buffer(body()); + + error_code ec; + size_t n = 0; +#if ANYHTTP_COROSIO + if (eof) + std::tie(ec, n) = co_await stop_after(100ms, request.write_eof(buffer)); + else + std::tie(ec, n) = co_await stop_after(100ms, request.write(buffer)); +#else + if (eof) + std::tie(ec, n) = + co_await request.async_write_eof(buffer, cancel_after(100ms, as_tuple)); + else + std::tie(ec, n) = co_await request.async_write(buffer, cancel_after(100ms, as_tuple)); +#endif + EXPECT_EQ(ec, errc::operation_canceled); + EXPECT_GT(n, 0u); + EXPECT_LT(n, buffer.size()); + std::println("cancelled after taking {} of {} bytes", n, buffer.size()); + + check(co_await request.write_eof(buffer + n)); + auto response = check(co_await request.get_response()); + EXPECT_EQ(co_await read(response), std::format("{} intact", buffer.size())); + }; + } +}; + +INSTANTIATE_TEST_SUITE_P(CancelledWrite, CancelledWrite, ValuesIn(protocols()), NameGenerator); + +TEST_P(CancelledWrite, WHEN_write_is_cancelled_THEN_the_rest_can_follow) +{ + if (GetParam() == anyhttp::Protocol::h1) + GTEST_SKIP(); // a cancelled write ends the connection, see above + cancel_then_finish(false); +} + +TEST_P(CancelledWrite, WHEN_write_eof_is_cancelled_THEN_the_rest_can_follow) +{ + if (GetParam() == anyhttp::Protocol::h1) + GTEST_SKIP(); // a cancelled write ends the connection, see above + cancel_then_finish(true); +} From 14556f71ad2ce5f126b60b010e79a82b309d193c Mon Sep 17 00:00:00 2001 From: Peter Eisenlohr Date: Sun, 8 Nov 2026 19:54:52 +0000 Subject: [PATCH 4/5] docs: cancellation, abandoning and draining A page of its own for what used to be a list of cases in the guide. There are three ways to stop something in progress: cancelling one operation, which completes right away, abandoning a message or session by destroying its handle, and draining, the ordinary operations that finish what was started and may wait for the peer. The page has what a cancelled operation promises, what each one leaves behind, how writes differ by protocol, and how to finish a body after a cancelled write. The guide points there now, and end-of-body.rst says what cancelling write_eof() does to the end of the body since it is taken back where the stack has not been given it. Co-Authored-By: Claude Opus 5.5 --- docs/cancellation.rst | 111 +++++++++++++++++++++++++++++++++++++ docs/end-of-body.rst | 11 ++-- docs/index.rst | 1 + docs/programmers-guide.rst | 19 ++----- 4 files changed, 125 insertions(+), 17 deletions(-) create mode 100644 docs/cancellation.rst diff --git a/docs/cancellation.rst b/docs/cancellation.rst new file mode 100644 index 0000000..db1cc87 --- /dev/null +++ b/docs/cancellation.rst @@ -0,0 +1,111 @@ +.. cpp:namespace:: anyhttp + +Cancellation +============ + +There are three ways to stop something that is in progress, and only the first of them is +cancellation: + +* **Cancelling** stops one operation, through the mechanism of the I/O runtime. It completes right + away, whatever that leaves behind. +* **Abandoning** a message or a session, by destroying its handle, ends what it had open, abruptly. + There is no operation to complete, and nothing goes on in the background. +* **Draining** finishes what was started: writing the rest of a body and ending it, reading one to + its end, :cpp:func:`Session::shutdown`. These are ordinary operations, which may wait for the peer + for as long as the peer takes. That is why they are separate, and why a time limit goes around + them rather than into cancellation. + +Cancelling +---------- + +On ASIO, an operation is cancelled through the cancellation slot of its completion token, which is +what ``cancel_after`` and ``||`` use. On COROSIO, it is cancelled through the ``std::stop_token`` of +the coroutine that awaits it. :cpp:func:`when_either` works on both, see :doc:`programmers-guide`. + +There is one kind of cancellation. On ASIO, that is a request with ``cancellation_type::terminal``, +the one ``cancel_after`` sends, and ``||`` sends it as well. A request for ``partial`` or ``total`` +alone is ignored, which is what Asio asks of an operation that cannot promise what those stand for. + +What a cancelled operation promises +----------------------------------- + +1. It **completes right away**. It does not wait for the peer, for flow control, or for a frame to + be finished, so it never needs a time limit of its own. Its completion is still queued like any + other, not run from inside the cancellation. +2. It completes with an error that compares equal to ``errc::operation_canceled``, unless it had + finished already, in which case it reports how it finished. +3. A read reports the bytes it has delivered, a write **how many bytes of its buffer it has taken**. + Those are no longer the caller's: what is left to write starts after them, at ``buffer + n``. + Bytes taken need not have reached the peer, as some protocols copy them first. +4. The buffer of the operation has to stay valid until it completes, cancelled or not. Once it has + completed, nothing refers to it any more. +5. It leaves the message either **usable**, so that it can be drained, or **broken**, so that it can + only be abandoned. The cancellation does not say which; the next operation does, by failing. + +What it leaves behind +--------------------- + +.. list-table:: + :header-rows: 1 + :stub-columns: 1 + + * - Operation + - After it has been cancelled + * - :cpp:func:`Client::connect ` + - Nothing: over HTTP/3, the connection being set up ends with it. + * - :cpp:func:`get_response() ` + - Usable: it can be started again. Over HTTP/1.1 only if nothing of the response had arrived. + * - :cpp:func:`Reader::read_some` + - Usable: what arrives afterwards is read as usual. + * - :cpp:func:`Writer::write`, :cpp:func:`Writer::write_eof` + - Depends on the protocol, see below. + * - :cpp:func:`Session::shutdown` + - The orderly end is not waited for any more, and the session takes no more requests. Destroy + it to end the connection. + +Whether a cancelled write leaves its body usable depends on how the protocol takes it: + +* **HTTP/1.1** sends the body straight out of the caller's buffer, as one chunk, so a write that + stops in the middle leaves a chunk that can't be finished. The connection stops sending, and every + further write fails. ``n`` is what of the body has gone onto the connection. +* **HTTP/2** over nghttp2 version 1 copies what flow control lets through. The stream stays usable, + and ``n`` is what has been copied. +* **HTTP/2** over nghttp2 version 2 and **HTTP/3** copy a request body, a bounded piece at a time, + so a request stays usable, and ``n`` is what has been copied. A response is handed over by + reference until the peer has it, see :doc:`http2-library`. If the cancellation leaves bytes of it + outstanding, the stream is reset and the response is broken; if not, it stays usable, and ``n`` + is what the stack has been given. Plain QUIC streams copy, unless + :cpp:member:`quic::Config::zero_copy` says to hand over by reference. + +Draining +-------- + +A write that was cancelled with its message left usable is finished by writing ``buffer + n``, the +part it has not taken, and then ending the body, or with ``write_eof(buffer + n)`` in one go. That +holds for a cancelled ``write_eof()`` as well: if the end of the body had not been given to the +protocol stack yet, it is taken back with the write, and if it had been, ending the body again is +the idempotent re-end of :doc:`end-of-body`. Either way, ``write_eof(buffer + n)`` is right. + +This needs the ``n`` of the cancelled write. :cpp:func:`when_either` does not hand back the result +of the operation it cancels, so for a write that is to be drained, cancel it through its own token +on ASIO, ``cancel_after`` for a time limit, and with a stop token of its own on COROSIO. + +A drain can wait for the peer just like the write before it, so it is where a time limit belongs. +Cancelling the drain is cancellation again: it completes right away, and if the message is broken +then, all that is left is abandoning it. + +A coroutine that has been cancelled itself, and wants to drain before it returns, needs +:cpp:func:`reset_cancellation` and :cpp:func:`shielded` to get to its next ``co_await``. + +Abandoning +---------- + +Destroying a :cpp:class:`Writer`, or the request or response it belongs to, before the body has +ended cuts the body short: over HTTP/2 and HTTP/3, the stream is reset. Destroying a +:cpp:class:`Reader` before the end of its body stops reading: over HTTP/3 with a STOP_SENDING, over +HTTP/2 by letting flow control stop the peer, so that a response can still go out, and over +HTTP/1.1, where the rest of a body can't be skipped, by closing the connection. + +Destroying a :cpp:class:`Session`, or :cpp:func:`Session::reset`, ends the connection, see +:doc:`programmers-guide`. None of this waits for anything or leaves anything running: what is still +in progress on that message or session completes with an error. diff --git a/docs/end-of-body.rst b/docs/end-of-body.rst index 3a3d3a0..cfedf06 100644 --- a/docs/end-of-body.rst +++ b/docs/end-of-body.rst @@ -51,7 +51,10 @@ one (``partial_message``), without touching the connection that is gone. Cancellation ------------ -Cancelling ``write_eof()`` does not take the end of the body back. The end is declared the moment the -operation starts, and no protocol can un-send a FIN. Cancellation detaches the handler, and issuing -``write_eof()`` again adopts the end that is already on its way rather than failing. Intent and -delivery are tracked separately for exactly this reason. +Cancelling ``write_eof()`` takes the end of the body back if the protocol stack has not been given +it yet: the body is open again, and what the write has not taken can follow. Once the stack has it, +the end stands, as no protocol can un-send a FIN, and issuing ``write_eof()`` again is the +idempotent re-end of rule 2 -- or, over HTTP/3, adopts the end that is still on its way. Which of the +two a cancelled ``write_eof()`` leaves depends on how far it had come, so going on with +``write_eof(buffer + n)``, where ``n`` is what it has taken, is right either way. See +:doc:`cancellation`. diff --git a/docs/index.rst b/docs/index.rst index 9077c84..7fc2017 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -35,6 +35,7 @@ report `__. :caption: How it works io-runtimes + cancellation end-of-body concurrent-requests alt-svc diff --git a/docs/programmers-guide.rst b/docs/programmers-guide.rst index a6d16cf..a3891fe 100644 --- a/docs/programmers-guide.rst +++ b/docs/programmers-guide.rst @@ -335,19 +335,12 @@ against something else, with :cpp:func:`when_either`, which cancels the loser: :start-after: // [timeout] :end-before: // [timeout-end] -A cancelled operation completes with an error that compares equal to ``errc::operation_canceled``, -which is the way to ask, as a timer of the runtime has its own spelling of it. What that leaves behind -depends on how far it had come, and on the protocol: - -* A cancelled :cpp:func:`get_response() ` can be started again. -* Cancelling :cpp:func:`Writer::write_eof` does not take the end of the body back, see - :ref:`end-of-body-cancellation`. -* A write that has started putting data on a TCP connection leaves part of a message behind. Over - HTTP/1.1, that is the end of the connection for further requests, see - :doc:`concurrent-requests`. -* The response of a server over HTTP/2 version 2 and HTTP/3 is handed to the protocol stack by - reference, without a copy, until the peer has it. As that can't be taken back, cancelling a write - with bytes outstanding resets the stream. A request body is copied, so it can go on afterwards. +A cancelled operation completes right away, with an error that compares equal to +``errc::operation_canceled``, which is the way to ask, as a timer of the runtime has its own spelling +of it. A cancelled write also says how much of its buffer it has taken. Whether the message can go on +afterwards depends on the operation and the protocol: a cancelled read can always be followed by +another, a cancelled write over HTTP/1.1 ends the connection. :doc:`cancellation` has all of it, and +how to finish a message after a cancelled write. A coroutine that has been cancelled and wants to clean up by awaiting something else, as a last ``write_eof()``, needs :cpp:func:`reset_cancellation` and :cpp:func:`shielded` to get there. From df5c9dd2003b00866dad323a7afd45be71053547 Mon Sep 17 00:00:00 2001 From: Peter Eisenlohr Date: Sun, 8 Nov 2026 21:20:02 +0000 Subject: [PATCH 5/5] test: delay() can be cancelled A new file for what runtime.hpp offers generic code, with the same expectations on both runtimes. A delay cancelled from outside completes right away with operation_canceled: on ASIO through a terminal signal on its completion's slot, on COROSIO through its stop token. when_either() does not wait for the delay it cancels. On ASIO, partial or total alone does not reach the timer, since co_spawn() lets only terminal cancellation through by default. Co-Authored-By: Claude Opus 5.5 --- test/test_runtime.cpp | 167 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 167 insertions(+) create mode 100644 test/test_runtime.cpp diff --git a/test/test_runtime.cpp b/test/test_runtime.cpp new file mode 100644 index 0000000..3be038f --- /dev/null +++ b/test/test_runtime.cpp @@ -0,0 +1,167 @@ +// +// What anyhttp/runtime.hpp offers generic code, checked on the runtime of the tree: the same +// expectations on ASIO and COROSIO, with only the way a test cancels something told apart. +// +#include "anyhttp/runtime.hpp" + +#if ANYHTTP_ASIO +#include +#include +#include +#include +#else +#include +#include + +#include +#endif + +#include +#include + +#include +#include +#include + +using namespace anyhttp; +using namespace std::chrono_literals; +using namespace testing; + +namespace +{ + +using Clock = std::chrono::steady_clock; + +/// Longer than any test here may take: a delay of this that is not cancelled fails the test. +constexpr auto forever = 10s; + +#if ANYHTTP_ASIO +using IoContext = boost::asio::io_context; + +/// Runs a task that is cancelled from outside, the way `cancel_after` does it: by emitting +/// \p type on the cancellation slot bound to its completion. +class Cancellable +{ +public: + explicit Cancellable( + boost::asio::cancellation_type type = boost::asio::cancellation_type::terminal) + : type_(type) + { + } + + void start(const Executor& executor, Task task, std::optional& result) + { + boost::asio::co_spawn( + executor, std::move(task), + boost::asio::bind_cancellation_slot( + signal_.slot(), [&result](const std::exception_ptr& ep, error_code ec) { + if (ep) + ADD_FAILURE() << "threw instead of returning an error"; + else + result = ec; + })); + } + + void cancel() { signal_.emit(type_); } + +private: + boost::asio::cancellation_type type_; + boost::asio::cancellation_signal signal_; +}; +#else +using IoContext = boost::corosio::io_context; + +/// Runs a task that is cancelled from outside: through the stop token it is run with. +class Cancellable +{ +public: + void start(const Executor& executor, Task task, std::optional& result) + { + auto run = [](Task task, std::optional& result) -> Task { + result = co_await std::move(task); + }; + boost::capy::run_async(executor, stop_.get_token())(run(std::move(task), result)); + } + + void cancel() { stop_.request_stop(); } + +private: + std::stop_source stop_; +}; +#endif + +} // namespace + +// ================================================================================================= + +TEST(Runtime, WHEN_delay_is_cancelled_THEN_it_completes_right_away_with_canceled) +{ + IoContext context; + Cancellable cancellable; + std::optional result; + + cancellable.start(context.get_executor(), delay(forever), result); + run_later(context.get_executor(), [&] { cancellable.cancel(); }); // once the timer waits + const auto start = Clock::now(); + context.run(); + + EXPECT_LT(Clock::now() - start, 1s); + ASSERT_TRUE(result); + EXPECT_EQ(*result, errc::operation_canceled) << result->message(); +} + +TEST(Runtime, WHEN_delay_is_not_cancelled_THEN_it_completes_without_error_once_it_has_passed) +{ + IoContext context; + Cancellable cancellable; + std::optional result; + + cancellable.start(context.get_executor(), delay(50ms), result); + const auto start = Clock::now(); + context.run(); + + EXPECT_GE(Clock::now() - start, 50ms); + ASSERT_TRUE(result); + EXPECT_FALSE(*result) << result->message(); +} + +TEST(Runtime, WHEN_the_other_task_of_when_either_finishes_first_THEN_delay_is_cancelled) +{ + IoContext context; + std::optional index; + + auto race = [](std::optional& index) -> Task { + auto result = co_await when_either(delay(forever), delay(10ms)); + index = result.index(); + }; + launch(context.get_executor(), race(index)); + const auto start = Clock::now(); + context.run(); + + EXPECT_LT(Clock::now() - start, 1s) << "when_either() waited for the delay it cancelled"; + ASSERT_TRUE(index); + EXPECT_EQ(*index, 1); +} + +#if ANYHTTP_ASIO +// +// There is one kind of cancellation (docs/cancellation.rst): ASIO's awaitable lets only terminal +// through, by default, so a request for partial or total alone does not reach the timer. +// +TEST(Runtime, WHEN_partial_or_total_cancellation_alone_is_requested_THEN_delay_runs_its_course) +{ + IoContext context; + Cancellable cancellable(boost::asio::cancellation_type::partial | + boost::asio::cancellation_type::total); + std::optional result; + + cancellable.start(context.get_executor(), delay(50ms), result); + run_later(context.get_executor(), [&] { cancellable.cancel(); }); + const auto start = Clock::now(); + context.run(); + + EXPECT_GE(Clock::now() - start, 50ms); + ASSERT_TRUE(result); + EXPECT_FALSE(*result) << result->message(); +} +#endif