loky/process_executor.py
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 as e:
            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)
            sys.exit(1)
        if call_item is None:
            # Notify queue management thread about clean worker shutdown
            result_queue.put(pid)
            with worker_exit_lock:
                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:
                    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()
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 as e:
            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)
            sys.exit(1)
        if call_item is None:
            # Notify queue management thread about clean worker shutdown
            result_queue.put(pid)
            with worker_exit_lock:
                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:
                    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()
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 as e : 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 ) sys . exit ( 1 ) if call_item is None : # Notify queue management thread about clean worker shutdown result_queue . put ( pid ) with worker_exit_lock : 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 : 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 ( )
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.
Evaluates calls from call_queue and places the results in result_queue .