This is an archive of the discontinued Mercurial Phabricator instance.

Differential D3179

wireproto: port heads command to wire protocol v2
ClosedPublic

Authored by indygreg on Apr 7 2018, 1:53 AM.

Details

Reviewers
yuja
Group Reviewers
hg-reviewers
Commits
rHG0b7475ea38cf: wireproto: port heads command to wire protocol v2
Summary

After much thought and consideration, wire protocol version 2's
commands will be defined in different functions from the existing
commands. This will make it easier to implement these commands
because it won't require shoehorning things like response formatting
and argument declaration into the same APIs.

For example, wire protocol version 1 requires that commands declare
a fixed and ordered list of argument names. It isn't really possible
to insert new arguments or have optional arguments without
breaking backwards compatibility. Wire protocol version 2, however,
uses CBOR maps for passing arguments. So arguments a) can be
optional b) can be added without BC c) can be strongly typed.

This commit starts our trek towards reimplementing the wire protocol
for version 2 with the heads command. It is pretty similar to the
existing heads command. One added feature is it can be told to
operate on only public phase changesets. This is useful for
making discovery faster when a repo has tens of thousands of
draft phase heads (such as Mozilla's "try" repository).

The HTTPv2 server-side protocol has had its getargs() implementation
updated to reflect that arguments are a map and not a list.

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

indygreg created this revision.Apr 7 2018, 1:53 AM
yuja accepted this revision.Apr 9 2018, 10:58 AM
This revision is now accepted and ready to land.Apr 9 2018, 10:58 AM
yuja added inline comments.Apr 9 2018, 11:07 AM
mercurial/wireproto.py
526

My two cents. I think it's better to pass args dict without expansion.
**args means we can't use repo and proto as parameter names.

1209

Maybe we'll need a separate v2 command module?

Closed by commit rHG0b7475ea38cf: wireproto: port heads command to wire protocol v2 (authored by indygreg). · Explain WhyApr 9 2018, 11:13 AM
This revision was automatically updated to reflect the committed changes.
indygreg added inline comments.Apr 9 2018, 2:00 PM
mercurial/wireproto.py
526

I agree about not using `**args. It is there mostly for compatibility with existing command handlers. Once we have a distinct set of handlers, we can ditch **args`.

1209

I also think it makes sense to separate things into separate modules. However, there are a number of cycles involved. I'll try to do this. Not sure if I'll make it before 4.6 freeze though.

Revision Contents
Changeset List

Diff 7896

mercurial/help/internals/wireprotocol.txt

hooks, etc). hooks, etc).
In some cases, the server may respond with a ``bundle2`` bundle. In this In some cases, the server may respond with a ``bundle2`` bundle. In this
case, the response type is ``stream``. For the HTTP version 1 transport, the case, the response type is ``stream``. For the HTTP version 1 transport, the
response is zlib compressed. response is zlib compressed.
The server may also respond with a generic error type, which contains a string The server may also respond with a generic error type, which contains a string
indicating the failure. indicating the failure.
Frame-Based Protocol Commands
=============================
**Experimental and under active development**
This section documents the wire protocol commands exposed to transports
using the frame-based protocol. The set of commands exposed through
these transports is distinct from the set of commands exposed to legacy
transports.
The frame-based protocol uses CBOR to encode command execution requests.
All command arguments must be mapped to a specific or set of CBOR data
types.
The response to many commands is also CBOR. There is no common response
format: each command defines its own response format.
TODO require node type be specified, as N bytes of binary node value
could be ambiguous once SHA-1 is replaced.
heads
-----
Obtain DAG heads in the repository.
The command accepts the following arguments:
publiconly (optional)
(boolean) If set, operate on the DAG for public phase changesets only.
Non-public (i.e. draft) phase DAG heads will not be returned.
The response is a CBOR array of bytestrings defining changeset nodes
of DAG heads. The array can be empty if the repository is empty or no
changesets satisfied the request.
TODO consider exposing phase of heads in response

mercurial/wireproto.py

