This is an archive of the discontinued Mercurial Phabricator instance.

Differential D6475

merge: fix race that could cause wrong size in dirstate
ClosedPublic

Authored by valentin.gatienbaron on Jun 3 2019, 9:00 AM.

Details

Reviewers
durin42
martinvonz
Group Reviewers
hg-reviewers
Commits
rHG87a34c767384: merge: fix race that could cause wrong size in dirstate
rHG773b02226e93: merge: fix race that could cause wrong size in dirstate
Summary

The problem is that hg merge/update/etc work the following way:

  1. figure out what files to update
  2. apply the update to disk
  3. apply the update to in-memory dirstate
  4. write dirstate

where step3 looks at the filesystem and assumes it sees the result of
step2. If a file is changed between step2 and step3, step3 will record
incorrect information in the dirstate.

I avoid this by passing the size step3 needs directly from step2, for
the common path (not implemented for change/delete conflicts for
instance).

I didn't fix the same race for the exec bit for now, because it's less
likely to be problematic and I had trouble due to the fact that the
dirstate stores the permissions differently from the manifest (st_mode
vs '' 'l' 'x'), in combination with tests that pretend that symlinks
are not supported.

However, I moved the lstat from step3 to step2, which should tighten
the race window markedly, both for the exec bit and for the mtime.

Diff Detail

Repository
rHG Mercurial
Lint
Lint Skipped
Unit
Unit Tests Skipped

Event Timeline

durin42 added a comment.Jun 12 2019, 12:07 PM

I'm broadly happy with this, but had one point where I need clarification and I'd appreciate splitting the worker.py bit into its own change.

mercurial/merge.py
1805

I'm a little confused by this - it looks like we unconditionally set None for some filedatas? why?

mercurial/worker.py
86

Could we make the worker.py change discrete? I think it merits its own commit even if it's in service of this change.

valentin.gatienbaron edited the summary of this revision. (Show Details)Jun 12 2019, 1:25 PM
valentin.gatienbaron updated this revision to Diff 15463.
martinvonz added inline comments.Jun 12 2019, 2:05 PM
mercurial/context.py
1769

Does overlayworkingfilectx also need to implement this function?

1806

Was this necessary for this patch? Why?

martinvonz added a comment.Jun 12 2019, 2:07 PM

Do you need to update the metadata in Phabricator to tell it that D6515 is its new parent?

durin42 added a comment.Jun 12 2019, 2:12 PM
In D6475#94820, @martinvonz wrote:

Do you need to update the metadata in Phabricator to tell it that D6515 is its new parent?

Looks like that is needed. Note that as of today, @ on hg-committed's phabricator extension fixes this up for you when you phabsend ;)

valentin.gatienbaron added inline comments.Jun 12 2019, 3:55 PM
mercurial/context.py
1806

This change is needed to get the size of the written data in merge.batchget.

mercurial/merge.py
1805

It's simpler to have the invariant that every action has a filedata, so that's why I set something. Now the actual filedata information is not available here, because it's not that easy to make it available. These ACTION_GET can come from:

  • delete/change conflicts. In this case, the file contents is computed in filemerge, so one would need to pass around the filedata until the mergestate, or perhaps recompute it in the mergestate class
  • an extension calling the mergestate api to declare a file as clean

Putting None in the list makes the filedata be determined from disk as it always was at the base. So this is simply not fixing the race for these cases.

valentin.gatienbaron added inline comments.Jun 13 2019, 2:38 AM
mercurial/context.py
1769

This is used only by rebase, and given that rebase seems to be very well tested and that the tests pass, empirically, the answer seems to be no. I think what's happening is that this method is called only when isinmemory is false (otherwise step 3 from the description doesn't happen and so we'd be adding calls to lstat where before there were none). The size returned by write is also used only when isinmemory is false.

valentin.gatienbaron added a commit: rHG773b02226e93: merge: fix race that could cause wrong size in dirstate.Jun 13 2019, 10:03 AM
This revision was not accepted when it landed; it landed in state Needs Review.
Closed by commit rHG773b02226e93: merge: fix race that could cause wrong size in dirstate (authored by valentin.gatienbaron). · Explain Why
This revision was automatically updated to reflect the committed changes.
martinvonz added a comment.Jun 26 2019, 5:21 PM

Try this:

hg co -C 4.9; hg debugrebuilddirstate; hg co 5.0; hg st

Before this patch, that produced an empty output as it should. After this patch, it lists hundreds of modified files (536 to be specific). A workaround is to add --config worker.enabled=no. Can you look into it? It's obviously a pretty serious bug that needs to be fixed before we release a new version.

martinvonz added a comment.Jun 26 2019, 5:27 PM
In D6475#95878, @martinvonz wrote:

Try this:
hg co -C 4.9; hg debugrebuilddirstate; hg co 5.0; hg st
Before this patch, that produced an empty output as it should. After this patch, it lists hundreds of modified files (536 to be specific). A workaround is to add --config worker.enabled=no. Can you look into it? It's obviously a pretty serious bug that needs to be fixed before we release a new version.

I forgot to say that what seems wrong in the dirstate is the size (probably other fields too, but size is probably the important one).

valentin.gatienbaron added a comment.Jun 27 2019, 4:08 AM
In D6475#95879, @martinvonz wrote:
In D6475#95878, @martinvonz wrote:

