You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by ka...@apache.org on 2016/11/11 00:10:51 UTC
[24/50] [abbrv] hadoop git commit: HADOOP-13782. Make MutableRates
metrics thread-local write, aggregate-on-read. Contributed by Erik Krogen.
HADOOP-13782. Make MutableRates metrics thread-local write, aggregate-on-read. Contributed by Erik Krogen.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/77c13c38
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/77c13c38
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/77c13c38
Branch: refs/heads/YARN-4752
Commit: 77c13c385774c51766fe505397fa916754ac08d4
Parents: 2a65eb1
Author: Zhe Zhang <zh...@apache.org>
Authored: Tue Nov 8 16:07:36 2016 -0800
Committer: Zhe Zhang <zh...@apache.org>
Committed: Tue Nov 8 16:07:36 2016 -0800
----------------------------------------------------------------------
.../hadoop/ipc/metrics/RpcDetailedMetrics.java | 4 +-
.../metrics2/lib/MutableMetricsFactory.java | 5 +
.../hadoop/metrics2/lib/MutableRates.java | 6 +
.../lib/MutableRatesWithAggregation.java | 148 +++++++++++++++++++
.../hadoop/metrics2/lib/TestMutableMetrics.java | 148 ++++++++++++++++++-
5 files changed, 306 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/77c13c38/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/metrics/RpcDetailedMetrics.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/metrics/RpcDetailedMetrics.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/metrics/RpcDetailedMetrics.java
index 7414364..ad36742 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/metrics/RpcDetailedMetrics.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/metrics/RpcDetailedMetrics.java
@@ -24,7 +24,7 @@ import org.apache.hadoop.metrics2.annotation.Metric;
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.MutableRates;
+import org.apache.hadoop.metrics2.lib.MutableRatesWithAggregation;
/**
* This class is for maintaining RPC method related statistics
@@ -34,7 +34,7 @@ import org.apache.hadoop.metrics2.lib.MutableRates;
@Metrics(about="Per method RPC metrics", context="rpcdetailed")
public class RpcDetailedMetrics {
- @Metric MutableRates rates;
+ @Metric MutableRatesWithAggregation rates;
static final Log LOG = LogFactory.getLog(RpcDetailedMetrics.class);
final MetricsRegistry registry;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/77c13c38/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/lib/MutableMetricsFactory.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/lib/MutableMetricsFactory.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/lib/MutableMetricsFactory.java
index ae2cb4f..b926c4b 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/lib/MutableMetricsFactory.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/lib/MutableMetricsFactory.java
@@ -67,6 +67,11 @@ public class MutableMetricsFactory {
if (cls == MutableRates.class) {
return new MutableRates(registry);
}
+ if (cls == MutableRatesWithAggregation.class) {
+ MutableRatesWithAggregation rates = new MutableRatesWithAggregation();
+ registry.add(info.name(), rates);
+ return rates;
+ }
if (cls == MutableStat.class) {
return registry.newStat(info.name(), info.description(),
annotation.sampleName(), annotation.valueName(),
http://git-wip-us.apache.org/repos/asf/hadoop/blob/77c13c38/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/lib/MutableRates.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/lib/MutableRates.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/lib/MutableRates.java
index 121c292..1074e87 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/lib/MutableRates.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/lib/MutableRates.java
@@ -33,6 +33,12 @@ import org.apache.hadoop.metrics2.MetricsRecordBuilder;
/**
* Helper class to manage a group of mutable rate metrics
+ *
+ * This class synchronizes all accesses to the metrics it
+ * contains, so it should not be used in situations where
+ * there is high contention on the metrics.
+ * {@link MutableRatesWithAggregation} is preferable in that
+ * situation.
*/
@InterfaceAudience.Public
@InterfaceStability.Evolving
http://git-wip-us.apache.org/repos/asf/hadoop/blob/77c13c38/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/lib/MutableRatesWithAggregation.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/lib/MutableRatesWithAggregation.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/lib/MutableRatesWithAggregation.java
new file mode 100644
index 0000000..64eae03
--- /dev/null
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/lib/MutableRatesWithAggregation.java
@@ -0,0 +1,148 @@
+/**
+ * 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.metrics2.lib;
+
+import com.google.common.collect.Sets;
+import java.lang.ref.WeakReference;
+import java.lang.reflect.Method;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedDeque;
+import java.util.concurrent.ConcurrentMap;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.metrics2.MetricsRecordBuilder;
+import org.apache.hadoop.metrics2.util.SampleStat;
+
+
+/**
+ * Helper class to manage a group of mutable rate metrics.
+ *
+ * Each thread will maintain a local rate count, and upon snapshot,
+ * these values will be aggregated into a global rate. This class
+ * should only be used for long running threads, as any metrics
+ * produced between the last snapshot and the death of a thread
+ * will be lost. This allows for significantly higher concurrency
+ * than {@link MutableRates}. See HADOOP-24420.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+public class MutableRatesWithAggregation extends MutableMetric {
+ static final Log LOG = LogFactory.getLog(MutableRatesWithAggregation.class);
+ private final Map<String, MutableRate> globalMetrics = new HashMap<>();
+ private final Set<Class<?>> protocolCache = Sets.newHashSet();
+
+ private final ConcurrentLinkedDeque<WeakReference<ConcurrentMap<String, ThreadSafeSampleStat>>>
+ weakReferenceQueue = new ConcurrentLinkedDeque<>();
+ private final ThreadLocal<ConcurrentMap<String, ThreadSafeSampleStat>>
+ threadLocalMetricsMap = new ThreadLocal<>();
+
+ /**
+ * Initialize the registry with all the methods in a protocol
+ * so they all show up in the first snapshot.
+ * Convenient for JMX implementations.
+ * @param protocol the protocol class
+ */
+ public void init(Class<?> protocol) {
+ if (protocolCache.contains(protocol)) {
+ return;
+ }
+ protocolCache.add(protocol);
+ for (Method method : protocol.getDeclaredMethods()) {
+ String name = method.getName();
+ LOG.debug(name);
+ addMetricIfNotExists(name);
+ }
+ }
+
+ /**
+ * Add a rate sample for a rate metric.
+ * @param name of the rate metric
+ * @param elapsed time
+ */
+ public void add(String name, long elapsed) {
+ ConcurrentMap<String, ThreadSafeSampleStat> localStats =
+ threadLocalMetricsMap.get();
+ if (localStats == null) {
+ localStats = new ConcurrentHashMap<>();
+ threadLocalMetricsMap.set(localStats);
+ weakReferenceQueue.add(new WeakReference<>(localStats));
+ }
+ ThreadSafeSampleStat stat = localStats.get(name);
+ if (stat == null) {
+ stat = new ThreadSafeSampleStat();
+ localStats.put(name, stat);
+ }
+ stat.add(elapsed);
+ }
+
+ @Override
+ public synchronized void snapshot(MetricsRecordBuilder rb, boolean all) {
+ Iterator<WeakReference<ConcurrentMap<String, ThreadSafeSampleStat>>> iter =
+ weakReferenceQueue.iterator();
+ while (iter.hasNext()) {
+ ConcurrentMap<String, ThreadSafeSampleStat> map = iter.next().get();
+ if (map == null) {
+ // Thread has died; clean up its state
+ iter.remove();
+ } else {
+ // Aggregate the thread's local samples into the global metrics
+ for (Map.Entry<String, ThreadSafeSampleStat> entry : map.entrySet()) {
+ String name = entry.getKey();
+ MutableRate globalMetric = addMetricIfNotExists(name);
+ entry.getValue().snapshotInto(globalMetric);
+ }
+ }
+ }
+ for (MutableRate globalMetric : globalMetrics.values()) {
+ globalMetric.snapshot(rb, all);
+ }
+ }
+
+ private synchronized MutableRate addMetricIfNotExists(String name) {
+ MutableRate metric = globalMetrics.get(name);
+ if (metric == null) {
+ metric = new MutableRate(name, name, false);
+ globalMetrics.put(name, metric);
+ }
+ return metric;
+ }
+
+ private static class ThreadSafeSampleStat {
+
+ private SampleStat stat = new SampleStat();
+
+ synchronized void add(double x) {
+ stat.add(x);
+ }
+
+ synchronized void snapshotInto(MutableRate metric) {
+ if (stat.numSamples() > 0) {
+ metric.add(stat.numSamples(), Math.round(stat.total()));
+ stat.reset();
+ }
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/77c13c38/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/metrics2/lib/TestMutableMetrics.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/metrics2/lib/TestMutableMetrics.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/metrics2/lib/TestMutableMetrics.java
index 9161df5..1faa361 100644
--- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/metrics2/lib/TestMutableMetrics.java
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/metrics2/lib/TestMutableMetrics.java
@@ -19,9 +19,7 @@
package org.apache.hadoop.metrics2.lib;
import static org.apache.hadoop.metrics2.lib.Interns.info;
-import static org.apache.hadoop.test.MetricsAsserts.assertCounter;
-import static org.apache.hadoop.test.MetricsAsserts.assertGauge;
-import static org.apache.hadoop.test.MetricsAsserts.mockMetricsRecordBuilder;
+import static org.apache.hadoop.test.MetricsAsserts.*;
import static org.mockito.AdditionalMatchers.eq;
import static org.mockito.AdditionalMatchers.geq;
import static org.mockito.AdditionalMatchers.leq;
@@ -29,10 +27,15 @@ import static org.mockito.Matchers.anyLong;
import static org.mockito.Matchers.eq;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
+import static org.junit.Assert.*;
import java.util.Map;
import java.util.Map.Entry;
+import java.util.Random;
+import java.util.concurrent.CountDownLatch;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.metrics2.MetricsRecordBuilder;
import org.apache.hadoop.metrics2.util.Quantile;
import org.junit.Test;
@@ -42,6 +45,7 @@ import org.junit.Test;
*/
public class TestMutableMetrics {
+ private static final Log LOG = LogFactory.getLog(TestMutableMetrics.class);
private final double EPSILON = 1e-42;
/**
@@ -129,6 +133,144 @@ public class TestMutableMetrics {
assertGauge("BarAvgTime", 0.0, rb);
}
+ @Test public void testMutableRatesWithAggregationInit() {
+ MetricsRecordBuilder rb = mockMetricsRecordBuilder();
+ MutableRatesWithAggregation rates = new MutableRatesWithAggregation();
+
+ rates.init(TestProtocol.class);
+ rates.snapshot(rb, false);
+
+ assertCounter("FooNumOps", 0L, rb);
+ assertGauge("FooAvgTime", 0.0, rb);
+ assertCounter("BarNumOps", 0L, rb);
+ assertGauge("BarAvgTime", 0.0, rb);
+ }
+
+ @Test public void testMutableRatesWithAggregationSingleThread() {
+ MutableRatesWithAggregation rates = new MutableRatesWithAggregation();
+
+ rates.add("foo", 1);
+ rates.add("bar", 5);
+
+ MetricsRecordBuilder rb = mockMetricsRecordBuilder();
+ rates.snapshot(rb, false);
+ assertCounter("FooNumOps", 1L, rb);
+ assertGauge("FooAvgTime", 1.0, rb);
+ assertCounter("BarNumOps", 1L, rb);
+ assertGauge("BarAvgTime", 5.0, rb);
+
+ rates.add("foo", 1);
+ rates.add("foo", 3);
+ rates.add("bar", 6);
+
+ rb = mockMetricsRecordBuilder();
+ rates.snapshot(rb, false);
+ assertCounter("FooNumOps", 3L, rb);
+ assertGauge("FooAvgTime", 2.0, rb);
+ assertCounter("BarNumOps", 2L, rb);
+ assertGauge("BarAvgTime", 6.0, rb);
+ }
+
+ @Test public void testMutableRatesWithAggregationManyThreads()
+ throws InterruptedException {
+ final MutableRatesWithAggregation rates = new MutableRatesWithAggregation();
+
+ final int n = 10;
+ long[] opCount = new long[n];
+ double[] opTotalTime = new double[n];
+
+ for (int i = 0; i < n; i++) {
+ opCount[i] = 0;
+ opTotalTime[i] = 0;
+ // Initialize so that the getLongCounter() method doesn't complain
+ rates.add("metric" + i, 0);
+ }
+
+ Thread[] threads = new Thread[n];
+ final CountDownLatch firstAddsFinished = new CountDownLatch(threads.length);
+ final CountDownLatch firstSnapshotsFinished = new CountDownLatch(1);
+ final CountDownLatch secondAddsFinished =
+ new CountDownLatch(threads.length);
+ final CountDownLatch secondSnapshotsFinished = new CountDownLatch(1);
+ long seed = new Random().nextLong();
+ LOG.info("Random seed = " + seed);
+ final Random sleepRandom = new Random(seed);
+ for (int tIdx = 0; tIdx < threads.length; tIdx++) {
+ final int threadIdx = tIdx;
+ threads[threadIdx] = new Thread() {
+ @Override
+ public void run() {
+ try {
+ for (int i = 0; i < 1000; i++) {
+ rates.add("metric" + (i % n), (i / n) % 2 == 0 ? 1 : 2);
+ // Sleep so additions can be interleaved with snapshots
+ Thread.sleep(sleepRandom.nextInt(5));
+ }
+ firstAddsFinished.countDown();
+
+ // Make sure all threads stay alive long enough for the first
+ // snapshot to complete; else their metrics may be lost to GC
+ firstSnapshotsFinished.await();
+
+ // Let half the threads continue with more metrics and let half die
+ if (threadIdx % 2 == 0) {
+ for (int i = 0; i < 1000; i++) {
+ rates.add("metric" + (i % n), (i / n) % 2 == 0 ? 1 : 2);
+ }
+ secondAddsFinished.countDown();
+ secondSnapshotsFinished.await();
+ } else {
+ secondAddsFinished.countDown();
+ }
+ } catch (InterruptedException e) {
+ // Ignore
+ }
+ }
+ };
+ }
+ for (Thread t : threads) {
+ t.start();
+ }
+ // Snapshot concurrently with additions but aggregate the totals into
+ // opCount / opTotalTime
+ for (int i = 0; i < 100; i++) {
+ snapshotMutableRatesWithAggregation(rates, opCount, opTotalTime);
+ Thread.sleep(sleepRandom.nextInt(20));
+ }
+ firstAddsFinished.await();
+ // Final snapshot to grab any remaining metrics and then verify that
+ // the totals are as expected
+ snapshotMutableRatesWithAggregation(rates, opCount, opTotalTime);
+ for (int i = 0; i < n; i++) {
+ assertEquals("metric" + i + " count", 1001, opCount[i]);
+ assertEquals("metric" + i + " total", 1500, opTotalTime[i], 1.0);
+ }
+ firstSnapshotsFinished.countDown();
+
+ // After half of the threads die, ensure that the remaining ones still
+ // add metrics correctly and that snapshot occurs correctly
+ secondAddsFinished.await();
+ snapshotMutableRatesWithAggregation(rates, opCount, opTotalTime);
+ for (int i = 0; i < n; i++) {
+ assertEquals("metric" + i + " count", 1501, opCount[i]);
+ assertEquals("metric" + i + " total", 2250, opTotalTime[i], 1.0);
+ }
+ secondSnapshotsFinished.countDown();
+ }
+
+ private static void snapshotMutableRatesWithAggregation(
+ MutableRatesWithAggregation rates, long[] opCount, double[] opTotalTime) {
+ MetricsRecordBuilder rb = mockMetricsRecordBuilder();
+ rates.snapshot(rb, true);
+ for (int i = 0; i < opCount.length; i++) {
+ long prevOpCount = opCount[i];
+ long newOpCount = getLongCounter("Metric" + i + "NumOps", rb);
+ opCount[i] = newOpCount;
+ double avgTime = getDoubleGauge("Metric" + i + "AvgTime", rb);
+ opTotalTime[i] += avgTime * (newOpCount - prevOpCount);
+ }
+ }
+
/**
* Tests that when using {@link MutableStat#add(long, long)}, even with a high
* sample count, the mean does not lose accuracy.
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org