This is an archive of the discontinued Mercurial Phabricator instance.

Differential D1926

exchange: send bundle2 stream clones uncompressed
ClosedPublic

Authored by indygreg on Jan 20 2018, 7:44 PM.

Details

Reviewers
durin42
Group Reviewers
hg-reviewers
Commits
rHGa84dbc87dae9: exchange: send bundle2 stream clones uncompressed
Summary

Stream clones don't compress well. And compression undermines
a point of stream clones which is to trade significant CPU
reductions by increasing size.

Building upon our introduction of metadata to communicate bundle
information back to callers of exchange.getbundlechunks(), we add
an attribute to the bundler that communicates whether the bundle is
best left uncompressed. We return this attribute as part of the bundle
metadata. And the wire protocol honors it when determining whether
to compress the wire protocol response.

The added test demonstrates that the raw result from the wire
protocol is not compressed. It also demonstrates that the server
will serve stream responses when the feature isn't enabled. We'll
address that in another commit.

The effect of this change is that server-side CPU usage for bundle2
stream clones is significantly reduced by removing zstd compression.
For the mozilla-unified repository:

before: 37.69 user 8.01 system
after: 27.38 user 7.34 system

Assuming things are CPU bound, that ~10s reduction would translate to
faster clones on the client. zstd can decompress at >1 GB/s. So the
overhead from decompression on the client is small in the grand scheme
of things. But if zlib compression were being used, the overhead would
be much greater.

Diff Detail

Repository
rHG Mercurial
Lint
Lint Skipped
Unit
Unit Tests Skipped

Event Timeline

indygreg created this revision.Jan 20 2018, 7:44 PM
lothiraldan added a subscriber: lothiraldan.Jan 22 2018, 11:02 AM
lothiraldan added inline comments.
mercurial/bundle2.py
605

The double-negation (preferuncompressed = False) sounds complicated. Some other parts of the code are also using uncompressed but maybe we could use prefercompressed = True here as an improvement.

indygreg updated this revision to Diff 4984.Jan 22 2018, 3:32 PM
durin42 accepted this revision.Jan 22 2018, 4:09 PM
This revision is now accepted and ready to land.Jan 22 2018, 4:09 PM
Closed by commit rHGa84dbc87dae9: exchange: send bundle2 stream clones uncompressed (authored by indygreg). · Explain WhyJan 22 2018, 4:22 PM
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

Diff 4984

mercurial/bundle2.py

def __init__(self, ui, capabilities=()): def __init__(self, ui, capabilities=()):
self.ui = ui self.ui = ui
self._params = [] self._params = []
self._parts = [] self._parts = []
self.capabilities = dict(capabilities) self.capabilities = dict(capabilities)
self._compengine = util.compengines.forbundletype('UN') self._compengine = util.compengines.forbundletype('UN')
self._compopts = None self._compopts = None
# If compression is being handled by a consumer of the raw
# data (e.g. the wire protocol), unsetting this flag tells
# consumers that the bundle is best left uncompressed.
self.prefercompressed = True
lothiraldanUnsubmitted
Not Done

The double-negation (preferuncompressed = False) sounds complicated. Some other parts of the code are also using uncompressed but maybe we could use prefercompressed = True here as an improvement.

lothiraldan: The double-negation (prefer**un**compressed = **False**) sounds complicated. Some other parts…
def setcompression(self, alg, compopts=None): def setcompression(self, alg, compopts=None):
"""setup core part compression to <alg>""" """setup core part compression to <alg>"""
if alg in (None, 'UN'): if alg in (None, 'UN'):
return return
assert not any(n.lower() == 'compression' for n, v in self._params) assert not any(n.lower() == 'compression' for n, v in self._params)
self.addparam('Compression', alg) self.addparam('Compression', alg)
self._compengine = util.compengines.forbundletype(alg) self._compengine = util.compengines.forbundletype(alg)

mercurial/exchange.py

