update execute nodes

liulhdarks · Dec 1, 2015 · 773b9e2 · 773b9e2
1 parent bb4c5be
commit 773b9e2
Show file tree

Hide file tree

Showing 5 changed files with 16 additions and 4 deletions.
diff --git a/src/main/java/darks/grid/balance/GridBalance.java b/src/main/java/darks/grid/balance/GridBalance.java
@@ -16,11 +16,14 @@
  */
 package darks.grid.balance;
 
+import java.util.List;
+
 import darks.grid.beans.GridNode;
 
 public abstract class GridBalance
 {
 
 	public abstract GridNode getBalanceNode();
 
+	public abstract List<GridNode> getTargetNodes();
 }
diff --git a/src/main/java/darks/grid/balance/RollPolingBalance.java b/src/main/java/darks/grid/balance/RollPolingBalance.java
@@ -66,4 +66,11 @@ private synchronized boolean initNodes()
 		}
 		return nodesList != null;
 	}
+
+	@Override
+	public synchronized List<GridNode> getTargetNodes()
+	{
+		return nodesList;
+	}
+
 }
diff --git a/src/main/java/darks/grid/executor/job/JobExecutor.java b/src/main/java/darks/grid/executor/job/JobExecutor.java
@@ -168,7 +168,6 @@ public String toSimpleString()
 			.append(getStatusType()).append('\t')
 			.append(System.currentTimeMillis() - getTimestamp()).append('/')
 			.append(job.getTimeout()).append(' ')
-			.append(System.currentTimeMillis() - getTimestamp()).append(' ')
 			.append(getDelay());
 		if (isCanncel())
 			buf.append(' ').append("CANCELED");

diff --git a/src/main/java/darks/grid/executor/task/mapred/MapReduceExecutor.java b/src/main/java/darks/grid/executor/task/mapred/MapReduceExecutor.java
@@ -24,6 +24,7 @@
 import org.slf4j.LoggerFactory;
 
 import darks.grid.GridRuntime;
+import darks.grid.balance.GridBalance;
 import darks.grid.beans.GridMessage;
 import darks.grid.beans.GridNode;
 import darks.grid.executor.ExecuteConfig;
@@ -55,8 +56,10 @@ public MapReduceExecutor(MapReduceTask<T, R> task,
     protected R execute()
     {
         ExecuteConfig config = getConfig();
-        List<GridNode> nodesList = GridRuntime.nodes().getSnapshotNodes();
-        task.initialize(nodesList, config.getBalance());
+        GridBalance balance = config.getBalance();
+        List<GridNode> nodesList = balance.getTargetNodes() != null ? 
+        		balance.getTargetNodes() : GridRuntime.nodes().getSnapshotNodes();
+        task.initialize(nodesList, balance);
         int jobCount = nodesList.size();
         if (config.getCallType() == CallType.SINGLE)
             jobCount = 1;

diff --git a/src/main/java/darks/grid/network/handler/msg/HEART_ALIVE.java b/src/main/java/darks/grid/network/handler/msg/HEART_ALIVE.java
@@ -46,7 +46,7 @@ public void handler(GridSession session, GridMessage msg) throws Exception
 			node.setHeartAliveTime(System.currentTimeMillis());
 			node.context().setMachineInfo(meta.context().getMachineInfo());
 		}
-		node.setPingDelay(arriveTime - meta.getTimestamp());
+		node.setPingDelay(Math.abs(arriveTime - meta.getTimestamp()));
 		//HEART ALIVE REPLY
 		if (msg.getType() == GridMessage.MSG_HEART_ALIVE)
 			GridRuntime.events().publish(EventsChannel.SYSTEM_CHANNEL, GridEvent.HEART_ALIVE_REPLY, msg);