This is an archive of the discontinued Mercurial Phabricator instance.

Differential D10778

changegroup: fix deltachunk API to be consistent from one class to another
ClosedPublic

Authored by marmoute on May 28 2021, 6:34 PM.

Details

Reviewers
Alphare
Group Reviewers
hg-reviewers
Commits
rHG3f00665bbea0: changegroup: fix deltachunk API to be consistent from one class to another
Summary

Depending of the subclass the 8th index of chunkdata items was either a
sidedata dict of a proto_flags integer. We have not fixed the inconsistency and
we already return fixed "delta" items from deltaiter.

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.

Revision Contents
Changeset List

Diff 28289

hgext/remotefilelog/shallowbundle.py

if not repo.shallowmatch(f): if not repo.shallowmatch(f):
fl = repo.file(f) fl = repo.file(f)
deltas = source.deltaiter() deltas = source.deltaiter()
fl.addgroup(deltas, revmap, trp) fl.addgroup(deltas, revmap, trp)
continue continue
chain = None chain = None
while True: while True:
# returns: (node, p1, p2, cs, deltabase, delta, flags) or None # returns: None or (
# node,
# p1,
# p2,
# cs,
# deltabase,
# delta,
# flags,
# sidedata,
# proto_flags
# )
revisiondata = source.deltachunk(chain) revisiondata = source.deltachunk(chain)
if not revisiondata: if not revisiondata:
break break
chain = revisiondata[0] chain = revisiondata[0]
revisiondatas[(f, chain)] = revisiondata revisiondatas[(f, chain)] = revisiondata
queue.append((f, chain)) queue.append((f, chain))
return True return True
skipcount = 0 skipcount = 0
# Prefetch the non-bundled revisions that we will need # Prefetch the non-bundled revisions that we will need
prefetchfiles = [] prefetchfiles = []
for f, node in queue: for f, node in queue:
revisiondata = revisiondatas[(f, node)] revisiondata = revisiondatas[(f, node)]
# revisiondata: (node, p1, p2, cs, deltabase, delta, flags) # revisiondata: (node, p1, p2, cs, deltabase, delta, flags, sdata, pfl)
dependents = [revisiondata[1], revisiondata[2], revisiondata[4]] dependents = [revisiondata[1], revisiondata[2], revisiondata[4]]
for dependent in dependents: for dependent in dependents:
if dependent == repo.nullid or (f, dependent) in revisiondatas: if dependent == repo.nullid or (f, dependent) in revisiondatas:
continue continue
prefetchfiles.append((f, hex(dependent))) prefetchfiles.append((f, hex(dependent)))
repo.fileservice.prefetch(prefetchfiles) repo.fileservice.prefetch(prefetchfiles)
# Apply the revisions in topological order such that a revision # Apply the revisions in topological order such that a revision
# is only written once it's deltabase and parents have been written. # is only written once it's deltabase and parents have been written.
while queue: while queue:
f, node = queue.pop(0) f, node = queue.pop(0)
if (f, node) in processed: if (f, node) in processed:
continue continue
skipcount += 1 skipcount += 1
if skipcount > len(queue) + 1: if skipcount > len(queue) + 1:
raise error.Abort(_(b"circular node dependency")) raise error.Abort(_(b"circular node dependency"))
fl = repo.file(f) fl = repo.file(f)
revisiondata = revisiondatas[(f, node)] revisiondata = revisiondatas[(f, node)]
# revisiondata: (node, p1, p2, cs, deltabase, delta, flags) # revisiondata: (node, p1, p2, cs, deltabase, delta, flags, sdata, pfl)
node, p1, p2, linknode, deltabase, delta, flags, sidedata = revisiondata (
node,
p1,
p2,
linknode,
deltabase,
delta,
flags,
sidedata,
proto_flags,
) = revisiondata
if not available(f, node, f, deltabase): if not available(f, node, f, deltabase):
continue continue
base = fl.rawdata(deltabase) base = fl.rawdata(deltabase)
text = mdiff.patch(base, delta) text = mdiff.patch(base, delta)
if not isinstance(text, bytes): if not isinstance(text, bytes):
text = bytes(text) text = bytes(text)

mercurial/changegroup.py

