hadoop-common-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From cnaur...@apache.org
Subject git commit: HADOOP-11105. MetricsSystemImpl could leak memory in registered callbacks. Contributed by Chuan Liu.
Date Thu, 18 Sep 2014 22:38:08 GMT
Repository: hadoop
Updated Branches:
  refs/heads/trunk 2c3da25fd -> 1942364ef


HADOOP-11105. MetricsSystemImpl could leak memory in registered callbacks. Contributed by
Chuan Liu.


Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/1942364e
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/1942364e
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/1942364e

Branch: refs/heads/trunk
Commit: 1942364ef14396e9bd94a87c0d901ff9abe1d42a
Parents: 2c3da25
Author: cnauroth <cnauroth@apache.org>
Authored: Thu Sep 18 15:36:43 2014 -0700
Committer: cnauroth <cnauroth@apache.org>
Committed: Thu Sep 18 15:36:43 2014 -0700

----------------------------------------------------------------------
 .../hadoop/metrics2/impl/MetricsSystemImpl.java | 35 ++++++++++++++++----
 1 file changed, 28 insertions(+), 7 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hadoop/blob/1942364e/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/impl/MetricsSystemImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/impl/MetricsSystemImpl.java
b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/impl/MetricsSystemImpl.java
index 722abd9..2107e68 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/impl/MetricsSystemImpl.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/impl/MetricsSystemImpl.java
@@ -83,7 +83,12 @@ public class MetricsSystemImpl extends MetricsSystem implements MetricsSource
{
   private final Map<String, MetricsSource> allSources;
   private final Map<String, MetricsSinkAdapter> sinks;
   private final Map<String, MetricsSink> allSinks;
+
+  // The callback list is used by register(Callback callback), while
+  // the callback map is used by register(String name, String desc, T sink)
   private final List<Callback> callbacks;
+  private final Map<String, Callback> namedCallbacks;
+
   private final MetricsCollectorImpl collector;
   private final MetricsRegistry registry = new MetricsRegistry(MS_NAME);
   @Metric({"Snapshot", "Snapshot stats"}) MutableStat snapshotStat;
@@ -119,6 +124,7 @@ public class MetricsSystemImpl extends MetricsSystem implements MetricsSource
{
     sourceConfigs = Maps.newHashMap();
     sinkConfigs = Maps.newHashMap();
     callbacks = Lists.newArrayList();
+    namedCallbacks = Maps.newHashMap();
     injectedTags = Lists.newArrayList();
     collector = new MetricsCollectorImpl();
     if (prefix != null) {
@@ -178,11 +184,13 @@ public class MetricsSystemImpl extends MetricsSystem implements MetricsSource
{
       return;
     }
     for (Callback cb : callbacks) cb.preStart();
+    for (Callback cb : namedCallbacks.values()) cb.preStart();
     configure(prefix);
     startTimer();
     monitoring = true;
     LOG.info(prefix +" metrics system started");
     for (Callback cb : callbacks) cb.postStart();
+    for (Callback cb : namedCallbacks.values()) cb.postStart();
   }
 
   @Override
@@ -198,6 +206,7 @@ public class MetricsSystemImpl extends MetricsSystem implements MetricsSource
{
       return;
     }
     for (Callback cb : callbacks) cb.preStop();
+    for (Callback cb : namedCallbacks.values()) cb.preStop();
     LOG.info("Stopping "+ prefix +" metrics system...");
     stopTimer();
     stopSources();
@@ -206,6 +215,7 @@ public class MetricsSystemImpl extends MetricsSystem implements MetricsSource
{
     monitoring = false;
     LOG.info(prefix +" metrics system stopped.");
     for (Callback cb : callbacks) cb.postStop();
+    for (Callback cb : namedCallbacks.values()) cb.postStop();
   }
 
   @Override public synchronized <T>
@@ -224,7 +234,7 @@ public class MetricsSystemImpl extends MetricsSystem implements MetricsSource
{
     }
     // We want to re-register the source to pick up new config when the
     // metrics system restarts.
-    register(new AbstractCallback() {
+    register(name, new AbstractCallback() {
       @Override public void postStart() {
         registerSource(finalName, finalDesc, s);
       }
@@ -241,6 +251,9 @@ public class MetricsSystemImpl extends MetricsSystem implements MetricsSource
{
     if (allSources.containsKey(name)) {
       allSources.remove(name);
     }
+    if (namedCallbacks.containsKey(name)) {
+      namedCallbacks.remove(name);
+    }
   }
 
   synchronized
@@ -268,7 +281,7 @@ public class MetricsSystemImpl extends MetricsSystem implements MetricsSource
{
     }
     // We want to re-register the sink to pick up new config
     // when the metrics system restarts.
-    register(new AbstractCallback() {
+    register(name, new AbstractCallback() {
       @Override public void postStart() {
         register(name, description, sink);
       }
@@ -289,9 +302,16 @@ public class MetricsSystemImpl extends MetricsSystem implements MetricsSource
{
 
   @Override
   public synchronized void register(final Callback callback) {
-    callbacks.add((Callback) Proxy.newProxyInstance(
-        callback.getClass().getClassLoader(), new Class<?>[] { Callback.class },
-        new InvocationHandler() {
+    callbacks.add((Callback) getProxyForCallback(callback));
+  }
+
+  private synchronized void register(String name, final Callback callback) {
+    namedCallbacks.put(name, (Callback) getProxyForCallback(callback));
+  }
+
+  private Object getProxyForCallback(final Callback callback) {
+    return Proxy.newProxyInstance(callback.getClass().getClassLoader(),
+        new Class<?>[] { Callback.class }, new InvocationHandler() {
           @Override
           public Object invoke(Object proxy, Method method, Object[] args)
               throws Throwable {
@@ -299,11 +319,11 @@ public class MetricsSystemImpl extends MetricsSystem implements MetricsSource
{
               return method.invoke(callback, args);
             } catch (Exception e) {
               // These are not considered fatal.
-              LOG.warn("Caught exception in callback "+ method.getName(), e);
+              LOG.warn("Caught exception in callback " + method.getName(), e);
             }
             return null;
           }
-        }));
+        });
   }
 
   @Override
@@ -572,6 +592,7 @@ public class MetricsSystemImpl extends MetricsSystem implements MetricsSource
{
     allSources.clear();
     allSinks.clear();
     callbacks.clear();
+    namedCallbacks.clear();
     if (mbeanName != null) {
       MBeans.unregister(mbeanName);
       mbeanName = null;


Mime
View raw message