| indygreg | |
| Alphare |
| hg-reviewers |
This help use to fix transaction safety on repos. See next changesets for details.
| No Linters Available |
| No Unit Test Coverage |
| Path | Packages | |||
|---|---|---|---|---|
| M | mercurial/configitems.py (1 line) | |||
| M | mercurial/localrepo.py (2 lines) | |||
| M | mercurial/revlog.py (47 lines) | |||
| M | mercurial/revlogutils/docket.py (32 lines) |
| Commit | Parents | Author | Summary | Date |
|---|---|---|---|---|
| ee298a1ff292 | c837bc05f18c | Pierre-Yves David | May 3 2021, 6:34 AM |
| ) | ) | ||||
| 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 | ||||
| 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] | ||||
| 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 | ||||
docket*, but will fix in flight, as with the first docket changeset