1# This is a variant of the very old (early 90's) file 2# Demo/threads/bug.py. It simply provokes a number of threads into 3# trying to import the same module "at the same time". 4# There are no pleasant failure modes -- most likely is that Python 5# complains several times about module random having no attribute 6# randrange, and then Python hangs. 7 8import unittest 9from test.test_support import verbose, TestFailed, import_module 10thread = import_module('thread') 11 12critical_section = thread.allocate_lock() 13done = thread.allocate_lock() 14 15def task(): 16 global N, critical_section, done 17 import random 18 x = random.randrange(1, 3) 19 critical_section.acquire() 20 N -= 1 21 # Must release critical_section before releasing done, else the main 22 # thread can exit and set critical_section to None as part of global 23 # teardown; then critical_section.release() raises AttributeError. 24 finished = N == 0 25 critical_section.release() 26 if finished: 27 done.release() 28 29def test_import_hangers(): 30 import sys 31 if verbose: 32 print "testing import hangers ...", 33 34 import test.threaded_import_hangers 35 try: 36 if test.threaded_import_hangers.errors: 37 raise TestFailed(test.threaded_import_hangers.errors) 38 elif verbose: 39 print "OK." 40 finally: 41 # In case this test is run again, make sure the helper module 42 # gets loaded from scratch again. 43 del sys.modules['test.threaded_import_hangers'] 44 45# Tricky: When regrtest imports this module, the thread running regrtest 46# grabs the import lock and won't let go of it until this module returns. 47# All other threads attempting an import hang for the duration. Since 48# this test spawns threads that do little *but* import, we can't do that 49# successfully until after this module finishes importing and regrtest 50# regains control. To make this work, a special case was added to 51# regrtest to invoke a module's "test_main" function (if any) after 52# importing it. 53 54def test_main(): # magic name! see above 55 global N, done 56 57 import imp 58 if imp.lock_held(): 59 # This triggers on, e.g., from test import autotest. 60 raise unittest.SkipTest("can't run when import lock is held") 61 62 done.acquire() 63 for N in (20, 50) * 3: 64 if verbose: 65 print "Trying", N, "threads ...", 66 for i in range(N): 67 thread.start_new_thread(task, ()) 68 done.acquire() 69 if verbose: 70 print "OK." 71 done.release() 72 73 test_import_hangers() 74 75if __name__ == "__main__": 76 test_main() 77