( )⚙ D6581 update: fix spurious unclean status bug shown by previous commit

This is an archive of the discontinued Mercurial Phabricator instance.

Differential D6581

update: fix spurious unclean status bug shown by previous commit
ClosedPublic

Authored by valentin.gatienbaron on Jun 27 2019, 7:00 AM.

Details

Reviewers
martinvonz
Group Reviewers
hg-reviewers
Commits
rHGd29db0a0c4eb: update: fix spurious unclean status bug shown by previous commit
Summary

The crux of the problem is:

  • the dirstate is corrupted (the sizes/dates are assigned to the wrong files)
  • because when worker.worker is used with a return value (batchget in merge.py here), the return value when worker.worker effectively parallelizes is permuted
  • this is because worker.worker's partition of input and combination of output values are not inverses of one another: it split [1,2,3,4,5,6] into [[1,3,5],[2,4,6]], but combines that into [1,3,5,2,4,6].

Given that worker.worker doesn't call its function argument on contiguous
chunks on the input arguments, sticking with lists means we'd need to
know the relation between the inputs of worker.worker function argument
(for instance, requiring that every input element is mapped to exactly
one output element). It seems better to instead switch return values to
dicts, which can combined reliably with a straighforward restriction.

Diff Detail

Repository
rHG Mercurial
Lint
Lint Skipped
Unit
Unit Tests Skipped

Event Timeline

martinvonz accepted this revision.Jun 27 2019, 11:05 AM
martinvonz added a subscriber: martinvonz.

Thanks for fixing this quickly! The fix looks very straight-forward too.

This revision is now accepted and ready to land.Jun 27 2019, 11:05 AM

Revision Contents
Changeset List

PathPackages
Mmercurial/merge.py (17 lines)
Mmercurial/worker.py (37 lines)
Mtests/test-simple-update.t (18 lines)

Diff 15669

mercurial/merge.py

def batchget(repo, mctx, wctx, wantfiledata, actions): def batchget(repo, mctx, wctx, wantfiledata, actions):
"""apply gets to the working directory """apply gets to the working directory
mctx is the context to get from mctx is the context to get from
Yields arbitrarily many (False, tuple) for progress updates, followed by Yields arbitrarily many (False, tuple) for progress updates, followed by
exactly one (True, filedata). When wantfiledata is false, filedata is an exactly one (True, filedata). When wantfiledata is false, filedata is an
empty list. When wantfiledata is true, filedata[i] is a triple (mode, size, empty dict. When wantfiledata is true, filedata[f] is a triple (mode, size,
mtime) of the file written for action[i]. mtime) of the file f written for each action.
""" """
filedata = [] filedata = {}
verbose = repo.ui.verbose verbose = repo.ui.verbose
fctx = mctx.filectx fctx = mctx.filectx
ui = repo.ui ui = repo.ui
i = 0 i = 0
with repo.wvfs.backgroundclosing(ui, expectedcount=len(actions)): with repo.wvfs.backgroundclosing(ui, expectedcount=len(actions)):
for f, (flags, backup), msg in actions: for f, (flags, backup), msg in actions:
repo.ui.debug(" %s: %s -> g\n" % (f, msg)) repo.ui.debug(" %s: %s -> g\n" % (f, msg))
if verbose: if verbose:
atomictemp = ui.configbool("experimental", "update.atomic-file") atomictemp = ui.configbool("experimental", "update.atomic-file")
size = wfctx.write(fctx(f).data(), flags, size = wfctx.write(fctx(f).data(), flags,
backgroundclose=True, backgroundclose=True,
atomictemp=atomictemp) atomictemp=atomictemp)
if wantfiledata: if wantfiledata:
s = wfctx.lstat() s = wfctx.lstat()
mode = s.st_mode mode = s.st_mode
mtime = s[stat.ST_MTIME] mtime = s[stat.ST_MTIME]
filedata.append((mode, size, mtime)) # for dirstate.normal filedata[f] = ((mode, size, mtime)) # for dirstate.normal
if i == 100: if i == 100:
yield False, (i, f) yield False, (i, f)
i = 0 i = 0
i += 1 i += 1
if i > 0: if i > 0:
yield False, (i, f) yield False, (i, f)
yield True, filedata yield True, filedata
# get in parallel. # get in parallel.
threadsafe = repo.ui.configbool('experimental', threadsafe = repo.ui.configbool('experimental',
'worker.wdir-get-thread-safe') 'worker.wdir-get-thread-safe')
prog = worker.worker(repo.ui, cost, batchget, prog = worker.worker(repo.ui, cost, batchget,
(repo, mctx, wctx, wantfiledata), (repo, mctx, wctx, wantfiledata),
actions[ACTION_GET], actions[ACTION_GET],
threadsafe=threadsafe, threadsafe=threadsafe,
hasretval=True) hasretval=True)
getfiledata = [] getfiledata = {}
for final, res in prog: for final, res in prog:
if final: if final:
getfiledata = res getfiledata = res
else: else:
i, item = res i, item = res
progress.increment(step=i, item=item) progress.increment(step=i, item=item)
updated = len(actions[ACTION_GET]) updated = len(actions[ACTION_GET])
extraactions = ms.actions() extraactions = ms.actions()
if extraactions: if extraactions:
mfiles = set(a[0] for a in actions[ACTION_MERGE]) mfiles = set(a[0] for a in actions[ACTION_MERGE])
for k, acts in extraactions.iteritems(): for k, acts in extraactions.iteritems():
actions[k].extend(acts) actions[k].extend(acts)
if k == ACTION_GET and wantfiledata: if k == ACTION_GET and wantfiledata:
# no filedata until mergestate is updated to provide it # no filedata until mergestate is updated to provide it
getfiledata.extend([None] * len(acts)) for a in acts:
getfiledata[a[0]] = None
# Remove these files from actions[ACTION_MERGE] as well. This is # Remove these files from actions[ACTION_MERGE] as well. This is
# important because in recordupdates, files in actions[ACTION_MERGE] # important because in recordupdates, files in actions[ACTION_MERGE]
# are processed after files in other actions, and the merge driver # are processed after files in other actions, and the merge driver
# might add files to those actions via extraactions above. This can # might add files to those actions via extraactions above. This can
# lead to a file being recorded twice, with poor results. This is # lead to a file being recorded twice, with poor results. This is
# especially problematic for actions[ACTION_REMOVE] (currently only # especially problematic for actions[ACTION_REMOVE] (currently only
# possible with the merge driver in the initial merge process; # possible with the merge driver in the initial merge process;
# interrupted merges don't go through this flow). # interrupted merges don't go through this flow).
for f, args, msg in actions.get(ACTION_EXEC, []): for f, args, msg in actions.get(ACTION_EXEC, []):
repo.dirstate.normallookup(f) repo.dirstate.normallookup(f)
# keep # keep
for f, args, msg in actions.get(ACTION_KEEP, []): for f, args, msg in actions.get(ACTION_KEEP, []):
pass pass
# get # get
for i, (f, args, msg) in enumerate(actions.get(ACTION_GET, [])): for f, args, msg in actions.get(ACTION_GET, []):
if branchmerge: if branchmerge:
repo.dirstate.otherparent(f) repo.dirstate.otherparent(f)
else: else:
parentfiledata = getfiledata[i] if getfiledata else None parentfiledata = getfiledata[f] if getfiledata else None
repo.dirstate.normal(f, parentfiledata=parentfiledata) repo.dirstate.normal(f, parentfiledata=parentfiledata)
# merge # merge
for f, args, msg in actions.get(ACTION_MERGE, []): for f, args, msg in actions.get(ACTION_MERGE, []):
f1, f2, fa, move, anc = args f1, f2, fa, move, anc = args
if branchmerge: if branchmerge:
# We've done a branch merge, mark this file as merged # We've done a branch merge, mark this file as merged
# so that we properly record the merger later # so that we properly record the merger later

