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

Python code coverage for Lib/test/fork_wait.py

#countcontent
1n/a"""This test case provides support for checking forking and wait behavior.
2n/a
3n/aTo test different wait behavior, override the wait_impl method.
4n/a
5n/aWe want fork1() semantics -- only the forking thread survives in the
6n/achild after a fork().
7n/a
8n/aOn some systems (e.g. Solaris without posix threads) we find that all
9n/aactive threads survive in the child after a fork(); this is an error.
10n/a"""
11n/a
12n/aimport os, sys, time, unittest
13n/aimport test.support as support
14n/a_thread = support.import_module('_thread')
15n/a
16n/aLONGSLEEP = 2
17n/aSHORTSLEEP = 0.5
18n/aNUM_THREADS = 4
19n/a
20n/aclass ForkWait(unittest.TestCase):
21n/a
22n/a def setUp(self):
23n/a self.alive = {}
24n/a self.stop = 0
25n/a
26n/a def f(self, id):
27n/a while not self.stop:
28n/a self.alive[id] = os.getpid()
29n/a try:
30n/a time.sleep(SHORTSLEEP)
31n/a except OSError:
32n/a pass
33n/a
34n/a def wait_impl(self, cpid):
35n/a for i in range(10):
36n/a # waitpid() shouldn't hang, but some of the buildbots seem to hang
37n/a # in the forking tests. This is an attempt to fix the problem.
38n/a spid, status = os.waitpid(cpid, os.WNOHANG)
39n/a if spid == cpid:
40n/a break
41n/a time.sleep(2 * SHORTSLEEP)
42n/a
43n/a self.assertEqual(spid, cpid)
44n/a self.assertEqual(status, 0, "cause = %d, exit = %d" % (status&0xff, status>>8))
45n/a
46n/a @support.reap_threads
47n/a def test_wait(self):
48n/a for i in range(NUM_THREADS):
49n/a _thread.start_new(self.f, (i,))
50n/a
51n/a # busy-loop to wait for threads
52n/a deadline = time.monotonic() + 10.0
53n/a while len(self.alive) < NUM_THREADS:
54n/a time.sleep(0.1)
55n/a if deadline < time.monotonic():
56n/a break
57n/a
58n/a a = sorted(self.alive.keys())
59n/a self.assertEqual(a, list(range(NUM_THREADS)))
60n/a
61n/a prefork_lives = self.alive.copy()
62n/a
63n/a if sys.platform in ['unixware7']:
64n/a cpid = os.fork1()
65n/a else:
66n/a cpid = os.fork()
67n/a
68n/a if cpid == 0:
69n/a # Child
70n/a time.sleep(LONGSLEEP)
71n/a n = 0
72n/a for key in self.alive:
73n/a if self.alive[key] != prefork_lives[key]:
74n/a n += 1
75n/a os._exit(n)
76n/a else:
77n/a # Parent
78n/a try:
79n/a self.wait_impl(cpid)
80n/a finally:
81n/a # Tell threads to die
82n/a self.stop = 1