merged value-map hash once

fraillt · fraillt · commit 502a208bf957 · 2024-11-16T21:40:47.000+02:00
diff --git a/opentelemetry-sdk/src/metrics/internal/hashed.rs b/opentelemetry-sdk/src/metrics/internal/hashed.rs
@@ -0,0 +1,133 @@
+use std::{
+    borrow::{Borrow, Cow},
+    hash::{BuildHasher, DefaultHasher, Hash, Hasher},
+    ops::Deref,
+};
+
+/// Hash value only once, works with references and owned types.
+pub(crate) struct Hashed<'a, T>
+where
+    T: ToOwned + ?Sized,
+{
+    value: Cow<'a, T>,
+    hash: u64,
+}
+
+impl<'a, T> Hashed<'a, T>
+where
+    T: ToOwned + Hash + ?Sized,
+{
+    pub(crate) fn from_borrowed(value: &'a T) -> Self {
+        let hash = calc_hash(&value);
+        Self {
+            value: Cow::Borrowed(value),
+            hash,
+        }
+    }
+
+    pub(crate) fn from_owned(value: <T as ToOwned>::Owned) -> Self {
+        let hash = calc_hash(value.borrow());
+        Self {
+            value: Cow::Owned(value),
+            hash,
+        }
+    }
+
+    pub(crate) fn into_owned(self) -> Hashed<'static, T> {
+        let value = self.value.into_owned();
+        Hashed {
+            value: Cow::Owned(value),
+            hash: self.hash,
+        }
+    }
+
+    pub(crate) fn into_inner_owned(self) -> T::Owned {
+        self.value.into_owned()
+    }
+}
+
+fn calc_hash<T>(value: T) -> u64
+where
+    T: Hash,
+{
+    let mut hasher = DefaultHasher::default();
+    value.hash(&mut hasher);
+    hasher.finish()
+}
+
+impl<T> Clone for Hashed<'_, T>
+where
+    T: ToOwned + ?Sized,
+{
+    fn clone(&self) -> Self {
+        Self {
+            value: self.value.clone(),
+            hash: self.hash,
+        }
+    }
+
+    fn clone_from(&mut self, source: &Self) {
+        self.value.clone_from(&source.value);
+        self.hash = source.hash;
+    }
+}
+
+impl<T> Hash for Hashed<'_, T>
+where
+    T: ToOwned + Hash + ?Sized,
+{
+    fn hash<H: Hasher>(&self, state: &mut H) {
+        state.write_u64(self.hash);
+    }
+}
+
+impl<T> PartialEq for Hashed<'_, T>
+where
+    T: ToOwned + PartialEq + ?Sized,
+{
+    fn eq(&self, other: &Self) -> bool {
+        self.value.as_ref() == other.value.as_ref()
+    }
+}
+
+impl<T> Eq for Hashed<'_, T> where T: ToOwned + Eq + ?Sized {}
+
+impl<T> Deref for Hashed<'_, T>
+where
+    T: ToOwned + ?Sized,
+{
+    type Target = T;
+
+    fn deref(&self) -> &Self::Target {
+        self.value.deref()
+    }
+}
+
+/// Used to make [`Hashed`] values no-op in [`HashMap`](std::collections::HashMap) or [`HashSet`](std::collections::HashSet).
+/// For all other keys types (except for [`u64`]) it will panic.
+#[derive(Default, Clone)]
+pub(crate) struct HashedNoOpBuilder {
+    hashed: u64,
+}
+
+impl Hasher for HashedNoOpBuilder {
+    fn finish(&self) -> u64 {
+        self.hashed
+    }
+
+    fn write(&mut self, _bytes: &[u8]) {
+        panic!("Only works with `Hashed` value")
+    }
+
+    fn write_u64(&mut self, i: u64) {
+        self.hashed = i;
+    }
+}
+
+impl BuildHasher for HashedNoOpBuilder {
+    type Hasher = HashedNoOpBuilder;
+
+    fn build_hasher(&self) -> Self::Hasher {
+        HashedNoOpBuilder::default()
+    }
+}
diff --git a/opentelemetry-sdk/src/metrics/internal/mod.rs b/opentelemetry-sdk/src/metrics/internal/mod.rs
@@ -1,5 +1,6 @@
 mod aggregate;
 mod exponential_histogram;
