use new-style sender algo customization in stream_scheduler

Author: ericniebler

.clang-format

@@ -83,14 +83,13 @@ KeepEmptyLinesAtTheStartOfBlocks: true
 LambdaBodyIndentation: Signature
 LineEnding: LF
 Macros: [
-  'STDEXEC_MEMFN_DECL(X)=X',
-  'STDEXEC_MEMFN_DECL(X,Y)=X,Y',
-  'STDEXEC_MEMFN_DECL(X,Y,Z)=X,Y,Z',
-  'STDEXEC_ATTRIBUTE(X)=[[]]',
+  'STDEXEC_MEMFN_DECL(...)=__VA_ARGS__',
+  'STDEXEC_ATTRIBUTE(X)=//',
   'STDEXEC_NO_UNIQUE_ADDRESS=[[no_unique_address]]',
   'STDEXEC_IMMOVABLE_NO_UNIQUE_ADDRESS=[[no_unique_address]]',
   'STDEXEC_MISSING_MEMBER(X,Y)=true',
-  'STDEXEC_DEFINE_MEMBER(X)=void foo() {}'
+  'STDEXEC_DEFINE_MEMBER(X)=void foo() {}',
+  'STDEXEC_AUTO_RETURN(...)=->decltype(auto){ return __VA_ARGS__; }',
 ]
 MaxEmptyLinesToKeep: 2
 NamespaceIndentation: All

.clangd

@@ -35,7 +35,7 @@ CompileFlags:
 
 # Use clang++ in CUDA mode to provide intellisense for files with `nvexec` in their path
 If:
-  PathMatch: (include|examples|test)/nvexec.*
+  PathMatch: .*\b(include|examples|test)/nvexec.*
 CompileFlags:
   Compiler: clang++
   Add:
@@ -54,6 +54,7 @@ CompileFlags:
     - "-ferror-limit=0"
     - "-fmacro-backtrace-limit=0"
     - "-ftemplate-backtrace-limit=0"
+    - "-std=gnu++20"
   Remove:
     - "-stdpar*"
     # strip CUDA fatbin args

examples/benchmark/static_thread_pool_bulk_enqueue.cpp

