How to immediately re-raise an exception thrown in any of the worker threads?
<p>I'm using <code>ThreadPoolExecutor</code> and need to abort the whole computation in case any of the worker threads fails.</p>

<p><strong>Example 1.</strong> This prints <em>Success</em> regardless of the error, since <code>ThreadPoolExecutor</code> does not re-raise exceptions automatically.</p>

<pre><code>from concurrent.futures import ThreadPoolExecutor

def task():
    raise ValueError

with ThreadPoolExecutor() as executor:
    executor.submit(task)
print('Success')
</code></pre>

<p><strong>Example 2.</strong> This correctly crashes the main thread because <code>.result()</code> re-raises exceptions. But it waits for the first task to finish, so the main thread experiences the exception with a delay.</p>

<pre><code>import time
from concurrent.futures import ThreadPoolExecutor

def task(should_raise):
    time.sleep(1)
    if should_raise:
        raise ValueError

with ThreadPoolExecutor() as executor:
    executor.submit(task, False).result()
    executor.submit(task, True).result()
print('Success')
</code></pre>

<p>How can I notice a worker exception in the main thread (more or less) immediately after it occurred, to handle the failure and abort the remaining workers?</p>

<p>First of all, we have to submit the tasks before requesting their results. Otherwise, the threads are not even running in parallel:</p>

<pre><code>futures = []
with ThreadPoolExecutor() as executor:
    futures.append(executor.submit(good_task))
    futures.append(executor.submit(bad_task))
for future in futures:
    future.result()
</code></pre>

<p>Now we can store the exception information in a variable that is available to both the main thread and the worker threads:</p>

<pre><code>exc_info = None
</code></pre>

<p>The main thread cannot really kill its sub processes, so we have the workers check for the exception information to be set and stop:</p>

<pre><code>def good_task():
    global exc_info
    while not exc_info:
        time.sleep(0.1)

def bad_task():
    global exc_info
    time.sleep(0.2)
    try:
        raise ValueError()
    except Exception:
        exc_info = sys.exc_info()
</code></pre>

<p>After all threads terminated, the main thread can check the variable holding the exception information. If it's populated, we re-raise the exception:</p>

<pre><code>if exc_info:
    raise exc_info[0].with_traceback(exc_info[1], exc_info[2])
print('Success')
</code></pre>

python multithreading python-3.x exception threadpoolexecutor