( )⚙ D4919 wireprotov2: pass ui into clientreactor and serverreactor

This is an archive of the discontinued Mercurial Phabricator instance.

Differential D4919

wireprotov2: pass ui into clientreactor and serverreactor
ClosedPublic

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

Details

Reviewers
None
Group Reviewers
hg-reviewers
Commits
rHG293835e0fff7: wireprotov2: pass ui into clientreactor and serverreactor
Summary

This will allow us to use config options to influence compression
settings.

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.Oct 8 2018, 8:29 PM
durin42 added a subscriber: durin42.Oct 10 2018, 10:46 AM

I'm a little bummed to pass all of UI down in there, but probably not worth addressing until the design settles down more.

Closed by commit rHG293835e0fff7: wireprotov2: pass ui into clientreactor and serverreactor (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 11783

mercurial/httppeer.py

def _callcompressable(self, cmd, **args): def _callcompressable(self, cmd, **args):
return self._callstream(cmd, _compressible=True, **args) return self._callstream(cmd, _compressible=True, **args)
def _abort(self, exception): def _abort(self, exception):
raise exception raise exception
def sendv2request(ui, opener, requestbuilder, apiurl, permission, requests, def sendv2request(ui, opener, requestbuilder, apiurl, permission, requests,
redirect): redirect):
reactor = wireprotoframing.clientreactor(hasmultiplesend=False, reactor = wireprotoframing.clientreactor(ui,
hasmultiplesend=False,
buffersends=True) buffersends=True)
handler = wireprotov2peer.clienthandler(ui, reactor, handler = wireprotov2peer.clienthandler(ui, reactor,
opener=opener, opener=opener,
requestbuilder=requestbuilder) requestbuilder=requestbuilder)
url = '%s/%s' % (apiurl, permission) url = '%s/%s' % (apiurl, permission)

mercurial/wireprotoframing.py

to facilitate graceful shutdown? to facilitate graceful shutdown?
Active requests that haven't been responded to aren't tracked. This means Active requests that haven't been responded to aren't tracked. This means
that if we receive a command and instruct its dispatch, another command that if we receive a command and instruct its dispatch, another command
with its request ID can come in over the wire and there will be a race with its request ID can come in over the wire and there will be a race
between who responds to what. between who responds to what.
""" """
def __init__(self, deferoutput=False): def __init__(self, ui, deferoutput=False):
"""Construct a new server reactor. """Construct a new server reactor.
``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._ui = ui
self._deferoutput = deferoutput self._deferoutput = deferoutput
self._state = 'initial' 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.
represents the request this data is for. represents the request this data is for.
The ``data`` key contains the decoded data from the server. The ``data`` key contains the decoded data from the server.
``expectmore`` and ``eos`` evaluate to True when more response data ``expectmore`` and ``eos`` evaluate to True when more response data
is expected to follow or we're at the end of the response stream, is expected to follow or we're at the end of the response stream,
respectively. respectively.
""" """
def __init__(self, hasmultiplesend=False, buffersends=True): def __init__(self, ui, hasmultiplesend=False, buffersends=True):
"""Create a new instance. """Create a new instance.
``hasmultiplesend`` indicates whether multiple sends are supported ``hasmultiplesend`` indicates whether multiple sends are supported
by the transport. When True, it is possible to send commands immediately by the transport. When True, it is possible to send commands immediately
instead of buffering until the caller signals an intent to finish a instead of buffering until the caller signals an intent to finish a
send operation. send operation.
``buffercommands`` indicates whether sends should be buffered until the ``buffercommands`` indicates whether sends should be buffered until the
last request has been issued. last request has been issued.
""" """
self._ui = ui
self._hasmultiplesend = hasmultiplesend self._hasmultiplesend = hasmultiplesend
self._buffersends = buffersends self._buffersends = buffersends
self._canissuecommands = True self._canissuecommands = True
self._cansend = True self._cansend = True
self._nextrequestid = 1 self._nextrequestid = 1
# We only support a single outgoing stream for now. # We only support a single outgoing stream for now.

mercurial/wireprotov2server.py

if not ui.configbool('experimental', 'web.api.debugreflect'): if not ui.configbool('experimental', 'web.api.debugreflect'):
res.status = b'404 Not Found' res.status = b'404 Not Found'
res.headers[b'Content-Type'] = b'text/plain' res.headers[b'Content-Type'] = b'text/plain'
res.setbodybytes(_('debugreflect service not available')) res.setbodybytes(_('debugreflect service not available'))
return return
# We assume we have a unified framing protocol request body. # We assume we have a unified framing protocol request body.
reactor = wireprotoframing.serverreactor() reactor = wireprotoframing.serverreactor(ui)
states = [] states = []
while True: while True:
frame = wireprotoframing.readframe(req.bodyfh) frame = wireprotoframing.readframe(req.bodyfh)
if not frame: if not frame:
states.append(b'received: <no frame>') states.append(b'received: <no frame>')
break break
"""Post-validation handler for HTTPv2 requests. """Post-validation handler for HTTPv2 requests.
Called when the HTTP request contains unified frame-based protocol Called when the HTTP request contains unified frame-based protocol
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(ui, deferoutput=True)
seencommand = False seencommand = False
outstream = reactor.makeoutputstream() 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

tests/test-wireproto-clientreactor.py

from __future__ import absolute_import from __future__ import absolute_import
import unittest import unittest
from mercurial import ( from mercurial import (
error, error,
ui as uimod,
wireprotoframing as framing, wireprotoframing as framing,
) )
from mercurial.utils import ( from mercurial.utils import (
cborutil, cborutil,
) )
ffs = framing.makeframefromhumanstring ffs = framing.makeframefromhumanstring
globalui = uimod.ui()
def sendframe(reactor, frame): def sendframe(reactor, frame):
"""Send a frame bytearray to a reactor.""" """Send a frame bytearray to a reactor."""
header = framing.parseheader(frame) header = framing.parseheader(frame)
payload = frame[framing.FRAME_HEADER_SIZE:] payload = frame[framing.FRAME_HEADER_SIZE:]
assert len(payload) == header.length assert len(payload) == header.length
return reactor.onframerecv(framing.frame(header.requestid, return reactor.onframerecv(framing.frame(header.requestid,
header.streamid, header.streamid,
header.streamflags, header.streamflags,
header.typeid, header.typeid,
header.flags, header.flags,
payload)) payload))
class SingleSendTests(unittest.TestCase): class SingleSendTests(unittest.TestCase):
"""A reactor that can only send once rejects subsequent sends.""" """A reactor that can only send once rejects subsequent sends."""
if not getattr(unittest.TestCase, 'assertRaisesRegex', False): if not getattr(unittest.TestCase, 'assertRaisesRegex', False):
# Python 3.7 deprecates the regex*p* version, but 2.7 lacks # Python 3.7 deprecates the regex*p* version, but 2.7 lacks
# the regex version. # the regex version.
assertRaisesRegex = (# camelcase-required assertRaisesRegex = (# camelcase-required
unittest.TestCase.assertRaisesRegexp) unittest.TestCase.assertRaisesRegexp)
def testbasic(self): def testbasic(self):
reactor = framing.clientreactor(hasmultiplesend=False, buffersends=True) reactor = framing.clientreactor(globalui,
hasmultiplesend=False,
buffersends=True)
request, action, meta = reactor.callcommand(b'foo', {}) request, action, meta = reactor.callcommand(b'foo', {})
self.assertEqual(request.state, b'pending') self.assertEqual(request.state, b'pending')
self.assertEqual(action, b'noop') self.assertEqual(action, b'noop')
action, meta = reactor.flushcommands() action, meta = reactor.flushcommands()
self.assertEqual(action, b'sendframes') self.assertEqual(action, b'sendframes')
for frame in meta[b'framegen']: for frame in meta[b'framegen']:
self.assertEqual(request.state, b'sending') self.assertEqual(request.state, b'sending')
self.assertEqual(request.state, b'sent') self.assertEqual(request.state, b'sent')
with self.assertRaisesRegex(error.ProgrammingError, with self.assertRaisesRegex(error.ProgrammingError,
'cannot issue new commands'): 'cannot issue new commands'):
reactor.callcommand(b'foo', {}) reactor.callcommand(b'foo', {})
with self.assertRaisesRegex(error.ProgrammingError, with self.assertRaisesRegex(error.ProgrammingError,
'cannot issue new commands'): 'cannot issue new commands'):
reactor.callcommand(b'foo', {}) reactor.callcommand(b'foo', {})
class NoBufferTests(unittest.TestCase): class NoBufferTests(unittest.TestCase):
"""A reactor without send buffering sends requests immediately.""" """A reactor without send buffering sends requests immediately."""
def testbasic(self): def testbasic(self):
reactor = framing.clientreactor(hasmultiplesend=True, buffersends=False) reactor = framing.clientreactor(globalui,
hasmultiplesend=True,
buffersends=False)
request, action, meta = reactor.callcommand(b'command1', {}) request, action, meta = reactor.callcommand(b'command1', {})
self.assertEqual(request.requestid, 1) self.assertEqual(request.requestid, 1)
self.assertEqual(action, b'sendframes') self.assertEqual(action, b'sendframes')
self.assertEqual(request.state, b'pending') self.assertEqual(request.state, b'pending')
for frame in meta[b'framegen']: for frame in meta[b'framegen']:
class BadFrameRecvTests(unittest.TestCase): class BadFrameRecvTests(unittest.TestCase):
if not getattr(unittest.TestCase, 'assertRaisesRegex', False): if not getattr(unittest.TestCase, 'assertRaisesRegex', False):
# Python 3.7 deprecates the regex*p* version, but 2.7 lacks # Python 3.7 deprecates the regex*p* version, but 2.7 lacks
# the regex version. # the regex version.
assertRaisesRegex = (# camelcase-required assertRaisesRegex = (# camelcase-required
unittest.TestCase.assertRaisesRegexp) unittest.TestCase.assertRaisesRegexp)
def testoddstream(self): def testoddstream(self):
reactor = framing.clientreactor() reactor = framing.clientreactor(globalui)
action, meta = sendframe(reactor, ffs(b'1 1 0 1 0 foo')) action, meta = sendframe(reactor, ffs(b'1 1 0 1 0 foo'))
self.assertEqual(action, b'error') self.assertEqual(action, b'error')
self.assertEqual(meta[b'message'], self.assertEqual(meta[b'message'],
b'received frame with odd numbered stream ID: 1') b'received frame with odd numbered stream ID: 1')
def testunknownstream(self): def testunknownstream(self):
reactor = framing.clientreactor() reactor = framing.clientreactor(globalui)
action, meta = sendframe(reactor, ffs(b'1 0 0 1 0 foo')) action, meta = sendframe(reactor, ffs(b'1 0 0 1 0 foo'))
self.assertEqual(action, b'error') self.assertEqual(action, b'error')
self.assertEqual(meta[b'message'], self.assertEqual(meta[b'message'],
b'received frame on unknown stream without beginning ' b'received frame on unknown stream without beginning '
b'of stream flag set') b'of stream flag set')
def testunhandledframetype(self): def testunhandledframetype(self):
reactor = framing.clientreactor(buffersends=False) reactor = framing.clientreactor(globalui, buffersends=False)
request, action, meta = reactor.callcommand(b'foo', {}) request, action, meta = reactor.callcommand(b'foo', {})
for frame in meta[b'framegen']: for frame in meta[b'framegen']:
pass pass
with self.assertRaisesRegex(error.ProgrammingError, with self.assertRaisesRegex(error.ProgrammingError,
'unhandled frame type'): 'unhandled frame type'):
sendframe(reactor, ffs(b'1 0 stream-begin text-output 0 foo')) sendframe(reactor, ffs(b'1 0 stream-begin text-output 0 foo'))
class StreamTests(unittest.TestCase): class StreamTests(unittest.TestCase):
def testmultipleresponseframes(self): def testmultipleresponseframes(self):
reactor = framing.clientreactor(buffersends=False) reactor = framing.clientreactor(globalui, buffersends=False)
request, action, meta = reactor.callcommand(b'foo', {}) request, action, meta = reactor.callcommand(b'foo', {})
self.assertEqual(action, b'sendframes') self.assertEqual(action, b'sendframes')
for f in meta[b'framegen']: for f in meta[b'framegen']:
pass pass
action, meta = sendframe( action, meta = sendframe(
reactor, reactor,
ffs(b'%d 0 stream-begin command-response 0 foo' % ffs(b'%d 0 stream-begin command-response 0 foo' %
request.requestid)) request.requestid))
self.assertEqual(action, b'responsedata') self.assertEqual(action, b'responsedata')
action, meta = sendframe( action, meta = sendframe(
reactor, reactor,
ffs(b'%d 0 0 command-response eos bar' % request.requestid)) ffs(b'%d 0 0 command-response eos bar' % request.requestid))
self.assertEqual(action, b'responsedata') self.assertEqual(action, b'responsedata')
class RedirectTests(unittest.TestCase): class RedirectTests(unittest.TestCase):
def testredirect(self): def testredirect(self):
reactor = framing.clientreactor(buffersends=False) reactor = framing.clientreactor(globalui, buffersends=False)
redirect = { redirect = {
b'targets': [b'a', b'b'], b'targets': [b'a', b'b'],
b'hashes': [b'sha256'], b'hashes': [b'sha256'],
} }
request, action, meta = reactor.callcommand( request, action, meta = reactor.callcommand(
b'foo', {}, redirect=redirect) b'foo', {}, redirect=redirect)
self.assertEqual(action, b'sendframes') self.assertEqual(action, b'sendframes')
frames = list(meta[b'framegen']) frames = list(meta[b'framegen'])
self.assertEqual(len(frames), 1) self.assertEqual(len(frames), 1)
self.assertEqual(frames[0], self.assertEqual(frames[0],
ffs(b'1 1 stream-begin command-request new ' ffs(b'1 1 stream-begin command-request new '
b"cbor:{b'name': b'foo', " b"cbor:{b'name': b'foo', "
b"b'redirect': {b'targets': [b'a', b'b'], " b"b'redirect': {b'targets': [b'a', b'b'], "
b"b'hashes': [b'sha256']}}")) b"b'hashes': [b'sha256']}}"))
class StreamSettingsTests(unittest.TestCase): class StreamSettingsTests(unittest.TestCase):
def testnoflags(self): def testnoflags(self):
reactor = framing.clientreactor(buffersends=False) reactor = framing.clientreactor(globalui, buffersends=False)
request, action, meta = reactor.callcommand(b'foo', {}) request, action, meta = reactor.callcommand(b'foo', {})
for f in meta[b'framegen']: for f in meta[b'framegen']:
pass pass
action, meta = sendframe(reactor, action, meta = sendframe(reactor,
ffs(b'1 2 stream-begin stream-settings 0 ')) ffs(b'1 2 stream-begin stream-settings 0 '))
self.assertEqual(action, b'error') self.assertEqual(action, b'error')
self.assertEqual(meta, { self.assertEqual(meta, {
b'message': b'stream encoding settings frame must have ' b'message': b'stream encoding settings frame must have '
b'continuation or end of stream flag set', b'continuation or end of stream flag set',
}) })
def testconflictflags(self): def testconflictflags(self):
reactor = framing.clientreactor(buffersends=False) reactor = framing.clientreactor(globalui, buffersends=False)
request, action, meta = reactor.callcommand(b'foo', {}) request, action, meta = reactor.callcommand(b'foo', {})
for f in meta[b'framegen']: for f in meta[b'framegen']:
pass pass
action, meta = sendframe(reactor, action, meta = sendframe(reactor,
ffs(b'1 2 stream-begin stream-settings continuation|eos ')) ffs(b'1 2 stream-begin stream-settings continuation|eos '))
self.assertEqual(action, b'error') self.assertEqual(action, b'error')
self.assertEqual(meta, { self.assertEqual(meta, {
b'message': b'stream encoding settings frame cannot have both ' b'message': b'stream encoding settings frame cannot have both '
b'continuation and end of stream flags set', b'continuation and end of stream flags set',
}) })
def testemptypayload(self): def testemptypayload(self):
reactor = framing.clientreactor(buffersends=False) reactor = framing.clientreactor(globalui, buffersends=False)
request, action, meta = reactor.callcommand(b'foo', {}) request, action, meta = reactor.callcommand(b'foo', {})
for f in meta[b'framegen']: for f in meta[b'framegen']:
pass pass
action, meta = sendframe(reactor, action, meta = sendframe(reactor,
ffs(b'1 2 stream-begin stream-settings eos ')) ffs(b'1 2 stream-begin stream-settings eos '))
self.assertEqual(action, b'error') self.assertEqual(action, b'error')
self.assertEqual(meta, { self.assertEqual(meta, {
b'message': b'stream encoding settings frame did not contain ' b'message': b'stream encoding settings frame did not contain '
b'CBOR data' b'CBOR data'
}) })
def testbadcbor(self): def testbadcbor(self):
reactor = framing.clientreactor(buffersends=False) reactor = framing.clientreactor(globalui, buffersends=False)
request, action, meta = reactor.callcommand(b'foo', {}) request, action, meta = reactor.callcommand(b'foo', {})
for f in meta[b'framegen']: for f in meta[b'framegen']:
pass pass
action, meta = sendframe(reactor, action, meta = sendframe(reactor,
ffs(b'1 2 stream-begin stream-settings eos badvalue')) ffs(b'1 2 stream-begin stream-settings eos badvalue'))
self.assertEqual(action, b'error') self.assertEqual(action, b'error')
def testsingleobject(self): def testsingleobject(self):
reactor = framing.clientreactor(buffersends=False) reactor = framing.clientreactor(globalui, buffersends=False)
request, action, meta = reactor.callcommand(b'foo', {}) request, action, meta = reactor.callcommand(b'foo', {})
for f in meta[b'framegen']: for f in meta[b'framegen']:
pass pass
action, meta = sendframe(reactor, action, meta = sendframe(reactor,
ffs(b'1 2 stream-begin stream-settings eos cbor:b"identity"')) ffs(b'1 2 stream-begin stream-settings eos cbor:b"identity"'))
self.assertEqual(action, b'noop') self.assertEqual(action, b'noop')
self.assertEqual(meta, {}) self.assertEqual(meta, {})
def testmultipleobjects(self): def testmultipleobjects(self):
reactor = framing.clientreactor(buffersends=False) reactor = framing.clientreactor(globalui, buffersends=False)
request, action, meta = reactor.callcommand(b'foo', {}) request, action, meta = reactor.callcommand(b'foo', {})
for f in meta[b'framegen']: for f in meta[b'framegen']:
pass pass
data = b''.join([ data = b''.join([
b''.join(cborutil.streamencode(b'identity')), b''.join(cborutil.streamencode(b'identity')),
b''.join(cborutil.streamencode({b'foo', b'bar'})), b''.join(cborutil.streamencode({b'foo', b'bar'})),
]) ])
action, meta = sendframe(reactor, action, meta = sendframe(reactor,
ffs(b'1 2 stream-begin stream-settings eos %s' % data)) ffs(b'1 2 stream-begin stream-settings eos %s' % data))
self.assertEqual(action, b'noop') self.assertEqual(action, b'noop')
self.assertEqual(meta, {}) self.assertEqual(meta, {})
def testmultipleframes(self): def testmultipleframes(self):
reactor = framing.clientreactor(buffersends=False) reactor = framing.clientreactor(globalui, buffersends=False)
request, action, meta = reactor.callcommand(b'foo', {}) request, action, meta = reactor.callcommand(b'foo', {})
for f in meta[b'framegen']: for f in meta[b'framegen']:
pass pass
data = b''.join(cborutil.streamencode(b'identity')) data = b''.join(cborutil.streamencode(b'identity'))
action, meta = sendframe(reactor, action, meta = sendframe(reactor,

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 (
ui as uimod,
util, util,
wireprotoframing as framing, wireprotoframing as framing,
) )
from mercurial.utils import ( from mercurial.utils import (
cborutil, 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) ui = uimod.ui()
return framing.serverreactor(ui, deferoutput=deferoutput)
def sendframes(reactor, gen): def sendframes(reactor, gen):
"""Send a generator of frame bytearray to a reactor. """Send a generator of frame bytearray to a reactor.
Emits a generator of results from ``onframerecv()`` calls. Emits a generator of results from ``onframerecv()`` calls.
""" """
for frame in gen: for frame in gen:
header = framing.parseheader(frame) header = framing.parseheader(frame)