This is an archive of the discontinued Mercurial Phabricator instance.

Differential D1954

bundle: add the possibility to bundle a stream v2 part
ClosedPublic

Authored by lothiraldan on Jan 31 2018, 11:25 AM.

Details

Reviewers
indygreg
Group Reviewers
hg-reviewers
Commits
rHG6f467adf9f05: bundle: add the possibility to bundle a stream v2 part
Summary

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

lothiraldan created this revision.Jan 31 2018, 11:25 AM
lothiraldan updated this revision to Diff 5152.Feb 2 2018, 4:30 AM
lothiraldan added a comment.Feb 2 2018, 4:33 AM

Now that we have more time to review this series, this is the part I am the least comfortable with. I think we could/should extract the stream part generator somewhere else than exchange.py. Maybe bundle2.py directly but I'm not sure. Any ideas?

indygreg requested changes to this revision.Mar 19 2018, 4:39 PM
indygreg added a subscriber: indygreg.
In D1954#33576, @lothiraldan wrote:

Now that we have more time to review this series, this is the part I am the least comfortable with. I think we could/should extract the stream part generator somewhere else than exchange.py. Maybe bundle2.py directly but I'm not sure. Any ideas?

Yup. You've stumbled upon something: we have 2 variations of code that produce bundle2 data. We have bundle2._addpartsfromopts() and exchange.getbundlechunks(). There is redundancy between the two. For example, creation of the changegroup part.

I would *very* much like to see this code unified. Probably in bundle2.py. What makes it difficult to unify is that each implementation is determining *what data should I generate* via different mechanisms. bundle2.py is using a dict of [content] options. exchange.py is using bundle2 capabilities, which come from the client.

If you squint hard enough, these are both almost the same thing. I would like to think that we could normalize b2caps and content options down to a single data structure that described what to generate and we could pass that into a common *generate bundle2 content* function. I /think/ we want to unify on content options. But I'm not sure.

Things are a bit more difficult than they should be because bundle2 is both used for data at rest and in flight. It represents both static data and the request/results of an RPC call over the wire protocol. e.g. the check:*, error:*, and reply:* parts don't make a lot of sense for local-only bundle2 bundles. Those are there mainly to facilitate wire protocol features. So any unified *create a bundle* or *apply a bundle* functionality need to account for the fact that there are special bundle2 parts and functionality that are unique to the wire protocol.

Anyway, I'd fix the immediate concern of having to import exchange by moving the code for creating this bundle2 part into bundle2.py and having the part generator in exchange call into it.

This revision now requires changes to proceed.Mar 19 2018, 4:39 PM
lothiraldan updated this revision to Diff 7376.Mar 30 2018, 8:55 AM
indygreg accepted this revision.Mar 30 2018, 1:10 PM
This revision is now accepted and ready to land.Mar 30 2018, 1:10 PM
Closed by commit rHG6f467adf9f05: bundle: add the possibility to bundle a stream v2 part (authored by lothiraldan). · Explain WhyMar 30 2018, 1:14 PM
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

PathPackages
Mmercurial/bundle2.py (34 lines)
Mmercurial/exchange.py (36 lines)

Diff 7385

mercurial/bundle2.py

part.addparam('version', cg.version) part.addparam('version', cg.version)
if 'clcount' in cg.extras: if 'clcount' in cg.extras:
part.addparam('nbchanges', '%d' % cg.extras['clcount'], part.addparam('nbchanges', '%d' % cg.extras['clcount'],
mandatory=False) mandatory=False)
if opts.get('phases') and repo.revs('%ln and secret()', if opts.get('phases') and repo.revs('%ln and secret()',
outgoing.missingheads): outgoing.missingheads):
part.addparam('targetphase', '%d' % phases.secret, mandatory=False) part.addparam('targetphase', '%d' % phases.secret, mandatory=False)
if opts.get('streamv2', False):
addpartbundlestream2(bundler, repo, stream=True)
if opts.get('tagsfnodescache', True): if opts.get('tagsfnodescache', True):
addparttagsfnodescache(repo, bundler, outgoing) addparttagsfnodescache(repo, bundler, outgoing)
if opts.get('revbranchcache', True): if opts.get('revbranchcache', True):
addpartrevbranchcache(repo, bundler, outgoing) addpartrevbranchcache(repo, bundler, outgoing)
if opts.get('obsolescence', False): if opts.get('obsolescence', False):
obsmarkers = repo.obsstore.relevantmarkers(outgoing.missing) obsmarkers = repo.obsstore.relevantmarkers(outgoing.missing)
yield utf8branch yield utf8branch
for n in sorted(nodes): for n in sorted(nodes):
yield n yield n
for n in sorted(closed): for n in sorted(closed):
yield n yield n
bundler.newpart('cache:rev-branch-cache', data=generate()) bundler.newpart('cache:rev-branch-cache', data=generate())
def _formatrequirementsspec(requirements):
return urlreq.quote(','.join(sorted(requirements)))
def _formatrequirementsparams(requirements):
requirements = _formatrequirementsspec(requirements)
params = "%s%s" % (urlreq.quote("requirements="), requirements)
return params
def addpartbundlestream2(bundler, repo, **kwargs):
if not kwargs.get('stream', False):
return
if not streamclone.allowservergeneration(repo):
raise error.Abort(_('stream data requested but server does not allow '
'this feature'),
hint=_('well-behaved clients should not be '
'requesting stream data from servers not '
'advertising it; the client may be buggy'))
# Stream clones don't compress well. And compression undermines a
# goal of stream clones, which is to be fast. Communicate the desire
# to avoid compression to consumers of the bundle.
bundler.prefercompressed = False
filecount, bytecount, it = streamclone.generatev2(repo)
requirements = _formatrequirementsspec(repo.requirements)
part = bundler.newpart('stream2', data=it)
part.addparam('bytecount', '%d' % bytecount, mandatory=True)
part.addparam('filecount', '%d' % filecount, mandatory=True)
part.addparam('requirements', requirements, mandatory=True)
def buildobsmarkerspart(bundler, markers): def buildobsmarkerspart(bundler, markers):
"""add an obsmarker part to the bundler with <markers> """add an obsmarker part to the bundler with <markers>
No part is created if markers is empty. No part is created if markers is empty.
Raises ValueError if the bundler doesn't support any known obsmarker format. Raises ValueError if the bundler doesn't support any known obsmarker format.
""" """
if not markers: if not markers:
return None return None

