diff --git a/README.md b/README.md index 0fe42ee..cff9d2f 100644 --- a/README.md +++ b/README.md @@ -214,6 +214,7 @@ let parser = NetflowParser::builder() ``` **Cache Behavior:** +- Empty template caches reserve no entry storage; each cache grows as templates arrive - When the cache is full, the least recently used template is evicted - Templates are keyed by template ID (per source) - Each parser instance maintains its own template cache diff --git a/src/lib.rs b/src/lib.rs index 79223b1..0cde9a6 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -27,7 +27,10 @@ use std::sync::Arc; /// Count non-expired templates in an LRU cache, respecting TTL if configured. fn count_valid_templates( - cache: &lru::LruCache>, + cache: &variable_versions::lazy_lru::LazyLruCache< + u16, + variable_versions::ttl::TemplateWithTtl, + >, ttl_config: &Option, ) -> usize { match ttl_config { diff --git a/src/variable_versions/ipfix/parser.rs b/src/variable_versions/ipfix/parser.rs index 3886f72..f71f9cc 100644 --- a/src/variable_versions/ipfix/parser.rs +++ b/src/variable_versions/ipfix/parser.rs @@ -19,6 +19,7 @@ use crate::template_store::{ use crate::variable_versions::config::DEFAULT_MAX_RECORDS_PER_FLOWSET; use crate::variable_versions::enterprise_registry::EnterpriseFieldRegistry; use crate::variable_versions::field_value::FieldValue; +use crate::variable_versions::lazy_lru::LazyLruCache; use crate::variable_versions::metrics::CacheMetricsInner; use crate::variable_versions::template_events::TemplateProtocol; use crate::variable_versions::ttl::{TemplateWithTtl, TtlConfig}; @@ -33,7 +34,6 @@ use crate::variable_versions::{ use crate::{NetflowError, NetflowPacket, ParsedNetflow}; use crate::variable_versions::fast_parse::{parse_u8, parse_u16_be}; -use lru::LruCache; use nom::IResult; use nom::combinator::complete; use nom::multi::many0; @@ -88,10 +88,10 @@ impl IPFixParser { .transpose()?; Ok(Self { - templates: LruCache::new(cache_size), - v9_templates: LruCache::new(cache_size), - ipfix_options_templates: LruCache::new(cache_size), - v9_options_templates: LruCache::new(cache_size), + templates: LazyLruCache::new(cache_size), + v9_templates: LazyLruCache::new(cache_size), + ipfix_options_templates: LazyLruCache::new(cache_size), + v9_options_templates: LazyLruCache::new(cache_size), ttl_config: config.ttl_config, max_template_cache_size: config.max_template_cache_size, max_field_count: config.max_field_count, @@ -159,7 +159,7 @@ impl IPFixParser { /// checker reaching for splits. #[allow(clippy::too_many_arguments)] fn install_restored( - cache: &mut LruCache>>, + cache: &mut LazyLruCache>>, template_id: u16, arc: &Arc, ttl_enabled: bool, @@ -835,7 +835,7 @@ impl HasTemplateId for V9OptionsTemplate { /// metrics. Returns the IDs of any entries the LRU evicted to make room so the /// caller can mirror the eviction into the secondary template store. fn insert_templates( - cache: &mut LruCache>>, + cache: &mut LazyLruCache>>, templates: &[T], ttl_enabled: bool, metrics: &mut CacheMetricsInner, diff --git a/src/variable_versions/ipfix/types.rs b/src/variable_versions/ipfix/types.rs index 5176e67..3ad1445 100644 --- a/src/variable_versions/ipfix/types.rs +++ b/src/variable_versions/ipfix/types.rs @@ -9,6 +9,7 @@ use crate::template_store::TemplateStore; use crate::variable_versions::PendingFlowCache; use crate::variable_versions::enterprise_registry::EnterpriseFieldRegistry; use crate::variable_versions::field_value::FieldValue; +use crate::variable_versions::lazy_lru::LazyLruCache; use crate::variable_versions::metrics::CacheMetricsInner; use crate::variable_versions::template_events::TemplateProtocol; use crate::variable_versions::ttl::{TemplateWithTtl, TtlConfig}; @@ -24,7 +25,6 @@ use crate::variable_versions::v9::{ Template as V9Template, }; -use lru::LruCache; use std::sync::Arc; use crate::variable_versions::TemplateId; @@ -37,12 +37,12 @@ pub type IPFixFlowRecord = Vec; /// Supports both native IPFIX templates and V9-style templates embedded in IPFIX messages. #[derive(Debug)] pub struct IPFixParser { - pub(crate) templates: LruCache>>, - pub(crate) v9_templates: LruCache>>, + pub(crate) templates: LazyLruCache>>, + pub(crate) v9_templates: LazyLruCache>>, pub(crate) ipfix_options_templates: - LruCache>>, + LazyLruCache>>, pub(crate) v9_options_templates: - LruCache>>, + LazyLruCache>>, pub(crate) ttl_config: Option, pub(crate) max_template_cache_size: usize, pub(crate) max_field_count: usize, diff --git a/src/variable_versions/lazy_lru.rs b/src/variable_versions/lazy_lru.rs new file mode 100644 index 0000000..a5e1691 --- /dev/null +++ b/src/variable_versions/lazy_lru.rs @@ -0,0 +1,164 @@ +use lru::LruCache; +use std::borrow::Borrow; +use std::fmt; +use std::hash::Hash; +use std::num::NonZeroUsize; + +/// An LRU cache that allocates its backing storage on first insertion. +/// +/// `lru::LruCache::new(capacity)` reserves the full configured capacity. +/// Template parsers own several independent caches, most of which remain unused +/// for a given exporter. This wrapper retains the same capacity and eviction +/// semantics without reserving memory for empty caches. +pub(crate) struct LazyLruCache { + capacity: NonZeroUsize, + cache: Option>, +} + +impl fmt::Debug for LazyLruCache { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + // Preserve the parser's existing debug representation. + formatter + .debug_struct("LruCache") + .field("len", &self.len()) + .field("cap", &self.capacity) + .finish() + } +} + +impl LazyLruCache { + pub(crate) fn new(capacity: NonZeroUsize) -> Self { + Self { + capacity, + cache: None, + } + } + + fn cache_mut(&mut self) -> &mut LruCache { + let capacity = self.capacity; + self.cache.get_or_insert_with(|| { + // `unbounded()` starts with an empty hash map. Resizing it before + // the first insertion sets the eviction limit without reserving + // storage for every possible entry. + let mut cache = LruCache::unbounded(); + cache.resize(capacity); + cache + }) + } + + fn release_if_empty(&mut self) { + if self.cache.as_ref().is_some_and(LruCache::is_empty) { + self.cache = None; + } + } + + pub(crate) fn push(&mut self, key: K, value: V) -> Option<(K, V)> { + self.cache_mut().push(key, value) + } + + #[cfg(test)] + pub(crate) fn put(&mut self, key: K, value: V) -> Option { + self.cache_mut().put(key, value) + } + + pub(crate) fn peek(&self, key: &Q) -> Option<&V> + where + K: Borrow, + Q: Hash + Eq + ?Sized, + { + self.cache.as_ref()?.peek(key) + } + + pub(crate) fn pop(&mut self, key: &Q) -> Option + where + K: Borrow, + Q: Hash + Eq + ?Sized, + { + let value = self.cache.as_mut()?.pop(key); + self.release_if_empty(); + value + } + + pub(crate) fn promote(&mut self, key: &Q) -> bool + where + K: Borrow, + Q: Hash + Eq + ?Sized, + { + self.cache.as_mut().is_some_and(|cache| cache.promote(key)) + } + + pub(crate) fn resize(&mut self, capacity: NonZeroUsize) { + self.capacity = capacity; + if let Some(cache) = self.cache.as_mut() { + cache.resize(capacity); + } + self.release_if_empty(); + } + + pub(crate) fn clear(&mut self) { + self.cache = None; + } + + pub(crate) fn len(&self) -> usize { + self.cache.as_ref().map_or(0, LruCache::len) + } + + #[cfg(test)] + pub(crate) fn cap(&self) -> NonZeroUsize { + self.capacity + } + + pub(crate) fn iter(&self) -> impl Iterator { + self.cache.iter().flat_map(|cache| cache.iter()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn reads_and_resizes_do_not_allocate_an_empty_cache() { + let mut cache = LazyLruCache::::new(NonZeroUsize::new(10).unwrap()); + + assert!(cache.cache.is_none()); + assert_eq!(cache.cap().get(), 10); + assert_eq!(cache.len(), 0); + assert_eq!(cache.peek(&1), None); + assert!(!cache.promote(&1)); + assert_eq!(cache.pop(&1), None); + assert_eq!(cache.iter().count(), 0); + + cache.resize(NonZeroUsize::new(20).unwrap()); + assert!(cache.cache.is_none()); + assert_eq!(cache.cap().get(), 20); + } + + #[test] + fn first_insert_allocates_and_preserves_lru_eviction() { + let mut cache = LazyLruCache::new(NonZeroUsize::new(2).unwrap()); + + assert_eq!(cache.push(1, "one"), None); + assert!(cache.cache.is_some()); + assert_eq!(cache.push(2, "two"), None); + assert!(cache.promote(&1)); + assert_eq!(cache.push(3, "three"), Some((2, "two"))); + assert_eq!(cache.peek(&1), Some(&"one")); + assert_eq!(cache.peek(&2), None); + assert_eq!(cache.peek(&3), Some(&"three")); + } + + #[test] + fn removing_the_last_entry_releases_storage() { + let mut cache = LazyLruCache::new(NonZeroUsize::new(2).unwrap()); + cache.put(1, "one"); + + assert_eq!(cache.pop(&1), Some("one")); + assert!(cache.cache.is_none()); + + cache.put(2, "two"); + cache.clear(); + assert!(cache.cache.is_none()); + assert_eq!(cache.cap().get(), 2); + } +} diff --git a/src/variable_versions/mod.rs b/src/variable_versions/mod.rs index 4d3941a..6c65cda 100644 --- a/src/variable_versions/mod.rs +++ b/src/variable_versions/mod.rs @@ -73,6 +73,7 @@ pub(crate) mod fast_parse; pub mod field_types; pub mod field_value; pub mod ipfix; +pub(crate) mod lazy_lru; pub mod metrics; pub(crate) mod pending_flows; pub mod template_events; @@ -144,7 +145,7 @@ pub(crate) fn calculate_padding(body_size: usize) -> &'static [u8] { /// Returns None if the template doesn't exist or has expired. #[inline] pub(crate) fn get_valid_template( - cache: &mut lru::LruCache>>, + cache: &mut lazy_lru::LazyLruCache>>, id: &TemplateId, ttl_config: &Option, metrics: &mut CacheMetricsInner, @@ -170,7 +171,7 @@ pub(crate) fn get_valid_template( /// Used in replay paths where the parse may fail — promotion should only /// happen on a successful replay (callers promote manually on success). pub(crate) fn peek_valid_template( - cache: &mut lru::LruCache>>, + cache: &mut lazy_lru::LazyLruCache>>, id: &TemplateId, ttl_config: &Option, metrics: &mut CacheMetricsInner, diff --git a/src/variable_versions/v9/parser.rs b/src/variable_versions/v9/parser.rs index 81deec2..5a1042d 100644 --- a/src/variable_versions/v9/parser.rs +++ b/src/variable_versions/v9/parser.rs @@ -19,6 +19,7 @@ use crate::template_store::{ use crate::variable_versions::config::DEFAULT_MAX_RECORDS_PER_FLOWSET; use crate::variable_versions::enterprise_registry::EnterpriseFieldRegistry; use crate::variable_versions::field_value::FieldValue; +use crate::variable_versions::lazy_lru::LazyLruCache; use crate::variable_versions::metrics::CacheMetricsInner; use crate::variable_versions::template_events::TemplateProtocol; use crate::variable_versions::ttl::{TemplateWithTtl, TtlConfig}; @@ -28,7 +29,6 @@ use crate::variable_versions::{ }; use crate::{NetflowError, NetflowPacket, ParsedNetflow}; -use lru::LruCache; use nom::IResult; use nom::bytes::complete::take; use nom::error::{Error as NomError, ErrorKind}; @@ -39,8 +39,9 @@ use std::sync::Arc; /// Stateful NetFlow V9 parser with LRU template caching and optional pending flow support. #[derive(Debug)] pub struct V9Parser { - pub(crate) templates: LruCache>>, - pub(crate) options_templates: LruCache>>, + pub(crate) templates: LazyLruCache>>, + pub(crate) options_templates: + LazyLruCache>>, pub(crate) ttl_config: Option, pub(crate) max_template_cache_size: usize, pub(crate) max_field_count: usize, @@ -108,8 +109,8 @@ impl V9Parser { .transpose()?; Ok(Self { - templates: LruCache::new(cache_size), - options_templates: LruCache::new(cache_size), + templates: LazyLruCache::new(cache_size), + options_templates: LazyLruCache::new(cache_size), ttl_config: config.ttl_config, max_template_cache_size: config.max_template_cache_size, max_field_count: config.max_field_count, @@ -191,7 +192,7 @@ impl V9Parser { /// borrows at the call site without re-borrowing the whole parser. #[allow(clippy::too_many_arguments)] fn install_restored_template( - cache: &mut LruCache>>, + cache: &mut LazyLruCache>>, template_id: u16, arc: &Arc, ttl_enabled: bool, diff --git a/tests/lazy_template_cache_allocation.rs b/tests/lazy_template_cache_allocation.rs new file mode 100644 index 0000000..7b77db3 --- /dev/null +++ b/tests/lazy_template_cache_allocation.rs @@ -0,0 +1,108 @@ +use netflow_parser::NetflowParser; +use std::alloc::{GlobalAlloc, Layout, System}; +use std::hint::black_box; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; + +struct CountingAllocator; + +static TRACKING: AtomicBool = AtomicBool::new(false); +static REQUESTED_BYTES: AtomicUsize = AtomicUsize::new(0); +static LIVE_BYTES: AtomicUsize = AtomicUsize::new(0); + +fn record_allocation(size: usize) { + if TRACKING.load(Ordering::Relaxed) { + REQUESTED_BYTES.fetch_add(size, Ordering::Relaxed); + } +} + +unsafe impl GlobalAlloc for CountingAllocator { + unsafe fn alloc(&self, layout: Layout) -> *mut u8 { + let pointer = unsafe { System.alloc(layout) }; + if !pointer.is_null() { + LIVE_BYTES.fetch_add(layout.size(), Ordering::Relaxed); + record_allocation(layout.size()); + } + pointer + } + + unsafe fn alloc_zeroed(&self, layout: Layout) -> *mut u8 { + let pointer = unsafe { System.alloc_zeroed(layout) }; + if !pointer.is_null() { + LIVE_BYTES.fetch_add(layout.size(), Ordering::Relaxed); + record_allocation(layout.size()); + } + pointer + } + + unsafe fn dealloc(&self, pointer: *mut u8, layout: Layout) { + LIVE_BYTES.fetch_sub(layout.size(), Ordering::Relaxed); + unsafe { System.dealloc(pointer, layout) }; + } + + unsafe fn realloc(&self, pointer: *mut u8, old: Layout, new_size: usize) -> *mut u8 { + let new_pointer = unsafe { System.realloc(pointer, old, new_size) }; + if !new_pointer.is_null() { + if new_size >= old.size() { + LIVE_BYTES.fetch_add(new_size - old.size(), Ordering::Relaxed); + } else { + LIVE_BYTES.fetch_sub(old.size() - new_size, Ordering::Relaxed); + } + record_allocation(new_size); + } + new_pointer + } +} + +#[global_allocator] +static ALLOCATOR: CountingAllocator = CountingAllocator; + +#[test] +fn parser_construction_does_not_reserve_template_cache_storage() { + // Warm process-global allocator and hashing state before measuring the parser. + drop(NetflowParser::builder().with_cache_size(1).build().unwrap()); + + let baseline = LIVE_BYTES.load(Ordering::SeqCst); + REQUESTED_BYTES.store(0, Ordering::SeqCst); + TRACKING.store(true, Ordering::SeqCst); + let parser = black_box(NetflowParser::default()); + TRACKING.store(false, Ordering::SeqCst); + + let requested = REQUESTED_BYTES.load(Ordering::SeqCst); + let retained = LIVE_BYTES.load(Ordering::SeqCst).saturating_sub(baseline); + eprintln!("parser construction requested={requested} retained={retained}"); + + const CONSTRUCTION_BUDGET: usize = 16 * 1024; + assert!( + requested <= CONSTRUCTION_BUDGET, + "parser construction requested {requested} bytes; template caches must allocate on first use" + ); + assert!( + retained <= CONSTRUCTION_BUDGET, + "parser construction retained {retained} bytes; template caches must allocate on first use" + ); + + drop(parser); + + let mut parser = NetflowParser::builder() + .with_cache_size(100_000) + .build() + .unwrap(); + let template = [ + 0, 9, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 1, 0, 0, 0, 12, 1, 0, 0, 1, 0, + 1, 0, 4, + ]; + REQUESTED_BYTES.store(0, Ordering::SeqCst); + TRACKING.store(true, Ordering::SeqCst); + let result = black_box(parser.parse_bytes(&template)); + TRACKING.store(false, Ordering::SeqCst); + + assert!(result.error.is_none()); + assert_eq!(parser.v9_cache_info().current_size, 1); + assert_eq!(parser.v9_cache_info().max_size_per_cache, 100_000); + let requested = REQUESTED_BYTES.load(Ordering::SeqCst); + eprintln!("first template insertion requested={requested}"); + assert!( + requested <= 64 * 1024, + "first template insertion requested {requested} bytes; the cache must grow incrementally" + ); +}