PyXR

c:\python24\lib \ test \ test_threaded_import.py



0001 # This is a variant of the very old (early 90's) file
0002 # Demo/threads/bug.py.  It simply provokes a number of threads into
0003 # trying to import the same module "at the same time".
0004 # There are no pleasant failure modes -- most likely is that Python
0005 # complains several times about module random having no attribute
0006 # randrange, and then Python hangs.
0007 
0008 import thread
0009 from test.test_support import verbose, TestSkipped
0010 
0011 critical_section = thread.allocate_lock()
0012 done = thread.allocate_lock()
0013 
0014 def task():
0015     global N, critical_section, done
0016     import random
0017     x = random.randrange(1, 3)
0018     critical_section.acquire()
0019     N -= 1
0020     # Must release critical_section before releasing done, else the main
0021     # thread can exit and set critical_section to None as part of global
0022     # teardown; then critical_section.release() raises AttributeError.
0023     finished = N == 0
0024     critical_section.release()
0025     if finished:
0026         done.release()
0027 
0028 # Tricky:  When regrtest imports this module, the thread running regrtest
0029 # grabs the import lock and won't let go of it until this module returns.
0030 # All other threads attempting an import hang for the duration.  Since
0031 # this test spawns threads that do little *but* import, we can't do that
0032 # successfully until after this module finishes importing and regrtest
0033 # regains control.  To make this work, a special case was added to
0034 # regrtest to invoke a module's "test_main" function (if any) after
0035 # importing it.
0036 
0037 def test_main():        # magic name!  see above
0038     global N, done
0039 
0040     import imp
0041     if imp.lock_held():
0042         # This triggers on, e.g., from test import autotest.
0043         raise TestSkipped("can't run when import lock is held")
0044 
0045     done.acquire()
0046     for N in (20, 50) * 3:
0047         if verbose:
0048             print "Trying", N, "threads ...",
0049         for i in range(N):
0050             thread.start_new_thread(task, ())
0051         done.acquire()
0052         if verbose:
0053             print "OK."
0054     done.release()
0055 
0056 if __name__ == "__main__":
0057     test_main()
0058 

Generated by PyXR 0.9.4
SourceForge.net Logo