Implements batch reads for server pushes.

Author: mzimbres

README.md

@@ -674,22 +674,34 @@ https://lists.boost.org/Archives/boost/2023/01/253944.php.
 
 ## Changelog
 
-### develop (incorporates changes to conform the boost review and more)
+### develop
 
 * Deprecates the `async_receive` overload that takes a response. Users
-  should now first call `set_receive_response` to avoid contantly seting
-  the same response.
+  should now first call `set_receive_response` to avoid constantly and
+  unnecessarily setting the same response.
 
 * Uses `std::function` to type erase the response adapter. This change
   should not influence users in any way but allowed important
-  simplification in the connections internals. This resulted in big
-  performance improvement where one of my benchmark programs passed
-  from 190k/s to 473k/s.
+  simplification in the connections internals. This resulted in
+  massive performance improvement.
 
 * The connection has a new member `get_usage()` that returns the
-  connection usage information, such as number of bytes writen,
+  connection usage information, such as number of bytes written,
   received etc.
 
+* There are massive performance improvements in the consuming of
+  server pushes which are now communicated with an `asio::channel` and
+  therefore can be buffered which avoids blocking the socket read-loop.
+  Batch reads are also supported by means of `channel.try_send` and
+  buffered messages can be consumed synchronously with
+  `connection::receive`. The function `boost::redis::cancel_one` has
+  been added to simplify processing multiple server pushes contained
+  in the same `generic_response`.  *IMPORTANT*: These changes may
+  result in more than one push in the response when
+  `connection::async_receive` resumes. The user must therefore be
+  careful when calling `resp.clear()`: either ensure that all message
+  have been processed or just use `consume_one`.
+
 ### v1.4.2 (incorporates changes to conform the boost review and more)
 
 * Adds `boost::redis::config::database_index` to make it possible to

examples/cpp20_subscriber.cpp

@@ -22,9 +22,11 @@ namespace asio = boost::asio;
 using namespace std::chrono_literals;
 using boost::redis::request;
 using boost::redis::generic_response;
+using boost::redis::consume_one;
 using boost::redis::logger;
 using boost::redis::config;
 using boost::redis::ignore;
+using boost::redis::error;
 using boost::system::error_code;
 using boost::redis::connection;
 using signal_set = asio::deferred_t::as_default_on_t<asio::signal_set>;
@@ -58,20 +60,28 @@ receiver(std::shared_ptr<connection> conn) -> asio::awaitable<void>
    // Loop while reconnection is enabled
    while (conn->will_reconnect()) {
 
-      // Reconnect to channels.
+      // Reconnect to the channels.
       co_await conn->async_exec(req, ignore, asio::deferred);
 
       // Loop reading Redis pushs messages.
       for (error_code ec;;) {
-         co_await conn->async_receive(asio::redirect_error(asio::use_awaitable, ec));
+         // First tries to read any buffered pushes.
+         conn->receive(ec);
+         if (ec == error::sync_receive_push_failed) {
+            ec = {};
+            co_await conn->async_receive(asio::redirect_error(asio::use_awaitable, ec));
+         }
+
          if (ec)
             break; // Connection lost, break so we can reconnect to channels.
+
          std::cout
             << resp.value().at(1).value
             << " " << resp.value().at(2).value
             << " " << resp.value().at(3).value
             << std::endl;
-         resp.value().clear();
+
+         consume_one(resp);
       }
    }
 }

include/boost/redis/adapter/detail/adapters.hpp

