def __init__(self):

self._closed = False

self._lock = threading.Lock()

self._reader, self._writer = mp.Pipe(duplex=False)

def close(self):

# Please note that we do not take the self._lock when

# calling clear() (to avoid deadlocking) so this method can

# only be called safely from the same thread as all calls to

# clear() even if you hold the lock. Otherwise we

# might try to read from the closed pipe.

with self._lock:

if not self._closed:

self._closed = True

self._writer.close()

self._reader.close()

def wakeup(self):

with self._lock:

if not self._closed:

self._writer.send_bytes(b"")

def clear(self):

if self._closed:

raise RuntimeError('operation on closed _ThreadWakeup')

while self._reader.poll():

global _global_shutdown

_global_shutdown = True

items = list(_threads_wakeups.items())

for _, thread_wakeup in items:

# call not protected by ProcessPoolExecutor._shutdown_lock

thread_wakeup.wakeup()

for t, _ in items:

def __init__(self, tb):

self.tb = tb

def __str__(self):

def __init__(self, exc, tb):

tb = ''.join(format_exception(type(exc), exc, tb))

self.exc = exc

# Traceback object needs to be garbage-collected as its frames

# contain references to all the objects in the exception scope

self.exc.__traceback__ = None

self.tb = '\n"""\n%s"""' % tb

def __reduce__(self):

def __init__(self, future, fn, args, kwargs):

self.future = future

self.fn = fn

self.args = args

def __init__(self, work_id, exception=None, result=None, exit_pid=None):

self.work_id = work_id

self.exception = exception

self.result = result

def __init__(self, work_id, fn, args, kwargs):

self.work_id = work_id

self.fn = fn

self.args = args

"""Safe Queue set exception to the future object linked to a job"""

def __init__(self, max_size=0, *, ctx, pending_work_items, thread_wakeup):

self.pending_work_items = pending_work_items

self.thread_wakeup = thread_wakeup

super().__init__(max_size, ctx=ctx)

def _on_queue_feeder_error(self, e, obj):

if isinstance(obj, _CallItem):

tb = format_exception(type(e), e, e.__traceback__)

e.__cause__ = _RemoteTraceback('\n"""\n{}"""'.format(''.join(tb)))

work_item = self.pending_work_items.pop(obj.work_id, None)

self.thread_wakeup.wakeup()

# work_item can be None if another process terminated. In this

# case, the executor_manager_thread fails all work_items

# with BrokenProcessPool

if work_item is not None:

work_item.future.set_exception(e)

else:

""" Processes a chunk of an iterable passed to map.

exit_pid=None):

"""Safely send back the given result or exception"""

try:

result_queue.put(_ResultItem(work_id, result=result,

exception=exception, exit_pid=exit_pid))

except BaseException as e:

exc = _ExceptionWithTraceback(e, e.__traceback__)

result_queue.put(_ResultItem(work_id, exception=exc,

"""Evaluates calls from call_queue and places the results in result_queue.

if initializer is not None:

try:

initializer(*initargs)

except BaseException:

_base.LOGGER.critical('Exception in initializer:', exc_info=True)

# The parent will notice that the process stopped and

# mark the pool broken

return

num_tasks = 0

exit_pid = None

while True:

call_item = call_queue.get(block=True)

if call_item is None:

# Wake up queue management thread

result_queue.put(os.getpid())

return

if max_tasks is not None:

num_tasks += 1

if num_tasks >= max_tasks:

exit_pid = os.getpid()

try:

r = call_item.fn(*call_item.args, **call_item.kwargs)

except BaseException as e:

exc = _ExceptionWithTraceback(e, e.__traceback__)

_sendback_result(result_queue, call_item.work_id, exception=exc,

exit_pid=exit_pid)

else:

_sendback_result(result_queue, call_item.work_id, result=r,

exit_pid=exit_pid)

del r

# Liberate the resource as soon as possible, to avoid holding onto

# open files or shared memory that is not needed anymore

del call_item

if exit_pid is not None:

initargs, max_tasks_per_child, processes):

"""Start one worker process and record it in *processes* by pid."""

