Cancellation: one kind, and a write says how much it has taken

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 <client::Client::connect>`
+     - Nothing: over HTTP/3, the connection being set up ends with it.
+   * - :cpp:func:`get_response() <client::Request::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.

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`.

docs/index.rst

@@ -35,6 +35,7 @@ report <https://pgit.github.io/anyhttp/coverage/>`__.
    :caption: How it works
 
    io-runtimes
+   cancellation
    end-of-body
    concurrent-requests
    alt-svc

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() <client::Request::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.

docs/snippets.cpp

@@ -64,7 +64,8 @@ Task<void> 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<std::string> 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<size_t> read_at_most(Session session, boost::urls::url url, size_t limit)
 Task<void> 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<void> 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<std::string> 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);

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 <typename Signature, typename F>
 inline void on_cancel(Completion<Signature>& 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<F>(on_cancel)](boost::asio::cancellation_type_t) mutable { //
+      slot.assign([on_cancel = std::forward<F>(on_cancel)](
+                     boost::asio::cancellation_type_t type) mutable { //
+         if ((type & boost::asio::cancellation_type::terminal) !=
+             boost::asio::cancellation_type::none)
             on_cancel();
-         });
+      });
 }
 
 // -------------------------------------------------------------------------------------------------

include/anyhttp/common.hpp

@@ -139,8 +139,10 @@ using ReadSomeHandler = Completion<ReadSome>;
 using WriteSome = void(error_code, size_t);
 using WriteSomeHandler = Completion<WriteSome>;
 
-/// 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<Write>;
 
 /// The completion signature of an operation that has nothing to report but how it went.

include/anyhttp/h1/io.hpp

@@ -156,15 +156,28 @@ IoTask<size_t> 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 <typename Stream, bool isRequest, typename Body, typename Fields>
-IoTask<size_t> write_some(Stream& stream, http::serializer<isRequest, Body, Fields>& serializer)
+IoTask<size_t> write_some(Stream& stream, http::serializer<isRequest, Body, Fields>& serializer,
+                          BodyWatch* watch = nullptr)
 {
    if (serializer.is_done())
       co_return std::tuple{error_code{}, size_t{0}};
@@ -186,6 +199,21 @@ IoTask<size_t> write_some(Stream& stream, http::serializer<isRequest, Body, Fiel
    auto [write_ec, n] = co_await io::write(stream, buffers);
    if (!write_ec)
       serializer.consume(n);
+
+   if (watch)
+   {
+      const auto* begin = static_cast<const char*>(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<const char*>(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<size_t> write_some(Stream& stream, http::serializer<isRequest, Body, Fiel
  * here, this also stops when a \c buffer_body has been written that is not the last.
  */
 template <typename Stream, typename Serializer, typename Predicate>
-IoTask<size_t> write_until(Stream& stream, Serializer& serializer, Predicate done)
+IoTask<size_t> 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<size_t> write_header(Stream& stream, Serializer& serializer)
 
 /// Writes all there is, as Beast's \c http::async_write() with a serializer.
 template <typename Stream, typename Serializer>
-IoTask<size_t> write(Stream& stream, Serializer& serializer)
+IoTask<size_t> 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.

include/anyhttp/h2/stream.hpp

@@ -141,16 +141,24 @@ class NGHttp2Stream : public std::enable_shared_from_this<NGHttp2Stream>
    // 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;
 
    //
    // How the end of the outgoing body travels: async_write_eof() sets \c eof_requested, and the
    // 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;