Lib/test/test_threaded_import.py

   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
   8 import thread
   9 from test.test_support import verbose, TestSkipped
  10
  11 critical_section = thread.allocate_lock()
  12 done = thread.allocate_lock()
  13
  14 def task():
  15     global N, critical_section, done
  16     import random
  17     x = random.randrange(1, 3)
  18     critical_section.acquire()
  19     N -= 1
  20     # Must release critical_section before releasing done, else the main
  21     # thread can exit and set critical_section to None as part of global
  22     # teardown; then critical_section.release() raises AttributeError.
  23     finished = N == 0
  24     critical_section.release()
  25     if finished:
  26         done.release()
  27
  28 # Tricky:  When regrtest imports this module, the thread running regrtest
  29 # grabs the import lock and won't let go of it until this module returns.
  30 # All other threads attempting an import hang for the duration.  Since
  31 # this test spawns threads that do little *but* import, we can't do that
  32 # successfully until after this module finishes importing and regrtest
  33 # regains control.  To make this work, a special case was added to
  34 # regrtest to invoke a module's "test_main" function (if any) after
  35 # importing it.
  36
  37 def test_main():        # magic name!  see above
  38     global N, done
  39
  40     import imp
  41     if imp.lock_held():
  42         # This triggers on, e.g., from test import autotest.
  43         raise TestSkipped("can't run when import lock is held")
  44
  45     done.acquire()
  46     for N in (20, 50) * 3:
  47         if verbose:
  48             print "Trying", N, "threads ...",
  49         for i in range(N):
  50             thread.start_new_thread(task, ())
  51         done.acquire()
  52         if verbose:
  53             print "OK."
  54
  55 if __name__ == "__main__":
  56     test_main()