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
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 ) : 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.
Manages the communication between this process and the worker processes .