async_exec now uses a sans-io strategy (#250)

async_exec now supports partial and total cancellation when the request is pending
Added docs on async_exec's cancellation support
Added detail::exec_fsm and macros for coroutine emulation
Requests are now removed from the multiplexer when an error is encountered

Author: anarthal

include/boost/redis/connection.hpp

@@ -10,6 +10,7 @@
 #include <boost/redis/adapter/adapt.hpp>
 #include <boost/redis/adapter/any_adapter.hpp>
 #include <boost/redis/config.hpp>
+#include <boost/redis/detail/exec_fsm.hpp>
 #include <boost/redis/detail/health_checker.hpp>
 #include <boost/redis/detail/helper.hpp>
 #include <boost/redis/detail/multiplexer.hpp>
@@ -24,7 +25,6 @@
 
 #include <boost/asio/any_completion_handler.hpp>
 #include <boost/asio/any_io_executor.hpp>
-#include <boost/asio/associated_immediate_executor.hpp>
 #include <boost/asio/basic_stream_socket.hpp>
 #include <boost/asio/bind_executor.hpp>
 #include <boost/asio/buffer.hpp>
@@ -33,6 +33,7 @@
 #include <boost/asio/deferred.hpp>
 #include <boost/asio/experimental/channel.hpp>
 #include <boost/asio/experimental/parallel_group.hpp>
+#include <boost/asio/immediate.hpp>
 #include <boost/asio/io_context.hpp>
 #include <boost/asio/ip/tcp.hpp>
 #include <boost/asio/prepend.hpp>
@@ -110,68 +111,41 @@ using exec_notifier_type = asio::experimental::channel<
 
 template <class Conn>
 struct exec_op {
-   using req_info_type = typename multiplexer::elem;
    using executor_type = typename Conn::executor_type;
 
    Conn* conn_ = nullptr;
    std::shared_ptr<exec_notifier_type<executor_type>> notifier_ = nullptr;
-   std::shared_ptr<req_info_type> info_ = nullptr;
-   asio::coroutine coro_{};
+   detail::exec_fsm fsm_;
 
    template <class Self>
    void operator()(Self& self, system::error_code = {}, std::size_t = 0)
    {
-      BOOST_ASIO_CORO_REENTER(coro_)
-      {
-         // Check whether the user wants to wait for the connection to
-         // be stablished.
-         if (info_->get_request().get_config().cancel_if_not_connected && !conn_->is_open()) {
-            BOOST_ASIO_CORO_YIELD
-            asio::dispatch(
-               asio::get_associated_immediate_executor(self, self.get_io_executor()),
-               std::move(self));
-            return self.complete(error::not_connected, 0);
-         }
-
-         conn_->mpx_.add(info_);
-         if (conn_->trigger_write()) {
-            conn_->writer_timer_.cancel();
-         }
-
-EXEC_OP_WAIT:
-         BOOST_ASIO_CORO_YIELD
-         notifier_->async_receive(std::move(self));
-
-         if (info_->get_error()) {
-            self.complete(info_->get_error(), 0);
-            return;
-         }
-
-         if (is_cancelled(self)) {
-            if (!conn_->mpx_.remove(info_)) {
-               using c_t = asio::cancellation_type;
-               auto const c = self.get_cancellation_state().cancelled();
-               if ((c & c_t::terminal) != c_t::none) {
-                  // Cancellation requires closing the connection
-                  // otherwise it stays in inconsistent state.
-                  conn_->cancel(operation::run);
-                  return self.complete(asio::error::operation_aborted, 0);
-               } else {
-                  // Can't implement other cancelation types, ignoring.
-                  self.get_cancellation_state().clear();
-
-                  // TODO: Find out a better way to ignore
-                  // cancelation.
-                  goto EXEC_OP_WAIT;
-               }
-            } else {
-               // Cancelation honored.
-               self.complete(asio::error::operation_aborted, 0);
+      while (true) {
+         // Invoke the state machine
+         auto act = fsm_.resume(conn_->is_open(), self.get_cancellation_state().cancelled());
+
+         // Do what the FSM said
+         switch (act.type()) {
+            case detail::exec_action_type::setup_cancellation:
+               self.reset_cancellation_state(asio::enable_total_cancellation());
+               continue;  // this action does not require yielding
+            case detail::exec_action_type::immediate:
+               asio::async_immediate(self.get_io_executor(), std::move(self));
+               return;
+            case detail::exec_action_type::notify_writer:
+               conn_->writer_timer_.cancel();
+               continue;  // this action does not require yielding
+            case detail::exec_action_type::wait_for_response:
+               notifier_->async_receive(std::move(self));
+               return;
+            case detail::exec_action_type::cancel_run:
+               conn_->cancel(operation::run);
+               continue;  // this action does not require yielding
+            case detail::exec_action_type::done:
+               notifier_.reset();
+               self.complete(act.error(), act.bytes_read());
                return;
-            }
          }
-
-         self.complete(info_->get_error(), info_->get_read_size());
       }
    }
 };
@@ -612,6 +586,16 @@ class basic_connection {
     *
     *  Where the second parameter is the size of the response received
     *  in bytes.
+    *
+    * @par Per-operation cancellation
+    * This operation supports per-operation cancellation. The following cancellation types
+    * are supported:
+    *
+    *   - `asio::cancellation_type_t::terminal`. Always supported. May cause the current
+    *     `async_run` operation to be cancelled.
+    *   - `asio::cancellation_type_t::partial` and `asio::cancellation_type_t::total`.
+    *     Supported only if the request hasn't been written to the network yet.
+    *
     */
    template <
       class Response = ignore_t,
@@ -644,7 +628,7 @@ class basic_connection {
       });
 
       return asio::async_compose<CompletionToken, void(system::error_code, std::size_t)>(
-         detail::exec_op<this_type>{this, notifier, info},
+         detail::exec_op<this_type>{this, notifier, detail::exec_fsm(mpx_, std::move(info))},
          token,
          writer_timer_);
    }

include/boost/redis/detail/coroutine.hpp

@@ -0,0 +1,36 @@
+//
+// Copyright (c) 2025 Marcelo Zimbres Silva ([email protected]),
+// Ruben Perez Hidalgo (rubenperez038 at gmail dot com)
+//
+// Distributed under the Boost Software License, Version 1.0. (See accompanying
+// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
+//
+
+#ifndef BOOST_REDIS_DETAIL_COROUTINE_HPP
+#define BOOST_REDIS_DETAIL_COROUTINE_HPP
+
+// asio::coroutine uses __COUNTER__ internally, which can trigger
+// ODR violations if we use them in header-only code. These manifest as
+// extremely hard-to-debug bugs only present in release builds.
+// Use this instead when doing coroutines in non-template code.
+// Adapted from Boost.MySQL.
+
+// Coroutine state is represented as an integer (resume_point_var).
+// Every yield gets assigned a unique value (resume_point_id).
+// Yielding sets the next resume point, returns, and sets a case label for re-entering.
+// Coroutines need to switch on resume_point_var to re-enter.
+
+// Enclosing this in a scope allows placing the macro inside a brace-less for/while loop
+// The empty scope after the case label is required because labels can't be at the end of a compound statement
+#define BOOST_REDIS_YIELD(resume_point_var, resume_point_id, ...) \
+   {                                                              \
+      resume_point_var = resume_point_id;                         \
+      return __VA_ARGS__;                                         \
+      case resume_point_id:                                       \
+      {                                                           \
+      }                                                           \
+   }
+
+#define BOOST_REDIS_CORO_INITIAL case 0:
+
+#endif

include/boost/redis/detail/exec_fsm.hpp

@@ -0,0 +1,72 @@
+//
+// Copyright (c) 2025 Marcelo Zimbres Silva ([email protected]),
+// Ruben Perez Hidalgo (rubenperez038 at gmail dot com)
+//
+// Distributed under the Boost Software License, Version 1.0. (See accompanying
+// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
+//
+
+#ifndef BOOST_REDIS_EXEC_FSM_HPP
+#define BOOST_REDIS_EXEC_FSM_HPP
+
+#include <boost/redis/detail/multiplexer.hpp>
+
+#include <boost/asio/cancellation_type.hpp>
+#include <boost/system/error_code.hpp>
+
+#include <cstddef>
+#include <memory>
+
+// Sans-io algorithm for async_exec, as a finite state machine
+
+namespace boost::redis::detail {
+
+// What should we do next?
+enum class exec_action_type
+{
+   setup_cancellation,  // Set up the cancellation types supported by the composed operation
+   immediate,           // Invoke asio::async_immediate to avoid re-entrancy problems
+   done,                // Call the final handler
+   notify_writer,       // Notify the writer task
+   wait_for_response,   // Wait to be notified
+   cancel_run,          // Cancel the connection's run operation
+};
+
+class exec_action {
+   exec_action_type type_;
+   system::error_code ec_;
+   std::size_t bytes_read_;
+
+public:
+   exec_action(exec_action_type type) noexcept
+   : type_{type}
+   { }
+
+   exec_action(system::error_code ec, std::size_t bytes_read = 0u) noexcept
+   : type_{exec_action_type::done}
+   , ec_{ec}
+   , bytes_read_{bytes_read}
+   { }
+
+   exec_action_type type() const { return type_; }
+   system::error_code error() const { return ec_; }
+   std::size_t bytes_read() const { return bytes_read_; }
+};
+
+class exec_fsm {
+   int resume_point_{0};
+   multiplexer* mpx_{nullptr};
+   std::shared_ptr<multiplexer::elem> elem_;
+
+public:
+   exec_fsm(multiplexer& mpx, std::shared_ptr<multiplexer::elem> elem) noexcept
+   : mpx_(&mpx)
+   , elem_(std::move(elem))
+   { }
+
+   exec_action resume(bool connection_is_open, asio::cancellation_type_t cancel_state);
+};
+
+}  // namespace boost::redis::detail
+
+#endif  // BOOST_REDIS_CONNECTOR_HPP

include/boost/redis/detail/multiplexer.hpp

@@ -43,7 +43,11 @@ struct multiplexer {
 
       void set_done_callback(std::function<void()> f) noexcept { done_ = std::move(f); };
 
-      auto notify_done() noexcept { done_(); }
+      auto notify_done() noexcept -> void
+      {
+         status_ = status::done;
+         done_();
+      }
 
       auto notify_error(system::error_code ec) noexcept -> void;
 
@@ -65,6 +69,12 @@ struct multiplexer {
          return status_ == status::staged;
       }
 
+      [[nodiscard]]
+      bool is_done() const noexcept
+      {
+         return status_ == status::done;
+      }
+
       void mark_written() noexcept { status_ = status::written; }
 
       void mark_staged() noexcept { status_ = status::staged; }
@@ -86,9 +96,10 @@ struct multiplexer {
    private:
       enum class status
       {
-         waiting,
-         staged,
-         written
+         waiting,  // the request hasn't been written yet
+         staged,   // we've issued the write for this request, but it hasn't finished yet
+         written,  // the request has been written successfully
+         done,     // the request has completed and the done callback has been invoked
       };
 
       request const* req_;

include/boost/redis/impl/exec_fsm.ipp

@@ -0,0 +1,94 @@
+//
+// Copyright (c) 2025 Marcelo Zimbres Silva ([email protected]),
+// Ruben Perez Hidalgo (rubenperez038 at gmail dot com)
+//
+// Distributed under the Boost Software License, Version 1.0. (See accompanying
+// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
+//
+
+#ifndef BOOST_REDIS_EXEC_FSM_IPP
+#define BOOST_REDIS_EXEC_FSM_IPP
+
+#include <boost/redis/detail/coroutine.hpp>
+#include <boost/redis/detail/exec_fsm.hpp>
+#include <boost/redis/request.hpp>
+
+#include <boost/asio/error.hpp>
+#include <boost/assert.hpp>
+
+namespace boost::redis::detail {
+
+inline bool is_cancellation(asio::cancellation_type_t type)
+{
+   return !!(
+      type & (asio::cancellation_type_t::total | asio::cancellation_type_t::partial |
+              asio::cancellation_type_t::terminal));
+}
+
+}  // namespace boost::redis::detail
+
+boost::redis::detail::exec_action boost::redis::detail::exec_fsm::resume(
+   bool connection_is_open,
+   asio::cancellation_type_t cancel_state)
+{
+   switch (resume_point_) {
+      BOOST_REDIS_CORO_INITIAL
+
+      // Check whether the user wants to wait for the connection to
+      // be established.
+      if (elem_->get_request().get_config().cancel_if_not_connected && !connection_is_open) {
+         BOOST_REDIS_YIELD(resume_point_, 1, exec_action_type::immediate)
+         elem_.reset();  // Deallocate memory before finalizing
+         return system::error_code(error::not_connected);
+      }
+
+      // No more immediate errors. Set up the supported cancellation types.
+      // This is required to get partial and total cancellations.
+      // This is a potentially allocating operation, so do it as late as we can.
+      BOOST_REDIS_YIELD(resume_point_, 2, exec_action_type::setup_cancellation)
+
+      // Add the request to the multiplexer
+      mpx_->add(elem_);
+
+      // Notify the writer task that there is work to do. If the task is not
+      // listening (e.g. it's already writing or the connection is not healthy),
+      // this is a no-op. Since this is sync, no cancellation can happen here.
+      BOOST_REDIS_YIELD(resume_point_, 3, exec_action_type::notify_writer)
+
+      while (true) {
+         // Wait until we get notified. This will return once the request completes,
+         // or upon any kind of cancellation
+         BOOST_REDIS_YIELD(resume_point_, 4, exec_action_type::wait_for_response)
+
+         // If the request has completed (with error or not), we're done
+         if (elem_->is_done()) {
+            exec_action act{elem_->get_error(), elem_->get_read_size()};
+            elem_.reset();  // Deallocate memory before finalizing
+            return act;
+         }
+
+         // If we're cancelled, try to remove the request from the queue. This will only
+         // succeed if the request is waiting (wasn't written yet)
+         if (is_cancellation(cancel_state) && mpx_->remove(elem_)) {
+            elem_.reset();  // Deallocate memory before finalizing
+            return exec_action{asio::error::operation_aborted};
+         }
+
+         // If we hit a terminal cancellation, tear down the connection.
+         // Otherwise, go back to waiting.
+         // TODO: we could likely do better here and mark the request as cancelled, removing
+         // the done callback and the adapter. But this requires further exploration
+         if (!!(cancel_state & asio::cancellation_type_t::terminal)) {
+            BOOST_REDIS_YIELD(resume_point_, 5, exec_action_type::cancel_run)
+            elem_.reset();  // Deallocate memory before finalizing
+            return exec_action{asio::error::operation_aborted};
+         }
+      }
+   }
+
+   // We should never get here
+   BOOST_ASSERT(false);
+   return exec_action{system::error_code()};
+}
+
+#endif