From 08a587972a7955bce188f131ad37ec164b539fab Mon Sep 17 00:00:00 2001 From: Cra3z Date: Mon, 24 Aug 2026 13:54:52 +0800 Subject: [PATCH 1/2] Fix `exec::thread_pool_base` customization of `bulk_unchunked` and correctly handle non-parallel execution policies --- include/exec/thread_pool_base.hpp | 72 +++++++++++++++++++++++++------ 1 file changed, 58 insertions(+), 14 deletions(-) diff --git a/include/exec/thread_pool_base.hpp b/include/exec/thread_pool_base.hpp index ed4f44c6f..b964cd0d4 100644 --- a/include/exec/thread_pool_base.hpp +++ b/include/exec/thread_pool_base.hpp @@ -28,6 +28,7 @@ import stdexec; #else # include "../stdexec/__detail/__execution_fwd.hpp" +# include "../stdexec/__detail/__bulk.hpp" # include "../stdexec/__detail/__connect.hpp" # include "../stdexec/__detail/__env.hpp" # include "../stdexec/__detail/__meta.hpp" @@ -81,7 +82,9 @@ namespace experimental::execution struct domain : STDEXEC::default_domain { - template Sender, class Env> + template + requires sender_for + || sender_for static constexpr auto transform_sender(STDEXEC::set_value_t, Sender&& sndr, Env const & env) { auto& [tag, data, child] = sndr; @@ -91,12 +94,33 @@ namespace experimental::execution { auto sch = STDEXEC::get_completion_scheduler(STDEXEC::get_env(child), env); - using sender_t = - scheduler::template bulk_sender_t; - return sender_t{*sch.pool_, - STDEXEC::__forward_like(child), - shape, - STDEXEC::__forward_like(fun)}; + + using policy_type = STDEXEC::__decay_t; + constexpr bool parallelize = + STDEXEC::__same_as + || STDEXEC::__same_as; + + if constexpr (sender_for) + { + using sender_t = + scheduler::template bulk_sender_t; + return sender_t{*sch.pool_, + STDEXEC::__forward_like(child), + shape, + STDEXEC::__forward_like(fun), + parallelize}; + } + else + { + using fun_t = STDEXEC::__bulk::__as_bulk_chunked_fn>; + using sender_t = + scheduler::template bulk_sender_t; + return sender_t{*sch.pool_, + STDEXEC::__forward_like(child), + shape, + fun_t{STDEXEC::__forward_like(fun)}, + parallelize}; + } } else { @@ -110,9 +134,6 @@ namespace experimental::execution STDEXEC::_WITH_ENVIRONMENT_(Env)>(); } } - - template Sender, class Env> - static constexpr auto transform_sender(STDEXEC::set_value_t, Sender&& sndr, Env const & env); }; struct scheduler @@ -197,6 +218,7 @@ namespace experimental::execution Receiver rcvr_; Shape shape_; Fun fun_; + bool parallelize_; std::atomic finished_threads_{0}; std::atomic thread_with_exception_{0}; @@ -205,6 +227,11 @@ namespace experimental::execution [[nodiscard]] auto num_agents_required() const -> std::uint32_t { + if (!parallelize_) + { + return 1; + } + // With work stealing, is std::min necessary, or can we feel free to ask for more agents (tasks) // than we can actually deal with at one time? return static_cast( @@ -219,11 +246,16 @@ namespace experimental::execution data_); } - explicit bulk_shared_state(DerivedPoolType& pool, Receiver rcvr, Shape shape, Fun fun) + explicit bulk_shared_state(DerivedPoolType& pool, + Receiver rcvr, + Shape shape, + Fun fun, + bool parallelize) : pool_(pool) , rcvr_{static_cast(rcvr)} , shape_{shape} , fun_{fun} + , parallelize_{parallelize} , thread_with_exception_{num_agents_required()} { this->execute_ = [](_pool_::task_base* t, std::uint32_t tid) noexcept @@ -379,8 +411,13 @@ namespace experimental::execution STDEXEC::start(inner_op_); } - bulk_opstate(DerivedPoolType& pool, Shape shape, Fun fun, CvSender&& sndr, Receiver rcvr) - : shared_state_(pool, static_cast(rcvr), shape, fun) + bulk_opstate(DerivedPoolType& pool, + Shape shape, + Fun fun, + bool parallelize, + CvSender&& sndr, + Receiver rcvr) + : shared_state_(pool, static_cast(rcvr), shape, fun, parallelize) , inner_op_{STDEXEC::connect(static_cast(sndr), bulk_rcvr{shared_state_})} {} @@ -414,11 +451,16 @@ namespace experimental::execution using bulk_opstate_t = bulk_opstate, Receiver, Shape, Fun>; - explicit bulk_sender(DerivedPoolType& pool, Sender sndr, Shape shape, Fun fun) + explicit bulk_sender(DerivedPoolType& pool, + Sender sndr, + Shape shape, + Fun fun, + bool parallelize) : pool_(pool) , sndr_(std::move(sndr)) , shape_(shape) , fun_(std::move(fun)) + , parallelize_(parallelize) {} template Self, STDEXEC::receiver Receiver> @@ -436,6 +478,7 @@ namespace experimental::execution return bulk_opstate_t{self.pool_, self.shape_, static_cast(self).fun_, + self.parallelize_, static_cast(self).sndr_, static_cast(rcvr)}; } @@ -473,6 +516,7 @@ namespace experimental::execution Sender sndr_; Shape shape_; Fun fun_; + bool parallelize_; }; template From 72147215728b4840a3e901542d3266d3d547c0a4 Mon Sep 17 00:00:00 2001 From: Cra3z Date: Mon, 24 Aug 2026 17:57:02 +0800 Subject: [PATCH 2/2] `num_agents_required()` should be 0 when shape is 0 --- include/exec/thread_pool_base.hpp | 9 +++------ 1 file changed, 3 insertions(+), 6 deletions(-) diff --git a/include/exec/thread_pool_base.hpp b/include/exec/thread_pool_base.hpp index b964cd0d4..f1f4d51bc 100644 --- a/include/exec/thread_pool_base.hpp +++ b/include/exec/thread_pool_base.hpp @@ -227,15 +227,12 @@ namespace experimental::execution [[nodiscard]] auto num_agents_required() const -> std::uint32_t { - if (!parallelize_) - { - return 1; - } + auto const parallelism = parallelize_ ? pool_.available_parallelism() + : static_cast(1); // With work stealing, is std::min necessary, or can we feel free to ask for more agents (tasks) // than we can actually deal with at one time? - return static_cast( - (std::min) (shape_, static_cast(pool_.available_parallelism()))); + return static_cast((std::min) (shape_, static_cast(parallelism))); } template