You can interrupt the main thread via the _thread.interrupt_main() function.
In this tutorial you will discover how to interrupt the main thread from another thread in Python.
Let’s get started.
Need to Interrupt Main Thread
A thread is a thread of execution in a computer program.
Every Python program has at least one thread of execution called the main thread. Both processes and threads are created and managed by the underlying operating system.
Sometimes we may need to create additional threads in our program in order to execute code concurrently.
Python provides the ability to create and manage new threads via the threading module and the threading.Thread class.
You can learn more about Python threads in the guide:
In concurrent programming, we may need to interrupt the main thread from another thread.
This may be for many reasons, such as:
- A major change to the program or external system.
- A request from a user or another system.
- A resource is no longer valid or available.
How can we interrupt the main thread from another thread?
Run loops using all CPUs, download your FREE book to learn how.
How to Interrupt the Main Thread
The main thread can be interrupted via the _thread.interrupt_main() function.
Simulate the effect of a signal arriving in the main thread.
— _thread — Low-level threading API
This function can be called from another thread which will by default raise a signal.SIGINT (interruption signal) that will be sent to the main thread. This is the signal sent when the user presses Ctrl-C on the command prompt.
For example:
|
1 2 3 |
... # interrupt the main thread _thread.interrupt_main() |
On Python v3.10 and above, the signal number can be specified as an argument to the _thread.interrupt_main() function.
If given, signum is the number of the signal to simulate. If signum is not given, signal.SIGINT is simulated.
— _thread — Low-level threading API
For example:
|
1 2 3 |
... # interrupt the main thread _thread.interrupt_main(signum=signal.SIGKILL) |
If the SIGINT is not handled in the main thread, it will unravel and terminate the main thread. If there are no other threads running, the process will terminate.
The SIGINT can be handled by the main thread.
One approach is to use a try-except block and catch a KeyboardInterrupt exception.
For example:
|
1 2 3 4 5 6 7 |
... # except a sigint try: ... except KeyboardInterrupt: # handle sigint ... |
An alternate approach is for the main thread to register a handler function for the signal.SIGINT.
This can be achieved using the signal.signal() function and specifying the signal number to handle and the function to handle it.
For example:
|
1 2 3 4 5 6 7 |
# handle single def handle_sigint(signalnum, handler): # ... ... # handle sigint signal(SIGINT, handle_sigint) |
The interruption from the other thread may not have an immediate effect.
This is because the interruption is handled by the Python interpreter itself.
A thread can use this function to interrupt the main thread, though there is no guarantee that the interruption will happen immediately.
— _thread — Low-level threading API
As such, if the main thread is blocked, the signal will not be received and processed by the main thread until after it stops blocking.
Some examples of blocking calls the main thread may be performing that will delay the processing of the interrupt include:
- Sleeping, e.g. a call to time.sleep()
- IO, e.g. reading or writing to file or socket.
- Waiting, e.g. attempting to acquire a lock, semaphore or similar concurrency primitive.
You can learn more about blocking calls in this tutorial:
Now that we know how to interrupt the main thread, let’s look at some worked examples.
Interrupt Main Thread and Handle With Exception
We can explore how to interrupt the main thread and handle the interrupt with a try-except pattern.
In this example we will create a new thread that will block for a while to simulate doing work, then interrupt the main thread. The main thread will start the new thread, then sleep in a loop to simulate work. When interrupted, the main thread will handle the interrupt with a try-except block and will then terminate.
First, we can define a function to execute in a new thread.
The function will block for a moment by calling the time.sleep() function for three seconds. It will then report a message that it is interrupting the main thread, then calls the _thread.interrupt_main() function.
The task() function below implements this.
|
1 2 3 4 5 6 7 |
# task executed in a new thread def task(): # block for a moment sleep(3) # interrupt the main thread print('Interrupting main thread now') interrupt_main() |
Next, in the main thread we will create a new threading.Thread instance and configure it to execute the task() function. The thread can then be started by calling the start() function.
|
1 2 3 4 |
... # start the new thread thread = Thread(target=task) thread.start() |
We can then accept the SIGINT signal for being interrupted and use a try-except pattern to catch a raised KeyboardInterrupt.
In the body of the try-except will block forever to simulate work, printing a message and sleeping for half a second each iteration.
The KeyboardInterrupt is handled by printing a message and terminating the thread with a call to the sys.exit() function.
|
1 2 3 4 5 6 7 8 9 10 11 |
... # handle being interrupted try: # wait around while True: print('Main thread waiting...') sleep(0.5) except KeyboardInterrupt: # terminate main thread print('Main interrupted! Exiting.') sys.exit() |
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 |
# SuperFastPython.com # example of interrupting the main thread and handle with try-except from time import sleep from threading import Thread from _thread import interrupt_main import sys # task executed in a new thread def task(): # block for a moment sleep(3) # interrupt the main thread print('Interrupting main thread now') interrupt_main() # start the new thread thread = Thread(target=task) thread.start() # handle being interrupted try: # wait around while True: print('Main thread waiting...') sleep(0.5) except KeyboardInterrupt: # terminate main thread print('Main interrupted! Exiting.') sys.exit() |
Running the example first creates and starts the new thread.
The main thread then loops, blocking for half a second at a time and reporting a message each iteration.
The new thread starts, blocking for three seconds. It then sends a SIGINT to the main thread and terminates.
The main thread receives the SIGINT signal which raises a KeyboardInterrupt
The KeyboardInterrupt is handled by reporting a message and exiting the main thread, which in turn terminates the program.
This demonstrates how a new thread can interrupt the main thread and how the main thread might expect and handle the interruption with a try-except pattern.
|
1 2 3 4 5 6 7 8 |
Main thread waiting... Main thread waiting... Main thread waiting... Main thread waiting... Main thread waiting... Main thread waiting... Interrupting main thread now Main interrupted! Exiting. |
Next, let’s look at how the main thread might handle an interruption using a signal handler.
Free Python Threading Course
Download your FREE threading PDF cheat sheet and get BONUS access to my free 7-day crash course on the threading API.
Discover how to use the Python threading module including how to create and start new threads and how to use a mutex locks and semaphores
Interrupt Main Thread and Handle With Signal Handler
We can explore how the main thread can handle an expected interruption using a signal handler.
This can be achieved by defining a new function to handle the interruption and registering this function to be called to handle the SIGINT when it occurs.
The example in the previous section can be updated to demonstrate this approach.
First, we must define a new function to handle the SIGINT. This function must take the signal number and the current stack frame as arguments.
In this case we can handle the signal by reporting a message and terminating the thread.
The handle_sigint() function below implements this.
|
1 2 3 4 5 |
# handle single def handle_sigint(signalnum, frame): # terminate print('Main interrupted! Exiting.') sys.exit() |
Next, in the main thread, we can register our function to handle the SIGINT via the signal.signal() function.
|
1 2 3 |
... # register the signal handler for this process signal(SIGINT, handle_sigint) |
And that’s 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 |
# SuperFastPython.com # example of interrupting the main thread and handle with signal handler from time import sleep from threading import Thread from _thread import interrupt_main from signal import signal from signal import SIGINT import sys # handle single def handle_sigint(signalnum, frame): # terminate print('Main interrupted! Exiting.') sys.exit() # task executed in a new thread def task(): # block for a moment sleep(3) # interrupt the main thread print('Interrupting main thread now') interrupt_main() # register the signal handler for this process signal(SIGINT, handle_sigint) # start the new thread thread = Thread(target=task) thread.start() # wait around while True: print('Main thread waiting...') sleep(0.5) |
Running the example first creates and starts the new thread.
The main thread then loops, blocking for half a second at a time and reporting a message each iteration.
The new thread starts, blocking for three seconds. It then sends a SIGINT to the main thread and terminates.
The main thread receives the SIGINT signal which is handled by our custom signal handler function. The function responds by reporting a message and exiting the main thread, which in turn terminates the program.
This demonstrates how a new thread can interrupt the main thread and how the main thread might expect and handle the interruption with a registered signal handler.
|
1 2 3 4 5 6 7 8 |
Main thread waiting... Main thread waiting... Main thread waiting... Main thread waiting... Main thread waiting... Main thread waiting... Interrupting main thread now Main interrupted! Exiting. |
Next, let’s look at an example where the interruption to the main thread is delayed.
Overwhelmed by the python concurrency APIs?
Find relief, download my FREE Python Concurrency Mind Maps
Example of Delayed Interruption of Main Thread
We can explore how an interruption of the main thread may be delayed.
This can occur if the main thread is blocked, such as in a call to sleep, by performing a blocking IO call, or by waiting on a concurrency primitive.
In this example we will have the main thread block for an extended period by calling the sleep function. The main thread will be interrupted in the middle of this sleep, but will not respond until after the blocking call has finished.
This can be achieved by updating the first example above that handles the interruption using the try-except block.
We can update the main thread so that instead of looping forever and sleeping for a fraction of a second, that instead, there is no loop and a single call to the time.sleep() function for seven seconds.
|
1 2 3 4 5 6 7 8 9 10 |
... # handle being interrupted try: # block for a long time print('Main thread waiting...') sleep(7) except KeyboardInterrupt: # terminate main thread print('Main interrupted! Exiting.') sys.exit() |
That’s 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 |
# SuperFastPython.com # example of main thread cannot be interrupting while sleeping from time import sleep from threading import Thread from _thread import interrupt_main import sys # task executed in a new thread def task(): # block for a moment sleep(3) # interrupt the main thread print('Interrupting main thread now') interrupt_main() # start the new thread thread = Thread(target=task) thread.start() # handle being interrupted try: # block for a long time print('Main thread waiting...') sleep(7) except KeyboardInterrupt: # terminate main thread print('Main interrupted! Exiting.') sys.exit() |
Running the example first creates and starts the new thread.
The main thread then reports a message and blocks for an extended period.
The new thread starts, blocking for three seconds. It then sends a SIGINT to the main thread and terminates.
The main thread is still blocking and continues to block until the sleep() function returns.
The SIGINT signal is then received which raises a KeyboardInterrupt The KeyboardInterrupt is handled by reporting a message and exiting the main thread, which in turn terminates the program.
This demonstrates how a new thread can interrupt the main thread and how the main thread will not respond immediately if it is occupied with a blocking function call.
|
1 2 3 |
Main thread waiting... Interrupting main thread now Main interrupted! Exiting. |
Further Reading
This section provides additional resources that you may find helpful.
Python Threading Books
- Python Threading Jump-Start, Jason Brownlee (my book!)
- Threading API Interview Questions
- Threading Module API Cheat Sheet
I also recommend specific chapters in the following books:
- Python Cookbook, David Beazley and Brian Jones, 2013.
- See: Chapter 12: Concurrency
- Effective Python, Brett Slatkin, 2019.
- See: Chapter 7: Concurrency and Parallelism
- Python in a Nutshell, Alex Martelli, et al., 2017.
- See: Chapter: 14: Threads and Processes
Guides
- Python Threading: The Complete Guide
- Python ThreadPoolExecutor: The Complete Guide
- Python ThreadPool: The Complete Guide
APIs
References
Takeaways
You now know how to interrupt the main thread in Python.
Do you have any questions?
Ask your questions in the comments below and I will do my best to answer.
Photo by Kirk Thornton on Unsplash
Leave a Reply