From f5db109e61c44f92e1d2e75edceac99057be579a Mon Sep 17 00:00:00 2001 From: Minghao Li Date: Fri, 17 Jun 2022 11:37:10 +0800 Subject: [PATCH 1/4] fix(sync): get last max snapshot index --- include/libs/sync/sync.h | 1 + source/dnode/mnode/impl/src/mndSync.c | 8 +++-- source/dnode/mnode/impl/src/mndTrans.c | 8 +++-- source/libs/sync/inc/syncRaftCfg.h | 6 ++++ source/libs/sync/src/syncMain.c | 32 +++++++++++++++--- source/libs/sync/src/syncRaftCfg.c | 41 ++++++++++++++++++----- source/libs/sync/src/syncRaftLog.c | 26 +++++++------- source/libs/sync/src/syncSnapshot.c | 10 +++--- source/libs/sync/test/syncRaftCfgTest.cpp | 19 ++++++++++- 9 files changed, 114 insertions(+), 37 deletions(-) diff --git a/include/libs/sync/sync.h b/include/libs/sync/sync.h index fc9d419b1f..f49030466e 100644 --- a/include/libs/sync/sync.h +++ b/include/libs/sync/sync.h @@ -192,6 +192,7 @@ bool syncEnvIsStart(); const char* syncStr(ESyncState state); bool syncIsRestoreFinish(int64_t rid); int32_t syncGetSnapshotMeta(int64_t rid, struct SSnapshotMeta* sMeta); +int32_t syncGetSnapshotMetaByIndex(int64_t rid, SyncIndex snapshotIndex, struct SSnapshotMeta* sMeta); int32_t syncReconfig(int64_t rid, const SSyncCfg* pNewCfg); diff --git a/source/dnode/mnode/impl/src/mndSync.c b/source/dnode/mnode/impl/src/mndSync.c index 3a243e36d3..388a9d82c4 100644 --- a/source/dnode/mnode/impl/src/mndSync.c +++ b/source/dnode/mnode/impl/src/mndSync.c @@ -69,7 +69,8 @@ void mndSyncCommitMsg(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cbM if (cbMeta.index - sdbGetApplyIndex(pMnode->pSdb) > 100) { SSnapshotMeta sMeta = {0}; - if (syncGetSnapshotMeta(pMnode->syncMgmt.sync, &sMeta) == 0) { + // if (syncGetSnapshotMeta(pMnode->syncMgmt.sync, &sMeta) == 0) { + if (syncGetSnapshotMetaByIndex(pMnode->syncMgmt.sync, cbMeta.index, &sMeta) == 0) { sdbSetCurConfig(pMnode->pSdb, sMeta.lastConfigIndex); } sdbWriteFile(pMnode->pSdb); @@ -89,7 +90,10 @@ void mndRestoreFinish(struct SSyncFSM *pFsm) { SMnode *pMnode = pFsm->data; SSnapshotMeta sMeta = {0}; - if (syncGetSnapshotMeta(pMnode->syncMgmt.sync, &sMeta) == 0) { + // if (syncGetSnapshotMeta(pMnode->syncMgmt.sync, &sMeta) == 0) { + + SyncIndex snapshotIndex = sdbGetApplyIndex(pMnode->pSdb); + if (syncGetSnapshotMetaByIndex(pMnode->syncMgmt.sync, snapshotIndex, &sMeta) == 0) { sdbSetCurConfig(pMnode->pSdb, sMeta.lastConfigIndex); } diff --git a/source/dnode/mnode/impl/src/mndTrans.c b/source/dnode/mnode/impl/src/mndTrans.c index 1631c9825b..a50d8da37e 100644 --- a/source/dnode/mnode/impl/src/mndTrans.c +++ b/source/dnode/mnode/impl/src/mndTrans.c @@ -22,8 +22,8 @@ #include "mndSync.h" #include "mndUser.h" -#define TRANS_VER_NUMBER 1 -#define TRANS_ARRAY_SIZE 8 +#define TRANS_VER_NUMBER 1 +#define TRANS_ARRAY_SIZE 8 #define TRANS_RESERVE_SIZE 64 static SSdbRaw *mndTransActionEncode(STrans *pTrans); @@ -1426,7 +1426,9 @@ void mndTransPullup(SMnode *pMnode) { } SSnapshotMeta sMeta = {0}; - if (syncGetSnapshotMeta(pMnode->syncMgmt.sync, &sMeta) == 0) { + // if (syncGetSnapshotMeta(pMnode->syncMgmt.sync, &sMeta) == 0) { + SyncIndex snapshotIndex = sdbGetApplyIndex(pMnode->pSdb); + if (syncGetSnapshotMetaByIndex(pMnode->syncMgmt.sync, snapshotIndex, &sMeta) == 0) { sdbSetCurConfig(pMnode->pSdb, sMeta.lastConfigIndex); } sdbWriteFile(pMnode->pSdb); diff --git a/source/libs/sync/inc/syncRaftCfg.h b/source/libs/sync/inc/syncRaftCfg.h index e72e1e7be7..cd64402738 100644 --- a/source/libs/sync/inc/syncRaftCfg.h +++ b/source/libs/sync/inc/syncRaftCfg.h @@ -29,6 +29,8 @@ extern "C" { #define CONFIG_FILE_LEN 1024 +#define MAX_CONFIG_INDEX_COUNT 512 + typedef struct SRaftCfg { SSyncCfg cfg; TdFilePtr pFile; @@ -36,6 +38,10 @@ typedef struct SRaftCfg { int8_t isStandBy; int8_t snapshotEnable; SyncIndex lastConfigIndex; + + SyncIndex configIndexArr[MAX_CONFIG_INDEX_COUNT]; + int32_t configIndexCount; + } SRaftCfg; SRaftCfg *raftCfgOpen(const char *path); diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c index 00786e5015..278288eb1e 100644 --- a/source/libs/sync/src/syncMain.c +++ b/source/libs/sync/src/syncMain.c @@ -418,6 +418,29 @@ int32_t syncGetSnapshotMeta(int64_t rid, struct SSnapshotMeta* sMeta) { return 0; } +int32_t syncGetSnapshotMetaByIndex(int64_t rid, SyncIndex snapshotIndex, struct SSnapshotMeta* sMeta) { + SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid); + if (pSyncNode == NULL) { + return -1; + } + assert(rid == pSyncNode->rid); + + ASSERT(pSyncNode->pRaftCfg->configIndexCount >= 1); + SyncIndex lastIndex = (pSyncNode->pRaftCfg->configIndexArr)[0]; + + for (int i = 0; i < pSyncNode->pRaftCfg->configIndexCount; ++i) { + if ((pSyncNode->pRaftCfg->configIndexArr)[i] > lastIndex && + (pSyncNode->pRaftCfg->configIndexArr)[i] <= snapshotIndex) { + lastIndex = (pSyncNode->pRaftCfg->configIndexArr)[i]; + } + } + sMeta->lastConfigIndex = lastIndex; + sTrace("sync get snapshot meta by index:%ld lastConfigIndex:%ld", snapshotIndex, sMeta->lastConfigIndex); + + taosReleaseRef(tsNodeRefId, pSyncNode->rid); + return 0; +} + const char* syncGetMyRoleStr(int64_t rid) { const char* s = syncUtilState2String(syncGetMyRole(rid)); return s; @@ -2195,8 +2218,8 @@ static int32_t syncNodeConfigChange(SSyncNode* ths, SRpcMsg* pRpcMsg, SSyncRaftE char* newStr = syncCfg2Str(&newSyncCfg); syncUtilJson2Line(oldStr); syncUtilJson2Line(newStr); - snprintf(tmpbuf, sizeof(tmpbuf), "config change from %d to %d, %s --> %s", oldSyncCfg.replicaNum, - newSyncCfg.replicaNum, oldStr, newStr); + snprintf(tmpbuf, sizeof(tmpbuf), "config change from %d to %d, index:%ld, %s --> %s", oldSyncCfg.replicaNum, + newSyncCfg.replicaNum, pEntry->index, oldStr, newStr); taosMemoryFree(oldStr); taosMemoryFree(newStr); @@ -2212,8 +2235,8 @@ static int32_t syncNodeConfigChange(SSyncNode* ths, SRpcMsg* pRpcMsg, SSyncRaftE char* newStr = syncCfg2Str(&newSyncCfg); syncUtilJson2Line(oldStr); syncUtilJson2Line(newStr); - snprintf(tmpbuf, sizeof(tmpbuf), "config change2 from %d to %d, %s --> %s", oldSyncCfg.replicaNum, - newSyncCfg.replicaNum, oldStr, newStr); + snprintf(tmpbuf, sizeof(tmpbuf), "config change2 from %d to %d, index:%ld, %s --> %s", oldSyncCfg.replicaNum, + newSyncCfg.replicaNum, pEntry->index, oldStr, newStr); taosMemoryFree(oldStr); taosMemoryFree(newStr); @@ -2295,6 +2318,7 @@ int32_t syncNodeCommit(SSyncNode* ths, SyncIndex beginIndex, SyncIndex endIndex, } // restore finish + // if only snapshot, a noop entry will be append, so syncLogLastIndex is always ok if (pEntry->index == ths->pLogStore->syncLogLastIndex(ths->pLogStore)) { if (ths->restoreFinish == false) { if (ths->pFsm->FpRestoreFinishCb != NULL) { diff --git a/source/libs/sync/src/syncRaftCfg.c b/source/libs/sync/src/syncRaftCfg.c index 45e00aca2c..2d51f1f6f0 100644 --- a/source/libs/sync/src/syncRaftCfg.c +++ b/source/libs/sync/src/syncRaftCfg.c @@ -85,16 +85,11 @@ cJSON *syncCfg2Json(SSyncCfg *pSyncCfg) { } return pRoot; - /* - cJSON *pJson = cJSON_CreateObject(); - cJSON_AddItemToObject(pJson, "SSyncCfg", pRoot); - return pJson; - */ } char *syncCfg2Str(SSyncCfg *pSyncCfg) { cJSON *pJson = syncCfg2Json(pSyncCfg); - char * serialized = cJSON_Print(pJson); + char *serialized = cJSON_Print(pJson); cJSON_Delete(pJson); return serialized; } @@ -154,6 +149,16 @@ cJSON *raftCfg2Json(SRaftCfg *pRaftCfg) { snprintf(buf64, sizeof(buf64), "%ld", pRaftCfg->lastConfigIndex); cJSON_AddStringToObject(pRoot, "lastConfigIndex", buf64); + cJSON_AddNumberToObject(pRoot, "configIndexCount", pRaftCfg->configIndexCount); + cJSON *pIndexArr = cJSON_CreateArray(); + cJSON_AddItemToObject(pRoot, "configIndexArr", pIndexArr); + for (int i = 0; i < pRaftCfg->configIndexCount; ++i) { + snprintf(buf64, sizeof(buf64), "%ld", (pRaftCfg->configIndexArr)[i]); + cJSON *pIndexObj = cJSON_CreateObject(); + cJSON_AddStringToObject(pIndexObj, "index", buf64); + cJSON_AddItemToArray(pIndexArr, pIndexObj); + } + cJSON *pJson = cJSON_CreateObject(); cJSON_AddItemToObject(pJson, "RaftCfg", pRoot); return pJson; @@ -161,7 +166,7 @@ cJSON *raftCfg2Json(SRaftCfg *pRaftCfg) { char *raftCfg2Str(SRaftCfg *pRaftCfg) { cJSON *pJson = raftCfg2Json(pRaftCfg); - char * serialized = cJSON_Print(pJson); + char *serialized = cJSON_Print(pJson); cJSON_Delete(pJson); return serialized; } @@ -177,6 +182,9 @@ int32_t raftCfgCreateFile(SSyncCfg *pCfg, SRaftCfgMeta meta, const char *path) { raftCfg.isStandBy = meta.isStandBy; raftCfg.snapshotEnable = meta.snapshotEnable; raftCfg.lastConfigIndex = meta.lastConfigIndex; + raftCfg.configIndexCount = 1; + memset(raftCfg.configIndexArr, 0, sizeof(raftCfg.configIndexArr)); + raftCfg.configIndexArr[0] = -1; char *s = raftCfg2Str(&raftCfg); char buf[CONFIG_FILE_LEN] = {0}; @@ -207,7 +215,24 @@ int32_t raftCfgFromJson(const cJSON *pRoot, SRaftCfg *pRaftCfg) { cJSON *pJsonLastConfigIndex = cJSON_GetObjectItem(pJson, "lastConfigIndex"); pRaftCfg->lastConfigIndex = atoll(cJSON_GetStringValue(pJsonLastConfigIndex)); - cJSON * pJsonSyncCfg = cJSON_GetObjectItem(pJson, "SSyncCfg"); + cJSON *pJsonConfigIndexCount = cJSON_GetObjectItem(pJson, "configIndexCount"); + pRaftCfg->configIndexCount = cJSON_GetNumberValue(pJsonConfigIndexCount); + + cJSON *pIndexArr = cJSON_GetObjectItem(pJson, "configIndexArr"); + int arraySize = cJSON_GetArraySize(pIndexArr); + assert(arraySize == pRaftCfg->configIndexCount); + + memset(pRaftCfg->configIndexArr, 0, sizeof(pRaftCfg->configIndexArr)); + for (int i = 0; i < arraySize; ++i) { + cJSON *pIndexObj = cJSON_GetArrayItem(pIndexArr, i); + assert(pIndexObj != NULL); + + cJSON *pIndex = cJSON_GetObjectItem(pIndexObj, "index"); + assert(cJSON_IsString(pIndex)); + (pRaftCfg->configIndexArr)[i] = atoll(pIndex->valuestring); + } + + cJSON *pJsonSyncCfg = cJSON_GetObjectItem(pJson, "SSyncCfg"); int32_t code = syncCfgFromJson(pJsonSyncCfg, &(pRaftCfg->cfg)); ASSERT(code == 0); diff --git a/source/libs/sync/src/syncRaftLog.c b/source/libs/sync/src/syncRaftLog.c index 7c6d97a96e..87dfef5fdd 100644 --- a/source/libs/sync/src/syncRaftLog.c +++ b/source/libs/sync/src/syncRaftLog.c @@ -410,20 +410,18 @@ SyncTerm logStoreLastTerm(SSyncLogStore* pLogStore) { } int32_t logStoreUpdateCommitIndex(SSyncLogStore* pLogStore, SyncIndex index) { - /* - SSyncLogStoreData* pData = pLogStore->data; - SWal* pWal = pData->pWal; - // assert(walCommit(pWal, index) == 0); - int32_t code = walCommit(pWal, index); - if (code != 0) { - int32_t err = terrno; - const char* errStr = tstrerror(err); - int32_t linuxErr = errno; - const char* linuxErrMsg = strerror(errno); - sError("walCommit error, err:%d %X, msg:%s, linuxErr:%d, linuxErrMsg:%s", err, err, errStr, linuxErr, - linuxErrMsg); ASSERT(0); - } - */ + SSyncLogStoreData* pData = pLogStore->data; + SWal* pWal = pData->pWal; + // assert(walCommit(pWal, index) == 0); + int32_t code = walCommit(pWal, index); + if (code != 0) { + int32_t err = terrno; + const char* errStr = tstrerror(err); + int32_t linuxErr = errno; + const char* linuxErrMsg = strerror(errno); + sError("walCommit error, err:%d %X, msg:%s, linuxErr:%d, linuxErrMsg:%s", err, err, errStr, linuxErr, linuxErrMsg); + ASSERT(0); + } return 0; } diff --git a/source/libs/sync/src/syncSnapshot.c b/source/libs/sync/src/syncSnapshot.c index ef2011cc26..4973d36295 100644 --- a/source/libs/sync/src/syncSnapshot.c +++ b/source/libs/sync/src/syncSnapshot.c @@ -421,7 +421,7 @@ cJSON *snapshotSender2Json(SSyncSnapshotSender *pSender) { char *snapshotSender2Str(SSyncSnapshotSender *pSender) { cJSON *pJson = snapshotSender2Json(pSender); - char *serialized = cJSON_Print(pJson); + char * serialized = cJSON_Print(pJson); cJSON_Delete(pJson); return serialized; } @@ -542,7 +542,7 @@ cJSON *snapshotReceiver2Json(SSyncSnapshotReceiver *pReceiver) { cJSON_AddStringToObject(pFromId, "addr", u64buf); { uint64_t u64 = pReceiver->fromId.addr; - cJSON *pTmp = pFromId; + cJSON * pTmp = pFromId; char host[128] = {0}; uint16_t port; syncUtilU642Addr(u64, host, sizeof(host), &port); @@ -566,7 +566,7 @@ cJSON *snapshotReceiver2Json(SSyncSnapshotReceiver *pReceiver) { char *snapshotReceiver2Str(SSyncSnapshotReceiver *pReceiver) { cJSON *pJson = snapshotReceiver2Json(pReceiver); - char *serialized = cJSON_Print(pJson); + char * serialized = cJSON_Print(pJson); cJSON_Delete(pJson); return serialized; } @@ -671,8 +671,8 @@ int32_t syncNodeOnSnapshotSendCb(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) { char *newStr = syncCfg2Str(&newSyncCfg); syncUtilJson2Line(oldStr); syncUtilJson2Line(newStr); - snprintf(tmpbuf, sizeof(tmpbuf), "config change3 from %d to %d, %s --> %s", oldSyncCfg.replicaNum, - newSyncCfg.replicaNum, oldStr, newStr); + snprintf(tmpbuf, sizeof(tmpbuf), "config change3 from %d to %d, index:%ld, %s --> %s", + oldSyncCfg.replicaNum, newSyncCfg.replicaNum, pMsg->lastConfigIndex, oldStr, newStr); taosMemoryFree(oldStr); taosMemoryFree(newStr); diff --git a/source/libs/sync/test/syncRaftCfgTest.cpp b/source/libs/sync/test/syncRaftCfgTest.cpp index 8c6a704e2d..9460f0ad7a 100644 --- a/source/libs/sync/test/syncRaftCfgTest.cpp +++ b/source/libs/sync/test/syncRaftCfgTest.cpp @@ -27,6 +27,14 @@ SRaftCfg* createRaftCfg() { } pCfg->isStandBy = taosGetTimestampSec() % 100; + pCfg->configIndexCount = 5; + for (int i = 0; i < MAX_CONFIG_INDEX_COUNT; ++i) { + (pCfg->configIndexArr)[i] = -1; + } + for (int i = 0; i < pCfg->configIndexCount; ++i) { + (pCfg->configIndexArr)[i] = i * 100; + } + return pCfg; } @@ -100,6 +108,15 @@ void test5() { pCfg->isStandBy += 2; pCfg->snapshotEnable += 3; pCfg->lastConfigIndex += 1000; + + pCfg->configIndexCount = 5; + for (int i = 0; i < MAX_CONFIG_INDEX_COUNT; ++i) { + (pCfg->configIndexArr)[i] = -1; + } + for (int i = 0; i < pCfg->configIndexCount; ++i) { + (pCfg->configIndexArr)[i] = i * 100; + } + raftCfgPersist(pCfg); printf("%s update json file: %s myIndex->%d \n", (char*)__FUNCTION__, "./test3_raft_cfg.json", pCfg->cfg.myIndex); @@ -118,6 +135,6 @@ int main() { test3(); test4(); test5(); - + return 0; } From 28b32682246270336710717942cdda0f17d435b4 Mon Sep 17 00:00:00 2001 From: Shengliang Guan Date: Fri, 17 Jun 2022 11:40:18 +0800 Subject: [PATCH 2/4] fix(sync) add trace log --- source/libs/sync/src/syncMain.c | 8 ++++---- source/libs/sync/src/syncSnapshot.c | 4 ++-- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c index 343a0e5597..dbf7bc94e0 100644 --- a/source/libs/sync/src/syncMain.c +++ b/source/libs/sync/src/syncMain.c @@ -2197,8 +2197,8 @@ static int32_t syncNodeConfigChange(SSyncNode* ths, SRpcMsg* pRpcMsg, SSyncRaftE char* newStr = syncCfg2Str(&newSyncCfg); syncUtilJson2Line(oldStr); syncUtilJson2Line(newStr); - snprintf(tmpbuf, sizeof(tmpbuf), "config change from %d to %d, %s --> %s", oldSyncCfg.replicaNum, - newSyncCfg.replicaNum, oldStr, newStr); + snprintf(tmpbuf, sizeof(tmpbuf), "config change from %d to %d, index:%ld, %s --> %s", oldSyncCfg.replicaNum, + newSyncCfg.replicaNum, pEntry->index, oldStr, newStr); taosMemoryFree(oldStr); taosMemoryFree(newStr); @@ -2214,8 +2214,8 @@ static int32_t syncNodeConfigChange(SSyncNode* ths, SRpcMsg* pRpcMsg, SSyncRaftE char* newStr = syncCfg2Str(&newSyncCfg); syncUtilJson2Line(oldStr); syncUtilJson2Line(newStr); - snprintf(tmpbuf, sizeof(tmpbuf), "config change2 from %d to %d, %s --> %s", oldSyncCfg.replicaNum, - newSyncCfg.replicaNum, oldStr, newStr); + snprintf(tmpbuf, sizeof(tmpbuf), "config change2 from %d to %d, index:%ld, %s --> %s", oldSyncCfg.replicaNum, + newSyncCfg.replicaNum, pEntry->index, oldStr, newStr); taosMemoryFree(oldStr); taosMemoryFree(newStr); diff --git a/source/libs/sync/src/syncSnapshot.c b/source/libs/sync/src/syncSnapshot.c index ef2011cc26..c8d2866738 100644 --- a/source/libs/sync/src/syncSnapshot.c +++ b/source/libs/sync/src/syncSnapshot.c @@ -671,8 +671,8 @@ int32_t syncNodeOnSnapshotSendCb(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) { char *newStr = syncCfg2Str(&newSyncCfg); syncUtilJson2Line(oldStr); syncUtilJson2Line(newStr); - snprintf(tmpbuf, sizeof(tmpbuf), "config change3 from %d to %d, %s --> %s", oldSyncCfg.replicaNum, - newSyncCfg.replicaNum, oldStr, newStr); + snprintf(tmpbuf, sizeof(tmpbuf), "config change3 from %d to %d, index:%ld, %s --> %s", + oldSyncCfg.replicaNum, newSyncCfg.replicaNum, pMsg->lastConfigIndex, oldStr, newStr); taosMemoryFree(oldStr); taosMemoryFree(newStr); From 455a8da0adb94f1a7000c35be32fe2038f88fc64 Mon Sep 17 00:00:00 2001 From: Minghao Li Date: Fri, 17 Jun 2022 13:08:41 +0800 Subject: [PATCH 3/4] refactor(sync): add config index to json --- source/libs/sync/inc/syncRaftCfg.h | 9 +++++---- source/libs/sync/src/syncMain.c | 2 ++ source/libs/sync/src/syncRaftCfg.c | 7 +++++++ 3 files changed, 14 insertions(+), 4 deletions(-) diff --git a/source/libs/sync/inc/syncRaftCfg.h b/source/libs/sync/inc/syncRaftCfg.h index cd64402738..9969a0b974 100644 --- a/source/libs/sync/inc/syncRaftCfg.h +++ b/source/libs/sync/inc/syncRaftCfg.h @@ -47,14 +47,15 @@ typedef struct SRaftCfg { SRaftCfg *raftCfgOpen(const char *path); int32_t raftCfgClose(SRaftCfg *pRaftCfg); int32_t raftCfgPersist(SRaftCfg *pRaftCfg); +int32_t raftCfgAddConfigIndex(SRaftCfg *pRaftCfg, SyncIndex configIndex); -cJSON * syncCfg2Json(SSyncCfg *pSyncCfg); -char * syncCfg2Str(SSyncCfg *pSyncCfg); +cJSON *syncCfg2Json(SSyncCfg *pSyncCfg); +char *syncCfg2Str(SSyncCfg *pSyncCfg); int32_t syncCfgFromJson(const cJSON *pRoot, SSyncCfg *pSyncCfg); int32_t syncCfgFromStr(const char *s, SSyncCfg *pSyncCfg); -cJSON * raftCfg2Json(SRaftCfg *pRaftCfg); -char * raftCfg2Str(SRaftCfg *pRaftCfg); +cJSON *raftCfg2Json(SRaftCfg *pRaftCfg); +char *raftCfg2Str(SRaftCfg *pRaftCfg); int32_t raftCfgFromJson(const cJSON *pRoot, SRaftCfg *pRaftCfg); int32_t raftCfgFromStr(const char *s, SRaftCfg *pRaftCfg); diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c index 278288eb1e..7177ce4fe4 100644 --- a/source/libs/sync/src/syncMain.c +++ b/source/libs/sync/src/syncMain.c @@ -2307,6 +2307,8 @@ int32_t syncNodeCommit(SSyncNode* ths, SyncIndex beginIndex, SyncIndex endIndex, // config change if (pEntry->originalRpcType == TDMT_SYNC_CONFIG_CHANGE) { + raftCfgAddConfigIndex(ths->pRaftCfg, pEntry->index); + raftCfgPersist(ths->pRaftCfg); code = syncNodeConfigChange(ths, &rpcMsg, pEntry); ASSERT(code == 0); } diff --git a/source/libs/sync/src/syncRaftCfg.c b/source/libs/sync/src/syncRaftCfg.c index 2d51f1f6f0..8831704d7c 100644 --- a/source/libs/sync/src/syncRaftCfg.c +++ b/source/libs/sync/src/syncRaftCfg.c @@ -66,6 +66,13 @@ int32_t raftCfgPersist(SRaftCfg *pRaftCfg) { return 0; } +int32_t raftCfgAddConfigIndex(SRaftCfg *pRaftCfg, SyncIndex configIndex) { + ASSERT(pRaftCfg->configIndexCount <= MAX_CONFIG_INDEX_COUNT); + (pRaftCfg->configIndexArr)[pRaftCfg->configIndexCount] = configIndex; + ++(pRaftCfg->configIndexCount); + return 0; +} + cJSON *syncCfg2Json(SSyncCfg *pSyncCfg) { char u64buf[128] = {0}; cJSON *pRoot = cJSON_CreateObject(); From 9ab28e0b8868c6f397631fb24afb64e4f8c25ec6 Mon Sep 17 00:00:00 2001 From: Minghao Li Date: Fri, 17 Jun 2022 14:05:52 +0800 Subject: [PATCH 4/4] refactor(sync): add last config index in fsm cbMeta --- include/libs/sync/sync.h | 2 ++ source/libs/sync/inc/syncInt.h | 3 ++- source/libs/sync/src/syncAppendEntries.c | 15 ++++++++----- source/libs/sync/src/syncMain.c | 28 ++++++++++++++++++++---- 4 files changed, 38 insertions(+), 10 deletions(-) diff --git a/include/libs/sync/sync.h b/include/libs/sync/sync.h index f49030466e..e8c4faa240 100644 --- a/include/libs/sync/sync.h +++ b/include/libs/sync/sync.h @@ -62,6 +62,7 @@ typedef struct SSyncCfg { typedef struct SFsmCbMeta { SyncIndex index; + SyncIndex lastConfigIndex; bool isWeak; int32_t code; ESyncState state; @@ -75,6 +76,7 @@ typedef struct SReConfigCbMeta { int32_t code; SyncIndex index; SyncTerm term; + SyncIndex lastConfigIndex; SyncTerm currentTerm; SSyncCfg oldCfg; SSyncCfg newCfg; diff --git a/source/libs/sync/inc/syncInt.h b/source/libs/sync/inc/syncInt.h index b00c7cbda1..b6225c79cd 100644 --- a/source/libs/sync/inc/syncInt.h +++ b/source/libs/sync/inc/syncInt.h @@ -171,7 +171,8 @@ void syncNodeClose(SSyncNode* pSyncNode); int32_t syncNodePropose(SSyncNode* pSyncNode, const SRpcMsg* pMsg, bool isWeak); // option -bool syncNodeSnapshotEnable(SSyncNode* pSyncNode); +bool syncNodeSnapshotEnable(SSyncNode* pSyncNode); +SyncIndex syncNodeGetSnapshotConfigIndex(SSyncNode* pSyncNode, SyncIndex snapshotLastApplyIndex); // ping -------------- int32_t syncNodePing(SSyncNode* pSyncNode, const SRaftId* destRaftId, SyncPing* pMsg); diff --git a/source/libs/sync/src/syncAppendEntries.c b/source/libs/sync/src/syncAppendEntries.c index 7b2f79e24c..16350a8f40 100644 --- a/source/libs/sync/src/syncAppendEntries.c +++ b/source/libs/sync/src/syncAppendEntries.c @@ -208,8 +208,9 @@ int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) { SRpcMsg rpcMsg; syncEntry2OriginalRpc(pRollBackEntry, &rpcMsg); - SFsmCbMeta cbMeta; + SFsmCbMeta cbMeta = {0}; cbMeta.index = pRollBackEntry->index; + cbMeta.lastConfigIndex = syncNodeGetSnapshotConfigIndex(ths, cbMeta.index); cbMeta.isWeak = pRollBackEntry->isWeak; cbMeta.code = 0; cbMeta.state = ths->state; @@ -234,8 +235,9 @@ int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) { if (ths->pFsm != NULL) { // if (ths->pFsm->FpPreCommitCb != NULL && pAppendEntry->originalRpcType != TDMT_SYNC_NOOP) { if (ths->pFsm->FpPreCommitCb != NULL && syncUtilUserPreCommit(pAppendEntry->originalRpcType)) { - SFsmCbMeta cbMeta; + SFsmCbMeta cbMeta = {0}; cbMeta.index = pAppendEntry->index; + cbMeta.lastConfigIndex = syncNodeGetSnapshotConfigIndex(ths, cbMeta.index); cbMeta.isWeak = pAppendEntry->isWeak; cbMeta.code = 2; cbMeta.state = ths->state; @@ -266,8 +268,9 @@ int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) { if (ths->pFsm != NULL) { // if (ths->pFsm->FpPreCommitCb != NULL && pAppendEntry->originalRpcType != TDMT_SYNC_NOOP) { if (ths->pFsm->FpPreCommitCb != NULL && syncUtilUserPreCommit(pAppendEntry->originalRpcType)) { - SFsmCbMeta cbMeta; + SFsmCbMeta cbMeta = {0}; cbMeta.index = pAppendEntry->index; + cbMeta.lastConfigIndex = syncNodeGetSnapshotConfigIndex(ths, cbMeta.index); cbMeta.isWeak = pAppendEntry->isWeak; cbMeta.code = 3; cbMeta.state = ths->state; @@ -696,8 +699,9 @@ static int32_t syncNodeMakeLogSame(SSyncNode* ths, SyncAppendEntries* pMsg) { SRpcMsg rpcMsg; syncEntry2OriginalRpc(pRollBackEntry, &rpcMsg); - SFsmCbMeta cbMeta; + SFsmCbMeta cbMeta = {0}; cbMeta.index = pRollBackEntry->index; + cbMeta.lastConfigIndex = syncNodeGetSnapshotConfigIndex(ths, cbMeta.index); cbMeta.isWeak = pRollBackEntry->isWeak; cbMeta.code = 0; cbMeta.state = ths->state; @@ -725,8 +729,9 @@ static int32_t syncNodePreCommit(SSyncNode* ths, SSyncRaftEntry* pEntry) { syncEntry2OriginalRpc(pEntry, &rpcMsg); if (ths->pFsm != NULL) { if (ths->pFsm->FpPreCommitCb != NULL && syncUtilUserPreCommit(pEntry->originalRpcType)) { - SFsmCbMeta cbMeta; + SFsmCbMeta cbMeta = {0}; cbMeta.index = pEntry->index; + cbMeta.lastConfigIndex = syncNodeGetSnapshotConfigIndex(ths, cbMeta.index); cbMeta.isWeak = pEntry->isWeak; cbMeta.code = 2; cbMeta.state = ths->state; diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c index 7177ce4fe4..a2cb461a54 100644 --- a/source/libs/sync/src/syncMain.c +++ b/source/libs/sync/src/syncMain.c @@ -441,6 +441,21 @@ int32_t syncGetSnapshotMetaByIndex(int64_t rid, SyncIndex snapshotIndex, struct return 0; } +SyncIndex syncNodeGetSnapshotConfigIndex(SSyncNode* pSyncNode, SyncIndex snapshotLastApplyIndex) { + ASSERT(pSyncNode->pRaftCfg->configIndexCount >= 1); + SyncIndex lastIndex = (pSyncNode->pRaftCfg->configIndexArr)[0]; + + for (int i = 0; i < pSyncNode->pRaftCfg->configIndexCount; ++i) { + if ((pSyncNode->pRaftCfg->configIndexArr)[i] > lastIndex && + (pSyncNode->pRaftCfg->configIndexArr)[i] <= snapshotLastApplyIndex) { + lastIndex = (pSyncNode->pRaftCfg->configIndexArr)[i]; + } + } + + sTrace("sync syncNodeGetSnapshotConfigIndex index:%ld lastConfigIndex:%ld", snapshotLastApplyIndex, lastIndex); + return lastIndex; +} + const char* syncGetMyRoleStr(int64_t rid) { const char* s = syncUtilState2String(syncGetMyRole(rid)); return s; @@ -2065,8 +2080,9 @@ int32_t syncNodeOnClientRequestCb(SSyncNode* ths, SyncClientRequest* pMsg) { if (ths->pFsm != NULL) { // if (ths->pFsm->FpPreCommitCb != NULL && pEntry->originalRpcType != TDMT_SYNC_NOOP) { if (ths->pFsm->FpPreCommitCb != NULL && syncUtilUserPreCommit(pEntry->originalRpcType)) { - SFsmCbMeta cbMeta; + SFsmCbMeta cbMeta = {0}; cbMeta.index = pEntry->index; + cbMeta.lastConfigIndex = syncNodeGetSnapshotConfigIndex(ths, cbMeta.index); cbMeta.isWeak = pEntry->isWeak; cbMeta.code = 0; cbMeta.state = ths->state; @@ -2087,8 +2103,9 @@ int32_t syncNodeOnClientRequestCb(SSyncNode* ths, SyncClientRequest* pMsg) { if (ths->pFsm != NULL) { // if (ths->pFsm->FpPreCommitCb != NULL && pEntry->originalRpcType != TDMT_SYNC_NOOP) { if (ths->pFsm->FpPreCommitCb != NULL && syncUtilUserPreCommit(pEntry->originalRpcType)) { - SFsmCbMeta cbMeta; + SFsmCbMeta cbMeta = {0}; cbMeta.index = pEntry->index; + cbMeta.lastConfigIndex = syncNodeGetSnapshotConfigIndex(ths, cbMeta.index); cbMeta.isWeak = pEntry->isWeak; cbMeta.code = 1; cbMeta.state = ths->state; @@ -2152,11 +2169,12 @@ static int32_t syncDoLeaderTransfer(SSyncNode* ths, SRpcMsg* pRpcMsg, SSyncRaftE } */ if (ths->pFsm->FpLeaderTransferCb != NULL) { - SFsmCbMeta cbMeta; + SFsmCbMeta cbMeta = {0}; cbMeta.code = 0; cbMeta.currentTerm = ths->pRaftStore->currentTerm; cbMeta.flag = 0; cbMeta.index = pEntry->index; + cbMeta.lastConfigIndex = syncNodeGetSnapshotConfigIndex(ths, cbMeta.index); cbMeta.isWeak = pEntry->isWeak; cbMeta.seqNum = pEntry->seqNum; cbMeta.state = ths->state; @@ -2258,6 +2276,7 @@ static int32_t syncNodeConfigChange(SSyncNode* ths, SRpcMsg* pRpcMsg, SSyncRaftE cbMeta.code = 0; cbMeta.currentTerm = ths->pRaftStore->currentTerm; cbMeta.index = pEntry->index; + cbMeta.lastConfigIndex = syncNodeGetSnapshotConfigIndex(ths, pEntry->index); cbMeta.term = pEntry->term; cbMeta.newCfg = newSyncCfg; cbMeta.oldCfg = oldSyncCfg; @@ -2292,8 +2311,9 @@ int32_t syncNodeCommit(SSyncNode* ths, SyncIndex beginIndex, SyncIndex endIndex, // user commit if (ths->pFsm->FpCommitCb != NULL && syncUtilUserCommit(pEntry->originalRpcType)) { - SFsmCbMeta cbMeta; + SFsmCbMeta cbMeta = {0}; cbMeta.index = pEntry->index; + cbMeta.lastConfigIndex = syncNodeGetSnapshotConfigIndex(ths, cbMeta.index); cbMeta.isWeak = pEntry->isWeak; cbMeta.code = 0; cbMeta.state = ths->state;