@@ -139,11 +139,7 @@ class simple_impl {
    void on_value_available(Result&) {}
 
    template <class String>
-   void
-   operator()(
-      Result& result,
-      resp3::basic_node<String> const& n,
-      system::error_code& ec)
+   void operator()(Result& result, resp3::basic_node<String> const& n, system::error_code& ec)
    {
       if (is_aggregate(n.data_type)) {
          ec = redis::error::expects_resp3_simple_type;
@@ -164,11 +160,7 @@ class set_impl {
       { hint_ = std::end(result); }
 
    template <class String>
-   void
-   operator()(
-      Result& result,
-      resp3::basic_node<String> const& nd,
-      system::error_code& ec)
+   void operator()(Result& result, resp3::basic_node<String> const& nd, system::error_code& ec)
    {
       if (is_aggregate(nd.data_type)) {
          if (nd.data_type != resp3::type::set)
@@ -200,11 +192,7 @@ class map_impl {
       { current_ = std::end(result); }
 
    template <class String>
-   void
-   operator()(
-      Result& result,
-      resp3::basic_node<String> const& nd,
-      system::error_code& ec)
+   void operator()(Result& result, resp3::basic_node<String> const& nd, system::error_code& ec)
    {
       if (is_aggregate(nd.data_type)) {
          if (element_multiplicity(nd.data_type) != 2)
@@ -239,11 +227,7 @@ class vector_impl {
    void on_value_available(Result& ) { }
 
    template <class String>
-   void
-   operator()(
-      Result& result,
-      resp3::basic_node<String> const& nd,
-      system::error_code& ec)
+   void operator()(Result& result, resp3::basic_node<String> const& nd, system::error_code& ec)
    {
       if (is_aggregate(nd.data_type)) {
          auto const m = element_multiplicity(nd.data_type);
@@ -264,11 +248,7 @@ class array_impl {
    void on_value_available(Result& ) { }
 
    template <class String>
-   void
-   operator()(
-      Result& result,
-      resp3::basic_node<String> const& nd,
-      system::error_code& ec)
+   void operator()(Result& result, resp3::basic_node<String> const& nd, system::error_code& ec)
    {
       if (is_aggregate(nd.data_type)) {
 	 if (i_ != -1) {
@@ -300,11 +280,7 @@ struct list_impl {
    void on_value_available(Result& ) { }
 
    template <class String>
-   void
-   operator()(
-      Result& result,
-      resp3::basic_node<String> const& nd,
-      system::error_code& ec)
+   void operator()(Result& result, resp3::basic_node<String> const& nd, system::error_code& ec)
    {
       if (!is_aggregate(nd.data_type)) {
         BOOST_ASSERT(nd.aggregate_size == 1);
@@ -397,10 +373,7 @@ class wrapper<result<Result>> {
    }
 
    template <class String>
-   void
-   operator()(
-      resp3::basic_node<String> const& nd,
-      system::error_code& ec)
+   void operator()(resp3::basic_node<String> const& nd, system::error_code& ec)
    {
       BOOST_ASSERT_MSG(!!result_, "Unexpected null pointer");
 

include/boost/redis/adapter/detail/response_traits.hpp

@@ -24,8 +24,7 @@ namespace boost::redis::adapter::detail
 class ignore_adapter {
 public:
    template <class String>
-   void
-   operator()(std::size_t, resp3::basic_node<String> const& nd, system::error_code& ec)
+   void operator()(std::size_t, resp3::basic_node<String> const& nd, system::error_code& ec)
    {
       switch (nd.data_type) {
          case resp3::type::simple_error: ec = redis::error::resp3_simple_error; break;
@@ -61,11 +60,7 @@ class static_adapter {
       { return size;}
 
    template <class String>
-   void
-   operator()(
-      std::size_t i,
-      resp3::basic_node<String> const& nd,
-      system::error_code& ec)
+   void operator()(std::size_t i, resp3::basic_node<String> const& nd, system::error_code& ec)
    {
       using std::visit;
       // I am usure whether this should be an error or an assertion.
@@ -91,11 +86,7 @@ class vector_adapter {
       { return static_cast<std::size_t>(-1);}
 
    template <class String>
-   void
-   operator()(
-      std::size_t,
-      resp3::basic_node<String> const& nd,
-      system::error_code& ec)
+   void operator()(std::size_t, resp3::basic_node<String> const& nd, system::error_code& ec)
    {
       adapter_(nd, ec);
    }

include/boost/redis/connection.hpp

@@ -188,6 +188,23 @@ class basic_connection {
    auto async_receive(CompletionToken token = CompletionToken{})
       { return impl_.async_receive(std::move(token)); }
 
+   
+   /** @brief Receives server pushes synchronously without blocking.
+    *
+    *  Receives a server push synchronously by calling `try_receive` on
+    *  the underlying channel. If the operation fails because
+    *  `try_receive` returns `false`, `ec` will be set to
+    *  `boost::redis::error::sync_receive_push_failed`.
+    *
+    *  @param ec Contains the error if any occurred.
+    *
+    *  @returns The number of bytes read from the socket.
+    */
+   std::size_t receive(system::error_code& ec)
+   {
+      return impl_.receive(ec);
+   }
+
    template <
       class Response = ignore_t,
       class CompletionToken = asio::default_completion_token_t<executor_type>
@@ -367,6 +384,12 @@ class connection {
    auto async_receive(CompletionToken token)
       { return impl_.async_receive(std::move(token)); }
 
+   /// Calls `boost::redis::basic_connection::receive`.
+   std::size_t receive(system::error_code& ec)
+   {
+      return impl_.receive(ec);
+   }
+
    /// Calls `boost::redis::basic_connection::async_exec`.
    template <class Response, class CompletionToken>
    auto async_exec(request const& req, Response& resp, CompletionToken token)

include/boost/redis/detail/connection_base.hpp

@@ -348,8 +348,10 @@ struct reader_op {
          }
 
          if (res_.first == parse_result::push) {
-            BOOST_ASIO_CORO_YIELD
-            conn_->receive_channel_.async_send(ec, res_.second, std::move(self));
+            if (!conn_->receive_channel_.try_send(ec, res_.second)) {
+               BOOST_ASIO_CORO_YIELD
+               conn_->receive_channel_.async_send(ec, res_.second, std::move(self));
+            }
 
             if (ec) {
                logger_.trace("reader-op: error. Exiting ...");
@@ -398,7 +400,7 @@ class connection_base {
    : ctx_{method}
    , stream_{std::make_unique<next_layer_type>(ex, ctx_)}
    , writer_timer_{ex}
-   , receive_channel_{ex}
+   , receive_channel_{ex, 256}
    , runner_{ex, {}}
    , dbuf_{read_buffer_, max_read_size}
    {
@@ -470,6 +472,26 @@ class connection_base {
    auto async_receive(CompletionToken token)
       { return receive_channel_.async_receive(std::move(token)); }
 
+   std::size_t receive(system::error_code& ec)
+   {
+      std::size_t size = 0;
+
+      auto f = [&](system::error_code const& ec2, std::size_t n)
+      {
+         ec = ec2;
+         size = n;
+      };
+
+      auto const res = receive_channel_.try_receive(f);
+      if (ec)
+         return 0;
+
+      if (!res)
+         ec = error::sync_receive_push_failed;
+
+      return size;
+   }
+
    template <class Logger, class CompletionToken>
    auto async_run(config const& cfg, Logger l, CompletionToken token)
    {

include/boost/redis/error.hpp

@@ -75,6 +75,12 @@ enum class error
 
    /// SSL handshake timeout
    ssl_handshake_timeout,
+
+   /// Can't receive push synchronously without blocking
+   sync_receive_push_failed,
+
+   /// Incompatible node depth.
+   incompatible_node_depth,
 };
 
 /** \internal

include/boost/redis/impl/error.ipp

@@ -41,6 +41,9 @@ struct error_category_impl : system::error_category {
 	 case error::resolve_timeout: return "Resolve timeout.";
 	 case error::connect_timeout: return "Connect timeout.";
 	 case error::pong_timeout: return "Pong timeout.";
+	 case error::ssl_handshake_timeout: return "SSL handshake timeout.";
+	 case error::sync_receive_push_failed: return "Can't receive server push synchronously without blocking.";
+	 case error::incompatible_node_depth: return "Incompatible node depth.";
 	 default: BOOST_ASSERT(false); return "Boost.Redis error.";
       }
    }

include/boost/redis/impl/response.ipp

@@ -0,0 +1,48 @@
+/* Copyright (c) 2018-2022 Marcelo Zimbres Silva ([email protected])
+ *
+ * Distributed under the Boost Software License, Version 1.0. (See
+ * accompanying file LICENSE.txt)
+ */
+
+#include <boost/redis/response.hpp>
+#include <boost/redis/error.hpp>
+#include <boost/assert.hpp>
+
+namespace boost::redis
+{
+
+void consume_one(generic_response& r, system::error_code& ec)
+{
+   if (r.has_error())
+      return; // Nothing to consume.
+
+   if (std::empty(r.value()))
+      return; // Nothing to consume.
+
+   auto const depth = r.value().front().depth;
+
+   // To simplify we will refuse to consume any data-type that is not
+   // a root node. I think there is no use for that and it is complex
+   // since it requires updating parent nodes.
+   if (depth != 0) {
+      ec = error::incompatible_node_depth;
+      return;
+   }
+
+   auto f = [depth](auto const& e)
+      { return e.depth == depth; };
+
+   auto match = std::find_if(std::next(std::cbegin(r.value())), std::cend(r.value()), f);
+
+   r.value().erase(std::cbegin(r.value()), match);
+}
+
+void consume_one(generic_response& r)
+{
+   system::error_code ec;
+   consume_one(r, ec);
+   if (ec)
+      throw system::system_error(ec);
+}
+
+} // boost::redis::resp3