p = mp_context.Process(

target=_process_worker,

args=(call_queue,

result_queue,

initializer,

initargs,

max_tasks_per_child))

p.start()

"""Manages the communication between this process and the worker processes.

def __init__(self, executor):

# Store references to necessary internals of the executor.

# A _ThreadWakeup to allow waking up the queue_manager_thread from the

# main Thread and avoid deadlocks caused by permanently locked queues.

self.thread_wakeup = executor._executor_manager_thread_wakeup

self.shutdown_lock = executor._shutdown_lock

# A weakref.ref to the ProcessPoolExecutor that owns this thread. Used

# to determine if the ProcessPoolExecutor has been garbage collected

# and that the manager can exit.

# When the executor gets garbage collected, the weakref callback

# will wake up the queue management thread so that it can terminate

# if there is no pending work item.

def weakref_cb(_,

thread_wakeup=self.thread_wakeup,

mp_util_debug=mp.util.debug):

mp_util_debug('Executor collected: triggering callback for'

' QueueManager wakeup')

thread_wakeup.wakeup()

self.executor_reference = weakref.ref(executor, weakref_cb)

# A list of the ctx.Process instances used as workers.

self.processes = executor._processes

# A ctx.Queue that will be filled with _CallItems derived from

# _WorkItems for processing by the process workers.

self.call_queue = executor._call_queue

# A ctx.SimpleQueue of _ResultItems generated by the process workers.

self.result_queue = executor._result_queue

# A queue.Queue of work ids e.g. Queue([5, 6, ...]).

self.work_ids_queue = executor._work_ids

# Maximum number of tasks a worker process can execute before

# exiting safely

self.max_tasks_per_child = executor._max_tasks_per_child

# gh-119592: Needed to size worker replacement, and immutable, so

# keep a copy rather than reading it back through the executor

# weakref. The rest of the spawn configuration is deliberately NOT

# copied here: holding user-provided objects (initializer,

# initargs, mp_context) in this always-reachable running thread

# could keep the executor itself reachable through them, breaking

# garbage-collection-triggered shutdown.

self.max_workers = executor._max_workers

# A dict mapping work ids to _WorkItems e.g.

# {5: <_WorkItem...>, 6: <_WorkItem...>, ...}

self.pending_work_items = executor._pending_work_items

super().__init__()

def run(self):

# Main loop for the executor manager thread.

while True:

# gh-109047: During Python finalization, self.call_queue.put()

# creation of a thread can fail with RuntimeError.

try:

self.add_call_item_to_queue()

except BaseException as exc:

cause = format_exception(exc)

self.terminate_broken(cause)

return

result_item, is_broken, cause = self.wait_result_broken_or_wakeup()

if is_broken:

self.terminate_broken(cause)

return

if result_item is not None:

self.process_result_item(result_item)

process_exited = result_item.exit_pid is not None

if process_exited:

p = self.processes.pop(result_item.exit_pid)

p.join()

# Delete reference to result_item to avoid keeping references

# while waiting on new results.

del result_item

if process_exited:

with self.shutdown_lock:

broken = self._replace_dead_worker()

if broken is not None:

self.terminate_broken(*broken)

return

elif executor := self.executor_reference():

executor._idle_worker_semaphore.release()

del executor

if self.is_shutting_down():

self.flag_executor_shutting_down()

# When only canceled futures remain in pending_work_items, our

# next call to wait_result_broken_or_wakeup would hang forever.

# This makes sure we have some running futures or none at all.

self.add_call_item_to_queue()

# Since no new work items can be added, it is safe to shutdown

# this thread if there are no pending work items.

if not self.pending_work_items:

self.join_executor_internals()

return

def _replace_dead_worker(self):

"""Spawn a replacement for a worker that exited at its

assert self.shutdown_lock.locked()

cause = None

message = None

executor = self.executor_reference()

if executor is None:

# gh-152967: The executor was garbage collected; nothing can

# spawn a replacement worker for it anymore.

