( )⚙ D10028 sidedata-exchange: add `wanted_sidedata` and `sidedata_computers` to repos

This is an archive of the discontinued Mercurial Phabricator instance.

Differential D10028

sidedata-exchange: add `wanted_sidedata` and `sidedata_computers` to repos
ClosedPublic

Authored by Alphare on Feb 19 2021, 6:16 AM.

Details

Reviewers
marmoute
Group Reviewers
hg-reviewers
Commits
rHGbc2519513ae0: sidedata-exchange: add `wanted_sidedata` and `sidedata_computers` to repos
Summary

Each repo will advertise the sidedata categories it requires (categories being
unique and canonical), and have a set of "computers", functions to generate
sidedata from (repo, revlog, rev, previous_sidedata), for a given category.
The set of computers can be a superset of the set of the wanted categories, but
not smaller: repos are expected to be coherent in their handling of sidedata.

Diff Detail

Repository
rHG Mercurial
Branch
default
Lint
No Linters Available
Unit
No Unit Test Coverage

Event Timeline

Alphare created this revision.Feb 19 2021, 6:16 AM
baymax updated this revision to Diff 25995.Mar 1 2021, 11:52 AM

✅ refresh by Heptapod after a successful CI run (🐙 💚)

marmoute requested changes to this revision.Mar 2 2021, 2:00 PM
marmoute added a subscriber: marmoute.
marmoute added inline comments.
mercurial/metadata.py
819–825

We should use a registrar here (or at least a small helper function), that way, changing the implementation/processing will be simple in the futur.

What do you think ?

This revision now requires changes to proceed.Mar 2 2021, 2:00 PM
Alphare marked an inline comment as done.Mar 4 2021, 10:04 AM
Alphare updated this revision to Diff 26091.Mar 4 2021, 10:18 AM
marmoute requested changes to this revision.Mar 9 2021, 11:36 AM
marmoute added inline comments.
mercurial/bundle2.py
1815

Can you add a small docstring to clarify the semantic of the function ?

mercurial/changegroup.py
948

If you want to test for None, explicitly test for None (is None). Otherwise you introduce subtle bugs for our future self.

mercurial/exchange.py
423–435

Why do we need multiples layer of loop here . I would expect an API to return a the appropriate object for a categories and that's it.

So, what is going on here ?

1628

Same question here.

This revision now requires changes to proceed.Mar 9 2021, 11:36 AM
Alphare marked 2 inline comments as done.Mar 10 2021, 1:42 PM
Alphare added inline comments.
mercurial/exchange.py
423–435

Categories are advertised as a flat list, but can affect any or all of the revlog kinds. This checks that a sidedata computer is registered for at least one revlog kind.

Alphare updated this revision to Diff 26222.Mar 10 2021, 2:12 PM
marmoute accepted this revision.Mar 11 2021, 1:13 PM
marmoute added inline comments.
mercurial/exchange.py
423–435

Ha, yes…

marmoute requested changes to this revision.Mar 11 2021, 1:21 PM
marmoute added inline comments.
mercurial/exchange.py
1633–1641

If I understand correctly, this piece of code will be reached only if a the local client flag some sidedata as "locally wanted" without having the means to generate them. right ? The message could be clearer:

