Python code coverage for Lib/test/lock_tests.py

#	count	content
1	n/a	"""
2	n/a	Various tests for synchronization primitives.
3	n/a	"""
4	n/a
5	n/a	import sys
6	n/a	import time
7	n/a	from _thread import start_new_thread, TIMEOUT_MAX
8	n/a	import threading
9	n/a	import unittest
10	n/a	import weakref
11	n/a
12	n/a	from test import support
13	n/a
14	n/a
15	n/a	def _wait():
16	n/a	# A crude wait/yield function not relying on synchronization primitives.
17	n/a	time.sleep(0.01)
18	n/a
19	n/a	class Bunch(object):
20	n/a	"""
21	n/a	A bunch of threads.
22	n/a	"""
23	n/a	def __init__(self, f, n, wait_before_exit=False):
24	n/a	"""
25	n/a	Construct a bunch of `n` threads running the same function `f`.
26	n/a	If `wait_before_exit` is True, the threads won't terminate until
27	n/a	do_finish() is called.
28	n/a	"""
29	n/a	self.f = f
30	n/a	self.n = n
31	n/a	self.started = []
32	n/a	self.finished = []
33	n/a	self._can_exit = not wait_before_exit
34	n/a	def task():
35	n/a	tid = threading.get_ident()
36	n/a	self.started.append(tid)
37	n/a	try:
38	n/a	f()
39	n/a	finally:
40	n/a	self.finished.append(tid)
41	n/a	while not self._can_exit:
42	n/a	_wait()
43	n/a	try:
44	n/a	for i in range(n):
45	n/a	start_new_thread(task, ())
46	n/a	except:
47	n/a	self._can_exit = True
48	n/a	raise
49	n/a
50	n/a	def wait_for_started(self):
51	n/a	while len(self.started) < self.n:
52	n/a	_wait()
53	n/a
54	n/a	def wait_for_finished(self):
55	n/a	while len(self.finished) < self.n:
56	n/a	_wait()
57	n/a
58	n/a	def do_finish(self):
59	n/a	self._can_exit = True
60	n/a
61	n/a
62	n/a	class BaseTestCase(unittest.TestCase):
63	n/a	def setUp(self):
64	n/a	self._threads = support.threading_setup()
65	n/a
66	n/a	def tearDown(self):
67	n/a	support.threading_cleanup(*self._threads)
68	n/a	support.reap_children()
69	n/a
70	n/a	def assertTimeout(self, actual, expected):
71	n/a	# The waiting and/or time.time() can be imprecise, which
72	n/a	# is why comparing to the expected value would sometimes fail
73	n/a	# (especially under Windows).
74	n/a	self.assertGreaterEqual(actual, expected * 0.6)
75	n/a	# Test nothing insane happened
76	n/a	self.assertLess(actual, expected * 10.0)
77	n/a
78	n/a
79	n/a	class BaseLockTests(BaseTestCase):
80	n/a	"""
81	n/a	Tests for both recursive and non-recursive locks.
82	n/a	"""
83	n/a
84	n/a	def test_constructor(self):
85	n/a	lock = self.locktype()
86	n/a	del lock
87	n/a
88	n/a	def test_repr(self):
89	n/a	lock = self.locktype()
90	n/a	self.assertRegex(repr(lock), "<unlocked .* object (.)?at .>")
91	n/a	del lock
92	n/a
93	n/a	def test_locked_repr(self):
94	n/a	lock = self.locktype()
95	n/a	lock.acquire()
96	n/a	self.assertRegex(repr(lock), "<locked .* object (.)?at .>")
97	n/a	del lock
98	n/a
99	n/a	def test_acquire_destroy(self):
100	n/a	lock = self.locktype()
101	n/a	lock.acquire()
102	n/a	del lock
103	n/a
104	n/a	def test_acquire_release(self):
105	n/a	lock = self.locktype()
106	n/a	lock.acquire()
107	n/a	lock.release()
108	n/a	del lock
109	n/a
110	n/a	def test_try_acquire(self):
111	n/a	lock = self.locktype()
112	n/a	self.assertTrue(lock.acquire(False))
113	n/a	lock.release()
114	n/a
115	n/a	def test_try_acquire_contended(self):
116	n/a	lock = self.locktype()
117	n/a	lock.acquire()
118	n/a	result = []
119	n/a	def f():
120	n/a	result.append(lock.acquire(False))
121	n/a	Bunch(f, 1).wait_for_finished()
122	n/a	self.assertFalse(result[0])
123	n/a	lock.release()
124	n/a
125	n/a	def test_acquire_contended(self):
126	n/a	lock = self.locktype()
127	n/a	lock.acquire()
128	n/a	N = 5
129	n/a	def f():
130	n/a	lock.acquire()
131	n/a	lock.release()
132	n/a
133	n/a	b = Bunch(f, N)
134	n/a	b.wait_for_started()
135	n/a	_wait()
136	n/a	self.assertEqual(len(b.finished), 0)
137	n/a	lock.release()
138	n/a	b.wait_for_finished()
139	n/a	self.assertEqual(len(b.finished), N)
140	n/a
141	n/a	def test_with(self):
142	n/a	lock = self.locktype()
143	n/a	def f():
144	n/a	lock.acquire()
145	n/a	lock.release()
146	n/a	def _with(err=None):
147	n/a	with lock:
148	n/a	if err is not None:
149	n/a	raise err
150	n/a	_with()
151	n/a	# Check the lock is unacquired
152	n/a	Bunch(f, 1).wait_for_finished()
153	n/a	self.assertRaises(TypeError, _with, TypeError)
154	n/a	# Check the lock is unacquired
155	n/a	Bunch(f, 1).wait_for_finished()
156	n/a
157	n/a	def test_thread_leak(self):
158	n/a	# The lock shouldn't leak a Thread instance when used from a foreign
159	n/a	# (non-threading) thread.
160	n/a	lock = self.locktype()
161	n/a	def f():
162	n/a	lock.acquire()
163	n/a	lock.release()
164	n/a	n = len(threading.enumerate())
165	n/a	# We run many threads in the hope that existing threads ids won't
166	n/a	# be recycled.
167	n/a	Bunch(f, 15).wait_for_finished()
168	n/a	if len(threading.enumerate()) != n:
169	n/a	# There is a small window during which a Thread instance's
170	n/a	# target function has finished running, but the Thread is still
171	n/a	# alive and registered. Avoid spurious failures by waiting a
172	n/a	# bit more (seen on a buildbot).
173	n/a	time.sleep(0.4)
174	n/a	self.assertEqual(n, len(threading.enumerate()))
175	n/a
176	n/a	def test_timeout(self):
177	n/a	lock = self.locktype()
178	n/a	# Can't set timeout if not blocking
179	n/a	self.assertRaises(ValueError, lock.acquire, 0, 1)
180	n/a	# Invalid timeout values
181	n/a	self.assertRaises(ValueError, lock.acquire, timeout=-100)
182	n/a	self.assertRaises(OverflowError, lock.acquire, timeout=1e100)
183	n/a	self.assertRaises(OverflowError, lock.acquire, timeout=TIMEOUT_MAX + 1)
184	n/a	# TIMEOUT_MAX is ok
185	n/a	lock.acquire(timeout=TIMEOUT_MAX)
186	n/a	lock.release()
187	n/a	t1 = time.time()
188	n/a	self.assertTrue(lock.acquire(timeout=5))
189	n/a	t2 = time.time()
190	n/a	# Just a sanity test that it didn't actually wait for the timeout.
191	n/a	self.assertLess(t2 - t1, 5)
192	n/a	results = []
193	n/a	def f():
194	n/a	t1 = time.time()
195	n/a	results.append(lock.acquire(timeout=0.5))
196	n/a	t2 = time.time()
197	n/a	results.append(t2 - t1)
198	n/a	Bunch(f, 1).wait_for_finished()
199	n/a	self.assertFalse(results[0])
200	n/a	self.assertTimeout(results[1], 0.5)
201	n/a
202	n/a	def test_weakref_exists(self):
203	n/a	lock = self.locktype()
204	n/a	ref = weakref.ref(lock)
205	n/a	self.assertIsNotNone(ref())
206	n/a
207	n/a	def test_weakref_deleted(self):
208	n/a	lock = self.locktype()
209	n/a	ref = weakref.ref(lock)
210	n/a	del lock
211	n/a	self.assertIsNone(ref())
212	n/a
213	n/a
214	n/a	class LockTests(BaseLockTests):
215	n/a	"""
216	n/a	Tests for non-recursive, weak locks
217	n/a	(which can be acquired and released from different threads).
218	n/a	"""
219	n/a	def test_reacquire(self):
220	n/a	# Lock needs to be released before re-acquiring.
221	n/a	lock = self.locktype()
222	n/a	phase = []
223	n/a	def f():
224	n/a	lock.acquire()
225	n/a	phase.append(None)
226	n/a	lock.acquire()
227	n/a	phase.append(None)
228	n/a	start_new_thread(f, ())
229	n/a	while len(phase) == 0:
230	n/a	_wait()
231	n/a	_wait()
232	n/a	self.assertEqual(len(phase), 1)
233	n/a	lock.release()
234	n/a	while len(phase) == 1:
235	n/a	_wait()
236	n/a	self.assertEqual(len(phase), 2)
237	n/a
238	n/a	def test_different_thread(self):
239	n/a	# Lock can be released from a different thread.
240	n/a	lock = self.locktype()
241	n/a	lock.acquire()
242	n/a	def f():
243	n/a	lock.release()
244	n/a	b = Bunch(f, 1)
245	n/a	b.wait_for_finished()
246	n/a	lock.acquire()
247	n/a	lock.release()
248	n/a
249	n/a	def test_state_after_timeout(self):
250	n/a	# Issue #11618: check that lock is in a proper state after a
251	n/a	# (non-zero) timeout.
252	n/a	lock = self.locktype()
253	n/a	lock.acquire()
254	n/a	self.assertFalse(lock.acquire(timeout=0.01))
255	n/a	lock.release()
256	n/a	self.assertFalse(lock.locked())
257	n/a	self.assertTrue(lock.acquire(blocking=False))
258	n/a
259	n/a
260	n/a	class RLockTests(BaseLockTests):
261	n/a	"""
262	n/a	Tests for recursive locks.
263	n/a	"""
264	n/a	def test_reacquire(self):
265	n/a	lock = self.locktype()
266	n/a	lock.acquire()
267	n/a	lock.acquire()
268	n/a	lock.release()
269	n/a	lock.acquire()
270	n/a	lock.release()
271	n/a	lock.release()
272	n/a
273	n/a	def test_release_unacquired(self):
274	n/a	# Cannot release an unacquired lock
275	n/a	lock = self.locktype()
276	n/a	self.assertRaises(RuntimeError, lock.release)
277	n/a	lock.acquire()
278	n/a	lock.acquire()
279	n/a	lock.release()
280	n/a	lock.acquire()
281	n/a	lock.release()
282	n/a	lock.release()
283	n/a	self.assertRaises(RuntimeError, lock.release)
284	n/a
285	n/a	def test_release_save_unacquired(self):
286	n/a	# Cannot _release_save an unacquired lock
287	n/a	lock = self.locktype()
288	n/a	self.assertRaises(RuntimeError, lock._release_save)
289	n/a	lock.acquire()
290	n/a	lock.acquire()
291	n/a	lock.release()
292	n/a	lock.acquire()
293	n/a	lock.release()
294	n/a	lock.release()
295	n/a	self.assertRaises(RuntimeError, lock._release_save)
296	n/a
297	n/a	def test_different_thread(self):
298	n/a	# Cannot release from a different thread
299	n/a	lock = self.locktype()
300	n/a	def f():
301	n/a	lock.acquire()
302	n/a	b = Bunch(f, 1, True)
303	n/a	try:
304	n/a	self.assertRaises(RuntimeError, lock.release)
305	n/a	finally:
306	n/a	b.do_finish()
307	n/a
308	n/a	def test__is_owned(self):
309	n/a	lock = self.locktype()
310	n/a	self.assertFalse(lock._is_owned())
311	n/a	lock.acquire()
312	n/a	self.assertTrue(lock._is_owned())
313	n/a	lock.acquire()
314	n/a	self.assertTrue(lock._is_owned())
315	n/a	result = []
316	n/a	def f():
317	n/a	result.append(lock._is_owned())
318	n/a	Bunch(f, 1).wait_for_finished()
319	n/a	self.assertFalse(result[0])
320	n/a	lock.release()
321	n/a	self.assertTrue(lock._is_owned())
322	n/a	lock.release()
323	n/a	self.assertFalse(lock._is_owned())
324	n/a
325	n/a
326	n/a	class EventTests(BaseTestCase):
327	n/a	"""
328	n/a	Tests for Event objects.
329	n/a	"""
330	n/a
331	n/a	def test_is_set(self):
332	n/a	evt = self.eventtype()
333	n/a	self.assertFalse(evt.is_set())
334	n/a	evt.set()
335	n/a	self.assertTrue(evt.is_set())
336	n/a	evt.set()
337	n/a	self.assertTrue(evt.is_set())
338	n/a	evt.clear()
339	n/a	self.assertFalse(evt.is_set())
340	n/a	evt.clear()
341	n/a	self.assertFalse(evt.is_set())
342	n/a
343	n/a	def _check_notify(self, evt):
344	n/a	# All threads get notified
345	n/a	N = 5
346	n/a	results1 = []
347	n/a	results2 = []
348	n/a	def f():
349	n/a	results1.append(evt.wait())
350	n/a	results2.append(evt.wait())
351	n/a	b = Bunch(f, N)
352	n/a	b.wait_for_started()
353	n/a	_wait()
354	n/a	self.assertEqual(len(results1), 0)
355	n/a	evt.set()
356	n/a	b.wait_for_finished()
357	n/a	self.assertEqual(results1, [True] * N)
358	n/a	self.assertEqual(results2, [True] * N)
359	n/a
360	n/a	def test_notify(self):
361	n/a	evt = self.eventtype()
362	n/a	self._check_notify(evt)
363	n/a	# Another time, after an explicit clear()
364	n/a	evt.set()
365	n/a	evt.clear()
366	n/a	self._check_notify(evt)
367	n/a
368	n/a	def test_timeout(self):
369	n/a	evt = self.eventtype()
370	n/a	results1 = []
371	n/a	results2 = []
372	n/a	N = 5
373	n/a	def f():
374	n/a	results1.append(evt.wait(0.0))
375	n/a	t1 = time.time()
376	n/a	r = evt.wait(0.5)
377	n/a	t2 = time.time()
378	n/a	results2.append((r, t2 - t1))
379	n/a	Bunch(f, N).wait_for_finished()
380	n/a	self.assertEqual(results1, [False] * N)
381	n/a	for r, dt in results2:
382	n/a	self.assertFalse(r)
383	n/a	self.assertTimeout(dt, 0.5)
384	n/a	# The event is set
385	n/a	results1 = []
386	n/a	results2 = []
387	n/a	evt.set()
388	n/a	Bunch(f, N).wait_for_finished()
389	n/a	self.assertEqual(results1, [True] * N)
390	n/a	for r, dt in results2:
391	n/a	self.assertTrue(r)
392	n/a
393	n/a	def test_set_and_clear(self):
394	n/a	# Issue #13502: check that wait() returns true even when the event is
395	n/a	# cleared before the waiting thread is woken up.
396	n/a	evt = self.eventtype()
397	n/a	results = []
398	n/a	N = 5
399	n/a	def f():
400	n/a	results.append(evt.wait(1))
401	n/a	b = Bunch(f, N)
402	n/a	b.wait_for_started()
403	n/a	time.sleep(0.5)
404	n/a	evt.set()
405	n/a	evt.clear()
406	n/a	b.wait_for_finished()
407	n/a	self.assertEqual(results, [True] * N)
408	n/a
409	n/a	def test_reset_internal_locks(self):
410	n/a	# ensure that condition is still using a Lock after reset
411	n/a	evt = self.eventtype()
412	n/a	with evt._cond:
413	n/a	self.assertFalse(evt._cond.acquire(False))
414	n/a	evt._reset_internal_locks()
415	n/a	with evt._cond:
416	n/a	self.assertFalse(evt._cond.acquire(False))
417	n/a
418	n/a
419	n/a	class ConditionTests(BaseTestCase):
420	n/a	"""
421	n/a	Tests for condition variables.
422	n/a	"""
423	n/a
424	n/a	def test_acquire(self):
425	n/a	cond = self.condtype()
426	n/a	# Be default we have an RLock: the condition can be acquired multiple
427	n/a	# times.
428	n/a	cond.acquire()
429	n/a	cond.acquire()
430	n/a	cond.release()
431	n/a	cond.release()
432	n/a	lock = threading.Lock()
433	n/a	cond = self.condtype(lock)
434	n/a	cond.acquire()
435	n/a	self.assertFalse(lock.acquire(False))
436	n/a	cond.release()
437	n/a	self.assertTrue(lock.acquire(False))
438	n/a	self.assertFalse(cond.acquire(False))
439	n/a	lock.release()
440	n/a	with cond:
441	n/a	self.assertFalse(lock.acquire(False))
442	n/a
443	n/a	def test_unacquired_wait(self):
444	n/a	cond = self.condtype()
445	n/a	self.assertRaises(RuntimeError, cond.wait)
446	n/a
447	n/a	def test_unacquired_notify(self):
448	n/a	cond = self.condtype()
449	n/a	self.assertRaises(RuntimeError, cond.notify)
450	n/a
451	n/a	def _check_notify(self, cond):
452	n/a	# Note that this test is sensitive to timing. If the worker threads
453	n/a	# don't execute in a timely fashion, the main thread may think they
454	n/a	# are further along then they are. The main thread therefore issues
455	n/a	# _wait() statements to try to make sure that it doesn't race ahead
456	n/a	# of the workers.
457	n/a	# Secondly, this test assumes that condition variables are not subject
458	n/a	# to spurious wakeups. The absence of spurious wakeups is an implementation
459	n/a	# detail of Condition Cariables in current CPython, but in general, not
460	n/a	# a guaranteed property of condition variables as a programming
461	n/a	# construct. In particular, it is possible that this can no longer
462	n/a	# be conveniently guaranteed should their implementation ever change.
463	n/a	N = 5
464	n/a	results1 = []
465	n/a	results2 = []
466	n/a	phase_num = 0
467	n/a	def f():
468	n/a	cond.acquire()
469	n/a	result = cond.wait()
470	n/a	cond.release()
471	n/a	results1.append((result, phase_num))
472	n/a	cond.acquire()
473	n/a	result = cond.wait()
474	n/a	cond.release()
475	n/a	results2.append((result, phase_num))
476	n/a	b = Bunch(f, N)
477	n/a	b.wait_for_started()
478	n/a	_wait()
479	n/a	self.assertEqual(results1, [])
480	n/a	# Notify 3 threads at first
481	n/a	cond.acquire()
482	n/a	cond.notify(3)
483	n/a	_wait()
484	n/a	phase_num = 1
485	n/a	cond.release()
486	n/a	while len(results1) < 3:
487	n/a	_wait()
488	n/a	self.assertEqual(results1, [(True, 1)] * 3)
489	n/a	self.assertEqual(results2, [])
490	n/a	# first wait, to ensure all workers settle into cond.wait() before
491	n/a	# we continue. See issue #8799
492	n/a	_wait()
493	n/a	# Notify 5 threads: they might be in their first or second wait
494	n/a	cond.acquire()
495	n/a	cond.notify(5)
496	n/a	_wait()
497	n/a	phase_num = 2
498	n/a	cond.release()
499	n/a	while len(results1) + len(results2) < 8:
500	n/a	_wait()
501	n/a	self.assertEqual(results1, [(True, 1)] * 3 + [(True, 2)] * 2)
502	n/a	self.assertEqual(results2, [(True, 2)] * 3)
503	n/a	_wait() # make sure all workers settle into cond.wait()
504	n/a	# Notify all threads: they are all in their second wait
505	n/a	cond.acquire()
506	n/a	cond.notify_all()
507	n/a	_wait()
508	n/a	phase_num = 3
509	n/a	cond.release()
510	n/a	while len(results2) < 5:
511	n/a	_wait()
512	n/a	self.assertEqual(results1, [(True, 1)] * 3 + [(True,2)] * 2)
513	n/a	self.assertEqual(results2, [(True, 2)] * 3 + [(True, 3)] * 2)
514	n/a	b.wait_for_finished()
515	n/a
516	n/a	def test_notify(self):
517	n/a	cond = self.condtype()
518	n/a	self._check_notify(cond)
519	n/a	# A second time, to check internal state is still ok.
520	n/a	self._check_notify(cond)
521	n/a
522	n/a	def test_timeout(self):
523	n/a	cond = self.condtype()
524	n/a	results = []
525	n/a	N = 5
526	n/a	def f():
527	n/a	cond.acquire()
528	n/a	t1 = time.time()
529	n/a	result = cond.wait(0.5)
530	n/a	t2 = time.time()
531	n/a	cond.release()
532	n/a	results.append((t2 - t1, result))
533	n/a	Bunch(f, N).wait_for_finished()
534	n/a	self.assertEqual(len(results), N)
535	n/a	for dt, result in results:
536	n/a	self.assertTimeout(dt, 0.5)
537	n/a	# Note that conceptually (that"s the condition variable protocol)
538	n/a	# a wait() may succeed even if no one notifies us and before any
539	n/a	# timeout occurs. Spurious wakeups can occur.
540	n/a	# This makes it hard to verify the result value.
541	n/a	# In practice, this implementation has no spurious wakeups.
542	n/a	self.assertFalse(result)
543	n/a
544	n/a	def test_waitfor(self):
545	n/a	cond = self.condtype()
546	n/a	state = 0
547	n/a	def f():
548	n/a	with cond:
549	n/a	result = cond.wait_for(lambda : state==4)
550	n/a	self.assertTrue(result)
551	n/a	self.assertEqual(state, 4)
552	n/a	b = Bunch(f, 1)
553	n/a	b.wait_for_started()
554	n/a	for i in range(4):
555	n/a	time.sleep(0.01)
556	n/a	with cond:
557	n/a	state += 1
558	n/a	cond.notify()
559	n/a	b.wait_for_finished()
560	n/a
561	n/a	def test_waitfor_timeout(self):
562	n/a	cond = self.condtype()
563	n/a	state = 0
564	n/a	success = []
565	n/a	def f():
566	n/a	with cond:
567	n/a	dt = time.time()
568	n/a	result = cond.wait_for(lambda : state==4, timeout=0.1)
569	n/a	dt = time.time() - dt
570	n/a	self.assertFalse(result)
571	n/a	self.assertTimeout(dt, 0.1)
572	n/a	success.append(None)
573	n/a	b = Bunch(f, 1)
574	n/a	b.wait_for_started()
575	n/a	# Only increment 3 times, so state == 4 is never reached.
576	n/a	for i in range(3):
577	n/a	time.sleep(0.01)
578	n/a	with cond:
579	n/a	state += 1
580	n/a	cond.notify()
581	n/a	b.wait_for_finished()
582	n/a	self.assertEqual(len(success), 1)
583	n/a
584	n/a
585	n/a	class BaseSemaphoreTests(BaseTestCase):
586	n/a	"""
587	n/a	Common tests for {bounded, unbounded} semaphore objects.
588	n/a	"""
589	n/a
590	n/a	def test_constructor(self):
591	n/a	self.assertRaises(ValueError, self.semtype, value = -1)
592	n/a	self.assertRaises(ValueError, self.semtype, value = -sys.maxsize)
593	n/a
594	n/a	def test_acquire(self):
595	n/a	sem = self.semtype(1)
596	n/a	sem.acquire()
597	n/a	sem.release()
598	n/a	sem = self.semtype(2)
599	n/a	sem.acquire()
600	n/a	sem.acquire()
601	n/a	sem.release()
602	n/a	sem.release()
603	n/a
604	n/a	def test_acquire_destroy(self):
605	n/a	sem = self.semtype()
606	n/a	sem.acquire()
607	n/a	del sem
608	n/a
609	n/a	def test_acquire_contended(self):
610	n/a	sem = self.semtype(7)
611	n/a	sem.acquire()
612	n/a	N = 10
613	n/a	results1 = []
614	n/a	results2 = []
615	n/a	phase_num = 0
616	n/a	def f():
617	n/a	sem.acquire()
618	n/a	results1.append(phase_num)
619	n/a	sem.acquire()
620	n/a	results2.append(phase_num)
621	n/a	b = Bunch(f, 10)
622	n/a	b.wait_for_started()
623	n/a	while len(results1) + len(results2) < 6:
624	n/a	_wait()
625	n/a	self.assertEqual(results1 + results2, [0] * 6)
626	n/a	phase_num = 1
627	n/a	for i in range(7):
628	n/a	sem.release()
629	n/a	while len(results1) + len(results2) < 13:
630	n/a	_wait()
631	n/a	self.assertEqual(sorted(results1 + results2), [0] * 6 + [1] * 7)
632	n/a	phase_num = 2
633	n/a	for i in range(6):
634	n/a	sem.release()
635	n/a	while len(results1) + len(results2) < 19:
636	n/a	_wait()
637	n/a	self.assertEqual(sorted(results1 + results2), [0] * 6 + [1] * 7 + [2] * 6)
638	n/a	# The semaphore is still locked
639	n/a	self.assertFalse(sem.acquire(False))
640	n/a	# Final release, to let the last thread finish
641	n/a	sem.release()
642	n/a	b.wait_for_finished()
643	n/a
644	n/a	def test_try_acquire(self):
645	n/a	sem = self.semtype(2)
646	n/a	self.assertTrue(sem.acquire(False))
647	n/a	self.assertTrue(sem.acquire(False))
648	n/a	self.assertFalse(sem.acquire(False))
649	n/a	sem.release()
650	n/a	self.assertTrue(sem.acquire(False))
651	n/a
652	n/a	def test_try_acquire_contended(self):
653	n/a	sem = self.semtype(4)
654	n/a	sem.acquire()
655	n/a	results = []
656	n/a	def f():
657	n/a	results.append(sem.acquire(False))
658	n/a	results.append(sem.acquire(False))
659	n/a	Bunch(f, 5).wait_for_finished()
660	n/a	# There can be a thread switch between acquiring the semaphore and
661	n/a	# appending the result, therefore results will not necessarily be
662	n/a	# ordered.
663	n/a	self.assertEqual(sorted(results), [False] * 7 + [True] * 3 )
664	n/a
665	n/a	def test_acquire_timeout(self):
666	n/a	sem = self.semtype(2)
667	n/a	self.assertRaises(ValueError, sem.acquire, False, timeout=1.0)
668	n/a	self.assertTrue(sem.acquire(timeout=0.005))
669	n/a	self.assertTrue(sem.acquire(timeout=0.005))
670	n/a	self.assertFalse(sem.acquire(timeout=0.005))
671	n/a	sem.release()
672	n/a	self.assertTrue(sem.acquire(timeout=0.005))
673	n/a	t = time.time()
674	n/a	self.assertFalse(sem.acquire(timeout=0.5))
675	n/a	dt = time.time() - t
676	n/a	self.assertTimeout(dt, 0.5)
677	n/a
678	n/a	def test_default_value(self):
679	n/a	# The default initial value is 1.
680	n/a	sem = self.semtype()
681	n/a	sem.acquire()
682	n/a	def f():
683	n/a	sem.acquire()
684	n/a	sem.release()
685	n/a	b = Bunch(f, 1)
686	n/a	b.wait_for_started()
687	n/a	_wait()
688	n/a	self.assertFalse(b.finished)
689	n/a	sem.release()
690	n/a	b.wait_for_finished()
691	n/a
692	n/a	def test_with(self):
693	n/a	sem = self.semtype(2)
694	n/a	def _with(err=None):
695	n/a	with sem:
696	n/a	self.assertTrue(sem.acquire(False))
697	n/a	sem.release()
698	n/a	with sem:
699	n/a	self.assertFalse(sem.acquire(False))
700	n/a	if err:
701	n/a	raise err
702	n/a	_with()
703	n/a	self.assertTrue(sem.acquire(False))
704	n/a	sem.release()
705	n/a	self.assertRaises(TypeError, _with, TypeError)
706	n/a	self.assertTrue(sem.acquire(False))
707	n/a	sem.release()
708	n/a
709	n/a	class SemaphoreTests(BaseSemaphoreTests):
710	n/a	"""
711	n/a	Tests for unbounded semaphores.
712	n/a	"""
713	n/a
714	n/a	def test_release_unacquired(self):
715	n/a	# Unbounded releases are allowed and increment the semaphore's value
716	n/a	sem = self.semtype(1)
717	n/a	sem.release()
718	n/a	sem.acquire()
719	n/a	sem.acquire()
720	n/a	sem.release()
721	n/a
722	n/a
723	n/a	class BoundedSemaphoreTests(BaseSemaphoreTests):
724	n/a	"""
725	n/a	Tests for bounded semaphores.
726	n/a	"""
727	n/a
728	n/a	def test_release_unacquired(self):
729	n/a	# Cannot go past the initial value
730	n/a	sem = self.semtype()
731	n/a	self.assertRaises(ValueError, sem.release)
732	n/a	sem.acquire()
733	n/a	sem.release()
734	n/a	self.assertRaises(ValueError, sem.release)
735	n/a
736	n/a
737	n/a	class BarrierTests(BaseTestCase):
738	n/a	"""
739	n/a	Tests for Barrier objects.
740	n/a	"""
741	n/a	N = 5
742	n/a	defaultTimeout = 2.0
743	n/a
744	n/a	def setUp(self):
745	n/a	self.barrier = self.barriertype(self.N, timeout=self.defaultTimeout)
746	n/a	def tearDown(self):
747	n/a	self.barrier.abort()
748	n/a
749	n/a	def run_threads(self, f):
750	n/a	b = Bunch(f, self.N-1)
751	n/a	f()
752	n/a	b.wait_for_finished()
753	n/a
754	n/a	def multipass(self, results, n):
755	n/a	m = self.barrier.parties
756	n/a	self.assertEqual(m, self.N)
757	n/a	for i in range(n):
758	n/a	results[0].append(True)
759	n/a	self.assertEqual(len(results[1]), i * m)
760	n/a	self.barrier.wait()
761	n/a	results[1].append(True)
762	n/a	self.assertEqual(len(results[0]), (i + 1) * m)
763	n/a	self.barrier.wait()
764	n/a	self.assertEqual(self.barrier.n_waiting, 0)
765	n/a	self.assertFalse(self.barrier.broken)
766	n/a
767	n/a	def test_barrier(self, passes=1):
768	n/a	"""
769	n/a	Test that a barrier is passed in lockstep
770	n/a	"""
771	n/a	results = [[],[]]
772	n/a	def f():
773	n/a	self.multipass(results, passes)
774	n/a	self.run_threads(f)
775	n/a
776	n/a	def test_barrier_10(self):
777	n/a	"""
778	n/a	Test that a barrier works for 10 consecutive runs
779	n/a	"""
780	n/a	return self.test_barrier(10)
781	n/a
782	n/a	def test_wait_return(self):
783	n/a	"""
784	n/a	test the return value from barrier.wait
785	n/a	"""
786	n/a	results = []
787	n/a	def f():
788	n/a	r = self.barrier.wait()
789	n/a	results.append(r)
790	n/a
791	n/a	self.run_threads(f)
792	n/a	self.assertEqual(sum(results), sum(range(self.N)))
793	n/a
794	n/a	def test_action(self):
795	n/a	"""
796	n/a	Test the 'action' callback
797	n/a	"""
798	n/a	results = []
799	n/a	def action():
800	n/a	results.append(True)
801	n/a	barrier = self.barriertype(self.N, action)
802	n/a	def f():
803	n/a	barrier.wait()
804	n/a	self.assertEqual(len(results), 1)
805	n/a
806	n/a	self.run_threads(f)
807	n/a
808	n/a	def test_abort(self):
809	n/a	"""
810	n/a	Test that an abort will put the barrier in a broken state
811	n/a	"""
812	n/a	results1 = []
813	n/a	results2 = []
814	n/a	def f():
815	n/a	try:
816	n/a	i = self.barrier.wait()
817	n/a	if i == self.N//2:
818	n/a	raise RuntimeError
819	n/a	self.barrier.wait()
820	n/a	results1.append(True)
821	n/a	except threading.BrokenBarrierError:
822	n/a	results2.append(True)
823	n/a	except RuntimeError:
824	n/a	self.barrier.abort()
825	n/a	pass
826	n/a
827	n/a	self.run_threads(f)
828	n/a	self.assertEqual(len(results1), 0)
829	n/a	self.assertEqual(len(results2), self.N-1)
830	n/a	self.assertTrue(self.barrier.broken)
831	n/a
832	n/a	def test_reset(self):
833	n/a	"""
834	n/a	Test that a 'reset' on a barrier frees the waiting threads
835	n/a	"""
836	n/a	results1 = []
837	n/a	results2 = []
838	n/a	results3 = []
839	n/a	def f():
840	n/a	i = self.barrier.wait()
841	n/a	if i == self.N//2:
842	n/a	# Wait until the other threads are all in the barrier.
843	n/a	while self.barrier.n_waiting < self.N-1:
844	n/a	time.sleep(0.001)
845	n/a	self.barrier.reset()
846	n/a	else:
847	n/a	try:
848	n/a	self.barrier.wait()
849	n/a	results1.append(True)
850	n/a	except threading.BrokenBarrierError:
851	n/a	results2.append(True)
852	n/a	# Now, pass the barrier again
853	n/a	self.barrier.wait()
854	n/a	results3.append(True)
855	n/a
856	n/a	self.run_threads(f)
857	n/a	self.assertEqual(len(results1), 0)
858	n/a	self.assertEqual(len(results2), self.N-1)
859	n/a	self.assertEqual(len(results3), self.N)
860	n/a
861	n/a
862	n/a	def test_abort_and_reset(self):
863	n/a	"""
864	n/a	Test that a barrier can be reset after being broken.
865	n/a	"""
866	n/a	results1 = []
867	n/a	results2 = []
868	n/a	results3 = []
869	n/a	barrier2 = self.barriertype(self.N)
870	n/a	def f():
871	n/a	try:
872	n/a	i = self.barrier.wait()
873	n/a	if i == self.N//2:
874	n/a	raise RuntimeError
875	n/a	self.barrier.wait()
876	n/a	results1.append(True)
877	n/a	except threading.BrokenBarrierError:
878	n/a	results2.append(True)
879	n/a	except RuntimeError:
880	n/a	self.barrier.abort()
881	n/a	pass
882	n/a	# Synchronize and reset the barrier. Must synchronize first so
883	n/a	# that everyone has left it when we reset, and after so that no
884	n/a	# one enters it before the reset.
885	n/a	if barrier2.wait() == self.N//2:
886	n/a	self.barrier.reset()
887	n/a	barrier2.wait()
888	n/a	self.barrier.wait()
889	n/a	results3.append(True)
890	n/a
891	n/a	self.run_threads(f)
892	n/a	self.assertEqual(len(results1), 0)
893	n/a	self.assertEqual(len(results2), self.N-1)
894	n/a	self.assertEqual(len(results3), self.N)
895	n/a
896	n/a	def test_timeout(self):
897	n/a	"""
898	n/a	Test wait(timeout)
899	n/a	"""
900	n/a	def f():
901	n/a	i = self.barrier.wait()
902	n/a	if i == self.N // 2:
903	n/a	# One thread is late!
904	n/a	time.sleep(1.0)
905	n/a	# Default timeout is 2.0, so this is shorter.
906	n/a	self.assertRaises(threading.BrokenBarrierError,
907	n/a	self.barrier.wait, 0.5)
908	n/a	self.run_threads(f)
909	n/a
910	n/a	def test_default_timeout(self):
911	n/a	"""
912	n/a	Test the barrier's default timeout
913	n/a	"""
914	n/a	# create a barrier with a low default timeout
915	n/a	barrier = self.barriertype(self.N, timeout=0.3)
916	n/a	def f():
917	n/a	i = barrier.wait()
918	n/a	if i == self.N // 2:
919	n/a	# One thread is later than the default timeout of 0.3s.
920	n/a	time.sleep(1.0)
921	n/a	self.assertRaises(threading.BrokenBarrierError, barrier.wait)
922	n/a	self.run_threads(f)
923	n/a
924	n/a	def test_single_thread(self):
925	n/a	b = self.barriertype(1)
926	n/a	b.wait()
927	n/a	b.wait()