"""TestCases for multi-threaded access to a DB. """ import os import sys import time import errno import shutil import tempfile from pprint import pprint from random import random try: True, False except NameError: True = 1 False = 0 DASH = '-' try: from threading import Thread, currentThread have_threads = True except ImportError: have_threads = False import unittest from test_all import verbose try: # For Pythons w/distutils pybsddb from bsddb3 import db, dbutils except ImportError: # For Python 2.3 from bsddb import db, dbutils #---------------------------------------------------------------------- class BaseThreadedTestCase(unittest.TestCase): dbtype = db.DB_UNKNOWN # must be set in derived class dbopenflags = 0 dbsetflags = 0 envflags = 0 def setUp(self): if verbose: dbutils._deadlock_VerboseFile = sys.stdout homeDir = os.path.join(os.path.dirname(sys.argv[0]), 'db_home') self.homeDir = homeDir try: os.mkdir(homeDir) except OSError, e: if e.errno <> errno.EEXIST: raise self.env = db.DBEnv() self.setEnvOpts() self.env.open(homeDir, self.envflags | db.DB_CREATE) self.filename = self.__class__.__name__ + '.db' self.d = db.DB(self.env) if self.dbsetflags: self.d.set_flags(self.dbsetflags) self.d.open(self.filename, self.dbtype, self.dbopenflags|db.DB_CREATE) def tearDown(self): self.d.close() self.env.close() shutil.rmtree(self.homeDir) def setEnvOpts(self): pass def makeData(self, key): return DASH.join([key] * 5) #---------------------------------------------------------------------- class ConcurrentDataStoreBase(BaseThreadedTestCase): dbopenflags = db.DB_THREAD envflags = db.DB_THREAD | db.DB_INIT_CDB | db.DB_INIT_MPOOL readers = 0 # derived class should set writers = 0 records = 1000 def test01_1WriterMultiReaders(self): if verbose: print '\n', '-=' * 30 print "Running %s.test01_1WriterMultiReaders..." % \ self.__class__.__name__ threads = [] for x in range(self.writers): wt = Thread(target = self.writerThread, args = (self.d, self.records, x), name = 'writer %d' % x, )#verbose = verbose) threads.append(wt) for x in range(self.readers): rt = Thread(target = self.readerThread, args = (self.d, x), name = 'reader %d' % x, )#verbose = verbose) threads.append(rt) for t in threads: t.start() for t in threads: t.join() def writerThread(self, d, howMany, writerNum): #time.sleep(0.01 * writerNum + 0.01) name = currentThread().getName() start = howMany * writerNum stop = howMany * (writerNum + 1) - 1 if verbose: print "%s: creating records %d - %d" % (name, start, stop) for x in range(start, stop): key = '%04d' % x dbutils.DeadlockWrap(d.put, key, self.makeData(key), max_retries=12) if verbose and x % 100 == 0: print "%s: records %d - %d finished" % (name, start, x) if verbose: print "%s: finished creating records" % name ## # Each write-cursor will be exclusive, the only one that can update the DB... ## if verbose: print "%s: deleting a few records" % name ## c = d.cursor(flags = db.DB_WRITECURSOR) ## for x in range(10): ## key = int(random() * howMany) + start ## key = '%04d' % key ## if d.has_key(key): ## c.set(key) ## c.delete() ## c.close() if verbose: print "%s: thread finished" % name def readerThread(self, d, readerNum): time.sleep(0.01 * readerNum) name = currentThread().getName() for loop in range(5): c = d.cursor() count = 0 rec = c.first() while rec: count += 1 key, data = rec self.assertEqual(self.makeData(key), data) rec = c.next() if verbose: print "%s: found %d records" % (name, count) c.close() time.sleep(0.05) if verbose: print "%s: thread finished" % name class BTreeConcurrentDataStore(ConcurrentDataStoreBase): dbtype = db.DB_BTREE writers = 2 readers = 10 records = 1000 class HashConcurrentDataStore(ConcurrentDataStoreBase): dbtype = db.DB_HASH writers = 2 readers = 10 records = 1000 #---------------------------------------------------------------------- class SimpleThreadedBase(BaseThreadedTestCase): dbopenflags = db.DB_THREAD envflags = db.DB_THREAD | db.DB_INIT_MPOOL | db.DB_INIT_LOCK readers = 5 writers = 3 records = 1000 def setEnvOpts(self): self.env.set_lk_detect(db.DB_LOCK_DEFAULT) def test02_SimpleLocks(self): if verbose: print '\n', '-=' * 30 print "Running %s.test02_SimpleLocks..." % self.__class__.__name__ threads = [] for x in range(self.writers): wt = Thread(target = self.writerThread, args = (self.d, self.records, x), name = 'writer %d' % x, )#verbose = verbose) threads.append(wt) for x in range(self.readers): rt = Thread(target = self.readerThread, args = (self.d, x), name = 'reader %d' % x, )#verbose = verbose) threads.append(rt) for t in threads: t.start() for t in threads: t.join() def writerThread(self, d, howMany, writerNum): name = currentThread().getName() start = howMany * writerNum stop = howMany * (writerNum + 1) - 1 if verbose: print "%s: creating records %d - %d" % (name, start, stop) # create a bunch of records for x in xrange(start, stop): key = '%04d' % x dbutils.DeadlockWrap(d.put, key, self.makeData(key), max_retries=12) if verbose and x % 100 == 0: print "%s: records %d - %d finished" % (name, start, x) # do a bit or reading too if random() <= 0.05: for y in xrange(start, x): key = '%04d' % x data = dbutils.DeadlockWrap(d.get, key, max_retries=12) self.assertEqual(data, self.makeData(key)) # flush them try: dbutils.DeadlockWrap(d.sync, max_retries=12) except db.DBIncompleteError, val: if verbose: print "could not complete sync()..." # read them back, deleting a few for x in xrange(start, stop): key = '%04d' % x data = dbutils.DeadlockWrap(d.get, key, max_retries=12) if verbose and x % 100 == 0: print "%s: fetched record (%s, %s)" % (name, key, data) self.assertEqual(data, self.makeData(key)) if random() <= 0.10: dbutils.DeadlockWrap(d.delete, key, max_retries=12) if verbose: print "%s: deleted record %s" % (name, key) if verbose: print "%s: thread finished" % name def readerThread(self, d, readerNum): time.sleep(0.01 * readerNum) name = currentThread().getName() for loop in range(5): c = d.cursor() count = 0 rec = dbutils.DeadlockWrap(c.first, max_retries=10) while rec: count += 1 key, data = rec self.assertEqual(self.makeData(key), data) rec = dbutils.DeadlockWrap(c.next, max_retries=10) if verbose: print "%s: found %d records" % (name, count) c.close() time.sleep(0.05) if verbose: print "%s: thread finished" % name class BTreeSimpleThreaded(SimpleThreadedBase): dbtype = db.DB_BTREE class HashSimpleThreaded(SimpleThreadedBase): dbtype = db.DB_HASH #---------------------------------------------------------------------- class ThreadedTransactionsBase(BaseThreadedTestCase): dbopenflags = db.DB_THREAD | db.DB_AUTO_COMMIT envflags = (db.DB_THREAD | db.DB_INIT_MPOOL | db.DB_INIT_LOCK | db.DB_INIT_LOG | db.DB_INIT_TXN ) readers = 0 writers = 0 records = 2000 txnFlag = 0 def setEnvOpts(self): #self.env.set_lk_detect(db.DB_LOCK_DEFAULT) pass def test03_ThreadedTransactions(self): if verbose: print '\n', '-=' * 30 print "Running %s.test03_ThreadedTransactions..." % \ self.__class__.__name__ threads = [] for x in range(self.writers): wt = Thread(target = self.writerThread, args = (self.d, self.records, x), name = 'writer %d' % x, )#verbose = verbose) threads.append(wt) for x in range(self.readers): rt = Thread(target = self.readerThread, args = (self.d, x), name = 'reader %d' % x, )#verbose = verbose) threads.append(rt) dt = Thread(target = self.deadlockThread) dt.start() for t in threads: t.start() for t in threads: t.join() self.doLockDetect = False dt.join() def doWrite(self, d, name, start, stop): finished = False while not finished: try: txn = self.env.txn_begin(None, self.txnFlag) for x in range(start, stop): key = '%04d' % x d.put(key, self.makeData(key), txn) if verbose and x % 100 == 0: print "%s: records %d - %d finished" % (name, start, x) txn.commit() finished = True except (db.DBLockDeadlockError, db.DBLockNotGrantedError), val: if verbose: print "%s: Aborting transaction (%s)" % (name, val[1]) txn.abort() time.sleep(0.05) def writerThread(self, d, howMany, writerNum): name = currentThread().getName() start = howMany * writerNum stop = howMany * (writerNum + 1) - 1 if verbose: print "%s: creating records %d - %d" % (name, start, stop) step = 100 for x in range(start, stop, step): self.doWrite(d, name, x, min(stop, x+step)) if verbose: print "%s: finished creating records" % name if verbose: print "%s: deleting a few records" % name finished = False while not finished: try: recs = [] txn = self.env.txn_begin(None, self.txnFlag) for x in range(10): key = int(random() * howMany) + start key = '%04d' % key data = d.get(key, None, txn, db.DB_RMW) if data is not None: d.delete(key, txn) recs.append(key) txn.commit() finished = True if verbose: print "%s: deleted records %s" % (name, recs) except (db.DBLockDeadlockError, db.DBLockNotGrantedError), val: if verbose: print "%s: Aborting transaction (%s)" % (name, val[1]) txn.abort() time.sleep(0.05) if verbose: print "%s: thread finished" % name def readerThread(self, d, readerNum): time.sleep(0.01 * readerNum + 0.05) name = currentThread().getName() for loop in range(5): finished = False while not finished: try: txn = self.env.txn_begin(None, self.txnFlag) c = d.cursor(txn) count = 0 rec = c.first() while rec: count += 1 key, data = rec self.assertEqual(self.makeData(key), data) rec = c.next() if verbose: print "%s: found %d records" % (name, count) c.close() txn.commit() finished = True except (db.DBLockDeadlockError, db.DBLockNotGrantedError), val: if verbose: print "%s: Aborting transaction (%s)" % (name, val[1]) c.close() txn.abort() time.sleep(0.05) time.sleep(0.05) if verbose: print "%s: thread finished" % name def deadlockThread(self): self.doLockDetect = True while self.doLockDetect: time.sleep(0.5) try: aborted = self.env.lock_detect( db.DB_LOCK_RANDOM, db.DB_LOCK_CONFLICT) if verbose and aborted: print "deadlock: Aborted %d deadlocked transaction(s)" \ % aborted except db.DBError: pass class BTreeThreadedTransactions(ThreadedTransactionsBase): dbtype = db.DB_BTREE writers = 3 readers = 5 records = 2000 class HashThreadedTransactions(ThreadedTransactionsBase): dbtype = db.DB_HASH writers = 1 readers = 5 records = 2000 class BTreeThreadedNoWaitTransactions(ThreadedTransactionsBase): dbtype = db.DB_BTREE writers = 3 readers = 5 records = 2000 txnFlag = db.DB_TXN_NOWAIT class HashThreadedNoWaitTransactions(ThreadedTransactionsBase): dbtype = db.DB_HASH writers = 1 readers = 5 records = 2000 txnFlag = db.DB_TXN_NOWAIT #---------------------------------------------------------------------- def test_suite(): suite = unittest.TestSuite() if have_threads: suite.addTest(unittest.makeSuite(BTreeConcurrentDataStore)) suite.addTest(unittest.makeSuite(HashConcurrentDataStore)) suite.addTest(unittest.makeSuite(BTreeSimpleThreaded)) suite.addTest(unittest.makeSuite(HashSimpleThreaded)) suite.addTest(unittest.makeSuite(BTreeThreadedTransactions)) suite.addTest(unittest.makeSuite(HashThreadedTransactions)) suite.addTest(unittest.makeSuite(BTreeThreadedNoWaitTransactions)) suite.addTest(unittest.makeSuite(HashThreadedNoWaitTransactions)) else: print "Threads not available, skipping thread tests." return suite if __name__ == '__main__': unittest.main(defaultTest='test_suite')