Fix concurrency issue with Exponential Histogram (#5749)

Synchronize writes to exponential histogram. An alternative would be to use explicit locks, but this made the code more complex and did not yield significant performance improvements over using a synchronized method. Also adds a concurrency test that reproduced the reported issue and verifies this fixes it.

Fixes gh-5740

Author: lenin-jaganathan

concurrency-tests/build.gradle

@@ -8,6 +8,7 @@ dependencies {
     implementation project(":micrometer-core")
 //    implementation("io.micrometer:micrometer-core:1.14.1")
     implementation project(":micrometer-registry-prometheus")
+    implementation project(":micrometer-registry-otlp")
 //    implementation("io.micrometer:micrometer-registry-prometheus:1.14.1")
     runtimeOnly(libs.logbackLatest)
 }

concurrency-tests/src/jcstress/java/io/micrometer/concurrencytests/Base2ExponentialHistogramConcurrencyTests.java

@@ -0,0 +1,87 @@
+/*
+ * Copyright 2024 VMware, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.micrometer.concurrencytests;
+
+import static org.openjdk.jcstress.annotations.Expect.*;
+import static org.openjdk.jcstress.annotations.Expect.ACCEPTABLE;
+import static org.openjdk.jcstress.annotations.Expect.FORBIDDEN;
+
+import org.openjdk.jcstress.annotations.*;
+import org.openjdk.jcstress.infra.results.*;
+import io.micrometer.registry.otlp.internal.Base2ExponentialHistogram;
+import io.micrometer.registry.otlp.internal.CumulativeBase2ExponentialHistogram;
+
+public class Base2ExponentialHistogramConcurrencyTests {
+
+    @JCStressTest
+    @Outcome(id = "null, 5, 2", expect = ACCEPTABLE, desc = "Read after all writes")
+    @Outcome(id = "null, 20, 0", expect = ACCEPTABLE, desc = "Read before write")
+    @Outcome(id = { "null, 20, 1" }, expect = ACCEPTABLE, desc = "Reading after single " + "write")
+    @Outcome(
+            id = { "class java.lang.ArrayIndexOutOfBoundsException, 20, 0",
+                    "class java.lang.ArrayIndexOutOfBoundsException, 20, 1" },
+            expect = FORBIDDEN, desc = "Exception in recording thread")
+    @Outcome(
+            id = "class java.lang.ArrayIndexOutOfBoundsException, class java.lang.ArrayIndexOutOfBoundsException, null",
+            expect = FORBIDDEN, desc = "Exception in both reading and writing threads")
+    @Outcome(id = "null, class java.lang.ArrayIndexOutOfBoundsException, null", expect = FORBIDDEN,
+            desc = "Exception in reading thread")
+    @Outcome(expect = UNKNOWN)
+    @State
+    public static class RescalingAndConcurrentReading {
+
+        Base2ExponentialHistogram exponentialHistogram = new CumulativeBase2ExponentialHistogram(20, 40, 0, null);
+
+        @Actor
+        public void actor1(LLL_Result r) {
+            try {
+                exponentialHistogram.recordDouble(2);
+            }
+            catch (Exception e) {
+                r.r1 = e.getClass();
+            }
+        }
+
+        @Actor
+        public void actor2(LLL_Result r) {
+            try {
+                exponentialHistogram.recordDouble(4);
+            }
+            catch (Exception e) {
+                r.r1 = e.getClass();
+            }
+        }
+
+        @Actor
+        public void actor3(LLL_Result r) {
+            try {
+                exponentialHistogram.takeSnapshot(2, 6, 4);
+                r.r2 = exponentialHistogram.getLatestExponentialHistogramSnapshot().scale();
+                r.r3 = exponentialHistogram.getLatestExponentialHistogramSnapshot()
+                    .positive()
+                    .bucketCounts()
+                    .stream()
+                    .mapToLong(Long::longValue)
+                    .sum();
+            }
+            catch (Exception e) {
+                r.r2 = e.getClass();
+            }
+        }
+
+    }
+
+}

implementations/micrometer-registry-otlp/src/main/java/io/micrometer/registry/otlp/internal/Base2ExponentialHistogram.java

@@ -167,15 +167,16 @@ public void recordDouble(double value) {
             zeroCount.increment();
             return;
         }
+        recordToHistogram(value);
+    }
 
+    private synchronized void recordToHistogram(final double value) {
         int index = base2IndexProvider.getIndexForValue(value);
         if (!circularCountHolder.increment(index, 1)) {
-            synchronized (this) {
-                int downScaleFactor = getDownScaleFactor(index);
-                downScale(downScaleFactor);
-                index = base2IndexProvider.getIndexForValue(value);
-                circularCountHolder.increment(index, 1);
-            }
+            int downScaleFactor = getDownScaleFactor(index);
+            downScale(downScaleFactor);
+            index = base2IndexProvider.getIndexForValue(value);
+            circularCountHolder.increment(index, 1);
         }
     }
 

implementations/micrometer-registry-otlp/src/main/java/io/micrometer/registry/otlp/internal/CircularCountHolder.java

@@ -15,7 +15,7 @@
  */
 package io.micrometer.registry.otlp.internal;
 
-import java.util.concurrent.atomic.AtomicLongArray;
+import java.util.Arrays;
 
 /**
  * The CircularCountHolder is inspired from <a href=
@@ -25,7 +25,7 @@
  */
 class CircularCountHolder {
 
-    private final AtomicLongArray counts;
+    private final long[] counts;
 
     private final int length;
 
@@ -37,7 +37,7 @@ class CircularCountHolder {
 
     CircularCountHolder(int size) {
         this.length = size;
-        this.counts = new AtomicLongArray(size);
+        this.counts = new long[size];
         this.baseIndex = Integer.MIN_VALUE;
         this.startIndex = Integer.MIN_VALUE;
         this.endIndex = Integer.MIN_VALUE;
@@ -52,7 +52,7 @@ int getEndIndex() {
     }
 
     long getValueAtIndex(int index) {
-        return counts.get(getRelativeIndex(index));
+        return counts[getRelativeIndex(index)];
     }
 
     boolean isEmpty() {
@@ -64,7 +64,7 @@ boolean increment(int index, long incrementBy) {
             this.baseIndex = index;
             this.startIndex = index;
             this.endIndex = index;
-            this.counts.addAndGet(0, incrementBy);
+            this.counts[0] = this.counts[0] + incrementBy;
             return true;
         }
 
@@ -81,7 +81,8 @@ else if (index < startIndex) {
             startIndex = index;
         }
 
-        counts.addAndGet(getRelativeIndex(index), incrementBy);
+        final int relativeIndex = getRelativeIndex(index);
+        counts[relativeIndex] = counts[relativeIndex] + incrementBy;
         return true;
     }
 
@@ -97,9 +98,7 @@ else if (result < 0) {
     }
 
     void reset() {
-        for (int i = 0; i < counts.length(); i++) {
-            counts.set(i, 0);
-        }
+        Arrays.fill(counts, 0);
         this.baseIndex = Integer.MIN_VALUE;
         this.endIndex = Integer.MIN_VALUE;
         this.startIndex = Integer.MIN_VALUE;