diff --git a/README.md b/README.md index 3846a29..39f9938 100644 --- a/README.md +++ b/README.md @@ -28,13 +28,32 @@ awaitable echo(server::Request request, server::Response response) std::array 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 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) - async_write(buffer) + async_write(buffer, eof) detach() destroy() } diff --git a/include/anyhttp/any_async_stream.hpp b/include/anyhttp/any_async_stream.hpp index 68abba0..6b56faf 100644 --- a/include/anyhttp/any_async_stream.hpp +++ b/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( [this](ReadWriteHandler handler, const ConstBufferSequence& buffers) { -#if defined(USE_ASIO_LINEARISE) - using namespace asio; - using Adapter = detail::buffer_sequence_adapter; - std::array 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( [this](ReadWriteHandler handler, const MutableBufferSequence& buffers) { -#if defined(USE_ASIO_LINEARISE) - using namespace asio; - using Adapter = detail::buffer_sequence_adapter; - 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); } }; diff --git a/include/anyhttp/buffer_array.hpp b/include/anyhttp/buffer_array.hpp index f15f57c..f17a61a 100644 --- a/include/anyhttp/buffer_array.hpp +++ b/include/anyhttp/buffer_array.hpp @@ -14,11 +14,8 @@ #include #include -// #include -// #include #include -#include #include #include diff --git a/include/anyhttp/client.hpp b/include/anyhttp/client.hpp index 7c54046..dbb6cb7 100644 --- a/include/anyhttp/client.hpp +++ b/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 requires(boost::asio::is_mutable_buffer_sequence::value) @@ -106,6 +113,12 @@ 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 auto async_write(asio::const_buffer buffer, CompletionToken&& token = CompletionToken()) { @@ -113,13 +126,41 @@ class Request auto executor = asio::get_associated_executor(token); // , get_executor()); return asio::async_initiate( 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 + 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( + 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 + auto async_write_eof(CompletionToken&& token = CompletionToken()) + { + return async_write_eof(asio::const_buffer{}, std::forward(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; }; diff --git a/include/anyhttp/common.hpp b/include/anyhttp/common.hpp index eb87ef6..ebb5268 100644 --- a/include/anyhttp/common.hpp +++ b/include/anyhttp/common.hpp @@ -6,6 +6,7 @@ #include #include +#include #include #include #include @@ -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 +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), + ... args = std::forward(args)]() mutable { // + std::move(handler)(std::move(args)...); + }); +} + +// ================================================================================================= + namespace impl { class Reader : public std::enable_shared_from_this @@ -139,6 +165,18 @@ class Reader : public std::enable_shared_from_this virtual ~Reader() = default; virtual asio::any_io_executor get_executor() const noexcept = 0; virtual std::optional 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 virtual ~Writer() = default; virtual asio::any_io_executor get_executor() const noexcept = 0; virtual void content_length(std::optional 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); diff --git a/include/anyhttp/detail/h2_session_details.hpp b/include/anyhttp/detail/h2_session_details.hpp index 03c1155..0b32fe3 100644 --- a/include/anyhttp/detail/h2_session_details.hpp +++ b/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 #include diff --git a/include/anyhttp/formatter.hpp b/include/anyhttp/formatter.hpp index a6df6b6..6fc1022 100644 --- a/include/anyhttp/formatter.hpp +++ b/include/anyhttp/formatter.hpp @@ -11,11 +11,7 @@ #include #include - #include -#include - -namespace rv = std::ranges::views; // ================================================================================================= diff --git a/include/anyhttp/h2_backend.hpp b/include/anyhttp/h2_backend.hpp index adf5f75..fa092d5 100644 --- a/include/anyhttp/h2_backend.hpp +++ b/include/anyhttp/h2_backend.hpp @@ -27,15 +27,17 @@ using SslStream = boost::asio::ssl::stream; std::shared_ptr make_server_session(server::Server::Impl& server, boost::asio::any_io_executor executor, - SslStream&& stream); + boost::asio::ip::tcp::socket&& socket); std::shared_ptr make_server_session(server::Server::Impl& server, boost::asio::any_io_executor executor, - AnyAsyncStream&& stream); + SslStream&& stream); std::shared_ptr make_server_session(server::Server::Impl& server, boost::asio::any_io_executor executor, - boost::asio::ip::tcp::socket&& socket); + AnyAsyncStream&& stream); + +// ------------------------------------------------------------------------------------------------- std::shared_ptr make_client_session(client::Client::Impl& client, boost::asio::any_io_executor executor, diff --git a/include/anyhttp/h2_session.hpp b/include/anyhttp/h2_session.hpp index fd2c360..b5c895a 100644 --- a/include/anyhttp/h2_session.hpp +++ b/include/anyhttp/h2_session.hpp @@ -160,6 +160,8 @@ class ServerReference server::Server::Impl* m_server = nullptr; }; +// ------------------------------------------------------------------------------------------------- + template class ServerSession : public ServerReference, public NGHttp2SessionImpl { @@ -178,12 +180,10 @@ class ServerSession : public ServerReference, public NGHttp2SessionImpl 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 do_session(Buffer&& data) override; }; -// ------------------------------------------------------------------------------------------------- +// ================================================================================================= class ClientReference { @@ -199,6 +199,8 @@ class ClientReference client::Client::Impl* m_client = nullptr; }; +// ------------------------------------------------------------------------------------------------- + template class ClientSession : public ClientReference, public NGHttp2SessionImpl { @@ -217,8 +219,6 @@ class ClientSession : public ClientReference, public NGHttp2SessionImpl 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 do_session(Buffer&& data) override; }; diff --git a/include/anyhttp/h2_stream.hpp b/include/anyhttp/h2_stream.hpp index 41d0284..4e56329 100644 --- a/include/anyhttp/h2_stream.hpp +++ b/include/anyhttp/h2_stream.hpp @@ -7,6 +7,7 @@ #include #include +#include #include #include #include @@ -17,6 +18,9 @@ #include #include +#include +#include +#include namespace anyhttp::nghttp2 { @@ -41,6 +45,11 @@ class NGHttp2Reader : public Interface boost::url_view url() const override; NGHttp2Stream* stream; + asio::any_io_executor executor; // kept as a copy so a detached reader can still complete + + /// What a read past detach() reports, latched by detach(): the stream may be gone, but a body + /// that was read to its clean end keeps ending in \c eof, a truncated one in partial_message. + error_code detached_ec{boost::beast::http::error::partial_message}; }; // ------------------------------------------------------------------------------------------------- @@ -54,13 +63,15 @@ class NGHttp2Writer : public Base asio::any_io_executor get_executor() const noexcept override; void content_length(std::optional content_length) override; - void async_write(WriteHandler&& handler, asio::const_buffer buffer) override; + void async_write(WriteHandler&& handler, asio::const_buffer buffer, bool eof) override; void detach() override; void async_submit(StatusHandler&& handler, unsigned int status_code, const Fields& headers); void async_get_response(client::Request::GetResponseHandler&& handler); NGHttp2Stream* stream; + asio::any_io_executor executor; // kept as a copy so a detached writer can still complete + bool detached_eof_submitted = false; // latched by detach(): the body was cleanly ended std::optional m_content_length; }; @@ -79,7 +90,7 @@ class NGHttp2Stream : public std::enable_shared_from_this /** * True after we have received an EOF flag from the peer. After this, no more buffers will be - * added. But the might be still some buffers left to be deliver to the user. + * added. But there might be still some buffers left to be deliver to the user. */ bool eof_received = false; @@ -133,6 +144,16 @@ class NGHttp2Stream : public std::enable_shared_from_this asio::const_buffer write_buffer; // undefined unless write_handler is set WriteHandler write_handler; 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. + // + bool eof_requested = false; bool eof_submitted = false; // @@ -149,6 +170,13 @@ class NGHttp2Stream : public std::enable_shared_from_this std::string logPrefix; std::string method; boost::urls::url url; + + // + // Headers as they arrive from the peer. They are only collected while debug logging is on, + // because they are logged as one block after the request or status line, which is only known + // once all headers of the frame have been seen. See log_received_headers(). + // + std::vector> received_headers; std::optional status_code; std::optional content_length; @@ -243,7 +271,7 @@ class NGHttp2Stream : public std::enable_shared_from_this // ---------------------------------------------------------------------------------------------- - void async_write(WriteHandler handler, asio::const_buffer buffer); + void async_write(WriteHandler handler, asio::const_buffer buffer, bool eof); void resume(); @@ -257,6 +285,9 @@ class NGHttp2Stream : public std::enable_shared_from_this void deliver_response(); void on_request(); + /// Log and discard the headers collected by on_header_callback(). + void log_received_headers(); + impl::Reader* reader = nullptr; impl::Writer* writer = nullptr; diff --git a/include/anyhttp/h3_common.hpp b/include/anyhttp/h3_common.hpp index 7bcbb41..63e6b1e 100644 --- a/include/anyhttp/h3_common.hpp +++ b/include/anyhttp/h3_common.hpp @@ -1,7 +1,5 @@ #pragma once -#include "anyhttp/common.hpp" - #include #include diff --git a/include/anyhttp/h3_session.hpp b/include/anyhttp/h3_session.hpp index d47106e..580cb8e 100644 --- a/include/anyhttp/h3_session.hpp +++ b/include/anyhttp/h3_session.hpp @@ -1,7 +1,5 @@ #pragma once -#include "anyhttp/common.hpp" -#include "anyhttp/h3_common.hpp" #include "anyhttp/session_impl.hpp" #include @@ -35,7 +33,7 @@ class Http3Stream; // loop, the write loop, the timers, the flow control and every callback bridge below are shared. // // What the roles still own themselves is how datagrams reach the connection (the server -// demultiplexes many connections over one shared socket by connection ID, the client owns a +// de-multiplexes many connections over one shared socket by connection ID, the client owns a // connect()ed socket with exactly one peer), how a dead connection is torn down, and how streams // come into being (accepted from the peer vs. opened by async_submit()). Those are the virtuals // at the bottom. diff --git a/include/anyhttp/h3_stream.hpp b/include/anyhttp/h3_stream.hpp index 0936e15..c963ee0 100644 --- a/include/anyhttp/h3_stream.hpp +++ b/include/anyhttp/h3_stream.hpp @@ -1,7 +1,6 @@ #pragma once #include "anyhttp/common.hpp" -#include "anyhttp/h3_common.hpp" #include #include @@ -141,11 +140,19 @@ class Http3Stream : public std::enable_shared_from_this std::vector> in_flight_writes; // Staged: retired chunks, kept alive for // the stream's lifetime because ngtcp2 may // still retransmit from them + // + // Whether the active write ends the body. data_reader() then hands nghttp3 + // NGHTTP3_DATA_FLAG_EOF along with the write's last bytes -- one QUIC STREAM frame carrying + // both the tail of the body and the FIN -- rather than needing a write of its own for it. + // bool write_is_eof = false; WriteHandler write_handler; uint64_t write_token = 0; uint64_t next_write_token = 1; - bool eof_submitted = false; // user signalled EOF via an empty write + bool eof_submitted = false; // user ended the body via async_write_eof() + bool fin_offered = false; // ... and data_reader() has handed the FIN flag to nghttp3; between + // the two lies a cancelled async_write_eof(), which rolls + // eof_submitted back when the FIN is still owed // // Lifecycle. @@ -164,10 +171,16 @@ class Http3Stream : public std::enable_shared_from_this void on_eof(); void call_read_handler(); + /// True once the peer ended the body and every byte of it has been delivered to the reader. + bool reading_finished() const noexcept + { + return eof_received && read_head.size() == 0 && incoming.size() == 0; + } + // // Data flow from user land back to nghttp3 (outgoing body). // - void start_write(WriteHandler&& handler, asio::const_buffer buffer); + void start_write(WriteHandler&& handler, asio::const_buffer buffer, bool eof); nghttp3_ssize data_reader(nghttp3_vec* vec, size_t veccnt, uint32_t* pflags); void on_write_acked(size_t n); // ZeroCopy: nghttp3 acked_stream_data void on_write_offered(size_t n); // Staged: bytes ngtcp2 committed to a packet @@ -221,7 +234,10 @@ template class Http3Reader : public Interface { public: - explicit Http3Reader(Http3Stream& s) : stream(&s) { s.reader = this; } + explicit Http3Reader(Http3Stream& s) : stream(&s), executor(s.get_executor()) + { + s.reader = this; + } ~Http3Reader() override { if (stream) @@ -231,11 +247,7 @@ class Http3Reader : public Interface } } - asio::any_io_executor get_executor() const noexcept override - { - assert(stream); - return stream->get_executor(); - } + asio::any_io_executor get_executor() const noexcept override { return executor; } std::optional content_length() const noexcept override { @@ -253,17 +265,25 @@ class Http3Reader : public Interface void async_read_some(asio::mutable_buffer buffer, ReadSomeHandler&& handler) override { - if (!stream) + // + // An empty buffer is not a request to read anything: complete right away, without looking + // at whether the body has ended or the stream is even still there -- as ASIO does for a + // zero-length read. + // + if (asio::buffer_size(buffer) == 0) { - std::move(handler)(boost::beast::http::error::partial_message, 0); + complete_immediately(std::move(handler), executor, error_code{}, size_t{0}); return; } - if (asio::buffer_size(buffer) == 0) + + // + // The stream is gone; detach() latched how the body stood at that point, so a cleanly + // finished body keeps reporting eof (as the Reader contract requires) and a truncated one + // keeps reporting partial_message. + // + if (!stream) { - asio::any_completion_executor ex = - asio::get_associated_immediate_executor(handler, stream->get_executor()); - ex.execute([handler = std::move(handler)]() mutable - { std::move(handler)(boost::system::error_code{}, 0); }); + complete_immediately(std::move(handler), executor, detached_ec, size_t{0}); return; } @@ -291,9 +311,24 @@ class Http3Reader : public Interface stream->call_read_handler(); } - void detach() override { stream = nullptr; } + void detach() override + { + // + // The stream is going away first (session teardown outliving this exchange). Remember how + // the body stood, so that reads issued from now on keep answering per the Reader contract. + // + assert(stream); + detached_ec = stream->reading_finished() + ? error_code{asio::error::eof} + : error_code{boost::beast::http::error::partial_message}; + stream = nullptr; + } Http3Stream* stream; + asio::any_io_executor executor; // kept as a copy so a detached reader can still complete + + /// What a read past detach() reports: eof for a body read to its clean end, else truncation. + error_code detached_ec{boost::beast::http::error::partial_message}; }; // ------------------------------------------------------------------------------------------------- @@ -302,7 +337,10 @@ template class Http3Writer : public Base { public: - explicit Http3Writer(Http3Stream& s) : stream(&s) { s.writer = this; } + explicit Http3Writer(Http3Stream& s) : stream(&s), executor(s.get_executor()) + { + s.writer = this; + } ~Http3Writer() override { if (stream) @@ -312,11 +350,7 @@ class Http3Writer : public Base } } - asio::any_io_executor get_executor() const noexcept override - { - assert(stream); - return stream->get_executor(); - } + asio::any_io_executor get_executor() const noexcept override { return executor; } void content_length(std::optional len) override { @@ -324,16 +358,33 @@ class Http3Writer : public Base stream->response_content_length = len; } - void async_write(WriteHandler&& handler, asio::const_buffer buffer) override + void async_write(WriteHandler&& handler, asio::const_buffer buffer, bool eof) override { - if (!stream || stream->closed) + if (stream) { - std::move(handler)( - boost::system::errc::make_error_code(boost::system::errc::connection_reset)); + // everything -- including a write against a stream ngtcp2 has already torn down -- is + // start_write()'s to decide, so that ending a body twice and writing past its end are + // answered the same way whatever became of the stream since + stream->start_write(std::move(handler), buffer, eof); return; } - stream->start_write(std::move(handler), buffer); + // + // The stream itself is gone, but the entry ladder of the Writer contract still applies, + // answered from the state detach() latched: an empty non-EOF write stays a free no-op, a + // body that was cleanly ended keeps answering as such -- bare re-end idempotent, data + // broken_pipe -- and only a stream that vanished mid-body is a connection error. + // + const bool empty = asio::buffer_size(buffer) == 0; + error_code ec; + if (empty && !eof) + ec = {}; + else if (detached_body_ended) + ec = empty ? error_code{} + : boost::system::errc::make_error_code(boost::system::errc::broken_pipe); + else + ec = boost::system::errc::make_error_code(boost::system::errc::connection_reset); + complete_immediately(std::move(handler), executor, ec); } void async_submit(StatusHandler&& handler, unsigned int status_code, const Fields& fields) @@ -348,9 +399,18 @@ class Http3Writer : public Base std::move(handler)(boost::system::error_code{}); } - void detach() override { stream = nullptr; } + void detach() override + { + // remember whether the body was cleanly ended -- intent accepted *and* the FIN handed to + // nghttp3 -- so writes issued after this still answer per the Writer contract + assert(stream); + detached_body_ended = stream->eof_submitted && stream->fin_offered; + stream = nullptr; + } Http3Stream* stream; + asio::any_io_executor executor; // kept as a copy so a detached writer can still complete + bool detached_body_ended = false; // latched by detach(), see there }; // ================================================================================================= diff --git a/include/anyhttp/request_handlers.hpp b/include/anyhttp/request_handlers.hpp index 8a806ce..c463e2b 100644 --- a/include/anyhttp/request_handlers.hpp +++ b/include/anyhttp/request_handlers.hpp @@ -2,8 +2,9 @@ #include "anyhttp/client.hpp" #include "anyhttp/server.hpp" +#include "anyhttp/literals.hpp" -#include +#include #include #include #include @@ -12,9 +13,11 @@ #include #include #include +#include #include #include +#include #include @@ -72,12 +75,43 @@ awaitable discard(server::Request request, server::Response response); // ================================================================================================= -awaitable send(client::Request& request, size_t bytes); +awaitable generate(client::Request& request, size_t bytes); awaitable read(client::Response& response); -awaitable count(client::Response& response); + +// +// Reads and discards whatever is left of an incoming body, and returns how much that was. +// +// This is the plain shape of an ASIO read loop against the anyhttp reader interface: read until +// EOF, and let anything else -- a reset stream, a connection that went away mid-body -- come out +// as an exception. +// +template +awaitable drain(Reader& reader) +{ + size_t bytes = 0; + std::array buffer; + for (;;) + { + auto [ec, n] = co_await reader.async_read_some(asio::buffer(buffer), asio::as_tuple); + bytes += n; + + // the regular end of the body is not something to report as an error + if (ec == asio::error::eof) + { + logd("drain: EOF after reading {} bytes", bytes); + co_return bytes; + } + else if (ec) + { + logw("drain: \x1b[1;31m{}\x1b[0m after reading {} bytes, throwing", what(ec), bytes); + throw boost::system::system_error(ec); + } + } +} + awaitable> try_receive(client::Response& response); awaitable try_receive(client::Response& response, boost::system::error_code& ec); -awaitable read_response(client::Request& request); +awaitable count_response(client::Request& request); awaitable> try_read_response(client::Request& request); awaitable send_eof(client::Request& request); @@ -104,7 +138,7 @@ awaitable send(Writer& request, Range range) // For a non-contiguous range, we need to copy into a buffer first. // template - requires (!std::ranges::contiguous_range) + requires(!std::ranges::contiguous_range) awaitable send(Writer& request, Range range) { logd("send:"); @@ -114,7 +148,7 @@ awaitable send(Writer& request, Range range) { const auto end = std::ranges::copy(chunk, buffer.data()).out; const auto n = end - buffer.data(); - bytes += n; // FIXME: count after async_write + bytes += n; // FIXME: count after async_write #if 0 #if defined(NDEBUG) co_await request.async_write(asio::buffer(buffer.data(), n)); @@ -154,17 +188,6 @@ awaitable send(Writer& request, Range range) template awaitable sendAndDrop(client::Request request, Range range) { -#if 0 - try - { - co_return co_await send(request, std::move(range)); - } - catch (const boost::system::system_error& ec) - { - loge("sendAndDrop: (range) {}", ec.code().message()); - throw; - } -#else using namespace asio; auto ex = co_await this_coro::executor; if (auto [ep] = co_await co_spawn(ex, send(request, std::move(range)), as_tuple); ep) @@ -172,7 +195,6 @@ awaitable sendAndDrop(client::Request request, Range range) loge("sendAndDrop: {}", what(ep)); std::rethrow_exception(ep); } -#endif } // ------------------------------------------------------------------------------------------------- @@ -185,9 +207,9 @@ awaitable sendAndForceEOF(Writer& request, Range range) if (auto [ep] = co_await co_spawn(ex, send(request, std::move(range)), as_tuple); ep) { loge("sendAndForceEOF: {}", what(ep)); - co_await asio::this_coro::reset_cancellation_state(); + co_await this_coro::reset_cancellation_state(); } - auto [ec] = co_await request.async_write({}, as_tuple(deferred)); + std::ignore = co_await request.async_write_eof(as_tuple); } // ------------------------------------------------------------------------------------------------- @@ -200,20 +222,17 @@ inline awaitable generate(server::Request request, server::Response respon { namespace rv = std::ranges::views; - size_t length = 0; - const auto param = request.url().params().get_or("length"); - auto [ptr, ec] = std::from_chars(param.data(), param.data() + param.size(), length); - if (ec != std::errc{} || ptr != param.data() + param.size()) + const auto length = request.get_param_as("length"); + if (!length) { - logw("generate: invalid length '{}'", param); co_await response.async_submit(400, {}); - co_await response.async_write({}); + co_await response.async_write_eof(); co_return; } - logd("generate: {} bytes", length); - co_await response.async_submit(200, fields({{"Content-Length", length}})); - co_await sendAndForceEOF(response, rv::iota(uint8_t(0)) | rv::take(length)); + logd("generate: {} bytes", *length); + co_await response.async_submit(200, fields({{"Content-Length", *length}})); + co_await sendAndForceEOF(response, rv::iota(uint8_t(0)) | rv::take(*length)); } // ------------------------------------------------------------------------------------------------- diff --git a/include/anyhttp/server.hpp b/include/anyhttp/server.hpp index 5a9bb6e..280653e 100644 --- a/include/anyhttp/server.hpp +++ b/include/anyhttp/server.hpp @@ -4,11 +4,18 @@ #include #include -#include +#include #include +#include + #include +#include +#include +#include +#include + using namespace std::chrono_literals; namespace anyhttp::server @@ -61,7 +68,49 @@ class Request boost::url_view url() const; std::optional content_length() const noexcept; + /** + * Looks up a query parameter and converts its value to \c T. + * + * Returns \c std::nullopt if the parameter is missing, has no value at all, or if its value + * does not convert to \c T -- the latter is logged as a warning. Use \c value_or() for a + * default: + * + * \code + * auto delay = request.get_param_as("delay").value_or(0); + * \endcode + */ + template + std::optional get_param_as(std::string_view name) const + { + const auto u = url(); // keep the url_view alive: params() only references it + const auto params = u.params(); + const auto it = params.find(name); + if (it == params.end() || !(*it).has_value) + return std::nullopt; + + const std::string& value = (*it).value; + + // + // lexical_cast wraps a negative number around into an unsigned type -- "-1" arrives as + // SIZE_MAX -- which is never what a caller asking for an unsigned type wants. + // + if (std::is_integral_v && std::is_unsigned_v && value.starts_with('-')) + ; // invalid value (reported below) + else if (T converted; boost::conversion::try_lexical_convert(value, converted)) + return converted; + + logw("get_param_as: invalid value '{}' for parameter '{}'", value, name); + return std::nullopt; + } + public: + /** + * Reads a part of the request 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 auto async_read_some(boost::asio::mutable_buffer buffer, CompletionToken&& token = CompletionToken()) @@ -111,43 +160,65 @@ class Response auto async_submit(unsigned int status_code, const Fields& headers, CompletionToken&& token = CompletionToken()) { + // binding the executor lets tokens that need one -- cancel_after's timer -- find it here return boost::asio::async_initiate( - [this](StatusHandler handler, unsigned int status_code, const Fields& headers) { // - async_submit_any(std::move(handler), status_code, headers); - }, + asio::bind_executor(get_executor(), + [this](StatusHandler handler, unsigned int status_code, + const Fields& headers) { // + async_submit_any(std::move(handler), status_code, headers); + }), token, status_code, headers); } + /** + * Writes \p buffer as part of the response body, which stays open for more. + * + * An empty buffer writes nothing and completes immediately -- use \c async_write_eof() to end + * the body. + */ template auto async_write(asio::const_buffer buffer, CompletionToken&& token = CompletionToken()) { + // binding the executor lets tokens that need one -- cancel_after's timer -- find it here return boost::asio::async_initiate( - [this](WriteHandler handler, asio::const_buffer buffer) { // - async_write_any(std::move(handler), buffer); - }, + asio::bind_executor(get_executor(), + [this](WriteHandler handler, asio::const_buffer buffer) { // + async_write_any(std::move(handler), buffer, false); + }), token, buffer); } - // https://github.com/chriskohlhoff/asio/blob/231cb29bab30f82712fcd54faaea42424cc6e710/asio/src/tests/unit/co_composed.cpp#L45 + /** + * Writes \p buffer as the last part of the response 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 auto async_write_eof(asio::const_buffer buffer, CompletionToken&& token = CompletionToken()) { - return asio::async_initiate( - asio::co_composed( - [this](auto state, asio::const_buffer buffer, - asio::any_io_executor executor) mutable -> void { // - // FIXME: error handling - co_await async_write(buffer); - co_await async_write({}); - co_return {boost::system::error_code{}}; - }, - get_executor()), - token, buffer, get_executor()); + // binding the executor lets tokens that need one -- cancel_after's timer -- find it here + return boost::asio::async_initiate( + asio::bind_executor(get_executor(), + [this](WriteHandler handler, asio::const_buffer buffer) { // + async_write_any(std::move(handler), buffer, true); + }), + token, buffer); + } + + /// Ends the response body without writing anything more. + template + auto async_write_eof(CompletionToken&& token = CompletionToken()) + { + return async_write_eof(asio::const_buffer{}, std::forward(token)); } private: void async_submit_any(StatusHandler&& handler, unsigned int status_code, const Fields& headers); - void async_write_any(WriteHandler&& handler, asio::const_buffer buffer); + void async_write_any(WriteHandler&& handler, asio::const_buffer buffer, bool eof); std::shared_ptr impl; }; diff --git a/src/client.cpp b/src/client.cpp index e9eb58e..33f7200 100644 --- a/src/client.cpp +++ b/src/client.cpp @@ -35,10 +35,10 @@ Request::~Request() { reset(); } // ------------------------------------------------------------------------------------------------- -void Request::async_write_any(WriteHandler&& handler, asio::const_buffer buffer) +void Request::async_write_any(WriteHandler&& handler, asio::const_buffer buffer, bool eof) { if (impl) - impl->async_write(std::move(handler), buffer); + impl->async_write(std::move(handler), buffer, eof); else std::move(handler)(boost::asio::error::bad_descriptor); } diff --git a/src/client_main.cpp b/src/client_main.cpp index fc576fd..36e2ce5 100644 --- a/src/client_main.cpp +++ b/src/client_main.cpp @@ -1,4 +1,5 @@ #include "anyhttp/client.hpp" +#include "anyhttp/request_handlers.hpp" // for drain() #include "anyhttp/session.hpp" #include @@ -23,30 +24,8 @@ using namespace boost::asio::experimental::awaitable_operators; awaitable send(Request& request, std::string_view hello) { logd("send: sending string of {} bytes...", hello.size()); - co_await request.async_write(asio::buffer(hello)); - logd("send: sending string of {} bytes... done, sending EOF...", hello.size()); - co_await request.async_write({}); - logd("send: sending string of {} bytes... done, sending EOF... done", hello.size()); -} - -awaitable read_response(Request& request) -{ - logd("receive: waiting for response..."); - auto response = co_await request.async_get_response(); - logd("receive: waiting for response... done"); - size_t total = 0; - std::array buffer; - for (;;) - { - logd("receive: async_read_some..."); - size_t n = co_await response.async_read_some(asio::buffer(buffer)); - logd("receive: async_read_some... done, read {} bytes", n); - if (n == 0) - break; - total += n; - } - logd("receive: done, total {} bytes", total); - co_return total; + co_await request.async_write_eof(asio::buffer(hello)); + logd("send: sending string of {} bytes... done", hello.size()); } awaitable do_request(Session& session, boost::urls::url url) @@ -64,7 +43,7 @@ awaitable do_request(Session& session, boost::urls::url url) auto result = co_await (send(request, bytes) && receive(request)); assert(bytes == result); #else - co_await (send(request, hello) && read_response(request)); + co_await (send(request, hello) && count_response(request)); logi("do_request: done"); #endif } diff --git a/src/common.cpp b/src/common.cpp index 1ff2de6..72fea3a 100644 --- a/src/common.cpp +++ b/src/common.cpp @@ -1,4 +1,5 @@ #include +#include namespace anyhttp { @@ -69,7 +70,7 @@ boost::system::error_code code(const std::exception_ptr& ptr) } std::string what(const boost::system::error_code& ec) { return ec.message(); } - +std::string what(const boost::system::system_error& ex) { return what(ex.code()); } std::string what(const std::exception_ptr& ptr) { if (!ptr) diff --git a/src/file_handler.cpp b/src/file_handler.cpp index 26562f0..58138b5 100644 --- a/src/file_handler.cpp +++ b/src/file_handler.cpp @@ -1,6 +1,7 @@ #include "anyhttp/file_handler.hpp" + #include "anyhttp/formatter.hpp" // IWYU pragma: keep -#include "anyhttp/request_handlers.hpp" // for send() +#include "anyhttp/request_handlers.hpp" // for drain() #include #include @@ -197,7 +198,7 @@ unsigned status_for(const error_code& ec) awaitable respond(server::Response& response, unsigned status) { co_await response.async_submit(status, fields({{"Content-Length", 0}})); - co_await response.async_write({}); + co_await response.async_write_eof(); } // @@ -330,9 +331,7 @@ awaitable serve_file(server::Request request, server::Response response, f // has to close the connection when a handler leaves the request unparsed, which would // truncate the response we are about to write. // - std::array discard; - while (co_await request.async_read_some(asio::buffer(discard)) > 0) - ; + co_await drain(request); const std::string path = request.url().path(); const auto entry = g_cache.get(path, prefix, root); @@ -355,10 +354,12 @@ awaitable serve_file(server::Request request, server::Response response, f // evicted or replaced meanwhile. Note that touching a mapped page may block on disk I/O, // which no amount of chunking would avoid. // - if (file.size() > 0) - co_await send(response, file.bytes()); // an empty write already means EOF, don't send two - - co_await response.async_write({}); + // Body and the end of it go out in one call: the mapped pages reach the transport by + // reference, and whatever ends the message -- last chunk, END_STREAM, FIN -- travels with the + // last of them instead of costing a write of its own. An empty file, which cannot be mmap()ed + // at all, is then simply an empty buffer and ends the same way. + // + co_await response.async_write_eof(asio::buffer(file.bytes())); } } // namespace anyhttp diff --git a/src/h1_session.cpp b/src/h1_session.cpp index 22118c9..3cd57be 100644 --- a/src/h1_session.cpp +++ b/src/h1_session.cpp @@ -67,7 +67,8 @@ class BeastReader : public Interface { public: inline BeastReader(BeastSession& session_, Stream& stream_, Buffer& buffer_) - : session(&session_), stream(stream_), buffer(buffer_) + : session(&session_), stream(stream_), buffer(buffer_), + m_executor(session_.get_executor()) // survives detach(), see get_executor() { parser.body_limit(std::numeric_limits::max()); } @@ -127,12 +128,24 @@ class BeastReader : public Interface assert(!reading); - if (body_buffer.size() == 0 || parser.is_done()) + // + // Everything that can be answered without touching the connection, each case with the error + // code the Reader contract prescribes for it (see common.hpp): a zero-length read is not a + // read and reports nothing, wherever the body stands; a parser that is done keeps reporting + // the end of the body, ASIO-style, session or no session; and a read past a detached, + // unfinished parser is a truncation the caller must hear about. + // + if (body_buffer.size() == 0 || parser.is_done() || !session) { - any_completion_executor ex = get_associated_immediate_executor(handler, get_executor()); - ex.execute([handler = std::move(handler)]() mutable { // - std::move(handler)(boost::system::error_code{}, 0); - }); + error_code ec; + if (body_buffer.size() == 0) + ec = {}; + else if (parser.is_done()) + ec = asio::error::eof; + else + ec = boost::beast::http::error::partial_message; + + complete_immediately(std::move(handler), get_executor(), ec, size_t{0}); return; } @@ -154,6 +167,11 @@ class BeastReader : public Interface if (ec == beast::http::error::need_buffer) ec = {}; // FIXME: maybe we should keep 'need_buffer' to avoid extra empty round trip + // + // Nothing of the body came out of this round -- either the parser is now done, in which + // case the retry below turns straight into the EOF completion above, or it just needs + // more input. Either way there is nothing to hand to the caller yet. + // if (!ec && payload == 0) async_read_some(body_buffer, std::move(handler)); else @@ -172,13 +190,14 @@ class BeastReader : public Interface stream, buffer, parser, bind_executor(ex, bind_cancellation_slot(cs, std::move(cb)))); } - asio::any_io_executor get_executor() const noexcept { return session->get_executor(); } + asio::any_io_executor get_executor() const noexcept { return m_executor; } inline auto logPrefix() const { return session ? session->logPrefix() : "DETACHED"; } BeastSession* session; Stream& stream; Buffer& buffer; Parser parser; + asio::any_io_executor m_executor; // kept as a copy so a detached reader can still complete std::optional m_status_code = 0; boost::url m_url; bool reading = false; @@ -226,7 +245,8 @@ class WriterBase : public Parent { public: inline WriterBase(BeastSession& session_, Stream& stream_) - : session(&session_), stream(stream_) + : session(&session_), stream(stream_), + m_executor(session_.get_executor()) // survives detach(), see get_executor() { } @@ -241,7 +261,7 @@ class WriterBase : public Parent // ---------------------------------------------------------------------------------------------- - asio::any_io_executor get_executor() const noexcept override { return session->get_executor(); } + asio::any_io_executor get_executor() const noexcept override { return m_executor; } void detach() override { @@ -249,41 +269,59 @@ class WriterBase : public Parent session = nullptr; } - template - requires std::invocable - inline void complete_immediately(Handler&& handler, Args&&... args) + void async_write(WriteHandler&& handler, asio::const_buffer buffer, bool eof) override { - auto ex = asio::get_associated_immediate_executor(handler, get_executor()); - ex.execute([handler = std::forward(handler), ...args = std::forward(args)] mutable { // - std::move(handler)(std::move(args)...); - }); - } + const bool empty = buffer.size() == 0; + + // + // The protocol-independent entry ladder, in the order the Writer contract in common.hpp + // prescribes: a zero-length non-EOF write is a free no-op wherever the body stands (it + // would otherwise turn into an empty chunk); after the body has ended, data has no body + // left to belong to -- through either entry point -- while a bare re-end is idempotent. + // + if (empty && !eof) + { + complete_immediately(std::move(handler), get_executor(), error_code{}); + return; + } + + if (eof_submitted) + { + if (!empty) + mloge("async_write: body has already been ended"); + complete_immediately(std::move(handler), get_executor(), + empty ? error_code{} : errc::make_error_code(errc::broken_pipe)); + return; + } - void async_write(WriteHandler&& handler, asio::const_buffer buffer) override - { if (cancelled) { mloge("async_write: already canceled"); - complete_immediately(std::move(handler), errc::make_error_code(errc::operation_canceled)); + complete_immediately(std::move(handler), get_executor(), + errc::make_error_code(errc::operation_canceled)); return; } - logd("async_write: {} bytes", buffer.size()); - assert(!writing); writing = true; - if (buffer.size() == 0) - mlogd("async_write: write EOF"); - else - mlogd("async_write: {} bytes (chunked={} content_length={})", buffer.size(), - message.chunked(), message.has_content_length()); + mlogd("async_write: {} bytes (eof={} chunked={} content_length={})", buffer.size(), eof, + message.chunked(), message.has_content_length()); + // + // The last body buffer and the end of the body go into the same serializer pass: with + // 'more' cleared, beast emits the data and the terminating chunk (or just the data, for a + // content-length delimited body) in one go, so ending a body costs no extra write. + // +#if BOOST_BEAST_VERSION < 359 // https://github.com/boostorg/beast/issues/3032 // make sure to set 'nullptr' on empty size, otherwise beast may serialize an empty chunk message.body().data = buffer.size() ? const_cast(buffer.data()) : nullptr; +#else + message.body().data = const_cast(buffer.data()); +#endif message.body().size = buffer.size(); - message.body().more = buffer.size() != 0; // empty buffer --> EOF + message.body().more = !eof; // // With 'chunked' transfer encoding, the serializer will automatically emit a chunk as @@ -298,7 +336,7 @@ class WriterBase : public Parent auto ex = get_associated_executor(handler, get_executor()); auto alloc = get_associated_allocator(handler); - auto cb = [this, self = Parent::shared_from_this(), expected = buffer.size(), + auto cb = [this, self = Parent::shared_from_this(), expected = buffer.size(), eof, handler = std::move(handler)] // (boost::system::error_code ec, size_t n) mutable { @@ -348,6 +386,14 @@ class WriterBase : public Parent } */ + // + // Only now is the body really ended: a cancelled or failed EOF write never got its + // terminating bytes onto the wire, and latching the flag at accept time would let a + // retried async_write_eof() report success for a body the peer sees as truncated. + // + if (!ec && eof) + eof_submitted = true; + std::move(handler)(ec); }; @@ -382,9 +428,11 @@ class WriterBase : public Parent Stream& stream; Message message; Serializer serializer{message}; + asio::any_io_executor m_executor; // kept as a copy so a detached writer can still complete bool writing = false; bool cancelled = false; bool response_requested = false; + bool eof_submitted = false; }; // ------------------------------------------------------------------------------------------------- @@ -478,7 +526,7 @@ class RequestWriter message.content_length(boost::none); } - asio::any_io_executor get_executor() const noexcept { return session->get_executor(); } + using super::get_executor; void async_submit(StatusHandler&& handler, unsigned int status_code, const Fields& headers) override @@ -531,7 +579,9 @@ class RequestWriter if (!ec) { http::response_parser::value_type& msg = reader->parser.get(); - mlogd("async_read_header: len={} {} {}", len, msg.result_int(), msg.reason()); + mlogd("{} {}", msg.result_int(), msg.reason()); + for (const auto& header : msg) + mlogd(" \x1b[1;34m{}\x1b[0m: {}", header.name_string(), header.value()); } else mlogw("async_read_header: {} len={}", ec.message(), len); @@ -670,9 +720,6 @@ awaitable ServerSession::do_session(Buffer&& buffer) auto& request = parser.get(); const bool need_eof = request.need_eof(); - mlogd("{} {} (need_eof={})", request.method_string(), request.target(), request.need_eof()); - for (auto& header : request) - mlogd(" \x1b[1;34m{}\x1b[0m: {}", header.name_string(), header.value()); // if (auto url = boost::urls::parse_relative_ref(request.target()); url.has_value()) if (auto url = boost::urls::parse_uri_reference(request.target()); url.has_value()) @@ -698,6 +745,10 @@ awaitable ServerSession::do_session(Buffer&& buffer) mlogw("ignoring invalid host header: {}", request[http::field::host]); } + mlogd("{} {} (need_eof={})", request.method_string(), reader->m_url.buffer(), need_eof); + for (auto& header : request) + mlogd(" \x1b[1;34m{}\x1b[0m: {}", header.name_string(), header.value()); + // // Prepare response. // @@ -832,8 +883,6 @@ template void ClientSession::async_submit(SubmitHandler&& handler, boost::urls::url url, const Fields& headers) { - mlogd("submit: {}", url.buffer()); - auto writer = std::make_unique>(*this, m_stream); wx = writer.get(); auto& request = writer->message; @@ -848,6 +897,10 @@ void ClientSession::async_submit(SubmitHandler&& handler, boost::urls::u if (!request.has_content_length()) request.chunked(true); + mlogd("{} {}", request.method_string(), url.buffer()); + for (const auto& header : request) + mlogd(" \x1b[1;34m{}\x1b[0m: {}", header.name_string(), header.value()); + // // TODO: make writer shared? put into queue // diff --git a/src/h2_session.cpp b/src/h2_session.cpp index a776e55..a2bab2e 100644 --- a/src/h2_session.cpp +++ b/src/h2_session.cpp @@ -1,25 +1,31 @@ #include "anyhttp/h2_session.hpp" -#include "anyhttp/h2_backend.hpp" #include "anyhttp/client.hpp" #include "anyhttp/common.hpp" -#include "anyhttp/detail/h2_session_details.hpp" +#include "anyhttp/detail/h2_session_details.hpp" // IWYU pragma: keep #include "anyhttp/formatter.hpp" // IWYU pragma: keep +#include "anyhttp/h2_backend.hpp" #include "anyhttp/h2_common.hpp" #include "anyhttp/h2_stream.hpp" +#include + #include #include #include #include -#include #include #include #include #include +#include + #include #include + +#include + #include #include @@ -76,8 +82,6 @@ int on_begin_headers_callback(nghttp2_session*, const nghttp2_frame* frame, void { auto handler = static_cast(user_data); - logd("[{}] on_begin_header_callback:", handler->logPrefix(frame)); - if (frame->hd.type != NGHTTP2_HEADERS || frame->headers.cat != NGHTTP2_HCAT_REQUEST) return 0; @@ -98,11 +102,16 @@ int on_header_callback(nghttp2_session* session, const nghttp2_frame* frame, con auto handler = static_cast(user_data); auto name = make_string_view(name_, namelen_); auto value = make_string_view(value_, valuelen_); - logd("[{}] \x1b[1;34m{}\x1b[0m: {}", handler->logPrefix(frame), name, value); auto stream = handler->find_stream(frame->hd.stream_id); assert(stream); + // + // Headers are logged as a block, after the request or status line, see on_frame_recv_callback(). + // + if (spdlog::default_logger_raw()->should_log(spdlog::level::debug)) + stream->received_headers.emplace_back(name, value); + try { if (name == ":method") @@ -229,6 +238,14 @@ int on_frame_recv_callback(nghttp2_session* session, const nghttp2_frame* frame, case NGHTTP2_HEADERS: { assert(stream); + using namespace boost::beast::http; + if (frame->headers.cat == NGHTTP2_HCAT_REQUEST) + logd("[{}] {} {}", stream->logPrefix, stream->method, stream->url.buffer()); + else if (frame->headers.cat == NGHTTP2_HCAT_RESPONSE && stream->status_code) + logd("[{}] {} {}", stream->logPrefix, *stream->status_code, + obsolete_reason(int_to_status(*stream->status_code))); + stream->log_received_headers(); + if (frame->headers.cat == NGHTTP2_HCAT_REQUEST) stream->on_request(); else if (frame->headers.cat == NGHTTP2_HCAT_RESPONSE) @@ -406,8 +423,9 @@ void NGHttp2Session::async_submit(SubmitHandler&& handler, boost::urls::url url, std::string scheme(url.scheme()); std::string target(url.encoded_target()); std::string authority(url.host_address()); - auto nva = std::vector(); - // nva.reserve(4 + headers.size()); + + auto nva = boost::container::small_vector(); + nva.reserve(4 + std::distance(headers.begin(), headers.end())); nva.push_back(make_nv_ls(":method", method)); nva.push_back(make_nv_ls(":scheme", scheme)); nva.push_back(make_nv_ls(":path", target)); @@ -419,10 +437,10 @@ void NGHttp2Session::async_submit(SubmitHandler&& handler, boost::urls::url url, logw("[{}] async_submit: invalid header '{}': setting pseudo headers is not allowed", stream->logPrefix, item.name_string()); - logi("[{}] async_submit: {}: {}", stream->logPrefix, item.name_string(), item.value()); nva.push_back(make_nv_ls(item.name_string(), item.value())); } + logd("[{}] {} {}", stream->logPrefix, method, url.buffer()); for (auto nv : nva) logd("[{0}] \x1b[1;34m{1:n}\x1b[0m: {1:v}", stream->logPrefix, nv); diff --git a/src/h2_stream.cpp b/src/h2_stream.cpp index a654c4a..15dce01 100644 --- a/src/h2_stream.cpp +++ b/src/h2_stream.cpp @@ -16,11 +16,15 @@ #include #include #include + #include #include #include #include + +#include + #include #include @@ -36,7 +40,8 @@ namespace anyhttp::nghttp2 // ================================================================================================= template -NGHttp2Reader::NGHttp2Reader(NGHttp2Stream& stream) : stream(&stream) +NGHttp2Reader::NGHttp2Reader(NGHttp2Stream& stream) + : stream(&stream), executor(stream.get_executor()) { stream.reader = this; } @@ -54,6 +59,16 @@ NGHttp2Reader::~NGHttp2Reader() template void NGHttp2Reader::detach() { + // + // The stream is going away first (session teardown, or the stream close outliving this + // exchange). Remember how the body stood, so that reads issued from now on keep answering per + // the Reader contract: a body that was read to its clean end keeps reporting \c eof, anything + // else is a truncation. + // + assert(stream); + detached_ec = stream->reading_finished() + ? error_code{asio::error::eof} + : error_code{boost::beast::http::error::partial_message}; stream = nullptr; } @@ -62,8 +77,7 @@ void NGHttp2Reader::detach() template asio::any_io_executor NGHttp2Reader::get_executor() const noexcept { - assert(stream); - return stream->get_executor(); + return executor; } template @@ -93,37 +107,26 @@ template void NGHttp2Reader::async_read_some(boost::asio::mutable_buffer buffer, ReadSomeHandler&& handler) { - if (!stream) + // + // An empty buffer is not a request to read anything: complete right away, without looking at + // whether the body has ended or the stream is even still there -- as ASIO does for a + // zero-length read. + // + if (asio::buffer_size(buffer) == 0) { - logw("[] async_read_some: stream already gone"); - // - // FIXME: This may return the wrong error code in some situations. For example, in the - // cancellation testcases, it may happen that the stream gets destroyed before the - // user has seen the 'partial_message' error from async_read_some(). - // - // As a solution, we might need to store the error code in the reader, so it can be - // delivered on the next call to async_read_some(). - // - // It is still a bit unclear what should happen if the user calls async_read_some() - // after that. This is arguably a misuse of the interface, when the user knows that - // the stream is gone, but it should still be handled gracefully. - // - std::move(handler)(boost::beast::http::error::partial_message, 0); - // std::move(handler)(boost::asio::error::operation_aborted, 0); + complete_immediately(std::move(handler), get_executor(), error_code{}, size_t{0}); return; } // - // Given an empty buffer, we can't do anything. This includes signalling EOF, because that is - // done using an empty buffer as well. TODO: That design doesn't match the way ASIO usually - // signals EOF, which is using asio::error::eof. We should do it like that, too. + // The stream is gone; detach() latched how the body stood at that point, so a cleanly finished + // body keeps reporting eof (as the Reader contract requires) and a truncated one keeps + // reporting partial_message. // - if (asio::buffer_size(buffer) == 0) + if (!stream) { - any_completion_executor ex = get_associated_immediate_executor(handler, get_executor()); - ex.execute([handler = std::move(handler)]() mutable { // - std::move(handler)(boost::system::error_code{}, 0); - }); + logw("[] async_read_some: stream already gone ({})", detached_ec.message()); + complete_immediately(std::move(handler), get_executor(), detached_ec, size_t{0}); return; } @@ -156,7 +159,8 @@ void NGHttp2Reader::async_read_some(boost::asio::mutable_buffer buffer, // ================================================================================================= template -NGHttp2Writer::NGHttp2Writer(NGHttp2Stream& stream) : stream(&stream) +NGHttp2Writer::NGHttp2Writer(NGHttp2Stream& stream) + : stream(&stream), executor(stream.get_executor()) { stream.writer = this; } @@ -174,6 +178,10 @@ NGHttp2Writer::~NGHttp2Writer() template void NGHttp2Writer::detach() { + // remember whether the body was cleanly ended, so writes issued after this still answer + // per the Writer contract -- see async_write() above + assert(stream); + detached_eof_submitted = stream->eof_submitted; stream = nullptr; } @@ -182,8 +190,7 @@ void NGHttp2Writer::detach() template asio::any_io_executor NGHttp2Writer::get_executor() const noexcept { - assert(stream); - return stream->get_executor(); + return executor; } template @@ -203,15 +210,15 @@ void NGHttp2Writer::async_submit(StatusHandler&& handler, unsigned int sta return; } - logd("[{}] {} {}", stream->logPrefix, status_code, - boost::beast::http::obsolete_reason(boost::beast::http::int_to_status(status_code))); + using namespace boost::beast::http; + logd("[{}] {} {}", stream->logPrefix, status_code, obsolete_reason(int_to_status(status_code))); - auto nva = std::vector(); - // nva.reserve(3 + headers.size()); + const std::string status_code_str = std::format("{}", status_code); + const std::string date = format_http_date(std::chrono::system_clock::now()); - std::string status_code_str = std::format("{}", status_code); + auto nva = boost::container::small_vector(); + nva.reserve(3 + std::distance(headers.begin(), headers.end())); nva.push_back(make_nv_ls(":status", status_code_str)); - std::string date = format_http_date(std::chrono::system_clock::now()); nva.push_back(make_nv_ls("date", date)); for (auto&& item : headers) @@ -251,15 +258,32 @@ void NGHttp2Writer::async_submit(StatusHandler&& handler, unsigned int sta } template -void NGHttp2Writer::async_write(WriteHandler&& handler, asio::const_buffer buffer) +void NGHttp2Writer::async_write(WriteHandler&& handler, asio::const_buffer buffer, bool eof) { - if (!stream) + if (stream) { - logw("[] async_write: stream already gone"); - swap_and_invoke(handler, boost::asio::error::basic_errors::connection_aborted); + stream->async_write(std::move(handler), buffer, eof); + return; } + + // + // The stream is gone, but the entry ladder of the Writer contract still applies, answered + // from the state detach() latched: an empty non-EOF write stays a free no-op, a body that was + // cleanly ended keeps answering as such -- bare re-end idempotent, data broken_pipe -- and + // only a stream that vanished mid-body is a connection error. + // + const bool empty = asio::buffer_size(buffer) == 0; + error_code ec; + if (empty && !eof) + ec = {}; + else if (detached_eof_submitted) + ec = empty ? error_code{} : errc::make_error_code(errc::broken_pipe); else - stream->async_write(std::move(handler), buffer); + { + logw("[] async_write: stream already gone"); + ec = boost::asio::error::basic_errors::connection_aborted; + } + complete_immediately(std::move(handler), executor, ec); } template @@ -438,15 +462,16 @@ void NGHttp2Stream::call_read_handler(asio::const_buffer view) } // - // Signal EOF if there is no more data to read. - // TODO: Use errc::eof instead, like ASIO does. + // No data left and the peer is done: report the end of the body the way ASIO does everywhere + // else. Keep reporting it for as long as reads keep being issued -- a handler may re-arm from + // within, and every read past the end of a body ends the same way. // - else + else if (eof_received) { - if (eof_received && m_read_handler) + while (m_read_handler) { logd("[{}] read_callback: delivering EOF...", logPrefix); - swap_and_invoke(m_read_handler, boost::system::error_code{}, 0); + swap_and_invoke(m_read_handler, error_code{asio::error::eof}, 0); // // At this point, in testcases like "IgnoreRequest", the stream may already have been // deleted. This is because invoking the read handler eventually continues a coroutine, @@ -455,8 +480,8 @@ void NGHttp2Stream::call_read_handler(asio::const_buffer view) // To avoid this, deleting the stream is post()ed in on_stream_close_callback() // logd("[{}] read_callback: delivering EOF... done", logPrefix); - return; } + return; } // @@ -558,21 +583,61 @@ NGHttp2Stream::~NGHttp2Stream() // ================================================================================================= -void NGHttp2Stream::async_write(WriteHandler handler, asio::const_buffer buffer) +void NGHttp2Stream::async_write(WriteHandler handler, asio::const_buffer buffer, bool eof) { + const bool empty = buffer.size() == 0; + + // + // The protocol-independent entry ladder, in the order the Writer contract in common.hpp + // prescribes: a zero-length non-EOF write is a free no-op wherever the body stands; after the + // body has been ended, data has no body left to belong to -- through either entry point -- + // while a bare re-end is answered by how far the end has actually gotten. + // + if (empty && !eof) + { + complete_immediately(std::move(handler), get_executor(), error_code{}); + return; + } + + if (eof_requested) + { + if (!empty) + { + loge("[{}] async_write: body has already been ended", logPrefix); + complete_immediately(std::move(handler), get_executor(), + errc::make_error_code(errc::broken_pipe)); + 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. + // + if (eof_submitted) + { + complete_immediately(std::move(handler), get_executor(), error_code{}); + return; + } + } + if (closed) { logw("[{}] async_write: stream already closed", logPrefix); - std::move(handler)(errc::make_error_code(errc::operation_canceled)); + complete_immediately(std::move(handler), get_executor(), + errc::make_error_code(errc::operation_canceled)); return; } assert(!write_handler); - logd("[{}] async_write: buffer={} is_deferred={}", logPrefix, buffer.size(), is_deferred); + logd("[{}] async_write: buffer={} eof={} is_deferred={}", logPrefix, buffer.size(), eof, + is_deferred); - assert(!write_handler); write_buffer = buffer; + eof_requested |= eof; write_handler = std::move(handler); auto slot = asio::get_associated_cancellation_slot(write_handler); @@ -680,14 +745,6 @@ ssize_t NGHttp2Stream::producer_callback(uint8_t* buf, size_t length, uint32_t* return NGHTTP2_ERR_DEFERRED; } - // - // TODO: Try to avoid the extra round trip through this callback on EOF. Currently, EOF is - // signalled by an empty send buffer, but if that was done using an extra flag, we could - // return NGHTTP2_DATA_FLAG_EOF earlier. - // - // However, that is not supported by the interface of an async write stream. Writing an empty - // buffer shouldn't do anything special. - // size_t copied = 0; if (write_buffer.size()) { @@ -702,6 +759,19 @@ ssize_t NGHttp2Stream::producer_callback(uint8_t* buf, size_t length, uint32_t* write_buffer += copied; if (write_buffer.size() == 0) { + // + // The last bytes of a write that ends the body carry END_STREAM themselves: setting the + // flag on this very DATA frame is what spares an async_write_eof() with a payload the + // extra, empty frame -- and the extra round trip through this callback -- that ending a + // body used to cost. + // + if (eof_requested) + { + logd("[{}] write callback: EOF along with the last {} bytes", logPrefix, copied); + eof_submitted = true; + *data_flags |= NGHTTP2_DATA_FLAG_EOF; + } + logd("[{}] write callback: running handler...", logPrefix); swap_and_invoke(write_handler, boost::system::error_code{}); if (write_handler) @@ -714,18 +784,26 @@ ssize_t NGHttp2Stream::producer_callback(uint8_t* buf, size_t length, uint32_t* } else { + // an empty write is only ever accepted as the end of the body, see async_write() + assert(eof_requested); logd("[{}] write callback: EOF", logPrefix); eof_submitted = true; - swap_and_invoke(write_handler, boost::system::error_code{}); *data_flags |= NGHTTP2_DATA_FLAG_EOF; + swap_and_invoke(write_handler, boost::system::error_code{}); } return copied; } +void NGHttp2Stream::log_received_headers() +{ + for (const auto& [name, value] : received_headers) + logd("[{}] \x1b[1;34m{}\x1b[0m: {}", logPrefix, name, value); + received_headers.clear(); +} + void NGHttp2Stream::on_response() { - logd("[{}] on_response:", logPrefix); has_response = true; deliver_response(); } @@ -752,8 +830,6 @@ void NGHttp2Stream::deliver_response() void NGHttp2Stream::on_request() { - logd("[{}] on_request: {}", logPrefix, url.buffer()); - // // An incoming new request should be put into a queue of the server session. From there, // new requests can then be retrieved asynchronously by the user. diff --git a/src/h3_client.cpp b/src/h3_client.cpp index fc700aa..e40b504 100644 --- a/src/h3_client.cpp +++ b/src/h3_client.cpp @@ -32,12 +32,16 @@ #include #include +#include +#include + #include #include -#include #include +#include + #include #include @@ -272,7 +276,8 @@ void Http3ClientStream::on_pseudo_header(std::string_view name, std::string_view void Http3ClientStream::on_headers_complete() { - logd("[{}] response headers: status={}", log_prefix, status_code); + using namespace boost::beast::http; + logd("[{}] {} {}", log_prefix, status_code, obsolete_reason(int_to_status(status_code))); log_headers(log_prefix, std::exchange(received_headers, {})); deliver_response(); } @@ -309,8 +314,8 @@ bool Http3ClientStream::submit_request(const boost::urls::url& request_url, cons std::string target(request_url.encoded_target()); std::string authority(request_url.host_address()); - std::vector nva; - nva.reserve(16); // small typical header count; vector will grow if needed + auto nva = boost::container::small_vector(); + nva.reserve(4 + std::distance(headers.begin(), headers.end())); nva.push_back(make_nv(":method", method_str)); nva.push_back(make_nv(":scheme", scheme)); nva.push_back(make_nv(":path", target)); @@ -321,9 +326,11 @@ bool Http3ClientStream::submit_request(const boost::urls::url& request_url, cons if (item.name_string().starts_with(':')) logw("[{}] async_submit: invalid header '{}': setting pseudo headers is not allowed", log_prefix, item.name_string()); + nva.push_back(make_nv(item.name_string(), item.value())); } + logd("[{}] {} {}", log_prefix, method_str, request_url.buffer()); return submit_headers(nva, true /* request */); } @@ -331,11 +338,11 @@ void Http3ClientStream::async_get_response(client::Request::GetResponseHandler&& { if (response_delivered) { - auto ec = asio::error::basic_errors::already_started; asio::any_completion_executor ex = asio::get_associated_immediate_executor(handler, get_executor()); - ex.execute([handler = std::move(handler), ec]() mutable - { std::move(handler)(ec, client::Response{nullptr}); }); + ex.execute([handler = std::move(handler)]() mutable { // + std::move(handler)(asio::error::basic_errors::already_started, client::Response{nullptr}); + }); return; } @@ -663,8 +670,9 @@ void Http3ClientSession::async_submit(SubmitHandler&& handler, boost::urls::url wake_write(); post(get_executor(), [handler = std::move(handler), - writer = std::make_unique(*stream)]() mutable - { std::move(handler)(boost::system::error_code{}, client::Request{std::move(writer)}); }); + writer = std::make_unique(*stream)]() mutable { // + std::move(handler)(boost::system::error_code{}, client::Request{std::move(writer)}); + }); } // ================================================================================================= @@ -705,7 +713,7 @@ awaitable> async_connect_http3(asio::any_io_execu if (!session->ready()) throw boost::system::system_error(errc::make_error_code(errc::connection_refused)); - co_return std::static_pointer_cast(session); + co_return session; } // ================================================================================================= diff --git a/src/h3_common.cpp b/src/h3_common.cpp index fd4b361..7662782 100644 --- a/src/h3_common.cpp +++ b/src/h3_common.cpp @@ -2,6 +2,7 @@ // Small helpers shared by the HTTP/3 server and client, see anyhttp/h3_common.hpp. // #include "anyhttp/h3_common.hpp" +#include "anyhttp/common.hpp" // IWYU pragma: keep #include diff --git a/src/h3_server.cpp b/src/h3_server.cpp index fef5e93..15df393 100644 --- a/src/h3_server.cpp +++ b/src/h3_server.cpp @@ -9,14 +9,14 @@ // (server::Response) into the same `RequestHandler` used by the HTTP/1.1 and HTTP/2 backends. // // What is genuinely server-side here: the TLS server context, the UDP receive path (many -// connections over one socket, demultiplexed by connection ID) and the closing/draining period +// connections over one socket, de-multiplexed by connection ID) and the closing/draining period // bookkeeping that goes with being the endpoint that stays around. All of it sits behind // `Http3Server` (anyhttp/h3_backend.hpp), so the generic server in server_impl.cpp dispatches to // HTTP/3 without ever seeing an ngtcp2 or nghttp3 type. // // Threading: with Config::use_strand, each Http3ServerSession lives on its own strand -- the unit // of serialization is the QUIC *connection* (one ngtcp2_conn/nghttp3_conn pair), not the CID: many -// CIDs alias one connection. udp_receive_loop() is a single coroutine that only demultiplexes: it +// CIDs alias one connection. udp_receive_loop() is a single coroutine that only de-multiplexes: it // copies each datagram, groups them by session and posts one batch per session to that session's // strand (process_quic_batch()), where all ngtcp2/nghttp3 work, the timers and the request // handlers run. The CID demux table is the only cross-connection state and is guarded by @@ -27,7 +27,7 @@ // migration, ECN. // -#include "anyhttp/client_impl.hpp" +#include "anyhttp/client_impl.hpp" // IWYU pragma: keep #include "anyhttp/formatter.hpp" // IWYU pragma: keep #include "anyhttp/h3_backend.hpp" #include "anyhttp/h3_common.hpp" @@ -425,7 +425,7 @@ struct QuicBatch // // The server's HTTP/3 half (see anyhttp/h3_backend.hpp): the UDP socket every QUIC connection -// shares, the receive loop demultiplexing datagrams onto them by connection ID, and the table +// shares, the receive loop de-multiplexing datagrams onto them by connection ID, and the table // doing that lookup. The sessions themselves are owned by Server::Impl's session registry, like // the TCP-based ones -- what is kept here is only what routing packets needs. // @@ -955,7 +955,7 @@ int Http3ServerImpl::udp_on_read(Endpoint& ep) // // Datagrams collected per session over the whole batch. Each session gets its accumulated // batch posted to its strand once, below, after every datagram the socket had queued has been - // demultiplexed -- so one aggregate pass on the strand can pack a whole response into a + // de-multiplexed -- so one aggregate pass on the strand can pack a whole response into a // single GSO sendmsg() instead of dribbling it out per datagram. // boost::container::small_flat_map, QuicBatch, 32> batches; @@ -1076,8 +1076,9 @@ int Http3ServerImpl::udp_on_read(Endpoint& ep) { asio::post(session->get_executor(), [self = shared_from_this(), owner = owner(), session, - batch = std::move(batch)]() mutable - { self->process_quic_batch(session, std::move(batch)); }); + batch = std::move(batch)]() mutable { // + self->process_quic_batch(session, std::move(batch)); + }); } return 0; diff --git a/src/h3_session.cpp b/src/h3_session.cpp index cf34568..cbc0d21 100644 --- a/src/h3_session.cpp +++ b/src/h3_session.cpp @@ -5,6 +5,7 @@ // #include "anyhttp/h3_session.hpp" #include "anyhttp/h3_stream.hpp" +#include "anyhttp/h3_common.hpp" #include "anyhttp/literals.hpp" #include "anyhttp/tls.hpp" @@ -312,7 +313,7 @@ void Http3Session::arm_timer_from_ngtcp2() if (closed_ || !conn_) return; - auto expiry = ngtcp2_conn_get_expiry(conn_); + auto expiry = ngtcp2_conn_get_expiry2(conn_); if (expiry == UINT64_MAX) { // diff --git a/src/h3_stream.cpp b/src/h3_stream.cpp index 3792660..56a7e4d 100644 --- a/src/h3_stream.cpp +++ b/src/h3_stream.cpp @@ -5,6 +5,7 @@ // #include "anyhttp/h3_stream.hpp" #include "anyhttp/formatter.hpp" // IWYU pragma: keep +#include "anyhttp/h3_common.hpp" #include "anyhttp/h3_session.hpp" #include @@ -98,6 +99,8 @@ void Http3Stream::call_read_handler() if (!read_handler || call_read_handler_active) return; + call_read_handler_active = true; + // // The loop below may resume a coroutine that drops the last owning reference to this stream // (e.g. the Response gets destroyed once EOF is delivered) -- or to the whole Session, when @@ -108,7 +111,6 @@ void Http3Stream::call_read_handler() auto self = shared_from_this(); auto session_guard = session.shared_from_this(); - call_read_handler_active = true; size_t consumed = 0; while (read_handler) { @@ -157,8 +159,8 @@ void Http3Stream::call_read_handler() if (eof_received) { - // 0-byte read = EOF, matching the beast/nghttp2 convention. - swap_and_invoke(read_handler, boost::system::error_code{}, 0); + // the end of the body, reported the way ASIO reports it everywhere else + swap_and_invoke(read_handler, error_code{asio::error::eof}, 0); continue; } @@ -188,38 +190,69 @@ void Http3Stream::call_read_handler() // Outgoing body // ================================================================================================= -void Http3Stream::start_write(WriteHandler&& handler, asio::const_buffer buffer) +void Http3Stream::start_write(WriteHandler&& handler, asio::const_buffer buffer, bool eof) { auto n = asio::buffer_size(buffer); - const bool is_eof = (n == 0); - logd("[{}] start_write: n={} is_eof={}", log_prefix, n, is_eof); + logd("[{}] start_write: n={} eof={}", log_prefix, n, eof); + + auto complete_immediately = [&](error_code ec) { // + anyhttp::complete_immediately(std::move(handler), get_executor(), ec); + }; + + // + // The protocol-independent entry ladder, in the order the Writer contract in common.hpp + // prescribes: a zero-length non-EOF write is a free no-op wherever the body stands; after the + // body has been ended, data has no body left to belong to -- through either entry point -- + // while a bare re-end is answered by how far the end has actually gotten. + // + if (n == 0 && !eof) + { + complete_immediately(error_code{}); + return; + } // // Once accepted, the caller's intent to end the body is final: this is what tells // delete_writer() the body ended where it was meant to, so it need not reset the stream. An // earlier cancellation just makes for a shorter body than planned -- legitimate here, and a - // declared content-length is still enforced by the peer. + // declared content-length is still enforced by the peer. (Cancelling an EOF write whose FIN + // never reached nghttp3 rolls the intent back, see bind_write_cancellation() -- such a write + // does not land here again.) // - if (is_eof && eof_submitted) + if (eof_submitted) { + if (n > 0) + { + loge("[{}] start_write: body has already been ended", log_prefix); + complete_immediately(errc::make_error_code(errc::broken_pipe)); + return; + } + // - // The body was already ended. If that FIN is still pending (its handler was detached by - // cancellation, see bind_write_cancellation()), adopt this handler so it completes when the - // FIN actually goes out; otherwise the FIN is long gone and there is nothing left to do. + // A bare re-end of an already-ended body. If that FIN is still pending on a live stream + // (its handler was detached by cancellation), adopt this handler so it completes when the + // FIN actually goes out. If the FIN made it to nghttp3, there is nothing left to do. And + // if the stream died with the FIN still owed, it will never go out -- that is a reset, not + // a harmless no-op. // - if (write_active && write_is_eof) + if (write_active && write_is_eof && !closed) { + assert(!write_handler); // only a detached FIN may be adopted, never a live handler logd("[{}] start_write: FIN already pending, adopting handler", log_prefix); bind_write_cancellation(handler, write_token); write_handler = std::move(handler); + return; } - else if (handler) - { - asio::any_completion_executor ex = - asio::get_associated_immediate_executor(handler, get_executor()); - ex.execute([handler = std::move(handler)]() mutable - { std::move(handler)(boost::system::error_code{}); }); - } + + complete_immediately(closed && !fin_offered ? errc::make_error_code(errc::connection_reset) + : error_code{}); + return; + } + + if (closed) + { + logw("[{}] start_write: stream already closed", log_prefix); + complete_immediately(errc::make_error_code(errc::connection_reset)); return; } @@ -230,7 +263,7 @@ void Http3Stream::start_write(WriteHandler&& handler, asio::const_buffer buffer) // assert(!write_active); - if (is_eof) + if (eof) eof_submitted = true; const uint64_t token = next_write_token++; @@ -243,7 +276,7 @@ void Http3Stream::start_write(WriteHandler&& handler, asio::const_buffer buffer) write_source_copied = 0; write_chunk.clear(); write_confirmed = 0; - write_is_eof = is_eof; + write_is_eof = eof; write_token = token; write_handler = std::move(handler); @@ -271,13 +304,18 @@ void Http3Stream::bind_write_cancellation(WriteHandler& handler, uint64_t token) if (write_token != token || !write_handler) return; // already completed naturally before the cancellation was delivered - if (write_is_eof) + if (write_is_eof && asio::buffer_size(write_source) == 0) { // - // The body has already been declared ended, and a FIN cannot be un-sent -- it may just - // still be waiting for flow control credit. Detach the handler but leave the write - // active so it still goes out: abandoning it would leave the stream half-open forever, - // with the peer waiting for an end that never comes. + // A bare end-of-body carries no memory of the caller's, and a FIN cannot be un-sent -- + // it may just still be waiting for flow control credit. Detach the handler but leave + // the write active so it still goes out: abandoning it would leave the stream half-open + // forever, with the peer waiting for an end that never comes. + // + // A *data-carrying* EOF write does not get this treatment: its buffer belongs to the + // caller, who is free to destroy it the moment the handler runs, so it must be cancelled + // exactly like any other data write below -- keeping it active would leave nghttp3 and + // ngtcp2 pointing into freed memory. // logd("[{}] async_write: \x1b[1;31mcancelled\x1b[0m ({}), FIN still pending", log_prefix, ct); @@ -325,6 +363,19 @@ void Http3Stream::bind_write_cancellation(WriteHandler& handler, uint64_t token) write_chunk.clear(); // moved-from } + // + // If this write was to end the body but its FIN never reached nghttp3, the body is not + // ended after all: roll the intent back, so that a later async_write_eof() takes the normal + // path and actually sends the FIN of the (now shorter) body, instead of completing as a + // no-op while the peer waits forever. In ZeroCopy mode the intent is rolled back + // unconditionally -- there, a FIN that did reach nghttp3 went with unacknowledged bytes, + // which is exactly the case that just reset the stream, so nothing of it survives either + // way. Only a Staged write whose final chunk (FIN included) was already carved keeps its + // end standing: that FIN rides out with the retired chunks all by itself. + // + if (write_is_eof && (write_mode == WriteMode::ZeroCopy || !fin_offered)) + eof_submitted = false; + write_active = false; write_source = {}; // make sure to post this -- otherwise "MAIN COROUTINE DID NOT COMPLETE" happens @@ -342,6 +393,26 @@ nghttp3_ssize Http3Stream::data_reader(nghttp3_vec* vec, size_t veccnt, uint32_t if (!write_active) return NGHTTP3_ERR_WOULDBLOCK; + const size_t total = asio::buffer_size(write_source); + + // + // The one place that decides whether the vec about to be returned carries the last bytes of + // the body, so the FIN can ride along with them instead of costing a callback -- and a QUIC + // packet -- of its own. In ZeroCopy mode the single offer below hands out everything at once; + // in Staged mode write_source_copied reaches the end with the final chunk. Note that flagging + // the FIN does not complete the write: in ZeroCopy it still completes on acknowledgement, in + // Staged on confirmation of that final chunk. + // + auto flag_eof_if_last = [&] + { + const size_t handed = write_mode == WriteMode::ZeroCopy ? write_offered : write_source_copied; + if (write_is_eof && handed == total) + { + *pflags |= NGHTTP3_DATA_FLAG_EOF; + fin_offered = true; + } + }; + if (write_mode == WriteMode::ZeroCopy) { // @@ -352,7 +423,6 @@ nghttp3_ssize Http3Stream::data_reader(nghttp3_vec* vec, size_t veccnt, uint32_t // same vec to ngtcp2_conn_writev_stream() and advances nghttp3 by whatever went into the // packet. // - const size_t total = asio::buffer_size(write_source); if (write_offered < total) { auto* base = static_cast(write_source.data()) + write_offered; @@ -360,6 +430,7 @@ nghttp3_ssize Http3Stream::data_reader(nghttp3_vec* vec, size_t veccnt, uint32_t vec[0].len = total - write_offered; write_offered = total; // don't offer these bytes twice -- see class comment above // write_active + flag_eof_if_last(); return 1; } } @@ -370,6 +441,7 @@ nghttp3_ssize Http3Stream::data_reader(nghttp3_vec* vec, size_t veccnt, uint32_t vec[0].base = write_chunk.data() + write_offered; vec[0].len = write_chunk.size() - write_offered; write_offered = write_chunk.size(); // don't re-offer these bytes on a repeat call + flag_eof_if_last(); return 1; } @@ -389,7 +461,7 @@ nghttp3_ssize Http3Stream::data_reader(nghttp3_vec* vec, size_t veccnt, uint32_t if (!write_chunk.empty()) in_flight_writes.emplace_back(std::move(write_chunk)); - const size_t remaining = asio::buffer_size(write_source) - write_source_copied; + const size_t remaining = total - write_source_copied; if (remaining > 0) { const size_t take = std::min(remaining, kWriteChunkSize); @@ -400,6 +472,7 @@ nghttp3_ssize Http3Stream::data_reader(nghttp3_vec* vec, size_t veccnt, uint32_t write_confirmed = 0; vec[0].base = write_chunk.data(); vec[0].len = write_chunk.size(); + flag_eof_if_last(); return 1; } } @@ -413,11 +486,20 @@ nghttp3_ssize Http3Stream::data_reader(nghttp3_vec* vec, size_t veccnt, uint32_t return NGHTTP3_ERR_WOULDBLOCK; // - // The EOF marker (write_source is always empty for it) completes as soon as nghttp3 has taken - // the FIN: unlike body data, a FIN carries no memory of the caller's that we would have to - // keep alive until it is acknowledged. + // A write that ends the body with data still in flight already flagged EOF above, together + // with that data, and completes when the data does -- there is nothing to do here but repeat + // the flag, should nghttp3 ask again. // *pflags |= NGHTTP3_DATA_FLAG_EOF; + fin_offered = true; + if (total > 0) + return 0; + + // + // A bare end-of-body, on the other hand, completes as soon as nghttp3 has taken the FIN: + // unlike body data, it carries no memory of the caller's that we would have to keep alive + // until it is acknowledged. + // finish_active_write(); return 0; } @@ -434,8 +516,10 @@ void Http3Stream::on_write_acked(size_t n) // if (write_mode != WriteMode::ZeroCopy) return; // a staged write is long done by the time its bytes are acknowledged - if (n == 0 || !write_active || write_is_eof) + if (n == 0 || !write_active) return; + if (asio::buffer_size(write_source) == 0) + return; // a bare end-of-body completes in data_reader(), with nothing left to acknowledge write_acked = std::min(write_acked + n, asio::buffer_size(write_source)); logd("[{}] on_write_acked: {} bytes, {}/{} acknowledged", log_prefix, n, write_acked, @@ -456,8 +540,10 @@ void Http3Stream::on_write_offered(size_t n) // if (write_mode != WriteMode::Staged) return; // a zero-copy write completes on acknowledgement, not on handover - if (n == 0 || !write_active || write_is_eof) + if (n == 0 || !write_active) return; + if (asio::buffer_size(write_source) == 0) + return; // a bare end-of-body completes in data_reader(), with no chunk to confirm n = std::min(n, write_chunk.size() - write_confirmed); write_confirmed += n; @@ -518,8 +604,9 @@ void Http3Stream::finish_active_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. // - asio::post(get_executor(), [self = shared_from_this(), handler = std::move(handler)]() mutable - { swap_and_invoke(handler, boost::system::error_code{}); }); + asio::post(get_executor(), [self = shared_from_this(), handler = std::move(handler)]() mutable { + std::move(handler)(boost::system::error_code{}); + }); } // ================================================================================================= @@ -642,13 +729,17 @@ void Http3Stream::fail(boost::system::error_code ec) // A write waiting for its data to be acknowledged will never see those acknowledgements now: // ngtcp2 drops whatever of this stream is still in flight. It also stops touching the caller's // buffer, which is all the wait was ever for, so complete the write -- as failed, because the - // body did not make it -- instead of leaving it pending forever. + // body did not make it -- instead of leaving it pending forever. This must happen even when + // the handler was detached by cancellation (a pending FIN): leaving such a write marked + // active would let start_write() adopt a fresh handler onto a stream nghttp3 will never poll + // again. // - if (write_active && write_handler) + if (write_active) { write_active = false; write_source = {}; - swap_and_invoke(write_handler, ec ? ec : errc::make_error_code(errc::connection_reset)); + if (write_handler) + swap_and_invoke(write_handler, ec ? ec : errc::make_error_code(errc::connection_reset)); } maybe_close(); diff --git a/src/request_handlers.cpp b/src/request_handlers.cpp index 672a2fb..96ff995 100644 --- a/src/request_handlers.cpp +++ b/src/request_handlers.cpp @@ -1,6 +1,7 @@ #include "anyhttp/request_handlers.hpp" #include "anyhttp/client.hpp" #include "anyhttp/formatter.hpp" // IWYU pragma: keep +#include "anyhttp/literals.hpp" #include "anyhttp/server.hpp" #include @@ -74,8 +75,7 @@ awaitable dump(server::Request request, server::Response response) auto body = str.str(); co_await response.async_submit( 200, fields({{"Content-Length", body.size()}, {"Content-Type", "text/plain"}})); - co_await response.async_write(asio::buffer(body)); - co_await response.async_write({}, deferred); + co_await response.async_write_eof(asio::buffer(body)); } awaitable echo(server::Request request, server::Response response) @@ -85,26 +85,31 @@ awaitable echo(server::Request request, server::Response response) co_await response.async_submit(200, {}); - std::array buffer; + std::array 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(); } awaitable not_found(server::Response response) { co_await response.async_submit(404, {}); - co_await response.async_write({}); + co_await response.async_write_eof(); } awaitable not_found(server::Request, server::Response response) { co_await response.async_submit(404, {}); - co_await response.async_write({}); + co_await response.async_write_eof(); } awaitable eat_request(server::Request request, server::Response response) @@ -112,26 +117,15 @@ awaitable eat_request(server::Request request, server::Response response) logd("eat_request: going to eat {} bytes", request.content_length().value_or(-1)); co_await response.async_submit(200, {}); - co_await response.async_write({}); + co_await response.async_write_eof(); - size_t bytes = 0; try { - std::array buffer; - for (;;) - { - size_t n = co_await request.async_read_some(asio::buffer(buffer)); - if (n == 0) - break; - - logd("eat_request: ate {} bytes", n); - bytes += n; - } - logd("eat_request: ate {} bytes", bytes); + logd("eat_request: ate {} bytes", co_await drain(request)); } catch (const boost::system::system_error& e) { - logi("eat_request: ate {} bytes, then caught exception: {}", bytes, e.code().message()); + logi("eat_request: caught exception: {}", e.code().message()); throw; } @@ -155,7 +149,7 @@ awaitable discard(server::Request request, server::Response response) { co // ================================================================================================= -awaitable send(client::Request& request, size_t bytes) +awaitable generate(client::Request& request, size_t bytes) { return sendAndForceEOF(request, rv::iota(uint8_t{0}) | rv::take(bytes)); } @@ -163,97 +157,60 @@ awaitable send(client::Request& request, size_t bytes) awaitable read(client::Response& response) { std::string body; - std::array buffer; + std::array buffer; for (;;) { - size_t n = co_await response.async_read_some(asio::buffer(buffer)); - if (n == 0) - break; - + auto [ec, n] = co_await response.async_read_some(asio::buffer(buffer), as_tuple); body += std::string_view(buffer.data(), n); - logd("read: {}, total {}", n, body.size()); - } - - logi("read: EOF after reading {} bytes", body.size()); - co_return body; -} - -awaitable count(client::Response& response) -{ - size_t bytes = 0; - std::array buffer; - for (;;) - { - size_t n = co_await response.async_read_some(asio::buffer(buffer)); - if (n == 0) - break; + if (ec == asio::error::eof) + { + logd("read: EOF after reading {} bytes", body.size()); + co_return std::move(body); + } + else if (ec) + { + loge("receive: \x1b[1;31m{}\x1b[0m after reading {} bytes", ec.message(), body.size()); + throw boost::system::system_error(ec); + } - bytes += n; - logd("count: {}, total {}", n, bytes); + logd("read: {}, total {}", n, body.size()); } - - logi("count: EOF after reading {} bytes", bytes); - co_return bytes; } awaitable> try_receive(client::Response& response) { size_t bytes = 0; - std::array buffer; + std::array buffer; for (;;) { auto [ec, n] = co_await response.async_read_some(asio::buffer(buffer), as_tuple); - // co_await yield(); bytes += n; - if (ec || n == 0) + + // the regular end of the body is not something to report as an error + if (ec == asio::error::eof) + { + logd("receive: EOF after reading {} bytes", bytes); + co_return std::make_tuple(bytes, error_code{}); + } + else if (ec) + { + loge("receive: \x1b[1;31m{}\x1b[0m after reading {} bytes", ec.message(), bytes); co_return std::make_tuple(bytes, ec); + } } } awaitable try_receive(client::Response& response, error_code& ec) { -#if 0 size_t bytes; std::tie(bytes, ec) = co_await try_receive(response); -#else - ec = {}; - size_t bytes = 0, count = 0; - std::array buffer; - try - { - for (;;) - { - size_t n = co_await response.async_read_some(asio::buffer(buffer)); - if (n == 0) - break; - - // do NOT 'respawn' read handler in first round, see NGHttp2Stream::call_handler_loop() - // if (count++ == 0) - // co_await yield(); - // co_await yield(); - - bytes += n; - logd("receive: {}, total {}", n, bytes); - } - } - catch (const boost::system::system_error& ex) - { - ec = ex.code(); - loge("receive: \x1b[1;31n{}\x1b[0m after reading {} bytes", ex.code().message(), bytes); - co_return bytes; - } - - // co_await sleep(100ms); - - logi("receive: EOF after reading {} bytes", bytes); co_return bytes; -#endif } -awaitable read_response(client::Request& request) +awaitable count_response(client::Request& request) { auto response = co_await request.async_get_response(); - co_return co_await count(response); + co_return co_await drain(response); } awaitable> try_read_response(client::Request& request) @@ -261,7 +218,7 @@ awaitable> try_read_response(client::Request& request) try { auto response = co_await request.async_get_response(); - co_return co_await count(response); + co_return co_await drain(response); } catch (const boost::system::system_error& ex) { @@ -269,26 +226,18 @@ awaitable> try_read_response(client::Request& request) } } -awaitable send_eof(client::Request& request) -{ - co_await request.async_write({}); - // logi("send: finishing request..."); - // auto [ec] = co_await request.async_write({}, as_tuple(deferred)); - // logi("send: finishing request... done ({})", ec.message()); -} +awaitable send_eof(client::Request& request) { co_await request.async_write_eof(); } awaitable h2spec(server::Request request, server::Response response) { co_await yield(10); // FIXME: without this, one more testcase fails std::array buffer; - size_t n = co_await request.async_read_some(asio::buffer(buffer)); + co_await request.async_read_some(asio::buffer(buffer), as_tuple); constexpr auto hello = "Hello, World!\n"sv; co_await response.async_submit(200, fields({{"Content-Length", hello.size()}})); - co_await response.async_write(asio::buffer(hello)); - co_await response.async_write({}); - while (co_await request.async_read_some(asio::buffer(buffer)) > 0) - ; + co_await response.async_write_eof(asio::buffer(hello)); + co_await drain(request); } // ================================================================================================= diff --git a/src/server.cpp b/src/server.cpp index 40e7022..b1372f7 100644 --- a/src/server.cpp +++ b/src/server.cpp @@ -90,10 +90,10 @@ void Response::async_submit_any(StatusHandler&& handler, unsigned int status_cod impl->async_submit(std::move(handler), status_code, std::move(headers)); } -void Response::async_write_any(WriteHandler&& handler, asio::const_buffer buffer) +void Response::async_write_any(WriteHandler&& handler, asio::const_buffer buffer, bool eof) { assert(impl); - impl->async_write(std::move(handler), buffer); + impl->async_write(std::move(handler), buffer, eof); } // ================================================================================================= diff --git a/src/server_impl.cpp b/src/server_impl.cpp index 06ddc0c..be114e5 100644 --- a/src/server_impl.cpp +++ b/src/server_impl.cpp @@ -221,12 +221,12 @@ class TestStream : public AnyAsyncStream::Impl executor_type get_executor() noexcept override { return socket_.get_executor(); } ip::tcp::socket& get_socket() final { return socket_; } - void async_write_impl(ReadWriteHandler handler, ConstBuffers buffers) final + void async_write_some(ReadWriteHandler handler, ConstBuffers buffers) final { socket_.async_write_some(buffers, std::move(handler)); } - void async_read_impl(ReadWriteHandler handler, MutableBuffers buffers) final + void async_read_some(ReadWriteHandler handler, MutableBuffers buffers) final { socket_.async_read_some(buffers, std::move(handler)); } diff --git a/src/server_main.cpp b/src/server_main.cpp index fc01311..f5181af 100644 --- a/src/server_main.cpp +++ b/src/server_main.cpp @@ -13,6 +13,7 @@ #include +#include #include #include #include @@ -27,7 +28,7 @@ namespace po = boost::program_options; struct Config { - bool verbose = false; + size_t verbose = 0; size_t threads = 1; server::Config server{.port = 8080}; }; @@ -40,8 +41,8 @@ std::expected parseConfig(int argc, char* argv[]) po::options_description desc("Allowed options"); auto opts = desc.add_options(); opts("help,h", "produce help message"); - opts("verbose,v", po::bool_switch(&config.verbose)->default_value(false), - "enable verbose logging"); + opts("verbose,v", po::value>()->zero_tokens()->composing(), + "enable verbose logging (repeat for trace level)"); opts("threads,t", po::value(&config.threads)->default_value(1), "number of threads to run"); opts("port,p", po::value(&config.server.port)->default_value(config.server.port), "listening port"); @@ -53,8 +54,13 @@ std::expected parseConfig(int argc, char* argv[]) po::variables_map vm; try { - po::store(po::parse_command_line(argc, argv, desc), vm); + auto parsed = po::parse_command_line(argc, argv, desc); + po::store(parsed, vm); po::notify(vm); + + // 'verbose' takes no argument, so its parsed value is always empty -- count occurrences + config.verbose = std::ranges::count_if(parsed.options, [](const po::option& option) + { return option.string_key == "verbose"; }); } catch (const po::error& error) { @@ -94,7 +100,9 @@ int main(int argc, char* argv[]) if (!config) return config.error(); - if (config->verbose) + if (config->verbose >= 2) + spdlog::set_level(spdlog::level::trace); + else if (config->verbose) spdlog::set_level(spdlog::level::debug); else spdlog::set_level(spdlog::level::info); diff --git a/src/utils.cpp b/src/utils.cpp index 267c9de..7210c21 100644 --- a/src/utils.cpp +++ b/src/utils.cpp @@ -2,17 +2,14 @@ #include -#include - // ================================================================================================= +#if defined(GITHUB_ACTIONS) || defined(NDEBUG) +size_t run(boost::asio::io_context& context) { return context.run(); } +#else +#include size_t run(boost::asio::io_context& context) { -#if defined(GITHUB_ACTIONS) - return context.run(); -#elif defined(NDEBUG) - return context.run(); -#else size_t i = 0; using namespace std::chrono; auto t0 = steady_clock::now(); @@ -29,8 +26,8 @@ size_t run(boost::asio::io_context& context) // clang-format off } return i; -#endif } +#endif // ================================================================================================= diff --git a/test/test_server.cpp b/test/test_server.cpp index 857e920..89657d2 100644 --- a/test/test_server.cpp +++ b/test/test_server.cpp @@ -1,6 +1,7 @@ #include "anyhttp/client.hpp" #include "anyhttp/file_handler.hpp" #include "anyhttp/formatter.hpp" // IWYU pragma: keep +#include "anyhttp/literals.hpp" #include "anyhttp/request_handlers.hpp" #include "anyhttp/server.hpp" #include "anyhttp/session.hpp" @@ -27,9 +28,6 @@ #include -#include -#include - #include #include #include @@ -205,21 +203,8 @@ class Server : public testing::TestWithParam { logd("{} ({})", request.url().path(), request.url().buffer()); - auto url = request.url(); - auto params = url.params(); - if (auto it = params.find("delay"); it != params.end()) - { - try - { - using ms = std::chrono::milliseconds; - auto delay_ms = boost::lexical_cast((*it).value); - co_await sleep(ms{delay_ms}); - } - catch (boost::bad_lexical_cast&) - { - loge("invalid number: {}", (*it).value); - } - } + if (auto delay = request.get_param_as("delay")) + co_await sleep(std::chrono::milliseconds{*delay}); if (request.url().path() == "/echo") co_await echo(std::move(request), std::move(response)); @@ -767,8 +752,8 @@ TEST_P(ClientAsync, WHEN_post_data_THEN_receive_echo) test = [this](Session session) -> awaitable { auto request = co_await session.async_submit(url.set_path("echo"), {}); - size_t bytes = 1024; // * 1024 * 1024; - auto count = co_await (send(request, bytes) && read_response(request)); + size_t bytes = 1024; + auto count = co_await (generate(request, bytes) && count_response(request)); EXPECT_EQ(bytes, count); }; } @@ -778,7 +763,7 @@ TEST_P(ClientAsync, WHEN_post_without_path_THEN_error_404) test = [this](Session session) -> awaitable { auto request = co_await session.async_submit(url.set_path(""), {}); - co_await send(request, 1024); + co_await generate(request, 1024); auto [ec, response] = co_await request.async_get_response(as_tuple); EXPECT_TRUE(ec); }; @@ -789,10 +774,10 @@ TEST_P(ClientAsync, WHEN_post_to_unknown_path_THEN_error_404) test = [this](Session session) -> awaitable { auto request = co_await session.async_submit(url.set_path("unknown"), {}); - co_await send(request, 1024 * 1024); + co_await generate(request, 1_m); auto response = co_await request.async_get_response(); EXPECT_EQ(response.status_code(), 404); - auto received = co_await count(response); + auto received = co_await drain(response); }; } @@ -801,7 +786,7 @@ TEST_P(ClientAsync, WHEN_server_discards_request_THEN_error_500) test = [this](Session session) -> awaitable { auto request = co_await session.async_submit(url.set_path("discard"), {}); - co_await send(request, 1024); + co_await generate(request, 1024); auto [ec, response] = co_await request.async_get_response(as_tuple); EXPECT_TRUE(ec); }; @@ -812,7 +797,7 @@ TEST_P(ClientAsync, WHEN_server_discards_request_delayed_THEN_error_500) test = [this](Session session) -> awaitable { auto request = co_await session.async_submit(url.set_path("detach"), {}); - co_await send(request, 1024); + co_await generate(request, 1024); auto [ec, response] = co_await request.async_get_response(as_tuple); EXPECT_TRUE(ec); }; @@ -836,7 +821,7 @@ TEST_P(ClientAsync, WHEN_invalid_port_in_host_header_THEN_reports_error) Fields fields; fields.set("Host", "host:12345x"); auto request = co_await session.async_submit(url.set_path("echo"), fields); - auto response = co_await (send_eof(request) && read_response(request)); + auto response = co_await (send_eof(request) && count_response(request)); }; } @@ -925,7 +910,7 @@ TEST_P(ClientAsync, WHEN_client_cancels_write_THEN_can_resume) // control timing, so don't assert either way here. What matters is that ending the // upload and draining the response together complete the exchange. // - auto received = co_await (send_eof(request) && count(response)); + auto received = co_await (send_eof(request) && drain(response)); EXPECT_GT(received, 0); } else @@ -935,7 +920,7 @@ TEST_P(ClientAsync, WHEN_client_cancels_write_THEN_can_resume) EXPECT_EQ(code(ep), boost::system::errc::operation_canceled); // as we have no control over when the send window is re-opened, wait for it in parallel - auto received = co_await (send_eof(request) && count(response)); + auto received = co_await (send_eof(request) && drain(response)); EXPECT_GT(received, 0); } }; @@ -963,10 +948,10 @@ TEST_P(ClientAsync, YieldFuzz) co_await yield(dist(gen)); co_await response.async_write(asio::buffer(msg)); co_await yield(dist(gen)); - co_await response.async_write({}); + co_await response.async_write_eof(); co_await yield(dist(gen)); std::array data; - co_await request.async_read_some(asio::buffer(data)); + co_await request.async_read_some(asio::buffer(data), as_tuple); }; test = [this](Session session) -> awaitable { @@ -982,13 +967,124 @@ TEST_P(ClientAsync, YieldFuzz) fields.set("Content-Length", "0"); auto request = co_await session.async_submit(url, fields); co_await yield(dist(gen)); - co_await request.async_write({}); + co_await request.async_write_eof(); co_await yield(dist(gen)); - co_await read_response(request); + co_await count_response(request); } }; } +// +// The end of an incoming body is an error code, not a zero-sized read -- and it keeps being +// reported for every read issued after it. A zero-length buffer, on the other hand, says nothing +// about the body at all: it completes immediately, at the end of a body just as anywhere else. +// +TEST_P(ClientAsync, WHEN_body_ends_THEN_read_reports_eof) +{ + static const auto hello = "Hello, World!"sv; + custom = [this](server::Request request, server::Response response) -> awaitable + { + co_await drain(request); + co_await response.async_submit(200, fields({{"Content-Length", hello.size()}})); + co_await response.async_write_eof(asio::buffer(hello)); + }; + test = [this](Session session) -> awaitable + { + auto request = co_await session.async_submit(url); + co_await request.async_write_eof(); + auto response = co_await request.async_get_response(); + + std::string body; + std::array buffer; // small on purpose: several reads before the end + for (;;) + { + auto [ec, n] = co_await response.async_read_some(asio::buffer(buffer), as_tuple); + if (ec) + { + EXPECT_EQ(ec, asio::error::eof); + EXPECT_EQ(n, 0u); + break; + } + body.append(buffer.data(), n); + } + EXPECT_EQ(body, hello); + + // + // Reading past the end of a body says the same thing again -- also once the protocol layer + // has torn the underlying stream down in the meantime, which the yield gives it every + // opportunity to do (both sides of the exchange are finished by now). + // + co_await yield(20); + auto [ec, n] = co_await response.async_read_some(asio::buffer(buffer), as_tuple); + EXPECT_EQ(ec, asio::error::eof); + + // ... but a zero-length read is not a read, and reports nothing + std::array empty; + std::tie(ec, n) = co_await response.async_read_some(asio::buffer(empty), as_tuple); + EXPECT_FALSE(ec); + EXPECT_EQ(n, 0u); + }; +} + +// +// An empty async_write() no longer ends a body -- async_write_eof() does, and nothing else. So a +// message with an empty write in the middle of it still carries everything written after that. +// +TEST_P(ClientAsync, WHEN_empty_buffer_is_written_THEN_body_stays_open) +{ + static const auto tail = "still here"sv; + custom = [this](server::Request request, server::Response response) -> awaitable + { + EXPECT_EQ(co_await drain(request), 0u); + co_await response.async_submit(200, {}); + co_await response.async_write({}); // writes nothing, leaves the body open + co_await response.async_write_eof(asio::buffer(tail)); + }; + test = [this](Session session) -> awaitable + { + auto request = co_await session.async_submit(url); + co_await request.async_write({}); // likewise: the request body stays open + co_await request.async_write_eof(); + auto response = co_await request.async_get_response(); + EXPECT_EQ(co_await read(response), tail); + }; +} + +// +// Ending a body twice is harmless -- the second call has nothing left to do -- and an empty +// write stays a free no-op even then. Data after the end is neither: there is no body left for +// it to belong to, through whichever entry point it tries to sneak in. +// +TEST_P(ClientAsync, WHEN_written_after_eof_THEN_reports_broken_pipe) +{ + static const auto hello = "Hello, World!"sv; + custom = [this](server::Request request, server::Response response) -> awaitable + { + co_await drain(request); + co_await response.async_submit(200, fields({{"Content-Length", hello.size()}})); + co_await response.async_write_eof(asio::buffer(hello)); + + auto [ec] = co_await response.async_write_eof(as_tuple); + EXPECT_FALSE(ec); + + std::tie(ec) = co_await response.async_write({}, as_tuple); + EXPECT_FALSE(ec); + + std::tie(ec) = co_await response.async_write(asio::buffer(hello), as_tuple); + EXPECT_EQ(ec, boost::system::errc::broken_pipe); + + std::tie(ec) = co_await response.async_write_eof(asio::buffer(hello), as_tuple); + EXPECT_EQ(ec, boost::system::errc::broken_pipe); + }; + test = [this](Session session) -> awaitable + { + auto request = co_await session.async_submit(url); + co_await request.async_write_eof(); + auto response = co_await request.async_get_response(); + EXPECT_EQ(co_await read(response), hello); + }; +} + TEST_P(ClientAsync, HelloWorld) { static const auto hello = "Hello, World!"sv; @@ -1000,7 +1096,7 @@ TEST_P(ClientAsync, HelloWorld) test = [this](Session session) -> awaitable { auto request = co_await session.async_submit(url); - co_await request.async_write({}); + co_await request.async_write_eof(); auto response = co_await request.async_get_response(); auto body = co_await read(response); EXPECT_EQ(body, hello); @@ -1019,26 +1115,96 @@ TEST_P(ClientAsync, WHEN_server_writes_large_buffer_at_once_THEN_receives_all) { static const std::vector body = [] { - std::vector data(256 * 1024); + std::vector data(256_k); std::ranges::generate(data, [i = uint8_t(0)]() mutable { return i++; }); return data; }(); custom = [this](server::Request request, server::Response response) -> awaitable { - std::array buffer; - while (co_await request.async_read_some(asio::buffer(buffer)) > 0) - ; // drain the request -- HTTP/1.1 closes the connection on an unfinished parser + // drain the request -- HTTP/1.1 closes the connection on an unfinished parser + co_await drain(request); co_await response.async_submit(200, fields({{"Content-Length", body.size()}})); - co_await response.async_write(asio::buffer(body)); - co_await response.async_write({}); + co_await response.async_write_eof(asio::buffer(body)); }; test = [this](Session session) -> awaitable { auto request = co_await session.async_submit(url); - co_await request.async_write({}); - EXPECT_EQ(co_await read_response(request), body.size()); + co_await request.async_write_eof(); + EXPECT_EQ(co_await count_response(request), body.size()); + }; +} + +// +// Cancelling an async_write_eof() that carries data. The buffer goes back to the caller the +// moment the handler runs, so the backend must stop referencing it right there -- for HTTP/3's +// zero-copy path that means resetting the stream, exactly as for a cancelled plain write; under +// ASAN this test is what catches a backend that keeps pointing into the freed buffer. +// +TEST_P(ClientAsync, WHEN_server_cancels_write_eof_THEN_client_sees_truncated_body) +{ + static const std::vector body(8_m, 'x'); + + custom = [this](server::Request request, server::Response response) -> awaitable + { + co_await drain(request); + co_await response.async_submit(200, {}); + + // + // Far more than the peer's receive window, and the client below doesn't read a byte until + // this is over, so the write is guaranteed to still be in progress when it is cancelled. + // + auto [ec] = co_await response.async_write_eof(asio::buffer(body), + cancel_after(50ms, as_tuple)); + EXPECT_EQ(ec, boost::system::errc::operation_canceled); + }; + test = [this](Session session) -> awaitable + { + auto request = co_await session.async_submit(url); + co_await request.async_write_eof(); + auto response = co_await request.async_get_response(); + + // leave the body untouched until the cancellation above has hit, see the sibling testcase + asio::steady_timer timer(co_await this_coro::executor, 150ms); + co_await timer.async_wait(deferred); + + boost::system::error_code ec; + auto received = co_await try_receive(response, ec); + EXPECT_LT(received, body.size()); + EXPECT_EQ(ec, boost::beast::http::error::partial_message); + }; +} + +// +// Cancelling an async_write_eof() whose FIN never made it out must not leave the body in limbo: +// the intent to end it is rolled back, and a re-issued async_write_eof() ends the (now shorter) +// body for real -- instead of completing as a no-op while the peer waits forever for the end. +// +TEST_P(ClientAsync, WHEN_client_cancels_write_eof_THEN_can_still_end) +{ + if (GetParam() == anyhttp::Protocol::http11) + GTEST_SKIP(); // a chunked body cannot be cancelled correctly --> disconnects + + static const std::vector body(8_m, 'x'); + + test = [this](Session session) -> awaitable + { + co_await this_coro::throw_if_cancelled(false); + auto executor = co_await this_coro::executor; + auto request = co_await session.async_submit(url.set_path("echo")); + auto response = co_await request.async_get_response(); + + // far more than the send window, with nobody reading the echo yet: this cannot complete + auto write_eof = [&]() -> awaitable + { co_await request.async_write_eof(asio::buffer(body)); }; + auto [ep] = co_await co_spawn(executor, write_eof(), cancel_after(100ms, as_tuple)); + EXPECT_EQ(code(ep), boost::system::errc::operation_canceled); + + // the FIN never went out with the cancelled write, so the body can still be ended + auto received = co_await (send_eof(request) && drain(response)); + EXPECT_GT(received, 0u); + EXPECT_LT(received, body.size()); }; } @@ -1049,13 +1215,12 @@ TEST_P(ClientAsync, WHEN_server_writes_large_buffer_at_once_THEN_receives_all) // TEST_P(ClientAsync, WHEN_server_cancels_write_THEN_client_sees_truncated_body) { - static const std::vector body(8 * 1024 * 1024, 'x'); + static const std::vector body(8_m, 'x'); custom = [this](server::Request request, server::Response response) -> awaitable { - std::array buffer; - while (co_await request.async_read_some(asio::buffer(buffer)) > 0) - ; // drain the request -- HTTP/1.1 closes the connection on an unfinished parser + // drain the request -- HTTP/1.1 closes the connection on an unfinished parser + co_await drain(request); co_await response.async_submit(200, {}); @@ -1071,7 +1236,7 @@ TEST_P(ClientAsync, WHEN_server_cancels_write_THEN_client_sees_truncated_body) test = [this](Session session) -> awaitable { auto request = co_await session.async_submit(url); - co_await request.async_write({}); + co_await request.async_write_eof(); auto response = co_await request.async_get_response(); // @@ -1109,7 +1274,7 @@ class FileHandler : public ClientAsync write(root / "hello.txt", "Hello, File!"); write(root / "empty.txt", ""); write(root / "sub" / "nested.txt", "Nested!"); - write(root / "large.bin", std::string(256 * 1024, 'x')); + write(root / "large.bin", std::string(256_k, 'x')); write(root / "secret.txt", "no peeking"); write(root / "er.txt", "leaked"); // what "/customer.txt" resolves to without a segment check std::filesystem::permissions(root / "secret.txt", std::filesystem::perms::none); @@ -1144,7 +1309,7 @@ class FileHandler : public ClientAsync awaitable> get(Session& session, boost::urls::url target) { auto request = co_await session.async_submit(target, {}); - co_await request.async_write({}); + co_await request.async_write_eof(); auto response = co_await request.async_get_response(); auto body = co_await read(response); co_return std::make_tuple(response.status_code(), std::move(body)); @@ -1219,7 +1384,7 @@ TEST_P(FileHandler, WHEN_file_is_large_THEN_serves_all_of_it) { auto [status, body] = co_await get(session, "/custom/large.bin"); EXPECT_EQ(status, 200); - EXPECT_EQ(body, std::string(256 * 1024, 'x')); + EXPECT_EQ(body, std::string(256_k, 'x')); }; } @@ -1312,13 +1477,13 @@ TEST_P(ClientAsync, ServerYieldFirst) co_await yield(10); co_await response.async_submit(200, {}); co_await yield(10); - co_await response.async_write({}); + co_await response.async_write_eof(); }; test = [this](Session session) -> awaitable { auto request = co_await session.async_submit(url); - co_await request.async_write({}); - co_await read_response(request); + co_await request.async_write_eof(); + co_await count_response(request); }; } @@ -1389,18 +1554,21 @@ TEST_P(ClientAsync, Custom) std::array 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) + { + co_await response.async_write_eof(); co_return; + } + co_await response.async_write(asio::buffer(buffer, n)); } }; test = [this](Session session) -> awaitable { auto request = co_await session.async_submit(url, {}); - size_t bytes = 1024; - auto res = co_await (send(request, bytes) && read_response(request)); - EXPECT_EQ(bytes, res); + constexpr size_t bytes = 1024; + auto count = co_await (generate(request, bytes) && count_response(request)); + EXPECT_EQ(bytes, count); }; } @@ -1409,14 +1577,15 @@ TEST_P(ClientAsync, IgnoreRequest) custom = [this](server::Request request, server::Response response) -> awaitable { co_await response.async_submit(200, {}); - co_await response.async_write({}); + co_await response.async_write_eof(); }; test = [this](Session session) -> awaitable { Fields fields; fields.set("content-length", "0"); auto request = co_await session.async_submit(url, fields); - auto res = co_await (send(request, 0) && read_response(request)); + auto count = co_await (generate(request, 0) && count_response(request)); + EXPECT_EQ(count, 0); }; } @@ -1431,7 +1600,7 @@ TEST_P(ClientAsync, IgnoreRequestAndResponse) test = [this](Session session) -> awaitable { auto request = co_await session.async_submit(url, {}); - auto res = co_await (send(request, 0) && try_read_response(request)); + auto res = co_await (generate(request, 0) && try_read_response(request)); EXPECT_FALSE(res.has_value()); std::println("ERROR: {}", res.error().message()); }; @@ -1446,13 +1615,13 @@ TEST_P(ClientAsync, PostRange) auto request = co_await session.async_submit(url.set_path("echo"), {}); // co_await request.async_write(asio::buffer("ping"sv)); // FIXME: auto response = co_await request.async_get_response(); - // std::string s(10ul * 1024 * 1024, 'a'); + // std::string s(10_m, 'a'); // auto sender = send(request, std::string_view("blah")); - // auto sender = send(request, std::string(10ul * 1024 * 1024, 'a')); - auto sender = sendAndForceEOF(request, rv::iota(uint8_t(0)) | rv::take(1 * 1024 * 1024)); - auto received = co_await (std::move(sender) && count(response)); + // auto sender = send(request, std::string(10_m, 'a')); + auto sender = sendAndForceEOF(request, rv::iota(uint8_t(0)) | rv::take(1_m)); + auto received = co_await (std::move(sender) && drain(response)); loge("received: {}", received); - EXPECT_EQ(received, 1 * 1024 * 1024); + EXPECT_EQ(received, 1_m); }; } @@ -1461,10 +1630,10 @@ TEST_P(ClientAsync, PostRangeImmediate) test = [this](Session session) -> awaitable { auto request = co_await session.async_submit(url.set_path("echo"), {}); - auto sender = sendAndForceEOF(request, rv::iota(uint8_t(0)) | rv::take(1 * 1024 * 1024)); - auto received = co_await (std::move(sender) && read_response(request)); + auto sender = sendAndForceEOF(request, rv::iota(uint8_t(0)) | rv::take(1_m)); + auto received = co_await (std::move(sender) && count_response(request)); loge("received: {}", received); - EXPECT_EQ(received, 1 * 1024 * 1024); + EXPECT_EQ(received, 1_m); }; } @@ -1477,8 +1646,8 @@ TEST_P(ClientAsync, WHEN_request_is_sent_THEN_response_is_received_before_body_i auto request = co_await session.async_submit(url.set_path("echo"), {}); auto response = co_await request.async_get_response(); constexpr size_t bytes = 1024; - co_await send(request, bytes); - EXPECT_EQ(co_await count(response), bytes); + co_await generate(request, bytes); + EXPECT_EQ(co_await drain(response), bytes); }; } @@ -1496,18 +1665,16 @@ TEST_P(ClientAsync, WHEN_multiple_request_are_made_THEN_responses_are_received_i test = [this](Session session) -> awaitable { auto request1 = co_await session.async_submit(url.set_path("echo"), {}); - co_await request1.async_write(asio::buffer("Hello, Server #1!"sv)); - co_await request1.async_write({}); + co_await request1.async_write_eof(asio::buffer("Hello, Server #1!"sv)); auto request2 = co_await session.async_submit(url.set_path("echo"), {}); - co_await request2.async_write(asio::buffer("Hello, Server #2! XYZ"sv)); - co_await request2.async_write({}); + co_await request2.async_write_eof(asio::buffer("Hello, Server #2! XYZ"sv)); auto response1 = co_await request1.async_get_response(); - EXPECT_EQ(co_await count(response1), 17); + EXPECT_EQ(co_await drain(response1), 17); auto response2 = co_await request2.async_get_response(); - EXPECT_EQ(co_await count(response2), 21); + EXPECT_EQ(co_await drain(response2), 21); }; } @@ -1518,9 +1685,9 @@ TEST_P(ClientAsync, EatRequest) test = [this](Session session) -> awaitable { auto request = co_await session.async_submit(url.set_path("eat_request"), {}); - co_await send(request, 1024); + co_await generate(request, 1024); auto response = co_await request.async_get_response(); - auto received = co_await count(response); + auto received = co_await drain(response); EXPECT_EQ(received, 0); }; } @@ -1589,7 +1756,7 @@ TEST_P(ClientAsync, CancellationContentLength) { test = [this](Session session) -> awaitable { - const size_t length = 50ul * 1024 * 1024; + const size_t length = 50_m; const std::vector buffer(length); for (size_t i = 0; i <= 20; ++i) { @@ -1636,7 +1803,7 @@ TEST_P(ClientAsync, Cancellation) { test = [this](Session session) -> awaitable { - const size_t length = 50ul * 1024 * 1024; + const size_t length = 50_m; const std::vector buffer(length, 'a'); for (size_t i = 0; i <= 20; ++i) { @@ -1731,9 +1898,9 @@ TEST_P(ClientAsync, CancelAfter) std::tie(ec, response) = co_await request.async_get_response(as_tuple); EXPECT_FALSE(ec); - co_await request.async_write(asio::buffer("Hello, Client!"sv)); - co_await request.async_write({}); - auto received = co_await count(response); + constexpr auto msg = "Hello, Client!"sv; + co_await request.async_write_eof(asio::buffer(msg)); + EXPECT_EQ(co_await read(response), msg); }; } @@ -1745,7 +1912,7 @@ TEST_P(ClientAsync, WHEN_send_more_than_content_length_THEN_connection_is_reset) fields.set("content-length", "1024"); auto request = co_await session.async_submit(url.set_path("eat_request"), fields); auto response = co_await request.async_get_response(); - co_await count(response); + co_await drain(response); auto ex = co_await this_coro::executor; auto [ep] = co_await co_spawn(ex, send(request, rv::iota(uint8_t(0))), as_tuple);