Fangen Sie die Ausnahme eines Threads im Aufrufer-Thread in Python ab

Das Modul concurrent.futures macht es einfach, in separaten Threads (oder Prozessen) zu arbeiten und die daraus resultierenden Ausnahmen zu behandeln:

import concurrent.futures
import shutil

def copytree_with_dots(src_path, dst_path):
    with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
        # Execute the copy on a separate thread,
        # creating a future object to track progress.
        future = executor.submit(shutil.copytree, src_path, dst_path)

        while future.running():
            # Print pretty dots here.
            pass

        # Return the value returned by shutil.copytree(), None.
        # Raise any exceptions raised during the copy process.
        return future.result()

concurrent.futures ist in Python 3.2 enthalten und für frühere Versionen als das zurückportierte futures-Modul verfügbar.

Obwohl es nicht möglich ist, eine in einem anderen Thread geworfene Ausnahme direkt abzufangen, ist hier ein Code, um ganz transparent etwas zu erhalten, das dieser Funktionalität sehr nahe kommt. Ihr untergeordneter Thread muss die Klasse ExThread anstelle von threading.Thread unterteilen, und der übergeordnete Thread muss die Methode child_thread.join_with_exception() anstelle von child_thread.join() aufrufen, wenn er darauf wartet, dass der Thread seinen Job beendet.

Technische Details zu dieser Implementierung: Wenn der untergeordnete Thread eine Ausnahme auslöst, wird er über eine Queue an das übergeordnete Thread übergeben und erneut im übergeordneten Thread geworfen. Beachten Sie, dass bei diesem Ansatz nicht viel zu warten ist.

#!/usr/bin/env python

import sys
import threading
import Queue

class ExThread(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)
        self.__status_queue = Queue.Queue()

    def run_with_exception(self):
        """This method should be overriden."""
        raise NotImplementedError

    def run(self):
        """This method should NOT be overriden."""
        try:
            self.run_with_exception()
        except BaseException:
            self.__status_queue.put(sys.exc_info())
        self.__status_queue.put(None)

    def wait_for_exc_info(self):
        return self.__status_queue.get()

    def join_with_exception(self):
        ex_info = self.wait_for_exc_info()
        if ex_info is None:
            return
        else:
            raise ex_info[1]

class MyException(Exception):
    pass

class MyThread(ExThread):
    def __init__(self):
        ExThread.__init__(self)

    def run_with_exception(self):
        thread_name = threading.current_thread().name
        raise MyException("An error in thread '{}'.".format(thread_name))

def main():
    t = MyThread()
    t.start()
    try:
        t.join_with_exception()
    except MyException as ex:
        thread_name = threading.current_thread().name
        print "Caught a MyException in thread '{}': {}".format(thread_name, ex)

if __== '__main__':
    main()

Wenn eine Ausnahme in einem Thread auftritt, ist es am besten, sie im Aufrufer-Thread während join erneut anzuheben. Informationen zu der gerade behandelten Ausnahme erhalten Sie mit der Funktion sys.exc_info(). Diese Informationen können einfach als Eigenschaft des Thread-Objekts gespeichert werden, bis join aufgerufen wird. Anschließend können sie erneut aufgerufen werden.

Beachten Sie, dass ein Queue.Queue (wie in anderen Antworten vorgeschlagen) in diesem einfachen Fall nicht erforderlich ist, wenn der Thread höchstens eine Ausnahme auslöst und direkt nach dem Auslösen einer Ausnahme ausläuft. Wir vermeiden Rennenbedingungen, indem wir einfach auf den Abschluss des Threads warten.

Erweitern Sie beispielsweise ExcThread (unten) und überschreiben Sie excRun (anstelle von run).

Python 2.x:

import threading

