Gecko [ beriskala.com ]

Name	Size	Permission	Date
__init__.py	0 B	-rw-r--r--	2024-04-10 08:28
__init__.pyc	130 B	-rw-r--r--	2024-04-10 08:28
__init__.pyo	130 B	-rw-r--r--	2024-04-10 08:28
test_all.py	18.7 KB	-rw-r--r--	2024-04-10 08:28
test_all.pyc	23.95 KB	-rw-r--r--	2024-04-10 08:28
test_all.pyo	23.95 KB	-rw-r--r--	2024-04-10 08:28
test_associate.py	15.32 KB	-rw-r--r--	2024-04-10 08:28
test_associate.pyc	20.14 KB	-rw-r--r--	2024-04-10 08:28
test_associate.pyo	20.14 KB	-rw-r--r--	2024-04-10 08:28
test_basics.py	35.4 KB	-rw-r--r--	2024-04-10 08:28
test_basics.pyc	33.32 KB	-rw-r--r--	2024-04-10 08:28
test_basics.pyo	33.32 KB	-rw-r--r--	2024-04-10 08:28
test_compare.py	14.79 KB	-rw-r--r--	2024-04-10 08:28
test_compare.pyc	19.28 KB	-rw-r--r--	2024-04-10 08:28
test_compare.pyo	19.28 KB	-rw-r--r--	2024-04-10 08:28
test_compat.py	4.43 KB	-rw-r--r--	2024-04-10 08:28
test_compat.pyc	5.3 KB	-rw-r--r--	2024-04-10 08:28
test_compat.pyo	5.3 KB	-rw-r--r--	2024-04-10 08:28
test_cursor_pget_bug.py	1.83 KB	-rw-r--r--	2024-04-10 08:28
test_cursor_pget_bug.pyc	2.6 KB	-rw-r--r--	2024-04-10 08:28
test_cursor_pget_bug.pyo	2.6 KB	-rw-r--r--	2024-04-10 08:28
test_db.py	5.66 KB	-rw-r--r--	2024-04-10 08:28
test_db.pyc	8.58 KB	-rw-r--r--	2024-04-10 08:28
test_db.pyo	8.58 KB	-rw-r--r--	2024-04-10 08:28
test_dbenv.py	18.78 KB	-rw-r--r--	2024-04-10 08:28
test_dbenv.pyc	21.78 KB	-rw-r--r--	2024-04-10 08:28
test_dbenv.pyo	21.78 KB	-rw-r--r--	2024-04-10 08:28
test_dbobj.py	2.35 KB	-rw-r--r--	2024-04-10 08:28
test_dbobj.pyc	3.35 KB	-rw-r--r--	2024-04-10 08:28
test_dbobj.pyo	3.35 KB	-rw-r--r--	2024-04-10 08:28
test_dbshelve.py	11.42 KB	-rw-r--r--	2024-04-10 08:28
test_dbshelve.pyc	13.95 KB	-rw-r--r--	2024-04-10 08:28
test_dbshelve.pyo	13.95 KB	-rw-r--r--	2024-04-10 08:28
test_dbtables.py	14.98 KB	-rw-r--r--	2024-04-10 08:28
test_dbtables.pyc	11.49 KB	-rw-r--r--	2024-04-10 08:28
test_dbtables.pyo	11.49 KB	-rw-r--r--	2024-04-10 08:28
test_distributed_transactions.py	4.76 KB	-rw-r--r--	2024-04-10 08:28
test_distributed_transactions.pyc	5.53 KB	-rw-r--r--	2024-04-10 08:28
test_distributed_transactions.pyo	5.53 KB	-rw-r--r--	2024-04-10 08:28
test_early_close.py	7.27 KB	-rw-r--r--	2024-04-10 08:28
test_early_close.pyc	6.66 KB	-rw-r--r--	2024-04-10 08:28
test_early_close.pyo	6.66 KB	-rw-r--r--	2024-04-10 08:28
test_fileid.py	1.79 KB	-rw-r--r--	2024-04-10 08:28
test_fileid.pyc	2.37 KB	-rw-r--r--	2024-04-10 08:28
test_fileid.pyo	2.37 KB	-rw-r--r--	2024-04-10 08:28
test_get_none.py	2.24 KB	-rw-r--r--	2024-04-10 08:28
test_get_none.pyc	2.82 KB	-rw-r--r--	2024-04-10 08:28
test_get_none.pyo	2.82 KB	-rw-r--r--	2024-04-10 08:28
test_join.py	3.09 KB	-rw-r--r--	2024-04-10 08:28
test_join.pyc	3.54 KB	-rw-r--r--	2024-04-10 08:28
test_join.pyo	3.54 KB	-rw-r--r--	2024-04-10 08:28
test_lock.py	6.44 KB	-rw-r--r--	2024-04-10 08:28
test_lock.pyc	5.94 KB	-rw-r--r--	2024-04-10 08:28
test_lock.pyo	5.94 KB	-rw-r--r--	2024-04-10 08:28
test_misc.py	4.68 KB	-rw-r--r--	2024-04-10 08:28
test_misc.pyc	5.01 KB	-rw-r--r--	2024-04-10 08:28
test_misc.pyo	5.01 KB	-rw-r--r--	2024-04-10 08:28
test_pickle.py	1.93 KB	-rw-r--r--	2024-04-10 08:28
test_pickle.pyc	2.75 KB	-rw-r--r--	2024-04-10 08:28
test_pickle.pyo	2.75 KB	-rw-r--r--	2024-04-10 08:28
test_queue.py	3.92 KB	-rw-r--r--	2024-04-10 08:28
test_queue.pyc	3.9 KB	-rw-r--r--	2024-04-10 08:28
test_queue.pyo	3.9 KB	-rw-r--r--	2024-04-10 08:28
test_recno.py	8.55 KB	-rw-r--r--	2024-04-10 08:28
test_recno.pyc	7.61 KB	-rw-r--r--	2024-04-10 08:28
test_recno.pyo	7.61 KB	-rw-r--r--	2024-04-10 08:28
test_replication.py	20.97 KB	-rw-r--r--	2024-04-10 08:28
test_replication.pyc	15.54 KB	-rw-r--r--	2024-04-10 08:28
test_replication.pyo	15.54 KB	-rw-r--r--	2024-04-10 08:28
test_sequence.py	5.15 KB	-rw-r--r--	2024-04-10 08:28
test_sequence.pyc	6.56 KB	-rw-r--r--	2024-04-10 08:28
test_sequence.pyo	6.56 KB	-rw-r--r--	2024-04-10 08:28
test_thread.py	15.54 KB	-rw-r--r--	2024-04-10 08:28
test_thread.pyc	14.97 KB	-rw-r--r--	2024-04-10 08:28
test_thread.pyo	14.97 KB	-rw-r--r--	2024-04-10 08:28

Rename

"""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.assertEqual(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__

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__

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')