This is an archive of the discontinued Mercurial Phabricator instance.

Differential D9277

transaction: change list of journal entries into a dictionary
ClosedPublic

Authored by joerg.sonnenberger on Nov 7 2020, 4:32 PM.

Details

Reviewers
None
Group Reviewers
hg-reviewers
Commits
rHGa985c4fb23ca: transaction: change list of journal entries into a dictionary
rHG5b1214c16a39: transaction: change list of journal entries into a dictionary
Summary

The transaction object used to keep a mapping table of path names to
journal entries and a list of journal entries consisting of path and
file offset to truncate on rollback. The offsets are used in three
cases. repair.strip and rollback process all of them in one go, but they
care about the order. For them, it is perfectly reasonable to read the
journal back from disk as both operations already involve at least one
system call per journal entry. The other consumer is the revlog logic
for moving from inline to external data storage. It doesn't care about
the order of the journal and just needs to original offset stored.
Further optimisations are possible here to move the in-memory journal to
a set(), but without memoisation of the original revlog size this could
turn it into O(n^2) behavior in worst case when many revlogs need to
migrated.

Diff Detail

Repository
rHG Mercurial
Branch
default
Lint
No Linters Available
Unit
No Unit Test Coverage

Event Timeline

indygreg added a subscriber: indygreg.Nov 7 2020, 5:52 PM

I want a 2nd set of eyes on this.

I _think_ this may introduce a subtle bug regarding split revlogs. Before, we always recorded the offset of .i files before .d. After, because we sort, we'll truncate .d then .i. This leaves a window where a .i index references entries in a .d that don't exist. If a new reader comes along during that window, they'll read an index with no corresponding data. Unless we handle this case in the revlog reader code, I'd feel much better if we preserved insertion order and truncated .i before .d files so this race condition didn't exist.

joerg.sonnenberger added a comment.Nov 7 2020, 6:44 PM

I understand the concern and agree. I see two options of dealing with this without breaking the intention of the change here.
(1) Reverse the order. This would fix the immediate problem, but is more risky for similar classes of issues in other places, e.g. tree manifest chains.
(2) Adopt the logic from D9237 to read back the journal to recover the original order. Since both _playback and repair.strip will only ever be called once per transaction, it doesn't alter the performance much.

I'm inclined to go with (2).

indygreg added a comment.Nov 7 2020, 8:06 PM

Let's refactor the playback code to read from the journal file [handle]. That will preserve the order. There should be no meaningful performance implications here.

Once we do that, that frees you up to refactor the in-memory data structures, possibly even deleting them completely. I strongly prefer we do as much as possible in separate commits so we can bisect any performance regressions that may occur. I'm most worried about removal of the in-memory data structure regressing performance of large clones.

joerg.sonnenberger edited the summary of this revision. (Show Details)Nov 8 2020, 9:29 AM
joerg.sonnenberger updated this revision to Diff 23432.
joerg.sonnenberger added a comment.Nov 18 2020, 11:52 AM

The last update implemented the option to read to journal from disk, in case that wasn't clear from the description change.

joerg.sonnenberger added a commit: rHG5b1214c16a39: transaction: change list of journal entries into a dictionary.Nov 18 2020, 6:07 PM
This revision was not accepted when it landed; it landed in state Needs Review.
Closed by commit rHG5b1214c16a39: transaction: change list of journal entries into a dictionary (authored by joerg.sonnenberger). · Explain Why
This revision was automatically updated to reflect the committed changes.
mharbison72 added a subscriber: mharbison72.Dec 6 2020, 8:30 PM
mharbison72 added inline comments.
mercurial/transaction.py
403

This line crashes on Windows, using 60523483897c. I'm not sure why, since it implements __getattr__ and delegates to the internal file handle. Here's an example using hg split, though I have seen it in the test failures too.

