From 7e67dc032db609a9d4eb19e2eb7a1563880c1d10 Mon Sep 17 00:00:00 2001 From: Shengliang Guan Date: Thu, 29 Dec 2022 10:41:01 +0800 Subject: [PATCH 1/3] refact: rename variables --- source/libs/sync/inc/syncIndexMgr.h | 36 ++++---- source/libs/sync/src/syncIndexMgr.c | 124 ++++++++++++++-------------- 2 files changed, 78 insertions(+), 82 deletions(-) diff --git a/source/libs/sync/inc/syncIndexMgr.h b/source/libs/sync/inc/syncIndexMgr.h index 79b4fa0fbf..4e6ab284f8 100644 --- a/source/libs/sync/inc/syncIndexMgr.h +++ b/source/libs/sync/inc/syncIndexMgr.h @@ -25,29 +25,27 @@ extern "C" { // SIndexMgr ----------------------------- typedef struct SSyncIndexMgr { SRaftId (*replicas)[TSDB_MAX_REPLICA]; - SyncIndex index[TSDB_MAX_REPLICA]; - SyncTerm privateTerm[TSDB_MAX_REPLICA]; // for advanced function - - int64_t startTimeArr[TSDB_MAX_REPLICA]; - int64_t recvTimeArr[TSDB_MAX_REPLICA]; - + SyncIndex index[TSDB_MAX_REPLICA]; + SyncTerm privateTerm[TSDB_MAX_REPLICA]; // for advanced function + int64_t startTimeArr[TSDB_MAX_REPLICA]; + int64_t recvTimeArr[TSDB_MAX_REPLICA]; int32_t replicaNum; - SSyncNode *pSyncNode; + SSyncNode *pNode; } SSyncIndexMgr; -SSyncIndexMgr *syncIndexMgrCreate(SSyncNode *pSyncNode); -void syncIndexMgrUpdate(SSyncIndexMgr *pSyncIndexMgr, SSyncNode *pSyncNode); -void syncIndexMgrDestroy(SSyncIndexMgr *pSyncIndexMgr); -void syncIndexMgrClear(SSyncIndexMgr *pSyncIndexMgr); -void syncIndexMgrSetIndex(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId, SyncIndex index); -SyncIndex syncIndexMgrGetIndex(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId); +SSyncIndexMgr *syncIndexMgrCreate(SSyncNode *pNode); +void syncIndexMgrUpdate(SSyncIndexMgr *pIndexMgr, SSyncNode *pNode); +void syncIndexMgrDestroy(SSyncIndexMgr *pIndexMgr); +void syncIndexMgrClear(SSyncIndexMgr *pIndexMgr); +void syncIndexMgrSetIndex(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId, SyncIndex index); +SyncIndex syncIndexMgrGetIndex(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId); -void syncIndexMgrSetStartTime(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId, int64_t startTime); -int64_t syncIndexMgrGetStartTime(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId); -void syncIndexMgrSetRecvTime(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId, int64_t recvTime); -int64_t syncIndexMgrGetRecvTime(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId); -void syncIndexMgrSetTerm(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId, SyncTerm term); -SyncTerm syncIndexMgrGetTerm(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId); +void syncIndexMgrSetStartTime(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId, int64_t startTime); +int64_t syncIndexMgrGetStartTime(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId); +void syncIndexMgrSetRecvTime(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId, int64_t recvTime); +int64_t syncIndexMgrGetRecvTime(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId); +void syncIndexMgrSetTerm(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId, SyncTerm term); +SyncTerm syncIndexMgrGetTerm(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId); #ifdef __cplusplus } diff --git a/source/libs/sync/src/syncIndexMgr.c b/source/libs/sync/src/syncIndexMgr.c index 7933258e53..8ea5d782b7 100644 --- a/source/libs/sync/src/syncIndexMgr.c +++ b/source/libs/sync/src/syncIndexMgr.c @@ -17,56 +17,56 @@ #include "syncIndexMgr.h" #include "syncUtil.h" -SSyncIndexMgr *syncIndexMgrCreate(SSyncNode *pSyncNode) { - SSyncIndexMgr *pSyncIndexMgr = taosMemoryCalloc(1, sizeof(SSyncIndexMgr)); - if (pSyncIndexMgr == NULL) { +SSyncIndexMgr *syncIndexMgrCreate(SSyncNode *pNode) { + SSyncIndexMgr *pIndexMgr = taosMemoryCalloc(1, sizeof(SSyncIndexMgr)); + if (pIndexMgr == NULL) { terrno = TSDB_CODE_OUT_OF_MEMORY; return NULL; } - pSyncIndexMgr->replicas = &(pSyncNode->replicasId); - pSyncIndexMgr->replicaNum = pSyncNode->replicaNum; - pSyncIndexMgr->pSyncNode = pSyncNode; - syncIndexMgrClear(pSyncIndexMgr); + pIndexMgr->replicas = &(pNode->replicasId); + pIndexMgr->replicaNum = pNode->replicaNum; + pIndexMgr->pNode = pNode; + syncIndexMgrClear(pIndexMgr); - return pSyncIndexMgr; + return pIndexMgr; } -void syncIndexMgrUpdate(SSyncIndexMgr *pSyncIndexMgr, SSyncNode *pSyncNode) { - pSyncIndexMgr->replicas = &(pSyncNode->replicasId); - pSyncIndexMgr->replicaNum = pSyncNode->replicaNum; - pSyncIndexMgr->pSyncNode = pSyncNode; - syncIndexMgrClear(pSyncIndexMgr); +void syncIndexMgrUpdate(SSyncIndexMgr *pIndexMgr, SSyncNode *pNode) { + pIndexMgr->replicas = &(pNode->replicasId); + pIndexMgr->replicaNum = pNode->replicaNum; + pIndexMgr->pNode = pNode; + syncIndexMgrClear(pIndexMgr); } -void syncIndexMgrDestroy(SSyncIndexMgr *pSyncIndexMgr) { - if (pSyncIndexMgr != NULL) { - taosMemoryFree(pSyncIndexMgr); +void syncIndexMgrDestroy(SSyncIndexMgr *pIndexMgr) { + if (pIndexMgr != NULL) { + taosMemoryFree(pIndexMgr); } } -void syncIndexMgrClear(SSyncIndexMgr *pSyncIndexMgr) { - memset(pSyncIndexMgr->index, 0, sizeof(pSyncIndexMgr->index)); - memset(pSyncIndexMgr->privateTerm, 0, sizeof(pSyncIndexMgr->privateTerm)); +void syncIndexMgrClear(SSyncIndexMgr *pIndexMgr) { + memset(pIndexMgr->index, 0, sizeof(pIndexMgr->index)); + memset(pIndexMgr->privateTerm, 0, sizeof(pIndexMgr->privateTerm)); // int64_t timeNow = taosGetMonotonicMs(); int64_t timeNow = taosGetTimestampMs(); - for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) { - pSyncIndexMgr->startTimeArr[i] = 0; - pSyncIndexMgr->recvTimeArr[i] = timeNow; + for (int i = 0; i < pIndexMgr->replicaNum; ++i) { + pIndexMgr->startTimeArr[i] = 0; + pIndexMgr->recvTimeArr[i] = timeNow; } /* - for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) { - pSyncIndexMgr->index[i] = 0; + for (int i = 0; i < pIndexMgr->replicaNum; ++i) { + pIndexMgr->index[i] = 0; } */ } -void syncIndexMgrSetIndex(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId, SyncIndex index) { - for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) { - if (syncUtilSameId(&((*(pSyncIndexMgr->replicas))[i]), pRaftId)) { - (pSyncIndexMgr->index)[i] = index; +void syncIndexMgrSetIndex(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId, SyncIndex index) { + for (int i = 0; i < pIndexMgr->replicaNum; ++i) { + if (syncUtilSameId(&((*(pIndexMgr->replicas))[i]), pRaftId)) { + (pIndexMgr->index)[i] = index; return; } } @@ -77,8 +77,7 @@ void syncIndexMgrSetIndex(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId, char host[128]; uint16_t port; syncUtilU642Addr(pRaftId->addr, host, sizeof(host), &port); - sError("vgId:%d, index mgr set for %s:%d, index:%" PRId64 " error", pSyncIndexMgr->pSyncNode->vgId, host, port, - index); + sError("vgId:%d, index mgr set for %s:%d, index:%" PRId64 " error", pIndexMgr->pNode->vgId, host, port, index); } SSyncLogReplMgr *syncNodeGetLogReplMgr(SSyncNode *pNode, SRaftId *pDestId) { @@ -90,14 +89,14 @@ SSyncLogReplMgr *syncNodeGetLogReplMgr(SSyncNode *pNode, SRaftId *pDestId) { return NULL; } -SyncIndex syncIndexMgrGetIndex(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId) { - if (pSyncIndexMgr == NULL) { +SyncIndex syncIndexMgrGetIndex(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId) { + if (pIndexMgr == NULL) { return SYNC_INDEX_INVALID; } - for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) { - if (syncUtilSameId(&((*(pSyncIndexMgr->replicas))[i]), pRaftId)) { - SyncIndex idx = (pSyncIndexMgr->index)[i]; + for (int i = 0; i < pIndexMgr->replicaNum; ++i) { + if (syncUtilSameId(&((*(pIndexMgr->replicas))[i]), pRaftId)) { + SyncIndex idx = (pIndexMgr->index)[i]; return idx; } } @@ -105,10 +104,10 @@ SyncIndex syncIndexMgrGetIndex(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaf return SYNC_INDEX_INVALID; } -void syncIndexMgrSetStartTime(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId, int64_t startTime) { - for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) { - if (syncUtilSameId(&((*(pSyncIndexMgr->replicas))[i]), pRaftId)) { - (pSyncIndexMgr->startTimeArr)[i] = startTime; +void syncIndexMgrSetStartTime(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId, int64_t startTime) { + for (int i = 0; i < pIndexMgr->replicaNum; ++i) { + if (syncUtilSameId(&((*(pIndexMgr->replicas))[i]), pRaftId)) { + (pIndexMgr->startTimeArr)[i] = startTime; return; } } @@ -118,14 +117,14 @@ void syncIndexMgrSetStartTime(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaft char host[128]; uint16_t port; syncUtilU642Addr(pRaftId->addr, host, sizeof(host), &port); - sError("vgId:%d, index mgr set for %s:%d, start-time:%" PRId64 " error", pSyncIndexMgr->pSyncNode->vgId, host, port, + sError("vgId:%d, index mgr set for %s:%d, start-time:%" PRId64 " error", pIndexMgr->pNode->vgId, host, port, startTime); } -int64_t syncIndexMgrGetStartTime(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId) { - for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) { - if (syncUtilSameId(&((*(pSyncIndexMgr->replicas))[i]), pRaftId)) { - int64_t startTime = (pSyncIndexMgr->startTimeArr)[i]; +int64_t syncIndexMgrGetStartTime(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId) { + for (int i = 0; i < pIndexMgr->replicaNum; ++i) { + if (syncUtilSameId(&((*(pIndexMgr->replicas))[i]), pRaftId)) { + int64_t startTime = (pIndexMgr->startTimeArr)[i]; return startTime; } } @@ -133,10 +132,10 @@ int64_t syncIndexMgrGetStartTime(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pR return -1; } -void syncIndexMgrSetRecvTime(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId, int64_t recvTime) { - for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) { - if (syncUtilSameId(&((*(pSyncIndexMgr->replicas))[i]), pRaftId)) { - (pSyncIndexMgr->recvTimeArr)[i] = recvTime; +void syncIndexMgrSetRecvTime(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId, int64_t recvTime) { + for (int i = 0; i < pIndexMgr->replicaNum; ++i) { + if (syncUtilSameId(&((*(pIndexMgr->replicas))[i]), pRaftId)) { + (pIndexMgr->recvTimeArr)[i] = recvTime; return; } } @@ -146,14 +145,13 @@ void syncIndexMgrSetRecvTime(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftI char host[128]; uint16_t port; syncUtilU642Addr(pRaftId->addr, host, sizeof(host), &port); - sError("vgId:%d, index mgr set for %s:%d, recv-time:%" PRId64 " error", pSyncIndexMgr->pSyncNode->vgId, host, port, - recvTime); + sError("vgId:%d, index mgr set for %s:%d, recv-time:%" PRId64 " error", pIndexMgr->pNode->vgId, host, port, recvTime); } -int64_t syncIndexMgrGetRecvTime(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId) { - for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) { - if (syncUtilSameId(&((*(pSyncIndexMgr->replicas))[i]), pRaftId)) { - int64_t recvTime = (pSyncIndexMgr->recvTimeArr)[i]; +int64_t syncIndexMgrGetRecvTime(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId) { + for (int i = 0; i < pIndexMgr->replicaNum; ++i) { + if (syncUtilSameId(&((*(pIndexMgr->replicas))[i]), pRaftId)) { + int64_t recvTime = (pIndexMgr->recvTimeArr)[i]; return recvTime; } } @@ -161,10 +159,10 @@ int64_t syncIndexMgrGetRecvTime(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRa return -1; } -void syncIndexMgrSetTerm(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId, SyncTerm term) { - for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) { - if (syncUtilSameId(&((*(pSyncIndexMgr->replicas))[i]), pRaftId)) { - (pSyncIndexMgr->privateTerm)[i] = term; +void syncIndexMgrSetTerm(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId, SyncTerm term) { + for (int i = 0; i < pIndexMgr->replicaNum; ++i) { + if (syncUtilSameId(&((*(pIndexMgr->replicas))[i]), pRaftId)) { + (pIndexMgr->privateTerm)[i] = term; return; } } @@ -174,13 +172,13 @@ void syncIndexMgrSetTerm(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId, S char host[128]; uint16_t port; syncUtilU642Addr(pRaftId->addr, host, sizeof(host), &port); - sError("vgId:%d, index mgr set for %s:%d, term:%" PRIu64 " error", pSyncIndexMgr->pSyncNode->vgId, host, port, term); + sError("vgId:%d, index mgr set for %s:%d, term:%" PRIu64 " error", pIndexMgr->pNode->vgId, host, port, term); } -SyncTerm syncIndexMgrGetTerm(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId) { - for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) { - if (syncUtilSameId(&((*(pSyncIndexMgr->replicas))[i]), pRaftId)) { - SyncTerm term = (pSyncIndexMgr->privateTerm)[i]; +SyncTerm syncIndexMgrGetTerm(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId) { + for (int i = 0; i < pIndexMgr->replicaNum; ++i) { + if (syncUtilSameId(&((*(pIndexMgr->replicas))[i]), pRaftId)) { + SyncTerm term = (pIndexMgr->privateTerm)[i]; return term; } } From 4b7ab0a8392a77d558843588deb8befd8d469e1e Mon Sep 17 00:00:00 2001 From: Shengliang Guan Date: Thu, 29 Dec 2022 11:00:33 +0800 Subject: [PATCH 2/3] refact: adjust log --- source/libs/sync/src/syncIndexMgr.c | 73 +++++++++++++++-------------- 1 file changed, 37 insertions(+), 36 deletions(-) diff --git a/source/libs/sync/src/syncIndexMgr.c b/source/libs/sync/src/syncIndexMgr.c index 8ea5d782b7..0950ab7372 100644 --- a/source/libs/sync/src/syncIndexMgr.c +++ b/source/libs/sync/src/syncIndexMgr.c @@ -24,7 +24,7 @@ SSyncIndexMgr *syncIndexMgrCreate(SSyncNode *pNode) { return NULL; } - pIndexMgr->replicas = &(pNode->replicasId); + pIndexMgr->replicas = &pNode->replicasId; pIndexMgr->replicaNum = pNode->replicaNum; pIndexMgr->pNode = pNode; syncIndexMgrClear(pIndexMgr); @@ -33,7 +33,7 @@ SSyncIndexMgr *syncIndexMgrCreate(SSyncNode *pNode) { } void syncIndexMgrUpdate(SSyncIndexMgr *pIndexMgr, SSyncNode *pNode) { - pIndexMgr->replicas = &(pNode->replicasId); + pIndexMgr->replicas = &pNode->replicasId; pIndexMgr->replicaNum = pNode->replicaNum; pIndexMgr->pNode = pNode; syncIndexMgrClear(pIndexMgr); @@ -49,18 +49,11 @@ void syncIndexMgrClear(SSyncIndexMgr *pIndexMgr) { memset(pIndexMgr->index, 0, sizeof(pIndexMgr->index)); memset(pIndexMgr->privateTerm, 0, sizeof(pIndexMgr->privateTerm)); - // int64_t timeNow = taosGetMonotonicMs(); int64_t timeNow = taosGetTimestampMs(); for (int i = 0; i < pIndexMgr->replicaNum; ++i) { pIndexMgr->startTimeArr[i] = 0; pIndexMgr->recvTimeArr[i] = timeNow; } - - /* - for (int i = 0; i < pIndexMgr->replicaNum; ++i) { - pIndexMgr->index[i] = 0; - } - */ } void syncIndexMgrSetIndex(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId, SyncIndex index) { @@ -71,29 +64,27 @@ void syncIndexMgrSetIndex(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId, Sync } } - // maybe config change - // ASSERT(0); + char host[128]; + uint16_t port; + syncUtilU642Addr(pRaftId->addr, host, sizeof(host), &port); + sError("vgId:%d, indexmgr set index:%" PRId64 " for %s:%d failed", pIndexMgr->pNode->vgId, index, host, port); +} + +SSyncLogReplMgr *syncNodeGetLogReplMgr(SSyncNode *pNode, SRaftId *pRaftId) { + for (int i = 0; i < pNode->replicaNum; i++) { + if (syncUtilSameId(&pNode->replicasId[i], pRaftId)) { + return pNode->logReplMgrs[i]; + } + } char host[128]; uint16_t port; syncUtilU642Addr(pRaftId->addr, host, sizeof(host), &port); - sError("vgId:%d, index mgr set for %s:%d, index:%" PRId64 " error", pIndexMgr->pNode->vgId, host, port, index); -} - -SSyncLogReplMgr *syncNodeGetLogReplMgr(SSyncNode *pNode, SRaftId *pDestId) { - for (int i = 0; i < pNode->replicaNum; i++) { - if (syncUtilSameId(&(pNode->replicasId[i]), pDestId)) { - return pNode->logReplMgrs[i]; - } - } + sError("vgId:%d, indexmgr get replmgr from %s:%d failed", pNode->vgId, host, port); return NULL; } SyncIndex syncIndexMgrGetIndex(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId) { - if (pIndexMgr == NULL) { - return SYNC_INDEX_INVALID; - } - for (int i = 0; i < pIndexMgr->replicaNum; ++i) { if (syncUtilSameId(&((*(pIndexMgr->replicas))[i]), pRaftId)) { SyncIndex idx = (pIndexMgr->index)[i]; @@ -101,6 +92,10 @@ SyncIndex syncIndexMgrGetIndex(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId) } } + char host[128]; + uint16_t port; + syncUtilU642Addr(pRaftId->addr, host, sizeof(host), &port); + sError("vgId:%d, indexmgr get index from %s:%d failed", pIndexMgr->pNode->vgId, host, port); return SYNC_INDEX_INVALID; } @@ -112,13 +107,11 @@ void syncIndexMgrSetStartTime(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId, } } - // maybe config change - // ASSERT(0); char host[128]; uint16_t port; syncUtilU642Addr(pRaftId->addr, host, sizeof(host), &port); - sError("vgId:%d, index mgr set for %s:%d, start-time:%" PRId64 " error", pIndexMgr->pNode->vgId, host, port, - startTime); + sError("vgId:%d, indexmgr set start-time:%" PRId64 " for %s:%d failed", pIndexMgr->pNode->vgId, startTime, host, + port); } int64_t syncIndexMgrGetStartTime(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId) { @@ -128,7 +121,11 @@ int64_t syncIndexMgrGetStartTime(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftI return startTime; } } - ASSERT(0); + + char host[128]; + uint16_t port; + syncUtilU642Addr(pRaftId->addr, host, sizeof(host), &port); + sError("vgId:%d, indexmgr get start-time from %s:%d failed", pIndexMgr->pNode->vgId, host, port); return -1; } @@ -140,12 +137,10 @@ void syncIndexMgrSetRecvTime(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId, i } } - // maybe config change - // ASSERT(0); char host[128]; uint16_t port; syncUtilU642Addr(pRaftId->addr, host, sizeof(host), &port); - sError("vgId:%d, index mgr set for %s:%d, recv-time:%" PRId64 " error", pIndexMgr->pNode->vgId, host, port, recvTime); + sError("vgId:%d, indexmgr set recv-time:%" PRId64 " for %s:%d failed", pIndexMgr->pNode->vgId, recvTime, host, port); } int64_t syncIndexMgrGetRecvTime(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId) { @@ -156,6 +151,10 @@ int64_t syncIndexMgrGetRecvTime(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId } } + char host[128]; + uint16_t port; + syncUtilU642Addr(pRaftId->addr, host, sizeof(host), &port); + sError("vgId:%d, indexmgr get recv-time from %s:%d failed", pIndexMgr->pNode->vgId, host, port); return -1; } @@ -167,12 +166,10 @@ void syncIndexMgrSetTerm(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId, SyncT } } - // maybe config change - // ASSERT(0); char host[128]; uint16_t port; syncUtilU642Addr(pRaftId->addr, host, sizeof(host), &port); - sError("vgId:%d, index mgr set for %s:%d, term:%" PRIu64 " error", pIndexMgr->pNode->vgId, host, port, term); + sError("vgId:%d, indexmgr set term:%" PRId64 " for %s:%d failed", pIndexMgr->pNode->vgId, term, host, port); } SyncTerm syncIndexMgrGetTerm(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId) { @@ -182,6 +179,10 @@ SyncTerm syncIndexMgrGetTerm(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId) { return term; } } - ASSERT(0); + + char host[128]; + uint16_t port; + syncUtilU642Addr(pRaftId->addr, host, sizeof(host), &port); + sError("vgId:%d, indexmgr get term from %s:%d failed", pIndexMgr->pNode->vgId, host, port); return -1; } From b05c65cd47f2d3f5c9e51a988e816aefadae2878 Mon Sep 17 00:00:00 2001 From: Shengliang Guan Date: Thu, 29 Dec 2022 11:17:36 +0800 Subject: [PATCH 3/3] refact: rename variables --- source/libs/sync/inc/syncVoteMgr.h | 12 +++---- source/libs/sync/src/syncRespMgr.c | 16 ++++----- source/libs/sync/src/syncSnapshot.c | 2 +- source/libs/sync/src/syncVoteMgr.c | 54 ++++++++++++++--------------- 4 files changed, 41 insertions(+), 43 deletions(-) diff --git a/source/libs/sync/inc/syncVoteMgr.h b/source/libs/sync/inc/syncVoteMgr.h index 066a4dd76f..6e63a2b396 100644 --- a/source/libs/sync/inc/syncVoteMgr.h +++ b/source/libs/sync/inc/syncVoteMgr.h @@ -30,12 +30,12 @@ typedef struct SVotesGranted { SyncTerm term; int32_t quorum; bool toLeader; - SSyncNode *pSyncNode; + SSyncNode *pNode; } SVotesGranted; -SVotesGranted *voteGrantedCreate(SSyncNode *pSyncNode); +SVotesGranted *voteGrantedCreate(SSyncNode *pNode); void voteGrantedDestroy(SVotesGranted *pVotesGranted); -void voteGrantedUpdate(SVotesGranted *pVotesGranted, SSyncNode *pSyncNode); +void voteGrantedUpdate(SVotesGranted *pVotesGranted, SSyncNode *pNode); bool voteGrantedMajority(SVotesGranted *pVotesGranted); void voteGrantedVote(SVotesGranted *pVotesGranted, SyncRequestVoteReply *pMsg); void voteGrantedReset(SVotesGranted *pVotesGranted, SyncTerm term); @@ -45,12 +45,12 @@ typedef struct SVotesRespond { bool isRespond[TSDB_MAX_REPLICA]; int32_t replicaNum; SyncTerm term; - SSyncNode *pSyncNode; + SSyncNode *pNode; } SVotesRespond; -SVotesRespond *votesRespondCreate(SSyncNode *pSyncNode); +SVotesRespond *votesRespondCreate(SSyncNode *pNode); void votesRespondDestory(SVotesRespond *pVotesRespond); -void votesRespondUpdate(SVotesRespond *pVotesRespond, SSyncNode *pSyncNode); +void votesRespondUpdate(SVotesRespond *pVotesRespond, SSyncNode *pNode); bool votesResponded(SVotesRespond *pVotesRespond, const SRaftId *pRaftId); void votesRespondAdd(SVotesRespond *pVotesRespond, const SyncRequestVoteReply *pMsg); void votesRespondReset(SVotesRespond *pVotesRespond, SyncTerm term); diff --git a/source/libs/sync/src/syncRespMgr.c b/source/libs/sync/src/syncRespMgr.c index 503c56b1e7..b55aae4c76 100644 --- a/source/libs/sync/src/syncRespMgr.c +++ b/source/libs/sync/src/syncRespMgr.c @@ -118,12 +118,12 @@ static void syncRespCleanByTTL(SSyncRespMgr *pObj, int64_t ttl, bool rsp) { SRespStub *pStub = (SRespStub *)taosHashIterate(pObj->pRespHash, NULL); int cnt = 0; int sum = 0; - SSyncNode *pSyncNode = pObj->data; + SSyncNode *pNode = pObj->data; SArray *delIndexArray = taosArrayInit(4, sizeof(uint64_t)); if (delIndexArray == NULL) return; - sDebug("vgId:%d, resp manager begin clean by ttl", pSyncNode->vgId); + sDebug("vgId:%d, resp manager begin clean by ttl", pNode->vgId); while (pStub) { size_t len; void *key = taosHashGetKey(pStub, &len); @@ -140,20 +140,18 @@ static void syncRespCleanByTTL(SSyncRespMgr *pObj, int64_t ttl, bool rsp) { .lastConfigIndex = SYNC_INDEX_INVALID, .isWeak = false, .code = TSDB_CODE_SYN_TIMEOUT, - .state = pSyncNode->state, + .state = pNode->state, .seqNum = *pSeqNum, .term = SYNC_TERM_INVALID, - .currentTerm = pSyncNode->pRaftStore->currentTerm, + .currentTerm = pNode->pRaftStore->currentTerm, .flag = 0, }; pStub->rpcMsg.pCont = NULL; pStub->rpcMsg.contLen = 0; - // TODO: and make rpcMsg body, call commit cb - // pSyncNode->pFsm->FpCommitCb(pSyncNode->pFsm, &pStub->rpcMsg, cbMeta); SRpcMsg rpcMsg = {.info = pStub->rpcMsg.info, .code = TSDB_CODE_SYN_TIMEOUT}; - sInfo("vgId:%d, message handle:%p expired, type:%s ahandle:%p", pSyncNode->vgId, rpcMsg.info.handle, + sInfo("vgId:%d, message handle:%p expired, type:%s ahandle:%p", pNode->vgId, rpcMsg.info.handle, TMSG_INFO(pStub->rpcMsg.msgType), rpcMsg.info.ahandle); rpcSendResponse(&rpcMsg); } @@ -162,12 +160,12 @@ static void syncRespCleanByTTL(SSyncRespMgr *pObj, int64_t ttl, bool rsp) { } int32_t arraySize = taosArrayGetSize(delIndexArray); - sDebug("vgId:%d, resp manager end clean by ttl, sum:%d, cnt:%d, array-size:%d", pSyncNode->vgId, sum, cnt, arraySize); + sDebug("vgId:%d, resp manager end clean by ttl, sum:%d, cnt:%d, array-size:%d", pNode->vgId, sum, cnt, arraySize); for (int32_t i = 0; i < arraySize; ++i) { uint64_t *pSeqNum = taosArrayGet(delIndexArray, i); taosHashRemove(pObj->pRespHash, pSeqNum, sizeof(uint64_t)); - sDebug("vgId:%d, resp manager clean by ttl, seq:%" PRId64, pSyncNode->vgId, *pSeqNum); + sDebug("vgId:%d, resp manager clean by ttl, seq:%" PRId64, pNode->vgId, *pSeqNum); } taosArrayDestroy(delIndexArray); } diff --git a/source/libs/sync/src/syncSnapshot.c b/source/libs/sync/src/syncSnapshot.c index f2da2fb2ce..b68b735f46 100644 --- a/source/libs/sync/src/syncSnapshot.c +++ b/source/libs/sync/src/syncSnapshot.c @@ -406,7 +406,7 @@ void snapshotReceiverStart(SSyncSnapshotReceiver *pReceiver, SyncSnapshotSend *p } // just set start = false -// FpSnapshotStopWrite should not be called, assert writer == NULL +// FpSnapshotStopWrite should not be called void snapshotReceiverStop(SSyncSnapshotReceiver *pReceiver) { sRInfo(pReceiver, "snapshot receiver stop, not apply, writer:%p", pReceiver->pWriter); diff --git a/source/libs/sync/src/syncVoteMgr.c b/source/libs/sync/src/syncVoteMgr.c index 8a0a35ce33..6bd9625276 100644 --- a/source/libs/sync/src/syncVoteMgr.c +++ b/source/libs/sync/src/syncVoteMgr.c @@ -23,21 +23,21 @@ static void voteGrantedClearVotes(SVotesGranted *pVotesGranted) { pVotesGranted->votes = 0; } -SVotesGranted *voteGrantedCreate(SSyncNode *pSyncNode) { +SVotesGranted *voteGrantedCreate(SSyncNode *pNode) { SVotesGranted *pVotesGranted = taosMemoryCalloc(1, sizeof(SVotesGranted)); if (pVotesGranted == NULL) { terrno = TSDB_CODE_OUT_OF_MEMORY; return NULL; } - pVotesGranted->replicas = &(pSyncNode->replicasId); - pVotesGranted->replicaNum = pSyncNode->replicaNum; + pVotesGranted->replicas = &pNode->replicasId; + pVotesGranted->replicaNum = pNode->replicaNum; voteGrantedClearVotes(pVotesGranted); pVotesGranted->term = 0; - pVotesGranted->quorum = pSyncNode->quorum; + pVotesGranted->quorum = pNode->quorum; pVotesGranted->toLeader = false; - pVotesGranted->pSyncNode = pSyncNode; + pVotesGranted->pNode = pNode; return pVotesGranted; } @@ -48,33 +48,33 @@ void voteGrantedDestroy(SVotesGranted *pVotesGranted) { } } -void voteGrantedUpdate(SVotesGranted *pVotesGranted, SSyncNode *pSyncNode) { - pVotesGranted->replicas = &(pSyncNode->replicasId); - pVotesGranted->replicaNum = pSyncNode->replicaNum; +void voteGrantedUpdate(SVotesGranted *pVotesGranted, SSyncNode *pNode) { + pVotesGranted->replicas = &pNode->replicasId; + pVotesGranted->replicaNum = pNode->replicaNum; voteGrantedClearVotes(pVotesGranted); pVotesGranted->term = 0; - pVotesGranted->quorum = pSyncNode->quorum; + pVotesGranted->quorum = pNode->quorum; pVotesGranted->toLeader = false; - pVotesGranted->pSyncNode = pSyncNode; + pVotesGranted->pNode = pNode; } bool voteGrantedMajority(SVotesGranted *pVotesGranted) { return pVotesGranted->votes >= pVotesGranted->quorum; } void voteGrantedVote(SVotesGranted *pVotesGranted, SyncRequestVoteReply *pMsg) { if (!pMsg->voteGranted) { - sNFatal(pVotesGranted->pSyncNode, "vote granted should be true"); + sNFatal(pVotesGranted->pNode, "vote granted should be true"); return; } if (pMsg->term != pVotesGranted->term) { - sNTrace(pVotesGranted->pSyncNode, "vote grant term:%" PRId64 " not matched with msg term:%" PRId64, - pVotesGranted->term, pMsg->term); + sNTrace(pVotesGranted->pNode, "vote grant term:%" PRId64 " not matched with msg term:%" PRId64, pVotesGranted->term, + pMsg->term); return; } - if (!syncUtilSameId(&pVotesGranted->pSyncNode->myRaftId, &pMsg->destId)) { - sNFatal(pVotesGranted->pSyncNode, "vote granted raftId not matched with msg"); + if (!syncUtilSameId(&pVotesGranted->pNode->myRaftId, &pMsg->destId)) { + sNFatal(pVotesGranted->pNode, "vote granted raftId not matched with msg"); return; } @@ -86,7 +86,7 @@ void voteGrantedVote(SVotesGranted *pVotesGranted, SyncRequestVoteReply *pMsg) { } } if ((j == -1) || !(j >= 0 && j < pVotesGranted->replicaNum)) { - sNFatal(pVotesGranted->pSyncNode, "invalid msg srcId, index:%d", j); + sNFatal(pVotesGranted->pNode, "invalid msg srcId, index:%d", j); return; } @@ -96,7 +96,7 @@ void voteGrantedVote(SVotesGranted *pVotesGranted, SyncRequestVoteReply *pMsg) { } if (pVotesGranted->votes > pVotesGranted->replicaNum) { - sNFatal(pVotesGranted->pSyncNode, "votes:%d not matched with replicaNum:%d", pVotesGranted->votes, + sNFatal(pVotesGranted->pNode, "votes:%d not matched with replicaNum:%d", pVotesGranted->votes, pVotesGranted->replicaNum); return; } @@ -108,17 +108,17 @@ void voteGrantedReset(SVotesGranted *pVotesGranted, SyncTerm term) { pVotesGranted->toLeader = false; } -SVotesRespond *votesRespondCreate(SSyncNode *pSyncNode) { +SVotesRespond *votesRespondCreate(SSyncNode *pNode) { SVotesRespond *pVotesRespond = taosMemoryCalloc(1, sizeof(SVotesRespond)); if (pVotesRespond == NULL) { terrno = TSDB_CODE_OUT_OF_MEMORY; return NULL; } - pVotesRespond->replicas = &(pSyncNode->replicasId); - pVotesRespond->replicaNum = pSyncNode->replicaNum; + pVotesRespond->replicas = &pNode->replicasId; + pVotesRespond->replicaNum = pNode->replicaNum; pVotesRespond->term = 0; - pVotesRespond->pSyncNode = pSyncNode; + pVotesRespond->pNode = pNode; return pVotesRespond; } @@ -129,11 +129,11 @@ void votesRespondDestory(SVotesRespond *pVotesRespond) { } } -void votesRespondUpdate(SVotesRespond *pVotesRespond, SSyncNode *pSyncNode) { - pVotesRespond->replicas = &(pSyncNode->replicasId); - pVotesRespond->replicaNum = pSyncNode->replicaNum; +void votesRespondUpdate(SVotesRespond *pVotesRespond, SSyncNode *pNode) { + pVotesRespond->replicas = &pNode->replicasId; + pVotesRespond->replicaNum = pNode->replicaNum; pVotesRespond->term = 0; - pVotesRespond->pSyncNode = pSyncNode; + pVotesRespond->pNode = pNode; } bool votesResponded(SVotesRespond *pVotesRespond, const SRaftId *pRaftId) { @@ -149,7 +149,7 @@ bool votesResponded(SVotesRespond *pVotesRespond, const SRaftId *pRaftId) { void votesRespondAdd(SVotesRespond *pVotesRespond, const SyncRequestVoteReply *pMsg) { if (pVotesRespond->term != pMsg->term) { - sNTrace(pVotesRespond->pSyncNode, "vote respond add error"); + sNTrace(pVotesRespond->pNode, "vote respond add error"); return; } @@ -160,7 +160,7 @@ void votesRespondAdd(SVotesRespond *pVotesRespond, const SyncRequestVoteReply *p } } - sNFatal(pVotesRespond->pSyncNode, "votes respond not found"); + sNFatal(pVotesRespond->pNode, "votes respond not found"); } void votesRespondReset(SVotesRespond *pVotesRespond, SyncTerm term) {