Merge pull request #1560 from lucteo/bulk-unchunked

Bulk unchunked

Author: ericniebler

include/exec/system_context.hpp

@@ -98,7 +98,8 @@ namespace exec {
   /// The execution domain of the parallel_scheduler, used for the purposes of customizing
   /// sender algorithms such as `bulk_chunked` and `bulk_unchunked`.
   struct __parallel_scheduler_domain : stdexec::default_domain {
-    /// Schedules new bulk chunked work.
+    template <__bulk_chunked_or_unchunked _Sender>
+    auto transform_sender(_Sender&& __sndr) const noexcept;
     template <__bulk_chunked_or_unchunked _Sender, class _Env>
     auto transform_sender(_Sender&& __sndr, const _Env& __env) const noexcept;
   };
@@ -374,9 +375,16 @@ namespace exec {
         auto __state = reinterpret_cast<_BulkState*>(this);
         if constexpr (_BulkState::__is_unchunked) {
           (void) __end; // not used
-          std::apply(
-            [&](auto&&... __args) { __state->__fun_(__begin, __args...); },
-            *reinterpret_cast<std::tuple<_As...>*>(__base_t::__arguments_data_));
+          // If we are not parallelizing, we need to run all the iterations sequentially.
+          uint32_t __increments = 1;
+          if constexpr (!_BulkState::__parallelize) {
+            __increments = __state->__size_;
+          }
+          for (uint32_t __i = __begin; __i < __begin + __increments; __i++) {
+            std::apply(
+              [&](auto&&... __args) { __state->__fun_(__i, __args...); },
+              *reinterpret_cast<std::tuple<_As...>*>(__base_t::__arguments_data_));
+          }
         } else {
           // If we are not parallelizing, we need to pass the entire range to the functor.
           if constexpr (!_BulkState::__parallelize) {
@@ -464,7 +472,8 @@ namespace exec {
         // Schedule the bulk work on the system scheduler.
         // This will invoke `execute` on our receiver multiple times, and then a completion signal (e.g., `set_value`).
         if constexpr (_BulkState::__is_unchunked) {
-          __scheduler->schedule_bulk_unchunked(__size, __storage, *__r);
+          __scheduler
+            ->schedule_bulk_unchunked(_BulkState::__parallelize ? __size : 1, __storage, *__r);
         } else {
           __scheduler
             ->schedule_bulk_chunked(_BulkState::__parallelize ? __size : 1, __storage, *__r);
@@ -677,9 +686,17 @@ namespace exec {
     template <class _Data, class _Previous>
     auto
       operator()(stdexec::bulk_unchunked_t, _Data&& __data, _Previous&& __previous) const noexcept {
-      auto [__unused_pol, __shape, __fn] = static_cast<_Data&&>(__data);
-      return __parallel_bulk_sender<true, _Previous, decltype(__shape), decltype(__fn), true>{
-        __sched_, static_cast<_Previous&&>(__previous), __shape, std::move(__fn)};
+      auto [__pol, __shape, __fn] = static_cast<_Data&&>(__data);
+      using __policy_t = std::remove_cvref_t<decltype(__pol.__get())>;
+      constexpr bool __parallelize = std::same_as<__policy_t, stdexec::parallel_policy>
+                                  || std::same_as<__policy_t, stdexec::parallel_unsequenced_policy>;
+      return __parallel_bulk_sender<
+        true,
+        _Previous,
+        decltype(__shape),
+        decltype(__fn),
+        __parallelize
+      >{__sched_, static_cast<_Previous&&>(__previous), __shape, std::move(__fn)};
     }
 
     parallel_scheduler __sched_;
@@ -690,6 +707,22 @@ namespace exec {
     using sender_concept = stdexec::sender_t;
   };
 
+  template <__bulk_chunked_or_unchunked _Sender>
+  auto __parallel_scheduler_domain::transform_sender(_Sender&& __sndr)
+    const noexcept {
+    if constexpr (stdexec::__completes_on<_Sender, parallel_scheduler>) {
+      auto __sched = stdexec::get_completion_scheduler<stdexec::set_value_t>(
+        stdexec::get_env(__sndr));
+      return stdexec::__sexpr_apply(
+        static_cast<_Sender&&>(__sndr), __transform_parallel_bulk_sender{__sched});
+    } else {
+      static_assert(
+        stdexec::__completes_on<_Sender, parallel_scheduler>,
+        "No parallel_scheduler instance can be found in the sender's "
+        "environment on which to schedule bulk work.");
+      return __not_a_sender<stdexec::__name_of<_Sender>>();
+    }
+  }
   template <__bulk_chunked_or_unchunked _Sender, class _Env>
   auto __parallel_scheduler_domain::transform_sender(_Sender&& __sndr, const _Env& __env)
     const noexcept {

include/stdexec/__detail/__bulk.hpp

@@ -141,7 +141,7 @@ namespace stdexec {
     template <>
     struct __bulk_traits<bulk_unchunked_t> {
       using __on_not_callable =
-        __callable_error<"In stdexec::bulk_unchunked(Sender, Shape, Function)..."_mstr>;
+        __callable_error<"In stdexec::bulk_unchunked(Sender, Policy, Shape, Function)..."_mstr>;
 
       // Curried function, after passing the required indices.
       template <class _Fun, class _Shape>
@@ -256,30 +256,7 @@ namespace stdexec {
     };
 
     struct bulk_chunked_t : __generic_bulk_t<bulk_chunked_t> { };
-
-    struct bulk_unchunked_t {
-      template <sender _Sender, integral _Shape, copy_constructible _Fun>
-      STDEXEC_ATTRIBUTE(host, device)
-      auto operator()(_Sender&& __sndr, _Shape __shape, _Fun __fun) const -> __well_formed_sender
-        auto {
-        auto __domain = __get_early_domain(__sndr);
-        return stdexec::transform_sender(
-          __domain,
-          __make_sexpr<bulk_unchunked_t>(
-            __data{par, __shape, static_cast<_Fun&&>(__fun)}, static_cast<_Sender&&>(__sndr)));
-      }
-
-      template <integral _Shape, copy_constructible _Fun>
-      STDEXEC_ATTRIBUTE(always_inline)
-      auto operator()(_Shape __shape, _Fun __fun) const
-        -> __binder_back<bulk_unchunked_t, _Shape, _Fun> {
-        return {
-          {static_cast<_Shape&&>(__shape), static_cast<_Fun&&>(__fun)},
-          {},
-          {}
-        };
-      }
-    };
+    struct bulk_unchunked_t : __generic_bulk_t<bulk_unchunked_t> { };
 
     template <class _AlgoTag>
     struct __bulk_impl_base : __sexpr_defaults {
@@ -340,7 +317,6 @@ namespace stdexec {
       //! This implements the core default behavior for `bulk_unchunked`:
       //! When setting value, it loops over the shape and invokes the function.
       //! Note: This is not done in concurrently. That is customized by the scheduler.
-      //! Calling it on a scheduler that is not concurrent is an error.
       static constexpr auto complete =
         []<class _Tag, class _State, class _Receiver, class... _Args>(
           __ignore,
@@ -349,17 +325,7 @@ namespace stdexec {
           _Tag,
           _Args&&... __args) noexcept -> void {
         if constexpr (std::same_as<_Tag, set_value_t>) {
-          // Intercept set_value and dispatch to the bulk operation.
           using __shape_t = decltype(__state.__shape_);
-          constexpr bool __scheduler_available = requires {
-            get_completion_scheduler<set_value_t>(get_env(__rcvr));
-          };
-          if constexpr (__scheduler_available) {
-            // This default implementation doesn't run a scheduler with concurrent progres guarantees.
-            constexpr auto __guarantee = get_forward_progress_guarantee(
-              get_completion_scheduler<set_value_t>(get_env(__rcvr)));
-            static_assert(__guarantee != forward_progress_guarantee::concurrent);
-          }
           if constexpr (noexcept(__state.__fun_(__shape_t{}, __args...))) {
             // The noexcept version that doesn't need try/catch:
             for (__shape_t __i{}; __i != __state.__shape_; ++__i) {

test/exec/test_system_context.cpp

@@ -221,7 +221,7 @@ TEST_CASE("simple bulk_unchunked task on system context", "[types][system_schedu
   std::thread::id pool_ids[num_tasks];
   exec::parallel_scheduler sched = exec::get_parallel_scheduler();
 
-  auto bulk_snd = ex::bulk_unchunked(ex::schedule(sched), num_tasks, [&](unsigned long id) {
+  auto bulk_snd = ex::bulk_unchunked(ex::schedule(sched), ex::par, num_tasks, [&](unsigned long id) {
     pool_ids[id] = std::this_thread::get_id();
   });
 
@@ -233,6 +233,26 @@ TEST_CASE("simple bulk_unchunked task on system context", "[types][system_schedu
   }
 }
 
+TEST_CASE("bulk_unchunked with seq will run everything on one thread", "[types][system_scheduler]") {
+  std::thread::id this_id = std::this_thread::get_id();
+  constexpr size_t num_tasks = 16;
+  std::thread::id pool_ids[num_tasks];
+  exec::parallel_scheduler sched = exec::get_parallel_scheduler();
+
+  auto bulk_snd = ex::bulk_unchunked(ex::schedule(sched), ex::seq, num_tasks, [&](unsigned long id) {
+    pool_ids[id] = std::this_thread::get_id();
+    std::this_thread::sleep_for(std::chrono::milliseconds{1});
+  });
+
+  ex::sync_wait(std::move(bulk_snd));
+
+  for (auto pool_id: pool_ids) {
+    REQUIRE(pool_id != std::thread::id{});
+    REQUIRE(this_id != pool_id);
+    REQUIRE(pool_id == pool_ids[0]); // All should be the same
+  }
+}
+
 TEST_CASE("bulk_chunked on parallel_scheduler performs chunking", "[types][system_scheduler]") {
   std::atomic<bool> has_chunking = false;