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

Merge remote-tracking branch 'origin/3.0' into fix/dnode

This commit is contained in:
Shengliang Guan 2022-06-15 18:25:43 +08:00
parent 405b561821 1b15c3439a
commit bb032f13f5
1 changed files with 31 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1808,37 +1808,39 @@ static void syncNodeEqElectTimer(void* param, void* tmrId) {
static void syncNodeEqHeartbeatTimer(void* param, void* tmrId) { static void syncNodeEqHeartbeatTimer(void* param, void* tmrId) {
SSyncNode* pSyncNode = (SSyncNode*)param; SSyncNode* pSyncNode = (SSyncNode*)param;
if (atomic_load_64(&pSyncNode->heartbeatTimerLogicClockUser) <= if (pSyncNode->replicaNum > 1) {
atomic_load_64(&pSyncNode->heartbeatTimerLogicClock)) { if (atomic_load_64(&pSyncNode->heartbeatTimerLogicClockUser) <=
SyncTimeout* pSyncMsg = atomic_load_64(&pSyncNode->heartbeatTimerLogicClock)) {
syncTimeoutBuild2(SYNC_TIMEOUT_HEARTBEAT, atomic_load_64(&pSyncNode->heartbeatTimerLogicClock), SyncTimeout* pSyncMsg =
pSyncNode->heartbeatTimerMS, pSyncNode->vgId, pSyncNode); syncTimeoutBuild2(SYNC_TIMEOUT_HEARTBEAT, atomic_load_64(&pSyncNode->heartbeatTimerLogicClock),
SRpcMsg rpcMsg; pSyncNode->heartbeatTimerMS, pSyncNode->vgId, pSyncNode);
syncTimeout2RpcMsg(pSyncMsg, &rpcMsg); SRpcMsg rpcMsg;
syncRpcMsgLog2((char*)"==syncNodeEqHeartbeatTimer==", &rpcMsg); syncTimeout2RpcMsg(pSyncMsg, &rpcMsg);
if (pSyncNode->FpEqMsg != NULL) { syncRpcMsgLog2((char*)"==syncNodeEqHeartbeatTimer==", &rpcMsg);
int32_t code = pSyncNode->FpEqMsg(pSyncNode->msgcb, &rpcMsg); if (pSyncNode->FpEqMsg != NULL) {
if (code != 0) { int32_t code = pSyncNode->FpEqMsg(pSyncNode->msgcb, &rpcMsg);
sError("vgId:%d sync enqueue timer msg error, code:%d", pSyncNode->vgId, code); if (code != 0) {
rpcFreeCont(rpcMsg.pCont); sError("vgId:%d sync enqueue timer msg error, code:%d", pSyncNode->vgId, code);
syncTimeoutDestroy(pSyncMsg); rpcFreeCont(rpcMsg.pCont);
return; syncTimeoutDestroy(pSyncMsg);
return;
}
} else {
sError("syncNodeEqHeartbeatTimer FpEqMsg is NULL");
}
syncTimeoutDestroy(pSyncMsg);
if (syncEnvIsStart()) {
taosTmrReset(syncNodeEqHeartbeatTimer, pSyncNode->heartbeatTimerMS, pSyncNode, gSyncEnv->pTimerManager,
&pSyncNode->pHeartbeatTimer);
} else {
sError("sync env is stop, syncNodeEqHeartbeatTimer");
} }
} else { } else {
sError("syncNodeEqHeartbeatTimer FpEqMsg is NULL"); sTrace("==syncNodeEqHeartbeatTimer== heartbeatTimerLogicClock:%" PRIu64 ", heartbeatTimerLogicClockUser:%" PRIu64
"",
pSyncNode->heartbeatTimerLogicClock, pSyncNode->heartbeatTimerLogicClockUser);
} }
syncTimeoutDestroy(pSyncMsg);
if (syncEnvIsStart()) {
taosTmrReset(syncNodeEqHeartbeatTimer, pSyncNode->heartbeatTimerMS, pSyncNode, gSyncEnv->pTimerManager,
&pSyncNode->pHeartbeatTimer);
} else {
sError("sync env is stop, syncNodeEqHeartbeatTimer");
}
} else {
sTrace("==syncNodeEqHeartbeatTimer== heartbeatTimerLogicClock:%" PRIu64 ", heartbeatTimerLogicClockUser:%" PRIu64
"",
pSyncNode->heartbeatTimerLogicClock, pSyncNode->heartbeatTimerLogicClockUser);
} }
} }
@ -2184,7 +2186,7 @@ int32_t syncNodeCommit(SSyncNode* ths, SyncIndex beginIndex, SyncIndex endIndex,
ths->pFsm->FpRestoreFinishCb(ths->pFsm); ths->pFsm->FpRestoreFinishCb(ths->pFsm);
} }
ths->restoreFinish = true; ths->restoreFinish = true;
sDebug("vgId:%d sync event restore finish", ths->vgId); sDebug("vgId:%d sync event restore finish, index:%ld", ths->vgId, pEntry->index);
} }
} }