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. 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/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(); - }); + }); } // ------------------------------------------------------------------------------------------------- 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..2a86283 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); @@ -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); +} 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 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