Cause python to exit if any thread has an exception

Question

I have a python3 program that starts a second thread (besides the main thread) for handling some events asynchronously. Ideally, my program works without a flaw and never has an unhandled exceptions. But stuff happens. When/if there is an exception, I want the whole interpreter to exit with an error code as if it had been a single thread. Is that possible?

Right now, if an exception occurs on the spawned thread, it prints out the usual error information, but doesn't exit. The main thread just keeps going.

Example

import threading
import time

def countdown(initial):
    while True:
        print(initial[0])
        initial = initial[1:]
        time.sleep(1)

if __name__ == '__main__':
    helper = threading.Thread(target=countdown, args=['failsoon'])
    helper.start()
    time.sleep(0.5)
    #countdown('THISWILLTAKELONGERTOFAILBECAUSEITSMOREDATA')
    countdown('FAST')

The countdown will eventually fail to access [0] from the string because it's been emptied causing an IndexError: string index out of range error. The goal is that whether the main or helper dies first, the whole program dies alltogether, but the stack trace info is still output.

Solutions Tried

After some digging, my thought was to use sys.excepthook. I added the following:

def killAll(etype, value, tb):
    print('KILL ALL')
    traceback.print_exception(etype, value, tb)
    os.kill(os.getpid(), signal.SIGKILL)

sys.excepthook = killAll

This works if the main thread is the one that dies first. But in the other case it does not. This seems to be a known issue (https://bugs.python.org/issue1230540). I will try some of the workarounds there.

While the example shows a main thread and a helper thread which I created, I'm interested in the general case where I may be running someone else's library that launches a thread.

Answer 1

Well, you could simply raise an error in your thread and have the main thread handle and report that error. From there you could even terminate the program.

For example on your worker thread:

try:
    self.result = self.do_something_dangerous()
except Exception as e:
    import sys
    self.exc_info = sys.exc_info()

and on main thread:

if self.exc_info:
    raise self.exc_info[1].with_traceback(self.exc_info[2])
return self.result

So to give you a more complete picture, your code might look like this:

import threading

class ExcThread(threading.Thread):
    def excRun(self):
        pass
        #Where your core program will run

    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 
            # DON'T rethrow,
            # just complete the function such as storing
            # variables or states as needed

    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.with_traceback(self.exc[2])

(I added an extra line to keep track of which thread is causing the error in case you have multiple threads, it's also good practice to name them)

Answer 2

My solution ended up being a happy marriage between the solution posted here and the SIGKILL solution piece from above. I added the following killall.py submodule to my package:

import threading
import sys
import traceback
import os
import signal


def sendKillSignal(etype, value, tb):
    print('KILL ALL')
    traceback.print_exception(etype, value, tb)
    os.kill(os.getpid(), signal.SIGKILL)


original_init = threading.Thread.__init__
def patched_init(self, *args, **kwargs):
    print("thread init'ed")
    original_init(self, *args, **kwargs)
    original_run = self.run
    def patched_run(*args, **kw):
        try:
            original_run(*args, **kw)
        except:
            sys.excepthook(*sys.exc_info())
    self.run = patched_run


def install():
    sys.excepthook = sendKillSignal
    threading.Thread.__init__ = patched_init

And then ran the install right away before any other threads are launched (of my own creation or from other imported libraries).

Answer 3

I'm a bit late to this party, but here's the solution I came up with by combining these two answers and a little but of my own magic.

"""
Special killer thread
"""
import threading


class NuclearThead(threading.Thread):
"""
Thread that kills the entire program if unhandled
"""
def __init__(self, *args, **kwargs):
    super().__init__(*args, **kwargs)
    self.__exception: Exception | None = None

def run(self):
    try:
        super().run()
    except Exception as exc:
        self.__exception = exc

def join(self, *args, **kwargs):
    super().join(*args, **kwargs)
    if self.__exception:
        raise ThreadException from self.__exception


class ThreadException(Exception):
    """
    General exception raised by NuclearThread
    """

Answer 4

Just wanted to share my simple solution.

In my case I wanted the exception to display as normal but then immediately stop the program. I was able to accomplish this by starting a timer thread with a small delay to call os._exit before raising the exception.

import os
import threading

def raise_and_exit(args):
    threading.Timer(0.01, os._exit, args=(1,)).start()
    raise args[0]

threading.excepthook = raise_and_exit

Answer 5

Python 3.8 added threading.excepthook which makes it possible to handle this more cleanly.

I wrote the package "unhandled_exit" to do just that. It basically adds os._exit(1) to after the default handler. This means you get the normal backtrace before the process exits.

Package is published to pypi here: https://pypi.org/project/unhandled_exit/
Code is here: https://github.com/rfjakob/unhandled_exit/blob/master/unhandled_exit/\_\_init__.py

Usage is simply:

import unhandled_exit
unhandled_exit.activate()