optimizations

Author: tislib

examples/sensor_test/main.rs

@@ -12,7 +12,7 @@ fn main() {
     let readings_count = 100_000_000;
 
     // 1. Initialize StageEngine
-    let engine = StageEngine::<Reading, Reading>::with_capacity(readings_count * 64);
+    let engine = StageEngine::<Reading, Reading>::with_capacity(readings_count);
 
     // 2. Add Aggregation Stage: Reading -> Summary
     let mut engine = engine

src/pipe/delta.rs

@@ -2,14 +2,15 @@ use crate::stage::{OutputCollector, Stage};
 use bytemuck::Pod;
 use std::collections::HashMap;
 use std::marker::PhantomData;
+use fxhash::FxHashMap;
 
 /// Compares the current item with the previous item associated with the same key.
 ///
 /// This stage is useful for calculating changes or deltas between events in a stream.
 pub struct Delta<K, T, Out, F, L> {
     key_fn: F,
     logic: L,
-    last_values: HashMap<K, T>,
+    last_values: FxHashMap<K, T>,
     _phantom: PhantomData<(T, Out)>,
 }
 
@@ -25,7 +26,7 @@ where
         Self {
             key_fn,
             logic,
-            last_values: HashMap::new(),
+            last_values: FxHashMap::default(),
             _phantom: PhantomData,
         }
     }

src/pipe/stateful.rs

@@ -2,6 +2,7 @@ use crate::stage::{OutputCollector, Stage};
 use bytemuck::Pod;
 use std::collections::HashMap;
 use std::marker::PhantomData;
+use fxhash::FxHashMap;
 
 /// Maintains per-key state for stateful aggregations or processing.
 ///
@@ -11,7 +12,7 @@ pub struct Stateful<K, In, Out, KF, IF, FF> {
     key_fn: KF,
     init_fn: IF,
     fold_fn: FF,
-    storage: HashMap<K, Out>,
+    storage: FxHashMap<K, Out>,
     _phantom: PhantomData<In>,
 }
 
@@ -29,7 +30,7 @@ where
             key_fn,
             init_fn,
             fold_fn,
-            storage: HashMap::new(),
+            storage: FxHashMap::default(),
             _phantom: PhantomData,
         }
     }