This is an archive of the discontinued Mercurial Phabricator instance.

Differential D4916

wireprotov2: handle sender protocol settings frames
ClosedPublic

Authored by indygreg on Oct 8 2018, 8:29 PM.

Details

Reviewers
None
Group Reviewers
hg-reviewers
Commits
rHG327d40b94bed: wireprotov2: handle sender protocol settings frames
Summary

We teach the server reactor to handle the optional sender protocol
settings frames, which can only be sent at the beginning of frame
exchange.

Right now, we simply decode the data and record the sender protocol
settings on the server reactor instance: we don't yet do anything
meaningful with the data.

Diff Detail

Repository
rHG Mercurial
Lint
Lint Skipped
Unit
Unit Tests Skipped

Event Timeline

indygreg created this revision.Oct 8 2018, 8:29 PM
Closed by commit rHG327d40b94bed: wireprotov2: handle sender protocol settings frames (authored by indygreg). · Explain WhyOct 10 2018, 10:57 AM
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

Diff 11755

mercurial/wireprotoframing.py

payload) payload)
def ensureserverstream(stream): def ensureserverstream(stream):
if stream.streamid % 2: if stream.streamid % 2:
raise error.ProgrammingError('server should only write to even ' raise error.ProgrammingError('server should only write to even '
'numbered streams; %d is not even' % 'numbered streams; %d is not even' %
stream.streamid) stream.streamid)
DEFAULT_PROTOCOL_SETTINGS = {
'contentencodings': [b'identity'],
}
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.
``deferoutput`` can be used to indicate that no output frames should be ``deferoutput`` can be used to indicate that no output frames should be
instructed to be sent until input has been exhausted. In this mode, instructed to be sent until input has been exhausted. In this mode,
events that would normally generate output frames (such as a command events that would normally generate output frames (such as a command
response being ready) will instead defer instructing the consumer to response being ready) will instead defer instructing the consumer to
send those frames. This is useful for half-duplex transports where the send those frames. This is useful for half-duplex transports where the
sender cannot receive until all data has been transmitted. sender cannot receive until all data has been transmitted.
""" """
self._deferoutput = deferoutput self._deferoutput = deferoutput
self._state = 'idle' self._state = 'initial'
self._nextoutgoingstreamid = 2 self._nextoutgoingstreamid = 2
self._bufferedframegens = [] self._bufferedframegens = []
# stream id -> stream instance for all active streams from the client. # stream id -> stream instance for all active streams from the client.
self._incomingstreams = {} self._incomingstreams = {}
self._outgoingstreams = {} self._outgoingstreams = {}
# request id -> dict of commands that are actively being received. # request id -> dict of commands that are actively being received.
self._receivingcommands = {} self._receivingcommands = {}
# Request IDs that have been received and are actively being processed. # Request IDs that have been received and are actively being processed.
# Once all output for a request has been sent, it is removed from this # Once all output for a request has been sent, it is removed from this
# set. # set.
self._activecommands = set() self._activecommands = set()
self._protocolsettingsdecoder = None
# Sender protocol settings are optional. Set implied default values.
self._sendersettings = dict(DEFAULT_PROTOCOL_SETTINGS)
def onframerecv(self, frame): def onframerecv(self, frame):
"""Process a frame that has been received off the wire. """Process a frame that has been received off the wire.
Returns a dict with an ``action`` key that details what action, Returns a dict with an ``action`` key that details what action,
if any, the consumer should take next. if any, the consumer should take next.
""" """
if not frame.streamid % 2: if not frame.streamid % 2:
self._state = 'errored' self._state = 'errored'
self._state = 'errored' self._state = 'errored'
raise error.ProgrammingError('support for decoding stream payloads ' raise error.ProgrammingError('support for decoding stream payloads '
'not yet implemented') 'not yet implemented')
if frame.streamflags & STREAM_FLAG_END_STREAM: if frame.streamflags & STREAM_FLAG_END_STREAM:
del self._incomingstreams[frame.streamid] del self._incomingstreams[frame.streamid]
handlers = { handlers = {
'initial': self._onframeinitial,
'protocol-settings-receiving': self._onframeprotocolsettings,
'idle': self._onframeidle, 'idle': self._onframeidle,
'command-receiving': self._onframecommandreceiving, 'command-receiving': self._onframecommandreceiving,
'errored': self._onframeerrored, 'errored': self._onframeerrored,
} }
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)
'continuation flags set')) 'continuation flags set'))
if not new and not continuation: if not new and not continuation:
self._state = 'errored' self._state = 'errored'
return self._makeerrorresult( return self._makeerrorresult(
_('received command request frame with neither new nor ' _('received command request frame with neither new nor '
'continuation flags set')) 'continuation flags set'))
def _onframeinitial(self, frame):
# Called when we receive a frame when in the "initial" state.
if frame.typeid == FRAME_TYPE_SENDER_PROTOCOL_SETTINGS:
self._state = 'protocol-settings-receiving'
self._protocolsettingsdecoder = cborutil.bufferingdecoder()
return self._onframeprotocolsettings(frame)
elif frame.typeid == FRAME_TYPE_COMMAND_REQUEST:
self._state = 'idle'
return self._onframeidle(frame)
else:
self._state = 'errored'
return self._makeerrorresult(
_('expected sender protocol settings or command request '
'frame; got %d') % frame.typeid)
def _onframeprotocolsettings(self, frame):
assert self._state == 'protocol-settings-receiving'
assert self._protocolsettingsdecoder is not None
if frame.typeid != FRAME_TYPE_SENDER_PROTOCOL_SETTINGS:
self._state = 'errored'
return self._makeerrorresult(
_('expected sender protocol settings frame; got %d') %
frame.typeid)
more = frame.flags & FLAG_SENDER_PROTOCOL_SETTINGS_CONTINUATION
eos = frame.flags & FLAG_SENDER_PROTOCOL_SETTINGS_EOS
if more and eos:
self._state = 'errored'
return self._makeerrorresult(
_('sender protocol settings frame cannot have both '
'continuation and end of stream flags set'))
if not more and not eos:
self._state = 'errored'
return self._makeerrorresult(
_('sender protocol settings frame must have continuation or '
'end of stream flag set'))
# TODO establish limits for maximum amount of data that can be
# buffered.
try:
self._protocolsettingsdecoder.decode(frame.payload)
except Exception as e:
self._state = 'errored'
return self._makeerrorresult(
_('error decoding CBOR from sender protocol settings frame: %s')
% stringutil.forcebytestr(e))
if more:
return self._makewantframeresult()
assert eos
decoded = self._protocolsettingsdecoder.getavailable()
self._protocolsettingsdecoder = None
if not decoded:
self._state = 'errored'
return self._makeerrorresult(
_('sender protocol settings frame did not contain CBOR data'))
elif len(decoded) > 1:
self._state = 'errored'
return self._makeerrorresult(
_('sender protocol settings frame contained multiple CBOR '
'values'))
d = decoded[0]
if b'contentencodings' in d:
self._sendersettings['contentencodings'] = d[b'contentencodings']
self._state = 'idle'
return self._makewantframeresult()
def _onframeidle(self, frame): def _onframeidle(self, frame):
# The only frame type that should be received in this state is a # The only frame type that should be received in this state is a
# command request. # command request.
if frame.typeid != FRAME_TYPE_COMMAND_REQUEST: if frame.typeid != FRAME_TYPE_COMMAND_REQUEST:
self._state = 'errored' self._state = 'errored'
return self._makeerrorresult( return self._makeerrorresult(
_('expected command request frame; got %d') % frame.typeid) _('expected command request frame; got %d') % frame.typeid)

