Adding two level hashing in metrics hashmap

opentelemetry-sdk/Cargo.toml

@@ -12,12 +12,14 @@ rust-version = "1.65"
 [dependencies]
 opentelemetry = { version = "0.21", path = "../opentelemetry/" }
 opentelemetry-http = { version = "0.10", path = "../opentelemetry-http", optional = true }
+ahash = { version = "0.8", optional = true }
 async-std = { workspace = true, features = ["unstable"], optional = true }
 async-trait = { workspace = true, optional = true }
 crossbeam-channel = { version = "0.5", optional = true }
 futures-channel = "0.3"
 futures-executor = { workspace = true }
 futures-util = { workspace = true, features = ["std", "sink", "async-await-macro"] }
+hashbrown = { version = "0.14", optional = true }
 once_cell = { workspace = true }
 ordered-float = { workspace = true }
 percent-encoding = { version = "2.0", optional = true }
@@ -54,6 +56,7 @@ testing = ["opentelemetry/testing", "trace", "metrics", "logs", "rt-async-std",
 rt-tokio = ["tokio", "tokio-stream"]
 rt-tokio-current-thread = ["tokio", "tokio-stream"]
 rt-async-std = ["async-std"]
+use_hashbrown = ["hashbrown", "ahash"]
 
 [[bench]]
 name = "context"

opentelemetry-sdk/src/metrics/internal/sum.rs

@@ -1,22 +1,35 @@
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::{
-    collections::{hash_map::Entry, HashMap},
-    sync::Mutex,
+    sync::{Arc, Mutex},
     time::SystemTime,
 };
 
 use crate::attributes::AttributeSet;
 use crate::metrics::data::{self, Aggregation, DataPoint, Temporality};
 use opentelemetry::{global, metrics::MetricsError};
+use std::hash::{Hash, Hasher};
+
+#[cfg(feature = "use_hashbrown")]
+use ahash::AHasher;
+#[cfg(feature = "use_hashbrown")]
+use hashbrown::{hash_map::Entry, HashMap};
+
+#[cfg(not(feature = "use_hashbrown"))]
+use std::collections::{
+    hash_map::{DefaultHasher, Entry},
+    HashMap,
+};
 
 use super::{
     aggregate::{is_under_cardinality_limit, STREAM_OVERFLOW_ATTRIBUTE_SET},
     AtomicTracker, Number,
 };
 
+type BucketValue<T> = Mutex<Option<HashMap<AttributeSet, T>>>;
+type Buckets<T> = Arc<[BucketValue<T>; 256]>;
 /// The storage for sums.
 struct ValueMap<T: Number<T>> {
-    values: Mutex<HashMap<AttributeSet, T>>,
+    buckets: Buckets<T>,
     has_no_value_attribute_value: AtomicBool,
     no_attribute_value: T::AtomicTracker,
 }
@@ -29,12 +42,41 @@
 
 impl<T: Number<T>> ValueMap<T> {
     fn new() -> Self {
+        let buckets = std::iter::repeat_with(|| Mutex::new(None))
+            .take(256)
+            .collect::<Vec<_>>()
+            .try_into()
+            .unwrap_or_else(|_| panic!("Incorrect length"));
+
         ValueMap {
-            values: Mutex::new(HashMap::new()),
+            buckets: Arc::new(buckets),
             has_no_value_attribute_value: AtomicBool::new(false),
             no_attribute_value: T::new_atomic_tracker(),
         }
     }
+
+    // Hash function to determine the bucket
+    fn hash_to_bucket(key: &AttributeSet) -> u8 {
+        #[cfg(not(feature = "use_hashbrown"))]
+        let mut hasher = DefaultHasher::new();
+        #[cfg(feature = "use_hashbrown")]
+        let mut hasher = AHasher::default();
+
+        key.hash(&mut hasher);
+        // Use the 8 least significant bits directly, avoiding the modulus operation.
+        hasher.finish() as u8
+    }
+
+    // Calculate the total length of data points across all buckets.
+    fn total_data_points_count(&self) -> usize {
+        self.buckets
+            .iter()
+            .map(|bucket_mutex| {
+                let locked_bucket = bucket_mutex.lock().unwrap();
+                locked_bucket.as_ref().map_or(0, |bucket| bucket.len())
+            })
+            .sum::<usize>()
+    }
 }
 
 impl<T: Number<T>> ValueMap<T> {
@@ -43,22 +85,33 @@
             self.no_attribute_value.add(measurement);
             self.has_no_value_attribute_value
                 .store(true, Ordering::Release);
-        } else if let Ok(mut values) = self.values.lock() {
-            let size = values.len();
-            match values.entry(attrs) {
-                Entry::Occupied(mut occupied_entry) => {
-                    let sum = occupied_entry.get_mut();
-                    *sum += measurement;
-                }
-                Entry::Vacant(vacant_entry) => {
-                    if is_under_cardinality_limit(size) {
-                        vacant_entry.insert(measurement);
-                    } else {
-                        values
-                            .entry(STREAM_OVERFLOW_ATTRIBUTE_SET.clone())
-                            .and_modify(|val| *val += measurement)
-                            .or_insert(measurement);
-                        global::handle_error(MetricsError::Other("Warning: Maximum data points for metric stream exceeded. Entry added to overflow.".into()));
+        } else {
+            let bucket_index = Self::hash_to_bucket(&attrs) as usize; // Ensure index is usize for array indexing
+            let bucket_mutex = &self.buckets[bucket_index];
+            let mut bucket_guard = bucket_mutex.lock().unwrap();
+
+            if bucket_guard.is_none() {
+                *bucket_guard = Some(HashMap::new()); // Initialize the bucket if it's None
+            }
+
+            if let Some(ref mut values) = *bucket_guard {
+                let size = values.len();
+                match values.entry(attrs) {
+                    Entry::Occupied(mut occupied_entry) => {
+                        let sum = occupied_entry.get_mut();
+                        *sum += measurement;
+                    }
+                    Entry::Vacant(vacant_entry) => {
+                        if is_under_cardinality_limit(size) {
+                            vacant_entry.insert(measurement);
+                        } else {
+                            // TBD - Update total_count ??
+                            values
+                                .entry(STREAM_OVERFLOW_ATTRIBUTE_SET.clone())
+                                .and_modify(|val| *val += measurement)
+                                .or_insert(measurement);
+                            global::handle_error(MetricsError::Other("Warning: Maximum data points for metric stream exceeded. Entry added to overflow.".into()));
+                        }
                     }
                 }
             }
@@ -112,16 +165,10 @@
         s_data.is_monotonic = self.monotonic;
         s_data.data_points.clear();
 
-        let mut values = match self.value_map.values.lock() {
-            Ok(v) => v,
-            Err(_) => return (0, None),
-        };
-
-        let n = values.len() + 1;
-        if n > s_data.data_points.capacity() {
-            s_data
-                .data_points
-                .reserve_exact(n - s_data.data_points.capacity());
+        let total_len: usize = self.value_map.total_data_points_count() + 1;
+        if total_len > s_data.data_points.capacity() {
+            let additional_space_needed = total_len - s_data.data_points.capacity();
+            s_data.data_points.reserve_exact(additional_space_needed);
         }
 
         let prev_start = self.start.lock().map(|start| *start).unwrap_or(t);
@@ -139,14 +186,19 @@
             });
         }
 
-        for (attrs, value) in values.drain() {
-            s_data.data_points.push(DataPoint {
-                attributes: attrs,
-                start_time: Some(prev_start),
-                time: Some(t),
-                value,
-                exemplars: vec![],
-            });
+        for bucket_mutex in self.value_map.buckets.iter() {
+            if let Some(ref mut locked_bucket) = *bucket_mutex.lock().unwrap() {
+                for (attrs, value) in locked_bucket.drain() {
+                    s_data.data_points.push(DataPoint {
+                        attributes: attrs,
+                        start_time: Some(*self.start.lock().unwrap()),
+                        time: Some(t),
+                        value,
+                        exemplars: vec![],
+                    });
+                }
+                // The bucket is automatically cleared by the .drain() method
+            }
         }
 
         // The delta collection cycle resets.
@@ -181,16 +233,10 @@
         s_data.is_monotonic = self.monotonic;
         s_data.data_points.clear();
 
-        let values = match self.value_map.values.lock() {
-            Ok(v) => v,
-            Err(_) => return (0, None),
-        };
-
-        let n = values.len() + 1;
-        if n > s_data.data_points.capacity() {
-            s_data
-                .data_points
-                .reserve_exact(n - s_data.data_points.capacity());
+        let total_len: usize = self.value_map.total_data_points_count() + 1;
+        if total_len > s_data.data_points.capacity() {
+            let additional_space_needed = total_len - s_data.data_points.capacity();
+            s_data.data_points.reserve_exact(additional_space_needed);
         }
 
         let prev_start = self.start.lock().map(|start| *start).unwrap_or(t);
@@ -213,14 +259,18 @@
         // are unbounded number of attribute sets being aggregated. Attribute
         // sets that become "stale" need to be forgotten so this will not
         // overload the system.
-        for (attrs, value) in values.iter() {
-            s_data.data_points.push(DataPoint {
-                attributes: attrs.clone(),
-                start_time: Some(prev_start),
-                time: Some(t),
-                value: *value,
-                exemplars: vec![],
-            });
+        for bucket_mutex in self.value_map.buckets.iter() {
+            if let Some(ref locked_bucket) = *bucket_mutex.lock().unwrap() {
+                for (attrs, value) in locked_bucket.iter() {
+                    s_data.data_points.push(DataPoint {
+                        attributes: attrs.clone(),
+                        start_time: Some(*self.start.lock().unwrap()), // Consider last reset time
+                        time: Some(t),
+                        value: *value,
+                        exemplars: vec![],
+                    });
+                }
+            }
         }
 
         (
@@ -274,18 +324,13 @@
         s_data.temporality = Temporality::Delta;
         s_data.is_monotonic = self.monotonic;
 
-        let mut values = match self.value_map.values.lock() {
-            Ok(v) => v,
-            Err(_) => return (0, None),
-        };
-
-        let n = values.len() + 1;
-        if n > s_data.data_points.capacity() {
-            s_data
-                .data_points
-                .reserve_exact(n - s_data.data_points.capacity());
+        let total_len: usize = self.value_map.total_data_points_count() + 1;
+        if total_len > s_data.data_points.capacity() {
+            let additional_space_needed = total_len - s_data.data_points.capacity();
+            s_data.data_points.reserve_exact(additional_space_needed);
         }
-        let mut new_reported = HashMap::with_capacity(n);
+
+        let mut new_reported = HashMap::with_capacity(total_len);
         let mut reported = match self.reported.lock() {
             Ok(r) => r,
             Err(_) => return (0, None),
@@ -305,19 +350,23 @@
             });
         }
 
-        let default = T::default();
-        for (attrs, value) in values.drain() {
-            let delta = value - *reported.get(&attrs).unwrap_or(&default);
-            if delta != default {
-                new_reported.insert(attrs.clone(), value);
+        for bucket_mutex in self.value_map.buckets.iter() {
+            if let Some(ref mut locked_bucket) = *bucket_mutex.lock().unwrap() {
+                let default = T::default();
+                for (attrs, value) in locked_bucket.drain() {
+                    let delta = value - *reported.get(&attrs).unwrap_or(&default);
+                    if delta != default {
+                        new_reported.insert(attrs.clone(), value);
+                    }
+                    s_data.data_points.push(DataPoint {
+                        attributes: attrs.clone(),
+                        start_time: Some(prev_start),
+                        time: Some(t),
+                        value: delta,
+                        exemplars: vec![],
+                    });
+                }
             }
-            s_data.data_points.push(DataPoint {
-                attributes: attrs.clone(),
-                start_time: Some(prev_start),
-                time: Some(t),
-                value: delta,
-                exemplars: vec![],
-            });
         }
 
         // The delta collection cycle resets.
@@ -356,18 +405,13 @@
         s_data.temporality = Temporality::Cumulative;
         s_data.is_monotonic = self.monotonic;
 
-        let values = match self.value_map.values.lock() {
-            Ok(v) => v,
-            Err(_) => return (0, None),
-        };
-
-        let n = values.len() + 1;
-        if n > s_data.data_points.capacity() {
-            s_data
-                .data_points
-                .reserve_exact(n - s_data.data_points.capacity());
+        let total_len: usize = self.value_map.total_data_points_count() + 1;
+        if total_len > s_data.data_points.capacity() {
+            let additional_space_needed = total_len - s_data.data_points.capacity();
+            s_data.data_points.reserve_exact(additional_space_needed);
         }
-        let mut new_reported = HashMap::with_capacity(n);
+
+        let mut new_reported = HashMap::with_capacity(total_len);
         let mut reported = match self.reported.lock() {
             Ok(r) => r,
             Err(_) => return (0, None),
@@ -388,18 +432,22 @@
         }
 
         let default = T::default();
-        for (attrs, value) in values.iter() {
-            let delta = *value - *reported.get(attrs).unwrap_or(&default);
-            if delta != default {
-                new_reported.insert(attrs.clone(), *value);
+        for bucket_mutex in self.value_map.buckets.iter() {
+            if let Some(ref locked_bucket) = *bucket_mutex.lock().unwrap() {
+                for (attrs, value) in locked_bucket.iter() {
+                    let delta = *value - *reported.get(attrs).unwrap_or(&default);
+                    if delta != default {
+                        new_reported.insert(attrs.clone(), *value);
+                    }
+                    s_data.data_points.push(DataPoint {
+                        attributes: attrs.clone(),
+                        start_time: Some(prev_start),
+                        time: Some(t),
+                        value: *value, // For cumulative, we use the value directly without calculating delta
+                        exemplars: vec![],
+                    });
+                }
             }
-            s_data.data_points.push(DataPoint {
-                attributes: attrs.clone(),
-                start_time: Some(prev_start),
-                time: Some(t),
-                value: delta,
-                exemplars: vec![],
-            });
         }
 
         *reported = new_reported;

opentelemetry-sdk/src/testing/metrics/in_memory_exporter.rs

@@ -285,6 +285,9 @@ impl TemporalitySelector for InMemoryMetricsExporter {
 #[async_trait]
 impl PushMetricsExporter for InMemoryMetricsExporter {
     async fn export(&self, metrics: &mut ResourceMetrics) -> Result<()> {
+        if metrics.scope_metrics.is_empty() || metrics.scope_metrics[0].metrics.is_empty() {
+            return Ok(());
+        }
         self.metrics
             .lock()
             .map(|mut metrics_guard| {

stress/Cargo.toml

@@ -24,7 +24,7 @@ ctrlc = "3.2.5"
 lazy_static = "1.4.0"
 num_cpus = "1.15.0"
 opentelemetry = { path = "../opentelemetry", features = ["metrics", "logs", "trace"] }
-opentelemetry_sdk = { path = "../opentelemetry-sdk", features = ["metrics", "logs", "trace"] }
+opentelemetry_sdk = { path = "../opentelemetry-sdk", features = ["metrics", "logs", "trace", "use_hashbrown"] }
 opentelemetry-appender-tracing = { path = "../opentelemetry-appender-tracing"}
 rand = { version = "0.8.4", features = ["small_rng"] }
 tracing = { workspace = true, features = ["std"]}