worker.py
227 lines
| 7.4 KiB
| text/x-python
|
PythonLexer
/ mercurial / worker.py
|
r18635 | # worker.py - master-slave parallelism support | ||
# | ||||
# Copyright 2013 Facebook, Inc. | ||||
# | ||||
# This software may be used and distributed according to the terms of the | ||||
# GNU General Public License version 2 or any later version. | ||||
|
r25992 | from __future__ import absolute_import | ||
import errno | ||||
import os | ||||
import signal | ||||
import sys | ||||
from .i18n import _ | ||||
|
r30396 | from . import ( | ||
|
r30635 | encoding, | ||
|
r30396 | error, | ||
|
r30639 | pycompat, | ||
|
r30521 | scmutil, | ||
|
r30396 | util, | ||
) | ||||
|
r18635 | |||
def countcpus(): | ||||
'''try to count the number of CPUs on the system''' | ||||
|
r26568 | |||
# posix | ||||
|
r18635 | try: | ||
|
r26568 | n = int(os.sysconf('SC_NPROCESSORS_ONLN')) | ||
if n > 0: | ||||
return n | ||||
except (AttributeError, ValueError): | ||||
pass | ||||
# windows | ||||
try: | ||||
|
r30635 | n = int(encoding.environ['NUMBER_OF_PROCESSORS']) | ||
|
r26568 | if n > 0: | ||
return n | ||||
except (KeyError, ValueError): | ||||
pass | ||||
return 1 | ||||
|
r18636 | |||
def _numworkers(ui): | ||||
s = ui.config('worker', 'numcpus') | ||||
if s: | ||||
try: | ||||
n = int(s) | ||||
if n >= 1: | ||||
return n | ||||
except ValueError: | ||||
|
r26587 | raise error.Abort(_('number of cpus must be an integer')) | ||
|
r18636 | return min(max(countcpus(), 4), 32) | ||
|
r30639 | if pycompat.osname == 'posix': | ||
|
r18636 | _startupcost = 0.01 | ||
else: | ||||
_startupcost = 1e30 | ||||
def worthwhile(ui, costperop, nops): | ||||
'''try to determine whether the benefit of multiple processes can | ||||
outweigh the cost of starting them''' | ||||
linear = costperop * nops | ||||
workers = _numworkers(ui) | ||||
benefit = linear - (_startupcost * workers + linear / workers) | ||||
return benefit >= 0.15 | ||||
|
r18637 | |||
|
r18638 | def worker(ui, costperarg, func, staticargs, args): | ||
'''run a function, possibly in parallel in multiple worker | ||||
processes. | ||||
returns a progress iterator | ||||
costperarg - cost of a single task | ||||
func - function to run | ||||
staticargs - arguments to pass to every invocation of the function | ||||
args - arguments to split into chunks, to pass to individual | ||||
workers | ||||
''' | ||||
if 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) | ||||
|
r18708 | oldhandler = signal.getsignal(signal.SIGINT) | ||
signal.signal(signal.SIGINT, signal.SIG_IGN) | ||||
|
r30413 | pids, problem = set(), [0] | ||
|
r30410 | def killworkers(): | ||
|
r30423 | # unregister SIGCHLD handler as all children will be killed. This | ||
# function shouldn't be interrupted by another SIGCHLD; otherwise pids | ||||
# could be updated while iterating, which would cause inconsistency. | ||||
signal.signal(signal.SIGCHLD, oldchldhandler) | ||||
|
r30410 | # if one worker bails, there's no good reason to wait for the rest | ||
for p in pids: | ||||
try: | ||||
os.kill(p, signal.SIGTERM) | ||||
except OSError as err: | ||||
if err.errno != errno.ESRCH: | ||||
raise | ||||
|
r30412 | def waitforworkers(blocking=True): | ||
|
r30414 | for pid in pids.copy(): | ||
p = st = 0 | ||||
while True: | ||||
try: | ||||
p, st = os.waitpid(pid, (0 if blocking else os.WNOHANG)) | ||||
|
r30422 | break | ||
|
r30414 | except OSError as e: | ||
if e.errno == errno.EINTR: | ||||
continue | ||||
elif e.errno == errno.ECHILD: | ||||
|
r30425 | # child would already be reaped, but pids yet been | ||
# updated (maybe interrupted just after waitpid) | ||||
pids.discard(pid) | ||||
break | ||||
|
r30414 | else: | ||
raise | ||||
|
r31063 | if not p: | ||
# skip subsequent steps, because child process should | ||||
# be still running in this case | ||||
continue | ||||
pids.discard(p) | ||||
st = _exitstatus(st) | ||||
|
r30410 | if st and not problem[0]: | ||
problem[0] = st | ||||
|
r30415 | def sigchldhandler(signum, frame): | ||
waitforworkers(blocking=False) | ||||
|
r30424 | if problem[0]: | ||
killworkers() | ||||
|
r30415 | oldchldhandler = signal.signal(signal.SIGCHLD, sigchldhandler) | ||
|
r18638 | for pargs in partition(args, workers): | ||
pid = os.fork() | ||||
if pid == 0: | ||||
|
r18708 | signal.signal(signal.SIGINT, oldhandler) | ||
|
r30415 | signal.signal(signal.SIGCHLD, oldchldhandler) | ||
|
r30521 | |||
def workerfunc(): | ||||
|
r18638 | os.close(rfd) | ||
for i, item in func(*(staticargs + (pargs,))): | ||||
os.write(wfd, '%d %s\n' % (i, item)) | ||||
|
r30521 | |||
# make sure we use os._exit in all code paths. otherwise the worker | ||||
# may do some clean-ups which could cause surprises like deadlock. | ||||
# see sshpeer.cleanup for example. | ||||
try: | ||||
scmutil.callcatch(ui, workerfunc) | ||||
|
r18638 | except KeyboardInterrupt: | ||
os._exit(255) | ||||
|
r30521 | except: # never return, therefore no re-raises | ||
try: | ||||
ui.traceback() | ||||
finally: | ||||
os._exit(255) | ||||
else: | ||||
os._exit(0) | ||||
|
r30413 | pids.add(pid) | ||
|
r18638 | os.close(wfd) | ||
fp = os.fdopen(rfd, 'rb', 0) | ||||
def cleanup(): | ||||
signal.signal(signal.SIGINT, oldhandler) | ||||
|
r30416 | waitforworkers() | ||
|
r30415 | signal.signal(signal.SIGCHLD, oldchldhandler) | ||
|
r18709 | status = problem[0] | ||
if status: | ||||
if status < 0: | ||||
os.kill(os.getpid(), -status) | ||||
sys.exit(status) | ||||
|
r18638 | try: | ||
|
r30396 | for line in util.iterfile(fp): | ||
|
r18638 | l = line.split(' ', 1) | ||
yield int(l[0]), l[1][:-1] | ||||
except: # re-raises | ||||
|
r18709 | killworkers() | ||
|
r18638 | cleanup() | ||
raise | ||||
cleanup() | ||||
|
r18707 | def _posixexitstatus(code): | ||
'''convert a posix exit status into the same form returned by | ||||
os.spawnv | ||||
returns None if the process was stopped instead of exiting''' | ||||
if os.WIFEXITED(code): | ||||
return os.WEXITSTATUS(code) | ||||
elif os.WIFSIGNALED(code): | ||||
return -os.WTERMSIG(code) | ||||
|
r30639 | if pycompat.osname != 'nt': | ||
|
r18638 | _platformworker = _posixworker | ||
|
r18707 | _exitstatus = _posixexitstatus | ||
|
r18638 | |||
|
r18637 | def partition(lst, nslices): | ||
|
r28181 | '''partition a list into N slices of roughly equal size | ||
The current strategy takes every Nth element from the input. If | ||||
we ever write workers that need to preserve grouping in input | ||||
we should consider allowing callers to specify a partition strategy. | ||||
|
r28292 | |||
mpm is not a fan of this partitioning strategy when files are involved. | ||||
In his words: | ||||
Single-threaded Mercurial makes a point of creating and visiting | ||||
files in a fixed order (alphabetical). When creating files in order, | ||||
a typical filesystem is likely to allocate them on nearby regions on | ||||
disk. Thus, when revisiting in the same order, locality is maximized | ||||
and various forms of OS and disk-level caching and read-ahead get a | ||||
chance to work. | ||||
This effect can be quite significant on spinning disks. I discovered it | ||||
circa Mercurial v0.4 when revlogs were named by hashes of filenames. | ||||
Tarring a repo and copying it to another disk effectively randomized | ||||
the revlog ordering on disk by sorting the revlogs by hash and suddenly | ||||
performance of my kernel checkout benchmark dropped by ~10x because the | ||||
"working set" of sectors visited no longer fit in the drive's cache and | ||||
the workload switched from streaming to random I/O. | ||||
What we should really be doing is have workers read filenames from a | ||||
ordered queue. This preserves locality and also keeps any worker from | ||||
getting more than one file out of balance. | ||||
|
r28181 | ''' | ||
for i in range(nslices): | ||||
yield lst[i::nslices] | ||||