From b15ecfd25f4ca7feb2a45ebe1a375b3b442a9aa4 Mon Sep 17 00:00:00 2001 From: Steve Gerbino Date: Fri, 30 Oct 2026 19:30:03 +0100 Subject: [PATCH 1/2] refactor!: own buffer_param and fix content encoding never finishing Capy removed buffer_param and buffer_slice and made consuming_buffers private. Add http::buffer_param, a copy of capy's, and use it in the type-erased sources and sinks, parser::read, and the content coding filter. Replace the parser's buffer_slice with buffer_copy's at_most argument, and adopt capy's renamed buffer functions. The filter now skips empty input buffers. The serializer's final call passes two empty buffers, which used to count as more input, so the encoder never received finish and a gzip encoded message never completed. buffer_param exposes one window of buffers at a time, so the filter counts the remaining bytes itself and parser::read keeps copying until the body is drained. Add tests for gzip encoding and for reading into many buffers. --- include/boost/http/concept/read_source.hpp | 2 +- include/boost/http/io/any_buffer_sink.hpp | 10 +- include/boost/http/io/any_buffer_source.hpp | 6 +- include/boost/http/io/any_read_source.hpp | 4 +- include/boost/http/io/any_write_sink.hpp | 8 +- include/boost/http/io/buffer_param.hpp | 268 +++++++++++++ include/boost/http/json/json_sink.hpp | 4 +- include/boost/http/parser.hpp | 14 +- include/boost/http/serializer.hpp | 16 +- include/boost/http/test/read_source.hpp | 4 +- include/boost/http/test/write_sink.hpp | 2 +- src/detail/buffer_utils.hpp | 8 +- src/detail/filter.cpp | 26 +- src/parser.cpp | 9 +- test/unit/io/any_buffer_sink.cpp | 2 +- test/unit/io/any_buffer_source.cpp | 4 +- test/unit/io/buffer_param.cpp | 401 ++++++++++++++++++++ test/unit/parser.cpp | 38 ++ test/unit/serializer.cpp | 91 ++++- 19 files changed, 856 insertions(+), 61 deletions(-) create mode 100644 include/boost/http/io/buffer_param.hpp create mode 100644 test/unit/io/buffer_param.cpp diff --git a/include/boost/http/concept/read_source.hpp b/include/boost/http/concept/read_source.hpp index 910203d7..c2eaf4ce 100644 --- a/include/boost/http/concept/read_source.hpp +++ b/include/boost/http/concept/read_source.hpp @@ -63,7 +63,7 @@ namespace http { Successful partial reads are not permitted; either the entire buffer is filled or the operation returns with an error. - If `buffer_empty( buffers )` is `true`, the operation completes + If `buffers_empty( buffers )` is `true`, the operation completes immediately with `!ec` and `n` equal to 0. When the buffer sequence contains multiple buffers, each buffer is diff --git a/include/boost/http/io/any_buffer_sink.hpp b/include/boost/http/io/any_buffer_sink.hpp index 685edf7b..2427dde8 100644 --- a/include/boost/http/io/any_buffer_sink.hpp +++ b/include/boost/http/io/any_buffer_sink.hpp @@ -14,7 +14,7 @@ #include #include #include -#include +#include #include #include #include @@ -1042,7 +1042,7 @@ template capy::task> any_buffer_sink::write_some(CB buffers) { - capy::buffer_param bp(buffers); + buffer_param bp(buffers); auto src = bp.data(); if(src.empty()) co_return {std::error_code(), 0}; @@ -1075,7 +1075,7 @@ template capy::task> any_buffer_sink::write(CB buffers) { - capy::buffer_param bp(buffers); + buffer_param bp(buffers); std::size_t total = 0; // Native WriteSink path @@ -1186,7 +1186,7 @@ any_buffer_sink::write_eof(CB buffers) // Native WriteSink path if(vt_->construct_write_eof_buffers_awaitable) { - capy::const_buffer_param bp(buffers); + const_buffer_param bp(buffers); std::size_t total = 0; for(;;) @@ -1215,7 +1215,7 @@ any_buffer_sink::write_eof(CB buffers) } // Synthesized path: prepare + capy::buffer_copy + commit + commit_eof - capy::buffer_param bp(buffers); + buffer_param bp(buffers); std::size_t total = 0; for(;;) diff --git a/include/boost/http/io/any_buffer_source.hpp b/include/boost/http/io/any_buffer_source.hpp index 31a91f55..6cee1b39 100644 --- a/include/boost/http/io/any_buffer_source.hpp +++ b/include/boost/http/io/any_buffer_source.hpp @@ -14,7 +14,7 @@ #include #include #include -#include +#include #include #include #include @@ -758,7 +758,7 @@ template capy::task> any_buffer_source::read_some(MB buffers) { - capy::buffer_param bp(buffers); + buffer_param bp(buffers); auto dest = bp.data(); if(dest.empty()) co_return {std::error_code(), 0}; @@ -782,7 +782,7 @@ template capy::task> any_buffer_source::read(MB buffers) { - capy::buffer_param bp(buffers); + buffer_param bp(buffers); std::size_t total = 0; // Native ReadSource path diff --git a/include/boost/http/io/any_read_source.hpp b/include/boost/http/io/any_read_source.hpp index 9d613f9c..e792b30b 100644 --- a/include/boost/http/io/any_read_source.hpp +++ b/include/boost/http/io/any_read_source.hpp @@ -14,7 +14,7 @@ #include #include #include -#include +#include #include #include #include @@ -572,7 +572,7 @@ template capy::task> any_read_source::read(MB buffers) { - capy::buffer_param bp(buffers); + buffer_param bp(buffers); std::size_t total = 0; for(;;) diff --git a/include/boost/http/io/any_write_sink.hpp b/include/boost/http/io/any_write_sink.hpp index c5c6a81e..429d8d62 100644 --- a/include/boost/http/io/any_write_sink.hpp +++ b/include/boost/http/io/any_write_sink.hpp @@ -14,7 +14,7 @@ #include #include #include -#include +#include #include #include #include @@ -222,7 +222,7 @@ class any_write_sink Writes data from the provided buffer sequence. The operation completes when all bytes have been consumed, or an error occurs. Forwards to the underlying sink's `write` operation, - windowed through @ref capy::buffer_param when the sequence exceeds + windowed through @ref buffer_param when the sequence exceeds the per-call buffer limit. @param buffers The buffer sequence containing data to write. @@ -854,7 +854,7 @@ template capy::task> any_write_sink::write(CB buffers) { - capy::const_buffer_param bp(buffers); + const_buffer_param bp(buffers); std::size_t total = 0; for(;;) @@ -877,7 +877,7 @@ template capy::task> any_write_sink::write_eof(CB buffers) { - capy::const_buffer_param bp(buffers); + const_buffer_param bp(buffers); std::size_t total = 0; for(;;) diff --git a/include/boost/http/io/buffer_param.hpp b/include/boost/http/io/buffer_param.hpp new file mode 100644 index 00000000..1c5de9a7 --- /dev/null +++ b/include/boost/http/io/buffer_param.hpp @@ -0,0 +1,268 @@ +// +// Copyright (c) 2025 Vinnie Falco (vinnie.falco@gmail.com) +// Copyright (c) 2026 Michael Vandeberg +// Copyright (c) 2026 Steve Gerbino +// +// Distributed under the Boost Software License, Version 1.0. (See accompanying +// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt) +// +// Official repository: https://github.com/cppalliance/http +// + +/* + COROUTINE BUFFER SEQUENCE LIFETIME REQUIREMENT + =============================================== + Buffer sequence parameters in coroutine APIs MUST be passed BY VALUE, + never by reference. When a coroutine suspends, reference parameters may + dangle if the caller's object goes out of scope before resumption. + + CORRECT: task<> read_some(MutableBufferSequence auto buffers) + WRONG: task<> read_some(MutableBufferSequence auto& buffers) + WRONG: task<> read_some(MutableBufferSequence auto const& buffers) + + The buffer_param class works with this model: it takes a const& in its + constructor (for the non-coroutine scope) but the caller's template + function accepts the buffer sequence by value, ensuring the sequence + lives in the coroutine frame. +*/ + +#ifndef BOOST_HTTP_IO_BUFFER_PARAM_HPP +#define BOOST_HTTP_IO_BUFFER_PARAM_HPP + +#include +#include + +#include +#include +#include + +namespace boost { +namespace http { + +/** A buffer sequence wrapper providing windowed access. + + This template class wraps any buffer sequence and provides + incremental access through a sliding window of buffer + descriptors. It handles both const and mutable buffer + sequences automatically. + + @par Coroutine Lifetime Requirement + + When used in coroutine APIs, the outer template function + MUST accept the buffer sequence parameter BY VALUE: + + @code + // CORRECT + capy::task<> write(capy::ConstBufferSequence auto buffers); + + // WRONG - dangling reference + capy::task<> write(capy::ConstBufferSequence auto& buffers); + @endcode + + Pass-by-value ensures the buffer sequence is copied into + the coroutine frame and remains valid across suspension + points. References would dangle when the caller's scope + exits before the coroutine resumes. + + @par Purpose + + When iterating through large buffer sequences, it is often + more efficient to process buffers in batches rather than + one at a time. This class maintains a window of up to a + fixed, implementation-defined number of buffer descriptors + (currently 16). It refills the window from the underlying + sequence as buffers are consumed. + + @par Example + + Create a `buffer_param` from any buffer sequence and use + `data()` to get the current window of buffers. After + processing some bytes, call `consume()` to advance through + the sequence. + + @code + capy::task<> send(capy::ConstBufferSequence auto buffers) + { + buffer_param bp(buffers); + while(true) + { + auto bufs = bp.data(); + if(bufs.empty()) + break; + auto n = co_await do_something(bufs); + bp.consume(n); + } + } + @endcode + + @par Virtual Interface Pattern + + This class enables passing arbitrary buffer sequences through + a virtual function boundary. The template function captures + the buffer sequence by value and drives the iteration, while + the virtual function receives a simple span. Plain CTAD + (`buffer_param bp(buffers)`) deduces `BS`'s own buffer type, so a + mutable sequence yields `span`. That does not match + `write_impl`'s `span` parameter. Use @ref const_buffer_param + to force `const_buffer` storage regardless of what `BS` is: + + @code + class base + { + public: + template + capy::task<> write(BS buffers) + { + const_buffer_param bp(buffers); + while(true) + { + auto bufs = bp.data(); + if(bufs.empty()) + break; + std::size_t n = 0; + co_await write_impl(bufs, n); + bp.consume(n); + } + } + + protected: + virtual capy::task<> write_impl( + std::span buffers, + std::size_t& bytes_written) = 0; + }; + @endcode + + @tparam BS The buffer sequence type. Must satisfy + `capy::ConstBufferSequence`. +*/ +template + requires capy::ConstBufferSequence +class buffer_param +{ +public: + /// Names `const_buffer` when `MakeConst`, else `BS`'s own buffer type. + using buffer_type = std::conditional_t< + MakeConst, + capy::const_buffer, + capy::buffer_type>; + +private: + decltype(capy::buffer_sequence_begin(std::declval())) it_; + decltype(capy::buffer_sequence_end(std::declval())) end_; + union { + int dummy_; + buffer_type arr_[capy::detail::max_iovec_]; + }; + std::size_t size_ = 0; + std::size_t pos_ = 0; + + void + refill() + { + pos_ = 0; + size_ = 0; + for(; it_ != end_ && size_ < capy::detail::max_iovec_; ++it_) + { + buffer_type buf(*it_); + if(buf.size() > 0) + ::new(&arr_[size_++]) buffer_type(buf); + } + } + +public: + /** Construct from a buffer sequence. + + @param bs The buffer sequence to wrap. The caller must + ensure the buffer sequence remains valid for the + lifetime of this object. + */ + explicit + buffer_param(BS const& bs) + : it_(capy::buffer_sequence_begin(bs)) + , end_(capy::buffer_sequence_end(bs)) + , dummy_(0) + { + refill(); + } + + /** Return the current window of buffer descriptors. + + Returns a span of buffer descriptors representing the + currently available portion of the buffer sequence. + The span contains at most a fixed, implementation-defined + number of buffers (currently 16). + + When the current window is exhausted, this function + automatically refills from the underlying sequence. + + @return A span of buffer descriptors. Empty span + indicates no more data is available. + */ + std::span + data() + { + if(pos_ >= size_) + refill(); + if(size_ == 0) + return {}; + return {arr_ + pos_, size_ - pos_}; + } + + /** Check if more buffers exist beyond the current window. + + Returns `true` if the underlying buffer sequence has + additional buffers that have not yet been loaded into + the current window. Call after @ref data to determine + whether the current window is the last one. + + @return `true` if more buffers remain in the sequence. + */ + bool + more() const noexcept + { + return it_ != end_; + } + + /** Consume bytes from the buffer sequence. + + Advances the current position by `n` bytes, consuming + data from the front of the sequence. Partially consumed + buffers are adjusted in place. + + @param n Number of bytes to consume. + */ + void + consume(std::size_t n) + { + while(n > 0 && pos_ < size_) + { + auto avail = arr_[pos_].size(); + if(n < avail) + { + arr_[pos_] += n; + n = 0; + } + else + { + n -= avail; + ++pos_; + } + } + } +}; + +/** Deduce the sequence type from the constructor argument. + + @tparam BS The buffer sequence type. +*/ +template +buffer_param(BS const&) -> buffer_param; + +/// Forces `buffer_param` to store windows as `const_buffer`, whatever `BS` is. +template +using const_buffer_param = buffer_param; + +} // namespace http +} // namespace boost + +#endif diff --git a/include/boost/http/json/json_sink.hpp b/include/boost/http/json/json_sink.hpp index 40664833..91df6c7f 100644 --- a/include/boost/http/json/json_sink.hpp +++ b/include/boost/http/json/json_sink.hpp @@ -62,8 +62,8 @@ class json_sink { std::error_code ec; std::size_t total = 0; - auto const end = capy::end(buffers); - for(auto it = capy::begin(buffers); it != end; ++it) + auto const end = capy::buffer_sequence_end(buffers); + for(auto it = capy::buffer_sequence_begin(buffers); it != end; ++it) { capy::const_buffer buf(*it); auto n = parser_.write( diff --git a/include/boost/http/parser.hpp b/include/boost/http/parser.hpp index ce99062e..2edf97b0 100644 --- a/include/boost/http/parser.hpp +++ b/include/boost/http/parser.hpp @@ -17,9 +17,9 @@ #include #include -#include #include #include +#include #include #include #include @@ -601,11 +601,11 @@ capy::task> parser:: read(Stream& stream, MB buffers) { - if(capy::buffer_empty(buffers)) + if(capy::buffers_empty(buffers)) co_return {std::error_code(), 0}; std::size_t total = 0; - capy::consuming_buffers dest(buffers); + buffer_param dest(buffers); for(;;) { @@ -614,15 +614,17 @@ read(Stream& stream, MB buffers) if(got_header()) { - auto body_data = pull_body(); - if(capy::buffer_size(body_data) > 0) + // dest.data() is one window, so drain the body over several copies + for(auto body_data = pull_body(); + capy::buffer_size(body_data) > 0; + body_data = pull_body()) { std::size_t copied = capy::buffer_copy(dest.data(), body_data); consume_body(copied); total += copied; dest.consume(copied); - if(capy::buffer_empty(dest.data())) + if(capy::buffers_empty(dest.data())) co_return {std::error_code(), total}; } diff --git a/include/boost/http/serializer.hpp b/include/boost/http/serializer.hpp index d8326689..e2ddc2cc 100644 --- a/include/boost/http/serializer.hpp +++ b/include/boost/http/serializer.hpp @@ -663,7 +663,7 @@ class serializer::sink co_return {std::error_code(cbs.error())}; } - if(capy::buffer_empty(*cbs)) + if(capy::buffers_empty(*cbs)) { // advance state machine sr_->consume(0); @@ -713,7 +713,7 @@ class serializer::sink co_return {std::error_code(cbs.error())}; } - if(capy::buffer_empty(*cbs)) + if(capy::buffers_empty(*cbs)) { // advance state machine sr_->consume(0); @@ -760,7 +760,7 @@ class serializer::sink co_return {cbs.error(), 0}; } - if(capy::buffer_empty(*cbs)) + if(capy::buffers_empty(*cbs)) { // advance state machine sr_->consume(0); @@ -776,8 +776,8 @@ class serializer::sink // Write body data through stream_prepare/commit std::size_t total = 0; - for(auto it = capy::begin(buffers); - it != capy::end(buffers); ++it) + for(auto it = capy::buffer_sequence_begin(buffers); + it != capy::buffer_sequence_end(buffers); ++it) { capy::const_buffer src = *it; while(src.size() != 0) @@ -809,7 +809,7 @@ class serializer::sink co_return {cbs.error(), total}; } - if(capy::buffer_empty(*cbs)) + if(capy::buffers_empty(*cbs)) { // advance state machine sr_->consume(0); @@ -864,7 +864,7 @@ class serializer::sink co_return {cbs.error(), n}; } - if(capy::buffer_empty(*cbs)) + if(capy::buffers_empty(*cbs)) { // advance state machine sr_->consume(0); @@ -910,7 +910,7 @@ class serializer::sink co_return {std::error_code(cbs.error())}; } - if(capy::buffer_empty(*cbs)) + if(capy::buffers_empty(*cbs)) { // advance state machine sr_->consume(0); diff --git a/include/boost/http/test/read_source.hpp b/include/boost/http/test/read_source.hpp index 36e5de7c..86113294 100644 --- a/include/boost/http/test/read_source.hpp +++ b/include/boost/http/test/read_source.hpp @@ -162,7 +162,7 @@ class read_source capy::results await_resume() { - if(capy::buffer_empty(buffers_)) + if(capy::buffers_empty(buffers_)) return {std::error_code(), 0}; if(canceled_) @@ -237,7 +237,7 @@ class read_source capy::results await_resume() { - if(capy::buffer_empty(buffers_)) + if(capy::buffers_empty(buffers_)) return {std::error_code(), 0}; if(canceled_) diff --git a/include/boost/http/test/write_sink.hpp b/include/boost/http/test/write_sink.hpp index eb355768..5616c515 100644 --- a/include/boost/http/test/write_sink.hpp +++ b/include/boost/http/test/write_sink.hpp @@ -195,7 +195,7 @@ class write_sink capy::results await_resume() { - if(capy::buffer_empty(buffers_)) + if(capy::buffers_empty(buffers_)) return {std::error_code(), 0}; if(canceled_) diff --git a/src/detail/buffer_utils.hpp b/src/detail/buffer_utils.hpp index da8b3616..cfabce5b 100644 --- a/src/detail/buffer_utils.hpp +++ b/src/detail/buffer_utils.hpp @@ -21,10 +21,12 @@ namespace detail { template auto make_span(BufferSequence const& bs) -> - boost::span::value_type const> + boost::span::value_type const> { - return { &*capy::begin(bs), - std::size_t(std::distance(capy::begin(bs), capy::end(bs))) }; + return { &*capy::buffer_sequence_begin(bs), + std::size_t(std::distance( + capy::buffer_sequence_begin(bs), capy::buffer_sequence_end(bs))) }; } } // detail diff --git a/src/detail/filter.cpp b/src/detail/filter.cpp index c5eb99f0..c08cbff8 100644 --- a/src/detail/filter.cpp +++ b/src/detail/filter.cpp @@ -10,8 +10,8 @@ #include "src/detail/filter.hpp" -#include -#include +#include +#include namespace boost { namespace http { @@ -22,7 +22,7 @@ namespace { // Returns true if the slice's current data view contains at most one // non-empty buffer (i.e., we are processing the last logical chunk). template -bool single_or_empty(Slice const& s) +bool single_or_empty(Slice& s) { auto d = s.data(); auto it = d.begin(); @@ -42,8 +42,12 @@ process( std::array in_seq, bool more) -> results { - auto out = capy::consuming_buffers(out_seq); - auto in = capy::consuming_buffers(in_seq); + buffer_param out(out_seq); + buffer_param in(in_seq); + + // data() spans only the current window, so track totals separately + std::size_t out_left = capy::buffer_size(out_seq); + std::size_t in_left = capy::buffer_size(in_seq); results rv; bool p_more = true; @@ -51,7 +55,7 @@ process( { if(!more && p_more && single_or_empty(in)) { - if(capy::buffer_size(out.data()) < min_out_buffer()) + if(out_left < min_out_buffer()) { rv.out_short = true; return rv; @@ -59,8 +63,8 @@ process( p_more = false; } - auto ob = capy::front(out.data()); - auto ib = capy::front(in.data()); + auto ob = capy::buffers_front(out.data()); + auto ib = capy::buffers_front(in.data()); auto rs = do_process(ob, ib, p_more); rv.in_bytes += rs.in_bytes; @@ -80,11 +84,13 @@ process( out.consume(rs.out_bytes); in.consume(rs.in_bytes); + out_left -= rs.out_bytes; + in_left -= rs.in_bytes; - if(capy::buffer_size(out.data()) == 0) + if(out_left == 0) return rv; - if(capy::buffer_size(in.data()) == 0 && rs.out_bytes < ob.size()) + if(in_left == 0 && rs.out_bytes < ob.size()) return rv; } } diff --git a/src/parser.cpp b/src/parser.cpp index 467d310d..4147778f 100644 --- a/src/parser.cpp +++ b/src/parser.cpp @@ -20,8 +20,7 @@ #include #include -#include -#include +#include #include #include #include @@ -1123,9 +1122,6 @@ class parser::impl { const std::size_t chunk_avail = clamp(chunk_remain_, cb0_.size()); - auto cb0_data = cb0_.data(); - auto chunk = capy::buffer_slice( - cb0_data, 0, chunk_avail); if(body_limit_remain() < chunk_avail) { @@ -1137,7 +1133,8 @@ class parser::impl // in_place style auto copied = capy::buffer_copy( cb1_.prepare(cb1_.capacity()), - chunk); + cb0_.data(), + chunk_avail); chunk_remain_ -= copied; body_avail_ += copied; body_total_ += copied; diff --git a/test/unit/io/any_buffer_sink.cpp b/test/unit/io/any_buffer_sink.cpp index f12625b6..b428a9e1 100644 --- a/test/unit/io/any_buffer_sink.cpp +++ b/test/unit/io/any_buffer_sink.cpp @@ -167,7 +167,7 @@ class buffer_write_sink await_resume() { self_->write_api_used_ = true; - if(buffer_empty(buffers_)) + if(buffers_empty(buffers_)) return {std::error_code(), 0}; auto ec = self_->f_.maybe_fail(); if(ec) return {ec, 0}; diff --git a/test/unit/io/any_buffer_source.cpp b/test/unit/io/any_buffer_source.cpp index d59d4c4c..0f7cda23 100644 --- a/test/unit/io/any_buffer_source.cpp +++ b/test/unit/io/any_buffer_source.cpp @@ -144,7 +144,7 @@ class buffer_read_source await_resume() { self_->read_api_used_ = true; - if(buffer_empty(buffers_)) + if(buffers_empty(buffers_)) return {std::error_code(), 0}; auto ec = self_->f_.maybe_fail(); if(ec) @@ -182,7 +182,7 @@ class buffer_read_source await_resume() { self_->read_api_used_ = true; - if(buffer_empty(buffers_)) + if(buffers_empty(buffers_)) return {std::error_code(), 0}; auto ec = self_->f_.maybe_fail(); if(ec) diff --git a/test/unit/io/buffer_param.cpp b/test/unit/io/buffer_param.cpp new file mode 100644 index 00000000..cd2a5602 --- /dev/null +++ b/test/unit/io/buffer_param.cpp @@ -0,0 +1,401 @@ +// +// Copyright (c) 2025 Vinnie Falco (vinnie.falco@gmail.com) +// +// Distributed under the Boost Software License, Version 1.0. (See accompanying +// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt) +// +// Official repository: https://github.com/cppalliance/http +// + +// Test that header file is self-contained. +#include + +#include "test_helpers.hpp" + +#include +#include +#include +#include + +namespace boost { +namespace http { + +namespace { + +// Helper to compute total bytes in a span +std::size_t +total_size(std::span bufs) +{ + std::size_t n = 0; + for(auto const& b : bufs) + n += b.size(); + return n; +} + +std::size_t +total_size(std::span bufs) +{ + std::size_t n = 0; + for(auto const& b : bufs) + n += b.size(); + return n; +} + +// Helper to copy data from buffer span to string +std::string +to_string(std::span bufs) +{ + std::string result; + for(auto const& b : bufs) + result.append( + static_cast(b.data()), + b.size()); + return result; +} + +// Helper to consume all data from a buffer_param +template +std::string +consume_all(buffer_param& bp) +{ + std::string result; + while(true) + { + auto bufs = bp.data(); + if(bufs.empty()) + break; + result += to_string(bufs); + bp.consume(total_size(bufs)); + } + return result; +} + +} // namespace + +struct buffer_param_test +{ + void + testConstSingleBuffer() + { + // Single capy::const_buffer + { + std::string data = "Hello, World!"; + capy::const_buffer buf(data.data(), data.size()); + + buffer_param bp(buf); + auto result = consume_all(bp); + BOOST_TEST_EQ(result, data); + } + + // Empty buffer + { + capy::const_buffer buf; + buffer_param bp(buf); + auto result = consume_all(bp); + BOOST_TEST(result.empty()); + } + } + + void + testConstMultipleBuffers() + { + // Vector of buffers + { + std::string s1 = "Hello, "; + std::string s2 = "World"; + std::string s3 = "!"; + + std::vector bufs = { + capy::const_buffer(s1.data(), s1.size()), + capy::const_buffer(s2.data(), s2.size()), + capy::const_buffer(s3.data(), s3.size()) + }; + + buffer_param bp(bufs); + auto result = consume_all(bp); + BOOST_TEST_EQ(result, "Hello, World!"); + } + + // With empty buffers interspersed (should be skipped) + { + std::string s1 = "AB"; + std::string s2 = "CD"; + + std::vector bufs = { + capy::const_buffer(), + capy::const_buffer(s1.data(), s1.size()), + capy::const_buffer(), + capy::const_buffer(s2.data(), s2.size()), + capy::const_buffer() + }; + + buffer_param bp(bufs); + auto result = consume_all(bp); + BOOST_TEST_EQ(result, "ABCD"); + } + } + + void + testConstPartialConsume() + { + std::string data = "0123456789"; + capy::const_buffer buf(data.data(), data.size()); + + // Consume 3 bytes at a time + { + buffer_param bp(buf); + + auto bufs = bp.data(); + BOOST_TEST_EQ(total_size(bufs), 10u); + + // Consume 3 + bp.consume(3); + bufs = bp.data(); + BOOST_TEST_EQ(total_size(bufs), 7u); + BOOST_TEST_EQ( + static_cast(bufs[0].data())[0], + '3'); + + // Consume 4 more + bp.consume(4); + bufs = bp.data(); + BOOST_TEST_EQ(total_size(bufs), 3u); + BOOST_TEST_EQ( + static_cast(bufs[0].data())[0], + '7'); + + // Consume remaining + bp.consume(3); + bufs = bp.data(); + BOOST_TEST(bufs.empty()); + } + } + + void + testConstLargeSequence() + { + // More than 16 buffers (tests window refill) + std::vector strings; + std::vector bufs; + std::string expected; + + for(int i = 0; i < 50; ++i) + { + strings.push_back(std::to_string(i) + ","); + expected += strings.back(); + } + + for(auto const& s : strings) + bufs.emplace_back(s.data(), s.size()); + + buffer_param bp(bufs); + auto result = consume_all(bp); + BOOST_TEST_EQ(result, expected); + } + + void + testMutableSingleBuffer() + { + // Mutable buffer - write into it + { + char data[32] = {}; + capy::mutable_buffer buf(data, sizeof(data)); + + buffer_param bp(buf); + auto bufs = bp.data(); + + BOOST_TEST_EQ(bufs.size(), 1u); + BOOST_TEST_EQ(bufs[0].size(), sizeof(data)); + + // Write some data + std::memcpy(bufs[0].data(), "Test", 4); + bp.consume(4); + + bufs = bp.data(); + BOOST_TEST_EQ(bufs[0].size(), sizeof(data) - 4); + + // Write more + std::memcpy(bufs[0].data(), "Data", 4); + bp.consume(4); + + BOOST_TEST_EQ(std::string(data, 8), "TestData"); + } + } + + void + testMutableMultipleBuffers() + { + char buf1[8] = {}; + char buf2[8] = {}; + char buf3[8] = {}; + + std::vector bufs = { + capy::mutable_buffer(buf1, sizeof(buf1)), + capy::mutable_buffer(buf2, sizeof(buf2)), + capy::mutable_buffer(buf3, sizeof(buf3)) + }; + + buffer_param bp(bufs); + + // Fill all buffers through the param + std::size_t total = 0; + int chunk = 0; + while(true) + { + auto spans = bp.data(); + if(spans.empty()) + break; + + for(auto& b : spans) + { + std::memset(b.data(), 'A' + chunk, b.size()); + total += b.size(); + } + bp.consume(total_size(spans)); + ++chunk; + } + + BOOST_TEST_EQ(total, 24u); + + // Verify all buffers were written + BOOST_TEST_EQ(std::string(buf1, 8), "AAAAAAAA"); + BOOST_TEST_EQ(std::string(buf2, 8), "AAAAAAAA"); + BOOST_TEST_EQ(std::string(buf3, 8), "AAAAAAAA"); + } + + void + testBufferType() + { + // Verify buffer_type is correct for const sequences + { + std::vector bufs; + buffer_param bp(bufs); + static_assert(std::is_same_v< + decltype(bp)::buffer_type, capy::const_buffer>); + } + + // Verify buffer_type is correct for mutable sequences + { + std::vector bufs; + buffer_param bp(bufs); + static_assert(std::is_same_v< + decltype(bp)::buffer_type, capy::mutable_buffer>); + } + } + + void + testMore() + { + // Single buffer — no more after first window + { + std::string data = "Hello"; + capy::const_buffer buf(data.data(), data.size()); + buffer_param bp(buf); + + auto bufs = bp.data(); + BOOST_TEST(! bufs.empty()); + BOOST_TEST(! bp.more()); + } + + // Empty sequence — no more + { + std::vector bufs; + buffer_param bp(bufs); + auto d = bp.data(); + BOOST_TEST(d.empty()); + BOOST_TEST(! bp.more()); + } + + // Exactly max_iovec_ buffers — fits in one window + { + std::vector strings(capy::detail::max_iovec_, "x"); + std::vector bufs; + for(auto const& s : strings) + bufs.emplace_back(s.data(), s.size()); + + buffer_param bp(bufs); + auto d = bp.data(); + BOOST_TEST(! d.empty()); + BOOST_TEST(! bp.more()); + } + + // One more than max_iovec_ — needs two windows + { + std::vector strings( + capy::detail::max_iovec_ + 1, "x"); + std::vector bufs; + for(auto const& s : strings) + bufs.emplace_back(s.data(), s.size()); + + buffer_param bp(bufs); + auto d = bp.data(); + BOOST_TEST(! d.empty()); + BOOST_TEST(bp.more()); + + bp.consume(total_size(d)); + d = bp.data(); + BOOST_TEST(! d.empty()); + BOOST_TEST(! bp.more()); + + bp.consume(total_size(d)); + d = bp.data(); + BOOST_TEST(d.empty()); + } + } + + void + testPartialByteConsumption() + { + std::string data = "ABCDEFGHIJ"; + capy::const_buffer buf(data.data(), data.size()); + + buffer_param bp(buf); + std::string result; + std::size_t n = 5; + + while(n > 0) + { + auto bufs = bp.data(); + if(bufs.empty()) + break; + auto chunk = (std::min)(n, total_size(bufs)); + // Copy only 'chunk' bytes + std::size_t copied = 0; + for(auto const& b : bufs) + { + auto take = (std::min)(b.size(), chunk - copied); + result.append( + static_cast(b.data()), + take); + copied += take; + if(copied >= chunk) + break; + } + bp.consume(chunk); + n -= chunk; + } + + BOOST_TEST_EQ(result, "ABCDE"); + } + + void + run() + { + testConstSingleBuffer(); + testConstMultipleBuffers(); + testConstPartialConsume(); + testConstLargeSequence(); + testMutableSingleBuffer(); + testMutableMultipleBuffers(); + testBufferType(); + testMore(); + testPartialByteConsumption(); + } +}; + +TEST_SUITE( + buffer_param_test, + "boost.http.io.buffer_param"); + +} // http +} // boost diff --git a/test/unit/parser.cpp b/test/unit/parser.cpp index d540838b..0dbb3279 100644 --- a/test/unit/parser.cpp +++ b/test/unit/parser.cpp @@ -2264,6 +2264,43 @@ struct parser_coro_test BOOST_TEST(r.success); } + // More destination buffers than one copy window holds, with the + // whole message already buffered: every body byte must arrive + // before read reports eof. + void + testReadManyBuffers() + { + capy::test::fuse f; + auto r = f.armed([&](capy::test::fuse&) -> capy::task<> + { + capy::test::read_stream rs(f); + rs.provide( + "HTTP/1.1 200 OK\r\n" + "Content-Length: 26\r\n" + "\r\n" + "abcdefghijklmnopqrstuvwxyz"); + + response_parser pr(res_cfg_); + pr.reset(); + pr.start(); + + char storage[32] = {}; + std::array bufs; + for(std::size_t i = 0; i < bufs.size(); ++i) + bufs[i] = capy::mutable_buffer(storage + i, 1); + auto [ec, n] = co_await pr.read(rs, bufs); + ignore_unused(ec); + + if(! pr.is_complete()) + co_return; + + BOOST_TEST_EQ(n, 26u); + BOOST_TEST(core::string_view(storage, n) == + "abcdefghijklmnopqrstuvwxyz"); + }); + BOOST_TEST(r.success); + } + void testReadChunked() { @@ -2352,6 +2389,7 @@ struct parser_coro_test testSourceForBodyTooLarge(); testRead(); testReadBuffers(); + testReadManyBuffers(); testReadChunked(); testReadEof(); testReadOverflow(); diff --git a/test/unit/serializer.cpp b/test/unit/serializer.cpp index 6726c5a6..c36aa90d 100644 --- a/test/unit/serializer.cpp +++ b/test/unit/serializer.cpp @@ -13,7 +13,6 @@ #include #include -#include #include #include #include @@ -22,6 +21,11 @@ #include #include +#ifdef BOOST_HTTP_HAS_ZLIB +#include +#include +#endif + #include "test_helpers.hpp" #include @@ -61,19 +65,23 @@ struct serializer_test read_some(serializer& sr) { auto bufs = sr.prepare().value(); - auto cbs = capy::buffer_slice(bufs, 0, 256); BOOST_TEST(!sr.is_done()); // We limit buffer consumption to necessitate // multiple calls to serializer::prepare() and // serializer::consume(), allowing tests to cover // state management within these functions + std::size_t budget = 256; std::string s; - for( auto buf : cbs) + for(auto it = capy::buffer_sequence_begin(bufs); + it != capy::buffer_sequence_end(bufs) && budget > 0; ++it) { + capy::const_buffer buf(*it); + std::size_t const n = (std::min)(buf.size(), budget); s.append( reinterpret_cast(buf.data()), - buf.size()); - sr.consume(buf.size()); + n); + sr.consume(n); + budget -= n; } return s; } @@ -1278,6 +1286,75 @@ struct serializer_test BOOST_TEST(r.success); } +#ifdef BOOST_HTTP_HAS_ZLIB + // The final filter call passes empty input buffers; the encoder + // must still be told to finish or the serializer never completes. + void + testGzipFinishes() + { + auto& ctx = capy::get_system_context(); + if(! ctx.find_service()) + zlib::install_deflate_service(ctx); + + serializer_config c; + c.apply_gzip_encoder = true; + auto const cfg = make_serializer_config(c); + + auto drain = [](serializer& sr) + { + std::string out; + for(int i = 0; i < 1000 && ! sr.is_done(); ++i) + { + auto rv = sr.prepare(); + if(rv.has_error()) + { + BOOST_TEST(rv.error() == error::need_data); + return out; + } + for(auto it = capy::buffer_sequence_begin(rv.value()); + it != capy::buffer_sequence_end(rv.value()); ++it) + { + capy::const_buffer b(*it); + out.append( + static_cast(b.data()), b.size()); + } + sr.consume(capy::buffer_size(rv.value())); + } + return out; + }; + + response res; + res.set(field::content_encoding, "gzip"); + res.set_chunked(true); + + // empty body + { + serializer sr(cfg); + sr.set_message(res); + sr.start(); + auto out = drain(sr); + BOOST_TEST(sr.is_done()); + BOOST_TEST(out.ends_with("0\r\n\r\n")); + } + + // streamed body + { + serializer sr(cfg); + sr.set_message(res); + sr.start_stream(); + std::string body(1000, 'x'); + auto mb = sr.stream_prepare(); + auto n = capy::buffer_copy(mb, capy::make_buffer(body)); + BOOST_TEST_EQ(n, body.size()); + sr.stream_commit(n); + sr.stream_close(); + auto out = drain(sr); + BOOST_TEST(sr.is_done()); + BOOST_TEST(out.ends_with("0\r\n\r\n")); + } + } +#endif + void run() { @@ -1306,6 +1383,10 @@ struct serializer_test // any_buffer_sink wrapper tests (WriteSink) testAnyBufferSinkWrite(); testAnyBufferSinkWriteWithEof(); + +#ifdef BOOST_HTTP_HAS_ZLIB + testGzipFinishes(); +#endif } }; From 324fff596a5aba76c1d4b7623c8025499d08add8 Mon Sep 17 00:00:00 2001 From: Steve Gerbino Date: Fri, 30 Oct 2026 22:55:37 +0100 Subject: [PATCH 2/2] fix: release zlib streams when content filters are destroyed The deflate and inflate filters never called deflate_end or inflate_end, so every gzip or deflate message leaked the zlib state. The brotli filter already releases its encoder in its destructor. --- src/parser.cpp | 5 +++++ src/serializer.cpp | 5 +++++ 2 files changed, 10 insertions(+) diff --git a/src/parser.cpp b/src/parser.cpp index 4147778f..e73cb3bb 100644 --- a/src/parser.cpp +++ b/src/parser.cpp @@ -338,6 +338,11 @@ class zlib_filter detail::throw_system_error(ec); } + ~zlib_filter() + { + svc_.inflate_end(strm_); + } + private: virtual results diff --git a/src/serializer.cpp b/src/serializer.cpp index 20103537..58bfa2cb 100644 --- a/src/serializer.cpp +++ b/src/serializer.cpp @@ -156,6 +156,11 @@ class zlib_filter detail::throw_system_error(ec); } + ~zlib_filter() + { + svc_.deflate_end(strm_); + } + private: virtual std::size_t