def dispatch(repo, proto, command): def dispatch(repo, proto, command):
repo = getdispatchrepo(repo, proto, command) repo = getdispatchrepo(repo, proto, command)
transportversion = wireprototypes.TRANSPORTS[proto.name]['version'] transportversion = wireprototypes.TRANSPORTS[proto.name]['version']
commandtable = commandsv2 if transportversion == 2 else commands commandtable = commandsv2 if transportversion == 2 else commands
func, spec = commandtable[command] func, spec = commandtable[command]
args = proto.getargs(spec) args = proto.getargs(spec)
# Version 1 protocols define arguments as a list. Version 2 uses a dict.
if isinstance(args, list):
return func(repo, proto, *args) return func(repo, proto, *args)
elif isinstance(args, dict):
return func(repo, proto, **args)
yujaUnsubmitted
Not Done

My two cents. I think it's better to pass args dict without expansion.
**args means we can't use repo and proto as parameter names.

yuja: My two cents. I think it's better to pass `args` dict without expansion. `**args` means we…
indygregAuthorUnsubmitted
Not Done

I agree about not using `**args. It is there mostly for compatibility with existing command handlers. Once we have a distinct set of handlers, we can ditch **args`.

indygreg: I agree about not using ``**args``. It is there mostly for compatibility with existing command…
else:
raise error.ProgrammingError('unexpected type returned from '
'proto.getargs(): %s' % type(args))
def options(cmd, keys, others): def options(cmd, keys, others):
opts = {} opts = {}
for k in keys: for k in keys:
if k in others: if k in others:
opts[k] = others[k] opts[k] = others[k]
del others[k] del others[k]
if others: if others:
bundler.addpart(bundle2.bundlepart('error:abort', bundler.addpart(bundle2.bundlepart('error:abort',
manargs, advargs)) manargs, advargs))
chunks = bundler.getchunks() chunks = bundler.getchunks()
prefercompressed = False prefercompressed = False
return wireprototypes.streamres( return wireprototypes.streamres(
gen=chunks, prefer_uncompressed=not prefercompressed) gen=chunks, prefer_uncompressed=not prefercompressed)
@wireprotocommand('heads', permission='pull') @wireprotocommand('heads', permission='pull', transportpolicy=POLICY_V1_ONLY)
def heads(repo, proto): def heads(repo, proto):
h = repo.heads() h = repo.heads()
return wireprototypes.bytesresponse(encodelist(h) + '\n') return wireprototypes.bytesresponse(encodelist(h) + '\n')
@wireprotocommand('hello', permission='pull') @wireprotocommand('hello', permission='pull')
def hello(repo, proto): def hello(repo, proto):
"""Called as part of SSH handshake to obtain server info. """Called as part of SSH handshake to obtain server info.
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', bundler.newpart('error:pushraced',
[('message', stringutil.forcebytestr(exc))]) [('message', stringutil.forcebytestr(exc))])
return wireprototypes.streamreslegacy(gen=bundler.getchunks()) return wireprototypes.streamreslegacy(gen=bundler.getchunks())
# Wire protocol version 2 commands only past this point.
yujaUnsubmitted
Not Done

Maybe we'll need a separate v2 command module?

yuja: Maybe we'll need a separate v2 command module?
indygregAuthorUnsubmitted
Not Done

I also think it makes sense to separate things into separate modules. However, there are a number of cycles involved. I'll try to do this. Not sure if I'll make it before 4.6 freeze though.

indygreg: I also think it makes sense to separate things into separate modules. However, there are a…
@wireprotocommand('heads', args='publiconly', permission='pull',
transportpolicy=POLICY_V2_ONLY)
def headsv2(repo, proto, publiconly=False):
if publiconly:
repo = repo.filtered('immutable')
return wireprototypes.cborresponse(repo.heads())

mercurial/wireprotoframing.py

