asioexec::thread_unsafe_completion_token & _use_sender

For general purpose (i.e. potentially multithreaded) use the
asynchronous operations which result when passing the asioexec::
completion_token and ::use_sender completion tokens must use a recursive
mutex internally. However if the user knows that no multithreaded use
will occur this recursive mutex is pure overhead. Provided the
asioexec::thread_unsafe_completion_token and _use_sender completion
tokens which do not make use of a recursive mutex for the aforementioned
use case.

Author: RobertLeahy

include/asioexec/completion_token.hpp

@@ -136,10 +136,7 @@ namespace asioexec {
       ::STDEXEC::set_stopped_t()
     >;
 
-    template <typename, typename>
-    class completion_handler;
-
-    template <typename Signatures, typename Receiver>
+    template <typename Mutex, typename Signatures, typename Receiver>
     struct operation_state_base {
       class frame_;
 
@@ -153,7 +150,8 @@ namespace asioexec {
 
       Receiver r_;
       asio_impl::cancellation_signal signal_;
-      std::recursive_mutex m_;
+      [[no_unique_address]]
+      Mutex m_;
       frame_* frames_{nullptr};
       std::exception_ptr ex_;
       bool abandoned_{false};
@@ -251,11 +249,12 @@ namespace asioexec {
         callback_;
     };
 
-    template <typename Signatures, typename Receiver>
+    template <typename Mutex, typename Signatures, typename Receiver>
     class completion_handler {
-      operation_state_base<Signatures, Receiver>* self_;
+      using operation_state_type_ = operation_state_base<Mutex, Signatures, Receiver>;
+      operation_state_type_* self_;
      public:
-      explicit completion_handler(operation_state_base<Signatures, Receiver>& self) noexcept
+      explicit completion_handler(operation_state_type_& self) noexcept
         : self_(&self) {
       }
 
@@ -270,7 +269,7 @@ namespace asioexec {
           //  When this goes out of scope it might send set stopped or set error, or
           //  it might defer that to the executor frames above us on the call stack
           //  (if any)
-          const typename operation_state_base<Signatures, Receiver>::frame_ frame(*self_);
+          const typename operation_state_type_::frame_ frame(*self_);
           self_->abandoned_ = true;
         }
       }
@@ -309,15 +308,20 @@ namespace asioexec {
         return self_->signal_.slot();
       }
 
-      operation_state_base<Signatures, Receiver>& state() const noexcept {
+      operation_state_type_& state() const noexcept {
         STDEXEC_ASSERT(self_);
         return *self_;
       }
     };
 
-    template <typename Signatures, typename Receiver, typename Initiation, typename Args>
-    class operation_state : operation_state_base<Signatures, Receiver> {
-      using base_ = operation_state_base<Signatures, Receiver>;
+    template <
+      typename Mutex,
+      typename Signatures,
+      typename Receiver,
+      typename Initiation,
+      typename Args>
+    class operation_state : operation_state_base<Mutex, Signatures, Receiver> {
+      using base_ = operation_state_base<Mutex, Signatures, Receiver>;
       Initiation init_;
       Args args_;
      public:
@@ -339,7 +343,7 @@ namespace asioexec {
             [&](auto&&... args) {
               std::invoke(
                 static_cast<Initiation&&>(init_),
-                completion_handler<Signatures, Receiver>(*this),
+                completion_handler<Mutex, Signatures, Receiver>(*this),
                 static_cast<decltype(args)&&>(args)...);
             },
             std::move(args_));
@@ -362,9 +366,12 @@ namespace asioexec {
       }
     };
 
-    template <typename Signatures, typename Initiation, typename... Args>
+    template <typename Mutex, typename Signatures, typename Initiation, typename... Args>
     class sender {
       using args_type_ = std::tuple<std::decay_t<Args>...>;
+      template <typename Receiver>
+      using operation_state_type_ =
+        operation_state<Mutex, Signatures, std::remove_cvref_t<Receiver>, Initiation, args_type_>;
      public:
       using sender_concept = ::STDEXEC::sender_t;
 
@@ -388,32 +395,25 @@ namespace asioexec {
       template <typename Receiver>
         requires ::STDEXEC::receiver_of<
           std::remove_cvref_t<Receiver>,
-          ::STDEXEC::completion_signatures_of_t<const sender&, ::STDEXEC::env_of_t<Receiver>>
-        >
-      constexpr auto connect(Receiver&& receiver) const & noexcept(
-        std::is_nothrow_constructible_v<
-          operation_state<Signatures, std::remove_cvref_t<Receiver>, Initiation, args_type_>,
-          Receiver,
-          const Initiation&,
-          const args_type_&
-        >) {
-        return operation_state<Signatures, std::remove_cvref_t<Receiver>, Initiation, args_type_>(
-          static_cast<Receiver&&>(receiver), init_, args_);
+          ::STDEXEC::completion_signatures_of_t<const sender&, ::STDEXEC::env_of_t<Receiver>>>
+      constexpr auto connect(Receiver&& receiver) const & noexcept(std::is_nothrow_constructible_v<
+                                                                   operation_state_type_<Receiver>,
+                                                                   Receiver,
+                                                                   const Initiation&,
+                                                                   const args_type_&>) {
+        return operation_state_type_<Receiver>(static_cast<Receiver&&>(receiver), init_, args_);
       }
 
       template <typename Receiver>
         requires ::STDEXEC::receiver_of<
           std::remove_cvref_t<Receiver>,
-          ::STDEXEC::completion_signatures_of_t<sender, ::STDEXEC::env_of_t<Receiver>>
-        >
-      constexpr auto connect(Receiver&& receiver) && noexcept(
-        std::is_nothrow_constructible_v<
-          operation_state<Signatures, std::remove_cvref_t<Receiver>, Initiation, args_type_>,
-          Receiver,
-          Initiation,
-          args_type_
-        >) {
-        return operation_state<Signatures, std::remove_cvref_t<Receiver>, Initiation, args_type_>(
+          ::STDEXEC::completion_signatures_of_t<sender, ::STDEXEC::env_of_t<Receiver>>>
+      constexpr auto connect(Receiver&& receiver) && noexcept(std::is_nothrow_constructible_v<
+                                                              operation_state_type_<Receiver>,
+                                                              Receiver,
+                                                              Initiation,
+                                                              args_type_>) {
+        return operation_state_type_<Receiver>(
           static_cast<Receiver&&>(receiver),
           static_cast<Initiation&&>(init_),
           static_cast<args_type_&&>(args_));
@@ -423,9 +423,10 @@ namespace asioexec {
       args_type_ args_;
     };
 
-    template <typename Signatures, typename Receiver, typename Executor>
+    template <typename Mutex, typename Signatures, typename Receiver, typename Executor>
     class executor {
-      operation_state_base<Signatures, Receiver>& self_;
+      using operation_state_type_ = operation_state_base<Mutex, Signatures, Receiver>;
+      operation_state_type_& self_;
       Executor ex_;
 
       template <typename F>
@@ -435,9 +436,7 @@ namespace asioexec {
         };
       }
      public:
-      constexpr explicit executor(
-        operation_state_base<Signatures, Receiver>& self,
-        const Executor& ex) noexcept
+      constexpr explicit executor(operation_state_type_& self, const Executor& ex) noexcept
         : self_(self)
         , ex_(ex) {
       }
@@ -456,7 +455,7 @@ namespace asioexec {
         }
       constexpr decltype(auto) prefer(Args&&... args) const noexcept {
         const auto ex = asio_impl::prefer(ex_, static_cast<Args&&>(args)...);
-        return executor<Signatures, Receiver, std::remove_cvref_t<decltype(ex)>>(self_, ex);
+        return executor<Mutex, Signatures, Receiver, std::remove_cvref_t<decltype(ex)>>(self_, ex);
       }
 
       template <typename... Args>
@@ -465,7 +464,7 @@ namespace asioexec {
         }
       constexpr decltype(auto) require(Args&&... args) const noexcept {
         const auto ex = asio_impl::require(ex_, static_cast<Args&&>(args)...);
-        return executor<Signatures, Receiver, std::remove_cvref_t<decltype(ex)>>(self_, ex);
+        return executor<Mutex, Signatures, Receiver, std::remove_cvref_t<decltype(ex)>>(self_, ex);
       }
 
       template <typename T>
@@ -516,59 +515,78 @@ namespace asioexec {
       bool operator!=(const executor& rhs) const = default;
     };
 
+    template <typename Mutex>
+    struct t {
+      static constexpr auto as_default_on = ::asioexec::as_default_on<t>;
+      template <typename IoObject>
+      using as_default_on_t = ::asioexec::as_default_on_t<t, IoObject>;
+    };
+
+    struct null_basic_lockable {
+      constexpr void lock() noexcept {
+      }
+      constexpr void unlock() noexcept {
+      }
+    };
+
   } // namespace detail::completion_token
 
-  struct completion_token_t {
-    static constexpr auto as_default_on = asioexec::as_default_on<completion_token_t>;
-    template <typename IoObject>
-    using as_default_on_t = asioexec::as_default_on_t<completion_token_t, IoObject>;
-  };
+  using completion_token_t = detail::completion_token::t<std::recursive_mutex>;
 
   inline const completion_token_t completion_token{};
 
+  using thread_unsafe_completion_token_t =
+    detail::completion_token::t<detail::completion_token::null_basic_lockable>;
+
+  inline const thread_unsafe_completion_token_t thread_unsafe_completion_token{};
+
 } // namespace asioexec
 
 namespace ASIOEXEC_ASIO_NAMESPACE {
 
-  template <typename... Signatures>
-  struct async_result<::asioexec::completion_token_t, Signatures...> {
+  template <typename Mutex, typename... Signatures>
+  struct async_result<::asioexec::detail::completion_token::t<Mutex>, Signatures...> {
     template <typename Initiation, typename... Args>
       requires(std::is_constructible_v<std::decay_t<Args>, Args> && ...)
-    static constexpr auto
-      initiate(Initiation&& i, const ::asioexec::completion_token_t&, Args&&... args) {
+    static constexpr auto initiate(
+      Initiation&& i,
+      const ::asioexec::detail::completion_token::t<Mutex>&,
+      Args&&... args) {
       return ::asioexec::detail::completion_token::sender<
+        Mutex,
         ::asioexec::detail::completion_token::completion_signatures<Signatures...>,
         std::remove_cvref_t<Initiation>,
-        Args...
-      >(static_cast<Initiation&&>(i), static_cast<Args&&>(args)...);
+        Args...>(static_cast<Initiation&&>(i), static_cast<Args&&>(args)...);
     }
   };
 
-  template <typename Signatures, typename Receiver, typename Executor>
+  template <typename Mutex, typename Signatures, typename Receiver, typename Executor>
   struct associated_executor<
-    ::asioexec::detail::completion_token::completion_handler<Signatures, Receiver>,
-    Executor
-  > {
-    using type = ::asioexec::detail::completion_token::executor<Signatures, Receiver, Executor>;
+    ::asioexec::detail::completion_token::completion_handler<Mutex, Signatures, Receiver>,
+    Executor> {
+    using type =
+      ::asioexec::detail::completion_token::executor<Mutex, Signatures, Receiver, Executor>;
 
     static type get(
-      const ::asioexec::detail::completion_token::completion_handler<Signatures, Receiver>& h,
+      const ::asioexec::detail::completion_token::completion_handler<Mutex, Signatures, Receiver>&
+        h,
       const Executor& ex) noexcept {
       return type(h.state(), ex);
     }
   };
 
-  template <typename Signatures, typename Receiver, typename Allocator>
+  template <typename Mutex, typename Signatures, typename Receiver, typename Allocator>
     requires requires(const Receiver& r) { ::STDEXEC::get_allocator(::STDEXEC::get_env(r)); }
   struct associated_allocator<
-    ::asioexec::detail::completion_token::completion_handler<Signatures, Receiver>,
+    ::asioexec::detail::completion_token::completion_handler<Mutex, Signatures, Receiver>,
     Allocator
   > {
     using type = std::remove_cvref_t<decltype(::STDEXEC::get_allocator(
       ::STDEXEC::get_env(std::declval<const Receiver&>())))>;
 
     static type get(
-      const ::asioexec::detail::completion_token::completion_handler<Signatures, Receiver>& h,
+      const ::asioexec::detail::completion_token::completion_handler<Mutex, Signatures, Receiver>&
+        h,
       const Allocator&) noexcept {
       return ::STDEXEC::get_allocator(::STDEXEC::get_env(h.state().r_));
     }

include/asioexec/use_sender.hpp

@@ -23,6 +23,7 @@
 #include <asioexec/completion_token.hpp>
 #include <concepts>
 #include <exception>
+#include <mutex>
 #include <stdexec/execution.hpp>
 #include <system_error>
 #include <type_traits>
@@ -190,30 +191,38 @@ namespace asioexec {
     template <typename Sender>
     explicit sender(Sender) -> sender<Sender>;
 
+    template <typename Mutex>
+    struct t {
+      static constexpr auto as_default_on = ::asioexec::as_default_on<t>;
+      template <typename IoObject>
+      using as_default_on_t = ::asioexec::as_default_on_t<t, IoObject>;
+    };
+
   } // namespace detail::use_sender
 
-  struct use_sender_t {
-    static constexpr auto as_default_on = asioexec::as_default_on<use_sender_t>;
-    template <typename IoObject>
-    using as_default_on_t = asioexec::as_default_on_t<use_sender_t, IoObject>;
-  };
+  using use_sender_t = detail::use_sender::t<std::recursive_mutex>;
 
   inline const use_sender_t use_sender{};
 
+  using thread_unsafe_use_sender_t =
+    detail::use_sender::t<detail::completion_token::null_basic_lockable>;
+
+  inline const thread_unsafe_use_sender_t thread_unsafe_use_sender{};
+
 } // namespace asioexec
 
 namespace ASIOEXEC_ASIO_NAMESPACE {
 
-  template <typename... Signatures>
-  struct async_result<::asioexec::use_sender_t, Signatures...> {
+  template <typename Mutex, typename... Signatures>
+  struct async_result<::asioexec::detail::use_sender::t<Mutex>, Signatures...> {
     template <typename Initiation, typename... Args>
       requires(std::is_constructible_v<std::decay_t<Args>, Args> && ...)
     static constexpr auto
-      initiate(Initiation&& i, const ::asioexec::use_sender_t&, Args&&... args) {
+      initiate(Initiation&& i, const ::asioexec::detail::use_sender::t<Mutex>&, Args&&... args) {
       return ::asioexec::detail::use_sender::sender(
-        async_result<::asioexec::completion_token_t, Signatures...>::initiate(
+        async_result<::asioexec::detail::completion_token::t<Mutex>, Signatures...>::initiate(
           static_cast<Initiation&&>(i),
-          ::asioexec::completion_token,
+          ::asioexec::detail::completion_token::t<Mutex>{},
           static_cast<Args&&>(args)...));
     }
   };

test/asioexec/test_completion_token.cpp

@@ -162,7 +162,7 @@ namespace {
     asio_impl::io_context ctx;
     asio_impl::system_timer t(ctx);
     t.expires_after(std::chrono::years(1));
-    auto sender = t.async_wait(completion_token);
+    auto sender = t.async_wait(thread_unsafe_completion_token);
     static_assert(set_equivalent<
                   completion_signatures_of_t<decltype(sender), env<>>,
                   completion_signatures<

test/asioexec/test_use_sender.cpp

@@ -82,7 +82,7 @@ namespace {
     asio_impl::io_context ctx;
     asio_impl::system_timer t(ctx);
     t.expires_after(std::chrono::years(1));
-    auto sender = t.async_wait(use_sender);
+    auto sender = t.async_wait(thread_unsafe_use_sender);
     static_assert(::STDEXEC::sender_in<decltype(sender)>);
     static_assert(
       ::STDEXEC::sender_of<