diff --git a/Cargo.toml b/Cargo.toml index 2d4a15b..f3700d6 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "quick_cache" -version = "0.6.24" +version = "0.7.0" edition = "2021" description = "Lightweight and high performance concurrent cache" repository = "https://github.com/arthurprs/quick-cache" diff --git a/examples/eviction_listener.rs b/examples/eviction_listener.rs index 4861993..e5e5725 100644 --- a/examples/eviction_listener.rs +++ b/examples/eviction_listener.rs @@ -7,8 +7,6 @@ struct EvictionListener(mpsc::Sender<(u64, u64)>); impl Lifecycle for EvictionListener { type RequestState = (); - fn begin_request(&self) -> Self::RequestState {} - fn on_evict(&self, _state: &mut Self::RequestState, key: u64, val: u64) { let _ = self.0.send((key, val)); } diff --git a/fuzz/Cargo.lock b/fuzz/Cargo.lock index 0f19630..3c864f2 100644 --- a/fuzz/Cargo.lock +++ b/fuzz/Cargo.lock @@ -162,7 +162,7 @@ dependencies = [ [[package]] name = "quick_cache" -version = "0.6.17" +version = "0.7.0" dependencies = [ "ahash", "equivalent", diff --git a/fuzz/fuzz_targets/fuzz_sync_cache.rs b/fuzz/fuzz_targets/fuzz_sync_cache.rs index d5765f1..16238ed 100644 --- a/fuzz/fuzz_targets/fuzz_sync_cache.rs +++ b/fuzz/fuzz_targets/fuzz_sync_cache.rs @@ -31,10 +31,6 @@ impl Weighter for MyWeighter { impl Lifecycle for MyLifecycle { type RequestState = Vec<(u16, Value)>; - fn begin_request(&self) -> Self::RequestState { - Default::default() - } - fn is_pinned(&self, _key: &u16, val: &Value) -> bool { val.pinned } @@ -104,13 +100,15 @@ fn run(input: Input) { match op { Op::Insert(k, v, pinned) => { // eprintln!("insert {k} {v}"); - let evicted = cache.insert_with_lifecycle( + let mut evicted = Vec::new(); + cache.insert_with_lifecycle( k, Value { original: v, current: v, pinned, }, + &mut evicted, ); placeholders.remove(&k); // if k is present it must have value v @@ -121,15 +119,20 @@ fn run(input: Input) { Op::Replace(k, v, pinned) => { // eprintln!("replace {k} {v}"); placeholders.remove(&k); - if let Ok(evicted) = cache.replace_with_lifecycle( - k, - Value { - original: v, - current: v, - pinned, - }, - false, - ) { + let mut evicted = Vec::new(); + if cache + .replace_with_lifecycle( + k, + Value { + original: v, + current: v, + pinned, + }, + false, + &mut evicted, + ) + .is_ok() + { // if k is present it must have value v let peek = cache.peek(&k); assert!(peek.is_none() || peek.unwrap().original == v); @@ -155,7 +158,8 @@ fn run(input: Input) { current: v, pinned, }; - if let Ok(evicted) = p.insert_with_lifecycle(value) { + let mut evicted = Vec::new(); + if p.insert_with_lifecycle(value, &mut evicted).is_ok() { let peek = cache.peek(&k); assert!(peek.is_none() || peek.unwrap().original == v); check_evicted(k, peek, evicted); diff --git a/fuzz/fuzz_targets/fuzz_unsync_cache.rs b/fuzz/fuzz_targets/fuzz_unsync_cache.rs index 6ead4d4..02dc73b 100644 --- a/fuzz/fuzz_targets/fuzz_unsync_cache.rs +++ b/fuzz/fuzz_targets/fuzz_unsync_cache.rs @@ -26,10 +26,6 @@ impl Weighter for MyWeighter { impl Lifecycle for MyLifecycle { type RequestState = Vec<(u16, Value)>; - fn begin_request(&self) -> Self::RequestState { - Default::default() - } - fn is_pinned(&self, _key: &u16, val: &Value) -> bool { val.pinned } @@ -99,13 +95,15 @@ fn run(input: Input) { match op { Op::Insert(k, v, pinned) => { // eprintln!("insert {k} {v}"); - let evicted = cache.insert_with_lifecycle( + let mut evicted = Vec::new(); + cache.insert_with_lifecycle( k, Value { original: v, current: v, pinned, }, + &mut evicted, ); // if k is present it must have value v let peek = cache.peek(&k).copied(); @@ -125,15 +123,20 @@ fn run(input: Input) { } Op::Replace(k, v, pinned) => { // eprintln!("replace {k} {v}"); - if let Ok(evicted) = cache.replace_with_lifecycle( - k, - Value { - original: v, - current: v, - pinned, - }, - false, - ) { + let mut evicted = Vec::new(); + if cache + .replace_with_lifecycle( + k, + Value { + original: v, + current: v, + pinned, + }, + false, + &mut evicted, + ) + .is_ok() + { // if k is present it must have value v let peek = cache.peek(&k).copied(); assert!(peek.is_none() || peek.unwrap().original == v); @@ -146,11 +149,15 @@ fn run(input: Input) { let (inserted, evicted) = match cache.get_ref_or_guard(&k) { Ok(_) => (false, Vec::new()), Err(g) => { - let evicted = g.insert_with_lifecycle(Value { - original: v, - current: v, - pinned, - }); + let mut evicted = Vec::new(); + g.insert_with_lifecycle( + Value { + original: v, + current: v, + pinned, + }, + &mut evicted, + ); (true, evicted) } }; diff --git a/fuzz/fuzz_targets/fuzz_unsync_cache_pinstate.rs b/fuzz/fuzz_targets/fuzz_unsync_cache_pinstate.rs index d26a3aa..68257c0 100644 --- a/fuzz/fuzz_targets/fuzz_unsync_cache_pinstate.rs +++ b/fuzz/fuzz_targets/fuzz_unsync_cache_pinstate.rs @@ -34,10 +34,6 @@ impl Weighter for MyWeighter { impl Lifecycle for MyLifecycle { type RequestState = Vec<(u16, Value)>; - fn begin_request(&self) -> Self::RequestState { - Default::default() - } - fn is_pinned(&self, _key: &u16, val: &Value) -> bool { let mut pinned = val.pinned.get(); if pinned.remaining != 0 { @@ -114,13 +110,15 @@ fn run(input: Input) { match op { Op::Insert(k, v, pinned) => { // eprintln!("insert {k} {v}"); - let evicted = cache.insert_with_lifecycle( + let mut evicted = Vec::new(); + cache.insert_with_lifecycle( k, Value { original: v, current: v, pinned: Cell::new(pinned), }, + &mut evicted, ); // if k is present it must have value v let peek = cache.peek(&k).cloned(); @@ -138,15 +136,20 @@ fn run(input: Input) { } Op::Replace(k, v, pinned) => { // eprintln!("replace {k} {v}"); - if let Ok(evicted) = cache.replace_with_lifecycle( - k, - Value { - original: v, - current: v, - pinned: Cell::new(pinned), - }, - false, - ) { + let mut evicted = Vec::new(); + if cache + .replace_with_lifecycle( + k, + Value { + original: v, + current: v, + pinned: Cell::new(pinned), + }, + false, + &mut evicted, + ) + .is_ok() + { // if k is present it must have value v let peek = cache.peek(&k).cloned(); assert!(peek.is_none() || peek.as_ref().unwrap().original == v); @@ -159,11 +162,15 @@ fn run(input: Input) { let (inserted, evicted) = match cache.get_ref_or_guard(&k) { Ok(_) => (false, Vec::new()), Err(g) => { - let evicted = g.insert_with_lifecycle(Value { - original: v, - current: v, - pinned: Cell::new(pinned), - }); + let mut evicted = Vec::new(); + g.insert_with_lifecycle( + Value { + original: v, + current: v, + pinned: Cell::new(pinned), + }, + &mut evicted, + ); (true, evicted) } }; diff --git a/src/lib.rs b/src/lib.rs index 433d0f0..4db978a 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -164,8 +164,31 @@ impl Weighter for UnitWeighter { /// Hooks into the lifetime of the cache items. /// /// The functions should be small and very fast, otherwise the cache performance might be negatively affected. +/// +/// # Request state +/// +/// Operations that may evict items thread a [`RequestState`](Lifecycle::RequestState) +/// through the eviction hooks. It is a per-request accumulator: the cache constructs a +/// fresh one via [`Default`], the `on_evict*`/`before_evict` hooks record into it, and it +/// is finalized by its own [`Drop`] (which, for example, releases evicted items _after_ +/// the shard lock is dropped). +/// +/// The `_with_lifecycle` cache methods take `&mut RequestState`, letting a caller drive +/// several operations against a single state — e.g. to batch the eviction work or inspect +/// evicted items before dropping it: +/// +/// ```ignore +/// let mut lcs = Default::default(); +/// cache.insert_with_lifecycle(k1, v1, &mut lcs); +/// cache.insert_with_lifecycle(k2, v2, &mut lcs); +/// // inspect `lcs` here if desired; evicted items are released when it drops +/// ``` pub trait Lifecycle { - type RequestState; + /// Per-request accumulator threaded through the eviction hooks. + /// + /// Constructed via [`Default`] at the start of each request and finalized by its + /// [`Drop`]. Keep it cheap to create and drop. + type RequestState: Default; /// Returns whether the item is pinned. Items that are pinned can't be evicted. /// Note that a pinned item can still be replaced with get_mut, insert, replace and similar APIs. @@ -181,9 +204,6 @@ pub trait Lifecycle { false } - /// Called before the insert request starts, e.g.: insert, replace. - fn begin_request(&self) -> Self::RequestState; - /// Called when a cache item is about to be evicted. /// Note that value replacement (e.g. insertions for the same key) won't call this method. /// @@ -233,16 +253,6 @@ pub trait Lifecycle { fn on_evict_hot(&self, state: &mut Self::RequestState, key: Key, val: Val) { self.on_evict(state, key, val) } - - /// Called after a request finishes, e.g.: insert, replace. - /// - /// Notes: - /// This will _not_ be called when using `_with_lifecycle` apis, which will return the RequestState instead. - /// This will _not_ be called if the request errored (e.g. a replace didn't find a value to replace). - /// If needed, Drop for RequestState can be used to detect these cases. - #[allow(unused_variables)] - #[inline] - fn end_request(&self, state: Self::RequestState) {} } /// The memory used by the cache diff --git a/src/shard.rs b/src/shard.rs index 746e39d..11769fc 100644 --- a/src/shard.rs +++ b/src/shard.rs @@ -1325,7 +1325,7 @@ impl< } } - pub fn set_capacity(&mut self, new_weight_capacity: u64) { + pub fn set_capacity(&mut self, new_weight_capacity: u64, lcs: &mut L::RequestState) { // Guard against division by zero when old capacity is 0 (produces inf/NaN ratios) if self.weight_capacity == 0 { self.weight_capacity = new_weight_capacity; @@ -1344,11 +1344,7 @@ impl< } // Evict items if we're over the new capacity - let mut lcs = self.lifecycle.begin_request(); - while self.weight_hot + self.weight_cold > self.weight_capacity - && self.advance_cold(&mut lcs) - {} - self.lifecycle.end_request(lcs); + while self.weight_hot + self.weight_cold > self.weight_capacity && self.advance_cold(lcs) {} // Trim ghost entries if needed while self.num_non_resident > self.capacity_non_resident { self.advance_ghost(); diff --git a/src/sync.rs b/src/sync.rs index 8810ec2..363fde0 100644 --- a/src/sync.rs +++ b/src/sync.rs @@ -59,7 +59,6 @@ pub struct Cache< hash_builder: B, shards: Box<[RwLock>>]>, shards_mask: u64, - lifecycle: L, } impl Cache { @@ -170,7 +169,6 @@ impl< shards: shards.into_boxed_slice(), hash_builder, shards_mask: num_shards - 1, - lifecycle, } } @@ -407,28 +405,32 @@ impl< /// /// Returns `Ok` if the entry was admitted and `Err(_)` if it wasn't. pub fn replace(&self, key: Key, value: Val, soft: bool) -> Result<(), (Key, Val)> { - let lcs = self.replace_with_lifecycle(key, value, soft)?; - self.lifecycle.end_request(lcs); - Ok(()) + let mut lcs = Default::default(); + self.replace_with_lifecycle(key, value, soft, &mut lcs) } - /// Inserts an item in the cache, but _only_ if an entry with key `key` already exists. + /// Inserts an item in the cache, but _only_ if an entry with key `key` already exists, + /// recording any evicted items into the given lifecycle request state. /// If `soft` is set, the replace operation won't affect the "hotness" of the entry, /// even if the value is replaced. /// /// Returns `Ok` if the entry was admitted and `Err(_)` if it wasn't. + /// + /// `lcs` is a [`Lifecycle::RequestState`]; construct one with `Default::default()`. + /// The same `&mut lcs` can be threaded through multiple operations to batch the + /// eviction work. Evicted items are released when `lcs` is dropped. pub fn replace_with_lifecycle( &self, key: Key, value: Val, soft: bool, - ) -> Result { - let mut lcs = self.lifecycle.begin_request(); + lcs: &mut L::RequestState, + ) -> Result<(), (Key, Val)> { let (shard, hash) = self.shard_for(&key).unwrap(); shard .write() - .insert(&mut lcs, hash, key, value, InsertStrategy::Replace { soft })?; - Ok(lcs) + .insert(lcs, hash, key, value, InsertStrategy::Replace { soft })?; + Ok(()) } /// Retains only the items specified by the predicate. @@ -445,8 +447,8 @@ impl< /// Inserts an item in the cache with key `key`. pub fn insert(&self, key: Key, value: Val) { - let lcs = self.insert_with_lifecycle(key, value); - self.lifecycle.end_request(lcs); + let mut lcs = Default::default(); + self.insert_with_lifecycle(key, value, &mut lcs); } /// Attempts to insert an item in the cache with key `key` without blocking. @@ -454,44 +456,48 @@ impl< /// could not be acquired without blocking. Lock contention is the only failure /// mode: the inputs are returned so the caller can retry or discard them. pub fn try_insert(&self, key: Key, value: Val) -> Result<(), (Key, Val)> { - let lcs = self.try_insert_with_lifecycle(key, value)?; - self.lifecycle.end_request(lcs); - Ok(()) + let mut lcs = Default::default(); + self.try_insert_with_lifecycle(key, value, &mut lcs) } - /// Inserts an item in the cache with key `key`. - pub fn insert_with_lifecycle(&self, key: Key, value: Val) -> L::RequestState { - let mut lcs = self.lifecycle.begin_request(); + /// Inserts an item in the cache with key `key`, recording any evicted items into the + /// given lifecycle request state. + /// + /// `lcs` is a [`Lifecycle::RequestState`]; construct one with `Default::default()`. + /// The same `&mut lcs` can be threaded through multiple operations to batch the + /// eviction work. Evicted items are released when `lcs` is dropped. + pub fn insert_with_lifecycle(&self, key: Key, value: Val, lcs: &mut L::RequestState) { let (shard, hash) = self.shard_for(&key).unwrap(); let result = shard .write() - .insert(&mut lcs, hash, key, value, InsertStrategy::Insert); + .insert(lcs, hash, key, value, InsertStrategy::Insert); // result cannot err with the Insert strategy debug_assert!(result.is_ok()); - lcs } - /// Attempts to insert an item in the cache with key `key` without blocking. - /// Returns `Ok(lcs)` with the lifecycle request state if the item was inserted, - /// or `Err((key, value))` if the shard lock could not be acquired without blocking. - /// Lock contention is the only failure mode: the inputs are returned so the - /// caller can retry or discard them. + /// Attempts to insert an item in the cache with key `key` without blocking, recording + /// any evicted items into the given lifecycle request state. + /// Returns `Ok(())` if the item was inserted, or `Err((key, value))` if the shard lock + /// could not be acquired without blocking. Lock contention is the only failure mode: + /// the inputs are returned so the caller can retry or discard them. + /// + /// `lcs` is a [`Lifecycle::RequestState`]; construct one with `Default::default()`. + /// The same `&mut lcs` can be threaded through multiple operations to batch the + /// eviction work. Evicted items are released when `lcs` is dropped. pub fn try_insert_with_lifecycle( &self, key: Key, value: Val, - ) -> Result { - // Tradeoff: begin_request is called before acquiring the shard lock to avoid holding - // the lock during potentially expensive lifecycle initialization. - let mut lcs = self.lifecycle.begin_request(); + lcs: &mut L::RequestState, + ) -> Result<(), (Key, Val)> { let (shard, hash) = self.shard_for(&key).unwrap(); match shard.try_write() { Some(mut shard) => { - let result = shard.insert(&mut lcs, hash, key, value, InsertStrategy::Insert); + let result = shard.insert(lcs, hash, key, value, InsertStrategy::Insert); // result cannot err with the Insert strategy debug_assert!(result.is_ok()); - Ok(lcs) + Ok(()) } _ => Err((key, value)), } @@ -547,7 +553,9 @@ impl< let shard_weight_cap = new_weight_capacity.saturating_add(self.shards.len() as u64 - 1) / self.shards.len() as u64; for shard in &*self.shards { - shard.write().set_capacity(shard_weight_cap); + let mut lcs = Default::default(); + // `lcs` drops after this statement's lock guard, releasing evicted items outside the lock. + shard.write().set_capacity(shard_weight_cap, &mut lcs); } } @@ -574,7 +582,7 @@ impl< if let Some(v) = shard.read().get(hash, key) { return GuardResult::Value(v.clone()); } - PlaceholderGuard::join(&self.lifecycle, shard, hash, key, timeout) + PlaceholderGuard::join(shard, hash, key, timeout) } /// Gets or inserts an item in the cache with key `key`. @@ -618,7 +626,7 @@ impl< if let Some(v) = shard.read().get(hash, key) { return Ok(v.clone()); } - match JoinFuture::new(&self.lifecycle, shard, hash, key).await { + match JoinFuture::new(shard, hash, key).await { JoinResult::Filled(Some(shared)) => { // SAFETY: Filled means the value was set by the loader. return Ok(unsafe { shared.value().unwrap_unchecked().clone() }); @@ -724,21 +732,16 @@ impl< EntryOrPlaceholder::Replaced(shared, old_val) => { drop(shard_guard); return EntryResult::Replaced( - PlaceholderGuard::start_loading(&self.lifecycle, shard, shared), + PlaceholderGuard::start_loading(shard, shared), old_val, ); } EntryOrPlaceholder::NewPlaceholder(shared) => { drop(shard_guard); - return EntryResult::Vacant(PlaceholderGuard::start_loading( - &self.lifecycle, - shard, - shared, - )); + return EntryResult::Vacant(PlaceholderGuard::start_loading(shard, shared)); } EntryOrPlaceholder::ExistingPlaceholder(shared) => { match PlaceholderGuard::wait_for_placeholder( - &self.lifecycle, shard, shard_guard, shared, @@ -783,16 +786,14 @@ impl< EntryOrPlaceholder::Replaced(shared, old_val) => { drop(shard_guard); Ok(EntryResult::Replaced( - PlaceholderGuard::start_loading(&self.lifecycle, shard, shared), + PlaceholderGuard::start_loading(shard, shared), old_val, )) } EntryOrPlaceholder::NewPlaceholder(shared) => { drop(shard_guard); Ok(EntryResult::Vacant(PlaceholderGuard::start_loading( - &self.lifecycle, - shard, - shared, + shard, shared, ))) } EntryOrPlaceholder::ExistingPlaceholder(_) => Err(()), @@ -800,7 +801,7 @@ impl< }; match result { Ok(entry_result) => return entry_result, - Err(()) => match JoinFuture::new(&self.lifecycle, shard, hash, key).await { + Err(()) => match JoinFuture::new(shard, hash, key).await { JoinResult::Filled(_) => continue, JoinResult::Guard(g) => return EntryResult::Vacant(g), JoinResult::Timeout => unsafe { unreachable_unchecked() }, @@ -938,11 +939,6 @@ impl Lifecycle for DefaultLifecycle { // overhead (e.g. a vector) for this default lifecycle. type RequestState = [Option<(Key, Val)>; 2]; - #[inline] - fn begin_request(&self) -> Self::RequestState { - [None, None] - } - #[inline] fn on_evict(&self, state: &mut Self::RequestState, key: Key, val: Val) { if std::mem::needs_drop::<(Key, Val)>() { @@ -1750,19 +1746,24 @@ mod tests { #[test] fn test_try_insert_with_lifecycle() { let cache = Cache::new(100); + let mut lcs = Default::default(); - // Successful insert returns the lifecycle request state. - let result = cache.try_insert_with_lifecycle(1, 10); - assert!(result.is_ok()); - let lcs = result.ok().unwrap(); - cache.lifecycle.end_request(lcs); + // Successful insert records evictions into the provided request state. + assert_eq!(cache.try_insert_with_lifecycle(1, 10, &mut lcs), Ok(())); assert_eq!(cache.get(&1), Some(10)); + // The same request state can be threaded through several operations. + assert_eq!(cache.try_insert_with_lifecycle(2, 20, &mut lcs), Ok(())); + assert_eq!(cache.get(&2), Some(20)); + // Contended when a read lock is held. let guards: Vec<_> = cache.shards.iter().map(|s| s.read()).collect(); - assert_eq!(cache.try_insert_with_lifecycle(2, 20), Err((2, 20))); + assert_eq!( + cache.try_insert_with_lifecycle(3, 30, &mut lcs), + Err((3, 30)) + ); drop(guards); - assert_eq!(cache.get(&2), None); + assert_eq!(cache.get(&3), None); } #[test] diff --git a/src/sync_placeholder.rs b/src/sync_placeholder.rs index 5da45e2..23cb867 100644 --- a/src/sync_placeholder.rs +++ b/src/sync_placeholder.rs @@ -87,7 +87,6 @@ enum LoadingState { } pub struct PlaceholderGuard<'a, Key, Val, We, B, L> { - lifecycle: &'a L, shard: &'a RwLock>>, shared: SharedPlaceholder, inserted: bool, @@ -196,7 +195,6 @@ pub enum EntryResult<'a, Key, Val, We, B, L, T> { impl<'a, Key, Val, We, B, L> PlaceholderGuard<'a, Key, Val, We, B, L> { #[inline] pub fn start_loading( - lifecycle: &'a L, shard: &'a RwLock>>, shared: SharedPlaceholder, ) -> Self { @@ -205,7 +203,6 @@ impl<'a, Key, Val, We, B, L> PlaceholderGuard<'a, Key, Val, We, B, L> { LoadingState::Loading )); PlaceholderGuard { - lifecycle, shard, shared, inserted: false, @@ -216,7 +213,6 @@ impl<'a, Key, Val, We, B, L> PlaceholderGuard<'a, Key, Val, We, B, L> { // or a guard if the caller got the guard. #[inline] fn handle_notification( - lifecycle: &'a L, shard: &'a RwLock>>, shared: SharedPlaceholder, ) -> Result, PlaceholderGuard<'a, Key, Val, We, B, L>> { @@ -225,7 +221,7 @@ impl<'a, Key, Val, We, B, L> PlaceholderGuard<'a, Key, Val, We, B, L> { if shared.value().is_some() { Ok(shared) } else { - Err(PlaceholderGuard::start_loading(lifecycle, shard, shared)) + Err(PlaceholderGuard::start_loading(shard, shared)) } } @@ -265,7 +261,6 @@ impl< > PlaceholderGuard<'a, Key, Val, We, B, L> { pub fn join( - lifecycle: &'a L, shard: &'a RwLock>>, hash: u64, key: &Q, @@ -278,12 +273,12 @@ impl< let shared = match shard_guard.get_or_placeholder(hash, key) { Ok((_, v)) => return GuardResult::Value(v.clone()), Err((shared, true)) => { - return GuardResult::Guard(Self::start_loading(lifecycle, shard, shared)); + return GuardResult::Guard(Self::start_loading(shard, shared)); } Err((shared, false)) => shared, }; let mut deadline = timeout.map(Ok); - match Self::wait_for_placeholder(lifecycle, shard, shard_guard, shared, deadline.as_mut()) { + match Self::wait_for_placeholder(shard, shard_guard, shared, deadline.as_mut()) { JoinResult::Filled(shared) => unsafe { // SAFETY: Filled means the value was set by the loader. GuardResult::Value(shared.unwrap_unchecked().value().unwrap_unchecked().clone()) @@ -302,7 +297,6 @@ impl< /// call. On first use the duration is converted in-place to `Err(instant)` so that /// callers that retry (e.g. `entry`) preserve the original deadline across calls. pub(crate) fn wait_for_placeholder( - lifecycle: &'a L, shard: &'a RwLock>>, shard_guard: RwLockWriteGuard<'a, CacheShard>>, shared: SharedPlaceholder, @@ -345,7 +339,7 @@ impl< if let Some(instant) = deadline { let remaining = instant.saturating_duration_since(Instant::now()); if remaining.is_zero() { - return Self::join_timeout(lifecycle, shard, shared, parked_thread, ¬ified); + return Self::join_timeout(shard, shared, parked_thread, ¬ified); } #[cfg(not(fuzzing))] thread::park_timeout(remaining); @@ -354,7 +348,7 @@ impl< thread::park(); } if notified.load(Ordering::Acquire) { - return match Self::handle_notification(lifecycle, shard, shared) { + return match Self::handle_notification(shard, shared) { Ok(shared) => JoinResult::Filled(Some(shared)), Err(g) => JoinResult::Guard(g), }; @@ -364,7 +358,6 @@ impl< #[cold] fn join_timeout( - lifecycle: &'a L, shard: &'a RwLock>>>, shared: Arc>, // when timeout is zero, the thread may have not been added to the waiters list @@ -375,7 +368,7 @@ impl< match state.loading { LoadingState::Loading if notified.load(Ordering::Acquire) => { drop(state); // Drop state guard to avoid a deadlock with start_loading - JoinResult::Guard(PlaceholderGuard::start_loading(lifecycle, shard, shared)) + JoinResult::Guard(PlaceholderGuard::start_loading(shard, shared)) } LoadingState::Loading => { if parked_thread.is_some() { @@ -414,18 +407,24 @@ impl< /// A placeholder can be removed as a result of a `remove` call /// or a non-placeholder `insert` with the same key. pub fn insert(self, value: Val) -> Result<(), Val> { - let lifecycle = self.lifecycle; - let lcs = self.insert_with_lifecycle(value)?; - lifecycle.end_request(lcs); - Ok(()) + let mut lcs = Default::default(); + self.insert_with_lifecycle(value, &mut lcs) } - /// Inserts the value into the placeholder + /// Inserts the value into the placeholder, recording any evicted items into the given + /// lifecycle request state. /// /// Returns Err if the placeholder isn't in the cache anymore. /// A placeholder can be removed as a result of a `remove` call /// or a non-placeholder `insert` with the same key. - pub fn insert_with_lifecycle(mut self, value: Val) -> Result { + /// + /// `lcs` is a [`Lifecycle::RequestState`]; construct one with `Default::default()`. + /// Evicted items are released when `lcs` is dropped. + pub fn insert_with_lifecycle( + mut self, + value: Val, + lcs: &mut L::RequestState, + ) -> Result<(), Val> { unsafe { self.shared.value.set(value.clone()).unwrap_unchecked() }; let referenced; { @@ -446,11 +445,10 @@ impl< // - the placeholder will be removed here, if it still exists self.inserted = true; - let mut lcs = self.lifecycle.begin_request(); self.shard .write() - .replace_placeholder(&mut lcs, &self.shared, referenced, value)?; - Ok(lcs) + .replace_placeholder(lcs, &self.shared, referenced, value)?; + Ok(()) } } @@ -510,7 +508,6 @@ impl std::fmt::Debug for PlaceholderGuard<'_, Key, Val, We, /// be moved after the first poll, keeping that pointer valid. The pointer is /// cleaned up in `drop_pending_waiter` before the struct is destroyed. pub(crate) struct JoinFuture<'a, 'b, Q: ?Sized, Key, Val, We, B, L> { - lifecycle: &'a L, shard: &'a RwLock>>, hash: u64, key: &'b Q, @@ -530,13 +527,11 @@ enum JoinFutureState { impl<'a, 'b, Q: ?Sized, Key, Val, We, B, L> JoinFuture<'a, 'b, Q, Key, Val, We, B, L> { pub(crate) fn new( - lifecycle: &'a L, shard: &'a RwLock>>, hash: u64, key: &'b Q, ) -> Self { Self { - lifecycle, shard, hash, key, @@ -561,7 +556,7 @@ impl JoinFuture<'_, '_, Q, Key, Val, We, B, L> { // The write guard was abandoned elsewhere, this future was notified but didn't get polled. // So we get and drop the guard here to handle the side effects. drop(state); // Drop state guard to avoid a deadlock with start_loading - let _ = PlaceholderGuard::start_loading(self.lifecycle, self.shard, shared); + let _ = PlaceholderGuard::start_loading(self.shard, shared); } LoadingState::Loading => { // Remove ourselves from the waiters list @@ -607,7 +602,6 @@ impl< // fields. The `notified` field's address (registered in the waiter list) stays // stable because Pin guarantees the future won't be moved. let this = unsafe { self.get_unchecked_mut() }; - let lifecycle = this.lifecycle; let shard = this.shard; match &mut this.state { JoinFutureState::Created => { @@ -621,7 +615,7 @@ impl< this.state = JoinFutureState::Done; drop(shard_guard); Poll::Ready(JoinResult::Guard(PlaceholderGuard::start_loading( - lifecycle, shard, shared, + shard, shared, ))) } Err((shared, false)) => { @@ -680,12 +674,10 @@ impl< else { unsafe { unreachable_unchecked() } }; - Poll::Ready( - match PlaceholderGuard::handle_notification(lifecycle, shard, shared) { - Ok(shared) => JoinResult::Filled(Some(shared)), - Err(g) => JoinResult::Guard(g), - }, - ) + Poll::Ready(match PlaceholderGuard::handle_notification(shard, shared) { + Ok(shared) => JoinResult::Filled(Some(shared)), + Err(g) => JoinResult::Guard(g), + }) } JoinFutureState::Done => panic!("Polled after ready"), } diff --git a/src/unsync.rs b/src/unsync.rs index ae7f1d5..65af821 100644 --- a/src/unsync.rs +++ b/src/unsync.rs @@ -212,31 +212,35 @@ impl, B: BuildHasher, L: Lifecycle Result<(), (Key, Val)> { - let lcs = self.replace_with_lifecycle(key, value, soft)?; - self.shard.lifecycle.end_request(lcs); - Ok(()) + let mut lcs = Default::default(); + self.replace_with_lifecycle(key, value, soft, &mut lcs) } - /// Replaces an item in the cache, but only if it already exists. + /// Replaces an item in the cache, but only if it already exists, recording any evicted + /// items into the given lifecycle request state. /// If `soft` is set, the replace operation won't affect the "hotness" of the key, /// even if the value is replaced. /// /// Returns `Ok` if the entry was admitted and `Err(_)` if it wasn't. + /// + /// `lcs` is a [`Lifecycle::RequestState`]; construct one with `Default::default()`. + /// The same `&mut lcs` can be threaded through multiple operations to batch the + /// eviction work. Evicted items are released when `lcs` is dropped. pub fn replace_with_lifecycle( &mut self, key: Key, value: Val, soft: bool, - ) -> Result { - let mut lcs = self.shard.lifecycle.begin_request(); + lcs: &mut L::RequestState, + ) -> Result<(), (Key, Val)> { self.shard.insert( - &mut lcs, + lcs, self.shard.hash(&key), key, value, InsertStrategy::Replace { soft }, )?; - Ok(lcs) + Ok(()) } /// Retains only the items specified by the predicate. @@ -265,9 +269,8 @@ impl, B: BuildHasher, L: Lifecycle idx, Err((plh, _)) => { let v = with()?; - let mut lcs = self.shard.lifecycle.begin_request(); + let mut lcs = Default::default(); let replaced = self.shard.replace_placeholder(&mut lcs, &plh, false, v); - self.shard.lifecycle.end_request(lcs); debug_assert!(replaced.is_ok(), "unsync replace_placeholder can't fail"); plh.idx } @@ -291,10 +294,9 @@ impl, B: BuildHasher, L: Lifecycle idx, Err((plh, _)) => { let v = with()?; - let mut lcs = self.shard.lifecycle.begin_request(); + let mut lcs = Default::default(); let replaced = self.shard.replace_placeholder(&mut lcs, &plh, false, v); debug_assert!(replaced.is_ok(), "unsync replace_placeholder can't fail"); - self.shard.lifecycle.end_request(lcs); plh.idx } }; @@ -349,15 +351,19 @@ impl, B: BuildHasher, L: Lifecycle L::RequestState { - let mut lcs = self.shard.lifecycle.begin_request(); + /// Inserts an item in the cache with key `key`, recording any evicted items into the + /// given lifecycle request state. + /// + /// `lcs` is a [`Lifecycle::RequestState`]; construct one with `Default::default()`. + /// The same `&mut lcs` can be threaded through multiple operations to batch the + /// eviction work. Evicted items are released when `lcs` is dropped. + pub fn insert_with_lifecycle(&mut self, key: Key, value: Val, lcs: &mut L::RequestState) { let result = self.shard.insert( - &mut lcs, + lcs, self.shard.hash(&key), key, value, @@ -365,7 +371,6 @@ impl, B: BuildHasher, L: Lifecycle, B: BuildHasher, L: Lifecycle Clone for DefaultLifecycle { impl Lifecycle for DefaultLifecycle { type RequestState = (); - - #[inline] - fn begin_request(&self) -> Self::RequestState {} } #[derive(Debug, Clone)] @@ -462,29 +465,22 @@ impl, B: BuildHasher, L: Lifecycle L::RequestState { - self.insert_internal(value, true).unwrap() + let mut lcs = Default::default(); + self.insert_with_lifecycle(value, &mut lcs); } - #[inline] - fn insert_internal(mut self, value: Val, return_lcs: bool) -> Option { - let mut lcs = self.cache.shard.lifecycle.begin_request(); - let replaced = - self.cache - .shard - .replace_placeholder(&mut lcs, &self.placeholder, false, value); + /// Inserts the value into the placeholder, recording any evicted items into the given + /// lifecycle request state. + /// + /// `lcs` is a [`Lifecycle::RequestState`]; construct one with `Default::default()`. + /// Evicted items are released when `lcs` is dropped. + pub fn insert_with_lifecycle(mut self, value: Val, lcs: &mut L::RequestState) { + let replaced = self + .cache + .shard + .replace_placeholder(lcs, &self.placeholder, false, value); debug_assert!(replaced.is_ok(), "unsync replace_placeholder can't fail"); self.inserted = true; - if return_lcs { - Some(lcs) - } else { - self.cache.shard.lifecycle.end_request(lcs); - None - } } }