mirror of
https://github.com/python/cpython.git
synced 2025-10-17 12:18:23 +00:00
Issue #23715: signal.sigwaitinfo() and signal.sigtimedwait() are now retried
when interrupted by a signal not in the *sigset* parameter, if the signal handler does not raise an exception. signal.sigtimedwait() recomputes the timeout with a monotonic clock when it is retried. Remove test_signal.test_sigwaitinfo_interrupted() because sigwaitinfo() doesn't raise InterruptedError anymore if it is interrupted by a signal not in its sigset parameter.
This commit is contained in:
parent
a3c0202eb5
commit
a453cd8d85
5 changed files with 99 additions and 56 deletions
|
@ -264,11 +264,47 @@ class TimeEINTRTest(EINTRBaseTest):
|
|||
self.assertGreaterEqual(dt, self.sleep_time)
|
||||
|
||||
|
||||
@unittest.skipUnless(hasattr(signal, "setitimer"), "requires setitimer()")
|
||||
class SignalEINTRTest(EINTRBaseTest):
|
||||
""" EINTR tests for the signal module. """
|
||||
|
||||
def test_sigtimedwait(self):
|
||||
t0 = time.monotonic()
|
||||
signal.sigtimedwait([], self.sleep_time)
|
||||
dt = time.monotonic() - t0
|
||||
self.assertGreaterEqual(dt, self.sleep_time)
|
||||
|
||||
def test_sigwaitinfo(self):
|
||||
signum = signal.SIGUSR1
|
||||
pid = os.getpid()
|
||||
|
||||
old_handler = signal.signal(signum, lambda *args: None)
|
||||
self.addCleanup(signal.signal, signum, old_handler)
|
||||
|
||||
t0 = time.monotonic()
|
||||
child_pid = os.fork()
|
||||
if child_pid == 0:
|
||||
# child
|
||||
try:
|
||||
self._sleep()
|
||||
os.kill(pid, signum)
|
||||
finally:
|
||||
os._exit(0)
|
||||
else:
|
||||
# parent
|
||||
signal.sigwaitinfo([signum])
|
||||
dt = time.monotonic() - t0
|
||||
os.waitpid(child_pid, 0)
|
||||
|
||||
self.assertGreaterEqual(dt, self.sleep_time)
|
||||
|
||||
|
||||
def test_main():
|
||||
support.run_unittest(
|
||||
OSEINTRTest,
|
||||
SocketEINTRTest,
|
||||
TimeEINTRTest)
|
||||
TimeEINTRTest,
|
||||
SignalEINTRTest)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
|
|
@ -936,35 +936,6 @@ class PendingSignalsTests(unittest.TestCase):
|
|||
signum = signal.SIGALRM
|
||||
self.assertRaises(ValueError, signal.sigtimedwait, [signum], -1.0)
|
||||
|
||||
@unittest.skipUnless(hasattr(signal, 'sigwaitinfo'),
|
||||
'need signal.sigwaitinfo()')
|
||||
# Issue #18238: sigwaitinfo() can be interrupted on Linux (raises
|
||||
# InterruptedError), but not on AIX
|
||||
@unittest.skipIf(sys.platform.startswith("aix"),
|
||||
'signal.sigwaitinfo() cannot be interrupted on AIX')
|
||||
def test_sigwaitinfo_interrupted(self):
|
||||
self.wait_helper(signal.SIGUSR1, '''
|
||||
def test(signum):
|
||||
import errno
|
||||
|
||||
hndl_called = True
|
||||
def alarm_handler(signum, frame):
|
||||
hndl_called = False
|
||||
|
||||
signal.signal(signal.SIGALRM, alarm_handler)
|
||||
signal.alarm(1)
|
||||
try:
|
||||
signal.sigwaitinfo([signal.SIGUSR1])
|
||||
except OSError as e:
|
||||
if e.errno == errno.EINTR:
|
||||
if not hndl_called:
|
||||
raise Exception("SIGALRM handler not called")
|
||||
else:
|
||||
raise Exception("Expected EINTR to be raised by sigwaitinfo")
|
||||
else:
|
||||
raise Exception("Expected EINTR to be raised by sigwaitinfo")
|
||||
''')
|
||||
|
||||
@unittest.skipUnless(hasattr(signal, 'sigwait'),
|
||||
'need signal.sigwait()')
|
||||
@unittest.skipUnless(hasattr(signal, 'pthread_sigmask'),
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue