Added support for full CRUD queries.

Needs more testing

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

@@ -33,31 +33,36 @@ public class ClientRunner {
                 forAddress(ConfigConstants.HOST, ConfigConstants.PORT)
                 .usePlaintext()
                 .build();
-        String key = "dummy", value = "dummy";
-        NetworkServiceGrpc.NetworkServiceBlockingStub stub = NetworkServiceGrpc.newBlockingStub(channel);
-        ArrayList<Request> request = new ArrayList<>();
+        double averageReadTime = 0, averageWriteTime = 0;
+        for (int i = 1; i <= 10; i++) {
 
-        // create a metadata block
-        request.add(RequestBuilder.buildRequest(MessageConstants.METADATA_CREATE,
-                0, value.length(), key,
-                0, MessageConstants.METADATA_ACCESS_PRIVATE, this.clientID, value));
-        Packet packet = RequestBuilder.buildPacket(request);
-        Timestamp timestampCreateStart = Timestamp.from(Instant.now());
-        Packet response = stub.createMetadata(packet);
-        System.out.println(response);
-        System.out.println("MDS create time taken in ms: " + (Timestamp.from(Instant.now()).getTime()
-                - timestampCreateStart.getTime()));
+            String key = Integer.toString((int) (Math.random() * Integer.MAX_VALUE)), value = "dummy";
+            NetworkServiceGrpc.NetworkServiceBlockingStub stub = NetworkServiceGrpc.newBlockingStub(channel);
+            ArrayList<Request> request = new ArrayList<>();
 
-        // read back the metadata
-        request.clear();
-        request.add(RequestBuilder.buildRequest(MessageConstants.METADATA_READ,
-                0, 0, key, 0, MessageConstants.METADATA_ACCESS_PRIVATE, this.clientID, ""));
-        packet = RequestBuilder.buildPacket(request);
-        Timestamp timestampReadStart = Timestamp.from(Instant.now());
-        response = stub.readMetadata(packet);
-        System.out.println(response);
-        System.out.println("MDS create time taken in ms: " + (Timestamp.from(Instant.now()).getTime()
-                - timestampReadStart.getTime()));
+            // create a metadata block
+            request.add(RequestBuilder.buildRequest(MessageConstants.METADATA_CREATE,
+                    0, value.length(), key,
+                    0, MessageConstants.METADATA_ACCESS_PRIVATE, this.clientID, value));
+            Packet packet = RequestBuilder.buildPacket(request);
+            long timestampCreateStart = System.currentTimeMillis();
+            Packet response = stub.createMetadata(packet);
+
+            averageWriteTime = (averageWriteTime * (i - 1) + (System.currentTimeMillis() - timestampCreateStart)) / i;
+            // read back the metadata
+            request.clear();
+            request.add(RequestBuilder.buildRequest(MessageConstants.METADATA_READ,
+                    0, 0, key, 0, MessageConstants.METADATA_ACCESS_PRIVATE, this.clientID, ""));
+            packet = RequestBuilder.buildPacket(request);
+            long timestampReadStart = System.currentTimeMillis();
+            response = stub.readMetadata(packet);
+            averageReadTime = (averageReadTime * (i - 1) + (System.currentTimeMillis() - timestampReadStart)) / i;
+
+//            if (i % 100 == 0)
+                System.out.println("Average MDS read time: "
+                    + averageReadTime
+                    + " ms average create time: " + averageWriteTime + "ms");
+        }
         channel.shutdown();
         return id;
     }
