From 0e08d5109d18d0ee71215116e5fc0a6acb51bec5 Mon Sep 17 00:00:00 2001 From: Gregor Zeitlinger Date: Thu, 23 Jul 2026 08:00:47 +0000 Subject: [PATCH 1/9] perf: experiment with striped classic histogram accumulator Signed-off-by: Gregor Zeitlinger --- .../core/metrics/ClassicOnlyAccumulator.java | 120 ++++++++++++++++++ .../metrics/core/metrics/Histogram.java | 48 +++++-- .../metrics/core/metrics/HistogramTest.java | 112 ++++++++++++++++ 3 files changed, 272 insertions(+), 8 deletions(-) create mode 100644 prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulator.java diff --git a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulator.java b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulator.java new file mode 100644 index 000000000..e4f5b30ff --- /dev/null +++ b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulator.java @@ -0,0 +1,120 @@ +package io.prometheus.metrics.core.metrics; + +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.atomic.AtomicLong; + +/** + * Experimental accumulator for classic-only histogram data points. + * + *

Each recording thread owns a cell with two buffers. A snapshot advances the global epoch, + * waits only for observations that had already entered the previous epoch, and then drains the + * inactive buffers. Recording threads therefore never contend on a shared monitor. + * + *

Cells are retained for the lifetime of the data point so observations made by short-lived + * threads remain available to later snapshots. 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 ConcurrentLinkedQueue cells = new ConcurrentLinkedQueue<>(); + private final ThreadLocal threadCell = + new ThreadLocal() { + @Override + protected Cell initialValue() { + Cell cell = new Cell(bucketCount); + cells.add(cell); + return cell; + } + }; + + // 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) { + long observedEpoch = epoch.get(); + cell.writingEpoch = observedEpoch; + if (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 waiting on writingEpoch. + cell.writingEpoch = NOT_WRITING; + } + } + } + + @SuppressWarnings("ThreadPriorityCheck") + synchronized Snapshot snapshot() { + long inactiveEpoch = epoch.getAndIncrement(); + int inactiveBuffer = (int) (inactiveEpoch & 1); + + for (Cell cell : cells) { + while (cell.writingEpoch == inactiveEpoch) { + Thread.yield(); + } + 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; + } + + return new Snapshot(collectedBuckets.clone(), collectedCount, collectedSum); + } + + private static final class Cell { + private final CellBuffer[] buffers; + private volatile long writingEpoch = NOT_WRITING; + + 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; + } + } +} diff --git a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Histogram.java b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Histogram.java index c4bb1f5fe..5ead70345 100644 --- a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Histogram.java +++ b/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,24 @@ 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) { + return classicOnlyAccumulator.snapshot().sum; + } return sum.sum(); } @Override public long getCount() { + if (classicOnlyAccumulator != null) { + return classicOnlyAccumulator.snapshot().count; + } return count.sum(); } @@ -242,7 +251,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 +267,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 +279,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 +310,15 @@ private void doObserve(double value, boolean fromBuffer) { private HistogramSnapshot.HistogramDataPointSnapshot collect(Labels labels) { Exemplars exemplars = exemplarSampler != null ? exemplarSampler.collect() : Exemplars.EMPTY; + if (classicOnlyAccumulator != null) { + 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 +360,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 buckets) { boolean newBucketCreated = false; int bucketIndex; diff --git a/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/HistogramTest.java b/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/HistogramTest.java index cbfd5fade..127fd6a3e 100644 --- a/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/HistogramTest.java +++ b/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/HistogramTest.java @@ -41,6 +41,7 @@ import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.LongAdder; import java.util.stream.Collectors; import org.junit.jupiter.api.AfterEach; @@ -1598,6 +1599,117 @@ void testObserveMultithreaded() assertThat(executor.awaitTermination(5, TimeUnit.SECONDS)).isTrue(); } + @Test + void testClassicOnlyCollectWhileObserversAreRunning() throws Exception { + Histogram histogram = Histogram.builder().name("test").classicOnly().build(); + DistributionDataPoint dataPoint = histogram.labelValues(); + int observerCount = 8; + int observationsPerThread = 25_000; + ExecutorService executor = Executors.newFixedThreadPool(observerCount + 1); + CountDownLatch ready = new CountDownLatch(observerCount); + CountDownLatch start = new CountDownLatch(1); + AtomicBoolean observersFinished = new AtomicBoolean(); + List> observers = new ArrayList<>(); + + for (int i = 0; i < observerCount; i++) { + observers.add( + executor.submit( + () -> { + ready.countDown(); + start.await(); + for (int observation = 0; observation < observationsPerThread; observation++) { + dataPoint.observe(1.25); + } + return null; + })); + } + + Future collector = + executor.submit( + () -> { + long previousCount = 0; + start.await(); + while (!observersFinished.get()) { + HistogramSnapshot.HistogramDataPointSnapshot snapshot = + histogram.collect().getDataPoints().get(0); + assertClassicOnlySnapshotIsCoherent(snapshot, previousCount); + previousCount = snapshot.getCount(); + } + return null; + }); + + assertThat(ready.await(5, TimeUnit.SECONDS)).isTrue(); + start.countDown(); + for (Future observer : observers) { + observer.get(10, TimeUnit.SECONDS); + } + observersFinished.set(true); + collector.get(10, TimeUnit.SECONDS); + + HistogramSnapshot.HistogramDataPointSnapshot snapshot = + histogram.collect().getDataPoints().get(0); + assertClassicOnlySnapshotIsCoherent(snapshot, observerCount * observationsPerThread); + assertThat(dataPoint.getCount()).isEqualTo(observerCount * observationsPerThread); + assertThat(dataPoint.getSum()) + .isCloseTo(observerCount * observationsPerThread * 1.25, offset(0.000_001)); + + executor.shutdown(); + assertThat(executor.awaitTermination(5, TimeUnit.SECONDS)).isTrue(); + } + + @Test + void testClassicOnlyRetainsShortLivedThreadCellsAndClearCreatesFreshDataPoint() throws Exception { + Histogram histogram = + Histogram.builder().name("test").classicOnly().labelNames("status").build(); + DistributionDataPoint oldDataPoint = histogram.labelValues("200"); + int threadCount = 100; + ExecutorService executor = Executors.newFixedThreadPool(threadCount); + CountDownLatch start = new CountDownLatch(1); + List> observations = new ArrayList<>(); + for (int i = 0; i < threadCount; i++) { + observations.add( + executor.submit( + () -> { + start.await(); + oldDataPoint.observe(2.0); + return null; + })); + } + start.countDown(); + for (Future observation : observations) { + observation.get(5, TimeUnit.SECONDS); + } + executor.shutdown(); + assertThat(executor.awaitTermination(5, TimeUnit.SECONDS)).isTrue(); + + assertThat(oldDataPoint.getCount()).isEqualTo(threadCount); + assertThat(oldDataPoint.getSum()).isEqualTo(threadCount * 2.0); + assertThat(getBucket(histogram, 2.5, "status", "200").getCount()).isEqualTo(threadCount); + + histogram.clear(); + DistributionDataPoint newDataPoint = histogram.labelValues("200"); + assertThat(newDataPoint).isNotSameAs(oldDataPoint); + assertThat(newDataPoint.getCount()).isZero(); + assertThat(newDataPoint.getSum()).isZero(); + + newDataPoint.observe(3.0); + assertThat(newDataPoint.getCount()).isOne(); + assertThat(newDataPoint.getSum()).isEqualTo(3.0); + assertThat(oldDataPoint.getCount()).isEqualTo(threadCount); + assertThat(oldDataPoint.getSum()).isEqualTo(threadCount * 2.0); + } + + private static void assertClassicOnlySnapshotIsCoherent( + HistogramSnapshot.HistogramDataPointSnapshot snapshot, long minimumCount) { + assertThat(snapshot.getCount()).isGreaterThanOrEqualTo(minimumCount); + assertThat(snapshot.getSum()).isCloseTo(snapshot.getCount() * 1.25, offset(0.000_001)); + long bucketTotal = 0; + for (ClassicHistogramBucket bucket : snapshot.getClassicBuckets()) { + bucketTotal += bucket.getCount(); + } + assertThat(bucketTotal).isEqualTo(snapshot.getCount()); + } + @Test void testNativeResetDuration() { // Test that nativeResetDuration can be configured without error and the histogram From 6fe299560f57865c3ca33d399163a0639f84b47e Mon Sep 17 00:00:00 2001 From: Gregor Zeitlinger Date: Thu, 23 Jul 2026 08:04:40 +0000 Subject: [PATCH 2/9] docs: update core API diff Signed-off-by: Gregor Zeitlinger --- docs/apidiffs/current_vs_latest/prometheus-metrics-core.txt | 2 ++ 1 file changed, 2 insertions(+) diff --git a/docs/apidiffs/current_vs_latest/prometheus-metrics-core.txt b/docs/apidiffs/current_vs_latest/prometheus-metrics-core.txt index ffb4a1d52..136f7f6f1 100644 --- a/docs/apidiffs/current_vs_latest/prometheus-metrics-core.txt +++ b/docs/apidiffs/current_vs_latest/prometheus-metrics-core.txt @@ -1,4 +1,6 @@ Comparing source compatibility of prometheus-metrics-core-1.8.1-SNAPSHOT.jar against prometheus-metrics-core-1.8.0.jar *** MODIFIED CLASS: PUBLIC io.prometheus.metrics.core.exemplars.ExemplarSampler (not serializable) === CLASS FILE FORMAT VERSION: 52.0 <- 52.0 +*** MODIFIED CLASS: PUBLIC io.prometheus.metrics.core.metrics.Histogram$DataPoint (not serializable) + === CLASS FILE FORMAT VERSION: 52.0 <- 52.0 From 9a0c7c4aae6e1bb3995916677d34a59be9832c82 Mon Sep 17 00:00:00 2001 From: Gregor Zeitlinger Date: Tue, 18 Aug 2026 14:13:05 +0000 Subject: [PATCH 3/9] fix: harden classic histogram accumulator concurrency Signed-off-by: Gregor Zeitlinger --- .../benchmarks/HistogramBenchmark.java | 33 ++++++ .../core/metrics/ClassicOnlyAccumulator.java | 72 ++++++++++--- .../metrics/ClassicOnlyAccumulatorTest.java | 101 ++++++++++++++++++ 3 files changed, 194 insertions(+), 12 deletions(-) create mode 100644 prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorTest.java diff --git a/benchmarks/src/main/java/io/prometheus/metrics/benchmarks/HistogramBenchmark.java b/benchmarks/src/main/java/io/prometheus/metrics/benchmarks/HistogramBenchmark.java index 41b7097db..50cba01a2 100644 --- a/benchmarks/src/main/java/io/prometheus/metrics/benchmarks/HistogramBenchmark.java +++ b/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( diff --git a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulator.java b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulator.java index e4f5b30ff..b4b272701 100644 --- a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulator.java +++ b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulator.java @@ -1,6 +1,9 @@ package io.prometheus.metrics.core.metrics; -import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; /** @@ -10,25 +13,27 @@ * waits only for observations that had already entered the previous epoch, and then drains the * inactive buffers. Recording threads therefore never contend on a shared monitor. * - *

Cells are retained for the lifetime of the data point so observations made by short-lived - * threads remain available to later snapshots. A cell is static and does not reference its owning - * accumulator, so a thread-local value cannot retain a removed or cleared data point. + *

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; + // A stalled recorder must not make a scrape wait indefinitely. The next snapshot will retry + // the buffer after the recorder has left its epoch. + private static final long SNAPSHOT_WAIT_NANOS = TimeUnit.MILLISECONDS.toNanos(1); private final int bucketCount; private final AtomicLong epoch = new AtomicLong(); - private final ConcurrentLinkedQueue cells = new ConcurrentLinkedQueue<>(); + private final Set cells = ConcurrentHashMap.newKeySet(); private final ThreadLocal threadCell = new ThreadLocal() { @Override protected Cell initialValue() { - Cell cell = new Cell(bucketCount); - cells.add(cell); - return cell; + return new Cell(bucketCount); } }; @@ -45,9 +50,18 @@ protected Cell initialValue() { 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; - if (epoch.get() != 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; } @@ -64,14 +78,17 @@ void observe(int bucket, double value) { } } - @SuppressWarnings("ThreadPriorityCheck") + @SuppressWarnings({"ModifyCollectionInEnhancedForLoop", "ThreadPriorityCheck"}) synchronized Snapshot snapshot() { long inactiveEpoch = epoch.getAndIncrement(); int inactiveBuffer = (int) (inactiveEpoch & 1); + long waitDeadline = System.nanoTime() + SNAPSHOT_WAIT_NANOS; for (Cell cell : cells) { - while (cell.writingEpoch == inactiveEpoch) { - Thread.yield(); + if (!awaitInactiveBuffer(cell, inactiveBuffer, waitDeadline)) { + // 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++) { @@ -82,14 +99,45 @@ synchronized Snapshot snapshot() { 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); } + @SuppressWarnings("ThreadPriorityCheck") + private boolean awaitInactiveBuffer(Cell cell, int inactiveBuffer, long waitDeadline) { + while (true) { + 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 inactiveEpoch: draining while that writer is active would race + // with its plain bucket writes. + if (writingEpoch == NOT_WRITING || (writingEpoch & 1) != inactiveBuffer) { + return true; + } + if (System.nanoTime() >= waitDeadline) { + return false; + } + Thread.yield(); + } + } + + 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)}; diff --git a/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorTest.java b/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorTest.java new file mode 100644 index 000000000..7e9c02fc3 --- /dev/null +++ b/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorTest.java @@ -0,0 +1,101 @@ +package io.prometheus.metrics.core.metrics; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively; + +import java.lang.reflect.Field; +import java.time.Duration; +import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.Test; + +class ClassicOnlyAccumulatorTest { + + @Test + void stalledWriterDoesNotBlockSnapshotAndIsCollectedLater() throws Exception { + ClassicOnlyAccumulator accumulator = new ClassicOnlyAccumulator(2); + accumulator.observe(0, 1.0); + + Object cell = onlyCell(accumulator); + Field writingEpoch = cell.getClass().getDeclaredField("writingEpoch"); + writingEpoch.setAccessible(true); + writingEpoch.setLong(cell, 0); + + ClassicOnlyAccumulator.Snapshot skipped = + assertTimeoutPreemptively(Duration.ofMillis(500), accumulator::snapshot); + assertThat(skipped.count).isZero(); + + writingEpoch.setLong(cell, -1); + // The first post-release snapshot flips to the other buffer. The following one revisits the + // delayed writer's buffer and must retain its observation. + accumulator.snapshot(); + ClassicOnlyAccumulator.Snapshot collected = accumulator.snapshot(); + assertThat(collected.count).isEqualTo(1); + assertThat(collected.sum).isEqualTo(1.0); + } + + @Test + void emptyCellsAreReclaimedAndCanBeReused() throws Exception { + ClassicOnlyAccumulator accumulator = new ClassicOnlyAccumulator(1); + accumulator.observe(0, 1.0); + Set cells = cells(accumulator); + assertThat(cells).hasSize(1); + + accumulator.snapshot(); + assertThat(cells).isEmpty(); + + accumulator.observe(0, 2.0); + assertThat(cells).hasSize(1); + accumulator.snapshot(); + assertThat(cells).isEmpty(); + } + + @Test + void concurrentWritersAndSnapshotsPreserveJmmVisibility() throws Exception { + ClassicOnlyAccumulator accumulator = new ClassicOnlyAccumulator(3); + int writers = 8; + int observationsPerWriter = 10_000; + CountDownLatch start = new CountDownLatch(1); + ExecutorService executor = Executors.newFixedThreadPool(writers); + try { + for (int writer = 0; writer < writers; writer++) { + int bucket = writer % 3; + executor.submit( + () -> { + start.await(); + for (int i = 0; i < observationsPerWriter; i++) { + accumulator.observe(bucket, bucket + 1.0); + } + return null; + }); + } + start.countDown(); + executor.shutdown(); + while (!executor.awaitTermination(10, TimeUnit.MILLISECONDS)) { + accumulator.snapshot(); + } + } finally { + executor.shutdownNow(); + } + + ClassicOnlyAccumulator.Snapshot snapshot = accumulator.snapshot(); + snapshot = accumulator.snapshot(); + assertThat(snapshot.count).isEqualTo(writers * observationsPerWriter); + assertThat(snapshot.buckets).containsExactly(30_000, 30_000, 20_000); + assertThat(snapshot.sum).isEqualTo(150_000.0); + } + + private static Object onlyCell(ClassicOnlyAccumulator accumulator) throws Exception { + return cells(accumulator).iterator().next(); + } + + @SuppressWarnings("unchecked") + private static Set cells(ClassicOnlyAccumulator accumulator) throws Exception { + Field cells = ClassicOnlyAccumulator.class.getDeclaredField("cells"); + cells.setAccessible(true); + return (Set) cells.get(accumulator); + } +} From ebe3d8e821bc553c95b86fcb0bcd53bd822b9d41 Mon Sep 17 00:00:00 2001 From: Gregor Zeitlinger Date: Tue, 18 Aug 2026 14:25:36 +0000 Subject: [PATCH 4/9] fix: bound striped accumulator snapshot latency Signed-off-by: Gregor Zeitlinger --- .../core/metrics/ClassicOnlyAccumulator.java | 34 ++++------- .../metrics/core/metrics/Histogram.java | 6 ++ .../metrics/ClassicOnlyAccumulatorTest.java | 60 +++++++++++++++++++ 3 files changed, 78 insertions(+), 22 deletions(-) diff --git a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulator.java b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulator.java index b4b272701..788eb2f39 100644 --- a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulator.java +++ b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulator.java @@ -2,7 +2,6 @@ import java.util.Set; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; @@ -22,9 +21,6 @@ final class ClassicOnlyAccumulator { private static final long NOT_WRITING = -1; - // A stalled recorder must not make a scrape wait indefinitely. The next snapshot will retry - // the buffer after the recorder has left its epoch. - private static final long SNAPSHOT_WAIT_NANOS = TimeUnit.MILLISECONDS.toNanos(1); private final int bucketCount; private final AtomicLong epoch = new AtomicLong(); @@ -78,14 +74,17 @@ void observe(int bucket, double value) { } } - @SuppressWarnings({"ModifyCollectionInEnhancedForLoop", "ThreadPriorityCheck"}) + @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); - long waitDeadline = System.nanoTime() + SNAPSHOT_WAIT_NANOS; for (Cell cell : cells) { - if (!awaitInactiveBuffer(cell, inactiveBuffer, waitDeadline)) { + 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; @@ -113,21 +112,12 @@ && isEmpty(cell.buffers[1]) return new Snapshot(collectedBuckets.clone(), collectedCount, collectedSum); } - @SuppressWarnings("ThreadPriorityCheck") - private boolean awaitInactiveBuffer(Cell cell, int inactiveBuffer, long waitDeadline) { - while (true) { - 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 inactiveEpoch: draining while that writer is active would race - // with its plain bucket writes. - if (writingEpoch == NOT_WRITING || (writingEpoch & 1) != inactiveBuffer) { - return true; - } - if (System.nanoTime() >= waitDeadline) { - return false; - } - Thread.yield(); - } + 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) { diff --git a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Histogram.java b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Histogram.java index 5ead70345..5e62a5283 100644 --- a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Histogram.java +++ b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Histogram.java @@ -232,6 +232,8 @@ private DataPoint() { @Override public double getSum() { if (classicOnlyAccumulator != null) { + // A paused recorder may make this exact value temporarily stale. The next call retries + // its buffer rather than blocking the scrape indefinitely. return classicOnlyAccumulator.snapshot().sum; } return sum.sum(); @@ -240,6 +242,8 @@ public double getSum() { @Override public long getCount() { if (classicOnlyAccumulator != null) { + // A paused recorder may make this exact value temporarily stale. The next call retries + // its buffer rather than blocking the scrape indefinitely. return classicOnlyAccumulator.snapshot().count; } return count.sum(); @@ -311,6 +315,8 @@ 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; + // the following collection retries its buffer without an unbounded wait. ClassicOnlyAccumulator.Snapshot snapshot = classicOnlyAccumulator.snapshot(); return new HistogramSnapshot.HistogramDataPointSnapshot( ClassicHistogramBuckets.of(classicUpperBounds, snapshot.buckets), diff --git a/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorTest.java b/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorTest.java index 7e9c02fc3..c9bd5e483 100644 --- a/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorTest.java +++ b/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorTest.java @@ -37,6 +37,66 @@ void stalledWriterDoesNotBlockSnapshotAndIsCollectedLater() throws Exception { assertThat(collected.sum).isEqualTo(1.0); } + @Test + void stalledCellDoesNotPreventHealthyCellsFromBeingCollected() throws Exception { + ClassicOnlyAccumulator accumulator = new ClassicOnlyAccumulator(1); + accumulator.observe(0, 1.0); + for (int i = 0; i < 4; i++) { + Thread recorder = new Thread(() -> accumulator.observe(0, 1.0)); + recorder.start(); + recorder.join(); + } + + Object stalledCell = onlyCell(accumulator); + Field writingEpoch = stalledCell.getClass().getDeclaredField("writingEpoch"); + writingEpoch.setAccessible(true); + writingEpoch.setLong(stalledCell, 0); + + ClassicOnlyAccumulator.Snapshot first = accumulator.snapshot(); + // The four healthy cells are drained even though the first cell consumes its own wait budget. + assertThat(first.count).isEqualTo(4); + + writingEpoch.setLong(stalledCell, -1); + accumulator.snapshot(); + ClassicOnlyAccumulator.Snapshot finalSnapshot = accumulator.snapshot(); + assertThat(finalSnapshot.count).isEqualTo(5); + } + + @Test + void activeWriterCanResumeAfterAStaleSnapshot() throws Exception { + ClassicOnlyAccumulator accumulator = new ClassicOnlyAccumulator(1); + accumulator.observe(0, 1.0); + Object cell = onlyCell(accumulator); + Field writingEpoch = cell.getClass().getDeclaredField("writingEpoch"); + writingEpoch.setAccessible(true); + CountDownLatch entered = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + Thread writer = + new Thread( + () -> { + try { + writingEpoch.setLong(cell, 0); + entered.countDown(); + release.await(); + writingEpoch.setLong(cell, -1); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } catch (IllegalAccessException e) { + throw new AssertionError(e); + } + }); + writer.start(); + entered.await(); + + ClassicOnlyAccumulator.Snapshot stale = accumulator.snapshot(); + assertThat(stale.count).isZero(); + release.countDown(); + writer.join(); + accumulator.snapshot(); + ClassicOnlyAccumulator.Snapshot resumed = accumulator.snapshot(); + assertThat(resumed.count).isEqualTo(1); + } + @Test void emptyCellsAreReclaimedAndCanBeReused() throws Exception { ClassicOnlyAccumulator accumulator = new ClassicOnlyAccumulator(1); From 3d5852fb7e767930990c1b1e2f89284f23ad7f68 Mon Sep 17 00:00:00 2001 From: Gregor Zeitlinger Date: Tue, 18 Aug 2026 14:28:34 +0000 Subject: [PATCH 5/9] docs: clarify striped accumulator snapshot semantics Signed-off-by: Gregor Zeitlinger --- .../metrics/core/metrics/ClassicOnlyAccumulator.java | 8 ++++---- .../prometheus/metrics/core/metrics/Histogram.java | 12 ++++++------ 2 files changed, 10 insertions(+), 10 deletions(-) diff --git a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulator.java b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulator.java index 788eb2f39..6f460ca33 100644 --- a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulator.java +++ b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulator.java @@ -8,9 +8,9 @@ /** * Experimental accumulator for classic-only histogram data points. * - *

Each recording thread owns a cell with two buffers. A snapshot advances the global epoch, - * waits only for observations that had already entered the previous epoch, and then drains the - * inactive buffers. Recording threads therefore never contend on a shared monitor. + *

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. * *

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 @@ -68,7 +68,7 @@ void observe(int bucket, double value) { buffer.count++; return; } finally { - // Publishes all plain writes above to a snapshot waiting on writingEpoch. + // Publishes all plain writes above to a snapshot observing writingEpoch. cell.writingEpoch = NOT_WRITING; } } diff --git a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Histogram.java b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Histogram.java index 5e62a5283..c47ff6083 100644 --- a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Histogram.java +++ b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Histogram.java @@ -232,8 +232,8 @@ private DataPoint() { @Override public double getSum() { if (classicOnlyAccumulator != null) { - // A paused recorder may make this exact value temporarily stale. The next call retries - // its buffer rather than blocking the scrape indefinitely. + // 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(); @@ -242,8 +242,8 @@ public double getSum() { @Override public long getCount() { if (classicOnlyAccumulator != null) { - // A paused recorder may make this exact value temporarily stale. The next call retries - // its buffer rather than blocking the scrape indefinitely. + // 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(); @@ -315,8 +315,8 @@ 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; - // the following collection retries its buffer without an unbounded wait. + // 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), From c08aaed6ae0fcb4100f7fe58963f9fe774f0012e Mon Sep 17 00:00:00 2001 From: Gregor Zeitlinger Date: Tue, 18 Aug 2026 15:08:54 +0000 Subject: [PATCH 6/9] test: add jcstress coverage for classic accumulators Signed-off-by: Gregor Zeitlinger --- pom.xml | 1 + prometheus-metrics-jcstress/README.md | 18 ++++ prometheus-metrics-jcstress/pom.xml | 86 +++++++++++++++++++ .../ClassicOnlyAccumulatorNoLossTest.java | 34 ++++++++ ...ClassicOnlyAccumulatorPublicationTest.java | 37 ++++++++ ...ssicOnlyAccumulatorReRegistrationTest.java | 37 ++++++++ ...assicOnlyAccumulatorStalledWriterTest.java | 48 +++++++++++ .../ClassicOnlyAccumulatorStressSupport.java | 60 +++++++++++++ 8 files changed, 321 insertions(+) create mode 100644 prometheus-metrics-jcstress/README.md create mode 100644 prometheus-metrics-jcstress/pom.xml create mode 100644 prometheus-metrics-jcstress/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorNoLossTest.java create mode 100644 prometheus-metrics-jcstress/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorPublicationTest.java create mode 100644 prometheus-metrics-jcstress/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorReRegistrationTest.java create mode 100644 prometheus-metrics-jcstress/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorStalledWriterTest.java create mode 100644 prometheus-metrics-jcstress/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorStressSupport.java diff --git a/pom.xml b/pom.xml index 5cebbcb0b..ec065b245 100644 --- a/pom.xml +++ b/pom.xml @@ -42,6 +42,7 @@ prometheus-metrics-annotations prometheus-metrics-bom prometheus-metrics-core + prometheus-metrics-jcstress prometheus-metrics-config prometheus-metrics-model prometheus-metrics-tracer diff --git a/prometheus-metrics-jcstress/README.md b/prometheus-metrics-jcstress/README.md new file mode 100644 index 000000000..0a34d803b --- /dev/null +++ b/prometheus-metrics-jcstress/README.md @@ -0,0 +1,18 @@ +# JCStress tests + +This module contains the JCStress coverage for the concurrent metric +accumulators. It is intentionally separate from the regular unit-test +suite because JCStress uses an isolated, generated test harness. + +Build the harness and run the accumulator tests with: + +```bash +./mvnw -pl prometheus-metrics-jcstress -am package -DskipTests +java -jar prometheus-metrics-jcstress/target/jcstress.jar \ + -t ClassicOnlyAccumulator -iters 10 -f 1 +``` + +The tests cover epoch publication, delayed writers, cell reclamation and +re-registration, and eventual observation visibility. A snapshot may be +intentionally stale while a writer is paused; the tests accept that outcome +but require the observation to become visible after the writer is released. diff --git a/prometheus-metrics-jcstress/pom.xml b/prometheus-metrics-jcstress/pom.xml new file mode 100644 index 000000000..c06c6fad7 --- /dev/null +++ b/prometheus-metrics-jcstress/pom.xml @@ -0,0 +1,86 @@ + + + 4.0.0 + + + io.prometheus + client_java + 1.8.1-SNAPSHOT + + + prometheus-metrics-jcstress + jar + + Prometheus Metrics JCStress Tests + JCStress tests for concurrent Prometheus metric implementations + + + 0.16 + jcstress + + + + + io.prometheus + prometheus-metrics-core + ${project.version} + + + org.openjdk.jcstress + jcstress-core + ${jcstress.version} + + + + + + + org.apache.maven.plugins + maven-compiler-plugin + 3.15.0 + + + -Xlint:all,-serial,-processing,-options + -Werror + --should-stop=ifError=FLOW + -XDcompilePolicy=simple + -XDaddTypeAnnotationsToSymbol=true + -Xplugin:ErrorProne -Xep:AlmostJavadoc:OFF -Xep:MissingSummary:OFF -Xep:LongDoubleConversion:OFF -Xep:StringSplitter:OFF -Xep:ThreadPriorityCheck:OFF -XepExcludedPaths:(.*/generated/.*|.*/src/test/java/.*|.*/examples/.*|.*/integration-tests/.*|.*/target/generated-sources/.*) -XepOpt:NullAway:AnnotatedPackages=io.prometheus.metrics + + + + org.openjdk.jcstress + jcstress-core + ${jcstress.version} + + + + + + org.apache.maven.plugins + maven-shade-plugin + 3.6.2 + + + jcstress-uberjar + package + + shade + + + ${uberjar.name} + + + org.openjdk.jcstress.Main + + + META-INF/TestList + + + + + + + + + diff --git a/prometheus-metrics-jcstress/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorNoLossTest.java b/prometheus-metrics-jcstress/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorNoLossTest.java new file mode 100644 index 000000000..2170fd4e5 --- /dev/null +++ b/prometheus-metrics-jcstress/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorNoLossTest.java @@ -0,0 +1,34 @@ +package io.prometheus.metrics.core.metrics; + +import org.openjdk.jcstress.annotations.Actor; +import org.openjdk.jcstress.annotations.Arbiter; +import org.openjdk.jcstress.annotations.Expect; +import org.openjdk.jcstress.annotations.JCStressTest; +import org.openjdk.jcstress.annotations.Outcome; +import org.openjdk.jcstress.annotations.State; +import org.openjdk.jcstress.infra.results.JD_Result; + +@JCStressTest +@Outcome(id = "2, 3.0", expect = Expect.ACCEPTABLE, desc = "Concurrent observations are visible with coherent count and sum.") +@State +public class ClassicOnlyAccumulatorNoLossTest { + private final ClassicOnlyAccumulator accumulator = new ClassicOnlyAccumulator(1); + + @Actor + public void observeOne() { + accumulator.observe(0, 1.0); + } + + @Actor + public void observeTwo() { + accumulator.observe(0, 2.0); + } + + @Arbiter + public void collect(JD_Result result) { + accumulator.snapshot(); + ClassicOnlyAccumulator.Snapshot snapshot = accumulator.snapshot(); + result.r1 = snapshot.count; + result.r2 = snapshot.sum; + } +} diff --git a/prometheus-metrics-jcstress/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorPublicationTest.java b/prometheus-metrics-jcstress/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorPublicationTest.java new file mode 100644 index 000000000..27d99db0d --- /dev/null +++ b/prometheus-metrics-jcstress/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorPublicationTest.java @@ -0,0 +1,37 @@ +package io.prometheus.metrics.core.metrics; + +import org.openjdk.jcstress.annotations.Actor; +import org.openjdk.jcstress.annotations.Arbiter; +import org.openjdk.jcstress.annotations.Expect; +import org.openjdk.jcstress.annotations.JCStressTest; +import org.openjdk.jcstress.annotations.Outcome; +import org.openjdk.jcstress.annotations.State; +import org.openjdk.jcstress.infra.results.II_Result; + +@JCStressTest +@Outcome(id = "1, 1", expect = Expect.ACCEPTABLE, desc = "A completed observation is published to a subsequent snapshot.") +@State +public class ClassicOnlyAccumulatorPublicationTest { + private final ClassicOnlyAccumulator accumulator = new ClassicOnlyAccumulator(1); + private volatile boolean observed; + + @Actor + public void observe() { + accumulator.observe(0, 1.0); + observed = true; + } + + @Actor + public void snapshot(II_Result result) { + while (!observed) { + Thread.yield(); + } + result.r1 = accumulator.snapshot().count > 0 ? 1 : 0; + } + + @Arbiter + public void finalSnapshot(II_Result result) { + accumulator.snapshot(); + result.r2 = accumulator.snapshot().count > 0 ? 1 : 0; + } +} diff --git a/prometheus-metrics-jcstress/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorReRegistrationTest.java b/prometheus-metrics-jcstress/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorReRegistrationTest.java new file mode 100644 index 000000000..2f26db3e4 --- /dev/null +++ b/prometheus-metrics-jcstress/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorReRegistrationTest.java @@ -0,0 +1,37 @@ +package io.prometheus.metrics.core.metrics; + +import org.openjdk.jcstress.annotations.Actor; +import org.openjdk.jcstress.annotations.Arbiter; +import org.openjdk.jcstress.annotations.Expect; +import org.openjdk.jcstress.annotations.JCStressTest; +import org.openjdk.jcstress.annotations.Outcome; +import org.openjdk.jcstress.annotations.State; +import org.openjdk.jcstress.infra.results.J_Result; + +@JCStressTest +@Outcome(id = "2", expect = Expect.ACCEPTABLE, desc = "A reclaimed thread cell is re-registered without losing observations.") +@State +public class ClassicOnlyAccumulatorReRegistrationTest { + private final ClassicOnlyAccumulator accumulator = new ClassicOnlyAccumulator(1); + + @Actor + public void observeReclaimAndReuse() { + accumulator.observe(0, 1.0); + accumulator.snapshot(); + accumulator.observe(0, 2.0); + } + + @Actor + public void concurrentSnapshots() { + // Keep a second actor so the test still runs through JCStress's actor/arbiter protocol. The + // reclamation and re-registration happen on one recording thread; concurrent snapshots are + // covered by the stalled-writer and no-loss tests. + } + + @Arbiter + public void collect(J_Result result) { + accumulator.snapshot(); + accumulator.snapshot(); + result.r1 = accumulator.snapshot().count; + } +} diff --git a/prometheus-metrics-jcstress/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorStalledWriterTest.java b/prometheus-metrics-jcstress/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorStalledWriterTest.java new file mode 100644 index 000000000..8ace5e5b9 --- /dev/null +++ b/prometheus-metrics-jcstress/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorStalledWriterTest.java @@ -0,0 +1,48 @@ +package io.prometheus.metrics.core.metrics; + +import org.openjdk.jcstress.annotations.Actor; +import org.openjdk.jcstress.annotations.Arbiter; +import org.openjdk.jcstress.annotations.Expect; +import org.openjdk.jcstress.annotations.JCStressTest; +import org.openjdk.jcstress.annotations.Outcome; +import org.openjdk.jcstress.annotations.State; +import org.openjdk.jcstress.infra.results.II_Result; +import javax.annotation.Nullable; + +@JCStressTest +@Outcome(id = "0, 1", expect = Expect.ACCEPTABLE, desc = "A paused writer is skipped and collected after it resumes.") +@State +public class ClassicOnlyAccumulatorStalledWriterTest { + private final ClassicOnlyAccumulator accumulator = new ClassicOnlyAccumulator(1); + private volatile @Nullable Object stalledCell; + private volatile boolean cellReady; + + @Actor + public void pauseWriter() { + stalledCell = ClassicOnlyAccumulatorStressSupport.createStalledCell(accumulator); + cellReady = true; + } + + @Actor + public void snapshot(II_Result result) { + while (!cellReady) { + Thread.yield(); + } + result.r1 = accumulator.snapshot().count > 0 ? 1 : 0; + } + + @Arbiter + public void resumeAndCollect(II_Result result) { + Object cell = stalledCell; + if (cell == null) { + throw new AssertionError("Writer did not publish its cell"); + } + ClassicOnlyAccumulatorStressSupport.releaseStalledCell(cell); + // The observing actor may have advanced the epoch before creating the stalled cell. + // Four flips cover both buffer parities and leave the result independent of that race. + accumulator.snapshot(); + accumulator.snapshot(); + accumulator.snapshot(); + result.r2 = accumulator.snapshot().count > 0 ? 1 : 0; + } +} diff --git a/prometheus-metrics-jcstress/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorStressSupport.java b/prometheus-metrics-jcstress/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorStressSupport.java new file mode 100644 index 000000000..40e8b780b --- /dev/null +++ b/prometheus-metrics-jcstress/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorStressSupport.java @@ -0,0 +1,60 @@ +package io.prometheus.metrics.core.metrics; + +import java.lang.reflect.Field; +import java.lang.reflect.Method; +import java.util.Set; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; + +final class ClassicOnlyAccumulatorStressSupport { + private static final Field CELLS = field(ClassicOnlyAccumulator.class, "cells"); + private static final Field EPOCH = field(ClassicOnlyAccumulator.class, "epoch"); + private static final Field THREAD_CELL = field(ClassicOnlyAccumulator.class, "threadCell"); + private static final Method SET_ADD = method(Set.class, "add", Object.class); + + private ClassicOnlyAccumulatorStressSupport() {} + + static Object createStalledCell(ClassicOnlyAccumulator accumulator) { + try { + Object cell = ((ThreadLocal) THREAD_CELL.get(accumulator)).get(); + SET_ADD.invoke(CELLS.get(accumulator), cell); + Field registered = field(cell.getClass(), "registered"); + ((AtomicBoolean) registered.get(cell)).set(true); + long epoch = ((AtomicLong) EPOCH.get(accumulator)).get(); + field(cell.getClass(), "writingEpoch").setLong(cell, epoch); + Object buffer = ((Object[]) field(cell.getClass(), "buffers").get(cell))[(int) (epoch & 1)]; + ((long[]) field(buffer.getClass(), "buckets").get(buffer))[0] = 1; + field(buffer.getClass(), "count").setLong(buffer, 1); + field(buffer.getClass(), "sum").setDouble(buffer, 1.0); + return cell; + } catch (ReflectiveOperationException e) { + throw new LinkageError("Unable to model a paused writer", e); + } + } + + static void releaseStalledCell(Object cell) { + try { + field(cell.getClass(), "writingEpoch").setLong(cell, -1); + } catch (ReflectiveOperationException e) { + throw new LinkageError("Unable to resume a paused writer", e); + } + } + + private static Field field(Class type, String name) { + try { + Field field = type.getDeclaredField(name); + field.setAccessible(true); + return field; + } catch (ReflectiveOperationException e) { + throw new ExceptionInInitializerError(e); + } + } + + private static Method method(Class type, String name, Class... parameterTypes) { + try { + return type.getMethod(name, parameterTypes); + } catch (ReflectiveOperationException e) { + throw new ExceptionInInitializerError(e); + } + } +} From 37792360e56453df50d02598bdfbda33fbc0d63a Mon Sep 17 00:00:00 2001 From: Gregor Zeitlinger Date: Tue, 18 Aug 2026 15:12:27 +0000 Subject: [PATCH 7/9] style: satisfy jcstress module lint Signed-off-by: Gregor Zeitlinger --- prometheus-metrics-jcstress/pom.xml | 11 ++++++++++- .../metrics/ClassicOnlyAccumulatorNoLossTest.java | 5 ++++- .../ClassicOnlyAccumulatorPublicationTest.java | 5 ++++- .../ClassicOnlyAccumulatorReRegistrationTest.java | 5 ++++- .../ClassicOnlyAccumulatorStalledWriterTest.java | 7 +++++-- 5 files changed, 27 insertions(+), 6 deletions(-) diff --git a/prometheus-metrics-jcstress/pom.xml b/prometheus-metrics-jcstress/pom.xml index c06c6fad7..87f4d96f5 100644 --- a/prometheus-metrics-jcstress/pom.xml +++ b/prometheus-metrics-jcstress/pom.xml @@ -45,7 +45,16 @@ --should-stop=ifError=FLOW -XDcompilePolicy=simple -XDaddTypeAnnotationsToSymbol=true - -Xplugin:ErrorProne -Xep:AlmostJavadoc:OFF -Xep:MissingSummary:OFF -Xep:LongDoubleConversion:OFF -Xep:StringSplitter:OFF -Xep:ThreadPriorityCheck:OFF -XepExcludedPaths:(.*/generated/.*|.*/src/test/java/.*|.*/examples/.*|.*/integration-tests/.*|.*/target/generated-sources/.*) -XepOpt:NullAway:AnnotatedPackages=io.prometheus.metrics + + -Xplugin:ErrorProne + -Xep:AlmostJavadoc:OFF + -Xep:MissingSummary:OFF + -Xep:LongDoubleConversion:OFF + -Xep:StringSplitter:OFF + -Xep:ThreadPriorityCheck:OFF + -XepExcludedPaths:(.*/generated/.*|.*/src/test/java/.*|.*/examples/.*|.*/integration-tests/.*|.*/target/generated-sources/.*) + -XepOpt:NullAway:AnnotatedPackages=io.prometheus.metrics + diff --git a/prometheus-metrics-jcstress/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorNoLossTest.java b/prometheus-metrics-jcstress/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorNoLossTest.java index 2170fd4e5..9a5796c02 100644 --- a/prometheus-metrics-jcstress/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorNoLossTest.java +++ b/prometheus-metrics-jcstress/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorNoLossTest.java @@ -9,7 +9,10 @@ import org.openjdk.jcstress.infra.results.JD_Result; @JCStressTest -@Outcome(id = "2, 3.0", expect = Expect.ACCEPTABLE, desc = "Concurrent observations are visible with coherent count and sum.") +@Outcome( + id = "2, 3.0", + expect = Expect.ACCEPTABLE, + desc = "Concurrent observations are visible with coherent count and sum.") @State public class ClassicOnlyAccumulatorNoLossTest { private final ClassicOnlyAccumulator accumulator = new ClassicOnlyAccumulator(1); diff --git a/prometheus-metrics-jcstress/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorPublicationTest.java b/prometheus-metrics-jcstress/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorPublicationTest.java index 27d99db0d..ea3145106 100644 --- a/prometheus-metrics-jcstress/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorPublicationTest.java +++ b/prometheus-metrics-jcstress/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorPublicationTest.java @@ -9,7 +9,10 @@ import org.openjdk.jcstress.infra.results.II_Result; @JCStressTest -@Outcome(id = "1, 1", expect = Expect.ACCEPTABLE, desc = "A completed observation is published to a subsequent snapshot.") +@Outcome( + id = "1, 1", + expect = Expect.ACCEPTABLE, + desc = "A completed observation is published to a subsequent snapshot.") @State public class ClassicOnlyAccumulatorPublicationTest { private final ClassicOnlyAccumulator accumulator = new ClassicOnlyAccumulator(1); diff --git a/prometheus-metrics-jcstress/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorReRegistrationTest.java b/prometheus-metrics-jcstress/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorReRegistrationTest.java index 2f26db3e4..ab54d7d4e 100644 --- a/prometheus-metrics-jcstress/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorReRegistrationTest.java +++ b/prometheus-metrics-jcstress/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorReRegistrationTest.java @@ -9,7 +9,10 @@ import org.openjdk.jcstress.infra.results.J_Result; @JCStressTest -@Outcome(id = "2", expect = Expect.ACCEPTABLE, desc = "A reclaimed thread cell is re-registered without losing observations.") +@Outcome( + id = "2", + expect = Expect.ACCEPTABLE, + desc = "A reclaimed thread cell is re-registered without losing observations.") @State public class ClassicOnlyAccumulatorReRegistrationTest { private final ClassicOnlyAccumulator accumulator = new ClassicOnlyAccumulator(1); diff --git a/prometheus-metrics-jcstress/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorStalledWriterTest.java b/prometheus-metrics-jcstress/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorStalledWriterTest.java index 8ace5e5b9..e7ac86e52 100644 --- a/prometheus-metrics-jcstress/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorStalledWriterTest.java +++ b/prometheus-metrics-jcstress/src/main/java/io/prometheus/metrics/core/metrics/ClassicOnlyAccumulatorStalledWriterTest.java @@ -1,5 +1,6 @@ package io.prometheus.metrics.core.metrics; +import javax.annotation.Nullable; import org.openjdk.jcstress.annotations.Actor; import org.openjdk.jcstress.annotations.Arbiter; import org.openjdk.jcstress.annotations.Expect; @@ -7,10 +8,12 @@ import org.openjdk.jcstress.annotations.Outcome; import org.openjdk.jcstress.annotations.State; import org.openjdk.jcstress.infra.results.II_Result; -import javax.annotation.Nullable; @JCStressTest -@Outcome(id = "0, 1", expect = Expect.ACCEPTABLE, desc = "A paused writer is skipped and collected after it resumes.") +@Outcome( + id = "0, 1", + expect = Expect.ACCEPTABLE, + desc = "A paused writer is skipped and collected after it resumes.") @State public class ClassicOnlyAccumulatorStalledWriterTest { private final ClassicOnlyAccumulator accumulator = new ClassicOnlyAccumulator(1); From b2f7e948fe088f14d1798ca5e61a35af8537225c Mon Sep 17 00:00:00 2001 From: Gregor Zeitlinger Date: Tue, 18 Aug 2026 15:15:01 +0000 Subject: [PATCH 8/9] build: register jcstress module in bom Signed-off-by: Gregor Zeitlinger --- prometheus-metrics-bom/pom.xml | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/prometheus-metrics-bom/pom.xml b/prometheus-metrics-bom/pom.xml index c2c9935f7..ff04d1342 100644 --- a/prometheus-metrics-bom/pom.xml +++ b/prometheus-metrics-bom/pom.xml @@ -114,6 +114,11 @@ prometheus-metrics-instrumentation-jvm ${project.version} + + io.prometheus + prometheus-metrics-jcstress + ${project.version} + io.prometheus prometheus-metrics-model From e3e9831f89753bb4666704c0df3e427e7034805c Mon Sep 17 00:00:00 2001 From: Gregor Zeitlinger Date: Tue, 18 Aug 2026 15:23:52 +0000 Subject: [PATCH 9/9] fix: compile jcstress module on older jdks Signed-off-by: Gregor Zeitlinger --- prometheus-metrics-jcstress/pom.xml | 10 ---------- 1 file changed, 10 deletions(-) diff --git a/prometheus-metrics-jcstress/pom.xml b/prometheus-metrics-jcstress/pom.xml index 87f4d96f5..c0b73d5b9 100644 --- a/prometheus-metrics-jcstress/pom.xml +++ b/prometheus-metrics-jcstress/pom.xml @@ -45,16 +45,6 @@ --should-stop=ifError=FLOW -XDcompilePolicy=simple -XDaddTypeAnnotationsToSymbol=true - - -Xplugin:ErrorProne - -Xep:AlmostJavadoc:OFF - -Xep:MissingSummary:OFF - -Xep:LongDoubleConversion:OFF - -Xep:StringSplitter:OFF - -Xep:ThreadPriorityCheck:OFF - -XepExcludedPaths:(.*/generated/.*|.*/src/test/java/.*|.*/examples/.*|.*/integration-tests/.*|.*/target/generated-sources/.*) - -XepOpt:NullAway:AnnotatedPackages=io.prometheus.metrics -