diff --git a/crates/host_env/src/signal.rs b/crates/host_env/src/signal.rs index 60b06d63e..f5205d841 100644 --- a/crates/host_env/src/signal.rs +++ b/crates/host_env/src/signal.rs @@ -395,13 +395,28 @@ pub fn wakeup_fd_is_socket(fd: libc::SOCKET) -> io::Result { 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, -) { +) -> Option { if signum == libc::SIGINT && let Some(handle) = sigint_event { @@ -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 } } diff --git a/crates/vm/src/signal.rs b/crates/vm/src/signal.rs index eda531bbc..8bef84d33 100644 --- a/crates/vm/src/signal.rs +++ b/crates/vm/src/signal.rs @@ -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() diff --git a/crates/vm/src/stdlib/_signal.rs b/crates/vm/src/stdlib/_signal.rs index 1c2bc7925..348f5bc92 100644 --- a/crates/vm/src/stdlib/_signal.rs +++ b/crates/vm/src/stdlib/_signal.rs @@ -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; @@ -344,7 +351,7 @@ pub(crate) mod _signal { #[pyfunction] fn set_wakeup_fd(args: SetWakeupFdArgs, vm: &VirtualMachine) -> PyResult { - // 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, @@ -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 { @@ -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. diff --git a/extra_tests/snippets/stdlib_signal.py b/extra_tests/snippets/stdlib_signal.py index 39f103157..c03bc70ab 100644 --- a/extra_tests/snippets/stdlib_signal.py +++ b/extra_tests/snippets/stdlib_signal.py @@ -106,3 +106,115 @@ if "win" not in sys.platform and sys.platform != "wasi": 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(): + 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()