message = ("The ProcessPoolExecutor was garbage collected with "

"work pending after its last worker process exited "

"upon reaching max_tasks_per_child; the pending work "

"can never be run.")

elif executor._force_shutting_down:

# terminate_workers()/kill_workers() is tearing the pool down;

# a replacement worker would escape the kill and run work

# items that were enqueued before it.

message = ("A worker process exited while the pool was being "

"forcefully shut down; work that was still enqueued "

"will not be run.")

elif self.pending_work_items or not self.is_shutting_down():

# gh-115634: Do not consult the executor's

# _idle_worker_semaphore here: it counts task completions, not

# idle workers, so it can hold a stale token released by the

# now-dead worker. Trusting such a token would leave the pool

# a worker short, deadlocking once all workers reach their

# task limit. Spawning from this (manager) thread is safe

# despite gh-90622 because max_tasks_per_child is rejected for

# the "fork" start method.

if len(self.processes) < self.max_workers:

# gh-119592: Spawn using state owned by this thread and

# configuration read through the live weakref (which

# shutdown() never clears), not the executor state that

# shutdown(wait=False) clears concurrently.

try:

_spawn_worker(executor._mp_context, self.call_queue,

self.result_queue, executor._initializer,

executor._initargs,

self.max_tasks_per_child, self.processes)

except Exception as exc:

# While other workers remain the pool has merely lost

# capacity and they keep draining the queue; with none

# left the failure is reported below.

cause = format_exception(exc)

message = ("A replacement worker process could not be "

"started, leaving the pool without workers "

"to run the remaining work.")

del executor

if not self.processes and (self.pending_work_items

or cause is not None):

# No worker processes remain and no replacement can be

# spawned: any remaining work items can never run. A spawn

# failure breaks the pool even with nothing pending; leaving

# a zero-worker pool alive would hang a later submit() on a

# stale _idle_worker_semaphore token instead of raising.

return (cause, message)

return None

def add_call_item_to_queue(self):

# Fills call_queue with _WorkItems from pending_work_items.

# This function never blocks.

while True:

if self.call_queue.full():

return

try:

work_id = self.work_ids_queue.get(block=False)

except queue.Empty:

return

else:

work_item = self.pending_work_items[work_id]

if work_item.future.set_running_or_notify_cancel():

self.call_queue.put(_CallItem(work_id,

work_item.fn,

work_item.args,

work_item.kwargs),

block=True)

else:

del self.pending_work_items[work_id]

continue

def wait_result_broken_or_wakeup(self):

# Wait for a result to be ready in the result_queue while checking

# that all worker processes are still running, or for a wake up

# signal send. The wake up signals come either from new tasks being

# submitted, from the executor being shutdown/gc-ed, or from the

# shutdown of the python interpreter.

result_reader = self.result_queue._reader

assert not self.thread_wakeup._closed

wakeup_reader = self.thread_wakeup._reader

readers = [result_reader, wakeup_reader]

worker_sentinels = [p.sentinel for p in list(self.processes.values())]

ready = mp.connection.wait(readers + worker_sentinels)

cause = None

is_broken = True

result_item = None

if result_reader in ready:

try:

result_item = result_reader.recv()

is_broken = False

except BaseException as exc:

cause = format_exception(exc)

elif wakeup_reader in ready:

is_broken = False

self.thread_wakeup.clear()

return result_item, is_broken, cause

def process_result_item(self, result_item):

# Process the received a result_item. This can be either the PID of a

# worker that exited gracefully or a _ResultItem

# Received a _ResultItem so mark the future as completed.

work_item = self.pending_work_items.pop(result_item.work_id, None)

# work_item can be None if another process terminated (see above)

if work_item is not None:

if result_item.exception is not None:

work_item.future.set_exception(result_item.exception)

else:

work_item.future.set_result(result_item.result)

def is_shutting_down(self):

# Check whether we should start shutting down the executor.

executor = self.executor_reference()

# No more work items can be added if:

# - The interpreter is shutting down OR

# - The executor that owns this worker has been collected OR

# - The executor that owns this worker has been shutdown.

return (_global_shutdown or executor is None

or executor._shutdown_thread)

