Python code coverage for Lib/bsddb/test/test_thread.py

#	count	content
1	n/a	"""TestCases for multi-threaded access to a DB.
2	n/a	"""
3	n/a
4	n/a	import os
5	n/a	import sys
6	n/a	import time
7	n/a	import errno
8	n/a	from random import random
9	n/a
10	n/a	DASH = '-'
11	n/a
12	n/a	try:
13	n/a	WindowsError
14	n/a	except NameError:
15	n/a	class WindowsError(Exception):
16	n/a	pass
17	n/a
18	n/a	import unittest
19	n/a	from test_all import db, dbutils, test_support, verbose, have_threads, \
20	n/a	get_new_environment_path, get_new_database_path
21	n/a
22	n/a	if have_threads :
23	n/a	from threading import Thread
24	n/a	if sys.version_info[0] < 3 :
25	n/a	from threading import currentThread
26	n/a	else :
27	n/a	from threading import current_thread as currentThread
28	n/a
29	n/a
30	n/a	#----------------------------------------------------------------------
31	n/a
32	n/a	class BaseThreadedTestCase(unittest.TestCase):
33	n/a	dbtype = db.DB_UNKNOWN # must be set in derived class
34	n/a	dbopenflags = 0
35	n/a	dbsetflags = 0
36	n/a	envflags = 0
37	n/a
38	n/a	if sys.version_info < (2, 4) :
39	n/a	def assertTrue(self, expr, msg=None):
40	n/a	self.failUnless(expr,msg=msg)
41	n/a
42	n/a	def setUp(self):
43	n/a	if verbose:
44	n/a	dbutils._deadlock_VerboseFile = sys.stdout
45	n/a
46	n/a	self.homeDir = get_new_environment_path()
47	n/a	self.env = db.DBEnv()
48	n/a	self.setEnvOpts()
49	n/a	self.env.open(self.homeDir, self.envflags \| db.DB_CREATE)
50	n/a
51	n/a	self.filename = self.__class__.__name__ + '.db'
52	n/a	self.d = db.DB(self.env)
53	n/a	if self.dbsetflags:
54	n/a	self.d.set_flags(self.dbsetflags)
55	n/a	self.d.open(self.filename, self.dbtype, self.dbopenflags\|db.DB_CREATE)
56	n/a
57	n/a	def tearDown(self):
58	n/a	self.d.close()
59	n/a	self.env.close()
60	n/a	test_support.rmtree(self.homeDir)
61	n/a
62	n/a	def setEnvOpts(self):
63	n/a	pass
64	n/a
65	n/a	def makeData(self, key):
66	n/a	return DASH.join([key] * 5)
67	n/a
68	n/a
69	n/a	#----------------------------------------------------------------------
70	n/a
71	n/a
72	n/a	class ConcurrentDataStoreBase(BaseThreadedTestCase):
73	n/a	dbopenflags = db.DB_THREAD
74	n/a	envflags = db.DB_THREAD \| db.DB_INIT_CDB \| db.DB_INIT_MPOOL
75	n/a	readers = 0 # derived class should set
76	n/a	writers = 0
77	n/a	records = 1000
78	n/a
79	n/a	def test01_1WriterMultiReaders(self):
80	n/a	if verbose:
81	n/a	print '\n', '-=' * 30
82	n/a	print "Running %s.test01_1WriterMultiReaders..." % \
83	n/a	self.__class__.__name__
84	n/a
85	n/a	keys=range(self.records)
86	n/a	import random
87	n/a	random.shuffle(keys)
88	n/a	records_per_writer=self.records//self.writers
89	n/a	readers_per_writer=self.readers//self.writers
90	n/a	self.assertEqual(self.records,self.writers*records_per_writer)
91	n/a	self.assertEqual(self.readers,self.writers*readers_per_writer)
92	n/a	self.assertTrue((records_per_writer%readers_per_writer)==0)
93	n/a	readers = []
94	n/a
95	n/a	for x in xrange(self.readers):
96	n/a	rt = Thread(target = self.readerThread,
97	n/a	args = (self.d, x),
98	n/a	name = 'reader %d' % x,
99	n/a	)#verbose = verbose)
100	n/a	if sys.version_info[0] < 3 :
101	n/a	rt.setDaemon(True)
102	n/a	else :
103	n/a	rt.daemon = True
104	n/a	readers.append(rt)
105	n/a
106	n/a	writers=[]
107	n/a	for x in xrange(self.writers):
108	n/a	a=keys[records_per_writerx:records_per_writer(x+1)]
109	n/a	a.sort() # Generate conflicts
110	n/a	b=readers[readers_per_writerx:readers_per_writer(x+1)]
111	n/a	wt = Thread(target = self.writerThread,
112	n/a	args = (self.d, a, b),
113	n/a	name = 'writer %d' % x,
114	n/a	)#verbose = verbose)
115	n/a	writers.append(wt)
116	n/a
117	n/a	for t in writers:
118	n/a	if sys.version_info[0] < 3 :
119	n/a	t.setDaemon(True)
120	n/a	else :
121	n/a	t.daemon = True
122	n/a	t.start()
123	n/a
124	n/a	for t in writers:
125	n/a	t.join()
126	n/a	for t in readers:
127	n/a	t.join()
128	n/a
129	n/a	def writerThread(self, d, keys, readers):
130	n/a	if sys.version_info[0] < 3 :
131	n/a	name = currentThread().getName()
132	n/a	else :
133	n/a	name = currentThread().name
134	n/a
135	n/a	if verbose:
136	n/a	print "%s: creating records %d - %d" % (name, start, stop)
137	n/a
138	n/a	count=len(keys)//len(readers)
139	n/a	count2=count
140	n/a	for x in keys :
141	n/a	key = '%04d' % x
142	n/a	dbutils.DeadlockWrap(d.put, key, self.makeData(key),
143	n/a	max_retries=12)
144	n/a	if verbose and x % 100 == 0:
145	n/a	print "%s: records %d - %d finished" % (name, start, x)
146	n/a
147	n/a	count2-=1
148	n/a	if not count2 :
149	n/a	readers.pop().start()
150	n/a	count2=count
151	n/a
152	n/a	if verbose:
153	n/a	print "%s: finished creating records" % name
154	n/a
155	n/a	if verbose:
156	n/a	print "%s: thread finished" % name
157	n/a
158	n/a	def readerThread(self, d, readerNum):
159	n/a	if sys.version_info[0] < 3 :
160	n/a	name = currentThread().getName()
161	n/a	else :
162	n/a	name = currentThread().name
163	n/a
164	n/a	for i in xrange(5) :
165	n/a	c = d.cursor()
166	n/a	count = 0
167	n/a	rec = c.first()
168	n/a	while rec:
169	n/a	count += 1
170	n/a	key, data = rec
171	n/a	self.assertEqual(self.makeData(key), data)
172	n/a	rec = c.next()
173	n/a	if verbose:
174	n/a	print "%s: found %d records" % (name, count)
175	n/a	c.close()
176	n/a
177	n/a	if verbose:
178	n/a	print "%s: thread finished" % name
179	n/a
180	n/a
181	n/a	class BTreeConcurrentDataStore(ConcurrentDataStoreBase):
182	n/a	dbtype = db.DB_BTREE
183	n/a	writers = 2
184	n/a	readers = 10
185	n/a	records = 1000
186	n/a
187	n/a
188	n/a	class HashConcurrentDataStore(ConcurrentDataStoreBase):
189	n/a	dbtype = db.DB_HASH
190	n/a	writers = 2
191	n/a	readers = 10
192	n/a	records = 1000
193	n/a
194	n/a
195	n/a	#----------------------------------------------------------------------
196	n/a
197	n/a	class SimpleThreadedBase(BaseThreadedTestCase):
198	n/a	dbopenflags = db.DB_THREAD
199	n/a	envflags = db.DB_THREAD \| db.DB_INIT_MPOOL \| db.DB_INIT_LOCK
200	n/a	readers = 10
201	n/a	writers = 2
202	n/a	records = 1000
203	n/a
204	n/a	def setEnvOpts(self):
205	n/a	self.env.set_lk_detect(db.DB_LOCK_DEFAULT)
206	n/a
207	n/a	def test02_SimpleLocks(self):
208	n/a	if verbose:
209	n/a	print '\n', '-=' * 30
210	n/a	print "Running %s.test02_SimpleLocks..." % self.__class__.__name__
211	n/a
212	n/a
213	n/a	keys=range(self.records)
214	n/a	import random
215	n/a	random.shuffle(keys)
216	n/a	records_per_writer=self.records//self.writers
217	n/a	readers_per_writer=self.readers//self.writers
218	n/a	self.assertEqual(self.records,self.writers*records_per_writer)
219	n/a	self.assertEqual(self.readers,self.writers*readers_per_writer)
220	n/a	self.assertTrue((records_per_writer%readers_per_writer)==0)
221	n/a
222	n/a	readers = []
223	n/a	for x in xrange(self.readers):
224	n/a	rt = Thread(target = self.readerThread,
225	n/a	args = (self.d, x),
226	n/a	name = 'reader %d' % x,
227	n/a	)#verbose = verbose)
228	n/a	if sys.version_info[0] < 3 :
229	n/a	rt.setDaemon(True)
230	n/a	else :
231	n/a	rt.daemon = True
232	n/a	readers.append(rt)
233	n/a
234	n/a	writers = []
235	n/a	for x in xrange(self.writers):
236	n/a	a=keys[records_per_writerx:records_per_writer(x+1)]
237	n/a	a.sort() # Generate conflicts
238	n/a	b=readers[readers_per_writerx:readers_per_writer(x+1)]
239	n/a	wt = Thread(target = self.writerThread,
240	n/a	args = (self.d, a, b),
241	n/a	name = 'writer %d' % x,
242	n/a	)#verbose = verbose)
243	n/a	writers.append(wt)
244	n/a
245	n/a	for t in writers:
246	n/a	if sys.version_info[0] < 3 :
247	n/a	t.setDaemon(True)
248	n/a	else :
249	n/a	t.daemon = True
250	n/a	t.start()
251	n/a
252	n/a	for t in writers:
253	n/a	t.join()
254	n/a	for t in readers:
255	n/a	t.join()
256	n/a
257	n/a	def writerThread(self, d, keys, readers):
258	n/a	if sys.version_info[0] < 3 :
259	n/a	name = currentThread().getName()
260	n/a	else :
261	n/a	name = currentThread().name
262	n/a	if verbose:
263	n/a	print "%s: creating records %d - %d" % (name, start, stop)
264	n/a
265	n/a	count=len(keys)//len(readers)
266	n/a	count2=count
267	n/a	for x in keys :
268	n/a	key = '%04d' % x
269	n/a	dbutils.DeadlockWrap(d.put, key, self.makeData(key),
270	n/a	max_retries=12)
271	n/a
272	n/a	if verbose and x % 100 == 0:
273	n/a	print "%s: records %d - %d finished" % (name, start, x)
274	n/a
275	n/a	count2-=1
276	n/a	if not count2 :
277	n/a	readers.pop().start()
278	n/a	count2=count
279	n/a
280	n/a	if verbose:
281	n/a	print "%s: thread finished" % name
282	n/a
283	n/a	def readerThread(self, d, readerNum):
284	n/a	if sys.version_info[0] < 3 :
285	n/a	name = currentThread().getName()
286	n/a	else :
287	n/a	name = currentThread().name
288	n/a
289	n/a	c = d.cursor()
290	n/a	count = 0
291	n/a	rec = dbutils.DeadlockWrap(c.first, max_retries=10)
292	n/a	while rec:
293	n/a	count += 1
294	n/a	key, data = rec
295	n/a	self.assertEqual(self.makeData(key), data)
296	n/a	rec = dbutils.DeadlockWrap(c.next, max_retries=10)
297	n/a	if verbose:
298	n/a	print "%s: found %d records" % (name, count)
299	n/a	c.close()
300	n/a
301	n/a	if verbose:
302	n/a	print "%s: thread finished" % name
303	n/a
304	n/a
305	n/a	class BTreeSimpleThreaded(SimpleThreadedBase):
306	n/a	dbtype = db.DB_BTREE
307	n/a
308	n/a
309	n/a	class HashSimpleThreaded(SimpleThreadedBase):
310	n/a	dbtype = db.DB_HASH
311	n/a
312	n/a
313	n/a	#----------------------------------------------------------------------
314	n/a
315	n/a
316	n/a	class ThreadedTransactionsBase(BaseThreadedTestCase):
317	n/a	dbopenflags = db.DB_THREAD \| db.DB_AUTO_COMMIT
318	n/a	envflags = (db.DB_THREAD \|
319	n/a	db.DB_INIT_MPOOL \|
320	n/a	db.DB_INIT_LOCK \|
321	n/a	db.DB_INIT_LOG \|
322	n/a	db.DB_INIT_TXN
323	n/a	)
324	n/a	readers = 0
325	n/a	writers = 0
326	n/a	records = 2000
327	n/a	txnFlag = 0
328	n/a
329	n/a	def setEnvOpts(self):
330	n/a	#self.env.set_lk_detect(db.DB_LOCK_DEFAULT)
331	n/a	pass
332	n/a
333	n/a	def test03_ThreadedTransactions(self):
334	n/a	if verbose:
335	n/a	print '\n', '-=' * 30
336	n/a	print "Running %s.test03_ThreadedTransactions..." % \
337	n/a	self.__class__.__name__
338	n/a
339	n/a	keys=range(self.records)
340	n/a	import random
341	n/a	random.shuffle(keys)
342	n/a	records_per_writer=self.records//self.writers
343	n/a	readers_per_writer=self.readers//self.writers
344	n/a	self.assertEqual(self.records,self.writers*records_per_writer)
345	n/a	self.assertEqual(self.readers,self.writers*readers_per_writer)
346	n/a	self.assertTrue((records_per_writer%readers_per_writer)==0)
347	n/a
348	n/a	readers=[]
349	n/a	for x in xrange(self.readers):
350	n/a	rt = Thread(target = self.readerThread,
351	n/a	args = (self.d, x),
352	n/a	name = 'reader %d' % x,
353	n/a	)#verbose = verbose)
354	n/a	if sys.version_info[0] < 3 :
355	n/a	rt.setDaemon(True)
356	n/a	else :
357	n/a	rt.daemon = True
358	n/a	readers.append(rt)
359	n/a
360	n/a	writers = []
361	n/a	for x in xrange(self.writers):
362	n/a	a=keys[records_per_writerx:records_per_writer(x+1)]
363	n/a	b=readers[readers_per_writerx:readers_per_writer(x+1)]
364	n/a	wt = Thread(target = self.writerThread,
365	n/a	args = (self.d, a, b),
366	n/a	name = 'writer %d' % x,
367	n/a	)#verbose = verbose)
368	n/a	writers.append(wt)
369	n/a
370	n/a	dt = Thread(target = self.deadlockThread)
371	n/a	if sys.version_info[0] < 3 :
372	n/a	dt.setDaemon(True)
373	n/a	else :
374	n/a	dt.daemon = True
375	n/a	dt.start()
376	n/a
377	n/a	for t in writers:
378	n/a	if sys.version_info[0] < 3 :
379	n/a	t.setDaemon(True)
380	n/a	else :
381	n/a	t.daemon = True
382	n/a	t.start()
383	n/a
384	n/a	for t in writers:
385	n/a	t.join()
386	n/a	for t in readers:
387	n/a	t.join()
388	n/a
389	n/a	self.doLockDetect = False
390	n/a	dt.join()
391	n/a
392	n/a	def writerThread(self, d, keys, readers):
393	n/a	if sys.version_info[0] < 3 :
394	n/a	name = currentThread().getName()
395	n/a	else :
396	n/a	name = currentThread().name
397	n/a
398	n/a	count=len(keys)//len(readers)
399	n/a	while len(keys):
400	n/a	try:
401	n/a	txn = self.env.txn_begin(None, self.txnFlag)
402	n/a	keys2=keys[:count]
403	n/a	for x in keys2 :
404	n/a	key = '%04d' % x
405	n/a	d.put(key, self.makeData(key), txn)
406	n/a	if verbose and x % 100 == 0:
407	n/a	print "%s: records %d - %d finished" % (name, start, x)
408	n/a	txn.commit()
409	n/a	keys=keys[count:]
410	n/a	readers.pop().start()
411	n/a	except (db.DBLockDeadlockError, db.DBLockNotGrantedError), val:
412	n/a	if verbose:
413	n/a	if sys.version_info < (2, 6) :
414	n/a	print "%s: Aborting transaction (%s)" % (name, val[1])
415	n/a	else :
416	n/a	print "%s: Aborting transaction (%s)" % (name,
417	n/a	val.args[1])
418	n/a	txn.abort()
419	n/a
420	n/a	if verbose:
421	n/a	print "%s: thread finished" % name
422	n/a
423	n/a	def readerThread(self, d, readerNum):
424	n/a	if sys.version_info[0] < 3 :
425	n/a	name = currentThread().getName()
426	n/a	else :
427	n/a	name = currentThread().name
428	n/a
429	n/a	finished = False
430	n/a	while not finished:
431	n/a	try:
432	n/a	txn = self.env.txn_begin(None, self.txnFlag)
433	n/a	c = d.cursor(txn)
434	n/a	count = 0
435	n/a	rec = c.first()
436	n/a	while rec:
437	n/a	count += 1
438	n/a	key, data = rec
439	n/a	self.assertEqual(self.makeData(key), data)
440	n/a	rec = c.next()
441	n/a	if verbose: print "%s: found %d records" % (name, count)
442	n/a	c.close()
443	n/a	txn.commit()
444	n/a	finished = True
445	n/a	except (db.DBLockDeadlockError, db.DBLockNotGrantedError), val:
446	n/a	if verbose:
447	n/a	if sys.version_info < (2, 6) :
448	n/a	print "%s: Aborting transaction (%s)" % (name, val[1])
449	n/a	else :
450	n/a	print "%s: Aborting transaction (%s)" % (name,
451	n/a	val.args[1])
452	n/a	c.close()
453	n/a	txn.abort()
454	n/a
455	n/a	if verbose:
456	n/a	print "%s: thread finished" % name
457	n/a
458	n/a	def deadlockThread(self):
459	n/a	self.doLockDetect = True
460	n/a	while self.doLockDetect:
461	n/a	time.sleep(0.05)
462	n/a	try:
463	n/a	aborted = self.env.lock_detect(
464	n/a	db.DB_LOCK_RANDOM, db.DB_LOCK_CONFLICT)
465	n/a	if verbose and aborted:
466	n/a	print "deadlock: Aborted %d deadlocked transaction(s)" \
467	n/a	% aborted
468	n/a	except db.DBError:
469	n/a	pass
470	n/a
471	n/a
472	n/a	class BTreeThreadedTransactions(ThreadedTransactionsBase):
473	n/a	dbtype = db.DB_BTREE
474	n/a	writers = 2
475	n/a	readers = 10
476	n/a	records = 1000
477	n/a
478	n/a	class HashThreadedTransactions(ThreadedTransactionsBase):
479	n/a	dbtype = db.DB_HASH
480	n/a	writers = 2
481	n/a	readers = 10
482	n/a	records = 1000
483	n/a
484	n/a	class BTreeThreadedNoWaitTransactions(ThreadedTransactionsBase):
485	n/a	dbtype = db.DB_BTREE
486	n/a	writers = 2
487	n/a	readers = 10
488	n/a	records = 1000
489	n/a	txnFlag = db.DB_TXN_NOWAIT
490	n/a
491	n/a	class HashThreadedNoWaitTransactions(ThreadedTransactionsBase):
492	n/a	dbtype = db.DB_HASH
493	n/a	writers = 2
494	n/a	readers = 10
495	n/a	records = 1000
496	n/a	txnFlag = db.DB_TXN_NOWAIT
497	n/a
498	n/a
499	n/a	#----------------------------------------------------------------------
500	n/a
501	n/a	def test_suite():
502	n/a	suite = unittest.TestSuite()
503	n/a
504	n/a	if have_threads:
505	n/a	suite.addTest(unittest.makeSuite(BTreeConcurrentDataStore))
506	n/a	suite.addTest(unittest.makeSuite(HashConcurrentDataStore))
507	n/a	suite.addTest(unittest.makeSuite(BTreeSimpleThreaded))
508	n/a	suite.addTest(unittest.makeSuite(HashSimpleThreaded))
509	n/a	suite.addTest(unittest.makeSuite(BTreeThreadedTransactions))
510	n/a	suite.addTest(unittest.makeSuite(HashThreadedTransactions))
511	n/a	suite.addTest(unittest.makeSuite(BTreeThreadedNoWaitTransactions))
512	n/a	suite.addTest(unittest.makeSuite(HashThreadedNoWaitTransactions))
513	n/a
514	n/a	else:
515	n/a	print "Threads not available, skipping thread tests."
516	n/a
517	n/a	return suite
518	n/a
519	n/a
520	n/a	if __name__ == '__main__':
521	n/a	unittest.main(defaultTest='test_suite')