Reduce cost of no attribute counters

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

@@ -6,10 +6,26 @@
 
 use core::fmt;
 use std::ops::{Add, AddAssign, Sub};
+use std::sync::atomic::{AtomicI64, AtomicU64, Ordering};
+use std::sync::Mutex;
 
 pub(crate) use aggregate::{AggregateBuilder, ComputeAggregation, Measure};
 pub(crate) use exponential_histogram::{EXPO_MAX_SCALE, EXPO_MIN_SCALE};
 
+/// Marks a type that can have a value added and retrieved atomically. Required since
+/// different types have different backing atomic mechanisms
+pub(crate) trait AtomicTracker<T>: Sync + Send + 'static {
+    fn add(&self, value: T);
+    fn get_value(&self) -> T;
+    fn get_and_reset_value(&self) -> T;
+}
+
+/// Marks a type that can have an atomic tracker generated for it
+pub(crate) trait AtomicallyUpdate<T> {
+    type AtomicTracker: AtomicTracker<T>;
+    fn new_atomic_tracker() -> Self::AtomicTracker;
+}
+
 pub(crate) trait Number<T>:
     Add<Output = T>
     + AddAssign
@@ -23,6 +39,7 @@
     + Send
     + Sync
     + 'static
+    + AtomicallyUpdate<T>
 {
     fn min() -> Self;
     fn max() -> Self;
@@ -71,3 +88,87 @@
         self
     }
 }
+
+impl AtomicTracker<u64> for AtomicU64 {
+    fn add(&self, value: u64) {
+        self.fetch_add(value, Ordering::Relaxed);
+    }
+
+    fn get_value(&self) -> u64 {
+        self.load(Ordering::Relaxed)
+    }
+
+    fn get_and_reset_value(&self) -> u64 {
+        self.swap(0, Ordering::Relaxed)
+    }
+}
+
+impl AtomicallyUpdate<u64> for u64 {
+    type AtomicTracker = AtomicU64;
+
+    fn new_atomic_tracker() -> Self::AtomicTracker {
+        AtomicU64::new(0)
+    }
+}
+
+impl AtomicTracker<i64> for AtomicI64 {
+    fn add(&self, value: i64) {
+        self.fetch_add(value, Ordering::Relaxed);
+    }
+
+    fn get_value(&self) -> i64 {
+        self.load(Ordering::Relaxed)
+    }
+
+    fn get_and_reset_value(&self) -> i64 {
+        self.swap(0, Ordering::Relaxed)
+    }
+}
+
+impl AtomicallyUpdate<i64> for i64 {
+    type AtomicTracker = AtomicI64;
+
+    fn new_atomic_tracker() -> Self::AtomicTracker {
+        AtomicI64::new(0)
+    }
+}
+
+pub(crate) struct F64AtomicTracker {
+    inner: Mutex<f64>, // Floating points don't have true atomics, so we need to use mutex for them
+}
+
+impl F64AtomicTracker {
+    fn new() -> Self {
+        F64AtomicTracker {
+            inner: Mutex::new(0.0),
+        }
+    }
+}
+
+impl AtomicTracker<f64> for F64AtomicTracker {
+    fn add(&self, value: f64) {
+        let mut guard = self.inner.lock().expect("F64 mutex was poisoned");
+        *guard += value;
+    }
+
+    fn get_value(&self) -> f64 {
+        let guard = self.inner.lock().expect("F64 mutex was poisoned");
+        *guard
+    }
+
+    fn get_and_reset_value(&self) -> f64 {
+        let mut guard = self.inner.lock().expect("F64 mutex was poisoned");
+        let value = *guard;
+        *guard = 0.0;
+
+        value
+    }
+}
+
+impl AtomicallyUpdate<f64> for f64 {
+    type AtomicTracker = F64AtomicTracker;
+
+    fn new_atomic_tracker() -> Self::AtomicTracker {
+        F64AtomicTracker::new()
+    }
+}

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

