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