Changeset 614 in tailor

Timestamp:

08/17/05 19:49:32 (8 years ago)

Author:

lele@…

Hash name:

20050817174932-97f81-ed13a41218cda9683aafa49551b6432786b3e1a1

Message:

Make the state file an interator, and use a journal to avoid rewrite

Location:

vcpx

Files:

• changes.py (modified) (1 diff)
• cvsps.py (modified) (3 diffs)
• source.py (modified) (5 diffs)
• tests/__init__.py (modified) (1 diff)
• project.py (modified) (2 diffs)
• tests/statefile.py (added)

vcpx/changes.py

@@ -117 +117 @@
         self.entries = entries or []
         self.unidiff = None        # This is the unidiff of the whole changeset
+
+    def __eq__(self, other):
+        return (self.revision == other.revision and
+                self.date == other.date and
+                self.author == other.author)
+
+    def __ne__(self, other):
+        return (self.revision <> other.revision or
+                self.date <> other.date or
+                self.author <> other.author)
 
     def addEntry(self, entry, revision):

vcpx/cvsps.py

@@ -285 +285 @@
 
         if timestamp == 'INITIAL':
-            timestamp = csets[0].date.isoformat(sep=' ')
+            cset = csets.next()
+            timestamp = cset.date.isoformat(sep=' ')
+        else:
+            cset = None
 
         if not exists(join(self.basedir, 'CVS')):
@@ -320 +323 @@
 
         found = False
-        while csets:
-            cset = csets.pop(0)
+        if cset is None:
+            try:
+                cset = csets.next()
+            except StopIteration:
+                cset = None
+        while cset is not None:
             for m in cset.entries:
                 info = entries.getFileInfo(m.name)
@@ -333 +340 @@
                 last = cset
                 break
+
+            try:
+                cset = csets.next()
+            except StopIteration:
+                cset = None
 
         if not found:

vcpx/source.py

@@ -83 +83 @@
         conflicts = []
 
-        if not self.pending:
-            return last, conflicts
-
         try:
-            while self.pending:
-                c = self.pending[0]
-
+            for c in self.state_file:
                 # Give the opportunity to subclasses to stop the application
                 # of the queue, before the application of the patch by the
@@ -123 +118 @@
                         replay(c)
 
-                # Remove it from the queue and remember it for the finally
-                # clause
-                last = self.pending.pop(0)
+                # Remember it for the finally clause and notify the state
+                # file so that it gets removed from the queue
+                last = c
+                self.state_file.applied()
 
                 # Another hook (last==c here)
@@ -132 +128 @@
         finally:
             # For whatever reason we exit the loop, save the last state
-            if last is not None:
-                self.state_file.write(last.revision, self.pending)
+            self.state_file.finalize()
 
         return last, conflicts
@@ -181 +176 @@
         """
         Load the pending changesets from the state file, or query the
-        upstream repository if there's none.
-        """
-
-        revision, self.pending = self.state_file.load()
-        if not self.pending:
-            self.pending = self._getUpstreamChangesets(revision or sincerev)
-        return self.pending
+        upstream repository if there's none. Return an iterator over
+        pending changesets.
+        """
+
+        pending = len(self.state_file)
+        if pending == 0:
+            last = self.state_file.lastAppliedChangeset()
+            if last:
+                revision = last.revision
+            else:
+                revision = sincerev
+            changesets = self._getUpstreamChangesets(revision)
+            self.state_file.setPendingChangesets(changesets)
+        return self.state_file
 
     def _getUpstreamChangesets(self, sincerev):
@@ -219 +221 @@
 
         last = self._checkoutUpstreamRevision(revision)
-        self.state_file.write(last.revision, getattr(self, 'pending', None))
+        self.state_file.applied(last)
+        # Some backend may have already fetched the remaining history
+        pending = getattr(self, 'pending', None)
+        if pending:
+            self.state_file.setPendingChangesets(pending)
         return last
 

vcpx/tests/__init__.py

@@ -12 +12 @@
 from svn import *
 from config import *
+from statefile import *
 from tailor import *
 

vcpx/project.py

@@ -14 +14 @@
 __docformat__ = 'reStructuredText'
 
+from cPickle import load, dump
+
 class StateFile(object):
     """
     State file that stores current revision and pending changesets.
