Skip to content

Commit 44712eb

Browse files
youknowdotyouknowdot
committed
Report Windows signal wakeup socket errors
Capture Winsock failures with bounded atomic storage and report them through sys.unraisablehook on the interpreter thread. Honor warn_on_full_buffer for WSAEWOULDBLOCK. Include bounded Windows socket regressions without changing Unix or CRT wakeup writes. Assisted-by: OpenAI Codex:model identifier unavailable
1 parent f39da82 commit 44712eb

4 files changed

Lines changed: 208 additions & 10 deletions

File tree

‎crates/host_env/src/signal.rs‎

Lines changed: 28 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -395,13 +395,28 @@ pub fn wakeup_fd_is_socket(fd: libc::SOCKET) -> io::Result<bool> {
395395
Ok(false)
396396
}
397397

398+
#[cfg(windows)]
399+
pub struct WakeupSendError(pub i32);
400+
401+
#[cfg(windows)]
402+
impl WakeupSendError {
403+
pub fn is_would_block(&self) -> bool {
404+
self.0 == windows_sys::Win32::Networking::WinSock::WSAEWOULDBLOCK
405+
}
406+
407+
// Only convert to an allocating error outside the OS signal handler.
408+
pub fn into_io_error(self) -> io::Error {
409+
io::Error::from_raw_os_error(self.0)
410+
}
411+
}
412+
398413
#[cfg(windows)]
399414
pub fn notify_signal(
400415
signum: i32,
401416
wakeup_fd: libc::SOCKET,
402417
wakeup_is_socket: bool,
403418
sigint_event: Option<isize>,
404-
) {
419+
) -> Option<WakeupSendError> {
405420
if signum == libc::SIGINT
406421
&& let Some(handle) = sigint_event
407422
{
@@ -411,23 +426,31 @@ pub fn notify_signal(
411426
}
412427

413428
if wakeup_fd == INVALID_SOCKET {
414-
return;
429+
return None;
415430
}
416431

417432
let sigbyte = signum as u8;
418433
if wakeup_is_socket {
419-
unsafe {
420-
let _ = windows_sys::Win32::Networking::WinSock::send(
434+
let result = unsafe {
435+
windows_sys::Win32::Networking::WinSock::send(
421436
wakeup_fd,
422437
&sigbyte as *const u8 as *const _,
423438
1,
424439
0,
425-
);
440+
)
441+
};
442+
if result < 0 {
443+
Some(WakeupSendError(unsafe {
444+
windows_sys::Win32::Networking::WinSock::WSAGetLastError()
445+
}))
446+
} else {
447+
None
426448
}
427449
} else {
428450
unsafe {
429451
let _ = libc::write(wakeup_fd as _, &sigbyte as *const u8 as *const _, 1);
430452
}
453+
None
431454
}
432455
}
433456

‎crates/vm/src/signal.rs‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -103,6 +103,16 @@ fn trigger_signals(vm: &VirtualMachine) -> PyResult<()> {
103103
}
104104
let _guard = SignalHandlerGuard;
105105

106+
#[cfg(all(feature = "host_env", windows))]
107+
if vm.is_main_thread() && vm.state.is_main_interpreter() {
108+
// Report before a Python signal handler can raise out of the current
109+
// error-reporting scope.
110+
crate::stdlib::_signal::_signal::report_wakeup_send_errors(vm);
111+
} else if crate::stdlib::_signal::_signal::wakeup_send_errors_pending() {
112+
// A different interpreter/thread must leave the report for the main one.
113+
set_triggered();
114+
}
115+
106116
let signal_handlers = vm
107117
.signal_handlers
108118
.get()

‎crates/vm/src/stdlib/_signal.rs‎

Lines changed: 58 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -73,6 +73,13 @@ pub(crate) mod _signal {
7373
}
7474
}
7575

76+
#[cfg(windows)]
77+
static WARN_ON_FULL_BUFFER: atomic::AtomicBool = atomic::AtomicBool::new(true);
78+
79+
// Bounded pending-call storage: signal handlers cannot allocate or lock.
80+
#[cfg(windows)]
81+
static WAKEUP_SEND_ERRORS: [atomic::AtomicI32; 32] = [const { atomic::AtomicI32::new(0) }; 32];
82+
7683
#[cfg(any(unix, windows, target_os = "wasi"))]
7784
#[allow(unused_imports)]
7885
pub use host_signal::SIG_ERR;
@@ -344,7 +351,7 @@ pub(crate) mod _signal {
344351

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

385392
#[cfg(windows)]
386-
WAKEUP_IS_SOCKET.store(is_socket, Ordering::Relaxed);
393+
{
394+
WARN_ON_FULL_BUFFER.store(args.warn_on_full_buffer, Ordering::Relaxed);
395+
WAKEUP_IS_SOCKET.store(is_socket, Ordering::Relaxed);
396+
}
387397

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

535-
host_signal::notify_signal(
545+
#[cfg(not(windows))]
546+
host_signal::notify_signal(signum, WAKEUP.load(Ordering::Relaxed));
547+
548+
#[cfg(windows)]
549+
let error = host_signal::notify_signal(
536550
signum,
537551
WAKEUP.load(Ordering::Relaxed),
538-
#[cfg(windows)]
539552
WAKEUP_IS_SOCKET.load(Ordering::Relaxed),
540-
#[cfg(windows)]
541553
signal::get_sigint_event(),
542554
);
555+
#[cfg(windows)]
556+
if let Some(error) = error
557+
&& (WARN_ON_FULL_BUFFER.load(Ordering::Relaxed) || !error.is_would_block())
558+
{
559+
for pending in &WAKEUP_SEND_ERRORS {
560+
if pending
561+
.compare_exchange(0, error.0, Ordering::Relaxed, Ordering::Relaxed)
562+
.is_ok()
563+
{
564+
// The VM may have handled the signal on another thread
565+
// before the failed wakeup send completed.
566+
signal::set_triggered();
567+
break;
568+
}
569+
}
570+
}
571+
}
572+
573+
#[cfg(windows)]
574+
pub(crate) fn wakeup_send_errors_pending() -> bool {
575+
WAKEUP_SEND_ERRORS
576+
.iter()
577+
.any(|error| error.load(Ordering::Relaxed) != 0)
578+
}
579+
580+
#[cfg(windows)]
581+
pub(crate) fn report_wakeup_send_errors(vm: &VirtualMachine) {
582+
for pending in &WAKEUP_SEND_ERRORS {
583+
let error = pending.swap(0, Ordering::Relaxed);
584+
if error != 0 {
585+
vm.run_unraisable(
586+
host_signal::WakeupSendError(error)
587+
.into_io_error()
588+
.into_pyexception(vm),
589+
Some(
590+
"Exception ignored while trying to send to the signal wakeup fd".to_owned(),
591+
),
592+
vm.ctx.none(),
593+
);
594+
}
595+
}
543596
}
544597

545598
/// Reset wakeup fd after fork in child process.

‎extra_tests/snippets/stdlib_signal.py‎

Lines changed: 112 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -106,3 +106,115 @@ def arm_other(signum, frame):
106106
signal.signal(signal.SIGALRM, signal.SIG_DFL)
107107
signal.signal(signal.SIGUSR1, signal.SIG_DFL)
108108
signal.signal(signal.SIGUSR2, signal.SIG_DFL)
109+
110+
111+
def check_windows_wakeup_reporting():
112+
if sys.platform != "win32":
113+
return
114+
115+
import errno
116+
import socket
117+
118+
def fill_socket(write):
119+
# A nonblocking Windows send can briefly fail before TCP has filled
120+
# the receiving socket. Require repeated one-byte failures to settle.
121+
deadline = time.monotonic() + 5.0
122+
written = 0
123+
blocked = 0
124+
for _ in range(4096):
125+
assert time.monotonic() < deadline, "socket fill timed out"
126+
assert written < 16 * 1024 * 1024, "socket fill exceeded byte bound"
127+
try:
128+
written += write.send(b"x" * 65536)
129+
except BlockingIOError:
130+
try:
131+
written += write.send(b"x")
132+
except BlockingIOError:
133+
blocked += 1
134+
if blocked == 4:
135+
return
136+
time.sleep(0.050)
137+
continue
138+
blocked = 0
139+
raise AssertionError("socket did not fill within the attempt bound")
140+
141+
# Use numeric loopback addresses and bounded connection/receive calls.
142+
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as listener:
143+
listener.settimeout(5.0)
144+
listener.bind(("127.0.0.1", 0))
145+
listener.listen(1)
146+
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as write:
147+
write.settimeout(5.0)
148+
write.connect(listener.getsockname())
149+
read, _ = listener.accept()
150+
with read:
151+
read.settimeout(5.0)
152+
read.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 4096)
153+
write.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 4096)
154+
write.setblocking(False)
155+
old_handler = signal.getsignal(signal.SIGINT)
156+
old_hook = sys.unraisablehook
157+
old_fd = signal.set_wakeup_fd(-1)
158+
errors = []
159+
delivered = []
160+
161+
def raise_wakeup_signal():
162+
delivered.clear()
163+
# Reinstall for each raise: Windows CRT handlers can reset.
164+
signal.signal(
165+
signal.SIGINT, lambda signum, frame: delivered.append(signum)
166+
)
167+
signal.raise_signal(signal.SIGINT)
168+
assert delivered == [signal.SIGINT]
169+
170+
try:
171+
sys.unraisablehook = errors.append
172+
signal.set_wakeup_fd(write.fileno())
173+
raise_wakeup_signal()
174+
assert read.recv(1) == bytes([signal.SIGINT])
175+
assert not errors
176+
177+
for options in (
178+
{},
179+
{"warn_on_full_buffer": True},
180+
{"warn_on_full_buffer": False},
181+
{},
182+
):
183+
fill_socket(write)
184+
signal.set_wakeup_fd(write.fileno(), **options)
185+
raise_wakeup_signal()
186+
if options.get("warn_on_full_buffer", True):
187+
assert len(errors) == 1, errors
188+
error = errors.pop()
189+
assert error.exc_type is BlockingIOError
190+
assert error.exc_value.winerror == errno.WSAEWOULDBLOCK
191+
assert error.exc_value.errno in (
192+
errno.EAGAIN,
193+
errno.EWOULDBLOCK,
194+
)
195+
assert error.object is None
196+
assert error.exc_traceback is error.exc_value.__traceback__
197+
assert error.err_msg == (
198+
"Exception ignored while trying to send "
199+
"to the signal wakeup fd"
200+
)
201+
else:
202+
assert not errors
203+
204+
# False suppresses a full buffer, not other send failures.
205+
signal.set_wakeup_fd(write.fileno(), warn_on_full_buffer=False)
206+
write.close()
207+
raise_wakeup_signal()
208+
assert len(errors) == 1, errors
209+
assert errors[0].exc_type is OSError
210+
assert errors[0].exc_value.winerror == errno.WSAENOTSOCK
211+
assert errors[0].err_msg == (
212+
"Exception ignored while trying to send to the signal wakeup fd"
213+
)
214+
finally:
215+
signal.set_wakeup_fd(old_fd)
216+
signal.signal(signal.SIGINT, old_handler)
217+
sys.unraisablehook = old_hook
218+
219+
220+
check_windows_wakeup_reporting()

0 commit comments

Comments
 (0)