@@ -71,12 +76,7 @@ public class ClientRunner {
         Set<Callable<String>> callables = new HashSet<>();
         for (int i = 0; i < parallelCount; i++) {
             int finalI = i;
-            callables.add(new Callable<String>() {
-                @Override
-                public String call() throws Exception {
-                    return clientRunner.initClient(Integer.toString(finalI));
-                }
-            });
+            callables.add(() -> clientRunner.initClient(Integer.toString(finalI)));
         }
 
         List<Future<String>> futures = executorService.invokeAll(callables);

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

@@ -6,6 +6,7 @@ import hpdos.grpc.Request;
 import hpdos.grpc.Response;
 import hpdos.lib.StorageModel;
 import hpdos.lib.StorageService;
+import hpdos.lib.StoredModel;
 import hpdos.message.MessageConstants;
 import hpdos.message.ResponseBuilder;
 
@@ -26,34 +27,81 @@ public class IOHandler {
                 request.getAccessType(),
                 request.getClientID(),
                 request.getValue());
-        boolean status = storageService.create(request.getKey(), blob);
-        if (status) {
-            ack = ResponseBuilder.buildAck(MessageConstants.INIT_VERSION,
-                    blob.getDataSize(),
-                    blob.getKey(),
-                    blob.getCrc(),
-                    blob.getValue());
+        StoredModel storedData = storageService.create(request.getKey(), blob);
+        if (storedData.getStatus() == MessageConstants.STATUS_OK) {
+            ack = ResponseBuilder.buildAck(storedData.getData().getVersion(),
+                    storedData.getData().getDataSize(),
+                    storedData.getData().getKey(),
+                    storedData.getData().getCrc(),
+                    storedData.getData().getValue());
+
+            return ResponseBuilder.buildResponsePacket(
+                    MessageConstants.METADATA_CREATE, MessageConstants.STATUS_OK,
+                    ack, null);
         }
         else {
             nack = ResponseBuilder.buildNack(MessageConstants.INVALID_VERSION,
                     request.getKey());
-        }
-        if (status)
             return ResponseBuilder.buildResponsePacket(
-                    MessageConstants.METADATA_CREATE, MessageConstants.STATUS_OK,
-                    ack, null);
-        return ResponseBuilder.buildResponsePacket(
-                MessageConstants.METADATA_CREATE, MessageConstants.STATUS_IO_WRITE_FAILED,
-                null, nack);
+                    MessageConstants.METADATA_CREATE, storedData.getStatus(),
+                    null, nack);
+        }
     }
 
     public Response update(Request request) {
-        return null;
+        Ack ack = null; Nack nack = null;
+        StorageModel blob = new StorageModel(
+                request.getVersion(),
+                request.getDataSize(),
+                request.getKey(),
+                request.getAccessType(),
+                request.getClientID(),
+                request.getValue());
+        StoredModel storedData = storageService.update(request.getKey(), blob);
+        if (storedData.getStatus() == MessageConstants.STATUS_OK) {
+            ack = ResponseBuilder.buildAck(storedData.getData().getVersion(),
+                    storedData.getData().getDataSize(),
+                    storedData.getData().getKey(),
+                    storedData.getData().getCrc(),
+                    storedData.getData().getValue());
+
+            return ResponseBuilder.buildResponsePacket(
+                    MessageConstants.METADATA_CREATE, MessageConstants.STATUS_OK,
+                    ack, null);
+        }
+        else {
+            int version = (storedData.getStatus() == MessageConstants.STATUS_UPDATE_VERSION_MISMATCH)?
+                    storedData.getData().getVersion(): MessageConstants.INVALID_VERSION;
+            nack = ResponseBuilder.buildNack(version, request.getKey());
+            return ResponseBuilder.buildResponsePacket(
+                    MessageConstants.METADATA_CREATE, storedData.getStatus(),
+                    null, nack);
+        }
     }
 
     public Response delete(Request request) {
-        return null;
+        Ack ack = null; Nack nack = null;
 
+        StoredModel storedData = storageService.delete(request.getKey(), request.getVersion());
+        if (storedData.getStatus() == MessageConstants.STATUS_OK) {
+            ack = ResponseBuilder.buildAck(storedData.getData().getVersion(),
+                    storedData.getData().getDataSize(),
+                    storedData.getData().getKey(),
+                    storedData.getData().getCrc(),
+                    storedData.getData().getValue());
+
+            return ResponseBuilder.buildResponsePacket(
+                    MessageConstants.METADATA_CREATE, MessageConstants.STATUS_OK,
+                    ack, null);
+        }
+        else {
+            int version = (storedData.getStatus() == MessageConstants.STATUS_UPDATE_VERSION_MISMATCH)?
+                    storedData.getData().getVersion(): MessageConstants.INVALID_VERSION;
+            nack = ResponseBuilder.buildNack(version, request.getKey());
+            return ResponseBuilder.buildResponsePacket(
+                    MessageConstants.METADATA_CREATE, storedData.getStatus(),
+                    null, nack);
+        }
     }
 
     public Response read(Request request) {
@@ -65,24 +113,24 @@ public class IOHandler {
             nack = ResponseBuilder.buildNack(MessageConstants.INVALID_VERSION,
                     request.getKey());
         } else {
-            StorageModel blob = storageService.readByKey(request.getKey());
-            if (blob == null) {
+            StoredModel readData = storageService.readByKey(request.getKey());
+            if (readData.getStatus() == MessageConstants.STATUS_KEY_NOT_FOUND) {
                 status = MessageConstants.STATUS_KEY_NOT_FOUND;
                 nack = ResponseBuilder.buildNack(MessageConstants.INVALID_VERSION,
                         request.getKey());
 
-            } else if (request.getAccessType() == MessageConstants.METADATA_ACCESS_PRIVATE &&
-                    !request.getClientID().equals(blob.getOwner())) {
-                status = MessageConstants.STATUS_OWNER_MISMATCH;
+            } else if (readData.getData().getAccessType() == MessageConstants.METADATA_ACCESS_PRIVATE &&
+                    !request.getClientID().equals(readData.getData().getOwner())) {
+                status = MessageConstants.STATUS_UNAUTHORIZED_PRIVATE_KEY_ACCESS;
                 nack = ResponseBuilder.buildNack(MessageConstants.INVALID_VERSION,
                         request.getKey());
             } else {
                 status = MessageConstants.STATUS_OK;
-                ack = ResponseBuilder.buildAck(MessageConstants.INIT_VERSION,
-                        blob.getDataSize(),
-                        blob.getKey(),
-                        blob.getCrc(),
-                        blob.getValue());
+                ack = ResponseBuilder.buildAck(readData.getData().getVersion(),
+                        readData.getData().getDataSize(),
+                        readData.getData().getKey(),
+                        readData.getData().getCrc(),
+                        readData.getData().getValue());
             }
         }
         return ResponseBuilder.buildResponsePacket(

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

 package hpdos.handler;
 
+import com.google.common.base.Stopwatch;
 import hpdos.grpc.*;
 import hpdos.lib.ReplicationService;
 import hpdos.message.RequestBuilder;
@@ -34,10 +35,48 @@ public class NetworkHandler extends NetworkServiceGrpc.NetworkServiceImplBase {
     @Override
     public void createMetadata(Packet requestPacket, StreamObserver<Packet> responseObserver) {
         System.out.println("new create request " + requestPacket);
+        Stopwatch stopwatch = Stopwatch.createUnstarted();
+        stopwatch.start();
         for (Request request: requestPacket.getRequestList()) {
             ioHandler.create(request);
         }
+        stopwatch.stop();
+
+        System.out.println("Added to local memory " + stopwatch);
+        stopwatch.reset();
+        stopwatch.start();
+        Packet packet = replicate(requestPacket);
+        stopwatch.stop();
+        System.out.println("Replication time " + stopwatch);
+        responseObserver.onNext(packet);
+        responseObserver.onCompleted();
+    }
+
+    @Override
+    public void updateMetadata(Packet requestPacket, StreamObserver<Packet> responseObserver) {
+        System.out.println("new create request " + requestPacket);
+        for (Request request: requestPacket.getRequestList()) {
+            ioHandler.update(request);
+        }
+        System.out.println("Added to local memory");
+        Packet packet = replicate(requestPacket);
+        responseObserver.onNext(packet);
+        responseObserver.onCompleted();
+    }
+
+    @Override
+    public void deleteMetadata(Packet requestPacket, StreamObserver<Packet> responseObserver) {
+        System.out.println("new create request " + requestPacket);
+        for (Request request: requestPacket.getRequestList()) {
+            ioHandler.delete(request);
+        }
         System.out.println("Added to local memory");
+        Packet packet = replicate(requestPacket);
+        responseObserver.onNext(packet);
+        responseObserver.onCompleted();
+    }
+
+    private Packet replicate(Packet requestPacket) {
         ReplicationRequest replicationRequest = RequestBuilder.
                 buildReplicationRequest(new ArrayList<>(requestPacket.getRequestList()));
         ReplicationResponse replicationResponse = null;
@@ -50,17 +89,6 @@ public class NetworkHandler extends NetworkServiceGrpc.NetworkServiceImplBase {
         System.out.println("replication complete");
         Packet packet = ResponseBuilder.buildPacket(new ArrayList<>(replicationResponse.getResponseList()));
         System.out.println(packet);
-        responseObserver.onNext(packet);
-        responseObserver.onCompleted();
-    }
-
-    @Override
-    public void updateMetadata(Packet request, StreamObserver<Packet> responseObserver) {
-        super.updateMetadata(request, responseObserver);
-    }
-
-    @Override
-    public void deleteMetadata(Packet request, StreamObserver<Packet> responseObserver) {
-        super.deleteMetadata(request, responseObserver);
+        return packet;
     }
 }

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

 package hpdos.lib;
 
+import hpdos.message.MessageConstants;
+
 import java.util.HashMap;
 
-public class MemoryStorageService implements StorageService{
+public class MemoryStorageService implements StorageService {
     private final HashMap<String, StorageModel> memoryKVStore;
     public MemoryStorageService() {
         this.memoryKVStore = new HashMap<>();
     }
     @Override
-    public boolean create(String key, StorageModel value) {
+    public StoredModel create(String key, StorageModel value) {
         try {
             if (memoryKVStore.containsKey(key))
-                return false;
+                return new StoredModel(null, MessageConstants.STATUS_KEY_EXISTS);
             memoryKVStore.put(key, value);
+            return new StoredModel(value, MessageConstants.STATUS_OK);
         } catch (Exception e) {
-            return false;
+            return null;
         }
-        return true;
     }
 
     @Override
-    public StorageModel readByKey(String key) {
-        return memoryKVStore.get(key);
+    public StoredModel readByKey(String key) {
+        if (!memoryKVStore.containsKey(key))
+            return new StoredModel(null, MessageConstants.STATUS_KEY_NOT_FOUND);
+        return new StoredModel(memoryKVStore.get(key), MessageConstants.STATUS_OK);
     }
 
     @Override
-    public boolean update(String key, StorageModel value) {
+    public StoredModel update(String key, StorageModel value) {
         if (!memoryKVStore.containsKey(key))
-            return false;
+            return new StoredModel(null, MessageConstants.STATUS_KEY_NOT_FOUND);
+        StorageModel previousValue = memoryKVStore.get(key);
+        if (previousValue.getVersion() != value.getVersion())
+            return new StoredModel(previousValue, MessageConstants.STATUS_UPDATE_VERSION_MISMATCH);
+        value.setVersion((previousValue.getVersion() + 1) % Integer.MAX_VALUE); // version wraps around
         memoryKVStore.put(key, value);
-        return true;
+        return new StoredModel(value, MessageConstants.STATUS_OK);
     }
 
     @Override
-    public boolean delete(String key) {
-        try {
-            memoryKVStore.remove(key);
-        } catch (Exception e) {
-            return false;
-        }
-        return true;
+    public StoredModel delete(String key, int version) {
+        if (!memoryKVStore.containsKey(key))
+            return new StoredModel(null, MessageConstants.STATUS_KEY_NOT_FOUND);
+        StorageModel previousValue = memoryKVStore.get(key);
+        if (previousValue.getVersion() != version)
+            return new StoredModel(previousValue, MessageConstants.STATUS_UPDATE_VERSION_MISMATCH);
+        return new StoredModel(memoryKVStore.remove(key), MessageConstants.STATUS_OK);
     }
 }

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

@@ -82,3 +82,4 @@ public class StorageModel {
         this.owner = owner;
     }
 }
+

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

 package hpdos.lib;
 
 public interface StorageService {
-    boolean create(String key, StorageModel value);
-    StorageModel readByKey(String key);
-    boolean update(String key, StorageModel value);
-    boolean delete(String key);
+    StoredModel create(String key, StorageModel value);
+    StoredModel readByKey(String key);
+    StoredModel update(String key, StorageModel value);
+    StoredModel delete(String key, int version);
 }

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

+package hpdos.lib;
+
+public class StoredModel {
+    private StorageModel data;
+    private int status;
+
+    public StoredModel(StorageModel data, int status) {
+        this.data = data;
+        this.status = status;
+    }
+
+    public StorageModel getData() {
+        return data;
+    }
+
+    public void setData(StorageModel data) {
+        this.data = data;
+    }
+
+    public int getStatus() {
+        return status;
+    }
+
+    public void setStatus(int status) {
+        this.status = status;
+    }
+}

code/hpdos_server/app/src/main/java/hpdos/message/MessageConstants.java

@@ -13,12 +13,14 @@ public class MessageConstants {
 
     // Distinguishing ACK and NACK packets
     public static final int STATUS_OK = 200;
-    public static final int STATUS_OWNER_MISMATCH = 401;
-    public static final int STATUS_REPLICATE_FAILED = 402;
+    public static final int STATUS_UNAUTHORIZED_PRIVATE_KEY_ACCESS = 401;
+    public static final int STATUS_REPLICATION_FAILED = 402;
     public static final int STATUS_SERVER_NOT_MASTER = 403;
     public static final int STATUS_KEY_NOT_FOUND = 404;
     public static final int STATUS_REPLICATION_TIMEOUT = 405;
     public static final int STATUS_IO_WRITE_FAILED = 406;
+    public static final int STATUS_KEY_EXISTS = 407;
+    public static final int STATUS_UPDATE_VERSION_MISMATCH = 408;
 
     // 100 to 199 - HPDOS System internal operations
     public static final int MASTER_HEARTBEAT = 100;