class ExcThread(threading.Thread):
  def excRun(self):
    pass

  def run(self):
    self.exc = None
    try:
      # Possibly throws an exception
      self.excRun()
    except:
      import sys
      self.exc = sys.exc_info()
      # Save details of the exception thrown but don't rethrow,
      # just complete the function

  def join(self):
    threading.Thread.join(self)
    if self.exc:
      msg = "Thread '%s' threw an exception: %s" % (self.getName(), self.exc[1])
      new_exc = Exception(msg)
      raise new_exc.__class__, new_exc, self.exc[2]

Python 3.x:

Die 3-Argument-Form für raise ist in Python 3 nicht mehr vorhanden. Ändern Sie daher die letzte Zeile in:

raise new_exc.with_traceback(self.exc[2])

Auf diese Frage gibt es viele wirklich komische Antworten. Bin ich zu sehr vereinfacht, weil mir dies für die meisten Dinge ausreichend erscheint.

from threading import Thread

class PropagatingThread(Thread):
    def run(self):
        self.exc = None
        try:
            if hasattr(self, '_Thread__target'):
                # Thread uses name mangling prior to Python 3.
                self.ret = self._Thread__target(*self._Thread__args, **self._Thread__kwargs)
            else:
                self.ret = self._target(*self._args, **self._kwargs)
        except BaseException as e:
            self.exc = e

    def join(self):
        super(PropagatingThread, self).join()
        if self.exc:
            raise self.exc
        return self.ret

Wenn Sie sicher sind, dass Sie immer nur mit der einen oder der anderen Version von Python arbeiten, können Sie die run()-Methode auf die entstellte Version reduzieren (wenn Sie nur mit Python-Versionen 3 arbeiten). oder nur die saubere Version (wenn Sie nur mit Python-Versionen ab 3 laufen).

Verwendungsbeispiel:

def f(*args, **kwargs)
    print(args)
    print(kwargs)
    raise Exception('I suck')

t = PropagatingThread(target=f, args=(5,), kwargs={'hello':'world'})
t.start()
t.join()

Und Sie werden sehen, dass die Ausnahme beim Beitritt im anderen Thread ausgelöst wird.

Dies war ein unangenehmes kleines Problem, und ich würde gerne meine Lösung einwerfen. Einige andere Lösungen, die ich gefunden habe (async.io zum Beispiel), sahen vielversprechend aus, präsentierten jedoch auch eine Art Blackbox. Die Warteschlangen-/Ereignisschleifenmethode verbindet Sie mit einer bestimmten Implementierung. Der Concurrent-Futures-Quellcode besteht jedoch aus nur 1000 Zeilen und ist leicht zu verstehen . So konnte ich mein Problem leicht lösen: Ad-hoc-Worker-Threads ohne viel Setup erstellen und Ausnahmen im Haupt-Thread abfangen können.

Meine Lösung verwendet die API für gleichzeitige Futures und die Threading-API. Damit können Sie einen Worker erstellen, der Ihnen sowohl den Thread als auch die Zukunft gibt. Auf diese Weise können Sie dem Thread beitreten, um auf das Ergebnis zu warten:

worker = Worker(test)
thread = worker.start()
thread.join()
print(worker.future.result())

... oder Sie können den Arbeiter einfach einen Rückruf senden lassen, wenn Sie fertig sind:

worker = Worker(test)
thread = worker.start(lambda x: print('callback', x))

... oder Sie können eine Schleife machen, bis das Ereignis abgeschlossen ist:

worker = Worker(test)
thread = worker.start()

while True:
    print("waiting")
    if worker.future.done():
        exc = worker.future.exception()
        print('exception?', exc)
        result = worker.future.result()
        print('result', result)           
        break
    time.sleep(0.25)

Hier ist der Code:

from concurrent.futures import Future
import threading
import time

class Worker(object):
    def __init__(self, fn, args=()):
        self.future = Future()
        self._fn = fn
        self._args = args

    def start(self, cb=None):
        self._cb = cb
        self.future.set_running_or_notify_cancel()
        thread = threading.Thread(target=self.run, args=())
        thread.daemon = True #this will continue thread execution after the main thread runs out of code - you can still ctrl + c or kill the process
        thread.start()
        return thread

    def run(self):
        try:
            self.future.set_result(self._fn(*self._args))
        except BaseException as e:
            self.future.set_exception(e)

        if(self._cb):
            self._cb(self.future.result())

