Mercurial > hg
view mercurial/worker.py @ 25561:50a6c3c55db1 stable
parsers: do not cache RevlogError type (issue4451)
Index lookups raise RevlogError when the lookup fails. The previous
implementation was caching a reference to the RevlogError type in a
static variable. This assumed that the "mercurial.error" module was
only loaded once and there was only a single copy of it floating
around in memory. Unfortunately, in some situations - including
certain mod_wsgi configurations - this was not the case: the
"mercurial.error" module could be reloaded. It was possible for a
"RevlogError" reference from the first interpreter to be used by
a second interpreter. While the underlying thing was a
"mercurial.error.RevlogError," the object IDs were different, so
the Python code in revlog.py was failing to catch the exception! This
error has existed since the C index lookup code was implemented in
changeset e8d37b78acfb, which was first released in Mercurial 2.2 in
2012.
http://emptysqua.re/blog/python-c-extensions-and-mod-wsgi/#static-variables-are-shared
contains more details.
This patch removes the caching of the RevlogError type from the
function.
Since pretty much the entire function was refactored and the return
value of the function wasn't used, I changed the function signature
to not return anything.
For reasons unknown to me, we were calling PyErr_SetObject()
with the type of RevlogError and an instance of RevlogError. This
was equivalent to the Python code "raise RevlogError(RevlogError)".
This seemed wonky and completely unnecessary. The Python code only
cares about the type of the exception, not its contents. So I got
rid of this complexity.
This is my first Python C extension patch. Please give extra scrutiny
to it during review.
author | Gregory Szorc <gregory.szorc@gmail.com> |
---|---|
date | Fri, 12 Jun 2015 14:43:59 -0700 |
parents | b3e51675f98e |
children | 328739ea70c3 |
line wrap: on
line source
# 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. from i18n import _ import errno, os, signal, sys, threading import util def countcpus(): '''try to count the number of CPUs on the system''' # posix try: n = int(os.sysconf('SC_NPROCESSORS_ONLN')) if n > 0: return n except (AttributeError, ValueError): pass # windows try: n = int(os.environ['NUMBER_OF_PROCESSORS']) if n > 0: return n except (KeyError, ValueError): pass return 1 def _numworkers(ui): s = ui.config('worker', 'numcpus') if s: try: n = int(s) if n >= 1: return n except ValueError: raise util.Abort(_('number of cpus must be an integer')) return min(max(countcpus(), 4), 32) if os.name == 'posix': _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 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) oldhandler = signal.getsignal(signal.SIGINT) signal.signal(signal.SIGINT, signal.SIG_IGN) pids, problem = [], [0] for pargs in partition(args, workers): pid = os.fork() if pid == 0: signal.signal(signal.SIGINT, oldhandler) try: os.close(rfd) for i, item in func(*(staticargs + (pargs,))): os.write(wfd, '%d %s\n' % (i, item)) os._exit(0) except KeyboardInterrupt: os._exit(255) # other exceptions are allowed to propagate, we rely # on lock.py's pid checks to avoid release callbacks pids.append(pid) pids.reverse() os.close(wfd) fp = os.fdopen(rfd, 'rb', 0) def killworkers(): # 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, err: if err.errno != errno.ESRCH: raise def waitforworkers(): for _pid in pids: st = _exitstatus(os.wait()[1]) if st and not problem[0]: problem[0] = st killworkers() t = threading.Thread(target=waitforworkers) t.start() def cleanup(): signal.signal(signal.SIGINT, oldhandler) t.join() status = problem[0] if status: if status < 0: os.kill(os.getpid(), -status) sys.exit(status) try: for line in fp: l = line.split(' ', 1) yield int(l[0]), l[1][:-1] except: # re-raises killworkers() cleanup() raise cleanup() 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) if os.name != 'nt': _platformworker = _posixworker _exitstatus = _posixexitstatus def partition(lst, nslices): '''partition a list into N slices of equal size''' n = len(lst) chunk, slop = n / nslices, n % nslices end = 0 for i in xrange(nslices): start = end end = start + chunk if slop: end += 1 slop -= 1 yield lst[start:end]