Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1###############################################################################
2# Re-implementation of the ProcessPoolExecutor more robust to faults
3#
4# author: Thomas Moreau and Olivier Grisel
5#
6# adapted from concurrent/futures/process_pool_executor.py (17/02/2017)
7# * Add an extra management thread to detect executor_manager_thread failures,
8# * Improve the shutdown process to avoid deadlocks,
9# * Add timeout for workers,
10# * More robust pickling process.
11#
12# Copyright 2009 Brian Quinlan. All Rights Reserved.
13# Licensed to PSF under a Contributor Agreement.
15"""Implements ProcessPoolExecutor.
17The follow diagram and text describe the data-flow through the system:
19|======================= In-process =====================|== Out-of-process ==|
21+----------+ +----------+ +--------+ +-----------+ +---------+
22| | => | Work Ids | | | | Call Q | | Process |
23| | +----------+ | | +-----------+ | Pool |
24| | | ... | | | | ... | +---------+
25| | | 6 | => | | => | 5, call() | => | |
26| | | 7 | | | | ... | | |
27| Process | | ... | | Local | +-----------+ | Process |
28| Pool | +----------+ | Worker | | #1..n |
29| Executor | | Thread | | |
30| | +----------- + | | +-----------+ | |
31| | <=> | Work Items | <=> | | <= | Result Q | <= | |
32| | +------------+ | | +-----------+ | |
33| | | 6: call() | | | | ... | | |
34| | | future | +--------+ | 4, result | | |
35| | | ... | | 3, except | | |
36+----------+ +------------+ +-----------+ +---------+
38Executor.submit() called:
39- creates a uniquely numbered _WorkItem and adds it to the "Work Items" dict
40- adds the id of the _WorkItem to the "Work Ids" queue
42Local worker thread:
43- reads work ids from the "Work Ids" queue and looks up the corresponding
44 WorkItem from the "Work Items" dict: if the work item has been cancelled then
45 it is simply removed from the dict, otherwise it is repackaged as a
46 _CallItem and put in the "Call Q". New _CallItems are put in the "Call Q"
47 until "Call Q" is full. NOTE: the size of the "Call Q" is kept small because
48 calls placed in the "Call Q" can no longer be cancelled with Future.cancel().
49- reads _ResultItems from "Result Q", updates the future stored in the
50 "Work Items" dict and deletes the dict entry
52Process #1..n:
53- reads _CallItems from "Call Q", executes the calls, and puts the resulting
54 _ResultItems in "Result Q"
55"""
58__author__ = "Thomas Moreau (thomas.moreau.2010@gmail.com)"
61import faulthandler
62import os
63import gc
64import sys
65import queue
66import struct
67import weakref
68import warnings
69import itertools
70import traceback
71import threading
72from time import time, sleep
73import multiprocessing as mp
74from functools import partial
75from pickle import PicklingError
76from concurrent.futures import Executor, InvalidStateError
77from concurrent.futures._base import LOGGER
78from concurrent.futures.process import BrokenProcessPool as _BPPException
79from multiprocessing.connection import wait
81from ._base import Future
82from .backend import get_context
83from .backend.context import cpu_count, _MAX_WINDOWS_WORKERS
84from .backend.queues import Queue, SimpleQueue
85from .backend.reduction import set_loky_pickler, get_loky_pickler_name
86from .backend.utils import kill_process_tree, get_exitcodes_terminated_worker
87from .initializers import _prepare_initializer
90# Mechanism to prevent infinite process spawning. When a worker of a
91# ProcessPoolExecutor nested in MAX_DEPTH Executor tries to create a new
92# Executor, a LokyRecursionError is raised
93MAX_DEPTH = int(os.environ.get("LOKY_MAX_DEPTH", 10))
94_CURRENT_DEPTH = 0
96# Minimum time interval between two consecutive memory leak protection checks.
97_MEMORY_LEAK_CHECK_DELAY = 1.0
99# Number of bytes of memory usage allowed over the reference process size.
100# Workloads whose tasks differ a lot in size can legitimately exceed the default
101# without leaking, so it is configurable; it is read in the worker processes,
102# which inherit the environment of the process that created the executor.
103_MAX_MEMORY_LEAK_SIZE = int(
104 float(os.environ.get("LOKY_MAX_MEMORY_LEAK_SIZE", 3e8))
105)
108try:
109 from psutil import Process
111 _USE_PSUTIL = True
113 def _get_memory_usage(pid, force_gc=False):
114 if force_gc:
115 gc.collect()
117 mem_size = Process(pid).memory_info().rss
118 mp.util.debug(f"psutil return memory size: {mem_size}")
119 return mem_size
121except ImportError:
122 _USE_PSUTIL = False
125class _RecycledWorkerPid(int):
126 """PID of a worker that the executor itself decided to shut down.
128 Behaves like the plain PID the workers usually send back, but lets the
129 manager thread tell a recycling it triggered on purpose from a worker that
130 stopped on its own.
131 """
134class _ThreadWakeup:
135 def __init__(self):
136 self._closed = False
137 self._reader, self._writer = mp.Pipe(duplex=False)
139 def close(self):
140 if not self._closed:
141 self._closed = True
142 self._writer.close()
143 self._reader.close()
145 def wakeup(self):
146 if not self._closed:
147 self._writer.send_bytes(b"")
149 def clear(self):
150 if not self._closed:
151 while self._reader.poll():
152 self._reader.recv_bytes()
155class _ExecutorFlags:
156 """necessary references to maintain executor states without preventing gc
158 It permits to keep the information needed by executor_manager_thread
159 and crash_detection_thread to maintain the pool without preventing the
160 garbage collection of unreferenced executors.
161 """
163 def __init__(self, shutdown_lock):
165 self.shutdown = False
166 self.broken = None
167 self.kill_workers = False
168 self.shutdown_lock = shutdown_lock
170 def flag_as_shutting_down(self, kill_workers=None):
171 with self.shutdown_lock:
172 self.shutdown = True
173 if kill_workers is not None:
174 self.kill_workers = kill_workers
176 def flag_as_broken(self, broken):
177 with self.shutdown_lock:
178 self.shutdown = True
179 self.broken = broken
182# Prior to 3.9, executor_manager_thread is created as daemon thread. This means
183# that it is not joined automatically when the interpreter is shutting down.
184# To work around this problem, an exit handler is installed to tell the
185# thread to exit when the interpreter is shutting down and then waits until
186# it finishes. The thread needs to be daemonized because the atexit hooks are
187# called after all non daemonized threads are joined.
188#
189# Starting 3.9, there exists a specific atexit hook to be called before joining
190# the threads so the executor_manager_thread does not need to be daemonized
191# anymore.
192#
193# The atexit hooks are registered when starting the first ProcessPoolExecutor
194# to avoid import having an effect on the interpreter.
196_global_shutdown = False
197_global_shutdown_lock = threading.Lock()
198_threads_wakeups = weakref.WeakKeyDictionary()
201def _python_exit():
202 global _global_shutdown
203 _global_shutdown = True
205 # Materialize the list of items to avoid error due to iterating over
206 # changing size dictionary.
207 items = list(_threads_wakeups.items())
208 if len(items) > 0:
209 mp.util.debug(
210 f"Interpreter shutting down. Waking up {len(items)}"
211 f"executor_manager_thread:\n{items}"
212 )
214 # Wake up the executor_manager_thread's so they can detect the interpreter
215 # is shutting down and exit.
216 for _, (shutdown_lock, thread_wakeup) in items:
217 with shutdown_lock:
218 thread_wakeup.wakeup()
220 # Collect the executor_manager_thread's to make sure we exit cleanly.
221 for thread, _ in items:
222 # This locks is to prevent situations where an executor is gc'ed in one
223 # thread while the atexit finalizer is running in another thread.
224 with _global_shutdown_lock:
225 thread.join()
228# With the fork context, _thread_wakeups is propagated to children.
229# Clear it after fork to avoid some situation that can cause some
230# freeze when joining the workers.
231mp.util.register_after_fork(_threads_wakeups, lambda obj: obj.clear())
234# Module variable to register the at_exit call
235process_pool_executor_at_exit = None
237# Controls how many more calls than processes will be queued in the call queue.
238# A smaller number will mean that processes spend more time idle waiting for
239# work while a larger number will make Future.cancel() succeed less frequently
240# (Futures in the call queue cannot be cancelled).
241EXTRA_QUEUED_CALLS = 1
244class _RemoteTraceback(Exception):
245 """Embed stringification of remote traceback in local traceback"""
247 def __init__(self, tb=None):
248 self.tb = f'\n"""\n{tb}"""'
250 def __str__(self):
251 return self.tb
254# Do not inherit from BaseException to mirror
255# concurrent.futures.process._ExceptionWithTraceback
256class _ExceptionWithTraceback:
257 def __init__(self, exc):
258 tb = getattr(exc, "__traceback__", None)
259 if tb is None:
260 _, _, tb = sys.exc_info()
261 tb = traceback.format_exception(type(exc), exc, tb)
262 tb = "".join(tb)
263 self.exc = exc
264 self.tb = tb
266 def __reduce__(self):
267 return _rebuild_exc, (self.exc, self.tb)
270def _rebuild_exc(exc, tb):
271 exc.__cause__ = _RemoteTraceback(tb)
272 return exc
275class _WorkItem:
277 __slots__ = ["future", "fn", "args", "kwargs"]
279 def __init__(self, future, fn, args, kwargs):
280 self.future = future
281 self.fn = fn
282 self.args = args
283 self.kwargs = kwargs
286class _ResultItem:
287 def __init__(self, work_id, exception=None, result=None):
288 self.work_id = work_id
289 self.exception = exception
290 self.result = result
293class _CallItem:
294 def __init__(self, work_id, fn, args, kwargs):
295 self.work_id = work_id
296 self.fn = fn
297 self.args = args
298 self.kwargs = kwargs
300 # Store the current loky_pickler so it is correctly set in the worker
301 self.loky_pickler = get_loky_pickler_name()
303 def __call__(self):
304 set_loky_pickler(self.loky_pickler)
305 return self.fn(*self.args, **self.kwargs)
307 def __repr__(self):
308 return (
309 f"CallItem({self.work_id}, {self.fn}, {self.args}, {self.kwargs})"
310 )
313class _SafeQueue(Queue):
314 """Safe Queue set exception to the future object linked to a job"""
316 def __init__(
317 self,
318 max_size=0,
319 ctx=None,
320 pending_work_items=None,
321 running_work_items=None,
322 thread_wakeup=None,
323 shutdown_lock=None,
324 executor_flags=None,
325 reducers=None,
326 ):
327 self.thread_wakeup = thread_wakeup
328 self.shutdown_lock = shutdown_lock
329 self.executor_flags = executor_flags
330 self.pending_work_items = pending_work_items
331 self.running_work_items = running_work_items
332 super().__init__(max_size, reducers=reducers, ctx=ctx)
334 def _on_queue_feeder_error(self, e, obj):
335 if not isinstance(obj, _CallItem):
336 super()._on_queue_feeder_error(e, obj)
337 return
338 # work_item can be None if another process terminated. In this
339 # case, the executor_manager_thread fails all work_items with
340 # BrokenProcessPool
341 work_item = self.pending_work_items.pop(obj.work_id, None)
342 try:
343 # format traceback only works on python3
344 if isinstance(e, struct.error):
345 raised_error = RuntimeError(
346 "The task could not be sent to the workers as it is too "
347 "large for `send_bytes`."
348 )
349 else:
350 raised_error = PicklingError(
351 "Could not pickle the task to send it to the workers."
352 )
353 tb = traceback.format_exception(
354 type(e), e, getattr(e, "__traceback__", None)
355 )
356 raised_error.__cause__ = _RemoteTraceback("".join(tb))
357 self.running_work_items.remove(obj.work_id)
358 except BaseException as hook_exc:
359 # An error here would kill the feeder thread silently and leave
360 # every later task unsent: flag the executor as broken instead
361 raised_error = BrokenProcessPool(
362 "The call queue feeder thread crashed: the executor is "
363 "broken and the pending tasks have been cancelled."
364 )
365 tb = traceback.format_exception(
366 type(hook_exc), hook_exc, hook_exc.__traceback__
367 )
368 raised_error.__cause__ = _RemoteTraceback("".join(tb))
369 LOGGER.critical(
370 "Exception in call queue feeder error hook:", exc_info=True
371 )
372 self.executor_flags.flag_as_broken(raised_error)
373 if work_item is not None:
374 work_item.future.set_exception(raised_error)
375 del work_item
376 with self.shutdown_lock:
377 self.thread_wakeup.wakeup()
380def _get_chunks(chunksize, *iterables):
381 """Iterates over zip()ed iterables in chunks."""
382 it = zip(*iterables)
383 while True:
384 chunk = tuple(itertools.islice(it, chunksize))
385 if not chunk:
386 return
387 yield chunk
390def _process_chunk(fn, chunk):
391 """Processes a chunk of an iterable passed to map.
393 Runs the function passed to map() on a chunk of the
394 iterable passed to map.
396 This function is run in a separate process.
398 """
399 return [fn(*args) for args in chunk]
402def _sendback_result(result_queue, work_id, result=None, exception=None):
403 """Safely send back the given result or exception"""
404 try:
405 result_queue.put(
406 _ResultItem(work_id, result=result, exception=exception)
407 )
408 except BaseException as e:
409 exc = _ExceptionWithTraceback(e)
410 result_queue.put(_ResultItem(work_id, exception=exc))
413def _enable_faulthandler_if_needed():
414 if "PYTHONFAULTHANDLER" in os.environ:
415 # Respect the environment variable to configure faulthandler. This
416 # makes it possible to never enable faulthandler in the loky workers by
417 # setting PYTHONFAULTHANDLER=0 explicitly in the environment.
418 mp.util.debug(
419 f"faulthandler explicitly configured by environment variable: "
420 f"PYTHONFAULTHANDLER={os.environ['PYTHONFAULTHANDLER']}."
421 )
422 else:
423 if faulthandler.is_enabled():
424 # Fault handler is already enabled, possibly via a custom
425 # initializer to customize the behavior.
426 mp.util.debug("faulthandler already enabled.")
427 else:
428 # Enable faulthandler by default with default paramaters otherwise.
429 mp.util.debug(
430 "Enabling faulthandler to report tracebacks on worker crashes."
431 )
432 faulthandler.enable()
435def _process_worker(
436 call_queue,
437 result_queue,
438 initializer,
439 initargs,
440 processes_management_lock,
441 timeout,
442 worker_exit_lock,
443 current_depth,
444):
445 """Evaluates calls from call_queue and places the results in result_queue.
447 This worker is run in a separate process.
449 Args:
450 call_queue: A ctx.Queue of _CallItems that will be read and
451 evaluated by the worker.
452 result_queue: A ctx.Queue of _ResultItems that will written
453 to by the worker.
454 initializer: A callable initializer, or None
455 initargs: A tuple of args for the initializer
456 processes_management_lock: A ctx.Lock avoiding worker timeout while
457 some workers are being spawned.
458 timeout: maximum time to wait for a new item in the call_queue. If that
459 time is expired, the worker will shutdown.
460 worker_exit_lock: Lock to avoid flagging the executor as broken on
461 workers timeout.
462 current_depth: Nested parallelism level, to avoid infinite spawning.
463 """
464 if initializer is not None:
465 try:
466 initializer(*initargs)
467 except BaseException:
468 LOGGER.critical("Exception in initializer:", exc_info=True)
469 # The parent will notice that the process stopped and
470 # mark the pool broken
471 return
473 # set the global _CURRENT_DEPTH mechanism to limit recursive call
474 global _CURRENT_DEPTH
475 _CURRENT_DEPTH = current_depth
476 _process_reference_size = None
477 _last_memory_leak_check = None
478 pid = os.getpid()
480 mp.util.debug(f"Worker started with timeout={timeout}")
481 _enable_faulthandler_if_needed()
483 while True:
484 try:
485 call_item = call_queue.get(block=True, timeout=timeout)
486 if call_item is None:
487 mp.util.info("Shutting down worker on sentinel")
488 except queue.Empty:
489 mp.util.info(f"Shutting down worker after timeout {timeout:0.3f}s")
490 if processes_management_lock.acquire(block=False):
491 processes_management_lock.release()
492 call_item = None
493 else:
494 mp.util.info("Could not acquire processes_management_lock")
495 continue
496 except BaseException:
497 previous_tb = traceback.format_exc()
498 try:
499 result_queue.put(_RemoteTraceback(previous_tb))
500 except BaseException:
501 # If we cannot format correctly the exception, at least print
502 # the traceback.
503 print(previous_tb)
504 mp.util.debug("Exiting with code 1")
505 sys.exit(1)
506 if call_item is None:
507 # Notify queue management thread about worker shutdown
508 result_queue.put(pid)
509 is_clean = worker_exit_lock.acquire(True, timeout=30)
511 # Early notify any loky executor running in this worker process
512 # (nested parallelism) that this process is about to shutdown to
513 # avoid a deadlock waiting undifinitely for the worker to finish.
514 _python_exit()
516 if is_clean:
517 mp.util.debug("Exited cleanly")
518 else:
519 mp.util.info("Main process did not release worker_exit")
520 return
521 try:
522 r = call_item()
523 except BaseException as e:
524 exc = _ExceptionWithTraceback(e)
525 result_queue.put(_ResultItem(call_item.work_id, exception=exc))
526 else:
527 _sendback_result(result_queue, call_item.work_id, result=r)
528 del r
530 # Free the resource as soon as possible, to avoid holding onto
531 # open files or shared memory that is not needed anymore
532 del call_item
534 if _USE_PSUTIL:
535 if _process_reference_size is None:
536 # Make reference measurement after the first call
537 _process_reference_size = _get_memory_usage(pid, force_gc=True)
538 _last_memory_leak_check = time()
539 continue
540 if time() - _last_memory_leak_check > _MEMORY_LEAK_CHECK_DELAY:
541 mem_usage = _get_memory_usage(pid)
542 _last_memory_leak_check = time()
543 if mem_usage - _process_reference_size < _MAX_MEMORY_LEAK_SIZE:
544 # Memory usage stays within bounds: everything is fine.
545 continue
547 # Check again memory usage; this time take the measurement
548 # after a forced garbage collection to break any reference
549 # cycles.
550 mem_usage = _get_memory_usage(pid, force_gc=True)
551 _last_memory_leak_check = time()
552 if mem_usage - _process_reference_size < _MAX_MEMORY_LEAK_SIZE:
553 # The GC managed to free the memory: everything is fine.
554 continue
556 # The process is leaking memory: let the master process
557 # know that we need to start a new worker.
558 mp.util.info("Memory leak detected: shutting down worker")
559 result_queue.put(_RecycledWorkerPid(pid))
560 with worker_exit_lock:
561 mp.util.debug("Exit due to memory leak")
562 return
563 else:
564 # if psutil is not installed, trigger gc.collect events
565 # regularly to limit potential memory leaks due to reference cycles
566 if _last_memory_leak_check is None or (
567 time() - _last_memory_leak_check > _MEMORY_LEAK_CHECK_DELAY
568 ):
569 gc.collect()
570 _last_memory_leak_check = time()
573class _ExecutorManagerThread(threading.Thread):
574 """Manages the communication between this process and the worker processes.
576 The manager is run in a local thread.
578 Args:
579 executor: A reference to the ProcessPoolExecutor that owns
580 this thread. A weakref will be own by the manager as well as
581 references to internal objects used to introspect the state of
582 the executor.
583 """
585 def __init__(self, executor):
586 # Store references to necessary internals of the executor.
588 # A _ThreadWakeup to allow waking up the executor_manager_thread from
589 # the main Thread and avoid deadlocks caused by permanently
590 # locked queues.
591 self.thread_wakeup = executor._executor_manager_thread_wakeup
592 self.shutdown_lock = executor._shutdown_lock
594 # A weakref.ref to the ProcessPoolExecutor that owns this thread. Used
595 # to determine if the ProcessPoolExecutor has been garbage collected
596 # and that the manager can exit.
597 # When the executor gets garbage collected, the weakref callback
598 # will wake up the queue management thread so that it can terminate
599 # if there is no pending work item.
600 def weakref_cb(
601 _,
602 thread_wakeup=self.thread_wakeup,
603 shutdown_lock=self.shutdown_lock,
604 ):
605 if mp is not None:
606 # At this point, the multiprocessing module can already be
607 # garbage collected. We only log debug info when still
608 # possible.
609 mp.util.debug(
610 "Executor collected: triggering callback for"
611 " QueueManager wakeup"
612 )
613 with shutdown_lock:
614 thread_wakeup.wakeup()
616 self.executor_reference = weakref.ref(executor, weakref_cb)
618 # The flags of the executor
619 self.executor_flags = executor._flags
621 # A list of the ctx.Process instances used as workers.
622 self.processes = executor._processes
624 # A ctx.Queue that will be filled with _CallItems derived from
625 # _WorkItems for processing by the process workers.
626 self.call_queue = executor._call_queue
628 # A ctx.SimpleQueue of _ResultItems generated by the process workers.
629 self.result_queue = executor._result_queue
631 # A queue.Queue of work ids e.g. Queue([5, 6, ...]).
632 self.work_ids_queue = executor._work_ids
634 # A dict mapping work ids to _WorkItems e.g.
635 # {5: <_WorkItem...>, 6: <_WorkItem...>, ...}
636 self.pending_work_items = executor._pending_work_items
638 # A list of the work_ids that are currently running
639 self.running_work_items = executor._running_work_items
641 # A lock to avoid concurrent shutdown of workers on timeout and spawn
642 # of new processes or shut down
643 self.processes_management_lock = executor._processes_management_lock
645 # Long-running executors can recycle workers many times over their
646 # lifetime; the user only needs to hear about it once.
647 self.recycling_warned = False
649 super().__init__(name="ExecutorManagerThread")
650 if sys.version_info < (3, 9):
651 self.daemon = True
653 def run(self):
654 # Main loop for the executor manager thread.
655 try:
656 self._run_loop()
657 except BaseException as exc:
658 # Nothing in the loop is expected to raise, so this is a bug, a
659 # warning filter turning one of the warnings below into an error,
660 # or a failure of the runtime itself, such as a queue that cannot
661 # start its feeder thread while the interpreter is shutting down
662 # (python/cpython#109047). Letting the exception kill this thread
663 # would leave every pending future unresolved and the executor
664 # looking healthy, so a caller waiting on a result would hang
665 # forever.
666 # Flag the executor as broken instead, with the error as cause.
667 bpe = BrokenProcessPool(
668 "The executor manager thread crashed: the executor is broken "
669 "and the pending tasks have been cancelled."
670 )
671 tb = traceback.format_exception(type(exc), exc, exc.__traceback__)
672 bpe.__cause__ = _RemoteTraceback("".join(tb))
673 # Also log it: with no pending future, nothing else reports it
674 LOGGER.critical(
675 "Exception in executor manager thread:", exc_info=True
676 )
677 self.terminate_broken(bpe)
679 def _run_loop(self):
680 while True:
681 self.add_call_item_to_queue()
683 result_item, is_broken, bpe = self.wait_result_broken_or_wakeup()
685 # The call queue feeder thread flags the executor broken and
686 # wakes us up when it cannot report a task failure itself
687 if not is_broken and self.executor_flags.broken is not None:
688 is_broken, bpe = True, self.executor_flags.broken
689 if is_broken:
690 self.terminate_broken(bpe)
691 return
692 if result_item is not None:
693 self.process_result_item(result_item)
694 # Delete reference to result_item to avoid keeping references
695 # while waiting on new results.
696 del result_item
698 if self.is_shutting_down():
699 self.flag_executor_shutting_down()
701 # When only canceled futures remain in pending_work_items, our
702 # next call to wait_result_broken_or_wakeup would hang forever.
703 # This makes sure we have some running futures or none at all.
704 self.add_call_item_to_queue()
706 # Since no new work items can be added, it is safe to shutdown
707 # this thread if there are no pending work items.
708 if not self.pending_work_items:
709 self.join_executor_internals()
710 return
712 def add_call_item_to_queue(self):
713 # Fills call_queue with _WorkItems from pending_work_items.
714 # This function never blocks.
715 while True:
716 if self.call_queue.full():
717 return
718 try:
719 work_id = self.work_ids_queue.get(block=False)
720 except queue.Empty:
721 return
722 else:
723 work_item = self.pending_work_items[work_id]
725 if work_item.future.set_running_or_notify_cancel():
726 self.running_work_items += [work_id]
727 self.call_queue.put(
728 _CallItem(
729 work_id,
730 work_item.fn,
731 work_item.args,
732 work_item.kwargs,
733 ),
734 block=True,
735 )
736 else:
737 del self.pending_work_items[work_id]
738 continue
740 def wait_result_broken_or_wakeup(self):
741 # Wait for a result to be ready in the result_queue while checking
742 # that all worker processes are still running, or for a wake up
743 # signal send. The wake up signals come either from new tasks being
744 # submitted, from the executor being shutdown/gc-ed, or from the
745 # shutdown of the python interpreter.
746 result_reader = self.result_queue._reader
747 wakeup_reader = self.thread_wakeup._reader
748 readers = [result_reader, wakeup_reader]
749 worker_sentinels = [p.sentinel for p in list(self.processes.values())]
750 ready = wait(readers + worker_sentinels)
752 bpe = None
753 is_broken = True
754 result_item = None
755 if result_reader in ready:
756 try:
757 result_item = result_reader.recv()
758 if isinstance(result_item, _RemoteTraceback):
759 bpe = BrokenProcessPool(
760 "A task has failed to un-serialize. Please ensure that"
761 " the arguments of the function are all picklable."
762 )
763 bpe.__cause__ = result_item
764 else:
765 is_broken = False
766 except BaseException as e:
767 bpe = BrokenProcessPool(
768 "A result has failed to un-serialize. Please ensure that "
769 "the objects returned by the function are always "
770 "picklable."
771 )
772 tb = traceback.format_exception(
773 type(e), e, getattr(e, "__traceback__", None)
774 )
775 bpe.__cause__ = _RemoteTraceback("".join(tb))
777 elif wakeup_reader in ready:
778 # This is simply a wake-up event that might either trigger putting
779 # more tasks in the queue or trigger the clean up of resources.
780 is_broken = False
781 else:
782 # A worker has terminated and we don't know why, set the state of
783 # the executor as broken
784 exit_codes = ""
785 if sys.platform != "win32":
786 # In Windows, introspecting terminated workers exitcodes seems
787 # unstable, therefore they are not appended in the exception
788 # message.
789 exit_codes = (
790 "\nThe exit codes of the workers are "
791 f"{get_exitcodes_terminated_worker(self.processes)}"
792 )
793 mp.util.debug(
794 "A worker unexpectedly terminated. Workers that "
795 "might have caused the breakage: "
796 + str(
797 {
798 p.name: p.exitcode
799 for p in list(self.processes.values())
800 if p is not None and p.sentinel in ready
801 }
802 )
803 )
804 bpe = TerminatedWorkerError(
805 "A worker process managed by the executor was unexpectedly "
806 "terminated. This could be caused by a segmentation fault "
807 "while calling the function or by an excessive memory usage "
808 "causing the Operating System to kill the worker.\n"
809 f"{exit_codes}\n"
810 "Detailed tracebacks of the workers should have been printed "
811 "to stderr in the executor process if faulthandler was not "
812 "disabled."
813 )
815 self.thread_wakeup.clear()
817 return result_item, is_broken, bpe
819 def process_result_item(self, result_item):
820 # Process the received a result_item. This can be either the PID of a
821 # worker that exited gracefully or a _ResultItem
823 if isinstance(result_item, int):
824 # Clean shutdown of a worker using its PID, either on request
825 # by the executor.shutdown method or by the timeout of the worker
826 # itself: we should not mark the executor as broken.
827 with self.processes_management_lock:
828 p = self.processes.pop(result_item, None)
830 # p can be None if the executor is concurrently shutting down.
831 if p is not None:
832 p._worker_exit_lock.release()
833 mp.util.debug(
834 f"joining {p.name} when processing {p.pid} as result_item"
835 )
836 p.join()
837 del p
839 # Make sure the executor have the right number of worker, even if a
840 # worker timeout while some jobs were submitted. If some work is
841 # pending or there is less processes than running items, we need to
842 # start a new Process and raise a warning.
843 n_pending = len(self.pending_work_items)
844 n_running = len(self.running_work_items)
845 if n_pending - n_running > 0 or n_running > len(self.processes):
846 executor = self.executor_reference()
847 if (
848 executor is not None
849 and len(self.processes) < executor._max_workers
850 ):
851 with executor._processes_management_lock:
852 executor._adjust_process_count()
853 executor = None
854 # Warn only once the pool is back to full strength, so
855 # that a filter turning this into an error costs no worker
856 if not isinstance(result_item, _RecycledWorkerPid):
857 warnings.warn(
858 "A worker stopped while some jobs were given to "
859 "the executor. This can be caused by a too short "
860 "worker timeout or by a memory leak triggered by the "
861 "OS-level out-of-memory (OOM) process killer.",
862 UserWarning,
863 )
864 elif not self.recycling_warned:
865 # Recycling is a deliberate decision of the executor and
866 # it is transparent to the caller, so repeating it for
867 # every worker of a long-running executor is just noise.
868 self.recycling_warned = True
869 warnings.warn(
870 "A worker was restarted while some jobs were given "
871 "to the executor because it was using more than "
872 f"{_MAX_MEMORY_LEAK_SIZE / 1e6:.0f} MB more memory "
873 "than when it started, which usually indicates a "
874 "memory leak. Further restarts of this executor "
875 "will not be reported.",
876 UserWarning,
877 )
878 else:
879 # Received a _ResultItem so mark the future as completed.
880 work_item = self.pending_work_items.pop(result_item.work_id, None)
881 # work_item can be None if another process terminated (see above)
882 if work_item is not None:
883 if result_item.exception is not None:
884 work_item.future.set_exception(result_item.exception)
885 else:
886 work_item.future.set_result(result_item.result)
887 self.running_work_items.remove(result_item.work_id)
889 def is_shutting_down(self):
890 # Check whether we should start shutting down the executor.
891 executor = self.executor_reference()
892 # No more work items can be added if:
893 # - The interpreter is shutting down OR
894 # - The executor that owns this thread is not broken AND
895 # * The executor that owns this worker has been collected OR
896 # * The executor that owns this worker has been shutdown.
897 # If the executor is broken, it should be detected in the next loop.
898 return _global_shutdown or (
899 (executor is None or self.executor_flags.shutdown)
900 and not self.executor_flags.broken
901 )
903 def terminate_broken(self, bpe):
904 # Terminate the executor because it is in a broken state. The bpe
905 # argument can be used to display more information on the error that
906 # lead the executor into becoming broken.
908 # Mark the process pool broken so that submits fail right now.
909 self.executor_flags.flag_as_broken(bpe)
911 # Mark pending tasks as failed.
912 for work_item in self.pending_work_items.values():
913 try:
914 work_item.future.set_exception(bpe)
915 except InvalidStateError:
916 # set_exception() fails if the future is cancelled: ignore it.
917 # Trying to check if the future is cancelled before calling
918 # set_exception() would leave a race condition if the future is
919 # cancelled between the check and set_exception().
920 pass
921 # Delete references to object. See issue16284
922 del work_item
923 self.pending_work_items.clear()
925 # Terminate remaining workers forcibly: the queues or their
926 # locks may be in a dirty state and block forever.
927 self.kill_workers(reason="broken executor")
929 # clean up resources
930 self.join_executor_internals()
932 def flag_executor_shutting_down(self):
933 # Flag the executor as shutting down and cancel remaining tasks if
934 # requested as early as possible if it is not gc-ed yet.
935 self.executor_flags.flag_as_shutting_down()
937 # Cancel pending work items if requested.
938 if self.executor_flags.kill_workers:
939 # Drain the queued ids first: add_call_item_to_queue runs again
940 # right after this and looks each of them up in pending_work_items
941 while True:
942 try:
943 self.work_ids_queue.get(block=False)
944 except queue.Empty:
945 break
946 while self.pending_work_items:
947 _, work_item = self.pending_work_items.popitem()
948 work_item.future.set_exception(
949 ShutdownExecutorError(
950 "The Executor was shutdown with `kill_workers=True` "
951 "before this job could complete."
952 )
953 )
954 del work_item
956 # Kill the remaining worker forcibly to no waste time joining them
957 self.kill_workers(reason="executor shutting down")
959 def kill_workers(self, reason=""):
960 # Terminate the remaining workers using SIGKILL. This function also
961 # terminates descendant workers of the children in case there is some
962 # nested parallelism.
963 while self.processes:
964 _, p = self.processes.popitem()
965 mp.util.debug(f"terminate process {p.name}, reason: {reason}")
966 try:
967 kill_process_tree(p)
968 except ProcessLookupError: # pragma: no cover
969 pass
971 def shutdown_workers(self):
972 # shutdown all workers in self.processes
974 # Create a list to avoid RuntimeError due to concurrent modification of
975 # processes. nb_children_alive is thus an upper bound. Also release the
976 # processes' _worker_exit_lock to accelerate the shutdown procedure, as
977 # there is no need for hand-shake here.
978 with self.processes_management_lock:
979 n_children_to_stop = 0
980 for p in list(self.processes.values()):
981 mp.util.debug(f"releasing worker exit lock on {p.name}")
982 p._worker_exit_lock.release()
983 n_children_to_stop += 1
985 mp.util.debug(f"found {n_children_to_stop} processes to stop")
987 # Send the right number of sentinels, to make sure all children are
988 # properly terminated. Do it with a mechanism that avoid hanging on
989 # Full queue when all workers have already been shutdown.
990 n_sentinels_sent = 0
991 cooldown_time = 0.001
992 while (
993 n_sentinels_sent < n_children_to_stop
994 and self.get_n_children_alive() > 0
995 ):
996 for _ in range(n_children_to_stop - n_sentinels_sent):
997 try:
998 self.call_queue.put_nowait(None)
999 n_sentinels_sent += 1
1000 except queue.Full as e:
1001 if cooldown_time > 5.0:
1002 mp.util.info(
1003 "failed to send all sentinels and exit with error."
1004 f"\ncall_queue size={self.call_queue._maxsize}; "
1005 f" full is {self.call_queue.full()}; "
1006 )
1007 raise e
1008 mp.util.info(
1009 "full call_queue prevented to send all sentinels at "
1010 "once, waiting..."
1011 )
1012 sleep(cooldown_time)
1013 cooldown_time *= 1.2
1014 break
1016 mp.util.debug(f"sent {n_sentinels_sent} sentinels to the call queue")
1018 def join_executor_internals(self):
1019 self.shutdown_workers()
1021 # Release the queue's resources as soon as possible. Flag the feeder
1022 # thread for clean exit to avoid having the crash detection thread flag
1023 # the Executor as broken during the shutdown. This is safe as either:
1024 # * We don't need to communicate with the workers anymore
1025 # * There is nothing left in the Queue buffer except None sentinels
1026 mp.util.debug("closing call_queue")
1027 self.call_queue.close()
1028 self.call_queue.join_thread()
1030 # Closing result_queue
1031 mp.util.debug("closing result_queue")
1032 self.result_queue.close()
1034 mp.util.debug("closing thread_wakeup")
1035 with self.shutdown_lock:
1036 self.thread_wakeup.close()
1038 # If .join() is not called on the created processes then
1039 # some ctx.Queue methods may deadlock on macOS.
1040 with self.processes_management_lock:
1041 mp.util.debug(f"joining {len(self.processes)} processes")
1042 n_joined_processes = 0
1043 while True:
1044 try:
1045 pid, p = self.processes.popitem()
1046 mp.util.debug(f"joining process {p.name} with pid {pid}")
1047 p.join()
1048 n_joined_processes += 1
1049 except KeyError:
1050 break
1052 mp.util.debug(
1053 "executor management thread clean shutdown of "
1054 f"{n_joined_processes} workers"
1055 )
1057 def get_n_children_alive(self):
1058 # This is an upper bound on the number of children alive.
1059 with self.processes_management_lock:
1060 return sum(p.is_alive() for p in list(self.processes.values()))
1063_system_limits_checked = False
1064_system_limited = None
1067def _check_system_limits():
1068 global _system_limits_checked, _system_limited
1069 if _system_limits_checked and _system_limited:
1070 raise NotImplementedError(_system_limited)
1071 _system_limits_checked = True
1072 try:
1073 nsems_max = os.sysconf("SC_SEM_NSEMS_MAX")
1074 except (AttributeError, ValueError):
1075 # sysconf not available or setting not available
1076 return
1077 if nsems_max == -1:
1078 # undetermined limit, assume that limit is determined
1079 # by available memory only
1080 return
1081 if nsems_max >= 256:
1082 # minimum number of semaphores available
1083 # according to POSIX
1084 return
1085 _system_limited = (
1086 f"system provides too few semaphores ({nsems_max} available, "
1087 "256 necessary)"
1088 )
1089 raise NotImplementedError(_system_limited)
1092def _chain_from_iterable_of_lists(iterable):
1093 """
1094 Specialized implementation of itertools.chain.from_iterable.
1095 Each item in *iterable* should be a list. This function is
1096 careful not to keep references to yielded objects.
1097 """
1098 for element in iterable:
1099 element.reverse()
1100 while element:
1101 yield element.pop()
1104def _check_max_depth(context):
1105 # Limit the maxmal recursion level
1106 global _CURRENT_DEPTH
1107 if context.get_start_method() == "fork" and _CURRENT_DEPTH > 0:
1108 raise LokyRecursionError(
1109 "Could not spawn extra nested processes at depth superior to "
1110 "MAX_DEPTH=1. It is not possible to increase this limit when "
1111 "using the 'fork' start method."
1112 )
1114 if 0 < MAX_DEPTH and _CURRENT_DEPTH + 1 > MAX_DEPTH:
1115 raise LokyRecursionError(
1116 "Could not spawn extra nested processes at depth superior to "
1117 f"MAX_DEPTH={MAX_DEPTH}. If this is intendend, you can change "
1118 "this limit with the LOKY_MAX_DEPTH environment variable."
1119 )
1122class LokyRecursionError(RuntimeError):
1123 """A process tries to spawn too many levels of nested processes."""
1126class BrokenProcessPool(_BPPException):
1127 """
1128 Raised when the executor is broken while a future was in the running state.
1129 The cause can an error raised when unpickling the task in the worker
1130 process or when unpickling the result value in the parent process. It can
1131 also be caused by a worker process being terminated unexpectedly.
1132 """
1135class TerminatedWorkerError(BrokenProcessPool):
1136 """
1137 Raised when a process in a ProcessPoolExecutor terminated abruptly
1138 while a future was in the running state.
1139 """
1142# Alias for backward compat (for code written for loky 1.1.4 and earlier). Do
1143# not use in new code.
1144BrokenExecutor = BrokenProcessPool
1147class ShutdownExecutorError(RuntimeError):
1148 """
1149 Raised when a ProcessPoolExecutor is shutdown while a future was in the
1150 running or pending state.
1151 """
1154class ProcessPoolExecutor(Executor):
1156 _at_exit = None
1158 def __init__(
1159 self,
1160 max_workers=None,
1161 job_reducers=None,
1162 result_reducers=None,
1163 timeout=None,
1164 context=None,
1165 initializer=None,
1166 initargs=(),
1167 env=None,
1168 ):
1169 """Initializes a new ProcessPoolExecutor instance.
1171 Args:
1172 max_workers: int, optional (default: cpu_count())
1173 The maximum number of processes that can be used to execute the
1174 given calls. If None or not given then as many worker processes
1175 will be created as the number of CPUs the current process
1176 can use.
1177 job_reducers, result_reducers: dict(type: reducer_func)
1178 Custom reducer for pickling the jobs and the results from the
1179 Executor. If only `job_reducers` is provided, `result_reducer`
1180 will use the same reducers
1181 timeout: int, optional (default: None)
1182 Idle workers exit after timeout seconds. If a new job is
1183 submitted after the timeout, the executor will start enough
1184 new Python processes to make sure the pool of workers is full.
1185 context: A multiprocessing context to launch the workers. This
1186 object should provide SimpleQueue, Queue and Process.
1187 initializer: An callable used to initialize worker processes.
1188 initargs: A tuple of arguments to pass to the initializer.
1189 env: A dict of environment variable to overwrite in the child
1190 process. The environment variables are set before any module is
1191 loaded. Note that this only works with the loky context.
1192 """
1193 _check_system_limits()
1195 if max_workers is None:
1196 self._max_workers = cpu_count()
1197 else:
1198 if max_workers <= 0:
1199 raise ValueError("max_workers must be greater than 0")
1200 self._max_workers = max_workers
1202 if (
1203 sys.platform == "win32"
1204 and self._max_workers > _MAX_WINDOWS_WORKERS
1205 ):
1206 warnings.warn(
1207 f"On Windows, max_workers cannot exceed {_MAX_WINDOWS_WORKERS} "
1208 "due to limitations of the operating system."
1209 )
1210 self._max_workers = _MAX_WINDOWS_WORKERS
1212 if context is None:
1213 context = get_context()
1214 self._context = context
1215 self._env = env
1217 self._initializer, self._initargs = _prepare_initializer(
1218 initializer, initargs
1219 )
1220 _check_max_depth(self._context)
1222 if result_reducers is None:
1223 result_reducers = job_reducers
1225 # Timeout
1226 self._timeout = timeout
1228 # Management thread
1229 self._executor_manager_thread = None
1231 # Map of pids to processes
1232 self._processes = {}
1234 # Internal variables of the ProcessPoolExecutor
1235 self._processes = {}
1236 self._queue_count = 0
1237 self._pending_work_items = {}
1238 self._running_work_items = []
1239 self._work_ids = queue.Queue()
1240 self._processes_management_lock = self._context.Lock()
1241 self._executor_manager_thread = None
1242 self._shutdown_lock = threading.Lock()
1244 # _ThreadWakeup is a communication channel used to interrupt the wait
1245 # of the main loop of executor_manager_thread from another thread (e.g.
1246 # when calling executor.submit or executor.shutdown). We do not use the
1247 # _result_queue to send wakeup signals to the executor_manager_thread
1248 # as it could result in a deadlock if a worker process dies with the
1249 # _result_queue write lock still acquired.
1250 #
1251 # _shutdown_lock must be locked to access _ThreadWakeup.wakeup.
1252 self._executor_manager_thread_wakeup = _ThreadWakeup()
1254 # Flag to hold the state of the Executor. This permits to introspect
1255 # the Executor state even once it has been garbage collected.
1256 self._flags = _ExecutorFlags(self._shutdown_lock)
1258 # Finally setup the queues for interprocess communication
1259 self._setup_queues(job_reducers, result_reducers)
1261 mp.util.debug("ProcessPoolExecutor is setup")
1263 def _setup_queues(self, job_reducers, result_reducers, queue_size=None):
1264 # Make the call queue slightly larger than the number of processes to
1265 # prevent the worker processes from idling. But don't make it too big
1266 # because futures in the call queue cannot be cancelled.
1267 if queue_size is None:
1268 queue_size = 2 * self._max_workers + EXTRA_QUEUED_CALLS
1269 self._call_queue = _SafeQueue(
1270 max_size=queue_size,
1271 pending_work_items=self._pending_work_items,
1272 running_work_items=self._running_work_items,
1273 thread_wakeup=self._executor_manager_thread_wakeup,
1274 shutdown_lock=self._shutdown_lock,
1275 executor_flags=self._flags,
1276 reducers=job_reducers,
1277 ctx=self._context,
1278 )
1279 # Killed worker processes can produce spurious "broken pipe"
1280 # tracebacks in the queue's own worker thread. But we detect killed
1281 # processes anyway, so silence the tracebacks.
1282 self._call_queue._ignore_epipe = True
1284 self._result_queue = SimpleQueue(
1285 reducers=result_reducers, ctx=self._context
1286 )
1288 def _start_executor_manager_thread(self):
1289 if self._executor_manager_thread is None:
1290 mp.util.debug("_start_executor_manager_thread called")
1292 # Start the processes so that their sentinels are known.
1293 self._executor_manager_thread = _ExecutorManagerThread(self)
1294 self._executor_manager_thread.start()
1296 # register this executor in a mechanism that ensures it will wakeup
1297 # when the interpreter is exiting.
1298 _threads_wakeups[self._executor_manager_thread] = (
1299 self._shutdown_lock,
1300 self._executor_manager_thread_wakeup,
1301 )
1303 global process_pool_executor_at_exit
1304 if process_pool_executor_at_exit is None:
1305 # Ensure that the _python_exit function will be called before
1306 # the multiprocessing.Queue._close finalizers which have an
1307 # exitpriority of 10.
1309 if sys.version_info < (3, 9):
1310 process_pool_executor_at_exit = mp.util.Finalize(
1311 None, _python_exit, exitpriority=20
1312 )
1313 else:
1314 process_pool_executor_at_exit = threading._register_atexit(
1315 _python_exit
1316 )
1318 def _adjust_process_count(self):
1319 while len(self._processes) < self._max_workers:
1320 worker_exit_lock = self._context.BoundedSemaphore(1)
1321 args = (
1322 self._call_queue,
1323 self._result_queue,
1324 self._initializer,
1325 self._initargs,
1326 self._processes_management_lock,
1327 self._timeout,
1328 worker_exit_lock,
1329 _CURRENT_DEPTH + 1,
1330 )
1331 worker_exit_lock.acquire()
1332 try:
1333 # Try to spawn the process with some environment variable to
1334 # overwrite but it only works with the loky context for now.
1335 p = self._context.Process(
1336 target=_process_worker, args=args, env=self._env
1337 )
1338 except TypeError:
1339 p = self._context.Process(target=_process_worker, args=args)
1340 p._worker_exit_lock = worker_exit_lock
1341 p.start()
1342 self._processes[p.pid] = p
1343 mp.util.debug(
1344 f"Adjusted process count to {self._max_workers}: "
1345 f"{[(p.name, pid) for pid, p in self._processes.items()]}"
1346 )
1348 def _ensure_executor_running(self):
1349 """ensures all workers and management thread are running"""
1350 with self._processes_management_lock:
1351 if len(self._processes) != self._max_workers:
1352 self._adjust_process_count()
1353 self._start_executor_manager_thread()
1355 def submit(self, fn, *args, **kwargs):
1356 with self._flags.shutdown_lock:
1357 if self._flags.broken is not None:
1358 raise self._flags.broken
1359 if self._flags.shutdown:
1360 raise ShutdownExecutorError(
1361 "cannot schedule new futures after shutdown"
1362 )
1364 # Cannot submit a new calls once the interpreter is shutting down.
1365 # This check avoids spawning new processes at exit.
1366 if _global_shutdown:
1367 raise RuntimeError(
1368 "cannot schedule new futures after interpreter shutdown"
1369 )
1371 f = Future()
1372 w = _WorkItem(f, fn, args, kwargs)
1374 self._pending_work_items[self._queue_count] = w
1375 self._work_ids.put(self._queue_count)
1376 self._queue_count += 1
1377 # Wake up queue management thread
1378 self._executor_manager_thread_wakeup.wakeup()
1380 self._ensure_executor_running()
1381 return f
1383 submit.__doc__ = Executor.submit.__doc__
1385 def map(self, fn, *iterables, **kwargs):
1386 """Returns an iterator equivalent to map(fn, iter).
1388 Args:
1389 fn: A callable that will take as many arguments as there are
1390 passed iterables.
1391 timeout: The maximum number of seconds to wait. If None, then there
1392 is no limit on the wait time.
1393 chunksize: If greater than one, the iterables will be chopped into
1394 chunks of size chunksize and submitted to the process pool.
1395 If set to one, the items in the list will be sent one at a
1396 time.
1398 Returns:
1399 An iterator equivalent to: map(func, *iterables) but the calls may
1400 be evaluated out-of-order.
1402 Raises:
1403 TimeoutError: If the entire result iterator could not be generated
1404 before the given timeout.
1405 Exception: If fn(*args) raises for any values.
1406 """
1407 timeout = kwargs.get("timeout", None)
1408 chunksize = kwargs.get("chunksize", 1)
1409 if chunksize < 1:
1410 raise ValueError("chunksize must be >= 1.")
1412 results = super().map(
1413 partial(_process_chunk, fn),
1414 _get_chunks(chunksize, *iterables),
1415 timeout=timeout,
1416 )
1417 return _chain_from_iterable_of_lists(results)
1419 def shutdown(self, wait=True, kill_workers=False):
1420 mp.util.debug(f"shutting down executor {self}")
1422 self._flags.flag_as_shutting_down(kill_workers)
1423 executor_manager_thread = self._executor_manager_thread
1424 executor_manager_thread_wakeup = self._executor_manager_thread_wakeup
1426 if executor_manager_thread_wakeup is not None:
1427 # Wake up queue management thread
1428 with self._shutdown_lock:
1429 self._executor_manager_thread_wakeup.wakeup()
1431 if executor_manager_thread is not None and wait:
1432 # This locks avoids concurrent join if the interpreter
1433 # is shutting down.
1434 with _global_shutdown_lock:
1435 executor_manager_thread.join()
1436 _threads_wakeups.pop(executor_manager_thread, None)
1438 # To reduce the risk of opening too many files, remove references to
1439 # objects that use file descriptors.
1440 self._executor_manager_thread = None
1441 self._executor_manager_thread_wakeup = None
1442 self._call_queue = None
1443 self._result_queue = None
1444 self._processes_management_lock = None
1446 shutdown.__doc__ = Executor.shutdown.__doc__