[codex] Fix thread pool coroutine scheduler

README.md

@@ -29,7 +29,7 @@ This fork addresses critical bugs, adds utilities for high-throughput and async
 
 - **Completion transitions** (`completion<T>`) -- 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,34 +126,52 @@ cl /std:c++14 /Ox /W3 tcp_release.cpp
 
 ## Utilities
 
+<!-- utility APIs from include/boost/sml/utility/co_sm.hpp and include/boost/sml/utility/thread_pool_scheduler.hpp -->
+
 ### co_sm (C++20)
 
 Coroutine-driven state machine wrapper with configurable scheduling and allocation policies. Requires C++20 with coroutine support.
 
 ```cpp
 #include <stateforward/sml/utility/co_sm.hpp>
+#include <stateforward/sml/utility/thread_pool_scheduler.hpp>
 
 namespace utility = sml::utility;
 
-// Default: inline scheduler, pooled coroutine allocator
+// Default: FIFO scheduler, pooled coroutine allocator
 utility::co_sm<my_fsm> sm{};
 
 // Synchronous dispatch (same as regular sm)
 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<my_fsm, utility::policy::coroutine_scheduler<pool_scheduler>> 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<Capacity, InlineBytes>` | FIFO queue with bounded inline storage |
+| `inline_scheduler` | Runs tasks immediately |
+| `fifo_scheduler<Capacity, InlineBytes>` | FIFO queue with bounded inline storage (default) |
+| `thread_pool_scheduler<Workers, Capacity, InlineBytes>` | Starts `process_event_async` on worker threads and returns a task to await later; concurrent same-actor overlap is rejected |
 | `coroutine_scheduler<TScheduler>` | 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 |

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)

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 <array>
+#include <atomic>
+#include <boost/sml.hpp>
+#include <boost/sml/utility/co_sm.hpp>
+#include <boost/sml/utility/thread_pool_scheduler.hpp>
+#include <chrono>
+#include <cstdint>
+#include <exception>
+#include <iostream>
+#include <thread>
+
+namespace sml = boost::sml;
+
+struct tick {
+  std::atomic<std::uint64_t>* 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<tick> / 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<tick> / work = "ready"_s
+    );
+    // clang-format on
+  }
+};
+
+template <class fn>
+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<std::chrono::nanoseconds>(finish - start);
+}
+
+void print_duration(const std::chrono::nanoseconds elapsed, const std::size_t dispatches) {
+  const auto total_ms = std::chrono::duration_cast<std::chrono::milliseconds>(elapsed).count();
+  const auto ns_per_dispatch = dispatches == 0u ? 0u : static_cast<std::size_t>(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 <class machine, std::size_t children_count, std::size_t rounds>
+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<scheduler_t>,
+      sml::utility::policy::coroutine_allocator<sml::utility::policy::pooled_coroutine_allocator<4096, children_count * 2>>>;
+
+  constexpr std::size_t dispatches = children_count * rounds;
+
+  std::array<std::atomic<std::uint64_t>, children_count> blocking_calls{};
+  std::array<child_sm_t, children_count> 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<std::atomic<std::uint64_t>, children_count> async_calls{};
+  std::array<child_sm_t, children_count> async_children{};
+  std::cout << label << " start-all await-later: ";
+  print_duration(
+      measure([&] {
+        std::array<sml::utility::bool_task, children_count> 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_child, children_count, 10000>("noop");
+  run_case<sleep_child, children_count, 1000>("10us work");
+#endif
+}