def _terminate_broken(self, cause, bpe_message=None):

# Terminate the executor because it is in a broken state. The cause

# argument can be used to display more information on the error that

# lead the executor into becoming broken. bpe_message overrides the

# default message on the BrokenProcessPool set on pending futures.

# Mark the process pool broken so that submits fail right now.

executor = self.executor_reference()

if executor is not None:

executor._broken = ('A child process terminated '

'abruptly, the process pool is not '

'usable anymore')

executor._shutdown_thread = True

executor = None

# All pending tasks are to be marked failed with a

# BrokenProcessPool error, as separate instances to avoid sharing

# a traceback (gh-101267).

cause_tb = None

if cause is not None:

cause_tb = f"\n'''\n{''.join(cause)}'''"

if bpe_message is None:

bpe_message = ("A process in the process pool was terminated "

"abruptly while the future was running or pending.")

# Mark pending tasks as failed.

for work_id, work_item in self.pending_work_items.items():

bpe = BrokenProcessPool(bpe_message)

if cause_tb is not None:

bpe.__cause__ = _RemoteTraceback(cause_tb)

try:

work_item.future.set_exception(bpe)

except _base.InvalidStateError:

# set_exception() fails if the future is cancelled: ignore it.

# Trying to check if the future is cancelled before calling

# set_exception() would leave a race condition if the future is

# cancelled between the check and set_exception().

pass

# Delete references to object. See issue16284

del work_item

self.pending_work_items.clear()

# Terminate remaining workers forcibly: the queues or their

# locks may be in a dirty state and block forever.

for p in self.processes.values():

p.terminate()

self.call_queue._terminate_broken()

# clean up resources

self._join_executor_internals(broken=True)

def terminate_broken(self, cause, bpe_message=None):

with self.shutdown_lock:

self._terminate_broken(cause, bpe_message)

def flag_executor_shutting_down(self):

# Flag the executor as shutting down and cancel remaining tasks if

# requested as early as possible if it is not gc-ed yet.

executor = self.executor_reference()

if executor is not None:

executor._shutdown_thread = True

# Cancel pending work items if requested.

if executor._cancel_pending_futures:

# Cancel all pending futures and update pending_work_items

# to only have futures that are currently running.

new_pending_work_items = {}

for work_id, work_item in self.pending_work_items.items():

if not work_item.future.cancel():

new_pending_work_items[work_id] = work_item

self.pending_work_items = new_pending_work_items

# Drain work_ids_queue since we no longer need to

# add items to the call queue.

while True:

try:

self.work_ids_queue.get_nowait()

except queue.Empty:

break

# Make sure we do this only once to not waste time looping

# on running processes over and over.

executor._cancel_pending_futures = False

def shutdown_workers(self):

n_children_to_stop = self.get_n_children_alive()

n_sentinels_sent = 0

# Send the right number of sentinels, to make sure all children are

# properly terminated.

while (n_sentinels_sent < n_children_to_stop

and self.get_n_children_alive() > 0):

for i in range(n_children_to_stop - n_sentinels_sent):

try:

self.call_queue.put_nowait(None)

n_sentinels_sent += 1

except queue.Full:

break

def join_executor_internals(self):

with self.shutdown_lock:

self._join_executor_internals()

def _join_executor_internals(self, broken=False):

# If broken, call_queue was closed and so can no longer be used.

if not broken:

self.shutdown_workers()

# Release the queue's resources as soon as possible.

self.call_queue.close()

self.call_queue.join_thread()

self.thread_wakeup.close()

# If .join() is not called on the created processes then

# some ctx.Queue methods may deadlock on Mac OS X.

for p in self.processes.values():

if broken:

p.terminate()

p.join()

def get_n_children_alive(self):

# This is an upper bound on the number of children alive.

global _system_limits_checked, _system_limited

if _system_limits_checked:

if _system_limited:

raise NotImplementedError(_system_limited)

_system_limits_checked = True

try:

import multiprocessing.synchronize # noqa: F401

except ImportError:

