From 05bf7e4c9073350ef8738814f873809f69a6f2a5 Mon Sep 17 00:00:00 2001 From: Gregor Zeitlinger Date: Thu, 23 Jul 2026 10:08:01 +0000 Subject: [PATCH 1/5] fix: prevent buffer stripe index overflow Signed-off-by: Gregor Zeitlinger --- .../prometheus/metrics/core/metrics/Buffer.java | 6 +++++- .../metrics/core/metrics/BufferTest.java | 15 +++++++++++++++ 2 files changed, 20 insertions(+), 1 deletion(-) create mode 100644 prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/BufferTest.java diff --git a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Buffer.java b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Buffer.java index 1c47f867c..c2017995e 100644 --- a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Buffer.java +++ b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Buffer.java @@ -43,7 +43,7 @@ class Buffer { } boolean append(double value) { - int index = Math.abs((int) Thread.currentThread().getId()) % stripedObservationCounts.length; + int index = stripeIndex(Thread.currentThread().getId(), stripedObservationCounts.length); AtomicLong observationCountForThread = stripedObservationCounts[index]; long count = observationCountForThread.incrementAndGet(); if ((count & bufferActiveBit) == 0) { @@ -54,6 +54,10 @@ boolean append(double value) { } } + static int stripeIndex(long threadId, int stripeCount) { + return (int) Math.floorMod(threadId, stripeCount); + } + private void doAppend(double amount) { appendLock.lock(); try { diff --git a/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/BufferTest.java b/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/BufferTest.java new file mode 100644 index 000000000..dccd3b4eb --- /dev/null +++ b/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/BufferTest.java @@ -0,0 +1,15 @@ +package io.prometheus.metrics.core.metrics; + +import static org.assertj.core.api.Assertions.assertThat; + +import org.junit.jupiter.api.Test; + +class BufferTest { + + @Test + void stripeIndexDoesNotOverflowWhenThreadIdNarrowsToIntegerMinValue() { + assertThat(Buffer.stripeIndex(2_147_483_648L, 3)).isEqualTo(2); + assertThat(Buffer.stripeIndex(2_147_483_648L, 6)).isEqualTo(2); + assertThat(Buffer.stripeIndex(2_147_483_648L, 12)).isEqualTo(8); + } +} From 976d485d748c0775e384cdc9c46c4e4f1f0adce9 Mon Sep 17 00:00:00 2001 From: Gregor Zeitlinger Date: Thu, 23 Jul 2026 10:14:50 +0000 Subject: [PATCH 2/5] fix: bound observation buffering during collection Signed-off-by: Gregor Zeitlinger --- .../prometheus-metrics-core.txt | 2 + .../metrics/core/metrics/Buffer.java | 249 ++++++++++++------ .../metrics/core/metrics/Histogram.java | 31 ++- .../metrics/core/metrics/Summary.java | 14 +- .../metrics/core/metrics/BufferTest.java | 177 +++++++++++++ 5 files changed, 380 insertions(+), 93 deletions(-) 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 diff --git a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Buffer.java b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Buffer.java index c2017995e..cd29214ec 100644 --- a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Buffer.java +++ b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Buffer.java @@ -2,55 +2,122 @@ import io.prometheus.metrics.model.snapshots.DataPointSnapshot; import java.util.Arrays; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.ReentrantLock; import java.util.function.Consumer; import java.util.function.Function; import java.util.function.Supplier; +import javax.annotation.Nullable; /** - * Metrics support concurrent write and scrape operations. + * Coordinates concurrent metric observations with collection. * - *

This is implemented by switching to a Buffer when the scrape starts, and applying the values - * from the buffer after the scrape ends. + *

Collection activates a generation. Observations that start after activation are appended to + * that generation while the collector waits for observations from the previous phase to finish. The + * collector then creates a snapshot, deactivates the generation, and replays its buffered + * observations into the live metric state. */ class Buffer { - private static final long bufferActiveBit = 1L << 63; + private static final long DEFAULT_MAX_SPIN_WAIT_NANOS = TimeUnit.SECONDS.toNanos(1); + private static final int DEFAULT_MAX_BUFFER_SIZE = 1_000_000; + private static final int INITIAL_BUFFER_SIZE = 128; + + /** Observations buffered during one collection cycle. */ + private static final class Generation { + private double[] values = new double[0]; + private int size; + private boolean active = true; + } + // Tracking observation counts requires an AtomicLong for coordination between recording and // collecting. AtomicLong does much worse under contention than the LongAdder instances used - // elsewhere to hold aggregated state. To improve, we stripe the AtomicLong into N instances, - // where N is the number of available processors. Each record operation chooses the appropriate - // instance to use based on the modulo of its thread id and N. This is a more naive / simple - // implementation compared to the striping used under the hood in java.util.concurrent classes - // like LongAdder - contention and hot spots can still occur if recording thread ids happen to - // resolve to the same index. Further improvement is possible. + // elsewhere to hold aggregated state. To reduce contention, the count is striped across the + // available processors. This is simpler than the striping used by LongAdder, so hot spots remain + // possible when several recording threads resolve to the same stripe. private final AtomicLong[] stripedObservationCounts; - private double[] observationBuffer = new double[0]; - private int bufferPos = 0; - private boolean reset = false; - + // phaseTransition() makes appendPhase odd while the collector changes generations. An appender + // only buffers its observation if it sees the same even phase before and after incrementing its + // stripe. appendersInFlight closes the gap between those two phase reads. + private final AtomicLong appendersInFlight = new AtomicLong(); + private final AtomicLong appendPhase = new AtomicLong(); + private final ReentrantLock observationLock = new ReentrantLock(); + private boolean reset; + private long observationCountOffset; + @Nullable private volatile Generation activeGeneration; ReentrantLock appendLock = new ReentrantLock(); ReentrantLock runLock = new ReentrantLock(); - Condition bufferFilled = appendLock.newCondition(); + private final Condition bufferSpaceAvailable = appendLock.newCondition(); + private final long maxSpinWaitNanos; + private final int maxBufferSize; + private final Runnable beforeAppendLock; Buffer() { + this(DEFAULT_MAX_SPIN_WAIT_NANOS, DEFAULT_MAX_BUFFER_SIZE, () -> {}); + } + + Buffer(long maxSpinWaitNanos) { + this(maxSpinWaitNanos, DEFAULT_MAX_BUFFER_SIZE, () -> {}); + } + + Buffer(long maxSpinWaitNanos, int maxBufferSize, Runnable beforeAppendLock) { + if (maxBufferSize <= 0) { + throw new IllegalArgumentException("maxBufferSize must be positive"); + } + this.maxSpinWaitNanos = maxSpinWaitNanos; + this.maxBufferSize = maxBufferSize; + this.beforeAppendLock = beforeAppendLock; stripedObservationCounts = new AtomicLong[Runtime.getRuntime().availableProcessors()]; for (int i = 0; i < stripedObservationCounts.length; i++) { - stripedObservationCounts[i] = new AtomicLong(0); + stripedObservationCounts[i] = new AtomicLong(); } } boolean append(double value) { - int index = stripeIndex(Thread.currentThread().getId(), stripedObservationCounts.length); - AtomicLong observationCountForThread = stripedObservationCounts[index]; - long count = observationCountForThread.incrementAndGet(); - if ((count & bufferActiveBit) == 0) { - return false; // sign bit not set -> buffer not active. - } else { - doAppend(value); + AtomicLong counter = + stripedObservationCounts[ + stripeIndex(Thread.currentThread().getId(), stripedObservationCounts.length)]; + appendersInFlight.incrementAndGet(); + long phase = appendPhase.get(); + long count = counter.incrementAndGet(); + boolean phaseChanged = appendPhase.get() != phase; + appendersInFlight.decrementAndGet(); + if (phaseChanged || (phase & 1L) != 0 || (count & bufferActiveBit) == 0) { + return false; + } + Generation generation = activeGeneration; + if (generation == null) { + return false; + } + beforeAppendLock.run(); + appendLock.lock(); + try { + Generation current = activeGeneration; + if (current != generation || !generation.active) { + return false; + } + while (generation.size >= maxBufferSize && generation.active) { + bufferSpaceAvailable.awaitUninterruptibly(); + } + if (!generation.active) { + return false; + } + if (generation.size >= generation.values.length) { + int doubled = + generation.values.length > maxBufferSize / 2 + ? maxBufferSize + : generation.values.length * 2; + generation.values = + Arrays.copyOf( + generation.values, + Math.min(maxBufferSize, Math.max(INITIAL_BUFFER_SIZE, Math.max(1, doubled)))); + } + generation.values[generation.size++] = value; return true; + } finally { + appendLock.unlock(); } } @@ -58,89 +125,109 @@ static int stripeIndex(long threadId, int stripeCount) { return (int) Math.floorMod(threadId, stripeCount); } - private void doAppend(double amount) { - appendLock.lock(); - try { - if (bufferPos >= observationBuffer.length) { - observationBuffer = Arrays.copyOf(observationBuffer, observationBuffer.length + 128); - } - observationBuffer[bufferPos] = amount; - bufferPos++; + void reset() { + reset = true; + } - bufferFilled.signalAll(); + T observeDirect(Supplier observeFunction) { + observationLock.lock(); + try { + return observeFunction.get(); } finally { - appendLock.unlock(); + observationLock.unlock(); } } - /** Must be called by the runnable in the run() method. */ - void reset() { - reset = true; + @SuppressWarnings({"NullAway", "ThreadPriorityCheck"}) + T run( + Function complete, + Supplier createResult, + Consumer observeFunction) { + return run(complete, createResult, observeFunction, true); } - @SuppressWarnings("ThreadPriorityCheck") + @SuppressWarnings({"NullAway", "ThreadPriorityCheck"}) T run( Function complete, Supplier createResult, - Consumer observeFunction) { + Consumer observeFunction, + boolean failOnTimeout) { + Generation generation = new Generation(); double[] buffer; int bufferSize; - T result; - + boolean timedOut = false; + T result = null; runLock.lock(); try { - // Signal that the buffer is active. - long expectedCount = 0L; - for (AtomicLong observationCount : stripedObservationCounts) { - expectedCount += observationCount.getAndAdd(bufferActiveBit); + phaseTransition(); + long expectedCount; + appendLock.lock(); + try { + activeGeneration = generation; + long total = 0; + for (AtomicLong counter : stripedObservationCounts) { + total += counter.getAndAdd(bufferActiveBit); + } + expectedCount = total - observationCountOffset; + } finally { + appendLock.unlock(); } - + appendPhase.incrementAndGet(); + long deadline = System.nanoTime() + maxSpinWaitNanos; while (!complete.apply(expectedCount)) { - // Wait until all in-flight threads have added their observations to the histogram / - // summary. - // we can't use a condition here, because the other thread doesn't have a lock as it's on - // the fast path. - Thread.yield(); - } - result = createResult.get(); - - // Signal that the buffer is inactive. - long expectedBufferSize = 0; - if (reset) { - for (AtomicLong observationCount : stripedObservationCounts) { - expectedBufferSize += observationCount.getAndSet(0) & ~bufferActiveBit; - } - reset = false; - } else { - for (AtomicLong observationCount : stripedObservationCounts) { - expectedBufferSize += observationCount.addAndGet(bufferActiveBit); + if (System.nanoTime() - deadline >= 0) { + timedOut = true; + break; } + Thread.yield(); } - expectedBufferSize -= expectedCount; - - appendLock.lock(); + observationLock.lock(); try { - while (bufferPos < expectedBufferSize) { - // Wait until all in-flight threads have added their observations to the buffer. - bufferFilled.await(); - } + result = timedOut ? null : createResult.get(); } finally { - appendLock.unlock(); + try { + phaseTransition(); + appendLock.lock(); + try { + generation.active = false; + for (AtomicLong counter : stripedObservationCounts) { + counter.addAndGet(bufferActiveBit); + } + if (reset) { + observationCountOffset += expectedCount; + reset = false; + } + activeGeneration = null; + buffer = generation.values; + bufferSize = generation.size; + generation.values = new double[0]; + generation.size = 0; + bufferSpaceAvailable.signalAll(); + } finally { + appendLock.unlock(); + } + appendPhase.incrementAndGet(); + for (int i = 0; i < bufferSize; i++) { + observeFunction.accept(buffer[i]); + } + } finally { + observationLock.unlock(); + } } - - buffer = observationBuffer; - bufferSize = bufferPos; - observationBuffer = new double[0]; - bufferPos = 0; - } catch (InterruptedException e) { - throw new RuntimeException(e); + if (timedOut && failOnTimeout) { + throw new IllegalStateException("Timed out while waiting for in-flight observations."); + } + return result; } finally { runLock.unlock(); } + } - for (int i = 0; i < bufferSize; i++) { - observeFunction.accept(buffer[i]); + @SuppressWarnings("ThreadPriorityCheck") + private void phaseTransition() { + appendPhase.incrementAndGet(); + while (appendersInFlight.get() != 0) { + Thread.yield(); } - return result; } } 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..faddd0f27 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 @@ -243,7 +243,8 @@ public void observe(double value) { return; } if (!buffer.append(value)) { - doObserve(value, false); + boolean nativeBucketCreated = buffer.observeDirect(() -> doObserve(value)); + maybeResetOrScaleDown(value, nativeBucketCreated); } if (exemplarSampler != null) { exemplarSampler.observe(value); @@ -257,14 +258,15 @@ public void observeWithExemplar(double value, Labels labels) { return; } if (!buffer.append(value)) { - doObserve(value, false); + boolean nativeBucketCreated = buffer.observeDirect(() -> doObserve(value)); + maybeResetOrScaleDown(value, nativeBucketCreated); } if (exemplarSampler != null) { exemplarSampler.observeWithExemplar(value, labels); } } - private void doObserve(double value, boolean fromBuffer) { + private boolean doObserve(double value) { // 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. @@ -287,6 +289,11 @@ private void doObserve(double value, boolean fromBuffer) { count .increment(); // must be the last step, because count is used to signal that the operation // is complete. + return nativeBucketCreated; + } + + private void doObserve(double value, boolean fromBuffer) { + boolean nativeBucketCreated = doObserve(value); if (!fromBuffer) { // maybeResetOrScaleDown will switch to the buffer, // which won't work if we are currently still processing observations from the buffer. @@ -302,7 +309,7 @@ private void doObserve(double value, boolean fromBuffer) { private HistogramSnapshot.HistogramDataPointSnapshot collect(Labels labels) { Exemplars exemplars = exemplarSampler != null ? exemplarSampler.collect() : Exemplars.EMPTY; return buffer.run( - expectedCount -> count.sum() == expectedCount, + expectedCount -> count.sum() >= expectedCount, () -> { if (classicUpperBounds.length == 0) { // native only @@ -437,14 +444,15 @@ private void maybeResetOrScaleDown(double value, boolean nativeBucketCreated) { // If nativeSchema < initialNativeSchema the histogram has been scaled down. // So if resetDurationExpired we will reset it to restore the original native schema. buffer.run( - expectedCount -> count.sum() == expectedCount, + expectedCount -> count.sum() >= expectedCount, () -> { if (maybeReset()) { wasReset.set(true); } return null; }, - v -> doObserve(v, true)); + v -> doObserve(v, true), + false); } else if (nativeBucketCreated) { // If a new bucket was created we need to check if nativeMaxBuckets is exceeded // and scale down if so. @@ -453,7 +461,11 @@ private void maybeResetOrScaleDown(double value, boolean nativeBucketCreated) { if (wasReset.get()) { // We just discarded the newly observed value. Observe it again. if (!buffer.append(value)) { - doObserve(value, true); + buffer.observeDirect( + () -> { + doObserve(value, true); + return null; + }); } } } @@ -468,7 +480,7 @@ private void maybeScaleDown(AtomicBoolean wasReset) { return; } buffer.run( - expectedCount -> count.sum() == expectedCount, + expectedCount -> count.sum() >= expectedCount, () -> { // Now we are in the synchronized block while new observations go into the buffer. // Check again if we need to limit the bucket size, because another thread might @@ -488,7 +500,8 @@ private void maybeScaleDown(AtomicBoolean wasReset) { doubleBucketWidth(); return null; }, - v -> doObserve(v, true)); + v -> doObserve(v, true), + false); } // maybeReset is called in the synchronized block while new observations go into the buffer. diff --git a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Summary.java b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Summary.java index 043f31129..832c10619 100644 --- a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Summary.java +++ b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Summary.java @@ -185,7 +185,11 @@ public void observe(double value) { return; } if (!buffer.append(value)) { - doObserve(value); + buffer.observeDirect( + () -> { + doObserve(value); + return null; + }); } if (exemplarSampler != null) { exemplarSampler.observe(value); @@ -198,7 +202,11 @@ public void observeWithExemplar(double value, Labels labels) { return; } if (!buffer.append(value)) { - doObserve(value); + buffer.observeDirect( + () -> { + doObserve(value); + return null; + }); } if (exemplarSampler != null) { exemplarSampler.observeWithExemplar(value, labels); @@ -217,7 +225,7 @@ private void doObserve(double amount) { private SummarySnapshot.SummaryDataPointSnapshot collect(Labels labels) { return buffer.run( - expectedCount -> count.sum() == expectedCount, + expectedCount -> count.sum() >= expectedCount, // Note: Exemplars are currently hard-coded as empty for Summary metrics. // While exemplars are sampled during observe() and observeWithExemplar() calls // via the exemplarSampler field, they are not included in the snapshot to maintain diff --git a/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/BufferTest.java b/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/BufferTest.java index dccd3b4eb..abc7ae991 100644 --- a/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/BufferTest.java +++ b/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/BufferTest.java @@ -1,7 +1,17 @@ package io.prometheus.metrics.core.metrics; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatExceptionOfType; +import io.prometheus.metrics.model.snapshots.CounterSnapshot; +import io.prometheus.metrics.model.snapshots.Labels; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; import org.junit.jupiter.api.Test; class BufferTest { @@ -12,4 +22,171 @@ void stripeIndexDoesNotOverflowWhenThreadIdNarrowsToIntegerMinValue() { assertThat(Buffer.stripeIndex(2_147_483_648L, 6)).isEqualTo(2); assertThat(Buffer.stripeIndex(2_147_483_648L, 12)).isEqualTo(8); } + + @Test + void timeoutDeactivatesBufferAndReplaysBufferedObservations() throws InterruptedException { + Buffer buffer = new Buffer(TimeUnit.SECONDS.toNanos(1)); + CountDownLatch spinWaitStarted = new CountDownLatch(1); + AtomicBoolean keepWaiting = new AtomicBoolean(true); + List replayedObservations = new ArrayList<>(); + AtomicBoolean timedOut = new AtomicBoolean(false); + + Thread runner = + new Thread( + () -> { + try { + buffer.run( + expectedCount -> { + spinWaitStarted.countDown(); + return !keepWaiting.get(); + }, + () -> new CounterSnapshot.CounterDataPointSnapshot(0, Labels.EMPTY, null, 0), + replayedObservations::add); + } catch (IllegalStateException expected) { + timedOut.set(true); + } + }); + runner.start(); + + assertThat(spinWaitStarted.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(buffer.append(1.0)).isTrue(); + runner.join(5_000); + + assertThat(timedOut).isTrue(); + assertThat(replayedObservations).containsExactly(1.0); + assertThat(buffer.append(2.0)).isFalse(); + } + + @Test + void timeoutDoesNotCreateSnapshot() { + Buffer buffer = new Buffer(TimeUnit.MILLISECONDS.toNanos(1)); + + assertThatExceptionOfType(IllegalStateException.class) + .isThrownBy( + () -> + buffer.run( + expectedCount -> false, + () -> { + throw new AssertionError("snapshot should not be created"); + }, + ignored -> {})) + .withMessage("Timed out while waiting for in-flight observations."); + } + + @Test + void stalledAppenderAfterStripeActivationDoesNotBlockRun() throws InterruptedException { + CountDownLatch started = new CountDownLatch(1); + CountDownLatch proceed = new CountDownLatch(1); + CountDownLatch stalled = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + AtomicReference appended = new AtomicReference<>(); + Buffer buffer = + new Buffer( + TimeUnit.SECONDS.toNanos(1), + 16, + () -> { + stalled.countDown(); + try { + release.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }); + Thread runner = + new Thread( + () -> + buffer.run( + ignored -> { + started.countDown(); + return proceed.getCount() == 0; + }, + () -> new CounterSnapshot.CounterDataPointSnapshot(0, Labels.EMPTY, null, 0), + ignored -> {})); + runner.start(); + assertThat(started.await(5, TimeUnit.SECONDS)).isTrue(); + Thread appender = new Thread(() -> appended.set(buffer.append(1.0))); + appender.start(); + assertThat(stalled.await(5, TimeUnit.SECONDS)).isTrue(); + proceed.countDown(); + runner.join(5_000); + assertThat(runner.isAlive()).isFalse(); + release.countDown(); + appender.join(5_000); + assertThat(appended).hasValue(false); + } + + @Test + void lateAppenderCannotBeAddedToTheNextGeneration() throws InterruptedException { + CountDownLatch firstRunStarted = new CountDownLatch(1); + CountDownLatch firstRunMayFinish = new CountDownLatch(1); + CountDownLatch stalled = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + CountDownLatch secondRunStarted = new CountDownLatch(1); + AtomicBoolean appended = new AtomicBoolean(); + AtomicLong completedObservations = new AtomicLong(); + Buffer buffer = + new Buffer( + TimeUnit.SECONDS.toNanos(1), + 16, + () -> { + stalled.countDown(); + try { + release.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }); + Thread firstRun = + new Thread( + () -> + buffer.run( + ignored -> { + firstRunStarted.countDown(); + return firstRunMayFinish.getCount() == 0; + }, + () -> new CounterSnapshot.CounterDataPointSnapshot(0, Labels.EMPTY, null, 0), + ignored -> {})); + firstRun.start(); + assertThat(firstRunStarted.await(5, TimeUnit.SECONDS)).isTrue(); + + Thread appender = + new Thread( + () -> { + appended.set(buffer.append(1.0)); + if (!appended.get()) { + buffer.observeDirect( + () -> { + completedObservations.incrementAndGet(); + return null; + }); + } + }); + appender.start(); + assertThat(stalled.await(5, TimeUnit.SECONDS)).isTrue(); + + firstRunMayFinish.countDown(); + firstRun.join(5_000); + assertThat(firstRun.isAlive()).isFalse(); + + Thread secondRun = + new Thread( + () -> + buffer.run( + expectedCount -> { + secondRunStarted.countDown(); + return completedObservations.get() >= expectedCount; + }, + () -> new CounterSnapshot.CounterDataPointSnapshot(0, Labels.EMPTY, null, 0), + ignored -> {})); + secondRun.start(); + assertThat(secondRunStarted.await(5, TimeUnit.SECONDS)).isTrue(); + release.countDown(); + appender.join(5_000); + secondRun.join(5_000); + + assertThat(appender.isAlive()).isFalse(); + assertThat(secondRun.isAlive()).isFalse(); + assertThat(appended).isFalse(); + assertThat(completedObservations).hasValue(1); + } } From 599ddca1ee2a342f60eecaad09b0cbd3937c0517 Mon Sep 17 00:00:00 2001 From: Gregor Zeitlinger Date: Thu, 23 Jul 2026 11:28:38 +0000 Subject: [PATCH 3/5] fix: tolerate short collection stalls Signed-off-by: Gregor Zeitlinger --- .../main/java/io/prometheus/metrics/core/metrics/Buffer.java | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Buffer.java b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Buffer.java index cd29214ec..f077cbc2c 100644 --- a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Buffer.java +++ b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Buffer.java @@ -21,7 +21,9 @@ */ class Buffer { private static final long bufferActiveBit = 1L << 63; - private static final long DEFAULT_MAX_SPIN_WAIT_NANOS = TimeUnit.SECONDS.toNanos(1); + // Keep collection bounded without failing healthy scrapes during short periods of scheduler or + // CI-host contention. + private static final long DEFAULT_MAX_SPIN_WAIT_NANOS = TimeUnit.SECONDS.toNanos(5); private static final int DEFAULT_MAX_BUFFER_SIZE = 1_000_000; private static final int INITIAL_BUFFER_SIZE = 128; From 4d69da18904fe3104da063c3392e427eed18540d Mon Sep 17 00:00:00 2001 From: Gregor Zeitlinger Date: Thu, 27 Aug 2026 17:05:51 +0000 Subject: [PATCH 4/5] fix: address buffer collection review feedback Signed-off-by: Gregor Zeitlinger --- .../metrics/core/metrics/Buffer.java | 77 ++++++---- .../metrics/core/metrics/Histogram.java | 28 +--- .../metrics/core/metrics/Summary.java | 2 +- .../metrics/core/metrics/BufferTest.java | 131 ++++++++++++++++-- 4 files changed, 183 insertions(+), 55 deletions(-) diff --git a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Buffer.java b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Buffer.java index f077cbc2c..7ef9beb1f 100644 --- a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Buffer.java +++ b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Buffer.java @@ -1,5 +1,7 @@ package io.prometheus.metrics.core.metrics; +import static java.util.Objects.requireNonNull; + import io.prometheus.metrics.model.snapshots.DataPointSnapshot; import java.util.Arrays; import java.util.concurrent.TimeUnit; @@ -18,18 +20,25 @@ * that generation while the collector waits for observations from the previous phase to finish. The * collector then creates a snapshot, deactivates the generation, and replays its buffered * observations into the live metric state. + * + *

The default collection wait is five seconds. A generation is capped at one million buffered + * observations (about eight MiB of double storage) to keep a stalled collection from growing + * without bound; the cap applies backpressure rather than dropping observations. */ class Buffer { - private static final long bufferActiveBit = 1L << 63; + private static final long BUFFER_ACTIVE_BIT = 1L << 63; + private static final double[] EMPTY_BUFFER = new double[0]; + // Keep collection bounded without failing healthy scrapes during short periods of scheduler or - // CI-host contention. + // CI-host contention. The one-million-observation cap uses at most 8 MiB for one generation; + // it is deliberately an internal safeguard rather than a data-loss policy. private static final long DEFAULT_MAX_SPIN_WAIT_NANOS = TimeUnit.SECONDS.toNanos(5); private static final int DEFAULT_MAX_BUFFER_SIZE = 1_000_000; private static final int INITIAL_BUFFER_SIZE = 128; /** Observations buffered during one collection cycle. */ private static final class Generation { - private double[] values = new double[0]; + private double[] values = EMPTY_BUFFER; private int size; private boolean active = true; } @@ -40,11 +49,9 @@ private static final class Generation { // available processors. This is simpler than the striping used by LongAdder, so hot spots remain // possible when several recording threads resolve to the same stripe. private final AtomicLong[] stripedObservationCounts; - // phaseTransition() makes appendPhase odd while the collector changes generations. An appender - // only buffers its observation if it sees the same even phase before and after incrementing its - // stripe. appendersInFlight closes the gap between those two phase reads. + // phaseTransition() waits for appenders that are between entering append() and updating their + // stripe. This closes the handoff window around the collector's getAndAdd(BUFFER_ACTIVE_BIT). private final AtomicLong appendersInFlight = new AtomicLong(); - private final AtomicLong appendPhase = new AtomicLong(); private final ReentrantLock observationLock = new ReentrantLock(); private boolean reset; private long observationCountOffset; @@ -82,11 +89,16 @@ boolean append(double value) { stripedObservationCounts[ stripeIndex(Thread.currentThread().getId(), stripedObservationCounts.length)]; appendersInFlight.incrementAndGet(); - long phase = appendPhase.get(); - long count = counter.incrementAndGet(); - boolean phaseChanged = appendPhase.get() != phase; - appendersInFlight.decrementAndGet(); - if (phaseChanged || (phase & 1L) != 0 || (count & bufferActiveBit) == 0) { + long count; + try { + count = counter.incrementAndGet(); + } finally { + appendersInFlight.decrementAndGet(); + } + // The active bit is the exact handoff decision. In particular, do not use the phase transition + // as an additional gate: an observation that increments its stripe in that window must either + // be included in expectedCount or be buffered in the current generation. + if ((count & BUFFER_ACTIVE_BIT) == 0) { return false; } Generation generation = activeGeneration; @@ -101,7 +113,12 @@ boolean append(double value) { return false; } while (generation.size >= maxBufferSize && generation.active) { - bufferSpaceAvailable.awaitUninterruptibly(); + try { + bufferSpaceAvailable.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return false; + } } if (!generation.active) { return false; @@ -113,8 +130,7 @@ boolean append(double value) { : generation.values.length * 2; generation.values = Arrays.copyOf( - generation.values, - Math.min(maxBufferSize, Math.max(INITIAL_BUFFER_SIZE, Math.max(1, doubled)))); + generation.values, Math.min(maxBufferSize, Math.max(INITIAL_BUFFER_SIZE, doubled))); } generation.values[generation.size++] = value; return true; @@ -132,6 +148,11 @@ void reset() { } T observeDirect(Supplier observeFunction) { + // In steady state this is the lock-free path used before this buffer was introduced. Keep the + // lock only while a generation is active, so direct observations cannot race collection/replay. + if (activeGeneration == null) { + return observeFunction.get(); + } observationLock.lock(); try { return observeFunction.get(); @@ -140,15 +161,16 @@ T observeDirect(Supplier observeFunction) { } } - @SuppressWarnings({"NullAway", "ThreadPriorityCheck"}) + @SuppressWarnings("ThreadPriorityCheck") T run( Function complete, Supplier createResult, Consumer observeFunction) { - return run(complete, createResult, observeFunction, true); + return requireNonNull(run(complete, createResult, observeFunction, true)); } - @SuppressWarnings({"NullAway", "ThreadPriorityCheck"}) + @SuppressWarnings("ThreadPriorityCheck") + @Nullable T run( Function complete, Supplier createResult, @@ -168,13 +190,12 @@ T run( activeGeneration = generation; long total = 0; for (AtomicLong counter : stripedObservationCounts) { - total += counter.getAndAdd(bufferActiveBit); + total += counter.getAndAdd(BUFFER_ACTIVE_BIT); } expectedCount = total - observationCountOffset; } finally { appendLock.unlock(); } - appendPhase.incrementAndGet(); long deadline = System.nanoTime() + maxSpinWaitNanos; while (!complete.apply(expectedCount)) { if (System.nanoTime() - deadline >= 0) { @@ -193,25 +214,26 @@ T run( try { generation.active = false; for (AtomicLong counter : stripedObservationCounts) { - counter.addAndGet(bufferActiveBit); + counter.addAndGet(BUFFER_ACTIVE_BIT); } if (reset) { observationCountOffset += expectedCount; reset = false; } - activeGeneration = null; buffer = generation.values; bufferSize = generation.size; - generation.values = new double[0]; + generation.values = EMPTY_BUFFER; generation.size = 0; bufferSpaceAvailable.signalAll(); } finally { appendLock.unlock(); } - appendPhase.incrementAndGet(); for (int i = 0; i < bufferSize; i++) { observeFunction.accept(buffer[i]); } + // Keep the inactive generation visible until replay completes. An appender that loses the + // generation race must take observationLock before observing directly. + activeGeneration = null; } finally { observationLock.unlock(); } @@ -227,8 +249,13 @@ T run( @SuppressWarnings("ThreadPriorityCheck") private void phaseTransition() { - appendPhase.incrementAndGet(); + long deadline = System.nanoTime() + maxSpinWaitNanos; while (appendersInFlight.get() != 0) { + if (System.nanoTime() - deadline >= 0) { + // A late appender uses the active-bit decision and generation identity, so proceeding is + // safe even if a suspended appender did not reach its stripe in time. + return; + } Thread.yield(); } } 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 faddd0f27..d03ac9b97 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 @@ -292,24 +292,10 @@ private boolean doObserve(double value) { return nativeBucketCreated; } - private void doObserve(double value, boolean fromBuffer) { - boolean nativeBucketCreated = doObserve(value); - if (!fromBuffer) { - // maybeResetOrScaleDown will switch to the buffer, - // which won't work if we are currently still processing observations from the buffer. - // The reason is that before switching to the buffer we wait for all pending observations to - // be counted. - // If we do this while still applying observations from the buffer, the pending observations - // from - // the buffer will never be counted, and the buffer.run() method will wait forever. - maybeResetOrScaleDown(value, nativeBucketCreated); - } - } - private HistogramSnapshot.HistogramDataPointSnapshot collect(Labels labels) { Exemplars exemplars = exemplarSampler != null ? exemplarSampler.collect() : Exemplars.EMPTY; return buffer.run( - expectedCount -> count.sum() >= expectedCount, + expectedCount -> count.sum() == expectedCount, () -> { if (classicUpperBounds.length == 0) { // native only @@ -346,7 +332,7 @@ private HistogramSnapshot.HistogramDataPointSnapshot collect(Labels labels) { createdTimeMillis); } }, - v -> doObserve(v, true)); + this::doObserve); } private boolean addToNativeBucket(double value, ConcurrentHashMap buckets) { @@ -444,14 +430,14 @@ private void maybeResetOrScaleDown(double value, boolean nativeBucketCreated) { // If nativeSchema < initialNativeSchema the histogram has been scaled down. // So if resetDurationExpired we will reset it to restore the original native schema. buffer.run( - expectedCount -> count.sum() >= expectedCount, + expectedCount -> count.sum() == expectedCount, () -> { if (maybeReset()) { wasReset.set(true); } return null; }, - v -> doObserve(v, true), + this::doObserve, false); } else if (nativeBucketCreated) { // If a new bucket was created we need to check if nativeMaxBuckets is exceeded @@ -463,7 +449,7 @@ private void maybeResetOrScaleDown(double value, boolean nativeBucketCreated) { if (!buffer.append(value)) { buffer.observeDirect( () -> { - doObserve(value, true); + doObserve(value); return null; }); } @@ -480,7 +466,7 @@ private void maybeScaleDown(AtomicBoolean wasReset) { return; } buffer.run( - expectedCount -> count.sum() >= expectedCount, + expectedCount -> count.sum() == expectedCount, () -> { // Now we are in the synchronized block while new observations go into the buffer. // Check again if we need to limit the bucket size, because another thread might @@ -500,7 +486,7 @@ private void maybeScaleDown(AtomicBoolean wasReset) { doubleBucketWidth(); return null; }, - v -> doObserve(v, true), + this::doObserve, false); } diff --git a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Summary.java b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Summary.java index 832c10619..d75ac6c74 100644 --- a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Summary.java +++ b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Summary.java @@ -225,7 +225,7 @@ private void doObserve(double amount) { private SummarySnapshot.SummaryDataPointSnapshot collect(Labels labels) { return buffer.run( - expectedCount -> count.sum() >= expectedCount, + expectedCount -> count.sum() == expectedCount, // Note: Exemplars are currently hard-coded as empty for Summary metrics. // While exemplars are sampled during observe() and observeWithExemplar() calls // via the exemplarSampler field, they are not included in the snapshot to maintain diff --git a/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/BufferTest.java b/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/BufferTest.java index abc7ae991..279f112e7 100644 --- a/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/BufferTest.java +++ b/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/BufferTest.java @@ -27,7 +27,6 @@ void stripeIndexDoesNotOverflowWhenThreadIdNarrowsToIntegerMinValue() { void timeoutDeactivatesBufferAndReplaysBufferedObservations() throws InterruptedException { Buffer buffer = new Buffer(TimeUnit.SECONDS.toNanos(1)); CountDownLatch spinWaitStarted = new CountDownLatch(1); - AtomicBoolean keepWaiting = new AtomicBoolean(true); List replayedObservations = new ArrayList<>(); AtomicBoolean timedOut = new AtomicBoolean(false); @@ -38,14 +37,16 @@ void timeoutDeactivatesBufferAndReplaysBufferedObservations() throws Interrupted buffer.run( expectedCount -> { spinWaitStarted.countDown(); - return !keepWaiting.get(); + return false; }, () -> new CounterSnapshot.CounterDataPointSnapshot(0, Labels.EMPTY, null, 0), replayedObservations::add); } catch (IllegalStateException expected) { timedOut.set(true); } - }); + }, + "buffer-timeout-runner"); + runner.setDaemon(true); runner.start(); assertThat(spinWaitStarted.await(5, TimeUnit.SECONDS)).isTrue(); @@ -73,6 +74,111 @@ void timeoutDoesNotCreateSnapshot() { .withMessage("Timed out while waiting for in-flight observations."); } + @Test + void fullBufferUnblocksAppenderWhenGenerationIsDeactivated() throws InterruptedException { + CountDownLatch runStarted = new CountDownLatch(1); + CountDownLatch secondAppenderEntered = new CountDownLatch(1); + AtomicLong beforeAppendCount = new AtomicLong(); + AtomicReference appended = new AtomicReference<>(); + AtomicBoolean timedOut = new AtomicBoolean(); + Buffer buffer = + new Buffer( + TimeUnit.MILLISECONDS.toNanos(250), + 1, + () -> { + if (beforeAppendCount.incrementAndGet() == 2) { + secondAppenderEntered.countDown(); + } + }); + Thread runner = + new Thread( + () -> { + try { + buffer.run( + ignored -> { + runStarted.countDown(); + return false; + }, + () -> new CounterSnapshot.CounterDataPointSnapshot(0, Labels.EMPTY, null, 0), + ignored -> {}); + } catch (IllegalStateException expected) { + timedOut.set(true); + } + }, + "buffer-full-runner"); + runner.setDaemon(true); + runner.start(); + assertThat(runStarted.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(buffer.append(1.0)).isTrue(); + + Thread appender = new Thread(() -> appended.set(buffer.append(2.0)), "buffer-full-appender"); + appender.setDaemon(true); + appender.start(); + assertThat(secondAppenderEntered.await(5, TimeUnit.SECONDS)).isTrue(); + runner.join(5_000); + appender.join(5_000); + + assertThat(timedOut).isTrue(); + assertThat(appender.isAlive()).isFalse(); + assertThat(appended).hasValue(false); + } + + @Test + void interruptedAppenderLeavesBoundedBufferWait() throws InterruptedException { + CountDownLatch runStarted = new CountDownLatch(1); + CountDownLatch secondAppenderEntered = new CountDownLatch(1); + AtomicLong beforeAppendCount = new AtomicLong(); + AtomicBoolean interrupted = new AtomicBoolean(); + AtomicReference appended = new AtomicReference<>(); + Buffer buffer = + new Buffer( + TimeUnit.SECONDS.toNanos(1), + 1, + () -> { + if (beforeAppendCount.incrementAndGet() == 2) { + secondAppenderEntered.countDown(); + } + }); + Thread runner = + new Thread( + () -> { + try { + buffer.run( + ignored -> { + runStarted.countDown(); + return false; + }, + () -> new CounterSnapshot.CounterDataPointSnapshot(0, Labels.EMPTY, null, 0), + ignored -> {}); + } catch (IllegalStateException expected) { + // The runner is only used to hold the generation open for this test. + } + }, + "buffer-interrupt-runner"); + runner.setDaemon(true); + runner.start(); + assertThat(runStarted.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(buffer.append(1.0)).isTrue(); + + Thread appender = + new Thread( + () -> { + appended.set(buffer.append(2.0)); + interrupted.set(Thread.currentThread().isInterrupted()); + }, + "buffer-interrupt-appender"); + appender.setDaemon(true); + appender.start(); + assertThat(secondAppenderEntered.await(5, TimeUnit.SECONDS)).isTrue(); + appender.interrupt(); + appender.join(5_000); + runner.join(5_000); + + assertThat(appender.isAlive()).isFalse(); + assertThat(appended).hasValue(false); + assertThat(interrupted).isTrue(); + } + @Test void stalledAppenderAfterStripeActivationDoesNotBlockRun() throws InterruptedException { CountDownLatch started = new CountDownLatch(1); @@ -101,10 +207,13 @@ void stalledAppenderAfterStripeActivationDoesNotBlockRun() throws InterruptedExc return proceed.getCount() == 0; }, () -> new CounterSnapshot.CounterDataPointSnapshot(0, Labels.EMPTY, null, 0), - ignored -> {})); + ignored -> {}), + "buffer-stalled-runner"); + runner.setDaemon(true); runner.start(); assertThat(started.await(5, TimeUnit.SECONDS)).isTrue(); - Thread appender = new Thread(() -> appended.set(buffer.append(1.0))); + Thread appender = new Thread(() -> appended.set(buffer.append(1.0)), "buffer-stalled-appender"); + appender.setDaemon(true); appender.start(); assertThat(stalled.await(5, TimeUnit.SECONDS)).isTrue(); proceed.countDown(); @@ -145,7 +254,9 @@ void lateAppenderCannotBeAddedToTheNextGeneration() throws InterruptedException return firstRunMayFinish.getCount() == 0; }, () -> new CounterSnapshot.CounterDataPointSnapshot(0, Labels.EMPTY, null, 0), - ignored -> {})); + ignored -> {}), + "buffer-first-runner"); + firstRun.setDaemon(true); firstRun.start(); assertThat(firstRunStarted.await(5, TimeUnit.SECONDS)).isTrue(); @@ -160,7 +271,9 @@ void lateAppenderCannotBeAddedToTheNextGeneration() throws InterruptedException return null; }); } - }); + }, + "buffer-late-appender"); + appender.setDaemon(true); appender.start(); assertThat(stalled.await(5, TimeUnit.SECONDS)).isTrue(); @@ -177,7 +290,9 @@ void lateAppenderCannotBeAddedToTheNextGeneration() throws InterruptedException return completedObservations.get() >= expectedCount; }, () -> new CounterSnapshot.CounterDataPointSnapshot(0, Labels.EMPTY, null, 0), - ignored -> {})); + ignored -> {}), + "buffer-second-runner"); + secondRun.setDaemon(true); secondRun.start(); assertThat(secondRunStarted.await(5, TimeUnit.SECONDS)).isTrue(); release.countDown(); From a1657911077e0905b4281eec8e6d4c3a466ac779 Mon Sep 17 00:00:00 2001 From: Gregor Zeitlinger Date: Fri, 28 Aug 2026 10:36:28 +0000 Subject: [PATCH 5/5] fix: remove buffer hot-path coordination Signed-off-by: Gregor Zeitlinger --- .../metrics/core/metrics/Buffer.java | 32 ++----------- .../metrics/core/metrics/BufferTest.java | 47 +------------------ 2 files changed, 5 insertions(+), 74 deletions(-) diff --git a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Buffer.java b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Buffer.java index 7ef9beb1f..6dc68f8e6 100644 --- a/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Buffer.java +++ b/prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Buffer.java @@ -49,9 +49,6 @@ private static final class Generation { // available processors. This is simpler than the striping used by LongAdder, so hot spots remain // possible when several recording threads resolve to the same stripe. private final AtomicLong[] stripedObservationCounts; - // phaseTransition() waits for appenders that are between entering append() and updating their - // stripe. This closes the handoff window around the collector's getAndAdd(BUFFER_ACTIVE_BIT). - private final AtomicLong appendersInFlight = new AtomicLong(); private final ReentrantLock observationLock = new ReentrantLock(); private boolean reset; private long observationCountOffset; @@ -88,16 +85,10 @@ boolean append(double value) { AtomicLong counter = stripedObservationCounts[ stripeIndex(Thread.currentThread().getId(), stripedObservationCounts.length)]; - appendersInFlight.incrementAndGet(); - long count; - try { - count = counter.incrementAndGet(); - } finally { - appendersInFlight.decrementAndGet(); - } - // The active bit is the exact handoff decision. In particular, do not use the phase transition - // as an additional gate: an observation that increments its stripe in that window must either - // be included in expectedCount or be buffered in the current generation. + long count = counter.incrementAndGet(); + // The active bit is the exact handoff decision. An observation either increments its stripe + // before the collector's getAndAdd(BUFFER_ACTIVE_BIT) and takes the direct path, or sees the + // active bit and is buffered in the current generation. if ((count & BUFFER_ACTIVE_BIT) == 0) { return false; } @@ -183,7 +174,6 @@ T run( T result = null; runLock.lock(); try { - phaseTransition(); long expectedCount; appendLock.lock(); try { @@ -209,7 +199,6 @@ T run( result = timedOut ? null : createResult.get(); } finally { try { - phaseTransition(); appendLock.lock(); try { generation.active = false; @@ -246,17 +235,4 @@ T run( runLock.unlock(); } } - - @SuppressWarnings("ThreadPriorityCheck") - private void phaseTransition() { - long deadline = System.nanoTime() + maxSpinWaitNanos; - while (appendersInFlight.get() != 0) { - if (System.nanoTime() - deadline >= 0) { - // A late appender uses the active-bit decision and generation identity, so proceeding is - // safe even if a suspended appender did not reach its stripe in time. - return; - } - Thread.yield(); - } - } } diff --git a/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/BufferTest.java b/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/BufferTest.java index 279f112e7..3093110d3 100644 --- a/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/BufferTest.java +++ b/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/BufferTest.java @@ -179,51 +179,6 @@ void interruptedAppenderLeavesBoundedBufferWait() throws InterruptedException { assertThat(interrupted).isTrue(); } - @Test - void stalledAppenderAfterStripeActivationDoesNotBlockRun() throws InterruptedException { - CountDownLatch started = new CountDownLatch(1); - CountDownLatch proceed = new CountDownLatch(1); - CountDownLatch stalled = new CountDownLatch(1); - CountDownLatch release = new CountDownLatch(1); - AtomicReference appended = new AtomicReference<>(); - Buffer buffer = - new Buffer( - TimeUnit.SECONDS.toNanos(1), - 16, - () -> { - stalled.countDown(); - try { - release.await(); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } - }); - Thread runner = - new Thread( - () -> - buffer.run( - ignored -> { - started.countDown(); - return proceed.getCount() == 0; - }, - () -> new CounterSnapshot.CounterDataPointSnapshot(0, Labels.EMPTY, null, 0), - ignored -> {}), - "buffer-stalled-runner"); - runner.setDaemon(true); - runner.start(); - assertThat(started.await(5, TimeUnit.SECONDS)).isTrue(); - Thread appender = new Thread(() -> appended.set(buffer.append(1.0)), "buffer-stalled-appender"); - appender.setDaemon(true); - appender.start(); - assertThat(stalled.await(5, TimeUnit.SECONDS)).isTrue(); - proceed.countDown(); - runner.join(5_000); - assertThat(runner.isAlive()).isFalse(); - release.countDown(); - appender.join(5_000); - assertThat(appended).hasValue(false); - } - @Test void lateAppenderCannotBeAddedToTheNextGeneration() throws InterruptedException { CountDownLatch firstRunStarted = new CountDownLatch(1); @@ -287,7 +242,7 @@ void lateAppenderCannotBeAddedToTheNextGeneration() throws InterruptedException buffer.run( expectedCount -> { secondRunStarted.countDown(); - return completedObservations.get() >= expectedCount; + return completedObservations.get() == expectedCount; }, () -> new CounterSnapshot.CounterDataPointSnapshot(0, Labels.EMPTY, null, 0), ignored -> {}),