diff --git a/.github/workflows/async-ecosystem.yml b/.github/workflows/async-ecosystem.yml index d7908b7..be3b996 100644 --- a/.github/workflows/async-ecosystem.yml +++ b/.github/workflows/async-ecosystem.yml @@ -112,5 +112,12 @@ jobs: run: | uv run --no-sync python -c "import asyncpg, sqlalchemy, uvloop, platform; from importlib.metadata import version; print(platform.platform(), platform.python_version()); print({p: version(p) for p in ('asyncpg', 'sqlalchemy', 'uvloop')})" docker exec ${{ job.services.postgres.id }} postgres --version + - name: Measure cancelled timers and closed statement caches + run: | + for event_loop in asyncio uvloop rsloop; do + uv run --no-sync python benches/timer_cancellation.py --loop "$event_loop" --case timers + uv run --no-sync python benches/timer_cancellation.py --loop "$event_loop" --case timeout + uv run --no-sync python benches/timer_cancellation.py --loop "$event_loop" --case postgres --count 500 + done - name: Compare PostgreSQL lifecycles run: uv run --no-sync python scripts/run_python_tests.py tests/ecosystem/test_postgresql.py -m ecosystem -ra diff --git a/benches/timer_cancellation.py b/benches/timer_cancellation.py new file mode 100644 index 0000000..c067a3a --- /dev/null +++ b/benches/timer_cancellation.py @@ -0,0 +1,245 @@ +"""Fresh-process lifetime/RSS and timer throughput probe for issue #118. + +Run with the same release build and interpreter for before/after throughput. +PostgreSQL requires RSLOOP_POSTGRES_DSN pointing to an isolated database. +""" + +from __future__ import annotations + +import argparse +import asyncio +import gc +import importlib +import json +import os +import platform +import statistics +import sys +import time +import tracemalloc +import weakref +from importlib.metadata import version +from pathlib import Path + + +class Payload: + def __init__(self): + self.blob = bytearray(4096) + + def fire(self): + raise AssertionError("cancelled timer fired") + + +def rss_bytes(): + if sys.platform == "linux": + for line in Path("/proc/self/status").read_text().splitlines(): + if line.startswith("VmRSS:"): + return int(line.split()[1]) * 1024 + return None + + +def live(references): + gc.collect() + return sum(reference() is not None for reference in references) + + +async def memory_case(count, mode): + loop = asyncio.get_running_loop() + sentinel = loop.call_later(1800, lambda: None) + references = [] + gc.collect() + rss_before = rss_bytes() + tracemalloc.start() + before = tracemalloc.get_traced_memory()[0] + started = time.perf_counter() + try: + if mode == "timers": + payload = None + for _ in range(count): + payload = Payload() + references.append(weakref.ref(payload)) + loop.call_later(3600, payload.fire).cancel() + del payload + else: + + async def request(): + async with asyncio.timeout(3600): + await asyncio.sleep(0) + return Payload() + + for offset in range(0, count, 500): + tasks = [ + loop.create_task(request()) for _ in range(min(500, count - offset)) + ] + await asyncio.gather(*tasks) + references.extend(weakref.ref(task) for task in tasks) + del tasks + immediate = live(references) + held_immediate = tracemalloc.get_traced_memory()[0] - before + rss_immediate = rss_bytes() + for _ in range(3): + await asyncio.sleep(0) + after_turn = live(references) + held_after_turn = tracemalloc.get_traced_memory()[0] - before + # Subtract only this probe's weakref/list storage, as in the issue. + probe_bytes = sys.getsizeof(references) + sum( + sys.getsizeof(ref) for ref in references + ) + return { + "count": count, + "alive_immediately": immediate, + "alive_after_turn": after_turn, + "python_bytes_immediately": held_immediate - probe_bytes, + "python_bytes_after_turn": held_after_turn - probe_bytes, + "traced_peak_bytes": tracemalloc.get_traced_memory()[1], + "rss_before": rss_before, + "rss_immediately": rss_immediate, + "rss_after_turn": rss_bytes(), + "seconds": time.perf_counter() - started, + } + finally: + tracemalloc.stop() + sentinel.cancel() + + +def postgres_counts(): + gc.collect() + counts = dict.fromkeys(("_StatementCache", "PreparedStatementState", "Protocol"), 0) + for obj in gc.get_objects(): + name = type(obj).__name__ + if name in counts: + counts[name] += 1 + return counts + + +async def postgres_case(count): + asyncpg = importlib.import_module("asyncpg") + + loop = asyncio.get_running_loop() + sentinel = loop.call_later(1800, lambda: None) + before = postgres_counts() + rss_before = rss_bytes() + started = time.perf_counter() + + async def connection(): + conn = await asyncpg.connect( + os.environ["RSLOOP_POSTGRES_DSN"], max_cached_statement_lifetime=3600 + ) + try: + for index in range(3): + query = conn.fetchval(f"SELECT $1::int + {index}", index) + if hasattr(asyncio, "timeout"): + async with asyncio.timeout(3600): + value = await query + else: + value = await asyncio.wait_for(query, 3600) + assert value == 2 * index + finally: + await conn.close() + + try: + for offset in range(0, count, 10): + await asyncio.gather( + *(connection() for _ in range(min(10, count - offset))) + ) + for _ in range(3): + await asyncio.sleep(0) + after = postgres_counts() + return { + "count": count, + "seconds": time.perf_counter() - started, + "live_object_delta": {name: after[name] - before[name] for name in before}, + "rss_before": rss_before, + "rss_after": rss_bytes(), + "asyncpg": version("asyncpg"), + } + finally: + sentinel.cancel() + + +async def performance_case(count): + # Reuse the repository's workloads, without tracemalloc or GC in timed spans. + from compare_event_loops import bench_callbacks + from scheduler_workloads import bench_timers + + samples = {} + for mode in ( + "timers_retained", + "timers_discarded", + "timers_cancelled", + "timers_mixed", + ): + await bench_timers("probe", 1000, 1000, mode) + samples[mode] = [ + (await bench_timers("probe", count, 1000, mode)).seconds for _ in range(7) + ] + await bench_callbacks("probe", 1000) + samples["call_soon"] = [ + (await bench_callbacks("probe", count)).seconds for _ in range(7) + ] + return { + "count": count, + "samples_seconds": samples, + "median_seconds": { + mode: statistics.median(values) for mode, values in samples.items() + }, + } + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--loop", choices=("asyncio", "rsloop", "uvloop"), required=True + ) + parser.add_argument( + "--case", + choices=("timers", "timeout", "postgres", "performance"), + required=True, + ) + parser.add_argument("--count", type=int, default=20000) + args = parser.parse_args() + if args.case == "timeout" and not hasattr(asyncio, "timeout"): + print( + json.dumps( + { + "loop": args.loop, + "case": args.case, + "skipped": "asyncio.timeout requires Python 3.11", + } + ) + ) + return + if args.count <= 0: + parser.error("--count must be positive") + if args.loop == "asyncio": + factory = asyncio.new_event_loop + else: + factory = __import__(args.loop).new_event_loop + loop = factory() + try: + if args.case == "postgres": + operation = postgres_case(args.count) + elif args.case == "performance": + operation = performance_case(args.count) + else: + operation = memory_case(args.count, args.case) + result = loop.run_until_complete(operation) + finally: + loop.close() + print( + json.dumps( + dict( + loop=args.loop, + case=args.case, + python=platform.python_version(), + platform=platform.platform(), + loop_version=version(args.loop) if args.loop != "asyncio" else None, + **result, + ), + sort_keys=True, + ) + ) + + +if __name__ == "__main__": + main() diff --git a/src/engine/callbacks.rs b/src/engine/callbacks.rs index fafce30..43df01d 100644 --- a/src/engine/callbacks.rs +++ b/src/engine/callbacks.rs @@ -1,8 +1,8 @@ //! Callback values and Python handle wrappers used by the loop engine. use std::sync::{ - Arc, Weak, - atomic::{AtomicBool, Ordering}, + Arc, Mutex, Weak, + atomic::{AtomicBool, AtomicUsize, Ordering}, }; use pyo3::{prelude::*, types::PyTuple}; @@ -341,6 +341,103 @@ impl PyHandle { } } +/// Mutable ownership exists only for timers: ordinary ready callbacks keep +/// their allocation size and lock-free invocation path. +pub struct TimerCallback { + id: CallbackId, + state: Mutex, + cancelled: AtomicBool, + cancelled_timers: Arc, +} + +struct TimerState { + ready: Option, + queued: bool, +} + +impl TimerCallback { + #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "TimerCallback"))] + pub fn new(ready: ReadyCallback, cancelled_timers: Arc) -> Self { + Self { + id: ready.id(), + state: Mutex::new(TimerState { + ready: Some(ready), + queued: true, + }), + cancelled: AtomicBool::new(false), + cancelled_timers, + } + } + + #[inline] + #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "TimerCallback"))] + pub fn id(&self) -> CallbackId { + self.id + } + + #[inline] + #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "TimerCallback"))] + pub fn cancelled(&self) -> bool { + self.cancelled.load(Ordering::Relaxed) + } + + /// Linearize cancellation and detach Python ownership under the same lock + /// as invocation. Drop outside it: DECREF can run reentrant finalizers. + /// The Python handle calls this while attached; Rust callers can also use + /// PyO3's deferred decref when detached. + #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "TimerCallback"))] + pub fn cancel(&self) { + let ready = { + let mut state = self.state.lock().expect("poisoned timer state"); + if self.cancelled.swap(true, Ordering::Relaxed) { + return; + } + if state.queued { + self.cancelled_timers.fetch_add(1, Ordering::Relaxed); + } + state.ready.take() + }; + drop(ready); + } + + /// Queue removal and cancellation share the lock so repeated cancellation, + /// cross-thread removal and loop shutdown count each tombstone once. + #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "TimerCallback"))] + pub(super) fn dequeue(&self) { + let mut state = self.state.lock().expect("poisoned timer state"); + if state.queued && self.cancelled() { + self.cancelled_timers.fetch_sub(1, Ordering::Relaxed); + } + state.queued = false; + } + + /// The invocation owns its payload until completion. A concurrent cancel + /// can prevent a not-yet-claimed invocation, but cannot invalidate a + /// running callback's arguments/context. No lock crosses Python + /// execution. + #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "TimerCallback"))] + pub(super) fn take_ready(&self) -> Option { + self.state + .lock() + .expect("poisoned timer state") + .ready + .take() + } +} + +impl Drop for TimerCallback { + #[cfg_attr( + feature = "profile", + hotpath::measure(impl_type = "") + )] + fn drop(&mut self) { + let state = self.state.get_mut().expect("poisoned timer state"); + if state.queued && *self.cancelled.get_mut() { + self.cancelled_timers.fetch_sub(1, Ordering::Relaxed); + } + } +} + /// Python's cancellable `asyncio.TimerHandle` equivalent. /// /// The weak callback reference lets cancellation update a live scheduled @@ -355,7 +452,7 @@ impl PyHandle { pub struct PyTimerHandle { callback_id: CallbackId, when: f64, - callback: Weak, + callback: Weak, cancelled: AtomicBool, } @@ -363,7 +460,7 @@ impl PyTimerHandle { #[inline] /// Creates a timer handle for a callback scheduled at loop time `when`. #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "PyTimerHandle"))] - pub fn new(callback_id: CallbackId, when: f64, callback: &Arc) -> Self { + pub fn new(callback_id: CallbackId, when: f64, callback: &Arc) -> Self { Self { callback_id, when, @@ -518,11 +615,92 @@ mod tests { }); } + #[test] + fn timer_cancellation_racing_queue_removal_counts_once() { + crate::initialize_python_for_tests(); + Python::attach(|py| { + let count = Arc::new(AtomicUsize::new(0)); + for id in 0..128 { + let timer = TimerCallback::new( + callback_with_args(py, id, PyTuple::empty(py).unbind()), + Arc::clone(&count), + ); + std::thread::scope(|scope| { + scope.spawn(|| { + timer.cancel(); + timer.cancel(); + }); + scope.spawn(|| { + timer.dequeue(); + timer.dequeue(); + }); + }); + assert!(timer.cancelled()); + assert!(timer.take_ready().is_none()); + assert_eq!(count.load(Ordering::Relaxed), 0); + } + }); + } + + #[test] + fn timer_claim_racing_cancel_never_invokes_twice_or_loses_ownership() { + crate::initialize_python_for_tests(); + Python::attach(|py| { + for id in 0..128 { + let count = Arc::new(AtomicUsize::new(0)); + let timer = TimerCallback::new( + callback_with_args(py, id, PyTuple::empty(py).unbind()), + Arc::clone(&count), + ); + timer.dequeue(); + let claimed = std::thread::scope(|scope| { + scope.spawn(|| timer.cancel()); + scope + .spawn(|| timer.take_ready()) + .join() + .expect("claim thread") + }); + if let Some(callback) = claimed { + assert_eq!( + callback + .invoke(py) + .expect("claimed invocation") + .extract::>(py) + .expect("result"), + Vec::::new() + ); + } + assert!(timer.take_ready().is_none()); + assert_eq!(count.load(Ordering::Relaxed), 0); + } + }); + } + + #[test] + fn dropping_queued_cancelled_timer_restores_count() { + crate::initialize_python_for_tests(); + Python::attach(|py| { + let count = Arc::new(AtomicUsize::new(0)); + let timer = TimerCallback::new( + callback_with_args(py, 1, PyTuple::empty(py).unbind()), + Arc::clone(&count), + ); + timer.cancel(); + timer.cancel(); + assert_eq!(count.load(Ordering::Relaxed), 1); + drop(timer); + assert_eq!(count.load(Ordering::Relaxed), 0); + }); + } + #[test] fn timer_handle_cancels_live_callback_and_tolerates_a_dropped_one() { crate::initialize_python_for_tests(); Python::attach(|py| { - let ready = Arc::new(callback_with_args(py, 12, PyTuple::empty(py).unbind())); + let ready = Arc::new(TimerCallback::new( + callback_with_args(py, 12, PyTuple::empty(py).unbind()), + Arc::new(AtomicUsize::new(0)), + )); let timer = PyTimerHandle::new(12, 1.25, &ready); assert!((timer.when() - 1.25).abs() < f64::EPSILON); assert!(!timer.cancelled()); @@ -535,7 +713,10 @@ mod tests { "" ); - let dropped = Arc::new(callback_with_args(py, 13, PyTuple::empty(py).unbind())); + let dropped = Arc::new(TimerCallback::new( + callback_with_args(py, 13, PyTuple::empty(py).unbind()), + Arc::new(AtomicUsize::new(0)), + )); let dropped_timer = PyTimerHandle::new(13, 2.0, &dropped); drop(dropped); dropped_timer diff --git a/src/engine/commands.rs b/src/engine/commands.rs index 6ffb0bc..945fb0e 100644 --- a/src/engine/commands.rs +++ b/src/engine/commands.rs @@ -7,7 +7,7 @@ use std::{ use pyo3::prelude::*; -use super::callbacks::{PyHandle, ReadyCallback}; +use super::callbacks::{PyHandle, ReadyCallback, TimerCallback}; use crate::{ fd_ops::RawFd, transport::{ @@ -23,6 +23,7 @@ use crate::{ /// GIL. pub enum ReadyItem { Callback(Arc), + TimerCallback(Arc), HandleCallback(Py), FutureSetResult(Box), FutureSetException(Box), diff --git a/src/engine/loop_core.rs b/src/engine/loop_core.rs index 5d3a671..eef4fe0 100644 --- a/src/engine/loop_core.rs +++ b/src/engine/loop_core.rs @@ -12,7 +12,7 @@ use std::{ pin::Pin, sync::{ Arc, Mutex, - atomic::{AtomicBool, AtomicU64, Ordering}, + atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}, mpsc, }, task::{Context, Poll, Waker}, @@ -30,7 +30,7 @@ use pyo3::{ #[cfg(unix)] use super::commands::{ConnectCompletion, TcpReaderStart}; use super::{ - callbacks::{CallbackArgs, CallbackId, CallbackKind, ReadyCallback}, + callbacks::{CallbackArgs, CallbackId, CallbackKind, ReadyCallback, TimerCallback}, commands::{ AcceptedConnection, FutureCompletion, LoopCommand, LoopFutureCommand, LoopIoCommand, LoopRunCommand, LoopTransportCommand, ReadyItem, @@ -232,6 +232,14 @@ impl<'a> LocalTimers<'a> { .expect("poisoned pending timers"), ); } + // Like asyncio, amortize linear heap rebuilding over large bursts. + // Small queues retain at most 100 tombstones; larger queues compact + // once more than half of their entries are cancelled. These relaxed + // atomics are statistical hints; per-timer ownership uses a mutex. + let cancelled = self.core.cancelled_timers.load(Ordering::Relaxed); + if cancelled > 100 && cancelled > self.heap.len() / 2 { + self.heap.compact_cancelled(ready); + } let now = Instant::now(); while self .heap @@ -241,7 +249,8 @@ impl<'a> LocalTimers<'a> { let entry = self.heap.pop().expect("timer heap was nonempty"); // Even cancelled callbacks are released by the ready drain, never // while the heap is mutably borrowed: finalizers can schedule work. - ready.push_back(ReadyItem::Callback(entry.callback)); + entry.callback.dequeue(); + ready.push_back(ReadyItem::TimerCallback(entry.callback)); } } @@ -382,6 +391,7 @@ pub struct LoopCore { active_ready_dispatch: Mutex>, pending_timers: Mutex>, pending_timers_dirty: AtomicBool, + cancelled_timers: Arc, // Wakes the loop thread when a producer enqueues a ready item. Held in an // Arc so the park future (`WaitForWake`) can own a clone under `py.detach`. wake: Arc, @@ -414,6 +424,7 @@ impl LoopCore { active_ready_dispatch: Mutex::new(None), pending_timers: Mutex::new(BinaryHeap::new()), pending_timers_dirty: AtomicBool::new(false), + cancelled_timers: Arc::new(AtomicUsize::new(0)), wake: Arc::new(LoopWake::new()), }); @@ -609,16 +620,19 @@ impl LoopCore { callback: Py, args: Py, context: Option>, - ) -> PyResult<(Arc, f64)> { + ) -> PyResult<(Arc, f64)> { let (captured, context_needs_run) = capture_context(py, context)?; - let ready = Arc::new(ReadyCallback::new( - py, - self.next_callback_id(), - CallbackKind::Timer, - callback, - args, - captured, - context_needs_run, + let ready = Arc::new(TimerCallback::new( + ReadyCallback::new( + py, + self.next_callback_id(), + CallbackKind::Timer, + callback, + args, + captured, + context_needs_run, + ), + Arc::clone(&self.cancelled_timers), )); let deadline = Instant::now() + delay; @@ -832,6 +846,14 @@ impl LoopCore { break; } } + ReadyItem::TimerCallback(timer) => { + if let Some(callback) = timer.take_ready() { + if let Some(err) = self.execute_ready(py, Some(&loop_obj), &callback)? { + ready_error = Some(err); + break; + } + } + } ReadyItem::FutureSetResult(payload) => { let FutureCompletion { future, value } = *payload; let future = future.bind(py); @@ -1503,6 +1525,9 @@ impl LoopCore { .try_enqueue_local_ready(ReadyItem::Callback(callback)) .or_else(|item| self.try_enqueue_active_ready(item)) .map_err(|item| match item { + ReadyItem::TimerCallback(_) => { + unreachable!("ordinary enqueue cannot produce timers") + } ReadyItem::Callback(callback) => LoopCommand::ScheduleReady(callback), ReadyItem::HandleCallback(handle) => LoopCommand::ScheduleReadyHandle(handle), ReadyItem::Stop => LoopCommand::RequestStop, @@ -1757,6 +1782,49 @@ mod wake_tests { }); } + #[test] + fn cancelled_timer_burst_is_compacted_behind_an_earlier_live_timer() { + crate::initialize_python_for_tests(); + Python::attach(|py| { + let core = LoopCore::new(); + core.schedule_timer( + py, + Duration::from_secs(1800), + py.None(), + PyTuple::empty(py).unbind(), + None, + ) + .expect("sentinel"); + for _ in 0..256 { + let (callback, _) = core + .schedule_timer( + py, + Duration::from_secs(3600), + py.None(), + PyTuple::empty(py).unbind(), + None, + ) + .expect("timer"); + callback.cancel(); + callback.cancel(); + } + let mut local = LocalTimers::new(&core); + let mut ready = VecDeque::new(); + local.collect(&mut ready); + let mut remaining = BinaryHeap::new(); + local.heap.drain_into(&mut remaining); + assert_eq!( + remaining.len(), + 1, + "cancelled shells retained behind sentinel" + ); + drop(local); + drop(remaining); + drop(ready); + core.close().expect("close loop"); + }); + } + struct WakeCounter(AtomicUsize); impl ArcWake for WakeCounter { diff --git a/src/engine/timer_entry.rs b/src/engine/timer_entry.rs index d2df46f..3448bf9 100644 --- a/src/engine/timer_entry.rs +++ b/src/engine/timer_entry.rs @@ -24,7 +24,7 @@ fn compare_timer_parts( pub(super) struct TimerEntry { pub(super) when: Instant, pub(super) seq: u64, - pub(super) callback: Arc, + pub(super) callback: Arc, } /// Immediately due timers normally arrive in deadline order and use a FIFO. @@ -48,6 +48,30 @@ impl TimerQueue { self.ordered.is_empty() && self.unordered.is_empty() } + #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "TimerQueue"))] + pub(super) fn len(&self) -> usize { + self.ordered.len() + self.unordered.len() + } + + /// Retain live entries in linear time, rebuilding the heap only after a + /// cancellation burst. Transfer removed shells to the ready drain rather + /// than dropping shared callback ownership under a queue borrow. + #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "TimerQueue"))] + pub(super) fn compact_cancelled(&mut self, ready: &mut VecDeque) { + let mut retain = |entry: &TimerEntry| { + if !entry.callback.cancelled() { + return true; + } + entry.callback.dequeue(); + ready.push_back(super::commands::ReadyItem::TimerCallback(Arc::clone( + &entry.callback, + ))); + false + }; + self.ordered.retain(&mut retain); + self.unordered.retain(retain); + } + #[cfg_attr(feature = "profile", hotpath::measure(impl_type = "TimerQueue"))] pub(super) fn push(&mut self, entry: TimerEntry, immediate: bool) { // TimerEntry's comparison is reversed for the min-deadline heap. @@ -203,20 +227,23 @@ mod tests { use pyo3::{prelude::*, types::PyTuple}; use super::{TimerEntry, TimerQueue}; - use crate::engine::callbacks::{CallbackKind, ReadyCallback}; + use crate::engine::callbacks::{CallbackKind, ReadyCallback, TimerCallback}; fn timer_entry(py: Python<'_>, when: Instant, seq: u64) -> TimerEntry { TimerEntry { when, seq, - callback: Arc::new(ReadyCallback::new( - py, - seq, - CallbackKind::Timer, - py.None(), - PyTuple::empty(py).unbind(), - py.None(), - false, + callback: Arc::new(TimerCallback::new( + ReadyCallback::new( + py, + seq, + CallbackKind::Timer, + py.None(), + PyTuple::empty(py).unbind(), + py.None(), + false, + ), + Arc::new(std::sync::atomic::AtomicUsize::new(0)), )), } } @@ -246,6 +273,40 @@ mod tests { }); } + #[test] + fn compaction_preserves_live_deadlines_and_ties_in_both_queues() { + crate::initialize_python_for_tests(); + Python::attach(|py| { + let now = Instant::now(); + let mut queue = TimerQueue::new(); + // FIFO prefix plus scattered heap arrivals, with equal deadlines. + for (seq, delay, immediate, cancelled) in [ + (0, 0, true, false), + (1, 0, true, true), + (2, 0, true, false), + (3, 30, false, true), + (4, 10, false, false), + (5, 10, false, false), + (6, 20, false, true), + ] { + let entry = timer_entry(py, now + Duration::from_millis(delay), seq); + if cancelled { + entry.callback.cancel(); + } + queue.push(entry, immediate); + } + let mut removed = std::collections::VecDeque::new(); + queue.compact_cancelled(&mut removed); + assert_eq!(removed.len(), 3); + assert_eq!(queue.len(), 4); + let mut live = Vec::new(); + while let Some(entry) = queue.pop() { + live.push(entry.seq); + } + assert_eq!(live, [0, 2, 4, 5]); + }); + } + proptest::proptest! { #[test] fn timer_queue_matches_heap_through_insert_pop_and_merge( diff --git a/tests/ecosystem/test_postgresql.py b/tests/ecosystem/test_postgresql.py index 092e890..ba40425 100644 --- a/tests/ecosystem/test_postgresql.py +++ b/tests/ecosystem/test_postgresql.py @@ -1,6 +1,7 @@ """Real PostgreSQL contracts; opt in with a dedicated RSLOOP_POSTGRES_DSN.""" import asyncio +import gc import os import uuid from contextlib import asynccontextmanager @@ -255,3 +256,47 @@ async def main(): await observer.close() loop.run_until_complete(asyncio.wait_for(main(), 20)) + + +def test_closed_connections_release_statement_cache_before_expiry(loop, postgres_dsn): + """A live earlier timer must not pin cache expiry callbacks after close (#118).""" + names = {"_StatementCache", "PreparedStatementState", "Protocol"} + + def counts(): + gc.collect() + result = dict.fromkeys(names, 0) + for obj in gc.get_objects(): + name = type(obj).__name__ + if name in result: + result[name] += 1 + return result + + async def main(): + sentinel = loop.call_later(1800, lambda: None) + before = counts() + + async def connection(): + conn = await asyncpg.connect( + postgres_dsn, max_cached_statement_lifetime=3600 + ) + try: + for index in range(3): + assert ( + await conn.fetchval(f"SELECT $1::int + {index}", index) + == 2 * index + ) + finally: + await conn.close() + + try: + for _ in range(5): + await asyncio.gather(*(connection() for _ in range(10))) + for _ in range(3): + await asyncio.sleep(0) + after = counts() + delta = {name: after[name] - before[name] for name in names} + assert all(value <= 0 for value in delta.values()), delta + finally: + sentinel.cancel() + + loop.run_until_complete(asyncio.wait_for(main(), 30)) diff --git a/tests/test_timer_cancellation.py b/tests/test_timer_cancellation.py new file mode 100644 index 0000000..ab7c04a --- /dev/null +++ b/tests/test_timer_cancellation.py @@ -0,0 +1,311 @@ +"""Cancelled timers release Python ownership before their heap deadline (#118).""" + +from __future__ import annotations + +import asyncio +import contextvars +import gc +import threading +import weakref + +import pytest +import rsloop + + +@pytest.fixture(params=["asyncio", "rsloop", "uvloop"]) +def loop(request): + factory = { + "asyncio": asyncio.new_event_loop, + "rsloop": rsloop.new_event_loop, + }.get(request.param) + if factory is None: + factory = pytest.importorskip("uvloop").new_event_loop + event_loop = factory() + yield event_loop + event_loop.close() + + +class Payload: + def __init__(self): + self.blob = bytearray(4096) + + def __call__(self, *args): + pytest.fail("cancelled callback executed") + + +def make_timer(loop, ownership, many=False): + payload = Payload() + reference = weakref.ref(payload) + if ownership == "callback": + handle = loop.call_later(3600, payload) + elif ownership == "args": + args = (payload, payload) if many else (payload,) + handle = loop.call_later(3600, lambda *args: None, *args) + else: + var = contextvars.ContextVar("timer_payload") + context = contextvars.Context() + context.run(var.set, payload) + handle = loop.call_later(3600, lambda: None, context=context) + return handle, reference + + +@pytest.mark.parametrize("ownership", ["callback", "args", "context"]) +@pytest.mark.parametrize("running", [False, True]) +def test_cancel_releases_python_references_immediately(loop, ownership, running): + def check(): + sentinel = loop.call_later(1800, lambda: None) + try: + handle, reference = make_timer(loop, ownership) + assert reference() is not None + handle.cancel() + handle.cancel() + # No GC or loop turn is needed for these acyclic payloads. + if ownership == "context" and not isinstance(loop, rsloop.Loop): + # CPython and uvloop retain Context on the user-held handle. + assert reference() is not None + else: + assert reference() is None + assert handle.cancelled() + finally: + sentinel.cancel() + + async def main(): + check() + + if running: + loop.run_until_complete(main()) + else: + check() + + +def test_cancel_releases_many_arguments_and_captured_context(loop): + sentinel = loop.call_later(1800, lambda: None) + try: + handle, reference = make_timer(loop, "args", many=True) + handle.cancel() + assert reference() is None + var = contextvars.ContextVar("implicit_timer_payload") + payload = Payload() + reference = weakref.ref(payload) + token = var.set(payload) + handle = loop.call_later(3600, lambda: None) + var.reset(token) + del payload + handle.cancel() + if isinstance(loop, rsloop.Loop): + assert reference() is None + else: + assert reference() is not None + finally: + sentinel.cancel() + + +def test_cancelled_burst_behind_live_timer_releases_all_payloads(loop): + async def main(): + sentinel = loop.call_later(1800, lambda: None) + try: + references = [] + for _ in range(1000): + handle, reference = make_timer(loop, "callback") + references.append(reference) + handle.cancel() + await asyncio.sleep(0) + gc.collect() + assert all(reference() is None for reference in references) + finally: + sentinel.cancel() + + loop.run_until_complete(main()) + + +@pytest.mark.parametrize("mode", ["timeout", "wait_for"]) +def test_completed_timeout_tasks_and_results_are_released(loop, mode): + if mode == "timeout" and not hasattr(asyncio, "timeout"): + pytest.skip("asyncio.timeout requires Python 3.11") + + async def request(): + if mode == "timeout": + async with asyncio.timeout(3600): + await asyncio.sleep(0) + else: + await asyncio.wait_for(asyncio.sleep(0), 3600) + return Payload() + + async def main(): + sentinel = loop.call_later(1800, lambda: None) + try: + tasks = [loop.create_task(request()) for _ in range(100)] + await asyncio.gather(*tasks) + references = [weakref.ref(task) for task in tasks] + references.extend(weakref.ref(task.result()) for task in tasks) + del tasks + # Release gather's completion callbacks; no deadline may expire. + for _ in range(3): + await asyncio.sleep(0) + gc.collect() + assert all(reference() is None for reference in references) + finally: + sentinel.cancel() + + loop.run_until_complete(main()) + + +@pytest.mark.parametrize("ownership", ["callback", "args", "context"]) +def test_cancel_finalizers_can_reenter_and_close_loop(loop, ownership): + if not isinstance(loop, rsloop.Loop): + pytest.skip("reentrant close is a separate rsloop lifecycle contract") + events = [] + handles = [] + + class Callback: + def __call__(self): + pytest.fail("cancelled callback executed") + + def __del__(self): + events.append("released") + handles[0].cancel() + if not loop.is_closed(): + loop.call_later(0, lambda: None).cancel() + loop.close() + + if ownership == "callback": + handles.append(loop.call_later(3600, Callback())) + elif ownership == "args": + handles.append(loop.call_later(3600, lambda arg: None, Callback())) + else: + var = contextvars.ContextVar("finalizer_payload") + context = contextvars.Context() + context.run(var.set, Callback()) + handles.append(loop.call_later(3600, lambda: None, context=context)) + del context + handles[0].cancel() + released_immediately = events == ["released"] + if not released_immediately: + loop.close() + assert released_immediately + assert handles[0].cancelled() + assert loop.is_closed() + + +def test_cancel_after_dequeue_skips_callback(loop): + events = [] + when = loop.time() + # call_later's deadlines give the cancellation callback a strictly earlier + # position on loops without stable ordering for equal deadlines. + loop.call_at(when - 1, lambda: handle.cancel()) + handle = loop.call_at(when, lambda: events.append("unexpected")) + loop.run_until_complete(asyncio.sleep(0)) + assert events == [] + + +def test_concurrent_repeated_cancel_releases_payload_once(loop): + handle, reference = make_timer(loop, "callback") + barrier = threading.Barrier(9) + + def cancel(): + barrier.wait(timeout=5) + for _ in range(100): + handle.cancel() + + threads = [threading.Thread(target=cancel) for _ in range(8)] + for thread in threads: + thread.start() + barrier.wait(timeout=5) + for thread in threads: + thread.join(timeout=5) + assert not thread.is_alive() + assert handle.cancelled() + assert reference() is None + + +def test_cancel_running_callback_preserves_arguments_and_context(loop): + entered = threading.Event() + cancelled = threading.Event() + var = contextvars.ContextVar("running_timer_value") + context = contextvars.Context() + context.run(var.set, "captured") + events = [] + + def callback(payload): + entered.set() + assert cancelled.wait(timeout=5) + assert payload.blob == bytearray(4096) + assert var.get() == "captured" + events.append("finished") + + payload = Payload() + reference = weakref.ref(payload) + handle = loop.call_later(0, callback, payload, context=context) + del payload + + def cancel(): + if entered.wait(timeout=5): + handle.cancel() + cancelled.set() + + thread = threading.Thread(target=cancel) + thread.start() + try: + loop.run_until_complete(asyncio.sleep(0)) + assert events == ["finished"] + gc.collect() + assert reference() is None + finally: + cancelled.set() + thread.join(timeout=5) + assert not thread.is_alive() + + +@pytest.mark.parametrize("mode", ["timeout", "wait_for"]) +def test_expired_timeout_still_cancels_awaited_work(loop, mode): + if mode == "timeout" and not hasattr(asyncio, "timeout"): + pytest.skip("asyncio.timeout requires Python 3.11") + + async def main(): + finalized = [] + + async def work(): + try: + await asyncio.Future() + finally: + finalized.append("cancelled") + + with pytest.raises(asyncio.TimeoutError): + if mode == "timeout": + async with asyncio.timeout(0.01): + await work() + else: + await asyncio.wait_for(work(), 0.01) + assert finalized == ["cancelled"] + assert len(asyncio.all_tasks()) == 1 + + loop.run_until_complete(main()) + + +def test_periodic_timers_do_not_retain_cancelled_payloads(loop): + async def main(): + started = [asyncio.Event(), asyncio.Event()] + + async def heartbeat(index): + started[index].set() + while True: + await asyncio.sleep(0.01) + + beats = [loop.create_task(heartbeat(index)) for index in range(2)] + try: + await asyncio.gather(*(event.wait() for event in started)) + references = [] + for _ in range(100): + handle, reference = make_timer(loop, "callback") + handle.cancel() + references.append(reference) + for _ in range(3): + await asyncio.sleep(0.01) + gc.collect() + assert all(reference() is None for reference in references) + finally: + for beat in beats: + beat.cancel() + await asyncio.gather(*beats, return_exceptions=True) + + loop.run_until_complete(main())