Merge pull request #81 from zopefoundation/jm-merge-4

Merge 4.x branch

CHANGES.rst

@@ -5,7 +5,7 @@
 5.0.0a2 (unreleased)
 ====================
 
-See the 4.4.0 release.
+See the 4.4.x releases.
 
 5.0.0a1 (2016-06-20)
 ====================
@@ -23,6 +23,12 @@ Concurrency Control (MVCC) implementation:
   This change allows server-nased storages like ZEO and NEO to be
   implemented more simply and cleanly.
 
+4.4.1 (2016-07-01)
+==================
+
+Added IMultiCommitStorage to directly represent the changes in the 4.4.0
+release and to make complient storages introspectable.
+
 4.4.0 (2016-06-30)
 ==================
 

src/ZODB/BaseStorage.py

@@ -221,7 +221,7 @@ class BaseStorage(UndoLogCompatible):
             if transaction is not self._transaction:
                 raise POSException.StorageTransactionError(
                     "tpc_vote called with wrong transaction")
-            self._vote()
+            return self._vote()
 
     def _vote(self):
         """Subclasses should redefine this to supply transaction vote actions.
@@ -333,8 +333,8 @@ def copy(source, dest, verbose=0):
                              r.data_txn, transaction)
             else:
                 pre = preget(oid, None)
-                s = dest.store(oid, pre, r.data, r.version, transaction)
-                preindex[oid] = s
+                dest.store(oid, pre, r.data, r.version, transaction)
+                preindex[oid] = tid
 
         dest.tpc_vote(transaction)
         dest.tpc_finish(transaction)

src/ZODB/Connection.py

@@ -504,12 +504,12 @@ class Connection(ExportImport, object):
                 raise InvalidObjectReference(obj, obj._p_jar)
             elif oid in self._added:
                 assert obj._p_serial == z64
-            elif obj._p_changed:
-                self._modified.append(oid)
-            else:
+            elif oid in self._creating or not obj._p_changed:
                 # Nothing to do.  It's been said that it's legal, e.g., for
                 # an object to set _p_changed to false after it's been
                 # changed and registered.
+                # And new objects that are registered after any referrer are
+                # already processed.
                 continue
 
             self._store_objects(ObjectWriter(obj), transaction)
@@ -677,7 +677,7 @@ class Connection(ExportImport, object):
             raise
 
         if s:
-            if type(s[0]) is bytes:
+            if type(next(iter(s))) is bytes:
                 for oid in s:
                     self._handle_serial(oid)
                 return

src/ZODB/blob.py

@@ -710,9 +710,8 @@ class BlobStorageMixin(object):
         """Stores data that has a BLOB attached."""
         assert not version, "Versions aren't supported."
         serial = self.store(oid, oldserial, data, '', transaction)
-        self._blob_storeblob(oid, serial, blobfilename)
-
-        return self._tid
+        self._blob_storeblob(oid, self._tid, blobfilename)
+        return serial
 
     def temporaryDirectory(self):
         return self.fshelper.temp_dir
@@ -761,8 +760,9 @@ class BlobStorage(BlobStorageMixin):
         # We need to override the base storage's tpc_finish instead of
         # providing a _finish method because methods found on the proxied
         # object aren't rebound to the proxy
-        self.__storage.tpc_finish(*arg, **kw)
+        tid = self.__storage.tpc_finish(*arg, **kw)
         self._blob_tpc_finish()
+        return tid
 
     def tpc_abort(self, *arg, **kw):
         # We need to override the base storage's abort instead of

src/ZODB/interfaces.py

@@ -722,12 +722,8 @@ class IStorage(Interface):
         called while the storage transaction lock is held.  It takes
         the new transaction id generated by the transaction.
 
-        The return value must be the committed tid. It is used to set the
-        serial for objects whose ids were passed to previous store calls
-        in the same transaction.
-
-        For compatibility, the return value can also be None, in which case
-        store/tpc_vote must return the serial of stored objects.
+        The return value may be None or the transaction id of the
+        committed transaction, as described in IMultiCommitStorage.
         """
 
     def tpc_vote(transaction):
@@ -743,20 +739,53 @@ class IStorage(Interface):
         without an error, then there must not be an error if
         tpc_finish or tpc_abort is called subsequently.
 
-        The return value can be either None or a sequence of oids for which
-        a conflict was resolved.
-
-        For compatibility, the return value can also be a sequence of object-id
+        The return value can be None or a sequence of object-id
         and serial pairs giving new serials for objects whose ids were
         passed to previous store calls in the same transaction. The serial
         can be the special value ZODB.ConflictResolution.ResolvedSerial to
         indicate that a conflict occurred and that the object should be
         invalidated.
 
