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

refactor(sync): optimized heartbeat timer

source/libs/sync/inc/syncEnv.h

@@ -29,6 +29,7 @@ extern "C" {
 #define ELECT_TIMER_MS_MAX   (ELECT_TIMER_MS_MIN * 2)
 #define ELECT_TIMER_MS_RANGE (ELECT_TIMER_MS_MAX - ELECT_TIMER_MS_MIN)
 #define HEARTBEAT_TIMER_MS   1000
+#define HEARTBEAT_TICK_NUM   20
 
 typedef struct SSyncEnv {
   uint8_t isStart;

source/libs/sync/inc/syncInt.h

@@ -61,6 +61,7 @@ typedef struct SSyncHbTimerData {
   SSyncTimer* pTimer;
   SRaftId     destId;
   uint64_t    logicClock;
+  int64_t     execTime;
   int64_t     rid;
 } SSyncHbTimerData;
 

source/libs/sync/inc/syncUtil.h

@@ -94,7 +94,7 @@ void syncLogRecvLocalCmd(SSyncNode* pSyncNode, const SyncLocalCmd* pMsg, const c
 void syncLogSendAppendEntriesReply(SSyncNode* pSyncNode, const SyncAppendEntriesReply* pMsg, const char* s);
 void syncLogRecvAppendEntriesReply(SSyncNode* pSyncNode, const SyncAppendEntriesReply* pMsg, const char* s);
 
-void syncLogSendHeartbeat(SSyncNode* pSyncNode, const SyncHeartbeat* pMsg, bool printX, int64_t timerElapsed);
+void syncLogSendHeartbeat(SSyncNode* pSyncNode, const SyncHeartbeat* pMsg, bool printX, int64_t timerElapsed, int64_t execTime);
 void syncLogRecvHeartbeat(SSyncNode* pSyncNode, const SyncHeartbeat* pMsg, int64_t timeDiff);
 
 void syncLogSendHeartbeatReply(SSyncNode* pSyncNode, const SyncHeartbeatReply* pMsg, const char* s);

source/libs/sync/src/syncMain.c

@@ -698,6 +698,7 @@ static int32_t syncHbTimerInit(SSyncNode* pSyncNode, SSyncTimer* pSyncTimer, SRa
 
 static int32_t syncHbTimerStart(SSyncNode* pSyncNode, SSyncTimer* pSyncTimer) {
   int32_t ret = 0;
+  int64_t tsNow = taosGetTimestampMs();
   if (syncIsInit()) {
     SSyncHbTimerData* pData = syncHbTimerDataAcquire(pSyncTimer->hbDataRid);
     if (pData == NULL) {
@@ -705,15 +706,16 @@ static int32_t syncHbTimerStart(SSyncNode* pSyncNode, SSyncTimer* pSyncTimer) {
       pData->rid = syncHbTimerDataAdd(pData);
     }
     pSyncTimer->hbDataRid = pData->rid;
-    pSyncTimer->timeStamp = taosGetTimestampMs();
+    pSyncTimer->timeStamp = tsNow;
 
     pData->syncNodeRid = pSyncNode->rid;
     pData->pTimer = pSyncTimer;
     pData->destId = pSyncTimer->destId;
     pData->logicClock = pSyncTimer->logicClock;
+    pData->execTime = tsNow + pSyncTimer->timerMS;
 
-    taosTmrReset(pSyncTimer->timerCb, pSyncTimer->timerMS, (void*)(pData->rid), syncEnv()->pTimerManager,
-                 &pSyncTimer->pTimer);
+    taosTmrReset(pSyncTimer->timerCb, pSyncTimer->timerMS / HEARTBEAT_TICK_NUM, (void*)(pData->rid),
+                 syncEnv()->pTimerManager, &pSyncTimer->pTimer);
   } else {
     sError("vgId:%d, start ctrl hb timer error, sync env is stop", pSyncNode->vgId);
   }
@@ -1979,6 +1981,7 @@ static void syncNodeEqHeartbeatTimer(void* param, void* tmrId) {
 
 static void syncNodeEqPeerHeartbeatTimer(void* param, void* tmrId) {
   int64_t hbDataRid = (int64_t)param;
+  int64_t tsNow = taosGetTimestampMs();
 
   SSyncHbTimerData* pData = syncHbTimerDataAcquire(hbDataRid);
   if (pData == NULL) {
@@ -2023,35 +2026,51 @@ static void syncNodeEqPeerHeartbeatTimer(void* param, void* tmrId) {
     int64_t msgLogicClock = atomic_load_64(&pData->logicClock);
 
     if (timerLogicClock == msgLogicClock) {
+      if (tsNow > pData->execTime) {
+#if 0        
+        sTrace(
+            "vgId:%d, hbDataRid:%ld,  EXECUTE this step-------- heartbeat tsNow:%ld, exec:%ld, tsNow-exec:%ld, "
+            "---------",
+            pSyncNode->vgId, hbDataRid, tsNow, pData->execTime, tsNow - pData->execTime);
+#endif
+
+        pData->execTime += pSyncTimer->timerMS;
+
+        SRpcMsg rpcMsg = {0};
+        (void)syncBuildHeartbeat(&rpcMsg, pSyncNode->vgId);
+
+        SyncHeartbeat* pSyncMsg = rpcMsg.pCont;
+        pSyncMsg->srcId = pSyncNode->myRaftId;
+        pSyncMsg->destId = pData->destId;
+        pSyncMsg->term = pSyncNode->pRaftStore->currentTerm;
+        pSyncMsg->commitIndex = pSyncNode->commitIndex;
+        pSyncMsg->minMatchIndex = syncMinMatchIndex(pSyncNode);
+        pSyncMsg->privateTerm = 0;
+        pSyncMsg->timeStamp = tsNow;
+
+        // update reset time
+        int64_t timerElapsed = tsNow - pSyncTimer->timeStamp;
+        pSyncTimer->timeStamp = tsNow;
+
+        // send msg
+        syncLogSendHeartbeat(pSyncNode, pSyncMsg, false, timerElapsed, pData->execTime);
+        syncNodeSendHeartbeat(pSyncNode, &pSyncMsg->destId, &rpcMsg);
+      } else {
+#if 0        
+        sTrace(
+            "vgId:%d, hbDataRid:%ld,  pass this step-------- heartbeat tsNow:%ld, exec:%ld, tsNow-exec:%ld, ---------",
+            pSyncNode->vgId, hbDataRid, tsNow, pData->execTime, tsNow - pData->execTime);
+#endif
+      }
+
       if (syncIsInit()) {
         // sTrace("vgId:%d, reset peer hb timer", pSyncNode->vgId);
-        taosTmrReset(syncNodeEqPeerHeartbeatTimer, pSyncTimer->timerMS, (void*)hbDataRid, syncEnv()->pTimerManager,
-                     &pSyncTimer->pTimer);
+        taosTmrReset(syncNodeEqPeerHeartbeatTimer, pSyncTimer->timerMS / HEARTBEAT_TICK_NUM, (void*)hbDataRid,
+                     syncEnv()->pTimerManager, &pSyncTimer->pTimer);
       } else {
         sError("sync env is stop, reset peer hb timer error");
       }
 
-      SRpcMsg rpcMsg = {0};
-      (void)syncBuildHeartbeat(&rpcMsg, pSyncNode->vgId);
-
-      // update reset time
-      int64_t tsNow = taosGetTimestampMs();
-      int64_t timerElapsed = tsNow - pSyncTimer->timeStamp;
-      pSyncTimer->timeStamp = tsNow;
-
-      SyncHeartbeat* pSyncMsg = rpcMsg.pCont;
-      pSyncMsg->srcId = pSyncNode->myRaftId;
-      pSyncMsg->destId = pData->destId;
-      pSyncMsg->term = pSyncNode->pRaftStore->currentTerm;
-      pSyncMsg->commitIndex = pSyncNode->commitIndex;
-      pSyncMsg->minMatchIndex = syncMinMatchIndex(pSyncNode);
-      pSyncMsg->privateTerm = 0;
-      pSyncMsg->timeStamp = tsNow;
-
-      // send msg
-      syncLogSendHeartbeat(pSyncNode, pSyncMsg, false, timerElapsed);
-      syncNodeSendHeartbeat(pSyncNode, &pSyncMsg->destId, &rpcMsg);
-
     } else {
       sTrace("vgId:%d, do not send hb, timerLogicClock:%" PRId64 ", msgLogicClock:%" PRId64 "", pSyncNode->vgId,
              timerLogicClock, msgLogicClock);

source/libs/sync/src/syncReplication.c

@@ -230,7 +230,7 @@ int32_t syncNodeHeartbeatPeers(SSyncNode* pSyncNode) {
     pSyncMsg->timeStamp = ts;
 
     // send msg
-    syncLogSendHeartbeat(pSyncNode, pSyncMsg, true, 0);
+    syncLogSendHeartbeat(pSyncNode, pSyncMsg, true, 0, 0);
     syncNodeSendHeartbeat(pSyncNode, &pSyncMsg->destId, &rpcMsg);
   }
 

source/libs/sync/src/syncUtil.c

@@ -438,7 +438,7 @@ void syncLogRecvAppendEntriesReply(SSyncNode* pSyncNode, const SyncAppendEntries
           host, port, pMsg->term, pMsg->privateTerm, pMsg->success, pMsg->lastSendIndex, pMsg->matchIndex, s);
 }
 
-void syncLogSendHeartbeat(SSyncNode* pSyncNode, const SyncHeartbeat* pMsg, bool printX, int64_t timerElapsed) {
+void syncLogSendHeartbeat(SSyncNode* pSyncNode, const SyncHeartbeat* pMsg, bool printX, int64_t timerElapsed, int64_t execTime) {
   if (!(sDebugFlag & DEBUG_TRACE)) return;
 
   char     host[64];
@@ -453,8 +453,8 @@ void syncLogSendHeartbeat(SSyncNode* pSyncNode, const SyncHeartbeat* pMsg, bool
   } else {
     sNTrace(pSyncNode,
             "send sync-heartbeat to %s:%d {term:%" PRId64 ", cmt:%" PRId64 ", min-match:%" PRId64 ", ts:%" PRId64
-            "}, timer-elapsed:%" PRId64,
-            host, port, pMsg->term, pMsg->commitIndex, pMsg->minMatchIndex, pMsg->timeStamp, timerElapsed);
+            "}, timer-elapsed:%" PRId64 ", next-exec:%" PRId64,
+            host, port, pMsg->term, pMsg->commitIndex, pMsg->minMatchIndex, pMsg->timeStamp, timerElapsed, execTime);
   }
 }