Iteration with Expiration state

[peter-scholtens]

Similar as mentioned in #235, I try to store and read-in my Cache content locally during a re-start of my server. However, this means the expiration state will be reset: the frequency is set to zero and the TTL/TTI maximized. Would it be possible to iterate with Expiration state? So this state information can be stored too.

[tatsuya6502]

Hi. Currently (as of moka v0.11.x), there is no way to get and set the frequency and TTL/TTI timers for cached entry. I will create feature requests (GitHub Issues) for them.

For now, I think you can implement TTL/TTI part by using per-entry expiration policy. I will try to write a sample code later, but the basic idea will be to store your value with the last modified and last accessed timestamps.

// You may need to use an async lock if you use `future::Cache` of
// v0.12.0 or newer, but I am not sure yet.
use std::sync::RwLock;
// You cannot use `std::time::Instant` as it cannot be converted to/
// from the wall-clock time.
use time::OffsetDateTime;

pub struct MyValue<V> {
    value: V,
    last_modified: OffsetDateTime,
    last_accessed: RwLock<OffsetDateTime>,
}

You will set these timestamps when inserting, and update the last_accessed when Expiry::expire_after_read is called.

When restoring the cache entries, sort the values by their last_accessed ascending (older to newer), then insert them one by one. You will skip entries that have been already expired by TTL/TTI.

Let me know which cache type you are using: future::Cache or sync::Cache?

[tatsuya6502]

also a hasher implementation that can be serialized and deserialized

This can be achieved by using AHash as it provides a way to build a hasher from given seeds (u64 × 4). If we generate the seeds and save them, we can recreate the same hasher using them.

https://docs.rs/ahash/latest/ahash/random_state/struct.RandomState.html#method.with_seeds

[peter-scholtens]

Let me know which cache type you are using: future::Cache or sync::Cache?

For convenience (of understanding on my part) I use the sync::Cache but I wonder if that was the correct choice. Given that my webserver now processes a http PUT/GET request with the following sequence:

Axum (async) -> Moka (sync) -> Deadpool/Diesel (async)

I though switching threads make no sense when looking up the cache, it is all in the RAM, right?, but then I forgot the cache misses. So if I understand correctly, the penalty I built-in is switching heavy processes instead of lightweight threads. So it would be better to make all async, I guess.

[tatsuya6502]

Currently (as of moka v0.11.x), there is no way to get and set the frequency and TTL/TTI timers for cached entry. I will create feature requests (GitHub Issues) for them.

OK. I created the followings:

• Provide a way to iterate entries with their metadata #312
• Provide a way to get a read-only snapshot of the FrequencySketch of Cache #313
• Provide a way to restore a Cache from entries with metadata and a FrequencySketch snapshot #314

[tatsuya6502]

For now, I think you can implement TTL/TTI part by using per-entry expiration policy. I will try to write a sample code later

Here is the example. It uses future::Cache but I believe it will also work with sync::Cache by removing .await.

// Cargo.toml
//
// [dependencies]
// async-trait = "0.1.73"
// moka = { version = "0.11.3", features = ["future"] }
// time = "0.3.28"
// tokio = { version = "1.32.0", features = ["rt-multi-thread", "sync", "time", "macros"] }

use std::{
    sync::{
        atomic::{AtomicBool, Ordering},
        Arc, Mutex,
    },
    time::{Duration, Instant},
};

use moka::{future::Cache, Expiry};
use time::OffsetDateTime;

/// A value that is stored in the cache.
#[derive(Debug)]
pub struct MyValue<V> {
    pub value: V,
    last_modified: OffsetDateTime,
    last_accessed: Arc<Mutex<OffsetDateTime>>,
    /// `true` if this value is newly created. `false` if this value is
    /// backed up and restored.
    is_new: AtomicBool,
}

impl<V> MyValue<V>
where
    V: Clone + std::fmt::Debug,
{
    pub fn new(value: V) -> Self {
        let now = OffsetDateTime::now_utc();
        Self {
            value,
            last_modified: now,
            last_accessed: Arc::new(Mutex::new(now)),
            is_new: AtomicBool::new(true),
        }
    }
}

