This is an archive of the discontinued Mercurial Phabricator instance.

Differential D730

revlog: add revmap back to revlog.addgroup
ClosedPublic

Authored by durham on Sep 18 2017, 10:08 PM.

Details

Reviewers
indygreg
martinvonz
Group Reviewers
hg-reviewers
Commits
rHG1db9abf407c5: revlog: add revmap back to revlog.addgroup
Summary

The recent c8b6ed51386b patch removed the linkmapper argument from addgroup, as
part of trying to make addgroup more agnostic from the changegroup format. It
turns out that the changegroup can't resolve linkrevs while iterating over the
deltas, because applying the deltas might affect the linkrev resolution. For
example, when applying a series of changelog entries, the linkmapper just
returns len(cl). If we're iterating over the deltas without applying them to the
changelog, this results in incorrect linkrevs. This was caught by the hgsql
extension, which reads the revisions before applying them.

The fix is to return linknodes as part of the delta iterator, and let the
consumer choose what to do.

Diff Detail

Repository
rHG Mercurial
Lint
Lint Skipped
Unit
Unit Tests Skipped

Event Timeline

durham created this revision.Sep 18 2017, 10:08 PM
martinvonz added a subscriber: martinvonz.Sep 19 2017, 1:31 PM

A previous patch removed the revmap argument from addgroup

s/A previous patch/The recent c8b6ed51/ to make it easier for others

s/revmap/linkmapper/

The fix is to return linknodes as part of the delta iterator, and let the
consumer choose what to do.

Great, that seems easier to reason about too.

It will also make it easy to update debugbundle to use deltaiter. Could you do that too?

mercurial/changegroup.py
191–192

I'll repeat my comment from D688: This now seems unnecessary since the first thing you do is to convert it back into a very similar tuple. Do you think the is still useful? Maybe use a named tuple if we want the names? Either way, that's not for this patch, of course.

durham marked an inline comment as done.Sep 20 2017, 12:43 PM

I've updated the message and attached 3 more commits to the stack that update debugbundle, bundlerepo, and the deltachunk fix you suggested.

mercurial/changegroup.py
191–192

I've added a diff to the end of the stack to change it to a tuple. No need for names I think, since the return value is only consumed internal to the class.

martinvonz accepted this revision.Sep 20 2017, 1:19 PM
This revision is now accepted and ready to land.Sep 20 2017, 1:19 PM
durham marked an inline comment as done.Sep 20 2017, 1:56 PM
Closed by commit rHG1db9abf407c5: revlog: add revmap back to revlog.addgroup (authored by durham). · Explain Why
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

PathPackages
Mmercurial/changegroup.py (21 lines)
Mmercurial/revlog.py (5 lines)
Mtests/test-revlog-raw.py (9 lines)

Diff 1888

mercurial/changegroup.py

def deltachunk(self, prevnode): def deltachunk(self, prevnode):
l = self._chunklength() l = self._chunklength()
if not l: if not l:
return {} return {}
headerdata = readexactly(self._stream, self.deltaheadersize) headerdata = readexactly(self._stream, self.deltaheadersize)
header = struct.unpack(self.deltaheader, headerdata) header = struct.unpack(self.deltaheader, headerdata)
delta = readexactly(self._stream, l - self.deltaheadersize) delta = readexactly(self._stream, l - self.deltaheadersize)
node, p1, p2, deltabase, cs, flags = self._deltaheader(header, prevnode) node, p1, p2, deltabase, cs, flags = self._deltaheader(header, prevnode)
return {'node': node, 'p1': p1, 'p2': p2, 'cs': cs, return {'node': node, 'p1': p1, 'p2': p2, 'cs': cs,
'deltabase': deltabase, 'delta': delta, 'flags': flags} 'deltabase': deltabase, 'delta': delta, 'flags': flags}
martinvonzUnsubmitted
Done

I'll repeat my comment from D688: This now seems unnecessary since the first thing you do is to convert it back into a very similar tuple. Do you think the is still useful? Maybe use a named tuple if we want the names? Either way, that's not for this patch, of course.

