Add external-completion scheduler policy for co_sm

include/boost/sml/utility/co_sm.hpp

@@ -303,6 +303,24 @@ concept has_try_run_immediate = requires(TScheduler scheduler) {
   } -> std::same_as<bool>;
 };
 
+// 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 <class TScheduler>
+concept external_completion_scheduler_contract = requires {
+  { TScheduler::external_completion } -> std::convertible_to<bool>;
+  { TScheduler::npos } -> std::convertible_to<std::size_t>;
+  typename TScheduler::completion_event;
+} && static_cast<bool>(TScheduler::external_completion) && requires(TScheduler scheduler) {
+  { scheduler.has_required() } -> std::convertible_to<bool>;
+  { scheduler.sweep_next_fired() } -> std::same_as<std::size_t>;
+  { scheduler.consume_next_fired_required() } -> std::same_as<std::size_t>;
+  scheduler.wait_any();
+  scheduler.reset_dispatch_state();
+};
+
 template <class TAllocatorPolicy>
 concept valid_coroutine_allocator_policy = requires { typename TAllocatorPolicy::allocator_type; };
 
@@ -578,27 +596,76 @@ class co_sm {
 
   template <class TEvent>
   bool process_event(const TEvent& event) {
-    return state_machine_.process_event(event);
+    if constexpr (policy::external_completion_scheduler_contract<scheduler_type>) {
+      // 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 <class TEvent>
   bool_task process_event_async(TEvent&& event) {
     using event_t = typename std::decay<TEvent>::type;
     event_t event_copy(static_cast<TEvent&&>(event));
 
-    if constexpr (std::is_same_v<scheduler_type, policy::inline_scheduler>) {
+    if constexpr (policy::external_completion_scheduler_contract<scheduler_type>) {
+      return bool_task::from_value(process_event(event_copy));
+    } else if constexpr (std::is_same_v<scheduler_type, policy::inline_scheduler>) {
       return bool_task::from_value(state_machine_.process_event(event_copy));
-    }
-
-    if constexpr (policy::has_try_run_immediate<scheduler_type>) {
-      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<scheduler_type>) {
+        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_t&&>(event_copy));
+      return process_event_async_impl(std::allocator_arg, allocator_, *this, static_cast<event_t&&>(event_copy));
+    }
   }
 
   template <class TState>
@@ -627,6 +694,13 @@ class co_sm {
     co_return co_await process_event_awaitable<typename std::decay<TEvent>::type>{self, static_cast<TEvent&&>(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 <class TEvent>
   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

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(<semaphore>)
+#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 <array>
+#include <atomic>
+#include <cstddef>
+#include <cstdint>
+#include <semaphore>
+
+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<completion_source*>(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 <std::size_t>
+  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<std::uint8_t> 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 <std::size_t SourceCount = 8>
+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<std::size_t>(-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 <class TFn>
+  void schedule(TFn&& fn) noexcept(noexcept(static_cast<TFn&&>(fn)())) {
+    static_cast<TFn&&>(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<completion_source, SourceCount> 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

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 <boost/sml/utility/external_completion.hpp>
+#include <stateforward/sml.hpp>
+
+#endif

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)