Hilo de eliminación de Python

Vaibhhav Khetarpal 30 enero 2023
  1. Generar excepciones en un hilo para matar un hilo en Python
  2. Use trace para matar un hilo en Python
  3. Crear/restablecer un indicador de parada para cerrar un hilo en Python
  4. Use el módulo multiprocessing para matar un hilo en Python
  5. Establezca el subproceso dado como un subproceso de Daemon para eliminar un subproceso en Python
  6. Use la función oculta _stop() para matar un hilo en Python
Hilo de eliminación de Python

Aunque está marcado como una mala práctica de programación entre los programadores, aún puede ser necesario cerrar un hilo a veces en Python. Este tutorial demuestra los diferentes medios por los cuales podemos matar un hilo en Python.

El inconveniente de poner fin abruptamente a una amenaza puede dejar una tarea abierta en segundo plano, lo que genera un problema.

Además, Python no proporciona ningún medio para matar directamente un hilo en Python, lo que significa encontrar lagunas y formas indirectas de implementar esta tarea esencial.

Ahora, nos centraremos y explicaremos las diversas formas en que podemos cerrar un hilo en Python.

Generar excepciones en un hilo para matar un hilo en Python

Este método utiliza la función PyThreadState_SetAsyncExc(), que genera una excepción en el subproceso dado de forma asíncrona.

El siguiente código genera una excepción en un hilo para eliminar un hilo 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()

Cuando se ejecuta el código, y tan pronto como genera una excepción, la función run() se elimina ya que el control del programa puede omitir el bloque try del controlador de excepciones.

Luego se llama a la función join() para dar el golpe final y eliminar la función run().

Use trace para matar un hilo en Python

Otra forma de implementar la misma tarea de matar un subproceso en Python es instalando un rastreo en el subproceso dado, alterando la ejecución del subproceso.

El siguiente código usa rastros para cerrar un hilo 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()

El código anterior proporciona el siguiente resultado.

The function begins
1
2
3
4
5

Aquí utilizamos la clase KThread, un subconjunto de la clase original threading.Thread. La clase KThread hace que la función kill() esté implementada en el código.

Crear/restablecer un indicador de parada para cerrar un hilo en Python

Se puede declarar un indicador de parada en el código, lo que hará que detenga la ejecución del subproceso cuando lo encuentre.

El siguiente código crea un indicador de parada para cerrar un hilo 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.")

Use el módulo multiprocessing para matar un hilo en Python

El módulo de multiprocessing permite generar procesos, con un método y funcionamiento similar al del módulo threading, ya que ambos utilizan una API.

El terminate() puede matar un proceso dado, lo cual es relativamente más seguro y menos complejo que matar un hilo en sí.

El siguiente código usa el módulo de multiprocessing para matar un hilo 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")

Establezca el subproceso dado como un subproceso de Daemon para eliminar un subproceso en Python

Los subprocesos daemon son subprocesos que se eliminan automáticamente cuando finaliza el programa principal. Podemos configurar un subproceso dado como un subproceso daemon para eliminar el subproceso en particular en Python.

El siguiente código establece el subproceso dado como un subproceso daemon para eliminar un subproceso 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()

Use la función oculta _stop() para matar un hilo en Python

Aunque no está documentada, una función oculta _stop() puede implementar la tarea de matar un hilo en Python.

El siguiente código usa la función oculta _stop() para matar un hilo 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

Artículo relacionado - Python Thread