+        The return value can also be a sequence of object ids, as
+        described in IMultiCommitStorage.tpc_vote.
+
         After the tpc_vote call, all solved conflicts must have been notified,
         either from tpc_vote or store for objects passed to store.
         """
 
+
+class IMultiCommitStorage(IStorage):
+    """A multi-commit storage can commit multiple transactions at once.
+
+    It's likely that future versions of ZODB will require all storages
+    to provide this interface.
+    """
+
+    def store(oid, serial, data, version, transaction):
+        """Store data for the object id, oid.
+
+        See IStorage.store. For objects implementing this interface,
+        the return value is always None.
+        """
+
+    def tpc_finish(transaction, func = lambda tid: None):
+        """Finish the transaction, making any transaction changes permanent.
+
+        See IStorage.store. For objects implementing this interface,
+        the return value must be the committed tid. It is used to set the
+        serial for objects whose ids were passed to previous store calls
+        in the same transaction.
+        """
+
+    def tpc_vote(transaction):
+        """Provide a storage with an opportunity to veto a transaction
+
+        See IStorage.store. For objects implementing this interface,
+        the return value can be either None or a sequence of oids for which
+        a conflict was resolved.
+        """
+
+
 class IStorageRestoreable(IStorage):
     """Copying Transactions
 
@@ -893,8 +922,8 @@ class IStorageUndoable(IStorage):
         This method must only be called in the first phase of
         two-phase commit (after tpc_begin but before tpc_vote). It
         returns a serial (transaction id) and a sequence of object ids
