json_eval_rs/jsoneval/
eval_cache.rs

1#[cfg(feature = "parallel")]
2use dashmap::DashMap;
3#[cfg(not(feature = "parallel"))]
4use std::cell::RefCell;
5#[cfg(not(feature = "parallel"))]
6use std::collections::HashMap;
7
8use ahash::AHasher;
9use indexmap::IndexSet;
10use serde_json::Value;
11use std::hash::{Hash, Hasher};
12use std::sync::atomic::{AtomicUsize, Ordering};
13use std::sync::Arc;
14
15
16
17/// Hash a serde_json::Value directly without intermediate string allocation
18#[inline]
19fn hash_value(value: &Value, hasher: &mut AHasher) {
20    match value {
21        Value::Null => 0u8.hash(hasher),
22        Value::Bool(b) => {
23            1u8.hash(hasher);
24            b.hash(hasher);
25        }
26        Value::Number(n) => {
27            2u8.hash(hasher);
28            n.as_f64().unwrap_or(0.0).to_bits().hash(hasher);
29        }
30        Value::String(s) => {
31            3u8.hash(hasher);
32            s.hash(hasher);
33        }
34        Value::Array(arr) => {
35            4u8.hash(hasher);
36            arr.len().hash(hasher);
37            for v in arr {
38                hash_value(v, hasher);
39            }
40        }
41        Value::Object(map) => {
42            5u8.hash(hasher);
43            map.len().hash(hasher);
44            for (k, v) in map {
45                k.hash(hasher);
46                hash_value(v, hasher);
47            }
48        }
49    }
50}
51
52
53/// Cache key: combines evaluation logic ID with hash of all dependent values
54/// Zero-copy design: stores references to logic and dependency paths
55#[derive(Debug, Clone, PartialEq, Eq, Hash)]
56pub struct CacheKey {
57    /// Evaluation key (e.g., "$params.foo")
58    pub eval_key: String,
59    /// Single hash of all dependency values combined (for efficiency)
60    pub deps_hash: u64,
61}
62
63impl CacheKey {
64    /// Create cache key from evaluation key and dependency values
65    /// Dependencies are pre-filtered by caller (JSONEval)
66    /// Hashes all dependency values together in one pass for efficiency
67    pub fn new(
68        eval_key: String,
69        dependencies: &IndexSet<String>,
70        values: &[(String, &Value)],
71    ) -> Self {
72        let value_map: std::collections::HashMap<&str, &Value> =
73            values.iter().map(|(k, v)| (k.as_str(), *v)).collect();
74
75        let mut hasher = AHasher::default();
76        for dep_key in dependencies.iter() {
77            dep_key.hash(&mut hasher);
78            if let Some(value) = value_map.get(dep_key.as_str()) {
79                hash_value(value, &mut hasher);
80            } else {
81                0u8.hash(&mut hasher);
82            }
83        }
84
85        let deps_hash = hasher.finish();
86
87        Self {
88            eval_key,
89            deps_hash,
90        }
91    }
92
93
94    /// Create a simple cache key without dependencies (for evaluations with no deps)
95    pub fn simple(eval_key: String) -> Self {
96        Self {
97            eval_key,
98            deps_hash: 0, // No dependencies = hash of 0
99        }
100    }
101}
102
103/// Zero-copy cache store
104/// With parallel feature: Uses DashMap for thread-safe concurrent access
105/// Without parallel feature: Uses HashMap + RefCell for ultra-fast single-threaded access
106/// Values are stored behind Arc to enable cheap cloning
107pub struct EvalCache {
108    #[cfg(feature = "parallel")]
109    /// Cache storage: DashMap for thread-safe concurrent access
110    cache: DashMap<CacheKey, Arc<Value>>,
111
112    #[cfg(not(feature = "parallel"))]
113    /// Cache storage: HashMap + RefCell for ultra-fast single-threaded access
114    cache: RefCell<HashMap<CacheKey, Arc<Value>>>,
115
116    /// Cache statistics (atomic for thread safety)
117    hits: AtomicUsize,
118    misses: AtomicUsize,
119}
120
121impl EvalCache {
122    /// Create a new empty cache
123    pub fn new() -> Self {
124        Self {
125            #[cfg(feature = "parallel")]
126            cache: DashMap::new(),
127            #[cfg(not(feature = "parallel"))]
128            cache: RefCell::new(HashMap::new()),
129            hits: AtomicUsize::new(0),
130            misses: AtomicUsize::new(0),
131        }
132    }
133
134    /// Create cache with preallocated capacity
135    pub fn with_capacity(capacity: usize) -> Self {
136        Self {
137            #[cfg(feature = "parallel")]
138            cache: DashMap::with_capacity(capacity),
139            #[cfg(not(feature = "parallel"))]
140            cache: RefCell::new(HashMap::with_capacity(capacity)),
141            hits: AtomicUsize::new(0),
142            misses: AtomicUsize::new(0),
143        }
144    }
145
146    /// Get cached result (zero-copy via Arc clone)
147    /// Returns None if not cached
148    #[cfg(feature = "parallel")]
149    /// Thread-safe: can be called concurrently
150    #[inline]
151    pub fn get(&self, key: &CacheKey) -> Option<Arc<Value>> {
152        if let Some(value) = self.cache.get(key) {
153            self.hits.fetch_add(1, Ordering::Relaxed);
154            Some(Arc::clone(value.value()))
155        } else {
156            self.misses.fetch_add(1, Ordering::Relaxed);
157            None
158        }
159    }
160
161    /// Get cached result (zero-copy via Arc clone)
162    /// Returns None if not cached
163    #[cfg(not(feature = "parallel"))]
164    /// Ultra-fast single-threaded access
165    #[inline]
166    pub fn get(&self, key: &CacheKey) -> Option<Arc<Value>> {
167        if let Some(value) = self.cache.borrow().get(key) {
168            self.hits.fetch_add(1, Ordering::Relaxed);
169            Some(Arc::clone(value))
170        } else {
171            self.misses.fetch_add(1, Ordering::Relaxed);
172            None
173        }
174    }
175
176    /// Insert result into cache (wraps in Arc for zero-copy sharing)
177    #[cfg(feature = "parallel")]
178    /// Thread-safe: can be called concurrently
179    #[inline]
180    pub fn insert(&self, key: CacheKey, value: Value) {
181        self.cache.insert(key, Arc::new(value));
182    }
183
184    /// Insert result into cache (wraps in Arc for zero-copy sharing)
185    #[cfg(not(feature = "parallel"))]
186    /// Ultra-fast single-threaded access
187    #[inline]
188    pub fn insert(&self, key: CacheKey, value: Value) {
189        self.cache.borrow_mut().insert(key, Arc::new(value));
190    }
191
192    /// Insert with Arc-wrapped value (zero-copy if already Arc)
193    #[cfg(feature = "parallel")]
194    /// Thread-safe: can be called concurrently
195    #[inline]
196    pub fn insert_arc(&self, key: CacheKey, value: Arc<Value>) {
197        self.cache.insert(key, value);
198    }
199
200    /// Insert with Arc-wrapped value (zero-copy if already Arc)
201    #[cfg(not(feature = "parallel"))]
202    /// Ultra-fast single-threaded access
203    #[inline]
204    pub fn insert_arc(&self, key: CacheKey, value: Arc<Value>) {
205        self.cache.borrow_mut().insert(key, value);
206    }
207
208    /// Clear all cached entries
209    #[cfg(feature = "parallel")]
210    pub fn clear(&self) {
211        self.cache.clear();
212        self.hits.store(0, Ordering::Relaxed);
213        self.misses.store(0, Ordering::Relaxed);
214    }
215
216    /// Clear all cached entries
217    #[cfg(not(feature = "parallel"))]
218    pub fn clear(&self) {
219        self.cache.borrow_mut().clear();
220        self.hits.store(0, Ordering::Relaxed);
221        self.misses.store(0, Ordering::Relaxed);
222    }
223
224    /// Get cache hit rate (0.0 to 1.0)
225    pub fn hit_rate(&self) -> f64 {
226        let hits = self.hits.load(Ordering::Relaxed);
227        let misses = self.misses.load(Ordering::Relaxed);
228        let total = hits + misses;
229        if total == 0 {
230            0.0
231        } else {
232            hits as f64 / total as f64
233        }
234    }
235
236    /// Get cache statistics
237    #[cfg(feature = "parallel")]
238    pub fn stats(&self) -> CacheStats {
239        CacheStats {
240            hits: self.hits.load(Ordering::Relaxed),
241            misses: self.misses.load(Ordering::Relaxed),
242            entries: self.cache.len(),
243            hit_rate: self.hit_rate(),
244        }
245    }
246
247    /// Get cache statistics
248    #[cfg(not(feature = "parallel"))]
249    pub fn stats(&self) -> CacheStats {
250        CacheStats {
251            hits: self.hits.load(Ordering::Relaxed),
252            misses: self.misses.load(Ordering::Relaxed),
253            entries: self.cache.borrow().len(),
254            hit_rate: self.hit_rate(),
255        }
256    }
257
258    /// Get number of cached entries
259    #[cfg(feature = "parallel")]
260    #[inline]
261    pub fn len(&self) -> usize {
262        self.cache.len()
263    }
264
265    /// Get number of cached entries
266    #[cfg(not(feature = "parallel"))]
267    #[inline]
268    pub fn len(&self) -> usize {
269        self.cache.borrow().len()
270    }
271
272    /// Check if cache is empty
273    #[cfg(feature = "parallel")]
274    #[inline]
275    pub fn is_empty(&self) -> bool {
276        self.cache.is_empty()
277    }
278
279    /// Check if cache is empty
280    #[cfg(not(feature = "parallel"))]
281    #[inline]
282    pub fn is_empty(&self) -> bool {
283        self.cache.borrow().is_empty()
284    }
285
286    /// Remove specific entry
287    #[cfg(feature = "parallel")]
288    #[inline]
289    pub fn remove(&self, key: &CacheKey) -> Option<Arc<Value>> {
290        self.cache.remove(key).map(|(_, v)| v)
291    }
292
293    /// Remove specific entry
294    #[cfg(not(feature = "parallel"))]
295    #[inline]
296    pub fn remove(&self, key: &CacheKey) -> Option<Arc<Value>> {
297        self.cache.borrow_mut().remove(key)
298    }
299
300    /// Remove entries based on a predicate function
301    /// Predicate returns true to keep the entry, false to remove it
302    #[cfg(feature = "parallel")]
303    pub fn retain<F>(&self, predicate: F)
304    where
305        F: Fn(&CacheKey, &Arc<Value>) -> bool,
306    {
307        self.cache.retain(|k, v| predicate(k, v));
308    }
309
310    /// Remove entries based on a predicate function
311    /// Predicate returns true to keep the entry, false to remove it
312    #[cfg(not(feature = "parallel"))]
313    pub fn retain<F>(&self, predicate: F)
314    where
315        F: Fn(&CacheKey, &Arc<Value>) -> bool,
316    {
317        self.cache.borrow_mut().retain(|k, v| predicate(k, v));
318    }
319
320    /// Invalidate cache entries that depend on changed paths
321    /// Efficiently removes only affected entries
322    #[cfg(feature = "parallel")]
323    pub fn invalidate_dependencies(&self, changed_paths: &[String]) {
324        // Build a set of changed path hashes for fast lookup
325        let changed_hashes: IndexSet<String> = changed_paths.iter().cloned().collect();
326
327        // Remove cache entries whose eval_key is in the changed set
328        self.cache
329            .retain(|cache_key, _| !changed_hashes.contains(&cache_key.eval_key));
330    }
331
332    /// Invalidate cache entries that depend on changed paths
333    /// Efficiently removes only affected entries
334    #[cfg(not(feature = "parallel"))]
335    pub fn invalidate_dependencies(&self, changed_paths: &[String]) {
336        // Build a set of changed path hashes for fast lookup
337        let changed_hashes: IndexSet<String> = changed_paths.iter().cloned().collect();
338
339        // Remove cache entries whose eval_key is in the changed set
340        self.cache
341            .borrow_mut()
342            .retain(|cache_key, _| !changed_hashes.contains(&cache_key.eval_key));
343    }
344}
345
346impl Default for EvalCache {
347    fn default() -> Self {
348        Self::new()
349    }
350}
351
352/// Cache statistics
353#[derive(Debug, Clone, Copy)]
354pub struct CacheStats {
355    pub hits: usize,
356    pub misses: usize,
357    pub entries: usize,
358    pub hit_rate: f64,
359}
360
361impl std::fmt::Display for CacheStats {
362    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
363        write!(
364            f,
365            "Cache Stats: {} entries, {} hits, {} misses, {:.2}% hit rate",
366            self.entries,
367            self.hits,
368            self.misses,
369            self.hit_rate * 100.0
370        )
371    }
372}
373
374#[cfg(test)]
375mod tests {
376    use super::*;
377    use serde_json::json;
378
379    #[test]
380    fn test_cache_key_creation() {
381        let eval_key = "$params.foo".to_string();
382        let mut deps = IndexSet::new();
383        deps.insert("$params.bar".to_string());
384        deps.insert("data.value".to_string());
385
386        let val1 = json!(42);
387        let val2 = json!("test");
388        let values = vec![
389            ("$params.bar".to_string(), &val1),
390            ("data.value".to_string(), &val2),
391        ];
392
393        let key1 = CacheKey::new(eval_key.clone(), &deps, &values);
394        let key2 = CacheKey::new(eval_key.clone(), &deps, &values);
395
396        // Same inputs should produce same key
397        assert_eq!(key1, key2);
398    }
399
400    #[test]
401    fn test_cache_key_different_values() {
402        let eval_key = "$params.foo".to_string();
403        let mut deps = IndexSet::new();
404        deps.insert("data.value".to_string());
405
406        let val1 = json!(42);
407        let val2 = json!(43);
408        let values1 = vec![("data.value".to_string(), &val1)];
409        let values2 = vec![("data.value".to_string(), &val2)];
410
411        let key1 = CacheKey::new(eval_key.clone(), &deps, &values1);
412        let key2 = CacheKey::new(eval_key.clone(), &deps, &values2);
413
414        // Different values should produce different keys
415        assert_ne!(key1, key2);
416    }
417
418    #[test]
419    fn test_cache_operations() {
420        let cache = EvalCache::new();
421
422        let key = CacheKey::simple("test".to_string());
423        let value = json!({"result": 42});
424
425        // Test miss
426        assert!(cache.get(&key).is_none());
427        assert_eq!(cache.stats().misses, 1);
428
429        // Insert and test hit
430        cache.insert(key.clone(), value.clone());
431        assert_eq!(cache.get(&key).unwrap().as_ref(), &value);
432        assert_eq!(cache.stats().hits, 1);
433
434        // Test stats
435        let stats = cache.stats();
436        assert_eq!(stats.entries, 1);
437        assert_eq!(stats.hit_rate, 0.5); // 1 hit, 1 miss
438    }
439
440    #[test]
441    fn test_cache_clear() {
442        let cache = EvalCache::new();
443        cache.insert(CacheKey::simple("test".to_string()), json!(42));
444
445        assert_eq!(cache.len(), 1);
446        cache.clear();
447        assert_eq!(cache.len(), 0);
448        assert_eq!(cache.stats().hits, 0);
449    }
450
451    #[test]
452    fn test_invalidate_dependencies() {
453        let cache = EvalCache::new();
454
455        // Add cache entries
456        cache.insert(CacheKey::simple("$params.foo".to_string()), json!(1));
457        cache.insert(CacheKey::simple("$params.bar".to_string()), json!(2));
458        cache.insert(CacheKey::simple("$params.baz".to_string()), json!(3));
459
460        assert_eq!(cache.len(), 3);
461
462        // Invalidate one path
463        cache.invalidate_dependencies(&["$params.foo".to_string()]);
464
465        assert_eq!(cache.len(), 2);
466        assert!(cache
467            .get(&CacheKey::simple("$params.foo".to_string()))
468            .is_none());
469        assert!(cache
470            .get(&CacheKey::simple("$params.bar".to_string()))
471            .is_some());
472    }
473}
json_eval_rs/jsoneval/eval_cache.rs

json_eval_rs/jsoneval/
eval_cache.rs
