This is an archive of the discontinued Mercurial Phabricator instance.

Differential D1388

bundle2: implement consume() API on unbundlepart
ClosedPublic

Authored by indygreg on Nov 13 2017, 11:59 PM.

Details

Reviewers
durin42
Group Reviewers
hg-reviewers
Commits
rHGdb5038525718: bundle2: implement consume() API on unbundlepart
Summary

We want bundle parts to not be seekable by default. That means
eliminating the generic seek() method.

A common pattern in bundle2.py is to seek to the end of the part
data. This is mainly used by the part iteration code to ensure
the underlying stream is advanced to the next bundle part.

In this commit, we establish a dedicated API for consuming a
bundle2 part data. We switch users of seek() to it.

The old implementation of seek(0, os.SEEK_END) would effectively
call self.read(). The new implementation calls self.read(32768)
in a loop. The old implementation would therefore assemble a
buffer to hold all remaining data being seeked over. For seeking
over large bundle parts, this would involve a large allocation and
a lot of overhead to collect intermediate data! This overhead can
be seen in the results for hg perfbundleread:

! bundle2 iterparts()
! wall 10.891305 comb 10.820000 user 7.990000 sys 2.830000 (best of 3)
! wall 8.070791 comb 8.060000 user 7.180000 sys 0.880000 (best of 3)
! bundle2 part seek()
! wall 12.991478 comb 10.390000 user 7.720000 sys 2.670000 (best of 3)
! wall 10.370142 comb 10.350000 user 7.430000 sys 2.920000 (best of 3)

Of course, skipping over large payload data isn't likely very common.
So I doubt the performance wins will be observed in the wild.

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.Nov 13 2017, 11:59 PM
durin42 accepted this revision.Nov 20 2017, 6:40 PM
This revision is now accepted and ready to land.Nov 20 2017, 6:40 PM
Closed by commit rHGdb5038525718: bundle2: implement consume() API on unbundlepart (authored by indygreg). · Explain WhyNov 20 2017, 6:49 PM
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

PathPackages
Mmercurial/bundle2.py (31 lines)

Diff 3692

mercurial/bundle2.py

def __enter__(self): def __enter__(self):
def func(): def func():
itr = enumerate(self.unbundler.iterparts()) itr = enumerate(self.unbundler.iterparts())
for count, p in itr: for count, p in itr:
self.count = count self.count = count
self.current = p self.current = p
yield p yield p
p.seek(0, os.SEEK_END) p.consume()
self.current = None self.current = None
self.iterator = func() self.iterator = func()
return self.iterator return self.iterator
def __exit__(self, type, exc, tb): def __exit__(self, type, exc, tb):
if not self.iterator: if not self.iterator:
return return
# Only gracefully abort in a normal exception situation. User aborts # Only gracefully abort in a normal exception situation. User aborts
# like Ctrl+C throw a KeyboardInterrupt which is not a base Exception, # like Ctrl+C throw a KeyboardInterrupt which is not a base Exception,
# and should not gracefully cleanup. # and should not gracefully cleanup.
if isinstance(exc, Exception): if isinstance(exc, Exception):
# Any exceptions seeking to the end of the bundle at this point are # Any exceptions seeking to the end of the bundle at this point are
# almost certainly related to the underlying stream being bad. # almost certainly related to the underlying stream being bad.
# And, chances are that the exception we're handling is related to # And, chances are that the exception we're handling is related to
# getting in that bad state. So, we swallow the seeking error and # getting in that bad state. So, we swallow the seeking error and
# re-raise the original error. # re-raise the original error.
seekerror = False seekerror = False
try: try:
if self.current: if self.current:
# consume the part content to not corrupt the stream. # consume the part content to not corrupt the stream.
self.current.seek(0, os.SEEK_END) self.current.consume()
for part in self.iterator: for part in self.iterator:
# consume the bundle content # consume the bundle content
part.seek(0, os.SEEK_END) part.consume()
except Exception: except Exception:
seekerror = True seekerror = True
# Small hack to let caller code distinguish exceptions from bundle2 # Small hack to let caller code distinguish exceptions from bundle2
# processing from processing the old format. This is mostly needed # processing from processing the old format. This is mostly needed
# to handle different return codes to unbundle according to the type # to handle different return codes to unbundle according to the type
# of bundle. We should probably clean up or drop this return code # of bundle. We should probably clean up or drop this return code
# craziness in a future version. # craziness in a future version.
self.params self.params
# From there, payload need to be decompressed # From there, payload need to be decompressed
self._fp = self._compengine.decompressorreader(self._fp) self._fp = self._compengine.decompressorreader(self._fp)
indebug(self.ui, 'start extraction of bundle2 parts') indebug(self.ui, 'start extraction of bundle2 parts')
headerblock = self._readpartheader() headerblock = self._readpartheader()
while headerblock is not None: while headerblock is not None:
part = seekableunbundlepart(self.ui, headerblock, self._fp) part = seekableunbundlepart(self.ui, headerblock, self._fp)
yield part yield part
# Seek to the end of the part to force it's consumption so the next # Ensure part is fully consumed so we can start reading the next
# part can be read. But then seek back to the beginning so the # part.
# code consuming this generator has a part that starts at 0. part.consume()
part.seek(0, os.SEEK_END) # But then seek back to the beginning so the code consuming this
# generator has a part that starts at 0.
part.seek(0, os.SEEK_SET) part.seek(0, os.SEEK_SET)
headerblock = self._readpartheader() headerblock = self._readpartheader()
indebug(self.ui, 'end of bundle2 stream') indebug(self.ui, 'end of bundle2 stream')
def _readpartheader(self): def _readpartheader(self):
"""reads a part header size and return the bytes blob """reads a part header size and return the bytes blob
returns None if empty""" returns None if empty"""
hardabort = False hardabort = False
try: try:
_processpart(op, part) _processpart(op, part)
except (SystemExit, KeyboardInterrupt): except (SystemExit, KeyboardInterrupt):
hardabort = True hardabort = True
raise raise
finally: finally:
if not hardabort: if not hardabort:
part.seek(0, os.SEEK_END) part.consume()
self.ui.debug('bundle2-input-stream-interrupt:' self.ui.debug('bundle2-input-stream-interrupt:'
' closing out of band context\n') ' closing out of band context\n')
class interruptoperation(object): class interruptoperation(object):
"""A limited operation to be use by part handler during interruption """A limited operation to be use by part handler during interruption
It only have access to an ui object. It only have access to an ui object.
""" """
self._payloadstream = util.chunkbuffer(self._payloadchunks()) self._payloadstream = util.chunkbuffer(self._payloadchunks())
# we read the data, tell it # we read the data, tell it
self._initialized = True self._initialized = True
def _payloadchunks(self): def _payloadchunks(self):
"""Generator of decoded chunks in the payload.""" """Generator of decoded chunks in the payload."""
return decodepayloadchunks(self.ui, self._fp) return decodepayloadchunks(self.ui, self._fp)
def consume(self):
"""Read the part payload until completion.
By consuming the part data, the underlying stream read offset will
be advanced to the next part (or end of stream).
"""
if self.consumed:
return
chunk = self.read(32768)
while chunk:
self._pos += len(chunk)
chunk = self.read(32768)
def read(self, size=None): def read(self, size=None):
"""read payload data""" """read payload data"""
if not self._initialized: if not self._initialized:
self._readheader() self._readheader()
if size is None: if size is None:
data = self._payloadstream.read() data = self._payloadstream.read()
else: else:
data = self._payloadstream.read(size) data = self._payloadstream.read(size)