Begin a framework for testing invalidation-during-load scenario.

trunk/src/ZODB/tests/test_storage.py

@@ -11,6 +11,14 @@
 # FOR A PARTICULAR PURPOSE
 #
 ##############################################################################
+"""A storage used for unittests.
+
+The primary purpose of this module is to have a minimal multi-version
+storage to use for unit tests.  MappingStorage isn't sufficient.
+Since even a minimal storage has some complexity, we run standard
+storage tests against the test storage.
+"""
+
 import bisect
 import threading
 import unittest
@@ -41,8 +49,9 @@ class MinimalMemoryStorage(BaseStorage, object):
     """Simple in-memory storage that supports revisions.
 
     This storage is needed to test multi-version concurrency control.
-    It is similar to MappingStorage, but keeps multiple revisions.
-    It does not support versions.
+    It is similar to MappingStorage, but keeps multiple revisions.  It
+    does not support versions.  It doesn't implement operations like
+    pack(), because they aren't necessary for testing.
     """
 
     def __init__(self):
@@ -55,6 +64,10 @@ class MinimalMemoryStorage(BaseStorage, object):
     def isCurrent(self, oid, serial):
         return serial == self._cur[oid]
 
+    def hook(self, oid, tid, version):
+        # A hook for testing
+        pass
+
     def __len__(self):
         return len(self._index)
 
@@ -66,6 +79,7 @@ class MinimalMemoryStorage(BaseStorage, object):
         try:
             assert not version
             tid = self._cur[oid]
+            self.hook(oid, tid, version)
             return self._index[(oid, tid)], tid, ""
         finally:
             self._lock_release()

trunk/src/ZODB/tests/testmvcc.py

@@ -57,6 +57,9 @@ Now open a second connection.
 >>> cn2 = db.open()
 >>> txn2 = cn2.setLocalTransaction()
 
+Connection high-water mark
+--------------------------
+
 The ZODB Connection tracks a transaction high-water mark, which
 represents the latest transaction id that can be read by the current
 transaction and still present a consistent view of the database.  When
@@ -86,13 +89,15 @@ a transaction and process invalidations.
 >>> cn.sync()
 >>> cn._txn_time
 
+Basic functionality
+-------------------
+
 The next bit of code includes a simple MVCC test.  One transaction
 will begin and modify "a."  The other transaction will then modify "b"
 and commit.
 
 >>> r1 = cn1.root()
 >>> r1["a"].value = 2
-
 >>> cn1.getTransaction().commit()
 >>> txn = db.lastTransaction()
 
@@ -133,6 +138,161 @@ Traceback (most recent call last):
  ...
 ConflictError: database conflict error (oid 0000000000000001, class ZODB.tests.MinPO.MinPO)
 
+The failed commit aborted the current transaction, so we can try
+again.  This example will demonstrate that we can commit a transaction
+if we don't modify the object that isn't current.
+
+>>> cn2._txn_time
+
+>>> r1 = cn1.root()
+>>> r1["a"].value = 3
+>>> cn1.getTransaction().commit()
+>>> txn = db.lastTransaction()
+>>> cn2._txn_time == txn
+True
+
+>>> r2["b"].value = r2["a"].value + 1
+>>> r2["b"].value
+3
+>>> txn2.commit()
+>>> cn2._txn_time
+
+Object cache
+------------
+
+A Connection keeps objects in its cache so that multiple database
+references will always point to the same Python object.  At
+transaction boundaries, objects modified by other transactions are
+ghostified so that the next transaction doesn't see stale state.  We
+need to be sure the non-current objects loaded by MVCC are always
+ghosted.  It should be trivial, because MVCC is only used when an
+invalidation has been received for an object.
+
+First get the database back in an initial state.
+
+>>> cn1.sync()
+>>> r1["a"].value = 0
+>>> r1["b"].value = 0
+>>> cn1.getTransaction().commit()
+
+>>> cn2.sync()
+>>> r2["a"].value
+0
+>>> r2["b"].value = 1
+>>> cn2.getTransaction().commit()
+
+>>> r1["b"].value
+0
+>>> cn1.sync()
+>>> r1["b"]._p_state
+-1
+
+Closing the connection and commit a transaction should have the same effect.
+
+>>> def testit():
+...     cn1.sync()
+...     r1["a"].value = 0
+...     r1["b"].value = 0
+...     cn1.getTransaction().commit()
+...     cn2.sync()
+...     r2["b"].value = 1
+...     cn2.getTransaction().commit()
+
+>>> testit()
+>>> r1["a"].value = 1
+>>> cn1.getTransaction().commit()
+>>> r1["b"]._p_state
+-1
+
+When a connection is closed, it is saved by the database.  It will be
+reused by the next open() call (along with its object cache).
+
+>>> testit()
+>>> r1["a"].value = 1
+>>> cn1.close()
+>>> cn3 = db.open()
+>>> cn1 is cn3
+True
+>>> cn1 = cn3
+>>> r1 = cn1.root()
+
+It's not just that every object is a ghost.  The root was in the
+cache, so our first reference to it doesn't return a ghost.
+
+>>> r1._p_state
+0
+>>> r1["b"]._p_state
+-1
+
+Late invalidation
+-----------------
+
+The combination of ZEO and MVCC adds more complexity.  Since
+invalidations are delivered asynchronously by ZEO, it is possible for
+an invalidation to arrive just after a request to load the invalidated
+object is sent.  The connection can't use the just-loaded data,
+because the invalidation arrived first.  The complexity for MVCC is
+that it must check for invalidated objects after it has loaded them,
+just in case.
+
+Rather than add all the complexity of ZEO to these tests, the
+MinimalMemoryStorage has a hook.  We'll write a subclass that will
+deliver an invalidation when it loads an object.  The hook allows us
+to test the Connection code.
+
+>>> class TestStorage(MinimalMemoryStorage):
+...    def __init__(self):
+...        self.hooked = {}
+...        self.count = 0
+...        super(TestStorage, self).__init__()
+...    def registerDB(self, db, limit):
+...        self.db = db
+...    def hook(self, oid, tid, version):
+...        if oid in self.hooked:
+...            self.db.invalidate(tid, {oid:1})
+...            self.count += 1
+
+Now we'll repeat all the setup that was done earlier.
+
+>>> ts = TestStorage()
+>>> db = DB(ts)
+>>> cn1 = db.open()
+>>> txn1 = cn1.setLocalTransaction()
+>>> r1 = cn1.root()
+>>> r1["a"] = MinPO(0)
+>>> r1["b"] = MinPO(0)
+>>> cn1.getTransaction().commit()
+>>> cn1.cacheMinimize()
+
+>>> oid = r1["b"]._p_oid
+>>> ts.hooked[oid] = 1
+
+This test isn't quite rght yet, because it gets a ReadConflictError
+instead of getting a non-current revision.  Stil, it demonstrates that
+the basic mechanism for sending an invalidation during a load works.
+
+>>> oid in cn1._invalidated
+False
+>>> r1["b"]._p_state
+-1
+>>> r1["b"]._p_activate()
+Traceback (most recent call last):
+  ...
+ReadConflictError: database read conflict error (oid 0000000000000002, class ZODB.tests.MinPO.MinPO)
+>>> oid in cn1._invalidated
+True
+>>> ts.count
+1
+
+_p_independent() still has the desired effect.
+
+We still get a non-current version if the invalidation occurs while we
+are loading the current revision.  Can that happen without ZEO?
+
+Error cases:
+- storage doesn't have an earlier revision
+- MVCC returns current revision
+
 """
 
 import doctest