This is an archive of the discontinued Mercurial Phabricator instance.

Differential D4226

repository: establish API for emitting revision deltas
ClosedPublic

Authored by indygreg on Aug 9 2018, 7:12 PM.

Details

Reviewers
None
Group Reviewers
hg-reviewers
Commits
rHGb41d023a412a: repository: establish API for emitting revision deltas
Summary

With our revision delta and revision delta request interfaces
defined, it is now time to define a method on storage interfaces
for using them.

So far, the only storage interface that is well-defined and used
is file storage. So that is the only interface we need to add a
method on.

We define an `emitrevisiondeltas()` method that takes an
iterable of `irevisiondeltarequest`s and turns them into
`irevisiondelta` instances.

changegroup._handlerevisiondeltarequest() and the looping logic
from changegroup.deltagroup() has effectively been moved to
revlog.emitrevisiondeltas().

Our filelog wrapper class proxies its emitrevisiondeltas() to
the internal revlog instance.

The simple store test extension used to verify sanity of storage
abstractions has also implemented emitrevisiondeltas() for
file storage and the test harness when run with this extension doesn't
seem to exhibit any regressions.

Rather than create a shared type to represent revision deltas,
each storage backend has its own type and the class name identifies
where the revision delta was derived from.

Diff Detail

Repository
rHG Mercurial
Lint
Automatic diff as part of commit; lint not applicable.
Unit
Automatic diff as part of commit; unit tests not applicable.

Event Timeline

indygreg created this revision.Aug 9 2018, 7:12 PM
indygreg added a comment.Aug 9 2018, 10:38 PM

There might be room to structure this as a pair of APIs: build delta requests and generate them. But I'm having a difficult time reconciling how ellipsis mode would fit in and how callbacks would work

A benefit of having a pair of functions is that the storage implementation can emit any object it wants to describe the revision deltas that should be generated. Of course, at the point where the storage implementation can figure out the delta request logic, you might as well have a single function in the storage layer that replaces changegroup.deltagroup()!

One of the reasons I would like to have an abstraction for delta requests and a single function to process them is it can lead to easy-to-drop-in caching. For example, it should be possible to write an extension that uses redis or some other in-memory store to service the revisiondeltarequest -> revisiondelta handling.

indygreg updated this revision to Diff 10321.Aug 10 2018, 1:36 PM
durin42 added a subscriber: durin42.Aug 22 2018, 12:26 PM

Nice work, this layering feels saner to me.

Regarding the request/result split for caching, while I agree that caching deltas is something worth pursuing (it was an enormous win on code.google.com!) it's also not something anyone has cared a ton about, and I could also see it living in the store layer directly.

Closed by commit rHGb41d023a412a: repository: establish API for emitting revision deltas (authored by indygreg). · Explain WhyAug 22 2018, 12:37 PM
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

Diff 10511

mercurial/changegroup.py

