What's the correct way to clean up after an interrupted event loop?

Question

I have an event loop that runs some co-routines as part of a command line tool. The user may interrupt the tool with the usual Ctrl + C, at which point I want to clean up properly after the interrupted event loop.

Here's what I tried.

import asyncio


@asyncio.coroutine
def shleepy_time(seconds):
    print("Shleeping for {s} seconds...".format(s=seconds))
    yield from asyncio.sleep(seconds)


if __name__ == '__main__':
    loop = asyncio.get_event_loop()

    # Side note: Apparently, async() will be deprecated in 3.4.4.
    # See: https://docs.python.org/3.4/library/asyncio-task.html#asyncio.async
    tasks = [
        asyncio.async(shleepy_time(seconds=5)),
        asyncio.async(shleepy_time(seconds=10))
    ]

    try:
        loop.run_until_complete(asyncio.gather(*tasks))
    except KeyboardInterrupt as e:
        print("Caught keyboard interrupt. Canceling tasks...")

        # This doesn't seem to be the correct solution.
        for t in tasks:
            t.cancel()
    finally:
        loop.close()

Running this and hitting Ctrl + C yields:

$ python3 asyncio-keyboardinterrupt-example.py 
Shleeping for 5 seconds...
Shleeping for 10 seconds...
^CCaught keyboard interrupt. Canceling tasks...
Task was destroyed but it is pending!
task: <Task pending coro=<shleepy_time() running at asyncio-keyboardinterrupt-example.py:7> wait_for=<Future cancelled> cb=[gather.<locals>._done_callback(1)() at /usr/local/Cellar/python3/3.4.3/Frameworks/Python.framework/Versions/3.4/lib/python3.4/asyncio/tasks.py:587]>
Task was destroyed but it is pending!
task: <Task pending coro=<shleepy_time() running at asyncio-keyboardinterrupt-example.py:7> wait_for=<Future cancelled> cb=[gather.<locals>._done_callback(0)() at /usr/local/Cellar/python3/3.4.3/Frameworks/Python.framework/Versions/3.4/lib/python3.4/asyncio/tasks.py:587]>

Clearly, I didn't clean up correctly. I thought perhaps calling cancel() on the tasks would be the way to do it.

What's the correct way to clean up after an interrupted event loop?

Answer 1

When you CTRL+C, the event loop gets stopped, so your calls to t.cancel() don't actually take effect. For the tasks to be cancelled, you need to start the loop back up again.

Here's how you can handle it:

import asyncio

@asyncio.coroutine
def shleepy_time(seconds):
    print("Shleeping for {s} seconds...".format(s=seconds))
    yield from asyncio.sleep(seconds)


if __name__ == '__main__':
    loop = asyncio.get_event_loop()

    # Side note: Apparently, async() will be deprecated in 3.4.4.
    # See: https://docs.python.org/3.4/library/asyncio-task.html#asyncio.async
    tasks = asyncio.gather(
        asyncio.async(shleepy_time(seconds=5)),
        asyncio.async(shleepy_time(seconds=10))
    )

    try:
        loop.run_until_complete(tasks)
    except KeyboardInterrupt as e:
        print("Caught keyboard interrupt. Canceling tasks...")
        tasks.cancel()
        loop.run_forever()
        tasks.exception()
    finally:
        loop.close()

Once we catch KeyboardInterrupt, we call tasks.cancel() and then start the loop up again. run_forever will actually exit as soon as tasks gets cancelled (note that cancelling the Future returned by asyncio.gather also cancels all the Futures inside of it), because the interrupted loop.run_until_complete call added a done_callback to tasks that stops the loop. So, when we cancel tasks, that callback fires, and the loop stops. At that point we call tasks.exception, just to avoid getting a warning about not fetching the exception from the _GatheringFuture.

Answer 2

Note for Python 3.7+: The below is now implemented as part of the standard library asyncio.run function – Replace the below with sys.exit(loop.run(amain(loop))) once you are ready to upgrade! (If you want to print the message, simply move that try…except-clause into amain.)

Updated for Python 3.6+: Add call to loop.shutdown_asyncgens to avoid memory leaks by asynchronous generators that weren't fully used.

The following solution, inspired by some of the other answers, should work in almost all cases and does not depend on you manually keeping track of tasks that need to be cleaned up on Ctrl+C:

loop = asyncio.get_event_loop()
try:
    # Here `amain(loop)` is the core coroutine that may spawn any
    # number of tasks
    sys.exit(loop.run_until_complete(amain(loop)))
except KeyboardInterrupt:
    # Optionally show a message if the shutdown may take a while
    print("Attempting graceful shutdown, press Ctrl+C again to exit…", flush=True)
    
    # Do not show `asyncio.CancelledError` exceptions during shutdown
    # (a lot of these may be generated, skip this if you prefer to see them)
    def shutdown_exception_handler(loop, context):
        if "exception" not in context \
        or not isinstance(context["exception"], asyncio.CancelledError):
            loop.default_exception_handler(context)
    loop.set_exception_handler(shutdown_exception_handler)
    
    # Handle shutdown gracefully by waiting for all tasks to be cancelled
    tasks = asyncio.gather(*asyncio.Task.all_tasks(loop=loop), loop=loop, return_exceptions=True)
    tasks.add_done_callback(lambda t: loop.stop())
    tasks.cancel()
    
    # Keep the event loop running until it is either destroyed or all
    # tasks have really terminated
    while not tasks.done() and not loop.is_closed():
        loop.run_forever()
finally:
    loop.run_until_complete(loop.shutdown_asyncgens())
    loop.close()

The above code will obtain all currently tasks from the event loop using asyncio.Task.all_tasks and place them in a single combined future using asyncio.gather. All tasks in that future (which are all currently running tasks) are then canceled using the future's .cancel() method. The return_exceptions=True then ensures that all the received asyncio.CancelledError exceptions are stored instead of causing the future to become errored.

The above code will also override the default exception handler to prevent the generated asyncio.CancelledError exceptions from being logged.

Update from 2020-12-17: Dropped compatiblity code for Python 3.5.

Answer 3

In Python 3.7+ it's recommended that you use asyncio.run to start an async main function.

asyncio.run will take care of creating the event loop for your program and ensure that the event loop is closed and all tasks are cleaned when the main function exits (including due to a KeyboardInterrupt exception).

It is roughly analogous to the following (see asyncio/runners.py):

def run(coro, *, debug=False):
    """`asyncio.run` is new in Python 3.7"""
    loop = asyncio.get_event_loop()
    try:
        loop.set_debug(debug)
        return loop.run_until_complete(coro)
    finally:
        try:
            all_tasks = asyncio.gather(*asyncio.all_tasks(loop), return_exceptions=True)
            all_tasks.cancel()
            with contextlib.suppress(asyncio.CancelledError):
                loop.run_until_complete(all_tasks)
            loop.run_until_complete(loop.shutdown_asyncgens())
        finally:
            loop.close()

Answer 4

Unless you are on Windows, set up event-loop based signal handlers for SIGINT (and also SIGTERM so you can run it as a service). In these handlers, you may either exit the event loop immediately, or initiate some kind of cleanup sequence and exit later.

Example in official Python documentation: https://docs.python.org/3.4/library/asyncio-eventloop.html#set-signal-handlers-for-sigint-and-sigterm