This is an archive of the discontinued Mercurial Phabricator instance.

Differential D1862

wireproto: split streamres into legacy and modern case
ClosedPublic

Authored by joerg.sonnenberger on Jan 15 2018, 6:31 PM.

Details

Reviewers
None
Group Reviewers
hg-reviewers
Commits
rHGa39a9df7ecca: wireproto: split streamres into legacy and modern case
Summary

A couple of commands currently require transmission of uncompressed
frames with the old MIME type. Split this case from streamres into
a new streamres_legacy class. Streamline the remaining code accordingly.

Add a new flag to streamres to request uncompressed streams. This is
useful for sending data that is already compressed like a pre-built
bundle. Expect clients to support uncompressed data. For older clients,
zlib will still be used.

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.

Revision Contents
Changeset List

Diff 4959

hgext/largefiles/proto.py

# Mercurial core provides no way to give the length of a streamres # Mercurial core provides no way to give the length of a streamres
# (and reading the entire file into RAM would be ill-advised), we # (and reading the entire file into RAM would be ill-advised), we
# just send the length on the first line of the response, like the # just send the length on the first line of the response, like the
# ssh proto does for string responses. # ssh proto does for string responses.
def generator(): def generator():
yield '%d\n' % length yield '%d\n' % length
for chunk in util.filechunkiter(f): for chunk in util.filechunkiter(f):
yield chunk yield chunk
return wireproto.streamres(gen=generator()) return wireproto.streamres_legacy(gen=generator())
def statlfile(repo, proto, sha): def statlfile(repo, proto, sha):
'''Server command for checking if a largefile is present - returns '2\n' if '''Server command for checking if a largefile is present - returns '2\n' if
the largefile is missing, '0\n' if it seems to be in good condition. the largefile is missing, '0\n' if it seems to be in good condition.
The value 1 is reserved for mismatched checksum, but that is too expensive The value 1 is reserved for mismatched checksum, but that is too expensive
to be verified on every stat and must be caught be running 'hg verify' to be verified on every stat and must be caught be running 'hg verify'
server side.''' server side.'''

mercurial/hgweb/protocol.py

return val return val
def _client(self): def _client(self):
return 'remote:%s:%s:%s' % ( return 'remote:%s:%s:%s' % (
self.req.env.get('wsgi.url_scheme') or 'http', self.req.env.get('wsgi.url_scheme') or 'http',
urlreq.quote(self.req.env.get('REMOTE_HOST', '')), urlreq.quote(self.req.env.get('REMOTE_HOST', '')),
urlreq.quote(self.req.env.get('REMOTE_USER', ''))) urlreq.quote(self.req.env.get('REMOTE_USER', '')))
def responsetype(self, v1compressible=False): def responsetype(self, prefer_uncompressed):
"""Determine the appropriate response type and compression settings. """Determine the appropriate response type and compression settings.
The ``v1compressible`` argument states whether the response with
application/mercurial-0.1 media types should be zlib compressed.
Returns a tuple of (mediatype, compengine, engineopts). Returns a tuple of (mediatype, compengine, engineopts).
""" """
# For now, if it isn't compressible in the old world, it's never
# compressible. We can change this to send uncompressed 0.2 payloads
# later.
if not v1compressible:
return HGTYPE, None, None
# Determine the response media type and compression engine based # Determine the response media type and compression engine based
# on the request parameters. # on the request parameters.
protocaps = decodevaluefromheaders(self.req, r'X-HgProto').split(' ') protocaps = decodevaluefromheaders(self.req, r'X-HgProto').split(' ')
if '0.2' in protocaps: if '0.2' in protocaps:
# All clients are expected to support uncompressed data.
if prefer_uncompressed:
return HGTYPE2, util._noopengine(), {}
# Default as defined by wire protocol spec. # Default as defined by wire protocol spec.
compformats = ['zlib', 'none'] compformats = ['zlib', 'none']
for cap in protocaps: for cap in protocaps:
if cap.startswith('comp='): if cap.startswith('comp='):
compformats = cap[5:].split(',') compformats = cap[5:].split(',')
break break
# Now find an agreed upon compression format. # Now find an agreed upon compression format.
return HGTYPE, util.compengines['zlib'], opts return HGTYPE, util.compengines['zlib'], opts
def iscmd(cmd): def iscmd(cmd):
return cmd in wireproto.commands return cmd in wireproto.commands
def call(repo, req, cmd): def call(repo, req, cmd):
p = webproto(req, repo.ui) p = webproto(req, repo.ui)
def genversion2(gen, compress, engine, engineopts): def genversion2(gen, engine, engineopts):
# application/mercurial-0.2 always sends a payload header # application/mercurial-0.2 always sends a payload header
# identifying the compression engine. # identifying the compression engine.
name = engine.wireprotosupport().name name = engine.wireprotosupport().name
assert 0 < len(name) < 256 assert 0 < len(name) < 256
yield struct.pack('B', len(name)) yield struct.pack('B', len(name))
yield name yield name
if compress:
for chunk in engine.compressstream(gen, opts=engineopts):
yield chunk
else:
for chunk in gen: for chunk in gen:
yield chunk yield chunk
rsp = wireproto.dispatch(repo, p, cmd) rsp = wireproto.dispatch(repo, p, cmd)
if isinstance(rsp, bytes): if isinstance(rsp, bytes):
req.respond(HTTP_OK, HGTYPE, body=rsp) req.respond(HTTP_OK, HGTYPE, body=rsp)
return [] return []
elif isinstance(rsp, wireproto.streamres_legacy):
gen = rsp.gen
req.respond(HTTP_OK, HGTYPE)
return gen
elif isinstance(rsp, wireproto.streamres): elif isinstance(rsp, wireproto.streamres):
gen = rsp.gen gen = rsp.gen
# This code for compression should not be streamres specific. It # This code for compression should not be streamres specific. It
# is here because we only compress streamres at the moment. # is here because we only compress streamres at the moment.
mediatype, engine, engineopts = p.responsetype(rsp.v1compressible) mediatype, engine, engineopts = p.responsetype(rsp.prefer_uncompressed)
if mediatype == HGTYPE and rsp.v1compressible:
gen = engine.compressstream(gen, engineopts) gen = engine.compressstream(gen, engineopts)
elif mediatype == HGTYPE2:
gen = genversion2(gen, rsp.v1compressible, engine, engineopts) if mediatype == HGTYPE2:
gen = genversion2(gen, engine, engineopts)
req.respond(HTTP_OK, mediatype) req.respond(HTTP_OK, mediatype)
return gen return gen
elif isinstance(rsp, wireproto.pushres): elif isinstance(rsp, wireproto.pushres):
val = p.restore() val = p.restore()
rsp = '%d\n%s' % (rsp.res, val) rsp = '%d\n%s' % (rsp.res, val)
req.respond(HTTP_OK, HGTYPE, body=rsp) req.respond(HTTP_OK, HGTYPE, body=rsp)
return [] return []