from . import ( from . import (
dagop, dagop,
error, error,
match as matchmod, match as matchmod,
mdiff, mdiff,
phases, phases,
pycompat, pycompat,
repository, repository,
revlog,
util, util,
) )
from .utils import ( from .utils import (
interfaceutil, interfaceutil,
stringutil, stringutil,
) )
class revisiondeltarequest(object): class revisiondeltarequest(object):
node = attr.ib() node = attr.ib()
linknode = attr.ib() linknode = attr.ib()
p1node = attr.ib() p1node = attr.ib()
p2node = attr.ib() p2node = attr.ib()
basenode = attr.ib() basenode = attr.ib()
ellipsis = attr.ib(default=False) ellipsis = attr.ib(default=False)
@interfaceutil.implementer(repository.irevisiondelta)
@attr.s(slots=True, frozen=True)
class revisiondelta(object):
node = attr.ib()
p1node = attr.ib()
p2node = attr.ib()
basenode = attr.ib()
linknode = attr.ib()
flags = attr.ib()
baserevisionsize = attr.ib()
revision = attr.ib()
delta = attr.ib()
def _revisiondeltatochunks(delta, headerfn): def _revisiondeltatochunks(delta, headerfn):
"""Serialize a revisiondelta to changegroup chunks.""" """Serialize a revisiondelta to changegroup chunks."""
# The captured revision delta may be encoded as a delta against # The captured revision delta may be encoded as a delta against
# a base revision or as a full revision. The changegroup format # a base revision or as a full revision. The changegroup format
# requires that everything on the wire be deltas. So for full # requires that everything on the wire be deltas. So for full
# revisions, we need to invent a header that says to rewrite # revisions, we need to invent a header that says to rewrite
# data. # data.
# The one invariant we *know* holds is that the new (potentially # The one invariant we *know* holds is that the new (potentially
# bogus) DAG shape will be valid if we order the nodes in the # bogus) DAG shape will be valid if we order the nodes in the
# order that they're introduced in dramatis personae by the # order that they're introduced in dramatis personae by the
# changelog, so what we do is we sort the non-changelog histories # changelog, so what we do is we sort the non-changelog histories
# by the order in which they are used by the changelog. # by the order in which they are used by the changelog.
key = lambda n: cl.rev(lookup(n)) key = lambda n: cl.rev(lookup(n))
return [store.rev(n) for n in sorted(nodes, key=key)] return [store.rev(n) for n in sorted(nodes, key=key)]
def _handlerevisiondeltarequest(store, request, prevnode):
"""Obtain a revisiondelta from a revisiondeltarequest"""
node = request.node
rev = store.rev(node)
# Requesting a full revision.
if request.basenode == nullid:
baserev = nullrev
# Requesting an explicit revision.
elif request.basenode is not None:
baserev = store.rev(request.basenode)
# Allowing us to choose.
else:
p1, p2 = store.parentrevs(rev)
dp = store.deltaparent(rev)
if dp == nullrev and store.storedeltachains:
# Avoid sending full revisions when delta parent is null. Pick prev
# in that case. It's tempting to pick p1 in this case, as p1 will
# be smaller in the common case. However, computing a delta against
# p1 may require resolving the raw text of p1, which could be
# expensive. The revlog caches should have prev cached, meaning
# less CPU for changegroup generation. There is likely room to add
# a flag and/or config option to control this behavior.
baserev = store.rev(prevnode)
elif dp == nullrev:
# revlog is configured to use full snapshot for a reason,
# stick to full snapshot.
baserev = nullrev
elif dp not in (p1, p2, store.rev(prevnode)):
# Pick prev when we can't be sure remote has the base revision.
baserev = store.rev(prevnode)
else:
baserev = dp
if baserev != nullrev and not store.candelta(baserev, rev):
baserev = nullrev
revision = None
delta = None
baserevisionsize = None
if store.iscensored(baserev) or store.iscensored(rev):
try:
revision = store.revision(node, raw=True)
except error.CensoredNodeError as e:
revision = e.tombstone
if baserev != nullrev:
baserevisionsize = store.rawsize(baserev)
elif baserev == nullrev:
revision = store.revision(node, raw=True)
else:
delta = store.revdiff(baserev, rev)
extraflags = revlog.REVIDX_ELLIPSIS if request.ellipsis else 0
return revisiondelta(
node=node,
p1node=request.p1node,
p2node=request.p2node,
linknode=request.linknode,
basenode=store.node(baserev),
flags=store.flags(rev) | extraflags,
baserevisionsize=baserevisionsize,
revision=revision,
delta=delta,
)
def _makenarrowdeltarequest(cl, store, ischangelog, rev, node, linkrev, def _makenarrowdeltarequest(cl, store, ischangelog, rev, node, linkrev,
linknode, clrevtolocalrev, fullclnodes, linknode, clrevtolocalrev, fullclnodes,
precomputedellipsis): precomputedellipsis):
linkparents = precomputedellipsis[linkrev] linkparents = precomputedellipsis[linkrev]
def local(clrev): def local(clrev):
"""Turn a changelog revnum into a local revnum. """Turn a changelog revnum into a local revnum.
The ellipsis dag is stored as revnums on the changelog, The ellipsis dag is stored as revnums on the changelog,
# We expect the first pass to be fast, so we only engage the progress # We expect the first pass to be fast, so we only engage the progress
# meter for constructing the revision deltas. # meter for constructing the revision deltas.
progress = None progress = None
if units is not None: if units is not None:
progress = repo.ui.makeprogress(_('bundling'), unit=units, progress = repo.ui.makeprogress(_('bundling'), unit=units,
total=len(requests)) total=len(requests))
prevnode = store.node(revs[0]) for i, delta in enumerate(store.emitrevisiondeltas(requests)):
for i, request in enumerate(requests):
if progress: if progress:
progress.update(i + 1) progress.update(i + 1)
delta = _handlerevisiondeltarequest(store, request, prevnode)
yield delta yield delta
prevnode = request.node
if progress: if progress:
progress.complete() progress.complete()
class cgpacker(object): class cgpacker(object):
def __init__(self, repo, filematcher, version, allowreorder, def __init__(self, repo, filematcher, version, allowreorder,
builddeltaheader, manifestsend, builddeltaheader, manifestsend,
forcedeltaparentprev=False, forcedeltaparentprev=False,
bundlecaps=None, ellipses=False, bundlecaps=None, ellipses=False,

mercurial/filelog.py

return self._revlog.checkhash(text, node, p1=p1, p2=p2, rev=rev) return self._revlog.checkhash(text, node, p1=p1, p2=p2, rev=rev)
def revision(self, node, _df=None, raw=False): def revision(self, node, _df=None, raw=False):
return self._revlog.revision(node, _df=_df, raw=raw) return self._revlog.revision(node, _df=_df, raw=raw)
def revdiff(self, rev1, rev2): def revdiff(self, rev1, rev2):
return self._revlog.revdiff(rev1, rev2) return self._revlog.revdiff(rev1, rev2)
def emitrevisiondeltas(self, requests):
return self._revlog.emitrevisiondeltas(requests)
def addrevision(self, revisiondata, transaction, linkrev, p1, p2, def addrevision(self, revisiondata, transaction, linkrev, p1, p2,
node=None, flags=revlog.REVIDX_DEFAULT_FLAGS, node=None, flags=revlog.REVIDX_DEFAULT_FLAGS,
cachedelta=None): cachedelta=None):
return self._revlog.addrevision(revisiondata, transaction, linkrev, return self._revlog.addrevision(revisiondata, transaction, linkrev,
p1, p2, node=node, flags=flags, p1, p2, node=node, flags=flags,
cachedelta=cachedelta) cachedelta=cachedelta)
def addgroup(self, deltas, linkmapper, transaction, addrevisioncb=None): def addgroup(self, deltas, linkmapper, transaction, addrevisioncb=None):

mercurial/repository.py

"""Obtain a delta between two revision numbers. """Obtain a delta between two revision numbers.
Operates on raw data in the store (``revision(node, raw=True)``). Operates on raw data in the store (``revision(node, raw=True)``).
The returned data is the result of ``bdiff.bdiff`` on the raw The returned data is the result of ``bdiff.bdiff`` on the raw
revision data. revision data.
""" """
def emitrevisiondeltas(requests):
"""Produce ``irevisiondelta`` from ``irevisiondeltarequest``s.
Given an iterable of objects conforming to the ``irevisiondeltarequest``
interface, emits objects conforming to the ``irevisiondelta``
interface.
This method is a generator.
``irevisiondelta`` should be emitted in the same order of
``irevisiondeltarequest`` that was passed in.
The emitted objects MUST conform by the results of
``irevisiondeltarequest``. Namely, they must respect any requests
for building a delta from a specific ``basenode`` if defined.
When sending deltas, implementations must take into account whether
the client has the base delta before encoding a delta against that
revision. A revision encountered previously in ``requests`` is
always a suitable base revision. An example of a bad delta is a delta
against a non-ancestor revision. Another example of a bad delta is a
delta against a censored revision.
"""
class ifilemutation(interfaceutil.Interface): class ifilemutation(interfaceutil.Interface):
"""Storage interface for mutation events of a tracked file.""" """Storage interface for mutation events of a tracked file."""
def add(filedata, meta, transaction, linkrev, p1, p2): def add(filedata, meta, transaction, linkrev, p1, p2):
"""Add a new revision to the store. """Add a new revision to the store.
Takes file data, dictionary of metadata, a transaction, linkrev, Takes file data, dictionary of metadata, a transaction, linkrev,
and parent nodes. and parent nodes.

mercurial/revlog.py

attr, attr,
) )
from . import ( from . import (
ancestor, ancestor,
error, error,
mdiff, mdiff,
policy, policy,
pycompat, pycompat,
repository,
templatefilters, templatefilters,
util, util,
) )
from .utils import ( from .utils import (
interfaceutil,
stringutil, stringutil,
) )
parsers = policy.importmod(r'parsers') parsers = policy.importmod(r'parsers')
# Aliased for performance. # Aliased for performance.
_zlibdecompress = zlib.decompress _zlibdecompress = zlib.decompress
node = attr.ib() node = attr.ib()
p1 = attr.ib() p1 = attr.ib()
p2 = attr.ib() p2 = attr.ib()
btext = attr.ib() btext = attr.ib()
textlen = attr.ib() textlen = attr.ib()
cachedelta = attr.ib() cachedelta = attr.ib()
flags = attr.ib() flags = attr.ib()
@interfaceutil.implementer(repository.irevisiondelta)
@attr.s(slots=True, frozen=True)
class revlogrevisiondelta(object):
node = attr.ib()
p1node = attr.ib()
p2node = attr.ib()
basenode = attr.ib()
linknode = attr.ib()
flags = attr.ib()
baserevisionsize = attr.ib()
revision = attr.ib()
delta = attr.ib()
# index v0: # index v0:
# 4 bytes: offset # 4 bytes: offset
# 4 bytes: compressed length # 4 bytes: compressed length
# 4 bytes: base rev # 4 bytes: base rev
# 4 bytes: link rev # 4 bytes: link rev
# 20 bytes: parent 1 nodeid # 20 bytes: parent 1 nodeid
# 20 bytes: parent 2 nodeid # 20 bytes: parent 2 nodeid
# 20 bytes: nodeid # 20 bytes: nodeid
return (dd, di) return (dd, di)
def files(self): def files(self):
res = [self.indexfile] res = [self.indexfile]
if not self._inline: if not self._inline:
res.append(self.datafile) res.append(self.datafile)
return res return res
def emitrevisiondeltas(self, requests):
frev = self.rev
prevrev = None
for request in requests:
node = request.node
rev = frev(node)
if prevrev is None:
prevrev = self.index[rev][5]
# Requesting a full revision.
if request.basenode == nullid:
baserev = nullrev
# Requesting an explicit revision.
elif request.basenode is not None:
baserev = frev(request.basenode)
# Allowing us to choose.
else:
p1rev, p2rev = self.parentrevs(rev)
deltaparentrev = self.deltaparent(rev)
# Avoid sending full revisions when delta parent is null. Pick
# prev in that case. It's tempting to pick p1 in this case, as
# p1 will be smaller in the common case. However, computing a
# delta against p1 may require resolving the raw text of p1,
# which could be expensive. The revlog caches should have prev
# cached, meaning less CPU for delta generation. There is
# likely room to add a flag and/or config option to control this
# behavior.
if deltaparentrev == nullrev and self.storedeltachains:
baserev = prevrev
# Revlog is configured to use full snapshot for a reason.
# Stick to full snapshot.
elif deltaparentrev == nullrev:
baserev = nullrev
# Pick previous when we can't be sure the base is available
# on consumer.
elif deltaparentrev not in (p1rev, p2rev, prevrev):
baserev = prevrev
else:
baserev = deltaparentrev
if baserev != nullrev and not self.candelta(baserev, rev):
baserev = nullrev
revision = None
delta = None
baserevisionsize = None
if self.iscensored(baserev) or self.iscensored(rev):
try:
revision = self.revision(node, raw=True)
except error.CensoredNodeError as e:
revision = e.tombstone
if baserev != nullrev:
baserevisionsize = self.rawsize(baserev)
elif baserev == nullrev:
revision = self.revision(node, raw=True)
else:
delta = self.revdiff(baserev, rev)
extraflags = REVIDX_ELLIPSIS if request.ellipsis else 0
yield revlogrevisiondelta(
node=node,
p1node=request.p1node,
p2node=request.p2node,
linknode=request.linknode,
basenode=self.node(baserev),
flags=self.flags(rev) | extraflags,
baserevisionsize=baserevisionsize,
revision=revision,
delta=delta)
prevrev = rev
DELTAREUSEALWAYS = 'always' DELTAREUSEALWAYS = 'always'
DELTAREUSESAMEREVS = 'samerevs' DELTAREUSESAMEREVS = 'samerevs'
DELTAREUSENEVER = 'never' DELTAREUSENEVER = 'never'
DELTAREUSEFULLADD = 'fulladd' DELTAREUSEFULLADD = 'fulladd'
DELTAREUSEALL = {'always', 'samerevs', 'never', 'fulladd'} DELTAREUSEALL = {'always', 'samerevs', 'never', 'fulladd'}