impl<V> Clone for MyValue<V>
where
    V: Clone,
{
    fn clone(&self) -> Self {
        Self {
            value: self.value.clone(),
            last_modified: self.last_modified,
            last_accessed: Arc::clone(&self.last_accessed),
            is_new: AtomicBool::new(self.is_new.load(Ordering::Acquire)),
        }
    }
}

/// A custom expiry policy.
#[derive(Clone)]
pub struct MyExpiry {
    ttl: Duration,
    tti: Duration,
    base_instant: Instant,
    base_datetime: OffsetDateTime,
}

impl MyExpiry {
    pub fn new(ttl: Duration, tti: Duration) -> Self {
        let (base_instant, base_datetime) = (Instant::now(), OffsetDateTime::now_utc());
        Self {
            ttl,
            tti,
            base_instant,
            base_datetime,
        }
    }
}

impl<K, V> Expiry<K, MyValue<V>> for MyExpiry
where
    K: std::fmt::Display,
{
    fn expire_after_create(
        &self,
        key: &K,
        value: &MyValue<V>,
        current_time: Instant,
    ) -> Option<Duration> {
        let method = "expire_after_create";

        if value.is_new.load(Ordering::Acquire) {
            println!("{method} - key {key} is new. Set the expiry to the TTI");
            value.is_new.store(false, Ordering::Release);
            Some(self.tti)
        } else {
            println!("{method} - key {key} has been restored.");
            Some(self.remaining_duration(
                method,
                key,
                value.last_modified,
                *value.last_accessed.lock().unwrap(),
                current_time,
            ))
        }
    }

    fn expire_after_update(
        &self,
        key: &K,
        _value: &MyValue<V>,
        _current_time: Instant,
        _current_duration: Option<Duration>,
    ) -> Option<Duration> {
        println!("expire_after_update - Extending expiry of key {key} by the TTI");
        Some(self.tti)
    }

    fn expire_after_read(
        &self,
        key: &K,
        value: &MyValue<V>,
        current_time: Instant,
        _current_duration: Option<Duration>,
        _last_modified_at: Instant,
    ) -> Option<Duration> {
        let method = "expire_after_read";

        // Update the last accessed time.
        let new_last_accessed = self.instant_to_dt(current_time);
        *value.last_accessed.lock().unwrap() = new_last_accessed;

        Some(self.remaining_duration(
            method,
            key,
            value.last_modified,
            new_last_accessed,
            current_time,
        ))
    }
}

impl MyExpiry {
    fn remaining_duration<K>(
        &self,
        method_name: &str,
        key: &K,
        last_modified: OffsetDateTime,
        last_accessed: OffsetDateTime,
        current_time: Instant,
    ) -> Duration
    where
        K: std::fmt::Display,
    {
        let lm = self.dt_to_instant(last_modified);
        let la = self.dt_to_instant(last_accessed);
        let remaining_ttl = self.ttl.saturating_sub(current_time.duration_since(lm));
        let remaining_tti = self.tti.saturating_sub(current_time.duration_since(la));

        let zero = Duration::default();

        if remaining_tti == zero || remaining_ttl == zero {
            println!("{method_name} - Expiring key {key} now");
            zero
        } else if remaining_tti < remaining_ttl {
            println!(
                "{} - Extending expiry of key {} by the remaining duration until TTI ({:?})",
                method_name, key, remaining_tti,
            );
            remaining_tti
        } else {
            println!(
                "{} - Extending expiry of key {} by remaining duration until TTL ({:?})",
                method_name, key, remaining_ttl
            );
            remaining_ttl
        }
    }

    fn instant_to_dt(&self, instant: Instant) -> OffsetDateTime {
        let offset = instant.duration_since(self.base_instant);
        self.base_datetime + offset
    }

    fn dt_to_instant(&self, dt: OffsetDateTime) -> Instant {
        let offset = dt - self.base_datetime;
        self.base_instant + offset
    }
}

