Customize bulk_chunked instead of bulk for static thread pool.

Author: lucteo

include/exec/static_thread_pool.hpp

@@ -195,11 +195,11 @@ namespace exec {
 #endif
 
       template <class Fun, class Shape, class... Args>
-        requires __callable<Fun, Shape, Args&...>
+        requires __callable<Fun, Shape, Shape, Args&...>
       using bulk_non_throwing = //
         __mbool<
           // If function invocation doesn't throw
-          __nothrow_callable<Fun, Shape, Args&...> &&
+          __nothrow_callable<Fun, Shape, Shape, Args&...> &&
       // and emplacing a tuple doesn't throw
 #if STDEXEC_MSVC()
           __bulk_non_throwing<Args...>::__v
@@ -236,7 +236,7 @@ namespace exec {
 
       struct transform_bulk {
         template <class Data, class Sender>
-        auto operator()(bulk_t, Data&& data, Sender&& sndr) {
+        auto operator()(bulk_chunked_t, Data&& data, Sender&& sndr) {
           auto [pol, shape, fun] = static_cast<Data&&>(data);
           // TODO: handle non-par execution policies
           return bulk_sender_t<Sender, decltype(shape), decltype(fun)>{
@@ -265,7 +265,7 @@ namespace exec {
      public:
       struct domain : stdexec::default_domain {
         // For eager customization
-        template <sender_expr_for<bulk_t> Sender>
+        template <sender_expr_for<bulk_chunked_t> Sender>
         auto transform_sender(Sender&& sndr) const noexcept {
           if constexpr (__completes_on<Sender, static_thread_pool_::scheduler>) {
             auto sched = get_completion_scheduler<set_value_t>(get_env(sndr));
@@ -279,8 +279,8 @@ namespace exec {
           }
         }
 
-        // transform the generic bulk sender into a parallel thread-pool bulk sender
-        template <sender_expr_for<bulk_t> Sender, class Env>
+        // transform the generic bulk_chunked sender into a parallel thread-pool bulk sender
+        template <sender_expr_for<bulk_chunked_t> Sender, class Env>
         auto transform_sender(Sender&& sndr, const Env& env) const noexcept {
           if constexpr (__completes_on<Sender, static_thread_pool_::scheduler>) {
             auto sched = get_completion_scheduler<set_value_t>(get_env(sndr));
@@ -679,9 +679,8 @@ namespace exec {
 
       for (std::uint32_t index = 0; index < threadCount; ++index) {
         threadStates_[index].emplace(this, index, params, numa_);
-        threadIndexByNumaNode_.push_back(
-          thread_index_by_numa_node{
-            .numa_node = threadStates_[index]->numa_node(), .thread_index = index});
+        threadIndexByNumaNode_.push_back(thread_index_by_numa_node{
+          .numa_node = threadStates_[index]->numa_node(), .thread_index = index});
       }
 
       // NOLINTNEXTLINE(modernize-use-ranges) we still support platforms without the std::ranges algorithms
@@ -1158,9 +1157,7 @@ namespace exec {
               // In the case that the shape is much larger than the total number of threads,
               // then each call to computation will call the function many times.
               auto [begin, end] = even_share(sh_state.shape_, tid, total_threads);
-              for (Shape i = begin; i < end; ++i) {
-                sh_state.fun_(i, args...);
-              }
+              sh_state.fun_(begin, end, args...);
             };
 
             auto completion = [&](auto&... args) {

test/stdexec/algos/adaptors/test_bulk.cpp

@@ -892,7 +892,6 @@ namespace {
     }
   }
 
-  // TODO: also add similar tests for bulk_chunked and bulk_unchunked
   TEST_CASE("eager customization of bulk works with static thread pool", "[adaptors][bulk]") {
     exec::static_thread_pool pool{4};
     ex::scheduler auto sch = pool.get_scheduler();
@@ -907,7 +906,6 @@ namespace {
 
       auto snd = ex::just() //
                | ex::continues_on(sch) | ex::bulk(ex::par, tids.size(), fun);
-      CHECK(std::equal_to<void*>()(&snd.pool_, &pool));
       stdexec::sync_wait(std::move(snd));
 
       // All the work should not have run on the same thread
@@ -918,7 +916,34 @@ namespace {
     }
   }
 
-  // TODO: also add similar tests for bulk_chunked and bulk_unchunked
+  TEST_CASE(
+    "eager customization of bulk_chunked works with static thread pool",
+    "[adaptors][bulk]") {
+    exec::static_thread_pool pool{4};
+    ex::scheduler auto sch = pool.get_scheduler();
+
+    SECTION("Without values in the set_value channel") {
+      std::vector<std::thread::id> tids(42);
+
+      auto fun = [&tids](std::size_t b, std::size_t e) {
+        while (b < e) {
+          tids[b++] = std::this_thread::get_id();
+        }
+        std::this_thread::sleep_for(std::chrono::milliseconds{10});
+      };
+
+      auto snd = ex::just() //
+               | ex::continues_on(sch) | ex::bulk_chunked(ex::par, tids.size(), fun);
+      stdexec::sync_wait(std::move(snd));
+
+      // All the work should not have run on the same thread
+      const auto actual = static_cast<std::size_t>(std::count(tids.begin(), tids.end(), tids[0]));
+      const std::size_t wrong = tids.size();
+
+      CHECK(actual != wrong);
+    }
+  }
+
   TEST_CASE("lazy customization of bulk works with static thread pool", "[adaptors][bulk]") {
     exec::static_thread_pool pool{4};
     ex::scheduler auto sch = pool.get_scheduler();
@@ -943,6 +968,32 @@ namespace {
     }
   }
 
+  TEST_CASE("lazy customization of bulk_chunked works with static thread pool", "[adaptors][bulk]") {
+    exec::static_thread_pool pool{4};
+    ex::scheduler auto sch = pool.get_scheduler();
+
+    SECTION("Without values in the set_value channel") {
+      std::vector<std::thread::id> tids(42);
+
+      auto fun = [&tids](std::size_t b, std::size_t e) {
+        while (b < e) {
+          tids[b++] = std::this_thread::get_id();
+        }
+        std::this_thread::sleep_for(std::chrono::milliseconds{10});
+      };
+
+      auto snd = ex::just() //
+               | ex::bulk_chunked(ex::par, tids.size(), fun);
+      stdexec::sync_wait(stdexec::starts_on(sch, std::move(snd)));
+
+      // All the work should not have run on the same thread
+      const auto actual = static_cast<std::size_t>(std::count(tids.begin(), tids.end(), tids[0]));
+      const std::size_t wrong = tids.size();
+
+      CHECK(actual != wrong);
+    }
+  }
+
   TEST_CASE("default bulk works with non-default constructible types", "[adaptors][bulk]") {
     ex::sender auto s = ex::just(non_default_constructible{42})
                       | ex::bulk(ex::par, 1, [](int, auto&) { });
@@ -986,8 +1037,7 @@ namespace {
     auto snd = ex::just(std::string{"hello"})
              | exec::on(
                  sched, //
-                 ex::bulk(ex::par, 1, [&called](int, std::string x) { called = true; }))
-             | exec::write(stdexec::prop{ex::get_scheduler, inline_scheduler()});
+                 ex::bulk(ex::par, 1, [&called](int, std::string x) { called = true; }));
     wait_for_value(std::move(snd), std::string{"hijacked"});
     REQUIRE_FALSE(called);
   }