Fix regression that did not trigger events in nested calls within an already running transition (#451)

* chore: Processing block protected by a threading.Lock

* fix: Regression that did not trigger events in nested calls within an already running transition. Closes #449

Author: fgmacedo

docs/processing_model.md

@@ -91,6 +91,10 @@ Note that the events `connect` and `connection_succeed` are executed sequentiall
 
 ## Non-RTC model
 
+```{deprecated} 2.3.2
+`StateMachine.rtc` option is deprecated. We'll keep only the **run-to-completion** (RTC) model.
+```
+
 In contrast, in a non-RTC (synchronous) processing model, the state machine starts executing nested events
 while processing a parent event. This means that when an event is triggered, the state machine
 chains the processing when another event was triggered as a result of the first event.

docs/releases/1.0.1.md

@@ -217,7 +217,7 @@ single `Transition`.
 
 ## Deprecated features in 1.0
 
-### Statemachine class
+### Statemachine class deprecations
 
 - `StateMachine.run` is deprecated in favor of `StateMachine.send`.
 - `StateMachine.allowed_transitions` is deprecated in favor of `StateMachine.allowed_events`.

docs/releases/2.3.2.md

@@ -43,10 +43,15 @@ See {ref}`listeners` for more details.
 
 ## Bugfixes in 2.3.2
 
--
+- Fixes [#449](https://github.com/fgmacedo/python-statemachine/issues/449): Regression that did not trigger events
+  in nested calls within an already running transition.
+
 
 ## Deprecation notes
 
-### Statemachine class
+### Statemachine class deprecations in 2.3.2
+
+Deprecations that will be removed on the next major release:
 
 - `StateMachine.add_observer` is deprecated in favor of `StateMachine.add_listener`.
+- `StateMachine.rtc` option is deprecated. We'll keep only the **run-to-completion** (RTC) model.

statemachine/event.py

@@ -15,15 +15,15 @@ def __init__(self, name: str):
     def __repr__(self):
         return f"{type(self).__name__}({self.name!r})"
 
-    async def trigger(self, machine: "StateMachine", *args, **kwargs):
+    def trigger(self, machine: "StateMachine", *args, **kwargs):
         trigger_data = TriggerData(
             machine=machine,
             event=self.name,
             args=args,
             kwargs=kwargs,
         )
-
-        return await machine._process(trigger_data)
+        machine._put_nonblocking(trigger_data)
+        return machine._processing_loop()
 
 
 def trigger_event_factory(event_instance: Event):

statemachine/event_data.py

@@ -47,6 +47,7 @@ class EventData:
     """The destination :ref:`State` of the :ref:`transition`."""
 
     result: "Any | None" = None
+
     executed: bool = False
 
     def __post_init__(self):

statemachine/statemachine.py

@@ -2,6 +2,7 @@
 from collections import deque
 from copy import deepcopy
 from functools import partial
+from threading import Lock
 from typing import TYPE_CHECKING
 from typing import Any
 from typing import Dict
@@ -81,7 +82,7 @@ def __init__(
         self.allow_event_without_transition = allow_event_without_transition
         self.__initialized = False
         self.__rtc = rtc
-        self.__processing: bool = False
+        self.__processing = Lock()
         self._external_queue: deque = deque()
         self._callbacks_registry = CallbacksRegistry()
         self._states_for_instance: Dict[State, State] = {}
@@ -117,12 +118,17 @@ def __repr__(self):
 
     def __deepcopy__(self, memo):
         deepcopy_method = self.__deepcopy__
-        self.__deepcopy__ = None
-        try:
-            cp = deepcopy(self, memo)
-        finally:
-            self.__deepcopy__ = deepcopy_method
-            cp.__deepcopy__ = deepcopy_method
+        lock = self.__processing
+        with lock:
+            self.__deepcopy__ = None
+            self.__processing = None
+            try:
+                cp = deepcopy(self, memo)
+                cp.__processing = Lock()
+            finally:
+                self.__deepcopy__ = deepcopy_method
+                cp.__deepcopy__ = deepcopy_method
+                self.__processing = lock
         cp._callbacks_registry.clear()
         cp._register_callbacks([])
         cp.add_listener(*cp._listeners.keys())
@@ -312,7 +318,11 @@ async def _trigger(self, trigger_data: TriggerData):
 
         return event_data.result if event_data else None
 
-    async def _process(self, trigger_data: TriggerData):
+    def _put_nonblocking(self, trigger_data: TriggerData):
+        """Put the trigger on the queue without blocking the caller."""
+        self._external_queue.append(trigger_data)
+
+    async def _processing_loop(self):
         """Process event triggers.
 
         The simplest implementation is the non-RTC (synchronous),
@@ -331,31 +341,24 @@ async def _process(self, trigger_data: TriggerData):
             will be processed sequentially (and not nested).
 
         """
+
         if not self.__rtc:
             # The machine is in "synchronous" mode
+            trigger_data = self._external_queue.popleft()
             return await self._trigger(trigger_data)
 
-        # The machine is in "queued" mode
-        # Add the trigger to queue and start processing in a loop.
-        self._external_queue.append(trigger_data)
-
         # We make sure that only the first event enters the processing critical section,
         # next events will only be put on the queue and processed by the same loop.
-        if self.__processing:
-            return
-
-        return await self._processing_loop()
-
-    async def _processing_loop(self):
-        """Execute the triggers in the queue in order until the queue is empty"""
-        self.__processing = True
+        if not self.__processing.acquire(blocking=False):
+            return None
 
         # We will collect the first result as the processing result to keep backwards compatibility
         # so we need to use a sentinel object instead of `None` because the first result may
         # be also `None`, and on this case the `first_result` may be overridden by another result.
         sentinel = object()
         first_result = sentinel
         try:
+            # Execute the triggers in the queue in FIFO order until the queue is empty
             while self._external_queue:
                 trigger_data = self._external_queue.popleft()
                 try:
@@ -368,7 +371,7 @@ async def _processing_loop(self):
                     self._external_queue.clear()
                     raise
         finally:
-            self.__processing = False
+            self.__processing.release()
         return first_result if first_result is not sentinel else None
 
     async def _activate(self, event_data: EventData):
@@ -412,7 +415,7 @@ def send(self, event: str, *args, **kwargs):
         coro = self.async_send(event, *args, **kwargs)
         return run_async_from_sync(coro)
 
-    async def async_send(self, event: str, *args, **kwargs):
+    def async_send(self, event: str, *args, **kwargs):
         """Send an :ref:`Event` to the state machine.
 
         .. seealso::
@@ -421,7 +424,7 @@ async def async_send(self, event: str, *args, **kwargs):
 
         """
         event_instance: Event = Event(event)
-        return await event_instance.trigger(self, *args, **kwargs)
+        return event_instance.trigger(self, *args, **kwargs)
 
     def _get_callbacks(self, key) -> CallbacksExecutor:
         return self._callbacks_registry[key]

tests/testcases/issue449.md

@@ -37,8 +37,6 @@ Exercise:
 
 
 ```py
->>> import pytest
->>> pytest.xfail("This test is a regression on 2.3.0+ due to asyncio support.")
 >>> example = ExampleStateMachine()
 Entering state initial. Event: __initial__