From 9d9fc27082bf3f869bdab1a7c2f92f6e17a25fed Mon Sep 17 00:00:00 2001 From: Sumi Jeong <125195487+sigmaith@users.noreply.github.com> Date: Thu, 16 Jul 2026 17:39:48 +0900 Subject: [PATCH] `_queue`: check for pending signals during blocking `get()` (#8252) * _queue: check for pending signals during blocking get() Semaphore::acquire() waited on the condvar in a single uninterrupted call, so a Python-level signal handler never ran until the wait itself returned. check_signals() is only invoked from the bytecode dispatch loop, and this native wait never passed through it -- CPython chunks its equivalent blocking wait for the same reason. Cap each wait at a short interval and call check_signals() between chunks, only giving up once the caller's real deadline has actually elapsed. The mutex guard must be dropped before check_signals() runs: a signal handler that calls back into the same queue on the same thread would otherwise try to relock a mutex it's still holding and deadlock against itself. Add a regression snippet that arms a SIGALRM well before a helper thread unblocks the queue, and asserts the handler ran near the alarm instead of only once get() returned. Refs #8250 Assisted-by: Claude * _queue: correct SIGNAL_CHECK_INTERVAL doc comment The old comment claimed this mirrors CPython's PyThread_acquire_lock_timed, but CPython doesn't chunk its wait either -- it makes one sem_timedwait call and relies on the OS reporting EINTR. parking_lot's Condvar has no equivalent signal to surface, so polling is a workaround for that gap, not a mirror of CPython's approach. Assisted-by: Claude * Apply suggestions from code review Co-authored-by: Jeong, YunWon <69878+youknowone@users.noreply.github.com> * _queue: fix comment indentation (cargo fmt) Assisted-by: Claude --------- Co-authored-by: Jeong, YunWon <69878+youknowone@users.noreply.github.com> --- crates/stdlib/src/_queue.rs | 74 +++++++++++++++------ extra_tests/snippets/stdlib_queue_signal.py | 52 +++++++++++++++ 2 files changed, 104 insertions(+), 22 deletions(-) create mode 100644 extra_tests/snippets/stdlib_queue_signal.py diff --git a/crates/stdlib/src/_queue.rs b/crates/stdlib/src/_queue.rs index 65456bd35..6b150e4c6 100644 --- a/crates/stdlib/src/_queue.rs +++ b/crates/stdlib/src/_queue.rs @@ -31,6 +31,17 @@ mod _queue { const INITIAL_RING_BUF_CAPACITY: usize = 8; + /// `parking_lot`'s `Condvar` doesn't expose a mid-wait signal to us (unlike + /// CPython's raw `sem_timedwait`, which reports `EINTR`), so we poll instead. + // FIXME: interim stopgap. The signal already interrupts the wait with EINTR + // (SA_RESTART cleared via `siginterrupt`), but `parking_lot::Condvar` swallows + // it and re-parks, forcing this poll. Replace with a shared interruptible + // timed-wait that surfaces EINTR (`poll`/wakeup-fd, portable incl. macOS; or + // `sem_timedwait` where available) -- `_thread` lock and `Thread.join` share + // this defect. Then this constant and the chunking loop go away. + #[cfg(feature = "threading")] + const SIGNAL_CHECK_INTERVAL: Duration = Duration::from_millis(50); + #[pyattr] #[pyclass(module = "_queue", name = "Empty", base = PyException)] #[repr(transparent)] @@ -72,31 +83,50 @@ mod _queue { self.cond.notify_one(); } - /// Returns `true` if the semaphore was acquired, `false` on timeout. - #[must_use] - fn acquire(&self, block: bool, deadline: Option, vm: &VirtualMachine) -> bool { - let mut count = self.mutex.lock(); + /// `Ok(true)` if acquired, `Ok(false)` on timeout, `Err` if a signal + /// handler raised (e.g. `KeyboardInterrupt`) while we were waiting. + fn acquire( + &self, + block: bool, + deadline: Option, + vm: &VirtualMachine, + ) -> PyResult { loop { - if *count > 0 { - *count -= 1; - return true; - } + // Guard must be dropped before check_signals() below, since a + // signal handler may call back into this same queue. + { + let mut count = self.mutex.lock(); - if !block { - return false; - } - - match deadline { - Some(dl) => { - let result = vm.allow_threads(|| self.cond.wait_until(&mut count, dl)); - if result.timed_out() && *count == 0 { - return false; - } + if *count > 0 { + *count -= 1; + return Ok(true); } - None => { - vm.allow_threads(|| self.cond.wait(&mut count)); + + if !block { + return Ok(false); + } + + let now = Instant::now(); + let chunk_deadline = deadline.map_or_else( + || now + SIGNAL_CHECK_INTERVAL, + |dl| dl.min(now + SIGNAL_CHECK_INTERVAL), + ); + + vm.allow_threads(|| self.cond.wait_until(&mut count, chunk_deadline)); + + if *count > 0 { + *count -= 1; + return Ok(true); + } + + if let Some(dl) = deadline + && Instant::now() >= dl + { + return Ok(false); } } + + vm.check_signals()?; } } } @@ -227,7 +257,7 @@ mod _queue { #[cfg(feature = "threading")] { - if !self.sem.acquire(block, deadline, vm) { + if !self.sem.acquire(block, deadline, vm)? { return Err(empty_error(vm)); } } @@ -239,7 +269,7 @@ mod _queue { fn get_nowait(&self, vm: &VirtualMachine) -> PyResult { #[cfg(feature = "threading")] { - if !self.sem.acquire(false, None, vm) { + if !self.sem.acquire(false, None, vm)? { return Err(empty_error(vm)); } } diff --git a/extra_tests/snippets/stdlib_queue_signal.py b/extra_tests/snippets/stdlib_queue_signal.py new file mode 100644 index 000000000..fce6bcbb1 --- /dev/null +++ b/extra_tests/snippets/stdlib_queue_signal.py @@ -0,0 +1,52 @@ +"""A blocking queue.SimpleQueue.get() must stay responsive to signals. + +SimpleQueue.get() waits on a Condvar. A single uninterrupted wait blocks +Python-level signal handlers (including the default KeyboardInterrupt) until +the wait itself returns, since signals are only delivered at bytecode +safepoints. A regression shows up as the handler firing only once get() +unblocks, instead of promptly when the signal actually arrives. +""" + +import queue +import signal +import sys +import threading +import time + +if sys.platform.startswith("win"): + print("skipped (no SIGALRM)") + raise SystemExit(0) + +q = queue.SimpleQueue() +start = time.time() +handled_at = [] + + +def handler(signum, frame): + handled_at.append(time.time() - start) + + +signal.signal(signal.SIGALRM, handler) +signal.setitimer(signal.ITIMER_REAL, 0.3) + + +def unblock_later(): + time.sleep(1.5) + q.put("unblock") + + +threading.Thread(target=unblock_later, daemon=True).start() + +item = q.get() # blocks until unblock_later() wakes us up +elapsed = time.time() - start + +assert item == "unblock", item +assert handled_at, "signal handler never ran" +# The handler should fire around t=0.3s, when the timer was started, not t=1.5s +# (when get() finally unblocked). +assert handled_at[0] < elapsed - 0.5, ( + f"signal handled at {handled_at[0]:.2f}s but get() only returned at " + f"{elapsed:.2f}s -- signal was not processed while blocked" +) + +print("ok")