( )⚙ D688 changegroup: remove changegroup dependency from revlog.addgroup

This is an archive of the discontinued Mercurial Phabricator instance.

Differential D688

changegroup: remove changegroup dependency from revlog.addgroup
ClosedPublic

Authored by durham on Sep 11 2017, 8:02 PM.

Details

Reviewers
indygreg
Group Reviewers
hg-reviewers
Commits
rHGc8b6ed51386b: changegroup: remove changegroup dependency from revlog.addgroup
Summary

Previously revlog.addgroup would accept a changegroup and a linkmapper and use
it to iterate of the deltas. As part of untangling the revlog-changegroup
interdependency, let's move the changegroup delta iteration logic to it's own
function and pass the simple iterator to the revlog instead.

This will make it easier to introduce non-revlogs stores in the future, without
reinventing any changegroup specific logic.

Diff Detail

Repository
rHG Mercurial
Lint
Lint Skipped
Unit
Unit Tests Skipped

Event Timeline

durham created this revision.Sep 11 2017, 8:02 PM
martinvonz added a subscriber: martinvonz.Sep 12 2017, 6:40 PM
martinvonz added inline comments.
mercurial/revlog.py
1908

nit: If I'm reading this right, you're implicitly saying that "chain" was unnecessary before this patch too. Is that correct? If so, could you extract that to separate patch and explain in the commit message why it's safe?

durham planned changes to this revision.Sep 13 2017, 1:44 PM
durham added inline comments.
mercurial/revlog.py
1908

Not quite. Chain is necessary for the cg.deltachunk call. Since that call has moved, so has the chain declaration.

However, there is another spot below where I say it's unnecessary to store the result of 'self._addrevision(node, ...)', since it will always return node. I could change that to just be chain = none in a separate diff.

durham updated this revision to Diff 1788.Sep 13 2017, 1:49 PM
martinvonz added inline comments.Sep 13 2017, 2:08 PM
mercurial/revlog.py
1908

Oh, that "for chunkdata in iter(lambda: cg.deltachunk(chain), {})" was tricky. Even after seeing that that was the only use, it took me a while to understand that it actually read the updated variable on every iteration. Thanks for extracting that tricky bit to a separate patch.

Closed by commit rHGc8b6ed51386b: changegroup: remove changegroup dependency from revlog.addgroup (authored by durham). · Explain WhySep 13 2017, 2:18 PM
This revision was automatically updated to reflect the committed changes.
martinvonz added inline comments.Sep 13 2017, 2:20 PM
mercurial/changegroup.py
192–193

This is now a little unnecessary, since the first thing you do is to put it back in a tuple (much like the output from _deltaheader). Still worth the complexity and performance cost for compatibility maybe or should we make this method return a tuple?

Revision Contents
Changeset List

PathPackages
Mmercurial/changegroup.py (33 lines)
Mmercurial/revlog.py (26 lines)
Mtests/test-revlog-raw.py (19 lines)

Diff 1745

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
Not Done

This is now a little unnecessary, since the first thing you do is to put it back in a tuple (much like the output from _deltaheader). Still worth the complexity and performance cost for compatibility maybe or should we make this method return a tuple?

