diff --git a/.gitignore b/.gitignore index a99041f..6c89aab 100644 --- a/.gitignore +++ b/.gitignore @@ -5,6 +5,7 @@ docs/_build gtest-parallel-logs report.xml build +build-* secrets Testing *.pcap diff --git a/include/anyhttp/file_handler.hpp b/include/anyhttp/file_handler.hpp new file mode 100644 index 0000000..ad87006 --- /dev/null +++ b/include/anyhttp/file_handler.hpp @@ -0,0 +1,23 @@ +#pragma once + +#include "anyhttp/server.hpp" + +#include +#include + +namespace anyhttp +{ + +// ================================================================================================= + +/** + * Serve a single file from below \p root, mapped into memory with mmap(). The part of the request + * path below \p prefix is taken as the path relative to \p root; anything that would escape + * \p root ("..", a symlink pointing outside) is rejected with 404. + */ +awaitable serve_file(server::Request request, server::Response response, + std::filesystem::path root, std::string prefix); + +// ================================================================================================= + +} // namespace anyhttp diff --git a/include/anyhttp/server.hpp b/include/anyhttp/server.hpp index 5075b1f..5a9bb6e 100644 --- a/include/anyhttp/server.hpp +++ b/include/anyhttp/server.hpp @@ -21,6 +21,24 @@ struct Config std::string listen_address = "::"; uint16_t port = 8080; bool use_strand = false; + + // + // HTTP/3 only: how long a QUIC connection may go without a packet from its peer before it is + // dropped. This is the only way a peer that vanished without a CONNECTION_CLOSE -- a killed + // client, a machine that went to sleep -- is ever noticed, so it also bounds how long its + // session and streams stay around. 30s is what the ngtcp2 examples use. + // + std::chrono::nanoseconds idle_timeout = 30s; + + // + // HTTP/3 only, testing aid: probability (0.0 ... 1.0) with which an individual QUIC datagram + // is thrown away instead of being processed (rx) or actually sent (tx). This exercises loss + // recovery -- retransmits, PTO, ACK handling -- without needing a lossy network. Dropping + // happens per QUIC packet, i.e. GRO-coalesced datagrams are dropped individually and TX + // GSO batching is bypassed while `drop_rate_tx` is non-zero. + // + double drop_rate_rx = 0.0; + double drop_rate_tx = 0.0; }; // ================================================================================================= @@ -135,8 +153,7 @@ class Response // ================================================================================================= -using RequestHandler = std::function; -using RequestHandlerCoro = std::function(Request, Response)>; +using RequestHandler = std::function(Request, Response)>; class Server { @@ -151,7 +168,6 @@ class Server executor_type get_executor() const noexcept; void setRequestHandler(RequestHandler&& handler); - void setRequestHandlerCoro(RequestHandlerCoro&& handler); asio::ip::tcp::endpoint local_endpoint() const; diff --git a/include/anyhttp/server_impl.hpp b/include/anyhttp/server_impl.hpp index 7744dd9..2063fb4 100644 --- a/include/anyhttp/server_impl.hpp +++ b/include/anyhttp/server_impl.hpp @@ -79,12 +79,7 @@ class Server::Impl : public std::enable_shared_from_this } void setRequestHandler(RequestHandler&& handler) { m_requestHandler = std::move(handler); } - void setRequestHandler(RequestHandlerCoro&& handler) - { - m_requestHandlerCoro = std::move(handler); - } const RequestHandler& requestHandler() const { return m_requestHandler; } - const RequestHandlerCoro& requestHandlerCoro() const { return m_requestHandlerCoro; } asio::awaitable udp_receive_loop(); int udp_on_read(Endpoint& ep); @@ -110,7 +105,6 @@ class Server::Impl : public std::enable_shared_from_this std::unordered_map> m_quic_handlers; RequestHandler m_requestHandler; - RequestHandlerCoro m_requestHandlerCoro; bool m_stopped = false; }; diff --git a/src/beast_session.cpp b/src/beast_session.cpp index 003f42c..0a0841e 100644 --- a/src/beast_session.cpp +++ b/src/beast_session.cpp @@ -395,6 +395,7 @@ class ResponseWriter WriterBase>; public: + using super::logPrefix; using super::message; using super::serializer; using super::stream; @@ -426,6 +427,10 @@ class ResponseWriter if (message.find(http::field::server) == message.end()) message.set(http::field::server, "anyhttp"); + mlogd("{} {}", message.result_int(), message.reason()); + for (const auto& header : message) + mlogd(" \x1b[1;34m{}\x1b[0m: {}", header.name_string(), header.value()); + // // TODO: For bundling writing the header and body, we should just post the writing here, // giving an async_write the chance to add a body to the message first. @@ -716,7 +721,7 @@ awaitable ServerSession::do_session(Buffer&& buffer) // server::Request request_wrapper(std::move(reader)); server::Response response_wrapper(std::move(writer)); - if (auto& handler = server().requestHandlerCoro()) + if (auto& handler = server().requestHandler()) { try { diff --git a/src/client_impl_udp.cpp b/src/client_impl_udp.cpp index 7eba893..00d501d 100644 --- a/src/client_impl_udp.cpp +++ b/src/client_impl_udp.cpp @@ -95,6 +95,15 @@ struct TlsClientContext static constexpr unsigned char alpn[] = "\x02h3"; SSL_CTX_set_alpn_protos(ctx, alpn, sizeof(alpn) - 1); + // + // Same order as the server, so AES-128 GCM is also picked against peers that leave the + // choice to the client. + // + if (SSL_CTX_set_ciphersuites(ctx, "TLS_AES_128_GCM_SHA256:TLS_AES_256_GCM_SHA384:" + "TLS_CHACHA20_POLY1305_SHA256") != 1) + throw std::runtime_error(std::string{"SSL_CTX_set_ciphersuites: "} + + ERR_error_string(ERR_get_error(), nullptr)); + // // TODO: verify the server certificate (e.g. against pki/out/root.pem) instead of accepting // anything. @@ -133,6 +142,23 @@ nghttp3_nv make_nv(std::string_view name, std::string_view value) return nv; } +/// Logs a block of headers, one per line, in the same style as the received ones. +void log_headers(std::string_view log_prefix, const std::vector& nva) +{ + for (const auto& nv : nva) + logd("[{}] \x1b[1;34m{}\x1b[0m: {}", log_prefix, + std::string_view(reinterpret_cast(nv.name), nv.namelen), + std::string_view(reinterpret_cast(nv.value), nv.valuelen)); +} + +/// Same, for a header block buffered up by the recv_header callback. +void log_headers(std::string_view log_prefix, + const std::vector>& headers) +{ + for (const auto& [name, value] : headers) + logd("[{}] \x1b[1;34m{}\x1b[0m: {}", log_prefix, name, value); +} + void ngtcp2_log_printf(void* /*user*/, const char* fmt, ...) noexcept { if (!spdlog::default_logger()->should_log(spdlog::level::trace)) @@ -183,6 +209,13 @@ class Http3ClientStream : public std::enable_shared_from_this unsigned int status_code = 0; Fields response_fields; std::optional content_length; + + // + // The header block as it arrived, buffered so that h3_cb_end_headers() can log it in one go, + // below the status line, instead of one stray line per header as they come in. Only filled + // when debug logging is on, and dropped again as soon as it has been logged. + // + std::vector> received_headers; bool headers_received = false; bool response_delivered = false; client::Request::GetResponseHandler response_handler; @@ -232,6 +265,7 @@ class Http3ClientStream : public std::enable_shared_from_this // std::deque> pending_read; asio::const_buffer read_head; + asio::const_buffer incoming; // chunk on_data_chunk() is delivering, not yet taken bool eof_received = false; ReadSomeHandler read_handler; asio::mutable_buffer read_handler_buffer; @@ -623,10 +657,27 @@ void Http3ClientStream::on_data_chunk(const uint8_t* data, size_t len) { if (len == 0) return; - pending_read.emplace_back(data, data + len); - if (read_head.size() == 0) - read_head = asio::buffer(pending_read.front()); + + // + // nghttp3 hands us a view into the packet it is parsing, valid only until this callback + // returns. Offer it to a waiting reader as it stands before copying it anywhere: a handler + // that keeps a read outstanding -- the usual shape -- takes the bytes with a single copy, and + // the vector that would otherwise carry them (a malloc, a copy in, a copy out and a free, per + // QUIC packet, so about fifty of each per 64k of request body) is never created at all. Only + // what the reader could not take is parked for later. + // + auto self = shared_from_this(); // a resumed reader may drop the last reference to this stream + incoming = asio::const_buffer{data, len}; call_read_handler(); + + if (incoming.size() > 0) + { + auto* rest = static_cast(incoming.data()); + pending_read.emplace_back(rest, rest + incoming.size()); + if (read_head.size() == 0) + read_head = asio::buffer(pending_read.front()); + incoming = {}; + } } void Http3ClientStream::on_eof() @@ -661,17 +712,45 @@ void Http3ClientStream::call_read_handler() size_t consumed = 0; while (read_handler) { - if (asio::buffer_size(read_head) > 0) + if (read_head.size() > 0 || incoming.size() > 0) { - auto copied = asio::buffer_copy(read_handler_buffer, read_head); - read_head += copied; - consumed += copied; - if (read_head.size() == 0) + // + // Fill the caller's buffer from as many queued chunks as it takes, rather than stopping + // at the end of the first one. Each chunk is what arrived in a single QUIC packet -- a + // little over a kilobyte -- so handing them out one per read turns a 64k body into ~48 + // reads, and a handler that answers every read with a write (an echo) pays a full + // round trip for each of them, because a body write only completes once the peer has + // acknowledged it (see the comment above write_active). + // + auto dest = read_handler_buffer; + size_t copied = 0; + while (dest.size() > 0 && read_head.size() > 0) { - pending_read.pop_front(); - read_head = - pending_read.empty() ? asio::const_buffer{} : asio::buffer(pending_read.front()); + auto n = asio::buffer_copy(dest, read_head); + dest += n; + read_head += n; + copied += n; + if (read_head.size() == 0) + { + pending_read.pop_front(); + read_head = + pending_read.empty() ? asio::const_buffer{} : asio::buffer(pending_read.front()); + } } + + // + // ... and last from the chunk being delivered right now, which on_data_chunk() offers + // through `incoming` instead of parking it in a vector of its own first. Queued chunks + // go first: they arrived earlier. + // + if (dest.size() > 0 && incoming.size() > 0) + { + auto n = asio::buffer_copy(dest, incoming); + incoming += n; + copied += n; + } + + consumed += copied; swap_and_invoke(read_handler, boost::system::error_code{}, copied); continue; } @@ -712,9 +791,6 @@ void Http3ClientStream::start_write(WriteHandler&& handler, asio::const_buffer b const bool is_eof = (n == 0); logd("[{}] start_write: n={} is_eof={}", log_prefix, n, is_eof); - // Only one async_write() may be active at a time -- see the class comment above write_active. - assert(!write_active); - // // Once accepted, the caller's intent to end the request 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 @@ -744,6 +820,13 @@ void Http3ClientStream::start_write(WriteHandler&& handler, asio::const_buffer b return; } + // + // Only one async_write() may be active at a time -- see the class comment above write_active. + // The re-issued EOF handled above is not an exception to that: it adopts the FIN that is + // already in flight instead of starting a write of its own, and has returned by now. + // + assert(!write_active); + if (is_eof) eof_submitted = true; @@ -891,11 +974,29 @@ void Http3ClientStream::on_write_consumed(size_t n) n = std::min(n, write_chunk.size() - write_confirmed); write_confirmed += n; + if (write_confirmed < write_chunk.size()) + return; + // The write is fully done once its current chunk is confirmed and there is no more of // write_source left to carve into further chunks -- data_reader() advances write_chunk/ // write_source_copied otherwise, so this is the terminal state. - if (write_confirmed == write_chunk.size() && write_source_copied == asio::buffer_size(write_source)) + if (write_source_copied == asio::buffer_size(write_source)) + { finish_active_write(); + return; + } + + // + // There is more of write_source to carve into chunks, but nghttp3 may have asked for data + // while this chunk was offered and still unconfirmed, in which case data_reader() answered + // NGHTTP3_ERR_WOULDBLOCK -- and a blocked stream is never polled again until it is explicitly + // resumed. Now that the chunk is confirmed, there is something new to hand out, so unblock + // the stream. Without this, any single async_write() larger than kWriteChunkSize stalls here + // forever, with the response body truncated and no FIN. + // + if (auto h3 = session.h3()) + nghttp3_conn_resume_stream(h3, id); + session.wake_write(); } void Http3ClientStream::finish_active_write() @@ -1001,6 +1102,7 @@ void Http3ClientStream::delete_reader() auto self = shared_from_this(); // see delete_writer() pending_read.clear(); read_head = {}; + incoming = {}; maybe_close(); } @@ -1154,7 +1256,13 @@ int Http3ClientSession::init(asio::ip::udp::endpoint remote) ngtcp2_settings settings; ngtcp2_settings_default(&settings); settings.initial_ts = ngtcp2::util::timestamp(); - settings.log_printf = &ngtcp2_log_printf; + + // + // See the server-side counterpart: ngtcp2 does the full frame formatting before calling this, + // so only install it when trace logging is actually enabled. + // + if (spdlog::default_logger_raw()->should_log(spdlog::level::trace)) + settings.log_printf = &ngtcp2_log_printf; ngtcp2_transport_params params; ngtcp2_transport_params_default(¶ms); @@ -1272,7 +1380,16 @@ void Http3ClientSession::close() for (auto& stream : streams) stream->fail(errc::make_error_code(errc::connection_reset)); - if (conn_ && !ngtcp2_conn_in_closing_period(conn_) && !ngtcp2_conn_in_draining_period(conn_)) + // + // An idle-timed-out (or dropped) connection is discarded silently: RFC 9000 has no + // CONNECTION_CLOSE for it, and there is nobody left listening anyway -- writing one would + // just put a packet on a path whose peer has been gone for a full idle period. + // + const bool silent = last_error_.type == NGTCP2_CCERR_TYPE_IDLE_CLOSE || + last_error_.type == NGTCP2_CCERR_TYPE_DROP_CONN; + + if (conn_ && !silent && !ngtcp2_conn_in_closing_period(conn_) && + !ngtcp2_conn_in_draining_period(conn_)) { std::array closebuf; ngtcp2_path_storage ps; @@ -1523,7 +1640,16 @@ int Http3ClientSession::handle_expiry() auto now = ngtcp2::util::timestamp(); if (auto rv = ngtcp2_conn_handle_expiry(conn_, now); rv != 0) { - logw("[{}] ngtcp2_conn_handle_expiry: {}", log_prefix_, ngtcp2_strerror(rv)); + // + // NGTCP2_ERR_IDLE_CLOSE is how a connection whose peer stopped talking ends: a normal + // end of life, not a failure worth a warning. close() then discards it silently, see + // there. + // + if (rv == NGTCP2_ERR_IDLE_CLOSE) + logi("[{}] idle timeout, dropping connection", log_prefix_); + else + logw("[{}] ngtcp2_conn_handle_expiry: {}", log_prefix_, ngtcp2_strerror(rv)); + ngtcp2_ccerr_set_liberr(&last_error_, rv, nullptr, 0); return handle_error(rv); } @@ -1832,7 +1958,8 @@ int Http3ClientSession::h3_cb_recv_header(nghttp3_conn*, int64_t stream_id, int3 if (!s) return 0; - logd("[{}] \x1b[1;34m{}\x1b[0m: {}", s->log_prefix, name_view, value_view); + if (spdlog::default_logger_raw()->should_log(spdlog::level::debug)) + s->received_headers.emplace_back(name_view, value_view); try { @@ -1867,6 +1994,7 @@ int Http3ClientSession::h3_cb_end_headers(nghttp3_conn*, int64_t stream_id, int return 0; logd("[{}] response headers: status={}", s->log_prefix, s->status_code); + log_headers(s->log_prefix, std::exchange(s->received_headers, {})); s->headers_received = true; s->deliver_response(); return 0; @@ -1967,6 +2095,7 @@ void Http3ClientSession::async_submit(SubmitHandler&& handler, boost::urls::url } logd("[{}] async_submit: new stream ID: {}", stream->log_prefix, stream_id); + log_headers(stream->log_prefix, nva); wake_write(); post(get_executor(), [handler = std::move(handler), diff --git a/src/file_handler.cpp b/src/file_handler.cpp new file mode 100644 index 0000000..26562f0 --- /dev/null +++ b/src/file_handler.cpp @@ -0,0 +1,364 @@ +#include "anyhttp/file_handler.hpp" +#include "anyhttp/formatter.hpp" // IWYU pragma: keep +#include "anyhttp/request_handlers.hpp" // for send() + +#include +#include +#include +#include + +#include +#include +#include +#include +#include + +using namespace std::string_view_literals; +using namespace anyhttp; +using boost::system::error_code; + +// ================================================================================================= + +namespace +{ + +namespace fs = std::filesystem; + +error_code from_errno(int error) { return {error, boost::system::system_category()}; } + +// +// A file mapped into memory for as long as this object lives. An empty file maps to an empty +// span: mmap() rejects a zero length, and there is nothing to send anyway. +// +class MappedFile +{ +public: + MappedFile() = default; + MappedFile(MappedFile&& other) noexcept + : m_data(std::exchange(other.m_data, nullptr)), m_size(std::exchange(other.m_size, 0)), + m_mtime(other.m_mtime), m_id(other.m_id) + { + } + MappedFile& operator=(MappedFile&& other) noexcept + { + std::swap(m_data, other.m_data); + std::swap(m_size, other.m_size); + std::swap(m_mtime, other.m_mtime); + std::swap(m_id, other.m_id); + return *this; + } + ~MappedFile() + { + if (m_data) + ::munmap(m_data, m_size); + } + + static expected open(const fs::path& path) + { + const int fd = ::open(path.c_str(), O_RDONLY | O_CLOEXEC); + if (fd < 0) + return std::unexpected(from_errno(errno)); + auto close = defer([fd] { ::close(fd); }); + + struct stat st{}; + if (::fstat(fd, &st) < 0) + return std::unexpected(from_errno(errno)); + + // + // A directory can be open()ed, but not mapped -- and we do not serve listings anyway. + // Anything else that is not a regular file (FIFO, device, socket) has no size to speak of. + // + if (!S_ISREG(st.st_mode)) + return std::unexpected(from_errno(S_ISDIR(st.st_mode) ? EISDIR : EINVAL)); + + MappedFile file; + file.m_size = static_cast(st.st_size); + file.m_mtime = std::chrono::system_clock::from_time_t(st.st_mtime); + file.m_id = Identity{st}; + if (file.m_size == 0) + return file; + + void* data = ::mmap(nullptr, file.m_size, PROT_READ, MAP_PRIVATE, fd, 0); + if (data == MAP_FAILED) + return std::unexpected(from_errno(errno)); + + ::posix_madvise(data, file.m_size, POSIX_MADV_SEQUENTIAL); + file.m_data = data; + return file; + } + + // + // What the mapping was made from, so a cached mapping can be checked against the file that is + // on disk now. st_ino/st_dev catch a replaced file (the usual atomic rename), the rest catches + // a file modified in place. + // + struct Identity + { + dev_t dev; + ino_t ino; + off_t size; + decltype(std::declval().st_mtim) mtim; + + explicit Identity(const struct stat& st = {}) + : dev(st.st_dev), ino(st.st_ino), size(st.st_size), mtim(st.st_mtim) + { + } + bool operator==(const Identity& other) const noexcept + { + return dev == other.dev && ino == other.ino && size == other.size && + mtim.tv_sec == other.mtim.tv_sec && mtim.tv_nsec == other.mtim.tv_nsec; + } + }; + + size_t size() const noexcept { return m_size; } + auto mtime() const noexcept { return m_mtime; } + const Identity& identity() const noexcept { return m_id; } + std::span bytes() const noexcept + { + return {static_cast(m_data), m_size}; + } + +private: + void* m_data = nullptr; + size_t m_size = 0; + std::chrono::system_clock::time_point m_mtime; + Identity m_id{}; +}; + +// +// Map the part of the request path below 'prefix' onto 'root'. weakly_canonical() resolves ".." +// and symlinks, so comparing the result against the canonical root catches any attempt to reach +// outside of it. +// +expected resolve(std::string_view path, std::string_view prefix, const fs::path& root) +{ + const auto reject = std::unexpected(from_errno(ENOENT)); + + if (!path.starts_with(prefix)) + return reject; + path.remove_prefix(prefix.size()); + if (!path.empty() && !path.starts_with('/')) + return reject; // 'prefix' matched in the middle of a segment, e.g. "/testament" for "/test" + + while (path.starts_with('/')) + path.remove_prefix(1); + + std::error_code ec; + const auto base = fs::weakly_canonical(root, ec); + if (ec) + return std::unexpected(from_errno(ec.value())); + + const auto file = fs::weakly_canonical(base / fs::path(path), ec); + if (ec) + return std::unexpected(from_errno(ec.value())); + + const auto relative = file.lexically_relative(base); + if (relative.empty() || *relative.begin() == "..") + return reject; + + return file; +} + +std::string_view content_type(const fs::path& path) +{ + static constexpr std::pair types[] = { + {".css", "text/css"}, {".gif", "image/gif"}, {".htm", "text/html"}, + {".html", "text/html"}, {".jpeg", "image/jpeg"}, {".jpg", "image/jpeg"}, + {".js", "text/javascript"}, {".json", "application/json"}, {".pdf", "application/pdf"}, + {".png", "image/png"}, {".svg", "image/svg+xml"}, {".txt", "text/plain"}, + {".xml", "application/xml"}, {".zip", "application/zip"}}; + + auto extension = path.extension().string(); + std::ranges::transform(extension, extension.begin(), + [](unsigned char ch) { return std::tolower(ch); }); + + const auto it = + std::ranges::find(types, extension, &std::pair::first); + return it == std::ranges::end(types) ? "application/octet-stream"sv : it->second; +} + +unsigned status_for(const error_code& ec) +{ + switch (ec.value()) + { + case ENOENT: + case ENOTDIR: + case EISDIR: + case ENAMETOOLONG: + return 404; + case EACCES: + case EPERM: + return 403; + default: + return 500; + } +} + +awaitable respond(server::Response& response, unsigned status) +{ + co_await response.async_submit(status, fields({{"Content-Length", 0}})); + co_await response.async_write({}); +} + +// +// Everything serving one request needs, computed once per file instead of once per request: the +// mapping plus the response headers derived from it. +// +struct CachedFile +{ + fs::path resolved; + MappedFile file; + std::string last_modified; + std::string_view content_type; +}; + +// +// Mapping a file per request costs an mmap()/munmap() pair plus the faults to populate the +// mapping, which measured as the single largest per-request cost when serving a file that is +// already in the page cache -- larger than resolving the path, and larger than the QUIC send +// path itself. Keeping the mapping alive across requests removes all of that. +// +// A hit still stat()s the file, so a file replaced or modified on disk is picked up on the next +// request; that is one syscall instead of the ~17 a full resolve-and-map takes. What a hit does +// *not* redo is resolve(), so re-pointing a symlink along the path is only noticed once the file +// it used to point at changes. That direction is safe -- a stale entry can only keep serving a +// file that already passed the containment check -- but it is why this is a cache of resolved +// paths and not a cache of open files. +// +class FileCache +{ +public: + // Bounded so that a large tree cannot pin unbounded address space; least recently used first. + static constexpr size_t max_entries = 256; + static constexpr size_t max_bytes = 64u << 20; + + expected> get(const std::string& request_path, + std::string_view prefix, const fs::path& root) + { + if (auto hit = lookup(request_path)) + { + struct stat st{}; + if (::stat(hit->resolved.c_str(), &st) == 0 && + MappedFile::Identity{st} == hit->file.identity()) + return hit; + } + + // + // Miss, or the file changed underneath us. Build the entry outside the lock: mapping is the + // expensive part and two requests racing on the same path may as well both do it. + // + const auto resolved = resolve(request_path, prefix, root); + if (!resolved) + return std::unexpected(resolved.error()); + + auto mapped = MappedFile::open(*resolved); + if (!mapped) + return std::unexpected(mapped.error()); + + // Read mtime() before the move, rather than relying on argument evaluation order. + auto last_modified = format_http_date(mapped->mtime()); + auto entry = std::make_shared(*resolved, std::move(*mapped), + std::move(last_modified), + content_type(*resolved)); + insert(request_path, entry); + return entry; + } + +private: + std::shared_ptr lookup(const std::string& key) + { + const std::lock_guard lock{m_mutex}; + const auto it = m_entries.find(key); + if (it == m_entries.end()) + return nullptr; + m_lru.splice(m_lru.begin(), m_lru, it->second.lru); // most recently used first + return it->second.entry; + } + + void insert(const std::string& key, std::shared_ptr entry) + { + const std::lock_guard lock{m_mutex}; + + if (const auto it = m_entries.find(key); it != m_entries.end()) + { + m_bytes -= it->second.entry->file.size(); + m_lru.erase(it->second.lru); + m_entries.erase(it); + } + + m_bytes += entry->file.size(); + m_lru.push_front(key); + m_entries.emplace(key, Slot{std::move(entry), m_lru.begin()}); + + // + // Keep at least the entry just inserted, so a file larger than the byte budget still gets + // served from the cache rather than being evicted immediately every time. + // + while (m_entries.size() > 1 && (m_entries.size() > max_entries || m_bytes > max_bytes)) + { + const auto victim = m_entries.find(m_lru.back()); + m_bytes -= victim->second.entry->file.size(); + m_entries.erase(victim); + m_lru.pop_back(); + } + } + + struct Slot + { + std::shared_ptr entry; + std::list::iterator lru; + }; + + std::mutex m_mutex; + std::unordered_map m_entries; + std::list m_lru; + size_t m_bytes = 0; +}; + +FileCache g_cache; + +} // namespace + +namespace anyhttp +{ + +awaitable serve_file(server::Request request, server::Response response, fs::path root, + std::string prefix) +{ + // + // Read the request body to EOF before responding. A GET normally carries none, but HTTP/1.1 + // 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) + ; + + const std::string path = request.url().path(); + const auto entry = g_cache.get(path, prefix, root); + if (!entry) + { + logw("serve_file: {}: {}", path, entry.error().message()); + co_await respond(response, status_for(entry.error())); + co_return; + } + + const auto& file = (*entry)->file; + logd("serve_file: {} ({} bytes)", (*entry)->resolved.native(), file.size()); + co_await response.async_submit(200, fields({{"Content-Length", file.size()}, + {"Content-Type", (*entry)->content_type}, + {"Last-Modified", (*entry)->last_modified}})); + + // + // The shared_ptr lives in the coroutine frame, so the mapping stays valid across the write no + // matter how we leave -- normally, by exception or by cancellation -- even if the entry is + // 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({}); +} + +} // namespace anyhttp diff --git a/src/nghttp2_stream.cpp b/src/nghttp2_stream.cpp index 8ded962..cb8a099 100644 --- a/src/nghttp2_stream.cpp +++ b/src/nghttp2_stream.cpp @@ -17,6 +17,7 @@ #include #include #include +#include #include #include @@ -202,7 +203,8 @@ void NGHttp2Writer::async_submit(StatusHandler&& handler, unsigned int sta return; } - logd("[{}] async_submit: {}", stream->logPrefix, status_code); + logd("[{}] {} {}", stream->logPrefix, status_code, + boost::beast::http::obsolete_reason(boost::beast::http::int_to_status(status_code))); auto nva = std::vector(); // nva.reserve(3 + headers.size()); @@ -227,6 +229,9 @@ void NGHttp2Writer::async_submit(StatusHandler&& handler, unsigned int sta nva.push_back(make_nv_ls("content-length", length_str)); } + for (auto nv : nva) + logd("[{0}] \x1b[1;34m{1:n}\x1b[0m: {1:v}", stream->logPrefix, nv); + // TODO: If we already know that there is no body, don't set a producer. nghttp2_data_provider2 prd; prd.source.ptr = stream; @@ -762,10 +767,8 @@ void NGHttp2Stream::on_request() server::Response response(std::make_unique>(*this)); auto& server = dynamic_cast(parent).server(); - if (auto& handler = server.requestHandlerCoro()) + if (auto& handler = server.requestHandler()) co_spawn(get_executor(), handler(std::move(request), std::move(response)), detached); - else if (auto& handler = server.requestHandler()) - server.requestHandler()(std::move(request), std::move(response)); else { loge("[{}] on_request: no request handler!", logPrefix); diff --git a/src/request_handlers.cpp b/src/request_handlers.cpp index 0c62b12..672a2fb 100644 --- a/src/request_handlers.cpp +++ b/src/request_handlers.cpp @@ -109,7 +109,7 @@ awaitable not_found(server::Request, server::Response response) awaitable eat_request(server::Request request, server::Response response) { - logi("eat_request: going to eat {} bytes", request.content_length().value_or(-1)); + logd("eat_request: going to eat {} bytes", request.content_length().value_or(-1)); co_await response.async_submit(200, {}); co_await response.async_write({}); @@ -127,7 +127,7 @@ awaitable eat_request(server::Request request, server::Response response) logd("eat_request: ate {} bytes", n); bytes += n; } - logi("eat_request: ate {} bytes", bytes); + logd("eat_request: ate {} bytes", bytes); } catch (const boost::system::system_error& e) { diff --git a/src/server.cpp b/src/server.cpp index 3215597..40e7022 100644 --- a/src/server.cpp +++ b/src/server.cpp @@ -116,11 +116,6 @@ void Server::setRequestHandler(RequestHandler&& handler) impl->setRequestHandler(std::move(handler)); } -void Server::setRequestHandlerCoro(RequestHandlerCoro&& handler) -{ - impl->setRequestHandler(std::move(handler)); -} - asio::any_io_executor Server::get_executor() const noexcept { return impl->get_executor(); } asio::ip::tcp::endpoint Server::local_endpoint() const { return impl->local_endpoint(); } diff --git a/src/server_impl_udp.cpp b/src/server_impl_udp.cpp index fd0d07e..5bb6573 100644 --- a/src/server_impl_udp.cpp +++ b/src/server_impl_udp.cpp @@ -46,12 +46,14 @@ #include #include +#include #include #include #include #include #include #include +#include #include #include #include @@ -74,6 +76,10 @@ struct Endpoint { ngtcp2::Address addr; int fd; + + // Testing aid, see server::Config::drop_rate_rx/tx. + double drop_rate_rx = 0.0; + double drop_rate_tx = 0.0; }; // ================================================================================================= @@ -110,6 +116,15 @@ struct TlsServerContext SSL_OP_NO_ANTI_REPLAY); SSL_CTX_set_mode(ctx, SSL_MODE_RELEASE_BUFFERS); + // + // Prefer AES-128 over AES-256 GCM, like ngtcp2's example server does. Combined with + // SSL_OP_CIPHER_SERVER_PREFERENCE above, this is what actually picks the bulk cipher. + // + if (SSL_CTX_set_ciphersuites(ctx, "TLS_AES_128_GCM_SHA256:TLS_AES_256_GCM_SHA384:" + "TLS_CHACHA20_POLY1305_SHA256") != 1) + throw std::runtime_error(std::string{"SSL_CTX_set_ciphersuites: "} + + ERR_error_string(ERR_get_error(), nullptr)); + SSL_CTX_set_alpn_select_cb(ctx, &TlsServerContext::alpn_select_cb, nullptr); if (SSL_CTX_use_PrivateKey_file(ctx, "pki/out/server-key.pem", SSL_FILETYPE_PEM) != 1) @@ -171,11 +186,34 @@ std::string cid_key(const uint8_t* data, size_t len) // ------------------------------------------------------------------------------------------------- -int send_udp(int fd, const sockaddr* sa, socklen_t salen, std::span data) +// +// Testing aid: rolls the dice for a single QUIC packet. `rate` is the probability of the packet +// being dropped, 0.0 (never, the default) to 1.0 (always). The generator is deliberately not +// seeded deterministically -- this is meant to shake out loss handling over many runs, not to +// reproduce one exact sequence. +// +bool drop_packet(double rate) +{ + if (rate <= 0.0) + return false; + + static thread_local std::mt19937 rng{std::random_device{}()}; + return std::uniform_real_distribution{0.0, 1.0}(rng) < rate; +} + +// ------------------------------------------------------------------------------------------------- + +int send_udp(const Endpoint& ep, const sockaddr* sa, socklen_t salen, std::span data) { + if (drop_packet(ep.drop_rate_tx)) + { + // logw("*** dropping outgoing packet ({} bytes) ***", data.size()); + return 0; // pretend it went out; ngtcp2 will retransmit + } + for (;;) { - auto n = ::sendto(fd, data.data(), data.size(), 0, sa, salen); + auto n = ::sendto(ep.fd, data.data(), data.size(), 0, sa, salen); if (n == -1) { if (errno == EINTR) @@ -195,15 +233,16 @@ int send_udp(int fd, const sockaddr* sa, socklen_t salen, std::span data, - size_t gso_size, bool& no_gso) +int send_udp_gso(const Endpoint& ep, const sockaddr* sa, socklen_t salen, + std::span data, size_t gso_size, bool& no_gso) { - if (no_gso || data.size() <= gso_size) + // With TX dropping enabled, go packet by packet so each one can be dropped individually. + if (no_gso || data.size() <= gso_size || ep.drop_rate_tx > 0.0) { for (; !data.empty();) { auto len = std::min(gso_size, data.size()); - if (send_udp(fd, sa, salen, data.first(len)) != 0) + if (send_udp(ep, sa, salen, data.first(len)) != 0) return -1; data = data.subspan(len); } @@ -229,7 +268,7 @@ int send_udp_gso(int fd, const sockaddr* sa, socklen_t salen, std::span& nva) +{ + for (const auto& nv : nva) + logd("[{}] \x1b[1;34m{}\x1b[0m: {}", log_prefix, + std::string_view(reinterpret_cast(nv.name), nv.namelen), + std::string_view(reinterpret_cast(nv.value), nv.valuelen)); +} + +/// Same, for a header block buffered up by the recv_header callback. +void log_headers(std::string_view log_prefix, + const std::vector>& headers) +{ + for (const auto& [name, value] : headers) + logd("[{}] \x1b[1;34m{}\x1b[0m: {}", log_prefix, name, value); +} + void ngtcp2_log_printf(void* /*user*/, const char* fmt, ...) noexcept { if (!spdlog::default_logger()->should_log(spdlog::level::trace)) @@ -282,14 +338,6 @@ void ngtcp2_log_printf(void* /*user*/, const char* fmt, ...) noexcept class Http3Session; class Http3Stream; -// -// Bound on how much of the caller's async_write() buffer we copy into write_chunk at a time (see -// Http3Stream's write_* members) -- copying is paced by how much nghttp3/ngtcp2 actually drains, -// rather than copying a huge caller buffer in one synchronous allocation+memcpy, mirroring -// nghttp2's own per-call copy into its frame buffer. -// -inline constexpr size_t kWriteChunkSize = 16 * 1024; - class Http3Stream : public std::enable_shared_from_this { public: @@ -308,6 +356,13 @@ class Http3Stream : public std::enable_shared_from_this std::optional content_length; Fields request_fields; + // + // The header block as it arrived, buffered so that h3_cb_end_headers() can log it in one go, + // below the request line, instead of one stray line per header as they come in. Only filled + // when debug logging is on, and dropped again as soon as it has been logged. + // + std::vector> received_headers; + // // Response state (populated by user via Http3Writer). // @@ -322,6 +377,7 @@ class Http3Stream : public std::enable_shared_from_this // std::deque> pending_read; asio::const_buffer read_head; // view of pending_read.front() not yet delivered + asio::const_buffer incoming; // chunk on_data_chunk() is delivering, not yet taken bool eof_received = false; ReadSomeHandler read_handler; asio::mutable_buffer read_handler_buffer; @@ -333,35 +389,34 @@ class Http3Stream : public std::enable_shared_from_this // this is flat per-stream state rather than a queue of pending writes. See the client-side // counterpart (Http3ClientStream in client_impl_udp.cpp) for the fuller rationale. // - // write_source is the caller's buffer, referenced (not copied) the way asio::async_write - // generally requires -- it must stay valid until write_handler fires (and no longer: nghttp3 - // only ever gets pointers into write_chunk, our own copy, so once the handler has fired -- - // including via cancellation -- the caller's buffer is no longer touched). - // - // write_chunk is a bounded (<= kWriteChunkSize) slice of write_source, lazily refilled by - // data_reader() as it's drained. write_offered and write_confirmed are tracked separately - // because nghttp3 may call data_reader() several times in a row for the same stream before ever - // reporting consumption back via on_write_consumed() -- e.g. to gather more vecs than fit in a - // single call. If data_reader() just kept re-handing out write_chunk[0, write_chunk.size()) - // unconditionally (tracking only write_confirmed), nghttp3 would treat each repeat offer as - // *additional*, distinct stream bytes and duplicate the content on the wire. write_offered - // marks how much has already been handed to nghttp3 (whether or not it has been placed in a - // packet yet) so a repeat call sees nothing new and gets NGHTTP3_ERR_WOULDBLOCK instead. + // write_source is the caller's buffer, passed on to nghttp3 by reference: data_reader() points + // the nghttp3_vec straight into it, so the response body is never copied on its way down to the + // nghttp3/ngtcp2 boundary, however large it is -- the mmap()ed file of serve_file() travels + // from the page cache into QUIC packets without an intermediate byte. + // + // What that costs is *when* the write completes. ngtcp2 keeps pointing into this memory for as + // long as the bytes may still have to be retransmitted (it only ever copies the nghttp3_vec + // descriptors, never the payload), and running the write handler is what releases the caller's + // buffer -- so the handler has to wait for the data to be acknowledged. This is the model + // nghttp3 documents for its read_data callback: "the application must retain data until they + // are safe to free; it is notified by nghttp3_acked_stream_data". HTTP/2 completes a write as + // soon as nghttp2 has copied it into its own frame buffer, so a single async_write() there is + // done roughly a memcpy later, and here roughly a round trip later. + // + // write_offered tracks how much of write_source has been handed to nghttp3, which may ask + // again before any of it goes out and would take a repeated offer as *additional*, distinct + // stream bytes -- duplicating the body on the wire -- so a repeat call gets + // NGHTTP3_ERR_WOULDBLOCK instead. write_acked tracks what came back through nghttp3's + // acked_stream_data callback; the write is complete once that has caught up with write_source. // bool write_active = false; asio::const_buffer write_source; - size_t write_source_copied = 0; - std::vector write_chunk; size_t write_offered = 0; - size_t write_confirmed = 0; + size_t write_acked = 0; bool write_is_eof = false; WriteHandler write_handler; uint64_t write_token = 0; uint64_t next_write_token = 1; - - std::vector> in_flight_writes; // kept alive for the stream's lifetime -- - // ngtcp2 may still need this memory for - // retransmission until acked bool eof_submitted = false; // user signalled EOF via empty write bool eof_sent_to_h3 = false; // NGHTTP3_DATA_FLAG_EOF returned @@ -385,11 +440,11 @@ class Http3Stream : public std::enable_shared_from_this void submit_response(); void start_write(WriteHandler&& handler, asio::const_buffer buffer); nghttp3_ssize data_reader(nghttp3_vec* vec, size_t veccnt, uint32_t* pflags); - void on_write_consumed(size_t n); + void on_write_acked(size_t n); private: void bind_write_cancellation(WriteHandler& handler, uint64_t token); // arms cancellation - void finish_active_write(); // completes the active write once fully handed to nghttp3 + void finish_active_write(); // completes the active write and releases the caller's buffer public: // Called from either reader or writer destructor. @@ -424,6 +479,16 @@ class Http3Session : public Session::Impl int on_read(const ngtcp2_pkt_info& pi, std::span data, const ngtcp2::Address& remote); int write_streams(); + + // + // on_read() does not write; it only marks the session here, and Server::Impl calls + // flush_write() once the whole receive batch has been fed to ngtcp2. Writing per datagram + // means every aggregate pass only sees what happened to be queued at that instant, so a + // response goes out as several small GSO batches instead of one big one. + // + void defer_write() noexcept { write_pending_ = true; } + int flush_write(); + ngtcp2_ssize write_pkt(ngtcp2_path* path, ngtcp2_pkt_info* pi, uint8_t* dest, size_t destlen, ngtcp2_tstamp ts); void update_timer(); @@ -527,6 +592,8 @@ class Http3Session : public Session::Impl // // nghttp3 callback bridges // + static int h3_cb_acked_stream_data(nghttp3_conn*, int64_t stream_id, uint64_t datalen, + void* user, void*); static int h3_cb_stream_close(nghttp3_conn*, int64_t stream_id, uint64_t app_error_code, void* user, void*); static int h3_cb_recv_data(nghttp3_conn*, int64_t stream_id, const uint8_t* data, size_t datalen, @@ -570,6 +637,9 @@ class Http3Session : public Session::Impl std::string log_prefix_; + bool write_pending_ = false; // set by on_read(), acted on by flush_write() + bool write_posted_ = false; // a wake_write() flush is already on the way + std::vector conn_closebuf_; // buffered CONNECTION_CLOSE packet // Aggregated TX buffer: ngtcp2_conn_write_aggregate_pkt2() packs as many same-sized @@ -740,10 +810,27 @@ void Http3Stream::on_data_chunk(const uint8_t* data, size_t len) { if (len == 0) return; - pending_read.emplace_back(data, data + len); - if (read_head.size() == 0) - read_head = asio::buffer(pending_read.front()); + + // + // nghttp3 hands us a view into the packet it is parsing, valid only until this callback + // returns. Offer it to a waiting reader as it stands before copying it anywhere: a handler + // that keeps a read outstanding -- the usual shape -- takes the bytes with a single copy, and + // the vector that would otherwise carry them (a malloc, a copy in, a copy out and a free, per + // QUIC packet, so about fifty of each per 64k of request body) is never created at all. Only + // what the reader could not take is parked for later. + // + auto self = shared_from_this(); // a resumed reader may drop the last reference to this stream + incoming = asio::const_buffer{data, len}; call_read_handler(); + + if (incoming.size() > 0) + { + auto* rest = static_cast(incoming.data()); + pending_read.emplace_back(rest, rest + incoming.size()); + if (read_head.size() == 0) + read_head = asio::buffer(pending_read.front()); + incoming = {}; + } } void Http3Stream::on_eof() @@ -778,17 +865,45 @@ void Http3Stream::call_read_handler() size_t consumed = 0; while (read_handler) { - if (asio::buffer_size(read_head) > 0) + if (read_head.size() > 0 || incoming.size() > 0) { - auto copied = asio::buffer_copy(read_handler_buffer, read_head); - read_head += copied; - consumed += copied; - if (read_head.size() == 0) + // + // Fill the caller's buffer from as many queued chunks as it takes, rather than stopping + // at the end of the first one. Each chunk is what arrived in a single QUIC packet -- a + // little over a kilobyte -- so handing them out one per read turns a 64k body into ~48 + // reads, and a handler that answers every read with a write (an echo) pays a full + // round trip for each of them, because a body write only completes once the peer has + // acknowledged it (see the comment above write_active). + // + auto dest = read_handler_buffer; + size_t copied = 0; + while (dest.size() > 0 && read_head.size() > 0) + { + auto n = asio::buffer_copy(dest, read_head); + dest += n; + read_head += n; + copied += n; + if (read_head.size() == 0) + { + pending_read.pop_front(); + read_head = + pending_read.empty() ? asio::const_buffer{} : asio::buffer(pending_read.front()); + } + } + + // + // ... and last from the chunk being delivered right now, which on_data_chunk() offers + // through `incoming` instead of parking it in a vector of its own first. Queued chunks + // go first: they arrived earlier. + // + if (dest.size() > 0 && incoming.size() > 0) { - pending_read.pop_front(); - read_head = - pending_read.empty() ? asio::const_buffer{} : asio::buffer(pending_read.front()); + auto n = asio::buffer_copy(dest, incoming); + incoming += n; + copied += n; } + + consumed += copied; swap_and_invoke(read_handler, boost::system::error_code{}, copied); continue; } @@ -842,7 +957,9 @@ void Http3Stream::submit_response() auto status_str = std::to_string(response_status); std::vector nva; nva.reserve(16); // small typical header count; vector will grow if needed + auto date_str = format_http_date(std::chrono::system_clock::now()); nva.push_back(make_nv(":status", status_str)); + nva.push_back(make_nv("date", date_str)); nva.push_back(make_nv("server", "anyhttp-quic/0.1")); if (response_content_length) @@ -877,7 +994,10 @@ void Http3Stream::submit_response() return; } response_submitted = true; - logd("[{}] response submitted (status={})", log_prefix, response_status); + + using namespace boost::beast::http; + logd("[{}] {} {}", log_prefix, response_status, obsolete_reason(int_to_status(response_status))); + log_headers(log_prefix, nva); } void Http3Stream::start_write(WriteHandler&& handler, asio::const_buffer buffer) @@ -886,9 +1006,6 @@ void Http3Stream::start_write(WriteHandler&& handler, asio::const_buffer buffer) const bool is_eof = (n == 0); logd("[{}] start_write: n={} is_eof={}", log_prefix, n, is_eof); - // Only one async_write() may be active at a time -- see the class comment above write_active. - assert(!write_active); - // // Once accepted, the caller's intent to end the response 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 @@ -918,6 +1035,13 @@ void Http3Stream::start_write(WriteHandler&& handler, asio::const_buffer buffer) return; } + // + // Only one async_write() may be active at a time -- see the class comment above write_active. + // The re-issued EOF handled above is not an exception to that: it adopts the FIN that is + // already in flight instead of starting a write of its own, and has returned by now. + // + assert(!write_active); + if (is_eof) eof_submitted = true; @@ -926,10 +1050,8 @@ void Http3Stream::start_write(WriteHandler&& handler, asio::const_buffer buffer) write_active = true; write_source = buffer; // referenced, not copied -- see class comment above write_active - write_source_copied = 0; - write_chunk.clear(); write_offered = 0; - write_confirmed = 0; + write_acked = 0; write_is_eof = is_eof; write_token = token; write_handler = std::move(handler); @@ -952,11 +1074,8 @@ void Http3Stream::bind_write_cancellation(WriteHandler& handler, uint64_t token) cs.assign([this, token](asio::cancellation_type_t ct) { // - // Cancellation completes the write immediately: nghttp3/ngtcp2 only ever hold pointers into - // write_chunk (our own copy), never into the caller's buffer, so the un-copied remainder of - // write_source can simply be abandoned. Bytes already offered to nghttp3 still go out (they - // can't be un-offered), so write_chunk is retired to in_flight_writes to keep that memory - // alive. The caller may issue a fresh async_write() as soon as the handler fires. + // Cancellation completes the write immediately, without waiting for the acknowledgements + // it would normally complete on -- see below for what that costs. // if (write_token != token || !write_handler) return; // already completed naturally before the cancellation was delivered @@ -977,10 +1096,28 @@ void Http3Stream::bind_write_cancellation(WriteHandler& handler, uint64_t token) return; } logd("[{}] async_write: \x1b[1;31mcancelled\x1b[0m ({})", log_prefix, ct); - if (!write_chunk.empty()) - in_flight_writes.emplace_back(std::move(write_chunk)); - write_chunk.clear(); // moved-from + + // + // The handler runs now, and the caller is free to destroy its buffer the moment it does -- + // but nghttp3/ngtcp2 point straight into that buffer (see the class comment above + // write_active), so whatever was offered and is not acknowledged yet has to stop being + // referenced first. Only a reset can guarantee that: RESET_STREAM makes ngtcp2 drop the + // stream's queued data and keeps it from reclaiming in-flight bytes for retransmission. + // That costs nothing in expressiveness -- a body cut short mid-write is truncated, which + // is exactly what delete_writer() resets the stream for as well. + // + // A write that never got to offer a byte, or whose bytes are all acknowledged already, + // leaves nothing behind and lets the stream carry on unharmed. + // + if (write_offered > write_acked && !closed) + { + logw("[{}] async_write: cancelled with {} bytes unacknowledged, resetting stream", + log_prefix, write_offered - write_acked); + session.reset_stream(id, NGHTTP3_H3_REQUEST_CANCELLED); + closed = true; + } write_active = false; + write_source = {}; // make sure to post this -- otherwise "MAIN COROUTINE DID NOT COMPLETE" happens asio::post(get_executor(), [handler = std::move(write_handler)]() mutable { std::move(handler)(errc::make_error_code(errc::operation_canceled)); }); @@ -995,79 +1132,61 @@ nghttp3_ssize Http3Stream::data_reader(nghttp3_vec* vec, size_t veccnt, uint32_t if (!write_active) return NGHTTP3_ERR_WOULDBLOCK; - if (write_offered < write_chunk.size()) - { - 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 -- see - // class comment above write_active - return 1; - } - - // - // Current chunk fully offered. If it hasn't been confirmed yet (on_write_consumed()), there's - // nothing new until that happens -- see the class comment above write_active on why we can't - // just carve off the next slice of write_source early. - // - if (write_confirmed < write_chunk.size()) - return NGHTTP3_ERR_WOULDBLOCK; - // - // The current chunk is fully drained; retire it (ngtcp2 may still need this exact memory for - // retransmission until acked) and pull the next bounded slice out of write_source, if any. + // Hand out what is left of the caller's buffer, by reference and in one go: a nghttp3_vec is + // just a pointer and a length, so there is nothing to be gained from slicing it up, and + // nghttp3 gets to frame the whole thing as a single DATA frame. It picks up the rest by + // itself as packets are filled -- see write_pkt(), which keeps feeding the same vec to + // ngtcp2_conn_writev_stream() and advances nghttp3 by whatever went into the packet. // - 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; - if (remaining > 0) - { - const size_t take = std::min(remaining, kWriteChunkSize); - auto* src = static_cast(write_source.data()) + write_source_copied; - write_chunk.assign(src, src + take); - write_source_copied += take; - write_offered = write_chunk.size(); - write_confirmed = 0; - vec[0].base = write_chunk.data(); - vec[0].len = write_chunk.size(); + const size_t total = asio::buffer_size(write_source); + if (write_offered < total) + { + auto* base = static_cast(write_source.data()) + write_offered; + vec[0].base = const_cast(base); // nghttp3 reads through this, never writes + vec[0].len = total - write_offered; + write_offered = total; // don't offer these bytes twice -- see class comment above + // write_active return 1; } // - // Nothing left in write_source either. If this is the EOF marker (write_source is always - // empty), retire it now -- a FIN carries no stream bytes, so there is nothing for - // on_write_consumed() to report back. A non-EOF write with nothing left to offer is instead - // retired from on_write_consumed() once its last chunk is confirmed (see there). + // Everything has been offered. For a body write there is nothing new until the caller starts + // the next one (which resumes the stream), so block here rather than returning 0 bytes -- + // returning 0 without NGHTTP3_DATA_FLAG_EOF would tell nghttp3 the body ended. // if (!write_is_eof) 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. + // *pflags |= NGHTTP3_DATA_FLAG_EOF; eof_sent_to_h3 = true; finish_active_write(); return 0; } -void Http3Stream::on_write_consumed(size_t n) +void Http3Stream::on_write_acked(size_t n) { // - // n is the number of bytes of *stream* data ngtcp2 just committed to a packet, which also - // includes the HTTP/3 HEADERS frame nghttp3 sends ahead of any body -- e.g. the very first - // write_pkt() call after submit_response() drains the headers before there is an active write - // yet. Only attribute bytes once there is an active, non-EOF write to charge them against; - // clamp defensively in case a single packet still straddles the header/body boundary. + // n counts *application* data acknowledged on this stream -- nghttp3 accounts for the HTTP/3 + // framing it puts around the body itself, so unlike ngtcp2's stream offsets these bytes are + // exactly the ones the caller handed us. All of them belong to the write currently active: a + // write only completes once every byte it offered is acknowledged, so nothing can still be + // outstanding from an earlier one. Clamp defensively anyway -- an accounting mismatch should + // complete the write early, not run write_acked past the end of the buffer. // if (n == 0 || !write_active || write_is_eof) return; - n = std::min(n, write_chunk.size() - write_confirmed); - write_confirmed += n; + write_acked = std::min(write_acked + n, asio::buffer_size(write_source)); + logd("[{}] on_write_acked: {} bytes, {}/{} acknowledged", log_prefix, n, write_acked, + asio::buffer_size(write_source)); - // The write is fully done once its current chunk is confirmed and there is no more of - // write_source left to carve into further chunks -- data_reader() advances write_chunk/ - // write_source_copied otherwise, so this is the terminal state. - if (write_confirmed == write_chunk.size() && - write_source_copied == asio::buffer_size(write_source)) + if (write_acked == asio::buffer_size(write_source)) finish_active_write(); } @@ -1076,19 +1195,16 @@ void Http3Stream::finish_active_write() assert(write_active); // - // ngtcp2 may still need this memory for retransmission until the bytes are acked; rather than - // tracking acks precisely, keep every chunk alive for the life of the stream (in_flight_writes - // is freed on stream destruction). + // Invoking the handler hands the caller's buffer back to it, so this must only ever run when + // nothing points into it any more: every offered byte acknowledged (on_write_acked()), or no + // bytes offered at all (the EOF marker). // - if (!write_chunk.empty()) - in_flight_writes.emplace_back(std::move(write_chunk)); - write_chunk.clear(); // moved-from + write_source = {}; auto handler = std::move(write_handler); write_active = false; if (handler) swap_and_invoke(handler, boost::system::error_code{}); - } // ------------------------------------------------------------------------------------------------- @@ -1098,6 +1214,7 @@ void Http3Stream::delete_reader() auto self = shared_from_this(); // see delete_writer() pending_read.clear(); read_head = {}; + incoming = {}; // // The handler dropped the Request without reading the body to its end (e.g. not_found(), which @@ -1258,11 +1375,11 @@ void Http3Session::destroy() noexcept ngtcp2_pkt_info pi; ngtcp2_path_storage_zero(&ps); - auto nwrite = ngtcp2_conn_write_connection_close(conn_, &ps.path, &pi, closebuf.data(), - closebuf.size(), &last_error_, - ngtcp2::util::timestamp()); + auto nwrite = + ngtcp2_conn_write_connection_close(conn_, &ps.path, &pi, closebuf.data(), closebuf.size(), + &last_error_, ngtcp2::util::timestamp()); if (nwrite > 0) - send_udp(ep_.fd, ps.path.remote.addr, ps.path.remote.addrlen, + send_udp(ep_, ps.path.remote.addr, ps.path.remote.addrlen, {closebuf.data(), static_cast(nwrite)}); } @@ -1302,13 +1419,26 @@ void Http3Session::wake_write() // Reader/Writer destructor that runs as part of *this* session's own teardown (e.g. a // still-in-flight request/response destroyed by Server::Impl cancelling everything on // shutdown), at which point shared_from_this() would throw bad_weak_ptr. + // + // One flush per wake, not one per submission. A response submits its headers, its body and + // its EOF separately, and posting for each means the first pass writes everything and the + // rest walk the connection for nothing -- and still re-arm the timer on the way out. Arming + // once and clearing when the flush runs is what ngtcp2's example server gets for free from + // ev_io_start() on an already-active watcher. + // + if (write_posted_) + return; + write_posted_ = true; + asio::post(get_executor(), [self = weak_from_this()] { auto session = std::static_pointer_cast(self.lock()); - if (!session || session->closed_) + if (!session) return; - if (session->write_streams() == 0) - session->update_timer(); + session->write_posted_ = false; + if (session->closed_) + return; + session->flush_write(); }); } @@ -1352,7 +1482,14 @@ int Http3Session::init(const ngtcp2_cid& dcid, const ngtcp2_cid& scid, uint32_t ngtcp2_settings settings; ngtcp2_settings_default(&settings); settings.initial_ts = ngtcp2::util::timestamp(); - settings.log_printf = &ngtcp2_log_printf; + + // + // Only install the log callback when trace logging is actually enabled: ngtcp2 formats every + // frame of every packet into a string *before* invoking it, so a callback that discards its + // input still pays for the full formatting. A NULL log_printf makes ngtcp2 skip that work. + // + if (spdlog::default_logger_raw()->should_log(spdlog::level::trace)) + settings.log_printf = &ngtcp2_log_printf; ngtcp2_transport_params params; ngtcp2_transport_params_default(¶ms); @@ -1362,7 +1499,7 @@ int Http3Session::init(const ngtcp2_cid& dcid, const ngtcp2_cid& scid, uint32_t params.initial_max_data = 1_m; params.initial_max_streams_bidi = 100; params.initial_max_streams_uni = 3; - params.max_idle_timeout = std::chrono::nanoseconds(30s).count(); + params.max_idle_timeout = server_.config().idle_timeout.count(); params.original_dcid = dcid; params.original_dcid_present = 1; @@ -1443,10 +1580,10 @@ int Http3Session::on_read(const ngtcp2_pkt_info& pi, std::span da return handle_error(rv); } - if (auto wrv = write_streams(); wrv != 0) - return wrv; - - update_timer(); + // + // Deliberately no write here -- see defer_write(). + // + defer_write(); return 0; } @@ -1498,10 +1635,9 @@ ngtcp2_ssize Http3Session::write_pkt(ngtcp2_path* path, ngtcp2_pkt_info* pi, uin if (fin) flags |= NGTCP2_WRITE_STREAM_FLAG_FIN; - auto nwrite = - ngtcp2_conn_writev_stream(conn_, path, pi, dest, destlen, &ndatalen, flags, stream_id, - reinterpret_cast(vec.data()), - static_cast(sveccnt), ts); + auto nwrite = ngtcp2_conn_writev_stream( + conn_, path, pi, dest, destlen, &ndatalen, flags, stream_id, + reinterpret_cast(vec.data()), static_cast(sveccnt), ts); if (nwrite < 0) { @@ -1543,8 +1679,6 @@ ngtcp2_ssize Http3Session::write_pkt(ngtcp2_path* path, ngtcp2_pkt_info* pi, uin loge("[{}] nghttp3_conn_add_write_offset: {}", log_prefix_, nghttp3_strerror(rv)); return NGTCP2_ERR_CALLBACK_FAILURE; } - if (auto s = find_stream(stream_id)) - s->on_write_consumed(static_cast(ndatalen)); } continue; default: @@ -1563,8 +1697,6 @@ ngtcp2_ssize Http3Session::write_pkt(ngtcp2_path* path, ngtcp2_pkt_info* pi, uin loge("[{}] nghttp3_conn_add_write_offset: {}", log_prefix_, nghttp3_strerror(rv)); return NGTCP2_ERR_CALLBACK_FAILURE; } - if (auto s = find_stream(stream_id)) - s->on_write_consumed(static_cast(ndatalen)); } return nwrite; @@ -1585,9 +1717,9 @@ int Http3Session::write_streams() ngtcp2_path_storage_zero(&ps); size_t gso_size = 0; - auto nwrite = ngtcp2_conn_write_aggregate_pkt2(conn_, &ps.path, &pi, tx_buf_.data(), - tx_buf_.size(), &gso_size, &write_pkt_cb, 0, - ngtcp2::util::timestamp()); + auto nwrite = + ngtcp2_conn_write_aggregate_pkt2(conn_, &ps.path, &pi, tx_buf_.data(), tx_buf_.size(), + &gso_size, &write_pkt_cb, 0, ngtcp2::util::timestamp()); if (nwrite < 0) { loge("[{}] ngtcp2_conn_write_aggregate_pkt2: {}", log_prefix_, @@ -1602,12 +1734,28 @@ int Http3Session::write_streams() if (nwrite == 0) return 0; - return send_udp_gso(ep_.fd, ps.path.remote.addr, ps.path.remote.addrlen, + return send_udp_gso(ep_, ps.path.remote.addr, ps.path.remote.addrlen, {tx_buf_.data(), static_cast(nwrite)}, gso_size, no_gso_); } // ------------------------------------------------------------------------------------------------- +int Http3Session::flush_write() +{ + write_pending_ = false; + + if (closed_ || !conn_) + return 0; + + if (auto rv = write_streams(); rv != 0) + return rv; + + update_timer(); + return 0; +} + +// ------------------------------------------------------------------------------------------------- + void Http3Session::update_timer() { arm_timer_from_ngtcp2(); } void Http3Session::arm_timer_from_ngtcp2() @@ -1644,14 +1792,22 @@ int Http3Session::handle_expiry() auto now = ngtcp2::util::timestamp(); if (auto rv = ngtcp2_conn_handle_expiry(conn_, now); rv != 0) { - logw("[{}] ngtcp2_conn_handle_expiry: {}", log_prefix_, ngtcp2_strerror(rv)); + // + // NGTCP2_ERR_IDLE_CLOSE is how a connection whose peer simply stopped talking ends -- + // an interrupted client leaves one behind per connection it had open -- so it is a + // normal end of life, not a failure worth a warning. handle_error() takes it from here + // either way; what makes it special is that it discards the connection silently, see + // there. + // + if (rv == NGTCP2_ERR_IDLE_CLOSE) + logi("[{}] idle timeout, dropping connection", log_prefix_); + else + logw("[{}] ngtcp2_conn_handle_expiry: {}", log_prefix_, ngtcp2_strerror(rv)); + ngtcp2_ccerr_set_liberr(&last_error_, rv, nullptr, 0); return handle_error(rv); } - if (auto rv = write_streams(); rv != 0) - return rv; - update_timer(); - return 0; + return flush_write(); } // ------------------------------------------------------------------------------------------------- @@ -1662,10 +1818,20 @@ int Http3Session::handle_error(int /*rv*/) return -1; closed_ = true; - // Idle timeout and drop-conn need no CONNECTION_CLOSE packet. + // + // Idle timeout and drop-conn need no CONNECTION_CLOSE packet -- and with no packet there is + // no closing period either, so none of the cleanup in Server::Impl::udp_on_read() can ever + // run for this session: it is reached from the expiry timer precisely because nothing is + // arriving any more. Drop the session from the demux map right here instead, or it would sit + // in m_quic_handlers for the lifetime of the server, holding streams whose request handlers + // are still waiting on a peer that went away. What is left of it then dies with do_session(). + // if (last_error_.type == NGTCP2_CCERR_TYPE_IDLE_CLOSE || last_error_.type == NGTCP2_CCERR_TYPE_DROP_CONN) { + auto self = weak_from_this().lock(); // erase_quic_session() may drop the last reference + timer_.cancel(); + server_.erase_quic_session(this); signal_done(); return -1; } @@ -1686,7 +1852,7 @@ int Http3Session::handle_error(int /*rv*/) { conn_closebuf_.resize(static_cast(nwrite)); logi("[{}] sending CONNECTION_CLOSE", log_prefix_); - send_udp(ep_.fd, ps.path.remote.addr, ps.path.remote.addrlen, + send_udp(ep_, ps.path.remote.addr, ps.path.remote.addrlen, {conn_closebuf_.data(), conn_closebuf_.size()}); } else @@ -1729,7 +1895,7 @@ void Http3Session::resend_conn_close() if (!path) return; logd("[{}] resending CONNECTION_CLOSE", log_prefix_); - send_udp(ep_.fd, path->remote.addr, path->remote.addrlen, + send_udp(ep_, path->remote.addr, path->remote.addrlen, {conn_closebuf_.data(), conn_closebuf_.size()}); } @@ -1908,6 +2074,7 @@ int Http3Session::setup_http3() return 0; nghttp3_callbacks h3cb{}; + h3cb.acked_stream_data = &Http3Session::h3_cb_acked_stream_data; h3cb.stream_close = &Http3Session::h3_cb_stream_close; h3cb.recv_data = &Http3Session::h3_cb_recv_data; h3cb.deferred_consume = &Http3Session::h3_cb_deferred_consume; @@ -1968,6 +2135,29 @@ int Http3Session::setup_http3() // nghttp3 callbacks // ------------------------------------------------------------------------------------------------- +// +// The only notification that the peer is done with response body bytes we handed out by +// reference, and hence that the caller's buffer may be released -- see the comment above +// Http3Stream::write_active. +// +int Http3Session::h3_cb_acked_stream_data(nghttp3_conn*, int64_t stream_id, uint64_t datalen, + void* user, void*) +{ + auto self = static_cast(user); + auto stream = self->find_stream(stream_id); + if (!stream) + return 0; + + // + // Completing a write resumes the application, which may drop the last reference to this + // session -- while ngtcp2 is still in the middle of processing the ACK that got us here. + // weak_from_this(), not shared_from_this(): the ACK may well arrive during teardown. + // + auto session_guard = self->weak_from_this().lock(); + stream->on_write_acked(static_cast(datalen)); + return 0; +} + int Http3Session::h3_cb_stream_close(nghttp3_conn*, int64_t stream_id, uint64_t /*app_error*/, void* user, void*) { @@ -1979,6 +2169,19 @@ int Http3Session::h3_cb_stream_close(nghttp3_conn*, int64_t stream_id, uint64_t // Waiting readers/writers should see the close now. if (s->read_handler) swap_and_invoke(s->read_handler, boost::system::error_code{}, 0); + + // + // 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. + // + if (s->write_active && s->write_handler) + { + s->write_active = false; + s->write_source = {}; + swap_and_invoke(s->write_handler, errc::make_error_code(errc::connection_reset)); + } s->maybe_close(); } if (ngtcp2_conn_is_server(self->conn_)) @@ -2036,7 +2239,8 @@ int Http3Session::h3_cb_recv_header(nghttp3_conn*, int64_t stream_id, int32_t /* if (!s) return 0; - logd("[{}] \x1b[1;34m{}\x1b[0m: {}", s->log_prefix, name_view, value_view); + if (spdlog::default_logger_raw()->should_log(spdlog::level::debug)) + s->received_headers.emplace_back(name_view, value_view); try { @@ -2082,6 +2286,7 @@ int Http3Session::h3_cb_end_headers(nghttp3_conn*, int64_t stream_id, int /*fin* return 0; logd("[{}] {} {}", s->log_prefix, s->method, s->url.buffer()); + log_headers(s->log_prefix, std::exchange(s->received_headers, {})); // // Build user-facing Request/Response and dispatch through the shared handler. @@ -2090,10 +2295,8 @@ int Http3Session::h3_cb_end_headers(nghttp3_conn*, int64_t stream_id, int /*fin* server::Response response(std::make_unique>(*s)); auto& sv = self->server_; - if (auto& handler = sv.requestHandlerCoro()) + if (auto& handler = sv.requestHandler()) co_spawn(self->get_executor(), handler(std::move(request), std::move(response)), detached); - else if (auto& handler = sv.requestHandler()) - handler(std::move(request), std::move(response)); else { loge("[{}] no request handler set", s->log_prefix); @@ -2177,10 +2380,22 @@ int Server::Impl::udp_on_read(Endpoint& ep) msg_ctrl[CMSG_SPACE(sizeof(int)) + CMSG_SPACE(sizeof(in6_pktinfo)) + CMSG_SPACE(sizeof(int))]; msg.msg_control = msg_ctrl; + // + // Sessions that received something in this batch. They are written once, below, after every + // datagram the socket had queued has been handed to ngtcp2 -- so one aggregate pass can pack + // a whole response into a single GSO sendmsg() instead of dribbling it out per datagram. + // + std::vector> pending; + auto mark_pending = [&pending](const std::shared_ptr& session) + { + if (std::ranges::find(pending, session) == pending.end()) + pending.push_back(session); + }; + for (size_t pktcnt = 0; pktcnt < 32; ++pktcnt) { if (pktcnt) - logd("- - {} - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - ", pktcnt); + logd("- - {} - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - ", pktcnt); msg.msg_namelen = sizeof(su); msg.msg_controllen = sizeof(msg_ctrl); @@ -2190,7 +2405,7 @@ int Server::Impl::udp_on_read(Endpoint& ep) { if (errno != EAGAIN && errno != EWOULDBLOCK && errno != ENOTCONN) loge("recvmsg: {}", strerror(errno)); - return 0; + break; // socket drained (or broken): fall through to the write pass } if (nread < 22) @@ -2229,11 +2444,20 @@ int Server::Impl::udp_on_read(Endpoint& ep) auto all_data = std::span{buf.data(), static_cast(nread)}; const size_t seg_size = gro_size > 0 ? gro_size : all_data.size(); - while (!all_data.empty()) + for (size_t segcnt = 0; !all_data.empty(); ++segcnt) { + if (segcnt) + logd("- {} - - - - - - - - - - - - - - -", segcnt); + auto data = all_data.subspan(0, std::min(seg_size, all_data.size())); all_data = all_data.subspan(data.size()); + if (drop_packet(ep.drop_rate_rx)) + { + // logw("*** dropping received packet ({} bytes) ***", data.size()); + continue; + } + ngtcp2_version_cid vc; auto rv = ngtcp2_pkt_decode_version_cid(&vc, data.data(), data.size(), QUIC_SCIDLEN); if (rv != 0) @@ -2283,6 +2507,8 @@ int Server::Impl::udp_on_read(Endpoint& ep) auto lock = std::lock_guard(self->m_sessionMutex); self->m_sessions.erase(session); }); + + mark_pending(session); } else { @@ -2306,7 +2532,11 @@ int Server::Impl::udp_on_read(Endpoint& ep) continue; } - if (session->on_read(pi, data, *remote) != 0 && session->closed()) + if (session->on_read(pi, data, *remote) == 0) + { + mark_pending(session); + } + else if (session->closed()) { // // Only erase immediately when not in closing/draining period. @@ -2324,6 +2554,21 @@ int Server::Impl::udp_on_read(Endpoint& ep) } } } + + // + // One write pass per session, after the whole batch has been read. + // + for (const auto& session : pending) + { + if (session->flush_write() == 0 || !session->closed()) + continue; + + auto* conn = session->conn(); + if (!conn || (!ngtcp2_conn_in_closing_period(conn) && !ngtcp2_conn_in_draining_period(conn))) + std::erase_if(m_quic_handlers, + [&](const auto& kv) { return kv.second.get() == session.get(); }); + } + return 0; } @@ -2347,6 +2592,8 @@ awaitable Server::Impl::udp_receive_loop() Endpoint ep{}; ep.fd = m_udp_socket->native_handle(); + ep.drop_rate_rx = m_config.drop_rate_rx; + ep.drop_rate_tx = m_config.drop_rate_tx; auto local = m_udp_socket->local_endpoint(); auto data = local.data(); std::memcpy(&ep.addr.su, data, local.size()); diff --git a/src/server_main.cpp b/src/server_main.cpp index 5de244e..fc01311 100644 --- a/src/server_main.cpp +++ b/src/server_main.cpp @@ -1,3 +1,4 @@ +#include "anyhttp/file_handler.hpp" #include "anyhttp/request_handlers.hpp" #include "anyhttp/server.hpp" #include "anyhttp/utils.hpp" @@ -44,6 +45,10 @@ std::expected parseConfig(int argc, char* argv[]) 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"); + opts("drop-rx", po::value(&config.server.drop_rate_rx)->default_value(0.0), + "HTTP/3 testing: probability (0.0 .. 1.0) of dropping a received QUIC packet"); + opts("drop-tx", po::value(&config.server.drop_rate_tx)->default_value(0.0), + "HTTP/3 testing: probability (0.0 .. 1.0) of dropping a QUIC packet before sending it"); po::variables_map vm; try @@ -70,6 +75,16 @@ std::expected parseConfig(int argc, char* argv[]) return std::unexpected(1); } + for (auto [name, rate] : {std::pair{"drop-rx", config.server.drop_rate_rx}, + std::pair{"drop-tx", config.server.drop_rate_tx}}) + { + if (rate < 0.0 || rate > 1.0) + { + std::println(std::cerr, "--{} must be between 0.0 and 1.0", name); + return std::unexpected(1); + } + } + return {std::move(config)}; } @@ -97,7 +112,7 @@ int main(int argc, char* argv[]) server.reset(); }); - server->setRequestHandlerCoro( + server->setRequestHandler( [](server::Request request, server::Response response) -> awaitable { std::string path = request.url().path(); @@ -111,6 +126,8 @@ int main(int argc, char* argv[]) co_await dump(std::move(request), std::move(response)); else if (path == "/discard") co_return; + else if (path == "/test" || path.starts_with("/test/")) + co_await serve_file(std::move(request), std::move(response), "test", "/test"); else if (path == "/eat_request") co_await eat_request(std::move(request), std::move(response)); else if (path == "/" || path == "/h2spec") diff --git a/test/test_server.cpp b/test/test_server.cpp index 5a12afd..60bb8fc 100644 --- a/test/test_server.cpp +++ b/test/test_server.cpp @@ -1,4 +1,5 @@ #include "anyhttp/client.hpp" +#include "anyhttp/file_handler.hpp" #include "anyhttp/formatter.hpp" // IWYU pragma: keep #include "anyhttp/request_handlers.hpp" #include "anyhttp/server.hpp" @@ -42,6 +43,8 @@ #include +#include + #include #include @@ -183,7 +186,7 @@ class Server : public testing::TestWithParam // strand is created after accepting a new connection. // server.emplace(context.get_executor(), config); - server->setRequestHandlerCoro( + server->setRequestHandler( [this](server::Request request, server::Response response) -> awaitable { logd("{} ({})", request.url().path(), request.url().buffer()); @@ -218,7 +221,7 @@ class Server : public testing::TestWithParam co_await dump(std::move(request), std::move(response)); else if (request.url().path() == "/detach") co_await detach(std::move(request), std::move(response)); - else if (request.url().path() == "/custom") + else if (request.url().path().starts_with("/custom")) co_await custom(std::move(request), std::move(response)); else co_await not_found(std::move(request), std::move(response)); @@ -958,6 +961,302 @@ TEST_P(ClientAsync, HelloWorld) // ------------------------------------------------------------------------------------------------- +// +// A single async_write() larger than what the transport hands to its peer in one go, i.e. the +// whole body in one call instead of chunk by chunk. HTTP/3 has to keep offering the same buffer +// to nghttp3 across many packets here, and complete the write only once all of it is +// acknowledged. +// +TEST_P(ClientAsync, WHEN_server_writes_large_buffer_at_once_THEN_receives_all) +{ + static const std::vector body = [] + { + std::vector data(256 * 1024); + 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 + + co_await response.async_submit(200, fields({{"Content-Length", body.size()}})); + co_await response.async_write(asio::buffer(body)); + co_await response.async_write({}); + }; + 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()); + }; +} + +// +// Cancelling a response write mid-body. HTTP/3 hands the caller's buffer to nghttp3 by reference, +// so bytes already offered and not yet acknowledged cannot simply be abandoned -- the stream is +// reset instead, which is what the peer would see anyway for a body that stops short of its end. +// +TEST_P(ClientAsync, WHEN_server_cancels_write_THEN_client_sees_truncated_body) +{ + static const std::vector body(8 * 1024 * 1024, '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 + + 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 executor = co_await this_coro::executor; + auto [ep] = co_await co_spawn(executor, send(response, std::span(body)), + cancel_after(50ms, as_tuple)); + EXPECT_EQ(code(ep), boost::system::errc::operation_canceled); + }; + test = [this](Session session) -> awaitable + { + auto request = co_await session.async_submit(url); + co_await request.async_write({}); + auto response = co_await request.async_get_response(); + + // + // Leave the response body untouched for now: whatever the server manages to send fills up + // the receive window and stays there, so its write cannot run to completion before the + // cancellation above hits. + // + 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); + std::println("received {} of {} bytes ({})", received, body.size(), ec.message()); + EXPECT_LT(received, body.size()); + EXPECT_EQ(ec, boost::beast::http::error::partial_message); + }; +} + +// ================================================================================================= + +// +// serve_file() mounted on "/custom", serving a directory tree created fresh for each testcase. +// +class FileHandler : public ClientAsync +{ +public: + void SetUp() override + { + base = std::filesystem::temp_directory_path() / + std::format("anyhttp-file-handler-{}", ::getpid()); + root = base / "docroot"; + std::filesystem::remove_all(base); + std::filesystem::create_directories(root / "sub"); + + 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 / "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); + std::filesystem::create_symlink(base / "outside.txt", root / "escape.txt"); + + // just outside the root, reachable only by escaping it -- so a missing check shows up as + // content served instead of a 404 + write(base / "outside.txt", "outside"); + + ClientAsync::SetUp(); + + custom = [this](server::Request request, server::Response response) -> awaitable { + co_await serve_file(std::move(request), std::move(response), root, "/custom"); + }; + } + + void TearDown() override + { + ClientAsync::TearDown(); + std::filesystem::remove_all(base); + } + + static void write(const std::filesystem::path& path, std::string_view content) + { + std::ofstream(path, std::ios::binary).write(content.data(), content.size()); + } + + // + // Requests \p target and returns status code and body. The request is finished right away -- + // serve_file() ignores the request body, but still has to consume it. + // + awaitable> get(Session& session, boost::urls::url target) + { + auto request = co_await session.async_submit(target, {}); + co_await request.async_write({}); + 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)); + } + + awaitable> get(Session& session, std::string_view path) + { + co_return co_await get(session, boost::urls::url(url).set_path(path)); + } + + /// Target with a percent-encoded path, passed to the server as-is. + boost::urls::url encoded(std::string_view path) const + { + auto target = boost::urls::url(url); + target.set_encoded_path(path); + return target; + } + +protected: + std::filesystem::path base; ///< holds the docroot and the file just outside of it + std::filesystem::path root; ///< what serve_file() is mounted on +}; + +INSTANTIATE_TEST_SUITE_P(FileHandler, FileHandler, + ::testing::Values(anyhttp::Protocol::http11, anyhttp::Protocol::h2, + anyhttp::Protocol::h3), + NameGenerator); + +// ------------------------------------------------------------------------------------------------- + +TEST_P(FileHandler, WHEN_file_exists_THEN_serves_content) +{ + test = [this](Session session) -> awaitable + { + auto [status, body] = co_await get(session, "/custom/hello.txt"); + EXPECT_EQ(status, 200); + EXPECT_EQ(body, "Hello, File!"); + }; +} + +TEST_P(FileHandler, WHEN_file_is_in_subdirectory_THEN_serves_content) +{ + test = [this](Session session) -> awaitable + { + auto [status, body] = co_await get(session, "/custom/sub/nested.txt"); + EXPECT_EQ(status, 200); + EXPECT_EQ(body, "Nested!"); + }; +} + +// +// An empty file cannot be mmap()ed at all, and must not send the empty buffer that already means +// EOF twice. +// +TEST_P(FileHandler, WHEN_file_is_empty_THEN_serves_empty_body) +{ + test = [this](Session session) -> awaitable + { + auto [status, body] = co_await get(session, "/custom/empty.txt"); + EXPECT_EQ(status, 200); + EXPECT_EQ(body, ""); + }; +} + +// +// Large enough that a single async_write() spans many QUIC packets, so the HTTP/3 write path has +// to keep handing out the caller's buffer across several write_pkt() rounds. +// +TEST_P(FileHandler, WHEN_file_is_large_THEN_serves_all_of_it) +{ + test = [this](Session session) -> awaitable + { + auto [status, body] = co_await get(session, "/custom/large.bin"); + EXPECT_EQ(status, 200); + EXPECT_EQ(body, std::string(256 * 1024, 'x')); + }; +} + +TEST_P(FileHandler, WHEN_file_does_not_exist_THEN_error_404) +{ + test = [this](Session session) -> awaitable + { + auto [status, body] = co_await get(session, "/custom/missing.txt"); + EXPECT_EQ(status, 404); + EXPECT_EQ(body, ""); + }; +} + +// +// A directory can be open()ed but not mapped, and we do not serve listings. +// +TEST_P(FileHandler, WHEN_path_is_a_directory_THEN_error_404) +{ + test = [this](Session session) -> awaitable + { + EXPECT_EQ(std::get<0>(co_await get(session, "/custom/sub")), 404); + EXPECT_EQ(std::get<0>(co_await get(session, "/custom/")), 404); + }; +} + +TEST_P(FileHandler, WHEN_path_escapes_the_root_THEN_error_404) +{ + test = [this](Session session) -> awaitable + { + EXPECT_EQ(std::get<0>(co_await get(session, "/custom/../outside.txt")), 404); + EXPECT_EQ(std::get<0>(co_await get(session, "/custom/sub/../../outside.txt")), 404); + EXPECT_EQ(std::get<0>(co_await get(session, encoded("/custom/%2e%2e/outside.txt"))), 404); + }; +} + +// +// weakly_canonical() resolves the link, so a link out of the root is caught like any other escape. +// +TEST_P(FileHandler, WHEN_symlink_points_outside_the_root_THEN_error_404) +{ + test = [this](Session session) -> awaitable + { + EXPECT_EQ(std::get<0>(co_await get(session, "/custom/escape.txt")), 404); + }; +} + +// +// The mount prefix must match whole path segments, not just any leading characters: stripping +// "/custom" off "/customer.txt" would otherwise serve "er.txt" out of the docroot. +// +TEST_P(FileHandler, WHEN_prefix_matches_mid_segment_THEN_error_404) +{ + test = [this](Session session) -> awaitable + { + auto [status, body] = co_await get(session, "/customer.txt"); + EXPECT_EQ(status, 404); + EXPECT_EQ(body, ""); + EXPECT_EQ(std::get<0>(co_await get(session, "/customer/hello.txt")), 404); + }; +} + +TEST_P(FileHandler, WHEN_file_is_not_readable_THEN_error_403) +{ + if (::geteuid() == 0) + GTEST_SKIP() << "running as root, permissions do not apply"; + + test = [this](Session session) -> awaitable + { + auto [status, body] = co_await get(session, "/custom/secret.txt"); + EXPECT_EQ(status, 403); + EXPECT_EQ(body, ""); + }; +} + +TEST_P(FileHandler, WHEN_same_file_is_requested_twice_THEN_serves_it_twice) +{ + test = [this](Session session) -> awaitable + { + EXPECT_EQ(std::get<1>(co_await get(session, "/custom/hello.txt")), "Hello, File!"); + EXPECT_EQ(std::get<1>(co_await get(session, "/custom/hello.txt")), "Hello, File!"); + }; +} + +// ------------------------------------------------------------------------------------------------- + TEST_P(ClientAsync, ServerYieldFirst) { custom = [this](server::Request request, server::Response response) -> awaitable @@ -1478,3 +1777,113 @@ TEST_P(ClientAsync, DISABLED_SpawnAndForget) } // ================================================================================================= + +// +// A QUIC peer that goes away without a word -- a killed client, a machine that went to sleep in +// the middle of a request -- leaves the server nothing to react to: no CONNECTION_CLOSE arrives, +// and no further packet ever will. Only the idle timer can notice, and dropping the connection +// when it fires is what releases the session, its streams, and the request handlers suspended on +// them. +// +// Note that this is *not* what a client calling Session::reset() looks like: that one says +// goodbye, and the server cleans up right away by way of the draining period. +// +class Http3IdleTimeout : public testing::Test +{ +protected: + static constexpr auto IdleTimeout = 500ms; + + void SetUp() override + { + setupLogging(); + + server.emplace(context.get_executor(), server::Config{.listen_address = "127.0.0.2", + .port = 0, + .idle_timeout = IdleTimeout}); + server->setRequestHandler( + [this](server::Request request, server::Response response) -> awaitable + { + co_await response.async_submit(200, {}); + + // + // Wait for a request body that never comes: this first read is where the handler is + // suspended when the client freezes, and it must be resumed -- with an error -- once + // the server gives up on the connection. + // + std::array buffer; + auto [ec, n] = co_await request.async_read_some(asio::buffer(buffer), as_tuple); + handler_result.set_value(ec); + }); + + url.set_port_number(server->local_endpoint().port()); + } + + asio::io_context context; + std::optional server; + + std::promise handler_result; + boost::urls::url url{"http://127.0.0.2/echo"}; +}; + +// ------------------------------------------------------------------------------------------------- + +TEST_F(Http3IdleTimeout, WHEN_client_vanishes_in_flight_THEN_idle_timer_drops_the_session) +{ + auto result = handler_result.get_future(); + + // + // The server has to keep running while the client is frozen, so it gets a thread of its own. + // + std::jthread server_thread([this] { context.run(); }); + boost::scope::scope_exit stop_server([this] { context.stop(); }); + + // + // The client runs on its own io_context, which is what makes freezing it possible: stopping + // that context takes the client off the air mid-request without unwinding anything, so no + // CONNECTION_CLOSE is ever sent -- just like a client process that was killed. + // + asio::io_context client_context; + client::Client client(client_context.get_executor(), + client::Config{.url = url, .protocol = anyhttp::Protocol::h3}); + + // + // Session, request and response are kept out here rather than in the coroutine frame, which is + // destroyed as soon as the coroutine below returns: unwinding them would reset the stream, and + // that is a packet -- the one thing this client must not send. + // + std::optional session; + std::optional request; + std::optional response; + + bool responded = false; + co_spawn(client_context, [&]() -> awaitable + { + session = co_await client.async_connect(); + request = co_await session->async_submit(url, {}); + response = co_await request->async_get_response(); + responded = true; + }, detached); + + // + // Run the client just far enough to have the request open and answered, then stop running it: + // from here on it never touches its socket again. + // + while (client_context.run_one() && !responded); + ASSERT_TRUE(responded) << "client never received a response"; + std::println("=== freezing the client, request still in flight ==="); + + // + // From here on the server is on its own. Without the idle timer dropping the connection, the + // request handler stays suspended in async_read_some() forever, and the session it belongs to + // sits in the server's connection table for good. + // + ASSERT_EQ(result.wait_for(5s), std::future_status::ready) << "request handler never completed"; + EXPECT_EQ(result.get(), boost::system::errc::connection_reset); + + // + // Only now, on the way out, is the frozen client allowed to unwind: doing so earlier would + // have sent the CONNECTION_CLOSE that this test is all about not sending. + // +} + +// =================================================================================================