ยปCore Development>Code coverage>Lib/test/test_fork1.py

Python code coverage for Lib/test/test_fork1.py

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