tests/simplestorerepo.py

from mercurial.i18n import _ from mercurial.i18n import _
from mercurial.node import ( from mercurial.node import (
bin, bin,
hex, hex,
nullid, nullid,
nullrev, nullrev,
) )
from mercurial.thirdparty import ( from mercurial.thirdparty import (
attr,
cbor, cbor,
) )
from mercurial import ( from mercurial import (
ancestor, ancestor,
bundlerepo, bundlerepo,
error, error,
extensions, extensions,
localrepo, localrepo,
if len(node) != 20: if len(node) != 20:
raise ValueError('expected 20 byte node') raise ValueError('expected 20 byte node')
def validaterev(rev): def validaterev(rev):
if not isinstance(rev, int): if not isinstance(rev, int):
raise ValueError('expected int') raise ValueError('expected int')
@interfaceutil.implementer(repository.irevisiondelta)
@attr.s(slots=True, frozen=True)
class simplestorerevisiondelta(object):
node = attr.ib()
p1node = attr.ib()
p2node = attr.ib()
basenode = attr.ib()
linknode = attr.ib()
flags = attr.ib()
baserevisionsize = attr.ib()
revision = attr.ib()
delta = attr.ib()
@interfaceutil.implementer(repository.ifilestorage) @interfaceutil.implementer(repository.ifilestorage)
class filestorage(object): class filestorage(object):
"""Implements storage for a tracked path. """Implements storage for a tracked path.
Data is stored in the VFS in a directory corresponding to the tracked Data is stored in the VFS in a directory corresponding to the tracked
path. path.
Index data is stored in an ``index`` file using CBOR. Index data is stored in an ``index`` file using CBOR.
validaterev(rev2) validaterev(rev2)
node1 = self.node(rev1) node1 = self.node(rev1)
node2 = self.node(rev2) node2 = self.node(rev2)
return mdiff.textdiff(self.revision(node1, raw=True), return mdiff.textdiff(self.revision(node1, raw=True),
self.revision(node2, raw=True)) self.revision(node2, raw=True))
def emitrevisiondeltas(self, requests):
for request in requests:
node = request.node
rev = self.rev(node)
if request.basenode == nullid:
baserev = nullrev
elif request.basenode is not None:
baserev = self.rev(request.basenode)
else:
# This is a test extension and we can do simple things
# for choosing a delta parent.
baserev = self.deltaparent(rev)
if baserev != nullrev and not self.candelta(baserev, rev):
baserev = nullrev
revision = None
delta = None
baserevisionsize = None
if self.iscensored(baserev) or self.iscensored(rev):
try:
revision = self.revision(node, raw=True)
except error.CensoredNodeError as e:
revision = e.tombstone
if baserev != nullrev:
baserevisionsize = self.rawsize(baserev)
elif baserev == nullrev:
revision = self.revision(node, raw=True)
else:
delta = self.revdiff(baserev, rev)
extraflags = revlog.REVIDX_ELLIPSIS if request.ellipsis else 0
yield simplestorerevisiondelta(
node=node,
p1node=request.p1node,
p2node=request.p2node,
linknode=request.linknode,
basenode=self.node(baserev),
flags=self.flags(rev) | extraflags,
baserevisionsize=baserevisionsize,
revision=revision,
delta=delta)
def headrevs(self): def headrevs(self):
# Assume all revisions are heads by default. # Assume all revisions are heads by default.
revishead = {rev: True for rev in self._indexbyrev} revishead = {rev: True for rev in self._indexbyrev}
for rev, entry in self._indexbyrev.items(): for rev, entry in self._indexbyrev.items():
# Unset head flag for all seen parents. # Unset head flag for all seen parents.
revishead[self.rev(entry[b'p1'])] = False revishead[self.rev(entry[b'p1'])] = False
revishead[self.rev(entry[b'p2'])] = False revishead[self.rev(entry[b'p2'])] = False

tests/test-check-interfaces.py

changegroup, changegroup,
bundlerepo, bundlerepo,
filelog, filelog,
httppeer, httppeer,
localrepo, localrepo,
manifest, manifest,
pycompat, pycompat,
repository, repository,
revlog,
sshpeer, sshpeer,
statichttprepo, statichttprepo,
ui as uimod, ui as uimod,
unionrepo, unionrepo,
vfs as vfsmod, vfs as vfsmod,
wireprotoserver, wireprotoserver,
wireprototypes, wireprototypes,
wireprotov1peer, wireprotov1peer,
# Conforms to imanifestrevisionwritable. # Conforms to imanifestrevisionwritable.
checkzobject(mctx.new()) checkzobject(mctx.new())
checkzobject(mctx.copy()) checkzobject(mctx.copy())
# Conforms to imanifestdict. # Conforms to imanifestdict.
checkzobject(mctx.read()) checkzobject(mctx.read())
ziverify.verifyClass(repository.irevisiondelta, ziverify.verifyClass(repository.irevisiondelta,
changegroup.revisiondelta) revlog.revlogrevisiondelta)
ziverify.verifyClass(repository.irevisiondeltarequest, ziverify.verifyClass(repository.irevisiondeltarequest,
changegroup.revisiondeltarequest) changegroup.revisiondeltarequest)
rd = changegroup.revisiondelta( rd = revlog.revlogrevisiondelta(
node=b'', node=b'',
p1node=b'', p1node=b'',
p2node=b'', p2node=b'',
basenode=b'', basenode=b'',
linknode=b'', linknode=b'',
flags=b'', flags=b'',
baserevisionsize=None, baserevisionsize=None,
revision=b'', revision=b'',