This is an archive of the discontinued Mercurial Phabricator instance.

Differential D2858

wireproto: define and implement responses in framing protocol
ClosedPublic

Authored by indygreg on Mar 14 2018, 12:01 PM.

Details

Reviewers
durin42
Group Reviewers
hg-reviewers
Commits
rHG61393f888dfe: wireproto: define and implement responses in framing protocol
Summary

Previously, we only had client-side frame types defined. This commit
defines and implements basic support for server-side frame types.

We introduce two frame types - one for representing the raw bytes
result of a command and another for representing error results.
The types are quite primitive and behavior will expand over time.
But you have to start somewhere.

Our server reactor gains methods to react to an intent to send a
response. Again, following the "sans I/O" pattern, the reactor
doesn't actually send the data. Instead, it gives the caller a
generator to frames that it can send out over the wire.

Diff Detail

Repository
rHG Mercurial
Lint
Lint Skipped
Unit
Unit Tests Skipped

Event Timeline

indygreg created this revision.Mar 14 2018, 12:01 PM
indygreg updated this revision to Diff 7051.Mar 14 2018, 9:10 PM
indygreg updated this revision to Diff 7143.Mar 19 2018, 7:59 PM
durin42 added a subscriber: durin42.Mar 21 2018, 9:15 PM
durin42 added inline comments.
mercurial/wireprotoframing.py
347–349

I wonder if we'd be better off with an enumerated set of types that are attr.s instances instead of a dict, but let's not worry about it for now...

durin42 accepted this revision.Mar 21 2018, 9:32 PM
This revision is now accepted and ready to land.Mar 21 2018, 9:32 PM
Closed by commit rHG61393f888dfe: wireproto: define and implement responses in framing protocol (authored by indygreg). · Explain WhyMar 21 2018, 9:40 PM
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

Diff 7143

mercurial/help/internals/wireprotocol.txt

Command data continuation. When set, the data for this command Command data continuation. When set, the data for this command
continues into a subsequent frame. continues into a subsequent frame.
0x02 0x02
End of data. When set, command data has been fully sent to the End of data. When set, command data has been fully sent to the
server. The command has been fully issued and no new data for this server. The command has been fully issued and no new data for this
command will be sent. The next frame will belong to a new command. command will be sent. The next frame will belong to a new command.
Bytes Response Data (``0x04``)
------------------------------
This frame contains raw bytes response data to an issued command.
The following flag values are defined for this type:
0x01
Data continuation. When set, an additional frame containing raw
response data will follow.
0x02
End of data. When sent, the response data has been fully sent and
no additional frames for this response will be sent.
The ``0x01`` flag is mutually exclusive with the ``0x02`` flag.
Error Response (``0x05``)
-------------------------
An error occurred when processing a request. This could indicate
a protocol-level failure or an application level failure depending
on the flags for this message type.
The payload for this type is an error message that should be
displayed to the user.
The following flag values are defined for this type:
0x01
The error occurred at the transport/protocol level. If set, the
connection should be closed.
0x02
The error occurred at the application level. e.g. invalid command.
Issuing Commands Issuing Commands
---------------- ----------------
A client can request that a remote run a command by sending it A client can request that a remote run a command by sending it
frames defining that command. This logical stream is composed of frames defining that command. This logical stream is composed of
1 ``Command Request`` frame, 0 or more ``Command Argument`` frames, 1 ``Command Request`` frame, 0 or more ``Command Argument`` frames,
and 0 or more ``Command Data`` frames. and 0 or more ``Command Data`` frames.