martinvonz: I'll repeat my comment from D688: This now seems unnecessary since the first thing you do is to…
durhamAuthorUnsubmitted
Not Done

I've added a diff to the end of the stack to change it to a tuple. No need for names I think, since the return value is only consumed internal to the class.

durham: I've added a diff to the end of the stack to change it to a tuple. No need for names I think…
def getchunks(self): def getchunks(self):
"""returns all the chunks contains in the bundle """returns all the chunks contains in the bundle
Used when you need to forward the binary stream to a file or another Used when you need to forward the binary stream to a file or another
network API. To do so, it parse the changegroup data, otherwise it will network API. To do so, it parse the changegroup data, otherwise it will
block in case of sshrepo because it don't know the end of the stream. block in case of sshrepo because it don't know the end of the stream.
""" """
# We know that we'll never have more manifests than we had # We know that we'll never have more manifests than we had
# changesets. # changesets.
self.callback = prog(_('manifests'), numchanges) self.callback = prog(_('manifests'), numchanges)
# no need to check for empty manifest group here: # no need to check for empty manifest group here:
# if the result of the merge of 1 and 2 is the same in 3 and 4, # if the result of the merge of 1 and 2 is the same in 3 and 4,
# no new manifest will be created and the manifest group will # no new manifest will be created and the manifest group will
# be empty during the pull # be empty during the pull
self.manifestheader() self.manifestheader()
deltas = self.deltaiter(revmap) deltas = self.deltaiter()
repo.manifestlog._revlog.addgroup(deltas, trp) repo.manifestlog._revlog.addgroup(deltas, revmap, trp)
repo.ui.progress(_('manifests'), None) repo.ui.progress(_('manifests'), None)
self.callback = None self.callback = None
def apply(self, repo, tr, srctype, url, targetphase=phases.draft, def apply(self, repo, tr, srctype, url, targetphase=phases.draft,
expectedtotal=None): expectedtotal=None):
"""Add the changegroup returned by source.read() to this repo. """Add the changegroup returned by source.read() to this repo.
srctype is a string like 'push', 'pull', or 'unbundle'. url is srctype is a string like 'push', 'pull', or 'unbundle'. url is
the URL of the repo where this changegroup is coming from. the URL of the repo where this changegroup is coming from.
self._count += 1 self._count += 1
self.callback = prog(_('changesets'), expectedtotal) self.callback = prog(_('changesets'), expectedtotal)
efiles = set() efiles = set()
def onchangelog(cl, node): def onchangelog(cl, node):
efiles.update(cl.readfiles(node)) efiles.update(cl.readfiles(node))
self.changelogheader() self.changelogheader()
deltas = self.deltaiter(csmap) deltas = self.deltaiter()
cgnodes = cl.addgroup(deltas, trp, addrevisioncb=onchangelog) cgnodes = cl.addgroup(deltas, csmap, trp, addrevisioncb=onchangelog)
efiles = len(efiles) efiles = len(efiles)
if not cgnodes: if not cgnodes:
repo.ui.develwarn('applied empty changegroup', repo.ui.develwarn('applied empty changegroup',
config='empty-changegroup') config='empty-changegroup')
clend = len(cl) clend = len(cl)
changesets = clend - clstart changesets = clend - clstart
repo.ui.progress(_('changesets'), None) repo.ui.progress(_('changesets'), None)
repo.ui.flush() repo.ui.flush()
# never return 0 here: # never return 0 here:
if deltaheads < 0: if deltaheads < 0:
ret = deltaheads - 1 ret = deltaheads - 1
else: else:
ret = deltaheads + 1 ret = deltaheads + 1
return ret return ret
def deltaiter(self, linkmapper): def deltaiter(self):
""" """
returns an iterator of the deltas in this changegroup returns an iterator of the deltas in this changegroup
Useful for passing to the underlying storage system to be stored. Useful for passing to the underlying storage system to be stored.
""" """
chain = None chain = None
for chunkdata in iter(lambda: self.deltachunk(chain), {}): for chunkdata in iter(lambda: self.deltachunk(chain), {}):
node = chunkdata['node'] node = chunkdata['node']
p1 = chunkdata['p1'] p1 = chunkdata['p1']
p2 = chunkdata['p2'] p2 = chunkdata['p2']
cs = chunkdata['cs'] cs = chunkdata['cs']
deltabase = chunkdata['deltabase'] deltabase = chunkdata['deltabase']
delta = chunkdata['delta'] delta = chunkdata['delta']
flags = chunkdata['flags'] flags = chunkdata['flags']
link = linkmapper(cs)
chain = node chain = node
yield (node, p1, p2, link, deltabase, delta, flags) yield (node, p1, p2, cs, deltabase, delta, flags)
class cg2unpacker(cg1unpacker): class cg2unpacker(cg1unpacker):
"""Unpacker for cg2 streams. """Unpacker for cg2 streams.
cg2 streams add support for generaldelta, so the delta header cg2 streams add support for generaldelta, so the delta header
format is slightly different. All other features about the data format is slightly different. All other features about the data
remain the same. remain the same.
""" """
def _unpackmanifests(self, repo, revmap, trp, prog, numchanges): def _unpackmanifests(self, repo, revmap, trp, prog, numchanges):
super(cg3unpacker, self)._unpackmanifests(repo, revmap, trp, prog, super(cg3unpacker, self)._unpackmanifests(repo, revmap, trp, prog,
numchanges) numchanges)
for chunkdata in iter(self.filelogheader, {}): for chunkdata in iter(self.filelogheader, {}):
# If we get here, there are directory manifests in the changegroup # If we get here, there are directory manifests in the changegroup
d = chunkdata["filename"] d = chunkdata["filename"]
repo.ui.debug("adding %s revisions\n" % d) repo.ui.debug("adding %s revisions\n" % d)
dirlog = repo.manifestlog._revlog.dirlog(d) dirlog = repo.manifestlog._revlog.dirlog(d)
deltas = self.deltaiter(revmap) deltas = self.deltaiter()
if not dirlog.addgroup(deltas, trp): if not dirlog.addgroup(deltas, revmap, trp):
raise error.Abort(_("received dir revlog group is empty")) raise error.Abort(_("received dir revlog group is empty"))
class headerlessfixup(object): class headerlessfixup(object):
def __init__(self, fh, h): def __init__(self, fh, h):
self._h = h self._h = h
self._fh = fh self._fh = fh
def read(self, n): def read(self, n):
if self._h: if self._h:
files += 1 files += 1
f = chunkdata["filename"] f = chunkdata["filename"]
repo.ui.debug("adding %s revisions\n" % f) repo.ui.debug("adding %s revisions\n" % f)
repo.ui.progress(_('files'), files, unit=_('files'), repo.ui.progress(_('files'), files, unit=_('files'),
total=expectedfiles) total=expectedfiles)
fl = repo.file(f) fl = repo.file(f)
o = len(fl) o = len(fl)
try: try:
deltas = source.deltaiter(revmap) deltas = source.deltaiter()
if not fl.addgroup(deltas, trp): if not fl.addgroup(deltas, revmap, trp):
raise error.Abort(_("received file revlog group is empty")) raise error.Abort(_("received file revlog group is empty"))
except error.CensoredBaseError as e: except error.CensoredBaseError as e:
raise error.Abort(_("received delta base is censored: %s") % e) raise error.Abort(_("received delta base is censored: %s") % e)
revisions += len(fl) - o revisions += len(fl) - o
if f in needfiles: if f in needfiles:
needs = needfiles[f] needs = needfiles[f]
for new in xrange(o, len(fl)): for new in xrange(o, len(fl)):
n = fl.node(new) n = fl.node(new)

