trying out distributed mode

code/hpdos_client/app/src/main/java/HpdosClient/ClientRunner.java

@@ -32,16 +32,18 @@ public class ClientRunner {
     private final Properties properties;
     private final List<StorageModel> generatedPacket;
     private Semaphore limiter = null;
+    private final int cpuCount;
     public ClientRunner() {
         clientID = UUID.randomUUID().toString();
         generatedPacket = Collections.synchronizedList(new ArrayList<>());
         properties = new Properties();
+        cpuCount = Runtime.getRuntime().availableProcessors();
         try {
             InputStream inputStream = new FileInputStream(propertiesFile);
             this.properties.load(inputStream);
             System.out.println((String) properties.get("app.concurrency"));
             concurrency = Integer.parseInt((String) properties.get("app.concurrency"));
-            this.limiter = new Semaphore(concurrency);
+            this.limiter = new Semaphore(concurrency / cpuCount);
             runtime = Integer.parseInt((String) properties.get("app.runtime"));
 
             this.cCreate = Integer.parseInt((String) properties.get("app.cycle_create"));
@@ -152,9 +154,7 @@ public class ClientRunner {
 //                        System.out.println("packet " + packet);
                         for (Response response: packet.getResponseList()) {
                             if (response.getStatus() == MessageConstants.STATUS_OK)
-                                if (!sendPacket.getKey().equals(response.getAck().getKey()))
-                                    System.out.println("read kand ho gaya");
-                                else {
+                                if (sendPacket.getKey().equals(response.getAck().getKey())) {
                                     synchronized (sendPacket) {
                                         sendPacket.updateData(response.getAck());
                                     }
@@ -182,9 +182,7 @@ public class ClientRunner {
 //                        System.out.println("packet " + packet);
                         for (Response response: packet.getResponseList()) {
                             if (response.getStatus() == MessageConstants.STATUS_OK)
-                                if (!sendPacket.getKey().equals(response.getAck().getKey()))
-                                    System.out.println("update kand ho gaya");
-                                else {
+                                if (sendPacket.getKey().equals(response.getAck().getKey())) {
                                     synchronized (sendPacket) {
                                         sendPacket.updateData(response.getAck());
                                     }
@@ -210,9 +208,7 @@ public class ClientRunner {
 //                        System.out.println("packet " + packet);
                         for (Response response: packet.getResponseList()) {
                             if (response.getStatus() == MessageConstants.STATUS_OK)
-                                if (!sendPacket.getKey().equals(response.getAck().getKey()))
-                                    System.out.println("delete kand ho gaya");
-                                else {
+                                if (sendPacket.getKey().equals(response.getAck().getKey())) {
                                     synchronized (sendPacket) {
                                         generatedPacket.remove(sendPacket);
                                     }
@@ -296,7 +292,7 @@ public class ClientRunner {
         latch.await();
         storageService.cleanup();
     }
-    public static void main(String[] args) throws InterruptedException {
+    public static void main(String[] args) throws InterruptedException, ExecutionException {
         propertiesFile = args[0];
         ClientRunner clientRunner = new ClientRunner();
         System.out.println(clientRunner.getGreeting());
@@ -317,7 +313,15 @@ public class ClientRunner {
             clientRunner.experimentEnded = true;
             System.out.println("Experiment ended");
         });
-        clientRunner.runExperiment(storageService);
+        ExecutorService experimentExecutors = Executors.newFixedThreadPool(clientRunner.cpuCount);
+        Set<Callable<Double>> callables = new HashSet<>();
+        for (int i = 0; i < clientRunner.cpuCount; i++) {
+            callables.add(() -> clientRunner.runExperiment(storageService));
+        }
+        List<Future<Double>> futures = experimentExecutors.invokeAll(callables);
+        for (Future<Double> future: futures) {
+            future.get();
+        }
         long endTime = System.currentTimeMillis();
         System.out.println("Waiting for system to settle down");
         Thread.sleep(2000);