Złap wyjątek wątku w wątku wywołującym w języku Python

Jestem bardzo nowy w Pythonie i ogólnie programowaniu wielowątkowym. Zasadniczo mam skrypt, który skopiuje pliki do innej lokalizacji. Chciałbym, aby został on umieszczony w innym wątku, dzięki czemu mogę wysyłać dane ....wskazujące, że skrypt nadal działa.

Problem, który mam, polega na tym, że jeśli nie można skopiować plików, wygeneruje wyjątek. Jest to OK, jeśli działa w głównym wątku; jednak posiadanie następującego kodu nie działa:

try:
    threadClass = TheThread(param1, param2, etc.)
    threadClass.start()   ##### **Exception takes place here**
except:
    print "Caught an exception"

W samej klasie wątków próbowałem ponownie zgłosić wyjątek, ale to nie działa. Widziałem ludzi tutaj zadających podobne pytania, ale wszyscy wydają się robić coś bardziej konkretnego niż to, co próbuję zrobić (i nie do końca rozumiem oferowane rozwiązania). Widziałem ludzi wspominających o użyciu sys.exc_info(), jednak nie wiem, gdzie i jak go używać.

Każda pomoc jest mile widziana!

EDYCJA: Kod dla klasy wątku znajduje się poniżej:

class TheThread(threading.Thread):
    def __init__(self, sourceFolder, destFolder):
        threading.Thread.__init__(self)
        self.sourceFolder = sourceFolder
        self.destFolder = destFolder

    def run(self):
        try:
           shul.copytree(self.sourceFolder, self.destFolder)
        except:
           raise

Problem polega na tym, że thread_obj.start()natychmiast wraca. Wątek potomny, który odrodziłeś, wykonuje we własnym kontekście, z własnym stosem. Każdy występujący wyjątek występuje w kontekście wątku podrzędnego i znajduje się na swoim stosie. Jednym ze sposobów, które mogę teraz wymyślić, aby przekazać te informacje do wątku nadrzędnego, jest użycie pewnego rodzaju przekazywania wiadomości, abyś mógł to sprawdzić.

Spróbuj tego dla rozmiaru:

import sys
import threading
import Queue


class ExcThread(threading.Thread):

    def __init__(self, bucket):
        threading.Thread.__init__(self)
        self.bucket = bucket

    def run(self):
        try:
            raise Exception('An error occured here.')
        except Exception:
            self.bucket.put(sys.exc_info())


def main():
    bucket = Queue.Queue()
    thread_obj = ExcThread(bucket)
    thread_obj.start()

    while True:
        try:
            exc = bucket.get(block=False)
        except Queue.Empty:
            pass
        else:
            exc_type, exc_obj, exc_trace = exc
            # deal with the exception
            print exc_type, exc_obj
            print exc_trace

        thread_obj.join(0.1)
        if thread_obj.isAlive():
            continue
        else:
            break


if __name__ == '__main__':
    main()

concurrent.futuresModuł ułatwia działają w odrębnych wątków (lub procesów) i obsługiwać wszelkie wynikające wyjątki:

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.futuresjest dołączony Pythonie 3.2 i jest dostępny na przeniesiona futuresmodułu do wcześniejszych wersji.

Istnieje wiele naprawdę dziwnie skomplikowanych odpowiedzi na to pytanie. Upraszczam to, ponieważ wydaje mi się to wystarczające dla większości rzeczy.

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

Jeśli masz pewność, że kiedykolwiek będziesz działał tylko na jednej lub drugiej wersji Pythona, możesz zredukować run()metodę do tylko zniekształconej wersji (jeśli będziesz działał tylko na wersjach Pythona przed 3), lub tylko czysta wersja (jeśli będziesz działał tylko na wersjach Python zaczynających się od 3).

Przykładowe użycie:

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

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

Po dołączeniu zobaczysz wyjątek zgłoszony w drugim wątku.

Jeśli używasz sixlub tylko w języku Python 3, możesz poprawić informacje śledzenia stosu, które otrzymujesz po ponownym zgłoszeniu wyjątku. Zamiast samego stosu w punkcie łączenia możesz zawinąć wyjątek wewnętrzny w nowy wyjątek zewnętrzny i uzyskać oba ślady stosu za pomocą

six.raise_from(RuntimeError('Exception in thread'),self.exc)

lub

raise RuntimeError('Exception in thread') from self.exc

Chociaż nie jest możliwe bezpośrednie złapanie wyjątku zgłoszonego w innym wątku, oto kod, aby dość transparentnie uzyskać coś bardzo zbliżonego do tej funkcjonalności. Wątek podrzędny musi podklasować ExThreadklasę zamiast, threading.Threada wątek nadrzędny musi wywoływać child_thread.join_with_exception()metodę zamiast child_thread.join()czekać na zakończenie wątku.

