diff --git a/examples/benchmark/asio_thread_pool.cpp b/examples/benchmark/asio_thread_pool.cpp index ef693fe9f1..77fc6c89e9 100644 --- a/examples/benchmark/asio_thread_pool.cpp +++ b/examples/benchmark/asio_thread_pool.cpp @@ -47,7 +47,7 @@ struct RunThread buffer.size(), pmr::null_memory_resource()}; pmr::polymorphic_allocator alloc(&resource); - auto [start, end] = exec::_pool_::even_share(total_scheds, tid, pool.available_parallelism()); + auto [start, end] = exec::_pool::_even_share(total_scheds, tid, pool.available_parallelism()); std::size_t scheds = end - start; std::atomic counter{scheds}; auto env = stdexec::prop{stdexec::get_allocator, alloc}; @@ -68,7 +68,7 @@ struct RunThread --scheds; } #else - auto [start, end] = exec::_pool_::even_share(total_scheds, tid, pool.available_parallelism()); + auto [start, end] = exec::_pool::_even_share(total_scheds, tid, pool.available_parallelism()); std::size_t scheds = end - start; std::atomic counter{scheds}; while (scheds) diff --git a/examples/benchmark/static_thread_pool.cpp b/examples/benchmark/static_thread_pool.cpp index 9ffbb2480b..d332c7c5a0 100644 --- a/examples/benchmark/static_thread_pool.cpp +++ b/examples/benchmark/static_thread_pool.cpp @@ -49,7 +49,7 @@ struct RunThread buffer.size(), pmr::null_memory_resource()}; pmr::polymorphic_allocator alloc(&resource); - auto [start, end] = exec::_pool_::even_share(total_scheds, tid, pool.available_parallelism()); + auto [start, end] = exec::_pool::_even_share(total_scheds, tid, pool.available_parallelism()); std::size_t scheds = end - start; std::atomic counter{scheds}; auto env = stdexec::prop{stdexec::get_allocator, alloc}; @@ -70,7 +70,7 @@ struct RunThread --scheds; } #else - auto [start, end] = exec::_pool_::even_share(total_scheds, tid, pool.available_parallelism()); + auto [start, end] = exec::_pool::_even_share(total_scheds, tid, pool.available_parallelism()); std::size_t scheds = end - start; std::atomic counter{scheds}; while (scheds) @@ -110,4 +110,4 @@ auto main(int argc, char** argv) -> int { my_numa_distribution numa{}; my_main(argc, argv, numa); -} \ No newline at end of file +} diff --git a/examples/benchmark/static_thread_pool_bulk_enqueue.cpp b/examples/benchmark/static_thread_pool_bulk_enqueue.cpp index 61fd9530fe..7e46d34f21 100644 --- a/examples/benchmark/static_thread_pool_bulk_enqueue.cpp +++ b/examples/benchmark/static_thread_pool_bulk_enqueue.cpp @@ -45,11 +45,11 @@ struct RunThread pmr::monotonic_buffer_resource rsrc{buffer.data(), buffer.size()}; pmr::polymorphic_allocator alloc{&rsrc}; auto env = stdexec::prop{stdexec::get_allocator, alloc}; - auto [start, end] = exec::_pool_::even_share(total_scheds, tid, pool.available_parallelism()); + 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() | stdexec::write_env(env); #else - auto [start, end] = exec::_pool_::even_share(total_scheds, tid, pool.available_parallelism()); + 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(); #endif diff --git a/examples/benchmark/static_thread_pool_bulk_enqueue_nested.cpp b/examples/benchmark/static_thread_pool_bulk_enqueue_nested.cpp index f6fdd13ba8..68519e5295 100644 --- a/examples/benchmark/static_thread_pool_bulk_enqueue_nested.cpp +++ b/examples/benchmark/static_thread_pool_bulk_enqueue_nested.cpp @@ -46,11 +46,11 @@ struct RunThread pmr::monotonic_buffer_resource rsrc{buffer.data(), buffer.size()}; pmr::polymorphic_allocator alloc{&rsrc}; auto env = stdexec::prop{stdexec::get_allocator, alloc}; - auto [start, end] = exec::_pool_::even_share(total_scheds, tid, pool.available_parallelism()); + 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() | stdexec::write_env(env); #else - auto [start, end] = exec::_pool_::even_share(total_scheds, tid, pool.available_parallelism()); + 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(); #endif stdexec::sync_wait(stdexec::starts_on(scheduler, iterate)); diff --git a/examples/benchmark/static_thread_pool_nested.cpp b/examples/benchmark/static_thread_pool_nested.cpp index 8e08218fb7..0e7a6b85a9 100644 --- a/examples/benchmark/static_thread_pool_nested.cpp +++ b/examples/benchmark/static_thread_pool_nested.cpp @@ -47,7 +47,7 @@ struct RunThread buffer.size(), pmr::null_memory_resource()}; pmr::polymorphic_allocator alloc(&resource); - auto [start, end] = exec::_pool_::even_share(total_scheds, tid, pool.available_parallelism()); + auto [start, end] = exec::_pool::_even_share(total_scheds, tid, pool.available_parallelism()); std::size_t scheds = end - start; std::atomic counter{scheds}; auto env = stdexec::prop{stdexec::get_allocator, alloc}; @@ -74,7 +74,7 @@ struct RunThread } })); #else - auto [start, end] = exec::_pool_::even_share(total_scheds, tid, pool.available_parallelism()); + auto [start, end] = exec::_pool::_even_share(total_scheds, tid, pool.available_parallelism()); std::size_t scheds = end - start; std::atomic counter{scheds}; stdexec::sync_wait(stdexec::schedule(scheduler) @@ -110,4 +110,4 @@ struct RunThread auto main(int argc, char** argv) -> int { my_main(argc, argv); -} \ No newline at end of file +} diff --git a/examples/benchmark/taskflow_thread_pool.cpp b/examples/benchmark/taskflow_thread_pool.cpp index 10db91980e..3ae5a21a3a 100644 --- a/examples/benchmark/taskflow_thread_pool.cpp +++ b/examples/benchmark/taskflow_thread_pool.cpp @@ -47,7 +47,7 @@ struct RunThread buffer.size(), pmr::null_memory_resource()}; pmr::polymorphic_allocator alloc(&resource); - auto [start, end] = exec::_pool_::even_share(total_scheds, tid, pool.available_parallelism()); + auto [start, end] = exec::_pool::_even_share(total_scheds, tid, pool.available_parallelism()); std::size_t scheds = end - start; std::atomic counter{scheds}; auto env = stdexec::prop{stdexec::get_allocator, alloc}; @@ -68,7 +68,7 @@ struct RunThread --scheds; } #else - auto [start, end] = exec::_pool_::even_share(total_scheds, tid, pool.available_parallelism()); + auto [start, end] = exec::_pool::_even_share(total_scheds, tid, pool.available_parallelism()); std::size_t scheds = end - start; std::atomic counter{scheds}; while (scheds) diff --git a/examples/benchmark/tbb_thread_pool.cpp b/examples/benchmark/tbb_thread_pool.cpp index 1d3514ee5b..b1ab0f73f2 100644 --- a/examples/benchmark/tbb_thread_pool.cpp +++ b/examples/benchmark/tbb_thread_pool.cpp @@ -47,7 +47,7 @@ struct RunThread buffer.size(), pmr::null_memory_resource()}; pmr::polymorphic_allocator alloc(&resource); - auto [start, end] = exec::_pool_::even_share(total_scheds, tid, pool.available_parallelism()); + auto [start, end] = exec::_pool::_even_share(total_scheds, tid, pool.available_parallelism()); std::size_t scheds = end - start; std::atomic counter{scheds}; auto env = stdexec::prop{stdexec::get_allocator, alloc}; @@ -68,7 +68,7 @@ struct RunThread --scheds; } #else - auto [start, end] = exec::_pool_::even_share(total_scheds, tid, pool.available_parallelism()); + auto [start, end] = exec::_pool::_even_share(total_scheds, tid, pool.available_parallelism()); std::size_t scheds = end - start; std::atomic counter{scheds}; while (scheds) diff --git a/examples/benchmark/tbb_thread_pool_nested.cpp b/examples/benchmark/tbb_thread_pool_nested.cpp index b8e448668a..2a964811ae 100644 --- a/examples/benchmark/tbb_thread_pool_nested.cpp +++ b/examples/benchmark/tbb_thread_pool_nested.cpp @@ -41,7 +41,7 @@ struct RunThread { break; } - auto [start, end] = exec::_pool_::even_share(total_scheds, tid, pool.available_parallelism()); + auto [start, end] = exec::_pool::_even_share(total_scheds, tid, pool.available_parallelism()); std::size_t scheds = end - start; tbb::task_group tg{}; stdexec::sync_wait(stdexec::schedule(scheduler) diff --git a/examples/nvexec/maxwell/std.cuh b/examples/nvexec/maxwell/std.cuh index bec9097a2a..2f497a54c6 100644 --- a/examples/nvexec/maxwell/std.cuh +++ b/examples/nvexec/maxwell/std.cuh @@ -26,7 +26,7 @@ #include template -auto even_share(Shape n, std::size_t rank, std::size_t size) noexcept -> std::pair +auto _even_share(Shape n, std::size_t rank, std::size_t size) noexcept -> std::pair { auto const avg_per_thread = n / size; auto const n_big_share = avg_per_thread + 1; @@ -64,7 +64,7 @@ void run_std(float dt, auto e_updater = update_e(time.get(), dt, accessor); auto initializer = grid_initializer(dt, accessor); - auto [begin, end] = even_share(accessor.cells, tid, n_threads); + auto [begin, end] = _even_share(accessor.cells, tid, n_threads); for (std::size_t i = begin; i < end; i++) { diff --git a/examples/nvexec/maxwell_distributed.cu b/examples/nvexec/maxwell_distributed.cu index c3158f52f4..52cfa9d8d0 100644 --- a/examples/nvexec/maxwell_distributed.cu +++ b/examples/nvexec/maxwell_distributed.cu @@ -26,7 +26,7 @@ # include # include -static auto even_share(std::size_t n, std::size_t rank, std::size_t size) noexcept +static auto _even_share(std::size_t n, std::size_t rank, std::size_t size) noexcept -> std::pair { auto const avg_per_thread = n / size; @@ -457,7 +457,7 @@ auto main(int argc, char *argv[]) -> int bool const write_wtk = value(params, "write-vtk"); std::size_t const n_iterations = value(params, "iterations", 1000); std::size_t const N = value(params, "N", 512); - auto const [row_begin, row_end] = even_share(N, rank, size); + auto const [row_begin, row_end] = _even_share(N, rank, size); std::size_t const rank_begin = row_begin * N; std::size_t const rank_end = row_end * N; distributed::grid_t grid{N, rank_begin, rank_end}; diff --git a/examples/problem.cpp b/examples/problem.cpp index a61e541c4c..8a97c6fa45 100644 --- a/examples/problem.cpp +++ b/examples/problem.cpp @@ -42,7 +42,7 @@ struct receiver_t template void check_if_pool_domain() { - using pool_domain = exec::_pool_::_static_thread_pool::domain; + using pool_domain = exec::_pool::_static_thread_pool::domain; static_assert(std::is_same_v); } diff --git a/include/exec/asio/asio_thread_pool.hpp b/include/exec/asio/asio_thread_pool.hpp index ff9ab883de..aea62d1ccb 100644 --- a/include/exec/asio/asio_thread_pool.hpp +++ b/include/exec/asio/asio_thread_pool.hpp @@ -55,9 +55,7 @@ namespace experimental::execution::asio private: friend struct ::exec::thread_pool_base; - - template - friend struct exec::_pool_::opstate; + friend struct ::exec::thread_pool_access; [[nodiscard]] static constexpr auto forward_progress_guarantee() -> STDEXEC::forward_progress_guarantee @@ -65,7 +63,7 @@ namespace experimental::execution::asio return STDEXEC::forward_progress_guarantee::parallel; } - void enqueue(exec::_pool_::task_base* task, std::uint32_t tid = 0) noexcept + void enqueue(exec::_pool::_task_base* task, std::uint32_t tid = 0) noexcept { asio_impl::post(pool_, [task, tid] { task->execute_(task, /*tid=*/tid); }); } diff --git a/include/exec/libdispatch_queue.hpp b/include/exec/libdispatch_queue.hpp index d37a994d86..ec7b53cccd 100644 --- a/include/exec/libdispatch_queue.hpp +++ b/include/exec/libdispatch_queue.hpp @@ -47,9 +47,9 @@ namespace experimental::execution namespace __libdispatch { - struct task_base + struct _task_base { - void (*execute)(task_base *) noexcept; + void (*execute)(_task_base *) noexcept; }; template @@ -224,7 +224,7 @@ namespace experimental::execution { bool operator==(libdispatch_queue const &) const = default; - void submit(__libdispatch::task_base *f) + void submit(__libdispatch::_task_base *f) { auto queue = dispatch_get_global_queue(priority, 0); dispatch_async_f(queue, f, reinterpret_cast(f->execute)); @@ -241,7 +241,7 @@ namespace experimental::execution namespace __libdispatch { template - struct operation : task_base + struct operation : _task_base { libdispatch_queue &queue; Receiver receiver; @@ -250,7 +250,7 @@ namespace experimental::execution : queue(*queue_arg) , receiver(std::move(rcvr)) { - this->execute = [](task_base *t) noexcept + this->execute = [](_task_base *t) noexcept { auto &op = *static_cast(t); auto stoken = STDEXEC::get_stop_token(STDEXEC::get_env(op.receiver)); @@ -269,7 +269,7 @@ namespace experimental::execution }; } - void enqueue(task_base *op) const + void enqueue(_task_base *op) const { queue.submit(op); } @@ -366,7 +366,7 @@ namespace experimental::execution template struct bulk_shared_state { - struct bulk_task : task_base + struct bulk_task : _task_base { bulk_shared_state *sh_state_; Shape task_id_; @@ -375,7 +375,7 @@ namespace experimental::execution : sh_state_(sh_state_arg) , task_id_(task_id_arg) { - this->execute = [](task_base *t) noexcept + this->execute = [](_task_base *t) noexcept { auto &sh_state = *static_cast(t)->sh_state_; auto task_id = static_cast(t)->task_id_; diff --git a/include/exec/static_thread_pool.hpp b/include/exec/static_thread_pool.hpp index 90205c4f5a..f110663b31 100644 --- a/include/exec/static_thread_pool.hpp +++ b/include/exec/static_thread_pool.hpp @@ -23,31 +23,26 @@ import std; import stdexec; #else -# include "../stdexec/__detail/__bulk.hpp" # include "../stdexec/__detail/__domain.hpp" # include "../stdexec/__detail/__execution_fwd.hpp" # include "../stdexec/__detail/__execution_legacy.hpp" // IWYU pragma: keep # include "../stdexec/__detail/__get_completion_signatures.hpp" # include "../stdexec/__detail/__intrusive_queue.hpp" -# include "../stdexec/__detail/__manual_lifetime.hpp" # include "../stdexec/__detail/__meta.hpp" # include "../stdexec/__detail/__optional.hpp" # include "../stdexec/__detail/__receivers.hpp" # include "../stdexec/__detail/__tuple.hpp" # include "../stdexec/__detail/__type_traits.hpp" -# include "../stdexec/__detail/__variant.hpp" # if !STDEXEC_USE_MODULES() # include # include # include -# include # include # include # include # include # include -# include # include # endif #endif @@ -57,136 +52,84 @@ import stdexec; #include "detail/numa.hpp" #include "detail/xorshift.hpp" -#include "completion_signatures.hpp" #include "sender_for.hpp" #include "sequence/iterate.hpp" -#include "sequence_senders.hpp" +#include "thread_pool_base.hpp" STDEXEC_PRAGMA_PUSH() STDEXEC_PRAGMA_IGNORE_EDG(is_constant_evaluated_in_nonconstexpr_context) namespace experimental::execution { - struct bwos_params - { - std::size_t numBlocks{32}; - std::size_t blockSize{8}; - }; - - struct CANNOT_DISPATCH_THE_BULK_ALGORITHM_TO_THE_STATIC_THREAD_POOL_SCHEDULER; - struct BECAUSE_THERE_IS_NO_STATIC_THREAD_POOL_SCHEDULER_IN_THE_ENVIRONMENT; - struct ADD_A_CONTINUES_ON_TRANSITION_TO_THE_STATIC_THREAD_POOL_SCHEDULER_BEFORE_THE_BULK_ALGORITHM; - struct CANNOT_DISPATCH_THE_ITERATE_ALGORITHM_TO_THE_STATIC_THREAD_POOL_SCHEDULER; struct BECAUSE_THERE_IS_NO_STATIC_THREAD_POOL_SCHEDULER_IN_THE_ENVIRONMENT; struct ADD_A_CONTINUES_ON_TRANSITION_TO_THE_STATIC_THREAD_POOL_SCHEDULER_BEFORE_THE_ITERATE_ALGORITHM; - namespace _pool_ + struct static_thread_pool; + + namespace _pool { using namespace STDEXEC; - //! @brief Splits `n` into `size` chunks distributing `n % size` evenly - //! between ranks. - //! - //! @returns `[begin, end)` range in `n` for a given `rank`. - //! - //! @par Example: - //! - //! @code{cpp} - //! // n_items thread n_threads - //! even_share( 11, 0, 3); // -> [0, 4) -> 4 items - //! even_share( 11, 1, 3); // -> [4, 8) -> 4 items - //! even_share( 11, 2, 3); // -> [8, 11) -> 3 items - //! @endcode - template - constexpr auto even_share(Shape n, std::size_t rank, std::size_t size) noexcept // - -> std::pair + struct _remote_queue { - STDEXEC_ASSERT(n >= 0); - using ushape_t = std::make_unsigned_t; - auto const avg_per_thread = static_cast(n) / size; - auto const n_big_share = avg_per_thread + 1; - auto const big_shares = static_cast(n) % size; - auto const is_big_share = rank < big_shares; - auto const begin = is_big_share - ? n_big_share * rank - : n_big_share * big_shares + (rank - big_shares) * avg_per_thread; - auto const end = begin + (is_big_share ? n_big_share : avg_per_thread); - - return std::make_pair(static_cast(begin), static_cast(end)); - } - - namespace schedule_all_ - { - template - struct sequence; - } // namespace schedule_all_ - - struct task_base - { - void (*execute_)(task_base*, std::uint32_t tid) noexcept = nullptr; - task_base* next_ = nullptr; - }; - - struct remote_queue - { - explicit remote_queue(std::size_t nthreads) noexcept + explicit _remote_queue(std::size_t nthreads) noexcept : queues_(nthreads) {} - explicit remote_queue(remote_queue* next, std::size_t nthreads) noexcept + explicit _remote_queue(_remote_queue* next, std::size_t nthreads) noexcept : next_(next) , queues_(nthreads) {} - remote_queue* next_{}; - std::vector<__atomic_intrusive_queue<&task_base::next_>> queues_{}; - std::thread::id id_{std::this_thread::get_id()}; + _remote_queue* next_{}; + std::vector<__atomic_intrusive_queue<&_task_base::next_>> queues_{}; + std::thread::id id_{std::this_thread::get_id()}; // This marks whether the submitter is a thread in the pool or not. std::size_t index_{(std::numeric_limits::max)()}; }; - enum class remote_poll_mode + enum class _remote_poll_mode { - speculative, - before_sleep + _speculative, + _before_sleep }; - struct remote_queue_list + struct _remote_queue_list { private: - __std::atomic head_; - remote_queue* tail_; - std::size_t nthreads_; - remote_queue this_remotes_; + __std::atomic<_remote_queue*> head_; + _remote_queue* tail_; + std::size_t nthreads_; + _remote_queue this_remotes_; public: - explicit remote_queue_list(std::size_t nthreads) noexcept + explicit _remote_queue_list(std::size_t nthreads) noexcept : head_{&this_remotes_} , tail_{&this_remotes_} , nthreads_(nthreads) , this_remotes_(nthreads) {} - ~remote_queue_list() noexcept + ~_remote_queue_list() noexcept { - remote_queue* head = head_.load(__std::memory_order_acquire); + _remote_queue* head = head_.load(__std::memory_order_acquire); while (head != tail_) { - remote_queue* tmp = std::exchange(head, head->next_); + _remote_queue* tmp = std::exchange(head, head->next_); delete tmp; } } - auto pop_all_reversed(std::size_t tid, remote_poll_mode mode) noexcept - -> __intrusive_queue<&task_base::next_> + auto pop_all_reversed(std::size_t tid, _remote_poll_mode mode) noexcept + -> __intrusive_queue<&_task_base::next_> { - remote_queue* head = head_.load(__std::memory_order_acquire); - __intrusive_queue<&task_base::next_> tasks{}; + _remote_queue* head = head_.load(__std::memory_order_acquire); + __intrusive_queue<&_task_base::next_> tasks{}; while (head != nullptr) { auto& queue = head->queues_[tid]; - if (mode == remote_poll_mode::before_sleep || !queue.empty()) + if (mode == _remote_poll_mode::_before_sleep || !queue.empty()) { tasks.append(queue.pop_all_reversed()); } @@ -195,20 +138,19 @@ namespace experimental::execution return tasks; } - auto get() -> remote_queue* + auto get() -> _remote_queue* { - thread_local std::thread::id this_id = std::this_thread::get_id(); - remote_queue* head = head_.load(__std::memory_order_acquire); - remote_queue* queue = head; + _remote_queue* head = head_.load(__std::memory_order_acquire); + _remote_queue* queue = head; while (queue != tail_) { - if (queue->id_ == this_id) + if (queue->id_ == std::this_thread::get_id()) { return queue; } queue = queue->next_; } - auto* new_head = new remote_queue{head, nthreads_}; + auto* new_head = new _remote_queue{head, nthreads_}; while (!head_.compare_exchange_weak(head, new_head, __std::memory_order_acq_rel)) { new_head->next_ = head; @@ -217,62 +159,23 @@ namespace experimental::execution } }; - class _static_thread_pool + class _static_thread_pool : public thread_pool_base { + friend struct static_thread_pool; + template struct _opstate; - template - struct _bulk_sender; - - template - struct _bulk_shared_state; - - template - struct _bulk_receiver; - - template - struct _bulk_opstate; - - struct _transform_bulk - { - template Tag, - class Data, - class CvSender> - auto operator()(Tag, Data&& data, CvSender&& sndr) const - { - auto [pol, shape, fun] = static_cast(data); - using policy_t = std::remove_cvref_t; - constexpr bool parallelize = - __one_of; - - if constexpr (__same_as) - { - // Turn a bulk_unchunked into a bulk_chunked opstate - using fun_t = STDEXEC::__bulk::__as_bulk_chunked_fn; - using sender_t = _bulk_sender>; - return sender_t{pool_, static_cast(sndr), shape, fun_t(std::move(fun))}; - } - else - { - using fun_t = decltype(fun); - using sender_t = _bulk_sender>; - return sender_t{pool_, static_cast(sndr), shape, std::move(fun)}; - } - } - - _static_thread_pool& pool_; - }; - struct _transform_iterate { template - auto operator()(exec::iterate_t, Range&& range) -> schedule_all_::sequence<__decay_t> + auto operator()(exec::iterate_t, Range&& range) + -> _schedule_all::_sequence<__decay_t, static_thread_pool> { return {static_cast(range), pool_}; } - _static_thread_pool& pool_; + static_thread_pool& pool_; }; static auto _hardware_concurrency() noexcept -> unsigned int @@ -282,38 +185,12 @@ namespace experimental::execution } public: - struct domain : STDEXEC::default_domain + // The bulk algorithms are handled by the generic pool domain; only `iterate` + // is specific to this pool. + struct domain : _pool::_domain { - // transform the generic bulk_chunked sender into a parallel thread-pool - // bulk sender - template - requires __one_of, bulk_chunked_t, bulk_unchunked_t> - constexpr auto - transform_sender(STDEXEC::set_value_t, Sender&& sndr, Env const & env) const noexcept - { - auto& [tag, data, child] = sndr; - auto& [pol, shape, fun] = data; - - if constexpr (__completes_on) - { - auto sched = STDEXEC::get_completion_scheduler(get_env(child), - env); - static_assert(std::is_same_v); - return __apply(_transform_bulk{*sched.pool_}, static_cast(sndr)); - } - else - { - return STDEXEC::__not_a_sender< - STDEXEC::_WHAT_( - CANNOT_DISPATCH_THE_BULK_ALGORITHM_TO_THE_STATIC_THREAD_POOL_SCHEDULER), - STDEXEC::_WHY_(BECAUSE_THERE_IS_NO_STATIC_THREAD_POOL_SCHEDULER_IN_THE_ENVIRONMENT), - STDEXEC::_WHERE_(STDEXEC::_IN_ALGORITHM_, tag_of_t), - STDEXEC::_TO_FIX_THIS_ERROR_( - ADD_A_CONTINUES_ON_TRANSITION_TO_THE_STATIC_THREAD_POOL_SCHEDULER_BEFORE_THE_BULK_ALGORITHM), - STDEXEC::_WITH_SENDER_, - STDEXEC::_WITH_ENVIRONMENT_(Env)>(); - } - } + // Bring in the base class's transform_sender for bulk_[un]chunked + using _pool::_domain::transform_sender; template Sender, class Env> constexpr auto @@ -321,32 +198,31 @@ namespace experimental::execution { if constexpr (__completes_on) { - auto sched = STDEXEC::get_completion_scheduler(get_env(sndr), - env); - return __apply(_transform_iterate{*sched.pool_}, static_cast(sndr)); + auto sched = STDEXEC::get_completion_scheduler(get_env(sndr), env); + auto& pool = thread_pool_access::get_pool(sched); + return __apply(_transform_iterate{pool}, static_cast(sndr)); } else { - return STDEXEC::__not_a_sender< - STDEXEC::_WHAT_( - CANNOT_DISPATCH_THE_ITERATE_ALGORITHM_TO_THE_STATIC_THREAD_POOL_SCHEDULER), - STDEXEC::_WHY_(BECAUSE_THERE_IS_NO_STATIC_THREAD_POOL_SCHEDULER_IN_THE_ENVIRONMENT), - STDEXEC::_WHERE_(STDEXEC::_IN_ALGORITHM_, exec::iterate_t), - STDEXEC::_TO_FIX_THIS_ERROR_( + return __not_a_sender< + _WHAT_(CANNOT_DISPATCH_THE_ITERATE_ALGORITHM_TO_THE_STATIC_THREAD_POOL_SCHEDULER), + _WHY_(BECAUSE_THERE_IS_NO_STATIC_THREAD_POOL_SCHEDULER_IN_THE_ENVIRONMENT), + _WHERE_(_IN_ALGORITHM_, exec::iterate_t), + _TO_FIX_THIS_ERROR_( ADD_A_CONTINUES_ON_TRANSITION_TO_THE_STATIC_THREAD_POOL_SCHEDULER_BEFORE_THE_ITERATE_ALGORITHM), - STDEXEC::_WITH_SENDER_, - STDEXEC::_WITH_ENVIRONMENT_(Env)>(); + _WITH_SENDER_, + _WITH_ENVIRONMENT_(Env)>(); } } }; - public: _static_thread_pool(); explicit _static_thread_pool(std::uint32_t threadCount, bwos_params params = {}, numa_policy numa = get_numa_policy()); ~_static_thread_pool(); + public: struct scheduler { using scheduler_concept = scheduler_tag; @@ -355,6 +231,8 @@ namespace experimental::execution template friend struct _opstate; + friend thread_pool_access; // needs access to pool_ + class _sender { struct attrs @@ -372,7 +250,7 @@ namespace experimental::execution } _static_thread_pool& pool_; - remote_queue* queue_; + _remote_queue* queue_; }; public: @@ -407,7 +285,7 @@ namespace experimental::execution friend struct scheduler; explicit _sender(_static_thread_pool& pool, - remote_queue* queue, + _remote_queue* queue, std::size_t threadIndex, nodemask const & constraints) noexcept : pool_(pool) @@ -417,7 +295,7 @@ namespace experimental::execution {} _static_thread_pool& pool_; - remote_queue* queue_; + _remote_queue* queue_; std::size_t threadIndex_{(std::numeric_limits::max)()}; nodemask constraints_{}; }; @@ -432,7 +310,7 @@ namespace experimental::execution {} explicit scheduler(_static_thread_pool& pool, - remote_queue& queue, + _remote_queue& queue, nodemask const * mask = &nodemask::any()) noexcept : pool_(&pool) , queue_{&queue} @@ -440,7 +318,7 @@ namespace experimental::execution {} explicit scheduler(_static_thread_pool& pool, - remote_queue& queue, + _remote_queue& queue, std::size_t threadIndex) noexcept : pool_(&pool) , queue_{&queue} @@ -448,7 +326,7 @@ namespace experimental::execution {} _static_thread_pool* pool_; - remote_queue* queue_; + _remote_queue* queue_; nodemask const * nodemask_ = &nodemask::any(); std::size_t thread_idx_{(std::numeric_limits::max)()}; @@ -481,6 +359,7 @@ namespace experimental::execution } }; + // NOLINTNEXTLINE(bugprone-derived-method-shadowing-base-method) auto get_scheduler() noexcept -> scheduler { return scheduler{*this}; @@ -498,10 +377,10 @@ namespace experimental::execution return scheduler{*this, *get_remote_queue(), constraints}; } - auto get_remote_queue() noexcept -> remote_queue* + auto get_remote_queue() noexcept -> _remote_queue* { - remote_queue* queue = remotes_.get(); - std::size_t index = 0; + _remote_queue* queue = remotes_.get(); + std::size_t index = 0; for (std::thread& t: threads_) { if (t.get_id() == queue->id_) @@ -517,6 +396,7 @@ namespace experimental::execution void request_stop() noexcept; + // NOLINTNEXTLINE(bugprone-derived-method-shadowing-base-method) [[nodiscard]] auto available_parallelism() const noexcept -> std::uint32_t { @@ -529,38 +409,37 @@ namespace experimental::execution return params_; } - void enqueue(task_base* task, nodemask const & contraints = nodemask::any()) noexcept; - void enqueue(remote_queue& queue, - task_base* task, + void enqueue(_task_base* task, nodemask const & contraints = nodemask::any()) noexcept; + void enqueue(_remote_queue& queue, + _task_base* task, nodemask const & contraints = nodemask::any()) noexcept; - void enqueue(remote_queue& queue, task_base* task, std::size_t thread_index) noexcept; - - //! Enqueue a contiguous span of tasks across task queues. - //! Note: We use the concrete `Task` because we enqueue - //! tasks `task + 0`, `task + 1`, etc. so std::span - //! wouldn't be correct. - //! This is O(n_threads) on the calling thread. - template Task> + void enqueue(_remote_queue& queue, _task_base* task, std::size_t thread_index) noexcept; + + //! Enqueue a contiguous span of tasks across task queues. Note: We use + //! the concrete `Task` because we enqueue tasks `task + 0`, `task + 1`, + //! etc. so std::span<_task_base> wouldn't be correct. This is + //! O(n_threads) on the calling thread. + template Task> void bulk_enqueue(std::span tasks) noexcept; - void bulk_enqueue(remote_queue& queue, - __intrusive_queue<&task_base::next_> tasks, - std::size_t tasks_size, + void bulk_enqueue(_remote_queue& queue, + __intrusive_queue<&_task_base::next_> tasks, + std::size_t tasks_size, nodemask const & constraints = nodemask::any()) noexcept; private: - class workstealing_victim + class _workstealing_victim { public: - explicit workstealing_victim( - bwos::lifo_queue>* queue, - std::uint32_t index, - int numa_node) noexcept + explicit _workstealing_victim( + bwos::lifo_queue<_task_base*, numa_allocator<_task_base*>>* queue, + std::uint32_t index, + int numa_node) noexcept : queue_(queue) , index_(index) , numa_node_(numa_node) {} - auto try_steal() noexcept -> task_base* + auto try_steal() noexcept -> _task_base* { return queue_->steal_front(); } @@ -578,14 +457,14 @@ namespace experimental::execution } private: - bwos::lifo_queue>* queue_; - std::uint32_t index_; - int numa_node_; + bwos::lifo_queue<_task_base*, numa_allocator<_task_base*>>* queue_; + std::uint32_t index_; + int numa_node_; }; - struct thread_state_base + struct _thread_state_base { - explicit thread_state_base(std::uint32_t index, numa_policy const & numa) noexcept + explicit _thread_state_base(std::uint32_t index, numa_policy const & numa) noexcept : index_(index) , numa_node_(numa.thread_index_to_node(index)) {} @@ -594,23 +473,23 @@ namespace experimental::execution int numa_node_; }; - class thread_state : private thread_state_base + class _thread_state : private _thread_state_base { public: struct pop_result { - task_base* task; + _task_base* task; std::uint32_t queue_index; }; - explicit thread_state(_static_thread_pool* pool, - std::uint32_t index, - bwos_params params, - numa_policy const & numa) noexcept - : thread_state_base(index, numa) + explicit _thread_state(_static_thread_pool* pool, + std::uint32_t index, + bwos_params params, + numa_policy const & numa) noexcept + : _thread_state_base(index, numa) , local_queue_(params.numBlocks, params.blockSize, - numa_allocator(this->numa_node_)) + numa_allocator<_task_base*>(this->numa_node_)) , state_(state::running) , pool_(pool) { @@ -619,15 +498,15 @@ namespace experimental::execution } auto pop() -> pop_result; - void push_local(task_base* task); - void push_local(__intrusive_queue<&task_base::next_>&& tasks); + void push_local(_task_base* task); + void push_local(__intrusive_queue<&_task_base::next_>&& tasks); auto notify() -> bool; void request_stop(); - void victims(std::vector const & victims) + void victims(std::vector<_workstealing_victim> const & victims) { - for (workstealing_victim v: victims) + for (_workstealing_victim v: victims) { if (v.index() == index_) { @@ -654,9 +533,9 @@ namespace experimental::execution return numa_node_; } - auto as_victim() noexcept -> workstealing_victim + auto as_victim() noexcept -> _workstealing_victim { - return workstealing_victim{&local_queue_, index_, numa_node_}; + return _workstealing_victim{&local_queue_, index_, numa_node_}; } private: @@ -669,8 +548,8 @@ namespace experimental::execution }; auto try_pop() -> pop_result; - auto try_remote(remote_poll_mode mode) -> pop_result; - auto try_steal(std::span victims) -> pop_result; + auto try_remote(_remote_poll_mode mode) -> pop_result; + auto try_steal(std::span<_workstealing_victim> victims) -> pop_result; auto try_steal_near() -> pop_result; auto try_steal_any() -> pop_result; @@ -680,50 +559,50 @@ namespace experimental::execution void set_sleeping(); void clear_sleeping(); - bwos::lifo_queue> local_queue_; - __intrusive_queue<&task_base::next_> pending_queue_{}; - std::mutex mut_{}; - std::condition_variable cv_{}; - bool stop_requested_{false}; - std::vector near_victims_{}; - std::vector all_victims_{}; - __std::atomic state_; - _static_thread_pool* pool_; - xorshift rng_{}; + bwos::lifo_queue<_task_base*, numa_allocator<_task_base*>> local_queue_; + __intrusive_queue<&_task_base::next_> pending_queue_{}; + std::mutex mut_{}; + std::condition_variable cv_{}; + bool stop_requested_{false}; + std::vector<_workstealing_victim> near_victims_{}; + std::vector<_workstealing_victim> all_victims_{}; + __std::atomic state_; + _static_thread_pool* pool_; + xorshift rng_{}; }; void run(std::uint32_t index) noexcept; void join() noexcept; alignas(64) __std::atomic num_active_{}; - alignas(64) remote_queue_list remotes_; - std::uint32_t thread_count_; - std::uint32_t max_steals_{thread_count_ + 1}; - bwos_params params_; - std::vector threads_; - std::vector<__optional> thread_states_; - numa_policy numa_; - - struct thread_index_by_numa_node + alignas(64) _remote_queue_list remotes_; + std::uint32_t thread_count_; + std::uint32_t max_steals_{thread_count_ + 1}; + bwos_params params_; + std::vector threads_; + std::vector<__optional<_thread_state>> thread_states_; + numa_policy numa_; + + struct _thread_index_by_numa_node { int numa_node; std::size_t thread_index; - friend auto operator==(thread_index_by_numa_node const & lhs, - thread_index_by_numa_node const & rhs) noexcept -> bool + friend auto operator==(_thread_index_by_numa_node const & lhs, + _thread_index_by_numa_node const & rhs) noexcept -> bool { return lhs.numa_node == rhs.numa_node; } friend auto - operator<=>(thread_index_by_numa_node const & lhs, - thread_index_by_numa_node const & rhs) noexcept -> std::strong_ordering + operator<=>(_thread_index_by_numa_node const & lhs, + _thread_index_by_numa_node const & rhs) noexcept -> std::strong_ordering { return lhs.numa_node <=> rhs.numa_node; } }; - std::vector thread_index_by_numa_node_; + std::vector<_thread_index_by_numa_node> thread_index_by_numa_node_; [[nodiscard]] auto num_threads(int numa) const noexcept -> std::size_t; @@ -754,12 +633,12 @@ namespace experimental::execution { thread_states_[index].emplace(this, index, params, numa_); thread_index_by_numa_node_.push_back( - thread_index_by_numa_node{.numa_node = thread_states_[index]->numa_node(), - .thread_index = index}); + _thread_index_by_numa_node{.numa_node = thread_states_[index]->numa_node(), + .thread_index = index}); } std::ranges::sort(thread_index_by_numa_node_); - std::vector victims{}; + std::vector<_workstealing_victim> victims{}; for (auto& state: thread_states_) { victims.emplace_back(state->as_victim()); @@ -826,15 +705,16 @@ namespace experimental::execution threads_.clear(); } - inline void _static_thread_pool::enqueue(task_base* task, nodemask const & constraints) noexcept + inline void + _static_thread_pool::enqueue(_task_base* task, nodemask const & constraints) noexcept { this->enqueue(*get_remote_queue(), task, constraints); } inline auto _static_thread_pool::num_threads(int numa) const noexcept -> std::size_t { - thread_index_by_numa_node key{.numa_node = numa, .thread_index = 0}; - auto it = std::ranges::lower_bound(thread_index_by_numa_node_, key); + _thread_index_by_numa_node key{.numa_node = numa, .thread_index = 0}; + auto it = std::ranges::lower_bound(thread_index_by_numa_node_, key); if (it == thread_index_by_numa_node_.end()) { return 0; @@ -864,8 +744,8 @@ namespace experimental::execution _static_thread_pool::get_thread_index(int node_index, std::size_t thread_index) const noexcept -> std::size_t { - thread_index_by_numa_node key{.numa_node = node_index, .thread_index = 0}; - auto it = std::ranges::lower_bound(thread_index_by_numa_node_, key); + _thread_index_by_numa_node key{.numa_node = node_index, .thread_index = 0}; + auto it = std::ranges::lower_bound(thread_index_by_numa_node_, key); STDEXEC_ASSERT(it != thread_index_by_numa_node_.end()); std::advance(it, thread_index); return it->thread_index; @@ -898,13 +778,13 @@ namespace experimental::execution return target_index; } - inline void _static_thread_pool::enqueue(remote_queue& queue, - task_base* task, + inline void _static_thread_pool::enqueue(_remote_queue& queue, + _task_base* task, nodemask const & constraints) noexcept { - static thread_local std::thread::id this_id = std::this_thread::get_id(); - remote_queue* correct_queue = this_id == queue.id_ ? &queue : get_remote_queue(); - std::size_t idx = correct_queue->index_; + _remote_queue* correct_queue = std::this_thread::get_id() == queue.id_ ? &queue + : get_remote_queue(); + std::size_t idx = correct_queue->index_; if (idx < thread_states_.size()) { auto this_node = static_cast(thread_states_[idx]->numa_node()); @@ -922,9 +802,9 @@ namespace experimental::execution } } - inline void _static_thread_pool::enqueue(remote_queue& queue, - task_base* task, - std::size_t thread_index) noexcept + inline void _static_thread_pool::enqueue(_remote_queue& queue, + _task_base* task, + std::size_t thread_index) noexcept { thread_index %= thread_count_; if (queue.queues_[thread_index].push_front(task)) @@ -933,7 +813,7 @@ namespace experimental::execution } } - template Task> + template Task> void _static_thread_pool::bulk_enqueue(std::span tasks) noexcept { auto& queue = *this->get_remote_queue(); @@ -950,14 +830,15 @@ namespace experimental::execution // See `if (is_last_thread)` above. } - inline void _static_thread_pool::bulk_enqueue(remote_queue& queue, - __intrusive_queue<&task_base::next_> tasks, - std::size_t tasks_size, + inline void _static_thread_pool::bulk_enqueue(_remote_queue& queue, + __intrusive_queue<&_task_base::next_> tasks, + std::size_t tasks_size, nodemask const & constraints) noexcept { - static thread_local std::thread::id const this_id = std::this_thread::get_id(); - remote_queue* const correct_queue = this_id == queue.id_ ? &queue : get_remote_queue(); - std::size_t const idx = correct_queue->index_; + _remote_queue* const correct_queue = std::this_thread::get_id() == queue.id_ + ? &queue + : get_remote_queue(); + std::size_t const idx = correct_queue->index_; if (idx < thread_states_.size()) { auto this_node = static_cast(thread_states_[idx]->numa_node()); @@ -971,12 +852,12 @@ namespace experimental::execution std::uint32_t const total_threads = available_parallelism(); for (std::uint32_t i = 0; i < total_threads; ++i) { - auto [begin, end] = even_share(tasks_size, i, total_threads); + auto [begin, end] = _pool::_even_share(tasks_size, i, total_threads); if (begin == end) { continue; } - __intrusive_queue<&task_base::next_> tmp{}; + __intrusive_queue<&_task_base::next_> tmp{}; for (auto j = begin; j < end; ++j) { tmp.push_back(tasks.pop_front()); @@ -987,20 +868,21 @@ namespace experimental::execution } inline void - move_pending_to_local(__intrusive_queue<&task_base::next_>& pending_queue, - bwos::lifo_queue>& local_queue) + move_pending_to_local(__intrusive_queue<&_task_base::next_>& pending_queue, + bwos::lifo_queue<_task_base*, numa_allocator<_task_base*>>& local_queue) { auto const last = local_queue.push_back(pending_queue.begin(), pending_queue.end()); - __intrusive_queue<&task_base::next_> tmp{}; + __intrusive_queue<&_task_base::next_> tmp{}; tmp.splice(tmp.begin(), pending_queue, pending_queue.begin(), last); tmp.clear(); } - inline auto _static_thread_pool::thread_state::try_remote(remote_poll_mode mode) - -> _static_thread_pool::thread_state::pop_result + inline auto _static_thread_pool::_thread_state::try_remote(_remote_poll_mode mode) + -> _static_thread_pool::_thread_state::pop_result { - pop_result result{.task = nullptr, .queue_index = index_}; - __intrusive_queue<&task_base::next_> remotes = pool_->remotes_.pop_all_reversed(index_, mode); + pop_result result{.task = nullptr, .queue_index = index_}; + __intrusive_queue<&_task_base::next_> remotes = pool_->remotes_.pop_all_reversed(index_, + mode); pending_queue_.append(std::move(remotes)); if (!pending_queue_.empty()) { @@ -1012,7 +894,7 @@ namespace experimental::execution } inline auto - _static_thread_pool::thread_state::try_pop() -> _static_thread_pool::thread_state::pop_result + _static_thread_pool::_thread_state::try_pop() -> _static_thread_pool::_thread_state::pop_result { pop_result result{.task = nullptr, .queue_index = index_}; result.task = local_queue_.pop_back(); @@ -1020,11 +902,12 @@ namespace experimental::execution { return result; } - return try_remote(remote_poll_mode::speculative); + return try_remote(_remote_poll_mode::_speculative); } - inline auto _static_thread_pool::thread_state::try_steal(std::span victims) - -> _static_thread_pool::thread_state::pop_result + inline auto + _static_thread_pool::_thread_state::try_steal(std::span<_workstealing_victim> victims) + -> _static_thread_pool::_thread_state::pop_result { if (victims.empty()) { @@ -1038,19 +921,19 @@ namespace experimental::execution return {.task = v.try_steal(), .queue_index = v.index()}; } - inline auto _static_thread_pool::thread_state::try_steal_near() - -> _static_thread_pool::thread_state::pop_result + inline auto _static_thread_pool::_thread_state::try_steal_near() + -> _static_thread_pool::_thread_state::pop_result { return try_steal(near_victims_); } - inline auto _static_thread_pool::thread_state::try_steal_any() - -> _static_thread_pool::thread_state::pop_result + inline auto _static_thread_pool::_thread_state::try_steal_any() + -> _static_thread_pool::_thread_state::pop_result { return try_steal(all_victims_); } - inline void _static_thread_pool::thread_state::push_local(task_base* task) + inline void _static_thread_pool::_thread_state::push_local(_task_base* task) { if (!local_queue_.push_back(task)) { @@ -1059,18 +942,18 @@ namespace experimental::execution } inline void - _static_thread_pool::thread_state::push_local(__intrusive_queue<&task_base::next_>&& tasks) + _static_thread_pool::_thread_state::push_local(__intrusive_queue<&_task_base::next_>&& tasks) { pending_queue_.prepend(std::move(tasks)); } - inline void _static_thread_pool::thread_state::set_sleeping() + inline void _static_thread_pool::_thread_state::set_sleeping() { pool_->num_active_.fetch_sub(1u << 16u, __std::memory_order_relaxed); } // wakeup a worker thread and maintain the invariant that we always one active thief as long as a potential victim is awake - inline void _static_thread_pool::thread_state::clear_sleeping() + inline void _static_thread_pool::_thread_state::clear_sleeping() { std::uint32_t const num_active = pool_->num_active_.fetch_add(1u << 16u, __std::memory_order_relaxed); @@ -1080,14 +963,14 @@ namespace experimental::execution } } - inline void _static_thread_pool::thread_state::set_stealing() + inline void _static_thread_pool::_thread_state::set_stealing() { std::uint32_t const diff = 1u - (1u << 16u); pool_->num_active_.fetch_add(diff, __std::memory_order_relaxed); } // put a thief to sleep but maintain the invariant that we always have one active thief as long as a potential victim is awake - inline void _static_thread_pool::thread_state::clear_stealing() + inline void _static_thread_pool::_thread_state::clear_stealing() { constexpr std::uint32_t diff = 1 - (1u << 16u); std::uint32_t const num_active = pool_->num_active_.fetch_sub(diff, @@ -1100,7 +983,7 @@ namespace experimental::execution } } - inline void _static_thread_pool::thread_state::notify_one_sleeping() + inline void _static_thread_pool::_thread_state::notify_one_sleeping() { std::uniform_int_distribution dist(0, pool_->thread_count_ - 1); std::uint32_t start_index = dist(rng_); @@ -1118,8 +1001,8 @@ namespace experimental::execution } } - inline auto _static_thread_pool::thread_state::pop() // - -> _static_thread_pool::thread_state::pop_result + inline auto _static_thread_pool::_thread_state::pop() // + -> _static_thread_pool::_thread_state::pop_result { pop_result result = try_pop(); while (!result.task) @@ -1161,7 +1044,7 @@ namespace experimental::execution // The relaxed empty probe is safe during normal polling, but the // running-to-sleeping boundary must perform the CAS dequeue so work // published before the transition cannot be missed. - result = try_remote(remote_poll_mode::before_sleep); + result = try_remote(_remote_poll_mode::_before_sleep); if (result.task) { state expected_sleeping = state::sleeping; @@ -1186,7 +1069,7 @@ namespace experimental::execution return result; } - inline auto _static_thread_pool::thread_state::notify() -> bool + inline auto _static_thread_pool::_thread_state::notify() -> bool { if (state_.exchange(state::notified, __std::memory_order_release) == state::sleeping) { @@ -1199,7 +1082,7 @@ namespace experimental::execution return false; } - inline void _static_thread_pool::thread_state::request_stop() + inline void _static_thread_pool::_thread_state::request_stop() { { std::lock_guard lock{mut_}; @@ -1209,7 +1092,7 @@ namespace experimental::execution } template - struct _static_thread_pool::_opstate : task_base + struct _static_thread_pool::_opstate : _task_base { void start() & noexcept { @@ -1220,11 +1103,11 @@ namespace experimental::execution friend class scheduler::_sender; explicit _opstate(_static_thread_pool& pool, - remote_queue* queue, + _remote_queue* queue, Receiver rcvr, std::size_t tid, nodemask const & constraints) noexcept - : task_base{&_opstate::do_execute_} + : _task_base{&_opstate::do_execute_} , pool_(pool) , queue_(queue) , rcvr_(static_cast(rcvr)) @@ -1232,7 +1115,7 @@ namespace experimental::execution , constraints_{constraints} {} - void enqueue_(task_base* op) const + void enqueue_(_task_base* op) const { if (thread_index_ < pool_.available_parallelism()) { @@ -1244,7 +1127,7 @@ namespace experimental::execution } } - static void do_execute_(task_base* t, [[maybe_unused]] std::uint32_t const tid) noexcept + static void do_execute_(_task_base* t, [[maybe_unused]] std::uint32_t const tid) noexcept { auto& op = *static_cast<_opstate*>(t); auto stoken = get_stop_token(get_env(op.rcvr_)); @@ -1264,624 +1147,24 @@ namespace experimental::execution } _static_thread_pool& pool_; - remote_queue* queue_; + _remote_queue* queue_; Receiver rcvr_; std::size_t thread_index_{}; nodemask constraints_{}; }; - //////////////////////////////////////////////////////////////////////////// - // What follows is the implementation for parallel bulk execution on _static_thread_pool. - template - struct _static_thread_pool::_bulk_sender - { - using sender_concept = sender_tag; - - template - using _bulk_opstate_t = - _static_thread_pool::_bulk_opstate; - - explicit _bulk_sender(_static_thread_pool& pool, Sender sndr, Shape shape, Fun fun) - noexcept(__nothrow_move_constructible) - : pool_(pool) - , sndr_(static_cast(sndr)) - , shape_(shape) - , fun_(static_cast(fun)) - {} - - template <__decays_to<_bulk_sender> Self, receiver Receiver> - STDEXEC_EXPLICIT_THIS_BEGIN(auto connect)(this Self&& self, Receiver rcvr) - noexcept(__nothrow_constructible_from<_bulk_opstate_t, - _static_thread_pool&, - Shape, - Fun, - Sender, - Receiver>) -> _bulk_opstate_t - { - return _bulk_opstate_t{self.pool_, - self.shape_, - self.fun_, - static_cast(self).sndr_, - static_cast(rcvr)}; - } - STDEXEC_EXPLICIT_THIS_END(connect) - - template <__decays_to<_bulk_sender> Self, class... Env> - static consteval auto get_completion_signatures() - { - return exec::transform_completion_signatures( - STDEXEC::get_completion_signatures<__copy_cvref_t, __fwd_env_t...>(), - []() - { - if constexpr (!__nothrow_decay_copyable) - { - return exec::throw_compile_time_error<_WHAT_( - _PREDECESSOR_RESULTS_ARE_NOT_DECAY_COPYABLE_), - _WHERE_(_IN_ALGORITHM_, bulk_t), - _WITH_SENDER_<__copy_cvref_t>, - _WITH_ENVIRONMENT_(Env...)>(); - } - else if constexpr (!__callable&...>) - { - return exec::throw_compile_time_error< - _WHAT_(_FUNCTION_IS_NOT_CALLABLE_WITH_THE_GIVEN_ARGUMENTS_), - _WHERE_(_IN_ALGORITHM_, bulk_t), - _WITH_FUNCTION_(Fun&), - _WITH_ARGUMENTS_(Shape, Shape, __decay_t & ...)>(); - } - else if constexpr (__nothrow_callable&...>) - { - return completion_signatures...)>(); - } - else - { - return completion_signatures...), - set_error_t(std::exception_ptr)>(); - } - }); - } - - [[nodiscard]] - auto get_env() const noexcept -> env_of_t - { - return STDEXEC::get_env(sndr_); - } - - private: - _static_thread_pool& pool_; - Sender sndr_; - Shape shape_; - Fun fun_; - }; - - consteval auto _get_apply_fn() noexcept - { - return [](Fun& fun, Tuple& tupl) noexcept( - __nothrow_applicable) -> void - { - __apply(std::move(fun), tupl); - }; - } - - //! The customized opstate state for `STDEXEC::bulk` opstates - template - struct _static_thread_pool::_bulk_shared_state - { - using variant_t = - __value_types_of_t, __qq<__decayed_tuple>, __qq<__variant>>; - - //! Construct from a pool, receiver, shape, and function. - //! Allocates O(min(shape, available_parallelism())) memory. - explicit _bulk_shared_state(_static_thread_pool& pool, Receiver rcvr, Shape shape, Fun fun) - : pool_{pool} - , rcvr_{static_cast(rcvr)} - , shape_{shape} - , fun_{fun} - , thread_with_exception_{num_agents_required()} - , tasks_{num_agents_required(), bulk_task{this}} - {} - - //! The number of agents required is the minimum of `shape_` and the available parallelism. - //! That is, we don't need an agent for each of the shape values. - [[nodiscard]] - auto num_agents_required() const noexcept -> std::uint32_t - { - if constexpr (Parallelize) - { - return Shape{} < shape_ - ? static_cast( - __umin({std::size_t(shape_), std::size_t(pool_.available_parallelism())})) - : 0; - } - else - { - return static_cast(1); - } - } - - private: - friend struct _bulk_receiver; - - //! The actual `bulk_task` holds a pointer to the shared state - //! and its `execute_` function reads from that shared state. - struct bulk_task : task_base - { - explicit bulk_task(_bulk_shared_state* sh_state) noexcept - : task_base{&bulk_task::do_execute_} - , sh_state_(sh_state) - {} - - private: - static void do_execute_(task_base* t, std::uint32_t const tid) noexcept - { - auto& sh_state = *static_cast(t)->sh_state_; - auto total_threads = sh_state.num_agents_required(); - // Each computation does one or more call to the the bulk function. 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 const [begin, end] = _pool_::even_share(sh_state.shape_, tid, total_threads); - auto const applicator = std::bind_front(std::ref(sh_state.fun_), begin, end); - auto const computation = std::bind_front(__apply, applicator); - auto const completion = STDEXEC::__flip(std::bind_front(__apply, STDEXEC::set_value)); - - if constexpr (noexcept(__visit(computation, sh_state.data_))) - { - __visit(computation, sh_state.data_); - if (sh_state.finished_threads_.fetch_add(1) + 1 == total_threads) // last thread? - __visit(completion, std::move(sh_state.data_), std::move(sh_state.rcvr_)); - } - else - { - STDEXEC_TRY - { - __visit(computation, sh_state.data_); - } - STDEXEC_CATCH_ALL - { - std::uint32_t expected = total_threads; - if (sh_state.thread_with_exception_.compare_exchange_strong( - expected, - tid, - __std::memory_order_relaxed, - __std::memory_order_relaxed)) - { - sh_state.exception_ = std::current_exception(); - } - } - - if (sh_state.finished_threads_.fetch_add(1) + 1 == total_threads) // last thread? - { - if (sh_state.exception_) - STDEXEC::set_error(std::move(sh_state.rcvr_), std::move(sh_state.exception_)); - else - __visit(completion, std::move(sh_state.data_), std::move(sh_state.rcvr_)); - } - } - } - - _bulk_shared_state* sh_state_; - }; - - variant_t data_{STDEXEC::__no_init}; - _static_thread_pool& pool_; - Receiver rcvr_; - Shape shape_; - Fun fun_; - - __std::atomic finished_threads_{0}; - __std::atomic thread_with_exception_{0}; - std::exception_ptr exception_; - std::vector tasks_; - }; - - //! A customized receiver to allow parallel execution of `STDEXEC::bulk` opstates: - template - struct _static_thread_pool::_bulk_receiver - { - using receiver_concept = receiver_tag; - - void enqueue() noexcept - { - STDEXEC_ASSERT(shared_state_.tasks_.size() == shared_state_.num_agents_required()); - shared_state_.pool_.bulk_enqueue(std::span{shared_state_.tasks_}); - } - - template - void set_value(As&&... as) noexcept - { - STDEXEC_TRY - { - shared_state_.data_.template emplace<__decayed_tuple>(static_cast(as)...); - - if (Shape{} < shared_state_.shape_) - { - enqueue(); - } - else - { - __visit(STDEXEC::__flip(std::bind_front(__apply, STDEXEC::set_value)), - std::move(shared_state_.data_), - static_cast(shared_state_.rcvr_)); - } - } - STDEXEC_CATCH_ALL - { - if constexpr (__nothrow_decay_copyable) - STDEXEC::__std::unreachable(); - else - STDEXEC::set_error(static_cast(shared_state_.rcvr_), - std::current_exception()); - } - } - - template - void set_error(Error&& error) noexcept - { - STDEXEC::set_error(std::move(shared_state_.rcvr_), static_cast(error)); - } - - void set_stopped() noexcept - { - STDEXEC::set_stopped(std::move(shared_state_.rcvr_)); - } - - auto get_env() const noexcept -> env_of_t - { - return STDEXEC::get_env(shared_state_.rcvr_); - } - - _bulk_shared_state& shared_state_; - }; - - template - struct _static_thread_pool::_bulk_opstate - { - using receiver_t = _bulk_receiver; - using shared_state_t = _bulk_shared_state; - using inner_opstate_t = connect_result_t; - - void start() & noexcept - { - STDEXEC::start(inner_op_); - } - - explicit _bulk_opstate(_static_thread_pool& pool, - Shape shape, - Fun fun, - CvSender&& sndr, - Receiver rcvr) - : shared_state_(pool, static_cast(rcvr), shape, fun) - , inner_op_{STDEXEC::connect(static_cast(sndr), receiver_t{shared_state_})} - {} - - shared_state_t shared_state_; - inner_opstate_t inner_op_; - }; - - namespace schedule_all_ - { - template - auto get_allocator(Rcvr const & rcvr) - { - if constexpr (__callable>) - { - return STDEXEC::get_allocator(STDEXEC::get_env(rcvr)); - } - else - { - return std::allocator{}; - } - } - - template - using allocator_of_t = decltype(get_allocator(__declval())); - - template - struct opstate_base - { - Range range_; - _static_thread_pool& pool_; - std::mutex start_mutex_{}; - bool has_started_{false}; - __intrusive_queue<&task_base::next_> tasks_{}; - std::size_t tasks_size_{}; - __std::atomic countdown_{std::ranges::size(range_)}; - }; - - template - struct item_opstate : task_base - { - explicit item_opstate(ItemReceiver&& item_receiver, - std::ranges::iterator_t it, - opstate_base* parent) - : task_base{&item_opstate::do_execute_} - , item_receiver_(static_cast(item_receiver)) - , it_(it) - , parent_(parent) - {} - - void start() & noexcept - { - std::unique_lock lock{parent_->start_mutex_}; - if (!parent_->has_started_) - { - parent_->tasks_.push_back(static_cast(this)); - parent_->tasks_size_ += 1; - } - else - { - lock.unlock(); - parent_->pool_.enqueue(static_cast(this)); - } - } - - private: - static void do_execute_(task_base* base, std::uint32_t /* tid */) noexcept - { - auto op = static_cast(base); - STDEXEC::set_value(static_cast(op->item_receiver_), *op->it_); - } - - ItemReceiver item_receiver_; - std::ranges::iterator_t it_; - opstate_base* parent_; - }; - - template - struct item_sender - { - using sender_concept = sender_tag; - using completion_signatures = - STDEXEC::completion_signatures)>; - - struct attrs - { - _static_thread_pool* pool_; - - auto query(get_completion_scheduler_t, __ignore = {}) noexcept - -> _static_thread_pool::scheduler - { - return pool_->get_scheduler(); - } - }; - - auto get_env() const noexcept -> attrs - { - return {&op_->pool_}; - } - - template - requires receiver_of - auto connect(ItemReceiver rcvr) const noexcept - { - return item_opstate{static_cast(rcvr), it_, op_}; - } - - opstate_base* op_; - std::ranges::iterator_t it_; - }; - - template - struct opstate_base_with_receiver : opstate_base - { - explicit opstate_base_with_receiver(Range range, _static_thread_pool& pool, Receiver rcvr) - : opstate_base{std::move(range), pool} - , rcvr_(static_cast(rcvr)) - {} - - Receiver rcvr_; - }; - - template - struct next_receiver - { - using receiver_concept = receiver_tag; - - void set_value() noexcept - { - std::size_t countdown = op_->countdown_.fetch_sub(1, __std::memory_order_relaxed); - if (countdown == 1) - { - STDEXEC::set_value(static_cast(op_->rcvr_)); - } - } - - void set_stopped() noexcept - { - std::size_t countdown = op_->countdown_.fetch_sub(1, __std::memory_order_relaxed); - if (countdown == 1) - { - STDEXEC::set_value(static_cast(op_->rcvr_)); - } - } - - auto get_env() const noexcept -> env_of_t - { - return STDEXEC::get_env(op_->rcvr_); - } - - opstate_base_with_receiver* op_; - }; - - template - class opstate : opstate_base_with_receiver - { - public: - explicit opstate(Range range, _static_thread_pool& pool, Receiver rcvr) - : opstate_base_with_receiver{std::move(range), - pool, - static_cast(rcvr)} - , items_(std::ranges::size(this->range_), item_allocator_t(get_allocator(this->rcvr_))) - {} - - ~opstate() - { - for (std::size_t i = 0; i < items_constructed_; ++i) - { - items_[i].__destroy(); - } - } - - void start() & noexcept - { - std::size_t size = items_.size(); - if (size == 0) - { - STDEXEC::set_value(std::move(this->rcvr_)); - return; - } - - std::size_t nthreads = this->pool_.available_parallelism(); - STDEXEC_ASSERT(nthreads > 0); - bwos_params params = this->pool_.params(); - std::size_t local_size = params.blockSize * params.numBlocks; - std::size_t chunk_size = __umax( - {std::size_t{1}, __umin({size / nthreads, local_size * nthreads})}); - auto& remote_queue = *this->pool_.get_remote_queue(); - auto it = std::ranges::begin(this->range_); - std::size_t i0 = 0; - - STDEXEC_TRY - { - for (std::size_t i = 0; i < size; ++i) - { - items_[i].__construct_from(STDEXEC::connect, - set_next(this->rcvr_, item_sender_t{this, it + i}), - next_receiver_t{this}); - ++items_constructed_; - } - } - STDEXEC_CATCH_ALL - { - STDEXEC::set_error(std::move(this->rcvr_), std::current_exception()); - return; - } - - while (i0 + chunk_size < size) - { - for (std::size_t i = i0; i < i0 + chunk_size; ++i) - { - STDEXEC::start(items_[i].__get()); - } - - std::unique_lock lock{this->start_mutex_}; - this->pool_.bulk_enqueue(remote_queue, std::move(this->tasks_), this->tasks_size_); - this->tasks_size_ = 0; - lock.unlock(); - i0 += chunk_size; - } - - for (std::size_t i = i0; i < size; ++i) - { - STDEXEC::start(items_[i].__get()); - } - - std::unique_lock lock{this->start_mutex_}; - this->has_started_ = true; - this->pool_.bulk_enqueue(remote_queue, std::move(this->tasks_), this->tasks_size_); - } - - private: - using allocator_t = allocator_of_t; - using item_sender_t = item_sender; - using next_sender_t = next_sender_of_t; - using next_receiver_t = next_receiver; - using item_opstate_t = connect_result_t; - - using item_allocator_t = std::allocator_traits::template rebind_alloc< - STDEXEC::__manual_lifetime>; - - std::vector<__manual_lifetime, item_allocator_t> items_; - std::size_t items_constructed_{}; - }; - - template - struct sequence - { - using sender_concept = sequence_sender_tag; - using item_types = exec::item_types>; - using completion_signatures = - STDEXEC::completion_signatures; - - explicit sequence(Range range, _static_thread_pool& pool) - : range_(static_cast(range)) - , pool_(&pool) - {} + } // namespace _pool - template Receiver> - auto subscribe(Receiver rcvr) && noexcept -> opstate - { - return opstate{std::move(range_), *pool_, std::move(rcvr)}; - } - - template Receiver> - requires __decay_copyable - auto subscribe(Receiver rcvr) const & noexcept -> opstate - { - return opstate{range_, *pool_, std::move(rcvr)}; - } - - private: - Range range_; - _static_thread_pool* pool_; - }; - } // namespace schedule_all_ - - struct schedule_all_t; - } // namespace _pool_ - - struct static_thread_pool : private _pool_::_static_thread_pool + struct static_thread_pool : _pool::_static_thread_pool { - friend struct _pool_::schedule_all_t; - using task_base = _pool_::task_base; - static_thread_pool() = default; explicit static_thread_pool(std::uint32_t thread_count, bwos_params params = {}, numa_policy numa = get_numa_policy()) - : _pool_::_static_thread_pool(thread_count, params, std::move(numa)) + : _pool::_static_thread_pool(thread_count, params, std::move(numa)) {} - - // struct scheduler; - using _pool_::_static_thread_pool::scheduler; - - // scheduler get_scheduler() noexcept; - using _pool_::_static_thread_pool::get_scheduler; - - // scheduler get_scheduler_on_thread(std::size_t thread_index) noexcept; - using _pool_::_static_thread_pool::get_scheduler_on_thread; - - // scheduler get_constrained_scheduler(const nodemask& constraints) noexcept; - using _pool_::_static_thread_pool::get_constrained_scheduler; - - // void request_stop() noexcept; - using _pool_::_static_thread_pool::request_stop; - - // std::uint32_t available_parallelism() const; - using _pool_::_static_thread_pool::available_parallelism; - - // bwos_params params() const; - using _pool_::_static_thread_pool::params; }; - - namespace _pool_ - { - struct schedule_all_t - { - template - auto operator()(static_thread_pool& pool, Range&& range) const - { - return schedule_all_::sequence<__decay_t>{static_cast(range), pool}; - } - }; - } // namespace _pool_ - - inline constexpr _pool_::schedule_all_t schedule_all{}; - } // namespace experimental::execution STDEXEC_MODULE_EXPORT diff --git a/include/exec/taskflow/taskflow_thread_pool.hpp b/include/exec/taskflow/taskflow_thread_pool.hpp index 3743acaf17..b21e963198 100644 --- a/include/exec/taskflow/taskflow_thread_pool.hpp +++ b/include/exec/taskflow/taskflow_thread_pool.hpp @@ -38,19 +38,17 @@ namespace experimental::execution::taskflow { return static_cast(executor_.num_workers()); } + private: + friend thread_pool_access; + [[nodiscard]] static constexpr auto forward_progress_guarantee() -> STDEXEC::forward_progress_guarantee { return STDEXEC::forward_progress_guarantee::parallel; } - friend struct ::exec::thread_pool_base; - - template - friend struct exec::_pool_::opstate; - - void enqueue(exec::_pool_::task_base* task, std::uint32_t tid = 0) noexcept + void enqueue(exec::_pool::_task_base* task, std::uint32_t tid = 0) noexcept { executor_.silent_async([task, tid] { task->execute_(task, /*tid=*/tid); }); } diff --git a/include/exec/tbb/tbb_thread_pool.hpp b/include/exec/tbb/tbb_thread_pool.hpp index e020509b93..e9f709f99b 100644 --- a/include/exec/tbb/tbb_thread_pool.hpp +++ b/include/exec/tbb/tbb_thread_pool.hpp @@ -42,10 +42,7 @@ namespace experimental::execution::tbb } private: - friend struct ::exec::thread_pool_base; - - template - friend struct exec::_pool_::opstate; + friend thread_pool_access; [[nodiscard]] static constexpr auto forward_progress_guarantee() -> STDEXEC::forward_progress_guarantee @@ -53,7 +50,7 @@ namespace experimental::execution::tbb return STDEXEC::forward_progress_guarantee::parallel; } - void enqueue(exec::_pool_::task_base* task, std::uint32_t tid = 0) noexcept + void enqueue(exec::_pool::_task_base* task, std::uint32_t tid = 0) noexcept { arena_.enqueue([task, tid] { task->execute_(task, /*tid=*/tid); }); } diff --git a/include/exec/thread_pool_base.hpp b/include/exec/thread_pool_base.hpp index b42638417f..db4a1c32ed 100644 --- a/include/exec/thread_pool_base.hpp +++ b/include/exec/thread_pool_base.hpp @@ -28,28 +28,36 @@ import stdexec; # include "../stdexec/__detail/__bulk.hpp" # include "../stdexec/__detail/__connect.hpp" # include "../stdexec/__detail/__env.hpp" +# include "../stdexec/__detail/__intrusive_queue.hpp" +# include "../stdexec/__detail/__manual_lifetime.hpp" # include "../stdexec/__detail/__meta.hpp" # include "../stdexec/__detail/__operation_states.hpp" # include "../stdexec/__detail/__receivers.hpp" # include "../stdexec/__detail/__schedulers.hpp" -# include "../stdexec/__detail/__transform_completion_signatures.hpp" +# include "../stdexec/__detail/__stop_token.hpp" # include "../stdexec/__detail/__type_traits.hpp" # include "../stdexec/__detail/__utility.hpp" +# include "../stdexec/__detail/__variant.hpp" # include # include # include # include # include -# include +# include +# include +# include +# include +# include # include # include -# include +# include #endif +#include "completion_behavior.hpp" #include "completion_signatures.hpp" #include "sender_for.hpp" -#include "static_thread_pool.hpp" +#include "sequence_senders.hpp" namespace experimental::execution { @@ -57,6 +65,99 @@ namespace experimental::execution struct BECAUSE_THERE_IS_NO_POOL_SCHEDULER_IN_THE_ENVIRONMENT; struct ADD_A_CONTINUES_ON_TRANSITION_TO_THE_POOL_SCHEDULER_BEFORE_THE_BULK_ALGORITHM; + struct schedule_all_t; + + template + struct thread_pool_base; + + namespace _pool + { + struct _task_base; + + template + struct _domain; + + template + struct _opstate; + + template + struct _bulk_receiver; + + template + struct _bulk_state; + + template + struct _sender; + + template + struct _scheduler; + + namespace _schedule_all + { + template + struct _item_opstate; + } // namespace _schedule_all + } // namespace _pool + + struct bwos_params + { + std::size_t numBlocks{32}; + std::size_t blockSize{8}; + }; + + struct thread_pool_access + { + //! Pools whose queues are intrusive (a task can be in at most one queue at a time) + //! opt in to per-agent bulk tasks by providing `bulk_enqueue(std::span)`. + //! Task `i` of the span runs with `tid == i`. + template + static constexpr bool has_bulk_enqueue_for_tasks = + requires(Pool& pool, std::span<_pool::_task_base> tasks) { pool.bulk_enqueue(tasks); }; + + //! Gets the pool from a pool-provided scheduler type. The scheduler must befriend + //! `thread_pool_access` and have a `pool_` member convertible to a pointer to `Pool`. + template + static auto get_pool(Scheduler const & sch) noexcept -> Pool& + { + return static_cast(*sch.pool_); + } + + template + static constexpr auto forward_progress_guarantee(Pool& pool) noexcept // + -> STDEXEC::forward_progress_guarantee + { + return pool.forward_progress_guarantee(); + } + + template + static void enqueue(Pool& pool, _pool::_task_base* task) noexcept + { + static_assert(noexcept(pool.enqueue(task))); + pool.enqueue(task); + } + + template + static void enqueue(Pool& pool, _pool::_task_base* task, std::uint32_t tid) noexcept + { + static_assert(noexcept(pool.enqueue(task, tid))); + pool.enqueue(task, tid); + } + + template + static void bulk_enqueue(Pool& pool, _pool::_task_base* task, std::uint32_t n_threads) noexcept + { + static_assert(noexcept(pool.bulk_enqueue(task, n_threads))); + pool.bulk_enqueue(task, n_threads); + } + + template + static void bulk_enqueue(Pool& pool, std::span tasks) noexcept + { + static_assert(noexcept(pool.bulk_enqueue(tasks))); + pool.bulk_enqueue(tasks); + } + }; + //! This is a P2300-style thread pool wrapping base class, which its docs //! describe as "A class that represents an explicit, user-managed task //! scheduler arena." @@ -65,486 +166,841 @@ namespace experimental::execution //! and //! * template auto execute(F &&f) -> decltype(f()) //! - namespace _pool_ + namespace _pool { - template - struct opstate; + using namespace STDEXEC; - using task_base = exec::static_thread_pool::task_base; - } // namespace _pool_ + template + struct _opstate; - template // CRTP - class thread_pool_base - { - template - friend struct _pool_::opstate; + struct _task_base + { + void (*execute_)(_task_base*, std::uint32_t tid) noexcept = nullptr; + _task_base* next_ = nullptr; + }; - friend DerivedPoolType; + //! @brief Splits `n` into `size` chunks distributing `n % size` evenly + //! between ranks. + //! + //! @returns `[begin, end)` range in `n` for a given `rank`. + //! + //! @par Example: + //! + //! @code{cpp} + //! // n_items thread n_threads + //! _even_share( 11, 0, 3); // -> [0, 4) -> 4 items + //! _even_share( 11, 1, 3); // -> [4, 8) -> 4 items + //! _even_share( 11, 2, 3); // -> [8, 11) -> 3 items + //! @endcode + template + constexpr auto _even_share(Shape n, std::size_t rank, std::size_t size) noexcept // + -> std::pair + { + STDEXEC_ASSERT(n >= 0); + using ushape_t = std::make_unsigned_t; + auto const avg_per_thread = static_cast(n) / size; + auto const n_big_share = avg_per_thread + 1; + auto const big_shares = static_cast(n) % size; + auto const is_big_share = rank < big_shares; + auto const begin = is_big_share + ? n_big_share * rank + : n_big_share * big_shares + (rank - big_shares) * avg_per_thread; + auto const end = begin + (is_big_share ? n_big_share : avg_per_thread); + + return std::make_pair(static_cast(begin), static_cast(end)); + } - thread_pool_base() = default; + template + struct _sender + { + using sender_concept = STDEXEC::sender_tag; + + template + static consteval auto get_completion_signatures() noexcept + { + if constexpr (STDEXEC::unstoppable_token>) + { + return STDEXEC::completion_signatures(); + } + else + { + return STDEXEC::completion_signatures(); + } + } - public: - struct scheduler; + template + auto query(STDEXEC::get_completion_scheduler_t) const noexcept // + -> Pool::scheduler + { + return pool_.get_scheduler(); + } - struct domain : STDEXEC::default_domain + template + auto query(STDEXEC::get_completion_domain_t) const noexcept -> Pool::domain + { + return {}; + } + + auto get_env() const noexcept -> _sender const & + { + return *this; + } + + template + auto connect(Receiver rcvr) const -> _opstate + { + return _opstate{pool_, std::move(rcvr)}; + } + + private: + friend struct _scheduler; + + explicit _sender(Pool& pool) noexcept + : pool_(pool) + {} + + Pool& pool_; + }; + + template + struct _scheduler { - template - requires sender_for - || sender_for - static constexpr auto transform_sender(STDEXEC::set_value_t, Sender&& sndr, Env const & env) + using scheduler_concept = scheduler_tag; + + auto operator==(_scheduler const &) const -> bool = default; + + [[nodiscard]] + constexpr auto + query(get_forward_progress_guarantee_t) const noexcept -> forward_progress_guarantee { - auto& [tag, data, child] = sndr; - auto& [pol, shape, fun] = data; - using tag_t = decltype(tag); - using child_t = decltype(child); - using shape_t = decltype(shape); - using fun_t = decltype(fun); + return thread_pool_access::forward_progress_guarantee(*pool_); + } + + template <__one_of Tag> + [[nodiscard]] + constexpr auto query(get_completion_scheduler_t) const noexcept -> _scheduler + { + return *this; + } + + template <__one_of Tag> + [[nodiscard]] + constexpr auto query(get_completion_domain_t) const noexcept -> Pool::domain + { + return {}; + } + + template <__one_of Tag> + [[nodiscard]] + constexpr auto query(__get_completion_behavior_t) const noexcept + { + return exec::completion_behavior::asynchronous; + } + + [[nodiscard]] + auto schedule() const noexcept -> _sender + { + return _sender{*pool_}; + } + + private: + friend thread_pool_access; + friend struct _domain; // needs access to pool_ + friend struct thread_pool_base; // needs access to ctor + + explicit _scheduler(Pool& pool) noexcept + : pool_(std::addressof(pool)) + {} + + Pool* pool_; + }; + + template + struct _bulk_task : _task_base + { + explicit _bulk_task(State* state) noexcept + : _task_base{&_bulk_task::do_execute_} + , state_(state) + {} + + private: + static void do_execute_(_task_base* task, std::uint32_t tid) noexcept + { + static_cast<_bulk_task*>(task)->state_->run(tid); + } - if constexpr (STDEXEC::__completes_on) + State* state_; + }; + + template + struct _bulk_state + { + using task_t = _bulk_task<_bulk_state>; + + //! If true, the pool wants one task node per agent (e.g., because its queues are + //! intrusive); otherwise a single node is enqueued once per agent. + static constexpr bool per_agent_tasks = thread_pool_access::has_bulk_enqueue_for_tasks; + + explicit _bulk_state(Pool& pool, Receiver rcvr, Shape shape, Fun fun, bool parallelize) + noexcept(__nothrow_decay_copyable && !per_agent_tasks) + : pool_(pool) + , rcvr_{std::move(rcvr)} + , shape_{shape} + , fun_{std::move(fun)} + , parallelize_{parallelize} + , thread_with_exception_{num_agents_required()} + , tasks_(make_tasks()) + {} + + [[nodiscard]] + auto num_agents_required() const noexcept -> std::uint32_t + { + using shape_t = std::common_type_t; + auto const parallelism = parallelize_ ? pool_.available_parallelism() : std::uint32_t(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 Shape{} < shape_ ? std::uint32_t(std::min(shape_, parallelism)) + : std::uint32_t(0); + } + + void enqueue() noexcept + { + if constexpr (per_agent_tasks) + { + STDEXEC_ASSERT(tasks_.size() == num_agents_required()); + thread_pool_access::bulk_enqueue(pool_, std::span{tasks_}); + } + else { - auto sch = - STDEXEC::get_completion_scheduler(STDEXEC::get_env(child), env); + thread_pool_access::bulk_enqueue(pool_, &tasks_, num_agents_required()); + } + } + + void run(std::uint32_t const tid) noexcept + { + auto& self = *this; + auto total_threads = self.num_agents_required(); + + // Each computation does one or more call to the the bulk function. 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 const [begin, end] = _pool::_even_share(self.shape_, tid, total_threads); + auto const applicator = std::bind_front(std::ref(self.fun_), begin, end); + auto const computation = std::bind_front(__apply, applicator); + auto const completion = STDEXEC::__flip(std::bind_front(__apply, STDEXEC::set_value)); + + if constexpr (noexcept(__visit(computation, self.data_))) + { + __visit(computation, self.data_); + if (self.finished_threads_.fetch_add(1) + 1 == total_threads) // last thread? + __visit(completion, std::move(self.data_), std::move(self.rcvr_)); + } + else + { + STDEXEC_TRY + { + __visit(computation, self.data_); + } + STDEXEC_CATCH_ALL + { + std::uint32_t expected = total_threads; + if (self.thread_with_exception_.compare_exchange_strong(expected, + tid, + std::memory_order_relaxed, + std::memory_order_relaxed)) + { + self.exception_ = std::current_exception(); + } + } - using policy_type = STDEXEC::__decay_t; - constexpr bool parallelize = STDEXEC::__one_of; + if (self.finished_threads_.fetch_add(1) + 1 == total_threads) // last thread? + { + if (self.exception_) + STDEXEC::set_error(std::move(self.rcvr_), std::move(self.exception_)); + else + __visit(completion, std::move(self.data_), std::move(self.rcvr_)); + } + } + } + + Variant data_{__no_init}; + Pool& pool_; + Receiver rcvr_; + Shape shape_; + Fun fun_; + bool parallelize_; + + std::atomic finished_threads_{0}; + std::atomic thread_with_exception_{0}; + std::exception_ptr exception_; + + // Either one task per agent, or a single task enqueued once per agent. + // NB: the tasks point back at this object, which must not move after construction. + using tasks_t = std::conditional_t, task_t>; + tasks_t tasks_; + + private: + auto make_tasks() -> tasks_t + { + if constexpr (per_agent_tasks) + return tasks_t(num_agents_required(), task_t{this}); + else + return task_t{this}; + } + }; - if constexpr (sender_for) + template + struct _bulk_receiver + { + using receiver_concept = receiver_tag; + + template + void set_value(As&&... as) noexcept + { + STDEXEC_TRY + { + state_->data_.template emplace<__decayed_tuple>(static_cast(as)...); + + if (Shape{} < state_->shape_) { - using sender_t = scheduler::template bulk_sender_t; - return sender_t{*sch.pool_, - STDEXEC::__forward_like(child), - shape, - STDEXEC::__forward_like(fun), - parallelize}; + state_->enqueue(); } else { - using chunked_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, - chunked_fun_t{STDEXEC::__forward_like(fun)}, - parallelize}; + __visit(STDEXEC::__flip(std::bind_front(__apply, STDEXEC::set_value)), + std::move(state_->data_), + std::move(state_->rcvr_)); } } - else + STDEXEC_CATCH_ALL { - return STDEXEC::__not_a_sender< - STDEXEC::_WHAT_(CANNOT_DISPATCH_BULK_ALGORITHM_TO_THE_POOL_SCHEDULER), - STDEXEC::_WHY_(BECAUSE_THERE_IS_NO_POOL_SCHEDULER_IN_THE_ENVIRONMENT), - STDEXEC::_WHERE_(STDEXEC::_IN_ALGORITHM_, tag_t), - STDEXEC::_TO_FIX_THIS_ERROR_( - ADD_A_CONTINUES_ON_TRANSITION_TO_THE_POOL_SCHEDULER_BEFORE_THE_BULK_ALGORITHM), - STDEXEC::_WITH_SENDER_, - STDEXEC::_WITH_ENVIRONMENT_(Env)>(); + if constexpr (__nothrow_decay_copyable) + __std::unreachable(); + else + STDEXEC::set_error(std::move(state_->rcvr_), std::current_exception()); } } + + template + void set_error(Error&& err) noexcept + { + STDEXEC::set_error(std::move(state_->rcvr_), static_cast(err)); + } + + void set_stopped() noexcept + { + STDEXEC::set_stopped(std::move(state_->rcvr_)); + } + + auto get_env() const noexcept -> env_of_t + { + return STDEXEC::get_env(state_->rcvr_); + } + + _bulk_state* state_; + }; + + template + struct _bulk_opstate + { + using variant_t = value_types_of_t, __decayed_tuple, __variant>; + using receiver_t = _bulk_receiver; + + explicit _bulk_opstate(Pool& pool, + Shape shape, + Fun fun, + bool parallelize, + CvSender&& sndr, + Receiver rcvr) + noexcept(__nothrow_decay_copyable && __nothrow_connectable) + : state_(pool, std::move(rcvr), shape, std::move(fun), parallelize) + , inner_op_{STDEXEC::connect(static_cast(sndr), receiver_t{&state_})} + {} + + void start() & noexcept + { + STDEXEC::start(inner_op_); + } + + private: + _bulk_state state_; + connect_result_t inner_op_; }; - struct scheduler + template + constexpr auto _get_bulk_sender_cmplsigs_transform() { - using scheduler_concept = STDEXEC::scheduler_tag; + return []() + { + if constexpr (!__nothrow_decay_copyable) + { + return exec::throw_compile_time_error< // + _WHAT_(_PREDECESSOR_RESULTS_ARE_NOT_DECAY_COPYABLE_), + _WHERE_(_IN_ALGORITHM_, bulk_t), + _WITH_SENDER_, + __fn_t<_WITH_ENVIRONMENT_, Env>...>(); + } + else if constexpr (!__callable&...>) + { + return exec::throw_compile_time_error< + _WHAT_(_FUNCTION_IS_NOT_CALLABLE_WITH_THE_GIVEN_ARGUMENTS_), + _WHERE_(_IN_ALGORITHM_, bulk_t), + _WITH_FUNCTION_(Fun&), + _WITH_ARGUMENTS_(Shape, Shape, __decay_t & ...)>(); + } + else if constexpr (__nothrow_callable&...>) + { + return completion_signatures...)>(); + } + else + { + return completion_signatures...), + set_error_t(std::exception_ptr)>(); + } + }; + } + + template + struct _bulk_sender + { + using sender_concept = sender_tag; + + template + using bulk_opstate_t = + _bulk_opstate, Receiver, Shape, Fun>; + + explicit _bulk_sender(Pool& pool, Sender sndr, Shape shape, Fun fun, bool parallelize) + : pool_(pool) + , sndr_(std::move(sndr)) + , shape_(shape) + , fun_(std::move(fun)) + , parallelize_(parallelize) + {} + + template <__decays_to<_bulk_sender> Self, receiver Receiver> + STDEXEC_EXPLICIT_THIS_BEGIN(auto connect)(this Self&& self, Receiver rcvr) + STDEXEC_AUTO_RETURN(bulk_opstate_t(self.pool_, + Shape(self.shape_), + static_cast(self).fun_, + self.parallelize_, + static_cast(self).sndr_, + std::move(rcvr))); + STDEXEC_EXPLICIT_THIS_END(connect) + + template <__decays_to<_bulk_sender> Self, class... Env> + static consteval auto get_completion_signatures() + { + using sender_t = __copy_cvref_t; + return exec::transform_completion_signatures( + STDEXEC::get_completion_signatures(), + _pool::_get_bulk_sender_cmplsigs_transform()); + } + + struct attrs + { + template , __forwarding_query Tag, class... As> + requires __queryable_with<_Env2, Tag, As...> + auto query(Tag, As&&... as) const noexcept(__nothrow_queryable_with<_Env2, Tag, As...>) + -> decltype(auto) + { + return STDEXEC::get_env(sndr_.sndr_).query(Tag(), static_cast(as)...); + } + + _bulk_sender const & sndr_; + }; + + [[nodiscard]] + auto get_env() const noexcept -> attrs + { + return {*this}; + } private: - template - friend struct _pool_::opstate; + Pool& pool_; + Sender sndr_; + Shape shape_; + Fun fun_; + bool parallelize_; + }; - class sender + template + struct _domain : default_domain + { + protected: + //! Lets pool-specific derived domains reach the pool without befriending + //! `thread_pool_access` (friendship is not inherited). + template + static auto _get_pool(Scheduler const & sch) noexcept -> Pool& { - public: - using sender_concept = STDEXEC::sender_tag; + return thread_pool_access::get_pool(sch); + } + + public: + template + requires sender_for || sender_for + static constexpr auto transform_sender(set_value_t, Sender&& sndr, Env const & env) + { + auto& [tag, data, child] = sndr; + auto& [pol, shape, fun] = data; + using tag_t = decltype(tag); + using child_t = decltype(child); + using shape_t = decltype(shape); + using fun_t = decltype(fun); - template - static consteval auto get_completion_signatures() noexcept + if constexpr (__completes_on) { - if constexpr (STDEXEC::unstoppable_token>) + auto sch = STDEXEC::get_completion_scheduler(STDEXEC::get_env(child), env); + static_assert(std::is_same_v); + auto& pool = thread_pool_access::get_pool(sch); + + using policy_type = __decay_t; + constexpr bool parallelize = + __one_of; + + if constexpr (sender_for) { - return STDEXEC::completion_signatures(); + return _bulk_sender{pool, + STDEXEC::__forward_like(child), + shape, + STDEXEC::__forward_like(fun), + parallelize}; } else { - return STDEXEC::completion_signatures(); + using chunked_fun_t = __bulk::__as_bulk_chunked_fn; + return _bulk_sender{pool, + STDEXEC::__forward_like(child), + shape, + chunked_fun_t{STDEXEC::__forward_like(fun)}, + parallelize}; } } - - template - auto query(STDEXEC::get_completion_scheduler_t) const noexcept // - -> DerivedPoolType::scheduler + else { - return pool_.get_scheduler(); + return __not_a_sender< + _WHAT_(CANNOT_DISPATCH_BULK_ALGORITHM_TO_THE_POOL_SCHEDULER), + _WHY_(BECAUSE_THERE_IS_NO_POOL_SCHEDULER_IN_THE_ENVIRONMENT), + _WHERE_(_IN_ALGORITHM_, tag_t), + _TO_FIX_THIS_ERROR_( + ADD_A_CONTINUES_ON_TRANSITION_TO_THE_POOL_SCHEDULER_BEFORE_THE_BULK_ALGORITHM), + _WITH_SENDER_, + _WITH_ENVIRONMENT_(Env)>(); } + } + }; - template - auto query(STDEXEC::get_completion_domain_t) const noexcept -> domain + namespace _schedule_all + { + template + auto _get_allocator(Receiver const & rcvr) + { + if constexpr (__callable>) { - return {}; + return STDEXEC::get_allocator(STDEXEC::get_env(rcvr)); } - - auto get_env() const noexcept -> sender const & + else { - return *this; + return std::allocator{}; } + } - template - auto connect(Receiver rcvr) const -> _pool_::opstate - { - return _pool_::opstate{pool_, std::move(rcvr)}; - } + template + using _allocator_of_t = decltype(_schedule_all::_get_allocator(__declval())); - private: - using derived_pool_type = DerivedPoolType; - friend struct derived_pool_type::scheduler; + template + struct _state_base + { + Range range_; + Pool& pool_; + std::mutex start_mutex_{}; + bool has_started_{false}; + __intrusive_queue<&_task_base::next_> tasks_{}; + std::size_t tasks_size_{}; + __std::atomic countdown_{std::ranges::size(range_)}; + }; - explicit sender(DerivedPoolType& pool) noexcept - : pool_(pool) + template + struct _state : _state_base + { + explicit _state(Range range, Pool& pool, Receiver rcvr) + : _state_base{std::move(range), pool} + , rcvr_(static_cast(rcvr)) {} - DerivedPoolType& pool_; + Receiver rcvr_; }; - template - struct bulk_shared_state : _pool_::task_base - { - using variant_t = STDEXEC::value_types_of_t, - STDEXEC::__decayed_tuple, - STDEXEC::__variant>; - - explicit bulk_shared_state(DerivedPoolType& pool, - Receiver rcvr, - Shape shape, - Fun fun, - bool parallelize) - noexcept(STDEXEC::__nothrow_decay_copyable) - : _pool_::task_base{&do_execute_} - , pool_(pool) - , rcvr_{std::move(rcvr)} - , shape_{shape} - , fun_{std::move(fun)} - , parallelize_{parallelize} - , thread_with_exception_{num_agents_required()} + template + struct _item_opstate : _task_base + { + explicit _item_opstate(ItemReceiver&& item_receiver, + std::ranges::iterator_t it, + _state_base* parent) + : _task_base{&_item_opstate::do_execute_} + , item_rcvr_(static_cast(item_receiver)) + , it_(it) + , parent_(parent) {} - [[nodiscard]] - auto num_agents_required() const noexcept -> std::uint32_t - { - using common_shape_t = std::common_type_t; - auto const parallelism = parallelize_ ? pool_.available_parallelism() : std::uint32_t(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 Shape{} < shape_ ? std::uint32_t(std::min(shape_, parallelism)) - : std::uint32_t(0); - } - - static void do_execute_(_pool_::task_base* task, std::uint32_t const tid) noexcept + void start() & noexcept { - using namespace STDEXEC; - auto& self = *static_cast(task); - auto total_threads = self.num_agents_required(); - - // Each computation does one or more call to the the bulk function. 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 const [begin, end] = _pool_::even_share(self.shape_, tid, total_threads); - auto const applicator = std::bind_front(std::ref(self.fun_), begin, end); - auto const computation = std::bind_front(__apply, applicator); - auto const completion = STDEXEC::__flip(std::bind_front(__apply, STDEXEC::set_value)); - - if constexpr (noexcept(__visit(computation, self.data_))) + std::unique_lock lock{parent_->start_mutex_}; + if (!parent_->has_started_) { - __visit(computation, self.data_); - if (self.finished_threads_.fetch_add(1) + 1 == total_threads) // last thread? - __visit(completion, std::move(self.data_), std::move(self.rcvr_)); + parent_->tasks_.push_back(static_cast<_task_base*>(this)); + parent_->tasks_size_ += 1; } else { - STDEXEC_TRY - { - __visit(computation, self.data_); - } - STDEXEC_CATCH_ALL - { - std::uint32_t expected = total_threads; - if (self.thread_with_exception_.compare_exchange_strong(expected, - tid, - std::memory_order_relaxed, - std::memory_order_relaxed)) - { - self.exception_ = std::current_exception(); - } - } - - if (self.finished_threads_.fetch_add(1) + 1 == total_threads) // last thread? - { - if (self.exception_) - STDEXEC::set_error(std::move(self.rcvr_), std::move(self.exception_)); - else - __visit(completion, std::move(self.data_), std::move(self.rcvr_)); - } + lock.unlock(); + thread_pool_access::enqueue(parent_->pool_, static_cast<_task_base*>(this)); } } - variant_t data_{STDEXEC::__no_init}; - DerivedPoolType& pool_; - Receiver rcvr_; - Shape shape_; - Fun fun_; - bool parallelize_; + private: + static void do_execute_(_task_base* base, std::uint32_t /* tid */) noexcept + { + auto op = static_cast<_item_opstate*>(base); + STDEXEC::set_value(static_cast(op->item_rcvr_), *op->it_); + } - std::atomic finished_threads_{0}; - std::atomic thread_with_exception_{0}; - std::exception_ptr exception_; + ItemReceiver item_rcvr_; + std::ranges::iterator_t it_; + _state_base* parent_; }; - template - struct bulk_receiver + template + struct _item_sender { - using receiver_concept = STDEXEC::receiver_tag; - using shared_state_t = bulk_shared_state; + using sender_concept = sender_tag; + using completion_signatures = + STDEXEC::completion_signatures)>; - template - void set_value(As&&... as) noexcept + struct attrs { - using namespace STDEXEC; - STDEXEC_TRY - { - shared_state_.data_.template emplace<__decayed_tuple>(static_cast(as)...); + Pool* pool_; - if (Shape{} < shared_state_.shape_) - shared_state_.pool_.bulk_enqueue(&shared_state_, shared_state_.num_agents_required()); - else - __visit(STDEXEC::__flip(std::bind_front(__apply, STDEXEC::set_value)), - std::move(shared_state_.data_), - std::move(shared_state_.rcvr_)); - } - STDEXEC_CATCH_ALL + auto + query(get_completion_scheduler_t, __ignore = {}) noexcept -> Pool::scheduler { - if constexpr (__nothrow_decay_copyable) - __std::unreachable(); - else - STDEXEC::set_error(std::move(shared_state_.rcvr_), std::current_exception()); + return pool_->get_scheduler(); } - } + }; - template - void set_error(Error&& err) noexcept + auto get_env() const noexcept -> attrs { - STDEXEC::set_error(std::move(shared_state_.rcvr_), static_cast(err)); + return {&op_->pool_}; } - void set_stopped() noexcept + template + auto connect(ItemReceiver rcvr) const noexcept { - STDEXEC::set_stopped(std::move(shared_state_.rcvr_)); + using item_opstate_t = _item_opstate; + return item_opstate_t{static_cast(rcvr), it_, op_}; } - auto get_env() const noexcept -> STDEXEC::env_of_t + _state_base* op_; + std::ranges::iterator_t it_; + }; + + template + struct _next_receiver + { + using receiver_concept = receiver_tag; + + void set_value() noexcept { - return STDEXEC::get_env(shared_state_.rcvr_); + std::size_t countdown = op_->countdown_.fetch_sub(1, __std::memory_order_relaxed); + if (countdown == 1) + { + STDEXEC::set_value(static_cast(op_->rcvr_)); + } } - shared_state_t& shared_state_; - }; - - template - struct bulk_opstate - { - using receiver_t = bulk_receiver; - - explicit bulk_opstate(DerivedPoolType& pool, - Shape shape, - Fun fun, - bool parallelize, - CvSender&& sndr, - Receiver rcvr) - noexcept(STDEXEC::__nothrow_decay_copyable - && STDEXEC::__nothrow_connectable) - : shared_state_(pool, std::move(rcvr), shape, std::move(fun), parallelize) - , inner_op_{STDEXEC::connect(static_cast(sndr), receiver_t{shared_state_})} - {} + void set_stopped() noexcept + { + std::size_t countdown = op_->countdown_.fetch_sub(1, __std::memory_order_relaxed); + if (countdown == 1) + { + STDEXEC::set_value(static_cast(op_->rcvr_)); + } + } - void start() & noexcept + auto get_env() const noexcept -> env_of_t { - STDEXEC::start(inner_op_); + return STDEXEC::get_env(op_->rcvr_); } - private: - bulk_shared_state shared_state_; - STDEXEC::connect_result_t inner_op_; + _state* op_; }; - template - struct bulk_sender - { - using sender_concept = STDEXEC::sender_tag; - - template - using bulk_opstate_t = - bulk_opstate, Receiver, Shape, 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 + class _opstate : _state + { + public: + explicit _opstate(Range range, Pool& pool, Receiver rcvr) + : _state{std::move(range), pool, static_cast(rcvr)} + , items_(std::ranges::size(this->range_), + item_allocator_t(_schedule_all::_get_allocator(this->rcvr_))) {} - template Self, STDEXEC::receiver Receiver> - STDEXEC_EXPLICIT_THIS_BEGIN(auto connect)(this Self&& self, Receiver rcvr) - STDEXEC_AUTO_RETURN(bulk_opstate_t(self.pool_, - Shape(self.shape_), - static_cast(self).fun_, - self.parallelize_, - static_cast(self).sndr_, - std::move(rcvr))); - STDEXEC_EXPLICIT_THIS_END(connect) - - template Self, class... Env> - static consteval auto get_completion_signatures() + ~_opstate() { - using namespace STDEXEC; - return exec::transform_completion_signatures( - STDEXEC::get_completion_signatures, Env...>(), - []() - { - if constexpr (!__nothrow_decay_copyable) - { - return exec::throw_compile_time_error< - _WHAT_(_PREDECESSOR_RESULTS_ARE_NOT_DECAY_COPYABLE_), - _WHERE_(_IN_ALGORITHM_, bulk_t), - _WITH_SENDER_<__copy_cvref_t>, - __fn_t<_WITH_ENVIRONMENT_, Env>...>(); - } - else if constexpr (!__callable&...>) - { - return exec::throw_compile_time_error< - _WHAT_(_FUNCTION_IS_NOT_CALLABLE_WITH_THE_GIVEN_ARGUMENTS_), - _WHERE_(_IN_ALGORITHM_, bulk_t), - _WITH_FUNCTION_(Fun&), - _WITH_ARGUMENTS_(Shape, Shape, __decay_t & ...)>(); - } - else if constexpr (__nothrow_callable&...>) - { - return completion_signatures...)>(); - } - else - { - return completion_signatures...), - set_error_t(std::exception_ptr)>(); - } - }); + for (std::size_t i = 0; i < items_constructed_; ++i) + { + items_[i].__destroy(); + } } - struct attrs + void start() & noexcept { - template , - STDEXEC::__forwarding_query Tag, - class... As> - requires STDEXEC::__queryable_with<_Env2, Tag, As...> - auto query(Tag, As&&... as) const - noexcept(STDEXEC::__nothrow_queryable_with<_Env2, Tag, As...>) -> decltype(auto) + std::size_t size = items_.size(); + if (size == 0) { - return STDEXEC::get_env(sndr_.sndr_).query(Tag(), static_cast(as)...); + STDEXEC::set_value(std::move(this->rcvr_)); + return; } - bulk_sender const & sndr_; - }; + std::size_t nthreads = this->pool_.available_parallelism(); + STDEXEC_ASSERT(nthreads > 0); + bwos_params params = this->pool_.params(); + std::size_t local_size = params.blockSize * params.numBlocks; + std::size_t chunk_size = __umax( + {std::size_t{1}, __umin({size / nthreads, local_size * nthreads})}); + auto& remote_queue = *this->pool_.get_remote_queue(); + auto it = std::ranges::begin(this->range_); + std::size_t i0 = 0; - [[nodiscard]] - auto get_env() const noexcept -> attrs - { - return {*this}; - } + STDEXEC_TRY + { + for (std::size_t i = 0; i < size; ++i) + { + items_[i].__construct_from(STDEXEC::connect, + set_next(this->rcvr_, item_sender_t{this, it + i}), + next_receiver_t{this}); + ++items_constructed_; + } + } + STDEXEC_CATCH_ALL + { + STDEXEC::set_error(std::move(this->rcvr_), std::current_exception()); + return; + } - private: - DerivedPoolType& pool_; - Sender sndr_; - Shape shape_; - Fun fun_; - bool parallelize_; - }; + while (i0 + chunk_size < size) + { + for (std::size_t i = i0; i < i0 + chunk_size; ++i) + { + STDEXEC::start(items_[i].__get()); + } + + std::unique_lock lock{this->start_mutex_}; + this->pool_.bulk_enqueue(remote_queue, std::move(this->tasks_), this->tasks_size_); + this->tasks_size_ = 0; + lock.unlock(); + i0 += chunk_size; + } - template - using bulk_sender_t = bulk_sender, Shape, Fun>; + for (std::size_t i = i0; i < size; ++i) + { + STDEXEC::start(items_[i].__get()); + } - friend thread_pool_base; + std::unique_lock lock{this->start_mutex_}; + this->has_started_ = true; + this->pool_.bulk_enqueue(remote_queue, std::move(this->tasks_), this->tasks_size_); + } - explicit scheduler(DerivedPoolType& pool) noexcept - : pool_(&pool) - {} + private: + using allocator_t = _allocator_of_t; + using item_sender_t = _item_sender; + using next_sender_t = next_sender_of_t; + using next_receiver_t = _next_receiver; + using item_opstate_t = connect_result_t; - DerivedPoolType* pool_; + using item_allocator_t = std::allocator_traits::template rebind_alloc< + __manual_lifetime>; - public: - auto operator==(scheduler const &) const -> bool = default; + std::vector<__manual_lifetime, item_allocator_t> items_; + std::size_t items_constructed_{}; + }; - [[nodiscard]] - constexpr auto query(STDEXEC::get_forward_progress_guarantee_t) const noexcept - -> STDEXEC::forward_progress_guarantee + template + struct _sequence { - return pool_->forward_progress_guarantee(); - } + using sender_concept = sequence_sender_tag; + using item_types = exec::item_types<_item_sender>; + using completion_signatures = + STDEXEC::completion_signatures; + + explicit _sequence(Range range, Pool& pool) + : range_(static_cast(range)) + , pool_(std::addressof(pool)) + {} - template Tag> - [[nodiscard]] - constexpr auto query(STDEXEC::get_completion_scheduler_t) const noexcept -> scheduler - { - return *this; - } + template Receiver> + auto subscribe(Receiver rcvr) && noexcept -> _opstate + { + return _opstate{std::move(range_), *pool_, std::move(rcvr)}; + } - template Tag> - [[nodiscard]] - constexpr auto query(STDEXEC::get_completion_domain_t) const noexcept -> domain - { - return {}; - } + template Receiver> + requires __decay_copyable + auto subscribe(Receiver rcvr) const & noexcept -> _opstate + { + return _opstate{range_, *pool_, std::move(rcvr)}; + } - template Tag> - [[nodiscard]] - constexpr auto query(STDEXEC::__get_completion_behavior_t) const noexcept - { - return STDEXEC::__completion_behavior::__asynchronous; - } + private: + Range range_; + Pool* pool_; + }; + } // namespace _schedule_all + } // namespace _pool - [[nodiscard]] - auto schedule() const noexcept -> sender - { - return sender{*pool_}; - } - }; + template // CRTP + struct thread_pool_base + { + using domain = _pool::_domain; + using scheduler = _pool::_scheduler; [[nodiscard]] auto get_scheduler() noexcept -> scheduler { - return scheduler{static_cast(*this)}; + return scheduler{static_cast(*this)}; } [[nodiscard]] auto available_parallelism() const -> std::uint32_t { - return static_cast(*this).available_parallelism(); + return static_cast(*this).available_parallelism(); } + protected: + // Protected (rather than private + `friend Pool`) so that `Pool` may derive from + // `thread_pool_base` indirectly, through an implementation class. + thread_pool_base() = default; + private: - void enqueue(_pool_::task_base* task, std::uint32_t tid = 0) noexcept + friend Pool; + friend thread_pool_access; + + void enqueue(_pool::_task_base* task) noexcept + { + thread_pool_access::enqueue(static_cast(*this), task); + } + + void enqueue(_pool::_task_base* task, std::uint32_t tid) noexcept { - auto& self = static_cast(*this); - static_assert(noexcept(self.enqueue(task, tid))); - self.enqueue(task, tid); + thread_pool_access::enqueue(static_cast(*this), task, tid); } - void bulk_enqueue(_pool_::task_base* task, std::uint32_t n_threads) noexcept + void bulk_enqueue(_pool::_task_base* task, std::uint32_t n_threads) noexcept { for (std::uint32_t tid = 0; tid < n_threads; ++tid) { @@ -553,29 +1009,29 @@ namespace experimental::execution } }; - namespace _pool_ + namespace _pool { - template - struct opstate : _pool_::task_base + template + struct _opstate : _task_base { - explicit opstate(PoolType& pool, Receiver rcvr) noexcept - : _pool_::task_base{&do_execute_} + explicit _opstate(Pool& pool, Receiver rcvr) noexcept + : _task_base{&do_execute_} , pool_(pool) , rcvr_(std::move(rcvr)) {} void start() & noexcept { - pool_.enqueue(this); + thread_pool_access::enqueue(pool_, this); } private: - static void do_execute_(_pool_::task_base* task, std::uint32_t) noexcept + static void do_execute_(_task_base* task, std::uint32_t) noexcept { - auto& self = *static_cast(task); + auto& self = *static_cast<_opstate*>(task); auto stoken = STDEXEC::get_stop_token(STDEXEC::get_env(self.rcvr_)); - if constexpr (STDEXEC::unstoppable_token) + if constexpr (unstoppable_token) { // NOLINT(bugprone-branch-clone) STDEXEC::set_value(std::move(self.rcvr_)); } @@ -589,10 +1045,21 @@ namespace experimental::execution } } - PoolType& pool_; - Receiver rcvr_; + Pool& pool_; + Receiver rcvr_; }; - } // namespace _pool_ + } // namespace _pool + + struct schedule_all_t + { + template + auto operator()(Pool& pool, Range&& range) const + { + return _pool::_schedule_all::_sequence{static_cast(range), pool}; + } + }; + + inline constexpr schedule_all_t schedule_all{}; } // namespace experimental::execution namespace exec = experimental::execution; diff --git a/include/exec/when_any.hpp b/include/exec/when_any.hpp index 2bde0e1d72..bc3f0ba422 100644 --- a/include/exec/when_any.hpp +++ b/include/exec/when_any.hpp @@ -76,7 +76,7 @@ namespace experimental::execution template struct __opstate_base { - __opstate_base(_Receiver&& __rcvr, std::size_t __n_senders) + explicit __opstate_base(_Receiver&& __rcvr, std::size_t __n_senders) : __count_{__n_senders} , __rcvr_{static_cast<_Receiver&&>(__rcvr)} {} @@ -119,6 +119,7 @@ namespace experimental::execution template void notify(_Tag, _Args&&... __args) noexcept { + using __error_t = __tuple; using __result_t = __decayed_tuple<_Tag, _Args...>; bool __expect = false; if (__emplaced_.compare_exchange_strong(__expect, @@ -126,26 +127,23 @@ namespace experimental::execution __std::memory_order_relaxed, __std::memory_order_relaxed)) { - // This emplacement can happen only once - if constexpr ((__nothrow_decay_copyable<_Args> && ...)) + STDEXEC_TRY { + // This emplacement can happen only once __result_.template emplace<__result_t>(_Tag{}, static_cast<_Args&&>(__args)...); } - else + STDEXEC_CATCH_ALL { - STDEXEC_TRY - { - __result_.template emplace<__result_t>(_Tag{}, static_cast<_Args&&>(__args)...); - } - STDEXEC_CATCH_ALL - { - using __error_t = __tuple; + if constexpr ((__nothrow_decay_copyable<_Args> && ...)) + __std::unreachable(); + else __result_.template emplace<__error_t>(set_error_t{}, std::current_exception()); - } } + // stop pending operations __stop_source_.request_stop(); } + __arrive(); } @@ -158,11 +156,13 @@ namespace experimental::execution { __on_stop_.reset(); auto stop_token = get_stop_token(get_env(__rcvr_)); + if (stop_token.stop_requested()) { STDEXEC::set_stopped(static_cast<_Receiver&&>(__rcvr_)); return; } + std::move(__result_).__complete(__rcvr_); } } @@ -227,8 +227,9 @@ namespace experimental::execution void start() & noexcept { - this->__on_stop_.emplace(get_stop_token(get_env(this->__rcvr_)), - typename __op_base_t::__forward_stop_request{this}); + using __fwd_stop_t = __op_base_t::__forward_stop_request; + this->__on_stop_.emplace(get_stop_token(get_env(this->__rcvr_)), __fwd_stop_t{this}); + if (this->__stop_source_.stop_requested()) { STDEXEC::set_stopped(static_cast<_Receiver&&>(this->__rcvr_)); diff --git a/include/exec/windows/windows_thread_pool.hpp b/include/exec/windows/windows_thread_pool.hpp index 87db4c4f10..ab9f76f3e9 100644 --- a/include/exec/windows/windows_thread_pool.hpp +++ b/include/exec/windows/windows_thread_pool.hpp @@ -1148,7 +1148,7 @@ namespace experimental::execution::__win32 //! Splits `[0, n)` into `size` chunks, distributing `n % size` evenly //! between the first ranks, and returns the chunk of `rank`. static auto - even_share(Shape n, std::uint32_t rank, std::uint32_t size) noexcept -> std::pair + _even_share(Shape n, std::uint32_t rank, std::uint32_t size) noexcept -> std::pair { using ushape_t = std::make_unsigned_t; auto const avg_per_agent = static_cast(n) / size; @@ -1192,7 +1192,7 @@ namespace experimental::execution::__win32 STDEXEC::__std::memory_order_relaxed); STDEXEC_ASSERT(agent < self.num_agents_); - auto const [begin, end] = even_share(self.shape_, agent, self.num_agents_); + auto const [begin, end] = _even_share(self.shape_, agent, self.num_agents_); auto const applicator = std::bind_front(std::ref(self.fun_), begin, end); auto const computation = std::bind_front(STDEXEC::__apply, applicator); diff --git a/test/exec/asio/test_asio_thread_pool.cpp b/test/exec/asio/test_asio_thread_pool.cpp index f9f84e8b6e..1c49d91df8 100644 --- a/test/exec/asio/test_asio_thread_pool.cpp +++ b/test/exec/asio/test_asio_thread_pool.cpp @@ -19,12 +19,12 @@ #include -#include -#include - #include - #include +#include +#include + +#include #if STDEXEC_USE_MODULES() import std; diff --git a/test/exec/taskflow/test_taskflow_thread_pool.cpp b/test/exec/taskflow/test_taskflow_thread_pool.cpp index 6f3922e511..0be2a7c04e 100644 --- a/test/exec/taskflow/test_taskflow_thread_pool.cpp +++ b/test/exec/taskflow/test_taskflow_thread_pool.cpp @@ -20,6 +20,7 @@ #include #include +#include #include #include diff --git a/test/exec/tbb/test_tbb_thread_pool.cpp b/test/exec/tbb/test_tbb_thread_pool.cpp index 35af533911..bcceabde84 100644 --- a/test/exec/tbb/test_tbb_thread_pool.cpp +++ b/test/exec/tbb/test_tbb_thread_pool.cpp @@ -22,6 +22,7 @@ #include #include +#include #include #include diff --git a/test/exec/test_scheduler_concept.cpp b/test/exec/test_scheduler_concept.cpp index 9a9360fbcc..6c3b6027c9 100644 --- a/test/exec/test_scheduler_concept.cpp +++ b/test/exec/test_scheduler_concept.cpp @@ -61,7 +61,7 @@ namespace return ex::forward_progress_guarantee::parallel; } - void enqueue(exec::_pool_::task_base* task, std::uint32_t tid = 0) noexcept + void enqueue(exec::_pool::_task_base* task, std::uint32_t tid = 0) noexcept { ++enqueued_; task->execute_(task, tid); diff --git a/test/exec/test_thread_pool_base.cpp b/test/exec/test_thread_pool_base.cpp index 58aac6c9cf..c2ccb728a8 100644 --- a/test/exec/test_thread_pool_base.cpp +++ b/test/exec/test_thread_pool_base.cpp @@ -38,20 +38,23 @@ namespace return parallelism_; } + std::uint32_t parallelism_ = 1; + std::uint32_t enqueued_ = 0; + + private: + friend exec::thread_pool_access; + [[nodiscard]] static constexpr auto forward_progress_guarantee() noexcept -> ex::forward_progress_guarantee { return ex::forward_progress_guarantee::parallel; } - void enqueue(exec::_pool_::task_base *task, std::uint32_t tid = 0) noexcept + void enqueue(exec::_pool::_task_base *task, std::uint32_t tid = 0) noexcept { ++enqueued_; task->execute_(task, tid); } - - std::uint32_t parallelism_ = 1; - std::uint32_t enqueued_ = 0; }; struct value_capture_error