+mod hashed;
 mod histogram;
 mod last_value;
 mod precomputed_sum;
@@ -16,13 +17,14 @@ use std::sync::{Arc, Mutex, RwLock};
 use aggregate::is_under_cardinality_limit;
 pub(crate) use aggregate::{AggregateBuilder, ComputeAggregation, Measure};
 pub(crate) use exponential_histogram::{EXPO_MAX_SCALE, EXPO_MIN_SCALE};
+use hashed::{Hashed, HashedNoOpBuilder};
 use once_cell::sync::Lazy;
 use opentelemetry::{otel_warn, KeyValue};
 
-use crate::metrics::AttributeSet;
+use super::sort_and_dedup;
 
-pub(crate) static STREAM_OVERFLOW_ATTRIBUTES: Lazy<Vec<KeyValue>> =
-    Lazy::new(|| vec![KeyValue::new("otel.metric.overflow", "true")]);
+pub(crate) static STREAM_OVERFLOW_ATTRIBUTES: Lazy<Hashed<'static, [KeyValue]>> =
+    Lazy::new(|| Hashed::from_owned(vec![KeyValue::new("otel.metric.overflow", "true")]));
 
 pub(crate) trait Aggregator {
     /// A static configuration that is needed in order to initialize aggregator.
@@ -60,10 +62,10 @@ where
     // for performance reasons, no_attribs tracker
     no_attribs: NoAttribs<A>,
     // for performance reasons, to handle attributes in the provided order
-    all_attribs: RwLock<HashMap<Vec<KeyValue>, Arc<A>>>,
+    all_attribs: RwLock<HashMap<Hashed<'static, [KeyValue]>, Arc<A>, HashedNoOpBuilder>>,
     // different order of attribute keys should still map to same tracker instance
     // this helps to achieve that and also enables implementing collection efficiently
-    sorted_attribs: Mutex<HashMap<Vec<KeyValue>, Arc<A>>>,
+    sorted_attribs: Mutex<HashMap<Hashed<'static, [KeyValue]>, Arc<A>, HashedNoOpBuilder>>,
     /// Configuration for an Aggregator
     config: A::InitConfig,
 }
@@ -78,8 +80,8 @@ where
                 tracker: A::create(&config),
                 is_set: AtomicBool::new(false),
             },
-            all_attribs: RwLock::new(HashMap::new()),
-            sorted_attribs: Mutex::new(HashMap::new()),
+            all_attribs: RwLock::new(HashMap::default()),
+            sorted_attribs: Mutex::new(HashMap::default()),
             config,
         }
     }
@@ -91,10 +93,12 @@ where
             return;
         }
 
+        let attributes = Hashed::from_borrowed(attributes);
+
         // Try to retrieve and update the tracker with the attributes in the provided order first
         match self.all_attribs.read() {
             Ok(trackers) => {
-                if let Some(tracker) = trackers.get(attributes) {
+                if let Some(tracker) = trackers.get(&attributes) {
                     tracker.update(value);
                     return;
                 }
@@ -103,7 +107,7 @@ where
         };
 
         // Get or create a tracker
-        let sorted_attrs = AttributeSet::from(attributes).into_vec();
+        let sorted_attrs = Hashed::from_owned(sort_and_dedup(&attributes));
         let Ok(mut sorted_trackers) = self.sorted_attribs.lock() else {
             return;
         };
@@ -135,7 +139,7 @@ where
         let Ok(mut all_trackers) = self.all_attribs.write() else {
             return;
         };
-        all_trackers.insert(attributes.to_vec(), new_tracker);
+        all_trackers.insert(attributes.into_owned(), new_tracker);
     }
 
     /// Iterate through all attribute sets and populate `DataPoints` in readonly mode.
@@ -160,7 +164,7 @@ where
         }
 
         for (attrs, tracker) in trackers.into_iter() {
-            dest.push(map_fn(attrs, &tracker));
+            dest.push(map_fn(attrs.clone().into_inner_owned(), &tracker));
         }
     }
 
