diff --git a/source/libs/sync/src/syncCommit.c b/source/libs/sync/src/syncCommit.c index c75d23d96d..38b39e04cf 100644 --- a/source/libs/sync/src/syncCommit.c +++ b/source/libs/sync/src/syncCommit.c @@ -18,6 +18,7 @@ #include "syncInt.h" #include "syncRaftLog.h" #include "syncRaftStore.h" +#include "syncUtil.h" // \* Leader i advances its commitIndex. // \* This is done as a separate step from handling AppendEntries responses, @@ -50,7 +51,8 @@ void syncMaybeAdvanceCommitIndex(SSyncNode* pSyncNode) { SyncIndex newCommitIndex = pSyncNode->commitIndex; for (SyncIndex index = pSyncNode->pLogStore->getLastIndex(pSyncNode->pLogStore); index > pSyncNode->commitIndex; ++index) { - if (syncAgree(pSyncNode, index)) { + bool agree = syncAgree(pSyncNode, index); + if (agree) { // term SSyncRaftEntry* pEntry = pSyncNode->pLogStore->getEntry(pSyncNode->pLogStore, index); assert(pEntry != NULL); @@ -97,14 +99,19 @@ void syncMaybeAdvanceCommitIndex(SSyncNode* pSyncNode) { } bool syncAgreeIndex(SSyncNode* pSyncNode, SRaftId* pRaftId, SyncIndex index) { - SyncIndex matchIndex = syncIndexMgrGetIndex(pSyncNode->pMatchIndex, pRaftId); - - // b for debug - bool b = false; - if (matchIndex >= index) { - b = true; + // I am leader, I agree + if (syncUtilSameId(pRaftId, &(pSyncNode->myRaftId)) && pSyncNode->state == TAOS_SYNC_STATE_LEADER) { + return true; } - return b; + + // follower agree + SyncIndex matchIndex = syncIndexMgrGetIndex(pSyncNode->pMatchIndex, pRaftId); + if (matchIndex >= index) { + return true; + } + + // not agree + return false; } bool syncAgree(SSyncNode* pSyncNode, SyncIndex index) { diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c index 6c2ef0c85b..6ed9229625 100644 --- a/source/libs/sync/src/syncMain.c +++ b/source/libs/sync/src/syncMain.c @@ -60,11 +60,14 @@ int64_t syncStart(const SSyncInfo* pSyncInfo) { int32_t ret = 0; SSyncNode* pSyncNode = syncNodeOpen(pSyncInfo); assert(pSyncNode != NULL); + + // todo : return ref id return ret; } void syncStop(int64_t rid) { - SSyncNode* pSyncNode = NULL; // get pointer from rid + // todo : get pointer from rid + SSyncNode* pSyncNode = NULL; syncNodeClose(pSyncNode); } @@ -74,8 +77,10 @@ int32_t syncReconfig(int64_t rid, const SSyncCfg* pSyncCfg) { } int32_t syncForwardToPeer(int64_t rid, const SRpcMsg* pMsg, bool isWeak) { - int32_t ret = 0; - SSyncNode* pSyncNode = NULL; // get pointer from rid + int32_t ret = 0; + + // todo : get pointer from rid + SSyncNode* pSyncNode = NULL; if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) { SyncClientRequest* pSyncMsg = syncClientRequestBuild2(pMsg, 0, isWeak); SRpcMsg rpcMsg; @@ -86,13 +91,14 @@ int32_t syncForwardToPeer(int64_t rid, const SRpcMsg* pMsg, bool isWeak) { } else { sTrace("syncForwardToPeer not leader, %s", syncUtilState2String(pSyncNode->state)); - ret = -1; // need define err code !! + ret = -1; // todo : need define err code !! } return ret; } ESyncState syncGetMyRole(int64_t rid) { - SSyncNode* pSyncNode = NULL; // get pointer from rid + // todo : get pointer from rid + SSyncNode* pSyncNode = NULL; return pSyncNode->state; } @@ -195,7 +201,7 @@ SSyncNode* syncNodeOpen(const SSyncInfo* pSyncInfo) { // init TLA+ log vars pSyncNode->pLogStore = logStoreCreate(pSyncNode); assert(pSyncNode->pLogStore != NULL); - pSyncNode->commitIndex = 0; + pSyncNode->commitIndex = SYNC_INDEX_INVALID; // init ping timer pSyncNode->pPingTimer = NULL; @@ -774,9 +780,6 @@ static int32_t syncNodeOnClientRequestCb(SSyncNode* ths, SyncClientRequest* pMsg if (ths->state == TAOS_SYNC_STATE_LEADER) { ths->pLogStore->appendEntry(ths->pLogStore, pEntry); - // only myself, maybe commit - syncMaybeAdvanceCommitIndex(ths); - // start replicate right now! syncNodeReplicate(ths); @@ -791,6 +794,9 @@ static int32_t syncNodeOnClientRequestCb(SSyncNode* ths, SyncClientRequest* pMsg } rpcFreeCont(rpcMsg.pCont); + // only myself, maybe commit + syncMaybeAdvanceCommitIndex(ths); + } else { // pre commit SRpcMsg rpcMsg; diff --git a/source/libs/sync/test/CMakeLists.txt b/source/libs/sync/test/CMakeLists.txt index 2821764448..0c7608f902 100644 --- a/source/libs/sync/test/CMakeLists.txt +++ b/source/libs/sync/test/CMakeLists.txt @@ -30,6 +30,7 @@ add_executable(syncPingSelfTest "") add_executable(syncElectTest "") add_executable(syncEncodeTest "") add_executable(syncWriteTest "") +add_executable(syncReplicateTest "") target_sources(syncTest @@ -160,6 +161,10 @@ target_sources(syncWriteTest PRIVATE "syncWriteTest.cpp" ) +target_sources(syncReplicateTest + PRIVATE + "syncReplicateTest.cpp" +) target_include_directories(syncTest @@ -327,6 +332,11 @@ target_include_directories(syncWriteTest "${CMAKE_SOURCE_DIR}/include/libs/sync" "${CMAKE_CURRENT_SOURCE_DIR}/../inc" ) +target_include_directories(syncReplicateTest + PUBLIC + "${CMAKE_SOURCE_DIR}/include/libs/sync" + "${CMAKE_CURRENT_SOURCE_DIR}/../inc" +) target_link_libraries(syncTest @@ -457,6 +467,10 @@ target_link_libraries(syncWriteTest sync gtest_main ) +target_link_libraries(syncReplicateTest + sync + gtest_main +) enable_testing() diff --git a/source/libs/sync/test/syncReplicateTest.cpp b/source/libs/sync/test/syncReplicateTest.cpp new file mode 100644 index 0000000000..502263cfbf --- /dev/null +++ b/source/libs/sync/test/syncReplicateTest.cpp @@ -0,0 +1,126 @@ +#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 = 3; +int32_t myIndex = 0; + +SRaftId ids[TSDB_MAX_REPLICA]; +SSyncInfo syncInfo; +SSyncFSM* pFsm; +SWal* pWal; +SSyncNode* gSyncNode; + +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), "./elect_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), "./elect_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); + } + + 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->FpOnSyncTimeout = pSyncNode->FpOnTimeout; + 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); + + gSyncNode = syncInitTest(); + assert(gSyncNode != NULL); + syncNodePrint2((char*)"", gSyncNode); + + initRaftId(gSyncNode); + + //--------------------------- + while (1) { + sTrace("while 1 sleep, state: %d, %s, electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, electTimerMS:%d", + gSyncNode->state, syncUtilState2String(gSyncNode->state), gSyncNode->electTimerLogicClock, + gSyncNode->electTimerLogicClockUser, gSyncNode->electTimerMS); + taosMsleep(1000); + } + + return 0; +}