fork_wait.py

"""This test case provides support for checking forking and wait behavior.

To test different wait behavior, override the wait_impl method.

We want fork1() semantics -- only the forking thread survives in the
child after a fork().

On some systems (e.g. Solaris without posix threads) we find that all
active threads survive in the child after a fork(); this is an error.
"""

importos, sys, time, unittest
importthreading
importtest.supportassupport


LONGSLEEP=2
SHORTSLEEP=0.5
NUM_THREADS=4

classForkWait(unittest.TestCase):

defsetUp(self):
self._threading_key=support.threading_setup()
self.alive= {}
self.stop=0
self.threads= []

deftearDown(self):
# Stop threads
self.stop=1
forthreadinself.threads:
thread.join()
thread=None
self.threads.clear()
support.threading_cleanup(*self._threading_key)

deff(self, id):
whilenotself.stop:
self.alive[id] =os.getpid()
try:
time.sleep(SHORTSLEEP)
exceptOSError:
pass

defwait_impl(self, cpid):
foriinrange(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)
ifspid==cpid:
break
time.sleep(2*SHORTSLEEP)

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

deftest_wait(self):
foriinrange(NUM_THREADS):
thread=threading.Thread(target=self.f, args=(i,))
thread.start()
self.threads.append(thread)

# busy-loop to wait for threads
deadline=time.monotonic() +10.0
whilelen(self.alive) <NUM_THREADS:
time.sleep(0.1)
ifdeadline<time.monotonic():
break

a=sorted(self.alive.keys())
self.assertEqual(a, list(range(NUM_THREADS)))

prefork_lives=self.alive.copy()

ifsys.platformin ['unixware7']:
cpid=os.fork1()
else:
cpid=os.fork()

ifcpid==0:
# Child
time.sleep(LONGSLEEP)
n=0
forkeyinself.alive:
ifself.alive[key] !=prefork_lives[key]:
n+=1
os._exit(n)
else:
# Parent
self.wait_impl(cpid)