diff --git a/.gitignore b/.gitignore index f631b4d..6b6d924 100644 --- a/.gitignore +++ b/.gitignore @@ -12,3 +12,4 @@ cmake_install.cmake compile_commands.json Build build/ +.github/ diff --git a/docs/IMPLEMENTATION_PLAN_runtime-state.md b/docs/IMPLEMENTATION_PLAN_runtime-state.md index 2d9053d..d0826f2 100644 --- a/docs/IMPLEMENTATION_PLAN_runtime-state.md +++ b/docs/IMPLEMENTATION_PLAN_runtime-state.md @@ -1,6 +1,6 @@ -# Runtime State Engine Implementation Plan +# Runtime State Manager Implementation Plan -This document is a step-by-step TDD-first implementation plan for the runtime state engine feature. It maps spec items to phased implementation tasks. Follow the phases sequentially and run unit tests after each phase. +This document is a step-by-step TDD-first implementation plan for the runtime state manager feature. It maps spec items to phased implementation tasks. Follow the phases sequentially and run unit tests after each phase. Repository layout assumptions - include/: public headers @@ -8,46 +8,110 @@ Repository layout assumptions - test/: unit tests - CMake macros like ADD_TEST_EXECUTABLE are available and used for registering test executables. -Phase 0: Already completed (spec & header) -- `docs/specs/runtime-state-machine-spec.md` updated with ownership, run(onError), cancel, wait(timeout)/shared_future, priority/timed variants and tests. -- Public header `include/dmn-runtime-state.hpp` added declaring the API. - -Phase 1: Add test skeletons and minimal stubs to compile -- Add test: `test/dmn-test-runtime-state.cpp` (skeleton) -- Add CMake: include `dmn-test-runtime-state` in `test/CMakeLists.txt` -- Add Phase-1 stub: `src/dmn-runtime-state.cpp` implementing minimal methods (compile-friendly): - - Dmn_Runtime_State ctor/dtor - - run() returns false - - cancel() is a no-op - - wait() returns immediately - - getFuture() returns ready shared_future +Phase 0: API contract and header preparation +- `docs/specs/runtime-state-machine-spec.md` defines ownership, run(onError), + cancel, wait(timeout)/shared_future, priority/timed variants, and tests. +- `Dmn_Runtime_State_Manager` uses the inherited + `Dmn_Singleton::createInstance()` factory and + therefore returns `std::shared_ptr`. It must not + declare a conflicting reference-returning factory. +- `Dmn_Runtime_Manager::isRunInAsyncThread()` is part of the public runtime + API so runtime-state can reject run()/wait()/wait_for() calls from the async + thread. +- Update `include/dmn-runtime-state.hpp` to reflect these contracts before + adding its implementation. +- The selected lifecycle contract is: a pre-run shared future remains pending; + no configured state makes run() return false without terminalizing; cancel() + before run() terminalizes as cancelled; failed futures rethrow the captured + exception from get(); and runtime-thread run()/wait()/wait_for() calls throw + in all build configurations. + +Phase 1: Construct the singleton manager (complete) +- Correct `include/dmn-runtime-state.hpp` so + `Dmn_Runtime_State_Manager` has a protected constructor, public destructor, + and friends `Dmn_Singleton`. Do not declare a + conflicting `createInstance()` method; use the inherited shared-pointer + factory. The public destructor is required by the singleton's default + `std::shared_ptr` deleter. +- Add `src/dmn-runtime-state.cpp` containing the manager constructor and + destructor definitions. +- Add `test/dmn-test-runtime-state.cpp`, with a focused unit test that + calls `Dmn_Runtime_State_Manager::createInstance()`, verifies the returned + shared pointer is non-null, and verifies repeated calls return the same + manager address. +- Add `dmn-test-runtime-state` to `test/CMakeLists.txt`. +- Add `src/dmn-runtime-state.cpp` and `include/dmn-runtime-state.hpp` to the + `dmn` target in + `src/CMakeLists.txt`. Add the header to `include/dmn.hpp`. Verify: - cmake -B build -DCMAKE_BUILD_TYPE=Debug - cmake --build build -- ctest --test-dir build --output-on-failure +- ctest --test-dir build -R dmn-test-runtime-state --output-on-failure + +Completed follow-on increment: State-handle creation +- Add the `DmnRuntimeStatePtr` alias for + `std::shared_ptr`. +- Implement `Dmn_Runtime_State_Manager::createState(std::string_view)` to + construct and return a new `Dmn_Runtime_State`. +- Define the runtime-state constructor, destructor, and default no-op + lifecycle hooks required to link the concrete polymorphic type. +- Extend `dmn-test-runtime-state` to verify `createState()` returns a + non-null handle, that `Dmn_Runtime_State` derives from `Dmn_State`, and + that inherited state configuration and stepping remain accessible for + compatibility testing. +- Remove the shadowing `Dmn_Runtime_State` declarations of `setStateFnc()`, + `setNext()`, and `setEnd()`. `runNext()` is re-exposed as protected and is + available to `Dmn_Runtime_State_Manager` through friendship. +- Add `Dmn_State::hasStateFncs()` as a public query for whether the client + configured at least one state function, excluding the internal + initialization function. +- Do not retain created states in the manager yet. Retention begins only when + a later `run()` implementation queues a state. + +Phase 2: Terminal-state primitive and lifecycle unit tests (complete) +- Do not make the manager advance state transitions implicitly: it controls + when `runNext()` executes, while a state function uses its `Dmn_State &` + parameter to call `setNext()` or `setEnd()`. +- Require clients to finish configuring state functions before successful + submission, because configuration is not synchronized with runtime execution. +- Implement the completion promise/shared_future pair, terminal flags, and a + single idempotent terminal transition helper. +- Implement the selected no-state and cancel-before-run behavior. +- Add targeted tests for a pending pre-run future, rejected unconfigured + run(), and cancel-before-run terminalization. -Phase 2: Basic runtime enqueue & single-step execution -- Implement run() to atomically set queued flag and enqueue a Dmn_Runtime_Job to `Dmn_Runtime_Manager::addJob()` (immediate) or `addTimedJob()` (delay). Use Dmn_Runtime_Job::Priority. -- Engine will retain an internal shared_ptr to the state while queued; store it in `std::unordered_map> m_pendingStates;` keyed by pointer or generated id. -- The job's m_fnc must create a coroutine task (TaskFncType) that: +Verify: +- cmake -B build -DCMAKE_BUILD_TYPE=Debug +- cmake --build build +- ctest --test-dir build -R dmn-test-runtime-state --output-on-failure + +Phase 3: Basic runtime enqueue & single-step execution (complete) +- Implement run() to set the mutex-protected queued flag and enqueue a + Dmn_Runtime_Job to `Dmn_Runtime_Manager::addJob()` (immediate) or + `addTimedJob()` (initial delay). Use Dmn_Runtime_Job::Priority. +- The manager retains an internal shared_ptr to the state while queued or + running in `std::unordered_map m_pendingStates`. +- The job's m_fnc creates a coroutine task (TaskFncType) that: - locks a weak_ptr to the state - checks isCancelled(); if set, call setEnd() and finalize - calls runNext() once (in try/catch) - - if still active, repost by calling addJob() again - - if terminal, set completion promise and erase engine internal shared_ptr + - if still active, repost immediately by calling addJob() again + - if terminal, set completion promise and erase manager internal shared_ptr - Wire `m_completionPromise` and `m_completionSharedFuture` so getFuture() returns `m_completionSharedFuture`. Tests expected to pass after this phase: - RuntimeState_BasicFlow - RuntimeState_GetFuture_PreRun_MultipleWaiters (shared_future works) -Phase 3: Exception capture and onError forwarding +Phase 4: Exception capture and onError forwarding (complete) - Wrap runNext() call in try/catch inside the runtime job. - On exception: - store `std::current_exception()` in the state - set `m_failed` flag - - set `m_completionPromise` (set_exception or set_value, but capture ep for diagnostics) + - set `m_completionPromise` with the captured exception so + `getFuture().get()` rethrows it - invoke onError callback forwarded via job.m_onErrorFnc - Update run() to forward client-provided onError into the runtime job creation @@ -55,25 +119,45 @@ Tests expected to pass: - RuntimeState_RunOnErrorCallback - state_exception_marks_failed -Phase 4: Cancel semantics & destructor-while-queued -- Implement cancel() to set atomic m_cancelled. -- Ensure runtime job checks m_cancelled before runNext() and calls setEnd() if true. -- Ensure engine internal shared_ptr map is created when run() enqueues; it must hold the shared_ptr until terminal. -- Implement destructor_while_queued test to validate engine holds state alive. - -Phase 5: Priority/timed behavior and fairness -- Implement run(priority, delay) mapping to addJob/addTimedJob. If delay > 0 use addTimedJob. -- Add tests verifying that priority ordering affects execution order. -- Consider fairness: ensure engine uses runtime priority queues and doesn't monopolize the runtime. - -Phase 6: Runtime-thread detection & runtime safety -- Detect runtime context using `Dmn_Runtime_Manager::isRunInAsyncThread()`. -- In run() and wait(), if called on runtime thread: assert in debug builds and throw `std::runtime_error` in release builds. -- Implement safe unit/integration tests for detection (special harness that posts a runtime job which attempts to call wait() and expects an exception). - -Phase 7: Polish, stress tests, documentation -- Add stress tests, runtime integration tests, and code comments. -- Document known limitations and example usage. +Phase 5: Complete lifecycle and scheduling coverage (complete) +- Added focused named Google Test cases for singleton/state creation, + unconfigured and pre-run cancellation behavior, normal execution, failure + propagation, queued cancellation, manager-retained lifetime, priority + ordering, delayed initial submission, and runtime-thread rejection. +- The queued-cancellation test verifies no user-defined step executes after + cancellation and that the inherited `Dmn_State` is finalized. +- The retained-lifetime test verifies a client can release its handle after + submission and that the manager releases its final ownership after terminal + completion. +- The priority and delay tests verify `run(priority, delay, onError)` maps + correctly to runtime scheduling behavior. +- The runtime-thread test verifies `run()`, `wait()`, and `wait_for()` throw + `std::runtime_error` from the runtime async thread. + +Phase 6: Drain-and-cancel manager shutdown (complete) +- Added `Dmn_Runtime_State_Manager::shutdown()`, which permanently rejects new + state submissions while allowing callers to create non-runnable handles. +- Shutdown snapshots the manager-retained handles, requests cooperative + cancellation outside the manager mutex, and waits for all captured states to + reach terminal cancellation before returning. +- A state step already executing may finish its callback, but its terminal + outcome is cancellation when shutdown requested it. Queued states finalize + without running another user-defined callback. +- Shutdown is idempotent and rejects calls from the runtime async thread to + avoid deadlock. +- Added coverage for drain waiting, running and queued state cancellation, and + rejection of a post-shutdown submission. + +Phase 7: Integration, stress, and documentation (complete) +- Added multi-state integration coverage for serialized execution and runtime + async-thread affinity, plus failure-isolation coverage for independent + queued states. +- Added concurrent client coverage for create/run/cancel/getFuture/wait + operations across 24 states, and shutdown stress coverage for 32 queued + states behind a running callback. +- Added a public usage example that documents runtime initialization from the + main thread, explicit state-manager shutdown while the runtime loop is + active, and runtime shutdown only after state draining completes. Developer checklist for each commit - Keep commits small and focused. @@ -81,14 +165,17 @@ Developer checklist for each commit - Run `ctest --test-dir build --output-on-failure` after each phase and fix failing tests or update the Phase implementation accordingly. Notes and gotchas -- Use weak_ptr in runtime job to avoid reference cycles; the engine's internal shared_ptr keeps the object alive while queued. -- Use atomic compare_exchange to set queued flag and avoid races for multiple-concurrent run() calls. +- Use weak_ptr in runtime job to avoid reference cycles; the manager's internal shared_ptr keeps the object alive while queued. +- Use the state mutex to set the queued flag and avoid races for multiple + concurrent run() calls. - Use std::shared_future to support multiple waiters. -- Be careful to release engine internal shared_ptr only after the completion promise is fulfilled and after finalization is complete. +- Be careful to release manager internal shared_ptr only after the completion promise is fulfilled and after finalization is complete. - Use runtime's addJob/addTimedJob APIs and forward onError callback using Dmn_Runtime_Job::OnErrorFncType. +- The manager exposes one drain-and-cancel shutdown mode and no + concurrency-configuration API; state steps execute in the process-wide + runtime async context. Example commands - Configure & build: cmake -B build -DCMAKE_BUILD_TYPE=Debug - Build: cmake --build build -j$(nproc) - Run tests: ctest --test-dir build --output-on-failure - diff --git a/docs/specs/runtime-state-machine-plan.md b/docs/specs/runtime-state-machine-plan.md index ce93913..64b0964 100644 --- a/docs/specs/runtime-state-machine-plan.md +++ b/docs/specs/runtime-state-machine-plan.md @@ -1,41 +1,80 @@ -# Implementation Plan: Runtime State Engine +# Implementation Plan: Runtime State Manager ## 1. Goal -Implement the runtime state engine feature as a singleton runtime-owned state machine manager built on the existing `dmn-runtime` and `dmn-state` components. +Implement the runtime state manager feature as a singleton runtime-owned state machine manager built on the existing `dmn-runtime` and `dmn-state` components. ## 2. Core Design Decisions -### Decision 1: engine is singleton and runtime-owned -The engine follows the same singleton model as `Dmn_Runtime_Manager`. It owns the execution policy for state objects and routes state execution through the runtime scheduler. +### Decision 1: manager is singleton and runtime-owned +The manager follows the same singleton model as `Dmn_Runtime_Manager`. It owns the execution policy for state objects and routes state execution through the runtime scheduler. ### Decision 2: state objects subclass `Dmn_State` Each runtime state object remains a state machine, but adds async lifecycle metadata and `wait()` support. This keeps the base state semantics while adding runtime execution ownership. ### Decision 3: all state execution is serialized -The engine does not allow independent parallel execution of state steps across runtime state objects. It posts work to the runtime scheduler in serialized form. +The manager does not allow independent parallel execution of state steps across runtime state objects. It posts work to the runtime scheduler in serialized form. ### Decision 4: `run()` is async-only The client never directly executes state logic in its own thread. `run()` only queues runtime tasks. The runtime executes each state step and then re-posts the next task until the machine is complete. -## 3. Phase 1: Define the engine and object model +## 3. Phase 1: Construct the manager singleton (complete) ### Tasks -- define `Dmn_Runtime_State_Engine` singleton API -- define `Dmn_Runtime_State` subclass of `Dmn_State` -- add runtime lifecycle flags (`queued`, `running`, `completed`, `failed`, `cancelled`) -- add `wait()` synchronization primitive -- confirm object ownership and destroy semantics +- define `Dmn_Runtime_State_Manager` singleton API +- use the inherited `Dmn_Singleton` shared-pointer `createInstance()` factory +- grant `Dmn_Singleton` access to the protected + manager constructor through a friend declaration +- add a focused unit test that constructs the singleton, verifies the returned + shared pointer is non-null, and verifies repeated calls return the same + manager instance +- this phase deliberately deferred lifecycle flags, waiting, scheduling, and + ownership retention to later phases + +### Deliverables + +- public manager class declaration +- manager constructor/destructor implementation +- registered `dmn-test-runtime-state` target +- singleton construction test + +### Completed follow-on increment: Create state handles + +- define `DmnRuntimeStatePtr` as `std::shared_ptr` +- implement `Dmn_Runtime_State_Manager::createState()` to return a newly + constructed concrete runtime state +- define the runtime-state constructor, destructor, and default no-op + lifecycle hooks +- extend the runtime-state test to verify non-null state creation and + `Dmn_State` inheritance and compatibility stepping +- remove shadowing declarations of `setStateFnc()`, `setNext()`, and + `setEnd()` so clients use the inherited `Dmn_State` API; re-expose + `runNext()` as protected and grant the manager friend access +- add public `Dmn_State::hasStateFncs()` to identify whether the client + configured at least one state function +- retain a manager-owned state handle after successful runtime queueing and + release it after terminal completion + +## 4. Phase 2: Implement the completion model (complete) + +### Tasks + +- keep transition selection in state functions through their `Dmn_State &` + parameter; the runtime manager controls when `runNext()` executes and + reposts work, not which transition is selected +- add runtime lifecycle flags (`queued`, `running`, `completed`, `failed`, + `cancelled`) +- add promise/shared-future completion synchronization for `wait()` and + `wait_for()` +- implement pre-run rejection and cancel-before-run terminal semantics ### Deliverables -- public engine class declaration -- public state class declaration - lifecycle state model -- wait mechanism design +- completion waiting design -## 4. Phase 2: Integrate with runtime scheduler +## 5. Phase 3: Integrate with runtime scheduler (complete) ### Tasks @@ -47,18 +86,17 @@ The client never directly executes state logic in its own thread. `run()` only q ### Deliverables - runtime job adapter for state objects -- serialized engine dispatcher +- serialized manager dispatcher - sequential execution loop -## 5. Phase 3: State lifecycle and completion +## 6. Phase 4: State lifecycle and completion (complete) ### Tasks -- implement transitioned completion behavior -- implement failed state handling for thrown exceptions -- implement cancellation and shutdown handling -- notify waiters exactly once +- implement terminal completion, failure, and cancellation behavior +- notify waiters exactly once through the stored shared future - avoid re-enqueuing after terminal state +- defer shutdown-mode API and tests to the shutdown phase ### Deliverables @@ -66,40 +104,54 @@ The client never directly executes state logic in its own thread. `run()` only q - exception-safe cleanup - completion synchronization contract -## 6. Phase 4: API ergonomics and compatibility +## 7. Phase 5: Complete lifecycle and scheduling test coverage (complete) ### Tasks -- keep `Dmn_State` unchanged for synchronous scenarios -- provide a runtime-specific object for async execution -- preserve `setStateFnc()`, `setNext()`, `setEnd()`, and `runNext()` semantics -- document `wait()` semantics and restrictions +- Added named Google Test cases for cancellation while queued, manager-retained + lifetime after the client drops its handle, priority ordering, timed initial + submission, and runtime-thread rejection. +- The runtime-thread harness verifies `run()`, `wait()`, and `wait_for()` + throw `std::runtime_error` from the runtime async thread. +- The test suite verifies runtime-managed states retain `Dmn_State` + configuration and transition semantics. ### Deliverables -- developer-facing API contract -- usage examples for initialization, run, and wait -- compatibility note for existing runtime and state users +- independently reported lifecycle and scheduling tests +- regression coverage for state lifetime and runtime-thread safety -## 7. Phase 5: Validation +## 8. Phase 6: Drain-and-cancel manager shutdown (complete) -### Tests to add +### Tasks + +- Added `Dmn_Runtime_State_Manager::shutdown()`. +- Reject new runs after shutdown begins while preserving the existing + non-null `createState()` factory behavior. +- Cancel retained states cooperatively and wait for terminal cancellation + outside the manager mutex. +- Reject shutdown from the runtime async thread and release retained manager + handles as states become terminal. + +### Required tests -- state object created from engine -- multiple state objects serialized in order -- `run()` enqueues runtime work and completes successfully -- waiting on completion returns after all steps finish -- exception sets failed terminal state -- repeated run is rejected or ignored as designed -- cancellation during queued or running state does not corrupt runtime +- shutdown waits for a running state callback to complete +- queued states are cancelled without running their user callback +- post-shutdown submission is rejected +- completion-future notification and manager-retention cleanup -### Validation commands +## 9. Phase 7: Integration and documentation (complete) + +### Tasks -- `cmake -B build -DCMAKE_BUILD_TYPE=Debug` -- `cmake --build build` -- `ctest --test-dir build --output-on-failure` +- Added multi-state serialization and failure-isolation integration tests. +- Added concurrent lifecycle coverage for state creation, submission, + cancellation, future retrieval, and waiting. +- Added shutdown stress coverage for a running state and 32 queued states. +- Added user documentation for runtime initialization, configuration, + submission, completion waiting, state-manager shutdown, and runtime exit. -## 8. Risks and Checkpoints +## 10. Risks and Checkpoints ### Risk: one state object re-enters itself Checkpoint: ensure `run()` only posts a single pending job and does not recursively run a state before the previous task finishes. @@ -110,23 +162,23 @@ Checkpoint: `wait()` must never run inside the runtime async thread; it must blo ### Risk: queue corruption during failure/cancel Checkpoint: all failed/cancelled states must terminate the loop cleanly and never re-post further runtime tasks. -## 9. Definition of Ready +## 11. Definition of Ready Implementation can begin once: -- the engine singleton contract is approved +- the manager singleton contract is approved - the state object subclass behavior is approved - the serialization rules are agreed - `wait()` and failure semantics are documented -## 10. Definition of Done +## 12. Definition of Done The feature is done when: -- the runtime state engine is implemented as a singleton +- the runtime state manager is implemented as a singleton - runtime state objects subclass `Dmn_State` - `run()` schedules async execution through `Dmn_Runtime_Manager` -- all state objects are serialized through the runtime engine +- all state objects are serialized through the runtime manager - client `wait()` supports async completion tracking - failure and cancellation paths are verified by tests - the library remains backward compatible with existing runtime/state APIs diff --git a/docs/specs/runtime-state-machine-spec.md b/docs/specs/runtime-state-machine-spec.md index 3d343ec..72f278d 100644 --- a/docs/specs/runtime-state-machine-spec.md +++ b/docs/specs/runtime-state-machine-spec.md @@ -1,34 +1,50 @@ -# Feature Spec: Runtime State Engine +# Feature Spec: Runtime State Manager -Status: Draft +Status: Draft — Phases 1-7 implemented. + +## Implementation Status + +The current implementation provides the singleton manager, managed state +handles, completion futures, runtime scheduling, manager-held lifetime, +one-shot submission, cooperative cancellation, failure capture, and runtime +error callback forwarding. `run()` supports priority and an initial delay; +later steps are reposted immediately at the submitted priority. + +`run()`, `wait()`, and `wait_for()` reject calls from the runtime async thread. +Focused tests cover queued cancellation, manager-retained lifetime, priority +ordering, timed initial submission, runtime-thread rejection, +drain-and-cancel shutdown, multi-state serialization, failure isolation, and +concurrent lifecycle operations. The manager exposes one shutdown mode and no +configurable concurrency. The current runtime architecture serializes all +state steps through the process-wide runtime async thread. ## 1. Summary -This feature introduces a new runtime-owned state execution engine that combines the existing `dmn-runtime` scheduler with the `dmn-state` finite-state helper. The result is a singleton runtime service that creates state objects for clients, serializes their execution through the runtime, and lets callers wait for completion without forcing the client thread to execute each state step directly. +This feature introduces a new runtime-owned state execution manager that combines the existing `dmn-runtime` scheduler with the `dmn-state` finite-state helper. The result is a singleton runtime service that creates state objects for clients, serializes their execution through the runtime, and lets callers wait for completion without forcing the client thread to execute each state step directly. The new feature preserves the library’s current design philosophy: - `Dmn_Runtime_Manager` remains the process-wide scheduler and signal manager. - `Dmn_State` remains the lightweight state-machine primitive. -- the new runtime state engine owns the scheduling and serialized execution policy. +- the new runtime state manager owns the scheduling and serialized execution policy. - each state object is a runtime-managed, asynchronously executed state machine instance. The primary behavior is: -1. client obtains a state handle from the `Dmn_Runtime_State_Engine` singleton; +1. client obtains a state handle from the `Dmn_Runtime_State_Manager` singleton; 2. client configures the state(s) on that object; 3. client calls `statehandle->run()` (optionally with priority / delay / onError handler); -4. `run()` enqueues a runtime task into the runtime engine and returns a boolean indicating whether the enqueue succeeded; -5. the runtime engine continues to repost tasks until the state object reaches its terminal condition; errors occurring in async execution invoke a client-provided onError callback (if supplied) and are captured on the state handle; -6. all state objects created by the engine are executed in serialized order through `Dmn_Runtime_Manager` (subject to the engine's serialization policy); +4. `run()` enqueues a runtime task into the runtime manager and returns a boolean indicating whether the enqueue succeeded; +5. the runtime manager continues to repost tasks until the state object reaches its terminal condition; errors occurring in async execution invoke a client-provided onError callback (if supplied) and are captured on the state handle; +6. all state objects created by the manager are executed in serialized order through `Dmn_Runtime_Manager` (subject to the manager's serialization policy); 7. client may call `statehandle->wait()` or use the returned shared_future to block or asynchronously observe completion. ## 2. Design Objective -The existing `dmn-state` component is synchronous and client-driven. It calls `runNext()` directly in the caller thread. That is useful for local control flow, but not for runtime-managed workflows. The new runtime state engine changes the ownership model: +The existing `dmn-state` component is synchronous and client-driven. It calls `runNext()` directly in the caller thread. That is useful for local control flow, but not for runtime-managed workflows. The new runtime state manager changes the ownership model: - the state object remains a state machine definition and execution state, -- the runtime engine owns when the state steps are executed, +- the runtime manager owns when the state steps are executed, - state execution is serialized within the runtime’s async context, - the client receives an async completion signal via `wait()` or a shared_future rather than manually stepping the machine. @@ -38,11 +54,11 @@ This makes the feature a natural fit for handshake flows, retries, startup/teard ### In Scope -- singleton runtime state engine class modeled after `Dmn_Runtime_Manager` -- state objects returned from the engine as managed handles (shared ownership) and that subclass `Dmn_State` +- singleton runtime state manager class modeled after `Dmn_Runtime_Manager` +- state objects returned from the manager as managed handles (shared ownership) and that subclass `Dmn_State` - state registration API for user-defined states -- `run()` scheduling through the runtime engine with onError callback forwarding to `dmn-runtime` semantics -- serialized execution of all runtime state objects (engine-global by default) +- `run()` scheduling through the runtime manager with onError callback forwarding to `dmn-runtime` semantics +- serialized execution of all runtime state objects (manager-global by default) - `wait()` completion API for async execution with optional timeout and a shared_future-based async alternative - shutdown, error, cancellation, and terminal-state handling - tests for startup, normal completion, errors, cancellation, and lifetime edge cases @@ -63,35 +79,51 @@ Key runtime types and semantics reused: - `Dmn_Runtime_Job::Priority` and `Dmn_Runtime_Manager::addJob()` / `addTimedJob()` - `Dmn_Runtime_Job::OnErrorFncType` (signature: `std::function`) — the onError callback type used by runtime jobs -- `Dmn_Runtime_Manager` owns the singleton async thread and provides `isRunInAsyncThread()` to detect runtime-thread context +- `Dmn_Runtime_Manager` owns the singleton async thread and exposes the public + `isRunInAsyncThread()` query to detect runtime-thread context. This query + does not alter scheduling state and is required by runtime-managed clients + to reject operations that would deadlock the runtime thread. ## 5. Functional Requirements -### FR-1: Runtime state engine existence +### FR-1: Runtime state manager existence -A singleton class named `Dmn_Runtime_State_Engine` must exist and follow the same singleton creation conventions as `Dmn_Runtime_Manager`. +A singleton class named `Dmn_Runtime_State_Manager` must exist and follow the same singleton creation conventions as `Dmn_Runtime_Manager`. -The engine must: +The manager must: -- provide `createInstance()` or equivalent singleton factory consistent with `Dmn_Singleton` -- own a runtime-managed execution queue for state objects +- use the inherited + `Dmn_Singleton::createInstance()` factory, which + returns `std::shared_ptr` +- retain submitted state handles while routing their work through the + process-wide runtime scheduler - ensure all state execution is scheduled through `Dmn_Runtime_Manager` +The manager must not declare a same-named `createInstance()` with a different +return type. A forwarding convenience function is permitted only when it has a +distinct name and preserves the singleton's shared ownership semantics. + +The implemented manager and state handle live in +`include/dmn-runtime-state.hpp` and `src/dmn-runtime-state.cpp`; the +`dmn-test-runtime-state` target exercises the currently implemented baseline. +Manager shutdown modes and broader multi-state integration/stress coverage +remain outstanding. + ### FR-2: Client-managed state object creation and ownership -Clients must be able to obtain a managed handle to a state object from the runtime state engine. +Clients must be able to obtain a managed handle to a state object from the runtime state manager. Ownership model (required): - `createState()` MUST return a managed handle type: `std::shared_ptr` (alias `DmnRuntimeStatePtr`) following the existing dmn pattern used by other components (for example, `dmn-dmesg`). -- The engine MUST retain a `std::shared_ptr` to the state object while it is queued or running. This guarantees the object remains alive until it reaches a terminal state even if the client drops its handle. -- When the object becomes terminal (completed/failed/cancelled), the engine releases its internal shared_ptr; any remaining client-held shared_ptr keeps the object alive until all references are dropped. -- Clients may intentionally drop their handle to rely on engine ownership for fire-and-forget semantics. +- The manager MUST retain a `std::shared_ptr` to the state object while it is queued or running. This guarantees the object remains alive until it reaches a terminal state even if the client drops its handle. +- When the object becomes terminal (completed/failed/cancelled), the manager releases its internal shared_ptr; any remaining client-held shared_ptr keeps the object alive until all references are dropped. +- Clients may intentionally drop their handle to rely on manager ownership for fire-and-forget semantics. The resulting object must: - be a concrete state object type derived from `Dmn_State` -- be created from the runtime state engine singleton and returned as a shared_ptr handle +- be created from the runtime state manager singleton and returned as a shared_ptr handle - carry runtime-managed lifecycle metadata - be configured by calling `setStateFnc()` or equivalent state registration methods @@ -99,11 +131,26 @@ The resulting object must: A client must be able to define one or more states on the returned runtime state object via the compatible `Dmn_State` interface. -The engine must support: +`Dmn_Runtime_State` MUST inherit, rather than redeclare or override, +`Dmn_State::setStateFnc()`, `setNext()`, and `setEnd()`. State functions +therefore retain the base callback signature, `std::function`. + +Clients configure state functions before calling `run()`. They do not directly +advance or terminate the machine from outside a state function: the runtime +manager controls when `runNext()` executes and whether another job is posted. +A state function uses its `Dmn_State &` parameter to call `setNext()` or +`setEnd()` when it needs to select the next transition or terminate the +machine, preserving existing `Dmn_State` semantics. + +`Dmn_State::hasStateFncs()` is a public query that returns true when at least +one client-defined state function exists. It excludes the internal +initialization function and is used by `Dmn_Runtime_State::run()` to reject an +unconfigured state without terminalizing it. -- sequential state definition by appending state functors -- explicit transition to next state via `setNext()` and `setEnd()` semantics -- state functors that are valid in the same pattern as `Dmn_State` +Configuration must be complete before a successful `run()` call. Modifying +state functions after submission is unsupported because the runtime may +execute `runNext()` concurrently with the client thread. ### FR-4: `run()` dispatches work to runtime and error callback forwarding @@ -113,30 +160,42 @@ Behavior: - `run()` may be called from any client thread EXCEPT the runtime async thread (calls from the runtime thread are disallowed — see `run()` thread policy below). - `run()` must schedule asynchronous work on the singleton `Dmn_Runtime_Manager` async thread by calling `addJob()` or `addTimedJob()` as appropriate. -- `run()` returns `true` if the state was successfully queued (or started) and `false` if the enqueue failed (invalid state, already terminal, or internal error). +- `run()` returns `true` if the state was successfully queued and `false` when + the state is unconfigured, cancelled, terminal, already active, or the + manager has shut down. Runtime submission exceptions propagate after the + state clears its queued marker. - `run()` is a one-shot operation for each state handle: subsequent `run()` calls after the first successful enqueue MUST be no-ops and MUST return `false`. - The optional `onError` callback provided to `run()` MUST be forwarded to the underlying `dmn-runtime` job so that asynchronous runtime failures invoke the client callback when the runtime job reports an error. Use `Dmn_Runtime_Job::OnErrorFncType` as the canonical type. -- The runtime work must execute exactly one next state step of the state object per posted job; after the step executes, the engine reposts another job (or finalizes) until terminal. +- The runtime work must execute exactly one next state step of the state object per posted job; after the step executes, the manager reposts another job (or finalizes) until terminal. - `run()` MUST NOT execute state logic synchronously in the caller thread. Priority and timed variants: -- `run()` must accept an optional `Dmn_Runtime_Job::Priority` parameter and an optional timeout/delay parameter so callers can control priority and optionally schedule timed runs. The engine will map these directly to `Dmn_Runtime_Manager::addJob()` (immediate) or `addTimedJob()` (timed). +- `run()` must accept an optional `Dmn_Runtime_Job::Priority` parameter and an + optional delay so callers can control priority and schedule a timed initial + run. The manager maps these directly to `Dmn_Runtime_Manager::addJob()` + (immediate) or `addTimedJob()` (timed). -Thread policy for `run()` and `wait()`: +Thread policy for `run()`, `wait()`, and `wait_for()`: -- If `run()` or `wait()` is called from inside the runtime async thread (detected via `Dmn_Runtime_Manager::isRunInAsyncThread()`), the implementation MUST assert in debug builds and throw `std::runtime_error` (or similar) in non-debug builds. This prevents deadlocks and enforces the rule that runtime-internal callbacks should not block the runtime. +- If `run()`, `wait()`, or `wait_for()` is called from inside the runtime async + thread (detected via `Dmn_Runtime_Manager::isRunInAsyncThread()`), the + implementation MUST throw `std::runtime_error`. This prevents deadlocks and + enforces the rule that runtime-internal callbacks should not block the + runtime. +- This policy applies in every build configuration, permitting direct, + deterministic unit testing without a debug-only death test. ### FR-5: Serialized execution across all state objects -All state objects created from the runtime state engine must execute in serialized form through the shared runtime scheduler by default. +All state objects created from the runtime state manager must execute in serialized form through the shared runtime scheduler by default. Requirements: -- no two state objects may run their next step concurrently in the same runtime engine (default global serialization) +- no two state objects may run their next step concurrently in the same runtime manager (default global serialization) - state object execution order must follow runtime job ordering/priority semantics -- state step tasks must be single-threaded relative to engine execution -- the engine MAY provide configuration for relaxed concurrency (optional extension) but default behavior must be serialized to match the spec +- state step tasks must be single-threaded relative to manager execution +- the manager MAY provide configuration for relaxed concurrency (optional extension) but default behavior must be serialized to match the spec ### FR-6: Completion waiting via `wait()` and async alternatives @@ -146,8 +205,13 @@ Behavior: - `wait()` blocks until the state object reaches its terminal state - `wait()` MUST support an overload `wait_for(std::chrono::duration<...> timeout)` that returns a boolean indicating whether the wait observed terminal completion before the timeout -- `wait()` MUST NOT be called from the runtime async thread; calling from the runtime thread is undefined — implementations MUST assert (debug) and throw (release) when detected -- The object MUST provide a `std::shared_future getFuture()` so callers can use non-blocking or async wait patterns. The shared_future MUST be available immediately after state creation (before run()) and MUST become ready on terminal state regardless of whether run() was ever called. +- `wait()` MUST NOT be called from the runtime async thread; implementations + MUST throw `std::runtime_error` in every build configuration when detected +- The object MUST provide a `std::shared_future getFuture()` so callers + can use non-blocking or async wait patterns. The shared_future MUST be + available immediately after state creation (before run()) but remains pending + until the object reaches a terminal state. It becomes ready when execution + completes, fails, or is cancelled before or after submission. - Multiple threads/callers MAY wait concurrently on the same shared_future or call `wait()` concurrently; the implementation must support multiple waiters. ### FR-7: Terminal state and completion semantics @@ -176,26 +240,39 @@ The runtime state object MUST provide a `cancel()` method that is cooperative in Semantics: - `cancel()` is idempotent and sets the object's canceled flag and prevents further non-cooperative steps from being scheduled -- `cancel()` does NOT asynchronously preempt a currently executing state functor unless that functor explicitly checks a cancellation token exposed by the state and co-operates +- `cancel()` does NOT asynchronously preempt a currently executing state + functor. A callback that requires cooperative early exit may capture its + `Dmn_Runtime_State` handle and query `isCancelled()`. - when the runtime task executes and detects the cancel flag is set, it MUST call `setEnd()` before invoking `runNext()` so that the state finalizes deterministically rather than executing further steps -- the engine must provide a way for state functors to query the cancel flag (e.g. `isCancelled()` or a cancellation token) so they can early-exit and perform graceful cleanup -- on shutdown, the engine must support at least two modes: graceful (finish current step and queued states) and immediate (mark queued as cancelled and notify waiters). The exact mode shall be selectable via engine shutdown API. +- `isCancelled()` provides the cancellation query for callbacks that capture + their runtime-state handle and need to early-exit or perform cleanup +- calling `cancel()` before `run()` transitions the object to cancelled + terminal state immediately, completes its shared future, and causes any later + `run()` call to return false +- `Dmn_Runtime_State_Manager::shutdown()` provides the current + drain-and-cancel shutdown mode. It rejects new submissions, requests + cancellation for all submitted states, and waits for their terminal futures. + A currently executing callback may finish, but publishes cancellation when + shutdown requested it; queued states finalize without another user callback. + Calling shutdown from the runtime async thread throws `std::runtime_error`. ### FR-9: Error propagation and onError callback -If a state callback throws while running inside the runtime-managed async thread, the runtime state engine must: +If a state callback throws while running inside the runtime-managed async thread, the runtime state manager must: - capture the exception - set the state object to failed terminal state - notify any waiting client via `wait()` or shared_future - invoke the onError callback supplied to `run()` (if any) with the runtime's error details using `Dmn_Runtime_Job::OnErrorFncType` - avoid corrupting the runtime scheduler internal state +- complete the shared future with the captured exception so `getFuture().get()` + rethrows it ## 6. Non-Functional Requirements ### NFR-1: Singleton and runtime-thread ownership -The runtime state engine must preserve the runtime’s model: client code may call APIs from any thread, but the actual state execution must be marshaled to the runtime async thread. +The runtime state manager must preserve the runtime’s model: client code may call APIs from any thread, but the actual state execution must be marshaled to the runtime async thread. ### NFR-2: Backward compatibility @@ -203,57 +280,63 @@ This feature must not break the existing `Dmn_Runtime_Manager` or `Dmn_State` AP ### NFR-3: Determinism -State execution within one runtime engine must be deterministic with respect to queue ordering and posting order when preservation of ordering is requested by the client. +State execution within one runtime manager must be deterministic with respect to queue ordering and posting order when preservation of ordering is requested by the client. ### NFR-4: Controlled memory lifetime (explicit) -The engine MUST hold a `std::shared_ptr` to any queued/running state object until terminal state is reached. This mirrors existing dmn ownership patterns (see `dmn-dmesg` for a similar handle + engine-owned shared_ptr pattern). +The manager MUST hold a `std::shared_ptr` to any queued/running state object until terminal state is reached. This mirrors existing dmn ownership patterns (see `dmn-dmesg` for a similar handle + manager-owned shared_ptr pattern). ### NFR-5: Thread safety for wait semantics `wait()` must be implemented using synchronization primitives appropriate for cross-thread signaling (`std::condition_variable`, flag, or equivalent) and must not busy-spin. Use of a `std::shared_future` simplifies multi-waiter semantics. -## 7. Proposed API Shape +## 7. Current API Shape -This API is proposed to match the existing library naming and runtime conventions while reflecting the ownership, error/cancel, priority/timed semantics and thread-safety rules requested. +The following summarizes the currently implemented public API. It reflects the +library naming, ownership, error/cancellation, priority/timed, and thread-safety +contracts described in this specification. ```cpp namespace dmn { -class Dmn_Runtime_State_Engine - : public Dmn_Singleton { -public: - static auto createInstance() -> Dmn_Runtime_State_Engine &; - - class Dmn_Runtime_State; +class Dmn_Runtime_State; +using DmnRuntimeStatePtr = std::shared_ptr; - // handle type returned to clients. Follows existing dmn shared ownership pattern. - using DmnRuntimeStatePtr = std::shared_ptr; +class Dmn_Runtime_State_Manager + : public Dmn_Singleton { +public: + // Inherited factory: + // std::shared_ptr createInstance(); - // createState returns a shared_ptr handle. Engine will also keep a shared_ptr while the + // createState returns a shared_ptr handle. Manager will also keep a shared_ptr while the // state is queued or running to guarantee lifetime. DmnRuntimeStatePtr createState(std::string_view name); - // optional engine-level shutdown / configuration APIs omitted for brevity + // Cancel retained states and wait for terminal completion. New run() calls + // are rejected after shutdown begins. + void shutdown(); }; -class Dmn_Runtime_State : public Dmn_State { +class Dmn_Runtime_State + : public Dmn_State, + public std::enable_shared_from_this { + friend class Dmn_Runtime_State_Manager; + public: - using FncType = std::function; using OnErrorFnc = Dmn_Runtime_Job::OnErrorFncType; // std::function explicit Dmn_Runtime_State(std::string_view name); - // state configuration (same as Dmn_State) - void setStateFnc(FncType fnc, int index = 0); - void setNext(int index); - void setNext(); - void setEnd(); + // State configuration methods are inherited unchanged from Dmn_State. + // The runtime manager controls execution by calling protected runNext(). + +protected: + using Dmn_State::runNext; + +public: - // lifecycle APIs - // run: returns true when the enqueue succeeded; false on failure (already terminal, invalid, runtime busy) - // optional onError is forwarded to the runtime job so callers get notified of async job failure. - // priority and timed overloads are supported and map to addJob()/addTimedJob(). + // Lifecycle APIs. A successful run retains the state in the manager until + // it reaches a completed, failed, or cancelled terminal outcome. bool run(Dmn_Runtime_Job::Priority priority = Dmn_Runtime_Job::Priority::kMedium, const std::chrono::steady_clock::duration &delay = std::chrono::steady_clock::duration::zero(), OnErrorFnc onError = {}); @@ -271,7 +354,7 @@ public: std::shared_future getFuture(); // introspection - // isRunning() indicates the handle has been queued or is actively running inside the engine + // isRunning() indicates the handle has been queued or is actively running inside the manager bool isRunning() const; bool isCompleted() const; bool isFailed() const; @@ -284,23 +367,21 @@ protected: void onCancelled(); private: - // synchronization / state - std::mutex m_waitMutex; - std::condition_variable m_waitCv; + enum class Terminal_State { kCompleted, kFailed, kCancelled }; - // promise/future pair: keep a shared_future so multiple waiters are supported + // Synchronization and completion state. + std::mutex m_mutex; std::promise m_completionPromise; - std::shared_future m_completionSharedFuture; // initialized from m_completionPromise.get_future().share() - - std::atomic_bool m_running{false}; // set when a runtime task is active - std::atomic_bool m_completed{false}; // terminal - std::atomic_bool m_failed{false}; // terminal - std::atomic_bool m_cancelled{false}; // set by cancel() - std::atomic_bool m_queued{false}; // set before enqueue to avoid duplicates - - // captured failure for diagnostics - std::mutex m_failureMutex; - std::exception_ptr m_failureEp{nullptr}; + std::shared_future m_completionSharedFuture; + + std::exception_ptr m_failure; + bool m_queued{}; + bool m_running{}; + bool m_started{}; + bool m_completed{}; + bool m_failed{}; + bool m_cancelled{}; + bool m_terminal{}; }; } // namespace dmn @@ -308,13 +389,25 @@ private: ### API Notes -- `Dmn_Runtime_State_Engine::createState()` returns a `std::shared_ptr` handle. The engine retains a shared_ptr while the state is queued/running to ensure safe lifetime. -- `Dmn_Runtime_State::run(priority, delay, onError)` returns a boolean: true on successful enqueue, false if enqueue failed (e.g., already terminal or invalid state). If `delay` is non-zero, the engine must use `addTimedJob()`; otherwise `addJob()` is used. +- `Dmn_Runtime_State_Manager::createInstance()` is inherited from + `Dmn_Singleton` and returns `std::shared_ptr`. + Clients retain that manager handle while using it. +- `Dmn_Runtime_State_Manager::createState()` returns a + `std::shared_ptr` handle. The manager retains a shared_ptr + while the state is queued/running to ensure safe lifetime. +- `Dmn_Runtime_State::run(priority, delay, onError)` returns true on successful + enqueue and false when the state is unconfigured, cancelled, terminal, + already active, or its manager has shut down. A runtime submission exception + propagates after clearing the queued marker. If `delay` is non-zero, the + first job uses `addTimedJob()`; later state steps use `addJob()`. - `run()` accepts an optional onError callback that uses the runtime's `Dmn_Runtime_Job::OnErrorFncType` signature and is forwarded to the runtime job. - `run()` is one-shot: a successful `run()` prevents subsequent `run()` calls from enqueueing again; such subsequent calls return `false` (no-op). This avoids duplicate enqueues across threads. - `cancel()` is cooperative: it sets a cancelled flag. The runtime job, before calling `runNext()`, must check `isCancelled()` and call `setEnd()` if the state has been cancelled so that the state finalizes without executing further steps. - `wait()` supports a timeout variant and `getFuture()` returns a `std::shared_future` that can be used by multiple waiters. The shared_future is valid immediately after createState() is called and resolves when the state reaches a terminal condition. -- Calls to `run()` or `wait()` from inside the runtime async thread must assert/throw. Use `Dmn_Runtime_Manager::isRunInAsyncThread()` to detect and enforce this. +- Calls to `run()`, `wait()`, or `wait_for()` from inside the runtime async + thread must throw `std::runtime_error`. Use the public + `Dmn_Runtime_Manager::isRunInAsyncThread()` query to detect and enforce + this. ## 8. Execution Model @@ -322,33 +415,37 @@ private: A runtime state object has the following lifecycle: -1. Created by `Dmn_Runtime_State_Engine::createState()` and returned as a shared_ptr handle +1. Created by `Dmn_Runtime_State_Manager::createState()` and returned as a shared_ptr handle 2. Configured by setting state functors (`setStateFnc`, etc.) 3. Idle before `run()` is called -4. Queued for runtime execution after `run()` (engine retains shared_ptr and sets `m_queued` atomically) +4. Queued for runtime execution after `run()` (manager retains shared_ptr and + sets `m_queued` while holding the state mutex) 5. Running inside runtime async thread -6. Finalized once terminal condition reached (engine releases internal shared_ptr) +6. Finalized once terminal condition reached (manager releases internal shared_ptr) 7. Client may call `wait()` at any time after submission or use the shared_future returned by `getFuture()` -### 8.2 `run()` exact semantics and atomic queued flag +### 8.2 `run()` exact semantics and mutex-protected queued flag When `statehandle->run(priority, delay, onError)` is called: 1. the state object must be validated for legal execution -2. the implementation MUST atomically set the `m_queued` flag (e.g., using compare_exchange) to avoid races where multiple threads try to enqueue simultaneously; only the thread that successfully sets `m_queued` proceeds to request the engine to retain an internal shared_ptr and submit the runtime job +2. the implementation MUST set `m_queued` while holding the state mutex to + avoid races where multiple threads try to enqueue simultaneously; only the + thread that observes it unset proceeds to request the manager to retain an + internal shared_ptr and submit the runtime job 3. if `m_queued` was already true or the object is terminal, `run()` returns false (no-op) -4. the engine stores a shared_ptr to the object (ensuring lifetime) and schedules a runtime job via `addJob()` or `addTimedJob()` depending on `delay` +4. the manager stores a shared_ptr to the object (ensuring lifetime) and schedules a runtime job via `addJob()` or `addTimedJob()` depending on `delay` 5. the scheduled job executes exactly one next state step using the state object 6. before calling `runNext()`, the runtime job MUST check the cancel flag; if cancelled, it MUST call `setEnd()` and finalize instead of running the step -7. after the state step finishes, the runtime engine checks whether another state remains and reposts a new runtime job if appropriate -8. if no state remains, the object transitions to completed terminal state, the engine notifies waiters (set promise) and releases its internal shared_ptr +7. after the state step finishes, the runtime manager checks whether another state remains and reposts a new runtime job if appropriate +8. if no state remains, the object transitions to completed terminal state, the manager notifies waiters (set promise) and releases its internal shared_ptr 9. if a state callback throws, the exception is captured, the object transitions to failed, waiters are notified, and the optional onError callback is invoked -This loop continues until no more states to run. The runtime engine is responsible for re-posting tasks while the object remains active. +This loop continues until no more states to run. The runtime manager is responsible for re-posting tasks while the object remains active. ### 8.3 Serialization requirement and optional config -All state object tasks must be executed in serialized form through the runtime queue by default. This strict global serialization simplifies reasoning and matches the project's stated intent. Implementations SHOULD provide configuration for relaxed concurrency (e.g., per-engine worker count) as an optional extension, but that must be explicitly chosen and documented by the caller. +All state object tasks must be executed in serialized form through the runtime queue by default. This strict global serialization simplifies reasoning and matches the project's stated intent. Implementations SHOULD provide configuration for relaxed concurrency (e.g., per-manager worker count) as an optional extension, but that must be explicitly chosen and documented by the caller. ## 9. State Object Contract @@ -362,18 +459,25 @@ It must retain: - init/finalize behavior inherited from the base - default state sequencing model -The runtime layer adds async ownership, completion signaling, cancellation token, and error forwarding on top of the base semantics. +The runtime layer must not hide or redeclare the base configuration methods. +It adds async ownership, completion signaling, an `isCancelled()` query, and +error forwarding on top of the base semantics. The manager controls when +`runNext()` is invoked; state functions control their transitions with the +inherited `Dmn_State &` API. ### 9.2 Cancellation contract - `cancel()` sets the cancellation flag and is safe to call from any thread. - The runtime job MUST observe `isCancelled()` before executing `runNext()` and call `setEnd()` to force deterministic finalization. -- State functors MAY query `isCancelled()` if they want to implement cooperative cancellation. +- State functors that need to cooperate with cancellation may capture their + `Dmn_Runtime_State` handle and query `isCancelled()`; their callback + parameter remains `Dmn_State &` for base-API compatibility. ### 9.3 `wait()` behavior and deadlock avoidance - `wait()` blocks until the object is terminal. -- Implementations MUST detect calls from the runtime async thread and assert/throw as described. +- Implementations MUST detect calls from the runtime async thread and throw + `std::runtime_error` as described. - `wait_for(timeout)` returns a bool indicating whether the wait observed terminal completion before the timeout expired. - `getFuture()` returns a `std::shared_future` available immediately after creation and resolves on terminal state. @@ -381,7 +485,9 @@ The runtime layer adds async ownership, completion signaling, cancellation token ### 10.1 No state configured -If no state is defined before `run()`, the engine must not enqueue an invalid task. The object should either immediately transition to failed with a clear exception captured (and optional onError invoked), or the run() call should return false indicating invalid configuration. +If no state is defined before `run()`, the manager must not enqueue an invalid +task. `run()` returns false, leaves the state unsubmitted and non-terminal, and +does not invoke onError. ### 10.2 Repeated `run()` calls @@ -390,7 +496,9 @@ A runtime state object must not be run multiple times simultaneously. Behavior: - The first successful `run()` enqueues the object and returns true. -- Subsequent `run()` calls (concurrent or later) MUST return false (no-op). This avoids duplicate enqueues and is thread-safe due to the atomic `m_queued` guard. +- Subsequent `run()` calls (concurrent or later) MUST return false (no-op). + This avoids duplicate enqueues and is thread-safe due to the mutex-protected + `m_queued` guard. ### 10.3 Finalized or canceled states @@ -398,7 +506,7 @@ Once finalized, failed, or canceled, no further state step may be scheduled. ### 10.4 Exception propagation and onError -If a state callback throws while inside the runtime engine: +If a state callback throws while inside the runtime manager: - capture the exception in `std::exception_ptr` stored on the object - transition object to failed terminal state @@ -408,66 +516,65 @@ If a state callback throws while inside the runtime engine: ### 10.5 Shutdown race and modes -Engine shutdown must support at least two modes (configurable when shutting down): +The manager provides one drain-and-cancel shutdown mode: -- graceful: do not accept new state runs; let currently running state steps complete and drain queued states before finalizing them (clients may call wait() to observe completion) -- immediate: cancel queued (but not yet running) states and mark them cancelled; running steps proceed to completion or observe cancellation cooperatively +- shutdown rejects new `run()` submissions but preserves `createState()` as a + non-null handle factory; +- it requests cancellation for all manager-retained states and waits for their + completion futures without holding the manager mutex; +- queued states finalize as cancelled without another user callback; +- a running callback may return normally, but the runtime publishes + cancellation rather than completion once shutdown has requested it. -The spec requires tests for both modes. +The runtime main loop must remain active while shutdown drains queued work. ## 11. Serialization and Scheduling Contract -The engine is responsible for ensuring serialized processing across all state instances it creates (by default). Each queued job should trigger exactly one `runNext()` invocation and then repost if required. +The manager is responsible for ensuring serialized processing across all state instances it creates (by default). Each queued job should trigger exactly one `runNext()` invocation and then repost if required. -Provide clear priority mapping between engine jobs and other runtime jobs. The engine must not starve other runtime jobs; use the runtime's priority scheme and document how engine tasks are enqueued. +Provide clear priority mapping between manager jobs and other runtime jobs. The manager must not starve other runtime jobs; use the runtime's priority scheme and document how manager tasks are enqueued. ## 12. Test Plan (expanded) -### Unit Tests (additions focusing on the requested gaps) - -- `create_state_from_runtime_engine` -- `state_run_posts_runtime_job` -- `state_wait_completes_after_all_steps` -- `state_serialization_maintains_order` -- `state_exception_marks_failed` -- `state_cancel_marks_cancelled` -- `state_no_state_defined_fails_cleanly` -- `run_is_rejected_when_state_already_active` - -Additional edge-case tests (required): - -- `destructor_while_queued`: verify engine-held shared_ptr keeps object alive while queued and that finalization occurs correctly when the client drops its handle -- `wait_from_runtime_thread_detected`: ensure wait() from runtime thread asserts/throws -- `cancel_before_run_or_queued`: calling cancel() before a queued step prevents runNext() and finalizes the state -- `cancel_during_queued_execution`: cancellation while queued leads the runtime job to call setEnd() rather than executing further steps -- `run_onerror_callback_invoked`: ensure onError passed to run() is invoked when the runtime job fails -- `shutdown_graceful_vs_immediate`: test both shutdown modes and their effects on queued and running states -- `getFuture_shared_waiters_before_run`: verify multiple waiters using getFuture() before run() all get signaled on terminal condition -- `run_priority_and_timed_variant`: verify run(priority, delay) maps to addJob/addTimedJob and honors priority ordering - -### Integration Tests - -- run multiple runtime state objects in the same runtime engine -- verify that state steps are serialized in posting order -- verify wait() and shared_future complete after sequence completion -- verify runtime jobs remain valid when the state object reaches terminal condition -- stress test many queued states to measure the effect of global serialization - -### Stress / Regression Tests - -- repeated create/run/wait cycles -- many state objects queued sequentially -- failure followed by cleanup and reuse -- shutdown while state objects are queued +### Implemented Unit Tests + +`test/dmn-test-runtime-state.cpp` contains focused Google Test cases: + +- `CreatesSingletonManagerAndStateHandle` +- `RejectsUnconfiguredAndPreRunCancelledStates` +- `ExecutesStatesAndReportsStateFailures` +- `CancelsQueuedStateWithoutRunningUserStep` +- `RetainsStateUntilCompletionAfterClientHandleReleased` +- `HonorsPriorityOrdering` +- `DelaysInitialSubmission` +- `RejectsRunAndWaitOperationsFromRuntimeThread` +- `SerializesMultipleStateExecutions` +- `IsolatesStateFailureFromOtherQueuedStates` +- `HandlesConcurrentStateLifecycleOperations` +- `ShutdownCancelsPendingStatesAndRejectsNewSubmissions` + +Together, these cover creation, normal execution, future completion, failure +propagation and onError forwarding, pre-run and queued cancellation, +manager-retained lifetime, priority ordering, timed initial submission, +runtime-thread restrictions, multi-state serialization, failure isolation, +concurrent lifecycle operations, and shutdown draining. + +### Integration and Stress Coverage + +The implemented tests submit multiple state objects through the same manager, +verify that user callbacks do not overlap and run in the runtime async +context, and verify that a failure does not block an unrelated queued state. +The concurrent lifecycle test exercises 24 client threads, while the shutdown +stress test drains 32 queued states behind an executing callback. ## 13. Acceptance Criteria The feature is accepted when all of the following are true: -- clients can create runtime-managed state objects from a singleton engine using a shared_ptr handle +- clients can create runtime-managed state objects from a singleton manager using a shared_ptr handle - state objects subclass `Dmn_State` and retain base state semantics -- `statehandle->run(priority, delay, onError)` schedules work into the runtime engine and returns true/false to indicate success -- state execution is serialized through the runtime engine by default +- `statehandle->run(priority, delay, onError)` schedules work into the runtime manager and returns true/false to indicate success +- state execution is serialized through the runtime manager by default - `statehandle->wait()` and `wait_for()` block until runtime completion or terminal failure and `getFuture()` is available for async waiting (shared_future) - `cancel()` cooperatively finalizes execution and prevents future steps; cancel semantics are documented and tested - exceptions and cancellation leave the runtime in a valid state and optional onError callbacks are invoked @@ -476,16 +583,20 @@ The feature is accepted when all of the following are true: ## 14. Risks and Mitigations (updated) ### Risk: re-entrant scheduling loop -Mitigation: `run()` must schedule a single runtime task per state step and stop when the terminal condition is reached. No unbounded recursive scheduling loop is allowed. Atomic queued flag prevents duplicate enqueues. +Mitigation: `run()` must schedule a single runtime task per state step and stop +when the terminal condition is reached. No unbounded recursive scheduling loop +is allowed. A mutex-protected queued flag prevents duplicate enqueues. ### Risk: wait deadlock -Mitigation: do not call `wait()` from inside the runtime async thread. Document and assert this in debug builds; throw in release builds. Prefer future-based wait from runtime thread contexts. +Mitigation: do not call `wait()` from inside the runtime async thread. Throw +`std::runtime_error` in every build configuration. Prefer future-based waiting +from runtime thread contexts. ### Risk: queued state object lifetime issues -Mitigation: engine must hold a `std::shared_ptr` while queued. Returning a shared_ptr to clients and holding an internal shared_ptr mirrors the existing dmn pattern used in other subsystems (see `dmn-dmesg`). +Mitigation: manager must hold a `std::shared_ptr` while queued. Returning a shared_ptr to clients and holding an internal shared_ptr mirrors the existing dmn pattern used in other subsystems (see `dmn-dmesg`). ### Risk: serializer starvation -Mitigation: the runtime engine must keep job postings small, deterministic, and bounded; the runtime's priority and scheduling mechanisms should be used to avoid starvation. Consider adding a configurable fairness mechanism for long-running state functors. +Mitigation: the runtime manager must keep job postings small, deterministic, and bounded; the runtime's priority and scheduling mechanisms should be used to avoid starvation. Consider adding a configurable fairness mechanism for long-running state functors. ## 15. Implementation Notes @@ -497,59 +608,73 @@ Implementation should reuse: - `Dmn_State` for the state machine mechanics - `Dmn_Runtime_Job` and `Dmn_Runtime_Task` for runtime dispatch -The runtime state engine should primarily add: +The runtime state manager should primarily add: - state object lifecycle tracking using shared_ptr handles - queueing and serialization logic -- `wait()` synchronization (condition_variable + promise/shared_future) +- `wait()` synchronization through the promise/shared_future pair - terminal-state finalization and onError callback forwarding - cooperative cancel() semantics Sample usage (illustrative): ```cpp -using StatePtr = dmn::Dmn_Runtime_State_Engine::DmnRuntimeStatePtr; - -// create -StatePtr s = dmn::Dmn_Runtime_State_Engine::createInstance().createState("example"); - -// configure -s->setStateFnc([](dmn::Dmn_Runtime_State &st){ /* step 0 */ }, 0); -s->setNext(); -s->setStateFnc([](dmn::Dmn_Runtime_State &st){ /* step 1 */ }, 1); -s->setEnd(); - -// run with onError callback, medium priority, immediate -bool ok = s->run(Dmn_Runtime_Job::Priority::kMedium, std::chrono::steady_clock::duration::zero(), - [](std::exception_ptr &ep){ /* log or inspect error */ }); +#include +#include + +using StatePtr = dmn::DmnRuntimeStatePtr; + +// Initialize the process-wide runtime from the main thread before workers are +// created. The state manager can then create and submit managed state handles. +auto runtime = dmn::Dmn_Runtime_Manager<>::createInstance(); +auto manager = dmn::Dmn_Runtime_State_Manager::createInstance(); +StatePtr s = manager->createState("example"); + +// Configure using the inherited Dmn_State API. The manager invokes runNext(); +// each state function selects its own transition. +s->setStateFnc([](dmn::Dmn_State &st) { + /* step work */ + st.setEnd(); +}); + +const bool ok = s->run( + dmn::Dmn_Runtime_Job::Priority::kMedium, + std::chrono::steady_clock::duration::zero(), + [](std::exception_ptr &ep) { /* log or inspect error */ }); if (!ok) { /* handle enqueue failure */ } -// wait (blocking) +// enterMainLoop() keeps the runtime active while scheduled state work drains. +std::thread runtimeMainLoop{[runtime] { runtime->enterMainLoop(); }}; + +// Wait for normal completion, failure, or cancellation. s->wait(); -// or async -auto fut = s->getFuture(); -fut.wait(); +// Explicitly drain/cancel outstanding runtime states before stopping runtime. +manager->shutdown(); +runtime->exitMainLoop(); +runtimeMainLoop.join(); ``` ## 16. Definition of Done The feature is complete when: -- the singleton runtime state engine is designed and documented with shared_ptr handle ownership +- the singleton runtime state manager is designed and documented with + `Dmn_Singleton` shared-pointer ownership - the runtime state object class is specified and matches the required async semantics (run returning bool, onError forwarding, cancel, wait/timeout/shared_future) -- `run()` and `wait()` behavior are documented and tested +- `run()`, `wait()`, and `wait_for()` behavior are documented and tested - serialized execution through `Dmn_Runtime_Manager` is verified - shutdown, failure, and cancellation semantics are validated - example usage patterns are included in the documentation ## 17. Recommended Milestones -1. Create the engine singleton and state object base model (shared_ptr handle) -2. Add runtime scheduling and serialized execution loop with onError forwarding, priority and timed variants -3. Add completion/failure/cancel tracking, wait(timeout), and shared_future support -4. Add tests for execution order, completion, failures, destructor-while-queued, and shutdown modes -5. Validate the runtime integration with `Dmn_Runtime_Manager` +1. Complete the manager singleton, state-handle, lifecycle, and scheduling + implementation (complete). +2. Add focused coverage for cancellation, ownership, priority/timing, and + runtime-thread safety (complete). +3. Define and implement drain-and-cancel manager shutdown (complete). +4. Add multi-state integration and concurrency stress coverage. --- diff --git a/include/dmn-runtime-state.hpp b/include/dmn-runtime-state.hpp index 1a8c372..f3d1e7c 100644 --- a/include/dmn-runtime-state.hpp +++ b/include/dmn-runtime-state.hpp @@ -2,30 +2,51 @@ * Copyright © 2026 Chee Bin HOH. All rights reserved. * * @file dmn-runtime-state.hpp - * @brief Public API for the Runtime State Engine: a runtime-owned state - * execution engine that composes the existing dmn-runtime scheduler - * with the dmn-state finite-state helper. + * @brief Runtime-scheduled finite-state-machine execution and lifetime + * management. * * @author Chee Bin HOH * @date 2026-08-31 * - * This header declares the public types used by clients to create and - * manage runtime-managed state machine instances. Implementation details - * are intentionally omitted from the header; refer to the implementation - * (.cpp) and specification documents for full behavior. + * Overview + * -------- + * This header combines @ref Dmn_State with @ref Dmn_Runtime_Manager to execute + * state-machine steps asynchronously on the process-wide runtime thread. + * Clients create a @ref Dmn_Runtime_State through + * @ref Dmn_Runtime_State_Manager, configure it with the inherited + * @ref Dmn_State API, and submit it with Dmn_Runtime_State::run(). + * + * Ownership and Lifetime + * ---------------------- + * State handles are shared pointers. Once a state is submitted, the manager + * retains an owning handle until it reaches a terminal outcome, ensuring queued + * and running work cannot access a destroyed state. Implementations should use + * shared_from_this() only after a state is owned by a @c std::shared_ptr. + * + * Thread Safety + * ------------- + * Public lifecycle operations and state inspection are thread-safe. State + * functors and lifecycle hooks execute in the runtime async thread. Blocking + * wait and shutdown operations are prohibited from that thread to avoid + * deadlock. */ #ifndef DMN_RUNTIME_STATE_HPP_ #define DMN_RUNTIME_STATE_HPP_ #include "dmn-runtime.hpp" +#include "dmn-singleton.hpp" #include "dmn-state.hpp" #include +#include #include #include #include +#include +#include #include +#include namespace dmn { @@ -35,30 +56,28 @@ namespace dmn { * * Dmn_Runtime_State subclasses Dmn_State and adds asynchronous runtime * ownership semantics: a client obtains a shared_ptr handle from the - * engine, configures state functors using the same Dmn_State API, then + * manager, configures state functors using the same Dmn_State API, then * calls run() to schedule execution on the global runtime thread. * - * Important behavior (summary): - * - createState() returns std::shared_ptr — the engine - * also holds a shared_ptr while the state is queued or running to - * guarantee lifetime. - * - run(priority, delay, onError) enqueues the state for execution; it - * returns true when enqueue succeeded and false otherwise. The optional - * onError callback uses Dmn_Runtime_Job::OnErrorFncType and is forwarded - * into the runtime job. - * - cancel() is cooperative: runtime tasks must observe isCancelled() and - * call setEnd() before runNext() when cancelled. - * - wait(), wait_for(), and getFuture() provide blocking and non-blocking - * completion waiting. getFuture() returns a std::shared_future so - * multiple waiters are supported. - * - Calling run() or wait() from inside the runtime async thread is - * disallowed: implementations MUST assert/throw when detected. + * Lifecycle + * --------- + * A state is configured, submitted once, and then completes, fails, or is + * cancelled. Cancellation is cooperative: it prevents subsequent state steps + * but does not interrupt a functor already executing. Completion is published + * through a @c std::shared_future, which supports multiple waiters. */ -class Dmn_Runtime_State : public Dmn_State { +class Dmn_Runtime_State + : public Dmn_State, + public std::enable_shared_from_this { + friend class Dmn_Runtime_State_Manager; + public: - using FncType = std::function; - using OnErrorFnc = Dmn_Runtime_Job::OnErrorFncType; // std::function + /** + * @brief Callback invoked by the runtime when a state step throws. + * + * The callback receives the exception captured by the runtime job. + */ + using OnErrorFnc = Dmn_Runtime_Job::OnErrorFncType; /** * @brief Construct a runtime-managed state object with a human-readable name. @@ -67,57 +86,52 @@ class Dmn_Runtime_State : public Dmn_State { explicit Dmn_Runtime_State(std::string_view name); /** - * @brief Virtual destructor. Implementation should ensure safe teardown. + * @brief Destroy the state after all owning handles have been released. */ virtual ~Dmn_Runtime_State() noexcept; - /* State configuration (inherited semantics from Dmn_State) */ - - /** - * @brief Set the functor for a state slot. - * @param fnc The functor to be called for the state step. - * @param index 1-based index for the state slot (0 uses default behavior). - * - * This method preserves Dmn_State semantics and allows clients to define - * the machine steps. - */ - void setStateFnc(FncType fnc, int index = 0); - - /** - * @brief Set the next user state by index (1-based). - * @param index 1..m_states.size() selects the next state. - */ - void setNext(int index); + Dmn_Runtime_State(const Dmn_Runtime_State &) = delete; + Dmn_Runtime_State &operator=(const Dmn_Runtime_State &) = delete; + Dmn_Runtime_State(Dmn_Runtime_State &&) = delete; + Dmn_Runtime_State &operator=(Dmn_Runtime_State &&) = delete; - /** - * @brief Convenience: set the next state to the sequential next slot. - */ - void setNext(); + /* Execution control */ /** - * @brief Mark the machine to finalize after the current step. + * @brief Request cooperative cancellation of this state. + * + * The cancellation is idempotent and thread-safe. It does NOT preempt a + * currently-running functor unless that functor explicitly observes the + * cancel flag via isCancelled(). The runtime job must check isCancelled() + * and call setEnd() prior to invoking runNext() if cancellation is set. + * + * A state cancelled before submission becomes terminal immediately. A + * submitted state becomes terminal when the runtime observes the request. */ - void setEnd(); - - /* Lifecycle API */ + void cancel(); /** * @brief Schedule this state handle for runtime execution. * * @param priority Job priority to use when enqueuing (maps to * Dmn_Runtime_Job::Priority). - * @param delay If non-zero, the job is scheduled via addTimedJob() after this - * delay. + * @param delay If non-zero, the first job is scheduled via addTimedJob() + * after this delay. Later state steps are posted immediately. * @param onError Optional error callback forwarded to the runtime job. The * type matches Dmn_Runtime_Job::OnErrorFncType. - * @return true if the state was successfully queued; false if enqueue - * failed (already terminal, duplicate run, invalid configuration). + * @return true if the state was successfully queued; false for an already + * terminal, cancelled, active, or unconfigured state, or when the + * manager has shut down. * * Notes: * - run() is one-shot for a given handle: the first successful call enqueues * the state; subsequent calls return false. * - Calling run() from inside the runtime async thread is disallowed and - * will assert/throw. + * throws std::runtime_error. + * + * @throws std::bad_weak_ptr if this object is not owned by a + * @c std::shared_ptr. + * @throws std::runtime_error if called from the runtime async thread. */ bool run(Dmn_Runtime_Job::Priority priority = Dmn_Runtime_Job::Priority::kMedium, @@ -125,102 +139,247 @@ class Dmn_Runtime_State : public Dmn_State { std::chrono::steady_clock::duration::zero(), OnErrorFnc onError = {}); + /* Completion waiting */ + /** - * @brief Request cooperative cancellation of this state. + * @brief Return a shared_future that becomes ready when the state reaches + * a terminal condition (completed/failed/cancelled). * - * The cancellation is idempotent and thread-safe. It does NOT preempt a - * currently-running functor unless that functor explicitly observes the - * cancel flag via isCancelled(). The runtime job must check isCancelled() - * and call setEnd() prior to invoking runNext() if cancellation is set. + * The shared_future is available immediately after the state is created + * so callers may register waiters before run() is called. + * + * @return A copyable completion future. Calling @c get() on it rethrows a + * state-step failure. */ - void cancel(); + std::shared_future getFuture(); /** * @brief Block until the state reaches a terminal condition. * - * Calling wait() from the runtime async thread is disallowed; implementations - * should assert or throw if detected. See getFuture() for async waiting. + * Calling wait() from the runtime async thread is disallowed and throws + * std::runtime_error. See getFuture() for async waiting. + * + * @throws std::runtime_error if called from the runtime async thread. */ void wait(); /** - * @brief Block until terminal or timeout. + * @brief Block until the state is terminal or the timeout expires. + * @param timeout Maximum duration to wait. * @return true if terminal observed before timeout, false otherwise. + * @throws std::runtime_error if called from the runtime async thread. */ template bool wait_for(const std::chrono::duration &timeout); - /** - * @brief Return a shared_future that becomes ready when the state reaches - * a terminal condition (completed/failed/cancelled). - * - * The shared_future is available immediately after the state is created - * so callers may register waiters before run() is called. - */ - std::shared_future getFuture(); + /* State inspection */ /** - * @brief Introspection helpers. - * - * isRunning(): queued or actively running inside the engine. - * isCompleted(): terminal success. - * isFailed(): terminal failure. - * isCancelled(): cancellation requested. + * @brief Return whether cancellation was requested or completed. */ - bool isRunning() const; + bool isCancelled() const; + + /** @brief Return whether the state completed successfully. */ bool isCompleted() const; + + /** @brief Return whether a state step failed. */ bool isFailed() const; - bool isCancelled() const; + + /** @brief Return whether the state is queued or executing a step. */ + bool isRunning() const; protected: + /** + * @brief Restrict state-machine advancement to this type and its manager. + * + * Derived classes may use @ref runNext to implement specialized execution, + * but clients cannot drive a runtime-managed state directly. + */ + using Dmn_State::runNext; + /** * @brief Lifecycle hooks for derived implementations. * * Subclasses may override these to observe state lifecycle transitions. The * default implementations are no-ops. */ + /** @brief Called once before the first state step executes. */ virtual void onStarted(); + + /** @brief Called after normal terminal completion is published. */ virtual void onCompleted(); + + /** + * @brief Called after a state-step failure is published. + * @param ep Exception raised by the failed state step. + */ virtual void onFailed(std::exception_ptr ep); + + /** @brief Called after cancellation is published as terminal. */ virtual void onCancelled(); private: - // Implementation details are intentionally private and placed in the - // corresponding .cpp. Refer to the specification for the concurrency and - // lifecycle invariants the implementation must satisfy. + enum class Terminal_State { kCompleted, kFailed, kCancelled }; + + /** @brief Begin a runtime step and invoke @ref onStarted exactly once. */ + bool beginStep(); + + /** + * @brief Publish one terminal outcome and invoke its corresponding hook. + * @param terminalState Outcome to publish. + * @param failure Exception to associate with a failed outcome. + * + * @note A pending cancellation request takes precedence over normal + * completion so shutdown cannot publish conflicting terminal states. + */ + void complete(Terminal_State terminalState, std::exception_ptr failure = {}); + /** @brief Clear submission state after the manager declines or cannot queue + * it. */ + void resetQueuedAfterSubmission(); + + mutable std::mutex m_mutex{}; ///< Protects lifecycle state and completion. + std::promise m_completionPromise{}; ///< Fulfilled on terminal state. + std::shared_future + m_completionSharedFuture{}; ///< Copyable completion notification. + std::exception_ptr m_failure{}; ///< Captured state-function failure. + bool m_queued{}; ///< True after successful submission setup. + bool m_running{}; ///< True after the first state step begins until terminal. + bool m_started{}; ///< Ensures onStarted runs once. + bool m_completed{}; ///< Terminal normal-completion marker. + bool m_failed{}; ///< Terminal failure marker. + bool m_cancelled{}; ///< Cancellation requested or terminal. + bool m_terminal{}; ///< Guards one-time completion publication. }; /** - * @class Dmn_Runtime_State_Engine - * @brief Singleton factory and manager for runtime-managed states. + * @brief Shared ownership handle for a runtime-managed state. * - * Responsibilities: - * - provide createState() returning a shared_ptr handle to a Dmn_Runtime_State - * - retain a shared_ptr to queued/running state objects to guarantee lifetime - * - integrate with Dmn_Runtime_Manager by creating runtime jobs for state steps + * The manager returns this handle from createState() and retains an additional + * handle after successful submission until the state becomes terminal. Retain + * this handle to configure, submit, inspect, or await the state. */ -class Dmn_Runtime_State_Engine - : public Dmn_Singleton { +using DmnRuntimeStatePtr = std::shared_ptr; + +/** + * @class Dmn_Runtime_State_Manager + * @brief Singleton manager for runtime-managed states. + * + * The manager is created through its inherited @ref createInstance factory. + * It creates state handles and retains submitted states until a terminal + * outcome, thereby preventing pending runtime work from outliving its state. + */ +class Dmn_Runtime_State_Manager + : public Dmn_Singleton { + friend class Dmn_Singleton; + public: - using DmnRuntimeStatePtr = std::shared_ptr; + /** + * @brief Destroy the runtime state manager singleton. + * + * The destructor does not drain pending runtime work. Call @ref shutdown + * while the runtime main loop remains active before releasing the manager. + */ + virtual ~Dmn_Runtime_State_Manager() noexcept; + + Dmn_Runtime_State_Manager(const Dmn_Runtime_State_Manager &obj) = delete; + Dmn_Runtime_State_Manager & + operator=(const Dmn_Runtime_State_Manager &obj) = delete; + Dmn_Runtime_State_Manager(Dmn_Runtime_State_Manager &&obj) = delete; + Dmn_Runtime_State_Manager & + operator=(Dmn_Runtime_State_Manager &&obj) = delete; /** - * @brief Obtain the singleton engine instance. + * @brief Create a client-owned runtime state handle. + * + * The manager does not retain the returned handle until run() successfully + * queues the state for execution. This factory remains available after + * @ref shutdown, but such a handle cannot be submitted. + * + * @param name Human-readable state name used for diagnostics. + * @return A newly constructed state owned by the caller. */ - static auto createInstance() -> Dmn_Runtime_State_Engine &; + DmnRuntimeStatePtr createState(std::string_view name = ""); /** - * @brief Create a new runtime-managed state object and return a shared_ptr - * handle to the client. The engine will also hold a shared_ptr while - * the state is queued or running. + * @brief Cancel submitted states and wait for the runtime to drain them. + * + * Shutdown is idempotent. It permanently rejects new state submissions: + * handles created after shutdown remain configurable, but @ref + * Dmn_Runtime_State::run returns false. Submitted states are cancelled + * cooperatively; an executing user step may finish before publishing + * cancellation, while queued steps finalize without running another + * user-defined callback. + * + * This method waits without holding the manager mutex until every state + * retained when shutdown began has reached a terminal outcome. The runtime + * main loop must remain active so queued work can observe cancellation. + * + * @throws std::runtime_error if called from the runtime async thread. */ - DmnRuntimeStatePtr createState(std::string_view name); + void shutdown(); protected: - Dmn_Runtime_State_Engine(); - virtual ~Dmn_Runtime_State_Engine() noexcept; + /** + * @brief Construct the process-wide runtime state manager. + * + * Construction is restricted to @ref Dmn_Singleton. The optional name is + * retained for diagnostics; callers normally use the inherited + * @c createInstance() factory without arguments. + * + * @param name Human-readable manager name for diagnostics. + */ + Dmn_Runtime_State_Manager(std::string_view name = ""); + +private: + friend class Dmn_Runtime_State; + + /** + * @brief Retain and submit a state for its first or subsequent step. + * + * The retained handle guarantees state lifetime until @ref releaseState. + * @return true when the job was accepted; false if shutdown rejects an + * initial submission. + */ + bool enqueueState(DmnRuntimeStatePtr state, + Dmn_Runtime_Job::Priority priority, + const std::chrono::steady_clock::duration &delay, + Dmn_Runtime_State::OnErrorFnc onError); + + /** + * @brief Execute one state step and repost or terminally release it. + * + * @param state Non-owning state handle captured by a runtime job. + * @param priority Priority to reuse when posting the next step. + * @param onError Runtime error callback for the next step. + */ + void executeStateStep(std::weak_ptr state, + Dmn_Runtime_Job::Priority priority, + Dmn_Runtime_State::OnErrorFnc onError); + + /** + * @brief Release the manager's retained handle after terminal completion. + * @param state State whose manager-owned handle is removed. + */ + void releaseState(const Dmn_Runtime_State *state); + + std::mutex m_pendingStatesMutex{}; ///< Protects retained state handles. + std::unordered_map + m_pendingStates{}; ///< States retained while queued or running. + bool m_shutdown{}; ///< Protected by m_pendingStatesMutex; rejects new runs. + std::string m_name{}; ///< Human-readable manager name for diagnostics. }; +template +bool Dmn_Runtime_State::wait_for( + const std::chrono::duration &timeout) { + if (Dmn_Runtime_Manager<>::createInstance()->isRunInAsyncThread()) { + throw std::runtime_error( + "Dmn_Runtime_State::wait_for cannot run in the runtime async thread"); + } + + return getFuture().wait_for(timeout) == std::future_status::ready; +} + } // namespace dmn #endif // DMN_RUNTIME_STATE_HPP_ diff --git a/include/dmn-runtime.hpp b/include/dmn-runtime.hpp index 8bbdeb9..c687443 100644 --- a/include/dmn-runtime.hpp +++ b/include/dmn-runtime.hpp @@ -351,6 +351,10 @@ class Dmn_Runtime_Manager */ void exitMainLoop(); + /** @brief Return true if the caller is running on the singleton async thread. + */ + auto isRunInAsyncThread() -> bool; + /** * @brief Register a signal handler hook for a particular signal number. * Handlers are invoked by the runtime in a safe context (not from @@ -393,10 +397,6 @@ class Dmn_Runtime_Manager /** @brief Invoke all registered hooks for @p signo in the async context. */ void execSignalHandlerHookInternal(int signo); - /** @brief Return true if the caller is running on the singleton async thread. - */ - auto isRunInAsyncThread() -> bool; - /** @brief Insert @p hook into the internal map for @p signo. */ void registerSignalHandlerHookInternal(int signo, SignalHandlerHook &&hook); diff --git a/include/dmn-state.hpp b/include/dmn-state.hpp index cf4825a..a8111fe 100644 --- a/include/dmn-state.hpp +++ b/include/dmn-state.hpp @@ -102,6 +102,16 @@ class Dmn_State { */ auto isFinalized() -> bool; + /** + * @brief Return whether the client configured at least one state function. + * + * Excludes the internal initialization function installed during + * construction. + * + * @return true when at least one user-defined state function exists. + */ + bool hasStateFncs() const noexcept; + /** * @brief Execute the next state step. * @return true if the state machine remains active after running the step; diff --git a/include/dmn.hpp b/include/dmn.hpp index 42fabe3..e52d4e1 100644 --- a/include/dmn.hpp +++ b/include/dmn.hpp @@ -36,6 +36,7 @@ #include "dmn-pipe.hpp" #include "dmn-proc.hpp" #include "dmn-pub-sub.hpp" +#include "dmn-runtime-state.hpp" #include "dmn-runtime.hpp" #include "dmn-singleton.hpp" #include "dmn-socket.hpp" diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index ecc99dd..4fbcafa 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -7,6 +7,7 @@ add_library(dmn ${CMAKE_CURRENT_SOURCE_DIR}/dmn-dmesgnet.cpp ${CMAKE_CURRENT_SOURCE_DIR}/dmn-proc.cpp ${CMAKE_CURRENT_SOURCE_DIR}/dmn-runtime.cpp + ${CMAKE_CURRENT_SOURCE_DIR}/dmn-runtime-state.cpp ${CMAKE_CURRENT_SOURCE_DIR}/dmn-socket.cpp ${CMAKE_CURRENT_SOURCE_DIR}/dmn-state.cpp @@ -58,6 +59,7 @@ target_sources(dmn ${PROJECT_SOURCE_DIR}/include/dmn-singleton.hpp ${PROJECT_SOURCE_DIR}/include/dmn-state.hpp ${PROJECT_SOURCE_DIR}/include/dmn-runtime.hpp + ${PROJECT_SOURCE_DIR}/include/dmn-runtime-state.hpp ${PROJECT_SOURCE_DIR}/include/dmn-runtime-task.hpp ${PROJECT_SOURCE_DIR}/include/dmn-socket.hpp @@ -177,4 +179,3 @@ target_link_libraries(dmn-kafka-sender dmn ${rdkafka_LIB} ) - diff --git a/src/dmn-runtime-state.cpp b/src/dmn-runtime-state.cpp new file mode 100644 index 0000000..7a5a6e0 --- /dev/null +++ b/src/dmn-runtime-state.cpp @@ -0,0 +1,357 @@ +/** + * Copyright © 2026 Chee Bin HOH. All rights reserved. + * + * @file dmn-runtime-state.cpp + * @brief Runtime-state lifecycle, scheduling, and manager-retention + * implementation. + * + * Implementation Notes + * -------------------- + * Lifecycle flags and the completion promise are synchronized by + * Dmn_Runtime_State::m_mutex. The manager separately protects its retained + * state handles while jobs are queued or running. Hooks always run after the + * lifecycle mutex is released so derived implementations can safely inspect + * state or call other public APIs. + */ + +#include "dmn-runtime-state.hpp" + +#include +#include +#include +#include +#include +#include +#include +#include + +namespace dmn { + +Dmn_Runtime_State::Dmn_Runtime_State(std::string_view name) + : Dmn_State{name}, + m_completionSharedFuture{m_completionPromise.get_future().share()} {} + +Dmn_Runtime_State::~Dmn_Runtime_State() {} + +void Dmn_Runtime_State::cancel() { + bool completeNow{}; + { + std::lock_guard lock{m_mutex}; + if (m_terminal || m_cancelled) { + return; + } + + m_cancelled = true; + completeNow = !m_queued; + } + + if (completeNow) { + complete(Terminal_State::kCancelled); + } +} + +bool Dmn_Runtime_State::run(Dmn_Runtime_Job::Priority priority, + const std::chrono::steady_clock::duration &delay, + OnErrorFnc onError) { + auto runtime = Dmn_Runtime_Manager<>::createInstance(); + if (runtime->isRunInAsyncThread()) { + throw std::runtime_error( + "Dmn_Runtime_State::run cannot run in the runtime async thread"); + } + + if (!hasStateFncs()) { + return false; + } + + auto self = shared_from_this(); + + { + std::lock_guard lock{m_mutex}; + if (m_terminal || m_cancelled || m_queued) { + return false; + } + + m_queued = true; + } + + if (Dmn_Runtime_State_Manager::createInstance()->enqueueState( + std::move(self), priority, delay, std::move(onError))) { + return true; + } + + resetQueuedAfterSubmission(); + return false; +} + +std::shared_future Dmn_Runtime_State::getFuture() { + std::lock_guard lock{m_mutex}; + + return m_completionSharedFuture; +} + +void Dmn_Runtime_State::wait() { + if (Dmn_Runtime_Manager<>::createInstance()->isRunInAsyncThread()) { + throw std::runtime_error( + "Dmn_Runtime_State::wait cannot run in the runtime async thread"); + } + + getFuture().wait(); +} + +bool Dmn_Runtime_State::isCancelled() const { + std::lock_guard lock{m_mutex}; + + return m_cancelled; +} + +bool Dmn_Runtime_State::isCompleted() const { + std::lock_guard lock{m_mutex}; + return m_completed; +} + +bool Dmn_Runtime_State::isFailed() const { + std::lock_guard lock{m_mutex}; + + return m_failed; +} + +bool Dmn_Runtime_State::isRunning() const { + std::lock_guard lock{m_mutex}; + + return m_queued || m_running; +} + +bool Dmn_Runtime_State::beginStep() { + bool callOnStarted{}; + { + std::lock_guard lock{m_mutex}; + if (m_terminal || m_cancelled) { + return false; + } + + m_running = true; + if (!m_started) { + m_started = true; + callOnStarted = true; + } + } + + if (callOnStarted) { + onStarted(); + } + + return true; +} + +void Dmn_Runtime_State::complete(Terminal_State terminalState, + std::exception_ptr failure) { + bool callOnCompleted{}; + bool callOnFailed{}; + bool callOnCancelled{}; + { + std::lock_guard lock{m_mutex}; + if (m_terminal) { + return; + } + + m_terminal = true; + m_queued = false; + m_running = false; + // Cancellation wins over a step that returned normally during shutdown. + if (terminalState == Terminal_State::kCompleted && m_cancelled) { + terminalState = Terminal_State::kCancelled; + } + + switch (terminalState) { + case Terminal_State::kCompleted: + m_completed = true; + callOnCompleted = true; + m_completionPromise.set_value(); + break; + + case Terminal_State::kFailed: + m_failed = true; + m_failure = failure; + callOnFailed = true; + m_completionPromise.set_exception(failure); + break; + + case Terminal_State::kCancelled: + m_cancelled = true; + callOnCancelled = true; + m_completionPromise.set_value(); + break; + } + } + + if (callOnCompleted) { + onCompleted(); + } else if (callOnFailed) { + onFailed(failure); + } else if (callOnCancelled) { + onCancelled(); + } +} + +void Dmn_Runtime_State::resetQueuedAfterSubmission() { + bool completeNow{}; + { + std::lock_guard lock{m_mutex}; + m_queued = false; + completeNow = m_cancelled && !m_terminal; + } + + if (completeNow) { + complete(Terminal_State::kCancelled); + } +} + +void Dmn_Runtime_State::onStarted() {} + +void Dmn_Runtime_State::onCompleted() {} + +void Dmn_Runtime_State::onFailed(std::exception_ptr ep) { (void)ep; } + +void Dmn_Runtime_State::onCancelled() {} + +/** + * @brief Initialize the singleton manager's diagnostic name. + * + * Runtime state scheduling and manager-side ownership are initialized when a + * state is submitted for execution. + * + * @param name Human-readable manager name for diagnostics. + */ +Dmn_Runtime_State_Manager::Dmn_Runtime_State_Manager(std::string_view name) + : m_name{name} {} + +/** + * @brief Destroy the runtime state manager. + */ +Dmn_Runtime_State_Manager::~Dmn_Runtime_State_Manager() {} + +/** + * @brief Construct a client-owned runtime state handle. + * + * Manager retention begins after @ref Dmn_Runtime_State::run successfully + * queues the state for execution. + * + * @param name Human-readable state name used for diagnostics. + * @return A newly constructed runtime-managed state handle. + */ +DmnRuntimeStatePtr +Dmn_Runtime_State_Manager::createState(std::string_view name) { + return std::make_shared(name); +} + +void Dmn_Runtime_State_Manager::shutdown() { + auto runtime = Dmn_Runtime_Manager<>::createInstance(); + if (runtime->isRunInAsyncThread()) { + throw std::runtime_error( + "Dmn_Runtime_State_Manager::shutdown cannot run in the runtime async " + "thread"); + } + + std::vector pendingStates; + { + std::lock_guard lock{m_pendingStatesMutex}; + m_shutdown = true; + pendingStates.reserve(m_pendingStates.size()); + for (const auto &[state, handle] : m_pendingStates) { + (void)state; + pendingStates.emplace_back(handle); + } + } + + for (const auto &state : pendingStates) { + state->cancel(); + } + + for (const auto &state : pendingStates) { + state->getFuture().wait(); + } +} + +bool Dmn_Runtime_State_Manager::enqueueState( + DmnRuntimeStatePtr state, Dmn_Runtime_Job::Priority priority, + const std::chrono::steady_clock::duration &delay, + Dmn_Runtime_State::OnErrorFnc onError) { + { + std::lock_guard lock{m_pendingStatesMutex}; + const auto existing = m_pendingStates.find(state.get()); + if (m_shutdown && existing == m_pendingStates.end()) { + return false; + } + + if (existing == m_pendingStates.end()) { + m_pendingStates.emplace(state.get(), state); + } + } + + const std::weak_ptr weakState{state}; + auto schedule = [this, weakState, priority, + onError](const Dmn_Runtime_Job &) mutable { + executeStateStep(weakState, priority, std::move(onError)); + }; + + try { + auto runtime = Dmn_Runtime_Manager<>::createInstance(); + if (delay == std::chrono::steady_clock::duration::zero()) { + runtime->addJob(std::move(schedule), priority, std::move(onError)); + } else { + runtime->addTimedJob(std::move(schedule), delay, priority, + std::move(onError)); + } + } catch (...) { + releaseState(state.get()); + state->resetQueuedAfterSubmission(); + + throw; + } + + return true; +} + +void Dmn_Runtime_State_Manager::executeStateStep( + std::weak_ptr weakState, + Dmn_Runtime_Job::Priority priority, Dmn_Runtime_State::OnErrorFnc onError) { + auto state = weakState.lock(); + if (!state) { + return; + } + + try { + if (!state->beginStep()) { + state->setEnd(); + (void)state->runNext(); + state->complete(Dmn_Runtime_State::Terminal_State::kCancelled); + releaseState(state.get()); + + return; + } + + if (!state->runNext()) { + state->complete(Dmn_Runtime_State::Terminal_State::kCompleted); + releaseState(state.get()); + + return; + } + } catch (...) { + auto failure = std::current_exception(); + state->complete(Dmn_Runtime_State::Terminal_State::kFailed, failure); + releaseState(state.get()); + + std::rethrow_exception(failure); + } + + (void)enqueueState(std::move(state), priority, + std::chrono::steady_clock::duration::zero(), + std::move(onError)); +} + +void Dmn_Runtime_State_Manager::releaseState(const Dmn_Runtime_State *state) { + std::lock_guard lock{m_pendingStatesMutex}; + m_pendingStates.erase(state); +} + +} // namespace dmn diff --git a/src/dmn-state.cpp b/src/dmn-state.cpp index 09ef760..1c33410 100644 --- a/src/dmn-state.cpp +++ b/src/dmn-state.cpp @@ -41,6 +41,8 @@ auto Dmn_State::isInitialized() -> bool { return m_initialized; } auto Dmn_State::isFinalized() -> bool { return m_finalized; } +bool Dmn_State::hasStateFncs() const noexcept { return m_states.size() > 1; } + auto Dmn_State::runNext() -> bool { // preferred assertion: use an explicit cast so it always compiles assert(static_cast(*this) && "runNext called after finalize"); diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index bb83fba..2c5aff6 100644 --- a/test/CMakeLists.txt +++ b/test/CMakeLists.txt @@ -64,6 +64,7 @@ ADD_TEST_EXECUTABLE(dmn dmn-test-runtime-queue-lf dmn-test-runtime-queue-lf-2 dmn-test-runtime-queue-lf-3 + dmn-test-runtime-state dmn-test-singleton dmn-test-socket dmn-test-state @@ -140,6 +141,7 @@ if (ENABLE_VALGRIND) dmn-test-runtime-queue-lf dmn-test-runtime-queue-lf-2 dmn-test-runtime-queue-lf-3 + dmn-test-runtime-state dmn-test-singleton dmn-test-socket dmn-test-state diff --git a/test/dmn-test-runtime-state.cpp b/test/dmn-test-runtime-state.cpp new file mode 100644 index 0000000..bed6537 --- /dev/null +++ b/test/dmn-test-runtime-state.cpp @@ -0,0 +1,498 @@ +/** + * Copyright © 2024 - 2025 Chee Bin HOH. All rights reserved. + * + * @file dmn-test-runtime-state.cpp + * @brief Unit tests for runtime-managed state lifecycle and scheduling. + */ + +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "dmn-runtime-state.hpp" + +namespace { + +class Runtime_Main_Loop { +public: + explicit Runtime_Main_Loop( + std::shared_ptr> runtime) + : m_runtime{std::move(runtime)}, + m_thread{[this]() { m_runtime->enterMainLoop(); }} {} + + ~Runtime_Main_Loop() { stop(); } + + Runtime_Main_Loop(const Runtime_Main_Loop &) = delete; + Runtime_Main_Loop &operator=(const Runtime_Main_Loop &) = delete; + + void stop() { + if (!m_stopped) { + m_runtime->exitMainLoop(); + m_thread.join(); + m_stopped = true; + } + } + +private: + std::shared_ptr> m_runtime; + std::thread m_thread; + bool m_stopped{}; +}; + +auto runtime() -> std::shared_ptr> { + return dmn::Dmn_Runtime_Manager<>::createInstance(); +} + +auto stateManager() -> std::shared_ptr { + return dmn::Dmn_Runtime_State_Manager::createInstance(); +} + +} // namespace + +TEST(DmnRuntimeState, CreatesSingletonManagerAndStateHandle) { + auto first = stateManager(); + auto second = stateManager(); + + EXPECT_NE(first, nullptr); + EXPECT_EQ(first.get(), second.get()); + + static_assert(std::is_base_of_v); + EXPECT_NE(first->createState("state"), nullptr); +} + +TEST(DmnRuntimeState, RejectsUnconfiguredAndPreRunCancelledStates) { + using namespace std::chrono_literals; + + auto manager = stateManager(); + auto unconfigured = manager->createState("unconfigured"); + auto unconfiguredFuture = unconfigured->getFuture(); + EXPECT_FALSE(unconfigured->run()); + EXPECT_EQ(unconfiguredFuture.wait_for(0ms), std::future_status::timeout); + + auto cancelled = manager->createState("cancelled"); + auto cancelledFuture = cancelled->getFuture(); + auto secondCancelledFuture = cancelled->getFuture(); + cancelled->cancel(); + cancelled->cancel(); + + EXPECT_TRUE(cancelled->isCancelled()); + EXPECT_EQ(cancelledFuture.wait_for(0ms), std::future_status::ready); + EXPECT_EQ(secondCancelledFuture.wait_for(0ms), std::future_status::ready); + EXPECT_NO_THROW(cancelledFuture.get()); + EXPECT_NO_THROW(secondCancelledFuture.get()); + EXPECT_FALSE(cancelled->run()); +} + +TEST(DmnRuntimeState, ExecutesStatesAndReportsStateFailures) { + using namespace std::chrono_literals; + + auto manager = stateManager(); + int stateCount{}; + auto state = manager->createState("count-to-three"); + state->setStateFnc([&stateCount](dmn::Dmn_State ¤t) { + if (++stateCount >= 3) { + current.setEnd(); + } + }); + + std::atomic_bool onErrorCalled{}; + auto failed = manager->createState("failed-state"); + failed->setStateFnc( + [](dmn::Dmn_State &) { throw std::runtime_error{"state failure"}; }); + auto failedFuture = failed->getFuture(); + + EXPECT_TRUE(state->run()); + EXPECT_TRUE(failed->run(dmn::Dmn_Runtime_Job::Priority::kMedium, + std::chrono::steady_clock::duration::zero(), + [&onErrorCalled](std::exception_ptr &failure) { + onErrorCalled = static_cast(failure); + })); + + Runtime_Main_Loop loop{runtime()}; + EXPECT_TRUE(state->wait_for(5s)); + EXPECT_TRUE(failed->wait_for(5s)); + loop.stop(); + + EXPECT_TRUE(state->isCompleted()); + EXPECT_FALSE(state->isRunning()); + EXPECT_TRUE(static_cast(*state).isInitialized()); + EXPECT_TRUE(static_cast(*state).isFinalized()); + EXPECT_EQ(stateCount, 3); + EXPECT_TRUE(failed->isFailed()); + EXPECT_TRUE(onErrorCalled); + EXPECT_THROW(failedFuture.get(), std::runtime_error); +} + +TEST(DmnRuntimeState, SerializesMultipleStateExecutions) { + using namespace std::chrono_literals; + + constexpr int stateCount = 8; + std::atomic_int activeSteps{}; + std::atomic_int completedSteps{}; + std::atomic_bool concurrentExecution{}; + std::atomic_bool executedInRuntimeThread{true}; + auto manager = stateManager(); + auto runtimeInstance = runtime(); + std::vector states; + states.reserve(stateCount); + + for (int index = 0; index < stateCount; ++index) { + auto state = manager->createState("serialized-state"); + state->setStateFnc([&activeSteps, &completedSteps, &concurrentExecution, + &executedInRuntimeThread, + runtimeInstance](dmn::Dmn_State ¤t) { + if (activeSteps.fetch_add(1) != 0) { + concurrentExecution = true; + } + + executedInRuntimeThread = + executedInRuntimeThread && runtimeInstance->isRunInAsyncThread(); + std::this_thread::sleep_for(1ms); + ++completedSteps; + activeSteps.fetch_sub(1); + current.setEnd(); + }); + EXPECT_TRUE(state->run()); + states.emplace_back(std::move(state)); + } + + Runtime_Main_Loop loop{runtimeInstance}; + for (const auto &state : states) { + EXPECT_TRUE(state->wait_for(5s)); + EXPECT_TRUE(state->isCompleted()); + } + loop.stop(); + + EXPECT_EQ(completedSteps.load(), stateCount); + EXPECT_EQ(activeSteps.load(), 0); + EXPECT_FALSE(concurrentExecution.load()); + EXPECT_TRUE(executedInRuntimeThread.load()); +} + +TEST(DmnRuntimeState, IsolatesStateFailureFromOtherQueuedStates) { + using namespace std::chrono_literals; + + std::atomic_bool errorCallbackCalled{}; + std::atomic_int successfulStepCount{}; + auto manager = stateManager(); + auto failed = manager->createState("isolated-failure"); + failed->setStateFnc( + [](dmn::Dmn_State &) { throw std::runtime_error{"expected failure"}; }); + auto failedFuture = failed->getFuture(); + + auto successful = manager->createState("isolated-success"); + successful->setStateFnc([&successfulStepCount](dmn::Dmn_State ¤t) { + ++successfulStepCount; + current.setEnd(); + }); + + EXPECT_TRUE(failed->run(dmn::Dmn_Runtime_Job::Priority::kHigh, + std::chrono::steady_clock::duration::zero(), + [&errorCallbackCalled](std::exception_ptr &error) { + errorCallbackCalled = static_cast(error); + })); + EXPECT_TRUE(successful->run(dmn::Dmn_Runtime_Job::Priority::kLow)); + + Runtime_Main_Loop loop{runtime()}; + EXPECT_TRUE(failed->wait_for(5s)); + EXPECT_TRUE(successful->wait_for(5s)); + loop.stop(); + + EXPECT_TRUE(failed->isFailed()); + EXPECT_THROW(failedFuture.get(), std::runtime_error); + EXPECT_TRUE(errorCallbackCalled.load()); + EXPECT_TRUE(successful->isCompleted()); + EXPECT_EQ(successfulStepCount.load(), 1); +} + +TEST(DmnRuntimeState, CancelsQueuedStateWithoutRunningUserStep) { + using namespace std::chrono_literals; + + std::atomic_int stepCount{}; + auto state = stateManager()->createState("queued-cancelled-state"); + state->setStateFnc([&stepCount](dmn::Dmn_State &) { ++stepCount; }); + auto completion = state->getFuture(); + + // No main loop is running, so cancellation occurs before this queued job can + // enter its user-defined state step. + EXPECT_TRUE(state->run()); + EXPECT_TRUE(state->isRunning()); + state->cancel(); + EXPECT_TRUE(state->isCancelled()); + EXPECT_EQ(completion.wait_for(0ms), std::future_status::timeout); + + Runtime_Main_Loop loop{runtime()}; + EXPECT_TRUE(state->wait_for(5s)); + loop.stop(); + + EXPECT_TRUE(state->isCancelled()); + EXPECT_FALSE(state->isRunning()); + EXPECT_EQ(stepCount.load(), 0); + EXPECT_TRUE(static_cast(*state).isFinalized()); +} + +TEST(DmnRuntimeState, RetainsStateUntilCompletionAfterClientHandleReleased) { + using namespace std::chrono_literals; + + std::atomic_int stepCount{}; + auto state = stateManager()->createState("manager-retained-state"); + state->setStateFnc([&stepCount](dmn::Dmn_State ¤t) { + ++stepCount; + current.setEnd(); + }); + auto completion = state->getFuture(); + std::weak_ptr weakState{state}; + + EXPECT_TRUE(state->run()); + state.reset(); + + // Runtime jobs capture only weak ownership, so this must be the manager's + // pending-state reference keeping the submitted state alive. + EXPECT_FALSE(weakState.expired()); + + Runtime_Main_Loop loop{runtime()}; + EXPECT_EQ(completion.wait_for(5s), std::future_status::ready); + loop.stop(); + + EXPECT_EQ(stepCount.load(), 1); + EXPECT_TRUE(weakState.expired()); +} + +TEST(DmnRuntimeState, HonorsPriorityOrdering) { + using namespace std::chrono_literals; + + std::vector executionOrder; + auto manager = stateManager(); + auto high = manager->createState("high-priority"); + auto medium = manager->createState("medium-priority"); + auto low = manager->createState("low-priority"); + const auto configure = [&executionOrder](dmn::DmnRuntimeStatePtr state, + char marker) { + state->setStateFnc([&executionOrder, marker](dmn::Dmn_State ¤t) { + executionOrder.push_back(marker); + current.setEnd(); + }); + }; + configure(high, 'H'); + configure(medium, 'M'); + configure(low, 'L'); + + // Submit in reverse priority order so the observed order comes from runtime + // priority scheduling rather than submission order. + EXPECT_TRUE(low->run(dmn::Dmn_Runtime_Job::Priority::kLow)); + EXPECT_TRUE(medium->run(dmn::Dmn_Runtime_Job::Priority::kMedium)); + EXPECT_TRUE(high->run(dmn::Dmn_Runtime_Job::Priority::kHigh)); + + Runtime_Main_Loop loop{runtime()}; + EXPECT_TRUE(high->wait_for(5s)); + EXPECT_TRUE(medium->wait_for(5s)); + EXPECT_TRUE(low->wait_for(5s)); + loop.stop(); + + EXPECT_EQ(executionOrder, (std::vector{'H', 'M', 'L'})); +} + +TEST(DmnRuntimeState, DelaysInitialSubmission) { + using namespace std::chrono_literals; + + constexpr auto delay = 100ms; + dmn::Clock::time_point executedAt{}; + auto state = stateManager()->createState("delayed-state"); + state->setStateFnc([&executedAt](dmn::Dmn_State ¤t) { + executedAt = dmn::Clock::now(); + current.setEnd(); + }); + + const auto submittedAt = dmn::Clock::now(); + EXPECT_TRUE(state->run(dmn::Dmn_Runtime_Job::Priority::kMedium, delay)); + + Runtime_Main_Loop loop{runtime()}; + EXPECT_EQ(state->getFuture().wait_for(delay / 2), + std::future_status::timeout); + EXPECT_TRUE(state->wait_for(5s)); + loop.stop(); + + EXPECT_GE(executedAt - submittedAt, delay); +} + +TEST(DmnRuntimeState, RejectsRunAndWaitOperationsFromRuntimeThread) { + using namespace std::chrono_literals; + + auto state = stateManager()->createState("runtime-thread-state"); + std::atomic_bool runRejected{}; + std::atomic_bool waitRejected{}; + std::atomic_bool waitForRejected{}; + std::promise completed; + auto completedFuture = completed.get_future(); + auto runtimeInstance = runtime(); + + runtimeInstance->addJob( + [&state, &runRejected, &waitRejected, &waitForRejected, + &completed](const dmn::Dmn_Runtime_Job &) -> dmn::Dmn_Runtime_Task { + try { + (void)state->run(); + } catch (const std::runtime_error &) { + runRejected = true; + } + + try { + state->wait(); + } catch (const std::runtime_error &) { + waitRejected = true; + } + + try { + (void)state->wait_for(0ms); + } catch (const std::runtime_error &) { + waitForRejected = true; + } + + completed.set_value(); + co_return; + }); + + Runtime_Main_Loop loop{runtimeInstance}; + EXPECT_EQ(completedFuture.wait_for(5s), std::future_status::ready); + loop.stop(); + + EXPECT_TRUE(runRejected.load()); + EXPECT_TRUE(waitRejected.load()); + EXPECT_TRUE(waitForRejected.load()); +} + +TEST(DmnRuntimeState, HandlesConcurrentStateLifecycleOperations) { + using namespace std::chrono_literals; + + constexpr int stateCount = 24; + std::atomic_int completedUserSteps{}; + std::atomic_int cancelledUserSteps{}; + std::atomic_int submissionFailures{}; + std::atomic_int waitFailures{}; + std::barrier startSubmissions{stateCount}; + auto manager = stateManager(); + Runtime_Main_Loop loop{runtime()}; + std::vector clients; + clients.reserve(stateCount); + + for (int index = 0; index < stateCount; ++index) { + clients.emplace_back([index, &manager, &completedUserSteps, + &cancelledUserSteps, &submissionFailures, + &waitFailures, &startSubmissions]() { + const bool cancelState = index % 2 == 0; + auto state = manager->createState("concurrent-state"); + state->setStateFnc([cancelState, &completedUserSteps, + &cancelledUserSteps](dmn::Dmn_State ¤t) { + if (cancelState) { + ++cancelledUserSteps; + } else { + ++completedUserSteps; + } + + current.setEnd(); + }); + + startSubmissions.arrive_and_wait(); + if (!state->run(dmn::Dmn_Runtime_Job::Priority::kMedium, + cancelState ? 50ms : 0ms)) { + ++submissionFailures; + return; + } + + if (cancelState) { + state->cancel(); + } + + const auto completion = state->getFuture(); + if (completion.wait_for(5s) != std::future_status::ready || + (cancelState ? !state->isCancelled() : !state->isCompleted())) { + ++waitFailures; + } + }); + } + + for (auto &client : clients) { + client.join(); + } + loop.stop(); + + EXPECT_EQ(submissionFailures.load(), 0); + EXPECT_EQ(waitFailures.load(), 0); + EXPECT_EQ(completedUserSteps.load(), stateCount / 2); + EXPECT_EQ(cancelledUserSteps.load(), 0); +} + +TEST(DmnRuntimeState, ShutdownCancelsPendingStatesAndRejectsNewSubmissions) { + using namespace std::chrono_literals; + + auto manager = stateManager(); + std::promise blockingStepStarted; + auto blockingStepStartedFuture = blockingStepStarted.get_future(); + std::promise allowBlockingStepToFinish; + auto allowBlockingStepToFinishFuture = allowBlockingStepToFinish.get_future(); + + auto runningState = manager->createState("shutdown-running-state"); + runningState->setStateFnc( + [&blockingStepStarted, + &allowBlockingStepToFinishFuture](dmn::Dmn_State ¤t) { + blockingStepStarted.set_value(); + allowBlockingStepToFinishFuture.wait(); + current.setEnd(); + }); + auto runningStateFuture = runningState->getFuture(); + + constexpr int queuedStateCount = 32; + std::atomic_int queuedStepCount{}; + std::vector queuedStates; + std::vector> queuedStateFutures; + queuedStates.reserve(queuedStateCount); + queuedStateFutures.reserve(queuedStateCount); + for (int index = 0; index < queuedStateCount; ++index) { + auto queuedState = manager->createState("shutdown-queued-state"); + queuedState->setStateFnc( + [&queuedStepCount](dmn::Dmn_State &) { ++queuedStepCount; }); + queuedStateFutures.emplace_back(queuedState->getFuture()); + EXPECT_TRUE(queuedState->run(dmn::Dmn_Runtime_Job::Priority::kLow)); + queuedStates.emplace_back(std::move(queuedState)); + } + + EXPECT_TRUE(runningState->run(dmn::Dmn_Runtime_Job::Priority::kHigh)); + + Runtime_Main_Loop loop{runtime()}; + EXPECT_EQ(blockingStepStartedFuture.wait_for(5s), std::future_status::ready); + + std::promise shutdownReturned; + auto shutdownReturnedFuture = shutdownReturned.get_future(); + std::thread shutdownThread{[&manager, &shutdownReturned]() { + manager->shutdown(); + shutdownReturned.set_value(); + }}; + + EXPECT_EQ(shutdownReturnedFuture.wait_for(50ms), std::future_status::timeout); + allowBlockingStepToFinish.set_value(); + EXPECT_EQ(shutdownReturnedFuture.wait_for(5s), std::future_status::ready); + shutdownThread.join(); + loop.stop(); + + EXPECT_EQ(runningStateFuture.wait_for(0ms), std::future_status::ready); + EXPECT_TRUE(runningState->isCancelled()); + for (const auto &queuedStateFuture : queuedStateFutures) { + EXPECT_EQ(queuedStateFuture.wait_for(0ms), std::future_status::ready); + } + for (const auto &queuedState : queuedStates) { + EXPECT_TRUE(queuedState->isCancelled()); + } + EXPECT_EQ(queuedStepCount.load(), 0); + + auto postShutdownState = manager->createState("post-shutdown-state"); + postShutdownState->setStateFnc([](dmn::Dmn_State &) {}); + EXPECT_FALSE(postShutdownState->run()); +} diff --git a/test/dmn-test-state.cpp b/test/dmn-test-state.cpp index d99e581..64e6d44 100644 --- a/test/dmn-test-state.cpp +++ b/test/dmn-test-state.cpp @@ -23,6 +23,7 @@ int main(int argc, char *argv[]) { EXPECT_TRUE(s1); EXPECT_TRUE(!s1.isInitialized()); EXPECT_TRUE(!s1.isFinalized()); + EXPECT_FALSE(s1.hasStateFncs()); s1.runNext(); EXPECT_TRUE(s1); @@ -59,6 +60,8 @@ int main(int argc, char *argv[]) { } }); + EXPECT_TRUE(s3.hasStateFncs()); + while (s3) { s3.runNext(); }