From 3b4149ba3268a9460f3c8fcfd8f52cacf929bd6a Mon Sep 17 00:00:00 2001 From: Minghao Li Date: Wed, 9 Mar 2022 10:15:40 +0800 Subject: [PATCH 1/3] sync refactor --- source/libs/sync/inc/syncVoteMgr.h | 6 +- source/libs/sync/src/syncVoteMgr.c | 51 +++++- source/libs/sync/test/CMakeLists.txt | 14 ++ .../libs/sync/test/syncVotesRespondTest.cpp | 156 ++++++++++++++++++ 4 files changed, 221 insertions(+), 6 deletions(-) create mode 100644 source/libs/sync/test/syncVotesRespondTest.cpp diff --git a/source/libs/sync/inc/syncVoteMgr.h b/source/libs/sync/inc/syncVoteMgr.h index a769bfbccd..ae9cfe8d01 100644 --- a/source/libs/sync/inc/syncVoteMgr.h +++ b/source/libs/sync/inc/syncVoteMgr.h @@ -31,8 +31,8 @@ extern "C" { // SVotesGranted ----------------------------- typedef struct SVotesGranted { SRaftId (*replicas)[TSDB_MAX_REPLICA]; - int32_t replicaNum; bool isGranted[TSDB_MAX_REPLICA]; + int32_t replicaNum; int32_t votes; SyncTerm term; int32_t quorum; @@ -61,7 +61,9 @@ SVotesRespond *votesRespondCreate(SSyncNode *pSyncNode); void votesRespondDestory(SVotesRespond *pVotesRespond); bool votesResponded(SVotesRespond *pVotesRespond, const SRaftId *pRaftId); void votesRespondAdd(SVotesRespond *pVotesRespond, const SyncRequestVoteReply *pMsg); -void Reset(SVotesRespond *pVotesRespond, SyncTerm term); +void votesRespondReset(SVotesRespond *pVotesRespond, SyncTerm term); +cJSON * votesRespond2Json(SVotesRespond *pVotesRespond); +char * votesRespond2Str(SVotesRespond *pVotesRespond); #ifdef __cplusplus } diff --git a/source/libs/sync/src/syncVoteMgr.c b/source/libs/sync/src/syncVoteMgr.c index 893a4e3a55..a2f10ce339 100644 --- a/source/libs/sync/src/syncVoteMgr.c +++ b/source/libs/sync/src/syncVoteMgr.c @@ -153,8 +153,8 @@ bool votesResponded(SVotesRespond *pVotesRespond, const SRaftId *pRaftId) { void votesRespondAdd(SVotesRespond *pVotesRespond, const SyncRequestVoteReply *pMsg) { assert(pVotesRespond->term == pMsg->term); for (int i = 0; i < pVotesRespond->replicaNum; ++i) { - if (syncUtilSameId(&(*pVotesRespond->replicas)[i], &pMsg->srcId)) { - assert(pVotesRespond->isRespond[i] == false); + if (syncUtilSameId(&((*(pVotesRespond->replicas))[i]), &pMsg->srcId)) { + // assert(pVotesRespond->isRespond[i] == false); pVotesRespond->isRespond[i] = true; return; } @@ -162,9 +162,52 @@ void votesRespondAdd(SVotesRespond *pVotesRespond, const SyncRequestVoteReply *p assert(0); } -void Reset(SVotesRespond *pVotesRespond, SyncTerm term) { +void votesRespondReset(SVotesRespond *pVotesRespond, SyncTerm term) { pVotesRespond->term = term; + memset(pVotesRespond->isRespond, 0, sizeof(pVotesRespond->isRespond)); + /* + for (int i = 0; i < pVotesRespond->replicaNum; ++i) { + pVotesRespond->isRespond[i] = false; + } + */ +} + +cJSON *votesRespond2Json(SVotesRespond *pVotesRespond) { + char u64buf[128]; + cJSON *pRoot = cJSON_CreateObject(); + + cJSON_AddNumberToObject(pRoot, "replicaNum", pVotesRespond->replicaNum); + cJSON *pReplicas = cJSON_CreateArray(); + cJSON_AddItemToObject(pRoot, "replicas", pReplicas); for (int i = 0; i < pVotesRespond->replicaNum; ++i) { - pVotesRespond->isRespond[i] = false; + cJSON_AddItemToArray(pReplicas, syncUtilRaftId2Json(&(*(pVotesRespond->replicas))[i])); } + int respondNum = 0; + int *arr = (int *)malloc(sizeof(int) * pVotesRespond->replicaNum); + for (int i = 0; i < pVotesRespond->replicaNum; ++i) { + arr[i] = pVotesRespond->isRespond[i]; + if (pVotesRespond->isRespond[i]) { + respondNum++; + } + } + cJSON *pIsRespond = cJSON_CreateIntArray(arr, pVotesRespond->replicaNum); + free(arr); + cJSON_AddItemToObject(pRoot, "isRespond", pIsRespond); + cJSON_AddNumberToObject(pRoot, "respondNum", respondNum); + + snprintf(u64buf, sizeof(u64buf), "%lu", pVotesRespond->term); + cJSON_AddStringToObject(pRoot, "term", u64buf); + snprintf(u64buf, sizeof(u64buf), "%p", pVotesRespond->pSyncNode); + cJSON_AddStringToObject(pRoot, "pSyncNode", u64buf); + + cJSON *pJson = cJSON_CreateObject(); + cJSON_AddItemToObject(pJson, "SVotesRespond", pRoot); + return pJson; +} + +char *votesRespond2Str(SVotesRespond *pVotesRespond) { + cJSON *pJson = votesRespond2Json(pVotesRespond); + char * serialized = cJSON_Print(pJson); + cJSON_Delete(pJson); + return serialized; } \ No newline at end of file diff --git a/source/libs/sync/test/CMakeLists.txt b/source/libs/sync/test/CMakeLists.txt index 18b7748105..8172a42a80 100644 --- a/source/libs/sync/test/CMakeLists.txt +++ b/source/libs/sync/test/CMakeLists.txt @@ -13,6 +13,7 @@ add_executable(syncIndexTest "") add_executable(syncInitTest "") add_executable(syncUtilTest "") add_executable(syncVotesGrantedTest "") +add_executable(syncVotesRespondTest "") target_sources(syncTest @@ -75,6 +76,10 @@ target_sources(syncVotesGrantedTest PRIVATE "syncVotesGrantedTest.cpp" ) +target_sources(syncVotesRespondTest + PRIVATE + "syncVotesRespondTest.cpp" +) target_include_directories(syncTest @@ -152,6 +157,11 @@ target_include_directories(syncVotesGrantedTest "${CMAKE_SOURCE_DIR}/include/libs/sync" "${CMAKE_CURRENT_SOURCE_DIR}/../inc" ) +target_include_directories(syncVotesRespondTest + PUBLIC + "${CMAKE_SOURCE_DIR}/include/libs/sync" + "${CMAKE_CURRENT_SOURCE_DIR}/../inc" +) target_link_libraries(syncTest @@ -214,6 +224,10 @@ target_link_libraries(syncVotesGrantedTest sync gtest_main ) +target_link_libraries(syncVotesRespondTest + sync + gtest_main +) enable_testing() diff --git a/source/libs/sync/test/syncVotesRespondTest.cpp b/source/libs/sync/test/syncVotesRespondTest.cpp new file mode 100644 index 0000000000..74d42cd531 --- /dev/null +++ b/source/libs/sync/test/syncVotesRespondTest.cpp @@ -0,0 +1,156 @@ +#include +#include +#include "syncEnv.h" +#include "syncIO.h" +#include "syncInt.h" +#include "syncRaftStore.h" +#include "syncUtil.h" +#include "syncVoteMgr.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 = 3; +int32_t myIndex = 0; + +SRaftId ids[TSDB_MAX_REPLICA]; +SSyncInfo syncInfo; +SSyncFSM* pFsm; +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", "./test_path"); + snprintf(syncInfo.walPath, sizeof(syncInfo.walPath), "%s", "./test_wal_path"); + + 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->pSyncNode = 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); + 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); + + SSyncNode* pSyncNode = syncInitTest(); + assert(pSyncNode != NULL); + + char* serialized = syncNode2Str(pSyncNode); + printf("%s\n", serialized); + free(serialized); + + initRaftId(pSyncNode); + + SVotesRespond* pVotesRespond = votesRespondCreate(pSyncNode); + assert(pVotesRespond != NULL); + + printf("---------------------------------------\n"); + { + char* serialized = votesRespond2Str(pVotesRespond); + assert(serialized != NULL); + printf("%s\n", serialized); + free(serialized); + } + + SyncTerm term = 1234; + printf("---------------------------------------\n"); + votesRespondReset(pVotesRespond, term); + { + char* serialized = votesRespond2Str(pVotesRespond); + assert(serialized != NULL); + printf("%s\n", serialized); + free(serialized); + } + + for (int i = 0; i < replicaNum; ++i) { + SyncRequestVoteReply* reply = SyncRequestVoteReplyBuild(); + reply->destId = pSyncNode->myRaftId; + reply->srcId = ids[i]; + reply->term = term; + reply->voteGranted = true; + + votesRespondAdd(pVotesRespond, reply); + { + char* serialized = votesRespond2Str(pVotesRespond); + assert(serialized != NULL); + printf("%s\n", serialized); + free(serialized); + } + + votesRespondAdd(pVotesRespond, reply); + { + char* serialized = votesRespond2Str(pVotesRespond); + assert(serialized != NULL); + printf("%s\n", serialized); + free(serialized); + } + } + + printf("---------------------------------------\n"); + votesRespondReset(pVotesRespond, 123456789); + { + char* serialized = votesRespond2Str(pVotesRespond); + assert(serialized != NULL); + printf("%s\n", serialized); + free(serialized); + } + + votesRespondDestory(pVotesRespond); + return 0; +} From d3b1d5d8f1063928d958cff873bd639748eca668 Mon Sep 17 00:00:00 2001 From: Minghao Li Date: Wed, 9 Mar 2022 10:58:22 +0800 Subject: [PATCH 2/3] sync refactor --- source/libs/sync/inc/syncIndexMgr.h | 49 +++++++ source/libs/sync/src/syncIndexMgr.c | 99 +++++++++++++ source/libs/sync/test/CMakeLists.txt | 14 ++ source/libs/sync/test/syncIndexMgrTest.cpp | 156 +++++++++++++++++++++ 4 files changed, 318 insertions(+) create mode 100644 source/libs/sync/inc/syncIndexMgr.h create mode 100644 source/libs/sync/src/syncIndexMgr.c create mode 100644 source/libs/sync/test/syncIndexMgrTest.cpp diff --git a/source/libs/sync/inc/syncIndexMgr.h b/source/libs/sync/inc/syncIndexMgr.h new file mode 100644 index 0000000000..7116ae9d46 --- /dev/null +++ b/source/libs/sync/inc/syncIndexMgr.h @@ -0,0 +1,49 @@ +/* + * Copyright (c) 2019 TAOS Data, Inc. + * + * This program is free software: you can use, redistribute, and/or modify + * it under the terms of the GNU Affero General Public License, version 3 + * or later ("AGPL"), as published by the Free Software Foundation. + * + * This program is distributed in the hope that it will be useful, but WITHOUT + * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or + * FITNESS FOR A PARTICULAR PURPOSE. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ + +#ifndef _TD_LIBS_SYNC_INDEX_MGR_H +#define _TD_LIBS_SYNC_INDEX_MGR_H + +#ifdef __cplusplus +extern "C" { +#endif + +#include +#include +#include +#include "syncInt.h" +#include "taosdef.h" + +// SIndexMgr ----------------------------- +typedef struct SSyncIndexMgr { + SRaftId (*replicas)[TSDB_MAX_REPLICA]; + SyncIndex index[TSDB_MAX_REPLICA]; + int32_t replicaNum; + SSyncNode *pSyncNode; +} SSyncIndexMgr; + +SSyncIndexMgr *syncIndexMgrCreate(SSyncNode *pSyncNode); +void syncIndexMgrDestroy(SSyncIndexMgr *pSyncIndexMgr); +void syncIndexMgrClear(SSyncIndexMgr *pSyncIndexMgr); +void syncIndexMgrSetIndex(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId, SyncIndex index); +SyncIndex syncIndexMgrGetIndex(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId); +cJSON * syncIndexMgr2Json(SSyncIndexMgr *pSyncIndexMgr); +char * syncIndexMgr2Str(SSyncIndexMgr *pSyncIndexMgr); + +#ifdef __cplusplus +} +#endif + +#endif /*_TD_LIBS_SYNC_INDEX_MGR_H*/ diff --git a/source/libs/sync/src/syncIndexMgr.c b/source/libs/sync/src/syncIndexMgr.c new file mode 100644 index 0000000000..86b719ba1e --- /dev/null +++ b/source/libs/sync/src/syncIndexMgr.c @@ -0,0 +1,99 @@ +/* + * Copyright (c) 2019 TAOS Data, Inc. + * + * This program is free software: you can use, redistribute, and/or modify + * it under the terms of the GNU Affero General Public License, version 3 + * or later ("AGPL"), as published by the Free Software Foundation. + * + * This program is distributed in the hope that it will be useful, but WITHOUT + * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or + * FITNESS FOR A PARTICULAR PURPOSE. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ + +#include "syncIndexMgr.h" +#include "syncUtil.h" + +// SMatchIndex ----------------------------- + +SSyncIndexMgr *syncIndexMgrCreate(SSyncNode *pSyncNode) { + SSyncIndexMgr *pSyncIndexMgr = malloc(sizeof(SSyncIndexMgr)); + assert(pSyncIndexMgr != NULL); + memset(pSyncIndexMgr, 0, sizeof(SSyncIndexMgr)); + + pSyncIndexMgr->replicas = &(pSyncNode->replicasId); + pSyncIndexMgr->replicaNum = pSyncNode->replicaNum; + pSyncIndexMgr->pSyncNode = pSyncNode; + syncIndexMgrClear(pSyncIndexMgr); + + return pSyncIndexMgr; +} + +void syncIndexMgrDestroy(SSyncIndexMgr *pSyncIndexMgr) { + if (pSyncIndexMgr != NULL) { + free(pSyncIndexMgr); + } +} + +void syncIndexMgrClear(SSyncIndexMgr *pSyncIndexMgr) { + memset(pSyncIndexMgr->index, 0, sizeof(pSyncIndexMgr->index)); + /* + for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) { + pSyncIndexMgr->index[i] = 0; + } + */ +} + +void syncIndexMgrSetIndex(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId, SyncIndex index) { + for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) { + if (syncUtilSameId(&((*(pSyncIndexMgr->replicas))[i]), pRaftId)) { + (pSyncIndexMgr->index)[i] = index; + return; + } + } + assert(0); +} + +SyncIndex syncIndexMgrGetIndex(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId) { + for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) { + if (syncUtilSameId(&((*(pSyncIndexMgr->replicas))[i]), pRaftId)) { + SyncIndex idx = (pSyncIndexMgr->index)[i]; + return idx; + } + } + assert(0); +} + +cJSON *syncIndexMgr2Json(SSyncIndexMgr *pSyncIndexMgr) { + char u64buf[128]; + cJSON *pRoot = cJSON_CreateObject(); + + cJSON_AddNumberToObject(pRoot, "replicaNum", pSyncIndexMgr->replicaNum); + cJSON *pReplicas = cJSON_CreateArray(); + cJSON_AddItemToObject(pRoot, "replicas", pReplicas); + for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) { + cJSON_AddItemToArray(pReplicas, syncUtilRaftId2Json(&(*(pSyncIndexMgr->replicas))[i])); + } + int respondNum = 0; + int *arr = (int *)malloc(sizeof(int) * pSyncIndexMgr->replicaNum); + for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) { + arr[i] = pSyncIndexMgr->index[i]; + } + cJSON *pIsRespond = cJSON_CreateIntArray(arr, pSyncIndexMgr->replicaNum); + free(arr); + snprintf(u64buf, sizeof(u64buf), "%p", pSyncIndexMgr->pSyncNode); + cJSON_AddStringToObject(pRoot, "pSyncNode", u64buf); + + cJSON *pJson = cJSON_CreateObject(); + cJSON_AddItemToObject(pJson, "pSyncIndexMgr", pRoot); + return pJson; +} + +char *syncIndexMgr2Str(SSyncIndexMgr *pSyncIndexMgr) { + cJSON *pJson = syncIndexMgr2Json(pSyncIndexMgr); + char *serialized = cJSON_Print(pJson); + cJSON_Delete(pJson); + return serialized; +} \ No newline at end of file diff --git a/source/libs/sync/test/CMakeLists.txt b/source/libs/sync/test/CMakeLists.txt index 8172a42a80..5a5186c7e2 100644 --- a/source/libs/sync/test/CMakeLists.txt +++ b/source/libs/sync/test/CMakeLists.txt @@ -14,6 +14,7 @@ add_executable(syncInitTest "") add_executable(syncUtilTest "") add_executable(syncVotesGrantedTest "") add_executable(syncVotesRespondTest "") +add_executable(syncIndexMgrTest "") target_sources(syncTest @@ -80,6 +81,10 @@ target_sources(syncVotesRespondTest PRIVATE "syncVotesRespondTest.cpp" ) +target_sources(syncIndexMgrTest + PRIVATE + "syncIndexMgrTest.cpp" +) target_include_directories(syncTest @@ -162,6 +167,11 @@ target_include_directories(syncVotesRespondTest "${CMAKE_SOURCE_DIR}/include/libs/sync" "${CMAKE_CURRENT_SOURCE_DIR}/../inc" ) +target_include_directories(syncIndexMgrTest + PUBLIC + "${CMAKE_SOURCE_DIR}/include/libs/sync" + "${CMAKE_CURRENT_SOURCE_DIR}/../inc" +) target_link_libraries(syncTest @@ -228,6 +238,10 @@ target_link_libraries(syncVotesRespondTest sync gtest_main ) +target_link_libraries(syncIndexMgrTest + sync + gtest_main +) enable_testing() diff --git a/source/libs/sync/test/syncIndexMgrTest.cpp b/source/libs/sync/test/syncIndexMgrTest.cpp new file mode 100644 index 0000000000..3edde509f8 --- /dev/null +++ b/source/libs/sync/test/syncIndexMgrTest.cpp @@ -0,0 +1,156 @@ +#include +#include +#include "syncEnv.h" +#include "syncIO.h" +#include "syncInt.h" +#include "syncRaftStore.h" +#include "syncUtil.h" +#include "syncVoteMgr.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 = 3; +int32_t myIndex = 0; + +SRaftId ids[TSDB_MAX_REPLICA]; +SSyncInfo syncInfo; +SSyncFSM* pFsm; +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", "./test_path"); + snprintf(syncInfo.walPath, sizeof(syncInfo.walPath), "%s", "./test_wal_path"); + + 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->pSyncNode = 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); + 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); + + SSyncNode* pSyncNode = syncInitTest(); + assert(pSyncNode != NULL); + + char* serialized = syncNode2Str(pSyncNode); + printf("%s\n", serialized); + free(serialized); + + initRaftId(pSyncNode); + + SVotesGranted* pVotesGranted = voteGrantedCreate(pSyncNode); + assert(pVotesGranted != NULL); + + printf("---------------------------------------\n"); + { + char* serialized = voteGranted2Str(pVotesGranted); + assert(serialized != NULL); + printf("%s\n", serialized); + free(serialized); + } + + SyncTerm term = 1234; + printf("---------------------------------------\n"); + voteGrantedReset(pVotesGranted, term); + { + char* serialized = voteGranted2Str(pVotesGranted); + assert(serialized != NULL); + printf("%s\n", serialized); + free(serialized); + } + + for (int i = 0; i < replicaNum; ++i) { + SyncRequestVoteReply* reply = SyncRequestVoteReplyBuild(); + reply->destId = pSyncNode->myRaftId; + reply->srcId = ids[i]; + reply->term = term; + reply->voteGranted = true; + + voteGrantedVote(pVotesGranted, reply); + { + char* serialized = voteGranted2Str(pVotesGranted); + assert(serialized != NULL); + printf("%s\n", serialized); + free(serialized); + } + + voteGrantedVote(pVotesGranted, reply); + { + char* serialized = voteGranted2Str(pVotesGranted); + assert(serialized != NULL); + printf("%s\n", serialized); + free(serialized); + } + } + + printf("---------------------------------------\n"); + voteGrantedReset(pVotesGranted, 123456789); + { + char* serialized = voteGranted2Str(pVotesGranted); + assert(serialized != NULL); + printf("%s\n", serialized); + free(serialized); + } + + voteGrantedDestroy(pVotesGranted); + return 0; +} From 2fcc970d79727ba9ac3d06e4ff1fc67e771d960e Mon Sep 17 00:00:00 2001 From: Minghao Li Date: Wed, 9 Mar 2022 11:27:22 +0800 Subject: [PATCH 3/3] sync refactor --- source/libs/sync/inc/syncInt.h | 7 +++- source/libs/sync/src/syncIndexMgr.c | 5 ++- source/libs/sync/test/syncIndexMgrTest.cpp | 49 ++++++++-------------- 3 files changed, 25 insertions(+), 36 deletions(-) diff --git a/source/libs/sync/inc/syncInt.h b/source/libs/sync/inc/syncInt.h index 447b75a5e8..8b77e292c4 100644 --- a/source/libs/sync/inc/syncInt.h +++ b/source/libs/sync/inc/syncInt.h @@ -103,6 +103,9 @@ typedef struct SVotesGranted SVotesGranted; struct SVotesRespond; typedef struct SVotesRespond SVotesRespond; +struct SSyncIndexMgr; +typedef struct SSyncIndexMgr SSyncIndexMgr; + typedef struct SRaftId { SyncNodeId addr; // typedef uint64_t SyncNodeId; SyncGroupId vgId; // typedef int32_t SyncGroupId; @@ -148,8 +151,8 @@ typedef struct SSyncNode { SVotesRespond* pVotesRespond; // tla+ leader vars - SHashObj* pNextIndex; - SHashObj* pMatchIndex; + SSyncIndexMgr* pNextIndex; + SSyncIndexMgr* pMatchIndex; // tla+ log vars SSyncLogStore* pLogStore; diff --git a/source/libs/sync/src/syncIndexMgr.c b/source/libs/sync/src/syncIndexMgr.c index 86b719ba1e..fff54638e2 100644 --- a/source/libs/sync/src/syncIndexMgr.c +++ b/source/libs/sync/src/syncIndexMgr.c @@ -81,8 +81,9 @@ cJSON *syncIndexMgr2Json(SSyncIndexMgr *pSyncIndexMgr) { for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) { arr[i] = pSyncIndexMgr->index[i]; } - cJSON *pIsRespond = cJSON_CreateIntArray(arr, pSyncIndexMgr->replicaNum); + cJSON *pIndex = cJSON_CreateIntArray(arr, pSyncIndexMgr->replicaNum); free(arr); + cJSON_AddItemToObject(pRoot, "index", pIndex); snprintf(u64buf, sizeof(u64buf), "%p", pSyncIndexMgr->pSyncNode); cJSON_AddStringToObject(pRoot, "pSyncNode", u64buf); @@ -93,7 +94,7 @@ cJSON *syncIndexMgr2Json(SSyncIndexMgr *pSyncIndexMgr) { char *syncIndexMgr2Str(SSyncIndexMgr *pSyncIndexMgr) { cJSON *pJson = syncIndexMgr2Json(pSyncIndexMgr); - char *serialized = cJSON_Print(pJson); + char * serialized = cJSON_Print(pJson); cJSON_Delete(pJson); return serialized; } \ No newline at end of file diff --git a/source/libs/sync/test/syncIndexMgrTest.cpp b/source/libs/sync/test/syncIndexMgrTest.cpp index 3edde509f8..4e4cd9222b 100644 --- a/source/libs/sync/test/syncIndexMgrTest.cpp +++ b/source/libs/sync/test/syncIndexMgrTest.cpp @@ -1,4 +1,5 @@ -#include +#include "syncIndexMgr.h" +//#include #include #include "syncEnv.h" #include "syncIO.h" @@ -97,60 +98,44 @@ int main(int argc, char** argv) { initRaftId(pSyncNode); - SVotesGranted* pVotesGranted = voteGrantedCreate(pSyncNode); - assert(pVotesGranted != NULL); + SSyncIndexMgr* pSyncIndexMgr = syncIndexMgrCreate(pSyncNode); + assert(pSyncIndexMgr != NULL); printf("---------------------------------------\n"); { - char* serialized = voteGranted2Str(pVotesGranted); + char* serialized = syncIndexMgr2Str(pSyncIndexMgr); assert(serialized != NULL); printf("%s\n", serialized); free(serialized); } - SyncTerm term = 1234; + syncIndexMgrSetIndex(pSyncIndexMgr, &ids[0], 100); + syncIndexMgrSetIndex(pSyncIndexMgr, &ids[1], 200); + syncIndexMgrSetIndex(pSyncIndexMgr, &ids[2], 300); + printf("---------------------------------------\n"); - voteGrantedReset(pVotesGranted, term); { - char* serialized = voteGranted2Str(pVotesGranted); + char* serialized = syncIndexMgr2Str(pSyncIndexMgr); assert(serialized != NULL); printf("%s\n", serialized); free(serialized); } - for (int i = 0; i < replicaNum; ++i) { - SyncRequestVoteReply* reply = SyncRequestVoteReplyBuild(); - reply->destId = pSyncNode->myRaftId; - reply->srcId = ids[i]; - reply->term = term; - reply->voteGranted = true; - - voteGrantedVote(pVotesGranted, reply); - { - char* serialized = voteGranted2Str(pVotesGranted); - assert(serialized != NULL); - printf("%s\n", serialized); - free(serialized); - } - - voteGrantedVote(pVotesGranted, reply); - { - char* serialized = voteGranted2Str(pVotesGranted); - assert(serialized != NULL); - printf("%s\n", serialized); - free(serialized); - } + printf("---------------------------------------\n"); + for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) { + SyncIndex idx = syncIndexMgrGetIndex(pSyncIndexMgr, &ids[i]); + printf("index %d : %lu \n", i, idx); } + syncIndexMgrClear(pSyncIndexMgr); printf("---------------------------------------\n"); - voteGrantedReset(pVotesGranted, 123456789); { - char* serialized = voteGranted2Str(pVotesGranted); + char* serialized = syncIndexMgr2Str(pSyncIndexMgr); assert(serialized != NULL); printf("%s\n", serialized); free(serialized); } - voteGrantedDestroy(pVotesGranted); + syncIndexMgrDestroy(pSyncIndexMgr); return 0; }