enh(sync): add snapshotEnable2

include/libs/sync/syncTools.h

@@ -394,9 +394,26 @@ int32_t syncNodeOnRequestVoteCb(SSyncNode* ths, SyncRequestVote* pMsg);
 int32_t syncNodeOnRequestVoteReplyCb(SSyncNode* ths, SyncRequestVoteReply* pMsg);
 int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg);
 int32_t syncNodeOnAppendEntriesReplyCb(SSyncNode* ths, SyncAppendEntriesReply* pMsg);
+int32_t syncNodeOnRequestVoteSnapshotCb(SSyncNode* ths, SyncRequestVote* pMsg);
+int32_t syncNodeOnRequestVoteReplySnapshotCb(SSyncNode* ths, SyncRequestVoteReply* pMsg);
+int32_t syncNodeOnAppendEntriesSnapshotCb(SSyncNode* ths, SyncAppendEntries* pMsg);
+int32_t syncNodeOnAppendEntriesReplySnapshotCb(SSyncNode* ths, SyncAppendEntriesReply* pMsg);
+
+// option ----------------------------------
+typedef int32_t (*FpOnPingCb)(SSyncNode* ths, SyncPing* pMsg);
+typedef int32_t (*FpOnPingReplyCb)(SSyncNode* ths, SyncPingReply* pMsg);
+typedef int32_t (*FpOnClientRequestCb)(SSyncNode* ths, SyncClientRequest* pMsg);
+typedef int32_t (*FpOnRequestVoteCb)(SSyncNode* ths, SyncRequestVote* pMsg);
+typedef int32_t (*FpOnRequestVoteReplyCb)(SSyncNode* ths, SyncRequestVoteReply* pMsg);
+typedef int32_t (*FpOnAppendEntriesCb)(SSyncNode* ths, SyncAppendEntries* pMsg);
+typedef int32_t (*FpOnAppendEntriesReplyCb)(SSyncNode* ths, SyncAppendEntriesReply* pMsg);
+typedef int32_t (*FpOnTimeoutCb)(SSyncNode* pSyncNode, SyncTimeout* pMsg);
 
 // ---------------------------------------------
 
+bool syncNodeSnapshotEnable(SSyncNode* pSyncNode);
+// ---------------------------------------------
+
 #ifdef __cplusplus
 }
 #endif

source/dnode/mnode/impl/inc/mndInt.h

@@ -19,6 +19,7 @@
 #include "mndDef.h"
 
 #include "sdb.h"
+#include "sync.h"
 #include "syncTools.h"
 #include "tcache.h"
 #include "tdatablock.h"

source/dnode/mnode/impl/src/mnode.c

