Solved problem with having multiple dstores connect. Problem with receiving...

Solved problem with having multiple dstores connect. Problem with receiving messages not solved though.

ClientMain.java

@@ -45,7 +45,9 @@ public class ClientMain {
 			String[] files = {"AllStar.txt", "Unknown.txt", "PumpkinHill.txt", "SnowHalation.txt", "Grandad.txt"};
 			
 			for(String file : files) {
-				try { client.store(new File(file)); } catch(IOException e) { e.printStackTrace(); }
+				try { client.store(new File(file)); Thread.sleep(500);} 
+				catch(IOException e) { e.printStackTrace(); }
+				catch(InterruptedException e) {e.printStackTrace();}
 			}
 			
 			/*

Controller.java

@@ -17,7 +17,7 @@ public class Controller {
 	protected int rebalancePeriod; //How long to wait to start the next rebalance operation, in milliseconds
 	
 	protected class IndexEntry {
-		protected int filesize;
+		protected long filesize;
 		protected List<Integer> storedBy;
 		protected int numberToStore;
 		protected String status;
@@ -29,11 +29,11 @@ public class Controller {
 			status = "store in progress";
 		}
 		
-		public synchronized void setFilesize(int filesize) {
+		public synchronized void setFilesize(long filesize) {
 			this.filesize = filesize;
 		}
 		
-		public synchronized int getFilesize() {
+		public synchronized long getFilesize() {
 			return filesize;
 		}
 		
@@ -74,12 +74,17 @@ public class Controller {
 		}
 	}
 	
+	protected class RebalanceMessages {
+		public Map<Integer,List<String>> dstoreFiles;
+		public RebalanceMessages() {dstoreFiles = null;}
+	}
+	
 	protected class Reloader extends ArrayList<Integer> {
-		public int filesize;
+		public long filesize;
 	}
 	
 	protected Map<Integer,DstoreConnection> dstores;
-	protected Map<Integer,List<String>> rebalanceMessages;
+	protected RebalanceMessages rebalanceMessages;
 	protected Map<String,IndexEntry> index;
 	protected Map<Socket,Reloader> loadRequests;
 	
@@ -89,6 +94,7 @@ public class Controller {
 		this.timeout = timeout;
 		this.rebalancePeriod = rebalancePeriod;
 		dstores = Collections.synchronizedMap(new HashMap<Integer,DstoreConnection>());
+		rebalanceMessages = new RebalanceMessages();
 		index = Collections.synchronizedMap(new HashMap<String,IndexEntry>());
 		loadRequests = Collections.synchronizedMap(new HashMap<Socket,Reloader>());
 	}
@@ -147,9 +153,11 @@ public class Controller {
 								e.printStackTrace();
 							}
 							
-							while(!client.isClosed()) {
+							String clientMessage = "";
+							do {
 								try {
-									String clientMessage = in.readLine();
+									clientMessage = in.readLine();
+									System.out.println(clientMessage);
 									if(clientMessage != null) {
 										handleMessage(clientMessage.split(" "), client);
 									}
@@ -158,13 +166,14 @@ public class Controller {
 									e.printStackTrace();
 								}
 							}
+							while(clientMessage != null);
 							System.out.println("Client closed");
 						}).start();
 					}
 				}
 				catch(Exception e) {
 					//Log error
-					e.printStackTrace();
+					System.out.println("Controller error while accepting connections!");
 					System.out.println("Continue...");
 				}
 			}
@@ -213,19 +222,22 @@ public class Controller {
 		}
 		else {
 			//Log error
+			System.out.println("Malformed message received by Controller");
 		}
 	}
 	
 	void store(Socket client, String filename, String filesizeString) throws Exception {
-		int filesize = -1;
+		long filesize = -1;
 		try {
-			filesize = Integer.parseInt(filesizeString);
+			filesize = Long.parseLong(filesizeString);
 			if(filesize < 1) {
 				//Log error
+				System.out.println("A client is trying to store a file with size < 1");
 			}
 		}
 		catch(NumberFormatException e) {
 			//Log error
+			System.out.println("Client has not provided an integer as a filesize");
 		}
 		
 		try {
@@ -258,21 +270,24 @@ public class Controller {
 				message = message + " " + thisStore.intValue();
 				new Thread(() -> {
 					try {
-						String[] receivedMessage = dstores.get(thisStore).receive().split(" ");
+						String[] receivedMessage = dstores.get(thisStore).receive("STORE_ACK").split(" ");
 						if(receivedMessage[0].equals("STORE_ACK")) {
 							try {
 								storeAck(thisStore, receivedMessage[1]);
 							}
 							catch(Exception e) {
 								//Log error
+								System.out.println("Error processing store ack from dstore " + thisStore);
 							}
 						}
 						else {
 							//Log error
+							System.out.println("Dstore " + thisStore + " should have sent STORE_ACK but Controller received " + receivedMessage[0]);
 						}
 					}
 					catch(NullPointerException e) {
-						removeDstore(thisStore);
+						e.printStackTrace();
+						//removeDstore(thisStore);
 					}
 				}).start();
 			}
@@ -291,6 +306,7 @@ public class Controller {
 			
 			if(entry.getStoredBy().size() < rFactor) {
 				//Log error
+				System.out.println("Not all STORE_ACKs have been received");
 			}
 			
 			//Update index to "store complete"
@@ -308,7 +324,7 @@ public class Controller {
 	void storeAck(Integer port, String filename) throws Exception {
 		if(!index.containsKey(filename)) {
 			//Throw logging exception
-			return;
+			throw new Exception("Index does not contain " + filename);
 		}
 			
 		IndexEntry thisEntry = index.get(filename);
@@ -381,16 +397,18 @@ public class Controller {
 			for(Integer dstore : entry.getStoredBy()) {
 				new Thread(() -> {
 					try {
-						String[] message = dstores.get(dstore).sendAndReceive("REMOVE " + filename).split(" ");
+						String[] message = dstores.get(dstore).sendAndReceive("REMOVE " + filename, "REMOVE_ACK").split(" ");
 						if(message[0].equals("REMOVE_ACK") && message[1].equals(filename)) {
 							entry.removeStoredBy(dstore.intValue());
 						}
 						else {
 							//Log error
+							System.out.println("Dstore " + dstore + " should have sent REMOVE_ACK but Controller received " + message[0]);
 						}
 					}
 					catch(NullPointerException e) {
-						removeDstore(dstore);
+						e.printStackTrace();
+						//removeDstore(dstore);
 					}
 				}).start();
 			}
@@ -407,6 +425,7 @@ public class Controller {
 			
 			if(entry.getStoredBy().size() > 0) {
 				//Log error
+				System.out.println("Not all REMOVE_ACKs have been received");
 			}
 			
 			//Update index to "remove complete"
@@ -442,10 +461,12 @@ public class Controller {
 	}
 	
 	void rebalance() throws Exception {
-		if(rebalanceMessages != null) return;
+		boolean success = true;
 		Map<Integer,List<String>> dstoreFiles = new HashMap<Integer,List<String>>();
-		synchronized(dstoreFiles) {
-			rebalanceMessages = dstoreFiles;
+		
+		synchronized(rebalanceMessages) {
+			if(rebalanceMessages.dstoreFiles != null) return;
+			rebalanceMessages.dstoreFiles = dstoreFiles;
 			try {
 				//Send LIST message to each Dstore and receive their file list
 				for(Integer dstore : dstores.keySet()) {
@@ -457,14 +478,16 @@ public class Controller {
 							receiveDstoreList(dstore.intValue(), message);
 						}
 						catch(NullPointerException e) {
-							removeDstore(dstore);
+							e.printStackTrace();
+							//removeDstore(dstore);
 						}
 					}).start();
 				}
 				
-				dstoreFiles.wait(timeout);
+				rebalanceMessages.wait(timeout);
 				if(dstoreFiles.size() < dstores.size()) {
 					//Log error
+					System.out.println("Not all file lists have been received");
 				}
 				
 				//Create a new file allocation so that:
@@ -560,9 +583,10 @@ public class Controller {
 					String finalMessage = message;
 					new Thread(() -> {
 						try {
-							String returnMessage = dstores.get(thisStore).sendAndReceive(finalMessage);
+							String returnMessage = dstores.get(thisStore).sendAndReceive(finalMessage, "REBALANCE_COMPLETE");
 							if(!returnMessage.equals("REBALANCE_COMPLETE")) {
 								//Log error
+								System.out.println("Dstore " + thisStore + " should have sent REBALANCE_COMPLETE but Controller received " + returnMessage);
 							}
 							synchronized(acksReceived) {
 								acksReceived.incr();
@@ -572,7 +596,8 @@ public class Controller {
 							}
 						}
 						catch(NullPointerException e) {
-							removeDstore(thisStore);
+							e.printStackTrace();
+							//removeDstore(thisStore);
 						}
 					}).start();
 				}
@@ -584,6 +609,8 @@ public class Controller {
 						acksReceived.wait(timeout);
 						if(acksReceived.getValue() < storeOrder.size()) {
 							//Restart rebalance operation
+							System.out.println("Not all REBALANCE_COMPLETEs received. Restarting rebalance operation...");
+							success = false;
 						}
 					}
 					catch(InterruptedException e) {
@@ -595,14 +622,19 @@ public class Controller {
 				e.printStackTrace();
 			}
 			finally {
-				rebalanceMessages = null;
+				rebalanceMessages.dstoreFiles = null;
+				System.out.println("There are " + dstores.size() + " dstores connected");
+				for(String i : index.keySet()) {
+					System.out.print(i);
 				}
+				System.out.print("\n");
 			}
 		}
 		
-	void receiveDstoreList(int port, String[] list) {
-		if(rebalanceMessages == null) return;
+		if(!success) rebalance();
+	}
 	
+	void receiveDstoreList(int port, String[] list) {
 		List<String> toList = new ArrayList<String>();
 		if(!list[0].equals("ERROR_EMPTY")) {
 			for(String file : list) {
@@ -617,8 +649,10 @@ public class Controller {
 		}
 		
 		synchronized(rebalanceMessages) {
-			rebalanceMessages.put(port, toList);
-			if(rebalanceMessages.size() == dstores.size()) {
+			if(rebalanceMessages.dstoreFiles == null) return;
+			
+			rebalanceMessages.dstoreFiles.put(port, toList);
+			if(rebalanceMessages.dstoreFiles.size() == dstores.size()) {
 				rebalanceMessages.notify();
 			}
 		}

Dstore.java

@@ -13,7 +13,7 @@ public class Dstore {
 	protected int cport; //Controller's port to talk to
 	protected int timeout; //in milliseconds
 	protected File fileFolder; //Where to store the data locally
-	protected Map<String,Integer> fileSizes;
+	protected Map<String,Long> fileSizes;
 	
 	protected Socket controllerSocket;
 	protected BufferedReader controllerIn;
@@ -33,7 +33,10 @@ public class Dstore {
 			if(!fileFolder.mkdir()) throw new Exception("Folder could not be created");
 		}
 		
-		fileSizes = new HashMap<String,Integer>();
+		fileSizes = new HashMap<String,Long>();
+		for(File file : fileFolder.listFiles()) {
+			fileSizes.put(file.getName(), Long.valueOf(file.length()));
+		}
 	}
 	
 	public static void main(String[] args) {
@@ -108,7 +111,7 @@ public class Dstore {
 	
 	void handleMessage(String[] message, Socket client) throws Exception {
 		if(message[0].equals("STORE")) {
-			store(client, message[1], Integer.parseInt(message[2]));
+			store(client, message[1], Long.parseLong(message[2]));
 		}
 		else if(message[0].equals("LOAD_DATA")) {
 			load(client, message[1]);
@@ -124,10 +127,11 @@ public class Dstore {
 		}
 		else {
 			//Log error and continue (throw exception?)
+			System.out.println("Dstore " + port + " has received a malformed message");
 		}
 	}
 	
-	void store(Socket client, String filename, int filesize) throws Exception {
+	void store(Socket client, String filename, long filesize) throws Exception {
 		new Thread(() -> {
 			try {
 				//Send ACK message to client
@@ -151,7 +155,7 @@ public class Dstore {
 				controllerOut.flush();
 				
 				if(fileSizes.containsKey(filename)) fileSizes.remove(filename);
-				fileSizes.put(filename, filesize);
+				fileSizes.put(filename, Long.valueOf(filesize));
 			}
 			catch(IOException e) {
 				e.printStackTrace();
@@ -284,8 +288,10 @@ public class Dstore {
 							out.flush();
 							
 							BufferedReader in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
-							if(!in.readLine().equals("ACK")) {
+							String receivedMessage = in.readLine();
+							if(!receivedMessage.equals("ACK")) {
 								//Log error
+								System.out.println("Dstore " + dstore + " should have sent ACK but " + port + " received " + receivedMessage);
 							}
 							
 							byte[] content = new byte[8];

DstoreConnection.java

@@ -8,10 +8,13 @@ import java.util.HashMap;
 import java.util.Collection;
 
 public class DstoreConnection {
+	protected final int MAX_QUEUE_SIZE = 50;
+	
 	protected Socket socket;
 	protected BufferedReader reader;
 	protected PrintWriter writer;
 	protected boolean available;
+	protected List<String> queue;
 	
 	public DstoreConnection(Socket socket) {
 		this.socket = socket;
@@ -19,6 +22,7 @@ public class DstoreConnection {
 			reader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
 			writer = new PrintWriter(socket.getOutputStream());
 			available = true;
+			queue = new ArrayList<String>();
 		}
 		catch(IOException e) {
 			e.printStackTrace();
@@ -30,7 +34,7 @@ public class DstoreConnection {
 		}
 	}
 	
-	public String sendAndReceive(String message) throws NullPointerException {
+	public String sendAndReceive(String message, String expectedMessage) throws NullPointerException {
 		System.out.println("Getting lock...");
 		synchronized(this) {
 			try {
@@ -38,7 +42,8 @@ public class DstoreConnection {
 				if(!available) return "ERROR";
 				writer.println(message);
 				writer.flush();
-				return localReceive();
+				System.out.println("Controller sent " + message);
+				return localReceive(expectedMessage);
 			}
 			catch(NullPointerException e) {
 				System.out.println("Dstore disconnected");
@@ -48,21 +53,50 @@ public class DstoreConnection {
 		}
 	}
 	
-	public String receive() throws NullPointerException {
+	public String sendAndReceive(String message) throws NullPointerException {
+		return sendAndReceive(message, null);
+	}
+	
+	public String receive(String expectedMessage) throws NullPointerException {
 		System.out.println("Getting lock...");
 		synchronized(this) {
 			System.out.println("Lock acquired");
 			if(!available) return "ERROR";
-			return localReceive();
+			
+			//Check the queue for the message before trying to receive any new messages (if no expected message is specified, return the head of the queue)
+			for(int i=0; i<queue.size(); i++) {
+				String message = queue.get(i);
+				if(expectedMessage == null || message.equals(expectedMessage)) {
+					queue.remove(message);
+					return message;
+				}
+			}
+			
+			return localReceive(expectedMessage);
 		}
 	}
 	
-	protected String localReceive() throws NullPointerException {
+	public String receive() throws NullPointerException {
+		return receive(null);
+	}
+	
+	protected String localReceive(String expectedMessage) throws NullPointerException {
 		try {
-			String returnMessage = "";
-			while(returnMessage.equals("")) {
+			String returnMessage = null;
+			do {
 				returnMessage = reader.readLine();
+				if(returnMessage == null) {
+					System.out.println("Dstore disconnected");
+					available = false;
+					throw new NullPointerException();
+				}
+				if(expectedMessage != null && !returnMessage.equals(expectedMessage)) {
+					queue.add(returnMessage);
+					if(queue.size() > MAX_QUEUE_SIZE) queue.remove(0);
+					returnMessage = null;
+				}
 			}
+			while(returnMessage == null);
 			System.out.println("Controller received " + returnMessage);
 			return returnMessage;
 		}
@@ -70,10 +104,5 @@ public class DstoreConnection {
 			e.printStackTrace();
 			return "";
 		}
-		catch(NullPointerException e) {
-			System.out.println("Dstore disconnected");
-			available = false;
-			throw new NullPointerException();
-		}
 	}
 }

client_1618914098636.log

-Cannot connect to the Controller on port 8082

client_1618914115362.log

-Connection established to port 8080
-Message sent to port 8080: LIST
-List operation started
-Message received from port 8080: ERROR
-ERROR: Connection closed by the Controller
-List operation failed
-ERROR: File to store does not exist (absolute path: /home/danimal/Documents/comp2207-distributed-filesystem/Clipboard01.pdf)
-ERROR: File to store does not exist (absolute path: /home/danimal/Documents/comp2207-distributed-filesystem/Clipboard01.pdf)
-ERROR: File to store does not exist (absolute path: /home/danimal/Documents/comp2207-distributed-filesystem/Clipboard01.jpg)
-Message sent to port 8080: LIST
-List operation started
-Message received from port 8080: null
-ERROR: Connection closed by the Controller
-List operation failed
-Message sent to port 8080: LIST
-List operation started
-Message received from port 8080: null
-ERROR: Connection closed by the Controller
-List operation failed

client_1618916214881.log

-Connection established to port 8080
-Message sent to port 8080: LIST
-List operation started
-Message received from port 8080: null
-ERROR: Connection closed by the Controller
-List operation failed
-ERROR: File to store does not exist (absolute path: /home/danimal/Documents/comp2207-distributed-filesystem/Clipboard01.pdf)
-ERROR: File to store does not exist (absolute path: /home/danimal/Documents/comp2207-distributed-filesystem/Clipboard01.pdf)
-ERROR: File to store does not exist (absolute path: /home/danimal/Documents/comp2207-distributed-filesystem/Clipboard01.jpg)
-Message sent to port 8080: LIST
-List operation started
-Message received from port 8080: null
-ERROR: Connection closed by the Controller
-List operation failed
-Message sent to port 8080: LIST
-List operation started
-Message received from port 8080: null
-ERROR: Connection closed by the Controller
-List operation failed

client_1618917382748.log

-Connection established to port 8080
-Message sent to port 8080: LIST
-List operation started
-Message received from port 8080: null
-ERROR: Connection closed by the Controller
-List operation failed
-ERROR: File to store does not exist (absolute path: /home/danimal/Documents/comp2207-distributed-filesystem/Clipboard01.pdf)
-ERROR: File to store does not exist (absolute path: /home/danimal/Documents/comp2207-distributed-filesystem/Clipboard01.pdf)
-ERROR: File to store does not exist (absolute path: /home/danimal/Documents/comp2207-distributed-filesystem/Clipboard01.jpg)
-Message sent to port 8080: LIST
-List operation started
-Message received from port 8080: null
-ERROR: Connection closed by the Controller
-List operation failed
-Message sent to port 8080: LIST
-List operation started
-Message received from port 8080: null
-ERROR: Connection closed by the Controller
-List operation failed

client_1618917653577.log

-Connection established to port 8080
-Message sent to port 8080: LIST
-List operation started
-Timeout expired while reading from port 8080
-List operation failed
-ERROR: File to store does not exist (absolute path: /home/danimal/Documents/comp2207-distributed-filesystem/Clipboard01.pdf)
-ERROR: File to store does not exist (absolute path: /home/danimal/Documents/comp2207-distributed-filesystem/Clipboard01.pdf)
-ERROR: File to store does not exist (absolute path: /home/danimal/Documents/comp2207-distributed-filesystem/Clipboard01.jpg)
-Message sent to port 8080: LIST
-List operation started
-Timeout expired while reading from port 8080
-List operation failed
-Message sent to port 8080: LIST
-List operation started
-Timeout expired while reading from port 8080
-List operation failed

client_1618917705050.log

-Connection established to port 8080
-Message sent to port 8080: LIST
-List operation started
-Timeout expired while reading from port 8080
-List operation failed
-ERROR: File to store does not exist (absolute path: /home/danimal/Documents/comp2207-distributed-filesystem/Clipboard01.pdf)
-ERROR: File to store does not exist (absolute path: /home/danimal/Documents/comp2207-distributed-filesystem/Clipboard01.pdf)
-ERROR: File to store does not exist (absolute path: /home/danimal/Documents/comp2207-distributed-filesystem/Clipboard01.jpg)
-Message sent to port 8080: LIST
-List operation started
-Timeout expired while reading from port 8080
-List operation failed
-Message sent to port 8080: LIST
-List operation started
-Timeout expired while reading from port 8080
-List operation failed

client_1618918167755.log

-Connection established to port 8080
-Message sent to port 8080: LIST
-List operation started
-Timeout expired while reading from port 8080
-List operation failed
-ERROR: File to store does not exist (absolute path: /home/danimal/Documents/comp2207-distributed-filesystem/Clipboard01.pdf)
-ERROR: File to store does not exist (absolute path: /home/danimal/Documents/comp2207-distributed-filesystem/Clipboard01.pdf)
-ERROR: File to store does not exist (absolute path: /home/danimal/Documents/comp2207-distributed-filesystem/Clipboard01.jpg)
-Message sent to port 8080: LIST
-List operation started
-Timeout expired while reading from port 8080
-List operation failed
-Message sent to port 8080: LIST
-List operation started
-Timeout expired while reading from port 8080
-List operation failed