Simplifies async operations.

Author: mzimbres

include/boost/redis/connection.hpp

@@ -21,43 +21,6 @@
 #include <limits>
 
 namespace boost::redis {
-namespace detail
-{
-template <class Connection, class Logger>
-struct reconnection_op {
-   Connection* conn_ = nullptr;
-   Logger logger_;
-   asio::coroutine coro_{};
-
-   template <class Self>
-   void operator()(Self& self, system::error_code ec = {})
-   {
-      BOOST_ASIO_CORO_REENTER (coro_) for (;;)
-      {
-         BOOST_ASIO_CORO_YIELD
-         conn_->impl_.async_run(conn_->cfg_, logger_, std::move(self));
-         conn_->cancel(operation::receive);
-         logger_.on_connection_lost(ec);
-         if (!conn_->will_reconnect()) {
-            conn_->cancel(operation::reconnection);
-            self.complete(ec);
-            return;
-         }
-
-         conn_->timer_.expires_after(conn_->cfg_.reconnect_wait_interval);
-         BOOST_ASIO_CORO_YIELD
-         conn_->timer_.async_wait(std::move(self));
-         BOOST_REDIS_CHECK_OP0(;)
-         if (!conn_->will_reconnect()) {
-            self.complete(asio::error::operation_aborted);
-            return;
-         }
-
-         conn_->reset_stream();
-      }
-   }
-};
-} // detail
 
 /** @brief A SSL connection to the Redis server.
  *  @ingroup high-level-api
@@ -100,7 +63,6 @@ class basic_connection {
       asio::ssl::context ctx = asio::ssl::context{asio::ssl::context::tlsv12_client},
       std::size_t max_read_size = (std::numeric_limits<std::size_t>::max)())
    : impl_{ex, std::move(ctx), max_read_size}
-   , timer_{ex}
    { }
 
    /// Contructs from a context.
@@ -158,14 +120,7 @@ class basic_connection {
       Logger l = Logger{},
       CompletionToken token = CompletionToken{})
    {
-      using this_type = basic_connection<executor_type>;
-
-      cfg_ = cfg;
-      l.set_prefix(cfg_.log_prefix);
-      return asio::async_compose
-         < CompletionToken
-         , void(system::error_code)
-         >(detail::reconnection_op<this_type, Logger>{this, l}, token, timer_);
+      return impl_.async_run(cfg, l, std::move(token));
    }
 
    /** @brief Receives server side pushes asynchronously.
@@ -272,22 +227,11 @@ class basic_connection {
     *  @param op: The operation to be cancelled.
     */
    void cancel(operation op = operation::all)
-   {
-      switch (op) {
-         case operation::reconnection:
-         case operation::all:
-            cfg_.reconnect_wait_interval = std::chrono::seconds::zero();
-            timer_.cancel();
-            break;
-         default: /* ignore */;
-      }
-
-      impl_.cancel(op);
-   }
+      { impl_.cancel(op); }
 
    /// Returns true if the connection was canceled.
    bool will_reconnect() const noexcept
-      { return cfg_.reconnect_wait_interval != std::chrono::seconds::zero();}
+      { return impl_.will_reconnect();}
 
    /// Returns the ssl context.
    auto const& get_ssl_context() const noexcept
@@ -315,17 +259,7 @@ class basic_connection {
       { return impl_.get_usage(); }
 
 private:
-   using timer_type =
-      asio::basic_waitable_timer<
-         std::chrono::steady_clock,
-         asio::wait_traits<std::chrono::steady_clock>,
-         Executor>;
-
-   template <class, class> friend struct detail::reconnection_op;
-
-   config cfg_;
    detail::connection_base<executor_type> impl_;
-   timer_type timer_;
 };
 
 /** \brief A basic_connection that type erases the executor.

include/boost/redis/detail/connection_base.hpp

@@ -313,13 +313,6 @@ struct reader_op {
 
             logger_.on_read(ec, n);
 
-            // EOF is not treated as error.
-            if (ec == asio::error::eof) {
-               logger_.trace("reader-op: EOF received. Exiting ...");
-               conn_->cancel(operation::run);
-               return self.complete(ec);
-            }
-
             // The connection is not viable after an error.
             if (ec) {
                logger_.trace("reader-op: error. Exiting ...");
@@ -429,7 +422,17 @@ class connection_base {
    /// Cancels specific operations.
    void cancel(operation op)
    {
+      // TODO: Simplify
+      switch (op) {
+         case operation::reconnection:
+         case operation::all:
+            cfg_.reconnect_wait_interval = std::chrono::seconds::zero();
+            break;
+         default: /* ignore */;
+      }
+
       runner_.cancel(op);
