Skip to content

Implements lazy buffer rotations #415

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mzimbres merged 1 commit into from
May 9, 2026
Merged

Implements lazy buffer rotations #415

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions include/boost/redis/config.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
*/
Expand Down
2 changes: 1 addition & 1 deletion include/boost/redis/connection.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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());
Expand Down
6 changes: 4 additions & 2 deletions include/boost/redis/detail/exec_one_fsm.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@

namespace boost::redis::detail {

class read_buffer;
class multiplexer;

// What should we do next?
enum class exec_one_action_type
Expand Down Expand Up @@ -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);
Expand Down
5 changes: 3 additions & 2 deletions include/boost/redis/detail/multiplexer.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);

Expand All @@ -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;
Expand All @@ -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<multiplexer::elem>;
Expand Down
99 changes: 89 additions & 10 deletions include/boost/redis/detail/read_buffer.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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)
Expand All @@ -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<char>;

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<std::size_t>(-1);
std::size_t max_size = static_cast<std::size_t>(-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;
Expand All @@ -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);

Expand All @@ -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<char> buffer_;
std::size_t append_buf_begin_ = 0;
std::size_t consumed_ = 0;
std::size_t prepared_begin_ = 0;
};

} // namespace boost::redis::detail
Expand Down
12 changes: 6 additions & 6 deletions include/boost/redis/impl/exec_one_fsm.ipp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -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;

Expand All @@ -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.
Expand Down
20 changes: 12 additions & 8 deletions include/boost/redis/impl/multiplexer.ipp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -174,13 +174,18 @@ std::pair<consume_result, std::size_t> 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
{
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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<multiplexer::elem>
Expand Down
Loading