[Python-checkins] python/dist/src/Lib/test test_bsddb.py,1.14,1.15

[greg at users.sourceforge.net]

Update of /cvsroot/python/python/dist/src/Lib/test
In directory sc8-pr-cvs1:/tmp/cvs-serv30540/Lib/test

Modified Files:
	test_bsddb.py 
Log Message:
* Fix the singlethreaded deadlocks occurring in the simple bsddb interface.
* Add support for multiple iterator/generator objects at once on the simple
  bsddb _DBWithCursor interface.


Index: test_bsddb.py
===================================================================
RCS file: /cvsroot/python/python/dist/src/Lib/test/test_bsddb.py,v
retrieving revision 1.14
retrieving revision 1.15
diff -C2 -d -r1.14 -r1.15
*** test_bsddb.py	13 Sep 2003 05:51:09 -0000	1.14
--- test_bsddb.py	2 Nov 2003 09:10:16 -0000	1.15
***************
*** 3,7 ****
     Adapted to unittest format and expanded scope by Raymond Hettinger
  """
! import os
  import bsddb
  import dbhash # Just so we know it's imported
--- 3,7 ----
     Adapted to unittest format and expanded scope by Raymond Hettinger
  """
! import os, sys
  import bsddb
  import dbhash # Just so we know it's imported
***************
*** 93,96 ****
--- 93,147 ----
          self.f.clear()
          self.assertEqual(len(self.f), 0)
+ 
+     def test__no_deadlock_first(self, debug=0):
+         # do this so that testers can see what function we're in in
+         # verbose mode when we deadlock.
+         sys.stdout.flush()
+ 
+         # in pybsddb's _DBWithCursor this causes an internal DBCursor
+         # object is created.  Other test_ methods in this class could
+         # inadvertently cause the deadlock but an explicit test is needed.
+         if debug: print "A"
+         k,v = self.f.first()
+         if debug: print "B", k
+         self.f[k] = "deadlock.  do not pass go.  do not collect $200."
+         if debug: print "C"
+         # if the bsddb implementation leaves the DBCursor open during
+         # the database write and locking+threading support is enabled
+         # the cursor's read lock will deadlock the write lock request..
+ 
+         # test the iterator interface (if present)
+         if hasattr(self, 'iteritems'):
+             if debug: print "D"
+             k,v = self.f.iteritems()
+             if debug: print "E"
+             self.f[k] = "please don't deadlock"
+             if debug: print "F"
+             while 1:
+                 try:
+                     k,v = self.f.iteritems()
+                 except StopIteration:
+                     break
+             if debug: print "F2"
+ 
+             i = iter(self.f)
+             if debug: print "G"
+             while i:
+                 try:
+                     if debug: print "H"
+                     k = i.next()
+                     if debug: print "I"
+                     self.f[k] = "deadlocks-r-us"
+                     if debug: print "J"
+                 except StopIteration:
+                     i = None
+             if debug: print "K"
+ 
+         # test the legacy cursor interface mixed with writes
+         self.assert_(self.f.first()[0] in self.d)
+         k = self.f.next()[0]
+         self.assert_(k in self.d)
+         self.f[k] = "be gone with ye deadlocks"
+         self.assert_(self.f[k], "be gone with ye deadlocks")
  
      def test_popitem(self):