From 4142ddcdfcf8c07e26975f7526ed49986c0d4067 Mon Sep 17 00:00:00 2001 From: Arnab Nandy Date: Wed, 8 Jul 2026 16:13:38 +0530 Subject: [PATCH 1/3] #2287 fix: bound buffer spin wait during collection Signed-off-by: Arnab Nandy --- .../metrics/core/metrics/Buffer.java | 23 ++++++- .../metrics/core/metrics/BufferTest.java | 66 +++++++++++++++++++ 2 files changed, 87 insertions(+), 2 deletions(-) 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..7236971f8 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,7 +1,10 @@ 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; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.ReentrantLock; @@ -18,6 +21,7 @@ class Buffer { private static final long bufferActiveBit = 1L << 63; + private static final long DEFAULT_MAX_SPIN_WAIT_NANOS = TimeUnit.SECONDS.toNanos(1); // 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, @@ -34,8 +38,14 @@ class Buffer { ReentrantLock appendLock = new ReentrantLock(); ReentrantLock runLock = new ReentrantLock(); Condition bufferFilled = appendLock.newCondition(); + private final long maxSpinWaitNanos; Buffer() { + this(DEFAULT_MAX_SPIN_WAIT_NANOS); + } + + Buffer(long maxSpinWaitNanos) { + this.maxSpinWaitNanos = maxSpinWaitNanos; stripedObservationCounts = new AtomicLong[Runtime.getRuntime().availableProcessors()]; for (int i = 0; i < stripedObservationCounts.length; i++) { stripedObservationCounts[i] = new AtomicLong(0); @@ -82,6 +92,7 @@ T run( double[] buffer; int bufferSize; T result; + boolean timedOut = false; runLock.lock(); try { @@ -91,14 +102,19 @@ T run( expectedCount += observationCount.getAndAdd(bufferActiveBit); } + 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. + if (System.nanoTime() - deadline >= 0) { + timedOut = true; + break; + } Thread.yield(); } - result = createResult.get(); + result = timedOut ? null : createResult.get(); // Signal that the buffer is inactive. long expectedBufferSize = 0; @@ -137,6 +153,9 @@ T run( for (int i = 0; i < bufferSize; i++) { observeFunction.accept(buffer[i]); } - return result; + if (timedOut) { + throw new IllegalStateException("Timed out while waiting for in-flight observations."); + } + return requireNonNull(result); } } 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..869bb7822 --- /dev/null +++ b/prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/BufferTest.java @@ -0,0 +1,66 @@ +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 org.junit.jupiter.api.Test; + +class BufferTest { + + @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."); + } +} From 50a4eddace5133188581b2f253a654317fc0c2fb Mon Sep 17 00:00:00 2001 From: Arnab Nandy Date: Wed, 8 Jul 2026 16:33:48 +0530 Subject: [PATCH 2/3] #2287 fix: allow buffer maintenance callbacks to return null Signed-off-by: Arnab Nandy --- .../java/io/prometheus/metrics/core/metrics/Buffer.java | 6 ++---- 1 file changed, 2 insertions(+), 4 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 7236971f8..1a3c48b8c 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,7 +1,5 @@ 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; @@ -84,7 +82,7 @@ void reset() { reset = true; } - @SuppressWarnings("ThreadPriorityCheck") + @SuppressWarnings({"NullAway", "ThreadPriorityCheck"}) T run( Function complete, Supplier createResult, @@ -156,6 +154,6 @@ T run( if (timedOut) { throw new IllegalStateException("Timed out while waiting for in-flight observations."); } - return requireNonNull(result); + return result; } } From 09f04ed1e3ef48adf4ae551a2d6ea35f77a72cc8 Mon Sep 17 00:00:00 2001 From: Arnab Nandy Date: Wed, 8 Jul 2026 16:51:27 +0530 Subject: [PATCH 3/3] #2282 fix: avoid overflow in buffer stripe index Signed-off-by: Arnab Nandy --- .../java/io/prometheus/metrics/core/metrics/Buffer.java | 6 +++++- .../io/prometheus/metrics/core/metrics/BufferTest.java | 7 +++++++ 2 files changed, 12 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 1a3c48b8c..b3a52652c 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 @@ -51,7 +51,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) { @@ -62,6 +62,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 index 869bb7822..7f8e09dbf 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 @@ -14,6 +14,13 @@ 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); + } + @Test void timeoutDeactivatesBufferAndReplaysBufferedObservations() throws InterruptedException { Buffer buffer = new Buffer(TimeUnit.SECONDS.toNanos(1));