Use ValueMap for Gauge - Throughput increased around 8x

opentelemetry-sdk/benches/metric_gauge.rs

@@ -6,7 +6,7 @@
     RAM: 64.0 GB
     | Test                           | Average time|
     |--------------------------------|-------------|
-    | Gauge_Add                      | 483.78 ns   |
+    | Gauge_Add                      | 178.37 ns   |
 */
 
 use criterion::{criterion_group, criterion_main, Criterion};

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

@@ -1,82 +1,71 @@
 use std::{
-    collections::{hash_map::Entry, HashMap},
-    sync::Mutex,
+    collections::HashSet,
+    sync::{atomic::Ordering, Arc},
     time::SystemTime,
 };
 
-use crate::{metrics::data::DataPoint, metrics::AttributeSet};
-use opentelemetry::{global, metrics::MetricsError, KeyValue};
+use crate::metrics::data::DataPoint;
+use opentelemetry::KeyValue;
 
-use super::{
-    aggregate::{is_under_cardinality_limit, STREAM_OVERFLOW_ATTRIBUTE_SET},
-    Number,
-};
-
-/// Timestamped measurement data.
-struct DataPointValue<T> {
-    timestamp: SystemTime,
-    value: T,
-}
+use super::{Assign, AtomicTracker, Number, ValueMap};
 
 /// Summarizes a set of measurements as the last one made.
-#[derive(Default)]
-pub(crate) struct LastValue<T> {
-    values: Mutex<HashMap<AttributeSet, DataPointValue<T>>>,
+pub(crate) struct LastValue<T: Number<T>> {
+    value_map: ValueMap<T, Assign>,
 }
 
 impl<T: Number<T>> LastValue<T> {
     pub(crate) fn new() -> Self {
-        Self::default()
+        LastValue {
+            value_map: ValueMap::new(),
+        }
     }
 
     pub(crate) fn measure(&self, measurement: T, attrs: &[KeyValue]) {
-        let d: DataPointValue<T> = DataPointValue {
-            timestamp: SystemTime::now(),
-            value: measurement,
-        };
-
-        let attrs: AttributeSet = attrs.into();
-        if let Ok(mut values) = self.values.lock() {
-            let size = values.len();
-            match values.entry(attrs) {
-                Entry::Occupied(mut occupied_entry) => {
-                    occupied_entry.insert(d);
-                }
-                Entry::Vacant(vacant_entry) => {
-                    if is_under_cardinality_limit(size) {
-                        vacant_entry.insert(d);
-                    } else {
-                        values.insert(STREAM_OVERFLOW_ATTRIBUTE_SET.clone(), d);
-                        global::handle_error(MetricsError::Other("Warning: Maximum data points for metric stream exceeded. Entry added to overflow.".into()));
-                    }
-                }
-            }
-        }
+        self.value_map.measure(measurement, attrs);
     }
 
     pub(crate) fn compute_aggregation(&self, dest: &mut Vec<DataPoint<T>>) {
+        let t = SystemTime::now();
         dest.clear();
-        let mut values = match self.values.lock() {
-            Ok(guard) if !guard.is_empty() => guard,
-            _ => return,
-        };
 
-        let n = values.len();
+        // Max number of data points need to account for the special casing
+        // of the no attribute value + overflow attribute.
+        let n = self.value_map.count.load(Ordering::SeqCst) + 2;
         if n > dest.capacity() {
             dest.reserve_exact(n - dest.capacity());
         }
 
-        for (attrs, value) in values.drain() {
+        if self
+            .value_map
+            .has_no_attribute_value
+            .swap(false, Ordering::AcqRel)
+        {
             dest.push(DataPoint {
-                attributes: attrs
-                    .iter()
-                    .map(|(k, v)| KeyValue::new(k.clone(), v.clone()))
-                    .collect(),
-                time: Some(value.timestamp),
-                value: value.value,
+                attributes: vec![],
                 start_time: None,
+                time: Some(t),
+                value: self.value_map.no_attribute_tracker.get_and_reset_value(),
                 exemplars: vec![],
             });
         }
+
+        let mut trackers = match self.value_map.trackers.write() {
+            Ok(v) => v,
+            _ => return,
+        };
+
+        let mut seen = HashSet::new();
+        for (attrs, tracker) in trackers.drain() {
+            if seen.insert(Arc::as_ptr(&tracker)) {
+                dest.push(DataPoint {
+                    attributes: attrs.clone(),
+                    start_time: None,
+                    time: Some(t),
+                    value: tracker.get_value(),
+                    exemplars: vec![],
+                });
+            }
+        }
     }
 }

stress/src/metrics_gauge.rs

@@ -3,7 +3,7 @@
     OS: Ubuntu 22.04.4 LTS (5.15.153.1-microsoft-standard-WSL2)
     Hardware: Intel(R) Xeon(R) Platinum 8370C CPU @ 2.80GHz, 16vCPUs,
     RAM: 64.0 GB
-    ~1.5 M/sec
+    ~11.5 M/sec
 */
 
 use lazy_static::lazy_static;