Feature: add statemachine for dledger (#128)

* feature: add statemachine add statemachine caller

* feature: add fsm caller to dledgerEntryPusher

* feature: Improve the function of the state machine

* style: add apach header

* style: add apache header

Author: hzh0425

src/main/java/io/openmessaging/storage/dledger/DLedgerEntryPusher.java

@@ -23,6 +23,8 @@
 import io.openmessaging.storage.dledger.protocol.DLedgerResponseCode;
 import io.openmessaging.storage.dledger.protocol.PushEntryRequest;
 import io.openmessaging.storage.dledger.protocol.PushEntryResponse;
+import io.openmessaging.storage.dledger.statemachine.StateMachine;
+import io.openmessaging.storage.dledger.statemachine.StateMachineCaller;
 import io.openmessaging.storage.dledger.store.DLedgerMemoryStore;
 import io.openmessaging.storage.dledger.store.DLedgerStore;
 import io.openmessaging.storage.dledger.store.file.DLedgerMmapFileStore;
@@ -34,6 +36,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CompletableFuture;
@@ -65,6 +68,8 @@ public class DLedgerEntryPusher {
 
     private Map<String, EntryDispatcher> dispatcherMap = new HashMap<>();
 
+    private Optional<StateMachineCaller> fsmCaller;
+
     public DLedgerEntryPusher(DLedgerConfig dLedgerConfig, MemberState memberState, DLedgerStore dLedgerStore,
         DLedgerRpcService dLedgerRpcService) {
         this.dLedgerConfig = dLedgerConfig;
@@ -96,6 +101,10 @@ public void shutdown() {
         }
     }
 
+    public void registerStateMachine(final Optional<StateMachineCaller> fsmCaller) {
+        this.fsmCaller = fsmCaller;
+    }
+
     public CompletableFuture<PushEntryResponse> handlePush(PushEntryRequest request) throws Exception {
         return entryHandler.handlePush(request);
     }
@@ -175,6 +184,11 @@ public void wakeUpDispatchers() {
         }
     }
 
+    private void updateCommittedIndex(final long term, final long committedIndex) {
+        dLedgerStore.updateCommittedIndex(term, committedIndex);
+        this.fsmCaller.ifPresent(caller -> caller.onCommitted(committedIndex));
+    }
+
     /**
      * This thread will check the quorum index and complete the pending requests.
      */
@@ -236,7 +250,7 @@ public void doWork() {
                         .sorted(Comparator.reverseOrder())
                         .collect(Collectors.toList());
                 long quorumIndex = sortedWaterMarks.get(sortedWaterMarks.size() / 2);
-                dLedgerStore.updateCommittedIndex(currTerm, quorumIndex);
+                updateCommittedIndex(currTerm, quorumIndex);
                 ConcurrentMap<Long, TimeoutFuture<AppendEntryResponse>> responses = pendingAppendResponsesByTerm.get(currTerm);
                 boolean needCheck = false;
                 int ackNum = 0;
@@ -830,7 +844,7 @@ private void handleDoAppend(long writeIndex, PushEntryRequest request,
                 DLedgerEntry entry = dLedgerStore.appendAsFollower(request.getEntry(), request.getTerm(), request.getLeaderId());
                 PreConditions.check(entry.getIndex() == writeIndex, DLedgerResponseCode.INCONSISTENT_STATE);
                 future.complete(buildResponse(request, DLedgerResponseCode.SUCCESS.getCode()));
-                dLedgerStore.updateCommittedIndex(request.getTerm(), request.getCommitIndex());
+                updateCommittedIndex(request.getTerm(), request.getCommitIndex());
             } catch (Throwable t) {
                 logger.error("[HandleDoWrite] writeIndex={}", writeIndex, t);
                 future.complete(buildResponse(request, DLedgerResponseCode.INCONSISTENT_STATE.getCode()));
@@ -857,7 +871,7 @@ private CompletableFuture<PushEntryResponse> handleDoCommit(long committedIndex,
             try {
                 PreConditions.check(committedIndex == request.getCommitIndex(), DLedgerResponseCode.UNKNOWN);
                 PreConditions.check(request.getType() == PushEntryRequest.Type.COMMIT, DLedgerResponseCode.UNKNOWN);
-                dLedgerStore.updateCommittedIndex(request.getTerm(), committedIndex);
+                updateCommittedIndex(request.getTerm(), committedIndex);
                 future.complete(buildResponse(request, DLedgerResponseCode.SUCCESS.getCode()));
             } catch (Throwable t) {
                 logger.error("[HandleDoCommit] committedIndex={}", request.getCommitIndex(), t);
@@ -875,7 +889,7 @@ private CompletableFuture<PushEntryResponse> handleDoTruncate(long truncateIndex
                 long index = dLedgerStore.truncate(request.getEntry(), request.getTerm(), request.getLeaderId());
                 PreConditions.check(index == truncateIndex, DLedgerResponseCode.INCONSISTENT_STATE);
                 future.complete(buildResponse(request, DLedgerResponseCode.SUCCESS.getCode()));
-                dLedgerStore.updateCommittedIndex(request.getTerm(), request.getCommitIndex());
+                updateCommittedIndex(request.getTerm(), request.getCommitIndex());
             } catch (Throwable t) {
                 logger.error("[HandleDoTruncate] truncateIndex={}", truncateIndex, t);
                 future.complete(buildResponse(request, DLedgerResponseCode.INCONSISTENT_STATE.getCode()));
@@ -891,7 +905,7 @@ private void handleDoBatchAppend(long writeIndex, PushEntryRequest request,
                     dLedgerStore.appendAsFollower(entry, request.getTerm(), request.getLeaderId());
                 }
                 future.complete(buildBatchAppendResponse(request, DLedgerResponseCode.SUCCESS.getCode()));
-                dLedgerStore.updateCommittedIndex(request.getTerm(), request.getCommitIndex());
+                updateCommittedIndex(request.getTerm(), request.getCommitIndex());
             } catch (Throwable t) {
                 logger.error("[HandleDoBatchAppend]", t);
             }

src/main/java/io/openmessaging/storage/dledger/DLedgerServer.java

@@ -37,6 +37,8 @@
 import io.openmessaging.storage.dledger.protocol.PushEntryResponse;
 import io.openmessaging.storage.dledger.protocol.VoteRequest;
 import io.openmessaging.storage.dledger.protocol.VoteResponse;
+import io.openmessaging.storage.dledger.statemachine.StateMachine;
+import io.openmessaging.storage.dledger.statemachine.StateMachineCaller;
 import io.openmessaging.storage.dledger.store.DLedgerMemoryStore;
 import io.openmessaging.storage.dledger.store.DLedgerStore;
 import io.openmessaging.storage.dledger.store.file.DLedgerMmapFileStore;
@@ -49,12 +51,12 @@
 import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Optional;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.CompletableFuture;
 
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -71,6 +73,7 @@ public class DLedgerServer implements DLedgerProtocolHander {
     private DLedgerLeaderElector dLedgerLeaderElector;
 
     private ScheduledExecutorService executorService;
+    private Optional<StateMachineCaller> fsmCaller;
 
     public DLedgerServer(DLedgerConfig dLedgerConfig) {
         this.dLedgerConfig = dLedgerConfig;
@@ -87,7 +90,6 @@ public DLedgerServer(DLedgerConfig dLedgerConfig) {
         });
     }
 
-
     public void startup() {
         this.dLedgerStore.startup();
         this.dLedgerRpcService.startup();
@@ -102,6 +104,7 @@ public void shutdown() {
         this.dLedgerRpcService.shutdown();
         this.dLedgerStore.shutdown();
         executorService.shutdown();
+        this.fsmCaller.ifPresent(StateMachineCaller::shutdown);
     }
 
     private DLedgerStore createDLedgerStore(String storeType, DLedgerConfig config, MemberState memberState) {
@@ -116,6 +119,18 @@ public MemberState getMemberState() {
         return memberState;
     }
 
+    public void registerStateMachine(final StateMachine fsm) {
+        final StateMachineCaller fsmCaller = new StateMachineCaller(this.dLedgerStore, fsm);
+        fsmCaller.start();
+        this.fsmCaller = Optional.of(fsmCaller);
+        // Register state machine caller to entry pusher
+        this.dLedgerEntryPusher.registerStateMachine(this.fsmCaller);
+    }
+
+    public StateMachine getStateMachine() {
+        return this.fsmCaller.map(StateMachineCaller::getStateMachine).orElse(null);
+    }
+
     @Override public CompletableFuture<HeartBeatResponse> handleHeartBeat(HeartBeatRequest request) throws Exception {
         try {
 
@@ -190,12 +205,12 @@ public CompletableFuture<AppendEntryResponse> handleAppend(AppendEntryRequest re
                         }
                         // only wait last entry ack is ok
                         BatchAppendFuture<AppendEntryResponse> batchAppendFuture =
-                                (BatchAppendFuture<AppendEntryResponse>) dLedgerEntryPusher.waitAck(resEntry, true);
+                            (BatchAppendFuture<AppendEntryResponse>) dLedgerEntryPusher.waitAck(resEntry, true);
                         batchAppendFuture.setPositions(positions);
                         return batchAppendFuture;
                     }
                     throw new DLedgerException(DLedgerResponseCode.REQUEST_WITH_EMPTY_BODYS, "BatchAppendEntryRequest" +
-                            " with empty bodys");
+                        " with empty bodys");
                 } else {
                     DLedgerEntry dLedgerEntry = new DLedgerEntry();
                     dLedgerEntry.setBody(request.getBody());
@@ -278,7 +293,8 @@ public CompletableFuture<PullEntriesResponse> handlePull(PullEntriesRequest requ
     }
 
     @Override
-    public CompletableFuture<LeadershipTransferResponse> handleLeadershipTransfer(LeadershipTransferRequest request) throws Exception {
+    public CompletableFuture<LeadershipTransferResponse> handleLeadershipTransfer(
+        LeadershipTransferRequest request) throws Exception {
         try {
             PreConditions.check(memberState.getSelfId().equals(request.getRemoteId()), DLedgerResponseCode.UNKNOWN_MEMBER, "%s != %s", request.getRemoteId(), memberState.getSelfId());
             PreConditions.check(memberState.getGroup().equals(request.getGroup()), DLedgerResponseCode.UNKNOWN_GROUP, "%s != %s", request.getGroup(), memberState.getGroup());
@@ -385,7 +401,7 @@ private void checkPreferredLeader() {
             }
         }
         logger.info("preferredLeaderId = {}, which has the smallest fall behind index = {} and is decided to be transferee.", preferredLeaderId, minFallBehind);
-        
+
         if (minFallBehind < dLedgerConfig.getMaxLeadershipTransferWaitIndex()) {
             LeadershipTransferRequest request = new LeadershipTransferRequest();
             request.setTerm(memberState.currTerm());

src/main/java/io/openmessaging/storage/dledger/snapshot/SnapshotReader.java

@@ -0,0 +1,23 @@
+/*
+ * Copyright 2017-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.openmessaging.storage.dledger.snapshot;
+
+/**
+ * Reader for snapshot
+ */
+public interface SnapshotReader {
+}

src/main/java/io/openmessaging/storage/dledger/snapshot/SnapshotWriter.java

@@ -0,0 +1,23 @@
+/*
+ * Copyright 2017-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.openmessaging.storage.dledger.snapshot;
+
+/**
+ * Writer for snapshot
+ */
+public interface SnapshotWriter {
+}

src/main/java/io/openmessaging/storage/dledger/statemachine/CommittedEntryIterator.java

@@ -0,0 +1,62 @@
+/*
+ * Copyright 2017-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.openmessaging.storage.dledger.statemachine;
+
+import java.util.Iterator;
+import java.util.concurrent.atomic.AtomicLong;
+
+import io.openmessaging.storage.dledger.entry.DLedgerEntry;
+import io.openmessaging.storage.dledger.store.DLedgerStore;
+
+/**
+ * The iterator implementation of committed entries.
+ */
+public class CommittedEntryIterator implements Iterator<DLedgerEntry> {
+
+    private final DLedgerStore dLedgerStore;
+    private final long         committedIndex;
+    private final AtomicLong   applyingIndex;
+    private long               currentIndex;
+
+    public CommittedEntryIterator(final DLedgerStore dLedgerStore, final long committedIndex,
+                                  final AtomicLong applyingIndex, final long lastAppliedIndex) {
+        this.dLedgerStore = dLedgerStore;
+        this.committedIndex = committedIndex;
+        this.applyingIndex = applyingIndex;
+        this.currentIndex = lastAppliedIndex;
+    }
+
+    @Override
+    public boolean hasNext() {
+        return this.currentIndex < this.committedIndex;
+    }
+
+    @Override
+    public DLedgerEntry next() {
+        ++this.currentIndex;
+        if (this.currentIndex <= this.committedIndex) {
+            final DLedgerEntry dLedgerEntry = this.dLedgerStore.get(this.currentIndex);
+            this.applyingIndex.set(this.currentIndex);
+            return dLedgerEntry;
+        }
+        return null;
+    }
+
+    public long getIndex() {
+        return this.currentIndex;
+    }
+}

src/main/java/io/openmessaging/storage/dledger/statemachine/StateMachine.java

@@ -0,0 +1,59 @@
+/*
+ * Copyright 2017-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.openmessaging.storage.dledger.statemachine;
+
+import java.util.concurrent.CompletableFuture;
+
+import io.openmessaging.storage.dledger.snapshot.SnapshotReader;
+import io.openmessaging.storage.dledger.snapshot.SnapshotWriter;
+
+/**
+ * Finite state machine, which should be implemented by user.
+ */
+public interface StateMachine {
+
+    /**
+     * Update the user statemachine with a batch a tasks that can be accessed
+     * through |iterator|.
+     *
+     * @param iter iterator of committed entry
+     */
+    void onApply(final CommittedEntryIterator iter);
+
+    /**
+     * User defined snapshot generate function, this method will block StateMachine#onApply(Iterator).
+     * Call done.run(status) when snapshot finished.
+     *
+     * @param writer snapshot writer
+     * @param done   callback
+     */
+    void onSnapshotSave(final SnapshotWriter writer, final CompletableFuture<Boolean> done);
+
+    /**
+     * User defined snapshot load function.
+     *
+     * @param reader snapshot reader
+     * @return true on success
+     */
+    boolean onSnapshotLoad(final SnapshotReader reader);
+
+    /**
+     * Invoked once when the raft node was shut down.
+     * Default do nothing
+     */
+    void onShutdown();
+}