From 70c3f6c06d24daae19ee4ffbeb77fec53218007c Mon Sep 17 00:00:00 2001 From: youknowdot Date: Tue, 20 Oct 2026 04:08:49 -0700 Subject: [PATCH] Align signal wakeup errors and delivery with Python 3.15 (#9199) * Align signal wakeup error handling with Python 3.15 Assisted-by: OpenAI Codex:model version unavailable * Preserve CRT error state during signal wakeup writes Assisted-by: OpenAI Codex:model version unavailable * Represent nullable Windows CRT parameter handlers safely Assisted-by: OpenAI Codex:model version unavailable * Rework signal wakeup error delivery to follow signalmodule.c - Report pending wakeup write/send errors after Python signal handlers and VM pending calls, and re-arm the eval breaker when a handler or pending call raises. - Save and restore errno once in run_signal around trip_signal instead of inside each write path; stop preserving _doserrno. - Add crt_fd::write_noraise and use it for the Unix and Windows wakeup writes, sharing the single-write helper with _write. - Keep pending wakeup errors across fork. - Revert the suppress_iph handler type change and the unrelated quote change in test_signal. Assisted-by: Claude Code:claude-opus-5-5 * Represent nullable Windows CRT parameter handlers as Option _set_thread_local_invalid_parameter_handler takes and returns a nullable function pointer; declare it as Option so a NULL previous handler is restored as None. Assisted-by: Claude Code:claude-opus-5-5 --------- Co-authored-by: Jeong YunWon --- Lib/test/test_signal.py | 3 +- crates/host_env/src/crt_fd.rs | 35 +++++++-- crates/host_env/src/macros.rs | 8 +- crates/host_env/src/signal.rs | 50 +++++++++---- crates/vm/src/signal.rs | 82 ++++++++++++-------- crates/vm/src/stdlib/_signal.rs | 128 +++++++++++++++++++++----------- 6 files changed, 207 insertions(+), 99 deletions(-) diff --git a/Lib/test/test_signal.py b/Lib/test/test_signal.py index a4af92d52..8971e9972 100644 --- a/Lib/test/test_signal.py +++ b/Lib/test/test_signal.py @@ -14,7 +14,7 @@ import time import unittest from test import support from test.support import ( - is_apple, is_apple_mobile, os_helper, threading_helper + force_not_colorized, is_apple, is_apple_mobile, os_helper, threading_helper ) from test.support.script_helper import assert_python_ok, spawn_python try: @@ -355,6 +355,7 @@ class WakeupSignalTests(unittest.TestCase): @unittest.skipIf(_testcapi is None, 'need _testcapi') @unittest.skipUnless(hasattr(os, "pipe"), "requires os.pipe()") + @force_not_colorized def test_wakeup_write_error(self): # Issue #16105: write() errors in the C signal handler should not # pass silently. diff --git a/crates/host_env/src/crt_fd.rs b/crates/host_env/src/crt_fd.rs index 5b2cf685b..5b3ba4690 100644 --- a/crates/host_env/src/crt_fd.rs +++ b/crates/host_env/src/crt_fd.rs @@ -437,9 +437,30 @@ pub fn as_handle(fd: Borrowed<'_>) -> io::Result> { fn _write(fd: Raw, buf: &[u8]) -> io::Result { let count = cmp::min(buf.len(), MAX_RW); + write_errno(fd, &buf[..count]).map_err(|errno| { + cfg_select! { + windows => crate::os::io_error_from_errno(errno), + _ => io::Error::from_raw_os_error(errno), + } + }) +} + +/// `_Py_write_noraise`: retry on EINTR and return the errno on failure +/// without allocating, so the OS signal handler can use it. +pub fn write_noraise(fd: Raw, buf: &[u8]) -> Result { + let count = cmp::min(buf.len(), MAX_RW); + loop { + match write_errno(fd, &buf[..count]) { + Err(libc::EINTR) => continue, + result => return result, + } + } +} + +fn write_errno(fd: Raw, buf: &[u8]) -> Result { #[cfg(windows)] { - let mut count = count; + let mut count = buf.len(); loop { unsafe { c::_set_doserrno(0) }; let n = unsafe { suppress_iph!(c::write(fd, buf.as_ptr() as _, count as _)) }; @@ -450,18 +471,22 @@ fn _write(fd: Raw, buf: &[u8]) -> io::Result { // The CRT reports a successful zero-byte WriteFile as ENOSPC. // A nonblocking pipe may still accept a smaller prefix. if errno != libc::ENOSPC || dos_error() != 0 { - return Err(crate::os::io_error_from_errno(errno)); + return Err(errno); } count /= 2; if count == 0 { - return Err(crate::os::io_error_from_errno(libc::EAGAIN)); + return Err(libc::EAGAIN); } } } #[cfg(not(windows))] { - let n = cvt(unsafe { suppress_iph!(c::write(fd, buf.as_ptr() as _, count as _)) })?; - Ok(n as usize) + let n = unsafe { suppress_iph!(c::write(fd, buf.as_ptr() as _, buf.len() as _)) }; + if n < 0 { + Err(crate::os::get_errno()) + } else { + Ok(n as usize) + } } } diff --git a/crates/host_env/src/macros.rs b/crates/host_env/src/macros.rs index 08d00e592..a975997ce 100644 --- a/crates/host_env/src/macros.rs +++ b/crates/host_env/src/macros.rs @@ -12,9 +12,9 @@ macro_rules! suppress_iph { #[cfg(all(windows, target_env = "msvc"))] macro_rules! __suppress_iph_impl { ($e:expr) => {{ - let old = $crate::__macro_private::_set_thread_local_invalid_parameter_handler( + let old = $crate::__macro_private::_set_thread_local_invalid_parameter_handler(Some( $crate::__macro_private::silent_iph_handler, - ); + )); let ret = $e; $crate::__macro_private::_set_thread_local_invalid_parameter_handler(old); ret @@ -43,8 +43,8 @@ pub mod __macro_private { #[cfg(target_env = "msvc")] unsafe extern "C" { pub fn _set_thread_local_invalid_parameter_handler( - pNew: InvalidParamHandler, - ) -> InvalidParamHandler; + pNew: Option, + ) -> Option; } #[cfg(target_env = "msvc")] diff --git a/crates/host_env/src/signal.rs b/crates/host_env/src/signal.rs index f5205d841..ffe47dc41 100644 --- a/crates/host_env/src/signal.rs +++ b/crates/host_env/src/signal.rs @@ -395,18 +395,33 @@ pub fn wakeup_fd_is_socket(fd: libc::SOCKET) -> io::Result { Ok(false) } -#[cfg(windows)] -pub struct WakeupSendError(pub i32); +#[cfg(any(unix, windows))] +pub enum WakeupError { + Write(i32), + #[cfg(windows)] + Send(i32), +} -#[cfg(windows)] -impl WakeupSendError { +#[cfg(any(unix, windows))] +impl WakeupError { pub fn is_would_block(&self) -> bool { - self.0 == windows_sys::Win32::Networking::WinSock::WSAEWOULDBLOCK + match *self { + Self::Write(errno) => errno == libc::EAGAIN || errno == libc::EWOULDBLOCK, + #[cfg(windows)] + Self::Send(error) => error == 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) + match self { + #[cfg(unix)] + Self::Write(errno) => io::Error::from_raw_os_error(errno), + #[cfg(windows)] + Self::Write(errno) => crate::os::io_error_from_errno(errno), + #[cfg(windows)] + Self::Send(error) => io::Error::from_raw_os_error(error), + } } } @@ -416,7 +431,7 @@ pub fn notify_signal( wakeup_fd: libc::SOCKET, wakeup_is_socket: bool, sigint_event: Option, -) -> Option { +) -> Option { if signum == libc::SIGINT && let Some(handle) = sigint_event { @@ -440,21 +455,30 @@ pub fn notify_signal( ) }; if result < 0 { - Some(WakeupSendError(unsafe { + Some(WakeupError::Send(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 + crate::crt_fd::write_noraise(wakeup_fd as _, &[sigbyte]) + .err() + .map(WakeupError::Write) } } -#[cfg(any(unix, target_os = "wasi"))] +#[cfg(unix)] +pub fn notify_signal(signum: i32, wakeup_fd: i32) -> Option { + if wakeup_fd == -1 { + return None; + } + crate::crt_fd::write_noraise(wakeup_fd, &[signum as u8]) + .err() + .map(WakeupError::Write) +} + +#[cfg(target_os = "wasi")] pub fn notify_signal(signum: i32, wakeup_fd: i32) { if wakeup_fd == -1 { return; diff --git a/crates/vm/src/signal.rs b/crates/vm/src/signal.rs index 4ea023505..1cf741534 100644 --- a/crates/vm/src/signal.rs +++ b/crates/vm/src/signal.rs @@ -134,48 +134,68 @@ 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 can_handle_signals = vm.is_main_thread() && vm.state.is_main_interpreter(); + if can_handle_signals { + let signal_handlers = vm + .signal_handlers + .get() + .expect("should never fail since we check above"); - let signal_handlers = vm - .signal_handlers - .get() - .expect("should never fail since we check above"); + for (signum, trigger) in TRIGGERS.iter().enumerate().skip(1) { + let triggered = trigger.swap(false, Ordering::Relaxed); + if !triggered { + continue; + } - for (signum, trigger) in TRIGGERS.iter().enumerate().skip(1) { - let triggered = trigger.swap(false, Ordering::Relaxed); - if !triggered { - continue; + // SAFETY: TRIGGERS has the same length as the signal_handlers + let signum = unsafe { SignalNum::new_unchecked(signum as i32) }; + + // Read the handler out and drop the borrow before running it. A + // handler is free to call signal.signal(), which takes the same cell + // mutably, and a live read borrow turns that into a panic. + let handler = signal_handlers.borrow()[signum].clone(); + + if let Some(handler) = handler + && let Some(callable) = handler.to_callable() + && let Err(exc) = callable.invoke((signum.as_i32(), vm.ctx.none()), vm) + { + // On error, re-schedule a call to trigger_signals() for the + // remaining signals and the pending wakeup errors. + set_triggered(); + return Err(exc); + } } - - // SAFETY: TRIGGERS has the same length as the signal_handlers - let signum = unsafe { SignalNum::new_unchecked(signum as i32) }; - - // Read the handler out and drop the borrow before running it. A - // handler is free to call signal.signal(), which takes the same cell - // mutably, and a live read borrow turns that into a panic. - let handler = signal_handlers.borrow()[signum].clone(); - - if let Some(handler) = handler - && let Some(callable) = handler.to_callable() - { - callable.invoke((signum.as_i32(), vm.ctx.none()), vm)?; + } else { + let os_signals_pending = TRIGGERS + .iter() + .skip(1) + .any(|trigger| trigger.load(Ordering::Relaxed)); + #[cfg(all(feature = "host_env", any(unix, windows)))] + let os_signals_pending = + os_signals_pending || crate::stdlib::_signal::_signal::wakeup_errors_pending(); + if os_signals_pending { + // Leave OS signals for the main interpreter's main thread. Re-arm + // before running VM-local callbacks, which may raise. + set_triggered(); } } if let Some(signal_rx) = &vm.signal_rx { for f in signal_rx.rx.try_iter() { - f(vm)?; + if let Err(exc) = f(vm) { + set_triggered(); + return Err(exc); + } } } + // Wakeup fd errors are pending calls that only the main interpreter's + // main thread runs. + #[cfg(all(feature = "host_env", any(unix, windows)))] + if can_handle_signals { + crate::stdlib::_signal::_signal::report_wakeup_errors(vm); + } + Ok(()) } diff --git a/crates/vm/src/stdlib/_signal.rs b/crates/vm/src/stdlib/_signal.rs index 79867ff52..7deec7d5d 100644 --- a/crates/vm/src/stdlib/_signal.rs +++ b/crates/vm/src/stdlib/_signal.rs @@ -79,10 +79,11 @@ 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(any(unix, windows))] + static WAKEUP_WRITE_ERRORS: [atomic::AtomicI32; 32] = [const { atomic::AtomicI32::new(0) }; 32]; #[cfg(windows)] static WAKEUP_SEND_ERRORS: [atomic::AtomicI32; 32] = [const { atomic::AtomicI32::new(0) }; 32]; @@ -357,8 +358,6 @@ pub(crate) mod _signal { #[pyfunction] fn set_wakeup_fd(args: SetWakeupFdArgs, vm: &VirtualMachine) -> PyResult { - // 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, _ => args.fd, @@ -394,12 +393,10 @@ pub(crate) mod _signal { } let old_fd = WAKEUP.swap(fd, Ordering::Relaxed); + WARN_ON_FULL_BUFFER.store(args.warn_on_full_buffer, Ordering::Relaxed); #[cfg(windows)] - { - WARN_ON_FULL_BUFFER.store(args.warn_on_full_buffer, Ordering::Relaxed); - WAKEUP_IS_SOCKET.store(is_socket, Ordering::Relaxed); - } + WAKEUP_IS_SOCKET.store(is_socket, Ordering::Relaxed); #[cfg(windows)] if old_fd == INVALID_WAKEUP { @@ -544,60 +541,101 @@ pub(crate) mod _signal { #[cfg(any(unix, windows, target_os = "wasi"))] pub extern "C" fn run_signal(signum: i32) { + // Asynchronously executing signal handlers should not mutate errno + // under the feet of the interrupted code. + let save_errno = rustpython_host_env::os::get_errno(); + trip_signal(signum); + rustpython_host_env::os::set_errno(save_errno); + } + + #[cfg(any(unix, windows, target_os = "wasi"))] + fn trip_signal(signum: i32) { signal::TRIGGERS[signum as usize].store(true, Ordering::Relaxed); signal::set_triggered(); - #[cfg(not(windows))] + #[cfg(target_os = "wasi")] host_signal::notify_signal(signum, WAKEUP.load(Ordering::Relaxed)); - #[cfg(windows)] - let error = host_signal::notify_signal( - signum, - WAKEUP.load(Ordering::Relaxed), - WAKEUP_IS_SOCKET.load(Ordering::Relaxed), - signal::get_sigint_event(), - ); - #[cfg(windows)] - if let Some(error) = error - && (WARN_ON_FULL_BUFFER.load(Ordering::Relaxed) || !error.is_would_block()) + #[cfg(any(unix, windows))] { - 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; + 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(), + ); + if let Some(error) = error + && (WARN_ON_FULL_BUFFER.load(Ordering::Relaxed) || !error.is_would_block()) + { + let (errors, errno) = match error { + host_signal::WakeupError::Write(errno) => (&WAKEUP_WRITE_ERRORS, errno), + #[cfg(windows)] + host_signal::WakeupError::Send(errno) => (&WAKEUP_SEND_ERRORS, errno), + }; + for pending in errors { + if pending + .compare_exchange(0, errno, Ordering::Relaxed, Ordering::Relaxed) + .is_ok() + { + // The VM may have handled the signal on another thread + // before the failed wakeup write completed. + signal::set_triggered(); + break; + } } } } } - #[cfg(windows)] - pub(crate) fn wakeup_send_errors_pending() -> bool { - WAKEUP_SEND_ERRORS + #[cfg(any(unix, windows))] + pub(crate) fn wakeup_errors_pending() -> bool { + let pending = WAKEUP_WRITE_ERRORS .iter() - .any(|error| error.load(Ordering::Relaxed) != 0) + .any(|error| error.load(Ordering::Relaxed) != 0); + #[cfg(windows)] + let pending = pending + || WAKEUP_SEND_ERRORS + .iter() + .any(|error| error.load(Ordering::Relaxed) != 0); + pending } - #[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(), - ); + #[cfg(any(unix, windows))] + pub(crate) fn report_wakeup_errors(vm: &VirtualMachine) { + fn report( + errors: &[atomic::AtomicI32], + make_error: fn(i32) -> host_signal::WakeupError, + operation: &str, + vm: &VirtualMachine, + ) { + for pending in errors { + let errno = pending.swap(0, Ordering::Relaxed); + if errno != 0 { + vm.run_unraisable( + make_error(errno).into_io_error().into_pyexception(vm), + Some(format!( + "Exception ignored while trying to {operation} to the signal wakeup fd" + )), + vm.ctx.none(), + ); + } } } + report( + &WAKEUP_WRITE_ERRORS, + host_signal::WakeupError::Write, + "write", + vm, + ); + #[cfg(windows)] + report( + &WAKEUP_SEND_ERRORS, + host_signal::WakeupError::Send, + "send", + vm, + ); } /// Reset wakeup fd after fork in child process.