Adds per-operation cancellation support to async_run and cleans up cancellation

include/boost/redis/detail/helper.hpp

@@ -17,18 +17,6 @@ auto is_cancelled(T const& self)
    return self.get_cancellation_state().cancelled() != asio::cancellation_type_t::none;
 }
 
-#define BOOST_REDIS_CHECK_OP0(X)                                   \
-   if (ec || redis::detail::is_cancelled(self)) {                  \
-      X self.complete(!!ec ? ec : asio::error::operation_aborted); \
-      return;                                                      \
-   }
-
-#define BOOST_REDIS_CHECK_OP1(X)                                       \
-   if (ec || redis::detail::is_cancelled(self)) {                      \
-      X self.complete(!!ec ? ec : asio::error::operation_aborted, {}); \
-      return;                                                          \
-   }
-
 }  // namespace boost::redis::detail
 
 #endif  // BOOST_REDIS_HELPER_HPP

include/boost/redis/detail/reader_fsm.hpp

@@ -22,15 +22,13 @@ class reader_fsm {
    struct action {
       enum class type
       {
-         setup_cancellation,
          read_some,
          needs_more,
          notify_push_receiver,
-         cancel_run,
          done,
       };
 
-      type type_ = type::setup_cancellation;
+      type type_ = type::done;
       std::size_t push_size_ = 0u;
       system::error_code ec_ = {};
    };
@@ -40,11 +38,10 @@ class reader_fsm {
    action resume(
       std::size_t bytes_read,
       system::error_code ec,
-      asio::cancellation_type_t /*cancel_state*/);
+      asio::cancellation_type_t cancel_state);
 
 private:
    int resume_point_{0};
-   action action_after_resume_;
    action::type next_read_type_ = action::type::read_some;
    multiplexer* mpx_ = nullptr;
    std::pair<consume_result, std::size_t> res_{consume_result::needs_more, 0u};

include/boost/redis/detail/redis_stream.hpp

@@ -99,6 +99,9 @@ class redis_stream {
 
             if (obj.transport_ == transport_type::unix_socket) {
 #ifdef BOOST_ASIO_HAS_LOCAL_SOCKETS
+               // Discard any existing state
+               obj.unix_socket_.close(ec);
+
                // Directly connect to the socket
                BOOST_ASIO_CORO_YIELD
                obj.unix_socket_.async_connect(
@@ -119,7 +122,9 @@ class redis_stream {
             } else {
                // ssl::stream doesn't support being re-used. If we're to use
                // TLS and the stream has been used, re-create it.
-               // Must be done before anything else is done on the stream
+               // Must be done before anything else is done on the stream.
+               // Note that we don't need to close the socket here because
+               // range connect does it for us.
                if (cfg->use_ssl && obj.ssl_stream_used_)
                   obj.reset_stream();
 
@@ -269,19 +274,11 @@ class redis_stream {
       }
    }
 
-   // Cleanup
+   // Cancels resolve operations. Resolve operations don't support per-operation
+   // cancellation, but resolvers have a cancel() function. Resolve operations are
+   // in general blocking and run in a separate thread. cancel() has effect only
+   // if the operation hasn't started yet. Still, trying is better than nothing
    void cancel_resolve() { resolv_.cancel(); }
-
-   void close()
-   {
-      system::error_code ec;
-      if (stream_.next_layer().is_open())
-         stream_.next_layer().close(ec);
-#ifdef BOOST_ASIO_HAS_LOCAL_SOCKETS
-      if (unix_socket_.is_open())
-         unix_socket_.close(ec);
-#endif
-   }
 };
 
 }  // namespace detail

include/boost/redis/impl/connection.ipp

@@ -21,6 +21,33 @@ logger detail::make_stderr_logger(logger::level lvl, std::string prefix)
    });
 }
 
