blob: 11e63d3e8304af4269f5b5ebc7fc56f46c66c30f [file] [log] [blame]
Skip Montanaro4533f602001-08-20 20:28:48 +00001# Very rudimentary test of threading module
2
Benjamin Petersonee8712c2008-05-20 21:35:26 +00003import test.support
Antoine Pitrouc4d78642011-05-05 20:17:32 +02004from test.support import verbose, strip_python_stderr, import_module
Antoine Pitrou8e6e0fd2012-04-19 23:55:01 +02005from test.script_helper import assert_python_ok
6
Skip Montanaro4533f602001-08-20 20:28:48 +00007import random
Georg Brandl0c77a822008-06-10 16:37:50 +00008import re
Guido van Rossumcd16bf62007-06-13 18:07:49 +00009import sys
Antoine Pitrouc4d78642011-05-05 20:17:32 +020010_thread = import_module('_thread')
11threading = import_module('threading')
Skip Montanaro4533f602001-08-20 20:28:48 +000012import time
Tim Peters84d54892005-01-08 06:03:17 +000013import unittest
Christian Heimesd3eb5a152008-02-24 00:38:49 +000014import weakref
Alexandre Vassalotti93f2cd22009-07-22 04:54:52 +000015import os
Antoine Pitrouc4d78642011-05-05 20:17:32 +020016from test.script_helper import assert_python_ok, assert_python_failure
Gregory P. Smith4b129d22011-01-04 00:51:50 +000017import subprocess
Skip Montanaro4533f602001-08-20 20:28:48 +000018
Antoine Pitrou557934f2009-11-06 22:41:14 +000019from test import lock_tests
20
Tim Peters84d54892005-01-08 06:03:17 +000021# A trivial mutable counter.
22class Counter(object):
23 def __init__(self):
24 self.value = 0
25 def inc(self):
26 self.value += 1
27 def dec(self):
28 self.value -= 1
29 def get(self):
30 return self.value
Skip Montanaro4533f602001-08-20 20:28:48 +000031
32class TestThread(threading.Thread):
Tim Peters84d54892005-01-08 06:03:17 +000033 def __init__(self, name, testcase, sema, mutex, nrunning):
34 threading.Thread.__init__(self, name=name)
35 self.testcase = testcase
36 self.sema = sema
37 self.mutex = mutex
38 self.nrunning = nrunning
39
Skip Montanaro4533f602001-08-20 20:28:48 +000040 def run(self):
Christian Heimes4fbc72b2008-03-22 00:47:35 +000041 delay = random.random() / 10000.0
Skip Montanaro4533f602001-08-20 20:28:48 +000042 if verbose:
Jeffrey Yasskinca674122008-03-29 05:06:52 +000043 print('task %s will run for %.1f usec' %
Benjamin Petersonfdbea962008-08-18 17:33:47 +000044 (self.name, delay * 1e6))
Tim Peters84d54892005-01-08 06:03:17 +000045
Christian Heimes4fbc72b2008-03-22 00:47:35 +000046 with self.sema:
47 with self.mutex:
48 self.nrunning.inc()
49 if verbose:
50 print(self.nrunning.get(), 'tasks are running')
Benjamin Petersonc9c0f202009-06-30 23:06:06 +000051 self.testcase.assertTrue(self.nrunning.get() <= 3)
Tim Peters84d54892005-01-08 06:03:17 +000052
Christian Heimes4fbc72b2008-03-22 00:47:35 +000053 time.sleep(delay)
54 if verbose:
Benjamin Petersonfdbea962008-08-18 17:33:47 +000055 print('task', self.name, 'done')
Benjamin Peterson672b8032008-06-11 19:14:14 +000056
Christian Heimes4fbc72b2008-03-22 00:47:35 +000057 with self.mutex:
58 self.nrunning.dec()
Benjamin Petersonc9c0f202009-06-30 23:06:06 +000059 self.testcase.assertTrue(self.nrunning.get() >= 0)
Christian Heimes4fbc72b2008-03-22 00:47:35 +000060 if verbose:
61 print('%s is finished. %d tasks are running' %
Benjamin Petersonfdbea962008-08-18 17:33:47 +000062 (self.name, self.nrunning.get()))
Benjamin Peterson672b8032008-06-11 19:14:14 +000063
Skip Montanaro4533f602001-08-20 20:28:48 +000064
Antoine Pitroub0e9bd42009-10-27 20:05:26 +000065class BaseTestCase(unittest.TestCase):
66 def setUp(self):
67 self._threads = test.support.threading_setup()
68
69 def tearDown(self):
70 test.support.threading_cleanup(*self._threads)
71 test.support.reap_children()
72
73
74class ThreadTests(BaseTestCase):
Skip Montanaro4533f602001-08-20 20:28:48 +000075
Tim Peters84d54892005-01-08 06:03:17 +000076 # Create a bunch of threads, let each do some work, wait until all are
77 # done.
78 def test_various_ops(self):
79 # This takes about n/3 seconds to run (about n/3 clumps of tasks,
80 # times about 1 second per clump).
81 NUMTASKS = 10
82
83 # no more than 3 of the 10 can run at once
84 sema = threading.BoundedSemaphore(value=3)
85 mutex = threading.RLock()
86 numrunning = Counter()
87
88 threads = []
89
90 for i in range(NUMTASKS):
91 t = TestThread("<thread %d>"%i, self, sema, mutex, numrunning)
92 threads.append(t)
Benjamin Petersonc9c0f202009-06-30 23:06:06 +000093 self.assertEqual(t.ident, None)
94 self.assertTrue(re.match('<TestThread\(.*, initial\)>', repr(t)))
Tim Peters84d54892005-01-08 06:03:17 +000095 t.start()
96
97 if verbose:
Guido van Rossumbe19ed72007-02-09 05:37:30 +000098 print('waiting for all tasks to complete')
Tim Peters84d54892005-01-08 06:03:17 +000099 for t in threads:
Tim Peters711906e2005-01-08 07:30:42 +0000100 t.join(NUMTASKS)
Benjamin Petersonc9c0f202009-06-30 23:06:06 +0000101 self.assertTrue(not t.is_alive())
102 self.assertNotEqual(t.ident, 0)
Benjamin Petersond23f8222009-04-05 19:13:16 +0000103 self.assertFalse(t.ident is None)
Brett Cannon3f5f2262010-07-23 15:50:52 +0000104 self.assertTrue(re.match('<TestThread\(.*, stopped -?\d+\)>',
105 repr(t)))
Tim Peters84d54892005-01-08 06:03:17 +0000106 if verbose:
Guido van Rossumbe19ed72007-02-09 05:37:30 +0000107 print('all tasks done')
Tim Peters84d54892005-01-08 06:03:17 +0000108 self.assertEqual(numrunning.get(), 0)
109
Benjamin Petersond23f8222009-04-05 19:13:16 +0000110 def test_ident_of_no_threading_threads(self):
111 # The ident still must work for the main thread and dummy threads.
112 self.assertFalse(threading.currentThread().ident is None)
113 def f():
114 ident.append(threading.currentThread().ident)
115 done.set()
116 done = threading.Event()
117 ident = []
118 _thread.start_new_thread(f, ())
119 done.wait()
120 self.assertFalse(ident[0] is None)
Antoine Pitrouca13a0d2009-11-08 00:30:04 +0000121 # Kill the "immortal" _DummyThread
122 del threading._active[ident[0]]
Benjamin Petersond23f8222009-04-05 19:13:16 +0000123
Thomas Wouters0e3f5912006-08-11 14:57:12 +0000124 # run with a small(ish) thread stack size (256kB)
125 def test_various_ops_small_stack(self):
126 if verbose:
Guido van Rossumbe19ed72007-02-09 05:37:30 +0000127 print('with 256kB thread stack size...')
Thomas Wouters0e3f5912006-08-11 14:57:12 +0000128 try:
129 threading.stack_size(262144)
Georg Brandl2067bfd2008-05-25 13:05:15 +0000130 except _thread.error:
Alexandre Vassalotti93f2cd22009-07-22 04:54:52 +0000131 raise unittest.SkipTest(
132 'platform does not support changing thread stack size')
Thomas Wouters0e3f5912006-08-11 14:57:12 +0000133 self.test_various_ops()
134 threading.stack_size(0)
135
136 # run with a large thread stack size (1MB)
137 def test_various_ops_large_stack(self):
138 if verbose:
Guido van Rossumbe19ed72007-02-09 05:37:30 +0000139 print('with 1MB thread stack size...')
Thomas Wouters0e3f5912006-08-11 14:57:12 +0000140 try:
141 threading.stack_size(0x100000)
Georg Brandl2067bfd2008-05-25 13:05:15 +0000142 except _thread.error:
Alexandre Vassalotti93f2cd22009-07-22 04:54:52 +0000143 raise unittest.SkipTest(
144 'platform does not support changing thread stack size')
Thomas Wouters0e3f5912006-08-11 14:57:12 +0000145 self.test_various_ops()
146 threading.stack_size(0)
147
Tim Peters711906e2005-01-08 07:30:42 +0000148 def test_foreign_thread(self):
149 # Check that a "foreign" thread can use the threading module.
150 def f(mutex):
Antoine Pitroub0872682009-11-09 16:08:16 +0000151 # Calling current_thread() forces an entry for the foreign
Tim Peters711906e2005-01-08 07:30:42 +0000152 # thread to get made in the threading._active map.
Antoine Pitroub0872682009-11-09 16:08:16 +0000153 threading.current_thread()
Tim Peters711906e2005-01-08 07:30:42 +0000154 mutex.release()
155
156 mutex = threading.Lock()
157 mutex.acquire()
Georg Brandl2067bfd2008-05-25 13:05:15 +0000158 tid = _thread.start_new_thread(f, (mutex,))
Tim Peters711906e2005-01-08 07:30:42 +0000159 # Wait for the thread to finish.
160 mutex.acquire()
Benjamin Peterson577473f2010-01-19 00:09:57 +0000161 self.assertIn(tid, threading._active)
Ezio Melottie9615932010-01-24 19:26:24 +0000162 self.assertIsInstance(threading._active[tid], threading._DummyThread)
Tim Peters711906e2005-01-08 07:30:42 +0000163 del threading._active[tid]
Tim Peters84d54892005-01-08 06:03:17 +0000164
Thomas Wouters00ee7ba2006-08-21 19:07:27 +0000165 # PyThreadState_SetAsyncExc() is a CPython-only gimmick, not (currently)
166 # exposed at the Python level. This test relies on ctypes to get at it.
167 def test_PyThreadState_SetAsyncExc(self):
Antoine Pitrouc4d78642011-05-05 20:17:32 +0200168 ctypes = import_module("ctypes")
Thomas Wouters00ee7ba2006-08-21 19:07:27 +0000169
170 set_async_exc = ctypes.pythonapi.PyThreadState_SetAsyncExc
171
172 class AsyncExc(Exception):
173 pass
174
175 exception = ctypes.py_object(AsyncExc)
176
Antoine Pitroube4d8092009-10-18 18:27:17 +0000177 # First check it works when setting the exception from the same thread.
Victor Stinner2a129742011-05-30 23:02:52 +0200178 tid = threading.get_ident()
Antoine Pitroube4d8092009-10-18 18:27:17 +0000179
180 try:
181 result = set_async_exc(ctypes.c_long(tid), exception)
182 # The exception is async, so we might have to keep the VM busy until
183 # it notices.
184 while True:
185 pass
186 except AsyncExc:
187 pass
188 else:
Benjamin Petersona0dfa822009-11-13 02:25:08 +0000189 # This code is unreachable but it reflects the intent. If we wanted
190 # to be smarter the above loop wouldn't be infinite.
Antoine Pitroube4d8092009-10-18 18:27:17 +0000191 self.fail("AsyncExc not raised")
192 try:
193 self.assertEqual(result, 1) # one thread state modified
194 except UnboundLocalError:
Benjamin Petersona0dfa822009-11-13 02:25:08 +0000195 # The exception was raised too quickly for us to get the result.
Antoine Pitroube4d8092009-10-18 18:27:17 +0000196 pass
197
Thomas Wouters00ee7ba2006-08-21 19:07:27 +0000198 # `worker_started` is set by the thread when it's inside a try/except
199 # block waiting to catch the asynchronously set AsyncExc exception.
200 # `worker_saw_exception` is set by the thread upon catching that
201 # exception.
202 worker_started = threading.Event()
203 worker_saw_exception = threading.Event()
204
205 class Worker(threading.Thread):
206 def run(self):
Victor Stinner2a129742011-05-30 23:02:52 +0200207 self.id = threading.get_ident()
Thomas Wouters00ee7ba2006-08-21 19:07:27 +0000208 self.finished = False
209
210 try:
211 while True:
212 worker_started.set()
213 time.sleep(0.1)
214 except AsyncExc:
215 self.finished = True
216 worker_saw_exception.set()
217
218 t = Worker()
Benjamin Petersonfdbea962008-08-18 17:33:47 +0000219 t.daemon = True # so if this fails, we don't hang Python at shutdown
Thomas Wouters00ee7ba2006-08-21 19:07:27 +0000220 t.start()
221 if verbose:
Guido van Rossumbe19ed72007-02-09 05:37:30 +0000222 print(" started worker thread")
Thomas Wouters00ee7ba2006-08-21 19:07:27 +0000223
224 # Try a thread id that doesn't make sense.
225 if verbose:
Guido van Rossumbe19ed72007-02-09 05:37:30 +0000226 print(" trying nonsensical thread id")
Thomas Wouters00ee7ba2006-08-21 19:07:27 +0000227 result = set_async_exc(ctypes.c_long(-1), exception)
228 self.assertEqual(result, 0) # no thread states modified
229
230 # Now raise an exception in the worker thread.
231 if verbose:
Guido van Rossumbe19ed72007-02-09 05:37:30 +0000232 print(" waiting for worker thread to get started")
Benjamin Petersond23f8222009-04-05 19:13:16 +0000233 ret = worker_started.wait()
234 self.assertTrue(ret)
Thomas Wouters00ee7ba2006-08-21 19:07:27 +0000235 if verbose:
Guido van Rossumbe19ed72007-02-09 05:37:30 +0000236 print(" verifying worker hasn't exited")
Benjamin Petersonc9c0f202009-06-30 23:06:06 +0000237 self.assertTrue(not t.finished)
Thomas Wouters00ee7ba2006-08-21 19:07:27 +0000238 if verbose:
Guido van Rossumbe19ed72007-02-09 05:37:30 +0000239 print(" attempting to raise asynch exception in worker")
Thomas Wouters00ee7ba2006-08-21 19:07:27 +0000240 result = set_async_exc(ctypes.c_long(t.id), exception)
241 self.assertEqual(result, 1) # one thread state modified
242 if verbose:
Guido van Rossumbe19ed72007-02-09 05:37:30 +0000243 print(" waiting for worker to say it caught the exception")
Thomas Wouters00ee7ba2006-08-21 19:07:27 +0000244 worker_saw_exception.wait(timeout=10)
Benjamin Petersonc9c0f202009-06-30 23:06:06 +0000245 self.assertTrue(t.finished)
Thomas Wouters00ee7ba2006-08-21 19:07:27 +0000246 if verbose:
Guido van Rossumbe19ed72007-02-09 05:37:30 +0000247 print(" all OK -- joining worker")
Thomas Wouters00ee7ba2006-08-21 19:07:27 +0000248 if t.finished:
249 t.join()
250 # else the thread is still running, and we have no way to kill it
251
Gregory P. Smith3fdd9642010-02-28 18:57:46 +0000252 def test_limbo_cleanup(self):
253 # Issue 7481: Failure to start thread should cleanup the limbo map.
254 def fail_new_thread(*args):
255 raise threading.ThreadError()
256 _start_new_thread = threading._start_new_thread
257 threading._start_new_thread = fail_new_thread
258 try:
259 t = threading.Thread(target=lambda: None)
Gregory P. Smithf50f1682010-03-01 03:13:36 +0000260 self.assertRaises(threading.ThreadError, t.start)
261 self.assertFalse(
262 t in threading._limbo,
263 "Failed to cleanup _limbo map on failure of Thread.start().")
Gregory P. Smith3fdd9642010-02-28 18:57:46 +0000264 finally:
265 threading._start_new_thread = _start_new_thread
266
Christian Heimes7d2ff882007-11-30 14:35:04 +0000267 def test_finalize_runnning_thread(self):
268 # Issue 1402: the PyGILState_Ensure / _Release functions may be called
269 # very late on python exit: on deallocation of a running thread for
270 # example.
Antoine Pitrouc4d78642011-05-05 20:17:32 +0200271 import_module("ctypes")
Christian Heimes7d2ff882007-11-30 14:35:04 +0000272
Antoine Pitrouc4d78642011-05-05 20:17:32 +0200273 rc, out, err = assert_python_failure("-c", """if 1:
Georg Brandl2067bfd2008-05-25 13:05:15 +0000274 import ctypes, sys, time, _thread
Christian Heimes7d2ff882007-11-30 14:35:04 +0000275
Christian Heimes4fbc72b2008-03-22 00:47:35 +0000276 # This lock is used as a simple event variable.
Georg Brandl2067bfd2008-05-25 13:05:15 +0000277 ready = _thread.allocate_lock()
Christian Heimes4fbc72b2008-03-22 00:47:35 +0000278 ready.acquire()
279
Christian Heimes7d2ff882007-11-30 14:35:04 +0000280 # Module globals are cleared before __del__ is run
281 # So we save the functions in class dict
282 class C:
283 ensure = ctypes.pythonapi.PyGILState_Ensure
284 release = ctypes.pythonapi.PyGILState_Release
285 def __del__(self):
286 state = self.ensure()
287 self.release(state)
288
289 def waitingThread():
290 x = C()
Christian Heimes4fbc72b2008-03-22 00:47:35 +0000291 ready.release()
Christian Heimes7d2ff882007-11-30 14:35:04 +0000292 time.sleep(100)
293
Georg Brandl2067bfd2008-05-25 13:05:15 +0000294 _thread.start_new_thread(waitingThread, ())
Christian Heimes4fbc72b2008-03-22 00:47:35 +0000295 ready.acquire() # Be sure the other thread is waiting.
Christian Heimes7d2ff882007-11-30 14:35:04 +0000296 sys.exit(42)
Antoine Pitrouc4d78642011-05-05 20:17:32 +0200297 """)
Christian Heimes7d2ff882007-11-30 14:35:04 +0000298 self.assertEqual(rc, 42)
299
Neal Norwitzf5c7c2e2008-04-05 04:47:45 +0000300 def test_finalize_with_trace(self):
301 # Issue1733757
302 # Avoid a deadlock when sys.settrace steps into threading._shutdown
Antoine Pitrouc4d78642011-05-05 20:17:32 +0200303 assert_python_ok("-c", """if 1:
Neal Norwitzf5c7c2e2008-04-05 04:47:45 +0000304 import sys, threading
305
306 # A deadlock-killer, to prevent the
307 # testsuite to hang forever
308 def killer():
309 import os, time
310 time.sleep(2)
311 print('program blocked; aborting')
312 os._exit(2)
313 t = threading.Thread(target=killer)
Benjamin Petersonfdbea962008-08-18 17:33:47 +0000314 t.daemon = True
Neal Norwitzf5c7c2e2008-04-05 04:47:45 +0000315 t.start()
316
317 # This is the trace function
318 def func(frame, event, arg):
Benjamin Peterson672b8032008-06-11 19:14:14 +0000319 threading.current_thread()
Neal Norwitzf5c7c2e2008-04-05 04:47:45 +0000320 return func
321
322 sys.settrace(func)
Antoine Pitrouc4d78642011-05-05 20:17:32 +0200323 """)
Neal Norwitzf5c7c2e2008-04-05 04:47:45 +0000324
Antoine Pitrou011bd622009-10-20 21:52:47 +0000325 def test_join_nondaemon_on_shutdown(self):
326 # Issue 1722344
327 # Raising SystemExit skipped threading._shutdown
Antoine Pitrouc4d78642011-05-05 20:17:32 +0200328 rc, out, err = assert_python_ok("-c", """if 1:
Antoine Pitrou011bd622009-10-20 21:52:47 +0000329 import threading
330 from time import sleep
331
332 def child():
333 sleep(1)
334 # As a non-daemon thread we SHOULD wake up and nothing
335 # should be torn down yet
336 print("Woke up, sleep function is:", sleep)
337
338 threading.Thread(target=child).start()
339 raise SystemExit
Antoine Pitrouc4d78642011-05-05 20:17:32 +0200340 """)
341 self.assertEqual(out.strip(),
Antoine Pitrou899d1c62009-10-23 21:55:36 +0000342 b"Woke up, sleep function is: <built-in function sleep>")
Antoine Pitrouc4d78642011-05-05 20:17:32 +0200343 self.assertEqual(err, b"")
Neal Norwitzf5c7c2e2008-04-05 04:47:45 +0000344
Christian Heimes1af737c2008-01-23 08:24:23 +0000345 def test_enumerate_after_join(self):
346 # Try hard to trigger #1703448: a thread is still returned in
347 # threading.enumerate() after it has been join()ed.
348 enum = threading.enumerate
Antoine Pitrouc3b07572009-11-13 22:19:19 +0000349 old_interval = sys.getswitchinterval()
Christian Heimes1af737c2008-01-23 08:24:23 +0000350 try:
Jeffrey Yasskinca674122008-03-29 05:06:52 +0000351 for i in range(1, 100):
Antoine Pitrouc3b07572009-11-13 22:19:19 +0000352 sys.setswitchinterval(i * 0.0002)
Christian Heimes1af737c2008-01-23 08:24:23 +0000353 t = threading.Thread(target=lambda: None)
354 t.start()
355 t.join()
356 l = enum()
Ezio Melottib58e0bd2010-01-23 15:40:09 +0000357 self.assertNotIn(t, l,
Christian Heimes1af737c2008-01-23 08:24:23 +0000358 "#1703448 triggered after %d trials: %s" % (i, l))
359 finally:
Antoine Pitrouc3b07572009-11-13 22:19:19 +0000360 sys.setswitchinterval(old_interval)
Christian Heimes1af737c2008-01-23 08:24:23 +0000361
Christian Heimesd3eb5a152008-02-24 00:38:49 +0000362 def test_no_refcycle_through_target(self):
363 class RunSelfFunction(object):
364 def __init__(self, should_raise):
365 # The links in this refcycle from Thread back to self
366 # should be cleaned up when the thread completes.
367 self.should_raise = should_raise
368 self.thread = threading.Thread(target=self._run,
369 args=(self,),
370 kwargs={'yet_another':self})
371 self.thread.start()
372
373 def _run(self, other_ref, yet_another):
374 if self.should_raise:
375 raise SystemExit
376
377 cyclic_object = RunSelfFunction(should_raise=False)
378 weak_cyclic_object = weakref.ref(cyclic_object)
379 cyclic_object.thread.join()
380 del cyclic_object
Raymond Hettinger7beae8a2011-01-06 05:34:17 +0000381 self.assertIsNone(weak_cyclic_object(),
Ezio Melottib3aedd42010-11-20 19:04:17 +0000382 msg=('%d references still around' %
383 sys.getrefcount(weak_cyclic_object())))
Christian Heimesd3eb5a152008-02-24 00:38:49 +0000384
385 raising_cyclic_object = RunSelfFunction(should_raise=True)
386 weak_raising_cyclic_object = weakref.ref(raising_cyclic_object)
387 raising_cyclic_object.thread.join()
388 del raising_cyclic_object
Raymond Hettinger7beae8a2011-01-06 05:34:17 +0000389 self.assertIsNone(weak_raising_cyclic_object(),
Ezio Melottib3aedd42010-11-20 19:04:17 +0000390 msg=('%d references still around' %
391 sys.getrefcount(weak_raising_cyclic_object())))
Christian Heimesd3eb5a152008-02-24 00:38:49 +0000392
Benjamin Petersonb3085c92008-09-01 23:09:31 +0000393 def test_old_threading_api(self):
394 # Just a quick sanity check to make sure the old method names are
395 # still present
Benjamin Petersonf0923f52008-08-18 22:10:13 +0000396 t = threading.Thread()
Benjamin Petersonb3085c92008-09-01 23:09:31 +0000397 t.isDaemon()
398 t.setDaemon(True)
399 t.getName()
400 t.setName("name")
401 t.isAlive()
402 e = threading.Event()
403 e.isSet()
404 threading.activeCount()
Benjamin Petersonf0923f52008-08-18 22:10:13 +0000405
Brian Curtin81a4a6a2010-07-23 16:30:10 +0000406 def test_repr_daemon(self):
407 t = threading.Thread()
408 self.assertFalse('daemon' in repr(t))
409 t.daemon = True
410 self.assertTrue('daemon' in repr(t))
Brett Cannon3f5f2262010-07-23 15:50:52 +0000411
Antoine Pitrou0bd4deb2011-02-25 22:07:43 +0000412 def test_deamon_param(self):
413 t = threading.Thread()
414 self.assertFalse(t.daemon)
415 t = threading.Thread(daemon=False)
416 self.assertFalse(t.daemon)
417 t = threading.Thread(daemon=True)
418 self.assertTrue(t.daemon)
419
Antoine Pitrou8e6e0fd2012-04-19 23:55:01 +0200420 @unittest.skipUnless(hasattr(os, 'fork'), 'test needs fork()')
421 def test_dummy_thread_after_fork(self):
422 # Issue #14308: a dummy thread in the active list doesn't mess up
423 # the after-fork mechanism.
424 code = """if 1:
425 import _thread, threading, os, time
426
427 def background_thread(evt):
428 # Creates and registers the _DummyThread instance
429 threading.current_thread()
430 evt.set()
431 time.sleep(10)
432
433 evt = threading.Event()
434 _thread.start_new_thread(background_thread, (evt,))
435 evt.wait()
436 assert threading.active_count() == 2, threading.active_count()
437 if os.fork() == 0:
438 assert threading.active_count() == 1, threading.active_count()
439 os._exit(0)
440 else:
441 os.wait()
442 """
443 _, out, err = assert_python_ok("-c", code)
444 self.assertEqual(out, b'')
445 self.assertEqual(err, b'')
446
Christian Heimes1af737c2008-01-23 08:24:23 +0000447
Antoine Pitroub0e9bd42009-10-27 20:05:26 +0000448class ThreadJoinOnShutdown(BaseTestCase):
Jesse Nollera8513972008-07-17 16:49:17 +0000449
Victor Stinner26d31862011-07-01 14:26:24 +0200450 # Between fork() and exec(), only async-safe functions are allowed (issues
451 # #12316 and #11870), and fork() from a worker thread is known to trigger
452 # problems with some operating systems (issue #3863): skip problematic tests
453 # on platforms known to behave badly.
454 platforms_to_skip = ('freebsd4', 'freebsd5', 'freebsd6', 'netbsd5',
Stefan Krahfc4aa762013-01-17 23:29:54 +0100455 'os2emx', 'hp-ux11')
Victor Stinner26d31862011-07-01 14:26:24 +0200456
Jesse Nollera8513972008-07-17 16:49:17 +0000457 def _run_and_join(self, script):
458 script = """if 1:
459 import sys, os, time, threading
460
461 # a thread, which waits for the main program to terminate
462 def joiningfunc(mainthread):
463 mainthread.join()
464 print('end of thread')
Antoine Pitrou5fe291f2008-09-06 23:00:03 +0000465 # stdout is fully buffered because not a tty, we have to flush
466 # before exit.
467 sys.stdout.flush()
Jesse Nollera8513972008-07-17 16:49:17 +0000468 \n""" + script
469
Antoine Pitrouc4d78642011-05-05 20:17:32 +0200470 rc, out, err = assert_python_ok("-c", script)
471 data = out.decode().replace('\r', '')
Benjamin Petersonad703dc2008-07-17 17:02:57 +0000472 self.assertEqual(data, "end of main\nend of thread\n")
Jesse Nollera8513972008-07-17 16:49:17 +0000473
474 def test_1_join_on_shutdown(self):
475 # The usual case: on exit, wait for a non-daemon thread
476 script = """if 1:
477 import os
478 t = threading.Thread(target=joiningfunc,
479 args=(threading.current_thread(),))
480 t.start()
481 time.sleep(0.1)
482 print('end of main')
483 """
484 self._run_and_join(script)
485
Alexandre Vassalotti93f2cd22009-07-22 04:54:52 +0000486 @unittest.skipUnless(hasattr(os, 'fork'), "needs os.fork()")
Victor Stinner26d31862011-07-01 14:26:24 +0200487 @unittest.skipIf(sys.platform in platforms_to_skip, "due to known OS bug")
Jesse Nollera8513972008-07-17 16:49:17 +0000488 def test_2_join_in_forked_process(self):
489 # Like the test above, but from a forked interpreter
Jesse Nollera8513972008-07-17 16:49:17 +0000490 script = """if 1:
491 childpid = os.fork()
492 if childpid != 0:
493 os.waitpid(childpid, 0)
494 sys.exit(0)
495
496 t = threading.Thread(target=joiningfunc,
497 args=(threading.current_thread(),))
498 t.start()
499 print('end of main')
500 """
501 self._run_and_join(script)
502
Alexandre Vassalotti93f2cd22009-07-22 04:54:52 +0000503 @unittest.skipUnless(hasattr(os, 'fork'), "needs os.fork()")
Victor Stinner26d31862011-07-01 14:26:24 +0200504 @unittest.skipIf(sys.platform in platforms_to_skip, "due to known OS bug")
Antoine Pitrou5fe291f2008-09-06 23:00:03 +0000505 def test_3_join_in_forked_from_thread(self):
Jesse Nollera8513972008-07-17 16:49:17 +0000506 # Like the test above, but fork() was called from a worker thread
507 # In the forked process, the main Thread object must be marked as stopped.
Alexandre Vassalotti93f2cd22009-07-22 04:54:52 +0000508
Jesse Nollera8513972008-07-17 16:49:17 +0000509 script = """if 1:
510 main_thread = threading.current_thread()
511 def worker():
512 childpid = os.fork()
513 if childpid != 0:
514 os.waitpid(childpid, 0)
515 sys.exit(0)
516
517 t = threading.Thread(target=joiningfunc,
518 args=(main_thread,))
519 print('end of main')
520 t.start()
521 t.join() # Should not block: main_thread is already stopped
522
523 w = threading.Thread(target=worker)
524 w.start()
525 """
526 self._run_and_join(script)
527
Gregory P. Smith96c886c2011-01-03 21:06:12 +0000528 def assertScriptHasOutput(self, script, expected_output):
Antoine Pitrouc4d78642011-05-05 20:17:32 +0200529 rc, out, err = assert_python_ok("-c", script)
530 data = out.decode().replace('\r', '')
Gregory P. Smith96c886c2011-01-03 21:06:12 +0000531 self.assertEqual(data, expected_output)
532
533 @unittest.skipUnless(hasattr(os, 'fork'), "needs os.fork()")
Victor Stinner26d31862011-07-01 14:26:24 +0200534 @unittest.skipIf(sys.platform in platforms_to_skip, "due to known OS bug")
Gregory P. Smith96c886c2011-01-03 21:06:12 +0000535 def test_4_joining_across_fork_in_worker_thread(self):
536 # There used to be a possible deadlock when forking from a child
537 # thread. See http://bugs.python.org/issue6643.
538
Gregory P. Smith96c886c2011-01-03 21:06:12 +0000539 # The script takes the following steps:
540 # - The main thread in the parent process starts a new thread and then
541 # tries to join it.
542 # - The join operation acquires the Lock inside the thread's _block
543 # Condition. (See threading.py:Thread.join().)
544 # - We stub out the acquire method on the condition to force it to wait
545 # until the child thread forks. (See LOCK ACQUIRED HERE)
546 # - The child thread forks. (See LOCK HELD and WORKER THREAD FORKS
547 # HERE)
548 # - The main thread of the parent process enters Condition.wait(),
549 # which releases the lock on the child thread.
550 # - The child process returns. Without the necessary fix, when the
551 # main thread of the child process (which used to be the child thread
552 # in the parent process) attempts to exit, it will try to acquire the
553 # lock in the Thread._block Condition object and hang, because the
554 # lock was held across the fork.
555
556 script = """if 1:
557 import os, time, threading
558
559 finish_join = False
560 start_fork = False
561
562 def worker():
563 # Wait until this thread's lock is acquired before forking to
564 # create the deadlock.
565 global finish_join
566 while not start_fork:
567 time.sleep(0.01)
568 # LOCK HELD: Main thread holds lock across this call.
569 childpid = os.fork()
570 finish_join = True
571 if childpid != 0:
572 # Parent process just waits for child.
573 os.waitpid(childpid, 0)
574 # Child process should just return.
575
576 w = threading.Thread(target=worker)
577
578 # Stub out the private condition variable's lock acquire method.
579 # This acquires the lock and then waits until the child has forked
580 # before returning, which will release the lock soon after. If
581 # someone else tries to fix this test case by acquiring this lock
Ezio Melotti13925002011-03-16 11:05:33 +0200582 # before forking instead of resetting it, the test case will
Gregory P. Smith96c886c2011-01-03 21:06:12 +0000583 # deadlock when it shouldn't.
584 condition = w._block
585 orig_acquire = condition.acquire
586 call_count_lock = threading.Lock()
587 call_count = 0
588 def my_acquire():
589 global call_count
590 global start_fork
591 orig_acquire() # LOCK ACQUIRED HERE
592 start_fork = True
593 if call_count == 0:
594 while not finish_join:
595 time.sleep(0.01) # WORKER THREAD FORKS HERE
596 with call_count_lock:
597 call_count += 1
598 condition.acquire = my_acquire
599
600 w.start()
601 w.join()
602 print('end of main')
603 """
604 self.assertScriptHasOutput(script, "end of main\n")
605
606 @unittest.skipUnless(hasattr(os, 'fork'), "needs os.fork()")
Victor Stinner26d31862011-07-01 14:26:24 +0200607 @unittest.skipIf(sys.platform in platforms_to_skip, "due to known OS bug")
Gregory P. Smith96c886c2011-01-03 21:06:12 +0000608 def test_5_clear_waiter_locks_to_avoid_crash(self):
609 # Check that a spawned thread that forks doesn't segfault on certain
610 # platforms, namely OS X. This used to happen if there was a waiter
611 # lock in the thread's condition variable's waiters list. Even though
612 # we know the lock will be held across the fork, it is not safe to
613 # release locks held across forks on all platforms, so releasing the
614 # waiter lock caused a segfault on OS X. Furthermore, since locks on
615 # OS X are (as of this writing) implemented with a mutex + condition
616 # variable instead of a semaphore, while we know that the Python-level
617 # lock will be acquired, we can't know if the internal mutex will be
618 # acquired at the time of the fork.
619
Gregory P. Smith96c886c2011-01-03 21:06:12 +0000620 script = """if True:
621 import os, time, threading
622
623 start_fork = False
624
625 def worker():
626 # Wait until the main thread has attempted to join this thread
627 # before continuing.
628 while not start_fork:
629 time.sleep(0.01)
630 childpid = os.fork()
631 if childpid != 0:
632 # Parent process just waits for child.
633 (cpid, rc) = os.waitpid(childpid, 0)
634 assert cpid == childpid
635 assert rc == 0
636 print('end of worker thread')
637 else:
638 # Child process should just return.
639 pass
640
641 w = threading.Thread(target=worker)
642
643 # Stub out the private condition variable's _release_save method.
644 # This releases the condition's lock and flips the global that
645 # causes the worker to fork. At this point, the problematic waiter
646 # lock has been acquired once by the waiter and has been put onto
647 # the waiters list.
648 condition = w._block
649 orig_release_save = condition._release_save
650 def my_release_save():
651 global start_fork
652 orig_release_save()
653 # Waiter lock held here, condition lock released.
654 start_fork = True
655 condition._release_save = my_release_save
656
657 w.start()
658 w.join()
659 print('end of main thread')
660 """
661 output = "end of worker thread\nend of main thread\n"
662 self.assertScriptHasOutput(script, output)
663
Charles-François Natali8e6fe642012-03-24 20:36:09 +0100664 @unittest.skipIf(sys.platform in platforms_to_skip, "due to known OS bug")
Antoine Pitrou0d5e52d2011-05-04 20:02:30 +0200665 def test_6_daemon_threads(self):
666 # Check that a daemon thread cannot crash the interpreter on shutdown
667 # by manipulating internal structures that are being disposed of in
668 # the main thread.
669 script = """if True:
670 import os
671 import random
672 import sys
673 import time
674 import threading
675
676 thread_has_run = set()
677
678 def random_io():
679 '''Loop for a while sleeping random tiny amounts and doing some I/O.'''
Antoine Pitrou0d5e52d2011-05-04 20:02:30 +0200680 while True:
Victor Stinnera6d2c762011-06-30 18:20:11 +0200681 in_f = open(os.__file__, 'rb')
Antoine Pitrou0d5e52d2011-05-04 20:02:30 +0200682 stuff = in_f.read(200)
Victor Stinnera6d2c762011-06-30 18:20:11 +0200683 null_f = open(os.devnull, 'wb')
Antoine Pitrou0d5e52d2011-05-04 20:02:30 +0200684 null_f.write(stuff)
685 time.sleep(random.random() / 1995)
686 null_f.close()
687 in_f.close()
688 thread_has_run.add(threading.current_thread())
689
690 def main():
691 count = 0
692 for _ in range(40):
693 new_thread = threading.Thread(target=random_io)
694 new_thread.daemon = True
695 new_thread.start()
696 count += 1
697 while len(thread_has_run) < count:
698 time.sleep(0.001)
699 # Trigger process shutdown
700 sys.exit(0)
701
702 main()
703 """
704 rc, out, err = assert_python_ok('-c', script)
705 self.assertFalse(err)
706
Charles-François Natali6d0d24e2012-02-02 20:31:42 +0100707 @unittest.skipUnless(hasattr(os, 'fork'), "needs os.fork()")
Charles-François Natalib2c9e9a2012-02-08 21:29:11 +0100708 @unittest.skipIf(sys.platform in platforms_to_skip, "due to known OS bug")
Charles-François Natali6d0d24e2012-02-02 20:31:42 +0100709 def test_reinit_tls_after_fork(self):
710 # Issue #13817: fork() would deadlock in a multithreaded program with
711 # the ad-hoc TLS implementation.
712
713 def do_fork_and_wait():
714 # just fork a child process and wait it
715 pid = os.fork()
716 if pid > 0:
717 os.waitpid(pid, 0)
718 else:
719 os._exit(0)
720
721 # start a bunch of threads that will fork() child processes
722 threads = []
723 for i in range(16):
724 t = threading.Thread(target=do_fork_and_wait)
725 threads.append(t)
726 t.start()
727
728 for t in threads:
729 t.join()
730
Jesse Nollera8513972008-07-17 16:49:17 +0000731
Antoine Pitroub0e9bd42009-10-27 20:05:26 +0000732class ThreadingExceptionTests(BaseTestCase):
Guido van Rossumcd16bf62007-06-13 18:07:49 +0000733 # A RuntimeError should be raised if Thread.start() is called
734 # multiple times.
735 def test_start_thread_again(self):
736 thread = threading.Thread()
737 thread.start()
738 self.assertRaises(RuntimeError, thread.start)
739
Guido van Rossumcd16bf62007-06-13 18:07:49 +0000740 def test_joining_current_thread(self):
Benjamin Peterson672b8032008-06-11 19:14:14 +0000741 current_thread = threading.current_thread()
742 self.assertRaises(RuntimeError, current_thread.join);
Guido van Rossumcd16bf62007-06-13 18:07:49 +0000743
744 def test_joining_inactive_thread(self):
745 thread = threading.Thread()
746 self.assertRaises(RuntimeError, thread.join)
747
748 def test_daemonize_active_thread(self):
749 thread = threading.Thread()
750 thread.start()
Benjamin Petersonfdbea962008-08-18 17:33:47 +0000751 self.assertRaises(RuntimeError, setattr, thread, "daemon", True)
Guido van Rossumcd16bf62007-06-13 18:07:49 +0000752
Antoine Pitroufcf81fd2011-02-28 22:03:34 +0000753 def test_releasing_unacquired_lock(self):
754 lock = threading.Lock()
755 self.assertRaises(RuntimeError, lock.release)
756
Ned Deily9a7c5242011-05-28 00:19:56 -0700757 @unittest.skipUnless(sys.platform == 'darwin', 'test macosx problem')
758 def test_recursion_limit(self):
759 # Issue 9670
760 # test that excessive recursion within a non-main thread causes
761 # an exception rather than crashing the interpreter on platforms
762 # like Mac OS X or FreeBSD which have small default stack sizes
763 # for threads
764 script = """if True:
765 import threading
766
767 def recurse():
768 return recurse()
769
770 def outer():
771 try:
772 recurse()
773 except RuntimeError:
774 pass
775
776 w = threading.Thread(target=outer)
777 w.start()
778 w.join()
779 print('end of main thread')
780 """
781 expected_output = "end of main thread\n"
782 p = subprocess.Popen([sys.executable, "-c", script],
Antoine Pitroub8b6a682012-06-29 19:40:35 +0200783 stdout=subprocess.PIPE, stderr=subprocess.PIPE)
Ned Deily9a7c5242011-05-28 00:19:56 -0700784 stdout, stderr = p.communicate()
785 data = stdout.decode().replace('\r', '')
Antoine Pitroub8b6a682012-06-29 19:40:35 +0200786 self.assertEqual(p.returncode, 0, "Unexpected error: " + stderr.decode())
Ned Deily9a7c5242011-05-28 00:19:56 -0700787 self.assertEqual(data, expected_output)
Guido van Rossumcd16bf62007-06-13 18:07:49 +0000788
Antoine Pitrou557934f2009-11-06 22:41:14 +0000789class LockTests(lock_tests.LockTests):
790 locktype = staticmethod(threading.Lock)
791
Antoine Pitrou434736a2009-11-10 18:46:01 +0000792class PyRLockTests(lock_tests.RLockTests):
793 locktype = staticmethod(threading._PyRLock)
794
Charles-François Natali6b671b22012-01-28 11:36:04 +0100795@unittest.skipIf(threading._CRLock is None, 'RLock not implemented in C')
Antoine Pitrou434736a2009-11-10 18:46:01 +0000796class CRLockTests(lock_tests.RLockTests):
797 locktype = staticmethod(threading._CRLock)
Antoine Pitrou557934f2009-11-06 22:41:14 +0000798
799class EventTests(lock_tests.EventTests):
800 eventtype = staticmethod(threading.Event)
801
802class ConditionAsRLockTests(lock_tests.RLockTests):
803 # An Condition uses an RLock by default and exports its API.
804 locktype = staticmethod(threading.Condition)
805
806class ConditionTests(lock_tests.ConditionTests):
807 condtype = staticmethod(threading.Condition)
808
809class SemaphoreTests(lock_tests.SemaphoreTests):
810 semtype = staticmethod(threading.Semaphore)
811
812class BoundedSemaphoreTests(lock_tests.BoundedSemaphoreTests):
813 semtype = staticmethod(threading.BoundedSemaphore)
814
Kristján Valur Jónsson3be00032010-10-28 09:43:10 +0000815class BarrierTests(lock_tests.BarrierTests):
816 barriertype = staticmethod(threading.Barrier)
Antoine Pitrou557934f2009-11-06 22:41:14 +0000817
Victor Stinner754851f2011-04-19 23:58:51 +0200818
Tim Peters84d54892005-01-08 06:03:17 +0000819def test_main():
Antoine Pitrou434736a2009-11-10 18:46:01 +0000820 test.support.run_unittest(LockTests, PyRLockTests, CRLockTests, EventTests,
Antoine Pitrou557934f2009-11-06 22:41:14 +0000821 ConditionAsRLockTests, ConditionTests,
822 SemaphoreTests, BoundedSemaphoreTests,
823 ThreadTests,
824 ThreadJoinOnShutdown,
825 ThreadingExceptionTests,
Victor Stinnerd5c355c2011-04-30 14:53:09 +0200826 BarrierTests,
Antoine Pitrou557934f2009-11-06 22:41:14 +0000827 )
Tim Peters84d54892005-01-08 06:03:17 +0000828
829if __name__ == "__main__":
830 test_main()