EOF semantics -- don't use empty read as EOF marker, but asio::error::eof to better conform to ASIO

README.md

@@ -28,13 +28,32 @@ awaitable<void> echo(server::Request request, server::Response response)
    std::array<uint8_t, 64 * 1024> buffer;
    for (;;)
    {
-      size_t n = co_await request.async_read_some(asio::buffer(buffer));
-      co_await response.async_write(asio::buffer(buffer, n));
-      if (n == 0)
+      auto [ec, n] = co_await request.async_read_some(asio::buffer(buffer), as_tuple);
+      if (ec == asio::error::eof)
          break;
+      if (ec)
+         throw boost::system::system_error(ec);
+
+      co_await response.async_write(asio::buffer(buffer, n));
    }
+
+   co_await response.async_write_eof();
 }
 ```
+
+The end of an incoming body is reported the way ASIO reports it everywhere else: `asio::error::eof`
+with zero bytes. A body cut short -- a reset stream, a connection that went away mid-message --
+completes with `http::error::partial_message` instead, so the two stay distinguishable.
+
+The end of an *outgoing* body is stated explicitly, with `async_write_eof()`. It takes a buffer of
+its own, so the last of the body and the end of it go out together -- one DATA frame with
+END_STREAM, one QUIC STREAM frame with FIN, one last chunk -- instead of costing a second, empty
+write:
+
+```C++
+   co_await response.async_submit(200, fields({{"Content-Length", body.size()}}));
+   co_await response.async_write_eof(asio::buffer(body));
+```
 ### Client
 ```c++
 awaitable<void> do_session(Client& client, boost::urls::url url)
