incubator-giraph-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ach...@apache.org
Subject svn commit: r1214406 - in /incubator/giraph/trunk: ./ src/main/java/org/apache/giraph/benchmark/ src/main/java/org/apache/giraph/comm/ src/main/java/org/apache/giraph/examples/ src/main/java/org/apache/giraph/graph/ src/main/java/org/apache/giraph/utils/
Date Wed, 14 Dec 2011 19:06:10 GMT
Author: aching
Date: Wed Dec 14 19:06:10 2011
New Revision: 1214406

URL: http://svn.apache.org/viewvc?rev=1214406&view=rev
Log:
GIRAPH-104: Save half of maximum memory used from messaging. (aching)


Added:
    incubator/giraph/trunk/src/main/java/org/apache/giraph/utils/MemoryUtils.java
Modified:
    incubator/giraph/trunk/CHANGELOG
    incubator/giraph/trunk/src/main/java/org/apache/giraph/benchmark/RandomMessageBenchmark.java
    incubator/giraph/trunk/src/main/java/org/apache/giraph/comm/BasicRPCCommunications.java
    incubator/giraph/trunk/src/main/java/org/apache/giraph/examples/LongSumAggregator.java
    incubator/giraph/trunk/src/main/java/org/apache/giraph/graph/BspServiceWorker.java
    incubator/giraph/trunk/src/main/java/org/apache/giraph/graph/GraphMapper.java
    incubator/giraph/trunk/src/main/java/org/apache/giraph/graph/WorkerContext.java

Modified: incubator/giraph/trunk/CHANGELOG
URL: http://svn.apache.org/viewvc/incubator/giraph/trunk/CHANGELOG?rev=1214406&r1=1214405&r2=1214406&view=diff
==============================================================================
--- incubator/giraph/trunk/CHANGELOG (original)
+++ incubator/giraph/trunk/CHANGELOG Wed Dec 14 19:06:10 2011
@@ -2,6 +2,8 @@ Giraph Change Log
 
 Release 0.70.0 - unreleased
 
+  GIRAPH-104: Save half of maximum memory used from messaging. (aching)
+
   GIRAPH-10: Aggregators are not exported. (claudio)
 
   GIRAPH-100: GIRAPH-100 - Data input sampling and testing

Modified: incubator/giraph/trunk/src/main/java/org/apache/giraph/benchmark/RandomMessageBenchmark.java
URL: http://svn.apache.org/viewvc/incubator/giraph/trunk/src/main/java/org/apache/giraph/benchmark/RandomMessageBenchmark.java?rev=1214406&r1=1214405&r2=1214406&view=diff
==============================================================================
--- incubator/giraph/trunk/src/main/java/org/apache/giraph/benchmark/RandomMessageBenchmark.java
(original)
+++ incubator/giraph/trunk/src/main/java/org/apache/giraph/benchmark/RandomMessageBenchmark.java
Wed Dec 14 19:06:10 2011
@@ -23,52 +23,253 @@ import org.apache.commons.cli.CommandLin
 import org.apache.commons.cli.HelpFormatter;
 import org.apache.commons.cli.Options;
 import org.apache.commons.cli.PosixParser;
+import org.apache.giraph.examples.LongSumAggregator;
+import org.apache.giraph.graph.EdgeListVertex;
 import org.apache.giraph.graph.GiraphJob;
-import org.apache.giraph.graph.Vertex;
+import org.apache.giraph.graph.WorkerContext;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.io.BytesWritable;
 import org.apache.hadoop.io.DoubleWritable;
 import org.apache.hadoop.io.LongWritable;
 import org.apache.hadoop.util.Tool;
 import org.apache.hadoop.util.ToolRunner;
-
+import org.apache.log4j.Logger;
 import java.util.Iterator;
 import java.util.Random;
 
 /**
- * Random Message Benchmark for evaluating the message delivery feature
+ * Random Message Benchmark for evaluating the messaging performance.
  */