mercurial/wireprotoframing.py

) )
FRAME_HEADER_SIZE = 4 FRAME_HEADER_SIZE = 4
DEFAULT_MAX_FRAME_SIZE = 32768 DEFAULT_MAX_FRAME_SIZE = 32768
FRAME_TYPE_COMMAND_NAME = 0x01 FRAME_TYPE_COMMAND_NAME = 0x01
FRAME_TYPE_COMMAND_ARGUMENT = 0x02 FRAME_TYPE_COMMAND_ARGUMENT = 0x02
FRAME_TYPE_COMMAND_DATA = 0x03 FRAME_TYPE_COMMAND_DATA = 0x03
FRAME_TYPE_BYTES_RESPONSE = 0x04
FRAME_TYPE_ERROR_RESPONSE = 0x05
FRAME_TYPES = { FRAME_TYPES = {
b'command-name': FRAME_TYPE_COMMAND_NAME, b'command-name': FRAME_TYPE_COMMAND_NAME,
b'command-argument': FRAME_TYPE_COMMAND_ARGUMENT, b'command-argument': FRAME_TYPE_COMMAND_ARGUMENT,
b'command-data': FRAME_TYPE_COMMAND_DATA, b'command-data': FRAME_TYPE_COMMAND_DATA,
b'bytes-response': FRAME_TYPE_BYTES_RESPONSE,
b'error-response': FRAME_TYPE_ERROR_RESPONSE,
} }
FLAG_COMMAND_NAME_EOS = 0x01 FLAG_COMMAND_NAME_EOS = 0x01
FLAG_COMMAND_NAME_HAVE_ARGS = 0x02 FLAG_COMMAND_NAME_HAVE_ARGS = 0x02
FLAG_COMMAND_NAME_HAVE_DATA = 0x04 FLAG_COMMAND_NAME_HAVE_DATA = 0x04
FLAGS_COMMAND = { FLAGS_COMMAND = {
b'eos': FLAG_COMMAND_NAME_EOS, b'eos': FLAG_COMMAND_NAME_EOS,
FLAG_COMMAND_DATA_CONTINUATION = 0x01 FLAG_COMMAND_DATA_CONTINUATION = 0x01
FLAG_COMMAND_DATA_EOS = 0x02 FLAG_COMMAND_DATA_EOS = 0x02
FLAGS_COMMAND_DATA = { FLAGS_COMMAND_DATA = {
b'continuation': FLAG_COMMAND_DATA_CONTINUATION, b'continuation': FLAG_COMMAND_DATA_CONTINUATION,
b'eos': FLAG_COMMAND_DATA_EOS, b'eos': FLAG_COMMAND_DATA_EOS,
} }
FLAG_BYTES_RESPONSE_CONTINUATION = 0x01
FLAG_BYTES_RESPONSE_EOS = 0x02
FLAGS_BYTES_RESPONSE = {
b'continuation': FLAG_BYTES_RESPONSE_CONTINUATION,
b'eos': FLAG_BYTES_RESPONSE_EOS,
}
FLAG_ERROR_RESPONSE_PROTOCOL = 0x01
FLAG_ERROR_RESPONSE_APPLICATION = 0x02
FLAGS_ERROR_RESPONSE = {
b'protocol': FLAG_ERROR_RESPONSE_PROTOCOL,
b'application': FLAG_ERROR_RESPONSE_APPLICATION,
}
# Maps frame types to their available flags. # Maps frame types to their available flags.
FRAME_TYPE_FLAGS = { FRAME_TYPE_FLAGS = {
FRAME_TYPE_COMMAND_NAME: FLAGS_COMMAND, FRAME_TYPE_COMMAND_NAME: FLAGS_COMMAND,
FRAME_TYPE_COMMAND_ARGUMENT: FLAGS_COMMAND_ARGUMENT, FRAME_TYPE_COMMAND_ARGUMENT: FLAGS_COMMAND_ARGUMENT,
FRAME_TYPE_COMMAND_DATA: FLAGS_COMMAND_DATA, FRAME_TYPE_COMMAND_DATA: FLAGS_COMMAND_DATA,
FRAME_TYPE_BYTES_RESPONSE: FLAGS_BYTES_RESPONSE,
FRAME_TYPE_ERROR_RESPONSE: FLAGS_ERROR_RESPONSE,
} }
ARGUMENT_FRAME_HEADER = struct.Struct(r'<HH') ARGUMENT_FRAME_HEADER = struct.Struct(r'<HH')
def makeframe(frametype, frameflags, payload): def makeframe(frametype, frameflags, payload):
"""Assemble a frame into a byte array.""" """Assemble a frame into a byte array."""
# TODO assert size of payload. # TODO assert size of payload.
frame = bytearray(FRAME_HEADER_SIZE + len(payload)) frame = bytearray(FRAME_HEADER_SIZE + len(payload))
assert datafh.read(1) == b'' assert datafh.read(1) == b''
done = True done = True
yield makeframe(FRAME_TYPE_COMMAND_DATA, flags, data) yield makeframe(FRAME_TYPE_COMMAND_DATA, flags, data)
if done: if done:
break break
def createbytesresponseframesfrombytes(data,
maxframesize=DEFAULT_MAX_FRAME_SIZE):
"""Create a raw frame to send a bytes response from static bytes input.
Returns a generator of bytearrays.
"""
# Simple case of a single frame.
if len(data) <= maxframesize:
yield makeframe(FRAME_TYPE_BYTES_RESPONSE,
FLAG_BYTES_RESPONSE_EOS, data)
return
offset = 0
while True:
chunk = data[offset:offset + maxframesize]
offset += len(chunk)
done = offset == len(data)
if done:
flags = FLAG_BYTES_RESPONSE_EOS
else:
flags = FLAG_BYTES_RESPONSE_CONTINUATION
yield makeframe(FRAME_TYPE_BYTES_RESPONSE, flags, chunk)
if done:
break
def createerrorframe(msg, protocol=False, application=False):
# TODO properly handle frame size limits.
assert len(msg) <= DEFAULT_MAX_FRAME_SIZE
flags = 0
if protocol:
flags |= FLAG_ERROR_RESPONSE_PROTOCOL
if application:
flags |= FLAG_ERROR_RESPONSE_APPLICATION
yield makeframe(FRAME_TYPE_ERROR_RESPONSE, flags, msg)
class serverreactor(object): class serverreactor(object):
"""Holds state of a server handling frame-based protocol requests. """Holds state of a server handling frame-based protocol requests.
This class is the "brain" of the unified frame-based protocol server This class is the "brain" of the unified frame-based protocol server
component. While the protocol is stateless from the perspective of component. While the protocol is stateless from the perspective of
requests/commands, something needs to track which frames have been requests/commands, something needs to track which frames have been
received, what frames to expect, etc. This class is that thing. received, what frames to expect, etc. This class is that thing.
describing any follow-up action(s) to take. The first element is the describing any follow-up action(s) to take. The first element is the
name of an action to perform. The second is a data structure (usually name of an action to perform. The second is a data structure (usually
a dict) specific to that action that contains more information. e.g. a dict) specific to that action that contains more information. e.g.
if the server wants to send frames back to the client, the data structure if the server wants to send frames back to the client, the data structure
will contain a reference to those frames. will contain a reference to those frames.
Valid actions that consumers can be instructed to take are: Valid actions that consumers can be instructed to take are:
sendframes
Indicates that frames should be sent to the client. The ``framegen``
key contains a generator of frames that should be sent. The server
assumes that all frames are sent to the client.
error error
Indicates that an error occurred. Consumer should probably abort. Indicates that an error occurred. Consumer should probably abort.
runcommand runcommand
Indicates that the consumer should run a wire protocol command. Details Indicates that the consumer should run a wire protocol command. Details
of the command to run are given in the data structure. of the command to run are given in the data structure.
wantframe wantframe
} }
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(frametype, frameflags, payload) return meth(frametype, frameflags, payload)
def onbytesresponseready(self, data):
"""Signal that a bytes response is ready to be sent to the client.
The raw bytes response is passed as an argument.
"""
return 'sendframes', {
'framegen': createbytesresponseframesfrombytes(data),
}
durin42Unsubmitted
Not Done

