test_fork1.py 3.69 KB
Newer Older
1 2 3
"""This test checks for correct fork() behavior.
"""

4
import _imp as imp
5
import os
6 7
import signal
import sys
8
import time
9

10
from test.fork_wait import ForkWait
11 12 13
from test.support import (run_unittest, reap_children, get_attribute,
                          import_module, verbose)

14
threading = import_module('threading')
15 16 17

# Skip test if fork does not exist.
get_attribute(os, 'fork')
18

19 20
class ForkTest(ForkWait):
    def wait_impl(self, cpid):
21 22 23 24 25 26 27 28
        for i in range(10):
            # waitpid() shouldn't hang, but some of the buildbots seem to hang
            # in the forking tests.  This is an attempt to fix the problem.
            spid, status = os.waitpid(cpid, os.WNOHANG)
            if spid == cpid:
                break
            time.sleep(1.0)

29 30 31
        self.assertEqual(spid, cpid)
        self.assertEqual(status, 0, "cause = %d, exit = %d" % (status&0xff, status>>8))

32 33
    def test_threaded_import_lock_fork(self):
        """Check fork() in main thread works while a subthread is doing an import"""
34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49
        import_started = threading.Event()
        fake_module_name = "fake test module"
        partial_module = "partial"
        complete_module = "complete"
        def importer():
            imp.acquire_lock()
            sys.modules[fake_module_name] = partial_module
            import_started.set()
            time.sleep(0.01) # Give the other thread time to try and acquire.
            sys.modules[fake_module_name] = complete_module
            imp.release_lock()
        t = threading.Thread(target=importer)
        t.start()
        import_started.wait()
        pid = os.fork()
        try:
50 51 52
            # PyOS_BeforeFork should have waited for the import to complete
            # before forking, so the child can recreate the import lock
            # correctly, but also won't see a partially initialised module
53 54 55 56 57
            if not pid:
                m = __import__(fake_module_name)
                if m == complete_module:
                    os._exit(0)
                else:
58 59
                    if verbose > 1:
                        print("Child encountered partial module")
60 61 62 63 64 65 66 67 68 69 70 71 72
                    os._exit(1)
            else:
                t.join()
                # Exitcode 1 means the child got a partial module (bad.) No
                # exitcode (but a hang, which manifests as 'got pid 0')
                # means the child deadlocked (also bad.)
                self.wait_impl(pid)
        finally:
            try:
                os.kill(pid, signal.SIGKILL)
            except OSError:
                pass

73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105

    def test_nested_import_lock_fork(self):
        """Check fork() in main thread works while the main thread is doing an import"""
        # Issue 9573: this used to trigger RuntimeError in the child process
        def fork_with_import_lock(level):
            release = 0
            in_child = False
            try:
                try:
                    for i in range(level):
                        imp.acquire_lock()
                        release += 1
                    pid = os.fork()
                    in_child = not pid
                finally:
                    for i in range(release):
                        imp.release_lock()
            except RuntimeError:
                if in_child:
                    if verbose > 1:
                        print("RuntimeError in child")
                    os._exit(1)
                raise
            if in_child:
                os._exit(0)
            self.wait_impl(pid)

        # Check this works with various levels of nested
        # import in the main thread
        for level in range(5):
            fork_with_import_lock(level)


106 107
def test_main():
    run_unittest(ForkTest)
108
    reap_children()
109

110 111
if __name__ == "__main__":
    test_main()