def _queue_management_worker(executor_reference, processes, pending_work_items, work_ids_queue, call_queue, result_queue):
 executor = None
 def shutting_down():
  return (_shutdown or (executor is None) or executor._shutdown_thread)
 def shutdown_worker():
  nb_children_alive = sum((p.is_alive() for p in processes.values()))
  for i in range(0, nb_children_alive):
   call_queue.put_nowait(None)
  call_queue.close()
  for p in processes.values():
   p.join()
 reader = result_queue._reader
 while True:
  _add_call_item_to_queue(pending_work_items, work_ids_queue, call_queue)
  sentinels = [p.sentinel for p in processes.values()]
  assert sentinels
  ready = wait(([reader] + sentinels))
  if (reader in ready):
   result_item = reader.recv()
  else:
   executor = executor_reference()
   if (executor is not None):
    executor._broken = True
    executor._shutdown_thread = True
    executor = None
   for (work_id, work_item) in pending_work_items.items():
    work_item.future.set_exception(BrokenProcessPool('A process in the process pool was terminated abruptly while the future was running or pending.'))
    del work_item
   pending_work_items.clear()
   for p in processes.values():
    p.terminate()
   shutdown_worker()
   return
  if isinstance(result_item, int):
   assert shutting_down()
   p = processes.pop(result_item)
   p.join()
   if (not processes):
    shutdown_worker()
    return
  elif (result_item is not None):
   work_item = pending_work_items.pop(result_item.work_id, None)
   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)
    del work_item
  executor = executor_reference()
  if shutting_down():
   try:
    if (not pending_work_items):
     shutdown_worker()
     return
   except Full:
    pass
  executor = None

def _queue_management_worker ( executor_reference , processes , pending_work_items , work_ids_queue , call_queue , result_queue ) : executor = None def shutting_down ( ) : return ( _shutdown or ( executor is None ) or executor . _shutdown_thread ) def shutdown_worker ( ) : nb_children_alive = sum ( ( p . is_alive ( ) for p in processes . values ( ) ) ) for i in range ( 0 , nb_children_alive ) : call_queue . put_nowait ( None ) call_queue . close ( ) for p in processes . values ( ) : p . join ( ) reader = result_queue . _reader while True : _add_call_item_to_queue ( pending_work_items , work_ids_queue , call_queue ) sentinels = [ p . sentinel for p in processes . values ( ) ] assert sentinels ready = wait ( ( [ reader ] + sentinels ) ) if ( reader in ready ) : result_item = reader . recv ( ) else : executor = executor_reference ( ) if ( executor is not None ) : executor . _broken = True executor . _shutdown_thread = True executor = None for ( work_id , work_item ) in pending_work_items . items ( ) : work_item . future . set_exception ( BrokenProcessPool ( 'A process in the process pool was terminated abruptly while the future was running or pending.' ) ) del work_item pending_work_items . clear ( ) for p in processes . values ( ) : p . terminate ( ) shutdown_worker ( ) return if isinstance ( result_item , int ) : assert shutting_down ( ) p = processes . pop ( result_item ) p . join ( ) if ( not processes ) : shutdown_worker ( ) return elif ( result_item is not None ) : work_item = pending_work_items . pop ( result_item . work_id , None ) 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 ) del work_item executor = executor_reference ( ) if shutting_down ( ) : try : if ( not pending_work_items ) : shutdown_worker ( ) return except Full : pass executor = None
manages the communication between this process and the worker processes .