This is an archive of the discontinued Mercurial Phabricator instance.

Differential D3960

worker: use one pipe per posix worker and select() in parent process
ClosedPublic

Authored by hooper on Jul 17 2018, 6:44 PM.

Details

Reviewers
None
Group Reviewers
hg-reviewers
Commits
rHG9e6afe7fca31: worker: use one pipe per posix worker and select() in parent process
Summary

This allows us to pass results larger than PIPE_BUF through the pipes without
interleaving them. This is necessary now because "hg fix" sends file contents
as the result from workers.

Diff Detail

Repository
rHG Mercurial
Lint
Automatic diff as part of commit; lint not applicable.
Unit
Automatic diff as part of commit; unit tests not applicable.

Event Timeline

hooper created this revision.Jul 17 2018, 6:44 PM
yuja added a subscriber: yuja.Jul 18 2018, 7:37 AM

@@ -138,7 +138,15 @@

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))

  1. make sure we use os._exit in all worker code paths. otherwise the
  2. worker may do some clean-ups which could cause surprises like
  3. deadlock. see sshpeer.cleanup for example.

@@ -175,8 +183,10 @@

        finally:
            os._exit(ret & 255)
pids.add(pid)
  • os.close(wfd)
  • fp = os.fdopen(rfd, r'rb', 0)

+ fps = []
+ for rfd, wfd in pipes:
+ os.close(wfd)
+ fps.append(os.fdopen(rfd, r'rb', 0))

This isn't enough. For child processes, all pipe fds except for the last
wfd have to be closed at the beginning of workerfunc().

+ rlist, wlist, xlist = select.select(fps, [], fps)

Can you rewrite it to use the selectors module? commandserver.py has an
example.

hooper updated this revision to Diff 9626.Jul 18 2018, 2:45 PM
hooper added a comment.Jul 18 2018, 2:59 PM

Do you want to move the selector import stuff to pycompat?

yuja added a comment.Jul 19 2018, 8:15 AM

Queued, thanks. We'll probably need selector.close() somewhere.

Do you want to move the selector import stuff to pycompat?

Sounds nice. We'll have to pull the latest selectors2 module to get rid of
reference cycle. The issue addressed by a568a46751b6 appears to be fixed
in upstream.

Closed by commit rHG9e6afe7fca31: worker: use one pipe per posix worker and select() in parent process (authored by hooper). · Explain WhyJul 19 2018, 8:18 AM
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

PathPackages
Mmercurial/worker.py (46 lines)

Diff 9628

mercurial/worker.py

