Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
33 changes: 28 additions & 5 deletions crates/host_env/src/signal.rs
Original file line number Diff line number Diff line change
Expand Up @@ -395,13 +395,28 @@ pub fn wakeup_fd_is_socket(fd: libc::SOCKET) -> io::Result<bool> {
Ok(false)
}

#[cfg(windows)]
pub struct WakeupSendError(pub i32);

#[cfg(windows)]
impl WakeupSendError {
pub fn is_would_block(&self) -> bool {
self.0 == windows_sys::Win32::Networking::WinSock::WSAEWOULDBLOCK
}

// Only convert to an allocating error outside the OS signal handler.
pub fn into_io_error(self) -> io::Error {
io::Error::from_raw_os_error(self.0)
}
}

#[cfg(windows)]
pub fn notify_signal(
signum: i32,
wakeup_fd: libc::SOCKET,
wakeup_is_socket: bool,
sigint_event: Option<isize>,
) {
) -> Option<WakeupSendError> {
if signum == libc::SIGINT
&& let Some(handle) = sigint_event
{
Expand All @@ -411,23 +426,31 @@ pub fn notify_signal(
}

if wakeup_fd == INVALID_SOCKET {
return;
return None;
}

let sigbyte = signum as u8;
if wakeup_is_socket {
unsafe {
let _ = windows_sys::Win32::Networking::WinSock::send(
let result = unsafe {
windows_sys::Win32::Networking::WinSock::send(
wakeup_fd,
&sigbyte as *const u8 as *const _,
1,
0,
);
)
};
if result < 0 {
Some(WakeupSendError(unsafe {
windows_sys::Win32::Networking::WinSock::WSAGetLastError()
}))
} else {
None
}
} else {
unsafe {
let _ = libc::write(wakeup_fd as _, &sigbyte as *const u8 as *const _, 1);
}
None
}
}

Expand Down
10 changes: 10 additions & 0 deletions crates/vm/src/signal.rs
Original file line number Diff line number Diff line change
Expand Up @@ -103,6 +103,16 @@ fn trigger_signals(vm: &VirtualMachine) -> PyResult<()> {
}
let _guard = SignalHandlerGuard;

#[cfg(all(feature = "host_env", windows))]
if vm.is_main_thread() && vm.state.is_main_interpreter() {
// Report before a Python signal handler can raise out of the current
// error-reporting scope.
crate::stdlib::_signal::_signal::report_wakeup_send_errors(vm);
} else if crate::stdlib::_signal::_signal::wakeup_send_errors_pending() {
// A different interpreter/thread must leave the report for the main one.
set_triggered();
}

let signal_handlers = vm
.signal_handlers
.get()
Expand Down
63 changes: 58 additions & 5 deletions crates/vm/src/stdlib/_signal.rs
Original file line number Diff line number Diff line change
Expand Up @@ -73,6 +73,13 @@ pub(crate) mod _signal {
}
}

#[cfg(windows)]
static WARN_ON_FULL_BUFFER: atomic::AtomicBool = atomic::AtomicBool::new(true);

// Bounded pending-call storage: signal handlers cannot allocate or lock.
#[cfg(windows)]
static WAKEUP_SEND_ERRORS: [atomic::AtomicI32; 32] = [const { atomic::AtomicI32::new(0) }; 32];

#[cfg(any(unix, windows, target_os = "wasi"))]
#[allow(unused_imports)]
pub use host_signal::SIG_ERR;
Expand Down Expand Up @@ -344,7 +351,7 @@ pub(crate) mod _signal {

#[pyfunction]
fn set_wakeup_fd(args: SetWakeupFdArgs, vm: &VirtualMachine) -> PyResult<i64> {
// TODO: implement warn_on_full_buffer
// TODO: implement warn_on_full_buffer for the remaining wakeup write paths.
let _ = args.warn_on_full_buffer;
let fd = cfg_select! {
windows => args.fd.0,
Expand Down Expand Up @@ -383,7 +390,10 @@ pub(crate) mod _signal {
let old_fd = WAKEUP.swap(fd, Ordering::Relaxed);

#[cfg(windows)]
WAKEUP_IS_SOCKET.store(is_socket, Ordering::Relaxed);
{
WARN_ON_FULL_BUFFER.store(args.warn_on_full_buffer, Ordering::Relaxed);
WAKEUP_IS_SOCKET.store(is_socket, Ordering::Relaxed);
}

#[cfg(windows)]
if old_fd == INVALID_WAKEUP {
Expand Down Expand Up @@ -532,14 +542,57 @@ pub(crate) mod _signal {
signal::TRIGGERS[signum as usize].store(true, Ordering::Relaxed);
signal::set_triggered();

host_signal::notify_signal(
#[cfg(not(windows))]
host_signal::notify_signal(signum, WAKEUP.load(Ordering::Relaxed));

#[cfg(windows)]
let error = host_signal::notify_signal(
signum,
WAKEUP.load(Ordering::Relaxed),
#[cfg(windows)]
WAKEUP_IS_SOCKET.load(Ordering::Relaxed),
#[cfg(windows)]
signal::get_sigint_event(),
);
#[cfg(windows)]
if let Some(error) = error
&& (WARN_ON_FULL_BUFFER.load(Ordering::Relaxed) || !error.is_would_block())
{
for pending in &WAKEUP_SEND_ERRORS {
if pending
.compare_exchange(0, error.0, Ordering::Relaxed, Ordering::Relaxed)
.is_ok()
{
// The VM may have handled the signal on another thread
// before the failed wakeup send completed.
signal::set_triggered();
break;
}
}
}
}

#[cfg(windows)]
pub(crate) fn wakeup_send_errors_pending() -> bool {
WAKEUP_SEND_ERRORS
.iter()
.any(|error| error.load(Ordering::Relaxed) != 0)
}

#[cfg(windows)]
pub(crate) fn report_wakeup_send_errors(vm: &VirtualMachine) {
for pending in &WAKEUP_SEND_ERRORS {
let error = pending.swap(0, Ordering::Relaxed);
if error != 0 {
vm.run_unraisable(
host_signal::WakeupSendError(error)
.into_io_error()
.into_pyexception(vm),
Some(
"Exception ignored while trying to send to the signal wakeup fd".to_owned(),
),
vm.ctx.none(),
);
}
}
}

/// Reset wakeup fd after fork in child process.
Expand Down
112 changes: 112 additions & 0 deletions extra_tests/snippets/stdlib_signal.py
Original file line number Diff line number Diff line change
Expand Up @@ -106,3 +106,115 @@ def arm_other(signum, frame):
signal.signal(signal.SIGALRM, signal.SIG_DFL)
signal.signal(signal.SIGUSR1, signal.SIG_DFL)
signal.signal(signal.SIGUSR2, signal.SIG_DFL)


def check_windows_wakeup_reporting():

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Remove the disallowed test-logic addition

This adds a new test function with many assertions to a test file, but the repository's critical test policy forbids modifying test assertions, logic, or data and limits acceptable test-file edits to expected-failure decorator changes. Drop this test-file addition or obtain an explicit policy exception before landing the runtime change.

AGENTS.md reference: AGENTS.md:L273-L279

Useful? React with 👍 / 👎.

if sys.platform != "win32":
return

import errno
import socket

def fill_socket(write):
# A nonblocking Windows send can briefly fail before TCP has filled
# the receiving socket. Require repeated one-byte failures to settle.
deadline = time.monotonic() + 5.0
written = 0
blocked = 0
for _ in range(4096):
assert time.monotonic() < deadline, "socket fill timed out"
assert written < 16 * 1024 * 1024, "socket fill exceeded byte bound"
try:
written += write.send(b"x" * 65536)
except BlockingIOError:
try:
written += write.send(b"x")
except BlockingIOError:
blocked += 1
if blocked == 4:
return
time.sleep(0.050)
continue
blocked = 0
raise AssertionError("socket did not fill within the attempt bound")

# Use numeric loopback addresses and bounded connection/receive calls.
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as listener:
listener.settimeout(5.0)
listener.bind(("127.0.0.1", 0))
listener.listen(1)
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as write:
write.settimeout(5.0)
write.connect(listener.getsockname())
read, _ = listener.accept()
with read:
read.settimeout(5.0)
read.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 4096)
write.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 4096)
write.setblocking(False)
old_handler = signal.getsignal(signal.SIGINT)
old_hook = sys.unraisablehook
old_fd = signal.set_wakeup_fd(-1)
errors = []
delivered = []

def raise_wakeup_signal():
delivered.clear()
# Reinstall for each raise: Windows CRT handlers can reset.
signal.signal(
signal.SIGINT, lambda signum, frame: delivered.append(signum)
)
signal.raise_signal(signal.SIGINT)
assert delivered == [signal.SIGINT]

try:
sys.unraisablehook = errors.append
signal.set_wakeup_fd(write.fileno())
raise_wakeup_signal()
assert read.recv(1) == bytes([signal.SIGINT])
assert not errors

for options in (
{},
{"warn_on_full_buffer": True},
{"warn_on_full_buffer": False},
{},
):
fill_socket(write)
signal.set_wakeup_fd(write.fileno(), **options)
raise_wakeup_signal()
if options.get("warn_on_full_buffer", True):
assert len(errors) == 1, errors
error = errors.pop()
assert error.exc_type is BlockingIOError
assert error.exc_value.winerror == errno.WSAEWOULDBLOCK
assert error.exc_value.errno in (
errno.EAGAIN,
errno.EWOULDBLOCK,
)
assert error.object is None
assert error.exc_traceback is error.exc_value.__traceback__
assert error.err_msg == (
"Exception ignored while trying to send "
"to the signal wakeup fd"
)
else:
assert not errors

# False suppresses a full buffer, not other send failures.
signal.set_wakeup_fd(write.fileno(), warn_on_full_buffer=False)
write.close()
raise_wakeup_signal()
assert len(errors) == 1, errors
assert errors[0].exc_type is OSError
assert errors[0].exc_value.winerror == errno.WSAENOTSOCK
assert errors[0].err_msg == (
"Exception ignored while trying to send to the signal wakeup fd"
)
finally:
signal.set_wakeup_fd(old_fd)
signal.signal(signal.SIGINT, old_handler)
sys.unraisablehook = old_hook


check_windows_wakeup_reporting()
Loading