Avoids suspending the fsm when a sync push notification is possible.

Author: mzimbres

include/boost/redis/connection.hpp

@@ -223,7 +223,7 @@ struct reader_op {
 public:
    reader_op(Conn& conn) noexcept
    : conn_{&conn}
-   , fsm_{conn.mpx_}
+   , fsm_{conn.mpx_, [&conn](std::size_t size){return conn.receive_channel_.try_send(system::error_code(), size);}}
    { }
 
    template <class Self>
@@ -252,12 +252,7 @@ struct reader_op {
                   std::move(self));
                return;
             case reader_fsm::action::type::notify_push_receiver:
-               if (conn_->receive_channel_.try_send(ec, act.push_size_)) {
-                  continue;
-               } else {
-                  conn_->receive_channel_.async_send(ec, act.push_size_, std::move(self));
-                  return;
-               }
+               conn_->receive_channel_.async_send(ec, act.push_size_, std::move(self));
                return;
             case reader_fsm::action::type::cancel_run: conn_->cancel(operation::run); continue;
             case reader_fsm::action::type::done:       self.complete(act.ec_); return;

include/boost/redis/detail/reader_fsm.hpp

@@ -13,11 +13,14 @@
 #include <boost/system/error_code.hpp>
 
 #include <cstddef>
+#include <functional>
 
 namespace boost::redis::detail {
 
 class reader_fsm {
 public:
+   using push_notifier_type = std::function<bool(std::size_t)>;
+
    struct action {
       enum class type
       {
@@ -34,7 +37,8 @@ class reader_fsm {
       system::error_code ec_ = {};
    };
 
-   explicit reader_fsm(multiplexer& mpx) noexcept;
+   explicit
+   reader_fsm(multiplexer& mpx, push_notifier_type push_notifier) noexcept;
 
    action resume(
       std::size_t bytes_read,
@@ -46,6 +50,7 @@ class reader_fsm {
    action action_after_resume_;
    action::type next_read_type_ = action::type::append_some;
    multiplexer* mpx_ = nullptr;
+   push_notifier_type push_notifier_;
    std::pair<tribool, std::size_t> res_{std::make_pair(std::nullopt, 0)};
 };
 

include/boost/redis/impl/reader_fsm.ipp

@@ -10,8 +10,9 @@
 
 namespace boost::redis::detail {
 
-reader_fsm::reader_fsm(multiplexer& mpx) noexcept
+reader_fsm::reader_fsm(multiplexer& mpx, push_notifier_type push_notifier) noexcept
 : mpx_{&mpx}
+, push_notifier_{std::move(push_notifier)}
 { }
 
 reader_fsm::action reader_fsm::resume(
@@ -48,20 +49,13 @@ reader_fsm::action reader_fsm::resume(
                break;
             }
 
-            if (!res_.first.value()) {
+            if (!res_.first.value() && !push_notifier_(res_.second)) {
                BOOST_REDIS_YIELD(resume_point_, 6, action::type::notify_push_receiver, res_.second)
                if (ec) {
                   action_after_resume_ = {action::type::done, 0u, ec};
                   BOOST_REDIS_YIELD(resume_point_, 7, action::type::cancel_run)
                   return action_after_resume_;
                }
-            } else {
-               // TODO: Here we should notify the exec operation that
-               // it can be completed. This will improve log clarity
-               // and will make this code symmetrical in how it
-               // handles pushes and other messages. The new action
-               // type can be named notify_exec. To do that we need to
-               // refactor the multiplexer.
             }
          }
       }

test/test_reader_fsm.cpp

@@ -35,12 +35,16 @@ std::ostream& operator<<(std::ostream& os, reader_fsm::action::type t)
 // Operators
 namespace {
 
+// Sync push notifier that always fails so that the fsm has to suspend
+// so that the caller can notify it asynchronously.
+auto sync_push_notifier = [](std::size_t){return false;};
+
 void test_push()
 {
    multiplexer mpx;
    generic_response resp;
    mpx.set_receive_response(resp);
-   reader_fsm fsm{mpx};
+   reader_fsm fsm{mpx, sync_push_notifier};
    error_code ec;
    action act;
 
@@ -85,7 +89,7 @@ void test_read_needs_more()
    multiplexer mpx;
    generic_response resp;
    mpx.set_receive_response(resp);
-   reader_fsm fsm{mpx};
+   reader_fsm fsm{mpx, sync_push_notifier};
    error_code ec;
    action act;
 
@@ -130,7 +134,7 @@ void test_read_error()
    multiplexer mpx;
    generic_response resp;
    mpx.set_receive_response(resp);
-   reader_fsm fsm{mpx};
+   reader_fsm fsm{mpx, sync_push_notifier};
    error_code ec;
    action act;
 
@@ -160,7 +164,7 @@ void test_parse_error()
    multiplexer mpx;
    generic_response resp;
    mpx.set_receive_response(resp);
-   reader_fsm fsm{mpx};
+   reader_fsm fsm{mpx, sync_push_notifier};
    error_code ec;
    action act;
 
@@ -190,7 +194,7 @@ void test_push_deliver_error()
    multiplexer mpx;
    generic_response resp;
    mpx.set_receive_response(resp);
-   reader_fsm fsm{mpx};
+   reader_fsm fsm{mpx, sync_push_notifier};
    error_code ec;
    action act;
 
@@ -219,10 +223,53 @@ void test_push_deliver_error()
    BOOST_TEST_EQ(act.ec_, error_code{net::error::operation_aborted});
 }
 
+void test_sync_send_push()
+{
+   auto push_notifier = [i = 0](std::size_t) mutable
+   {
+      // When i == 3 we are simulating that the push channel became
+      // full and an async notification is necessary.
+      return ++i != 3;
+   };
+
+   multiplexer mpx;
+   generic_response resp;
+   mpx.set_receive_response(resp);
+   reader_fsm fsm{mpx, push_notifier};
+   error_code ec;
+   action act;
+
+   // Initiate
+   act = fsm.resume(0, ec, cancellation_type_t::none);
+   BOOST_TEST_EQ(act.type_, action::type::setup_cancellation);
+   act = fsm.resume(0, ec, cancellation_type_t::none);
+   BOOST_TEST_EQ(act.type_, action::type::append_some);
+
+   // The fsm is asking for data.
+   mpx.get_read_buffer().append(">1\r\n+msg1\r\n");
+   mpx.get_read_buffer().append(">1\r\n+msg2 \r\n");
+   mpx.get_read_buffer().append(">1\r\n+msg3  \r\n");
+   auto const bytes_read = mpx.get_read_buffer().size();
+
+   // Only the last push has to be notified asynchronously. The other
+   // ones have been notified synchronously by the callback passed in
+   // the constructor.
+   act = fsm.resume(bytes_read, ec, cancellation_type_t::none);
+   BOOST_TEST_EQ(act.type_, action::type::notify_push_receiver);
+   BOOST_TEST_EQ(act.push_size_, 13u);
+   BOOST_TEST_EQ(act.ec_, error_code());
+
+   // All pushes were delivered so the fsm should demand more data
+   act = fsm.resume(0, ec, cancellation_type_t::none);
+   BOOST_TEST_EQ(act.type_, action::type::append_some);
+   BOOST_TEST_EQ(act.ec_, error_code());
+}
+
 }  // namespace
 
 int main()
 {
+   test_sync_send_push();
    test_push_deliver_error();
    test_read_needs_more();
    test_push();