diff --git a/source/libs/sync/inc/syncInt.h b/source/libs/sync/inc/syncInt.h index 635ebd63080e94ef8e8bad5b0b55947d328bd745..a158430a0ff6d361c072fb02d4bdf6317732ae5d 100644 --- a/source/libs/sync/inc/syncInt.h +++ b/source/libs/sync/inc/syncInt.h @@ -248,6 +248,7 @@ char* syncNode2SimpleStr(const SSyncNode* pSyncNode); bool syncNodeInConfig(SSyncNode* pSyncNode, const SSyncCfg* config); void syncNodeDoConfigChange(SSyncNode* pSyncNode, SSyncCfg* newConfig, SyncIndex lastConfigChangeIndex); SyncIndex syncMinMatchIndex(SSyncNode* pSyncNode); +char* syncNodePeerState2Str(const SSyncNode* pSyncNode); SSyncNode* syncNodeAcquire(int64_t rid); void syncNodeRelease(SSyncNode* pNode); diff --git a/source/libs/sync/src/syncCommit.c b/source/libs/sync/src/syncCommit.c index f377b086717731b5836f71c56b59c58acee17d42..b9d7789ca2032bb2ad35d7647bd1513489f5f09d 100644 --- a/source/libs/sync/src/syncCommit.c +++ b/source/libs/sync/src/syncCommit.c @@ -170,6 +170,8 @@ static inline int64_t syncNodeAbs64(int64_t a, int64_t b) { } int32_t syncNodeDynamicQuorum(const SSyncNode* pSyncNode) { + return pSyncNode->quorum; + int32_t quorum = 1; // self int64_t timeNow = taosGetTimestampMs(); diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c index e102ae31a20f4e88a539f6d6b494291f34a64843..ba4ce8def37e38876db6a3b633a662d965b6230f 100644 --- a/source/libs/sync/src/syncMain.c +++ b/source/libs/sync/src/syncMain.c @@ -322,6 +322,38 @@ SyncIndex syncMinMatchIndex(SSyncNode* pSyncNode) { return minMatchIndex; } +char* syncNodePeerState2Str(const SSyncNode* pSyncNode) { + int32_t len = 128; + int32_t useLen = 0; + int32_t leftLen = len - useLen; + char* pStr = taosMemoryMalloc(len); + memset(pStr, 0, len); + + char* p = pStr; + int32_t use = snprintf(p, leftLen, "{"); + useLen += use; + leftLen -= use; + + for (int32_t i = 0; i < pSyncNode->replicaNum; ++i) { + SPeerState* pState = syncNodeGetPeerState((SSyncNode*)pSyncNode, &(pSyncNode->replicasId[i])); + ASSERT(pState != NULL); + + p = pStr + useLen; + use = snprintf(p, leftLen, "%d:%ld,%ld, ", i, pState->lastSendIndex, pState->lastSendTime); + useLen += use; + leftLen -= use; + } + + p = pStr + useLen; + use = snprintf(p, leftLen, "}"); + useLen += use; + leftLen -= use; + + // sTrace("vgId:%d, ------------------ syncNodePeerState2Str:%s", pSyncNode->vgId, pStr); + + return pStr; +} + int32_t syncBeginSnapshot(int64_t rid, int64_t lastApplyIndex) { SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); if (pSyncNode == NULL) { @@ -1825,8 +1857,6 @@ char* syncNode2Str(const SSyncNode* pSyncNode) { } inline void syncNodeEventLog(const SSyncNode* pSyncNode, char* str) { - int32_t userStrLen = strlen(str); - SSnapshot snapshot = {.data = NULL, .lastApplyIndex = -1, .lastApplyTerm = 0}; if (pSyncNode->pFsm != NULL && pSyncNode->pFsm->FpGetSnapshotInfo != NULL) { pSyncNode->pFsm->FpGetSnapshotInfo(pSyncNode->pFsm, &snapshot); @@ -1845,6 +1875,9 @@ inline void syncNodeEventLog(const SSyncNode* pSyncNode, char* str) { printStr = pCfgStr; } + char* peerStateStr = syncNodePeerState2Str(pSyncNode); + int32_t userStrLen = strlen(str) + strlen(peerStateStr); + if (userStrLen < 256) { char logBuf[256 + 256]; if (pSyncNode != NULL && pSyncNode->pRaftCfg != NULL && pSyncNode->pRaftStore != NULL) { @@ -1854,13 +1887,13 @@ inline void syncNodeEventLog(const SSyncNode* pSyncNode, char* str) { ", sby:%d, " "stgy:%d, bch:%d, " "r-num:%d, " - "lcfg:%" PRId64 ", chging:%d, rsto:%d, dquorum:%d, elt:%" PRId64 ", hb:%" PRId64 ", %s", + "lcfg:%" PRId64 ", chging:%d, rsto:%d, dquorum:%d, elt:%" PRId64 ", hb:%" PRId64 ", %s, %s", pSyncNode->vgId, syncUtilState2String(pSyncNode->state), str, pSyncNode->pRaftStore->currentTerm, pSyncNode->commitIndex, logBeginIndex, logLastIndex, pSyncNode->minMatchIndex, snapshot.lastApplyIndex, snapshot.lastApplyTerm, pSyncNode->pRaftCfg->isStandBy, pSyncNode->pRaftCfg->snapshotStrategy, pSyncNode->pRaftCfg->batchSize, pSyncNode->replicaNum, pSyncNode->pRaftCfg->lastConfigIndex, pSyncNode->changing, pSyncNode->restoreFinish, syncNodeDynamicQuorum(pSyncNode), - pSyncNode->electTimerLogicClockUser, pSyncNode->heartbeatTimerLogicClockUser, printStr); + pSyncNode->electTimerLogicClockUser, pSyncNode->heartbeatTimerLogicClockUser, peerStateStr, printStr); } else { snprintf(logBuf, sizeof(logBuf), "%s", str); } @@ -1878,19 +1911,20 @@ inline void syncNodeEventLog(const SSyncNode* pSyncNode, char* str) { ", sby:%d, " "stgy:%d, bch:%d, " "r-num:%d, " - "lcfg:%" PRId64 ", chging:%d, rsto:%d, dquorum:%d, elt:%" PRId64 ", hb:%" PRId64 ", %s", + "lcfg:%" PRId64 ", chging:%d, rsto:%d, dquorum:%d, elt:%" PRId64 ", hb:%" PRId64 ", %s, %s", pSyncNode->vgId, syncUtilState2String(pSyncNode->state), str, pSyncNode->pRaftStore->currentTerm, pSyncNode->commitIndex, logBeginIndex, logLastIndex, pSyncNode->minMatchIndex, snapshot.lastApplyIndex, snapshot.lastApplyTerm, pSyncNode->pRaftCfg->isStandBy, pSyncNode->pRaftCfg->snapshotStrategy, pSyncNode->pRaftCfg->batchSize, pSyncNode->replicaNum, pSyncNode->pRaftCfg->lastConfigIndex, pSyncNode->changing, pSyncNode->restoreFinish, syncNodeDynamicQuorum(pSyncNode), - pSyncNode->electTimerLogicClockUser, pSyncNode->heartbeatTimerLogicClockUser, printStr); + pSyncNode->electTimerLogicClockUser, pSyncNode->heartbeatTimerLogicClockUser, peerStateStr, printStr); } else { snprintf(s, len, "%s", str); } // sDebug("%s", s); // sInfo("%s", s); sTrace("%s", s); + taosMemoryFree(peerStateStr); taosMemoryFree(s); } @@ -3441,7 +3475,9 @@ SPeerState* syncNodeGetPeerState(SSyncNode* ths, const SRaftId* pDestId) { bool syncNodeNeedSendAppendEntries(SSyncNode* ths, const SRaftId* pDestId, const SyncAppendEntries* pMsg) { SPeerState* pState = syncNodeGetPeerState(ths, pDestId); - ASSERT(pState != NULL); + if (pState == NULL) { + return false; + } SyncIndex sendIndex = pMsg->prevLogIndex + 1; int64_t tsNow = taosGetTimestampMs(); diff --git a/source/libs/sync/src/syncReplication.c b/source/libs/sync/src/syncReplication.c index 431e1b46d09177cf5b625747a3460679c77dfe37..4231033e5d540741d31da35df3aca1be7cd099ca 100644 --- a/source/libs/sync/src/syncReplication.c +++ b/source/libs/sync/src/syncReplication.c @@ -165,8 +165,10 @@ int32_t syncNodeSendAppendEntries(SSyncNode* pSyncNode, const SRaftId* destRaftI SPeerState* pState = syncNodeGetPeerState(pSyncNode, destRaftId); ASSERT(pState != NULL); - pState->lastSendIndex = pMsg->prevLogIndex + 1; - pState->lastSendTime = taosGetTimestampMs(); + if (pMsg->dataLen > 0) { + pState->lastSendIndex = pMsg->prevLogIndex + 1; + pState->lastSendTime = taosGetTimestampMs(); + } return ret; }