Captura la excepción de un hilo en el hilo del llamador en Python

Soy muy nuevo en Python y en la programación multiproceso en general. Básicamente, tengo un script que copiará archivos a otra ubicación. Me gustaría que esto se coloque en otro hilo para que pueda mostrar ....que el script todavía se está ejecutando.

El problema que tengo es que si los archivos no se pueden copiar, arrojará una excepción. Esto está bien si se ejecuta en el hilo principal; sin embargo, tener el siguiente código no funciona:

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

En la clase de hilo en sí, intenté volver a lanzar la excepción, pero no funciona. He visto a personas aquí hacer preguntas similares, pero todos parecen estar haciendo algo más específico de lo que estoy tratando de hacer (y no entiendo muy bien las soluciones ofrecidas). He visto a personas mencionar el uso de sys.exc_info(), sin embargo, no sé dónde ni cómo usarlo.

¡Toda ayuda es muy apreciada!

EDITAR: El código para la clase de subproceso está a continuación:

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

El problema es ese thread_obj.start() vuelve de inmediato. El subproceso hijo que generó se ejecuta en su propio contexto, con su propia pila. Cualquier excepción que ocurra allí está en el contexto del subproceso secundario y está en su propia pila. Una forma en la que puedo pensar en este momento para comunicar esta información al hilo primario es mediante el uso de algún tipo de paso de mensajes, por lo que podría investigar eso.

Probar esto para el tamaño:

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

El concurrent.futuresmódulo simplifica el trabajo en subprocesos (o procesos) separados y maneja las excepciones resultantes:

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.futuresse incluye con Python 3.2, y está disponible como módulo con respaldofutures para versiones anteriores.

Hay muchas respuestas realmente extrañas a esta pregunta. ¿Estoy simplificando demasiado esto, porque esto me parece suficiente para la mayoría de las cosas?

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

Si está seguro de que solo se ejecutará en una u otra versión de Python, puede reducir el run()método a la versión destrozada (si solo se ejecutará en versiones de Python anteriores a la 3), o solo la versión limpia (si solo se ejecutará en versiones de Python que comiencen con 3).

Ejemplo de uso:

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

Y verá la excepción planteada en el otro hilo cuando se una.

Si está utilizando sixo solo en Python 3, puede mejorar la información de seguimiento de la pila que obtiene cuando se vuelve a generar la excepción. En lugar de solo la pila en el punto de la unión, puede ajustar la excepción interna en una nueva excepción externa y obtener ambas trazas de pila con

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

o

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

Aunque no es posible detectar directamente una excepción lanzada en un hilo diferente, aquí hay un código para obtener de manera bastante transparente algo muy cercano a esta funcionalidad. Su subproceso secundario debe subclasificar la ExThreadclase en lugar de threading.Thready el subproceso primario debe llamar al child_thread.join_with_exception()método en lugar de child_thread.join()cuando espera que el subproceso termine su trabajo.

Detalles técnicos de esta implementación: cuando el subproceso secundario genera una excepción, se pasa al primario a través de Queueay se vuelve a generar en el subproceso principal. Tenga en cuenta que no hay ocupados esperando en este enfoque.

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

Si ocurre una excepción en un hilo, la mejor manera es volver a subirlo en el hilo del llamante durante join. Puede obtener información sobre la excepción que se está manejando actualmente utilizando la sys.exc_info()función. Esta información simplemente puede almacenarse como una propiedad del objeto de subproceso hastajoin se llame, momento en el que se puede volver a generar.

Tenga en cuenta que a Queue.Queue(como se sugiere en otras respuestas) no es necesario en este caso simple donde el hilo arroja como máximo 1 excepción y se completa justo después de arrojar una excepción . Evitamos las condiciones de carrera simplemente esperando que se complete el hilo.