_system_limited = (

"This Python build lacks multiprocessing.synchronize, usually due "

"to named semaphores being unavailable on this platform."

)

raise NotImplementedError(_system_limited)

try:

nsems_max = os.sysconf("SC_SEM_NSEMS_MAX")

except (AttributeError, ValueError):

# sysconf not available or setting not available

return

if nsems_max == -1:

# indetermined limit, assume that limit is determined

# by available memory only

return

if nsems_max >= 256:

# minimum number of semaphores available

# according to POSIX

return

_system_limited = ("system provides too few semaphores (%d"

" available, 256 necessary)" % nsems_max)

"""

for element in iterable:

element.reverse()

while element:

"""

def __init__(self, max_workers=None, mp_context=None,

initializer=None, initargs=(), *, max_tasks_per_child=None):

"""Initializes a new ProcessPoolExecutor instance.

_check_system_limits()

if max_workers is None:

self._max_workers = os.process_cpu_count() or 1

if sys.platform == 'win32':

self._max_workers = min(_MAX_WINDOWS_WORKERS,

self._max_workers)

else:

if max_workers <= 0:

raise ValueError("max_workers must be greater than 0")

elif (sys.platform == 'win32' and

max_workers > _MAX_WINDOWS_WORKERS):

raise ValueError(

f"max_workers must be <= {_MAX_WINDOWS_WORKERS}")

self._max_workers = max_workers

if mp_context is None:

if max_tasks_per_child is not None:

mp_context = mp.get_context("spawn")

else:

mp_context = mp.get_context()

self._mp_context = mp_context

# https://github.com/python/cpython/issues/90622

self._safe_to_dynamically_spawn_children = (

self._mp_context.get_start_method(allow_none=False) != "fork")

if initializer is not None and not callable(initializer):

raise TypeError("initializer must be a callable")

self._initializer = initializer

self._initargs = initargs

if max_tasks_per_child is not None:

if not isinstance(max_tasks_per_child, int):

raise TypeError("max_tasks_per_child must be an integer")

elif max_tasks_per_child <= 0:

raise ValueError("max_tasks_per_child must be >= 1")

if self._mp_context.get_start_method(allow_none=False) == "fork":

# https://github.com/python/cpython/issues/90622

raise ValueError("max_tasks_per_child is incompatible with"

" the 'fork' multiprocessing start method;"

" supply a different mp_context.")

self._max_tasks_per_child = max_tasks_per_child

# Management thread

self._executor_manager_thread = None

# Map of pids to processes

self._processes = {}

# Shutdown is a two-step process.

self._shutdown_thread = False

self._shutdown_lock = threading.Lock()

self._idle_worker_semaphore = threading.Semaphore(0)

self._broken = False

self._queue_count = 0

self._pending_work_items = {}

self._cancel_pending_futures = False

self._force_shutting_down = False

# _ThreadWakeup is a communication channel used to interrupt the wait

# of the main loop of executor_manager_thread from another thread (e.g.

# when calling executor.submit or executor.shutdown). We do not use the

# _result_queue to send wakeup signals to the executor_manager_thread

# as it could result in a deadlock if a worker process dies with the

# _result_queue write lock still acquired.

#

# Care must be taken to only call clear and close from the

# executor_manager_thread, since _ThreadWakeup.clear() is not protected

# by a lock.

self._executor_manager_thread_wakeup = _ThreadWakeup()

# Create communication channels for the executor

# Make the call queue slightly larger than the number of processes to

# prevent the worker processes from idling. But don't make it too big

# because futures in the call queue cannot be cancelled.

queue_size = self._max_workers + EXTRA_QUEUED_CALLS

self._call_queue = _SafeQueue(

max_size=queue_size, ctx=self._mp_context,

pending_work_items=self._pending_work_items,

thread_wakeup=self._executor_manager_thread_wakeup)

# Killed worker processes can produce spurious "broken pipe"

# tracebacks in the queue's own worker thread. But we detect killed

# processes anyway, so silence the tracebacks.

self._call_queue._ignore_epipe = True

self._result_queue = mp_context.SimpleQueue()

self._work_ids = queue.Queue()

def _start_executor_manager_thread(self):

if self._executor_manager_thread is None:

# Start the processes so that their sentinels are known.

if not self._safe_to_dynamically_spawn_children: # ie, using fork.

self._launch_processes()

self._executor_manager_thread = _ExecutorManagerThread(self)

self._executor_manager_thread.start()

_threads_wakeups[self._executor_manager_thread] = \

self._executor_manager_thread_wakeup

def _adjust_process_count(self):

# gh-132969: avoid error when state is reset and executor is still running,

# which will happen when shutdown(wait=False) is called.

if self._processes is None:

return

# gh-152967: A forceful shutdown is in progress; a worker spawned

# here could escape its process snapshot and keep running work.

if self._force_shutting_down:

return

# if there's an idle process, we don't need to spawn a new one.

if self._idle_worker_semaphore.acquire(blocking=False):

return

process_count = len(self._processes)

if process_count < self._max_workers:

# gh-90622: spawning a child via fork while another thread is

# running can deadlock in the child. submit() only calls this

# method when using a non-fork start method.

assert (self._safe_to_dynamically_spawn_children

or not self._executor_manager_thread), (

'https://github.com/python/cpython/issues/90622')

self._spawn_process()

def _launch_processes(self):

# https://github.com/python/cpython/issues/90622

assert not self._executor_manager_thread, (

'Processes cannot be fork()ed after the thread has started, '

'deadlock in the child processes could result.')

for _ in range(len(self._processes), self._max_workers):

self._spawn_process()

def _spawn_process(self):

_spawn_worker(self._mp_context, self._call_queue,

self._result_queue, self._initializer,

self._initargs, self._max_tasks_per_child,

self._processes)

def submit(self, fn, /, *args, **kwargs):

with self._shutdown_lock:

if self._broken:

raise BrokenProcessPool(self._broken)

if self._shutdown_thread:

raise RuntimeError('cannot schedule new futures after shutdown')

if _global_shutdown:

raise RuntimeError('cannot schedule new futures after '

'interpreter shutdown')

f = _base.Future()

w = _WorkItem(f, fn, args, kwargs)

self._pending_work_items[self._queue_count] = w

self._work_ids.put(self._queue_count)

self._queue_count += 1

# Wake up queue management thread

self._executor_manager_thread_wakeup.wakeup()

if self._safe_to_dynamically_spawn_children:

self._adjust_process_count()

self._start_executor_manager_thread()

return f

submit.__doc__ = _base.Executor.submit.__doc__

def map(self, fn, *iterables, timeout=None, chunksize=1, buffersize=None):

"""Returns an iterator equivalent to map(fn, iter).

