home *** CD-ROM | disk | FTP | other *** search
/ Freelog 33 / Freelog033.iso / Progr / Python-2.2.1.exe / TEST_THREADED_IMPORT.PY < prev    next >
Encoding:
Python Source  |  2002-02-22  |  2.0 KB  |  57 lines
  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_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()
  57.