loky/process_executor.py
def _queue_management_worker(executor_reference,
                             executor_flags,
                             processes,
                             pending_work_items,
                             running_work_items,
                             work_ids_queue,
                             call_queue,
                             result_queue,
                             thread_wakeup,
                             processes_management_lock):
    """Manages the communication between this process and the worker processes.

    This function is run in a local thread.

    Args:
        executor_reference: A weakref.ref to the ProcessPoolExecutor that owns
            this thread. Used to determine if the ProcessPoolExecutor has been
            garbage collected and that this function can exit.
        executor_flags: A ExecutorFlags holding internal states of the
            ProcessPoolExecutor. It permits to know if the executor is broken
            even the object has been gc.
        process: A list of the ctx.Process instances used as
            workers.
        pending_work_items: A dict mapping work ids to _WorkItems e.g.
            {5: <_WorkItem...>, 6: <_WorkItem...>, ...}
        work_ids_queue: A queue.Queue of work ids e.g. Queue([5, 6, ...]).
        call_queue: A ctx.Queue that will be filled with _CallItems
            derived from _WorkItems for processing by the process workers.
        result_queue: A ctx.SimpleQueue of _ResultItems generated by the
            process workers.
        thread_wakeup: A _ThreadWakeup to allow waking up the
            queue_manager_thread from the main Thread and avoid deadlocks
            caused by permanently locked queues.
    """
    executor = None

    def is_shutting_down():
        # No more work items can be added if:
        #   - The interpreter is shutting down OR
        #   - The executor that own this worker is not broken AND
        #        * The executor that owns this worker has been collected OR
        #        * The executor that owns this worker has been shutdown.
        # If the executor is broken, it should be detected in the next loop.
        return (_global_shutdown or
                ((executor is None or executor_flags.shutdown)
                 and not executor_flags.broken))

    def shutdown_all_workers():
        mp.util.debug("queue management thread shutting down")
        executor_flags.flag_as_shutting_down()
        # Create a list to avoid RuntimeError due to concurrent modification of
        # processes. nb_children_alive is thus an upper bound. Also release the
        # processes' _worker_exit_lock to accelerate the shutdown procedure, as
        # there is no need for hand-shake here.
        with processes_management_lock:
            n_children_alive = 0
            for p in list(processes.values()):
                p._worker_exit_lock.release()
                n_children_alive += 1
        n_children_to_stop = 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 n_children_alive > 0:
            for i in range(n_children_to_stop - n_sentinels_sent):
                try:
                    call_queue.put_nowait(None)
                    n_sentinels_sent += 1
                except Full:
                    break
            with processes_management_lock:
                n_children_alive = sum(
                    p.is_alive() for p in list(processes.values())
                )

        # Release the queue's resources as soon as possible. Flag the feeder
        # thread for clean exit to avoid having the crash detection thread flag
        # the Executor as broken during the shutdown. This is safe as either:
        #  * We don't need to communicate with the workers anymore
        #  * There is nothing left in the Queue buffer except None sentinels
        mp.util.debug("closing call_queue")
        call_queue.close()

        mp.util.debug("joining processes")
        # If .join() is not called on the created processes then
        # some ctx.Queue methods may deadlock on Mac OS X.
        while processes:
            _, p = processes.popitem()
            p.join()
        mp.util.debug("queue management thread clean shutdown of worker "
                      "processes: {}".format(list(processes)))

    result_reader = result_queue._reader
    wakeup_reader = thread_wakeup._reader
    readers = [result_reader, wakeup_reader]

    while True:
        _add_call_item_to_queue(pending_work_items,
                                running_work_items,
                                work_ids_queue,
                                call_queue)
        # 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.
        worker_sentinels = [p.sentinel for p in list(processes.values())]
        ready = wait(readers + worker_sentinels)

        broken = ("A worker process managed by the executor was unexpectedly "
                  "terminated. This could be caused by a segmentation fault "
                  "while calling the function or by an excessive memory usage "
                  "causing the Operating System to kill the worker.", None,
                  TerminatedWorkerError)
        if result_reader in ready:
            try:
                result_item = result_reader.recv()
                broken = None
                if isinstance(result_item, _RemoteTraceback):
                    broken = ("A task has failed to un-serialize. Please "
                              "ensure that the arguments of the function are "
                              "all picklable.", result_item.tb,
                              BrokenProcessPool)
            except BaseException as e:
                tb = getattr(e, "__traceback__", None)
                if tb is None:
                    _, _, tb = sys.exc_info()
                broken = ("A result has failed to un-serialize. Please "
                          "ensure that the objects returned by the function "
                          "are always picklable.",
                          traceback.format_exception(type(e), e, tb),
                          BrokenProcessPool)
        elif wakeup_reader in ready:
            broken = None
            result_item = None
        thread_wakeup.clear()
        if broken is not None:
            msg, cause_tb, exc_type = broken
            if (issubclass(exc_type, TerminatedWorkerError) and
                    (sys.platform != "win32")):
                # In Windows, introspecting terminated workers exitcodes seems
                # unstable, therefore they are not appended in the exception
                # message.
                msg += " The exit codes of the workers are {}".format(
                    get_exitcodes_terminated_worker(processes))

            bpe = exc_type(msg)
            if cause_tb is not None:
                bpe = set_cause(bpe, _RemoteTraceback(
                          "\n'''\n{}'''".format(''.join(cause_tb))))
            # Mark the process pool broken so that submits fail right now.
            executor_flags.flag_as_broken(bpe)

            # All futures in flight must be marked failed
            for work_id, work_item in pending_work_items.items():
                work_item.future.set_exception(bpe)
                # Delete references to object. See issue16284
                del work_item
            pending_work_items.clear()

            # Terminate remaining workers forcibly: the queues or their
            # locks may be in a dirty state and block forever.
            while processes:
                _, p = processes.popitem()
                mp.util.debug('terminate process {}'.format(p.name))
                try:
                    recursive_terminate(p)
                except ProcessLookupError:  # pragma: no cover
                    pass

            shutdown_all_workers()
            return
        if isinstance(result_item, int):
            # Clean shutdown of a worker using its PID, either on request
            # by the executor.shutdown method or by the timeout of the worker
            # itself: we should not mark the executor as broken.
            with processes_management_lock:
                p = processes.pop(result_item, None)

            # p can be None is the executor is concurrently shutting down.
            if p is not None:
                p._worker_exit_lock.release()
                p.join()
                del p

            # Make sure the executor have the right number of worker, even if a
            # worker timeout while some jobs were submitted. If some work is
            # pending or there is less processes than running items, we need to
            # start a new Process and raise a warning.
            n_pending = len(pending_work_items)
            n_running = len(running_work_items)
            if (n_pending - n_running > 0 or n_running > len(processes)):
                executor = executor_reference()
                if (executor is not None
                        and len(processes) < executor._max_workers):
                    warnings.warn(
                        "A worker stopped while some jobs were given to the "
                        "executor. This can be caused by a too short worker "
                        "timeout or by a memory leak.", UserWarning
                    )
                    executor._adjust_process_count()
                    executor = None

        elif result_item is not None:
            work_item = pending_work_items.pop(result_item.work_id, None)
            # work_item can be None if another process terminated
            if work_item is not None:
                if result_item.exception:
                    work_item.future.set_exception(result_item.exception)
                else:
                    work_item.future.set_result(result_item.result)
                # Delete references to object. See issue16284
                del work_item
                running_work_items.remove(result_item.work_id)
            # Delete reference to result_item
            del result_item

        # Check whether we should start shutting down.
        executor = 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.
        if is_shutting_down():
            # bpo-33097: Make sure that the executor is flagged as shutting
            # down even if it is shutdown by the interpreter exiting.
            with executor_flags.shutdown_lock:
                executor_flags.shutdown = True
            if executor_flags.kill_workers:
                while pending_work_items:
                    _, work_item = pending_work_items.popitem()
                    work_item.future.set_exception(ShutdownExecutorError(
                        "The Executor was shutdown before this job could "
                        "complete."))
                    del work_item
                # Terminate remaining workers forcibly: the queues or their
                # locks may be in a dirty state and block forever.
                while processes:
                    _, p = processes.popitem()
                    recursive_terminate(p)
                shutdown_all_workers()
                return
            # Since no new work items can be added, it is safe to shutdown
            # this thread if there are no pending work items.
            if not pending_work_items:
                shutdown_all_workers()
                return
        elif executor_flags.broken:
            return
        executor = None