... und die Testfunktion:

def test(*args):
    print('args are', args)
    time.sleep(2)
    raise Exception('foo')

concurrent.futures.as_completed

https://docs.python.org/3.7/library/concurrent.futures.html#concurrent.futures.as_completed

Die folgende Lösung:

• kehrt sofort zum Hauptthread zurück, wenn eine Ausnahme aufgerufen wird
• erfordert keine zusätzlichen benutzerdefinierten Klassen, da Folgendes nicht benötigt wird:
  • eine explizite Queue
  • um eine Ausnahme zu Ihrem Arbeitsthread hinzuzufügen 

Quelle:

#!/usr/bin/env python3

import concurrent.futures
import time

def func_that_raises(do_raise):
    for i in range(3):
        print(i)
        time.sleep(0.1)
    if do_raise:
        raise Exception()
    for i in range(3):
        print(i)
        time.sleep(0.1)

with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
    futures = []
    futures.append(executor.submit(func_that_raises, False))
    futures.append(executor.submit(func_that_raises, True))
    for future in concurrent.futures.as_completed(futures):
        print(repr(future.exception()))

Mögliche Ausgabe:

0
0
1
1
2
2
0
Exception()
1
2
None

Es ist leider nicht möglich, Futures zu töten, um die anderen zu kündigen, da einer ausfällt:

• concurrent.features; Python: concurrent.futures Wie kann ich es abbrechen?
• threading: Gibt es eine Möglichkeit, einen Thread zu töten?
• C pthreads: Kill Thread in der Pthread Library

Wenn du so etwas machst:

for future in concurrent.futures.as_completed(futures):
    if future.exception() is not None:
        raise future.exception()

dann fängt die Variable with und wartet, bis der zweite Thread beendet ist, bevor Sie fortfahren. Folgendes verhält sich ähnlich:

for future in concurrent.futures.as_completed(futures):
    future.result()

da future.result() die Ausnahme erneut auslöst, falls eine aufgetreten ist.

Wenn Sie den gesamten Python-Prozess beenden möchten, kommen Sie möglicherweise mit os._exit(0) aus. Dies bedeutet jedoch wahrscheinlich, dass Sie einen Refactor benötigen.

Getestet auf Python 3.6.7, Ubuntu 18.04.

Als Noobie für Threading habe ich lange gebraucht, um zu verstehen, wie der Code von Mateusz Kobos (oben) implementiert wird. Hier ist eine klarere Version, um zu verstehen, wie man sie benutzt. 

#!/usr/bin/env python

import sys
import threading
import Queue

class ExThread(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)
        self.__status_queue = Queue.Queue()

    def run_with_exception(self):
        """This method should be overriden."""
        raise NotImplementedError

    def run(self):
        """This method should NOT be overriden."""
        try:
            self.run_with_exception()
        except Exception:
            self.__status_queue.put(sys.exc_info())
        self.__status_queue.put(None)

    def wait_for_exc_info(self):
        return self.__status_queue.get()

    def join_with_exception(self):
        ex_info = self.wait_for_exc_info()
        if ex_info is None:
            return
        else:
            raise ex_info[1]

class MyException(Exception):
    pass

class MyThread(ExThread):
    def __init__(self):
        ExThread.__init__(self)

    # This overrides the "run_with_exception" from class "ExThread"
    # Note, this is where the actual thread to be run lives. The thread
    # to be run could also call a method or be passed in as an object
    def run_with_exception(self):
        # Code will function until the int
        print "sleeping 5 seconds"
        import time
        for i in 1, 2, 3, 4, 5:
            print i
            time.sleep(1) 
        # Thread should break here
        int("str")
# I'm honestly not sure why these appear here? So, I removed them. 
# Perhaps Mateusz can clarify?        
#         thread_name = threading.current_thread().name
#         raise MyException("An error in thread '{}'.".format(thread_name))

