Skip to content

bpo-36780: Add wait_at_exit to ThreadPoolExecutor.shutdown. #13250

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
hniksic wants to merge 1 commit into
base: main
Choose a base branch
from
Open

bpo-36780: Add wait_at_exit to ThreadPoolExecutor.shutdown. #13250

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 17 additions & 0 deletions Doc/library/concurrent.futures.rst
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -159,6 +159,23 @@ And::
.. versionchanged:: 3.7
Added the *initializer* and *initargs* arguments.

.. method:: shutdown(wait=True, wait_at_exit=True)

Like :meth:`Executor.shutdown`, with the additional option to
suppress waiting for running futures at Python exit.

Normally, this method waits for all submitted jobs to finish
before returning. Specifying ``wait=False`` makes the method
return immediately, but when the Python process is about to
exit, it will still wait for the remaining futures to complete.

Specifying ``False`` for *wait_at_exit* inhibits waiting at
process exit. It is typically used along with ``wait=False``,
so ``shutdown(wait=False, wait_at_exit=False)`` abandons the
executor and its futures. This is useful when the submitted
jobs are possibly stuck, and waiting for them would make the
process hang.


.. _threadpoolexecutor-example:

Expand Down
5 changes: 4 additions & 1 deletion Lib/concurrent/futures/thread.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -211,11 +211,14 @@ def _initializer_failed(self):
if work_item is not None:
work_item.future.set_exception(BrokenThreadPool(self._broken))

def shutdown(self, wait=True):
def shutdown(self, wait=True, wait_at_exit=True):
Copy link
Contributor

@brianquinlan brianquinlan May 14, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

shutdown is defined in Executor because Executor is the abstract superclass for both ThreadPoolExecutor and ProcessPoolExecutor. Unless there is a very strong reason not to, this method should work the same in both executors.

Copy link
Contributor Author

@hniksic hniksic May 14, 2019 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@brianquinlan This was intentional - I tested shutdown(wait=False) with ProcessPoolExecutor, and found that it raised exceptions and hanged the process at exit. (Not just hanged in the sense of waiting for the pending futures, but completely hanged, even when the futures exited.) So the new functionality is only available in and documented for ThreadPoolExecutor.

For example, when I run this script on Python 3.7:

import time, concurrent.futures

pool = concurrent.futures.ProcessPoolExecutor()

pool.submit(time.sleep, 5)
print(1)

pool.shutdown(wait=False)
print(2)

The expected behavior is for the program to print 1 and 2 and then to wait for 5 seconds before exiting. Instead, it prints 1 and 2, but hangs at exit with the following output:

$ python3.7 ~/Desktop/x
1
2
Error in atexit._run_exitfuncs:
Traceback (most recent call last):
  File "/usr/lib/python3.7/concurrent/futures/process.py", line 101, in _python_exit
    thread_wakeup.wakeup()
  File "/usr/lib/python3.7/concurrent/futures/process.py", line 89, in wakeup
    self._writer.send_bytes(b"")
  File "/usr/lib/python3.7/multiprocessing/connection.py", line 183, in send_bytes
    self._check_closed()
  File "/usr/lib/python3.7/multiprocessing/connection.py", line 136, in _check_closed
    raise OSError("handle is closed")
Exception in thread QueueManagerThread:
Traceback (most recent call last):
  File "/usr/lib/python3.7/threading.py", line 917, in _bootstrap_inner
    self.run()
  File "/usr/lib/python3.7/threading.py", line 865, in run
    self._target(*self._args, **self._kwargs)
  File "/usr/lib/python3.7/concurrent/futures/process.py", line 368, in _queue_management_worker
    thread_wakeup.clear()
  File "/usr/lib/python3.7/concurrent/futures/process.py", line 92, in clear
    while self._reader.poll():
  File "/usr/lib/python3.7/multiprocessing/connection.py", line 255, in poll
    self._check_closed()
  File "/usr/lib/python3.7/multiprocessing/connection.py", line 136, in _check_closed
    raise OSError("handle is closed")
OSError: handle is closed
OSError: handle is closed

Copy link
Contributor

@brianquinlan brianquinlan Nov 15, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I really wanted ThreadPoolExecutor and ProcessPoolExecutor to have the same API when I designed them.

Do you have any bandwidth to debug this? If not, I could take a look.

Copy link
Contributor Author

@hniksic hniksic Nov 15, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please do take a look if you can. I am not acquainted with the implementation of ProcessPoolExecutor, so it would take quite some time for me to trace what's going on.

It would of course be ideal if both classes supported the new flag.

Copy link
Contributor

@brianquinlan brianquinlan Jan 6, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I was playing with ProcessPoolExecutor and it seems like there are a bunch of problems that are triggered when pool.shutdown(wait=False) is used. I filed a bug for one issue: https://bugs.python.org/issue39205

Do you think that your PR could hold off until I have a chance to sort some of this out?

Copy link
Contributor Author

@hniksic hniksic Jan 6, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do you think that your PR could hold off until I have a chance to sort some of this out?

Sure, thanks for asking. We have a workaround, so it's no problem to wait for the proper solution. It's just that the workaround is so extremely ugly, involving monkey patch of a private method, that we'd definitely prefer the proper fix to land eventually.

with self._shutdown_lock:
self._shutdown = True
self._work_queue.put(None)
if wait:
for t in self._threads:
t.join()
if not wait_at_exit:
for t in self._threads:
_threads_queues.pop(t, None)
shutdown.__doc__ = _base.Executor.shutdown.__doc__