From 1f68371920beb287db7079c6bc7edad65a0c35a0 Mon Sep 17 00:00:00 2001 From: Brummel Date: Sun, 29 Mar 2026 21:40:30 +0200 Subject: [PATCH] Add symbol-based cache eviction Introduces a reference counting mechanism for symbols within the `FileCache`. When a `SymbolChunkIter` is created, it increments the reference count for its symbol. When the iterator is dropped, the reference count is decremented. If a symbol's reference count drops to zero, all cached data associated with that symbol is evicted from the cache, freeing up memory. This prevents stale data from being held indefinitely when no active iterators are using it. --- examples/data_stream_print.myc | 8 ++++ src/ast/data_server/cache.rs | 82 ++++++++++++++++++++++++++++++++++ src/ast/data_server/mod.rs | 16 +++++++ 3 files changed, 106 insertions(+) create mode 100644 examples/data_stream_print.myc diff --git a/examples/data_stream_print.myc b/examples/data_stream_print.myc new file mode 100644 index 0000000..99b14d6 --- /dev/null +++ b/examples/data_stream_print.myc @@ -0,0 +1,8 @@ +;; Skip: data output to stdout +(do + (def ohlc (create-m1-stream "EURUSD")) + + ;; Sink: print returns void, so nothing is propagated downstream + (pipe ohlc + (fn [bar] + (print (.close bar))))) diff --git a/src/ast/data_server/cache.rs b/src/ast/data_server/cache.rs index e35aaf9..6bcd112 100644 --- a/src/ast/data_server/cache.rs +++ b/src/ast/data_server/cache.rs @@ -9,6 +9,9 @@ use super::records::{DataFormat, M1Parsed, TickParsed}; use std::collections::{HashMap, HashSet}; use std::sync::{Arc, Condvar, Mutex, RwLock}; +/// Per-symbol reference count for cache eviction. +type SymbolRefCounts = Mutex, usize>>; + /// Uniquely identifies one data file on disk. #[derive(Clone, Debug, Hash, Eq, PartialEq)] pub struct FileKey { @@ -40,6 +43,9 @@ pub struct FileCache { /// Signaled whenever a file finishes loading (insert completes). /// Waiters re-check `is_cached` after each wake-up. loaded: Condvar, + /// Per-symbol reference counts. When a count drops to zero, all + /// cached files for that symbol are evicted from both maps. + symbol_refs: SymbolRefCounts, } impl Default for FileCache { @@ -55,6 +61,7 @@ impl FileCache { tick: RwLock::new(HashMap::new()), loading: Mutex::new(HashSet::new()), loaded: Condvar::new(), + symbol_refs: Mutex::new(HashMap::new()), } } @@ -132,6 +139,46 @@ impl FileCache { guard = self.loaded.wait(guard).unwrap(); } } + + // -- Symbol-level reference counting ------------------------------------ + + /// Increments the reference count for a symbol. Call this when creating + /// a new [`SymbolChunkIter`] so the cached data stays alive. + pub fn retain_symbol(&self, symbol: &Arc) { + *self + .symbol_refs + .lock() + .unwrap() + .entry(Arc::clone(symbol)) + .or_insert(0) += 1; + } + + /// Decrements the reference count for a symbol. When the count reaches + /// zero, all cached files (M1 and tick) for that symbol are evicted. + pub fn release_symbol(&self, symbol: &Arc) { + let mut refs = self.symbol_refs.lock().unwrap(); + if let Some(count) = refs.get_mut(symbol) { + *count -= 1; + if *count == 0 { + refs.remove(symbol); + // Drop the lock before acquiring write locks on the caches + drop(refs); + self.evict_symbol(symbol); + } + } + } + + /// Removes all cached entries for the given symbol from both maps. + fn evict_symbol(&self, symbol: &Arc) { + self.m1 + .write() + .unwrap() + .retain(|key, _| key.symbol != *symbol); + self.tick + .write() + .unwrap() + .retain(|key, _| key.symbol != *symbol); + } } #[cfg(test)] @@ -233,4 +280,39 @@ mod tests { // Waiter should now see the cached data assert!(handle.join().unwrap()); } + + #[test] + fn test_symbol_retain_release() { + let cache = FileCache::new(); + let sym: Arc = Arc::from("EURUSD"); + let key1 = test_key("EURUSD", 2017, 1, DataFormat::M1); + let key2 = test_key("EURUSD", 2017, 2, DataFormat::M1); + let key3 = test_key("EURUSD", 2017, 1, DataFormat::Tick); + let other = test_key("GBPUSD", 2020, 6, DataFormat::M1); + + // Insert data for both symbols + cache.insert_m1(key1.clone(), vec![]); + cache.insert_m1(key2.clone(), vec![]); + cache.insert_tick(key3.clone(), vec![]); + cache.insert_m1(other.clone(), vec![]); + + // Two consumers retain EURUSD + cache.retain_symbol(&sym); + cache.retain_symbol(&sym); + + // First release: refcount drops to 1 — data stays + cache.release_symbol(&sym); + assert!(cache.is_cached(&key1)); + assert!(cache.is_cached(&key2)); + assert!(cache.is_cached(&key3)); + + // Second release: refcount drops to 0 — all EURUSD evicted + cache.release_symbol(&sym); + assert!(!cache.is_cached(&key1)); + assert!(!cache.is_cached(&key2)); + assert!(!cache.is_cached(&key3)); + + // GBPUSD is untouched + assert!(cache.is_cached(&other)); + } } diff --git a/src/ast/data_server/mod.rs b/src/ast/data_server/mod.rs index 6cd4e46..eae5083 100644 --- a/src/ast/data_server/mod.rs +++ b/src/ast/data_server/mod.rs @@ -190,6 +190,7 @@ impl DataServer { /// The iterator yields `Arc<[M1Parsed]>` chunks in chronological order, /// prefetching the next file while the current one is consumed. pub fn stream_m1(self: &Arc, symbol: &str) -> Option> { + let symbol_arc = self.index.symbols.get_key_value(symbol)?.0.clone(); let files: Vec = self .index .symbols @@ -209,8 +210,11 @@ impl DataServer { self.prefetch_m1(&files[1]); } + self.cache.retain_symbol(&symbol_arc); + Some(SymbolChunkIter { server: Arc::clone(self), + symbol: symbol_arc, files, file_idx: 0, chunk_idx: 0, @@ -220,6 +224,7 @@ impl DataServer { /// Returns a chunk iterator over all tick data for a symbol. pub fn stream_tick(self: &Arc, symbol: &str) -> Option> { + let symbol_arc = self.index.symbols.get_key_value(symbol)?.0.clone(); let files: Vec = self .index .symbols @@ -238,8 +243,11 @@ impl DataServer { self.prefetch_tick(&files[1]); } + self.cache.retain_symbol(&symbol_arc); + Some(SymbolChunkIter { server: Arc::clone(self), + symbol: symbol_arc, files, file_idx: 0, chunk_idx: 0, @@ -346,6 +354,8 @@ impl DataServer { /// `LoadDataFile(nextFileInfo)` prefetch call. pub struct SymbolChunkIter { server: Arc, + /// Symbol name, held for `Drop`-based cache release. + symbol: Arc, files: Vec, file_idx: usize, chunk_idx: usize, @@ -354,6 +364,12 @@ pub struct SymbolChunkIter { current_file_chunks: Option>, } +impl Drop for SymbolChunkIter { + fn drop(&mut self) { + self.server.cache.release_symbol(&self.symbol); + } +} + impl SymbolChunkIter { /// Returns the next chunk of M1 records, or `None` when exhausted. pub fn next_chunk(&mut self) -> Option> {