This is an automated email from the ASF dual-hosted git repository.
jojochuang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/master by this push:
new 21cdfda6081 HDDS-9377. Replace MutableStat with lock-free
ConcurrentMutableStat in OMLockMetrics and PerformanceMetrics (#11085)
21cdfda6081 is described below
commit 21cdfda6081c31893601b1985050f5d58259d243
Author: Andrey Yarovoy <[email protected]>
AuthorDate: Tue Aug 25 19:16:38 2026 +0300
HDDS-9377. Replace MutableStat with lock-free ConcurrentMutableStat in
OMLockMetrics and PerformanceMetrics (#11085)
Generated-by: Claude Opus 4.8
---
.../hadoop/ozone/util/ConcurrentMutableStat.java | 134 ++++++++++
.../hadoop/ozone/util/PerformanceMetrics.java | 5 +-
.../ozone/util/ConcurrentMutableStatBenchmark.java | 286 +++++++++++++++++++++
.../ozone/util/TestConcurrentMutableStat.java | 172 +++++++++++++
.../apache/hadoop/ozone/om/lock/OMLockMetrics.java | 29 +--
5 files changed, 607 insertions(+), 19 deletions(-)
diff --git
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/util/ConcurrentMutableStat.java
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/util/ConcurrentMutableStat.java
new file mode 100644
index 00000000000..fe7c512f473
--- /dev/null
+++
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/util/ConcurrentMutableStat.java
@@ -0,0 +1,134 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ozone.util;
+
+import java.util.concurrent.atomic.LongAccumulator;
+import java.util.concurrent.atomic.LongAdder;
+import org.apache.hadoop.metrics2.MetricsRecordBuilder;
+import org.apache.hadoop.metrics2.lib.MutableStat;
+import org.apache.hadoop.metrics2.util.SampleStat;
+
+/**
+ * A {@link MutableStat} whose hot-path {@link #add(long)} is non-blocking
+ * under concurrent callers. Each call accumulates in {@link LongAdder} /
+ * {@link LongAccumulator} cells and the pending totals are drained into the
+ * parent's running state lazily — only when metrics are read via
+ * {@link #snapshot}, {@link #lastStat}, or {@link #toString}.
+ *
+ * <p>This avoids the {@code synchronized} contention of the base class when
+ * many threads release locks simultaneously and all attempt to record a
+ * measurement on the same metric instance.
+ *
+ * <p><b>Standard deviation accuracy:</b> {@code drainPending()} batches all
+ * pending samples except the min and max into a single
+ * {@code super.add(n, sum)} call. {@link
org.apache.hadoop.metrics2.util.SampleStat}
+ * treats a batch as {@code n} identical samples equal to their mean, so the
+ * within-batch variance is lost. The reported standard deviation is therefore
+ * underestimated. Callers that need accurate standard deviation should use
+ * {@link org.apache.hadoop.metrics2.lib.MutableStat} directly, or pass
+ * {@code extended=false} to suppress the stdev metric.
+ */
+public class ConcurrentMutableStat extends MutableStat {
+
+ private final LongAdder pendingSum = new LongAdder();
+ private final LongAdder pendingCount = new LongAdder();
+ /** Cell-striped min accumulator: identity = Long.MAX_VALUE, function =
Math::min. */
+ private final LongAccumulator pendingMin = new LongAccumulator(Math::min,
Long.MAX_VALUE);
+ /** Cell-striped max accumulator: identity = Long.MIN_VALUE, function =
Math::max. */
+ private final LongAccumulator pendingMax = new LongAccumulator(Math::max,
Long.MIN_VALUE);
+
+ public ConcurrentMutableStat(String name, String description,
+ String sampleName, String valueName, boolean extended) {
+ super(name, description, sampleName, valueName, extended);
+ }
+
+ /**
+ * Accumulates {@code value} without acquiring any lock.
+ * The value is reflected in consumers on the next {@link #snapshot} call.
+ * {@code setChanged()} is deferred to {@link #drainPending()} to avoid
+ * concurrent volatile writes on the same field from all calling threads.
+ */
+ @Override
+ public void add(long value) {
+ pendingSum.add(value);
+ pendingCount.increment();
+ pendingMin.accumulate(value);
+ pendingMax.accumulate(value);
+ }
+
+ @Override
+ public synchronized void snapshot(MetricsRecordBuilder builder, boolean all)
{
+ drainPending();
+ super.snapshot(builder, all);
+ }
+
+ @Override
+ public SampleStat lastStat() {
+ drainPending();
+ return super.lastStat();
+ }
+
+ @Override
+ public String toString() {
+ drainPending();
+ return super.toString();
+ }
+
+ /**
+ * Moves accumulated pending samples into the parent stat under its lock.
+ *
+ * <p>Min and max are drained via individual {@code super.add()} calls so
+ * that {@code MutableStat.minMax} is kept correct. The remaining
+ * {@code n - 2} samples are batched for efficiency. In the rare case where
+ * a concurrent {@code add()} incremented the count before the min/max
+ * accumulators ran (sentinel identity values), the whole batch falls back to
+ * a bulk add; the actual min/max will be captured in the next drain.
+ *
+ * <p>Safe to call from unsynchronized contexts; {@code super.add(long)} is
+ * {@code synchronized(this)}, and Java intrinsic locks are reentrant so
+ * calling this from the already-locked {@link #snapshot} path is fine.
+ */
+ private void drainPending() {
+ long n = pendingCount.sumThenReset();
+ if (n == 0) {
+ return;
+ }
+ long sum = pendingSum.sumThenReset();
+
+ setChanged();
+
+ long min = pendingMin.getThenReset();
+ long max = pendingMax.getThenReset();
+
+ if (min == Long.MAX_VALUE || max == Long.MIN_VALUE) {
+ // Race: count was incremented before accumulate() ran. Fall back to
+ // bulk add; min/max for these items will be captured in the next drain.
+ super.add(n, sum);
+ return;
+ }
+
+ // Drain min and max individually so MutableStat.minMax is updated.
+ super.add(min);
+ if (n > 1) {
+ super.add(max);
+ if (n > 2) {
+ super.add(n - 2, sum - min - max);
+ }
+ }
+ }
+}
diff --git
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/util/PerformanceMetrics.java
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/util/PerformanceMetrics.java
index eaa17c0278f..c3ea31908d8 100644
---
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/util/PerformanceMetrics.java
+++
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/util/PerformanceMetrics.java
@@ -23,7 +23,6 @@
import org.apache.hadoop.metrics2.MetricsRecordBuilder;
import org.apache.hadoop.metrics2.lib.MetricsRegistry;
import org.apache.hadoop.metrics2.lib.MutableQuantiles;
-import org.apache.hadoop.metrics2.lib.MutableStat;
/**
* The {@code PerformanceMetrics} class encapsulates a collection of related
@@ -32,7 +31,7 @@
* snapshot their values for reporting.
*/
public class PerformanceMetrics implements Closeable {
- private final MutableStat stat;
+ private final ConcurrentMutableStat stat;
private final List<MutableQuantiles> quantiles;
private final MutableMinMax minMax;
@@ -70,7 +69,7 @@ public static synchronized <T> Map<String,
PerformanceMetrics> initializeMetrics
public PerformanceMetrics(
MetricsRegistry registry, String name, String description,
String sampleName, String valueName, int[] intervals) {
- stat = registry.newStat(name, description, sampleName, valueName, false);
+ stat = new ConcurrentMutableStat(name, description, sampleName, valueName,
false);
quantiles = MetricUtil.createQuantiles(registry, name, description,
sampleName, valueName, intervals);
minMax = new MutableMinMax(registry, name, description, valueName);
}
diff --git
a/hadoop-hdds/common/src/test/java/org/apache/hadoop/ozone/util/ConcurrentMutableStatBenchmark.java
b/hadoop-hdds/common/src/test/java/org/apache/hadoop/ozone/util/ConcurrentMutableStatBenchmark.java
new file mode 100644
index 00000000000..e7417e1a647
--- /dev/null
+++
b/hadoop-hdds/common/src/test/java/org/apache/hadoop/ozone/util/ConcurrentMutableStatBenchmark.java
@@ -0,0 +1,286 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ozone.util;
+
+import java.util.concurrent.BrokenBarrierException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.LongAdder;
+import org.apache.hadoop.metrics2.lib.MutableStat;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Scalability benchmark comparing {@link MutableStat} and
+ * {@link ConcurrentMutableStat} under two load patterns that model the
+ * thundering-herd observed when many threads release an
+ * {@code OzoneManagerLock} read-lock at the same time.
+ *
+ * <p><b>Burst</b>: all N threads release simultaneously via a
+ * {@link java.util.concurrent.CyclicBarrier}. Reports wall time for the full
+ * burst to complete. {@code MutableStat} serialises the work — burst time
+ * scales linearly with N. {@code ConcurrentMutableStat} uses cell-striped
+ * accumulators — burst time scales sub-linearly.
+ *
+ * <p><b>Steady-state</b>: all N threads call {@code add()} in a tight loop for
+ * a fixed window. Reports total ops/ms. {@code MutableStat} throughput is
+ * capped by the serialised mutex regardless of thread count.
+ * {@code ConcurrentMutableStat} throughput grows with N.
+ *
+ * <p>Each measurement method uses its own concrete stat type so the JIT can
+ * devirtualize {@code add()} monomorphically and eliminate dispatch overhead.
+ *
+ * <p>Run with:
+ * <pre>
+ * mvn test -pl :hdds-common \
+ * -Dtest=ConcurrentMutableStatBenchmark \
+ * -Dgroups=benchmark -Dexcluded-test-groups= \
+ * -Dsurefire.failIfNoSpecifiedTests=false
+ * </pre>
+ */
+@Tag("benchmark")
+public class ConcurrentMutableStatBenchmark {
+
+ private static final int[] THREAD_COUNTS = {1, 10, 20, 40, 60, 80};
+
+ // Burst scenario
+ /** Number of {@code add()} calls each thread makes per burst. */
+ private static final int ADDS_PER_BURST = 500;
+ private static final int WARMUP_BURSTS = 200;
+ private static final int MEASURE_BURSTS = 300;
+
+ // Steady-state scenario
+ /** Warm-up window per thread before the measured window starts. */
+ private static final int STEADY_WARMUP_MS = 200;
+ /** Measurement window per thread for steady-state throughput. */
+ private static final int STEADY_MEASURE_MS = 500;
+ /** Ops per {@code System.nanoTime()} poll to amortise timer-call overhead.
*/
+ private static final int STEADY_BATCH = 1_000;
+
+ @Test
+ public void benchmarkBurstScalability() throws Exception {
+ System.out.println();
+ System.out.printf("%-9s %-24s %-32s %s%n",
+ "Threads", "MutableStat µs/burst", "ConcurrentMutableStat µs/burst",
"Speedup");
+ System.out.println(
+
"--------------------------------------------------------------------------------------------");
+
+ for (int threads : THREAD_COUNTS) {
+ double baseUs = measureMutableStatBurst(threads);
+ double concUs = measureConcurrentMutableStatBurst(threads);
+ double speedup = baseUs > 0 ? baseUs / concUs : Double.NaN;
+ System.out.printf("%-9d %-24s %-32s %.1fx%n",
+ threads, formatUs(baseUs), formatUs(concUs), speedup);
+ }
+ }
+
+ private static double measureMutableStatBurst(int threads) throws Exception {
+ MutableStat stat = new MutableStat("base", "baseline", "Ops", "Time",
false);
+ ExecutorService pool = Executors.newFixedThreadPool(threads);
+ CyclicBarrier barrier = new CyclicBarrier(threads + 1);
+
+ for (int t = 0; t < threads; t++) {
+ pool.submit(() -> {
+ try {
+ while (!Thread.currentThread().isInterrupted()) {
+ barrier.await();
+ for (int i = 0; i < ADDS_PER_BURST; i++) {
+ stat.add(ThreadLocalRandom.current().nextLong(1, 10_000));
+ }
+ barrier.await();
+ }
+ } catch (InterruptedException | BrokenBarrierException e) {
+ Thread.currentThread().interrupt();
+ }
+ });
+ }
+
+ runBursts(barrier, WARMUP_BURSTS);
+ long ns = runBursts(barrier, MEASURE_BURSTS);
+ pool.shutdownNow();
+ pool.awaitTermination(5, TimeUnit.SECONDS);
+ return (double) ns / MEASURE_BURSTS / 1000.0;
+ }
+
+ private static double measureConcurrentMutableStatBurst(int threads) throws
Exception {
+ ConcurrentMutableStat stat =
+ new ConcurrentMutableStat("conc", "concurrent", "Ops", "Time", false);
+ ExecutorService pool = Executors.newFixedThreadPool(threads);
+ CyclicBarrier barrier = new CyclicBarrier(threads + 1);
+
+ for (int t = 0; t < threads; t++) {
+ pool.submit(() -> {
+ try {
+ while (!Thread.currentThread().isInterrupted()) {
+ barrier.await();
+ for (int i = 0; i < ADDS_PER_BURST; i++) {
+ stat.add(ThreadLocalRandom.current().nextLong(1, 10_000));
+ }
+ barrier.await();
+ }
+ } catch (InterruptedException | BrokenBarrierException e) {
+ Thread.currentThread().interrupt();
+ }
+ });
+ }
+
+ runBursts(barrier, WARMUP_BURSTS);
+ long ns = runBursts(barrier, MEASURE_BURSTS);
+ pool.shutdownNow();
+ pool.awaitTermination(5, TimeUnit.SECONDS);
+ return (double) ns / MEASURE_BURSTS / 1000.0;
+ }
+
+ @Test
+ public void benchmarkSteadyStateThroughput() throws Exception {
+ System.out.println();
+ System.out.printf("=== Steady-State (continuous load, %d ms window)
===%n", STEADY_MEASURE_MS);
+ System.out.printf("%-9s %-24s %-32s %s%n",
+ "Threads", "MutableStat ops/ms", "ConcurrentMutableStat ops/ms",
"Speedup");
+ System.out.println(
+
"--------------------------------------------------------------------------------------------");
+
+ for (int threads : THREAD_COUNTS) {
+ double baseOps = measureMutableStatSteady(threads);
+ double concOps = measureConcurrentMutableStatSteady(threads);
+ double speedup = baseOps > 0 ? concOps / baseOps : Double.NaN;
+ System.out.printf("%-9d %-24s %-32s %.1fx%n",
+ threads, formatOpsMs(baseOps), formatOpsMs(concOps), speedup);
+ }
+ }
+
+ private static double measureMutableStatSteady(int threads) throws Exception
{
+ MutableStat stat = new MutableStat("base", "baseline", "Ops", "Time",
false);
+ LongAdder totalOps = new LongAdder();
+ CountDownLatch start = new CountDownLatch(1);
+ CountDownLatch done = new CountDownLatch(threads);
+ ExecutorService pool = Executors.newFixedThreadPool(threads);
+ for (int t = 0; t < threads; t++) {
+ pool.submit(() -> {
+ try {
+ start.await();
+ ThreadLocalRandom rng = ThreadLocalRandom.current();
+ long warmupEnd = System.nanoTime() + (long) STEADY_WARMUP_MS *
1_000_000L;
+ long measureEnd = warmupEnd + (long) STEADY_MEASURE_MS *
1_000_000L;
+ while (System.nanoTime() < warmupEnd) {
+ for (int i = 0; i < STEADY_BATCH; i++) {
+ stat.add(rng.nextLong(1, 10_000));
+ }
+ }
+ long ops = 0;
+ while (System.nanoTime() < measureEnd) {
+ for (int i = 0; i < STEADY_BATCH; i++) {
+ stat.add(rng.nextLong(1, 10_000));
+ }
+ ops += STEADY_BATCH;
+ }
+ totalOps.add(ops);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } finally {
+ done.countDown();
+ }
+ });
+ }
+ start.countDown();
+ if (!done.await(60, TimeUnit.SECONDS)) {
+ throw new AssertionError("steady-state threads did not finish within 60
s");
+ }
+ pool.shutdownNow();
+ pool.awaitTermination(5, TimeUnit.SECONDS);
+ return (double) totalOps.sum() / STEADY_MEASURE_MS;
+ }
+
+ private static double measureConcurrentMutableStatSteady(int threads) throws
Exception {
+ ConcurrentMutableStat stat =
+ new ConcurrentMutableStat("conc", "concurrent", "Ops", "Time", false);
+ LongAdder totalOps = new LongAdder();
+ CountDownLatch start = new CountDownLatch(1);
+ CountDownLatch done = new CountDownLatch(threads);
+ ExecutorService pool = Executors.newFixedThreadPool(threads);
+ for (int t = 0; t < threads; t++) {
+ pool.submit(() -> {
+ try {
+ start.await();
+ ThreadLocalRandom rng = ThreadLocalRandom.current();
+ long warmupEnd = System.nanoTime() + (long) STEADY_WARMUP_MS *
1_000_000L;
+ long measureEnd = warmupEnd + (long) STEADY_MEASURE_MS *
1_000_000L;
+ while (System.nanoTime() < warmupEnd) {
+ for (int i = 0; i < STEADY_BATCH; i++) {
+ stat.add(rng.nextLong(1, 10_000));
+ }
+ }
+ long ops = 0;
+ while (System.nanoTime() < measureEnd) {
+ for (int i = 0; i < STEADY_BATCH; i++) {
+ stat.add(rng.nextLong(1, 10_000));
+ }
+ ops += STEADY_BATCH;
+ }
+ totalOps.add(ops);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } finally {
+ done.countDown();
+ }
+ });
+ }
+ start.countDown();
+ if (!done.await(60, TimeUnit.SECONDS)) {
+ throw new AssertionError("steady-state threads did not finish within 60
s");
+ }
+ pool.shutdownNow();
+ pool.awaitTermination(5, TimeUnit.SECONDS);
+ return (double) totalOps.sum() / STEADY_MEASURE_MS;
+ }
+
+ private static String formatOpsMs(double opsPerMs) {
+ if (opsPerMs >= 1_000_000) {
+ return String.format("%.2f G ops/ms", opsPerMs / 1_000_000);
+ } else if (opsPerMs >= 1_000) {
+ return String.format("%.2f k ops/ms", opsPerMs / 1_000);
+ }
+ return String.format("%.2f ops/ms", opsPerMs);
+ }
+
+ /**
+ * Runs {@code bursts} rounds through the driver side of the barrier and
+ * returns total elapsed nanoseconds.
+ */
+ private static long runBursts(CyclicBarrier barrier, int bursts) throws
Exception {
+ long total = 0;
+ for (int i = 0; i < bursts; i++) {
+ long t0 = System.nanoTime();
+ barrier.await();
+ barrier.await();
+ total += System.nanoTime() - t0;
+ }
+ return total;
+ }
+
+ private static String formatUs(double us) {
+ if (us >= 1000.0) {
+ return String.format("%.2f ms", us / 1000.0);
+ }
+ return String.format("%.2f µs", us);
+ }
+}
diff --git
a/hadoop-hdds/common/src/test/java/org/apache/hadoop/ozone/util/TestConcurrentMutableStat.java
b/hadoop-hdds/common/src/test/java/org/apache/hadoop/ozone/util/TestConcurrentMutableStat.java
new file mode 100644
index 00000000000..3ce9239af09
--- /dev/null
+++
b/hadoop-hdds/common/src/test/java/org/apache/hadoop/ozone/util/TestConcurrentMutableStat.java
@@ -0,0 +1,172 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ozone.util;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import org.apache.hadoop.metrics2.util.SampleStat;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Unit tests for {@link ConcurrentMutableStat}.
+ */
+public class TestConcurrentMutableStat {
+
+ @Test
+ public void testSingleThreadedCountAndMean() {
+ ConcurrentMutableStat stat = newStat();
+ for (int i = 1; i <= 10; i++) {
+ stat.add(i); // sum=55, count=10
+ }
+ SampleStat last = stat.lastStat();
+ assertEquals(10, last.numSamples());
+ assertEquals(5.5, last.mean(), 0.001);
+ }
+
+ @Test
+ public void testSingleThreadedMinMax() {
+ ConcurrentMutableStat stat = newStat();
+ stat.add(5);
+ stat.add(1);
+ stat.add(10);
+ stat.add(3);
+ SampleStat last = stat.lastStat();
+ assertEquals(4, last.numSamples());
+ assertEquals(1.0, last.min(), 0.001);
+ assertEquals(10.0, last.max(), 0.001);
+ }
+
+ @Test
+ public void testSingleValueMinEqualsMax() {
+ ConcurrentMutableStat stat = newStat();
+ stat.add(42);
+ SampleStat last = stat.lastStat();
+ assertEquals(1, last.numSamples());
+ assertEquals(42.0, last.min(), 0.001);
+ assertEquals(42.0, last.max(), 0.001);
+ }
+
+ @Test
+ public void testToStringContainsSampleCount() {
+ ConcurrentMutableStat stat = newStat();
+ for (int i = 0; i < 7; i++) {
+ stat.add(100);
+ }
+ assertThat(stat.toString()).contains("Samples = 7");
+ }
+
+ @Test
+ public void testConcurrentAddCount() throws InterruptedException {
+ ConcurrentMutableStat stat = newStat();
+ int threads = 14;
+ int addsPerThread = 1000;
+
+ CountDownLatch start = new CountDownLatch(1);
+ CountDownLatch done = new CountDownLatch(threads);
+ ExecutorService pool = Executors.newFixedThreadPool(threads);
+
+ for (int t = 0; t < threads; t++) {
+ pool.submit(() -> {
+ try {
+ start.await();
+ for (int i = 0; i < addsPerThread; i++) {
+ stat.add(1);
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } finally {
+ done.countDown();
+ }
+ });
+ }
+ start.countDown();
+ assertTrue(done.await(30, TimeUnit.SECONDS));
+ pool.shutdown();
+
+ assertEquals((long) threads * addsPerThread, stat.lastStat().numSamples());
+ }
+
+ @Test
+ public void testConcurrentAddMaxValue() throws InterruptedException {
+ ConcurrentMutableStat stat = newStat();
+ int threads = 14;
+ CountDownLatch done = new CountDownLatch(threads);
+ ExecutorService pool = Executors.newFixedThreadPool(threads);
+
+ for (int t = 0; t < threads; t++) {
+ long value = (t + 1) * 10L; // 10, 20, ..., 140
+ pool.submit(() -> {
+ stat.add(value);
+ done.countDown();
+ });
+ }
+ assertTrue(done.await(10, TimeUnit.SECONDS));
+ pool.shutdown();
+
+ SampleStat last = stat.lastStat();
+ assertEquals(threads, last.numSamples());
+ assertEquals(140.0, last.max(), 0.001);
+ }
+
+ @Test
+ public void testConcurrentAddMinValue() throws InterruptedException {
+ ConcurrentMutableStat stat = newStat();
+ int threads = 14;
+ CountDownLatch done = new CountDownLatch(threads);
+ ExecutorService pool = Executors.newFixedThreadPool(threads);
+
+ for (int t = 0; t < threads; t++) {
+ long value = (t + 1) * 10L; // 10, 20, ..., 140
+ pool.submit(() -> {
+ stat.add(value);
+ done.countDown();
+ });
+ }
+ assertTrue(done.await(10, TimeUnit.SECONDS));
+ pool.shutdown();
+
+ assertEquals(10.0, stat.lastStat().min(), 0.001);
+ }
+
+ @Test
+ public void testMultipleAddsAndDrains() {
+ ConcurrentMutableStat stat = newStat();
+
+ stat.add(1);
+ stat.add(3);
+ // first drain via toString; interval not reset by toString, samples
accumulate
+ assertThat(stat.toString()).contains("Samples = 2");
+
+ stat.add(5);
+ stat.add(7);
+ // interval accumulates until snapshot(); all four samples visible
+ assertThat(stat.toString()).contains("Samples = 4");
+ assertEquals(4.0, stat.lastStat().mean(), 0.001);
+ assertEquals(7.0, stat.lastStat().max(), 0.001);
+ }
+
+ private static ConcurrentMutableStat newStat() {
+ return new ConcurrentMutableStat("test", "test stat", "Ops", "Time", true);
+ }
+}
diff --git
a/hadoop-ozone/interface-storage/src/main/java/org/apache/hadoop/ozone/om/lock/OMLockMetrics.java
b/hadoop-ozone/interface-storage/src/main/java/org/apache/hadoop/ozone/om/lock/OMLockMetrics.java
index 1541819e9a9..0805d533bdd 100644
---
a/hadoop-ozone/interface-storage/src/main/java/org/apache/hadoop/ozone/om/lock/OMLockMetrics.java
+++
b/hadoop-ozone/interface-storage/src/main/java/org/apache/hadoop/ozone/om/lock/OMLockMetrics.java
@@ -24,9 +24,8 @@
import org.apache.hadoop.metrics2.MetricsSystem;
import org.apache.hadoop.metrics2.annotation.Metrics;
import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
-import org.apache.hadoop.metrics2.lib.MetricsRegistry;
-import org.apache.hadoop.metrics2.lib.MutableStat;
import org.apache.hadoop.ozone.OzoneConsts;
+import org.apache.hadoop.ozone.util.ConcurrentMutableStat;
/**
* This class is for maintaining the various Ozone Manager Lock Metrics.
@@ -37,26 +36,24 @@ public final class OMLockMetrics implements MetricsSource {
private static final String SOURCE_NAME =
OMLockMetrics.class.getSimpleName();
- private final MetricsRegistry registry;
- private final MutableStat readLockWaitingTimeMsStat;
- private final MutableStat readLockHeldTimeMsStat;
- private final MutableStat writeLockWaitingTimeMsStat;
- private final MutableStat writeLockHeldTimeMsStat;
+ private final ConcurrentMutableStat readLockWaitingTimeMsStat;
+ private final ConcurrentMutableStat readLockHeldTimeMsStat;
+ private final ConcurrentMutableStat writeLockWaitingTimeMsStat;
+ private final ConcurrentMutableStat writeLockHeldTimeMsStat;
private OMLockMetrics() {
- registry = new MetricsRegistry(SOURCE_NAME);
- readLockWaitingTimeMsStat = registry.newStat("ReadLockWaitingTime",
+ readLockWaitingTimeMsStat = new
ConcurrentMutableStat("ReadLockWaitingTime",
"Time (in milliseconds) spent waiting for acquiring the read lock",
- "Ops", "Time", true);
- readLockHeldTimeMsStat = registry.newStat("ReadLockHeldTime",
+ "Ops", "Time", false);
+ readLockHeldTimeMsStat = new ConcurrentMutableStat("ReadLockHeldTime",
"Time (in milliseconds) spent holding the read lock",
- "Ops", "Time", true);
- writeLockWaitingTimeMsStat = registry.newStat("WriteLockWaitingTime",
+ "Ops", "Time", false);
+ writeLockWaitingTimeMsStat = new
ConcurrentMutableStat("WriteLockWaitingTime",
"Time (in milliseconds) spent waiting for acquiring the write lock",
- "Ops", "Time", true);
- writeLockHeldTimeMsStat = registry.newStat("WriteLockHeldTime",
+ "Ops", "Time", false);
+ writeLockHeldTimeMsStat = new ConcurrentMutableStat("WriteLockHeldTime",
"Time (in milliseconds) spent holding the write lock",
- "Ops", "Time", true);
+ "Ops", "Time", false);
}
/**
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
