PR feedback. Split terminate_workers into terminate_workers and kill_workers.

Remove the ability to give an arbitrary signal, though leave plumbing to make it easy enough to re-add if needed.

Author: csm10495

Doc/library/concurrent.futures.rst

@@ -415,13 +415,24 @@ to a :class:`ProcessPoolExecutor` will result in deadlock.
       require the *fork* start method for :class:`ProcessPoolExecutor` you must
       explicitly pass ``mp_context=multiprocessing.get_context("fork")``.
 
-   .. method:: terminate_workers(signal=signal.SIGTERM)
+   .. method:: terminate_workers()
 
-      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. Internally, it will also call
-      :meth:`Executor.shutdown` to ensure that all other resources associated with
-      the executor are freed.
+      Attempt to terminate all living worker processes immediately by calling
+      :meth:`Process.terminate <multiprocessing.Process.terminate>` on each of them.
+      Internally, it will also call :meth:`Executor.shutdown` to ensure that all
+      other resources associated with the executor are freed.
+
+      After calling this method the caller should no longer submit tasks to the
+      executor.
+
+      .. versionadded:: next
+
+   .. method:: kill_workers()
+
+      Attempt to kill all living worker processes immediately by calling
+      :meth:`Process.terminate <multiprocessing.Process.kill>` on each of them.
+      Internally, it will also call:meth:`Executor.shutdown` to ensure that all
+      other resources associated with the executor are freed.
 
       After calling this method the caller should no longer submit tasks to the
       executor.

Doc/whatsnew/3.14.rst

@@ -440,8 +440,9 @@ contextvars
   (Contributed by Andrew Svetlov in :gh:`129889`.)
 
 
-* Add :meth:`concurrent.futures.ProcessPoolExecutor.terminate_workers` as
-  a way to terminate all living worker processes in the given pool.
+* 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

Lib/concurrent/futures/process.py

@@ -45,6 +45,7 @@
 
 __author__ = 'Brian Quinlan ([email protected])'
 
+import enum
 import os
 from concurrent.futures import _base
 import queue
