transaction.py
512 lines
| 18.0 KiB
| text/x-python
|
PythonLexer
/ mercurial / transaction.py
|
r17424 | # transaction.py - simple journaling scheme for mercurial | ||
|
r0 | # | ||
# This transaction scheme is intended to gracefully handle program | ||||
# errors and interruptions. More serious failures like system crashes | ||||
# can be recovered with an fsck-like tool. As the whole repository is | ||||
# effectively log-structured, this should amount to simply truncating | ||||
# anything that isn't referenced in the changelog. | ||||
# | ||||
|
r2859 | # Copyright 2005, 2006 Matt Mackall <mpm@selenic.com> | ||
|
r0 | # | ||
|
r8225 | # This software may be used and distributed according to the terms of the | ||
|
r10263 | # GNU General Public License version 2 or any later version. | ||
|
r0 | |||
|
r3891 | from i18n import _ | ||
|
r20886 | import errno | ||
|
r20882 | import error, util | ||
|
r8289 | |||
|
r23313 | version = 2 | ||
|
r23064 | |||
|
r8289 | def active(func): | ||
def _active(self, *args, **kwds): | ||||
if self.count == 0: | ||||
raise error.Abort(_( | ||||
'cannot use transaction when it is already committed/aborted')) | ||||
return func(self, *args, **kwds) | ||||
return _active | ||||
|
r0 | |||
|
r23311 | def _playback(journal, report, opener, vfsmap, entries, backupentries, | ||
unlink=True): | ||||
|
r22204 | for f, o, _ignore in entries: | ||
|
r8294 | if o or not unlink: | ||
try: | ||||
|
r13400 | fp = opener(f, 'a') | ||
fp.truncate(o) | ||||
fp.close() | ||||
|
r9686 | except IOError: | ||
|
r8294 | report(_("failed to truncate %s\n") % f) | ||
raise | ||||
else: | ||||
try: | ||||
|
r20084 | opener.unlink(f) | ||
|
r9686 | except (IOError, OSError), inst: | ||
|
r8294 | if inst.errno != errno.ENOENT: | ||
raise | ||||
|
r20882 | |||
backupfiles = [] | ||||
|
r23309 | for l, f, b, c in backupentries: | ||
|
r23312 | if l not in vfsmap and c: | ||
report("couldn't handle %s: unknown cache location %s\n" | ||||
% (b, l)) | ||||
|
r23311 | vfs = vfsmap[l] | ||
|
r23312 | try: | ||
if f and b: | ||||
filepath = vfs.join(f) | ||||
backuppath = vfs.join(b) | ||||
try: | ||||
util.copyfile(backuppath, filepath) | ||||
backupfiles.append(b) | ||||
except IOError: | ||||
report(_("failed to recover %s\n") % f) | ||||
else: | ||||
target = f or b | ||||
try: | ||||
vfs.unlink(target) | ||||
except (IOError, OSError), inst: | ||||
if inst.errno != errno.ENOENT: | ||||
raise | ||||
except (IOError, OSError, util.Abort), inst: | ||||
if not c: | ||||
|
r23278 | raise | ||
|
r20882 | |||
|
r20087 | opener.unlink(journal) | ||
|
r20882 | backuppath = "%s.backupfiles" % journal | ||
if opener.exists(backuppath): | ||||
opener.unlink(backuppath) | ||||
|
r23312 | try: | ||
for f in backupfiles: | ||||
if opener.exists(f): | ||||
opener.unlink(f) | ||||
except (IOError, OSError, util.Abort), inst: | ||||
# only pure backup file remains, it is sage to ignore any error | ||||
pass | ||||
|
r8294 | |||
|
r1559 | class transaction(object): | ||
|
r23310 | def __init__(self, report, opener, vfsmap, journal, after=None, | ||
|
r23513 | createmode=None): | ||
|
r20881 | """Begin a new transaction | ||
Begins a new transaction that allows rolling back writes in the event of | ||||
an exception. | ||||
* `after`: called after the transaction has been committed | ||||
* `createmode`: the mode of the journal file that will be created | ||||
""" | ||||
|
r1806 | self.count = 1 | ||
|
r11230 | self.usages = 1 | ||
|
r582 | self.report = report | ||
|
r23310 | # a vfs to the store content | ||
|
r0 | self.opener = opener | ||
|
r23310 | # a map to access file in various {location -> vfs} | ||
vfsmap = vfsmap.copy() | ||||
vfsmap[''] = opener # set default value | ||||
self._vfsmap = vfsmap | ||||
|
r95 | self.after = after | ||
|
r0 | self.entries = [] | ||
|
r23249 | self.map = {} | ||
|
r23279 | self.journal = journal | ||
self._queue = [] | ||||
# a dict of arguments to be passed to hooks | ||||
self.hookargs = {} | ||||
self.file = opener.open(self.journal, "w") | ||||
|
r23309 | # a list of ('location', 'path', 'backuppath', cache) entries. | ||
|
r23311 | # - if 'backuppath' is empty, no file existed at backup time | ||
# - if 'path' is empty, this is a temporary transaction file | ||||
# - 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' | ||||
# (cache is currently unused) | ||||
|
r23249 | self._backupentries = [] | ||
self._backupmap = {} | ||||
self._backupjournal = "%s.backupfiles" % journal | ||||
self._backupsfile = opener.open(self._backupjournal, 'w') | ||||
self._backupsfile.write('%d\n' % version) | ||||
|
r23279 | |||
|
r6065 | if createmode is not None: | ||
|
r20087 | opener.chmod(self.journal, createmode & 0666) | ||
|
r23249 | opener.chmod(self._backupjournal, createmode & 0666) | ||
|
r0 | |||
|
r22078 | # hold file generations to be performed on commit | ||
self._filegenerators = {} | ||||
|
r23543 | # hold callback to write pending data for hooks | ||
|
r23202 | self._pendingcallback = {} | ||
# True is any pending data have been written ever | ||||
self._anypending = False | ||||
|
r23204 | # holds callback to call when writing the transaction | ||
self._finalizecallback = {} | ||||
|
r23543 | # hold callback for post transaction close | ||
|
r23220 | self._postclosecallback = {} | ||
|
r23764 | # holds callbacks to call during abort | ||
self._abortcallback = {} | ||||
|
r22078 | |||
|
r0 | def __del__(self): | ||
|
r558 | if self.journal: | ||
|
r9693 | self._abort() | ||
|
r0 | |||
|
r8289 | @active | ||
|
r8363 | def startgroup(self): | ||
|
r23250 | """delay registration of file entry | ||
This is used by strip to delay vision of strip offset. The transaction | ||||
sees either none or all of the strip actions to be done.""" | ||||
|
r23251 | self._queue.append([]) | ||
|
r8363 | |||
@active | ||||
def endgroup(self): | ||||
|
r23250 | """apply delayed registration of file entry. | ||
This is used by strip to delay vision of strip offset. The transaction | ||||
sees either none or all of the strip actions to be done.""" | ||||
|
r8363 | q = self._queue.pop() | ||
|
r23253 | for f, o, data in q: | ||
self._addentry(f, o, data) | ||||
|
r8363 | |||
@active | ||||
|
r2084 | def add(self, file, offset, data=None): | ||
|
r23252 | """record the state of an append-only file before update""" | ||
|
r23249 | if file in self.map or file in self._backupmap: | ||
|
r10282 | return | ||
|
r8363 | if self._queue: | ||
|
r23251 | self._queue[-1].append((file, offset, data)) | ||
|
r8363 | return | ||
|
r23253 | self._addentry(file, offset, data) | ||
def _addentry(self, file, offset, data): | ||||
"""add a append-only entry to memory and on-disk state""" | ||||
if file in self.map or file in self._backupmap: | ||||
return | ||||
|
r2084 | self.entries.append((file, offset, data)) | ||
self.map[file] = len(self.entries) - 1 | ||||
|
r0 | # add enough data to the journal to do the truncate | ||
self.file.write("%s\0%d\n" % (file, offset)) | ||||
self.file.flush() | ||||
|
r8289 | @active | ||
|
r23316 | def addbackup(self, file, hardlink=True, location=''): | ||
|
r20882 | """Adds a backup of the file to the transaction | ||
Calling addbackup() creates a hardlink backup of the specified file | ||||
that is used to recover the file in the event of the transaction | ||||
aborting. | ||||
* `file`: the file path, relative to .hg/store | ||||
* `hardlink`: use a hardlink to quickly create the backup | ||||
""" | ||||
|
r23251 | if self._queue: | ||
msg = 'cannot use transaction.addbackup inside "group"' | ||||
raise RuntimeError(msg) | ||||
|
r20882 | |||
|
r23249 | if file in self.map or file in self._backupmap: | ||
|
r20882 | return | ||
|
r23582 | vfs = self._vfsmap[location] | ||
dirname, filename = vfs.split(file) | ||||
|
r23315 | backupfilename = "%s.backup.%s" % (self.journal, filename) | ||
|
r23581 | backupfile = vfs.reljoin(dirname, backupfilename) | ||
|
r22663 | if vfs.exists(file): | ||
filepath = vfs.join(file) | ||||
|
r23314 | backuppath = vfs.join(backupfile) | ||
|
r20882 | util.copyfiles(filepath, backuppath, hardlink=hardlink) | ||
else: | ||||
|
r23278 | backupfile = '' | ||
|
r20882 | |||
|
r23316 | self._addbackupentry((location, file, backupfile, False)) | ||
|
r23283 | |||
def _addbackupentry(self, entry): | ||||
"""register a new backup entry and write it to disk""" | ||||
self._backupentries.append(entry) | ||||
|
r23249 | self._backupmap[file] = len(self._backupentries) - 1 | ||
|
r23309 | self._backupsfile.write("%s\0%s\0%s\0%d\n" % entry) | ||
|
r23249 | self._backupsfile.flush() | ||
|
r20882 | |||
@active | ||||
|
r23354 | def registertmp(self, tmpfile, location=''): | ||
|
r23291 | """register a temporary transaction file | ||
|
r23355 | Such files will be deleted when the transaction exits (on both | ||
failure and success). | ||||
|
r23291 | """ | ||
|
r23354 | self._addbackupentry((location, '', tmpfile, False)) | ||
|
r23291 | |||
@active | ||||
|
r23317 | def addfilegenerator(self, genid, filenames, genfunc, order=0, | ||
location=''): | ||||
|
r22078 | """add a function to generates some files at transaction commit | ||
The `genfunc` argument is a function capable of generating proper | ||||
content of each entry in the `filename` tuple. | ||||
At transaction close time, `genfunc` will be called with one file | ||||
object argument per entries in `filenames`. | ||||
The transaction itself is responsible for the backup, creation and | ||||
final write of such file. | ||||
The `genid` argument is used to ensure the same set of file is only | ||||
generated once. Call to `addfilegenerator` for a `genid` already | ||||
present will overwrite the old entry. | ||||
The `order` argument may be used to control the order in which multiple | ||||
generator will be executed. | ||||
|
r23317 | |||
The `location` arguments may be used to indicate the files are located | ||||
outside of the the standard directory for transaction. It should match | ||||
|
r23543 | one of the key of the `transaction.vfsmap` dictionary. | ||
|
r22078 | """ | ||
|
r22663 | # For now, we are unable to do proper backup and restore of custom vfs | ||
# but for bookmarks that are handled outside this mechanism. | ||||
|
r23317 | self._filegenerators[genid] = (order, filenames, genfunc, location) | ||
|
r22078 | |||
|
r23356 | def _generatefiles(self, suffix=''): | ||
|
r23102 | # write files registered for generation | ||
|
r23357 | any = False | ||
|
r23102 | for entry in sorted(self._filegenerators.values()): | ||
|
r23357 | any = True | ||
|
r23317 | order, filenames, genfunc, location = entry | ||
vfs = self._vfsmap[location] | ||||
|
r23102 | files = [] | ||
try: | ||||
for name in filenames: | ||||
|
r23356 | name += suffix | ||
if suffix: | ||||
self.registertmp(name, location=location) | ||||
else: | ||||
self.addbackup(name, location=location) | ||||
|
r23102 | files.append(vfs(name, 'w', atomictemp=True)) | ||
genfunc(*files) | ||||
finally: | ||||
for f in files: | ||||
f.close() | ||||
|
r23357 | return any | ||
|
r23102 | |||
|
r22078 | @active | ||
|
r2084 | def find(self, file): | ||
if file in self.map: | ||||
return self.entries[self.map[file]] | ||||
|
r23249 | if file in self._backupmap: | ||
return self._backupentries[self._backupmap[file]] | ||||
|
r2084 | return None | ||
|
r8289 | @active | ||
|
r2084 | def replace(self, file, offset, data=None): | ||
|
r8363 | ''' | ||
replace can only replace already committed entries | ||||
that are not pending in the queue | ||||
''' | ||||
|
r2084 | if file not in self.map: | ||
raise KeyError(file) | ||||
index = self.map[file] | ||||
self.entries[index] = (file, offset, data) | ||||
self.file.write("%s\0%d\n" % (file, offset)) | ||||
self.file.flush() | ||||
|
r8289 | @active | ||
|
r1806 | def nest(self): | ||
self.count += 1 | ||||
|
r11230 | self.usages += 1 | ||
|
r1806 | return self | ||
|
r11230 | def release(self): | ||
if self.count > 0: | ||||
self.usages -= 1 | ||||
|
r11685 | # if the transaction scopes are left without being closed, fail | ||
|
r11230 | if self.count > 0 and self.usages == 0: | ||
self._abort() | ||||
|
r1806 | def running(self): | ||
return self.count > 0 | ||||
|
r23202 | def addpending(self, category, callback): | ||
"""add a callback to be called when the transaction is pending | ||||
|
r23280 | The transaction will be given as callback's first argument. | ||
|
r23202 | Category is a unique identifier to allow overwriting an old callback | ||
with a newer callback. | ||||
""" | ||||
self._pendingcallback[category] = callback | ||||
@active | ||||
def writepending(self): | ||||
'''write pending file to temporary version | ||||
This is used to allow hooks to view a transaction before commit''' | ||||
categories = sorted(self._pendingcallback) | ||||
for cat in categories: | ||||
# remove callback since the data will have been flushed | ||||
|
r23280 | any = self._pendingcallback.pop(cat)(self) | ||
|
r23202 | self._anypending = self._anypending or any | ||
|
r23358 | self._anypending |= self._generatefiles(suffix='.pending') | ||
|
r23202 | return self._anypending | ||
|
r8289 | @active | ||
|
r23204 | def addfinalize(self, category, callback): | ||
"""add a callback to be called when the transaction is closed | ||||
|
r23281 | The transaction will be given as callback's first argument. | ||
|
r23204 | Category is a unique identifier to allow overwriting old callbacks with | ||
newer callbacks. | ||||
""" | ||||
self._finalizecallback[category] = callback | ||||
@active | ||||
|
r23220 | def addpostclose(self, category, callback): | ||
"""add a callback to be called after the transaction is closed | ||||
|
r23282 | The transaction will be given as callback's first argument. | ||
|
r23220 | Category is a unique identifier to allow overwriting an old callback | ||
with a newer callback. | ||||
""" | ||||
self._postclosecallback[category] = callback | ||||
@active | ||||
|
r23764 | def addabort(self, category, callback): | ||
"""add a callback to be called when the transaction is aborted. | ||||
The transaction will be given as the first argument to the callback. | ||||
Category is a unique identifier to allow overwriting an old callback | ||||
with a newer callback. | ||||
""" | ||||
self._abortcallback[category] = callback | ||||
@active | ||||
|
r0 | def close(self): | ||
|
r9220 | '''commit the transaction''' | ||
|
r23290 | if self.count == 1: | ||
|
r23103 | self._generatefiles() | ||
|
r23204 | categories = sorted(self._finalizecallback) | ||
for cat in categories: | ||||
|
r23281 | self._finalizecallback[cat](self) | ||
|
r20881 | |||
|
r1806 | self.count -= 1 | ||
if self.count != 0: | ||||
return | ||||
|
r0 | self.file.close() | ||
|
r23249 | self._backupsfile.close() | ||
|
r23291 | # cleanup temporary files | ||
|
r23312 | for l, f, b, c in self._backupentries: | ||
if l not in self._vfsmap and c: | ||||
self.report("couldn't remote %s: unknown cache location %s\n" | ||||
% (b, l)) | ||||
continue | ||||
|
r23311 | vfs = self._vfsmap[l] | ||
if not f and b and vfs.exists(b): | ||||
|
r23312 | try: | ||
vfs.unlink(b) | ||||
except (IOError, OSError, util.Abort), inst: | ||||
if not c: | ||||
raise | ||||
# Abort may be raise by read only opener | ||||
self.report("couldn't remote %s: %s\n" | ||||
% (vfs.join(b), inst)) | ||||
|
r0 | self.entries = [] | ||
|
r95 | if self.after: | ||
|
r785 | self.after() | ||
|
r20087 | if self.opener.isfile(self.journal): | ||
self.opener.unlink(self.journal) | ||||
|
r23249 | if self.opener.isfile(self._backupjournal): | ||
self.opener.unlink(self._backupjournal) | ||||
|
r23727 | for l, _f, b, c in self._backupentries: | ||
|
r23312 | if l not in self._vfsmap and c: | ||
self.report("couldn't remote %s: unknown cache location" | ||||
"%s\n" % (b, l)) | ||||
continue | ||||
|
r23311 | vfs = self._vfsmap[l] | ||
if b and vfs.exists(b): | ||||
|
r23312 | try: | ||
vfs.unlink(b) | ||||
except (IOError, OSError, util.Abort), inst: | ||||
if not c: | ||||
raise | ||||
# Abort may be raise by read only opener | ||||
self.report("couldn't remote %s: %s\n" | ||||
% (vfs.join(b), inst)) | ||||
|
r23249 | self._backupentries = [] | ||
|
r573 | self.journal = None | ||
|
r23220 | # run post close action | ||
categories = sorted(self._postclosecallback) | ||||
for cat in categories: | ||||
|
r23282 | self._postclosecallback[cat](self) | ||
|
r0 | |||
|
r8289 | @active | ||
|
r0 | def abort(self): | ||
|
r9220 | '''abort the transaction (generally called on error, or when the | ||
transaction is not explicitly committed before going out of | ||||
scope)''' | ||||
|
r8289 | self._abort() | ||
def _abort(self): | ||||
|
r8290 | self.count = 0 | ||
|
r11230 | self.usages = 0 | ||
|
r8290 | self.file.close() | ||
|
r23249 | self._backupsfile.close() | ||
|
r8290 | |||
|
r10228 | try: | ||
|
r23249 | if not self.entries and not self._backupentries: | ||
|
r10228 | if self.journal: | ||
|
r20087 | self.opener.unlink(self.journal) | ||
|
r23249 | if self._backupjournal: | ||
self.opener.unlink(self._backupjournal) | ||||
|
r10228 | return | ||
|
r0 | |||
|
r10228 | self.report(_("transaction abort!\n")) | ||
|
r0 | |||
|
r108 | try: | ||
|
r23764 | for cat in sorted(self._abortcallback): | ||
self._abortcallback[cat](self) | ||||
|
r23311 | _playback(self.journal, self.report, self.opener, self._vfsmap, | ||
|
r23249 | self.entries, self._backupentries, False) | ||
|
r8294 | self.report(_("rollback completed\n")) | ||
|
r16689 | except Exception: | ||
|
r8294 | self.report(_("rollback failed - please run hg recover\n")) | ||
finally: | ||||
self.journal = None | ||||
|
r8290 | |||
|
r23311 | def rollback(opener, vfsmap, file, report): | ||
|
r20882 | """Rolls back the transaction contained in the given file | ||
Reads the entries in the specified file, and the corresponding | ||||
'*.backupfiles' file, to recover from an incomplete transaction. | ||||
* `file`: a file containing a list of entries, specifying where | ||||
to truncate each file. The file should contain a list of | ||||
file\0offset pairs, delimited by newlines. The corresponding | ||||
'*.backupfiles' file should contain a list of file\0backupfile | ||||
pairs, delimited by \0. | ||||
""" | ||||
|
r8294 | entries = [] | ||
|
r20882 | backupentries = [] | ||
|
r8294 | |||
|
r20087 | fp = opener.open(file) | ||
|
r13400 | lines = fp.readlines() | ||
fp.close() | ||||
for l in lines: | ||||
|
r20524 | try: | ||
f, o = l.split('\0') | ||||
entries.append((f, int(o), None)) | ||||
except ValueError: | ||||
report(_("couldn't read journal entry %r!\n") % l) | ||||
|
r0 | |||
|
r20882 | backupjournal = "%s.backupfiles" % file | ||
if opener.exists(backupjournal): | ||||
fp = opener.open(backupjournal) | ||||
|
r23065 | lines = fp.readlines() | ||
if lines: | ||||
ver = lines[0][:-1] | ||||
|
r23064 | if ver == str(version): | ||
|
r23065 | for line in lines[1:]: | ||
if line: | ||||
# Shave off the trailing newline | ||||
line = line[:-1] | ||||
|
r23309 | l, f, b, c = line.split('\0') | ||
backupentries.append((l, f, b, bool(c))) | ||||
|
r23064 | else: | ||
|
r23309 | report(_("journal was created by a different version of " | ||
|
r23064 | "Mercurial")) | ||
|
r20882 | |||
|
r23311 | _playback(file, report, opener, vfsmap, entries, backupentries) | ||