Struct moka::sync::Cache

pub struct Cache<K, V, S = RandomState> { /* fields omitted */ }

A thread-safe concurrent in-memory cache.

Cache supports full concurrency of retrievals and a high expected concurrency for updates.

Cache utilizes a lock-free concurrent hash table SegmentedHashMap from the moka-cht crate for the central key-value storage. Cache performs a best-effort bounding of the map using an entry replacement algorithm to determine which entries to evict when the capacity is exceeded.

Examples

Cache entries are manually added using insert or get_or_insert_with method, and are stored in the cache until either evicted or manually invalidated.

Here’s an example of reading and updating a cache by using multiple threads:

use moka::sync::Cache;

use std::thread;

fn value(n: usize) -> String {
    format!("value {}", n)
}

const NUM_THREADS: usize = 16;
const NUM_KEYS_PER_THREAD: usize = 64;

// Create a cache that can store up to 10,000 entries.
let cache = Cache::new(10_000);

// Spawn threads and read and update the cache simultaneously.
let threads: Vec<_> = (0..NUM_THREADS)
    .map(|i| {
        // To share the same cache across the threads, clone it.
        // This is a cheap operation.
        let my_cache = cache.clone();
        let start = i * NUM_KEYS_PER_THREAD;
        let end = (i + 1) * NUM_KEYS_PER_THREAD;

        thread::spawn(move || {
            // Insert 64 entries. (NUM_KEYS_PER_THREAD = 64)
            for key in start..end {
                my_cache.insert(key, value(key));
                // get() returns Option<String>, a clone of the stored value.
                assert_eq!(my_cache.get(&key), Some(value(key)));
            }

            // Invalidate every 4 element of the inserted entries.
            for key in (start..end).step_by(4) {
                my_cache.invalidate(&key);
            }
        })
    })
    .collect();

// Wait for all threads to complete.
threads.into_iter().for_each(|t| t.join().expect("Failed"));

// Verify the result.
for key in 0..(NUM_THREADS * NUM_KEYS_PER_THREAD) {
    if key % 4 == 0 {
        assert_eq!(cache.get(&key), None);
    } else {
        assert_eq!(cache.get(&key), Some(value(key)));
    }
}

If you want to atomically initialize and insert a value when the key is not present, you might want to check other insertion methods get_or_insert_with and get_or_try_insert_with.

Avoiding to clone the value at get

The return type of get method is Option<V> instead of Option<&V>. Every time get is called for an existing key, it creates a clone of the stored value V and returns it. This is because the Cache allows concurrent updates from threads so a value stored in the cache can be dropped or replaced at any time by any other thread. get cannot return a reference &V as it is impossible to guarantee the value outlives the reference.

If you want to store values that will be expensive to clone, wrap them by std::sync::Arc before storing in a cache. Arc is a thread-safe reference-counted pointer and its clone() method is cheap.

Size-based Eviction

use std::convert::TryInto;
use moka::sync::Cache;

// Evict based on the number of entries in the cache.
let cache = Cache::builder()
    // Up to 10,000 entries.
    .max_capacity(10_000)
    // Create the cache.
    .build();
cache.insert(1, "one".to_string());

// Evict based on the byte length of strings in the cache.
let cache = Cache::builder()
    // A weigher closure takes &K and &V and returns a u32
    // representing the relative size of the entry.
    .weigher(|_key, value: &String| -> u32 {
        value.len().try_into().unwrap_or(u32::MAX)
    })
    // This cache will hold up to 32MiB of values.
    .max_capacity(32 * 1024 * 1024)
    .build();
cache.insert(2, "two".to_string());

If your cache should not grow beyond a certain size, use the max_capacity method of the CacheBuilder to set the upper bound. The cache will try to evict entries that have not been used recently or very often.

At the cache creation time, a weigher closure can be set by the weigher method of the CacheBuilder. A weigher closure takes &K and &V as the arguments and returns a u32 representing the relative size of the entry:

• If the weigher is not set, the cache will treat each entry has the same size of 1. This means the cache will be bounded by the number of entries.
• If the weigher is set, the cache will call the weigher to calculate the weighted size (relative size) on an entry. This means the cache will be bounded by the total weighted size of entries.

Note that weighted sizes are not used when making eviction selections.

Time-based Expirations

Cache supports the following expiration policies:

• Time to live: A cached entry will be expired after the specified duration past from insert.
• Time to idle: A cached entry will be expired after the specified duration past from get or insert.

use moka::sync::Cache;
use std::time::Duration;

let cache = Cache::builder()
    // Time to live (TTL): 30 minutes
    .time_to_live(Duration::from_secs(30 * 60))
    // Time to idle (TTI):  5 minutes
    .time_to_idle(Duration::from_secs( 5 * 60))
    // Create the cache.
    .build();

// This entry will expire after 5 minutes (TTI) if there is no get().
cache.insert(0, "zero");

// This get() will extend the entry life for another 5 minutes.
cache.get(&0);

// Even though we keep calling get(), the entry will expire
// after 30 minutes (TTL) from the insert().

