Wait for timeout or event being set for asyncio.Event

Question

I have a class with a method that looks like this:

# self.stop_event -> threading.Event
def run(self):
    while not self.stop_event.wait(3):  # i.e. repeat every 3 sec
        pass  # do stuff

The idea is that several of these are run in their own thread and at some point one thread does stop_event.set(), which naturally stops all others. I want to switch to asyncio for this, because the tasks in run are mostly sleeping and doing IO. Thus, I got to:

# self.stop_event -> asyncio.Event
async def run(self):
    while not self.stop_event.is_set():
        await asyncio.sleep(3)
        pass  # do stuff

The problem is that the asyncio.Event cannot be waited on, so when it is set, there are at most 3 seconds to wait before the method completes. This is a problem, because the sleep time may be minutes. Currently, I am working around this by wrapping the run in an asyncio.Task and then cancelling it like event_loop.call_soon(the_task.cancel).

I want to ask if there is a better way to achieve the above? Is there a way I can wait on an asyncio.Event with a timeout somehow, similar to the threading.Event?

Answer 1

Is there a way I can wait on an asyncio.Event with a timeout somehow, similar to the threading.Event?

asyncio.wait_for supports conveniently adding timeouts to any awaited coroutine. An emulation of the timeout feature of threading.Event.wait for asyncio.Event could look like this:

async def event_wait(evt, timeout):
    # suppress TimeoutError because we'll return False in case of timeout
    with contextlib.suppress(asyncio.TimeoutError):
        await asyncio.wait_for(evt.wait(), timeout)
    return evt.is_set()

This allows a run almost exactly like the one that used threading.Event:

async def run(self):
    while not await event_wait(self.stop_event, 3):
        pass  # do stuff