Python code coverage for Lib/test/test_thread.py

#	count	content
1	n/a	import os
2	n/a	import unittest
3	n/a	import random
4	n/a	from test import support
5	n/a	thread = support.import_module('_thread')
6	n/a	import time
7	n/a	import sys
8	n/a	import weakref
9	n/a
10	n/a	from test import lock_tests
11	n/a
12	n/a	NUMTASKS = 10
13	n/a	NUMTRIPS = 3
14	n/a
15	n/a	_print_mutex = thread.allocate_lock()
16	n/a
17	n/a	def verbose_print(arg):
18	n/a	"""Helper function for printing out debugging output."""
19	n/a	if support.verbose:
20	n/a	with _print_mutex:
21	n/a	print(arg)
22	n/a
23	n/a	class BasicThreadTest(unittest.TestCase):
24	n/a
25	n/a	def setUp(self):
26	n/a	self.done_mutex = thread.allocate_lock()
27	n/a	self.done_mutex.acquire()
28	n/a	self.running_mutex = thread.allocate_lock()
29	n/a	self.random_mutex = thread.allocate_lock()
30	n/a	self.created = 0
31	n/a	self.running = 0
32	n/a	self.next_ident = 0
33	n/a
34	n/a
35	n/a	class ThreadRunningTests(BasicThreadTest):
36	n/a
37	n/a	def newtask(self):
38	n/a	with self.running_mutex:
39	n/a	self.next_ident += 1
40	n/a	verbose_print("creating task %s" % self.next_ident)
41	n/a	thread.start_new_thread(self.task, (self.next_ident,))
42	n/a	self.created += 1
43	n/a	self.running += 1
44	n/a
45	n/a	def task(self, ident):
46	n/a	with self.random_mutex:
47	n/a	delay = random.random() / 10000.0
48	n/a	verbose_print("task %s will run for %sus" % (ident, round(delay*1e6)))
49	n/a	time.sleep(delay)
50	n/a	verbose_print("task %s done" % ident)
51	n/a	with self.running_mutex:
52	n/a	self.running -= 1
53	n/a	if self.created == NUMTASKS and self.running == 0:
54	n/a	self.done_mutex.release()
55	n/a
56	n/a	def test_starting_threads(self):
57	n/a	# Basic test for thread creation.
58	n/a	for i in range(NUMTASKS):
59	n/a	self.newtask()
60	n/a	verbose_print("waiting for tasks to complete...")
61	n/a	self.done_mutex.acquire()
62	n/a	verbose_print("all tasks done")
63	n/a
64	n/a	def test_stack_size(self):
65	n/a	# Various stack size tests.
66	n/a	self.assertEqual(thread.stack_size(), 0, "initial stack size is not 0")
67	n/a
68	n/a	thread.stack_size(0)
69	n/a	self.assertEqual(thread.stack_size(), 0, "stack_size not reset to default")
70	n/a
71	n/a	@unittest.skipIf(os.name not in ("nt", "posix"), 'test meant for nt and posix')
72	n/a	def test_nt_and_posix_stack_size(self):
73	n/a	try:
74	n/a	thread.stack_size(4096)
75	n/a	except ValueError:
76	n/a	verbose_print("caught expected ValueError setting "
77	n/a	"stack_size(4096)")
78	n/a	except thread.error:
79	n/a	self.skipTest("platform does not support changing thread stack "
80	n/a	"size")
81	n/a
82	n/a	fail_msg = "stack_size(%d) failed - should succeed"
83	n/a	for tss in (262144, 0x100000, 0):
84	n/a	thread.stack_size(tss)
85	n/a	self.assertEqual(thread.stack_size(), tss, fail_msg % tss)
86	n/a	verbose_print("successfully set stack_size(%d)" % tss)
87	n/a
88	n/a	for tss in (262144, 0x100000):
89	n/a	verbose_print("trying stack_size = (%d)" % tss)
90	n/a	self.next_ident = 0
91	n/a	self.created = 0
92	n/a	for i in range(NUMTASKS):
93	n/a	self.newtask()
94	n/a
95	n/a	verbose_print("waiting for all tasks to complete")
96	n/a	self.done_mutex.acquire()
97	n/a	verbose_print("all tasks done")
98	n/a
99	n/a	thread.stack_size(0)
100	n/a
101	n/a	def test__count(self):
102	n/a	# Test the _count() function.
103	n/a	orig = thread._count()
104	n/a	mut = thread.allocate_lock()
105	n/a	mut.acquire()
106	n/a	started = []
107	n/a	def task():
108	n/a	started.append(None)
109	n/a	mut.acquire()
110	n/a	mut.release()
111	n/a	thread.start_new_thread(task, ())
112	n/a	while not started:
113	n/a	time.sleep(0.01)
114	n/a	self.assertEqual(thread._count(), orig + 1)
115	n/a	# Allow the task to finish.
116	n/a	mut.release()
117	n/a	# The only reliable way to be sure that the thread ended from the
118	n/a	# interpreter's point of view is to wait for the function object to be
119	n/a	# destroyed.
120	n/a	done = []
121	n/a	wr = weakref.ref(task, lambda _: done.append(None))
122	n/a	del task
123	n/a	while not done:
124	n/a	time.sleep(0.01)
125	n/a	self.assertEqual(thread._count(), orig)
126	n/a
127	n/a	def test_save_exception_state_on_error(self):
128	n/a	# See issue #14474
129	n/a	def task():
130	n/a	started.release()
131	n/a	raise SyntaxError
132	n/a	def mywrite(self, *args):
133	n/a	try:
134	n/a	raise ValueError
135	n/a	except ValueError:
136	n/a	pass
137	n/a	real_write(self, *args)
138	n/a	c = thread._count()
139	n/a	started = thread.allocate_lock()
140	n/a	with support.captured_output("stderr") as stderr:
141	n/a	real_write = stderr.write
142	n/a	stderr.write = mywrite
143	n/a	started.acquire()
144	n/a	thread.start_new_thread(task, ())
145	n/a	started.acquire()
146	n/a	while thread._count() > c:
147	n/a	time.sleep(0.01)
148	n/a	self.assertIn("Traceback", stderr.getvalue())
149	n/a
150	n/a
151	n/a	class Barrier:
152	n/a	def __init__(self, num_threads):
153	n/a	self.num_threads = num_threads
154	n/a	self.waiting = 0
155	n/a	self.checkin_mutex = thread.allocate_lock()
156	n/a	self.checkout_mutex = thread.allocate_lock()
157	n/a	self.checkout_mutex.acquire()
158	n/a
159	n/a	def enter(self):
160	n/a	self.checkin_mutex.acquire()
161	n/a	self.waiting = self.waiting + 1
162	n/a	if self.waiting == self.num_threads:
163	n/a	self.waiting = self.num_threads - 1
164	n/a	self.checkout_mutex.release()
165	n/a	return
166	n/a	self.checkin_mutex.release()
167	n/a
168	n/a	self.checkout_mutex.acquire()
169	n/a	self.waiting = self.waiting - 1
170	n/a	if self.waiting == 0:
171	n/a	self.checkin_mutex.release()
172	n/a	return
173	n/a	self.checkout_mutex.release()
174	n/a
175	n/a
176	n/a	class BarrierTest(BasicThreadTest):
177	n/a
178	n/a	def test_barrier(self):
179	n/a	self.bar = Barrier(NUMTASKS)
180	n/a	self.running = NUMTASKS
181	n/a	for i in range(NUMTASKS):
182	n/a	thread.start_new_thread(self.task2, (i,))
183	n/a	verbose_print("waiting for tasks to end")
184	n/a	self.done_mutex.acquire()
185	n/a	verbose_print("tasks done")
186	n/a
187	n/a	def task2(self, ident):
188	n/a	for i in range(NUMTRIPS):
189	n/a	if ident == 0:
190	n/a	# give it a good chance to enter the next
191	n/a	# barrier before the others are all out
192	n/a	# of the current one
193	n/a	delay = 0
194	n/a	else:
195	n/a	with self.random_mutex:
196	n/a	delay = random.random() / 10000.0
197	n/a	verbose_print("task %s will run for %sus" %
198	n/a	(ident, round(delay * 1e6)))
199	n/a	time.sleep(delay)
200	n/a	verbose_print("task %s entering %s" % (ident, i))
201	n/a	self.bar.enter()
202	n/a	verbose_print("task %s leaving barrier" % ident)
203	n/a	with self.running_mutex:
204	n/a	self.running -= 1
205	n/a	# Must release mutex before releasing done, else the main thread can
206	n/a	# exit and set mutex to None as part of global teardown; then
207	n/a	# mutex.release() raises AttributeError.
208	n/a	finished = self.running == 0
209	n/a	if finished:
210	n/a	self.done_mutex.release()
211	n/a
212	n/a	class LockTests(lock_tests.LockTests):
213	n/a	locktype = thread.allocate_lock
214	n/a
215	n/a
216	n/a	class TestForkInThread(unittest.TestCase):
217	n/a	def setUp(self):
218	n/a	self.read_fd, self.write_fd = os.pipe()
219	n/a
220	n/a	@unittest.skipIf(sys.platform.startswith('win'),
221	n/a	"This test is only appropriate for POSIX-like systems.")
222	n/a	@support.reap_threads
223	n/a	def test_forkinthread(self):
224	n/a	def thread1():
225	n/a	try:
226	n/a	pid = os.fork() # fork in a thread
227	n/a	except RuntimeError:
228	n/a	os._exit(1) # exit the child
229	n/a
230	n/a	if pid == 0: # child
231	n/a	try:
232	n/a	os.close(self.read_fd)
233	n/a	os.write(self.write_fd, b"OK")
234	n/a	finally:
235	n/a	os._exit(0)
236	n/a	else: # parent
237	n/a	os.close(self.write_fd)
238	n/a
239	n/a	thread.start_new_thread(thread1, ())
240	n/a	self.assertEqual(os.read(self.read_fd, 2), b"OK",
241	n/a	"Unable to fork() in thread")
242	n/a
243	n/a	def tearDown(self):
244	n/a	try:
245	n/a	os.close(self.read_fd)
246	n/a	except OSError:
247	n/a	pass
248	n/a
249	n/a	try:
250	n/a	os.close(self.write_fd)
251	n/a	except OSError:
252	n/a	pass
253	n/a
254	n/a
255	n/a	if __name__ == "__main__":
256	n/a	unittest.main()