diff --git a/include/boost/redis/config.hpp b/include/boost/redis/config.hpp index 98bb7624..a55c2d87 100644 --- a/include/boost/redis/config.hpp +++ b/include/boost/redis/config.hpp @@ -322,8 +322,8 @@ struct config { /** @brief Maximum size of the socket read-buffer in bytes. * - * Sets a limit on how much data is allowed to be read into the - * read buffer. It can be used to prevent DDOS. + * Sets a limit on how large the read-buffer is allowed to grow. It can be + * used to prevent DDOS. * * When using Sentinel, this setting applies to masters, replicas and Sentinels. */ diff --git a/include/boost/redis/connection.hpp b/include/boost/redis/connection.hpp index 46184c2c..5065b364 100644 --- a/include/boost/redis/connection.hpp +++ b/include/boost/redis/connection.hpp @@ -301,7 +301,7 @@ struct exec_one_op { void operator()(Self& self, system::error_code ec = {}, std::size_t bytes_written = 0u) { exec_one_action act = fsm_.resume( - conn_->st_.mpx.get_read_buffer(), + conn_->st_.mpx, ec, bytes_written, self.get_cancellation_state().cancelled()); diff --git a/include/boost/redis/detail/exec_one_fsm.hpp b/include/boost/redis/detail/exec_one_fsm.hpp index 4ad3bd3e..84a54a43 100644 --- a/include/boost/redis/detail/exec_one_fsm.hpp +++ b/include/boost/redis/detail/exec_one_fsm.hpp @@ -21,7 +21,7 @@ namespace boost::redis::detail { -class read_buffer; +class multiplexer; // What should we do next? enum class exec_one_action_type @@ -57,8 +57,10 @@ class exec_one_fsm { , remaining_responses_(expected_responses) { } + // Instead of using the read_buffer directly we use its facade in the + // multiplexer because it keeps track of usage information. exec_one_action resume( - read_buffer& buffer, + multiplexer& mpx, system::error_code ec, std::size_t bytes_transferred, asio::cancellation_type_t cancel_state); diff --git a/include/boost/redis/detail/multiplexer.hpp b/include/boost/redis/detail/multiplexer.hpp index 797c9493..c8c313c4 100644 --- a/include/boost/redis/detail/multiplexer.hpp +++ b/include/boost/redis/detail/multiplexer.hpp @@ -191,7 +191,7 @@ class multiplexer { auto get_prepared_read_buffer() noexcept -> read_buffer::span_type; [[nodiscard]] - auto prepare_read() noexcept -> system::error_code; + auto prepare_read()-> system::error_code; void commit_read(std::size_t read_size); @@ -209,7 +209,7 @@ class multiplexer { void set_config(config const& cfg); private: - void commit_usage(bool is_push, read_buffer::consume_result res); + void commit_usage(bool is_push, std::size_t consumed); [[nodiscard]] auto is_next_push(std::string_view data) const noexcept -> bool; @@ -229,6 +229,7 @@ class multiplexer { bool cancel_run_called_ = false; usage usage_; any_adapter receive_adapter_; + std::size_t append_size_ = 4096u; }; auto make_elem(request const& req, any_adapter adapter) -> std::shared_ptr; diff --git a/include/boost/redis/detail/read_buffer.hpp b/include/boost/redis/detail/read_buffer.hpp index 965845ec..b5573a4a 100644 --- a/include/boost/redis/detail/read_buffer.hpp +++ b/include/boost/redis/detail/read_buffer.hpp @@ -1,4 +1,4 @@ -/* Copyright (c) 2018-2025 Marcelo Zimbres Silva (mzimbres@gmail.com) +/* Copyright (c) 2018-2026 Marcelo Zimbres Silva (mzimbres@gmail.com) * * Distributed under the Boost Software License, Version 1.0. (See * accompanying file LICENSE.txt) @@ -17,24 +17,95 @@ namespace boost::redis::detail { +// Buffer class used in read operations that implements lazy rotations. It is +// split in three main parts +// +// 1. Consumed: Bytes that can be discarded but haven't to avoid rotating +// the buffer unnecessarily. +// +// 2. Commited: Area with data that is in use by the client code. +// +// 3. Prepared: Area waiting for data to be copied into it. +// +// The dynamics of the read-buffer is exemplified below +// +// 1. The buffer is empty and app start +// +// || +// +// 2. Client code calls "prepare" to reserve n bytes at the end of the +// buffer +// +// |+++++++++++++++++| +// +// 3. Bytes are read from the socket into that area and commited with the +// commit member function +// +// |-----------| +// +// 4. The steps above are repeated until the amount of data needed by the +// application is reached +// +// |-----------| +// |-----------+++++++++++++++++| Prepare +// |-----------------| Commit +// |-----------------++++++++++++++++| Prepare +// |---------------------------| Commit +// ... +// +// 5. Commited bytes are processed by the client and consumed. The consume +// op won't discard any data but increase an offset instead +// +// |============---------------| Consume +// +// 6. If preparing would cause the capacity to have to be increased we first +// discard already consumed data to perhaps avoid the reallocation +// +// |---------------| After the rotation +// |---------------++++++++++++++++| When prepare returns +// +// Buffer rotations can in principle be implemented both in the consume and in +// the prepare operation, the latter however produces better results because +// Redis commands are often very small e.g. "+OK\r\n" and a single read from +// the socket might bring in multiple responses, for example +// +// | Contains 100 responses | +// | | +// |=========-------------------------------| +// +// Consuming each response one by one would produce +// +// |=========-------------------------------| +// |===========-----------------------------| +// |=============---------------------------| +// |===============-------------------------| +// |=================-----------------------| +// |===================---------------------| +// +// When a prepare call comes we would like the consumed bytes to be zero so no +// reallocation must be performed, that can only be implemented in the consume +// op if it rotates eagerly i.e. on every consume call, something we don't want. + class read_buffer { public: using span_type = span; - struct consume_result { - std::size_t consumed; + struct prepare_result { std::size_t rotated; + system::error_code ec; }; // See config.hpp for the meaning of these parameters. struct config { - std::size_t read_buffer_append_size = 4096u; - std::size_t max_read_size = static_cast(-1); + std::size_t max_size = static_cast(-1); }; + read_buffer() noexcept = default; + read_buffer(config cfg) noexcept; + // Prepare the buffer to receive more data. [[nodiscard]] - auto prepare() -> system::error_code; + auto prepare(std::size_t append_size) -> prepare_result; [[nodiscard]] auto get_prepared() noexcept -> span_type; @@ -46,9 +117,7 @@ class read_buffer { void clear(); - // Consumes committed data by rotating the remaining data to the - // front of the buffer. - auto consume(std::size_t size) -> consume_result; + auto consume(std::size_t n) -> std::size_t; void reserve(std::size_t n); @@ -58,10 +127,20 @@ class read_buffer { void set_config(config const& cfg) noexcept { cfg_ = cfg; }; + // Returns the total size: consumed + commited + prepared. + std::size_t size() const noexcept + { return buffer_.size(); } + + std::size_t capacity() const noexcept + { return buffer_.capacity(); } + private: + bool needs_rotation(std::size_t append_size) const noexcept; + config cfg_ = config{}; std::vector buffer_; - std::size_t append_buf_begin_ = 0; + std::size_t consumed_ = 0; + std::size_t prepared_begin_ = 0; }; } // namespace boost::redis::detail diff --git a/include/boost/redis/impl/exec_one_fsm.ipp b/include/boost/redis/impl/exec_one_fsm.ipp index b4a7250b..50373986 100644 --- a/include/boost/redis/impl/exec_one_fsm.ipp +++ b/include/boost/redis/impl/exec_one_fsm.ipp @@ -27,7 +27,7 @@ namespace boost::redis::detail { exec_one_action exec_one_fsm::resume( - read_buffer& buffer, + multiplexer& mpx, system::error_code ec, std::size_t bytes_transferred, asio::cancellation_type_t cancel_state) @@ -49,10 +49,10 @@ exec_one_action exec_one_fsm::resume( return system::error_code{}; // Read responses until we're done - buffer.clear(); + mpx.get_read_buffer().clear(); while (true) { // Prepare the buffer to read some data - ec = buffer.prepare(); + ec = mpx.prepare_read(); if (ec) return ec; @@ -66,16 +66,16 @@ exec_one_action exec_one_fsm::resume( return ec; // Commit the data into the buffer - buffer.commit(bytes_transferred); + mpx.commit_read(bytes_transferred); // Consume the data until we run out or all the responses have been read - while (resp3::parse(parser_, buffer.get_commited(), adapter_, ec)) { + while (resp3::parse(parser_, mpx.get_read_buffer().get_commited(), adapter_, ec)) { // Check for errors if (ec) return ec; // We've finished parsing a response - buffer.consume(parser_.get_consumed()); + mpx.get_read_buffer().consume(parser_.get_consumed()); parser_.reset(); // When no more responses remain, we're done. diff --git a/include/boost/redis/impl/multiplexer.ipp b/include/boost/redis/impl/multiplexer.ipp index b6d118ce..f548b2ab 100644 --- a/include/boost/redis/impl/multiplexer.ipp +++ b/include/boost/redis/impl/multiplexer.ipp @@ -174,13 +174,18 @@ std::pair multiplexer::consume(system::error_code& parser_.reset(); auto const res = read_buffer_.consume(consumed); commit_usage(ret == consume_result::got_push, res); - return std::make_pair(ret, res.consumed); + return std::make_pair(ret, res); } return std::make_pair(consume_result::needs_more, consumed); } -auto multiplexer::prepare_read() noexcept -> system::error_code { return read_buffer_.prepare(); } +auto multiplexer::prepare_read() -> system::error_code +{ + auto const res = read_buffer_.prepare(append_size_); + usage_.bytes_rotated += res.rotated; + return res.ec; +} auto multiplexer::get_prepared_read_buffer() noexcept -> read_buffer::span_type { @@ -286,18 +291,16 @@ void multiplexer::cancel_on_conn_lost() }); } -void multiplexer::commit_usage(bool is_push, read_buffer::consume_result res) +void multiplexer::commit_usage(bool is_push, std::size_t consumed) { if (is_push) { usage_.pushes_received += 1; - usage_.push_bytes_received += res.consumed; + usage_.push_bytes_received += consumed; on_push_ = false; } else { usage_.responses_received += 1; - usage_.response_bytes_received += res.consumed; + usage_.response_bytes_received += consumed; } - - usage_.bytes_rotated += res.rotated; } bool multiplexer::is_next_push(std::string_view data) const noexcept @@ -359,7 +362,8 @@ void multiplexer::set_receive_adapter(any_adapter adapter) void multiplexer::set_config(config const& cfg) { - read_buffer_.set_config({cfg.read_buffer_append_size, cfg.max_read_size}); + append_size_ = cfg.read_buffer_append_size; + read_buffer_.set_config({cfg.max_read_size}); } auto make_elem(request const& req, any_adapter adapter) -> std::shared_ptr diff --git a/include/boost/redis/impl/read_buffer.ipp b/include/boost/redis/impl/read_buffer.ipp index b3705549..e817251b 100644 --- a/include/boost/redis/impl/read_buffer.ipp +++ b/include/boost/redis/impl/read_buffer.ipp @@ -13,66 +13,108 @@ namespace boost::redis::detail { -system::error_code read_buffer::prepare() +read_buffer::read_buffer(config cfg) noexcept +: cfg_{cfg} { - BOOST_ASSERT(append_buf_begin_ == buffer_.size()); +} + +bool read_buffer::needs_rotation(std::size_t append_size) const noexcept +{ + if (consumed_ == 0u) + return false; + + // If preparing would cause the capacity to have to be increased we first + // discard already consumed data to perhaps avoid the reallocation. + auto const capacity = buffer_.capacity(); + auto const size = buffer_.size(); + auto const remaining = capacity - size; + if (remaining < append_size) + return true; + + // Rotate if increasing the buffer size would require more than max. + return buffer_.size() > (cfg_.max_size - append_size); +} + +read_buffer::prepare_result read_buffer::prepare(std::size_t append_size) +{ + BOOST_ASSERT(prepared_begin_ == buffer_.size()); - auto const new_size = append_buf_begin_ + cfg_.read_buffer_append_size; + if (append_size > cfg_.max_size) { + return {0u, error::exceeds_maximum_read_buffer_size}; + } + + std::size_t rotated = 0u; + if (needs_rotation(append_size)) { + BOOST_ASSERT(consumed_ != 0u); + buffer_.erase(buffer_.begin(), buffer_.begin() + consumed_); + rotated = buffer_.size(); + + BOOST_ASSERT(consumed_ <= prepared_begin_); + prepared_begin_ -= consumed_; + consumed_ = 0u; + } + + auto const new_size = prepared_begin_ + append_size; - if (new_size > cfg_.max_read_size) { - return error::exceeds_maximum_read_buffer_size; + if (new_size > cfg_.max_size) { + return {rotated, error::exceeds_maximum_read_buffer_size}; } - buffer_.resize(new_size); - return {}; + { + auto const start_capacity = buffer_.capacity(); + auto const start_size = buffer_.size(); + buffer_.resize(new_size); + auto const end_capacity = buffer_.capacity(); + if (end_capacity > start_capacity) + rotated += start_size; + } + + return {rotated, {}}; } void read_buffer::commit(std::size_t read_size) { - BOOST_ASSERT(buffer_.size() >= (append_buf_begin_ + read_size)); - buffer_.resize(append_buf_begin_ + read_size); - append_buf_begin_ = buffer_.size(); + BOOST_ASSERT(buffer_.size() >= (prepared_begin_ + read_size)); + buffer_.resize(prepared_begin_ + read_size); + prepared_begin_ = buffer_.size(); } auto read_buffer::get_prepared() noexcept -> span_type { auto const size = buffer_.size(); - return make_span(buffer_.data() + append_buf_begin_, size - append_buf_begin_); + return make_span(buffer_.data() + prepared_begin_, size - prepared_begin_); } auto read_buffer::get_commited() const noexcept -> std::string_view { - return {buffer_.data(), append_buf_begin_}; + return {buffer_.data() + consumed_, prepared_begin_ - consumed_}; } void read_buffer::clear() { buffer_.clear(); - append_buf_begin_ = 0; + consumed_ = 0; + prepared_begin_ = 0; } -read_buffer::consume_result -read_buffer::consume(std::size_t size) +std::size_t read_buffer::consume(std::size_t n) { // For convenience, if the requested size is larger than the // committed buffer we cap it to the maximum. - if (size > append_buf_begin_) - size = append_buf_begin_; - - buffer_.erase(buffer_.begin(), buffer_.begin() + size); - auto const rotated = size == 0u ? 0u : buffer_.size(); - - BOOST_ASSERT(append_buf_begin_ >= size); - append_buf_begin_ -= size; + auto const consumable = prepared_begin_ - consumed_; + if (n > consumable) + n = consumable; - return {size, rotated}; + consumed_ += n; + BOOST_ASSERT(consumed_ <= prepared_begin_); + return n; } void read_buffer::reserve(std::size_t n) { buffer_.reserve(n); } bool operator==(read_buffer const& lhs, read_buffer const& rhs) { - return lhs.buffer_ == rhs.buffer_ && lhs.append_buf_begin_ == rhs.append_buf_begin_; + return lhs.buffer_ == rhs.buffer_ && lhs.prepared_begin_ == rhs.prepared_begin_; } bool operator!=(read_buffer const& lhs, read_buffer const& rhs) { return !(lhs == rhs); } diff --git a/include/boost/redis/impl/sentinel_utils.hpp b/include/boost/redis/impl/sentinel_utils.hpp index 36586629..dcb8efd1 100644 --- a/include/boost/redis/impl/sentinel_utils.hpp +++ b/include/boost/redis/impl/sentinel_utils.hpp @@ -25,6 +25,7 @@ #include #include #include +#include namespace boost::redis::detail { diff --git a/test/test_exec_one_fsm.cpp b/test/test_exec_one_fsm.cpp index 1e5d7c02..93c90bae 100644 --- a/test/test_exec_one_fsm.cpp +++ b/test/test_exec_one_fsm.cpp @@ -8,7 +8,7 @@ #include #include -#include +#include #include #include #include @@ -31,6 +31,7 @@ using detail::exec_one_fsm; using detail::exec_one_action; using detail::exec_one_action_type; using detail::read_buffer; +using detail::multiplexer; using boost::system::error_code; using boost::asio::cancellation_type_t; using parse_event = any_adapter::parse_event; @@ -109,20 +110,20 @@ void test_success() // Setup std::vector events; exec_one_fsm fsm{make_snoop_adapter(events), 2u}; - read_buffer buff; + multiplexer mpx; // Write the request - auto act = fsm.resume(buff, error_code(), 0u, cancellation_type_t::none); + auto act = fsm.resume(mpx, error_code(), 0u, cancellation_type_t::none); BOOST_TEST_EQ(act, exec_one_action_type::write); // FSM should now ask for data - act = fsm.resume(buff, error_code(), 25u, cancellation_type_t::none); + act = fsm.resume(mpx, error_code(), 25u, cancellation_type_t::none); BOOST_TEST_EQ(act, exec_one_action_type::read_some); // Read the entire response in one go constexpr std::string_view payload = "$5\r\nhello\r\n*1\r\n+goodbye\r\n"; - copy_to(buff, payload); - act = fsm.resume(buff, error_code(), payload.size(), cancellation_type_t::none); + copy_to(mpx.get_read_buffer(), payload); + act = fsm.resume(mpx, error_code(), payload.size(), cancellation_type_t::none); BOOST_TEST_EQ(act, exec_one_action_type::done); // Verify the adapter calls @@ -144,14 +145,14 @@ void test_no_expected_response() // Setup std::vector events; exec_one_fsm fsm{make_snoop_adapter(events), 0u}; - read_buffer buff; + multiplexer mpx; // Write the request - auto act = fsm.resume(buff, error_code(), 0u, cancellation_type_t::none); + auto act = fsm.resume(mpx, error_code(), 0u, cancellation_type_t::none); BOOST_TEST_EQ(act, exec_one_action_type::write); // FSM shouldn't ask for data - act = fsm.resume(buff, error_code(), 25u, cancellation_type_t::none); + act = fsm.resume(mpx, error_code(), 25u, cancellation_type_t::none); BOOST_TEST_EQ(act, error_code()); // No adapter calls should be done @@ -164,28 +165,28 @@ void test_short_reads() // Setup std::vector events; exec_one_fsm fsm{make_snoop_adapter(events), 2u}; - read_buffer buff; + multiplexer mpx; // Write the request - auto act = fsm.resume(buff, error_code(), 0u, cancellation_type_t::none); + auto act = fsm.resume(mpx, error_code(), 0u, cancellation_type_t::none); BOOST_TEST_EQ(act, exec_one_action_type::write); // FSM should now ask for data - act = fsm.resume(buff, error_code(), 25u, cancellation_type_t::none); + act = fsm.resume(mpx, error_code(), 25u, cancellation_type_t::none); BOOST_TEST_EQ(act, exec_one_action_type::read_some); // Read fragments constexpr std::string_view payload = "$5\r\nhello\r\n*1\r\n+goodbye\r\n"; - copy_to(buff, payload.substr(0, 6u)); - act = fsm.resume(buff, error_code(), 6u, cancellation_type_t::none); + copy_to(mpx.get_read_buffer(), payload.substr(0, 6u)); + act = fsm.resume(mpx, error_code(), 6u, cancellation_type_t::none); BOOST_TEST_EQ(act, exec_one_action_type::read_some); - copy_to(buff, payload.substr(6, 10u)); - act = fsm.resume(buff, error_code(), 10u, cancellation_type_t::none); + copy_to(mpx.get_read_buffer(), payload.substr(6, 10u)); + act = fsm.resume(mpx, error_code(), 10u, cancellation_type_t::none); BOOST_TEST_EQ(act, exec_one_action_type::read_some); - copy_to(buff, payload.substr(16)); - act = fsm.resume(buff, error_code(), payload.substr(16).size(), cancellation_type_t::none); + copy_to(mpx.get_read_buffer(), payload.substr(16)); + act = fsm.resume(mpx, error_code(), payload.substr(16).size(), cancellation_type_t::none); BOOST_TEST_EQ(act, exec_one_action_type::done); // Verify the adapter calls @@ -207,14 +208,14 @@ void test_write_error() // Setup std::vector events; exec_one_fsm fsm{make_snoop_adapter(events), 2u}; - read_buffer buff; + multiplexer mpx; // Write the request - auto act = fsm.resume(buff, error_code(), 0u, cancellation_type_t::none); + auto act = fsm.resume(mpx, error_code(), 0u, cancellation_type_t::none); BOOST_TEST_EQ(act, exec_one_action_type::write); // Write error - act = fsm.resume(buff, asio::error::connection_reset, 10u, cancellation_type_t::none); + act = fsm.resume(mpx, asio::error::connection_reset, 10u, cancellation_type_t::none); BOOST_TEST_EQ(act, error_code(asio::error::connection_reset)); } @@ -223,14 +224,14 @@ void test_write_cancel() // Setup std::vector events; exec_one_fsm fsm{make_snoop_adapter(events), 2u}; - read_buffer buff; + multiplexer mpx; // Write the request - auto act = fsm.resume(buff, error_code(), 0u, cancellation_type_t::none); + auto act = fsm.resume(mpx, error_code(), 0u, cancellation_type_t::none); BOOST_TEST_EQ(act, exec_one_action_type::write); // Edge case where the operation finished successfully but with the cancellation state set - act = fsm.resume(buff, error_code(), 10u, cancellation_type_t::terminal); + act = fsm.resume(mpx, error_code(), 10u, cancellation_type_t::terminal); BOOST_TEST_EQ(act, error_code(asio::error::operation_aborted)); } @@ -240,18 +241,18 @@ void test_read_error() // Setup std::vector events; exec_one_fsm fsm{make_snoop_adapter(events), 2u}; - read_buffer buff; + multiplexer mpx; // Write the request - auto act = fsm.resume(buff, error_code(), 0u, cancellation_type_t::none); + auto act = fsm.resume(mpx, error_code(), 0u, cancellation_type_t::none); BOOST_TEST_EQ(act, exec_one_action_type::write); // FSM should now ask for data - act = fsm.resume(buff, error_code(), 25u, cancellation_type_t::none); + act = fsm.resume(mpx, error_code(), 25u, cancellation_type_t::none); BOOST_TEST_EQ(act, exec_one_action_type::read_some); // Read error - act = fsm.resume(buff, asio::error::network_reset, 0u, cancellation_type_t::none); + act = fsm.resume(mpx, asio::error::network_reset, 0u, cancellation_type_t::none); BOOST_TEST_EQ(act, error_code(asio::error::network_reset)); } @@ -260,19 +261,19 @@ void test_read_cancelled() // Setup std::vector events; exec_one_fsm fsm{make_snoop_adapter(events), 2u}; - read_buffer buff; + multiplexer mpx; // Write the request - auto act = fsm.resume(buff, error_code(), 0u, cancellation_type_t::none); + auto act = fsm.resume(mpx, error_code(), 0u, cancellation_type_t::none); BOOST_TEST_EQ(act, exec_one_action_type::write); // FSM should now ask for data - act = fsm.resume(buff, error_code(), 25u, cancellation_type_t::none); + act = fsm.resume(mpx, error_code(), 25u, cancellation_type_t::none); BOOST_TEST_EQ(act, exec_one_action_type::read_some); // Edge case where the operation finished successfully but with the cancellation state set - copy_to(buff, "$5\r\n"); - act = fsm.resume(buff, error_code(), 4u, cancellation_type_t::terminal); + copy_to(mpx.get_read_buffer(), "$5\r\n"); + act = fsm.resume(mpx, error_code(), 4u, cancellation_type_t::terminal); BOOST_TEST_EQ(act, error_code(asio::error::operation_aborted)); } @@ -282,15 +283,15 @@ void test_buffer_prepare_error() // Setup std::vector events; exec_one_fsm fsm{make_snoop_adapter(events), 2u}; - read_buffer buff; - buff.set_config({4096u, 8u}); // max size is 8 bytes + multiplexer mpx; + mpx.get_read_buffer().set_config({8u}); // max size is 8 bytes // Write the request - auto act = fsm.resume(buff, error_code(), 0u, cancellation_type_t::none); + auto act = fsm.resume(mpx, error_code(), 0u, cancellation_type_t::none); BOOST_TEST_EQ(act, exec_one_action_type::write); // When preparing the buffer, we encounter an error - act = fsm.resume(buff, error_code(), 25u, cancellation_type_t::none); + act = fsm.resume(mpx, error_code(), 25u, cancellation_type_t::none); BOOST_TEST_EQ(act, error_code(error::exceeds_maximum_read_buffer_size)); } @@ -300,20 +301,20 @@ void test_parse_error() // Setup std::vector events; exec_one_fsm fsm{make_snoop_adapter(events), 2u}; - read_buffer buff; + multiplexer mpx; // Write the request - auto act = fsm.resume(buff, error_code(), 0u, cancellation_type_t::none); + auto act = fsm.resume(mpx, error_code(), 0u, cancellation_type_t::none); BOOST_TEST_EQ(act, exec_one_action_type::write); // FSM should now ask for data - act = fsm.resume(buff, error_code(), 25u, cancellation_type_t::none); + act = fsm.resume(mpx, error_code(), 25u, cancellation_type_t::none); BOOST_TEST_EQ(act, exec_one_action_type::read_some); // The response contains an invalid message constexpr std::string_view payload = "$bad\r\n"; - copy_to(buff, payload); - act = fsm.resume(buff, error_code(), payload.size(), cancellation_type_t::none); + copy_to(mpx.get_read_buffer(), payload); + act = fsm.resume(mpx, error_code(), payload.size(), cancellation_type_t::none); BOOST_TEST_EQ(act, error_code(error::not_a_number)); } @@ -326,20 +327,20 @@ void test_adapter_error() ec = error::empty_field; }}; exec_one_fsm fsm{std::move(adapter), 2u}; - read_buffer buff; + multiplexer mpx; // Write the request - auto act = fsm.resume(buff, error_code(), 0u, cancellation_type_t::none); + auto act = fsm.resume(mpx, error_code(), 0u, cancellation_type_t::none); BOOST_TEST_EQ(act, exec_one_action_type::write); // FSM should now ask for data - act = fsm.resume(buff, error_code(), 25u, cancellation_type_t::none); + act = fsm.resume(mpx, error_code(), 25u, cancellation_type_t::none); BOOST_TEST_EQ(act, exec_one_action_type::read_some); // Read the entire response in one go constexpr std::string_view payload = "$5\r\nhello\r\n*1\r\n+goodbye\r\n"; - copy_to(buff, payload); - act = fsm.resume(buff, error_code(), payload.size(), cancellation_type_t::none); + copy_to(mpx.get_read_buffer(), payload); + act = fsm.resume(mpx, error_code(), payload.size(), cancellation_type_t::none); BOOST_TEST_EQ(act, error_code(error::empty_field)); } diff --git a/test/test_read_buffer.cpp b/test/test_read_buffer.cpp index 85c0b07c..515e2dd6 100644 --- a/test/test_read_buffer.cpp +++ b/test/test_read_buffer.cpp @@ -14,89 +14,225 @@ using namespace boost::redis; using detail::read_buffer; using boost::system::error_code; +// NOTE1: The vector allocation behavior depends on the implementation. +// The test might not pass if we start testing on a new platform. + namespace { -void test_prepare_error() +void test_prepare_equals_max() { - read_buffer buf; + read_buffer buf{{10}}; - // Usual case, max size is bigger then requested size. - buf.set_config({10, 10}); - auto ec = buf.prepare(); - BOOST_TEST_EQ(ec, error_code()); - buf.commit(10); + // Corner case: prepare equals max. + auto const res = buf.prepare(10); + BOOST_TEST_EQ(res.ec, error_code()); +} - // Corner case, max size is equal to the requested size. - buf.set_config({10, 20}); - ec = buf.prepare(); - BOOST_TEST_EQ(ec, error_code()); - buf.commit(10); - buf.consume(20); +// TODO: bigger than max can happen in two situations. Check both +void test_prepare_bigger_than_max() +{ + read_buffer buf{{10}}; + auto const res = buf.prepare(11); + BOOST_TEST_EQ(res.ec, error_code{error::exceeds_maximum_read_buffer_size}); +} - auto const tmp = buf; +/* | max size | + * + * 1. |++++++++++++++++| prepare(16) + * 2. |----------------| commit(16) + * 3. |=========-------| consume(9) + * + * In this state the buffer has size 16 and a maximum configure size 20. + * Preparing for another 5 bytes exceeds the max size by one but should not + * fail since the 9 bytes in the front should be rotated by the implementation. + * + * 4. |=========-------| consume(9) + * 5. |-------+++++| prepare(5) + * + */ +void test_consume_avoids_prepare_max_error() +{ + read_buffer buf{{20}}; - // Error case, max size is smaller to the requested size. - buf.set_config({10, 9}); - ec = buf.prepare(); - BOOST_TEST_EQ(ec, error_code{error::exceeds_maximum_read_buffer_size}); + auto res = buf.prepare(16); + BOOST_TEST_EQ(res.ec, error_code()); + buf.commit(16); + auto consumed = buf.consume(9); + BOOST_TEST_EQ(consumed, 9u); - // Check that an error call has no side effects. - BOOST_TEST(buf == tmp); + res = buf.prepare(5); + BOOST_TEST_EQ(res.ec, error_code()); + BOOST_TEST_EQ(res.rotated, 7u); } void test_prepare_consume_only_committed_data() { - read_buffer buf; + read_buffer buf{{10}}; + + auto res = buf.prepare(10); + BOOST_TEST(!res.ec); + BOOST_TEST_EQ(res.rotated, 0u); + + // No data has been committed yet so nothing can be consumed. + auto consumed = buf.consume(5); + BOOST_TEST_EQ(consumed, 0u); + + buf.commit(10); + consumed = buf.consume(5); - buf.set_config({10, 10}); - auto ec = buf.prepare(); - BOOST_TEST(!ec); + // All five bytes should have been consumed. + BOOST_TEST_EQ(consumed, 5u); - auto res = buf.consume(5); + consumed = buf.consume(7); - // No data has been committed yet so nothing can be consummed. - BOOST_TEST_EQ(res.consumed, 0u); + // Only the remaining five bytes can be consumed + BOOST_TEST_EQ(consumed, 5u); +} - // If nothing was consumed, nothing got rotated. +void test_check_buffer_size() +{ + read_buffer buf{{10}}; + + auto res = buf.prepare(10); + BOOST_TEST_EQ(res.ec, error_code()); BOOST_TEST_EQ(res.rotated, 0u); - buf.commit(10); - res = buf.consume(5); + BOOST_TEST_EQ(buf.get_prepared().size(), 10u); +} - // All five bytes should have been consumed. - BOOST_TEST_EQ(res.consumed, 5u); +void test_prepared_erased_after_commit() +{ + read_buffer buf; - // We added a total of 10 bytes and consumed 5, that means, 5 were - // rotated. - BOOST_TEST_EQ(res.rotated, 5u); + auto res = buf.prepare(10); + BOOST_TEST_EQ(res.ec, error_code()); + BOOST_TEST_EQ(res.rotated, 0u); - res = buf.consume(7); + buf.commit(7); + auto prep = buf.get_prepared().size(); + BOOST_TEST_EQ(prep, 0u); - // Only the remaining five bytes can be consumed - BOOST_TEST_EQ(res.consumed, 5u); + res = buf.prepare(10); + prep = buf.get_prepared().size(); + BOOST_TEST_EQ(prep, 10u); +} - // No bytes to rotated. +/* 1. |++++++++++| - prepare(10) + * 2. |-------| - commit(7) + * 3. |-------++++++++++| - prepare(10) + * 4. |--------------| - commit(7) + * 5. |======--------| - consume(5) + */ +void test_prep_commit_consume_sizes() +{ + read_buffer buf; + + // 1. + auto res = buf.prepare(10); + BOOST_TEST_EQ(res.ec, error_code()); BOOST_TEST_EQ(res.rotated, 0u); + + // 2. + buf.commit(7); + BOOST_TEST_EQ(buf.size(), 7u); + auto prep = buf.get_prepared().size(); + BOOST_TEST_EQ(prep, 0u); + BOOST_TEST_EQ(buf.get_commited().size(), 7u); + + // 3. + res = buf.prepare(10); + BOOST_TEST_EQ(buf.size(), 17u); + prep = buf.get_prepared().size(); + BOOST_TEST_EQ(prep, 10u); + + // 4. + buf.commit(7); + BOOST_TEST_EQ(buf.size(), 14u); + prep = buf.get_prepared().size(); + BOOST_TEST_EQ(prep, 0u); + BOOST_TEST_EQ(buf.get_commited().size(), 14u); + + // 5. + buf.consume(5); + BOOST_TEST_EQ(buf.size(), 14u); + prep = buf.get_prepared().size(); + BOOST_TEST_EQ(prep, 0u); + BOOST_TEST_EQ(buf.get_commited().size(), 9u); } -void test_check_buffer_size() +/* | capacity | + * + * 1. |++++++++++++++++++++| prepare(20) + * 2. |--------------------| commit(20) + * 3. |===============-----| consume(15) + * 4. |-----++++++| prepare(6) + * + * Without rotation that last step would cause reallocation. The implementation + * should avoid that. + */ +void test_prepare_rotates_to_avoid_realloc() { read_buffer buf; + buf.reserve(25); + BOOST_TEST_EQ(buf.capacity(), 25); - buf.set_config({10, 10}); - auto ec = buf.prepare(); - BOOST_TEST_EQ(ec, error_code()); + // 1. + auto res = buf.prepare(20); + BOOST_TEST_EQ(res.ec, error_code()); - BOOST_TEST_EQ(buf.get_prepared().size(), 10u); + // 2. + buf.commit(20); + + // 3. + auto consumed = buf.consume(15); + BOOST_TEST_EQ(consumed, 15u); + + // 4. + res = buf.prepare(6); + BOOST_TEST_EQ(res.ec, error_code()); + BOOST_TEST_EQ(res.rotated, 5u); + BOOST_TEST_EQ(buf.capacity(), 25); +} + +/* 1. |++++++++++++++++| prepare(16) + * 2. |----------| commit(10) + * 3. |----------++++++++++| prepare(10) + * + * Step 3. should rotate no data since there is no consumed data. + */ +void test_no_rotation_when_consumed_zero() +{ + read_buffer buf; + buf.reserve(20); + BOOST_TEST_EQ(buf.capacity(), 20); + + // 1. + auto res = buf.prepare(16); + BOOST_TEST_EQ(res.ec, error_code()); + BOOST_TEST_EQ(res.rotated, 0u); + + // 2. + buf.commit(10); + + // 3. + res = buf.prepare(10); + BOOST_TEST_EQ(res.ec, error_code()); + BOOST_TEST_EQ(res.rotated, 0u); } } // namespace int main() { - test_prepare_error(); test_prepare_consume_only_committed_data(); test_check_buffer_size(); + test_prepared_erased_after_commit(); + test_prep_commit_consume_sizes(); + test_prepare_equals_max(); + test_prepare_bigger_than_max(); + test_consume_avoids_prepare_max_error(); + test_prepare_rotates_to_avoid_realloc(); + test_no_rotation_when_consumed_zero(); return boost::report_errors(); }