diff --git a/README.md b/README.md index b4d4b596..16649168 100644 --- a/README.md +++ b/README.md @@ -29,7 +29,7 @@ This fork addresses critical bugs, adds utilities for high-throughput and async - **Completion transitions** (`completion`) -- post-event transitions with origin event propagation - **Coroutine state machines** (`co_sm`) -- C++20 coroutine-driven SM with configurable scheduler and allocator policies -- **`thread_pool_scheduler`** -- opt-in C++20 multi-consumer work pool with a lifetime-safe, deadlock-free fork/join latch for action-level inter-op parallelism (warm workers, fixed task ring, no allocation on dispatch) +- **`thread_pool_scheduler`** -- opt-in C++20 multi-consumer work pool for nonblocking coroutine dispatch plus explicit blocking fork/join APIs (warm workers, fixed task ring, no allocation on dispatch) - **`sm_pool`** -- pool-based container for managing thousands of SM instances with indexed and batch dispatch - **`dispatch_table`** -- compile-time dispatch table for ID-based event routing at zero runtime overhead @@ -126,16 +126,19 @@ cl /std:c++14 /Ox /W3 tcp_release.cpp ## Utilities + + ### co_sm (C++20) Coroutine-driven state machine wrapper with configurable scheduling and allocation policies. Requires C++20 with coroutine support. ```cpp #include +#include namespace utility = sml::utility; -// Default: inline scheduler, pooled coroutine allocator +// Default: FIFO scheduler, pooled coroutine allocator utility::co_sm sm{}; // Synchronous dispatch (same as regular sm) @@ -143,17 +146,32 @@ sm.process_event(my_event{}); // Asynchronous dispatch via coroutine auto task = sm.process_event_async(my_event{}); -bool accepted = task.result(); + +// result() is nonblocking: call it only after completion, or co_await the task. +if (task.await_ready()) { + bool accepted = task.result(); +} + +// Thread-pool dispatch starts work on a worker and lets the caller await later. +using pool_scheduler = utility::policy::thread_pool_scheduler<4>; +utility::co_sm> pooled{}; +auto pooled_task = pooled.process_event_async(my_event{}); +// ...do local work... +while (!pooled_task.await_ready()) {} +bool pooled_accepted = pooled_task.result(); ``` **Scheduler policies:** | Policy | Description | |--------|-------------| -| `inline_scheduler` | Runs tasks immediately (default) | -| `fifo_scheduler` | FIFO queue with bounded inline storage | +| `inline_scheduler` | Runs tasks immediately | +| `fifo_scheduler` | FIFO queue with bounded inline storage (default) | +| `thread_pool_scheduler` | Starts `process_event_async` on worker threads and returns a task to await later; concurrent same-actor overlap is rejected | | `coroutine_scheduler` | Wraps any scheduler for the coroutine path | +`thread_pool_scheduler::schedule(fn)` is nonblocking. Use `run_or_schedule_and_wait(fn)` or `thread_pool_scheduler::join_group` at RTC call sites that need an immediate join. Destroying an incomplete `bool_task` is a hard contract violation and terminates; keep the task alive until it is ready or awaited. + **Allocator policies:** | Policy | Description | diff --git a/benchmark/simple/CMakeLists.txt b/benchmark/simple/CMakeLists.txt index 48b5d91f..f7d6d8ca 100644 --- a/benchmark/simple/CMakeLists.txt +++ b/benchmark/simple/CMakeLists.txt @@ -13,6 +13,14 @@ add_example(simple_sc benchmark_simple_sc sc.cpp) if (NOT IS_MSVC_2015) add_example(simple_co_sm benchmark_simple_co_sm co_sm.cpp) + add_example(simple_thread_pool_co_sm benchmark_simple_thread_pool_co_sm thread_pool_co_sm.cpp) + find_package(Threads REQUIRED) + target_link_libraries(simple_thread_pool_co_sm PRIVATE Threads::Threads) + if (IS_COMPILER_OPTION_MSVC_LIKE) + target_compile_options(simple_thread_pool_co_sm PRIVATE /UCHECK_COMPILE_TIME) + else() + target_compile_options(simple_thread_pool_co_sm PRIVATE -UCHECK_COMPILE_TIME) + endif() if (EXISTS "${CMAKE_CURRENT_SOURCE_DIR}/sml_player_sm.hpp") add_example(simple_sml benchmark_simple_sml sml.cpp) diff --git a/benchmark/simple/thread_pool_co_sm.cpp b/benchmark/simple/thread_pool_co_sm.cpp new file mode 100644 index 00000000..4b0b350b --- /dev/null +++ b/benchmark/simple/thread_pool_co_sm.cpp @@ -0,0 +1,140 @@ +// +// Copyright (c) 2026 stateforward +// +// Distributed under the Boost Software License, Version 1.0. +// (See accompanying file LICENSE_1_0.txt or copy at +// http://www.boost.org/LICENSE_1_0.txt) +// +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace sml = boost::sml; + +struct tick { + std::atomic* calls = nullptr; +}; + +struct noop_child { + auto operator()() const { + using namespace sml; + const auto record = [](const tick& event) noexcept { event.calls->fetch_add(1u, std::memory_order_relaxed); }; + // clang-format off + return make_transition_table( + *"ready"_s + event / record = "ready"_s + ); + // clang-format on + } +}; + +struct sleep_child { + auto operator()() const { + using namespace sml; + const auto work = [](const tick& event) noexcept { + std::this_thread::sleep_for(std::chrono::microseconds(10)); + event.calls->fetch_add(1u, std::memory_order_relaxed); + }; + // clang-format off + return make_transition_table( + *"ready"_s + event / work = "ready"_s + ); + // clang-format on + } +}; + +template +std::chrono::nanoseconds measure(fn&& fn_in) noexcept { + const auto start = std::chrono::high_resolution_clock::now(); + fn_in(); + const auto finish = std::chrono::high_resolution_clock::now(); + return std::chrono::duration_cast(finish - start); +} + +void print_duration(const std::chrono::nanoseconds elapsed, const std::size_t dispatches) { + const auto total_ms = std::chrono::duration_cast(elapsed).count(); + const auto ns_per_dispatch = dispatches == 0u ? 0u : static_cast(elapsed.count()) / dispatches; + std::cout << "execution speed: " << total_ms << "ms (" << ns_per_dispatch << "ns/dispatch)" << std::endl; +} + +#if BOOST_SML_UTILITY_CO_SM_ENABLED && BOOST_SML_UTILITY_THREAD_POOL_SCHEDULER_ENABLED +template +void run_case(const char* label) { + using scheduler_t = sml::utility::policy::thread_pool_scheduler<1, 1024, 128>; + using child_sm_t = sml::utility::co_sm< + machine, sml::utility::policy::coroutine_scheduler, + sml::utility::policy::coroutine_allocator>>; + + constexpr std::size_t dispatches = children_count * rounds; + + std::array, children_count> blocking_calls{}; + std::array blocking_children{}; + std::cout << label << " blocking run_or_schedule_and_wait: "; + print_duration( + measure([&] { + for (std::size_t round = 0; round < rounds; ++round) { + for (std::size_t i = 0u; i < children_count; ++i) { + auto& child_sm = blocking_children[i]; + const bool submitted = + child_sm.scheduler().run_or_schedule_and_wait([&child_sm, &blocking_calls, i]() noexcept { + (void)child_sm.process_event(tick{&blocking_calls[i]}); + }); + if (!submitted) { + std::terminate(); + } + } + } + }), + dispatches); + std::uint64_t blocking_total = 0u; + for (const auto& calls : blocking_calls) { + blocking_total += calls.load(std::memory_order_relaxed); + } + if (blocking_total != dispatches) { + std::terminate(); + } + + std::array, children_count> async_calls{}; + std::array async_children{}; + std::cout << label << " start-all await-later: "; + print_duration( + measure([&] { + std::array tasks{}; + for (std::size_t round = 0; round < rounds; ++round) { + for (std::size_t i = 0u; i < children_count; ++i) { + tasks[i] = async_children[i].process_event_async(tick{&async_calls[i]}); + } + for (auto& task : tasks) { + while (!task.await_ready()) { + sml::utility::policy::cpu_relax(); + } + if (!task.result()) { + std::terminate(); + } + } + } + }), + dispatches); + std::uint64_t async_total = 0u; + for (const auto& calls : async_calls) { + async_total += calls.load(std::memory_order_relaxed); + } + if (async_total != dispatches) { + std::terminate(); + } +} +#endif + +int main() { +#if BOOST_SML_UTILITY_CO_SM_ENABLED && BOOST_SML_UTILITY_THREAD_POOL_SCHEDULER_ENABLED + constexpr std::size_t children_count = 8u; + run_case("noop"); + run_case("10us work"); +#endif +} diff --git a/include/boost/sml/utility/co_sm.hpp b/include/boost/sml/utility/co_sm.hpp index cbe6c780..95d212fb 100644 --- a/include/boost/sml/utility/co_sm.hpp +++ b/include/boost/sml/utility/co_sm.hpp @@ -24,6 +24,7 @@ #if BOOST_SML_UTILITY_CO_SM_ENABLED #include +#include #include #include #include @@ -210,8 +211,15 @@ class pooled_coroutine_allocator { static_assert(SlotCount > 0, "pooled_coroutine_allocator slot count must be non-zero"); pooled_coroutine_allocator() noexcept { reset_freelist(); } + pooled_coroutine_allocator(const pooled_coroutine_allocator&) noexcept { reset_freelist(); } + pooled_coroutine_allocator& operator=(const pooled_coroutine_allocator&) noexcept { + lock_scope lock{lock_}; + reset_freelist(); + return *this; + } void* allocate(const std::size_t size, const std::size_t alignment) { + lock_scope lock{lock_}; if (size <= SlotSize && alignment <= alignof(pool_slot) && free_head_ != invalid_index) { const std::size_t slot_index = free_head_; free_head_ = next_free_[slot_index]; @@ -226,6 +234,7 @@ class pooled_coroutine_allocator { return; } + lock_scope lock{lock_}; if (size <= SlotSize && alignment <= alignof(pool_slot) && is_pool_pointer(ptr)) { const std::size_t slot_index = slot_index_for(ptr); next_free_[slot_index] = free_head_; @@ -244,6 +253,16 @@ class pooled_coroutine_allocator { alignas(std::max_align_t) std::array storage{}; }; + struct lock_scope { + explicit lock_scope(std::atomic_flag& flag) noexcept : flag(flag) { + while (flag.test_and_set(std::memory_order_acquire)) { + } + } + ~lock_scope() noexcept { flag.clear(std::memory_order_release); } + + std::atomic_flag& flag; + }; + bool is_pool_pointer(void* ptr) const noexcept { const auto* begin = reinterpret_cast(slots_.data()); const auto* end = begin + sizeof(slots_); @@ -274,6 +293,7 @@ class pooled_coroutine_allocator { std::array slots_{}; std::array next_free_{}; std::size_t free_head_ = 0; + mutable std::atomic_flag lock_ = ATOMIC_FLAG_INIT; }; template @@ -296,6 +316,16 @@ concept strict_ordering_scheduler_contract = } && static_cast(TScheduler::guarantees_fifo) && static_cast(TScheduler::single_consumer) && static_cast(TScheduler::run_to_completion); +template +concept async_coroutine_scheduler_contract = requires { + { TScheduler::multi_consumer } -> std::convertible_to; + { TScheduler::run_to_completion } -> std::convertible_to; +} && static_cast(TScheduler::multi_consumer) && !static_cast(TScheduler::run_to_completion); + +template +concept coroutine_scheduler_contract = + strict_ordering_scheduler_contract || async_coroutine_scheduler_contract; + template concept has_try_run_immediate = requires(TScheduler scheduler) { { @@ -340,6 +370,7 @@ class bool_task { struct promise_type { using allocate_fn = void* (*)(void*, std::size_t, std::size_t); using deallocate_fn = void (*)(void*, void*, std::size_t, std::size_t) noexcept; + using completion_fn = void (*)(void*) noexcept; struct frame_header { void* allocator_ctx = nullptr; @@ -436,9 +467,26 @@ class bool_task { deallocate_frame(frame_ptr); } + static constexpr unsigned running = 0u; + static constexpr unsigned awaiting = 1u; + static constexpr unsigned completing = 2u; + static constexpr unsigned completed = 3u; + + promise_type() = default; + + template + promise_type(std::allocator_arg_t, TAllocator&, TOwner& owner, TArgs&&...) noexcept + : completion_ctx(&owner), + completion(+[](void* ctx) noexcept { static_cast(ctx)->release_async_dispatch_active(); }) {} + bool value = false; +#if !BOOST_SML_DISABLE_EXCEPTIONS std::exception_ptr exception = nullptr; +#endif std::coroutine_handle<> continuation = std::noop_coroutine(); + std::atomic state = running; + void* completion_ctx = nullptr; + completion_fn completion = nullptr; bool_task get_return_object() noexcept { return bool_task{handle_type::from_promise(*this)}; } std::suspend_never initial_suspend() noexcept { return {}; } @@ -446,14 +494,33 @@ class bool_task { auto final_suspend() noexcept { struct continuation_awaiter { bool await_ready() const noexcept { return false; } - std::coroutine_handle<> await_suspend(handle_type h) const noexcept { return h.promise().continuation; } + std::coroutine_handle<> await_suspend(handle_type h) const noexcept { + auto& promise = h.promise(); + const unsigned previous = promise.state.exchange(completing, std::memory_order_acq_rel); + auto continuation = previous == awaiting ? promise.continuation : std::noop_coroutine(); + auto completion = promise.completion; + void* completion_ctx = promise.completion_ctx; + promise.state.store(completed, std::memory_order_release); + if (completion != nullptr) { + completion(completion_ctx); + } + return continuation; + } void await_resume() const noexcept {} }; return continuation_awaiter{}; } void return_value(const bool v) noexcept { value = v; } - void unhandled_exception() noexcept { exception = std::current_exception(); } + void unhandled_exception() noexcept { +#if !BOOST_SML_DISABLE_EXCEPTIONS + exception = std::current_exception(); +#else + std::terminate(); +#endif + } + + bool is_complete() const noexcept { return state.load(std::memory_order_acquire) == completed; } }; bool_task() = default; @@ -466,11 +533,7 @@ class bool_task { return task; } - ~bool_task() { - if (handle_) { - handle_.destroy(); - } - } + ~bool_task() { destroy_if_complete(); } bool_task(const bool_task&) = delete; bool_task& operator=(const bool_task&) = delete; @@ -488,9 +551,7 @@ class bool_task { return *this; } - if (handle_) { - handle_.destroy(); - } + destroy_if_complete(); handle_ = std::exchange(other.handle_, {}); has_immediate_value_ = other.has_immediate_value_; immediate_value_ = other.immediate_value_; @@ -503,15 +564,29 @@ class bool_task { if (has_immediate_value_) { return true; } - return (handle_ == nullptr) || handle_.done(); + return (handle_ == nullptr) || handle_.promise().is_complete(); } std::coroutine_handle<> await_suspend(std::coroutine_handle<> awaiting) noexcept { if (handle_ == nullptr) { return std::noop_coroutine(); } - handle_.promise().continuation = awaiting; - return handle_; + auto& promise = handle_.promise(); + promise.continuation = awaiting; + const unsigned previous = promise.state.exchange(promise_type::awaiting, std::memory_order_acq_rel); + if (previous == promise_type::completed) { + promise.state.store(promise_type::completed, std::memory_order_release); + return awaiting; + } + if (previous == promise_type::completing) { + while (!promise.is_complete()) { + } + return awaiting; + } + if (previous == promise_type::awaiting) { + std::terminate(); + } + return std::noop_coroutine(); } bool await_resume() { return result(); } @@ -523,24 +598,32 @@ class bool_task { if (handle_ == nullptr) { return false; } - if (!handle_.done()) { + if (!handle_.promise().is_complete()) { #if !BOOST_SML_DISABLE_EXCEPTIONS throw std::logic_error("bool_task result() called before coroutine completion"); #else std::terminate(); #endif } - if (handle_.promise().exception) { #if !BOOST_SML_DISABLE_EXCEPTIONS + if (handle_.promise().exception) { std::rethrow_exception(handle_.promise().exception); -#else - std::terminate(); -#endif } +#endif return handle_.promise().value; } private: + void destroy_if_complete() noexcept { + if (handle_ == nullptr) { + return; + } + if (!handle_.promise().is_complete()) { + std::terminate(); + } + handle_.destroy(); + } + handle_type handle_{}; bool has_immediate_value_ = false; bool immediate_value_ = false; @@ -561,8 +644,8 @@ class co_sm { using state_machine_type = sm; static_assert(policy::valid_coroutine_scheduler, "scheduler_type must provide schedule(fn)"); - static_assert(policy::strict_ordering_scheduler_contract, - "scheduler_type must guarantee FIFO ordering, single-consumer dispatch, and run-to-completion"); + static_assert(policy::coroutine_scheduler_contract, + "scheduler_type must be strict run-to-completion or an async coroutine scheduler"); static_assert(policy::valid_coroutine_allocator, "allocator_type must provide allocate(size, alignment) and noexcept deallocate(ptr, size, alignment)"); @@ -641,6 +724,25 @@ class co_sm { accepted = state_machine_.process_event(completion_event_t{index}) && accepted; } return accepted; + } else if constexpr (policy::async_coroutine_scheduler_contract) { + if (!async_dispatch_active_.try_acquire()) { + if (active_async_dispatch_owner_ == this) { + return state_machine_.process_event(event); + } + return false; + } + struct active_reset { + async_dispatch_flag& active; + ~active_reset() noexcept { active.release(); } + } reset{async_dispatch_active_}; + struct active_owner_scope { + const co_sm* previous; + explicit active_owner_scope(const co_sm* current) noexcept : previous(active_async_dispatch_owner_) { + active_async_dispatch_owner_ = current; + } + ~active_owner_scope() noexcept { active_async_dispatch_owner_ = previous; } + } owner{this}; + return state_machine_.process_event(event); } else { return state_machine_.process_event(event); } @@ -648,14 +750,34 @@ class co_sm { template bool_task process_event_async(TEvent&& event) { - using event_t = typename std::decay::type; - event_t event_copy(static_cast(event)); - if constexpr (policy::external_completion_scheduler_contract) { + using event_t = typename std::decay::type; + event_t event_copy(static_cast(event)); return bool_task::from_value(process_event(event_copy)); } else if constexpr (std::is_same_v) { + using event_t = typename std::decay::type; + event_t event_copy(static_cast(event)); return bool_task::from_value(state_machine_.process_event(event_copy)); + } else if constexpr (policy::async_coroutine_scheduler_contract) { + if (!async_dispatch_active_.try_acquire()) { + return bool_task::from_value(false); + } + using event_t = typename std::decay::type; +#if !BOOST_SML_DISABLE_EXCEPTIONS + try { + event_t event_copy(static_cast(event)); + return process_event_async_impl(std::allocator_arg, allocator_, *this, static_cast(event_copy)); + } catch (...) { + async_dispatch_active_.release(); + throw; + } +#else + event_t event_copy(static_cast(event)); + return process_event_async_impl(std::allocator_arg, allocator_, *this, static_cast(event_copy)); +#endif } else { + using event_t = typename std::decay::type; + event_t event_copy(static_cast(event)); if constexpr (policy::has_try_run_immediate) { bool accepted = false; if (scheduler_.try_run_immediate( @@ -706,7 +828,9 @@ class co_sm { co_sm& self; TEvent event_value; bool accepted = false; +#if !BOOST_SML_DISABLE_EXCEPTIONS std::exception_ptr exception{}; +#endif bool await_ready() noexcept { if constexpr (std::is_same_v) { @@ -718,15 +842,25 @@ class co_sm { void await_suspend(std::coroutine_handle<> handle) { self.scheduler_.schedule([this, handle]() mutable { + struct active_owner_scope { + const co_sm* previous; + explicit active_owner_scope(const co_sm* current) noexcept : previous(active_async_dispatch_owner_) { + active_async_dispatch_owner_ = current; + } + ~active_owner_scope() noexcept { active_async_dispatch_owner_ = previous; } + }; + { + active_owner_scope owner{&self}; #if !BOOST_SML_DISABLE_EXCEPTIONS - try { - accepted = self.state_machine_.process_event(event_value); - } catch (...) { - exception = std::current_exception(); - } + try { + accepted = self.state_machine_.process_event(event_value); + } catch (...) { + exception = std::current_exception(); + } #else - accepted = self.state_machine_.process_event(event_value); + accepted = self.state_machine_.process_event(event_value); #endif + } handle.resume(); }); } @@ -744,10 +878,34 @@ class co_sm { state_machine_type state_machine_{}; scheduler_type scheduler_{}; allocator_type allocator_{}; + + void release_async_dispatch_active() noexcept { + if constexpr (policy::async_coroutine_scheduler_contract) { + async_dispatch_active_.release(); + } + } + + friend struct bool_task::promise_type; + + struct async_dispatch_flag { + async_dispatch_flag() = default; + async_dispatch_flag(const async_dispatch_flag&) noexcept {} + async_dispatch_flag& operator=(const async_dispatch_flag&) noexcept { + active.store(false, std::memory_order_relaxed); + return *this; + } + + bool try_acquire() noexcept { return !active.exchange(true, std::memory_order_acquire); } + void release() noexcept { active.store(false, std::memory_order_release); } + + std::atomic active{false}; + }; // Depth of synchronous external-completion dispatches on this machine; // nested calls (actions re-entering process_event) skip reset/sweep/drain // so the outermost dispatch's run-to-completion drain stays intact. std::size_t external_dispatch_depth_ = 0; + async_dispatch_flag async_dispatch_active_{}; + inline static thread_local const co_sm* active_async_dispatch_owner_ = nullptr; }; } // namespace utility diff --git a/include/boost/sml/utility/thread_pool_scheduler.hpp b/include/boost/sml/utility/thread_pool_scheduler.hpp index 6aea85e5..7e197762 100644 --- a/include/boost/sml/utility/thread_pool_scheduler.hpp +++ b/include/boost/sml/utility/thread_pool_scheduler.hpp @@ -101,6 +101,60 @@ class thread_pool_scheduler { thread_pool_scheduler(thread_pool_scheduler&&) = delete; thread_pool_scheduler& operator=(thread_pool_scheduler&&) = delete; + class join_group { + public: + join_group() = default; + ~join_group() = default; + + join_group(const join_group&) = delete; + join_group& operator=(const join_group&) = delete; + join_group(join_group&&) = delete; + join_group& operator=(join_group&&) = delete; + + bool wait() noexcept { + // Spin-join on pending_ rather than blocking on a per-group semaphore. + // The group is caller-owned and typically stack-reused across fork/joins, + // so a notify-based wakeup is unsafe: the waiter could observe completion, + // return, and destroy the group before the last completer finishes its + // release()/notify, faulting on freed semaphore state. With a plain spin, + // a completer's final touch of the group is its pending_ decrement, and + // wait() returns only after observing pending_ == 0 (all decrements done), + // so nothing accesses the group after the caller may destroy it. The wait + // is a bounded RTC fork/join over already-submitted lanes, so the producer + // core would otherwise be idle; spinning gives the lowest join latency. + while (pending_.load(std::memory_order_acquire) != 0u) { + cpu_relax(); + } + // pending_ == 0 means every completer is done touching the group, so the + // owning thread can now read and clear accepted_ with no contention. The + // clear lets a stack-reused join_group start the next round clean: without + // it, a single rejected lane would make wait() return false on every later + // round even when all lanes succeed. + const bool accepted = accepted_.load(std::memory_order_acquire); + accepted_.store(true, std::memory_order_release); + return accepted; + } + + private: + friend class thread_pool_scheduler; + + void start_one() noexcept { pending_.fetch_add(1u, std::memory_order_acq_rel); } + + void reject_one() noexcept { + accepted_.store(false, std::memory_order_release); + complete_one(); + } + + void reject() noexcept { accepted_.store(false, std::memory_order_release); } + + void complete_one() noexcept { pending_.fetch_sub(1u, std::memory_order_acq_rel); } + + static void complete_one(void* ctx) noexcept { static_cast(ctx)->complete_one(); } + + std::atomic pending_ = 0u; + std::atomic accepted_ = true; + }; + template bool try_run_immediate(fn&& fn_in) noexcept(noexcept(std::forward(fn_in)())) { if (queued_or_running_.load(std::memory_order_acquire) != 0u) { @@ -122,7 +176,6 @@ class thread_pool_scheduler { } std::forward(fn_in)(); - immediate_run_count_.fetch_add(1u, std::memory_order_relaxed); return true; } @@ -131,16 +184,37 @@ class thread_pool_scheduler { return try_submit_with_completion(std::forward(fn_in), nullptr, nullptr); } - // Detached hard-contract wrapper for call sites that have already proven - // scheduler lifetime and queue capacity. Actor-facing RTC paths must use - // thread_pool_scheduler_ref::schedule or run_or_schedule_and_wait. template - void submit(fn&& fn_in) noexcept { + bool try_submit(join_group& group, fn&& fn_in) noexcept { + if (is_current_thread_worker()) { + group.reject(); + return false; + } + + group.start_one(); + const bool submitted = try_submit_with_completion(std::forward(fn_in), &group, join_group::complete_one); + if (!submitted) { + group.reject_one(); + return false; + } + return true; + } + + template + void schedule(fn&& fn_in) noexcept { if (!try_submit(std::forward(fn_in))) { std::terminate(); } } + // Detached hard-contract wrapper for call sites that have already proven + // scheduler lifetime and queue capacity. RTC call sites use + // run_or_schedule_and_wait or thread_pool_scheduler::join_group. + template + void submit(fn&& fn_in) noexcept { + schedule(std::forward(fn_in)); + } + // Unconditionally noexcept (like fifo_scheduler::schedule): when this falls to // the scheduled path the task runs through a noexcept worker thunk, which // cannot propagate an exception back to the waiter, so the API is consistently @@ -170,12 +244,6 @@ class thread_pool_scheduler { return true; } - std::uint64_t immediate_run_count() const noexcept { return immediate_run_count_.load(std::memory_order_relaxed); } - - std::uint64_t scheduled_run_count() const noexcept { return scheduled_run_count_.load(std::memory_order_relaxed); } - - std::uint64_t worker_run_count() const noexcept { return worker_run_count_.load(std::memory_order_relaxed); } - bool is_current_thread_worker() const noexcept { return running_on_this_worker(); } template @@ -190,8 +258,6 @@ class thread_pool_scheduler { queued_or_running_.fetch_sub(1u, std::memory_order_acq_rel); return false; } - - scheduled_run_count_.fetch_add(1u, std::memory_order_relaxed); ready_.release(); return true; } @@ -344,7 +410,6 @@ class thread_pool_scheduler { void (*completion_fn)(void*) noexcept = slot->completion_fn; slot->completion_ctx = nullptr; slot->completion_fn = nullptr; - worker_run_count_.fetch_add(1u, std::memory_order_relaxed); queued_or_running_.fetch_sub(1u, std::memory_order_acq_rel); slot->sequence.store(pos + capacity, std::memory_order_release); if (completion_fn != nullptr) { @@ -419,123 +484,9 @@ class thread_pool_scheduler { std::atomic queued_or_running_ = 0u; std::atomic inline_active_ = false; std::atomic stopping_ = false; - std::atomic immediate_run_count_ = 0u; - std::atomic scheduled_run_count_ = 0u; - std::atomic worker_run_count_ = 0u; inline static thread_local const thread_pool_scheduler* active_worker_scheduler_ = nullptr; }; -template -class thread_pool_scheduler_ref { - public: - static constexpr bool guarantees_fifo = scheduler::guarantees_fifo; - static constexpr bool single_consumer = scheduler::single_consumer; - static constexpr bool multi_consumer = scheduler::multi_consumer; - static constexpr bool owns_workers = false; - static constexpr bool run_to_completion = true; - static constexpr std::size_t static_worker_count = scheduler::static_worker_count; - static constexpr std::size_t static_capacity = scheduler::static_capacity; - - thread_pool_scheduler_ref() = delete; - explicit thread_pool_scheduler_ref(scheduler& scheduler_in) noexcept : scheduler_(&scheduler_in) {} - - class join_group { - public: - join_group() = default; - ~join_group() = default; - - join_group(const join_group&) = delete; - join_group& operator=(const join_group&) = delete; - join_group(join_group&&) = delete; - join_group& operator=(join_group&&) = delete; - - bool wait() noexcept { - // Spin-join on pending_ rather than blocking on a per-group semaphore. - // The group is caller-owned and typically stack-reused across fork/joins, - // so a notify-based wakeup is unsafe: the waiter could observe completion, - // return, and destroy the group before the last completer finishes its - // release()/notify, faulting on freed semaphore state. With a plain spin, - // a completer's final touch of the group is its pending_ decrement, and - // wait() returns only after observing pending_ == 0 (all decrements done), - // so nothing accesses the group after the caller may destroy it. The wait - // is a bounded RTC fork/join over already-submitted lanes, so the producer - // core would otherwise be idle; spinning gives the lowest join latency. - while (pending_.load(std::memory_order_acquire) != 0u) { - cpu_relax(); - } - // pending_ == 0 means every completer is done touching the group, so the - // owning thread can now read and clear accepted_ with no contention. The - // clear lets a stack-reused join_group start the next round clean: without - // it, a single rejected lane would make wait() return false on every later - // round even when all lanes succeed. - const bool accepted = accepted_.load(std::memory_order_acquire); - accepted_.store(true, std::memory_order_release); - return accepted; - } - - private: - friend class thread_pool_scheduler_ref; - - void start_one() noexcept { pending_.fetch_add(1u, std::memory_order_acq_rel); } - - void reject_one() noexcept { - accepted_.store(false, std::memory_order_release); - complete_one(); - } - - void reject() noexcept { accepted_.store(false, std::memory_order_release); } - - void complete_one() noexcept { pending_.fetch_sub(1u, std::memory_order_acq_rel); } - - static void complete_one(void* ctx) noexcept { static_cast(ctx)->complete_one(); } - - std::atomic pending_ = 0u; - std::atomic accepted_ = true; - }; - - template - bool try_run_immediate(fn&& fn_in) noexcept(noexcept(std::forward(fn_in)())) { - return scheduler_->try_run_immediate(std::forward(fn_in)); - } - - template - bool try_submit(join_group& group, fn&& fn_in) noexcept { - if (scheduler_->is_current_thread_worker()) { - group.reject(); - return false; - } - - group.start_one(); - const bool submitted = scheduler_->try_submit_with_completion(std::forward(fn_in), &group, join_group::complete_one); - if (!submitted) { - group.reject_one(); - return false; - } - return true; - } - - template - void schedule(fn&& fn_in) noexcept { - if (!scheduler_->run_or_schedule_and_wait(std::forward(fn_in))) { - std::terminate(); - } - } - - template - bool run_or_schedule_and_wait(fn&& fn_in) noexcept { - return scheduler_->run_or_schedule_and_wait(std::forward(fn_in)); - } - - std::uint64_t immediate_run_count() const noexcept { return scheduler_->immediate_run_count(); } - - std::uint64_t scheduled_run_count() const noexcept { return scheduler_->scheduled_run_count(); } - - std::uint64_t worker_run_count() const noexcept { return scheduler_->worker_run_count(); } - - private: - scheduler* scheduler_ = nullptr; -}; - } // namespace policy } // namespace utility diff --git a/test/ft/CMakeLists.txt b/test/ft/CMakeLists.txt index be82b57c..a6d5eeb8 100644 --- a/test/ft/CMakeLists.txt +++ b/test/ft/CMakeLists.txt @@ -23,6 +23,11 @@ add_executable(test_thread_pool_scheduler thread_pool_scheduler.cpp) add_test(test_thread_pool_scheduler test_thread_pool_scheduler) target_link_libraries(test_thread_pool_scheduler -lpthread) +add_executable(test_co_sm_thread_pool co_sm_thread_pool.cpp) +add_test(test_co_sm_thread_pool test_co_sm_thread_pool) +find_package(Threads REQUIRED) +target_link_libraries(test_co_sm_thread_pool PRIVATE Threads::Threads) + add_executable(test_external_completion external_completion.cpp) add_test(test_external_completion test_external_completion) target_link_libraries(test_external_completion -lpthread) diff --git a/test/ft/co_sm.cpp b/test/ft/co_sm.cpp index 818ff341..aff4c608 100644 --- a/test/ft/co_sm.cpp +++ b/test/ft/co_sm.cpp @@ -250,13 +250,13 @@ test co_sm_deferred_scheduler_not_done_then_done = [] { expect(sm.is(s1)); }; +#if !BOOST_SML_DISABLE_EXCEPTIONS utility::bool_task throwing_bool_task() { throw std::runtime_error("test"); co_return true; } test bool_task_exception_rethrow = [] { -#if !BOOST_SML_DISABLE_EXCEPTIONS auto task = throwing_bool_task(); bool got_runtime_error = false; try { @@ -265,8 +265,8 @@ test bool_task_exception_rethrow = [] { got_runtime_error = true; } expect(got_runtime_error); -#endif }; +#endif test bool_task_await_suspend_with_non_empty_handle = [] { using sm_t = utility::co_sm, @@ -275,7 +275,7 @@ test bool_task_await_suspend_with_non_empty_handle = [] { sm_t sm{}; auto task = sm.process_event_async(e1{}); const auto resumed_handle = task.await_suspend(std::noop_coroutine()); - expect(resumed_handle != std::noop_coroutine()); + expect(resumed_handle == std::noop_coroutine()); sm.scheduler().run_pending(); expect(task.result()); expect(sm.is(s1)); diff --git a/test/ft/co_sm_thread_pool.cpp b/test/ft/co_sm_thread_pool.cpp new file mode 100644 index 00000000..00efc665 --- /dev/null +++ b/test/ft/co_sm_thread_pool.cpp @@ -0,0 +1,448 @@ +// +// Copyright (c) 2026 stateforward +// +// Distributed under the Boost Software License, Version 1.0. +// (See accompanying file LICENSE_1_0.txt or copy at +// http://www.boost.org/LICENSE_1_0.txt) +// +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#if !defined(_WIN32) +#include +#include +#endif + +namespace sml = boost::sml; + +#if BOOST_SML_UTILITY_CO_SM_ENABLED && BOOST_SML_UTILITY_THREAD_POOL_SCHEDULER_ENABLED +namespace utility = sml::utility; +namespace policy = utility::policy; + +using scheduler_t = policy::thread_pool_scheduler<1, 64, 128>; + +static_assert(policy::async_coroutine_scheduler_contract, + "thread_pool_scheduler is the async coroutine scheduler policy"); +static_assert(policy::valid_coroutine_scheduler, "thread_pool_scheduler provides schedule(fn)"); + +namespace { + +struct e1 {}; +struct e_nested {}; +struct e_probe {}; +const auto idle = sml::state; +const auto s1 = sml::state; + +template +bool wait_until(predicate&& done) { + const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(5); + while (!done()) { + if (std::chrono::steady_clock::now() >= deadline) { + return false; + } + std::this_thread::yield(); + } + return true; +} + +void block_worker(scheduler_t& scheduler, std::atomic& entered, std::atomic& release) { + scheduler.schedule([&entered, &release]() noexcept { + entered.store(true, std::memory_order_release); + while (!release.load(std::memory_order_acquire)) { + policy::cpu_relax(); + } + }); + expect(wait_until([&entered] { return entered.load(std::memory_order_acquire); })); +} + +struct simple_context { + std::atomic* calls = nullptr; +}; + +struct simple_machine { + auto operator()() const { + using namespace sml; + const auto record = [](simple_context& context) { context.calls->fetch_add(1, std::memory_order_acq_rel); }; + // clang-format off + return make_transition_table( + *idle + event / record = s1, + s1 + event / record = s1 + ); + // clang-format on + } +}; + +using simple_sm = utility::co_sm, + policy::coroutine_allocator>, simple_context>; + +utility::bool_task await_task(utility::bool_task& task, std::atomic& resumed, bool& accepted) { + accepted = co_await task; + resumed.store(true, std::memory_order_release); + co_return accepted; +} + +utility::bool_task await_then_dispatch_again(simple_sm& sm, utility::bool_task& task, std::atomic& resumed, + bool& first_accepted, bool& second_accepted) { + first_accepted = co_await task; + auto second = sm.process_event_async(e1{}); + second_accepted = co_await second; + resumed.store(true, std::memory_order_release); + co_return first_accepted&& second_accepted; +} + +struct fixed_allocator { + std::size_t allocate_calls = 0; + std::size_t deallocate_calls = 0; + bool heap_used = false; + bool used = false; + alignas(std::max_align_t) unsigned char storage[4096]{}; + + void* allocate(const std::size_t size, const std::size_t alignment) { + ++allocate_calls; + if (used || size > sizeof(storage) || alignment > alignof(std::max_align_t)) { + heap_used = true; + return nullptr; + } + used = true; + return storage; + } + + void deallocate(void* ptr, std::size_t, std::size_t) noexcept { + ++deallocate_calls; + expect(ptr == storage); + used = false; + } +}; + +using fixed_alloc_sm = utility::co_sm, + policy::coroutine_allocator, simple_context>; + +struct blocking_context { + std::atomic* entered = nullptr; + std::atomic* release = nullptr; +}; + +struct blocking_machine { + auto operator()() const { + using namespace sml; + const auto block = [](blocking_context& context) { + context.entered->store(true, std::memory_order_release); + while (!context.release->load(std::memory_order_acquire)) { + policy::cpu_relax(); + } + }; + // clang-format off + return make_transition_table( + *idle + event / block = s1 + ); + // clang-format on + } +}; + +using blocking_sm = utility::co_sm, + policy::coroutine_allocator>, blocking_context>; + +struct nested_context { + void* self = nullptr; + bool (*nested_probe)(void*) = nullptr; + std::atomic* calls = nullptr; + std::atomic* nested_accepts = nullptr; +}; + +struct nested_machine { + auto operator()() const { + using namespace sml; + const auto enter_nested = [](nested_context& context) { + context.calls->fetch_add(1, std::memory_order_acq_rel); + if (context.nested_probe(context.self)) { + context.nested_accepts->fetch_add(1, std::memory_order_acq_rel); + } + }; + const auto record_probe = [](nested_context& context) { context.calls->fetch_add(1, std::memory_order_acq_rel); }; + // clang-format off + return make_transition_table( + *idle + event / enter_nested = s1, + idle + event / record_probe = idle, + s1 + event / record_probe = s1 + ); + // clang-format on + } +}; + +using nested_sm = utility::co_sm, + policy::coroutine_allocator>, nested_context>; + +#if !BOOST_SML_DISABLE_EXCEPTIONS +struct throwing_move_event { + throwing_move_event() = default; + throwing_move_event(const throwing_move_event&) noexcept = default; + throwing_move_event(throwing_move_event&&) { + if (throw_on_move) { + throw std::runtime_error("move failed"); + } + } + + static bool throw_on_move; +}; + +bool throwing_move_event::throw_on_move = false; + +struct throwing_move_machine { + auto operator()() const { + using namespace sml; + const auto record = [](simple_context& context) { context.calls->fetch_add(1, std::memory_order_acq_rel); }; + // clang-format off + return make_transition_table( + *idle + event / record = s1 + ); + // clang-format on + } +}; + +using throwing_move_sm = + utility::co_sm, + policy::coroutine_allocator>, simple_context>; +#endif + +#if !defined(_WIN32) +void destroy_incomplete_task_child() { + std::atomic calls{0}; + simple_context context{&calls}; + simple_sm sm{context}; + std::atomic worker_entered{false}; + std::atomic release_worker{false}; + block_worker(sm.scheduler(), worker_entered, release_worker); + + auto task = sm.process_event_async(e1{}); + expect(!task.await_ready()); +} +#endif + +} // namespace + +test thread_pool_co_sm_process_event_async_starts_now_and_awaits_later = [] { + std::atomic calls{0}; + simple_context context{&calls}; + simple_sm sm{context}; + std::atomic worker_entered{false}; + std::atomic release_worker{false}; + block_worker(sm.scheduler(), worker_entered, release_worker); + + auto task = sm.process_event_async(e1{}); + expect(!task.await_ready()); + + int local_work = 0; + for (int i = 0; i < 7; ++i) { + ++local_work; + } + expect(7 == local_work); + expect(0 == calls.load(std::memory_order_acquire)); + + release_worker.store(true, std::memory_order_release); + expect(wait_until([&task] { return task.await_ready(); })); + expect(task.result()); + expect(1 == calls.load(std::memory_order_acquire)); + expect(sm.is(s1)); +}; + +test thread_pool_co_sm_await_resumes_after_worker_completion = [] { + std::atomic calls{0}; + simple_context context{&calls}; + simple_sm sm{context}; + std::atomic worker_entered{false}; + std::atomic release_worker{false}; + block_worker(sm.scheduler(), worker_entered, release_worker); + + auto task = sm.process_event_async(e1{}); + std::atomic resumed{false}; + bool accepted = false; + auto waiter = await_task(task, resumed, accepted); + + expect(!resumed.load(std::memory_order_acquire)); + expect(!waiter.await_ready()); + release_worker.store(true, std::memory_order_release); + expect(wait_until([&waiter] { return waiter.await_ready(); })); + expect(resumed.load(std::memory_order_acquire)); + expect(1 == calls.load(std::memory_order_acquire)); + expect(waiter.result()); + expect(accepted); + expect(task.result()); +}; + +test thread_pool_co_sm_await_can_dispatch_same_actor_again = [] { + std::atomic calls{0}; + simple_context context{&calls}; + simple_sm sm{context}; + std::atomic worker_entered{false}; + std::atomic release_worker{false}; + block_worker(sm.scheduler(), worker_entered, release_worker); + + auto task = sm.process_event_async(e1{}); + std::atomic resumed{false}; + bool first_accepted = false; + bool second_accepted = false; + auto waiter = await_then_dispatch_again(sm, task, resumed, first_accepted, second_accepted); + + expect(!resumed.load(std::memory_order_acquire)); + release_worker.store(true, std::memory_order_release); + expect(wait_until([&waiter] { return waiter.await_ready(); })); + expect(resumed.load(std::memory_order_acquire)); + expect(waiter.result()); + expect(first_accepted); + expect(second_accepted); + expect(2 == calls.load(std::memory_order_acquire)); +}; + +test thread_pool_co_sm_incomplete_task_destruction_terminates = [] { +#if !defined(_WIN32) + const pid_t pid = fork(); + expect(pid >= 0); + if (pid == 0) { + destroy_incomplete_task_child(); + _exit(0); + } + + int status = 0; + expect(waitpid(pid, &status, 0) == pid); + expect(WIFSIGNALED(status) || (WIFEXITED(status) && WEXITSTATUS(status) != 0)); +#else + expect(true); +#endif +}; + +test thread_pool_co_sm_fixed_allocator_avoids_heap_fallback = [] { + std::atomic calls{0}; + simple_context context{&calls}; + fixed_alloc_sm sm{context}; + + { + auto task = sm.process_event_async(e1{}); + expect(wait_until([&task] { return task.await_ready(); })); + expect(task.result()); + } + + expect(1u == sm.allocator().allocate_calls); + expect(1u == sm.allocator().deallocate_calls); + expect(!sm.allocator().heap_used); + expect(1 == calls.load(std::memory_order_acquire)); +}; + +test thread_pool_co_sm_rejects_concurrent_dispatch_to_same_actor = [] { + std::atomic calls{0}; + simple_context context{&calls}; + simple_sm sm{context}; + std::atomic worker_entered{false}; + std::atomic release_worker{false}; + block_worker(sm.scheduler(), worker_entered, release_worker); + + auto first = sm.process_event_async(e1{}); + auto second = sm.process_event_async(e1{}); + + expect(!first.await_ready()); + expect(second.await_ready()); + expect(!second.result()); + release_worker.store(true, std::memory_order_release); + expect(wait_until([&first] { return first.await_ready(); })); + expect(first.result()); + expect(1 == calls.load(std::memory_order_acquire)); +}; + +test thread_pool_co_sm_process_event_allows_same_stack_reentry = [] { + std::atomic calls{0}; + std::atomic nested_accepts{0}; + nested_context context{}; + nested_sm sm{context}; + context.self = &sm; + context.nested_probe = [](void* self) { return static_cast(self)->process_event(e_probe{}); }; + context.calls = &calls; + context.nested_accepts = &nested_accepts; + + auto task = sm.process_event_async(e_nested{}); + + expect(wait_until([&task] { return task.await_ready(); })); + expect(task.result()); + expect(2 == calls.load(std::memory_order_acquire)); + expect(1 == nested_accepts.load(std::memory_order_acquire)); + expect(sm.is(s1)); +}; + +test thread_pool_co_sm_direct_process_event_allows_same_stack_reentry = [] { + std::atomic calls{0}; + std::atomic nested_accepts{0}; + nested_context context{}; + nested_sm sm{context}; + context.self = &sm; + context.nested_probe = [](void* self) { return static_cast(self)->process_event(e_probe{}); }; + context.calls = &calls; + context.nested_accepts = &nested_accepts; + + expect(sm.process_event(e_nested{})); + + expect(2 == calls.load(std::memory_order_acquire)); + expect(1 == nested_accepts.load(std::memory_order_acquire)); + expect(sm.is(s1)); +}; + +test thread_pool_co_sm_setup_exception_releases_actor_guard = [] { +#if !BOOST_SML_DISABLE_EXCEPTIONS + std::atomic calls{0}; + simple_context context{&calls}; + throwing_move_sm sm{context}; + + throwing_move_event event{}; + throwing_move_event::throw_on_move = true; + auto failed = sm.process_event_async(event); + throwing_move_event::throw_on_move = false; + + expect(failed.await_ready()); + bool got_runtime_error = false; + try { + (void)failed.result(); + } catch (const std::runtime_error&) { + got_runtime_error = true; + } + expect(got_runtime_error); + + auto recovered = sm.process_event_async(throwing_move_event{}); + expect(wait_until([&recovered] { return recovered.await_ready(); })); + expect(recovered.result()); + expect(1 == calls.load(std::memory_order_acquire)); +#else + expect(true); +#endif +}; + +test thread_pool_co_sm_different_actors_run_concurrently = [] { + std::atomic entered1{false}; + std::atomic entered2{false}; + std::atomic release{false}; + blocking_context context1{&entered1, &release}; + blocking_context context2{&entered2, &release}; + blocking_sm sm1{context1}; + blocking_sm sm2{context2}; + + auto task1 = sm1.process_event_async(e1{}); + auto task2 = sm2.process_event_async(e1{}); + + expect(wait_until([&] { return entered1.load(std::memory_order_acquire) && entered2.load(std::memory_order_acquire); })); + expect(!task1.await_ready()); + expect(!task2.await_ready()); + + release.store(true, std::memory_order_release); + expect(wait_until([&] { return task1.await_ready() && task2.await_ready(); })); + expect(task1.result()); + expect(task2.result()); + expect(sm1.is(s1)); + expect(sm2.is(s1)); +}; + +#else +test thread_pool_co_sm_disabled_without_cxx20_coroutines_and_semaphore = [] { expect(true); }; +#endif diff --git a/test/ft/thread_pool_scheduler.cpp b/test/ft/thread_pool_scheduler.cpp index 0f1c98d2..c6770ba1 100644 --- a/test/ft/thread_pool_scheduler.cpp +++ b/test/ft/thread_pool_scheduler.cpp @@ -8,13 +8,13 @@ #include #include #include +#include #include #if BOOST_SML_UTILITY_THREAD_POOL_SCHEDULER_ENABLED namespace policy = boost::sml::utility::policy; using pool_t = policy::thread_pool_scheduler<8>; -using pool_ref_t = policy::thread_pool_scheduler_ref; // A non-noexcept callable: the scheduled wait/schedule paths must stay // unconditionally noexcept (their worker thunk cannot propagate), while the @@ -22,30 +22,31 @@ using pool_ref_t = policy::thread_pool_scheduler_ref; struct may_throw_callable { void operator()() const {} }; -static_assert(noexcept(std::declval().run_or_schedule_and_wait(may_throw_callable{})), +static_assert(noexcept(std::declval().run_or_schedule_and_wait(may_throw_callable{})), "run_or_schedule_and_wait must be unconditionally noexcept"); -static_assert(noexcept(std::declval().schedule(may_throw_callable{})), - "schedule must be unconditionally noexcept"); +static_assert(noexcept(std::declval().schedule(may_throw_callable{})), "schedule must be unconditionally noexcept"); static_assert(!noexcept(std::declval().try_run_immediate(may_throw_callable{})), "try_run_immediate stays conditionally noexcept (inline path propagates)"); test thread_pool_scheduler_runs_task_inline_when_idle = [] { - pool_t pool{}; - pool_ref_t scheduler{pool}; + pool_t scheduler{}; int calls = 0; - const bool ran = scheduler.run_or_schedule_and_wait([&calls] { ++calls; }); + const auto caller = std::this_thread::get_id(); + std::thread::id ran_on{}; + const bool ran = scheduler.run_or_schedule_and_wait([&] { + ++calls; + ran_on = std::this_thread::get_id(); + }); expect(ran); expect(1 == calls); // An idle pool runs the work on the calling thread rather than a worker. - expect(1u == scheduler.immediate_run_count()); - expect(0u == scheduler.worker_run_count()); + expect(caller == ran_on); }; test thread_pool_scheduler_fork_join_runs_every_lane = [] { - pool_t pool{}; - pool_ref_t scheduler{pool}; + pool_t scheduler{}; std::atomic calls{0}; - pool_ref_t::join_group group{}; + pool_t::join_group group{}; for (std::size_t lane = 0; lane < 8u; ++lane) { scheduler.try_submit(group, [&calls] { calls.fetch_add(1, std::memory_order_relaxed); }); } @@ -58,13 +59,12 @@ test thread_pool_scheduler_fork_join_runs_every_lane = [] { // must not make a later all-success round report failure. (try_submit from inside // a worker is rejected, since a worker cannot fork into its own pool.) test thread_pool_scheduler_reused_join_group_clears_prior_rejection = [] { - pool_t pool{}; - pool_ref_t scheduler{pool}; - pool_ref_t::join_group group{}; + pool_t scheduler{}; + pool_t::join_group group{}; // Round 1: drive a worker that tries to submit into `group` -> rejected. std::atomic rejected_on_worker{false}; - pool_ref_t::join_group driver{}; + pool_t::join_group driver{}; scheduler.try_submit(driver, [&] { rejected_on_worker.store(!scheduler.try_submit(group, [] {}), std::memory_order_release); }); driver.wait(); @@ -85,14 +85,13 @@ test thread_pool_scheduler_reused_join_group_clears_prior_rejection = [] { // destroy-during-release semaphore could strand a wakeup. The lifetime-safe // spin-join must complete every round without hanging. (Reproduced reliably with // thousands of rounds; kept here as a standing guard.) -test thread_pool_scheduler_ref_fork_join_survives_rapid_repeated_rounds = [] { +test thread_pool_scheduler_fork_join_survives_rapid_repeated_rounds = [] { constexpr std::size_t lanes = 8u; constexpr int rounds = 20000; - pool_t pool{}; - pool_ref_t scheduler{pool}; + pool_t scheduler{}; std::atomic calls{0}; for (int round = 0; round < rounds; ++round) { - pool_ref_t::join_group group{}; + pool_t::join_group group{}; for (std::size_t lane = 0; lane < lanes; ++lane) { scheduler.try_submit(group, [&calls] { calls.fetch_add(1, std::memory_order_relaxed); }); }