This is an archive of the discontinued Mercurial Phabricator instance.

Differential D2947

wireproto: explicit API to create outgoing streams
ClosedPublic

Authored by indygreg on Mar 26 2018, 2:12 PM.

Details

Reviewers
durin42
Group Reviewers
hg-reviewers
Commits
rHG5fadc63ac99f: wireproto: explicit API to create outgoing streams
Summary

It is better to create outgoing streams through the reactor so the
reactor knows about what streams are active and can track them
accordingly.

Test output changes slightly because frames from subsequent responses
no longer have the "stream begin" stream flag set because the stream
is now used across all responses.

Diff Detail

Repository
rHG Mercurial
Lint
Lint Skipped
Unit
Unit Tests Skipped

Event Timeline

indygreg created this revision.Mar 26 2018, 2:12 PM
indygreg updated this revision to Diff 7326.Mar 26 2018, 5:35 PM
indygreg updated this revision to Diff 7436.Mar 30 2018, 4:32 PM
durin42 accepted this revision.Apr 3 2018, 11:28 AM
This revision is now accepted and ready to land.Apr 3 2018, 11:28 AM
Closed by commit rHG5fadc63ac99f: wireproto: explicit API to create outgoing streams (authored by indygreg). · Explain WhyApr 3 2018, 12:49 PM
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

Diff 7436

mercurial/wireprotoframing.py

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 = 'idle'
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 = {}
# 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()
def onframerecv(self, frame): def onframerecv(self, frame):
def onapplicationerror(self, stream, requestid, msg): def onapplicationerror(self, stream, requestid, msg):
ensureserverstream(stream) ensureserverstream(stream)
return 'sendframes', { return 'sendframes', {
'framegen': createerrorframe(stream, requestid, msg, 'framegen': createerrorframe(stream, requestid, msg,
application=True), application=True),
} }
def makeoutputstream(self):
"""Create a stream to be used for sending data to the client."""
streamid = self._nextoutgoingstreamid
self._nextoutgoingstreamid += 2
s = stream(streamid)
self._outgoingstreams[streamid] = s
return s
def _makeerrorresult(self, msg): def _makeerrorresult(self, msg):
return 'error', { return 'error', {
'message': msg, 'message': msg,
} }
def _makeruncommandresult(self, requestid): def _makeruncommandresult(self, requestid):
entry = self._receivingcommands[requestid] entry = self._receivingcommands[requestid]
del self._receivingcommands[requestid] del self._receivingcommands[requestid]

mercurial/wireprotoserver.py