+
+    It behaves as an iterator, and source backends loop over not yet
+    applied changesets, calling .applied() after each one: that writes
+    the applied changeset in a `journal` file, much more atomic than
+    rewriting the whole archive each time.
+
+    When the source backend finishes it's job, either because there
+    are no more pending changeset or stopped by an error, it calls
+    .finalize(), that in presence of a journal file adjust the
+    archive filtering out already applied changesets.
+
+    Should an hard error prevent .finalize() call, it will happen
+    automatically next time the state file is loaded.
     """
+
     def __init__(self, fname, config):
         self.filename = fname
+        self.archive = None
+        self.last_applied = None
+        self.current = None
+        self.queuelen = 0
+
+    def _load(self):
+        """
+        Open the pickle file and load the first two items, respectively
+        the revision and the number of pending changesets.
+        """
+
+        # Take care of the journal file, if present.
+        self.finalize()
+
+        self.current = None
+        try:
+            self.archive = open(self.filename)
+            self.last_applied = load(self.archive)
+            self.queuelen = load(self.archive)
+        except IOError:
+            self.archive = None
+            self.last_applied = None
+            self.queuelen = 0
+
+    def _write(self, changesets):
+        """
+        Write the state file.
+        """
+
+        sf = open(self.filename, 'w')
+        dump(self.last_applied, sf)
+        dump(len(changesets), sf)
+        for cs in changesets:
+            dump(cs, sf)
+        sf.close()
 
     def __str__(self):
         return self.filename
 
-    def load(self):
-        """
-        Read the source revision and pending changesets from the state file.
-        """
-
-        from cPickle import load
-
-        try:
-            sf = open(self.filename)
-            revision, changesets = load(sf)
+    def __iter__(self):
+        return self
+
+    def next(self):
+        if not self.archive:
+            raise StopIteration
+        try:
+            self.current = load(self.archive)
+        except EOFError:
+            self.archive.close()
+            self.archive = None
+            raise StopIteration
+        self.queuelen -= 1
+        return self.current
+
+    def __len__(self):
+        if self.archive is None:
+            self._load()
+        return self.queuelen
+
+    def applied(self, current=None):
+        """
+        Write the applied changeset to the journal file.
+        """
+
+        self.last_applied = current or self.current
+        journal = open(self.filename + '.journal', 'w')
+        dump(self.last_applied, journal)
+        journal.close()
+
+    def finalize(self):
+        """
+        If there is a journal file, adjust the archive accordingly,
+        dropping already applied changesets.
+        """
+
+        from os.path import exists
+        from os import unlink, rename
+
+        if self.archive is not None:
+            self.archive.close()
+            self.archive = None
+
+        if not exists(self.filename + '.journal'):
+            return
+
+        # Load last applied changeset from the journal
+        journal = open(self.filename + '.journal')
+        last_applied = load(journal)
+        journal.close()
+
+        # If there is an actual archive (ie, this is not bootstrap time)
+        # load the changesets from there, skipping the changesets until
+        # the last_applied one, then transfer the remaining to the new
+        # archive.
+        if exists(self.filename):
+            old = open(self.filename)
+            load(old) # last applied
+            queuelen = load(old)
+            cs = load(old)
+
+            # Skip already applied changesets
+            while cs <> last_applied:
+                queuelen -= 1
+                cs = load(old)
+
+            sf = open(self.filename + '.new', 'w')
+            dump(last_applied, sf)
+            dump(queuelen-1, sf)
+
+            while True:
+                try:
+                    cs = load(old)
+                except EOFError:
+                    break
+                dump(cs, sf)
             sf.close()
-        except IOError:
-            revision = None
-            changesets = None
-
-        return revision, changesets
-
-    def write(self, revision, changesets):
-        """
-        Write current source revision and pending changesets in the state file.
-        """
-
-        from cPickle import dump
-
-        sf = open(self.filename, 'w')
-        dump((revision, changesets), sf)
-        sf.close()
+            old.close()
+            unlink(self.filename)
+            rename(sf.name, self.filename)
+        else:
+            sf = open(self.filename, 'w')
+            dump(last_applied, sf)
+            dump(0, sf)
+            sf.close()
+
+        unlink(journal.name)
+
+    def lastAppliedChangeset(self):
+        """
+        Return the last applied changeset, if any, None otherwise.
+        """
+
+        if self.archive is None:
+            self._load()
+        return self.last_applied
+
+    def setPendingChangesets(self, changesets):
+        """
+        Write pending changesets to the state file.
+        """
+
+        if self.archive is not None:
+            self.archive.close()
+            self.archive = None
+
+        self._write(changesets)
+        self._load()
 
 
@@ -179 +311 @@
         """
         Return True if the project exists, False otherwise.
-
-        Check for the existence of the state file to decide.
-        """
-
-        from os.path import exists
-
-        return exists(self.state_file.filename)
+        """
+
+        return self.state_file.lastAppliedChangeset() is not None
 
     def workingDir(self):