yield stream.makeframe(requestid=requestid, yield stream.makeframe(requestid=requestid,
typeid=FRAME_TYPE_COMMAND_DATA, typeid=FRAME_TYPE_COMMAND_DATA,
flags=flags, flags=flags,
payload=data) payload=data)
if done: if done:
break break
def createbytesresponseframesfrombytes(stream, requestid, data, def createbytesresponseframesfrombytes(stream, requestid, data, iscbor=False,
maxframesize=DEFAULT_MAX_FRAME_SIZE): maxframesize=DEFAULT_MAX_FRAME_SIZE):
"""Create a raw frame to send a bytes response from static bytes input. """Create a raw frame to send a bytes response from static bytes input.
Returns a generator of bytearrays. Returns a generator of bytearrays.
""" """
# Simple case of a single frame. # Simple case of a single frame.
if len(data) <= maxframesize: if len(data) <= maxframesize:
flags = FLAG_BYTES_RESPONSE_EOS
if iscbor:
flags |= FLAG_BYTES_RESPONSE_CBOR
yield stream.makeframe(requestid=requestid, yield stream.makeframe(requestid=requestid,
typeid=FRAME_TYPE_BYTES_RESPONSE, typeid=FRAME_TYPE_BYTES_RESPONSE,
flags=FLAG_BYTES_RESPONSE_EOS, flags=flags,
payload=data) payload=data)
return return
offset = 0 offset = 0
while True: while True:
chunk = data[offset:offset + maxframesize] chunk = data[offset:offset + maxframesize]
offset += len(chunk) offset += len(chunk)
done = offset == len(data) done = offset == len(data)
if done: if done:
flags = FLAG_BYTES_RESPONSE_EOS flags = FLAG_BYTES_RESPONSE_EOS
else: else:
flags = FLAG_BYTES_RESPONSE_CONTINUATION flags = FLAG_BYTES_RESPONSE_CONTINUATION
if iscbor:
flags |= FLAG_BYTES_RESPONSE_CBOR
yield stream.makeframe(requestid=requestid, yield stream.makeframe(requestid=requestid,
typeid=FRAME_TYPE_BYTES_RESPONSE, typeid=FRAME_TYPE_BYTES_RESPONSE,
flags=flags, flags=flags,
payload=chunk) payload=chunk)
if done: if done:
break break
} }
meth = handlers.get(self._state) meth = handlers.get(self._state)
if not meth: if not meth:
raise error.ProgrammingError('unhandled state: %s' % self._state) raise error.ProgrammingError('unhandled state: %s' % self._state)
return meth(frame) return meth(frame)
def onbytesresponseready(self, stream, requestid, data): def onbytesresponseready(self, stream, requestid, data, iscbor=False):
"""Signal that a bytes response is ready to be sent to the client. """Signal that a bytes response is ready to be sent to the client.
The raw bytes response is passed as an argument. The raw bytes response is passed as an argument.
""" """
ensureserverstream(stream) ensureserverstream(stream)
def sendframes(): def sendframes():
for frame in createbytesresponseframesfrombytes(stream, requestid, for frame in createbytesresponseframesfrombytes(stream, requestid,
data): data,
iscbor=iscbor):
yield frame yield frame
self._activecommands.remove(requestid) self._activecommands.remove(requestid)
result = sendframes() result = sendframes()
if self._deferoutput: if self._deferoutput:
self._bufferedframegens.append(result) self._bufferedframegens.append(result)

mercurial/wireprotoserver.py