#[tokio::main]
async fn main() {
    const MAX_CAPACITY: u64 = 100;
    const TTL_SECS: u64 = 8;
    const TTI_SECS: u64 = 3;

    let expiry = MyExpiry::new(Duration::from_secs(TTL_SECS), Duration::from_secs(TTI_SECS));

    let cache = Cache::builder()
        .max_capacity(MAX_CAPACITY)
        .expire_after(expiry.clone())
        .build();

    println!("Started at {}", OffsetDateTime::now_utc());

    println!("\nInserting key 0 at {}", OffsetDateTime::now_utc());
    cache.insert(0, MyValue::new("Zero")).await; // expires 3s after start
    println!("\nInserting key 1 at {}", OffsetDateTime::now_utc());
    cache.insert(1, MyValue::new("One")).await; //  expires 3s after start
    println!("\nInserting key 2 at {}", OffsetDateTime::now_utc());
    cache.insert(2, MyValue::new("Two")).await; //  expires 3s after start

    tokio::time::sleep(Duration::from_secs(2)).await;
    println!("\n{} (2 seconds from start)", OffsetDateTime::now_utc());
    println!("\nGetting key 0 at {}", OffsetDateTime::now_utc());
    cache.get(&0); // expires 5s after start (2s + TTI)
    println!("\nGetting key 1 at {}", OffsetDateTime::now_utc());
    cache.get(&1); // expires 5s after start (2s + TTI)

    assert!(cache.contains_key(&0));
    assert!(cache.contains_key(&1));
    assert!(cache.contains_key(&2));
    println!("\nVerified all keys 0, 1 and 2 still exist");

    tokio::time::sleep(Duration::from_secs(2)).await;
    println!("\n{} (4 seconds from start)", OffsetDateTime::now_utc()); // 4 secs after start
    println!("\nGetting key 0 at {}", OffsetDateTime::now_utc());
    cache.get(&0); // expires 7s after start (4 + TTI)

    assert!(cache.contains_key(&0));
    assert!(cache.contains_key(&1));
    assert!(!cache.contains_key(&2)); // already expired
    println!("\nVerified key 2 has expired");

    println!("\nBacking up all entries");

    // Save the values to reinsert them later.
    let mut entries = cache.iter().map(|(k, v)| (*k, v)).collect::<Vec<_>>();

    drop(cache);

    println!("\nRecreating the cache");
    // Recreate the cache with the same expiry.
    let cache = Cache::builder()
        .max_capacity(MAX_CAPACITY)
        .expire_after(expiry)
        .build();

    // Sort the entries by last accessed time (older to newer).
    entries.sort_unstable_by_key(|(_, v)| *v.last_accessed.lock().unwrap());

    // Reinsert the entries with the same last modified and last accessed times.
    for (k, v) in entries {
        println!("\nReinserting key {} at {}", k, OffsetDateTime::now_utc());
        cache.insert(k, v).await;
    }

    println!("\n{} (4 seconds from start)", OffsetDateTime::now_utc());
    assert!(cache.contains_key(&0));
    assert!(cache.contains_key(&1));
    println!("\nVerified keys 0 and 1 still exist");

    tokio::time::sleep(Duration::from_secs(2)).await;
    println!("\n{} (6 seconds from start)", OffsetDateTime::now_utc());
    println!("\nGetting key 0 at {}", OffsetDateTime::now_utc());
    cache.get(&0); // expires 8s after start (TTL is used befause 6 + TTI > TTL)

    assert!(cache.contains_key(&0));
    assert!(!cache.contains_key(&1)); // already expired
    println!("\nVerified key 1 has expired");

    tokio::time::sleep(Duration::from_secs(3)).await;
    println!("\n{} (9 seconds from start)", OffsetDateTime::now_utc());
    assert!(!cache.contains_key(&0)); // already expired
    println!("\nVerified key 0 has expired");

    println!("\nDone");
}

EDIT: Updated the code; removed the generic parameter V from MyExpiry.

[tatsuya6502]

In general, we would recommend to use future::Cache in async code.

Axum (async) -> Moka (sync) -> Deadpool/Diesel (async)

Your code (request handler) for Axum is executed by an executor thread of Tokio. The same executor thread will run Moka's codes no matter if it is sync::Cache or future::Cache. Some API of the Cache will acquire a lock before calling user supplied closure, which can take some time to finish. If you use sync::Cache, the executor thread is blocked in such a case. I wrote some details here.

Up to Moka v0.11.x, future::Cache does not support some features that sync::Cache supports (e.g. disabling the thread pool), and I know some users had to use sync::Cache for that reason. But we are working on it and going to release v0.12.0 soon. (It is currently beta.2). Its future::Cache supports all features that sync::Cache supports.