@@ -368,6 +368,48 @@ int32_t mndProcessSyncMsg(SRpcMsg *pMsg) {
   syncRpcMsgLog2(logBuf, pMsg);
   taosMemoryFree(syncNodeStr);
 
+
+  // ugly! use function pointer 
+  if (syncNodeSnapshotEnable(pSyncNode)) {
+    if (pMsg->msgType == TDMT_VND_SYNC_TIMEOUT) {
+      SyncTimeout *pSyncMsg = syncTimeoutFromRpcMsg2(pMsg);
+      code = syncNodeOnTimeoutCb(pSyncNode, pSyncMsg);
+      syncTimeoutDestroy(pSyncMsg);
+    } else if (pMsg->msgType == TDMT_VND_SYNC_PING) {
+      SyncPing *pSyncMsg = syncPingFromRpcMsg2(pMsg);
+      code = syncNodeOnPingCb(pSyncNode, pSyncMsg);
+      syncPingDestroy(pSyncMsg);
+    } else if (pMsg->msgType == TDMT_VND_SYNC_PING_REPLY) {
+      SyncPingReply *pSyncMsg = syncPingReplyFromRpcMsg2(pMsg);
+      code = syncNodeOnPingReplyCb(pSyncNode, pSyncMsg);
+      syncPingReplyDestroy(pSyncMsg);
+    } else if (pMsg->msgType == TDMT_VND_SYNC_CLIENT_REQUEST) {
+      SyncClientRequest *pSyncMsg = syncClientRequestFromRpcMsg2(pMsg);
+      code = syncNodeOnClientRequestCb(pSyncNode, pSyncMsg);
+      syncClientRequestDestroy(pSyncMsg);
+
+    } else if (pMsg->msgType == TDMT_VND_SYNC_REQUEST_VOTE) {
+      SyncRequestVote *pSyncMsg = syncRequestVoteFromRpcMsg2(pMsg);
+      code = syncNodeOnRequestVoteSnapshotCb(pSyncNode, pSyncMsg);
+      syncRequestVoteDestroy(pSyncMsg);
+    } else if (pMsg->msgType == TDMT_VND_SYNC_REQUEST_VOTE_REPLY) {
+      SyncRequestVoteReply *pSyncMsg = syncRequestVoteReplyFromRpcMsg2(pMsg);
+      code = syncNodeOnRequestVoteReplySnapshotCb(pSyncNode, pSyncMsg);
+      syncRequestVoteReplyDestroy(pSyncMsg);
+    } else if (pMsg->msgType == TDMT_VND_SYNC_APPEND_ENTRIES) {
+      SyncAppendEntries *pSyncMsg = syncAppendEntriesFromRpcMsg2(pMsg);
+      code = syncNodeOnAppendEntriesSnapshotCb(pSyncNode, pSyncMsg);
+      syncAppendEntriesDestroy(pSyncMsg);
+    } else if (pMsg->msgType == TDMT_VND_SYNC_APPEND_ENTRIES_REPLY) {
+      SyncAppendEntriesReply *pSyncMsg = syncAppendEntriesReplyFromRpcMsg2(pMsg);
+      code = syncNodeOnAppendEntriesReplySnapshotCb(pSyncNode, pSyncMsg);
+      syncAppendEntriesReplyDestroy(pSyncMsg);
+    } else {
+      mError("failed to process msg:%p since invalid type:%s", pMsg, TMSG_INFO(pMsg->msgType));
+      code = TAOS_SYNC_PROPOSE_OTHER_ERROR;
+    }
+
+  } else {
     if (pMsg->msgType == TDMT_VND_SYNC_TIMEOUT) {
       SyncTimeout *pSyncMsg = syncTimeoutFromRpcMsg2(pMsg);
       code = syncNodeOnTimeoutCb(pSyncNode, pSyncMsg);
@@ -404,6 +446,7 @@ int32_t mndProcessSyncMsg(SRpcMsg *pMsg) {
       mError("failed to process msg:%p since invalid type:%s", pMsg, TMSG_INFO(pMsg->msgType));
       code = TAOS_SYNC_PROPOSE_OTHER_ERROR;
     }
+  }
 
   return code;
 }

source/libs/sync/inc/syncAppendEntries.h

@@ -93,6 +93,7 @@ extern "C" {
 //       /\ UNCHANGED <<candidateVars, leaderVars>>
 //
 int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg);
+int32_t syncNodeOnAppendEntriesSnapshotCb(SSyncNode* ths, SyncAppendEntries* pMsg);
 
 #ifdef __cplusplus
 }

source/libs/sync/inc/syncAppendEntriesReply.h

@@ -41,6 +41,7 @@ extern "C" {
 //    /\ UNCHANGED <<serverVars, candidateVars, logVars, elections>>
 //
 int32_t syncNodeOnAppendEntriesReplyCb(SSyncNode* ths, SyncAppendEntriesReply* pMsg);
+int32_t syncNodeOnAppendEntriesReplySnapshotCb(SSyncNode* ths, SyncAppendEntriesReply* pMsg);
 
 #ifdef __cplusplus
 }

source/libs/sync/inc/syncInt.h

