uasyncio.ThreadSafeFlag: is this use valid?

[peterhinch]

The ThreadSafeFlag has a limitation compared to Event in that it is self-clearing, so only one task may wait on it. This would appear to have a simple solution, which is to create a task which waits on the TSF and triggers an Event. I have used this, and it cropped up again here. The technique seems to work, and I can't see a flaw.

I have a nagging doubt that if it was this easy, @dpgeorge would have written it this way. What is the view of this:

import uasyncio as asyncio
from pyb import Timer

class ThreadSafeEvent(asyncio.Event):
    def __init__(self):
        super().__init__()
        self._tsf = asyncio.ThreadSafeFlag()
        asyncio.create_task(self._run())

    async def _run(self):
        while True:
            await self._tsf.wait()
            super().set()

    def set(self):
        self._tsf.set()

# Test/demo
async def waiter(n, evt):
    await evt.wait()
    print(f"Waiter {n} got event")

async def main():
    evt = ThreadSafeEvent()
    tim = pyb.Timer(4, freq=1, callback=lambda t : evt.set())
    while True:
        tasks = [asyncio.create_task(waiter(n + 1, evt)) for n in range(4)]
        await asyncio.gather(*tasks)
        evt.clear()
        print("Cleared event")

asyncio.run(main())

[jimmo]

This looks like a neat solution! I don't see a problem with it. Conceptually it's very similar to how TSF is used in aioble, for example to queue writes on characteristics.

I have a nagging doubt that if it was this easy, @dpgeorge would have written it this way.

What should/could have been written this way?

I see this TSE as being a very useful complement to TSF and E, but I think TSF is still useful on its own. For one thing it is very light weight and doesn't have the overhead of a task.

Perhaps we should consider adding this TSE to asyncio too? I still wish we'd put TSF in a different module though...

