*: Use compare_exchange and fix sync import (#381)

* src/: Replace deprecated compare_and_swap with compare_exchange(_weak)

`compare_and_swap` is deprecated in favor of `compare_exchange` and
`compare_exchange_weak`. This commit replaces the former with either of
the two latter options.

Signed-off-by: Max Inden <[email protected]>

* static/metric/: use TokenStream from proc_macro

Signed-off-by: Federico Garcia Ronca <[email protected]>

Co-authored-by: Federico Garcia Ronca <[email protected]>

Author: mxinden

src/atomic64.rs

@@ -110,10 +110,13 @@ impl Atomic for AtomicF64 {
         loop {
             let current = self.inner.load(Ordering::Acquire);
             let new = u64_to_f64(current) + delta;
-            let swapped = self
-                .inner
-                .compare_and_swap(current, f64_to_u64(new), Ordering::Release);
-            if swapped == current {
+            let result = self.inner.compare_exchange_weak(
+                current,
+                f64_to_u64(new),
+                Ordering::Release,
+                Ordering::Relaxed,
+            );
+            if result.is_ok() {
                 return;
             }
         }
@@ -205,9 +208,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)
+    /// Stores a value into the atomic integer if the current value is the same
+    /// as the current value.
+    ///
+    /// This function is allowed to spuriously fail even when the comparison
+    /// succeeds, which can result in more efficient code on some platforms. The
+    /// return value is a result indicating whether the new value was written
+    /// and containing the previous value.
+    ///
+    /// See [`StdAtomicU64`] for details.
+    pub(crate) fn compare_exchange_weak(
+        &self,
+        current: u64,
+        new: u64,
+        success: Ordering,
+        failure: Ordering,
+    ) -> Result<u64, u64> {
+        self.inner
+            .compare_exchange_weak(current, new, success, failure)
     }
 
     /// Increment the value by a given amount with the provided memory ordering.

src/histogram.rs

@@ -410,18 +410,21 @@ impl HistogramCore {
         // shards were flipped, all in-progress `observe` calls are done. With
         // all of them done, the cold shard is now in a consistent state.
         //
-        // `observe` uses `Release` ordering. `compare_and_swap` needs to use
+        // `observe` uses `Release` ordering. `compare_exchange` needs to use
         // `Acquire` ordering to ensure that (1) one sees all the previous
         // `observe` stores to the counter and (2) to ensure the below shard
         // modifications happen after this point, thus the shard is not modified
         // by any `observe` operations.
-        while overall_count
-            != cold_shard.count.compare_and_swap(
+        while cold_shard
+            .count
+            .compare_exchange_weak(
                 overall_count,
                 // While at it, reset cold shard count on success.
                 0,
                 Ordering::Acquire,
+                Ordering::Acquire,
             )
+            .is_err()
         {}
 
         // Get cold shard sum and reset to 0.

src/timer.rs

@@ -45,7 +45,10 @@ const CHECK_UPDATE_INTERVAL: Duration = Duration::from_millis(200);
 
 /// Ensures background updater is running, which will call `now_millis` periodically.
 pub fn ensure_updater() {
-    if !UPDATER_IS_RUNNING.compare_and_swap(false, true, Ordering::SeqCst) {
+    if UPDATER_IS_RUNNING
+        .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
+        .is_ok()
+    {
         std::thread::Builder::new()
             .name("time updater".to_owned())
             .spawn(|| loop {

static-metric/src/auto_flush_from.rs

@@ -1,7 +1,7 @@
 // Copyright 2019 TiKV Project Authors. Licensed under Apache-2.0.
 
+use proc_macro::TokenStream;
 use proc_macro2::Span;
-use syn::export::TokenStream2;
 use syn::parse::{Parse, ParseStream};
 use syn::token::*;
 use syn::*;
@@ -38,7 +38,7 @@ impl Parse for AutoFlushFromDef {
 }
 
 impl AutoFlushFromDef {
-    pub fn auto_flush_from(&self) -> TokenStream2 {
+    pub fn auto_flush_from(&self) -> TokenStream {
         let inner_class_name = self.inner_class_name.clone();
         let class_name = self.class_name.clone();
         let source_var_name = self.source_var_name.clone();
@@ -50,13 +50,14 @@ impl AutoFlushFromDef {
             }
             None => quote! {},
         };
-        quote! {
+        let token_stream_inner = quote! {
             {
                 thread_local! {
                     static INNER: #inner_class_name = #inner_class_name::from(& #source_var_name)#update_duration;
                 }
                 #class_name::from(&INNER)
             }
-        }
+        };
+        TokenStream::from(token_stream_inner)
     }
 }

static-metric/src/lib.rs

@@ -63,7 +63,7 @@ pub fn make_auto_flush_static_metric(input: TokenStream) -> TokenStream {
 #[proc_macro]
 pub fn auto_flush_from(input: TokenStream) -> TokenStream {
     let def: AutoFlushFromDef = syn::parse(input).unwrap();
-    def.auto_flush_from().into()
+    def.auto_flush_from()
 }
 
 /// Register a `CounterVec` and create static metrics from it.