Put a cache in front of a slow async lookup

You wrap a slow async function in a fibre_cache AsyncCache with a loader, a TTL and a capacity, then watch the output as repeated reads turn into hits, ten concurrent misses turn into one backend call and old entries expire and get evicted.

fibre_cache = "0"

Set up the project#

Start a new binary crate and add fibre_cache and tokio.

cargo new user_cache
cd user_cache

Put this in the [dependencies] section of Cargo.toml:

[dependencies]
fibre_cache = "0"
tokio = { version = "1", features = ["full"] }

fibre_cache turns on its tokio feature by default, which is what lets an async loader run without any extra setup. If you use a different runtime you pass your own spawner to the builder instead.

Write the slow lookup#

The thing we want to cache is a function that takes a user id and returns a name after a 300 ms wait. It prints a line every time it runs so you can see each trip to the "backend". Replace src/main.rs with this:

use std::time::{Duration, Instant};

async fn slow_lookup(user_id: u32) -> String {
    println!("  backend: loading user {user_id}");
    tokio::time::sleep(Duration::from_millis(300)).await;
    match user_id {
        1 => "Ada".to_string(),
        2 => "Grace".to_string(),
        _ => format!("user-{user_id}"),
    }
}

#[tokio::main]
async fn main() {
    let started = Instant::now();
    for id in [1, 1, 2, 1] {
        let name = slow_lookup(id).await;
        println!("user {id} is {name}");
    }
    println!("4 lookups took {:.1?}", started.elapsed());
}

Run it:

$ cargo run
  backend: loading user 1
user 1 is Ada
  backend: loading user 1
user 1 is Ada
  backend: loading user 2
user 2 is Grace
  backend: loading user 1
user 1 is Ada
4 lookups took 1.2s

Every request goes to the backend, including the three for user 1.

Put the cache in front of it#

Now build an AsyncCache whose loader calls slow_lookup. The loader returns a tuple of the value and its cost. Here every user costs 1, so the capacity is a count of entries.

use std::time::Duration;

use fibre_cache::{AsyncCache, CacheBuilder};

async fn slow_lookup(user_id: u32) -> String {
    println!("  backend: loading user {user_id}");
    tokio::time::sleep(Duration::from_millis(300)).await;
    match user_id {
        1 => "Ada".to_string(),
        2 => "Grace".to_string(),
        _ => format!("user-{user_id}"),
    }
}

fn build_cache() -> AsyncCache<u32, String> {
    CacheBuilder::default()
        .capacity(100)
        .shards(4)
        .maintenance_chance(16)
        .time_to_live(Duration::from_secs(2))
        .async_loader(|user_id: u32| async move { (slow_lookup(user_id).await, 1) })
        .build_async()
        .expect("cache config is valid")
}

#[tokio::main]
async fn main() {
    let cache = build_cache();

    for id in [1, 1, 2, 1] {
        let name = cache.fetch_with(&id).await;
        println!("user {id} is {name}");
    }

    let m = cache.metrics();
    println!("hits: {}, misses: {}", m.hits, m.misses);
}

Here is what each builder call does:

• capacity(100) caps the total cost of everything in the cache at 100.
• shards(4) splits the cache into 4 shards, each with its own lock and its own share of the capacity. The default is based on the CPU count, which is more shards than a cache of 100 entries needs.
• maintenance_chance(16) runs maintenance on roughly 1 in 16 inserts, on top of the background janitor.
• time_to_live(2s) makes every entry expire two seconds after it was loaded.
• async_loader(...) is what fetch_with calls when a key is missing.

fetch_with returns an Arc<String>. It looks the key up and, on a miss, runs the loader, stores the result and hands it back. Run it again:

  backend: loading user 1
user 1 is Ada
user 1 is Ada
  backend: loading user 2
user 2 is Grace
user 1 is Ada
hits: 2, misses: 2

The backend only runs for the first request for each user. The second and third reads of user 1 come straight from memory, which the metrics confirm with two hits and two misses.

Send many requests for the same key at once#

A plain "check the map, call the backend, insert" cache has a gap: if ten requests miss the same key at the same moment, all ten call the backend. fetch_with closes that gap. The first caller runs the loader and the others wait for its result.

AsyncCache is a cheap handle you can clone into each task. Add this to the end of main, just before the let m = cache.metrics(); line:

    let mut tasks = Vec::new();
    for _ in 0..10 {
        let cache = cache.clone();
        tasks.push(tokio::spawn(async move { cache.fetch_with(&42).await }));
    }
    for task in tasks {
        let name = task.await.expect("task panicked");
        assert_eq!(*name, "user-42");
    }
    println!("10 concurrent requests for user 42 are done");

Run it:

  backend: loading user 1
user 1 is Ada
user 1 is Ada
  backend: loading user 2
user 2 is Grace
user 1 is Ada
  backend: loading user 42
