I have a fairly complex application involving a GUI front-end and several other classes, several of which are state machines based on the excellent pytransitions library. The application hangs at various times, and being multi-threaded it is difficult to debug, but I have reproduced a minimal example here:
from transitions import Machine
import time
States = ["OFF", "ON"]
Transitions = [{"trigger": "PowerUp", "source": "OFF", "dest": "ON"}]
class GUI:
def __init__(self):
self.machine1 = Machine_1(self)
self.machine2 = Machine_2(self)
def on_button_pressed(self):
self.machine2.trigger("PowerUp")
self.machine1.trigger("PowerUp")
class Machine_1:
def __init__(self, parent):
self.parent = parent
self.fsm = Machine(model=self, states=States, transitions=Transitions, initial="OFF")
def on_enter_ON(self):
print("machine1 is on!")
self.parent.machine2.signal = True
class Machine_2:
def __init__(self, parent):
self.parent = parent
self.fsm = Machine(model=self, states=States, transitions=Transitions, initial="OFF")
self.signal = False
def on_enter_ON(self):
print("machine2 waits for machine1 ...")
while self.signal is False:
time.sleep(0.1)
print("machine2 on!")
i.e. When machine1 is "turned on", it sends a signal to machine2. machine2 must wait for this signal before executing its logic for the "on" state.
Using the Python REPL as the main loop:
>>> import test
>>> gui = test.GUI()
>>> gui.machine1.state
'off'
>>> gui.machine2.state
'off'
>>> gui.on_button_pressed()
machine2 waits for machine1 ...
(never returns)
machine2 is stuck in its while loop, waiting for the signal from machine1, which never arrives because machine1 is never triggered.
If I re-order the sequence of triggers in gui.on_button_pressed():
self.machine1.trigger("PowerUp")
self.machine2.trigger("PowerUp")
... everything works:
>>> gui.on_button_pressed()
machine1 is on!
machine2 waits for machine1 ...
machine2 is on!
What is going on here? Is the trigger() method supposed to block? Is it documented somewhere, when does it return? Or am I using the on_enter_state callbacks in an unsupported way? Is there a recommended pattern I should be using?
The core Machine of pytransitions is neither threaded nor asynchronous. This means that when a callback does not return, the whole event processing and thus the trigger function will block. There are a variety of ways to deal with this. Which way to chose depends on your architecture. Providing MWE for all approaches is a bit too much for a SO answer but I can briefly outline how solutions could look like.
Since Machine_1 already has a reference to Machine_2 and Machine_2 is waiting for an event from Machine_1, you could model that transition explicitly:
Transitions = [{"trigger": "PowerUp", "source": "OFF", "dest": "ON"}]
# ...
self.machine2 = Machine_2(self)
self.machine2.add_transition("Machine1PoweredUp", "OFF", "ON")
# ...
def on_enter_ON(self):
print("machine1 is on!")
self.parent.machine2.Machine1PoweredUp()
Note that when you have machines that share the same states and transitions, you can use ONE machine and multiple models.
def Model_1():
def on_enter_ON(self):
# a dispatched event is forwarded to all models attached to the
# machine in question. We consider `parent` to be such a machine here.
self.parent.dispatch("Machine1PoweredUp")
s1 = Model_1()
s2 = Model_2()
# we set `ignore_invalid_triggers` since `s1` need to process the fired
# event as well even though it is already in state `ON`
m = Machine(model=[s1, s2], ignore_invalid_triggers=True)
s1.parent = m
s1.PowerUp()
You can use the shared machine as some sort of 'event bus' which can reduce coupling and dependencies.
Instead of blocking in your main loop, you could make Machine_2 'attempt' a transition with a condition and return. Conditions can either be the name of a model method or a reference to a callable. Since pytransitions always adds convenience state check functions to models, we can just pass the state check is_ON from machine1 to the transition definition. The condition will halt and return when the condition is not met, enabling a sequential workflow.
self.machine2.add_transition("PowerUp", "OFF", "ON", conditions=machine1.is_ON)
# ...
while not machine2.is_ON:
time.sleep(0.1)
machine2.PowerUp()
You should note though, that when on_enter_<state> callbacks are executed, the machine/model is already considered to be in that state. For your particular use case that would mean that machine1.is_ON returns True even though the on_enter_ON callback of Machine_1 is still processed.
You could create a thread for your machine 1 state check in machine 2 as well. To 'unblock' the transition, I'd suggest adding a temporary state such as Booting to show that Machine_2 is not ready yet.
class Machine_2:
def check_state(self):
while self.signal is False:
time.sleep(0.1)
self.Machine1Ready() # transition from 'Booting' to 'ON'
def on_enter_ON(self):
print("machine2 on!")
def on_enter_BOOTING(self):
print("machine2 waits for machine1 ...")
threading.Thread(target=self.check_state).start()
transitions features a LockedMachine in case multiple threads need to access the same machine. However, threaded access to state machines has some pitfalls and is considerably harder to debug in my experience.
Here is a short example of how AsyncMachine could be used to achieve something like this. This example is for illustrative purposes and should not be reproduced that way:
from transitions.extensions.asyncio import AsyncMachine
import asyncio
class Model1:
done = False
async def on_enter_ON(self):
await asyncio.sleep(0.1)
print("Machine 1 done")
self.done = True
model1 = Model1()
class Model2:
done = False
async def on_enter_ON(self):
while not model1.done:
print("Machine 1 not ready yet")
await asyncio.sleep(0.05)
print("Machine 2 done")
model2 = Model2()
m = AsyncMachine(model=[model1, model2], states=["OFF", "ON"], initial="OFF")
asyncio.get_event_loop().run_until_complete(asyncio.gather(model1.to_ON(), model2.to_ON()))