@@ -173,7 +177,7 @@ where
         // reset sorted trackers so new attributes set will be written into new hashmap
         let trackers = match self.sorted_attribs.lock() {
             Ok(mut trackers) => {                
-                let new = HashMap::with_capacity(trackers.len());
+                let new = HashMap::with_capacity_and_hasher(trackers.len(), HashedNoOpBuilder::default());
                 replace(trackers.deref_mut(), new)
             }
             Err(_) => return,
@@ -195,7 +199,7 @@ where
 
         for (attrs, tracker) in trackers.into_iter() {
             let tracker = Arc::into_inner(tracker).expect("the only instance");
-            dest.push(map_fn(attrs, tracker));
+            dest.push(map_fn(attrs.into_inner_owned(), tracker));
         }
     }
 }
@@ -403,6 +407,7 @@ impl AtomicallyUpdate<f64> for f64 {
 
 #[cfg(test)]
 mod tests {
+
     use super::*;
 
     #[test]
diff --git a/opentelemetry-sdk/src/metrics/mod.rs b/opentelemetry-sdk/src/metrics/mod.rs
@@ -115,23 +115,27 @@ pub(crate) struct AttributeSet(Vec<KeyValue>, u64);
 
 impl From<&[KeyValue]> for AttributeSet {
     fn from(values: &[KeyValue]) -> Self {
-        let mut seen_keys = HashSet::with_capacity(values.len());
-        let vec = values
-            .iter()
-            .rev()
-            .filter_map(|kv| {
-                if seen_keys.insert(kv.key.clone()) {
-                    Some(kv.clone())
-                } else {
-                    None
-                }
-            })
-            .collect::<Vec<_>>();
-
-        AttributeSet::new(vec)
+        AttributeSet::new(sort_and_dedup(values))
     }
 }
 
+pub(crate) fn sort_and_dedup(values: &[KeyValue]) -> Vec<KeyValue> {
+    let mut seen_keys = HashSet::with_capacity(values.len());
+    let mut vec = values
+        .iter()
+        .rev()
+        .filter_map(|kv| {
+            if seen_keys.insert(kv.key.clone()) {
+                Some(kv.clone())
+            } else {
+                None
+            }
+        })
+        .collect::<Vec<_>>();
+    vec.sort_unstable_by(|a, b| a.key.cmp(&b.key));
+    vec
+}
+
 fn calculate_hash(values: &[KeyValue]) -> u64 {
     let mut hasher = DefaultHasher::new();
     values.iter().fold(&mut hasher, |mut hasher, item| {
@@ -142,8 +146,7 @@ fn calculate_hash(values: &[KeyValue]) -> u64 {
 }
 
 impl AttributeSet {
-    fn new(mut values: Vec<KeyValue>) -> Self {
-        values.sort_unstable_by(|a, b| a.key.cmp(&b.key));
+    fn new(values: Vec<KeyValue>) -> Self {
         let hash = calculate_hash(&values);
         AttributeSet(values, hash)
     }
@@ -152,11 +155,6 @@ impl AttributeSet {
     pub(crate) fn iter(&self) -> impl Iterator<Item = (&Key, &Value)> {
         self.0.iter().map(|kv| (&kv.key, &kv.value))
     }
-
-    /// Returns the underlying Vec of KeyValue pairs
-    pub(crate) fn into_vec(self) -> Vec<KeyValue> {
-        self.0
-    }
 }
 
 impl Hash for AttributeSet {
