sync refactor

source/libs/sync/inc/syncCommit.h

@@ -26,7 +26,32 @@ extern "C" {
 #include "syncInt.h"
 #include "taosdef.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 syncMaybeAdvanceCommitIndex(SSyncNode* pSyncNode);
+bool syncAgreeIndex(SSyncNode* pSyncNode, SRaftId* pRaftId, SyncIndex index);
+bool syncAgree(SSyncNode* pSyncNode, SyncIndex index);
 
 #ifdef __cplusplus
 }

source/libs/sync/src/syncCommit.c

@@ -46,25 +46,61 @@ void syncMaybeAdvanceCommitIndex(SSyncNode* pSyncNode) {
   syncIndexMgrLog2("==syncNodeMaybeAdvanceCommitIndex== pMatchIndex", pSyncNode->pMatchIndex);
 
   // update commit index
+  SyncIndex newCommitIndex = pSyncNode->commitIndex;
+  for (SyncIndex index = pSyncNode->pLogStore->getLastIndex(pSyncNode->pLogStore); index > pSyncNode->commitIndex;
+       ++index) {
+    if (syncAgree(pSyncNode, index)) {
+      newCommitIndex = index;
+      break;
+    }
+  }
 
-  if (pSyncNode->pFsm != NULL) {
+  if (newCommitIndex > pSyncNode->commitIndex) {
-    SyncIndex beginIndex = SYNC_INDEX_INVALID;
+    SyncIndex beginIndex = pSyncNode->commitIndex + 1;
-    SyncIndex endIndex = SYNC_INDEX_INVALID;
+    SyncIndex endIndex = newCommitIndex;
-    for (SyncIndex i = beginIndex; i <= endIndex; ++i) {
+    pSyncNode->commitIndex = newCommitIndex;
-      if (i != SYNC_INDEX_INVALID) {
-        SSyncRaftEntry* pEntry = pSyncNode->pLogStore->getEntry(pSyncNode->pLogStore, i);
-        assert(pEntry != NULL);
 
-        SRpcMsg rpcMsg;
+    if (pSyncNode->pFsm != NULL) {
-        syncEntry2OriginalRpc(pEntry, &rpcMsg);
+      for (SyncIndex i = beginIndex; i <= endIndex; ++i) {
+        if (i != SYNC_INDEX_INVALID) {
+          SSyncRaftEntry* pEntry = pSyncNode->pLogStore->getEntry(pSyncNode->pLogStore, i);
+          assert(pEntry != NULL);
 
-        if (pSyncNode->pFsm->FpCommitCb != NULL) {
+          SRpcMsg rpcMsg;
-          pSyncNode->pFsm->FpCommitCb(pSyncNode->pFsm, &rpcMsg, pEntry->index, pEntry->isWeak, 0);
+          syncEntry2OriginalRpc(pEntry, &rpcMsg);
+
+          if (pSyncNode->pFsm->FpCommitCb != NULL) {
+            pSyncNode->pFsm->FpCommitCb(pSyncNode->pFsm, &rpcMsg, pEntry->index, pEntry->isWeak, 0);
+          }
+
+          rpcFreeCont(rpcMsg.pCont);
+          syncEntryDestory(pEntry);
         }
-
-        rpcFreeCont(rpcMsg.pCont);
-        syncEntryDestory(pEntry);
       }
     }
   }
+}
+
+bool syncAgreeIndex(SSyncNode* pSyncNode, SRaftId* pRaftId, SyncIndex index) {
+  SyncIndex matchIndex = syncIndexMgrGetIndex(pSyncNode->pMatchIndex, pRaftId);
+
+  // b for debug
+  bool b = false;
+  if (matchIndex >= index) {
+    b = true;
+  }
+  return b;
+}
+
+bool syncAgree(SSyncNode* pSyncNode, SyncIndex index) {
+  int agreeCount = 0;
+  for (int i = 0; i < pSyncNode->replicaNum; ++i) {
+    if (syncAgreeIndex(pSyncNode, &(pSyncNode->replicasId[i]), index)) {
+      ++agreeCount;
+    }
+    if (agreeCount >= pSyncNode->quorum) {
+      return true;
+    }
+  }
+  return false;
 }