summaryrefslogtreecommitdiffstats
path: root/Lib/test/test_fork1.py
blob: 8192c38a4491d4d11a045425f946ba8d83c744e0 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
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
106
107
108
109
110
111
"""This test checks for correct fork() behavior.
"""

import imp
import os
import signal
import sys
import time

from test.fork_wait import ForkWait
from test.support import (run_unittest, reap_children, get_attribute,
                          import_module, verbose)

threading = import_module('threading')

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

class ForkTest(ForkWait):
    def wait_impl(self, cpid):
        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)

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

    def test_threaded_import_lock_fork(self):
        """Check fork() in main thread works while a subthread is doing an import"""
        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:
            # 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
            if not pid:
                m = __import__(fake_module_name)
                if m == complete_module:
                    os._exit(0)
                else:
                    if verbose > 1:
                        print("Child encountered partial module")
                    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


    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)


def test_main():
    run_unittest(ForkTest)
    reap_children()

if __name__ == "__main__":
    test_main()