Python Kill Thread

Vaibhhav Khetarpal 30 janvier 2023
  1. Lever des exceptions dans un thread pour tuer un thread en Python
  2. Utilisez trace pour tuer un thread en Python
  3. Créer/réinitialiser un indicateur d’arrêt pour tuer un thread en Python
  4. Utiliser le module multiprocessing pour tuer un thread en Python
  5. Définir le thread donné en tant que thread démon pour tuer un thread en Python
  6. Utilisez la fonction cachée _stop() pour tuer un thread en Python
Python Kill Thread

Bien que cela soit signalé comme une mauvaise pratique de programmation parmi les programmeurs, il peut toujours être nécessaire de tuer un thread parfois en Python. Ce tutoriel montre les différents moyens par lesquels nous pouvons tuer un thread en Python.

L’inconvénient de mettre fin brusquement à une menace peut laisser une tâche ouverte en arrière-plan, entraînant un problème.

De plus, Python ne fournit aucun moyen de tuer directement un thread en Python, ce qui signifie trouver des failles et des moyens indirects d’implémenter cette tâche essentielle.

Maintenant, nous allons nous concentrer sur et expliquer les différentes façons de tuer un thread en Python.

Lever des exceptions dans un thread pour tuer un thread en Python

Cette méthode utilise la fonction PyThreadState_SetAsyncExc(), qui lève une exception dans le thread donné de manière asynchrone.

Le code suivant lève une exception dans un thread pour tuer un thread en 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()

Lorsque le code est exécuté, et dès qu’il lève une exception, la fonction run() est tuée car le contrôle du programme peut contourner le bloc try du gestionnaire d’exceptions.

La fonction join() est alors appelée pour donner le coup de grâce et tuer la fonction run().

Utilisez trace pour tuer un thread en Python

Une autre façon d’implémenter la même tâche de tuer un thread en Python est d’installer une trace dans le thread donné, modifiant l’exécution du thread.

Le code suivant utilise des traces pour tuer un thread en 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()

Le code ci-dessus fournit la sortie suivante.

The function begins
1
2
3
4
5

Ici, nous utilisons la classe KThread, un sous-ensemble de la classe originale threading.Thread. La classe KThread rend la fonction kill() implémentée dans le code.

Créer/réinitialiser un indicateur d’arrêt pour tuer un thread en Python

Un indicateur d’arrêt peut être déclaré dans le code, ce qui le fera arrêter l’exécution du thread lorsqu’il le rencontrera.

Le code suivant crée un indicateur d’arrêt pour tuer un thread en 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.")

Utiliser le module multiprocessing pour tuer un thread en Python

Le module multiprocessing permet de spawner des processus, dont la méthode et le fonctionnement sont similaires au module threading car tous deux utilisent une API.

Le terminate() peut tuer un processus donné, ce qui est relativement plus sûr et moins complexe que de tuer un thread lui-même.

Le code suivant utilise le module multiprocessing pour tuer un thread en 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")

Définir le thread donné en tant que thread démon pour tuer un thread en Python

Les threads démons sont des threads qui sont automatiquement tués lorsque le programme principal est terminé. Nous pouvons définir un thread donné en tant que thread démon pour tuer le thread particulier en Python.

Le code suivant définit le thread donné en tant que thread démon pour tuer un thread en 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()

Utilisez la fonction cachée _stop() pour tuer un thread en Python

Bien que non documentée, une fonction cachée _stop() peut implémenter la tâche de tuer un thread en Python.

Le code suivant utilise la fonction cachée _stop() pour tuer un thread en 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()
Vaibhhav Khetarpal avatar Vaibhhav Khetarpal avatar

Vaibhhav is an IT professional who has a strong-hold in Python programming and various projects under his belt. He has an eagerness to discover new things and is a quick learner.

LinkedIn

Article connexe - Python Thread