# Copyright 21 May 2005 - (c) 2005 Jake Edge <jake@edge2.net> # Copyright 21 May 2005 - (c) 2005 Jake Edge <jake@edge2.net>
# Copyright 2005-2007 Matt Mackall <mpm@selenic.com> # Copyright 2005-2007 Matt Mackall <mpm@selenic.com>
# #
# This software may be used and distributed according to the terms of the # This software may be used and distributed according to the terms of the
# GNU General Public License version 2 or any later version. # GNU General Public License version 2 or any later version.
from __future__ import absolute_import from __future__ import absolute_import
import contextlib import contextlib
import struct import struct
import sys import sys
import threading import threading
from .i18n import _ from .i18n import _
from .thirdparty import (
cbor,
)
from .thirdparty.zope import ( from .thirdparty.zope import (
interface as zi, interface as zi,
) )
from . import ( from . import (
encoding, encoding,
error, error,
hook, hook,
pycompat, pycompat,
res.status = b'200 OK' res.status = b'200 OK'
res.headers[b'Content-Type'] = FRAMINGTYPE res.headers[b'Content-Type'] = FRAMINGTYPE
if isinstance(rsp, wireprototypes.bytesresponse): if isinstance(rsp, wireprototypes.bytesresponse):
action, meta = reactor.onbytesresponseready(outstream, action, meta = reactor.onbytesresponseready(outstream,
command['requestid'], command['requestid'],
rsp.data) rsp.data)
elif isinstance(rsp, wireprototypes.cborresponse):
encoded = cbor.dumps(rsp.value, canonical=True)
action, meta = reactor.onbytesresponseready(outstream,
command['requestid'],
encoded,
iscbor=True)
else: else:
action, meta = reactor.onapplicationerror( action, meta = reactor.onapplicationerror(
_('unhandled response type from wire proto command')) _('unhandled response type from wire proto command'))
if action == 'sendframes': if action == 'sendframes':
res.setbodygen(meta['framegen']) res.setbodygen(meta['framegen'])
return True return True
elif action == 'noop': elif action == 'noop':
def name(self): def name(self):
return HTTPV2 return HTTPV2
def getargs(self, args): def getargs(self, args):
data = {} data = {}
for k in args.split(): for k in args.split():
if k == '*': if k == '*':
raise NotImplementedError('do not support * args') raise NotImplementedError('do not support * args')
else: elif k in self._args:
data[k] = self._args[k] data[k] = self._args[k]
return [data[k] for k in args.split()] return data
def getprotocaps(self): def getprotocaps(self):
# Protocol capabilities are currently not implemented for HTTP V2. # Protocol capabilities are currently not implemented for HTTP V2.
return set() return set()
def getpayload(self): def getpayload(self):
raise NotImplementedError raise NotImplementedError

mercurial/wireprototypes.py

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.
Like ``streamres``, but sends an uncompressed data for "version 1" clients Like ``streamres``, but sends an uncompressed data for "version 1" clients
using the application/mercurial-0.1 media type. using the application/mercurial-0.1 media type.
""" """
def __init__(self, gen=None): def __init__(self, gen=None):
self.gen = gen self.gen = gen
class cborresponse(object):
"""Encode the response value as CBOR."""
def __init__(self, v):
self.value = v
class baseprotocolhandler(zi.Interface): class baseprotocolhandler(zi.Interface):
"""Abstract base class for wire protocol handlers. """Abstract base class for wire protocol handlers.
A wire protocol handler serves as an interface between protocol command A wire protocol handler serves as an interface between protocol command
handlers and the wire protocol transport layer. Protocol handlers provide handlers and the wire protocol transport layer. Protocol handlers provide
methods to read command arguments, redirect stdio for the duration of methods to read command arguments, redirect stdio for the duration of
the request, handle response types, etc. the request, handle response types, etc.
""" """
name = zi.Attribute( name = zi.Attribute(
"""The name of the protocol implementation. """The name of the protocol implementation.
Used for uniquely identifying the transport type. Used for uniquely identifying the transport type.
""") """)
def getargs(args): def getargs(args):
"""return the value for arguments in <args> """return the value for arguments in <args>
returns a list of values (same order as <args>)""" For version 1 transports, returns a list of values in the same
order they appear in ``args``. For version 2 transports, returns
a dict mapping argument name to value.
"""
def getprotocaps(): def getprotocaps():
"""Returns the list of protocol-level capabilities of client """Returns the list of protocol-level capabilities of client
Returns a list of capabilities as declared by the client for Returns a list of capabilities as declared by the client for
the current request (or connection for stateful protocol handlers).""" the current request (or connection for stateful protocol handlers)."""
def getpayload(): def getpayload():

tests/test-wireproto-command-heads.t

  • This file was added.
$ . $TESTDIR/wireprotohelpers.sh
$ hg init server
$ enablehttpv2 server
$ cd server
$ hg debugdrawdag << EOF
> H I J
> | | |
> E F G
> | |/
> C D
> |/
> B
> |
> A
> EOF
$ hg phase --force --secret J
$ hg phase --public E
$ hg log -r 'E + H + I + G + J' -T '{rev}:{node} {desc} {phase}\n'
4:78d2dca436b2f5b188ac267e29b81e07266d38fc E public
7:ae492e36b0c8339ffaf328d00b85b4525de1165e H draft
8:1d6f6b91d44aaba6d5e580bc30a9948530dbe00b I draft
6:29446d2dc5419c5f97447a8bc062e4cc328bf241 G draft
9:dec04b246d7cbb670c6689806c05ad17c835284e J secret
$ hg serve -p $HGPORT -d --pid-file hg.pid -E error.log
$ cat hg.pid > $DAEMON_PIDS
All non-secret heads returned by default
$ sendhttpv2peer << EOF
> command heads
> EOF
creating http peer for wire protocol version 2
sending heads command
s> POST /api/exp-http-v2-0001/ro/heads HTTP/1.1\r\n
s> Accept-Encoding: identity\r\n
s> accept: application/mercurial-exp-framing-0003\r\n
s> content-type: application/mercurial-exp-framing-0003\r\n
s> content-length: 20\r\n
s> host: $LOCALIP:$HGPORT\r\n (glob)
s> user-agent: Mercurial debugwireproto\r\n
s> \r\n
s> \x0c\x00\x00\x01\x00\x01\x01\x11\xa1DnameEheads
s> makefile('rb', None)
s> HTTP/1.1 200 OK\r\n
s> Server: testing stub value\r\n
s> Date: $HTTP_DATE$\r\n
s> Content-Type: application/mercurial-exp-framing-0003\r\n
s> Transfer-Encoding: chunked\r\n
s> \r\n
s> 48\r\n
s> @\x00\x00\x01\x00\x02\x01F
s> \x83T\x1dok\x91\xd4J\xab\xa6\xd5\xe5\x80\xbc0\xa9\x94\x850\xdb\xe0\x0bT\xaeI.6\xb0\xc83\x9f\xfa\xf3(\xd0\x0b\x85\xb4R]\xe1\x16^T)Dm-\xc5A\x9c_\x97Dz\x8b\xc0b\xe4\xcc2\x8b\xf2A
s> \r\n
received frame(size=64; request=1; stream=2; streamflags=stream-begin; type=bytes-response; flags=eos|cbor)
s> 0\r\n
s> \r\n
response: [[b'\x1dok\x91\xd4J\xab\xa6\xd5\xe5\x80\xbc0\xa9\x94\x850\xdb\xe0\x0b', b'\xaeI.6\xb0\xc83\x9f\xfa\xf3(\xd0\x0b\x85\xb4R]\xe1\x16^', b')Dm-\xc5A\x9c_\x97Dz\x8b\xc0b\xe4\xcc2\x8b\xf2A']]
Requesting just the public heads works
$ sendhttpv2peer << EOF
> command heads
> publiconly 1
> EOF
creating http peer for wire protocol version 2
sending heads command
s> POST /api/exp-http-v2-0001/ro/heads HTTP/1.1\r\n
s> Accept-Encoding: identity\r\n
s> accept: application/mercurial-exp-framing-0003\r\n
s> content-type: application/mercurial-exp-framing-0003\r\n
s> content-length: 39\r\n
s> host: $LOCALIP:$HGPORT\r\n (glob)
s> user-agent: Mercurial debugwireproto\r\n
s> \r\n
s> \x1f\x00\x00\x01\x00\x01\x01\x11\xa2Dargs\xa1JpubliconlyA1DnameEheads
s> makefile('rb', None)
s> HTTP/1.1 200 OK\r\n
s> Server: testing stub value\r\n
s> Date: $HTTP_DATE$\r\n
s> Content-Type: application/mercurial-exp-framing-0003\r\n
s> Transfer-Encoding: chunked\r\n
s> \r\n
s> 1e\r\n
s> \x16\x00\x00\x01\x00\x02\x01F
s> \x81Tx\xd2\xdc\xa46\xb2\xf5\xb1\x88\xac&~)\xb8\x1e\x07&m8\xfc
s> \r\n
received frame(size=22; request=1; stream=2; streamflags=stream-begin; type=bytes-response; flags=eos|cbor)
s> 0\r\n
s> \r\n
response: [[b'x\xd2\xdc\xa46\xb2\xf5\xb1\x88\xac&~)\xb8\x1e\x07&m8\xfc']]
$ cat error.log