| """TestCases for multi-threaded access to a DB. |
| """ |
| |
| import os |
| import sys |
| import time |
| import errno |
| from random import random |
| |
| DASH = '-' |
| |
| try: |
| WindowsError |
| except NameError: |
| class WindowsError(Exception): |
| pass |
| |
| import unittest |
| from test_all import db, dbutils, test_support, verbose, have_threads, \ |
| get_new_environment_path, get_new_database_path |
| |
| if have_threads : |
| from threading import Thread |
| if sys.version_info[0] < 3 : |
| from threading import currentThread |
| else : |
| from threading import current_thread as currentThread |
| |
| |
| #---------------------------------------------------------------------- |
| |
| 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 |
| |
| self.homeDir = get_new_environment_path() |
| self.env = db.DBEnv() |
| self.setEnvOpts() |
| self.env.open(self.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() |
| test_support.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__ |
| |
| keys=range(self.records) |
| import random |
| random.shuffle(keys) |
| records_per_writer=self.records//self.writers |
| readers_per_writer=self.readers//self.writers |
| self.assertEqual(self.records,self.writers*records_per_writer) |
| self.assertEqual(self.readers,self.writers*readers_per_writer) |
| self.assertTrue((records_per_writer%readers_per_writer)==0) |
| readers = [] |
| |
| for x in xrange(self.readers): |
| rt = Thread(target = self.readerThread, |
| args = (self.d, x), |
| name = 'reader %d' % x, |
| )#verbose = verbose) |
| if sys.version_info[0] < 3 : |
| rt.setDaemon(True) |
| else : |
| rt.daemon = True |
| readers.append(rt) |
| |
| writers=[] |
| for x in xrange(self.writers): |
| a=keys[records_per_writer*x:records_per_writer*(x+1)] |
| a.sort() # Generate conflicts |
| b=readers[readers_per_writer*x:readers_per_writer*(x+1)] |
| wt = Thread(target = self.writerThread, |
| args = (self.d, a, b), |
| name = 'writer %d' % x, |
| )#verbose = verbose) |
| writers.append(wt) |
| |
| for t in writers: |
| if sys.version_info[0] < 3 : |
| t.setDaemon(True) |
| else : |
| t.daemon = True |
| t.start() |
| |
| for t in writers: |
| t.join() |
| for t in readers: |
| t.join() |
| |
| def writerThread(self, d, keys, readers): |
| if sys.version_info[0] < 3 : |
| name = currentThread().getName() |
| else : |
| name = currentThread().name |
| |
| if verbose: |
| print "%s: creating records %d - %d" % (name, start, stop) |
| |
| count=len(keys)//len(readers) |
| count2=count |
| for x in keys : |
| 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) |
| |
| count2-=1 |
| if not count2 : |
| readers.pop().start() |
| count2=count |
| |
| if verbose: |
| print "%s: finished creating records" % name |
| |
| if verbose: |
| print "%s: thread finished" % name |
| |
| def readerThread(self, d, readerNum): |
| if sys.version_info[0] < 3 : |
| name = currentThread().getName() |
| else : |
| name = currentThread().name |
| |
| for i in xrange(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() |
| |
| 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 = 10 |
| writers = 2 |
| 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__ |
| |
| |
| keys=range(self.records) |
| import random |
| random.shuffle(keys) |
| records_per_writer=self.records//self.writers |
| readers_per_writer=self.readers//self.writers |
| self.assertEqual(self.records,self.writers*records_per_writer) |
| self.assertEqual(self.readers,self.writers*readers_per_writer) |
| self.assertTrue((records_per_writer%readers_per_writer)==0) |
| |
| readers = [] |
| for x in xrange(self.readers): |
| rt = Thread(target = self.readerThread, |
| args = (self.d, x), |
| name = 'reader %d' % x, |
| )#verbose = verbose) |
| if sys.version_info[0] < 3 : |
| rt.setDaemon(True) |
| else : |
| rt.daemon = True |
| readers.append(rt) |
| |
| writers = [] |
| for x in xrange(self.writers): |
| a=keys[records_per_writer*x:records_per_writer*(x+1)] |
| a.sort() # Generate conflicts |
| b=readers[readers_per_writer*x:readers_per_writer*(x+1)] |
| wt = Thread(target = self.writerThread, |
| args = (self.d, a, b), |
| name = 'writer %d' % x, |
| )#verbose = verbose) |
| writers.append(wt) |
| |
| for t in writers: |
| if sys.version_info[0] < 3 : |
| t.setDaemon(True) |
| else : |
| t.daemon = True |
| t.start() |
| |
| for t in writers: |
| t.join() |
| for t in readers: |
| t.join() |
| |
| def writerThread(self, d, keys, readers): |
| if sys.version_info[0] < 3 : |
| name = currentThread().getName() |
| else : |
| name = currentThread().name |
| if verbose: |
| print "%s: creating records %d - %d" % (name, start, stop) |
| |
| count=len(keys)//len(readers) |
| count2=count |
| for x in keys : |
| 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) |
| |
| count2-=1 |
| if not count2 : |
| readers.pop().start() |
| count2=count |
| |
| if verbose: |
| print "%s: thread finished" % name |
| |
| def readerThread(self, d, readerNum): |
| if sys.version_info[0] < 3 : |
| name = currentThread().getName() |
| else : |
| name = currentThread().name |
| |
| 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() |
| |
| 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__ |
| |
| keys=range(self.records) |
| import random |
| random.shuffle(keys) |
| records_per_writer=self.records//self.writers |
| readers_per_writer=self.readers//self.writers |
| self.assertEqual(self.records,self.writers*records_per_writer) |
| self.assertEqual(self.readers,self.writers*readers_per_writer) |
| self.assertTrue((records_per_writer%readers_per_writer)==0) |
| |
| readers=[] |
| for x in xrange(self.readers): |
| rt = Thread(target = self.readerThread, |
| args = (self.d, x), |
| name = 'reader %d' % x, |
| )#verbose = verbose) |
| if sys.version_info[0] < 3 : |
| rt.setDaemon(True) |
| else : |
| rt.daemon = True |
| readers.append(rt) |
| |
| writers = [] |
| for x in xrange(self.writers): |
| a=keys[records_per_writer*x:records_per_writer*(x+1)] |
| b=readers[readers_per_writer*x:readers_per_writer*(x+1)] |
| wt = Thread(target = self.writerThread, |
| args = (self.d, a, b), |
| name = 'writer %d' % x, |
| )#verbose = verbose) |
| writers.append(wt) |
| |
| dt = Thread(target = self.deadlockThread) |
| if sys.version_info[0] < 3 : |
| dt.setDaemon(True) |
| else : |
| dt.daemon = True |
| dt.start() |
| |
| for t in writers: |
| if sys.version_info[0] < 3 : |
| t.setDaemon(True) |
| else : |
| t.daemon = True |
| t.start() |
| |
| for t in writers: |
| t.join() |
| for t in readers: |
| t.join() |
| |
| self.doLockDetect = False |
| dt.join() |
| |
| def writerThread(self, d, keys, readers): |
| if sys.version_info[0] < 3 : |
| name = currentThread().getName() |
| else : |
| name = currentThread().name |
| |
| count=len(keys)//len(readers) |
| while len(keys): |
| try: |
| txn = self.env.txn_begin(None, self.txnFlag) |
| keys2=keys[:count] |
| for x in keys2 : |
| 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() |
| keys=keys[count:] |
| readers.pop().start() |
| except (db.DBLockDeadlockError, db.DBLockNotGrantedError), val: |
| if verbose: |
| if sys.version_info < (2, 6) : |
| print "%s: Aborting transaction (%s)" % (name, val[1]) |
| else : |
| print "%s: Aborting transaction (%s)" % (name, |
| val.args[1]) |
| txn.abort() |
| |
| if verbose: |
| print "%s: thread finished" % name |
| |
| def readerThread(self, d, readerNum): |
| if sys.version_info[0] < 3 : |
| name = currentThread().getName() |
| else : |
| name = currentThread().name |
| |
| 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: |
| if sys.version_info < (2, 6) : |
| print "%s: Aborting transaction (%s)" % (name, val[1]) |
| else : |
| print "%s: Aborting transaction (%s)" % (name, |
| val.args[1]) |
| c.close() |
| txn.abort() |
| |
| if verbose: |
| print "%s: thread finished" % name |
| |
| def deadlockThread(self): |
| self.doLockDetect = True |
| while self.doLockDetect: |
| time.sleep(0.05) |
| 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 = 2 |
| readers = 10 |
| records = 1000 |
| |
| class HashThreadedTransactions(ThreadedTransactionsBase): |
| dbtype = db.DB_HASH |
| writers = 2 |
| readers = 10 |
| records = 1000 |
| |
| class BTreeThreadedNoWaitTransactions(ThreadedTransactionsBase): |
| dbtype = db.DB_BTREE |
| writers = 2 |
| readers = 10 |
| records = 1000 |
| txnFlag = db.DB_TXN_NOWAIT |
| |
| class HashThreadedNoWaitTransactions(ThreadedTransactionsBase): |
| dbtype = db.DB_HASH |
| writers = 2 |
| readers = 10 |
| records = 1000 |
| 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') |
