xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

update sync vote & reply

This commit is contained in:
Minglei Jin 2024-07-25 12:15:00 +08:00
parent 970a5f31a3
commit 5caa366464
2 changed files with 15 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
source/libs/sync/src/syncRequestVote.c
View File

@ -20,7 +20,6 @@
#include "syncRaftStore.h" #include "syncRaftStore.h"
#include "syncUtil.h" #include "syncUtil.h"
#include "syncVoteMgr.h" #include "syncVoteMgr.h"
#include "syncUtil.h"
// TLA+ Spec // TLA+ Spec
// HandleRequestVoteRequest(i, j, m) == // HandleRequestVoteRequest(i, j, m) ==
@ -95,7 +94,8 @@ int32_t syncNodeOnRequestVote(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
// if already drop replica, do not process // if already drop replica, do not process
if (!syncNodeInRaftGroup(ths, &pMsg->srcId)) { if (!syncNodeInRaftGroup(ths, &pMsg->srcId)) {
syncLogRecvRequestVote(ths, pMsg, -1, "not in my config"); syncLogRecvRequestVote(ths, pMsg, -1, "not in my config");
return -1;
TAOS_RETURN(TSDB_CODE_FAILED);
} }
bool logOK = syncNodeOnRequestVoteLogOK(ths, pMsg); bool logOK = syncNodeOnRequestVoteLogOK(ths, pMsg);
@ -122,8 +122,8 @@ int32_t syncNodeOnRequestVote(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
// send msg // send msg
SRpcMsg rpcMsg = {0}; SRpcMsg rpcMsg = {0};
ret = syncBuildRequestVoteReply(&rpcMsg, ths->vgId);
ASSERT(ret == 0); TAOS_CHECK_RETURN(syncBuildRequestVoteReply(&rpcMsg, ths->vgId));
SyncRequestVoteReply* pReply = rpcMsg.pCont; SyncRequestVoteReply* pReply = rpcMsg.pCont;
pReply->srcId = ths->myRaftId; pReply->srcId = ths->myRaftId;
@ -138,5 +138,6 @@ int32_t syncNodeOnRequestVote(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
syncNodeSendMsgById(&pReply->destId, ths, &rpcMsg); syncNodeSendMsgById(&pReply->destId, ths, &rpcMsg);
if (resetElect) syncNodeResetElectTimer(ths); if (resetElect) syncNodeResetElectTimer(ths);
return 0;
TAOS_RETURN(TSDB_CODE_SUCCESS);
} }

14
source/libs/sync/src/syncRequestVoteReply.c
View File

@ -45,19 +45,22 @@ int32_t syncNodeOnRequestVoteReply(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
// if already drop replica, do not process // if already drop replica, do not process
if (!syncNodeInRaftGroup(ths, &(pMsg->srcId))) { if (!syncNodeInRaftGroup(ths, &(pMsg->srcId))) {
syncLogRecvRequestVoteReply(ths, pMsg, "not in my config"); syncLogRecvRequestVoteReply(ths, pMsg, "not in my config");
return -1;
TAOS_RETURN(TSDB_CODE_FAILED);
} }
SyncTerm currentTerm = raftStoreGetTerm(ths); SyncTerm currentTerm = raftStoreGetTerm(ths);
// drop stale response // drop stale response
if (pMsg->term < currentTerm) { if (pMsg->term < currentTerm) {
syncLogRecvRequestVoteReply(ths, pMsg, "drop stale response"); syncLogRecvRequestVoteReply(ths, pMsg, "drop stale response");
return -1;
TAOS_RETURN(TSDB_CODE_FAILED);
} }
if (pMsg->term > currentTerm) { if (pMsg->term > currentTerm) {
syncLogRecvRequestVoteReply(ths, pMsg, "error term"); syncLogRecvRequestVoteReply(ths, pMsg, "error term");
syncNodeStepDown(ths, pMsg->term); syncNodeStepDown(ths, pMsg->term);
return -1;
TAOS_RETURN(TSDB_CODE_FAILED);
} }
syncLogRecvRequestVoteReply(ths, pMsg, ""); syncLogRecvRequestVoteReply(ths, pMsg, "");
@ -69,7 +72,8 @@ int32_t syncNodeOnRequestVoteReply(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
if (ths->pVotesRespond->term != pMsg->term) { if (ths->pVotesRespond->term != pMsg->term) {
sNError(ths, "vote respond error vote-respond-mgr term:%" PRIu64 ", msg term:%" PRIu64 "", sNError(ths, "vote respond error vote-respond-mgr term:%" PRIu64 ", msg term:%" PRIu64 "",
ths->pVotesRespond->term, pMsg->term); ths->pVotesRespond->term, pMsg->term);
return -1;
TAOS_RETURN(TSDB_CODE_FAILED);
} }
votesRespondAdd(ths->pVotesRespond, pMsg); votesRespondAdd(ths->pVotesRespond, pMsg);
@ -93,5 +97,5 @@ int32_t syncNodeOnRequestVoteReply(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
} }
} }
return 0; TAOS_RETURN(TSDB_CODE_SUCCESS);
} }