Btw, when we support awaiting hardware events (e.g. pin edges) should we allow multiple waiters on one pin? (I don't think this would work with the current proposal)

[peterhinch]

Thanks for confirming that this looks valid.

I'm not sure what you mean by

Btw, when we support awaiting hardware events (e.g. pin edges) should we allow multiple waiters on one pin? (I don't think this would work with the current proposal)

Where is the difference between a hard timer ISR and one for a pin? I tried the following and it seems to work OK.

import uasyncio as asyncio
from machine import Pin

class ThreadSafeEvent(asyncio.Event):
    def __init__(self):
        super().__init__()
        self._tsf = asyncio.ThreadSafeFlag()
        asyncio.create_task(self._run())

    async def _run(self):
        while True:
            await self._tsf.wait()
            super().set()

    def set(self):
        self._tsf.set()

# Demo with hard pin IRQ - link Y1 and Y2:
async def waiter(n, evt):
    await evt.wait()
    print(f"Waiter {n} got event")

async def trigger(pin):
    while True:
        await asyncio.sleep(1)
        pin(not pin())

async def main():
    evt = ThreadSafeEvent()
    out = Pin("Y1", Pin.OUT)
    inp = Pin("Y2", Pin.IN)
    irq = inp.irq(handler=lambda p : evt.set(), trigger=Pin.IRQ_FALLING, hard=True)
    asyncio.create_task(trigger(out))
    while True:
        tasks = [asyncio.create_task(waiter(n + 1, evt)) for n in range(4)]
        await asyncio.gather(*tasks)
        evt.clear()
        print("Cleared event")

asyncio.run(main())

[dpgeorge]

Yes we should eventually move TSF to another module to get CPython compatibility, eg asyncio_micropython.ThreadSafeFlag.

[jimmo]

Where is the difference between a hard timer ISR and one for a pin?

@peterhinch Oh I meant when we make this more integrated. i.e. my hope is that we might eventually support being able to just write something along the lines of:

p = machine.Pin("X4", machine.Pin.IN)
await p.edge(machine.Pin.RISING)

What happens if two things await the same pin at the same time?

[dpgeorge]

The reason TSF is the way it is, is so that it does not allocate any memory and .set() can execute from within a hard IRQ. As such it cannot queue waiting tasks when .set() is called, like Event can.

That said, it should be possible to implement a TSE without the overhead of an additional task. Something like this:

import uasyncio as asyncio
from pyb import Timer


class ThreadSafeEvent(asyncio.Event):
    def __init__(self):
        super().__init__()
        self._waiting_on_tsf = False
        self._tsf = asyncio.ThreadSafeFlag()

    def set(self):
        self._tsf.set()

    async def wait(self):
        if self._waiting_on_tsf == False:
            self._waiting_on_tsf = True
            await self._tsf.wait()
            super().set()
            self._waiting_on_tsf = False
        else:
            await super().wait()


# Test/demo
async def waiter(n, evt):
    await evt.wait()
    print(f"Waiter {n} got event")


async def main():
    evt = ThreadSafeEvent()
    tim = pyb.Timer(4, freq=1, callback=lambda t: evt.set())
    while True:
        tasks = [asyncio.create_task(waiter(n + 1, evt)) for n in range(4)]
        await asyncio.gather(*tasks)
        evt.clear()
        print("Cleared event")


asyncio.run(main())

In the above, the first task to wait on the TSE will wait on the TSF. Subsequent tasks will wait on the Event. Then when the first one wakes up, it wakes the rest.

I think there might be a problem though if the first task is cancelled while waiting, then the remaining tasks stay waiting forever. That may or may not be easy to fix with a try-finally.

[peterhinch]

I'll see if I can figure out solution to that.

[peterhinch]

Taking the above code, in the case where the first task is cancelled, the only solution I've found is to launch a minimal task waiting on the ThreadSafeFlag as below. The overhead (~200 bytes on STM) is only incurred in this particular case.

class ThreadSafeEvent(asyncio.Event):
    def __init__(self):
        super().__init__()
        self._waiting_on_tsf = False
        self._tsf = asyncio.ThreadSafeFlag()

    def set(self):
        self._tsf.set()

    async def _waiter(self):  # Runs on cancellation of first task
        await self._tsf.wait()
        super().set()
        self._waiting_on_tsf = False

    async def wait(self):
        if self._waiting_on_tsf == False:
            self._waiting_on_tsf = True
            try:
                await self._tsf.wait()
                super().set()
                self._waiting_on_tsf = False
            except asyncio.CancelledError:
                asyncio.create_task(self._waiter())
                raise  # Propagate the cancellation immediately
        else:
            await super().wait()

I'd be interested to see if anyone can come up with something better.

Full testable code follows:

import uasyncio as asyncio
from pyb import Timer

class ThreadSafeEvent(asyncio.Event):
    def __init__(self):
        super().__init__()
        self._waiting_on_tsf = False
        self._tsf = asyncio.ThreadSafeFlag()

    def set(self):
        self._tsf.set()

    async def _waiter(self):
        await self._tsf.wait()
        super().set()
        self._waiting_on_tsf = False

    async def wait(self):
        if self._waiting_on_tsf == False:
            self._waiting_on_tsf = True
            try:
                await self._tsf.wait()
                super().set()
                self._waiting_on_tsf = False
            except asyncio.CancelledError:
                asyncio.create_task(self._waiter())
                raise
        else:
            await super().wait()

# Test/demo
async def waiter(n, evt):
    try:
        await evt.wait()
        print(f"Waiter {n} got event")
    except asyncio.CancelledError:
        print(f"Waiter {n} cancelled")

async def can(task):
    await asyncio.sleep_ms(100)
    task.cancel()

async def main():
    evt = ThreadSafeEvent()
    tim = pyb.Timer(4, freq=1, callback=lambda t: evt.set())
    nt = 0
    while True:
        tasks = [asyncio.create_task(waiter(n + 1, evt)) for n in range(4)]
        asyncio.create_task(can(tasks[nt]))
        await asyncio.gather(*tasks, return_exceptions=True)
        evt.clear()
        print("Cleared event")
        nt = (nt + 1) % 4


asyncio.run(main())

[rkompass]

With

def set_event(evt):
    while True:
        sleep_ms(1000)
        evt.set()

and

    _thread.start_new_thread(set_event, (evt,))
#    tim = Timer(freq=1, callback=lambda t: evt.set())

in main() this runs with the same output on my RP Pico. So it seems to work too if the event is triggered from another task, in the setup without GIL.

The only alternative to ThreadSafeEvent._waiter() I can imagine would be interrupting one of the instances of the TSE hanging in
await super().wait() and restarting the wait() function. I assume this is not possible in asyncio.?.

[peterhinch]

I can't see a way to do that.