@@ -1,3 +1,4 @@
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::{
     collections::{hash_map::Entry, HashMap},
     sync::Mutex,
@@ -10,26 +11,39 @@
 
 use super::{
     aggregate::{is_under_cardinality_limit, STREAM_OVERFLOW_ATTRIBUTE_SET},
-    Number,
+    AtomicTracker, Number,
 };
 
 /// The storage for sums.
-#[derive(Default)]
 struct ValueMap<T: Number<T>> {
     values: Mutex<HashMap<AttributeSet, T>>,
+    has_no_value_attribute_value: AtomicBool,
+    no_attribute_value: T::AtomicTracker,
+}
+
+impl<T: Number<T>> Default for ValueMap<T> {
+    fn default() -> Self {
+        ValueMap::new()
+    }
 }
 
 impl<T: Number<T>> ValueMap<T> {
     fn new() -> Self {
         ValueMap {
             values: Mutex::new(HashMap::new()),
+            has_no_value_attribute_value: AtomicBool::new(false),
+            no_attribute_value: T::new_atomic_tracker(),
         }
     }
 }
 
 impl<T: Number<T>> ValueMap<T> {
     fn measure(&self, measurement: T, attrs: AttributeSet) {
-        if let Ok(mut values) = self.values.lock() {
+        if attrs.is_empty() {
+            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) => {
@@ -103,14 +117,28 @@
             Err(_) => return (0, None),
         };
 
-        let n = values.len();
+        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 prev_start = self.start.lock().map(|start| *start).unwrap_or(t);
+        if self
+            .value_map
+            .has_no_value_attribute_value
+            .swap(false, Ordering::AcqRel)
+        {
+            s_data.data_points.push(DataPoint {
+                attributes: AttributeSet::default(),
+                start_time: Some(prev_start),
+                time: Some(t),
+                value: self.value_map.no_attribute_value.get_and_reset_value(),
+                exemplars: vec![],
+            });
+        }
+
         for (attrs, value) in values.drain() {
             s_data.data_points.push(DataPoint {
                 attributes: attrs,
@@ -126,7 +154,10 @@
             *start = t;
         }
 
-        (n, new_agg.map(|a| Box::new(a) as Box<_>))
+        (
+            s_data.data_points.len(),
+            new_agg.map(|a| Box::new(a) as Box<_>),
+        )
     }
 
     pub(crate) fn cumulative(
@@ -155,14 +186,29 @@
             Err(_) => return (0, None),
         };
 
-        let n = values.len();
+        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 prev_start = self.start.lock().map(|start| *start).unwrap_or(t);
+
+        if self
+            .value_map
+            .has_no_value_attribute_value
+            .load(Ordering::Acquire)
+        {
+            s_data.data_points.push(DataPoint {
+                attributes: AttributeSet::default(),
+                start_time: Some(prev_start),
+                time: Some(t),
+                value: self.value_map.no_attribute_value.get_value(),
+                exemplars: vec![],
+            });
+        }
+
         // TODO: This will use an unbounded amount of memory if there
         // are unbounded number of attribute sets being aggregated. Attribute
         // sets that become "stale" need to be forgotten so this will not
@@ -177,7 +223,10 @@
             });
         }
 
-        (n, new_agg.map(|a| Box::new(a) as Box<_>))
+        (
+            s_data.data_points.len(),
+            new_agg.map(|a| Box::new(a) as Box<_>),
+        )
     }
 }
 
@@ -230,7 +279,7 @@
             Err(_) => return (0, None),
         };
 
-        let n = values.len();
+        let n = values.len() + 1;
         if n > s_data.data_points.capacity() {
             s_data
                 .data_points
@@ -242,6 +291,20 @@
             Err(_) => return (0, None),
         };
 
+        if self
+            .value_map
+            .has_no_value_attribute_value
+            .swap(false, Ordering::AcqRel)
+        {
+            s_data.data_points.push(DataPoint {
+                attributes: AttributeSet::default(),
+                start_time: Some(prev_start),
+                time: Some(t),
+                value: self.value_map.no_attribute_value.get_and_reset_value(),
+                exemplars: vec![],
+            });
+        }
+
         let default = T::default();
         for (attrs, value) in values.drain() {
             let delta = value - *reported.get(&attrs).unwrap_or(&default);
@@ -265,7 +328,10 @@
         *reported = new_reported;
         drop(reported); // drop before values guard is dropped
 
-        (n, new_agg.map(|a| Box::new(a) as Box<_>))
+        (
+            s_data.data_points.len(),
+            new_agg.map(|a| Box::new(a) as Box<_>),
+        )
     }
 
     pub(crate) fn cumulative(
@@ -295,7 +361,7 @@
             Err(_) => return (0, None),
         };
 
-        let n = values.len();
+        let n = values.len() + 1;
         if n > s_data.data_points.capacity() {
             s_data
                 .data_points
@@ -307,6 +373,20 @@
             Err(_) => return (0, None),
         };
 
+        if self
+            .value_map
+            .has_no_value_attribute_value
+            .load(Ordering::Acquire)
+        {
+            s_data.data_points.push(DataPoint {
+                attributes: AttributeSet::default(),
+                start_time: Some(prev_start),
+                time: Some(t),
+                value: self.value_map.no_attribute_value.get_value(),
+                exemplars: vec![],
+            });
+        }
+
         let default = T::default();
         for (attrs, value) in values.iter() {
             let delta = *value - *reported.get(attrs).unwrap_or(&default);
@@ -325,6 +405,9 @@
         *reported = new_reported;
         drop(reported); // drop before values guard is dropped
 
-        (n, new_agg.map(|a| Box::new(a) as Box<_>))
+        (
+            s_data.data_points.len(),
+            new_agg.map(|a| Box::new(a) as Box<_>),
+        )
     }
 }