From cee0f4473db19f447a17ae787c825deedf60d14a Mon Sep 17 00:00:00 2001 From: Cra3z <3324654761@qq.com> Date: Thu, 10 Sep 2026 22:57:17 +0800 Subject: [PATCH 1/6] Add default parallel_scheduler_backend --- CMakeLists.txt | 6 + .../default_parallel_scheduler_backend.hpp | 34 +++ .../execution/detail/parallel_scheduler.hpp | 1 + .../execution/detail/thread_pool_backend.hpp | 247 ++++++++++++++++++ src/beman/execution/CMakeLists.txt | 11 + .../parallel_scheduler_replacement.cppm | 1 + src/beman/execution/thread_pool_backend.cppm | 12 + .../exec-parallel-scheduler.test.cpp | 188 +------------ 8 files changed, 317 insertions(+), 183 deletions(-) create mode 100644 include/beman/execution/detail/default_parallel_scheduler_backend.hpp create mode 100644 include/beman/execution/detail/thread_pool_backend.hpp create mode 100644 src/beman/execution/thread_pool_backend.cppm diff --git a/CMakeLists.txt b/CMakeLists.txt index 3d80e7dd..7afed45c 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -94,6 +94,12 @@ option( ${PROJECT_IS_TOP_LEVEL} ) +option( + BEMAN_EXECUTION_WITH_DEFAULT_PARALLEL_SCHEDULER_BACKEND + "Enable building default parallel_scheduler backend implementation. Default: ON. Values: { ON, OFF }." + ON +) + add_subdirectory(src/beman/execution) if(NOT BEMAN_USE_MODULES) diff --git a/include/beman/execution/detail/default_parallel_scheduler_backend.hpp b/include/beman/execution/detail/default_parallel_scheduler_backend.hpp new file mode 100644 index 00000000..ae1b321d --- /dev/null +++ b/include/beman/execution/detail/default_parallel_scheduler_backend.hpp @@ -0,0 +1,34 @@ +// include/beman/execution/detail/default_parallel_scheduler_backend.hpp -*-C++-*- +// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + +#ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_DEFAULT_PARALLEL_SCHEDULER_BACKEND +#define INCLUDED_BEMAN_EXECUTION_DETAIL_DEFAULT_PARALLEL_SCHEDULER_BACKEND +#ifdef BEMAN_EXECUTION_WITH_DEFAULT_PARALLEL_SCHEDULER_BACKEND + +#include +#ifdef BEMAN_HAS_IMPORT_STD +import std; +#else +#include +#endif +#ifdef BEMAN_HAS_MODULES +import beman.execution.detail.parallel_scheduler_replacement; +import beman.execution.detail.thread_pool_backend; +#else +#include +#include +#endif + +// ---------------------------------------------------------------------------- + +namespace beman::execution::parallel_scheduler_replacement { +inline auto query_parallel_scheduler_backend() -> ::std::shared_ptr { + static auto backend = ::std::make_shared<::beman::execution::detail::thread_pool_backend>(); + return backend; +} +} // namespace beman::execution::parallel_scheduler_replacement + +// ---------------------------------------------------------------------------- + +#endif // BEMAN_EXECUTION_WITH_DEFAULT_PARALLEL_SCHEDULER_BACKEND +#endif // INCLUDED_BEMAN_EXECUTION_DETAIL_DEFAULT_PARALLEL_SCHEDULER_BACKEND diff --git a/include/beman/execution/detail/parallel_scheduler.hpp b/include/beman/execution/detail/parallel_scheduler.hpp index 2110f523..8d600c5b 100644 --- a/include/beman/execution/detail/parallel_scheduler.hpp +++ b/include/beman/execution/detail/parallel_scheduler.hpp @@ -45,6 +45,7 @@ import beman.execution.detail.stop_token_of_t; #else #include #include +#include #include #include #include diff --git a/include/beman/execution/detail/thread_pool_backend.hpp b/include/beman/execution/detail/thread_pool_backend.hpp new file mode 100644 index 00000000..e780a5e8 --- /dev/null +++ b/include/beman/execution/detail/thread_pool_backend.hpp @@ -0,0 +1,247 @@ +// include/beman/execution/detail/thread_pool_backend.hpp -*-C++-*- +// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + +#ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_THREAD_POOL_BACKEND +#define INCLUDED_BEMAN_EXECUTION_DETAIL_THREAD_POOL_BACKEND + +#include +#ifdef BEMAN_HAS_IMPORT_STD +import std; +#else +#include +#include +#include +#include +#include +#include +#include +#include +#include +#endif +#ifdef BEMAN_HAS_MODULES +import beman.execution.detail.parallel_scheduler_replacement; +import beman.execution.detail.unreachable; +#else +#include +#include +#endif + +// ---------------------------------------------------------------------------- + +namespace beman::execution::detail { +class thread_pool_backend_base + : public ::beman::execution::parallel_scheduler_replacement::parallel_scheduler_backend { + protected: + struct task { + task() noexcept : next(nullptr) {} + + task(const task&) = delete; + + task(task&&) = delete; + + virtual ~task() = default; + + auto operator=(const task&) -> task& = delete; + + auto operator=(task&&) -> task& = delete; + + virtual auto exec(::std::pmr::polymorphic_allocator<>) noexcept -> void = 0; + + task* next; + }; + + struct schedule_task : task { + explicit schedule_task(::beman::execution::parallel_scheduler_replacement::receiver_proxy& p) noexcept + : proxy(p) {} + + auto exec(::std::pmr::polymorphic_allocator<> alloc) noexcept -> void override { + auto& proxy_ref = proxy; + alloc.delete_object(this); + proxy_ref.set_value(); + } + + ::beman::execution::parallel_scheduler_replacement::receiver_proxy& proxy; + }; + + struct bulk_task : task { + struct shared_state_type { + ::std::span tasks; + ::std::atomic<::std::size_t> counter; + }; + + bulk_task(::std::shared_ptr counter, + ::beman::execution::parallel_scheduler_replacement::bulk_item_receiver_proxy& proxy, + ::std::size_t i, + ::std::size_t j) noexcept + : shared_state(::std::move(counter)), proxy(proxy), i(i), j(j) {} + + auto exec(::std::pmr::polymorphic_allocator<> alloc) noexcept -> void override { + proxy.execute(i, j); + if (shared_state->counter.fetch_sub(1uz, ::std::memory_order_acq_rel) == 1uz) { + auto& proxy_ref = proxy; + ::std::ranges::destroy(shared_state->tasks); + alloc.deallocate_object(shared_state->tasks.data(), shared_state->tasks.size()); + proxy_ref.set_value(); + } + } + + ::std::shared_ptr shared_state; + ::beman::execution::parallel_scheduler_replacement::bulk_item_receiver_proxy& proxy; + ::std::size_t i; + ::std::size_t j; + }; + + public: + thread_pool_backend_base() = default; + + thread_pool_backend_base(const thread_pool_backend_base&) = delete; + + thread_pool_backend_base(thread_pool_backend_base&&) = delete; + + ~thread_pool_backend_base() override = default; + + auto operator=(const thread_pool_backend_base&) -> thread_pool_backend_base& = delete; + + auto operator=(thread_pool_backend_base&&) -> thread_pool_backend_base& = delete; + + auto shutdown() -> void { + ::std::unique_lock guard{mtx}; + shutdown_requested = true; + guard.unlock(); + cv.notify_all(); + } + + auto schedule(::beman::execution::parallel_scheduler_replacement::receiver_proxy& proxy, + ::std::span<::std::byte>) noexcept -> void override { + try { + ::std::unique_lock guard{mtx}; + ::std::pmr::polymorphic_allocator<> alloc{&mempool}; + auto* t = alloc.new_object(proxy); + if (tasks_end == nullptr) { + tasks_begin = t; + } else { + tasks_end->next = t; + } + tasks_end = t; + guard.unlock(); + cv.notify_one(); + } catch (...) { + proxy.set_error(::std::current_exception()); + } + } + + auto schedule_bulk(::std::size_t shape, + ::std::size_t chunk_length, + ::beman::execution::parallel_scheduler_replacement::bulk_item_receiver_proxy& proxy, + ::std::span<::std::byte> storage) noexcept -> void { + if (shape == 0uz) { + schedule(proxy, storage); + return; + } + + const ::std::size_t chunk_count = (shape + chunk_length - 1uz) / chunk_length; + ::std::pmr::polymorphic_allocator<> alloc{&mempool}; + ::std::shared_ptr shared_state; + bulk_task* batch = nullptr; + try { + batch = alloc.allocate_object(chunk_count); + shared_state = ::std::allocate_shared( + alloc, ::std::span(batch, chunk_count), chunk_count); + } catch (...) { + if (batch) { + alloc.deallocate_object(batch, chunk_count); + } + proxy.set_error(::std::current_exception()); + return; + } + + bulk_task* prev = nullptr; + for (::std::size_t i = 0; i < chunk_count; ++i) { + const ::std::size_t begin = i * chunk_length; + const ::std::size_t end = ::std::min(begin + chunk_length, shape); + // NOLINTBEGIN(*-pointer-arithmetic-on-polymorphic-object, *-ctr56-cpp) + ::std::construct_at(batch + i, shared_state, proxy, begin, end); + if (prev) { + prev->next = &batch[i]; + } + prev = &batch[i]; + // NOLINTEND(*-pointer-arithmetic-on-polymorphic-object, *-ctr56-cpp) + } + + ::std::unique_lock guard{mtx}; + if (tasks_end == nullptr) { + tasks_begin = batch; + } else { + tasks_end->next = batch; + } + tasks_end = prev; + guard.unlock(); + cv.notify_all(); + } + + auto schedule_bulk_chunked(::std::size_t shape, + ::beman::execution::parallel_scheduler_replacement::bulk_item_receiver_proxy& proxy, + ::std::span<::std::byte> storage) noexcept -> void override { + const ::std::size_t chunk_length = (shape + num_threads - 1uz) / num_threads; + schedule_bulk(shape, chunk_length, proxy, storage); + } + + auto schedule_bulk_unchunked(::std::size_t shape, + ::beman::execution::parallel_scheduler_replacement::bulk_item_receiver_proxy& proxy, + ::std::span<::std::byte> storage) noexcept -> void override { + schedule_bulk(shape, 1uz, proxy, storage); + } + + protected: + inline static ::std::size_t num_threads = ::std::thread::hardware_concurrency(); + inline static ::std::pmr::synchronized_pool_resource mempool; + bool shutdown_requested = false; + ::std::mutex mtx; + ::std::condition_variable cv; + task* tasks_begin = nullptr; + task* tasks_end = nullptr; +}; + +struct thread_pool_backend : ::beman::execution::detail::thread_pool_backend_base { + explicit thread_pool_backend(::std::in_place_t) : workers(num_threads, &mempool) {} + + thread_pool_backend() : thread_pool_backend(::std::in_place) { + for (::std::size_t i = 0; i < num_threads; ++i) { + workers[i] = ::std::thread([this]() noexcept { this->run(); }); + } + } + + ~thread_pool_backend() override { + shutdown(); + for (auto& worker : workers) { + worker.join(); + } + } + + private: + auto run() noexcept -> void { + while (true) { + ::std::unique_lock guard{mtx}; + cv.wait(guard, [this]() noexcept { return tasks_begin != nullptr || shutdown_requested; }); + if (shutdown_requested && tasks_begin == nullptr) { + return; + } + auto task = tasks_begin; + tasks_begin = task->next; + if (tasks_begin == nullptr) { + tasks_end = nullptr; + } + task->next = nullptr; + guard.unlock(); + task->exec(&mempool); + } + } + + ::std::pmr::vector<::std::thread> workers; +}; + +} // namespace beman::execution::detail + +// ---------------------------------------------------------------------------- + +#endif // #ifdef INCLUDED_BEMAN_EXECUTION_DETAIL_THREAD_POOL_BACKEND diff --git a/src/beman/execution/CMakeLists.txt b/src/beman/execution/CMakeLists.txt index fe0d2430..7317dc37 100644 --- a/src/beman/execution/CMakeLists.txt +++ b/src/beman/execution/CMakeLists.txt @@ -75,6 +75,7 @@ target_sources( ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/decays_to.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/default_domain.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/default_impls.hpp + ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/default_parallel_scheduler_backend.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/dependent_sender.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/dependent_sender_error.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/execution_policy.hpp @@ -222,6 +223,7 @@ target_sources( ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/task.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/task_scheduler.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/then.hpp + ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/thread_pool_backend.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/transform_sender.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/try_query.hpp ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/type_list.hpp @@ -244,6 +246,14 @@ target_sources( ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/write_env.hpp ) +if (BEMAN_EXECUTION_WITH_DEFAULT_PARALLEL_SCHEDULER_BACKEND) + target_compile_definitions( + ${BEMAN_EXECUTION_TARGET_NAME} + INTERFACE + BEMAN_EXECUTION_WITH_DEFAULT_PARALLEL_SCHEDULER_BACKEND + ) +endif () + if(BEMAN_USE_MODULES) target_sources( ${BEMAN_EXECUTION_TARGET_PREFIX} @@ -446,6 +456,7 @@ if(BEMAN_USE_MODULES) task.cppm task_scheduler.cppm then.cppm + thread_pool_backend.cppm transform_sender.cppm try_query.cppm type_list.cppm diff --git a/src/beman/execution/parallel_scheduler_replacement.cppm b/src/beman/execution/parallel_scheduler_replacement.cppm index 252df8ee..952ede25 100644 --- a/src/beman/execution/parallel_scheduler_replacement.cppm +++ b/src/beman/execution/parallel_scheduler_replacement.cppm @@ -3,6 +3,7 @@ module; // SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception #include +#include export module beman.execution.detail.parallel_scheduler_replacement; diff --git a/src/beman/execution/thread_pool_backend.cppm b/src/beman/execution/thread_pool_backend.cppm new file mode 100644 index 00000000..4ce67e23 --- /dev/null +++ b/src/beman/execution/thread_pool_backend.cppm @@ -0,0 +1,12 @@ +module; +// src/beman/execution/thread_pool_backend.cppm -*-C++-*- +// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + +#include + +export module beman.execution.detail.thread_pool_backend; + +namespace beman::execution::detail { +export using beman::execution::detail::thread_pool_backend_base; +export using beman::execution::detail::thread_pool_backend; +} // namespace beman::execution::detail diff --git a/tests/beman/execution/exec-parallel-scheduler.test.cpp b/tests/beman/execution/exec-parallel-scheduler.test.cpp index dae8fe12..b525c6db 100644 --- a/tests/beman/execution/exec-parallel-scheduler.test.cpp +++ b/tests/beman/execution/exec-parallel-scheduler.test.cpp @@ -27,6 +27,7 @@ import std; #ifdef BEMAN_HAS_MODULES import beman.execution; import beman.execution.detail.schedule_result_t; +import beman.execution.detail.thread_pool_backend; #else #include #endif @@ -57,190 +58,9 @@ struct backend : replaceability::parallel_scheduler_backend { ::std::span<::std::byte>) noexcept -> void override {} }; -struct thread_pool_base : replaceability::parallel_scheduler_backend { - struct task { - task() = default; - - task(const task&) = delete; - - task(task&&) = delete; - - virtual ~task() = default; - - auto operator=(const task&) -> task& = delete; - - auto operator=(task&&) -> task& = delete; - - virtual auto exec() noexcept -> void = 0; - }; - - struct schedule_task : task { - explicit schedule_task(replaceability::receiver_proxy& p) noexcept : proxy(p) {} - - auto exec() noexcept -> void override { proxy.set_value(); } - - replaceability::receiver_proxy& proxy; - }; - - struct bulk_shared_state { - std::atomic counter; - std::exception_ptr exception; - }; - - struct bulk_task : task { - bulk_task(std::shared_ptr counter, - replaceability::bulk_item_receiver_proxy& proxy, - std::size_t i, - std::size_t j) noexcept - : shared_state(std::move(counter)), proxy(proxy), i(i), j(j) {} - - auto exec() noexcept -> void override { - proxy.execute(i, j); - if (shared_state->counter.fetch_sub(1uz, std::memory_order_acq_rel) == 1uz) { - if (shared_state->exception) { - proxy.set_error(shared_state->exception); - } else { - proxy.set_value(); - } - } - } - - std::shared_ptr shared_state; - replaceability::bulk_item_receiver_proxy& proxy; - std::size_t i; - std::size_t j; - }; - - thread_pool_base() = default; - - thread_pool_base(const thread_pool_base&) = delete; - - ~thread_pool_base() override = 0; - - auto operator=(const thread_pool_base&) = delete; - - auto shutdown() -> void { - std::unique_lock guard{mtx}; - shutdown_requested = true; - guard.unlock(); - cv.notify_all(); - } - - auto schedule(replaceability::receiver_proxy& proxy, ::std::span<::std::byte>) noexcept -> void override { - try { - auto t = std::make_unique(proxy); - std::unique_lock guard{mtx}; - tasks.push(std::move(t)); - guard.unlock(); - cv.notify_one(); - } catch (...) { - proxy.set_error(std::current_exception()); - } - } - - auto schedule_bulk(std::size_t shape, - std::size_t chunk_length, - replaceability::bulk_item_receiver_proxy& proxy, - std::span<::std::byte> storage) noexcept -> void { - if (shape == 0uz) { - schedule(proxy, storage); - return; - } - const std::size_t chunk_count = (shape + chunk_length - 1uz) / chunk_length; - std::shared_ptr shared_state; - try { - shared_state = std::make_shared(chunk_count); - } catch (...) { - proxy.set_error(std::current_exception()); - return; - } - std::unique_lock guard{mtx}; - for (std::size_t i = 0; i < chunk_count; ++i) { - try { - const std::size_t begin = i * chunk_length; - const std::size_t end = std::min(begin + chunk_length, shape); - tasks.push(std::make_unique(shared_state, proxy, begin, end)); - } catch (...) { - const std::size_t n = chunk_count - i; // the count of `bulk_task` which are not enqueued successfully - - guard.unlock(); - if (i == 1uz) { - cv.notify_one(); - } else if (i > 1uz) { - cv.notify_all(); - } - - // happens-before `proxy.set_value()/proxy.set_error(...)` in `bulk_task::exec` - shared_state->exception = std::current_exception(); - - if (shared_state->counter.fetch_sub(n, std::memory_order_acq_rel) == n) { - proxy.set_error(shared_state->exception); - } - return; - } - } - guard.unlock(); - cv.notify_all(); - } - - auto schedule_bulk_chunked(::std::size_t shape, - replaceability::bulk_item_receiver_proxy& proxy, - ::std::span<::std::byte> storage) noexcept -> void override { - const std::size_t chunk_length = (shape + num_threads - 1uz) / num_threads; - schedule_bulk(shape, chunk_length, proxy, storage); - } - - auto schedule_bulk_unchunked(::std::size_t shape, - replaceability::bulk_item_receiver_proxy& proxy, - ::std::span<::std::byte> storage) noexcept -> void override { - schedule_bulk(shape, 1uz, proxy, storage); - } - - protected: - static constexpr std::size_t num_threads = 4uz; - bool shutdown_requested = false; - std::mutex mtx; - std::condition_variable cv; - std::queue> tasks; -}; - -thread_pool_base::~thread_pool_base() = default; - -struct thread_pool_backend : thread_pool_base { - thread_pool_backend() { - for (std::size_t i = 0; i < num_threads; ++i) { - workers[i] = std::thread([this]() noexcept { this->run(); }); - } - } - - ~thread_pool_backend() override { - shutdown(); - for (auto& worker : workers) { - worker.join(); - } - } - - private: - auto run() noexcept -> void { - while (true) { - std::unique_lock guard{mtx}; - cv.wait(guard, [this]() noexcept { return !tasks.empty() || shutdown_requested; }); - if (shutdown_requested && tasks.empty()) { - return; - } - auto task = std::move(tasks.front()); - tasks.pop(); - guard.unlock(); - task->exec(); - } - } - - std::thread workers[num_threads]; -}; - // for GCC and Clang, enable -fopenmp for both compiling and linking; for MSVC, use the /openmp:llvm compiler option. #ifdef _OPENMP -struct openmp_backend : thread_pool_base { +struct openmp_backend : test_detail::thread_pool_backend_base { openmp_backend() { designee = std::thread{[this]() noexcept { #pragma omp parallel num_threads(num_threads) @@ -345,16 +165,18 @@ auto test_parallel_scheduler_schedule() -> void { } } // namespace +#ifndef BEMAN_EXECUTION_WITH_DEFAULT_PARALLEL_SCHEDULER_BACKEND namespace beman::execution::parallel_scheduler_replacement { auto query_parallel_scheduler_backend() -> std::shared_ptr { #ifdef _OPENMP static auto backend = std::make_shared<::openmp_backend>(); #else - static auto backend = std::make_shared<::thread_pool_backend>(); + static auto backend = std::make_shared<::test_detail::thread_pool_backend>(); #endif return backend; } } // namespace beman::execution::parallel_scheduler_replacement +#endif // BEMAN_EXECUTION_WITH_DEFAULT_PARALLEL_SCHEDULER_BACKEND TEST(exec_parallel_scheduler) { test_parallel_scheduler_synopsis(); From bc680eadf549e26fbbf874ac9232b8acaad8762e Mon Sep 17 00:00:00 2001 From: Cra3z <3324654761@qq.com> Date: Thu, 10 Sep 2026 23:08:39 +0800 Subject: [PATCH 2/6] Fix module export error --- src/beman/execution/parallel_scheduler.cppm | 1 + src/beman/execution/parallel_scheduler_replacement.cppm | 1 - 2 files changed, 1 insertion(+), 1 deletion(-) diff --git a/src/beman/execution/parallel_scheduler.cppm b/src/beman/execution/parallel_scheduler.cppm index 14076f84..eba7ece9 100644 --- a/src/beman/execution/parallel_scheduler.cppm +++ b/src/beman/execution/parallel_scheduler.cppm @@ -3,6 +3,7 @@ module; // SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception #include +#include export module beman.execution.detail.parallel_scheduler; diff --git a/src/beman/execution/parallel_scheduler_replacement.cppm b/src/beman/execution/parallel_scheduler_replacement.cppm index 952ede25..252df8ee 100644 --- a/src/beman/execution/parallel_scheduler_replacement.cppm +++ b/src/beman/execution/parallel_scheduler_replacement.cppm @@ -3,7 +3,6 @@ module; // SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception #include -#include export module beman.execution.detail.parallel_scheduler_replacement; From bed9ff731849945fd809e9d50b2156a9f2287e98 Mon Sep 17 00:00:00 2001 From: Cra3z Date: Fri, 11 Sep 2026 09:56:48 +0800 Subject: [PATCH 3/6] Refactor thread pool backend --- .../execution/detail/thread_pool_backend.hpp | 268 ++++++++++-------- .../exec-parallel-scheduler.test.cpp | 44 +-- 2 files changed, 153 insertions(+), 159 deletions(-) diff --git a/include/beman/execution/detail/thread_pool_backend.hpp b/include/beman/execution/detail/thread_pool_backend.hpp index e780a5e8..05dd8861 100644 --- a/include/beman/execution/detail/thread_pool_backend.hpp +++ b/include/beman/execution/detail/thread_pool_backend.hpp @@ -4,26 +4,29 @@ #ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_THREAD_POOL_BACKEND #define INCLUDED_BEMAN_EXECUTION_DETAIL_THREAD_POOL_BACKEND +#include #include #ifdef BEMAN_HAS_IMPORT_STD import std; #else +#include +#include #include #include +#include #include -#include #include -#include +#include +#include #include #include -#include #endif #ifdef BEMAN_HAS_MODULES import beman.execution.detail.parallel_scheduler_replacement; -import beman.execution.detail.unreachable; +import beman.execution.detail.psched_bulk_sender; #else #include -#include +#include #endif // ---------------------------------------------------------------------------- @@ -32,65 +35,101 @@ namespace beman::execution::detail { class thread_pool_backend_base : public ::beman::execution::parallel_scheduler_replacement::parallel_scheduler_backend { protected: - struct task { - task() noexcept : next(nullptr) {} + struct task_base { + task_base() noexcept : next(nullptr) {} - task(const task&) = delete; + task_base(const task_base&) = delete; - task(task&&) = delete; + task_base(task_base&&) = delete; - virtual ~task() = default; + virtual ~task_base() = default; - auto operator=(const task&) -> task& = delete; + auto operator=(const task_base&) -> task_base& = delete; - auto operator=(task&&) -> task& = delete; + auto operator=(task_base&&) -> task_base& = delete; - virtual auto exec(::std::pmr::polymorphic_allocator<>) noexcept -> void = 0; + virtual auto exec() noexcept -> void = 0; - task* next; + task_base* next; }; - struct schedule_task : task { + struct schedule_task : task_base { explicit schedule_task(::beman::execution::parallel_scheduler_replacement::receiver_proxy& p) noexcept : proxy(p) {} - auto exec(::std::pmr::polymorphic_allocator<> alloc) noexcept -> void override { + auto exec() noexcept -> void override { auto& proxy_ref = proxy; - alloc.delete_object(this); + ::std::destroy_at(this); proxy_ref.set_value(); } ::beman::execution::parallel_scheduler_replacement::receiver_proxy& proxy; }; - struct bulk_task : task { - struct shared_state_type { - ::std::span tasks; - ::std::atomic<::std::size_t> counter; + // `schedule_task` is small enough to fit directly into the pre-allocated storage provided by `schedule()` + static_assert(sizeof(schedule_task) <= psched_storage_size && alignof(schedule_task) <= psched_storage_alignment); + + struct single_bulk_task : task_base { + single_bulk_task(::beman::execution::parallel_scheduler_replacement::bulk_item_receiver_proxy& p, + ::std::size_t shape) noexcept + : proxy(p), shape(shape) {} + + auto exec() noexcept -> void override { + proxy.execute(0uz, shape); + auto& proxy_ref = proxy; + ::std::destroy_at(this); + proxy_ref.set_value(); + } + + ::beman::execution::parallel_scheduler_replacement::bulk_item_receiver_proxy& proxy; + ::std::size_t shape; + }; + + // `single_bulk_task` is small enough to fit directly into the pre-allocated storage provided by + // `schedule_bulk_chunked()`/`schedule_bulk_unchunked()` + static_assert(sizeof(single_bulk_task) <= psched_storage_size && + alignof(single_bulk_task) <= psched_storage_alignment); + + struct batched_bulk_task : task_base { + struct cookie_type { + cookie_type(batched_bulk_task* head, ::std::size_t chunk_count) noexcept + : head(head), chunk_count(chunk_count), ref_count(chunk_count) {} + batched_bulk_task* head; + ::std::size_t chunk_count; + ::std::atomic<::std::size_t> ref_count; }; - bulk_task(::std::shared_ptr counter, - ::beman::execution::parallel_scheduler_replacement::bulk_item_receiver_proxy& proxy, - ::std::size_t i, - ::std::size_t j) noexcept - : shared_state(::std::move(counter)), proxy(proxy), i(i), j(j) {} + batched_bulk_task(cookie_type* cookie, + ::beman::execution::parallel_scheduler_replacement::bulk_item_receiver_proxy& proxy, + ::std::size_t i, + ::std::size_t j) noexcept + : cookie(cookie), proxy(proxy), i(i), j(j) {} - auto exec(::std::pmr::polymorphic_allocator<> alloc) noexcept -> void override { + auto exec() noexcept -> void override { proxy.execute(i, j); - if (shared_state->counter.fetch_sub(1uz, ::std::memory_order_acq_rel) == 1uz) { - auto& proxy_ref = proxy; - ::std::ranges::destroy(shared_state->tasks); - alloc.deallocate_object(shared_state->tasks.data(), shared_state->tasks.size()); + if (cookie->ref_count.fetch_sub(1uz, ::std::memory_order_acq_rel) == 1uz) { + auto head = cookie->head; + const auto chunk_count = cookie->chunk_count; + auto& proxy_ref = proxy; + ::std::destroy_at(cookie); + ::std::destroy_n(head, chunk_count); + ::operator delete( + head, chunk_count * sizeof(batched_bulk_task), ::std::align_val_t{alignof(batched_bulk_task)}); proxy_ref.set_value(); } } - ::std::shared_ptr shared_state; + cookie_type* cookie; ::beman::execution::parallel_scheduler_replacement::bulk_item_receiver_proxy& proxy; ::std::size_t i; ::std::size_t j; }; + // The cookie of a batch lives in the pre-allocated storage too, so a batch costs exactly one allocation: + // the chunk array itself. + static_assert(sizeof(batched_bulk_task::cookie_type) <= psched_storage_size && + alignof(batched_bulk_task::cookie_type) <= psched_storage_alignment); + public: thread_pool_backend_base() = default; @@ -104,7 +143,7 @@ class thread_pool_backend_base auto operator=(thread_pool_backend_base&&) -> thread_pool_backend_base& = delete; - auto shutdown() -> void { + auto shutdown() noexcept -> void { ::std::unique_lock guard{mtx}; shutdown_requested = true; guard.unlock(); @@ -112,22 +151,27 @@ class thread_pool_backend_base } auto schedule(::beman::execution::parallel_scheduler_replacement::receiver_proxy& proxy, - ::std::span<::std::byte>) noexcept -> void override { - try { - ::std::unique_lock guard{mtx}; - ::std::pmr::polymorphic_allocator<> alloc{&mempool}; - auto* t = alloc.new_object(proxy); - if (tasks_end == nullptr) { - tasks_begin = t; - } else { - tasks_end->next = t; - } - tasks_end = t; - guard.unlock(); - cv.notify_one(); - } catch (...) { - proxy.set_error(::std::current_exception()); - } + ::std::span<::std::byte> storage) noexcept -> void override { + push_back(::std::construct_at(reinterpret_cast(storage.data()), proxy)); // nothrow! + } + + auto schedule_bulk_chunked(::std::size_t shape, + ::beman::execution::parallel_scheduler_replacement::bulk_item_receiver_proxy& proxy, + ::std::span<::std::byte> storage) noexcept -> void override { + const ::std::size_t chunk_length = (shape + num_threads() - 1uz) / num_threads(); + schedule_bulk(shape, chunk_length, proxy, storage); + } + + auto schedule_bulk_unchunked(::std::size_t shape, + ::beman::execution::parallel_scheduler_replacement::bulk_item_receiver_proxy& proxy, + ::std::span<::std::byte> storage) noexcept -> void override { + schedule_bulk_chunked(shape, proxy, storage); + } + + protected: + [[nodiscard]] static auto num_threads() noexcept -> ::std::size_t { + static const ::std::size_t count = ::std::max(1u, ::std::thread::hardware_concurrency()); + return count; } auto schedule_bulk(::std::size_t shape, @@ -139,105 +183,97 @@ class thread_pool_backend_base return; } - const ::std::size_t chunk_count = (shape + chunk_length - 1uz) / chunk_length; - ::std::pmr::polymorphic_allocator<> alloc{&mempool}; - ::std::shared_ptr shared_state; - bulk_task* batch = nullptr; + const ::std::size_t chunk_count = (shape + chunk_length - 1uz) / chunk_length; try { - batch = alloc.allocate_object(chunk_count); - shared_state = ::std::allocate_shared( - alloc, ::std::span(batch, chunk_count), chunk_count); - } catch (...) { - if (batch) { - alloc.deallocate_object(batch, chunk_count); + if (chunk_count == 1uz) { + push_back(::std::construct_at(reinterpret_cast(storage.data()), proxy, shape)); + } else { + auto head = static_cast(::operator new( + chunk_count * sizeof(batched_bulk_task), ::std::align_val_t{alignof(batched_bulk_task)})); + // NOLINTBEGIN(*-reinterpret-cast, *-pointer-arithmetic-on-polymorphic-object, *-ctr56-cpp) + auto cookie = ::std::construct_at( + reinterpret_cast(storage.data()), head, chunk_count); + + batched_bulk_task* prev = nullptr; + for (::std::size_t i = 0; i < chunk_count; ++i) { + const ::std::size_t begin = i * chunk_length; + const ::std::size_t end = ::std::min(begin + chunk_length, shape); + auto task = ::std::construct_at(head + i, cookie, proxy, begin, end); + if (prev) { + prev->next = task; + } + prev = task; + } + // NOLINTEND(*-reinterpret-cast, *-pointer-arithmetic-on-polymorphic-object, *-ctr56-cpp) + push_back(head, chunk_count); } + } catch (...) { proxy.set_error(::std::current_exception()); - return; } + } - bulk_task* prev = nullptr; - for (::std::size_t i = 0; i < chunk_count; ++i) { - const ::std::size_t begin = i * chunk_length; - const ::std::size_t end = ::std::min(begin + chunk_length, shape); - // NOLINTBEGIN(*-pointer-arithmetic-on-polymorphic-object, *-ctr56-cpp) - ::std::construct_at(batch + i, shared_state, proxy, begin, end); - if (prev) { - prev->next = &batch[i]; + auto push_back(task_base* t, ::std::size_t n = 1uz) noexcept -> void { + ::std::unique_lock guard{mtx}; + for (::std::size_t i = 0; i < n; ++i) { + if (auto prev_back = ::std::exchange(back, t)) { + prev_back->next = t; + } else { + front = t; } - prev = &batch[i]; - // NOLINTEND(*-pointer-arithmetic-on-polymorphic-object, *-ctr56-cpp) + t = t->next; } - - ::std::unique_lock guard{mtx}; - if (tasks_end == nullptr) { - tasks_begin = batch; + assert(t == nullptr); + guard.unlock(); + if (n == 1uz) { + cv.notify_one(); } else { - tasks_end->next = batch; + cv.notify_all(); } - tasks_end = prev; - guard.unlock(); - cv.notify_all(); } - auto schedule_bulk_chunked(::std::size_t shape, - ::beman::execution::parallel_scheduler_replacement::bulk_item_receiver_proxy& proxy, - ::std::span<::std::byte> storage) noexcept -> void override { - const ::std::size_t chunk_length = (shape + num_threads - 1uz) / num_threads; - schedule_bulk(shape, chunk_length, proxy, storage); - } - - auto schedule_bulk_unchunked(::std::size_t shape, - ::beman::execution::parallel_scheduler_replacement::bulk_item_receiver_proxy& proxy, - ::std::span<::std::byte> storage) noexcept -> void override { - schedule_bulk(shape, 1uz, proxy, storage); + [[nodiscard]] auto pop_front() noexcept -> task_base* { + ::std::unique_lock guard{mtx}; + cv.wait(guard, [this] { return front != nullptr || shutdown_requested; }); + if (front == back) { + back = nullptr; + } + return front ? ::std::exchange(front, front->next) : nullptr; } protected: - inline static ::std::size_t num_threads = ::std::thread::hardware_concurrency(); - inline static ::std::pmr::synchronized_pool_resource mempool; - bool shutdown_requested = false; - ::std::mutex mtx; - ::std::condition_variable cv; - task* tasks_begin = nullptr; - task* tasks_end = nullptr; + bool shutdown_requested = false; + ::std::mutex mtx; + ::std::condition_variable cv; + task_base* front = nullptr; + task_base* back = nullptr; }; struct thread_pool_backend : ::beman::execution::detail::thread_pool_backend_base { - explicit thread_pool_backend(::std::in_place_t) : workers(num_threads, &mempool) {} + explicit thread_pool_backend(::std::in_place_t) : workers(::std::make_unique<::std::thread[]>(num_threads())) {} thread_pool_backend() : thread_pool_backend(::std::in_place) { - for (::std::size_t i = 0; i < num_threads; ++i) { - workers[i] = ::std::thread([this]() noexcept { this->run(); }); + for (auto& worker : ::std::span(workers.get(), num_threads())) { + worker = ::std::thread(&thread_pool_backend::run, this); } } ~thread_pool_backend() override { shutdown(); - for (auto& worker : workers) { - worker.join(); + for (auto& worker : ::std::span(workers.get(), num_threads())) { + if (worker.joinable()) { + worker.join(); + } } } private: auto run() noexcept -> void { - while (true) { - ::std::unique_lock guard{mtx}; - cv.wait(guard, [this]() noexcept { return tasks_begin != nullptr || shutdown_requested; }); - if (shutdown_requested && tasks_begin == nullptr) { - return; - } - auto task = tasks_begin; - tasks_begin = task->next; - if (tasks_begin == nullptr) { - tasks_end = nullptr; - } - task->next = nullptr; - guard.unlock(); - task->exec(&mempool); + while (auto task = pop_front()) { + task->exec(); } } - ::std::pmr::vector<::std::thread> workers; + ::std::unique_ptr<::std::thread[]> workers; }; } // namespace beman::execution::detail diff --git a/tests/beman/execution/exec-parallel-scheduler.test.cpp b/tests/beman/execution/exec-parallel-scheduler.test.cpp index b525c6db..1b6c60c0 100644 --- a/tests/beman/execution/exec-parallel-scheduler.test.cpp +++ b/tests/beman/execution/exec-parallel-scheduler.test.cpp @@ -30,6 +30,7 @@ import beman.execution.detail.schedule_result_t; import beman.execution.detail.thread_pool_backend; #else #include +#include #endif namespace { @@ -58,45 +59,6 @@ struct backend : replaceability::parallel_scheduler_backend { ::std::span<::std::byte>) noexcept -> void override {} }; -// for GCC and Clang, enable -fopenmp for both compiling and linking; for MSVC, use the /openmp:llvm compiler option. -#ifdef _OPENMP -struct openmp_backend : test_detail::thread_pool_backend_base { - openmp_backend() { - designee = std::thread{[this]() noexcept { -#pragma omp parallel num_threads(num_threads) - { -#pragma omp single - { - while (true) { - std::unique_lock guard{mtx}; - cv.wait(guard, [this]() noexcept { return !tasks.empty() || shutdown_requested; }); - if (shutdown_requested && tasks.empty()) { - break; - } - auto front = std::move(tasks.front()); - tasks.pop(); - guard.unlock(); - auto front_ptr = front.release(); -#pragma omp task firstprivate(front_ptr) - { - std::unique_ptr{front_ptr}->exec(); - } - } - } - } - }}; - } - - ~openmp_backend() override { - shutdown(); - designee.join(); - } - - private: - std::thread designee; -}; -#endif - auto test_parallel_scheduler_synopsis() -> void { static_assert(!::std::default_initializable); static_assert(::std::copy_constructible); @@ -168,11 +130,7 @@ auto test_parallel_scheduler_schedule() -> void { #ifndef BEMAN_EXECUTION_WITH_DEFAULT_PARALLEL_SCHEDULER_BACKEND namespace beman::execution::parallel_scheduler_replacement { auto query_parallel_scheduler_backend() -> std::shared_ptr { -#ifdef _OPENMP - static auto backend = std::make_shared<::openmp_backend>(); -#else static auto backend = std::make_shared<::test_detail::thread_pool_backend>(); -#endif return backend; } } // namespace beman::execution::parallel_scheduler_replacement From 2f6652d6e22f4adbeb9ae564c2650003a7397c8a Mon Sep 17 00:00:00 2001 From: Cra3z Date: Fri, 11 Sep 2026 17:17:01 +0800 Subject: [PATCH 4/6] Update README --- README.md | 12 +++++++----- .../detail/parallel_scheduler_replacement.hpp | 1 - 2 files changed, 7 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index 65302974..63a125fb 100644 --- a/README.md +++ b/README.md @@ -4,7 +4,7 @@ SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception --> -[![Library Status](https://raw.githubusercontent.com/bemanproject/beman/refs/heads/main/images/badges/beman_badge-beman_library_under_development.svg)](https://github.com/bemanproject/beman/blob/main/docs/beman_library_maturity_model.md#the-beman-library-maturity-model)![Standard Target](https://github.com/bemanproject/beman/blob/main/images/badges/cpp26.svg)[![Coverage](https://coveralls.io/repos/github/bemanproject/execution/badge.svg?branch=main)](https://coveralls.io/github/bemanproject/execution?branch=main)[![Compiler Explorer Example](https://img.shields.io/badge/Try%20it%20on%20Compiler%20Explorer-grey?logo=compilerexplorer&logoColor=67c52a)](https://godbolt.org/z/jeMEWGYbM) +[![Library Status](https://raw.githubusercontent.com/bemanproject/beman/refs/heads/main/images/badges/beman_badge-beman_library_under_development.svg)](https://github.com/bemanproject/beman/blob/main/docs/beman_library_maturity_model.md#the-beman-library-maturity-model)![Standard Target](https://github.com/bemanproject/beman/blob/main/images/badges/cpp26.svg)![Build Status](https://github.com/bemanproject/execution/actions/workflows/ci_tests.yml/badge.svg)[![Coverage](https://coveralls.io/repos/github/bemanproject/execution/badge.svg?branch=main)](https://coveralls.io/github/bemanproject/execution?branch=main)[![Compiler Explorer Example](https://img.shields.io/badge/Try%20it%20on%20Compiler%20Explorer-grey?logo=compilerexplorer&logoColor=67c52a)](https://godbolt.org/z/jeMEWGYbM) @@ -64,10 +64,12 @@ You can disable building tests by setting CMake option `BEMAN_EXECUTION_BUILD_TE You can disable building examples by setting CMake option `BEMAN_EXECUTION_BUILD_EXAMPLES` to `OFF` when configuring the project. - -| Library | Linux | MacOS | Windows | -| ------- | ----- | ----- | ------- | -| build | ![Linux build status](https://github.com/bemanproject/execution/actions/workflows/linux.yml/badge.svg) | ![MacOS build status](https://github.com/bemanproject/execution/actions/workflows/macos.yml/badge.svg) | ![Window build status](https://github.com/bemanproject/execution/actions/workflows/windows.yml/badge.svg) | +By default, `beman.execution` +provides [query_parallel_scheduler_backend](https://eel.is/c++draft/exec.parschedrepl.query). If you want to use a +custom +`parallel_scheduler_backend`, you can disable the default `query_parallel_scheduler_backend` +implementation by setting CMake option +`BEMAN_EXECUTION_WITH_DEFAULT_PARALLEL_SCHEDULER_BACKEND` to `OFF` when configuring the project. The following instructions build the library and the examples: diff --git a/include/beman/execution/detail/parallel_scheduler_replacement.hpp b/include/beman/execution/detail/parallel_scheduler_replacement.hpp index b1951b31..27bce3a5 100644 --- a/include/beman/execution/detail/parallel_scheduler_replacement.hpp +++ b/include/beman/execution/detail/parallel_scheduler_replacement.hpp @@ -67,7 +67,6 @@ struct parallel_scheduler_backend { -> void = 0; }; -// TODO(P2079R10): provide the project-supported link-time replaceability hook. auto query_parallel_scheduler_backend() -> ::std::shared_ptr; } // namespace beman::execution::parallel_scheduler_replacement From bfcb469e6380bb085c66aae62f50207d92318556 Mon Sep 17 00:00:00 2001 From: Cra3z <3324654761@qq.com> Date: Fri, 11 Sep 2026 21:27:11 +0800 Subject: [PATCH 5/6] Add workaround for gcc15 ICE --- include/beman/execution/detail/parallel_scheduler.hpp | 2 +- include/beman/execution/detail/thread_pool_backend.hpp | 2 +- src/beman/execution/parallel_scheduler.cppm | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/include/beman/execution/detail/parallel_scheduler.hpp b/include/beman/execution/detail/parallel_scheduler.hpp index 8d600c5b..c1d2e704 100644 --- a/include/beman/execution/detail/parallel_scheduler.hpp +++ b/include/beman/execution/detail/parallel_scheduler.hpp @@ -45,7 +45,7 @@ import beman.execution.detail.stop_token_of_t; #else #include #include -#include +#include // IWYU pragma: keep #include #include #include diff --git a/include/beman/execution/detail/thread_pool_backend.hpp b/include/beman/execution/detail/thread_pool_backend.hpp index 05dd8861..9a89af17 100644 --- a/include/beman/execution/detail/thread_pool_backend.hpp +++ b/include/beman/execution/detail/thread_pool_backend.hpp @@ -249,7 +249,7 @@ class thread_pool_backend_base }; struct thread_pool_backend : ::beman::execution::detail::thread_pool_backend_base { - explicit thread_pool_backend(::std::in_place_t) : workers(::std::make_unique<::std::thread[]>(num_threads())) {} + explicit thread_pool_backend(::std::in_place_t) : workers(::new ::std::thread[num_threads()]) {} thread_pool_backend() : thread_pool_backend(::std::in_place) { for (auto& worker : ::std::span(workers.get(), num_threads())) { diff --git a/src/beman/execution/parallel_scheduler.cppm b/src/beman/execution/parallel_scheduler.cppm index eba7ece9..c5380546 100644 --- a/src/beman/execution/parallel_scheduler.cppm +++ b/src/beman/execution/parallel_scheduler.cppm @@ -2,8 +2,8 @@ module; // src/beman/execution/parallel_scheduler.cppm -*-C++-*- // SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception +#include // IWYU pragma: keep #include -#include export module beman.execution.detail.parallel_scheduler; From 76caffab1adbacd14b3acfd90dffb42215e5da43 Mon Sep 17 00:00:00 2001 From: Cra3z <3324654761@qq.com> Date: Fri, 11 Sep 2026 21:36:37 +0800 Subject: [PATCH 6/6] Format code --- src/beman/execution/CMakeLists.txt | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/src/beman/execution/CMakeLists.txt b/src/beman/execution/CMakeLists.txt index 7317dc37..f3eed017 100644 --- a/src/beman/execution/CMakeLists.txt +++ b/src/beman/execution/CMakeLists.txt @@ -246,13 +246,12 @@ target_sources( ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/write_env.hpp ) -if (BEMAN_EXECUTION_WITH_DEFAULT_PARALLEL_SCHEDULER_BACKEND) +if(BEMAN_EXECUTION_WITH_DEFAULT_PARALLEL_SCHEDULER_BACKEND) target_compile_definitions( ${BEMAN_EXECUTION_TARGET_NAME} - INTERFACE - BEMAN_EXECUTION_WITH_DEFAULT_PARALLEL_SCHEDULER_BACKEND + INTERFACE BEMAN_EXECUTION_WITH_DEFAULT_PARALLEL_SCHEDULER_BACKEND ) -endif () +endif() if(BEMAN_USE_MODULES) target_sources(