Add 2-state (boolean) asyncio synchronization primitive

Ideas
1

UPD: A more general multi-valued State asyncio primitive with State(bool, False) as a special case:

class State[T]:
    def __init__(
        self, values: Sequence[T] | type[Enum] | type[bool],
        default: T,
    ): ...
    async def wait_value(self, value: T) -> T: ...
    async def wait_changed(self) -> T: ...
    def set(self, value: T) -> None: ...
    def value(self) -> T: ...

Imagine that we have boolean variable, like “ready”, “healthy”, etc., and we want to listen for its state change or reaching specific on/off state. There is already asyncio primitive Event that allows to await for the state change, but it works in one direction only.

What do you think about asyncio having a Toggle primitive with API similar to the one below?

I met this use case when implementing liveness status monitor for python app in k8s, but possible applications look broad enough to let me ask this question.

class Toggle:
    def __init__(self, start_off: bool = True):
        self._on, self._off = asyncio.Event(), asyncio.Event()
        (self._off if start_off else self._on).set()

    def is_on(self) -> bool: return self._on.is_set()
    def is_off(self) -> bool: return self._off.is_set()

    async def wait_on(self) -> Literal[True]:
        return await self._on.wait()

    async def wait_off(self) -> Literal[True]:
        return await self._off.wait()

    async def wait_toggle(self) -> Literal[True]:
        return await (self._off if self.is_on() else self._on).wait()

    def turn_on(self) -> None:
        self._on.set(); self._off.clear()

    def turn_off(self) -> None:
        self._on.clear(); self._off.set()

    def toggle(self) -> None:
        if self.is_on():
            self.turn_off()
        else:
            self.turn_on()

Here one can see direct reuse of Event, but Toggle can be implemented independently using logic similar to Event.

2

Or maybe call it Boolean:

class Boolean:
    def __init__(self, initial: bool = False): ...
    async def wait_true(self) -> Literal[True]: ...
    async def wait_false(self) -> Literal[True]: ...
    async def wait_toggle(self) -> Literal[True]: ...
    def set(self, value: bool) -> None: ...
    def toggle(self) -> None: ...
    def is_true(self) -> bool: ...
    def is_false(self) -> bool: ...
    def __bool__(self) -> bool: ...
3

Or even generalize to State? Btw, this will make coding async state machines super easy.

class State[T]:
    def __init__(
        self, values: Sequence[T] | type[Enum] | type[bool],
        default: T,
    ): ...
    async def wait_value(self, value: T) -> T: ...
    async def wait_changed(self) -> T: ...
    def set(self, value: T) -> None: ...
    def value(self) -> T: ...
4

Regarding:

    async def wait_off(self) -> Literal[True]:
        return self._off.wait()

I think there is an await missing: return await self._off.wait(). If not, could you please show an example how is the API supposed to be used ?

1 Like
5

Await is not missing here, it just returns the result of asyncio.Event.wait() which is an Awaitable[Literal[True]]. I don’t have a specific use case how to use the-always-True value, just mimicking asyncio.Event for API consistence. The method could be re-written as

def wait_off(self) -> Awaitable[Literal[True]]:
        return self._off.wait()

but async def is more expressive as part of interface.

6

Did you try it? The two versions are not equivalent. The sync function is clear, but the async function will not work as expected, it would need to be called this way:

await (await mytoggle.wait_off())

because the inner await returns a coroutine.

1 Like
7

Yes, my bad, you’re absolutely right, added missing awaits.

8

See GitHub - groove-x/trio-util: Utility library for the Python Trio async/await framework for prior art on async values

1 Like
9

Thank you! Didn’t know about it. Really like the AsyncValue & AsyncBool interface.