diff --git a/AGENTS.md b/AGENTS.md index 49062228..fc4db48a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -39,7 +39,14 @@ The engine follows the SCXML run-to-completion (RTC) model with two processing l ### Event queues - `send()` → **external queue** (processed after current macrostep ends). -- `raise_()` → **internal queue** (processed within the current macrostep, before external events). +- `raise_()`, `send(internal=True)`, `Event(internal=True)` → **internal queue** (processed + within the current macrostep, before external events), but only when raised while the + machine runs a step in the same thread or task (`_running_engine` in `engines/base.py`, a + `ContextVar` so it is right for threads and coroutines alike). From outside a step there + is no macrostep to join: the trigger goes to the external queue, so its caller gets the + result, the exception or `TransitionNotAllowed`. +- `delay` and `internal` together raise `InvalidDefinition`: a step takes no time. +- Each external event is followed by its macrostep before the next external event is taken. ### Error handling (`catch_errors_as_events`) diff --git a/docs/async.md b/docs/async.md index 313d03f8..9c3bb367 100644 --- a/docs/async.md +++ b/docs/async.md @@ -155,10 +155,11 @@ resolves each event's future as it processes the queue. Callers that didn't acquire the lock simply `await` their future. ```{note} -Futures are only created for **external** events sent from outside the -processing loop. Events triggered from within callbacks (via `send()` or -`raise_()`) follow the {ref}`run-to-completion ` model — they -are enqueued and processed within the current macrostep. +Futures are created for every event sent from outside the processing loop, +including one declared `internal=True`, which goes to the external queue +when there is no macrostep to join. Events triggered from within callbacks +(via `send()` or `raise_()`) follow the {ref}`run-to-completion ` +model: they are enqueued and processed by the running loop. ``` If an exception occurs during processing (with `catch_errors_as_events=False`), diff --git a/docs/processing_model.md b/docs/processing_model.md index d8181316..7dd99a58 100644 --- a/docs/processing_model.md +++ b/docs/processing_model.md @@ -76,17 +76,51 @@ After the macrostep completes, the engine picks the next event from the ### Event queues -The engine maintains two separate FIFO queues: +The engine maintains two separate queues, each ordered by due time. Events due at the same +moment run in the order they were sent. | Queue | How to enqueue | When processed | |---|---|---| -| **Internal** | {func}`raise_() ` or `send(..., internal=True)` | Within the current macrostep | -| **External** | {func}`send() ` | After the current macrostep ends | +| **Internal** | {func}`raise_() `, `send(..., internal=True)`, or an event declared with `Event(..., internal=True)`, from a callback | Within the current macrostep | +| **External** | {func}`send() `, and any event triggered from outside a callback | After the current macrostep ends | This distinction matters when you trigger events from inside callbacks. Using `raise_()` ensures the event is handled as part of the current processing cycle, while `send()` defers it to after the machine reaches -a stable configuration. +a stable configuration. An event declared `internal=True` behaves like +`raise_()` without spelling it at each call: + +```py +>>> from statemachine import Event, State, StateChart + +>>> class Muster(StateChart): +... camp = State(initial=True) +... riding = State(final=True) +... besieged = State(final=True) +... +... summon = camp.to.itself(internal=True, on="call") +... ride = Event(camp.to(riding), internal=True) +... siege = camp.to(besieged) +... +... def call(self): +... self.send("siege") # after this macrostep +... self.ride() # within this macrostep, so first + +>>> sm = Muster() +>>> sm.send("summon") +>>> sm.riding.is_active +True + +``` + +The internal queue belongs to the macrostep in progress. Triggered from +outside a callback, there is no macrostep to join: an internal event goes +to the external queue, and the call returns its result like `send()`. + +A delay cannot be internal, since a macrostep takes no time. Declaring or +sending an event with both `delay` and `internal=True`, or calling +`raise_()` with a `delay`, raises {class}`~statemachine.exceptions.InvalidDefinition`. +Use `send(..., delay=...)` to deliver an event later. ```{seealso} See {ref}`sending-events` for examples of `send()` vs `raise_()`. diff --git a/docs/releases/3.2.2.md b/docs/releases/3.2.2.md index 47b25b89..81e11eab 100644 --- a/docs/releases/3.2.2.md +++ b/docs/releases/3.2.2.md @@ -117,8 +117,7 @@ Delays are now measured on `time.monotonic()`, which no setting of the wall cloc `Event(dark.to(lit), delay=50)` rebuilt the event without its `delay`, so `BeaconsOfGondor.light.delay` was `0` and the event fired immediately instead of being queued. -The same happened to `internal`. Both are now preserved, although `internal` still has no -effect at trigger time: +The same happened to `internal`. Each is now preserved: ```py >>> class BeaconsOfGondor(StateChart): @@ -132,6 +131,62 @@ effect at trigger time: ``` +### `internal` declared on an `Event` ignored at trigger time + +`go = Event(a.to(b), internal=True)` kept `internal` on the declaration, but neither `sm.go()` +nor `sm.send("go")` carried it to the queue, so the event always went to the external queue. Both +now honor the declaration: triggered from a callback, the event runs within the current +macrostep, before any pending external event. + +Honoring it exposed how the internal queue treated a trigger that no running step had raised. +On the async engine, the first `await sm.go()` on a fresh machine was lost before the initial +state was entered. The caller got `None` instead of the callback result, and no +`TransitionNotAllowed` on a `StateMachine`. The internal queue now receives only what the +machine raises while it runs a step. Triggered from outside, an internal event goes to the +external queue and returns its result, as `send()` does: + +```py +>>> class Beacons(StateChart): +... dark = State(initial=True) +... lit = State(final=True) +... +... light = Event(dark.to(lit, on="kindle"), internal=True) +... +... def kindle(self): +... return "kindled" + +>>> Beacons().light() +'kindled' + +``` + +A delay never reached the internal queue either: `raise_("go", delay=50)` fired at once. A +macrostep takes no time, and SCXML forbids `delay` on a `` to `#_internal`, so a delayed +internal event is now an invalid definition, whether declared, sent, raised, or loaded +from an SCXML, JSON or YAML `` to `#_internal` (also rejected by the JSON Schema): + +```py +>>> class DelayedBeacons(StateChart): +... dark = State(initial=True) +... lit = State(final=True) +... +... light = Event(dark.to(lit), delay=50, internal=True) +Traceback (most recent call last): +... +InvalidDefinition: Event 'light' cannot be both delayed and internal: ... + +``` + +Drop `internal` to deliver the event later on the external queue, or drop `delay` to run it +within the current macrostep. An explicit `send(..., internal=False)` overrides the declaration, so +`send("light", delay=50, internal=False)` delivers a declared-internal event later. + +Contributed by [@chiliec](https://github.com/chiliec). + +[#655](https://github.com/fgmacedo/python-statemachine/issues/655), +[#663](https://github.com/fgmacedo/python-statemachine/pull/663), +[#668](https://github.com/fgmacedo/python-statemachine/pull/668). + ### `is_active` on a state nested in a parallel region Reaching a child state through its parent (`sm.door.shut`) returned the class-level `State`, @@ -167,3 +222,38 @@ Reported by [@MartinBernstorff](https://github.com/MartinBernstorff). [#651](https://github.com/fgmacedo/python-statemachine/issues/651), [#652](https://github.com/fgmacedo/python-statemachine/pull/652). + +### Each external event gets its own macrostep + +With several external events already queued, the engine took them one after the other and +only then ran the internal events and eventless transitions they had enabled. An internal +event raised while handling the first external event ran after the second, breaking the +run-to-completion order the {ref}`processing model ` describes: + +```py +>>> class Beacons(StateChart): +... dark = State(initial=True) +... lit = State(final=True) +... besieged = State(final=True) +... +... signal = dark.to.itself(internal=True, on="relay") +... light = dark.to(lit) +... siege = dark.to(besieged) +... +... def relay(self): +... self.raise_("light") +... self.send("siege") + +>>> sm = Beacons() +>>> sm.send("signal") +>>> sm.lit.is_active +True + +``` + +Each external event is now followed by its macrostep before the next one is taken, on both +engines. Eventless transitions are now checked after every external event instead of once per +batch, so a burst of queued external events costs one guard evaluation each, and an event +queued behind one that reaches a final state is no longer processed. + +[#668](https://github.com/fgmacedo/python-statemachine/pull/668). diff --git a/docs/statechart.md b/docs/statechart.md index e06ed251..3c8e3fe4 100644 --- a/docs/statechart.md +++ b/docs/statechart.md @@ -88,6 +88,24 @@ True ``` +Called from outside a callback, `raise_()` has no macrostep to join, so it +behaves like `send()` and returns the result: + +```py +>>> class Greeter(StateChart): +... idle = State(initial=True) +... done = State(final=True) +... +... greet = idle.to(done, on="say_hello") +... +... def say_hello(self): +... return "hello" + +>>> Greeter().raise_("greet") +'hello' + +``` + Both methods accept arbitrary keyword arguments that are forwarded to all callbacks via {ref}`dependency injection `. @@ -115,6 +133,10 @@ sm.cancel_event("my_timeout") On the async engine, a coroutine awaiting the `send()` of a cancelled event gets `None`. +Only `send()` takes a delay. An internal event runs within the current macrostep, +which takes no time, so `raise_(..., delay=...)` and `send(..., delay=..., internal=True)` +raise {class}`~statemachine.exceptions.InvalidDefinition`. + ```{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 9ba27f89..4c17a0e1 100644 --- a/statemachine/engines/async_.py +++ b/statemachine/engines/async_.py @@ -1,5 +1,4 @@ import asyncio -import contextvars from collections.abc import Callable from itertools import chain from time import monotonic @@ -13,19 +12,12 @@ from ..state import State from .base import _ERROR_EXECUTION from .base import BaseEngine +from .base import _processing_step +from .base import _running_engine if TYPE_CHECKING: from ..transition import Transition -# ContextVar to distinguish reentrant calls (from within callbacks) from -# concurrent external calls. asyncio propagates context to child tasks -# (e.g., those created by asyncio.gather in the callback system), so a -# ContextVar set in the processing loop is visible in all callbacks. -# Independent external coroutines have their own context where this is False. -_in_processing_loop: contextvars.ContextVar[bool] = contextvars.ContextVar( - "_in_processing_loop", default=False -) - class AsyncEngine(BaseEngine): """Async engine with full StateChart support. @@ -35,16 +27,16 @@ class AsyncEngine(BaseEngine): """ def put(self, trigger_data: TriggerData, internal: bool = False): - """Override to attach an asyncio.Future for external events. + """Override to attach an asyncio.Future to a trigger put from outside the loop. - Futures are only created when: - - The event is external (not internal) + Such a trigger always lands on the external queue, and its caller awaits the future + for its result. Futures are only created when: - No future is already attached - There is a running asyncio loop - The call is NOT from within the processing loop (reentrant calls from callbacks must not get futures, as that would deadlock) """ - if not internal and trigger_data.future is None and not _in_processing_loop.get(): + if trigger_data.future is None and _running_engine.get() is None: try: loop = asyncio.get_running_loop() trigger_data.future = loop.create_future() @@ -357,21 +349,23 @@ async def processing_loop(self, caller_future: "asyncio.Future[object] | None" = return await caller_future return None - _ctx_token = _in_processing_loop.set(True) self._debug("%s Processing loop started: %s", self._log_id, self.sm.current_state_value) first_result = self._sentinel try: - took_events = True - while took_events and self.running: - self.clear_cache() - took_events = await self._run_macrostep() + with _processing_step(self): + took_events = True + while took_events and self.running: + self.clear_cache() + 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() + # Spawn invoke handlers for states entered during this macrostep. + await self._invoke_manager.spawn_pending_async() + self._check_root_final_state() - took_external_events, first_result = await self._take_external_events(first_result) - took_events = took_events or took_external_events + took_external_events, first_result = await self._take_due_external_event( + first_result + ) + took_events = took_events or took_external_events except Exception as exc: if caller_future is not None: @@ -385,7 +379,6 @@ async def processing_loop(self, caller_future: "asyncio.Future[object] | None" = else: raise finally: - _in_processing_loop.reset(_ctx_token) self._processing.release() self._debug("%s Processing loop ended", self._log_id) @@ -424,45 +417,44 @@ async def _run_macrostep(self) -> bool: 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. + async def _take_due_external_event(self, first_result): + """Take the external event at the head of the queue, if it is 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. + One at a time, so the macrostep that follows runs the internal events and eventless + transitions it enabled before the next external event is taken. + + Returns whether the queue held an event, 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_due(monotonic()) - if external_event is None: - 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, - ) + if self.external_queue.is_empty(): + return False, first_result + + self.clear_cache() + external_event = self.external_queue.pop_due(monotonic()) + if external_event is None: + await asyncio.sleep(self.sm._loop_sleep_in_ms) + return True, first_result + + 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. + if external_event.event == "__initial__": + transitions = self._initial_transitions(external_event) + await self._enter_states(transitions, external_event, OrderedSet(), OrderedSet()) + return True, first_result - # 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 + result = await self._take_external_event(external_event) + if first_result is self._sentinel: + first_result = result + return True, first_result async def _take_external_event(self, external_event: TriggerData): """Run one external event and settle its future. diff --git a/statemachine/engines/base.py b/statemachine/engines/base.py index 0fed1acb..81cafdfa 100644 --- a/statemachine/engines/base.py +++ b/statemachine/engines/base.py @@ -1,5 +1,8 @@ import logging from collections.abc import Callable +from collections.abc import Iterator +from contextlib import contextmanager +from contextvars import ContextVar from dataclasses import dataclass from dataclasses import field from heapq import heappop @@ -26,6 +29,24 @@ logger = logging.getLogger(__name__) +_running_engine: "ContextVar[BaseEngine | None]" = ContextVar("_running_engine", default=None) +"""The engine whose step the current thread or task is running, if any. + +A context variable is per thread and per asyncio task: asyncio copies it into the tasks a +step creates, so callbacks see it, while another thread or an independent coroutine sending +to the same machine does not. +""" + + +@contextmanager +def _processing_step(engine: "BaseEngine") -> Iterator[None]: + """Mark the current context as running a step of ``engine``, until the block ends.""" + token = _running_engine.set(engine) + try: + yield + finally: + _running_engine.reset(token) + @dataclass(frozen=True, unsafe_hash=True, eq=True) class StateTransition: @@ -133,10 +154,17 @@ def clear_cache(self): self._cache.clear() def put(self, trigger_data: TriggerData, internal: bool = False): - """Put the trigger on the queue without blocking the caller.""" + """Put the trigger on the queue without blocking the caller. + + An ``internal`` trigger joins the internal queue only when this machine raises it while + running a step. Put from outside, including from a step of another machine, there is + no macrostep to join, so it goes to the external queue, where its caller gets its + result. + """ if not self.running and not self.sm.allow_event_without_transition: raise TransitionNotAllowed(trigger_data.event, self.sm.configuration) + internal = internal and _running_engine.get() is self if internal: self.internal_queue.put(trigger_data) else: diff --git a/statemachine/engines/sync.py b/statemachine/engines/sync.py index e15c8d8f..d4bcc334 100644 --- a/statemachine/engines/sync.py +++ b/statemachine/engines/sync.py @@ -12,6 +12,7 @@ from ..state import State from ..transition import Transition from .base import BaseEngine +from .base import _processing_step class SyncEngine(BaseEngine): @@ -38,7 +39,8 @@ def activate_initial_state(self, **kwargs): transitions = self._initial_transitions(trigger_data) self._processing.acquire(blocking=False) try: - self._enter_states(transitions, trigger_data, OrderedSet(), OrderedSet()) + with _processing_step(self): + self._enter_states(transitions, trigger_data, OrderedSet(), OrderedSet()) finally: self._processing.release() return self.processing_loop() @@ -64,17 +66,20 @@ def processing_loop(self): self._debug("%s Processing loop started: %s", self._log_id, self.sm.current_state_value) first_result = self._sentinel try: - took_events = True - while took_events and self.running: - self.clear_cache() - took_events = self._run_macrostep() + with _processing_step(self): + took_events = True + while took_events and self.running: + self.clear_cache() + 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() + # Spawn invoke handlers for states entered during this macrostep. + self._invoke_manager.spawn_pending_sync() + self._check_root_final_state() - took_external_events, first_result = self._take_external_events(first_result) - took_events = took_events or took_external_events + took_external_events, first_result = self._take_due_external_event( + first_result + ) + took_events = took_events or took_external_events finally: self._processing.release() self._debug("%s Processing loop ended", self._log_id) @@ -107,37 +112,38 @@ def _run_macrostep(self) -> bool: took_events = True self.microstep(list(enabled_transitions), internal_event) - def _take_external_events(self, first_result): - """Take external events in due order until the queue is empty or holds one not yet due. + def _take_due_external_event(self, first_result): + """Take the external event at the head of the queue, if it is 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. + One at a time, so the macrostep that follows runs the internal events and eventless + transitions it enabled before the next external event is taken. + + Returns whether the queue held an event, 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_due(monotonic()) - if external_event is None: - 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, - ) + if self.external_queue.is_empty(): + return False, first_result + + self.clear_cache() + external_event = self.external_queue.pop_due(monotonic()) + if external_event is None: + sleep(self.sm._loop_sleep_in_ms) + return True, first_result + + self._macrostep_count += 1 + self._microstep_count = 0 + self._debug( + "%s macrostep %d: event=%s", + self._log_id, + self._macrostep_count, + external_event.event, + ) - result = self._take_external_event(external_event) - if first_result is self._sentinel: - first_result = result - return took_events, first_result + result = self._take_external_event(external_event) + if first_result is self._sentinel: + first_result = result + return True, first_result def _take_external_event(self, external_event: TriggerData): """Run one external event. diff --git a/statemachine/event.py b/statemachine/event.py index 2c23f920..4ae6e2b3 100644 --- a/statemachine/event.py +++ b/statemachine/event.py @@ -46,6 +46,20 @@ def _expand_event_id(key: str) -> str: } +def raise_delayed_internal(event_id: str): + """Reject an event that asks to be both delayed and internal. + + The internal queue runs within the current macrostep, which takes no time. + """ + raise InvalidDefinition( + _( + "Event {id!r} cannot be both delayed and internal: the internal queue runs " + "within the current macrostep, which takes no time. Drop 'internal' to " + "deliver it later on the external queue, or drop 'delay' to run it now." + ).format(id=event_id) + ) + + class Event(AddCallbacksMixin, str): """An event triggers a signal that something has happened. @@ -100,6 +114,9 @@ def __new__( _has_real_id = id is not None id = str(id) if _has_real_id else f"__event__{uuid4().hex}" + if delay and internal: + raise_delayed_internal(id) + instance = super().__new__(cls, id) instance.id = id instance.delay = delay @@ -146,7 +163,9 @@ def __get__(self, instance, owner): """ if instance is None: return self - return BoundEvent(id=self.id, name=self.name, delay=self.delay, _sm=instance) + return BoundEvent( + id=self.id, name=self.name, delay=self.delay, internal=self.internal, _sm=instance + ) def put(self, *args, send_id: "str | None" = None, **kwargs): # The `__call__` is declared here to help IDEs knowing that an `Event` diff --git a/statemachine/io/actions.py b/statemachine/io/actions.py index cc5c17ac..11056e56 100644 --- a/statemachine/io/actions.py +++ b/statemachine/io/actions.py @@ -17,6 +17,7 @@ from ..event import BoundEvent from ..event import Event +from ..event import raise_delayed_internal from ..statemachine import StateChart from .evaluators import Evaluator from .evaluators import protected_attrs @@ -400,6 +401,9 @@ def create_send_action_callable( # noqa: C901 ) -> Callable: content: Any = () _valid_targets = (None, "#_internal", "internal", "#_parent", "parent") + internal = action.target in ("#_internal", "internal") + if internal and (action.delay or action.delayexpr): + raise_delayed_internal(action.event or action.eventexpr or "") if action.content: content = (evaluator.eval_literal(action.content),) delay_expr = ( @@ -433,8 +437,6 @@ def send_action(*args, **kwargs): # noqa: C901 _send_to_parent(action, evaluator, **kwargs) return - internal = target in ("#_internal", "internal") - send_id = None if action.id: send_id = action.id diff --git a/statemachine/io/schemas/statechart.schema.json b/statemachine/io/schemas/statechart.schema.json index b9000073..091d707b 100644 --- a/statemachine/io/schemas/statechart.schema.json +++ b/statemachine/io/schemas/statechart.schema.json @@ -117,7 +117,14 @@ "params": { "type": "array", "items": { "$ref": "#/$defs/param" } }, "content": { "type": "string" } }, - "additionalProperties": false + "additionalProperties": false, + "if": { + "required": ["target"], + "properties": { "target": { "enum": ["#_internal", "internal"] } } + }, + "then": { + "not": { "anyOf": [{ "required": ["delay"] }, { "required": ["delayexpr"] }] } + } }, "foreach": { "type": "object", diff --git a/statemachine/statemachine.py b/statemachine/statemachine.py index ba7398b4..fa805ed9 100644 --- a/statemachine/statemachine.py +++ b/statemachine/statemachine.py @@ -467,7 +467,7 @@ def send( *args, delay: float = 0, send_id: "str | None" = None, - internal: bool = False, + internal: "bool | None" = None, **kwargs, ) -> Any: """Send an :ref:`Event` to the state machine. @@ -477,6 +477,10 @@ def send( :param delay: A time delay in milliseconds to process the event. Default is 0. :param send_id: An identifier for the event, used with ``cancel_event()`` to cancel delayed events. + :param internal: Put the event on the internal queue, so it runs within the current + macrostep. Defaults to the event's declaration. Honored from a callback; sent from + outside a step, the event goes to the external queue and its result is returned. + Cannot be combined with ``delay``. :param kwargs: Additional keyword arguments to pass to the event. .. seealso:: @@ -488,6 +492,8 @@ def send( delay = ( delay if delay else know_event and know_event.delay or 0 ) # first the param, then the event, or 0 + if internal is None: + internal = bool(know_event and know_event.internal) event_instance = BoundEvent( id=event, name=event_name, delay=delay, internal=internal, _sm=self ) @@ -503,7 +509,9 @@ def raise_( Events on the internal queue are processed immediately within the current macrostep, before any pending external events. This is equivalent to calling - ``send(..., internal=True)``. + ``send(..., internal=True)``: called from outside a callback, there is no macrostep + to join, so the event goes to the external queue and its result is returned. A + ``delay`` other than 0 is an :class:`~statemachine.exceptions.InvalidDefinition`. .. seealso:: diff --git a/tests/io/test_validation.py b/tests/io/test_validation.py index a7c13f83..030a78a4 100644 --- a/tests/io/test_validation.py +++ b/tests/io/test_validation.py @@ -166,6 +166,19 @@ def test_invalid_callback_ref_in_nested_action_list(self): } ) + @pytest.mark.parametrize("target", ["#_internal", "internal"]) + @pytest.mark.parametrize("delay", ["delay", "delayexpr"]) + def test_invalid_delayed_send_to_internal(self, target, delay): + send = {"event": "e", "target": target, delay: "1s"} + with pytest.raises(InvalidDefinition, match="failed schema validation"): + validate_document({"states": {"a": {"initial": True, "enter": [{"send": send}]}}}) + + @pytest.mark.parametrize( + "send", [{"event": "e", "target": "#_internal"}, {"event": "e", "delay": "1s"}] + ) + def test_valid_send_to_internal_or_delayed(self, send): + validate_document({"states": {"a": {"initial": True, "enter": [{"send": send}]}}}) + def test_invalid_invoke_extra_key(self): with pytest.raises(InvalidDefinition, match="failed schema validation"): validate_document({"states": {"a": {"initial": True, "invoke": {"bogus": 1}}}}) diff --git a/tests/test_event_ordering.py b/tests/test_event_ordering.py index 168f994f..53b945e6 100644 --- a/tests/test_event_ordering.py +++ b/tests/test_event_ordering.py @@ -2,7 +2,8 @@ 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. +must still come out in the order they went in. And each external event gets its +own macrostep: what it enables runs before the next external event is taken. Theme: the Riders of Rohan answering the muster, each in turn. """ @@ -56,3 +57,53 @@ async def test_external_events_run_in_the_order_sent(self, sm_runner, riders): await sm_runner.processing_loop(sm) assert sm.answered == list(range(riders)) + + +class RelayedBeacons(StateChart): + dark = State(initial=True) + lit = State(final=True) + besieged = State(final=True) + + signal = dark.to.itself(internal=True, on="relay") + light = dark.to(lit) + siege = dark.to(besieged) + + def relay(self): + BoundEvent(id="light", name="Light", internal=True, _sm=self).put() + + +class WatchedBeacons(StateChart): + dark = State(initial=True) + lit = State(final=True) + besieged = State(final=True) + + signal = dark.to.itself(internal=True, on="relay") + dark.to(lit, cond="seen") + siege = dark.to(besieged) + + def __init__(self, *args, **kwargs): + self.seen = False + super().__init__(*args, **kwargs) + + def relay(self): + self.seen = True + + +@pytest.mark.timeout(10) +@pytest.mark.parametrize( + "machine", + [ + pytest.param(RelayedBeacons, id="internal-event"), + pytest.param(WatchedBeacons, id="eventless-transition"), + ], +) +async def test_what_an_external_event_enables_runs_before_the_next_external_event( + sm_runner, machine +): + sm = await sm_runner.start(machine) + BoundEvent(id="signal", name="Signal", _sm=sm).put() + BoundEvent(id="siege", name="Siege", _sm=sm).put() + + await sm_runner.processing_loop(sm) + + assert sm.configuration_values == {"lit"} diff --git a/tests/test_events.py b/tests/test_events.py index fc00d675..7fd97c12 100644 --- a/tests/test_events.py +++ b/tests/test_events.py @@ -59,17 +59,25 @@ class StartMachine(StateChart): assert [e.name for e in StartMachine.events] == ["Start the machine"] assert StartMachine.start.name == "Start the machine" - def test_accept_delay_and_internal(self): + @pytest.mark.parametrize( + ("declared", "expected"), + [ + pytest.param({"delay": 50}, (50, False), id="delay"), + pytest.param({"internal": True}, (0, True), id="internal"), + ], + ) + def test_accept_delay_or_internal(self, declared, expected): class BeaconsOfGondor(StateChart): dark = State(initial=True) lit = State(final=True) - light = Event(dark.to(lit), delay=50, internal=True) + light = Event(dark.to(lit), **declared) (registered,) = BeaconsOfGondor.events - assert (registered.delay, registered.internal) == (50, True) - assert BeaconsOfGondor.light.delay == 50 - assert BeaconsOfGondor().light.delay == 50 + bound = BeaconsOfGondor().light + assert (registered.delay, registered.internal) == expected + assert (BeaconsOfGondor.light.delay, BeaconsOfGondor.light.internal) == expected + assert (bound.delay, bound.internal) == expected def test_derive_name_from_id(self): class StartMachine(StateChart): diff --git a/tests/test_internal_triggers.py b/tests/test_internal_triggers.py new file mode 100644 index 00000000..f0132d6c --- /dev/null +++ b/tests/test_internal_triggers.py @@ -0,0 +1,381 @@ +"""Which queue a trigger asked to be internal goes to, and what its caller gets back. + +The internal queue holds what the machine raises while it runs a step, so it is processed +within that macrostep, before any pending external event. A trigger put from outside a step +has no macrostep to join: it goes to the external queue, where its caller gets its result, +its exception or ``TransitionNotAllowed``. A delay can never be internal, since a step takes +no time; asking for both is an invalid definition. + +Theme: the beacons of Gondor, lit from inside the chain or called for from afar. +""" + +import asyncio +import json +import threading +from inspect import isawaitable + +import pytest +from statemachine.engines.base import _running_engine +from statemachine.exceptions import InvalidDefinition +from statemachine.exceptions import TransitionNotAllowed +from statemachine.io.loader import load +from statemachine.io.scxml.processor import SCXMLProcessor + +from statemachine import Event +from statemachine import State +from statemachine import StateChart +from statemachine import StateMachine + +DELAYED_INTERNAL = "cannot be both delayed and internal" + + +class Beacons(StateChart): + dark = State(initial=True) + lit = State(final=True) + + light = Event(dark.to(lit, on="kindle"), internal=True) + + def kindle(self): + return "kindled" + + +async def settle(result): + return await result if isawaitable(result) else result + + +class TestFromACallback: + async def test_a_declared_internal_event_runs_before_a_pending_external_event(self, sm_runner): + class Muster(StateChart): + camp = State(initial=True) + riding = State(final=True) + besieged = State(final=True) + + summon = camp.to.itself(internal=True, on="call") + ride = Event(camp.to(riding), internal=True) + siege = camp.to(besieged) + + def call(self): + self.siege.put() + self.ride.put() + + sm = await sm_runner.start(Muster) + + await sm_runner.send(sm, "summon") + + assert sm.configuration_values == {"riding"} + + async def test_an_internal_event_raised_while_entering_the_initial_state_runs_first( + self, sm_runner + ): + class Watch(StateChart): + dark = State(initial=True) + lit = State(final=True) + fallen = State(final=True) + + light = Event(dark.to(lit), internal=True) + fall = dark.to(fallen) + + def on_enter_dark(self): + self.fall.put() + self.light.put() + + sm = await sm_runner.start(Watch) + + assert sm.configuration_values == {"lit"} + + def test_send_with_internal_false_overrides_the_declaration(self): + class Muster(StateChart): + camp = State(initial=True) + riding = State(final=True) + besieged = State(final=True) + + summon = camp.to.itself(internal=True, on="call") + ride = Event(camp.to(riding), internal=True) + siege = camp.to(besieged) + + def call(self): + self.send("siege") + self.send("ride", internal=False) + + sm = Muster() + + sm.send("summon") + + assert sm.configuration_values == {"besieged"} + + +class TestFromOutside: + @pytest.mark.parametrize( + "trigger", + [ + pytest.param(lambda sm: sm.light(), id="method"), + pytest.param(lambda sm: sm.send("light"), id="send"), + pytest.param(lambda sm: sm.raise_("light"), id="raise"), + ], + ) + async def test_a_declared_internal_event_returns_the_callback_result(self, sm_runner, trigger): + sm = await sm_runner.start(Beacons) + + assert await settle(trigger(sm)) == "kindled" + assert sm.configuration_values == {"lit"} + + async def test_an_internal_event_with_no_transition_raises_on_a_state_machine(self, sm_runner): + class StrictBeacons(StateMachine): + dark = State(initial=True) + lit = State() + + light = dark.to(lit) + douse = Event(lit.to(dark), internal=True) + + sm = await sm_runner.start(StrictBeacons) + + with pytest.raises(TransitionNotAllowed): + await sm_runner.send(sm, "douse") + + async def test_an_exception_from_a_declared_internal_event_reaches_the_caller(self, sm_runner): + class FailingBeacons(StateChart): + catch_errors_as_events = False + + dark = State(initial=True) + lit = State(final=True) + + light = Event(dark.to(lit, on="kindle"), internal=True) + + def kindle(self): + raise RuntimeError("the fire would not catch") + + sm = await sm_runner.start(FailingBeacons) + + with pytest.raises(RuntimeError, match="the fire would not catch"): + await sm_runner.send(sm, "light") + + async def test_a_fresh_async_machine_runs_a_declared_internal_event_on_its_first_await( + self, + ): + class AsyncBeacons(Beacons): + async def kindle(self): + return "kindled" + + sm = AsyncBeacons() + + assert await sm.light() == "kindled" + assert sm.configuration_values == {"lit"} + + @pytest.mark.parametrize("outcome", ["completes", "raises"]) + async def test_the_processing_loop_leaves_the_context_as_it_found_it(self, sm_runner, outcome): + class Signal(StateChart): + catch_errors_as_events = False + + dark = State(initial=True) + lit = State(final=True) + + light = Event(dark.to(lit, on="kindle"), internal=True) + fail = dark.to.itself(internal=True, on="stumble") + + def kindle(self): + return "kindled" + + def stumble(self): + raise RuntimeError("stumbled") + + sm = await sm_runner.start(Signal) + if outcome == "raises": + with pytest.raises(RuntimeError, match="stumbled"): + await sm_runner.send(sm, "fail") + + assert _running_engine.get() is None + assert await settle(sm.raise_("light")) == "kindled" + + +class TestWhoIsRunningTheStep: + """The internal queue belongs to the step of one machine, in one thread or task.""" + + @pytest.mark.timeout(10) + async def test_a_coroutine_outside_the_loop_gets_its_result_while_another_holds_it(self): + paused = asyncio.Event() + resume = asyncio.Event() + + class Relay(StateChart): + dark = State(initial=True) + waiting = State() + lit = State(final=True) + + hold = dark.to(waiting, on="pause") + light = Event(waiting.to(lit, on="kindle"), internal=True) + + async def pause(self): + paused.set() + await resume.wait() + return "held" + + def kindle(self): + return "kindled" + + async def light(): + return await sm.light() + + sm = Relay() + await sm.activate_initial_state() + holding = asyncio.create_task(sm.send("hold")) + await paused.wait() + lighting = asyncio.create_task(light()) + await asyncio.sleep(0) + resume.set() + + assert await asyncio.gather(holding, lighting) == ["held", "kindled"] + + @pytest.mark.timeout(10) + def test_a_thread_raising_while_another_runs_the_loop_queues_behind_its_sends(self): + in_step = threading.Event() + release = threading.Event() + + class Watch(StateChart): + dark = State(initial=True) + waiting = State() + lit = State(final=True) + besieged = State(final=True) + + hold = dark.to(waiting, on="pause") + light = waiting.to(lit) + siege = waiting.to(besieged) + + def pause(self): + in_step.set() + release.wait(timeout=5) + + sm = Watch() + holder = threading.Thread(target=sm.send, args=("hold",)) + holder.start() + assert in_step.wait(timeout=5) + + sm.send("siege") + sm.raise_("light") + release.set() + holder.join(timeout=5) + + assert sm.configuration_values == {"besieged"} + + async def test_a_callback_raising_on_another_machine_gets_its_result(self, sm_runner): + other = await sm_runner.start(Beacons) + + class Lighter(StateChart): + idle = State(initial=True) + done = State(final=True) + + reach = idle.to(done, on="light_other") + + def light_other(self): + return other.raise_("light") + + sm = await sm_runner.start(Lighter) + + assert await settle(await sm_runner.send(sm, "reach")) == "kindled" + assert other.configuration_values == {"lit"} + + async def test_a_callback_triggering_a_disallowed_event_on_another_machine_raises( + self, sm_runner + ): + class StrictBeacons(StateMachine): + dark = State(initial=True) + lit = State() + + light = dark.to(lit) + douse = Event(lit.to(dark), internal=True) + + other = await sm_runner.start(StrictBeacons) + + class Douser(StateChart): + catch_errors_as_events = False + + idle = State(initial=True) + done = State(final=True) + + reach = idle.to(done, on="douse_other") + + def douse_other(self): + return other.douse() + + sm = await sm_runner.start(Douser) + + with pytest.raises(TransitionNotAllowed): + await settle(await sm_runner.send(sm, "reach")) + + +class TestDelayedInternal: + def test_a_declaration_cannot_be_both_delayed_and_internal(self): + with pytest.raises(InvalidDefinition, match=DELAYED_INTERNAL): + + class DelayedBeacons(StateChart): + dark = State(initial=True) + lit = State(final=True) + + light = Event(dark.to(lit), delay=50, internal=True) + + def test_a_compound_declaration_cannot_be_both_delayed_and_internal(self): + with pytest.raises(InvalidDefinition, match=DELAYED_INTERNAL): + + class DelayedBeacons(StateChart): + class chain(State.Compound): + dark = State(initial=True) + lit = State(final=True) + + light = Event(dark.to(lit), delay=50, internal=True) + + @pytest.mark.parametrize( + "trigger", + [ + pytest.param(lambda sm: sm.raise_("light", delay=50), id="raise"), + pytest.param(lambda sm: sm.send("light", delay=50, internal=True), id="send"), + pytest.param(lambda sm: sm.send("light", delay=50), id="declared-internal"), + ], + ) + async def test_a_trigger_cannot_be_both_delayed_and_internal(self, sm_runner, trigger): + sm = await sm_runner.start(Beacons) + + with pytest.raises(InvalidDefinition, match=DELAYED_INTERNAL): + await settle(trigger(sm)) + + assert sm.configuration_values == {"dark"} + + @pytest.mark.timeout(10) + async def test_a_declared_internal_event_can_be_sent_later_as_external(self, sm_runner): + sm = await sm_runner.start(Beacons) + + assert await sm_runner.send(sm, "light", delay=10, internal=False) == "kindled" + + @pytest.mark.parametrize("delay", ['delay="50ms"', "delayexpr=\"'50ms'\""]) + @pytest.mark.parametrize("target", ["#_internal", "internal"]) + def test_an_scxml_send_to_internal_with_a_delay_fails_to_load(self, target, delay): + scxml = f""" + + + + + + + + + + """ + + with pytest.raises(InvalidDefinition, match=DELAYED_INTERNAL): + SCXMLProcessor().parse_scxml("delayed_internal", scxml) + + @pytest.mark.parametrize("validate", [True, False]) + def test_a_native_send_to_internal_with_a_delay_fails_to_load(self, validate): + document = { + "name": "DelayedInternal", + "states": { + "dark": { + "initial": True, + "enter": [ + {"send": {"event": "light", "target": "#_internal", "delay": "50ms"}} + ], + "transitions": [{"event": "light", "target": "lit"}], + }, + "lit": {"final": True}, + }, + } + + with pytest.raises(InvalidDefinition): + load(json.dumps(document), format="json", validate=validate) diff --git a/tests/test_statechart_compound.py b/tests/test_statechart_compound.py index de1c52f3..61dafc01 100644 --- a/tests/test_statechart_compound.py +++ b/tests/test_statechart_compound.py @@ -337,19 +337,24 @@ class shire(State.Compound): await sm_runner.send(sm, "visit_pub") assert {"shire", "green_dragon"} == set(sm.configuration_values) - def test_name_delay_and_internal_are_preserved(self): + @pytest.mark.parametrize( + ("declared", "expected"), + [ + pytest.param({"delay": 50}, (50, False), id="delay"), + pytest.param({"internal": True}, (0, True), id="internal"), + ], + ) + def test_name_delay_and_internal_are_preserved(self, declared, expected): class NamedEvent(StateChart): class shire(State.Compound): bag_end = State(initial=True) green_dragon = State(final=True) - visit_pub = Event( - bag_end.to(green_dragon), name="Visit the pub", delay=50, internal=True - ) + visit_pub = Event(bag_end.to(green_dragon), name="Visit the pub", **declared) (registered,) = NamedEvent.events assert (registered.id, registered.name) == ("visit_pub", "Visit the pub") - assert (registered.delay, registered.internal) == (50, True) + assert (registered.delay, registered.internal) == expected assert NamedEvent.visit_pub.name == "Visit the pub" def test_expanded_id_drops_the_declared_arguments(self):