mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-08-29 05:05:03 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 8

Issue #12363: improve siginterrupt() tests

Browse source 
Backport commits 968b9ff9a059 and aff0a7b0cb12 from the default branch to 3.2
branch. Extract of the changelog messages:

"The previous tests used time.sleep() to synchronize two processes. If the host
was too slow, the test could fail.

The new tests only use one process, but they use a subprocess to:

- have only one thread
- have a timeout on the blocking read (select cannot be used in the test,
select always fail with EINTR, the kernel doesn't restart it)
- not touch signal handling of the parent process"

and

"Add a basic synchronization code between the child and the parent processes:
the child writes "ready" to stdout."

I replaced .communicate(timeout=3.0) by an explicit waiting loop using
Popen.poll().
This commit is contained in:
Victor Stinner 2011-07-01 15:24:50 +02:00
parent 26d318690f
commit 8d64248c69
1 changed files with 82 additions and 85 deletions
Download patch file Download diff file Expand all files Collapse all files

167
Lib/test/test_signal.py
View file

@ -1,13 +1,17 @@
import unittest import errno
from test import support
from contextlib import closing
import gc import gc
import os
import pickle import pickle
import select import select
import signal import signal
import subprocess import subprocess
import sys
import time
import traceback import traceback
import sys, os, time, errno import unittest
from test import support
from contextlib import closing
from test.script_helper import spawn_python
if sys.platform in ('os2', 'riscos'): if sys.platform in ('os2', 'riscos'):
raise unittest.SkipTest("Can't test signal on %s" % sys.platform) raise unittest.SkipTest("Can't test signal on %s" % sys.platform)
@ -276,103 +280,96 @@ class WakeupSignalTests(unittest.TestCase):
@unittest.skipIf(sys.platform == "win32", "Not valid on Windows") @unittest.skipIf(sys.platform == "win32", "Not valid on Windows")
class SiginterruptTest(unittest.TestCase): class SiginterruptTest(unittest.TestCase):
def setUp(self): def readpipe_interrupted(self, interrupt):
"""Install a no-op signal handler that can be set to allow
interrupts or not, and arrange for the original signal handler to be
re-installed when the test is finished.
"""
self.signum = signal.SIGUSR1
oldhandler = signal.signal(self.signum, lambda x,y: None)
self.addCleanup(signal.signal, self.signum, oldhandler)
def readpipe_interrupted(self):
"""Perform a read during which a signal will arrive. Return True if the """Perform a read during which a signal will arrive. Return True if the
read is interrupted by the signal and raises an exception. Return False read is interrupted by the signal and raises an exception. Return False
if it returns normally. if it returns normally.
""" """
# Create a pipe that can be used for the read. Also clean it up class Timeout(Exception):
# when the test is over, since nothing else will (but see below for pass
# the write end).
r, w = os.pipe()
self.addCleanup(os.close, r)
# Create another process which can send a signal to this one to try # use a subprocess to have only one thread, to have a timeout on the
# to interrupt the read. # blocking read and to not touch signal handling in this process
ppid = os.getpid() code = """if 1:
pid = os.fork() import errno
import os
import signal
import sys
if pid == 0: interrupt = %r
# Child code: sleep to give the parent enough time to enter the r, w = os.pipe()
# read() call (there's a race here, but it's really tricky to
# eliminate it); then signal the parent process. Also, sleep def handler(signum, frame):
# again to make it likely that the signal is delivered to the pass
# parent process before the child exits. If the child exits
# first, the write end of the pipe will be closed and the test print("ready")
# is invalid. sys.stdout.flush()
signal.signal(signal.SIGALRM, handler)
if interrupt is not None:
signal.siginterrupt(signal.SIGALRM, interrupt)
# run the test twice
for loop in range(2):
# send a SIGALRM in a second (during the read)
signal.alarm(1)
try:
# blocking call: read from a pipe without data
os.read(r, 1)
except OSError as err:
if err.errno != errno.EINTR:
raise
else:
sys.exit(2)
sys.exit(3)
""" % (interrupt,)
with spawn_python('-c', code) as process:
try: try:
time.sleep(0.2) # wait until the child process is loaded and has started
os.kill(ppid, self.signum) first_line = process.stdout.readline()
time.sleep(0.2)
finally:
# No matter what, just exit as fast as possible now.
exit_subprocess()
else:
# Parent code.
# Make sure the child is eventually reaped, else it'll be a
# zombie for the rest of the test suite run.
self.addCleanup(os.waitpid, pid, 0)
# Close the write end of the pipe. The child has a copy, so # Wait the process with a timeout of 3 seconds
# it's not really closed until the child exits. We need it to timeout = time.time() + 3.0
# close when the child exits so that in the non-interrupt case while True:
# the read eventually completes, otherwise we could just close if timeout < time.time():
# it *after* the test. raise Timeout()
os.close(w) status = process.poll()
if status is not None:
break
time.sleep(0.1)
# Try the read and report whether it is interrupted or not to stdout, stderr = process.communicate()
# the caller. except Timeout:
try: process.kill()
d = os.read(r, 1)
return False return False
except OSError as err: else:
if err.errno != errno.EINTR: stdout = first_line + stdout
raise exitcode = process.wait()
return True if exitcode not in (2, 3):
raise Exception("Child error (exit code %s): %s"
% (exitcode, stdout))
return (exitcode == 3)
def test_without_siginterrupt(self): def test_without_siginterrupt(self):
"""If a signal handler is installed and siginterrupt is not called # If a signal handler is installed and siginterrupt is not called
at all, when that signal arrives, it interrupts a syscall that's in # at all, when that signal arrives, it interrupts a syscall that's in
progress. # progress.
""" interrupted = self.readpipe_interrupted(None)
i = self.readpipe_interrupted() self.assertTrue(interrupted)
self.assertTrue(i)
# Arrival of the signal shouldn't have changed anything.
i = self.readpipe_interrupted()
self.assertTrue(i)
def test_siginterrupt_on(self): def test_siginterrupt_on(self):
"""If a signal handler is installed and siginterrupt is called with # If a signal handler is installed and siginterrupt is called with
a true value for the second argument, when that signal arrives, it # a true value for the second argument, when that signal arrives, it
interrupts a syscall that's in progress. # interrupts a syscall that's in progress.
""" interrupted = self.readpipe_interrupted(True)
signal.siginterrupt(self.signum, 1) self.assertTrue(interrupted)
i = self.readpipe_interrupted()
self.assertTrue(i)
# Arrival of the signal shouldn't have changed anything.
i = self.readpipe_interrupted()
self.assertTrue(i)
def test_siginterrupt_off(self): def test_siginterrupt_off(self):
"""If a signal handler is installed and siginterrupt is called with # If a signal handler is installed and siginterrupt is called with
a false value for the second argument, when that signal arrives, it # a false value for the second argument, when that signal arrives, it
does not interrupt a syscall that's in progress. # does not interrupt a syscall that's in progress.
""" interrupted = self.readpipe_interrupted(False)
signal.siginterrupt(self.signum, 0) self.assertFalse(interrupted)
i = self.readpipe_interrupted()
self.assertFalse(i)
# Arrival of the signal shouldn't have changed anything.
i = self.readpipe_interrupted()
self.assertFalse(i)
@unittest.skipIf(sys.platform == "win32", "Not valid on Windows") @unittest.skipIf(sys.platform == "win32", "Not valid on Windows")