sync refactor

source/libs/sync/src/syncAppendEntriesReply.c

@@ -51,10 +51,10 @@ int32_t syncNodeOnAppendEntriesReplyCb(SSyncNode* ths, SyncAppendEntriesReply* p
   assert(pMsg->term == ths->pRaftStore->currentTerm);
 
   if (pMsg->success) {
-    // nextIndex = reply.matchIndex + 1
+    // nextIndex'  = [nextIndex  EXCEPT ![i][j] = m.mmatchIndex + 1]
     syncIndexMgrSetIndex(ths->pNextIndex, &(pMsg->srcId), pMsg->matchIndex + 1);
 
-    // matchIndex = reply.matchIndex
+    // matchIndex' = [matchIndex EXCEPT ![i][j] = m.mmatchIndex]
     syncIndexMgrSetIndex(ths->pMatchIndex, &(pMsg->srcId), pMsg->matchIndex);
 
     // maybe commit
@@ -62,6 +62,8 @@ int32_t syncNodeOnAppendEntriesReplyCb(SSyncNode* ths, SyncAppendEntriesReply* p
 
   } else {
     SyncIndex nextIndex = syncIndexMgrGetIndex(ths->pNextIndex, &(pMsg->srcId));
+
+    // notice! int64, uint64
     if (nextIndex > SYNC_INDEX_BEGIN) {
       --nextIndex;
     } else {

source/libs/sync/src/syncCommit.c

@@ -0,0 +1,41 @@
+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "syncInt.h"
+
+// \* Leader i advances its commitIndex.
+// \* This is done as a separate step from handling AppendEntries responses,
+// \* in part to minimize atomic regions, and in part so that leaders of
+// \* single-server clusters are able to mark entries committed.
+// AdvanceCommitIndex(i) ==
+//     /\ state[i] = Leader
+//     /\ LET \* The set of servers that agree up through index.
+//            Agree(index) == {i} \cup {k \in Server :
+//                                          matchIndex[i][k] >= index}
+//            \* The maximum indexes for which a quorum agrees
+//            agreeIndexes == {index \in 1..Len(log[i]) :
+//                                 Agree(index) \in Quorum}
+//            \* New value for commitIndex'[i]
+//            newCommitIndex ==
+//               IF /\ agreeIndexes /= {}
+//                  /\ log[i][Max(agreeIndexes)].term = currentTerm[i]
+//               THEN
+//                   Max(agreeIndexes)
+//               ELSE
+//                   commitIndex[i]
+//        IN commitIndex' = [commitIndex EXCEPT ![i] = newCommitIndex]
+//     /\ UNCHANGED <<messages, serverVars, candidateVars, leaderVars, log>>
+//
+void syncNodeMaybeAdvanceCommitIndex(SSyncNode* pSyncNode) {}

source/libs/sync/src/syncMain.c

@@ -617,8 +617,6 @@ void syncNodeVoteForSelf(SSyncNode* pSyncNode) {
   syncRequestVoteReplyDestroy(pMsg);
 }
 
-void syncNodeMaybeAdvanceCommitIndex(SSyncNode* pSyncNode) {}
-
 // for debug --------------
 void syncNodePrint(SSyncNode* pObj) {
   char* serialized = syncNode2Str(pObj);