Try this:
hg co -C 4.9; hg debugrebuilddirstate; hg co 5.0; hg st
Before this patch, that produced an empty output as it should. After this patch, it lists hundreds of modified files (536 to be specific). A workaround is to add --config worker.enabled=no. Can you look into it? It's obviously a pretty serious bug that needs to be fixed before we release a new version.

I forgot to say that what seems wrong in the dirstate is the size (probably other fields too, but size is probably the important one).

Oops. I looked and the problem is clearly that the size/lstat for the dirstate are getting swapped between files, when over the parallelism threshold. I thought worker.py would split input arguments [0,1,2,3,4,5] into [[0,1,2], [3,4,5]] but it's actually [[0,2,4], [1,3,5]], so the lists of results are jumbled with parallelization.

Revision Contents
Changeset List

Diff 15327

hgext/largefiles/overrides.py

actions[standin] = ('a', None, 'keep standin') actions[standin] = ('a', None, 'keep standin')
else: # pick remote normal file else: # pick remote normal file
actions[lfile] = ('g', largs, 'replaces standin') actions[lfile] = ('g', largs, 'replaces standin')
actions[standin] = ('r', None, 'replaced by non-standin') actions[standin] = ('r', None, 'replaced by non-standin')
return actions, diverge, renamedelete return actions, diverge, renamedelete
@eh.wrapfunction(merge, 'recordupdates') @eh.wrapfunction(merge, 'recordupdates')
def mergerecordupdates(orig, repo, actions, branchmerge): def mergerecordupdates(orig, repo, actions, branchmerge, getfiledata):
if 'lfmr' in actions: if 'lfmr' in actions:
lfdirstate = lfutil.openlfdirstate(repo.ui, repo) lfdirstate = lfutil.openlfdirstate(repo.ui, repo)
for lfile, args, msg in actions['lfmr']: for lfile, args, msg in actions['lfmr']:
# this should be executed before 'orig', to execute 'remove' # this should be executed before 'orig', to execute 'remove'
# before all other actions # before all other actions
repo.dirstate.remove(lfile) repo.dirstate.remove(lfile)
# make sure lfile doesn't get synclfdirstate'd as normal # make sure lfile doesn't get synclfdirstate'd as normal
lfdirstate.add(lfile) lfdirstate.add(lfile)
lfdirstate.write() lfdirstate.write()
return orig(repo, actions, branchmerge) return orig(repo, actions, branchmerge, getfiledata)
# Override filemerge to prompt the user about how they wish to merge # Override filemerge to prompt the user about how they wish to merge
# largefiles. This will handle identical edits without prompting the user. # largefiles. This will handle identical edits without prompting the user.
@eh.wrapfunction(filemerge, '_filemerge') @eh.wrapfunction(filemerge, '_filemerge')
def overridefilemerge(origfn, premerge, repo, wctx, mynode, orig, fcd, fco, fca, def overridefilemerge(origfn, premerge, repo, wctx, mynode, orig, fcd, fco, fca,
labels=None): labels=None):
if not lfutil.isstandin(orig) or fcd.isabsent() or fco.isabsent(): if not lfutil.isstandin(orig) or fcd.isabsent() or fco.isabsent():
return origfn(premerge, repo, wctx, mynode, orig, fcd, fco, fca, return origfn(premerge, repo, wctx, mynode, orig, fcd, fco, fca,

hgext/narrow/narrowdirstate.py

from mercurial import ( from mercurial import (
error, error,
) )
def wrapdirstate(repo, dirstate): def wrapdirstate(repo, dirstate):
"""Add narrow spec dirstate ignore, block changes outside narrow spec.""" """Add narrow spec dirstate ignore, block changes outside narrow spec."""
def _editfunc(fn): def _editfunc(fn):
def _wrapper(self, *args): def _wrapper(self, *args, **kwargs):
narrowmatch = repo.narrowmatch() narrowmatch = repo.narrowmatch()
for f in args: for f in args:
if f is not None and not narrowmatch(f) and f not in self: if f is not None and not narrowmatch(f) and f not in self:
raise error.Abort(_("cannot track '%s' - it is outside " + raise error.Abort(_("cannot track '%s' - it is outside " +
"the narrow clone") % f) "the narrow clone") % f)
return fn(self, *args) return fn(self, *args, **kwargs)
return _wrapper return _wrapper
class narrowdirstate(dirstate.__class__): class narrowdirstate(dirstate.__class__):
# Prevent adding/editing/copying/deleting files that are outside the # Prevent adding/editing/copying/deleting files that are outside the
# sparse checkout # sparse checkout
@_editfunc @_editfunc
def normal(self, *args): def normal(self, *args, **kwargs):
return super(narrowdirstate, self).normal(*args) return super(narrowdirstate, self).normal(*args, **kwargs)
@_editfunc @_editfunc
def add(self, *args): def add(self, *args):
return super(narrowdirstate, self).add(*args) return super(narrowdirstate, self).add(*args)
@_editfunc @_editfunc
def normallookup(self, *args): def normallookup(self, *args):
return super(narrowdirstate, self).normallookup(*args) return super(narrowdirstate, self).normallookup(*args)

hgext/remotefilelog/__init__.py

def storewrapper(orig, requirements, path, vfstype): def storewrapper(orig, requirements, path, vfstype):
s = orig(requirements, path, vfstype) s = orig(requirements, path, vfstype)
if constants.SHALLOWREPO_REQUIREMENT in requirements: if constants.SHALLOWREPO_REQUIREMENT in requirements:
s = shallowstore.wrapstore(s) s = shallowstore.wrapstore(s)
return s return s
# prefetch files before update # prefetch files before update
def applyupdates(orig, repo, actions, wctx, mctx, overwrite, labels=None): def applyupdates(orig, repo, actions, wctx, mctx, overwrite, wantfiledata,
labels=None):
if isenabled(repo): if isenabled(repo):
manifest = mctx.manifest() manifest = mctx.manifest()
files = [] files = []
for f, args, msg in actions['g']: for f, args, msg in actions['g']:
files.append((f, hex(manifest[f]))) files.append((f, hex(manifest[f])))
# batch fetch the needed files from the server # batch fetch the needed files from the server
repo.fileservice.prefetch(files) repo.fileservice.prefetch(files)
return orig(repo, actions, wctx, mctx, overwrite, labels=labels) return orig(repo, actions, wctx, mctx, overwrite, wantfiledata,
labels=labels)
# Prefetch merge checkunknownfiles # Prefetch merge checkunknownfiles
def checkunknownfiles(orig, repo, wctx, mctx, force, actions, def checkunknownfiles(orig, repo, wctx, mctx, force, actions,
*args, **kwargs): *args, **kwargs):
if isenabled(repo): if isenabled(repo):
files = [] files = []
sparsematch = repo.maybesparsematch(mctx.rev()) sparsematch = repo.maybesparsematch(mctx.rev())
for f, (m, actionargs, msg) in actions.iteritems(): for f, (m, actionargs, msg) in actions.iteritems():

hgext/sparse.py

return orig(self, parent, allfiles, changedfiles) return orig(self, parent, allfiles, changedfiles)
extensions.wrapfunction(dirstate.dirstate, 'rebuild', _rebuild) extensions.wrapfunction(dirstate.dirstate, 'rebuild', _rebuild)
# Prevent adding files that are outside the sparse checkout # Prevent adding files that are outside the sparse checkout
editfuncs = ['normal', 'add', 'normallookup', 'copy', 'remove', 'merge'] editfuncs = ['normal', 'add', 'normallookup', 'copy', 'remove', 'merge']
hint = _('include file with `hg debugsparse --include <pattern>` or use ' + hint = _('include file with `hg debugsparse --include <pattern>` or use ' +
'`hg add -s <file>` to include file directory while adding') '`hg add -s <file>` to include file directory while adding')
for func in editfuncs: for func in editfuncs:
def _wrapper(orig, self, *args): def _wrapper(orig, self, *args, **kwargs):
sparsematch = self._sparsematcher sparsematch = self._sparsematcher
if not sparsematch.always(): if not sparsematch.always():
for f in args: for f in args:
if (f is not None and not sparsematch(f) and if (f is not None and not sparsematch(f) and
f not in self): f not in self):
raise error.Abort(_("cannot add '%s' - it is outside " raise error.Abort(_("cannot add '%s' - it is outside "
"the sparse checkout") % f, "the sparse checkout") % f,
hint=hint) hint=hint)
return orig(self, *args) return orig(self, *args, **kwargs)
extensions.wrapfunction(dirstate.dirstate, func, _wrapper) extensions.wrapfunction(dirstate.dirstate, func, _wrapper)
@command('debugsparse', [ @command('debugsparse', [
('I', 'include', False, _('include files in the sparse checkout')), ('I', 'include', False, _('include files in the sparse checkout')),
('X', 'exclude', False, _('exclude files in the sparse checkout')), ('X', 'exclude', False, _('exclude files in the sparse checkout')),
('d', 'delete', False, _('delete an include/exclude rule')), ('d', 'delete', False, _('delete an include/exclude rule')),
('f', 'force', False, _('allow changing rules even with pending changes')), ('f', 'force', False, _('allow changing rules even with pending changes')),
('', 'enable-profile', False, _('enables the specified profile')), ('', 'enable-profile', False, _('enables the specified profile')),

mercurial/context.py

def data(self): def data(self):
return self._repo.wread(self._path) return self._repo.wread(self._path)
def copysource(self): def copysource(self):
return self._repo.dirstate.copied(self._path) return self._repo.dirstate.copied(self._path)
def size(self): def size(self):
return self._repo.wvfs.lstat(self._path).st_size return self._repo.wvfs.lstat(self._path).st_size
def lstat(self):
martinvonzUnsubmitted
Not Done

Does overlayworkingfilectx also need to implement this function?

martinvonz: Does overlayworkingfilectx also need to implement this function?
valentin.gatienbaronAuthorUnsubmitted
Done

This is used only by rebase, and given that rebase seems to be very well tested and that the tests pass, empirically, the answer seems to be no. I think what's happening is that this method is called only when isinmemory is false (otherwise step 3 from the description doesn't happen and so we'd be adding calls to lstat where before there were none). The size returned by write is also used only when isinmemory is false.

valentin.gatienbaron: This is used only by rebase, and given that rebase seems to be very well tested and that the…
return self._repo.wvfs.lstat(self._path)
def date(self): def date(self):
t, tz = self._changectx.date() t, tz = self._changectx.date()
try: try:
return (self._repo.wvfs.lstat(self._path)[stat.ST_MTIME], tz) return (self._repo.wvfs.lstat(self._path)[stat.ST_MTIME], tz)
except OSError as err: except OSError as err:
if err.errno != errno.ENOENT: if err.errno != errno.ENOENT:
raise raise
return (t, tz) return (t, tz)
def remove(self, ignoremissing=False): def remove(self, ignoremissing=False):
"""wraps unlink for a repo's working directory""" """wraps unlink for a repo's working directory"""
rmdir = self._repo.ui.configbool('experimental', 'removeemptydirs') rmdir = self._repo.ui.configbool('experimental', 'removeemptydirs')
self._repo.wvfs.unlinkpath(self._path, ignoremissing=ignoremissing, self._repo.wvfs.unlinkpath(self._path, ignoremissing=ignoremissing,
rmdir=rmdir) rmdir=rmdir)
def write(self, data, flags, backgroundclose=False, **kwargs): def write(self, data, flags, backgroundclose=False, **kwargs):
"""wraps repo.wwrite""" """wraps repo.wwrite"""
self._repo.wwrite(self._path, data, flags, return self._repo.wwrite(self._path, data, flags,
martinvonzUnsubmitted
Not Done

Was this necessary for this patch? Why?

martinvonz: Was this necessary for this patch? Why?
valentin.gatienbaronAuthorUnsubmitted
Done

This change is needed to get the size of the written data in merge.batchget.

valentin.gatienbaron: This change is needed to get the size of the written data in merge.batchget.
backgroundclose=backgroundclose, backgroundclose=backgroundclose,
**kwargs) **kwargs)
def markcopied(self, src): def markcopied(self, src):
"""marks this file a copy of `src`""" """marks this file a copy of `src`"""
self._repo.dirstate.copy(src, self._path) self._repo.dirstate.copy(src, self._path)
def clearunknown(self): def clearunknown(self):
"""Removes conflicting items in the working directory so that """Removes conflicting items in the working directory so that
``write()`` can be called successfully. ``write()`` can be called successfully.

mercurial/dirstate.py

if entry is not None and entry[0] != 'r': if entry is not None and entry[0] != 'r':
raise error.Abort( raise error.Abort(
_('file %r in dirstate clashes with %r') % _('file %r in dirstate clashes with %r') %
(pycompat.bytestr(d), pycompat.bytestr(f))) (pycompat.bytestr(d), pycompat.bytestr(f)))
self._dirty = True self._dirty = True
self._updatedfiles.add(f) self._updatedfiles.add(f)
self._map.addfile(f, oldstate, state, mode, size, mtime) self._map.addfile(f, oldstate, state, mode, size, mtime)
def normal(self, f): def normal(self, f, parentfiledata=None):
'''Mark a file normal and clean.''' '''Mark a file normal and clean.
parentfiledata: (mode, size, mtime) of the clean file
parentfiledata should be computed from memory (for mode,
size), as or close as possible from the point where we
determined the file was clean, to limit the risk of the
file having been changed by an external process between the
moment where the file was determined to be clean and now.'''
if parentfiledata:
(mode, size, mtime) = parentfiledata
else:
s = os.lstat(self._join(f)) s = os.lstat(self._join(f))
mode = s.st_mode
size = s.st_size
mtime = s[stat.ST_MTIME] mtime = s[stat.ST_MTIME]
self._addpath(f, 'n', s.st_mode, self._addpath(f, 'n', mode, size & _rangemask, mtime & _rangemask)
s.st_size & _rangemask, mtime & _rangemask)
self._map.copymap.pop(f, None) self._map.copymap.pop(f, None)
if f in self._map.nonnormalset: if f in self._map.nonnormalset:
self._map.nonnormalset.remove(f) self._map.nonnormalset.remove(f)
if mtime > self._lastnormaltime: if mtime > self._lastnormaltime:
# Remember the most recent modification timeslot for status(), # Remember the most recent modification timeslot for status(),
# to make sure we won't miss future size-preserving file content # to make sure we won't miss future size-preserving file content
# modifications that happen within the same timeslot. # modifications that happen within the same timeslot.
self._lastnormaltime = mtime self._lastnormaltime = mtime

mercurial/merge.py

# merge.py - directory-level update/merge handling for Mercurial # merge.py - directory-level update/merge handling for Mercurial
# #
# Copyright 2006, 2007 Matt Mackall <mpm@selenic.com> # Copyright 2006, 2007 Matt Mackall <mpm@selenic.com>
# #
# 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 hashlib import hashlib
import shutil import shutil
import stat
import struct import struct
from .i18n import _ from .i18n import _
from .node import ( from .node import (
addednodeid, addednodeid,
bin, bin,
hex, hex,
modifiednodeid, modifiednodeid,
for f, (r, action) in self._results.iteritems(): for f, (r, action) in self._results.iteritems():
if action is not None: if action is not None:
actions[action].append((f, None, "merge result")) actions[action].append((f, None, "merge result"))
return actions return actions
def recordactions(self): def recordactions(self):
"""record remove/add/get actions in the dirstate""" """record remove/add/get actions in the dirstate"""
branchmerge = self._repo.dirstate.p2() != nullid branchmerge = self._repo.dirstate.p2() != nullid
recordupdates(self._repo, self.actions(), branchmerge) recordupdates(self._repo, self.actions(), branchmerge, None)
def queueremove(self, f): def queueremove(self, f):
"""queues a file to be removed from the dirstate """queues a file to be removed from the dirstate
Meant for use by custom merge drivers.""" Meant for use by custom merge drivers."""
self._results[f] = 0, ACTION_REMOVE self._results[f] = 0, ACTION_REMOVE
def queueadd(self, f): def queueadd(self, f):
yield i, f yield i, f
if cwd and not _getcwd(): if cwd and not _getcwd():
# cwd was removed in the course of removing files; print a helpful # cwd was removed in the course of removing files; print a helpful
# warning. # warning.
repo.ui.warn(_("current directory was removed\n" repo.ui.warn(_("current directory was removed\n"
"(consider changing to repo root: %s)\n") % repo.root) "(consider changing to repo root: %s)\n") % repo.root)
def batchget(repo, mctx, wctx, 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 tuples for progress updates Yields arbitrarily many (False, tuple) for progress updates, followed by
exactly one (True, filedata). When wantfiledata is false, filedata is an
empty list. When wantfiledata is true, filedata[i] is a triple (mode, size,
mtime) of the file written for action[i].
""" """
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:
repo.ui.note(_("getting %s\n") % f) repo.ui.note(_("getting %s\n") % f)
if backup: if backup:
# If a file or directory exists with the same name, back that # If a file or directory exists with the same name, back that
# up. Otherwise, look to see if there is a file that conflicts # up. Otherwise, look to see if there is a file that conflicts
# with a directory this file is in, and if so, back that up. # with a directory this file is in, and if so, back that up.
conflicting = f conflicting = f
if not repo.wvfs.lexists(f): if not repo.wvfs.lexists(f):
for p in util.finddirs(f): for p in util.finddirs(f):
if repo.wvfs.isfileorlink(p): if repo.wvfs.isfileorlink(p):
conflicting = p conflicting = p
break break
if repo.wvfs.lexists(conflicting): if repo.wvfs.lexists(conflicting):
orig = scmutil.backuppath(ui, repo, conflicting) orig = scmutil.backuppath(ui, repo, conflicting)
util.rename(repo.wjoin(conflicting), orig) util.rename(repo.wjoin(conflicting), orig)
wctx[f].clearunknown() wfctx = wctx[f]
wfctx.clearunknown()
atomictemp = ui.configbool("experimental", "update.atomic-file") atomictemp = ui.configbool("experimental", "update.atomic-file")
wctx[f].write(fctx(f).data(), flags, backgroundclose=True, size = wfctx.write(fctx(f).data(), flags,
backgroundclose=True,
atomictemp=atomictemp) atomictemp=atomictemp)
if wantfiledata:
s = wfctx.lstat()
mode = s.st_mode
mtime = s[stat.ST_MTIME]
filedata.append((mode, size, mtime)) # for dirstate.normal
if i == 100: if i == 100:
yield i, f yield False, (i, f)
i = 0 i = 0
i += 1 i += 1
if i > 0: if i > 0:
yield i, f yield False, (i, f)
yield True, filedata
def _prefetchfiles(repo, ctx, actions): def _prefetchfiles(repo, ctx, actions):
"""Invoke ``scmutil.prefetchfiles()`` for the files relevant to the dict """Invoke ``scmutil.prefetchfiles()`` for the files relevant to the dict
of merge actions. ``ctx`` is the context being merged in.""" of merge actions. ``ctx`` is the context being merged in."""
# Skipping 'a', 'am', 'f', 'r', 'dm', 'e', 'k', 'p' and 'pr', because they # Skipping 'a', 'am', 'f', 'r', 'dm', 'e', 'k', 'p' and 'pr', because they
# don't touch the context to be merged in. 'cd' is skipped, because # don't touch the context to be merged in. 'cd' is skipped, because
# changed/deleted never resolves to something from the remote side. # changed/deleted never resolves to something from the remote side.
ACTION_DIR_RENAME_MOVE_LOCAL, ACTION_DIR_RENAME_MOVE_LOCAL,
ACTION_LOCAL_DIR_RENAME_GET, ACTION_LOCAL_DIR_RENAME_GET,
ACTION_MERGE, ACTION_MERGE,
ACTION_EXEC, ACTION_EXEC,
ACTION_KEEP, ACTION_KEEP,
ACTION_PATH_CONFLICT, ACTION_PATH_CONFLICT,
ACTION_PATH_CONFLICT_RESOLVE)) ACTION_PATH_CONFLICT_RESOLVE))
def applyupdates(repo, actions, wctx, mctx, overwrite, labels=None): def applyupdates(repo, actions, wctx, mctx, overwrite, wantfiledata,
labels=None):
"""apply the merge action list to the working directory """apply the merge action list to the working directory
wctx is the working copy context wctx is the working copy context
mctx is the context to be merged into the working copy mctx is the context to be merged into the working copy
Return a tuple of counts (updated, merged, removed, unresolved) that Return a tuple of (counts, filedata), where counts is a tuple
describes how many files were affected by the update. (updated, merged, removed, unresolved) that describes how many
files were affected by the update, and filedata is as described in
batchget.
""" """
_prefetchfiles(repo, mctx, actions) _prefetchfiles(repo, mctx, actions)
updated, merged, removed = 0, 0, 0 updated, merged, removed = 0, 0, 0
ms = mergestate.clean(repo, wctx.p1().node(), mctx.node(), labels) ms = mergestate.clean(repo, wctx.p1().node(), mctx.node(), labels)
moves = [] moves = []
for m, l in actions.items(): for m, l in actions.items():
wctx[f].audit() wctx[f].audit()
wctx[f].write(wctx.filectx(f0).data(), wctx.filectx(f0).flags()) wctx[f].write(wctx.filectx(f0).data(), wctx.filectx(f0).flags())
wctx[f0].remove() wctx[f0].remove()
progress.increment(item=f) progress.increment(item=f)
# 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, (repo, mctx, wctx), prog = worker.worker(repo.ui, cost, batchget,
(repo, mctx, wctx, wantfiledata),
actions[ACTION_GET], actions[ACTION_GET],
threadsafe=threadsafe) threadsafe=threadsafe,
for i, item in prog: hasretval=True)
getfiledata = []
for final, res in prog:
if final:
getfiledata = res
else:
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])
if [a for a in actions[ACTION_GET] if a[0] == '.hgsubstate']: if [a for a in actions[ACTION_GET] if a[0] == '.hgsubstate']:
subrepoutil.submerge(repo, wctx, mctx, wctx, overwrite, labels) subrepoutil.submerge(repo, wctx, mctx, wctx, overwrite, labels)
# forget (manifest only, just log it) (must come first) # forget (manifest only, just log it) (must come first)
for f, args, msg in actions[ACTION_FORGET]: for f, args, msg in actions[ACTION_FORGET]:
repo.ui.debug(" %s: %s -> f\n" % (f, msg)) repo.ui.debug(" %s: %s -> f\n" % (f, msg))
merged += msmerged merged += msmerged
removed += msremoved removed += msremoved
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:
getfiledata.extend([None] * len(acts))
durin42Unsubmitted
Not Done

I'm a little confused by this - it looks like we unconditionally set None for some filedatas? why?

durin42: I'm a little confused by this - it looks like we unconditionally set None for some filedatas?
valentin.gatienbaronAuthorUnsubmitted
Done

It's simpler to have the invariant that every action has a filedata, so that's why I set something. Now the actual filedata information is not available here, because it's not that easy to make it available. These ACTION_GET can come from:

  • delete/change conflicts. In this case, the file contents is computed in filemerge, so one would need to pass around the filedata until the mergestate, or perhaps recompute it in the mergestate class
  • an extension calling the mergestate api to declare a file as clean

Putting None in the list makes the filedata be determined from disk as it always was at the base. So this is simply not fixing the race for these cases.

valentin.gatienbaron: It's simpler to have the invariant that every action has a filedata, so that's why I set…
# 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).
# #
# The real fix here is to have indexes by both file and action so # The real fix here is to have indexes by both file and action so
# that when the action for a file is changed it is automatically # that when the action for a file is changed it is automatically
# reflected in the other action lists. But that involves a more # reflected in the other action lists. But that involves a more
# complex data structure, so this will do for now. # complex data structure, so this will do for now.
# #
# We don't need to do the same operation for 'dc' and 'cd' because # We don't need to do the same operation for 'dc' and 'cd' because
# those lists aren't consulted again. # those lists aren't consulted again.
mfiles.difference_update(a[0] for a in acts) mfiles.difference_update(a[0] for a in acts)
actions[ACTION_MERGE] = [a for a in actions[ACTION_MERGE] actions[ACTION_MERGE] = [a for a in actions[ACTION_MERGE]
if a[0] in mfiles] if a[0] in mfiles]
progress.complete() progress.complete()
return updateresult(updated, merged, removed, unresolved) assert len(getfiledata) == (len(actions[ACTION_GET]) if wantfiledata else 0)
return updateresult(updated, merged, removed, unresolved), getfiledata
def recordupdates(repo, actions, branchmerge): def recordupdates(repo, actions, branchmerge, getfiledata):
"record merge actions to the dirstate" "record merge actions to the dirstate"
# remove (must come first) # remove (must come first)
for f, args, msg in actions.get(ACTION_REMOVE, []): for f, args, msg in actions.get(ACTION_REMOVE, []):
if branchmerge: if branchmerge:
repo.dirstate.remove(f) repo.dirstate.remove(f)
else: else:
repo.dirstate.drop(f) repo.dirstate.drop(f)
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 f, args, msg in actions.get(ACTION_GET, []): for i, (f, args, msg) in enumerate(actions.get(ACTION_GET, [])):
if branchmerge: if branchmerge:
repo.dirstate.otherparent(f) repo.dirstate.otherparent(f)
else: else:
repo.dirstate.normal(f) parentfiledata = getfiledata[i] if getfiledata else None
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
repo.dirstate.merge(f) repo.dirstate.merge(f)
and p1.node() == nullid and p1.node() == nullid
and len(actions[ACTION_GET]) >= fsmonitorthreshold and len(actions[ACTION_GET]) >= fsmonitorthreshold
and pycompat.sysplatform.startswith(('linux', 'darwin'))): and pycompat.sysplatform.startswith(('linux', 'darwin'))):
repo.ui.warn( repo.ui.warn(
_('(warning: large working directory being used without ' _('(warning: large working directory being used without '
'fsmonitor enabled; enable fsmonitor to improve performance; ' 'fsmonitor enabled; enable fsmonitor to improve performance; '
'see "hg help -e fsmonitor")\n')) 'see "hg help -e fsmonitor")\n'))
stats = applyupdates(repo, actions, wc, p2, overwrite, labels=labels) updatedirstate = not partial and not wc.isinmemory()
wantfiledata = updatedirstate and not branchmerge
stats, getfiledata = applyupdates(repo, actions, wc, p2, overwrite,
wantfiledata, labels=labels)
if not partial and not wc.isinmemory(): if updatedirstate:
with repo.dirstate.parentchange(): with repo.dirstate.parentchange():
repo.setparents(fp1, fp2) repo.setparents(fp1, fp2)
recordupdates(repo, actions, branchmerge) recordupdates(repo, actions, branchmerge, getfiledata)
# update completed, clear state # update completed, clear state
util.unlink(repo.vfs.join('updatestate')) util.unlink(repo.vfs.join('updatestate'))
if not branchmerge: if not branchmerge:
repo.dirstate.setbranch(p2.branch()) repo.dirstate.setbranch(p2.branch())
# If we're updating to a location, clean up any stale temporary includes # If we're updating to a location, clean up any stale temporary includes
# (ex: this happens during hg rebase --abort). # (ex: this happens during hg rebase --abort).