kwargs['heads'] = heads kwargs['heads'] = heads
kwargs['common'] = common kwargs['common'] = common
for name in getbundle2partsorder: for name in getbundle2partsorder:
func = getbundle2partsmapping[name] func = getbundle2partsmapping[name]
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
return info, bundler.getchunks() return info, bundler.getchunks()
@getbundle2partsgenerator('stream') @getbundle2partsgenerator('stream')
def _getbundlestream(bundler, repo, source, bundlecaps=None, def _getbundlestream(bundler, repo, source, bundlecaps=None,
b2caps=None, heads=None, common=None, **kwargs): b2caps=None, heads=None, common=None, **kwargs):
if not kwargs.get('stream', False): if not kwargs.get('stream', False):
return return
# 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) filecount, bytecount, it = streamclone.generatev2(repo)
requirements = ' '.join(sorted(repo.requirements)) requirements = ' '.join(sorted(repo.requirements))
part = bundler.newpart('stream', data=it) part = bundler.newpart('stream', data=it)
part.addparam('bytecount', '%d' % bytecount, mandatory=True) part.addparam('bytecount', '%d' % bytecount, mandatory=True)
part.addparam('filecount', '%d' % filecount, mandatory=True) part.addparam('filecount', '%d' % filecount, mandatory=True)
part.addparam('requirements', requirements, mandatory=True) part.addparam('requirements', requirements, mandatory=True)
part.addparam('version', 'v2', mandatory=True) part.addparam('version', 'v2', mandatory=True)

mercurial/wireproto.py

if not bundle1allowed(repo, 'pull'): if not bundle1allowed(repo, 'pull'):
if not exchange.bundle2requested(opts.get('bundlecaps')): if not exchange.bundle2requested(opts.get('bundlecaps')):
if proto.name == 'http': if proto.name == 'http':
return ooberror(bundle2required) return ooberror(bundle2required)
raise error.Abort(bundle2requiredmain, raise error.Abort(bundle2requiredmain,
hint=bundle2requiredhint) hint=bundle2requiredhint)
preferuncompressed = False prefercompressed = True
try: try:
if repo.ui.configbool('server', 'disablefullbundle'): if repo.ui.configbool('server', 'disablefullbundle'):
# Check to see if this is a full clone. # Check to see if this is a full clone.
clheads = set(repo.changelog.heads()) clheads = set(repo.changelog.heads())
changegroup = opts.get('cg', True) changegroup = opts.get('cg', True)
heads = set(opts.get('heads', set())) heads = set(opts.get('heads', set()))
common = set(opts.get('common', set())) common = set(opts.get('common', set()))
common.discard(nullid) common.discard(nullid)
if changegroup and not common and clheads == heads: if changegroup and not common and clheads == heads:
raise error.Abort( raise error.Abort(
_('server has pull-based clones disabled'), _('server has pull-based clones disabled'),
hint=_('remove --pull if specified or upgrade Mercurial')) hint=_('remove --pull if specified or upgrade Mercurial'))
info, chunks = exchange.getbundlechunks(repo, 'serve', info, chunks = exchange.getbundlechunks(repo, 'serve',
**pycompat.strkwargs(opts)) **pycompat.strkwargs(opts))
prefercompressed = info.get('prefercompressed', True)
except error.Abort as exc: except error.Abort as exc:
# cleanly forward Abort error to the client # cleanly forward Abort error to the client
if not exchange.bundle2requested(opts.get('bundlecaps')): if not exchange.bundle2requested(opts.get('bundlecaps')):
if proto.name == 'http': if proto.name == 'http':
return ooberror(str(exc) + '\n') return ooberror(str(exc) + '\n')
raise # cannot do better for bundle1 + ssh raise # cannot do better for bundle1 + ssh
# bundle2 request expect a bundle2 reply # bundle2 request expect a bundle2 reply
bundler = bundle2.bundle20(repo.ui) bundler = bundle2.bundle20(repo.ui)
manargs = [('message', str(exc))] manargs = [('message', str(exc))]
advargs = [] advargs = []
if exc.hint is not None: if exc.hint is not None:
advargs.append(('hint', exc.hint)) advargs.append(('hint', exc.hint))
bundler.addpart(bundle2.bundlepart('error:abort', bundler.addpart(bundle2.bundlepart('error:abort',
manargs, advargs)) manargs, advargs))
chunks = bundler.getchunks() chunks = bundler.getchunks()
preferuncompressed = True prefercompressed = False
return streamres(gen=chunks, prefer_uncompressed=preferuncompressed) return streamres(gen=chunks, prefer_uncompressed=not prefercompressed)
@wireprotocommand('heads') @wireprotocommand('heads')
def heads(repo, proto): def heads(repo, proto):
h = repo.heads() h = repo.heads()
return encodelist(h) + "\n" return encodelist(h) + "\n"
@wireprotocommand('hello') @wireprotocommand('hello')
def hello(repo, proto): def hello(repo, proto):

