mirror of
https://github.com/python/cpython.git
synced 2025-07-22 18:55:22 +00:00
0d63bacefd
Replace hardcoded timeout constants in tests with SHORT_TIMEOUT of test.support, so it's easier to ajdust this timeout for all tests at once. SHORT_TIMEOUT is 30 seconds by default, but it can be longer depending on --timeout command line option. The change makes almost all timeouts longer, except test_reap_children() of test_support which is made 2x shorter: SHORT_TIMEOUT should be enough. If this test starts to fail, LONG_TIMEOUT should be used instead. Uniformize also "from test import support" import in some test files.
111 lines
3.7 KiB
Python
111 lines
3.7 KiB
Python
"""This test checks for correct fork() behavior.
|
|
"""
|
|
|
|
import _imp as imp
|
|
import os
|
|
import signal
|
|
import sys
|
|
import threading
|
|
import time
|
|
import unittest
|
|
|
|
from test.fork_wait import ForkWait
|
|
from test import support
|
|
|
|
|
|
# Skip test if fork does not exist.
|
|
support.get_attribute(os, 'fork')
|
|
|
|
class ForkTest(ForkWait):
|
|
def wait_impl(self, cpid):
|
|
deadline = time.monotonic() + support.SHORT_TIMEOUT
|
|
while time.monotonic() <= deadline:
|
|
# 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(0.1)
|
|
|
|
self.assertEqual(spid, cpid)
|
|
self.assertEqual(status, 0, "cause = %d, exit = %d" % (status&0xff, status>>8))
|
|
|
|
def test_threaded_import_lock_fork(self):
|
|
"""Check fork() in main thread works while a subthread is doing an import"""
|
|
import_started = threading.Event()
|
|
fake_module_name = "fake test module"
|
|
partial_module = "partial"
|
|
complete_module = "complete"
|
|
def importer():
|
|
imp.acquire_lock()
|
|
sys.modules[fake_module_name] = partial_module
|
|
import_started.set()
|
|
time.sleep(0.01) # Give the other thread time to try and acquire.
|
|
sys.modules[fake_module_name] = complete_module
|
|
imp.release_lock()
|
|
t = threading.Thread(target=importer)
|
|
t.start()
|
|
import_started.wait()
|
|
pid = os.fork()
|
|
try:
|
|
# PyOS_BeforeFork should have waited for the import to complete
|
|
# before forking, so the child can recreate the import lock
|
|
# correctly, but also won't see a partially initialised module
|
|
if not pid:
|
|
m = __import__(fake_module_name)
|
|
if m == complete_module:
|
|
os._exit(0)
|
|
else:
|
|
if support.verbose > 1:
|
|
print("Child encountered partial module")
|
|
os._exit(1)
|
|
else:
|
|
t.join()
|
|
# Exitcode 1 means the child got a partial module (bad.) No
|
|
# exitcode (but a hang, which manifests as 'got pid 0')
|
|
# means the child deadlocked (also bad.)
|
|
self.wait_impl(pid)
|
|
finally:
|
|
try:
|
|
os.kill(pid, signal.SIGKILL)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def test_nested_import_lock_fork(self):
|
|
"""Check fork() in main thread works while the main thread is doing an import"""
|
|
# Issue 9573: this used to trigger RuntimeError in the child process
|
|
def fork_with_import_lock(level):
|
|
release = 0
|
|
in_child = False
|
|
try:
|
|
try:
|
|
for i in range(level):
|
|
imp.acquire_lock()
|
|
release += 1
|
|
pid = os.fork()
|
|
in_child = not pid
|
|
finally:
|
|
for i in range(release):
|
|
imp.release_lock()
|
|
except RuntimeError:
|
|
if in_child:
|
|
if support.verbose > 1:
|
|
print("RuntimeError in child")
|
|
os._exit(1)
|
|
raise
|
|
if in_child:
|
|
os._exit(0)
|
|
self.wait_impl(pid)
|
|
|
|
# Check this works with various levels of nested
|
|
# import in the main thread
|
|
for level in range(5):
|
|
fork_with_import_lock(level)
|
|
|
|
|
|
def tearDownModule():
|
|
support.reap_children()
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|