MDS now configurable via app.configs

Fixed data block size bug for the client

code/hpdos_client/app.config

 app.name="HPDOS-Client"
 app.version="0.1.4"
 app.thread_count=4
-app.runtime=30
+app.runtime=1
 app.data_size=10
-app.data_conversion_factor="B"
+app.data_conversion_factor=B
 app.private_ratio=0.8
 app.cycle_create=1
 app.cycle_read=4

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

@@ -17,8 +17,8 @@ import java.util.*;
 import java.util.concurrent.*;
 
 public class ClientRunner {
-    public static int parallelCount = 10;
-    public static int runtime = 0;
+    public static int parallelCount;
+    public static int runtime;
     private final String clientID;
     public static String propertiesFile;
     private int cCreate, cRead, cUpdate, cDelete;
@@ -26,6 +26,7 @@ public class ClientRunner {
     private List<Follower> replicaSet;
     private Queue<Long> createTime, updateTime, readTime, deleteTime;
     private final Properties properties;
+
     public ClientRunner() {
         clientID = UUID.randomUUID().toString();
         properties = new Properties();
@@ -100,8 +101,8 @@ public class ClientRunner {
     }
 
     private String createString() {
-        int dataSize = Integer.parseInt((String) properties.get("app.data_size"));
-        dataSize /= 2; // Java strings are 2B long
+        double dataSize = Double.parseDouble((String) properties.get("app.data_size"));
+        dataSize /= 2.0; // Java strings are 2B long
         String conversionFactor = (String) properties.get("app.data_conversion_factor");
         int multiplier = 1;
         switch (conversionFactor) {
@@ -109,7 +110,7 @@ public class ClientRunner {
             case "M": multiplier *= 1000;
             case "K": multiplier *= 1000;
         }
-        char[] data = new char[dataSize * multiplier];
+        char[] data = new char[(int)(dataSize * multiplier)];
         return new String(data);
     }
     public double runExperiment(String id, long experimentStartTime) {

code/hpdos_server/app.config

+app.name=HPDOS-Server
+app.version=0.1.6
+app.REPLICATION_TYPE=async
+app.REPLICATOR_THREAD_POOL_SIZE=140

code/hpdos_server/app/src/main/java/hpdos/ConfigConstants.java

@@ -5,10 +5,10 @@ public class ConfigConstants {
     public static final int PORT = 8080;
     public static final int HEARTBEAT_INTERVAL = 500;
     public static final int REPLICATION_TIMEOUT = 5000;
-
+    public static final String replicationAsync = "async";
     // Backend types 300-399
     public static final int BACKEND_IN_MEMORY = 300;
     public static final int LSM_BACKEND = 301;
     public static final int BINARY_TREE_BACKEND = 302;
-    public static final int REPLICATOR_THREAD_POOL_SIZE = 12;
+    public static final int REPLICATOR_THREAD_POOL_SIZE = Runtime.getRuntime().availableProcessors() * 2;
 }

code/hpdos_server/app/src/main/java/hpdos/MetadataServer.java

@@ -17,11 +17,10 @@ import io.grpc.ManagedChannelBuilder;
 import io.grpc.Server;
 import io.grpc.ServerBuilder;
 
+import java.io.FileInputStream;
 import java.io.IOException;
-import java.util.HashMap;
-import java.util.Timer;
-import java.util.TimerTask;
-import java.util.UUID;
+import java.io.InputStream;
+import java.util.*;
 
 public class MetadataServer {
     private Server server;
@@ -32,13 +31,17 @@ public class MetadataServer {
     private final String host;
     private IOHandler ioHandler;
     private ReplicationService replicationService;
+    private final Properties properties;
 
-    public MetadataServer() {
+    public MetadataServer(String propertiesFile) throws IOException {
         this.followers = new HashMap<>();
         this.serverID = UUID.randomUUID().toString();
         this.port = 10000 + (int)(Math.random() * 40000);
         this.host = "localhost";
         this.replicationService = null;
+        this.properties = new Properties();
+        InputStream inputStream = new FileInputStream(propertiesFile);
+        properties.load(inputStream);
     }
 
     public String getGreeting() {
@@ -139,7 +142,14 @@ public class MetadataServer {
         }
     }
     public static void main(String[] args) {
-        MetadataServer metaDataServer = new MetadataServer();
+
+        MetadataServer metaDataServer = null;
+        try {
+            metaDataServer = new MetadataServer(args[0]);
+        } catch (IOException e) {
+            System.out.println("Config file missing or unreadable");
+            System.exit(-1);
+        }
 
         System.out.println(metaDataServer.getGreeting());
         System.out.println("Starting Metadata service");
@@ -154,7 +164,8 @@ public class MetadataServer {
         System.out.println("Searching for MetadataMaster");
         metaDataServer.announceToMaster();
         if (metaDataServer.isMaster) {
-            metaDataServer.replicationService = new InlineReplicationService(metaDataServer.followers);
+            metaDataServer.replicationService = new InlineReplicationService(
+                    metaDataServer.followers, metaDataServer.properties);
             System.out.println("Started master replication module");
             boolean status = metaDataServer.startMasterServices();
             System.out.println("Master ID: " + metaDataServer.serverID);

code/hpdos_server/app/src/main/java/hpdos/handler/NetworkHandler.java

@@ -10,6 +10,7 @@ import io.grpc.stub.StreamObserver;
 
 import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.concurrent.ExecutionException;
 
 public class NetworkHandler extends NetworkServiceGrpc.NetworkServiceImplBase {
     private final IOHandler ioHandler;
@@ -102,10 +103,10 @@ public class NetworkHandler extends NetworkServiceGrpc.NetworkServiceImplBase {
 //            System.out.println("starting replication");
             Stopwatch stopwatch = Stopwatch.createUnstarted();
             stopwatch.start();
-            replicationResponse = replicationService.replicateMetadataAsync(replicationRequest);
+            replicationResponse = replicationService.replicateMetadata(replicationRequest);
             stopwatch.stop();
 //            System.out.println("Network handler replicate" + stopwatch);
-        } catch (InterruptedException e) {
+        } catch (InterruptedException | ExecutionException e) {
             e.printStackTrace();
         }
 //        System.out.println("replication complete");

code/hpdos_server/app/src/main/java/hpdos/lib/InlineReplicationService.java

@@ -15,6 +15,7 @@ import hpdos.message.ResponseBuilder;
 import io.grpc.ManagedChannel;
 import io.grpc.ManagedChannelBuilder;
 
+import javax.annotation.Nonnull;
 import java.util.*;
 import java.util.concurrent.*;
 
@@ -22,26 +23,40 @@ public class InlineReplicationService implements ReplicationService {
 
     private final HashMap<String, MasterFollower> followers;
     private final HashMap<String, ManagedChannel> channels;
-    private final ExecutorService executorService;
-    public InlineReplicationService(HashMap<String, MasterFollower> followers) {
+    private ExecutorService executorService;
+    private final boolean isReplicationAsync;
+    public InlineReplicationService(HashMap<String, MasterFollower> followers, Properties properties) {
         this.followers = followers;
         this.channels = new HashMap<>();
-        for (MasterFollower follower: this.followers.values()) {
-            ManagedChannel channel = ManagedChannelBuilder
-                    .forAddress(follower.getIp(), follower.getPort())
-                    .usePlaintext()
-                    .build();
-            channels.put(follower.getFollowerID(), channel);
+        String replicationType = (String) properties.get("app.REPLICATION_TYPE");
+        this.isReplicationAsync = replicationType.equals(ConfigConstants.replicationAsync);
+        if (!this.isReplicationAsync) {
+            int replicationThreadPoolSize;
+            if (properties.containsKey("app.REPLICATOR_THREAD_POOL_SIZE")) {
+                replicationThreadPoolSize = Integer.parseInt((String)
+                        properties.get("app.REPLICATOR_THREAD_POOL_SIZE"));
+            } else {
+                replicationThreadPoolSize = ConfigConstants.REPLICATOR_THREAD_POOL_SIZE;
+            }
+            this.executorService = Executors.newFixedThreadPool(replicationThreadPoolSize);
+            System.out.println("Creating synchronous replication pool. Pool size: " + replicationThreadPoolSize);
+        } else {
+            System.out.println("Replication to be handled using asynchronous handlers");
         }
-        this.executorService = Executors.newFixedThreadPool(ConfigConstants.REPLICATOR_THREAD_POOL_SIZE);
     }
 
     @Override
     public void cleanup() throws InterruptedException {
         for (ManagedChannel channel: channels.values())
             channel.shutdown();
-        executorService.shutdown();
-        executorService.awaitTermination(MessageConstants.STATUS_REPLICATION_TIMEOUT, TimeUnit.MILLISECONDS);
+        if (this.executorService != null) {
+            executorService.shutdown();
+            boolean status = executorService.
+                    awaitTermination(MessageConstants.STATUS_REPLICATION_TIMEOUT, TimeUnit.MILLISECONDS);
+            if (!status)
+                executorService.shutdownNow();
+
+        }
     }
 
     private void establishChannels() {
@@ -58,7 +73,14 @@ public class InlineReplicationService implements ReplicationService {
     }
 
     @Override
-    public ReplicationResponse replicateMetadata(ReplicationRequest replicationRequest)
+    public ReplicationResponse replicateMetadata(ReplicationRequest replicationRequest) throws ExecutionException, InterruptedException {
+        if (this.isReplicationAsync) {
+            return replicateMetadataAsync(replicationRequest);
+        } else {
+            return replicateMetadataSync(replicationRequest);
+        }
+    }
+    public ReplicationResponse replicateMetadataSync(ReplicationRequest replicationRequest)
             throws InterruptedException, ExecutionException {
 
         Set<Callable<ReplicationResponse>> callables = new HashSet<>();
@@ -104,7 +126,6 @@ public class InlineReplicationService implements ReplicationService {
      * @param replicationRequest replication request sent
      * @return replication response
      */
-    @Override
     public ReplicationResponse replicateMetadataAsync(ReplicationRequest replicationRequest) throws InterruptedException {
         CountDownLatch replicationWaiter = new CountDownLatch(this.followers.size());
         HashMap<String, Response> responseHashMap = new HashMap<>();
@@ -133,7 +154,7 @@ public class InlineReplicationService implements ReplicationService {
                 }
 
                 @Override
-                public void onFailure(Throwable t) {
+                public void onFailure(@Nonnull Throwable t) {
                     replicationWaiter.countDown();
                 }
             }, MoreExecutors.directExecutor());

code/hpdos_server/app/src/main/java/hpdos/lib/ReplicationService.java

@@ -8,7 +8,6 @@ import java.util.concurrent.ExecutionException;
 
 public interface ReplicationService {
     abstract ReplicationResponse replicateMetadata(ReplicationRequest replicationRequest) throws InterruptedException, ExecutionException;
-    abstract ReplicationResponse replicateMetadataAsync(ReplicationRequest replicationRequest) throws InterruptedException;
     abstract void cleanup() throws InterruptedException;
     abstract HashMap<String, MasterFollower> getFollowers();
 }

code/hpdos_server/app/src/test/java/hpdos/AppTest.java

@@ -4,11 +4,14 @@
 package hpdos;
 
 import org.junit.Test;
+
+import java.io.IOException;
+
 import static org.junit.Assert.*;
 
 public class AppTest {
-    @Test public void testAppHasAGreeting() {
-        MetadataServer classUnderTest = new MetadataServer();
+    @Test public void testAppHasAGreeting() throws IOException {
+        MetadataServer classUnderTest = new MetadataServer("dummypath");
         assertNotNull("app should have a greeting", classUnderTest.getGreeting());
     }
 }