From a76ab52463ac288cb30333ec6c78ea372c0452a4 Mon Sep 17 00:00:00 2001 From: Minh Vu Date: Thu, 8 Oct 2026 13:32:49 +0200 Subject: [PATCH 1/4] Fix bulk worker counts for narrow integer shapes --- include/exec/thread_pool_base.hpp | 4 +- test/exec/test_thread_pool_base.cpp | 111 ++++++++++++++++++++++++++-- 2 files changed, 109 insertions(+), 6 deletions(-) diff --git a/include/exec/thread_pool_base.hpp b/include/exec/thread_pool_base.hpp index 1317b0d92..591922ff4 100644 --- a/include/exec/thread_pool_base.hpp +++ b/include/exec/thread_pool_base.hpp @@ -43,6 +43,7 @@ import stdexec; # include # include # include +# include # include # include #endif @@ -230,6 +231,7 @@ namespace experimental::execution [[nodiscard]] auto num_agents_required() const -> std::uint32_t { + using common_shape_t = std::common_type_t; auto const parallelism = parallelize_ ? pool_.available_parallelism() : static_cast(1); @@ -237,7 +239,7 @@ namespace experimental::execution // ask for more agents (tasks) than we can actually deal with at one // time? return Shape{} < shape_ - ? static_cast((std::min) (shape_, static_cast(parallelism))) + ? static_cast((std::min) (shape_, parallelism)) : 0; } diff --git a/test/exec/test_thread_pool_base.cpp b/test/exec/test_thread_pool_base.cpp index 68fe40a08..7c87000a0 100644 --- a/test/exec/test_thread_pool_base.cpp +++ b/test/exec/test_thread_pool_base.cpp @@ -18,8 +18,10 @@ #include #include +#include #include #include +#include namespace ex = STDEXEC; @@ -32,7 +34,7 @@ namespace [[nodiscard]] auto available_parallelism() const noexcept -> std::uint32_t { - return 1; + return parallelism_; } [[nodiscard]] @@ -47,7 +49,8 @@ namespace task->execute_(task, tid); } - std::uint32_t enqueued_ = 0; + std::uint32_t parallelism_ = 1; + std::uint32_t enqueued_ = 0; }; struct value_capture_error @@ -126,15 +129,113 @@ namespace CHECK(pool.enqueued_ == 1); } - TEST_CASE("thread_pool_base bulk does not invoke the function with a negative shape", + TEMPLATE_TEST_CASE("thread_pool_base bulk handles pool sizes wider than the shape", + "[thread_pool_base][bulk]", + std::uint8_t, + std::int8_t, + std::uint16_t, + std::int16_t) + { + inline_test_thread_pool pool; + pool.parallelism_ = GENERATE(256u, 257u, 65'536u, 65'537u); + completion_state state; + std::array visits{}; + auto const shape = GENERATE(TestType{1}, TestType{3}, TestType{5}); + + SECTION("chunked") + { + auto sndr = ex::schedule(pool.get_scheduler()) + | ex::bulk_chunked(ex::par, + shape, + [&](TestType begin, TestType end) noexcept + { + for (auto i = begin; i < end; ++i) + { + ++visits[i]; + } + }); + auto op = ex::connect(std::move(sndr), counting_receiver{&state}); + ex::start(op); + } + + SECTION("unchunked") + { + auto sndr = ex::schedule(pool.get_scheduler()) + | ex::bulk_unchunked(ex::par, shape, [&](TestType i) noexcept { ++visits[i]; }); + auto op = ex::connect(std::move(sndr), counting_receiver{&state}); + ex::start(op); + } + + REQUIRE(state.completions_ == 1); + CHECK_FALSE(state.error_); + CHECK(pool.enqueued_ == static_cast(shape) + 1); + for (std::size_t i = 0; i < visits.size(); ++i) + { + CHECK(visits[i] == (i < static_cast(shape) ? 1 : 0)); + } + } + + TEST_CASE("thread_pool_base bulk preserves shapes wider than the worker count", + "[thread_pool_base][bulk]") + { + inline_test_thread_pool pool; + pool.parallelism_ = 2; + completion_state state; + auto const shape = GENERATE(std::uint64_t{1} << 32, + (std::numeric_limits::max)()); + std::uint64_t visited = 0; + + auto sndr = ex::schedule(pool.get_scheduler()) + | ex::bulk_chunked(ex::par, + shape, + [&](std::uint64_t begin, std::uint64_t end) noexcept + { visited += end - begin; }); + auto op = ex::connect(std::move(sndr), counting_receiver{&state}); + ex::start(op); + + CHECK(state.completions_ == 1); + CHECK_FALSE(state.error_); + CHECK(visited == shape); + CHECK(pool.enqueued_ == 3); + } + + TEST_CASE("thread_pool_base sequential bulk uses one worker with a narrow shape", + "[thread_pool_base][bulk]") + { + inline_test_thread_pool pool; + pool.parallelism_ = 256; + completion_state state; + int visited = 0; + int bulk_calls = 0; + + auto sndr = ex::schedule(pool.get_scheduler()) + | ex::bulk_chunked(ex::seq, + std::uint8_t{3}, + [&](std::uint8_t begin, std::uint8_t end) noexcept + { + visited += end - begin; + ++bulk_calls; + }); + auto op = ex::connect(std::move(sndr), counting_receiver{&state}); + ex::start(op); + + CHECK(state.completions_ == 1); + CHECK_FALSE(state.error_); + CHECK(visited == 3); + CHECK(bulk_calls == 1); + CHECK(pool.enqueued_ == 2); + } + + TEST_CASE("thread_pool_base bulk does not invoke the function with a nonpositive shape", "[thread_pool_base][bulk]") { inline_test_thread_pool pool; completion_state state; int bulk_calls = 0; - auto sndr = ex::schedule(pool.get_scheduler()) - | ex::bulk_chunked(ex::par, -1, [&](int, int) noexcept { ++bulk_calls; }); + auto const shape = GENERATE(-1, 0); + auto sndr = ex::schedule(pool.get_scheduler()) + | ex::bulk_chunked(ex::par, shape, [&](int, int) noexcept { ++bulk_calls; }); auto op = ex::connect(std::move(sndr), counting_receiver{&state}); ex::start(op); From 838754155a516c21af0d0a1229c104ee2433effa Mon Sep 17 00:00:00 2001 From: Minh Vu Date: Thu, 8 Oct 2026 14:11:57 +0200 Subject: [PATCH 2/4] Test signed 32-bit bulk shapes with very large pool sizes --- include/exec/thread_pool_base.hpp | 1 + test/exec/test_thread_pool_base.cpp | 5 +++-- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/include/exec/thread_pool_base.hpp b/include/exec/thread_pool_base.hpp index 591922ff4..7ce4899ce 100644 --- a/include/exec/thread_pool_base.hpp +++ b/include/exec/thread_pool_base.hpp @@ -38,6 +38,7 @@ import stdexec; # include "../stdexec/__detail/__transform_completion_signatures.hpp" # include "../stdexec/__detail/__type_traits.hpp" +# include # include # include # include diff --git a/test/exec/test_thread_pool_base.cpp b/test/exec/test_thread_pool_base.cpp index 7c87000a0..d16313dae 100644 --- a/test/exec/test_thread_pool_base.cpp +++ b/test/exec/test_thread_pool_base.cpp @@ -134,10 +134,11 @@ namespace std::uint8_t, std::int8_t, std::uint16_t, - std::int16_t) + std::int16_t, + std::int32_t) { inline_test_thread_pool pool; - pool.parallelism_ = GENERATE(256u, 257u, 65'536u, 65'537u); + pool.parallelism_ = GENERATE(256u, 257u, 65'536u, 65'537u, 0x80000000u, 0xffffffffu); completion_state state; std::array visits{}; auto const shape = GENERATE(TestType{1}, TestType{3}, TestType{5}); From 80b4fce87a7f4a524f450d1be43544858fb64d30 Mon Sep 17 00:00:00 2001 From: Minh Vu Date: Thu, 8 Oct 2026 14:26:52 +0200 Subject: [PATCH 3/4] Cover maximal uint8 bulk shape and explicitly include utility --- test/exec/test_thread_pool_base.cpp | 24 ++++++++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/test/exec/test_thread_pool_base.cpp b/test/exec/test_thread_pool_base.cpp index d16313dae..0fc68ae47 100644 --- a/test/exec/test_thread_pool_base.cpp +++ b/test/exec/test_thread_pool_base.cpp @@ -22,6 +22,7 @@ #include #include #include +#include namespace ex = STDEXEC; @@ -176,6 +177,29 @@ namespace } } + TEST_CASE("thread_pool_base bulk covers a maximal uint8 shape", "[thread_pool_base][bulk]") + { + inline_test_thread_pool pool; + pool.parallelism_ = 256; + completion_state state; + std::array visits{}; + + auto sndr = ex::schedule(pool.get_scheduler()) + | ex::bulk_unchunked(ex::par, + std::uint8_t{255}, + [&](std::uint8_t i) noexcept { ++visits[i]; }); + auto op = ex::connect(std::move(sndr), counting_receiver{&state}); + ex::start(op); + + CHECK(state.completions_ == 1); + CHECK_FALSE(state.error_); + CHECK(pool.enqueued_ == 256); + for (int count: visits) + { + CHECK(count == 1); + } + } + TEST_CASE("thread_pool_base bulk preserves shapes wider than the worker count", "[thread_pool_base][bulk]") { From 106f5af3e6831832a9c6cdd7ed3afeb023e022b7 Mon Sep 17 00:00:00 2001 From: Minh Vu Date: Fri, 9 Oct 2026 20:43:21 +0200 Subject: [PATCH 4/4] style: fix clang-format alignment --- test/exec/test_thread_pool_base.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/exec/test_thread_pool_base.cpp b/test/exec/test_thread_pool_base.cpp index 0fc68ae47..58aac6c9c 100644 --- a/test/exec/test_thread_pool_base.cpp +++ b/test/exec/test_thread_pool_base.cpp @@ -181,7 +181,7 @@ namespace { inline_test_thread_pool pool; pool.parallelism_ = 256; - completion_state state; + completion_state state; std::array visits{}; auto sndr = ex::schedule(pool.get_scheduler())