diff --git a/source/libs/sync/inc/syncInt.h b/source/libs/sync/inc/syncInt.h index 0e0042d526..0507bb3661 100644 --- a/source/libs/sync/inc/syncInt.h +++ b/source/libs/sync/inc/syncInt.h @@ -41,7 +41,7 @@ typedef struct SSyncRespMgr SSyncRespMgr; typedef struct SSyncSnapshotSender SSyncSnapshotSender; typedef struct SSyncSnapshotReceiver SSyncSnapshotReceiver; typedef struct SSyncTimer SSyncTimer; -typedef struct SSyncHbTimerData SSyncHbTimerData; +typedef struct SSyncHbTimerParam SSyncHbTimerParam; typedef struct SyncSnapshotSend SyncSnapshotSend; typedef struct SyncSnapshotRsp SyncSnapshotRsp; typedef struct SyncLocalCmd SyncLocalCmd; @@ -56,12 +56,13 @@ typedef struct SRaftId { SyncGroupId vgId; } SRaftId; -typedef struct SSyncHbTimerData { +typedef struct SSyncHbTimerParam { SSyncNode* pSyncNode; SSyncTimer* pTimer; SRaftId destId; uint64_t logicClock; -} SSyncHbTimerData; + int64_t executeTime; +} SSyncHbTimerParam; typedef struct SSyncTimer { void* pTimer; @@ -73,11 +74,12 @@ typedef struct SSyncTimer { void* pData; } SSyncTimer; -typedef struct SElectTimer { +typedef struct SElectTimerParam { uint64_t logicClock; SSyncNode* pSyncNode; + int64_t executeTime; void* pData; -} SElectTimer; +} SElectTimerParam; typedef struct SPeerState { SyncIndex lastSendIndex; @@ -153,6 +155,7 @@ typedef struct SSyncNode { uint64_t electTimerLogicClock; TAOS_TMR_CALLBACK FpElectTimerCB; // Timer Fp uint64_t electTimerCounter; + SElectTimerParam electTimerParam; // heartbeat timer tmr_h pHeartbeatTimer; @@ -161,6 +164,7 @@ typedef struct SSyncNode { uint64_t heartbeatTimerLogicClockUser; TAOS_TMR_CALLBACK FpHeartbeatTimerCB; // Timer Fp uint64_t heartbeatTimerCounter; + SSyncHbTimerParam hbTimerParam; // peer heartbeat timer SSyncTimer peerHeartbeatTimerArr[TSDB_MAX_REPLICA]; diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c index 036127bf9c..7484e93abe 100644 --- a/source/libs/sync/src/syncMain.c +++ b/source/libs/sync/src/syncMain.c @@ -665,7 +665,7 @@ static int32_t syncHbTimerInit(SSyncNode* pSyncNode, SSyncTimer* pSyncTimer, SRa static int32_t syncHbTimerStart(SSyncNode* pSyncNode, SSyncTimer* pSyncTimer) { int32_t ret = 0; if (syncIsInit()) { - SSyncHbTimerData* pData = taosMemoryMalloc(sizeof(SSyncHbTimerData)); + SSyncHbTimerParam* pData = taosMemoryMalloc(sizeof(SSyncHbTimerParam)); pData->pSyncNode = pSyncNode; pData->pTimer = pSyncTimer; pData->destId = pSyncTimer->destId; @@ -1106,12 +1106,13 @@ int32_t syncNodeStartElectTimer(SSyncNode* pSyncNode, int32_t ms) { if (syncIsInit()) { pSyncNode->electTimerMS = ms; - SElectTimer* pElectTimer = taosMemoryMalloc(sizeof(SElectTimer)); - pElectTimer->logicClock = pSyncNode->electTimerLogicClock; - pElectTimer->pSyncNode = pSyncNode; - pElectTimer->pData = NULL; + int64_t execTime = taosGetTimestampMs() + ms; + atomic_store_64(&(pSyncNode->electTimerParam.executeTime), execTime); + atomic_store_64(&(pSyncNode->electTimerParam.logicClock), pSyncNode->electTimerLogicClock); + pSyncNode->electTimerParam.pSyncNode = pSyncNode; + pSyncNode->electTimerParam.pData = NULL; - taosTmrReset(pSyncNode->FpElectTimerCB, pSyncNode->electTimerMS, pElectTimer, syncEnv()->pTimerManager, + taosTmrReset(pSyncNode->FpElectTimerCB, pSyncNode->electTimerMS, pSyncNode, syncEnv()->pTimerManager, &pSyncNode->pElectTimer); } else { @@ -1865,18 +1866,21 @@ static void syncNodeEqPingTimer(void* param, void* tmrId) { static void syncNodeEqElectTimer(void* param, void* tmrId) { if (!syncIsInit()) return; - SElectTimer* pElectTimer = param; - SSyncNode* pNode = pElectTimer->pSyncNode; + SSyncNode* pNode = (SSyncNode*)param; if (pNode == NULL) return; if (pNode->syncEqMsg == NULL) return; + int64_t tsNow = taosGetTimestampMs(); + if (tsNow < pNode->electTimerParam.executeTime) return; + SRpcMsg rpcMsg = {0}; - int32_t code = syncBuildTimeout(&rpcMsg, SYNC_TIMEOUT_ELECTION, pElectTimer->logicClock, pNode->electTimerMS, pNode); + int32_t code = + syncBuildTimeout(&rpcMsg, SYNC_TIMEOUT_ELECTION, pNode->electTimerParam.logicClock, pNode->electTimerMS, pNode); if (code != 0) { sError("failed to build elect msg"); - taosMemoryFree(pElectTimer); + return; } @@ -1887,21 +1891,9 @@ static void syncNodeEqElectTimer(void* param, void* tmrId) { if (code != 0) { sError("failed to sync enqueue elect msg since %s", terrstr()); rpcFreeCont(rpcMsg.pCont); - taosMemoryFree(pElectTimer); + return; } - - taosMemoryFree(pElectTimer); - -#if 0 - // reset timer ms - if (syncIsInit() && pNode->electBaseLine > 0) { - pNode->electTimerMS = syncUtilElectRandomMS(pNode->electBaseLine, 2 * pNode->electBaseLine); - taosTmrReset(syncNodeEqElectTimer, pNode->electTimerMS, pNode, syncEnv()->pTimerManager, &pNode->pElectTimer); - } else { - sError("sync env is stop, syncNodeEqElectTimer"); - } -#endif } static void syncNodeEqHeartbeatTimer(void* param, void* tmrId) { @@ -1938,9 +1930,9 @@ static void syncNodeEqHeartbeatTimer(void* param, void* tmrId) { } static void syncNodeEqPeerHeartbeatTimer(void* param, void* tmrId) { - SSyncHbTimerData* pData = (SSyncHbTimerData*)param; - SSyncNode* pSyncNode = pData->pSyncNode; - SSyncTimer* pSyncTimer = pData->pTimer; + SSyncHbTimerParam* pData = (SSyncHbTimerParam*)param; + SSyncNode* pSyncNode = pData->pSyncNode; + SSyncTimer* pSyncTimer = pData->pTimer; if (pSyncNode == NULL) { return; diff --git a/source/libs/sync/src/syncUtil.c b/source/libs/sync/src/syncUtil.c index fb1b07b0b6..1d0ad07c99 100644 --- a/source/libs/sync/src/syncUtil.c +++ b/source/libs/sync/src/syncUtil.c @@ -193,7 +193,7 @@ static void syncPeerState2Str(SSyncNode* pSyncNode, char* buf, int32_t bufLen) { } void syncPrintNodeLog(const char* flags, ELogLevel level, int32_t dflag, SSyncNode* pNode, const char* format, ...) { - if (pNode == NULL || pNode->pRaftCfg != NULL && pNode->pRaftStore == NULL || pNode->pLogStore == NULL) return; + if (pNode == NULL || pNode->pRaftCfg == NULL && pNode->pRaftStore == NULL || pNode->pLogStore == NULL) return; int64_t currentTerm = pNode->pRaftStore->currentTerm; // save error code, otherwise it will be overwritten @@ -554,4 +554,4 @@ void syncLogSendRequestVoteReply(SSyncNode* pSyncNode, const SyncRequestVoteRepl syncUtilU642Addr(pMsg->destId.addr, host, sizeof(host), &port); sNTrace(pSyncNode, "send sync-request-vote-reply to %s:%d {term:%" PRId64 ", grant:%d}, %s", host, port, pMsg->term, pMsg->voteGranted, s); -} \ No newline at end of file +}