diff --git a/source/libs/sync/src/syncIO.c b/source/libs/sync/src/syncIO.c index 19121632e20f55e1402898f352533cf9565c0249..fcfffea9c3c1154e5db56e82d6d394e010992dfa 100644 --- a/source/libs/sync/src/syncIO.c +++ b/source/libs/sync/src/syncIO.c @@ -29,7 +29,7 @@ static int32_t syncIODestroy(SSyncIO *io); static int32_t syncIOStartInternal(SSyncIO *io); static int32_t syncIOStopInternal(SSyncIO *io); -static void * syncIOConsumerFunc(void *param); +static void *syncIOConsumerFunc(void *param); static void syncIOProcessRequest(void *pParent, SRpcMsg *pMsg, SEpSet *pEpSet); static void syncIOProcessReply(void *pParent, SRpcMsg *pMsg, SEpSet *pEpSet); static int32_t syncIOAuth(void *parent, char *meterId, char *spi, char *encrypt, char *secret, char *ckey); @@ -234,9 +234,9 @@ static int32_t syncIOStopInternal(SSyncIO *io) { } static void *syncIOConsumerFunc(void *param) { - SSyncIO * io = param; + SSyncIO *io = param; STaosQall *qall; - SRpcMsg * pRpcMsg, rpcMsg; + SRpcMsg *pRpcMsg, rpcMsg; qall = taosAllocateQall(); while (1) { @@ -257,13 +257,6 @@ static void *syncIOConsumerFunc(void *param) { assert(pSyncMsg != NULL); io->FpOnSyncPing(io->pSyncNode, pSyncMsg); syncPingDestroy(pSyncMsg); - /* - pSyncMsg = syncPingBuild(pRpcMsg->contLen); - syncPingFromRpcMsg(pRpcMsg, pSyncMsg); - // memcpy(pSyncMsg, tmpRpcMsg.pCont, tmpRpcMsg.contLen); - io->FpOnSyncPing(io->pSyncNode, pSyncMsg); - syncPingDestroy(pSyncMsg); - */ } } else if (pRpcMsg->msgType == SYNC_PING_REPLY) { diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c index b92317ef3f75621a665b2fbb7ec8618f75e30a2b..98cf025093fcd060ab0bdf6360e50d0f7e3a4b24 100644 --- a/source/libs/sync/src/syncMain.c +++ b/source/libs/sync/src/syncMain.c @@ -122,7 +122,7 @@ int32_t syncPropose(int64_t rid, const SRpcMsg* pMsg, bool isWeak) { ret = 0; } else { - sTrace("syncForwardToPeer not leader, %s", syncUtilState2String(pSyncNode->state)); + sTrace("syncPropose not leader, %s", syncUtilState2String(pSyncNode->state)); ret = -1; // todo : need define err code !! } diff --git a/source/libs/sync/src/syncMessage.c b/source/libs/sync/src/syncMessage.c index 3ac4a7a1c083485fe776a92a59c2989b111966a7..f2344c6b6d413a2d8f3ec068a5c7aa974a6d212c 100644 --- a/source/libs/sync/src/syncMessage.c +++ b/source/libs/sync/src/syncMessage.c @@ -834,7 +834,7 @@ cJSON* syncRequestVote2Json(const SyncRequestVote* pMsg) { snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->term); cJSON_AddStringToObject(pRoot, "term", u64buf); - snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->lastLogIndex); + snprintf(u64buf, sizeof(u64buf), "%ld", pMsg->lastLogIndex); cJSON_AddStringToObject(pRoot, "lastLogIndex", u64buf); snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->lastLogTerm); cJSON_AddStringToObject(pRoot, "lastLogTerm", u64buf); @@ -1127,14 +1127,14 @@ cJSON* syncAppendEntries2Json(const SyncAppendEntries* pMsg) { snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->term); cJSON_AddStringToObject(pRoot, "term", u64buf); - snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->prevLogIndex); - cJSON_AddStringToObject(pRoot, "pre_log_index", u64buf); + snprintf(u64buf, sizeof(u64buf), "%ld", pMsg->prevLogIndex); + cJSON_AddStringToObject(pRoot, "prevLogIndex", u64buf); snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->prevLogTerm); cJSON_AddStringToObject(pRoot, "pre_log_term", u64buf); - snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->commitIndex); - cJSON_AddStringToObject(pRoot, "commit_index", u64buf); + snprintf(u64buf, sizeof(u64buf), "%ld", pMsg->commitIndex); + cJSON_AddStringToObject(pRoot, "commitIndex", u64buf); cJSON_AddNumberToObject(pRoot, "dataLen", pMsg->dataLen); char* s; @@ -1288,7 +1288,7 @@ cJSON* syncAppendEntriesReply2Json(const SyncAppendEntriesReply* pMsg) { snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->term); cJSON_AddStringToObject(pRoot, "term", u64buf); cJSON_AddNumberToObject(pRoot, "success", pMsg->success); - snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->matchIndex); + snprintf(u64buf, sizeof(u64buf), "%ld", pMsg->matchIndex); cJSON_AddStringToObject(pRoot, "matchIndex", u64buf); } diff --git a/source/libs/sync/test/CMakeLists.txt b/source/libs/sync/test/CMakeLists.txt index 653763b96bbb386ec7c471835804791c66e0a5c7..ec4351797435d5860a07316acea5d79ac89ce54a 100644 --- a/source/libs/sync/test/CMakeLists.txt +++ b/source/libs/sync/test/CMakeLists.txt @@ -33,6 +33,7 @@ add_executable(syncElectTest3 "") add_executable(syncEncodeTest "") add_executable(syncWriteTest "") add_executable(syncReplicateTest "") +add_executable(syncReplicateTest2 "") add_executable(syncRefTest "") @@ -176,6 +177,10 @@ target_sources(syncReplicateTest PRIVATE "syncReplicateTest.cpp" ) +target_sources(syncReplicateTest2 + PRIVATE + "syncReplicateTest2.cpp" +) target_sources(syncRefTest PRIVATE "syncRefTest.cpp" @@ -357,6 +362,11 @@ target_include_directories(syncReplicateTest "${CMAKE_SOURCE_DIR}/include/libs/sync" "${CMAKE_CURRENT_SOURCE_DIR}/../inc" ) +target_include_directories(syncReplicateTest2 + PUBLIC + "${CMAKE_SOURCE_DIR}/include/libs/sync" + "${CMAKE_CURRENT_SOURCE_DIR}/../inc" +) target_include_directories(syncRefTest PUBLIC "${CMAKE_SOURCE_DIR}/include/libs/sync" @@ -504,6 +514,10 @@ target_link_libraries(syncReplicateTest sync gtest_main ) +target_link_libraries(syncReplicateTest2 + sync + gtest_main +) target_link_libraries(syncRefTest sync gtest_main diff --git a/source/libs/sync/test/syncReplicateTest2.cpp b/source/libs/sync/test/syncReplicateTest2.cpp new file mode 100644 index 0000000000000000000000000000000000000000..352d2892bf6b4544251017c2dfd9f2b423b113dc --- /dev/null +++ b/source/libs/sync/test/syncReplicateTest2.cpp @@ -0,0 +1,199 @@ +#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 = 3; +int32_t myIndex = 0; + +SRaftId ids[TSDB_MAX_REPLICA]; +SSyncInfo syncInfo; +SSyncFSM *pFsm; +SWal * pWal; + +void CommitCb(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SyncIndex index, bool isWeak, int32_t code) { + char logBuf[256]; + snprintf(logBuf, sizeof(logBuf), "==callback== ==CommitCb== pFsm:%p, index:%ld, isWeak:%d, code:%d \n", pFsm, index, + isWeak, code); + syncRpcMsgPrint2(logBuf, (SRpcMsg *)pMsg); +} + +void PreCommitCb(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SyncIndex index, bool isWeak, int32_t code) { + char logBuf[256]; + snprintf(logBuf, sizeof(logBuf), "==callback== ==PreCommitCb== pFsm:%p, index:%ld, isWeak:%d, code:%d \n", pFsm, + index, isWeak, code); + syncRpcMsgPrint2(logBuf, (SRpcMsg *)pMsg); +} + +void RollBackCb(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SyncIndex index, bool isWeak, int32_t code) { + char logBuf[256]; + snprintf(logBuf, sizeof(logBuf), "==callback== ==RollBackCb== pFsm:%p, index:%ld, isWeak:%d, code:%d \n", pFsm, index, + isWeak, code); + syncRpcMsgPrint2(logBuf, (SRpcMsg *)pMsg); +} + +void initFsm() { + pFsm = (SSyncFSM *)malloc(sizeof(SSyncFSM)); + pFsm->FpCommitCb = CommitCb; + pFsm->FpPreCommitCb = PreCommitCb; + pFsm->FpRollBackCb = RollBackCb; +} + +int64_t 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), "./replicate2_test_%d", myIndex); + + 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; + + char tmpdir[128]; + snprintf(tmpdir, sizeof(tmpdir), "./replicate2_test_wal_%d", myIndex); + pWal = walOpen(tmpdir, &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); + } + + int64_t rid = syncStart(&syncInfo); + assert(rid > 0); + + SSyncNode *pSyncNode = (SSyncNode *)syncNodeAcquire(rid); + assert(pSyncNode != NULL); + + // pSyncNode->hbBaseLine = 500; + // pSyncNode->electBaseLine = 1500; + + 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->FpOnSyncTimeout = pSyncNode->FpOnTimeout; + gSyncIO->FpOnSyncClientRequest = pSyncNode->FpOnClientRequest; + gSyncIO->pSyncNode = pSyncNode; + + syncNodeRelease(pSyncNode); + + return rid; +} + +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); + } +} + +SRpcMsg *step0(int i) { + SRpcMsg *pMsg = (SRpcMsg *)malloc(sizeof(SRpcMsg)); + memset(pMsg, 0, sizeof(SRpcMsg)); + pMsg->msgType = 9999; + pMsg->contLen = 128; + pMsg->pCont = malloc(pMsg->contLen); + snprintf((char *)(pMsg->pCont), pMsg->contLen, "value-%u-%d", ports[myIndex], i); + return pMsg; +} + +SyncClientRequest *step1(const SRpcMsg *pMsg) { + SyncClientRequest *pRetMsg = syncClientRequestBuild2(pMsg, 123, true); + 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) { + myIndex = atoi(argv[1]); + } + + int32_t ret = syncIOStart((char *)"127.0.0.1", ports[myIndex]); + assert(ret == 0); + + initFsm(); + + ret = syncInit(); + assert(ret == 0); + + int64_t rid = syncNodeInit(); + assert(rid > 0); + + SSyncNode *pSyncNode = (SSyncNode *)syncNodeAcquire(rid); + assert(pSyncNode != NULL); + + syncNodePrint2((char *)"", pSyncNode); + initRaftId(pSyncNode); + + for (int i = 0; i < 30; ++i) { + // step0 + SRpcMsg *pMsg0 = step0(i); + syncRpcMsgPrint2((char *)"==step0==", pMsg0); + + syncPropose(rid, pMsg0, true); + taosMsleep(1000); + + free(pMsg0); + + sTrace( + "syncPropose sleep, state: %d, %s, term:%lu electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, " + "electTimerMS:%d", + pSyncNode->state, syncUtilState2String(pSyncNode->state), pSyncNode->pRaftStore->currentTerm, + pSyncNode->electTimerLogicClock, pSyncNode->electTimerLogicClockUser, pSyncNode->electTimerMS); + } + + while (1) { + sTrace( + "replicate sleep, state: %d, %s, term:%lu electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, " + "electTimerMS:%d", + pSyncNode->state, syncUtilState2String(pSyncNode->state), pSyncNode->pRaftStore->currentTerm, + pSyncNode->electTimerLogicClock, pSyncNode->electTimerLogicClockUser, pSyncNode->electTimerMS); + taosMsleep(1000); + } + + return 0; +}