Szczegóły techniczne tej implementacji: gdy wątek potomny zgłasza wyjątek, jest przekazywany do rodzica przez a Queuei ponownie wrzucany do wątku rodzica. Zauważ, że w tym podejściu nie ma żadnych zajęć.

#!/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 __name__ == '__main__':
    main()

Jeśli wyjątek występuje w wątku, najlepszym sposobem jest ponowne podniesienie go w wątku wywołującym podczas join. Możesz uzyskać informacje o aktualnie obsługiwanym wyjątku za pomocą sys.exc_info()funkcji. Informacje te można po prostu przechowywać jako właściwość obiektu wątku do momentu joinwywołania, w którym to momencie można je ponownie wywołać.

Zauważ, że Queue.Queue(jak zasugerowano w innych odpowiedziach) nie jest konieczne w tym prostym przypadku, w którym wątek zgłasza maksymalnie 1 wyjątek i kończy się zaraz po zgłoszeniu wyjątku . Unikamy warunków wyścigu, po prostu czekając na zakończenie wątku.

Na przykład rozszerz ExcThread(poniżej), zastępując excRun(zamiast 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:

Formularz 3 argumentu dla raisezniknął w Pythonie 3, więc zmień ostatni wiersz na:

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

concurrent.futures.as_completed

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

Następujące rozwiązanie:

• wraca do głównego wątku natychmiast po wywołaniu wyjątku
• nie wymaga żadnych dodatkowych klas zdefiniowanych przez użytkownika, ponieważ nie potrzebuje:
  • wyraźne Queue
  • aby dodać wyjątek oprócz twojego wątku roboczego

Źródło:

#!/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()))

Możliwe wyjście:

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

Niestety nie jest możliwe zabijanie kontraktów futures, aby anulować pozostałe, ponieważ jedna zawiedzie:

• concurrent.features; Python: concurrent.futures Jak to zrobić?
• threading: Czy jest jakiś sposób na zabicie Wątku?
• C pthreads: Kill Thread in Pthread Library

Jeśli robisz coś takiego:

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

następnie withłapie go i czeka na zakończenie drugiego wątku przed kontynuowaniem. Poniższe zachowuje się podobnie:

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

ponieważ future.result()ponownie podnosi wyjątek, jeśli taki wystąpił.

Jeśli chcesz wyjść z całego procesu Pythona, możesz uciec os._exit(0), ale prawdopodobnie oznacza to, że potrzebujesz refaktora.

Klasa niestandardowa z doskonałą semantyką wyjątków

Skończyło się na tym, że napisałem dla siebie idealny interfejs: Właściwy sposób na ograniczenie maksymalnej liczby wątków działających jednocześnie? sekcja „Przykład kolejki z obsługą błędów”. Ta klasa ma być zarówno wygodna, jak i dać ci całkowitą kontrolę nad przesyłaniem i obsługą wyników / błędów.

Testowane na Python 3.6.7, Ubuntu 18.04.

To był bardzo paskudny mały problem i chciałbym wrzucić moje rozwiązanie. Niektóre inne rozwiązania, które znalazłem (na przykład async.io), wyglądały obiecująco, ale także prezentowały trochę czarną skrzynkę. Podejście do kolejki / zdarzenia wiąże się z pewną implementacją. Jednoczesny kod źródłowy futures ma jednak tylko około 1000 linii i jest łatwy do zrozumienia . Pozwoliło mi to łatwo rozwiązać problem: tworzyć wątki robocze ad-hoc bez dużej konfiguracji i móc wychwytywać wyjątki w głównym wątku.

Moje rozwiązanie wykorzystuje współbieżny interfejs futures API i wątkowy interfejs API. Pozwala ci stworzyć pracownika, który da ci zarówno wątek, jak i przyszłość. W ten sposób możesz dołączyć do wątku, aby czekać na wynik:

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

... lub możesz pozwolić pracownikowi wysłać oddzwonienie po zakończeniu:

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

... lub możesz zapętlać, aż wydarzenie się zakończy:

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)

Oto kod:

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())

... i funkcja testowa:

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

Jako noobie do wątków zajęło mi dużo czasu, aby zrozumieć, jak zaimplementować kod Mateusza Kobosa (powyżej). Oto wyjaśniona wersja, która pomoże zrozumieć, jak z niej korzystać.

#!/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 __name__ == '__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

