From c56f49970a3f1dfe713423352e1dcd118425ae92 Mon Sep 17 00:00:00 2001 From: Minh Vu Date: Tue, 11 Aug 2026 01:10:59 +0200 Subject: [PATCH 1/8] Avoid RMWs when polling empty remote queues --- include/exec/static_thread_pool.hpp | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/include/exec/static_thread_pool.hpp b/include/exec/static_thread_pool.hpp index f8f12a058..2cc003de5 100644 --- a/include/exec/static_thread_pool.hpp +++ b/include/exec/static_thread_pool.hpp @@ -171,7 +171,11 @@ namespace experimental::execution __intrusive_queue<&task_base::next_> tasks{}; while (head != nullptr) { - tasks.append(head->queues_[tid].pop_all_reversed()); + auto& queue = head->queues_[tid]; + if (!queue.empty()) + { + tasks.append(queue.pop_all_reversed()); + } head = head->next_; } return tasks; From f2530568df66e0ea00b04eae985865f3af39c8d5 Mon Sep 17 00:00:00 2001 From: Minh Vu Date: Fri, 14 Aug 2026 13:54:14 +0200 Subject: [PATCH 2/8] Fix remote poll before worker sleep --- include/exec/static_thread_pool.hpp | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/include/exec/static_thread_pool.hpp b/include/exec/static_thread_pool.hpp index 2cc003de5..9b73b5397 100644 --- a/include/exec/static_thread_pool.hpp +++ b/include/exec/static_thread_pool.hpp @@ -165,14 +165,15 @@ namespace experimental::execution } } - auto pop_all_reversed(std::size_t tid) noexcept -> __intrusive_queue<&task_base::next_> + auto pop_all_reversed(std::size_t tid, bool force) noexcept + -> __intrusive_queue<&task_base::next_> { remote_queue* head = head_.load(__std::memory_order_acquire); __intrusive_queue<&task_base::next_> tasks{}; while (head != nullptr) { auto& queue = head->queues_[tid]; - if (!queue.empty()) + if (force || !queue.empty()) { tasks.append(queue.pop_all_reversed()); } @@ -649,7 +650,7 @@ namespace experimental::execution }; auto try_pop() -> pop_result; - auto try_remote() -> pop_result; + auto try_remote(bool force = false) -> pop_result; auto try_steal(std::span victims) -> pop_result; auto try_steal_near() -> pop_result; auto try_steal_any() -> pop_result; @@ -975,10 +976,12 @@ namespace experimental::execution } inline auto - _static_thread_pool::thread_state::try_remote() -> _static_thread_pool::thread_state::pop_result + _static_thread_pool::thread_state::try_remote(bool force) + -> _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_); + __intrusive_queue<&task_base::next_> remotes = + pool_->remotes_.pop_all_reversed(index_, force); pending_queue_.append(std::move(remotes)); if (!pending_queue_.empty()) { @@ -1133,7 +1136,7 @@ namespace experimental::execution state expected = state::running; if (state_.compare_exchange_weak(expected, state::sleeping, __std::memory_order_relaxed)) { - result = try_remote(); + result = try_remote(true); if (result.task) { return result; From 7885761b76ead00d7076e3e4f192cb0936cde711 Mon Sep 17 00:00:00 2001 From: Minh Vu Date: Fri, 14 Aug 2026 15:22:48 +0200 Subject: [PATCH 3/8] Harden remote queue sleep polling --- examples/benchmark/common.hpp | 7 ++- include/exec/static_thread_pool.hpp | 12 +++- test/exec/test_static_thread_pool.cpp | 87 +++++++++++++++++++++++++++ 3 files changed, 103 insertions(+), 3 deletions(-) diff --git a/examples/benchmark/common.hpp b/examples/benchmark/common.hpp index 30ebc10f4..a5e521441 100644 --- a/examples/benchmark/common.hpp +++ b/examples/benchmark/common.hpp @@ -23,6 +23,7 @@ #include #include #include +#include #include #include #include @@ -120,6 +121,10 @@ void my_main(int argc, char** argv, exec::numa_policy policy = exec::get_numa_po nthreads = std::atoi(argv[1]); } std::size_t total_scheds = 10'000'000; + if (argc > 2) + { + total_scheds = static_cast(std::strtoull(argv[2], nullptr, 10)); + } #ifndef STDEXEC_NO_MONOTONIC_BUFFER_RESOURCE std::vector> buffers; #endif @@ -192,4 +197,4 @@ void my_main(int argc, char** argv, exec::numa_policy policy = exec::get_numa_po auto [dur_ms, ops_per_sec, avg, max, min, stddev] = compute_perf(starts, ends, warmup, nRuns - 1, total_scheds); std::cout << avg << " | " << max << " | " << min << " | " << stddev << "\n"; -} \ No newline at end of file +} diff --git a/include/exec/static_thread_pool.hpp b/include/exec/static_thread_pool.hpp index 9b73b5397..3a423c3cf 100644 --- a/include/exec/static_thread_pool.hpp +++ b/include/exec/static_thread_pool.hpp @@ -1134,11 +1134,19 @@ namespace experimental::execution return result; } state expected = state::running; - if (state_.compare_exchange_weak(expected, state::sleeping, __std::memory_order_relaxed)) + if (state_.compare_exchange_weak(expected, + state::sleeping, + __std::memory_order_relaxed, + __std::memory_order_acquire)) { result = try_remote(true); if (result.task) { + state expected_sleeping = state::sleeping; + state_.compare_exchange_strong(expected_sleeping, + state::running, + __std::memory_order_relaxed, + __std::memory_order_relaxed); return result; } set_sleeping(); @@ -1158,7 +1166,7 @@ namespace experimental::execution inline auto _static_thread_pool::thread_state::notify() -> bool { - if (state_.exchange(state::notified, __std::memory_order_relaxed) == state::sleeping) + if (state_.exchange(state::notified, __std::memory_order_release) == state::sleeping) { { std::lock_guard lock{mut_}; diff --git a/test/exec/test_static_thread_pool.cpp b/test/exec/test_static_thread_pool.cpp index 01109af5a..cd5247961 100644 --- a/test/exec/test_static_thread_pool.cpp +++ b/test/exec/test_static_thread_pool.cpp @@ -16,17 +16,21 @@ #include #include +#include #include #include #include // IWYU pragma: keep #include +#include #include +#include #include #include #include #include #include +#include namespace ex = STDEXEC; namespace @@ -229,3 +233,86 @@ TEST_CASE("bulk on static_thread_pool executes on multiple threads, take 2", ex::sync_wait(std::move(sender)); REQUIRE(thread_ids.size() == num_of_threads); } + +TEST_CASE("static_thread_pool drains remote work after idle transitions", + "[types][static_thread_pool][stress]") +{ + constexpr std::size_t num_producers = 4; + constexpr std::size_t rounds = 10'000; + + std::latch ready{num_producers}; + std::atomic start{false}; + std::atomic stop{false}; + std::vector> completed(num_producers); + std::vector producers; + producers.reserve(num_producers); + for (auto& count: completed) + { + count.store(0, std::memory_order_relaxed); + } + + exec::static_thread_pool pool{1}; + auto scheduler = pool.get_scheduler(); + + for (std::size_t producer = 0; producer < num_producers; ++producer) + { + producers.emplace_back([&, producer] + { + ready.count_down(); + while (!start.load(std::memory_order_acquire)) + { + std::this_thread::yield(); + } + + auto* const producer_completed = &completed[producer]; + std::size_t expected = 0; + for (std::size_t round = 0; round < rounds && !stop.load(std::memory_order_relaxed); + ++round) + { + std::size_t const batch_size = (round % 4 == 0) ? 2 : 1; + expected += batch_size; + for (std::size_t i = 0; i < batch_size; ++i) + { + exec::start_detached( + ex::schedule(scheduler) + | ex::then([producer_completed] + { producer_completed->fetch_add(1, std::memory_order_relaxed); })); + } + + while (!stop.load(std::memory_order_relaxed) + && producer_completed->load(std::memory_order_relaxed) < expected) + { + std::this_thread::yield(); + } + std::this_thread::yield(); + } + }); + } + + ready.wait(); + start.store(true, std::memory_order_release); + + auto const expected = num_producers * rounds + num_producers * ((rounds + 3) / 4); + auto const deadline = std::chrono::steady_clock::now() + std::chrono::seconds(10); + auto completed_total = [&] + { + std::size_t result = 0; + for (auto const& count: completed) + { + result += count.load(std::memory_order_relaxed); + } + return result; + }; + + while (completed_total() < expected && std::chrono::steady_clock::now() < deadline) + { + std::this_thread::yield(); + } + stop.store(true, std::memory_order_release); + for (auto& producer: producers) + { + producer.join(); + } + + CHECK(completed_total() == expected); +} From 361048d751deb7f3e9a7e211b049df79e88da40f Mon Sep 17 00:00:00 2001 From: Minh Vu Date: Fri, 14 Aug 2026 16:41:28 +0200 Subject: [PATCH 4/8] Consume remote poll notifications with acquire RMW --- include/exec/static_thread_pool.hpp | 4 +- test/rrd/CMakeLists.txt | 2 +- test/rrd/static_thread_pool_remote_poll.cpp | 159 ++++++++++++++++++++ 3 files changed, 162 insertions(+), 3 deletions(-) create mode 100644 test/rrd/static_thread_pool_remote_poll.cpp diff --git a/include/exec/static_thread_pool.hpp b/include/exec/static_thread_pool.hpp index 3a423c3cf..55fb36ce9 100644 --- a/include/exec/static_thread_pool.hpp +++ b/include/exec/static_thread_pool.hpp @@ -1137,7 +1137,7 @@ namespace experimental::execution if (state_.compare_exchange_weak(expected, state::sleeping, __std::memory_order_relaxed, - __std::memory_order_acquire)) + __std::memory_order_relaxed)) { result = try_remote(true); if (result.task) @@ -1158,7 +1158,7 @@ namespace experimental::execution { lock.unlock(); } - state_.store(state::running, __std::memory_order_relaxed); + state_.exchange(state::running, __std::memory_order_acquire); result = try_pop(); } return result; diff --git a/test/rrd/CMakeLists.txt b/test/rrd/CMakeLists.txt index 36746b5a6..d8feafc67 100644 --- a/test/rrd/CMakeLists.txt +++ b/test/rrd/CMakeLists.txt @@ -54,7 +54,7 @@ function(add_relacy_test target_name) endfunction() set(relacy_tests async_scope bwos_lifo_queue intrusive_mpsc_queue split - sync_wait) + static_thread_pool_remote_poll sync_wait) foreach(test ${relacy_tests}) add_relacy_test(${test}) diff --git a/test/rrd/static_thread_pool_remote_poll.cpp b/test/rrd/static_thread_pool_remote_poll.cpp new file mode 100644 index 000000000..8349c42b6 --- /dev/null +++ b/test/rrd/static_thread_pool_remote_poll.cpp @@ -0,0 +1,159 @@ +/* + * Copyright (c) 2026 NVIDIA Corporation + * + * Licensed under the Apache License Version 2.0 with LLVM Exceptions + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * https://llvm.org/LICENSE.txt + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include + +#include + +struct remote_queue_node +{ + remote_queue_node* next_ = nullptr; +}; + +struct static_thread_pool_remote_poll : rl::test_suite +{ + static constexpr int running = 0; + static constexpr int sleeping = 1; + static constexpr int notified = 2; + + std::atomic state_{running}; + std::atomic head_{nullptr}; + std::atomic first_notification_published_{false}; + remote_queue_node first_{}; + remote_queue_node second_{}; + bool worker_would_sleep_ = false; + + void before() + { + state_.store(running, std::memory_order_relaxed); + head_.store(nullptr, std::memory_order_relaxed); + first_notification_published_.store(false, std::memory_order_relaxed); + first_.next_ = nullptr; + second_.next_ = nullptr; + worker_would_sleep_ = false; + } + + void publish(remote_queue_node& node) + { + auto* old_head = head_.load(std::memory_order_relaxed); + do + { + node.next_ = old_head; + } + while (!head_.compare_exchange_weak(old_head, &node, std::memory_order_acq_rel)); + } + + void notify() + { + state_.exchange(notified, std::memory_order_release); + } + + auto sees_second_node() -> bool + { + for (auto* node = head_.load(std::memory_order_acquire); node != nullptr; + node = node->next_) + { + if (node == &second_) + { + return true; + } + } + return false; + } + + void worker_poll() + { + if (sees_second_node()) + { + return; + } + + int expected = running; + if (!state_.compare_exchange_weak(expected, + sleeping, + std::memory_order_relaxed, + std::memory_order_relaxed)) + { + // This acquire RMW must consume a preceding notification or leave a + // later notification visible to the next sleep transition. + state_.exchange(running, std::memory_order_acquire); + if (sees_second_node()) + { + return; + } + + expected = running; + if (!state_.compare_exchange_weak(expected, + sleeping, + std::memory_order_relaxed, + std::memory_order_relaxed)) + { + return; + } + } + + if (sees_second_node()) + { + int expected_sleeping = sleeping; + state_.compare_exchange_strong(expected_sleeping, + running, + std::memory_order_relaxed, + std::memory_order_relaxed); + return; + } + + worker_would_sleep_ = true; + } + + void thread(unsigned thread_id) + { + if (thread_id == 0) + { + publish(first_); + notify(); + first_notification_published_.store(true, std::memory_order_release); + } + else if (thread_id == 1) + { + while (!first_notification_published_.load(std::memory_order_acquire)) + { + } + publish(second_); + notify(); + } + else + { + worker_poll(); + } + } + + void after() + { + if (worker_would_sleep_) + { + RL_ASSERT(state_.load(std::memory_order_acquire) != sleeping); + } + } +}; + +auto main() -> int +{ + rl::test_params p; + p.iteration_count = 50000; + p.execution_depth_limit = 10000; + p.search_type = rl::random_scheduler_type; + return rl::simulate(p) ? 0 : 1; +} From 349c69276903f184d8203d2f794c00c1bfba12e3 Mon Sep 17 00:00:00 2001 From: Minh Vu Date: Fri, 14 Aug 2026 18:55:15 +0200 Subject: [PATCH 5/8] Format static thread pool polling changes --- include/exec/static_thread_pool.hpp | 11 ++-- test/exec/test_static_thread_pool.cpp | 63 +++++++++++---------- test/rrd/static_thread_pool_remote_poll.cpp | 5 +- 3 files changed, 39 insertions(+), 40 deletions(-) diff --git a/include/exec/static_thread_pool.hpp b/include/exec/static_thread_pool.hpp index 55fb36ce9..41be137d5 100644 --- a/include/exec/static_thread_pool.hpp +++ b/include/exec/static_thread_pool.hpp @@ -165,8 +165,8 @@ namespace experimental::execution } } - auto pop_all_reversed(std::size_t tid, bool force) noexcept - -> __intrusive_queue<&task_base::next_> + auto + pop_all_reversed(std::size_t tid, bool force) noexcept -> __intrusive_queue<&task_base::next_> { remote_queue* head = head_.load(__std::memory_order_acquire); __intrusive_queue<&task_base::next_> tasks{}; @@ -975,13 +975,12 @@ namespace experimental::execution tmp.clear(); } - inline auto - _static_thread_pool::thread_state::try_remote(bool force) + inline auto _static_thread_pool::thread_state::try_remote(bool force) -> _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_, force); + __intrusive_queue<&task_base::next_> remotes = pool_->remotes_.pop_all_reversed(index_, + force); pending_queue_.append(std::move(remotes)); if (!pending_queue_.empty()) { diff --git a/test/exec/test_static_thread_pool.cpp b/test/exec/test_static_thread_pool.cpp index cd5247961..c79915d00 100644 --- a/test/exec/test_static_thread_pool.cpp +++ b/test/exec/test_static_thread_pool.cpp @@ -240,11 +240,11 @@ TEST_CASE("static_thread_pool drains remote work after idle transitions", constexpr std::size_t num_producers = 4; constexpr std::size_t rounds = 10'000; - std::latch ready{num_producers}; - std::atomic start{false}; - std::atomic stop{false}; + std::latch ready{num_producers}; + std::atomic start{false}; + std::atomic stop{false}; std::vector> completed(num_producers); - std::vector producers; + std::vector producers; producers.reserve(num_producers); for (auto& count: completed) { @@ -256,48 +256,49 @@ TEST_CASE("static_thread_pool drains remote work after idle transitions", for (std::size_t producer = 0; producer < num_producers; ++producer) { - producers.emplace_back([&, producer] - { - ready.count_down(); - while (!start.load(std::memory_order_acquire)) - { - std::this_thread::yield(); - } - - auto* const producer_completed = &completed[producer]; - std::size_t expected = 0; - for (std::size_t round = 0; round < rounds && !stop.load(std::memory_order_relaxed); - ++round) + producers.emplace_back( + [&, producer] { - std::size_t const batch_size = (round % 4 == 0) ? 2 : 1; - expected += batch_size; - for (std::size_t i = 0; i < batch_size; ++i) + ready.count_down(); + while (!start.load(std::memory_order_acquire)) { - exec::start_detached( - ex::schedule(scheduler) - | ex::then([producer_completed] - { producer_completed->fetch_add(1, std::memory_order_relaxed); })); + std::this_thread::yield(); } - while (!stop.load(std::memory_order_relaxed) - && producer_completed->load(std::memory_order_relaxed) < expected) + auto* const producer_completed = &completed[producer]; + std::size_t expected = 0; + for (std::size_t round = 0; round < rounds && !stop.load(std::memory_order_relaxed); + ++round) { + std::size_t const batch_size = (round % 4 == 0) ? 2 : 1; + expected += batch_size; + for (std::size_t i = 0; i < batch_size; ++i) + { + exec::start_detached( + ex::schedule(scheduler) + | ex::then([producer_completed] + { producer_completed->fetch_add(1, std::memory_order_relaxed); })); + } + + while (!stop.load(std::memory_order_relaxed) + && producer_completed->load(std::memory_order_relaxed) < expected) + { + std::this_thread::yield(); + } std::this_thread::yield(); } - std::this_thread::yield(); - } - }); + }); } ready.wait(); start.store(true, std::memory_order_release); - auto const expected = num_producers * rounds + num_producers * ((rounds + 3) / 4); - auto const deadline = std::chrono::steady_clock::now() + std::chrono::seconds(10); + auto const expected = num_producers * rounds + num_producers * ((rounds + 3) / 4); + auto const deadline = std::chrono::steady_clock::now() + std::chrono::seconds(10); auto completed_total = [&] { std::size_t result = 0; - for (auto const& count: completed) + for (auto const & count: completed) { result += count.load(std::memory_order_relaxed); } diff --git a/test/rrd/static_thread_pool_remote_poll.cpp b/test/rrd/static_thread_pool_remote_poll.cpp index 8349c42b6..43d67dd1d 100644 --- a/test/rrd/static_thread_pool_remote_poll.cpp +++ b/test/rrd/static_thread_pool_remote_poll.cpp @@ -34,7 +34,7 @@ struct static_thread_pool_remote_poll : rl::test_suite first_notification_published_{false}; remote_queue_node first_{}; remote_queue_node second_{}; - bool worker_would_sleep_ = false; + bool worker_would_sleep_ = false; void before() { @@ -63,8 +63,7 @@ struct static_thread_pool_remote_poll : rl::test_suite bool { - for (auto* node = head_.load(std::memory_order_acquire); node != nullptr; - node = node->next_) + for (auto* node = head_.load(std::memory_order_acquire); node != nullptr; node = node->next_) { if (node == &second_) { From ee93b235ef11ef04a8ae6256c6eaf53fee5aa633 Mon Sep 17 00:00:00 2001 From: Minh Vu Date: Sat, 15 Aug 2026 00:34:37 +0200 Subject: [PATCH 6/8] Clarify remote polling modes --- include/exec/static_thread_pool.hpp | 24 +++++++++++++++++------- 1 file changed, 17 insertions(+), 7 deletions(-) diff --git a/include/exec/static_thread_pool.hpp b/include/exec/static_thread_pool.hpp index 41be137d5..a8839bf0c 100644 --- a/include/exec/static_thread_pool.hpp +++ b/include/exec/static_thread_pool.hpp @@ -139,6 +139,12 @@ namespace experimental::execution std::size_t index_{(std::numeric_limits::max)()}; }; + enum class remote_poll_mode + { + speculative, + before_sleep + }; + struct remote_queue_list { private: @@ -166,14 +172,15 @@ namespace experimental::execution } auto - pop_all_reversed(std::size_t tid, bool force) noexcept -> __intrusive_queue<&task_base::next_> + 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{}; while (head != nullptr) { auto& queue = head->queues_[tid]; - if (force || !queue.empty()) + if (mode == remote_poll_mode::before_sleep || !queue.empty()) { tasks.append(queue.pop_all_reversed()); } @@ -650,7 +657,7 @@ namespace experimental::execution }; auto try_pop() -> pop_result; - auto try_remote(bool force = false) -> pop_result; + auto try_remote(remote_poll_mode mode) -> pop_result; auto try_steal(std::span victims) -> pop_result; auto try_steal_near() -> pop_result; auto try_steal_any() -> pop_result; @@ -975,12 +982,12 @@ namespace experimental::execution tmp.clear(); } - inline auto _static_thread_pool::thread_state::try_remote(bool force) + 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_, - force); + mode); pending_queue_.append(std::move(remotes)); if (!pending_queue_.empty()) { @@ -1000,7 +1007,7 @@ namespace experimental::execution { return result; } - return try_remote(); + return try_remote(remote_poll_mode::speculative); } inline auto _static_thread_pool::thread_state::try_steal(std::span victims) @@ -1138,7 +1145,10 @@ namespace experimental::execution __std::memory_order_relaxed, __std::memory_order_relaxed)) { - result = try_remote(true); + // 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); if (result.task) { state expected_sleeping = state::sleeping; From 16f3342fca8c604c751fbda0eb2e12657f497275 Mon Sep 17 00:00:00 2001 From: Minh Vu Date: Sat, 15 Aug 2026 01:14:58 +0200 Subject: [PATCH 7/8] Strengthen remote polling coverage --- examples/benchmark/common.hpp | 5 - test/exec/test_static_thread_pool.cpp | 149 +++++++++++-------- test/rrd/static_thread_pool_remote_poll.cpp | 154 ++++++++++++++++---- 3 files changed, 207 insertions(+), 101 deletions(-) diff --git a/examples/benchmark/common.hpp b/examples/benchmark/common.hpp index a5e521441..edb95014a 100644 --- a/examples/benchmark/common.hpp +++ b/examples/benchmark/common.hpp @@ -23,7 +23,6 @@ #include #include #include -#include #include #include #include @@ -121,10 +120,6 @@ void my_main(int argc, char** argv, exec::numa_policy policy = exec::get_numa_po nthreads = std::atoi(argv[1]); } std::size_t total_scheds = 10'000'000; - if (argc > 2) - { - total_scheds = static_cast(std::strtoull(argv[2], nullptr, 10)); - } #ifndef STDEXEC_NO_MONOTONIC_BUFFER_RESOURCE std::vector> buffers; #endif diff --git a/test/exec/test_static_thread_pool.cpp b/test/exec/test_static_thread_pool.cpp index c79915d00..485745063 100644 --- a/test/exec/test_static_thread_pool.cpp +++ b/test/exec/test_static_thread_pool.cpp @@ -26,6 +26,7 @@ #include #include #include +#include #include #include #include @@ -234,86 +235,106 @@ TEST_CASE("bulk on static_thread_pool executes on multiple threads, take 2", REQUIRE(thread_ids.size() == num_of_threads); } -TEST_CASE("static_thread_pool drains remote work after idle transitions", - "[types][static_thread_pool][stress]") +namespace { - constexpr std::size_t num_producers = 4; - constexpr std::size_t rounds = 10'000; - - std::latch ready{num_producers}; - std::atomic start{false}; - std::atomic stop{false}; - std::vector> completed(num_producers); - std::vector producers; - producers.reserve(num_producers); - for (auto& count: completed) + void run_remote_poll_stress(bool separate_schedulers) { - count.store(0, std::memory_order_relaxed); - } - - exec::static_thread_pool pool{1}; - auto scheduler = pool.get_scheduler(); + constexpr std::size_t num_producers = 4; + constexpr std::size_t rounds = 10'000; + + std::latch ready{num_producers}; + std::atomic start{false}; + std::atomic stop{false}; + std::vector> completed(num_producers); + std::vector producers; + producers.reserve(num_producers); + for (auto& count: completed) + { + count.store(0, std::memory_order_relaxed); + } - for (std::size_t producer = 0; producer < num_producers; ++producer) - { - producers.emplace_back( - [&, producer] - { - ready.count_down(); - while (!start.load(std::memory_order_acquire)) - { - std::this_thread::yield(); - } + exec::static_thread_pool pool{1}; + using scheduler_t = decltype(pool.get_scheduler()); + std::optional shared_scheduler; + if (!separate_schedulers) + { + shared_scheduler.emplace(pool.get_scheduler()); + } - auto* const producer_completed = &completed[producer]; - std::size_t expected = 0; - for (std::size_t round = 0; round < rounds && !stop.load(std::memory_order_relaxed); - ++round) + for (std::size_t producer = 0; producer < num_producers; ++producer) + { + producers.emplace_back( + [&, producer] { - std::size_t const batch_size = (round % 4 == 0) ? 2 : 1; - expected += batch_size; - for (std::size_t i = 0; i < batch_size; ++i) + auto scheduler = separate_schedulers ? pool.get_scheduler() : *shared_scheduler; + ready.count_down(); + while (!start.load(std::memory_order_acquire)) { - exec::start_detached( - ex::schedule(scheduler) - | ex::then([producer_completed] - { producer_completed->fetch_add(1, std::memory_order_relaxed); })); + std::this_thread::yield(); } - while (!stop.load(std::memory_order_relaxed) - && producer_completed->load(std::memory_order_relaxed) < expected) + auto* const producer_completed = &completed[producer]; + std::size_t expected = 0; + for (std::size_t round = 0; round < rounds && !stop.load(std::memory_order_relaxed); + ++round) { + std::size_t const batch_size = (round % 4 == 0) ? 2 : 1; + expected += batch_size; + for (std::size_t i = 0; i < batch_size; ++i) + { + exec::start_detached( + ex::schedule(scheduler) + | ex::then([producer_completed] + { producer_completed->fetch_add(1, std::memory_order_relaxed); })); + } + + while (!stop.load(std::memory_order_relaxed) + && producer_completed->load(std::memory_order_relaxed) < expected) + { + std::this_thread::yield(); + } std::this_thread::yield(); } - std::this_thread::yield(); - } - }); - } + }); + } - ready.wait(); - start.store(true, std::memory_order_release); + ready.wait(); + start.store(true, std::memory_order_release); - auto const expected = num_producers * rounds + num_producers * ((rounds + 3) / 4); - auto const deadline = std::chrono::steady_clock::now() + std::chrono::seconds(10); - auto completed_total = [&] - { - std::size_t result = 0; - for (auto const & count: completed) + auto const expected = num_producers * rounds + num_producers * ((rounds + 3) / 4); + auto const deadline = std::chrono::steady_clock::now() + std::chrono::seconds(10); + auto completed_total = [&] { - result += count.load(std::memory_order_relaxed); + std::size_t result = 0; + for (auto const & count: completed) + { + result += count.load(std::memory_order_relaxed); + } + return result; + }; + + while (completed_total() < expected && std::chrono::steady_clock::now() < deadline) + { + std::this_thread::yield(); + } + stop.store(true, std::memory_order_release); + for (auto& producer: producers) + { + producer.join(); } - return result; - }; - while (completed_total() < expected && std::chrono::steady_clock::now() < deadline) - { - std::this_thread::yield(); - } - stop.store(true, std::memory_order_release); - for (auto& producer: producers) - { - producer.join(); + CHECK(completed_total() == expected); } +} // namespace + +TEST_CASE("static_thread_pool drains remote work from a shared scheduler", + "[types][static_thread_pool][stress]") +{ + run_remote_poll_stress(false); +} - CHECK(completed_total() == expected); +TEST_CASE("static_thread_pool drains remote work from producer schedulers", + "[types][static_thread_pool][stress]") +{ + run_remote_poll_stress(true); } diff --git a/test/rrd/static_thread_pool_remote_poll.cpp b/test/rrd/static_thread_pool_remote_poll.cpp index 43d67dd1d..1faca3213 100644 --- a/test/rrd/static_thread_pool_remote_poll.cpp +++ b/test/rrd/static_thread_pool_remote_poll.cpp @@ -2,8 +2,8 @@ * Copyright (c) 2026 NVIDIA Corporation * * Licensed under the Apache License Version 2.0 with LLVM Exceptions - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at + * (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at * * https://llvm.org/LICENSE.txt * @@ -18,9 +18,15 @@ #include -struct remote_queue_node +struct task_node { - remote_queue_node* next_ = nullptr; + task_node* next_ = nullptr; +}; + +struct remote_queue +{ + remote_queue* next_ = nullptr; + std::atomic head_{nullptr}; }; struct static_thread_pool_remote_poll : rl::test_suite @@ -29,31 +35,68 @@ struct static_thread_pool_remote_poll : rl::test_suite state_{running}; - std::atomic head_{nullptr}; - std::atomic first_notification_published_{false}; - remote_queue_node first_{}; - remote_queue_node second_{}; - bool worker_would_sleep_ = false; + enum class remote_poll_mode + { + speculative, + before_sleep + }; + + struct poll_result + { + bool any = false; + bool second = false; + }; + + std::atomic state_{running}; + std::atomic remote_head_{nullptr}; + std::atomic first_notification_published_{false}; + remote_queue first_queue_{}; + remote_queue second_queue_{}; + task_node first_task_{}; + task_node first_extra_task_{}; + task_node second_task_{}; + bool worker_would_sleep_ = false; void before() { state_.store(running, std::memory_order_relaxed); - head_.store(nullptr, std::memory_order_relaxed); + remote_head_.store(nullptr, std::memory_order_relaxed); first_notification_published_.store(false, std::memory_order_relaxed); - first_.next_ = nullptr; - second_.next_ = nullptr; - worker_would_sleep_ = false; + first_queue_.next_ = nullptr; + second_queue_.next_ = nullptr; + first_queue_.head_.store(nullptr, std::memory_order_relaxed); + second_queue_.head_.store(nullptr, std::memory_order_relaxed); + first_task_.next_ = nullptr; + first_extra_task_.next_ = nullptr; + second_task_.next_ = nullptr; + worker_would_sleep_ = false; + } + + void publish_remote_queue(remote_queue& queue) + { + auto* old_head = remote_head_.load(std::memory_order_acquire); + do + { + queue.next_ = old_head; + } + while (!remote_head_.compare_exchange_weak(old_head, + &queue, + std::memory_order_acq_rel, + std::memory_order_acquire)); } - void publish(remote_queue_node& node) + auto push(remote_queue& queue, task_node& task) -> bool { - auto* old_head = head_.load(std::memory_order_relaxed); + auto* old_head = queue.head_.load(std::memory_order_relaxed); do { - node.next_ = old_head; + task.next_ = old_head; } - while (!head_.compare_exchange_weak(old_head, &node, std::memory_order_acq_rel)); + while (!queue.head_.compare_exchange_weak(old_head, + &task, + std::memory_order_acq_rel, + std::memory_order_acquire)); + return old_head == nullptr; } void notify() @@ -61,24 +104,62 @@ struct static_thread_pool_remote_poll : rl::test_suite bool + void enqueue(remote_queue& queue, task_node& task) + { + bool const was_empty = push(queue, task); + if (was_empty) + { + notify(); + } + } + + auto drain(remote_queue& queue) -> task_node* + { + auto* old_head = queue.head_.load(std::memory_order_relaxed); + while (!queue.head_.compare_exchange_weak(old_head, + nullptr, + std::memory_order_acq_rel, + std::memory_order_acquire)) + { + } + return old_head; + } + + auto poll_remote(remote_poll_mode mode) -> poll_result { - for (auto* node = head_.load(std::memory_order_acquire); node != nullptr; node = node->next_) + poll_result result{}; + auto* queue = remote_head_.load(std::memory_order_acquire); + while (queue != nullptr) { - if (node == &second_) + if (mode == remote_poll_mode::before_sleep + || queue->head_.load(std::memory_order_relaxed) != nullptr) { - return true; + for (auto* task = drain(*queue); task != nullptr; task = task->next_) + { + result.any = true; + result.second = result.second || task == &second_task_; + } } + queue = queue->next_; } - return false; + return result; } void worker_poll() { - if (sees_second_node()) + auto result = poll_remote(remote_poll_mode::speculative); + if (result.second) { return; } + if (result.any) + { + result = poll_remote(remote_poll_mode::speculative); + if (result.second) + { + return; + } + } int expected = running; if (!state_.compare_exchange_weak(expected, @@ -86,13 +167,20 @@ struct static_thread_pool_remote_poll : rl::test_suite Date: Sat, 15 Aug 2026 03:00:49 +0000 Subject: [PATCH 8/8] clang-format --- include/exec/static_thread_pool.hpp | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/include/exec/static_thread_pool.hpp b/include/exec/static_thread_pool.hpp index a8839bf0c..1a1444d8f 100644 --- a/include/exec/static_thread_pool.hpp +++ b/include/exec/static_thread_pool.hpp @@ -171,8 +171,7 @@ namespace experimental::execution } } - auto - pop_all_reversed(std::size_t tid, remote_poll_mode mode) noexcept + 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); @@ -986,8 +985,7 @@ namespace experimental::execution -> _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); + __intrusive_queue<&task_base::next_> remotes = pool_->remotes_.pop_all_reversed(index_, mode); pending_queue_.append(std::move(remotes)); if (!pending_queue_.empty()) {