mercurial/narrowspec.py

def _writeaddedfiles(repo, pctx, files): def _writeaddedfiles(repo, pctx, files):
actions = merge.emptyactions() actions = merge.emptyactions()
addgaction = actions[merge.ACTION_GET].append addgaction = actions[merge.ACTION_GET].append
mf = repo['.'].manifest() mf = repo['.'].manifest()
for f in files: for f in files:
if not repo.wvfs.exists(f): if not repo.wvfs.exists(f):
addgaction((f, (mf.flags(f), False), "narrowspec updated")) addgaction((f, (mf.flags(f), False), "narrowspec updated"))
merge.applyupdates(repo, actions, wctx=repo[None], merge.applyupdates(repo, actions, wctx=repo[None],
mctx=repo['.'], overwrite=False) mctx=repo['.'], overwrite=False, wantfiledata=False)
def checkworkingcopynarrowspec(repo): def checkworkingcopynarrowspec(repo):
storespec = repo.svfs.tryread(FILENAME) storespec = repo.svfs.tryread(FILENAME)
wcspec = repo.vfs.tryread(DIRSTATE_FILENAME) wcspec = repo.vfs.tryread(DIRSTATE_FILENAME)
if wcspec != storespec: if wcspec != storespec:
raise error.Abort(_("working copy's narrowspec is stale"), raise error.Abort(_("working copy's narrowspec is stale"),
hint=_("run 'hg tracked --update-working-copy'")) hint=_("run 'hg tracked --update-working-copy'"))

