bulk_unchunked also takes an execution policy, and it's not restricted to concurrent execution.

lucteo · lucteo · commit cf0a38fb2964 · 2025-06-25T22:11:21.000+03:00
diff --git a/include/exec/system_context.hpp b/include/exec/system_context.hpp
@@ -374,9 +374,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 +471,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 +685,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_;
diff --git a/include/stdexec/__detail/__bulk.hpp b/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) {
diff --git a/test/exec/test_system_context.cpp b/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();
   });
 
diff --git a/test/stdexec/algos/adaptors/test_bulk.cpp b/test/stdexec/algos/adaptors/test_bulk.cpp
@@ -75,7 +75,7 @@ namespace {
   }
 
   TEST_CASE("bulk_unchunked returns a sender", "[adaptors][bulk]") {
-    auto snd = ex::bulk_unchunked(ex::just(19), 8, [](int, int) { });
+    auto snd = ex::bulk_unchunked(ex::just(19), ex::par, 8, [](int, int) { });
     static_assert(ex::sender<decltype(snd)>);
     (void) snd;
   }
@@ -93,7 +93,7 @@ namespace {
   }
 
   TEST_CASE("bulk_unchunked with environment returns a sender", "[adaptors][bulk]") {
-    auto snd = ex::bulk_unchunked(ex::just(19), 8, [](int, int) { });
+    auto snd = ex::bulk_unchunked(ex::just(19), ex::par, 8, [](int, int) { });
     static_assert(ex::sender_in<decltype(snd), ex::env<>>);
     (void) snd;
   }
@@ -109,7 +109,7 @@ namespace {
   }
 
   TEST_CASE("bulk_unchunked can be piped", "[adaptors][bulk]") {
-    ex::sender auto snd = ex::just() | ex::bulk_unchunked(42, [](int) { });
+    ex::sender auto snd = ex::just() | ex::bulk_unchunked(ex::par, 42, [](int) { });
     (void) snd;
   }
 
@@ -135,11 +135,11 @@ namespace {
 
   TEST_CASE("bulk_unchunked keeps values_type from input sender", "[adaptors][bulk]") {
     constexpr int n = 42;
-    check_val_types<ex::__mset<pack<>>>(ex::just() | ex::bulk_unchunked(n, [](int) { }));
+    check_val_types<ex::__mset<pack<>>>(ex::just() | ex::bulk_unchunked(ex::par, n, [](int) { }));
     check_val_types<ex::__mset<pack<double>>>(
-      ex::just(4.2) | ex::bulk_unchunked(n, [](int, double) { }));
+      ex::just(4.2) | ex::bulk_unchunked(ex::par, n, [](int, double) { }));
     check_val_types<ex::__mset<pack<double, std::string>>>(
-      ex::just(4.2, std::string{}) | ex::bulk_unchunked(n, [](int, double, std::string) { }));
+      ex::just(4.2, std::string{}) | ex::bulk_unchunked(ex::par, n, [](int, double, std::string) { }));
   }
 
   TEST_CASE("bulk keeps error_types from input sender", "[adaptors][bulk]") {
@@ -193,17 +193,17 @@ namespace {
     error_scheduler<int> sched3{43};
 
     check_err_types<ex::__mset<>>(
-      ex::transfer_just(sched1) | ex::bulk_unchunked(n, [](int) noexcept { }));
+      ex::transfer_just(sched1) | ex::bulk_unchunked(ex::par, n, [](int) noexcept { }));
     check_err_types<ex::__mset<std::exception_ptr>>(
-      ex::transfer_just(sched2) | ex::bulk_unchunked(n, [](int) noexcept { }));
+      ex::transfer_just(sched2) | ex::bulk_unchunked(ex::par, n, [](int) noexcept { }));
     check_err_types<ex::__mset<int>>(
-      ex::just_error(n) | ex::bulk_unchunked(n, [](int) noexcept { }));
+      ex::just_error(n) | ex::bulk_unchunked(ex::par, n, [](int) noexcept { }));
     check_err_types<ex::__mset<int>>(
-      ex::transfer_just(sched3) | ex::bulk_unchunked(n, [](int) noexcept { }));
+      ex::transfer_just(sched3) | ex::bulk_unchunked(ex::par, n, [](int) noexcept { }));
 #if !STDEXEC_STD_NO_EXCEPTIONS()
     check_err_types<ex::__mset<std::exception_ptr, int>>(
       ex::transfer_just(sched3)
-      | ex::bulk_unchunked(n, [](int) { throw std::logic_error{"err"}; }));
+      | ex::bulk_unchunked(ex::par, n, [](int) { throw std::logic_error{"err"}; }));
 #endif
   }
 
