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

sync snapshot

This commit is contained in:
Minghao Li 2022-04-19 11:51:51 +08:00
parent 7066274cc7
commit 01008939f2
4 changed files with 236 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
include/libs/sync/sync.h
View File

@ -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;

6
include/libs/sync/syncTools.h
View File

@ -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;

14
source/libs/sync/test/CMakeLists.txt
View File

@ -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()

213
source/libs/sync/test/syncSnapshotTest.cpp Normal file
View File

@ -0,0 +1,213 @@
#include <gtest/gtest.h>
#include <stdio.h>
#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;
}