deltabase = p1 deltabase = p1
else: else:
deltabase = prevnode deltabase = prevnode
flags = 0 flags = 0
protocol_flags = 0 protocol_flags = 0
return node, p1, p2, deltabase, cs, flags, protocol_flags return node, p1, p2, deltabase, cs, flags, protocol_flags
def deltachunk(self, prevnode): def deltachunk(self, prevnode):
# Chunkdata: (node, p1, p2, cs, deltabase, delta, flags, sidedata, proto_flags)
l = self._chunklength() l = self._chunklength()
if not l: if not l:
return {} return {}
headerdata = readexactly(self._stream, self.deltaheadersize) headerdata = readexactly(self._stream, self.deltaheadersize)
header = self.deltaheader.unpack(headerdata) header = self.deltaheader.unpack(headerdata)
delta = readexactly(self._stream, l - self.deltaheadersize) delta = readexactly(self._stream, l - self.deltaheadersize)
header = self._deltaheader(header, prevnode) header = self._deltaheader(header, prevnode)
node, p1, p2, deltabase, cs, flags, protocol_flags = header node, p1, p2, deltabase, cs, flags, protocol_flags = header
return node, p1, p2, cs, deltabase, delta, flags, protocol_flags return node, p1, p2, cs, deltabase, delta, flags, {}, protocol_flags
def getchunks(self): def getchunks(self):
"""returns all the chunks contains in the bundle """returns all the chunks contains in the bundle
Used when you need to forward the binary stream to a file or another Used when you need to forward the binary stream to a file or another
network API. To do so, it parse the changegroup data, otherwise it will network API. To do so, it parse the changegroup data, otherwise it will
block in case of sshrepo because it don't know the end of the stream. block in case of sshrepo because it don't know the end of the stream.
""" """
def deltaiter(self): def deltaiter(self):
""" """
returns an iterator of the deltas in this changegroup returns an iterator of the deltas in this changegroup
Useful for passing to the underlying storage system to be stored. Useful for passing to the underlying storage system to be stored.
""" """
chain = None chain = None
for chunkdata in iter(lambda: self.deltachunk(chain), {}): for chunkdata in iter(lambda: self.deltachunk(chain), {}):
# Chunkdata: (node, p1, p2, cs, deltabase, delta, flags, sidedata) # Chunkdata: (node, p1, p2, cs, deltabase, delta, flags, sidedata, proto_flags)
yield chunkdata yield chunkdata[:8]
chain = chunkdata[0] chain = chunkdata[0]
class cg2unpacker(cg1unpacker): class cg2unpacker(cg1unpacker):
"""Unpacker for cg2 streams. """Unpacker for cg2 streams.
cg2 streams add support for generaldelta, so the delta header cg2 streams add support for generaldelta, so the delta header
format is slightly different. All other features about the data format is slightly different. All other features about the data
protocol_flags, node, p1, p2, deltabase, cs, flags = headertuple protocol_flags, node, p1, p2, deltabase, cs, flags = headertuple
return node, p1, p2, deltabase, cs, flags, protocol_flags return node, p1, p2, deltabase, cs, flags, protocol_flags
def deltachunk(self, prevnode): def deltachunk(self, prevnode):
res = super(cg4unpacker, self).deltachunk(prevnode) res = super(cg4unpacker, self).deltachunk(prevnode)
if not res: if not res:
return res return res
(node, p1, p2, cs, deltabase, delta, flags, protocol_flags) = res (
node,
p1,
p2,
cs,
deltabase,
delta,
flags,
sidedata,
protocol_flags,
) = res
assert not sidedata
sidedata = {} sidedata = {}
if protocol_flags & storageutil.CG_FLAG_SIDEDATA: if protocol_flags & storageutil.CG_FLAG_SIDEDATA:
sidedata_raw = getchunk(self._stream) sidedata_raw = getchunk(self._stream)
sidedata = sidedatamod.deserialize_sidedata(sidedata_raw) sidedata = sidedatamod.deserialize_sidedata(sidedata_raw)
return node, p1, p2, cs, deltabase, delta, flags, sidedata return (
node,
p1,
p2,
cs,
deltabase,
delta,
flags,
sidedata,
protocol_flags,
)
class headerlessfixup(object): class headerlessfixup(object):
def __init__(self, fh, h): def __init__(self, fh, h):
self._h = h self._h = h
self._fh = fh self._fh = fh
def read(self, n): def read(self, n):