Lib/test/test_threading.py

   1 # Very rudimentary test of threading module
   2
   3 import test.test_support
   4 from test.test_support import verbose
   5 import random
   6 import re
   7 import sys
   8 import threading
   9 import thread
  10 import time
  11 import unittest
  12 import weakref
  13
  14 from test import lock_tests
  15
  16 # A trivial mutable counter.
  17 class Counter(object):
  18     def __init__(self):
  19         self.value = 0
  20     def inc(self):
  21         self.value += 1
  22     def dec(self):
  23         self.value -= 1
  24     def get(self):
  25         return self.value
  26
  27 class TestThread(threading.Thread):
  28     def __init__(self, name, testcase, sema, mutex, nrunning):
  29         threading.Thread.__init__(self, name=name)
  30         self.testcase = testcase
  31         self.sema = sema
  32         self.mutex = mutex
  33         self.nrunning = nrunning
  34
  35     def run(self):
  36         delay = random.random() / 10000.0
  37         if verbose:
  38             print 'task %s will run for %.1f usec' % (
  39                 self.name, delay * 1e6)
  40
  41         with self.sema:
  42             with self.mutex:
  43                 self.nrunning.inc()
  44                 if verbose:
  45                     print self.nrunning.get(), 'tasks are running'
  46                 self.testcase.assertTrue(self.nrunning.get() <= 3)
  47
  48             time.sleep(delay)
  49             if verbose:
  50                 print 'task', self.name, 'done'
  51
  52             with self.mutex:
  53                 self.nrunning.dec()
  54                 self.testcase.assertTrue(self.nrunning.get() >= 0)
  55                 if verbose:
  56                     print '%s is finished. %d tasks are running' % (
  57                         self.name, self.nrunning.get())
  58
  59 class BaseTestCase(unittest.TestCase):
  60     def setUp(self):
  61         self._threads = test.test_support.threading_setup()
  62
  63     def tearDown(self):
  64         test.test_support.threading_cleanup(*self._threads)
  65         test.test_support.reap_children()
  66
  67
  68 class ThreadTests(BaseTestCase):
  69
  70     # Create a bunch of threads, let each do some work, wait until all are
  71     # done.
  72     def test_various_ops(self):
  73         # This takes about n/3 seconds to run (about n/3 clumps of tasks,
  74         # times about 1 second per clump).
  75         NUMTASKS = 10
  76
  77         # no more than 3 of the 10 can run at once
  78         sema = threading.BoundedSemaphore(value=3)
  79         mutex = threading.RLock()
  80         numrunning = Counter()
  81
  82         threads = []
  83
  84         for i in range(NUMTASKS):
  85             t = TestThread("<thread %d>"%i, self, sema, mutex, numrunning)
  86             threads.append(t)
  87             self.assertEqual(t.ident, None)
  88             self.assertTrue(re.match('<TestThread\(.*, initial\)>', repr(t)))
  89             t.start()
  90
  91         if verbose:
  92             print 'waiting for all tasks to complete'
  93         for t in threads:
  94             t.join(NUMTASKS)
  95             self.assertTrue(not t.is_alive())
  96             self.assertNotEqual(t.ident, 0)
  97             self.assertFalse(t.ident is None)
  98             self.assertTrue(re.match('<TestThread\(.*, \w+ -?\d+\)>', repr(t)))
  99         if verbose:
 100             print 'all tasks done'
 101         self.assertEqual(numrunning.get(), 0)
 102
 103     def test_ident_of_no_threading_threads(self):
 104         # The ident still must work for the main thread and dummy threads.
 105         self.assertFalse(threading.currentThread().ident is None)
 106         def f():
 107             ident.append(threading.currentThread().ident)
 108             done.set()
 109         done = threading.Event()
 110         ident = []
 111         thread.start_new_thread(f, ())
 112         done.wait()
 113         self.assertFalse(ident[0] is None)
 114         # Kill the "immortal" _DummyThread
 115         del threading._active[ident[0]]
 116
 117     # run with a small(ish) thread stack size (256kB)
 118     def test_various_ops_small_stack(self):
 119         if verbose:
 120             print 'with 256kB thread stack size...'
 121         try:
 122             threading.stack_size(262144)
 123         except thread.error:
 124             if verbose:
 125                 print 'platform does not support changing thread stack size'
 126             return
 127         self.test_various_ops()
 128         threading.stack_size(0)
 129
 130     # run with a large thread stack size (1MB)
 131     def test_various_ops_large_stack(self):
 132         if verbose:
 133             print 'with 1MB thread stack size...'
 134         try:
 135             threading.stack_size(0x100000)
 136         except thread.error:
 137             if verbose:
 138                 print 'platform does not support changing thread stack size'
 139             return
 140         self.test_various_ops()
 141         threading.stack_size(0)
 142
 143     def test_foreign_thread(self):
 144         # Check that a "foreign" thread can use the threading module.
 145         def f(mutex):
 146             # Calling current_thread() forces an entry for the foreign
 147             # thread to get made in the threading._active map.
 148             threading.current_thread()
 149             mutex.release()
 150
 151         mutex = threading.Lock()
 152         mutex.acquire()
 153         tid = thread.start_new_thread(f, (mutex,))
 154         # Wait for the thread to finish.
 155         mutex.acquire()
 156         self.assertTrue(tid in threading._active)
 157         self.assertTrue(isinstance(threading._active[tid],
 158                                 threading._DummyThread))
 159         del threading._active[tid]
 160
 161     # PyThreadState_SetAsyncExc() is a CPython-only gimmick, not (currently)
 162     # exposed at the Python level.  This test relies on ctypes to get at it.
 163     def test_PyThreadState_SetAsyncExc(self):
 164         try:
 165             import ctypes
 166         except ImportError:
 167             if verbose:
 168                 print "test_PyThreadState_SetAsyncExc can't import ctypes"
 169             return  # can't do anything
 170
 171         set_async_exc = ctypes.pythonapi.PyThreadState_SetAsyncExc
 172
 173         class AsyncExc(Exception):
 174             pass
 175
 176         exception = ctypes.py_object(AsyncExc)
 177
 178         # First check it works when setting the exception from the same thread.
 179         tid = thread.get_ident()
 180
 181         try:
 182             result = set_async_exc(ctypes.c_long(tid), exception)
 183             # The exception is async, so we might have to keep the VM busy until
 184             # it notices.
 185             while True:
 186                 pass
 187         except AsyncExc:
 188             pass
 189         else:
 190             # This code is unreachable but it reflects the intent. If we wanted
 191             # to be smarter the above loop wouldn't be infinite.
 192             self.fail("AsyncExc not raised")
 193         try:
 194             self.assertEqual(result, 1) # one thread state modified
 195         except UnboundLocalError:
 196             # The exception was raised too quickly for us to get the result.
 197             pass
 198
 199         # `worker_started` is set by the thread when it's inside a try/except
 200         # block waiting to catch the asynchronously set AsyncExc exception.
 201         # `worker_saw_exception` is set by the thread upon catching that
 202         # exception.
 203         worker_started = threading.Event()
 204         worker_saw_exception = threading.Event()
 205
 206         class Worker(threading.Thread):
 207             def run(self):
 208                 self.id = thread.get_ident()
 209                 self.finished = False
 210
 211                 try:
 212                     while True:
 213                         worker_started.set()
 214                         time.sleep(0.1)
 215                 except AsyncExc:
 216                     self.finished = True
 217                     worker_saw_exception.set()
 218
 219         t = Worker()
 220         t.daemon = True # so if this fails, we don't hang Python at shutdown
 221         t.start()
 222         if verbose:
 223             print "    started worker thread"
 224
 225         # Try a thread id that doesn't make sense.
 226         if verbose:
 227             print "    trying nonsensical thread id"
 228         result = set_async_exc(ctypes.c_long(-1), exception)
 229         self.assertEqual(result, 0)  # no thread states modified
 230
 231         # Now raise an exception in the worker thread.
 232         if verbose:
 233             print "    waiting for worker thread to get started"
 234         ret = worker_started.wait()
 235         self.assertTrue(ret)
 236         if verbose:
 237             print "    verifying worker hasn't exited"
 238         self.assertTrue(not t.finished)
 239         if verbose:
 240             print "    attempting to raise asynch exception in worker"
 241         result = set_async_exc(ctypes.c_long(t.id), exception)
 242         self.assertEqual(result, 1) # one thread state modified
 243         if verbose:
 244             print "    waiting for worker to say it caught the exception"
 245         worker_saw_exception.wait(timeout=10)
 246         self.assertTrue(t.finished)
 247         if verbose:
 248             print "    all OK -- joining worker"
 249         if t.finished:
 250             t.join()
 251         # else the thread is still running, and we have no way to kill it
 252
 253     def test_finalize_runnning_thread(self):
 254         # Issue 1402: the PyGILState_Ensure / _Release functions may be called
 255         # very late on python exit: on deallocation of a running thread for
 256         # example.
 257         try:
 258             import ctypes
 259         except ImportError:
 260             if verbose:
 261                 print("test_finalize_with_runnning_thread can't import ctypes")
 262             return  # can't do anything
 263
 264         import subprocess
 265         rc = subprocess.call([sys.executable, "-c", """if 1:
 266             import ctypes, sys, time, thread
 267
 268             # This lock is used as a simple event variable.
 269             ready = thread.allocate_lock()
 270             ready.acquire()
 271
 272             # Module globals are cleared before __del__ is run
 273             # So we save the functions in class dict
 274             class C:
 275                 ensure = ctypes.pythonapi.PyGILState_Ensure
 276                 release = ctypes.pythonapi.PyGILState_Release
 277                 def __del__(self):
 278                     state = self.ensure()
 279                     self.release(state)
 280
 281             def waitingThread():
 282                 x = C()
 283                 ready.release()
 284                 time.sleep(100)
 285
 286             thread.start_new_thread(waitingThread, ())
 287             ready.acquire()  # Be sure the other thread is waiting.
 288             sys.exit(42)
 289             """])
 290         self.assertEqual(rc, 42)
 291
 292     def test_finalize_with_trace(self):
 293         # Issue1733757
 294         # Avoid a deadlock when sys.settrace steps into threading._shutdown
 295         import subprocess
 296         rc = subprocess.call([sys.executable, "-c", """if 1:
 297             import sys, threading
 298
 299             # A deadlock-killer, to prevent the
 300             # testsuite to hang forever
 301             def killer():
 302                 import os, time
 303                 time.sleep(2)
 304                 print 'program blocked; aborting'
 305                 os._exit(2)
 306             t = threading.Thread(target=killer)
 307             t.daemon = True
 308             t.start()
 309
 310             # This is the trace function
 311             def func(frame, event, arg):
 312                 threading.current_thread()
 313                 return func
 314
 315             sys.settrace(func)
 316             """])
 317         self.assertFalse(rc == 2, "interpreted was blocked")
 318         self.assertTrue(rc == 0, "Unexpected error")
 319
 320     def test_join_nondaemon_on_shutdown(self):
 321         # Issue 1722344
 322         # Raising SystemExit skipped threading._shutdown
 323         import subprocess
 324         p = subprocess.Popen([sys.executable, "-c", """if 1:
 325                 import threading
 326                 from time import sleep
 327
 328                 def child():
 329                     sleep(1)
 330                     # As a non-daemon thread we SHOULD wake up and nothing
 331                     # should be torn down yet
 332                     print "Woke up, sleep function is:", sleep
 333
 334                 threading.Thread(target=child).start()
 335                 raise SystemExit
 336             """],
 337             stdout=subprocess.PIPE,
 338             stderr=subprocess.PIPE)
 339         stdout, stderr = p.communicate()
 340         self.assertEqual(stdout.strip(),
 341             "Woke up, sleep function is: <built-in function sleep>")
 342         stderr = re.sub(r"^\[\d+ refs\]", "", stderr, re.MULTILINE).strip()
 343         self.assertEqual(stderr, "")
 344
 345     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
 349         old_interval = sys.getcheckinterval()
 350         try:
 351             for i in xrange(1, 100):
 352                 # Try a couple times at each thread-switching interval
 353                 # to get more interleavings.
 354                 sys.setcheckinterval(i // 5)
 355                 t = threading.Thread(target=lambda: None)
 356                 t.start()
 357                 t.join()
 358                 l = enum()
 359                 self.assertFalse(t in l,
 360                     "#1703448 triggered after %d trials: %s" % (i, l))
 361         finally:
 362             sys.setcheckinterval(old_interval)
 363
 364     def test_no_refcycle_through_target(self):
 365         class RunSelfFunction(object):
 366             def __init__(self, should_raise):
 367                 # The links in this refcycle from Thread back to self
 368                 # should be cleaned up when the thread completes.
 369                 self.should_raise = should_raise
 370                 self.thread = threading.Thread(target=self._run,
 371                                                args=(self,),
 372                                                kwargs={'yet_another':self})
 373                 self.thread.start()
 374
 375             def _run(self, other_ref, yet_another):
 376                 if self.should_raise:
 377                     raise SystemExit
 378
 379         cyclic_object = RunSelfFunction(should_raise=False)
 380         weak_cyclic_object = weakref.ref(cyclic_object)
 381         cyclic_object.thread.join()
 382         del cyclic_object
 383         self.assertEquals(None, weak_cyclic_object(),
 384                           msg=('%d references still around' %
 385                                sys.getrefcount(weak_cyclic_object())))
 386
 387         raising_cyclic_object = RunSelfFunction(should_raise=True)
 388         weak_raising_cyclic_object = weakref.ref(raising_cyclic_object)
 389         raising_cyclic_object.thread.join()
 390         del raising_cyclic_object
 391         self.assertEquals(None, weak_raising_cyclic_object(),
 392                           msg=('%d references still around' %
 393                                sys.getrefcount(weak_raising_cyclic_object())))
 394
 395
 396 class ThreadJoinOnShutdown(BaseTestCase):
 397
 398     def _run_and_join(self, script):
 399         script = """if 1:
 400             import sys, os, time, threading
 401
 402             # a thread, which waits for the main program to terminate
 403             def joiningfunc(mainthread):
 404                 mainthread.join()
 405                 print 'end of thread'
 406         \n""" + script
 407
 408         import subprocess
 409         p = subprocess.Popen([sys.executable, "-c", script], stdout=subprocess.PIPE)
 410         rc = p.wait()
 411         data = p.stdout.read().replace('\r', '')
 412         self.assertEqual(data, "end of main\nend of thread\n")
 413         self.assertFalse(rc == 2, "interpreter was blocked")
 414         self.assertTrue(rc == 0, "Unexpected error")
 415
 416     def test_1_join_on_shutdown(self):
 417         # The usual case: on exit, wait for a non-daemon thread
 418         script = """if 1:
 419             import os
 420             t = threading.Thread(target=joiningfunc,
 421                                  args=(threading.current_thread(),))
 422             t.start()
 423             time.sleep(0.1)
 424             print 'end of main'
 425             """
 426         self._run_and_join(script)
 427
 428
 429     def test_2_join_in_forked_process(self):
 430         # Like the test above, but from a forked interpreter
 431         import os
 432         if not hasattr(os, 'fork'):
 433             return
 434         script = """if 1:
 435             childpid = os.fork()
 436             if childpid != 0:
 437                 os.waitpid(childpid, 0)
 438                 sys.exit(0)
 439
 440             t = threading.Thread(target=joiningfunc,
 441                                  args=(threading.current_thread(),))
 442             t.start()
 443             print 'end of main'
 444             """
 445         self._run_and_join(script)
 446
 447     def test_3_join_in_forked_from_thread(self):
 448         # Like the test above, but fork() was called from a worker thread
 449         # In the forked process, the main Thread object must be marked as stopped.
 450         import os
 451         if not hasattr(os, 'fork'):
 452             return
 453         # Skip platforms with known problems forking from a worker thread.
 454         # See http://bugs.python.org/issue3863.
 455         if sys.platform in ('freebsd4', 'freebsd5', 'freebsd6', 'os2emx'):
 456             print >>sys.stderr, ('Skipping test_3_join_in_forked_from_thread'
 457                                  ' due to known OS bugs on'), sys.platform
 458             return
 459         script = """if 1:
 460             main_thread = threading.current_thread()
 461             def worker():
 462                 childpid = os.fork()
 463                 if childpid != 0:
 464                     os.waitpid(childpid, 0)
 465                     sys.exit(0)
 466
 467                 t = threading.Thread(target=joiningfunc,
 468                                      args=(main_thread,))
 469                 print 'end of main'
 470                 t.start()
 471                 t.join() # Should not block: main_thread is already stopped
 472
 473             w = threading.Thread(target=worker)
 474             w.start()
 475             """
 476         self._run_and_join(script)
 477
 478
 479 class ThreadingExceptionTests(BaseTestCase):
 480     # A RuntimeError should be raised if Thread.start() is called
 481     # multiple times.
 482     def test_start_thread_again(self):
 483         thread = threading.Thread()
 484         thread.start()
 485         self.assertRaises(RuntimeError, thread.start)
 486
 487     def test_joining_current_thread(self):
 488         current_thread = threading.current_thread()
 489         self.assertRaises(RuntimeError, current_thread.join);
 490
 491     def test_joining_inactive_thread(self):
 492         thread = threading.Thread()
 493         self.assertRaises(RuntimeError, thread.join)
 494
 495     def test_daemonize_active_thread(self):
 496         thread = threading.Thread()
 497         thread.start()
 498         self.assertRaises(RuntimeError, setattr, thread, "daemon", True)
 499
 500
 501 class LockTests(lock_tests.LockTests):
 502     locktype = staticmethod(threading.Lock)
 503
 504 class RLockTests(lock_tests.RLockTests):
 505     locktype = staticmethod(threading.RLock)
 506
 507 class EventTests(lock_tests.EventTests):
 508     eventtype = staticmethod(threading.Event)
 509
 510 class ConditionAsRLockTests(lock_tests.RLockTests):
 511     # An Condition uses an RLock by default and exports its API.
 512     locktype = staticmethod(threading.Condition)
 513
 514 class ConditionTests(lock_tests.ConditionTests):
 515     condtype = staticmethod(threading.Condition)
 516
 517 class SemaphoreTests(lock_tests.SemaphoreTests):
 518     semtype = staticmethod(threading.Semaphore)
 519
 520 class BoundedSemaphoreTests(lock_tests.BoundedSemaphoreTests):
 521     semtype = staticmethod(threading.BoundedSemaphore)
 522
 523
 524 def test_main():
 525     test.test_support.run_unittest(LockTests, RLockTests, EventTests,
 526                                    ConditionAsRLockTests, ConditionTests,
 527                                    SemaphoreTests, BoundedSemaphoreTests,
 528                                    ThreadTests,
 529                                    ThreadJoinOnShutdown,
 530                                    ThreadingExceptionTests,
 531                                    )
 532
 533 if __name__ == "__main__":
 534     test_main()