def _queue_manangement_worker(executor_reference,
                              processes,
                              pending_work_items,
                              work_ids_queue,
                              call_queue,
                              result_queue,
                              shutdown_process_event):
    """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.
        process: A list of the multiprocessing.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 multiprocessing.Queue that will be filled with _CallItems
            derived from _WorkItems for processing by the process workers.
        result_queue: A multiprocessing.Queue of _ResultItems generated by the
            process workers.
        shutdown_process_event: A multiprocessing.Event used to signal the
            process workers that they should exit when their work queue is
            empty.
    """
    while True:
        _add_call_item_to_queue(pending_work_items,
                                work_ids_queue,
                                call_queue)

        try:
            result_item = result_queue.get(block=True, timeout=0.1)
        except queue.Empty:
            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 _shutdown or executor is None or executor._shutdown_thread:
                # 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_process_event.set()

                    # If .join() is not called on the created processes then
                    # some multiprocessing.Queue methods may deadlock on Mac OSX.
                    for p in processes:
                        p.join()
                    return
            del executor
        else:
            work_item = pending_work_items.pop(result_item.work_id)
            loop = work_item.loop
            loop.call_from_thread(_set_work_result, work_item, result_item)
            del result_item, work_item, loop