diff --git a/source/libs/sync/inc/syncInt.h b/source/libs/sync/inc/syncInt.h index 874763cd4565e6b830f7e5f0b31fe888aba4865d..447b75a5e8a8d5227a1d7af3c3d07d598f3f5f47 100644 --- a/source/libs/sync/inc/syncInt.h +++ b/source/libs/sync/inc/syncInt.h @@ -193,6 +193,7 @@ typedef struct SSyncNode { SSyncNode* syncNodeOpen(const SSyncInfo* pSyncInfo); void syncNodeClose(SSyncNode* pSyncNode); cJSON* syncNode2Json(const SSyncNode* pSyncNode); +char* syncNode2Str(const SSyncNode* pSyncNode); int32_t syncNodeSendMsgById(const SRaftId* destRaftId, SSyncNode* pSyncNode, SRpcMsg* pMsg); int32_t syncNodeSendMsgByInfo(const SNodeInfo* nodeInfo, SSyncNode* pSyncNode, SRpcMsg* pMsg); diff --git a/source/libs/sync/inc/syncUtil.h b/source/libs/sync/inc/syncUtil.h index 6cb1105b2c424dd1507d58b2e5413d94ad6cc954..f2c979da99730be198358b46e9260e9320df1258 100644 --- a/source/libs/sync/inc/syncUtil.h +++ b/source/libs/sync/inc/syncUtil.h @@ -47,6 +47,7 @@ int32_t syncUtilElectRandomMS(); int32_t syncUtilQuorum(int32_t replicaNum); cJSON* syncUtilNodeInfo2Json(const SNodeInfo* p); cJSON* syncUtilRaftId2Json(const SRaftId* p); +char* syncUtilRaftId2Str(const SRaftId* p); const char* syncUtilState2String(ESyncState state); #ifdef __cplusplus diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c index ad424b935390299cdb8cb0551158bdcb807cd9bb..3b8d716dbe5efc02d8b69706ca0b4ecfd5bf8a5e 100644 --- a/source/libs/sync/src/syncMain.c +++ b/source/libs/sync/src/syncMain.c @@ -291,6 +291,13 @@ cJSON* syncNode2Json(const SSyncNode* pSyncNode) { return pJson; } +char* syncNode2Str(const SSyncNode* pSyncNode) { + cJSON* pJson = syncNode2Json(pSyncNode); + char* serialized = cJSON_Print(pJson); + cJSON_Delete(pJson); + return serialized; +} + int32_t syncNodeSendMsgById(const SRaftId* destRaftId, SSyncNode* pSyncNode, SRpcMsg* pMsg) { SEpSet epSet; syncUtilraftId2EpSet(destRaftId, &epSet); diff --git a/source/libs/sync/src/syncUtil.c b/source/libs/sync/src/syncUtil.c index d2b413bdaad208a3189402e5d3ba10f646208b3b..3fb430a714bd8b7e1a390ca25cace75c1874f2c8 100644 --- a/source/libs/sync/src/syncUtil.c +++ b/source/libs/sync/src/syncUtil.c @@ -131,6 +131,13 @@ cJSON* syncUtilRaftId2Json(const SRaftId* p) { return pJson; } +char* syncUtilRaftId2Str(const SRaftId* p) { + cJSON* pJson = syncUtilRaftId2Json(p); + char* serialized = cJSON_Print(pJson); + cJSON_Delete(pJson); + return serialized; +} + const char* syncUtilState2String(ESyncState state) { if (state == TAOS_SYNC_STATE_FOLLOWER) { return "TAOS_SYNC_STATE_FOLLOWER"; diff --git a/source/libs/sync/src/syncVoteMgr.c b/source/libs/sync/src/syncVoteMgr.c index 407b7b1941d05497d8508522a7d5518786882724..3b6105ea0a89e7925b96cea8d77a94bb5a01a765 100644 --- a/source/libs/sync/src/syncVoteMgr.c +++ b/source/libs/sync/src/syncVoteMgr.c @@ -63,6 +63,7 @@ void voteGrantedVote(SVotesGranted *pVotesGranted, SyncRequestVoteReply *pMsg) { } } assert(j != -1); + assert(j >= 0 && j < pVotesGranted->replicaNum); if (pVotesGranted->isGranted[j] != true) { ++(pVotesGranted->votes); @@ -87,6 +88,14 @@ cJSON *voteGranted2Json(SVotesGranted *pVotesGranted) { for (int i = 0; i < pVotesGranted->replicaNum; ++i) { cJSON_AddItemToArray(pReplicas, syncUtilRaftId2Json(&(*(pVotesGranted->replicas)[i]))); } + int *arr = (int *)malloc(sizeof(int) * pVotesGranted->replicaNum); + for (int i = 0; i < pVotesGranted->replicaNum; ++i) { + arr[i] = pVotesGranted->isGranted[i]; + } + cJSON *pIsGranted = cJSON_CreateIntArray(arr, pVotesGranted->replicaNum); + free(arr); + cJSON_AddItemToObject(pRoot, "isGranted", pIsGranted); + cJSON_AddNumberToObject(pRoot, "votes", pVotesGranted->votes); snprintf(u64buf, sizeof(u64buf), "%lu", pVotesGranted->term); cJSON_AddStringToObject(pRoot, "term", u64buf); @@ -95,6 +104,9 @@ cJSON *voteGranted2Json(SVotesGranted *pVotesGranted) { snprintf(u64buf, sizeof(u64buf), "%p", pVotesGranted->pSyncNode); cJSON_AddStringToObject(pRoot, "pSyncNode", u64buf); + bool majority = voteGrantedMajority(pVotesGranted); + cJSON_AddNumberToObject(pRoot, "majority", majority); + cJSON *pJson = cJSON_CreateObject(); cJSON_AddItemToObject(pJson, "SVotesGranted", pRoot); return pJson; @@ -102,7 +114,7 @@ cJSON *voteGranted2Json(SVotesGranted *pVotesGranted) { char *voteGranted2Str(SVotesGranted *pVotesGranted) { cJSON *pJson = voteGranted2Json(pVotesGranted); - char * serialized = cJSON_Print(pJson); + char *serialized = cJSON_Print(pJson); cJSON_Delete(pJson); return serialized; } diff --git a/source/libs/sync/test/syncInitTest.cpp b/source/libs/sync/test/syncInitTest.cpp index 602297b2a6dec21ad5d530b0c45125cfef630c27..669c4e68a5c7b9cb5261a8f3db78a5db2c7751da 100644 --- a/source/libs/sync/test/syncInitTest.cpp +++ b/source/libs/sync/test/syncInitTest.cpp @@ -1,8 +1,10 @@ #include #include +#include "syncEnv.h" #include "syncIO.h" #include "syncInt.h" #include "syncRaftStore.h" +#include "syncUtil.h" void logTest() { sTrace("--- sync log test: trace"); @@ -13,13 +15,16 @@ void logTest() { sFatal("--- sync log test: fatal"); } -uint16_t ports[3] = {7010, 7110, 7210}; +uint16_t ports[] = {7010, 7110, 7210, 7310, 7410}; +int32_t replicaNum = 5; +int32_t myIndex = 0; -SSyncNode* syncInitTest() { - SSyncFSM* pFsm; +SRaftId ids[TSDB_MAX_REPLICA]; +SSyncInfo syncInfo; +SSyncFSM* pFsm; - SSyncInfo syncInfo; - syncInfo.vgId = 1; +SSyncNode* syncNodeInit() { + syncInfo.vgId = 1234; syncInfo.rpcClient = gSyncIO->clientRpc; syncInfo.FpSendMsg = syncIOSendMsg; syncInfo.queue = gSyncIO->pMsgQ; @@ -29,46 +34,66 @@ SSyncNode* syncInitTest() { snprintf(syncInfo.walPath, sizeof(syncInfo.walPath), "%s", "./test_wal_path"); SSyncCfg* pCfg = &syncInfo.syncCfg; - pCfg->myIndex = 0; - pCfg->replicaNum = 3; + pCfg->myIndex = myIndex; + pCfg->replicaNum = replicaNum; - pCfg->nodeInfo[0].nodePort = ports[0]; - snprintf(pCfg->nodeInfo[0].nodeFqdn, sizeof(pCfg->nodeInfo[0].nodeFqdn), "%s", "127.0.0.1"); - // taosGetFqdn(pCfg->nodeInfo[0].nodeFqdn); - - pCfg->nodeInfo[1].nodePort = ports[1]; - snprintf(pCfg->nodeInfo[1].nodeFqdn, sizeof(pCfg->nodeInfo[1].nodeFqdn), "%s", "127.0.0.1"); - // taosGetFqdn(pCfg->nodeInfo[1].nodeFqdn); - - pCfg->nodeInfo[2].nodePort = ports[2]; - snprintf(pCfg->nodeInfo[2].nodeFqdn, sizeof(pCfg->nodeInfo[2].nodeFqdn), "%s", "127.0.0.1"); - // taosGetFqdn(pCfg->nodeInfo[2].nodeFqdn); + 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->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; } -int main() { +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; - int32_t ret = syncIOStart((char*)"127.0.0.1", ports[0]); + 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); - cJSON* pJson = syncNode2Json(pSyncNode); - char* serialized = cJSON_Print(pJson); + char* serialized = syncNode2Str(pSyncNode); printf("%s\n", serialized); free(serialized); - cJSON_Delete(pJson); + + initRaftId(pSyncNode); return 0; } diff --git a/source/libs/sync/test/syncPingTest.cpp b/source/libs/sync/test/syncPingTest.cpp index 4e5f7b78b5965ab5ea9f3cebdeff4d02fb527506..450e097cc8a9e7d6c6c0cda6bd66b9c73c6af4d2 100644 --- a/source/libs/sync/test/syncPingTest.cpp +++ b/source/libs/sync/test/syncPingTest.cpp @@ -25,7 +25,8 @@ SSyncNode* doSync(int myIndex) { syncInfo.queue = gSyncIO->pMsgQ; syncInfo.FpEqMsg = syncIOEqMsg; syncInfo.pFsm = pFsm; - snprintf(syncInfo.path, sizeof(syncInfo.path), "%s", "./test_sync_ping"); + snprintf(syncInfo.path, sizeof(syncInfo.path), "%s", "./path"); + snprintf(syncInfo.walPath, sizeof(syncInfo.walPath), "%s", "./wal_path"); SSyncCfg* pCfg = &syncInfo.syncCfg; pCfg->myIndex = myIndex; diff --git a/source/libs/sync/test/syncVotesGrantedTest.cpp b/source/libs/sync/test/syncVotesGrantedTest.cpp index 97b807736b9416b7dea6e115ead1a09515a0fa75..3edde509f83e3ccf0eb07e0dc6b4604d4b3e90d6 100644 --- a/source/libs/sync/test/syncVotesGrantedTest.cpp +++ b/source/libs/sync/test/syncVotesGrantedTest.cpp @@ -4,6 +4,7 @@ #include "syncIO.h" #include "syncInt.h" #include "syncRaftStore.h" +#include "syncUtil.h" #include "syncVoteMgr.h" void logTest() { @@ -15,66 +16,141 @@ void logTest() { sFatal("--- sync log test: fatal"); } -uint16_t ports[3] = {7010, 7110, 7210}; +uint16_t ports[] = {7010, 7110, 7210, 7310, 7410}; +int32_t replicaNum = 3; +int32_t myIndex = 0; -SSyncNode* doSync(int myIndex) { - SSyncFSM* pFsm; +SRaftId ids[TSDB_MAX_REPLICA]; +SSyncInfo syncInfo; +SSyncFSM* pFsm; +SSyncNode* pSyncNode; - SSyncInfo syncInfo; - syncInfo.vgId = 1; +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_sync_ping"); + 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 = 3; + pCfg->replicaNum = replicaNum; - pCfg->nodeInfo[0].nodePort = ports[0]; - snprintf(pCfg->nodeInfo[0].nodeFqdn, sizeof(pCfg->nodeInfo[0].nodeFqdn), "%s", "127.0.0.1"); - // taosGetFqdn(pCfg->nodeInfo[0].nodeFqdn); + 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); + } - pCfg->nodeInfo[1].nodePort = ports[1]; - snprintf(pCfg->nodeInfo[1].nodeFqdn, sizeof(pCfg->nodeInfo[1].nodeFqdn), "%s", "127.0.0.1"); - // taosGetFqdn(pCfg->nodeInfo[1].nodeFqdn); - - pCfg->nodeInfo[2].nodePort = ports[2]; - snprintf(pCfg->nodeInfo[2].nodeFqdn, sizeof(pCfg->nodeInfo[2].nodeFqdn), "%s", "127.0.0.1"); - // taosGetFqdn(pCfg->nodeInfo[2].nodeFqdn); - - SSyncNode* pSyncNode = syncNodeOpen(&syncInfo); + 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; } -int main() { +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; - logTest(); - int myIndex = 0; + 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 = doSync(myIndex); - SVotesGranted* pVotesGranted = voteGrantedCreate(pSyncNode); - assert(pVotesGranted != NULL); + SSyncNode* pSyncNode = syncInitTest(); + assert(pSyncNode != NULL); - char* serialized = voteGranted2Str(pVotesGranted); - assert(serialized != 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; }