Backport fix for concurrent writing and packing.

The pack() implementation would leave a corrupted storage behind if
transactions committed while packing.

trunk/src/ZODB/FileStorage.py

@@ -115,7 +115,7 @@
 #   may have a back pointer to a version record or to a non-version
 #   record.
 #
-__version__='$Revision: 1.131 $'[11:-2]
+__version__='$Revision: 1.132 $'[11:-2]
 
 import base64
 from cPickle import Pickler, Unpickler, loads
@@ -1460,25 +1460,28 @@ class FileStorage(BaseStorage.BaseStorage,
             if opos is None:
                 return
             oldpath = self._file_name + ".old"
-            self._file.close()
+            self._lock_acquire()
             try:
-                if os.path.exists(oldpath):
-                    os.remove(oldpath)
-                os.rename(self._file_name, oldpath)
-            except Exception, msg:
-                self._file = open(self._file_name, 'r+b')
-                raise
+                self._file.close()
+                try:
+                    if os.path.exists(oldpath):
+                        os.remove(oldpath)
+                    os.rename(self._file_name, oldpath)
+                except Exception, msg:
+                    self._file = open(self._file_name, 'r+b')
+                    raise
 
-            # OK, we're beyond the point of no return
-            os.rename(self._file_name + '.pack', self._file_name)
-            self._file = open(self._file_name, 'r+b')
-            self._initIndex(p.index, p.vindex, p.tindex, p.tvindex)
-            self._pos = opos
-            self._save_index()
+                # OK, we're beyond the point of no return
+                os.rename(self._file_name + '.pack', self._file_name)
+                self._file = open(self._file_name, 'r+b')
+                self._initIndex(p.index, p.vindex, p.tindex, p.tvindex)
+                self._pos = opos
+                self._save_index()
+            finally:
+                self._lock_release()
         finally:
             if p.locked:
                 self._commit_lock_release()
-                self._lock_release()
             self._lock_acquire()
             self._packt = z64
             self._lock_release()

trunk/src/ZODB/fspack.py

@@ -45,6 +45,23 @@ except ImportError:
 class CorruptedError(Exception):
     pass
 
+class CorruptedDataError(CorruptedError):
+
+    def __init__(self, oid=None, buf=None, pos=None):
+        self.oid = oid
+        self.buf = buf
+        self.pos = pos
+
+    def __str__(self):
+        if self.oid:
+            msg = "Error reading oid %s.  Found %r" % (_fmt_oid(self.oid),
+                                                       self.buf)
+        else:
+            msg = "Error reading unknown oid.  Found %r" % self.buf
+        if self.pos:
+            msg += " at %d" % self.pos
+        return msg
+
 # the struct formats for the headers
 TRANS_HDR = ">8s8scHHH"
 DATA_HDR = ">8s8s8s8sH8s"
@@ -672,6 +689,14 @@ class FileStoragePacker(FileStorageFormatter):
             self._tfile.close()
             os.remove(self._name + ".pack")
             return None
+        self._commit_lock_acquire()
+        self.locked = True
+        self._lock_acquire()
+        try:
+            self._file.seek(0, 2)
+            self.file_end = self._file.tell()
+        finally:
+            self._lock_release()
         if ipos < self.file_end:
             self.copyRest(ipos)
 
@@ -795,34 +820,58 @@ class FileStoragePacker(FileStorageFormatter):
         # After the pack time, all data records are copied.
         # Copy one txn at a time, using copy() for data.
 
-        while ipos < self.file_end:
-            th = self._read_txn_header(ipos)
-            pos = self._tfile.tell()
-            self._copier.setTxnPos(pos)
-            self._tfile.write(th.asString())
-            tend = ipos + th.tlen
-            ipos += th.headerlen()
-
-            while ipos < tend:
-                h = self._read_data_header(ipos)
-                ipos += h.recordlen()
-                prev_txn = None
-                if h.plen:
-                    data = self._file.read(h.plen)
-                else:
-                    data = self.fetchBackpointer(h.oid, h.back)
-                    if h.back:
-                        prev_txn = self.getTxnFromData(h.oid, h.back)
-
-                self._copier.copy(h.oid, h.serial, data, h.version,
-                                  prev_txn, pos, self._tfile.tell())
-
-            tlen = self._tfile.tell() - pos
-            assert tlen == th.tlen
-            self._tfile.write(p64(tlen))
-            ipos += 8
-
-            self.index.update(self.tindex)
-            self.tindex.clear()
-            self.vindex.update(self.tvindex)
-            self.tvindex.clear()
+        # Release the commit lock every 20 copies
+        self._lock_counter = 0
+
+        try:
+            while 1:
+                ipos = self.copyOne(ipos)
+        except CorruptedDataError, err:
+            # The last call to copyOne() will raise
+            # CorruptedDataError, because it will attempt to read past
+            # the end of the file.  Double-check that the exception
+            # occurred for this reason.
+            self._file.seek(0, 2)
+            endpos = self._file.tell()
+            if endpos != err.pos:
+                raise
+
+    def copyOne(self, ipos):
+        # The call below will raise CorruptedDataError at EOF.
+        th = self._read_txn_header(ipos)
+        self._lock_counter += 1
+        if self._lock_counter % 20 == 0:
+            self._commit_lock_release()
+        pos = self._tfile.tell()
+        self._copier.setTxnPos(pos)
+        self._tfile.write(th.asString())
+        tend = ipos + th.tlen
+        ipos += th.headerlen()
+
+        while ipos < tend:
+            h = self._read_data_header(ipos)
+            ipos += h.recordlen()
+            prev_txn = None
+            if h.plen:
+                data = self._file.read(h.plen)
+            else:
+                data = self.fetchBackpointer(h.oid, h.back)
+                if h.back:
+                    prev_txn = self.getTxnFromData(h.oid, h.back)
+
+            self._copier.copy(h.oid, h.serial, data, h.version,
+                              prev_txn, pos, self._tfile.tell())
+
+        tlen = self._tfile.tell() - pos
+        assert tlen == th.tlen
+        self._tfile.write(p64(tlen))
+        ipos += 8
+
+        self.index.update(self.tindex)
+        self.tindex.clear()
+        self.vindex.update(self.tvindex)
+        self.tvindex.clear()
+        if self._lock_counter % 20 == 0:
+            self._commit_lock_acquire()
+        return ipos
+

trunk/src/ZODB/tests/PackableStorage.py

@@ -25,11 +25,15 @@ try:
 except ImportError:
     from StringIO import StringIO
 
+import threading
 import time
+
 from ZODB import DB
 from Persistence import Persistent
 from ZODB.referencesf import referencesf
-
+from ZODB.tests.MinPO import MinPO
+from ZODB.tests.StorageTestBase import snooze
+from ZODB.POSException import ConflictError
 
 ZERO = '\0'*8
 
@@ -376,3 +380,56 @@ class PackableStorage(PackableStorageBase):
         conn.sync()
 
         eq(root['obj'].value, 7)
+
+    def checkPackWhileWriting(self):
+        # A storage should allow some reading and writing during
+        # a pack.  This test attempts to exercise locking code
+        # in the storage to test that it is safe.  It generates
+        # a lot of revisions, so that pack takes a long time.
+
+        db = DB(self._storage)
+        conn = db.open()
+        root = conn.root()
+
+        for i in range(10):
+            root[i] = MinPO(i)
+        get_transaction().commit()
+
+        snooze()
+        packt = time.time()
+
+        for j in range(10):
+            for i in range(10):
+                root[i].value = MinPO(i)
+                get_transaction().commit()
+
+        threads = [ClientThread(db) for i in range(4)]
+        for t in threads:
+            t.start()
+        db.pack(packt)
+        for t in threads:
+            t.join(30)
+        for t in threads:
+            t.join(1)
+            self.assert_(not t.isAlive())
+
+        # iterator over the storage to make sure it's sane
+        iter = self._storage.iterator()
+        for txn in iter:
+            for data in txn:
+                pass
+        iter.close()
+
+class ClientThread(threading.Thread):
+
+    def __init__(self, db):
+        threading.Thread.__init__(self)
+        self.root = db.open().root()
+        
+    def run(self):
+        for j in range(50):
+            try:
+                self.root[j % 10].value = MinPO(j)
+                get_transaction().commit()
+            except ConflictError:
+                get_transaction().abort()