@@ -627,6 +628,18 @@ class BrokenProcessPool(_base.BrokenExecutor):
     while a future was in the running state.
     """
 
+class _TerminateOrKillOperation(enum.Enum):
+    """Enum for _terminate_or_kill_workers().
+
+    Used to determine the operation used by the
+    _terminate_or_kill_workers() method.
+    """
+    # Delegate to call process.terminate()
+    TERMINATE = 1
+
+    # Delegate to call process.kill()
+    KILL = 2
+
 
 class ProcessPoolExecutor(_base.Executor):
     def __init__(self, max_workers=None, mp_context=None,
@@ -857,19 +870,18 @@ def shutdown(self, wait=True, *, cancel_futures=False):
 
     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.
+    def _terminate_or_kill_workers(self, operation: _TerminateOrKillOperation):
+        """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).
-
-        Args:
-            signal: The signal to send to each worker process. Defaults to
-                signal.SIGTERM.
         """
+        if operation not in _TerminateOrKillOperation._member_map_.values():
+            raise ValueError(f"Unsupported operation: {operation}")
+
         processes = {}
         if self._processes:
             processes = self._processes.copy()
@@ -881,7 +893,7 @@ def terminate_workers(self, signal=signal.SIGTERM):
         if not processes:
             return
 
-        for pid, proc in processes.items():
+        for proc in processes.values():
             try:
                 if not proc.is_alive():
                     continue
@@ -890,7 +902,32 @@ def terminate_workers(self, signal=signal.SIGTERM):
                 continue
 
             try:
-                os.kill(pid, signal)
+                if operation == _TerminateOrKillOperation.TERMINATE:
+                    proc.terminate()
+                elif operation == _TerminateOrKillOperation.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._terminate_or_kill_workers(operation=_TerminateOrKillOperation.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._terminate_or_kill_workers(operation=_TerminateOrKillOperation.KILL)

Lib/test/test_concurrent_futures/test_process_pool.py

@@ -228,59 +228,70 @@ 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()
+    def test_process_pool_executor_terminate_kill_workers(self):
+        for function_name in ('terminate_workers', 'kill_workers'):
+            manager = multiprocessing.Manager()
+            q = manager.Queue()
 
-        with futures.ProcessPoolExecutor(max_workers=1) as executor:
-            executor.submit(_put_sleep_put, q)
+            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')
+                # 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()
+                getattr(executor, function_name)()
 
-            self.assertRaises(queue.Empty, q.get, timeout=1)
+                self.assertRaises(queue.Empty, q.get, timeout=1)
 
-    def test_process_pool_executor_terminate_workers_dead_workers(self):
-        with futures.ProcessPoolExecutor(max_workers=1) as executor:
-            future = executor.submit(os._exit, 1)
-            self.assertRaises(BrokenProcessPool, future.result)
+    def test_process_pool_executor_terminate_kill_workers_dead_workers(self):
+        for function_name in ('terminate_workers', 'kill_workers'):
+            with futures.ProcessPoolExecutor(max_workers=1) as executor:
+                future = executor.submit(os._exit, 1)
+                self.assertRaises(BrokenProcessPool, future.result)
 
-    @unittest.mock.patch('concurrent.futures.process.os.kill')
-    def test_process_pool_executor_terminate_workers_not_started_yet(self, mock_kill):
-        with futures.ProcessPoolExecutor(max_workers=1) as executor:
-            # The worker has not been started yet, terminate_workers should basically no-op
-            executor.terminate_workers()
+                # even though the pool is broken, this shouldn't raise
+                getattr(executor, function_name)()
 
-        mock_kill.assert_not_called()
+    def test_process_pool_executor_terminate_kill_workers_not_started_yet(self):
+        for function_name in ('terminate_workers', 'kill_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()
+            context_with_mocked_process = multiprocessing.get_context()
+            with unittest.mock.patch.object(context_with_mocked_process, 'Process') as mock_process:
 
-            executor.terminate_workers()
+                with futures.ProcessPoolExecutor(max_workers=1, mp_context=context_with_mocked_process) as executor:
+                    # The worker has not been started yet, terminate/kill_workers should basically no-op
+                    getattr(executor, function_name)()
+
+                mock_process.return_value.kill.assert_not_called()
+                mock_process.return_value.terminate.assert_not_called()
 
-            self.assertRaises(RuntimeError, executor.submit, time.sleep, 0)
+    def test_process_pool_executor_terminate_kill_workers_stops_pool(self):
+        for function_name in ('terminate_workers', 'kill_workers'):
+            with futures.ProcessPoolExecutor(max_workers=1) as executor:
+                executor.submit(time.sleep, 0).result()
 
-    @unittest.mock.patch('concurrent.futures.process.os.kill')
-    def test_process_pool_executor_terminate_workers_passes_signal(self, mock_kill):
+                getattr(executor, function_name)()
+
+                self.assertRaises(RuntimeError, executor.submit, time.sleep, 0)
+
+    def test_process_pool_executor_terminate_workers(self):
         with futures.ProcessPoolExecutor(max_workers=1) as executor:
-            future = executor.submit(time.sleep, 0)
-            future.result()
+            executor._terminate_or_kill_workers = unittest.mock.Mock()
+            executor.terminate_workers()
+
+        executor._terminate_or_kill_workers.assert_called_once_with(operation=futures.process._TerminateOrKillOperation.TERMINATE)
 
-            worker_process = list(executor._processes.values())[0]
-            executor.terminate_workers(signal.SIGABRT)
+    def test_process_pool_executor_kill_workers(self):
+        with futures.ProcessPoolExecutor(max_workers=1) as executor:
+            executor._terminate_or_kill_workers = unittest.mock.Mock()
+            executor.kill_workers()
 
-            mock_kill.assert_called_once_with(worker_process.pid, signal.SIGABRT)
+        executor._terminate_or_kill_workers.assert_called_once_with(operation=futures.process._TerminateOrKillOperation.KILL)
 
-    def test_process_pool_executor_terminate_workers_passes_even_bad_signals(self):
+    def test_process_pool_executor_terminate_or_kill_workers_invalid_operation(self):
         with futures.ProcessPoolExecutor(max_workers=1) as executor:
-            future = executor.submit(time.sleep, 0)
-            future.result()
+            self.assertRaises(ValueError, executor._terminate_or_kill_workers, operation='invalid operation'),
 
-            # 'potatoes' isn't a valid signal, so os.kill will raise a TypeError
-            self.assertRaises(TypeError, executor.terminate_workers, 'potatoes')
 
 
 create_executor_tests(globals(), ProcessPoolExecutorTest,

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

@@ -1,3 +1,4 @@
-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`.)
+* Add :meth:`concurrent.futures.ProcessPoolExecutor.terminate_workers` and
+  :meth:`concurrent.futures.ProcessPoolExecutor.terminate_workers` as
+  ways to terminate or kill all living worker processes in the given pool.
+  (Contributed by Charles Machalow in :gh:`128043`.)