This is an archive of the discontinued Mercurial Phabricator instance.

Differential D289

bundle2: seek part back during iteration
ClosedPublic

Authored by durham on Aug 8 2017, 8:05 PM.

Details

Reviewers
indygreg
Group Reviewers
hg-reviewers
Commits
rHG891118dcd279: bundle2: seek part back during iteration
Summary

Previously, iterparts would yield the part to users, then consume the part. This
changed the part after the user was given it and left it at the end, both of
which seem unexpected. Let's seek back to the beginning after we've consumed
it. I tried not seeking to the end at all, but that seems important for the
overall bundle2 consumption.

This is used in a future patch to let us move the bundlerepo
bundle2-changegroup-part to be handled entirely within the for loop, instead of
having to do a seek back to 0 after the entire loop finishes.

Diff Detail

Repository
rHG Mercurial
Lint
Lint Skipped
Unit
Unit Tests Skipped

Event Timeline

durham created this revision.Aug 8 2017, 8:05 PM
indygreg accepted this revision.Aug 14 2017, 9:46 PM
indygreg added a subscriber: indygreg.

This change feels a bit brittle to me, as it is relying on low-level implementation details of how bundle2 part handling and underlying stream consumption works. But I know from trying to touch this code that we have test coverage of it courtesy of bundlerepo. So I think we'll be fine.

This revision is now accepted and ready to land.Aug 14 2017, 9:46 PM
martinvonz added a subscriber: martinvonz.Aug 23 2017, 1:55 PM
martinvonz added inline comments.
mercurial/bundle2.py
834–835

Unrelated to this patch, but seems like we should define constants for "whence". I don't think seek(0, 2) is clear at all.

durham added a comment.Aug 23 2017, 3:08 PM

I agree it's brittle, but no more brittle than before, since we still had to seek back.

durham updated this revision to Diff 1216.Aug 23 2017, 3:35 PM
Closed by commit rHG891118dcd279: bundle2: seek part back during iteration (authored by durham). · Explain WhyAug 23 2017, 4:31 PM
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

PathPackages
Mmercurial/bundle2.py (4 lines)
Mmercurial/bundlerepo.py (1 line)

Diff 1216

mercurial/bundle2.py

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 = unbundlepart(self.ui, headerblock, self._fp) part = unbundlepart(self.ui, headerblock, self._fp)
yield part yield part
# Seek to the end of the part to force it's consumption so the next
# part can be read. But then seek back to the beginning so the
# code consuming this generator has a part that starts at 0.
part.seek(0, 2) part.seek(0, 2)
part.seek(0)
martinvonzUnsubmitted
Not Done

Unrelated to this patch, but seems like we should define constants for "whence". I don't think seek(0, 2) is clear at all.

martinvonz: Unrelated to this patch, but seems like we should define constants for "whence". I don't think…
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"""
headersize = self._unpack(_fpartheadersize)[0] headersize = self._unpack(_fpartheadersize)[0]

mercurial/bundlerepo.py

msg = _('Unsupported changegroup version: %s') msg = _('Unsupported changegroup version: %s')
raise error.Abort(msg % version) raise error.Abort(msg % version)
if self.bundle.compressed(): if self.bundle.compressed():
cgstream = self._writetempbundle(part.read, cgstream = self._writetempbundle(part.read,
".cg%sun" % version) ".cg%sun" % version)
if cgstream is None: if cgstream is None:
raise error.Abort(_('No changegroups found')) raise error.Abort(_('No changegroups found'))
cgstream.seek(0)
self.bundle = changegroup.getunbundler(version, cgstream, 'UN') self.bundle = changegroup.getunbundler(version, cgstream, 'UN')
elif self.bundle.compressed(): elif self.bundle.compressed():
f = self._writetempbundle(self.bundle.read, '.hg10un', f = self._writetempbundle(self.bundle.read, '.hg10un',
header='HG10UN') header='HG10UN')
self.bundlefile = self.bundle = exchange.readbundle(ui, f, self.bundlefile = self.bundle = exchange.readbundle(ui, f,
bundlename, bundlename,