[FIXED] Interrupt all asyncio.sleep currently executing

Issue

where

This is on Linux, Python 3.5.1.

what

I'm developing a monitor process with asyncio, whose tasks at various places await on asyncio.sleep calls of various durations.

There are points in time when I would like to be able to interrupt all said asyncio.sleep calls and let all tasks proceed normally, but I can't find how to do that. An example is for graceful shutdown of the monitor process.

how (failed assumption)

I thought that I could send an ALRM signal to that effect, but the process dies. I tried catching the ALRM signal with:

def sigalrm_sent(signum, frame):
    tse.logger.info("got SIGALRM")

signal.signal(signal.SIGALRM, sigalrm_sent)

Then I get the log line about catching SIGALRM, but the asyncio.sleep calls are not interrupted.

how (kludge)

At this point, I replaced all asyncio.sleep calls with calls to this coroutine:

async def interruptible_sleep(seconds):
    while seconds > 0 and not tse.stop_requested:
        duration = min(seconds, tse.TIME_QUANTUM)
        await asyncio.sleep(duration)
        seconds -= duration

So I only have to pick a TIME_QUANTUM that is not too small and not too large either.

but

Is there a way to interrupt all running asyncio.sleep calls and I am missing it?

Solution

Interrupting all running calls of asyncio.sleep seems a bit dangerous since it can be used in other parts of the code, for other purposes. Instead I would make a dedicated sleep coroutine that keeps track of its running calls. It is then possible to interrupt them all by canceling the corresponding tasks:

def make_sleep():
    async def sleep(delay, result=None, *, loop=None):
        coro = asyncio.sleep(delay, result=result, loop=loop)
        task = asyncio.ensure_future(coro)
        sleep.tasks.add(task)
        try:
            return await task
        except asyncio.CancelledError:
            return result
        finally:
            sleep.tasks.remove(task)

    sleep.tasks = set()
    sleep.cancel_all = lambda: sum(task.cancel() for task in sleep.tasks)
    return sleep

Example:

async def main(sleep, loop):
    for i in range(10):
        loop.create_task(sleep(i))
    await sleep(3)
    nb_cancelled = sleep.cancel_all()
    await asyncio.wait(sleep.tasks)
    return nb_cancelled

sleep = make_sleep()
loop = asyncio.get_event_loop()
result = loop.run_until_complete(main(sleep, loop)) 
print(result)  # Print '6'

For debugging purposes, loop.time = lambda: float('inf') also works.

Answered By - Vincent