From 83b639ae8f4f44184cad19d42070ca84bcd49e54 Mon Sep 17 00:00:00 2001 From: Minghao Li Date: Wed, 6 Jul 2022 14:58:38 +0800 Subject: [PATCH] refactor(sync): persist batch size --- source/libs/sync/inc/syncRaftCfg.h | 2 ++ source/libs/sync/src/syncAppendEntries.c | 14 ++++++---- source/libs/sync/src/syncAppendEntriesReply.c | 28 +++++++++++++------ source/libs/sync/src/syncMain.c | 1 + source/libs/sync/src/syncRaftCfg.c | 5 ++++ source/libs/sync/test/syncRaftCfgTest.cpp | 3 ++ 6 files changed, 39 insertions(+), 14 deletions(-) diff --git a/source/libs/sync/inc/syncRaftCfg.h b/source/libs/sync/inc/syncRaftCfg.h index 086a6aa074..fecd66e4da 100644 --- a/source/libs/sync/inc/syncRaftCfg.h +++ b/source/libs/sync/inc/syncRaftCfg.h @@ -36,6 +36,7 @@ typedef struct SRaftCfg { TdFilePtr pFile; char path[TSDB_FILENAME_LEN * 2]; int8_t isStandBy; + int32_t batchSize; int8_t snapshotStrategy; SyncIndex lastConfigIndex; @@ -62,6 +63,7 @@ int32_t raftCfgFromStr(const char *s, SRaftCfg *pRaftCfg); typedef struct SRaftCfgMeta { int8_t isStandBy; + int32_t batchSize; int8_t snapshotStrategy; SyncIndex lastConfigIndex; } SRaftCfgMeta; diff --git a/source/libs/sync/src/syncAppendEntries.c b/source/libs/sync/src/syncAppendEntries.c index c923ee3d1d..885ab1acae 100644 --- a/source/libs/sync/src/syncAppendEntries.c +++ b/source/libs/sync/src/syncAppendEntries.c @@ -834,7 +834,7 @@ int32_t syncNodeOnAppendEntriesSnapshot2Cb(SSyncNode* ths, SyncAppendEntriesBatc // // operation: // if hasAppendEntries && pMsg->prevLogIndex == ths->commitIndex, append entry - // match my-commit-index or my-commit-index + 1 + // match my-commit-index or my-commit-index + batchSize do { bool condition = (pMsg->term == ths->pRaftStore->currentTerm) && (ths->state == TAOS_SYNC_STATE_FOLLOWER) && (pMsg->prevLogIndex <= ths->commitIndex); @@ -928,11 +928,13 @@ int32_t syncNodeOnAppendEntriesSnapshot2Cb(SSyncNode* ths, SyncAppendEntriesBatc bool condition = condition1 || condition2; if (condition) { - char logBuf[128]; - snprintf(logBuf, sizeof(logBuf), - "recv sync-append-entries-batch, not match, pre-index:%ld, pre-term:%lu, datalen:%d", pMsg->prevLogIndex, - pMsg->prevLogTerm, pMsg->dataLen); - syncNodeEventLog(ths, logBuf); + do { + char logBuf[128]; + snprintf(logBuf, sizeof(logBuf), + "recv sync-append-entries-batch, not match, pre-index:%ld, pre-term:%lu, datalen:%d", + pMsg->prevLogIndex, pMsg->prevLogTerm, pMsg->dataLen); + syncNodeEventLog(ths, logBuf); + } while (0); // prepare response msg SyncAppendEntriesReply* pReply = syncAppendEntriesReplyBuild(ths->vgId); diff --git a/source/libs/sync/src/syncAppendEntriesReply.c b/source/libs/sync/src/syncAppendEntriesReply.c index f3206e9ccc..e4724fc999 100644 --- a/source/libs/sync/src/syncAppendEntriesReply.c +++ b/source/libs/sync/src/syncAppendEntriesReply.c @@ -109,19 +109,30 @@ int32_t syncNodeOnAppendEntriesReplyCb(SSyncNode* ths, SyncAppendEntriesReply* p } // only start once -static void syncNodeStartSnapshot(SSyncNode* ths, SyncIndex beginIndex, SyncIndex endIndex, SyncTerm lastApplyTerm, - SyncAppendEntriesReply* pMsg) { +static void syncNodeStartSnapshotOnce(SSyncNode* ths, SyncIndex beginIndex, SyncIndex endIndex, SyncTerm lastApplyTerm, + SyncAppendEntriesReply* pMsg) { // get sender SSyncSnapshotSender* pSender = syncNodeGetSnapshotSender(ths, &(pMsg->srcId)); ASSERT(pSender != NULL); + if (snapshotSenderIsStart(pSender)) { + do { + char* eventLog = snapshotSender2SimpleStr(pSender, "snapshot sender already start"); + syncNodeErrorLog(ths, eventLog); + taosMemoryFree(eventLog); + } while (0); + + return; + } + SSnapshot snapshot = { .data = NULL, .lastApplyIndex = endIndex, .lastApplyTerm = lastApplyTerm, .lastConfigIndex = SYNC_INDEX_INVALID}; - void* pReader = NULL; SSnapshotParam readerParam = {.start = beginIndex, .end = endIndex}; - ths->pFsm->FpSnapshotStartRead(ths->pFsm, &readerParam, &pReader); - if (!snapshotSenderIsStart(pSender) && pMsg->privateTerm < pSender->privateTerm) { + int32_t code = ths->pFsm->FpSnapshotStartRead(ths->pFsm, &readerParam, &pReader); + ASSERT(code == 0); + + if (pMsg->privateTerm < pSender->privateTerm) { ASSERT(pReader != NULL); snapshotSenderStart(pSender, readerParam, snapshot, pReader); @@ -178,7 +189,9 @@ int32_t syncNodeOnAppendEntriesReplySnapshot2Cb(SSyncNode* ths, SyncAppendEntrie // start snapshot SSnapshot oldSnapshot; ths->pFsm->FpGetSnapshotInfo(ths->pFsm, &oldSnapshot); - syncNodeStartSnapshot(ths, newMatchIndex + 1, oldSnapshot.lastApplyIndex, oldSnapshot.lastApplyTerm, pMsg); + ASSERT(oldSnapshot.lastApplyIndex >= newMatchIndex + 1); + syncNodeStartSnapshotOnce(ths, newMatchIndex + 1, oldSnapshot.lastApplyIndex, oldSnapshot.lastApplyTerm, + pMsg); // term maybe not ok? syncIndexMgrSetIndex(ths->pNextIndex, &(pMsg->srcId), oldSnapshot.lastApplyIndex + 1); syncIndexMgrSetIndex(ths->pMatchIndex, &(pMsg->srcId), newMatchIndex); @@ -187,7 +200,6 @@ int32_t syncNodeOnAppendEntriesReplySnapshot2Cb(SSyncNode* ths, SyncAppendEntrie } else { SyncIndex nextIndex = syncIndexMgrGetIndex(ths->pNextIndex, &(pMsg->srcId)); - // notice! int64, uint64 if (nextIndex > SYNC_INDEX_BEGIN) { --nextIndex; @@ -198,7 +210,7 @@ int32_t syncNodeOnAppendEntriesReplySnapshot2Cb(SSyncNode* ths, SyncAppendEntrie SSyncRaftEntry* pEntry; int32_t code = ths->pLogStore->syncLogGetEntry(ths->pLogStore, nextIndex, &pEntry); ASSERT(code == 0); - syncNodeStartSnapshot(ths, SYNC_INDEX_BEGIN, nextIndex, pEntry->term, pMsg); + syncNodeStartSnapshotOnce(ths, SYNC_INDEX_BEGIN, nextIndex, pEntry->term, pMsg); // get sender SSyncSnapshotSender* pSender = syncNodeGetSnapshotSender(ths, &(pMsg->srcId)); diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c index 2192418c50..814fad5858 100644 --- a/source/libs/sync/src/syncMain.c +++ b/source/libs/sync/src/syncMain.c @@ -846,6 +846,7 @@ SSyncNode* syncNodeOpen(const SSyncInfo* pOldSyncInfo) { meta.isStandBy = pSyncInfo->isStandBy; meta.snapshotStrategy = pSyncInfo->snapshotStrategy; meta.lastConfigIndex = SYNC_INDEX_INVALID; + meta.batchSize = pSyncInfo->batchSize; ret = raftCfgCreateFile((SSyncCfg*)&(pSyncInfo->syncCfg), meta, pSyncNode->configPath); ASSERT(ret == 0); diff --git a/source/libs/sync/src/syncRaftCfg.c b/source/libs/sync/src/syncRaftCfg.c index 7eb7eb0db1..6c381f6e7d 100644 --- a/source/libs/sync/src/syncRaftCfg.c +++ b/source/libs/sync/src/syncRaftCfg.c @@ -183,6 +183,7 @@ cJSON *raftCfg2Json(SRaftCfg *pRaftCfg) { cJSON_AddItemToObject(pRoot, "SSyncCfg", syncCfg2Json(&(pRaftCfg->cfg))); cJSON_AddNumberToObject(pRoot, "isStandBy", pRaftCfg->isStandBy); cJSON_AddNumberToObject(pRoot, "snapshotStrategy", pRaftCfg->snapshotStrategy); + cJSON_AddNumberToObject(pRoot, "batchSize", pRaftCfg->batchSize); char buf64[128]; snprintf(buf64, sizeof(buf64), "%ld", pRaftCfg->lastConfigIndex); @@ -228,6 +229,7 @@ int32_t raftCfgCreateFile(SSyncCfg *pCfg, SRaftCfgMeta meta, const char *path) { SRaftCfg raftCfg; raftCfg.cfg = *pCfg; raftCfg.isStandBy = meta.isStandBy; + raftCfg.batchSize = meta.batchSize; raftCfg.snapshotStrategy = meta.snapshotStrategy; raftCfg.lastConfigIndex = meta.lastConfigIndex; raftCfg.configIndexCount = 1; @@ -257,6 +259,9 @@ int32_t raftCfgFromJson(const cJSON *pRoot, SRaftCfg *pRaftCfg) { cJSON *pJsonIsStandBy = cJSON_GetObjectItem(pJson, "isStandBy"); pRaftCfg->isStandBy = cJSON_GetNumberValue(pJsonIsStandBy); + cJSON *pJsonBatchSize = cJSON_GetObjectItem(pJson, "batchSize"); + pRaftCfg->batchSize = cJSON_GetNumberValue(pJsonBatchSize); + cJSON *pJsonSnapshotStrategy = cJSON_GetObjectItem(pJson, "snapshotStrategy"); pRaftCfg->snapshotStrategy = cJSON_GetNumberValue(pJsonSnapshotStrategy); diff --git a/source/libs/sync/test/syncRaftCfgTest.cpp b/source/libs/sync/test/syncRaftCfgTest.cpp index a3773604fb..2823a7826b 100644 --- a/source/libs/sync/test/syncRaftCfgTest.cpp +++ b/source/libs/sync/test/syncRaftCfgTest.cpp @@ -26,6 +26,7 @@ SRaftCfg* createRaftCfg() { snprintf(((pCfg->cfg.nodeInfo)[i]).nodeFqdn, sizeof(((pCfg->cfg.nodeInfo)[i]).nodeFqdn), "100.200.300.%d", i); } pCfg->isStandBy = taosGetTimestampSec() % 100; + pCfg->batchSize = taosGetTimestampSec() % 100; pCfg->configIndexCount = 5; for (int i = 0; i < MAX_CONFIG_INDEX_COUNT; ++i) { @@ -84,6 +85,7 @@ void test3() { SRaftCfgMeta meta; meta.isStandBy = 7; meta.snapshotStrategy = 9; + meta.batchSize = 10; meta.lastConfigIndex = 789; raftCfgCreateFile(pCfg, meta, s); printf("%s create json file: %s \n", (char*)__FUNCTION__, s); @@ -109,6 +111,7 @@ void test5() { pCfg->cfg.myIndex = taosGetTimestampSec(); pCfg->isStandBy += 2; pCfg->snapshotStrategy += 3; + pCfg->batchSize += 4; pCfg->lastConfigIndex += 1000; pCfg->configIndexCount = 5;