( )⚙ D9208 clonebundles: move a bundle of clone bundle related code to a new module

This is an archive of the discontinued Mercurial Phabricator instance.

Differential D9208

clonebundles: move a bundle of clone bundle related code to a new module
ClosedPublic

Authored by marmoute on Oct 14 2020, 7:34 PM.

Details

Reviewers
None
Group Reviewers
hg-reviewers
Commits
rHG74271829ddc0: clonebundles: move a bundle of clone bundle related code to a new module
Summary

In the process on general clone bundle automatically, we need to make some
function available more widely. This is a good opportunity to extract a
significant amount of code from mercurial.exchange into a new
mercurial.bundlecaches. This make mercurial.exchange move under the 3K line
range (hooray…).

The module is called bundlecaches because I expect it to be eventually useful
for more than just clone bundle (like pull bunbles).

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

marmoute created this revision.Oct 14 2020, 7:34 PM
pulkit added a subscriber: pulkit.Oct 15 2020, 8:16 AM
pulkit added inline comments.
tests/test-pull-bundle.t
55 ↗(On Diff #23209)

seems like this and next one are unrelated change.

marmoute added a comment.Oct 15 2020, 9:51 AM

Ha yes, they sneaking in while debugging.

marmoute added a commit: rHG74271829ddc0: clonebundles: move a bundle of clone bundle related code to a new module.Oct 28 2020, 9:59 PM
This revision was not accepted when it landed; it landed in state Needs Review.
Closed by commit rHG74271829ddc0: clonebundles: move a bundle of clone bundle related code to a new module (authored by marmoute). · Explain Why
This revision was automatically updated to reflect the committed changes.
martinvonz added a subscriber: martinvonz.Oct 28 2020, 11:48 PM

Can you send a follow-up to mention the API change in relnotes/next? (I've said before that I'm skeptical that anyone reads the list of API changes, but I think the consensus has been to document it.)

Revision Contents
Changeset List

Diff 23338

hgext/lfs/__init__.py

from __future__ import absolute_import from __future__ import absolute_import
import sys import sys
from mercurial.i18n import _ from mercurial.i18n import _
from mercurial import ( from mercurial import (
bundlecaches,
config, config,
context, context,
error, error,
exchange,
extensions, extensions,
exthelper, exthelper,
filelog, filelog,
filesetlang, filesetlang,
localrepo, localrepo,
minifileset, minifileset,
node, node,
pycompat, pycompat,
context.basefilectx.islfs = wrapper.filectxislfs context.basefilectx.islfs = wrapper.filectxislfs
scmutil.fileprefetchhooks.add(b'lfs', wrapper._prefetchfiles) scmutil.fileprefetchhooks.add(b'lfs', wrapper._prefetchfiles)
# Make bundle choose changegroup3 instead of changegroup2. This affects # Make bundle choose changegroup3 instead of changegroup2. This affects
# "hg bundle" command. Note: it does not cover all bundle formats like # "hg bundle" command. Note: it does not cover all bundle formats like
# "packed1". Using "packed1" with lfs will likely cause trouble. # "packed1". Using "packed1" with lfs will likely cause trouble.
exchange._bundlespeccontentopts[b"v2"][b"cg.version"] = b"03" bundlecaches._bundlespeccontentopts[b"v2"][b"cg.version"] = b"03"
@eh.filesetpredicate(b'lfs()') @eh.filesetpredicate(b'lfs()')
def lfsfileset(mctx, x): def lfsfileset(mctx, x):
"""File that uses LFS storage.""" """File that uses LFS storage."""
# i18n: "lfs" is a keyword # i18n: "lfs" is a keyword
filesetlang.getargs(x, 0, 0, _(b"lfs takes no arguments")) filesetlang.getargs(x, 0, 0, _(b"lfs takes no arguments"))
ctx = mctx.ctx ctx = mctx.ctx

mercurial/bundlecaches.py

  • This file was added.
# bundlecaches.py - utility to deal with pre-computed bundle for servers
#
# This software may be used and distributed according to the terms of the
# GNU General Public License version 2 or any later version.
from .i18n import _
from .thirdparty import attr
from . import (
error,
sslutil,
util,
)
from .utils import stringutil
urlreq = util.urlreq
@attr.s
class bundlespec(object):
compression = attr.ib()
wirecompression = attr.ib()
version = attr.ib()
wireversion = attr.ib()
params = attr.ib()
contentopts = attr.ib()
# Maps bundle version human names to changegroup versions.
_bundlespeccgversions = {
b'v1': b'01',
b'v2': b'02',
b'packed1': b's1',
b'bundle2': b'02', # legacy
}
# Maps bundle version with content opts to choose which part to bundle
_bundlespeccontentopts = {
b'v1': {
b'changegroup': True,
b'cg.version': b'01',
b'obsolescence': False,
b'phases': False,
b'tagsfnodescache': False,
b'revbranchcache': False,
},
b'v2': {
b'changegroup': True,
b'cg.version': b'02',
b'obsolescence': False,
b'phases': False,
b'tagsfnodescache': True,
b'revbranchcache': True,
},
b'packed1': {b'cg.version': b's1'},
}
_bundlespeccontentopts[b'bundle2'] = _bundlespeccontentopts[b'v2']
_bundlespecvariants = {
b"streamv2": {
b"changegroup": False,
b"streamv2": True,
b"tagsfnodescache": False,
b"revbranchcache": False,
}
}
# Compression engines allowed in version 1. THIS SHOULD NEVER CHANGE.
_bundlespecv1compengines = {b'gzip', b'bzip2', b'none'}
def parsebundlespec(repo, spec, strict=True):
"""Parse a bundle string specification into parts.
Bundle specifications denote a well-defined bundle/exchange format.
The content of a given specification should not change over time in
order to ensure that bundles produced by a newer version of Mercurial are
readable from an older version.
The string currently has the form:
<compression>-<type>[;<parameter0>[;<parameter1>]]
Where <compression> is one of the supported compression formats
and <type> is (currently) a version string. A ";" can follow the type and
all text afterwards is interpreted as URI encoded, ";" delimited key=value
pairs.
If ``strict`` is True (the default) <compression> is required. Otherwise,
it is optional.
Returns a bundlespec object of (compression, version, parameters).
Compression will be ``None`` if not in strict mode and a compression isn't
defined.
An ``InvalidBundleSpecification`` is raised when the specification is
not syntactically well formed.
An ``UnsupportedBundleSpecification`` is raised when the compression or
bundle type/version is not recognized.
Note: this function will likely eventually return a more complex data
structure, including bundle2 part information.
"""
def parseparams(s):
if b';' not in s:
return s, {}
params = {}
version, paramstr = s.split(b';', 1)
for p in paramstr.split(b';'):
if b'=' not in p:
raise error.InvalidBundleSpecification(
_(
b'invalid bundle specification: '
b'missing "=" in parameter: %s'
)
% p
)
key, value = p.split(b'=', 1)
key = urlreq.unquote(key)
value = urlreq.unquote(value)
params[key] = value
return version, params
if strict and b'-' not in spec:
raise error.InvalidBundleSpecification(
_(
b'invalid bundle specification; '
b'must be prefixed with compression: %s'
)
% spec
)
if b'-' in spec:
compression, version = spec.split(b'-', 1)
if compression not in util.compengines.supportedbundlenames:
raise error.UnsupportedBundleSpecification(
_(b'%s compression is not supported') % compression
)
version, params = parseparams(version)
if version not in _bundlespeccgversions:
raise error.UnsupportedBundleSpecification(
_(b'%s is not a recognized bundle version') % version
)
else:
# Value could be just the compression or just the version, in which
# case some defaults are assumed (but only when not in strict mode).
assert not strict
spec, params = parseparams(spec)
if spec in util.compengines.supportedbundlenames:
compression = spec
version = b'v1'
# Generaldelta repos require v2.
if b'generaldelta' in repo.requirements:
version = b'v2'
# Modern compression engines require v2.
if compression not in _bundlespecv1compengines:
version = b'v2'
elif spec in _bundlespeccgversions:
if spec == b'packed1':
compression = b'none'
else:
compression = b'bzip2'
version = spec
else:
raise error.UnsupportedBundleSpecification(
_(b'%s is not a recognized bundle specification') % spec
)
# Bundle version 1 only supports a known set of compression engines.
if version == b'v1' and compression not in _bundlespecv1compengines:
raise error.UnsupportedBundleSpecification(
_(b'compression engine %s is not supported on v1 bundles')
% compression
)
# The specification for packed1 can optionally declare the data formats
# required to apply it. If we see this metadata, compare against what the
# repo supports and error if the bundle isn't compatible.
if version == b'packed1' and b'requirements' in params:
requirements = set(params[b'requirements'].split(b','))
missingreqs = requirements - repo.supportedformats
if missingreqs:
raise error.UnsupportedBundleSpecification(
_(b'missing support for repository features: %s')
% b', '.join(sorted(missingreqs))
)
# Compute contentopts based on the version
contentopts = _bundlespeccontentopts.get(version, {}).copy()
# Process the variants
if b"stream" in params and params[b"stream"] == b"v2":
variant = _bundlespecvariants[b"streamv2"]
contentopts.update(variant)
engine = util.compengines.forbundlename(compression)
compression, wirecompression = engine.bundletype()
wireversion = _bundlespeccgversions[version]
return bundlespec(
compression, wirecompression, version, wireversion, params, contentopts
)
def parseclonebundlesmanifest(repo, s):
"""Parses the raw text of a clone bundles manifest.
Returns a list of dicts. The dicts have a ``URL`` key corresponding
to the URL and other keys are the attributes for the entry.
"""
m = []
for line in s.splitlines():
fields = line.split()
if not fields:
continue
attrs = {b'URL': fields[0]}
for rawattr in fields[1:]:
key, value = rawattr.split(b'=', 1)
key = util.urlreq.unquote(key)
value = util.urlreq.unquote(value)
attrs[key] = value
# Parse BUNDLESPEC into components. This makes client-side
# preferences easier to specify since you can prefer a single
# component of the BUNDLESPEC.
if key == b'BUNDLESPEC':
try:
bundlespec = parsebundlespec(repo, value)
attrs[b'COMPRESSION'] = bundlespec.compression
attrs[b'VERSION'] = bundlespec.version
except error.InvalidBundleSpecification:
pass
except error.UnsupportedBundleSpecification:
pass
m.append(attrs)
return m
def isstreamclonespec(bundlespec):
# Stream clone v1
if bundlespec.wirecompression == b'UN' and bundlespec.wireversion == b's1':
return True
# Stream clone v2
if (
bundlespec.wirecompression == b'UN'
and bundlespec.wireversion == b'02'
and bundlespec.contentopts.get(b'streamv2')
):
return True
return False
def filterclonebundleentries(repo, entries, streamclonerequested=False):
"""Remove incompatible clone bundle manifest entries.
Accepts a list of entries parsed with ``parseclonebundlesmanifest``
and returns a new list consisting of only the entries that this client
should be able to apply.
There is no guarantee we'll be able to apply all returned entries because
the metadata we use to filter on may be missing or wrong.
"""
newentries = []
for entry in entries:
spec = entry.get(b'BUNDLESPEC')
if spec:
try:
bundlespec = parsebundlespec(repo, spec, strict=True)
# If a stream clone was requested, filter out non-streamclone
# entries.
if streamclonerequested and not isstreamclonespec(bundlespec):
repo.ui.debug(
b'filtering %s because not a stream clone\n'
% entry[b'URL']
)
continue
except error.InvalidBundleSpecification as e:
repo.ui.debug(stringutil.forcebytestr(e) + b'\n')
continue
except error.UnsupportedBundleSpecification as e:
repo.ui.debug(
b'filtering %s because unsupported bundle '
b'spec: %s\n' % (entry[b'URL'], stringutil.forcebytestr(e))
)
continue
# If we don't have a spec and requested a stream clone, we don't know
# what the entry is so don't attempt to apply it.
elif streamclonerequested:
repo.ui.debug(
b'filtering %s because cannot determine if a stream '
b'clone bundle\n' % entry[b'URL']
)
continue
if b'REQUIRESNI' in entry and not sslutil.hassni:
repo.ui.debug(
b'filtering %s because SNI not supported\n' % entry[b'URL']
)
continue
if b'REQUIREDRAM' in entry:
try:
requiredram = util.sizetoint(entry[b'REQUIREDRAM'])
except error.ParseError:
repo.ui.debug(
b'filtering %s due to a bad REQUIREDRAM attribute\n'
% entry[b'URL']
)
continue
actualram = repo.ui.estimatememory()
if actualram is not None and actualram * 0.66 < requiredram:
repo.ui.debug(
b'filtering %s as it needs more than 2/3 of system memory\n'
% entry[b'URL']
)
continue
newentries.append(entry)
return newentries
class clonebundleentry(object):
"""Represents an item in a clone bundles manifest.
This rich class is needed to support sorting since sorted() in Python 3
doesn't support ``cmp`` and our comparison is complex enough that ``key=``
won't work.
"""
def __init__(self, value, prefers):
self.value = value
self.prefers = prefers
def _cmp(self, other):
for prefkey, prefvalue in self.prefers:
avalue = self.value.get(prefkey)
bvalue = other.value.get(prefkey)
# Special case for b missing attribute and a matches exactly.
if avalue is not None and bvalue is None and avalue == prefvalue:
return -1
# Special case for a missing attribute and b matches exactly.
if bvalue is not None and avalue is None and bvalue == prefvalue:
return 1
# We can't compare unless attribute present on both.
if avalue is None or bvalue is None:
continue
# Same values should fall back to next attribute.
if avalue == bvalue:
continue
# Exact matches come first.
if avalue == prefvalue:
return -1
if bvalue == prefvalue:
return 1
# Fall back to next attribute.
continue
# If we got here we couldn't sort by attributes and prefers. Fall
# back to index order.
return 0
def __lt__(self, other):
return self._cmp(other) < 0
def __gt__(self, other):
return self._cmp(other) > 0
def __eq__(self, other):
return self._cmp(other) == 0
def __le__(self, other):
return self._cmp(other) <= 0
def __ge__(self, other):
return self._cmp(other) >= 0
def __ne__(self, other):
return self._cmp(other) != 0
def sortclonebundleentries(ui, entries):
prefers = ui.configlist(b'ui', b'clonebundleprefers')
if not prefers:
return list(entries)
def _split(p):
if b'=' not in p:
hint = _(b"each comma separated item should be key=value pairs")
raise error.Abort(
_(b"invalid ui.clonebundleprefers item: %s") % p, hint=hint
)
return p.split(b'=', 1)
prefers = [_split(p) for p in prefers]
items = sorted(clonebundleentry(v, prefers) for v in entries)
return [i.value for i in items]

mercurial/commands.py

wdirhex, wdirhex,
wdirrev, wdirrev,
) )
from .pycompat import open from .pycompat import open
from . import ( from . import (
archival, archival,
bookmarks, bookmarks,
bundle2, bundle2,
bundlecaches,
changegroup, changegroup,
cmdutil, cmdutil,
copies, copies,
debugcommands as debugcommandsmod, debugcommands as debugcommandsmod,
destutil, destutil,
dirstateguard, dirstateguard,
discovery, discovery,
encoding, encoding,
if b'rev' in opts: if b'rev' in opts:
revstrings = opts[b'rev'] revstrings = opts[b'rev']
revs = scmutil.revrange(repo, revstrings) revs = scmutil.revrange(repo, revstrings)
if revstrings and not revs: if revstrings and not revs:
raise error.Abort(_(b'no commits to bundle')) raise error.Abort(_(b'no commits to bundle'))
bundletype = opts.get(b'type', b'bzip2').lower() bundletype = opts.get(b'type', b'bzip2').lower()
try: try:
bundlespec = exchange.parsebundlespec(repo, bundletype, strict=False) bundlespec = bundlecaches.parsebundlespec(
repo, bundletype, strict=False
)
except error.UnsupportedBundleSpecification as e: except error.UnsupportedBundleSpecification as e:
raise error.Abort( raise error.Abort(
pycompat.bytestr(e), pycompat.bytestr(e),
hint=_(b"see 'hg help bundlespec' for supported values for --type"), hint=_(b"see 'hg help bundlespec' for supported values for --type"),
) )
cgversion = bundlespec.contentopts[b"cg.version"] cgversion = bundlespec.contentopts[b"cg.version"]
# Packed bundles are a pseudo bundle format for now. # Packed bundles are a pseudo bundle format for now.

mercurial/exchange.py

import weakref import weakref
from .i18n import _ from .i18n import _
from .node import ( from .node import (
hex, hex,
nullid, nullid,
nullrev, nullrev,
) )
from .thirdparty import attr
from . import ( from . import (
bookmarks as bookmod, bookmarks as bookmod,
bundle2, bundle2,
bundlecaches,
changegroup, changegroup,
discovery, discovery,
error, error,
exchangev2, exchangev2,
lock as lockmod, lock as lockmod,
logexchange, logexchange,
narrowspec, narrowspec,
obsolete, obsolete,
obsutil, obsutil,
phases, phases,
pushkey, pushkey,
pycompat, pycompat,
requirements, requirements,
scmutil, scmutil,
sslutil,
streamclone, streamclone,
url as urlmod, url as urlmod,
util, util,
wireprototypes, wireprototypes,
) )
from .utils import ( from .utils import (
hashutil, hashutil,
stringutil, stringutil,
) )
urlerr = util.urlerr urlerr = util.urlerr
urlreq = util.urlreq urlreq = util.urlreq
_NARROWACL_SECTION = b'narrowacl' _NARROWACL_SECTION = b'narrowacl'
# Maps bundle version human names to changegroup versions.
_bundlespeccgversions = {
b'v1': b'01',
b'v2': b'02',
b'packed1': b's1',
b'bundle2': b'02', # legacy
}
# Maps bundle version with content opts to choose which part to bundle
_bundlespeccontentopts = {
b'v1': {
b'changegroup': True,
b'cg.version': b'01',
b'obsolescence': False,
b'phases': False,
b'tagsfnodescache': False,
b'revbranchcache': False,
},
b'v2': {
b'changegroup': True,
b'cg.version': b'02',
b'obsolescence': False,
b'phases': False,
b'tagsfnodescache': True,
b'revbranchcache': True,
},
b'packed1': {b'cg.version': b's1'},
}
_bundlespeccontentopts[b'bundle2'] = _bundlespeccontentopts[b'v2']
_bundlespecvariants = {
b"streamv2": {
b"changegroup": False,
b"streamv2": True,
b"tagsfnodescache": False,
b"revbranchcache": False,
}
}
# Compression engines allowed in version 1. THIS SHOULD NEVER CHANGE.
_bundlespecv1compengines = {b'gzip', b'bzip2', b'none'}
@attr.s
class bundlespec(object):
compression = attr.ib()
wirecompression = attr.ib()
version = attr.ib()
wireversion = attr.ib()
params = attr.ib()
contentopts = attr.ib()
def parsebundlespec(repo, spec, strict=True):
"""Parse a bundle string specification into parts.
Bundle specifications denote a well-defined bundle/exchange format.
The content of a given specification should not change over time in
order to ensure that bundles produced by a newer version of Mercurial are
readable from an older version.
The string currently has the form:
<compression>-<type>[;<parameter0>[;<parameter1>]]
Where <compression> is one of the supported compression formats
and <type> is (currently) a version string. A ";" can follow the type and
all text afterwards is interpreted as URI encoded, ";" delimited key=value
pairs.
If ``strict`` is True (the default) <compression> is required. Otherwise,
it is optional.
Returns a bundlespec object of (compression, version, parameters).
Compression will be ``None`` if not in strict mode and a compression isn't
defined.
An ``InvalidBundleSpecification`` is raised when the specification is
not syntactically well formed.
An ``UnsupportedBundleSpecification`` is raised when the compression or
bundle type/version is not recognized.
Note: this function will likely eventually return a more complex data
structure, including bundle2 part information.
"""
def parseparams(s):
if b';' not in s:
return s, {}
params = {}
version, paramstr = s.split(b';', 1)
for p in paramstr.split(b';'):
if b'=' not in p:
raise error.InvalidBundleSpecification(
_(
b'invalid bundle specification: '
b'missing "=" in parameter: %s'
)
% p
)
key, value = p.split(b'=', 1)
key = urlreq.unquote(key)
value = urlreq.unquote(value)
params[key] = value
return version, params
if strict and b'-' not in spec:
raise error.InvalidBundleSpecification(
_(
b'invalid bundle specification; '
b'must be prefixed with compression: %s'
)
% spec
)
if b'-' in spec:
compression, version = spec.split(b'-', 1)
if compression not in util.compengines.supportedbundlenames:
raise error.UnsupportedBundleSpecification(
_(b'%s compression is not supported') % compression
)
version, params = parseparams(version)
if version not in _bundlespeccgversions:
raise error.UnsupportedBundleSpecification(
_(b'%s is not a recognized bundle version') % version
)
else:
# Value could be just the compression or just the version, in which
# case some defaults are assumed (but only when not in strict mode).
assert not strict
spec, params = parseparams(spec)
if spec in util.compengines.supportedbundlenames:
compression = spec
version = b'v1'
# Generaldelta repos require v2.
if b'generaldelta' in repo.requirements:
version = b'v2'
# Modern compression engines require v2.
if compression not in _bundlespecv1compengines:
version = b'v2'
elif spec in _bundlespeccgversions:
if spec == b'packed1':
compression = b'none'
else:
compression = b'bzip2'
version = spec
else:
raise error.UnsupportedBundleSpecification(
_(b'%s is not a recognized bundle specification') % spec
)
# Bundle version 1 only supports a known set of compression engines.
if version == b'v1' and compression not in _bundlespecv1compengines:
raise error.UnsupportedBundleSpecification(
_(b'compression engine %s is not supported on v1 bundles')
% compression
)
# The specification for packed1 can optionally declare the data formats
# required to apply it. If we see this metadata, compare against what the
# repo supports and error if the bundle isn't compatible.
if version == b'packed1' and b'requirements' in params:
requirements = set(params[b'requirements'].split(b','))
missingreqs = requirements - repo.supportedformats
if missingreqs:
raise error.UnsupportedBundleSpecification(
_(b'missing support for repository features: %s')
% b', '.join(sorted(missingreqs))
)
# Compute contentopts based on the version
contentopts = _bundlespeccontentopts.get(version, {}).copy()
# Process the variants
if b"stream" in params and params[b"stream"] == b"v2":
variant = _bundlespecvariants[b"streamv2"]
contentopts.update(variant)
engine = util.compengines.forbundlename(compression)
compression, wirecompression = engine.bundletype()
wireversion = _bundlespeccgversions[version]
return bundlespec(
compression, wirecompression, version, wireversion, params, contentopts
)
def readbundle(ui, fh, fname, vfs=None): def readbundle(ui, fh, fname, vfs=None):
header = changegroup.readexactly(fh, 4) header = changegroup.readexactly(fh, 4)
alg = None alg = None
if not fname: if not fname:
fname = b"stream" fname = b"stream"
if not header.startswith(b'HG') and header.startswith(b'\0'): if not header.startswith(b'HG') and header.startswith(b'\0'):
with remote.commandexecutor() as e: with remote.commandexecutor() as e:
res = e.callcommand(b'clonebundles', {}).result() res = e.callcommand(b'clonebundles', {}).result()
# If we call the wire protocol command, that's good enough to record the # If we call the wire protocol command, that's good enough to record the
# attempt. # attempt.
pullop.clonebundleattempted = True pullop.clonebundleattempted = True
entries = parseclonebundlesmanifest(repo, res) entries = bundlecaches.parseclonebundlesmanifest(repo, res)
if not entries: if not entries:
repo.ui.note( repo.ui.note(
_( _(
b'no clone bundles available on remote; ' b'no clone bundles available on remote; '
b'falling back to regular clone\n' b'falling back to regular clone\n'
) )
) )
return return
entries = filterclonebundleentries( entries = bundlecaches.filterclonebundleentries(
repo, entries, streamclonerequested=pullop.streamclonerequested repo, entries, streamclonerequested=pullop.streamclonerequested
) )
if not entries: if not entries:
# There is a thundering herd concern here. However, if a server # There is a thundering herd concern here. However, if a server
# operator doesn't advertise bundles appropriate for its clients, # operator doesn't advertise bundles appropriate for its clients,
# they deserve what's coming. Furthermore, from a client's # they deserve what's coming. Furthermore, from a client's
# perspective, no automatic fallback would mean not being able to # perspective, no automatic fallback would mean not being able to
# clone! # clone!
repo.ui.warn( repo.ui.warn(
_( _(
b'no compatible clone bundles available on server; ' b'no compatible clone bundles available on server; '
b'falling back to regular clone\n' b'falling back to regular clone\n'
) )
) )
repo.ui.warn( repo.ui.warn(
_(b'(you may want to report this to the server operator)\n') _(b'(you may want to report this to the server operator)\n')
) )
return return
entries = sortclonebundleentries(repo.ui, entries) entries = bundlecaches.sortclonebundleentries(repo.ui, entries)
url = entries[0][b'URL'] url = entries[0][b'URL']
repo.ui.status(_(b'applying clone bundle from %s\n') % url) repo.ui.status(_(b'applying clone bundle from %s\n') % url)
if trypullbundlefromurl(repo.ui, repo, url): if trypullbundlefromurl(repo.ui, repo, url):
repo.ui.status(_(b'finished applying clone bundle\n')) repo.ui.status(_(b'finished applying clone bundle\n'))
# Bundle failed. # Bundle failed.
# #
# We abort by default to avoid the thundering herd of # We abort by default to avoid the thundering herd of
# clients flooding a server that was expecting expensive # clients flooding a server that was expecting expensive
# clone load to be offloaded. # clone load to be offloaded.
elif repo.ui.configbool(b'ui', b'clonebundlefallback'): elif repo.ui.configbool(b'ui', b'clonebundlefallback'):
repo.ui.warn(_(b'falling back to normal clone\n')) repo.ui.warn(_(b'falling back to normal clone\n'))
else: else:
raise error.Abort( raise error.Abort(
_(b'error applying bundle'), _(b'error applying bundle'),
hint=_( hint=_(
b'if this error persists, consider contacting ' b'if this error persists, consider contacting '
b'the server operator or disable clone ' b'the server operator or disable clone '
b'bundles via ' b'bundles via '
b'"--config ui.clonebundles=false"' b'"--config ui.clonebundles=false"'
), ),
) )
def parseclonebundlesmanifest(repo, s):
"""Parses the raw text of a clone bundles manifest.
Returns a list of dicts. The dicts have a ``URL`` key corresponding
to the URL and other keys are the attributes for the entry.
"""
m = []
for line in s.splitlines():
fields = line.split()
if not fields:
continue
attrs = {b'URL': fields[0]}
for rawattr in fields[1:]:
key, value = rawattr.split(b'=', 1)
key = urlreq.unquote(key)
value = urlreq.unquote(value)
attrs[key] = value
# Parse BUNDLESPEC into components. This makes client-side
# preferences easier to specify since you can prefer a single
# component of the BUNDLESPEC.
if key == b'BUNDLESPEC':
try:
bundlespec = parsebundlespec(repo, value)
attrs[b'COMPRESSION'] = bundlespec.compression
attrs[b'VERSION'] = bundlespec.version
except error.InvalidBundleSpecification:
pass
except error.UnsupportedBundleSpecification:
pass
m.append(attrs)
return m
def isstreamclonespec(bundlespec):
# Stream clone v1
if bundlespec.wirecompression == b'UN' and bundlespec.wireversion == b's1':
return True
# Stream clone v2
if (
bundlespec.wirecompression == b'UN'
and bundlespec.wireversion == b'02'
and bundlespec.contentopts.get(b'streamv2')
):
return True
return False
def filterclonebundleentries(repo, entries, streamclonerequested=False):
"""Remove incompatible clone bundle manifest entries.
Accepts a list of entries parsed with ``parseclonebundlesmanifest``
and returns a new list consisting of only the entries that this client
should be able to apply.
There is no guarantee we'll be able to apply all returned entries because
the metadata we use to filter on may be missing or wrong.
"""
newentries = []
for entry in entries:
spec = entry.get(b'BUNDLESPEC')
if spec:
try:
bundlespec = parsebundlespec(repo, spec, strict=True)
# If a stream clone was requested, filter out non-streamclone
# entries.
if streamclonerequested and not isstreamclonespec(bundlespec):
repo.ui.debug(
b'filtering %s because not a stream clone\n'
% entry[b'URL']
)
continue
except error.InvalidBundleSpecification as e:
repo.ui.debug(stringutil.forcebytestr(e) + b'\n')
continue
except error.UnsupportedBundleSpecification as e:
repo.ui.debug(
b'filtering %s because unsupported bundle '
b'spec: %s\n' % (entry[b'URL'], stringutil.forcebytestr(e))
)
continue
# If we don't have a spec and requested a stream clone, we don't know
# what the entry is so don't attempt to apply it.
elif streamclonerequested:
repo.ui.debug(
b'filtering %s because cannot determine if a stream '
b'clone bundle\n' % entry[b'URL']
)
continue
if b'REQUIRESNI' in entry and not sslutil.hassni:
repo.ui.debug(
b'filtering %s because SNI not supported\n' % entry[b'URL']
)
continue
if b'REQUIREDRAM' in entry:
try:
requiredram = util.sizetoint(entry[b'REQUIREDRAM'])
except error.ParseError:
repo.ui.debug(
b'filtering %s due to a bad REQUIREDRAM attribute\n'
% entry[b'URL']
)
continue
actualram = repo.ui.estimatememory()
if actualram is not None and actualram * 0.66 < requiredram:
repo.ui.debug(
b'filtering %s as it needs more than 2/3 of system memory\n'
% entry[b'URL']
)
continue
newentries.append(entry)
return newentries
class clonebundleentry(object):
"""Represents an item in a clone bundles manifest.
This rich class is needed to support sorting since sorted() in Python 3
doesn't support ``cmp`` and our comparison is complex enough that ``key=``
won't work.
"""
def __init__(self, value, prefers):
self.value = value
self.prefers = prefers
def _cmp(self, other):
for prefkey, prefvalue in self.prefers:
avalue = self.value.get(prefkey)
bvalue = other.value.get(prefkey)
# Special case for b missing attribute and a matches exactly.
if avalue is not None and bvalue is None and avalue == prefvalue:
return -1
# Special case for a missing attribute and b matches exactly.
if bvalue is not None and avalue is None and bvalue == prefvalue:
return 1
# We can't compare unless attribute present on both.
if avalue is None or bvalue is None:
continue
# Same values should fall back to next attribute.
if avalue == bvalue:
continue
# Exact matches come first.
if avalue == prefvalue:
return -1
if bvalue == prefvalue:
return 1
# Fall back to next attribute.
continue
# If we got here we couldn't sort by attributes and prefers. Fall
# back to index order.
return 0
def __lt__(self, other):
return self._cmp(other) < 0
def __gt__(self, other):
return self._cmp(other) > 0
def __eq__(self, other):
return self._cmp(other) == 0
def __le__(self, other):
return self._cmp(other) <= 0
def __ge__(self, other):
return self._cmp(other) >= 0
def __ne__(self, other):
return self._cmp(other) != 0
def sortclonebundleentries(ui, entries):
prefers = ui.configlist(b'ui', b'clonebundleprefers')
if not prefers:
return list(entries)
def _split(p):
if b'=' not in p:
hint = _(b"each comma separated item should be key=value pairs")
raise error.Abort(
_(b"invalid ui.clonebundleprefers item: %s") % p, hint=hint
)
return p.split(b'=', 1)
prefers = [_split(p) for p in prefers]
items = sorted(clonebundleentry(v, prefers) for v in entries)
return [i.value for i in items]
def trypullbundlefromurl(ui, repo, url): def trypullbundlefromurl(ui, repo, url):
"""Attempt to apply a bundle from a URL.""" """Attempt to apply a bundle from a URL."""
with repo.lock(), repo.transaction(b'bundleurl') as tr: with repo.lock(), repo.transaction(b'bundleurl') as tr:
try: try:
fh = urlmod.open(ui, url) fh = urlmod.open(ui, url)
cg = readbundle(ui, fh, b'stream') cg = readbundle(ui, fh, b'stream')
if isinstance(cg, streamclone.streamcloneapplier): if isinstance(cg, streamclone.streamcloneapplier):

mercurial/wireprotov1server.py

from .node import ( from .node import (
hex, hex,
nullid, nullid,
) )
from .pycompat import getattr from .pycompat import getattr
from . import ( from . import (
bundle2, bundle2,
bundlecaches,
changegroup as changegroupmod, changegroup as changegroupmod,
discovery, discovery,
encoding, encoding,
error, error,
exchange, exchange,
pushkey as pushkeymod, pushkey as pushkeymod,
pycompat, pycompat,
streamclone, streamclone,
""" """
def decodehexstring(s): def decodehexstring(s):
return {binascii.unhexlify(h) for h in s.split(b';')} return {binascii.unhexlify(h) for h in s.split(b';')}
manifest = repo.vfs.tryread(b'pullbundles.manifest') manifest = repo.vfs.tryread(b'pullbundles.manifest')
if not manifest: if not manifest:
return None return None
res = exchange.parseclonebundlesmanifest(repo, manifest) res = bundlecaches.parseclonebundlesmanifest(repo, manifest)
res = exchange.filterclonebundleentries(repo, res) res = bundlecaches.filterclonebundleentries(repo, res)
if not res: if not res:
return None return None
cl = repo.unfiltered().changelog cl = repo.unfiltered().changelog
heads_anc = cl.ancestors([cl.rev(rev) for rev in heads], inclusive=True) heads_anc = cl.ancestors([cl.rev(rev) for rev in heads], inclusive=True)
common_anc = cl.ancestors([cl.rev(rev) for rev in common], inclusive=True) common_anc = cl.ancestors([cl.rev(rev) for rev in common], inclusive=True)
compformats = clientcompressionsupport(proto) compformats = clientcompressionsupport(proto)
for entry in res: for entry in res:
comp = entry.get(b'COMPRESSION') comp = entry.get(b'COMPRESSION')

tests/flagprocessorext.py

# coding=UTF-8 # coding=UTF-8
from __future__ import absolute_import from __future__ import absolute_import
import base64 import base64
import zlib import zlib
from mercurial import ( from mercurial import (
bundlecaches,
changegroup, changegroup,
exchange,
extensions, extensions,
revlog, revlog,
util, util,
) )
from mercurial.revlogutils import flagutil from mercurial.revlogutils import flagutil
# Test only: These flags are defined here only in the context of testing the # Test only: These flags are defined here only in the context of testing the
# behavior of the flag processor. The canonical way to add flags is to get in # behavior of the flag processor. The canonical way to add flags is to get in
wrapfunction(changegroup, 'allsupportedversions', allsupportedversions) wrapfunction(changegroup, 'allsupportedversions', allsupportedversions)
# Teach revlog about our test flags # Teach revlog about our test flags
flags = [REVIDX_NOOP, REVIDX_BASE64, REVIDX_GZIP, REVIDX_FAIL] flags = [REVIDX_NOOP, REVIDX_BASE64, REVIDX_GZIP, REVIDX_FAIL]
flagutil.REVIDX_KNOWN_FLAGS |= util.bitsfrom(flags) flagutil.REVIDX_KNOWN_FLAGS |= util.bitsfrom(flags)
revlog.REVIDX_FLAGS_ORDER.extend(flags) revlog.REVIDX_FLAGS_ORDER.extend(flags)
# Teach exchange to use changegroup 3 # Teach exchange to use changegroup 3
for k in exchange._bundlespeccontentopts.keys(): for k in bundlecaches._bundlespeccontentopts.keys():
exchange._bundlespeccontentopts[k][b"cg.version"] = b"03" bundlecaches._bundlespeccontentopts[k][b"cg.version"] = b"03"
# Register flag processors for each extension # Register flag processors for each extension
flagutil.addflagprocessor( flagutil.addflagprocessor(
REVIDX_NOOP, (noopdonothingread, noopdonothing, validatehash,) REVIDX_NOOP, (noopdonothingread, noopdonothing, validatehash,)
) )
flagutil.addflagprocessor( flagutil.addflagprocessor(
REVIDX_BASE64, (b64decode, b64encode, bypass,), REVIDX_BASE64, (b64decode, b64encode, bypass,),
) )
flagutil.addflagprocessor( flagutil.addflagprocessor(
REVIDX_GZIP, (gzipdecompress, gzipcompress, bypass) REVIDX_GZIP, (gzipdecompress, gzipcompress, bypass)
) )