# This is a variant of the very old (early 90's) file

# Demo/threads/bug.py.  It simply provokes a number of threads into

# trying to import the same module "at the same time".

# There are no pleasant failure modes -- most likely is that Python

# complains several times about module random having no attribute

# randrange, and then Python hangs.

import thread

from test.test_support import verbose, TestSkipped, TestFailed

critical_section = thread.allocate_lock()

done = thread.allocate_lock()

def task():

    global N, critical_section, done

    import random

    x = random.randrange(1, 3)

    critical_section.acquire()

    N -= 1

    # Must release critical_section before releasing done, else the main

    # thread can exit and set critical_section to None as part of global

    # teardown; then critical_section.release() raises AttributeError.

    finished = N == 0

    critical_section.release()

    if finished:

        done.release()

def test_import_hangers():

    import sys

    if verbose:

        print "testing import hangers ...",

    import test.threaded_import_hangers

    try:

        if test.threaded_import_hangers.errors:

            raise TestFailed(test.threaded_import_hangers.errors)

        elif verbose:

            print "OK."

    finally:

        # In case this test is run again, make sure the helper module

        # gets loaded from scratch again.

        del sys.modules['test.threaded_import_hangers']

# Tricky:  When regrtest imports this module, the thread running regrtest

# grabs the import lock and won't let go of it until this module returns.

# All other threads attempting an import hang for the duration.  Since

# this test spawns threads that do little *but* import, we can't do that

# successfully until after this module finishes importing and regrtest

# regains control.  To make this work, a special case was added to

# regrtest to invoke a module's "test_main" function (if any) after

# importing it.

def test_main():        # magic name!  see above

    global N, done

    import imp

    if imp.lock_held():

        # This triggers on, e.g., from test import autotest.

        raise TestSkipped("can't run when import lock is held")

    done.acquire()

    for N in (20, 50) * 3:

        if verbose:

            print "Trying", N, "threads ...",

        for i in range(N):

            thread.start_new_thread(task, ())

        done.acquire()

        if verbose:

            print "OK."

    done.release()

    test_import_hangers()

if __name__ == "__main__":

    test_main()