diff --git a/source/libs/sync/src/syncAppendEntries.c b/source/libs/sync/src/syncAppendEntries.c index 270180e3477b4c6359feb07a918de56ac097fcef..e4df93ca47c013348289f057a8a10063e6e6edfe 100644 --- a/source/libs/sync/src/syncAppendEntries.c +++ b/source/libs/sync/src/syncAppendEntries.c @@ -87,7 +87,10 @@ // int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) { int32_t ret = 0; - syncAppendEntriesLog2("==syncNodeOnAppendEntriesCb==", pMsg); + + char logBuf[128]; + snprintf(logBuf, sizeof(logBuf), "==syncNodeOnAppendEntriesCb== term:%lu", ths->pRaftStore->currentTerm); + syncAppendEntriesLog2(logBuf, pMsg); if (pMsg->term > ths->pRaftStore->currentTerm) { syncNodeUpdateTerm(ths, pMsg->term); diff --git a/source/libs/sync/src/syncAppendEntriesReply.c b/source/libs/sync/src/syncAppendEntriesReply.c index 9db9a3e8ac532790dbcffff40f80b5950adc983c..817974fd268fa4105c45d51c5b591a79167971dc 100644 --- a/source/libs/sync/src/syncAppendEntriesReply.c +++ b/source/libs/sync/src/syncAppendEntriesReply.c @@ -37,7 +37,10 @@ // int32_t syncNodeOnAppendEntriesReplyCb(SSyncNode* ths, SyncAppendEntriesReply* pMsg) { int32_t ret = 0; - syncAppendEntriesReplyLog2("==syncNodeOnAppendEntriesReplyCb==", pMsg); + + char logBuf[128]; + snprintf(logBuf, sizeof(logBuf), "==syncNodeOnAppendEntriesReplyCb== term:%lu", ths->pRaftStore->currentTerm); + syncAppendEntriesReplyLog2(logBuf, pMsg); if (pMsg->term < ths->pRaftStore->currentTerm) { sTrace("DropStaleResponse, receive term:%" PRIu64 ", current term:%" PRIu64 "", pMsg->term, diff --git a/source/libs/sync/src/syncCommit.c b/source/libs/sync/src/syncCommit.c index 38b39e04cf06b4dfbb6cc1f1da15756f7b50affe..0d4df8e6cf43aecce88b71145f1f0f4c1fe49744 100644 --- a/source/libs/sync/src/syncCommit.c +++ b/source/libs/sync/src/syncCommit.c @@ -50,8 +50,10 @@ void syncMaybeAdvanceCommitIndex(SSyncNode* pSyncNode) { // update commit index SyncIndex newCommitIndex = pSyncNode->commitIndex; for (SyncIndex index = pSyncNode->pLogStore->getLastIndex(pSyncNode->pLogStore); index > pSyncNode->commitIndex; - ++index) { + --index) { bool agree = syncAgree(pSyncNode, index); + sTrace("syncMaybeAdvanceCommitIndex syncAgree:%d, index:%ld, pSyncNode->commitIndex:%ld", agree, index, + pSyncNode->commitIndex); if (agree) { // term SSyncRaftEntry* pEntry = pSyncNode->pLogStore->getEntry(pSyncNode->pLogStore, index); @@ -70,6 +72,8 @@ void syncMaybeAdvanceCommitIndex(SSyncNode* pSyncNode) { SyncIndex beginIndex = pSyncNode->commitIndex + 1; SyncIndex endIndex = newCommitIndex; + sTrace("syncMaybeAdvanceCommitIndex sync commit %ld", newCommitIndex); + // update commit index pSyncNode->commitIndex = newCommitIndex; diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c index 6ed922962540d6d2327978a39335ef9bcb5ac2bd..829526512b4403bf88716910902faf2c6bfb3464 100644 --- a/source/libs/sync/src/syncMain.c +++ b/source/libs/sync/src/syncMain.c @@ -804,7 +804,7 @@ static int32_t syncNodeOnClientRequestCb(SSyncNode* ths, SyncClientRequest* pMsg if (ths->pFsm != NULL) { if (ths->pFsm->FpPreCommitCb != NULL) { - ths->pFsm->FpPreCommitCb(ths->pFsm, &rpcMsg, pEntry->index, pEntry->isWeak, -1); + ths->pFsm->FpPreCommitCb(ths->pFsm, &rpcMsg, pEntry->index, pEntry->isWeak, -2); } } rpcFreeCont(rpcMsg.pCont); diff --git a/source/libs/sync/src/syncRequestVote.c b/source/libs/sync/src/syncRequestVote.c index 062b0244bdd54ce611d7323d5e2115380f334733..e23748a81e7e8f8622c31d9bf98b381727a44c39 100644 --- a/source/libs/sync/src/syncRequestVote.c +++ b/source/libs/sync/src/syncRequestVote.c @@ -43,7 +43,10 @@ // int32_t syncNodeOnRequestVoteCb(SSyncNode* ths, SyncRequestVote* pMsg) { int32_t ret = 0; - syncRequestVoteLog2("==syncNodeOnRequestVoteCb==", pMsg); + + char logBuf[128]; + snprintf(logBuf, sizeof(logBuf), "==syncNodeOnRequestVoteCb== term:%lu", ths->pRaftStore->currentTerm); + syncRequestVoteLog2(logBuf, pMsg); if (pMsg->term > ths->pRaftStore->currentTerm) { syncNodeUpdateTerm(ths, pMsg->term); diff --git a/source/libs/sync/src/syncRequestVoteReply.c b/source/libs/sync/src/syncRequestVoteReply.c index 12603bb337ce6791599bd6755756ff3ef2e090a7..1531f701ff7605543c6c5abaabcbb1d88d894528 100644 --- a/source/libs/sync/src/syncRequestVoteReply.c +++ b/source/libs/sync/src/syncRequestVoteReply.c @@ -38,7 +38,10 @@ // int32_t syncNodeOnRequestVoteReplyCb(SSyncNode* ths, SyncRequestVoteReply* pMsg) { int32_t ret = 0; - syncRequestVoteReplyLog2("==syncNodeOnRequestVoteReplyCb==", pMsg); + + char logBuf[128]; + snprintf(logBuf, sizeof(logBuf), "==syncNodeOnRequestVoteReplyCb== term:%lu", ths->pRaftStore->currentTerm); + syncRequestVoteReplyLog2(logBuf, pMsg); if (pMsg->term < ths->pRaftStore->currentTerm) { sTrace("DropStaleResponse, receive term:%" PRIu64 ", current term:%" PRIu64 "", pMsg->term, diff --git a/source/libs/sync/test/syncElectTest.cpp b/source/libs/sync/test/syncElectTest.cpp index 502263cfbf736758298961211db6eda5876de2a3..0e0b57a0257001c717e3a0f23de5fbf2cac31cb7 100644 --- a/source/libs/sync/test/syncElectTest.cpp +++ b/source/libs/sync/test/syncElectTest.cpp @@ -116,10 +116,9 @@ int main(int argc, char** argv) { //--------------------------- while (1) { - sTrace("while 1 sleep, state: %d, %s, electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, electTimerMS:%d", - gSyncNode->state, syncUtilState2String(gSyncNode->state), gSyncNode->electTimerLogicClock, + sTrace("elect sleep, state: %d, %s, term:%lu electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, electTimerMS:%d", + gSyncNode->state, syncUtilState2String(gSyncNode->state), gSyncNode->pRaftStore->currentTerm, gSyncNode->electTimerLogicClock, gSyncNode->electTimerLogicClockUser, gSyncNode->electTimerMS); - taosMsleep(1000); } return 0; diff --git a/source/libs/sync/test/syncReplicateTest.cpp b/source/libs/sync/test/syncReplicateTest.cpp index 502263cfbf736758298961211db6eda5876de2a3..6c4fab24252656a3f8079465f1b07d192420a537 100644 --- a/source/libs/sync/test/syncReplicateTest.cpp +++ b/source/libs/sync/test/syncReplicateTest.cpp @@ -3,6 +3,8 @@ #include "syncEnv.h" #include "syncIO.h" #include "syncInt.h" +#include "syncMessage.h" +#include "syncRaftEntry.h" #include "syncRaftLog.h" #include "syncRaftStore.h" #include "syncUtil.h" @@ -22,18 +24,40 @@ int32_t myIndex = 0; SRaftId ids[TSDB_MAX_REPLICA]; SSyncInfo syncInfo; -SSyncFSM* pFsm; -SWal* pWal; -SSyncNode* gSyncNode; +SSyncFSM * pFsm; +SWal * pWal; +SSyncNode *gSyncNode; -SSyncNode* syncNodeInit() { +void CommitCb(struct SSyncFSM *pFsm, const SRpcMsg *pBuf, SyncIndex index, bool isWeak, int32_t code) { + printf("==CommitCb== pFsm:%p, index:%ld, isWeak:%d, code:%d \n", pFsm, index, isWeak, code); + syncRpcMsgPrint2((char *)"==CommitCb==", (SRpcMsg *)pBuf); +} + +void PreCommitCb(struct SSyncFSM *pFsm, const SRpcMsg *pBuf, SyncIndex index, bool isWeak, int32_t code) { + printf("==PreCommitCb== pFsm:%p, index:%ld, isWeak:%d, code:%d \n", pFsm, index, isWeak, code); + syncRpcMsgPrint2((char *)"==PreCommitCb==", (SRpcMsg *)pBuf); +} + +void RollBackCb(struct SSyncFSM *pFsm, const SRpcMsg *pBuf, SyncIndex index, bool isWeak, int32_t code) { + printf("==RollBackCb== pFsm:%p, index:%ld, isWeak:%d, code:%d \n", pFsm, index, isWeak, code); + syncRpcMsgPrint2((char *)"==RollBackCb==", (SRpcMsg *)pBuf); +} + +void initFsm() { + pFsm = (SSyncFSM *)malloc(sizeof(SSyncFSM)); + pFsm->FpCommitCb = CommitCb; + pFsm->FpPreCommitCb = PreCommitCb; + pFsm->FpRollBackCb = RollBackCb; +} + +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); + snprintf(syncInfo.path, sizeof(syncInfo.path), "./replicate_test_%d", myIndex); int code = walInit(); assert(code == 0); @@ -48,13 +72,13 @@ SSyncNode* syncNodeInit() { walCfg.level = TAOS_WAL_FSYNC; char tmpdir[128]; - snprintf(tmpdir, sizeof(tmpdir), "./elect_test_wal_%d", myIndex); + snprintf(tmpdir, sizeof(tmpdir), "./replicate_test_wal_%d", myIndex); pWal = walOpen(tmpdir, &walCfg); assert(pWal != NULL); syncInfo.pWal = pWal; - SSyncCfg* pCfg = &syncInfo.syncCfg; + SSyncCfg *pCfg = &syncInfo.syncCfg; pCfg->myIndex = myIndex; pCfg->replicaNum = replicaNum; @@ -64,60 +88,98 @@ SSyncNode* syncNodeInit() { // taosGetFqdn(pCfg->nodeInfo[0].nodeFqdn); } - SSyncNode* pSyncNode = syncNodeOpen(&syncInfo); + SSyncNode *pSyncNode = syncNodeOpen(&syncInfo); assert(pSyncNode != NULL); gSyncIO->FpOnSyncPing = pSyncNode->FpOnPing; + gSyncIO->FpOnSyncClientRequest = pSyncNode->FpOnClientRequest; 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(); } +SSyncNode *syncInitTest() { return syncNodeInit(); } -void initRaftId(SSyncNode* pSyncNode) { +void initRaftId(SSyncNode *pSyncNode) { for (int i = 0; i < replicaNum; ++i) { ids[i] = pSyncNode->replicasId[i]; - char* s = syncUtilRaftId2Str(&ids[i]); + char *s = syncUtilRaftId2Str(&ids[i]); printf("raftId[%d] : %s\n", i, s); free(s); } } -int main(int argc, char** argv) { +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]); + int32_t ret = syncIOStart((char *)"127.0.0.1", ports[myIndex]); assert(ret == 0); ret = syncEnvStart(); assert(ret == 0); + taosRemoveDir("./wal_test"); + + initFsm(); + gSyncNode = syncInitTest(); assert(gSyncNode != NULL); - syncNodePrint2((char*)"", gSyncNode); + syncNodePrint2((char *)"", gSyncNode); initRaftId(gSyncNode); - //--------------------------- + for (int i = 0; i < 30; ++i) { + // step0 + SRpcMsg *pMsg0 = step0(i); + syncRpcMsgPrint2((char *)"==step0==", pMsg0); + + // step1 + SyncClientRequest *pMsg1 = step1(pMsg0); + syncClientRequestPrint2((char *)"==step1==", pMsg1); + + SyncClientRequest *pSyncClientRequest = pMsg1; + SRpcMsg rpcMsg; + syncClientRequest2RpcMsg(pSyncClientRequest, &rpcMsg); + gSyncNode->FpEqMsg(gSyncNode->queue, &rpcMsg); + + taosMsleep(1000); + sTrace("replicate sleep, state: %d, %s, term:%lu electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, electTimerMS:%d", + gSyncNode->state, syncUtilState2String(gSyncNode->state), gSyncNode->pRaftStore->currentTerm, gSyncNode->electTimerLogicClock, + gSyncNode->electTimerLogicClockUser, gSyncNode->electTimerMS); + } + while (1) { - sTrace("while 1 sleep, state: %d, %s, electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, electTimerMS:%d", - gSyncNode->state, syncUtilState2String(gSyncNode->state), gSyncNode->electTimerLogicClock, + sTrace("replicate sleep, state: %d, %s, term:%lu electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, electTimerMS:%d", + gSyncNode->state, syncUtilState2String(gSyncNode->state), gSyncNode->pRaftStore->currentTerm, gSyncNode->electTimerLogicClock, gSyncNode->electTimerLogicClockUser, gSyncNode->electTimerMS); taosMsleep(1000); }