use crate::axes::{Axis, Operation}; use indexmap::{Equivalent, IndexMap}; use ndarray::{ArrayD, SliceInfoElem}; use std::any::Any; use std::fmt::{Debug, Formatter}; use std::hash::{Hash, Hasher}; use std::path::{Path, PathBuf}; use std::sync::{Arc, Mutex, OnceLock}; use crate::readers::Frame; /// maximum number of frames held in the cache const DEFAULT_FRAME_CACHE_SIZE: usize = 128; /// maximum number of materialized arrays held in the cache const DEFAULT_ARRAY_CACHE_SIZE: usize = 32; /// identity of the reader a frame was read from #[derive(Debug, Clone, PartialEq, Eq, Hash)] pub(crate) struct ReaderKey { pub(crate) name: String, pub(crate) path: PathBuf, pub(crate) series: usize, pub(crate) position: usize, } /// borrowed view of [`ReaderKey`] for cache lookups without allocation. /// hashes byte-identically to [`ReaderKey`] (str/String and Path/PathBuf hash /// the same), so it only matches the same reader identity. #[derive(Debug, PartialEq, Eq)] pub(crate) struct ReaderKeyRef<'a> { pub(crate) name: &'a str, pub(crate) path: &'a Path, pub(crate) series: usize, pub(crate) position: usize, } impl Hash for ReaderKeyRef<'_> { fn hash(&self, state: &mut H) { self.name.hash(state); self.path.hash(state); self.series.hash(state); self.position.hash(state); } } impl Equivalent for ReaderKeyRef<'_> { fn equivalent(&self, key: &ReaderKey) -> bool { self.name == key.name && self.path == key.path.as_path() && self.series == key.series && self.position == key.position } } impl ReaderKeyRef<'_> { pub(crate) fn to_owned(&self) -> ReaderKey { ReaderKey { name: self.name.to_string(), path: self.path.to_path_buf(), series: self.series, position: self.position, } } } pub(crate) type FrameKey = (ReaderKey, usize, usize, usize); /// borrowed view of [`FrameKey`] for cache lookups without allocation. #[derive(Debug, PartialEq, Eq)] pub(crate) struct FrameKeyRef<'a> { pub(crate) reader: ReaderKeyRef<'a>, pub(crate) c: usize, pub(crate) z: usize, pub(crate) t: usize, } impl Hash for FrameKeyRef<'_> { fn hash(&self, state: &mut H) { self.reader.hash(state); self.c.hash(state); self.z.hash(state); self.t.hash(state); } } impl Equivalent for FrameKeyRef<'_> { fn equivalent(&self, key: &FrameKey) -> bool { let (rk, c, z, t) = key; self.reader.equivalent(rk) && self.c == *c && self.z == *z && self.t == *t } } impl FrameKeyRef<'_> { pub(crate) fn to_owned(&self) -> FrameKey { (self.reader.to_owned(), self.c, self.z, self.t) } } /// identity of a materialized array in the process-wide cache #[derive(Debug, Clone, PartialEq, Eq, Hash)] pub(crate) struct ArrayKey { pub(crate) reader: ReaderKey, pub(crate) dtype: &'static str, pub(crate) slice: Vec, pub(crate) axes: Vec, pub(crate) operations: Vec<(Axis, Operation)>, } /// borrowed view of [`ArrayKey`] for cache lookups without allocation. #[derive(Debug, PartialEq, Eq)] pub(crate) struct ArrayKeyRef<'a> { pub(crate) reader: ReaderKeyRef<'a>, pub(crate) dtype: &'static str, pub(crate) slice: &'a [SliceInfoElem], pub(crate) axes: &'a [Axis], pub(crate) operations: &'a IndexMap, } impl Hash for ArrayKeyRef<'_> { fn hash(&self, state: &mut H) { self.reader.hash(state); self.dtype.hash(state); self.slice.hash(state); self.axes.hash(state); self.operations.len().hash(state); for (ax, op) in self.operations.iter() { ax.hash(state); op.hash(state); } } } impl Equivalent for ArrayKeyRef<'_> { fn equivalent(&self, key: &ArrayKey) -> bool { self.reader.equivalent(&key.reader) && self.dtype == key.dtype && self.slice == key.slice && self.axes == key.axes && self.operations.len() == key.operations.len() && self .operations .iter() .zip(&key.operations) .all(|((ax, op), (key_ax, key_op))| ax == key_ax && op == key_op) } } impl ArrayKeyRef<'_> { pub(crate) fn to_owned(&self) -> ArrayKey { ArrayKey { reader: self.reader.to_owned(), dtype: self.dtype, slice: self.slice.to_vec(), axes: self.axes.to_vec(), operations: self .operations .iter() .map(|(ax, op)| (*ax, op.clone())) .collect(), } } } /// process-wide LRU cache of frames, shared between all views and threads static GLOBAL_FRAME_CACHE: OnceLock = OnceLock::new(); /// thread-safe LRU cache of frames read from the underlying reader pub(crate) struct FrameCache { inner: Mutex, } struct FrameCacheInner { map: IndexMap>, capacity: usize, } impl Default for FrameCache { fn default() -> Self { Self::new(DEFAULT_FRAME_CACHE_SIZE) } } impl FrameCache { pub(crate) fn new(capacity: usize) -> Self { Self { inner: Mutex::new(FrameCacheInner { map: IndexMap::with_capacity(capacity), capacity, }), } } pub(crate) fn global() -> &'static FrameCache { GLOBAL_FRAME_CACHE.get_or_init(FrameCache::default) } pub(crate) fn capacity(&self) -> usize { self.inner.lock().unwrap().capacity } pub(crate) fn set_capacity(&self, capacity: usize) { let mut inner = self.inner.lock().unwrap(); inner.capacity = capacity; while inner.map.len() > inner.capacity { inner.map.shift_remove_index(0); } } pub(crate) fn get(&self, key: &Q) -> Option> where Q: ?Sized + Hash + Equivalent, { let mut inner = self.inner.lock().unwrap(); if let Some(idx) = inner.map.get_index_of(key) { let (key, frame) = inner.map.shift_remove_index(idx).unwrap(); inner.map.insert(key, frame.clone()); Some(frame) } else { None } } pub(crate) fn insert(&self, key: FrameKey, frame: Arc) { let mut inner = self.inner.lock().unwrap(); inner.map.insert(key, frame); while inner.map.len() > inner.capacity { inner.map.shift_remove_index(0); } } #[cfg(test)] pub(crate) fn len(&self) -> usize { self.inner.lock().unwrap().map.len() } } impl Debug for FrameCache { fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { f.debug_struct("FrameCache").finish_non_exhaustive() } } /// process-wide LRU cache of materialized arrays, shared between all views and threads static GLOBAL_ARRAY_CACHE: OnceLock = OnceLock::new(); /// thread-safe LRU cache of materialized arrays produced by `as_array_dyn` pub(crate) struct ArrayCache { inner: Mutex, } struct ArrayCacheInner { map: IndexMap>, capacity: usize, } impl Default for ArrayCache { fn default() -> Self { Self::new(DEFAULT_ARRAY_CACHE_SIZE) } } impl ArrayCache { pub(crate) fn new(capacity: usize) -> Self { Self { inner: Mutex::new(ArrayCacheInner { map: IndexMap::with_capacity(capacity), capacity, }), } } pub(crate) fn global() -> &'static ArrayCache { GLOBAL_ARRAY_CACHE.get_or_init(ArrayCache::default) } pub(crate) fn capacity(&self) -> usize { self.inner.lock().unwrap().capacity } pub(crate) fn set_capacity(&self, capacity: usize) { let mut inner = self.inner.lock().unwrap(); inner.capacity = capacity; while inner.map.len() > inner.capacity { inner.map.shift_remove_index(0); } } pub(crate) fn get(&self, key: &Q) -> Option>> where Q: ?Sized + Hash + Equivalent, T: Any + Send + Sync, { let mut inner = self.inner.lock().unwrap(); if let Some(idx) = inner.map.get_index_of(key) { let (key, array) = inner.map.shift_remove_index(idx).unwrap(); inner.map.insert(key, array.clone()); array.downcast::>().ok() } else { None } } pub(crate) fn insert(&self, key: ArrayKey, array: ArrayD) { let mut inner = self.inner.lock().unwrap(); inner.map.insert(key, Arc::new(array)); while inner.map.len() > inner.capacity { inner.map.shift_remove_index(0); } } #[cfg(test)] pub(crate) fn len(&self) -> usize { self.inner.lock().unwrap().map.len() } } impl Debug for ArrayCache { fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { f.debug_struct("ArrayCache").finish_non_exhaustive() } }