Reuses async_run parallel group for health-checker tasks.

Author: mzimbres

include/boost/redis/detail/connection_base.hpp

@@ -20,7 +20,6 @@
 #include <boost/system.hpp>
 #include <boost/asio/basic_stream_socket.hpp>
 #include <boost/asio/bind_executor.hpp>
-#include <boost/asio/experimental/parallel_group.hpp>
 #include <boost/asio/ip/tcp.hpp>
 #include <boost/asio/steady_timer.hpp>
 #include <boost/asio/write.hpp>

include/boost/redis/detail/health_checker.hpp

@@ -17,7 +17,6 @@
 #include <boost/asio/consign.hpp>
 #include <boost/asio/coroutine.hpp>
 #include <boost/asio/post.hpp>
-#include <boost/asio/experimental/parallel_group.hpp>
 #include <memory>
 #include <chrono>
 
@@ -36,6 +35,14 @@ class ping_op {
    {
       BOOST_ASIO_CORO_REENTER (coro_) for (;;)
       {
+         if (checker_->ping_interval_ == std::chrono::seconds::zero()) {
+            logger_.trace("ping-op: timeout disabled. Exiting ...");
+            BOOST_ASIO_CORO_YIELD
+            asio::post(std::move(self));
+            self.complete({});
+            return;
+         }
+
          if (checker_->checker_has_exited_) {
             logger_.trace("ping_op: checker has exited. Exiting ...");
             self.complete({});
@@ -77,6 +84,14 @@ class check_timeout_op {
    {
       BOOST_ASIO_CORO_REENTER (coro_) for (;;)
       {
+         if (checker_->ping_interval_ == std::chrono::seconds::zero()) {
+            logger_.trace("check-timeout-op: timeout disabled. Exiting ...");
+            BOOST_ASIO_CORO_YIELD
+            asio::post(std::move(self));
+            self.complete({});
+            return;
+         }
+
          checker_->wait_timer_.expires_after(2 * checker_->ping_interval_);
          BOOST_ASIO_CORO_YIELD
          checker_->wait_timer_.async_wait(std::move(self));
@@ -108,51 +123,6 @@ class check_timeout_op {
    }
 };
 
-template <class HealthChecker, class Connection, class Logger>
-class check_health_op {
-public:
-   HealthChecker* checker_ = nullptr;
-   Connection* conn_ = nullptr;
-   Logger logger_;
-   asio::coroutine coro_{};
-
-   template <class Self>
-   void
-   operator()(
-         Self& self,
-         std::array<std::size_t, 2> order = {},
-         system::error_code ec1 = {},
-         system::error_code ec2 = {})
-   {
-      BOOST_ASIO_CORO_REENTER (coro_)
-      {
-         if (checker_->ping_interval_ == std::chrono::seconds::zero()) {
-            logger_.trace("check-health-op: timeout disabled.");
-            BOOST_ASIO_CORO_YIELD
-            asio::post(std::move(self));
-            self.complete({});
-            return;
-         }
-
-         BOOST_ASIO_CORO_YIELD
-         asio::experimental::make_parallel_group(
-            [this](auto token) { return checker_->async_ping(*conn_, logger_, token); },
-            [this](auto token) { return checker_->async_check_timeout(*conn_, logger_, token);}
-         ).async_wait(
-            asio::experimental::wait_for_one(),
-            std::move(self));
-
-         logger_.on_check_health(ec1, ec2);
-
-         switch (order[0]) {
-            case 0: self.complete(ec1); return;
-            case 1: self.complete(ec2); return;
-            default: BOOST_ASSERT(false);
-         }
-      }
-   }
-};
-
 template <class Executor>
 class health_checker {
 private:
@@ -177,24 +147,6 @@ class health_checker {
       ping_interval_ = cfg.health_check_interval;
    }
 
-   template <
-      class Connection,
-      class Logger,
-      class CompletionToken = asio::default_completion_token_t<Executor>
-   >
-   auto
-   async_check_health(
-      Connection& conn,
-      Logger l,
-      CompletionToken token = CompletionToken{})
-   {
-      checker_has_exited_ = false;
-      return asio::async_compose
-         < CompletionToken
-         , void(system::error_code)
-         >(check_health_op<health_checker, Connection, Logger>{this, &conn, l}, token, conn);
-   }
-
    std::size_t cancel(operation op)
    {
       switch (op) {
@@ -209,7 +161,6 @@ class health_checker {
       return 0;
    }
 
-private:
    template <class Connection, class Logger, class CompletionToken>
    auto async_ping(Connection& conn, Logger l, CompletionToken token)
    {
@@ -228,9 +179,10 @@ class health_checker {
          >(check_timeout_op<health_checker, Connection, Logger>{this, &conn, l}, token, conn, wait_timer_);
    }
 
+private:
+
    template <class, class, class> friend class ping_op;
    template <class, class, class> friend class check_timeout_op;
-   template <class, class, class> friend class check_health_op;
 
    timer_type ping_timer_;
    timer_type wait_timer_;

include/boost/redis/detail/runner.hpp

@@ -32,6 +32,9 @@ namespace boost::redis::detail
 
 void push_hello(config const& cfg, request& req);
 
+// TODO: Can we avoid this whole function whose only purpose is to
+// check for an error in the hello response and complete with an error
+// so that the parallel group that starts it can exit?
 template <class Runner, class Connection, class Logger>
 struct hello_op {
    Runner* runner_ = nullptr;
@@ -50,10 +53,10 @@ struct hello_op {
          conn_->async_exec(runner_->hello_req_, runner_->hello_resp_, std::move(self));
          logger_.on_hello(ec, runner_->hello_resp_);
 
-         if (ec || runner_->has_error_in_response() || is_cancelled(self)) {
+         if (ec || runner_->has_error_in_response()) {
             logger_.trace("hello-op: error/canceled. Exiting ...");
             conn_->cancel(operation::run);
-            self.complete(!!ec ? ec : asio::error::operation_aborted);
+            self.complete(ec);
             return;
          }
 
@@ -70,7 +73,7 @@ class runner_op {
    Logger logger_;
    asio::coroutine coro_{};
 
-   using order_t = std::array<std::size_t, 4>;
+   using order_t = std::array<std::size_t, 5>;
 
 public:
    runner_op(Runner* runner, Connection* conn, Logger l)
@@ -85,7 +88,8 @@ class runner_op {
                   , system::error_code ec0 = {}
                   , system::error_code ec1 = {}
                   , system::error_code ec2 = {}
-                  , system::error_code ec3 = {})
+                  , system::error_code ec3 = {}
+                  , system::error_code ec4 = {})
    {
       BOOST_ASIO_CORO_REENTER (coro_) for (;;)
       {
@@ -130,18 +134,20 @@ class runner_op {
 
          // Note: Oder is important here because the writer might
          // trigger an async_write before the async_hello thereby
-         // causing authentication problems.
+         // causing an authentication problem.
          BOOST_ASIO_CORO_YIELD
          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 runner_->health_checker_.async_ping(*conn_, logger_, token); },
+            [this](auto token) { return runner_->health_checker_.async_check_timeout(*conn_, logger_, token);},
             [this](auto token) { return conn_->reader(logger_, token);},
             [this](auto token) { return conn_->writer(logger_, token);}
          ).async_wait(
             asio::experimental::wait_for_one_error(),
             std::move(self));
 
          // TODO: Unify these lines.
+         logger_.on_check_health(ec1, ec2);
          logger_.on_run(ec0, ec1);
          logger_.on_runner(ec0, ec1, ec2);
          logger_.on_connection_lost({});
@@ -151,7 +157,7 @@ class runner_op {
             return;
          }
 
-         if (order[0] == 1 && ec1 == error::pong_timeout) {
+         if (order[0] == 2 && ec2 == error::pong_timeout) {
             self.complete(ec1);
             return;
          }
@@ -163,7 +169,7 @@ class runner_op {
 
          if (!conn_->will_reconnect()) {
             conn_->cancel(operation::reconnection);
-            self.complete(ec2);
+            self.complete(ec3);
             return;
          }