Merge branch 'on-redux' into feature/cuda_reduce

Author: ericniebler

include/exec/async_scope.hpp

@@ -222,6 +222,7 @@ namespace exec {
         return __nest_operation_t<_Receiver>{
           __self.__scope_, ((_Self&&) __self).__c_, (_Receiver&&) __rcvr};
       }
+
       template <__decays_to<__nest_sender> _Self, class _Env>
       friend auto tag_invoke(get_completion_signatures_t, _Self&&, _Env&&)
         -> completion_signatures_of_t<__copy_cvref_t<_Self, _Constrained>, __env_t<_Env>> {
@@ -597,10 +598,17 @@ namespace exec {
 
     ////////////////////////////////////////////////////////////////////////////
     // async_scope::spawn implementation
+    template <class _Env>
+    using __spawn_env_t = __result_of<
+      __join_env,
+      _Env,
+      __env::__prop<get_stop_token_t, in_place_stop_token>,
+      __env::__prop<get_scheduler_t, __inln::__scheduler>>;
+
     template <class _EnvId>
     struct __spawn_op_base {
       using _Env = __t<_EnvId>;
-      __env_t<_Env> __env_;
+      __spawn_env_t<_Env> __env_;
       void (*__delete_)(__spawn_op_base*);
     };
 
@@ -623,7 +631,7 @@ namespace exec {
         std::terminate();
       }
 
-      friend const __env_t<_Env>& tag_invoke(get_env_t, const __spawn_rcvr& __self) noexcept {
+      friend const __spawn_env_t<_Env>& tag_invoke(get_env_t, const __spawn_rcvr& __self) noexcept {
         return __self.__op_->__env_;
       }
     };
@@ -638,8 +646,9 @@ namespace exec {
 
       template <__decays_to<_Sender> _Sndr>
       __spawn_op(_Sndr&& __sndr, _Env __env, const __impl* __scope)
-        : __spawn_op_base<_EnvId>{make_env((_Env&&) __env,
-                                    with(get_stop_token, __scope->__stop_source_.get_token())),
+        : __spawn_op_base<_EnvId>{__join_env((_Env&&) __env,
+          __mkprop(get_stop_token, __scope->__stop_source_.get_token()),
+          __mkprop(get_scheduler, __inln::__scheduler{})),
           [](__spawn_op_base<_EnvId>* __op) {
             delete static_cast<__spawn_op*>(__op);
           }}
@@ -682,7 +691,7 @@ namespace exec {
         return nest_result_t<_Constrained>{&__impl_, (_Constrained&&) __c};
       }
 
-      template <__movable_value _Env = empty_env, sender_in<__env_t<_Env>> _Sender>
+      template <__movable_value _Env = empty_env, sender_in<__spawn_env_t<_Env>> _Sender>
         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>;

include/exec/env.hpp

@@ -115,116 +115,7 @@ namespace exec {
 
   inline constexpr __read_with_default::__read_with_default_t read_with_default{};
 
-  namespace __write {
-    using namespace stdexec;
-
-    struct __write_t;
-
-    template <class _ReceiverId, class _Env>
-    struct __operation_base {
-      using _Receiver = __t<_ReceiverId>;
-      _Receiver __rcvr_;
-      const _Env __env_;
-    };
-
-    template <class _ReceiverId, class _Env>
-    struct __receiver {
-      using _Receiver = stdexec::__t<_ReceiverId>;
-
-      struct __t : receiver_adaptor<__t> {
-        _Receiver&& base() && noexcept {
-          return (_Receiver&&) __op_->__rcvr_;
-        }
-
-        const _Receiver& base() const & noexcept {
-          return __op_->__rcvr_;
-        }
-
-        auto get_env() const noexcept -> __env::__env_join_t<const _Env&, env_of_t<_Receiver>> {
-          return __join_env(__op_->__env_, stdexec::get_env(base()));
-        }
-
-        __operation_base<_ReceiverId, _Env>* __op_;
-      };
-    };
-
-    template <class _SenderId, class _ReceiverId, class _Env>
-    struct __operation : __operation_base<_ReceiverId, _Env> {
-      using _Sender = __t<_SenderId>;
-      using __base_t = __operation_base<_ReceiverId, _Env>;
-      using __receiver_t = __t<__receiver<_ReceiverId, _Env>>;
-      connect_result_t<_Sender, __receiver_t> __state_;
-
-      __operation(_Sender&& __sndr, auto&& __rcvr, auto&& __env)
-        : __base_t{(decltype(__rcvr)) __rcvr, (decltype(__env)) __env}
-        , __state_{stdexec::connect((_Sender&&) __sndr, __receiver_t{{}, this})} {
-      }
-
-      friend void tag_invoke(start_t, __operation& __self) noexcept {
-        start(__self.__state_);
-      }
-    };
-
-    template <class _SenderId, class _Env>
-    struct __sender {
-      using _Sender = stdexec::__t<_SenderId>;
-
-      template <class _Receiver>
-      using __receiver_t = stdexec::__t<__receiver<__id<_Receiver>, _Env>>;
-      template <class _Self, class _Receiver>
-      using __operation_t =
-        __operation<__id<__copy_cvref_t<_Self, _Sender>>, __id<_Receiver>, _Env>;
-
-      struct __t {
-        using is_sender = void;
-        using __id = __sender;
-        _Sender __sndr_;
-        _Env __env_;
-
-        template <__decays_to<__t> _Self, receiver _Receiver>
-          requires sender_to<__copy_cvref_t<_Self, _Sender>, __receiver_t<_Receiver>>
-        friend auto tag_invoke(connect_t, _Self&& __self, _Receiver __rcvr)
-          -> __operation_t<_Self, _Receiver> {
-          return {((_Self&&) __self).__sndr_, (_Receiver&&) __rcvr, ((_Self&&) __self).__env_};
-        }
-
-        friend auto tag_invoke(stdexec::get_env_t, const __t& __self) //
-          noexcept(stdexec::__nothrow_callable<stdexec::get_env_t, const _Sender&>)
-            -> stdexec::env_of_t<const _Sender&> {
-          return stdexec::get_env(__self.__sndr_);
-        }
-
-        template <__decays_to<__t> _Self, class _BaseEnv>
-        friend auto tag_invoke(get_completion_signatures_t, _Self&&, _BaseEnv&&)
-          -> stdexec::__completion_signatures_of_t<
-            __copy_cvref_t<_Self, _Sender>,
-            __env::__env_join_t<_Env, _BaseEnv>> {
-          return {};
-        }
-      };
-    };
-
-    struct __write_t {
-      template <class _Sender, class... _Funs>
-      using __sender_t =
-        __t<__sender<__id<__decay_t<_Sender>>, __env::__env_join_t<__env::__env_fn<_Funs>...>>>;
-
-      template <__is_not_instance_of<__env::__env_fn> _Sender, class... _Funs>
-        requires sender<_Sender>
-      auto operator()(_Sender&& __sndr, __env::__env_fn<_Funs>... __withs) const
-        -> __sender_t<_Sender, _Funs...> {
-        return {(_Sender&&) __sndr, __join_env(std::move(__withs)...)};
-      }
-
-      template <class... _Funs>
-      auto operator()(__env::__env_fn<_Funs>... __withs) const
-        -> __binder_back<__write_t, __env::__env_fn<_Funs>...> {
-        return {{}, {}, {std::move(__withs)...}};
-      }
-    };
-  } // namespace __write
-
-  inline constexpr __write::__write_t write{};
+  inline constexpr stdexec::__write_::__write_t write{};
 } // namespace exec
 
 #ifdef __EDG__

include/exec/static_thread_pool.hpp

@@ -93,7 +93,7 @@ namespace exec {
       }
 
       // For eager customization
-      template <stdexec::__lazy_sender_for<stdexec::bulk_t> Sender>
+      template <stdexec::sender_expr_for<stdexec::bulk_t> Sender>
       auto transform_sender(Sender&& sndr) const noexcept {
         auto sched = stdexec::get_completion_scheduler<stdexec::set_value_t>(
           stdexec::get_env(sndr));
@@ -106,7 +106,7 @@ namespace exec {
       }
 
       // transform the generic bulk sender into a parallel thread-pool bulk sender
-      template <stdexec::__lazy_sender_for<stdexec::bulk_t> Sender, class Env>
+      template <stdexec::sender_expr_for<stdexec::bulk_t> Sender, class Env>
         requires stdexec::__callable<stdexec::get_scheduler_t, Env>
       auto transform_sender(Sender&& sndr, const Env& env) const noexcept {
         auto sched = stdexec::get_scheduler(env);
@@ -182,17 +182,6 @@ namespace exec {
         return s.make_sender_();
       }
 
-      // template <class S, class Shape, class Fn>
-      // bulk_sender_t<S, Shape, Fn> make_bulk_sender_(S&& sndr, Shape shape, Fn fun) const {
-      //   return bulk_sender_t<S, Shape, Fn>{*pool_, (S&&) sndr, shape, (Fn&&) fun};
-      // }
-
-      // template <stdexec::sender S, std::integral Shape, class Fn>
-      // friend bulk_sender_t<S, Shape, Fn>
-      //   tag_invoke(stdexec::bulk_t, scheduler sch, S&& sndr, Shape shape, Fn fun) noexcept {
-      //   return sch.make_bulk_sender_((S&&) sndr, shape, (Fn&&) fun);
-      // }
-
       friend stdexec::forward_progress_guarantee
         tag_invoke(stdexec::get_forward_progress_guarantee_t, const static_thread_pool&) noexcept {
         return stdexec::forward_progress_guarantee::parallel;

include/nvexec/stream/reduce.cuh

@@ -188,17 +188,17 @@ namespace nvexec {
       // #if STDEXEC_FRIENDSHIP_IS_LEXICAL()
       //      private:
       //       template <class...>
-      //       friend struct stdexec::__basic_sender;
+      //       friend struct stdexec::__sexpr;
       // #endif
 
       template < sender Sender, __movable_value Init, __movable_value Fun = cub::Sum>
       auto operator()(Sender&& sndr, Init init, Fun fun) const {
         auto __domain = __get_sender_domain(sndr);
-        return __domain.transform_sender(make_sender<reduce_t>(
+        return __domain.transform_sender(make_sender_expr<reduce_t>(
           reduce_::__data{(Init&&) init, (Fun&&) fun}, (Sender&&) sndr));
       }
 
-      template <__lazy_sender_for<reduce_t> _Sender>
+      template <sender_expr_for<reduce_t> _Sender>
       static auto get_env(const _Sender&) noexcept {
         return empty_env{};
       }
@@ -213,7 +213,7 @@ namespace nvexec {
         }
       };
 
-      template <__lazy_sender_for<reduce_t> _Sender, receiver _Receiver>
+      template <sender_expr_for<reduce_t> _Sender, receiver _Receiver>
       //requires SOME CONSTRAINT HERE
       static auto connect(_Sender&& __sndr, _Receiver __rcvr) {
         return op{}; // return a dummy operation state to see if it compiles
@@ -231,7 +231,7 @@ namespace nvexec {
           completion_signatures<set_stopped_t()>,
           __mbind_back_q<_set_value_t, _Init, _Fun>>;
 
-      template <__lazy_sender_for<reduce_t> _Sender, class _Env>
+      template <sender_expr_for<reduce_t> _Sender, class _Env>
       static auto get_completion_signatures(_Sender&& __sndr, _Env&& env) {
         // what's the relationship(if it exists) between the lambdas types and the lambda types in `stream_domain::transform_sender`
         // apply_sender?
@@ -264,7 +264,7 @@ namespace nvexec {
           _Fun),
         tag_invoke_t(reduce_t, _Sender, _Init, _Fun)>;
 
-      template <__lazy_sender_for<reduce_t> _Sender, receiver _Receiver>
+      template <sender_expr_for<reduce_t> _Sender, receiver _Receiver>
       static auto connect(_Sender&& __sndr, _Receiver __rcvr) noexcept(
         __nothrow_callable< apply_sender_t, _Sender, reduce_::__connect_fn<_Receiver>>)
         -> __call_result_t< apply_sender_t, _Sender, reduce_::__connect_fn<_Receiver>> {

include/nvexec/stream_context.cuh

@@ -99,19 +99,19 @@ namespace nvexec {
       using __default_domain::transform_sender;
 
       // Lazy algorithm customizations require a recursive tree transformation
-      template <__lazy_sender Sender, class Env>
+      template <sender_expr Sender, class Env>
         requires _non_stream_sender<Sender> // no need to transform it a second time
       auto transform_sender(Sender&& sndr, const Env& env) const noexcept {
         return stdexec::apply_sender(
           (Sender&&) sndr,
           [&]<class Tag, class Data, class... Children>(Tag, Data&& data, Children&&... children) {
-            return make_sender<Tag, stream_domain>(
+            return make_sender_expr<Tag, stream_domain>(
               (Data&&) data, transform_sender((Children&&) children, env)...);
           });
       }
 
       // reduce senders get a special transformation
-      template <__lazy_sender_for<reduce_t> Sender, class Env>
+      template <sender_expr_for<reduce_t> Sender, class Env>
         requires _non_stream_sender<Sender> // no need to transform it a second time
       auto transform_sender(Sender&& sndr, const Env& env) const noexcept {
         return stdexec::apply_sender(