text: """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__
  
  
          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: 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.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)]
              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')
  