Mercurial > public > mercurial-scm > hg

comparison mercurial/wireprotoframing.py @ 37288:9bfcbe4f4745

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
wireproto: add streams to frame-based protocol Previously, the frame-based protocol was just a series of frames, with each frame associated with a request ID. In order to scale the protocol, we'll want to enable the use of compression. While it is possible to enable compression at the socket/pipe level, this has its disadvantages. The big one is it undermines the point of frames being standalone, atomic units that can be read and written: if you add compression above the framing protocol, you are back to having a stream-based protocol as opposed to something frame-based. So in order to preserve frames, compression needs to occur at the frame payload level. Compressing each frame's payload individually will limit compression ratios because the window size of the compressor will be limited by the max frame size, which is 32-64kb as currently defined. It will also add CPU overhead, as it is more efficient for compressors to operate on fewer, larger blocks of data than more, smaller blocks. So compressing each frame independently is out. This means we need to compress each frame's payload as if it is part of a larger stream. The simplest approach is to have 1 stream per connection. This could certainly work. However, it has disadvantages (documented below). We could also have 1 stream per RPC/command invocation. (This is the model HTTP/2 goes with.) This also has disadvantages. The main disadvantage to one global stream is that it has the very real potential to create CPU bottlenecks doing compression. Networks are only getting faster and the performance of single CPU cores has been relatively flat. Newer compression formats like zstandard offer better CPU cycle efficiency than predecessors like zlib. But it still all too common to saturate your CPU with compression overhead long before you saturate the network pipe. The main disadvantage with streams per request is that you can't reap the benefits of the compression context for multiple requests. For example, if you send 1000 RPC requests (or HTTP/2 requests for that matter), the response to each would have its own compression context. The overall size of the raw responses would be larger because compression contexts wouldn't be able to reference data from another request or response. The approach for streams as implemented in this commit is to support N streams per connection and for streams to potentially span requests and responses. As explained by the added internals docs, this facilitates servers and clients delegating independent streams and compression to independent threads / CPU cores. This helps alleviate the CPU bottleneck of compression. This design also allows compression contexts to be reused across requests/responses. This can result in improved compression ratios and less overhead for compressors and decompressors having to build new contexts. Another feature that was defined was the ability for individual frames within a stream to declare whether that individual frame's payload uses the content encoding (read: compression) defined by the stream. The idea here is that some servers may serve data from a combination of caches and dynamic resolution. Data coming from caches may be pre-compressed. We want to facilitate servers being able to essentially stream bytes from caches to the wire with minimal overhead. Being able to mix and match with frames are compressed within a stream enables these types of advanced server functionality. This commit defines the new streams mechanism. Basic code for supporting streams in frames has been added. But that code is seriously lacking and doesn't fully conform to the defined protocol. For example, we don't close any streams. And support for content encoding within streams is not yet implemented. The change was rather invasive and I didn't think it would be reasonable to implement the entire feature in a single commit. For the record, I would have loved to reuse an existing multiplexing protocol to build the new wire protocol on top of. However, I couldn't find a protocol that offers the performance and scaling characteristics that I desired. Namely, it should support multiple compression contexts to facilitate scaling out to multiple CPU cores and compression contexts should be able to live longer than single RPC requests. HTTP/2 *almost* fits the bill. But the semantics of HTTP message exchange state that streams can only live for a single request-response. We /could/ tunnel on top of HTTP/2 streams and frames with HEADER and DATA frames. But there's no guarantee that HTTP/2 libraries and proxies would allow us to use HTTP/2 streams and frames without the HTTP message exchange semantics defined in RFC 7540 Section 8. Other RPC protocols like gRPC tunnel are built on top of HTTP/2 and thus preserve its semantics of stream per RPC invocation. Even QUIC does this. We could attempt to invent a higher-level stream that spans HTTP/2 streams. But this would be violating HTTP/2 because there is no guarantee that HTTP/2 streams are routed to the same server. The best we can do - which is what this protocol does - is shoehorn all request and response data into a single HTTP message and create streams within. At that point, we've defined a Content-Type in HTTP parlance. It just so happens our media type can also work as a standalone, stream-based protocol, without leaning on HTTP or similar protocol. Differential Revision: https://phab.mercurial-scm.org/D2907
author Gregory Szorc <gregory.szorc@gmail.com>
date Mon, 26 Mar 2018 11:00:16 -0700
parents 3ed344546d9e
children 5fadc63ac99f
comparison
equal deleted inserted replaced
37285:3ed344546d9e 37288:9bfcbe4f4745
23 ) 23 )
24 from .utils import ( 24 from .utils import (
25 stringutil, 25 stringutil,
26 ) 26 )
27 27
28 FRAME_HEADER_SIZE = 6 28 FRAME_HEADER_SIZE = 8
29 DEFAULT_MAX_FRAME_SIZE = 32768 29 DEFAULT_MAX_FRAME_SIZE = 32768
30
31 STREAM_FLAG_BEGIN_STREAM = 0x01
32 STREAM_FLAG_END_STREAM = 0x02
33 STREAM_FLAG_ENCODING_APPLIED = 0x04
34
35 STREAM_FLAGS = {
36 b'stream-begin': STREAM_FLAG_BEGIN_STREAM,
37 b'stream-end': STREAM_FLAG_END_STREAM,
38 b'encoded': STREAM_FLAG_ENCODING_APPLIED,
39 }
30 40
31 FRAME_TYPE_COMMAND_NAME = 0x01 41 FRAME_TYPE_COMMAND_NAME = 0x01
32 FRAME_TYPE_COMMAND_ARGUMENT = 0x02 42 FRAME_TYPE_COMMAND_ARGUMENT = 0x02
33 FRAME_TYPE_COMMAND_DATA = 0x03 43 FRAME_TYPE_COMMAND_DATA = 0x03
34 FRAME_TYPE_BYTES_RESPONSE = 0x04 44 FRAME_TYPE_BYTES_RESPONSE = 0x04
35 FRAME_TYPE_ERROR_RESPONSE = 0x05 45 FRAME_TYPE_ERROR_RESPONSE = 0x05
36 FRAME_TYPE_TEXT_OUTPUT = 0x06 46 FRAME_TYPE_TEXT_OUTPUT = 0x06
47 FRAME_TYPE_STREAM_SETTINGS = 0x08
37 48
38 FRAME_TYPES = { 49 FRAME_TYPES = {
39 b'command-name': FRAME_TYPE_COMMAND_NAME, 50 b'command-name': FRAME_TYPE_COMMAND_NAME,
40 b'command-argument': FRAME_TYPE_COMMAND_ARGUMENT, 51 b'command-argument': FRAME_TYPE_COMMAND_ARGUMENT,
41 b'command-data': FRAME_TYPE_COMMAND_DATA, 52 b'command-data': FRAME_TYPE_COMMAND_DATA,
42 b'bytes-response': FRAME_TYPE_BYTES_RESPONSE, 53 b'bytes-response': FRAME_TYPE_BYTES_RESPONSE,
43 b'error-response': FRAME_TYPE_ERROR_RESPONSE, 54 b'error-response': FRAME_TYPE_ERROR_RESPONSE,
44 b'text-output': FRAME_TYPE_TEXT_OUTPUT, 55 b'text-output': FRAME_TYPE_TEXT_OUTPUT,
56 b'stream-settings': FRAME_TYPE_STREAM_SETTINGS,
45 } 57 }
46 58
47 FLAG_COMMAND_NAME_EOS = 0x01 59 FLAG_COMMAND_NAME_EOS = 0x01
48 FLAG_COMMAND_NAME_HAVE_ARGS = 0x02 60 FLAG_COMMAND_NAME_HAVE_ARGS = 0x02
49 FLAG_COMMAND_NAME_HAVE_DATA = 0x04 61 FLAG_COMMAND_NAME_HAVE_DATA = 0x04
92 FRAME_TYPE_COMMAND_ARGUMENT: FLAGS_COMMAND_ARGUMENT, 104 FRAME_TYPE_COMMAND_ARGUMENT: FLAGS_COMMAND_ARGUMENT,
93 FRAME_TYPE_COMMAND_DATA: FLAGS_COMMAND_DATA, 105 FRAME_TYPE_COMMAND_DATA: FLAGS_COMMAND_DATA,
94 FRAME_TYPE_BYTES_RESPONSE: FLAGS_BYTES_RESPONSE, 106 FRAME_TYPE_BYTES_RESPONSE: FLAGS_BYTES_RESPONSE,
95 FRAME_TYPE_ERROR_RESPONSE: FLAGS_ERROR_RESPONSE, 107 FRAME_TYPE_ERROR_RESPONSE: FLAGS_ERROR_RESPONSE,
96 FRAME_TYPE_TEXT_OUTPUT: {}, 108 FRAME_TYPE_TEXT_OUTPUT: {},
109 FRAME_TYPE_STREAM_SETTINGS: {},
97 } 110 }
98 111
99 ARGUMENT_FRAME_HEADER = struct.Struct(r'<HH') 112 ARGUMENT_FRAME_HEADER = struct.Struct(r'<HH')
100 113
101 @attr.s(slots=True) 114 @attr.s(slots=True)
102 class frameheader(object): 115 class frameheader(object):
103 """Represents the data in a frame header.""" 116 """Represents the data in a frame header."""
104 117
105 length = attr.ib() 118 length = attr.ib()
106 requestid = attr.ib() 119 requestid = attr.ib()
120 streamid = attr.ib()
121 streamflags = attr.ib()
107 typeid = attr.ib() 122 typeid = attr.ib()
108 flags = attr.ib() 123 flags = attr.ib()
109 124
110 @attr.s(slots=True) 125 @attr.s(slots=True)
111 class frame(object): 126 class frame(object):
112 """Represents a parsed frame.""" 127 """Represents a parsed frame."""
113 128
114 requestid = attr.ib() 129 requestid = attr.ib()
130 streamid = attr.ib()
131 streamflags = attr.ib()
115 typeid = attr.ib() 132 typeid = attr.ib()
116 flags = attr.ib() 133 flags = attr.ib()
117 payload = attr.ib() 134 payload = attr.ib()
118 135
119 def makeframe(requestid, typeid, flags, payload): 136 def makeframe(requestid, streamid, streamflags, typeid, flags, payload):
120 """Assemble a frame into a byte array.""" 137 """Assemble a frame into a byte array."""
121 # TODO assert size of payload. 138 # TODO assert size of payload.
122 frame = bytearray(FRAME_HEADER_SIZE + len(payload)) 139 frame = bytearray(FRAME_HEADER_SIZE + len(payload))
123 140
124 # 24 bits length 141 # 24 bits length
125 # 16 bits request id 142 # 16 bits request id
143 # 8 bits stream id
144 # 8 bits stream flags
126 # 4 bits type 145 # 4 bits type
127 # 4 bits flags 146 # 4 bits flags
128 147
129 l = struct.pack(r'<I', len(payload)) 148 l = struct.pack(r'<I', len(payload))
130 frame[0:3] = l[0:3] 149 frame[0:3] = l[0:3]
131 struct.pack_into(r'<H', frame, 3, requestid) 150 struct.pack_into(r'<HBB', frame, 3, requestid, streamid, streamflags)
132 frame[5] = (typeid << 4) | flags 151 frame[7] = (typeid << 4) | flags
133 frame[6:] = payload 152 frame[8:] = payload
134 153
135 return frame 154 return frame
136 155
137 def makeframefromhumanstring(s): 156 def makeframefromhumanstring(s):
138 """Create a frame from a human readable string 157 """Create a frame from a human readable string
139 158
140 Strings have the form: 159 Strings have the form:
141 160
142 <request-id> <type> <flags> <payload> 161 <request-id> <stream-id> <stream-flags> <type> <flags> <payload>
143 162
144 This can be used by user-facing applications and tests for creating 163 This can be used by user-facing applications and tests for creating
145 frames easily without having to type out a bunch of constants. 164 frames easily without having to type out a bunch of constants.
146 165
147 Request ID is an integer. 166 Request ID and stream IDs are integers.
148 167
149 Frame type and flags can be specified by integer or named constant. 168 Stream flags, frame type, and flags can be specified by integer or
169 named constant.
150 170
151 Flags can be delimited by `|` to bitwise OR them together. 171 Flags can be delimited by `|` to bitwise OR them together.
152 """ 172 """
153 requestid, frametype, frameflags, payload = s.split(b' ', 3) 173 fields = s.split(b' ', 5)
174 requestid, streamid, streamflags, frametype, frameflags, payload = fields
154 175
155 requestid = int(requestid) 176 requestid = int(requestid)
177 streamid = int(streamid)
178
179 finalstreamflags = 0
180 for flag in streamflags.split(b'|'):
181 if flag in STREAM_FLAGS:
182 finalstreamflags |= STREAM_FLAGS[flag]
183 else:
184 finalstreamflags |= int(flag)
156 185
157 if frametype in FRAME_TYPES: 186 if frametype in FRAME_TYPES:
158 frametype = FRAME_TYPES[frametype] 187 frametype = FRAME_TYPES[frametype]
159 else: 188 else:
160 frametype = int(frametype) 189 frametype = int(frametype)
167 else: 196 else:
168 finalflags |= int(flag) 197 finalflags |= int(flag)
169 198
170 payload = stringutil.unescapestr(payload) 199 payload = stringutil.unescapestr(payload)
171 200
172 return makeframe(requestid=requestid, typeid=frametype, 201 return makeframe(requestid=requestid, streamid=streamid,
202 streamflags=finalstreamflags, typeid=frametype,
173 flags=finalflags, payload=payload) 203 flags=finalflags, payload=payload)
174 204
175 def parseheader(data): 205 def parseheader(data):
176 """Parse a unified framing protocol frame header from a buffer. 206 """Parse a unified framing protocol frame header from a buffer.
177 207
178 The header is expected to be in the buffer at offset 0 and the 208 The header is expected to be in the buffer at offset 0 and the
179 buffer is expected to be large enough to hold a full header. 209 buffer is expected to be large enough to hold a full header.
180 """ 210 """
181 # 24 bits payload length (little endian) 211 # 24 bits payload length (little endian)
212 # 16 bits request ID
213 # 8 bits stream ID
214 # 8 bits stream flags
182 # 4 bits frame type 215 # 4 bits frame type
183 # 4 bits frame flags 216 # 4 bits frame flags
184 # ... payload 217 # ... payload
185 framelength = data[0] + 256 * data[1] + 16384 * data[2] 218 framelength = data[0] + 256 * data[1] + 16384 * data[2]
186 requestid = struct.unpack_from(r'<H', data, 3)[0] 219 requestid, streamid, streamflags = struct.unpack_from(r'<HBB', data, 3)
187 typeflags = data[5] 220 typeflags = data[7]
188 221
189 frametype = (typeflags & 0xf0) >> 4 222 frametype = (typeflags & 0xf0) >> 4
190 frameflags = typeflags & 0x0f 223 frameflags = typeflags & 0x0f
191 224
192 return frameheader(framelength, requestid, frametype, frameflags) 225 return frameheader(framelength, requestid, streamid, streamflags,
226 frametype, frameflags)
193 227
194 def readframe(fh): 228 def readframe(fh):
195 """Read a unified framing protocol frame from a file object. 229 """Read a unified framing protocol frame from a file object.
196 230
197 Returns a 3-tuple of (type, flags, payload) for the decoded frame or 231 Returns a 3-tuple of (type, flags, payload) for the decoded frame or
214 payload = fh.read(h.length) 248 payload = fh.read(h.length)
215 if len(payload) != h.length: 249 if len(payload) != h.length:
216 raise error.Abort(_('frame length error: expected %d; got %d') % 250 raise error.Abort(_('frame length error: expected %d; got %d') %
217 (h.length, len(payload))) 251 (h.length, len(payload)))
218 252
219 return frame(h.requestid, h.typeid, h.flags, payload) 253 return frame(h.requestid, h.streamid, h.streamflags, h.typeid, h.flags,
254 payload)
220 255
221 def createcommandframes(stream, requestid, cmd, args, datafh=None): 256 def createcommandframes(stream, requestid, cmd, args, datafh=None):
222 """Create frames necessary to transmit a request to run a command. 257 """Create frames necessary to transmit a request to run a command.
223 258
224 This is a generator of bytearrays. Each item represents a frame 259 This is a generator of bytearrays. Each item represents a frame
396 payload=b''.join(atomchunks)) 431 payload=b''.join(atomchunks))
397 432
398 class stream(object): 433 class stream(object):
399 """Represents a logical unidirectional series of frames.""" 434 """Represents a logical unidirectional series of frames."""
400 435
436 def __init__(self, streamid, active=False):
437 self.streamid = streamid
438 self._active = False
439
401 def makeframe(self, requestid, typeid, flags, payload): 440 def makeframe(self, requestid, typeid, flags, payload):
402 """Create a frame to be sent out over this stream. 441 """Create a frame to be sent out over this stream.
403 442
404 Only returns the frame instance. Does not actually send it. 443 Only returns the frame instance. Does not actually send it.
405 """ 444 """
406 return makeframe(requestid, typeid, flags, payload) 445 streamflags = 0
446 if not self._active:
447 streamflags |= STREAM_FLAG_BEGIN_STREAM
448 self._active = True
449
450 return makeframe(requestid, self.streamid, streamflags, typeid, flags,
451 payload)
452
453 def ensureserverstream(stream):
454 if stream.streamid % 2:
455 raise error.ProgrammingError('server should only write to even '
456 'numbered streams; %d is not even' %
457 stream.streamid)
407 458
408 class serverreactor(object): 459 class serverreactor(object):
409 """Holds state of a server handling frame-based protocol requests. 460 """Holds state of a server handling frame-based protocol requests.
410 461
411 This class is the "brain" of the unified frame-based protocol server 462 This class is the "brain" of the unified frame-based protocol server
481 sender cannot receive until all data has been transmitted. 532 sender cannot receive until all data has been transmitted.
482 """ 533 """
483 self._deferoutput = deferoutput 534 self._deferoutput = deferoutput
484 self._state = 'idle' 535 self._state = 'idle'
485 self._bufferedframegens = [] 536 self._bufferedframegens = []
537 # stream id -> stream instance for all active streams from the client.
538 self._incomingstreams = {}
486 # request id -> dict of commands that are actively being received. 539 # request id -> dict of commands that are actively being received.
487 self._receivingcommands = {} 540 self._receivingcommands = {}
488 # Request IDs that have been received and are actively being processed. 541 # Request IDs that have been received and are actively being processed.
489 # Once all output for a request has been sent, it is removed from this 542 # Once all output for a request has been sent, it is removed from this
490 # set. 543 # set.
494 """Process a frame that has been received off the wire. 547 """Process a frame that has been received off the wire.
495 548
496 Returns a dict with an ``action`` key that details what action, 549 Returns a dict with an ``action`` key that details what action,
497 if any, the consumer should take next. 550 if any, the consumer should take next.
498 """ 551 """
552 if not frame.streamid % 2:
553 self._state = 'errored'
554 return self._makeerrorresult(
555 _('received frame with even numbered stream ID: %d') %
556 frame.streamid)
557
558 if frame.streamid not in self._incomingstreams:
559 if not frame.streamflags & STREAM_FLAG_BEGIN_STREAM:
560 self._state = 'errored'
561 return self._makeerrorresult(
562 _('received frame on unknown inactive stream without '
563 'beginning of stream flag set'))
564
565 self._incomingstreams[frame.streamid] = stream(frame.streamid)
566
567 if frame.streamflags & STREAM_FLAG_ENCODING_APPLIED:
568 # TODO handle decoding frames
569 self._state = 'errored'
570 raise error.ProgrammingError('support for decoding stream payloads '
571 'not yet implemented')
572
573 if frame.streamflags & STREAM_FLAG_END_STREAM:
574 del self._incomingstreams[frame.streamid]
575
499 handlers = { 576 handlers = {
500 'idle': self._onframeidle, 577 'idle': self._onframeidle,
501 'command-receiving': self._onframecommandreceiving, 578 'command-receiving': self._onframecommandreceiving,
502 'errored': self._onframeerrored, 579 'errored': self._onframeerrored,
503 } 580 }
511 def onbytesresponseready(self, stream, requestid, data): 588 def onbytesresponseready(self, stream, requestid, data):
512 """Signal that a bytes response is ready to be sent to the client. 589 """Signal that a bytes response is ready to be sent to the client.
513 590
514 The raw bytes response is passed as an argument. 591 The raw bytes response is passed as an argument.
515 """ 592 """
593 ensureserverstream(stream)
594
516 def sendframes(): 595 def sendframes():
517 for frame in createbytesresponseframesfrombytes(stream, requestid, 596 for frame in createbytesresponseframesfrombytes(stream, requestid,
518 data): 597 data):
519 yield frame 598 yield frame
520 599
550 return 'sendframes', { 629 return 'sendframes', {
551 'framegen': makegen(), 630 'framegen': makegen(),
552 } 631 }
553 632
554 def onapplicationerror(self, stream, requestid, msg): 633 def onapplicationerror(self, stream, requestid, msg):
634 ensureserverstream(stream)
635
555 return 'sendframes', { 636 return 'sendframes', {
556 'framegen': createerrorframe(stream, requestid, msg, 637 'framegen': createerrorframe(stream, requestid, msg,
557 application=True), 638 application=True),
558 } 639 }
559 640