diff --git a/source/libs/sync/inc/syncRaftLog.h b/source/libs/sync/inc/syncRaftLog.h index ecdb544302..d59b3206b5 100644 --- a/source/libs/sync/inc/syncRaftLog.h +++ b/source/libs/sync/inc/syncRaftLog.h @@ -63,6 +63,9 @@ cJSON* logStore2Json(SSyncLogStore* pLogStore); char* logStore2Str(SSyncLogStore* pLogStore); +// for debug +void logStorePrint(SSyncLogStore* pLogStore); + #ifdef __cplusplus } #endif diff --git a/source/libs/sync/src/syncRaftLog.c b/source/libs/sync/src/syncRaftLog.c index d5735d9142..aa2595c199 100644 --- a/source/libs/sync/src/syncRaftLog.c +++ b/source/libs/sync/src/syncRaftLog.c @@ -115,11 +115,44 @@ SSyncRaftEntry* logStoreGetLastEntry(SSyncLogStore* pLogStore) { return pEntry; } -cJSON* logStore2Json(SSyncLogStore* pLogStore) {} +cJSON* logStore2Json(SSyncLogStore* pLogStore) { + char u64buf[128]; + + SSyncLogStoreData* pData = (SSyncLogStoreData*)pLogStore->data; + cJSON* pRoot = cJSON_CreateObject(); + snprintf(u64buf, sizeof(u64buf), "%p", pData->pSyncNode); + cJSON_AddStringToObject(pRoot, "pSyncNode", u64buf); + snprintf(u64buf, sizeof(u64buf), "%p", pData->pWal); + cJSON_AddStringToObject(pRoot, "pWal", u64buf); + snprintf(u64buf, sizeof(u64buf), "%lu", logStoreLastIndex(pLogStore)); + cJSON_AddStringToObject(pRoot, "LastIndex", u64buf); + snprintf(u64buf, sizeof(u64buf), "%lu", logStoreLastTerm(pLogStore)); + cJSON_AddStringToObject(pRoot, "LastTerm", u64buf); + + cJSON* pEntries = cJSON_CreateArray(); + cJSON_AddItemToObject(pRoot, "pEntries", pEntries); + SyncIndex lastIndex = logStoreLastIndex(pLogStore); + for (SyncIndex i = 1; i <= lastIndex; ++i) { + SSyncRaftEntry* pEntry = logStoreGetEntry(pLogStore, i); + cJSON_AddItemToArray(pEntries, syncEntry2Json(pEntry)); + syncEntryDestory(pEntry); + } + + cJSON* pJson = cJSON_CreateObject(); + cJSON_AddItemToObject(pJson, "SSyncLogStore", pRoot); + return pJson; +} char* logStore2Str(SSyncLogStore* pLogStore) { cJSON* pJson = logStore2Json(pLogStore); char* serialized = cJSON_Print(pJson); cJSON_Delete(pJson); return serialized; +} + +// for debug +void logStorePrint(SSyncLogStore* pLogStore) { + char* s = logStore2Str(pLogStore); + sTrace("%s", s); + free(s); } \ No newline at end of file diff --git a/source/libs/sync/test/CMakeLists.txt b/source/libs/sync/test/CMakeLists.txt index 5a5186c7e2..2163583f0a 100644 --- a/source/libs/sync/test/CMakeLists.txt +++ b/source/libs/sync/test/CMakeLists.txt @@ -15,6 +15,7 @@ add_executable(syncUtilTest "") add_executable(syncVotesGrantedTest "") add_executable(syncVotesRespondTest "") add_executable(syncIndexMgrTest "") +add_executable(syncLogStoreTest "") target_sources(syncTest @@ -85,6 +86,10 @@ target_sources(syncIndexMgrTest PRIVATE "syncIndexMgrTest.cpp" ) +target_sources(syncLogStoreTest + PRIVATE + "syncLogStoreTest.cpp" +) target_include_directories(syncTest @@ -172,6 +177,11 @@ target_include_directories(syncIndexMgrTest "${CMAKE_SOURCE_DIR}/include/libs/sync" "${CMAKE_CURRENT_SOURCE_DIR}/../inc" ) +target_include_directories(syncLogStoreTest + PUBLIC + "${CMAKE_SOURCE_DIR}/include/libs/sync" + "${CMAKE_CURRENT_SOURCE_DIR}/../inc" +) target_link_libraries(syncTest @@ -242,6 +252,10 @@ target_link_libraries(syncIndexMgrTest sync gtest_main ) +target_link_libraries(syncLogStoreTest + sync + gtest_main +) enable_testing() diff --git a/source/libs/sync/test/syncInitTest.cpp b/source/libs/sync/test/syncInitTest.cpp index f665c5a612..b6794544eb 100644 --- a/source/libs/sync/test/syncInitTest.cpp +++ b/source/libs/sync/test/syncInitTest.cpp @@ -53,6 +53,7 @@ SSyncNode* syncNodeInit() { gSyncIO->FpOnSyncAppendEntriesReply = pSyncNode->FpOnAppendEntriesReply; gSyncIO->FpOnSyncPing = pSyncNode->FpOnPing; gSyncIO->FpOnSyncPingReply = pSyncNode->FpOnPingReply; + gSyncIO->FpOnSyncTimeout = pSyncNode->FpOnTimeout; gSyncIO->pSyncNode = pSyncNode; return pSyncNode; @@ -88,11 +89,11 @@ int main(int argc, char** argv) { SSyncNode* pSyncNode = syncInitTest(); assert(pSyncNode != NULL); - char* serialized = syncNode2Str(pSyncNode); - printf("%s\n", serialized); - free(serialized); + syncNodePrint((char*)"syncInitTest", pSyncNode); initRaftId(pSyncNode); + //-------------------------------------------------------------- + return 0; -} +} \ No newline at end of file diff --git a/source/libs/sync/test/syncLogStoreTest.cpp b/source/libs/sync/test/syncLogStoreTest.cpp new file mode 100644 index 0000000000..a859186bbe --- /dev/null +++ b/source/libs/sync/test/syncLogStoreTest.cpp @@ -0,0 +1,132 @@ +#include +#include +#include "syncEnv.h" +#include "syncIO.h" +#include "syncInt.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* pSyncNode; + +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", "./"); + + SWalCfg walCfg; + memset(&walCfg, 0, sizeof(SWalCfg)); + walCfg.vgId = syncInfo.vgId; + walCfg.fsyncPeriod = 1000; + walCfg.retentionPeriod = 1000; + walCfg.rollPeriod = 1000; + walCfg.retentionSize = 100000; + walCfg.segSize = 100000; + walCfg.level = TAOS_WAL_WRITE; + pWal = walOpen("./wal_test", &walCfg); + + 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); + } + + pSyncNode = syncNodeOpen(&syncInfo); + assert(pSyncNode != NULL); + + gSyncIO->FpOnSyncPing = pSyncNode->FpOnPing; + gSyncIO->FpOnSyncPingReply = pSyncNode->FpOnPingReply; + gSyncIO->FpOnSyncRequestVote = pSyncNode->FpOnRequestVote; + gSyncIO->FpOnSyncRequestVoteReply = pSyncNode->FpOnRequestVoteReply; + gSyncIO->FpOnSyncAppendEntries = pSyncNode->FpOnAppendEntries; + gSyncIO->FpOnSyncAppendEntriesReply = pSyncNode->FpOnAppendEntriesReply; + gSyncIO->FpOnSyncPing = pSyncNode->FpOnPing; + gSyncIO->FpOnSyncPingReply = pSyncNode->FpOnPingReply; + gSyncIO->FpOnSyncTimeout = pSyncNode->FpOnTimeout; + gSyncIO->pSyncNode = pSyncNode; + + return pSyncNode; +} + +SSyncNode* syncInitTest() { return syncNodeInit(); } + +void logStoreTest() { + logStorePrint(pSyncNode->pLogStore); + for (int i = 0; i < 5; ++i) { + SSyncRaftEntry* pEntry; + pSyncNode->pLogStore->appendEntry(pSyncNode->pLogStore, pEntry); + } + logStorePrint(pSyncNode->pLogStore); + + pSyncNode->pLogStore->truncate(pSyncNode->pLogStore, 3); + logStorePrint(pSyncNode->pLogStore); +} + +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); + free(s); + } +} + +int main(int argc, char** argv) { + // taosInitLog((char *)"syncTest.log", 100000, 10); + tsAsyncLog = 0; + sDebugFlag = 143 + 64; + + myIndex = 0; + if (argc >= 2) { + myIndex = atoi(argv[1]); + } + + int32_t ret = syncIOStart((char*)"127.0.0.1", ports[myIndex]); + assert(ret == 0); + + ret = syncEnvStart(); + assert(ret == 0); + + pSyncNode = syncInitTest(); + assert(pSyncNode != NULL); + + syncNodePrint((char*)"syncLogStoreTest", pSyncNode); + + initRaftId(pSyncNode); + + //-------------------------------------------------------------- + + logStoreTest(); + + //-------------------------------------------------------------- + // walClose(pWal); + + return 0; +}