hg split -r tip
(leaving bookmark @)
switching to topic tag-cache-display
0 files updated, 0 files merged, 0 files removed, 0 files unresolved
reverting hg
reverting hgext\phabricator.py
reverting mercurial\dispatch.py
reverting setup.py
starting interactive selection
** Unknown exception encountered with possibly-broken third-party extension "mercurial_keyring" (version N/A)
** which supports versions unknown of Mercurial.
** Please disable "mercurial_keyring" and try your action again.
** If that fixes the bug please report it to https://bitbucket.org/Mekk/mercurial_keyring/issues
** Python 2.7.17 (v2.7.17:c2f86d86e6, Oct 19 2019, 21:01:17) [MSC v.1500 64 bit (AMD64)]
** Mercurial Distributed SCM (version 5.6+222-60523483897c+20201201)
** Extensions loaded: absorb, blackbox, evolve 10.2.0.dev, extdiff, fix, mercurial_keyring, mq, phabblocker, phabricator, rebase, show, strip, topic 0.21.0.dev
Traceback (most recent call last):
  File "C:\Users\Matt\hg\hg", line 43, in <module>

  File "C:\Users\Matt\hg\mercurial\dispatch.py", line 115, in run
    def run():
  File "C:\Users\Matt\hg\mercurial\dispatch.py", line 266, in dispatch
    except error.Abort as inst:
  File "C:\Users\Matt\hg\mercurial\dispatch.py", line 442, in _runcatch
    ui.flush()
  File "C:\Users\Matt\hg\mercurial\dispatch.py", line 451, in _callcatch

  File "C:\Users\Matt\hg\mercurial\scmutil.py", line 155, in callcatch
    return func()
  File "C:\Users\Matt\hg\mercurial\dispatch.py", line 432, in _runcatchfunc
    b"%s debugger specified "
  File "C:\Users\Matt\hg\mercurial\dispatch.py", line 1229, in _dispatch
    ui.warn(_(b"warning: --repository ignored\n"))
  File "C:\Users\Matt\hg\mercurial\dispatch.py", line 883, in runcommand
    repo,
  File "C:\Users\Matt\hg\mercurial\dispatch.py", line 1240, in _runcommand
    if repo and repo != req.repo:
  File "C:\Users\Matt\hg\mercurial\dispatch.py", line 1226, in <lambda>
    repo = repo.unfiltered()
  File "C:\Users\Matt\hg\mercurial\util.py", line 1867, in check
    return func(*args, **kwargs)
  File "C:\Users\Matt\hg\mercurial\util.py", line 1867, in check
    return func(*args, **kwargs)
  File "C:\Users\Matt\hg\hgext\mq.py", line 4237, in mqcommand
    return orig(ui, repo, *args, **kwargs)
  File "C:\Users\Matt\hg\mercurial\util.py", line 1867, in check
    return func(*args, **kwargs)
  File "C:\Users\Matt\AppData\Roaming\Python\Python27\site-packages\hgext3rd\evolve\cmdrewrite.py", line 1300, in cmdsplit
    lockmod.release(tr, lock, wlock)
  File "C:\Users\Matt\hg\mercurial\lock.py", line 397, in release
    lock.release()
  File "C:\Users\Matt\hg\mercurial\transaction.py", line 462, in release
    self._abort()
  File "C:\Users\Matt\hg\mercurial\transaction.py", line 661, in _abort
    entries = self.readjournal()
  File "C:\Users\Matt\hg\mercurial\transaction.py", line 44, in _active
    return func(self, *args, **kwds)
  File "C:\Users\Matt\hg\mercurial\transaction.py", line 421, in readjournal
    for l in self._file:
TypeError: 'mixedfilemodewrapper' object is not iterable
joerg.sonnenberger added inline comments.Dec 6 2020, 8:54 PM
mercurial/transaction.py
403

Right, so the Windows wrapper doesn't implement line iteration. Can you try using for l in self._file.readlines(): instead? It's a bit wasteful, but as this isn't the fast path anyway, I don't think we care too much.

Revision Contents
Changeset List