tests/test-wireproto-serverreactor.py

from __future__ import absolute_import, print_function from __future__ import absolute_import, print_function
import unittest import unittest
from mercurial.thirdparty import ( from mercurial.thirdparty import (
cbor, cbor,
) )
from mercurial import ( from mercurial import (
util, util,
wireprotoframing as framing, wireprotoframing as framing,
) )
from mercurial.utils import (
cborutil,
)
ffs = framing.makeframefromhumanstring ffs = framing.makeframefromhumanstring
OK = cbor.dumps({b'status': b'ok'}) OK = cbor.dumps({b'status': b'ok'})
def makereactor(deferoutput=False): def makereactor(deferoutput=False):
return framing.serverreactor(deferoutput=deferoutput) return framing.serverreactor(deferoutput=deferoutput)
}) })
def testunexpectedcommanddata(self): def testunexpectedcommanddata(self):
"""Command data frame when not running a command is an error.""" """Command data frame when not running a command is an error."""
result = self._sendsingleframe(makereactor(), result = self._sendsingleframe(makereactor(),
ffs(b'1 1 stream-begin command-data 0 ignored')) ffs(b'1 1 stream-begin command-data 0 ignored'))
self.assertaction(result, b'error') self.assertaction(result, b'error')
self.assertEqual(result[1], { self.assertEqual(result[1], {
b'message': b'expected command request frame; got 2', b'message': b'expected sender protocol settings or command request '
b'frame; got 2',
}) })
def testunexpectedcommanddatareceiving(self): def testunexpectedcommanddatareceiving(self):
"""Same as above except the command is receiving.""" """Same as above except the command is receiving."""
results = list(sendframes(makereactor(), [ results = list(sendframes(makereactor(), [
ffs(b'1 1 stream-begin command-request new|more ' ffs(b'1 1 stream-begin command-request new|more '
b"cbor:{b'name': b'ignored'}"), b"cbor:{b'name': b'ignored'}"),
ffs(b'1 1 0 command-data eos ignored'), ffs(b'1 1 0 command-data eos ignored'),
list(sendcommandframes(reactor, instream, 1, b'command1', {})) list(sendcommandframes(reactor, instream, 1, b'command1', {}))
outstream = reactor.makeoutputstream() outstream = reactor.makeoutputstream()
res = reactor.oncommandresponseready(outstream, 1, b'response') res = reactor.oncommandresponseready(outstream, 1, b'response')
list(res[1][b'framegen']) list(res[1][b'framegen'])
results = list(sendcommandframes(reactor, instream, 1, b'command1', {})) results = list(sendcommandframes(reactor, instream, 1, b'command1', {}))
self.assertaction(results[0], b'runcommand') self.assertaction(results[0], b'runcommand')
def testprotocolsettingsnoflags(self):
result = self._sendsingleframe(
makereactor(),
ffs(b'0 1 stream-begin sender-protocol-settings 0 '))
self.assertaction(result, b'error')
self.assertEqual(result[1], {
b'message': b'sender protocol settings frame must have '
b'continuation or end of stream flag set',
})
def testprotocolsettingsconflictflags(self):
result = self._sendsingleframe(
makereactor(),
ffs(b'0 1 stream-begin sender-protocol-settings continuation|eos '))
self.assertaction(result, b'error')
self.assertEqual(result[1], {
b'message': b'sender protocol settings frame cannot have both '
b'continuation and end of stream flags set',
})
def testprotocolsettingsemptypayload(self):
result = self._sendsingleframe(
makereactor(),
ffs(b'0 1 stream-begin sender-protocol-settings eos '))
self.assertaction(result, b'error')
self.assertEqual(result[1], {
b'message': b'sender protocol settings frame did not contain CBOR '
b'data',
})
def testprotocolsettingsmultipleobjects(self):
result = self._sendsingleframe(
makereactor(),
ffs(b'0 1 stream-begin sender-protocol-settings eos '
b'\x46foobar\x43foo'))
self.assertaction(result, b'error')
self.assertEqual(result[1], {
b'message': b'sender protocol settings frame contained multiple '
b'CBOR values',
})
def testprotocolsettingscontentencodings(self):
reactor = makereactor()
result = self._sendsingleframe(
reactor,
ffs(b'0 1 stream-begin sender-protocol-settings eos '
b'cbor:{b"contentencodings": [b"a", b"b"]}'))
self.assertaction(result, b'wantframe')
self.assertEqual(reactor._state, b'idle')
self.assertEqual(reactor._sendersettings[b'contentencodings'],
[b'a', b'b'])
def testprotocolsettingsmultipleframes(self):
reactor = makereactor()
data = b''.join(cborutil.streamencode({
b'contentencodings': [b'value1', b'value2'],
}))
results = list(sendframes(reactor, [
ffs(b'0 1 stream-begin sender-protocol-settings continuation %s' %
data[0:5]),
ffs(b'0 1 0 sender-protocol-settings eos %s' % data[5:]),
]))
self.assertEqual(len(results), 2)
self.assertaction(results[0], b'wantframe')
self.assertaction(results[1], b'wantframe')
self.assertEqual(reactor._state, b'idle')
self.assertEqual(reactor._sendersettings[b'contentencodings'],
[b'value1', b'value2'])
def testprotocolsettingsbadcbor(self):
result = self._sendsingleframe(
makereactor(),
ffs(b'0 1 stream-begin sender-protocol-settings eos badvalue'))
self.assertaction(result, b'error')
def testprotocolsettingsnoninitial(self):
# Cannot have protocol settings frames as non-initial frames.
reactor = makereactor()
stream = framing.stream(1)
results = list(sendcommandframes(reactor, stream, 1, b'mycommand', {}))
self.assertEqual(len(results), 1)
self.assertaction(results[0], b'runcommand')
result = self._sendsingleframe(
reactor,
ffs(b'0 1 0 sender-protocol-settings eos '))
self.assertaction(result, b'error')
self.assertEqual(result[1], {
b'message': b'expected command request frame; got 8',
})
if __name__ == '__main__': if __name__ == '__main__':
import silenttestrunner import silenttestrunner
silenttestrunner.main(__name__) silenttestrunner.main(__name__)