Por ejemplo, extienda ExcThread(abajo), anulando excRun(en lugar de 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:

El formulario de 3 argumentos para raisedesapareció en Python 3, así que cambie la última línea a:

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

La siguiente solución:

• vuelve al hilo principal inmediatamente cuando se llama a una excepción
• no requiere clases definidas por el usuario adicionales porque no necesita:
  • un explícito Queue
  • para agregar una excepción más alrededor de su hilo de trabajo

Fuente:

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

Salida posible:

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

Desafortunadamente, no es posible matar futuros para cancelar los otros ya que uno falla:

• concurrent.features; Python: concurrent.futures ¿Cómo hacer que sea cancelable?
• threading: ¿Hay alguna forma de matar un hilo?
• Subprocesos C: matar subprocesos en la biblioteca Pthread

Si haces algo como:

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

luego lo withatrapa y espera a que termine el segundo hilo antes de continuar. Lo siguiente se comporta de manera similar:

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

ya que future.result()vuelve a plantear la excepción si se produjo una.

Si desea salir de todo el proceso de Python, puede salirse con la suya os._exit(0) , pero esto probablemente significa que necesita un refactorizador.

Clase personalizada con semántica de excepción perfecta

Terminé codificando la interfaz perfecta para mí en: ¿ La forma correcta de limitar el número máximo de hilos que se ejecutan a la vez? sección "Ejemplo de cola con manejo de errores". Esa clase pretende ser conveniente y brindarle un control total sobre el envío y el manejo de resultados / errores.

Probado en Python 3.6.7, Ubuntu 18.04.

Este fue un pequeño problema desagradable, y me gustaría incluir mi solución. Algunas otras soluciones que encontré (por ejemplo, async.io) parecían prometedoras pero también presentaban un cuadro negro. El enfoque de bucle de cola / evento lo vincula a una determinada implementación. Sin embargo, el código fuente de futuros concurrentes es de alrededor de 1000 líneas y es fácil de comprender. . Me permitió resolver fácilmente mi problema: crear hilos de trabajo ad-hoc sin mucha configuración, y poder detectar excepciones en el hilo principal.

Mi solución utiliza la API de futuros concurrentes y la API de subprocesos. Le permite crear un trabajador que le proporciona tanto el hilo como el futuro. De esa manera, puedes unirte al hilo para esperar el resultado:

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

... o puede dejar que el trabajador solo envíe una devolución de llamada cuando haya terminado:

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

... o puedes recorrer hasta que se complete el evento:

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)

Aquí está el código:

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

... y la función de prueba:

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

Como novato en Threading, me llevó mucho tiempo entender cómo implementar el código de Mateusz Kobos (arriba). Aquí hay una versión aclarada para ayudar a entender cómo usarla.

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

De manera similar a RickardSjogren's sin Queue, sys, etc. pero también sin algunos oyentes a las señales: ejecute directamente un controlador de excepción que corresponda a un bloque except.

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

Solo self._callback y el bloque except en run () es adicional al subproceso normal.

Sé que llegué un poco tarde a la fiesta aquí, pero estaba teniendo un problema muy similar, pero incluía el uso de tkinter como GUI, y el mainloop hizo imposible usar cualquiera de las soluciones que dependen de .join (). Por lo tanto, adapté la solución dada en el EDIT de la pregunta original, pero la hice más general para que sea más fácil de entender para los demás.

Aquí está la nueva clase de hilo en acción:

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

Por supuesto, siempre puede hacer que maneje la excepción de alguna otra manera desde el inicio de sesión, como imprimirlo o hacer que se envíe a la consola.

Esto le permite usar la clase ExceptionThread exactamente como lo haría con la clase Thread, sin ninguna modificación especial.

Un método que me gusta se basa en el patrón de observación . Defino una clase de señal que mi hilo utiliza para emitir excepciones a los oyentes. También se puede usar para devolver valores de subprocesos. Ejemplo:

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'

No tengo suficiente experiencia trabajando con hilos para afirmar que este es un método completamente seguro. Pero me ha funcionado y me gusta la flexibilidad.

El uso de excepciones desnudas no es una buena práctica, ya que generalmente captura más de lo que negocia.

Sugeriría modificar el exceptpara capturar SOLO la excepción que le gustaría manejar. No creo que elevarlo tenga el efecto deseado, porque cuando vas a crear una instancia TheThreaden el exterior try, si genera una excepción, la asignación nunca va a suceder.

En su lugar, es posible que desee alertarlo y seguir adelante, como:

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

Luego, cuando se detecta esa excepción, puede manejarla allí. Luego, cuando el exterior trydetecta una excepción TheThread, sabe que no será el que ya manejó y lo ayudará a aislar el flujo de su proceso.

Una forma sencilla de detectar la excepción del hilo y comunicarse con el método de la persona que llama podría ser pasar un diccionario o una lista a worker método.

Ejemplo (pasar del diccionario al método de trabajo):

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

Hilo de envoltura con almacenamiento de excepción.

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 proporciona sync.WorkGroup que, en particular, propaga la excepción de los hilos de trabajo generados al hilo principal. Por ejemplo:

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

da lo siguiente cuando se ejecuta:

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

El código original de la pregunta sería simplemente:

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