( )⚙ D4473 wireprotoframing: buffer emitted data to reduce frame count

This is an archive of the discontinued Mercurial Phabricator instance.

Differential D4473

wireprotoframing: buffer emitted data to reduce frame count
ClosedPublic

Authored by indygreg on Sep 4 2018, 2:28 PM.

Details

Reviewers
None
Group Reviewers
hg-reviewers
Commits
rHG84bf6ded9317: wireprotoframing: buffer emitted data to reduce frame count
Summary

An upcoming commit introduces a wire protocol command that can emit
hundreds of thousands of small objects. Without a buffering layer,
we would emit a single, small frame for every object. Performance
profiling revealed this to be a source of significant overhead for
both client and server.

This commit introduces a very crude buffering layer so that we emit
fewer, bigger frames in such a scenario. This code will likely get
rewritten in the future to be part of the streams API, as we'll
need a similar strategy for compressing data. I don't want to think
about it too much at the moment though.

server
before: user 32.500+0.000 sys 1.160+0.000
after: user 20.230+0.010 sys 0.180+0.000

client
before: user 133.400+0.000 sys 93.120+0.000
after: user 68.370+0.000 sys 32.950+0.000

This appears to indicate we have significant overhead in the frame
processing code on both client and server. It might be worth profiling
that at some point...

Diff Detail

Repository
rHG Mercurial
Lint
Lint Skipped
Unit
Unit Tests Skipped

Event Timeline

indygreg created this revision.Sep 4 2018, 2:28 PM
Closed by commit rHG84bf6ded9317: wireprotoframing: buffer emitted data to reduce frame count (authored by indygreg). · Explain WhySep 12 2018, 10:43 AM
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

PathPackages
Mmercurial/wireprotoframing.py (105 lines)

Diff 10770

mercurial/wireprotoframing.py

if len(payload) > maxframesize: if len(payload) > maxframesize:
raise ValueError('cannot encode data in a single frame') raise ValueError('cannot encode data in a single frame')
yield stream.makeframe(requestid=requestid, yield stream.makeframe(requestid=requestid,
typeid=FRAME_TYPE_TEXT_OUTPUT, typeid=FRAME_TYPE_TEXT_OUTPUT,
flags=0, flags=0,
payload=payload) payload=payload)
class bufferingcommandresponseemitter(object):
"""Helper object to emit command response frames intelligently.
Raw command response data is likely emitted in chunks much smaller
than what can fit in a single frame. This class exists to buffer
chunks until enough data is available to fit in a single frame.
TODO we'll need something like this when compression is supported.
So it might make sense to implement this functionality at the stream
level.
"""
def __init__(self, stream, requestid, maxframesize=DEFAULT_MAX_FRAME_SIZE):
self._stream = stream
self._requestid = requestid
self._maxsize = maxframesize
self._chunks = []
self._chunkssize = 0
def send(self, data):
"""Send new data for emission.
Is a generator of new frames that were derived from the new input.
If the special input ``None`` is received, flushes all buffered
data to frames.
"""
if data is None:
for frame in self._flush():
yield frame
return
# There is a ton of potential to do more complicated things here.
# Our immediate goal is to coalesce small chunks into big frames,
# not achieve the fewest number of frames possible. So we go with
# a simple implementation:
#
# * If a chunk is too large for a frame, we flush and emit frames
# for the new chunk.
# * If a chunk can be buffered without total buffered size limits
# being exceeded, we do that.
# * If a chunk causes us to go over our buffering limit, we flush
# and then buffer the new chunk.
if len(data) > self._maxsize:
for frame in self._flush():
yield frame
# Now emit frames for the big chunk.
offset = 0
while True:
chunk = data[offset:offset + self._maxsize]
offset += len(chunk)
yield self._stream.makeframe(
self._requestid,
typeid=FRAME_TYPE_COMMAND_RESPONSE,
flags=FLAG_COMMAND_RESPONSE_CONTINUATION,
payload=chunk)
if offset == len(data):
return
# If we don't have enough to constitute a full frame, buffer and
# return.
if len(data) + self._chunkssize < self._maxsize:
self._chunks.append(data)
self._chunkssize += len(data)
return
# Else flush what we have and buffer the new chunk. We could do
# something more intelligent here, like break the chunk. Let's
# keep things simple for now.
for frame in self._flush():
yield frame
self._chunks.append(data)
self._chunkssize = len(data)
def _flush(self):
payload = b''.join(self._chunks)
assert len(payload) <= self._maxsize
self._chunks[:] = []
self._chunkssize = 0
yield self._stream.makeframe(
self._requestid,
typeid=FRAME_TYPE_COMMAND_RESPONSE,
flags=FLAG_COMMAND_RESPONSE_CONTINUATION,
payload=payload)
class stream(object): class stream(object):
"""Represents a logical unidirectional series of frames.""" """Represents a logical unidirectional series of frames."""
def __init__(self, streamid, active=False): def __init__(self, streamid, active=False):
self.streamid = streamid self.streamid = streamid
self._active = active self._active = active
def makeframe(self, requestid, typeid, flags, payload): def makeframe(self, requestid, typeid, flags, payload):
# functions. That is why we differentiate between exceptions raised # functions. That is why we differentiate between exceptions raised
# when iterating versus other exceptions that occur. # when iterating versus other exceptions that occur.
# #
# In all cases, when the function finishes, the request is fully # In all cases, when the function finishes, the request is fully
# handled and no new frames for it should be seen. # handled and no new frames for it should be seen.
def sendframes(): def sendframes():
emitted = False emitted = False
emitter = bufferingcommandresponseemitter(stream, requestid)
while True: while True:
try: try:
o = next(objs) o = next(objs)
except StopIteration: except StopIteration:
for frame in emitter.send(None):
yield frame
if emitted: if emitted:
yield createcommandresponseeosframe(stream, requestid) yield createcommandresponseeosframe(stream, requestid)
break break
except error.WireprotoCommandError as e: except error.WireprotoCommandError as e:
for frame in createcommanderrorresponse( for frame in createcommanderrorresponse(
stream, requestid, e.message, e.messageargs): stream, requestid, e.message, e.messageargs):
yield frame yield frame
break break
except Exception as e: except Exception as e:
for frame in createerrorframe(stream, requestid, for frame in createerrorframe(stream, requestid,
'%s' % e, '%s' % e,
errtype='server'): errtype='server'):
yield frame yield frame
break break
try: try:
if not emitted: if not emitted:
yield createcommandresponseokframe(stream, requestid) yield createcommandresponseokframe(stream, requestid)
emitted = True emitted = True
# TODO buffer chunks so emitted frame payloads can be for chunk in cborutil.streamencode(o):
# larger. for frame in emitter.send(chunk):
for frame in createbytesresponseframesfromgen(
stream, requestid, cborutil.streamencode(o)):
yield frame yield frame
except Exception as e: except Exception as e:
for frame in createerrorframe(stream, requestid, for frame in createerrorframe(stream, requestid,
'%s' % e, '%s' % e,
errtype='server'): errtype='server'):
yield frame yield frame
break break