Skip to content
Draft
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
41 changes: 26 additions & 15 deletions crates/stdlib/src/_queue.rs
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,9 @@ mod _queue {
}

const INITIAL_RING_BUF_CAPACITY: usize = 8;
#[cfg(feature = "threading")]
// Keep signal handlers responsive without excessively waking blocked threads.
const SIGNAL_CHECK_INTERVAL: Duration = Duration::from_millis(100);

#[pyattr]
#[pyclass(module = "_queue", name = "Empty", base = PyException)]
Expand Down Expand Up @@ -74,29 +77,37 @@ mod _queue {

/// Returns `true` if the semaphore was acquired, `false` on timeout.
#[must_use]
fn acquire(&self, block: bool, deadline: Option<Instant>, vm: &VirtualMachine) -> bool {
fn acquire(
&self,
block: bool,
deadline: Option<Instant>,
vm: &VirtualMachine,
) -> PyResult<bool> {
let mut count = self.mutex.lock();
loop {
if *count > 0 {
*count -= 1;
return true;
return Ok(true);
}

if !block {
return false;
return Ok(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;
}
}
None => {
vm.allow_threads(|| self.cond.wait(&mut count));
}
let now = Instant::now();
if deadline.is_some_and(|deadline| now >= deadline) {
return Ok(false);
}

let wait_deadline = deadline.map_or_else(
|| now + SIGNAL_CHECK_INTERVAL,
|deadline| deadline.min(now + SIGNAL_CHECK_INTERVAL),
);
vm.allow_threads(|| self.cond.wait_until(&mut count, wait_deadline));
// Signal handlers can access this queue, so they must run without its mutex.
drop(count);
vm.check_signals()?;
count = self.mutex.lock();
}
}
}
Expand Down Expand Up @@ -227,7 +238,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 +250,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
27 changes: 27 additions & 0 deletions extra_tests/snippets/stdlib_signal.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
import queue
import signal
import sys
import threading
import time

from testutils import assert_raises
Expand Down Expand Up @@ -41,3 +43,28 @@ def handler(signum, frame):
time.sleep(2.0)

assert signals == [signal.SIGALRM, signal.SIGALRM]

q = queue.SimpleQueue()
handled_at = []

def queue_handler(signum, frame):
handled_at.append(time.monotonic() - start)

signal.signal(signal.SIGALRM, queue_handler)
start = time.monotonic()
signal.setitimer(signal.ITIMER_REAL, 0.1)

def unblock():
time.sleep(1)
q.put("unblock")

thread = threading.Thread(target=unblock)
thread.start()
try:
assert q.get() == "unblock"
finally:
signal.setitimer(signal.ITIMER_REAL, 0)
thread.join()

# This leaves a 0.3-second margin before the thread unblocks at one second.
assert handled_at and handled_at[0] < 0.7
Loading