+
       if (op == operation::all) {
          cancel_impl(operation::run);
          cancel_impl(operation::receive);
@@ -490,8 +493,9 @@ class connection_base {
    template <class Logger, class CompletionToken>
    auto async_run(config const& cfg, Logger l, CompletionToken token)
    {
+      cfg_ = cfg;
       runner_.set_config(cfg);
-      l.set_prefix(runner_.get_config().log_prefix);
+      l.set_prefix(cfg.log_prefix);
       return runner_.async_run(*this, l, std::move(token));
    }
 
@@ -509,6 +513,9 @@ class connection_base {
    auto run_is_canceled() const noexcept
       { return cancel_run_called_; }
 
+   bool will_reconnect() const noexcept
+      { return cfg_.reconnect_wait_interval != std::chrono::seconds::zero();}
+
 private:
    using receive_channel_type = asio::experimental::channel<executor_type, void(system::error_code, std::size_t)>;
    using runner_type = runner<executor_type>;
@@ -517,7 +524,7 @@ class connection_base {
    using exec_notifier_type = receive_channel_type;
 
    auto use_ssl() const noexcept
-      { return runner_.get_config().use_ssl;}
+      { return cfg_.use_ssl;}
 
    auto cancel_on_conn_lost() -> std::size_t
    {
@@ -760,10 +767,8 @@ class connection_base {
    }
 
    template <class Logger, class CompletionToken>
-   auto async_run_lean(config const& cfg, Logger l, CompletionToken token)
+   auto async_run_lean(Logger l, CompletionToken token)
    {
-      runner_.set_config(cfg);
-      l.set_prefix(runner_.get_config().log_prefix);
       return asio::async_compose
          < CompletionToken
          , void(system::error_code)
@@ -948,6 +953,7 @@ class connection_base {
 
    using dyn_buffer_type = asio::dynamic_string_buffer<char, std::char_traits<char>, std::allocator<char>>;
 
+   config cfg_;
    std::string read_buffer_;
    dyn_buffer_type dbuf_;
    std::string write_buffer_;

include/boost/redis/detail/runner.hpp

@@ -87,7 +87,7 @@ class runner_op {
                   , system::error_code ec2 = {}
                   , std::size_t = 0)
    {
-      BOOST_ASIO_CORO_REENTER (coro_)
+      BOOST_ASIO_CORO_REENTER (coro_) for (;;)
       {
          BOOST_ASIO_CORO_YIELD
          runner_->resv_.async_resolve(asio::prepend(std::move(self), order_t {}));
@@ -133,17 +133,14 @@ class runner_op {
          asio::experimental::make_parallel_group(
             [this](auto token) { return runner_->async_hello(*conn_, logger_, token); },
             [this](auto token) { return runner_->health_checker_.async_check_health(*conn_, logger_, token); },
-            [this](auto token) { return conn_->async_run_lean(runner_->cfg_, logger_, token); }
+            [this](auto token) { return conn_->async_run_lean(logger_, token); }
          ).async_wait(
             asio::experimental::wait_for_one_error(),
             std::move(self));
 
+         // TODO: Unify these lines.
          logger_.on_runner(ec0, ec1, ec2);
-
-         if (is_cancelled(self)) {
-            self.complete(asio::error::operation_aborted);
-            return;
-         }
+         logger_.on_connection_lost({});
 
          if (order[0] == 0 && !!ec0) {
             self.complete(ec0);
@@ -155,7 +152,34 @@ class runner_op {
             return;
          }
 
-         self.complete(ec2);
+         // The receive operation must be cancelled because channel
+         // subscription does not survive a reconnection but requires
+         // re-subscription.
+         conn_->cancel(operation::receive);
+
+         if (!conn_->will_reconnect()) {
+            conn_->cancel(operation::reconnection);
+            self.complete(ec2);
+            return;
+         }
+
+         // It is safe to use the writer timer here because we are not
+         // connected.
+         conn_->writer_timer_.expires_after(conn_->cfg_.reconnect_wait_interval);
+
+         BOOST_ASIO_CORO_YIELD
+         conn_->writer_timer_.async_wait(asio::prepend(std::move(self), order_t {}));
+         if (ec0) {
+            self.complete(ec0);
+            return;
+         }
+
+         if (!conn_->will_reconnect()) {
+            self.complete(asio::error::operation_aborted);
+            return;
+         }
+
+         conn_->reset_stream();
       }
    }
 };
@@ -196,8 +220,6 @@ class runner {
          >(runner_op<runner, Connection, Logger>{this, &conn, l}, token, conn);
    }
 
-   config const& get_config() const noexcept {return cfg_;}
-
 private:
    using resolver_type = resolver<Executor>;
    using handshaker_type = detail::handshaker<Executor>;

test/CMakeLists.txt

@@ -46,7 +46,6 @@ make_test(test_conn_reconnect 20)
 make_test(test_conn_exec_cancel 20)
 make_test(test_conn_exec_cancel2 20)
 make_test(test_conn_echo_stress 20)
-make_test(test_conn_run_cancel 20)
 make_test(test_issue_50 20)
 make_test(test_issue_181 17)
 

test/common.hpp

@@ -24,5 +24,5 @@ run(
    boost::redis::config cfg = make_test_config(),
    boost::system::error_code ec = boost::asio::error::operation_aborted,
    boost::redis::operation op = boost::redis::operation::receive,
-   boost::redis::logger::level l = boost::redis::logger::level::disabled);
+   boost::redis::logger::level l = boost::redis::logger::level::debug);
 

test/test_conn_exec_retry.cpp

@@ -75,8 +75,13 @@ BOOST_AUTO_TEST_CASE(request_retry_false)
    conn->async_exec(req0, ignore, c0);
 
    auto cfg = make_test_config();
-   cfg.health_check_interval = 5s;
-   run(conn);
+   conn->async_run(cfg, {boost::redis::logger::level::debug},
+      [&](boost::system::error_code const& ec)
+      {
+         std::cout << "async_run: " << ec.message() << std::endl;
+         conn->cancel();
+      }
+   );
 
    ioc.run();
 }

test/test_conn_quit.cpp

@@ -20,24 +20,6 @@ using boost::redis::response;
 using boost::redis::ignore;
 using namespace std::chrono_literals;
 
-BOOST_AUTO_TEST_CASE(test_eof_no_error)
-{
-   request req;
-   req.get_config().cancel_on_connection_lost = false;
-   req.push("QUIT");
-
-   net::io_context ioc;
-   auto conn = std::make_shared<connection>(ioc);
-
-   conn->async_exec(req, ignore, [&](auto ec, auto) {
-      BOOST_TEST(!ec);
-      conn->cancel(operation::reconnection);
-   });
-
-   run(conn);
-   ioc.run();
-}
-
 // Test if quit causes async_run to exit.
 BOOST_AUTO_TEST_CASE(test_async_run_exits)
 {