@@ -137,14 +137,25 @@ typedef struct SSyncNode {
   uint64_t          heartbeatTimerCounter;
 
   // callback
+  FpOnPingCb               FpOnPing;
+  FpOnPingReplyCb          FpOnPingReply;
+  FpOnClientRequestCb      FpOnClientRequest;
+  FpOnTimeoutCb            FpOnTimeout;
+  FpOnRequestVoteCb        FpOnRequestVote;
+  FpOnRequestVoteReplyCb   FpOnRequestVoteReply;
+  FpOnAppendEntriesCb      FpOnAppendEntries;
+  FpOnAppendEntriesReplyCb FpOnAppendEntriesReply;
+
+  /*
   int32_t (*FpOnPing)(SSyncNode* ths, SyncPing* pMsg);
   int32_t (*FpOnPingReply)(SSyncNode* ths, SyncPingReply* pMsg);
   int32_t (*FpOnClientRequest)(SSyncNode* ths, SyncClientRequest* pMsg);
+  int32_t (*FpOnTimeout)(SSyncNode* pSyncNode, SyncTimeout* pMsg);
   int32_t (*FpOnRequestVote)(SSyncNode* ths, SyncRequestVote* pMsg);
   int32_t (*FpOnRequestVoteReply)(SSyncNode* ths, SyncRequestVoteReply* pMsg);
   int32_t (*FpOnAppendEntries)(SSyncNode* ths, SyncAppendEntries* pMsg);
   int32_t (*FpOnAppendEntriesReply)(SSyncNode* ths, SyncAppendEntriesReply* pMsg);
-  int32_t (*FpOnTimeout)(SSyncNode* pSyncNode, SyncTimeout* pMsg);
+  */
 
   // tools
   SSyncRespMgr* pSyncRespMgr;
@@ -164,6 +175,9 @@ void       syncNodeStart(SSyncNode* pSyncNode);
 void       syncNodeStartStandBy(SSyncNode* pSyncNode);
 void       syncNodeClose(SSyncNode* pSyncNode);
 
+// option
+bool syncNodeSnapshotEnable(SSyncNode* pSyncNode);
+
 // ping --------------
 int32_t syncNodePing(SSyncNode* pSyncNode, const SRaftId* destRaftId, SyncPing* pMsg);
 int32_t syncNodePingSelf(SSyncNode* pSyncNode);

source/libs/sync/inc/syncReplication.h

@@ -52,6 +52,7 @@ extern "C" {
 //    /\ UNCHANGED <<serverVars, candidateVars, leaderVars, logVars>>
 //
 int32_t syncNodeAppendEntriesPeers(SSyncNode* pSyncNode);
+int32_t syncNodeAppendEntriesPeersSnapshot(SSyncNode* pSyncNode);
 int32_t syncNodeReplicate(SSyncNode* pSyncNode);
 int32_t syncNodeAppendEntries(SSyncNode* pSyncNode, const SRaftId* destRaftId, const SyncAppendEntries* pMsg);
 

source/libs/sync/inc/syncRequestVote.h

@@ -50,6 +50,7 @@ extern "C" {
 //       /\ UNCHANGED <<state, currentTerm, candidateVars, leaderVars, logVars>>
 //
 int32_t syncNodeOnRequestVoteCb(SSyncNode* ths, SyncRequestVote* pMsg);
+int32_t syncNodeOnRequestVoteSnapshotCb(SSyncNode* ths, SyncRequestVote* pMsg);
 
 #ifdef __cplusplus
 }

source/libs/sync/inc/syncRequestVoteReply.h

@@ -45,6 +45,7 @@ extern "C" {
 //    /\ UNCHANGED <<serverVars, votedFor, leaderVars, logVars>>
 //
 int32_t syncNodeOnRequestVoteReplyCb(SSyncNode* ths, SyncRequestVoteReply* pMsg);
+int32_t syncNodeOnRequestVoteReplySnapshotCb(SSyncNode* ths, SyncRequestVoteReply* pMsg);
 
 #ifdef __cplusplus
 }

source/libs/sync/src/syncAppendEntries.c

@@ -427,3 +427,5 @@ int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) {
 
   return ret;
 }
+
+int32_t syncNodeOnAppendEntriesSnapshotCb(SSyncNode* ths, SyncAppendEntries* pMsg) { return 0; }
\ No newline at end of file

source/libs/sync/src/syncAppendEntriesReply.c

@@ -94,3 +94,5 @@ int32_t syncNodeOnAppendEntriesReplyCb(SSyncNode* ths, SyncAppendEntriesReply* p
 
   return ret;
 }
+
+int32_t syncNodeOnAppendEntriesReplySnapshotCb(SSyncNode* ths, SyncAppendEntriesReply* pMsg) { return 0; }
\ No newline at end of file

source/libs/sync/src/syncElection.c

@@ -15,6 +15,7 @@
 
 #include "syncElection.h"
 #include "syncMessage.h"
+#include "syncRaftCfg.h"
 #include "syncRaftStore.h"
 #include "syncVoteMgr.h"
 
@@ -73,7 +74,12 @@ int32_t syncNodeElect(SSyncNode* pSyncNode) {
     return ret;
   }
 
+  if (pSyncNode->pRaftCfg->snapshotEnable) {
+    ret = syncNodeRequestVotePeersSnapshot(pSyncNode);
+  } else {
     ret = syncNodeRequestVotePeers(pSyncNode);
+  }
+
   assert(ret == 0);
   syncNodeResetElectTimer(pSyncNode);
 

source/libs/sync/src/syncMain.c

@@ -555,11 +555,19 @@ SSyncNode* syncNodeOpen(const SSyncInfo* pOldSyncInfo) {
   pSyncNode->FpOnPing = syncNodeOnPingCb;
   pSyncNode->FpOnPingReply = syncNodeOnPingReplyCb;
   pSyncNode->FpOnClientRequest = syncNodeOnClientRequestCb;
+  pSyncNode->FpOnTimeout = syncNodeOnTimeoutCb;
+
+  if (pSyncNode->pRaftCfg->snapshotEnable) {
     pSyncNode->FpOnRequestVote = syncNodeOnRequestVoteCb;
     pSyncNode->FpOnRequestVoteReply = syncNodeOnRequestVoteReplyCb;
     pSyncNode->FpOnAppendEntries = syncNodeOnAppendEntriesCb;
     pSyncNode->FpOnAppendEntriesReply = syncNodeOnAppendEntriesReplyCb;
-  pSyncNode->FpOnTimeout = syncNodeOnTimeoutCb;
+  } else {
+    pSyncNode->FpOnRequestVote = syncNodeOnRequestVoteSnapshotCb;
+    pSyncNode->FpOnRequestVoteReply = syncNodeOnRequestVoteReplySnapshotCb;
+    pSyncNode->FpOnAppendEntries = syncNodeOnAppendEntriesSnapshotCb;
+    pSyncNode->FpOnAppendEntriesReply = syncNodeOnAppendEntriesReplySnapshotCb;
+  }
 
   // tools
   pSyncNode->pSyncRespMgr = syncRespMgrCreate(NULL, 0);
@@ -675,6 +683,9 @@ void syncNodeClose(SSyncNode* pSyncNode) {
   // taosMemoryFree(pSyncNode);
 }
 
+// option
+bool syncNodeSnapshotEnable(SSyncNode* pSyncNode) { return pSyncNode->pRaftCfg->snapshotEnable; }
+
 // ping --------------
 int32_t syncNodePing(SSyncNode* pSyncNode, const SRaftId* destRaftId, SyncPing* pMsg) {
   syncPingLog2((char*)"==syncNodePing==", pMsg);

source/libs/sync/src/syncReplication.c

@@ -16,6 +16,7 @@
 #include "syncReplication.h"
 #include "syncIndexMgr.h"
 #include "syncMessage.h"
+#include "syncRaftCfg.h"
 #include "syncRaftEntry.h"
 #include "syncRaftLog.h"
 #include "syncRaftStore.h"
@@ -113,9 +114,17 @@ int32_t syncNodeAppendEntriesPeers(SSyncNode* pSyncNode) {
   return ret;
 }
 
+int32_t syncNodeAppendEntriesPeersSnapshot(SSyncNode* pSyncNode) { return 0; }
+
 int32_t syncNodeReplicate(SSyncNode* pSyncNode) {
   // start replicate
-  int32_t ret = syncNodeAppendEntriesPeers(pSyncNode);
+  int32_t ret = 0;
+
+  if (pSyncNode->pRaftCfg->snapshotEnable) {
+    ret = syncNodeAppendEntriesPeersSnapshot(pSyncNode);
+  } else {
+    ret = syncNodeAppendEntriesPeers(pSyncNode);
+  }
   return ret;
 }
 

source/libs/sync/src/syncRequestVote.c

@@ -77,3 +77,5 @@ int32_t syncNodeOnRequestVoteCb(SSyncNode* ths, SyncRequestVote* pMsg) {
 
   return ret;
 }
+
+int32_t syncNodeOnRequestVoteSnapshotCb(SSyncNode* ths, SyncRequestVote* pMsg) { return 0; }
\ No newline at end of file

source/libs/sync/src/syncRequestVoteReply.c

@@ -92,3 +92,5 @@ int32_t syncNodeOnRequestVoteReplyCb(SSyncNode* ths, SyncRequestVoteReply* pMsg)
 
   return ret;
 }
+
+int32_t syncNodeOnRequestVoteReplySnapshotCb(SSyncNode* ths, SyncRequestVoteReply* pMsg) { return 0; }
\ No newline at end of file