Revert "gh-128041: Add `terminate_workers` and `kill_workers` methods to ProcessPoolExecutor (GH-128043)" by colesbury · Pull Request #130838 · python/cpython · GitHub
Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
24 changes: 0 additions & 24 deletions Doc/library/concurrent.futures.rst
5 changes: 0 additions & 5 deletions Doc/whatsnew/3.14.rst
Original file line number Diff line number Diff line change
Expand Up @@ -444,11 +444,6 @@ contextvars
* Support context manager protocol by :class:`contextvars.Token`.
(Contributed by Andrew Svetlov in :gh:`129889`.)

* Add :meth:`concurrent.futures.ProcessPoolExecutor.terminate_workers` and
:meth:`concurrent.futures.ProcessPoolExecutor.kill_workers` as
ways to terminate or kill all living worker processes in the given pool.
(Contributed by Charles Machalow in :gh:`128043`.)


ctypes
------
Expand Down
71 changes: 0 additions & 71 deletions Lib/concurrent/futures/process.py
Original file line number Diff line number Diff line change
Expand Up @@ -626,14 +626,6 @@ class BrokenProcessPool(_base.BrokenExecutor):
while a future was in the running state.
"""

_TERMINATE = "terminate"
_KILL = "kill"

_SHUTDOWN_CALLBACK_OPERATION = {
_TERMINATE,
_KILL
}


class ProcessPoolExecutor(_base.Executor):
def __init__(self, max_workers=None, mp_context=None,
Expand Down Expand Up @@ -863,66 +855,3 @@ def shutdown(self, wait=True, *, cancel_futures=False):
self._executor_manager_thread_wakeup = None

shutdown.__doc__ = _base.Executor.shutdown.__doc__

def _force_shutdown(self, operation):
"""Attempts to terminate or kill the executor's workers based off the
given operation. Iterates through all of the current processes and
performs the relevant task if the process is still alive.

After terminating workers, the pool will be in a broken state
and no longer usable (for instance, new tasks should not be
submitted).
"""
if operation not in _SHUTDOWN_CALLBACK_OPERATION:
raise ValueError(f"Unsupported operation: {operation!r}")

processes = {}
if self._processes:
processes = self._processes.copy()

# shutdown will invalidate ._processes, so we copy it right before
# calling. If we waited here, we would deadlock if a process decides not
# to exit.
self.shutdown(wait=False, cancel_futures=True)

if not processes:
return

for proc in processes.values():
try:
if not proc.is_alive():
continue
except ValueError:
# The process is already exited/closed out.
continue

try:
if operation == _TERMINATE:
proc.terminate()
elif operation == _KILL:
proc.kill()
except ProcessLookupError:
# The process just ended before our signal
continue

def terminate_workers(self):
"""Attempts to terminate the executor's workers.
Iterates through all of the current worker processes and terminates
each one that is still alive.

After terminating workers, the pool will be in a broken state
and no longer usable (for instance, new tasks should not be
submitted).
"""
return self._force_shutdown(operation=_TERMINATE)

def kill_workers(self):
"""Attempts to kill the executor's workers.
Iterates through all of the current worker processes and kills
each one that is still alive.

After killing workers, the pool will be in a broken state
and no longer usable (for instance, new tasks should not be
submitted).
"""
return self._force_shutdown(operation=_KILL)
97 changes: 0 additions & 97 deletions Lib/test/test_concurrent_futures/test_process_pool.py

This file was deleted.