mercurial/revlog.py

else: else:
offset += curr * self._io.size offset += curr * self._io.size
transaction.add(self.indexfile, offset, curr) transaction.add(self.indexfile, offset, curr)
ifh.write(entry) ifh.write(entry)
ifh.write(data[0]) ifh.write(data[0])
ifh.write(data[1]) ifh.write(data[1])
self.checkinlinesize(transaction, ifh) self.checkinlinesize(transaction, ifh)
def addgroup(self, deltas, transaction, addrevisioncb=None): def addgroup(self, deltas, linkmapper, transaction, addrevisioncb=None):
""" """
add a delta group add a delta group
given a set of deltas, add them to the revision log. the given a set of deltas, add them to the revision log. the
first delta is against its parent, which should be in our first delta is against its parent, which should be in our
log, the rest are against the previous delta. log, the rest are against the previous delta.
If ``addrevisioncb`` is defined, it will be called with arguments of If ``addrevisioncb`` is defined, it will be called with arguments of
dfh = self.opener(self.datafile, "a+") dfh = self.opener(self.datafile, "a+")
def flush(): def flush():
if dfh: if dfh:
dfh.flush() dfh.flush()
ifh.flush() ifh.flush()
try: try:
# loop through our set of deltas # loop through our set of deltas
for data in deltas: for data in deltas:
node, p1, p2, link, deltabase, delta, flags = data node, p1, p2, linknode, deltabase, delta, flags = data
link = linkmapper(linknode)
flags = flags or REVIDX_DEFAULT_FLAGS flags = flags or REVIDX_DEFAULT_FLAGS
nodes.append(node) nodes.append(node)
if node in self.nodemap: if node in self.nodemap:
# this can happen if two branches make the same change # this can happen if two branches make the same change
continue continue

