add a customizable algorithm submit that combines connect and start

Author: ericniebler

include/exec/async_scope.hpp

@@ -593,12 +593,13 @@ namespace exec {
       __future_state(_Sender __sndr, _Env __env, const __impl* __scope)
         : __future_state_base<_Completions, _Env>(static_cast<_Env&&>(__env), __scope)
         , __op_(
-            stdexec::connect(
+            stdexec::submit(
               static_cast<_Sender&&>(__sndr),
-              __future_receiver_t<_Sender, _Env>{this, __scope})) {
+              __future_receiver_t<_Sender, _Env>{this, __scope},
+              __ignore{})) {
       }
 
-      connect_result_t<_Sender, __future_receiver_t<_Sender, _Env>> __op_;
+      STDEXEC_ATTRIBUTE((no_unique_address)) submit_result_t<_Sender, __future_receiver_t<_Sender, _Env>, __ignore> __op_;
     };
 
     template <class _SenderId, class _EnvId>
@@ -736,14 +737,10 @@ namespace exec {
             [](__spawn_op_base<_EnvId>* __op) {
                 delete static_cast<__t*>(__op);
             }}
-          , __op_(stdexec::connect(static_cast<_Sndr&&>(__sndr), __spawn_receiver_t<_Env>{this})) {
+          , __data_(stdexec::submit(static_cast<_Sndr&&>(__sndr), __spawn_receiver_t<_Env>{this}, __ignore{})) {
         }
 
-        void start() & noexcept {
-          stdexec::start(__op_);
-        }
-
-        connect_result_t<_Sender, __spawn_receiver_t<_Env>> __op_;
+        STDEXEC_ATTRIBUTE((no_unique_address)) submit_result_t<_Sender, __spawn_receiver_t<_Env>, __ignore> __data_;
       };
     };
 
