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

fix/TD-31891-remove-void-mndSync

This commit is contained in:
dmchen 2024-09-05 10:38:04 +00:00
parent a662b59041
commit b71e7f402e
1 changed files with 33 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -238,7 +238,9 @@ static int32_t mndPostMgmtCode(SMnode *pMnode, int32_t code) {
pMgmt->transSec = 0; pMgmt->transSec = 0;
pMgmt->transSeq = 0; pMgmt->transSeq = 0;
pMgmt->errCode = code; pMgmt->errCode = code;
(void)tsem_post(&pMgmt->syncSem); if (tsem_post(&pMgmt->syncSem) < 0) {
mError("trans:%d, failed to post sem", transId);
}
if (pMgmt->errCode != 0) { if (pMgmt->errCode != 0) {
mError("trans:%d, failed to propose since %s, post sem", transId, tstrerror(pMgmt->errCode)); mError("trans:%d, failed to propose since %s, post sem", transId, tstrerror(pMgmt->errCode));
@ -310,11 +312,15 @@ void mndRestoreFinish(const SSyncFSM *pFsm, const SyncIndex commitIdx) {
} else { } else {
mInfo("vgId:1, sync restore finished"); mInfo("vgId:1, sync restore finished");
} }
(void)mndRefreshUserIpWhiteList(pMnode); int32_t code = mndRefreshUserIpWhiteList(pMnode);
if (code != 0) {
mError("vgId:1, failed to refresh user ip white list since %s", tstrerror(code));
mndSetRestored(pMnode, false);
}
SyncIndex fsmIndex = mndSyncAppliedIndex(pFsm); SyncIndex fsmIndex = mndSyncAppliedIndex(pFsm);
if (commitIdx != fsmIndex) { if (commitIdx != fsmIndex) {
mError("vgId:1, sync restore finished, but commitIdx:%" PRId64 " is not equal to appliedIdx:%" PRId64, commitIdx, mError("vgId:1, failed to sync restore, commitIdx:%" PRId64 " is not equal to appliedIdx:%" PRId64, commitIdx,
fsmIndex); fsmIndex);
mndSetRestored(pMnode, false); mndSetRestored(pMnode, false);
} }
@ -368,7 +374,9 @@ static void mndBecomeFollower(const SSyncFSM *pFsm) {
pMgmt->transSec = 0; pMgmt->transSec = 0;
pMgmt->transSeq = 0; pMgmt->transSeq = 0;
pMgmt->errCode = TSDB_CODE_SYN_NOT_LEADER; pMgmt->errCode = TSDB_CODE_SYN_NOT_LEADER;
(void)tsem_post(&pMgmt->syncSem); if (tsem_post(&pMgmt->syncSem) < 0) {
mError("failed to post sem");
}
} }
(void)taosThreadMutexUnlock(&pMgmt->lock); (void)taosThreadMutexUnlock(&pMgmt->lock);
@ -387,7 +395,9 @@ static void mndBecomeLearner(const SSyncFSM *pFsm) {
pMgmt->transSec = 0; pMgmt->transSec = 0;
pMgmt->transSeq = 0; pMgmt->transSeq = 0;
pMgmt->errCode = TSDB_CODE_SYN_NOT_LEADER; pMgmt->errCode = TSDB_CODE_SYN_NOT_LEADER;
(void)tsem_post(&pMgmt->syncSem); if (tsem_post(&pMgmt->syncSem) < 0) {
mError("failed to post sem");
}
} }
(void)taosThreadMutexUnlock(&pMgmt->lock); (void)taosThreadMutexUnlock(&pMgmt->lock);
} }
@ -487,13 +497,19 @@ int32_t mndInitSync(SMnode *pMnode) {
pNode->nodePort = pMgmt->replicas[i].port; pNode->nodePort = pMgmt->replicas[i].port;
tstrncpy(pNode->nodeFqdn, pMgmt->replicas[i].fqdn, sizeof(pNode->nodeFqdn)); tstrncpy(pNode->nodeFqdn, pMgmt->replicas[i].fqdn, sizeof(pNode->nodeFqdn));
pNode->nodeRole = pMgmt->nodeRoles[i]; pNode->nodeRole = pMgmt->nodeRoles[i];
(void)tmsgUpdateDnodeInfo(&pNode->nodeId, &pNode->clusterId, pNode->nodeFqdn, &pNode->nodePort); if (tmsgUpdateDnodeInfo(&pNode->nodeId, &pNode->clusterId, pNode->nodeFqdn, &pNode->nodePort) != true) {
mError("failed to open sync, tmsgUpdateDnodeInfo");
TAOS_RETURN(-1);
}
mInfo("vgId:1, index:%d ep:%s:%u dnode:%d cluster:%" PRId64, i, pNode->nodeFqdn, pNode->nodePort, pNode->nodeId, mInfo("vgId:1, index:%d ep:%s:%u dnode:%d cluster:%" PRId64, i, pNode->nodeFqdn, pNode->nodePort, pNode->nodeId,
pNode->clusterId); pNode->clusterId);
} }
int32_t code = 0; int32_t code = 0;
(void)tsem_init(&pMgmt->syncSem, 0, 0); if ((code = tsem_init(&pMgmt->syncSem, 0, 0)) < 0) {
mError("failed to open sync, tsem_init, since %s", tstrerror(code));
TAOS_RETURN(code);
}
pMgmt->sync = syncOpen(&syncInfo, 1); // always check pMgmt->sync = syncOpen(&syncInfo, 1); // always check
if (pMgmt->sync <= 0) { if (pMgmt->sync <= 0) {
if (terrno != 0) code = terrno; if (terrno != 0) code = terrno;
@ -511,7 +527,9 @@ void mndCleanupSync(SMnode *pMnode) {
syncStop(pMgmt->sync); syncStop(pMgmt->sync);
mInfo("mnode-sync is stopped, id:%" PRId64, pMgmt->sync); mInfo("mnode-sync is stopped, id:%" PRId64, pMgmt->sync);
(void)tsem_destroy(&pMgmt->syncSem); if (tsem_destroy(&pMgmt->syncSem) < 0) {
mError("failed to destroy sem");
}
(void)taosThreadMutexDestroy(&pMgmt->lock); (void)taosThreadMutexDestroy(&pMgmt->lock);
memset(pMgmt, 0, sizeof(SSyncMgmt)); memset(pMgmt, 0, sizeof(SSyncMgmt));
} }
@ -531,7 +549,9 @@ void mndSyncCheckTimeout(SMnode *pMnode) {
pMgmt->transSeq = 0; pMgmt->transSeq = 0;
terrno = TSDB_CODE_SYN_TIMEOUT; terrno = TSDB_CODE_SYN_TIMEOUT;
pMgmt->errCode = TSDB_CODE_SYN_TIMEOUT; pMgmt->errCode = TSDB_CODE_SYN_TIMEOUT;
(void)tsem_post(&pMgmt->syncSem); if (tsem_post(&pMgmt->syncSem) < 0) {
mError("failed to post sem");
}
} else { } else {
mDebug("trans:%d, waiting for sync confirm, start:%d cur:%d delta:%d seq:%" PRId64, pMgmt->transId, mDebug("trans:%d, waiting for sync confirm, start:%d cur:%d delta:%d seq:%" PRId64, pMgmt->transId,
pMgmt->transSec, curSec, curSec - pMgmt->transSec, pMgmt->transSeq); pMgmt->transSec, curSec, curSec - pMgmt->transSec, pMgmt->transSeq);
@ -572,7 +592,7 @@ int32_t mndSyncPropose(SMnode *pMnode, SSdbRaw *pRaw, int32_t transId) {
mInfo("trans:%d, is proposing and wait sem, seq:%" PRId64, transId, seq); mInfo("trans:%d, is proposing and wait sem, seq:%" PRId64, transId, seq);
pMgmt->transSeq = seq; pMgmt->transSeq = seq;
(void)taosThreadMutexUnlock(&pMgmt->lock); (void)taosThreadMutexUnlock(&pMgmt->lock);
(void)tsem_wait(&pMgmt->syncSem); code = tsem_wait(&pMgmt->syncSem);
} else if (code > 0) { } else if (code > 0) {
mInfo("trans:%d, confirm at once since replica is 1, continue execute", transId); mInfo("trans:%d, confirm at once since replica is 1, continue execute", transId);
pMgmt->transId = 0; pMgmt->transId = 0;
@ -624,7 +644,9 @@ void mndSyncStop(SMnode *pMnode) {
pMgmt->transId = 0; pMgmt->transId = 0;
pMgmt->transSec = 0; pMgmt->transSec = 0;
pMgmt->errCode = TSDB_CODE_APP_IS_STOPPING; pMgmt->errCode = TSDB_CODE_APP_IS_STOPPING;
(void)tsem_post(&pMgmt->syncSem); if (tsem_post(&pMgmt->syncSem) < 0) {
mError("failed to post sem");
}
} }
(void)taosThreadMutexUnlock(&pMgmt->lock); (void)taosThreadMutexUnlock(&pMgmt->lock);
} }