src/histogram: Make Histogram::observe atomic across collects

benches/histogram.rs

@@ -15,7 +15,9 @@
 
 extern crate test;
 
-use prometheus::{Histogram, HistogramOpts, HistogramVec};
+use prometheus::{core::Collector, Histogram, HistogramOpts, HistogramVec};
+use std::sync::{atomic, Arc};
+use std::thread;
 use test::Bencher;
 
 #[bench]
@@ -99,3 +101,33 @@ fn bench_local_histogram_coarse_timer(b: &mut Bencher) {
     b.iter(|| local.start_coarse_timer());
     local.flush();
 }
+
+#[bench]
+fn concurrent_observe_and_collect(b: &mut Bencher) {
+    let signal_exit = Arc::new(atomic::AtomicBool::new(false));
+    let opts = HistogramOpts::new("test_name", "test help").buckets(vec![1.0]);
+    let histogram = Histogram::with_opts(opts).unwrap();
+
+    let mut handlers = vec![];
+
+    for _ in 0..4 {
+        let histogram = histogram.clone();
+        let signal_exit = signal_exit.clone();
+        handlers.push(thread::spawn(move || {
+            while !signal_exit.load(atomic::Ordering::Relaxed) {
+                for _ in 0..1_000 {
+                    histogram.observe(1.0);
+                }
+
+                histogram.collect();
+            }
+        }));
+    }
+
+    b.iter(|| histogram.observe(1.0));
+
+    signal_exit.store(true, atomic::Ordering::Relaxed);
+    for handler in handlers {
+        handler.join().unwrap();
+    }
+}

src/atomic64.rs

@@ -125,6 +125,13 @@ impl Atomic for AtomicF64 {
     }
 }
 
+impl AtomicF64 {
+    /// Store the value, returning the previous value.
+    pub fn swap(&self, val: f64) -> f64 {
+        u64_to_f64(self.inner.swap(f64_to_u64(val), Ordering::Relaxed))
+    }
+}
+
 /// A atomic signed integer.
 #[derive(Debug)]
 pub struct AtomicI64 {
@@ -188,7 +195,7 @@ impl Atomic for AtomicU64 {
 
     #[inline]
     fn inc_by(&self, delta: Self::T) {
-        self.inner.fetch_add(delta, Ordering::Relaxed);
+        self.inc_by_with_ordering(delta, Ordering::Relaxed);
     }
 
     #[inline]
@@ -197,6 +204,24 @@ impl Atomic for AtomicU64 {
     }
 }
 
+impl AtomicU64 {
+    /// Get the value with the provided memory ordering.
+    pub fn compare_and_swap(&self, current: u64, new: u64, ordering: Ordering) -> u64 {
+        self.inner.compare_and_swap(current, new, ordering)
+    }
+
+    /// Increment the value by a given amount with the provided memory ordering.
+    pub fn inc_by_with_ordering(&self, delta: u64, ordering: Ordering) {
+        self.inner.fetch_add(delta, ordering);
+    }
+
+    /// Swap the current value with the given value, returning that previously
+    /// current value.
+    pub fn swap(&self, val: u64) -> u64 {
+        self.inner.swap(val, Ordering::Relaxed)
+    }
+}
+
 #[cfg(test)]
 mod test {
     use std::f64::consts::PI;