refactor(sync): SSyncSnapshotSender

source/libs/sync/inc/syncSnapshot.h

@@ -39,8 +39,8 @@ typedef struct SSyncSnapshotSender {
   bool       start;
   int32_t    seq;
   int32_t    ack;
-  void *     pReader;
-  void *     pCurrentBlock;
+  void      *pReader;
+  void      *pCurrentBlock;
   int32_t    blockLen;
   SSnapshot  snapshot;
   int64_t    sendingMS;
@@ -48,25 +48,24 @@ typedef struct SSyncSnapshotSender {
   int32_t    replicaIndex;
   SyncTerm   term;
   SyncTerm   privateTerm;
-  bool       finish;
+  bool       apply;
 } SSyncSnapshotSender;
 
-void snapshotSenderDoStart(SSyncSnapshotSender *pSender);
-
 SSyncSnapshotSender *snapshotSenderCreate(SSyncNode *pSyncNode, int32_t replicaIndex);
 void                 snapshotSenderDestroy(SSyncSnapshotSender *pSender);
-// void                 snapshotSenderStart(SSyncSnapshotSender *pSender);
-void    snapshotSenderStop(SSyncSnapshotSender *pSender);
-int32_t snapshotSend(SSyncSnapshotSender *pSender);
-int32_t snapshotReSend(SSyncSnapshotSender *pSender);
-cJSON * snapshotSender2Json(SSyncSnapshotSender *pSender);
-char *  snapshotSender2Str(SSyncSnapshotSender *pSender);
+bool                 snapshotSenderIsStart(SSyncSnapshotSender *pSender);
+void                 snapshotSenderStart(SSyncSnapshotSender *pSender);
+void                 snapshotSenderStop(SSyncSnapshotSender *pSender);
+int32_t              snapshotSend(SSyncSnapshotSender *pSender);
+int32_t              snapshotReSend(SSyncSnapshotSender *pSender);
+cJSON               *snapshotSender2Json(SSyncSnapshotSender *pSender);
+char                *snapshotSender2Str(SSyncSnapshotSender *pSender);
 
 typedef struct SSyncSnapshotReceiver {
   bool start;
 
   int32_t  ack;
-  void *   pWriter;
+  void    *pWriter;
   SyncTerm term;
   SyncTerm privateTerm;
 
@@ -78,8 +77,8 @@ SSyncSnapshotReceiver *snapshotReceiverCreate(SSyncNode *pSyncNode, int32_t repl
 void                   snapshotReceiverDestroy(SSyncSnapshotReceiver *pReceiver);
 void                   snapshotReceiverStart(SSyncSnapshotReceiver *pReceiver);
 void                   snapshotReceiverStop(SSyncSnapshotReceiver *pReceiver);
-cJSON *                snapshotReceiver2Json(SSyncSnapshotReceiver *pReceiver);
-char *                 snapshotReceiver2Str(SSyncSnapshotReceiver *pReceiver);
+cJSON                 *snapshotReceiver2Json(SSyncSnapshotReceiver *pReceiver);
+char                  *snapshotReceiver2Str(SSyncSnapshotReceiver *pReceiver);
 
 int32_t syncNodeOnSnapshotSendCb(SSyncNode *ths, SyncSnapshotSend *pMsg);
 int32_t syncNodeOnSnapshotRspCb(SSyncNode *ths, SyncSnapshotRsp *pMsg);

source/libs/sync/src/syncAppendEntriesReply.c

@@ -190,7 +190,7 @@ int32_t syncNodeOnAppendEntriesReplySnapshotCb(SSyncNode* ths, SyncAppendEntries
 
           } else {
             // start first time
-            snapshotSenderDoStart(pSender);
+            snapshotSenderStart(pSender);
             pSender->start = true;
             sentryIndex = pSender->snapshot.lastApplyIndex;
           }

source/libs/sync/src/syncSnapshot.c

@@ -20,7 +20,6 @@
 #include "syncUtil.h"
 #include "wal.h"
 
-// static void snapshotSenderDoStart(SSyncSnapshotSender *pSender);
 static void snapshotReceiverDoStart(SSyncSnapshotReceiver *pReceiver);
 
 SSyncSnapshotSender *snapshotSenderCreate(SSyncNode *pSyncNode, int32_t replicaIndex) {
@@ -45,27 +44,28 @@ SSyncSnapshotSender *snapshotSenderCreate(SSyncNode *pSyncNode, int32_t replicaI
     pSender->term = pSyncNode->pRaftStore->currentTerm;
     pSender->privateTerm = taosGetTimestampMs() + 100;
     pSender->pSyncNode->pFsm->FpGetSnapshot(pSender->pSyncNode->pFsm, &(pSender->snapshot));
-
-    pSender->finish = false;
+    pSender->apply = false;
   } else {
-    sInfo("snapshotSenderCreate cannot create sender");
+    sError("snapshotSenderCreate cannot create sender");
   }
-
   return pSender;
 }
 
 void snapshotSenderDestroy(SSyncSnapshotSender *pSender) {
   if (pSender != NULL) {
+    if (pSender->pCurrentBlock != NULL) {
+      taosMemoryFree(pSender->pCurrentBlock);
+    }
     taosMemoryFree(pSender);
   }
 }
 
+bool snapshotSenderIsStart(SSyncSnapshotSender *pSender) { return pSender->start; }
+
 // begin send snapshot (current term, seq begin)
-void snapshotSenderDoStart(SSyncSnapshotSender *pSender) {
-  // when start, increase term
-  ++(pSender->privateTerm);
+void snapshotSenderStart(SSyncSnapshotSender *pSender) {
+  ASSERT(!snapshotSenderIsStart(pSender));
 
-  pSender->term = pSender->pSyncNode->pRaftStore->currentTerm;
   pSender->seq = SYNC_SNAPSHOT_SEQ_BEGIN;
   pSender->ack = SYNC_SNAPSHOT_SEQ_INVALID;
 
@@ -74,9 +74,21 @@ void snapshotSenderDoStart(SSyncSnapshotSender *pSender) {
   int32_t ret = pSender->pSyncNode->pFsm->FpSnapshotStartRead(pSender->pSyncNode->pFsm, &(pSender->pReader));
   ASSERT(ret == 0);
 
+  if (pSender->pCurrentBlock != NULL) {
+    taosMemoryFree(pSender->pCurrentBlock);
+  }
+
+  pSender->blockLen = 0;
+
   // get current snapshot info
   pSender->pSyncNode->pFsm->FpGetSnapshot(pSender->pSyncNode->pFsm, &(pSender->snapshot));
 
+  pSender->sendingMS = SYNC_SNAPSHOT_RETRY_MS;
+  pSender->term = pSender->pSyncNode->pRaftStore->currentTerm;
+  ++(pSender->privateTerm);
+  pSender->apply = false;
+  pSender->start = true;
+
   // build begin msg
   SyncSnapshotSend *pMsg = syncSnapshotSendBuild(0, pSender->pSyncNode->vgId);
   pMsg->srcId = pSender->pSyncNode->myRaftId;
@@ -87,7 +99,7 @@ void snapshotSenderDoStart(SSyncSnapshotSender *pSender) {
   pMsg->seq = pSender->seq;  // SYNC_SNAPSHOT_SEQ_BEGIN
   pMsg->privateTerm = pSender->privateTerm;
 
-  // send
+  // send msg
   SRpcMsg rpcMsg;
   syncSnapshotSend2RpcMsg(pMsg, &rpcMsg);
   syncNodeSendMsgById(&(pMsg->destId), pSender->pSyncNode, &rpcMsg);
@@ -192,6 +204,7 @@ int32_t snapshotSend(SSyncSnapshotSender *pSender) {
     pSender->seq = SYNC_SNAPSHOT_SEQ_END;
   }
 
+  // build msg
   SyncSnapshotSend *pMsg = syncSnapshotSendBuild(pSender->blockLen, pSender->pSyncNode->vgId);
   pMsg->srcId = pSender->pSyncNode->myRaftId;
   pMsg->destId = (pSender->pSyncNode->replicasId)[pSender->replicaIndex];
@@ -202,6 +215,7 @@ int32_t snapshotSend(SSyncSnapshotSender *pSender) {
   pMsg->privateTerm = pSender->privateTerm;
   memcpy(pMsg->data, pSender->pCurrentBlock, pSender->blockLen);
 
+  // send msg
   SRpcMsg rpcMsg;
   syncSnapshotSend2RpcMsg(pMsg, &rpcMsg);
   syncNodeSendMsgById(&(pMsg->destId), pSender->pSyncNode, &rpcMsg);
@@ -215,7 +229,6 @@ int32_t snapshotSend(SSyncSnapshotSender *pSender) {
   taosMemoryFree(msgStr);
 
   syncSnapshotSendDestroy(pMsg);
-
   return 0;
 }
 
@@ -286,7 +299,7 @@ cJSON *snapshotSender2Json(SSyncSnapshotSender *pSender) {
     cJSON_AddStringToObject(pRoot, "term", u64buf);
     snprintf(u64buf, sizeof(u64buf), "%lu", pSender->privateTerm);
     cJSON_AddStringToObject(pRoot, "privateTerm", u64buf);
-    cJSON_AddNumberToObject(pRoot, "finish", pSender->finish);
+    cJSON_AddNumberToObject(pRoot, "apply", pSender->apply);
   }
 
   cJSON *pJson = cJSON_CreateObject();
@@ -296,7 +309,7 @@ cJSON *snapshotSender2Json(SSyncSnapshotSender *pSender) {
 
 char *snapshotSender2Str(SSyncSnapshotSender *pSender) {
   cJSON *pJson = snapshotSender2Json(pSender);
-  char * serialized = cJSON_Print(pJson);
+  char  *serialized = cJSON_Print(pJson);
   cJSON_Delete(pJson);
   return serialized;
 }
@@ -416,7 +429,7 @@ cJSON *snapshotReceiver2Json(SSyncSnapshotReceiver *pReceiver) {
 
 char *snapshotReceiver2Str(SSyncSnapshotReceiver *pReceiver) {
   cJSON *pJson = snapshotReceiver2Json(pReceiver);
-  char * serialized = cJSON_Print(pJson);
+  char  *serialized = cJSON_Print(pJson);
   cJSON_Delete(pJson);
   return serialized;
 }
@@ -448,7 +461,7 @@ int32_t syncNodeOnSnapshotSendCb(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) {
         pSyncNode->pFsm->FpSnapshotStopWrite(pSyncNode->pFsm, pReceiver->pWriter, true);
 
         pSyncNode->pLogStore->syncLogSetBeginIndex(pSyncNode->pLogStore, pMsg->lastIndex + 1);
-        char *    logSimpleStr = logStoreSimple2Str(pSyncNode->pLogStore);
+        char     *logSimpleStr = logStoreSimple2Str(pSyncNode->pLogStore);
         SSnapshot snapshot;
         pSyncNode->pFsm->FpGetSnapshot(pSyncNode->pFsm, &snapshot);
         sInfo(
@@ -539,7 +552,7 @@ int32_t syncNodeOnSnapshotRspCb(SSyncNode *pSyncNode, SyncSnapshotRsp *pMsg) {
     if (pMsg->term == pSyncNode->pRaftStore->currentTerm) {
       // receiver ack is finish, close sender
       if (pMsg->ack == SYNC_SNAPSHOT_SEQ_END) {
-        pSender->finish = true;
+        pSender->apply = true;
         snapshotSenderStop(pSender);
 
         // update nextIndex private term