( )⚙ D10628 revlogv2: track current index size in the docket

This is an archive of the discontinued Mercurial Phabricator instance.

Differential D10628

revlogv2: track current index size in the docket
ClosedPublic

Authored by marmoute on May 3 2021, 7:54 AM.

Details

Reviewers
indygreg
Alphare
Group Reviewers
hg-reviewers
Commits
rHG6597255a4f94: revlogv2: track current index size in the docket
Summary

This help use to fix transaction safety on repos. See next changesets for details.

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

marmoute created this revision.May 3 2021, 7:54 AM
baymax updated this revision to Diff 27471.May 3 2021, 9:32 PM

✅ refresh by Heptapod after a successful CI run (🐙 💚)

marmoute updated this revision to Diff 27544.May 4 2021, 10:13 AM
baymax updated this revision to Diff 27874.May 11 2021, 1:44 PM

✅ refresh by Heptapod after a successful CI run (🐙 💚)

Alphare added a subscriber: Alphare.May 17 2021, 10:36 AM
Alphare added inline comments.
mercurial/configitems.py
1160

docket*, but will fix in flight, as with the first docket changeset

Alphare accepted this revision.May 17 2021, 5:48 PM
This revision is now accepted and ready to land.May 17 2021, 5:48 PM
marmoute added a commit: rHG6597255a4f94: revlogv2: track current index size in the docket.May 17 2021, 5:51 PM
Closed by commit rHG6597255a4f94: revlogv2: track current index size in the docket (authored by marmoute). · Explain Why
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

Diff 28028

mercurial/configitems.py

# #
# * properly hide uncommitted content to other process # * properly hide uncommitted content to other process
# * expose transaction content hooks during pre-commit validation # * expose transaction content hooks during pre-commit validation
# * include management of a persistent nodemap in the main docket # * include management of a persistent nodemap in the main docket
# * enforce a "no-truncate" policy for mmap safety # * enforce a "no-truncate" policy for mmap safety
# - for censoring operation # - for censoring operation
# - for stripping operation # - for stripping operation
# - for rollback operation # - for rollback operation
# * proper streaming (race free) of the docket file
AlphareUnsubmitted
Not Done

docket*, but will fix in flight, as with the first docket changeset

Alphare: docket*, but will fix in flight, as with the first docket changeset
# * store the data size in the docket to simplify sidedata rewrite. # * store the data size in the docket to simplify sidedata rewrite.
# * track garbage data to evemtually allow rewriting -existing- sidedata. # * track garbage data to evemtually allow rewriting -existing- sidedata.
# * Exchange-wise, we will also need to do something more efficient than # * Exchange-wise, we will also need to do something more efficient than
# keeping references to the affected revlogs, especially memory-wise when # keeping references to the affected revlogs, especially memory-wise when
# rewriting sidedata. # rewriting sidedata.
# * sidedata compression # * sidedata compression
# * introduce a proper solution to reduce the number of filelog related files. # * introduce a proper solution to reduce the number of filelog related files.
# * Improvement to consider # * Improvement to consider

mercurial/localrepo.py

) )
hgvfs.createmode = store.createmode hgvfs.createmode = store.createmode
storevfs = store.vfs storevfs = store.vfs
storevfs.options = resolvestorevfsoptions(ui, requirements, features) storevfs.options = resolvestorevfsoptions(ui, requirements, features)
if requirementsmod.REVLOGV2_REQUIREMENT in requirements: if requirementsmod.REVLOGV2_REQUIREMENT in requirements:
features.add(repository.REPO_FEATURE_SIDE_DATA) features.add(repository.REPO_FEATURE_SIDE_DATA)
# the revlogv2 docket introduced race condition that we need to fix
features.discard(repository.REPO_FEATURE_STREAM_CLONE)
# The cache vfs is used to manage cache files. # The cache vfs is used to manage cache files.
cachevfs = vfsmod.vfs(cachepath, cacheaudited=True) cachevfs = vfsmod.vfs(cachepath, cacheaudited=True)
cachevfs.createmode = store.createmode cachevfs.createmode = store.createmode
# The cache vfs is used to manage cache files related to the working copy # The cache vfs is used to manage cache files related to the working copy
wcachevfs = vfsmod.vfs(wcachepath, cacheaudited=True) wcachevfs = vfsmod.vfs(wcachepath, cacheaudited=True)
wcachevfs.createmode = store.createmode wcachevfs.createmode = store.createmode

