Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
66 changes: 45 additions & 21 deletions crates/stdlib/src/_queue.rs
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,11 @@ 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.
#[cfg(feature = "threading")]
const SIGNAL_CHECK_INTERVAL: Duration = Duration::from_millis(50);

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

does this line has CPython reference?

@sigmaith sigmaith Jul 12, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No CPython citation for 50ms, It was empirical choice.
CPython doesn't chunk either; it makes one sem_timedwait call and relies on EINTR.
parking_lot's condvar doesn't expose signal interruptions to us the way CPython does, So I think polling is the workaround. Comment updated to reflect this.


#[pyattr]
#[pyclass(module = "_queue", name = "Empty", base = PyException)]
#[repr(transparent)]
Expand Down Expand Up @@ -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()?;
}
}
}
Expand Down Expand Up @@ -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));
}
}
Expand All @@ -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));
}
}
Expand Down
52 changes: 52 additions & 0 deletions extra_tests/snippets/stdlib_queue_signal.py
Original file line number Diff line number Diff line change
@@ -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")