PathPackages
Mmercurial/repair.py (9 lines)
Mmercurial/transaction.py (39 lines)

Diff 23432

mercurial/repair.py

with ui.uninterruptible(): with ui.uninterruptible():
try: try:
with repo.transaction(b"strip") as tr: with repo.transaction(b"strip") as tr:
# TODO this code violates the interface abstraction of the # TODO this code violates the interface abstraction of the
# transaction and makes assumptions that file storage is # transaction and makes assumptions that file storage is
# using append-only files. We'll need some kind of storage # using append-only files. We'll need some kind of storage
# API to handle stripping for us. # API to handle stripping for us.
offset = len(tr._entries) oldfiles = set(tr._offsetmap.keys())
tr.startgroup() tr.startgroup()
cl.strip(striprev, tr) cl.strip(striprev, tr)
stripmanifest(repo, striprev, tr, files) stripmanifest(repo, striprev, tr, files)
for fn in files: for fn in files:
repo.file(fn).strip(striprev, tr) repo.file(fn).strip(striprev, tr)
tr.endgroup() tr.endgroup()
for i in pycompat.xrange(offset, len(tr._entries)): entries = tr.readjournal()
file, troffset = tr._entries[i]
for file, troffset in entries:
if file in oldfiles:
continue
with repo.svfs(file, b'a', checkambig=True) as fp: with repo.svfs(file, b'a', checkambig=True) as fp:
fp.truncate(troffset) fp.truncate(troffset)
if troffset == 0: if troffset == 0:
repo.store.markremoved(file) repo.store.markremoved(file)
deleteobsmarkers(repo.obsstore, stripobsidx) deleteobsmarkers(repo.obsstore, stripobsidx)
del repo.obsstore del repo.obsstore
repo.invalidatevolatilesets() repo.invalidatevolatilesets()

mercurial/transaction.py

