mercurial-scm/hg-stable: mercurial/revlog.py comparison

comparison mercurial/revlog.py @ 47225:906a7bcaac86

revlog: introduce a mandatory `_writing` context to update revlog content 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. Differential Revision: https://phab.mercurial-scm.org/D10605

author	Pierre-Yves David <pierre-yves.david@octobus.net>
date	Mon, 03 May 2021 12:27:42 +0200
parents	100f061d88f6
children	042388bba644

comparison

equal deleted inserted replaced

-:100f061d88f6
+:906a7bcaac86
 # custom flags.
 self._flagprocessors = dict(flagutil.flagprocessors)
 # 2-tuple of file handles being used for active writing.
 self._writinghandles = None
+# prevent nesting of addgroup
+self._adding_group = None
 self._loadindex()
 self._concurrencychecker = concurrencychecker
 except error.RevlogError:
 if self._censorable and storageutil.iscensoredtext(text):
 raise error.CensoredNodeError(self.display_id, node, text)
 raise
-def _enforceinlinesize(self, tr, fp=None):
+def _enforceinlinesize(self, tr):
 """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
 revlog has grown too large to be an inline revlog, it will convert it
 to use multiple index and data files.
 _(b"%s not found in the transaction") % self._indexfile
 )
 trindex = 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.close()
 # We can't use the cached file handle after close(). So prevent
 # its usage.
 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:
-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):
 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._inline = False
 for i in self:
 e = self.index.entry_binary(i)
 if i == 0:
 header = self._format_flags | self._format_version
 header = self.index.pack_header(header)
 e = header + e
 fp.write(e)
 # the temp file replace the real index when we exit the context
 # manager
 tr.replace(self._indexfile, trindex * self.index.entry_size)
 nodemaputil.setup_persistent_nodemap(tr, self)
 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):
 """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(
 self,
 text,
 transaction,
 ):
 """add a raw revision with known flags, node and parents
 useful when reusing a revision not stored in this revlog (ex: received
 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(
 node,
 rawtext,
 transaction,
 link,
 p1,
 p2,
 flags,
 cachedelta,
-ifh,
-dfh,
 deltacomputer=deltacomputer,
 sidedata=sidedata,
 )
-finally:
-if dfh:
-dfh.close()
-ifh.close()
 def compress(self, data):
 """Generate a possibly-compressed representation of data."""
 if not data:
 return b'', data
 link,
 p1,
 p2,
 flags,
 cachedelta,
-ifh,
-dfh,
 alwayscache=False,
 deltacomputer=None,
 sidedata=None,
 ):
 """internal function to add revisions to the log
 or node in self.nodeconstants.wdirfilenodeids
 ):
 raise error.RevlogError(
 _(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:
-fh = ifh
+fh = self._writinghandles[0]
 else:
-fh = dfh
+fh = self._writinghandles[1]
 btext = [rawtext]
 curr = len(self)
 prev = curr - 1
 offset = self._get_data_offset(prev)
 if self._concurrencychecker:
+ifh, dfh = self._writinghandles
 if self._inline:
 # offset is "as if" it were in the .d file, so we need to add on
 # the size of the entry metadata.
 self._concurrencychecker(
 ifh, self._indexfile, offset + curr * self.index.entry_size
 header = self._format_flags | self._format_version
 header = self.index.pack_header(header)
 entry = header + entry
 self._writeentry(
 transaction,
-ifh,
-dfh,
 entry,
 deltainfo.data,
 link,
 offset,
 serialized_sidedata,
 # Sidedata for a previous rev has potentially been written after
 # this rev's end, so take the max.
 offset = max(self.end(rev), offset, sidedata_end)
 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
 # platforms. Windows requires that a file positioning call be made
 # when the file handle transitions between reads and writes. See
 # 3686fa2b8eee and the mixedfilemodewrapper in windows.py. On other
 # platforms, Python or the platform itself can be buggy. Some versions
 #
 # We work around this issue by inserting a seek() before writing.
 # 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
 # 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)
 if dfh:
 dfh.seek(0, os.SEEK_END)
 curr = len(self) - 1
 ifh.write(entry)
 ifh.write(data[0])
 ifh.write(data[1])
 if sidedata:
 ifh.write(sidedata)
-self._enforceinlinesize(transaction, ifh)
+self._enforceinlinesize(transaction)
 nodemaputil.setup_persistent_nodemap(transaction, self)
 def addgroup(
 self,
 deltas,
 If ``addrevisioncb`` is defined, it will be called with arguments of
 this revlog and the node that was added.
 """
-if self._writinghandles:
+if self._adding_group:
 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
 try:
-if True:
+with self._writing(transaction):
 deltacomputer = deltautil.deltacomputer(self)
 # loop through our set of deltas
 for data in deltas:
 (
 node,
 link,
 p1,
 p2,
 flags,
 (baserev, delta),
-ifh,
-dfh,
 alwayscache=alwayscache,
 deltacomputer=deltacomputer,
 sidedata=sidedata,
 )
 if addrevisioncb:
 addrevisioncb(self, rev)
 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:
-self._writinghandles = None
+self._adding_group = False
-if dfh:
-dfh.close()
-ifh.close()
 return not empty
 def iscensored(self, rev):
 """Check if a file revision is censored."""
 if not self._censorable:
 (sidedata, new_flags) = sidedatautil.run_sidedata_helpers(
 self, sidedata_helpers, sidedata, rev
 )
 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(
 node,
 rawtext,
 tr,
 linkrev,
 p1,
 p2,
 flags,
 cachedelta,
-ifh,
-dfh,
 deltacomputer=deltacomputer,
 sidedata=sidedata,
 )
-finally:
-if dfh:
-dfh.close()
-ifh.close()
 if addrevisioncb:
 addrevisioncb(self, rev, node)
 def censorrevision(self, tr, censornode, tombstone=b''):

Mercurial > public > mercurial-scm > hg-stable

comparison mercurial/revlog.py @ 47225:906a7bcaac86