( )⚙ D10605 revlog: introduce a mandatory `_writing` context to update revlog content

This is an archive of the discontinued Mercurial Phabricator instance.

Differential D10605

revlog: introduce a mandatory `_writing` context to update revlog content
ClosedPublic

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

Details

Reviewers
indygreg
Alphare
Group Reviewers
hg-reviewers
Commits
rHG906a7bcaac86: revlog: introduce a mandatory `_writing` context to update revlog content
Summary

Before this change, various revlog methods where managing the opening and
closing of the revlog files manually and passing the file descriptor alors the
call path. To simplify the tracking of the write operation by a future docket,
we need something more organised. As a result, we introduce a revlog._writing
context manager that will wrap each revlog update operation. The file
descriptor are kept in the existing revlog._writinghandles parameter that
was already used by the addgroup logic.

All this change is internal to the revlog only, the "public" interface is not
affected. The addrevision and addgroup logic are still responsible for
setup up this context. However this new context give us multiple benefits:

  • all writer use a same, unified, logic,
  • this context is programmatically enforced,
  • each write "session" as a clearly identified start and end.

The post-pull sidedata update logic is still doing writing by end and will be
adjusted in a later changesets.

This change affect the concurrency checker test, because register the state of
the file in the transaction sooner in addrevision (about as early as what
addgroup would do), so the abort is rollbacking the other commit. I don't want
to weaken the current main logic.

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:52 AM
baymax updated this revision to Diff 27520.May 3 2021, 9:44 PM

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

baymax updated this revision to Diff 27851.May 11 2021, 1:42 PM

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

Alphare accepted this revision.May 17 2021, 5:58 AM
This revision is now accepted and ready to land.May 17 2021, 5:58 AM
baymax updated this revision to Diff 27977.May 17 2021, 7:02 AM

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

Revision Contents
Changeset List

Diff 28004

mercurial/changelog.py

self._divert = True self._divert = True
self.opener = _divertopener(self._realopener, self._indexfile) self.opener = _divertopener(self._realopener, self._indexfile)
if self._divert: if self._divert:
return True return True
return False return False
def _enforceinlinesize(self, tr, fp=None): def _enforceinlinesize(self, tr):
if not self._delayed: if not self._delayed:
revlog.revlog._enforceinlinesize(self, tr, fp) revlog.revlog._enforceinlinesize(self, tr)
def read(self, nodeorrev): def read(self, nodeorrev):
"""Obtain data from a parsed changelog revision. """Obtain data from a parsed changelog revision.
Returns a 6-tuple of: Returns a 6-tuple of:
- manifest node in binary - manifest node in binary
- author/user as a localstr - author/user as a localstr

mercurial/revlog.py

