Modifications made so that only one connection per Dstore needs to open

Controller.java

@@ -83,52 +83,6 @@ public class Controller {
 		}
 	}
 	
-	protected class SyncList extends ArrayList<Integer> {
-		public SyncList() {
-			super();
-		}
-		
-		@Override
-		public boolean add(Integer i) {
-			synchronized(this) {
-				return super.add(i);
-			}
-		}
-		
-		@Override
-		public boolean addAll(Collection<? extends Integer> c) {
-			synchronized(this) {
-				return super.addAll(c);
-			}
-		}
-		
-		@Override
-		public Integer get(int i) {
-			synchronized(this) {
-				return super.get(i);
-			}
-		}
-		
-		@Override
-		public int size() {
-			synchronized(this) {
-				return super.size();
-			}
-		}
-		
-		public Integer remove(int i) {
-			synchronized(this) {
-				return super.remove(i);
-			}
-		}
-		
-		public boolean remove(Integer i) {
-			synchronized(this) {
-				return super.remove(i);
-			}
-		}
-	}
-	
 	protected class Reloader {
 		public boolean reload;
 		public Reloader() {
@@ -136,7 +90,7 @@ public class Controller {
 		}
 	}
 	
-	protected List<Integer> dstores;
+	protected Map<Integer,Socket> dstores;
 	protected Map<Integer,String[]> rebalanceMessages;
 	protected Map<String,IndexEntry> index;
 	
@@ -145,7 +99,7 @@ public class Controller {
 		this.rFactor = rFactor;
 		this.timeout = timeout;
 		this.rebalancePeriod = rebalancePeriod;
-		dstores = new SyncList();
+		dstores = Collections.synchronizedMap(new HashMap<Integer,Socket>());
 		index = new HashMap<String,IndexEntry>();
 	}
 	
@@ -201,16 +155,13 @@ public class Controller {
 	
 	void handleMessage(String[] message, Socket client) throws Exception {
 		if(message[0].equals("JOIN")) {
-			dstores.add(Integer.parseInt(message[1]));
+			dstores.add(Integer.parseInt(message[1]), client);
 			System.out.println("Dstore at " + message[1] + " joined");
 			rebalance();
 		}
 		else if(message[0].equals("STORE")) {
 			store(client, message[1]);
 		}
-		else if(message[0].equals("STORE_ACK")) {
-			storeAck(client, message[1]);
-		}
 		else if(message[0].equals("LOAD")) {
 			load(client, message[1]);
 		}
@@ -260,10 +211,19 @@ public class Controller {
 				//Send STORE_TO message
 				PrintWriter out = new PrintWriter(client.getOutputStream());
 				out.print("STORE_TO");
+				out.flush();
 				for(int port : storesToStore) {
 					out.print(" " + port);
+					new Thread(() -> {
+						String[] message = dstores.get(Integer.valueOf(port)).receive().split(" ");
+						if(message[0].equals("STORE_ACK")) {
+							storeAck(Integer.valueOf(port), message[1]);
+						}
+						else {
+							//Log error
+						}
+					}).start();
 				}
-				out.flush();
 				
 				//Wait for STORE_ACKs from datastores in storesToStore
 				try {
@@ -291,16 +251,14 @@ public class Controller {
 		}).start();
 	}
 	
-	void storeAck(Socket client, String filename) throws Exception {
-		new Thread(() -> {
+	void storeAck(Integer port, String filename) throws Exception {
 		if(!index.containsKey(filename)) {
 			//Throw logging exception
 			return;
 		}
 			
 		IndexEntry thisEntry = index.get(filename);
-			thisEntry.addStoredBy(Integer.valueOf(client.getPort()));
-		}).start();
+		thisEntry.addStoredBy(port);
 	}
 	
 	void load(Socket client, String filename) throws Exception {
@@ -380,13 +338,16 @@ public class Controller {
 				entry.status = "remove in progress";
 				
 				//Send REMOVE message to all Dstores storing the file
-				for(Integer dstore : entry.storedBy) {
-					Socket socket = new Socket(InetAddress.getLocalHost(), dstore.intValue());
-					PrintWriter out = new PrintWriter(socket.getOutputStream());
-					out.write("REMOVE " + filename);
-					out.flush();
-					out.close();
-					socket.close();
+				for(Integer dstore : entry.getStoredBy()) {
+					new Thread(() -> {
+						String[] message = dstores.get(dstore).sendAndReceive("REMOVE").split(" ");
+						if(message[0].equals("REMOVE_ACK") && message[1].equals(filename)) {
+							entry.removeStoredBy(dstore.intValue());
+						}
+						else {
+							//Log error
+						}
+					}).start();
 				}
 				
 				//Wait for REMOVE_ACKs from all Dstores which were sent the REMOVE message
@@ -440,13 +401,11 @@ public class Controller {
 			rebalanceMessages = dstoreFiles;
 			try {
 				//Send LIST message to each Dstore and receive their file list
-				for(Integer dstore : dstores) {
-					Socket socket = new Socket(InetAddress.getLocalHost(), dstore.intValue());
-					PrintWriter out = new PrintWriter(socket.getOutputStream());
-					out.write("LIST");
-					out.flush();
-					out.close();
-					socket.close();
+				for(Integer dstore : dstores.keySet()) {
+					new Thread(() -> {
+						String[] message = dstores.get(dstore).sendAndReceive("LIST").split(" ");
+						receiveDstoreList(dstore.intValue(), message);
+					}).start();
 				}
 				
 				dstoreFiles.wait(timeout);
@@ -458,11 +417,41 @@ public class Controller {
 				  //Each file appears rFactor times
 				  //Each file appears at most once on each datastore
 				  //Files are evenly distributed (Dstores differ in capacity by at most 1, no 2 datastores have identical file lists)
+				List<Integer> storeOrder = reshuffle(dstoreFiles.keySet());
+				List<String> fileList = new ArrayList<String>();
 				for(Integer i : reshuffle(dstoreFiles.keySet())) {
+					for(String s : dstoreFiles.get(i)) {
+						if(!fileList.contains(s)) {
+							fileList.add(s);
+						}
+					}
+				}
 				
+				Map<Integer,List<String>> newAlloc = new HashMap<Integer,String[]>();
+				int pos = 0;
+				int storeSize = Math.ceiling((fileList.size() * rFactor) / dstores.size());
+				for(Integer i : dstoreFiles.keySet()) {
+					newAlloc.add(i, new ArrayList<String>(storeSize));
+				}
+				for(String file : fileList) {
+					for(int j=0; j<rFactor; j++) {
+						newAlloc.get(pos).add(file);
+						pos ++;
+						if(pos >= newAlloc.size()) pos = 0;
+					}
 				}
 				
 				//Make a (files to send, files to remove) pair for each Dstore
+				/*
+				Map<Integer,String> outMessages = new HashMap<Integer,String>();
+				for(Integer dstore : storeOrder) {
+					String[] oldFiles = dstoreFiles.get(dstore);
+					List<String> newFiles = newAlloc.get(dstore);
+					for(String file : oldFiles) {
+						if(!newFiles.)
+					}
+				}
+				*/
 				
 				//Send the respective REBALANCE message to each Dstore
 				
@@ -471,20 +460,29 @@ public class Controller {
 			catch(IOException e) {
 				e.printStackTrace();
 			}
+			catch(Exception e) {
+				e.printStackTrace();
+			}
 			finally {
 				rebalanceMessages = null;
 			}
 		}).start();
 	}
 	
-	void receiveDstoreList(Socket client, String[] list) {
-		new Thread(() -> {
+	void receiveDstoreList(int port, String[] list) {
 		if(rebalanceMessages == null) return;
-			rebalanceMessages.add(Integer.valueOf(client.getPort()), list);
+		
+		for(String file : list) {
+			if(!index.containsKey(file)) {
+				//Log error
+				return; //Throw exception?
+			}
+		}
+		
+		rebalanceMessages.add(port, list);
 		if(rebalanceMessages.size() == dstores.size()) {
 			rebalanceMessages.notify();
 		}
-		}).start();
 	}
 	
 	List<Integer> reshuffle(Collection<Integer> col) {

Dstore.java

@@ -11,12 +11,17 @@ public class Dstore {
 	protected int cport; //Controller's port to talk to
 	protected int timeout; //in milliseconds
 	protected String fileFolder; //Where to store the data locally
+	protected Map<String,Integer> fileSizes;
+	
+	protected BufferedReader controllerIn;
+	protected PrintWriter controllerOut;
 	
 	public Dstore(int port, int cport, int timeout, String fileFolder) {
 		this.port = port;
 		this.cport = cport;
 		this.timeout = timeout;
 		this.fileFolder = fileFolder;
+		fileSizes = new HashMap<String,Integer>();
 	}
 	
 	public static void main(String[] args) {
@@ -42,10 +47,22 @@ public class Dstore {
 	public void start() {
 		try {
 			Socket socket = new Socket(InetAddress.getLocalHost(), cport);
-			BufferedWriter out = new BufferedWriter(new OutputStreamWriter(socket.getOutputStream()));
-			out.write("JOIN " + port);
-			out.close();
-			socket.close();
+			controllerIn = new BufferedReader(new InputStreamReader(socket.getInputStream()));
+			controllerOut = new PrintWriter(socket.getOutputStream());
+			controllerOut.print("JOIN " + port);
+			controllerOut.flush();
+			
+			new Thread(() -> {
+				while(true) {
+					try {
+						String[] message = controllerIn.readLine().split(" ");
+						handleMessage(message, socket, controllerIn);
+					}
+					catch(Exception e) {
+						e.printStackTrace();
+					}
+				}
+			}).start();
 			
 			ServerSocket server = new ServerSocket(port);
 			
@@ -76,13 +93,13 @@ public class Dstore {
 			load(client, message[1]);
 		}
 		else if(message[0].equals("REMOVE")) {
-			remove(client, message[1]);
+			remove(message[1]);
 		}
 		else if(message[0].equals("LIST")) {
-			list(client);
+			list();
 		}
 		else if(message[0].equals("REBALANCE")) {
-			rebalance(client, message);
+			rebalance(message);
 		}
 		else {
 			//Log error and continue (throw exception?)
@@ -115,6 +132,9 @@ public class Dstore {
 				controllerOut.print("STORE_ACK " + filename);
 				controllerOut.flush();
 				controllerOut.close();
+				
+				if(fileSizes.containsKey(filename)) fileSizes.remove(filename);
+				fileSizes.add(filename, filesize);
 			}
 			catch(IOException e) {
 				e.printStackTrace();
@@ -153,24 +173,22 @@ public class Dstore {
 		}).start();
 	}
 	
-	void remove(Socket client, String filename) throws Exception {
+	void remove(String filename) throws Exception {
 		new Thread(() -> {
 			try {
 				//Remove the file from fileFolder
 				Path path = new File(fileFolder + "/" + filename).toPath();
-				PrintWriter out = new PrintWriter(client.getOutputStream());
 				
 				if(Files.deleteIfExists(path)) {
 					//Send REMOVE_ACK message to client (the controller)
-					out.print("REMOVE_ACK");
+					controllerOut.print("REMOVE_ACK");
 				}
 				else {
 					//Send DOES NOT EXIST error
-					out.print("ERROR DOES_NOT_EXIST " + filename);
+					controllerOut.print("ERROR DOES_NOT_EXIST " + filename);
 				}
 				
-				out.flush();
-				out.close();
+				controllerOut.flush();
 			}
 			catch(IOException e) {
 				e.printStackTrace();
@@ -178,16 +196,14 @@ public class Dstore {
 		}).start();
 	}
 	
-	void list(Socket client) throws Exception {
+	void list() throws Exception {
 		new Thread(() -> {
 			try {
 				//Send a list of all files in fileFolder to client (the controller)
-				PrintWriter out = new PrintWriter(client.getOutputStream());
 				for(File file : new File(fileFolder).listFiles()) {
-					out.print(file.getName());
-					out.flush();
+					controllerOut.print(file.getName());
+					controllerOut.flush();
 				}
-				out.close();
 			}
 			catch(IOException e) {
 				e.printStackTrace();
@@ -195,7 +211,7 @@ public class Dstore {
 		}).start();
 	}
 	
-	void rebalance(Socket client, String[] message) throws Exception {
+	void rebalance(String[] message) throws Exception {
 		new Thread(() -> {
 			try {
 				//Interpret files to send and files to remove from the message
@@ -232,7 +248,34 @@ public class Dstore {
 				//Send each file to send to the Dstore at the specified port number
 				for(String filename : filesToSend.keySet()) {
 					for(Integer dstore : filesToSend.get(filename)) {
-						//Same store functions as used in the client object
+						try {
+							Socket socket = new Socket(InetAddress.getLocalHost(), dstore.intValue());
+							PrintWriter out = new PrintWriter(socket.getOutputStream());
+							out.write("STORE " + filename + " " + fileSizes.get(filename));
+							out.flush();
+							
+							BufferedReader in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
+							if(!in.readLine().equals("ACK")) {
+								//Log error
+							}
+							String line;
+							while(line = new BufferedReader(new FileInputStream(fileFolder + "/" + filename))) {
+								if(line == null) {
+									out.write("");
+									out.flush();
+									break;
+								}
+								else {
+									out.write(line);
+									out.flush();
+								}
+							}
+							out.close();
+							socket.close();
+						}
+						catch(IOException e) {
+							e.printStackTrace();
+						}
 					}
 				}
 				
@@ -242,10 +285,8 @@ public class Dstore {
 				}
 				
 				//Send REBALANCE_COMPLETE message to client (the controller)
-				PrintWriter out = new PrintWriter(client.getOutputStream());
-				out.print("REBALANCE COMPLETE");
-				out.flush();
-				out.close();
+				controllerOut.print("REBALANCE COMPLETE");
+				controllerOut.flush();
 			}
 			catch(IOException e) {
 				e.printStackTrace();

DstoreConnection.java

+import java.io.*;
+import java.net.*;
+import java.lang.Runnable;
+import java.util.List;
+import java.util.ArrayList;
+import java.util.Map;
+import java.util.HashMap;
+import java.util.Collection;
+
+public class DstoreConnection {
+	protected Socket socket;
+	protected BufferedReader reader;
+	protected PrintWriter writer;
+	
+	public DstoreConnection(Socket socket) {
+		this.socket = socket;
+		reader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
+		writer = new PrintWriter(socket.getOutputStream());
+	}
+	
+	public String sendAndReceive(String message) {
+		try {
+			writer.print(message);
+			writer.flush();
+		}
+		catch(IOException e) {
+			e.printStackTrace();
+			return "";
+		}
+		return receive();
+	}
+	
+	public String receive() {
+		try {
+			String returnMessage = reader.readLine();
+			return returnMessage;
+		}
+		catch(IOException e) {
+			e.printStackTrace();
+			return "";
+		}
+	}
+}