From 27e0f2aa8bfde64d4e4889728ee2c5e437795bac Mon Sep 17 00:00:00 2001 From: Cra3z Date: Mon, 14 Sep 2026 15:48:10 +0800 Subject: [PATCH 1/4] Decide whether to enable the default `query_parallel_scheduler_backend` based on the value of `BEMAN_EXECUTION_WITH_DEFAULT_PARALLEL_SCHEDULER_BACKEND` instead of whether it is defined --- include/beman/execution/detail/common.hpp | 4 ++++ .../detail/default_parallel_scheduler_backend.hpp | 2 +- src/beman/execution/CMakeLists.txt | 10 ++++------ tests/beman/execution/exec-parallel-scheduler.test.cpp | 2 +- 4 files changed, 10 insertions(+), 8 deletions(-) diff --git a/include/beman/execution/detail/common.hpp b/include/beman/execution/detail/common.hpp index ec320358..40f56288 100644 --- a/include/beman/execution/detail/common.hpp +++ b/include/beman/execution/detail/common.hpp @@ -17,6 +17,10 @@ #define BEMAN_SPECIALIZE_EXPORT template <> #endif +#ifndef BEMAN_EXECUTION_WITH_DEFAULT_PARALLEL_SCHEDULER_BACKEND +#define BEMAN_EXECUTION_WITH_DEFAULT_PARALLEL_SCHEDULER_BACKEND 1 +#endif + #define BEMAN_EXECUTION_TRY_EVAL(rcvr, expr) \ do { \ try { \ diff --git a/include/beman/execution/detail/default_parallel_scheduler_backend.hpp b/include/beman/execution/detail/default_parallel_scheduler_backend.hpp index ae1b321d..d89db4cc 100644 --- a/include/beman/execution/detail/default_parallel_scheduler_backend.hpp +++ b/include/beman/execution/detail/default_parallel_scheduler_backend.hpp @@ -3,7 +3,7 @@ #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 +#if BEMAN_EXECUTION_WITH_DEFAULT_PARALLEL_SCHEDULER_BACKEND #include #ifdef BEMAN_HAS_IMPORT_STD diff --git a/src/beman/execution/CMakeLists.txt b/src/beman/execution/CMakeLists.txt index f3eed017..baf59da1 100644 --- a/src/beman/execution/CMakeLists.txt +++ b/src/beman/execution/CMakeLists.txt @@ -246,12 +246,10 @@ 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() +target_compile_definitions( + ${BEMAN_EXECUTION_TARGET_NAME} + INTERFACE BEMAN_EXECUTION_WITH_DEFAULT_PARALLEL_SCHEDULER_BACKEND=$ +) if(BEMAN_USE_MODULES) target_sources( diff --git a/tests/beman/execution/exec-parallel-scheduler.test.cpp b/tests/beman/execution/exec-parallel-scheduler.test.cpp index 1b6c60c0..6ea8f5e9 100644 --- a/tests/beman/execution/exec-parallel-scheduler.test.cpp +++ b/tests/beman/execution/exec-parallel-scheduler.test.cpp @@ -127,7 +127,7 @@ auto test_parallel_scheduler_schedule() -> void { } } // namespace -#ifndef BEMAN_EXECUTION_WITH_DEFAULT_PARALLEL_SCHEDULER_BACKEND +#if !BEMAN_EXECUTION_WITH_DEFAULT_PARALLEL_SCHEDULER_BACKEND namespace beman::execution::parallel_scheduler_replacement { auto query_parallel_scheduler_backend() -> std::shared_ptr { static auto backend = std::make_shared<::test_detail::thread_pool_backend>(); From b2c90a13792af9c0ca74f684f59be1356ce68076 Mon Sep 17 00:00:00 2001 From: Cra3z Date: Mon, 14 Sep 2026 16:12:31 +0800 Subject: [PATCH 2/4] Format Code --- src/beman/execution/CMakeLists.txt | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/beman/execution/CMakeLists.txt b/src/beman/execution/CMakeLists.txt index baf59da1..3df1a8c5 100644 --- a/src/beman/execution/CMakeLists.txt +++ b/src/beman/execution/CMakeLists.txt @@ -248,7 +248,8 @@ target_sources( target_compile_definitions( ${BEMAN_EXECUTION_TARGET_NAME} - INTERFACE BEMAN_EXECUTION_WITH_DEFAULT_PARALLEL_SCHEDULER_BACKEND=$ + INTERFACE + BEMAN_EXECUTION_WITH_DEFAULT_PARALLEL_SCHEDULER_BACKEND=$ ) if(BEMAN_USE_MODULES) From fee0cbeec8d2e09e7ea39c52a3ba874f91f63361 Mon Sep 17 00:00:00 2001 From: Cra3z Date: Tue, 15 Sep 2026 10:54:50 +0800 Subject: [PATCH 3/4] Fix `thread_pool_backend::schedule_bulk_unchunked` --- .../execution/detail/psched_bulk_sender.hpp | 6 ++- include/beman/execution/detail/task.hpp | 2 +- .../beman/execution/detail/task_scheduler.hpp | 1 - .../execution/detail/thread_pool_backend.hpp | 35 ++++++++++------- .../exec-parallel-scheduler.test.cpp | 38 ++++++++++++++++++- 5 files changed, 64 insertions(+), 18 deletions(-) diff --git a/include/beman/execution/detail/psched_bulk_sender.hpp b/include/beman/execution/detail/psched_bulk_sender.hpp index 19360fda..fdc2c0fc 100644 --- a/include/beman/execution/detail/psched_bulk_sender.hpp +++ b/include/beman/execution/detail/psched_bulk_sender.hpp @@ -75,7 +75,7 @@ import beman.execution.detail.value_types_of_t; // ---------------------------------------------------------------------------- namespace beman::execution::detail { -inline constexpr ::std::size_t psched_storage_alignment = alignof(void*); +inline constexpr ::std::size_t psched_storage_alignment = alignof(::std::max_align_t); inline constexpr ::std::size_t psched_storage_size = 6uz * sizeof(void*); template struct psched_bulk_sender { @@ -101,12 +101,16 @@ struct psched_bulk_sender { auto get_env() const noexcept { return ::beman::execution::get_env(rcvr); } auto execute(::std::size_t begin, ::std::size_t end) noexcept -> void final { + assert(begin < end); + assert(IsChunked || !is_parallel_policy || end - begin == 1uz); const Shape first = is_parallel_policy ? static_cast(begin) : Shape(0); const Shape last = is_parallel_policy ? static_cast(end) : shape; const auto call_fn = [=, this](const Args&... args) { if constexpr (IsChunked) { + static_assert(::std::invocable); fn(first, last, args...); } else { + static_assert(::std::invocable); for (Shape i = first; i < last; ++i) { fn(i, args...); } diff --git a/include/beman/execution/detail/task.hpp b/include/beman/execution/detail/task.hpp index 0f24b82c..e839621d 100644 --- a/include/beman/execution/detail/task.hpp +++ b/include/beman/execution/detail/task.hpp @@ -4,11 +4,11 @@ #ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_TASK #define INCLUDED_BEMAN_EXECUTION_DETAIL_TASK +#include #include #ifdef BEMAN_HAS_IMPORT_STD import std; #else -#include #include #include #include diff --git a/include/beman/execution/detail/task_scheduler.hpp b/include/beman/execution/detail/task_scheduler.hpp index 0de0adcb..69b6a865 100644 --- a/include/beman/execution/detail/task_scheduler.hpp +++ b/include/beman/execution/detail/task_scheduler.hpp @@ -9,7 +9,6 @@ import std; #else #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 index 9a89af17..ac425353 100644 --- a/include/beman/execution/detail/thread_pool_backend.hpp +++ b/include/beman/execution/detail/thread_pool_backend.hpp @@ -92,11 +92,12 @@ class thread_pool_backend_base 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; + cookie_type(batched_bulk_task* head, ::std::size_t chunk_count, bool one_by_one) noexcept + : head(head), chunk_count(chunk_count), one_by_one(one_by_one), ref_count(chunk_count) {} + batched_bulk_task* head; + ::std::size_t chunk_count; + bool one_by_one; + ::std::size_t ref_count; }; batched_bulk_task(cookie_type* cookie, @@ -106,8 +107,16 @@ class thread_pool_backend_base : cookie(cookie), proxy(proxy), i(i), j(j) {} auto exec() noexcept -> void override { - proxy.execute(i, j); - if (cookie->ref_count.fetch_sub(1uz, ::std::memory_order_acq_rel) == 1uz) { + if (cookie->one_by_one) { + for (::std::size_t k = i; k < j; ++k) { + proxy.execute(k, k + 1uz); + } + } else { + proxy.execute(i, j); + } + + ::std::atomic_ref<::std::size_t> ref_count{cookie->ref_count}; + if (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; @@ -158,14 +167,13 @@ class thread_pool_backend_base 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); + schedule_bulk(shape, false, 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); + schedule_bulk(shape, true, proxy, storage); } protected: @@ -175,7 +183,7 @@ class thread_pool_backend_base } auto schedule_bulk(::std::size_t shape, - ::std::size_t chunk_length, + bool one_by_one, ::beman::execution::parallel_scheduler_replacement::bulk_item_receiver_proxy& proxy, ::std::span<::std::byte> storage) noexcept -> void { if (shape == 0uz) { @@ -183,7 +191,8 @@ class thread_pool_backend_base return; } - const ::std::size_t chunk_count = (shape + chunk_length - 1uz) / chunk_length; + const ::std::size_t chunk_length = (shape + num_threads() - 1uz) / num_threads(); + const ::std::size_t chunk_count = (shape + chunk_length - 1uz) / chunk_length; try { if (chunk_count == 1uz) { push_back(::std::construct_at(reinterpret_cast(storage.data()), proxy, shape)); @@ -192,7 +201,7 @@ class thread_pool_backend_base 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); + reinterpret_cast(storage.data()), head, chunk_count, one_by_one); batched_bulk_task* prev = nullptr; for (::std::size_t i = 0; i < chunk_count; ++i) { diff --git a/tests/beman/execution/exec-parallel-scheduler.test.cpp b/tests/beman/execution/exec-parallel-scheduler.test.cpp index 6ea8f5e9..4b816b05 100644 --- a/tests/beman/execution/exec-parallel-scheduler.test.cpp +++ b/tests/beman/execution/exec-parallel-scheduler.test.cpp @@ -100,9 +100,23 @@ auto test_parallel_scheduler_schedule() -> void { ASSERT(i == 114514); } { - test_std::sync_wait(test_std::schedule(sch) | test_std::bulk(test_std::par, 0uz, [](std::size_t) noexcept {})); + bool invoked = false; test_std::sync_wait(test_std::schedule(sch) | - test_std::bulk(test_std::unseq, 0uz, [](std::size_t) noexcept {})); + test_std::bulk(test_std::par, 0uz, [&invoked](std::size_t) noexcept { invoked = true; })); + ASSERT(not invoked); + test_std::sync_wait( + test_std::schedule(sch) | + test_std::bulk(test_std::unseq, 0uz, [&invoked](std::size_t) noexcept { invoked = true; })); + ASSERT(not invoked); + test_std::sync_wait(test_std::schedule(sch) | + test_std::bulk_chunked(test_std::par, 0uz, [&invoked](std::size_t, std::size_t) noexcept { + invoked = true; + })); + ASSERT(not invoked); + test_std::sync_wait( + test_std::schedule(sch) | + test_std::bulk_unchunked(test_std::par, 0uz, [&invoked](std::size_t) noexcept { invoked = true; })); + ASSERT(not invoked); } { for (auto size : {1uz, 4uz, 8uz, 16uz, 32uz}) { @@ -123,6 +137,26 @@ auto test_parallel_scheduler_schedule() -> void { ASSERT(vec[i] == 2 * static_cast(i) + 1); } } + + for (auto size : {1uz, 4uz, 8uz, 16uz, 32uz}) { + std::vector vec(size); + std::iota(vec.begin(), vec.end(), 0); + + test_std::sync_wait(test_std::schedule(sch) | + test_std::bulk_unchunked(test_std::par, vec.size(), [&vec](std::size_t i) noexcept { + vec[i] = 2 * vec[i]; + })); + for (std::size_t i = 0; i < vec.size(); ++i) { + ASSERT(vec[i] == 2 * static_cast(i)); + } + + test_std::sync_wait( + test_std::schedule(sch) | + test_std::bulk_unchunked(test_std::seq, vec.size(), [&vec](std::size_t i) noexcept { ++vec[i]; })); + for (std::size_t i = 0; i < vec.size(); ++i) { + ASSERT(vec[i] == 2 * static_cast(i) + 1); + } + } } } } // namespace From 06663cd2b413eb360fd75c7e3a46671c1d240652 Mon Sep 17 00:00:00 2001 From: Cra3z Date: Tue, 15 Sep 2026 11:02:35 +0800 Subject: [PATCH 4/4] Add missing headers --- include/beman/execution/detail/psched_bulk_sender.hpp | 1 + 1 file changed, 1 insertion(+) diff --git a/include/beman/execution/detail/psched_bulk_sender.hpp b/include/beman/execution/detail/psched_bulk_sender.hpp index fdc2c0fc..e2dd5fd1 100644 --- a/include/beman/execution/detail/psched_bulk_sender.hpp +++ b/include/beman/execution/detail/psched_bulk_sender.hpp @@ -4,6 +4,7 @@ #ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_PSCHED_BULK_SENDER #define INCLUDED_BEMAN_EXECUTION_DETAIL_PSCHED_BULK_SENDER +#include #include #ifdef BEMAN_HAS_IMPORT_STD import std;