foyer_memory/
cache.rs

1// Copyright 2025 foyer Project Authors
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::{borrow::Cow, fmt::Debug, future::Future, hash::Hash, ops::Deref, sync::Arc};
16
17use equivalent::Equivalent;
18use foyer_common::{
19    code::{DefaultHasher, HashBuilder, Key, Value},
20    event::EventListener,
21    future::Diversion,
22    metrics::Metrics,
23    properties::{Hint, Location, Properties, Source},
24    runtime::SingletonHandle,
25};
26use mixtrics::{metrics::BoxedRegistry, registry::noop::NoopMetricsRegistry};
27use pin_project::pin_project;
28use serde::{Deserialize, Serialize};
29use tokio::sync::oneshot;
30
31use crate::{
32    eviction::{
33        fifo::{Fifo, FifoConfig},
34        lfu::{Lfu, LfuConfig},
35        lru::{Lru, LruConfig},
36        s3fifo::{S3Fifo, S3FifoConfig},
37    },
38    raw::{FetchContext, FetchState, RawCache, RawCacheConfig, RawCacheEntry, RawFetch, Weighter},
39    Piece, Pipe, Result,
40};
41
42/// Entry properties for in-memory only cache.
43#[derive(Debug, Clone, Default)]
44pub struct CacheProperties {
45    ephemeral: bool,
46    hint: Hint,
47}
48
49impl CacheProperties {
50    /// Set the entry to be ephemeral.
51    ///
52    /// An ephemeral entry will be evicted immediately after all its holders drop it,
53    /// no matter if the capacity is reached.
54    pub fn with_ephemeral(mut self, ephemeral: bool) -> Self {
55        self.ephemeral = ephemeral;
56        self
57    }
58
59    /// Get if the entry is ephemeral.
60    pub fn ephemeral(&self) -> bool {
61        self.ephemeral
62    }
63
64    /// Set entry hint.
65    pub fn with_hint(mut self, hint: Hint) -> Self {
66        self.hint = hint;
67        self
68    }
69
70    /// Get entry hint.
71    pub fn hint(&self) -> Hint {
72        self.hint
73    }
74}
75
76impl Properties for CacheProperties {
77    fn with_ephemeral(self, ephemeral: bool) -> Self {
78        self.with_ephemeral(ephemeral)
79    }
80
81    fn ephemeral(&self) -> Option<bool> {
82        Some(self.ephemeral())
83    }
84
85    fn with_hint(self, hint: Hint) -> Self {
86        self.with_hint(hint)
87    }
88
89    fn hint(&self) -> Option<Hint> {
90        Some(self.hint())
91    }
92
93    fn with_location(self, _: Location) -> Self {
94        self
95    }
96
97    fn location(&self) -> Option<Location> {
98        None
99    }
100
101    fn with_source(self, _: Source) -> Self {
102        self
103    }
104
105    fn source(&self) -> Option<Source> {
106        None
107    }
108}
109
110pub type FifoCache<K, V, S = DefaultHasher, P = CacheProperties> = RawCache<Fifo<K, V, P>, S>;
111pub type FifoCacheEntry<K, V, S = DefaultHasher, P = CacheProperties> = RawCacheEntry<Fifo<K, V, P>, S>;
112pub type FifoFetch<K, V, ER, S = DefaultHasher, P = CacheProperties> = RawFetch<Fifo<K, V, P>, ER, S>;
113
114pub type S3FifoCache<K, V, S = DefaultHasher, P = CacheProperties> = RawCache<S3Fifo<K, V, P>, S>;
115pub type S3FifoCacheEntry<K, V, S = DefaultHasher, P = CacheProperties> = RawCacheEntry<S3Fifo<K, V, P>, S>;
116pub type S3FifoFetch<K, V, ER, S = DefaultHasher, P = CacheProperties> = RawFetch<S3Fifo<K, V, P>, ER, S>;
117
118pub type LruCache<K, V, S = DefaultHasher, P = CacheProperties> = RawCache<Lru<K, V, P>, S>;
119pub type LruCacheEntry<K, V, S = DefaultHasher, P = CacheProperties> = RawCacheEntry<Lru<K, V, P>, S>;
120pub type LruFetch<K, V, ER, S = DefaultHasher, P = CacheProperties> = RawFetch<Lru<K, V, P>, ER, S>;
121
122pub type LfuCache<K, V, S = DefaultHasher, P = CacheProperties> = RawCache<Lfu<K, V, P>, S>;
123pub type LfuCacheEntry<K, V, S = DefaultHasher, P = CacheProperties> = RawCacheEntry<Lfu<K, V, P>, S>;
124pub type LfuFetch<K, V, ER, S = DefaultHasher, P = CacheProperties> = RawFetch<Lfu<K, V, P>, ER, S>;
125
126/// A cached entry holder of the in-memory cache.
127#[derive(Debug)]
128pub enum CacheEntry<K, V, S = DefaultHasher, P = CacheProperties>
129where
130    K: Key,
131    V: Value,
132    S: HashBuilder,
133    P: Properties,
134{
135    /// A cached entry holder of the in-memory FIFO cache.
136    Fifo(FifoCacheEntry<K, V, S, P>),
137    /// A cached entry holder of the in-memory S3FIFO cache.
138    S3Fifo(S3FifoCacheEntry<K, V, S, P>),
139    /// A cached entry holder of the in-memory LRU cache.
140    Lru(LruCacheEntry<K, V, S, P>),
141    /// A cached entry holder of the in-memory LFU cache.
142    Lfu(LfuCacheEntry<K, V, S, P>),
143}
144
145impl<K, V, S, P> Clone for CacheEntry<K, V, S, P>
146where
147    K: Key,
148    V: Value,
149    S: HashBuilder,
150    P: Properties,
151{
152    fn clone(&self) -> Self {
153        match self {
154            Self::Fifo(entry) => Self::Fifo(entry.clone()),
155            Self::Lru(entry) => Self::Lru(entry.clone()),
156            Self::Lfu(entry) => Self::Lfu(entry.clone()),
157            Self::S3Fifo(entry) => Self::S3Fifo(entry.clone()),
158        }
159    }
160}
161
162impl<K, V, S, P> Deref for CacheEntry<K, V, S, P>
163where
164    K: Key,
165    V: Value,
166    S: HashBuilder,
167    P: Properties,
168{
169    type Target = V;
170
171    fn deref(&self) -> &Self::Target {
172        match self {
173            CacheEntry::Fifo(entry) => entry.deref(),
174            CacheEntry::Lru(entry) => entry.deref(),
175            CacheEntry::Lfu(entry) => entry.deref(),
176            CacheEntry::S3Fifo(entry) => entry.deref(),
177        }
178    }
179}
180
181impl<K, V, S, P> From<FifoCacheEntry<K, V, S, P>> for CacheEntry<K, V, S, P>
182where
183    K: Key,
184    V: Value,
185    S: HashBuilder,
186    P: Properties,
187{
188    fn from(entry: FifoCacheEntry<K, V, S, P>) -> Self {
189        Self::Fifo(entry)
190    }
191}
192
193impl<K, V, S, P> From<LruCacheEntry<K, V, S, P>> for CacheEntry<K, V, S, P>
194where
195    K: Key,
196    V: Value,
197    S: HashBuilder,
198    P: Properties,
199{
200    fn from(entry: LruCacheEntry<K, V, S, P>) -> Self {
201        Self::Lru(entry)
202    }
203}
204
205impl<K, V, S, P> From<LfuCacheEntry<K, V, S, P>> for CacheEntry<K, V, S, P>
206where
207    K: Key,
208    V: Value,
209    S: HashBuilder,
210    P: Properties,
211{
212    fn from(entry: LfuCacheEntry<K, V, S, P>) -> Self {
213        Self::Lfu(entry)
214    }
215}
216
217impl<K, V, S, P> From<S3FifoCacheEntry<K, V, S, P>> for CacheEntry<K, V, S, P>
218where
219    K: Key,
220    V: Value,
221    S: HashBuilder,
222    P: Properties,
223{
224    fn from(entry: S3FifoCacheEntry<K, V, S, P>) -> Self {
225        Self::S3Fifo(entry)
226    }
227}
228
229impl<K, V, S, P> CacheEntry<K, V, S, P>
230where
231    K: Key,
232    V: Value,
233    S: HashBuilder,
234    P: Properties,
235{
236    /// Key hash of the cached entry.
237    pub fn hash(&self) -> u64 {
238        match self {
239            CacheEntry::Fifo(entry) => entry.hash(),
240            CacheEntry::Lru(entry) => entry.hash(),
241            CacheEntry::Lfu(entry) => entry.hash(),
242            CacheEntry::S3Fifo(entry) => entry.hash(),
243        }
244    }
245
246    /// Key of the cached entry.
247    pub fn key(&self) -> &K {
248        match self {
249            CacheEntry::Fifo(entry) => entry.key(),
250            CacheEntry::Lru(entry) => entry.key(),
251            CacheEntry::Lfu(entry) => entry.key(),
252            CacheEntry::S3Fifo(entry) => entry.key(),
253        }
254    }
255
256    /// Value of the cached entry.
257    pub fn value(&self) -> &V {
258        match self {
259            CacheEntry::Fifo(entry) => entry.value(),
260            CacheEntry::Lru(entry) => entry.value(),
261            CacheEntry::Lfu(entry) => entry.value(),
262            CacheEntry::S3Fifo(entry) => entry.value(),
263        }
264    }
265
266    /// Properties of the cached entry.
267    pub fn properties(&self) -> &P {
268        match self {
269            CacheEntry::Fifo(entry) => entry.properties(),
270            CacheEntry::Lru(entry) => entry.properties(),
271            CacheEntry::Lfu(entry) => entry.properties(),
272            CacheEntry::S3Fifo(entry) => entry.properties(),
273        }
274    }
275
276    /// Weight of the cached entry.
277    pub fn weight(&self) -> usize {
278        match self {
279            CacheEntry::Fifo(entry) => entry.weight(),
280            CacheEntry::Lru(entry) => entry.weight(),
281            CacheEntry::Lfu(entry) => entry.weight(),
282            CacheEntry::S3Fifo(entry) => entry.weight(),
283        }
284    }
285
286    /// External reference count of the cached entry.
287    pub fn refs(&self) -> usize {
288        match self {
289            CacheEntry::Fifo(entry) => entry.refs(),
290            CacheEntry::Lru(entry) => entry.refs(),
291            CacheEntry::Lfu(entry) => entry.refs(),
292            CacheEntry::S3Fifo(entry) => entry.refs(),
293        }
294    }
295
296    /// If the cached entry is updated and outdated.
297    pub fn is_outdated(&self) -> bool {
298        match self {
299            CacheEntry::Fifo(entry) => entry.is_outdated(),
300            CacheEntry::Lru(entry) => entry.is_outdated(),
301            CacheEntry::Lfu(entry) => entry.is_outdated(),
302            CacheEntry::S3Fifo(entry) => entry.is_outdated(),
303        }
304    }
305
306    /// Get the piece of the entry record.
307    pub fn piece(&self) -> Piece<K, V, P> {
308        match self {
309            CacheEntry::Fifo(entry) => entry.piece(),
310            CacheEntry::Lru(entry) => entry.piece(),
311            CacheEntry::Lfu(entry) => entry.piece(),
312            CacheEntry::S3Fifo(entry) => entry.piece(),
313        }
314    }
315}
316
317/// Eviction algorithm config.
318#[derive(Debug, Clone, Serialize, Deserialize)]
319pub enum EvictionConfig {
320    /// FIFO eviction algorithm config.
321    Fifo(FifoConfig),
322    /// S3FIFO eviction algorithm config.
323    S3Fifo(S3FifoConfig),
324    /// LRU eviction algorithm config.
325    Lru(LruConfig),
326    /// LFU eviction algorithm config.
327    Lfu(LfuConfig),
328}
329
330impl From<FifoConfig> for EvictionConfig {
331    fn from(value: FifoConfig) -> EvictionConfig {
332        EvictionConfig::Fifo(value)
333    }
334}
335
336impl From<S3FifoConfig> for EvictionConfig {
337    fn from(value: S3FifoConfig) -> EvictionConfig {
338        EvictionConfig::S3Fifo(value)
339    }
340}
341
342impl From<LruConfig> for EvictionConfig {
343    fn from(value: LruConfig) -> EvictionConfig {
344        EvictionConfig::Lru(value)
345    }
346}
347
348impl From<LfuConfig> for EvictionConfig {
349    fn from(value: LfuConfig) -> EvictionConfig {
350        EvictionConfig::Lfu(value)
351    }
352}
353
354/// In-memory cache builder.
355pub struct CacheBuilder<K, V, S>
356where
357    K: Key,
358    V: Value,
359    S: HashBuilder,
360{
361    name: Cow<'static, str>,
362
363    capacity: usize,
364    shards: usize,
365    eviction_config: EvictionConfig,
366
367    hash_builder: S,
368    weighter: Arc<dyn Weighter<K, V>>,
369
370    event_listener: Option<Arc<dyn EventListener<Key = K, Value = V>>>,
371
372    registry: BoxedRegistry,
373    metrics: Option<Arc<Metrics>>,
374}
375
376impl<K, V> CacheBuilder<K, V, DefaultHasher>
377where
378    K: Key,
379    V: Value,
380{
381    /// Create a new in-memory cache builder.
382    pub fn new(capacity: usize) -> Self {
383        Self {
384            name: "foyer".into(),
385
386            capacity,
387            shards: 8,
388            eviction_config: LruConfig::default().into(),
389
390            hash_builder: Default::default(),
391            weighter: Arc::new(|_, _| 1),
392            event_listener: None,
393
394            registry: Box::new(NoopMetricsRegistry),
395            metrics: None,
396        }
397    }
398}
399
400impl<K, V, S> CacheBuilder<K, V, S>
401where
402    K: Key,
403    V: Value,
404    S: HashBuilder,
405{
406    /// Set the name of the foyer in-memory cache instance.
407    ///
408    /// foyer will use the name as the prefix of the metric names.
409    ///
410    /// Default: `foyer`.
411    pub fn with_name(mut self, name: impl Into<Cow<'static, str>>) -> Self {
412        self.name = name.into();
413        self
414    }
415
416    /// Set in-memory cache sharding count. Entries will be distributed to different shards based on their hash.
417    /// Operations on different shard can be parallelized.
418    pub fn with_shards(mut self, shards: usize) -> Self {
419        self.shards = shards;
420        self
421    }
422
423    /// Set in-memory cache eviction algorithm.
424    ///
425    /// The default value is a general-used w-TinyLFU algorithm.
426    pub fn with_eviction_config(mut self, eviction_config: impl Into<EvictionConfig>) -> Self {
427        self.eviction_config = eviction_config.into();
428        self
429    }
430
431    /// Set in-memory cache hash builder.
432    pub fn with_hash_builder<OS>(self, hash_builder: OS) -> CacheBuilder<K, V, OS>
433    where
434        OS: HashBuilder,
435    {
436        CacheBuilder {
437            name: self.name,
438            capacity: self.capacity,
439            shards: self.shards,
440            eviction_config: self.eviction_config,
441            hash_builder,
442            weighter: self.weighter,
443            event_listener: self.event_listener,
444            registry: self.registry,
445            metrics: self.metrics,
446        }
447    }
448
449    /// Set in-memory cache weighter.
450    pub fn with_weighter(mut self, weighter: impl Weighter<K, V>) -> Self {
451        self.weighter = Arc::new(weighter);
452        self
453    }
454
455    /// Set event listener.
456    pub fn with_event_listener(mut self, event_listener: Arc<dyn EventListener<Key = K, Value = V>>) -> Self {
457        self.event_listener = Some(event_listener);
458        self
459    }
460
461    /// Set metrics registry.
462    ///
463    /// Default: [`NoopMetricsRegistry`].
464    pub fn with_metrics_registry(mut self, registry: BoxedRegistry) -> CacheBuilder<K, V, S> {
465        self.registry = registry;
466        self
467    }
468
469    /// Set metrics.
470    ///
471    /// Note: `with_metrics` is only supposed to be called by other foyer components.
472    #[doc(hidden)]
473    pub fn with_metrics(mut self, metrics: Arc<Metrics>) -> Self {
474        self.metrics = Some(metrics);
475        self
476    }
477
478    /// Build in-memory cache with the given configuration.
479    pub fn build<P>(self) -> Cache<K, V, S, P>
480    where
481        P: Properties,
482    {
483        if self.capacity < self.shards {
484            tracing::warn!(
485                "The in-memory cache capacity({}) < shards({}).",
486                self.capacity,
487                self.shards
488            );
489        }
490
491        let metrics = self
492            .metrics
493            .unwrap_or_else(|| Arc::new(Metrics::new(self.name, &self.registry)));
494
495        match self.eviction_config {
496            EvictionConfig::Fifo(eviction_config) => Cache::Fifo(Arc::new(RawCache::new(RawCacheConfig {
497                capacity: self.capacity,
498                shards: self.shards,
499                eviction_config,
500                hash_builder: self.hash_builder,
501                weighter: self.weighter,
502                event_listener: self.event_listener,
503                metrics,
504            }))),
505            EvictionConfig::S3Fifo(eviction_config) => Cache::S3Fifo(Arc::new(RawCache::new(RawCacheConfig {
506                capacity: self.capacity,
507                shards: self.shards,
508                eviction_config,
509                hash_builder: self.hash_builder,
510                weighter: self.weighter,
511                event_listener: self.event_listener,
512                metrics,
513            }))),
514            EvictionConfig::Lru(eviction_config) => Cache::Lru(Arc::new(RawCache::new(RawCacheConfig {
515                capacity: self.capacity,
516                shards: self.shards,
517                eviction_config,
518                hash_builder: self.hash_builder,
519                weighter: self.weighter,
520                event_listener: self.event_listener,
521                metrics,
522            }))),
523            EvictionConfig::Lfu(eviction_config) => Cache::Lfu(Arc::new(RawCache::new(RawCacheConfig {
524                capacity: self.capacity,
525                shards: self.shards,
526                eviction_config,
527                hash_builder: self.hash_builder,
528                weighter: self.weighter,
529                event_listener: self.event_listener,
530                metrics,
531            }))),
532        }
533    }
534}
535
536/// In-memory cache with plug-and-play algorithms.
537pub enum Cache<K, V, S = DefaultHasher, P = CacheProperties>
538where
539    K: Key,
540    V: Value,
541    S: HashBuilder,
542    P: Properties,
543{
544    /// In-memory FIFO cache.
545    Fifo(Arc<FifoCache<K, V, S, P>>),
546    /// In-memory LRU cache.
547    Lru(Arc<LruCache<K, V, S, P>>),
548    /// In-memory LFU cache.
549    Lfu(Arc<LfuCache<K, V, S, P>>),
550    /// In-memory S3FIFO cache.
551    S3Fifo(Arc<S3FifoCache<K, V, S, P>>),
552}
553
554impl<K, V, S, P> Debug for Cache<K, V, S, P>
555where
556    K: Key,
557    V: Value,
558    S: HashBuilder,
559    P: Properties,
560{
561    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
562        match self {
563            Self::Fifo(_) => f.debug_tuple("Cache::FifoCache").finish(),
564            Self::S3Fifo(_) => f.debug_tuple("Cache::S3FifoCache").finish(),
565            Self::Lru(_) => f.debug_tuple("Cache::LruCache").finish(),
566            Self::Lfu(_) => f.debug_tuple("Cache::LfuCache").finish(),
567        }
568    }
569}
570
571impl<K, V, S, P> Clone for Cache<K, V, S, P>
572where
573    K: Key,
574    V: Value,
575    S: HashBuilder,
576    P: Properties,
577{
578    fn clone(&self) -> Self {
579        match self {
580            Self::Fifo(cache) => Self::Fifo(cache.clone()),
581            Self::S3Fifo(cache) => Self::S3Fifo(cache.clone()),
582            Self::Lru(cache) => Self::Lru(cache.clone()),
583            Self::Lfu(cache) => Self::Lfu(cache.clone()),
584        }
585    }
586}
587
588impl<K, V> Cache<K, V, DefaultHasher, CacheProperties>
589where
590    K: Key,
591    V: Value,
592{
593    /// Create a new in-memory cache builder with capacity.
594    pub fn builder(capacity: usize) -> CacheBuilder<K, V, DefaultHasher> {
595        CacheBuilder::new(capacity)
596    }
597}
598
599impl<K, V, S, P> Cache<K, V, S, P>
600where
601    K: Key,
602    V: Value,
603    S: HashBuilder,
604    P: Properties,
605{
606    /// Update capacity and evict overflowed entries.
607    #[cfg_attr(feature = "tracing", fastrace::trace(name = "foyer::memory::cache::resize"))]
608    pub fn resize(&self, capacity: usize) -> Result<()> {
609        match self {
610            Cache::Fifo(cache) => cache.resize(capacity),
611            Cache::S3Fifo(cache) => cache.resize(capacity),
612            Cache::Lru(cache) => cache.resize(capacity),
613            Cache::Lfu(cache) => cache.resize(capacity),
614        }
615    }
616
617    /// Insert cache entry to the in-memory cache.
618    #[cfg_attr(feature = "tracing", fastrace::trace(name = "foyer::memory::cache::insert"))]
619    pub fn insert(&self, key: K, value: V) -> CacheEntry<K, V, S, P> {
620        match self {
621            Cache::Fifo(cache) => cache.insert(key, value).into(),
622            Cache::S3Fifo(cache) => cache.insert(key, value).into(),
623            Cache::Lru(cache) => cache.insert(key, value).into(),
624            Cache::Lfu(cache) => cache.insert(key, value).into(),
625        }
626    }
627
628    /// Insert cache entry to the in-memory cache with properties.
629    #[cfg_attr(
630        feature = "tracing",
631        fastrace::trace(name = "foyer::memory::cache::insert_with_properties")
632    )]
633    pub fn insert_with_properties(&self, key: K, value: V, properties: P) -> CacheEntry<K, V, S, P> {
634        match self {
635            Cache::Fifo(cache) => cache.insert_with_properties(key, value, properties).into(),
636            Cache::S3Fifo(cache) => cache.insert_with_properties(key, value, properties).into(),
637            Cache::Lru(cache) => cache.insert_with_properties(key, value, properties).into(),
638            Cache::Lfu(cache) => cache.insert_with_properties(key, value, properties).into(),
639        }
640    }
641
642    /// Remove a cached entry with the given key from the in-memory cache.
643    #[cfg_attr(feature = "tracing", fastrace::trace(name = "foyer::memory::cache::remove"))]
644    pub fn remove<Q>(&self, key: &Q) -> Option<CacheEntry<K, V, S, P>>
645    where
646        Q: Hash + Equivalent<K> + ?Sized,
647    {
648        match self {
649            Cache::Fifo(cache) => cache.remove(key).map(CacheEntry::from),
650            Cache::S3Fifo(cache) => cache.remove(key).map(CacheEntry::from),
651            Cache::Lru(cache) => cache.remove(key).map(CacheEntry::from),
652            Cache::Lfu(cache) => cache.remove(key).map(CacheEntry::from),
653        }
654    }
655
656    /// Get cached entry with the given key from the in-memory cache.
657    #[cfg_attr(feature = "tracing", fastrace::trace(name = "foyer::memory::cache::get"))]
658    pub fn get<Q>(&self, key: &Q) -> Option<CacheEntry<K, V, S, P>>
659    where
660        Q: Hash + Equivalent<K> + ?Sized,
661    {
662        match self {
663            Cache::Fifo(cache) => cache.get(key).map(CacheEntry::from),
664            Cache::S3Fifo(cache) => cache.get(key).map(CacheEntry::from),
665            Cache::Lru(cache) => cache.get(key).map(CacheEntry::from),
666            Cache::Lfu(cache) => cache.get(key).map(CacheEntry::from),
667        }
668    }
669
670    /// Check if the in-memory cache contains a cached entry with the given key.
671    #[cfg_attr(feature = "tracing", fastrace::trace(name = "foyer::memory::cache::contains"))]
672    pub fn contains<Q>(&self, key: &Q) -> bool
673    where
674        Q: Hash + Equivalent<K> + ?Sized,
675    {
676        match self {
677            Cache::Fifo(cache) => cache.contains(key),
678            Cache::S3Fifo(cache) => cache.contains(key),
679            Cache::Lru(cache) => cache.contains(key),
680            Cache::Lfu(cache) => cache.contains(key),
681        }
682    }
683
684    /// Access the cached entry with the given key but don't return.
685    ///
686    /// Note: This method can be used to update the cache eviction information and order based on the algorithm.
687    #[cfg_attr(feature = "tracing", fastrace::trace(name = "foyer::memory::cache::touch"))]
688    pub fn touch<Q>(&self, key: &Q) -> bool
689    where
690        Q: Hash + Equivalent<K> + ?Sized,
691    {
692        match self {
693            Cache::Fifo(cache) => cache.touch(key),
694            Cache::S3Fifo(cache) => cache.touch(key),
695            Cache::Lru(cache) => cache.touch(key),
696            Cache::Lfu(cache) => cache.touch(key),
697        }
698    }
699
700    /// Clear the in-memory cache.
701    #[cfg_attr(feature = "tracing", fastrace::trace(name = "foyer::memory::cache::clear"))]
702    pub fn clear(&self) {
703        match self {
704            Cache::Fifo(cache) => cache.clear(),
705            Cache::S3Fifo(cache) => cache.clear(),
706            Cache::Lru(cache) => cache.clear(),
707            Cache::Lfu(cache) => cache.clear(),
708        }
709    }
710
711    /// Get the capacity of the in-memory cache.
712    pub fn capacity(&self) -> usize {
713        match self {
714            Cache::Fifo(cache) => cache.capacity(),
715            Cache::S3Fifo(cache) => cache.capacity(),
716            Cache::Lru(cache) => cache.capacity(),
717            Cache::Lfu(cache) => cache.capacity(),
718        }
719    }
720
721    /// Get the usage of the in-memory cache.
722    pub fn usage(&self) -> usize {
723        match self {
724            Cache::Fifo(cache) => cache.usage(),
725            Cache::S3Fifo(cache) => cache.usage(),
726            Cache::Lru(cache) => cache.usage(),
727            Cache::Lfu(cache) => cache.usage(),
728        }
729    }
730
731    /// Hash the given key with the hash builder of the cache.
732    pub fn hash<Q>(&self, key: &Q) -> u64
733    where
734        Q: Hash + ?Sized,
735    {
736        self.hash_builder().hash_one(key)
737    }
738
739    /// Get the hash builder of the in-memory cache.
740    pub fn hash_builder(&self) -> &Arc<S> {
741        match self {
742            Cache::Fifo(cache) => cache.hash_builder(),
743            Cache::S3Fifo(cache) => cache.hash_builder(),
744            Cache::Lru(cache) => cache.hash_builder(),
745            Cache::Lfu(cache) => cache.hash_builder(),
746        }
747    }
748
749    /// Get the shards of the in-memory cache.
750    pub fn shards(&self) -> usize {
751        match self {
752            Cache::Fifo(cache) => cache.shards(),
753            Cache::S3Fifo(cache) => cache.shards(),
754            Cache::Lru(cache) => cache.shards(),
755            Cache::Lfu(cache) => cache.shards(),
756        }
757    }
758
759    /// Set the pipe for the hybrid cache.
760    #[doc(hidden)]
761    pub fn set_pipe(&self, pipe: Box<dyn Pipe<Key = K, Value = V, Properties = P>>) {
762        match self {
763            Cache::Fifo(cache) => cache.set_pipe(pipe),
764            Cache::S3Fifo(cache) => cache.set_pipe(pipe),
765            Cache::Lru(cache) => cache.set_pipe(pipe),
766            Cache::Lfu(cache) => cache.set_pipe(pipe),
767        }
768    }
769
770    /// Evict all entries from the in-memory cache.
771    ///
772    /// Instead of [`Cache::clear`], [`Cache::evict_all`] will send the evicted pipe to the pipe.
773    /// It is useful when the cache is used as a hybrid cache.
774    pub fn evict_all(&self) {
775        match self {
776            Cache::Fifo(cache) => cache.evict_all(),
777            Cache::S3Fifo(cache) => cache.evict_all(),
778            Cache::Lru(cache) => cache.evict_all(),
779            Cache::Lfu(cache) => cache.evict_all(),
780        }
781    }
782
783    /// Evict all entries in the cache and offload them into the disk cache via the pipe if needed.
784    ///
785    /// This function obeys the io throttler of the disk cache and make sure all entries will be offloaded.
786    /// Therefore, this function is asynchronous.
787    #[cfg_attr(feature = "tracing", fastrace::trace(name = "foyer::memory::raw::offload"))]
788    pub async fn flush(&self) {
789        match self {
790            Cache::Fifo(cache) => cache.flush().await,
791            Cache::S3Fifo(cache) => cache.flush().await,
792            Cache::Lru(cache) => cache.flush().await,
793            Cache::Lfu(cache) => cache.flush().await,
794        }
795    }
796}
797
798/// A future that is used to get entry value from the remote storage for the in-memory cache.
799#[pin_project(project = FetchProj)]
800pub enum Fetch<K, V, ER, S = DefaultHasher, P = CacheProperties>
801where
802    K: Key,
803    V: Value,
804    S: HashBuilder,
805    P: Properties,
806{
807    /// A future that is used to get entry value from the remote storage for the in-memory FIFO cache.
808    Fifo(#[pin] FifoFetch<K, V, ER, S, P>),
809    /// A future that is used to get entry value from the remote storage for the in-memory S3FIFO cache.
810    S3Fifo(#[pin] S3FifoFetch<K, V, ER, S, P>),
811    /// A future that is used to get entry value from the remote storage for the in-memory LRU cache.
812    Lru(#[pin] LruFetch<K, V, ER, S, P>),
813    /// A future that is used to get entry value from the remote storage for the in-memory LFU cache.
814    Lfu(#[pin] LfuFetch<K, V, ER, S, P>),
815}
816
817impl<K, V, ER, S, P> From<FifoFetch<K, V, ER, S, P>> for Fetch<K, V, ER, S, P>
818where
819    K: Key,
820    V: Value,
821    S: HashBuilder,
822    P: Properties,
823{
824    fn from(entry: FifoFetch<K, V, ER, S, P>) -> Self {
825        Self::Fifo(entry)
826    }
827}
828
829impl<K, V, ER, S, P> From<S3FifoFetch<K, V, ER, S, P>> for Fetch<K, V, ER, S, P>
830where
831    K: Key,
832    V: Value,
833    S: HashBuilder,
834    P: Properties,
835{
836    fn from(entry: S3FifoFetch<K, V, ER, S, P>) -> Self {
837        Self::S3Fifo(entry)
838    }
839}
840
841impl<K, V, ER, S, P> From<LruFetch<K, V, ER, S, P>> for Fetch<K, V, ER, S, P>
842where
843    K: Key,
844    V: Value,
845    S: HashBuilder,
846    P: Properties,
847{
848    fn from(entry: LruFetch<K, V, ER, S, P>) -> Self {
849        Self::Lru(entry)
850    }
851}
852
853impl<K, V, ER, S, P> From<LfuFetch<K, V, ER, S, P>> for Fetch<K, V, ER, S, P>
854where
855    K: Key,
856    V: Value,
857    S: HashBuilder,
858    P: Properties,
859{
860    fn from(entry: LfuFetch<K, V, ER, S, P>) -> Self {
861        Self::Lfu(entry)
862    }
863}
864
865impl<K, V, ER, S, P> Future for Fetch<K, V, ER, S, P>
866where
867    K: Key,
868    V: Value,
869    ER: From<oneshot::error::RecvError>,
870    S: HashBuilder,
871    P: Properties,
872{
873    type Output = std::result::Result<CacheEntry<K, V, S, P>, ER>;
874
875    fn poll(self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> std::task::Poll<Self::Output> {
876        match self.project() {
877            FetchProj::Fifo(entry) => entry.poll(cx).map(|res| res.map(CacheEntry::from)),
878            FetchProj::S3Fifo(entry) => entry.poll(cx).map(|res| res.map(CacheEntry::from)),
879            FetchProj::Lru(entry) => entry.poll(cx).map(|res| res.map(CacheEntry::from)),
880            FetchProj::Lfu(entry) => entry.poll(cx).map(|res| res.map(CacheEntry::from)),
881        }
882    }
883}
884
885impl<K, V, ER, S, P> Fetch<K, V, ER, S, P>
886where
887    K: Key,
888    V: Value,
889    S: HashBuilder,
890    P: Properties,
891{
892    /// Get the fetch state.
893    pub fn state(&self) -> FetchState {
894        match self {
895            Fetch::Fifo(fetch) => fetch.state(),
896            Fetch::S3Fifo(fetch) => fetch.state(),
897            Fetch::Lru(fetch) => fetch.state(),
898            Fetch::Lfu(fetch) => fetch.state(),
899        }
900    }
901
902    /// Get the ext of the fetch.
903    #[doc(hidden)]
904    pub fn store(&self) -> &Option<FetchContext> {
905        match self {
906            Fetch::Fifo(fetch) => fetch.store(),
907            Fetch::S3Fifo(fetch) => fetch.store(),
908            Fetch::Lru(fetch) => fetch.store(),
909            Fetch::Lfu(fetch) => fetch.store(),
910        }
911    }
912}
913
914impl<K, V, S, P> Cache<K, V, S, P>
915where
916    K: Key + Clone,
917    V: Value,
918    S: HashBuilder,
919    P: Properties,
920{
921    /// Get the cached entry with the given key from the in-memory cache.
922    ///
923    /// Use `fetch` to fetch the cache value from the remote storage on cache miss.
924    ///
925    /// The concurrent fetch requests will be deduplicated.
926    #[cfg_attr(feature = "tracing", fastrace::trace(name = "foyer::memory::cache::fetch"))]
927    pub fn fetch<F, FU, ER>(&self, key: K, fetch: F) -> Fetch<K, V, ER, S, P>
928    where
929        F: FnOnce() -> FU,
930        FU: Future<Output = std::result::Result<V, ER>> + Send + 'static,
931        ER: Send + 'static + Debug,
932    {
933        match self {
934            Cache::Fifo(cache) => Fetch::from(cache.fetch(key, fetch)),
935            Cache::S3Fifo(cache) => Fetch::from(cache.fetch(key, fetch)),
936            Cache::Lru(cache) => Fetch::from(cache.fetch(key, fetch)),
937            Cache::Lfu(cache) => Fetch::from(cache.fetch(key, fetch)),
938        }
939    }
940
941    /// Get the cached entry with the given key and properties from the in-memory cache.
942    ///
943    /// Use `fetch` to fetch the cache value from the remote storage on cache miss.
944    ///
945    /// The concurrent fetch requests will be deduplicated.
946    #[cfg_attr(feature = "tracing", fastrace::trace(name = "foyer::memory::cache::fetch"))]
947    pub fn fetch_with_properties<F, FU, ER>(&self, key: K, properties: P, fetch: F) -> Fetch<K, V, ER, S, P>
948    where
949        F: FnOnce() -> FU,
950        FU: Future<Output = std::result::Result<V, ER>> + Send + 'static,
951        ER: Send + 'static + Debug,
952    {
953        match self {
954            Cache::Fifo(cache) => Fetch::from(cache.fetch_with_properties(key, properties, fetch)),
955            Cache::S3Fifo(cache) => Fetch::from(cache.fetch_with_properties(key, properties, fetch)),
956            Cache::Lru(cache) => Fetch::from(cache.fetch_with_properties(key, properties, fetch)),
957            Cache::Lfu(cache) => Fetch::from(cache.fetch_with_properties(key, properties, fetch)),
958        }
959    }
960
961    /// Get the cached entry with the given key from the in-memory cache.
962    ///
963    /// The fetch task will be spawned in the give `runtime`.
964    ///
965    /// Use `fetch` to fetch the cache value from the remote storage on cache miss.
966    ///
967    /// The concurrent fetch requests will be deduplicated.
968    #[doc(hidden)]
969    pub fn fetch_inner<F, FU, ER, ID>(
970        &self,
971        key: K,
972        properties: P,
973        fetch: F,
974        runtime: &SingletonHandle,
975    ) -> Fetch<K, V, ER, S, P>
976    where
977        F: FnOnce() -> FU,
978        FU: Future<Output = ID> + Send + 'static,
979        ER: Send + 'static + Debug,
980        ID: Into<Diversion<std::result::Result<V, ER>, FetchContext>>,
981    {
982        match self {
983            Cache::Fifo(cache) => Fetch::from(cache.fetch_inner(key, properties, fetch, runtime)),
984            Cache::Lru(cache) => Fetch::from(cache.fetch_inner(key, properties, fetch, runtime)),
985            Cache::Lfu(cache) => Fetch::from(cache.fetch_inner(key, properties, fetch, runtime)),
986            Cache::S3Fifo(cache) => Fetch::from(cache.fetch_inner(key, properties, fetch, runtime)),
987        }
988    }
989}
990
991#[cfg(test)]
992mod tests {
993    use std::{ops::Range, time::Duration};
994
995    use futures_util::future::join_all;
996    use itertools::Itertools;
997    use rand::{rngs::StdRng, seq::SliceRandom, Rng, SeedableRng};
998
999    use super::*;
1000    use crate::eviction::{fifo::FifoConfig, lfu::LfuConfig, lru::LruConfig, s3fifo::S3FifoConfig};
1001
1002    const CAPACITY: usize = 100;
1003    const SHARDS: usize = 4;
1004    const RANGE: Range<u64> = 0..1000;
1005    const OPS: usize = 10000;
1006    const CONCURRENCY: usize = 8;
1007
1008    fn fifo() -> Cache<u64, u64> {
1009        CacheBuilder::new(CAPACITY)
1010            .with_shards(SHARDS)
1011            .with_eviction_config(FifoConfig {})
1012            .build()
1013    }
1014
1015    fn lru() -> Cache<u64, u64> {
1016        CacheBuilder::new(CAPACITY)
1017            .with_shards(SHARDS)
1018            .with_eviction_config(LruConfig {
1019                high_priority_pool_ratio: 0.1,
1020            })
1021            .build()
1022    }
1023
1024    fn lfu() -> Cache<u64, u64> {
1025        CacheBuilder::new(CAPACITY)
1026            .with_shards(SHARDS)
1027            .with_eviction_config(LfuConfig {
1028                window_capacity_ratio: 0.1,
1029                protected_capacity_ratio: 0.8,
1030                cmsketch_eps: 0.001,
1031                cmsketch_confidence: 0.9,
1032            })
1033            .build()
1034    }
1035
1036    fn s3fifo() -> Cache<u64, u64> {
1037        CacheBuilder::new(CAPACITY)
1038            .with_shards(SHARDS)
1039            .with_eviction_config(S3FifoConfig {
1040                small_queue_capacity_ratio: 0.1,
1041                ghost_queue_capacity_ratio: 10.0,
1042                small_to_main_freq_threshold: 2,
1043            })
1044            .build()
1045    }
1046
1047    fn init_cache(cache: &Cache<u64, u64>, rng: &mut StdRng) {
1048        let mut v = RANGE.collect_vec();
1049        v.shuffle(rng);
1050        for i in v {
1051            cache.insert(i, i);
1052        }
1053    }
1054
1055    async fn operate(cache: &Cache<u64, u64>, rng: &mut StdRng) {
1056        let i = rng.random_range(RANGE);
1057        match rng.random_range(0..=3) {
1058            0 => {
1059                let entry = cache.insert(i, i);
1060                assert_eq!(*entry.key(), i);
1061                assert_eq!(entry.key(), entry.value());
1062            }
1063            1 => {
1064                if let Some(entry) = cache.get(&i) {
1065                    assert_eq!(*entry.key(), i);
1066                    assert_eq!(entry.key(), entry.value());
1067                }
1068            }
1069            2 => {
1070                cache.remove(&i);
1071            }
1072            3 => {
1073                let entry = cache
1074                    .fetch(i, || async move {
1075                        tokio::time::sleep(Duration::from_micros(10)).await;
1076                        Ok::<_, tokio::sync::oneshot::error::RecvError>(i)
1077                    })
1078                    .await
1079                    .unwrap();
1080                assert_eq!(*entry.key(), i);
1081                assert_eq!(entry.key(), entry.value());
1082            }
1083            _ => unreachable!(),
1084        }
1085    }
1086
1087    async fn case(cache: Cache<u64, u64>) {
1088        let mut rng = StdRng::seed_from_u64(42);
1089
1090        init_cache(&cache, &mut rng);
1091
1092        let handles = (0..CONCURRENCY)
1093            .map(|_| {
1094                let cache = cache.clone();
1095                let mut rng = rng.clone();
1096                tokio::spawn(async move {
1097                    for _ in 0..OPS {
1098                        operate(&cache, &mut rng).await;
1099                    }
1100                })
1101            })
1102            .collect_vec();
1103
1104        join_all(handles).await;
1105    }
1106
1107    #[tokio::test]
1108    async fn test_fifo_cache() {
1109        case(fifo()).await
1110    }
1111
1112    #[tokio::test]
1113    async fn test_lru_cache() {
1114        case(lru()).await
1115    }
1116
1117    #[tokio::test]
1118    async fn test_lfu_cache() {
1119        case(lfu()).await
1120    }
1121
1122    #[tokio::test]
1123    async fn test_s3fifo_cache() {
1124        case(s3fifo()).await
1125    }
1126}
foyer_memory/cache.rs

foyer_memory/
cache.rs
