concurrent.futures.ProcessPoolExecutor raises during shutdown

[apparebit]

Bug report

Bug description:

from concurrent.futures import ProcessPoolExecutor as Pool

def nope():
    pass

def mk_done(pool):
    def done(_fut):
        pool.shutdown(wait=True)
    return done

if __name__ == "__main__":
    pool = Pool()
    future = pool.submit(nope)
    future.add_done_callback(mk_done(pool))

ProcessPoolExecutor invokes future done callbacks from its manager thread. When one of those future callbacks invokes ProcessPoolExecutor.shutdown(), shut down fails with a RuntimeError: cannot join current thread because the implementation joins the manager thread without protecting against self-joins.

The relevant lines Lib/concurrent/futures/process.py:845-846 currently read:

if self._executor_manager_thread is not None and wait:
            self._executor_manager_thread.join()

but should probably be updated to:

t = self._executor_manager_thread
if t is not None and wait and threading.current_thread() != t:
            self._executor_manager_thread.join()

Invoking shutdown(False) avoids the problem but also makes it impossible to wait for pool completion. A more general solution would be to deprecate the wait parameter for shutdown, never waiting in that method, and add a separate method, say wait_for_shutdown, that uses a threading.Event to wake any waiting threads. Alas, correctly setting that Event is going to be a bit involved.

While I tested on 3.12, the bug is also present in 3.13 and current.

CPython versions tested on:

3.12

Operating systems tested on:

macOS

[csm10495]

The first time I looked at this issue, I thought that doing the current thread check and skipping the join could lead to an interpreter hang since the thread would be stuck waiting for join.

After thinking about it a bit more: Doing something similar to what you proposed (I think) would be fine since _threads_wakeups still has the reference to the thread if not garbage collected at interpreter shutdown, which would then have .join() called. I would probably try/except/pass the RuntimeError since that allows us to skip fetching the current thread and accomplish the same behavior.

It's still kind of odd though: a callback stops the executor, but others may try to submit still. I guess they'll just get RuntimeError since the pool would be shutdown at some point.

An alternative to changing the stdlib could be doing something like this instead:

from concurrent.futures import ProcessPoolExecutor as Pool, wait
from functools import partial

def nope():
    pass

def mk_done(_future_not_used, pool, all_futures):
    # No more tasks allowed to be submitted
    pool.shutdown(wait=False)

    # Wait for all current tasks to finish
    wait(all_futures)

if __name__ == "__main__":
    pool = Pool()
    future = pool.submit(nope)

    # if you have more futures, have them in the all_futures set.
    future.add_done_callback(partial(mk_done, pool=pool, all_futures={future}))

    # if you wait a bit and the done callback was called: calls to pool.submit() would raise a RuntimeError since the pool is shutdown.

That's not too complicated, but I can see the appeal of allowing the shutdown(wait=True) call to be made in the callback like you initially tried.

@picnixz have any thoughts on this? I'm willing to do a PR if we want to allow the call to not raise the RuntimeError in this particular case.

[apparebit]

Assuming that exception frames are cheap in Python, I like the idea of wrapping the call in a try/except.

Since I filed the bug report, I discovered alexwlchan/concurrently and adapted the source code to my own needs in apparebit/shantay. It seems less fiddly to me and nicely avoids the threading issue in this bug report. Just adding to the menu of options 🧐