You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@giraph.apache.org by ni...@apache.org on 2012/11/20 02:07:53 UTC
svn commit: r1411493 [1/2] - in /giraph/trunk: ./
giraph/src/main/java/org/apache/giraph/
giraph/src/main/java/org/apache/giraph/comm/netty/
giraph/src/main/java/org/apache/giraph/comm/netty/handler/
giraph/src/main/java/org/apache/giraph/graph/ giraph...
Author: nitay
Date: Tue Nov 20 01:07:49 2012
New Revision: 1411493
URL: http://svn.apache.org/viewvc?rev=1411493&view=rev
Log:
GIRAPH-421: Aggregate metrics up to master
Added:
giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/AggregatedMetric.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/AggregatedMetrics.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/GiraphTimer.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/GiraphTimerContext.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/LongAndTimeUnit.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/MetricsRegistryDebugger.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/ValueWithHostname.java
- copied, changed from r1411478, giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/ValueGauge.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/WorkerSuperstepMetrics.java
Removed:
giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/MetricGroup.java
Modified:
giraph/trunk/.reviewboardrc
giraph/trunk/giraph/src/main/java/org/apache/giraph/GiraphConfiguration.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/ByteCounter.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/NettyWorkerClientRequestProcessor.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/handler/RequestDecoder.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/handler/RequestEncoder.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/handler/RequestInfo.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/handler/RequestServerHandler.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/BspService.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/BspServiceMaster.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/BspServiceWorker.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/ComputeCallable.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/EdgeInputSplitsCallable.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/GraphMapper.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/InputSplitsCallable.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/VertexInputSplitsCallable.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/GiraphMetrics.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/GiraphMetricsRegistry.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/SuperstepMetricsRegistry.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/ValueGauge.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/utils/ByteArrayVertexIdMessageCollection.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/utils/FakeTime.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/utils/MemoryUtils.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/utils/SystemTime.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/utils/Time.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/utils/Times.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/utils/WritableUtils.java
giraph/trunk/giraph/src/main/java/org/apache/giraph/zk/PredicateLock.java
giraph/trunk/giraph/src/test/java/org/apache/giraph/graph/TestMutableVertex.java
giraph/trunk/giraph/src/test/java/org/apache/giraph/utils/BspUtilsTest.java
Modified: giraph/trunk/.reviewboardrc
URL: http://svn.apache.org/viewvc/giraph/trunk/.reviewboardrc?rev=1411493&r1=1411492&r2=1411493&view=diff
==============================================================================
--- giraph/trunk/.reviewboardrc (original)
+++ giraph/trunk/.reviewboardrc Tue Nov 20 01:07:49 2012
@@ -1,3 +1,3 @@
-GUESS_FIELDS = True
+GUESS_SUMMARY = True
REVIEWBOARD_URL = "https://reviews.apache.org/"
TARGET_GROUPS = "giraph"
Modified: giraph/trunk/giraph/src/main/java/org/apache/giraph/GiraphConfiguration.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/GiraphConfiguration.java?rev=1411493&r1=1411492&r2=1411493&view=diff
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/GiraphConfiguration.java (original)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/GiraphConfiguration.java Tue Nov 20 01:07:49 2012
@@ -914,21 +914,12 @@ public class GiraphConfiguration extends
}
/**
- * Should we dump metrics at the end of the job.
+ * Whether to track, print, and aggregate metrics.
*
- * @return true if we should dump metrics, false otherwise.
+ * @return true if metrics are enabled, false otherwise (default)
*/
- public boolean dumpMetricsAtEnd() {
- return getBoolean(METRICS_DUMP_AT_END, false);
- }
-
- /**
- * Should we print superstep metrics at end of superstep.
- *
- * @return true if we should print metrics, false otherwise.
- */
- public boolean printSuperstepMetrics() {
- return getBoolean(METRICS_SUPERSTEP_PRINT, false);
+ public boolean metricsEnabled() {
+ return getBoolean(METRICS_ENABLE, false);
}
/**
Modified: giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/ByteCounter.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/ByteCounter.java?rev=1411493&r1=1411492&r2=1411493&view=diff
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/ByteCounter.java (original)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/ByteCounter.java Tue Nov 20 01:07:49 2012
@@ -19,7 +19,6 @@
package org.apache.giraph.comm.netty;
import org.apache.giraph.metrics.GiraphMetrics;
-import org.apache.giraph.metrics.MetricGroup;
import org.apache.giraph.metrics.ResetSuperstepMetricsObserver;
import org.apache.giraph.metrics.SuperstepMetricsRegistry;
import org.apache.giraph.utils.SystemTime;
@@ -55,7 +54,7 @@ public class ByteCounter extends SimpleC
private static final Logger LOG =
Logger.getLogger(ByteCounter.class);
/** Class timer */
- private static final Time TIME = SystemTime.getInstance();
+ private static final Time TIME = SystemTime.get();
/** All bytes ever sent */
private final AtomicLong bytesSent = new AtomicLong();
/** Total sent requests */
@@ -87,14 +86,12 @@ public class ByteCounter extends SimpleC
@Override
public void newSuperstep(SuperstepMetricsRegistry superstepMetrics) {
- sentRequestsMeter = superstepMetrics.getMeter(MetricGroup.NETWORK,
- "sent-requests", "requests", TimeUnit.SECONDS);
- sentBytesHist = superstepMetrics.getHistogram(MetricGroup.NETWORK,
- "sent-bytes", false);
- receivedRequestsMeter = superstepMetrics.getMeter(MetricGroup.NETWORK,
- "received-requests", "request", TimeUnit.SECONDS);
- receivedBytesHist = superstepMetrics.getHistogram(MetricGroup.NETWORK,
- "received-bytes", false);
+ sentRequestsMeter = superstepMetrics.getMeter("sent-requests", "requests",
+ TimeUnit.SECONDS);
+ sentBytesHist = superstepMetrics.getHistogram("sent-bytes", false);
+ receivedRequestsMeter = superstepMetrics.getMeter("received-requests",
+ "request", TimeUnit.SECONDS);
+ receivedBytesHist = superstepMetrics.getHistogram("received-bytes", false);
}
@Override
Modified: giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/NettyWorkerClientRequestProcessor.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/NettyWorkerClientRequestProcessor.java?rev=1411493&r1=1411492&r2=1411493&view=diff
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/NettyWorkerClientRequestProcessor.java (original)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/NettyWorkerClientRequestProcessor.java Tue Nov 20 01:07:49 2012
@@ -44,7 +44,6 @@ import org.apache.giraph.graph.WorkerInf
import org.apache.giraph.graph.partition.Partition;
import org.apache.giraph.graph.partition.PartitionOwner;
import org.apache.giraph.metrics.GiraphMetrics;
-import org.apache.giraph.metrics.MetricGroup;
import org.apache.giraph.metrics.ValueGauge;
import org.apache.giraph.utils.ByteArrayVertexIdMessageCollection;
import org.apache.giraph.utils.PairList;
@@ -125,8 +124,7 @@ public class NettyWorkerClientRequestPro
// Per-Superstep Metrics.
// Since this object is not long lived we just initialize the metrics here.
GiraphMetrics gmr = GiraphMetrics.getInstance();
- msgsSentInSuperstep = new ValueGauge<Long>(gmr.perSuperstep(),
- MetricGroup.NETWORK, "msgs-sent");
+ msgsSentInSuperstep = new ValueGauge<Long>(gmr.perSuperstep(), "msgs-sent");
}
@Override
Modified: giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/handler/RequestDecoder.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/handler/RequestDecoder.java?rev=1411493&r1=1411492&r2=1411493&view=diff
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/handler/RequestDecoder.java (original)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/handler/RequestDecoder.java Tue Nov 20 01:07:49 2012
@@ -41,7 +41,7 @@ public class RequestDecoder extends OneT
private static final Logger LOG =
Logger.getLogger(RequestDecoder.class);
/** Time class to use */
- private static final Time TIME = SystemTime.getInstance();
+ private static final Time TIME = SystemTime.get();
/** Configuration */
private final ImmutableClassesGiraphConfiguration conf;
/** Byte counter to output */
Modified: giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/handler/RequestEncoder.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/handler/RequestEncoder.java?rev=1411493&r1=1411492&r2=1411493&view=diff
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/handler/RequestEncoder.java (original)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/handler/RequestEncoder.java Tue Nov 20 01:07:49 2012
@@ -35,7 +35,7 @@ import org.jboss.netty.handler.codec.one
*/
public class RequestEncoder extends OneToOneEncoder {
/** Time class to use */
- private static final Time TIME = SystemTime.getInstance();
+ private static final Time TIME = SystemTime.get();
/** Class logger */
private static final Logger LOG = Logger.getLogger(RequestEncoder.class);
/** Holds the place of the message length until known */
Modified: giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/handler/RequestInfo.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/handler/RequestInfo.java?rev=1411493&r1=1411492&r2=1411493&view=diff
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/handler/RequestInfo.java (original)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/handler/RequestInfo.java Tue Nov 20 01:07:49 2012
@@ -31,7 +31,7 @@ import java.util.Date;
*/
public class RequestInfo {
/** Time class to use */
- private static final Time TIME = SystemTime.getInstance();
+ private static final Time TIME = SystemTime.get();
/** Destination of the request */
private final InetSocketAddress destinationAddress;
/** When the request was started */
Modified: giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/handler/RequestServerHandler.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/handler/RequestServerHandler.java?rev=1411493&r1=1411492&r2=1411493&view=diff
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/handler/RequestServerHandler.java (original)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/comm/netty/handler/RequestServerHandler.java Tue Nov 20 01:07:49 2012
@@ -44,7 +44,7 @@ public abstract class RequestServerHandl
/** Number of bytes in the encoded response */
public static final int RESPONSE_BYTES = 13;
/** Time class to use */
- private static Time TIME = SystemTime.getInstance();
+ private static Time TIME = SystemTime.get();
/** Class logger */
private static final Logger LOG =
Logger.getLogger(RequestServerHandler.class);
Modified: giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/BspService.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/BspService.java?rev=1411493&r1=1411492&r2=1411493&view=diff
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/BspService.java (original)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/BspService.java Tue Nov 20 01:07:49 2012
@@ -140,6 +140,9 @@ public abstract class BspService<I exten
"_partitionStatsKey";
/** JSON message count key */
public static final String JSONOBJ_NUM_MESSAGES_KEY = "_numMsgsKey";
+ /** JSON metrics key */
+ public static final String JSONOBJ_METRICS_KEY = "_metricsKey";
+
/** JSON state key */
public static final String JSONOBJ_STATE_KEY = "_stateKey";
/** JSON application attempt key */
Modified: giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/BspServiceMaster.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/BspServiceMaster.java?rev=1411493&r1=1411492&r2=1411493&view=diff
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/BspServiceMaster.java (original)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/BspServiceMaster.java Tue Nov 20 01:07:49 2012
@@ -18,6 +18,7 @@
package org.apache.giraph.graph;
+import org.apache.commons.io.FilenameUtils;
import org.apache.giraph.GiraphConfiguration;
import org.apache.giraph.ImmutableClassesGiraphConfiguration;
import org.apache.giraph.bsp.ApplicationState;
@@ -35,15 +36,16 @@ import org.apache.giraph.graph.partition
import org.apache.giraph.graph.partition.PartitionStats;
import org.apache.giraph.graph.partition.PartitionUtils;
import org.apache.giraph.master.MasterObserver;
+import org.apache.giraph.metrics.AggregatedMetrics;
import org.apache.giraph.metrics.GiraphMetrics;
-import org.apache.giraph.metrics.MetricGroup;
+import org.apache.giraph.metrics.GiraphTimer;
+import org.apache.giraph.metrics.GiraphTimerContext;
import org.apache.giraph.metrics.ResetSuperstepMetricsObserver;
import org.apache.giraph.metrics.SuperstepMetricsRegistry;
-import org.apache.giraph.metrics.ValueGauge;
+import org.apache.giraph.metrics.WorkerSuperstepMetrics;
import org.apache.giraph.utils.ProgressableUtils;
import org.apache.giraph.utils.SystemTime;
import org.apache.giraph.utils.Time;
-import org.apache.giraph.utils.Times;
import org.apache.giraph.utils.WritableUtils;
import org.apache.giraph.zk.BspEvent;
import org.apache.giraph.zk.PredicateLock;
@@ -88,6 +90,7 @@ import java.util.TreeSet;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
/**
* ZooKeeper-based implementation of {@link CentralizedServiceMaster}.
@@ -111,7 +114,7 @@ public class BspServiceMaster<I extends
/** Default number of threads to use when writing input splits to zookeeper */
public static final int DEFAULT_INPUT_SPLIT_THREAD_COUNT = 1;
/** Time instance to use for timing */
- private static final Time TIME = SystemTime.getInstance();
+ private static final Time TIME = SystemTime.get();
/** Class logger */
private static final Logger LOG = Logger.getLogger(BspServiceMaster.class);
/** Am I the master? */
@@ -157,8 +160,8 @@ public class BspServiceMaster<I extends
private final MasterObserver[] observers;
// Per-Superstep Metrics
- /** MasterCompute time in msec */
- private ValueGauge<Long> masterComputeMs;
+ /** MasterCompute time */
+ private GiraphTimer masterComputeTimer;
/**
* Constructor for setting up the master.
@@ -202,8 +205,8 @@ public class BspServiceMaster<I extends
@Override
public void newSuperstep(SuperstepMetricsRegistry superstepMetrics) {
- masterComputeMs = new ValueGauge<Long>(superstepMetrics, MetricGroup.USER,
- "master-compute-call");
+ masterComputeTimer = new GiraphTimer(superstepMetrics,
+ "master-compute-call", TimeUnit.MILLISECONDS);
}
@Override
@@ -835,6 +838,8 @@ public class BspServiceMaster<I extends
* @return Global statistics aggregated on all worker statistics
*/
private GlobalStats aggregateWorkerStats(long superstep) {
+ ImmutableClassesGiraphConfiguration conf = getConfiguration();
+
Class<? extends PartitionStats> partitionStatsClass =
masterGraphPartitioner.createPartitionStats().getClass();
GlobalStats globalStats = new GlobalStats();
@@ -854,8 +859,11 @@ public class BspServiceMaster<I extends
"aggregateWorkerStats: InterruptedException", e);
}
+ AggregatedMetrics aggregatedMetrics = new AggregatedMetrics();
+
allPartitionStatsList.clear();
for (String finishedPath : workerFinishedPathList) {
+ String hostnamePartitionId = FilenameUtils.getName(finishedPath);
JSONObject workerFinishedInfoObj = null;
try {
byte [] zkData =
@@ -866,7 +874,7 @@ public class BspServiceMaster<I extends
Base64.decode(workerFinishedInfoObj.getString(
JSONOBJ_PARTITION_STATS_KEY)),
partitionStatsClass,
- getConfiguration());
+ conf);
for (PartitionStats partitionStats : statsList) {
globalStats.addPartitionStats(partitionStats);
allPartitionStatsList.add(partitionStats);
@@ -874,6 +882,16 @@ public class BspServiceMaster<I extends
globalStats.addMessageCount(
workerFinishedInfoObj.getLong(
JSONOBJ_NUM_MESSAGES_KEY));
+ if (conf.metricsEnabled() &&
+ workerFinishedInfoObj.has(JSONOBJ_METRICS_KEY)) {
+ WorkerSuperstepMetrics workerMetrics = new WorkerSuperstepMetrics();
+ WritableUtils.readFieldsFromByteArray(
+ Base64.decode(
+ workerFinishedInfoObj.getString(
+ JSONOBJ_METRICS_KEY)),
+ workerMetrics);
+ aggregatedMetrics.add(workerMetrics, hostnamePartitionId);
+ }
} catch (JSONException e) {
throw new IllegalStateException(
"aggregateWorkerStats: JSONException", e);
@@ -889,6 +907,10 @@ public class BspServiceMaster<I extends
}
}
+ if (conf.metricsEnabled()) {
+ aggregatedMetrics.print(superstep);
+ }
+
if (LOG.isInfoEnabled()) {
LOG.info("aggregateWorkerStats: Aggregation found " + globalStats +
" on superstep = " + getSuperstep());
@@ -1494,9 +1516,9 @@ public class BspServiceMaster<I extends
"runMasterCompute: Failed in access", e);
}
}
- long masterComputeBeginMs = TIME.getMilliseconds();
+ GiraphTimerContext timerContext = masterComputeTimer.time();
masterCompute.compute();
- masterComputeMs.set(Times.getMsSince(TIME, masterComputeBeginMs));
+ timerContext.stop();
}
/**
Modified: giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/BspServiceWorker.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/BspServiceWorker.java?rev=1411493&r1=1411492&r2=1411493&view=diff
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/BspServiceWorker.java (original)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/BspServiceWorker.java Tue Nov 20 01:07:49 2012
@@ -38,16 +38,14 @@ import org.apache.giraph.graph.partition
import org.apache.giraph.graph.partition.PartitionStore;
import org.apache.giraph.graph.partition.WorkerGraphPartitioner;
import org.apache.giraph.metrics.GiraphMetrics;
-import org.apache.giraph.metrics.MetricGroup;
+import org.apache.giraph.metrics.GiraphTimer;
+import org.apache.giraph.metrics.GiraphTimerContext;
import org.apache.giraph.metrics.ResetSuperstepMetricsObserver;
import org.apache.giraph.metrics.SuperstepMetricsRegistry;
-import org.apache.giraph.metrics.ValueGauge;
+import org.apache.giraph.metrics.WorkerSuperstepMetrics;
import org.apache.giraph.utils.LoggerUtils;
import org.apache.giraph.utils.MemoryUtils;
import org.apache.giraph.utils.ProgressableUtils;
-import org.apache.giraph.utils.SystemTime;
-import org.apache.giraph.utils.Time;
-import org.apache.giraph.utils.Times;
import org.apache.giraph.utils.WritableUtils;
import org.apache.giraph.zk.BspEvent;
import org.apache.giraph.zk.PredicateLock;
@@ -89,6 +87,7 @@ import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
/**
* ZooKeeper-based implementation of {@link CentralizedServiceWorker}.
@@ -105,9 +104,7 @@ public class BspServiceWorker<I extends
implements CentralizedServiceWorker<I, V, E, M>,
ResetSuperstepMetricsObserver {
/** Name of gauge for time spent waiting on other workers */
- public static final String GAUGE_WAITING_TIME = "waiting-ms";
- /** Time instance to use for timing */
- private static final Time TIME = SystemTime.getInstance();
+ public static final String TIMER_WAIT_REQUESTS = "wait-requests-us";
/** Class logger */
private static final Logger LOG = Logger.getLogger(BspServiceWorker.class);
/** My process health znode */
@@ -144,10 +141,10 @@ public class BspServiceWorker<I extends
private final WorkerAggregatorHandler aggregatorHandler;
// Per-Superstep Metrics
- /** msec spent in WorkerContext#postSuperstep */
- private ValueGauge<Long> wcPostSuperstepMs;
- /** msec spent waiting for other workers */
- private ValueGauge<Long> waitMs;
+ /** Timer for WorkerContext#postSuperstep */
+ private GiraphTimer wcPostSuperstepTimer;
+ /** Time spent waiting on requests to finish */
+ private GiraphTimer waitRequestsTimer;
/**
* Constructor for setting up the worker.
@@ -191,10 +188,10 @@ public class BspServiceWorker<I extends
@Override
public void newSuperstep(SuperstepMetricsRegistry superstepMetrics) {
- waitMs = new ValueGauge<Long>(superstepMetrics, MetricGroup.NETWORK,
- GAUGE_WAITING_TIME);
- wcPostSuperstepMs = new ValueGauge<Long>(superstepMetrics,
- MetricGroup.USER, "worker-context-post-superstep-ms");
+ waitRequestsTimer = new GiraphTimer(superstepMetrics,
+ TIMER_WAIT_REQUESTS, TimeUnit.MICROSECONDS);
+ wcPostSuperstepTimer = new GiraphTimer(superstepMetrics,
+ "worker-context-post-superstep", TimeUnit.MICROSECONDS);
}
@Override
@@ -712,12 +709,7 @@ else[HADOOP_NON_SECURE]*/
// of this worker
// 5. Let the master know it is finished.
// 6. Wait for the master's global stats, and check if done
- if (LOG.isInfoEnabled()) {
- LOG.info("finishSuperstep: Waiting on all requests, superstep " +
- getSuperstep() + " " +
- MemoryUtils.getRuntimeMemoryStats());
- }
- workerClient.waitAllRequests();
+ waitForRequestsToFinish();
graphState.getGraphMapper().notifyFinishedCommunication();
@@ -728,9 +720,9 @@ else[HADOOP_NON_SECURE]*/
if (getSuperstep() != INPUT_SUPERSTEP) {
getWorkerContext().setGraphState(graphState);
- long postSuperstepBeginMs = TIME.getMilliseconds();
+ GiraphTimerContext timerContext = wcPostSuperstepTimer.time();
getWorkerContext().postSuperstep();
- wcPostSuperstepMs.set(Times.getMsSince(TIME, postSuperstepBeginMs));
+ timerContext.stop();
getContext().progress();
}
@@ -742,22 +734,105 @@ else[HADOOP_NON_SECURE]*/
MemoryUtils.getRuntimeMemoryStats());
}
+ writeFinshedSuperstepInfoToZK(partitionStatsList, workerSentMessages);
+
+ LoggerUtils.setStatusAndLog(getContext(), LOG, Level.INFO,
+ "finishSuperstep: (waiting for rest " +
+ "of workers) " +
+ getGraphMapper().getMapFunctions().toString() +
+ " - Attempt=" + getApplicationAttempt() +
+ ", Superstep=" + getSuperstep());
+
+ String superstepFinishedNode =
+ getSuperstepFinishedPath(getApplicationAttempt(), getSuperstep());
+
+ waitForOtherWorkers(superstepFinishedNode);
+
+ GlobalStats globalStats = new GlobalStats();
+ WritableUtils.readFieldsFromZnode(
+ getZkExt(), superstepFinishedNode, false, null, globalStats);
+ if (LOG.isInfoEnabled()) {
+ LOG.info("finishSuperstep: Completed superstep " + getSuperstep() +
+ " with global stats " + globalStats);
+ }
+ incrCachedSuperstep();
+ getContext().setStatus("finishSuperstep: (all workers done) " +
+ getGraphMapper().getMapFunctions().toString() +
+ " - Attempt=" + getApplicationAttempt() +
+ ", Superstep=" + getSuperstep());
+
+ return new FinishedSuperstepStats(
+ globalStats.getHaltComputation(),
+ globalStats.getVertexCount(),
+ globalStats.getEdgeCount());
+ }
+
+ /**
+ * Wait for all the requests to finish.
+ */
+ private void waitForRequestsToFinish() {
+ if (LOG.isInfoEnabled()) {
+ LOG.info("finishSuperstep: Waiting on all requests, superstep " +
+ getSuperstep() + " " +
+ MemoryUtils.getRuntimeMemoryStats());
+ }
+ GiraphTimerContext timerContext = waitRequestsTimer.time();
+ workerClient.waitAllRequests();
+ timerContext.stop();
+ }
+
+ /**
+ * Wait for all the other Workers to finish the superstep.
+ *
+ * @param superstepFinishedNode ZooKeeper path to wait on.
+ */
+ private void waitForOtherWorkers(String superstepFinishedNode) {
+ try {
+ while (getZkExt().exists(superstepFinishedNode, true) == null) {
+ getSuperstepFinishedEvent().waitForever();
+ getSuperstepFinishedEvent().reset();
+ }
+ } catch (KeeperException e) {
+ throw new IllegalStateException(
+ "finishSuperstep: Failed while waiting for master to " +
+ "signal completion of superstep " + getSuperstep(), e);
+ } catch (InterruptedException e) {
+ throw new IllegalStateException(
+ "finishSuperstep: Failed while waiting for master to " +
+ "signal completion of superstep " + getSuperstep(), e);
+ }
+ }
+
+ /**
+ * Write finished superstep info to ZooKeeper.
+ *
+ * @param partitionStatsList List of partition stats from superstep.
+ * @param workerSentMessages Number of messages sent in superstep.
+ */
+ private void writeFinshedSuperstepInfoToZK(
+ List<PartitionStats> partitionStatsList, long workerSentMessages) {
Collection<PartitionStats> finalizedPartitionStats =
workerGraphPartitioner.finalizePartitionStats(
partitionStatsList, getPartitionStore());
List<PartitionStats> finalizedPartitionStatsList =
new ArrayList<PartitionStats>(finalizedPartitionStats);
- byte [] partitionStatsBytes =
+ byte[] partitionStatsBytes =
WritableUtils.writeListToByteArray(finalizedPartitionStatsList);
+ WorkerSuperstepMetrics metrics = new WorkerSuperstepMetrics();
+ metrics.readFromRegistry();
+ byte[] metricsBytes = WritableUtils.writeToByteArray(metrics);
+
JSONObject workerFinishedInfoObj = new JSONObject();
try {
workerFinishedInfoObj.put(JSONOBJ_PARTITION_STATS_KEY,
Base64.encodeBytes(partitionStatsBytes));
- workerFinishedInfoObj.put(JSONOBJ_NUM_MESSAGES_KEY,
- workerSentMessages);
+ workerFinishedInfoObj.put(JSONOBJ_NUM_MESSAGES_KEY, workerSentMessages);
+ workerFinishedInfoObj.put(JSONOBJ_METRICS_KEY,
+ Base64.encodeBytes(metricsBytes));
} catch (JSONException e) {
throw new RuntimeException(e);
}
+
String finishedWorkerPath =
getWorkerFinishedPath(getApplicationAttempt(), getSuperstep()) +
"/" + getHostnamePartitionId();
@@ -777,52 +852,6 @@ else[HADOOP_NON_SECURE]*/
throw new IllegalStateException("Creating " + finishedWorkerPath +
" failed with InterruptedException", e);
}
-
- LoggerUtils.setStatusAndLog(getContext(), LOG, Level.INFO,
- "finishSuperstep: (waiting for rest " +
- "of workers) " +
- getGraphMapper().getMapFunctions().toString() +
- " - Attempt=" + getApplicationAttempt() +
- ", Superstep=" + getSuperstep());
-
- String superstepFinishedNode =
- getSuperstepFinishedPath(getApplicationAttempt(), getSuperstep());
-
- long waitBeginMs = TIME.getMilliseconds();
- try {
- while (getZkExt().exists(superstepFinishedNode, true) == null) {
- getSuperstepFinishedEvent().waitForever();
- getSuperstepFinishedEvent().reset();
- }
- } catch (KeeperException e) {
- throw new IllegalStateException(
- "finishSuperstep: Failed while waiting for master to " +
- "signal completion of superstep " + getSuperstep(), e);
- } catch (InterruptedException e) {
- throw new IllegalStateException(
- "finishSuperstep: Failed while waiting for master to " +
- "signal completion of superstep " + getSuperstep(), e);
- }
-
- waitMs.set(Times.getMsSince(TIME, waitBeginMs));
-
- GlobalStats globalStats = new GlobalStats();
- WritableUtils.readFieldsFromZnode(
- getZkExt(), superstepFinishedNode, false, null, globalStats);
- if (LOG.isInfoEnabled()) {
- LOG.info("finishSuperstep: Completed superstep " + getSuperstep() +
- " with global stats " + globalStats);
- }
- incrCachedSuperstep();
- getContext().setStatus("finishSuperstep: (all workers done) " +
- getGraphMapper().getMapFunctions().toString() +
- " - Attempt=" + getApplicationAttempt() +
- ", Superstep=" + getSuperstep());
-
- return new FinishedSuperstepStats(
- globalStats.getHaltComputation(),
- globalStats.getVertexCount(),
- globalStats.getEdgeCount());
}
/**
@@ -901,7 +930,7 @@ else[HADOOP_NON_SECURE]*/
LOG.error("cleanup: Zookeeper failed to close with " + e);
}
- if (getConfiguration().dumpMetricsAtEnd()) {
+ if (getConfiguration().metricsEnabled()) {
GiraphMetrics.getInstance().dumpToStdout();
}
Modified: giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/ComputeCallable.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/ComputeCallable.java?rev=1411493&r1=1411492&r2=1411493&view=diff
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/ComputeCallable.java (original)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/ComputeCallable.java Tue Nov 20 01:07:49 2012
@@ -28,7 +28,6 @@ import org.apache.giraph.comm.netty.Nett
import org.apache.giraph.graph.partition.Partition;
import org.apache.giraph.graph.partition.PartitionStats;
import org.apache.giraph.metrics.GiraphMetrics;
-import org.apache.giraph.metrics.MetricGroup;
import org.apache.giraph.utils.SystemTime;
import org.apache.giraph.utils.Time;
import org.apache.giraph.utils.Times;
@@ -65,7 +64,7 @@ public class ComputeCallable<I extends W
/** Class logger */
private static final Logger LOG = Logger.getLogger(ComputeCallable.class);
/** Class time object */
- private static final Time TIME = SystemTime.getInstance();
+ private static final Time TIME = SystemTime.get();
/** Context */
private final Mapper<?, ?, ?, ?>.Context context;
/** Graph state (note that it is recreated in call() for locality) */
@@ -117,8 +116,7 @@ public class ComputeCallable<I extends W
GiraphMetrics metrics = GiraphMetrics.getInstance();
// Normally we would use ResetSuperstepMetricsObserver but this class is
// not long-lived, so just instantiating in the constructor is good enough.
- computeOneTimer = metrics.perSuperstep().getTimer(MetricGroup.COMPUTE,
- TIMER_COMPUTE_ONE);
+ computeOneTimer = metrics.perSuperstep().getTimer(TIMER_COMPUTE_ONE);
}
@Override
Modified: giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/EdgeInputSplitsCallable.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/EdgeInputSplitsCallable.java?rev=1411493&r1=1411492&r2=1411493&view=diff
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/EdgeInputSplitsCallable.java (original)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/EdgeInputSplitsCallable.java Tue Nov 20 01:07:49 2012
@@ -21,7 +21,6 @@ package org.apache.giraph.graph;
import org.apache.giraph.ImmutableClassesGiraphConfiguration;
import org.apache.giraph.metrics.GiraphMetrics;
import org.apache.giraph.metrics.GiraphMetricsRegistry;
-import org.apache.giraph.metrics.MetricGroup;
import org.apache.giraph.utils.LoggerUtils;
import org.apache.giraph.utils.MemoryUtils;
import org.apache.giraph.zk.ZooKeeperExt;
@@ -91,8 +90,7 @@ public class EdgeInputSplitsCallable<I e
// Initialize Metrics
GiraphMetricsRegistry jobMetrics = GiraphMetrics.getInstance().perJob();
- edgesLoadedCounter = jobMetrics.getCounter(MetricGroup.IO,
- COUNTER_EDGES_LOADED);
+ edgesLoadedCounter = jobMetrics.getCounter(COUNTER_EDGES_LOADED);
}
/**
Modified: giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/GraphMapper.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/GraphMapper.java?rev=1411493&r1=1411492&r2=1411493&view=diff
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/GraphMapper.java (original)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/GraphMapper.java Tue Nov 20 01:07:49 2012
@@ -27,16 +27,15 @@ import org.apache.giraph.graph.partition
import org.apache.giraph.graph.partition.PartitionStats;
import org.apache.giraph.metrics.GiraphMetrics;
import org.apache.giraph.metrics.GiraphMetricsRegistry;
-import org.apache.giraph.metrics.MetricGroup;
+import org.apache.giraph.metrics.GiraphTimer;
+import org.apache.giraph.metrics.GiraphTimerContext;
import org.apache.giraph.metrics.ResetSuperstepMetricsObserver;
import org.apache.giraph.metrics.SuperstepMetricsRegistry;
-import org.apache.giraph.metrics.ValueGauge;
import org.apache.giraph.utils.MemoryUtils;
import org.apache.giraph.utils.ProgressableUtils;
import org.apache.giraph.utils.ReflectionUtils;
import org.apache.giraph.utils.SystemTime;
import org.apache.giraph.utils.Time;
-import org.apache.giraph.utils.Times;
import org.apache.giraph.zk.ZooKeeperManager;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.filecache.DistributedCache;
@@ -44,6 +43,7 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapreduce.Mapper;
+import org.apache.hadoop.util.Progressable;
import org.apache.log4j.Appender;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;
@@ -65,6 +65,7 @@ import java.util.concurrent.BlockingQueu
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
/**
* This mapper that will execute the BSP graph tasks. Since this mapper will
@@ -86,17 +87,17 @@ public class GraphMapper<I extends Writa
}
/** Name of metric for superstep time in msec */
- public static final String GAUGE_SUPERSTEP_TIME = "superstep-time-ms";
+ public static final String TIMER_SUPERSTEP_TIME = "superstep-time-ms";
/** Name of metric for compute on all vertices in msec */
- public static final String GAUGE_COMPUTE_ALL = "compute-all-ms";
+ public static final String TIMER_COMPUTE_ALL = "compute-all-ms";
/** Name of metric for time from begin compute to first message sent */
- public static final String GAUGE_TIME_TO_FIRST_MSG =
+ public static final String TIMER_TIME_TO_FIRST_MSG =
"time-to-first-message-ms";
/** Name of metric for time from first message till last message flushed */
- public static final String GAUGE_COMMUNICATION_TIME = "communication-time-ms";
+ public static final String TIMER_COMMUNICATION_TIME = "communication-time-ms";
/** Time instance used for timing in this class */
- private static final Time TIME = SystemTime.getInstance();
+ private static final Time TIME = SystemTime.get();
/** Class logger */
private static final Logger LOG = Logger.getLogger(GraphMapper.class);
/** Coordination service worker */
@@ -122,25 +123,25 @@ public class GraphMapper<I extends Writa
// Per-Job Metrics
/** Timer for WorkerContext#preApplication() */
- private ValueGauge<Long> wcPreAppMs;
- /** Timer in msec for WorkerContext#postApplication() */
- private ValueGauge<Long> wcPostAppMs;
+ private GiraphTimer wcPreAppTimer;
+ /** Timer for WorkerContext#postApplication() */
+ private GiraphTimer wcPostAppTimer;
// Per-Superstep Metrics
- /** Time in msec for how long superstep took */
- private ValueGauge<Long> superstepMs;
- /** Time in msec for all compute() calls in a superstep */
- private ValueGauge<Long> computeAllMs;
- /** Time in msec when computation started */
- private long computeAllBeginMs;
- /** Milliseconds from starting compute to sending first message */
- private ValueGauge<Long> msecToFirstMessage;
- /** Time in msec first message sent was at */
- private long firstMessageTimeMs;
+ /** Time for how long superstep took */
+ private GiraphTimer superstepTimer;
+ /** Time for all compute() calls in a superstep */
+ private GiraphTimer computeAll;
+ /** Time from starting compute to sending first message */
+ private GiraphTimer timeToFirstMessage;
+ /** Context for timing time to first message above */
+ private GiraphTimerContext timeToFirstMessageTimerContext;
/** Time from first sent message till last message flushed. */
- private ValueGauge<Long> communicationTimer;
- /** Timer in msec for WorkerContext#preSuperstep() */
- private ValueGauge<Long> wcPreSuperstepMs;
+ private GiraphTimer communicationTimer;
+ /** Context for timing communication time above */
+ private GiraphTimerContext communicationTimerContext;
+ /** Timer for WorkerContext#preSuperstep() */
+ private GiraphTimer wcPreSuperstepTimer;
/** What kinds of functions to run on this mapper */
public enum MapFunctions {
@@ -341,7 +342,7 @@ public class GraphMapper<I extends Writa
}
// Set up GiraphMetrics
- GiraphMetrics.init(context);
+ GiraphMetrics.init(conf);
GiraphMetrics.getInstance().addSuperstepResetObserver(this);
initJobMetrics();
MemoryUtils.initMetrics();
@@ -452,24 +453,24 @@ public class GraphMapper<I extends Writa
*/
private void initJobMetrics() {
GiraphMetricsRegistry jobMetrics = GiraphMetrics.getInstance().perJob();
- wcPreAppMs = new ValueGauge<Long>(jobMetrics, MetricGroup.USER,
- "worker-context-pre-app-ms");
- wcPostAppMs = new ValueGauge<Long>(jobMetrics, MetricGroup.USER,
- "worker-context-post-app-ms");
+ wcPreAppTimer = new GiraphTimer(jobMetrics, "worker-context-pre-app",
+ TimeUnit.MILLISECONDS);
+ wcPostAppTimer = new GiraphTimer(jobMetrics, "worker-context-post-app",
+ TimeUnit.MILLISECONDS);
}
@Override
public void newSuperstep(SuperstepMetricsRegistry superstepMetrics) {
- superstepMs = new ValueGauge<Long>(superstepMetrics, MetricGroup.COMPUTE,
- GAUGE_SUPERSTEP_TIME);
- computeAllMs = new ValueGauge<Long>(superstepMetrics, MetricGroup.COMPUTE,
- GAUGE_COMPUTE_ALL);
- msecToFirstMessage = new ValueGauge<Long>(superstepMetrics,
- MetricGroup.NETWORK, GAUGE_TIME_TO_FIRST_MSG);
- communicationTimer = new ValueGauge<Long>(superstepMetrics,
- MetricGroup.NETWORK, GAUGE_COMMUNICATION_TIME);
- wcPreSuperstepMs = new ValueGauge<Long>(superstepMetrics,
- MetricGroup.USER, "worker-context-pre-superstep-ms");
+ superstepTimer = new GiraphTimer(superstepMetrics,
+ TIMER_SUPERSTEP_TIME, TimeUnit.MILLISECONDS);
+ computeAll = new GiraphTimer(superstepMetrics,
+ TIMER_COMPUTE_ALL, TimeUnit.MILLISECONDS);
+ timeToFirstMessage = new GiraphTimer(superstepMetrics,
+ TIMER_TIME_TO_FIRST_MSG, TimeUnit.MICROSECONDS);
+ communicationTimer = new GiraphTimer(superstepMetrics,
+ TIMER_COMMUNICATION_TIME, TimeUnit.MILLISECONDS);
+ wcPreSuperstepTimer = new GiraphTimer(superstepMetrics,
+ "worker-context-pre-superstep", TimeUnit.MILLISECONDS);
}
/**
@@ -477,13 +478,14 @@ public class GraphMapper<I extends Writa
*/
public void notifySentMessages() {
// We are tracking the time between when the compute started and the first
- // message get sent. We use zero to flag that we have already recorded it.
- long tmp = firstMessageTimeMs;
- if (tmp == 0) {
- synchronized (msecToFirstMessage) {
- if (firstMessageTimeMs == 0) {
- firstMessageTimeMs = TIME.getMilliseconds();
- msecToFirstMessage.set(firstMessageTimeMs - computeAllBeginMs);
+ // message get sent. We use null to flag that we have already recorded it.
+ GiraphTimerContext tmp = timeToFirstMessageTimerContext;
+ if (tmp != null) {
+ synchronized (timeToFirstMessage) {
+ if (timeToFirstMessageTimerContext != null) {
+ timeToFirstMessageTimerContext.stop();
+ timeToFirstMessageTimerContext = null;
+ communicationTimerContext = communicationTimer.time();
}
}
}
@@ -494,10 +496,14 @@ public class GraphMapper<I extends Writa
* and are done waiting for all messages to send.
*/
public void notifyFinishedCommunication() {
- if (firstMessageTimeMs != 0) {
- long commTimeMs = Times.getMsSince(TIME, firstMessageTimeMs);
- communicationTimer.set(commTimeMs);
- firstMessageTimeMs = 0;
+ GiraphTimerContext tmp = communicationTimerContext;
+ if (tmp != null) {
+ synchronized (communicationTimer) {
+ if (communicationTimerContext != null) {
+ communicationTimerContext.stop();
+ communicationTimerContext = null;
+ }
+ }
}
}
@@ -546,20 +552,7 @@ public class GraphMapper<I extends Writa
new GraphState<I, V, E, M>(serviceWorker.getSuperstep(),
numVertices, numEdges, context, this, null, aggregatorUsage));
- long preAppBeginMs = TIME.getMilliseconds();
- try {
- serviceWorker.getWorkerContext().preApplication();
- } catch (InstantiationException e) {
- LOG.fatal("map: preApplication failed in instantiation", e);
- throw new RuntimeException(
- "map: preApplication failed in instantiation", e);
- } catch (IllegalAccessException e) {
- LOG.fatal("map: preApplication failed in access", e);
- throw new RuntimeException(
- "map: preApplication failed in access", e);
- }
- wcPreAppMs.set(Times.getMsSince(TIME, preAppBeginMs));
- context.progress();
+ workerContextPreApp(context);
List<PartitionStats> partitionStatsList =
new ArrayList<PartitionStats>();
@@ -570,7 +563,7 @@ public class GraphMapper<I extends Writa
final long superstep = serviceWorker.getSuperstep();
GiraphMetrics.getInstance().resetSuperstepMetrics(superstep);
- long superstepBeginMs = TIME.getMilliseconds();
+ GiraphTimerContext superstepTimerContext = superstepTimer.time();
GraphState<I, V, E, M> graphState =
new GraphState<I, V, E, M>(superstep, numVertices, numEdges,
@@ -607,9 +600,9 @@ public class GraphMapper<I extends Writa
serviceWorker.prepareSuperstep();
serviceWorker.getWorkerContext().setGraphState(graphState);
- long preSuperstepBeginMs = TIME.getMilliseconds();
+ GiraphTimerContext perSuperstepTimer = wcPreSuperstepTimer.time();
serviceWorker.getWorkerContext().preSuperstep();
- wcPreSuperstepMs.set(Times.getMsSince(TIME, preSuperstepBeginMs));
+ perSuperstepTimer.stop();
context.progress();
MessageStoreByPartition<I, M> messageStore =
@@ -633,7 +626,8 @@ public class GraphMapper<I extends Writa
computePartitionIdQueue.add(partitionId);
}
- computeAllBeginMs = TIME.getMilliseconds();
+ GiraphTimerContext computeAllTimerContext = computeAll.time();
+ timeToFirstMessageTimerContext = timeToFirstMessage.time();
ExecutorService partitionExecutor =
Executors.newFixedThreadPool(numThreads,
@@ -659,7 +653,7 @@ public class GraphMapper<I extends Writa
}
partitionExecutor.shutdown();
- computeAllMs.set(Times.getMsSince(TIME, computeAllBeginMs));
+ computeAllTimerContext.stop();
}
finishedSuperstepStats =
@@ -667,8 +661,8 @@ public class GraphMapper<I extends Writa
numVertices = finishedSuperstepStats.getVertexCount();
numEdges = finishedSuperstepStats.getEdgeCount();
- superstepMs.set(Times.getMsSince(TIME, superstepBeginMs));
- if (conf.printSuperstepMetrics()) {
+ superstepTimerContext.stop();
+ if (conf.metricsEnabled()) {
GiraphMetrics.getInstance().perSuperstep().printSummary();
}
@@ -680,12 +674,34 @@ public class GraphMapper<I extends Writa
serviceWorker.getWorkerContext().setGraphState(
new GraphState<I, V, E, M>(serviceWorker.getSuperstep(),
numVertices, numEdges, context, this, null, aggregatorUsage));
- long postAppBeginMs = TIME.getMilliseconds();
+ GiraphTimerContext postAppTimerContext = wcPostAppTimer.time();
serviceWorker.getWorkerContext().postApplication();
- wcPostAppMs.set(Times.getMsSince(TIME, postAppBeginMs));
+ postAppTimerContext.stop();
context.progress();
}
+ /**
+ * Call to the WorkerContext before application begins.
+ *
+ * @param progressable thing to call progress on.
+ */
+ private void workerContextPreApp(Progressable progressable) {
+ GiraphTimerContext preAppTimerContext = wcPreAppTimer.time();
+ try {
+ serviceWorker.getWorkerContext().preApplication();
+ } catch (InstantiationException e) {
+ LOG.fatal("map: preApplication failed in instantiation", e);
+ throw new RuntimeException(
+ "map: preApplication failed in instantiation", e);
+ } catch (IllegalAccessException e) {
+ LOG.fatal("map: preApplication failed in access", e);
+ throw new RuntimeException(
+ "map: preApplication failed in access", e);
+ }
+ preAppTimerContext.stop();
+ progressable.progress();
+ }
+
@Override
public void cleanup(Context context)
throws IOException, InterruptedException {
Modified: giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/InputSplitsCallable.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/InputSplitsCallable.java?rev=1411493&r1=1411492&r2=1411493&view=diff
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/InputSplitsCallable.java (original)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/InputSplitsCallable.java Tue Nov 20 01:07:49 2012
@@ -63,7 +63,7 @@ public abstract class InputSplitsCallabl
/** Class logger */
private static final Logger LOG = Logger.getLogger(InputSplitsCallable.class);
/** Class time object */
- private static final Time TIME = SystemTime.getInstance();
+ private static final Time TIME = SystemTime.get();
/** Configuration */
protected final ImmutableClassesGiraphConfiguration<I, V, E, M>
configuration;
Modified: giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/VertexInputSplitsCallable.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/VertexInputSplitsCallable.java?rev=1411493&r1=1411492&r2=1411493&view=diff
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/VertexInputSplitsCallable.java (original)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/graph/VertexInputSplitsCallable.java Tue Nov 20 01:07:49 2012
@@ -22,7 +22,6 @@ import org.apache.giraph.ImmutableClasse
import org.apache.giraph.graph.partition.PartitionOwner;
import org.apache.giraph.metrics.GiraphMetrics;
import org.apache.giraph.metrics.GiraphMetricsRegistry;
-import org.apache.giraph.metrics.MetricGroup;
import org.apache.giraph.utils.LoggerUtils;
import org.apache.giraph.utils.MemoryUtils;
import org.apache.giraph.zk.ZooKeeperExt;
@@ -99,10 +98,8 @@ public class VertexInputSplitsCallable<I
// Initialize Metrics
GiraphMetricsRegistry jobMetrics = GiraphMetrics.getInstance().perJob();
- verticesLoadedCounter = jobMetrics.getCounter(MetricGroup.IO,
- COUNTER_VERTICES_LOADED);
- edgesLoadedCounter = jobMetrics.getCounter(MetricGroup.IO,
- COUNTER_EDGES_LOADED);
+ verticesLoadedCounter = jobMetrics.getCounter(COUNTER_VERTICES_LOADED);
+ edgesLoadedCounter = jobMetrics.getCounter(COUNTER_EDGES_LOADED);
}
/**
Added: giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/AggregatedMetric.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/AggregatedMetric.java?rev=1411493&view=auto
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/AggregatedMetric.java (added)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/AggregatedMetric.java Tue Nov 20 01:07:49 2012
@@ -0,0 +1,112 @@
+/*
+ * 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.giraph.metrics;
+
+/**
+ * An aggregator over metrics from multiple hosts. Computes min, max, and mean.
+ */
+public class AggregatedMetric {
+ /** Minimum value seen with the host that it came from */
+ private ValueWithHostname min;
+ /** Maximum value seen with the host that it came from */
+ private ValueWithHostname max;
+ /** Total of all the values seen */
+ private long sum;
+ /** Number of values seen */
+ private long count;
+
+ /**
+ * Create new aggregated metric.
+ */
+ public AggregatedMetric() {
+ min = new ValueWithHostname(Long.MAX_VALUE);
+ max = new ValueWithHostname(Long.MIN_VALUE);
+ }
+
+ /**
+ * Add another item to the aggregation.
+ *
+ * @param value long value to add
+ * @param hostnamePartitionId String hostname it came from
+ */
+ public void addItem(long value, String hostnamePartitionId) {
+ if (value < min.getValue()) {
+ min.set(value, hostnamePartitionId);
+ }
+ if (value > max.getValue()) {
+ max.set(value, hostnamePartitionId);
+ }
+ sum += value;
+ count++;
+ }
+
+ /**
+ * Whether this AggregatedMetric has any data.
+ *
+ * @return true if we have any data.
+ */
+ public boolean hasData() {
+ return count > 0;
+ }
+
+ /**
+ * Get minimum value together with host it came from.
+ *
+ * @return ValueWithHostname for minimum
+ */
+ public ValueWithHostname min() {
+ return min;
+ }
+
+ /**
+ * Get maximum value together with host it came from.
+ *
+ * @return ValueWithHostname for maximum
+ */
+ public ValueWithHostname max() {
+ return max;
+ }
+
+ /**
+ * Get total of all the values seen
+ *
+ * @return long total of values seen
+ */
+ public long sum() {
+ return sum;
+ }
+
+ /**
+ * Get average of all the values
+ *
+ * @return computed average of all the values
+ */
+ public double mean() {
+ return sum / (double) count;
+ }
+
+ /**
+ * Get number of values seen
+ *
+ * @return count of values
+ */
+ public long count() {
+ return count;
+ }
+}
Added: giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/AggregatedMetrics.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/AggregatedMetrics.java?rev=1411493&view=auto
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/AggregatedMetrics.java (added)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/AggregatedMetrics.java Tue Nov 20 01:07:49 2012
@@ -0,0 +1,170 @@
+/*
+ * 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.giraph.metrics;
+
+import org.apache.giraph.graph.BspServiceWorker;
+import org.apache.giraph.graph.GraphMapper;
+
+import com.google.common.collect.Maps;
+
+import java.io.PrintStream;
+import java.util.Map;
+
+/**
+ * Map of a bunch of aggregated metrics
+ */
+public class AggregatedMetrics {
+ /** counter for user compute time in milliseconds */
+ public static final String USER_COMPUTE_MS = "user-compute-ms";
+
+ /** Mapping from name to aggregated metric */
+ private Map<String, AggregatedMetric> metrics = Maps.newHashMap();
+
+ /**
+ * Add value from hostname for a metric.
+ *
+ * @param name String name of metric
+ * @param value long value to track
+ * @param hostnamePartitionId String host it came from
+ * @return this
+ */
+ public AggregatedMetrics add(String name, long value,
+ String hostnamePartitionId) {
+ AggregatedMetric aggregatedMetric = metrics.get(name);
+ if (aggregatedMetric == null) {
+ aggregatedMetric = new AggregatedMetric();
+ metrics.put(name, aggregatedMetric);
+ }
+ aggregatedMetric.addItem(value, hostnamePartitionId);
+ return this;
+ }
+
+ /**
+ * Add metrics from worker.
+ *
+ * @param workerMetrics WorkerSuperstepMetrics from work
+ * @param hostname String hostname of worker
+ * @return this
+ */
+ public AggregatedMetrics add(WorkerSuperstepMetrics workerMetrics,
+ String hostname) {
+ add(GraphMapper.TIMER_COMMUNICATION_TIME,
+ workerMetrics.getCommTimer(), hostname);
+ add(GraphMapper.TIMER_COMPUTE_ALL,
+ workerMetrics.getComputeAllTimer(), hostname);
+ add(GraphMapper.TIMER_TIME_TO_FIRST_MSG,
+ workerMetrics.getTimeToFirstMsg(), hostname);
+ add(BspServiceWorker.TIMER_WAIT_REQUESTS,
+ workerMetrics.getWaitRequestsTimer(), hostname);
+ add(USER_COMPUTE_MS,
+ workerMetrics.getUserComputeTime(), hostname);
+ return this;
+ }
+
+ /**
+ * Print the aggregated metrics to the stream provided.
+ *
+ * @param superstep long number of superstep.
+ * @param out PrintStream to write to.
+ * @return this
+ */
+ public AggregatedMetrics print(long superstep, PrintStream out) {
+ AggregatedMetric commTime = get(GraphMapper.TIMER_COMMUNICATION_TIME);
+ AggregatedMetric computeAll = get(GraphMapper.TIMER_COMPUTE_ALL);
+ AggregatedMetric timeToFirstMsg =
+ get(GraphMapper.TIMER_TIME_TO_FIRST_MSG);
+ AggregatedMetric waitRequestsMicros = get(
+ BspServiceWorker.TIMER_WAIT_REQUESTS);
+ AggregatedMetric userComputeTime = get(USER_COMPUTE_MS);
+
+ out.println();
+ out.println("--- METRICS: superstep " + superstep + " ---");
+ printAggregatedMetric(out, "compute all partitions", "ms", computeAll);
+ printAggregatedMetric(out, "user compute time", "ms", userComputeTime);
+ printAggregatedMetric(out, "network communication time", "ms", commTime);
+ printAggregatedMetric(out, "time to first message", "us", timeToFirstMsg);
+ printAggregatedMetric(out, "wait requests time", "us", waitRequestsMicros);
+
+ return this;
+ }
+
+ /**
+ * Print the aggregated metrics to the stream provided.
+ *
+ * @param superstep long number of superstep.
+ * @return this
+ */
+ public AggregatedMetrics print(long superstep) {
+ return print(superstep, System.out);
+ }
+
+ /**
+ * Print batch of lines for AggregatedMetric
+ *
+ * @param out PrintStream to write to
+ * @param header String header to print.
+ * @param timeUnit String time unit of metric
+ * @param aggregatedMetric AggregatedMetric to write
+ */
+ private void printAggregatedMetric(PrintStream out, String header,
+ String timeUnit,
+ AggregatedMetric aggregatedMetric) {
+ if (aggregatedMetric.hasData()) {
+ out.println(header);
+ out.println(" mean: " + aggregatedMetric.mean() + " " + timeUnit);
+ printValueFromHost(out, " slowest: ", timeUnit, aggregatedMetric.max());
+ printValueFromHost(out, " fastest: ", timeUnit, aggregatedMetric.min());
+ } else {
+ out.println(header + ": NO DATA");
+ }
+ }
+
+ /**
+ * Print a line for a value with the host it came from.
+ *
+ * @param out PrintStream to write to
+ * @param prefix String to write at beginning
+ * @param timeUnit String timeUnit of metric
+ * @param vh ValueWithHostname to write
+ */
+ private void printValueFromHost(PrintStream out, String prefix,
+ String timeUnit, ValueWithHostname vh) {
+ out.println(prefix + vh.getValue() + ' ' + timeUnit +
+ " from " + vh.getHostname());
+ }
+
+ /**
+ * Get AggregatedMetric with given name.
+ *
+ * @param name String metric to lookup.
+ * @return AggregatedMetric for given metric name.
+ */
+ public AggregatedMetric get(String name) {
+ return metrics.get(name);
+ }
+
+ /**
+ * Get map of all aggregated metrics.
+ *
+ * @return Map of all the aggregated metrics.
+ */
+ public Map<String, AggregatedMetric> getAll() {
+ return metrics;
+ }
+}
Modified: giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/GiraphMetrics.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/GiraphMetrics.java?rev=1411493&r1=1411492&r2=1411493&view=diff
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/GiraphMetrics.java (original)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/GiraphMetrics.java Tue Nov 20 01:07:49 2012
@@ -17,9 +17,8 @@
*/
package org.apache.giraph.metrics;
+import org.apache.giraph.GiraphConfiguration;
import org.apache.giraph.graph.BspService;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.mapreduce.Mapper;
import com.google.common.collect.Lists;
@@ -54,11 +53,10 @@ public class GiraphMetrics {
/**
* Initialize GiraphMetrics with Hadoop Context
*
- * @param context Hadoop Context to use.
+ * @param conf GiraphConfiguration to use.
*/
- private GiraphMetrics(Mapper.Context context) {
- Configuration conf = context.getConfiguration();
- perJob = new GiraphMetricsRegistry(conf, "giraph.job");
+ private GiraphMetrics(GiraphConfiguration conf) {
+ perJob = new GiraphMetricsRegistry(conf, "giraph", "job");
perSuperstep = new SuperstepMetricsRegistry(conf,
BspService.INPUT_SUPERSTEP);
}
@@ -75,10 +73,10 @@ public class GiraphMetrics {
/**
* Initialize singleton instance of GiraphMetrics.
*
- * @param context Hadoop Context to use.
+ * @param conf GiraphConfiguration to use.
*/
- public static void init(Mapper.Context context) {
- INSTANCE = new GiraphMetrics(context);
+ public static void init(GiraphConfiguration conf) {
+ INSTANCE = new GiraphMetrics(conf);
}
/**
Modified: giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/GiraphMetricsRegistry.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/GiraphMetricsRegistry.java?rev=1411493&r1=1411492&r2=1411493&view=diff
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/GiraphMetricsRegistry.java (original)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/GiraphMetricsRegistry.java Tue Nov 20 01:07:49 2012
@@ -19,12 +19,12 @@
package org.apache.giraph.metrics;
import org.apache.giraph.GiraphConfiguration;
-import org.apache.hadoop.conf.Configuration;
import com.yammer.metrics.core.Counter;
import com.yammer.metrics.core.Gauge;
import com.yammer.metrics.core.Histogram;
import com.yammer.metrics.core.Meter;
+import com.yammer.metrics.core.Metric;
import com.yammer.metrics.core.MetricName;
import com.yammer.metrics.core.MetricPredicate;
import com.yammer.metrics.core.MetricsRegistry;
@@ -33,6 +33,7 @@ import com.yammer.metrics.reporting.Cons
import com.yammer.metrics.reporting.JmxReporter;
import java.io.PrintStream;
+import java.util.Map;
import java.util.concurrent.TimeUnit;
/**
@@ -41,6 +42,8 @@ import java.util.concurrent.TimeUnit;
public class GiraphMetricsRegistry {
/** String name of group to use for metrics created */
private String groupName;
+ /** String type to use for metrics created */
+ private String type;
/** Internal Yammer registry used */
private final MetricsRegistry registry;
/** JmxReporter that send metrics to JMX */
@@ -59,10 +62,13 @@ public class GiraphMetricsRegistry {
*
* @param conf Hadoop Configuration to use.
* @param groupName String group to use for metrics.
+ * @param type String type to use for metrics.
*/
- public GiraphMetricsRegistry(Configuration conf, String groupName) {
+ public GiraphMetricsRegistry(GiraphConfiguration conf, String groupName,
+ String type) {
this.groupName = groupName;
- if (conf.getBoolean(GiraphConfiguration.METRICS_ENABLE, false)) {
+ this.type = type;
+ if (conf.metricsEnabled()) {
registry = new MetricsRegistry();
jmxReporter = new JmxReporter(registry);
jmxReporter.start();
@@ -73,6 +79,24 @@ public class GiraphMetricsRegistry {
}
/**
+ * Get map of all metrics.
+ *
+ * @return Map of all metrics held.
+ */
+ public Map<MetricName, Metric> getAll() {
+ return registry.allMetrics();
+ }
+
+ /**
+ * Get group name used for metrics.
+ *
+ * @return String group name.
+ */
+ public String getGroupName() {
+ return groupName;
+ }
+
+ /**
* Set group name used by this MetricsRegistry. Used for incrementing
* superstep number to create a new hierarchy of metrics per superstep.
*
@@ -83,6 +107,24 @@ public class GiraphMetricsRegistry {
}
/**
+ * Get type used for new metrics created
+ *
+ * @return String type to use for metrics
+ */
+ public String getType() {
+ return type;
+ }
+
+ /**
+ * Set type to use for new metrics
+ *
+ * @param type String type to use
+ */
+ public void setType(String type) {
+ this.type = type;
+ }
+
+ /**
* Dump all the metrics to the PrintStream provided.
*
* @param out PrintStream to write metrics to.
@@ -105,69 +147,61 @@ public class GiraphMetricsRegistry {
* Creates a new {@link com.yammer.metrics.core.Counter} and registers it
* under the given group and name.
*
- * @param group what type of metric this is
* @param name the name of the metric
* @return a new {@link com.yammer.metrics.core.Counter}
*/
- public Counter getCounter(MetricGroup group, String name) {
- return registry.newCounter(makeMetricName(group, name));
+ public Counter getCounter(String name) {
+ return registry.newCounter(makeMetricName(name));
}
/**
* Given a new {@link com.yammer.metrics.core.Gauge}, registers it under the
* given group and name.
*
- * @param group what type of metric this is
* @param name the name of the metric
* @param metric the metric
* @param <T> the type of the value returned by the metric
* @return {@code metric}
*/
- public <T> Gauge<T> getGauge(MetricGroup group, String name,
- Gauge<T> metric) {
- return registry.newGauge(makeMetricName(group, name), metric);
+ public <T> Gauge<T> getGauge(String name, Gauge<T> metric) {
+ return registry.newGauge(makeMetricName(name), metric);
}
/**
* Creates a new non-biased {@link com.yammer.metrics.core.Histogram} and
* registers it under the given group and name.
*
- * @param group what type of metric this is
* @param name the name of the metric
* @return a new {@link com.yammer.metrics.core.Histogram}
*/
- public Histogram getHistogram(MetricGroup group, String name) {
- return registry.newHistogram(makeMetricName(group, name), false);
+ public Histogram getHistogram(String name) {
+ return registry.newHistogram(makeMetricName(name), false);
}
/**
* Creates a new {@link Histogram} and registers it under the given group
* and name.
*
- * @param group what type of metric this is
* @param name the name of the metric
* @param biased whether or not the histogram should be biased
* @return a new {@link Histogram}
*/
- public Histogram getHistogram(MetricGroup group, String name,
- boolean biased) {
- return registry.newHistogram(makeMetricName(group, name), biased);
+ public Histogram getHistogram(String name, boolean biased) {
+ return registry.newHistogram(makeMetricName(name), biased);
}
/**
* Creates a new {@link com.yammer.metrics.core.Meter} and registers it under
* the given group and name.
*
- * @param group what type of metric this is
* @param name the name of the metric
* @param eventType the plural name of the type of events the meter is
* measuring (e.g., {@code "requests"})
* @param timeUnit the rate unit of the new meter
* @return a new {@link com.yammer.metrics.core.Meter}
*/
- public Meter getMeter(MetricGroup group, String name, String eventType,
- TimeUnit timeUnit) {
- return registry.newMeter(makeMetricName(group, name), eventType, timeUnit);
+ public Meter getMeter(String name, String eventType, TimeUnit timeUnit) {
+ return registry.newMeter(makeMetricName(name), eventType, timeUnit);
}
/**
@@ -175,38 +209,45 @@ public class GiraphMetricsRegistry {
* the given group and name, measuring elapsed time in milliseconds and
* invocations per second.
*
- * @param group what type of metric this is
* @param name the name of the metric
* @return a new {@link com.yammer.metrics.core.Timer}
*/
- public Timer getTimer(MetricGroup group, String name) {
- return getTimer(group, name, TimeUnit.MILLISECONDS, TimeUnit.SECONDS);
+ public Timer getTimer(String name) {
+ return getTimer(name, TimeUnit.MILLISECONDS, TimeUnit.SECONDS);
}
/**
* Creates a new {@link Timer} and registers it under the given
* group and name.
*
- * @param group what type of metric this is
* @param name the name of the metric
* @param durationUnit the duration scale unit of the new timer
* @param rateUnit the rate scale unit of the new timer
* @return a new {@link Timer}
*/
- public Timer getTimer(MetricGroup group, String name,
- TimeUnit durationUnit, TimeUnit rateUnit) {
- return registry.newTimer(makeMetricName(group, name),
- durationUnit, rateUnit);
+ public Timer getTimer(String name, TimeUnit durationUnit, TimeUnit rateUnit) {
+ return registry.newTimer(makeMetricName(name), durationUnit, rateUnit);
+ }
+
+ /**
+ * Get a Gauge that is already present in the MetricsRegistry
+ *
+ * @param name String name of Gauge
+ * @param <T> value type Gauge returns
+ * @return Gauge<T> from MetricsRegistry
+ */
+ public <T> Gauge<T> getExistingGauge(String name) {
+ Metric metric = registry.allMetrics().get(makeMetricName(name));
+ return metric instanceof Gauge ? (Gauge<T>) metric : null;
}
/**
* Create a MetricName using the job ID, group, and name.
*
- * @param group what type of metric this is
* @param name String name given to metric
* @return MetricName for use with MetricsRegistry
*/
- protected MetricName makeMetricName(MetricGroup group, String name) {
- return new MetricName(groupName, group.toString().toLowerCase(), name);
+ protected MetricName makeMetricName(String name) {
+ return new MetricName(groupName, type, name);
}
}
Added: giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/GiraphTimer.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/GiraphTimer.java?rev=1411493&view=auto
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/GiraphTimer.java (added)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/GiraphTimer.java Tue Nov 20 01:07:49 2012
@@ -0,0 +1,106 @@
+/*
+ * 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.giraph.metrics;
+
+import com.google.common.collect.ImmutableMap;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * A timer to record duration of an event in a given TimeUnit.
+ * GiraphTimer is actually just a single-value Yammer Gauge that has some
+ * methods to make timing things easier.
+ */
+public class GiraphTimer extends ValueGauge<Long> {
+ /** Mapping from TimeUnit to abbreviation used for printing */
+ private static final ImmutableMap<TimeUnit, String> TIME_UNIT_TO_ABBREV =
+ ImmutableMap.<TimeUnit, String>builder().
+ put(TimeUnit.DAYS, "days").
+ put(TimeUnit.HOURS, "hours").
+ put(TimeUnit.MICROSECONDS, "us").
+ put(TimeUnit.MILLISECONDS, "ms").
+ put(TimeUnit.MINUTES, "mins").
+ put(TimeUnit.NANOSECONDS, "ns").
+ put(TimeUnit.SECONDS, "secs").build();
+
+ /** dimension to measure things by */
+ private TimeUnit timeUnit;
+
+ /**
+ * Create new timer, add it to the registry.
+ *
+ * @param registry GiraphMetricsRegistry to add timer to
+ * @param name String name of timer
+ * @param timeUnit TimeUnit to measure in
+ */
+ public GiraphTimer(GiraphMetricsRegistry registry, String name,
+ TimeUnit timeUnit) {
+ super(registry, name);
+ this.timeUnit = timeUnit;
+ set(0L);
+ }
+
+ /**
+ * Get TimeUnit used.
+ *
+ * @return TimeUnit being used.
+ */
+ public TimeUnit getTimeUnit() {
+ return timeUnit;
+ }
+
+ /**
+ * Begin timing an event.
+ *
+ * @return GiraphTimerContext. Use stop() to end timing the event.
+ */
+ public GiraphTimerContext time() {
+ return new GiraphTimerContext(this);
+ }
+
+ /**
+ * Set value from a given TimeUnit, converting to our TimeUnit.
+ *
+ * @param value long measurement taken.
+ * @param valueTimeUnit TimeUnit measurement is in.
+ * @return this
+ */
+ public GiraphTimer set(long value, TimeUnit valueTimeUnit) {
+ set(timeUnit.convert(value, valueTimeUnit));
+ return this;
+ }
+
+ /**
+ * Get abbreviated string of TimeUnit.
+ *
+ * @return TimeUnit abbreviation.
+ */
+ public String getTimeUnitAbbrev() {
+ return TIME_UNIT_TO_ABBREV.get(timeUnit);
+ }
+
+ /**
+ * Get string representation of value
+ *
+ * @return String value and abbreviated time unit
+ */
+ public String valueAndUnit() {
+ return "" + value() + " " + getTimeUnitAbbrev();
+ }
+}
Added: giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/GiraphTimerContext.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/GiraphTimerContext.java?rev=1411493&view=auto
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/GiraphTimerContext.java (added)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/GiraphTimerContext.java Tue Nov 20 01:07:49 2012
@@ -0,0 +1,56 @@
+/*
+ * 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.giraph.metrics;
+
+import org.apache.giraph.utils.SystemTime;
+import org.apache.giraph.utils.Time;
+import org.apache.giraph.utils.Times;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Context to use with GiraphTimer for timing an event.
+ */
+public class GiraphTimerContext {
+ /** Object to use for time measurement */
+ private static Time TIME = SystemTime.get();
+
+ /** GiraphTimer to callback to with result of timing */
+ private GiraphTimer giraphTimer;
+ /** Beginning time to compare against */
+ private long startTime;
+
+ /**
+ * Create with GiraphTimer to callback to.
+ *
+ * @param giraphTimer GiraphTimer to use.
+ */
+ public GiraphTimerContext(GiraphTimer giraphTimer) {
+ this.giraphTimer = giraphTimer;
+ this.startTime = TIME.getNanoseconds();
+ }
+
+ /**
+ * End timing. Set time elapsed in GiraphTimer.
+ */
+ public void stop() {
+ giraphTimer.set(Times.getNanosecondsSince(TIME, startTime),
+ TimeUnit.NANOSECONDS);
+ }
+}
Added: giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/LongAndTimeUnit.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/LongAndTimeUnit.java?rev=1411493&view=auto
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/LongAndTimeUnit.java (added)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/LongAndTimeUnit.java Tue Nov 20 01:07:49 2012
@@ -0,0 +1,88 @@
+/*
+ * 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.giraph.metrics;
+
+import com.google.common.collect.ImmutableMap;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Pair of long,TimeUnit
+ */
+public class LongAndTimeUnit {
+ /** Mapping from TimeUnit to abbreviation used for printing */
+ private static final ImmutableMap<TimeUnit, String> TIME_UNIT_TO_ABBREV =
+ ImmutableMap.<TimeUnit, String>builder().
+ put(TimeUnit.DAYS, "days").
+ put(TimeUnit.HOURS, "hours").
+ put(TimeUnit.MICROSECONDS, "us").
+ put(TimeUnit.MILLISECONDS, "ms").
+ put(TimeUnit.MINUTES, "mins").
+ put(TimeUnit.NANOSECONDS, "ns").
+ put(TimeUnit.SECONDS, "secs").build();
+
+ /** value held */
+ private long value;
+ /** TimeUnit part */
+ private TimeUnit timeUnit;
+
+ @Override
+ public String toString() {
+ if (timeUnit == null) {
+ return String.valueOf(value);
+ } else {
+ return value + " " + TIME_UNIT_TO_ABBREV.get(timeUnit);
+ }
+ }
+
+ /**
+ * Get the value
+ *
+ * @return value
+ */
+ public long getValue() {
+ return value;
+ }
+
+ /**
+ * Set the value
+ *
+ * @param value to set
+ */
+ public void setValue(long value) {
+ this.value = value;
+ }
+
+ /**
+ * Get the TimeUnit
+ *
+ * @return TimeUnit
+ */
+ public TimeUnit getTimeUnit() {
+ return timeUnit;
+ }
+
+ /**
+ * Set the TimeUnit
+ * @param timeUnit to set
+ */
+ public void setTimeUnit(TimeUnit timeUnit) {
+ this.timeUnit = timeUnit;
+ }
+}
Added: giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/MetricsRegistryDebugger.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/MetricsRegistryDebugger.java?rev=1411493&view=auto
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/MetricsRegistryDebugger.java (added)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/MetricsRegistryDebugger.java Tue Nov 20 01:07:49 2012
@@ -0,0 +1,57 @@
+/*
+ * 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.giraph.metrics;
+
+import org.apache.log4j.Logger;
+
+import com.yammer.metrics.core.Metric;
+import com.yammer.metrics.core.MetricName;
+import com.yammer.metrics.core.MetricsRegistryListener;
+
+/**
+ * A simple MetricsRegistry listener that prints every Metric added/removed.
+ */
+public class MetricsRegistryDebugger implements MetricsRegistryListener {
+ /** logger instance */
+ private static final Logger LOG =
+ Logger.getLogger(MetricsRegistryDebugger.class);
+ /** static singleton instance */
+ private static final MetricsRegistryDebugger INSTANCE =
+ new MetricsRegistryDebugger();
+
+ /**
+ * Get singleton static instance of this class.
+ *
+ * @return singleton static instance.
+ */
+ public static MetricsRegistryDebugger get() {
+ return INSTANCE;
+ }
+
+ @Override
+ public void onMetricAdded(MetricName name, Metric metric) {
+ LOG.info("Adding metric " + name + " - " +
+ metric.getClass().getSimpleName());
+ }
+
+ @Override
+ public void onMetricRemoved(MetricName name) {
+ LOG.info("Removing metric " + name);
+ }
+}
Modified: giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/SuperstepMetricsRegistry.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/SuperstepMetricsRegistry.java?rev=1411493&r1=1411492&r2=1411493&view=diff
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/SuperstepMetricsRegistry.java (original)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/SuperstepMetricsRegistry.java Tue Nov 20 01:07:49 2012
@@ -18,15 +18,8 @@
package org.apache.giraph.metrics;
+import org.apache.giraph.GiraphConfiguration;
import org.apache.giraph.graph.BspService;
-import org.apache.giraph.graph.BspServiceWorker;
-import org.apache.giraph.graph.ComputeCallable;
-import org.apache.giraph.graph.GraphMapper;
-import org.apache.hadoop.conf.Configuration;
-
-import com.yammer.metrics.core.Gauge;
-import com.yammer.metrics.core.Metric;
-import com.yammer.metrics.core.Timer;
import java.io.PrintStream;
@@ -50,8 +43,8 @@ public class SuperstepMetricsRegistry ex
* @param conf Hadoop Configuration to use.
* @param superstep number of superstep to use as group for metrics.
*/
- public SuperstepMetricsRegistry(Configuration conf, long superstep) {
- super(conf, makeSuperstepGroupName(superstep));
+ public SuperstepMetricsRegistry(GiraphConfiguration conf, long superstep) {
+ super(conf, "giraph.superstep", String.valueOf(superstep));
this.superstep = superstep;
}
@@ -61,48 +54,17 @@ public class SuperstepMetricsRegistry ex
* @param superstep long number of superstep to use.
*/
public void setSuperstep(long superstep) {
- super.setGroupName(makeSuperstepGroupName(superstep));
+ super.setType(String.valueOf(superstep));
this.superstep = superstep;
}
/**
- * Create group name to use for superstep.
- *
- * @param superstep long value of superstep to use.
- * @return String group for superstep to use for metrics created.
- */
- private static String makeSuperstepGroupName(long superstep) {
- return "giraph.superstep." + superstep;
- }
-
- /**
* Print human readable summary of superstep metrics.
*
* @param out PrintStream to write to.
*/
public void printSummary(PrintStream out) {
- Long commTime = getGaugeValue(MetricGroup.NETWORK,
- GraphMapper.GAUGE_COMMUNICATION_TIME);
- Long computeAllTime = getGaugeValue(MetricGroup.COMPUTE,
- GraphMapper.GAUGE_COMPUTE_ALL);
- Long timeToFirstMsg = getGaugeValue(MetricGroup.NETWORK,
- GraphMapper.GAUGE_TIME_TO_FIRST_MSG);
- Long superstepTime = getGaugeValue(MetricGroup.COMPUTE,
- GraphMapper.GAUGE_SUPERSTEP_TIME);
- Long waitingMs = getGaugeValue(MetricGroup.NETWORK,
- BspServiceWorker.GAUGE_WAITING_TIME);
- Timer computeOne = getTimer(MetricGroup.COMPUTE,
- ComputeCallable.TIMER_COMPUTE_ONE);
- double userComputeTime = computeOne.mean() * computeOne.count();
-
- out.println("");
- out.println("Superstep " + superstep + ":");
- out.println(" superstep time: " + superstepTime + " ms");
- out.println(" time to first message: " + timeToFirstMsg + " ms");
- out.println(" compute time: " + computeAllTime + " ms");
- out.println(" user compute time: " + userComputeTime + " ms");
- out.println(" network communication time: " + commTime + " ms");
- out.println(" waiting time: " + waitingMs + " ms");
+ new WorkerSuperstepMetrics().readFromRegistry().print(superstep, out);
}
/**
@@ -111,31 +73,4 @@ public class SuperstepMetricsRegistry ex
public void printSummary() {
printSummary(System.out);
}
-
- /**
- * Get a Gauge that is already present in the MetricsRegistry
- *
- * @param group MetricGroup Gauge belongs to
- * @param name String name of Gauge
- * @param <T> value type Gauge returns
- * @return Gauge<T> from MetricsRegistry
- */
- private <T> Gauge<T> getExistingGauge(MetricGroup group, String name) {
- Metric metric = getInternalRegistry().allMetrics().
- get(makeMetricName(group, name));
- return metric instanceof Gauge ? (Gauge<T>) metric : null;
- }
-
- /**
- * Get value of Gauge that is already present in the MetricsRegistry
- *
- * @param group MetricGroup Gauge belongs to
- * @param name String name of Gauge
- * @param <T> value type Gauge returns
- * @return T value of Gauge<T> from MetricsRegistry
- */
- private <T> T getGaugeValue(MetricGroup group, String name) {
- Gauge<T> gauge = getExistingGauge(group, name);
- return gauge == null ? null : gauge.value();
- }
}
Modified: giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/ValueGauge.java
URL: http://svn.apache.org/viewvc/giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/ValueGauge.java?rev=1411493&r1=1411492&r2=1411493&view=diff
==============================================================================
--- giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/ValueGauge.java (original)
+++ giraph/trunk/giraph/src/main/java/org/apache/giraph/metrics/ValueGauge.java Tue Nov 20 01:07:49 2012
@@ -27,23 +27,21 @@ import com.yammer.metrics.core.Gauge;
*/
public class ValueGauge<T extends Number> extends Gauge<T> {
/** value held by this class */
- private T value;
+ private T val;
/**
* Constructor that registers Gauge in MetricsRegistry.
*
* @param registry GiraphMetricsRegistry to use.
- * @param group MetricGroup for Gauge.
* @param name String name of Gauge.
*/
- public ValueGauge(GiraphMetricsRegistry registry, MetricGroup group,
- String name) {
- registry.getGauge(group, name, this);
+ public ValueGauge(GiraphMetricsRegistry registry, String name) {
+ registry.getGauge(name, this);
}
@Override
public T value() {
- return value;
+ return val;
}
/**
@@ -52,7 +50,7 @@ public class ValueGauge<T extends Number
* @return double value
*/
public double getDouble() {
- return value.doubleValue();
+ return val != null ? val.doubleValue() : 0.0d;
}
/**
@@ -61,15 +59,17 @@ public class ValueGauge<T extends Number
* @return long value
*/
public long getLong() {
- return value.longValue();
+ return val != null ? val.longValue() : 0L;
}
/**
* Set value held by this object.
*
* @param value value to set.
+ * @return this
*/
- public void set(T value) {
- this.value = value;
+ public ValueGauge<T> set(T value) {
+ this.val = value;
+ return this;
}
}