feat: implement integer hasher for better performance

prosa/src/core/main.rs

@@ -20,12 +20,12 @@ use crate::otel::metrics::{Meter, MeterProvider as _};
 use crate::otel::trace::TracerProvider as _;
 use crate::otel::{InstrumentationScope, KeyValue};
 use crate::tracing::{debug, info, warn};
+use prosa_utils::hash::{BuildIntHasher, IntHashMap, IntHashSet};
 use std::sync::{
     Arc,
     atomic::{AtomicBool, Ordering},
 };
-use std::{borrow::Cow, collections::HashSet};
-use std::{collections::HashMap, fmt::Debug};
+use std::{borrow::Cow, fmt::Debug};
 use tokio::{signal, sync::mpsc};
 
 /// Trait to define a ProSA main processor that is runnable
@@ -259,13 +259,16 @@ where
     }
 }
 
+type ProcQueueMap<M> = IntHashMap<u32, ProcService<M>>;
+type ProcessorMap<M> = IntHashMap<u32, ProcQueueMap<M>>;
+
 /// Main ProSA task processor
 pub struct MainProc<M>
 where
     M: Sized + Clone + Tvf,
 {
     name: String,
-    processors: HashMap<u32, HashMap<u32, ProcService<M>>>,
+    processors: ProcessorMap<M>,
     services: Arc<ServiceTable<M>>,
     config: Option<Arc<ProsaConfig>>,
     internal_rx_queue: mpsc::Receiver<InternalMainMsg<M>>,
@@ -290,7 +293,7 @@ impl<M> MainProc<M>
 where
     M: Sized + Clone + Debug + Tvf + Default + 'static + std::marker::Send + std::marker::Sync,
 {
-    async fn remove_proc(&mut self, proc_id: u32) -> Option<HashMap<u32, ProcService<M>>> {
+    async fn remove_proc(&mut self, proc_id: u32) -> Option<ProcQueueMap<M>> {
         if let Some(proc) = self.processors.remove(&proc_id) {
             let mut new_services = (*self.services).clone();
             new_services.remove_proc_services(proc_id);
@@ -433,7 +436,7 @@ where
     fn create<S: Settings>(settings: &S, proc_capacity: Option<usize>) -> (Main<M>, MainProc<M>) {
         fn inner<M>(
             main: Main<M>,
-            processors: HashMap<u32, HashMap<u32, ProcService<M>>>,
+            processors: ProcessorMap<M>,
             internal_rx_queue: mpsc::Receiver<InternalMainMsg<M>>,
         ) -> (Main<M>, MainProc<M>)
         where
@@ -464,9 +467,9 @@ where
         }
         let (internal_tx_queue, internal_rx_queue) = mpsc::channel(2048);
         let processors = if let Some(capacity) = proc_capacity {
-            HashMap::with_capacity(capacity)
+            IntHashMap::with_capacity_and_hasher(capacity, BuildIntHasher::default())
         } else {
-            HashMap::new()
+            IntHashMap::with_hasher(BuildIntHasher::default())
         };
         inner(
             Main::new(internal_tx_queue, settings),
@@ -508,11 +511,11 @@ where
             })
             .build();
 
-        let mut proc_names = HashMap::new();
+        let mut proc_names = IntHashMap::with_hasher(BuildIntHasher::default());
 
         // Monitor processors objects
-        let mut crashed_proc: HashSet<u32> = HashSet::new();
-        let mut restarted_proc = HashMap::new();
+        let mut crashed_proc = IntHashSet::with_hasher(BuildIntHasher::default());
+        let mut restarted_proc = IntHashMap::with_hasher(BuildIntHasher::default());
         let processors_meter = self
             .meter
             .i64_gauge("prosa_processors")
@@ -580,9 +583,9 @@ where
                                 proc_service.insert(queue_id, proc);
                             } else {
                                 proc_names.insert(proc_id, proc.name().to_string());
-                                self.processors.insert(proc_id, HashMap::from([
-                                    (queue_id, proc),
-                                ]));
+                                self.processors.insert(proc_id, [(queue_id, proc)]
+                                .into_iter()
+                                .collect::<IntHashMap<_, _>>());
                             }
 
                             // Ask to the processor to load the service table

prosa/src/event/pending.rs

@@ -1,6 +1,9 @@
-use std::{cmp::Ordering, collections::HashMap, marker::PhantomData, ops::Add, time::Duration};
+use std::{cmp::Ordering, marker::PhantomData, ops::Add, time::Duration};
 
-use prosa_utils::msg::tvf::Tvf;
+use prosa_utils::{
+    hash::{BuildIntHasher, IntHashMap},
+    msg::tvf::Tvf,
+};
 use tokio::time::{Instant, Sleep, sleep_until};
 
 use crate::core::msg::Msg;
@@ -256,7 +259,7 @@ where
     T: Msg<M>,
     M: Sized + Clone + Tvf,
 {
-    pending_messages: HashMap<u64, T>,
+    pending_messages: IntHashMap<u64, T>,
     timers: Timers<u64>,
     phantom: PhantomData<M>,
 }
@@ -288,7 +291,10 @@ where
         M: Sized + Clone + Tvf,
     {
         PendingMsgs {
-            pending_messages: HashMap::with_capacity(capacity),
+            pending_messages: IntHashMap::with_capacity_and_hasher(
+                capacity,
+                BuildIntHasher::default(),
+            ),
             timers: Timers::with_capacity(capacity),
             phantom: PhantomData,
         }

prosa/src/event/queue.rs

@@ -1,5 +1,5 @@
 pub use crate::core::queue::SendError;
-pub use prosa_utils::queue::{QueueChecker, QueueError};
+pub use prosa_utils::queue::{IsInteger, QueueChecker, QueueError};
 
 /// Multi producer / Single consumer queue
 pub mod mpsc;

prosa/src/event/queue/mpsc.rs

@@ -11,11 +11,12 @@ pub trait Sender<T> {
     /// Try to send a value, return a Full error if the queue is full
     ///
     /// ```
-    /// use prosa::event::queue::{QueueChecker, QueueError, mpsc::Sender};
+    /// use prosa::event::queue::{IsInteger, QueueChecker, QueueError, mpsc::Sender};
     ///
     /// async fn sender_process<S, T, P>(sender: S)
     /// where
     ///     S: Sender<T> + QueueChecker<P>,
+    ///     P: IsInteger,
     ///     T: std::cmp::PartialEq + std::fmt::Debug + std::default::Default,
     /// {
     ///     if sender.is_full() {

prosa_book/src/ch02-01-tvf.md

@@ -138,6 +138,26 @@ Make sure to also implement:
 - `Clone`: if duplication of buffers is needed
 - `Debug`: To enable easy debugging and inspection
 
+### Use integer hashing for TVF fields
+
+TVF field identifiers are integers. When designing a TVF buffer backed by a hash
+table, use [`IntHashMap`](https://docs.rs/prosa-utils/latest/prosa_utils/hash/type.IntHashMap.html)
+instead of the standard `HashMap`. It uses ProSA's integer hasher and avoids the
+overhead of general-purpose hashing for field lookups.
+
+```rust,noplayground
+use prosa_utils::hash::IntHashMap;
+
+#[derive(Default)]
+struct MyOwnTvf {
+    fields: IntHashMap<usize, String>,
+}
+```
+
+`IntHashMap` supports only integer key types, including `usize`, which matches the
+field identifiers used by the `Tvf` trait. For an integer set, use
+[`IntHashSet`](https://docs.rs/prosa-utils/latest/prosa_utils/hash/type.IntHashSet.html).
+
 ## Declare your custom TVF
 
 When you implement your own TVF, you need to expose it in your Cargo.toml metadata as discussed in the previous chapter.

prosa_utils/src/config/ssl.rs

@@ -1,7 +1,7 @@
 //! Definition of SSL configuration
 
 use serde::{Deserialize, Serialize};
-use std::{collections::HashMap, fmt, time::Duration};
+use std::{collections::HashMap, fmt, io, path::Path, time::Duration};
 
 use super::ConfigError;
 
@@ -59,6 +59,30 @@ pub enum Store {
     },
 }
 
+impl TryFrom<&Path> for Store {
+    type Error = io::Error;
+
+    /// Initialize a Store with a certificate path
+    ///
+    /// ```
+    /// use std::path::Path;
+    /// use prosa_utils::config::ssl::Store;
+    ///
+    /// let store = Store::try_from(Path::new("cert.pem")).expect("Path should be valid");
+    /// assert_eq!(store, Store::File{ path: "cert.pem".into() });
+    /// ```
+    fn try_from(path: &Path) -> io::Result<Self> {
+        path.to_str()
+            .map(|s| Store::File {
+                path: s.to_string(),
+            })
+            .ok_or(io::Error::new(
+                io::ErrorKind::InvalidFilename,
+                "Invalid store path",
+            ))
+    }
+}
+
 impl fmt::Display for Store {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         match self {

prosa_utils/src/hash.rs

@@ -0,0 +1,165 @@
+//! Module for hashing helpers
+
+use std::{
+    hash::{BuildHasherDefault, Hasher},
+    marker::PhantomData,
+    num::{
+        NonZeroI8, NonZeroI16, NonZeroI32, NonZeroI64, NonZeroIsize, NonZeroU8, NonZeroU16,
+        NonZeroU32, NonZeroU64, NonZeroUsize,
+    },
+};
+
+mod sealed {
+    pub trait SealedInteger {}
+}
+
+/// Int Hasher
+#[derive(Debug, Default, Clone, Copy)]
+pub struct IntHasher<T: IsInteger>(u64, PhantomData<T>);
+
+/// IntHasher builder use for
+/// - [`IntHashSet`]
+/// - [`IntHashMap`]
+pub type BuildIntHasher<T> = BuildHasherDefault<IntHasher<T>>;
+
+/// IntHashSet for integer HashSet
+///
+/// ```
+/// use prosa_utils::hash::{BuildIntHasher, IntHashSet};
+///
+/// let mut int_hashset: IntHashSet<u32> = IntHashSet::with_capacity_and_hasher(1, BuildIntHasher::default());
+/// assert!(int_hashset.insert(2));
+/// assert!(int_hashset.insert(3));
+/// assert!(int_hashset.capacity() >= 2);
+/// ```
+pub type IntHashSet<T> = std::collections::HashSet<T, BuildIntHasher<T>>;
+
+/// IntHashMap for integer HashMap
+///
+/// ```
+/// use prosa_utils::hash::{BuildIntHasher, IntHashMap};
+///
+/// let mut int_hashmap: IntHashMap<i16, String> = IntHashMap::with_capacity_and_hasher(1, BuildIntHasher::default());
+/// assert!(int_hashmap.insert(2, "test2".to_string()).is_none());
+/// assert!(int_hashmap.insert(3, "test3".to_string()).is_none());
+/// assert!(int_hashmap.capacity() >= 2);
+/// ```
+pub type IntHashMap<K, V> = std::collections::HashMap<K, V, BuildIntHasher<K>>;
+
+/// Trait to identify integer types for implementations.
+///
+/// This trait is sealed and can only be implemented by `prosa_utils` for supported integer types.
+///
+/// ```
+/// use prosa_utils::hash::IsInteger;
+///
+/// /// When you implement the trait, P need to be an integer
+/// pub trait IntegerGetter<P: IsInteger> {
+///     /// Returns an integer
+///     fn get_int(&self) -> P;
+/// }
+/// ```
+///
+/// ```compile_fail
+/// struct CustomInteger(u64);
+///
+/// impl prosa_utils::hash::IsInteger for CustomInteger {}
+/// ```
+pub trait IsInteger: sealed::SealedInteger {}
+
+macro_rules! impl_is_integer {
+    ( $($integer:ty),+ $(,)? ) => {
+        $(
+            impl sealed::SealedInteger for $integer {}
+            impl IsInteger for $integer {}
+        )+
+    };
+}
+
+impl_is_integer!(
+    u8,
+    u16,
+    u32,
+    u64,
+    usize,
+    i8,
+    i16,
+    i32,
+    i64,
+    isize,
+    NonZeroU8,
+    NonZeroU16,
+    NonZeroU32,
+    NonZeroU64,
+    NonZeroUsize,
+    NonZeroI8,
+    NonZeroI16,
+    NonZeroI32,
+    NonZeroI64,
+    NonZeroIsize,
+);
+
+impl<T: IsInteger> Hasher for IntHasher<T> {
+    fn write(&mut self, _: &[u8]) {
+        panic!("Invalid use of IntHasher")
+    }
+
+    fn write_u8(&mut self, n: u8) {
+        self.0 = u64::from(n)
+    }
+    fn write_u16(&mut self, n: u16) {
+        self.0 = u64::from(n)
+    }
+    fn write_u32(&mut self, n: u32) {
+        self.0 = u64::from(n)
+    }
+    fn write_u64(&mut self, n: u64) {
+        self.0 = n
+    }
+    fn write_usize(&mut self, n: usize) {
+        self.0 = n as u64
+    }
+
+    fn write_i8(&mut self, n: i8) {
+        self.0 = n as u64
+    }
+    fn write_i16(&mut self, n: i16) {
+        self.0 = n as u64
+    }
+    fn write_i32(&mut self, n: i32) {
+        self.0 = n as u64
+    }
+    fn write_i64(&mut self, n: i64) {
+        self.0 = n as u64
+    }
+    fn write_isize(&mut self, n: isize) {
+        self.0 = n as u64
+    }
+
+    fn finish(&self) -> u64 {
+        self.0
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_hashset() {
+        let mut int_hashset =
+            IntHashSet::<u32>::with_capacity_and_hasher(1, BuildIntHasher::default());
+        assert!(int_hashset.insert(2));
+        assert!(int_hashset.insert(3));
+        assert!(int_hashset.capacity() >= 2);
+    }
+
+    #[test]
+    fn test_hashmap() {
+        let mut int_hashmap =
+            IntHashMap::<i16, String>::with_capacity_and_hasher(1, BuildIntHasher::default());
+        assert!(int_hashmap.insert(2, "test2".to_string()).is_none());
+        assert!(int_hashmap.insert(3, "test3".to_string()).is_none());
+        assert!(int_hashmap.capacity() >= 2);
+    }
+}

prosa_utils/src/lib.rs

@@ -17,3 +17,6 @@ pub mod config;
 
 #[cfg(feature = "queue")]
 pub mod queue;
+
+#[cfg(any(feature = "msg", feature = "queue"))]
+pub mod hash;