Advice to implementing state machine

Help and advice
1

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

2

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)
3

… 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())
4

You solution is exactly what I was going to suggest when I read your first post. Your “runner()” function, which calls a function f then puts the result into f to run on the next loop iteration, is often called a “trampoline” and it’s the standard trick to work around the lack of a tail call feature in a language. There’s a Wikipedia article for trampoline but it’s weirdly terrible

Your first version of the solution is a bit more Trionic e.g. notice how nursery.start_soon takes a coroutine function (like state1) rather than a coroutine (like state1()). You could still allow arguments to be passed by returning a tuple of (fn, args) (again that matches nursery.start_soon()), although that would be hard to make type hints for. Alternatively, you could use your same strategy but just use functools.partial whenever you want to pass a parameter (e.g., return partial(state2, "test") or return partial(state2, arg="test").