tests/test-revlog-raw.py

else: else:
# suboptimal deltaparent # suboptimal deltaparent
deltaparent = min(0, parentrev) deltaparent = min(0, parentrev)
return {'node': rlog.node(r), 'p1': pnode, 'p2': node.nullid, return {'node': rlog.node(r), 'p1': pnode, 'p2': node.nullid,
'cs': rlog.node(rlog.linkrev(r)), 'flags': rlog.flags(r), 'cs': rlog.node(rlog.linkrev(r)), 'flags': rlog.flags(r),
'deltabase': rlog.node(deltaparent), 'deltabase': rlog.node(deltaparent),
'delta': rlog.revdiff(deltaparent, r)} 'delta': rlog.revdiff(deltaparent, r)}
def deltaiter(self, linkmapper): def deltaiter(self):
chain = None chain = None
for chunkdata in iter(lambda: self.deltachunk(chain), {}): for chunkdata in iter(lambda: self.deltachunk(chain), {}):
node = chunkdata['node'] node = chunkdata['node']
p1 = chunkdata['p1'] p1 = chunkdata['p1']
p2 = chunkdata['p2'] p2 = chunkdata['p2']
cs = chunkdata['cs'] cs = chunkdata['cs']
deltabase = chunkdata['deltabase'] deltabase = chunkdata['deltabase']
delta = chunkdata['delta'] delta = chunkdata['delta']
flags = chunkdata['flags'] flags = chunkdata['flags']
link = linkmapper(cs)
chain = node chain = node
yield (node, p1, p2, link, deltabase, delta, flags) yield (node, p1, p2, cs, deltabase, delta, flags)
def linkmap(lnode): def linkmap(lnode):
return rlog.rev(lnode) return rlog.rev(lnode)
dlog = newrevlog(destname, recreate=True) dlog = newrevlog(destname, recreate=True)
dummydeltas = dummychangegroup().deltaiter(linkmap) dummydeltas = dummychangegroup().deltaiter()
dlog.addgroup(dummydeltas, tr) dlog.addgroup(dummydeltas, linkmap, tr)
return dlog return dlog
def lowlevelcopy(rlog, tr, destname=b'_destrevlog.i'): def lowlevelcopy(rlog, tr, destname=b'_destrevlog.i'):
'''Like addgroupcopy, but use the low level revlog._addrevision directly. '''Like addgroupcopy, but use the low level revlog._addrevision directly.
It exercises some code paths that are hard to reach easily otherwise. It exercises some code paths that are hard to reach easily otherwise.
''' '''
dlog = newrevlog(destname, recreate=True) dlog = newrevlog(destname, recreate=True)