use core::{fmt::Display, hash::Hash};
use alloc::boxed::Box;
use alloc::string::String;
use alloc::vec::Vec;
use hashbrown::{HashMap, HashSet};
use serde::{Serialize, de::DeserializeOwned};
use super::namespace::Namespace;
use super::storage::{Insertion, Origin, Storage};
use crate::bytes::Bytes;
#[derive(Debug)]
pub enum StoreError<K, V> {
#[allow(missing_docs)]
DuplicatedKey {
key: K,
value_previous: V,
value_updated: V,
},
#[allow(missing_docs)]
KeyOutOfSync {
key: K,
value_previous: V,
value_updated: V,
},
#[allow(missing_docs)]
Backend { key: K, error: String },
}
impl<K, V> StoreError<K, V> {
pub fn reason(&self) -> &str {
match self {
Self::DuplicatedKey { .. } => "the key was already stored with a different value",
Self::KeyOutOfSync { .. } => "another process stored the key first",
Self::Backend { error, .. } => error,
}
}
}
impl<K: core::fmt::Debug, V: core::fmt::Debug> Display for StoreError<K, V> {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
match self {
Self::DuplicatedKey {
key,
value_previous,
value_updated,
} => write!(
f,
"key {key:?} was already stored with a different value: \
kept {value_previous:?}, dropped {value_updated:?}"
),
Self::KeyOutOfSync {
key,
value_previous,
value_updated,
} => write!(
f,
"key {key:?} was stored concurrently: kept {value_previous:?}, \
dropped {value_updated:?}"
),
Self::Backend { key, error } => {
write!(f, "storing key {key:?} failed: {error}")
}
}
}
}
impl<K: core::fmt::Debug, V: core::fmt::Debug> core::error::Error for StoreError<K, V> {}
pub trait StoreKey: Serialize + DeserializeOwned + PartialEq + Eq + Hash + Clone {}
pub trait StoreValue: Serialize + DeserializeOwned + PartialEq + Eq + Clone {}
impl<T: Serialize + DeserializeOwned + PartialEq + Eq + Clone + Hash> StoreKey for T {}
impl<T: Serialize + DeserializeOwned + PartialEq + Eq + Clone> StoreValue for T {}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum CacheOption {
#[default]
Eager,
Lazy,
}
#[derive(Debug, Default)]
enum StorageOption {
#[default]
InMemory,
Environment(Namespace),
Explicit(Box<dyn Storage>, Namespace),
}
#[derive(Debug, Default)]
pub struct StoreOptions {
storage: StorageOption,
cache: CacheOption,
}
impl StoreOptions {
pub fn new() -> Self {
Self::default()
}
pub fn storage<N: Into<Namespace>>(mut self, namespace: N) -> Self {
self.storage = StorageOption::Environment(namespace.into());
self
}
pub fn storage_with<N: Into<Namespace>>(
mut self,
storage: Box<dyn Storage>,
namespace: N,
) -> Self {
self.storage = StorageOption::Explicit(storage, namespace.into());
self
}
pub fn cache(mut self, cache: CacheOption) -> Self {
self.cache = cache;
self
}
}
pub struct Store<K, V> {
entries: HashMap<K, V>,
known: HashSet<K>,
storage: Option<Box<dyn Storage>>,
namespace: Option<Namespace>,
cache: CacheOption,
loaded: bool,
generation: Option<u32>,
}
impl<K: StoreKey, V: StoreValue> Store<K, V> {
#[cfg_attr(
feature = "tracing",
tracing::instrument(level = "trace", skip_all, fields(options = ?options))
)]
pub fn new(options: StoreOptions) -> Self {
let (storage, namespace, generation) = match options.storage {
StorageOption::InMemory => (None, None, None),
StorageOption::Environment(namespace) => {
let generation = crate::environment::generation();
(
Some(super::storage::open(namespace.as_str())),
Some(namespace),
Some(generation),
)
}
StorageOption::Explicit(storage, namespace) => (Some(storage), Some(namespace), None),
};
let mut store = Self {
entries: HashMap::new(),
known: HashSet::new(),
storage,
namespace,
cache: options.cache,
loaded: false,
generation,
};
match (store.cache, &store.storage) {
(CacheOption::Eager, Some(_)) => store.sync(),
_ => store.loaded = true,
}
store
}
pub fn namespace(&self) -> Option<&Namespace> {
self.namespace.as_ref()
}
pub fn get(&self, key: &K) -> Option<&V> {
if self.stale() {
return None;
}
self.entries.get(key)
}
pub fn get_mut(&mut self, key: &K) -> Option<&mut V> {
self.reset_if_stale();
self.refresh_if_pending();
if matches!(self.cache, CacheOption::Lazy)
&& !self.entries.contains_key(key)
&& let Some(value) = self.fetch(key)
{
self.entries.insert(key.clone(), value);
}
self.entries.get_mut(key)
}
pub fn remove(&mut self, key: &K) -> Option<V> {
self.reset_if_stale();
self.refresh_if_pending();
let value = match self.entries.remove(key) {
Some(value) => Some(value),
None => match self.cache {
CacheOption::Eager => None,
CacheOption::Lazy => self.fetch(key),
},
};
if value.is_some() && self.storage.is_some() {
self.known.insert(key.clone());
}
value
}
pub fn insert(&mut self, key: K, value: V) -> Result<(), StoreError<K, V>> {
self.reset_if_stale();
self.refresh_if_pending();
let known = match self.entries.get(&key) {
Some(existing) if existing == &value => return Ok(()),
existing => existing.is_some() || self.known.contains(&key),
};
let Some(storage) = self.storage.as_deref() else {
return match self.entries.get(&key) {
Some(existing) => Err(StoreError::DuplicatedKey {
value_previous: existing.clone(),
value_updated: value,
key,
}),
None => {
self.entries.insert(key, value);
Ok(())
}
};
};
match write_through(storage, &key, &value) {
Written::Stored => {
self.record(key, value);
Ok(())
}
Written::Failed(error) => Err(StoreError::Backend { key, error }),
Written::Conflict(existing) => {
if matches!(self.cache, CacheOption::Eager) {
self.entries.insert(key.clone(), existing.clone());
} else {
self.known.insert(key.clone());
}
let (value_previous, value_updated) = (existing, value);
Err(if known {
StoreError::DuplicatedKey {
key,
value_previous,
value_updated,
}
} else {
StoreError::KeyOutOfSync {
key,
value_previous,
value_updated,
}
})
}
}
}
pub fn purge_key(&mut self, key: &K) -> Option<V> {
self.reset_if_stale();
self.refresh_if_pending();
let value = match self.entries.remove(key) {
Some(value) => Some(value),
None => match self.cache {
CacheOption::Eager => None,
CacheOption::Lazy => self.fetch(key),
},
};
self.known.remove(key);
if let Some(storage) = self.storage.as_deref() {
storage.purge_key(&encode(key));
}
value
}
pub fn clear(&mut self) {
self.reset_if_stale();
if self.storage.is_some() {
self.known.extend(self.entries.drain().map(|(key, _)| key));
} else {
self.entries.clear();
}
}
pub fn purge(&mut self) {
self.reset_if_stale();
self.entries.clear();
self.known.clear();
if let Some(storage) = self.storage.as_deref() {
storage.purge();
}
}
#[cfg_attr(
feature = "tracing",
tracing::instrument(level = "trace", skip_all, fields(namespace = ?self.namespace))
)]
pub fn sync(&mut self) {
self.reset_if_stale();
let Some(storage) = self.storage.as_deref() else {
self.loaded = true;
return;
};
let loading = storage.loading();
let entries = &mut self.entries;
storage.scan(&mut |key, value| {
if let Some((key, value)) = decode_entry::<K, V>(key, value) {
entries.insert(key, value);
}
});
self.loaded = !loading;
}
pub fn pending_load(&self) -> bool {
!self.loaded || self.stale()
}
pub fn scan<F: FnMut(K, V)>(&mut self, mut func: F) -> bool {
self.reset_if_stale();
let Some(storage) = self.storage.as_deref() else {
for (key, value) in self.entries.iter() {
func(key.clone(), value.clone());
}
return true;
};
let loading = storage.loading();
storage.scan(&mut |key, value| {
if let Some((key, value)) = decode_entry::<K, V>(key, value) {
func(key, value);
}
});
!loading
}
pub fn for_each<F: FnMut(&K, &V)>(&self, mut func: F) {
if self.stale() {
return;
}
for (key, value) in self.entries.iter() {
func(key, value);
}
}
pub fn len(&self) -> usize {
if self.stale() {
return 0;
}
self.entries.len()
}
pub fn is_empty(&self) -> bool {
self.len() == 0
}
fn fetch(&self, key: &K) -> Option<V> {
let bytes = self.storage.as_deref()?.get(&encode(key))?;
decode::<V>(&bytes)
}
fn record(&mut self, key: K, value: V) {
match self.cache {
CacheOption::Eager => {
self.entries.insert(key, value);
}
CacheOption::Lazy => {
self.known.insert(key);
}
}
}
fn refresh_if_pending(&mut self) {
if !self.loaded {
self.sync();
}
}
fn stale(&self) -> bool {
match self.generation {
Some(generation) => generation != crate::environment::generation(),
None => false,
}
}
fn reset_if_stale(&mut self) {
if !self.stale() {
return;
}
let (Some(namespace), Some(_)) = (&self.namespace, self.generation) else {
return;
};
log::debug!("Environment switched, resetting the store for {namespace}");
self.generation = Some(crate::environment::generation());
self.storage = Some(super::storage::open(namespace.as_str()));
self.entries.clear();
self.known.clear();
self.loaded = matches!(self.cache, CacheOption::Lazy);
}
}
impl<K, V> core::fmt::Debug for Store<K, V> {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
f.debug_struct("Store")
.field("namespace", &self.namespace)
.field("cache", &self.cache)
.field("entries", &self.entries.len())
.field("known", &self.known.len())
.field("storage", &self.storage)
.field("loaded", &self.loaded)
.finish()
}
}
impl<K: StoreKey, V: StoreValue> Display for Store<K, V> {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
match (&self.namespace, &self.storage) {
(Some(namespace), Some(storage)) => write!(
f,
"{namespace} ({} entries in {})",
self.len(),
storage.describe()
),
_ => write!(f, "in-memory ({} entries)", self.len()),
}
}
}
pub(crate) enum Written<V> {
Stored,
Conflict(V),
Failed(String),
}
pub(crate) fn write_through<K: StoreKey, V: StoreValue>(
storage: &dyn Storage,
key: &K,
value: &V,
) -> Written<V> {
let key_bytes = encode(key);
match storage.insert(&key_bytes, encode(value), Origin::Local) {
Insertion::Stored => Written::Stored,
Insertion::Failed(error) => Written::Failed(error),
Insertion::Conflict(existing) => match decode::<V>(&existing) {
Some(existing) if &existing != value => Written::Conflict(existing),
Some(_) => Written::Stored,
None => match storage.replace(&key_bytes, encode(value), Origin::Local) {
Insertion::Failed(error) => Written::Failed(error),
_ => Written::Stored,
},
},
}
}
pub(crate) fn encode<T: Serialize>(value: &T) -> Bytes {
let mut bytes = Vec::new();
ciborium::ser::into_writer(value, &mut bytes).expect("Can serialize data");
Bytes::from_bytes_vec(bytes)
}
pub(crate) fn decode<T: DeserializeOwned>(bytes: &[u8]) -> Option<T> {
match ciborium::de::from_reader(bytes) {
Ok(value) => Some(value),
Err(err) => {
log::warn!("Corrupted cache entry, ignoring it: {err}");
None
}
}
}
fn decode_entry<K: StoreKey, V: StoreValue>(key: &[u8], value: &[u8]) -> Option<(K, V)> {
Some((decode::<K>(key)?, decode::<V>(value)?))
}
#[cfg(all(test, feature = "cache"))]
mod tests {
use std::string::ToString;
use std::vec;
use super::*;
fn eager(path: &str) -> StoreOptions {
StoreOptions::new().storage(Namespace::new(path))
}
fn lazy(path: &str) -> StoreOptions {
eager(path).cache(CacheOption::Lazy)
}
#[test_log::test]
#[serial_test::serial]
#[cfg_attr(miri, ignore)]
fn test_cache_simple() {
let dir = tempfile::tempdir().unwrap();
crate::environment::set_root(dir.path());
let key1 = || "key1".to_string();
let key2 = || "key2".to_string();
let value1 = || "value1".to_string();
let value2 = || "value2".to_string();
let mut cache = Store::<String, String>::new(eager("test"));
cache.insert(key1(), value1()).unwrap();
cache.insert(key2(), value2()).unwrap();
let result = cache.insert(key1(), value2());
assert!(
result.is_err(),
"Can't reinsert the same key with a different value."
);
assert_eq!(cache.len(), 2);
let value1_actual = cache.get(&key1()).unwrap();
assert_eq!(value1_actual, &value1());
let value2_actual = cache.get(&key2()).unwrap();
assert_eq!(value2_actual, &value2());
}
#[test_log::test]
#[serial_test::serial]
#[cfg_attr(miri, ignore)]
fn test_on_disk_format_is_stable() {
use super::super::sqlite::{Database, db_file_name};
let dir = tempfile::tempdir().unwrap();
crate::environment::set_root(dir.path());
let namespace = Namespace::scoped("golden", "device0/matmul");
let mut cache = Store::<String, u32>::new(StoreOptions::new().storage(namespace));
cache.insert("shape=2x2".to_string(), 42).unwrap();
let expected_namespace =
std::format!("golden/{}/device0/matmul", env!("CARGO_PKG_VERSION"));
assert_eq!(cache.namespace().unwrap().as_str(), expected_namespace);
let path = dir.path().join(db_file_name(&crate::environment::active()));
assert!(path.exists(), "Database missing at {path:?}");
let database = Database::open(&path, true).unwrap();
let stored = database
.get(&expected_namespace, &encode(&"shape=2x2".to_string()))
.expect("Entry should be stored");
assert_eq!(decode::<u32>(&stored), Some(42));
}
#[test_log::test]
#[serial_test::serial]
#[cfg_attr(miri, ignore)]
fn test_entries_survive_reopen() {
let dir = tempfile::tempdir().unwrap();
crate::environment::set_root(dir.path());
let mut cache = Store::<String, u32>::new(eager("reopen"));
cache.insert("key".to_string(), 7).unwrap();
drop(cache);
let cache = Store::<String, u32>::new(eager("reopen"));
assert_eq!(cache.get(&"key".to_string()), Some(&7));
}
#[test_log::test]
#[serial_test::serial]
#[cfg_attr(miri, ignore)]
fn test_stores_are_isolated() {
let dir = tempfile::tempdir().unwrap();
crate::environment::set_root(dir.path());
let mut first = Store::<String, u32>::new(eager("device0/matmul"));
first.insert("key".to_string(), 1).unwrap();
let mut second = Store::<String, u32>::new(eager("device1/matmul"));
assert_eq!(second.get(&"key".to_string()), None);
second.insert("key".to_string(), 2).unwrap();
assert_eq!(first.get(&"key".to_string()), Some(&1));
assert_eq!(second.get(&"key".to_string()), Some(&2));
}
#[test_log::test]
#[serial_test::serial]
#[cfg_attr(miri, ignore)]
fn lazy_values_survive_reopen_and_load_lazily() {
let dir = tempfile::tempdir().unwrap();
crate::environment::set_root(dir.path());
let mut cache = Store::<String, Bytes>::new(lazy("ptx_sm90"));
cache
.insert(
"kernel_a".to_string(),
Bytes::from_bytes_vec(std::vec![1, 2, 3]),
)
.unwrap();
cache
.insert(
"kernel_b".to_string(),
Bytes::from_bytes_vec(std::vec![4, 5]),
)
.unwrap();
assert!(cache.is_empty());
drop(cache);
let mut cache = Store::<String, Bytes>::new(lazy("ptx_sm90"));
assert!(cache.is_empty());
assert_eq!(
cache.get_mut(&"kernel_a".to_string()).map(|v| v.to_vec()),
Some(std::vec![1, 2, 3])
);
assert_eq!(cache.len(), 1, "get_mut memoizes");
assert_eq!(
cache.remove(&"kernel_b".to_string()).map(|v| v.to_vec()),
Some(std::vec![4, 5])
);
assert_eq!(cache.len(), 1, "remove reads through without memoizing");
assert_eq!(cache.get_mut(&"missing".to_string()), None);
}
#[test_log::test]
#[serial_test::serial]
#[cfg_attr(miri, ignore)]
fn lazy_reinserting_a_different_value_errors() {
let dir = tempfile::tempdir().unwrap();
crate::environment::set_root(dir.path());
let mut cache = Store::<String, Bytes>::new(lazy("ptx_sm90"));
let kernel = |byte: u8| Bytes::from_bytes_vec(std::vec![byte]);
cache.insert("kernel".to_string(), kernel(1)).unwrap();
assert!(cache.insert("kernel".to_string(), kernel(1)).is_ok());
let error = cache.insert("kernel".to_string(), kernel(2));
assert!(matches!(error, Err(StoreError::DuplicatedKey { .. })));
assert!(cache.remove(&"kernel".to_string()).is_some());
let error = cache.insert("kernel".to_string(), kernel(2));
assert!(matches!(error, Err(StoreError::DuplicatedKey { .. })));
}
#[test_log::test]
#[serial_test::serial]
#[cfg_attr(miri, ignore)]
fn switching_environments_resets_bound_stores() {
let first = tempfile::tempdir().unwrap();
let second = tempfile::tempdir().unwrap();
crate::environment::set_root(first.path());
let mut store = Store::<String, u32>::new(eager("reset"));
store.insert("key".to_string(), 1).unwrap();
assert_eq!(store.get(&"key".to_string()), Some(&1));
crate::environment::set_root(second.path());
assert_eq!(store.get(&"key".to_string()), None);
assert_eq!(store.len(), 0);
assert!(store.pending_load());
store.insert("key".to_string(), 2).unwrap();
assert_eq!(store.get(&"key".to_string()), Some(&2));
crate::environment::set_root(first.path());
store.sync();
assert_eq!(store.get(&"key".to_string()), Some(&1));
}
#[test_log::test]
#[serial_test::serial]
#[cfg_attr(miri, ignore)]
fn unbound_stores_survive_environment_switches() {
let root = tempfile::tempdir().unwrap();
let mut store = Store::<String, u32>::new(StoreOptions::new());
store.insert("key".to_string(), 1).unwrap();
crate::environment::set_root(root.path());
assert_eq!(store.get(&"key".to_string()), Some(&1));
}
#[test_log::test]
#[serial_test::serial]
#[cfg_attr(miri, ignore)]
fn purge_key_deletes_one_entry_durably() {
let dir = tempfile::tempdir().unwrap();
crate::environment::set_root(dir.path());
let mut store = Store::<String, u32>::new(eager("purge_key"));
store.insert("gone".to_string(), 1).unwrap();
store.insert("kept".to_string(), 2).unwrap();
assert_eq!(store.purge_key(&"gone".to_string()), Some(1));
store.insert("gone".to_string(), 3).unwrap();
assert_eq!(store.purge_key(&"gone".to_string()), Some(3));
drop(store);
let store = Store::<String, u32>::new(eager("purge_key"));
assert_eq!(store.get(&"gone".to_string()), None);
assert_eq!(store.get(&"kept".to_string()), Some(&2));
}
#[test_log::test]
#[serial_test::serial]
#[cfg_attr(miri, ignore)]
fn clear_evicts_memory_but_not_the_storage() {
let dir = tempfile::tempdir().unwrap();
crate::environment::set_root(dir.path());
let mut store = Store::<String, u32>::new(eager("clear"));
store.insert("key".to_string(), 1).unwrap();
store.clear();
assert!(store.is_empty());
assert!(matches!(
store.insert("key".to_string(), 2),
Err(StoreError::DuplicatedKey { .. })
));
store.sync();
assert_eq!(store.get(&"key".to_string()), Some(&1));
}
#[test_log::test]
#[serial_test::serial]
#[cfg_attr(miri, ignore)]
fn purge_deletes_durably_and_frees_the_keys() {
let dir = tempfile::tempdir().unwrap();
crate::environment::set_root(dir.path());
let mut store = Store::<String, u32>::new(eager("purge"));
store.insert("kept".to_string(), 1).unwrap();
store.insert("gone".to_string(), 2).unwrap();
let mut other = Store::<String, u32>::new(eager("other"));
other.insert("kept".to_string(), 9).unwrap();
store.purge();
assert!(store.is_empty());
store.insert("kept".to_string(), 3).unwrap();
drop(store);
let store = Store::<String, u32>::new(eager("purge"));
assert_eq!(store.get(&"kept".to_string()), Some(&3));
assert_eq!(store.get(&"gone".to_string()), None);
assert_eq!(
Store::<String, u32>::new(eager("other")).get(&"kept".to_string()),
Some(&9)
);
}
#[test_log::test]
#[serial_test::serial]
#[cfg_attr(miri, ignore)]
fn scan_visits_the_storage_without_retaining() {
let dir = tempfile::tempdir().unwrap();
crate::environment::set_root(dir.path());
let mut store = Store::<String, u32>::new(lazy("scan"));
store.insert("a".to_string(), 1).unwrap();
store.insert("b".to_string(), 2).unwrap();
let mut seen = std::vec::Vec::new();
let complete = store.scan(|key, value| seen.push((key, value)));
seen.sort();
assert!(complete, "a synchronous storage is scanned in full");
assert_eq!(seen, std::vec![("a".to_string(), 1), ("b".to_string(), 2)]);
assert!(store.is_empty(), "nothing stays resident after a scan");
}
#[test]
fn in_memory_store_needs_no_storage() {
let mut store = Store::<String, u32>::new(StoreOptions::new());
store.insert("key".to_string(), 1).unwrap();
assert_eq!(store.get(&"key".to_string()), Some(&1));
assert!(store.insert("key".to_string(), 1).is_ok());
assert!(matches!(
store.insert("key".to_string(), 2),
Err(StoreError::DuplicatedKey { .. })
));
assert_eq!(store.remove(&"key".to_string()), Some(1));
store.insert("key".to_string(), 2).unwrap();
assert_eq!(store.get(&"key".to_string()), Some(&2));
}
}