perf: remove weakref indirection on engine.sm and extract _first_transition_that_matches

- Replace weakref + property with direct reference: eliminates 145k
  weakref deref + assert calls per benchmark cycle. The engine lifetime
  is tied to the SM — no leak risk with CPython's cyclic GC (PEP 442).
- Extract `first_transition_that_matches` closure into a proper method
  `_first_transition_that_matches` on both BaseEngine and AsyncEngine,
  avoiding re-creation of the function object on every _select_transitions
  call.

Author: fgmacedo

statemachine/engines/async_.py

@@ -16,7 +16,6 @@
 from .base import BaseEngine
 
 if TYPE_CHECKING:
-    from ..event import Event
     from ..transition import Transition
 
 # ContextVar to distinguish reentrant calls (from within callbacks) from
@@ -105,29 +104,32 @@ 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]
+        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 await self._conditions_match(transition, trigger_data)
+                ):
+                    return transition
+        return None
+
     async def _select_transitions(  # type: ignore[override]
         self, trigger_data: TriggerData, predicate: Callable
     ) -> "OrderedSet[Transition]":
         enabled_transitions: "OrderedSet[Transition]" = OrderedSet()
 
         atomic_states = (state for state in self.sm.configuration if state.is_atomic)
 
-        async def first_transition_that_matches(
-            state: State, event: "Event | None"
-        ) -> "Transition | None":
-            for s in chain([state], state.ancestors()):
-                transition: "Transition"
-                for transition in s.transitions:
-                    if (
-                        not transition.initial
-                        and predicate(transition, event)
-                        and await self._conditions_match(transition, trigger_data)
-                    ):
-                        return transition
-            return None
-
         for state in atomic_states:
-            transition = await first_transition_that_matches(state, trigger_data.event)
+            transition = await self._first_transition_that_matches(state, trigger_data, predicate)
             if transition is not None:
                 enabled_transitions.add(transition)
 

statemachine/engines/base.py

@@ -11,11 +11,8 @@
 from typing import Dict
 from typing import List
 from typing import cast
-from weakref import ReferenceType
-from weakref import ref
 
 from ..event import BoundEvent
-from ..event import Event
 from ..event_data import EventData
 from ..event_data import TriggerData
 from ..exceptions import InvalidDefinition
@@ -88,7 +85,7 @@ def remove(self, send_id: str):
 
 class BaseEngine:
     def __init__(self, sm: "StateChart"):
-        self._sm: ReferenceType["StateChart"] = ref(sm)
+        self.sm: "StateChart" = sm
         self.external_queue = EventQueue()
         self.internal_queue = EventQueue()
         self._sentinel = object()
@@ -105,12 +102,6 @@ def __init__(self, sm: "StateChart"):
     def empty(self):  # pragma: no cover
         return self.external_queue.is_empty()
 
-    @property
-    def sm(self) -> "StateChart":
-        sm = self._sm()
-        assert sm, "StateMachine has been destroyed"
-        return sm
-
     def clear_cache(self):
         """Clears the cache. Should be called at the start of each processing loop."""
         self._cache.clear()
@@ -347,6 +338,23 @@ def select_transitions(self, trigger_data: TriggerData) -> OrderedSet[Transition
         """
         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]:
@@ -356,23 +364,8 @@ def _select_transitions(
         # Get atomic states, TODO: sorted by document order
         atomic_states = (state for state in self.sm.configuration if state.is_atomic)
 
-        def first_transition_that_matches(
-            state: State, event: "Event | None"
-        ) -> "Transition | None":
-            for s in chain([state], state.ancestors()):
-                transition: Transition
-                for transition in s.transitions:
-                    if (
-                        not transition.initial
-                        and predicate(transition, event)
-                        and self._conditions_match(transition, trigger_data)
-                    ):
-                        return transition
-
-            return None
-
         for state in atomic_states:
-            transition = first_transition_that_matches(state, trigger_data.event)
+            transition = self._first_transition_that_matches(state, trigger_data, predicate)
             if transition is not None:
                 enabled_transitions.add(transition)