tests/test-clone-uncompressed.t

transferred 96.4 KB in * seconds (* */sec) (glob) transferred 96.4 KB in * seconds (* */sec) (glob)
$ ls -1 clone1/.hg/cache $ ls -1 clone1/.hg/cache
branch2-served branch2-served
rbc-names-v1 rbc-names-v1
rbc-revs-v1 rbc-revs-v1
#endif #endif
getbundle requests with stream=1 are uncompressed
$ get-with-headers.py $LOCALIP:$HGPORT '?cmd=getbundle' content-type --bodyfile body --hgproto '0.1 0.2 comp=zlib,none' --requestheader "x-hgarg-1=bundlecaps=HG20%2Cbundle2%3DHG20%250Abookmarks%250Achangegroup%253D01%252C02%250Adigests%253Dmd5%252Csha1%252Csha512%250Aerror%253Dabort%252Cunsupportedcontent%252Cpushraced%252Cpushkey%250Ahgtagsfnodes%250Alistkeys%250Aphases%253Dheads%250Apushkey%250Aremote-changegroup%253Dhttp%252Chttps&cg=0&common=0000000000000000000000000000000000000000&heads=c17445101a72edac06facd130d14808dfbd5c7c2&stream=1"
200 Script output follows
content-type: application/mercurial-0.2
$ f --size --hex --bytes 256 body
body: size=112328
0000: 04 6e 6f 6e 65 48 47 32 30 00 00 00 00 00 00 00 |.noneHG20.......|
0010: 72 06 53 54 52 45 41 4d 00 00 00 00 04 00 09 05 |r.STREAM........|
0020: 09 04 0c 2d 07 02 62 79 74 65 63 6f 75 6e 74 39 |...-..bytecount9|
0030: 38 37 35 38 66 69 6c 65 63 6f 75 6e 74 31 30 33 |8758filecount103|
0040: 30 72 65 71 75 69 72 65 6d 65 6e 74 73 64 6f 74 |0requirementsdot|
0050: 65 6e 63 6f 64 65 20 66 6e 63 61 63 68 65 20 67 |encode fncache g|
0060: 65 6e 65 72 61 6c 64 65 6c 74 61 20 72 65 76 6c |eneraldelta revl|
0070: 6f 67 76 31 20 73 74 6f 72 65 76 65 72 73 69 6f |ogv1 storeversio|
0080: 6e 76 32 00 00 10 00 73 08 42 64 61 74 61 2f 30 |nv2....s.Bdata/0|
0090: 2e 69 00 03 00 01 00 00 00 00 00 00 00 02 00 00 |.i..............|
00a0: 00 01 00 00 00 00 00 00 00 01 ff ff ff ff ff ff |................|
00b0: ff ff 80 29 63 a0 49 d3 23 87 bf ce fe 56 67 92 |...)c.I.#....Vg.|
00c0: 67 2c 69 d1 ec 39 00 00 00 00 00 00 00 00 00 00 |g,i..9..........|
00d0: 00 00 75 30 73 08 42 64 61 74 61 2f 31 2e 69 00 |..u0s.Bdata/1.i.|
00e0: 03 00 01 00 00 00 00 00 00 00 02 00 00 00 01 00 |................|
00f0: 00 00 00 00 00 00 01 ff ff ff ff ff ff ff ff f9 |................|
--uncompressed is an alias to --stream --uncompressed is an alias to --stream
#if stream-legacy #if stream-legacy
$ hg clone --uncompressed -U http://localhost:$HGPORT clone1-uncompressed $ hg clone --uncompressed -U http://localhost:$HGPORT clone1-uncompressed
streaming all changes streaming all changes
1027 files to transfer, 96.3 KB of data 1027 files to transfer, 96.3 KB of data
transferred 96.3 KB in * seconds (*/sec) (glob) transferred 96.3 KB in * seconds (*/sec) (glob)
searching for changes searching for changes