gh-128041: Add a terminate_workers method to ProcessPoolExecutor

Doc/library/concurrent.futures.rst

@@ -415,6 +415,19 @@
       require the *fork* start method for :class:`ProcessPoolExecutor` you must
       explicitly pass ``mp_context=multiprocessing.get_context("fork")``.
 
+   .. method:: terminate_workers(signal=signal.SIGTERM)
+
+      Attempt to terminate all living worker processes immediately by sending each
+      of them the given signal. If the signal is not specified, the default signal
+      :data:`signal.SIGTERM` is used.
+
+      After calling :meth:`ProcessPoolExecutor.terminate_workers`, the caller, should
+      no longer submit tasks to the executor. It is also recommended to still call
+      :meth:`ProcessPoolExecutor.shutdown` to ensure that all other resources
+      associated with the executor are freed.
+
+      .. versionadded:: 3.14
+
 .. _processpoolexecutor-example:
 
 ProcessPoolExecutor Example

Doc/whatsnew/3.14.rst

@@ -306,6 +306,9 @@ concurrent.futures
   incompatible *fork* start method you must explicitly request it by
   supplying a *mp_context* to :class:`concurrent.futures.ProcessPoolExecutor`.
   (Contributed by Gregory P.  Smith in :gh:`84559`.)
+* Add :meth:`concurrent.futures.ProcessPoolExecutor.terminate_workers` as
+  a way to terminate all living worker processes in the given pool.
+  (Contributed by Charles Machalow in :gh:`128043`.)
 
 ctypes
 ------

Lib/concurrent/futures/process.py

@@ -57,6 +57,7 @@
 import weakref
 from functools import partial
 import itertools
+import signal
 import sys
 from traceback import format_exception
 
@@ -855,3 +856,31 @@ def shutdown(self, wait=True, *, cancel_futures=False):
         self._executor_manager_thread_wakeup = None
 
     shutdown.__doc__ = _base.Executor.shutdown.__doc__
+
+    def terminate_workers(self, signal=signal.SIGTERM):
+        """Attempts to terminate the executor's workers using the given signal.
+        Iterates through all of the current processes and sends the given signal if
+        the process is still alive.
+
+        After terminating workers, the pool will be in a broken state and no longer usable.
+
+        Args:
+            signal: The signal to send to each worker process. Defaults to
+                signal.SIGTERM.
+        """
+        if not self._processes:
+            return
+
+        for pid, proc in self._processes.items():
+            try:
+                is_alive = proc.is_alive()
+            except ValueError:
+                # The process is already exited/closed out.
+                is_alive = False
+
+            if is_alive:
+                try:
+                    os.kill(pid, signal)
+                except ProcessLookupError:
+                    # The process just ended before our signal
+                    pass

Lib/test/test_concurrent_futures/test_process_pool.py

@@ -1,8 +1,11 @@
+import multiprocessing
 import os
+import queue
 import sys
 import threading
 import time
 import unittest
+import unittest.mock
 from concurrent import futures
 from concurrent.futures.process import BrokenProcessPool
 
@@ -22,6 +25,12 @@ def __init__(self, mgr):
     def __del__(self):
         self.event.set()
 
+def _put_sleep_put(queue):
+    """ Used as part of test_process_pool_executor_terminate_workers """
+    queue.put('started')
+    time.sleep(2)
+    queue.put('finished')
+
 
 class ProcessPoolExecutorTest(ExecutorTest):
 
@@ -218,6 +227,60 @@ def mock_start_new_thread(func, *args, **kwargs):
                     list(executor.map(mul, [(2, 3)] * 10))
             executor.shutdown()
 
+    def test_process_pool_executor_terminate_workers(self):
+        manager = multiprocessing.Manager()
+        q = manager.Queue()
+
+        with futures.ProcessPoolExecutor(max_workers=1) as executor:
+            executor.submit(_put_sleep_put, q)
+
+            # We should get started, but not finished since we'll terminate the workers just after
+            self.assertEqual(q.get(timeout=1), 'started')
+
+            executor.terminate_workers()
+
+            try:
+                q.get(timeout=1)
+                raise RuntimeError("Queue should not have gotten a second value")
+            except queue.Empty:
+                pass
+
+    def test_process_pool_executor_terminate_workers_dead_workers(self):
+        with futures.ProcessPoolExecutor(max_workers=1) as executor:
+            try:
+                executor.submit(os._exit, 1).result()
+            except BrokenProcessPool:
+                # BrokenProcessPool will be raised by our call to .result() since the worker will die
+                pass
+
+            # The worker has been killed already, terminate_workers should basically no-op
+            executor.terminate_workers()
+
+    def test_process_pool_executor_terminate_workers_not_started_yet(self):
+        with futures.ProcessPoolExecutor(max_workers=1) as executor:
+            # The worker has not been started yet, terminate_workers should basically no-op
+            executor.terminate_workers()
+
+    def test_process_pool_executor_terminate_workers_stops_pool(self):
+        with futures.ProcessPoolExecutor(max_workers=1) as executor:
+            executor.submit(time.sleep, 0).result()
+
+            executor.terminate_workers()
+
+            try:
+                executor.submit(time.sleep, 0).result()
+                raise RuntimeError("Should have raised BrokenProcessPool")
+            except BrokenProcessPool:
+                pass
+
+    @unittest.mock.patch('concurrent.futures.process.os.kill')
+    def test_process_pool_executor_terminate_workers_passes_signal(self, mock_kill):
+        with futures.ProcessPoolExecutor(max_workers=1) as executor:
+            executor.submit(time.sleep, 0).result()
+
+            executor.terminate_workers(9)
+            mock_kill.assert_called_once_with(list(executor._processes.values())[0].pid, 9)
+
 
 create_executor_tests(globals(), ProcessPoolExecutorTest,
                       executor_mixins=(ProcessPoolForkMixin,

Misc/NEWS.d/next/Library/2024-12-17-18-53-21.gh-issue-128041.W96kAr.rst

@@ -0,0 +1 @@
+Add a ``terminate_workers`` method to ``ProcessPoolExecutor`` to allow a way to attempt to force kill the worker processes.