imago/
async_lru_cache.rs

1//! Provides a least-recently-used cache with async access.
2//!
3//! To operate, this cache is bound to an I/O back-end object that provides the loading and
4//! flushing of cache entries.
5//!
6//! The cache holds `map.write()` during eviction I/O, so cross-cache flush dependencies
7//! (e.g. “flush cache B before evicting from cache A”) must be handled externally (see e.g.
8//! qcow2’s [`MetadataCaches`](../qcow2/cache/struct.MetadataCaches.html)).
9
10#![allow(dead_code)]
11
12use crate::sync_primitives::{RwLock, RwLockWriteGuard};
13#[cfg(feature = "async")]
14use futures::stream::{FuturesUnordered, StreamExt};
15use maybe_async::maybe_async;
16use std::collections::HashMap;
17use std::fmt::Debug;
18use std::hash::Hash;
19use std::sync::atomic::{AtomicUsize, Ordering};
20use std::sync::Arc;
21use std::{io, mem};
22use tracing::{error, instrument, trace};
23
24/// Cache entry structure, wrapping the cached object.
25pub(crate) struct AsyncLruCacheEntry<V> {
26    /// Cached object.
27    ///
28    /// Always set during operation, only cleared when trying to unwrap the `Arc` on eviction.
29    value: Option<Arc<V>>,
30
31    /// When this entry was last accessed.
32    last_used: AtomicUsize,
33}
34
35/// Least-recently-used cache with async access.
36struct AsyncLruCacheInner<
37    Key: Clone + Copy + Debug + PartialEq + Eq + Hash + Send + Sync,
38    Value: Send + Sync,
39    IoBackend: AsyncLruCacheBackend<Key = Key, Value = Value>,
40> {
41    /// I/O back-end that performs loading and flushing of cache entries.
42    backend: IoBackend,
43
44    /// Cache entries.
45    map: RwLock<HashMap<Key, AsyncLruCacheEntry<Value>>>,
46
47    /// Monotonically increasing counter to generate “timestamps”.
48    lru_timer: AtomicUsize,
49
50    /// Upper limit of how many entries to cache.
51    limit: usize,
52}
53
54/// Least-recently-used cache with async access.
55///
56/// Keeps the least recently used entries up to a limited count.  Accessing and flushing is
57/// async-aware.
58///
59/// `K` is the key used to uniquely identify cache entries, `V` is the cached data.
60pub(crate) struct AsyncLruCache<
61    K: Clone + Copy + Debug + PartialEq + Eq + Hash + Send + Sync,
62    V: Send + Sync,
63    B: AsyncLruCacheBackend<Key = K, Value = V>,
64>(Arc<AsyncLruCacheInner<K, V, B>>);
65
66/// Provides loading and flushing for cache entries.
67#[maybe_async(AFIT)]
68pub(crate) trait AsyncLruCacheBackend: Send + Sync {
69    /// Key type.
70    type Key: Clone + Copy + Debug + PartialEq + Eq + Hash + Send + Sync;
71    /// Value (object) type.
72    type Value: Send + Sync;
73
74    /// Load the given object.
75    #[allow(async_fn_in_trait)] // No need for Send
76    async fn load(&self, key: Self::Key) -> io::Result<Self::Value>;
77
78    /// Flush the given object.
79    ///
80    /// The implementation should itself check whether the object is dirty; `flush()` is called for
81    /// all evicted cache entries, regardless of whether they actually are dirty or not.
82    #[allow(async_fn_in_trait)] // No need for Send
83    async fn flush(&self, key: Self::Key, value: &Self::Value) -> io::Result<()>;
84
85    /// Drop the given object without flushing.
86    ///
87    /// The cache owner is invalidating the cache, evicting all objects without flushing them.  If
88    /// dropping the object as-is would cause problems (e.g. because it is verified not to be
89    /// dirty), those problems need to be resolved here.
90    ///
91    /// # Safety
92    /// Depending on the nature of the cache, this operation may be unsafe.  Must only be performed
93    /// if the cache owner requested it and guarantees it is safe.
94    unsafe fn evict(&self, key: Self::Key, value: Self::Value);
95}
96
97#[maybe_async]
98impl<
99        K: Clone + Copy + Debug + PartialEq + Eq + Hash + Send + Sync,
100        V: Send + Sync,
101        B: AsyncLruCacheBackend<Key = K, Value = V>,
102    > AsyncLruCache<K, V, B>
103{
104    /// Create a new cache.
105    ///
106    /// `size` is the maximum number of entries to keep in the cache.
107    pub fn new(backend: B, size: usize) -> Self {
108        AsyncLruCache(Arc::new(AsyncLruCacheInner {
109            backend,
110            map: Default::default(),
111            lru_timer: AtomicUsize::new(0),
112            limit: size,
113        }))
114    }
115
116    /// Retrieve an entry from the cache.
117    ///
118    /// If there is no entry yet, load it via the backend.
119    ///
120    /// If there is no more room in the cache for a new entry and `may_flush` is true, flush out
121    /// the oldest entry via `flush()` to make space.
122    ///
123    /// `Ok(None)` is returned if and only if there is no more room in the cache and `may_flush` is
124    /// false.
125    pub async fn get_or_insert(&self, key: K, may_flush: bool) -> io::Result<Option<Arc<V>>> {
126        self.0.get_or_insert(key, may_flush).await
127    }
128
129    /// Force-insert the given object into the cache.
130    ///
131    /// If there is an existing object under that key and `may_flush` is true, it is flushed first.
132    ///
133    /// If there is no existing object yet, i.e. a new entry must be created, but there is no more
134    /// room in the cache for this new entry, and `may_flush` is true, the oldest entry is flushed
135    /// out first to make space.
136    ///
137    /// On success, `Ok(true)` is returned.  `Ok(false)` is returned if and only if `may_flush` was
138    /// false, but an older cache entry would need to be flushed.
139    pub async fn insert(&self, key: K, value: Arc<V>, may_flush: bool) -> io::Result<bool> {
140        self.0.insert(key, value, may_flush).await
141    }
142
143    /// Flush all cache entries.
144    ///
145    /// Those entries are not evicted, but remain in the cache.
146    pub async fn flush(&self) -> io::Result<()> {
147        self.0.flush().await
148    }
149
150    /// Evict all cache entries.
151    ///
152    /// Evicts all cache entries without flushing them.
153    ///
154    /// # Safety
155    /// Depending on the nature of the cache, this operation may be unsafe.  Perform at your own
156    /// risk.
157    pub async unsafe fn invalidate(&self) -> io::Result<()> {
158        unsafe { self.0.invalidate() }.await
159    }
160}
161
162#[maybe_async]
163impl<
164        K: Clone + Copy + Debug + PartialEq + Eq + Hash + Send + Sync,
165        V: Send + Sync,
166        B: AsyncLruCacheBackend<Key = K, Value = V>,
167    > AsyncLruCacheInner<K, V, B>
168{
169    /// Ensure there is at least one free entry in the cache.
170    ///
171    /// If there are free entries, return `Ok(true)` immediately.
172    ///
173    /// If there are no free entries and `may_evict` is true, evict the least-recently-used entry
174    /// by flushing it via `backend.flush()`, then return `Ok(true)` on success.
175    ///
176    /// If there are no free entries and `may_evict` is false, return `Ok(false)`.
177    ///
178    /// Note that this function holds the write lock for its entire lifetime, so `backend.flush()`
179    /// must not call back into this cache directly or indirectly.  Cross-cache flush ordering must
180    /// be handled externally (e.g. by qcow2’s
181    /// [`MetadataCaches`](../qcow2/cache/struct.MetadataCaches.html)).
182    #[instrument(
183        level = "trace",
184        name = "AsyncLruCache::ensure_free_entry",
185        skip_all,
186        fields(self = &self as *const _ as usize),
187    )]
188    async fn ensure_free_entry(
189        &self,
190        map: &mut RwLockWriteGuard<'_, HashMap<K, AsyncLruCacheEntry<V>>>,
191        may_evict: bool,
192    ) -> io::Result<bool> {
193        if map.len() < self.limit {
194            return Ok(true);
195        } else if !may_evict {
196            return Ok(false);
197        }
198
199        while map.len() >= self.limit {
200            trace!("{} / {} used", map.len(), self.limit);
201
202            let now = self.lru_timer.load(Ordering::Relaxed);
203            let oldest = map
204                .iter()
205                .filter(|(_key, entry)| Arc::strong_count(entry.value()) == 1)
206                .fold((0, None), |oldest, (key, entry)| {
207                    // Users must not create weak references, and so we know that with a `strong_count`
208                    // of 1 (while holding the map’s write lock), no one can access this entry anymore
209                    // and we could safely drop it.
210                    assert_eq!(Arc::weak_count(entry.value()), 0);
211
212                    let age = now.wrapping_sub(entry.last_used.load(Ordering::Relaxed));
213                    if age >= oldest.0 {
214                        (age, Some(*key))
215                    } else {
216                        oldest
217                    }
218                });
219
220            let Some(oldest_key) = oldest.1 else {
221                error!("Cannot evict entry from cache; everything is in use");
222                return Err(io::Error::other(
223                    "Cannot evict entry from cache; everything is in use",
224                ));
225            };
226
227            trace!("Removing entry with key {oldest_key:?}, aged {}", oldest.0);
228
229            let oldest_entry = map.remove(&oldest_key).unwrap();
230
231            // We checked `strong_count` above to be 1, and there are no weak references, so the
232            // only reference to this entry must have been the one in the map.  We held the write
233            // lock throughout, there was no await point between the check and here, so the
234            // `strong_count` must still be 1 and we can thus safely unwrap the `Arc`.
235            let evicted_object = Arc::try_unwrap(oldest_entry.value.unwrap())
236                .unwrap_or_else(|_| panic!("entry has gained external references"));
237
238            trace!("Flushing {oldest_key:?}");
239            if let Err(err) = self.backend.flush(oldest_key, &evicted_object).await {
240                map.insert(
241                    oldest_key,
242                    AsyncLruCacheEntry {
243                        value: Some(Arc::new(evicted_object)),
244                        last_used: oldest_entry.last_used.load(Ordering::Relaxed).into(),
245                    },
246                );
247                return Err(err);
248            }
249        }
250
251        Ok(true)
252    }
253
254    /// Retrieve an entry from the cache.
255    ///
256    /// If there is no entry yet, load it via the backend.
257    ///
258    /// If there is no more room in the cache for a new entry and `may_flush` is true, flush out
259    /// the oldest entry via `flush()` to make space.
260    ///
261    /// `Ok(None)` is returned if and only if there is no more room in the cache and `may_flush` is
262    /// false.
263    ///
264    /// Users must not create weak references to the returned `Arc`.
265    async fn get_or_insert(&self, key: K, may_flush: bool) -> io::Result<Option<Arc<V>>> {
266        {
267            let map = self.map.read().await;
268            if let Some(entry) = map.get(&key) {
269                entry.last_used.store(
270                    self.lru_timer.fetch_add(1, Ordering::Relaxed),
271                    Ordering::Relaxed,
272                );
273                return Ok(Some(Arc::clone(entry.value())));
274            }
275        }
276
277        let mut map = self.map.write().await;
278        if let Some(entry) = map.get(&key) {
279            entry.last_used.store(
280                self.lru_timer.fetch_add(1, Ordering::Relaxed),
281                Ordering::Relaxed,
282            );
283            return Ok(Some(Arc::clone(entry.value())));
284        }
285
286        if !self.ensure_free_entry(&mut map, may_flush).await? {
287            return Ok(None);
288        }
289
290        let object = Arc::new(self.backend.load(key).await?);
291
292        let new_entry = AsyncLruCacheEntry {
293            value: Some(Arc::clone(&object)),
294            last_used: AtomicUsize::new(self.lru_timer.fetch_add(1, Ordering::Relaxed)),
295        };
296        map.insert(key, new_entry);
297
298        Ok(Some(object))
299    }
300
301    /// Force-insert the given object into the cache.
302    ///
303    /// If there is an existing object under that key and `may_flush` is true, it is flushed first.
304    ///
305    /// If there is no existing object yet, i.e. a new entry must be created, but there is no more
306    /// room in the cache for this new entry, and `may_flush` is true, the oldest entry is flushed
307    /// out first to make space.
308    ///
309    /// On success, `Ok(true)` is returned.  `Ok(false)` is returned if and only if `may_flush` was
310    /// false, but an older cache entry would need to be flushed.
311    async fn insert(&self, key: K, value: Arc<V>, may_flush: bool) -> io::Result<bool> {
312        let mut map = self.map.write().await;
313        if let Some(entry) = map.get_mut(&key) {
314            if !may_flush {
315                return Ok(false);
316            }
317
318            entry.last_used.store(
319                self.lru_timer.fetch_add(1, Ordering::Relaxed),
320                Ordering::Relaxed,
321            );
322            self.backend.flush(key, entry.value()).await?;
323            entry.value = Some(value);
324        } else {
325            if !self.ensure_free_entry(&mut map, may_flush).await? {
326                return Ok(false);
327            }
328
329            let new_entry = AsyncLruCacheEntry {
330                value: Some(value),
331                last_used: AtomicUsize::new(self.lru_timer.fetch_add(1, Ordering::Relaxed)),
332            };
333            map.insert(key, new_entry);
334        }
335
336        Ok(true)
337    }
338
339    /// Flush all cache entries.
340    ///
341    /// Those entries are not evicted, but remain in the cache.
342    #[instrument(
343        level = "trace",
344        name = "AsyncLruCache::flush",
345        skip_all,
346        fields(self = &self as *const _ as usize)
347    )]
348    async fn flush(&self) -> io::Result<()> {
349        #[cfg(feature = "async")]
350        let mut futs = FuturesUnordered::new();
351
352        let map = self.map.read().await;
353        let mut first_err: Option<io::Error> = None;
354        for (key, entry) in map.iter() {
355            let key = *key;
356            trace!("Flushing {key:?}");
357            #[cfg(feature = "async")]
358            futs.push({
359                let object = Arc::clone(entry.value());
360                async move { self.backend.flush(key, &object).await }
361            });
362            #[cfg(feature = "sync")]
363            if let Err(e) = self.backend.flush(key, entry.value()) {
364                first_err.get_or_insert(e);
365            }
366        }
367
368        #[cfg(feature = "async")]
369        while let Some(result) = futs.next().await {
370            if let Err(e) = result {
371                first_err.get_or_insert(e);
372            }
373        }
374        if let Some(e) = first_err {
375            Err(e)
376        } else {
377            Ok(())
378        }
379    }
380
381    /// Evict all cache entries.
382    ///
383    /// Evicts all cache entries without flushing them.
384    ///
385    /// # Safety
386    /// Depending on the nature of the cache, this operation may be unsafe.  Perform at your own
387    /// risk.
388    #[instrument(
389        level = "trace",
390        name = "AsyncLruCache::invalidate",
391        skip_all,
392        fields(self = &self as *const _ as usize)
393    )]
394    async unsafe fn invalidate(&self) -> io::Result<()> {
395        let mut in_use = Vec::new();
396
397        let mut map = self.map.write().await;
398        // Clear the map; we could use `.drain()`, but doing this allows the following loop to put
399        // objects back into the new map in case they cannot be evicted.
400        let old_map = mem::take(&mut *map);
401        for (key, mut entry) in old_map {
402            let object = entry.value.take().unwrap();
403            trace!("Evicting {key:?}");
404            match Arc::try_unwrap(object) {
405                Ok(object) => {
406                    // Caller guarantees this is safe
407                    unsafe { self.backend.evict(key, object) };
408                }
409
410                Err(arc) => {
411                    trace!("Entry is still in use, retaining it");
412                    entry.value = Some(arc);
413                    map.insert(key, entry);
414                    in_use.push(key);
415                }
416            }
417        }
418
419        if in_use.is_empty() {
420            Ok(())
421        } else {
422            Err(io::Error::other(format!(
423                "Cannot invalidate cache, entries still in use: {}",
424                in_use
425                    .iter()
426                    .map(|key| format!("{key:?}"))
427                    .collect::<Vec<String>>()
428                    .join(", "),
429            )))
430        }
431    }
432}
433
434impl<V> AsyncLruCacheEntry<V> {
435    /// Return the cached object.
436    fn value(&self) -> &Arc<V> {
437        self.value.as_ref().unwrap()
438    }
439}
440
441#[cfg(test)]
442mod tests {
443    use super::*;
444    use std::sync::atomic::AtomicUsize;
445
446    /// Minimal backend for testing: load returns the key, flush is a no-op
447    struct DummyBackend;
448
449    #[maybe_async(AFIT)]
450    impl AsyncLruCacheBackend for DummyBackend {
451        type Key = usize;
452        type Value = usize;
453
454        async fn load(&self, key: usize) -> io::Result<usize> {
455            Ok(key)
456        }
457
458        async fn flush(&self, _key: usize, _value: &usize) -> io::Result<()> {
459            Ok(())
460        }
461
462        unsafe fn evict(&self, _key: usize, _value: usize) {}
463    }
464
465    /// Backend that records flush calls in order
466    #[derive(Default)]
467    struct RecordingBackend {
468        flushed: std::sync::Mutex<Vec<(usize, usize)>>,
469    }
470
471    #[maybe_async(AFIT)]
472    impl AsyncLruCacheBackend for RecordingBackend {
473        type Key = usize;
474        type Value = usize;
475
476        async fn load(&self, key: usize) -> io::Result<usize> {
477            Ok(key)
478        }
479
480        async fn flush(&self, key: usize, value: &usize) -> io::Result<()> {
481            self.flushed.lock().unwrap().push((key, *value));
482            Ok(())
483        }
484
485        unsafe fn evict(&self, _key: usize, _value: usize) {}
486    }
487
488    #[maybe_async(AFIT)]
489    impl<B: AsyncLruCacheBackend> AsyncLruCacheBackend for Arc<B> {
490        type Key = <B as AsyncLruCacheBackend>::Key;
491        type Value = <B as AsyncLruCacheBackend>::Value;
492
493        async fn load(&self, key: Self::Key) -> io::Result<Self::Value> {
494            (**self).load(key).await
495        }
496
497        async fn flush(&self, key: Self::Key, value: &Self::Value) -> io::Result<()> {
498            (**self).flush(key, value).await
499        }
500
501        unsafe fn evict(&self, key: Self::Key, value: Self::Value) {
502            unsafe { (**self).evict(key, value) }
503        }
504    }
505
506    /// `flush()` must continue past individual entry errors and report the first one, not stop at
507    /// the first failure
508    #[maybe_async::test(feature = "sync", async(feature = "async", tokio::test))]
509    async fn test_flush_continues_past_errors() {
510        #[derive(Default)]
511        struct FailOddBackend {
512            flush_count: AtomicUsize,
513        }
514
515        #[maybe_async(AFIT)]
516        impl AsyncLruCacheBackend for FailOddBackend {
517            type Key = usize;
518            type Value = usize;
519
520            async fn load(&self, key: usize) -> io::Result<usize> {
521                Ok(key)
522            }
523
524            async fn flush(&self, key: usize, _value: &usize) -> io::Result<()> {
525                self.flush_count.fetch_add(1, Ordering::Relaxed);
526                if key % 2 == 1 {
527                    Err(io::Error::other("odd key"))
528                } else {
529                    Ok(())
530                }
531            }
532
533            unsafe fn evict(&self, _key: usize, _value: usize) {}
534        }
535
536        const ENTRIES: usize = 42;
537
538        let backend = Arc::new(FailOddBackend::default());
539        let cache = AsyncLruCache::new(Arc::clone(&backend), ENTRIES);
540
541        for i in 0..ENTRIES {
542            cache.get_or_insert(i, false).await.unwrap().unwrap();
543        }
544
545        let err = cache.flush().await.unwrap_err();
546        assert!(err.to_string().contains("odd key"));
547
548        assert_eq!(backend.flush_count.load(Ordering::Relaxed), ENTRIES);
549    }
550
551    /// Eviction must remove the least-recently-used entry
552    #[maybe_async::test(feature = "sync", async(feature = "async", tokio::test))]
553    async fn test_lru_eviction_order() {
554        const ENTRIES: usize = 3;
555
556        let backend = Arc::new(RecordingBackend::default());
557        let cache = AsyncLruCache::new(Arc::clone(&backend), ENTRIES);
558
559        for i in 0..ENTRIES {
560            cache.get_or_insert(i, false).await.unwrap().unwrap();
561        }
562
563        // Touch key 0 so it becomes most-recently-used
564        cache.get_or_insert(0, false).await.unwrap().unwrap();
565
566        // Insert one more key — must evict key 1 (the oldest untouched)
567        let entry = cache.get_or_insert(ENTRIES, false).await.unwrap();
568        assert_eq!(entry, None);
569        cache.get_or_insert(ENTRIES, true).await.unwrap().unwrap();
570
571        assert_eq!(*backend.flushed.lock().unwrap(), [(1, 1)]);
572    }
573
574    /// Entries with external `Arc` references must not be evicted
575    #[maybe_async::test(feature = "sync", async(feature = "async", tokio::test))]
576    async fn test_in_use_entries_not_evicted() {
577        let backend = Arc::new(RecordingBackend::default());
578        let cache = AsyncLruCache::new(Arc::clone(&backend), 2);
579
580        let held = cache.get_or_insert(0, false).await.unwrap().unwrap();
581        cache.get_or_insert(1, false).await.unwrap().unwrap();
582
583        // Insert key 2 — key 0 is oldest but in use, so key 1 must be evicted
584        let entry = cache.get_or_insert(2, false).await.unwrap();
585        assert_eq!(entry, None);
586        cache.get_or_insert(2, true).await.unwrap().unwrap();
587
588        assert_eq!(*backend.flushed.lock().unwrap(), [(1, 1)]);
589        assert_eq!(*held, 0);
590    }
591
592    /// When all entries are in use, eviction must fail with an error
593    #[maybe_async::test(feature = "sync", async(feature = "async", tokio::test))]
594    async fn test_cache_full_all_in_use() {
595        const ENTRIES: usize = 23;
596
597        let cache = AsyncLruCache::new(DummyBackend, ENTRIES);
598
599        let mut held = vec![];
600        for i in 0..ENTRIES {
601            held.push(cache.get_or_insert(i, false).await.unwrap().unwrap());
602        }
603
604        let entry = cache.get_or_insert(ENTRIES, false).await.unwrap();
605        assert_eq!(entry, None);
606        let err = cache.get_or_insert(ENTRIES, true).await.unwrap_err();
607        assert!(err.to_string().contains("everything is in use"));
608    }
609
610    /// `invalidate()` must retain entries that are still in use and evict the rest
611    #[maybe_async::test(feature = "sync", async(feature = "async", tokio::test))]
612    async fn test_invalidate_retains_in_use() {
613        let cache = AsyncLruCache::new(DummyBackend, 16);
614
615        let held = cache.get_or_insert(0, false).await.unwrap().unwrap();
616        cache.get_or_insert(1, false).await.unwrap().unwrap();
617        cache.get_or_insert(2, false).await.unwrap().unwrap();
618
619        let err = unsafe { cache.invalidate() }.await.unwrap_err();
620        assert!(err.to_string().contains("still in use"));
621
622        let from_cache = cache.get_or_insert(0, false).await.unwrap().unwrap();
623        assert!(Arc::ptr_eq(&from_cache, &held));
624        let from_cache = cache.get_or_insert(0, true).await.unwrap().unwrap();
625        assert!(Arc::ptr_eq(&from_cache, &held));
626
627        let len = cache.0.map.read().await.len();
628        assert_eq!(len, 1);
629    }
630
631    /// When eviction flush fails, the entry must be re-inserted and remain accessible
632    #[maybe_async::test(feature = "sync", async(feature = "async", tokio::test))]
633    async fn test_eviction_flush_failure_reinserts_entry() {
634        struct FailFlushBackend;
635
636        #[maybe_async(AFIT)]
637        impl AsyncLruCacheBackend for FailFlushBackend {
638            type Key = usize;
639            type Value = usize;
640
641            async fn load(&self, key: usize) -> io::Result<usize> {
642                Ok(key)
643            }
644
645            async fn flush(&self, _key: usize, _value: &usize) -> io::Result<()> {
646                Err(io::Error::other("flush failed"))
647            }
648
649            unsafe fn evict(&self, _key: usize, _value: usize) {}
650        }
651
652        const ENTRIES: usize = 2;
653
654        let cache = AsyncLruCache::new(FailFlushBackend, ENTRIES);
655
656        for i in 0..ENTRIES {
657            cache.get_or_insert(i, false).await.unwrap().unwrap();
658        }
659
660        // Cache is full
661        let entry = cache.get_or_insert(ENTRIES, false).await.unwrap();
662        assert_eq!(entry, None);
663        // And eviction flush fails
664        let err = cache.get_or_insert(ENTRIES, true).await.unwrap_err();
665        assert!(err.to_string().contains("flush failed"));
666
667        // All original entries must still be in the cache
668        let len = cache.0.map.read().await.len();
669        assert_eq!(len, ENTRIES);
670        for i in 0..ENTRIES {
671            let entry = cache.get_or_insert(i, false).await.unwrap().unwrap();
672            assert_eq!(*entry, i);
673        }
674
675        // New entry was never inserted
676        let entry = cache.get_or_insert(ENTRIES, false).await.unwrap();
677        assert_eq!(entry, None);
678        let err = cache.get_or_insert(ENTRIES, true).await.unwrap_err();
679        assert!(err.to_string().contains("flush failed"));
680    }
681
682    /// `insert()` over an existing key must flush the old value first
683    #[maybe_async::test(feature = "sync", async(feature = "async", tokio::test))]
684    async fn test_insert_flushes_existing() {
685        let backend = Arc::new(RecordingBackend::default());
686        let cache = AsyncLruCache::new(Arc::clone(&backend), 16);
687
688        cache.get_or_insert(5, false).await.unwrap().unwrap();
689        let inserted = cache.insert(5, Arc::new(55), false).await.unwrap();
690        assert!(!inserted);
691        let inserted = cache.insert(5, Arc::new(55), true).await.unwrap();
692        assert!(inserted);
693
694        assert_eq!(*backend.flushed.lock().unwrap(), [(5, 5)]);
695        let entry = *cache.get_or_insert(5, false).await.unwrap().unwrap();
696        assert_eq!(entry, 55);
697        let entry = *cache.get_or_insert(5, true).await.unwrap().unwrap();
698        assert_eq!(entry, 55);
699        let len = cache.0.map.read().await.len();
700        assert_eq!(len, 1);
701    }
702}
imago/async_lru_cache.rs

imago/
async_lru_cache.rs
