( )⚙ D4854 cborutil: change buffering strategy

This is an archive of the discontinued Mercurial Phabricator instance.

Differential D4854

cborutil: change buffering strategy
ClosedPublic

Authored by indygreg on Oct 3 2018, 12:51 PM.

Details

Reviewers
None
Group Reviewers
hg-reviewers
Commits
rHG62160d3077cd: cborutil: change buffering strategy
Summary

Profiling revealed that we were spending a lot of time on the
line that was concatenating the old buffer with the incoming data
when attempting to decode long byte strings, such as manifest
revisions.

Essentially, we were feeding N chunks of size len(X) << len(Y) into
decode() and continuously allocating a new, larger buffer to hold
the undecoded input. This created substantial memory churn and
slowed down execution.

Changing the code to aggregate pending chunks in a list until we
have enough data to fully decode the next atom makes things much
more efficient.

I don't have exact data, but I recall the old code spending >1s
on manifest fulltexts from the mozilla-unified repo. The new code
doesn't significantly appear in profile output.

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 3 2018, 12:51 PM
Closed by commit rHG62160d3077cd: cborutil: change buffering strategy (authored by indygreg). · Explain WhyOct 3 2018, 8:07 PM
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

PathPackages
Mmercurial/utils/cborutil.py (36 lines)

Diff 11661

mercurial/utils/cborutil.py

layer. All input that isn't consumed by ``sansiodecoder`` will be buffered layer. All input that isn't consumed by ``sansiodecoder`` will be buffered
and concatenated with any new input that arrives later. and concatenated with any new input that arrives later.
TODO consider adding limits as to the maximum amount of data that can TODO consider adding limits as to the maximum amount of data that can
be buffered. be buffered.
""" """
def __init__(self): def __init__(self):
self._decoder = sansiodecoder() self._decoder = sansiodecoder()
self._leftover = None self._chunks = []
self._wanted = 0
def decode(self, b): def decode(self, b):
"""Attempt to decode bytes to CBOR values. """Attempt to decode bytes to CBOR values.
Returns a tuple with the following fields: Returns a tuple with the following fields:
* Bool indicating whether new values are available for retrieval. * Bool indicating whether new values are available for retrieval.
* Integer number of bytes decoded from the new input. * Integer number of bytes decoded from the new input.
* Integer number of bytes wanted to decode the next value. * Integer number of bytes wanted to decode the next value.
""" """
# Our strategy for buffering is to aggregate the incoming chunks in a
# list until we've received enough data to decode the next item.
# This is slightly more complicated than using an ``io.BytesIO``
# or continuously concatenating incoming data. However, because it
# isn't constantly reallocating backing memory for a growing buffer,
# it prevents excessive memory thrashing and is significantly faster,
# especially in cases where the percentage of input chunks that don't
# decode into a full item is high.
if self._chunks:
# A previous call said we needed N bytes to decode the next item.
# But this call doesn't provide enough data. We buffer the incoming
# chunk without attempting to decode.
if len(b) < self._wanted:
self._chunks.append(b)
self._wanted -= len(b)
return False, 0, self._wanted
# Else we may have enough data to decode the next item. Aggregate
# old data with new and reset the buffer.
newlen = len(b)
self._chunks.append(b)
b = b''.join(self._chunks)
self._chunks = []
oldlen = len(b) - newlen
if self._leftover:
oldlen = len(self._leftover)
b = self._leftover + b
self._leftover = None
else: else:
b = b
oldlen = 0 oldlen = 0
available, readcount, wanted = self._decoder.decode(b) available, readcount, wanted = self._decoder.decode(b)
self._wanted = wanted
if readcount < len(b): if readcount < len(b):
self._leftover = b[readcount:] self._chunks.append(b[readcount:])
return available, readcount - oldlen, wanted return available, readcount - oldlen, wanted
def getavailable(self): def getavailable(self):
return self._decoder.getavailable() return self._decoder.getavailable()
def decodeall(b): def decodeall(b):
"""Decode all CBOR items present in an iterable of bytes. """Decode all CBOR items present in an iterable of bytes.