From 8060108585cb9cfc015d95899fb5def9036825b9 Mon Sep 17 00:00:00 2001 From: Minghao Li Date: Fri, 24 Jun 2022 13:50:23 +0800 Subject: [PATCH 1/5] refactor(sync): add trace log --- source/libs/sync/src/syncElection.c | 9 ++++++- source/libs/sync/src/syncReplication.c | 4 ++-- source/libs/sync/src/syncRequestVote.c | 33 +++++++++++++++++++------- 3 files changed, 35 insertions(+), 11 deletions(-) diff --git a/source/libs/sync/src/syncElection.c b/source/libs/sync/src/syncElection.c index e0f2b0fed2..738b17b9cb 100644 --- a/source/libs/sync/src/syncElection.c +++ b/source/libs/sync/src/syncElection.c @@ -105,9 +105,16 @@ int32_t syncNodeElect(SSyncNode* pSyncNode) { } int32_t syncNodeRequestVote(SSyncNode* pSyncNode, const SRaftId* destRaftId, const SyncRequestVote* pMsg) { - sTrace("syncNodeRequestVote pSyncNode:%p ", pSyncNode); int32_t ret = 0; + do { + char host[128]; + uint16_t port; + syncUtilU642Addr(destRaftId->addr, host, sizeof(host), &port); + sDebug("vgId:%d, send sync-request-vote to %s:%d, {term:%lu, last-index:%ld, last-term:%lu}", pSyncNode->vgId, host, + port, pMsg->term, pMsg->lastLogTerm, pMsg->lastLogIndex); + } while (0); + SRpcMsg rpcMsg; syncRequestVote2RpcMsg(pMsg, &rpcMsg); syncNodeSendMsgById(destRaftId, pSyncNode, &rpcMsg); diff --git a/source/libs/sync/src/syncReplication.c b/source/libs/sync/src/syncReplication.c index 02cd977e1e..46850758f1 100644 --- a/source/libs/sync/src/syncReplication.c +++ b/source/libs/sync/src/syncReplication.c @@ -224,8 +224,8 @@ int32_t syncNodeAppendEntries(SSyncNode* pSyncNode, const SRaftId* destRaftId, c uint16_t port; syncUtilU642Addr(destRaftId->addr, host, sizeof(host), &port); sDebug( - "vgId:%d, send sync-append-entries to %s:%d, term:%lu, pre-index:%ld, pre-term:%lu, pterm:%lu, commit:%ld, " - "datalen:%d", + "vgId:%d, send sync-append-entries to %s:%d, {term:%lu, pre-index:%ld, pre-term:%lu, pterm:%lu, commit:%ld, " + "datalen:%d}", pSyncNode->vgId, host, port, pMsg->term, pMsg->prevLogIndex, pMsg->prevLogTerm, pMsg->privateTerm, pMsg->commitIndex, pMsg->dataLen); } while (0); diff --git a/source/libs/sync/src/syncRequestVote.c b/source/libs/sync/src/syncRequestVote.c index def5c321ad..9b181e21e5 100644 --- a/source/libs/sync/src/syncRequestVote.c +++ b/source/libs/sync/src/syncRequestVote.c @@ -99,15 +99,20 @@ static bool syncNodeOnRequestVoteLogOK(SSyncNode* pSyncNode, SyncRequestVote* pM int32_t syncNodeOnRequestVoteSnapshotCb(SSyncNode* ths, SyncRequestVote* pMsg) { int32_t ret = 0; - // print log - char logBuf[128] = {0}; - snprintf(logBuf, sizeof(logBuf), "recv SyncRequestVote, currentTerm:%lu", ths->pRaftStore->currentTerm); - syncRequestVoteLog2(logBuf, pMsg); - // if already drop replica, do not process if (!syncNodeInRaftGroup(ths, &(pMsg->srcId)) && !ths->pRaftCfg->isStandBy) { - sInfo("recv SyncRequestVote maybe replica already dropped"); - return ret; + do { + char logBuf[128]; + char host[64]; + uint16_t port; + syncUtilU642Addr(pMsg->srcId.addr, host, sizeof(host), &port); + snprintf(logBuf, sizeof(logBuf), + "recv sync-request-vote from %s:%d, term:%lu, lindex:%ld, lterm:%lu, maybe replica already dropped", + host, port, pMsg->term, pMsg->lastLogIndex, pMsg->lastLogTerm); + syncNodeEventLog(ths, logBuf); + } while (0); + + return -1; } // maybe update term @@ -135,10 +140,22 @@ int32_t syncNodeOnRequestVoteSnapshotCb(SSyncNode* ths, SyncRequestVote* pMsg) { pReply->term = ths->pRaftStore->currentTerm; pReply->voteGranted = grant; + // trace log + do { + char logBuf[128]; + char host[64]; + uint16_t port; + syncUtilU642Addr(pMsg->srcId.addr, host, sizeof(host), &port); + snprintf(logBuf, sizeof(logBuf), + "recv sync-request-vote from %s:%d, term:%lu, lindex:%ld, lterm:%lu, reply-grant:%d", host, port, + pMsg->term, pMsg->lastLogIndex, pMsg->lastLogTerm, pReply->voteGranted); + syncNodeEventLog(ths, logBuf); + } while (0); + SRpcMsg rpcMsg; syncRequestVoteReply2RpcMsg(pReply, &rpcMsg); syncNodeSendMsgById(&pReply->destId, ths, &rpcMsg); syncRequestVoteReplyDestroy(pReply); - return ret; + return 0; } \ No newline at end of file From 8a8e42a9233f1ee2e6c1d9f1a9295adb94f55039 Mon Sep 17 00:00:00 2001 From: Minghao Li Date: Fri, 24 Jun 2022 14:04:58 +0800 Subject: [PATCH 2/5] refactor(sync): set error code in syncIsReady --- source/dnode/mnode/impl/src/mndSync.c | 3 ++- source/libs/sync/src/syncMain.c | 9 +++++++++ 2 files changed, 11 insertions(+), 1 deletion(-) diff --git a/source/dnode/mnode/impl/src/mndSync.c b/source/dnode/mnode/impl/src/mndSync.c index e3358b41df..b481f417d9 100644 --- a/source/dnode/mnode/impl/src/mndSync.c +++ b/source/dnode/mnode/impl/src/mndSync.c @@ -261,7 +261,8 @@ bool mndIsMaster(SMnode *pMnode) { SSyncMgmt *pMgmt = &pMnode->syncMgmt; if (!syncIsReady(pMgmt->sync)) { - terrno = TSDB_CODE_SYN_NOT_LEADER; + // get terrno from syncIsReady + // terrno = TSDB_CODE_SYN_NOT_LEADER; return false; } diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c index 42ba2b85b1..041afdb337 100644 --- a/source/libs/sync/src/syncMain.c +++ b/source/libs/sync/src/syncMain.c @@ -370,6 +370,15 @@ bool syncIsReady(int64_t rid) { ASSERT(rid == pSyncNode->rid); bool b = (pSyncNode->state == TAOS_SYNC_STATE_LEADER) && pSyncNode->restoreFinish; taosReleaseRef(tsNodeRefId, pSyncNode->rid); + + // if false, set error code + if (false == b) { + if (pSyncNode->state != TAOS_SYNC_STATE_LEADER) { + terrno = TSDB_CODE_SYN_NOT_LEADER; + } else { + terrno = TSDB_CODE_APP_NOT_READY; + } + } return b; } From 87defc2790ec69b3e10917c67302a12225b17e22 Mon Sep 17 00:00:00 2001 From: Minghao Li Date: Fri, 24 Jun 2022 14:26:31 +0800 Subject: [PATCH 3/5] refactor(sync): add interface: get retry epset --- include/libs/sync/sync.h | 1 + source/dnode/vnode/src/vnd/vnodeSync.c | 4 ++++ source/libs/sync/src/syncMain.c | 24 +++++++++++++++++++++--- 3 files changed, 26 insertions(+), 3 deletions(-) diff --git a/include/libs/sync/sync.h b/include/libs/sync/sync.h index 2bf49fa006..52d9b6f810 100644 --- a/include/libs/sync/sync.h +++ b/include/libs/sync/sync.h @@ -200,6 +200,7 @@ const char* syncGetMyRoleStr(int64_t rid); SyncTerm syncGetMyTerm(int64_t rid); SyncGroupId syncGetVgId(int64_t rid); void syncGetEpSet(int64_t rid, SEpSet* pEpSet); +void syncGetRetryEpSet(int64_t rid, SEpSet* pEpSet); int32_t syncPropose(int64_t rid, SRpcMsg* pMsg, bool isWeak); bool syncEnvIsStart(); const char* syncStr(ESyncState state); diff --git a/source/dnode/vnode/src/vnd/vnodeSync.c b/source/dnode/vnode/src/vnd/vnodeSync.c index 32090b774e..4c56180979 100644 --- a/source/dnode/vnode/src/vnd/vnodeSync.c +++ b/source/dnode/vnode/src/vnd/vnodeSync.c @@ -144,11 +144,15 @@ void vnodeProposeMsg(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) { vnodeAccumBlockMsg(pVnode, pMsg->msgType); } else if (code == -1 && terrno == TSDB_CODE_SYN_NOT_LEADER) { SEpSet newEpSet = {0}; + syncGetRetryEpSet(pVnode->sync, &newEpSet); + + /* syncGetEpSet(pVnode->sync, &newEpSet); SEp *pEp = &newEpSet.eps[newEpSet.inUse]; if (pEp->port == tsServerPort && strcmp(pEp->fqdn, tsLocalFqdn) == 0) { newEpSet.inUse = (newEpSet.inUse + 1) % newEpSet.numOfEps; } + */ vGTrace("vgId:%d, msg:%p is redirect since not leader, numOfEps:%d inUse:%d", vgId, pMsg, newEpSet.numOfEps, newEpSet.inUse); diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c index 041afdb337..c5af72c971 100644 --- a/source/libs/sync/src/syncMain.c +++ b/source/libs/sync/src/syncMain.c @@ -489,12 +489,30 @@ void syncGetEpSet(int64_t rid, SEpSet* pEpSet) { snprintf(pEpSet->eps[i].fqdn, sizeof(pEpSet->eps[i].fqdn), "%s", (pSyncNode->pRaftCfg->cfg.nodeInfo)[i].nodeFqdn); pEpSet->eps[i].port = (pSyncNode->pRaftCfg->cfg.nodeInfo)[i].nodePort; (pEpSet->numOfEps)++; - - sInfo("syncGetEpSet index:%d %s:%d", i, pEpSet->eps[i].fqdn, pEpSet->eps[i].port); + sInfo("vgId:%d sync get epset: index:%d %s:%d", pSyncNode->vgId, i, pEpSet->eps[i].fqdn, pEpSet->eps[i].port); } pEpSet->inUse = pSyncNode->pRaftCfg->cfg.myIndex; + sInfo("vgId:%d sync get epset in-use:%d", pSyncNode->vgId, pEpSet->inUse); - sInfo("syncGetEpSet pEpSet->inUse:%d ", pEpSet->inUse); + taosReleaseRef(tsNodeRefId, pSyncNode->rid); +} + +void syncGetRetryEpSet(int64_t rid, SEpSet* pEpSet) { + SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); + if (pSyncNode == NULL) { + memset(pEpSet, 0, sizeof(*pEpSet)); + return; + } + ASSERT(rid == pSyncNode->rid); + pEpSet->numOfEps = 0; + for (int i = 0; i < pSyncNode->pRaftCfg->cfg.replicaNum; ++i) { + snprintf(pEpSet->eps[i].fqdn, sizeof(pEpSet->eps[i].fqdn), "%s", (pSyncNode->pRaftCfg->cfg.nodeInfo)[i].nodeFqdn); + pEpSet->eps[i].port = (pSyncNode->pRaftCfg->cfg.nodeInfo)[i].nodePort; + (pEpSet->numOfEps)++; + sInfo("vgId:%d sync get retry epset: index:%d %s:%d", pSyncNode->vgId, i, pEpSet->eps[i].fqdn, pEpSet->eps[i].port); + } + pEpSet->inUse = (pSyncNode->pRaftCfg->cfg.myIndex + 1) % pEpSet->numOfEps; + sInfo("vgId:%d sync get retry epset in-use:%d", pSyncNode->vgId, pEpSet->inUse); taosReleaseRef(tsNodeRefId, pSyncNode->rid); } From 7695dbcce776f0429906059244c9b11135dd302c Mon Sep 17 00:00:00 2001 From: Minghao Li Date: Fri, 24 Jun 2022 15:28:01 +0800 Subject: [PATCH 4/5] refactor(sync): adjust election timer --- source/dnode/mnode/impl/src/mndSync.c | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/source/dnode/mnode/impl/src/mndSync.c b/source/dnode/mnode/impl/src/mndSync.c index b481f417d9..c58ce97588 100644 --- a/source/dnode/mnode/impl/src/mndSync.c +++ b/source/dnode/mnode/impl/src/mndSync.c @@ -198,6 +198,10 @@ int32_t mndInitSync(SMnode *pMnode) { return -1; } + // decrease election timer + setElectTimerMS(pMgmt->sync, 600); + setHeartbeatTimerMS(pMgmt->sync, 300); + mDebug("mnode-sync is opened, id:%" PRId64, pMgmt->sync); return 0; } From 58296f6a5abf559915cbc11a50150300a85d3c84 Mon Sep 17 00:00:00 2001 From: Minghao Li Date: Fri, 24 Jun 2022 16:44:45 +0800 Subject: [PATCH 5/5] refactor(sync): adjust log buf size --- source/libs/sync/src/syncRequestVote.c | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/source/libs/sync/src/syncRequestVote.c b/source/libs/sync/src/syncRequestVote.c index 9b181e21e5..3270b5c0e4 100644 --- a/source/libs/sync/src/syncRequestVote.c +++ b/source/libs/sync/src/syncRequestVote.c @@ -102,7 +102,7 @@ int32_t syncNodeOnRequestVoteSnapshotCb(SSyncNode* ths, SyncRequestVote* pMsg) { // if already drop replica, do not process if (!syncNodeInRaftGroup(ths, &(pMsg->srcId)) && !ths->pRaftCfg->isStandBy) { do { - char logBuf[128]; + char logBuf[256]; char host[64]; uint16_t port; syncUtilU642Addr(pMsg->srcId.addr, host, sizeof(host), &port); @@ -142,7 +142,7 @@ int32_t syncNodeOnRequestVoteSnapshotCb(SSyncNode* ths, SyncRequestVote* pMsg) { // trace log do { - char logBuf[128]; + char logBuf[256]; char host[64]; uint16_t port; syncUtilU642Addr(pMsg->srcId.addr, host, sizeof(host), &port);