perf: experiment with striped classic histogram accumulator

benchmarks/src/main/java/io/prometheus/metrics/benchmarks/HistogramBenchmark.java

@@ -10,9 +10,12 @@
 import io.opentelemetry.sdk.resources.Resource;
 import io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader;
 import io.prometheus.metrics.core.metrics.Histogram;
+import io.prometheus.metrics.model.snapshots.MetricSnapshot;
 import java.util.Arrays;
 import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.Level;
 import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
 import org.openjdk.jmh.annotations.State;
 import org.openjdk.jmh.annotations.Threads;
 
@@ -57,6 +60,24 @@ public PrometheusClassicHistogramPerThread() {
     }
   }
 
+  @State(Scope.Benchmark)
+  public static class PrometheusClassicHistogramAfterThreadChurn {
+
+    final Histogram noLabels = Histogram.builder().name("test").help("help").classicOnly().build();
+
+    @Setup(Level.Invocation)
+    public void createShortLivedRecorders() throws InterruptedException {
+      Thread[] recorders = new Thread[1_000];
+      for (int i = 0; i < 1_000; i++) {
+        recorders[i] = new Thread(() -> noLabels.observe(1.0));
+        recorders[i].start();
+      }
+      for (Thread recorder : recorders) {
+        recorder.join();
+      }
+    }
+  }
+
   @State(Scope.Benchmark)
   public static class PrometheusNativeHistogram {
 
@@ -173,6 +194,18 @@ public Histogram prometheusClassicPerThread(
     return histogram.noLabels;
   }
 
+  @Benchmark
+  public long prometheusClassicGetCountAfterThreadChurn(
+      PrometheusClassicHistogramAfterThreadChurn histogram) {
+    return histogram.noLabels.getCount();
+  }
+
+  @Benchmark
+  public MetricSnapshot prometheusClassicCollectAfterThreadChurn(
+      PrometheusClassicHistogramAfterThreadChurn histogram) {
+    return histogram.noLabels.collect();
+  }
+
   @Benchmark
   @Threads(4)
   public Histogram prometheusNative(

pom.xml

@@ -42,6 +42,7 @@
     <module>prometheus-metrics-annotations</module>
     <module>prometheus-metrics-bom</module>
     <module>prometheus-metrics-core</module>
+    <module>prometheus-metrics-jcstress</module>
     <module>prometheus-metrics-config</module>
     <module>prometheus-metrics-model</module>
     <module>prometheus-metrics-tracer</module>

prometheus-metrics-bom/pom.xml

@@ -114,6 +114,11 @@
         <artifactId>prometheus-metrics-instrumentation-jvm</artifactId>
         <version>${project.version}</version>
       </dependency>
+      <dependency>
+        <groupId>io.prometheus</groupId>
+        <artifactId>prometheus-metrics-jcstress</artifactId>
+        <version>${project.version}</version>
+      </dependency>
       <dependency>
         <groupId>io.prometheus</groupId>
         <artifactId>prometheus-metrics-model</artifactId>

prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulator.java

@@ -0,0 +1,158 @@
+package io.prometheus.metrics.core.metrics;
+
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Experimental accumulator for classic-only histogram data points.
+ *
+ * <p>Each recording thread owns a cell with two buffers. A snapshot advances the global epoch and
+ * drains inactive buffers that are not being written. It does not wait for a paused recorder;
+ * recording threads therefore never contend on a shared monitor or stall a scrape.
+ *
+ * <p>Cells remain registered until both buffers have been collected, after which they can be
+ * reclaimed and re-registered if their recording thread is reused. A cell is static and does not
+ * reference its owning accumulator, so a thread-local value cannot retain a removed or cleared data
+ * point.
+ */
+@SuppressWarnings("ThreadLocalUsage")
+final class ClassicOnlyAccumulator {
+
+  private static final long NOT_WRITING = -1;
+
+  private final int bucketCount;
+  private final AtomicLong epoch = new AtomicLong();
+  private final Set<Cell> cells = ConcurrentHashMap.newKeySet();
+  private final ThreadLocal<Cell> threadCell =
+      new ThreadLocal<Cell>() {
+        @Override
+        protected Cell initialValue() {
+          return new Cell(bucketCount);
+        }
+      };
+
+  // Accessed only while holding this accumulator's monitor.
+  private final long[] collectedBuckets;
+  private long collectedCount;
+  private double collectedSum;
+
+  ClassicOnlyAccumulator(int bucketCount) {
+    this.bucketCount = bucketCount;
+    this.collectedBuckets = new long[bucketCount];
+  }
+
+  void observe(int bucket, double value) {
+    Cell cell = threadCell.get();
+    while (true) {
+      // Cells are removed once both buffers are empty. A thread-local may outlive that removal, so
+      // re-register it before every recording attempt.
+      if (!cell.registered.get() || !cells.contains(cell)) {
+        if (cell.registered.compareAndSet(false, true) || !cells.contains(cell)) {
+          cells.add(cell);
+        }
+      }
+      long observedEpoch = epoch.get();
+      cell.writingEpoch = observedEpoch;
+      // The registration check closes the race with snapshot's empty-cell reclamation. If a
+      // snapshot removed this cell after the first check, do not write into an unregistered cell.
+      if (!cell.registered.get() || epoch.get() != observedEpoch) {
+        cell.writingEpoch = NOT_WRITING;
+        continue;
+      }
+      try {
+        CellBuffer buffer = cell.buffers[(int) (observedEpoch & 1)];
+        buffer.buckets[bucket]++;
+        buffer.sum += value;
+        buffer.count++;
+        return;
+      } finally {
+        // Publishes all plain writes above to a snapshot observing writingEpoch.
+        cell.writingEpoch = NOT_WRITING;
+      }
+    }
+  }
+
+  @SuppressWarnings("ModifyCollectionInEnhancedForLoop")
+  synchronized Snapshot snapshot() {
+    // A snapshot is intentionally allowed to be stale for a cell whose recorder is paused. Do not
+    // wait here: this keeps collect(), getCount(), and getSum() bounded by the registered-cell and
+    // bucket counts, independent of writer stalls, while a subsequent snapshot includes the
+    // delayed observation after the recorder publishes NOT_WRITING.
+    long inactiveEpoch = epoch.getAndIncrement();
+    int inactiveBuffer = (int) (inactiveEpoch & 1);
+
+    for (Cell cell : cells) {
+      if (!canDrainInactiveBuffer(cell, inactiveBuffer)) {
+        // The writer may be paused indefinitely. Leave this buffer untouched; a later snapshot
+        // will collect it after the writer has published NOT_WRITING.
+        continue;
+      }
+      CellBuffer buffer = cell.buffers[inactiveBuffer];
+      for (int i = 0; i < bucketCount; i++) {
+        collectedBuckets[i] += buffer.buckets[i];
+        buffer.buckets[i] = 0;
+      }
+      collectedCount += buffer.count;
+      collectedSum += buffer.sum;
+      buffer.count = 0;
+      buffer.sum = 0;
+
+      // Reclaim cells from short-lived recording threads once their observations have been
+      // collected. The registration check in observe makes this safe if the thread is reused.
+      if (cell.writingEpoch == NOT_WRITING
+          && isEmpty(cell.buffers[0])
+          && isEmpty(cell.buffers[1])
+          && cell.registered.compareAndSet(true, false)) {
+        cells.remove(cell);
+      }
+    }
+
+    return new Snapshot(collectedBuckets.clone(), collectedCount, collectedSum);
+  }
+
+  private static boolean canDrainInactiveBuffer(Cell cell, int inactiveBuffer) {
+    long writingEpoch = cell.writingEpoch;
+    // An old writer can still be in the same parity after a pair of epoch flips. It is not enough
+    // to compare with the current epoch: draining while that writer is active would race with its
+    // plain bucket writes.
+    return writingEpoch == NOT_WRITING || (writingEpoch & 1) != inactiveBuffer;
+  }
+
+  private static boolean isEmpty(CellBuffer buffer) {
+    return buffer.count == 0;
+  }
+
+  private static final class Cell {
+    private final CellBuffer[] buffers;
+    private volatile long writingEpoch = NOT_WRITING;
+    private final AtomicBoolean registered = new AtomicBoolean();
+
+    private Cell(int bucketCount) {
+      buffers = new CellBuffer[] {new CellBuffer(bucketCount), new CellBuffer(bucketCount)};
+    }
+  }
+
+  private static final class CellBuffer {
+    private final long[] buckets;
+    private long count;
+    private double sum;
+
+    private CellBuffer(int bucketCount) {
+      buckets = new long[bucketCount];
+    }
+  }
+
+  static final class Snapshot {
+    final long[] buckets;
+    final long count;
+    final double sum;
+
+    private Snapshot(long[] buckets, long count, double sum) {
+      this.buckets = buckets;
+      this.count = count;
+      this.sum = sum;
+    }
+  }
+}

prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Histogram.java

@@ -205,6 +205,7 @@ public class DataPoint implements DistributionDataPoint {
     private final LongAdder nativeZeroCount = new LongAdder();
     private final LongAdder count = new LongAdder();
     private final DoubleAdder sum = new DoubleAdder();
+    @Nullable private final ClassicOnlyAccumulator classicOnlyAccumulator;
     private volatile int nativeSchema =
         nativeInitialSchema; // integer in [-4, 8] or CLASSIC_HISTOGRAM
     private volatile double nativeZeroThreshold = Histogram.this.nativeMinZeroThreshold;
@@ -223,16 +224,28 @@ private DataPoint() {
       for (int i = 0; i < classicUpperBounds.length; i++) {
         classicBuckets[i] = new LongAdder();
       }
+      classicOnlyAccumulator =
+          isClassicOnly() ? new ClassicOnlyAccumulator(classicUpperBounds.length) : null;
       maybeScheduleNextReset();
     }
 
     @Override
     public double getSum() {
+      if (classicOnlyAccumulator != null) {
+        // A paused recorder may make this exact value temporarily stale. A later snapshot, after
+        // the intervening buffer rotation, retries its buffer rather than blocking the scrape.
+        return classicOnlyAccumulator.snapshot().sum;
+      }
       return sum.sum();
     }
 
     @Override
     public long getCount() {
+      if (classicOnlyAccumulator != null) {
+        // A paused recorder may make this exact value temporarily stale. A later snapshot, after
+        // the intervening buffer rotation, retries its buffer rather than blocking the scrape.
+        return classicOnlyAccumulator.snapshot().count;
+      }
       return count.sum();
     }
 
@@ -242,7 +255,9 @@ public void observe(double value) {
         // See https://github.com/prometheus/client_golang/issues/1275 on ignoring NaN observations.
         return;
       }
-      if (!buffer.append(value)) {
+      if (classicOnlyAccumulator != null) {
+        classicOnlyAccumulator.observe(findClassicBucket(value), value);
+      } else if (!buffer.append(value)) {
         doObserve(value, false);
       }
       if (exemplarSampler != null) {
@@ -256,7 +271,9 @@ public void observeWithExemplar(double value, Labels labels) {
         // See https://github.com/prometheus/client_golang/issues/1275 on ignoring NaN observations.
         return;
       }
-      if (!buffer.append(value)) {
+      if (classicOnlyAccumulator != null) {
+        classicOnlyAccumulator.observe(findClassicBucket(value), value);
+      } else if (!buffer.append(value)) {
         doObserve(value, false);
       }
       if (exemplarSampler != null) {
@@ -266,12 +283,8 @@ public void observeWithExemplar(double value, Labels labels) {
 
     private void doObserve(double value, boolean fromBuffer) {
       // classicUpperBounds is an empty array if this is a native histogram only.
-      for (int i = 0; i < classicUpperBounds.length; ++i) {
-        // The last bucket is +Inf, so we always increment.
-        if (value <= classicUpperBounds[i]) {
-          classicBuckets[i].add(1);
-          break;
-        }
+      if (classicUpperBounds.length > 0) {
+        classicBuckets[findClassicBucket(value)].add(1);
       }
       boolean nativeBucketCreated = false;
       if (Histogram.this.nativeInitialSchema != CLASSIC_HISTOGRAM) {
@@ -301,6 +314,17 @@ private void doObserve(double value, boolean fromBuffer) {
 
     private HistogramSnapshot.HistogramDataPointSnapshot collect(Labels labels) {
       Exemplars exemplars = exemplarSampler != null ? exemplarSampler.collect() : Exemplars.EMPTY;
+      if (classicOnlyAccumulator != null) {
+        // collect() is intentionally allowed to return a stale snapshot for a paused recorder; a
+        // later collection, after the intervening buffer rotation, retries its buffer.
+        ClassicOnlyAccumulator.Snapshot snapshot = classicOnlyAccumulator.snapshot();
+        return new HistogramSnapshot.HistogramDataPointSnapshot(
+            ClassicHistogramBuckets.of(classicUpperBounds, snapshot.buckets),
+            snapshot.sum,
+            labels,
+            exemplars,
+            createdTimeMillis);
+      }
       return buffer.run(
           expectedCount -> count.sum() == expectedCount,
           () -> {
@@ -342,6 +366,20 @@ private HistogramSnapshot.HistogramDataPointSnapshot collect(Labels labels) {
           v -> doObserve(v, true));
     }
 
+    private boolean isClassicOnly() {
+      return Histogram.this.nativeInitialSchema == CLASSIC_HISTOGRAM;
+    }
+
+    private int findClassicBucket(double value) {
+      for (int i = 0; i < classicUpperBounds.length; ++i) {
+        // The last bucket is +Inf, so we always return from this loop.
+        if (value <= classicUpperBounds[i]) {
+          return i;
+        }
+      }
+      throw new IllegalStateException("Classic histogram is missing the +Inf bucket.");
+    }
+
     private boolean addToNativeBucket(double value, ConcurrentHashMap<Integer, LongAdder> buckets) {
       boolean newBucketCreated = false;
       int bucketIndex;