Merge pull request #153 from boostorg/152-enable-reading-server-pushes-in-batches

152 enable reading server pushes in batches

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

@@ -92,7 +92,8 @@ class general_aggregate {
 
 public:
    explicit general_aggregate(Result* c = nullptr): result_(c) {}
-   void operator()(resp3::basic_node<std::string_view> const& nd, system::error_code&)
+   template <class String>
+   void operator()(resp3::basic_node<String> const& nd, system::error_code&)
    {
       BOOST_ASSERT_MSG(!!result_, "Unexpected null pointer");
       switch (nd.data_type) {
@@ -114,7 +115,8 @@ class general_simple {
 public:
    explicit general_simple(Node* t = nullptr) : result_(t) {}
 
-   void operator()(resp3::basic_node<std::string_view> const& nd, system::error_code&)
+   template <class String>
+   void operator()(resp3::basic_node<String> const& nd, system::error_code&)
    {
       BOOST_ASSERT_MSG(!!result_, "Unexpected null pointer");
       switch (nd.data_type) {
@@ -136,11 +138,8 @@ class simple_impl {
 public:
    void on_value_available(Result&) {}
 
-   void
-   operator()(
-      Result& result,
-      resp3::basic_node<std::string_view> const& n,
-      system::error_code& ec)
+   template <class String>
+   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;
@@ -160,11 +159,8 @@ class set_impl {
    void on_value_available(Result& result)
       { hint_ = std::end(result); }
 
-   void
-   operator()(
-      Result& result,
-      resp3::basic_node<std::string_view> const& nd,
-      system::error_code& ec)
+   template <class String>
+   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)
@@ -195,11 +191,8 @@ class map_impl {
    void on_value_available(Result& result)
       { current_ = std::end(result); }
 
-   void
-   operator()(
-      Result& result,
-      resp3::basic_node<std::string_view> const& nd,
-      system::error_code& ec)
+   template <class String>
+   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)
@@ -233,11 +226,8 @@ class vector_impl {
 public:
    void on_value_available(Result& ) { }
 
-   void
-   operator()(
-      Result& result,
-      resp3::basic_node<std::string_view> const& nd,
-      system::error_code& ec)
+   template <class String>
+   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);
@@ -257,11 +247,8 @@ class array_impl {
 public:
    void on_value_available(Result& ) { }
 
-   void
-   operator()(
-      Result& result,
-      resp3::basic_node<std::string_view> const& nd,
-      system::error_code& ec)
+   template <class String>
+   void operator()(Result& result, resp3::basic_node<String> const& nd, system::error_code& ec)
    {
       if (is_aggregate(nd.data_type)) {
 	 if (i_ != -1) {
@@ -292,11 +279,8 @@ struct list_impl {
 
    void on_value_available(Result& ) { }
 
-   void
-   operator()(
-      Result& result,
-      resp3::basic_node<std::string_view> const& nd,
-      system::error_code& ec)
+   template <class String>
+   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);
@@ -365,7 +349,8 @@ class wrapper<result<Result>> {
    response_type* result_;
    typename impl_map<Result>::type impl_;
 
-   bool set_if_resp3_error(resp3::basic_node<std::string_view> const& nd) noexcept
+   template <class String>
+   bool set_if_resp3_error(resp3::basic_node<String> const& nd) noexcept
    {
       switch (nd.data_type) {
          case resp3::type::null:
@@ -387,10 +372,8 @@ class wrapper<result<Result>> {
       }
    }
 
-   void
-   operator()(
-      resp3::basic_node<std::string_view> const& nd,
-      system::error_code& ec)
+   template <class String>
+   void operator()(resp3::basic_node<String> const& nd, system::error_code& ec)
    {
       BOOST_ASSERT_MSG(!!result_, "Unexpected null pointer");
 
@@ -414,7 +397,8 @@ class wrapper<result<std::optional<T>>> {
    response_type* result_;
    typename impl_map<T>::type impl_{};
 
-   bool set_if_resp3_error(resp3::basic_node<std::string_view> const& nd) noexcept
+   template <class String>
+   bool set_if_resp3_error(resp3::basic_node<String> const& nd) noexcept
    {
       switch (nd.data_type) {
          case resp3::type::blob_error:
@@ -429,9 +413,10 @@ class wrapper<result<std::optional<T>>> {
 public:
    explicit wrapper(response_type* o = nullptr) : result_(o) {}
 
+   template <class String>
    void
    operator()(
-      resp3::basic_node<std::string_view> const& nd,
+      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

@@ -23,8 +23,8 @@ namespace boost::redis::adapter::detail
 
 class ignore_adapter {
 public:
-   void
-   operator()(std::size_t, resp3::basic_node<std::string_view> const& nd, system::error_code& ec)
+   template <class String>
+   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;
@@ -59,11 +59,8 @@ class static_adapter {
    auto get_supported_response_size() const noexcept
       { return size;}
 
-   void
-   operator()(
-      std::size_t i,
-      resp3::basic_node<std::string_view> const& nd,
-      system::error_code& ec)
+   template <class String>
+   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.
@@ -88,11 +85,8 @@ class vector_adapter {
    get_supported_response_size() const noexcept
       { return static_cast<std::size_t>(-1);}
 
-   void
-   operator()(
-      std::size_t,
-      resp3::basic_node<std::string_view> const& nd,
-      system::error_code& ec)
+   template <class String>
+   void operator()(std::size_t, resp3::basic_node<String> const& nd, system::error_code& ec)
    {
       adapter_(nd, ec);
    }
@@ -142,7 +136,8 @@ class wrapper {
 public:
    explicit wrapper(Adapter adapter) : adapter_{adapter} {}
 
-   void operator()(resp3::basic_node<std::string_view> const& nd, system::error_code& ec)
+   template <class String>
+   void operator()(resp3::basic_node<String> const& nd, system::error_code& ec)
       { return adapter_(0, nd, ec); }
 
    [[nodiscard]]

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

@@ -116,7 +116,8 @@ class static_aggregate_adapter<result<Tuple>> {
       }
    }
 
-   void count(resp3::basic_node<std::string_view> const& nd)
+   template <class String>
+   void count(resp3::basic_node<String> const& nd)
    {
       if (nd.depth == 1) {
          if (is_aggregate(nd.data_type))
@@ -131,7 +132,8 @@ class static_aggregate_adapter<result<Tuple>> {
          ++i_;
    }
 
-   void operator()(resp3::basic_node<std::string_view> const& nd, system::error_code& ec)
+   template <class String>
+   void operator()(resp3::basic_node<String> const& nd, system::error_code& ec)
    {
       using std::visit;
 

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)
    {