if __== '__main__':
    # The code lives in MyThread in this example. So creating the MyThread 
    # object set the code to be run (but does not start it yet)
    t = MyThread()
    # This actually starts the thread
    t.start()
    print
    print ("Notice 't.start()' is considered to have completed, although" 
           " the countdown continues in its new thread. So you code "
           "can tinue into new processing.")
    # Now that the thread is running, the join allows for monitoring of it
    try:
        t.join_with_exception()
    # should be able to be replace "Exception" with specific error (untested)
    except Exception, e: 
        print
        print "Exceptioon was caught and control passed back to the main thread"
        print "Do some handling here...or raise a custom exception "
        thread_name = threading.current_thread().name
        e = ("Caught a MyException in thread: '" + 
             str(thread_name) + 
             "' [" + str(e) + "]")
        raise Exception(e) # Or custom class of exception, such as MyException

Ähnlich wie bei RickardSjogrens ohne Queue, sys etc., aber auch ohne einige Zuhörer, die Signale signalisieren: Führen Sie direkt einen Exception-Handler aus, der einem Except-Block entspricht.

#!/usr/bin/env python3

import threading

class ExceptionThread(threading.Thread):

    def __init__(self, callback=None, *args, **kwargs):
        """
        Redirect exceptions of thread to an exception handler.

        :param callback: function to handle occured exception
        :type callback: function(thread, exception)
        :param args: arguments for threading.Thread()
        :type args: Tuple
        :param kwargs: keyword arguments for threading.Thread()
        :type kwargs: dict
        """
        self._callback = callback
        super().__init__(*args, **kwargs)

    def run(self):
        try:
            if self._target:
                self._target(*self._args, **self._kwargs)
        except BaseException as e:
            if self._callback is None:
                raise e
            else:
                self._callback(self, e)
        finally:
            # Avoid a refcycle if the thread is running a function with
            # an argument that has a member that points to the thread.
            del self._target, self._args, self._kwargs, self._callback

Nur self._callback und der Exceptionsblock in run () ergänzen das normale threading.Thread.

Die Verwendung nackter Ausnahmen ist keine gute Praxis, da Sie normalerweise mehr fangen, als Sie erwarten.

Ich würde vorschlagen, die except so zu ändern, dass NUR die Ausnahme erfasst wird, die Sie behandeln möchten. Ich denke nicht, dass das Erhöhen die gewünschte Wirkung hat, denn wenn Sie TheThread in der äußeren try instanziieren, wird eine Zuweisung nie ausgelöst, wenn eine Ausnahme ausgelöst wird.

Stattdessen möchten Sie vielleicht einfach darauf aufmerksam machen und weitermachen, z.

def run(self):
    try:
       shul.copytree(self.sourceFolder, self.destFolder)
    except OSError, err:
       print err

Wenn diese Ausnahme dann abgefangen wird, können Sie sie dort behandeln. Wenn die äußere Variable try eine Ausnahme von TheThread feststellt, wissen Sie, dass es sich nicht um die bereits behandelte handelt, und hilft Ihnen, Ihren Prozessablauf zu isolieren.

Eine Methode, die ich sehr mag, basiert auf dem Beobachtermuster . Ich definiere eine Signalklasse, die mein Thread verwendet, um Ausnahmen für Zuhörer auszugeben. Es kann auch verwendet werden, um Werte aus Threads zurückzugeben. Beispiel:

import threading

class Signal:
    def __init__(self):
        self._subscribers = list()

    def emit(self, *args, **kwargs):
        for func in self._subscribers:
            func(*args, **kwargs)

    def connect(self, func):
        self._subscribers.append(func)

    def disconnect(self, func):
        try:
            self._subscribers.remove(func)
        except ValueError:
            raise ValueError('Function {0} not removed from {1}'.format(func, self))


