Python code coverage for Lib/test/test_threadsignals.py

#	count	content
1	n/a	"""PyUnit testing that threads honor our signal semantics"""
2	n/a
3	n/a	import unittest
4	n/a	import signal
5	n/a	import os
6	n/a	import sys
7	n/a	from test.support import run_unittest, import_module
8	n/a	thread = import_module('_thread')
9	n/a	import time
10	n/a
11	n/a	if (sys.platform[:3] == 'win'):
12	n/a	raise unittest.SkipTest("Can't test signal on %s" % sys.platform)
13	n/a
14	n/a	process_pid = os.getpid()
15	n/a	signalled_all=thread.allocate_lock()
16	n/a
17	n/a	USING_PTHREAD_COND = (sys.thread_info.name == 'pthread'
18	n/a	and sys.thread_info.lock == 'mutex+cond')
19	n/a
20	n/a	def registerSignals(for_usr1, for_usr2, for_alrm):
21	n/a	usr1 = signal.signal(signal.SIGUSR1, for_usr1)
22	n/a	usr2 = signal.signal(signal.SIGUSR2, for_usr2)
23	n/a	alrm = signal.signal(signal.SIGALRM, for_alrm)
24	n/a	return usr1, usr2, alrm
25	n/a
26	n/a
27	n/a	# The signal handler. Just note that the signal occurred and
28	n/a	# from who.
29	n/a	def handle_signals(sig,frame):
30	n/a	signal_blackboard[sig]['tripped'] += 1
31	n/a	signal_blackboard[sig]['tripped_by'] = thread.get_ident()
32	n/a
33	n/a	# a function that will be spawned as a separate thread.
34	n/a	def send_signals():
35	n/a	os.kill(process_pid, signal.SIGUSR1)
36	n/a	os.kill(process_pid, signal.SIGUSR2)
37	n/a	signalled_all.release()
38	n/a
39	n/a	class ThreadSignals(unittest.TestCase):
40	n/a
41	n/a	def test_signals(self):
42	n/a	# Test signal handling semantics of threads.
43	n/a	# We spawn a thread, have the thread send two signals, and
44	n/a	# wait for it to finish. Check that we got both signals
45	n/a	# and that they were run by the main thread.
46	n/a	signalled_all.acquire()
47	n/a	self.spawnSignallingThread()
48	n/a	signalled_all.acquire()
49	n/a	# the signals that we asked the kernel to send
50	n/a	# will come back, but we don't know when.
51	n/a	# (it might even be after the thread exits
52	n/a	# and might be out of order.) If we haven't seen
53	n/a	# the signals yet, send yet another signal and
54	n/a	# wait for it return.
55	n/a	if signal_blackboard[signal.SIGUSR1]['tripped'] == 0 \
56	n/a	or signal_blackboard[signal.SIGUSR2]['tripped'] == 0:
57	n/a	signal.alarm(1)
58	n/a	signal.pause()
59	n/a	signal.alarm(0)
60	n/a
61	n/a	self.assertEqual( signal_blackboard[signal.SIGUSR1]['tripped'], 1)
62	n/a	self.assertEqual( signal_blackboard[signal.SIGUSR1]['tripped_by'],
63	n/a	thread.get_ident())
64	n/a	self.assertEqual( signal_blackboard[signal.SIGUSR2]['tripped'], 1)
65	n/a	self.assertEqual( signal_blackboard[signal.SIGUSR2]['tripped_by'],
66	n/a	thread.get_ident())
67	n/a	signalled_all.release()
68	n/a
69	n/a	def spawnSignallingThread(self):
70	n/a	thread.start_new_thread(send_signals, ())
71	n/a
72	n/a	def alarm_interrupt(self, sig, frame):
73	n/a	raise KeyboardInterrupt
74	n/a
75	n/a	@unittest.skipIf(USING_PTHREAD_COND,
76	n/a	'POSIX condition variables cannot be interrupted')
77	n/a	# Issue #20564: sem_timedwait() cannot be interrupted on OpenBSD
78	n/a	@unittest.skipIf(sys.platform.startswith('openbsd'),
79	n/a	'lock cannot be interrupted on OpenBSD')
80	n/a	def test_lock_acquire_interruption(self):
81	n/a	# Mimic receiving a SIGINT (KeyboardInterrupt) with SIGALRM while stuck
82	n/a	# in a deadlock.
83	n/a	# XXX this test can fail when the legacy (non-semaphore) implementation
84	n/a	# of locks is used in thread_pthread.h, see issue #11223.
85	n/a	oldalrm = signal.signal(signal.SIGALRM, self.alarm_interrupt)
86	n/a	try:
87	n/a	lock = thread.allocate_lock()
88	n/a	lock.acquire()
89	n/a	signal.alarm(1)
90	n/a	t1 = time.time()
91	n/a	self.assertRaises(KeyboardInterrupt, lock.acquire, timeout=5)
92	n/a	dt = time.time() - t1
93	n/a	# Checking that KeyboardInterrupt was raised is not sufficient.
94	n/a	# We want to assert that lock.acquire() was interrupted because
95	n/a	# of the signal, not that the signal handler was called immediately
96	n/a	# after timeout return of lock.acquire() (which can fool assertRaises).
97	n/a	self.assertLess(dt, 3.0)
98	n/a	finally:
99	n/a	signal.signal(signal.SIGALRM, oldalrm)
100	n/a
101	n/a	@unittest.skipIf(USING_PTHREAD_COND,
102	n/a	'POSIX condition variables cannot be interrupted')
103	n/a	# Issue #20564: sem_timedwait() cannot be interrupted on OpenBSD
104	n/a	@unittest.skipIf(sys.platform.startswith('openbsd'),
105	n/a	'lock cannot be interrupted on OpenBSD')
106	n/a	def test_rlock_acquire_interruption(self):
107	n/a	# Mimic receiving a SIGINT (KeyboardInterrupt) with SIGALRM while stuck
108	n/a	# in a deadlock.
109	n/a	# XXX this test can fail when the legacy (non-semaphore) implementation
110	n/a	# of locks is used in thread_pthread.h, see issue #11223.
111	n/a	oldalrm = signal.signal(signal.SIGALRM, self.alarm_interrupt)
112	n/a	try:
113	n/a	rlock = thread.RLock()
114	n/a	# For reentrant locks, the initial acquisition must be in another
115	n/a	# thread.
116	n/a	def other_thread():
117	n/a	rlock.acquire()
118	n/a	thread.start_new_thread(other_thread, ())
119	n/a	# Wait until we can't acquire it without blocking...
120	n/a	while rlock.acquire(blocking=False):
121	n/a	rlock.release()
122	n/a	time.sleep(0.01)
123	n/a	signal.alarm(1)
124	n/a	t1 = time.time()
125	n/a	self.assertRaises(KeyboardInterrupt, rlock.acquire, timeout=5)
126	n/a	dt = time.time() - t1
127	n/a	# See rationale above in test_lock_acquire_interruption
128	n/a	self.assertLess(dt, 3.0)
129	n/a	finally:
130	n/a	signal.signal(signal.SIGALRM, oldalrm)
131	n/a
132	n/a	def acquire_retries_on_intr(self, lock):
133	n/a	self.sig_recvd = False
134	n/a	def my_handler(signal, frame):
135	n/a	self.sig_recvd = True
136	n/a	old_handler = signal.signal(signal.SIGUSR1, my_handler)
137	n/a	try:
138	n/a	def other_thread():
139	n/a	# Acquire the lock in a non-main thread, so this test works for
140	n/a	# RLocks.
141	n/a	lock.acquire()
142	n/a	# Wait until the main thread is blocked in the lock acquire, and
143	n/a	# then wake it up with this.
144	n/a	time.sleep(0.5)
145	n/a	os.kill(process_pid, signal.SIGUSR1)
146	n/a	# Let the main thread take the interrupt, handle it, and retry
147	n/a	# the lock acquisition. Then we'll let it run.
148	n/a	time.sleep(0.5)
149	n/a	lock.release()
150	n/a	thread.start_new_thread(other_thread, ())
151	n/a	# Wait until we can't acquire it without blocking...
152	n/a	while lock.acquire(blocking=False):
153	n/a	lock.release()
154	n/a	time.sleep(0.01)
155	n/a	result = lock.acquire() # Block while we receive a signal.
156	n/a	self.assertTrue(self.sig_recvd)
157	n/a	self.assertTrue(result)
158	n/a	finally:
159	n/a	signal.signal(signal.SIGUSR1, old_handler)
160	n/a
161	n/a	def test_lock_acquire_retries_on_intr(self):
162	n/a	self.acquire_retries_on_intr(thread.allocate_lock())
163	n/a
164	n/a	def test_rlock_acquire_retries_on_intr(self):
165	n/a	self.acquire_retries_on_intr(thread.RLock())
166	n/a
167	n/a	def test_interrupted_timed_acquire(self):
168	n/a	# Test to make sure we recompute lock acquisition timeouts when we
169	n/a	# receive a signal. Check this by repeatedly interrupting a lock
170	n/a	# acquire in the main thread, and make sure that the lock acquire times
171	n/a	# out after the right amount of time.
172	n/a	# NOTE: this test only behaves as expected if C signals get delivered
173	n/a	# to the main thread. Otherwise lock.acquire() itself doesn't get
174	n/a	# interrupted and the test trivially succeeds.
175	n/a	self.start = None
176	n/a	self.end = None
177	n/a	self.sigs_recvd = 0
178	n/a	done = thread.allocate_lock()
179	n/a	done.acquire()
180	n/a	lock = thread.allocate_lock()
181	n/a	lock.acquire()
182	n/a	def my_handler(signum, frame):
183	n/a	self.sigs_recvd += 1
184	n/a	old_handler = signal.signal(signal.SIGUSR1, my_handler)
185	n/a	try:
186	n/a	def timed_acquire():
187	n/a	self.start = time.time()
188	n/a	lock.acquire(timeout=0.5)
189	n/a	self.end = time.time()
190	n/a	def send_signals():
191	n/a	for _ in range(40):
192	n/a	time.sleep(0.02)
193	n/a	os.kill(process_pid, signal.SIGUSR1)
194	n/a	done.release()
195	n/a
196	n/a	# Send the signals from the non-main thread, since the main thread
197	n/a	# is the only one that can process signals.
198	n/a	thread.start_new_thread(send_signals, ())
199	n/a	timed_acquire()
200	n/a	# Wait for thread to finish
201	n/a	done.acquire()
202	n/a	# This allows for some timing and scheduling imprecision
203	n/a	self.assertLess(self.end - self.start, 2.0)
204	n/a	self.assertGreater(self.end - self.start, 0.3)
205	n/a	# If the signal is received several times before PyErr_CheckSignals()
206	n/a	# is called, the handler will get called less than 40 times. Just
207	n/a	# check it's been called at least once.
208	n/a	self.assertGreater(self.sigs_recvd, 0)
209	n/a	finally:
210	n/a	signal.signal(signal.SIGUSR1, old_handler)
211	n/a
212	n/a
213	n/a	def test_main():
214	n/a	global signal_blackboard
215	n/a
216	n/a	signal_blackboard = { signal.SIGUSR1 : {'tripped': 0, 'tripped_by': 0 },
217	n/a	signal.SIGUSR2 : {'tripped': 0, 'tripped_by': 0 },
218	n/a	signal.SIGALRM : {'tripped': 0, 'tripped_by': 0 } }
219	n/a
220	n/a	oldsigs = registerSignals(handle_signals, handle_signals, handle_signals)
221	n/a	try:
222	n/a	run_unittest(ThreadSignals)
223	n/a	finally:
224	n/a	registerSignals(*oldsigs)
225	n/a
226	n/a	if __name__ == '__main__':
227	n/a	test_main()