comparison mercurial/worker.py @ 49545:3556f0392808 stable

lfs: avoid closing connections when the worker doesn't fork

Probably not much more than an minor optimization, but could be useful in the
case of `hg verify` where missing blobs are fetched one at a time.

49544:abf471862b8e

    benefit = linear - (_STARTUP_COST * workers + linear / workers)
    return benefit >= 0.15


def worker(
    ui, costperarg, func, staticargs, args, hasretval=False, threadsafe=True
):
    """run a function, possibly in parallel in multiple worker
    processes.

    returns a progress iterator

    overlapping keys are a bad idea.

    threadsafe - whether work items are thread safe and can be executed using
    a thread-based worker. Should be disabled for CPU heavy tasks that don't
    release the GIL.
    """
    enabled = ui.configbool(b'worker', b'enabled')
    if enabled and _platformworker is _posixworker and not ismainthread():
        # The POSIX worker has to install a handler for SIGCHLD.
        # Python up to 3.9 only allows this in the main thread.
        enabled = False

    if enabled and worthwhile(ui, costperarg, len(args), threadsafe=threadsafe):
        return _platformworker(ui, func, staticargs, args, hasretval)
    return func(*staticargs + (args,))


def _posixworker(ui, func, staticargs, args, hasretval):
    workers = _numworkers(ui)
    oldhandler = signal.getsignal(signal.SIGINT)
    signal.signal(signal.SIGINT, signal.SIG_IGN)
    pids, problem = set(), [0]

    oldchldhandler = signal.signal(signal.SIGCHLD, sigchldhandler)
    ui.flush()
    parentpid = os.getpid()
    pipes = []
    retval = {}
    for pargs in partition(args, min(workers, len(args))):
        # Every worker gets its own pipe to send results on, so we don't have to
        # implement atomic writes larger than PIPE_BUF. Each forked process has
        # its own pipe's descriptors in the local variables, and the parent
        # process has the full list of pipe descriptors (and it doesn't really

        return os.WEXITSTATUS(code)
    elif os.WIFSIGNALED(code):
        return -(os.WTERMSIG(code))


def _windowsworker(ui, func, staticargs, args, hasretval):
    class Worker(threading.Thread):
        def __init__(
            self, taskqueue, resultqueue, func, staticargs, *args, **kwargs
        ):
            threading.Thread.__init__(self, *args, **kwargs)

49545:3556f0392808

    benefit = linear - (_STARTUP_COST * workers + linear / workers)
    return benefit >= 0.15


def worker(
    ui,
    costperarg,
    func,
    staticargs,
    args,
    hasretval=False,
    threadsafe=True,
    prefork=None,
):
    """run a function, possibly in parallel in multiple worker
    processes.

    returns a progress iterator

    overlapping keys are a bad idea.

    threadsafe - whether work items are thread safe and can be executed using
    a thread-based worker. Should be disabled for CPU heavy tasks that don't
    release the GIL.

    prefork - a parameterless Callable that is invoked prior to forking the
    process.  fork() is only used on non-Windows platforms, but is also not
    called on POSIX platforms if the work amount doesn't warrant a worker.
    """
    enabled = ui.configbool(b'worker', b'enabled')
    if enabled and _platformworker is _posixworker and not ismainthread():
        # The POSIX worker has to install a handler for SIGCHLD.
        # Python up to 3.9 only allows this in the main thread.
        enabled = False

    if enabled and worthwhile(ui, costperarg, len(args), threadsafe=threadsafe):
        return _platformworker(
            ui, func, staticargs, args, hasretval, prefork=prefork
        )
    return func(*staticargs + (args,))


def _posixworker(ui, func, staticargs, args, hasretval, prefork=None):
    workers = _numworkers(ui)
    oldhandler = signal.getsignal(signal.SIGINT)
    signal.signal(signal.SIGINT, signal.SIG_IGN)
    pids, problem = set(), [0]

    oldchldhandler = signal.signal(signal.SIGCHLD, sigchldhandler)
    ui.flush()
    parentpid = os.getpid()
    pipes = []
    retval = {}

    if prefork:
        prefork()

    for pargs in partition(args, min(workers, len(args))):
        # Every worker gets its own pipe to send results on, so we don't have to
        # implement atomic writes larger than PIPE_BUF. Each forked process has
        # its own pipe's descriptors in the local variables, and the parent
        # process has the full list of pipe descriptors (and it doesn't really

        return os.WEXITSTATUS(code)
    elif os.WIFSIGNALED(code):
        return -(os.WTERMSIG(code))


def _windowsworker(ui, func, staticargs, args, hasretval, prefork=None):
    class Worker(threading.Thread):
        def __init__(
            self, taskqueue, resultqueue, func, staticargs, *args, **kwargs
        ):
            threading.Thread.__init__(self, *args, **kwargs)