Merge pull request #11886 from taosdata/feature/3.0_mhli

enh: add flag in syncEnv, to deal with expired timer

include/libs/sync/sync.h

@@ -158,6 +158,8 @@ typedef enum {
 
 int32_t syncPropose(int64_t rid, const SRpcMsg* pMsg, bool isWeak);
 
+bool syncEnvIsStart();
+
 extern int32_t sDebugFlag;
 
 //-----------------------------------------

source/dnode/vnode/src/vnd/vnodeSvr.c

@@ -193,84 +193,89 @@ void smaHandleRes(void *pVnode, int64_t smaId, const SArray *data) {
 
 // sync integration
 int vnodeProcessSyncReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
-  SSyncNode *pSyncNode = syncNodeAcquire(pVnode->sync);
-  assert(pSyncNode != NULL);
 
-  ESyncState state = syncGetMyRole(pVnode->sync);
-  SyncTerm   currentTerm = syncGetMyTerm(pVnode->sync);
+  if (syncEnvIsStart()) {
 
-  SMsgHead *pHead = pMsg->pCont;
+    SSyncNode *pSyncNode = syncNodeAcquire(pVnode->sync);
+    assert(pSyncNode != NULL);
 
-  char  logBuf[512];
-  char *syncNodeStr = sync2SimpleStr(pVnode->sync);
-  snprintf(logBuf, sizeof(logBuf), "==vnodeProcessSyncReq== msgType:%d, syncNode: %s", pMsg->msgType, syncNodeStr);
-  syncRpcMsgLog2(logBuf, pMsg);
-  taosMemoryFree(syncNodeStr);
+    ESyncState state = syncGetMyRole(pVnode->sync);
+    SyncTerm   currentTerm = syncGetMyTerm(pVnode->sync);
 
-  SRpcMsg *pRpcMsg = pMsg;
+    SMsgHead *pHead = pMsg->pCont;
 
-  if (pRpcMsg->msgType == TDMT_VND_SYNC_TIMEOUT) {
-    SyncTimeout *pSyncMsg = syncTimeoutFromRpcMsg2(pRpcMsg);
-    assert(pSyncMsg != NULL);
+    char  logBuf[512];
+    char *syncNodeStr = sync2SimpleStr(pVnode->sync);
+    snprintf(logBuf, sizeof(logBuf), "==vnodeProcessSyncReq== msgType:%d, syncNode: %s", pMsg->msgType, syncNodeStr);
+    syncRpcMsgLog2(logBuf, pMsg);
+    taosMemoryFree(syncNodeStr);
 
-    syncNodeOnTimeoutCb(pSyncNode, pSyncMsg);
-    syncTimeoutDestroy(pSyncMsg);
+    SRpcMsg *pRpcMsg = pMsg;
 
-  } else if (pRpcMsg->msgType == TDMT_VND_SYNC_PING) {
-    SyncPing *pSyncMsg = syncPingFromRpcMsg2(pRpcMsg);
-    assert(pSyncMsg != NULL);
+    if (pRpcMsg->msgType == TDMT_VND_SYNC_TIMEOUT) {
+      SyncTimeout *pSyncMsg = syncTimeoutFromRpcMsg2(pRpcMsg);
+      assert(pSyncMsg != NULL);
 
-    syncNodeOnPingCb(pSyncNode, pSyncMsg);
-    syncPingDestroy(pSyncMsg);
+      syncNodeOnTimeoutCb(pSyncNode, pSyncMsg);
+      syncTimeoutDestroy(pSyncMsg);
 
-  } else if (pRpcMsg->msgType == TDMT_VND_SYNC_PING_REPLY) {
-    SyncPingReply *pSyncMsg = syncPingReplyFromRpcMsg2(pRpcMsg);
-    assert(pSyncMsg != NULL);
+    } else if (pRpcMsg->msgType == TDMT_VND_SYNC_PING) {
+      SyncPing *pSyncMsg = syncPingFromRpcMsg2(pRpcMsg);
+      assert(pSyncMsg != NULL);
 
-    syncNodeOnPingReplyCb(pSyncNode, pSyncMsg);
-    syncPingReplyDestroy(pSyncMsg);
+      syncNodeOnPingCb(pSyncNode, pSyncMsg);
+      syncPingDestroy(pSyncMsg);
 
-  } else if (pRpcMsg->msgType == TDMT_VND_SYNC_CLIENT_REQUEST) {
-    SyncClientRequest *pSyncMsg = syncClientRequestFromRpcMsg2(pRpcMsg);
-    assert(pSyncMsg != NULL);
+    } else if (pRpcMsg->msgType == TDMT_VND_SYNC_PING_REPLY) {
+      SyncPingReply *pSyncMsg = syncPingReplyFromRpcMsg2(pRpcMsg);
+      assert(pSyncMsg != NULL);
 
-    syncNodeOnClientRequestCb(pSyncNode, pSyncMsg);
-    syncClientRequestDestroy(pSyncMsg);
+      syncNodeOnPingReplyCb(pSyncNode, pSyncMsg);
+      syncPingReplyDestroy(pSyncMsg);
 
-  } else if (pRpcMsg->msgType == TDMT_VND_SYNC_REQUEST_VOTE) {
-    SyncRequestVote *pSyncMsg = syncRequestVoteFromRpcMsg2(pRpcMsg);
-    assert(pSyncMsg != NULL);
+    } else if (pRpcMsg->msgType == TDMT_VND_SYNC_CLIENT_REQUEST) {
+      SyncClientRequest *pSyncMsg = syncClientRequestFromRpcMsg2(pRpcMsg);
+      assert(pSyncMsg != NULL);
 
-    syncNodeOnRequestVoteCb(pSyncNode, pSyncMsg);
-    syncRequestVoteDestroy(pSyncMsg);
+      syncNodeOnClientRequestCb(pSyncNode, pSyncMsg);
+      syncClientRequestDestroy(pSyncMsg);
 
-  } else if (pRpcMsg->msgType == TDMT_VND_SYNC_REQUEST_VOTE_REPLY) {
-    SyncRequestVoteReply *pSyncMsg = syncRequestVoteReplyFromRpcMsg2(pRpcMsg);
-    assert(pSyncMsg != NULL);
+    } else if (pRpcMsg->msgType == TDMT_VND_SYNC_REQUEST_VOTE) {
+      SyncRequestVote *pSyncMsg = syncRequestVoteFromRpcMsg2(pRpcMsg);
+      assert(pSyncMsg != NULL);
 
-    syncNodeOnRequestVoteReplyCb(pSyncNode, pSyncMsg);
-    syncRequestVoteReplyDestroy(pSyncMsg);
+      syncNodeOnRequestVoteCb(pSyncNode, pSyncMsg);
+      syncRequestVoteDestroy(pSyncMsg);
 
-  } else if (pRpcMsg->msgType == TDMT_VND_SYNC_APPEND_ENTRIES) {
-    SyncAppendEntries *pSyncMsg = syncAppendEntriesFromRpcMsg2(pRpcMsg);
-    assert(pSyncMsg != NULL);
+    } else if (pRpcMsg->msgType == TDMT_VND_SYNC_REQUEST_VOTE_REPLY) {
+      SyncRequestVoteReply *pSyncMsg = syncRequestVoteReplyFromRpcMsg2(pRpcMsg);
+      assert(pSyncMsg != NULL);
 
-    syncNodeOnAppendEntriesCb(pSyncNode, pSyncMsg);
-    syncAppendEntriesDestroy(pSyncMsg);
+      syncNodeOnRequestVoteReplyCb(pSyncNode, pSyncMsg);
+      syncRequestVoteReplyDestroy(pSyncMsg);
 
-  } else if (pRpcMsg->msgType == TDMT_VND_SYNC_APPEND_ENTRIES_REPLY) {
-    SyncAppendEntriesReply *pSyncMsg = syncAppendEntriesReplyFromRpcMsg2(pRpcMsg);
-    assert(pSyncMsg != NULL);
+    } else if (pRpcMsg->msgType == TDMT_VND_SYNC_APPEND_ENTRIES) {
+      SyncAppendEntries *pSyncMsg = syncAppendEntriesFromRpcMsg2(pRpcMsg);
+      assert(pSyncMsg != NULL);
 
-    syncNodeOnAppendEntriesReplyCb(pSyncNode, pSyncMsg);
-    syncAppendEntriesReplyDestroy(pSyncMsg);
+      syncNodeOnAppendEntriesCb(pSyncNode, pSyncMsg);
+      syncAppendEntriesDestroy(pSyncMsg);
 
-  } else {
-    vError("==vnodeProcessSyncReq== error msg type:%d", pRpcMsg->msgType);
-  }
+    } else if (pRpcMsg->msgType == TDMT_VND_SYNC_APPEND_ENTRIES_REPLY) {
+      SyncAppendEntriesReply *pSyncMsg = syncAppendEntriesReplyFromRpcMsg2(pRpcMsg);
+      assert(pSyncMsg != NULL);
 
-  syncNodeRelease(pSyncNode);
+      syncNodeOnAppendEntriesReplyCb(pSyncNode, pSyncMsg);
+      syncAppendEntriesReplyDestroy(pSyncMsg);
 
+    } else {
+      vError("==vnodeProcessSyncReq== error msg type:%d", pRpcMsg->msgType);
+    }
+
+    syncNodeRelease(pSyncNode);
+  } else {
+    vError("==vnodeProcessSyncReq== error syncEnv stop");
+  }
   return 0;
 }
 

source/libs/sync/inc/syncEnv.h

@@ -39,6 +39,8 @@ extern "C" {
 #define EMPTY_RAFT_ID ((SRaftId){.addr = 0, .vgId = 0})
 
 typedef struct SSyncEnv {
+  uint8_t isStart;
+
   // tick timer
   tmr_h             pEnvTickTimer;
   int32_t           envTickTimerMS;

source/libs/sync/src/syncEnv.c

@@ -26,6 +26,14 @@ static int32_t   doSyncEnvStopTimer(SSyncEnv *pSyncEnv);
 static void      syncEnvTick(void *param, void *tmrId);
 // --------------------------------
 
+bool syncEnvIsStart() {
+  if (gSyncEnv == NULL) {
+    return false;
+  }
+
+  return atomic_load_8(&(gSyncEnv->isStart));
+}
+
 int32_t syncEnvStart() {
   int32_t ret = 0;
   taosSeedRand(taosGetTimestampSec());
@@ -88,12 +96,15 @@ static SSyncEnv *doSyncEnvStart() {
 
   // start tmr thread
   pSyncEnv->pTimerManager = taosTmrInit(1000, 50, 10000, "SYNC-ENV");
+
+  atomic_store_8(&(pSyncEnv->isStart), 1); 
   return pSyncEnv;
 }
 
 static int32_t doSyncEnvStop(SSyncEnv *pSyncEnv) {
   assert(pSyncEnv == gSyncEnv);
   if (pSyncEnv != NULL) {
+    atomic_store_8(&(pSyncEnv->isStart), 0); 
     taosTmrCleanUp(pSyncEnv->pTimerManager);
     taosMemoryFree(pSyncEnv);
   }