The “Signal Interrupt” or SIGINT signal is raised in a program when the user presses Ctrl-C.
This has the effect of interrupting and often terminating a Python program.
The SIGINT signal can be used to terminate asyncio programs as well, and we can use a custom signal handler function to perform cleanup activities before the asyncio program is closed.
In this tutorial, you will discover how asyncio programs can handle Control-C or SIGINT signals.
Let’s get started.
What is Control-C (SIGINT)
Control-C, or “Signal Interrupt” (SIGINT for short) in Unix-based systems, is a commonly used method to interrupt or terminate the execution of a program.
When we press Control-C in a terminal or send a SIGINT signal to a running process, it triggers an interrupt request, signaling to the program that it should stop its current execution.
The SIGINT signal is sent to a process by its controlling terminal when a user wishes to interrupt the process. This is typically initiated by pressing Ctrl+C …
— Signal (IPC)
In Python, when a program is running, it generally executes in a single thread. If a user sends a Control-C signal, the Python interpreter catches this signal and raises a KeyboardInterrupt exception.
This exception is designed to gracefully handle the interruption and allows the program to perform necessary cleanup actions before exiting.
In regular Python, we can handle this exception by using try and except blocks to catch the KeyboardInterrupt exception, enabling us to execute specific code when the program receives a SIGINT signal.
This approach allows the program to handle the interruption gracefully, such as closing open files, releasing resources, or performing any cleanup required before terminating the program.
How can we handle a Control-C or SIGINT in asyncio programs?
Run loops using all CPUs, download your FREE book to learn how.
What Happens in Asyncio When Control-C (SIGINT) is Received
Pressing Control-C while an asyncio program will send the SIGINT signal to the process.
By default, the SIGINT process will terminate the asyncio program, and this capability was introduced in Python version 3.11.
It does this by calling the cancel() method on the main task.
This raises a CancelledError in the main task if it is suspended.
When the main task exits, all other running tasks are canceled. This is a normal behavior of the main task.
When signal.SIGINT is raised by Ctrl-C, the custom signal handler cancels the main task by calling asyncio.Task.cancel() which raises asyncio.CancelledError inside the main task. This causes the Python stack to unwind, try/except and try/finally blocks can be used for resource cleanup.
— Asyncio Runners, Asyncio API Documentation
You can learn more about the special properties of the main task or “main coroutine” in the tutorial:
If the main task is not suspended and does not suspend after the SIGINT was sent to the process, then the main task will not be canceled and instead will run until completion, after which a KeyboardInterrupt will be raised.
After the main task is cancelled, asyncio.Runner.run() raises KeyboardInterrupt.
— Asyncio Runners, Asyncio API Documentation
Raising a KeyboardInterrupt exception in the main thread is the default behavior for a regular Python program, and provides a fallback if the asyncio program does not respond to the SIGINT.
When signal.SIGINT is raised by Ctrl-C, KeyboardInterrupt exception is raised in the main thread by default.
— Asyncio Runners, Asyncio API Documentation
Now that we know the default behavior of SIGINT in asyncio programs, let’s look at adding a custom handler.
How to Handle Control-C (SIGINT)
We can add a custom handler for SIGINT in our asyncio programs.
This capability was introduced in Python 3.11 and is only supported on Unix-based platforms (e.g. Linux and macOS, not Windows).
This can be achieved by defining a regular Python function to call when a SIGINT is received by the asyncio program, registering our function as a handler at the beginning of our asyncio program.
The handler function does not take any arguments by default.
For example:
|
1 2 3 |
# custom signal handler def signal_handler(): # ... |
We can then register it to handle SIGINT signals by first getting access to the asyncio event loop and then calling the add_signal_handler() method and specifying the signal type to handle and the function to call.
For example:
|
1 2 3 4 5 |
... # get the event loop loop = asyncio.get_running_loop() # add a custom signal handler loop.add_signal_handler(signal.SIGINT, signal_handler) |
We could then perform some custom actions in our handler, such as logging or reporting the request to shut down, close resources, like socket servers, and cancel running tasks.
For example, we might cancel all running tasks as follows:
|
1 2 3 4 5 |
# custom signal handler def signal_handler(): # cancel all tasks for task in asyncio.all_tasks(): task.cancel() |
Because the handler is a regular Python function, we cannot directly await coroutines or tasks.
Now that we know how to handle SIGINT in our asyncio programs, let’s look at some worked examples.
Free Python Asyncio Course
Download your FREE Asyncio PDF cheat sheet and get BONUS access to my free 7-day crash course on the Asyncio API.
Discover how to use the Python asyncio module including how to define, create, and run new coroutines and how to use non-blocking I/O.
Example of Control-C (SIGINT) in Asyncio
We can explore the default behavior of raising a SIGINT signal in an asyncio program.
In this case, we will define a task that reports a message and sleeps for a long duration. The main coroutine will run this task in the background, then raise a SIGINT signal and wait for it to take effect. We will use try-except blocks to handle exceptions in order to see what exceptions are raised in each task, if at all.
Firstly, we can define the background task that sleeps for an extended duration.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 |
# coroutine to perform some work async def work(): try: # report a message print('Work is starting...') # suspend a while await asyncio.sleep(10) # report a message print('Work is done') except asyncio.CancelledError as e: print('work() cancelled') except Exception as e: print(f'work() got: {e}') |
Next, we can define the main() coroutine.
Firstly we can create and schedule our work() coroutine as a background task, then suspend a moment to allow it to run.
|
1 2 3 4 5 |
... # create and schedule the event task = asyncio.create_task(work()) # wait a moment await asyncio.sleep(2) |
Next, we can raise a signal in the process, then sleep a moment to allow the main coroutine/task to respond to it.
We can raise a signal within our program via the signal.raise_signal() and pass it the details of the signal to raise, such as signal.SIGINT.
|
1 2 3 4 5 6 7 |
... # report progress print('Sending SIGINT') # raise a sigint in the current process signal.raise_signal(signal.SIGINT) # wait a moment await asyncio.sleep(2) |
Tying this together, the complete main() coroutine is listed below.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 |
# main coroutine async def main(): try: # create and schedule the event task = asyncio.create_task(work()) # wait a moment await asyncio.sleep(2) # report progress print('Sending SIGINT') # raise a sigint in the current process signal.raise_signal(signal.SIGINT) # wait a moment await asyncio.sleep(2) # report a final message print('Main is done') except asyncio.CancelledError as e: print('main() cancelled') except Exception as e: print(f'main() got: {e}') |
We can then start the event loop and run the main() coroutine.
|
1 2 3 |
... # start the event loop asyncio.run(main()) |
Tying this together, the complete example is listed below.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 |
# SuperFastPython.com # example of default SIGINT handling import signal import asyncio # coroutine to perform some work async def work(): try: # report a message print('Work is starting...') # suspend a while await asyncio.sleep(10) # report a message print('Work is done') except asyncio.CancelledError as e: print('work() cancelled') except Exception as e: print(f'work() got: {e}') # main coroutine async def main(): try: # create and schedule the event task = asyncio.create_task(work()) # wait a moment await asyncio.sleep(2) # report progress print('Sending SIGINT') # raise a sigint in the current process signal.raise_signal(signal.SIGINT) # wait a moment await asyncio.sleep(2) # report a final message print('Main is done') except asyncio.CancelledError as e: print('main() cancelled') except Exception as e: print(f'main() got: {e}') # start the event loop asyncio.run(main()) |
Running the example first creates the main() coroutine and starts the asyncio event loop to run it.
The main() coroutine runs and creates the work() coroutine and runs it as a background task. It then suspends a moment to allow the task to begin.
The work() coroutine runs and reports a message, it then sleeps for an extended period.
The main() coroutine results and raises a SIGINT in the current process that is running the asyncio event loop. It then suspends for a moment to allow the SIGINT to take effect.
The SIGNAL is handled by the asyncio event loop and the main() coroutine is canceled. A message is reported confirming that the main() coroutine was canceled.
Once the main() coroutine is canceled, the asyncio event loop then cancels all other running tasks, such as our work() task. This is the default behavior of the asyncio event loop when the main task exits. The work() task reports a message confirming it was canceled.
A KeyboardInterrupt exception is not raised because the asyncio program handled the SIGINT and the main thread did not need to handle it.
This highlights the default behavior when a SIGINT is raised in an asyncio program.
|
1 2 3 4 |
Work is starting... Sending SIGINT main() cancelled work() cancelled |
Next, let’s explore an example of a custom SIGINT handler in an asyncio program.
Overwhelmed by the python concurrency APIs?
Find relief, download my FREE Python Concurrency Mind Maps
Example of Control-C (SIGINT) Customer Handler in Asyncio
We can explore an example of a custom SIGINT handler in our asyncio program.
In this case, we can update the above example to define a custom SIGINT handler function, and to register it at the beginning of our program.
In this case, we will handle the SIGINT by doing nothing, allowing the asyncio program to run until completion.
Our custom signal handler function is listed below.
|
1 2 3 4 |
# custom signal handler def signal_handler(): # report progress print('Custom SIGINT handler running...') |
We can then register this function to be called and handle the SIGINT at the beginning of our program,
|
1 2 3 4 5 |
... # get the event loop loop = asyncio.get_running_loop() # add a custom signal handler loop.add_signal_handler(signal.SIGINT, signal_handler) |
Tying this together, the complete example is listed below.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 |
# SuperFastPython.com # example of custom SIGINT handling import signal import asyncio # custom signal handler def signal_handler(): # report progress print('Custom SIGINT handler running...') # coroutine to perform some work async def work(): try: # report a message print('Work is starting...') # suspend a while await asyncio.sleep(10) # report a message print('Work is done') except asyncio.CancelledError as e: print('work() cancelled') except Exception as e: print(f'work() got: {e}') # main coroutine async def main(): # get the event loop loop = asyncio.get_running_loop() # add a custom signal handler loop.add_signal_handler(signal.SIGINT, signal_handler) try: # create and schedule the event task = asyncio.create_task(work()) # wait a moment await asyncio.sleep(2) # report progress print('Sending SIGINT') # raise a sigint in the current process signal.raise_signal(signal.SIGINT) # wait a moment await asyncio.sleep(2) # report a final message print('Main is done') except asyncio.CancelledError as e: print('main() cancelled') except Exception as e: print(f'main() got: {e}') # start the event loop asyncio.run(main()) |
Running the example first creates the main() coroutine and starts the asyncio event loop to run it.
The main() coroutine runs and registers our signal_handler() function to handle any SIGINT signals that are raised.
The main() coroutine then creates the work() coroutine and runs it as a background task. It then suspends for a moment to allow the task to begin.
The work() coroutine runs and reports a message, it then sleeps for an extended period.
The main() coroutine results and raises a SIGINT in the current process that is running the asyncio event loop. It then suspends for a moment to allow the SIGINT to take effect.
The SIGNAL is handled by our custom signal_handler() function and reports a message.
The asyncio program is not terminated.
The main() coroutine resumes after its sleep, reports a final message, then terminates.
When the main() coroutine terminates, the work() coroutine is canceled, the normal behavior when the main task exits.
This highlights how we can use a custom handler function to handle SIGINT signals in our asyncio programs.
|
1 2 3 4 5 |
Work is starting... Sending SIGINT Custom SIGINT handler running... Main is done work() cancelled |
Next, let’s look at an example of the main() coroutine being too busy to handle a SIGINT signal.
Example of Control-C (SIGINT) Never Handled
We can explore an example where the main() coroutine is busy and never responds to a raised SIGINT.
This will prevent the asyncio program from handling the signal, which will then be handled by the main thread and raise a KeyboardInterrupt.
In this case, we can update the first example above so that the main() coroutine runs a loop that blocks the asyncio event loop.
For example, we can achieve this with calls to the time.sleep() function, not the asyncio.sleep() function.
|
1 2 3 4 |
... # perform blocking task for i in range(5): time.sleep(1) |
The updated main() coroutine with this change is listed below.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 |
# main coroutine async def main(): try: # create and schedule the event task = asyncio.create_task(work()) # wait a moment await asyncio.sleep(2) # report progress print('Sending SIGINT') # raise a sigint in the current process signal.raise_signal(signal.SIGINT) # perform blocking task for i in range(5): time.sleep(1) # report a final message print('Main is done') except asyncio.CancelledError as e: print('main() cancelled') except Exception as e: print(f'main() got: {e}') |
This does not suspend the main coroutine and allows other coroutines to run. Instead, it blocks the asyncio event loop and prevents any other coroutines from running.
As such, the main coroutine does not suspend after the SIGINT is raised and never handles it.
Tying this together, the complete example is listed below.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 |
# SuperFastPython.com # example of default SIGINT handling import signal import asyncio import time # coroutine to perform some work async def work(): try: # report a message print('Work is starting...') # suspend a while await asyncio.sleep(10) # report a message print('Work is done') except asyncio.CancelledError as e: print('work() cancelled') except Exception as e: print(f'work() got: {e}') # main coroutine async def main(): try: # create and schedule the event task = asyncio.create_task(work()) # wait a moment await asyncio.sleep(2) # report progress print('Sending SIGINT') # raise a sigint in the current process signal.raise_signal(signal.SIGINT) # perform blocking task for i in range(5): time.sleep(1) # report a final message print('Main is done') except asyncio.CancelledError as e: print('main() cancelled') except Exception as e: print(f'main() got: {e}') # start the event loop asyncio.run(main()) |
Running the example first creates the main() coroutine and starts the asyncio event loop to run it.
The main() coroutine runs and registers our signal_handler() function to handle any SIGINT signals that are raised.
The main() coroutine then creates the work() coroutine and runs it as a background task. It then suspends for a moment to allow the task to begin.
The work() coroutine runs and reports a message, it then sleeps for an extended period.
The main() coroutine results and raises a SIGINT in the current process that is running the asyncio event loop.
It then does not suspend and allows the event loop to handle the signal. Instead, it loops and executes a blocking call, then reports a final message and terminates.
When the main() coroutine terminates, the work() coroutine is canceled, the normal behavior when the main task exits.
Finally, the main thread of the program handles the SIGINT, and a KeyboardInterrupt is raised and not handled. This terminates the program.
This highlights that the main coroutine in an asyncio program can be too busy to respond to a SIGINT, which is then handled by the main thread after the asyncio program terminates.
|
1 2 3 4 5 6 7 8 9 10 11 |
Work is starting... Sending SIGINT Main is done work() cancelled Traceback (most recent call last): asyncio.exceptions.CancelledError During handling of the above exception, another exception occurred: KeyboardInterrupt |
Takeaways
You now know how asyncio programs can handle Control-C or SIGINT signals.
Did I make a mistake? See a typo?
I’m a simple humble human. Correct me, please!
Do you have any additional tips?
I’d love to hear about them!
Do you have any questions?
Ask your questions in the comments below and I will do my best to answer.
Photo by Bing Hui Yau on Unsplash
Thank you for the article! I wrote a small package: https://github.com/aliev/aioshutdown that allows capturing signals through a context manager in asyncio. Can be useful for graceful shutdown and can reduce the boilerplate code.
Thanks for sharing.