m = _(b'sidedata category requested by local side without local support: '%s'%s'")
This revision now requires changes to proceed.Mar 11 2021, 1:21 PM
Alphare marked 3 inline comments as done.Mar 12 2021, 5:34 AM
Alphare updated this revision to Diff 26256.Mar 12 2021, 6:34 AM
marmoute accepted this revision.Mar 12 2021, 1:10 PM
Alphare added a commit: rHGbc2519513ae0: sidedata-exchange: add `wanted_sidedata` and `sidedata_computers` to repos.Mar 15 2021, 2:33 PM
This revision was not accepted when it landed; it landed in state Needs Review.
Closed by commit rHGbc2519513ae0: sidedata-exchange: add `wanted_sidedata` and `sidedata_computers` to repos (authored by Alphare). · Explain Why
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

Diff 26222

mercurial/bundle2.py

def _formatrequirementsparams(requirements): def _formatrequirementsparams(requirements):
requirements = _formatrequirementsspec(requirements) requirements = _formatrequirementsspec(requirements)
params = b"%s%s" % (urlreq.quote(b"requirements="), requirements) params = b"%s%s" % (urlreq.quote(b"requirements="), requirements)
return params return params
def format_remote_wanted_sidedata(repo):
marmouteUnsubmitted
Done

Can you add a small docstring to clarify the semantic of the function ?

marmoute: Can you add a small docstring to clarify the semantic of the function ?
"""Formats a repo's wanted sidedata categories into a bytestring for
capabilities exchange."""
wanted = b""
if repo._wanted_sidedata:
wanted = b','.join(
pycompat.bytestr(c) for c in sorted(repo._wanted_sidedata)
)
return wanted
def read_remote_wanted_sidedata(remote):
sidedata_categories = remote.capable(b'exp-wanted-sidedata')
return read_wanted_sidedata(sidedata_categories)
def read_wanted_sidedata(formatted):
if formatted:
return set(formatted.split(b','))
return set()
def addpartbundlestream2(bundler, repo, **kwargs): def addpartbundlestream2(bundler, repo, **kwargs):
if not kwargs.get('stream', False): if not kwargs.get('stream', False):
return return
if not streamclone.allowservergeneration(repo): if not streamclone.allowservergeneration(repo):
raise error.Abort( raise error.Abort(
_( _(
b'stream data requested but server does not allow ' b'stream data requested but server does not allow '
@parthandler( @parthandler(
b'changegroup', b'changegroup',
( (
b'version', b'version',
b'nbchanges', b'nbchanges',
b'exp-sidedata', b'exp-sidedata',
b'exp-wanted-sidedata',
b'treemanifest', b'treemanifest',
b'targetphase', b'targetphase',
), ),
) )
def handlechangegroup(op, inpart): def handlechangegroup(op, inpart):
"""apply a changegroup part on the repo""" """apply a changegroup part on the repo"""
from . import localrepo from . import localrepo
msg = b"repository is using sidedata but the bundle source do not" msg = b"repository is using sidedata but the bundle source do not"
hint = b'this is currently unsupported' hint = b'this is currently unsupported'
raise error.Abort(msg, hint=hint) raise error.Abort(msg, hint=hint)
extrakwargs = {} extrakwargs = {}
targetphase = inpart.params.get(b'targetphase') targetphase = inpart.params.get(b'targetphase')
if targetphase is not None: if targetphase is not None:
extrakwargs['targetphase'] = int(targetphase) extrakwargs['targetphase'] = int(targetphase)
remote_sidedata = inpart.params.get(b'exp-wanted-sidedata')
extrakwargs['sidedata_categories'] = read_wanted_sidedata(remote_sidedata)
ret = _processchangegroup( ret = _processchangegroup(
op, op,
cg, cg,
tr, tr,
op.source, op.source,
b'bundle2', b'bundle2',
expectedtotal=nbchangesets, expectedtotal=nbchangesets,
**extrakwargs **extrakwargs
) )
part = bundler.newpart(b'changegroup', data=cgdata) part = bundler.newpart(b'changegroup', data=cgdata)
part.addparam(b'version', cgversion) part.addparam(b'version', cgversion)
if scmutil.istreemanifest(repo): if scmutil.istreemanifest(repo):
part.addparam(b'treemanifest', b'1') part.addparam(b'treemanifest', b'1')
if b'exp-sidedata-flag' in repo.requirements: if b'exp-sidedata-flag' in repo.requirements:
part.addparam(b'exp-sidedata', b'1') part.addparam(b'exp-sidedata', b'1')
wanted = format_remote_wanted_sidedata(repo)
part.addparam(b'exp-wanted-sidedata', wanted)
return bundler return bundler

mercurial/changegroup.py

self._builddeltaheader = builddeltaheader self._builddeltaheader = builddeltaheader
self._manifestsend = manifestsend self._manifestsend = manifestsend
self._ellipses = ellipses self._ellipses = ellipses
# Set of capabilities we can use to build the bundle. # Set of capabilities we can use to build the bundle.
if bundlecaps is None: if bundlecaps is None:
bundlecaps = set() bundlecaps = set()
self._bundlecaps = bundlecaps self._bundlecaps = bundlecaps
if remote_sidedata is None:
marmouteUnsubmitted
Done

If you want to test for None, explicitly test for None (is None). Otherwise you introduce subtle bugs for our future self.

marmoute: If you want to test for None, explicitly test for None (`is None`). Otherwise you introduce…
remote_sidedata = set()
self._remote_sidedata = remote_sidedata
self._isshallow = shallow self._isshallow = shallow
self._fullclnodes = fullnodes self._fullclnodes = fullnodes
# Maps ellipsis revs to their roots at the changelog level. # Maps ellipsis revs to their roots at the changelog level.
self._precomputedellipsis = ellipsisroots self._precomputedellipsis = ellipsisroots
self._repo = repo self._repo = repo

mercurial/exchange.py

if not pushop.remote.capable(b'unbundle'): if not pushop.remote.capable(b'unbundle'):
raise error.Abort( raise error.Abort(
_( _(
b'cannot push: destination does not support the ' b'cannot push: destination does not support the '
b'unbundle wire protocol command' b'unbundle wire protocol command'
) )
) )
for category in sorted(bundle2.read_remote_wanted_sidedata(pushop.remote)):
for kind, computers in repo._sidedata_computers.items():
if computers.get(category):
break
else:
raise error.Abort(
_(
b'cannot push: required sidedata category not supported'
b" by this client: '%s'"
)
% pycompat.bytestr(category)
)
# get lock as we might write phase data # get lock as we might write phase data
marmouteUnsubmitted
Not Done

Why do we need multiples layer of loop here . I would expect an API to return a the appropriate object for a categories and that's it.

So, what is going on here ?

marmoute: Why do we need multiples layer of loop here . I would expect an API to return a the appropriate…
AlphareAuthorUnsubmitted
Done

Categories are advertised as a flat list, but can affect any or all of the revlog kinds. This checks that a sidedata computer is registered for at least one revlog kind.

Alphare: Categories are advertised as a flat list, but can affect any or all of the revlog kinds. This…
marmouteUnsubmitted
Done

Ha, yes…

marmoute: Ha, yes…
wlock = lock = None wlock = lock = None
try: try:
# bundle2 push may receive a reply bundle touching bookmarks # bundle2 push may receive a reply bundle touching bookmarks
# requiring the wlock. Take it now to ensure proper ordering. # requiring the wlock. Take it now to ensure proper ordering.
maypushback = pushop.ui.configbool(b'experimental', b'bundle2.pushback') maypushback = pushop.ui.configbool(b'experimental', b'bundle2.pushback')
if ( if (
(not _forcebundle1(pushop)) (not _forcebundle1(pushop))
and maypushback and maypushback
cgversions = [ cgversions = [
v v
for v in cgversions for v in cgversions
if v in changegroup.supportedoutgoingversions(pushop.repo) if v in changegroup.supportedoutgoingversions(pushop.repo)
] ]
if not cgversions: if not cgversions:
raise error.Abort(_(b'no common changegroup version')) raise error.Abort(_(b'no common changegroup version'))
version = max(cgversions) version = max(cgversions)
remote_sidedata = bundle2.read_remote_wanted_sidedata(pushop.remote)
cgstream = changegroup.makestream( cgstream = changegroup.makestream(
pushop.repo, pushop.outgoing, version, b'push' pushop.repo,
pushop.outgoing,
version,
b'push',
bundlecaps=b2caps,
remote_sidedata=remote_sidedata,
) )
cgpart = bundler.newpart(b'changegroup', data=cgstream) cgpart = bundler.newpart(b'changegroup', data=cgstream)
if cgversions: if cgversions:
cgpart.addparam(b'version', version) cgpart.addparam(b'version', version)
if scmutil.istreemanifest(pushop.repo): if scmutil.istreemanifest(pushop.repo):
cgpart.addparam(b'treemanifest', b'1') cgpart.addparam(b'treemanifest', b'1')
if b'exp-sidedata-flag' in pushop.repo.requirements: if b'exp-sidedata-flag' in pushop.repo.requirements:
cgpart.addparam(b'exp-sidedata', b'1') cgpart.addparam(b'exp-sidedata', b'1')
if missing: if missing:
msg = _( msg = _(
b"required features are not" b"required features are not"
b" supported in the destination:" b" supported in the destination:"
b" %s" b" %s"
) % (b', '.join(sorted(missing))) ) % (b', '.join(sorted(missing)))
raise error.Abort(msg) raise error.Abort(msg)
for category in repo._wanted_sidedata:
marmouteUnsubmitted
Done

Same question here.

marmoute: Same question here.
for kind, computers in repo._sidedata_computers.items():
if computers.get(category):
break
else:
# This should never happen since repos are supposed to be able to
# generate the sidedata they require.
raise error.ProgrammingError(
_(
b'cannot pull: required sidedata category not supported'
b" by this client: '%s'"
)
% pycompat.bytestr(category)
)
marmouteUnsubmitted
Done

If I understand correctly, this piece of code will be reached only if a the local client flag some sidedata as "locally wanted" without having the means to generate them. right ? The message could be clearer:

m = _(b'sidedata category requested by local side without local support: '%s'%s'")
marmoute: If I understand correctly, this piece of code will be reached only if a the local client flag…
pullop.trmanager = transactionmanager(repo, b'pull', remote.url()) pullop.trmanager = transactionmanager(repo, b'pull', remote.url())
wlock = util.nullcontextmanager() wlock = util.nullcontextmanager()
if not bookmod.bookmarksinstore(repo): if not bookmod.bookmarksinstore(repo):
wlock = repo.wlock() wlock = repo.wlock()
with wlock, repo.lock(), pullop.trmanager: with wlock, repo.lock(), pullop.trmanager:
if confirm or ( if confirm or (
repo.ui.configbool(b"pull", b"confirm") and not repo.ui.plain() repo.ui.configbool(b"pull", b"confirm") and not repo.ui.plain()
): ):
pullop.repo.ui.status(_(b"requesting all changes\n")) pullop.repo.ui.status(_(b"requesting all changes\n"))
if obsolete.isenabled(pullop.repo, obsolete.exchangeopt): if obsolete.isenabled(pullop.repo, obsolete.exchangeopt):
remoteversions = bundle2.obsmarkersversion(pullop.remotebundle2caps) remoteversions = bundle2.obsmarkersversion(pullop.remotebundle2caps)
if obsolete.commonversion(remoteversions) is not None: if obsolete.commonversion(remoteversions) is not None:
kwargs[b'obsmarkers'] = True kwargs[b'obsmarkers'] = True
pullop.stepsdone.add(b'obsmarkers') pullop.stepsdone.add(b'obsmarkers')
_pullbundle2extraprepare(pullop, kwargs) _pullbundle2extraprepare(pullop, kwargs)
remote_sidedata = bundle2.read_remote_wanted_sidedata(pullop.remote)
if remote_sidedata:
kwargs[b'remote_sidedata'] = remote_sidedata
with pullop.remote.commandexecutor() as e: with pullop.remote.commandexecutor() as e:
args = dict(kwargs) args = dict(kwargs)
args[b'source'] = b'pull' args[b'source'] = b'pull'
bundle = e.callcommand(b'getbundle', args).result() bundle = e.callcommand(b'getbundle', args).result()
try: try:
op = bundle2.bundleoperation( op = bundle2.bundleoperation(
pullop.repo, pullop.gettransaction, source=b'pull' pullop.repo, pullop.gettransaction, source=b'pull'
part.addparam(b'nbchanges', b'%d' % len(outgoing.missing), mandatory=False) part.addparam(b'nbchanges', b'%d' % len(outgoing.missing), mandatory=False)
if scmutil.istreemanifest(repo): if scmutil.istreemanifest(repo):
part.addparam(b'treemanifest', b'1') part.addparam(b'treemanifest', b'1')
if b'exp-sidedata-flag' in repo.requirements: if b'exp-sidedata-flag' in repo.requirements:
part.addparam(b'exp-sidedata', b'1') part.addparam(b'exp-sidedata', b'1')
sidedata = bundle2.format_remote_wanted_sidedata(repo)
part.addparam(b'exp-wanted-sidedata', sidedata)
if ( if (
kwargs.get('narrow', False) kwargs.get('narrow', False)
and kwargs.get('narrow_acl', False) and kwargs.get('narrow_acl', False)
and (include or exclude) and (include or exclude)
): ):
# this is mandatory because otherwise ACL clients won't work # this is mandatory because otherwise ACL clients won't work
narrowspecpart = bundler.newpart(b'Narrow:responsespec') narrowspecpart = bundler.newpart(b'Narrow:responsespec')

mercurial/interfaces/repository.py

pass pass
def debugwireargs(one, two, three=None, four=None, five=None): def debugwireargs(one, two, three=None, four=None, five=None):
pass pass
def savecommitmessage(text): def savecommitmessage(text):
pass pass
def register_sidedata_computer(kind, category, keys, computer):
pass
def register_wanted_sidedata(category):
pass
class completelocalrepository( class completelocalrepository(
ilocalrepositorymain, ilocalrepositoryfilestorage ilocalrepositorymain, ilocalrepositoryfilestorage
): ):
"""Complete interface for a local repository.""" """Complete interface for a local repository."""
class iwireprotocolcommandcacher(interfaceutil.Interface): class iwireprotocolcommandcacher(interfaceutil.Interface):

mercurial/localrepo.py

exchange, exchange,
extensions, extensions,
filelog, filelog,
hook, hook,
lock as lockmod, lock as lockmod,
match as matchmod, match as matchmod,
mergestate as mergestatemod, mergestate as mergestatemod,
mergeutil, mergeutil,
metadata as metadatamod,
namespaces, namespaces,
narrowspec, narrowspec,
obsolete, obsolete,
pathutil, pathutil,
phases, phases,
pushkey, pushkey,
pycompat, pycompat,
rcutil, rcutil,
def __init__(self, repo, caps=None): def __init__(self, repo, caps=None):
super(localpeer, self).__init__() super(localpeer, self).__init__()
if caps is None: if caps is None:
caps = moderncaps.copy() caps = moderncaps.copy()
self._repo = repo.filtered(b'served') self._repo = repo.filtered(b'served')
self.ui = repo.ui self.ui = repo.ui
if repo._wanted_sidedata:
formatted = bundle2.format_remote_wanted_sidedata(repo)
caps.add(b'exp-wanted-sidedata=' + formatted)
self._caps = repo._restrictcapabilities(caps) self._caps = repo._restrictcapabilities(caps)
# Begin of _basepeer interface. # Begin of _basepeer interface.
def url(self): def url(self):
return self._repo.url() return self._repo.url()
def local(self): def local(self):
self._sparsematchercache = {} self._sparsematchercache = {}
self._extrafilterid = repoview.extrafilter(ui) self._extrafilterid = repoview.extrafilter(ui)
self.filecopiesmode = None self.filecopiesmode = None
if requirementsmod.COPIESSDC_REQUIREMENT in self.requirements: if requirementsmod.COPIESSDC_REQUIREMENT in self.requirements:
self.filecopiesmode = b'changeset-sidedata' self.filecopiesmode = b'changeset-sidedata'
self._wanted_sidedata = set()
self._sidedata_computers = {}
metadatamod.set_sidedata_spec_for_repo(self)
def _getvfsward(self, origfunc): def _getvfsward(self, origfunc):
"""build a ward for self.vfs""" """build a ward for self.vfs"""
rref = weakref.ref(self) rref = weakref.ref(self)
def checkvfs(path, mode=None): def checkvfs(path, mode=None):
ret = origfunc(path, mode=mode) ret = origfunc(path, mode=mode)
repo = rref() repo = rref()
if ( if (
def savecommitmessage(self, text): def savecommitmessage(self, text):
fp = self.vfs(b'last-message.txt', b'wb') fp = self.vfs(b'last-message.txt', b'wb')
try: try:
fp.write(text) fp.write(text)
finally: finally:
fp.close() fp.close()
return self.pathto(fp.name[len(self.root) + 1 :]) return self.pathto(fp.name[len(self.root) + 1 :])
def register_wanted_sidedata(self, category):
self._wanted_sidedata.add(pycompat.bytestr(category))
def register_sidedata_computer(self, kind, category, keys, computer):
if kind not in (b"changelog", b"manifest", b"filelog"):
msg = _(b"unexpected revlog kind '%s'.")
raise error.ProgrammingError(msg % kind)
category = pycompat.bytestr(category)
if category in self._sidedata_computers.get(kind, []):
msg = _(
b"cannot register a sidedata computer twice for category '%s'."
)
raise error.ProgrammingError(msg % category)
self._sidedata_computers.setdefault(kind, {})
self._sidedata_computers[kind][category] = (keys, computer)
# used to avoid circular references so destructors work # used to avoid circular references so destructors work
def aftertrans(files): def aftertrans(files):
renamefiles = [tuple(t) for t in files] renamefiles = [tuple(t) for t in files]
def a(): def a():
for vfs, src, dest in renamefiles: for vfs, src, dest in renamefiles:
# if src and dest refer to a same file, vfs.rename is a no-op, # if src and dest refer to a same file, vfs.rename is a no-op,

mercurial/metadata.py

from .node import ( from .node import (
nullid, nullid,
nullrev, nullrev,
) )
from . import ( from . import (
error, error,
pycompat, pycompat,
requirements as requirementsmod,
util, util,
) )
from .revlogutils import ( from .revlogutils import (
flagutil as sidedataflag, flagutil as sidedataflag,
sidedata as sidedatamod, sidedata as sidedatamod,
) )
def _getsidedata(srcrepo, rev): def _getsidedata(srcrepo, rev):
ctx = srcrepo[rev] ctx = srcrepo[rev]
files = compute_all_files_changes(ctx) files = compute_all_files_changes(ctx)
return encode_files_sidedata(files), files.has_copies_info return encode_files_sidedata(files), files.has_copies_info
def copies_sidedata_computer(repo, revlog, rev, existing_sidedata):
return _getsidedata(repo, rev)[0]
def set_sidedata_spec_for_repo(repo):
if requirementsmod.COPIESSDC_REQUIREMENT in repo.requirements:
repo.register_wanted_sidedata(sidedatamod.SD_FILES)
repo.register_sidedata_computer(
b"changelog",
sidedatamod.SD_FILES,
(sidedatamod.SD_FILES,),
copies_sidedata_computer,
)
def getsidedataadder(srcrepo, destrepo): def getsidedataadder(srcrepo, destrepo):
use_w = srcrepo.ui.configbool(b'experimental', b'worker.repository-upgrade') use_w = srcrepo.ui.configbool(b'experimental', b'worker.repository-upgrade')
if pycompat.iswindows or not use_w: if pycompat.iswindows or not use_w:
marmouteUnsubmitted
Done

We should use a registrar here (or at least a small helper function), that way, changing the implementation/processing will be simple in the futur.

What do you think ?

marmoute: We should use a registrar here (or at least a small helper function), that way, changing the…
return _get_simple_sidedata_adder(srcrepo, destrepo) return _get_simple_sidedata_adder(srcrepo, destrepo)
else: else:
return _get_worker_sidedata_adder(srcrepo, destrepo) return _get_worker_sidedata_adder(srcrepo, destrepo)
def _sidedata_worker(srcrepo, revs_queue, sidedata_queue, tokens): def _sidedata_worker(srcrepo, revs_queue, sidedata_queue, tokens):
"""The function used by worker precomputing sidedata """The function used by worker precomputing sidedata

mercurial/statichttprepo.py

vfsclass = build_opener(ui, authinfo) vfsclass = build_opener(ui, authinfo)
self.vfs = vfsclass(self.path) self.vfs = vfsclass(self.path)
self.cachevfs = vfsclass(self.vfs.join(b'cache')) self.cachevfs = vfsclass(self.vfs.join(b'cache'))
self._phasedefaults = [] self._phasedefaults = []
self.names = namespaces.namespaces() self.names = namespaces.namespaces()
self.filtername = None self.filtername = None
self._extrafilterid = None self._extrafilterid = None
self._wanted_sidedata = set()
try: try:
requirements = set(self.vfs.read(b'requires').splitlines()) requirements = set(self.vfs.read(b'requires').splitlines())
except IOError as inst: except IOError as inst:
if inst.errno != errno.ENOENT: if inst.errno != errno.ENOENT:
raise raise
requirements = set() requirements = set()

tests/test-check-interfaces.py

% (o.__class__.__name__, attr) % (o.__class__.__name__, attr)
) )
# Facilitates testing localpeer. # Facilitates testing localpeer.
class dummyrepo(object): class dummyrepo(object):
def __init__(self): def __init__(self):
self.ui = uimod.ui() self.ui = uimod.ui()
self._wanted_sidedata = set()
def filtered(self, name): def filtered(self, name):
pass pass
def _restrictcapabilities(self, caps): def _restrictcapabilities(self, caps):
pass pass