From d762219fe0561ee6862b634fbae5c1f4791ae1c2 Mon Sep 17 00:00:00 2001 From: Dmitry Agafonov <42949186+Malkiz223@users.noreply.github.com> Date: Sun, 30 Aug 2026 14:41:19 +0300 Subject: [PATCH] Avoid GC and value-finalizer deadlocks while the cache lock is held --- cachebox/_core.pyi | 91 ++++++--- src/internal/lazyheap.rs | 12 ++ src/policies/fifopolicy.rs | 21 +- src/policies/lfupolicy.rs | 21 +- src/policies/lrupolicy.rs | 24 ++- src/policies/nopolicy.rs | 24 ++- src/policies/rrpolicy.rs | 24 ++- src/policies/traits.rs | 13 ++ src/policies/ttlpolicy.rs | 34 +++- src/policies/vttlpolicy.rs | 22 +- src/policies/wrapped.rs | 245 ++++++++++++++++++---- src/pyclasses/cache.rs | 68 +++++-- src/pyclasses/fifocache.rs | 68 +++++-- src/pyclasses/lfucache.rs | 68 +++++-- src/pyclasses/lrucache.rs | 68 +++++-- src/pyclasses/rrcache.rs | 68 +++++-- src/pyclasses/ttlcache.rs | 68 +++++-- src/pyclasses/vttlcache.rs | 74 +++++-- tests/mixins.py | 402 +++++++++++++++++++++++++++++++++++++ tests/test_impls.py | 38 ++++ 20 files changed, 1267 insertions(+), 186 deletions(-) diff --git a/cachebox/_core.pyi b/cachebox/_core.pyi index e6a7dd2..ae6bba4 100644 --- a/cachebox/_core.pyi +++ b/cachebox/_core.pyi @@ -410,6 +410,10 @@ class Cache(BaseCacheImpl[KT, VT]): Note: Use `setdefault_with`, if computing the value is expensive or has side effectes. + + `getsizeof` runs with the internal lock released; if another thread + inserts the key meanwhile, that value wins, and if the losing + `getsizeof` raises, its exception still propagates to this caller. """ ... @@ -429,10 +433,11 @@ class Cache(BaseCacheImpl[KT, VT]): factory: The factory to call and get default value from if ``key`` is not in the cache. Warning: - if two threads miss the same key at once, `factory` can run - more than once; the value inserted first wins and is returned to - both. If `factory` raises, nothing is inserted and the exception - propagates. + if two threads miss the same key at once, `factory` (and + `getsizeof`) can run more than once; the value inserted first wins + and is returned to both callers that succeed. If the losing call's + `factory` or `getsizeof` raises, nothing more is inserted and the + exception still propagates to that caller. """ ... @@ -592,6 +597,10 @@ class FIFOCache(BaseCacheImpl[KT, VT]): Note: Use `setdefault_with`, if computing the value is expensive or has side effectes. + + `getsizeof` runs with the internal lock released; if another thread + inserts the key meanwhile, that value wins, and if the losing + `getsizeof` raises, its exception still propagates to this caller. """ ... @@ -611,10 +620,11 @@ class FIFOCache(BaseCacheImpl[KT, VT]): factory: The factory to call and get default value from if ``key`` is not in the cache. Warning: - if two threads miss the same key at once, `factory` can run - more than once; the value inserted first wins and is returned to - both. If `factory` raises, nothing is inserted and the exception - propagates. + if two threads miss the same key at once, `factory` (and + `getsizeof`) can run more than once; the value inserted first wins + and is returned to both callers that succeed. If the losing call's + `factory` or `getsizeof` raises, nothing more is inserted and the + exception still propagates to that caller. """ ... @@ -807,6 +817,10 @@ class RRCache(BaseCacheImpl[KT, VT]): Note: Use `setdefault_with`, if computing the value is expensive or has side effectes. + + `getsizeof` runs with the internal lock released; if another thread + inserts the key meanwhile, that value wins, and if the losing + `getsizeof` raises, its exception still propagates to this caller. """ ... @@ -826,10 +840,11 @@ class RRCache(BaseCacheImpl[KT, VT]): factory: The factory to call and get default value from if ``key`` is not in the cache. Warning: - if two threads miss the same key at once, `factory` can run - more than once; the value inserted first wins and is returned to - both. If `factory` raises, nothing is inserted and the exception - propagates. + if two threads miss the same key at once, `factory` (and + `getsizeof`) can run more than once; the value inserted first wins + and is returned to both callers that succeed. If the losing call's + `factory` or `getsizeof` raises, nothing more is inserted and the + exception still propagates to that caller. """ ... @@ -1015,6 +1030,10 @@ class LRUCache(BaseCacheImpl[KT, VT]): Note: Use `setdefault_with`, if computing the value is expensive or has side effectes. + + `getsizeof` runs with the internal lock released; if another thread + inserts the key meanwhile, that value wins, and if the losing + `getsizeof` raises, its exception still propagates to this caller. """ ... @@ -1034,10 +1053,11 @@ class LRUCache(BaseCacheImpl[KT, VT]): factory: The factory to call and get default value from if ``key`` is not in the cache. Warning: - if two threads miss the same key at once, `factory` can run - more than once; the value inserted first wins and is returned to - both. If `factory` raises, nothing is inserted and the exception - propagates. + if two threads miss the same key at once, `factory` (and + `getsizeof`) can run more than once; the value inserted first wins + and is returned to both callers that succeed. If the losing call's + `factory` or `getsizeof` raises, nothing more is inserted and the + exception still propagates to that caller. """ ... @@ -1264,6 +1284,10 @@ class LFUCache(BaseCacheImpl[KT, VT]): Note: Use `setdefault_with`, if computing the value is expensive or has side effectes. + + `getsizeof` runs with the internal lock released; if another thread + inserts the key meanwhile, that value wins, and if the losing + `getsizeof` raises, its exception still propagates to this caller. """ ... @@ -1283,10 +1307,11 @@ class LFUCache(BaseCacheImpl[KT, VT]): factory: The factory to call and get default value from if ``key`` is not in the cache. Warning: - if two threads miss the same key at once, `factory` can run - more than once; the value inserted first wins and is returned to - both. If `factory` raises, nothing is inserted and the exception - propagates. + if two threads miss the same key at once, `factory` (and + `getsizeof`) can run more than once; the value inserted first wins + and is returned to both callers that succeed. If the losing call's + `factory` or `getsizeof` raises, nothing more is inserted and the + exception still propagates to that caller. """ ... @@ -1472,6 +1497,10 @@ class TTLCache(BaseCacheImpl[KT, VT]): Note: Use `setdefault_with`, if computing the value is expensive or has side effectes. + + `getsizeof` runs with the internal lock released; if another thread + inserts the key meanwhile, that value wins, and if the losing + `getsizeof` raises, its exception still propagates to this caller. """ ... @@ -1491,10 +1520,11 @@ class TTLCache(BaseCacheImpl[KT, VT]): factory: The factory to call and get default value from if ``key`` is not in the cache. Warning: - if two threads miss the same key at once, `factory` can run - more than once; the value inserted first wins and is returned to - both. If `factory` raises, nothing is inserted and the exception - propagates. + if two threads miss the same key at once, `factory` (and + `getsizeof`) can run more than once; the value inserted first wins + and is returned to both callers that succeed. If the losing call's + `factory` or `getsizeof` raises, nothing more is inserted and the + exception still propagates to that caller. """ ... @@ -1734,6 +1764,10 @@ class VTTLCache(BaseCacheImpl[KT, VT]): Use `setdefault_with`, if computing the value is expensive or has side effectes. + `getsizeof` runs with the internal lock released; if another thread + inserts the key meanwhile, that value wins, and if the losing + `getsizeof` raises, its exception still propagates to this caller. + Args: key: The key to look up or insert. default: The value to insert if ``key`` is not in the cache. @@ -1760,10 +1794,11 @@ class VTTLCache(BaseCacheImpl[KT, VT]): ttl: An optional time-to-live duration for item. Warning: - if two threads miss the same key at once, `factory` can run - more than once; the value inserted first wins and is returned to - both. If `factory` raises, nothing is inserted and the exception - propagates. + if two threads miss the same key at once, `factory` (and + `getsizeof`) can run more than once; the value inserted first wins + and is returned to both callers that succeed. If the losing call's + `factory` or `getsizeof` raises, nothing more is inserted and the + exception still propagates to that caller. """ ... diff --git a/src/internal/lazyheap.rs b/src/internal/lazyheap.rs index b63f05b..a07434d 100644 --- a/src/internal/lazyheap.rs +++ b/src/internal/lazyheap.rs @@ -228,6 +228,18 @@ impl LazyHeap { self.is_sorted = true; } + /// Moves every element out into `out`, without sorting. + /// + /// The heap is empty and considered sorted after this call. + #[inline] + pub fn drain_into(&mut self, out: &mut Vec) { + out.reserve(self.data.len()); + while let Some(element) = self.unlink_back() { + out.push(element); + } + self.is_sorted = true; + } + /// Shrinks the backing buffer's capacity as close to its current length /// as possible. #[inline] diff --git a/src/policies/fifopolicy.rs b/src/policies/fifopolicy.rs index 4afc7fc..b2be35b 100644 --- a/src/policies/fifopolicy.rs +++ b/src/policies/fifopolicy.rs @@ -87,7 +87,8 @@ impl traits::VacantExt for Vacant<'_> { #[inline] fn evict(&mut self) -> pyo3::PyResult<()> { - self.policy.evict(self.shared)?; + let handle = self.policy.evict(self.shared)?; + self.policy.pending_drops.push(handle); Ok(()) } @@ -120,6 +121,10 @@ pub struct FIFOPolicy { /// Running total of all stored handles' sizes, maintained incrementally. currsize: usize, + /// Handles parked for destruction after the lock is released; + /// see [`super::traits::PolicyExt::pending_drops`]. + pending_drops: Vec, + /// Number of handles ever popped from the front of [`FIFOPolicy::entries`]. /// /// Because [`VecDeque`] indices shift on front-removal, naively keeping @@ -146,6 +151,7 @@ impl FIFOPolicy { table: hashbrown::raw::RawTable::with_capacity(capacity), entries: VecDeque::with_capacity(capacity), currsize: 0, + pending_drops: Vec::new(), front_offset: 0, } } @@ -314,6 +320,16 @@ impl PolicyExt for FIFOPolicy { Ok(front) } + #[inline(always)] + fn pending_drops(&mut self) -> &mut Vec { + &mut self.pending_drops + } + + #[inline(always)] + fn len(&self) -> usize { + self.entries.len() + } + #[inline] fn shrink_to_fit(&mut self, shared: &Self::Shared) { shared.generation_version().increment(); @@ -332,7 +348,7 @@ impl PolicyExt for FIFOPolicy { shared.generation_version().increment(); self.table.clear(); - self.entries.clear(); + self.pending_drops.extend(self.entries.drain(..)); self.currsize = 0; self.front_offset = 0; } @@ -392,6 +408,7 @@ impl PolicyExt for FIFOPolicy { table: self.table.clone(), entries, currsize: self.currsize, + pending_drops: Vec::new(), front_offset: self.front_offset, } } diff --git a/src/policies/lfupolicy.rs b/src/policies/lfupolicy.rs index d5ea80a..54189a1 100644 --- a/src/policies/lfupolicy.rs +++ b/src/policies/lfupolicy.rs @@ -199,7 +199,8 @@ impl traits::VacantExt for Vacant<'_> { #[inline] fn evict(&mut self) -> pyo3::PyResult<()> { - self.policy.evict(self.shared)?; + let handle = self.policy.evict(self.shared)?; + self.policy.pending_drops.push(handle); Ok(()) } @@ -226,6 +227,10 @@ pub struct LFUPolicy { /// Running total of all stored handles' sizes, maintained incrementally. currsize: usize, + + /// Handles parked for destruction after the lock is released; + /// see [`super::traits::PolicyExt::pending_drops`]. + pending_drops: Vec, } impl LFUPolicy { @@ -238,6 +243,7 @@ impl LFUPolicy { table: hashbrown::raw::RawTable::with_capacity(capacity), heap: lazyheap::LazyHeap::new(), currsize: 0, + pending_drops: Vec::new(), } } @@ -388,6 +394,16 @@ impl PolicyExt for LFUPolicy { Ok(handle) } + #[inline(always)] + fn pending_drops(&mut self) -> &mut Vec { + &mut self.pending_drops + } + + #[inline(always)] + fn len(&self) -> usize { + self.heap.len() + } + fn clear(&mut self, shared: &Self::Shared) { if self.heap.is_empty() { return; @@ -395,7 +411,7 @@ impl PolicyExt for LFUPolicy { shared.generation_version().increment(); self.table.clear_no_drop(); - self.heap.clear(); + self.heap.drain_into(&mut self.pending_drops); self.currsize = 0; } @@ -472,6 +488,7 @@ impl PolicyExt for LFUPolicy { table, heap, currsize: self.currsize, + pending_drops: Vec::new(), } } diff --git a/src/policies/lrupolicy.rs b/src/policies/lrupolicy.rs index 73f248b..829ca14 100644 --- a/src/policies/lrupolicy.rs +++ b/src/policies/lrupolicy.rs @@ -75,7 +75,8 @@ impl traits::VacantExt for Vacant<'_> { #[inline] fn evict(&mut self) -> pyo3::PyResult<()> { - self.policy.evict(self.shared)?; + let handle = self.policy.evict(self.shared)?; + self.policy.pending_drops.push(handle); Ok(()) } @@ -102,6 +103,10 @@ pub struct LRUPolicy { /// Running total of all stored handles' sizes, maintained incrementally. currsize: usize, + + /// Handles parked for destruction after the lock is released; + /// see [`super::traits::PolicyExt::pending_drops`]. + pending_drops: Vec, } impl LRUPolicy { @@ -114,6 +119,7 @@ impl LRUPolicy { table: hashbrown::raw::RawTable::with_capacity(capacity), list: linked_list::LinkedList::new(), currsize: 0, + pending_drops: Vec::new(), } } @@ -240,6 +246,16 @@ impl PolicyExt for LRUPolicy { Ok(handle) } + #[inline(always)] + fn pending_drops(&mut self) -> &mut Vec { + &mut self.pending_drops + } + + #[inline(always)] + fn len(&self) -> usize { + self.list.len() + } + #[inline] fn shrink_to_fit(&mut self, _shared: &Self::Shared) { self.table @@ -254,7 +270,10 @@ impl PolicyExt for LRUPolicy { shared.generation_version().increment(); self.table.clear_no_drop(); - self.list.clear(); + self.pending_drops.reserve(self.list.len()); + while let Some(handle) = self.list.pop_front() { + self.pending_drops.push(handle); + } self.currsize = 0; } @@ -319,6 +338,7 @@ impl PolicyExt for LRUPolicy { table, list: entries, currsize: self.currsize, + pending_drops: Vec::new(), } } diff --git a/src/policies/nopolicy.rs b/src/policies/nopolicy.rs index b45f05f..e08ab59 100644 --- a/src/policies/nopolicy.rs +++ b/src/policies/nopolicy.rs @@ -63,7 +63,8 @@ impl traits::VacantExt for Vacant<'_> { #[inline(always)] fn evict(&mut self) -> pyo3::PyResult<()> { - self.policy.evict(self.shared)?; + let handle = self.policy.evict(self.shared)?; + self.policy.pending_drops.push(handle); Ok(()) } @@ -87,6 +88,10 @@ pub struct NoPolicy { table: hashbrown::raw::RawTable, /// Running total of all stored handles' sizes, maintained incrementally. currsize: usize, + + /// Handles parked for destruction after the lock is released; + /// see [`super::traits::PolicyExt::pending_drops`]. + pending_drops: Vec, } impl NoPolicy { @@ -98,6 +103,7 @@ impl NoPolicy { Self { table: hashbrown::raw::RawTable::with_capacity(capacity), currsize: 0, + pending_drops: Vec::new(), } } @@ -174,6 +180,16 @@ impl PolicyExt for NoPolicy { )) } + #[inline(always)] + fn pending_drops(&mut self) -> &mut Vec { + &mut self.pending_drops + } + + #[inline(always)] + fn len(&self) -> usize { + self.table.len() + } + #[inline] fn shrink_to_fit(&mut self, shared: &Self::Shared) { shared.generation_version().increment(); @@ -185,8 +201,11 @@ impl PolicyExt for NoPolicy { if self.table.is_empty() { return; } - self.table.clear(); shared.generation_version().increment(); + self.pending_drops.reserve(self.table.len()); + for handle in self.table.drain() { + self.pending_drops.push(handle); + } self.currsize = 0; } @@ -241,6 +260,7 @@ impl PolicyExt for NoPolicy { Self { table, currsize: self.currsize, + pending_drops: Vec::new(), } } diff --git a/src/policies/rrpolicy.rs b/src/policies/rrpolicy.rs index fd60c6e..6afd1c9 100644 --- a/src/policies/rrpolicy.rs +++ b/src/policies/rrpolicy.rs @@ -65,7 +65,8 @@ impl traits::VacantExt for Vacant<'_> { #[inline(always)] fn evict(&mut self) -> pyo3::PyResult<()> { - self.policy.evict(self.shared)?; + let handle = self.policy.evict(self.shared)?; + self.policy.pending_drops.push(handle); Ok(()) } @@ -89,6 +90,10 @@ pub struct RRPolicy { table: hashbrown::raw::RawTable, /// Running total of all stored handles' sizes, maintained incrementally. currsize: usize, + + /// Handles parked for destruction after the lock is released; + /// see [`super::traits::PolicyExt::pending_drops`]. + pending_drops: Vec, } impl RRPolicy { @@ -100,6 +105,7 @@ impl RRPolicy { Self { table: hashbrown::raw::RawTable::with_capacity(capacity), currsize: 0, + pending_drops: Vec::new(), } } @@ -185,6 +191,16 @@ impl PolicyExt for RRPolicy { } } + #[inline(always)] + fn pending_drops(&mut self) -> &mut Vec { + &mut self.pending_drops + } + + #[inline(always)] + fn len(&self) -> usize { + self.table.len() + } + #[inline] fn shrink_to_fit(&mut self, shared: &Self::Shared) { shared.generation_version().increment(); @@ -196,8 +212,11 @@ impl PolicyExt for RRPolicy { if self.table.is_empty() { return; } - self.table.clear(); shared.generation_version().increment(); + self.pending_drops.reserve(self.table.len()); + for handle in self.table.drain() { + self.pending_drops.push(handle); + } self.currsize = 0; } @@ -252,6 +271,7 @@ impl PolicyExt for RRPolicy { Self { table, currsize: self.currsize, + pending_drops: Vec::new(), } } diff --git a/src/policies/traits.rs b/src/policies/traits.rs index 9a17dfa..591275a 100644 --- a/src/policies/traits.rs +++ b/src/policies/traits.rs @@ -97,6 +97,9 @@ pub trait PolicyExt: Sized { /// Returns the current total cumulative size consumed by all stored entries. fn current_size(&self) -> usize; + /// Returns the number of stored entries. + fn len(&self) -> usize; + /// Looks up a handle by `hash` and `eq`, applying policy side effects on hit. /// /// # Errors @@ -122,6 +125,16 @@ pub trait PolicyExt: Sized { /// Evicts a handle according to the policy algorithm, returning it. fn evict(&mut self, shared: &Self::Shared) -> pyo3::PyResult; + /// Returns the buffer of handles whose destruction is deferred until the + /// policy's lock is released. + /// + /// Dropping a handle can run Python code (the value's ``__del__``), and + /// running Python while the lock is held deadlocks if that code touches + /// the same cache. Internal operations that remove handles park them here + /// instead of dropping them; [`super::wrapped::PolicyGuard`] empties the + /// buffer right after it releases the lock. + fn pending_drops(&mut self) -> &mut Vec; + /// Removes all handles without shrinking the allocation. fn clear(&mut self, shared: &Self::Shared); diff --git a/src/policies/ttlpolicy.rs b/src/policies/ttlpolicy.rs index a4159b5..3bd014c 100644 --- a/src/policies/ttlpolicy.rs +++ b/src/policies/ttlpolicy.rs @@ -200,7 +200,8 @@ impl traits::VacantExt for Vacant<'_> { #[inline] fn evict(&mut self) -> pyo3::PyResult<()> { - self.policy.evict(self.shared)?; + let handle = self.policy.evict(self.shared)?; + self.policy.pending_drops.push(handle); Ok(()) } @@ -223,6 +224,10 @@ pub struct TTLPolicy { table: hashbrown::raw::RawTable, entries: VecDeque, currsize: usize, + + /// Handles parked for destruction after the lock is released; + /// see [`super::traits::PolicyExt::pending_drops`]. + pending_drops: Vec, front_offset: usize, } @@ -236,6 +241,7 @@ impl TTLPolicy { table: hashbrown::raw::RawTable::with_capacity(capacity), entries: VecDeque::with_capacity(capacity), currsize: 0, + pending_drops: Vec::new(), front_offset: 0, } } @@ -315,6 +321,18 @@ impl TTLPolicy { pub fn expire(&mut self, gv: &utils::GenerationVersion) { let now = std::time::SystemTime::now(); + // The queue is ordered by expiry: once the front has expired, the + // expired prefix is countable up front. + match self.entries.front() { + Some(front) if front.is_expired(now) => { + let expired = self + .entries + .partition_point(|handle| handle.is_expired(now)); + self.pending_drops.reserve(expired); + } + _ => return, + } + while let Some(handle) = self.entries.front() { if !handle.is_expired(now) { break; @@ -332,6 +350,7 @@ impl TTLPolicy { self.currsize = self.currsize.saturating_sub(front.size()); self.decrement_indexes(1, self.entries.len()); + self.pending_drops.push(front); } } @@ -441,6 +460,16 @@ impl PolicyExt for TTLPolicy { Ok(front) } + #[inline(always)] + fn pending_drops(&mut self) -> &mut Vec { + &mut self.pending_drops + } + + #[inline(always)] + fn len(&self) -> usize { + self.entries.len() + } + #[inline] fn shrink_to_fit(&mut self, shared: &Self::Shared) { shared.generation_version().increment(); @@ -458,7 +487,7 @@ impl PolicyExt for TTLPolicy { shared.generation_version().increment(); self.table.clear(); - self.entries.clear(); + self.pending_drops.extend(self.entries.drain(..)); self.currsize = 0; self.front_offset = 0; } @@ -523,6 +552,7 @@ impl PolicyExt for TTLPolicy { table: self.table.clone(), entries, currsize: self.currsize, + pending_drops: Vec::new(), front_offset: self.front_offset, } } diff --git a/src/policies/vttlpolicy.rs b/src/policies/vttlpolicy.rs index 54a596c..e6ede01 100644 --- a/src/policies/vttlpolicy.rs +++ b/src/policies/vttlpolicy.rs @@ -201,7 +201,8 @@ impl traits::VacantExt for Vacant<'_> { #[inline] fn evict(&mut self) -> pyo3::PyResult<()> { - self.policy.evict(self.shared)?; + let handle = self.policy.evict(self.shared)?; + self.policy.pending_drops.push(handle); Ok(()) } @@ -224,6 +225,10 @@ pub struct VTTLPolicy { table: hashbrown::raw::RawTable>, heap: lazyheap::LazyHeap, currsize: usize, + + /// Handles parked for destruction after the lock is released; + /// see [`super::traits::PolicyExt::pending_drops`]. + pending_drops: Vec, } impl VTTLPolicy { @@ -236,6 +241,7 @@ impl VTTLPolicy { table: hashbrown::raw::RawTable::with_capacity(capacity), heap: lazyheap::LazyHeap::new(), currsize: 0, + pending_drops: Vec::new(), } } @@ -282,6 +288,7 @@ impl VTTLPolicy { let handle = self.heap.pop_front(compare_fn!()).unwrap(); self.currsize = self.currsize.saturating_sub(handle.size); + self.pending_drops.push(handle); } } } @@ -384,6 +391,16 @@ impl PolicyExt for VTTLPolicy { Ok(handle) } + #[inline(always)] + fn pending_drops(&mut self) -> &mut Vec { + &mut self.pending_drops + } + + #[inline(always)] + fn len(&self) -> usize { + self.heap.len() + } + fn clear(&mut self, shared: &Self::Shared) { if self.heap.is_empty() { return; @@ -391,7 +408,7 @@ impl PolicyExt for VTTLPolicy { shared.generation_version().increment(); self.table.clear_no_drop(); - self.heap.clear(); + self.heap.drain_into(&mut self.pending_drops); self.currsize = 0; } @@ -473,6 +490,7 @@ impl PolicyExt for VTTLPolicy { table, heap, currsize: self.currsize, + pending_drops: Vec::new(), } } fn build_pickle( diff --git a/src/policies/wrapped.rs b/src/policies/wrapped.rs index c4a5b32..ccd1267 100644 --- a/src/policies/wrapped.rs +++ b/src/policies/wrapped.rs @@ -1,3 +1,5 @@ +use std::collections::VecDeque; + use pyo3::types::PyAnyMethods; use pyo3::types::PyTupleMethods; @@ -30,6 +32,70 @@ pub struct Wrapped { inner: parking_lot::Mutex

