metrics: enforce tighter Atomic ordering

Metric values are accessed from all threads, so
enforce Release/Acquire semantics to make sure
that we always use the latest values.

Signed-off-by: alindima <[email protected]>

Author: alindima

src/logger/src/metrics.rs

@@ -116,15 +116,15 @@ impl<T: Serialize> Metrics<T> {
     ///
     /// * `metrics_dest` - Buffer for JSON formatted metrics. Needs to implement `Write` and `Send`.
     pub fn init(&self, metrics_dest: Box<dyn Write + Send>) -> Result<(), MetricsError> {
-        if self.is_initialized.load(Ordering::Relaxed) {
+        if self.is_initialized.load(Ordering::Acquire) {
             return Err(MetricsError::AlreadyInitialized);
         }
         {
             let mut g = extract_guard(self.metrics_buf.lock());
 
             *g = Some(metrics_dest);
         }
-        self.is_initialized.store(true, Ordering::Relaxed);
+        self.is_initialized.store(true, Ordering::Release);
         Ok(())
     }
 
@@ -148,7 +148,7 @@ impl<T: Serialize> Metrics<T> {
     /// The alternative is to hold a Mutex over the entire function call, but this increases the
     /// known deadlock potential.
     pub fn write(&self) -> Result<bool, MetricsError> {
-        if self.is_initialized.load(Ordering::Relaxed) {
+        if self.is_initialized.load(Ordering::Acquire) {
             match serde_json::to_string(&self.app_metrics) {
                 Ok(msg) => {
                     if let Some(guard) = extract_guard(self.metrics_buf.lock()).as_mut() {
@@ -251,48 +251,45 @@ pub struct SharedIncMetric(AtomicUsize, AtomicUsize);
 pub struct SharedStoreMetric(AtomicUsize);
 
 impl IncMetric for SharedIncMetric {
-    // While the order specified for this operation is still Relaxed, the actual instruction will
-    // be an asm "LOCK; something" and thus atomic across multiple threads, simply because of the
-    // fetch_and_add (as opposed to "store(load() + 1)") implementation for atomics.
-    // TODO: would a stronger ordering make a difference here?
     fn add(&self, value: usize) {
-        self.0.fetch_add(value, Ordering::Relaxed);
+        self.0.fetch_add(value, Ordering::AcqRel);
     }
 
     fn count(&self) -> usize {
-        self.0.load(Ordering::Relaxed)
+        self.0.load(Ordering::Acquire)
     }
 }
 
 impl StoreMetric for SharedStoreMetric {
     fn fetch(&self) -> usize {
-        self.0.load(Ordering::Relaxed)
+        self.0.load(Ordering::Acquire)
     }
 
     fn store(&self, value: usize) {
-        self.0.store(value, Ordering::Relaxed);
+        self.0.store(value, Ordering::Release);
     }
 }
 
 impl Serialize for SharedIncMetric {
     /// Reset counters of each metrics. Here we suppose that Serialize's goal is to help with the
     /// flushing of metrics.
     /// !!! Any print of the metrics will also reset them. Use with caution !!!
+    /// !!! This function is not thread safe !!!
     fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
         // There's no serializer.serialize_usize() for some reason :(
-        let snapshot = self.0.load(Ordering::Relaxed);
-        let res = serializer.serialize_u64(snapshot as u64 - self.1.load(Ordering::Relaxed) as u64);
+        let snapshot = self.count();
+        let res = serializer.serialize_u64(snapshot as u64 - self.1.load(Ordering::Acquire) as u64);
 
         if res.is_ok() {
-            self.1.store(snapshot, Ordering::Relaxed);
+            self.1.store(snapshot, Ordering::Release);
         }
         res
     }
 }
 
 impl Serialize for SharedStoreMetric {
     fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
-        serializer.serialize_u64(self.0.load(Ordering::Relaxed) as u64)
+        serializer.serialize_u64(self.0.load(Ordering::Acquire) as u64)
     }
 }