def _queue_management_worker(executor_reference,
                             executor_flags,
                             processes,
                             pending_work_items,
                             running_work_items,
                             work_ids_queue,
                             call_queue,
                             result_queue,
                             thread_wakeup,
                             processes_management_lock):
    """Manages the communication between this process and the worker processes.

    This function is run in a local thread.

    Args:
        executor_reference: A weakref.ref to the ProcessPoolExecutor that owns
            this thread. Used to determine if the ProcessPoolExecutor has been
            garbage collected and that this function can exit.
        executor_flags: A ExecutorFlags holding internal states of the
            ProcessPoolExecutor. It permits to know if the executor is broken
            even the object has been gc.
        process: A list of the ctx.Process instances used as
            workers.
        pending_work_items: A dict mapping work ids to _WorkItems e.g.
            {5: <_WorkItem...>, 6: <_WorkItem...>, ...}
        work_ids_queue: A queue.Queue of work ids e.g. Queue([5, 6, ...]).
        call_queue: A ctx.Queue that will be filled with _CallItems
            derived from _WorkItems for processing by the process workers.
        result_queue: A ctx.SimpleQueue of _ResultItems generated by the
            process workers.
        thread_wakeup: A _ThreadWakeup to allow waking up the
            queue_manager_thread from the main Thread and avoid deadlocks
            caused by permanently locked queues.
    """
    executor = None

    def is_shutting_down():
        # No more work items can be added if:
        #   - The interpreter is shutting down OR
        #   - The executor that own this worker is not broken AND
        #        * The executor that owns this worker has been collected OR
        #        * The executor that owns this worker has been shutdown.
        # If the executor is broken, it should be detected in the next loop.
        return (_global_shutdown or
                ((executor is None or executor_flags.shutdown)
                 and not executor_flags.broken))

    def shutdown_all_workers():
        mp.util.debug("queue management thread shutting down")
        executor_flags.flag_as_shutting_down()
        # Create a list to avoid RuntimeError due to concurrent modification of
        # processes. nb_children_alive is thus an upper bound. Also release the
        # processes' _worker_exit_lock to accelerate the shutdown procedure, as
        # there is no need for hand-shake here.
        with processes_management_lock:
            n_children_alive = 0
            for p in list(processes.values()):
                p._worker_exit_lock.release()
                n_children_alive += 1
        n_children_to_stop = 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 n_children_alive > 0:
            for i in range(n_children_to_stop - n_sentinels_sent):
                try:
                    call_queue.put_nowait(None)
                    n_sentinels_sent += 1
                except Full:
                    break
            with processes_management_lock:
                n_children_alive = sum(
                    p.is_alive() for p in list(processes.values())
                )

        # Release the queue's resources as soon as possible. Flag the feeder
        # thread for clean exit to avoid having the crash detection thread flag
        # the Executor as broken during the shutdown. This is safe as either:
        #  * We don't need to communicate with the workers anymore
        #  * There is nothing left in the Queue buffer except None sentinels
        mp.util.debug("closing call_queue")
        call_queue.close()

        mp.util.debug("joining processes")
        # If .join() is not called on the created processes then
        # some ctx.Queue methods may deadlock on Mac OS X.
        while processes:
            _, p = processes.popitem()
            p.join()
        mp.util.debug("queue management thread clean shutdown of worker "
                      "processes: {}".format(list(processes)))

    result_reader = result_queue._reader
    wakeup_reader = thread_wakeup._reader
    readers = [result_reader, wakeup_reader]

    while True:
        _add_call_item_to_queue(pending_work_items,
                                running_work_items,
                                work_ids_queue,
                                call_queue)
        # 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.
        worker_sentinels = [p.sentinel for p in list(processes.values())]
        ready = wait(readers + worker_sentinels)

        broken = ("A worker process managed by the executor was unexpectedly "
                  "terminated. This could be caused by a segmentation fault "
                  "while calling the function or by an excessive memory usage "
                  "causing the Operating System to kill the worker.", None,
                  TerminatedWorkerError)
        if result_reader in ready:
            try:
                result_item = result_reader.recv()
                broken = None
                if isinstance(result_item, _RemoteTraceback):
                    broken = ("A task has failed to un-serialize. Please "
                              "ensure that the arguments of the function are "
                              "all picklable.", result_item.tb,
                              BrokenProcessPool)
            except BaseException as e:
                tb = getattr(e, "__traceback__", None)
                if tb is None:
                    _, _, tb = sys.exc_info()
                broken = ("A result has failed to un-serialize. Please "
                          "ensure that the objects returned by the function "
                          "are always picklable.",
                          traceback.format_exception(type(e), e, tb),
                          BrokenProcessPool)
        elif wakeup_reader in ready:
            broken = None
            result_item = None
        thread_wakeup.clear()
        if broken is not None:
            msg, cause_tb, exc_type = broken
            if (issubclass(exc_type, TerminatedWorkerError) and
                    (sys.platform != "win32")):
                # In Windows, introspecting terminated workers exitcodes seems
                # unstable, therefore they are not appended in the exception
                # message.
                msg += " The exit codes of the workers are {}".format(
                    get_exitcodes_terminated_worker(processes))

            bpe = exc_type(msg)
            if cause_tb is not None:
                bpe = set_cause(bpe, _RemoteTraceback(
                          "\n'''\n{}'''".format(''.join(cause_tb))))
            # Mark the process pool broken so that submits fail right now.
            executor_flags.flag_as_broken(bpe)

            # All futures in flight must be marked failed
            for work_id, work_item in pending_work_items.items():
                work_item.future.set_exception(bpe)
                # Delete references to object. See issue16284
                del work_item
            pending_work_items.clear()

            # Terminate remaining workers forcibly: the queues or their
            # locks may be in a dirty state and block forever.
            while processes:
                _, p = processes.popitem()
                mp.util.debug('terminate process {}'.format(p.name))
                try:
                    recursive_terminate(p)
                except ProcessLookupError:  # pragma: no cover
                    pass

            shutdown_all_workers()
            return
        if isinstance(result_item, int):
            # Clean shutdown of a worker using its PID, either on request
            # by the executor.shutdown method or by the timeout of the worker
            # itself: we should not mark the executor as broken.
            with processes_management_lock:
                p = processes.pop(result_item, None)

            # p can be None is the executor is concurrently shutting down.
            if p is not None:
                p._worker_exit_lock.release()
                p.join()
                del p

            # Make sure the executor have the right number of worker, even if a
            # worker timeout while some jobs were submitted. If some work is
            # pending or there is less processes than running items, we need to
            # start a new Process and raise a warning.
            n_pending = len(pending_work_items)
            n_running = len(running_work_items)
            if (n_pending - n_running > 0 or n_running > len(processes)):
                executor = executor_reference()
                if (executor is not None
                        and len(processes) < executor._max_workers):
                    warnings.warn(
                        "A worker stopped while some jobs were given to the "
                        "executor. This can be caused by a too short worker "
                        "timeout or by a memory leak.", UserWarning
                    )
                    executor._adjust_process_count()
                    executor = None

        elif result_item is not None:
            work_item = pending_work_items.pop(result_item.work_id, None)
            # work_item can be None if another process terminated
            if work_item is not None:
                if result_item.exception:
                    work_item.future.set_exception(result_item.exception)
                else:
                    work_item.future.set_result(result_item.result)
                # Delete references to object. See issue16284
                del work_item
                running_work_items.remove(result_item.work_id)
            # Delete reference to result_item
            del result_item

        # Check whether we should start shutting down.
        executor = 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.
        if is_shutting_down():
            # bpo-33097: Make sure that the executor is flagged as shutting
            # down even if it is shutdown by the interpreter exiting.
            with executor_flags.shutdown_lock:
                executor_flags.shutdown = True
            if executor_flags.kill_workers:
                while pending_work_items:
                    _, work_item = pending_work_items.popitem()
                    work_item.future.set_exception(ShutdownExecutorError(
                        "The Executor was shutdown before this job could "
                        "complete."))
                    del work_item
                # Terminate remaining workers forcibly: the queues or their
                # locks may be in a dirty state and block forever.
                while processes:
                    _, p = processes.popitem()
                    recursive_terminate(p)
                shutdown_all_workers()
                return
            # Since no new work items can be added, it is safe to shutdown
            # this thread if there are no pending work items.
            if not pending_work_items:
                shutdown_all_workers()
                return
        elif executor_flags.broken:
            return
        executor = None
Manages the communication between this process and the worker processes.

    This function is run in a local thread.

    Args:
        executor_reference: A weakref.ref to the ProcessPoolExecutor that owns
            this thread. Used to determine if the ProcessPoolExecutor has been
            garbage collected and that this function can exit.
        executor_flags: A ExecutorFlags holding internal states of the
            ProcessPoolExecutor. It permits to know if the executor is broken
            even the object has been gc.
        process: A list of the ctx.Process instances used as
            workers.
        pending_work_items: A dict mapping work ids to _WorkItems e.g.
            {5: <_WorkItem...>, 6: <_WorkItem...>, ...}
        work_ids_queue: A queue.Queue of work ids e.g. Queue([5, 6, ...]).
        call_queue: A ctx.Queue that will be filled with _CallItems
            derived from _WorkItems for processing by the process workers.
        result_queue: A ctx.SimpleQueue of _ResultItems generated by the
            process workers.
        thread_wakeup: A _ThreadWakeup to allow waking up the
            queue_manager_thread from the main Thread and avoid deadlocks
            caused by permanently locked queues.