diff --git a/docs/releases/3.2.2.md b/docs/releases/3.2.2.md
index b0a6405e..9ad2d1fa 100644
--- a/docs/releases/3.2.2.md
+++ b/docs/releases/3.2.2.md
@@ -76,6 +76,35 @@ Reported by [@Dolecor](https://github.com/Dolecor).
[#643](https://github.com/fgmacedo/python-statemachine/issues/643).
+### Events due at the same moment kept their order only by chance
+
+Both event queues are priority queues ordered by each trigger's due time, and triggers due at
+the same time came out in no particular order. A clock coarse enough to stamp two events alike
+could run them out of the order they were sent or raised: twenty events raised in one callback,
+all stamped with the same time, ran as `0, 2, 6, 14, 19, 18, ...`. Each trigger now also holds
+the order it was created in, and triggers due at the same moment run in that order, on both
+engines and both queues.
+
+### Cancelling a delayed event could disorder the others and strand its caller
+
+`cancel_event()` removed a trigger by filtering the list behind the queue's heap, and a filtered
+heap is not always a heap. The triggers left could then run out of due order: of ten delayed
+events due from 10 to 90 ms, cancelling the first left the one due at 50 ms running before the
+one due at 40 ms. The trigger's future went with it, so on the async engine a coroutine
+awaiting the cancelled event's `send()` was never answered.
+
+A cancelled trigger now stays where it is, marked as cancelled, and is dropped when it reaches
+the head of the queue, without the processing loop waiting for its due time. A coroutine
+awaiting it gets `None` as soon as it is cancelled.
+
+### Delays measured on the wall clock
+
+Every trigger was stamped with `time.time()` plus its delay, and the processing loop held a
+trigger until `time.time()` reached its stamp. A step of the wall clock, from a time sync or a
+change of the system time, moved every due time with it: stepped back, it held even an event
+sent with no delay until the clock caught up; stepped forward, it ran delayed events early.
+Delays are now measured on `time.monotonic()`, which no setting of the wall clock moves.
+
### `delay` and `internal` dropped from an explicit `Event`
`Event(dark.to(lit), delay=50)` rebuilt the event without its `delay`, so
diff --git a/docs/statechart.md b/docs/statechart.md
index 5b92d6a3..e06ed251 100644
--- a/docs/statechart.md
+++ b/docs/statechart.md
@@ -113,6 +113,8 @@ sm.send("timeout", delay=5000, send_id="my_timeout")
sm.cancel_event("my_timeout")
```
+On the async engine, a coroutine awaiting the `send()` of a cancelled event gets `None`.
+
```{note}
The delay is **blocking** in the sync engine — the processing loop sleeps until the
delay elapses, holding the calling thread. In the async engine, delays are scheduled
diff --git a/statemachine/engines/async_.py b/statemachine/engines/async_.py
index 0cd6c4a8..e3fe65aa 100644
--- a/statemachine/engines/async_.py
+++ b/statemachine/engines/async_.py
@@ -2,7 +2,7 @@
import contextvars
from collections.abc import Callable
from itertools import chain
-from time import time
+from time import monotonic
from typing import TYPE_CHECKING
from ..event_data import EventData
@@ -113,7 +113,7 @@ async def _conditions_match(self, transition: "Transition", trigger_data: Trigge
transition.cond.key, *args, on_error=on_error, **kwargs
)
- async def _first_transition_that_matches( # type: ignore[override]
+ async def _first_transition_that_matches(
self,
state: State,
trigger_data: TriggerData,
@@ -130,7 +130,7 @@ async def _first_transition_that_matches( # type: ignore[override]
return transition
return None
- async def _select_transitions( # type: ignore[override]
+ async def _select_transitions(
self, trigger_data: TriggerData, predicate: Callable
) -> "OrderedSet[Transition]":
enabled_transitions: "OrderedSet[Transition]" = OrderedSet()
@@ -147,7 +147,7 @@ async def _select_transitions( # type: ignore[override]
async def select_eventless_transitions(self, trigger_data: TriggerData):
return await self._select_transitions(trigger_data, lambda t, _e: t.is_eventless)
- async def select_transitions(self, trigger_data: TriggerData) -> "OrderedSet[Transition]": # type: ignore[override]
+ async def select_transitions(self, trigger_data: TriggerData) -> "OrderedSet[Transition]":
return await self._select_transitions(trigger_data, lambda t, e: t.match(e))
async def _execute_transition_content(
@@ -172,7 +172,7 @@ async def _execute_transition_content(
return result
- async def _exit_states( # type: ignore[override]
+ async def _exit_states(
self, enabled_transitions: "list[Transition]", trigger_data: TriggerData
) -> "OrderedSet[State]":
ordered_states, result = self._prepare_exit_states(enabled_transitions)
@@ -180,24 +180,22 @@ async def _exit_states( # type: ignore[override]
for info in ordered_states:
# Cancel invocations for this state before executing exit handlers.
- if info.state is not None: # pragma: no branch
- self._invoke_manager.cancel_for_state(info.state)
+ self._invoke_manager.cancel_for_state(info.state)
args, kwargs = await self._get_args_kwargs(
info.transition, trigger_data, source=info.state
)
- if info.state is not None: # pragma: no branch
- self._debug("%s Exiting state: %s", self._log_id, info.state)
- await self.sm._callbacks.async_call(
- info.state.exit.key, *args, on_error=on_error, **kwargs
- )
+ self._debug("%s Exiting state: %s", self._log_id, info.state)
+ await self.sm._callbacks.async_call(
+ info.state.exit.key, *args, on_error=on_error, **kwargs
+ )
self._remove_state_from_configuration(info.state)
return result
- async def _enter_states( # noqa: C901
+ async def _enter_states(
self,
enabled_transitions: "list[Transition]",
trigger_data: TriggerData,
@@ -329,19 +327,6 @@ async def microstep(self, transitions: "list[Transition]", trigger_data: Trigger
# --- Engine loop ---
- async def _run_microstep(self, enabled_transitions, trigger_data): # pragma: no cover
- """Run a microstep for internal/eventless transitions with error handling.
-
- Note: microstep() handles its own errors internally, so this try/except
- is a safety net that is not expected to be reached in normal operation.
- """
- try:
- await self.microstep(list(enabled_transitions), trigger_data)
- except InvalidDefinition:
- raise
- except Exception as e:
- self._handle_error(e, trigger_data)
-
async def activate_initial_state(self, **kwargs):
"""Activate the initial state.
@@ -354,14 +339,12 @@ async def activate_initial_state(self, **kwargs):
"""
return await self.processing_loop()
- async def processing_loop( # noqa: C901
- self, caller_future: "asyncio.Future[object] | None" = None
- ):
- """Process event triggers with the 3-phase macrostep architecture.
+ async def processing_loop(self, caller_future: "asyncio.Future[object] | None" = None):
+ """Process event triggers, a macrostep at a time.
- Phase 1: Eventless transitions + internal queue until quiescence.
- Phase 2: Remaining internal events (safety net for invoke-generated events).
- Phase 3: External events.
+ Each macrostep takes eventless transitions and internal events until neither enables a
+ transition, spawns the invocations of the states it entered, and then takes the
+ external events that are due.
When ``caller_future`` is provided, the caller can ``await`` it to
receive its own event's result — even if another coroutine holds the
@@ -381,110 +364,14 @@ async def processing_loop( # noqa: C901
took_events = True
while took_events and self.running:
self.clear_cache()
- took_events = False
- macrostep_done = False
-
- # Phase 1: eventless transitions and internal events
- while not macrostep_done:
- self._microstep_count = 0
- self._debug(
- "%s Macrostep %d: eventless/internal queue",
- self._log_id,
- self._macrostep_count,
- )
-
- self.clear_cache()
- internal_event = TriggerData(self.sm, event=None) # null object for eventless
- enabled_transitions = await self.select_eventless_transitions(internal_event)
- if not enabled_transitions:
- if self.internal_queue.is_empty():
- macrostep_done = True
- else:
- internal_event = self.internal_queue.pop()
- enabled_transitions = await self.select_transitions(internal_event)
- if enabled_transitions:
- self._debug(
- "%s Enabled transitions: %s", self._log_id, enabled_transitions
- )
- took_events = True
- await self._run_microstep(enabled_transitions, internal_event)
+ took_events = await self._run_macrostep()
# Spawn invoke handlers for states entered during this macrostep.
await self._invoke_manager.spawn_pending_async()
self._check_root_final_state()
- # Phase 2: remaining internal events
- while not self.internal_queue.is_empty(): # pragma: no cover
- internal_event = self.internal_queue.pop()
- enabled_transitions = await self.select_transitions(internal_event)
- if enabled_transitions:
- await self._run_microstep(enabled_transitions, internal_event)
-
- # Phase 3: external events
- self._debug("%s Macrostep %d: external queue", self._log_id, self._macrostep_count)
- while not self.external_queue.is_empty():
- self.clear_cache()
- took_events = True
- external_event = self.external_queue.pop()
- current_time = time()
- if external_event.execution_time > current_time:
- self.put(external_event, _delayed=True)
- await asyncio.sleep(self.sm._loop_sleep_in_ms)
- # Break to Phase 1 so internal events and eventless
- # transitions can be processed while we wait.
- break
-
- self._macrostep_count += 1
- self._microstep_count = 0
- self._debug(
- "%s macrostep %d: event=%s",
- self._log_id,
- self._macrostep_count,
- external_event.event,
- )
-
- # Handle lazy initial state activation.
- # Break out of phase 3 so the outer loop restarts from phase 1
- # (eventless/internal), ensuring internal events queued during
- # initial entry are processed before any external events.
- if external_event.event == "__initial__":
- transitions = self._initial_transitions(external_event)
- await self._enter_states(
- transitions, external_event, OrderedSet(), OrderedSet()
- )
- break
-
- # Finalize + autoforward for active invocations
- self._invoke_manager.handle_external_event(external_event)
-
- event_future = external_event.future
- try:
- enabled_transitions = await self.select_transitions(external_event)
- self._debug(
- "%s Enabled transitions: %s", self._log_id, enabled_transitions
- )
- if enabled_transitions:
- result = await self.microstep(
- list(enabled_transitions), external_event
- )
- self._resolve_future(event_future, result)
- if first_result is self._sentinel:
- first_result = result
- else:
- if not self.sm.allow_event_without_transition:
- tna = TransitionNotAllowed(
- external_event.event, self.sm.configuration
- )
- self._reject_future(event_future, tna)
- self._reject_pending_futures(tna)
- raise tna
- # Event allowed but no transition — resolve with None
- self._resolve_future(event_future, None)
- except Exception as exc:
- self._reject_future(event_future, exc)
- self._reject_pending_futures(exc)
- self.clear()
- raise
+ took_external_events, first_result = await self._take_external_events(first_result)
+ took_events = took_events or took_external_events
except Exception as exc:
if caller_future is not None:
@@ -510,6 +397,105 @@ async def processing_loop( # noqa: C901
return await caller_future
return result
+ async def _run_macrostep(self) -> bool:
+ """Take eventless transitions and internal events until neither enables a transition.
+
+ Returns whether any transition was taken. The internal queue is empty on return.
+ """
+ took_events = False
+ while True:
+ self._microstep_count = 0
+ self._debug(
+ "%s Macrostep %d: eventless/internal queue",
+ self._log_id,
+ self._macrostep_count,
+ )
+
+ self.clear_cache()
+ internal_event = TriggerData(self.sm, event=None) # null object for eventless
+ enabled_transitions = await self.select_eventless_transitions(internal_event)
+ if not enabled_transitions:
+ if self.internal_queue.is_empty():
+ return took_events
+ internal_event = self.internal_queue.pop()
+ enabled_transitions = await self.select_transitions(internal_event)
+ if enabled_transitions:
+ self._debug("%s Enabled transitions: %s", self._log_id, enabled_transitions)
+ took_events = True
+ await self.microstep(list(enabled_transitions), internal_event)
+
+ async def _take_external_events(self, first_result):
+ """Take external events in due order until the queue is empty or holds one not yet due.
+
+ Returns whether any event was taken from the queue, and ``first_result`` updated with
+ the result of the first event that enabled a transition.
+ """
+ self._debug("%s Macrostep %d: external queue", self._log_id, self._macrostep_count)
+ took_events = False
+ while not self.external_queue.is_empty():
+ self.clear_cache()
+ took_events = True
+ external_event = self.external_queue.pop()
+ current_time = monotonic()
+ if external_event.execution_time > current_time:
+ self.put(external_event, _delayed=True)
+ await asyncio.sleep(self.sm._loop_sleep_in_ms)
+ # Break to the macrostep so internal events and eventless
+ # transitions can be processed while we wait.
+ break
+
+ self._macrostep_count += 1
+ self._microstep_count = 0
+ self._debug(
+ "%s macrostep %d: event=%s",
+ self._log_id,
+ self._macrostep_count,
+ external_event.event,
+ )
+
+ # Handle lazy initial state activation.
+ # Break to the macrostep so internal events queued during
+ # initial entry are processed before any external events.
+ if external_event.event == "__initial__":
+ transitions = self._initial_transitions(external_event)
+ await self._enter_states(transitions, external_event, OrderedSet(), OrderedSet())
+ break
+
+ result = await self._take_external_event(external_event)
+ if first_result is self._sentinel:
+ first_result = result
+ return took_events, first_result
+
+ async def _take_external_event(self, external_event: TriggerData):
+ """Run one external event and settle its future.
+
+ Returns the event's microstep result, or the sentinel when it enabled no transition.
+ """
+ # Finalize + autoforward for active invocations
+ self._invoke_manager.handle_external_event(external_event)
+
+ event_future = external_event.future
+ try:
+ enabled_transitions = await self.select_transitions(external_event)
+ self._debug("%s Enabled transitions: %s", self._log_id, enabled_transitions)
+ if enabled_transitions:
+ result = await self.microstep(list(enabled_transitions), external_event)
+ self._resolve_future(event_future, result)
+ return result
+ if not self.sm.allow_event_without_transition:
+ tna = TransitionNotAllowed(external_event.event, self.sm.configuration)
+ self._reject_future(event_future, tna)
+ self._reject_pending_futures(tna)
+ raise tna
+ # Event allowed but no transition — resolve with None
+ self._resolve_future(event_future, None)
+ return self._sentinel
+ except Exception as exc:
+ self._reject_future(event_future, exc)
+ self._reject_pending_futures(exc)
+ self.clear()
+ raise
+
async def enabled_events(self, *args, **kwargs):
sm = self.sm
enabled = {}
diff --git a/statemachine/engines/base.py b/statemachine/engines/base.py
index 14ae301d..796eed36 100644
--- a/statemachine/engines/base.py
+++ b/statemachine/engines/base.py
@@ -2,7 +2,7 @@
from collections.abc import Callable
from dataclasses import dataclass
from dataclasses import field
-from itertools import chain
+from heapq import heappop
from queue import PriorityQueue
from queue import Queue
from threading import Lock
@@ -33,6 +33,13 @@ class StateTransition:
state: State
+#: What entering states needs, worked out before any callback runs: the states in entry order,
+#: those entered by default, the default history content, and the new configuration.
+_EntryPlan = tuple[
+ list[StateTransition], OrderedSet[StateTransition], dict[str, Any], OrderedSet[State]
+]
+
+
class EventQueue:
def __init__(self):
self.queue: Queue = PriorityQueue()
@@ -41,7 +48,13 @@ def __repr__(self):
return f"EventQueue({self.queue.queue!r}, size={self.queue.qsize()})"
def is_empty(self):
- return self.queue.qsize() == 0
+ """Whether the queue holds no trigger that has not been cancelled. Cancelled triggers at
+ the head are dropped first, so the :meth:`pop` that follows returns one that was not."""
+ with self.queue.mutex:
+ heap = self.queue.queue
+ while heap and heap[0].cancelled:
+ heappop(heap)
+ return not heap
def put(self, trigger_data: TriggerData):
"""Put the trigger on the queue without blocking the caller."""
@@ -67,15 +80,17 @@ def reject_futures(self, exc: Exception):
if future is not None and not future.done():
future.set_exception(exc)
- def remove(self, send_id: str):
- # We use the internal `queue` to make thins faster as the mutex
- # is protecting the block below
+ def cancel(self, send_id: str):
+ """Cancel every trigger with this ``send_id``, and answer a caller awaiting one with
+ ``None``. Each stays where it is, so the heap keeps its order, until :meth:`is_empty`
+ drops it from the head."""
with self.queue.mutex:
- self.queue.queue = [
- trigger_data
- for trigger_data in self.queue.queue
- if trigger_data.send_id != send_id
- ]
+ for trigger_data in self.queue.queue:
+ if trigger_data.send_id == send_id:
+ trigger_data.cancelled = True
+ future = trigger_data.future
+ if future is not None and not future.done():
+ future.set_result(None)
_ERROR_EXECUTION = "error.execution"
@@ -97,9 +112,6 @@ def __init__(self, sm: "StateChart"):
self._debug = logger.debug if logger.isEnabledFor(logging.DEBUG) else lambda *a, **k: None
self._root_parallel_final_pending: "State | None" = None
- def empty(self): # pragma: no cover
- return self.external_queue.is_empty()
-
def clear_cache(self):
"""Clears the cache. Should be called at the start of each processing loop."""
self._cache.clear()
@@ -122,15 +134,12 @@ def put(self, trigger_data: TriggerData, internal: bool = False, _delayed: bool
"internal" if internal else "external",
)
- def pop(self): # pragma: no cover
- return self.external_queue.pop()
-
def clear(self):
self.external_queue.clear()
def cancel_event(self, send_id: str):
"""Cancel the event with the given send_id."""
- self.external_queue.remove(send_id)
+ self.external_queue.cancel(send_id)
def _on_error_handler(self) -> "Callable[[Exception], None] | None":
"""Return a per-block error handler, or ``None``.
@@ -324,100 +333,6 @@ def get_effective_target_states(self, transition: Transition) -> OrderedSet[Stat
return targets
- def select_eventless_transitions(self, trigger_data: TriggerData):
- """
- Select the eventless transitions that match the trigger data.
- """
- return self._select_transitions(trigger_data, lambda t, _e: t.is_eventless)
-
- def select_transitions(self, trigger_data: TriggerData) -> OrderedSet[Transition]:
- """
- Select the transitions that match the trigger data.
- """
- return self._select_transitions(trigger_data, lambda t, e: t.match(e))
-
- def _first_transition_that_matches(
- self,
- state: State,
- trigger_data: TriggerData,
- predicate: Callable,
- ) -> "Transition | None":
- for s in chain([state], state.ancestors()):
- transition: Transition
- for transition in s.transitions:
- if (
- not transition.initial
- and predicate(transition, trigger_data.event)
- and self._conditions_match(transition, trigger_data)
- ):
- return transition
- return None
-
- def _select_transitions(
- self, trigger_data: TriggerData, predicate: Callable
- ) -> OrderedSet[Transition]:
- """Select the transitions that match the trigger data."""
- enabled_transitions = OrderedSet[Transition]()
-
- # Get atomic states, TODO: sorted by document order
- atomic_states = (state for state in self.sm.configuration if state.is_atomic)
-
- for state in atomic_states:
- transition = self._first_transition_that_matches(state, trigger_data, predicate)
- if transition is not None:
- enabled_transitions.add(transition)
-
- return self._filter_conflicting_transitions(enabled_transitions)
-
- def microstep(self, transitions: list[Transition], trigger_data: TriggerData):
- """Process a single set of transitions in a 'lock step'.
- This includes exiting states, executing transition content, and entering states.
- """
- self._microstep_count += 1
- self._debug(
- "%s macro:%d micro:%d transitions: %s",
- self._log_id,
- self._macrostep_count,
- self._microstep_count,
- transitions,
- )
- previous_configuration = self.sm.configuration
- try:
- result = self._execute_transition_content(
- transitions, trigger_data, lambda t: t.before.key
- )
-
- states_to_exit = self._exit_states(transitions, trigger_data)
- result += self._enter_states(
- transitions, trigger_data, states_to_exit, previous_configuration
- )
- except InvalidDefinition:
- self.sm.configuration = previous_configuration
- raise
- except Exception as e:
- self.sm.configuration = previous_configuration
- self._handle_error(e, trigger_data)
- return None
-
- try:
- self._execute_transition_content(
- transitions,
- trigger_data,
- lambda t: t.after.key,
- set_target_as_state=True,
- )
- except InvalidDefinition:
- raise
- except Exception as e:
- self._handle_error(e, trigger_data)
-
- if len(result) == 0:
- result = None
- elif len(result) == 1:
- result = result[0]
-
- return result
-
def _get_args_kwargs(
self,
transition: Transition,
@@ -479,7 +394,7 @@ def _prepare_exit_states(
state = info.state
for history in state.history:
if history.type.is_deep:
- history_value = [s for s in self.sm.configuration if s.is_descendant(state)] # noqa: E501
+ history_value = [s for s in self.sm.configuration if s.is_descendant(state)]
else: # shallow history
history_value = [s for s in self.sm.configuration if s.parent == state]
@@ -499,29 +414,6 @@ def _remove_state_from_configuration(self, state: State):
if not self.sm.atomic_configuration_update:
self.sm._config.discard(state)
- def _exit_states(
- self, enabled_transitions: list[Transition], trigger_data: TriggerData
- ) -> OrderedSet[State]:
- """Compute and process the states to exit for the given transitions."""
- ordered_states, result = self._prepare_exit_states(enabled_transitions)
- on_error = self._on_error_handler()
-
- for info in ordered_states:
- # Cancel invocations for this state before executing exit handlers.
- if info.state is not None: # pragma: no branch
- self._invoke_manager.cancel_for_state(info.state)
-
- args, kwargs = self._get_args_kwargs(info.transition, trigger_data, source=info.state)
-
- # Execute `onexit` handlers — same per-block error isolation as onentry.
- if info.state is not None: # pragma: no branch
- self._debug("%s Exiting state: %s", self._log_id, info.state)
- self.sm._callbacks.call(info.state.exit.key, *args, on_error=on_error, **kwargs)
-
- self._remove_state_from_configuration(info.state)
-
- return result
-
def _execute_transition_content(
self,
enabled_transitions: list[Transition],
@@ -549,7 +441,7 @@ def _prepare_entry_states(
enabled_transitions: list[Transition],
states_to_exit: OrderedSet[State],
previous_configuration: OrderedSet[State],
- ) -> "tuple[list[StateTransition], OrderedSet[StateTransition], dict[str, Any], OrderedSet[State]]": # noqa: E501
+ ) -> _EntryPlan:
"""Compute entry set, ordering, and new configuration. Pure computation, no callbacks.
Returns:
@@ -590,7 +482,7 @@ def stop(self):
self.running = False
try:
self._invoke_manager.cancel_all()
- except Exception: # pragma: no cover
+ except Exception:
self._debug("%s Error stopping engine", self._log_id, exc_info=True)
def __del__(self):
@@ -633,7 +525,7 @@ def _handle_final_state(self, target: State, on_entry_result: list):
if grandparent.parent is None:
self._root_parallel_final_pending = grandparent
- def _enter_states( # noqa: C901
+ def _enter_states(
self,
enabled_transitions: list[Transition],
trigger_data: TriggerData,
@@ -755,7 +647,7 @@ def compute_entry_set(
default_history_content,
)
- def add_descendant_states_to_enter( # noqa: C901
+ def add_descendant_states_to_enter(
self,
info: StateTransition,
states_to_enter,
@@ -775,121 +667,164 @@ def add_descendant_states_to_enter( # noqa: C901
state = info.state
if state and state.is_history:
- # Handle history state
- state = cast(HistoryState, state)
- parent_id = state.parent and state.parent.id
- default_history_content[parent_id] = [info]
- if state.id in self.sm.history_values:
- self._debug(
- "%s History state '%s.%s' %s restoring: '%s'",
- self._log_id,
- state.parent,
- state,
- state.type.value,
- [s.id for s in self.sm.history_values[state.id]],
- )
- for history_state in self.sm.history_values[state.id]:
- info_to_add = StateTransition(transition=info.transition, state=history_state)
- if state.type.is_deep:
- states_to_enter.add(info_to_add)
- else:
- self.add_descendant_states_to_enter(
- info_to_add,
- states_to_enter,
- states_for_default_entry,
- default_history_content,
- )
- for history_state in self.sm.history_values[state.id]:
- info_to_add = StateTransition(transition=info.transition, state=history_state)
- self.add_ancestor_states_to_enter(
- info_to_add,
- state.parent,
- states_to_enter,
- states_for_default_entry,
- default_history_content,
- )
- else:
- # Handle default history content
- self._debug(
- "%s History state '%s.%s' default content: %s",
- self._log_id,
- state.parent,
- state,
- [t.target.id for t in state.transitions if t.target],
- )
-
- for transition in state.transitions:
- target = cast(State, transition.target)
- info_history = StateTransition(transition=transition, state=target)
- default_history_content[parent_id].append(info_history)
- self.add_descendant_states_to_enter(
- info_history,
- states_to_enter,
- states_for_default_entry,
- default_history_content,
- ) # noqa: E501
- for transition in state.transitions:
- target = cast(State, transition.target)
- info_history = StateTransition(transition=transition, state=target)
-
- self.add_ancestor_states_to_enter(
- info_history,
- state.parent,
- states_to_enter,
- states_for_default_entry,
- default_history_content,
- ) # noqa: E501
+ self._add_history_states_to_enter(
+ info,
+ cast(HistoryState, state),
+ states_to_enter,
+ states_for_default_entry,
+ default_history_content,
+ )
return
# Add the state to the entry set
- if (
- self.sm.enable_self_transition_entries
- or not info.transition.internal
- or not (
- info.transition.is_self
- or (
- info.transition.target
- and info.transition.target.is_descendant(info.transition.source)
- )
- )
- ):
+ if self._enters_target(info.transition):
states_to_enter.add(info)
state = info.state
if state.parallel:
- for child_state in state.states:
- if not any( # pragma: no branch
- s.state.is_descendant(child_state) for s in states_to_enter
- ):
- info_to_add = StateTransition(transition=info.transition, state=child_state)
+ self._add_regions_to_enter(
+ info, states_to_enter, states_for_default_entry, default_history_content
+ )
+ elif state.is_compound:
+ self._add_initial_states_to_enter(
+ info, states_to_enter, states_for_default_entry, default_history_content
+ )
+
+ def _enters_target(self, transition: Transition) -> bool:
+ """Whether taking ``transition`` enters its target.
+
+ An internal transition to its own source, or to a descendant of it, does not, unless
+ the machine enables self-transition entries.
+ """
+ return (
+ self.sm.enable_self_transition_entries
+ or not transition.internal
+ or not (
+ transition.is_self
+ or (transition.target and transition.target.is_descendant(transition.source))
+ )
+ )
+
+ def _add_history_states_to_enter(
+ self,
+ info: StateTransition,
+ state: HistoryState,
+ states_to_enter,
+ states_for_default_entry,
+ default_history_content,
+ ):
+ """Add the states a history state stands for: those it recorded, or its default."""
+ parent_id = state.parent and state.parent.id
+ default_history_content[parent_id] = [info]
+ if state.id in self.sm.history_values:
+ self._debug(
+ "%s History state '%s.%s' %s restoring: '%s'",
+ self._log_id,
+ state.parent,
+ state,
+ state.type.value,
+ [s.id for s in self.sm.history_values[state.id]],
+ )
+ for history_state in self.sm.history_values[state.id]:
+ info_to_add = StateTransition(transition=info.transition, state=history_state)
+ if state.type.is_deep:
+ states_to_enter.add(info_to_add)
+ else:
self.add_descendant_states_to_enter(
info_to_add,
states_to_enter,
states_for_default_entry,
default_history_content,
)
- elif state.is_compound:
- states_for_default_entry.add(info)
- transition = next(t for t in state.transitions if t.initial)
- # Process all targets (supports multi-target initial transitions for parallel regions)
- for initial_target in transition.targets:
- info_initial = StateTransition(transition=transition, state=initial_target)
+ for history_state in self.sm.history_values[state.id]:
+ info_to_add = StateTransition(transition=info.transition, state=history_state)
+ self.add_ancestor_states_to_enter(
+ info_to_add,
+ state.parent,
+ states_to_enter,
+ states_for_default_entry,
+ default_history_content,
+ )
+ else:
+ # Handle default history content
+ self._debug(
+ "%s History state '%s.%s' default content: %s",
+ self._log_id,
+ state.parent,
+ state,
+ [t.target.id for t in state.transitions if t.target],
+ )
+
+ for transition in state.transitions:
+ target = cast(State, transition.target)
+ info_history = StateTransition(transition=transition, state=target)
+ default_history_content[parent_id].append(info_history)
self.add_descendant_states_to_enter(
- info_initial,
+ info_history,
states_to_enter,
states_for_default_entry,
default_history_content,
)
- for initial_target in transition.targets:
- info_initial = StateTransition(transition=transition, state=initial_target)
+ for transition in state.transitions:
+ target = cast(State, transition.target)
+ info_history = StateTransition(transition=transition, state=target)
+
self.add_ancestor_states_to_enter(
- info_initial,
- state,
+ info_history,
+ state.parent,
states_to_enter,
states_for_default_entry,
default_history_content,
)
+ def _add_regions_to_enter(
+ self,
+ info: StateTransition,
+ states_to_enter,
+ states_for_default_entry,
+ default_history_content,
+ ):
+ """Add each region of a parallel state, unless a state inside it is already entered."""
+ for child_state in info.state.states:
+ if not any(s.state.is_descendant(child_state) for s in states_to_enter):
+ info_to_add = StateTransition(transition=info.transition, state=child_state)
+ self.add_descendant_states_to_enter(
+ info_to_add,
+ states_to_enter,
+ states_for_default_entry,
+ default_history_content,
+ )
+
+ def _add_initial_states_to_enter(
+ self,
+ info: StateTransition,
+ states_to_enter,
+ states_for_default_entry,
+ default_history_content,
+ ):
+ """Add the states a compound state's initial transition targets."""
+ state = info.state
+ states_for_default_entry.add(info)
+ transition = next(t for t in state.transitions if t.initial)
+ # Process all targets (supports multi-target initial transitions for parallel regions)
+ for initial_target in transition.targets:
+ info_initial = StateTransition(transition=transition, state=initial_target)
+ self.add_descendant_states_to_enter(
+ info_initial,
+ states_to_enter,
+ states_for_default_entry,
+ default_history_content,
+ )
+ for initial_target in transition.targets:
+ info_initial = StateTransition(transition=transition, state=initial_target)
+ self.add_ancestor_states_to_enter(
+ info_initial,
+ state,
+ states_to_enter,
+ states_for_default_entry,
+ default_history_content,
+ )
+
def add_ancestor_states_to_enter(
self,
info: StateTransition,
@@ -952,9 +887,10 @@ def _check_root_final_state(self):
self.running = False
def is_in_final_state(self, state: State) -> bool:
- if state.is_compound:
- return any(s.final and s in self.sm.configuration for s in state.states)
- elif state.parallel: # pragma: no cover — requires nested parallel-in-parallel
+ """Whether every region of a parallel state, or a compound state, rests in a final child.
+
+ An atomic state has no children, so it never does.
+ """
+ if state.parallel:
return all(self.is_in_final_state(s) for s in state.states)
- else: # pragma: no cover — atomic states are never "in final state"
- return False
+ return any(s.final and s in self.sm.configuration for s in state.states)
diff --git a/statemachine/engines/sync.py b/statemachine/engines/sync.py
index 627b51ae..3cba86a9 100644
--- a/statemachine/engines/sync.py
+++ b/statemachine/engines/sync.py
@@ -1,6 +1,7 @@
+from collections.abc import Callable
+from itertools import chain
+from time import monotonic
from time import sleep
-from time import time
-from typing import TYPE_CHECKING
from statemachine.event import BoundEvent
from statemachine.orderedset import OrderedSet
@@ -8,26 +9,12 @@
from ..event_data import TriggerData
from ..exceptions import InvalidDefinition
from ..exceptions import TransitionNotAllowed
+from ..state import State
+from ..transition import Transition
from .base import BaseEngine
-if TYPE_CHECKING:
- from ..transition import Transition
-
class SyncEngine(BaseEngine):
- def _run_microstep(self, enabled_transitions, trigger_data):
- """Run a microstep for internal/eventless transitions with error handling.
-
- Note: microstep() handles its own errors internally, so this try/except
- is a safety net that is not expected to be reached in normal operation.
- """
- try:
- self.microstep(list(enabled_transitions), trigger_data)
- except InvalidDefinition:
- raise
- except Exception as e: # pragma: no cover
- self._handle_error(e, trigger_data)
-
def start(self, **kwargs):
if self.sm.current_state_value is not None:
return
@@ -56,7 +43,7 @@ def activate_initial_state(self, **kwargs):
self._processing.release()
return self.processing_loop()
- def processing_loop(self, caller_future=None): # noqa: C901
+ def processing_loop(self):
"""Process event triggers.
The event is put on a queue, and only the first event will have the result collected.
@@ -80,100 +67,217 @@ def processing_loop(self, caller_future=None): # noqa: C901
took_events = True
while took_events and self.running:
self.clear_cache()
- took_events = False
- # Execute the triggers in the queue in FIFO order until the queue is empty
- # while self._running and not self.external_queue.is_empty():
- macrostep_done = False
- enabled_transitions: "OrderedSet[Transition] | None" = None
-
- # handles eventless transitions and internal events
- while not macrostep_done:
- self._microstep_count = 0
- self._debug(
- "%s Macrostep %d: eventless/internal queue",
- self._log_id,
- self._macrostep_count,
- )
-
- self.clear_cache()
- internal_event = TriggerData(
- self.sm, event=None
- ) # this one is a "null object"
- enabled_transitions = self.select_eventless_transitions(internal_event)
- if not enabled_transitions:
- if self.internal_queue.is_empty():
- macrostep_done = True
- else:
- internal_event = self.internal_queue.pop()
- enabled_transitions = self.select_transitions(internal_event)
- if enabled_transitions:
- self._debug(
- "%s Enabled transitions: %s", self._log_id, enabled_transitions
- )
- took_events = True
- self._run_microstep(enabled_transitions, internal_event)
+ took_events = self._run_macrostep()
# Spawn invoke handlers for states entered during this macrostep.
self._invoke_manager.spawn_pending_sync()
self._check_root_final_state()
- # Process remaining internal events before external events.
- # Note: the macrostep loop above already drains the internal queue,
- # so this is a safety net per SCXML spec for invoke-generated events.
- while not self.internal_queue.is_empty(): # pragma: no cover
- internal_event = self.internal_queue.pop()
- enabled_transitions = self.select_transitions(internal_event)
- if enabled_transitions:
- self._run_microstep(enabled_transitions, internal_event)
-
- # Process external events
- self._debug("%s Macrostep %d: external queue", self._log_id, self._macrostep_count)
- while not self.external_queue.is_empty():
- self.clear_cache()
- took_events = True
- external_event = self.external_queue.pop()
- current_time = time()
- if external_event.execution_time > current_time:
- self.put(external_event, _delayed=True)
- sleep(self.sm._loop_sleep_in_ms)
- # Break to Phase 1 so internal events and eventless
- # transitions can be processed while we wait.
- break
-
- self._macrostep_count += 1
- self._microstep_count = 0
- self._debug(
- "%s macrostep %d: event=%s",
- self._log_id,
- self._macrostep_count,
- external_event.event,
- )
+ took_external_events, first_result = self._take_external_events(first_result)
+ took_events = took_events or took_external_events
+ finally:
+ self._processing.release()
+ self._debug("%s Processing loop ended", self._log_id)
+ return first_result if first_result is not self._sentinel else None
+
+ def _run_macrostep(self) -> bool:
+ """Take eventless transitions and internal events until neither enables a transition.
+
+ Returns whether any transition was taken. The internal queue is empty on return.
+ """
+ took_events = False
+ while True:
+ self._microstep_count = 0
+ self._debug(
+ "%s Macrostep %d: eventless/internal queue",
+ self._log_id,
+ self._macrostep_count,
+ )
- # Finalize + autoforward for active invocations
- self._invoke_manager.handle_external_event(external_event)
+ self.clear_cache()
+ internal_event = TriggerData(self.sm, event=None) # this one is a "null object"
+ enabled_transitions = self.select_eventless_transitions(internal_event)
+ if not enabled_transitions:
+ if self.internal_queue.is_empty():
+ return took_events
+ internal_event = self.internal_queue.pop()
+ enabled_transitions = self.select_transitions(internal_event)
+ if enabled_transitions:
+ self._debug("%s Enabled transitions: %s", self._log_id, enabled_transitions)
+ took_events = True
+ self.microstep(list(enabled_transitions), internal_event)
- enabled_transitions = self.select_transitions(external_event)
- self._debug("%s Enabled transitions: %s", self._log_id, enabled_transitions)
- if enabled_transitions:
- try:
- result = self.microstep(list(enabled_transitions), external_event)
- if first_result is self._sentinel:
- first_result = result
+ def _take_external_events(self, first_result):
+ """Take external events in due order until the queue is empty or holds one not yet due.
- except Exception:
- # We clear the queue as we don't have an expected behavior
- # and cannot keep processing
- self.clear()
- raise
+ Returns whether any event was taken from the queue, and ``first_result`` updated with
+ the result of the first event that enabled a transition.
+ """
+ self._debug("%s Macrostep %d: external queue", self._log_id, self._macrostep_count)
+ took_events = False
+ while not self.external_queue.is_empty():
+ self.clear_cache()
+ took_events = True
+ external_event = self.external_queue.pop()
+ current_time = monotonic()
+ if external_event.execution_time > current_time:
+ self.put(external_event, _delayed=True)
+ sleep(self.sm._loop_sleep_in_ms)
+ # Break to the macrostep so internal events and eventless
+ # transitions can be processed while we wait.
+ break
- else:
- if not self.sm.allow_event_without_transition:
- raise TransitionNotAllowed(external_event.event, self.sm.configuration)
+ self._macrostep_count += 1
+ self._microstep_count = 0
+ self._debug(
+ "%s macrostep %d: event=%s",
+ self._log_id,
+ self._macrostep_count,
+ external_event.event,
+ )
- finally:
- self._processing.release()
- self._debug("%s Processing loop ended", self._log_id)
- return first_result if first_result is not self._sentinel else None
+ result = self._take_external_event(external_event)
+ if first_result is self._sentinel:
+ first_result = result
+ return took_events, first_result
+
+ def _take_external_event(self, external_event: TriggerData):
+ """Run one external event.
+
+ Returns the event's microstep result, or the sentinel when it enabled no transition.
+ """
+ # Finalize + autoforward for active invocations
+ self._invoke_manager.handle_external_event(external_event)
+
+ enabled_transitions = self.select_transitions(external_event)
+ self._debug("%s Enabled transitions: %s", self._log_id, enabled_transitions)
+ if not enabled_transitions:
+ if not self.sm.allow_event_without_transition:
+ raise TransitionNotAllowed(external_event.event, self.sm.configuration)
+ return self._sentinel
+
+ try:
+ return self.microstep(list(enabled_transitions), external_event)
+ except Exception:
+ # We clear the queue as we don't have an expected behavior
+ # and cannot keep processing
+ self.clear()
+ raise
+
+ def select_eventless_transitions(self, trigger_data: TriggerData):
+ """
+ Select the eventless transitions that match the trigger data.
+ """
+ return self._select_transitions(trigger_data, lambda t, _e: t.is_eventless)
+
+ def select_transitions(self, trigger_data: TriggerData) -> OrderedSet[Transition]:
+ """
+ Select the transitions that match the trigger data.
+ """
+ return self._select_transitions(trigger_data, lambda t, e: t.match(e))
+
+ def _first_transition_that_matches(
+ self,
+ state: State,
+ trigger_data: TriggerData,
+ predicate: Callable,
+ ) -> "Transition | None":
+ for s in chain([state], state.ancestors()):
+ transition: Transition
+ for transition in s.transitions:
+ if (
+ not transition.initial
+ and predicate(transition, trigger_data.event)
+ and self._conditions_match(transition, trigger_data)
+ ):
+ return transition
+ return None
+
+ def _select_transitions(
+ self, trigger_data: TriggerData, predicate: Callable
+ ) -> OrderedSet[Transition]:
+ """Select the transitions that match the trigger data."""
+ enabled_transitions = OrderedSet[Transition]()
+
+ # Get atomic states, TODO: sorted by document order
+ atomic_states = (state for state in self.sm.configuration if state.is_atomic)
+
+ for state in atomic_states:
+ transition = self._first_transition_that_matches(state, trigger_data, predicate)
+ if transition is not None:
+ enabled_transitions.add(transition)
+
+ return self._filter_conflicting_transitions(enabled_transitions)
+
+ def microstep(self, transitions: list[Transition], trigger_data: TriggerData):
+ """Process a single set of transitions in a 'lock step'.
+ This includes exiting states, executing transition content, and entering states.
+ """
+ self._microstep_count += 1
+ self._debug(
+ "%s macro:%d micro:%d transitions: %s",
+ self._log_id,
+ self._macrostep_count,
+ self._microstep_count,
+ transitions,
+ )
+ previous_configuration = self.sm.configuration
+ try:
+ result = self._execute_transition_content(
+ transitions, trigger_data, lambda t: t.before.key
+ )
+
+ states_to_exit = self._exit_states(transitions, trigger_data)
+ result += self._enter_states(
+ transitions, trigger_data, states_to_exit, previous_configuration
+ )
+ except InvalidDefinition:
+ self.sm.configuration = previous_configuration
+ raise
+ except Exception as e:
+ self.sm.configuration = previous_configuration
+ self._handle_error(e, trigger_data)
+ return None
+
+ try:
+ self._execute_transition_content(
+ transitions,
+ trigger_data,
+ lambda t: t.after.key,
+ set_target_as_state=True,
+ )
+ except InvalidDefinition:
+ raise
+ except Exception as e:
+ self._handle_error(e, trigger_data)
+
+ if len(result) == 0:
+ result = None
+ elif len(result) == 1:
+ result = result[0]
+
+ return result
+
+ def _exit_states(
+ self, enabled_transitions: list[Transition], trigger_data: TriggerData
+ ) -> OrderedSet[State]:
+ """Compute and process the states to exit for the given transitions."""
+ ordered_states, result = self._prepare_exit_states(enabled_transitions)
+ on_error = self._on_error_handler()
+
+ for info in ordered_states:
+ # Cancel invocations for this state before executing exit handlers.
+ self._invoke_manager.cancel_for_state(info.state)
+
+ args, kwargs = self._get_args_kwargs(info.transition, trigger_data, source=info.state)
+
+ # Execute `onexit` handlers — same per-block error isolation as onentry.
+ self._debug("%s Exiting state: %s", self._log_id, info.state)
+ self.sm._callbacks.call(info.state.exit.key, *args, on_error=on_error, **kwargs)
+
+ self._remove_state_from_configuration(info.state)
+
+ return result
def enabled_events(self, *args, **kwargs):
sm = self.sm
diff --git a/statemachine/event_data.py b/statemachine/event_data.py
index 9eebfe41..07747721 100644
--- a/statemachine/event_data.py
+++ b/statemachine/event_data.py
@@ -1,6 +1,7 @@
from dataclasses import dataclass
from dataclasses import field
-from time import time
+from itertools import count
+from time import monotonic
from typing import TYPE_CHECKING
from typing import Any
@@ -10,6 +11,9 @@
from .statemachine import StateChart
from .transition import Transition
+_sequence = count()
+"""Numbers each trigger as it is created, so triggers due at the same time keep their order."""
+
@dataclass(order=True)
class TriggerData:
@@ -25,7 +29,13 @@ class TriggerData:
"""
execution_time: float = field(default=0.0)
- """The time at which the :ref:`Event` should run."""
+ """The time at which the :ref:`Event` should run, on the :func:`time.monotonic` clock, so
+ that a step of the wall clock neither holds nor hastens it."""
+
+ sequence: int = field(default=0, init=False)
+ """The order this trigger was created in, which breaks a tie between triggers due at the
+ same time: the queues are priority queues, and without it two triggers stamped alike come
+ out in no particular order."""
model: Any = field(init=False, compare=False)
"""A reference to the underlying model that holds the current :ref:`State`."""
@@ -43,10 +53,16 @@ class TriggerData:
result (or exception), allowing the caller to ``await`` it.
"""
+ cancelled: bool = field(default=False, compare=False, init=False)
+ """Whether :meth:`~statemachine.StateChart.cancel_event` cancelled this trigger. A cancelled
+ trigger stays in its queue, so the triggers around it keep their due order, and is dropped
+ when it reaches the head."""
+
def __post_init__(self):
self.model = self.machine.model
delay = self.event.delay if self.event and self.event.delay else 0
- self.execution_time = time() + (delay / 1000)
+ self.execution_time = monotonic() + (delay / 1000)
+ self.sequence = next(_sequence)
@dataclass
diff --git a/statemachine/statemachine.py b/statemachine/statemachine.py
index 5277c4fc..2f88ab1e 100644
--- a/statemachine/statemachine.py
+++ b/statemachine/statemachine.py
@@ -5,6 +5,7 @@
from typing import Any
from typing import Generic
from typing import TypeVar
+from typing import cast
from statemachine.orderedset import OrderedSet
@@ -146,7 +147,7 @@ def __init__(
listeners: "list[object] | None" = None,
**kwargs: Any,
):
- self.model: TModel = model if model is not None else Model() # type: ignore[assignment]
+ self.model: TModel = model if model is not None else cast(TModel, Model())
"""The external model object that holds domain state, or an internal
:class:`Model` instance when none is provided. See :ref:`domain models`."""
@@ -225,7 +226,11 @@ def activate_initial_state(self) -> Any:
return run_async_from_sync(result)
def _processing_loop(self, caller_future: "Any | None" = None) -> Any:
- result = self._engine.processing_loop(caller_future)
+ # Only the async engine gives an event a future, for its caller to await the result by.
+ if isinstance(self._engine, AsyncEngine):
+ result = self._engine.processing_loop(caller_future)
+ else:
+ result = self._engine.processing_loop()
if not isawaitable(result):
return result
return run_async_from_sync(result)
@@ -259,7 +264,7 @@ def __getstate__(self):
def __setstate__(self, state: dict[str, Any]) -> None:
listeners = state.pop("_listeners")
- self.__dict__.update(state) # type: ignore[attr-defined]
+ vars(self).update(state)
self._callbacks = CallbacksRegistry()
self._config = self._build_configuration()
self._listeners = {}
@@ -273,11 +278,11 @@ def __setstate__(self, state: dict[str, Any]) -> None:
self._engine.start()
def _get_initial_configuration(self):
- initial_state_values = (
- self.start_configuration_values
- if self.start_configuration_values
- else [self.initial_state.value] # type: ignore[union-attr]
- )
+ initial_state_values = self.start_configuration_values
+ if not initial_state_values:
+ # Only a machine with no states has no initial state, and none can be created.
+ assert self.initial_state is not None
+ initial_state_values = [self.initial_state.value]
try:
return [self.states_map[value] for value in initial_state_values]
except KeyError as err:
@@ -362,7 +367,7 @@ def _repr_html_(self):
return f'{self._repr_svg_()}
'
def _repr_svg_(self):
- return self._graph().create_svg().decode() # type: ignore[attr-defined]
+ return format(self, "svg")
def _graph(self):
from .contrib.diagram import DotGraphMachine
@@ -507,7 +512,10 @@ def raise_(
return self.send(event, *args, delay=delay, send_id=send_id, internal=True, **kwargs)
def cancel_event(self, send_id: str):
- """Cancel all the delayed events with the given ``send_id``."""
+ """Cancel all the delayed events with the given ``send_id``.
+
+ A caller awaiting one of them gets ``None``.
+ """
self._engine.cancel_event(send_id)
@property
diff --git a/tests/scxml/test_parallel_regions.py b/tests/scxml/test_parallel_regions.py
new file mode 100644
index 00000000..6f149827
--- /dev/null
+++ b/tests/scxml/test_parallel_regions.py
@@ -0,0 +1,39 @@
+"""A transition that targets a parallel state and a state inside one of its regions.
+
+The region holding the targeted state enters that state and not its initial one; every other
+region enters its initial state.
+
+Theme: the beacons, one of them already lit.
+"""
+
+import pytest
+from statemachine.io.scxml.processor import SCXMLProcessor
+
+pytestmark = pytest.mark.scxml
+
+BEACONS_SCXML = """
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+"""
+
+
+def test_a_region_with_a_targeted_state_does_not_also_enter_its_initial_state():
+ processor = SCXMLProcessor()
+ processor.parse_scxml("beacons", BEACONS_SCXML)
+ sm = processor.start()
+
+ sm.send("light")
+
+ assert set(sm.configuration_values) == {"beacons", "east", "east_lit", "west", "west_dark"}
diff --git a/tests/test_cancel_event.py b/tests/test_cancel_event.py
new file mode 100644
index 00000000..a1e1b220
--- /dev/null
+++ b/tests/test_cancel_event.py
@@ -0,0 +1,311 @@
+"""Cancelling delayed events with ``cancel_event()``.
+
+A cancelled trigger stays where it is in its queue, so the triggers around it keep their due
+order, and is dropped when it reaches the head. A caller awaiting it is answered with ``None``.
+
+Theme: signal fires, some of them called off before they are lit.
+"""
+
+import asyncio
+
+import pytest
+from statemachine.engines import async_
+from statemachine.engines import sync
+from statemachine.event import BoundEvent
+
+from statemachine import State
+from statemachine import StateChart
+from statemachine import event_data
+
+_CLOCKS = ("statemachine.event_data", "statemachine.engines.sync", "statemachine.engines.async_")
+
+
+class Signals(StateChart):
+ idle = State(initial=True)
+
+ signal = idle.to.itself(internal=True, on="answer")
+ stand_down = idle.to.itself(internal=True, on="call_off")
+
+ def __init__(self, *args, **kwargs):
+ self.answered = []
+ super().__init__(*args, **kwargs)
+
+ def answer(self, due):
+ self.answered.append(due)
+
+ def call_off(self, which):
+ self.cancel_event(which)
+
+
+class AsyncSignals(Signals):
+ async def answer(self, due):
+ self.answered.append(due)
+
+
+@pytest.fixture(autouse=True)
+def clock(monkeypatch):
+ """Run every test here on a monotonic clock that moves only when the test moves it."""
+ now = [1000.0]
+ for module in _CLOCKS:
+ monkeypatch.setattr(f"{module}.monotonic", lambda: now[0])
+ return now
+
+
+def signal(sm, due, send_id=None):
+ """Queue a signal due in ``due`` ms, without running the processing loop."""
+ return BoundEvent(id="signal", name="Signal", delay=due, _sm=sm).put(send_id=send_id, due=due)
+
+
+def test_every_engine_reads_the_test_clock(clock):
+ modules = (event_data, sync, async_)
+ assert [module.monotonic() for module in modules] == [1000.0, 1000.0, 1000.0]
+
+ clock[0] += 0.5
+
+ assert [module.monotonic() for module in modules] == [1000.5, 1000.5, 1000.5]
+
+
+@pytest.mark.timeout(10)
+@pytest.mark.parametrize(
+ "cancelled",
+ [
+ pytest.param("s0", id="first-due-and-first-queued"),
+ pytest.param("s6", id="due-in-the-middle"),
+ pytest.param("s8", id="last-due"),
+ pytest.param("s9", id="last-queued"),
+ ],
+)
+async def test_cancelling_a_delayed_event_keeps_the_rest_in_due_order(sm_runner, cancelled, clock):
+ sm = await sm_runner.start(Signals)
+ delays = [10, 50, 20, 60, 70, 30, 40, 80, 90, 15]
+ for number, delay in enumerate(delays):
+ signal(sm, delay, f"s{number}")
+
+ sm.cancel_event(cancelled)
+ clock[0] += 1.0
+ await sm_runner.processing_loop(sm)
+
+ assert sm.answered == sorted(set(delays) - {delays[int(cancelled[1:])]})
+
+
+@pytest.mark.timeout(10)
+async def test_cancelling_a_send_id_cancels_every_event_sent_with_it(sm_runner, clock):
+ sm = await sm_runner.start(Signals)
+ signal(sm, 10, "dawn")
+ signal(sm, 20, "dawn")
+ signal(sm, 30, "dusk")
+ signal(sm, 40, "dawn")
+
+ sm.cancel_event("dawn")
+ clock[0] += 1.0
+ await sm_runner.processing_loop(sm)
+
+ assert sm.answered == [30]
+
+
+@pytest.mark.timeout(10)
+async def test_cancelling_an_unknown_send_id_cancels_nothing(sm_runner, clock):
+ sm = await sm_runner.start(Signals)
+ signal(sm, 10, "dawn")
+ signal(sm, 20, "dusk")
+
+ sm.cancel_event("noon")
+ clock[0] += 1.0
+ await sm_runner.processing_loop(sm)
+
+ assert sm.answered == [10, 20]
+
+
+@pytest.mark.timeout(10)
+async def test_cancelling_two_send_ids_in_turn_cancels_each_and_nothing_else(sm_runner, clock):
+ sm = await sm_runner.start(Signals)
+ signal(sm, 10, "dawn")
+ signal(sm, 20, "dusk")
+ signal(sm, 30, "noon")
+
+ sm.cancel_event("dawn")
+ sm.cancel_event("dusk")
+ clock[0] += 1.0
+ await sm_runner.processing_loop(sm)
+
+ assert sm.answered == [30]
+
+
+@pytest.mark.timeout(10)
+async def test_a_cancel_with_nothing_queued_does_not_cancel_what_is_sent_later(sm_runner, clock):
+ sm = await sm_runner.start(Signals)
+
+ sm.cancel_event("dawn")
+ signal(sm, 10, "dawn")
+ clock[0] += 1.0
+ await sm_runner.processing_loop(sm)
+
+ assert sm.answered == [10]
+
+
+@pytest.mark.timeout(10)
+@pytest.mark.parametrize(
+ ("sent", "cancelled"),
+ [
+ pytest.param("dawn", "", id="empty"),
+ pytest.param("dawn", " ", id="space"),
+ pytest.param("dawn", " dawn", id="leading-space"),
+ pytest.param("dawn", "dawn ", id="trailing-space"),
+ pytest.param("dawn", "dawn\n", id="newline"),
+ pytest.param("dawn\r\n", "dawn\n", id="crlf-and-lf"),
+ pytest.param("dawn", "Dawn", id="case"),
+ pytest.param("dawn", "dawn\u200b", id="zero-width-space"),
+ pytest.param("dawn", "\ufeffdawn", id="byte-order-mark"),
+ pytest.param("da\0wn", "da", id="up-to-a-nul"),
+ pytest.param("caf\u00e9", "cafe\u0301", id="composed-and-decomposed"),
+ ],
+)
+async def test_a_send_id_cancels_only_events_sent_with_exactly_that_id(
+ sm_runner, sent, cancelled, clock
+):
+ sm = await sm_runner.start(Signals)
+ signal(sm, 10, sent)
+
+ sm.cancel_event(cancelled)
+ clock[0] += 1.0
+ await sm_runner.processing_loop(sm)
+
+ assert sm.answered == [10]
+
+
+@pytest.mark.timeout(10)
+@pytest.mark.parametrize(
+ "send_id",
+ [
+ pytest.param("", id="empty"),
+ pytest.param(" ", id="space"),
+ pytest.param("dawn\n", id="newline"),
+ pytest.param("da\0wn", id="nul"),
+ pytest.param("cafe\u0301", id="decomposed"),
+ pytest.param("\U0001f525", id="astral"),
+ pytest.param("\u6681", id="cjk"),
+ pytest.param("\u202edawn", id="right-to-left-override"),
+ pytest.param("x" * 255, id="255-characters"),
+ pytest.param("x" * 256, id="256-characters"),
+ pytest.param("x" * 257, id="257-characters"),
+ pytest.param("x" * 1024, id="1024-characters"),
+ pytest.param("x" * 2048, id="2048-characters"),
+ ],
+)
+async def test_any_string_is_a_send_id_that_cancels_only_its_own_events(sm_runner, send_id, clock):
+ sm = await sm_runner.start(Signals)
+ signal(sm, 10, send_id)
+ signal(sm, 20)
+
+ sm.cancel_event(send_id)
+ clock[0] += 1.0
+ await sm_runner.processing_loop(sm)
+
+ assert sm.answered == [20]
+
+
+@pytest.mark.timeout(10)
+async def test_cancelling_marks_each_trigger_sent_with_the_id(sm_runner):
+ sm = await sm_runner.start(Signals)
+ dawn = signal(sm, 10, "dawn")
+ dusk = signal(sm, 20, "dusk")
+
+ sm.cancel_event("dawn")
+
+ assert (dawn.cancelled, dusk.cancelled) == (True, False)
+
+
+@pytest.mark.timeout(10)
+async def test_an_event_can_be_cancelled_from_a_callback(sm_runner, clock):
+ sm = await sm_runner.start(Signals)
+ BoundEvent(id="stand_down", name="Stand down", _sm=sm).put(which="dawn")
+ signal(sm, 10, "dawn")
+ signal(sm, 20, "dusk")
+
+ clock[0] += 1.0
+ await sm_runner.processing_loop(sm)
+
+ assert sm.answered == [20]
+
+
+@pytest.mark.timeout(10)
+async def test_the_loop_never_waits_for_a_cancelled_event(sm_runner, monkeypatch):
+ sm = await sm_runner.start(Signals)
+ dawn = signal(sm, 10, "dawn")
+ sm.cancel_event("dawn")
+
+ def checked():
+ raise AssertionError("the loop checked whether a cancelled event was due")
+
+ for module in _CLOCKS[1:]:
+ monkeypatch.setattr(f"{module}.monotonic", checked)
+ await sm_runner.processing_loop(sm)
+
+ assert (dawn.cancelled, sm.answered) == (True, [])
+
+
+async def awaiting(sm, started=True):
+ """Queue two delayed signals, each awaited by its own task, and return the tasks.
+
+ The first task runs the processing loop, which waits on the clock for its signal. The
+ second signal, sent with ``send_id="dawn"`` and due first, is awaited through its trigger's
+ future by the second task, as ``send()`` awaits it, once that task has started. The
+ trigger is returned too.
+ """
+ holding = asyncio.create_task(sm.send("signal", delay=20, due=20))
+ await asyncio.sleep(0)
+ dawn = signal(sm, 10, "dawn")
+ waiting = asyncio.create_task(sm._processing_loop(dawn.future))
+ if started:
+ await asyncio.sleep(0)
+ return holding, waiting, dawn
+
+
+def settled(future):
+ """How a settled future was settled: cancelled, or answered with a value."""
+ if future.cancelled():
+ return "cancelled"
+ return f"answered with {future.result()!r}"
+
+
+@pytest.mark.timeout(10)
+@pytest.mark.parametrize("cancels", [1, 2])
+async def test_a_caller_awaiting_a_cancelled_event_gets_none(cancels, clock):
+ sm = AsyncSignals()
+ await sm.activate_initial_state()
+ holding, waiting, dawn = await awaiting(sm)
+
+ for _ in range(cancels):
+ sm.cancel_event("dawn")
+ await asyncio.sleep(0)
+
+ assert (settled(dawn.future), settled(waiting)) == ("answered with None", "answered with None")
+ clock[0] += 1.0
+ await holding
+ assert sm.answered == [20]
+
+
+@pytest.mark.timeout(10)
+@pytest.mark.parametrize(
+ ("started", "settled_as"),
+ [
+ pytest.param(False, "answered with None", id="before-it-runs"),
+ pytest.param(True, "cancelled", id="while-it-awaits"),
+ ],
+)
+async def test_a_caller_that_stops_waiting_leaves_its_event_cancellable(
+ started, settled_as, clock
+):
+ sm = AsyncSignals()
+ await sm.activate_initial_state()
+ holding, waiting, dawn = await awaiting(sm, started)
+
+ waiting.cancel()
+ with pytest.raises(asyncio.CancelledError):
+ await waiting
+ sm.cancel_event("dawn")
+ clock[0] += 1.0
+ await holding
+
+ assert settled(dawn.future) == settled_as
+ assert sm.answered == [20]
diff --git a/tests/test_event_ordering.py b/tests/test_event_ordering.py
new file mode 100644
index 00000000..f10842b8
--- /dev/null
+++ b/tests/test_event_ordering.py
@@ -0,0 +1,58 @@
+"""Events due at the same moment are processed in the order they were sent.
+
+Both queues are priority queues ordered by each trigger's due time. Triggers
+stamped with the same time, as a coarse clock stamps events sent close together,
+must still come out in the order they went in.
+
+Theme: the Riders of Rohan answering the muster, each in turn.
+"""
+
+import pytest
+from statemachine.event import BoundEvent
+
+from statemachine import State
+from statemachine import StateChart
+
+
+class Muster(StateChart):
+ camp = State(initial=True)
+
+ call = camp.to.itself(internal=True, on="call_riders")
+ ride = camp.to.itself(internal=True, on="answer")
+
+ def __init__(self, *args, **kwargs):
+ self.answered = []
+ super().__init__(*args, **kwargs)
+
+ def call_riders(self, riders):
+ for rider in range(riders):
+ BoundEvent(id="ride", name="Ride", internal=True, _sm=self).put(rider=rider)
+
+ def answer(self, rider):
+ self.answered.append(rider)
+
+
+@pytest.fixture(autouse=True)
+def stamped_alike(monkeypatch):
+ """Stamp every trigger in these tests with the same time."""
+ monkeypatch.setattr("statemachine.event_data.monotonic", lambda: 1000.0)
+
+
+@pytest.mark.timeout(10)
+@pytest.mark.parametrize("riders", [0, 1, 2, 20])
+class TestEventsDueTogether:
+ async def test_internal_events_run_in_the_order_raised(self, sm_runner, riders):
+ sm = await sm_runner.start(Muster)
+
+ await sm_runner.send(sm, "call", riders=riders)
+
+ assert sm.answered == list(range(riders))
+
+ async def test_external_events_run_in_the_order_sent(self, sm_runner, riders):
+ sm = await sm_runner.start(Muster)
+ for rider in range(riders):
+ BoundEvent(id="ride", name="Ride", _sm=sm).put(rider=rider)
+
+ await sm_runner.processing_loop(sm)
+
+ assert sm.answered == list(range(riders))
diff --git a/tests/test_invoke.py b/tests/test_invoke.py
index e31baad2..c9dc271c 100644
--- a/tests/test_invoke.py
+++ b/tests/test_invoke.py
@@ -1294,6 +1294,27 @@ def test_stop_child_machine_exception_swallowed(self):
_stop_child_machine(child)
+class TestEngineStopCleanup:
+ """Test BaseEngine.stop cancel_all exception handling."""
+
+ def test_stop_swallows_cancel_all_exception_and_stops(self):
+ """stop() swallows exceptions from cancel_all and still stops the engine."""
+
+ class SM(StateChart):
+ s1 = State(initial=True)
+ s2 = State(final=True)
+
+ go = s1.to(s2)
+
+ sm = SM()
+ engine = sm._engine
+ engine._invoke_manager.cancel_all = lambda: (_ for _ in ()).throw(RuntimeError("boom"))
+
+ engine.stop()
+
+ assert engine.running is False
+
+
class TestEngineDelCleanup:
"""Test BaseEngine.__del__ cancel_all exception handling."""
diff --git a/tests/test_statechart_delayed.py b/tests/test_statechart_delayed.py
index f1f7c21b..635fa974 100644
--- a/tests/test_statechart_delayed.py
+++ b/tests/test_statechart_delayed.py
@@ -8,13 +8,20 @@
"""
import asyncio
+import time
+from itertools import count
import pytest
+from statemachine.engines import async_
+from statemachine.engines import sync
from statemachine.event import BoundEvent
from statemachine import Event
from statemachine import State
from statemachine import StateChart
+from statemachine import event_data
+
+_CLOCKS = ("statemachine.event_data", "statemachine.engines.sync", "statemachine.engines.async_")
@pytest.mark.timeout(10)
@@ -97,3 +104,131 @@ class BeaconsOfGondor(StateChart):
await asyncio.sleep(0.1)
await sm_runner.processing_loop(sm)
assert "lit" in sm.configuration_values
+
+ async def test_delay_is_measured_on_the_monotonic_clock(self, sm_runner, monkeypatch):
+ """A delay comes due by ``time.monotonic``, which a wall-clock step cannot move.
+
+ The trigger is stamped and checked on the same monotonic clock: here a fake one that
+ moves only when the test moves it. A stamp on one clock checked against another would
+ hold the beacon unlit, and the wall clock fails the test if anything reads it.
+ """
+
+ class BeaconsOfGondor(StateChart):
+ dark = State(initial=True)
+ lit = State(final=True)
+
+ light = dark.to(lit)
+
+ def wall_clock():
+ raise AssertionError("a trigger was stamped or checked on the wall clock")
+
+ monkeypatch.setattr("time.time", wall_clock)
+ now = [1000.0]
+ for module in _CLOCKS:
+ monkeypatch.setattr(f"{module}.monotonic", lambda: now[0])
+ monkeypatch.setattr(f"{module}.time", wall_clock, raising=False)
+ sm = await sm_runner.start(BeaconsOfGondor)
+ BoundEvent(id="light", name="Light", delay=50, _sm=sm).put()
+ now[0] += 0.05
+
+ await sm_runner.processing_loop(sm)
+
+ assert "lit" in sm.configuration_values
+
+ def test_triggers_are_stamped_and_checked_on_the_monotonic_clock(self):
+ clocks = (event_data.monotonic, sync.monotonic, async_.monotonic)
+
+ assert clocks == (time.monotonic, time.monotonic, time.monotonic)
+
+ @pytest.mark.parametrize(
+ ("delay", "due"),
+ [
+ pytest.param(None, 1000.0, id="none"),
+ pytest.param(0, 1000.0, id="zero"),
+ pytest.param(1, 1000.001, id="one-millisecond"),
+ pytest.param(50, 1000.05, id="fifty-milliseconds"),
+ pytest.param(-50, 999.95, id="negative-already-due"),
+ ],
+ )
+ def test_a_delay_in_milliseconds_comes_due_that_many_thousandths_later(
+ self, monkeypatch, delay, due
+ ):
+ class BeaconsOfGondor(StateChart):
+ dark = State(initial=True)
+ lit = State(final=True)
+
+ light = dark.to(lit)
+
+ monkeypatch.setattr("statemachine.event_data.monotonic", lambda: 1000.0)
+ sm = BeaconsOfGondor()
+
+ trigger = BoundEvent(id="light", name="Light", delay=delay, _sm=sm).put()
+
+ assert trigger.execution_time == pytest.approx(due, abs=1e-9)
+
+ async def test_a_delayed_event_queued_from_a_callback_fires_when_due(
+ self, sm_runner, monkeypatch
+ ):
+ """The callback runs inside the processing loop, which then waits on the event it queued.
+
+ Every reading of the clock is 10 ms later than the last, so the beacon comes due after
+ a few turns of the loop however long each takes.
+ """
+
+ class BeaconsOfGondor(StateChart):
+ dark = State(initial=True)
+ first_lit = State()
+ all_lit = State(final=True)
+
+ light_first = dark.to(first_lit, after="spread")
+ light_all = first_lit.to(all_lit)
+
+ def spread(self):
+ BoundEvent(id="light_all", name="Light all", delay=50, _sm=self).put()
+
+ readings = count()
+ for module in _CLOCKS:
+ monkeypatch.setattr(f"{module}.monotonic", lambda: 1000.0 + next(readings) / 100)
+ sm = await sm_runner.start(BeaconsOfGondor)
+
+ await sm_runner.send(sm, "light_first")
+
+ assert sm.configuration_values == {"all_lit"}
+
+ @pytest.mark.parametrize(
+ "started",
+ [pytest.param(False, id="before-it-runs"), pytest.param(True, id="while-it-awaits")],
+ )
+ async def test_a_caller_that_stops_waiting_does_not_stop_its_delayed_event(
+ self, monkeypatch, started
+ ):
+ class Beacon(StateChart):
+ dark = State(initial=True)
+
+ signal = dark.to.itself(internal=True, on="flare")
+
+ def __init__(self, *args, **kwargs):
+ self.flares = []
+ super().__init__(*args, **kwargs)
+
+ async def flare(self, which):
+ self.flares.append(which)
+
+ now = [1000.0]
+ for module in _CLOCKS:
+ monkeypatch.setattr(f"{module}.monotonic", lambda: now[0])
+ sm = Beacon()
+ await sm.activate_initial_state()
+ holding = asyncio.create_task(sm.send("signal", delay=20, which="second"))
+ await asyncio.sleep(0)
+ waiting = asyncio.create_task(sm.send("signal", delay=10, which="first"))
+ if started:
+ await asyncio.sleep(0)
+
+ waiting.cancel()
+ with pytest.raises(asyncio.CancelledError):
+ await waiting
+ now[0] += 1.0
+ await holding
+
+ assert sm.flares == ["first", "second"]