@@ -66,6 +85,7 @@ namespace client {
    class Request {
       async_get_response()
       async_write(buffer)
+      async_write_eof(buffer)
    }
    class Client {
       async_connect()
@@ -87,7 +107,7 @@ namespace impl {
    class Writer {
       get_executor()
       content_length(optional<size_t>)
-      async_write(buffer)
+      async_write(buffer, eof)
       detach()
       destroy()
    }

include/anyhttp/any_async_stream.hpp

@@ -19,9 +19,6 @@
 namespace asio = boost::asio;
 namespace ip = asio::ip;
 
-// this is considerably slower, likely because buffer contents may get copied
-// #define USE_ASIO_LINEARISE
-
 namespace anyhttp
 {
 // =================================================================================================
@@ -59,17 +56,12 @@ class AnyAsyncStream
       virtual ~Impl() = default;
       virtual executor_type get_executor() noexcept = 0;
       virtual ip::tcp::socket& get_socket() = 0;
-#if defined(USE_ASIO_LINEARISE)
-      using ConstBuffers = asio::const_buffer;
-      using MutableBuffers = asio::mutable_buffer;
-      virtual void async_write_impl(ReadWriteHandler handler, asio::const_buffer buffer) = 0;
-      virtual void async_read_impl(ReadWriteHandler handler, asio::mutable_buffer buffer) = 0;
-#else
+
       using ConstBuffers = ConstBufferVector;
       using MutableBuffers = MutableBufferVector;
-      virtual void async_write_impl(ReadWriteHandler handler, ConstBufferVector buffer) = 0;
-      virtual void async_read_impl(ReadWriteHandler handler, MutableBufferVector buffer) = 0;
-#endif
+      virtual void async_write_some(ReadWriteHandler handler, ConstBufferVector buffer) = 0;
+      virtual void async_read_some(ReadWriteHandler handler, MutableBufferVector buffer) = 0;
+
       virtual void async_shutdown_impl(ShutdownHandler handler)
       {
          auto ex = boost::asio::get_associated_immediate_executor(handler, get_executor());
@@ -117,16 +109,9 @@ class AnyAsyncStream
       return boost::asio::async_initiate<CompletionToken, ReadWrite>(
          [this](ReadWriteHandler handler, const ConstBufferSequence& buffers)
       {
-#if defined(USE_ASIO_LINEARISE)
-         using namespace asio;
-         using Adapter = detail::buffer_sequence_adapter<const_buffer, ConstBufferSequence>;
-         std::array<uint8_t, Adapter::linearisation_storage_size> storage;
-         impl->async_write_impl(std::move(handler), Adapter::linearise(buffers, buffer(storage)));
-#else
-         impl->async_write_impl(std::move(handler),
+         impl->async_write_some(std::move(handler),
                                 ConstBufferVector{asio::buffer_sequence_begin(buffers),
                                                   asio::buffer_sequence_end(buffers)});
-#endif
       }, token, buffers);
    }
 
@@ -143,15 +128,9 @@ class AnyAsyncStream
       return boost::asio::async_initiate<CompletionToken, ReadWrite>(
          [this](ReadWriteHandler handler, const MutableBufferSequence& buffers)
       {
-#if defined(USE_ASIO_LINEARISE)
-         using namespace asio;
-         using Adapter = detail::buffer_sequence_adapter<mutable_buffer, MutableBufferSequence>;
-         impl->async_read_impl(std::move(handler), Adapter::first(buffers));
-#else
-         impl->async_read_impl(std::move(handler),
+         impl->async_read_some(std::move(handler),
                                MutableBufferVector{asio::buffer_sequence_begin(buffers),
                                                    asio::buffer_sequence_end(buffers)});
-#endif
       }, token, buffers);
    }
 };

include/anyhttp/buffer_array.hpp

@@ -14,11 +14,8 @@
 #include <anyhttp/concepts.hpp>
 
 #include <boost/asio/buffer.hpp>
-// #include <boost/beast/core/detail/config.hpp>
-// #include <boost/beast/core/detail/buffer_traits.hpp>
 
 #include <cstddef>
-#include <new>
 #include <span>
 #include <utility>
 

include/anyhttp/client.hpp

@@ -46,6 +46,13 @@ class Response
    int status_code() const noexcept;
 
 public:
+   /**
+    * Reads a part of the response body.
+    *
+    * The end of the body is reported as \c asio::error::eof with zero bytes, as ASIO does
+    * everywhere else, and so is every read after it. A body cut short by a reset stream or a lost
+    * connection completes with \c http::error::partial_message instead.
+    */
    template <typename Buffers,
              BOOST_ASIO_COMPLETION_TOKEN_FOR(ReadSome) CompletionToken = DefaultCompletionToken>
       requires(boost::asio::is_mutable_buffer_sequence<Buffers>::value)
@@ -106,20 +113,54 @@ class Request
    }
 
 public:
+   /**
+    * Writes \p buffer as part of the request body, which stays open for more.
+    *
+    * An empty buffer writes nothing and completes immediately -- use \c async_write_eof() to end
+    * the body.
+    */
    template <BOOST_ASIO_COMPLETION_TOKEN_FOR(Write) CompletionToken = DefaultCompletionToken>
    auto async_write(asio::const_buffer buffer, CompletionToken&& token = CompletionToken())
    {
       // FIXME: get_executor() breaks testcase SpawnAndForget because the impl is already gone there
       auto executor = asio::get_associated_executor(token); // , get_executor());
       return asio::async_initiate<CompletionToken, Write>(
          asio::bind_executor(executor, [this](auto&& handler, asio::const_buffer buffer) { //
-            async_write_any(std::move(handler), buffer);
+            async_write_any(std::move(handler), buffer, false);
          }),
          token, buffer);
    }
 
+   /**
+    * Writes \p buffer as the last part of the request body and ends it.
+    *
+    * Both go out together, so ending a body that has a tail of data left costs no more than
+    * writing that tail: no second, empty write and no extra round trip through the protocol
+    * stack. Re-ending an already-ended body with an empty buffer completes immediately and
+    * changes nothing; with data attached it completes with \c errc::broken_pipe, just as writing
+    * that data would -- there is no body left for it to belong to.
+    */
+   template <BOOST_ASIO_COMPLETION_TOKEN_FOR(Write) CompletionToken = DefaultCompletionToken>
+   auto async_write_eof(asio::const_buffer buffer, CompletionToken&& token = CompletionToken())
+   {
+      // see async_write() above for why the executor is not defaulted to get_executor()
+      auto executor = asio::get_associated_executor(token);
+      return asio::async_initiate<CompletionToken, Write>(
+         asio::bind_executor(executor, [this](auto&& handler, asio::const_buffer buffer) { //
+            async_write_any(std::move(handler), buffer, true);
+         }),
+         token, buffer);
+   }
+
+   /// Ends the request body without writing anything more.
+   template <BOOST_ASIO_COMPLETION_TOKEN_FOR(Write) CompletionToken = DefaultCompletionToken>
+   auto async_write_eof(CompletionToken&& token = CompletionToken())
+   {
+      return async_write_eof(asio::const_buffer{}, std::forward<CompletionToken>(token));
+   }
+
 private:
-   void async_write_any(WriteHandler&& handler, asio::const_buffer buffer);
+   void async_write_any(WriteHandler&& handler, asio::const_buffer buffer, bool eof);
    void async_get_response_any(GetResponseHandler&& handler);
    std::shared_ptr<Impl> impl;
 };

include/anyhttp/common.hpp

@@ -6,6 +6,7 @@
 
 #include <boost/asio/any_completion_handler.hpp>
 #include <boost/asio/any_io_executor.hpp>
+#include <boost/asio/associated_immediate_executor.hpp>
 #include <boost/asio/awaitable.hpp>
 #include <boost/asio/deferred.hpp>
 #include <boost/asio/ip/address.hpp>
@@ -131,6 +132,31 @@ inline void swap_and_invoke(F&& function, Args&&... args)
 
 // =================================================================================================
 
+/**
+ * Completes \p handler without doing any I/O, through its associated immediate executor (with
+ * \p fallback standing in when the handler has none). This is the one way an operation that has
+ * nothing asynchronous left to do may finish: invoking the handler straight from the initiating
+ * function would surprise callers that rely on the ASIO guarantee of not being re-entered.
+ *
+ * A handler that is empty (an \c any_completion_handler detached by cancellation) is quietly
+ * dropped -- there is nobody left to tell.
+ */
+template <typename Handler, typename... Args>
+inline void complete_immediately(Handler&& handler, const asio::any_io_executor& fallback,
+                                 Args&&... args)
+{
+   if (!handler)
+      return;
+
+   asio::any_completion_executor ex = asio::get_associated_immediate_executor(handler, fallback);
+   ex.execute([handler = std::forward<Handler>(handler),
+               ... args = std::forward<Args>(args)]() mutable { //
+      std::move(handler)(std::move(args)...);
+   });
+}
+
+// =================================================================================================
+
 namespace impl
 {
 class Reader : public std::enable_shared_from_this<Reader>
@@ -139,6 +165,18 @@ class Reader : public std::enable_shared_from_this<Reader>
    virtual ~Reader() = default;
    virtual asio::any_io_executor get_executor() const noexcept = 0;
    virtual std::optional<size_t> content_length() const noexcept = 0;
+
+   //
+   // Reads at most one buffer worth of the incoming body. The end of the body is reported the way
+   // ASIO reports it everywhere else: \c asio::error::eof with zero bytes, and again for every
+   // further read -- including reads issued after the underlying stream object is long gone. A
+   // body that ends before it was supposed to -- a reset stream, a connection that went away
+   // mid-message -- is reported as \c http::error::partial_message instead, so the two cases stay
+   // distinguishable.
+   //
+   // An empty buffer is not a request to do anything; it completes immediately with success and
+   // zero bytes, wherever the body stands.
+   //
    virtual void async_read_some(asio::mutable_buffer buffer, ReadSomeHandler&& handler) = 0;
    virtual void detach() = 0;
    virtual void destroy() {};
@@ -150,7 +188,22 @@ class Writer : public std::enable_shared_from_this<Writer>
    virtual ~Writer() = default;
    virtual asio::any_io_executor get_executor() const noexcept = 0;
    virtual void content_length(std::optional<size_t> content_length) = 0;
-   virtual void async_write(WriteHandler&& handler, asio::const_buffer buffer) = 0;
+
+   //
+   // Writes \p buffer and, if \p eof is set, ends the outgoing body after it. The two travel
+   // together on purpose: every backend can put the last bytes of a body and the flag that ends
+   // it into the same protocol element -- one DATA frame with END_STREAM (HTTP/2), one QUIC
+   // STREAM frame with FIN (HTTP/3), one last chunk (HTTP/1.1) -- so a message that ends with
+   // data needs no second, empty write to close it out.
+   //
+   // Every implementation answers the same entry ladder, in this order: an empty buffer with
+   // \p eof clear writes nothing at all and completes immediately with success, wherever the
+   // body stands -- it is not, as it once was, how a body is ended. Once the body has been ended,
+   // writing data -- through either entry point -- completes with \c errc::broken_pipe, while
+   // re-ending it with no data attached is an idempotent no-op. Only then do stream-level
+   // failures (closed, cancelled) get their say.
+   //
+   virtual void async_write(WriteHandler&& handler, asio::const_buffer buffer, bool eof) = 0;
    virtual void detach() = 0;
    virtual void destroy() {};
 };
@@ -195,6 +248,9 @@ boost::system::error_code code(const std::exception_ptr& ptr);
 /// Get error message from exception pointer, as used in the completion signature of \c co_spawn().
 std::string what(const std::exception_ptr& ptr);
 
+/// Get error message from a boost::system_error, as thrown by boost ASIO if not caught.
+std::string what(const boost::system::system_error& ex);
+
 /// Get error message from \c boost::system::error_code, used by ASIO.
 std::string what(const boost::system::error_code& ec);
 

include/anyhttp/detail/h2_session_details.hpp

@@ -7,7 +7,6 @@
 
 #include "anyhttp/any_async_stream.hpp"
 #include "anyhttp/h2_session.hpp"
-#include "anyhttp/session.hpp"
 
 #include <boost/asio/basic_stream_socket.hpp>
 #include <boost/asio/buffer.hpp>

include/anyhttp/formatter.hpp

@@ -11,11 +11,7 @@
 #include <boost/url/pct_string_view.hpp>
 
 #include <thread>
-
 #include <format>
-#include <ranges>
-
-namespace rv = std::ranges::views;
 
 // =================================================================================================
 

include/anyhttp/h2_backend.hpp

@@ -27,15 +27,17 @@ using SslStream = boost::asio::ssl::stream<boost::asio::ip::tcp::socket>;
 
 std::shared_ptr<Session::Impl> make_server_session(server::Server::Impl& server,
                                                    boost::asio::any_io_executor executor,
-                                                   SslStream&& stream);
+                                                   boost::asio::ip::tcp::socket&& socket);
 
 std::shared_ptr<Session::Impl> make_server_session(server::Server::Impl& server,
                                                    boost::asio::any_io_executor executor,
-                                                   AnyAsyncStream&& stream);
+                                                   SslStream&& stream);
 
 std::shared_ptr<Session::Impl> make_server_session(server::Server::Impl& server,
                                                    boost::asio::any_io_executor executor,
-                                                   boost::asio::ip::tcp::socket&& socket);
+                                                   AnyAsyncStream&& stream);
+
+// -------------------------------------------------------------------------------------------------
 
 std::shared_ptr<Session::Impl> make_client_session(client::Client::Impl& client,
                                                    boost::asio::any_io_executor executor,

include/anyhttp/h2_session.hpp

@@ -160,6 +160,8 @@ class ServerReference
    server::Server::Impl* m_server = nullptr;
 };
 