mercurial/exchange.py

return changegroup.cg1unpacker(fh, alg) return changegroup.cg1unpacker(fh, alg)
elif version.startswith('2'): elif version.startswith('2'):
return bundle2.getunbundler(ui, fh, magicstring=magic + version) return bundle2.getunbundler(ui, fh, magicstring=magic + version)
elif version == 'S1': elif version == 'S1':
return streamclone.streamcloneapplier(fh) return streamclone.streamcloneapplier(fh)
else: else:
raise error.Abort(_('%s: unknown bundle version %s') % (fname, version)) raise error.Abort(_('%s: unknown bundle version %s') % (fname, version))
def _formatrequirementsspec(requirements):
return urlreq.quote(','.join(sorted(requirements)))
def _formatrequirementsparams(requirements):
requirements = _formatrequirementsspec(requirements)
params = "%s%s" % (urlreq.quote("requirements="), requirements)
return params
def getbundlespec(ui, fh): def getbundlespec(ui, fh):
"""Infer the bundlespec from a bundle file handle. """Infer the bundlespec from a bundle file handle.
The input file handle is seeked and the original seek position is not The input file handle is seeked and the original seek position is not
restored. restored.
""" """
def speccompression(alg): def speccompression(alg):
try: try:
if not version: if not version:
raise error.Abort(_('could not identify changegroup version in ' raise error.Abort(_('could not identify changegroup version in '
'bundle')) 'bundle'))
return '%s-%s' % (comp, version) return '%s-%s' % (comp, version)
elif isinstance(b, streamclone.streamcloneapplier): elif isinstance(b, streamclone.streamcloneapplier):
requirements = streamclone.readbundle1header(fh)[2] requirements = streamclone.readbundle1header(fh)[2]
return 'none-packed1;%s' % _formatrequirementsparams(requirements) formatted = bundle2._formatrequirementsparams(requirements)
return 'none-packed1;%s' % formatted
else: else:
raise error.Abort(_('unknown bundle type: %s') % b) raise error.Abort(_('unknown bundle type: %s') % b)
def _computeoutgoing(repo, heads, common): def _computeoutgoing(repo, heads, common):
"""Computes which revs are outgoing given a set of common """Computes which revs are outgoing given a set of common
and a set of heads. and a set of heads.
This is a separate function so extensions can have access to This is a separate function so extensions can have access to
func(bundler, repo, source, bundlecaps=bundlecaps, b2caps=b2caps, func(bundler, repo, source, bundlecaps=bundlecaps, b2caps=b2caps,
**pycompat.strkwargs(kwargs)) **pycompat.strkwargs(kwargs))
info['prefercompressed'] = bundler.prefercompressed info['prefercompressed'] = bundler.prefercompressed
return info, bundler.getchunks() return info, bundler.getchunks()
@getbundle2partsgenerator('stream2') @getbundle2partsgenerator('stream2')
def _getbundlestream2(bundler, repo, source, bundlecaps=None, def _getbundlestream2(bundler, repo, *args, **kwargs):
b2caps=None, heads=None, common=None, **kwargs): return bundle2.addpartbundlestream2(bundler, repo, **kwargs)
if not kwargs.get('stream', False):
return
if not streamclone.allowservergeneration(repo):
raise error.Abort(_('stream data requested but server does not allow '
'this feature'),
hint=_('well-behaved clients should not be '
'requesting stream data from servers not '
'advertising it; the client may be buggy'))
# Stream clones don't compress well. And compression undermines a
# goal of stream clones, which is to be fast. Communicate the desire
# to avoid compression to consumers of the bundle.
bundler.prefercompressed = False
filecount, bytecount, it = streamclone.generatev2(repo)
requirements = _formatrequirementsspec(repo.requirements)
part = bundler.newpart('stream2', data=it)
part.addparam('bytecount', '%d' % bytecount, mandatory=True)
part.addparam('filecount', '%d' % filecount, mandatory=True)
part.addparam('requirements', requirements, mandatory=True)
@getbundle2partsgenerator('changegroup') @getbundle2partsgenerator('changegroup')
def _getbundlechangegrouppart(bundler, repo, source, bundlecaps=None, def _getbundlechangegrouppart(bundler, repo, source, bundlecaps=None,
b2caps=None, heads=None, common=None, **kwargs): b2caps=None, heads=None, common=None, **kwargs):
"""add a changegroup part to the requested bundle""" """add a changegroup part to the requested bundle"""
cgstream = None cgstream = None
if kwargs.get(r'cg', True): if kwargs.get(r'cg', True):
# build changegroup bundle here. # build changegroup bundle here.