, } +/// A lock guard that defers the destruction of removed values. +/// +/// Handles removed by internal operations are parked in the policy's +/// [`PolicyExt::pending_drops`] buffer. When this guard goes out of scope it +/// first releases the mutex and only then drops those handles, so Python code +/// running in a value's ``__del__`` never executes while the lock is held. +pub struct PolicyGuard<'a, P: PolicyExt> { + guard: std::mem::ManuallyDrop>, +} + +impl<'a, P: PolicyExt> PolicyGuard<'a, P> { + #[inline(always)] + fn new(guard: parking_lot::MutexGuard<'a, P>) -> Self { + Self { + guard: std::mem::ManuallyDrop::new(guard), + } + } +} + +impl<'a, P: PolicyExt> std::ops::Deref for PolicyGuard<'a, P> { + type Target = P; + + #[inline(always)] + fn deref(&self) -> &P { + &self.guard + } +} + +impl<'a, P: PolicyExt> std::ops::DerefMut for PolicyGuard<'a, P> { + #[inline(always)] + fn deref_mut(&mut self) -> &mut P { + &mut self.guard + } +} + +impl<'a, P: PolicyExt> Drop for PolicyGuard<'a, P> { + #[inline] + fn drop(&mut self) { + let buffer = self.guard.pending_drops(); + + if buffer.is_empty() { + // SAFETY: the guard is dropped exactly once per branch. + unsafe { std::mem::ManuallyDrop::drop(&mut self.guard) }; + return; + } + + if buffer.len() == 1 { + // `pop` keeps the buffer's allocation for the next eviction. + let handle = buffer.pop(); + // SAFETY: the guard is dropped exactly once per branch. + unsafe { std::mem::ManuallyDrop::drop(&mut self.guard) }; + // The lock is released: the destructor may run Python code. + drop(handle); + return; + } + + let pending = std::mem::take(buffer); + // SAFETY: the guard is dropped exactly once per branch. + unsafe { std::mem::ManuallyDrop::drop(&mut self.guard) }; + // The lock is released: the destructors may run Python code. + drop(pending); + } +} + impl Wrapped