mercurial/sparse.py

for file in tempincludes: for file in tempincludes:
if file in dirstate and not sparsematch(file): if file in dirstate and not sparsematch(file):
message = _('dropping temporarily included sparse files') message = _('dropping temporarily included sparse files')
actions.append((file, None, message)) actions.append((file, None, message))
dropped.append(file) dropped.append(file)
typeactions = mergemod.emptyactions() typeactions = mergemod.emptyactions()
typeactions['r'] = actions typeactions['r'] = actions
mergemod.applyupdates(repo, typeactions, repo[None], repo['.'], False) mergemod.applyupdates(repo, typeactions, repo[None], repo['.'], False,
wantfiledata=False)
# Fix dirstate # Fix dirstate
for file in dropped: for file in dropped:
dirstate.drop(file) dirstate.drop(file)
repo.vfs.unlink('tempsparse') repo.vfs.unlink('tempsparse')
repo._sparsesignaturecache.clear() repo._sparsesignaturecache.clear()
msg = _('cleaned up %d temporarily added file(s) from the ' msg = _('cleaned up %d temporarily added file(s) from the '
for file in temporaryfiles: for file in temporaryfiles:
if file in wctxmanifest: if file in wctxmanifest:
fctx = repo[None][file] fctx = repo[None][file]
actions.append((file, (fctx.flags(), False), message)) actions.append((file, (fctx.flags(), False), message))
typeactions = mergemod.emptyactions() typeactions = mergemod.emptyactions()
typeactions['g'] = actions typeactions['g'] = actions
mergemod.applyupdates(repo, typeactions, repo[None], repo['.'], mergemod.applyupdates(repo, typeactions, repo[None], repo['.'],
False) False, wantfiledata=False)
dirstate = repo.dirstate dirstate = repo.dirstate
for file, flags, msg in actions: for file, flags, msg in actions:
dirstate.normal(file) dirstate.normal(file)
profiles = activeconfig(repo)[2] profiles = activeconfig(repo)[2]
changedprofiles = profiles & files changedprofiles = profiles & files
# If an active profile changed during the update, refresh the checkout. # If an active profile changed during the update, refresh the checkout.
if old and not new: if old and not new:
dropped.append(file) dropped.append(file)
# Apply changes to disk # Apply changes to disk
typeactions = mergemod.emptyactions() typeactions = mergemod.emptyactions()
for f, (m, args, msg) in actions.iteritems(): for f, (m, args, msg) in actions.iteritems():
typeactions[m].append((f, args, msg)) typeactions[m].append((f, args, msg))
mergemod.applyupdates(repo, typeactions, repo[None], repo['.'], False) mergemod.applyupdates(repo, typeactions, repo[None], repo['.'], False,
wantfiledata=False)
# Fix dirstate # Fix dirstate
for file in added: for file in added:
dirstate.normal(file) dirstate.normal(file)
for file in dropped: for file in dropped:
dirstate.drop(file) dirstate.drop(file)

mercurial/worker.py

if not threadsafe and _DISALLOW_THREAD_UNSAFE: if not threadsafe and _DISALLOW_THREAD_UNSAFE:
return False return False
linear = costperop * nops linear = costperop * nops
workers = _numworkers(ui) workers = _numworkers(ui)
benefit = linear - (_STARTUP_COST * workers + linear / workers) benefit = linear - (_STARTUP_COST * workers + linear / workers)
return benefit >= 0.15 return benefit >= 0.15
def worker(ui, costperarg, func, staticargs, args, threadsafe=True): def worker(ui, costperarg, func, staticargs, args, hasretval=False,
durin42Unsubmitted
Not Done

Could we make the worker.py change discrete? I think it merits its own commit even if it's in service of this change.

durin42: Could we make the worker.py change discrete? I think it merits its own commit even if it's in…
threadsafe=True):
'''run a function, possibly in parallel in multiple worker '''run a function, possibly in parallel in multiple worker
processes. processes.
returns a progress iterator returns a progress iterator
costperarg - cost of a single task costperarg - cost of a single task
func - function to run 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
iterator then a list (encoded as an iterator that yield many (False, ..)
then a (True, list)). The resulting list is in the natural order.
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) return _platformworker(ui, func, staticargs, args, hasretval)
return func(*staticargs + (args,)) return func(*staticargs + (args,))
def _posixworker(ui, func, staticargs, args): def _posixworker(ui, func, staticargs, args, hasretval):
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.
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 = []
for pargs in partition(args, workers): retvals = []
for i, pargs in enumerate(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(result)) os.write(wfd, util.pickle.dumps((i, 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:
yield util.pickle.load(key.fileobj) i, res = util.pickle.load(key.fileobj)
if hasretval and res[0]:
retvals[i] = res[1]
else:
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:
yield True, sum(retvals, [])
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)
elif os.WIFSIGNALED(code): elif os.WIFSIGNALED(code):
return -os.WTERMSIG(code) return -os.WTERMSIG(code)
def _windowsworker(ui, func, staticargs, args): def _windowsworker(ui, func, staticargs, args, hasretval):
class Worker(threading.Thread): class Worker(threading.Thread):
def __init__(self, taskqueue, resultqueue, func, staticargs, *args, def __init__(self, taskqueue, resultqueue, func, staticargs, *args,
**kwargs): **kwargs):
threading.Thread.__init__(self, *args, **kwargs) threading.Thread.__init__(self, *args, **kwargs)
self._taskqueue = taskqueue self._taskqueue = taskqueue
self._resultqueue = resultqueue self._resultqueue = resultqueue
self._func = func self._func = func
self._staticargs = staticargs self._staticargs = staticargs
self._interrupted = False self._interrupted = False
self.daemon = True self.daemon = True
self.exception = None self.exception = None
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:
args = self._taskqueue.get_nowait() i, args = self._taskqueue.get_nowait()
for res in self._func(*self._staticargs + (args,)): for res in self._func(*self._staticargs + (args,)):
self._resultqueue.put(res) self._resultqueue.put((i, 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 = []
# 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 partition(args, workers * 20): for pargs in enumerate(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():
yield resultqueue.get() (i, res) = resultqueue.get()
if hasretval and res[0]:
retvals[i] = res[1]
else:
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():
yield resultqueue.get() (i, res) = resultqueue.get()
if hasretval and res[0]:
retvals[i] = res[1]
else:
yield res
if hasretval:
yield True, sum(retvals, [])
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-dirstate-race2.t

> def wrap(orig, *args, **kwargs): > def wrap(orig, *args, **kwargs):
> res = orig(*args, **kwargs) > res = orig(*args, **kwargs)
> with open("a", "w"): > with open("a", "w"):
> pass # just truncate the file > pass # just truncate the file
> return res > return res
> EOF > EOF
Do an update where file 'a' is changed between hg writing it to disk Do an update where file 'a' is changed between hg writing it to disk
and hg writing the dirstate. It results in a corrupted dirstate, which and hg writing the dirstate. The dirstate is correct nonetheless, and
stores the wrong size, and thus hg status shows spuriously modified so hg status correctly shows a as clean.
files.
$ hg up -r 0 --config extensions.race=$TESTTMP/dirstaterace.py $ hg up -r 0 --config extensions.race=$TESTTMP/dirstaterace.py
1 files updated, 0 files merged, 0 files removed, 0 files unresolved 1 files updated, 0 files merged, 0 files removed, 0 files unresolved
$ hg debugdirstate --no-dates $ hg debugdirstate --no-dates
n 644 0 (set |unset) a (re) n 644 2 (set |unset) a (re)
$ echo a > a; hg status; hg diff $ echo a > a; hg status; hg diff
M a

tests/test-dirstate.t

$ cat > ../dirstateexception.py <<EOF $ cat > ../dirstateexception.py <<EOF
> from __future__ import absolute_import > from __future__ import absolute_import
> from mercurial import ( > from mercurial import (
> error, > error,
> extensions, > extensions,
> merge, > merge,
> ) > )
> >
> def wraprecordupdates(orig, repo, actions, branchmerge): > def wraprecordupdates(*args):
> raise error.Abort("simulated error while recording dirstateupdates") > raise error.Abort("simulated error while recording dirstateupdates")
> >
> def reposetup(ui, repo): > def reposetup(ui, repo):
> extensions.wrapfunction(merge, 'recordupdates', wraprecordupdates) > extensions.wrapfunction(merge, 'recordupdates', wraprecordupdates)
> EOF > EOF
$ hg rm a $ hg rm a
$ hg commit -m 'rm a' $ hg commit -m 'rm a'