frames for evaluation. frames for evaluation.
""" """
# TODO Some HTTP clients are full duplex and can receive data before # TODO Some HTTP clients are full duplex and can receive data before
# the entire request is transmitted. Figure out a way to indicate support # the entire request is transmitted. Figure out a way to indicate support
# for that so we can opt into full duplex mode. # for that so we can opt into full duplex mode.
reactor = wireprotoframing.serverreactor(deferoutput=True) reactor = wireprotoframing.serverreactor(deferoutput=True)
seencommand = False seencommand = False
outstream = reactor.makeoutputstream()
while True: while True:
frame = wireprotoframing.readframe(req.bodyfh) frame = wireprotoframing.readframe(req.bodyfh)
if not frame: if not frame:
break break
action, meta = reactor.onframerecv(frame) action, meta = reactor.onframerecv(frame)
if action == 'wantframe': if action == 'wantframe':
# Need more data before we can do anything. # Need more data before we can do anything.
continue continue
elif action == 'runcommand': elif action == 'runcommand':
sentoutput = _httpv2runcommand(ui, repo, req, res, authedperm, sentoutput = _httpv2runcommand(ui, repo, req, res, authedperm,
reqcommand, reactor, meta, reqcommand, reactor, outstream,
issubsequent=seencommand) meta, issubsequent=seencommand)
if sentoutput: if sentoutput:
return return
seencommand = True seencommand = True
elif action == 'error': elif action == 'error':
# TODO define proper error mechanism. # TODO define proper error mechanism.
res.setbodygen(meta['framegen']) res.setbodygen(meta['framegen'])
elif action == 'noop': elif action == 'noop':
pass pass
else: else:
raise error.ProgrammingError('unhandled action from frame processor: %s' raise error.ProgrammingError('unhandled action from frame processor: %s'
% action) % action)
def _httpv2runcommand(ui, repo, req, res, authedperm, reqcommand, reactor, def _httpv2runcommand(ui, repo, req, res, authedperm, reqcommand, reactor,
command, issubsequent): outstream, command, issubsequent):
"""Dispatch a wire protocol command made from HTTPv2 requests. """Dispatch a wire protocol command made from HTTPv2 requests.
The authenticated permission (``authedperm``) along with the original The authenticated permission (``authedperm``) along with the original
command from the URL (``reqcommand``) are passed in. command from the URL (``reqcommand``) are passed in.
""" """
# We already validated that the session has permissions to perform the # We already validated that the session has permissions to perform the
# actions in ``authedperm``. In the unified frame protocol, the canonical # actions in ``authedperm``. In the unified frame protocol, the canonical
# command to run is expressed in a frame. However, the URL also requested # command to run is expressed in a frame. However, the URL also requested
res.headers[b'Content-Type'] = b'text/plain' res.headers[b'Content-Type'] = b'text/plain'
res.setbodybytes(_('command in frame must match command in URL')) res.setbodybytes(_('command in frame must match command in URL'))
return True return True
rsp = wireproto.dispatch(repo, proto, command['command']) rsp = wireproto.dispatch(repo, proto, command['command'])
res.status = b'200 OK' res.status = b'200 OK'
res.headers[b'Content-Type'] = FRAMINGTYPE res.headers[b'Content-Type'] = FRAMINGTYPE
stream = wireprotoframing.stream(2)
if isinstance(rsp, wireprototypes.bytesresponse): if isinstance(rsp, wireprototypes.bytesresponse):
action, meta = reactor.onbytesresponseready(stream, action, meta = reactor.onbytesresponseready(outstream,
command['requestid'], command['requestid'],
rsp.data) rsp.data)
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'])

tests/test-http-api-httpv2.t

s> Date: $HTTP_DATE$\r\n s> Date: $HTTP_DATE$\r\n
s> Content-Type: application/mercurial-exp-framing-0002\r\n s> Content-Type: application/mercurial-exp-framing-0002\r\n
s> Transfer-Encoding: chunked\r\n s> Transfer-Encoding: chunked\r\n
s> \r\n s> \r\n
s> *\r\n (glob) s> *\r\n (glob)
s> \x1d\x00\x00\x01\x00\x02\x01Bcustomreadonly bytes response s> \x1d\x00\x00\x01\x00\x02\x01Bcustomreadonly bytes response
s> \r\n s> \r\n
s> 25\r\n s> 25\r\n
s> \x1d\x00\x00\x03\x00\x02\x01Bcustomreadonly bytes response s> \x1d\x00\x00\x03\x00\x02\x00Bcustomreadonly bytes response
s> \r\n s> \r\n
s> 0\r\n s> 0\r\n
s> \r\n s> \r\n
Interleaved requests to "multirequest" are processed Interleaved requests to "multirequest" are processed
$ send << EOF $ send << EOF
> httprequest POST api/$HTTPV2/ro/multirequest > httprequest POST api/$HTTPV2/ro/multirequest
s> Date: $HTTP_DATE$\r\n s> Date: $HTTP_DATE$\r\n
s> Content-Type: application/mercurial-exp-framing-0002\r\n s> Content-Type: application/mercurial-exp-framing-0002\r\n
s> Transfer-Encoding: chunked\r\n s> Transfer-Encoding: chunked\r\n
s> \r\n s> \r\n
s> 8\r\n s> 8\r\n
s> \x00\x00\x00\x03\x00\x02\x01B s> \x00\x00\x00\x03\x00\x02\x01B
s> \r\n s> \r\n
s> 26\r\n s> 26\r\n
s> \x1e\x00\x00\x01\x00\x02\x01Bbookmarks \n s> \x1e\x00\x00\x01\x00\x02\x00Bbookmarks \n
s> namespaces \n s> namespaces \n
s> phases s> phases
s> \r\n s> \r\n
s> 0\r\n s> 0\r\n
s> \r\n s> \r\n
Restart server to disable read-write access Restart server to disable read-write access