self._report = report self._report = report
# a vfs to the store content # a vfs to the store content
self._opener = opener self._opener = opener
# a map to access file in various {location -> vfs} # a map to access file in various {location -> vfs}
vfsmap = vfsmap.copy() vfsmap = vfsmap.copy()
vfsmap[b''] = opener # set default value vfsmap[b''] = opener # set default value
self._vfsmap = vfsmap self._vfsmap = vfsmap
self._after = after self._after = after
self._entries = [] self._offsetmap = {}
self._map = {}
self._journal = journalname self._journal = journalname
self._undoname = undoname self._undoname = undoname
self._queue = [] self._queue = []
# A callback to do something just after releasing transaction. # A callback to do something just after releasing transaction.
if releasefn is None: if releasefn is None:
releasefn = lambda tr, success: None releasefn = lambda tr, success: None
self._releasefn = releasefn self._releasefn = releasefn
self._checkambigfiles = set() self._checkambigfiles = set()
if checkambigfiles: if checkambigfiles:
self._checkambigfiles.update(checkambigfiles) self._checkambigfiles.update(checkambigfiles)
self._names = [name] self._names = [name]
# A dict dedicated to precisely tracking the changes introduced in the # A dict dedicated to precisely tracking the changes introduced in the
# transaction. # transaction.
self.changes = {} self.changes = {}
# a dict of arguments to be passed to hooks # a dict of arguments to be passed to hooks
self.hookargs = {} self.hookargs = {}
self._file = opener.open(self._journal, b"w") self._file = opener.open(self._journal, b"w+")
# a list of ('location', 'path', 'backuppath', cache) entries. # a list of ('location', 'path', 'backuppath', cache) entries.
# - if 'backuppath' is empty, no file existed at backup time # - if 'backuppath' is empty, no file existed at backup time
# - if 'path' is empty, this is a temporary transaction file # - if 'path' is empty, this is a temporary transaction file
# - if 'location' is not empty, the path is outside main opener reach. # - if 'location' is not empty, the path is outside main opener reach.
# use 'location' value as a key in a vfsmap to find the right 'vfs' # use 'location' value as a key in a vfsmap to find the right 'vfs'
# (cache is currently unused) # (cache is currently unused)
self._backupentries = [] self._backupentries = []
sees either none or all of the strip actions to be done.""" sees either none or all of the strip actions to be done."""
q = self._queue.pop() q = self._queue.pop()
for f, o in q: for f, o in q:
self._addentry(f, o) self._addentry(f, o)
@active @active
def add(self, file, offset): def add(self, file, offset):
"""record the state of an append-only file before update""" """record the state of an append-only file before update"""
if file in self._map or file in self._backupmap: if file in self._offsetmap or file in self._backupmap:
return return
if self._queue: if self._queue:
self._queue[-1].append((file, offset)) self._queue[-1].append((file, offset))
return return
self._addentry(file, offset) self._addentry(file, offset)
def _addentry(self, file, offset): def _addentry(self, file, offset):
"""add a append-only entry to memory and on-disk state""" """add a append-only entry to memory and on-disk state"""
if file in self._map or file in self._backupmap: if file in self._offsetmap or file in self._backupmap:
return return
self._entries.append((file, offset)) self._offsetmap[file] = offset
self._map[file] = len(self._entries) - 1
# add enough data to the journal to do the truncate # add enough data to the journal to do the truncate
self._file.write(b"%s\0%d\n" % (file, offset)) self._file.write(b"%s\0%d\n" % (file, offset))
self._file.flush() self._file.flush()
@active @active
def addbackup(self, file, hardlink=True, location=b''): def addbackup(self, file, hardlink=True, location=b''):
"""Adds a backup of the file to the transaction """Adds a backup of the file to the transaction
Calling addbackup() creates a hardlink backup of the specified file Calling addbackup() creates a hardlink backup of the specified file
that is used to recover the file in the event of the transaction that is used to recover the file in the event of the transaction
aborting. aborting.
* `file`: the file path, relative to .hg/store * `file`: the file path, relative to .hg/store
* `hardlink`: use a hardlink to quickly create the backup * `hardlink`: use a hardlink to quickly create the backup
""" """
if self._queue: if self._queue:
msg = b'cannot use transaction.addbackup inside "group"' msg = b'cannot use transaction.addbackup inside "group"'
raise error.ProgrammingError(msg) raise error.ProgrammingError(msg)
if file in self._map or file in self._backupmap: if file in self._offsetmap or file in self._backupmap:
return return
vfs = self._vfsmap[location] vfs = self._vfsmap[location]
dirname, filename = vfs.split(file) dirname, filename = vfs.split(file)
backupfilename = b"%s.backup.%s" % (self._journal, filename) backupfilename = b"%s.backup.%s" % (self._journal, filename)
backupfile = vfs.reljoin(dirname, backupfilename) backupfile = vfs.reljoin(dirname, backupfilename)
if vfs.exists(file): if vfs.exists(file):
filepath = vfs.join(file) filepath = vfs.join(file)
backuppath = vfs.join(backupfile) backuppath = vfs.join(backupfile)
del files[:] del files[:]
finally: finally:
for f in files: for f in files:
f.discard() f.discard()
return any return any
@active @active
def findoffset(self, file): def findoffset(self, file):
if file in self._map: return self._offsetmap.get(file)
return self._entries[self._map[file]][1]
return None @active
def readjournal(self):
self._file.seek(0)
entries = []
for l in self._file:
mharbison72Unsubmitted
Not Done

This line crashes on Windows, using 60523483897c. I'm not sure why, since it implements __getattr__ and delegates to the internal file handle. Here's an example using hg split, though I have seen it in the test failures too.

