_queue: check for pending signals during blocking get()

sigmaith · sigmaith · commit 3409f2f04568 · 2026-07-11T15:32:36.000+09:00
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
diff --git a/crates/stdlib/src/_queue.rs b/crates/stdlib/src/_queue.rs
@@ -31,6 +31,11 @@ mod _queue {
 
     const INITIAL_RING_BUF_CAPACITY: usize = 8;
 
+    /// Wait chunk size so blocking waits stay responsive to signals, mirroring
+    /// CPython's `PyThread_acquire_lock_timed`.
+    #[cfg(feature = "threading")]
+    const SIGNAL_CHECK_INTERVAL: Duration = Duration::from_millis(50);
+
     #[pyattr]
     #[pyclass(module = "_queue", name = "Empty", base = PyException)]
     #[repr(transparent)]
@@ -72,31 +77,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 *count > 0 {
+                        *count -= 1;
+                        return Ok(true);
+                    }
 
-                if !block {
-                    return false;
-                }
+                    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),
+                    );
 
-                match deadline {
-                    Some(dl) => {
-                        let result = vm.allow_threads(|| self.cond.wait_until(&mut count, dl));
-                        if result.timed_out() && *count == 0 {
-                            return false;
-                        }
+                    vm.allow_threads(|| self.cond.wait_until(&mut count, chunk_deadline));
+
+                    if *count > 0 {
+                        *count -= 1;
+                        return Ok(true);
                     }
-                    None => {
-                        vm.allow_threads(|| self.cond.wait(&mut count));
+
+                    if let Some(dl) = deadline
+                        && Instant::now() >= dl
+                    {
+                        return Ok(false);
                     }
                 }
+
+                vm.check_signals()?;
             }
         }
     }
@@ -227,7 +251,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 +263,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));
                 }
             }
diff --git a/extra_tests/snippets/stdlib_queue_signal.py 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")
