From 01008939f2c441c4b28f01c6eb1963720b4000b5 Mon Sep 17 00:00:00 2001 From: Minghao Li Date: Tue, 19 Apr 2022 11:51:51 +0800 Subject: [PATCH] sync snapshot --- include/libs/sync/sync.h | 10 +- include/libs/sync/syncTools.h | 6 + source/libs/sync/test/CMakeLists.txt | 14 ++ source/libs/sync/test/syncSnapshotTest.cpp | 213 +++++++++++++++++++++ 4 files changed, 236 insertions(+), 7 deletions(-) create mode 100644 source/libs/sync/test/syncSnapshotTest.cpp diff --git a/include/libs/sync/sync.h b/include/libs/sync/sync.h index 86666f4c68..6bafa29a50 100644 --- a/include/libs/sync/sync.h +++ b/include/libs/sync/sync.h @@ -50,14 +50,10 @@ typedef struct SSyncCfg { SNodeInfo nodeInfo[TSDB_MAX_REPLICA]; } SSyncCfg; -typedef struct SRaftId { - SyncNodeId addr; - SyncGroupId vgId; -} SRaftId; - typedef struct SSnapshot { void* data; SyncIndex lastApplyIndex; + SyncTerm lastApplyTerm; } SSnapshot; typedef enum { @@ -80,8 +76,8 @@ typedef struct SSyncFSM { void (*FpPreCommitCb)(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SFsmCbMeta cbMeta); void (*FpRollBackCb)(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SFsmCbMeta cbMeta); - int32_t (*FpTakeSnapshot)(SSnapshot* snapshot); - int32_t (*FpRestoreSnapshot)(const SSnapshot* snapshot); + int32_t (*FpGetSnapshot)(struct SSyncFSM* pFsm, SSnapshot* pSnapshot); + int32_t (*FpRestoreSnapshot)(struct SSyncFSM* pFsm, const SSnapshot* snapshot); } SSyncFSM; diff --git a/include/libs/sync/syncTools.h b/include/libs/sync/syncTools.h index 6d28f54e29..567889191a 100644 --- a/include/libs/sync/syncTools.h +++ b/include/libs/sync/syncTools.h @@ -27,6 +27,12 @@ extern "C" { #include "trpc.h" #include "wal.h" +// ------------------ ds ------------------- +typedef struct SRaftId { + SyncNodeId addr; + SyncGroupId vgId; +} SRaftId; + // ------------------ control ------------------- struct SSyncNode; typedef struct SSyncNode SSyncNode; diff --git a/source/libs/sync/test/CMakeLists.txt b/source/libs/sync/test/CMakeLists.txt index 85fe285661..e2ac38150f 100644 --- a/source/libs/sync/test/CMakeLists.txt +++ b/source/libs/sync/test/CMakeLists.txt @@ -40,6 +40,7 @@ add_executable(syncRefTest "") add_executable(syncLogStoreCheck "") add_executable(syncRaftCfgTest "") add_executable(syncRespMgrTest "") +add_executable(syncSnapshotTest "") target_sources(syncTest @@ -210,6 +211,10 @@ target_sources(syncRespMgrTest PRIVATE "syncRespMgrTest.cpp" ) +target_sources(syncSnapshotTest + PRIVATE + "syncSnapshotTest.cpp" +) target_include_directories(syncTest @@ -422,6 +427,11 @@ target_include_directories(syncRespMgrTest "${CMAKE_SOURCE_DIR}/include/libs/sync" "${CMAKE_CURRENT_SOURCE_DIR}/../inc" ) +target_include_directories(syncSnapshotTest + PUBLIC + "${CMAKE_SOURCE_DIR}/include/libs/sync" + "${CMAKE_CURRENT_SOURCE_DIR}/../inc" +) target_link_libraries(syncTest @@ -592,6 +602,10 @@ target_link_libraries(syncRespMgrTest sync gtest_main ) +target_link_libraries(syncSnapshotTest + sync + gtest_main +) enable_testing() diff --git a/source/libs/sync/test/syncSnapshotTest.cpp b/source/libs/sync/test/syncSnapshotTest.cpp new file mode 100644 index 0000000000..e55f37d2d6 --- /dev/null +++ b/source/libs/sync/test/syncSnapshotTest.cpp @@ -0,0 +1,213 @@ +#include +#include +#include "syncEnv.h" +#include "syncIO.h" +#include "syncInt.h" +#include "syncMessage.h" +#include "syncRaftEntry.h" +#include "syncRaftLog.h" +#include "syncRaftStore.h" +#include "syncUtil.h" + +void logTest() { + sTrace("--- sync log test: trace"); + sDebug("--- sync log test: debug"); + sInfo("--- sync log test: info"); + sWarn("--- sync log test: warn"); + sError("--- sync log test: error"); + sFatal("--- sync log test: fatal"); +} + +uint16_t ports[] = {7010, 7110, 7210, 7310, 7410}; +int32_t replicaNum = 1; +int32_t myIndex = 0; + +SRaftId ids[TSDB_MAX_REPLICA]; +SSyncInfo syncInfo; +SSyncFSM * pFsm; +SWal * pWal; +SSyncNode *gSyncNode; +SyncIndex snapshotLastApplyIndex = SYNC_INDEX_INVALID; + +const char *pDir = "./syncSnapshotTest"; +const char *pWalDir = "./syncSnapshotTest_wal"; + +void CommitCb(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cbMeta) { + SyncIndex beginIndex = SYNC_INDEX_INVALID; + if (pFsm->FpGetSnapshot != NULL) { + SSnapshot snapshot; + pFsm->FpGetSnapshot(pFsm, &snapshot); + beginIndex = snapshot.lastApplyIndex; + } + + if (cbMeta.index > beginIndex) { + char logBuf[256]; + snprintf(logBuf, sizeof(logBuf), "==callback== ==CommitCb== pFsm:%p, index:%ld, isWeak:%d, code:%d, state:%d %s \n", + pFsm, cbMeta.index, cbMeta.isWeak, cbMeta.code, cbMeta.state, syncUtilState2String(cbMeta.state)); + syncRpcMsgLog2(logBuf, (SRpcMsg *)pMsg); + } else { + sTrace("==callback== ==CommitCb== do not apply again %ld", cbMeta.index); + } +} + +void PreCommitCb(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cbMeta) { + char logBuf[256]; + snprintf(logBuf, sizeof(logBuf), + "==callback== ==PreCommitCb== pFsm:%p, index:%ld, isWeak:%d, code:%d, state:%d %s \n", pFsm, cbMeta.index, + cbMeta.isWeak, cbMeta.code, cbMeta.state, syncUtilState2String(cbMeta.state)); + syncRpcMsgLog2(logBuf, (SRpcMsg *)pMsg); +} + +void RollBackCb(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cbMeta) { + char logBuf[256]; + snprintf(logBuf, sizeof(logBuf), "==callback== ==RollBackCb== pFsm:%p, index:%ld, isWeak:%d, code:%d, state:%d %s \n", + pFsm, cbMeta.index, cbMeta.isWeak, cbMeta.code, cbMeta.state, syncUtilState2String(cbMeta.state)); + syncRpcMsgLog2(logBuf, (SRpcMsg *)pMsg); +} + +int32_t GetSnapshotCb(struct SSyncFSM *pFsm, SSnapshot *pSnapshot) { + pSnapshot->data = NULL; + pSnapshot->lastApplyIndex = snapshotLastApplyIndex; + pSnapshot->lastApplyTerm = 100; + return 0; +} + +void initFsm() { + pFsm = (SSyncFSM *)taosMemoryMalloc(sizeof(SSyncFSM)); + pFsm->FpCommitCb = CommitCb; + pFsm->FpPreCommitCb = PreCommitCb; + pFsm->FpRollBackCb = RollBackCb; + pFsm->FpGetSnapshot = GetSnapshotCb; +} + +SSyncNode *syncNodeInit() { + syncInfo.vgId = 1234; + syncInfo.rpcClient = gSyncIO->clientRpc; + syncInfo.FpSendMsg = syncIOSendMsg; + syncInfo.queue = gSyncIO->pMsgQ; + syncInfo.FpEqMsg = syncIOEqMsg; + syncInfo.pFsm = pFsm; + snprintf(syncInfo.path, sizeof(syncInfo.path), "%s", pDir); + + int code = walInit(); + assert(code == 0); + SWalCfg walCfg; + memset(&walCfg, 0, sizeof(SWalCfg)); + walCfg.vgId = syncInfo.vgId; + walCfg.fsyncPeriod = 1000; + walCfg.retentionPeriod = 1000; + walCfg.rollPeriod = 1000; + walCfg.retentionSize = 1000; + walCfg.segSize = 1000; + walCfg.level = TAOS_WAL_FSYNC; + pWal = walOpen(pWalDir, &walCfg); + assert(pWal != NULL); + + syncInfo.pWal = pWal; + + SSyncCfg *pCfg = &syncInfo.syncCfg; + pCfg->myIndex = myIndex; + pCfg->replicaNum = replicaNum; + + for (int i = 0; i < replicaNum; ++i) { + pCfg->nodeInfo[i].nodePort = ports[i]; + snprintf(pCfg->nodeInfo[i].nodeFqdn, sizeof(pCfg->nodeInfo[i].nodeFqdn), "%s", "127.0.0.1"); + // taosGetFqdn(pCfg->nodeInfo[0].nodeFqdn); + } + + SSyncNode *pSyncNode = syncNodeOpen(&syncInfo); + assert(pSyncNode != NULL); + + gSyncIO->FpOnSyncPing = pSyncNode->FpOnPing; + gSyncIO->FpOnSyncClientRequest = pSyncNode->FpOnClientRequest; + gSyncIO->FpOnSyncPingReply = pSyncNode->FpOnPingReply; + gSyncIO->FpOnSyncRequestVote = pSyncNode->FpOnRequestVote; + gSyncIO->FpOnSyncRequestVoteReply = pSyncNode->FpOnRequestVoteReply; + gSyncIO->FpOnSyncAppendEntries = pSyncNode->FpOnAppendEntries; + gSyncIO->FpOnSyncAppendEntriesReply = pSyncNode->FpOnAppendEntriesReply; + gSyncIO->FpOnSyncTimeout = pSyncNode->FpOnTimeout; + gSyncIO->pSyncNode = pSyncNode; + + syncNodeStart(pSyncNode); + + return pSyncNode; +} + +SSyncNode *syncInitTest() { return syncNodeInit(); } + +void initRaftId(SSyncNode *pSyncNode) { + for (int i = 0; i < replicaNum; ++i) { + ids[i] = pSyncNode->replicasId[i]; + char *s = syncUtilRaftId2Str(&ids[i]); + printf("raftId[%d] : %s\n", i, s); + taosMemoryFree(s); + } +} + +SRpcMsg *step0() { + SRpcMsg *pMsg = (SRpcMsg *)taosMemoryMalloc(sizeof(SRpcMsg)); + memset(pMsg, 0, sizeof(SRpcMsg)); + pMsg->msgType = 9999; + pMsg->contLen = 32; + pMsg->pCont = taosMemoryMalloc(pMsg->contLen); + snprintf((char *)(pMsg->pCont), pMsg->contLen, "hello, world"); + return pMsg; +} + +SyncClientRequest *step1(const SRpcMsg *pMsg) { + SyncClientRequest *pRetMsg = syncClientRequestBuild2(pMsg, 123, true, 1000); + return pRetMsg; +} + +int main(int argc, char **argv) { + // taosInitLog((char *)"syncTest.log", 100000, 10); + tsAsyncLog = 0; + sDebugFlag = 143 + 64; + void logTest(); + + myIndex = 0; + if (argc >= 2) { + snapshotLastApplyIndex = atoi(argv[1]); + } + sTrace("--snapshotLastApplyIndex : %ld \n", snapshotLastApplyIndex); + + int32_t ret = syncIOStart((char *)"127.0.0.1", ports[myIndex]); + assert(ret == 0); + + ret = syncEnvStart(); + assert(ret == 0); + + // taosRemoveDir(pWalDir); + + initFsm(); + + gSyncNode = syncInitTest(); + assert(gSyncNode != NULL); + syncNodeLog2((char *)"", gSyncNode); + + initRaftId(gSyncNode); + + // step0 + SRpcMsg *pMsg0 = step0(); + syncRpcMsgLog2((char *)"==step0==", pMsg0); + + // step1 + SyncClientRequest *pMsg1 = step1(pMsg0); + syncClientRequestLog2((char *)"==step1==", pMsg1); + + for (int i = 0; i < 10; ++i) { + SyncClientRequest *pSyncClientRequest = pMsg1; + SRpcMsg rpcMsg; + syncClientRequest2RpcMsg(pSyncClientRequest, &rpcMsg); + gSyncNode->FpEqMsg(gSyncNode->queue, &rpcMsg); + + taosMsleep(1000); + } + + while (1) { + sTrace("while 1 sleep"); + taosMsleep(1000); + } + + return 0; +}