@@ -779,19 +776,18 @@ namespace exec {
         requires sender_to<nest_result_t<_Sender>, __spawn_receiver_t<_Env>>
       void spawn(_Sender&& __sndr, _Env __env = {}) {
         using __op_t = __spawn_operation_t<nest_result_t<_Sender>, _Env>;
-        // start is noexcept so we can assume that the operation will complete
-        // after this, which means we can rely on its self-ownership to ensure
-        // that it is eventually deleted
-        stdexec::start(*(
-          new __op_t{nest(static_cast<_Sender&&>(__sndr)), static_cast<_Env&&>(__env), &__impl_}));
+        // this will connect and start the operation, after which the operation state is
+        // responsible for deleting itself after it completes.
+        [[maybe_unused]]
+        auto* __op =
+          new __op_t{nest(static_cast<_Sender&&>(__sndr)), static_cast<_Env&&>(__env), &__impl_};
       }
 
       template <__movable_value _Env = env<>, sender_in<__env_t<_Env>> _Sender>
       auto spawn_future(_Sender&& __sndr, _Env __env = {}) -> __future_t<_Sender, _Env> {
         using __state_t = __future_state<nest_result_t<_Sender>, _Env>;
         auto __state = std::make_unique<__state_t>(
           nest(static_cast<_Sender&&>(__sndr)), static_cast<_Env&&>(__env), &__impl_);
-        stdexec::start(__state->__op_);
         return __future_t<_Sender, _Env>{std::move(__state)};
       }
 

include/nvexec/stream/start_detached.cuh

@@ -28,14 +28,40 @@
 
 namespace nvexec::_strm {
   namespace _start_detached {
+    struct submit_receiver {
+      using receiver_concept = receiver_t;
+      using __t = submit_receiver;
+      using __id = submit_receiver;
+
+      template <class... _As>
+      void set_value(_As&&...) noexcept {
+      }
+
+      template <class _Error>
+      [[noreturn]]
+      void set_error(_Error&&) noexcept {
+        // A detached operation failed. There is noplace for the error to go.
+        // This is unrecoverable, so we terminate.
+        std::terminate();
+      }
+
+      void set_stopped() noexcept {
+      }
+
+      [[nodiscard]]
+      auto get_env() const noexcept -> __root_env {
+        return {};
+      }
+    };
+
     template <class SenderId, class EnvId>
     struct operation {
       using Sender = __cvref_t<SenderId>;
       using Env = __t<EnvId>;
 
       explicit operation(Sender&& sndr, Env env)
         : env_(static_cast<Env&&>(env))
-        , op_state_(connect(static_cast<Sender&&>(sndr), receiver{{}, this})) {
+        , op_data_(submit(static_cast<Sender&&>(sndr), receiver{{}, this}, __ignore{})) {
       }
 
       // If the operation state was allocated with a user-provided allocator, then we must
@@ -85,7 +111,7 @@ namespace nvexec::_strm {
       };
 
       STDEXEC_ATTRIBUTE((no_unique_address)) Env env_;
-      connect_result_t<Sender, receiver> op_state_;
+      STDEXEC_ATTRIBUTE((no_unique_address)) submit_result_t<Sender, receiver> op_data_;
     };
   } // namespace _start_detached
 
@@ -94,8 +120,16 @@ namespace nvexec::_strm {
     template <class Sender, class Env = __root_env>
     void operator()(Sender&& sndr, Env&& env = {}) const noexcept(false) {
       using Op = _start_detached::operation<__cvref_id<Sender>, __id<__decay_t<Env>>>;
-      // Use the provided allocator, if any, to allocate the operation state.
-      if constexpr (__callable<get_allocator_t, Env>) {
+
+      // NON-STANDARD
+      if constexpr (
+        __same_as<Env, __root_env>
+        && __same_as<void, submit_result_t<Sender, _start_detached::submit_receiver>>) {
+        // If submit(sndr, rcvr) returns void, then no state needs to be kept alive
+        // for the operation. We can just call submit and return.
+        stdexec::submit(static_cast<Sender&&>(sndr), _start_detached::submit_receiver{});
+      } else if constexpr (__callable<get_allocator_t, Env>) {
+        // Use the provided allocator to allocate the operation state.
         auto alloc = get_allocator(env);
         using Alloc = decltype(alloc);
         using OpAlloc = typename std::allocator_traits<Alloc>::template rebind_alloc<Op>;
@@ -111,12 +145,12 @@ namespace nvexec::_strm {
           op_alloc, op, static_cast<Sender&&>(sndr), static_cast<Env&&>(env));
         // The operation state is now constructed, dismiss the scope guard.
         g.dismiss();
-        // This cannot throw:
-        stdexec::start(op->op_state_);
       } else {
         // The caller did not provide an allocator, so we use the default allocator.
+        [[maybe_unused]]
         Op* op = new Op{static_cast<Sender&&>(sndr), static_cast<Env&&>(env)};
-        start(op->op_state_);
+        // The operation has now started and is responsible for deleting itself when it
+        // completes.
       }
     }
   };

include/stdexec/__detail/__basic_sender.hpp

@@ -509,6 +509,15 @@ namespace stdexec {
           static_cast<_Self&&>(__self), static_cast<_Receiver&&>(__rcvr));
       }
 
+      template <__decays_to<__sexpr> _Self, /*receiver*/ class _Receiver>
+      STDEXEC_ATTRIBUTE((always_inline)) static auto submit(_Self&& __self, _Receiver&& __rcvr)        //
+        noexcept(__noexcept_of<__impl<_Self>::submit, _Self, _Receiver>) //
+        -> __msecond<
+          __if_c<__decays_to<_Self, __sexpr>>,
+          __result_of<__impl<_Self>::submit, _Self, _Receiver>> {
+        return __impl<_Self>::submit(static_cast<_Self&&>(__self), static_cast<_Receiver&&>(__rcvr));
+      }
+
       template <class _Sender, class _ApplyFn>
       STDEXEC_ATTRIBUTE((always_inline)) static auto apply(_Sender&& __sndr, _ApplyFn&& __fun) //
         noexcept(

include/stdexec/__detail/__just.hpp

@@ -52,6 +52,17 @@ namespace stdexec {
           },
           __state);
       };
+
+      static constexpr auto submit =
+        []<class _Sender, class _Receiver>(_Sender&& __sndr, _Receiver __rcvr) noexcept -> void {
+        static_assert(sender_expr_for<_Sender, _JustTag>);
+        auto&& __state = get_state(static_cast<_Sender&&>(__sndr), __rcvr);
+        __state.apply(
+          [&]<class... _Ts>(_Ts&&... __ts) noexcept {
+            __tag_t()(static_cast<_Receiver&&>(__rcvr), static_cast<_Ts&&>(__ts)...);
+          },
+          static_cast<decltype(__state)>(__state));
+      };
     };
 
     struct just_t {

include/stdexec/__detail/__let.hpp

@@ -27,6 +27,7 @@
 #include "__schedulers.hpp"
 #include "__sender_adaptor_closure.hpp"
 #include "__senders.hpp"
+#include "__submit.hpp"
 #include "__transform_sender.hpp"
 #include "__transform_completion_signatures.hpp"
 #include "__variant.hpp"
@@ -224,9 +225,10 @@ namespace stdexec {
         __result_receiver_t<_Receiver, _Scheduler>>;
 
     template <class _ResultSender, class _Scheduler, class _Receiver>
-    using __op_state_t = connect_result_t<
+    using __submit_result = submit_result_t<
       _ResultSender,
-      __checked_result_receiver_t<_ResultSender, _Scheduler, _Receiver>>;
+      __checked_result_receiver_t<_ResultSender, _Scheduler, _Receiver>,
+      __ignore>;
 
     template <class _SetTag, class _Fun, class _Sched, class... _Env>
     struct __transform_signal_fn {
@@ -353,10 +355,11 @@ namespace stdexec {
     //! Metafunction creating the operation state needed to connect the result of calling
     //! the sender factory function, `_Fun`, and passing its result to a receiver.
     template <class _Receiver, class _Fun, class _Set, class _Sched>
-    struct __op_state_for {
-      static_assert(!std::is_reference_v<_Sched>);
+    struct __submit_datum_for {
+      // compute the result of calling submit with the result of executing _Fun
+      // with _Args. if the result is void, substitute with __ignore.
       template <class... _Args>
-      using __f = __op_state_t<
+      using __f = __submit_result<
         __mcall<__result_sender_fn<_Set, _Fun, _Sched, env_of_t<_Receiver>>, _Args...>,
         _Sched,
         _Receiver>;
@@ -371,10 +374,10 @@ namespace stdexec {
       using __env_t = __result_env_t<_Sched, env_of_t<_Receiver>>;
       using __rcvr_t = __receiver_with_sched_t<_Receiver, _Sched>;
       using __result_variant = __variant_for<__monostate, _Tuples...>;
-      using __op_state_variant = //
+      using __submit_variant = //
         __variant_for<
           __monostate,
-          __mapply<__op_state_for<_Receiver, _Fun, _Set, _Sched>, _Tuples>...>;
+          __mapply<__submit_datum_for<_Receiver, _Fun, _Set, _Sched>, _Tuples>...>;
 
       template <class _ResultSender, class _OpState>
       auto __get_result_receiver(const _ResultSender&, _OpState& __op_state) -> decltype(auto) {
@@ -407,7 +410,7 @@ namespace stdexec {
       __result_variant __args_{};
       //! Variant type for holding the operation state from connecting
       //! the function result to the downstream receiver:
-      __op_state_variant __op_state3_{};
+      __submit_variant __storage_{};
     };
 
     //! Implementation of the `let_*_t` types, where `_Set` is, e.g., `set_value_t` for `let_value`.
@@ -493,8 +496,8 @@ namespace stdexec {
         };
 
       //! Helper function to actually invoke the function to produce `let_*`'s sender,
-      //! connect it to the downstream receiver, and start it.
-      //! This is the heart of `let_*`.
+      //! connect it to the downstream receiver, and start it. This is the heart of
+      //! `let_*`.
       template <class _State, class _OpState, class... _As>
       static void __bind_(_State& __state, _OpState& __op_state, _As&&... __as) {
         // Store the passed-in (received) args:
@@ -504,9 +507,8 @@ namespace stdexec {
         // Create a receiver based on the state, the computed sender, and the operation state:
         auto __rcvr2 = __state.__get_result_receiver(__sndr2, __op_state);
         // Connect the sender to the receiver and start it:
-        auto& __op2 = __state.__op_state3_.emplace_from(
-          stdexec::connect, std::move(__sndr2), std::move(__rcvr2));
-        stdexec::start(__op2);
+        __state.__storage_.emplace_from(
+          stdexec::submit, std::move(__sndr2), std::move(__rcvr2), __ignore{});
       }
 
       template <class _OpState, class... _As>

include/stdexec/__detail/__read_env.hpp

@@ -26,6 +26,7 @@
 #include "__optional.hpp"
 #include "__meta.hpp"
 #include "__receivers.hpp"
+#include "__submit.hpp"
 
 #include <exception>
 
@@ -119,6 +120,24 @@ namespace stdexec {
           stdexec::__set_value_invoke(static_cast<_Receiver&&>(__rcvr), __query_fn);
         }
       };
+
+      static constexpr auto submit = //
+        []<class _Sender, class _Receiver>(const _Sender& __sndr, _Receiver __rcvr) noexcept {
+          static_assert(sender_expr_for<_Sender, __read_env_t>);
+          using __query = __data_of<_Sender>;
+          using __result = __call_result_t<__query, env_of_t<_Receiver>>;
+          // When the query completes with a reference type, we can complete the receiver
+          // immediately; there is no need to return an operation state.
+          if constexpr (__same_as<__result, __result&&>) {
+            stdexec::__set_value_invoke(
+              static_cast<_Receiver&&>(__rcvr), __query(), stdexec::get_env(__rcvr));
+          } else {
+            // This will create a __state object to cache the query result and start the
+            // operation.
+            return __submit::__op_data<const _Sender&, _Receiver>(
+              __sndr, static_cast<_Receiver&&>(__rcvr));
+          }
+        };
     };
   } // namespace __read
 

include/stdexec/__detail/__start_detached.hpp

@@ -24,20 +24,47 @@
 #include "__receivers.hpp"
 #include "__env.hpp"
 #include "__scope.hpp"
+#include "__submit.hpp"
 #include "__transform_sender.hpp"
 
 namespace stdexec {
   /////////////////////////////////////////////////////////////////////////////
   // [execution.senders.consumer.start_detached]
   namespace __start_detached {
+    struct __submit_receiver {
+      using receiver_concept = receiver_t;
+      using __t = __submit_receiver;
+      using __id = __submit_receiver;
+
+      template <class... _As>
+      void set_value(_As&&...) noexcept {
+      }
+
+      template <class _Error>
+      [[noreturn]]
+      void set_error(_Error&&) noexcept {
+        // A detached operation failed. There is noplace for the error to go.
+        // This is unrecoverable, so we terminate.
+        std::terminate();
+      }
+
+      void set_stopped() noexcept {
+      }
+
+      [[nodiscard]]
+      auto get_env() const noexcept -> __root_env {
+        return {};
+      }
+    };
+
     template <class _SenderId, class _EnvId>
     struct __operation {
       using _Sender = __cvref_t<_SenderId>;
       using _Env = __t<_EnvId>;
 
       explicit __operation(_Sender&& __sndr, _Env __env)
         : __env_(static_cast<_Env&&>(__env))
-        , __op_state_(connect(static_cast<_Sender&&>(__sndr), __receiver{this})) {
+        , __op_data_(submit(static_cast<_Sender&&>(__sndr), __receiver{this}, __ignore{})) {
       }
 
       static void __destroy_delete(__operation* __self) noexcept {
@@ -85,7 +112,7 @@ namespace stdexec {
       };
 
       STDEXEC_ATTRIBUTE((no_unique_address)) _Env __env_;
-      connect_result_t<_Sender, __receiver> __op_state_;
+      STDEXEC_ATTRIBUTE((no_unique_address)) submit_result_t<_Sender, __receiver, __ignore> __op_data_;
     };
 
     struct start_detached_t {
@@ -117,13 +144,20 @@ namespace stdexec {
         requires sender_in<_Sender, __as_root_env_t<_Env>>
       void apply_sender(_Sender&& __sndr, _Env&& __env = {}) const noexcept(false) {
         using _Op = __operation<__cvref_id<_Sender>, __id<__decay_t<_Env>>>;
-        // Use the provided allocator, if any, to allocate the operation state.
-        if constexpr (__callable<get_allocator_t, _Env>) {
+
+        // NON-STANDARD
+        if constexpr (
+          __same_as<_Env, __root_env>
+          && __same_as<void, submit_result_t<_Sender, __submit_receiver>>) {
+          // If submit(sndr, rcvr) returns void, then no state needs to be kept alive
+          // for the operation. We can just call submit and return.
+          stdexec::submit(static_cast<_Sender&&>(__sndr), __submit_receiver{});
+        } else if constexpr (__callable<get_allocator_t, _Env>) {
+          // Use the provided allocator if any to allocate the operation state.
           auto __alloc = get_allocator(__env);
           using _Alloc = decltype(__alloc);
           using _OpAlloc = typename std::allocator_traits<_Alloc>::template rebind_alloc<_Op>;
-          // We use the allocator to allocate the operation state and also to construct
-          // it.
+          // We use the allocator to allocate the op state and also to construct it.
           _OpAlloc __op_alloc{__alloc};
           _Op* __op = std::allocator_traits<_OpAlloc>::allocate(__op_alloc, 1);
           __scope_guard __g{[__op, &__op_alloc]() noexcept {
@@ -135,12 +169,12 @@ namespace stdexec {
             __op_alloc, __op, static_cast<_Sender&&>(__sndr), static_cast<_Env&&>(__env));
           // The operation state is now constructed, dismiss the scope guard.
           __g.__dismiss();
-          // This cannot throw:
-          stdexec::start(__op->__op_state_);
         } else {
           // The caller did not provide an allocator, so we use the default allocator.
+          [[maybe_unused]]
           _Op* __op = new _Op(static_cast<_Sender&&>(__sndr), static_cast<_Env&&>(__env));
-          start(__op->__op_state_);
+          // The operation has now started and is responsible for deleting itself when it
+          // completes.
         }
       }
     };