mercurial/worker.py

38728:88be288e8ac1

import os
import signal
import sys
import threading
import time

from .i18n import _
from . import (
    encoding,
    error,

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

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

        if problem[0]:
            killworkers()
    oldchldhandler = signal.signal(signal.SIGCHLD, sigchldhandler)
    ui.flush()
    parentpid = os.getpid()
    for pargs in partition(args, workers):
        # make sure we use os._exit in all worker code paths. otherwise the
        # worker may do some clean-ups which could cause surprises like
        # deadlock. see sshpeer.cleanup for example.
        # override error handling *before* fork. this is necessary because
        # exception (signal) may arrive after fork, before "pid =" assignment

            if pid == 0:
                signal.signal(signal.SIGINT, oldhandler)
                signal.signal(signal.SIGCHLD, oldchldhandler)

                def workerfunc():
                    os.close(rfd)
                    for result in func(*(staticargs + (pargs,))):
                        os.write(wfd, util.pickle.dumps(result))
                    return 0

                except: # never returns, no re-raises
                    pass
                finally:
                    os._exit(ret & 255)
        pids.add(pid)
    os.close(wfd)
    fp = os.fdopen(rfd, r'rb', 0)
    def cleanup():
        signal.signal(signal.SIGINT, oldhandler)
        waitforworkers()
        signal.signal(signal.SIGCHLD, oldchldhandler)
        status = problem[0]
        if status:
            if status < 0:
                os.kill(os.getpid(), -status)
            sys.exit(status)
    try:
        while True:
            try:
                yield util.pickle.load(fp)
            except EOFError:
                break
            except IOError as e:
                if e.errno == errno.EINTR:
                    continue
                raise
    except: # re-raises
        killworkers()
        cleanup()
        raise
    cleanup()

38729:9e6afe7fca31

import os
import signal
import sys
import threading
import time

try:
    import selectors
    selectors.BaseSelector
except ImportError:
    from .thirdparty import selectors2 as selectors

from .i18n import _
from . import (
    encoding,
    error,

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

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

        if problem[0]:
            killworkers()
    oldchldhandler = signal.signal(signal.SIGCHLD, sigchldhandler)
    ui.flush()
    parentpid = os.getpid()
    pipes = []
    for pargs in partition(args, workers):
        # 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
        # care what order they're in).
        rfd, wfd = os.pipe()
        pipes.append((rfd, wfd))
        # make sure we use os._exit in all worker code paths. otherwise the
        # worker may do some clean-ups which could cause surprises like
        # deadlock. see sshpeer.cleanup for example.
        # override error handling *before* fork. this is necessary because
        # exception (signal) may arrive after fork, before "pid =" assignment

            if pid == 0:
                signal.signal(signal.SIGINT, oldhandler)
                signal.signal(signal.SIGCHLD, oldchldhandler)

                def workerfunc():
                    for r, w in pipes[:-1]:
                        os.close(r)
                        os.close(w)
                    os.close(rfd)
                    for result in func(*(staticargs + (pargs,))):
                        os.write(wfd, util.pickle.dumps(result))
                    return 0

                except: # never returns, no re-raises
                    pass
                finally:
                    os._exit(ret & 255)
        pids.add(pid)
    selector = selectors.DefaultSelector()
    for rfd, wfd in pipes:
        os.close(wfd)
        selector.register(os.fdopen(rfd, r'rb', 0), selectors.EVENT_READ)
    def cleanup():
        signal.signal(signal.SIGINT, oldhandler)
        waitforworkers()
        signal.signal(signal.SIGCHLD, oldchldhandler)
        status = problem[0]
        if status:
            if status < 0:
                os.kill(os.getpid(), -status)
            sys.exit(status)
    try:
        openpipes = len(pipes)
        while openpipes > 0:
            for key, events in selector.select():
                try:
                    yield util.pickle.load(key.fileobj)
                except EOFError:
                    selector.unregister(key.fileobj)
                    key.fileobj.close()
                    openpipes -= 1
                except IOError as e:
                    if e.errno == errno.EINTR:
                        continue
                    raise
    except: # re-raises
        killworkers()
        cleanup()
        raise
    cleanup()