• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1"""This test checks for correct fork() behavior.
2"""
3
4import _imp as imp
5import os
6import signal
7import sys
8import threading
9import time
10import unittest
11
12from test.fork_wait import ForkWait
13from test.support import reap_children, get_attribute, verbose
14
15
16# Skip test if fork does not exist.
17get_attribute(os, 'fork')
18
19class ForkTest(ForkWait):
20    def wait_impl(self, cpid):
21        deadline = time.monotonic() + 10.0
22        while time.monotonic() <= deadline:
23            # waitpid() shouldn't hang, but some of the buildbots seem to hang
24            # in the forking tests.  This is an attempt to fix the problem.
25            spid, status = os.waitpid(cpid, os.WNOHANG)
26            if spid == cpid:
27                break
28            time.sleep(0.1)
29
30        self.assertEqual(spid, cpid)
31        self.assertEqual(status, 0, "cause = %d, exit = %d" % (status&0xff, status>>8))
32
33    def test_threaded_import_lock_fork(self):
34        """Check fork() in main thread works while a subthread is doing an import"""
35        import_started = threading.Event()
36        fake_module_name = "fake test module"
37        partial_module = "partial"
38        complete_module = "complete"
39        def importer():
40            imp.acquire_lock()
41            sys.modules[fake_module_name] = partial_module
42            import_started.set()
43            time.sleep(0.01) # Give the other thread time to try and acquire.
44            sys.modules[fake_module_name] = complete_module
45            imp.release_lock()
46        t = threading.Thread(target=importer)
47        t.start()
48        import_started.wait()
49        pid = os.fork()
50        try:
51            # PyOS_BeforeFork should have waited for the import to complete
52            # before forking, so the child can recreate the import lock
53            # correctly, but also won't see a partially initialised module
54            if not pid:
55                m = __import__(fake_module_name)
56                if m == complete_module:
57                    os._exit(0)
58                else:
59                    if verbose > 1:
60                        print("Child encountered partial module")
61                    os._exit(1)
62            else:
63                t.join()
64                # Exitcode 1 means the child got a partial module (bad.) No
65                # exitcode (but a hang, which manifests as 'got pid 0')
66                # means the child deadlocked (also bad.)
67                self.wait_impl(pid)
68        finally:
69            try:
70                os.kill(pid, signal.SIGKILL)
71            except OSError:
72                pass
73
74
75    def test_nested_import_lock_fork(self):
76        """Check fork() in main thread works while the main thread is doing an import"""
77        # Issue 9573: this used to trigger RuntimeError in the child process
78        def fork_with_import_lock(level):
79            release = 0
80            in_child = False
81            try:
82                try:
83                    for i in range(level):
84                        imp.acquire_lock()
85                        release += 1
86                    pid = os.fork()
87                    in_child = not pid
88                finally:
89                    for i in range(release):
90                        imp.release_lock()
91            except RuntimeError:
92                if in_child:
93                    if verbose > 1:
94                        print("RuntimeError in child")
95                    os._exit(1)
96                raise
97            if in_child:
98                os._exit(0)
99            self.wait_impl(pid)
100
101        # Check this works with various levels of nested
102        # import in the main thread
103        for level in range(5):
104            fork_with_import_lock(level)
105
106
107def tearDownModule():
108    reap_children()
109
110if __name__ == "__main__":
111    unittest.main()
112