mirror of
https://github.com/RustPython/RustPython.git
synced 2026-06-02 19:39:49 +09:00
6b5c5a9e92
* Handle EINTR retry in os.write() (PEP 475) Add EINTR retry loop to os.write(), matching the existing pattern in os.read() and os.readinto(). Remove the expectedFailure marker from test_write in _test_eintr.py. * Add atomic snapshot for dict/dict_keys in extract_elements Add fast paths for dict and dict_keys types in extract_elements_with, matching _list_extend() in CPython Objects/listobject.c. Each branch takes an atomic snapshot under a single read lock, preventing race conditions from concurrent dict mutation without the GIL. Remove expectedFailure from test_thread_safety.
559 lines
18 KiB
Python
Vendored
559 lines
18 KiB
Python
Vendored
"""
|
|
This test suite exercises some system calls subject to interruption with EINTR,
|
|
to check that it is actually handled transparently.
|
|
It is intended to be run by the main test suite within a child process, to
|
|
ensure there is no background thread running (so that signals are delivered to
|
|
the correct thread).
|
|
Signals are generated in-process using setitimer(ITIMER_REAL), which allows
|
|
sub-second periodicity (contrarily to signal()).
|
|
"""
|
|
|
|
import contextlib
|
|
import faulthandler
|
|
import fcntl
|
|
import os
|
|
import platform
|
|
import select
|
|
import signal
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import textwrap
|
|
import time
|
|
import unittest
|
|
|
|
from test import support
|
|
from test.support import os_helper
|
|
from test.support import socket_helper
|
|
|
|
|
|
# gh-109592: Tolerate a difference of 20 ms when comparing timings
|
|
# (clock resolution)
|
|
CLOCK_RES = 0.020
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def kill_on_error(proc):
|
|
"""Context manager killing the subprocess if a Python exception is raised."""
|
|
with proc:
|
|
try:
|
|
yield proc
|
|
except:
|
|
proc.kill()
|
|
raise
|
|
|
|
|
|
@unittest.skipUnless(hasattr(signal, "setitimer"), "requires setitimer()")
|
|
class EINTRBaseTest(unittest.TestCase):
|
|
""" Base class for EINTR tests. """
|
|
|
|
# delay for initial signal delivery
|
|
signal_delay = 0.1
|
|
# signal delivery periodicity
|
|
signal_period = 0.1
|
|
# default sleep time for tests - should obviously have:
|
|
# sleep_time > signal_period
|
|
sleep_time = 0.2
|
|
|
|
def sighandler(self, signum, frame):
|
|
self.signals += 1
|
|
|
|
def setUp(self):
|
|
self.signals = 0
|
|
self.orig_handler = signal.signal(signal.SIGALRM, self.sighandler)
|
|
signal.setitimer(signal.ITIMER_REAL, self.signal_delay,
|
|
self.signal_period)
|
|
|
|
# Use faulthandler as watchdog to debug when a test hangs
|
|
# (timeout of 10 minutes)
|
|
faulthandler.dump_traceback_later(10 * 60, exit=True,
|
|
file=sys.__stderr__)
|
|
|
|
@staticmethod
|
|
def stop_alarm():
|
|
signal.setitimer(signal.ITIMER_REAL, 0, 0)
|
|
|
|
def tearDown(self):
|
|
self.stop_alarm()
|
|
signal.signal(signal.SIGALRM, self.orig_handler)
|
|
faulthandler.cancel_dump_traceback_later()
|
|
|
|
def subprocess(self, *args, **kw):
|
|
cmd_args = (sys.executable, '-c') + args
|
|
return subprocess.Popen(cmd_args, **kw)
|
|
|
|
def check_elapsed_time(self, elapsed):
|
|
self.assertGreaterEqual(elapsed, self.sleep_time - CLOCK_RES)
|
|
|
|
|
|
@unittest.skipUnless(hasattr(signal, "setitimer"), "requires setitimer()")
|
|
class OSEINTRTest(EINTRBaseTest):
|
|
""" EINTR tests for the os module. """
|
|
|
|
def new_sleep_process(self):
|
|
code = f'import time; time.sleep({self.sleep_time!r})'
|
|
return self.subprocess(code)
|
|
|
|
def _test_wait_multiple(self, wait_func):
|
|
num = 3
|
|
processes = [self.new_sleep_process() for _ in range(num)]
|
|
for _ in range(num):
|
|
wait_func()
|
|
# Call the Popen method to avoid a ResourceWarning
|
|
for proc in processes:
|
|
proc.wait()
|
|
|
|
def test_wait(self):
|
|
self._test_wait_multiple(os.wait)
|
|
|
|
@unittest.skipUnless(hasattr(os, 'wait3'), 'requires wait3()')
|
|
def test_wait3(self):
|
|
self._test_wait_multiple(lambda: os.wait3(0))
|
|
|
|
def _test_wait_single(self, wait_func):
|
|
proc = self.new_sleep_process()
|
|
wait_func(proc.pid)
|
|
# Call the Popen method to avoid a ResourceWarning
|
|
proc.wait()
|
|
|
|
def test_waitpid(self):
|
|
self._test_wait_single(lambda pid: os.waitpid(pid, 0))
|
|
|
|
@unittest.skipUnless(hasattr(os, 'wait4'), 'requires wait4()')
|
|
def test_wait4(self):
|
|
self._test_wait_single(lambda pid: os.wait4(pid, 0))
|
|
|
|
def _interrupted_reads(self):
|
|
"""Make a fd which will force block on read of expected bytes."""
|
|
rd, wr = os.pipe()
|
|
self.addCleanup(os.close, rd)
|
|
# wr closed explicitly by parent
|
|
|
|
# the payload below are smaller than PIPE_BUF, hence the writes will be
|
|
# atomic
|
|
data = [b"hello", b"world", b"spam"]
|
|
|
|
code = '\n'.join((
|
|
'import os, sys, time',
|
|
'',
|
|
'wr = int(sys.argv[1])',
|
|
f'data = {data!r}',
|
|
f'sleep_time = {self.sleep_time!r}',
|
|
'',
|
|
'for item in data:',
|
|
' # let the parent block on read()',
|
|
' time.sleep(sleep_time)',
|
|
' os.write(wr, item)',
|
|
))
|
|
|
|
proc = self.subprocess(code, str(wr), pass_fds=[wr])
|
|
with kill_on_error(proc):
|
|
os.close(wr)
|
|
for datum in data:
|
|
yield rd, datum
|
|
self.assertEqual(proc.wait(), 0)
|
|
|
|
def test_read(self):
|
|
for fd, expected in self._interrupted_reads():
|
|
self.assertEqual(expected, os.read(fd, len(expected)))
|
|
|
|
def test_readinto(self):
|
|
for fd, expected in self._interrupted_reads():
|
|
buffer = bytearray(len(expected))
|
|
self.assertEqual(os.readinto(fd, buffer), len(expected))
|
|
self.assertEqual(buffer, expected)
|
|
|
|
def test_write(self):
|
|
rd, wr = os.pipe()
|
|
self.addCleanup(os.close, wr)
|
|
# rd closed explicitly by parent
|
|
|
|
# we must write enough data for the write() to block
|
|
data = b"x" * support.PIPE_MAX_SIZE
|
|
|
|
code = '\n'.join((
|
|
'import io, os, sys, time',
|
|
'',
|
|
'rd = int(sys.argv[1])',
|
|
f'sleep_time = {self.sleep_time!r}',
|
|
f'data = b"x" * {support.PIPE_MAX_SIZE}',
|
|
'data_len = len(data)',
|
|
'',
|
|
'# let the parent block on write()',
|
|
'time.sleep(sleep_time)',
|
|
'',
|
|
'read_data = io.BytesIO()',
|
|
'while len(read_data.getvalue()) < data_len:',
|
|
' chunk = os.read(rd, 2 * data_len)',
|
|
' read_data.write(chunk)',
|
|
'',
|
|
'value = read_data.getvalue()',
|
|
'if value != data:',
|
|
' raise Exception(f"read error: {len(value)}'
|
|
' vs {data_len} bytes")',
|
|
))
|
|
|
|
proc = self.subprocess(code, str(rd), pass_fds=[rd])
|
|
with kill_on_error(proc):
|
|
os.close(rd)
|
|
written = 0
|
|
while written < len(data):
|
|
written += os.write(wr, memoryview(data)[written:])
|
|
self.assertEqual(proc.wait(), 0)
|
|
|
|
|
|
@unittest.skipUnless(hasattr(signal, "setitimer"), "requires setitimer()")
|
|
class SocketEINTRTest(EINTRBaseTest):
|
|
""" EINTR tests for the socket module. """
|
|
|
|
@unittest.skipUnless(hasattr(socket, 'socketpair'), 'needs socketpair()')
|
|
def _test_recv(self, recv_func):
|
|
rd, wr = socket.socketpair()
|
|
self.addCleanup(rd.close)
|
|
# wr closed explicitly by parent
|
|
|
|
# single-byte payload guard us against partial recv
|
|
data = [b"x", b"y", b"z"]
|
|
|
|
code = '\n'.join((
|
|
'import os, socket, sys, time',
|
|
'',
|
|
'fd = int(sys.argv[1])',
|
|
f'family = {int(wr.family)}',
|
|
f'sock_type = {int(wr.type)}',
|
|
f'data = {data!r}',
|
|
f'sleep_time = {self.sleep_time!r}',
|
|
'',
|
|
'wr = socket.fromfd(fd, family, sock_type)',
|
|
'os.close(fd)',
|
|
'',
|
|
'with wr:',
|
|
' for item in data:',
|
|
' # let the parent block on recv()',
|
|
' time.sleep(sleep_time)',
|
|
' wr.sendall(item)',
|
|
))
|
|
|
|
fd = wr.fileno()
|
|
proc = self.subprocess(code, str(fd), pass_fds=[fd])
|
|
with kill_on_error(proc):
|
|
wr.close()
|
|
for item in data:
|
|
self.assertEqual(item, recv_func(rd, len(item)))
|
|
self.assertEqual(proc.wait(), 0)
|
|
|
|
def test_recv(self):
|
|
self._test_recv(socket.socket.recv)
|
|
|
|
@unittest.skipUnless(hasattr(socket.socket, 'recvmsg'), 'needs recvmsg()')
|
|
def test_recvmsg(self):
|
|
self._test_recv(lambda sock, data: sock.recvmsg(data)[0])
|
|
|
|
def _test_send(self, send_func):
|
|
rd, wr = socket.socketpair()
|
|
self.addCleanup(wr.close)
|
|
# rd closed explicitly by parent
|
|
|
|
# we must send enough data for the send() to block
|
|
data = b"xyz" * (support.SOCK_MAX_SIZE // 3)
|
|
|
|
code = '\n'.join((
|
|
'import os, socket, sys, time',
|
|
'',
|
|
'fd = int(sys.argv[1])',
|
|
f'family = {int(rd.family)}',
|
|
f'sock_type = {int(rd.type)}',
|
|
f'sleep_time = {self.sleep_time!r}',
|
|
f'data = b"xyz" * {support.SOCK_MAX_SIZE // 3}',
|
|
'data_len = len(data)',
|
|
'',
|
|
'rd = socket.fromfd(fd, family, sock_type)',
|
|
'os.close(fd)',
|
|
'',
|
|
'with rd:',
|
|
' # let the parent block on send()',
|
|
' time.sleep(sleep_time)',
|
|
'',
|
|
' received_data = bytearray(data_len)',
|
|
' n = 0',
|
|
' while n < data_len:',
|
|
' n += rd.recv_into(memoryview(received_data)[n:])',
|
|
'',
|
|
'if received_data != data:',
|
|
' raise Exception(f"recv error: {len(received_data)}'
|
|
' vs {data_len} bytes")',
|
|
))
|
|
|
|
fd = rd.fileno()
|
|
proc = self.subprocess(code, str(fd), pass_fds=[fd])
|
|
with kill_on_error(proc):
|
|
rd.close()
|
|
written = 0
|
|
while written < len(data):
|
|
sent = send_func(wr, memoryview(data)[written:])
|
|
# sendall() returns None
|
|
written += len(data) if sent is None else sent
|
|
self.assertEqual(proc.wait(), 0)
|
|
|
|
def test_send(self):
|
|
self._test_send(socket.socket.send)
|
|
|
|
def test_sendall(self):
|
|
self._test_send(socket.socket.sendall)
|
|
|
|
@unittest.skipUnless(hasattr(socket.socket, 'sendmsg'), 'needs sendmsg()')
|
|
def test_sendmsg(self):
|
|
self._test_send(lambda sock, data: sock.sendmsg([data]))
|
|
|
|
def test_accept(self):
|
|
sock = socket.create_server((socket_helper.HOST, 0))
|
|
self.addCleanup(sock.close)
|
|
port = sock.getsockname()[1]
|
|
|
|
code = '\n'.join((
|
|
'import socket, time',
|
|
'',
|
|
f'host = {socket_helper.HOST!r}',
|
|
f'port = {port}',
|
|
f'sleep_time = {self.sleep_time!r}',
|
|
'',
|
|
'# let parent block on accept()',
|
|
'time.sleep(sleep_time)',
|
|
'with socket.create_connection((host, port)):',
|
|
' time.sleep(sleep_time)',
|
|
))
|
|
|
|
proc = self.subprocess(code)
|
|
with kill_on_error(proc):
|
|
client_sock, _ = sock.accept()
|
|
client_sock.close()
|
|
self.assertEqual(proc.wait(), 0)
|
|
|
|
# Issue #25122: There is a race condition in the FreeBSD kernel on
|
|
# handling signals in the FIFO device. Skip the test until the bug is
|
|
# fixed in the kernel.
|
|
# https://bugs.freebsd.org/bugzilla/show_bug.cgi?id=203162
|
|
@support.requires_freebsd_version(10, 3)
|
|
@unittest.skipUnless(hasattr(os, 'mkfifo'), 'needs mkfifo()')
|
|
def _test_open(self, do_open_close_reader, do_open_close_writer):
|
|
filename = os_helper.TESTFN
|
|
|
|
# Use a fifo: until the child opens it for reading, the parent will
|
|
# block when trying to open it for writing.
|
|
os_helper.unlink(filename)
|
|
try:
|
|
os.mkfifo(filename)
|
|
except PermissionError as exc:
|
|
self.skipTest(f'os.mkfifo(): {exc!r}')
|
|
self.addCleanup(os_helper.unlink, filename)
|
|
|
|
code = '\n'.join((
|
|
'import os, time',
|
|
'',
|
|
f'path = {filename!a}',
|
|
f'sleep_time = {self.sleep_time!r}',
|
|
'',
|
|
'# let the parent block',
|
|
'time.sleep(sleep_time)',
|
|
'',
|
|
do_open_close_reader,
|
|
))
|
|
|
|
proc = self.subprocess(code)
|
|
with kill_on_error(proc):
|
|
do_open_close_writer(filename)
|
|
self.assertEqual(proc.wait(), 0)
|
|
|
|
def python_open(self, path):
|
|
fp = open(path, 'w')
|
|
fp.close()
|
|
|
|
@unittest.skipIf(sys.platform == "darwin",
|
|
"hangs under macOS; see bpo-25234, bpo-35363")
|
|
def test_open(self):
|
|
self._test_open("fp = open(path, 'r')\nfp.close()",
|
|
self.python_open)
|
|
|
|
def os_open(self, path):
|
|
fd = os.open(path, os.O_WRONLY)
|
|
os.close(fd)
|
|
|
|
@unittest.skipIf(sys.platform == "darwin",
|
|
"hangs under macOS; see bpo-25234, bpo-35363")
|
|
@unittest.skipIf(sys.platform.startswith('netbsd'),
|
|
"hangs on NetBSD; see gh-137397")
|
|
def test_os_open(self):
|
|
self._test_open("fd = os.open(path, os.O_RDONLY)\nos.close(fd)",
|
|
self.os_open)
|
|
|
|
|
|
@unittest.skipUnless(hasattr(signal, "setitimer"), "requires setitimer()")
|
|
class TimeEINTRTest(EINTRBaseTest):
|
|
""" EINTR tests for the time module. """
|
|
|
|
def test_sleep(self):
|
|
t0 = time.monotonic()
|
|
time.sleep(self.sleep_time)
|
|
self.stop_alarm()
|
|
dt = time.monotonic() - t0
|
|
self.check_elapsed_time(dt)
|
|
|
|
|
|
@unittest.skipUnless(hasattr(signal, "setitimer"), "requires setitimer()")
|
|
# bpo-30320: Need pthread_sigmask() to block the signal, otherwise the test
|
|
# is vulnerable to a race condition between the child and the parent processes.
|
|
@unittest.skipUnless(hasattr(signal, 'pthread_sigmask'),
|
|
'need signal.pthread_sigmask()')
|
|
class SignalEINTRTest(EINTRBaseTest):
|
|
""" EINTR tests for the signal module. """
|
|
|
|
def check_sigwait(self, wait_func):
|
|
signum = signal.SIGUSR1
|
|
|
|
old_handler = signal.signal(signum, lambda *args: None)
|
|
self.addCleanup(signal.signal, signum, old_handler)
|
|
|
|
code = '\n'.join((
|
|
'import os, time',
|
|
f'pid = {os.getpid()}',
|
|
f'signum = {int(signum)}',
|
|
f'sleep_time = {self.sleep_time!r}',
|
|
'time.sleep(sleep_time)',
|
|
'os.kill(pid, signum)',
|
|
))
|
|
|
|
signal.pthread_sigmask(signal.SIG_BLOCK, [signum])
|
|
self.addCleanup(signal.pthread_sigmask, signal.SIG_UNBLOCK, [signum])
|
|
|
|
proc = self.subprocess(code)
|
|
with kill_on_error(proc):
|
|
wait_func(signum)
|
|
|
|
self.assertEqual(proc.wait(), 0)
|
|
|
|
@unittest.skipUnless(hasattr(signal, 'sigwaitinfo'),
|
|
'need signal.sigwaitinfo()')
|
|
def test_sigwaitinfo(self):
|
|
def wait_func(signum):
|
|
signal.sigwaitinfo([signum])
|
|
|
|
self.check_sigwait(wait_func)
|
|
|
|
@unittest.skipUnless(hasattr(signal, 'sigtimedwait'),
|
|
'need signal.sigwaitinfo()')
|
|
def test_sigtimedwait(self):
|
|
def wait_func(signum):
|
|
signal.sigtimedwait([signum], 120.0)
|
|
|
|
self.check_sigwait(wait_func)
|
|
|
|
|
|
@unittest.skipUnless(hasattr(signal, "setitimer"), "requires setitimer()")
|
|
class SelectEINTRTest(EINTRBaseTest):
|
|
""" EINTR tests for the select module. """
|
|
|
|
@unittest.expectedFailure # TODO: RUSTPYTHON
|
|
def test_select(self):
|
|
t0 = time.monotonic()
|
|
select.select([], [], [], self.sleep_time)
|
|
dt = time.monotonic() - t0
|
|
self.stop_alarm()
|
|
self.check_elapsed_time(dt)
|
|
|
|
@unittest.skip("TODO: RUSTPYTHON; timed out at the 10 minute mark")
|
|
@unittest.skipIf(sys.platform == "darwin",
|
|
"poll may fail on macOS; see issue #28087")
|
|
@unittest.skipUnless(hasattr(select, 'poll'), 'need select.poll')
|
|
def test_poll(self):
|
|
poller = select.poll()
|
|
|
|
t0 = time.monotonic()
|
|
poller.poll(self.sleep_time * 1e3)
|
|
dt = time.monotonic() - t0
|
|
self.stop_alarm()
|
|
self.check_elapsed_time(dt)
|
|
|
|
@unittest.skipUnless(hasattr(select, 'epoll'), 'need select.epoll')
|
|
def test_epoll(self):
|
|
poller = select.epoll()
|
|
self.addCleanup(poller.close)
|
|
|
|
t0 = time.monotonic()
|
|
poller.poll(self.sleep_time)
|
|
dt = time.monotonic() - t0
|
|
self.stop_alarm()
|
|
self.check_elapsed_time(dt)
|
|
|
|
@unittest.skipUnless(hasattr(select, 'kqueue'), 'need select.kqueue')
|
|
def test_kqueue(self):
|
|
kqueue = select.kqueue()
|
|
self.addCleanup(kqueue.close)
|
|
|
|
t0 = time.monotonic()
|
|
kqueue.control(None, 1, self.sleep_time)
|
|
dt = time.monotonic() - t0
|
|
self.stop_alarm()
|
|
self.check_elapsed_time(dt)
|
|
|
|
@unittest.skipUnless(hasattr(select, 'devpoll'), 'need select.devpoll')
|
|
def test_devpoll(self):
|
|
poller = select.devpoll()
|
|
self.addCleanup(poller.close)
|
|
|
|
t0 = time.monotonic()
|
|
poller.poll(self.sleep_time * 1e3)
|
|
dt = time.monotonic() - t0
|
|
self.stop_alarm()
|
|
self.check_elapsed_time(dt)
|
|
|
|
|
|
class FCNTLEINTRTest(EINTRBaseTest):
|
|
def _lock(self, lock_func, lock_name):
|
|
self.addCleanup(os_helper.unlink, os_helper.TESTFN)
|
|
rd1, wr1 = os.pipe()
|
|
rd2, wr2 = os.pipe()
|
|
for fd in (rd1, wr1, rd2, wr2):
|
|
self.addCleanup(os.close, fd)
|
|
code = textwrap.dedent(f"""
|
|
import fcntl, os, time
|
|
with open('{os_helper.TESTFN}', 'wb') as f:
|
|
fcntl.{lock_name}(f, fcntl.LOCK_EX)
|
|
os.write({wr1}, b"ok")
|
|
_ = os.read({rd2}, 2) # wait for parent process
|
|
time.sleep({self.sleep_time})
|
|
""")
|
|
proc = self.subprocess(code, pass_fds=[wr1, rd2])
|
|
with kill_on_error(proc):
|
|
with open(os_helper.TESTFN, 'wb') as f:
|
|
# synchronize the subprocess
|
|
ok = os.read(rd1, 2)
|
|
self.assertEqual(ok, b"ok")
|
|
|
|
# notify the child that the parent is ready
|
|
start_time = time.monotonic()
|
|
os.write(wr2, b"go")
|
|
|
|
# the child locked the file just a moment ago for 'sleep_time' seconds
|
|
# that means that the lock below will block for 'sleep_time' minus some
|
|
# potential context switch delay
|
|
lock_func(f, fcntl.LOCK_EX)
|
|
dt = time.monotonic() - start_time
|
|
self.stop_alarm()
|
|
self.check_elapsed_time(dt)
|
|
proc.wait()
|
|
|
|
# Issue 35633: See https://bugs.python.org/issue35633#msg333662
|
|
# skip test rather than accept PermissionError from all platforms
|
|
@unittest.expectedFailure # TODO: RUSTPYTHON; InterruptedError: [Errno 4] Interrupted system call
|
|
@unittest.skipIf(platform.system() == "AIX", "AIX returns PermissionError")
|
|
def test_lockf(self):
|
|
self._lock(fcntl.lockf, "lockf")
|
|
|
|
@unittest.expectedFailure # TODO: RUSTPYTHON; InterruptedError: [Errno 4] Interrupted system call
|
|
def test_flock(self):
|
|
self._lock(fcntl.flock, "flock")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|