parameters: (call_queue, result_queue, initializer, initargs,
                      processes_management_lock, timeout, worker_exit_lock,
                      current_depth)
docstring: Evaluates calls from call_queue and places the results in result_queue.
  
      This worker is run in a separate process.
  
      Args:
          call_queue: A ctx.Queue of _CallItems that will be read and
              evaluated by the worker.
          result_queue: A ctx.Queue of _ResultItems that will written
              to by the worker.
          initializer: A callable initializer, or None
          initargs: A tuple of args for the initializer
          process_management_lock: A ctx.Lock avoiding worker timeout while some
              workers are being spawned.
          timeout: maximum time to wait for a new item in the call_queue. If that
              time is expired, the worker will shutdown.
          worker_exit_lock: Lock to avoid flagging the executor as broken on
              workers timeout.
          current_depth: Nested parallelism level, to avoid infinite spawning.
docstring_summary: Evaluates calls from call_queue and places the results in result_queue.
docstring_tokens: Evaluates calls from call_queue and places the results in result_queue .
function: def _process_worker(call_queue, result_queue, initializer, initargs,
                      processes_management_lock, timeout, worker_exit_lock,
                      current_depth):
      """Evaluates calls from call_queue and places the results in result_queue.
  
      This worker is run in a separate process.
  
      Args:
          call_queue: A ctx.Queue of _CallItems that will be read and
              evaluated by the worker.
          result_queue: A ctx.Queue of _ResultItems that will written
              to by the worker.
          initializer: A callable initializer, or None
          initargs: A tuple of args for the initializer
          process_management_lock: A ctx.Lock avoiding worker timeout while some
              workers are being spawned.
          timeout: maximum time to wait for a new item in the call_queue. If that
              time is expired, the worker will shutdown.
          worker_exit_lock: Lock to avoid flagging the executor as broken on
              workers timeout.
          current_depth: Nested parallelism level, to avoid infinite spawning.
      """
      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
  
      # set the global _CURRENT_DEPTH mechanism to limit recursive call
      global _CURRENT_DEPTH
      _CURRENT_DEPTH = current_depth
      _process_reference_size = None
      _last_memory_leak_check = None
      pid = os.getpid()
  
      mp.util.debug('Worker started with timeout=%s' % timeout)
      while True:
          try:
              call_item = call_queue.get(block=True, timeout=timeout)
              if call_item is None:
                  mp.util.info("Shutting down worker on sentinel")
          except queue.Empty:
              mp.util.info("Shutting down worker after timeout %0.3fs"
                           % timeout)
              if processes_management_lock.acquire(block=False):
                  processes_management_lock.release()
                  call_item = None
              else:
                  mp.util.info("Could not acquire processes_management_lock")
                  continue
          except BaseException:
              previous_tb = traceback.format_exc()
              try:
                  result_queue.put(_RemoteTraceback(previous_tb))
              except BaseException:
                  # If we cannot format correctly the exception, at least print
                  # the traceback.
                  print(previous_tb)
              mp.util.debug('Exiting with code 1')
              sys.exit(1)
          if call_item is None:
              # Notify queue management thread about clean worker shutdown
              result_queue.put(pid)
              with worker_exit_lock:
                  mp.util.debug('Exited cleanly')
                  return
          try:
              r = call_item()
          except BaseException as e:
              exc = _ExceptionWithTraceback(e)
              result_queue.put(_ResultItem(call_item.work_id, exception=exc))
          else:
              _sendback_result(result_queue, call_item.work_id, result=r)
              del r
  
          # Free the resource as soon as possible, to avoid holding onto
          # open files or shared memory that is not needed anymore
          del call_item
  
          if _USE_PSUTIL:
              if _process_reference_size is None:
                  # Make reference measurement after the first call
                  _process_reference_size = _get_memory_usage(pid, force_gc=True)
                  _last_memory_leak_check = time()
                  continue
              if time() - _last_memory_leak_check > _MEMORY_LEAK_CHECK_DELAY:
                  mem_usage = _get_memory_usage(pid)
                  _last_memory_leak_check = time()
                  if mem_usage - _process_reference_size < _MAX_MEMORY_LEAK_SIZE:
                      # Memory usage stays within bounds: everything is fine.
                      continue
  
                  # Check again memory usage; this time take the measurement
                  # after a forced garbage collection to break any reference
                  # cycles.
                  mem_usage = _get_memory_usage(pid, force_gc=True)
                  _last_memory_leak_check = time()
                  if mem_usage - _process_reference_size < _MAX_MEMORY_LEAK_SIZE:
                      # The GC managed to free the memory: everything is fine.
                      continue
  
                  # The process is leaking memory: let the master process
                  # know that we need to start a new worker.
                  mp.util.info("Memory leak detected: shutting down worker")
                  result_queue.put(pid)
                  with worker_exit_lock:
                      mp.util.debug('Exit due to memory leak')
                      return
          else:
              # if psutil is not installed, trigger gc.collect events
              # regularly to limit potential memory leaks due to reference cycles
              if ((_last_memory_leak_check is None) or
                      (time() - _last_memory_leak_check >
                       _MEMORY_LEAK_CHECK_DELAY)):
                  gc.collect()
                  _last_memory_leak_check = time()
function_tokens: def _process_worker ( call_queue , result_queue , initializer , initargs , processes_management_lock , timeout , worker_exit_lock , current_depth ) : 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 # set the global _CURRENT_DEPTH mechanism to limit recursive call global _CURRENT_DEPTH _CURRENT_DEPTH = current_depth _process_reference_size = None _last_memory_leak_check = None pid = os . getpid ( ) mp . util . debug ( 'Worker started with timeout=%s' % timeout ) while True : try : call_item = call_queue . get ( block = True , timeout = timeout ) if call_item is None : mp . util . info ( "Shutting down worker on sentinel" ) except queue . Empty : mp . util . info ( "Shutting down worker after timeout %0.3fs" % timeout ) if processes_management_lock . acquire ( block = False ) : processes_management_lock . release ( ) call_item = None else : mp . util . info ( "Could not acquire processes_management_lock" ) continue except BaseException : previous_tb = traceback . format_exc ( ) try : result_queue . put ( _RemoteTraceback ( previous_tb ) ) except BaseException : # If we cannot format correctly the exception, at least print # the traceback. print ( previous_tb ) mp . util . debug ( 'Exiting with code 1' ) sys . exit ( 1 ) if call_item is None : # Notify queue management thread about clean worker shutdown result_queue . put ( pid ) with worker_exit_lock : mp . util . debug ( 'Exited cleanly' ) return try : r = call_item ( ) except BaseException as e : exc = _ExceptionWithTraceback ( e ) result_queue . put ( _ResultItem ( call_item . work_id , exception = exc ) ) else : _sendback_result ( result_queue , call_item . work_id , result = r ) del r # Free the resource as soon as possible, to avoid holding onto # open files or shared memory that is not needed anymore del call_item if _USE_PSUTIL : if _process_reference_size is None : # Make reference measurement after the first call _process_reference_size = _get_memory_usage ( pid , force_gc = True ) _last_memory_leak_check = time ( ) continue if time ( ) - _last_memory_leak_check > _MEMORY_LEAK_CHECK_DELAY : mem_usage = _get_memory_usage ( pid ) _last_memory_leak_check = time ( ) if mem_usage - _process_reference_size < _MAX_MEMORY_LEAK_SIZE : # Memory usage stays within bounds: everything is fine. continue # Check again memory usage; this time take the measurement # after a forced garbage collection to break any reference # cycles. mem_usage = _get_memory_usage ( pid , force_gc = True ) _last_memory_leak_check = time ( ) if mem_usage - _process_reference_size < _MAX_MEMORY_LEAK_SIZE : # The GC managed to free the memory: everything is fine. continue # The process is leaking memory: let the master process # know that we need to start a new worker. mp . util . info ( "Memory leak detected: shutting down worker" ) result_queue . put ( pid ) with worker_exit_lock : mp . util . debug ( 'Exit due to memory leak' ) return else : # if psutil is not installed, trigger gc.collect events # regularly to limit potential memory leaks due to reference cycles if ( ( _last_memory_leak_check is None ) or ( time ( ) - _last_memory_leak_check > _MEMORY_LEAK_CHECK_DELAY ) ) : gc . collect ( ) _last_memory_leak_check = time ( )