diff options
Diffstat (limited to 'src/main/java/derms/Replica4pkg/Replica4.java')
| -rw-r--r-- | src/main/java/derms/Replica4pkg/Replica4.java | 578 |
1 files changed, 578 insertions, 0 deletions
diff --git a/src/main/java/derms/Replica4pkg/Replica4.java b/src/main/java/derms/Replica4pkg/Replica4.java new file mode 100644 index 0000000..91bc303 --- /dev/null +++ b/src/main/java/derms/Replica4pkg/Replica4.java @@ -0,0 +1,578 @@ +package derms.Replica4pkg; + +import java.io.*; +import java.net.DatagramPacket; +import java.net.DatagramSocket; +import java.net.InetAddress; +import java.util.*; + +import javax.jws.WebMethod; + +import derms.Replica; +import derms.ReplicaManager; +import derms.Request; +import derms.Response; +import derms.util.TestLogger; +import derms.util.ThreadPool; + + + +public class Replica4 implements Replica { + + private ReplicaManager replicaManager; + private boolean alive = true; + private boolean byzFailure; + + private String serverName; + private int udpPort; + private Map<String, Map<String, Resource>> resourceMap; //resourceName -> (resourceID -> Resource) + private Map<String, Set<String>> coordinatorResources; //coordinatorID -> Set<resourceID> + private Map<String, Map<String, Integer>> coordinatorResourceDurations; //coordinatorID -> (resourceID -> allocatedDuration) + private Map<String, Integer> serverUdpPorts; //serverName -> UDP port + + public Replica4(ReplicaManager replicaManager){ + this.replicaManager = replicaManager; + this.resourceMap = new HashMap<>(); + this.coordinatorResources = new HashMap<>(); + this.coordinatorResourceDurations = new HashMap<>(); + startUDPListener(); + } + + @Override + public boolean isAlive() { + return alive; + } + + @Override + public void startProcess(int byzantine, int crash) { + // [TEST] Detect crash + if (crash == 1) { + alive = false; + } else { + alive = true; + } + + // [TEST] Detect byzantine failure + if (byzantine == 1) { + byzFailure = true; + } else { + byzFailure = false; + } + + System.out.println("[Replica 4] Process started."); + } + + @Override + public void processRequest(Request request) { + log(request.toString()); + + // [TEST] Simulate byzantine failure (return incorrect value) + if (byzFailure == true) { + Response response = new Response(request, replicaManager.getReplicaId(), "BYZANTINE FAILURE", false); + replicaManager.sendResponseToFE(response); + return; + } + + String status = ""; + try { + switch (request.getFunction()) { + case "addResource": + status = addResource(request.getResourceID(), request.getResourceType(), request.getDuration()); + break; + case "removeResource": + status = removeResource(request.getResourceID(), request.getDuration()); + break; + case "listResourceAvailability": + status = listResourceAvailability(request.getResourceType()); + break; + case "requestResource": + status = requestResource(request.getClientID(), request.getResourceID(), request.getDuration()); + break; + case "findResource": + status = findResource(request.getClientID(), request.getResourceType()); + break; + case "returnResource": + status = returnResource(request.getClientID(), request.getResourceID()); + break; + case "swapResource": + status = swapResource(request.getClientID(), request.getOldResourceID(), request.getOldResourceType(), request.getResourceID(), request.getResourceType()); + break; + default: + status = "Failure: unknown function '" + request.getFunction() + "'"; + } + } catch (Exception e) { + log(e.getMessage()); + status = "Failure: " + request.getFunction() + ": " + e.getMessage(); + } + + Response response = new Response(request, replicaManager.getReplicaId(), status, false); // TODO: isSuccess flag + log("Processed request " + request + "; response: " + response); + replicaManager.sendResponseToFE(response); + } + + @Override + public void restart() { + shutdown(); + + // [TEST] Restart process without byzantine failure or crash + TestLogger.log("REPLICA 2: {RESTARTED}"); + startProcess(0, 0); + } + + public void shutdown() { + log("Shutting down..."); + alive = false; + log("Finished shutting down."); + } + + @Override + public int getId() { + return 4; + } + + public synchronized void log(String message) { + String logMessage = new Date() + " - " + message; + System.out.println(logMessage); + + try (FileWriter fw = new FileWriter("Replica4_log.txt", true); + BufferedWriter bw = new BufferedWriter(fw); + PrintWriter out = new PrintWriter(bw)) { + out.println(logMessage); + } catch (IOException e) { + e.printStackTrace(); + } + } + + @WebMethod(exclude = true) + public void initReplica4(String serverName, int udpPort, Map<String, Integer> serverUdpPorts) { + this.serverName = serverName; + this.udpPort = udpPort; + this.serverUdpPorts = serverUdpPorts; + } + + //Start UDP Listener + private void startUDPListener() { + new Thread(() -> { + try (DatagramSocket socket = new DatagramSocket(udpPort)) { + byte[] buffer = new byte[4096]; + System.out.println(serverName + " UDP Server started on port " + udpPort); + + while (true) { + DatagramPacket request = new DatagramPacket(buffer, buffer.length); + socket.receive(request); + new Thread(() -> handleUDPRequest(request)).start(); + } + } catch (IOException e) { + e.printStackTrace(); + } + }).start(); + } + + //Handle UDP Request + private void handleUDPRequest(DatagramPacket requestPacket) { + try { + String requestData = new String(requestPacket.getData(), 0, requestPacket.getLength()); + String[] parts = requestData.split(":"); + String methodName = parts[0]; + String[] params = Arrays.copyOfRange(parts, 1, parts.length); + + String responseData = ""; + + switch (methodName) { + case "listResourceAvailability": + responseData = handleListResourceAvailability(params[0]); + break; + case "requestResource": + responseData = handleLocalResourceRequest(params[0], params[1], Integer.parseInt(params[2])); + break; + case "findResource": + responseData = handleFindResource(params[0], params[1]); + break; + case "returnResource": + responseData = handleReturnResource(params[0], params[1]); + break; + case "swapRequestResource": + responseData = handleSwapRequestResource(params[0], params[1], Integer.parseInt(params[2])); + break; + case "swapReturnResource": + responseData = handleSwapReturnResource(params[0], params[1]); + break; + case "checkCoordinatorResource": + responseData = checkCoordinatorHasResource(params[0], params[1]); + break; + case "getCoordinatorResourceDuration": + int duration = getCoordinatorResourceDuration(params[0], params[1]); + responseData = String.valueOf(duration); + break; + default: + responseData = "Invalid method"; + } + + //Send response + byte[] responseBytes = responseData.getBytes(); + DatagramPacket responsePacket = new DatagramPacket(responseBytes, responseBytes.length, requestPacket.getAddress(), requestPacket.getPort()); + DatagramSocket socket = new DatagramSocket(); + socket.send(responsePacket); + socket.close(); + } catch (Exception e) { + e.printStackTrace(); + } + } + + //Operation Handlers + private synchronized String handleListResourceAvailability(String resourceName) { + StringBuilder result = new StringBuilder(); + resourceName = resourceName.toUpperCase(); + Map<String, Resource> resources = resourceMap.get(resourceName); + + if (resources != null) { + for (Resource resource : resources.values()) { + if (resource.getDuration() > 0) { + result.append(resource.getResourceName()).append(" - ").append(serverName).append(" ").append(resource.getDuration()).append("\n"); + } + } + } + return result.toString(); + } + + private synchronized String handleLocalResourceRequest(String coordinatorID, String resourceID, int duration) { + for (Map<String, Resource> resources : resourceMap.values()) { + Resource resource = resources.get(resourceID); + if (resource != null) { + if (resource.getDuration() >= duration) { + resource.subDuration(duration); + coordinatorResources.computeIfAbsent(coordinatorID, k -> new HashSet<>()).add(resourceID); + coordinatorResourceDurations.computeIfAbsent(coordinatorID, k -> new HashMap<>()).put(resourceID, duration); + + log("Resource " + resourceID + " allocated to " + coordinatorID + " for duration " + duration); + return "Resource " + resourceID + " allocated successfully."; + } else { + return "Insufficient duration for resource " + resourceID; + } + } + } + return "Resource " + resourceID + " not found."; + } + + private synchronized String handleFindResource(String coordinatorID, String resourceName) { + StringBuilder result = new StringBuilder(); + resourceName = resourceName.toUpperCase(); + Set<String> resources = coordinatorResources.get(coordinatorID); + if (resources != null) { + Map<String, Integer> resourceDurations = coordinatorResourceDurations.get(coordinatorID); + if (resourceDurations != null) { + for (String resourceID : resources) { + Resource resource = getResourceByID(resourceID); + if (resource != null && resource.getResourceName().equals(resourceName)) { + Integer allocatedDuration = resourceDurations.get(resourceID); + if (allocatedDuration != null) { + result.append(resource.getResourceName()).append(" - ").append(resourceID).append(" ").append(allocatedDuration).append("\n"); + } + } + } + } + } + return result.toString(); + } + + private synchronized String handleReturnResource(String coordinatorID, String resourceID) { + Set<String> resources = coordinatorResources.get(coordinatorID); + if (resources != null && resources.contains(resourceID)) { + resources.remove(resourceID); + if (resources.isEmpty()) { + coordinatorResources.remove(coordinatorID); + } + Resource resource = getResourceByID(resourceID); + if (resource != null) { + Map<String, Integer> resourceDurations = coordinatorResourceDurations.get(coordinatorID); + if (resourceDurations != null) { + Integer allocatedDuration = resourceDurations.remove(resourceID); + if (allocatedDuration != null) { + resource.addDuration(allocatedDuration); + } + if (resourceDurations.isEmpty()) { + coordinatorResourceDurations.remove(coordinatorID); + } + } + log("Coordinator " + coordinatorID + " returned resource " + resourceID); + return "Resource " + resourceID + " returned successfully."; + } else { + return "Resource " + resourceID + " not found."; + } + } else { + return "You do not occupy resource " + resourceID + "."; + } + } + + //Helper Methods + private Resource getResourceByID(String resourceID) { + for (Map<String, Resource> resources : resourceMap.values()) { + if (resources.containsKey(resourceID)) { + return resources.get(resourceID); + } + } + return null; + } + + private synchronized String handleSwapRequestResource(String coordinatorID, String resourceID, int duration) { + //Attempt to allocate the resource to the coordinator + for (Map<String, Resource> resources : resourceMap.values()) { + Resource resource = resources.get(resourceID); + if (resource != null) { + if (resource.getDuration() >= duration) { + resource.subDuration(duration); + coordinatorResources.computeIfAbsent(coordinatorID, k -> new HashSet<>()).add(resourceID); + coordinatorResourceDurations.computeIfAbsent(coordinatorID, k -> new HashMap<>()).put(resourceID, duration); + log("Resource " + resourceID + " temporarily allocated to " + coordinatorID + " for swapping, duration " + duration); + return "Success"; + } else { + return "Insufficient duration for resource " + resourceID; + } + } + } + return "Resource " + resourceID + " not found."; + } + + private synchronized String handleSwapReturnResource(String coordinatorID, String resourceID) { + Set<String> resources = coordinatorResources.get(coordinatorID); + if (resources != null && resources.contains(resourceID)) { + resources.remove(resourceID); + if (resources.isEmpty()) { + coordinatorResources.remove(coordinatorID); + } + Resource resource = getResourceByID(resourceID); + if (resource != null) { + Map<String, Integer> resourceDurations = coordinatorResourceDurations.get(coordinatorID); + if (resourceDurations != null) { + Integer allocatedDuration = resourceDurations.remove(resourceID); + if (allocatedDuration != null) { + resource.addDuration(allocatedDuration); + } + if (resourceDurations.isEmpty()) { + coordinatorResourceDurations.remove(coordinatorID); + } + } + log("Coordinator " + coordinatorID + " resource " + resourceID + " allocation undone during swap"); + return "Success"; + } else { + return "Resource " + resourceID + " not found."; + } + } else { + return "Coordinator did not acquire resource " + resourceID; + } + } + + private String sendUDPRequest(String serverName, String methodName, String... params) { + String response = ""; + try { + InetAddress host = InetAddress.getByName("localhost"); + int port = serverUdpPorts.get(serverName); + String requestData = methodName + ":" + String.join(":", params); + + byte[] requestBytes = requestData.getBytes(); + DatagramPacket requestPacket = new DatagramPacket(requestBytes, requestBytes.length, host, port); + + DatagramSocket socket = new DatagramSocket(); + socket.send(requestPacket); + + byte[] buffer = new byte[4096]; + DatagramPacket responsePacket = new DatagramPacket(buffer, buffer.length); + socket.receive(responsePacket); + + response = new String(responsePacket.getData(), 0, responsePacket.getLength()); + socket.close(); + } catch (Exception e) { + e.printStackTrace(); + } + return response; + } + + //Implement ResponderInterface Methods + public synchronized String addResource(String resourceID, String resourceName, int duration) { + resourceName = resourceName.toUpperCase(); + Map<String, Resource> resources = resourceMap.computeIfAbsent(resourceName, k -> new HashMap<>()); + Resource resource = resources.get(resourceID); + + if (resource != null) { + if (duration > resource.getDuration()) { + resource.setDuration(duration); + log("Updated duration of resource " + resourceID); + return "Resource " + resourceID + " duration updated."; + } else { + log("Resource " + resourceID + " already exists with greater or equal duration."); + return "Resource " + resourceID + " already exists with greater or equal duration."; + } + } else { + resource = new Resource(resourceName, duration); + resources.put(resourceID, resource); + log("Added new resource " + resourceID); + return "Resource " + resourceID + " added successfully."; + } + } + + public synchronized String removeResource(String resourceID, int duration) { + for (Map<String, Resource> resources : resourceMap.values()) { + Resource resource = resources.get(resourceID); + if (resource != null) { + if (duration >= resource.getDuration()) { + resources.remove(resourceID); + log("Resource " + resourceID + " completely removed."); + return "Resource " + resourceID + " completely removed."; + } else { + resource.subDuration(duration); + log("Decreased duration of resource " + resourceID); + return "Resource " + resourceID + " duration decreased by " + duration + "."; + } + } + } + log("Resource " + resourceID + " not found."); + return "Resource " + resourceID + " not found."; + } + + public String listResourceAvailability(String resourceName) { + resourceName = resourceName.toUpperCase(); + StringBuilder result = new StringBuilder(); + + synchronized (this) { + Map<String, Resource> resources = resourceMap.get(resourceName); + if (resources != null) { + for (Resource resource : resources.values()) { + if (resource.getDuration() > 0) { + result.append(resource.getResourceName()).append(" - ").append(serverName).append(" ").append(resource.getDuration()).append("\n"); + } + } + } + } + + for (String otherServer : serverUdpPorts.keySet()) { + if (!otherServer.equals(serverName)) { + String response = sendUDPRequest(otherServer, "listResourceAvailability", resourceName); + result.append(response); + } + } + + return result.toString(); + } + + public String requestResource(String coordinatorID, String resourceID, int duration) { + String resourceServer = resourceID.substring(0, 3).toUpperCase(); + String response; + + if (resourceServer.equals(serverName)) { + response = handleLocalResourceRequest(coordinatorID, resourceID, duration); + } else { + response = sendUDPRequest(resourceServer, "requestResource", coordinatorID, resourceID, String.valueOf(duration)); + } + + log("Coordinator " + coordinatorID + " requested resource " + resourceID + ": " + response); + return response; + } + + public String findResource(String coordinatorID, String resourceName) { + resourceName = resourceName.toUpperCase(); + StringBuilder result = new StringBuilder(); + + for (String otherServer : serverUdpPorts.keySet()) { + String response; + if (otherServer.equals(serverName)) { + response = handleFindResource(coordinatorID, resourceName); + } else { + response = sendUDPRequest(otherServer, "findResource", coordinatorID, resourceName); + } + result.append(response); + } + + return result.toString(); + } + + public String returnResource(String coordinatorID, String resourceID) { + String resourceServer = resourceID.substring(0, 3).toUpperCase(); + String response; + + if (resourceServer.equals(serverName)) { + response = handleReturnResource(coordinatorID, resourceID); + } else { + response = sendUDPRequest(resourceServer, "returnResource", coordinatorID, resourceID); + } + + log("Coordinator " + coordinatorID + " returned resource " + resourceID + ": " + response); + return response; + } + + private synchronized int getCoordinatorResourceDuration(String coordinatorID, String resourceID) { + Map<String, Integer> resourceDurations = coordinatorResourceDurations.get(coordinatorID); + if (resourceDurations != null && resourceDurations.containsKey(resourceID)) { + return resourceDurations.get(resourceID); + } + return 0; + } + + private synchronized String checkCoordinatorHasResource(String coordinatorID, String resourceID) { + Set<String> resources = coordinatorResources.get(coordinatorID); + if (resources != null && resources.contains(resourceID)) { + return "true"; + } + return "false"; + } + + public String swapResource(String coordinatorID, String oldResourceID, String oldResourceType, String newResourceID, String newResourceType) { + String oldResourceServer = oldResourceID.substring(0, 3).toUpperCase(); + String newResourceServer = newResourceID.substring(0, 3).toUpperCase(); + + //Check if the coordinator has the old resource + String checkOldResourceResponse; + if (oldResourceServer.equals(serverName)) { + checkOldResourceResponse = checkCoordinatorHasResource(coordinatorID, oldResourceID); + } else { + checkOldResourceResponse = sendUDPRequest(oldResourceServer, "checkCoordinatorResource", coordinatorID, oldResourceID); + } + + if (!checkOldResourceResponse.equals("true")) { + return "Coordinator has not acquired the old resource."; + } + + //Get the duration allocated to the old resource + int duration; + if (oldResourceServer.equals(serverName)) { + duration = getCoordinatorResourceDuration(coordinatorID, oldResourceID); + } else { + String durationStr = sendUDPRequest(oldResourceServer, "getCoordinatorResourceDuration", coordinatorID, oldResourceID); + duration = Integer.parseInt(durationStr); + } + + //Attempt to acquire the new resource + String requestResponse; + if (newResourceServer.equals(serverName)) { + requestResponse = handleLocalResourceRequest(coordinatorID, newResourceID, duration); + } else { + requestResponse = sendUDPRequest(newResourceServer, "requestResource", coordinatorID, newResourceID, String.valueOf(duration)); + } + + if (!requestResponse.contains("allocated successfully")) { + return "Failed to acquire new resource: " + requestResponse; + } + + //Return the old resource + String returnResponse; + if (oldResourceServer.equals(serverName)) { + returnResponse = handleReturnResource(coordinatorID, oldResourceID); + } else { + returnResponse = sendUDPRequest(oldResourceServer, "returnResource", coordinatorID, oldResourceID); + } + + if (!returnResponse.contains("returned successfully")) { + //Undo the allocation of the new resource + if (newResourceServer.equals(serverName)) { + handleReturnResource(coordinatorID, newResourceID); + } else { + sendUDPRequest(newResourceServer, "returnResource", coordinatorID, newResourceID); + } + return "Failed to return old resource: " + returnResponse; + } + + log("Coordinator " + coordinatorID + " swapped resource " + oldResourceID + " with " + newResourceID); + + return "Resource swap successful."; + } +} + |