# worker.py - master-slave parallelism support # worker.py - master-slave parallelism support
# #
# Copyright 2013 Facebook, Inc. # Copyright 2013 Facebook, Inc.
# #
# This software may be used and distributed according to the terms of the # This software may be used and distributed according to the terms of the
# GNU General Public License version 2 or any later version. # GNU General Public License version 2 or any later version.
from __future__ import absolute_import from __future__ import absolute_import
import errno import errno
import os import os
import signal import signal
import sys import sys
import threading import threading
import time import time
try:
import selectors
selectors.BaseSelector
except ImportError:
from .thirdparty import selectors2 as selectors
from .i18n import _ from .i18n import _
from . import ( from . import (
encoding, encoding,
error, error,
pycompat, pycompat,
scmutil, scmutil,
util, util,
) )
workers workers
''' '''
enabled = ui.configbool('worker', 'enabled') enabled = ui.configbool('worker', 'enabled')
if enabled and worthwhile(ui, costperarg, len(args)): if enabled and worthwhile(ui, costperarg, len(args)):
return _platformworker(ui, func, staticargs, args) return _platformworker(ui, func, staticargs, args)
return func(*staticargs + (args,)) return func(*staticargs + (args,))
def _posixworker(ui, func, staticargs, args): def _posixworker(ui, func, staticargs, args):
rfd, wfd = os.pipe()
workers = _numworkers(ui) workers = _numworkers(ui)
oldhandler = signal.getsignal(signal.SIGINT) oldhandler = signal.getsignal(signal.SIGINT)
signal.signal(signal.SIGINT, signal.SIG_IGN) signal.signal(signal.SIGINT, signal.SIG_IGN)
pids, problem = set(), [0] pids, problem = set(), [0]
def killworkers(): def killworkers():
# unregister SIGCHLD handler as all children will be killed. This # unregister SIGCHLD handler as all children will be killed. This
# function shouldn't be interrupted by another SIGCHLD; otherwise pids # function shouldn't be interrupted by another SIGCHLD; otherwise pids
# could be updated while iterating, which would cause inconsistency. # could be updated while iterating, which would cause inconsistency.
problem[0] = st problem[0] = st
def sigchldhandler(signum, frame): def sigchldhandler(signum, frame):
waitforworkers(blocking=False) waitforworkers(blocking=False)
if problem[0]: if problem[0]:
killworkers() killworkers()
oldchldhandler = signal.signal(signal.SIGCHLD, sigchldhandler) oldchldhandler = signal.signal(signal.SIGCHLD, sigchldhandler)
ui.flush() ui.flush()
parentpid = os.getpid() parentpid = os.getpid()
pipes = []
for pargs in partition(args, workers): 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 # make sure we use os._exit in all worker code paths. otherwise the
# worker may do some clean-ups which could cause surprises like # worker may do some clean-ups which could cause surprises like
# deadlock. see sshpeer.cleanup for example. # deadlock. see sshpeer.cleanup for example.
# override error handling *before* fork. this is necessary because # override error handling *before* fork. this is necessary because
# exception (signal) may arrive after fork, before "pid =" assignment # exception (signal) may arrive after fork, before "pid =" assignment
# completes, and other exception handler (dispatch.py) can lead to # completes, and other exception handler (dispatch.py) can lead to
# unexpected code path without os._exit. # unexpected code path without os._exit.
ret = -1 ret = -1
try: try:
pid = os.fork() pid = os.fork()
if pid == 0: if pid == 0:
signal.signal(signal.SIGINT, oldhandler) signal.signal(signal.SIGINT, oldhandler)
signal.signal(signal.SIGCHLD, oldchldhandler) signal.signal(signal.SIGCHLD, oldchldhandler)
def workerfunc(): def workerfunc():
for r, w in pipes[:-1]:
os.close(r)
os.close(w)
os.close(rfd) os.close(rfd)
for result in func(*(staticargs + (pargs,))): for result in func(*(staticargs + (pargs,))):
os.write(wfd, util.pickle.dumps(result)) os.write(wfd, util.pickle.dumps(result))
return 0 return 0
ret = scmutil.callcatch(ui, workerfunc) ret = scmutil.callcatch(ui, workerfunc)
except: # parent re-raises, child never returns except: # parent re-raises, child never returns
if os.getpid() == parentpid: if os.getpid() == parentpid:
raise raise
exctype = sys.exc_info()[0] exctype = sys.exc_info()[0]
force = not issubclass(exctype, KeyboardInterrupt) force = not issubclass(exctype, KeyboardInterrupt)
ui.traceback(force=force) ui.traceback(force=force)
finally: finally:
if os.getpid() != parentpid: if os.getpid() != parentpid:
try: try:
ui.flush() ui.flush()
except: # never returns, no re-raises except: # never returns, no re-raises
pass pass
finally: finally:
os._exit(ret & 255) os._exit(ret & 255)
pids.add(pid) pids.add(pid)
selector = selectors.DefaultSelector()
for rfd, wfd in pipes:
os.close(wfd) os.close(wfd)
fp = os.fdopen(rfd, r'rb', 0) selector.register(os.fdopen(rfd, r'rb', 0), selectors.EVENT_READ)
def cleanup(): def cleanup():
signal.signal(signal.SIGINT, oldhandler) signal.signal(signal.SIGINT, oldhandler)
waitforworkers() waitforworkers()
signal.signal(signal.SIGCHLD, oldchldhandler) signal.signal(signal.SIGCHLD, oldchldhandler)
status = problem[0] status = problem[0]
if status: if status:
if status < 0: if status < 0:
os.kill(os.getpid(), -status) os.kill(os.getpid(), -status)
sys.exit(status) sys.exit(status)
try: try:
while True: openpipes = len(pipes)
while openpipes > 0:
for key, events in selector.select():
try: try:
yield util.pickle.load(fp) yield util.pickle.load(key.fileobj)
except EOFError: except EOFError:
break selector.unregister(key.fileobj)
key.fileobj.close()
openpipes -= 1
except IOError as e: except IOError as e:
if e.errno == errno.EINTR: if e.errno == errno.EINTR:
continue continue
raise raise
except: # re-raises except: # re-raises
killworkers() killworkers()
cleanup() cleanup()
raise raise
cleanup() cleanup()
def _posixexitstatus(code): def _posixexitstatus(code):
'''convert a posix exit status into the same form returned by '''convert a posix exit status into the same form returned by