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