hg split -r tip
(leaving bookmark @)
switching to topic tag-cache-display
0 files updated, 0 files merged, 0 files removed, 0 files unresolved
reverting hg
reverting hgext\phabricator.py
reverting mercurial\dispatch.py
reverting setup.py
starting interactive selection
** Unknown exception encountered with possibly-broken third-party extension "mercurial_keyring" (version N/A)
** which supports versions unknown of Mercurial.
** Please disable "mercurial_keyring" and try your action again.
** If that fixes the bug please report it to https://bitbucket.org/Mekk/mercurial_keyring/issues
** Python 2.7.17 (v2.7.17:c2f86d86e6, Oct 19 2019, 21:01:17) [MSC v.1500 64 bit (AMD64)]
** Mercurial Distributed SCM (version 5.6+222-60523483897c+20201201)
** Extensions loaded: absorb, blackbox, evolve 10.2.0.dev, extdiff, fix, mercurial_keyring, mq, phabblocker, phabricator, rebase, show, strip, topic 0.21.0.dev
Traceback (most recent call last):
  File "C:\Users\Matt\hg\hg", line 43, in <module>

  File "C:\Users\Matt\hg\mercurial\dispatch.py", line 115, in run
    def run():
  File "C:\Users\Matt\hg\mercurial\dispatch.py", line 266, in dispatch
    except error.Abort as inst:
  File "C:\Users\Matt\hg\mercurial\dispatch.py", line 442, in _runcatch
    ui.flush()
  File "C:\Users\Matt\hg\mercurial\dispatch.py", line 451, in _callcatch

  File "C:\Users\Matt\hg\mercurial\scmutil.py", line 155, in callcatch
    return func()
  File "C:\Users\Matt\hg\mercurial\dispatch.py", line 432, in _runcatchfunc
    b"%s debugger specified "
  File "C:\Users\Matt\hg\mercurial\dispatch.py", line 1229, in _dispatch
    ui.warn(_(b"warning: --repository ignored\n"))
  File "C:\Users\Matt\hg\mercurial\dispatch.py", line 883, in runcommand
    repo,
  File "C:\Users\Matt\hg\mercurial\dispatch.py", line 1240, in _runcommand
    if repo and repo != req.repo:
  File "C:\Users\Matt\hg\mercurial\dispatch.py", line 1226, in <lambda>
    repo = repo.unfiltered()
  File "C:\Users\Matt\hg\mercurial\util.py", line 1867, in check
    return func(*args, **kwargs)
  File "C:\Users\Matt\hg\mercurial\util.py", line 1867, in check
    return func(*args, **kwargs)
  File "C:\Users\Matt\hg\hgext\mq.py", line 4237, in mqcommand
    return orig(ui, repo, *args, **kwargs)
  File "C:\Users\Matt\hg\mercurial\util.py", line 1867, in check
    return func(*args, **kwargs)
  File "C:\Users\Matt\AppData\Roaming\Python\Python27\site-packages\hgext3rd\evolve\cmdrewrite.py", line 1300, in cmdsplit
    lockmod.release(tr, lock, wlock)
  File "C:\Users\Matt\hg\mercurial\lock.py", line 397, in release
    lock.release()
  File "C:\Users\Matt\hg\mercurial\transaction.py", line 462, in release
    self._abort()
  File "C:\Users\Matt\hg\mercurial\transaction.py", line 661, in _abort
    entries = self.readjournal()
  File "C:\Users\Matt\hg\mercurial\transaction.py", line 44, in _active
    return func(self, *args, **kwds)
  File "C:\Users\Matt\hg\mercurial\transaction.py", line 421, in readjournal
    for l in self._file:
TypeError: 'mixedfilemodewrapper' object is not iterable
mharbison72: This line crashes on Windows, using 60523483897c. I'm not sure why, since it implements…
joerg.sonnenbergerAuthorUnsubmitted
Done

Right, so the Windows wrapper doesn't implement line iteration. Can you try using for l in self._file.readlines(): instead? It's a bit wasteful, but as this isn't the fast path anyway, I don't think we care too much.

