xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

refactor(sync): add timer routines

This commit is contained in:
Minghao Li 2022-07-12 15:04:32 +08:00
parent 37ebcdeaaa
commit 96f9274fef
5 changed files with 21 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
source/dnode/mnode/impl/src/mndSync.c
View File

@ -199,6 +199,7 @@ int32_t mndInitSync(SMnode *pMnode) {
} }
// decrease election timer // decrease election timer
setPingTimerMS(pMgmt->sync, 5000);
setElectTimerMS(pMgmt->sync, 600); setElectTimerMS(pMgmt->sync, 600);
setHeartbeatTimerMS(pMgmt->sync, 300); setHeartbeatTimerMS(pMgmt->sync, 300);

2
source/dnode/vnode/src/vnd/vnodeSync.c
View File

@ -569,7 +569,7 @@ int32_t vnodeSyncOpen(SVnode *pVnode, char *path) {
return -1; return -1;
} }
setPingTimerMS(pVnode->sync, 3000); setPingTimerMS(pVnode->sync, 5000);
setElectTimerMS(pVnode->sync, 500); setElectTimerMS(pVnode->sync, 500);
setHeartbeatTimerMS(pVnode->sync, 100); setHeartbeatTimerMS(pVnode->sync, 100);
return 0; return 0;

2
source/libs/sync/inc/syncEnv.h
View File

@ -30,7 +30,7 @@ extern "C" {
#define TIMER_MAX_MS 0x7FFFFFFF #define TIMER_MAX_MS 0x7FFFFFFF
#define ENV_TICK_TIMER_MS 1000 #define ENV_TICK_TIMER_MS 1000
#define PING_TIMER_MS 1000 #define PING_TIMER_MS 5000
#define ELECT_TIMER_MS_MIN 1300 #define ELECT_TIMER_MS_MIN 1300
#define ELECT_TIMER_MS_MAX (ELECT_TIMER_MS_MIN * 2) #define ELECT_TIMER_MS_MAX (ELECT_TIMER_MS_MIN * 2)
#define ELECT_TIMER_MS_RANGE (ELECT_TIMER_MS_MAX - ELECT_TIMER_MS_MIN) #define ELECT_TIMER_MS_RANGE (ELECT_TIMER_MS_MAX - ELECT_TIMER_MS_MIN)

30
source/libs/sync/src/syncMain.c
View File

@ -1099,19 +1099,13 @@ void syncNodeStart(SSyncNode* pSyncNode) {
// Raft 3.6.2 Committing entries from previous terms // Raft 3.6.2 Committing entries from previous terms
syncNodeAppendNoop(pSyncNode); syncNodeAppendNoop(pSyncNode);
syncMaybeAdvanceCommitIndex(pSyncNode); syncMaybeAdvanceCommitIndex(pSyncNode);
} else {
return;
}
syncNodeBecomeFollower(pSyncNode, "first start"); syncNodeBecomeFollower(pSyncNode, "first start");
// int32_t ret = 0;
// ret = syncNodeStartPingTimer(pSyncNode);
// ASSERT(ret == 0);
if (gRaftDetailLog) {
syncNodeLog2("==state change become leader immediately==", pSyncNode);
} }
int32_t ret = 0;
ret = syncNodeStartPingTimer(pSyncNode);
ASSERT(ret == 0);
} }
void syncNodeStartStandBy(SSyncNode* pSyncNode) { void syncNodeStartStandBy(SSyncNode* pSyncNode) {
@ -1162,14 +1156,6 @@ void syncNodeClose(SSyncNode* pSyncNode) {
pSyncNode->pNewNodeReceiver = NULL; pSyncNode->pNewNodeReceiver = NULL;
} }
/*
if (pSyncNode->pSnapshot != NULL) {
taosMemoryFree(pSyncNode->pSnapshot);
}
*/
// tsem_destroy(&pSyncNode->restoreSem);
// free memory in syncFreeNode // free memory in syncFreeNode
// taosMemoryFree(pSyncNode); // taosMemoryFree(pSyncNode);
} }
@ -1234,7 +1220,7 @@ int32_t syncNodeStartPingTimer(SSyncNode* pSyncNode) {
&pSyncNode->pPingTimer); &pSyncNode->pPingTimer);
atomic_store_64(&pSyncNode->pingTimerLogicClock, pSyncNode->pingTimerLogicClockUser); atomic_store_64(&pSyncNode->pingTimerLogicClock, pSyncNode->pingTimerLogicClockUser);
} else { } else {
sError("sync env is stop, syncNodeStartPingTimer"); sError("vgId:%d, start ping timer error, sync env is stop", pSyncNode->vgId);
} }
return ret; return ret;
} }
@ -1255,7 +1241,7 @@ int32_t syncNodeStartElectTimer(SSyncNode* pSyncNode, int32_t ms) {
&pSyncNode->pElectTimer); &pSyncNode->pElectTimer);
atomic_store_64(&pSyncNode->electTimerLogicClock, pSyncNode->electTimerLogicClockUser); atomic_store_64(&pSyncNode->electTimerLogicClock, pSyncNode->electTimerLogicClockUser);
} else { } else {
sError("sync env is stop, syncNodeStartElectTimer"); sError("vgId:%d, start elect timer error, sync env is stop", pSyncNode->vgId);
} }
return ret; return ret;
} }
@ -1295,7 +1281,7 @@ int32_t syncNodeStartHeartbeatTimer(SSyncNode* pSyncNode) {
&pSyncNode->pHeartbeatTimer); &pSyncNode->pHeartbeatTimer);
atomic_store_64(&pSyncNode->heartbeatTimerLogicClock, pSyncNode->heartbeatTimerLogicClockUser); atomic_store_64(&pSyncNode->heartbeatTimerLogicClock, pSyncNode->heartbeatTimerLogicClockUser);
} else { } else {
sError("sync env is stop, syncNodeStartHeartbeatTimer"); sError("vgId:%d, start heartbeat timer error, sync env is stop", pSyncNode->vgId);
} }
return ret; return ret;
} }

