| FazBrowse GitHub Viewer | Trending | | Home |
| Tools: [Download Repo ZIP] [Original HTTPS Page] |
1 parent 7afa476 commit f97e409
5 files changed
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -415,6 +415,30 @@ to a :class:`ProcessPoolExecutor` will result in deadlock. | |||
| 415 | 415 | require the *fork* start method for :class:`ProcessPoolExecutor` you must | |
| 416 | 416 | explicitly pass ``mp_context=multiprocessing.get_context("fork")``. | |
| 417 | 417 | ||
| 418 | + .. method:: terminate_workers() | ||
| 419 | + | ||
| 420 | + Attempt to terminate all living worker processes immediately by calling | ||
| 421 | + :meth:`Process.terminate <multiprocessing.Process.terminate>` on each of them. | ||
| 422 | + Internally, it will also call :meth:`Executor.shutdown` to ensure that all | ||
| 423 | + other resources associated with the executor are freed. | ||
| 424 | + | ||
| 425 | + After calling this method the caller should no longer submit tasks to the | ||
| 426 | + executor. | ||
| 427 | + | ||
| 428 | + .. versionadded:: next | ||
| 429 | + | ||
| 430 | + .. method:: kill_workers() | ||
| 431 | + | ||
| 432 | + Attempt to kill all living worker processes immediately by calling | ||
| 433 | + :meth:`Process.kill <multiprocessing.Process.kill>` on each of them. | ||
| 434 | + Internally, it will also call :meth:`Executor.shutdown` to ensure that all | ||
| 435 | + other resources associated with the executor are freed. | ||
| 436 | + | ||
| 437 | + After calling this method the caller should no longer submit tasks to the | ||
| 438 | + executor. | ||
| 439 | + | ||
| 440 | + .. versionadded:: next | ||
| 441 | + | ||
| 418 | 442 | .. _processpoolexecutor-example: | |
| 419 | 443 | ||
| 420 | 444 | ProcessPoolExecutor Example | |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -444,6 +444,11 @@ contextvars | |||
| 444 | 444 | * Support context manager protocol by :class:`contextvars.Token`. | |
| 445 | 445 | (Contributed by Andrew Svetlov in :gh:`129889`.) | |
| 446 | 446 | ||
| 447 | + * Add :meth:`concurrent.futures.ProcessPoolExecutor.terminate_workers` and | ||
| 448 | + :meth:`concurrent.futures.ProcessPoolExecutor.kill_workers` as | ||
| 449 | + ways to terminate or kill all living worker processes in the given pool. | ||
| 450 | + (Contributed by Charles Machalow in :gh:`128043`.) | ||
| 451 | + | ||
| 447 | 452 | ||
| 448 | 453 | ctypes | |
| 449 | 454 | ------ | |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -626,6 +626,14 @@ class BrokenProcessPool(_base.BrokenExecutor): | |||
| 626 | 626 | while a future was in the running state. | |
| 627 | 627 | """ | |
| 628 | 628 | ||
| 629 | + _TERMINATE = "terminate" | ||
| 630 | + _KILL = "kill" | ||
| 631 | + | ||
| 632 | + _SHUTDOWN_CALLBACK_OPERATION = { | ||
| 633 | + _TERMINATE, | ||
| 634 | + _KILL | ||
| 635 | + } | ||
| 636 | + | ||
| 629 | 637 | ||
| 630 | 638 | class ProcessPoolExecutor(_base.Executor): | |
| 631 | 639 | def __init__(self, max_workers=None, mp_context=None, | |
@@ -855,3 +863,66 @@ def shutdown(self, wait=True, *, cancel_futures=False): | |||
| 855 | 863 | self._executor_manager_thread_wakeup = None | |
| 856 | 864 | ||
| 857 | 865 | shutdown.__doc__ = _base.Executor.shutdown.__doc__ | |
| 866 | + | ||
| 867 | + def _force_shutdown(self, operation): | ||
| 868 | + """Attempts to terminate or kill the executor's workers based off the | ||
| 869 | + given operation. Iterates through all of the current processes and | ||
| 870 | + performs the relevant task if the process is still alive. | ||
| 871 | + | ||
| 872 | + After terminating workers, the pool will be in a broken state | ||
| 873 | + and no longer usable (for instance, new tasks should not be | ||
| 874 | + submitted). | ||
| 875 | + """ | ||
| 876 | + if operation not in _SHUTDOWN_CALLBACK_OPERATION: | ||
| 877 | + raise ValueError(f"Unsupported operation: {operation!r}") | ||
| 878 | + | ||
| 879 | + processes = {} | ||
| 880 | + if self._processes: | ||
| 881 | + processes = self._processes.copy() | ||
| 882 | + | ||
| 883 | + # shutdown will invalidate ._processes, so we copy it right before | ||
| 884 | + # calling. If we waited here, we would deadlock if a process decides not | ||
| 885 | + # to exit. | ||
| 886 | + self.shutdown(wait=False, cancel_futures=True) | ||
| 887 | + | ||
| 888 | + if not processes: | ||
| 889 | + return | ||
| 890 | + | ||
| 891 | + for proc in processes.values(): | ||
| 892 | + try: | ||
| 893 | + if not proc.is_alive(): | ||
| 894 | + continue | ||
| 895 | + except ValueError: | ||
| 896 | + # The process is already exited/closed out. | ||
| 897 | + continue | ||
| 898 | + | ||
| 899 | + try: | ||
| 900 | + if operation == _TERMINATE: | ||
| 901 | + proc.terminate() | ||
| 902 | + elif operation == _KILL: | ||
| 903 | + proc.kill() | ||
| 904 | + except ProcessLookupError: | ||
| 905 | + # The process just ended before our signal | ||
| 906 | + continue | ||
| 907 | + | ||
| 908 | + def terminate_workers(self): | ||
| 909 | + """Attempts to terminate the executor's workers. | ||
| 910 | + Iterates through all of the current worker processes and terminates | ||
| 911 | + each one that is still alive. | ||
| 912 | + | ||
| 913 | + After terminating workers, the pool will be in a broken state | ||
| 914 | + and no longer usable (for instance, new tasks should not be | ||
| 915 | + submitted). | ||
| 916 | + """ | ||
| 917 | + return self._force_shutdown(operation=_TERMINATE) | ||
| 918 | + | ||
| 919 | + def kill_workers(self): | ||
| 920 | + """Attempts to kill the executor's workers. | ||
| 921 | + Iterates through all of the current worker processes and kills | ||
| 922 | + each one that is still alive. | ||
| 923 | + | ||
| 924 | + After killing workers, the pool will be in a broken state | ||
| 925 | + and no longer usable (for instance, new tasks should not be | ||
| 926 | + submitted). | ||
| 927 | + """ | ||
| 928 | + return self._force_shutdown(operation=_KILL) | ||
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -1,13 +1,17 @@ | |||
| 1 | 1 | import os | |
| 2 | + import queue | ||
| 3 | + import signal | ||
| 2 | 4 | import sys | |
| 3 | 5 | import threading | |
| 4 | 6 | import time | |
| 5 | 7 | import unittest | |
| 8 | + import unittest.mock | ||
| 6 | 9 | from concurrent import futures | |
| 7 | 10 | from concurrent.futures.process import BrokenProcessPool | |
| 8 | 11 | ||
| 9 | 12 | from test import support | |
| 10 | 13 | from test.support import hashlib_helper | |
| 14 | + from test.test_importlib.metadata.fixtures import parameterize | ||
| 11 | 15 | ||
| 12 | 16 | from .executor import ExecutorTest, mul | |
| 13 | 17 | from .util import ( | |
@@ -22,6 +26,19 @@ def __init__(self, mgr): | |||
| 22 | 26 | def __del__(self): | |
| 23 | 27 | self.event.set() | |
| 24 | 28 | ||
| 29 | + TERMINATE_WORKERS = futures.ProcessPoolExecutor.terminate_workers.__name__ | ||
| 30 | + KILL_WORKERS = futures.ProcessPoolExecutor.kill_workers.__name__ | ||
| 31 | + FORCE_SHUTDOWN_PARAMS = [ | ||
| 32 | + dict(function_name=TERMINATE_WORKERS), | ||
| 33 | + dict(function_name=KILL_WORKERS), | ||
| 34 | + ] | ||
| 35 | + | ||
| 36 | + def _put_sleep_put(queue): | ||
| 37 | + """ Used as part of test_terminate_workers """ | ||
| 38 | + queue.put('started') | ||
| 39 | + time.sleep(2) | ||
| 40 | + queue.put('finished') | ||
| 41 | + | ||
| 25 | 42 | ||
| 26 | 43 | class ProcessPoolExecutorTest(ExecutorTest): | |
| 27 | 44 | ||
@@ -218,6 +235,86 @@ def mock_start_new_thread(func, *args, **kwargs): | |||
| 218 | 235 | list(executor.map(mul, [(2, 3)] * 10)) | |
| 219 | 236 | executor.shutdown() | |
| 220 | 237 | ||
| 238 | + def test_terminate_workers(self): | ||
| 239 | + mock_fn = unittest.mock.Mock() | ||
| 240 | + with self.executor_type(max_workers=1) as executor: | ||
| 241 | + executor._force_shutdown = mock_fn | ||
| 242 | + executor.terminate_workers() | ||
| 243 | + | ||
| 244 | + mock_fn.assert_called_once_with(operation=futures.process._TERMINATE) | ||
| 245 | + | ||
| 246 | + def test_kill_workers(self): | ||
| 247 | + mock_fn = unittest.mock.Mock() | ||
| 248 | + with self.executor_type(max_workers=1) as executor: | ||
| 249 | + executor._force_shutdown = mock_fn | ||
| 250 | + executor.kill_workers() | ||
| 251 | + | ||
| 252 | + mock_fn.assert_called_once_with(operation=futures.process._KILL) | ||
| 253 | + | ||
| 254 | + def test_force_shutdown_workers_invalid_op(self): | ||
| 255 | + with self.executor_type(max_workers=1) as executor: | ||
| 256 | + self.assertRaises(ValueError, | ||
| 257 | + executor._force_shutdown, | ||
| 258 | + operation='invalid operation'), | ||
| 259 | + | ||
| 260 | + @parameterize(*FORCE_SHUTDOWN_PARAMS) | ||
| 261 | + def test_force_shutdown_workers(self, function_name): | ||
| 262 | + manager = self.get_context().Manager() | ||
| 263 | + q = manager.Queue() | ||
| 264 | + | ||
| 265 | + with self.executor_type(max_workers=1) as executor: | ||
| 266 | + executor.submit(_put_sleep_put, q) | ||
| 267 | + | ||
| 268 | + # We should get started, but not finished since we'll terminate the | ||
| 269 | + # workers just after | ||
| 270 | + self.assertEqual(q.get(timeout=5), 'started') | ||
| 271 | + | ||
| 272 | + worker_process = list(executor._processes.values())[0] | ||
| 273 | + getattr(executor, function_name)() | ||
| 274 | + worker_process.join() | ||
| 275 | + | ||
| 276 | + if function_name == TERMINATE_WORKERS or \ | ||
| 277 | + sys.platform == 'win32': | ||
| 278 | + # On windows, kill and terminate both send SIGTERM | ||
| 279 | + self.assertEqual(worker_process.exitcode, -signal.SIGTERM) | ||
| 280 | + elif function_name == KILL_WORKERS: | ||
| 281 | + self.assertEqual(worker_process.exitcode, -signal.SIGKILL) | ||
| 282 | + else: | ||
| 283 | + self.fail(f"Unknown operation: {function_name}") | ||
| 284 | + | ||
| 285 | + self.assertRaises(queue.Empty, q.get, timeout=1) | ||
| 286 | + | ||
| 287 | + @parameterize(*FORCE_SHUTDOWN_PARAMS) | ||
| 288 | + def test_force_shutdown_workers_dead_workers(self, function_name): | ||
| 289 | + with self.executor_type(max_workers=1) as executor: | ||
| 290 | + future = executor.submit(os._exit, 1) | ||
| 291 | + self.assertRaises(BrokenProcessPool, future.result) | ||
| 292 | + | ||
| 293 | + # even though the pool is broken, this shouldn't raise | ||
| 294 | + getattr(executor, function_name)() | ||
| 295 | + | ||
| 296 | + @parameterize(*FORCE_SHUTDOWN_PARAMS) | ||
| 297 | + def test_force_shutdown_workers_not_started_yet(self, function_name): | ||
| 298 | + ctx = self.get_context() | ||
| 299 | + with unittest.mock.patch.object(ctx, 'Process') as mock_process: | ||
| 300 | + with self.executor_type(max_workers=1, mp_context=ctx) as executor: | ||
| 301 | + # The worker has not been started yet, terminate/kill_workers | ||
| 302 | + # should basically no-op | ||
| 303 | + getattr(executor, function_name)() | ||
| 304 | + | ||
| 305 | + mock_process.return_value.kill.assert_not_called() | ||
| 306 | + mock_process.return_value.terminate.assert_not_called() | ||
| 307 | + | ||
| 308 | + @parameterize(*FORCE_SHUTDOWN_PARAMS) | ||
| 309 | + def test_force_shutdown_workers_stops_pool(self, function_name): | ||
| 310 | + with self.executor_type(max_workers=1) as executor: | ||
| 311 | + task = executor.submit(time.sleep, 0) | ||
| 312 | + self.assertIsNone(task.result()) | ||
| 313 | + | ||
| 314 | + getattr(executor, function_name)() | ||
| 315 | + | ||
| 316 | + self.assertRaises(RuntimeError, executor.submit, time.sleep, 0) | ||
| 317 | + | ||
| 221 | 318 | ||
| 222 | 319 | create_executor_tests(globals(), ProcessPoolExecutorTest, | |
| 223 | 320 | executor_mixins=(ProcessPoolForkMixin, | |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -0,0 +1,4 @@ | |||
| 1 | + Add :meth:`concurrent.futures.ProcessPoolExecutor.terminate_workers` and | ||
| 2 | + :meth:`concurrent.futures.ProcessPoolExecutor.kill_workers` as | ||
| 3 | + ways to terminate or kill all living worker processes in the given pool. | ||
| 4 | + (Contributed by Charles Machalow in :gh:`128043`.) | ||
| Back | FazBrowse Home | New Git URL |
0 commit comments