Fixed request rate for benchmarking script

code/hpdos_rdma_sal/app_benchmark.config

@@ -12,6 +12,6 @@ writes=0.8
 deletes=0.1
 clients=1
 requests_per_client=100000
-requests_per_second=1000000
+requests_per_second=5000000
 key_size=10
 value_size=10

code/hpdos_rdma_sal/src/main/java/hpdos/handlers/NetworkHandler.java

@@ -44,6 +44,8 @@ public class NetworkHandler {
     public int numberOfMasters;
     public Cache<String, String> cache;
 
+    public NetworkHandler(){}
+
     public NetworkHandler(int cacheSize, int numberOfMasters){
         this.numberOfMasters = numberOfMasters;
         this.cache = Caffeine.newBuilder().expireAfterWrite(30, TimeUnit.SECONDS).maximumSize(cacheSize).build();
@@ -82,6 +84,7 @@ public class NetworkHandler {
     {
         logger.info("Getting From local cache");
         // var value = this.cache.getIfPresent(key);
+        long startTime = System.nanoTime();
         String value = this.cache.getIfPresent(new String(key));
         if(value != null)
             return value.getBytes(StandardCharsets.UTF_8);
@@ -132,13 +135,15 @@ public class NetworkHandler {
             return response.getValue();
         }
 
-            
+        collectExperimentStatistics((System.nanoTime() - startTime)/1000, 101);  
         return null;
     }
 
     public int put(byte[] key,byte[] value) throws Exception
     {
+        
         logger.info("Putting KV pair on server ....");
+        long startTime = System.nanoTime();
         Request request= new Request();
         request.setRequestType(RequestType.PUT);
         request.setKey(key);
@@ -156,13 +161,15 @@ public class NetworkHandler {
         if(response.getAck() == AckType.SUCCESS){
             this.cache.put(new String(key), new String(value));
             logger.info("Updating cache..");
-        }  
+        }
+        collectExperimentStatistics((System.nanoTime() - startTime)/1000, 100);
         return response.getAck();
     }
 
     public int delete(byte[] key) throws Exception
     {
         logger.info("Invalidating Local Cache and sending delete request to server");
+        long startTime = System.nanoTime();
         this.cache.invalidate(new String(key));
         Request request= new Request();
         request.setRequestType(RequestType.DELETE);
@@ -177,6 +184,7 @@ public class NetworkHandler {
 
         DaRPCStream<Request, Response> stream = NetworkHandler.streams.get(ip);
         stream.request(request, response, false).get();
+        collectExperimentStatistics((System.nanoTime() - startTime)/1000, 102);
         return response.getAck();
     }
     // public Pair<byte[], byte[]> read(byte[] key) throws InterruptedException, ExecutionException, IOException{
@@ -263,4 +271,63 @@ public class NetworkHandler {
         }
     }
 
+    public static void collectExperimentStatistics(long time, int type){
+        // logger.debug("Time: " + time);
+        if (type == 100) {
+           BenchmarkingProperties.writeCount ++;
+           BenchmarkingProperties.writeTime = BenchmarkingProperties.writeTime + time;
+           // BenchmarkingProperties.actualReqPerSecond = BenchmarkingProperties.actualReqPerSecond + time;
+  
+        } 
+        else if (type == 101) {
+           BenchmarkingProperties.readCount ++;
+           BenchmarkingProperties.readTime = BenchmarkingProperties.readTime + time;
+           // BenchmarkingProperties.actualReqPerSecond = BenchmarkingProperties.actualReqPerSecond + time;
+        }
+        else if(type == 102){
+           BenchmarkingProperties.deleteCount ++;
+           BenchmarkingProperties.deleteTime = BenchmarkingProperties.deleteTime + time;
+           // BenchmarkingProperties.actualReqPerSecond = BenchmarkingProperties.actualReqPerSecond + time;
+        }
+     }
+
+     public void printStatistics(){
+        logger.debug("Total writes: " + BenchmarkingProperties.writeCount);
+        logger.debug("Total reads: " + BenchmarkingProperties.readCount);
+        logger.debug("Total deletes: " + BenchmarkingProperties.deleteCount);
+  
+        logger.debug("Total write time: " + BenchmarkingProperties.writeTime);
+        logger.debug("Total read time: " + BenchmarkingProperties.readTime);
+        logger.debug("Total delete time: " + BenchmarkingProperties.deleteTime);
+  
+        logger.debug("Average write time in microseconds: " + (BenchmarkingProperties.writeTime / BenchmarkingProperties.writeCount));
+        logger.debug("Average read time in microseconds: " + (BenchmarkingProperties.readTime / BenchmarkingProperties.readCount));
+        logger.debug("Average delete time in microseconds: " + (BenchmarkingProperties.deleteTime / BenchmarkingProperties.deleteCount));
+  
+        // logger.debug("Average Time/Request : " + BenchmarkingProperties.timePerRequest/(BenchmarkingProperties.writeCount + BenchmarkingProperties.readCount + BenchmarkingProperties.deleteCount));
+        // long avgTimePerReq = BenchmarkingProperties.timePerRequest/(BenchmarkingProperties.writeCount + BenchmarkingProperties.readCount + BenchmarkingProperties.deleteCount);
+        // logger.debug("Actual Requests/second : " + (1000000/avgTimePerReq));
+     }
+
 }
+
+class BenchmarkingProperties {
+    static Double reads;
+    static Double writes;
+    static Double deletes;
+    static int clients;
+    static long requestsPerClient;
+    static Double requestsPerSecond;
+    static int keySize;
+    static int valueSize;
+ 
+    // Result statistics
+    static int readCount;
+    static int writeCount;
+    static int deleteCount;
+    
+    static long readTime = 0;
+    static long writeTime = 0 ;
+    static long deleteTime = 0;
+    static long timePerRequest = 0;
+ }
\ No newline at end of file

code/hpdos_rdma_sal/src/main/java/hpdos/invalidationServer/InvalidationServer.java

@@ -26,15 +26,32 @@ public class InvalidationServer
     final static Logger logger = Logger.getLogger(InvalidationServer.class);
     Cache<String, String> cache;
     RdmaServerEndpoint<DaRPCServerEndpoint<InvalidationRequest, InvalidationResponse>> serverEp;
+    private long[] clusterAffinities;
     public InvalidationServer(Cache<String, String> cache2,String invalidationServerIP, int invalidationServerPort,  NetworkHandler networkHandler)throws Exception
     {
             this.cache = cache2; 
-            long[] clusterAffinities =  new long[1];
-            clusterAffinities[0] = 1<<1;
+            long[] clusterAffinities = new long[1];
+            clusterAffinities[0] = 1L << 1 ;
+            // clusterAffinities =  new long[4];
+            // for (int i = 0; i < 4; i++) {
+            //     clusterAffinities[i] = 1L << i ;    
+            // }
+            
             DaRPCServerGroup<InvalidationRequest, InvalidationResponse> group = null;
             InvalidationService  service = new InvalidationService(networkHandler);
 
-            group = DaRPCServerGroup.createServerGroup(service, clusterAffinities, -1, 1, true, 2, 2, 2, 1);    
+            group = DaRPCServerGroup.createServerGroup(service, clusterAffinities, -1, 1, false, 16, 16, 16, 4);    
+            // for (int i = 0; i < 4; i++)
+            // {
+            //     long cpu = 1L << i;
+            //     clusterAffinities[i] = cpu;
+            //     System.out.println(cpu);
+            // };
+            // DaRPCServerGroup<InvalidationRequest, InvalidationResponse> group = null;
+            // InvalidationService  service = new InvalidationService(networkHandler);
+
+            // group = DaRPCServerGroup.createServerGroup(service, clusterAffinities, -1, 0, false, 16, 16, 32, 4);
+            // group = DaRPCServerGroup.createServerGroup(service, clusterAffinities, 1000, 0, false, 16, 16, 16, 4);    
             serverEp = group.createServerEndpoint();
             InetSocketAddress address = new InetSocketAddress(invalidationServerIP, invalidationServerPort);
             serverEp.bind(address, 2);