add a customizable algorithm submit that combines connect and start (#1519)

Author: ericniebler

include/exec/async_scope.hpp

@@ -590,15 +590,27 @@ namespace exec {
     struct __future_state : __future_state_base<__future_completions_t<_Sender, _Env>, _Env> {
       using _Completions = __future_completions_t<_Sender, _Env>;
 
-      __future_state(_Sender __sndr, _Env __env, const __impl* __scope)
+      __future_state(connect_t, _Sender&& __sndr, _Env __env, const __impl* __scope)
         : __future_state_base<_Completions, _Env>(static_cast<_Env&&>(__env), __scope)
-        , __op_(
-            stdexec::connect(
-              static_cast<_Sender&&>(__sndr),
-              __future_receiver_t<_Sender, _Env>{this, __scope})) {
+        , __op_(static_cast<_Sender&&>(__sndr), __future_receiver_t<_Sender, _Env>{this, __scope}) {
       }
 
-      connect_result_t<_Sender, __future_receiver_t<_Sender, _Env>> __op_;
+      __future_state(_Sender __sndr, _Env __env, const __impl* __scope)
+        : __future_state(
+            connect,
+            static_cast<_Sender&&>(__sndr),
+            static_cast<_Env&&>(__env),
+            __scope) {
+        // If the operation completes synchronously, then the following line will cause
+        // the destruction of *this, which is not a problem because we used a delegating
+        // constructor, so *this is considered fully constructed.
+        __op_.submit(
+          static_cast<_Sender&&>(__sndr), __future_receiver_t<_Sender, _Env>{this, __scope});
+      }
+
+      STDEXEC_ATTRIBUTE((no_unique_address)) [[]] //
+        submit_result<_Sender, __future_receiver_t<_Sender, _Env>>
+          __op_{};
     };
 
     template <class _SenderId, class _EnvId>
@@ -729,21 +741,26 @@ namespace exec {
       using _Sender = stdexec::__t<_SenderId>;
 
       struct __t : __spawn_op_base<_EnvId> {
-        template <__decays_to<_Sender> _Sndr>
-        __t(_Sndr&& __sndr, _Env __env, const __impl* __scope)
+        __t(connect_t, _Sender&& __sndr, _Env __env, const __impl* __scope)
           : __spawn_op_base<_EnvId>{__env::__join(static_cast<_Env&&>(__env),
             __spawn_env_{__scope->__stop_source_.get_token()}),
             [](__spawn_op_base<_EnvId>* __op) {
                 delete static_cast<__t*>(__op);
             }}
-          , __op_(stdexec::connect(static_cast<_Sndr&&>(__sndr), __spawn_receiver_t<_Env>{this})) {
+          , __data_(static_cast<_Sender&&>(__sndr), __spawn_receiver_t<_Env>{this}) {
         }
 
-        void start() & noexcept {
-          stdexec::start(__op_);
+        __t(_Sender __sndr, _Env __env, const __impl* __scope)
+          : __t(connect, static_cast<_Sender&&>(__sndr), static_cast<_Env&&>(__env), __scope) {
+          // If the operation completes synchronously, then the following line will cause
+          // the destruction of *this, which is not a problem because we used a delegating
+          // constructor, so *this is considered fully constructed.
+          __data_.submit(static_cast<_Sender&&>(__sndr), __spawn_receiver_t<_Env>{this});
         }
 
-        connect_result_t<_Sender, __spawn_receiver_t<_Env>> __op_;
+        STDEXEC_ATTRIBUTE((no_unique_address)) [[]] //
+          submit_result<_Sender, __spawn_receiver_t<_Env>>
+            __data_;
       };
     };
 
@@ -779,19 +796,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/exec/just_from.hpp

@@ -134,6 +134,18 @@ namespace exec {
         noexcept(stdexec::__nothrow_decay_copyable<Rcvr, Fn const &>) -> _opstate<Rcvr, Fn> {
         return _opstate<Rcvr, Fn>{static_cast<Rcvr&&>(rcvr), _fn};
       }
+
+      template <class Rcvr>
+      STDEXEC_ATTRIBUTE((host, device)) auto submit(Rcvr rcvr) && noexcept -> void {
+        auto op = static_cast<_sndr_base&&>(*this).connect(static_cast<Rcvr&&>(rcvr));
+        stdexec::start(op);
+      }
+
+      template <class Rcvr>
+      STDEXEC_ATTRIBUTE((host, device)) auto submit(Rcvr rcvr) const & noexcept -> void {
+        auto op = this->connect(static_cast<Rcvr&&>(rcvr));
+        stdexec::start(op);
+      }
     };
 
     template <class Fn, class Tag = JustTag>

include/nvexec/stream/start_detached.cuh

@@ -28,14 +28,48 @@
 
 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)
+      explicit operation(connect_t, Sender&& sndr, Env env)
         : env_(static_cast<Env&&>(env))
-        , op_state_(connect(static_cast<Sender&&>(sndr), receiver{{}, this})) {
+        , op_data_(static_cast<Sender&&>(sndr), receiver{{}, this}) {
+      }
+
+      explicit operation(Sender&& sndr, Env env)
+        : operation(connect, static_cast<Sender&&>(sndr), static_cast<Env&&>(env)) {
+        // If the operation completes synchronously, then the following line will cause
+        // the destruction of *this, which is not a problem because we used a delegating
+        // constructor, so *this is considered fully constructed.
+        op_data_.submit(static_cast<Sender&&>(sndr), receiver{{}, this});
       }
 
       // If the operation state was allocated with a user-provided allocator, then we must
@@ -85,17 +119,33 @@ namespace nvexec::_strm {
       };
 
       STDEXEC_ATTRIBUTE((no_unique_address)) Env env_;
-      connect_result_t<Sender, receiver> op_state_;
+      STDEXEC_ATTRIBUTE((no_unique_address)) submit_result<Sender, receiver> op_data_;
     };
+
+    template <class Sender, class Env>
+    concept _use_submit = __submittable<Sender, submit_receiver> && __same_as<Env, __root_env>
+                        && __same_as<void, __submit_result_t<Sender, submit_receiver>>;
   } // namespace _start_detached
 
   template <>
   struct apply_sender_for<start_detached_t> {
     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>) {
+
+#if !STDEXEC_APPLE_CLANG() // There seems to be a codegen bug in apple clang that causes
+                           // `start_detached` to segfault when the code path below is
+                           // taken.
+      // BUGBUG NOT TO SPEC: the use of the non-standard `submit` algorithm here is a
+      // conforming extension.
+      if constexpr (_start_detached::_use_submit<Sender, Env>) {
+        // 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::__submit(static_cast<Sender&&>(sndr), _start_detached::submit_receiver{});
+      } else
+#endif
+        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 +161,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

@@ -168,7 +168,7 @@ namespace stdexec {
     };
 
     template <class _Sexpr, class _Receiver>
-    struct __state_box {
+    struct __state_box : __immovable {
       using __tag_t = typename __decay_t<_Sexpr>::__tag_t;
       using __state_t = __state_type_t<__tag_t, _Sexpr, _Receiver>;
 
@@ -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,8 @@ namespace stdexec {
         __result_receiver_t<_Receiver, _Scheduler>>;
 
     template <class _ResultSender, class _Scheduler, class _Receiver>
-    using __op_state_t = connect_result_t<
-      _ResultSender,
-      __checked_result_receiver_t<_ResultSender, _Scheduler, _Receiver>>;
+    using __submit_result =
+      submit_result<_ResultSender, __checked_result_receiver_t<_ResultSender, _Scheduler, _Receiver>>;
 
     template <class _SetTag, class _Fun, class _Sched, class... _Env>
     struct __transform_signal_fn {
@@ -353,10 +353,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 +372,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 +408,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 +494,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 +505,10 @@ 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);
+        using __result_t = decltype(submit_result{std::move(__sndr2), std::move(__rcvr2)});
+        auto& __op =
+          __state.__storage_.template emplace<__result_t>(std::move(__sndr2), std::move(__rcvr2));
+        __op.submit(std::move(__sndr2), std::move(__rcvr2));
       }
 
       template <class _OpState, class... _As>