Stop a Running Python Thread Initiatively with Examples: A Completed Guide – Python Tutorial

Suppose you have a function which is used to search files in a directory. This function will take a long time to finish.

For example, when you press Find button, we will start a python thread (search_thread_1) to search files in a directory.

However, we may press Find button again to start a new thread (search_thread_2) to search files  before search_thread_1 thread is finished. At that situation, we should stop search_thread_1 before starting to run search_thread_2.

The example is below:

import threading
import time
import inspect
import ctypes

def searchFiles(dir):
    print("start to search files in "+dir)
    for i in range(100):
        time.sleep(1)
        print("get file "+ str(i)+ " in "+ dir)
    print("search files end in "+ dir)

search_thread_1 = threading.Thread(target=searchFiles, args=["C:\\"])
search_thread_1.start() 
time.sleep(10)
search_thread_2 = threading.Thread(target=searchFiles, args=["E:\\"])

In this code, we define a searchFiles() function to show files found in a directory. First we create and start search_thread_1 thread to search files in C:\.

Then we create a new thread  search_thread_2 to search E:\. However, we should stop search_thread_1 before starting to run search_thread_2.

How to stop a python thread initiatively?

It is a pity that python does not provide functions to stop a running thread, we have to use other ways.

In this tutorial, we define a stop function to stop a running python thread.

def _async_raise(tid, exctype):
    """raises the exception, performs cleanup if needed"""
    tid = ctypes.c_long(tid)
    if not inspect.isclass(exctype):
        exctype = type(exctype)
    res = ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, ctypes.py_object(exctype))
    if res == 0:
        raise ValueError("invalid thread id")
    elif res != 1:
        # """if it returns a number greater than one, you're in trouble,
        # and you should call it again with exc=NULL to revert the effect"""
        ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, None)
        raise SystemError("PyThreadState_SetAsyncExc failed")
 
 
def stop_thread(thread):
    _async_raise(thread.ident, SystemExit)

How to use stop_thread()?

You can stop a running thread like this:

stop_thread(search_thread_1)
search_thread_2.start()

After we stop a running thread search_thread_1, we can start to run search_thread_2.

Run this python code, we can find the result is:

From the result, we can find search_thread_1 is stop successfully by us.