sync refactor

source/libs/sync/inc/syncCommit.h

@@ -0,0 +1,35 @@
+/*
+ * 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/>.
+ */
+
+#ifndef _TD_LIBS_SYNC_COMMIT_H
+#define _TD_LIBS_SYNC_COMMIT_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include <stdint.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include "syncInt.h"
+#include "taosdef.h"
+
+void syncMaybeAdvanceCommitIndex(SSyncNode* pSyncNode);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /*_TD_LIBS_SYNC_COMMIT_H*/

source/libs/sync/inc/syncInt.h

@@ -236,7 +236,6 @@ void syncNodeCandidate2Follower(SSyncNode* pSyncNode);
 // raft vote --------------
 void syncNodeVoteForTerm(SSyncNode* pSyncNode, SyncTerm term, SRaftId* pRaftId);
 void syncNodeVoteForSelf(SSyncNode* pSyncNode);
-void syncNodeMaybeAdvanceCommitIndex(SSyncNode* pSyncNode);
 
 // for debug --------------
 void syncNodePrint(SSyncNode* pObj);

source/libs/sync/src/syncAppendEntriesReply.c

@@ -14,6 +14,7 @@
  */
 
 #include "syncAppendEntriesReply.h"
+#include "syncCommit.h"
 #include "syncIndexMgr.h"
 #include "syncInt.h"
 #include "syncRaftLog.h"
@@ -59,7 +60,7 @@ int32_t syncNodeOnAppendEntriesReplyCb(SSyncNode* ths, SyncAppendEntriesReply* p
     syncIndexMgrSetIndex(ths->pMatchIndex, &(pMsg->srcId), pMsg->matchIndex);
 
     // maybe commit
-    syncNodeMaybeAdvanceCommitIndex(ths);
+    syncMaybeAdvanceCommitIndex(ths);
 
   } else {
     SyncIndex nextIndex = syncIndexMgrGetIndex(ths->pNextIndex, &(pMsg->srcId));

source/libs/sync/src/syncCommit.c

@@ -13,6 +13,7 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
+#include "syncCommit.h"
 #include "syncIndexMgr.h"
 #include "syncInt.h"
 #include "syncRaftLog.h"
@@ -40,7 +41,7 @@
 //        IN commitIndex' = [commitIndex EXCEPT ![i] = newCommitIndex]
 //     /\ UNCHANGED <<messages, serverVars, candidateVars, leaderVars, log>>
 //
-void syncNodeMaybeAdvanceCommitIndex(SSyncNode* pSyncNode) {
+void syncMaybeAdvanceCommitIndex(SSyncNode* pSyncNode) {
   syncIndexMgrLog2("==syncNodeMaybeAdvanceCommitIndex== pNextIndex", pSyncNode->pNextIndex);
   syncIndexMgrLog2("==syncNodeMaybeAdvanceCommitIndex== pMatchIndex", pSyncNode->pMatchIndex);
 

source/libs/sync/src/syncMain.c

@@ -17,6 +17,7 @@
 #include "sync.h"
 #include "syncAppendEntries.h"
 #include "syncAppendEntriesReply.h"
+#include "syncCommit.h"
 #include "syncElection.h"
 #include "syncEnv.h"
 #include "syncIndexMgr.h"
@@ -740,7 +741,7 @@ static int32_t syncNodeOnClientRequestCb(SSyncNode* ths, SyncClientRequest* pMsg
     ths->pLogStore->appendEntry(ths->pLogStore, pEntry);
 
     // only myself, maybe commit
-    syncNodeMaybeAdvanceCommitIndex(ths);
+    syncMaybeAdvanceCommitIndex(ths);
 
     // start replicate right now!
     syncNodeReplicate(ths);

source/libs/sync/test/syncWriteTest.cpp

@@ -162,7 +162,7 @@ int main(int argc, char **argv) {
   SyncClientRequest *pMsg1 = step1(pMsg0);
   syncClientRequestPrint2((char *)"==step1==", pMsg1);
 
-  for (int i = 0; i < 5; ++i) {
+  for (int i = 0; i < 10; ++i) {
     SyncClientRequest *pSyncClientRequest = pMsg1;
     SRpcMsg            rpcMsg;
     syncClientRequest2RpcMsg(pSyncClientRequest, &rpcMsg);