How to gracefully terminate an asyncio script with Ctrl-C?

Question

I've read every post I could find about how to gracefully handle a script with an asyncio event loop getting terminated with Ctrl-C, and I haven't been able to get any of them to work without printing one or more tracebacks as I do so. The answers are pretty much all over the place, and I haven't been able implement any of them into this small script:

import asyncio
import datetime
import functools
import signal


async def display_date(loop):
    end_time = loop.time() + 5.0
    while True:
        print(datetime.datetime.now())
        if (loop.time() + 1.0) >= end_time:
            break
        await asyncio.sleep(1)


def stopper(signame, loop):
    print("Got %s, stopping..." % signame)
    loop.stop()


loop = asyncio.get_event_loop()
for signame in ('SIGINT', 'SIGTERM'):
    loop.add_signal_handler(getattr(signal, signame), functools.partial(stopper, signame, loop))

loop.run_until_complete(display_date(loop))
loop.close()

What I want to happen is for the script to exit without printing any tracebacks following a Ctrl-C (or SIGTERM/SIGINT sent via kill). This code prints RuntimeError: Event loop stopped before Future completed. In the MANY other forms I've tried based on previous answers, I've gotten a plethora of other types of exception classes and error messages with no idea how to fix them. The code above is minimal right now, but some of the attempts I made earlier were anything but, and none of them were correct.

If you're able to modify the script so that it terminates gracefully, an explanation of why your way of doing it is the right way would be greatly appreciated.

ManuelSchneid3r · Answer 1 · 2020-10-23T10:13:58.990

29

Use signal handlers:

import asyncio
from signal import SIGINT, SIGTERM

async def main_coro():
    try:
        await awaitable()
    except asyncio.CancelledError:
        do_cleanup()

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    main_task = asyncio.ensure_future(main_coro())
    for signal in [SIGINT, SIGTERM]:
        loop.add_signal_handler(signal, main_task.cancel)
    try:
        loop.run_until_complete(main_task)
    finally:
        loop.close()

edited Oct 23 '20 at 10:13

answered Nov 13 '19 at 16:00

ManuelSchneid3r

15,850
12
65
103

If you're only writing the coroutine and not the main code, you can use `asyncio.get_running_loop()` and `asyncio.current_task()` to get the loop and task respectively, this way I added signal handlers while in the coroutine (allowing it to be called with `asyncio.run()`) – Scott Stevens Jan 21 '21 at 05:48
4

`asyncio.add_signal_handler()` on win32 raises NotImplementedError (python 3.7.3). – JimB Feb 05 '21 at 12:09

score 24 · Accepted Answer · edited Nov 14 '18 at 13:11

24

Stopping the event loop while it is running will never be valid.

Here, you need to catch the Ctrl-C, to indicate to Python that you wish to handle it yourself instead of displaying the default stacktrace. This can be done with a classic try/except:

coro = display_date(loop)
try:
    loop.run_until_complete(coro)
except KeyboardInterrupt:
    print("Received exit, exiting")

And, for your use-case, that's it! For a more real-life program, you would probably need to cleanup some resources. See also Graceful shutdown of asyncio coroutines

edited Nov 14 '18 at 13:11

Adam Hopkins

6,837
6
32
52

answered Feb 01 '18 at 15:18

Arthur

4,093
3
17
28

1

Thank you. As a clarification for any other readers, I had to remove the whole signal handler-adding `for` loop, and the final `loop.close()` was unnecessary as well. – JK Laiho Feb 02 '18 at 08:10
I can't believe how much time I wasted trying to get `add_signal_handler(SIGINT)` to work; thanks for pointing to a much simpler path that actually works. – Mike C Oct 29 '19 at 17:14
I also had to add `RuntimeError` on the exception because I was receiving `RuntimeError: Event loop stopped before Future completed.` – Leonardo Rick May 10 '21 at 01:47

How to gracefully terminate an asyncio script with Ctrl-C?

2 Answers2

Linked