Thread Safety

All methods provided by the Cache are considered thread-safe, and can be safely accessed by multiple concurrent threads.

• Cache<K, V, S> requires trait bounds Send, Sync and 'static for K (key), V (value) and S (hasher state).
• Cache<K, V, S> will implement Send and Sync.

Sharing a cache across threads

To share a cache across threads, do one of the followings:

• Create a clone of the cache by calling its clone method and pass it to other thread.
• Wrap the cache by a sync::OnceCell or sync::Lazy from once_cell create, and set it to a static variable.

Cloning is a cheap operation for Cache as it only creates thread-safe reference-counted pointers to the internal data structures.

Hashing Algorithm

By default, Cache uses a hashing algorithm selected to provide resistance against HashDoS attacks. It will be the same one used by std::collections::HashMap, which is currently SipHash 1-3.

While SipHash’s performance is very competitive for medium sized keys, other hashing algorithms will outperform it for small keys such as integers as well as large keys such as long strings. However those algorithms will typically not protect against attacks such as HashDoS.

The hashing algorithm can be replaced on a per-Cache basis using the build_with_hasher method of the CacheBuilder. Many alternative algorithms are available on crates.io, such as the aHash crate.

Implementations

impl<K, V> Cache<K, V, RandomState> where
    K: Hash + Eq + Send + Sync + 'static,
    V: Clone + Send + Sync + 'static, 

pub fn new(max_capacity: u64) -> Self

Constructs a new Cache<K, V> that will store up to the max_capacity.

To adjust various configuration knobs such as initial_capacity or time_to_live, use the CacheBuilder.

pub fn builder() -> CacheBuilder<K, V, Cache<K, V, RandomState>>

Returns a CacheBuilder, which can builds a Cache or SegmentedCache with various configuration knobs.

impl<K, V, S> Cache<K, V, S> where
    K: Hash + Eq + Send + Sync + 'static,
    V: Clone + Send + Sync + 'static,
    S: BuildHasher + Clone + Send + Sync + 'static, 

pub fn get<Q>(&self, key: &Q) -> Option<V> where
    Arc<K>: Borrow<Q>,
    Q: Hash + Eq + ?Sized, 

Returns a clone of the value corresponding to the key.

If you want to store values that will be expensive to clone, wrap them by std::sync::Arc before storing in a cache. Arc is a thread-safe reference-counted pointer and its clone() method is cheap.

The key may be any borrowed form of the cache’s key type, but Hash and Eq on the borrowed form must match those for the key type.

pub fn get_or_insert_with(&self, key: K, init: impl FnOnce() -> V) -> V

Ensures the value of the key exists by inserting the result of the init function if not exist, and returns a clone of the value.

This method prevents to evaluate the init closure multiple times on the same key even if the method is concurrently called by many threads; only one of the calls evaluates its closure, and other calls wait for that closure to complete.

Example

use moka::sync::Cache;
use std::{sync::Arc, thread, time::Duration};

const TEN_MIB: usize = 10 * 1024 * 1024; // 10MiB
let cache = Cache::new(100);

// Spawn four threads.
let threads: Vec<_> = (0..4_u8)
    .map(|task_id| {
        let my_cache = cache.clone();
        thread::spawn(move || {
            println!("Thread {} started.", task_id);

            // Try to insert and get the value for key1. Although all four
            // threads will call `get_or_insert_with` at the same time, the
            // `init` closure must be evaluated only once.
            let value = my_cache.get_or_insert_with("key1", || {
                println!("Thread {} inserting a value.", task_id);
                Arc::new(vec![0u8; TEN_MIB])
            });

            // Ensure the value exists now.
            assert_eq!(value.len(), TEN_MIB);
            thread::sleep(Duration::from_millis(10));
            assert!(my_cache.get(&"key1").is_some());

            println!("Thread {} got the value. (len: {})", task_id, value.len());
        })
    })
    .collect();

// Wait all threads to complete.
threads
    .into_iter()
    .for_each(|t| t.join().expect("Thread failed"));

Result

• The init closure was called exactly once by thread 1.
• Other threads were blocked until thread 1 inserted the value.

Thread 1 started.
Thread 0 started.
Thread 3 started.
Thread 2 started.
Thread 1 inserting a value.
Thread 2 got the value. (len: 10485760)
Thread 1 got the value. (len: 10485760)
Thread 0 got the value. (len: 10485760)
Thread 3 got the value. (len: 10485760)

Panics

This method panics when the init closure has been panicked. When it happens, only the caller whose init closure panicked will get the panic (e.g. only thread 1 in the above sample). If there are other calls in progress (e.g. thread 0, 2 and 3 above), this method will restart and resolve one of the remaining init closure.

pub fn get_or_try_insert_with<F, E>(&self, key: K, init: F) -> Result<V, Arc<E>> where
    F: FnOnce() -> Result<V, E>,
    E: Send + Sync + 'static, 

Try to ensure the value of the key exists by inserting an Ok result of the init closure if not exist, and returns a clone of the value or the Err returned by the closure.

