1"""This test case provides support for checking forking and wait behavior.
2
3To test different wait behavior, override the wait_impl method.
4
5We want fork1() semantics -- only the forking thread survives in the
6child after a fork().
7
8On some systems (e.g. Solaris without posix threads) we find that all
9active threads survive in the child after a fork(); this is an error.
10
11While BeOS doesn't officially support fork and native threading in
12the same application, the present example should work just fine.  DC
13"""
14
15import os, sys, time, unittest
16import test.test_support as test_support
17thread = test_support.import_module('thread')
18
19LONGSLEEP = 2
20SHORTSLEEP = 0.5
21NUM_THREADS = 4
22
23class ForkWait(unittest.TestCase):
24
25    def setUp(self):
26        self.alive = {}
27        self.stop = 0
28
29    def f(self, id):
30        while not self.stop:
31            self.alive[id] = os.getpid()
32            try:
33                time.sleep(SHORTSLEEP)
34            except IOError:
35                pass
36
37    def wait_impl(self, cpid):
38        for i in range(10):
39            # waitpid() shouldn't hang, but some of the buildbots seem to hang
40            # in the forking tests.  This is an attempt to fix the problem.
41            spid, status = os.waitpid(cpid, os.WNOHANG)
42            if spid == cpid:
43                break
44            time.sleep(2 * SHORTSLEEP)
45
46        self.assertEqual(spid, cpid)
47        self.assertEqual(status, 0, "cause = %d, exit = %d" % (status&0xff, status>>8))
48
49    def test_wait(self):
50        for i in range(NUM_THREADS):
51            thread.start_new(self.f, (i,))
52
53        time.sleep(LONGSLEEP)
54
55        a = self.alive.keys()
56        a.sort()
57        self.assertEqual(a, range(NUM_THREADS))
58
59        prefork_lives = self.alive.copy()
60
61        if sys.platform in ['unixware7']:
62            cpid = os.fork1()
63        else:
64            cpid = os.fork()
65
66        if cpid == 0:
67            # Child
68            time.sleep(LONGSLEEP)
69            n = 0
70            for key in self.alive:
71                if self.alive[key] != prefork_lives[key]:
72                    n += 1
73            os._exit(n)
74        else:
75            # Parent
76            self.wait_impl(cpid)
77            # Tell threads to die
78            self.stop = 1
79            time.sleep(2*SHORTSLEEP) # Wait for threads to die
80