@@ -241,7 +241,7 @@ namespace {
     static int counter3[n]{};
     std::fill_n(counter3, n, 0);
 
-    ex::sender auto snd = ex::just() | ex::bulk_unchunked(n, function<int, n, counter3>);
+    ex::sender auto snd = ex::just() | ex::bulk_unchunked(ex::par, n, function<int, n, counter3>);
     auto op = ex::connect(std::move(snd), expect_void_receiver{});
     ex::start(op);
 
@@ -283,7 +283,7 @@ namespace {
     int counter[n]{0};
     function_object_t<int> fn{counter};
 
-    ex::sender auto snd = ex::just() | ex::bulk_unchunked(n, fn);
+    ex::sender auto snd = ex::just() | ex::bulk_unchunked(ex::par, n, fn);
     auto op = ex::connect(std::move(snd), expect_void_receiver{});
     ex::start(op);
 
@@ -325,7 +325,7 @@ namespace {
     constexpr int n = 9;
     int counter[n]{0};
 
-    ex::sender auto snd = ex::just() | ex::bulk_unchunked(n, [&](int i) { counter[i]++; });
+    ex::sender auto snd = ex::just() | ex::bulk_unchunked(ex::par, n, [&](int i) { counter[i]++; });
     auto op = ex::connect(std::move(snd), expect_void_receiver{});
     ex::start(op);
 
@@ -409,7 +409,7 @@ namespace {
     constexpr int magic_number = 42;
     int counter[n]{0};
 
-    auto snd = ex::just(magic_number) | ex::bulk_unchunked(n, [&](int i, int val) {
+    auto snd = ex::just(magic_number) | ex::bulk_unchunked(ex::par, n, [&](int i, int val) {
                  if (val == magic_number) {
                    counter[i]++;
                  }
@@ -459,7 +459,7 @@ namespace {
     std::iota(vals_expected.begin(), vals_expected.end(), 0);
 
     auto snd = ex::just(std::move(vals))
-             | ex::bulk_unchunked(n, [&](std::size_t i, std::vector<int>& vals) {
+             | ex::bulk_unchunked(ex::par, n, [&](std::size_t i, std::vector<int>& vals) {
                  vals[i] = static_cast<int>(i);
                });
     auto op = ex::connect(std::move(snd), expect_value_receiver{vals_expected});
@@ -494,7 +494,7 @@ namespace {
     constexpr int n = 2;
 
     auto snd = ex::just(magic_number)
-             | ex::bulk_unchunked(n, [](int, int) { return function_object_t<int>{nullptr}; });
+             | ex::bulk_unchunked(ex::par, n, [](int, int) { return function_object_t<int>{nullptr}; });
 
     auto op = ex::connect(std::move(snd), expect_value_receiver{magic_number});
     ex::start(op);
@@ -522,7 +522,7 @@ namespace {
     constexpr int n = 2;
 
     auto snd = ex::just()
-             | ex::bulk_unchunked(n, [](int) -> int { throw std::logic_error{"err"}; });
+             | ex::bulk_unchunked(ex::par, n, [](int) -> int { throw std::logic_error{"err"}; });
     auto op = ex::connect(std::move(snd), expect_error_receiver{});
     ex::start(op);
   }
@@ -553,7 +553,7 @@ namespace {
     int called{};
 
     auto snd = ex::just_error(std::string{"err"})
-             | ex::bulk_unchunked(n, [&called](int) { called++; });
+             | ex::bulk_unchunked(ex::par, n, [&called](int) { called++; });
     auto op = ex::connect(std::move(snd), expect_error_receiver{std::string{"err"}});
     ex::start(op);
   }
@@ -580,7 +580,7 @@ namespace {
     constexpr int n = 2;
     int called{};
 
-    auto snd = ex::just_stopped() | ex::bulk_unchunked(n, [&called](int) { called++; });
+    auto snd = ex::just_stopped() | ex::bulk_unchunked(ex::par, n, [&called](int) { called++; });
     auto op = ex::connect(std::move(snd), expect_stopped_receiver{});
     ex::start(op);
   }
@@ -819,8 +819,8 @@ namespace {
         std::vector<int> counter(n, 42);
 
         auto snd = ex::transfer_just(sch)
-                 | ex::bulk_unchunked(n, [&counter](std::size_t idx) { counter[idx] = 0; })
-                 | ex::bulk_unchunked(n, [&counter](std::size_t idx) { counter[idx]++; });
+                 | ex::bulk_unchunked(ex::par, n, [&counter](std::size_t idx) { counter[idx] = 0; })
+                 | ex::bulk_unchunked(ex::par, n, [&counter](std::size_t idx) { counter[idx]++; });
         stdexec::sync_wait(std::move(snd));
 
         const std::size_t actual = static_cast<std::size_t>(
@@ -836,14 +836,14 @@ namespace {
         std::vector<int> counter(n, 42);
 
         auto snd = ex::transfer_just(sch, 42)
-                 | ex::bulk_unchunked(
+                 | ex::bulk_unchunked(ex::par, 
                      n,
                      [&counter](std::size_t idx, int val) {
                        if (val == 42) {
                          counter[idx] = 0;
                        }
                      })
-                 | ex::bulk_unchunked(n, [&counter](std::size_t idx, int val) {
+                 | ex::bulk_unchunked(ex::par, n, [&counter](std::size_t idx, int val) {
                      if (val == 42) {
                        counter[idx]++;
                      }
@@ -868,9 +868,9 @@ namespace {
 
         auto snd =
           ex::transfer_just(sch, std::move(vals))
-          | ex::bulk_unchunked(
+          | ex::bulk_unchunked(ex::par, 
             n, [](std::size_t idx, std::vector<int>& vals) { vals[idx] = static_cast<int>(idx); })
-          | ex::bulk_unchunked(n, [](std::size_t idx, std::vector<int>& vals) { ++vals[idx]; });
+          | ex::bulk_unchunked(ex::par, n, [](std::size_t idx, std::vector<int>& vals) { ++vals[idx]; });
         auto [vals_actual] = stdexec::sync_wait(std::move(snd)).value();
 
         CHECK(vals_actual == vals_expected);
@@ -881,7 +881,7 @@ namespace {
     SECTION("With exception") {
       constexpr int n = 9;
       auto snd = ex::transfer_just(sch)
-               | ex::bulk_unchunked(n, [](int) { throw std::runtime_error("bulk_unchunked"); });
+               | ex::bulk_unchunked(ex::par, n, [](int) { throw std::runtime_error("bulk_unchunked"); });
 
       CHECK_THROWS_AS(stdexec::sync_wait(std::move(snd)), std::runtime_error);
     }
@@ -894,9 +894,9 @@ namespace {
 
       stdexec::sender auto snd = stdexec::when_all(
         stdexec::schedule(sch)
-          | stdexec::bulk_unchunked(n, [&](std::size_t id) { counters_1[id]++; }),
+          | stdexec::bulk_unchunked(ex::par, n, [&](std::size_t id) { counters_1[id]++; }),
         stdexec::schedule(sch)
-          | stdexec::bulk_unchunked(n, [&](std::size_t id) { counters_2[id]++; }));
+          | stdexec::bulk_unchunked(ex::par, n, [&](std::size_t id) { counters_2[id]++; }));
 
       stdexec::sync_wait(std::move(snd));
 
@@ -1021,7 +1021,7 @@ namespace {
     "default bulk_unchunked works with non-default constructible types",
     "[adaptors][bulk]") {
     ex::sender auto s = ex::just(non_default_constructible{42})
-                      | ex::bulk_unchunked(1, [](int, auto&) { });
+                      | ex::bulk_unchunked(ex::par, 1, [](int, auto&) { });
     ex::sync_wait(std::move(s));
   }
 