+system::error_code detail::translate_parallel_group_errors(
+   std::array<std::size_t, 4u> order,
+   system::error_code setup_ec,
+   system::error_code health_check_ec,
+   system::error_code reader_ec,
+   system::error_code writer_ec)
+{
+   // The setup request is special: it might complete successfully,
+   // without causing the other tasks to exit.
+   // The other tasks will always complete with an error.
+
+   // If the setup task errored and was the first to exit, use its code
+   if (order[0] == 0u && setup_ec) {
+      return setup_ec;
+   }
+
+   // Use the code corresponding to the task that finished first,
+   // excluding the setup task
+   std::size_t task_number = order[0] == 0u ? order[1] : order[0];
+   switch (task_number) {
+      case 1u: return health_check_ec;
+      case 2u: return reader_ec;
+      case 3u: return writer_ec;
+      default: BOOST_ASSERT(false); return system::error_code();
+   }
+}
+
 connection::connection(executor_type ex, asio::ssl::context ctx, logger lgr)
 : impl_{std::move(ex), std::move(ctx), std::move(lgr)}
 { }

include/boost/redis/impl/connection_logger.ipp

@@ -25,11 +25,9 @@ namespace boost::redis::detail {
 auto to_string(reader_fsm::action::type t) noexcept -> char const*
 {
    switch (t) {
-      BOOST_REDIS_READER_SWITCH_CASE(setup_cancellation);
       BOOST_REDIS_READER_SWITCH_CASE(read_some);
       BOOST_REDIS_READER_SWITCH_CASE(needs_more);
       BOOST_REDIS_READER_SWITCH_CASE(notify_push_receiver);
-      BOOST_REDIS_READER_SWITCH_CASE(cancel_run);
       BOOST_REDIS_READER_SWITCH_CASE(done);
       default: return "action::type::<invalid type>";
    }

include/boost/redis/impl/reader_fsm.ipp

@@ -8,49 +8,63 @@
 #include <boost/redis/detail/multiplexer.hpp>
 #include <boost/redis/detail/reader_fsm.hpp>
 
+#include <boost/asio/cancellation_type.hpp>
+#include <boost/asio/error.hpp>
+
 namespace boost::redis::detail {
 
+inline bool is_terminal_cancel(asio::cancellation_type_t value)
+{
+   return (value & asio::cancellation_type_t::terminal) != asio::cancellation_type_t::none;
+}
+
 reader_fsm::reader_fsm(multiplexer& mpx) noexcept
 : mpx_{&mpx}
 { }
 
 reader_fsm::action reader_fsm::resume(
    std::size_t bytes_read,
    system::error_code ec,
-   asio::cancellation_type_t /*cancel_state*/)
+   asio::cancellation_type_t cancel_state)
 {
    switch (resume_point_) {
       BOOST_REDIS_CORO_INITIAL
-      BOOST_REDIS_YIELD(resume_point_, 1, action::type::setup_cancellation)
 
       for (;;) {
+         // Prepare the buffer for the read operation
          ec = mpx_->prepare_read();
          if (ec) {
-            action_after_resume_ = {action::type::done, 0, ec};
-            BOOST_REDIS_YIELD(resume_point_, 2, action::type::cancel_run)
-            return action_after_resume_;
+            return {action::type::done, 0, ec};
          }
 
-         BOOST_REDIS_YIELD(resume_point_, 3, next_read_type_)
+         // Read
+         BOOST_REDIS_YIELD(resume_point_, 1, next_read_type_)
+
+         // Process the bytes read, even if there was an error
          mpx_->commit_read(bytes_read);
+
+         // Check for read errors
          if (ec) {
             // TODO: If an error occurred but data was read (i.e.
             // bytes_read != 0) we should try to process that data and
             // deliver it to the user before calling cancel_run.
-            action_after_resume_ = {action::type::done, bytes_read, ec};
-            BOOST_REDIS_YIELD(resume_point_, 4, action::type::cancel_run)
-            return action_after_resume_;
+            return {action::type::done, bytes_read, ec};
+         }
+
+         // Check for cancellations
+         if (is_terminal_cancel(cancel_state)) {
+            return {action::type::done, 0u, asio::error::operation_aborted};
          }
 
+         // Process the data that we've read
          next_read_type_ = action::type::read_some;
          while (mpx_->get_read_buffer_size() != 0) {
             res_ = mpx_->consume(ec);
+
             if (ec) {
                // TODO: Perhaps log what has not been consumed to aid
                // debugging.
-               action_after_resume_ = {action::type::done, res_.second, ec};
-               BOOST_REDIS_YIELD(resume_point_, 5, action::type::cancel_run)
-               return action_after_resume_;
+               return {action::type::done, res_.second, ec};
             }
 
             if (res_.first == consume_result::needs_more) {
@@ -59,11 +73,12 @@ reader_fsm::action reader_fsm::resume(
             }
 
             if (res_.first == consume_result::got_push) {
-               BOOST_REDIS_YIELD(resume_point_, 6, action::type::notify_push_receiver, res_.second)
+               BOOST_REDIS_YIELD(resume_point_, 2, 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_;
+                  return {action::type::done, 0u, ec};
+               }
+               if (is_terminal_cancel(cancel_state)) {
+                  return {action::type::done, 0u, asio::error::operation_aborted};
                }
             } else {
                // TODO: Here we should notify the exec operation that

include/boost/redis/operation.hpp

@@ -16,22 +16,68 @@ namespace boost::redis {
  */
 enum class operation
 {
-   /// Resolve operation.
+   /**
+    * @brief (Deprecated) Resolve operation.
+    * 
+    * Cancelling a single resolve operation is probably not what you
+    * want, since there is no way to detect when a connection is performing name resolution.
+    * Use @ref operation::run to cancel the current @ref basic_connection::async_run operation,
+    * which includes name resolution.
+    */
    resolve,
-   /// Connect operation.
+
+   /**
+    * @brief (Deprecated) Connect operation.
+    * 
+    * Cancelling a single connect operation is probably not what you
+    * want, since there is no way to detect when a connection is performing a connect operation.
+    * Use @ref operation::run  to cancel the current @ref basic_connection::async_run operation,
+    * which includes connection establishment.
+    */
    connect,
-   /// SSL handshake operation.
+
+   /**
+    * @brief (Deprecated) SSL handshake operation.
+    * 
+    * Cancelling a single connect operation is probably not what you
+    * want, since there is no way to detect when a connection is performing an SSL handshake.
+    * Use @ref operation::run  to cancel the current @ref basic_connection::async_run operation,
+    * which includes the SSL handshake.
+    */
    ssl_handshake,
+
    /// Refers to `connection::async_exec` operations.
    exec,
+
    /// Refers to `connection::async_run` operations.
    run,
+
    /// Refers to `connection::async_receive` operations.
    receive,
-   /// Cancels reconnection.
+
+   /**
+    * @brief (Deprecated) Cancels reconnection.
+    * 
+    * Cancelling reconnection doesn't really cancel anything.
+    * It will only prevent further connections attempts from being
+    * made once the current connection encounters an error.
+    *
+    * Use @ref operation::run  to cancel the current @ref basic_connection::async_run operation,
+    * which includes reconnection. If you want to disable reconnection completely,
+    * set @ref config::reconnect_wait_interval to zero before calling `async_run`.
+    */
    reconnection,
-   /// Health check operation.
+
+   /**
+    * @brief (Deprecated) Health check operation.
+    * 
+    * Cancelling the health checker only is probably not what you want.
+    * Use @ref operation::run  to cancel the current @ref basic_connection::async_run operation,
+    * which includes the health checker. If you want to disable health checks completely,
+    * set @ref config::health_check_interval to zero before calling `async_run`.
+    */
    health_check,
+
    /// Refers to all operations.
    all,
 };

test/CMakeLists.txt

@@ -49,6 +49,7 @@ make_test(test_conn_quit)
 make_test(test_conn_exec_retry)
 make_test(test_conn_exec_error)
 make_test(test_run)
+make_test(test_conn_run_cancel)
 make_test(test_conn_check_health)
 make_test(test_conn_exec)
 make_test(test_conn_push)