I wonder if we'd be better off with an enumerated set of types that are attr.s instances instead of a dict, but let's not worry about it for now...

durin42: I wonder if we'd be better off with an enumerated set of types that are attr.s instances…
def onapplicationerror(self, msg):
return 'sendframes', {
'framegen': createerrorframe(msg, application=True),
}
def _makeerrorresult(self, msg): def _makeerrorresult(self, msg):
return 'error', { return 'error', {
'message': msg, 'message': msg,
} }
def _makeruncommandresult(self): def _makeruncommandresult(self):
return 'runcommand', { return 'runcommand', {
'command': self._activecommand, 'command': self._activecommand,

mercurial/wireprotoserver.py

if authedperm == b'ro': if authedperm == b'ro':
assert wirecommand.permission == 'pull' assert wirecommand.permission == 'pull'
elif authedperm == b'rw': elif authedperm == b'rw':
# We are allowed to access read-only commands under the rw URL. # We are allowed to access read-only commands under the rw URL.
assert wirecommand.permission in ('push', 'pull') assert wirecommand.permission in ('push', 'pull')
rsp = wireproto.dispatch(repo, proto, command['command']) rsp = wireproto.dispatch(repo, proto, command['command'])
# TODO use proper response format.
res.status = b'200 OK' res.status = b'200 OK'
res.headers[b'Content-Type'] = b'text/plain' res.headers[b'Content-Type'] = FRAMINGTYPE
if isinstance(rsp, wireprototypes.bytesresponse): if isinstance(rsp, wireprototypes.bytesresponse):
res.setbodybytes(rsp.data) action, meta = reactor.onbytesresponseready(rsp.data)
else:
action, meta = reactor.onapplicationerror(
_('unhandled response type from wire proto command'))
if action == 'sendframes':
res.setbodygen(meta['framegen'])
else: else:
res.setbodybytes(b'unhandled response type from wire proto ' raise error.ProgrammingError('unhandled event from reactor: %s' %
'command') action)
# Maps API name to metadata so custom API can be registered. # Maps API name to metadata so custom API can be registered.
API_HANDLERS = { API_HANDLERS = {
HTTPV2: { HTTPV2: {
'config': ('experimental', 'web.api.http-v2'), 'config': ('experimental', 'web.api.http-v2'),
'handler': _handlehttpv2request, 'handler': _handlehttpv2request,
}, },
} }

tests/test-http-api-httpv2.t

s> content-length: 18\r\n s> content-length: 18\r\n
s> host: $LOCALIP:$HGPORT\r\n (glob) s> host: $LOCALIP:$HGPORT\r\n (glob)
s> \r\n s> \r\n
s> \x0e\x00\x00\x11customreadonly s> \x0e\x00\x00\x11customreadonly
s> makefile('rb', None) s> makefile('rb', None)
s> HTTP/1.1 200 OK\r\n s> HTTP/1.1 200 OK\r\n
s> Server: testing stub value\r\n s> Server: testing stub value\r\n
s> Date: $HTTP_DATE$\r\n s> Date: $HTTP_DATE$\r\n
s> Content-Type: text/plain\r\n s> Content-Type: application/mercurial-exp-framing-0001\r\n
s> Content-Length: 29\r\n s> Transfer-Encoding: chunked\r\n
s> \r\n
s> 21\r\n
s> \x1d\x00\x00Bcustomreadonly bytes response
s> \r\n
s> 0\r\n
s> \r\n s> \r\n
s> customreadonly bytes response
Request to read-write command fails because server is read-only by default Request to read-write command fails because server is read-only by default
GET to read-write request yields 405 GET to read-write request yields 405
$ send << EOF $ send << EOF
> httprequest GET api/$HTTPV2/rw/customreadonly > httprequest GET api/$HTTPV2/rw/customreadonly
> user-agent: test > user-agent: test
s> content-length: 18\r\n s> content-length: 18\r\n
s> host: $LOCALIP:$HGPORT\r\n (glob) s> host: $LOCALIP:$HGPORT\r\n (glob)
s> \r\n s> \r\n
s> \x0e\x00\x00\x11customreadonly s> \x0e\x00\x00\x11customreadonly
s> makefile('rb', None) s> makefile('rb', None)
s> HTTP/1.1 200 OK\r\n s> HTTP/1.1 200 OK\r\n
s> Server: testing stub value\r\n s> Server: testing stub value\r\n
s> Date: $HTTP_DATE$\r\n s> Date: $HTTP_DATE$\r\n
s> Content-Type: text/plain\r\n s> Content-Type: application/mercurial-exp-framing-0001\r\n
s> Content-Length: 29\r\n s> Transfer-Encoding: chunked\r\n
s> \r\n
s> 21\r\n
s> \x1d\x00\x00Bcustomreadonly bytes response
s> \r\n
s> 0\r\n
s> \r\n s> \r\n
s> customreadonly bytes response
Authorized request for unknown command is rejected Authorized request for unknown command is rejected
$ send << EOF $ send << EOF
> httprequest POST api/$HTTPV2/rw/badcommand > httprequest POST api/$HTTPV2/rw/badcommand
> user-agent: test > user-agent: test
> accept: $MEDIATYPE > accept: $MEDIATYPE
> EOF > EOF

tests/test-wireproto-serverreactor.py

return results[0] return results[0]
def assertaction(self, res, expected): def assertaction(self, res, expected):
self.assertIsInstance(res, tuple) self.assertIsInstance(res, tuple)
self.assertEqual(len(res), 2) self.assertEqual(len(res), 2)
self.assertIsInstance(res[1], dict) self.assertIsInstance(res[1], dict)
self.assertEqual(res[0], expected) self.assertEqual(res[0], expected)
def assertframesequal(self, frames, framestrings):
expected = [ffs(s) for s in framestrings]
self.assertEqual(list(frames), expected)
def test1framecommand(self): def test1framecommand(self):
"""Receiving a command in a single frame yields request to run it.""" """Receiving a command in a single frame yields request to run it."""
reactor = makereactor() reactor = makereactor()
results = list(sendcommandframes(reactor, b'mycommand', {})) results = list(sendcommandframes(reactor, b'mycommand', {}))
self.assertEqual(len(results), 1) self.assertEqual(len(results), 1)
self.assertaction(results[0], 'runcommand') self.assertaction(results[0], 'runcommand')
self.assertEqual(results[0][1], { self.assertEqual(results[0][1], {
'command': b'mycommand', 'command': b'mycommand',
results = list(sendframes(makereactor(), frames)) results = list(sendframes(makereactor(), frames))
self.assertEqual(len(results), 2) self.assertEqual(len(results), 2)
self.assertaction(results[0], 'wantframe') self.assertaction(results[0], 'wantframe')
self.assertaction(results[1], 'error') self.assertaction(results[1], 'error')
self.assertEqual(results[1][1], { self.assertEqual(results[1][1], {
'message': b'command data frame without flags', 'message': b'command data frame without flags',
}) })
def testsimpleresponse(self):
"""Bytes response to command sends result frames."""
reactor = makereactor()
list(sendcommandframes(reactor, b'mycommand', {}))
result = reactor.onbytesresponseready(b'response')
self.assertaction(result, 'sendframes')
self.assertframesequal(result[1]['framegen'], [
b'bytes-response eos response',
])
def testmultiframeresponse(self):
"""Bytes response spanning multiple frames is handled."""
first = b'x' * framing.DEFAULT_MAX_FRAME_SIZE
second = b'y' * 100
reactor = makereactor()
list(sendcommandframes(reactor, b'mycommand', {}))
result = reactor.onbytesresponseready(first + second)
self.assertaction(result, 'sendframes')
self.assertframesequal(result[1]['framegen'], [
b'bytes-response continuation %s' % first,
b'bytes-response eos %s' % second,
])
def testapplicationerror(self):
reactor = makereactor()
list(sendcommandframes(reactor, b'mycommand', {}))
result = reactor.onapplicationerror(b'some message')
self.assertaction(result, 'sendframes')
self.assertframesequal(result[1]['framegen'], [
b'error-response application some message',
])
if __name__ == '__main__': if __name__ == '__main__':
import silenttestrunner import silenttestrunner
silenttestrunner.main(__name__) silenttestrunner.main(__name__)