tests/test-wireproto-serverreactor.py

self.assertEqual(result[1], { self.assertEqual(result[1], {
'message': b'missing frame flags on command frame', 'message': b'missing frame flags on command frame',
}) })
def testconflictingrequestidallowed(self): def testconflictingrequestidallowed(self):
"""Multiple fully serviced commands with same request ID is allowed.""" """Multiple fully serviced commands with same request ID is allowed."""
reactor = makereactor() reactor = makereactor()
results = [] results = []
outstream = framing.stream(2) outstream = reactor.makeoutputstream()
results.append(self._sendsingleframe( results.append(self._sendsingleframe(
reactor, ffs(b'1 1 stream-begin command-name eos command'))) reactor, ffs(b'1 1 stream-begin command-name eos command')))
result = reactor.onbytesresponseready(outstream, 1, b'response1') result = reactor.onbytesresponseready(outstream, 1, b'response1')
self.assertaction(result, 'sendframes') self.assertaction(result, 'sendframes')
list(result[1]['framegen']) list(result[1]['framegen'])
results.append(self._sendsingleframe( results.append(self._sendsingleframe(
reactor, ffs(b'1 1 0 command-name eos command'))) reactor, ffs(b'1 1 0 command-name eos command')))
result = reactor.onbytesresponseready(outstream, 1, b'response2') result = reactor.onbytesresponseready(outstream, 1, b'response2')
}) })
def testsimpleresponse(self): def testsimpleresponse(self):
"""Bytes response to command sends result frames.""" """Bytes response to command sends result frames."""
reactor = makereactor() reactor = makereactor()
instream = framing.stream(1) instream = framing.stream(1)
list(sendcommandframes(reactor, instream, 1, b'mycommand', {})) list(sendcommandframes(reactor, instream, 1, b'mycommand', {}))
outstream = framing.stream(2) outstream = reactor.makeoutputstream()
result = reactor.onbytesresponseready(outstream, 1, b'response') result = reactor.onbytesresponseready(outstream, 1, b'response')
self.assertaction(result, 'sendframes') self.assertaction(result, 'sendframes')
self.assertframesequal(result[1]['framegen'], [ self.assertframesequal(result[1]['framegen'], [
b'1 2 stream-begin bytes-response eos response', b'1 2 stream-begin bytes-response eos response',
]) ])
def testmultiframeresponse(self): def testmultiframeresponse(self):
"""Bytes response spanning multiple frames is handled.""" """Bytes response spanning multiple frames is handled."""
first = b'x' * framing.DEFAULT_MAX_FRAME_SIZE first = b'x' * framing.DEFAULT_MAX_FRAME_SIZE
second = b'y' * 100 second = b'y' * 100
reactor = makereactor() reactor = makereactor()
instream = framing.stream(1) instream = framing.stream(1)
list(sendcommandframes(reactor, instream, 1, b'mycommand', {})) list(sendcommandframes(reactor, instream, 1, b'mycommand', {}))
outstream = framing.stream(2) outstream = reactor.makeoutputstream()
result = reactor.onbytesresponseready(outstream, 1, first + second) result = reactor.onbytesresponseready(outstream, 1, first + second)
self.assertaction(result, 'sendframes') self.assertaction(result, 'sendframes')
self.assertframesequal(result[1]['framegen'], [ self.assertframesequal(result[1]['framegen'], [
b'1 2 stream-begin bytes-response continuation %s' % first, b'1 2 stream-begin bytes-response continuation %s' % first,
b'1 2 0 bytes-response eos %s' % second, b'1 2 0 bytes-response eos %s' % second,
]) ])
def testapplicationerror(self): def testapplicationerror(self):
reactor = makereactor() reactor = makereactor()
instream = framing.stream(1) instream = framing.stream(1)
list(sendcommandframes(reactor, instream, 1, b'mycommand', {})) list(sendcommandframes(reactor, instream, 1, b'mycommand', {}))
outstream = framing.stream(2) outstream = reactor.makeoutputstream()
result = reactor.onapplicationerror(outstream, 1, b'some message') result = reactor.onapplicationerror(outstream, 1, b'some message')
self.assertaction(result, 'sendframes') self.assertaction(result, 'sendframes')
self.assertframesequal(result[1]['framegen'], [ self.assertframesequal(result[1]['framegen'], [
b'1 2 stream-begin error-response application some message', b'1 2 stream-begin error-response application some message',
]) ])
def test1commanddeferresponse(self): def test1commanddeferresponse(self):
"""Responses when in deferred output mode are delayed until EOF.""" """Responses when in deferred output mode are delayed until EOF."""
reactor = makereactor(deferoutput=True) reactor = makereactor(deferoutput=True)
instream = framing.stream(1) instream = framing.stream(1)
results = list(sendcommandframes(reactor, instream, 1, b'mycommand', results = list(sendcommandframes(reactor, instream, 1, b'mycommand',
{})) {}))
self.assertEqual(len(results), 1) self.assertEqual(len(results), 1)
self.assertaction(results[0], 'runcommand') self.assertaction(results[0], 'runcommand')
outstream = framing.stream(2) outstream = reactor.makeoutputstream()
result = reactor.onbytesresponseready(outstream, 1, b'response') result = reactor.onbytesresponseready(outstream, 1, b'response')
self.assertaction(result, 'noop') self.assertaction(result, 'noop')
result = reactor.oninputeof() result = reactor.oninputeof()
self.assertaction(result, 'sendframes') self.assertaction(result, 'sendframes')
self.assertframesequal(result[1]['framegen'], [ self.assertframesequal(result[1]['framegen'], [
b'1 2 stream-begin bytes-response eos response', b'1 2 stream-begin bytes-response eos response',
]) ])
def testmultiplecommanddeferresponse(self): def testmultiplecommanddeferresponse(self):
reactor = makereactor(deferoutput=True) reactor = makereactor(deferoutput=True)
instream = framing.stream(1) instream = framing.stream(1)
list(sendcommandframes(reactor, instream, 1, b'command1', {})) list(sendcommandframes(reactor, instream, 1, b'command1', {}))
list(sendcommandframes(reactor, instream, 3, b'command2', {})) list(sendcommandframes(reactor, instream, 3, b'command2', {}))
outstream = framing.stream(2) outstream = reactor.makeoutputstream()
result = reactor.onbytesresponseready(outstream, 1, b'response1') result = reactor.onbytesresponseready(outstream, 1, b'response1')
self.assertaction(result, 'noop') self.assertaction(result, 'noop')
result = reactor.onbytesresponseready(outstream, 3, b'response2') result = reactor.onbytesresponseready(outstream, 3, b'response2')
self.assertaction(result, 'noop') self.assertaction(result, 'noop')
result = reactor.oninputeof() result = reactor.oninputeof()
self.assertaction(result, 'sendframes') self.assertaction(result, 'sendframes')
self.assertframesequal(result[1]['framegen'], [ self.assertframesequal(result[1]['framegen'], [
b'1 2 stream-begin bytes-response eos response1', b'1 2 stream-begin bytes-response eos response1',
b'3 2 0 bytes-response eos response2' b'3 2 0 bytes-response eos response2'
]) ])
def testrequestidtracking(self): def testrequestidtracking(self):
reactor = makereactor(deferoutput=True) reactor = makereactor(deferoutput=True)
instream = framing.stream(1) instream = framing.stream(1)
list(sendcommandframes(reactor, instream, 1, b'command1', {})) list(sendcommandframes(reactor, instream, 1, b'command1', {}))
list(sendcommandframes(reactor, instream, 3, b'command2', {})) list(sendcommandframes(reactor, instream, 3, b'command2', {}))
list(sendcommandframes(reactor, instream, 5, b'command3', {})) list(sendcommandframes(reactor, instream, 5, b'command3', {}))
# Register results for commands out of order. # Register results for commands out of order.
outstream = framing.stream(2) outstream = reactor.makeoutputstream()
reactor.onbytesresponseready(outstream, 3, b'response3') reactor.onbytesresponseready(outstream, 3, b'response3')
reactor.onbytesresponseready(outstream, 1, b'response1') reactor.onbytesresponseready(outstream, 1, b'response1')
reactor.onbytesresponseready(outstream, 5, b'response5') reactor.onbytesresponseready(outstream, 5, b'response5')
result = reactor.oninputeof() result = reactor.oninputeof()
self.assertaction(result, 'sendframes') self.assertaction(result, 'sendframes')
self.assertframesequal(result[1]['framegen'], [ self.assertframesequal(result[1]['framegen'], [
b'3 2 stream-begin bytes-response eos response3', b'3 2 stream-begin bytes-response eos response3',
'message': b'request with ID 1 is already active', 'message': b'request with ID 1 is already active',
}) })
def testduplicaterequestonactivecommandnosend(self): def testduplicaterequestonactivecommandnosend(self):
"""Same as above but we've registered a response but haven't sent it.""" """Same as above but we've registered a response but haven't sent it."""
reactor = makereactor() reactor = makereactor()
instream = framing.stream(1) instream = framing.stream(1)
list(sendcommandframes(reactor, instream, 1, b'command1', {})) list(sendcommandframes(reactor, instream, 1, b'command1', {}))
outstream = framing.stream(2) outstream = reactor.makeoutputstream()
reactor.onbytesresponseready(outstream, 1, b'response') reactor.onbytesresponseready(outstream, 1, b'response')
# We've registered the response but haven't sent it. From the # We've registered the response but haven't sent it. From the
# perspective of the reactor, the command is still active. # perspective of the reactor, the command is still active.
results = list(sendcommandframes(reactor, instream, 1, b'command1', {})) results = list(sendcommandframes(reactor, instream, 1, b'command1', {}))
self.assertaction(results[0], 'error') self.assertaction(results[0], 'error')
self.assertEqual(results[0][1], { self.assertEqual(results[0][1], {
'message': 'received frame for request that is still active: 1', 'message': 'received frame for request that is still active: 1',
}) })
def testduplicaterequestaftersend(self): def testduplicaterequestaftersend(self):
"""We can use a duplicate request ID after we've sent the response.""" """We can use a duplicate request ID after we've sent the response."""
reactor = makereactor() reactor = makereactor()
instream = framing.stream(1) instream = framing.stream(1)
list(sendcommandframes(reactor, instream, 1, b'command1', {})) list(sendcommandframes(reactor, instream, 1, b'command1', {}))
outstream = framing.stream(2) outstream = reactor.makeoutputstream()
res = reactor.onbytesresponseready(outstream, 1, b'response') res = reactor.onbytesresponseready(outstream, 1, b'response')
list(res[1]['framegen']) list(res[1]['framegen'])
results = list(sendcommandframes(reactor, instream, 1, b'command1', {})) results = list(sendcommandframes(reactor, instream, 1, b'command1', {}))
self.assertaction(results[0], 'runcommand') self.assertaction(results[0], 'runcommand')
if __name__ == '__main__': if __name__ == '__main__':
import silenttestrunner import silenttestrunner
silenttestrunner.main(__name__) silenttestrunner.main(__name__)