+// -------------------------------------------------------------------------------------------------
+
 template <typename Stream>
 class ServerSession : public ServerReference, public NGHttp2SessionImpl<Stream>
 {
@@ -178,12 +180,10 @@ class ServerSession : public ServerReference, public NGHttp2SessionImpl<Stream>
 public:
    ServerSession(server::Server::Impl& parent, any_io_executor executor, Stream&& stream);
 
-   // void async_submit(SubmitHandler&& handler, boost::urls::url url, const Fields& headers)
-   // override;
    awaitable<void> do_session(Buffer&& data) override;
 };
 
-// -------------------------------------------------------------------------------------------------
+// =================================================================================================
 
 class ClientReference
 {
@@ -199,6 +199,8 @@ class ClientReference
    client::Client::Impl* m_client = nullptr;
 };
 
+// -------------------------------------------------------------------------------------------------
+
 template <typename Stream>
 class ClientSession : public ClientReference, public NGHttp2SessionImpl<Stream>
 {
@@ -217,8 +219,6 @@ class ClientSession : public ClientReference, public NGHttp2SessionImpl<Stream>
 public:
    ClientSession(client::Client::Impl& parent, any_io_executor executor, Stream&& stream);
 
-   // void async_submit(SubmitHandler&& handler, boost::urls::url url, const Fields& headers)
-   // override;
    awaitable<void> do_session(Buffer&& data) override;
 };