Podobnie jak RickardSjogren bez kolejki, sys itp., Ale także bez niektórych detektorów sygnałów: bezpośrednio uruchom moduł obsługi wyjątków, który odpowiada blokowi wyjątków.

#!/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

Tylko self._callback i blok wyjątków w run () stanowią uzupełnienie normalnego wątku.

Wiem, że jestem trochę spóźniony na imprezę, ale miałem bardzo podobny problem, ale obejmował on użycie tkintera jako GUI, a mainloop uniemożliwił użycie dowolnego rozwiązania zależnego od .join (). Dlatego dostosowałem rozwiązanie podane w EDIT pierwotnego pytania, ale uczyniłem to bardziej ogólnym, aby ułatwić zrozumienie dla innych.

Oto nowa klasa wątków w akcji:

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 __name__ == "__main__":
    input = 'useful'

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

Oczywiście zawsze możesz sprawić, by obsługiwał wyjątek w inny sposób niż logowanie, na przykład wydrukowanie go lub wysłanie do konsoli.

Dzięki temu możesz używać klasy ExceptionThread dokładnie tak samo, jak klasy Thread, bez żadnych specjalnych modyfikacji.

Jedna metoda, którą lubię, opiera się na wzorcu obserwatora . Definiuję klasę sygnału, której mój wątek używa do wysyłania wyjątków dla słuchaczy. Można go również użyć do zwrócenia wartości z wątków. Przykład:

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 __name__ == '__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'

Nie mam wystarczającego doświadczenia w pracy z wątkami, aby stwierdzić, że jest to całkowicie bezpieczna metoda. Ale zadziałało dla mnie i podoba mi się elastyczność.

Używanie nagich wyjątków nie jest dobrą praktyką, ponieważ zwykle łapiesz więcej, niż się spodziewasz.

Sugerowałbym modyfikację, exceptaby ŁĄCZYĆ TYLKO wyjątek, który chcesz obsłużyć. Nie sądzę, aby podniesienie go miało pożądany efekt, ponieważ jeśli przejdziesz do tworzenia instancji TheThreadna zewnątrz try, jeśli pojawi się wyjątek, zadanie nigdy się nie wydarzy.

Zamiast tego możesz po prostu zaalarmować i przejść dalej, na przykład:

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

Następnie, gdy ten wyjątek zostanie złapany, możesz sobie z nim poradzić. Wtedy, gdy zewnętrzny trywychwytuje wyjątek TheThread, wiesz, że nie będzie to ten, z którym już sobie poradziłeś, i pomoże ci odizolować przebieg procesu.

Prostym sposobem wychwycenia wyjątku wątku i przekazania go z powrotem do metody wywołującej może być przekazanie słownika lub listy do workermetody.

Przykład (przekazanie słownika do metody roboczej):

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'])

Zawijaj wątek z przechowywaniem wyjątków.

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()

pygolang zapewnia sync.WorkGroup, która w szczególności propaguje wyjątek od spawnowanych wątków roboczych do głównego wątku. Na przykład:

#!/usr/bin/env python
"""This program demostrates how with sync.WorkGroup an exception raised in
spawned thread is propagated into main thread which spawned the worker."""

from __future__ import print_function
from golang import sync, context

def T1(ctx, *argv):
    print('T1: run ... %r' % (argv,))
    raise RuntimeError('T1: problem')

def T2(ctx):
    print('T2: ran ok')

def main():
    wg = sync.WorkGroup(context.background())
    wg.go(T1, [1,2,3])
    wg.go(T2)

    try:
        wg.wait()
    except Exception as e:
        print('Tmain: caught exception: %r\n' %e)
        # reraising to see full traceback
        raise

if __name__ == '__main__':
    main()

daje następujące po uruchomieniu:

T1: run ... ([1, 2, 3],)
T2: ran ok
Tmain: caught exception: RuntimeError('T1: problem',)

Traceback (most recent call last):
  File "./x.py", line 28, in <module>
    main()
  File "./x.py", line 21, in main
    wg.wait()
  File "golang/_sync.pyx", line 198, in golang._sync.PyWorkGroup.wait
    pyerr_reraise(pyerr)
  File "golang/_sync.pyx", line 178, in golang._sync.PyWorkGroup.go.pyrunf
    f(pywg._pyctx, *argv, **kw)
  File "./x.py", line 10, in T1
    raise RuntimeError('T1: problem')
RuntimeError: T1: problem

Oryginalny kod z pytania byłby po prostu:

    wg = sync.WorkGroup(context.background())

    def _(ctx):
        shul.copytree(sourceFolder, destFolder)
    wg.go(_)

    # waits for spawned worker to complete and, on error, reraises
    # its exception on the main thread.
    wg.wait()