* 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.
diff --git a/Lib/test/test_bsddb.py b/Lib/test/test_bsddb.py
index 73a2d96..ff8c355 100755
--- a/Lib/test/test_bsddb.py
+++ b/Lib/test/test_bsddb.py
@@ -2,7 +2,7 @@
"""Test script for the bsddb C module by Roger E. Masse
Adapted to unittest format and expanded scope by Raymond Hettinger
"""
-import os
+import os, sys
import bsddb
import dbhash # Just so we know it's imported
import unittest
@@ -93,6 +93,57 @@
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):
k, v = self.f.popitem()
self.assert_(k in self.d)