@@ -45,7 +45,7 @@ struct RunThread {
       auto env = exec::make_env(stdexec::prop{stdexec::get_allocator, alloc});
       auto [start, end] = exec::_pool_::even_share(total_scheds, tid, pool.available_parallelism());
       auto iterate = exec::schedule_all(pool, std::views::iota(start, end))
-                   | exec::ignore_all_values() | exec::write(env);
+                   | exec::ignore_all_values() | exec::write_env(env);
 #  else
       auto [start, end] = exec::_pool_::even_share(total_scheds, tid, pool.available_parallelism());
       auto iterate = exec::schedule_all(pool, std::views::iota(start, end))

examples/benchmark/static_thread_pool_bulk_enqueue_nested.cpp

@@ -46,7 +46,7 @@ struct RunThread {
       auto env = exec::make_env(stdexec::prop{stdexec::get_allocator, alloc});
       auto [start, end] = exec::_pool_::even_share(total_scheds, tid, pool.available_parallelism());
       auto iterate = exec::iterate(std::views::iota(start, end)) | exec::ignore_all_values()
-                   | exec::write(env);
+                   | exec::write_env(env);
 #  else
       auto [start, end] = exec::_pool_::even_share(total_scheds, tid, pool.available_parallelism());
       auto iterate = exec::iterate(std::views::iota(start, end)) | exec::ignore_all_values();

examples/nvexec/CMakeLists.txt

@@ -92,6 +92,9 @@ function(def_cpu_example example)
                 stdexec_executable_flags
     )
     set_source_files_properties(${source} PROPERTIES LANGUAGE ${_lang_cxx})
+    if (CMAKE_CXX_COMPILER_ID STREQUAL "Clang")
+        target_link_options(${target} PRIVATE -lc++abi -ltbb)
+    endif()
 endfunction()
 
 foreach(example ${nvexec_cpu_examples})

examples/nvexec/maxwell_cpu_mt.cpp

@@ -14,6 +14,8 @@
  * limitations under the License.
  */
 
+#include <iostream>
+
 #include "maxwell/snr.cuh"
 #include "maxwell/std.cuh"
 #include "maxwell/stdpar.cuh"

include/exec/any_sender_of.hpp

@@ -624,8 +624,8 @@ namespace exec {
           template <class _Rcvr>
             requires receiver_of<_Rcvr, completion_signatures<_Sigs...>>
                   && (__callable<__query_vfun_fn<_Rcvr>, _Queries> && ...)
-          STDEXEC_MEMFN_DECL(
-            auto __create_vtable)(this __mtype<__t>, __mtype<_Rcvr>) noexcept -> const __t* {
+          STDEXEC_MEMFN_DECL(auto __create_vtable)(this __mtype<__t>, __mtype<_Rcvr>) noexcept
+            -> const __t* {
             static const __t __vtable_{
               {__any_::__rcvr_vfun_fn(
                 static_cast<_Rcvr*>(nullptr), static_cast<_Sigs*>(nullptr))}...,
@@ -960,9 +960,8 @@ namespace exec {
         __immovable_operation_storage (*__connect_)(void*, __receiver_ref_t);
        private:
         template <sender_to<__receiver_ref_t> _Sender>
-        STDEXEC_MEMFN_DECL(
-          auto
-          __create_vtable)(this __mtype<__vtable>, __mtype<_Sender>) noexcept -> const __vtable* {
+        STDEXEC_MEMFN_DECL(auto __create_vtable)(this __mtype<__vtable>, __mtype<_Sender>) noexcept
+          -> const __vtable* {
           static const __vtable __vtable_{
             {*__create_vtable(__mtype<__query_vtable<_SenderQueries>>{}, __mtype<_Sender>{})},
             [](void* __object_pointer, __receiver_ref_t __receiver)
@@ -1085,8 +1084,8 @@ namespace exec {
        private:
         template <scheduler _Scheduler>
         STDEXEC_MEMFN_DECL(
-          auto
-          __create_vtable)(this __mtype<__vtable>, __mtype<_Scheduler>) noexcept -> const __vtable* {
+          auto __create_vtable)(this __mtype<__vtable>, __mtype<_Scheduler>) noexcept
+          -> const __vtable* {
           static const __vtable __vtable_{
             {*__create_vtable(
               __mtype<__query_vtable<_SchedulerQueries, false>>{}, __mtype<_Scheduler>{})},

include/exec/env.hpp

@@ -118,10 +118,10 @@ namespace exec {
 
       template <__decays_to<__sender> _Self, class _Receiver>
         requires receiver_of<_Receiver, __completions_t<env_of_t<_Receiver>>>
-      static auto connect(_Self&& __self, _Receiver __rcvr) //
+      static auto connect(_Self&& __Self, _Receiver __rcvr) //
         noexcept(std::is_nothrow_move_constructible_v<_Receiver>)
           -> __operation_t<_Tag, __default_t<env_of_t<_Receiver>>, _Receiver> {
-        return {{}, static_cast<_Self&&>(__self).__default_, static_cast<_Receiver&&>(__rcvr)};
+        return {{}, static_cast<_Self&&>(__Self).__default_, static_cast<_Receiver&&>(__rcvr)};
       }
 
       template <class _Env>
@@ -141,8 +141,73 @@ namespace exec {
 
   inline constexpr __read_with_default::__read_with_default_t read_with_default{};
 
+  [[deprecated("exec::write_env has been renamed to exec::write_env")]]
   inline constexpr stdexec::__write_::__write_env_t write{};
   inline constexpr stdexec::__write_::__write_env_t write_env{};
+
+  namespace __write_attrs {
+    using namespace stdexec;
+
+    template <class _SenderId, class _Attrs>
+    struct __sender {
+      using _Sender = stdexec::__t<_SenderId>;
+
+      struct __t {
+        using sender_concept = sender_t;
+        using __id = __sender;
+        _Sender __sndr_;
+        _Attrs __attrs_;
+
+        auto get_env() const noexcept -> __env::__join_t<const _Attrs&, env_of_t<_Sender>> {
+          return stdexec::__env::__join(__attrs_, stdexec::get_env(__sndr_));
+        }
+
+        template <__decays_to<__t> _Self, class... _Env>
+        static auto get_completion_signatures(_Self&&, _Env&&...)
+          -> completion_signatures_of_t<__copy_cvref_t<_Self, _Sender>, _Env...> {
+          return {};
+        }
+
+        template <__decays_to<__t> _Self, class _Receiver>
+          requires sender_in<__copy_cvref_t<_Self, _Sender>, env_of_t<_Receiver>>
+        static auto connect(_Self&& __self, _Receiver __rcvr)
+          -> connect_result_t<__copy_cvref_t<_Self, _Sender>, _Receiver> {
+          return stdexec::connect(std::forward<_Self>(__self).__sndr_, std::move(__rcvr));
+        }
+      };
+    };
+
+    struct __write_attrs_t {
+      template <class _Sender, class _Attrs>
+      STDEXEC_ATTRIBUTE((host, device))
+      auto operator()(_Sender snd, _Attrs __attrs_) const //
+        -> __write_attrs::__sender<_Sender, _Attrs> {
+        return __t<__write_attrs::__sender<__id<_Sender>, _Attrs>>{
+          static_cast<_Sender&&>(snd), static_cast<_Attrs&&>(__attrs_)};
+      }
+
+      template <class _Attrs>
+      struct __closure {
+        _Attrs __attrs_;
+
+        template <class _Sender>
+        STDEXEC_ATTRIBUTE((host, device))
+        friend auto operator|(_Sender __sndr_, __closure _clsr) {
+          return __t<__write_attrs::__sender<__id<_Sender>, _Attrs>>{
+            static_cast<_Sender&&>(__sndr_), static_cast<_Attrs&&>(_clsr.__attrs_)};
+        }
+      };
+
+      template <class _Attrs>
+      STDEXEC_ATTRIBUTE((host, device))
+      auto operator()(_Attrs __attrs_) const {
+        return __closure<_Attrs>{static_cast<_Attrs&&>(__attrs_)};
+      }
+    };
+  } // namespace __write_attrs
+
+  inline constexpr __write_attrs::__write_attrs_t write_attrs{};
+
 } // namespace exec
 
 STDEXEC_PRAGMA_POP()

include/exec/sequence.hpp

@@ -19,6 +19,9 @@
 #include <stdexec/__detail/__tuple.hpp>
 #include <stdexec/__detail/__variant.hpp>
 
+STDEXEC_PRAGMA_PUSH()
+STDEXEC_PRAGMA_IGNORE_GNU("-Wmissing-braces")
+
 namespace exec {
   namespace _seq {
     template <class... Sndrs>
@@ -96,7 +99,7 @@ namespace exec {
         // result is that the first sender in `sndrs` is not moved from, but the rest are.
         _ops.template emplace_from_at<0>(
           stdexec::connect,
-          stdexec::__tup::get<0>(static_cast<CvrefSndrs&&>(sndrs)),
+          sndrs.template __get<0>(static_cast<CvrefSndrs&&>(sndrs)),
           _rcvr_t<0>{this});
       }
 
@@ -107,7 +110,7 @@ namespace exec {
           if constexpr (Idx == sizeof...(Sndrs) + 1) {
             stdexec::set_value(static_cast<Rcvr&&>(_rcvr), static_cast<Args&&>(args)...);
           } else {
-            auto& sndr = stdexec::__tup::get<Idx>(_sndrs);
+            auto& sndr = _sndrs.template __get<Idx>(_sndrs);
             auto& op = _ops.template emplace_from_at<Idx>(
               stdexec::connect, std::move(sndr), _rcvr_t<Idx>{this});
             stdexec::start(op);
@@ -197,7 +200,7 @@ namespace exec {
     template <class... Sndrs>
       requires(sizeof...(Sndrs) > 1) && stdexec::__domain::__has_common_domain<Sndrs...>
     STDEXEC_ATTRIBUTE((host, device)) _sndr<Sndrs...> sequence_t::operator()(Sndrs... sndrs) const {
-      return _sndr<Sndrs...>{{}, {}, {{static_cast<Sndrs&&>(sndrs)}...}};
+      return _sndr<Sndrs...>{{}, {}, {static_cast<Sndrs&&>(sndrs)...}};
     }
   } // namespace _seq
 
@@ -215,3 +218,5 @@ namespace std {
     using type = stdexec::__m_at_c<I, exec::sequence_t, stdexec::__, Sndrs...>;
   };
 } // namespace std
+
+STDEXEC_PRAGMA_POP()

include/exec/sequence/any_sequence_of.hpp

@@ -67,8 +67,8 @@ namespace exec {
           template <class _Rcvr>
             requires sequence_receiver_of<_Rcvr, __item_types>
                   && (__callable<__query_vfun_fn<_Rcvr>, _Queries> && ...)
-          STDEXEC_MEMFN_DECL(
-            auto __create_vtable)(this __mtype<__t>, __mtype<_Rcvr>) noexcept -> const __t* {
+          STDEXEC_MEMFN_DECL(auto __create_vtable)(this __mtype<__t>, __mtype<_Rcvr>) noexcept
+            -> const __t* {
             static const __t __vtable_{
               {__rcvr_next_vfun_fn<_Rcvr>{}(static_cast<_NextSigs*>(nullptr))},
               {__any_::__rcvr_vfun_fn(
@@ -182,8 +182,8 @@ namespace exec {
 
         template <class _Sender>
           requires sequence_sender_to<_Sender, __receiver_ref_t>
-        STDEXEC_MEMFN_DECL(
-          auto __create_vtable)(this __mtype<__t>, __mtype<_Sender>) noexcept -> const __t* {
+        STDEXEC_MEMFN_DECL(auto __create_vtable)(this __mtype<__t>, __mtype<_Sender>) noexcept
+          -> const __t* {
           static const __t __vtable_{
             {*__create_vtable(__mtype<__query_vtable_t>{}, __mtype<_Sender>{})},
             [](void* __object_pointer, __receiver_ref_t __receiver)