diff --git a/include/boost/sml/utility/co_sm.hpp b/include/boost/sml/utility/co_sm.hpp index 65ddc486..cbe6c780 100644 --- a/include/boost/sml/utility/co_sm.hpp +++ b/include/boost/sml/utility/co_sm.hpp @@ -303,6 +303,24 @@ concept has_try_run_immediate = requires(TScheduler scheduler) { } -> std::same_as; }; +// Structural contract for schedulers that deliver externally fired completions +// (see utility/external_completion.hpp). Detected duck-typed so this header +// stays independent of the opt-in module. Such schedulers still honor the +// strict ordering contract: required completions are drained before the +// top-level dispatch returns, and delivery order is deterministic. +template +concept external_completion_scheduler_contract = requires { + { TScheduler::external_completion } -> std::convertible_to; + { TScheduler::npos } -> std::convertible_to; + typename TScheduler::completion_event; +} && static_cast(TScheduler::external_completion) && requires(TScheduler scheduler) { + { scheduler.has_required() } -> std::convertible_to; + { scheduler.sweep_next_fired() } -> std::same_as; + { scheduler.consume_next_fired_required() } -> std::same_as; + scheduler.wait_any(); + scheduler.reset_dispatch_state(); +}; + template concept valid_coroutine_allocator_policy = requires { typename TAllocatorPolicy::allocator_type; }; @@ -578,7 +596,54 @@ class co_sm { template bool process_event(const TEvent& event) { - return state_machine_.process_event(event); + if constexpr (policy::external_completion_scheduler_contract) { + // Only the outermost dispatch owns the completion machinery: a nested + // synchronous process_event from a transition action dispatches its + // trigger plainly, and any sources it marks required are drained by the + // enclosing dispatch before that dispatch returns. Resetting or + // draining from the nested call would clear required bits the outer + // drain still needs. + if (external_dispatch_depth_ > 0) { + return state_machine_.process_event(event); + } + + struct dispatch_depth_scope { + std::size_t& depth; + ~dispatch_depth_scope() noexcept { --depth; } + }; + ++external_dispatch_depth_; + dispatch_depth_scope depth_scope{external_dispatch_depth_}; + + // Synchronous drive with zero coroutine machinery: (1) commit sweep - + // completions that fired between dispatches are delivered as explicit + // completion events in ascending source order before the trigger runs; + // (2) the trigger dispatches, its actions may arm sources, hand them to + // workers, and mark them required; (3) the drain loop blocks on the + // scheduler's semaphore and delivers each required completion (lowest + // fired index first) until none remain, so nothing observable escapes + // this call and delivery happens only on this (the dispatching) + // thread. Required bits left by a previous aborted dispatch are + // cleared first. The fast path (no fired and no required sources) costs + // one sweep scan plus one mask check over the inline dispatch. + scheduler_.reset_dispatch_state(); + using completion_event_t = typename scheduler_type::completion_event; + for (std::size_t index = scheduler_.sweep_next_fired(); index != scheduler_type::npos; + index = scheduler_.sweep_next_fired()) { + (void)state_machine_.process_event(completion_event_t{index}); + } + bool accepted = state_machine_.process_event(event); + while (scheduler_.has_required()) { + const std::size_t index = scheduler_.consume_next_fired_required(); + if (index == scheduler_type::npos) { + scheduler_.wait_any(); + continue; + } + accepted = state_machine_.process_event(completion_event_t{index}) && accepted; + } + return accepted; + } else { + return state_machine_.process_event(event); + } } template @@ -586,19 +651,21 @@ class co_sm { using event_t = typename std::decay::type; event_t event_copy(static_cast(event)); - if constexpr (std::is_same_v) { + if constexpr (policy::external_completion_scheduler_contract) { + return bool_task::from_value(process_event(event_copy)); + } else if constexpr (std::is_same_v) { return bool_task::from_value(state_machine_.process_event(event_copy)); - } - - if constexpr (policy::has_try_run_immediate) { - bool accepted = false; - if (scheduler_.try_run_immediate( - [this, &event_copy, &accepted]() { accepted = state_machine_.process_event(event_copy); })) { - return bool_task::from_value(accepted); + } else { + if constexpr (policy::has_try_run_immediate) { + bool accepted = false; + if (scheduler_.try_run_immediate( + [this, &event_copy, &accepted]() { accepted = state_machine_.process_event(event_copy); })) { + return bool_task::from_value(accepted); + } } - } - return process_event_async_impl(std::allocator_arg, allocator_, *this, static_cast(event_copy)); + return process_event_async_impl(std::allocator_arg, allocator_, *this, static_cast(event_copy)); + } } template @@ -627,6 +694,13 @@ class co_sm { co_return co_await process_event_awaitable::type>{self, static_cast(event)}; } // GCOVR_EXCL_LINE + // External-completion dispatch: (1) commit sweep — completions fired between + // dispatches are delivered as explicit completion events in ascending source + // order before the trigger runs; (2) the trigger event dispatches; (3) any + // sources the machine marked required are awaited (suspending this coroutine) + // and re-entered as completion events until none remain. The caller's drive + // loop in process_event resumes the coroutine on the dispatching thread only. + template struct process_event_awaitable { co_sm& self; @@ -670,6 +744,10 @@ class co_sm { state_machine_type state_machine_{}; scheduler_type scheduler_{}; allocator_type allocator_{}; + // 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; }; } // namespace utility diff --git a/include/boost/sml/utility/external_completion.hpp b/include/boost/sml/utility/external_completion.hpp new file mode 100644 index 00000000..c47d6984 --- /dev/null +++ b/include/boost/sml/utility/external_completion.hpp @@ -0,0 +1,225 @@ +// +// 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) +// +#ifndef BOOST_SML_UTILITY_EXTERNAL_COMPLETION_HPP +#define BOOST_SML_UTILITY_EXTERNAL_COMPLETION_HPP + +#include "boost/sml.hpp" + +#if defined(_MSVC_LANG) +#define BOOST_SML_UTILITY_EXTERNAL_COMPLETION_LANG _MSVC_LANG +#else +#define BOOST_SML_UTILITY_EXTERNAL_COMPLETION_LANG __cplusplus +#endif + +// Opt-in module: external completion delivery needs C++20 coroutines plus a +// hosted std::counting_semaphore, so it compiles to nothing on freestanding or +// pre-C++20 toolchains, keeping the header-only core dependency-free. +#if BOOST_SML_UTILITY_EXTERNAL_COMPLETION_LANG >= 202002L && (defined(__cpp_impl_coroutine) || defined(__cpp_coroutines)) +#if defined(__has_include) +#if __has_include() +#define BOOST_SML_UTILITY_EXTERNAL_COMPLETION_ENABLED 1 +#endif +#endif +#endif +#ifndef BOOST_SML_UTILITY_EXTERNAL_COMPLETION_ENABLED +#define BOOST_SML_UTILITY_EXTERNAL_COMPLETION_ENABLED 0 +#endif + +#if BOOST_SML_UTILITY_EXTERNAL_COMPLETION_ENABLED +#include +#include +#include +#include +#include + +BOOST_SML_NAMESPACE_BEGIN + +namespace utility { + +// Generic completion trigger delivered into a machine's transition table by a +// co_sm running an external-completion scheduler policy. source_index names the +// completion_source that fired; machines map it onto domain state (for example +// a buffer slot) via their own guards and actions. +struct completion { + std::size_t source_index{}; +}; + +namespace policy { + +// One externally fired completion flag. Producers (worker threads, device +// callbacks) touch only fire(); every other member is single-consumer and runs +// on the dispatching thread. fire() never resumes a coroutine and never touches +// the state machine: it flips the flag and releases the scheduler's semaphore, +// so single-writer dispatch holds structurally no matter which thread fires. +class completion_source { + public: + // Trampoline compatible with thread_pool_scheduler::try_submit_with_completion. + // Only the transition into fired releases a wakeup permit, so a misbehaving + // producer that fires twice cannot accumulate permits or overflow the + // semaphore's max count. + static void fire(void* ctx) noexcept { + auto* source = static_cast(ctx); + if (source->state_.exchange(state_fired, std::memory_order_release) != state_fired) { + source->wakeup_->release(); + } + } + + void arm() noexcept { state_.store(state_pending, std::memory_order_relaxed); } + void reset() noexcept { state_.store(state_idle, std::memory_order_relaxed); } + bool is_fired() const noexcept { return state_.load(std::memory_order_acquire) == state_fired; } + bool is_idle() const noexcept { return state_.load(std::memory_order_acquire) == state_idle; } + + private: + template + friend class external_completion_scheduler; + + static constexpr std::uint8_t state_idle = 0; + static constexpr std::uint8_t state_pending = 1; + static constexpr std::uint8_t state_fired = 2; + + std::atomic state_{state_idle}; + std::counting_semaphore<>* wakeup_ = nullptr; +}; + +// Scheduler policy that lets a co_sm dispatch block on completions fired by +// other threads while still returning only when the whole dispatch is done: +// +// 1. commit sweep — completions that fired between dispatches are delivered +// as explicit `utility::completion` events, ascending index order, before +// the trigger event runs; +// 2. the trigger event dispatches; its actions may arm sources, hand them to +// workers, and mark some of them required for this dispatch; +// 3. required-wait drain — the dispatching thread consumes each required +// source (lowest fired index first) as a `utility::completion` event, +// blocking on the semaphore between fires, until none remain. +// +// All bookkeeping is single-consumer on the dispatching thread; producers +// interact only through completion_source::fire. The run-to-completion +// guarantees are honored because nothing escapes the top-level dispatch: +// required completions are drained before it returns, and unrequired +// (background) fires persist only as passive flags swept by the next dispatch. +// The drive is a plain loop, so the no-completion fast path adds only one +// sweep scan and one mask check over an inline dispatch. +// +// Contract: a source marked required must have been handed to a producer that +// will eventually fire it, or the drain loop cannot finish. Destroying the +// scheduler while producers still hold source pointers is undefined; owners +// drain or join in-flight work first. +template +class external_completion_scheduler { + public: + static_assert(SourceCount > 0, "external_completion_scheduler needs at least one source"); + static_assert(SourceCount <= 64, "required-source bookkeeping is a 64-bit mask"); + + static constexpr bool guarantees_fifo = true; + static constexpr bool single_consumer = true; + static constexpr bool run_to_completion = true; + static constexpr bool external_completion = true; + static constexpr std::size_t npos = static_cast(-1); + + using completion_event = completion; + + external_completion_scheduler() noexcept { + for (auto& source : sources_) { + source.wakeup_ = &wakeup_; + } + } + + external_completion_scheduler(const external_completion_scheduler&) = delete; + external_completion_scheduler& operator=(const external_completion_scheduler&) = delete; + external_completion_scheduler(external_completion_scheduler&&) = delete; + external_completion_scheduler& operator=(external_completion_scheduler&&) = delete; + + // Continuations run inline on the dispatching thread. + template + void schedule(TFn&& fn) noexcept(noexcept(static_cast(fn)())) { + static_cast(fn)(); + } + + completion_source& source(const std::size_t index) noexcept { return sources_[index]; } + + // Clears bookkeeping an aborted dispatch may have left behind (for example + // a state-machine action that threw after require()): required bits are + // meaningless outside the dispatch that created them, and stale bits would + // otherwise block the next drain forever. + void reset_dispatch_state() noexcept { required_ = 0; } + + // Marks a source as blocking the current dispatch. Consumer thread only + // (machine actions via an injected scheduler pointer). + void require(const std::size_t index) noexcept { required_ |= bit(index); } + + bool has_required() const noexcept { return required_ != 0u; } + + // Lowest fired source not marked required, consumed and returned for + // delivery as a completion event; npos when none. Consumer thread only. + std::size_t sweep_next_fired() noexcept { + for (std::size_t index = 0; index < SourceCount; ++index) { + if ((required_ & bit(index)) == 0u && sources_[index].is_fired()) { + consume_fired(index); + return index; + } + } + return npos; + } + + // Lowest fired source marked required, consumed (required bit cleared, + // source reset) and returned for delivery as a completion event; npos when + // none has fired yet. Consumer thread only. + std::size_t consume_next_fired_required() noexcept { + const std::size_t index = lowest_required_fired(); + if (index == npos) { + return npos; + } + required_ &= ~bit(index); + consume_fired(index); + return index; + } + + // Blocks the dispatching thread until some source fires. Permits accumulate, + // so a fire that lands before the wait is never lost; a permit consumed for + // a background (unrequired) fire simply re-checks and waits again, leaving + // that fire set for a later sweep. + void wait_any() noexcept { wakeup_.acquire(); } + + private: + static constexpr std::uint64_t bit(const std::size_t index) noexcept { return std::uint64_t{1} << index; } + + // Consuming a fired flag also burns the wakeup permit its fire released + // (non-blocking: the permit may already have been spent by a blocking wait + // in the drain loop). Fired flags are the ground truth - the drain only + // blocks after finding no consumable flag - so permit accounting cannot + // cause a hang; without this burn, background fires swept between + // dispatches would leak permits for the scheduler's lifetime. + void consume_fired(const std::size_t index) noexcept { + sources_[index].reset(); + (void)wakeup_.try_acquire(); + } + + std::size_t lowest_required_fired() const noexcept { + for (std::size_t index = 0; index < SourceCount; ++index) { + if ((required_ & bit(index)) != 0u && sources_[index].is_fired()) { + return index; + } + } + return npos; + } + + std::array sources_{}; + std::counting_semaphore<> wakeup_{0}; + std::uint64_t required_ = 0; +}; + +} // namespace policy +} // namespace utility + +BOOST_SML_NAMESPACE_END +#endif + +#undef BOOST_SML_UTILITY_EXTERNAL_COMPLETION_LANG + +#endif // BOOST_SML_UTILITY_EXTERNAL_COMPLETION_HPP diff --git a/include/stateforward/sml/utility/external_completion.hpp b/include/stateforward/sml/utility/external_completion.hpp new file mode 100644 index 00000000..2f9d294a --- /dev/null +++ b/include/stateforward/sml/utility/external_completion.hpp @@ -0,0 +1,7 @@ +#ifndef STATEFORWARD_SML_UTILITY_EXTERNAL_COMPLETION_HPP +#define STATEFORWARD_SML_UTILITY_EXTERNAL_COMPLETION_HPP + +#include +#include + +#endif diff --git a/test/ft/CMakeLists.txt b/test/ft/CMakeLists.txt index 5ac48a60..be82b57c 100644 --- a/test/ft/CMakeLists.txt +++ b/test/ft/CMakeLists.txt @@ -23,6 +23,10 @@ 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_external_completion external_completion.cpp) +add_test(test_external_completion test_external_completion) +target_link_libraries(test_external_completion -lpthread) + add_executable(test_constexpr constexpr.cpp) add_test(test_constexpr test_constexpr) diff --git a/test/ft/external_completion.cpp b/test/ft/external_completion.cpp new file mode 100644 index 00000000..7682b2a9 --- /dev/null +++ b/test/ft/external_completion.cpp @@ -0,0 +1,392 @@ +// +// 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; + +#if BOOST_SML_UTILITY_EXTERNAL_COMPLETION_ENABLED && BOOST_SML_UTILITY_CO_SM_ENABLED && \ + BOOST_SML_UTILITY_THREAD_POOL_SCHEDULER_ENABLED +namespace utility = sml::utility; + +using scheduler_t = utility::policy::external_completion_scheduler<8>; +using pool_t = utility::policy::thread_pool_scheduler<1>; + +static_assert(utility::policy::external_completion_scheduler_contract, + "external_completion_scheduler must satisfy the external completion contract"); +static_assert(!utility::policy::external_completion_scheduler_contract, + "inline_scheduler must not satisfy the external completion contract"); +static_assert(!utility::policy::external_completion_scheduler_contract>, + "fifo_scheduler must not satisfy the external completion contract"); +static_assert(utility::policy::strict_ordering_scheduler_contract, + "external completion drains before return, honoring the strict ordering contract"); + +namespace { + +constexpr std::size_t k_probe_marker = 99; + +struct test_context { + scheduler_t* scheduler = nullptr; + pool_t* pool = nullptr; + std::array log{}; + std::size_t log_count = 0; + std::thread::id completion_thread{}; + bool worker_delay = false; +}; + +struct e_require_via_worker { + std::size_t count{}; +}; +struct e_require_fired_inline { + std::size_t count{}; +}; +struct e_hand_to_worker_unrequired { + std::size_t index{}; +}; +struct e_probe {}; + +void fire_after_optional_delay(test_context& context, const std::size_t index) { + const bool delay = context.worker_delay; + (void)context.pool->try_submit_with_completion( + [delay]() noexcept { + if (delay) { + const auto until = std::chrono::steady_clock::now() + std::chrono::milliseconds(2); + while (std::chrono::steady_clock::now() < until) { + } + } + }, + &context.scheduler->source(index), &utility::policy::completion_source::fire); +} + +struct machine { + auto operator()() const { + using namespace sml; + + const auto a_require_via_worker = [](const e_require_via_worker& ev, test_context& context) { + for (std::size_t index = 0; index < ev.count; ++index) { + context.scheduler->source(index).arm(); + fire_after_optional_delay(context, index); + } + for (std::size_t index = 0; index < ev.count; ++index) { + context.scheduler->require(index); + } + }; + + const auto a_require_fired_inline = [](const e_require_fired_inline& ev, test_context& context) { + for (std::size_t index = 0; index < ev.count; ++index) { + context.scheduler->source(index).arm(); + } + // Fire in descending order on this thread: delivery must still ascend. + for (std::size_t reverse = ev.count; reverse > 0; --reverse) { + utility::policy::completion_source::fire(&context.scheduler->source(reverse - 1)); + } + for (std::size_t index = 0; index < ev.count; ++index) { + context.scheduler->require(index); + } + }; + + const auto a_hand_to_worker_unrequired = [](const e_hand_to_worker_unrequired& ev, test_context& context) { + context.scheduler->source(ev.index).arm(); + fire_after_optional_delay(context, ev.index); + }; + + const auto a_probe = [](const e_probe&, test_context& context) { context.log[context.log_count++] = k_probe_marker; }; + + const auto a_record = [](const utility::completion& ev, test_context& context) { + context.log[context.log_count++] = ev.source_index; + context.completion_thread = std::this_thread::get_id(); + }; + + // clang-format off + return make_transition_table( + *"ready"_s + event / a_require_via_worker + , "ready"_s + event / a_require_fired_inline + , "ready"_s + event / a_hand_to_worker_unrequired + , "ready"_s + event / a_probe + , "ready"_s + event / a_record + ); + // clang-format on + } +}; + +using co_sm_t = + utility::co_sm, + utility::policy::coroutine_allocator>, test_context>; + +struct fixture { + pool_t pool{}; + test_context context{}; + co_sm_t machine_instance; + + fixture() : machine_instance{context} { + context.scheduler = &machine_instance.scheduler(); + context.pool = &pool; + } +}; + +} // namespace + +test external_completion_required_drained_before_return = [] { + fixture f{}; + f.context.worker_delay = true; // force the dispatch coroutine to park + + expect(f.machine_instance.process_event(e_require_via_worker{3})); + + // Everything observable happened inside the dispatch: all three completions + // were delivered, nothing remains required, and the sources are reusable. + expect(3u == f.context.log_count); + expect(!f.machine_instance.scheduler().has_required()); + for (std::size_t index = 0; index < 3; ++index) { + expect(f.machine_instance.scheduler().source(index).is_idle()); + } +}; + +test external_completion_delivers_on_dispatching_thread = [] { + fixture f{}; + f.context.worker_delay = true; + + expect(f.machine_instance.process_event(e_require_via_worker{1})); + + expect(1u == f.context.log_count); + expect(std::this_thread::get_id() == f.context.completion_thread); +}; + +test external_completion_prefired_sources_deliver_ascending_without_park = [] { + fixture f{}; + + // Sources fire 2,1,0 on this thread before the drain runs: no suspension, + // and delivery is deterministic ascending regardless of fire order. + expect(f.machine_instance.process_event(e_require_fired_inline{3})); + + expect(3u == f.context.log_count); + expect(0u == f.context.log[0]); + expect(1u == f.context.log[1]); + expect(2u == f.context.log[2]); +}; + +test external_completion_background_fire_swept_before_next_trigger = [] { + fixture f{}; + + // Dispatch 1 hands source 5 to a worker without requiring it; the dispatch + // returns while the fire is still pending. + expect(f.machine_instance.process_event(e_hand_to_worker_unrequired{5})); + expect(0u == f.context.log_count); + + while (!f.machine_instance.scheduler().source(5).is_fired()) { + std::this_thread::yield(); + } + + // Dispatch 2: the commit sweep delivers completion(5) before the trigger. + expect(f.machine_instance.process_event(e_probe{})); + expect(2u == f.context.log_count); + expect(5u == f.context.log[0]); + expect(k_probe_marker == f.context.log[1]); + expect(f.machine_instance.scheduler().source(5).is_idle()); +}; + +test external_completion_dispatch_without_requires_never_parks = [] { + fixture f{}; + + expect(f.machine_instance.process_event(e_probe{})); + + expect(1u == f.context.log_count); + expect(k_probe_marker == f.context.log[0]); +}; + +test external_completion_async_wrapper_completes_inline = [] { + fixture f{}; + f.context.worker_delay = true; + + utility::bool_task task = f.machine_instance.process_event_async(e_require_via_worker{2}); + + expect(task.await_ready()); + expect(task.result()); + expect(2u == f.context.log_count); +}; + +test external_completion_double_fire_releases_one_permit = [] { + fixture f{}; + + // A misbehaving producer fires the same source twice: only the transition + // into fired may release a permit, so exactly one completion is delivered + // and no stray permit disturbs a later dispatch. + f.context.scheduler->source(2).arm(); + utility::policy::completion_source::fire(&f.context.scheduler->source(2)); + utility::policy::completion_source::fire(&f.context.scheduler->source(2)); + + expect(f.machine_instance.process_event(e_probe{})); + expect(2u == f.context.log_count); + expect(2u == f.context.log[0]); + expect(k_probe_marker == f.context.log[1]); + + f.context.log_count = 0; + expect(f.machine_instance.process_event(e_require_fired_inline{1})); + expect(1u == f.context.log_count); + expect(0u == f.context.log[0]); +}; + +#if !defined(BOOST_SML_DISABLE_EXCEPTIONS) || !BOOST_SML_DISABLE_EXCEPTIONS +struct e_require_then_throw {}; + +test external_completion_aborted_dispatch_does_not_poison_next = [] { + fixture f{}; + + struct throwing_machine { + auto operator()() const { + using namespace sml; + const auto a_require_then_throw = [](const e_require_then_throw&, test_context& context) { + context.scheduler->require(4); // never armed, never fired + throw std::runtime_error{"abort after require"}; + }; + const auto a_probe = [](const e_probe&, test_context& context) { context.log[context.log_count++] = k_probe_marker; }; + const auto a_record = [](const utility::completion& ev, test_context& context) { + context.log[context.log_count++] = ev.source_index; + }; + // clang-format off + return make_transition_table( + *"ready"_s + event / a_require_then_throw + , "ready"_s + event / a_probe + , "ready"_s + event / a_record + ); + // clang-format on + } + }; + + using throwing_co_sm = + utility::co_sm, + utility::policy::coroutine_allocator>, test_context>; + test_context context{}; + throwing_co_sm machine{context}; + context.scheduler = &machine.scheduler(); + + bool threw = false; + try { + (void)machine.process_event(e_require_then_throw{}); + } catch (const std::runtime_error&) { + threw = true; + } + expect(threw); + + // Pre-fix, the stale required bit for the never-fired source 4 would make + // this dispatch's drain block forever; post-fix it completes immediately. + expect(machine.process_event(e_probe{})); + expect(1u == context.log_count); + expect(k_probe_marker == context.log[0]); +}; +#endif + +struct e_require_then_nested_probe {}; + +test external_completion_nested_dispatch_preserves_outer_drain = [] { + struct nested_context { + scheduler_t* scheduler = nullptr; + pool_t* pool = nullptr; + std::array log{}; + std::size_t log_count = 0; + void* self = nullptr; + bool (*nested_probe)(void*) = nullptr; + }; + + struct nested_machine { + auto operator()() const { + using namespace sml; + const auto a_require_then_nested_probe = [](const e_require_then_nested_probe&, nested_context& context) { + context.scheduler->source(0).arm(); + (void)context.pool->try_submit_with_completion( + []() noexcept { + const auto until = std::chrono::steady_clock::now() + std::chrono::milliseconds(2); + while (std::chrono::steady_clock::now() < until) { + } + }, + &context.scheduler->source(0), &utility::policy::completion_source::fire); + context.scheduler->require(0); + // Synchronous re-entry: the nested dispatch must not clear the + // required bit the outer drain is about to wait on. + (void)context.nested_probe(context.self); + }; + const auto a_probe = [](const e_probe&, nested_context& context) { context.log[context.log_count++] = k_probe_marker; }; + const auto a_record = [](const utility::completion& ev, nested_context& context) { + context.log[context.log_count++] = ev.source_index; + }; + // clang-format off + return make_transition_table( + *"ready"_s + event / a_require_then_nested_probe + , "ready"_s + event / a_probe + , "ready"_s + event / a_record + ); + // clang-format on + } + }; + + using nested_co_sm = + utility::co_sm, + utility::policy::coroutine_allocator>, nested_context>; + pool_t pool{}; + nested_context context{}; + nested_co_sm machine{context}; + context.scheduler = &machine.scheduler(); + context.pool = &pool; + context.self = &machine; + context.nested_probe = [](void* self) { return static_cast(self)->process_event(e_probe{}); }; + + expect(machine.process_event(e_require_then_nested_probe{})); + + // The nested probe ran inside the outer dispatch, and the outer drain still + // delivered the required completion before returning. + expect(2u == context.log_count); + expect(k_probe_marker == context.log[0]); + expect(0u == context.log[1]); + expect(!machine.scheduler().has_required()); + expect(machine.scheduler().source(0).is_idle()); +}; + +test external_completion_background_fires_do_not_leak_permits = [] { + fixture f{}; + + // Each background fire releases one wakeup permit; the sweep that consumes + // its flag must burn that permit too, or permits accumulate for the + // scheduler's lifetime. After many cycles a worker-fired required dispatch + // must still drain correctly (a leak would surface as the drain loop + // spinning on stale permits or, at the extreme, fire() terminating on + // semaphore overflow). + for (std::size_t cycle = 0; cycle < 10000; ++cycle) { + f.context.scheduler->source(3).arm(); + utility::policy::completion_source::fire(&f.context.scheduler->source(3)); + expect(f.machine_instance.process_event(e_probe{})); + f.context.log_count = 0; + } + + f.context.worker_delay = true; + expect(f.machine_instance.process_event(e_require_via_worker{2})); + expect(2u == f.context.log_count); + expect(0u == f.context.log[0]); + expect(1u == f.context.log[1]); +}; + +test external_completion_sources_reusable_across_dispatches = [] { + fixture f{}; + + for (std::size_t round = 0; round < 3; ++round) { + f.context.log_count = 0; + expect(f.machine_instance.process_event(e_require_fired_inline{2})); + expect(2u == f.context.log_count); + expect(0u == f.context.log[0]); + expect(1u == f.context.log[1]); + } +}; +#else +test external_completion_disabled_without_cxx20_coroutines_and_semaphore = [] { expect(true); }; +#endif