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            // Flush the entry while it is still in the map, and only remove it once that has
230            // succeeded: If flushing fails, or this future is dropped before flushing is done, the
231            // entry must remain in the cache, or its modifications would be lost.
232            // We checked `strong_count` above to be 1, and there are no weak references, so the
233            // only reference to this entry is the one in the map.  We hold the write lock
234            // throughout, so nobody can gain a new reference while we flush.
235            trace!("Flushing {oldest_key:?}");
236            let oldest_entry = map.get(&oldest_key).unwrap();
237            self.backend.flush(oldest_key, oldest_entry.value()).await?;
238            map.remove(&oldest_key);
239        }
240
241        Ok(true)
242    }
243
244    /// Retrieve an entry from the cache.
245    ///
246    /// If there is no entry yet, load it via the backend.
247    ///
248    /// If there is no more room in the cache for a new entry and `may_flush` is true, flush out
249    /// the oldest entry via `flush()` to make space.
250    ///
251    /// `Ok(None)` is returned if and only if there is no more room in the cache and `may_flush` is
252    /// false.
253    ///
254    /// Users must not create weak references to the returned `Arc`.
255    async fn get_or_insert(&self, key: K, may_flush: bool) -> io::Result<Option<Arc<V>>> {
256        {
257            let map = self.map.read().await;
258            if let Some(entry) = map.get(&key) {
259                entry.last_used.store(
260                    self.lru_timer.fetch_add(1, Ordering::Relaxed),
261                    Ordering::Relaxed,
262                );
263                return Ok(Some(Arc::clone(entry.value())));
264            }
265        }
266
267        let mut map = self.map.write().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        if !self.ensure_free_entry(&mut map, may_flush).await? {
277            return Ok(None);
278        }
279
280        let object = Arc::new(self.backend.load(key).await?);
281
282        let new_entry = AsyncLruCacheEntry {
283            value: Some(Arc::clone(&object)),
284            last_used: AtomicUsize::new(self.lru_timer.fetch_add(1, Ordering::Relaxed)),
285        };
286        map.insert(key, new_entry);
287
288        Ok(Some(object))
289    }
290
291    /// Force-insert the given object into the cache.
292    ///
293    /// If there is an existing object under that key and `may_flush` is true, it is flushed first.
294    ///
295    /// If there is no existing object yet, i.e. a new entry must be created, but there is no more
296    /// room in the cache for this new entry, and `may_flush` is true, the oldest entry is flushed
297    /// out first to make space.
298    ///
299    /// On success, `Ok(true)` is returned.  `Ok(false)` is returned if and only if `may_flush` was
300    /// false, but an older cache entry would need to be flushed.
301    async fn insert(&self, key: K, value: Arc<V>, may_flush: bool) -> io::Result<bool> {
302        let mut map = self.map.write().await;
303        if let Some(entry) = map.get_mut(&key) {
304            if !may_flush {
305                return Ok(false);
306            }
307
308            entry.last_used.store(
309                self.lru_timer.fetch_add(1, Ordering::Relaxed),
310                Ordering::Relaxed,
311            );
312            self.backend.flush(key, entry.value()).await?;
313            entry.value = Some(value);
314        } else {
315            if !self.ensure_free_entry(&mut map, may_flush).await? {
316                return Ok(false);
317            }
318
319            let new_entry = AsyncLruCacheEntry {
320                value: Some(value),
321                last_used: AtomicUsize::new(self.lru_timer.fetch_add(1, Ordering::Relaxed)),
322            };
323            map.insert(key, new_entry);
324        }
325
326        Ok(true)
327    }
328
329    /// Flush all cache entries.
330    ///
331    /// Those entries are not evicted, but remain in the cache.
332    #[instrument(
333        level = "trace",
334        name = "AsyncLruCache::flush",
335        skip_all,
336        fields(self = &self as *const _ as usize)
337    )]
338    async fn flush(&self) -> io::Result<()> {
339        #[cfg(feature = "async")]
340        let mut futs = FuturesUnordered::new();
341
342        let map = self.map.read().await;
343        let mut first_err: Option<io::Error> = None;
344        for (key, entry) in map.iter() {
345            let key = *key;
346            trace!("Flushing {key:?}");
347            #[cfg(feature = "async")]
348            futs.push({
349                let object = Arc::clone(entry.value());
350                async move { self.backend.flush(key, &object).await }
351            });
352            #[cfg(feature = "sync")]
353            if let Err(e) = self.backend.flush(key, entry.value()) {
354                first_err.get_or_insert(e);
355            }
356        }
357
358        #[cfg(feature = "async")]
359        while let Some(result) = futs.next().await {
360            if let Err(e) = result {
361                first_err.get_or_insert(e);
362            }
363        }
364        if let Some(e) = first_err {
365            Err(e)
366        } else {
367            Ok(())
368        }
369    }
370
371    /// Evict all cache entries.
372    ///
373    /// Evicts all cache entries without flushing them.
374    ///
375    /// # Safety
376    /// Depending on the nature of the cache, this operation may be unsafe.  Perform at your own
377    /// risk.
378    #[instrument(
379        level = "trace",
380        name = "AsyncLruCache::invalidate",
381        skip_all,
382        fields(self = &self as *const _ as usize)
383    )]
384    async unsafe fn invalidate(&self) -> io::Result<()> {
385        let mut in_use = Vec::new();
386
387        let mut map = self.map.write().await;
388        // Clear the map; we could use `.drain()`, but doing this allows the following loop to put
389        // objects back into the new map in case they cannot be evicted.
390        let old_map = mem::take(&mut *map);
391        for (key, mut entry) in old_map {
392            let object = entry.value.take().unwrap();
393            trace!("Evicting {key:?}");
394            match Arc::try_unwrap(object) {
395                Ok(object) => {
396                    // Caller guarantees this is safe
397                    unsafe { self.backend.evict(key, object) };
398                }
399
400                Err(arc) => {
401                    trace!("Entry is still in use, retaining it");
402                    entry.value = Some(arc);
403                    map.insert(key, entry);
404                    in_use.push(key);
405                }
406            }
407        }
408
409        if in_use.is_empty() {
410            Ok(())
411        } else {
412            Err(io::Error::other(format!(
413                "Cannot invalidate cache, entries still in use: {}",
414                in_use
415                    .iter()
416                    .map(|key| format!("{key:?}"))
417                    .collect::<Vec<String>>()
418                    .join(", "),
419            )))
420        }
421    }
422}
423
424impl<V> AsyncLruCacheEntry<V> {
425    /// Return the cached object.
426    fn value(&self) -> &Arc<V> {
427        self.value.as_ref().unwrap()
428    }
429}
430
431#[cfg(test)]
432mod tests {
433    use super::*;
434    use std::sync::atomic::AtomicUsize;
435
436    /// Minimal backend for testing: load returns the key, flush is a no-op
437    struct DummyBackend;
438
439    #[maybe_async(AFIT)]
440    impl AsyncLruCacheBackend for DummyBackend {
441        type Key = usize;
442        type Value = usize;
443
444        async fn load(&self, key: usize) -> io::Result<usize> {
445            Ok(key)
446        }
447
448        async fn flush(&self, _key: usize, _value: &usize) -> io::Result<()> {
449            Ok(())
450        }
451
452        unsafe fn evict(&self, _key: usize, _value: usize) {}
453    }
454
455    /// Backend that records flush calls in order
456    #[derive(Default)]
457    struct RecordingBackend {
458        flushed: parking_lot::Mutex<Vec<(usize, usize)>>,
459    }
460
461    #[maybe_async(AFIT)]
462    impl AsyncLruCacheBackend for RecordingBackend {
463        type Key = usize;
464        type Value = usize;
465
466        async fn load(&self, key: usize) -> io::Result<usize> {
467            Ok(key)
468        }
469
470        async fn flush(&self, key: usize, value: &usize) -> io::Result<()> {
471            self.flushed.lock().push((key, *value));
472            Ok(())
473        }
474
475        unsafe fn evict(&self, _key: usize, _value: usize) {}
476    }
477
478    #[maybe_async(AFIT)]
479    impl<B: AsyncLruCacheBackend> AsyncLruCacheBackend for Arc<B> {
480        type Key = <B as AsyncLruCacheBackend>::Key;
481        type Value = <B as AsyncLruCacheBackend>::Value;
482
483        async fn load(&self, key: Self::Key) -> io::Result<Self::Value> {
484            (**self).load(key).await
485        }
486
487        async fn flush(&self, key: Self::Key, value: &Self::Value) -> io::Result<()> {
488            (**self).flush(key, value).await
489        }
490
491        unsafe fn evict(&self, key: Self::Key, value: Self::Value) {
492            unsafe { (**self).evict(key, value) }
493        }
494    }
495
496    /// `flush()` must continue past individual entry errors and report the first one, not stop at
497    /// the first failure
498    #[maybe_async::test(feature = "sync", async(feature = "async", tokio::test))]
499    async fn test_flush_continues_past_errors() {
500        #[derive(Default)]
501        struct FailOddBackend {
502            flush_count: AtomicUsize,
503        }
504
505        #[maybe_async(AFIT)]
506        impl AsyncLruCacheBackend for FailOddBackend {
507            type Key = usize;
508            type Value = usize;
509
510            async fn load(&self, key: usize) -> io::Result<usize> {
511                Ok(key)
512            }
513
514            async fn flush(&self, key: usize, _value: &usize) -> io::Result<()> {
515                self.flush_count.fetch_add(1, Ordering::Relaxed);
516                if key % 2 == 1 {
517                    Err(io::Error::other("odd key"))
518                } else {
519                    Ok(())
520                }
521            }
522
523            unsafe fn evict(&self, _key: usize, _value: usize) {}
524        }
525
526        const ENTRIES: usize = 42;
527
528        let backend = Arc::new(FailOddBackend::default());
529        let cache = AsyncLruCache::new(Arc::clone(&backend), ENTRIES);
530
531        for i in 0..ENTRIES {
532            cache.get_or_insert(i, false).await.unwrap().unwrap();
533        }
534
535        let err = cache.flush().await.unwrap_err();
536        assert!(err.to_string().contains("odd key"));
537
538        assert_eq!(backend.flush_count.load(Ordering::Relaxed), ENTRIES);
539    }
540
541    /// Eviction must remove the least-recently-used entry
542    #[maybe_async::test(feature = "sync", async(feature = "async", tokio::test))]
543    async fn test_lru_eviction_order() {
544        const ENTRIES: usize = 3;
545
546        let backend = Arc::new(RecordingBackend::default());
547        let cache = AsyncLruCache::new(Arc::clone(&backend), ENTRIES);
548
549        for i in 0..ENTRIES {
550            cache.get_or_insert(i, false).await.unwrap().unwrap();
551        }
552
553        // Touch key 0 so it becomes most-recently-used
554        cache.get_or_insert(0, false).await.unwrap().unwrap();
555
556        // Insert one more key — must evict key 1 (the oldest untouched)
557        let entry = cache.get_or_insert(ENTRIES, false).await.unwrap();
558        assert_eq!(entry, None);
559        cache.get_or_insert(ENTRIES, true).await.unwrap().unwrap();
560
561        assert_eq!(*backend.flushed.lock(), [(1, 1)]);
562    }
563
564    /// Entries with external `Arc` references must not be evicted
565    #[maybe_async::test(feature = "sync", async(feature = "async", tokio::test))]
566    async fn test_in_use_entries_not_evicted() {
567        let backend = Arc::new(RecordingBackend::default());
568        let cache = AsyncLruCache::new(Arc::clone(&backend), 2);
569
570        let held = cache.get_or_insert(0, false).await.unwrap().unwrap();
571        cache.get_or_insert(1, false).await.unwrap().unwrap();
572
573        // Insert key 2 — key 0 is oldest but in use, so key 1 must be evicted
574        let entry = cache.get_or_insert(2, false).await.unwrap();
575        assert_eq!(entry, None);
576        cache.get_or_insert(2, true).await.unwrap().unwrap();
577
578        assert_eq!(*backend.flushed.lock(), [(1, 1)]);
579        assert_eq!(*held, 0);
580    }
581
582    /// When all entries are in use, eviction must fail with an error
583    #[maybe_async::test(feature = "sync", async(feature = "async", tokio::test))]
584    async fn test_cache_full_all_in_use() {
585        const ENTRIES: usize = 23;
586
587        let cache = AsyncLruCache::new(DummyBackend, ENTRIES);
588
589        let mut held = vec![];
590        for i in 0..ENTRIES {
591            held.push(cache.get_or_insert(i, false).await.unwrap().unwrap());
592        }
593
594        let entry = cache.get_or_insert(ENTRIES, false).await.unwrap();
595        assert_eq!(entry, None);
596        let err = cache.get_or_insert(ENTRIES, true).await.unwrap_err();
597        assert!(err.to_string().contains("everything is in use"));
598    }
599
600    /// `invalidate()` must retain entries that are still in use and evict the rest
601    #[maybe_async::test(feature = "sync", async(feature = "async", tokio::test))]
602    async fn test_invalidate_retains_in_use() {
603        let cache = AsyncLruCache::new(DummyBackend, 16);
604
605        let held = cache.get_or_insert(0, false).await.unwrap().unwrap();
606        cache.get_or_insert(1, false).await.unwrap().unwrap();
607        cache.get_or_insert(2, false).await.unwrap().unwrap();
608
609        let err = unsafe { cache.invalidate() }.await.unwrap_err();
610        assert!(err.to_string().contains("still in use"));
611
612        let from_cache = cache.get_or_insert(0, false).await.unwrap().unwrap();
613        assert!(Arc::ptr_eq(&from_cache, &held));
614        let from_cache = cache.get_or_insert(0, true).await.unwrap().unwrap();
615        assert!(Arc::ptr_eq(&from_cache, &held));
616
617        let len = cache.0.map.read().await.len();
618        assert_eq!(len, 1);
619    }
620
621    /// When eviction flush fails, the entry must be re-inserted and remain accessible
622    #[maybe_async::test(feature = "sync", async(feature = "async", tokio::test))]
623    async fn test_eviction_flush_failure_reinserts_entry() {
624        struct FailFlushBackend;
625
626        #[maybe_async(AFIT)]
627        impl AsyncLruCacheBackend for FailFlushBackend {
628            type Key = usize;
629            type Value = usize;
630
631            async fn load(&self, key: usize) -> io::Result<usize> {
632                Ok(key)
633            }
634
635            async fn flush(&self, _key: usize, _value: &usize) -> io::Result<()> {
636                Err(io::Error::other("flush failed"))
637            }
638
639            unsafe fn evict(&self, _key: usize, _value: usize) {}
640        }
641
642        const ENTRIES: usize = 2;
643
644        let cache = AsyncLruCache::new(FailFlushBackend, ENTRIES);
645
646        for i in 0..ENTRIES {
647            cache.get_or_insert(i, false).await.unwrap().unwrap();
648        }
649
650        // Cache is full
651        let entry = cache.get_or_insert(ENTRIES, false).await.unwrap();
652        assert_eq!(entry, None);
653        // And eviction flush fails
654        let err = cache.get_or_insert(ENTRIES, true).await.unwrap_err();
655        assert!(err.to_string().contains("flush failed"));
656
657        // All original entries must still be in the cache
658        let len = cache.0.map.read().await.len();
659        assert_eq!(len, ENTRIES);
660        for i in 0..ENTRIES {
661            let entry = cache.get_or_insert(i, false).await.unwrap().unwrap();
662            assert_eq!(*entry, i);
663        }
664
665        // New entry was never inserted
666        let entry = cache.get_or_insert(ENTRIES, false).await.unwrap();
667        assert_eq!(entry, None);
668        let err = cache.get_or_insert(ENTRIES, true).await.unwrap_err();
669        assert!(err.to_string().contains("flush failed"));
670    }
671
672    /// `insert()` over an existing key must flush the old value first
673    #[maybe_async::test(feature = "sync", async(feature = "async", tokio::test))]
674    async fn test_insert_flushes_existing() {
675        let backend = Arc::new(RecordingBackend::default());
676        let cache = AsyncLruCache::new(Arc::clone(&backend), 16);
677
678        cache.get_or_insert(5, false).await.unwrap().unwrap();
679        let inserted = cache.insert(5, Arc::new(55), false).await.unwrap();
680        assert!(!inserted);
681        let inserted = cache.insert(5, Arc::new(55), true).await.unwrap();
682        assert!(inserted);
683
684        assert_eq!(*backend.flushed.lock(), [(5, 5)]);
685        let entry = *cache.get_or_insert(5, false).await.unwrap().unwrap();
686        assert_eq!(entry, 55);
687        let entry = *cache.get_or_insert(5, true).await.unwrap().unwrap();
688        assert_eq!(entry, 55);
689        let len = cache.0.map.read().await.len();
690        assert_eq!(len, 1);
691    }
692
693    /// When the future evicting an entry is dropped while flushing it, the entry must remain in
694    /// the cache (otherwise, its modifications would be lost)
695    #[cfg(feature = "async")]
696    #[tokio::test]
697    async fn test_dropped_eviction_flush_keeps_entry() {
698        use futures::FutureExt;
699
700        /// Backend whose flushes never complete, and that counts loads
701        #[derive(Default)]
702        struct PendingFlushBackend {
703            loads: AtomicUsize,
704        }
705
706        #[maybe_async(AFIT)]
707        impl AsyncLruCacheBackend for PendingFlushBackend {
708            type Key = usize;
709            type Value = usize;
710
711            async fn load(&self, key: usize) -> io::Result<usize> {
712                self.loads.fetch_add(1, Ordering::Relaxed);
713                Ok(key)
714            }
715
716            async fn flush(&self, _key: usize, _value: &usize) -> io::Result<()> {
717                std::future::pending().await
718            }
719
720            unsafe fn evict(&self, _key: usize, _value: usize) {}
721        }
722
723        let backend = Arc::new(PendingFlushBackend::default());
724        let cache = AsyncLruCache::new(Arc::clone(&backend), 2);
725
726        cache.get_or_insert(0, false).await.unwrap().unwrap();
727        cache.get_or_insert(1, false).await.unwrap().unwrap();
728        assert_eq!(backend.loads.load(Ordering::Relaxed), 2);
729
730        // Evicting the oldest entry (key 0) stays pending on its flush; drop it then
731        assert!(cache.get_or_insert(2, true).now_or_never().is_none());
732
733        // Check that both keys still exist, i.e. must not need to be loaded again
734        let entry = cache.get_or_insert(0, false).await.unwrap().unwrap();
735        assert_eq!(*entry, 0);
736        let entry = cache.get_or_insert(1, false).await.unwrap().unwrap();
737        assert_eq!(*entry, 1);
738
739        // And that the number of loads did not change
740        assert_eq!(backend.loads.load(Ordering::Relaxed), 2);
741    }
742}
imago/async_lru_cache.rs

imago/
async_lru_cache.rs
