Removes async_append_some

include/boost/redis/config.hpp

@@ -88,12 +88,20 @@ struct config {
     */
    std::chrono::steady_clock::duration reconnect_wait_interval = std::chrono::seconds{1};
 
-   /** @brief Maximum size of a socket read, in bytes.
+   /** @brief Maximum size of the 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.
     */
-   std::size_t max_read_size = (std::numeric_limits<std::size_t>::max)();
+   std::size_t max_read_buffer_size = (std::numeric_limits<std::size_t>::max)();
+
+   /** @brief read_buffer_append_size
+    *
+    * The size by which the read buffer grows when more space is
+    * needed. There is no need to set this too high because memory is
+    * reused and the growth will tend to zero.
+    */
+   std::size_t read_buffer_append_size = 4096;
 };
 
 }  // namespace boost::redis

include/boost/redis/connection.hpp

@@ -57,56 +57,6 @@
 namespace boost::redis {
 namespace detail {
 
-template <class AsyncReadStream, class DynamicBuffer>
-class append_some_op {
-private:
-   AsyncReadStream& stream_;
-   DynamicBuffer buf_;
-   std::size_t size_ = 0;
-   std::size_t tmp_ = 0;
-   asio::coroutine coro_{};
-
-public:
-   append_some_op(AsyncReadStream& stream, DynamicBuffer buf, std::size_t size)
-   : stream_{stream}
-   , buf_{std::move(buf)}
-   , size_{size}
-   { }
-
-   template <class Self>
-   void operator()(Self& self, system::error_code ec = {}, std::size_t n = 0)
-   {
-      BOOST_ASIO_CORO_REENTER(coro_)
-      {
-         tmp_ = buf_.size();
-         buf_.grow(size_);
-
-         BOOST_ASIO_CORO_YIELD
-         stream_.async_read_some(buf_.data(tmp_, size_), std::move(self));
-         if (ec) {
-            self.complete(ec, 0);
-            return;
-         }
-
-         buf_.shrink(buf_.size() - tmp_ - n);
-         self.complete({}, n);
-      }
-   }
-};
-
-template <class AsyncReadStream, class DynamicBuffer, class CompletionToken>
-auto async_append_some(
-   AsyncReadStream& stream,
-   DynamicBuffer buffer,
-   std::size_t size,
-   CompletionToken&& token)
-{
-   return asio::async_compose<CompletionToken, void(system::error_code, std::size_t)>(
-      append_some_op<AsyncReadStream, DynamicBuffer>{stream, buffer, size},
-      token,
-      stream);
-}
-
 template <class Executor>
 using exec_notifier_type = asio::experimental::channel<
    Executor,
@@ -209,33 +159,18 @@ struct writer_op {
 
 template <class Conn>
 struct reader_op {
-   using dyn_buffer_type = asio::dynamic_string_buffer<
-      char,
-      std::char_traits<char>,
-      std::allocator<char>>;
-
-   // TODO: Move this to config so the user can fine tune?
-   static constexpr std::size_t buffer_growth_hint = 4096;
-
    Conn* conn_;
-   detail::reader_fsm fsm_;
 
 public:
    reader_op(Conn& conn) noexcept
    : conn_{&conn}
-   , fsm_{conn.mpx_}
    { }
 
    template <class Self>
    void operator()(Self& self, system::error_code ec = {}, std::size_t n = 0)
    {
-      using dyn_buffer_type = asio::dynamic_string_buffer<
-         char,
-         std::char_traits<char>,
-         std::allocator<char>>;
-
       for (;;) {
-         auto act = fsm_.resume(n, ec, self.get_cancellation_state().cancelled());
+         auto act = conn_->read_fsm_.resume(n, ec, self.get_cancellation_state().cancelled());
 
          conn_->logger_.on_fsm_resume(act);
 
@@ -245,11 +180,10 @@ struct reader_op {
                continue;
             case reader_fsm::action::type::needs_more:
             case reader_fsm::action::type::append_some:
-               async_append_some(
-                  conn_->stream_,
-                  dyn_buffer_type{conn_->mpx_.get_read_buffer(), conn_->cfg_.max_read_size},
-                  conn_->mpx_.get_parser().get_suggested_buffer_growth(buffer_growth_hint),
-                  std::move(self));
+            {
+               auto const buf = conn_->read_fsm_.get_append_buffer();
+               conn_->stream_.async_read_some(asio::buffer(buf.first, buf.second), std::move(self));
+            }
                return;
             case reader_fsm::action::type::notify_push_receiver:
                if (conn_->receive_channel_.try_send(ec, act.push_size_)) {
@@ -343,6 +277,7 @@ class run_op {
             // If we were successful, run all the connection tasks
             if (!ec) {
                conn_->mpx_.reset();
+               conn_->read_fsm_.reset();
 
                // Note: Order is important here because the writer might
                // trigger an async_write before the async_hello thereby
@@ -450,6 +385,7 @@ class basic_connection {
    , reconnect_timer_{ex}
    , receive_channel_{ex, 256}
    , health_checker_{ex}
+   , read_fsm_{mpx_}
    , logger_{std::move(lgr)}
    {
       set_receive_response(ignore);
@@ -553,6 +489,7 @@ class basic_connection {
       cfg_ = cfg;
       health_checker_.set_config(cfg);
       handshaker_.set_config(cfg);
+      read_fsm_.set_config({cfg_.read_buffer_append_size, cfg_.max_read_buffer_size});
 
       return asio::async_compose<CompletionToken, void(system::error_code)>(
          detail::run_op<this_type>{this},
@@ -951,6 +888,7 @@ class basic_connection {
 
    config cfg_;
    detail::multiplexer mpx_;
+   detail::reader_fsm read_fsm_;
    detail::connection_logger logger_;
 };
 

include/boost/redis/detail/multiplexer.hpp

@@ -8,13 +8,13 @@
 #define BOOST_REDIS_MULTIPLEXER_HPP
 
 #include <boost/redis/adapter/adapt.hpp>
-#include <boost/redis/adapter/any_adapter.hpp>
-#include <boost/redis/config.hpp>
-#include <boost/redis/operation.hpp>
+#include <boost/redis/detail/read_buffer.hpp>
+#include <boost/redis/resp3/node.hpp>
+#include <boost/redis/resp3/parser.hpp>
 #include <boost/redis/resp3/type.hpp>
 #include <boost/redis/usage.hpp>
 
-#include <boost/asio/experimental/channel.hpp>
+#include <boost/system.hpp>
 
 #include <algorithm>
 #include <deque>
@@ -32,7 +32,8 @@ namespace detail {
 
 using tribool = std::optional<bool>;
 
-struct multiplexer {
+class multiplexer {
+public:
    using adapter_type = std::function<void(resp3::node_view const&, system::error_code&)>;
    using pipeline_adapter_type = std::function<
       void(std::size_t, resp3::node_view const&, system::error_code&)>;
@@ -127,7 +128,8 @@ struct multiplexer {
    // If the tribool contains no value more data is needed, otherwise
    // if the value is true the message consumed is a push.
    [[nodiscard]]
-   auto consume_next(system::error_code& ec) -> std::pair<tribool, std::size_t>;
+   auto consume_next(std::string_view data, system::error_code& ec)
+      -> std::pair<tribool, std::size_t>;
 
    auto add(std::shared_ptr<elem> const& ptr) -> void;
    auto reset() -> void;
@@ -156,18 +158,6 @@ struct multiplexer {
       return std::string_view{write_buffer_};
    }
 
-   [[nodiscard]]
-   auto get_read_buffer() noexcept -> std::string&
-   {
-      return read_buffer_;
-   }
-
-   [[nodiscard]]
-   auto get_read_buffer() const noexcept -> std::string const&
-   {
-      return read_buffer_;
-   }
-
    // TODO: Change signature to receive an adapter instead of a
    // response.
    template <class Response>
@@ -191,17 +181,18 @@ struct multiplexer {
    [[nodiscard]]
    auto is_waiting_response() const noexcept -> bool;
 
-   [[nodiscard]]
-   auto on_finish_parsing(bool is_push) -> std::size_t;
+   void commit_usage(bool is_push, std::size_t size);
 
    [[nodiscard]]
-   auto is_next_push() const noexcept -> bool;
+   auto is_next_push(std::string_view data) const noexcept -> bool;
 
    // Releases the number of requests that have been released.
    [[nodiscard]]
    auto release_push_requests() -> std::size_t;
 
-   std::string read_buffer_;
+   [[nodiscard]]
+   tribool consume_next_impl(std::string_view data, system::error_code& ec);
+
    std::string write_buffer_;
    std::deque<std::shared_ptr<elem>> reqs_;
    resp3::parser parser_{};

include/boost/redis/detail/read_buffer.hpp

@@ -0,0 +1,46 @@
+/* Copyright (c) 2018-2025 Marcelo Zimbres Silva ([email protected])
+ *
+ * Distributed under the Boost Software License, Version 1.0. (See
+ * accompanying file LICENSE.txt)
+ */
+
+#ifndef BOOST_REDIS_READ_BUFFER_HPP
+#define BOOST_REDIS_READ_BUFFER_HPP
+
+#include <cstddef>
+#include <string>
+#include <string_view>
+#include <utility>
+
+namespace boost::redis::detail {
+
+class read_buffer {
+public:
+   void prepare_append(
+      std::size_t append_size,
+      std::size_t max_buffer_size,
+      system::error_code& ec);
+
+   void commit_append(std::size_t read_size);
+
+   [[nodiscard]]
+   auto get_append_buffer() noexcept -> std::pair<char*, std::size_t>;
+
+   [[nodiscard]]
+   auto get_committed_buffer() const noexcept -> std::string_view;
+
+   [[nodiscard]]
+   auto get_committed_size() const noexcept -> std::size_t;
+
+   void clear();
+
+   void consume(std::size_t size);
+
+private:
+   std::string buffer_;
+   std::size_t append_buf_begin_ = 0;
+};
+
+}  // namespace boost::redis::detail
+
+#endif  // BOOST_REDIS_READ_BUFFER_HPP

include/boost/redis/detail/reader_fsm.hpp

@@ -18,6 +18,12 @@ namespace boost::redis::detail {
 
 class reader_fsm {
 public:
+   // See config.hpp for the meaning of these parameters.
+   struct config {
+      std::size_t read_buffer_append_size = 4096;
+      std::size_t max_read_buffer_size = -1;
+   };
+
    struct action {
       enum class type
       {
@@ -41,8 +47,20 @@ class reader_fsm {
       system::error_code ec,
       asio::cancellation_type_t /*cancel_state*/);
 
+   void set_config(config const& cfg) noexcept { cfg_ = cfg; };
+
+   void reset();
+
+   [[nodiscard]]
+   auto get_append_buffer() noexcept
+   {
+      return read_buffer_.get_append_buffer();
+   }
+
 private:
    int resume_point_{0};
+   read_buffer read_buffer_;
+   config cfg_;
    action action_after_resume_;
    action::type next_read_type_ = action::type::append_some;
    multiplexer* mpx_ = nullptr;

include/boost/redis/error.hpp

@@ -88,6 +88,9 @@ enum class error
 
    /// The configuration specified UNIX sockets with SSL, which is not supported.
    unix_sockets_ssl_unsupported,
+
+   /// The size of the read buffer would exceed it maximum configured value.
+   exceeds_maximum_read_buffer_size,
 };
 
 /**

include/boost/redis/impl/error.ipp

@@ -50,6 +50,8 @@ struct error_category_impl : system::error_category {
                    "supported by the system.";
          case error::unix_sockets_ssl_unsupported:
             return "The configuration specified UNIX sockets with SSL, which is not supported.";
+         case error::exceeds_maximum_read_buffer_size:
+            return "The size of the read buffer would exceed it maximum configured value";
          default: BOOST_ASSERT(false); return "Boost.Redis error.";
       }
    }