Use the same protobuf for both requests and log entries

We are now _really_ using the command-log pattern!

cloudata-keyvalue/src/main/java/com/cloudata/keyvalue/KeyValueServer.java

@@ -5,6 +5,7 @@
 import java.net.SocketAddress;
 import java.util.EnumSet;
 import java.util.List;
+import java.util.concurrent.Executors;
 
 import javax.servlet.DispatcherType;
 
@@ -24,6 +25,8 @@
 import com.cloudata.keyvalue.web.WebModule;
 import com.google.common.base.Throwables;
 import com.google.common.collect.Lists;
+import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
 import com.google.inject.Guice;
 import com.google.inject.Injector;
 import com.google.inject.servlet.GuiceFilter;
@@ -65,7 +68,8 @@ public synchronized void start() throws Exception {
         logDir.mkdirs();
         stateDir.mkdirs();
 
-        KeyValueStateMachine stateMachine = new KeyValueStateMachine();
+        ListeningExecutorService executor = MoreExecutors.listeningDecorator(Executors.newCachedThreadPool());
+        KeyValueStateMachine stateMachine = new KeyValueStateMachine(executor);
 
         ClusterConfig config = ClusterConfig.from(local, peers);
         this.raft = RaftService.newBuilder(config).logDir(logDir).timeout(300).build(stateMachine);

cloudata-keyvalue/src/main/java/com/cloudata/keyvalue/KeyValueStateMachine.java

@@ -4,6 +4,7 @@
 
 import java.io.File;
 import java.nio.ByteBuffer;
+import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutionException;
 
 import javax.annotation.Nonnull;
@@ -19,18 +20,20 @@
 
 import com.cloudata.btree.BtreeQuery;
 import com.cloudata.btree.Keyspace;
-import com.cloudata.keyvalue.KeyValueLog.KvEntry;
-import com.cloudata.keyvalue.operation.AppendOperation;
-import com.cloudata.keyvalue.operation.DeleteOperation;
-import com.cloudata.keyvalue.operation.IncrementOperation;
-import com.cloudata.keyvalue.operation.KeyOperation;
-import com.cloudata.keyvalue.operation.SetOperation;
+import com.cloudata.keyvalue.KeyValueProtocol.ActionResponse;
+import com.cloudata.keyvalue.KeyValueProtocol.KeyValueAction;
+import com.cloudata.keyvalue.operation.KeyValueOperation;
+import com.cloudata.keyvalue.operation.KeyValueOperations;
 import com.cloudata.values.Value;
+import com.google.common.base.Function;
+import com.google.common.base.Preconditions;
 import com.google.common.base.Throwables;
 import com.google.common.cache.CacheBuilder;
 import com.google.common.cache.CacheLoader;
 import com.google.common.cache.LoadingCache;
+import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.ListeningExecutorService;
 import com.google.protobuf.ByteString;
 import com.google.protobuf.InvalidProtocolBufferException;
 
@@ -41,7 +44,10 @@
     private File baseDir;
     final LoadingCache<Long, KeyValueStore> keyValueStoreCache;
 
-    public KeyValueStateMachine() {
+    private final ListeningExecutorService executor;
+
+    public KeyValueStateMachine(ListeningExecutorService executor) {
+        this.executor = executor;
         KeyValueStoreCacheLoader loader = new KeyValueStoreCacheLoader();
         this.keyValueStoreCache = CacheBuilder.newBuilder().recordStats().build(loader);
     }
@@ -65,7 +71,7 @@ public void init(RaftService raft, File stateDir) {
     // return raft.commit(entry.toByteArray());
     // }
 
-    public <V> V doActionSync(KeyOperation<V> operation) throws InterruptedException, RaftException {
+    public ActionResponse doActionSync(KeyValueOperation operation) throws InterruptedException, RaftException {
         try {
             return doActionAsync(operation).get();
         } catch (ExecutionException e) {
@@ -80,50 +86,53 @@ public void init(RaftService raft, File stateDir) {
         }
     }
 
-    public <V> ListenableFuture<V> doActionAsync(KeyOperation<V> operation) throws RaftException {
-        KvEntry entry = operation.serialize();
+    public ListenableFuture<ActionResponse> doActionAsync(final KeyValueOperation operation) throws RaftException {
+        if (operation.isReadOnly()) {
+            return executor.submit(new Callable<ActionResponse>() {
+
+                @Override
+                public ActionResponse call() throws Exception {
+                    // TODO: Need to check that we are the leader!!
+
+                    long storeId = operation.getStoreId();
+
+                    KeyValueStore keyValueStore = getKeyValueStore(storeId);
+
+                    keyValueStore.doAction(operation);
+
+                    return operation.getResult();
+                }
+            });
+        }
+
+        KeyValueAction entry = operation.serialize();
 
         log.debug("Proposing operation {}", entry.getAction());
 
-        return (ListenableFuture<V>) raft.commitAsync(entry.toByteArray());
+        return Futures.transform(raft.commitAsync(entry.toByteArray()), new Function<Object, ActionResponse>() {
+
+            @Override
+            public ActionResponse apply(Object input) {
+                Preconditions.checkArgument(input instanceof ActionResponse);
+                return (ActionResponse) input;
+            }
+
+        });
     }
 
     @Override
     public Object applyOperation(@Nonnull ByteBuffer op) {
         // TODO: We need to prevent repetition during replay
         // (we need idempotency)
         try {
-            KvEntry entry = KvEntry.parseFrom(ByteString.copyFrom(op));
+            KeyValueAction entry = KeyValueAction.parseFrom(ByteString.copyFrom(op));
             log.debug("Committing operation {}", entry.getAction());
 
             long storeId = entry.getStoreId();
 
             KeyValueStore keyValueStore = getKeyValueStore(storeId);
 
-            KeyOperation<?> operation;
-
-            switch (entry.getAction()) {
-
-            case APPEND:
-                operation = new AppendOperation(entry);
-                break;
-
-            case DELETE:
-                operation = new DeleteOperation(entry);
-                break;
-
-            case INCREMENT: {
-                operation = new IncrementOperation(entry);
-                break;
-            }
-
-            case SET:
-                operation = new SetOperation(entry);
-                break;
-
-            default:
-                throw new UnsupportedOperationException();
-            }
+            KeyValueOperation operation = KeyValueOperations.build(entry);
 
             keyValueStore.doAction(operation);
 
@@ -168,11 +177,13 @@ public KeyValueStore load(@Nonnull Long id) throws Exception {
         }
     }
 
+    // This should also really be an operation, but is special-cased for speed
     public Value get(long storeId, Keyspace keyspace, ByteString key) {
         KeyValueStore keyValueStore = getKeyValueStore(storeId);
         return keyValueStore.get(keyspace.mapToKey(key).asReadOnlyByteBuffer());
     }
 
+    // This function should really be an operation, but we want to support streaming
     public BtreeQuery scan(long storeId, Keyspace keyspace, ByteString keyPrefix) {
         KeyValueStore keyValueStore = getKeyValueStore(storeId);
         boolean stripKeyspace = true;

cloudata-keyvalue/src/main/java/com/cloudata/keyvalue/KeyValueStore.java

@@ -14,7 +14,7 @@
 import com.cloudata.btree.PageStore;
 import com.cloudata.btree.ReadOnlyTransaction;
 import com.cloudata.btree.WriteTransaction;
-import com.cloudata.keyvalue.operation.KeyOperation;
+import com.cloudata.keyvalue.operation.KeyValueOperation;
 import com.cloudata.values.Value;
 import com.google.protobuf.ByteString;
 
@@ -33,10 +33,16 @@ public KeyValueStore(File dir, boolean uniqueKeys) throws IOException {
         this.btree = new Btree(pageStore, uniqueKeys);
     }
 
-    public void doAction(KeyOperation<?> operation) {
-        try (WriteTransaction txn = btree.beginReadWrite()) {
-            txn.doAction(btree, operation);
-            txn.commit();
+    public void doAction(KeyValueOperation operation) {
+        if (operation.isReadOnly()) {
+            try (ReadOnlyTransaction txn = btree.beginReadOnly()) {
+                txn.doAction(btree, operation);
+            }
+        } else {
+            try (WriteTransaction txn = btree.beginReadWrite()) {
+                txn.doAction(btree, operation);
+                txn.commit();
+            }
         }
     }
 

cloudata-keyvalue/src/main/java/com/cloudata/keyvalue/operation/AppendOperation.java

@@ -1,62 +1,49 @@
 package com.cloudata.keyvalue.operation;
 
-import java.nio.ByteBuffer;
-
-import com.cloudata.keyvalue.KeyValueLog.KvAction;
-import com.cloudata.keyvalue.KeyValueLog.KvEntry;
+import com.cloudata.btree.Keyspace;
+import com.cloudata.keyvalue.KeyValueProtocol.ActionType;
+import com.cloudata.keyvalue.KeyValueProtocol.KeyValueAction;
+import com.cloudata.keyvalue.KeyValueProtocol.ResponseEntry;
 import com.cloudata.values.Value;
 import com.google.common.base.Preconditions;
 import com.google.protobuf.ByteString;
 
-public class AppendOperation implements KeyOperation<Integer> {
+public class AppendOperation extends KeyValueOperationBase {
 
-    private int newLength;
-    final KvEntry entry;
+    public AppendOperation(KeyValueAction entry) {
+        super(entry);
 
-    public AppendOperation(KvEntry entry) {
-        Preconditions.checkState(entry.getAction() == KvAction.APPEND);
+        Preconditions.checkState(entry.getAction() == ActionType.APPEND);
         Preconditions.checkState(!entry.getIfNotExists());
         Preconditions.checkState(!entry.hasIfValue());
-
-        this.entry = entry;
     }
 
     @Override
     public Value doAction(Value oldValue) {
         Value appendValue = Value.deserialize(entry.getValue().asReadOnlyByteBuffer());
 
+        ResponseEntry.Builder eb = response.addEntryBuilder();
+        eb.setKey(entry.getKey());
+
+        Value newValue;
         if (oldValue == null) {
-            this.newLength = entry.getValue().size();
-            return appendValue;
+            newValue = appendValue;
         } else {
-            Value appended = oldValue.concat(appendValue.asBytes());
-
-            this.newLength = appended.sizeAsBytes();
-            return appended;
+            newValue = oldValue.concat(appendValue.asBytes());
         }
-    }
-
-    @Override
-    public KvEntry serialize() {
-        return entry;
-    }
-
-    @Override
-    public Integer getResult() {
-        return newLength;
-    }
+        eb.setValue(ByteString.copyFrom(newValue.asBytes()));
+        eb.setChanged(true);
 
-    @Override
-    public ByteBuffer getKey() {
-        return entry.getKey().asReadOnlyByteBuffer();
+        return newValue;
     }
 
-    public static AppendOperation build(long storeId, ByteString qualifiedKey, Value value) {
-        KvEntry.Builder b = KvEntry.newBuilder();
-        b.setAction(KvAction.APPEND);
+    public static AppendOperation build(long storeId, Keyspace keyspace, ByteString key, Value value) {
+        KeyValueAction.Builder b = KeyValueAction.newBuilder();
+        b.setAction(ActionType.APPEND);
         b.setStoreId(storeId);
-        b.setKey(qualifiedKey);
-        b.setValue(ByteString.copyFrom(value.serialize()));
+        b.setKeyspaceId(keyspace.getKeyspaceId());
+        b.setKey(key);
+        b.setValue(ByteString.copyFrom(value.asBytes()));
         return new AppendOperation(b.build());
     }
 }

cloudata-keyvalue/src/main/java/com/cloudata/keyvalue/operation/CompoundOperation.java

@@ -0,0 +1,56 @@
+package com.cloudata.keyvalue.operation;
+
+import com.cloudata.btree.Btree;
+import com.cloudata.btree.Transaction;
+import com.cloudata.btree.operation.ComplexOperation;
+import com.cloudata.keyvalue.KeyValueProtocol.ActionResponse;
+import com.cloudata.keyvalue.KeyValueProtocol.ActionType;
+import com.cloudata.keyvalue.KeyValueProtocol.KeyValueAction;
+import com.google.common.base.Preconditions;
+
+public class CompoundOperation implements KeyValueOperation, ComplexOperation<ActionResponse> {
+
+    private final KeyValueAction entry;
+
+    final ActionResponse.Builder response = ActionResponse.newBuilder();
+
+    public CompoundOperation(KeyValueAction entry) {
+        Preconditions.checkState(entry.getAction() == ActionType.COMPOUND);
+        Preconditions.checkArgument(entry.hasStoreId());
+
+        this.entry = entry;
+    }
+
+    @Override
+    public KeyValueAction serialize() {
+        return entry;
+    }
+
+    @Override
+    public ActionResponse getResult() {
+        return response.build();
+    }
+
+    @Override
+    public void doAction(Btree btree, Transaction txn) {
+        for (KeyValueAction child : entry.getChildrenList()) {
+            if (child.getStoreId() != entry.getStoreId()) {
+                throw new IllegalArgumentException();
+            }
+            KeyValueOperation op = KeyValueOperations.build(child);
+            txn.doAction(btree, op);
+            response.addChildren(op.getResult());
+        }
+    }
+
+    @Override
+    public boolean isReadOnly() {
+        return KeyValueOperations.isReadOnly(entry);
+    }
+
+    @Override
+    public long getStoreId() {
+        return entry.getStoreId();
+    }
+
+}