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")