joerg.sonnenberger: Right, so the Windows wrapper doesn't implement line iteration. Can you try using `for l in…
file, troffset = l.split(b'\0')
entries.append((file, int(troffset)))
return entries
@active @active
def replace(self, file, offset): def replace(self, file, offset):
''' '''
replace can only replace already committed entries replace can only replace already committed entries
that are not pending in the queue that are not pending in the queue
''' '''
if file not in self._map: if file not in self._offsetmap:
raise KeyError(file) raise KeyError(file)
index = self._map[file] self._offsetmap[file] = offset
self._entries[index] = (file, offset)
self._file.write(b"%s\0%d\n" % (file, offset)) self._file.write(b"%s\0%d\n" % (file, offset))
self._file.flush() self._file.flush()
@active @active
def nest(self, name='<unnamed>'): def nest(self, name='<unnamed>'):
self._count += 1 self._count += 1
self._usages += 1 self._usages += 1
self._names.append(name) self._names.append(name)
vfs.unlink(b) vfs.unlink(b)
except (IOError, OSError, error.Abort) as inst: except (IOError, OSError, error.Abort) as inst:
if not c: if not c:
raise raise
# Abort may be raise by read only opener # Abort may be raise by read only opener
self._report( self._report(
b"couldn't remove %s: %s\n" % (vfs.join(b), inst) b"couldn't remove %s: %s\n" % (vfs.join(b), inst)
) )
self._entries = [] self._offsetmap = {}
self._writeundo() self._writeundo()
if self._after: if self._after:
self._after() self._after()
self._after = None # Help prevent cycles. self._after = None # Help prevent cycles.
if self._opener.isfile(self._backupjournal): if self._opener.isfile(self._backupjournal):
self._opener.unlink(self._backupjournal) self._opener.unlink(self._backupjournal)
if self._opener.isfile(self._journal): if self._opener.isfile(self._journal):
self._opener.unlink(self._journal) self._opener.unlink(self._journal)
assert name.startswith(self._journal), name assert name.startswith(self._journal), name
uname = name.replace(self._journal, self._undoname, 1) uname = name.replace(self._journal, self._undoname, 1)
u = vfs.reljoin(base, uname) u = vfs.reljoin(base, uname)
util.copyfile(vfs.join(b), vfs.join(u), hardlink=True) util.copyfile(vfs.join(b), vfs.join(u), hardlink=True)
undobackupfile.write(b"%s\0%s\0%s\0%d\n" % (l, f, u, c)) undobackupfile.write(b"%s\0%s\0%s\0%d\n" % (l, f, u, c))
undobackupfile.close() undobackupfile.close()
def _abort(self): def _abort(self):
entries = self.readjournal()
self._count = 0 self._count = 0
self._usages = 0 self._usages = 0
self._file.close() self._file.close()
self._backupsfile.close() self._backupsfile.close()
try: try:
if not self._entries and not self._backupentries: if not self._offsetmap and not self._backupentries:
if self._backupjournal: if self._backupjournal:
self._opener.unlink(self._backupjournal) self._opener.unlink(self._backupjournal)
if self._journal: if self._journal:
self._opener.unlink(self._journal) self._opener.unlink(self._journal)
return return
self._report(_(b"transaction abort!\n")) self._report(_(b"transaction abort!\n"))
try: try:
for cat in sorted(self._abortcallback): for cat in sorted(self._abortcallback):
self._abortcallback[cat](self) self._abortcallback[cat](self)
# Prevent double usage and help clear cycles. # Prevent double usage and help clear cycles.
self._abortcallback = None self._abortcallback = None
_playback( _playback(
self._journal, self._journal,
self._report, self._report,
self._opener, self._opener,
self._vfsmap, self._vfsmap,
self._entries, entries,
self._backupentries, self._backupentries,
False, False,
checkambigfiles=self._checkambigfiles, checkambigfiles=self._checkambigfiles,
) )
self._report(_(b"rollback completed\n")) self._report(_(b"rollback completed\n"))
except BaseException as exc: except BaseException as exc:
self._report(_(b"rollback failed - please run hg recover\n")) self._report(_(b"rollback failed - please run hg recover\n"))
self._report( self._report(