Merge pull request #3 from vicweeks/milestone3-patch0

Milestone3 patch0

Author: vicweeks

cs455/scaling/client/Client.class

@@ -1,13 +1,16 @@
 package cs455.scaling.client;
 
 import cs455.scaling.tasks.ClientTask;
+import cs455.scaling.util.ClientLogger;
 import java.nio.channels.SocketChannel;
 import java.net.InetSocketAddress;
 import java.io.IOException;
+import java.util.Timer;
 
 public class Client {
 
     private static SocketChannel socketChannel;
+    private static ClientLogger clientLogger = new ClientLogger();
 
     public static void main(String[] args) {
 
@@ -27,9 +30,11 @@ public static void main(String[] args) {
 	if (args.length == 4)
 	    debug = true;
 
+	c.printStatus();
+	
 	try {
 	    c.setUpChannel(serverHost, serverPort);
-	    ClientTask clientTask = new ClientTask(socketChannel, messageRate);
+	    ClientTask clientTask = new ClientTask(socketChannel, messageRate, clientLogger, debug);
 	    clientTask.run();
 	} catch (IOException ioe) {
 	    System.out.println(ioe.getMessage());
@@ -39,10 +44,18 @@ public static void main(String[] args) {
 
     }
 
+    private void printStatus() {
+	// logs stats to the console evry interval
+	Timer timer = new Timer();
+	int interval = 5000;
+	timer.schedule(clientLogger, interval, interval);
+    }
+    
     private void setUpChannel(String serverHost, int serverPort)
 	throws IOException, InterruptedException{
 	System.out.println("Setting up client...");
 	socketChannel = SocketChannel.open();
+	socketChannel.configureBlocking(false);
 	socketChannel.connect(new InetSocketAddress(serverHost, serverPort));
 	while(!socketChannel.finishConnect())
 	    Thread.sleep(100);

cs455/scaling/client/Client.java

@@ -2,6 +2,8 @@
 
 import cs455.scaling.threadpool.ThreadPoolManager;
 import cs455.scaling.tasks.ServerTask;
+import cs455.scaling.util.ServerLogger;
+import cs455.scaling.util.ThroughputLogger;
 import java.nio.channels.Selector;
 import java.nio.channels.SelectionKey;
 import java.nio.channels.ServerSocketChannel;
@@ -11,13 +13,16 @@
 import java.net.InetSocketAddress;
 import java.io.IOException;
 import java.util.Set;
+import java.util.Timer;
 
 public class Server {
 
     private static boolean debug;
     private static ThreadPoolManager tpm;
     private static ServerSocketChannel ssChannel;
     private static Selector serverSelector;
+    private static ServerLogger serverLogger = new ServerLogger();
+    
     //private final int clientInterestSet = SelectionKey.OP_READ | SelectionKey.OP_WRITE;
 
     public static void main(String[] args) {
@@ -37,6 +42,8 @@ public static void main(String[] args) {
 	if (args.length == 3)
 	    debug = true;	
 
+	s.printStatus();
+	
 	tpm = new ThreadPoolManager(threadPoolSize, debug);
 
 	try {
@@ -49,6 +56,13 @@ public static void main(String[] args) {
 
     }
 
+    private void printStatus() {
+	// log stats to the console every interval
+	Timer timer = new Timer();
+	int interval = 5000;
+	timer.schedule(serverLogger, interval, interval);
+    }
+    
     private void setupServerSocket(int portNumber) throws IOException {
 	System.out.println("Setting up server...");
 	ssChannel = ServerSocketChannel.open();
@@ -64,7 +78,7 @@ private void executeServerLoop() throws IOException {
 	    if (socketChannel != null)
 		registerIncomingKey(socketChannel);
 	    // Find read ready channels
-	    if (serverSelector.selectNow() > 0) {
+	    if (serverSelector.select(1000) > 0) {
 		Set<SelectionKey> selectedKeys = serverSelector.selectedKeys();
 
 		for (SelectionKey key : selectedKeys) {		    
@@ -74,14 +88,13 @@ private void executeServerLoop() throws IOException {
 			tpm.assignTask();
 		    }
 		    selectedKeys.remove(key);
-		}
-		
+		}		
 	    }
 	}
     }
-
+    
     private void createTask(SelectionKey key) throws IOException {
-	ServerTask task = new ServerTask(key);
+	ServerTask task = new ServerTask(key, serverLogger, debug);
 	tpm.addTaskToQueue(task);
     }
 
@@ -91,6 +104,8 @@ private void registerIncomingKey(SocketChannel socketChannel) throws IOException
 	try {
 	    socketChannel.configureBlocking(false);
 	    SelectionKey key = socketChannel.register(serverSelector, SelectionKey.OP_READ);
+	    ThroughputLogger logger = serverLogger.addClient();	    
+	    key.attach(logger);
 	} catch (ClosedChannelException cce) {
 	    System.out.println(cce.getMessage());
 	}

cs455/scaling/server/Server.class

@@ -1,6 +1,7 @@
 package cs455.scaling.tasks;
 
 import cs455.scaling.util.HashGenerator;
+import cs455.scaling.util.ClientLogger;
 import java.nio.channels.SocketChannel;
 import java.nio.ByteBuffer;
 import java.io.IOException;
@@ -13,26 +14,32 @@ public class ClientTask extends Thread {
     private ByteBuffer buf;
     private int messageRate;
     private LinkedList<String> packetHashcodes;
+    private ClientLogger logger;
     private final HashGenerator hashGen;
+    private boolean debug;
 
-    public ClientTask(SocketChannel socketChannel, int messageRate) {
+    public ClientTask(SocketChannel socketChannel, int messageRate,
+		      ClientLogger logger, boolean debug) {
 	this.socketChannel = socketChannel;
 	buf = ByteBuffer.allocate(8000);
 	this.messageRate = messageRate;
 	packetHashcodes = new LinkedList<String>();
 	hashGen = new HashGenerator();
+	this.logger = logger;
+	this.debug = debug;
     }
 
     public void run() {
 	while(!isInterrupted()) {
-	    // test
 	    try {
 		sendMessage();
-		Thread.sleep(1000 / messageRate);
+		receiveMessage();
+		Thread.sleep(1000 / messageRate);		
 	    } catch (IOException ioe) {
 		System.out.println(ioe.getMessage());
+		System.exit(0);
 	    } catch (InterruptedException ie) {
-		System.out.println(ie.getMessage());
+		System.out.println(ie.getMessage());	        
 	    }
 	}
     }
@@ -47,11 +54,39 @@ private void sendMessage() throws IOException {
         byte[] message = generateRandomBytes();
 	String messageHash = hashGen.SHA1FromBytes(message);
 	packetHashcodes.add(messageHash);
-	buf.clear();
-	buf.put(message);
-	buf.flip();
-	
-	while(buf.hasRemaining())
-	    socketChannel.write(buf);
+
+	ByteBuffer buffer = ByteBuffer.wrap(message);
+	socketChannel.write(buffer);
+
+	logger.addSent();
+    }
+
+    private void receiveMessage() {
+        try {
+	    int bytesRead = socketChannel.read(buf);
+	    if (bytesRead <= 0) { // nothing has been read
+		if (debug)
+		    System.out.println("Nothing was read. " + bytesRead);
+		return;
+	    }
+	    byte[] message = new byte[bytesRead];
+	    buf.flip();
+	    buf.get(message);
+	    checkHash(message);
+	    buf.clear();
+	} catch (IOException ioe) {
+	    System.out.println(ioe.getMessage());
+	}
+    }
+
+    private void checkHash(byte[] message) {
+	String messageHash = new String(message);
+	int hashIndex = packetHashcodes.indexOf(messageHash);
+	if (hashIndex != -1) { // found hash
+	    packetHashcodes.remove(hashIndex);
+	    logger.addReceived();
+	    if (debug)
+		System.out.println("Found matching hash: " + messageHash);
+	}
     }
 }

cs455/scaling/server/Server.java

@@ -1,33 +1,49 @@
 package cs455.scaling.tasks;
 
 import cs455.scaling.util.HashGenerator;
+import cs455.scaling.util.ThroughputLogger;
+import cs455.scaling.util.ServerLogger;
 import java.nio.channels.SelectionKey;
 import java.nio.channels.SocketChannel;
 import java.nio.ByteBuffer;
 import java.io.IOException;
 
 public class ServerTask implements Runnable {
 
+    private SelectionKey key;
     private final HashGenerator hashGen;
     private SocketChannel socketChannel;
     private ByteBuffer buf;
+    private ThroughputLogger logger;
+    private ServerLogger serverLogger;
+    private boolean debug;
 
-    public ServerTask(SelectionKey key) {
+    public ServerTask(SelectionKey key, ServerLogger serverLogger, boolean debug) {
 	hashGen = new HashGenerator();
 	socketChannel = (SocketChannel) key.channel();
 	buf = ByteBuffer.allocate(8000);
+	logger = (ThroughputLogger) key.attachment();
+	this.serverLogger = serverLogger;
+	this.debug = debug;
+	this.key = key;
+	key.interestOps(SelectionKey.OP_WRITE);
     }
 
     public void run() {
-	try {
+	try {	    
 	    int bytesRead = socketChannel.read(buf);
+	    if (bytesRead == -1) {
+		System.out.println("Connection terminated by the client.");
+		return;
+	    }
+	    
 	    byte[] message = new byte[bytesRead];
 	    buf.flip();
 	    buf.get(message);
+	    
 	    String messageHash = getHash(message);
-	    System.out.println(messageHash);
-	    buf.clear();
-	    //replyWithHash(messageHash);
+	    buf.clear();	    
+	    replyWithHash(messageHash);
 	} catch (IOException ioe) {
 	    System.out.println(ioe.getMessage());
 	}       
@@ -39,11 +55,17 @@ private String getHash(byte[] message) {
 
     private void replyWithHash(String messageHash) throws IOException {
 	byte[] replyMessage = messageHash.getBytes();
-	buf.put(replyMessage);
-	buf.flip();
 
-	while(buf.hasRemaining())
-	    socketChannel.write(buf);
+	ByteBuffer buffer = ByteBuffer.wrap(replyMessage);        
+	socketChannel.write(buffer);
+	
+	if (debug)
+	    System.out.println("Wrote reply " + messageHash + " to socketChannel");
+	
+	logger.processMessage();
+	serverLogger.processMessage();
+
+	key.interestOps(SelectionKey.OP_READ);
     }
 
 }