Merge pull request #2 from vicweeks/milestone2-patch0

Milestone2 patch0

Author: vicweeks

cs455/scaling/client/Client.java

@@ -1,10 +1,17 @@
 package cs455.scaling.client;
 
-public class Client {
+import cs455.scaling.tasks.ClientTask;
+import java.nio.channels.SocketChannel;
+import java.net.InetSocketAddress;
+import java.io.IOException;
 
+public class Client {
+    
+    private static SocketChannel socketChannel;
+    
     public static void main(String[] args) {
 
-	Client s = new Client();
+	Client c = new Client();
 
 	if (args.length < 3) {
             System.err.println(
@@ -19,7 +26,26 @@ public static void main(String[] args) {
 
 	if (args.length == 4)
 	    debug = true;
+
+	try {
+	    c.setUpChannel(serverHost, serverPort);
+	    ClientTask clientTask = new ClientTask(socketChannel, messageRate);
+	    clientTask.run();
+	} catch (IOException ioe) {
+	    System.out.println(ioe.getMessage());
+	} catch (InterruptedException ie) {
+	    System.out.println(ie.getMessage());
+	}
 
     }
+
+    private void setUpChannel(String serverHost, int serverPort)
+	throws IOException, InterruptedException{
+	System.out.println("Setting up client...");
+	socketChannel = SocketChannel.open();
+	socketChannel.connect(new InetSocketAddress(serverHost, serverPort));
+	while(!socketChannel.finishConnect())
+	    Thread.sleep(100);
+    }
 
 }

cs455/scaling/server/Server.java

@@ -1,6 +1,24 @@
 package cs455.scaling.server;
 
+import cs455.scaling.threadpool.ThreadPoolManager;
+import cs455.scaling.tasks.ServerTask;
+import java.nio.channels.Selector;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.ServerSocketChannel;
+import java.nio.channels.SocketChannel;
+import java.nio.channels.ClosedChannelException;
+import java.nio.ByteBuffer;
+import java.net.InetSocketAddress;
+import java.io.IOException;
+import java.util.Set;
+
 public class Server {
+
+    private static boolean debug;
+    private static ThreadPoolManager tpm;
+    private static ServerSocketChannel ssChannel;
+    private static Selector serverSelector;
+    //private final int clientInterestSet = SelectionKey.OP_READ | SelectionKey.OP_WRITE;
 
     public static void main(String[] args) {
 
@@ -14,11 +32,68 @@ public static void main(String[] args) {
 
 	int portNumber = Integer.parseInt(args[0]);
 	int threadPoolSize = Integer.parseInt(args[1]);
-	boolean debug = false;
+	debug = false;
 
 	if (args.length == 3)
 	    debug = true;	
+
+	tpm = new ThreadPoolManager(threadPoolSize, debug);
 
+	try {
+	    s.setupServerSocket(portNumber);
+	    serverSelector = Selector.open();
+	    s.executeServerLoop();	    
+	} catch (IOException ioe) {
+	    System.out.println(ioe.getMessage());
+	}
+			
+    }
+
+    private void setupServerSocket(int portNumber) throws IOException {
+	System.out.println("Setting up server...");
+	ssChannel = ServerSocketChannel.open();
+	ssChannel.socket().bind(new InetSocketAddress(portNumber));
+	ssChannel.configureBlocking(false);
+    }
+    
+    private void executeServerLoop() throws IOException {
+	while (true) {
+	    // Get new sockets
+	    SocketChannel socketChannel = ssChannel.accept();
+	    // Register new sockets
+	    if (socketChannel != null)
+		registerIncomingKey(socketChannel);
+	    // Find read ready channels
+	    if (serverSelector.selectNow() > 0) {
+		Set<SelectionKey> selectedKeys = serverSelector.selectedKeys();
+
+		for (SelectionKey key : selectedKeys) {		    
+		    if (key.isReadable()) {	  
+			// create and enqueue a task to read message from client
+			createTask(key);			
+			tpm.assignTask();
+		    }
+		    selectedKeys.remove(key);
+		}
+		
+	    }
+	}
+    }
+
+    private void createTask(SelectionKey key) throws IOException {
+	ServerTask task = new ServerTask(key);
+	tpm.addTaskToQueue(task);
+    }
+    
+    private void registerIncomingKey(SocketChannel socketChannel) throws IOException {
+	if (debug)
+	    System.out.println("Registering new client...");
+	try {
+	    socketChannel.configureBlocking(false);
+	    SelectionKey key = socketChannel.register(serverSelector, SelectionKey.OP_READ);
+	} catch (ClosedChannelException cce) {
+	    System.out.println(cce.getMessage());
+	}
     }
 
 }

cs455/scaling/tasks/ClientTask.java

@@ -0,0 +1,57 @@
+package cs455.scaling.tasks;
+
+import cs455.scaling.util.HashGenerator;
+import java.nio.channels.SocketChannel;
+import java.nio.ByteBuffer;
+import java.io.IOException;
+import java.util.Random;
+import java.util.LinkedList;
+
+public class ClientTask extends Thread {
+
+    private SocketChannel socketChannel;
+    private ByteBuffer buf;
+    private int messageRate;
+    private LinkedList<String> packetHashcodes;
+    private final HashGenerator hashGen;
+    
+    public ClientTask(SocketChannel socketChannel, int messageRate) {
+	this.socketChannel = socketChannel;
+	buf = ByteBuffer.allocate(8000);
+	this.messageRate = messageRate;
+	packetHashcodes = new LinkedList<String>();
+	hashGen = new HashGenerator();
+    }
+
+    public void run() {
+	while(!isInterrupted()) {
+	    // test
+	    try {
+		sendMessage();
+		Thread.sleep(1000 / messageRate);
+	    } catch (IOException ioe) {
+		System.out.println(ioe.getMessage());
+	    } catch (InterruptedException ie) {
+		System.out.println(ie.getMessage());
+	    }
+	}
+    }
+    
+    private byte[] generateRandomBytes() {
+	byte[] rBytes = new byte[8000];
+	new Random().nextBytes(rBytes);
+	return rBytes;
+    }
+
+    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);
+    }
+}

cs455/scaling/tasks/ServerTask.java

@@ -0,0 +1,49 @@
+package cs455.scaling.tasks;
+
+import cs455.scaling.util.HashGenerator;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.SocketChannel;
+import java.nio.ByteBuffer;
+import java.io.IOException;
+
+public class ServerTask implements Runnable {
+
+    private final HashGenerator hashGen;
+    private SocketChannel socketChannel;
+    private ByteBuffer buf;
+    
+    public ServerTask(SelectionKey key) {
+	hashGen = new HashGenerator();
+	socketChannel = (SocketChannel) key.channel();
+	buf = ByteBuffer.allocate(8000);
+    }
+
+    public void run() {
+	try {
+	    int bytesRead = socketChannel.read(buf);
+	    byte[] message = new byte[bytesRead];
+	    buf.flip();
+	    buf.get(message);
+	    String messageHash = getHash(message);
+	    System.out.println(messageHash);
+	    buf.clear();
+	    //replyWithHash(messageHash);
+	} catch (IOException ioe) {
+	    System.out.println(ioe.getMessage());
+	}       
+    }
+
+    private String getHash(byte[] message) {
+	return hashGen.SHA1FromBytes(message);
+    }
+
+    private void replyWithHash(String messageHash) throws IOException {
+	byte[] replyMessage = messageHash.getBytes();
+	buf.put(replyMessage);
+	buf.flip();
+	
+	while(buf.hasRemaining())
+	    socketChannel.write(buf);
+    }
+    
+}

cs455/scaling/threadpool/ThreadPoolManager.java

@@ -16,10 +16,32 @@ public ThreadPoolManager(int threadPoolSize, boolean debug) {
 	this.taskQueue = new LinkedList<Runnable>();
 	this.threadPool = new FixedThreadPool(threadPoolSize, debug);
     }
+
+    public void addTaskToQueue(Runnable task) {
+	taskQueue.add(task);       
+    }
+
+    public void assignTask() {
+	synchronized(taskQueue) {
+	    WorkerThread thread = threadPool.retrieveSpareWorker();
+	    Runnable task = taskQueue.pollFirst();
+	    if (task == null) {
+		if (debug)
+		    System.out.println("Task Queue is empty, waiting for tasks.");
+		    try {
+			taskQueue.wait();
+			task = taskQueue.removeFirst();
+		    } catch (InterruptedException ie) {
+			System.out.println(ie.getMessage());
+		}
+	    }
+	    thread.assignTask(task);
+	}
+    }
 
     public static void main(String[] args) {
 
-    //TODO: partial implementation
+    // partial implementation
     // Note: this component can be created and tested in isolation from the rest of the system
     // should allocate a given number of threads;
     // maintain a queue of pending tasks;

cs455/scaling/util/HashGenerator.java

@@ -0,0 +1,25 @@
+package cs455.scaling.util;
+
+import java.security.MessageDigest;
+import java.math.BigInteger;
+import java.security.NoSuchAlgorithmException;
+
+public class HashGenerator {
+
+    private MessageDigest digest = null;
+    
+    public HashGenerator() {
+	try {
+	    digest = MessageDigest.getInstance("SHA1");
+	} catch (NoSuchAlgorithmException nsae) {
+	    System.out.println(nsae.getMessage());
+	}
+    }
+
+    public String SHA1FromBytes(byte[] data) {
+	byte[] hash = digest.digest(data);
+	BigInteger hashInt = new BigInteger(1, hash);
+	return hashInt.toString(16);
+    }
+    
+}

makefile

@@ -4,7 +4,8 @@ PACKAGES = \
 	cs455/scaling/server \
 	cs455/scaling/client \
 	cs455/scaling/threadpool \
-	cs455/scaling/tasks
+	cs455/scaling/tasks \
+	cs455/scaling/util
 
 JC = javac
 JVM = java

setup

@@ -6,9 +6,9 @@ test_home=Classes/Spring_18/CS455/HW2-PC/CS455-HW2-PC
 
 for i in `cat machine_list`
 do
-    for value in {1..7}
+    for value in {1..2}
     do
 	echo 'logging into '${i}
-	gnome-terminal -- bash -c "ssh -t ${i} 'cd ${test_home};java cs455.scaling.client.Client richmond 4444 4;bash;'" &
+	gnome-terminal -- bash -c "ssh -t ${i} 'cd ${test_home};java cs455.scaling.client.Client richmond 4444 1 d;bash;'" &
     done
 done