10 concurrent requests for user 42 are done
hits: 11, misses: 3

There is exactly one backend: loading user 42 line for ten requests. One task counts as the miss and the nine that waited for it count as hits, which is why the totals moved from 2 and 2 to 11 and 3.

Let entries expire#

The cache was built with a two-second TTL. Once an entry is older than that, the background janitor removes it: fetch (a plain lookup that never calls the loader) returns None and fetch_with goes back to the backend. Add this after the concurrent block, again before the metrics line:

    tokio::time::sleep(Duration::from_millis(2500)).await;
    println!("after the ttl, fetch(1) returns {:?}", cache.fetch(&1).await);
    let name = cache.fetch_with(&1).await;
    println!("user 1 is {name}");

Run it:

  backend: loading user 1
user 1 is Ada
user 1 is Ada
  backend: loading user 2
user 2 is Grace
user 1 is Ada
  backend: loading user 42
10 concurrent requests for user 42 are done
after the ttl, fetch(1) returns None
  backend: loading user 1
user 1 is Ada
hits: 11, misses: 5

User 1 is loaded a second time because its first value is more than two seconds old. The failed fetch and the reload both count as misses. If some values need a different lifetime from the rest, insert_with_ttl lets you set one per entry.

Keep the cache bounded#

The capacity is 100, so the cache never holds more than 100 users. When inserts push the total cost over it, the janitor evicts entries until the total is back under the limit. Which entries go is up to the eviction policy. The default is TinyLFU, which keeps keys that are requested often over keys that were seen once.

To see it, insert 250 users directly with insert (key, value and cost), give the janitor a moment to run and read the metrics. Add this before the metrics line:

    for id in 1000..1250 {
        cache.insert(id, format!("user-{id}"), 1).await;
    }
    tokio::time::sleep(Duration::from_millis(500)).await;
    let m = cache.metrics();
    println!("within capacity: {}, evicted for capacity: {}", m.current_cost <= 100, m.evicted_by_capacity > 0);

within capacity: true, evicted for capacity: true

The cost stays at or under 100 however many users you insert. evicted_by_capacity counts what the janitor removed to keep it there. You can swap in another policy with cache_policy_factory on the builder if your access pattern suits LRU, SIEVE or one of the others better.

The whole program#

Here is the finished src/main.rs in one piece:

use std::time::Duration;

use fibre_cache::{AsyncCache, CacheBuilder};

async fn slow_lookup(user_id: u32) -> String {
    println!("  backend: loading user {user_id}");
    tokio::time::sleep(Duration::from_millis(300)).await;
    match user_id {
        1 => "Ada".to_string(),
        2 => "Grace".to_string(),
        _ => format!("user-{user_id}"),
    }
}

fn build_cache() -> AsyncCache<u32, String> {
    CacheBuilder::default()
        .capacity(100)
        .shards(4)
        .maintenance_chance(16)
        .time_to_live(Duration::from_secs(2))
        .async_loader(|user_id: u32| async move { (slow_lookup(user_id).await, 1) })
        .build_async()
        .expect("cache config is valid")
}

#[tokio::main]
async fn main() {
    let cache = build_cache();

    for id in [1, 1, 2, 1] {
        let name = cache.fetch_with(&id).await;
        println!("user {id} is {name}");
    }

    let mut tasks = Vec::new();
    for _ in 0..10 {
        let cache = cache.clone();
        tasks.push(tokio::spawn(async move { cache.fetch_with(&42).await }));
    }
    for task in tasks {
        let name = task.await.expect("task panicked");
        assert_eq!(*name, "user-42");
    }
    println!("10 concurrent requests for user 42 are done");

    tokio::time::sleep(Duration::from_millis(2500)).await;
    println!("after the ttl, fetch(1) returns {:?}", cache.fetch(&1).await);
    let name = cache.fetch_with(&1).await;
    println!("user 1 is {name}");

    for id in 1000..1250 {
        cache.insert(id, format!("user-{id}"), 1).await;
    }
    tokio::time::sleep(Duration::from_millis(500)).await;
    let m = cache.metrics();
    println!("within capacity: {}, evicted for capacity: {}", m.current_cost <= 100, m.evicted_by_capacity > 0);

    println!("hits: {}, misses: {}", m.hits, m.misses);
}

Where to go next#

This tutorial used the async handle. If your code is synchronous, the same builder has loader and build(), which give you a blocking Cache with the same fetch_with behaviour. The usage guide on GitHub covers that along with stale-while-revalidate, time-to-idle, the entry API, eviction listeners, bulk operations and snapshots: https://github.com/excsn/fibre/blob/main/cache/README.USAGE.md

The full API is on docs.rs at https://docs.rs/fibre_cache. You can also build the docs locally for the exact version in your lockfile:

cargo doc -p fibre_cache --open