mercurial/worker.py

func - function to run. It is expected to return a progress iterator. func - function to run. It is expected to return a progress iterator.
staticargs - arguments to pass to every invocation of the function staticargs - arguments to pass to every invocation of the function
args - arguments to split into chunks, to pass to individual args - arguments to split into chunks, to pass to individual
workers workers
hasretval - when True, func and the current function return an progress hasretval - when True, func and the current function return an progress
iterator then a list (encoded as an iterator that yield many (False, ..) iterator then a dict (encoded as an iterator that yield many (False, ..)
then a (True, list)). The resulting list is in the natural order. then a (True, dict)). The dicts are joined in some arbitrary order, so
overlapping keys are a bad idea.
threadsafe - whether work items are thread safe and can be executed using 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 a thread-based worker. Should be disabled for CPU heavy tasks that don't
release the GIL. release the GIL.
''' '''
enabled = ui.configbool('worker', 'enabled') enabled = ui.configbool('worker', 'enabled')
if enabled and worthwhile(ui, costperarg, len(args), threadsafe=threadsafe): if enabled and worthwhile(ui, costperarg, len(args), threadsafe=threadsafe):
return _platformworker(ui, func, staticargs, args, hasretval) return _platformworker(ui, func, staticargs, args, hasretval)
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 = [] pipes = []
retvals = [] retval = {}
for i, pargs in enumerate(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 # 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 # implement atomic writes larger than PIPE_BUF. Each forked process has
# its own pipe's descriptors in the local variables, and the parent # 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 # process has the full list of pipe descriptors (and it doesn't really
# care what order they're in). # care what order they're in).
rfd, wfd = os.pipe() rfd, wfd = os.pipe()
pipes.append((rfd, wfd)) pipes.append((rfd, wfd))
retvals.append(None)
# 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]: for r, w in pipes[:-1]:
os.close(r) os.close(r)
os.close(w) 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((i, 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)
signal.signal(signal.SIGCHLD, oldchldhandler) signal.signal(signal.SIGCHLD, oldchldhandler)
selector.close() selector.close()
return problem[0] return problem[0]
try: try:
openpipes = len(pipes) openpipes = len(pipes)
while openpipes > 0: while openpipes > 0:
for key, events in selector.select(): for key, events in selector.select():
try: try:
i, res = util.pickle.load(key.fileobj) res = util.pickle.load(key.fileobj)
if hasretval and res[0]: if hasretval and res[0]:
retvals[i] = res[1] retval.update(res[1])
else: else:
yield res yield res
except EOFError: except EOFError:
selector.unregister(key.fileobj) selector.unregister(key.fileobj)
key.fileobj.close() key.fileobj.close()
openpipes -= 1 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
status = cleanup() status = cleanup()
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)
if hasretval: if hasretval:
yield True, sum(retvals, []) yield True, retval
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
os.spawnv os.spawnv
returns None if the process was stopped instead of exiting''' returns None if the process was stopped instead of exiting'''
if os.WIFEXITED(code): if os.WIFEXITED(code):
return os.WEXITSTATUS(code) return os.WEXITSTATUS(code)
def interrupt(self): def interrupt(self):
self._interrupted = True self._interrupted = True
def run(self): def run(self):
try: try:
while not self._taskqueue.empty(): while not self._taskqueue.empty():
try: try:
i, args = self._taskqueue.get_nowait() args = self._taskqueue.get_nowait()
for res in self._func(*self._staticargs + (args,)): for res in self._func(*self._staticargs + (args,)):
self._resultqueue.put((i, res)) self._resultqueue.put(res)
# threading doesn't provide a native way to # threading doesn't provide a native way to
# interrupt execution. handle it manually at every # interrupt execution. handle it manually at every
# iteration. # iteration.
if self._interrupted: if self._interrupted:
return return
except pycompat.queue.Empty: except pycompat.queue.Empty:
break break
except Exception as e: except Exception as e:
# task and does not get to handle the interruption. # task and does not get to handle the interruption.
ui.warn(_("failed to kill worker threads while " ui.warn(_("failed to kill worker threads while "
"handling an exception\n")) "handling an exception\n"))
return return
workers = _numworkers(ui) workers = _numworkers(ui)
resultqueue = pycompat.queue.Queue() resultqueue = pycompat.queue.Queue()
taskqueue = pycompat.queue.Queue() taskqueue = pycompat.queue.Queue()
retvals = [] retval = {}
# partition work to more pieces than workers to minimize the chance # partition work to more pieces than workers to minimize the chance
# of uneven distribution of large tasks between the workers # of uneven distribution of large tasks between the workers
for pargs in enumerate(partition(args, workers * 20)): for pargs in partition(args, workers * 20):
retvals.append(None)
taskqueue.put(pargs) taskqueue.put(pargs)
for _i in range(workers): for _i in range(workers):
t = Worker(taskqueue, resultqueue, func, staticargs) t = Worker(taskqueue, resultqueue, func, staticargs)
threads.append(t) threads.append(t)
t.start() t.start()
try: try:
while len(threads) > 0: while len(threads) > 0:
while not resultqueue.empty(): while not resultqueue.empty():
(i, res) = resultqueue.get() res = resultqueue.get()
if hasretval and res[0]: if hasretval and res[0]:
retvals[i] = res[1] retval.update(res[1])
else: else:
yield res yield res
threads[0].join(0.05) threads[0].join(0.05)
finishedthreads = [_t for _t in threads if not _t.is_alive()] finishedthreads = [_t for _t in threads if not _t.is_alive()]
for t in finishedthreads: for t in finishedthreads:
if t.exception is not None: if t.exception is not None:
raise t.exception raise t.exception
threads.remove(t) threads.remove(t)
except (Exception, KeyboardInterrupt): # re-raises except (Exception, KeyboardInterrupt): # re-raises
trykillworkers() trykillworkers()
raise raise
while not resultqueue.empty(): while not resultqueue.empty():
(i, res) = resultqueue.get() res = resultqueue.get()
if hasretval and res[0]: if hasretval and res[0]:
retvals[i] = res[1] retval.update(res[1])
else: else:
yield res yield res
if hasretval: if hasretval:
yield True, sum(retvals, []) yield True, retval
if pycompat.iswindows: if pycompat.iswindows:
_platformworker = _windowsworker _platformworker = _windowsworker
else: else:
_platformworker = _posixworker _platformworker = _posixworker
_exitstatus = _posixexitstatus _exitstatus = _posixexitstatus
def partition(lst, nslices): def partition(lst, nslices):

tests/test-simple-update.t

$ hg ci -qAm 'add 100 files' $ hg ci -qAm 'add 100 files'
$ hg update null $ hg update null
0 files updated, 0 files merged, 100 files removed, 0 files unresolved 0 files updated, 0 files merged, 100 files removed, 0 files unresolved
$ hg update -v | grep 100 $ hg update -v | grep 100
getting 100 getting 100
100 files updated, 0 files merged, 0 files removed, 0 files unresolved 100 files updated, 0 files merged, 0 files removed, 0 files unresolved
$ hg status $ hg status
M 100
M 11
M 2
M 21
M 3
M 4
M 41
M 5
M 51
M 54
M 6
M 61
M 7
M 71
M 8
M 81
M 9
M 91
$ cd .. $ cd ..
#endif #endif