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>I think, I'll implement it like that:</p>

<p>I the main process, I create 2 queues:</p>

<ol>
<li>One to report exceptions,</li>
<li>One to notify cancellation.</li>
</ol>

<p>::</p>

<pre><code>import multiprocessing as mp

error_queue = mp.Queue()
cancel_queue = mp.Queue()
</code></pre>

<p>I create each <code>ThreadPoolExecutor</code>, and pass these queues as parameters.</p>

<pre><code>class MyExecutor(concurrent.futures.ThreadPoolExecutor):
    def __init__(self, error_queue, cancel_queue):
        self.error_queue : error_queue
        self.cancel_queue = cancel_queue
</code></pre>

<p>Each <code>ThreadPoolExecutor</code> has a main loop. In this loop, I first scan the <code>cancel_queue</code> to see if a "cancel" message is available.</p>

<p>In the main loop, I also implement an exception manager. And if an erreur occurs, I raise an exception:</p>

<pre><code>self.status = "running"
with True:  # &lt;- or something else
    if not self.cancel_queue.empty():
        self.status = "cancelled"
        break
    try:
        # normal processing
        ...
    except Exception as exc:
        # you can log the exception here for debug
        self.error_queue.put(exc)
        self.status = "error"
        break
    time.sleep(.1)
</code></pre>

<p>In the main process:</p>

<p>Run all <code>MyExecutor</code> instance.</p>

<p>Scan the error_queue:</p>

<pre><code>while True:
    if not error_queue.empty():
        cancel_queue.put("cancel")
    time.sleep(.1)
</code></pre>

python multithreading python-3.x exception threadpoolexecutor