Mercurial > hg > FederatedLinda

changeset 80:04bd4ae97e7c

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
RingTopology is finished.
author one
date Sun, 22 Nov 2009 13:59:05 +0900
parents 805645cf5ec0
children c001797f3fdb
files src/fdl/test/topology/MetaProtocolEngine.java src/fdl/test/topology/TopologyManagerEngine.java src/fdl/test/topology/ring/RingMetaProtocolEngine.java src/fdl/test/topology/ring/RingTopologyManagerEngine.java
diffstat 4 files changed, 74 insertions(+), 5 deletions(-) [+]
line wrap: on
line diff
--- a/src/fdl/test/topology/MetaProtocolEngine.java	Tue Nov 17 18:19:39 2009 +0900
+++ b/src/fdl/test/topology/MetaProtocolEngine.java	Sun Nov 22 13:59:05 2009 +0900
@@ -54,6 +54,7 @@
 	protected void initSendServer() {
 		sendLocalHostName();
 		connectSendServers();
+		sendLocalHostName();
 	}
 
 	protected void initTopologyManager() {
--- a/src/fdl/test/topology/TopologyManagerEngine.java	Tue Nov 17 18:19:39 2009 +0900
+++ b/src/fdl/test/topology/TopologyManagerEngine.java	Sun Nov 22 13:59:05 2009 +0900
@@ -46,6 +46,7 @@
 		makeConnection();
 		acceptNewNode();
 		sendLastToken();
+		checkConnection();
 	}
 	
 	// これを継承して、 Topology を形成
@@ -91,4 +92,22 @@
 		}
 	}
 	
+	protected void checkConnection() {
+		for (int i = 0; i < nodes.length; i++) {
+			PSXReply reply;
+			reply = manager.in(manageId);
+			do {
+				try {
+					manager.sync(1);
+				} catch (IOException e) {
+					e.printStackTrace();
+				}
+			} while (!reply.ready());
+			ByteBuffer data = reply.getData();
+			String ack = new String(data.array());
+			System.out.println("[DEBUG] ConnectedHost: " + ack);
+		}
+		System.out.println("[DEBUG] AllConnected");
+	}
+	
 }
--- a/src/fdl/test/topology/ring/RingMetaProtocolEngine.java	Tue Nov 17 18:19:39 2009 +0900
+++ b/src/fdl/test/topology/ring/RingMetaProtocolEngine.java	Sun Nov 22 13:59:05 2009 +0900
@@ -2,6 +2,8 @@
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.text.SimpleDateFormat;
+import java.util.Date;
 
 import fdl.MetaLinda;
 import fdl.PSXLinda;
@@ -21,10 +23,13 @@
 public class RingMetaProtocolEngine extends MetaProtocolEngine {
 	private int relayNum;
 	private static int relayId = 10;
-	
+	private Date startTime, endTime;
+	private SimpleDateFormat DF = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss.SSS");
 	public RingMetaProtocolEngine(MetaLinda ml, String managerHostName, int relayNum) {
 		super(ml, managerHostName);
 		this.relayNum = relayNum;
+		startTime = null;
+		endTime = null;
 	}
 	
 	@Override public void mainLoop() {
@@ -37,10 +42,14 @@
 	private void relayTuple(int tupleId) {
 		while (true) {
 			ByteBuffer data = receiveToken(tupleId);
+			if (startTime == null) {
+				startTime = new Date(); 
+			}
 			if (relayNum == 0) {
 				String token = new String(data.array());
 				if (!token.equals(lastToken)) {
 					// TODO: 実験結果をManagerに伝えるなどの処理
+					sendResult();
 					System.out.println("[Ring] relay finished: " + token);
 					// 実験終了を各ノードにリレーで伝える
 					sendToken(tupleId, ByteBuffer.wrap(lastToken.getBytes()));
@@ -48,10 +57,23 @@
 				return;
 			}
 			sendToken(tupleId, data);
+			System.out.println("[DEBUG] Relay...");
 			relayNum--;
 		}
 	}
 	
+	private void sendResult() {
+		this.endTime = new Date();
+		Long resultTime = new Long(endTime.getTime() - startTime.getTime()); 
+		ByteBuffer data = ByteBuffer.wrap(resultTime.toString().getBytes());
+		manager.out(relayId, data);
+		try {
+			manager.sync(1);
+		} catch (IOException e) {
+			e.printStackTrace();
+		}
+	}
+	
 	private ByteBuffer receiveToken(int tupleId) {
 		PSXReply reply = psxLocal.in(tupleId);
 		do {
--- a/src/fdl/test/topology/ring/RingTopologyManagerEngine.java	Tue Nov 17 18:19:39 2009 +0900
+++ b/src/fdl/test/topology/ring/RingTopologyManagerEngine.java	Sun Nov 22 13:59:05 2009 +0900
@@ -1,9 +1,16 @@
 package fdl.test.topology.ring;
 
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+
 import fdl.MetaLinda;
+import fdl.PSXReply;
 import fdl.test.topology.TopologyManagerEngine;
 
 public class RingTopologyManagerEngine extends TopologyManagerEngine {
+	private int relayId = 10;
 
 	// Constructor
 	public RingTopologyManagerEngine(MetaLinda ml, int nodeNum) {
@@ -12,12 +19,12 @@
 	
 	public void mainLoop() {
 		super.mainLoop();
-		
+		startRelay();
+		endRelay();
 	}
 	
 	protected void makeTopology() {
 		super.makeTopology();
-		startRelay();
 	}
 	
 	@Override protected void makeConnection() {
@@ -28,9 +35,29 @@
 	}
 	
 	private void startRelay() {
-		
-		
+		String relayString = "test";
+		ByteBuffer data = ByteBuffer.wrap(relayString.getBytes());
+		nodes[0].linda.out(relayId, data);
+		try {
+			nodes[0].linda.sync(1);
+		} catch (IOException e) {
+			e.printStackTrace();
+		}
 	}
 	
+	private void endRelay() {
+		PSXReply reply;
+		reply = manager.in(relayId);
+		do {
+			try {
+				manager.sync(1);
+			} catch (IOException e) {
+				e.printStackTrace();
+			}
+		} while (!reply.ready());
+		ByteBuffer data = reply.getData();
+		String resultData = new String(data.array());
+		System.out.println("[DEBUG] RelayTime: " + resultData);
+	}
 
 }