-public class RandomMessageBenchmark extends
-        Vertex<LongWritable, DoubleWritable, DoubleWritable, BytesWritable>
-        implements Tool {
+public class RandomMessageBenchmark implements Tool {
     /** Configuration from Configurable */
     private Configuration conf;
 
     /** How many supersteps to run */
-    public static String SUPERSTEP_COUNT =
+    public static final String SUPERSTEP_COUNT =
         "RandomMessageBenchmark.superstepCount";
-
     /** How many bytes per message */
-    public static String NUM_BYTES_PER_MESSAGE =
+    public static final String NUM_BYTES_PER_MESSAGE =
         "RandomMessageBenchmark.numBytesPerMessage";
-    /** How many bytes per message */
-    public static String NUM_MESSAGES_PER_VERTEX =
-        "RandomMessageBenchmark.numMessagesPerVertex";
+    /** Default bytes per message */
+    public static final int DEFAULT_NUM_BYTES_PER_MESSAGE = 16;
+    /** How many messages per edge */
+    public static final String NUM_MESSAGES_PER_EDGE=
+        "RandomMessageBenchmark.numMessagesPerEdge";
+    /** Default messages per edge */
+    public static final int DEFAULT_NUM_MESSAGES_PER_EDGE = 1;
+    /** All bytes sent during this superstep */
+    public static final String AGG_SUPERSTEP_TOTAL_BYTES =
+        "superstep total bytes sent";
+    /** All bytes sent during this application */
+    public static final String AGG_TOTAL_BYTES = "total bytes sent";
+    /** All messages during this superstep */
+    public static final String AGG_SUPERSTEP_TOTAL_MESSAGES =
+        "superstep total messages";
+    /** All messages during this application */
+    public static final String AGG_TOTAL_MESSAGES = "total messages";
+    /** All millis during this superstep */
+    public static final String AGG_SUPERSTEP_TOTAL_MILLIS =
+        "superstep total millis";
+    /** All millis during this application */
+    public static final String AGG_TOTAL_MILLIS = "total millis";
+    /** Workers for that superstep */
+    public static final String WORKERS = "workers";
+
+    /**
+     * {@link WorkerContext} forRandomMessageBenchmark.
+     */
+    private static class RandomMessageBenchmarkWorkerContext extends
+            WorkerContext {
+        /** Bytes to be sent */
+        private byte[] messageBytes;
+        /** Number of messages sent per edge */
+        private int numMessagesPerEdge = -1;
+        /** Number of supersteps */
+        private int numSupersteps = -1;
+        /** Random generator for random bytes message */
+        private final Random random = new Random(System.currentTimeMillis());
+        /** Start superstep millis */
+        private long startSuperstepMillis = 0;
+        /** Total bytes */
+        private long totalBytes = 0;
+        /** Total messages */
+        private long totalMessages = 0;
+        /** Total millis */
+        private long totalMillis = 0;
+        /** Class logger */
+        private static final Logger LOG =
+            Logger.getLogger(RandomMessageBenchmarkWorkerContext.class);
+
+        @Override
+        public void preApplication()
+                throws InstantiationException, IllegalAccessException {
+            messageBytes =
+                new byte[getContext().getConfiguration().
+                         getInt(NUM_BYTES_PER_MESSAGE,
+                               DEFAULT_NUM_BYTES_PER_MESSAGE)];
+            numMessagesPerEdge =
+                getContext().getConfiguration().
+                    getInt(NUM_MESSAGES_PER_EDGE,
+                           DEFAULT_NUM_MESSAGES_PER_EDGE);
+            numSupersteps = getContext().getConfiguration().
+                                getInt(SUPERSTEP_COUNT, -1);
+            registerAggregator(AGG_SUPERSTEP_TOTAL_BYTES,
+                LongSumAggregator.class);
+            registerAggregator(AGG_SUPERSTEP_TOTAL_MESSAGES,
+                LongSumAggregator.class);
+            registerAggregator(AGG_SUPERSTEP_TOTAL_MILLIS,
+                LongSumAggregator.class);
+            registerAggregator(WORKERS,
+                LongSumAggregator.class);
+        }
+
+        @Override
+        public void preSuperstep() {
+            LongSumAggregator superstepBytesAggregator =
+                (LongSumAggregator) getAggregator(AGG_SUPERSTEP_TOTAL_BYTES);
+            LongSumAggregator superstepMessagesAggregator =
+                (LongSumAggregator) getAggregator(AGG_SUPERSTEP_TOTAL_MESSAGES);
+            LongSumAggregator superstepMillisAggregator =
+                (LongSumAggregator) getAggregator(AGG_SUPERSTEP_TOTAL_MILLIS);
+            LongSumAggregator workersAggregator =
+                (LongSumAggregator) getAggregator(WORKERS);
+
+            // For timing and tracking the supersteps
+            // - superstep 0 starts the time, but cannot display any stats
+            //   since nothing has been aggregated yet
+            // - supersteps > 0 can display the stats
+            if (getSuperstep() == 0) {
+                startSuperstepMillis = System.currentTimeMillis();
+            } else {
+                totalBytes +=
+                        superstepBytesAggregator.getAggregatedValue().get();
+                totalMessages +=
+                        superstepMessagesAggregator.getAggregatedValue().get();
+                totalMillis +=
+                        superstepMillisAggregator.getAggregatedValue().get();
+                double superstepMegabytesPerSecond =
+                        superstepBytesAggregator.getAggregatedValue().get() *
+                        workersAggregator.getAggregatedValue().get() *
+                        1000d / 1024d / 1024d /
+                        superstepMillisAggregator.getAggregatedValue().get();
+                double megabytesPerSecond = totalBytes *
+                        workersAggregator.getAggregatedValue().get() *
+                        1000d / 1024d / 1024d / totalMillis;
+                double superstepMessagesPerSecond =
+                        superstepMessagesAggregator.getAggregatedValue().get() *
+                        workersAggregator.getAggregatedValue().get() * 1000d /
+                        superstepMillisAggregator.getAggregatedValue().get();
+                double messagesPerSecond = totalMessages *
+                        workersAggregator.getAggregatedValue().get() * 1000d /
+                        totalMillis;
+                if (LOG.isInfoEnabled()) {
+                    LOG.info("Outputing statistics for superstep " +
+                             getSuperstep());
+                    LOG.info(AGG_SUPERSTEP_TOTAL_BYTES + " : " +
+                             superstepBytesAggregator.getAggregatedValue());
+                    LOG.info(AGG_TOTAL_BYTES + " : " + totalBytes);
+                    LOG.info(AGG_SUPERSTEP_TOTAL_MESSAGES + " : " +
+                             superstepMessagesAggregator.getAggregatedValue());
+                    LOG.info(AGG_TOTAL_MESSAGES + " : " + totalMessages);
+                    LOG.info(AGG_SUPERSTEP_TOTAL_MILLIS + " : " +
+                             superstepMillisAggregator.getAggregatedValue());
+                    LOG.info(AGG_TOTAL_MILLIS + " : " + totalMillis);
+                    LOG.info(WORKERS + " : " +
+                             workersAggregator.getAggregatedValue());
+                    LOG.info("Superstep megabytes / second = " +
+                             superstepMegabytesPerSecond);
+                    LOG.info("Total megabytes / second = " +
+                             megabytesPerSecond);
+                    LOG.info("Superstep messages / second = " +
+                             superstepMessagesPerSecond);
+                    LOG.info("Total messages / second = " +
+                             messagesPerSecond);
+                    LOG.info("Superstep megabytes / second / worker = " +
+                             superstepMegabytesPerSecond /
+                             workersAggregator.getAggregatedValue().get());
+                    LOG.info("Total megabytes / second / worker = " +
+                             megabytesPerSecond /
+                             workersAggregator.getAggregatedValue().get());
+                    LOG.info("Superstep messages / second / worker = " +
+                             superstepMessagesPerSecond /
+                             workersAggregator.getAggregatedValue().get());
+                    LOG.info("Total messages / second / worker = " +
+                             messagesPerSecond /
+                             workersAggregator.getAggregatedValue().get());
+                }
+            }
+
+            superstepBytesAggregator.setAggregatedValue(
+                new LongWritable(0L));
+            superstepMessagesAggregator.setAggregatedValue(
+                new LongWritable(0L));
+            workersAggregator.setAggregatedValue(
+                new LongWritable(1L));
+            useAggregator(AGG_SUPERSTEP_TOTAL_BYTES);
+            useAggregator(AGG_SUPERSTEP_TOTAL_MILLIS);
+            useAggregator(AGG_SUPERSTEP_TOTAL_MESSAGES);
+            useAggregator(WORKERS);
+        }
 
-    /** Random generator for random bytes message */
-    private Random rnd = new Random(System.currentTimeMillis());
+        @Override
+        public void postSuperstep() {
+            LongSumAggregator superstepMillisAggregator =
+                (LongSumAggregator) getAggregator(AGG_SUPERSTEP_TOTAL_MILLIS);
+            long endSuperstepMillis = System.currentTimeMillis();
+            long superstepMillis = endSuperstepMillis - startSuperstepMillis;
+            startSuperstepMillis = endSuperstepMillis;
+            superstepMillisAggregator.setAggregatedValue(
+                new LongWritable(superstepMillis));
+        }
 
-    @Override
-    public void compute(Iterator<BytesWritable> msgIterator) {
-        byte [] message = new byte[getConf().getInt(NUM_BYTES_PER_MESSAGE, 16)];
-        int numMessage = getConf().getInt(NUM_MESSAGES_PER_VERTEX, 1);
-        if (getSuperstep() < getConf().getInt(SUPERSTEP_COUNT, -1)) {
-            for (int i=0; i < numMessage; i++) {
-                rnd.nextBytes(message);
-                sendMsgToAllEdges(new BytesWritable(message));
+        @Override
+        public void postApplication() {}
+
+        public byte[] getMessageBytes() {
+            return messageBytes;
+        }
+
+        public int getNumMessagePerEdge() {
+            return numMessagesPerEdge;
+        }
+
+        public int getNumSupersteps() {
+            return numSupersteps;
+        }
+
+        public void randomizeMessageBytes() {
+            random.nextBytes(messageBytes);
+        }
+    }
+
+    /**
+     * Actual message computation (messaging in this case)
+     */
+    public static class RandomMessageVertex extends EdgeListVertex<
+            LongWritable, DoubleWritable, DoubleWritable, BytesWritable> {
+
+        @Override
+        public void compute(Iterator<BytesWritable> msgIterator) {
+            RandomMessageBenchmarkWorkerContext workerContext =
+                (RandomMessageBenchmarkWorkerContext) getWorkerContext();
+            LongSumAggregator superstepBytesAggregator =
+                (LongSumAggregator) getAggregator(AGG_SUPERSTEP_TOTAL_BYTES);
+            LongSumAggregator superstepMessagesAggregator =
+                (LongSumAggregator) getAggregator(AGG_SUPERSTEP_TOTAL_MESSAGES);
+            if (getSuperstep() < workerContext.getNumSupersteps()) {
+                for (int i = 0; i < workerContext.getNumMessagePerEdge();
+                        i++) {
+                    workerContext.randomizeMessageBytes();
+                    sendMsgToAllEdges(
+                        new BytesWritable(workerContext.getMessageBytes()));
+                    long bytesSent = workerContext.getMessageBytes().length *
+                        getNumOutEdges();
+                    superstepBytesAggregator.aggregate(bytesSent);
+                    superstepMessagesAggregator.aggregate(getNumOutEdges());
+                }
+            } else {
+                voteToHalt();
             }
-        } else {
-            voteToHalt();
         }
     }
 
@@ -155,8 +356,9 @@ public class RandomMessageBenchmark exte
         int workers = Integer.parseInt(cmd.getOptionValue('w'));
         GiraphJob job = new GiraphJob(getConf(), getClass().getName());
         job.getConfiguration().setInt(GiraphJob.CHECKPOINT_FREQUENCY, 0);
-        job.setVertexClass(getClass());
+        job.setVertexClass(RandomMessageVertex.class);
         job.setVertexInputFormatClass(PseudoRandomVertexInputFormat.class);
+        job.setWorkerContextClass(RandomMessageBenchmarkWorkerContext.class);
         job.setWorkerConfiguration(workers, workers, 100.0f);
         job.getConfiguration().setLong(
             PseudoRandomVertexInputFormat.AGGREGATE_VERTICES,
@@ -171,7 +373,7 @@ public class RandomMessageBenchmark exte
             RandomMessageBenchmark.NUM_BYTES_PER_MESSAGE,
             Integer.parseInt(cmd.getOptionValue('b')));
         job.getConfiguration().setInt(
-            RandomMessageBenchmark.NUM_MESSAGES_PER_VERTEX,
+            RandomMessageBenchmark.NUM_MESSAGES_PER_EDGE,
             Integer.parseInt(cmd.getOptionValue('n')));
 
         boolean isVerbose = false;

Modified: incubator/giraph/trunk/src/main/java/org/apache/giraph/comm/BasicRPCCommunications.java
URL: http://svn.apache.org/viewvc/incubator/giraph/trunk/src/main/java/org/apache/giraph/comm/BasicRPCCommunications.java?rev=1214406&r1=1214405&r2=1214406&view=diff
==============================================================================
--- incubator/giraph/trunk/src/main/java/org/apache/giraph/comm/BasicRPCCommunications.java
(original)
+++ incubator/giraph/trunk/src/main/java/org/apache/giraph/comm/BasicRPCCommunications.java
Wed Dec 14 19:06:10 2011
@@ -48,8 +48,8 @@ import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Set;
-import java.util.TreeMap;
 import java.util.TreeSet;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
@@ -57,6 +57,7 @@ import java.util.concurrent.Future;
 import org.apache.giraph.graph.WorkerInfo;
 import org.apache.giraph.graph.partition.Partition;
 import org.apache.giraph.graph.partition.PartitionOwner;
+import org.apache.giraph.utils.MemoryUtils;
 
 /*if[HADOOP_FACEBOOK]
 import org.apache.hadoop.ipc.ProtocolSignature;
@@ -142,7 +143,7 @@ public abstract class BasicRPCCommunicat
      */
     private final Map<I, VertexMutations<I, V, E, M>>
         inVertexMutationsMap =
-            new TreeMap<I, VertexMutations<I, V, E, M>>();
+            new HashMap<I, VertexMutations<I, V, E, M>>();
 
     /** Maximum size of cached message list, before sending it out */
     private final int maxSize;
@@ -160,7 +161,7 @@ public abstract class BasicRPCCommunicat
     private class PeerConnection {
         /**
          * Map of outbound messages going to a particular remote server,
-         * mapping from vertex range (max vertex index) to list of messages.
+         * mapping from the destination vertex to a list of messages.
          * (Synchronized with itself).
          */
         private final Map<I, MsgList<M>> outMessagesPerPeer;
@@ -168,7 +169,6 @@ public abstract class BasicRPCCommunicat
          * Client interface: RPC proxy for remote server, this class for local
          */
         private final CommunicationsInterface<I, V, E, M> peer;
-        /** Maximum size of cached message list, before sending it out */
         /** Boolean, set to false when local client (self), true otherwise */
         private final boolean isProxy;
 
@@ -190,11 +190,18 @@ public abstract class BasicRPCCommunicat
         public CommunicationsInterface<I, V, E, M> getRPCProxy() {
             return peer;
         }
+
+        @Override
+        public String toString() {
+            return peer.getName() + ", proxy=" + isProxy;
+        }
     }
 
     private class PeerFlushExecutor implements Runnable {
         private final PeerConnection peerConnection;
         private final Mapper<?, ?, ?, ?>.Context context;
+        // Report on the status of this flusher if this interval was exceeded
+        private static final int REPORTING_INTERVAL_MIN_MILLIS = 60000;
 
         PeerFlushExecutor(PeerConnection peerConnection,
                           Mapper<?, ?, ?, ?>.Context context) {
@@ -206,11 +213,17 @@ public abstract class BasicRPCCommunicat
         public void run() {
             CommunicationsInterface<I, V, E, M> proxy
                 = peerConnection.getRPCProxy();
+            long startMillis = System.currentTimeMillis();
+            long lastReportedMillis = startMillis;
 
             try {
+                int verticesDone = 0;
                 synchronized (peerConnection.outMessagesPerPeer) {
-                    for (Entry<I, MsgList<M>> e :
-                        peerConnection.outMessagesPerPeer.entrySet()) {
+                    final int vertices =
+                        peerConnection.outMessagesPerPeer.size();
+                    while (!peerConnection.outMessagesPerPeer.isEmpty()) {
+                        Entry<I, MsgList<M>> e =
+                            peerConnection.outMessagesPerPeer.entrySet().iterator().next();
                         MsgList<M> msgList = e.getValue();
 
                         if (msgList.size() > 0) {
@@ -222,27 +235,10 @@ public abstract class BasicRPCCommunicat
                                         proxy.putMsg(e.getKey(), combinedMsg);
                                     }
                                 } else {
-                                    if (LOG.isDebugEnabled()) {
-                                        LOG.debug("putAllMessages: " +
-                                            proxy.getName() +
-                                            " putting (list) " + msgList +
-                                            " to " + e.getKey() +
-                                            ", proxy = " +
-                                            peerConnection.isProxy);
-                                    }
                                     proxy.putMsgList(e.getKey(), msgList);
                                 }
-                                msgList.clear();
                             } else {
                                 for (M msg : msgList) {
-                                    if (LOG.isDebugEnabled()) {
-                                        LOG.debug("putAllMessages: "
-                                            + proxy.getName() +
-                                            " putting " + msg +
-                                            " to " + e.getKey() +
-                                            ", proxy = " +
-                                            peerConnection.isProxy);
-                                    }
                                     if (msg == null) {
                                         throw new IllegalArgumentException(
                                             "putAllMessages: Cannot put " +
@@ -251,7 +247,38 @@ public abstract class BasicRPCCommunicat
                                     proxy.putMsg(e.getKey(), msg);
                                     context.progress();
                                 }
-                                msgList.clear();
+                            }
+                            msgList.clear();
+                        }
+
+                        // Clean up the memory with the message list
+                        msgList = null;
+                        peerConnection.outMessagesPerPeer.remove(e.getKey());
+                        e = null;
+
+                        ++verticesDone;
+                        long curMillis = System.currentTimeMillis();
+                        if ((lastReportedMillis +
+                                REPORTING_INTERVAL_MIN_MILLIS) < curMillis) {
+                            lastReportedMillis = curMillis;
+                            if (LOG.isInfoEnabled()) {
+                                float percentDone =
+                                    (100f * verticesDone) /
+                                    vertices;
+                                float minutesUsed =
+                                    (curMillis - startMillis) / 1000f / 60f;
+                                float minutesRemaining =
+                                    (minutesUsed * 100f / percentDone) -
+                                    minutesUsed;
+                                LOG.info("run: " + peerConnection + ", " +
+                                         verticesDone + " out of " +
+                                         vertices  +
+                                         " done in " + minutesUsed +
+                                         " minutes, " +
+                                         percentDone + "% done, ETA " +
+                                         minutesRemaining +
+                                         " minutes remaining, " +
+                                         MemoryUtils.getRuntimeMemoryStats());
                             }
                         }
                     }
@@ -297,14 +324,14 @@ public abstract class BasicRPCCommunicat
                     peerConnection.getRPCProxy();
 
                 if (combiner != null) {
-                        M combinedMsg = combiner.combine(destVertex,
-                                                         outMessage);
-                        if (combinedMsg != null) {
-                            proxy.putMsg(destVertex, combinedMsg);
-                        }
-                    } else {
-                        proxy.putMsgList(destVertex, outMessage);
+                    M combinedMsg = combiner.combine(destVertex,
+                                                     outMessage);
+                    if (combinedMsg != null) {
+                        proxy.putMsg(destVertex, combinedMsg);
                     }
+                } else {
+                    proxy.putMsgList(destVertex, outMessage);
+                }
             } catch (IOException e) {
                 LOG.error(e);
                 if (peerConnection.isProxy) {
@@ -560,7 +587,7 @@ end[HADOOP_FACEBOOK]*/
         synchronized(transientInMessages) {
             msgs = transientInMessages.get(vertex);
             if (msgs == null) {
-                msgs = new ArrayList<M>();
+                msgs = new ArrayList<M>(msgList.size());
                 transientInMessages.put(vertex, msgs);
             }
         }
@@ -833,7 +860,9 @@ end[HADOOP_FACEBOOK]*/
     @Override
     public long flush(Mapper<?, ?, ?, ?>.Context context) throws IOException {
         if (LOG.isInfoEnabled()) {
-            LOG.info("flush: starting for superstep " + service.getSuperstep());
+            LOG.info("flush: starting for superstep " +
+                      service.getSuperstep() + " " +
+                      MemoryUtils.getRuntimeMemoryStats());
         }
         for (List<M> msgList : inMessages.values()) {
             msgList.clear();
@@ -856,11 +885,20 @@ end[HADOOP_FACEBOOK]*/
             try {
                 future.get();
                 context.progress();
-            } catch (Exception e) {
-                throw new RuntimeException(e);
+            } catch (InterruptedException e) {
+                throw new IllegalStateException("flush: Got IOException", e);
+            } catch (ExecutionException e) {
+                throw new IllegalStateException(
+                    "flush: Got ExecutionException", e);
             }
         }
 
+        if (LOG.isInfoEnabled()) {
+            LOG.info("flush: ended for superstep " +
+                      service.getSuperstep() + " " +
+                      MemoryUtils.getRuntimeMemoryStats());
+        }
+
         long msgs = totalMsgsSentInSuperstep;
         totalMsgsSentInSuperstep = 0;
         return msgs;
@@ -869,7 +907,9 @@ end[HADOOP_FACEBOOK]*/
     @Override
     public void prepareSuperstep() {
         if (LOG.isInfoEnabled()) {
-            LOG.info("prepareSuperstep");
+            LOG.info("prepareSuperstep: Superstep " +
+                     service.getSuperstep() + " " +
+                     MemoryUtils.getRuntimeMemoryStats());
         }
         inPrepareSuperstep = true;
 

Modified: incubator/giraph/trunk/src/main/java/org/apache/giraph/examples/LongSumAggregator.java
URL: http://svn.apache.org/viewvc/incubator/giraph/trunk/src/main/java/org/apache/giraph/examples/LongSumAggregator.java?rev=1214406&r1=1214405&r2=1214406&view=diff
==============================================================================
--- incubator/giraph/trunk/src/main/java/org/apache/giraph/examples/LongSumAggregator.java
(original)
+++ incubator/giraph/trunk/src/main/java/org/apache/giraph/examples/LongSumAggregator.java
Wed Dec 14 19:06:10 2011
@@ -26,27 +26,30 @@ import org.apache.giraph.graph.Aggregato
  * Aggregator for summing up values.
  */
 public class LongSumAggregator implements Aggregator<LongWritable> {
+    /** Internal sum */
+    private long sum = 0;
 
-  private long sum = 0;
-
-  public void aggregate(long value) {
-      sum += value;
-  }
-
-  public void aggregate(LongWritable value) {
-      sum += value.get();
-  }
-
-  public void setAggregatedValue(LongWritable value) {
-      sum = value.get();
-  }
-
-  public LongWritable getAggregatedValue() {
-      return new LongWritable(sum);
-  }
-
-  public LongWritable createAggregatedValue() {
-      return new LongWritable();
-  }
-
+    public void aggregate(long value) {
+        sum += value;
+    }
+
+    @Override
+    public void aggregate(LongWritable value) {
+        sum += value.get();
+    }
+
+    @Override
+    public void setAggregatedValue(LongWritable value) {
+        sum = value.get();
+    }
+
+    @Override
+    public LongWritable getAggregatedValue() {
+        return new LongWritable(sum);
+    }
+
+    @Override
+    public LongWritable createAggregatedValue() {
+        return new LongWritable();
+    }
 }

Modified: incubator/giraph/trunk/src/main/java/org/apache/giraph/graph/BspServiceWorker.java
URL: http://svn.apache.org/viewvc/incubator/giraph/trunk/src/main/java/org/apache/giraph/graph/BspServiceWorker.java?rev=1214406&r1=1214405&r2=1214406&view=diff
==============================================================================
--- incubator/giraph/trunk/src/main/java/org/apache/giraph/graph/BspServiceWorker.java (original)
+++ incubator/giraph/trunk/src/main/java/org/apache/giraph/graph/BspServiceWorker.java Wed
Dec 14 19:06:10 2011
@@ -29,6 +29,7 @@ import org.apache.giraph.graph.partition
 import org.apache.giraph.graph.partition.PartitionOwner;
 import org.apache.giraph.graph.partition.PartitionStats;
 import org.apache.giraph.graph.partition.WorkerGraphPartitioner;
+import org.apache.giraph.utils.MemoryUtils;
 import org.apache.giraph.utils.WritableUtils;
 import org.apache.giraph.zk.BspEvent;
 import org.apache.giraph.zk.PredicateLock;
@@ -471,9 +472,7 @@ public class BspServiceWorker<
                 String status = "readVerticesFromInputSplit: Loaded " +
                     totalVerticesLoaded + " vertices and " +
                     totalEdgesLoaded + " edges " +
-                    ", totalMem = " + Runtime.getRuntime().totalMemory() +
-                    " maxMem ="  + Runtime.getRuntime().maxMemory() +
-                    " freeMem=" + Runtime.getRuntime().freeMemory() + " " +
+                    MemoryUtils.getRuntimeMemoryStats() + " " +
                     getGraphMapper().getMapFunctions().toString() +
                     " - Attempt=" + getApplicationAttempt() +
                     ", Superstep=" + getSuperstep();
@@ -899,11 +898,13 @@ public class BspServiceWorker<
         //
         // Master will coordinate the barriers and aggregate "doneness" of all
         // the vertices.  Each worker will:
-        // 1. Save aggregator values that are in use.
-        // 2. Report the statistics (vertices, edges, messages, etc.)
-        // of this worker
-        // 3. Let the master know it is finished.
-        // 4. Then it waits for the master to say whether to stop or not.
+        // 1. Flush the unsent messages
+        // 2. Execute user postSuperstep() if necessary.
+        // 3. Save aggregator values that are in use.
+        // 4. Report the statistics (vertices, edges, messages, etc.)
+        //    of this worker
+        // 5. Let the master know it is finished.
+        // 6. Wait for the master's global stats, and check if done
         long workerSentMessages = 0;
         try {
             workerSentMessages = commService.flush(getContext());
@@ -911,6 +912,17 @@ public class BspServiceWorker<
             throw new IllegalStateException(
                 "finishSuperstep: flush failed", e);
         }
+
+        if (getSuperstep() != INPUT_SUPERSTEP) {
+            getWorkerContext().postSuperstep();
+            getContext().progress();
+        }
+
+        if (LOG.isInfoEnabled()) {
+            LOG.info("finishSuperstep: Superstep " + getSuperstep() + " " +
+                      MemoryUtils.getRuntimeMemoryStats());
+        }
+
         JSONArray aggregatorValueArray =
             marshalAggregatorValues(getSuperstep());
         Collection<PartitionStats> finalizedPartitionStats =

Modified: incubator/giraph/trunk/src/main/java/org/apache/giraph/graph/GraphMapper.java
URL: http://svn.apache.org/viewvc/incubator/giraph/trunk/src/main/java/org/apache/giraph/graph/GraphMapper.java?rev=1214406&r1=1214405&r2=1214406&view=diff
==============================================================================
--- incubator/giraph/trunk/src/main/java/org/apache/giraph/graph/GraphMapper.java (original)
+++ incubator/giraph/trunk/src/main/java/org/apache/giraph/graph/GraphMapper.java Wed Dec
14 19:06:10 2011
@@ -22,6 +22,7 @@ import org.apache.giraph.bsp.Centralized
 import org.apache.giraph.graph.partition.Partition;
 import org.apache.giraph.graph.partition.PartitionOwner;
 import org.apache.giraph.graph.partition.PartitionStats;
+import org.apache.giraph.utils.MemoryUtils;
 import org.apache.giraph.utils.ReflectionUtils;
 import org.apache.giraph.zk.ZooKeeperManager;
 import org.apache.hadoop.conf.Configuration;
@@ -530,10 +531,7 @@ public class GraphMapper<I extends Writa
             }
 
             if (LOG.isDebugEnabled()) {
-                LOG.debug("map: totalMem=" +
-                          Runtime.getRuntime().totalMemory() +
-                          " maxMem=" + Runtime.getRuntime().maxMemory() +
-                          " freeMem=" + Runtime.getRuntime().freeMemory());
+                LOG.debug("map: " + MemoryUtils.getRuntimeMemoryStats());
             }
             context.progress();
 
@@ -576,6 +574,8 @@ public class GraphMapper<I extends Writa
                             basicVertex.getMsgList().iterator();
                         context.progress();
                         basicVertex.compute(vertexMsgIt);
+                        // Hint to GC to free the messages
+                        basicVertex.getMsgList().clear();
                     }
                     if (basicVertex.isHalted()) {
                         partitionStats.incrFinishedVertexCount();
@@ -585,15 +585,6 @@ public class GraphMapper<I extends Writa
                 }
                 partitionStatsList.add(partitionStats);
             }
-
-            serviceWorker.getWorkerContext().postSuperstep();
-            context.progress();
-            if (LOG.isInfoEnabled()) {
-                LOG.info("map: totalMem="
-                         + Runtime.getRuntime().totalMemory() +
-                         " maxMem=" + Runtime.getRuntime().maxMemory() +
-                         " freeMem=" + Runtime.getRuntime().freeMemory());
-            }
         } while (!serviceWorker.finishSuperstep(partitionStatsList));
         if (LOG.isInfoEnabled()) {
             LOG.info("map: BSP application done " +

Modified: incubator/giraph/trunk/src/main/java/org/apache/giraph/graph/WorkerContext.java
URL: http://svn.apache.org/viewvc/incubator/giraph/trunk/src/main/java/org/apache/giraph/graph/WorkerContext.java?rev=1214406&r1=1214405&r2=1214406&view=diff
==============================================================================
--- incubator/giraph/trunk/src/main/java/org/apache/giraph/graph/WorkerContext.java (original)
+++ incubator/giraph/trunk/src/main/java/org/apache/giraph/graph/WorkerContext.java Wed Dec
14 19:06:10 2011
@@ -24,7 +24,6 @@ import org.apache.hadoop.mapreduce.Mappe
 /**
  * WorkerContext allows for the execution of user code
  * on a per-worker basis. There's one WorkerContext per worker.
- *
  */
 @SuppressWarnings("rawtypes")
 public abstract class WorkerContext implements AggregatorUsage {
@@ -34,35 +33,35 @@ public abstract class WorkerContext impl
 	public void setGraphState(GraphState graphState) {
 		this.graphState = graphState;
 	}
-	
+
     /**
      * Initialize the WorkerContext.
-     * This method is executed once on each Worker before the first 
+     * This method is executed once on each Worker before the first
      * superstep starts.
-     * 
-     * @throws IllegalAccessException 
-     * @throws InstantiationException 
+     *
+     * @throws IllegalAccessException
+     * @throws InstantiationException
      */
 	public abstract void preApplication() throws InstantiationException,
 		IllegalAccessException;
-	
+
     /**
      * Finalize the WorkerContext.
-     * This method is executed once on each Worker after the last 
+     * This method is executed once on each Worker after the last
      * superstep ends.
      */
     public abstract void postApplication();
 
     /**
      * Execute user code.
-     * This method is executed once on each Worker before each 
+     * This method is executed once on each Worker before each
      * superstep starts.
      */
     public abstract void preSuperstep();
-    
+
     /**
      * Execute user code.
-     * This method is executed once on each Worker after each 
+     * This method is executed once on each Worker after each
      * superstep ends.
      */
     public abstract void postSuperstep();
@@ -75,7 +74,7 @@ public abstract class WorkerContext impl
     public long getSuperstep() {
         return graphState.getSuperstep();
     }
-    
+
     /**
      * Get the total (all workers) number of vertices that
      * existed in the previous superstep.
@@ -85,7 +84,7 @@ public abstract class WorkerContext impl
     public long getNumVertices() {
     	return graphState.getNumVertices();
     }
-    
+
     /**
      * Get the total (all workers) number of edges that
      * existed in the previous superstep.
@@ -95,7 +94,7 @@ public abstract class WorkerContext impl
     public long getNumEdges() {
     	return graphState.getNumEdges();
     }
-    
+
     /**
      * Get the mapper context
      *
@@ -104,7 +103,7 @@ public abstract class WorkerContext impl
     public Mapper.Context getContext() {
         return graphState.getContext();
     }
-    
+
     @Override
     public final <A extends Writable> Aggregator<A> registerAggregator(
             String name,

Added: incubator/giraph/trunk/src/main/java/org/apache/giraph/utils/MemoryUtils.java
URL: http://svn.apache.org/viewvc/incubator/giraph/trunk/src/main/java/org/apache/giraph/utils/MemoryUtils.java?rev=1214406&view=auto
==============================================================================
--- incubator/giraph/trunk/src/main/java/org/apache/giraph/utils/MemoryUtils.java (added)
+++ incubator/giraph/trunk/src/main/java/org/apache/giraph/utils/MemoryUtils.java Wed Dec
14 19:06:10 2011
@@ -0,0 +1,21 @@
+package org.apache.giraph.utils;
+
+/**
+ * Helper static methods for tracking memory usage.
+ */
+public class MemoryUtils {
+    /**
+     * Get stringified runtime memory stats
+     *
+     * @return String of all Runtime stats.
+     */
+    public static String getRuntimeMemoryStats() {
+        return "totalMem = " +
+               (Runtime.getRuntime().totalMemory() / 1024f / 1024f) +
+               "M, maxMem = "  +
+               (Runtime.getRuntime().maxMemory() / 1024f / 1024f) +
+               "M, freeMem = " +
+               (Runtime.getRuntime().freeMemory() / 1024f / 1024f)
+                + "M";
+    }
+}



Mime
View raw message