mercurial/sshserver.py

finally: finally:
if self.lock is not None: if self.lock is not None:
self.lock.release() self.lock.release()
sys.exit(0) sys.exit(0)
handlers = { handlers = {
str: sendresponse, str: sendresponse,
wireproto.streamres: sendstream, wireproto.streamres: sendstream,
wireproto.streamres_legacy: sendstream,
wireproto.pushres: sendpushresponse, wireproto.pushres: sendpushresponse,
wireproto.pusherr: sendpusherror, wireproto.pusherr: sendpusherror,
wireproto.ooberror: sendooberror, wireproto.ooberror: sendooberror,
} }
def serve_one(self): def serve_one(self):
cmd = self.fin.readline()[:-1] cmd = self.fin.readline()[:-1]
if cmd and cmd in wireproto.commands: if cmd and cmd in wireproto.commands:

mercurial/wireproto.py

# wire protocol command can either return a string or one of these classes. # wire protocol command can either return a string or one of these classes.
class streamres(object): class streamres(object):
"""wireproto reply: binary stream """wireproto reply: binary stream
The call was successful and the result is a stream. The call was successful and the result is a stream.
Accepts a generator containing chunks of data to be sent to the client. Accepts a generator containing chunks of data to be sent to the client.
``v1compressible`` indicates whether this data can be compressed to ``prefer_uncompressed`` indicates that the data is expected to be
"version 1" clients (technically: HTTP peers using uncompressable and that the stream should therefore use the ``none``
application/mercurial-0.1 media type). This flag should NOT be used on engine.
new commands because new clients should support a more modern compression
mechanism.
""" """
def __init__(self, gen=None, v1compressible=False): def __init__(self, gen=None, prefer_uncompressed=False):
self.gen = gen
self.prefer_uncompressed = prefer_uncompressed
class streamres_legacy(object):
"""wireproto reply: uncompressed binary stream
The call was successful and the result is a stream.
Accepts a generator containing chunks of data to be sent to the client.
Like ``streamres``, but sends an uncompressed data for "version 1" clients
using the application/mercurial-0.1 media type.
"""
def __init__(self, gen=None):
self.gen = gen self.gen = gen
self.v1compressible = v1compressible
class pushres(object): class pushres(object):
"""wireproto reply: success with simple integer return """wireproto reply: success with simple integer return
The call was successful and returned an integer contained in `self.res`. The call was successful and returned an integer contained in `self.res`.
""" """
def __init__(self, res): def __init__(self, res):
self.res = res self.res = res
@wireprotocommand('changegroup', 'roots') @wireprotocommand('changegroup', 'roots')
def changegroup(repo, proto, roots): def changegroup(repo, proto, roots):
nodes = decodelist(roots) nodes = decodelist(roots)
outgoing = discovery.outgoing(repo, missingroots=nodes, outgoing = discovery.outgoing(repo, missingroots=nodes,
missingheads=repo.heads()) missingheads=repo.heads())
cg = changegroupmod.makechangegroup(repo, outgoing, '01', 'serve') cg = changegroupmod.makechangegroup(repo, outgoing, '01', 'serve')
gen = iter(lambda: cg.read(32768), '') gen = iter(lambda: cg.read(32768), '')
return streamres(gen=gen, v1compressible=True) return streamres(gen=gen)
@wireprotocommand('changegroupsubset', 'bases heads') @wireprotocommand('changegroupsubset', 'bases heads')
def changegroupsubset(repo, proto, bases, heads): def changegroupsubset(repo, proto, bases, heads):
bases = decodelist(bases) bases = decodelist(bases)
heads = decodelist(heads) heads = decodelist(heads)
outgoing = discovery.outgoing(repo, missingroots=bases, outgoing = discovery.outgoing(repo, missingroots=bases,
missingheads=heads) missingheads=heads)
cg = changegroupmod.makechangegroup(repo, outgoing, '01', 'serve') cg = changegroupmod.makechangegroup(repo, outgoing, '01', 'serve')
gen = iter(lambda: cg.read(32768), '') gen = iter(lambda: cg.read(32768), '')
return streamres(gen=gen, v1compressible=True) return streamres(gen=gen)
@wireprotocommand('debugwireargs', 'one two *') @wireprotocommand('debugwireargs', 'one two *')
def debugwireargs(repo, proto, one, two, others): def debugwireargs(repo, proto, one, two, others):
# only accept optional args from the known set # only accept optional args from the known set
opts = options('debugwireargs', ['three', 'four'], others) opts = options('debugwireargs', ['three', 'four'], others)
return repo.debugwireargs(one, two, **pycompat.strkwargs(opts)) return repo.debugwireargs(one, two, **pycompat.strkwargs(opts))
@wireprotocommand('getbundle', '*') @wireprotocommand('getbundle', '*')
# 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))
return streamres(gen=bundler.getchunks(), v1compressible=True) return streamres(gen=bundler.getchunks())
return streamres(gen=chunks, v1compressible=True) return streamres(gen=chunks)
@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):
return '%s\n' % int(r) return '%s\n' % int(r)
@wireprotocommand('stream_out') @wireprotocommand('stream_out')
def stream(repo, proto): def stream(repo, proto):
'''If the server supports streaming clone, it advertises the "stream" '''If the server supports streaming clone, it advertises the "stream"
capability with a value representing the version and flags of the repo capability with a value representing the version and flags of the repo
it is serving. Client checks to see if it understands the format. it is serving. Client checks to see if it understands the format.
''' '''
return streamres(streamclone.generatev1wireproto(repo)) return streamres_legacy(streamclone.generatev1wireproto(repo))
@wireprotocommand('unbundle', 'heads') @wireprotocommand('unbundle', 'heads')
def unbundle(repo, proto, heads): def unbundle(repo, proto, heads):
their_heads = decodelist(heads) their_heads = decodelist(heads)
try: try:
proto.redirect() proto.redirect()
raise error.Abort(bundle2requiredmain, raise error.Abort(bundle2requiredmain,
hint=bundle2requiredhint) hint=bundle2requiredhint)
r = exchange.unbundle(repo, gen, their_heads, 'serve', r = exchange.unbundle(repo, gen, their_heads, 'serve',
proto._client()) proto._client())
if util.safehasattr(r, 'addpart'): if util.safehasattr(r, 'addpart'):
# The return looks streamable, we are in the bundle2 case and # The return looks streamable, we are in the bundle2 case and
# should return a stream. # should return a stream.
return streamres(gen=r.getchunks()) return streamres_legacy(gen=r.getchunks())
return pushres(r) return pushres(r)
finally: finally:
fp.close() fp.close()
os.unlink(tempname) os.unlink(tempname)
except (error.BundleValueError, error.Abort, error.PushRaced) as exc: except (error.BundleValueError, error.Abort, error.PushRaced) as exc:
# handle non-bundle2 case first # handle non-bundle2 case first
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))
except error.PushRaced as exc: except error.PushRaced as exc:
bundler.newpart('error:pushraced', [('message', str(exc))]) bundler.newpart('error:pushraced', [('message', str(exc))])
return streamres(gen=bundler.getchunks()) return streamres_legacy(gen=bundler.getchunks())