martinvonz: This is now a little unnecessary, since the first thing you do is to put it back in a tuple…
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()
repo.manifestlog._revlog.addgroup(self, revmap, trp) deltas = self.deltaiter(revmap)
repo.manifestlog._revlog.addgroup(deltas, 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()
cgnodes = cl.addgroup(self, csmap, trp, addrevisioncb=onchangelog) deltas = self.deltaiter(csmap)
cgnodes = cl.addgroup(deltas, 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):
"""
returns an iterator of the deltas in this changegroup
Useful for passing to the underlying storage system to be stored.
"""
chain = None
for chunkdata in iter(lambda: self.deltachunk(chain), {}):
node = chunkdata['node']
p1 = chunkdata['p1']
p2 = chunkdata['p2']
cs = chunkdata['cs']
deltabase = chunkdata['deltabase']
delta = chunkdata['delta']
flags = chunkdata['flags']
link = linkmapper(cs)
chain = node
yield (node, p1, p2, link, 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.
""" """
deltaheader = _CHANGEGROUPV2_DELTA_HEADER deltaheader = _CHANGEGROUPV2_DELTA_HEADER
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)
if not dirlog.addgroup(self, revmap, trp): deltas = self.deltaiter(revmap)
if not dirlog.addgroup(deltas, 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:
if not fl.addgroup(source, revmap, trp): deltas = source.deltaiter(revmap)
if not fl.addgroup(deltas, 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, cg, linkmapper, transaction, addrevisioncb=None): def addgroup(self, deltas, 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
transaction.add(self.datafile, end) transaction.add(self.datafile, end)
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
chain = None for data in deltas:
martinvonzUnsubmitted
Not Done

nit: If I'm reading this right, you're implicitly saying that "chain" was unnecessary before this patch too. Is that correct? If so, could you extract that to separate patch and explain in the commit message why it's safe?

martinvonz: nit: If I'm reading this right, you're implicitly saying that "chain" was unnecessary before…
durhamAuthorUnsubmitted
Not Done

Not quite. Chain is necessary for the cg.deltachunk call. Since that call has moved, so has the chain declaration.

However, there is another spot below where I say it's unnecessary to store the result of 'self._addrevision(node, ...)', since it will always return node. I could change that to just be chain = none in a separate diff.

durham: Not quite. Chain is necessary for the cg.deltachunk call. Since that call has moved, so has…
martinvonzUnsubmitted
Not Done

Oh, that "for chunkdata in iter(lambda: cg.deltachunk(chain), {})" was tricky. Even after seeing that that was the only use, it took me a while to understand that it actually read the updated variable on every iteration. Thanks for extracting that tricky bit to a separate patch.

martinvonz: Oh, that "for chunkdata in iter(lambda: cg.deltachunk(chain), {})" was tricky. Even after…
for chunkdata in iter(lambda: cg.deltachunk(chain), {}): node, p1, p2, link, deltabase, delta, flags = data
node = chunkdata['node'] flags = flags or REVIDX_DEFAULT_FLAGS
p1 = chunkdata['p1']
p2 = chunkdata['p2']
cs = chunkdata['cs']
deltabase = chunkdata['deltabase']
delta = chunkdata['delta']
flags = chunkdata['flags'] or REVIDX_DEFAULT_FLAGS
nodes.append(node) nodes.append(node)
link = linkmapper(cs)
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
chain = node
continue continue
for p in (p1, p2): for p in (p1, p2):
if p not in self.nodemap: if p not in self.nodemap:
raise LookupError(p, self.indexfile, raise LookupError(p, self.indexfile,
_('unknown parent')) _('unknown parent'))
if deltabase not in self.nodemap: if deltabase not in self.nodemap:
# We assume consumers of addrevisioncb will want to retrieve # We assume consumers of addrevisioncb will want to retrieve
# the added revision, which will require a call to # the added revision, which will require a call to
# revision(). revision() will fast path if there is a cache # revision(). revision() will fast path if there is a cache
# hit. So, we tell _addrevision() to always cache in this case. # hit. So, we tell _addrevision() to always cache in this case.
# We're only using addgroup() in the context of changegroup # We're only using addgroup() in the context of changegroup
# generation so the revision data can always be handled as raw # generation so the revision data can always be handled as raw
# by the flagprocessor. # by the flagprocessor.
chain = self._addrevision(node, None, transaction, link, self._addrevision(node, None, transaction, link,
p1, p2, flags, (baserev, delta), p1, p2, flags, (baserev, delta),
ifh, dfh, ifh, dfh,
alwayscache=bool(addrevisioncb)) alwayscache=bool(addrevisioncb))
if addrevisioncb: if addrevisioncb:
addrevisioncb(self, chain) addrevisioncb(self, node)
if not dfh and not self._inline: if not dfh and not self._inline:
# addrevision switched from inline to conventional # addrevision switched from inline to conventional
# reopen the index # reopen the index
ifh.close() ifh.close()
dfh = self.opener(self.datafile, "a+") dfh = self.opener(self.datafile, "a+")
ifh = self.opener(self.indexfile, "a+", ifh = self.opener(self.indexfile, "a+",
checkambig=self._checkambig) checkambig=self._checkambig)

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):
chain = None
for chunkdata in iter(lambda: self.deltachunk(chain), {}):
node = chunkdata['node']
p1 = chunkdata['p1']
p2 = chunkdata['p2']
cs = chunkdata['cs']
deltabase = chunkdata['deltabase']
delta = chunkdata['delta']
flags = chunkdata['flags']
link = linkmapper(cs)
chain = node
yield (node, p1, p2, link, 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)
dlog.addgroup(dummychangegroup(), linkmap, tr) dummydeltas = dummychangegroup().deltaiter(linkmap)
dlog.addgroup(dummydeltas, 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)