This method prevents to evaluate the init closure multiple times on the same key even if the method is concurrently called by many threads; only one of the calls evaluates its closure (as long as these closures return the same error type), and other calls wait for that closure to complete.

Example

use moka::sync::Cache;
use std::{path::Path, time::Duration, thread};

/// This function tries to get the file size in bytes.
fn get_file_size(thread_id: u8, path: impl AsRef<Path>) -> Result<u64, std::io::Error> {
    println!("get_file_size() called by thread {}.", thread_id);
    Ok(std::fs::metadata(path)?.len())
}

let cache = Cache::new(100);

// Spawn four threads.
let threads: Vec<_> = (0..4_u8)
    .map(|thread_id| {
        let my_cache = cache.clone();
        thread::spawn(move || {
            println!("Thread {} started.", thread_id);

            // Try to insert and get the value for key1. Although all four
            // threads will call `get_or_try_insert_with` at the same time,
            // get_file_size() must be called only once.
            let value = my_cache.get_or_try_insert_with(
                "key1",
                || get_file_size(thread_id, "./Cargo.toml"),
            );

            // Ensure the value exists now.
            assert!(value.is_ok());
            thread::sleep(Duration::from_millis(10));
            assert!(my_cache.get(&"key1").is_some());

            println!(
                "Thread {} got the value. (len: {})",
                thread_id,
                value.unwrap()
            );
        })
    })
    .collect();

// Wait all threads to complete.
threads
    .into_iter()
    .for_each(|t| t.join().expect("Thread failed"));

Result

• get_file_size() was called exactly once by thread 1.
• Other threads were blocked until thread 1 inserted the value.

Thread 1 started.
Thread 2 started.
get_file_size() called by thread 1.
Thread 3 started.
Thread 0 started.
Thread 2 got the value. (len: 1466)
Thread 0 got the value. (len: 1466)
Thread 1 got the value. (len: 1466)
Thread 3 got the value. (len: 1466)

Panics

This method panics when the init closure has been panicked. When it happens, only the caller whose init closure panicked will get the panic (e.g. only thread 1 in the above sample). If there are other calls in progress (e.g. thread 0, 2 and 3 above), this method will restart and resolve one of the remaining init closure.

pub fn insert(&self, key: K, value: V)

Inserts a key-value pair into the cache.

If the cache has this key present, the value is updated.

pub fn invalidate<Q>(&self, key: &Q) where
    Arc<K>: Borrow<Q>,
    Q: Hash + Eq + ?Sized, 

Discards any cached value for the key.

The key may be any borrowed form of the cache’s key type, but Hash and Eq on the borrowed form must match those for the key type.

pub fn invalidate_all(&self)

Discards all cached values.

This method returns immediately and a background thread will evict all the cached values inserted before the time when this method was called. It is guaranteed that the get method must not return these invalidated values even if they have not been evicted.

Like the invalidate method, this method does not clear the historic popularity estimator of keys so that it retains the client activities of trying to retrieve an item.

pub fn invalidate_entries_if<F>(
    &self,
    predicate: F
) -> Result<PredicateId, PredicateError> where
    F: Fn(&K, &V) -> bool + Send + Sync + 'static, 

Discards cached values that satisfy a predicate.

invalidate_entries_if takes a closure that returns true or false. This method returns immediately and a background thread will apply the closure to each cached value inserted before the time when invalidate_entries_if was called. If the closure returns true on a value, that value will be evicted from the cache.

Also the get method will apply the closure to a value to determine if it should have been invalidated. Therefore, it is guaranteed that the get method must not return invalidated values.

Note that you must call CacheBuilder::support_invalidation_closures at the cache creation time as the cache needs to maintain additional internal data structures to support this method. Otherwise, calling this method will fail with a PredicateError::InvalidationClosuresDisabled.

Like the invalidate method, this method does not clear the historic popularity estimator of keys so that it retains the client activities of trying to retrieve an item.

pub fn max_capacity(&self) -> Option<usize>

Returns the max_capacity of this cache.

pub fn time_to_live(&self) -> Option<Duration>

Returns the time_to_live of this cache.

pub fn time_to_idle(&self) -> Option<Duration>

Returns the time_to_idle of this cache.

pub fn num_segments(&self) -> usize

Returns the number of internal segments of this cache.

Cache always returns 1.

Trait Implementations

impl<K: Clone, V: Clone, S: Clone> Clone for Cache<K, V, S>

fn clone(&self) -> Cache<K, V, S>

Returns a copy of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<K, V, S> ConcurrentCacheExt<K, V> for Cache<K, V, S> where
    K: Hash + Eq + Send + Sync + 'static,
    V: Send + Sync + 'static,
    S: BuildHasher + Clone + Send + Sync + 'static, 

fn sync(&self)

Performs any pending maintenance operations needed by the cache.

impl<K, V, S> Send for Cache<K, V, S> where
    K: Send + Sync,
    V: Send + Sync,
    S: Send, 

impl<K, V, S> Sync for Cache<K, V, S> where
    K: Send + Sync,
    V: Send + Sync,
    S: Sync,