Interrupt the Main Thread in Python

Last Updated on September 12, 2022

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.

Table of Contents

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:

Threading in Python: The Complete 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:

...

# 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:

...

# 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:

...

# 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:

# 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:

Thread Blocking Call in Python

Now that we know how to interrupt the main thread, let’s look at some worked examples.

Download Now: Free Threading PDF Cheat Sheet

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.

# 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.

...

# 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.

...

# 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.

# 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.

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

Learn more

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.

# 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.

...

# register the signal handler for this process

signal(SIGINT, handle_sigint)

And that’s it.

Tying this together, the complete example is listed below.

# 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.

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.

...

# 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.

# 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.

Main thread waiting...

Interrupting main thread now

Main interrupted! Exiting.