Gibt es eine Möglichkeit, einen Thread zu löschen?

Question

Ist es möglich, einen laufenden Thread zu beenden, ohne irgendwelche Flags/Semaphoren/etc. zu setzen/zu überprüfen?

Answer 1

Es ist zwar schon ziemlich alt, este könnte für manche eine praktische Lösung sein:

Ein kleines Modul, das die Funktionalität des Threading-Moduls erweitert - erlaubt einem Thread, Ausnahmen im Kontext eines anderen Threads zu Threads. Durch das Auslösen SystemExit können Sie endlich Python-Threads beenden.

import threading
import ctypes     

def _async_raise(tid, excobj):
    res = ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, ctypes.py_object(excobj))
    if res == 0:
        raise ValueError("nonexistent 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, 0)
        raise SystemError("PyThreadState_SetAsyncExc failed")

class Thread(threading.Thread):
    def raise_exc(self, excobj):
        assert self.isAlive(), "thread must be started"
        for tid, tobj in threading._active.items():
            if tobj is self:
                _async_raise(tid, excobj)
                return

        # the thread was alive when we entered the loop, but was not found 
        # in the dict, hence it must have been already terminated. should we raise
        # an exception here? silently ignore?

    def terminate(self):
        # must raise the SystemExit type, instead of a SystemExit() instance
        # due to a bug in PyThreadState_SetAsyncExc
        self.raise_exc(SystemExit)

Auf diese Weise kann ein "Thread Ausnahmen im Kontext eines anderen Threads auslösen" und der beendete Thread kann die Beendigung behandeln, ohne regelmäßig ein Abbruch-Flag zu überprüfen.

Nach seinen Angaben ist jedoch Originalquelle gibt es einige Probleme mit diesem Code.

• Die Ausnahme wird nur bei der Ausführung von Python Bytecode ausgelöst. Wenn Ihr Thread eine native/eingebaute blockierende Funktion aufruft, wird die Ausnahme nur ausgelöst, wenn die Ausführung zum Python-Code zurückkehrt. [ ]
  • Es gibt auch ein Problem, wenn die eingebaute Funktion intern PyErr_Clear() aufruft, was zu einer Sie können versuchen, sie erneut auszulösen.
• Nur Ausnahmetypen können sicher ausgelöst werden. E
  • Zum Beispiel: t1.raise_exc(TypeError) und nicht t1.raise_exc(TypeError("blah")).
  • IMHO ist das ein Fehler, und ich habe ihn als solchen gemeldet. Für mehr Informationen, http://mail.python.org/pipermail/python-dev/2006-August/068158.html
• Ich habe darum gebeten, diese Funktion im eingebauten Thread-Modul darzustellen, aber da ctypes eine Standardbibliothek geworden ist (ab 2.5) und diese
  Merkmal wahrscheinlich nicht implementierungsunabhängig ist, kann es beibehalten werden
  nicht belichtet.

Answer 2

Wenn man davon ausgeht, dass man mehrere Threads der gleichen Funktion haben möchte, ist dies IMHO die einfachste Implementierung, um einen nach Id zu stoppen:

import time
from threading import Thread

def doit(id=0):
    doit.stop=0
    print("start id:%d"%id)
    while 1:
        time.sleep(1)
        print(".")
        if doit.stop==id:
            doit.stop=0
            break
    print("end thread %d"%id)

t5=Thread(target=doit, args=(5,))
t6=Thread(target=doit, args=(6,))

t5.start() ; t6.start()
time.sleep(2)
doit.stop =5  #kill t5
time.sleep(2)
doit.stop =6  #kill t6

Das Schöne daran ist, dass man hier mehrere gleiche und unterschiedliche Funktionen haben kann, und sie alle durch functionname.stop

Wenn Sie nur einen Thread der Funktion haben wollen, brauchen Sie sich die ID nicht zu merken. Einfach aufhören, wenn doit.stop > 0.

Answer 3

Python-Version: 3.8

Mit Daemon-Thread auszuführen, was wir wollten, wenn wir wollen, Daemon-Thread beendet werden, alles, was wir brauchen, ist die Eltern-Thread beenden, dann wird das System Daemon-Thread, die Eltern-Thread erstellt beenden.

Unterstützt auch Coroutine und Coroutine-Funktion.

def main():
    start_time = time.perf_counter()
    t1 = ExitThread(time.sleep, (10,), debug=False)
    t1.start()
    time.sleep(0.5)
    t1.exit()
    try:
        print(t1.result_future.result())
    except concurrent.futures.CancelledError:
        pass
    end_time = time.perf_counter()
    print(f"time cost {end_time - start_time:0.2f}")

Nachfolgend der Quellcode von ExitThread

import concurrent.futures
import threading
import typing
import asyncio

class _WorkItem(object):
    """ concurrent\futures\thread.py

    """

    def __init__(self, future, fn, args, kwargs, *, debug=None):
        self._debug = debug
        self.future = future
        self.fn = fn
        self.args = args
        self.kwargs = kwargs

    def run(self):
        if self._debug:
            print("ExitThread._WorkItem run")
        if not self.future.set_running_or_notify_cancel():
            return

        try:
            coroutine = None
            if asyncio.iscoroutinefunction(self.fn):
                coroutine = self.fn(*self.args, **self.kwargs)
            elif asyncio.iscoroutine(self.fn):
                coroutine = self.fn
            if coroutine is None:
                result = self.fn(*self.args, **self.kwargs)
            else:
                result = asyncio.run(coroutine)
            if self._debug:
                print("_WorkItem done")
        except BaseException as exc:
            self.future.set_exception(exc)
            # Break a reference cycle with the exception 'exc'
            self = None
        else:
            self.future.set_result(result)

class ExitThread:
    """ Like a stoppable thread

    Using coroutine for target then exit before running may cause RuntimeWarning.

    """

    def __init__(self, target: typing.Union[typing.Coroutine, typing.Callable] = None
                 , args=(), kwargs={}, *, daemon=None, debug=None):
        #
        self._debug = debug
        self._parent_thread = threading.Thread(target=self._parent_thread_run, name="ExitThread_parent_thread"
                                               , daemon=daemon)
        self._child_daemon_thread = None
        self.result_future = concurrent.futures.Future()
        self._workItem = _WorkItem(self.result_future, target, args, kwargs, debug=debug)
        self._parent_thread_exit_lock = threading.Lock()
        self._parent_thread_exit_lock.acquire()
        self._parent_thread_exit_lock_released = False  # When done it will be True
        self._started = False
        self._exited = False
        self.result_future.add_done_callback(self._release_parent_thread_exit_lock)

    def _parent_thread_run(self):
        self._child_daemon_thread = threading.Thread(target=self._child_daemon_thread_run
                                                     , name="ExitThread_child_daemon_thread"
                                                     , daemon=True)
        self._child_daemon_thread.start()
        # Block manager thread
        self._parent_thread_exit_lock.acquire()
        self._parent_thread_exit_lock.release()
        if self._debug:
            print("ExitThread._parent_thread_run exit")

    def _release_parent_thread_exit_lock(self, _future):
        if self._debug:
            print(f"ExitThread._release_parent_thread_exit_lock {self._parent_thread_exit_lock_released} {_future}")
        if not self._parent_thread_exit_lock_released:
            self._parent_thread_exit_lock_released = True
            self._parent_thread_exit_lock.release()

    def _child_daemon_thread_run(self):
        self._workItem.run()

    def start(self):
        if self._debug:
            print(f"ExitThread.start {self._started}")
        if not self._started:
            self._started = True
            self._parent_thread.start()

    def exit(self):
        if self._debug:
            print(f"ExitThread.exit exited: {self._exited} lock_released: {self._parent_thread_exit_lock_released}")
        if self._parent_thread_exit_lock_released:
            return
        if not self._exited:
            self._exited = True
            if not self.result_future.cancel():
                if self.result_future.running():
                    self.result_future.set_exception(concurrent.futures.CancelledError())

Answer 4

Wie bereits in @Kozyarchuk's Antwort die Installation von Verfolgungsarbeiten. Da diese Antwort keinen Code enthielt, finden Sie hier ein funktionierendes, einsatzbereites Beispiel:

import sys, threading, time 

class TraceThread(threading.Thread): 
    def __init__(self, *args, **keywords): 
        threading.Thread.__init__(self, *args, **keywords) 
        self.killed = False
    def start(self): 
        self._run = self.run 
        self.run = self.settrace_and_run
        threading.Thread.start(self) 
    def settrace_and_run(self): 
        sys.settrace(self.globaltrace) 
        self._run()
    def globaltrace(self, frame, event, arg): 
        return self.localtrace if event == 'call' else None
    def localtrace(self, frame, event, arg): 
        if self.killed and event == 'line': 
            raise SystemExit() 
        return self.localtrace 

def f(): 
    while True: 
        print('1') 
        time.sleep(2)
        print('2') 
        time.sleep(2)
        print('3') 
        time.sleep(2)

t = TraceThread(target=f) 
t.start() 
time.sleep(2.5) 
t.killed = True

Er stoppt nach dem Drucken 1 y 2 . 3 wird nicht gedruckt.

Answer 5

Wenn Sie wirklich die Möglichkeit benötigen, eine Unteraufgabe zu beenden, verwenden Sie eine alternative Implementierung. multiprocessing y gevent beide unterstützen das wahllose Töten eines "Fadens".

Das Threading von Python unterstützt keine Stornierung. Versuchen Sie es gar nicht erst. Ihr Code wird höchstwahrscheinlich blockieren, Speicher beschädigen oder auslaufen lassen oder andere unbeabsichtigte "interessante", schwer zu debuggende Effekte haben, die selten und nicht deterministisch auftreten.