mercurial/revlog.py

elif self._chunkcachesize & (self._chunkcachesize - 1): elif self._chunkcachesize & (self._chunkcachesize - 1):
raise error.RevlogError( raise error.RevlogError(
_(b'revlog chunk cache size %r is not a power of 2') _(b'revlog chunk cache size %r is not a power of 2')
% self._chunkcachesize % self._chunkcachesize
) )
force_nodemap = opts.get(b'devel-force-nodemap', False) force_nodemap = opts.get(b'devel-force-nodemap', False)
return new_header, mmapindexthreshold, force_nodemap return new_header, mmapindexthreshold, force_nodemap
def _get_data(self, filepath, mmap_threshold): def _get_data(self, filepath, mmap_threshold, size=None):
"""return a file content with or without mmap """return a file content with or without mmap
If the file is missing return the empty string""" If the file is missing return the empty string"""
try: try:
with self.opener(filepath) as fp: with self.opener(filepath) as fp:
if mmap_threshold is not None: if mmap_threshold is not None:
file_size = self.opener.fstat(fp).st_size file_size = self.opener.fstat(fp).st_size
if file_size >= mmap_threshold: if file_size >= mmap_threshold:
if size is not None:
# avoid potentiel mmap crash
size = min(file_size, size)
# TODO: should .close() to release resources without # TODO: should .close() to release resources without
# relying on Python GC # relying on Python GC
if size is None:
return util.buffer(util.mmapread(fp)) return util.buffer(util.mmapread(fp))
else:
return util.buffer(util.mmapread(fp, size))
if size is None:
return fp.read() return fp.read()
else:
return fp.read(size)
except IOError as inst: except IOError as inst:
if inst.errno != errno.ENOENT: if inst.errno != errno.ENOENT:
raise raise
return b'' return b''
def _loadindex(self): def _loadindex(self):
new_header, mmapindexthreshold, force_nodemap = self._init_opts() new_header, mmapindexthreshold, force_nodemap = self._init_opts()
index_data = entry_data index_data = entry_data
else: else:
self._docket_file = entry_point self._docket_file = entry_point
if self._initempty: if self._initempty:
self._docket = docketutil.default_docket(self, header) self._docket = docketutil.default_docket(self, header)
else: else:
self._docket = docketutil.parse_docket(self, entry_data) self._docket = docketutil.parse_docket(self, entry_data)
self._indexfile = self._docket.index_filepath() self._indexfile = self._docket.index_filepath()
index_data = self._get_data(self._indexfile, mmapindexthreshold) index_data = b''
index_size = self._docket.index_end
if index_size > 0:
index_data = self._get_data(
self._indexfile, mmapindexthreshold, size=index_size
)
if len(index_data) < index_size:
msg = _(b'too few index data for %s: got %d, expected %d')
msg %= (self.display_id, len(index_data), index_size)
raise error.RevlogError(msg)
self._inline = False self._inline = False
# generaldelta implied by version 2 revlogs. # generaldelta implied by version 2 revlogs.
self._generaldelta = True self._generaldelta = True
# the logic for persistent nodemap will be dealt with within the # the logic for persistent nodemap will be dealt with within the
# main docket, so disable it for now. # main docket, so disable it for now.
self._nodemap_file = None self._nodemap_file = None
if self.postfix is None or self.postfix == b'a': if self.postfix is None or self.postfix == b'a':
return self.opener(self._indexfile, mode=b"r") return self.opener(self._indexfile, mode=b"r")
def __index_write_fp(self): def __index_write_fp(self):
# You should not use this directly and use `_writing` instead # You should not use this directly and use `_writing` instead
try: try:
f = self.opener( f = self.opener(
self._indexfile, mode=b"r+", checkambig=self._checkambig self._indexfile, mode=b"r+", checkambig=self._checkambig
) )
if self._docket is None:
f.seek(0, os.SEEK_END) f.seek(0, os.SEEK_END)
else:
f.seek(self._docket.index_end, os.SEEK_SET)
return f return f
except IOError as inst: except IOError as inst:
if inst.errno != errno.ENOENT: if inst.errno != errno.ENOENT:
raise raise
return self.opener( return self.opener(
self._indexfile, mode=b"w+", checkambig=self._checkambig self._indexfile, mode=b"w+", checkambig=self._checkambig
) )
self._inline = False self._inline = False
for i in self: for i in self:
e = self.index.entry_binary(i) e = self.index.entry_binary(i)
if i == 0 and self._docket is None: if i == 0 and self._docket is None:
header = self._format_flags | self._format_version header = self._format_flags | self._format_version
header = self.index.pack_header(header) header = self.index.pack_header(header)
e = header + e e = header + e
fp.write(e) fp.write(e)
if self._docket is not None:
self._docket.index_end = fp.tell()
# the temp file replace the real index when we exit the context # the temp file replace the real index when we exit the context
# manager # manager
tr.replace(self._indexfile, trindex * self.index.entry_size) tr.replace(self._indexfile, trindex * self.index.entry_size)
nodemaputil.setup_persistent_nodemap(tr, self) nodemaputil.setup_persistent_nodemap(tr, self)
self._chunkclear() self._chunkclear()
if existing_handles: if existing_handles:
# We work around this issue by inserting a seek() before writing. # We work around this issue by inserting a seek() before writing.
# Note: This is likely not necessary on Python 3. However, because # Note: This is likely not necessary on Python 3. However, because
# the file handle is reused for reads and may be seeked there, we need # the file handle is reused for reads and may be seeked there, we need
# to be careful before changing this. # to be careful before changing this.
if self._writinghandles is None: if self._writinghandles is None:
msg = b'adding revision outside `revlog._writing` context' msg = b'adding revision outside `revlog._writing` context'
raise error.ProgrammingError(msg) raise error.ProgrammingError(msg)
ifh, dfh = self._writinghandles ifh, dfh = self._writinghandles
if self._docket is None:
ifh.seek(0, os.SEEK_END) ifh.seek(0, os.SEEK_END)
else:
ifh.seek(self._docket.index_end, os.SEEK_SET)
if dfh: if dfh:
dfh.seek(0, os.SEEK_END) dfh.seek(0, os.SEEK_END)
curr = len(self) - 1 curr = len(self) - 1
if not self._inline: if not self._inline:
transaction.add(self._datafile, offset) transaction.add(self._datafile, offset)
transaction.add(self._indexfile, curr * len(entry)) transaction.add(self._indexfile, curr * len(entry))
if data[0]: if data[0]:
dfh.write(data[0]) dfh.write(data[0])
dfh.write(data[1]) dfh.write(data[1])
if sidedata: if sidedata:
dfh.write(sidedata) dfh.write(sidedata)
ifh.write(entry) ifh.write(entry)
else: else:
offset += curr * self.index.entry_size offset += curr * self.index.entry_size
transaction.add(self._indexfile, offset) transaction.add(self._indexfile, offset)
ifh.write(entry) ifh.write(entry)
ifh.write(data[0]) ifh.write(data[0])
ifh.write(data[1]) ifh.write(data[1])
if sidedata: if sidedata:
ifh.write(sidedata) ifh.write(sidedata)
self._enforceinlinesize(transaction) self._enforceinlinesize(transaction)
if self._docket is not None:
self._docket.index_end = self._writinghandles[0].tell()
nodemaputil.setup_persistent_nodemap(transaction, self) nodemaputil.setup_persistent_nodemap(transaction, self)
def addgroup( def addgroup(
self, self,
deltas, deltas,
linkmapper, linkmapper,
transaction, transaction,
alwayscache=False, alwayscache=False,
end = self.start(rev) end = self.start(rev)
if not self._inline: if not self._inline:
transaction.add(self._datafile, end) transaction.add(self._datafile, end)
end = rev * self.index.entry_size end = rev * self.index.entry_size
else: else:
end += rev * self.index.entry_size end += rev * self.index.entry_size
transaction.add(self._indexfile, end) transaction.add(self._indexfile, end)
if self._docket is not None:
# XXX we could, leverage the docket while stripping. However it is
# not powerfull enough at the time of this comment
self._docket.index_end = end
self._docket.write(transaction, stripping=True)
# then reset internal state in memory to forget those revisions # then reset internal state in memory to forget those revisions
self._revisioncache = None self._revisioncache = None
self._chaininfocache = util.lrucachedict(500) self._chaininfocache = util.lrucachedict(500)
self._chunkclear() self._chunkclear()
del self.index[rev:-1] del self.index[rev:-1]

mercurial/revlogutils/docket.py

constants, constants,
) )
# Docket format # Docket format
# #
# * 4 bytes: revlog version # * 4 bytes: revlog version
# | This is mandatory as docket must be compatible with the previous # | This is mandatory as docket must be compatible with the previous
# | revlog index header. # | revlog index header.
S_HEADER = struct.Struct(constants.INDEX_HEADER.format) # * 8 bytes: size of index data
S_HEADER = struct.Struct(constants.INDEX_HEADER.format + 'L')
class RevlogDocket(object): class RevlogDocket(object):
"""metadata associated with revlog""" """metadata associated with revlog"""
def __init__(self, revlog, version_header=None): def __init__(self, revlog, version_header=None, index_end=0):
self._version_header = version_header self._version_header = version_header
self._dirty = False self._dirty = False
self._radix = revlog.radix self._radix = revlog.radix
self._path = revlog._docket_file self._path = revlog._docket_file
self._opener = revlog.opener self._opener = revlog.opener
self._index_end = index_end
def index_filepath(self): def index_filepath(self):
"""file path to the current index file associated to this docket""" """file path to the current index file associated to this docket"""
# very simplistic version at first # very simplistic version at first
return b"%s.idx" % self._radix return b"%s.idx" % self._radix
def write(self, transaction): @property
def index_end(self):
return self._index_end
@index_end.setter
def index_end(self, new_size):
if new_size != self._index_end:
self._index_end = new_size
self._dirty = True
def write(self, transaction, stripping=False):
"""write the modification of disk if any """write the modification of disk if any
This make the new content visible to all process""" This make the new content visible to all process"""
if self._dirty: if self._dirty:
if not stripping:
# XXX we could, leverage the docket while stripping. However it
# is not powerfull enough at the time of this comment
transaction.addbackup(self._path, location=b'store') transaction.addbackup(self._path, location=b'store')
with self._opener(self._path, mode=b'w', atomictemp=True) as f: with self._opener(self._path, mode=b'w', atomictemp=True) as f:
f.write(self._serialize()) f.write(self._serialize())
self._dirty = False self._dirty = False
def _serialize(self): def _serialize(self):
return S_HEADER.pack(self._version_header) data = (
self._version_header,
self._index_end,
)
return S_HEADER.pack(*data)
def default_docket(revlog, version_header): def default_docket(revlog, version_header):
"""given a revlog version a new docket object for the given revlog""" """given a revlog version a new docket object for the given revlog"""
if (version_header & 0xFFFF) != constants.REVLOGV2: if (version_header & 0xFFFF) != constants.REVLOGV2:
return None return None
docket = RevlogDocket(revlog, version_header=version_header) docket = RevlogDocket(revlog, version_header=version_header)
docket._dirty = True docket._dirty = True
return docket return docket
def parse_docket(revlog, data): def parse_docket(revlog, data):
"""given some docket data return a docket object for the given revlog""" """given some docket data return a docket object for the given revlog"""
header = S_HEADER.unpack(data[: S_HEADER.size]) header = S_HEADER.unpack(data[: S_HEADER.size])
(version_header,) = header version_header, index_size = header
docket = RevlogDocket( docket = RevlogDocket(
revlog, revlog,
version_header=version_header, version_header=version_header,
index_end=index_size,
) )
return docket return docket