-        for objects affected by the transaction.
-
+        for objects affected by the transaction. The serial is ignored
+        and may be None. The return from this method may be None.
         """
         # Used by DB (Actually, by TransactionalUndo)
 

src/ZODB/multicommitadapter.py

+"""Adapt non-IMultiCommitStorage storages to IMultiCommitStorage
+"""
+
+import zope.interface
+
+from .ConflictResolution import ResolvedSerial
+
+class MultiCommitAdapter:
+
+    def __init__(self, storage):
+        self._storage = storage
+        ifaces = zope.interface.providedBy(storage)
+        zope.interface.alsoProvides(self, ifaces)
+        self._resolved = set() # {OID}, here to make linters happy
+
+    def __getattr__(self, name):
+        v = getattr(self._storage, name)
+        self.__dict__[name] = v
+        return v
+
+    def tpc_begin(self, *args):
+        self._storage.tpc_begin(*args)
+        self._resolved = set()
+
+    def store(self, oid, *args):
+        if self._storage.store(oid, *args) == ResolvedSerial:
+            self._resolved.add(oid)
+
+    def storeBlob(self, oid, *args):
+        s = self._storage.storeBlob(oid, *args)
+        if s:
+            if isinstance(s, bytes):
+                s = ((oid, s), )
+
+            for oid, serial in s:
+                if s == ResolvedSerial:
+                    self._resolved.add(oid)
+
+    def undo(self, transaction_id, transaction):
+        r = self._storage.undo(transaction_id, transaction)
+        if r:
+            self._resolved.update(r[1])
+
+    def tpc_vote(self, *args):
+        s = self._storage.tpc_vote(*args)
+        for (oid, serial) in (s or ()):
+            if serial == ResolvedSerial:
+                self._resolved.add(oid)
+
+        return self._resolved
+
+    def tpc_finish(self, transaction, f=lambda tid: None):
+
+        t = []
+
+        def func(tid):
+            t.append(tid)
+            f(tid)
+
+        self._storage.tpc_finish(transaction, func)
+
+        return t[0]
+
+    def __len__(self):
+        return len(self._storage)

src/ZODB/mvccadapter.py

@@ -244,8 +244,11 @@ class UndoAdapterInstance(Base):
     def tpc_vote(self, transaction):
         result = self._storage.tpc_vote(transaction)
         if result:
-            for oid, serial in result:
-                self._undone.add(oid)
+            if isinstance(next(iter(result)), bytes):
+                self._undone.update(result)
+            else:
+                for oid, _ in result:
+                    self._undone.add(oid)
 
     def tpc_finish(self, transaction, func = lambda tid: None):
 

src/ZODB/tests/MVCCMappingStorage.py

@@ -112,7 +112,7 @@ class MVCCMappingStorage(MappingStorage):
 
     def tpc_finish(self, transaction, func = lambda tid: None):
         self._data_snapshot = None
-        MappingStorage.tpc_finish(self, transaction, func)
+        return MappingStorage.tpc_finish(self, transaction, func)
 
     def tpc_abort(self, transaction):
         self._data_snapshot = None

src/ZODB/tests/TransactionalUndoStorage.py

@@ -111,7 +111,12 @@ class TransactionalUndoStorage:
             undo_result = self._storage.undo(tid, t)
             if undo_result:
                 oids.extend(undo_result[1])
-        oids.extend(oid for (oid, _) in self._storage.tpc_vote(t) or ())
+        v = self._storage.tpc_vote(t)
+        if v:
+            if isinstance(next(iter(v)), bytes):
+                oids.extend(v)
+            else:
+                oids.extend(oid for (oid, _) in v)
         return oids
 
     def undo(self, tid, note):

src/ZODB/tests/blob_packing.txt

@@ -31,41 +31,20 @@ Put some revisions of a blob object in our database and on the filesystem:
     >>> import os
     >>> tids = []
     >>> times = []
-    >>> nothing = transaction.begin()
-    >>> times.append(new_time())
     >>> blob = Blob()
-    >>> with blob.open('w') as file:
-    ...     _ = file.write(b'this is blob data 0')
-    >>> root['blob'] = blob
-    >>> transaction.commit()
-    >>> tids.append(blob._p_serial)
 
-    >>> nothing = transaction.begin()
-    >>> times.append(new_time())
-    >>> with root['blob'].open('w') as file:
-    ...     _ = file.write(b'this is blob data 1')
-    >>> transaction.commit()
-    >>> tids.append(blob._p_serial)
-
-    >>> nothing = transaction.begin()
-    >>> times.append(new_time())
-    >>> with root['blob'].open('w') as file:
-    ...     _ = file.write(b'this is blob data 2')
-    >>> transaction.commit()
-    >>> tids.append(blob._p_serial)
-
-    >>> nothing = transaction.begin()
-    >>> times.append(new_time())
-    >>> with root['blob'].open('w') as file:
-    ...     _ = file.write(b'this is blob data 3')
-    >>> transaction.commit()
-    >>> tids.append(blob._p_serial)
-
-    >>> nothing = transaction.begin()
-    >>> times.append(new_time())
-    >>> with root['blob'].open('w') as file:
-    ...     _ = file.write(b'this is blob data 4')
-    >>> transaction.commit()
+    >>> for i in range(5):
+    ...     _ = transaction.begin()
+    ...     times.append(new_time())
+    ...     with blob.open('w') as file:
+    ...         _ = file.write(b'this is blob data ' + str(i).encode())
+    ...     if i:
+    ...         tids.append(blob._p_serial)
+    ...     else:
+    ...         root['blob'] = blob
+    ...     transaction.commit()
+
+    >>> blob._p_activate()
     >>> tids.append(blob._p_serial)
 
     >>> oid = root['blob']._p_oid

src/ZODB/tests/blob_transaction.txt

@@ -390,10 +390,6 @@ stored are discarded.
     ...                             '', t)
 
     >>> serials = blob_storage.tpc_vote(t)
-    >>> if s1 is None:
-    ...     s1 = [s for (oid, s) in serials if oid == blob._p_oid][0]
-    >>> if s2 is None:
-    ...     s2 = [s for (oid, s) in serials if oid == new_oid][0]
 
     >>> blob_storage.tpc_abort(t)
 
@@ -402,14 +398,7 @@ Now, the serial for the existing blob should be the same:
     >>> blob_storage.load(blob._p_oid, '') == (olddata, oldserial)
     True
 
-And we shouldn't be able to read the data that we saved:
-
-    >>> blob_storage.loadBlob(blob._p_oid, s1)
-    Traceback (most recent call last):
-    ...
-    POSKeyError: 'No blob file at <BLOB STORAGE PATH>'
-
-Of course the old data should be unaffected:
+The old data should be unaffected:
 
     >>> with open(blob_storage.loadBlob(blob._p_oid, oldserial)) as fp:
     ...     fp.read()
@@ -422,11 +411,6 @@ Similarly, the new object wasn't added to the storage:
     ...
     POSKeyError: 0x...
 
-    >>> blob_storage.loadBlob(blob._p_oid, s2)
-    Traceback (most recent call last):
-    ...
-    POSKeyError: 'No blob file at <BLOB STORAGE PATH>'
-
 .. clean up
 
     >>> tm1.abort()

src/ZODB/tests/testConnection.py

@@ -1011,9 +1011,14 @@ def doctest_lp485456_setattr_in_setstate_doesnt_cause_multiple_stores():
     storing '\x00\x00\x00\x00\x00\x00\x00\x00'
     storing '\x00\x00\x00\x00\x00\x00\x00\x01'
 
-    >>> conn.add(C())
+    Retry with the new object registered before its referrer.
+
+    >>> z = C()
+    >>> conn.add(z)
+    >>> conn.root.z = z
     >>> transaction.commit()
     storing '\x00\x00\x00\x00\x00\x00\x00\x02'
+    storing '\x00\x00\x00\x00\x00\x00\x00\x00'
 
     We still see updates:
 

src/ZODB/tests/testFileStorage.py

@@ -21,7 +21,6 @@ import transaction
 import ZODB.FileStorage
 import ZODB.tests.hexstorage
 import ZODB.tests.testblob
-import ZODB.tests.util
 import zope.testing.setupstack
 from ZODB import POSException
 from ZODB import DB
@@ -37,6 +36,7 @@ from ZODB.tests.StorageTestBase import MinPO, zodb_pickle
 from ZODB._compat import dump, dumps, _protocol
 
 from . import util
+from .. import multicommitadapter
 
 class FileStorageTests(
     StorageTestBase.StorageTestBase,
@@ -324,6 +324,12 @@ class FileStorageHexTests(FileStorageTests):
         self._storage = ZODB.tests.hexstorage.HexStorage(
             ZODB.FileStorage.FileStorage('FileStorageTests.fs',**kwargs))
 
+class MultiFileStorageTests(FileStorageTests):
+
+    def open(self, **kwargs):
+        self._storage = multicommitadapter.MultiCommitAdapter(
+            ZODB.FileStorage.FileStorage('FileStorageTests.fs', **kwargs))
+
 
 class FileStorageTestsWithBlobsEnabled(FileStorageTests):
 
@@ -333,6 +339,7 @@ class FileStorageTestsWithBlobsEnabled(FileStorageTests):
             kwargs['blob_dir'] = 'blobs'
         FileStorageTests.open(self, **kwargs)
 
+
 class FileStorageHexTestsWithBlobsEnabled(FileStorageTests):
 
     def open(self, **kwargs):
@@ -342,6 +349,16 @@ class FileStorageHexTestsWithBlobsEnabled(FileStorageTests):
         FileStorageTests.open(self, **kwargs)
         self._storage = ZODB.tests.hexstorage.HexStorage(self._storage)
 
+
+class MultiFileStorageTestsWithBlobsEnabled(MultiFileStorageTests):
+
+    def open(self, **kwargs):
+        if 'blob_dir' not in kwargs:
+            kwargs = kwargs.copy()
+            kwargs['blob_dir'] = 'blobs'
+        MultiFileStorageTests.open(self, **kwargs)
+
+
 class FileStorageRecoveryTest(
     StorageTestBase.StorageTestBase,
     RecoveryStorage.RecoveryStorage,
@@ -704,12 +721,13 @@ def test_suite():
         FileStorageNoRestoreRecoveryTest,
         FileStorageTestsWithBlobsEnabled, FileStorageHexTestsWithBlobsEnabled,
         AnalyzeDotPyTest,
+        MultiFileStorageTests, MultiFileStorageTestsWithBlobsEnabled,
         ]:
         suite.addTest(unittest.makeSuite(klass, "check"))
     suite.addTest(doctest.DocTestSuite(
         setUp=zope.testing.setupstack.setUpDirectory,
         tearDown=util.tearDown,
-        checker=ZODB.tests.util.checker))
+        checker=util.checker))
     suite.addTest(ZODB.tests.testblob.storage_reusable_suite(
         'BlobFileStorage',
         lambda name, blob_dir:
@@ -725,10 +743,18 @@ def test_suite():
         test_blob_storage_recovery=True,
         test_packing=True,
         ))
+    suite.addTest(ZODB.tests.testblob.storage_reusable_suite(
+        'BlobMultiFileStorage',
+        lambda name, blob_dir:
+        multicommitadapter.MultiCommitAdapter(
+            ZODB.FileStorage.FileStorage('%s.fs' % name, blob_dir=blob_dir)),
+        test_blob_storage_recovery=True,
+        test_packing=True,
+        ))
     suite.addTest(PackableStorage.IExternalGC_suite(
         lambda : ZODB.FileStorage.FileStorage(
             'data.fs', blob_dir='blobs', pack_gc=False)))
-    suite.layer = ZODB.tests.util.MininalTestLayer('testFileStorage')
+    suite.layer = util.MininalTestLayer('testFileStorage')
     return suite
 
 if __name__=='__main__':

src/ZODB/tests/testblob.py

@@ -712,8 +712,8 @@ def lp440234_Setting__p_changed_of_a_Blob_w_no_uncomitted_changes_is_noop():
     >>> blob = ZODB.blob.Blob(b'blah')
     >>> conn.add(blob)
     >>> transaction.commit()
-    >>> old_serial = blob._p_serial
     >>> blob._p_changed = True
+    >>> old_serial = blob._p_serial
     >>> transaction.commit()
     >>> with blob.open() as fp: fp.read()
     'blah'