cpython/Lib/test/fork_wait.py
Christian Heimes 05e8be17fd Merged revisions 60990-61002 via svnmerge from
svn+ssh://pythondev@svn.python.org/python/trunk

........
  r60990 | eric.smith | 2008-02-23 17:05:26 +0100 (Sat, 23 Feb 2008) | 1 line

  Removed duplicate Py_CHARMASK define.  It's already defined in Python.h.
........
  r60991 | andrew.kuchling | 2008-02-23 17:23:05 +0100 (Sat, 23 Feb 2008) | 4 lines

  #1330538: Improve comparison of xmlrpclib.DateTime and datetime instances.
  Remove automatic handling of datetime.date and datetime.time.
  This breaks backward compatibility, but python-dev discussion was strongly
  against this automatic conversion; see the bug for a link.
........
  r60994 | andrew.kuchling | 2008-02-23 17:39:43 +0100 (Sat, 23 Feb 2008) | 1 line

  #835521: Add index entries for various pickle-protocol methods and attributes
........
  r60995 | andrew.kuchling | 2008-02-23 18:10:46 +0100 (Sat, 23 Feb 2008) | 2 lines

  #1433694: minidom's .normalize() failed to set .nextSibling for last element.
  Fix by Malte Helmert
........
  r61000 | christian.heimes | 2008-02-23 18:40:11 +0100 (Sat, 23 Feb 2008) | 1 line

  Patch #2167 from calvin: Remove unused imports
........
  r61001 | christian.heimes | 2008-02-23 18:42:31 +0100 (Sat, 23 Feb 2008) | 1 line

  Patch #1957: syslogmodule: Release GIL when calling syslog(3)
........
  r61002 | christian.heimes | 2008-02-23 18:52:07 +0100 (Sat, 23 Feb 2008) | 2 lines

  Issue #2051 and patch from Alexander Belopolsky:
  Permission for pyc and pyo files are inherited from the py file.
........
2008-02-23 18:30:17 +00:00

74 lines
2.0 KiB
Python

"""This test case provides support for checking forking and wait behavior.
To test different wait behavior, overrise 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.
"""
import os, sys, time, thread, unittest
LONGSLEEP = 2
SHORTSLEEP = 0.5
NUM_THREADS = 4
class ForkWait(unittest.TestCase):
def setUp(self):
self.alive = {}
self.stop = 0
def f(self, id):
while not self.stop:
self.alive[id] = os.getpid()
try:
time.sleep(SHORTSLEEP)
except IOError:
pass
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(2 * SHORTSLEEP)
self.assertEquals(spid, cpid)
self.assertEquals(status, 0, "cause = %d, exit = %d" % (status&0xff, status>>8))
def test_wait(self):
for i in range(NUM_THREADS):
thread.start_new(self.f, (i,))
time.sleep(LONGSLEEP)
a = sorted(self.alive.keys())
self.assertEquals(a, list(range(NUM_THREADS)))
prefork_lives = self.alive.copy()
if sys.platform in ['unixware7']:
cpid = os.fork1()
else:
cpid = os.fork()
if cpid == 0:
# Child
time.sleep(LONGSLEEP)
n = 0
for key in self.alive:
if self.alive[key] != prefork_lives[key]:
n += 1
os._exit(n)
else:
# Parent
self.wait_impl(cpid)
# Tell threads to die
self.stop = 1
time.sleep(2*SHORTSLEEP) # Wait for threads to die