svn commit: r1294462 - /incubator/hama/trunk/graph/src/main/java/org/apache/hama/graph/GraphJobRunner.java

[ed...@apache.org]

Author: edwardyoon
Date: Tue Feb 28 05:01:58 2012
New Revision: 1294462

URL: http://svn.apache.org/viewvc?rev=1294462&view=rev
Log:
Exit if there's no update made

Modified:
    incubator/hama/trunk/graph/src/main/java/org/apache/hama/graph/GraphJobRunner.java

Modified: incubator/hama/trunk/graph/src/main/java/org/apache/hama/graph/GraphJobRunner.java
URL: http://svn.apache.org/viewvc/incubator/hama/trunk/graph/src/main/java/org/apache/hama/graph/GraphJobRunner.java?rev=1294462&r1=1294461&r2=1294462&view=diff
==============================================================================
--- incubator/hama/trunk/graph/src/main/java/org/apache/hama/graph/GraphJobRunner.java (original)
+++ incubator/hama/trunk/graph/src/main/java/org/apache/hama/graph/GraphJobRunner.java Tue Feb 28 05:01:58 2012
@@ -28,6 +28,7 @@ import java.util.Map.Entry;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.IntWritable;
 import org.apache.hadoop.io.MapWritable;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.io.Writable;
@@ -41,6 +42,8 @@ import org.apache.hama.util.KeyValuePair
 public class GraphJobRunner extends BSP {
   public static final Log LOG = LogFactory.getLog(GraphJobRunner.class);
   private Map<String, Vertex> vertices = new HashMap<String, Vertex>();
+  private String masterTask;
+  private String FLAG_MESSAGE = "hama.graph.msg.counts";
 
   @SuppressWarnings("unchecked")
   @Override
@@ -52,6 +55,7 @@ public class GraphJobRunner extends BSP 
     boolean updated = true;
     int iteration = 0;
     while (updated && iteration < maxIteration) {
+      int globalUpdateCounts = 0;
       peer.sync();
 
       MapWritable msg = null;
@@ -60,27 +64,50 @@ public class GraphJobRunner extends BSP 
 
         for (Entry<Writable, Writable> e : msg.entrySet()) {
           String vertexID = ((Text) e.getKey()).toString();
-          Writable value = e.getValue();
 
-          if (msgMap.containsKey(vertexID)) {
-            LinkedList<Writable> msgs = msgMap.get(vertexID);
-            msgs.add(value);
-            msgMap.put(vertexID, msgs);
+          if (vertexID.toString().equals(FLAG_MESSAGE)) {
+            if (((IntWritable) e.getValue()).get() == Integer.MIN_VALUE) {
+              updated = false;
+            } else {
+              globalUpdateCounts += ((IntWritable) e.getValue()).get();
+            }
           } else {
-            LinkedList<Writable> msgs = new LinkedList<Writable>();
-            msgs.add(value);
-            msgMap.put(vertexID, msgs);
-          }
+            Writable value = e.getValue();
 
+            if (msgMap.containsKey(vertexID)) {
+              LinkedList<Writable> msgs = msgMap.get(vertexID);
+              msgs.add(value);
+              msgMap.put(vertexID, msgs);
+            } else {
+              LinkedList<Writable> msgs = new LinkedList<Writable>();
+              msgs.add(value);
+              msgMap.put(vertexID, msgs);
+            }
+          }
         }
       }
 
-      if (msgMap.size() < 1) {
-        updated = false;
+      // exit if there's no update made
+      if (globalUpdateCounts == 0 && peer.getPeerName().equals(masterTask)
+          && peer.getSuperstepCount() > 1) {
+        MapWritable updatedCnt = new MapWritable();
+        updatedCnt.put(new Text(FLAG_MESSAGE), new IntWritable(
+            Integer.MIN_VALUE));
+
+        for (String peerName : peer.getAllPeerNames()) {
+          peer.send(peerName, updatedCnt);
+        }
       }
 
+      // send msgCounts to the master task
+      MapWritable updatedCnt = new MapWritable();
+      updatedCnt.put(new Text(FLAG_MESSAGE), new IntWritable(msgMap.size()));
+      peer.send(masterTask, updatedCnt);
+
       for (Map.Entry<String, LinkedList<Writable>> e : msgMap.entrySet()) {
-        vertices.get(e.getKey()).compute(e.getValue().iterator());
+        if (e.getValue().size() > 0) {
+          vertices.get(e.getKey()).compute(e.getValue().iterator());
+        }
       }
       iteration++;
     }
@@ -90,6 +117,8 @@ public class GraphJobRunner extends BSP 
   public void setup(BSPPeer peer) throws IOException, SyncException,
       InterruptedException {
     Configuration conf = peer.getConfiguration();
+    // Choose one as a master to collect global updates
+    masterTask = peer.getPeerName(0);
     LOG.debug("vertex class: " + conf.get("hama.graph.vertex.class"));
 
     KeyValuePair<? extends VertexWritable, ? extends VertexArrayWritable> next = null;