[Zodb-checkins] CVS: ZODB3/ZEO/tests - InvalidationTests.py:1.2
ConnectionTests.py:1.27 testConnection.py:1.14
Jeremy Hylton
jeremy at zope.com
Fri Jun 13 18:57:08 EDT 2003
Update of /cvs-repository/ZODB3/ZEO/tests
In directory cvs.zope.org:/tmp/cvs-serv3078/ZEO/tests
Modified Files:
ConnectionTests.py testConnection.py
Added Files:
InvalidationTests.py
Log Message:
Bacport various cache consistency bug fixes from the ZODB3-3_1-branch.
=== ZODB3/ZEO/tests/InvalidationTests.py 1.1 => 1.2 ===
--- /dev/null Fri Jun 13 17:57:08 2003
+++ ZODB3/ZEO/tests/InvalidationTests.py Fri Jun 13 17:56:37 2003
@@ -0,0 +1,294 @@
+##############################################################################
+#
+# Copyright (c) 2003 Zope Corporation and Contributors.
+# All Rights Reserved.
+#
+# This software is subject to the provisions of the Zope Public License,
+# Version 2.0 (ZPL). A copy of the ZPL should accompany this distribution.
+# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
+# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
+# FOR A PARTICULAR PURPOSE
+#
+##############################################################################
+
+from thread import get_ident
+import threading
+import time
+
+from BTrees.check import check, display
+from BTrees.OOBTree import OOBTree
+
+from ZEO.tests.TestThread import TestThread
+
+from ZODB.DB import DB
+from ZODB.POSException \
+ import ReadConflictError, ConflictError, VersionLockError
+import zLOG
+
+class StressThread(TestThread):
+
+ def __init__(self, testcase, db, stop, threadnum, startnum,
+ step=2, sleep=None):
+ TestThread.__init__(self, testcase)
+ self.db = db
+ self.stop = stop
+ self.threadnum = threadnum
+ self.startnum = startnum
+ self.step = step
+ self.sleep = sleep
+ self.added_keys = []
+
+ def testrun(self):
+ cn = self.db.open()
+ while not self.stop.isSet():
+ try:
+ tree = cn.root()["tree"]
+ break
+ except (ConflictError, KeyError):
+ get_transaction().abort()
+ cn.sync()
+ key = self.startnum
+ while not self.stop.isSet():
+ try:
+ tree[key] = self.threadnum
+ get_transaction().note("add key %s" % key)
+ get_transaction().commit()
+ if self.sleep:
+ time.sleep(self.sleep)
+ except (ReadConflictError, ConflictError), msg:
+ get_transaction().abort()
+ # sync() is necessary here to process invalidations
+ # if we get a read conflict. In the read conflict case,
+ # no objects were modified so cn never got registered
+ # with the transaction.
+ cn.sync()
+ else:
+ self.added_keys.append(key)
+ key += self.step
+ cn.close()
+
+class VersionStressThread(TestThread):
+
+ def __init__(self, testcase, db, stop, threadnum, startnum,
+ step=2, sleep=None):
+ TestThread.__init__(self, testcase)
+ self.db = db
+ self.stop = stop
+ self.threadnum = threadnum
+ self.startnum = startnum
+ self.step = step
+ self.sleep = sleep
+ self.added_keys = []
+
+ def log(self, msg):
+ zLOG.LOG("thread %d" % get_ident(), 0, msg)
+
+ def testrun(self):
+ self.log("thread begin")
+ commit = 0
+ key = self.startnum
+ while not self.stop.isSet():
+ version = "%s:%s" % (self.threadnum, key)
+ commit = not commit
+ self.log("attempt to add key=%s version=%s commit=%d" %
+ (key, version, commit))
+ if self.oneupdate(version, key, commit):
+ self.added_keys.append(key)
+ key += self.step
+
+ def oneupdate(self, version, key, commit=1):
+ # The mess of sleeps below were added to reduce the number
+ # of VersionLockErrors, based on empirical observation.
+ # It looks like the threads don't switch enough without
+ # the sleeps.
+
+ cn = self.db.open(version)
+ while not self.stop.isSet():
+ try:
+ tree = cn.root()["tree"]
+ break
+ except (ConflictError, KeyError):
+ get_transaction().abort()
+ cn.sync()
+ while not self.stop.isSet():
+ try:
+ tree[key] = self.threadnum
+ get_transaction().note("add key %d" % key)
+ get_transaction().commit()
+ if self.sleep:
+ time.sleep(self.sleep)
+ break
+ except (VersionLockError, ReadConflictError, ConflictError), msg:
+ self.log(msg)
+ get_transaction().abort()
+ # sync() is necessary here to process invalidations
+ # if we get a read conflict. In the read conflict case,
+ # no objects were modified so cn never got registered
+ # with the transaction.
+ cn.sync()
+ if self.sleep:
+ time.sleep(self.sleep)
+ try:
+ while not self.stop.isSet():
+ try:
+ if commit:
+ self.db.commitVersion(version)
+ get_transaction().note("commit version %s" % version)
+ else:
+ self.db.abortVersion(version)
+ get_transaction().note("abort version %s" % version)
+ get_transaction().commit()
+ if self.sleep:
+ time.sleep(self.sleep)
+ return commit
+ except ConflictError, msg:
+ self.log(msg)
+ get_transaction().abort()
+ cn.sync()
+ finally:
+ cn.close()
+ return 0
+
+class InvalidationTests:
+
+ level = 2
+ DELAY = 15
+
+ def _check_tree(self, cn, tree):
+ # Make sure the BTree is sane and that all the updates persisted
+ retries = 3
+ while retries:
+ retries -= 1
+ try:
+ check(tree)
+ tree._check()
+ except ReadConflictError:
+ if retries:
+ get_transaction().abort()
+ cn.sync()
+ else:
+ raise
+ except:
+ display(tree)
+ raise
+
+ def _check_threads(self, tree, *threads):
+ # Make sure the thread's view of the world is consistent with
+ # the actual database state.
+ for t in threads:
+ # If the test didn't add any keys, it didn't do what we expected.
+ self.assert_(t.added_keys)
+ for key in t.added_keys:
+ self.assert_(tree.has_key(key), key)
+
+ def go(self, stop, *threads):
+ # Run the threads
+ for t in threads:
+ t.start()
+ time.sleep(self.DELAY)
+ stop.set()
+ for t in threads:
+ t.cleanup()
+
+ def checkConcurrentUpdates2Storages(self):
+ self._storage = storage1 = self.openClientStorage()
+ storage2 = self.openClientStorage(cache="2")
+ db1 = DB(storage1)
+ db2 = DB(storage2)
+ stop = threading.Event()
+
+ cn = db1.open()
+ tree = cn.root()["tree"] = OOBTree()
+ get_transaction().commit()
+
+ # Run two threads that update the BTree
+ t1 = StressThread(self, db1, stop, 1, 1)
+ t2 = StressThread(self, db2, stop, 2, 2)
+ self.go(stop, t1, t2)
+
+ cn.sync()
+ self._check_tree(cn, tree)
+ self._check_threads(tree, t1, t2)
+
+ cn.close()
+ db1.close()
+ db2.close()
+
+ def checkConcurrentUpdates1Storage(self):
+ self._storage = storage1 = self.openClientStorage()
+ db1 = DB(storage1)
+ stop = threading.Event()
+
+ cn = db1.open()
+ tree = cn.root()["tree"] = OOBTree()
+ get_transaction().commit()
+
+ # Run two threads that update the BTree
+ t1 = StressThread(self, db1, stop, 1, 1, sleep=0.001)
+ t2 = StressThread(self, db1, stop, 2, 2, sleep=0.001)
+ self.go(stop, t1, t2)
+
+ cn.sync()
+ self._check_tree(cn, tree)
+ self._check_threads(tree, t1, t2)
+
+ cn.close()
+ db1.close()
+
+ def checkConcurrentUpdates2StoragesMT(self):
+ self._storage = storage1 = self.openClientStorage()
+ db1 = DB(storage1)
+ stop = threading.Event()
+
+ cn = db1.open()
+ tree = cn.root()["tree"] = OOBTree()
+ get_transaction().commit()
+
+ db2 = DB(self.openClientStorage(cache="2"))
+ # Run three threads that update the BTree.
+ # Two of the threads share a single storage so that it
+ # is possible for both threads to read the same object
+ # at the same time.
+
+ t1 = StressThread(self, db1, stop, 1, 1, 3)
+ t2 = StressThread(self, db2, stop, 2, 2, 3, 0.001)
+ t3 = StressThread(self, db2, stop, 3, 3, 3, 0.001)
+ self.go(stop, t1, t2, t3)
+
+ cn.sync()
+ self._check_tree(cn, tree)
+ self._check_threads(tree, t1, t2, t3)
+
+ cn.close()
+ db1.close()
+ db2.close()
+
+ def checkConcurrentUpdatesInVersions(self):
+ self._storage = storage1 = self.openClientStorage()
+ db1 = DB(storage1)
+ db2 = DB(self.openClientStorage(cache="2"))
+ stop = threading.Event()
+
+ cn = db1.open()
+ tree = cn.root()["tree"] = OOBTree()
+ get_transaction().commit()
+
+ # Run three threads that update the BTree.
+ # Two of the threads share a single storage so that it
+ # is possible for both threads to read the same object
+ # at the same time.
+
+ t1 = VersionStressThread(self, db1, stop, 1, 1, 3)
+ t2 = VersionStressThread(self, db2, stop, 2, 2, 3, 0.001)
+ t3 = VersionStressThread(self, db2, stop, 3, 3, 3, 0.001)
+ self.go(stop, t1, t2, t3)
+
+ cn.sync()
+ self._check_tree(cn, tree)
+ self._check_threads(tree, t1, t2, t3)
+
+ cn.close()
+ db1.close()
+ db2.close()
+
=== ZODB3/ZEO/tests/ConnectionTests.py 1.26 => 1.27 ===
--- ZODB3/ZEO/tests/ConnectionTests.py:1.26 Fri May 30 15:20:56 2003
+++ ZODB3/ZEO/tests/ConnectionTests.py Fri Jun 13 17:56:37 2003
@@ -30,6 +30,7 @@
from ZEO.Exceptions import ClientDisconnected
from ZEO.zrpc.marshal import Marshaller
from ZEO.tests import forker
+from ZEO.tests.InvalidationTests import InvalidationTests
from ZODB.DB import DB
from ZODB.Transaction import get_transaction, Transaction
@@ -198,7 +199,7 @@
self.fail("timed out waiting for storage to disconnect")
-class ConnectionTests(CommonSetupTearDown):
+class ConnectionTests(CommonSetupTearDown, InvalidationTests):
"""Tests that explicitly manage the server process.
To test the cache or re-connection, these test cases explicit
=== ZODB3/ZEO/tests/testConnection.py 1.13 => 1.14 ===
--- ZODB3/ZEO/tests/testConnection.py:1.13 Thu May 29 14:26:56 2003
+++ ZODB3/ZEO/tests/testConnection.py Fri Jun 13 17:56:37 2003
@@ -110,12 +110,9 @@
test_classes = [FileStorageConnectionTests,
FileStorageReconnectionTests,
- FileStorageTimeoutTests]
-
-test_classes.extend(
- [MappingStorageConnectionTests,
- MappingStorageTimeoutTests])
-
+ FileStorageTimeoutTests,
+ MappingStorageConnectionTests,
+ MappingStorageTimeoutTests]
import BDBStorage
if BDBStorage.is_available:
More information about the Zodb-checkins
mailing list