_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

@@ -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<Instant>, 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<Instant>,
+            vm: &VirtualMachine,
+        ) -> PyResult<bool> {
             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<PyObjectRef> {
             #[cfg(feature = "threading")]
             {
-                if !self.sem.acquire(false, None, vm) {
+                if !self.sem.acquire(false, None, vm)? {
                     return Err(empty_error(vm));
                 }
             }

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