12
source/libs/sync/src/syncTimeout.c
View File

@ -17,6 +17,11 @@
#include "syncElection.h" #include "syncElection.h"
#include "syncReplication.h" #include "syncReplication.h"
int32_t syncNodeTimerRoutine(SSyncNode* ths) {
syncNodeEventLog(ths, "timer routines ... ");
return 0;
}
int32_t syncNodeOnTimeoutCb(SSyncNode* ths, SyncTimeout* pMsg) { int32_t syncNodeOnTimeoutCb(SSyncNode* ths, SyncTimeout* pMsg) {
int32_t ret = 0; int32_t ret = 0;
syncTimeoutLog2("==syncNodeOnTimeoutCb==", pMsg); syncTimeoutLog2("==syncNodeOnTimeoutCb==", pMsg);
@ -24,8 +29,11 @@ int32_t syncNodeOnTimeoutCb(SSyncNode* ths, SyncTimeout* pMsg) {
if (pMsg->timeoutType == SYNC_TIMEOUT_PING) { if (pMsg->timeoutType == SYNC_TIMEOUT_PING) {
if (atomic_load_64(&ths->pingTimerLogicClockUser) <= pMsg->logicClock) { if (atomic_load_64(&ths->pingTimerLogicClockUser) <= pMsg->logicClock) {
++(ths->pingTimerCounter); ++(ths->pingTimerCounter);
// syncNodePingAll(ths); // syncNodePingAll(ths);
syncNodePingPeers(ths); // syncNodePingPeers(ths);
syncNodeTimerRoutine(ths);
} }
} else if (pMsg->timeoutType == SYNC_TIMEOUT_ELECTION) { } else if (pMsg->timeoutType == SYNC_TIMEOUT_ELECTION) {
@ -40,7 +48,7 @@ int32_t syncNodeOnTimeoutCb(SSyncNode* ths, SyncTimeout* pMsg) {
syncNodeReplicate(ths); syncNodeReplicate(ths);
} }
} else { } else {
sTrace("unknown timeoutType:%d", pMsg->timeoutType); sError("vgId:%d, unknown timeout-type:%d", ths->vgId, pMsg->timeoutType);
} }
return ret; return ret;