Dear trio comunity,
what is the suggested way to implement a simple state machine with trio
?
The intention is to avoid global state variable and additional state transition loop, something like this:
import trio
async def state1_broken():
print('this is state1')
await trio.sleep(0.01)
await state2_broken()
async def state2_broken():
print('this is state2')
await trio.sleep(0.01)
await state1_broken()
trio.run(state1_broken)
This naive implementation obviously does not work, since python
does not optimize tail recursive calls. It quickly raises RecursionError: maximum recursion depth exceeded
exception.
There are some state machine python libraries, but I am wandering if there is an effective way to solve it within trio
.
regards,
Zoran
Answering my own question… Not sure if this is the correct way to go, but introducing the runner
solves the problem:
from typing import *
import trio
State: TypeAlias = Callable[[], Awaitable['State']]
async def state1() -> State:
print('this is state1')
await trio.sleep(0.01)
return state2
async def state2() -> State:
print('this is state2')
await trio.sleep(0.01)
return state1
async def runner(initial: State) -> Never:
f = initial
while True:
f = await f()
trio.run(runner, state1)
… or even more simple which also allows passing arguments between states
from typing import *
import trio
State: TypeAlias = Awaitable['State']
async def state1() -> State:
print('this is state1')
await trio.sleep(0)
return state2('test')
async def state2(arg: str) -> State:
print('this is state2')
await trio.sleep(0)
return state1()
async def runner(initial: State) -> Never:
f = initial
while True:
f = await f
trio.run(runner, state1())