Viewing file:  fork_wait.py (2.29 KB) -rw-r--r--
Select action/file-type:
 (+) |  (+) |  (+) | Code (+) | Session (+) |  (+) | SDB (+) |  (+) |  (+) |  (+) |  (+) |  (+) |
"""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.
"""
import os, time, unittest
import threading
from test import support
from test.support import threading_helper
import warnings
LONGSLEEP = 2
SHORTSLEEP = 0.5
NUM_THREADS = 4
class ForkWait(unittest.TestCase):
def setUp(self):
self._threading_key = threading_helper.threading_setup()
self.alive = {}
self.stop = 0
self.threads = []
def tearDown(self):
# Stop threads
self.stop = 1
for thread in self.threads:
thread.join()
thread = None
self.threads.clear()
threading_helper.threading_cleanup(*self._threading_key)
def f(self, id):
while not self.stop:
self.alive[id] = os.getpid()
try:
time.sleep(SHORTSLEEP)
except OSError:
pass
def wait_impl(self, cpid, *, exitcode):
support.wait_process(cpid, exitcode=exitcode)
def test_wait(self):
for i in range(NUM_THREADS):
thread = threading.Thread(target=self.f, args=(i,))
thread.start()
self.threads.append(thread)
# busy-loop to wait for threads
for _ in support.sleeping_retry(support.SHORT_TIMEOUT):
if len(self.alive) >= NUM_THREADS:
break
a = sorted(self.alive.keys())
self.assertEqual(a, list(range(NUM_THREADS)))
prefork_lives = self.alive.copy()
# Ignore the warning about fork with threads.
with warnings.catch_warnings(category=DeprecationWarning,
action="ignore"):
if (cpid := os.fork()) == 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, exitcode=0)
|