self._srmingapsize = 262144 self._srmingapsize = 262144
# Make copy of flag processors so each revlog instance can support # Make copy of flag processors so each revlog instance can support
# custom flags. # custom flags.
self._flagprocessors = dict(flagutil.flagprocessors) self._flagprocessors = dict(flagutil.flagprocessors)
# 2-tuple of file handles being used for active writing. # 2-tuple of file handles being used for active writing.
self._writinghandles = None self._writinghandles = None
# prevent nesting of addgroup
self._adding_group = None
self._loadindex() self._loadindex()
self._concurrencychecker = concurrencychecker self._concurrencychecker = concurrencychecker
def _init_opts(self): def _init_opts(self):
"""process options (from above/config) to setup associated default revlog mode """process options (from above/config) to setup associated default revlog mode
_(b"integrity check failed on %s:%s") _(b"integrity check failed on %s:%s")
% (self.display_id, pycompat.bytestr(revornode)) % (self.display_id, pycompat.bytestr(revornode))
) )
except error.RevlogError: except error.RevlogError:
if self._censorable and storageutil.iscensoredtext(text): if self._censorable and storageutil.iscensoredtext(text):
raise error.CensoredNodeError(self.display_id, node, text) raise error.CensoredNodeError(self.display_id, node, text)
raise raise
def _enforceinlinesize(self, tr, fp=None): def _enforceinlinesize(self, tr):
"""Check if the revlog is too big for inline and convert if so. """Check if the revlog is too big for inline and convert if so.
This should be called after revisions are added to the revlog. If the This should be called after revisions are added to the revlog. If the
revlog has grown too large to be an inline revlog, it will convert it revlog has grown too large to be an inline revlog, it will convert it
to use multiple index and data files. to use multiple index and data files.
""" """
tiprev = len(self) - 1 tiprev = len(self) - 1
total_size = self.start(tiprev) + self.length(tiprev) total_size = self.start(tiprev) + self.length(tiprev)
if not self._inline or total_size < _maxinline: if not self._inline or total_size < _maxinline:
return return
troffset = tr.findoffset(self._indexfile) troffset = tr.findoffset(self._indexfile)
if troffset is None: if troffset is None:
raise error.RevlogError( raise error.RevlogError(
_(b"%s not found in the transaction") % self._indexfile _(b"%s not found in the transaction") % self._indexfile
) )
trindex = 0 trindex = 0
tr.add(self._datafile, 0) tr.add(self._datafile, 0)
if fp: existing_handles = False
if self._writinghandles is not None:
existing_handles = True
fp = self._writinghandles[0]
fp.flush() fp.flush()
fp.close() fp.close()
# We can't use the cached file handle after close(). So prevent # We can't use the cached file handle after close(). So prevent
# its usage. # its usage.
self._writinghandles = None self._writinghandles = None
if True: new_dfh = self._datafp(b'w+')
with self._indexfp(b'r') as ifh, self._datafp(b'w') as dfh: new_dfh.truncate(0) # drop any potentially existing data
try:
with self._indexfp(b'r') as read_ifh:
for r in self: for r in self:
dfh.write(self._getsegmentforrevs(r, r, df=ifh)[1]) new_dfh.write(self._getsegmentforrevs(r, r, df=read_ifh)[1])
if troffset <= self.start(r): if troffset <= self.start(r):
trindex = r trindex = r
new_dfh.flush()
with self._indexfp(b'w') as fp: with self.opener(self._indexfile, mode=b'w', atomictemp=True) as fp:
self._format_flags &= ~FLAG_INLINE_DATA self._format_flags &= ~FLAG_INLINE_DATA
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: if i == 0:
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)
# 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:
# switched from inline to conventional reopen the index
ifh = self._indexfp(b"a+")
self._writinghandles = (ifh, new_dfh)
new_dfh = None
finally:
if new_dfh is not None:
new_dfh.close()
def _nodeduplicatecallback(self, transaction, node): def _nodeduplicatecallback(self, transaction, node):
"""called when trying to add a node already stored.""" """called when trying to add a node already stored."""
@contextlib.contextmanager
def _writing(self, transaction):
if self._writinghandles is not None:
yield
else:
r = len(self)
dsize = 0
if r:
dsize = self.end(r - 1)
dfh = None
if not self._inline:
dfh = self._datafp(b"a+")
transaction.add(self._datafile, dsize)
try:
isize = r * self.index.entry_size
ifh = self._indexfp(b"a+")
if self._inline:
transaction.add(self._indexfile, dsize + isize)
else:
transaction.add(self._indexfile, isize)
try:
self._writinghandles = (ifh, dfh)
try:
yield
finally:
self._writinghandles = None
finally:
ifh.close()
finally:
if dfh is not None:
dfh.close()
def addrevision( def addrevision(
self, self,
text, text,
transaction, transaction,
link, link,
p1, p1,
p2, p2,
cachedelta=None, cachedelta=None,
cachedelta=None, cachedelta=None,
deltacomputer=None, deltacomputer=None,
sidedata=None, sidedata=None,
): ):
"""add a raw revision with known flags, node and parents """add a raw revision with known flags, node and parents
useful when reusing a revision not stored in this revlog (ex: received useful when reusing a revision not stored in this revlog (ex: received
over wire, or read from an external bundle). over wire, or read from an external bundle).
""" """
dfh = None with self._writing(transaction):
if not self._inline:
dfh = self._datafp(b"a+")
ifh = self._indexfp(b"a+")
try:
return self._addrevision( return self._addrevision(
node, node,
rawtext, rawtext,
transaction, transaction,
link, link,
p1, p1,
p2, p2,
flags, flags,
cachedelta, cachedelta,
ifh,
dfh,
deltacomputer=deltacomputer, deltacomputer=deltacomputer,
sidedata=sidedata, sidedata=sidedata,
) )
finally:
if dfh:
dfh.close()
ifh.close()
def compress(self, data): def compress(self, data):
"""Generate a possibly-compressed representation of data.""" """Generate a possibly-compressed representation of data."""
if not data: if not data:
return b'', data return b'', data
compressed = self._compressor.compress(data) compressed = self._compressor.compress(data)
node, node,
rawtext, rawtext,
transaction, transaction,
link, link,
p1, p1,
p2, p2,
flags, flags,
cachedelta, cachedelta,
ifh,
dfh,
alwayscache=False, alwayscache=False,
deltacomputer=None, deltacomputer=None,
sidedata=None, sidedata=None,
): ):
"""internal function to add revisions to the log """internal function to add revisions to the log
see addrevision for argument descriptions. see addrevision for argument descriptions.
) )
if ( if (
node == self.nodeconstants.wdirid node == self.nodeconstants.wdirid
or node in self.nodeconstants.wdirfilenodeids or node in self.nodeconstants.wdirfilenodeids
): ):
raise error.RevlogError( raise error.RevlogError(
_(b"%s: attempt to add wdir revision") % self.display_id _(b"%s: attempt to add wdir revision") % self.display_id
) )
if self._writinghandles is None:
msg = b'adding revision outside `revlog._writing` context'
raise error.ProgrammingError(msg)
if self._inline: if self._inline:
fh = ifh fh = self._writinghandles[0]
else: else:
fh = dfh fh = self._writinghandles[1]
btext = [rawtext] btext = [rawtext]
curr = len(self) curr = len(self)
prev = curr - 1 prev = curr - 1
offset = self._get_data_offset(prev) offset = self._get_data_offset(prev)
if self._concurrencychecker: if self._concurrencychecker:
ifh, dfh = self._writinghandles
if self._inline: if self._inline:
# offset is "as if" it were in the .d file, so we need to add on # offset is "as if" it were in the .d file, so we need to add on
# the size of the entry metadata. # the size of the entry metadata.
self._concurrencychecker( self._concurrencychecker(
ifh, self._indexfile, offset + curr * self.index.entry_size ifh, self._indexfile, offset + curr * self.index.entry_size
) )
else: else:
# Entries in the .i are a consistent size. # Entries in the .i are a consistent size.
self.index.append(e) self.index.append(e)
entry = self.index.entry_binary(curr) entry = self.index.entry_binary(curr)
if curr == 0: if curr == 0:
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)
entry = header + entry entry = header + entry
self._writeentry( self._writeentry(
transaction, transaction,
ifh,
dfh,
entry, entry,
deltainfo.data, deltainfo.data,
link, link,
offset, offset,
serialized_sidedata, serialized_sidedata,
) )
rawtext = btext[0] rawtext = btext[0]
offset = 0 offset = 0
for rev, entry in enumerate(self.index): for rev, entry in enumerate(self.index):
sidedata_end = entry[8] + entry[9] sidedata_end = entry[8] + entry[9]
# Sidedata for a previous rev has potentially been written after # Sidedata for a previous rev has potentially been written after
# this rev's end, so take the max. # this rev's end, so take the max.
offset = max(self.end(rev), offset, sidedata_end) offset = max(self.end(rev), offset, sidedata_end)
return offset return offset
def _writeentry( def _writeentry(self, transaction, entry, data, link, offset, sidedata):
self, transaction, ifh, dfh, entry, data, link, offset, sidedata
):
# Files opened in a+ mode have inconsistent behavior on various # Files opened in a+ mode have inconsistent behavior on various
# platforms. Windows requires that a file positioning call be made # platforms. Windows requires that a file positioning call be made
# when the file handle transitions between reads and writes. See # when the file handle transitions between reads and writes. See
# 3686fa2b8eee and the mixedfilemodewrapper in windows.py. On other # 3686fa2b8eee and the mixedfilemodewrapper in windows.py. On other
# platforms, Python or the platform itself can be buggy. Some versions # platforms, Python or the platform itself can be buggy. Some versions
# of Solaris have been observed to not append at the end of the file # of Solaris have been observed to not append at the end of the file
# if the file was seeked to before the end. See issue4943 for more. # if the file was seeked to before the end. See issue4943 for more.
# #
# 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:
msg = b'adding revision outside `revlog._writing` context'
raise error.ProgrammingError(msg)
ifh, dfh = self._writinghandles
ifh.seek(0, os.SEEK_END) ifh.seek(0, os.SEEK_END)
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, ifh) self._enforceinlinesize(transaction)
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,
addrevisioncb=None, addrevisioncb=None,
duplicaterevisioncb=None, duplicaterevisioncb=None,
): ):
""" """
add a delta group add a delta group
given a set of deltas, add them to the revision log. the given a set of deltas, add them to the revision log. the
first delta is against its parent, which should be in our first delta is against its parent, which should be in our
log, the rest are against the previous delta. log, the rest are against the previous delta.
If ``addrevisioncb`` is defined, it will be called with arguments of If ``addrevisioncb`` is defined, it will be called with arguments of
this revlog and the node that was added. this revlog and the node that was added.
""" """
if self._writinghandles: if self._adding_group:
raise error.ProgrammingError(b'cannot nest addgroup() calls') raise error.ProgrammingError(b'cannot nest addgroup() calls')
r = len(self) self._adding_group = True
end = 0
if r:
end = self.end(r - 1)
ifh = self._indexfp(b"a+")
isize = r * self.index.entry_size
if self._inline:
transaction.add(self._indexfile, end + isize)
dfh = None
else:
transaction.add(self._indexfile, isize)
transaction.add(self._datafile, end)
dfh = self._datafp(b"a+")
self._writinghandles = (ifh, dfh)
empty = True empty = True
try: try:
if True: with self._writing(transaction):
deltacomputer = deltautil.deltacomputer(self) deltacomputer = deltautil.deltacomputer(self)
# loop through our set of deltas # loop through our set of deltas
for data in deltas: for data in deltas:
( (
node, node,
p1, p1,
p2, p2,
linknode, linknode,
node, node,
None, None,
transaction, transaction,
link, link,
p1, p1,
p2, p2,
flags, flags,
(baserev, delta), (baserev, delta),
ifh,
dfh,
alwayscache=alwayscache, alwayscache=alwayscache,
deltacomputer=deltacomputer, deltacomputer=deltacomputer,
sidedata=sidedata, sidedata=sidedata,
) )
if addrevisioncb: if addrevisioncb:
addrevisioncb(self, rev) addrevisioncb(self, rev)
empty = False empty = False
if not dfh and not self._inline:
# addrevision switched from inline to conventional
# reopen the index
ifh.close()
dfh = self._datafp(b"a+")
ifh = self._indexfp(b"a+")
self._writinghandles = (ifh, dfh)
finally: finally:
self._writinghandles = None self._adding_group = False
if dfh:
dfh.close()
ifh.close()
return not empty return not empty
def iscensored(self, rev): def iscensored(self, rev):
"""Check if a file revision is censored.""" """Check if a file revision is censored."""
if not self._censorable: if not self._censorable:
return False return False
return self.flags(rev) & REVIDX_ISCENSORED return self.flags(rev) & REVIDX_ISCENSORED
sidedata = self.sidedata(rev) sidedata = self.sidedata(rev)
if sidedata_helpers is not None: if sidedata_helpers is not None:
(sidedata, new_flags) = sidedatautil.run_sidedata_helpers( (sidedata, new_flags) = sidedatautil.run_sidedata_helpers(
self, sidedata_helpers, sidedata, rev self, sidedata_helpers, sidedata, rev
) )
flags = flags | new_flags[0] & ~new_flags[1] flags = flags | new_flags[0] & ~new_flags[1]
ifh = destrevlog.opener( with destrevlog._writing(tr):
destrevlog._indexfile, b'a+', checkambig=False
)
dfh = None
if not destrevlog._inline:
dfh = destrevlog.opener(destrevlog._datafile, b'a+')
try:
destrevlog._addrevision( destrevlog._addrevision(
node, node,
rawtext, rawtext,
tr, tr,
linkrev, linkrev,
p1, p1,
p2, p2,
flags, flags,
cachedelta, cachedelta,
ifh,
dfh,
deltacomputer=deltacomputer, deltacomputer=deltacomputer,
sidedata=sidedata, sidedata=sidedata,
) )
finally:
if dfh:
dfh.close()
ifh.close()
if addrevisioncb: if addrevisioncb:
addrevisioncb(self, rev, node) addrevisioncb(self, rev, node)
def censorrevision(self, tr, censornode, tombstone=b''): def censorrevision(self, tr, censornode, tombstone=b''):
if self._format_version == REVLOGV0: if self._format_version == REVLOGV0:
raise error.RevlogError( raise error.RevlogError(
_(b'cannot censor with version %d revlogs') _(b'cannot censor with version %d revlogs')

tests/test-racy-mutations.t

rollback completed rollback completed
note: commit message saved in .hg/last-message.txt note: commit message saved in .hg/last-message.txt
note: use 'hg commit --logfile .hg/last-message.txt --edit' to reuse it note: use 'hg commit --logfile .hg/last-message.txt --edit' to reuse it
abort: 00changelog.i: file cursor at position 249, expected 121 abort: 00changelog.i: file cursor at position 249, expected 121
And no corruption in the changelog. And no corruption in the changelog.
$ hg debugrevlogindex -c $ hg debugrevlogindex -c
rev linkrev nodeid p1 p2 rev linkrev nodeid p1 p2
0 0 222799e2f90b 000000000000 000000000000 0 0 222799e2f90b 000000000000 000000000000
1 1 6f124f6007a0 222799e2f90b 000000000000 1 1 6f124f6007a0 222799e2f90b 000000000000 (missing-correct-output !)
And, because of transactions, there's none in the manifestlog either. And, because of transactions, there's none in the manifestlog either.
$ hg debugrevlogindex -m $ hg debugrevlogindex -m
rev linkrev nodeid p1 p2 rev linkrev nodeid p1 p2
0 0 7b7020262a56 000000000000 000000000000 0 0 7b7020262a56 000000000000 000000000000
1 1 ad3fe36d86d9 7b7020262a56 000000000000 1 1 ad3fe36d86d9 7b7020262a56 000000000000
#endif #endif

tests/test-revlog-raw.py

) )
from mercurial.revlogutils import ( from mercurial.revlogutils import (
constants, constants,
deltas, deltas,
flagutil, flagutil,
) )
class _NoTransaction(object):
"""transaction like object to update the nodemap outside a transaction"""
def __init__(self):
self._postclose = {}
def addpostclose(self, callback_id, callback_func):
self._postclose[callback_id] = callback_func
def registertmp(self, *args, **kwargs):
pass
def addbackup(self, *args, **kwargs):
pass
def add(self, *args, **kwargs):
pass
def addabort(self, *args, **kwargs):
pass
def _report(self, *args):
pass
# TESTTMP is optional. This makes it convenient to run without run-tests.py # TESTTMP is optional. This makes it convenient to run without run-tests.py
tvfs = vfs.vfs(encoding.environ.get(b'TESTTMP', b'/tmp')) tvfs = vfs.vfs(encoding.environ.get(b'TESTTMP', b'/tmp'))
# Enable generaldelta otherwise revlog won't use delta as expected by the test # Enable generaldelta otherwise revlog won't use delta as expected by the test
tvfs.options = { tvfs.options = {
b'generaldelta': True, b'generaldelta': True,
b'revlogv1': True, b'revlogv1': True,
b'sparse-revlog': True, b'sparse-revlog': True,
p p
for p in range(r) for p in range(r)
if rlog.flags(p) & revlog.REVIDX_EXTSTORED == 0 if rlog.flags(p) & revlog.REVIDX_EXTSTORED == 0
] ]
) )
text = None text = None
cachedelta = (deltaparent, rlog.revdiff(deltaparent, r)) cachedelta = (deltaparent, rlog.revdiff(deltaparent, r))
flags = rlog.flags(r) flags = rlog.flags(r)
ifh = dfh = None with dlog._writing(_NoTransaction()):
try:
ifh = dlog.opener(dlog._indexfile, b'a+')
if not dlog._inline:
dfh = dlog.opener(dlog._datafile, b'a+')
dlog._addrevision( dlog._addrevision(
rlog.node(r), text, tr, r, p1, p2, flags, cachedelta, ifh, dfh rlog.node(r),
text,
tr,
r,
p1,
p2,
flags,
cachedelta,
) )
finally:
if dfh is not None:
dfh.close()
if ifh is not None:
ifh.close()
return dlog return dlog
# Utilities to generate revisions for testing # Utilities to generate revisions for testing
def genbits(n): def genbits(n):
"""Given a number n, generate (2 ** (n * 2) + 1) numbers in range(2 ** n). """Given a number n, generate (2 ** (n * 2) + 1) numbers in range(2 ** n).