Python Kill Thread

Python Kill Thread

  1. Raise Exceptions in a Thread to Kill a Thread in Python
  2. Use trace to Kill a Thread in Python
  3. Create/Reset a Stop Flag to Kill a Thread in Python
  4. Use the multiprocessing Module to Kill a Thread in Python
  5. Set the Given Thread as a Daemon Thread to Kill a Thread in Python
  6. Use the Hidden _stop() Function to Kill a Thread in Python

Although it is flagged as a bad programming practice among programmers, it might still be necessary to kill a thread sometimes in Python. This tutorial demonstrates the different means by which we can kill a thread in Python.

The drawback of abruptly putting an end to a threat might leave a task open in the background, leading to a problem.

Moreover, Python does not provide any means to directly kill a thread in Python, which means finding loopholes and indirect ways to implement this essential task.

Now, we will focus on and explain the several ways we can kill a thread in Python.

Raise Exceptions in a Thread to Kill a Thread in Python

This method utilizes the PyThreadState_SetAsyncExc() function, which raises an exception in the given thread asynchronously.

The following code raises an exception in a thread to kill a thread in Python.

import threading
import ctypes
import time
class twe(threading.Thread):
    def __init__(self, name):
        threading.Thread.__init__(self)
        self.name = name        
    def run(self):
        try:
            while True:
                print('running ' + self.name)
        finally:
            print('ended')          
    def get_id(self):
        if hasattr(self, '_thread_id'):
            return self._thread_id
        for id, thread in threading._active.items():
            if thread is self:
                return id
    def raise_exception(self):
        thread_id = self.get_id()
        resu = ctypes.pythonapi.PyThreadState_SetAsyncExc(thread_id,
              ctypes.py_object(SystemExit))
        if resu > 1: 
            ctypes.pythonapi.PyThreadState_SetAsyncExc(thread_id, 0)
            print('Failure in raising exception')      
x = twe('Thread A')
x.start()
time.sleep(2)
x.raise_exception()
x.join()

When the code is run, and as soon as it raises an exception, the run() function is killed as the program control can bypass the try block of the exception handler.

The join() function is then called on to give the final blow and kill the run() function.

Use trace to Kill a Thread in Python

Another way to implement the same task of killing a thread in Python is by installing a trace in the given thread, altering the thread’s execution.

The following code uses traces to kill a thread in Python.

import sys
import time
import threading
import trace
class KThread(threading.Thread):
  def __init__(self, *args, **keywords):
    threading.Thread.__init__(self, *args, **keywords)
    self.killed = False
  def start(self):
    self.__run_backup = self.run
    self.run = self.__run     
    threading.Thread.start(self)
  def __run(self):
    sys.settrace(self.globaltrace)
    self.__run_backup()
    self.run = self.__run_backup
  def globaltrace(self, frame, why, arg):
    if why == 'call':
      return self.localtrace
    else:
      return None
  def localtrace(self, frame, why, arg):
    if self.killed:
      if why == 'line':
        raise SystemExit()
    return self.localtrace
  def kill(self):
    self.killed = True
def exfu():
  print('The function begins')
  for i in range(1,100):
    print(i)
    time.sleep(0.2)
  print('The function ends')
x = KThread(target=exfu)
x.start()
time.sleep(1)
x.kill()

The above code provides the following output.

The function begins
1
2
3
4
5

Here, we utilize the KThread class, a subset of the original threading.Thread class. The KThread class makes the kill() function implemented in the code.

Create/Reset a Stop Flag to Kill a Thread in Python

A stop flag can be declared in the code, which will make it stop the thread’s execution when encountered by the thread.

The following code creates a stop flag to kill a thread in Python.

import threading
import time
def frun():
    while True:
        print('thread running')
        global stop_threads
        if stop_threads:
            break
stop_threads = False
x = threading.Thread(target = frun)
x.start()
time.sleep(1)
stop_threads = True
x.join()
print('killed the thread.')

Use the multiprocessing Module to Kill a Thread in Python

The multiprocessing module makes it possible to spawn processes, with the method and working of it being similar to the threading module as both of them use an API.

The terminate() can kill a given process, which is relatively safer and less complex than killing a thread itself.

The following code uses the multiprocessing module to kill a thread in Python.

import multiprocessing
import time
def cp():
  while True:
   for i in range (20):  
    print ('Process: ', i)
    time.sleep(0.05)
x = multiprocessing.Process(target = cp)
x.start()
time.sleep(0.5)
x.terminate()
print("Terminated the child process")

Set the Given Thread as a Daemon Thread to Kill a Thread in Python

Daemon threads are threads that automatically get killed when the main program is terminated. We can set a given thread as a daemon thread to kill the particular thread in Python.

The following code sets the given thread as a daemon thread to kill a thread in Python.

import threading
import time
import sys
def exfu():
    while True:
        time.sleep(0.5)
        print('Thread alive, but it will die on program termination')
x = threading.Thread(target=exfu)
x.daemon = True
x.start()
time.sleep(2)
sys.exit()

Use the Hidden _stop() Function to Kill a Thread in Python

Although undocumented, a hidden _stop() function can implement the task of killing a thread in Python.

The following code uses the hidden _stop() function to kill a thread in Python.

import time
import threading
 
class th1(threading.Thread):
    def __init__(self, *args, **kwargs):
        super(th1, self).__init__(*args, **kwargs)
        self._stop = threading.Event()
    def stop(self):
        self._stop.set()
    def stopped(self):
        return self._stop.isSet()
    def run(self):
        while True:
            if self.stopped():
                return
            print("Hello, world!")
            time.sleep(1)
 
x = th1()
x.start()
time.sleep(5)
x.stop()
x.join()

Related Article - Python Thread

  • Python Thread Priority
  • Daemon Threads in Python
  • Join Threads in Python
  • Thread Lock in Python
  • Start A Thread in Python