if chunksize < 1:

raise ValueError("chunksize must be >= 1.")

results = super().map(partial(_process_chunk, fn),

itertools.batched(zip(*iterables), chunksize),

timeout=timeout,

buffersize=buffersize)

return _chain_from_iterable_of_lists(results)

def shutdown(self, wait=True, *, cancel_futures=False):

with self._shutdown_lock:

self._cancel_pending_futures = cancel_futures

self._shutdown_thread = True

if self._executor_manager_thread_wakeup is not None:

# Wake up queue management thread

self._executor_manager_thread_wakeup.wakeup()

if self._executor_manager_thread is not None and wait:

self._executor_manager_thread.join()

# To reduce the risk of opening too many files, remove references to

# objects that use file descriptors.

self._executor_manager_thread = None

self._call_queue = None

if self._result_queue is not None and wait:

self._result_queue.close()

self._result_queue = None

self._processes = None

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

if operation not in _SHUTDOWN_CALLBACK_OPERATION:

raise ValueError(f"Unsupported operation: {operation!r}")

# gh-152967: Stop the manager thread from spawning replacement

# workers before we copy the processes to signal: a worker spawned

# after the copy would survive the loop below and run enqueued

# work items. Taking the lock orders this against the manager's

# worker replacement, which runs under the same lock.

with self._shutdown_lock:

self._force_shutting_down = True

processes = {}

if self._processes:

processes = self._processes.copy()