class WorkerThread(threading.Thread):

    def __init__(self, *args, **kwargs):
        super(WorkerThread, self).__init__(*args, **kwargs)
        self.Exception = Signal()
        self.Result = Signal()

    def run(self):
        if self._Thread__target is not None:
            try:
                self._return_value = self._Thread__target(*self._Thread__args, **self._Thread__kwargs)
            except Exception as e:
                self.Exception.emit(e)
            else:
                self.Result.emit(self._return_value)

if __== '__main__':
    import time

    def handle_exception(exc):
        print exc.message

    def handle_result(res):
        print res

    def a():
        time.sleep(1)
        raise IOError('a failed')

    def b():
        time.sleep(2)
        return 'b returns'

    t = WorkerThread(target=a)
    t2 = WorkerThread(target=b)
    t.Exception.connect(handle_exception)
    t2.Result.connect(handle_result)
    t.start()
    t2.start()

    print 'Threads started'

    t.join()
    t2.join()
    print 'Done'

Ich habe nicht genug Erfahrung in der Arbeit mit Threads, um zu behaupten, dass dies eine absolut sichere Methode ist. Aber es hat für mich funktioniert und ich mag die Flexibilität.

Ich weiß, dass ich zu der Party etwas spät komme, aber ich hatte ein sehr ähnliches Problem, aber es beinhaltete die Verwendung von tkinter als GUI, und der Mainloop machte es unmöglich, eine der .join () - Lösungen zu verwenden. Deshalb habe ich die in der EDIT angegebene Lösung der ursprünglichen Frage angepasst, aber es allgemeiner gemacht, um es für andere verständlicher zu machen.

Hier ist die neue Thread-Klasse in Aktion:

import threading
import traceback
import logging


class ExceptionThread(threading.Thread):
    def __init__(self, *args, **kwargs):
        threading.Thread.__init__(self, *args, **kwargs)

    def run(self):
        try:
            if self._target:
                self._target(*self._args, **self._kwargs)
        except Exception:
            logging.error(traceback.format_exc())


def test_function_1(input):
    raise IndexError(input)


if __== "__main__":
    input = 'useful'

    t1 = ExceptionThread(target=test_function_1, args=[input])
    t1.start()

Natürlich können Sie die Ausnahme auch auf andere Weise von der Protokollierung abwickeln lassen, z. B. Ausdrucken oder Ausgabe auf der Konsole.

Auf diese Weise können Sie die ExceptionThread-Klasse genau wie die Thread-Klasse verwenden, ohne dass spezielle Änderungen erforderlich sind. 

Wrap-Thread mit Ausnahmespeicher.

import threading
import sys
class ExcThread(threading.Thread):

    def __init__(self, target, args = None):
        self.args = args if args else []
        self.target = target
        self.exc = None
        threading.Thread.__init__(self)

    def run(self):
        try:
            self.target(*self.args)
            raise Exception('An error occured here.')
        except Exception:
            self.exc=sys.exc_info()

def main():
    def hello(name):
        print(!"Hello, {name}!")
    thread_obj = ExcThread(target=hello, args=("Jack"))
    thread_obj.start()

    thread_obj.join()
    exc = thread_obj.exc
    if exc:
        exc_type, exc_obj, exc_trace = exc
        print(exc_type, ':',exc_obj, ":", exc_trace)

main()

Eine einfache Möglichkeit, die Ausnahme von Thread zu erkennen und zur aufrufenden Methode zurückzukehren, besteht darin, das Wörterbuch oder eine Liste an die worker-Methode zu übergeben.

Beispiel (Wörterbuch an die Workermethode übergeben): 

import threading

def my_method(throw_me):
    raise Exception(throw_me)

def worker(shared_obj, *args, **kwargs):
    try:
        shared_obj['target'](*args, **kwargs)
    except Exception as err:
        shared_obj['err'] = err

shared_obj = {'err':'', 'target': my_method}
throw_me = "Test"

th = threading.Thread(target=worker, args=(shared_obj, throw_me), kwargs={})
th.start()
th.join()

if shared_obj['err']:
    print(">>%s" % shared_obj['err'])