Use atomics instead of synchronized for Gauge (#482)

* Use atomic update instead of synchronized for Gauge
* Introduce DoubleAdder#set to avoid allocations in Gauge#set

Signed-off-by: Rudolf Rakos <[email protected]>

Author: Rudolf Rakos

simpleclient/src/main/java/io/prometheus/client/DoubleAdder.java

@@ -94,8 +94,37 @@ public void add(double x) {
      * @return the sum
      */
     public double sum() {
-        Cell[] as = cells;
-        double sum = Double.longBitsToDouble(base);
+        // On concurrent `sum` and `set`, it is acceptable to `get` an outdated `value`.
+        // On concurrent `sum` and `add`, it is acceptable to `get` an outdated `value`.
+        // On concurrent `sum` and `set` and `add`, it is possible to `get` an outdated `value`.
+
+        // Correctness is guaranteed by `volatile` memory access ordering and visibility semantics.
+        // Program order:
+        //  - writes in `set` - `busy` (CAS), `cells` (Wc), `base` (Wb), `busy`
+        //  - reads in `sum` - `cells` (Rc), `base` (Rb), `busy`, `cells` (Cc), `base` (Cb)
+        // Note that:
+        //  - `busy` is written after `cells` and `base`
+        //  - `busy` is read after `cells` and `base`, then `cells` and `base` is re-read after `busy`
+        // In other words:
+        //  - if we see the write to `busy`, then we must see the write to `cells` and `busy` on re-read
+        //  - if we don't see the write to `busy`, then we must retry as we have no guarantees
+        // Execution order (in the former case):
+        //  - serial
+        //    - old result - Rc, Rb, Cc, Cb, Wc, Wb
+        //    - new result - Wc, Wb, Rc, Rb, Cc, Cb
+        //  - concurrent
+        //    - old result - Rc, Wc, Rb, Wb, Cc, Cb - retry (superfluous)
+        //    - new result - Wc, Rc, Wb, Rb, Cc, Cb
+        //    - invalid result - Rc, Wc, Wb, Rb, Cc, Cb - retry
+        //    - invalid result - Wc, Rc, Rb, Wb, Cc, Cb - retry
+        Cell[] as = cells; long b = base;
+        while (as != null && !(busy == 0 && cells == as && base == b)) {
+            // busy waiting, retry loop
+            Thread.yield();
+            as = cells; b = base;
+        }
+
+        double sum = Double.longBitsToDouble(b);
         if (as != null) {
             int n = as.length;
             for (int i = 0; i < n; ++i) {
@@ -118,6 +147,41 @@ public void reset() {
         internalReset(0L);
     }
 
+    public void set(double x) {
+        // On concurrent `set` and `set`, it should be acceptable to lose one `set` measurement.
+        // On concurrent `set` and `add`, it should be acceptable to lose the `add` measurement.
+
+        // Correctness is ensured by different techniques:
+        //  - `set` waits on contention (blocking)
+        //  - `add` avoids contention (non-blocking)
+        //  - `sum` retries on conflicts (non-blocking)
+        // Performance characteristics by use cases:
+        //  - only `set` - `cells` is always `null` - no allocations
+        //  - only `add` - `cells` allocated on contention
+        //  - mixed `set` and `add` - `cells` allocated on contention, `cells` deallocated on `set`
+        for (;;) {
+            Cell[] as;
+            if ((as = cells) != null) { // have cells
+                if (busy == 0 && casBusy()) {
+                    try {
+                        if (cells == as) { // recheck under lock
+                            // update cells and base (not atomic)
+                            cells = null;
+                            base = Double.doubleToLongBits(x);
+                            break;
+                        }
+                    } finally {
+                        busy = 0;
+                    }
+                }
+            } else { // no cells
+                // update base (atomic)
+                base = Double.doubleToLongBits(x);
+                break;
+            }
+        }
+    }
+
     /**
      * Equivalent in effect to {@link #sum} followed by {@link
      * #reset}. This method may apply for example during quiescent

simpleclient/src/main/java/io/prometheus/client/Gauge.java

@@ -134,6 +134,7 @@ public void close() {
    * {@link SimpleCollector#remove} or {@link SimpleCollector#clear},
    */
   public static class Child {
+
     private final DoubleAdder value = new DoubleAdder();
 
     static TimeProvider timeProvider = new TimeProvider();
@@ -166,13 +167,7 @@ public void dec(double amt) {
      * Set the gauge to the given value.
      */
     public void set(double val) {
-      synchronized(this) {
-        value.reset();
-        // If get() were called here it'd see an invalid value, so use a lock.
-        // inc()/dec() don't need locks, as all the possible outcomes
-        // are still possible if set() were atomic so no new races are introduced.
-        value.add(val);
-      }
+      value.set(val);
     }
     /**
      * Set the gauge to the current unixtime.
@@ -234,9 +229,7 @@ public <E> E setToTime(Callable<E> timeable){
      * Get the value of the gauge.
      */
     public double get() {
-      synchronized(this) {
-        return value.sum();
-      }
+      return value.sum();
     }
   }