{ /// Wraps an existing policy alongside its shared (lock-free) data. pub fn new(policy: P, shared: P::Shared) -> Self { @@ -50,33 +116,74 @@ impl Wrapped

{ /// # Panics /// Panics if the mutex is poisoned. #[inline(always)] - pub fn policy(&self) -> parking_lot::MutexGuard<'_, P> { - self.inner.lock() + pub fn policy(&self) -> PolicyGuard<'_, P> { + PolicyGuard::new(self.inner.lock()) + } + + /// Acquires the mutex only if it is free, returning `None` otherwise. + /// + /// For callers that must never wait for the lock, such as `__traverse__`: + /// the thread holding the lock may be running Python code, and a garbage + /// collection pass landing there would deadlock the whole process. + #[inline(always)] + pub fn try_policy(&self) -> Option> { + self.inner.try_lock().map(PolicyGuard::new) } } #[inline(always)] fn insert_inner( - lock: &mut parking_lot::MutexGuard<'_, P>, + lock: &mut PolicyGuard<'_, P>, shared: &P::Shared, py: pyo3::Python<'_>, handle: P::Handle, ) -> pyo3::PyResult> { + match insert_attempt(lock, shared, py, handle) { + Ok(result) => Ok(result), + Err((err, rejected)) => { + if let Some(handle) = rejected { + lock.pending_drops().push(handle); + } + Err(err) + } + } +} + +/// The locked part of [`insert_inner`]. On failure the handle that did not +/// make it into the cache is handed back, so the caller can park it. +#[inline(always)] +fn insert_attempt( + lock: &mut PolicyGuard<'_, P>, + shared: &P::Shared, + py: pyo3::Python<'_>, + handle: P::Handle, +) -> Result, (pyo3::PyErr, Option)> { let handle_size = handle.size(); if handle_size > shared.maxsize() { - return Err(new_py_error!( + let err = new_py_error!( PyOverflowError, "handle size is more than the configured maximum size" - )); + ); + return Err((err, Some(handle))); } - let result = match lock.entry(py, handle.key(), shared)? { - PolicyEntry::Occupied(occupied) => Some(occupied.replace(handle)), - PolicyEntry::Vacant(mut vacant) => { + let mut result = match lock.entry(py, handle.key(), shared) { + Err(err) => return Err((err, Some(handle))), + Ok(PolicyEntry::Occupied(occupied)) => Some(occupied.replace(handle)), + Ok(PolicyEntry::Vacant(mut vacant)) => { // Evict if need + let mut eviction_failed = None; while vacant.would_exceed(handle_size) { - vacant.evict()?; + if let Err(err) = vacant.evict() { + eviction_failed = Some(err); + break; + } + } + + if let Some(err) = eviction_failed { + drop(vacant); + return Err((err, Some(handle))); } vacant.insert(handle); @@ -87,7 +194,10 @@ fn insert_inner( if result.is_some() { // For the `PolicyEntry::Occupied` case, evict after replacement while lock.current_size() > shared.maxsize() { - lock.evict(shared)?; + match lock.evict(shared) { + Ok(evicted) => lock.pending_drops().push(evicted), + Err(err) => return Err((err, result.take())), + } } } @@ -99,7 +209,7 @@ impl Wrapped

{ /// Returns the remaining size. Equals to `maxsize - current_size`. #[inline] pub fn remaining_size(&self) -> usize { - let policy = self.inner.lock(); + let policy = self.policy(); self.shared.maxsize().saturating_sub(policy.current_size()) } @@ -110,7 +220,7 @@ impl Wrapped

{ py: pyo3::Python<'_>, key: &::Key, ) -> pyo3::PyResult { - let mut lock = self.inner.lock(); + let mut lock = self.policy(); let handle = lock.get(py, key, &self.shared)?; Ok(handle.is_some()) @@ -124,7 +234,7 @@ impl Wrapped

{ #[inline] pub fn insert_no_lock( &self, - policy: &mut parking_lot::MutexGuard<'_, P>, + policy: &mut PolicyGuard<'_, P>, py: pyo3::Python<'_>, handle: P::Handle, ) -> pyo3::PyResult> { @@ -138,7 +248,7 @@ impl Wrapped

{ py: pyo3::Python<'_>, handle: P::Handle, ) -> pyo3::PyResult> { - let mut lock = self.inner.lock(); + let mut lock = self.policy(); self.insert_no_lock(&mut lock, py, handle) } @@ -150,7 +260,7 @@ impl Wrapped

{ py: pyo3::Python<'_>, key: &::Key, ) -> pyo3::PyResult> { - let mut lock = self.inner.lock(); + let mut lock = self.policy(); let entry = lock.entry(py, key, &self.shared)?; match entry { @@ -184,7 +294,16 @@ impl Wrapped

{ use pyo3::types::PyAnyMethods; use pyo3::types::PyDictMethods; - let mut lock = self.inner.lock(); + /// How many items are transformed before the lock is taken once for + /// all of them. This bounds the transient memory of an update, and + /// an unbounded iterable stays streaming. + const BATCH_SIZE: usize = 1024; + + let py = iterable.py(); + + // The iterable, the extraction and `transform` (with `getsizeof` + // inside) are Python; they run before the lock is taken. + let mut batch: VecDeque = VecDeque::new(); // Using [pyo3::ffi::PyObject_TypeCheck] and [Bound::cast_unchecked] is so faster than [Bound::cast] let is_dictionary = unsafe { @@ -193,36 +312,87 @@ impl Wrapped

{ if is_dictionary { let dict = unsafe { iterable.cast_unchecked::() }; + batch.reserve(BATCH_SIZE.min(dict.len())); for pair in dict.items() { let (key, value) = unsafe { pair.extract::<(alias::PyObject, alias::PyObject)>() .unwrap_unchecked() }; - insert_inner(&mut lock, &self.shared, pair.py(), transform(key, value)?)?; + batch.push_back(transform(key, value)?); + if batch.len() == BATCH_SIZE { + self.insert_batch(py, &mut batch)?; + } } + } else { + // By this we will support everything has `.items()` attribute, + // including our cache classes + let items_iterable = { + if let Some(items_attribute) = iterable.getattr_opt(c"items")? { + items_attribute.call0()? + } else { + iterable + } + }; - return Ok(()); - } + let hint = unsafe { pyo3::ffi::PyObject_LengthHint(items_iterable.as_ptr(), 0) }; + if hint < 0 { + return Err(pyo3::PyErr::fetch(py)); + } + batch.reserve(BATCH_SIZE.min(hint as usize)); + + for pair in items_iterable.try_iter()? { + let pair = pair?; + let (key, value) = pair.extract::<(alias::PyObject, alias::PyObject)>()?; - // By this we will support everything has `.items()` attribute, - // including our cache classes - let items_iterable = { - if let Some(items_attribute) = iterable.getattr_opt(c"items")? { - items_attribute.call0()? - } else { - iterable + batch.push_back(transform(key, value)?); + if batch.len() == BATCH_SIZE { + self.insert_batch(py, &mut batch)?; + } } - }; + } - for pair in items_iterable.try_iter()? { - let pair = pair?; - let (key, value) = pair.extract::<(alias::PyObject, alias::PyObject)>()?; + self.insert_batch(py, &mut batch) + } - insert_inner(&mut lock, &self.shared, pair.py(), transform(key, value)?)?; + /// Takes the lock once and inserts every buffered handle. Replaced + /// handles reuse the space opened at the front of the batch; on an error, + /// its unprocessed tail stays there too. The buffer is cleared only after + /// the lock guard is gone, so none of them is destroyed under the lock. + fn insert_batch( + &self, + py: pyo3::Python<'_>, + batch: &mut VecDeque, + ) -> pyo3::PyResult<()> { + if batch.is_empty() { + return Ok(()); } - Ok(()) + let count = batch.len(); + let result = { + let mut lock = self.policy(); + let mut result = Ok(()); + + for _ in 0..count { + let handle = batch.pop_front().unwrap(); + + match insert_inner(&mut lock, &self.shared, py, handle) { + // Reuse the batch allocation as the deferred-drop buffer. + Ok(Some(old)) => batch.push_back(old), + Ok(None) => {} + Err(err) => { + result = Err(err); + break; + } + } + } + + result + }; + + // The lock is gone: clearing may run Python finalizers. + batch.clear(); + result } /// Calls the `evict()` `n` times and returns count of removed items. @@ -236,12 +406,15 @@ impl Wrapped

{ return Ok(0); } - let mut lock = self.inner.lock(); + let mut lock = self.policy(); + + let expected = (n as usize).min(lock.len()); + lock.pending_drops().reserve(expected); let mut count: pyo3::ffi::Py_ssize_t = 0; while count < n { match lock.evict(&self.shared) { - Ok(_) => {} + Ok(evicted) => lock.pending_drops().push(evicted), Err(err) => { if !err.is_instance_of::(py) { return Err(err); @@ -260,7 +433,7 @@ impl Wrapped

{ #[inline] pub fn clone_ref(&self, py: pyo3::Python) -> Self { let shared = self.shared.clone_ref(py); - let policy = self.inner.lock().clone_ref(py); + let policy = self.policy().clone_ref(py); Self { shared, @@ -279,7 +452,7 @@ impl Wrapped

{ .push(self.shared.global_ttl())?; let mut tuple = builder.begin_tuple(P::PICKLE_SIZE)?; - self.inner.lock().build_pickle(&mut tuple)?; + self.policy().build_pickle(&mut tuple)?; tuple.end()?; Ok(builder.finish()) diff --git a/src/pyclasses/cache.rs b/src/pyclasses/cache.rs index 36e1eba..9e8dc3c 100644 --- a/src/pyclasses/cache.rs +++ b/src/pyclasses/cache.rs @@ -318,6 +318,10 @@ impl PyCache { /// /// Use `setdefault_with`, if computing the value is expensive or has side /// effects. + /// + /// `getsizeof` runs with the internal lock released; if another thread + /// inserts the key meanwhile, that value wins, and if the losing + /// `getsizeof` raises, its exception still propagates to this caller. #[pyo3(signature = (key, default=utils::OptionalArgument::Undefined))] fn setdefault( &self, @@ -332,10 +336,12 @@ impl PyCache { let inner = self.0.get(); let shared = inner.shared(); - let mut policy = inner.policy(); + { + let mut policy = inner.policy(); - if let Some(x) = policy.get(py, &key, inner.shared())? { - return Ok(x.value().clone_ref(py)); + if let Some(x) = policy.get(py, &key, inner.shared())? { + return Ok(x.value().clone_ref(py)); + } } let default_object = match default { @@ -352,7 +358,22 @@ impl PyCache { key, default_object.clone_ref(py), )?; - inner.insert_no_lock(&mut policy, py, handle)?; + + let mut policy = inner.policy(); + + let existing = policy + .get(py, handle.key(), inner.shared())? + .map(|x| x.value().clone_ref(py)); + if let Some(existing) = existing { + // Lost the race: the winner's value is returned, ours is parked. + policy.pending_drops().push(handle); + return Ok(existing); + } + + if let Some(old) = inner.insert_no_lock(&mut policy, py, handle)? { + // Only reachable when the key's __eq__ is inconsistent. + policy.pending_drops().push(old); + } Ok(default_object) } @@ -362,10 +383,11 @@ impl PyCache { /// Otherwise `factory` is called with the internal lock released, its /// result is inserted and returned. /// - /// Warning: if two threads miss the same key at once, `factory` can run - /// more than once; the value inserted first wins and is returned to - /// both. If `factory` raises, nothing is inserted and the exception - /// propagates. + /// Warning: if two threads miss the same key at once, `factory` (and + /// `getsizeof`) can run more than once; the value inserted first wins and + /// is returned to both callers that succeed. If the losing call's + /// `factory` or `getsizeof` raises, nothing more is inserted and the + /// exception still propagates to that caller. fn setdefault_with( &self, py: pyo3::Python, @@ -391,19 +413,28 @@ impl PyCache { // `factory` is Python code: a GC pass inside it would deadlock on `__traverse__` let default_object = factory.call0(py)?; - let mut policy = inner.policy(); - - if let Some(x) = policy.get(py, &key, inner.shared())? { - return Ok(x.value().clone_ref(py)); - } - let handle = nopolicy::Handle::with_precomputed_hash_key( py, shared.getsizeof(), key, default_object.clone_ref(py), )?; - inner.insert_no_lock(&mut policy, py, handle)?; + + let mut policy = inner.policy(); + + let existing = policy + .get(py, handle.key(), inner.shared())? + .map(|x| x.value().clone_ref(py)); + if let Some(existing) = existing { + // Lost the race: the winner's value is returned, ours is parked. + policy.pending_drops().push(handle); + return Ok(existing); + } + + if let Some(old) = inner.insert_no_lock(&mut policy, py, handle)? { + // Only reachable when the key's __eq__ is inconsistent. + policy.pending_drops().push(old); + } Ok(default_object) } @@ -651,7 +682,12 @@ impl PyCache { } let inner = self.0.get(); - let policy = inner.policy(); + // Never wait here: the lock holder may be running Python code, and a + // collection landing there would deadlock. Skipping a pass only keeps + // the contents alive until the next one. + let Some(policy) = inner.try_policy() else { + return Ok(()); + }; for handle_ref in unsafe { policy.table().iter() } { let handle = unsafe { handle_ref.as_ref() }; diff --git a/src/pyclasses/fifocache.rs b/src/pyclasses/fifocache.rs index 75056c2..512af1a 100644 --- a/src/pyclasses/fifocache.rs +++ b/src/pyclasses/fifocache.rs @@ -324,6 +324,10 @@ impl PyFIFOCache { /// /// Use `setdefault_with`, if computing the value is expensive or has side /// effectes. + /// + /// `getsizeof` runs with the internal lock released; if another thread + /// inserts the key meanwhile, that value wins, and if the losing + /// `getsizeof` raises, its exception still propagates to this caller. #[pyo3(signature = (key, default=utils::OptionalArgument::Undefined))] fn setdefault( &self, @@ -338,10 +342,12 @@ impl PyFIFOCache { let inner = self.0.get(); let shared = inner.shared(); - let mut policy = inner.policy(); + { + let mut policy = inner.policy(); - if let Some(x) = policy.get(py, &key, inner.shared())? { - return Ok(x.value().clone_ref(py)); + if let Some(x) = policy.get(py, &key, inner.shared())? { + return Ok(x.value().clone_ref(py)); + } } let default_object = match default { @@ -358,7 +364,22 @@ impl PyFIFOCache { key, default_object.clone_ref(py), )?; - inner.insert_no_lock(&mut policy, py, handle)?; + + let mut policy = inner.policy(); + + let existing = policy + .get(py, handle.key(), inner.shared())? + .map(|x| x.value().clone_ref(py)); + if let Some(existing) = existing { + // Lost the race: the winner's value is returned, ours is parked. + policy.pending_drops().push(handle); + return Ok(existing); + } + + if let Some(old) = inner.insert_no_lock(&mut policy, py, handle)? { + // Only reachable when the key's __eq__ is inconsistent. + policy.pending_drops().push(old); + } Ok(default_object) } @@ -368,10 +389,11 @@ impl PyFIFOCache { /// Otherwise `factory` is called with the internal lock released, its /// result is inserted and returned. /// - /// Warning: if two threads miss the same key at once, `factory` can run - /// more than once; the value inserted first wins and is returned to - /// both. If `factory` raises, nothing is inserted and the exception - /// propagates. + /// Warning: if two threads miss the same key at once, `factory` (and + /// `getsizeof`) can run more than once; the value inserted first wins and + /// is returned to both callers that succeed. If the losing call's + /// `factory` or `getsizeof` raises, nothing more is inserted and the + /// exception still propagates to that caller. fn setdefault_with( &self, py: pyo3::Python, @@ -397,19 +419,28 @@ impl PyFIFOCache { // `factory` is Python code: a GC pass inside it would deadlock on `__traverse__` let default_object = factory.call0(py)?; - let mut policy = inner.policy(); - - if let Some(x) = policy.get(py, &key, inner.shared())? { - return Ok(x.value().clone_ref(py)); - } - let handle = fifopolicy::Handle::with_precomputed_hash_key( py, shared.getsizeof(), key, default_object.clone_ref(py), )?; - inner.insert_no_lock(&mut policy, py, handle)?; + + let mut policy = inner.policy(); + + let existing = policy + .get(py, handle.key(), inner.shared())? + .map(|x| x.value().clone_ref(py)); + if let Some(existing) = existing { + // Lost the race: the winner's value is returned, ours is parked. + policy.pending_drops().push(handle); + return Ok(existing); + } + + if let Some(old) = inner.insert_no_lock(&mut policy, py, handle)? { + // Only reachable when the key's __eq__ is inconsistent. + policy.pending_drops().push(old); + } Ok(default_object) } @@ -679,7 +710,12 @@ impl PyFIFOCache { } let inner = self.0.get(); - let policy = inner.policy(); + // Never wait here: the lock holder may be running Python code, and a + // collection landing there would deadlock. Skipping a pass only keeps + // the contents alive until the next one. + let Some(policy) = inner.try_policy() else { + return Ok(()); + }; for handle in policy.entries().iter() { visit.call(handle.key().as_ref())?; diff --git a/src/pyclasses/lfucache.rs b/src/pyclasses/lfucache.rs index 663a5d1..54d53ac 100644 --- a/src/pyclasses/lfucache.rs +++ b/src/pyclasses/lfucache.rs @@ -342,6 +342,10 @@ impl PyLFUCache { /// /// Use `setdefault_with`, if computing the value is expensive or has side /// effectes. + /// + /// `getsizeof` runs with the internal lock released; if another thread + /// inserts the key meanwhile, that value wins, and if the losing + /// `getsizeof` raises, its exception still propagates to this caller. #[pyo3(signature = (key, default=utils::OptionalArgument::Undefined))] fn setdefault( &self, @@ -356,10 +360,12 @@ impl PyLFUCache { let inner = self.0.get(); let shared = inner.shared(); - let mut policy = inner.policy(); + { + let mut policy = inner.policy(); - if let Some(x) = policy.get(py, &key, inner.shared())? { - return Ok(x.value().clone_ref(py)); + if let Some(x) = policy.get(py, &key, inner.shared())? { + return Ok(x.value().clone_ref(py)); + } } let default_object = match default { @@ -377,7 +383,22 @@ impl PyLFUCache { default_object.clone_ref(py), 1, )?; - inner.insert_no_lock(&mut policy, py, handle)?; + + let mut policy = inner.policy(); + + let existing = policy + .get(py, handle.key(), inner.shared())? + .map(|x| x.value().clone_ref(py)); + if let Some(existing) = existing { + // Lost the race: the winner's value is returned, ours is parked. + policy.pending_drops().push(handle); + return Ok(existing); + } + + if let Some(old) = inner.insert_no_lock(&mut policy, py, handle)? { + // Only reachable when the key's __eq__ is inconsistent. + policy.pending_drops().push(old); + } Ok(default_object) } @@ -387,10 +408,11 @@ impl PyLFUCache { /// Otherwise, `factory` is called with the internal lock released, its /// result is inserted and returned. /// - /// Warning: if two threads miss the same key at once, `factory` can run - /// more than once; the value inserted first wins and is returned to - /// both. If `factory` raises, nothing is inserted and the exception - /// propagates. + /// Warning: if two threads miss the same key at once, `factory` (and + /// `getsizeof`) can run more than once; the value inserted first wins and + /// is returned to both callers that succeed. If the losing call's + /// `factory` or `getsizeof` raises, nothing more is inserted and the + /// exception still propagates to that caller. fn setdefault_with( &self, py: pyo3::Python, @@ -416,12 +438,6 @@ impl PyLFUCache { // `factory` is Python code: a GC pass inside it would deadlock on `__traverse__` let default_object = factory.call0(py)?; - let mut policy = inner.policy(); - - if let Some(x) = policy.get(py, &key, inner.shared())? { - return Ok(x.value().clone_ref(py)); - } - let handle = lfupolicy::FrequencyHandle::with_precomputed_hash_key( py, shared.getsizeof(), @@ -429,7 +445,22 @@ impl PyLFUCache { default_object.clone_ref(py), 1, )?; - inner.insert_no_lock(&mut policy, py, handle)?; + + let mut policy = inner.policy(); + + let existing = policy + .get(py, handle.key(), inner.shared())? + .map(|x| x.value().clone_ref(py)); + if let Some(existing) = existing { + // Lost the race: the winner's value is returned, ours is parked. + policy.pending_drops().push(handle); + return Ok(existing); + } + + if let Some(old) = inner.insert_no_lock(&mut policy, py, handle)? { + // Only reachable when the key's __eq__ is inconsistent. + policy.pending_drops().push(old); + } Ok(default_object) } @@ -744,7 +775,12 @@ impl PyLFUCache { } let inner = self.0.get(); - let policy = inner.policy(); + // Never wait here: the lock holder may be running Python code, and a + // collection landing there would deadlock. Skipping a pass only keeps + // the contents alive until the next one. + let Some(policy) = inner.try_policy() else { + return Ok(()); + }; for cursor in unsafe { policy.table().iter() } { let handle = unsafe { cursor.as_ref().element() }; diff --git a/src/pyclasses/lrucache.rs b/src/pyclasses/lrucache.rs index 1954e8c..4e71e21 100644 --- a/src/pyclasses/lrucache.rs +++ b/src/pyclasses/lrucache.rs @@ -351,6 +351,10 @@ impl PyLRUCache { /// /// Use `setdefault_with`, if computing the value is expensive or has side /// effectes. + /// + /// `getsizeof` runs with the internal lock released; if another thread + /// inserts the key meanwhile, that value wins, and if the losing + /// `getsizeof` raises, its exception still propagates to this caller. #[pyo3(signature = (key, default=utils::OptionalArgument::Undefined))] fn setdefault( &self, @@ -365,10 +369,12 @@ impl PyLRUCache { let inner = self.0.get(); let shared = inner.shared(); - let mut policy = inner.policy(); + { + let mut policy = inner.policy(); - if let Some(x) = policy.get(py, &key, inner.shared())? { - return Ok(x.value().clone_ref(py)); + if let Some(x) = policy.get(py, &key, inner.shared())? { + return Ok(x.value().clone_ref(py)); + } } let default_object = match default { @@ -385,7 +391,22 @@ impl PyLRUCache { key, default_object.clone_ref(py), )?; - inner.insert_no_lock(&mut policy, py, handle)?; + + let mut policy = inner.policy(); + + let existing = policy + .get(py, handle.key(), inner.shared())? + .map(|x| x.value().clone_ref(py)); + if let Some(existing) = existing { + // Lost the race: the winner's value is returned, ours is parked. + policy.pending_drops().push(handle); + return Ok(existing); + } + + if let Some(old) = inner.insert_no_lock(&mut policy, py, handle)? { + // Only reachable when the key's __eq__ is inconsistent. + policy.pending_drops().push(old); + } Ok(default_object) } @@ -395,10 +416,11 @@ impl PyLRUCache { /// Otherwise `factory` is called with the internal lock released, its /// result is inserted and returned. /// - /// Warning: if two threads miss the same key at once, `factory` can run - /// more than once; the value inserted first wins and is returned to - /// both. If `factory` raises, nothing is inserted and the exception - /// propagates. + /// Warning: if two threads miss the same key at once, `factory` (and + /// `getsizeof`) can run more than once; the value inserted first wins and + /// is returned to both callers that succeed. If the losing call's + /// `factory` or `getsizeof` raises, nothing more is inserted and the + /// exception still propagates to that caller. fn setdefault_with( &self, py: pyo3::Python, @@ -424,19 +446,28 @@ impl PyLRUCache { // `factory` is Python code: a GC pass inside it would deadlock on `__traverse__` let default_object = factory.call0(py)?; - let mut policy = inner.policy(); - - if let Some(x) = policy.get(py, &key, inner.shared())? { - return Ok(x.value().clone_ref(py)); - } - let handle = lrupolicy::Handle::with_precomputed_hash_key( py, shared.getsizeof(), key, default_object.clone_ref(py), )?; - inner.insert_no_lock(&mut policy, py, handle)?; + + let mut policy = inner.policy(); + + let existing = policy + .get(py, handle.key(), inner.shared())? + .map(|x| x.value().clone_ref(py)); + if let Some(existing) = existing { + // Lost the race: the winner's value is returned, ours is parked. + policy.pending_drops().push(handle); + return Ok(existing); + } + + if let Some(old) = inner.insert_no_lock(&mut policy, py, handle)? { + // Only reachable when the key's __eq__ is inconsistent. + policy.pending_drops().push(old); + } Ok(default_object) } @@ -725,7 +756,12 @@ impl PyLRUCache { } let inner = self.0.get(); - let policy = inner.policy(); + // Never wait here: the lock holder may be running Python code, and a + // collection landing there would deadlock. Skipping a pass only keeps + // the contents alive until the next one. + let Some(policy) = inner.try_policy() else { + return Ok(()); + }; for cursor in unsafe { policy.list().iter() } { let handle = unsafe { cursor.element() }; diff --git a/src/pyclasses/rrcache.rs b/src/pyclasses/rrcache.rs index 93b5eb7..0a6d5c5 100644 --- a/src/pyclasses/rrcache.rs +++ b/src/pyclasses/rrcache.rs @@ -322,6 +322,10 @@ impl PyRRCache { /// /// Use `setdefault_with`, if computing the value is expensive or has side /// effectes. + /// + /// `getsizeof` runs with the internal lock released; if another thread + /// inserts the key meanwhile, that value wins, and if the losing + /// `getsizeof` raises, its exception still propagates to this caller. #[pyo3(signature = (key, default=utils::OptionalArgument::Undefined))] fn setdefault( &self, @@ -336,10 +340,12 @@ impl PyRRCache { let inner = self.0.get(); let shared = inner.shared(); - let mut policy = inner.policy(); + { + let mut policy = inner.policy(); - if let Some(x) = policy.get(py, &key, inner.shared())? { - return Ok(x.value().clone_ref(py)); + if let Some(x) = policy.get(py, &key, inner.shared())? { + return Ok(x.value().clone_ref(py)); + } } let default_object = match default { @@ -356,7 +362,22 @@ impl PyRRCache { key, default_object.clone_ref(py), )?; - inner.insert_no_lock(&mut policy, py, handle)?; + + let mut policy = inner.policy(); + + let existing = policy + .get(py, handle.key(), inner.shared())? + .map(|x| x.value().clone_ref(py)); + if let Some(existing) = existing { + // Lost the race: the winner's value is returned, ours is parked. + policy.pending_drops().push(handle); + return Ok(existing); + } + + if let Some(old) = inner.insert_no_lock(&mut policy, py, handle)? { + // Only reachable when the key's __eq__ is inconsistent. + policy.pending_drops().push(old); + } Ok(default_object) } @@ -366,10 +387,11 @@ impl PyRRCache { /// Otherwise `factory` is called with the internal lock released, its /// result is inserted and returned. /// - /// Warning: if two threads miss the same key at once, `factory` can run - /// more than once; the value inserted first wins and is returned to - /// both. If `factory` raises, nothing is inserted and the exception - /// propagates. + /// Warning: if two threads miss the same key at once, `factory` (and + /// `getsizeof`) can run more than once; the value inserted first wins and + /// is returned to both callers that succeed. If the losing call's + /// `factory` or `getsizeof` raises, nothing more is inserted and the + /// exception still propagates to that caller. fn setdefault_with( &self, py: pyo3::Python, @@ -395,19 +417,28 @@ impl PyRRCache { // `factory` is Python code: a GC pass inside it would deadlock on `__traverse__` let default_object = factory.call0(py)?; - let mut policy = inner.policy(); - - if let Some(x) = policy.get(py, &key, inner.shared())? { - return Ok(x.value().clone_ref(py)); - } - let handle = rrpolicy::Handle::with_precomputed_hash_key( py, shared.getsizeof(), key, default_object.clone_ref(py), )?; - inner.insert_no_lock(&mut policy, py, handle)?; + + let mut policy = inner.policy(); + + let existing = policy + .get(py, handle.key(), inner.shared())? + .map(|x| x.value().clone_ref(py)); + if let Some(existing) = existing { + // Lost the race: the winner's value is returned, ours is parked. + policy.pending_drops().push(handle); + return Ok(existing); + } + + if let Some(old) = inner.insert_no_lock(&mut policy, py, handle)? { + // Only reachable when the key's __eq__ is inconsistent. + policy.pending_drops().push(old); + } Ok(default_object) } @@ -671,7 +702,12 @@ impl PyRRCache { } let inner = self.0.get(); - let policy = inner.policy(); + // Never wait here: the lock holder may be running Python code, and a + // collection landing there would deadlock. Skipping a pass only keeps + // the contents alive until the next one. + let Some(policy) = inner.try_policy() else { + return Ok(()); + }; for handle_ref in unsafe { policy.table().iter() } { let handle = unsafe { handle_ref.as_ref() }; diff --git a/src/pyclasses/ttlcache.rs b/src/pyclasses/ttlcache.rs index 7a56372..c78703a 100644 --- a/src/pyclasses/ttlcache.rs +++ b/src/pyclasses/ttlcache.rs @@ -327,6 +327,10 @@ impl PyTTLCache { /// /// Use `setdefault_with`, if computing the value is expensive or has side /// effects. + /// + /// `getsizeof` runs with the internal lock released; if another thread + /// inserts the key meanwhile, that value wins, and if the losing + /// `getsizeof` raises, its exception still propagates to this caller. #[pyo3(signature = (key, default=utils::OptionalArgument::Undefined))] fn setdefault( &self, @@ -341,10 +345,12 @@ impl PyTTLCache { let inner = self.0.get(); let shared = inner.shared(); - let mut policy = inner.policy(); + { + let mut policy = inner.policy(); - if let Some(x) = policy.get(py, &key, inner.shared())? { - return Ok(x.value().clone_ref(py)); + if let Some(x) = policy.get(py, &key, inner.shared())? { + return Ok(x.value().clone_ref(py)); + } } let default_object = match default { @@ -362,7 +368,22 @@ impl PyTTLCache { key, default_object.clone_ref(py), )?; - inner.insert_no_lock(&mut policy, py, handle)?; + + let mut policy = inner.policy(); + + let existing = policy + .get(py, handle.key(), inner.shared())? + .map(|x| x.value().clone_ref(py)); + if let Some(existing) = existing { + // Lost the race: the winner's value is returned, ours is parked. + policy.pending_drops().push(handle); + return Ok(existing); + } + + if let Some(old) = inner.insert_no_lock(&mut policy, py, handle)? { + // Only reachable when the key's __eq__ is inconsistent. + policy.pending_drops().push(old); + } Ok(default_object) } @@ -372,10 +393,11 @@ impl PyTTLCache { /// Otherwise, `factory` is called with the internal lock released, its /// result is inserted and returned. /// - /// Warning: if two threads miss the same key at once, `factory` can run - /// more than once; the value inserted first wins and is returned to - /// both. If `factory` raises, nothing is inserted and the exception - /// propagates. + /// Warning: if two threads miss the same key at once, `factory` (and + /// `getsizeof`) can run more than once; the value inserted first wins and + /// is returned to both callers that succeed. If the losing call's + /// `factory` or `getsizeof` raises, nothing more is inserted and the + /// exception still propagates to that caller. fn setdefault_with( &self, py: pyo3::Python, @@ -401,12 +423,6 @@ impl PyTTLCache { // `factory` is Python code: a GC pass inside it would deadlock on `__traverse__` let default_object = factory.call0(py)?; - let mut policy = inner.policy(); - - if let Some(x) = policy.get(py, &key, inner.shared())? { - return Ok(x.value().clone_ref(py)); - } - let handle = ttlpolicy::ExpiringHandle::with_precomputed_hash_key( py, shared.getsizeof(), @@ -414,7 +430,22 @@ impl PyTTLCache { key, default_object.clone_ref(py), )?; - inner.insert_no_lock(&mut policy, py, handle)?; + + let mut policy = inner.policy(); + + let existing = policy + .get(py, handle.key(), inner.shared())? + .map(|x| x.value().clone_ref(py)); + if let Some(existing) = existing { + // Lost the race: the winner's value is returned, ours is parked. + policy.pending_drops().push(handle); + return Ok(existing); + } + + if let Some(old) = inner.insert_no_lock(&mut policy, py, handle)? { + // Only reachable when the key's __eq__ is inconsistent. + policy.pending_drops().push(old); + } Ok(default_object) } @@ -816,7 +847,12 @@ impl PyTTLCache { } let inner = self.0.get(); - let policy = inner.policy(); + // Never wait here: the lock holder may be running Python code, and a + // collection landing there would deadlock. Skipping a pass only keeps + // the contents alive until the next one. + let Some(policy) = inner.try_policy() else { + return Ok(()); + }; for handle in policy.entries().iter() { visit.call(handle.key().as_ref())?; diff --git a/src/pyclasses/vttlcache.rs b/src/pyclasses/vttlcache.rs index 060dc8c..5553e87 100644 --- a/src/pyclasses/vttlcache.rs +++ b/src/pyclasses/vttlcache.rs @@ -303,6 +303,14 @@ impl PyVTTLCache { } } + /// Get `key`s value, or automatically insert `default` and return it. + /// + /// If `key` exists, its current value is returned and `default` is ignored. + /// Otherwise `default` is inserted for `key` (with `ttl`) and returned. + /// + /// `getsizeof` runs with the internal lock released; if another thread + /// inserts the key meanwhile, that value wins, and if the losing + /// `getsizeof` raises, its exception still propagates to this caller. #[pyo3(signature = (key, default=utils::OptionalArgument::Undefined, ttl=None))] fn setdefault( &self, @@ -322,10 +330,12 @@ impl PyVTTLCache { let inner = self.0.get(); let shared = inner.shared(); - let mut policy = inner.policy(); + { + let mut policy = inner.policy(); - if let Some(x) = policy.get(py, &key, inner.shared())? { - return Ok(x.value().clone_ref(py)); + if let Some(x) = policy.get(py, &key, inner.shared())? { + return Ok(x.value().clone_ref(py)); + } } let default_object = match default { @@ -343,10 +353,36 @@ impl PyVTTLCache { key, default_object.clone_ref(py), )?; - inner.insert_no_lock(&mut policy, py, handle)?; + + let mut policy = inner.policy(); + + let existing = policy + .get(py, handle.key(), inner.shared())? + .map(|x| x.value().clone_ref(py)); + if let Some(existing) = existing { + // Lost the race: the winner's value is returned, ours is parked. + policy.pending_drops().push(handle); + return Ok(existing); + } + + if let Some(old) = inner.insert_no_lock(&mut policy, py, handle)? { + // Only reachable when the key's __eq__ is inconsistent. + policy.pending_drops().push(old); + } Ok(default_object) } + /// Get `key`s value, or automatically create and insert one via `factory`. + /// + /// If `key` exists, its current value is returned and `factory` is not called. + /// Otherwise `factory` is called with the internal lock released, its + /// result is inserted (with `ttl`) and returned. + /// + /// Warning: if two threads miss the same key at once, `factory` (and + /// `getsizeof`) can run more than once; the value inserted first wins and + /// is returned to both callers that succeed. If the losing call's + /// `factory` or `getsizeof` raises, nothing more is inserted and the + /// exception still propagates to that caller. #[pyo3(signature = (key, factory, ttl=None))] fn setdefault_with( &self, @@ -378,12 +414,6 @@ impl PyVTTLCache { // `factory` is Python code: a GC pass inside it would deadlock on `__traverse__` let default_object = factory.call0(py)?; - let mut policy = inner.policy(); - - if let Some(x) = policy.get(py, &key, inner.shared())? { - return Ok(x.value().clone_ref(py)); - } - let handle = vttlpolicy::ExpiringHandle::with_precomputed_hash_key( py, shared.getsizeof(), @@ -391,7 +421,22 @@ impl PyVTTLCache { key, default_object.clone_ref(py), )?; - inner.insert_no_lock(&mut policy, py, handle)?; + + let mut policy = inner.policy(); + + let existing = policy + .get(py, handle.key(), inner.shared())? + .map(|x| x.value().clone_ref(py)); + if let Some(existing) = existing { + // Lost the race: the winner's value is returned, ours is parked. + policy.pending_drops().push(handle); + return Ok(existing); + } + + if let Some(old) = inner.insert_no_lock(&mut policy, py, handle)? { + // Only reachable when the key's __eq__ is inconsistent. + policy.pending_drops().push(old); + } Ok(default_object) } @@ -776,7 +821,12 @@ impl PyVTTLCache { } let inner = self.0.get(); - let policy = inner.policy(); + // Never wait here: the lock holder may be running Python code, and a + // collection landing there would deadlock. Skipping a pass only keeps + // the contents alive until the next one. + let Some(policy) = inner.try_policy() else { + return Ok(()); + }; for cursor in unsafe { policy.table().iter() } { let handle = unsafe { cursor.as_ref().element() }; diff --git a/tests/mixins.py b/tests/mixins.py index e135f48..07e1b03 100644 --- a/tests/mixins.py +++ b/tests/mixins.py @@ -101,6 +101,23 @@ def test_gc_traverse_clear(self): class InsertAndGetMixin(BaseMixin): + def test_key_eq_may_trigger_the_gc(self): + # __eq__ runs in the middle of a probe, with the lock held; a deadlock + # here would keep the GIL, so the call runs in a child process + name = type(self.create_cache()).__name__ + + try: + done = subprocess.run( + [sys.executable, "-c", EQ_TRIGGERING_GC, name], + capture_output=True, + text=True, + timeout=60, + ) + except subprocess.TimeoutExpired: + pytest.fail(f"{name}.get() with a colliding key never returned") + + assert done.stdout.strip() == "ok", done.stderr + def test_insert_returns_none_on_new_key(self): cache = self.create_cache() @@ -189,6 +206,22 @@ def test_popitem_updates_currsize(self): class SetDefaultMixin(BaseMixin): + def test_setdefault_getsizeof_may_touch_the_cache(self): + # a deadlock here would keep the GIL, so the call runs in a child process + name = type(self.create_cache()).__name__ + + try: + done = subprocess.run( + [sys.executable, "-c", GETSIZEOF_TOUCHING_CACHE, name, "setdefault"], + capture_output=True, + text=True, + timeout=60, + ) + except subprocess.TimeoutExpired: + pytest.fail(f"{name}.setdefault never returned") + + assert done.stdout.strip() == "ok", done.stderr + def test_setdefault_inserts_when_absent(self): cache = self.create_cache() @@ -205,6 +238,262 @@ def test_setdefault_returns_existing_value(self): assert cache.get("k") == "existing" +EQ_TRIGGERING_GC = """ +import gc +import sys + +import cachebox + +name = sys.argv[1] +cls = getattr(cachebox, name) +cache = cls(10, global_ttl=60) if name == "TTLCache" else cls(10) + + +class Key: + def __init__(self, name): + self.name = name + + def __hash__(self): + return 42 # same hash for every key, so lookups have to call __eq__ + + def __eq__(self, other): + gc.collect() + return self.name == other.name + + +cache.insert(Key("a"), 1) +assert cache.get(Key("b")) is None +print("ok") +""" + +DROPPED_VALUE_TOUCHING_CACHE = """ +import sys + +import cachebox + +name = sys.argv[1] +cls = getattr(cachebox, name) +cache = cls(2, global_ttl=60) if name == "TTLCache" else cls(2) + + +class Touchy: + def __del__(self): + cache.get("probe") + + +cache.update({"k": Touchy()}) +cache.update({"k": Touchy()}) # the replacement drops the old value + +try: + for i in range(4): + cache.insert(i, Touchy()) # evictions drop values +except OverflowError: + pass # Cache has no eviction algorithm + +try: + cache.drain(1) +except OverflowError: + pass # Cache has no eviction algorithm + +cache.clear() +print("ok") +""" + +BATCHED_UPDATE_KEEPS_FINISHED_BATCHES = """ +import sys + +import cachebox + +name = sys.argv[1] +cls = getattr(cachebox, name) +cache = cls(2000, global_ttl=60) if name == "TTLCache" else cls(2000) + + +class Touchy: + def __del__(self): + cache.get("probe") + + +def pairs(): + for i in range(1500): + yield (i, i) + yield (Touchy(),) # malformed, held only by this tuple + + +try: + cache.update(pairs()) +except ValueError: + inserted = len(cache) + if 0 < inserted < 1500: + print("ok") + else: + print(f"unexpected count {inserted}") +else: + print("the malformed item was accepted") +""" + +GETSIZEOF_FAILING_IN_UPDATE = """ +import operator +import sys + +import cachebox + +name = sys.argv[1] +cls = getattr(cachebox, name) +if name == "TTLCache": + cache = cls(5, global_ttl=60, getsizeof=operator.index) +else: + cache = cls(5, getsizeof=operator.index) + + +class Touchy: + def __del__(self): + cache.get("probe") + + +def pairs(): + # operator.index rejects two arguments with a C-level TypeError + yield ("k", Touchy()) + + +try: + cache.update(pairs()) +except TypeError: + print("ok") +else: + print("getsizeof did not fail") +""" + +MALFORMED_UPDATE_ITEM = """ +import sys + +import cachebox + +name = sys.argv[1] +cls = getattr(cachebox, name) +cache = cls(5, global_ttl=60) if name == "TTLCache" else cls(5) + + +class Touchy: + def __del__(self): + cache.get("probe") + + +def pairs(): + yield ("valid", "value") + # not a key/value pair; the tuple holds the only reference to the value + yield (Touchy(),) + + +try: + cache.update(pairs()) +except ValueError: + # the error came before any insert, so the valid pair is not in either + print("ok" if len(cache) == 0 else "the valid item leaked in") +else: + print("the malformed item was accepted") +""" + +REJECTED_UPDATE_DROPPING_VALUE = """ +import gc +import sys + +import cachebox + + +def sizeof(key, value): + return 9 if getattr(value, "tag", None) == "rejected" else 1 + + +name = sys.argv[1] +cls = getattr(cachebox, name) +if name == "TTLCache": + cache = cls(5, global_ttl=60, getsizeof=sizeof) +else: + cache = cls(5, getsizeof=sizeof) + +finalized = [] + + +class Touchy: + def __init__(self, tag): + self.tag = tag + + def __del__(self): + cache.get("probe") + finalized.append(self.tag) + + +def pairs(): + # each yielded tuple holds the only reference to its value + yield ("a", Touchy("inserted")) + yield ("b", Touchy("rejected")) # fails the size pre-check + yield ("c", Touchy("tail")) # never reaches its insert + + +try: + cache.update(pairs()) +except OverflowError: + # PyPy runs finalizers on a later collection, in no promised order + for _ in range(4): + gc.collect() + if {"rejected", "tail"} <= set(finalized): + print("ok") + else: + print("missing finalizers:", sorted(finalized)) +else: + print("the update was not rejected") +""" + +DROPPED_VALUE_TRIGGERING_GC = """ +import gc +import sys + +import cachebox + +name = sys.argv[1] +cls = getattr(cachebox, name) +cache = cls(10, global_ttl=60) if name == "TTLCache" else cls(10) + + +class Boom: + def __del__(self): + gc.collect() + + +cache.insert("k", Boom()) +cache.clear() +print("ok") +""" + +GETSIZEOF_TOUCHING_CACHE = """ +import sys + +import cachebox + + +def sizeof(key, value): + cache.get("probe") + return 1 + + +name = sys.argv[1] +variant = sys.argv[2] +cls = getattr(cachebox, name) +if name == "TTLCache": + cache = cls(10, global_ttl=60, getsizeof=sizeof) +else: + cache = cls(10, getsizeof=sizeof) + +if variant == "setdefault": + assert cache.setdefault("k", "v") == "v" + assert cache.setdefault("k", "other") == "v" +else: + assert cache.setdefault_with("k", lambda: "v") == "v" + assert cache.setdefault_with("k", lambda: "other") == "v" +print("ok") +""" + FACTORY_TOUCHING_CACHE = """ import gc import sys @@ -228,6 +517,22 @@ def factory(): class SetDefaultWithMixin(BaseMixin): + def test_setdefault_with_getsizeof_may_touch_the_cache(self): + # a deadlock here would keep the GIL, so the call runs in a child process + name = type(self.create_cache()).__name__ + + try: + done = subprocess.run( + [sys.executable, "-c", GETSIZEOF_TOUCHING_CACHE, name, "setdefault_with"], + capture_output=True, + text=True, + timeout=60, + ) + except subprocess.TimeoutExpired: + pytest.fail(f"{name}.setdefault_with never returned") + + assert done.stdout.strip() == "ok", done.stderr + def test_setdefault_with_inserts_when_absent(self): cache = self.create_cache() @@ -317,6 +622,71 @@ def test_delitem_missing_key_raises_keyerror(self): class UpdateMixin(BaseMixin): + def test_update_failing_mid_batch_keeps_the_finished_batches(self): + # a deadlock here would keep the GIL, so the call runs in a child process + name = type(self.create_cache()).__name__ + + try: + done = subprocess.run( + [sys.executable, "-c", BATCHED_UPDATE_KEEPS_FINISHED_BATCHES, name], + capture_output=True, + text=True, + timeout=60, + ) + except subprocess.TimeoutExpired: + pytest.fail(f"the failing update of {name} never returned") + + assert done.stdout.strip() == "ok", done.stderr + + def test_failing_getsizeof_in_update_may_touch_the_cache(self): + # a deadlock here would keep the GIL, so the call runs in a child process + name = type(self.create_cache()).__name__ + + try: + done = subprocess.run( + [sys.executable, "-c", GETSIZEOF_FAILING_IN_UPDATE, name], + capture_output=True, + text=True, + timeout=60, + ) + except subprocess.TimeoutExpired: + pytest.fail(f"the update of {name} with a failing getsizeof never returned") + + assert done.stdout.strip() == "ok", done.stderr + + def test_malformed_update_item_may_touch_the_cache(self): + # a deadlock here would keep the GIL, so the call runs in a child process + name = type(self.create_cache()).__name__ + + try: + done = subprocess.run( + [sys.executable, "-c", MALFORMED_UPDATE_ITEM, name], + capture_output=True, + text=True, + timeout=60, + ) + except subprocess.TimeoutExpired: + pytest.fail(f"the malformed update of {name} never returned") + + assert done.stdout.strip() == "ok", done.stderr + + def test_rejected_update_item_may_touch_the_cache(self): + # a deadlock here would keep the GIL, so the call runs in a child process + name = type(self.create_cache()).__name__ + + try: + done = subprocess.run( + [sys.executable, "-c", REJECTED_UPDATE_DROPPING_VALUE, name], + capture_output=True, + text=True, + timeout=60, + ) + except subprocess.TimeoutExpired: + pytest.fail(f"the rejected update of {name} never returned") + + # both parked values are finalized once the lock is gone, tail included + assert done.stdout.strip() == "ok", done.stdout + done.stderr + def test_update_from_dict(self): cache = self.create_cache() @@ -635,6 +1005,38 @@ def test_generation_version_on_popitem(self): class DrainClearShrinkMixin(BaseMixin): + def test_dropped_value_may_touch_the_cache(self): + # a deadlock here would keep the GIL, so the call runs in a child process + name = type(self.create_cache()).__name__ + + try: + done = subprocess.run( + [sys.executable, "-c", DROPPED_VALUE_TOUCHING_CACHE, name], + capture_output=True, + text=True, + timeout=60, + ) + except subprocess.TimeoutExpired: + pytest.fail(f"dropping a {name} value never returned") + + assert done.stdout.strip() == "ok", done.stderr + + def test_dropping_a_value_may_trigger_the_gc(self): + # a deadlock here would keep the GIL, so the call runs in a child process + name = type(self.create_cache()).__name__ + + try: + done = subprocess.run( + [sys.executable, "-c", DROPPED_VALUE_TRIGGERING_GC, name], + capture_output=True, + text=True, + timeout=60, + ) + except subprocess.TimeoutExpired: + pytest.fail(f"{name}.clear() never returned") + + assert done.stdout.strip() == "ok", done.stderr + def test_clear_removes_all_items(self): cache = self.create_cache() diff --git a/tests/test_impls.py b/tests/test_impls.py index 87b9f8a..f8231e9 100644 --- a/tests/test_impls.py +++ b/tests/test_impls.py @@ -11,6 +11,30 @@ from . import mixins +FAILED_REPLACEMENT_DROPPING_VALUE = """ +import cachebox + +cache = cachebox.Cache( + 10, getsizeof=lambda key, value: value if isinstance(value, int) else 1 +) + + +class Touchy: + def __del__(self): + cache.get("probe") + + +cache.insert("ballast", 5) +cache.insert("k", Touchy()) +try: + cache.insert("k", 9) # the replacement overflows and the eviction fails +except OverflowError: + print("ok") +else: + print("the replacement did not overflow") +""" + + class TestCache( mixins.InitializeMixin, mixins.InsertAndGetMixin, @@ -43,6 +67,20 @@ def test_popitem_overflow_error(self): with pytest.raises(OverflowError): cache.popitem() + def test_failed_replacement_still_drops_the_old_value_safely(self): + # a deadlock here would keep the GIL, so the call runs in a child process + try: + done = subprocess.run( + [sys.executable, "-c", FAILED_REPLACEMENT_DROPPING_VALUE], + capture_output=True, + text=True, + timeout=60, + ) + except subprocess.TimeoutExpired: + pytest.fail("the failing replacement never returned") + + assert done.stdout.strip() == "ok", done.stderr + def test_insert_overflow_error(self): cache = self.create_cache(5)