This is an archive of the discontinued Mercurial Phabricator instance.

Differential D706

bundle2: move processpart stream maintenance into part iterator
ClosedPublic

Authored by durham on Sep 13 2017, 8:38 PM.

Details

Reviewers
indygreg
Group Reviewers
hg-reviewers
Commits
rHGe71890f27767: bundle2: move processpart stream maintenance into part iterator
Summary

The processpart function also did some stream maintenance, so let's move it to
the part iterator as well, as part of moving all part iteration logic into the
class.

There is one place processpart is called outside of the normal loop, so we
manually handle the seek there.

The now-empty try/finally will be removed in a later patch, for ease of review.

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

durham created this revision.Sep 13 2017, 8:38 PM
indygreg requested changes to this revision.Sep 14 2017, 12:04 AM
indygreg added a subscriber: indygreg.
indygreg added inline comments.
mercurial/bundle2.py
1153–1162

I think there is a change in behavior here.

Before, _processpart() would catch SystemExit and KeyboardInterrupt and not attempt part.seek(). After this change, it always does the part.seek() courtesy of being in a finally block.

The problem with this is that we may attempt I/O after process termination is requested. This can lead to bad things, as the commit message for 9a9629b9416c details.

We mustn't attempt possibly blocking I/O after process termination is requested.

This revision now requires changes to proceed.Sep 14 2017, 12:04 AM
durham updated this revision to Diff 1818.Sep 14 2017, 1:21 PM
durham marked an inline comment as done.Sep 15 2017, 12:00 PM
Closed by commit rHGe71890f27767: bundle2: move processpart stream maintenance into part iterator (authored by durham). · Explain WhySep 18 2017, 4:40 PM
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

PathPackages
Mmercurial/bundle2.py (31 lines)

Diff 1872

mercurial/bundle2.py

class partiterator(object): class partiterator(object):
def __init__(self, repo, op, unbundler): def __init__(self, repo, op, unbundler):
self.repo = repo self.repo = repo
self.op = op self.op = op
self.unbundler = unbundler self.unbundler = unbundler
self.iterator = None self.iterator = None
self.count = 0 self.count = 0
self.current = None
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
yield p yield p
p.seek(0, 2)
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
if exc: if exc:
# If exiting or interrupted, do not attempt to seek the stream in
# the finally block below. This makes abort faster.
if (self.current and
not isinstance(exc, (SystemExit, KeyboardInterrupt))):
# consume the part content to not corrupt the stream.
self.current.seek(0, 2)
# 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:
for part in self.iterator: for part in self.iterator:
return ret return ret
def _processpart(op, part): def _processpart(op, part):
"""process a single part from a bundle """process a single part from a bundle
The part is guaranteed to have been fully consumed when the function exits The part is guaranteed to have been fully consumed when the function exits
(even if an exception is raised).""" (even if an exception is raised)."""
status = 'unknown' # used by debug output status = 'unknown' # used by debug output
hardabort = False
try: try:
try: try:
handler = parthandlermapping.get(part.type) handler = parthandlermapping.get(part.type)
if handler is None: if handler is None:
status = 'unsupported-type' status = 'unsupported-type'
raise error.BundleUnknownFeatureError(parttype=part.type) raise error.BundleUnknownFeatureError(parttype=part.type)
indebug(op.ui, 'found a handler for part %r' % part.type) indebug(op.ui, 'found a handler for part %r' % part.type)
unknownparams = part.mandatorykeys - handler.params unknownparams = part.mandatorykeys - handler.params
finally: finally:
if output is not None: if output is not None:
output = op.ui.popbuffer() output = op.ui.popbuffer()
if output: if output:
outpart = op.reply.newpart('output', data=output, outpart = op.reply.newpart('output', data=output,
mandatory=False) mandatory=False)
outpart.addparam( outpart.addparam(
'in-reply-to', pycompat.bytestr(part.id), mandatory=False) 'in-reply-to', pycompat.bytestr(part.id), mandatory=False)
# If exiting or interrupted, do not attempt to seek the stream in the
# finally block below. This makes abort faster.
except (SystemExit, KeyboardInterrupt):
hardabort = True
raise
finally: finally:
# consume the part content to not corrupt the stream. pass
if not hardabort:
part.seek(0, 2)
def decodecaps(blob): def decodecaps(blob):
"""decode a bundle2 caps bytes blob into a dictionary """decode a bundle2 caps bytes blob into a dictionary
The blob is a list of capabilities (one per line) The blob is a list of capabilities (one per line)
Capabilities may have values using a line of the form:: Capabilities may have values using a line of the form::
' opening out of band context\n') ' opening out of band context\n')
indebug(self.ui, 'bundle2 stream interruption, looking for a part.') indebug(self.ui, 'bundle2 stream interruption, looking for a part.')
headerblock = self._readpartheader() headerblock = self._readpartheader()
if headerblock is None: if headerblock is None:
indebug(self.ui, 'no part found during interruption.') indebug(self.ui, 'no part found during interruption.')
return return
part = unbundlepart(self.ui, headerblock, self._fp) part = unbundlepart(self.ui, headerblock, self._fp)
op = interruptoperation(self.ui) op = interruptoperation(self.ui)
hardabort = False
try:
_processpart(op, part) _processpart(op, part)
except (SystemExit, KeyboardInterrupt):
hardabort = True
raise
finally:
if not hardabort:
part.seek(0, 2)
self.ui.debug('bundle2-input-stream-interrupt:' self.ui.debug('bundle2-input-stream-interrupt:'
indygregUnsubmitted
Done

I think there is a change in behavior here.

Before, _processpart() would catch SystemExit and KeyboardInterrupt and not attempt part.seek(). After this change, it always does the part.seek() courtesy of being in a finally block.

The problem with this is that we may attempt I/O after process termination is requested. This can lead to bad things, as the commit message for 9a9629b9416c details.

We mustn't attempt possibly blocking I/O after process termination is requested.

indygreg: I think there is a change in behavior here. Before, _processpart() would catch SystemExit and…
' 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.
""" """