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

Merge pull request #19242 from taosdata/fix/TS-2349 

refact: adjust sync log
This commit is contained in:
Shengliang Guan 2022-12-29 13:22:45 +08:00 committed by GitHub
parent 4a142018da b05c65cd47
commit 1f9786886b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 147 additions and 152 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
source/libs/sync/inc/syncIndexMgr.h
View File

@ -27,27 +27,25 @@ typedef struct SSyncIndexMgr {
SRaftId (*replicas)[TSDB_MAX_REPLICA];
SyncIndex index[TSDB_MAX_REPLICA];
SyncTerm privateTerm[TSDB_MAX_REPLICA]; // for advanced function
int64_t startTimeArr[TSDB_MAX_REPLICA];
int64_t recvTimeArr[TSDB_MAX_REPLICA];
int32_t replicaNum;
SSyncNode *pSyncNode;
SSyncNode *pNode;
} SSyncIndexMgr;
SSyncIndexMgr *syncIndexMgrCreate(SSyncNode *pSyncNode);
void syncIndexMgrUpdate(SSyncIndexMgr *pSyncIndexMgr, SSyncNode *pSyncNode);
void syncIndexMgrDestroy(SSyncIndexMgr *pSyncIndexMgr);
void syncIndexMgrClear(SSyncIndexMgr *pSyncIndexMgr);
void syncIndexMgrSetIndex(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId, SyncIndex index);
SyncIndex syncIndexMgrGetIndex(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId);
SSyncIndexMgr *syncIndexMgrCreate(SSyncNode *pNode);
void syncIndexMgrUpdate(SSyncIndexMgr *pIndexMgr, SSyncNode *pNode);
void syncIndexMgrDestroy(SSyncIndexMgr *pIndexMgr);
void syncIndexMgrClear(SSyncIndexMgr *pIndexMgr);
void syncIndexMgrSetIndex(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId, SyncIndex index);
SyncIndex syncIndexMgrGetIndex(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId);
void syncIndexMgrSetStartTime(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId, int64_t startTime);
int64_t syncIndexMgrGetStartTime(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId);
void syncIndexMgrSetRecvTime(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId, int64_t recvTime);
int64_t syncIndexMgrGetRecvTime(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId);
void syncIndexMgrSetTerm(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId, SyncTerm term);
SyncTerm syncIndexMgrGetTerm(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId);
void syncIndexMgrSetStartTime(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId, int64_t startTime);
int64_t syncIndexMgrGetStartTime(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId);
void syncIndexMgrSetRecvTime(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId, int64_t recvTime);
int64_t syncIndexMgrGetRecvTime(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId);
void syncIndexMgrSetTerm(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId, SyncTerm term);
SyncTerm syncIndexMgrGetTerm(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId);
#ifdef __cplusplus
}

12
source/libs/sync/inc/syncVoteMgr.h
View File

@ -30,12 +30,12 @@ typedef struct SVotesGranted {
SyncTerm term;
int32_t quorum;
bool toLeader;
SSyncNode *pSyncNode;
SSyncNode *pNode;
} SVotesGranted;
SVotesGranted *voteGrantedCreate(SSyncNode *pSyncNode);
SVotesGranted *voteGrantedCreate(SSyncNode *pNode);
void voteGrantedDestroy(SVotesGranted *pVotesGranted);
void voteGrantedUpdate(SVotesGranted *pVotesGranted, SSyncNode *pSyncNode);
void voteGrantedUpdate(SVotesGranted *pVotesGranted, SSyncNode *pNode);
bool voteGrantedMajority(SVotesGranted *pVotesGranted);
void voteGrantedVote(SVotesGranted *pVotesGranted, SyncRequestVoteReply *pMsg);
void voteGrantedReset(SVotesGranted *pVotesGranted, SyncTerm term);
@ -45,12 +45,12 @@ typedef struct SVotesRespond {
bool isRespond[TSDB_MAX_REPLICA];
int32_t replicaNum;
SyncTerm term;
SSyncNode *pSyncNode;
SSyncNode *pNode;
} SVotesRespond;
SVotesRespond *votesRespondCreate(SSyncNode *pSyncNode);
SVotesRespond *votesRespondCreate(SSyncNode *pNode);
void votesRespondDestory(SVotesRespond *pVotesRespond);
void votesRespondUpdate(SVotesRespond *pVotesRespond, SSyncNode *pSyncNode);
void votesRespondUpdate(SVotesRespond *pVotesRespond, SSyncNode *pNode);
bool votesResponded(SVotesRespond *pVotesRespond, const SRaftId *pRaftId);
void votesRespondAdd(SVotesRespond *pVotesRespond, const SyncRequestVoteReply *pMsg);
void votesRespondReset(SVotesRespond *pVotesRespond, SyncTerm term);

179
source/libs/sync/src/syncIndexMgr.c
View File

@ -17,173 +17,172 @@
#include "syncIndexMgr.h"
#include "syncUtil.h"
SSyncIndexMgr *syncIndexMgrCreate(SSyncNode *pSyncNode) {
SSyncIndexMgr *pSyncIndexMgr = taosMemoryCalloc(1, sizeof(SSyncIndexMgr));
if (pSyncIndexMgr == NULL) {
SSyncIndexMgr *syncIndexMgrCreate(SSyncNode *pNode) {
SSyncIndexMgr *pIndexMgr = taosMemoryCalloc(1, sizeof(SSyncIndexMgr));
if (pIndexMgr == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
}
pSyncIndexMgr->replicas = &(pSyncNode->replicasId);
pSyncIndexMgr->replicaNum = pSyncNode->replicaNum;
pSyncIndexMgr->pSyncNode = pSyncNode;
syncIndexMgrClear(pSyncIndexMgr);
pIndexMgr->replicas = &pNode->replicasId;
pIndexMgr->replicaNum = pNode->replicaNum;
pIndexMgr->pNode = pNode;
syncIndexMgrClear(pIndexMgr);
return pSyncIndexMgr;
return pIndexMgr;
}
void syncIndexMgrUpdate(SSyncIndexMgr *pSyncIndexMgr, SSyncNode *pSyncNode) {
pSyncIndexMgr->replicas = &(pSyncNode->replicasId);
pSyncIndexMgr->replicaNum = pSyncNode->replicaNum;
pSyncIndexMgr->pSyncNode = pSyncNode;
syncIndexMgrClear(pSyncIndexMgr);
void syncIndexMgrUpdate(SSyncIndexMgr *pIndexMgr, SSyncNode *pNode) {
pIndexMgr->replicas = &pNode->replicasId;
pIndexMgr->replicaNum = pNode->replicaNum;
pIndexMgr->pNode = pNode;
syncIndexMgrClear(pIndexMgr);
}
void syncIndexMgrDestroy(SSyncIndexMgr *pSyncIndexMgr) {
if (pSyncIndexMgr != NULL) {
taosMemoryFree(pSyncIndexMgr);
void syncIndexMgrDestroy(SSyncIndexMgr *pIndexMgr) {
if (pIndexMgr != NULL) {
taosMemoryFree(pIndexMgr);
}
}
void syncIndexMgrClear(SSyncIndexMgr *pSyncIndexMgr) {
memset(pSyncIndexMgr->index, 0, sizeof(pSyncIndexMgr->index));
memset(pSyncIndexMgr->privateTerm, 0, sizeof(pSyncIndexMgr->privateTerm));
void syncIndexMgrClear(SSyncIndexMgr *pIndexMgr) {
memset(pIndexMgr->index, 0, sizeof(pIndexMgr->index));
memset(pIndexMgr->privateTerm, 0, sizeof(pIndexMgr->privateTerm));
// int64_t timeNow = taosGetMonotonicMs();
int64_t timeNow = taosGetTimestampMs();
for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) {
pSyncIndexMgr->startTimeArr[i] = 0;
pSyncIndexMgr->recvTimeArr[i] = timeNow;
for (int i = 0; i < pIndexMgr->replicaNum; ++i) {
pIndexMgr->startTimeArr[i] = 0;
pIndexMgr->recvTimeArr[i] = timeNow;
}
/*
for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) {
pSyncIndexMgr->index[i] = 0;
}
*/
}
void syncIndexMgrSetIndex(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId, SyncIndex index) {
for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) {
if (syncUtilSameId(&((*(pSyncIndexMgr->replicas))[i]), pRaftId)) {
(pSyncIndexMgr->index)[i] = index;
void syncIndexMgrSetIndex(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId, SyncIndex index) {
for (int i = 0; i < pIndexMgr->replicaNum; ++i) {
if (syncUtilSameId(&((*(pIndexMgr->replicas))[i]), pRaftId)) {
(pIndexMgr->index)[i] = index;
return;
}
}
// maybe config change
// ASSERT(0);
char host[128];
uint16_t port;
syncUtilU642Addr(pRaftId->addr, host, sizeof(host), &port);
sError("vgId:%d, indexmgr set index:%" PRId64 " for %s:%d failed", pIndexMgr->pNode->vgId, index, host, port);
}
SSyncLogReplMgr *syncNodeGetLogReplMgr(SSyncNode *pNode, SRaftId *pRaftId) {
for (int i = 0; i < pNode->replicaNum; i++) {
if (syncUtilSameId(&pNode->replicasId[i], pRaftId)) {
return pNode->logReplMgrs[i];
}
}
char host[128];
uint16_t port;
syncUtilU642Addr(pRaftId->addr, host, sizeof(host), &port);
sError("vgId:%d, index mgr set for %s:%d, index:%" PRId64 " error", pSyncIndexMgr->pSyncNode->vgId, host, port,
index);
}
SSyncLogReplMgr *syncNodeGetLogReplMgr(SSyncNode *pNode, SRaftId *pDestId) {
for (int i = 0; i < pNode->replicaNum; i++) {
if (syncUtilSameId(&(pNode->replicasId[i]), pDestId)) {
return pNode->logReplMgrs[i];
}
}
sError("vgId:%d, indexmgr get replmgr from %s:%d failed", pNode->vgId, host, port);
return NULL;
}
SyncIndex syncIndexMgrGetIndex(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId) {
if (pSyncIndexMgr == NULL) {
return SYNC_INDEX_INVALID;
}
for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) {
if (syncUtilSameId(&((*(pSyncIndexMgr->replicas))[i]), pRaftId)) {
SyncIndex idx = (pSyncIndexMgr->index)[i];
SyncIndex syncIndexMgrGetIndex(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId) {
for (int i = 0; i < pIndexMgr->replicaNum; ++i) {
if (syncUtilSameId(&((*(pIndexMgr->replicas))[i]), pRaftId)) {
SyncIndex idx = (pIndexMgr->index)[i];
return idx;
}
}
char host[128];
uint16_t port;
syncUtilU642Addr(pRaftId->addr, host, sizeof(host), &port);
sError("vgId:%d, indexmgr get index from %s:%d failed", pIndexMgr->pNode->vgId, host, port);
return SYNC_INDEX_INVALID;
}
void syncIndexMgrSetStartTime(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId, int64_t startTime) {
for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) {
if (syncUtilSameId(&((*(pSyncIndexMgr->replicas))[i]), pRaftId)) {
(pSyncIndexMgr->startTimeArr)[i] = startTime;
void syncIndexMgrSetStartTime(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId, int64_t startTime) {
for (int i = 0; i < pIndexMgr->replicaNum; ++i) {
if (syncUtilSameId(&((*(pIndexMgr->replicas))[i]), pRaftId)) {
(pIndexMgr->startTimeArr)[i] = startTime;
return;
}
}
// maybe config change
// ASSERT(0);
char host[128];
uint16_t port;
syncUtilU642Addr(pRaftId->addr, host, sizeof(host), &port);
sError("vgId:%d, index mgr set for %s:%d, start-time:%" PRId64 " error", pSyncIndexMgr->pSyncNode->vgId, host, port,
startTime);
sError("vgId:%d, indexmgr set start-time:%" PRId64 " for %s:%d failed", pIndexMgr->pNode->vgId, startTime, host,
port);
}
int64_t syncIndexMgrGetStartTime(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId) {
for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) {
if (syncUtilSameId(&((*(pSyncIndexMgr->replicas))[i]), pRaftId)) {
int64_t startTime = (pSyncIndexMgr->startTimeArr)[i];
int64_t syncIndexMgrGetStartTime(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId) {
for (int i = 0; i < pIndexMgr->replicaNum; ++i) {
if (syncUtilSameId(&((*(pIndexMgr->replicas))[i]), pRaftId)) {
int64_t startTime = (pIndexMgr->startTimeArr)[i];
return startTime;
}
}
ASSERT(0);
char host[128];
uint16_t port;
syncUtilU642Addr(pRaftId->addr, host, sizeof(host), &port);
sError("vgId:%d, indexmgr get start-time from %s:%d failed", pIndexMgr->pNode->vgId, host, port);
return -1;
}
void syncIndexMgrSetRecvTime(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId, int64_t recvTime) {
for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) {
if (syncUtilSameId(&((*(pSyncIndexMgr->replicas))[i]), pRaftId)) {
(pSyncIndexMgr->recvTimeArr)[i] = recvTime;
void syncIndexMgrSetRecvTime(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId, int64_t recvTime) {
for (int i = 0; i < pIndexMgr->replicaNum; ++i) {
if (syncUtilSameId(&((*(pIndexMgr->replicas))[i]), pRaftId)) {
(pIndexMgr->recvTimeArr)[i] = recvTime;
return;
}
}
// maybe config change
// ASSERT(0);
char host[128];
uint16_t port;
syncUtilU642Addr(pRaftId->addr, host, sizeof(host), &port);
sError("vgId:%d, index mgr set for %s:%d, recv-time:%" PRId64 " error", pSyncIndexMgr->pSyncNode->vgId, host, port,
recvTime);
sError("vgId:%d, indexmgr set recv-time:%" PRId64 " for %s:%d failed", pIndexMgr->pNode->vgId, recvTime, host, port);
}
int64_t syncIndexMgrGetRecvTime(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId) {
for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) {
if (syncUtilSameId(&((*(pSyncIndexMgr->replicas))[i]), pRaftId)) {
int64_t recvTime = (pSyncIndexMgr->recvTimeArr)[i];
int64_t syncIndexMgrGetRecvTime(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId) {
for (int i = 0; i < pIndexMgr->replicaNum; ++i) {
if (syncUtilSameId(&((*(pIndexMgr->replicas))[i]), pRaftId)) {
int64_t recvTime = (pIndexMgr->recvTimeArr)[i];
return recvTime;
}
}
char host[128];
uint16_t port;
syncUtilU642Addr(pRaftId->addr, host, sizeof(host), &port);
sError("vgId:%d, indexmgr get recv-time from %s:%d failed", pIndexMgr->pNode->vgId, host, port);
return -1;
}
void syncIndexMgrSetTerm(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId, SyncTerm term) {
for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) {
if (syncUtilSameId(&((*(pSyncIndexMgr->replicas))[i]), pRaftId)) {
(pSyncIndexMgr->privateTerm)[i] = term;
void syncIndexMgrSetTerm(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId, SyncTerm term) {
for (int i = 0; i < pIndexMgr->replicaNum; ++i) {
if (syncUtilSameId(&((*(pIndexMgr->replicas))[i]), pRaftId)) {
(pIndexMgr->privateTerm)[i] = term;
return;
}
}
// maybe config change
// ASSERT(0);
char host[128];
uint16_t port;
syncUtilU642Addr(pRaftId->addr, host, sizeof(host), &port);
sError("vgId:%d, index mgr set for %s:%d, term:%" PRIu64 " error", pSyncIndexMgr->pSyncNode->vgId, host, port, term);
sError("vgId:%d, indexmgr set term:%" PRId64 " for %s:%d failed", pIndexMgr->pNode->vgId, term, host, port);
}
SyncTerm syncIndexMgrGetTerm(SSyncIndexMgr *pSyncIndexMgr, const SRaftId *pRaftId) {
for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) {
if (syncUtilSameId(&((*(pSyncIndexMgr->replicas))[i]), pRaftId)) {
SyncTerm term = (pSyncIndexMgr->privateTerm)[i];
SyncTerm syncIndexMgrGetTerm(SSyncIndexMgr *pIndexMgr, const SRaftId *pRaftId) {
for (int i = 0; i < pIndexMgr->replicaNum; ++i) {
if (syncUtilSameId(&((*(pIndexMgr->replicas))[i]), pRaftId)) {
SyncTerm term = (pIndexMgr->privateTerm)[i];
return term;
}
}
ASSERT(0);
char host[128];
uint16_t port;
syncUtilU642Addr(pRaftId->addr, host, sizeof(host), &port);
sError("vgId:%d, indexmgr get term from %s:%d failed", pIndexMgr->pNode->vgId, host, port);
return -1;
}

16
source/libs/sync/src/syncRespMgr.c
View File

@ -118,12 +118,12 @@ static void syncRespCleanByTTL(SSyncRespMgr *pObj, int64_t ttl, bool rsp) {
SRespStub *pStub = (SRespStub *)taosHashIterate(pObj->pRespHash, NULL);
int cnt = 0;
int sum = 0;
SSyncNode *pSyncNode = pObj->data;
SSyncNode *pNode = pObj->data;
SArray *delIndexArray = taosArrayInit(4, sizeof(uint64_t));
if (delIndexArray == NULL) return;
sDebug("vgId:%d, resp manager begin clean by ttl", pSyncNode->vgId);
sDebug("vgId:%d, resp manager begin clean by ttl", pNode->vgId);
while (pStub) {
size_t len;
void *key = taosHashGetKey(pStub, &len);
@ -140,20 +140,18 @@ static void syncRespCleanByTTL(SSyncRespMgr *pObj, int64_t ttl, bool rsp) {
.lastConfigIndex = SYNC_INDEX_INVALID,
.isWeak = false,
.code = TSDB_CODE_SYN_TIMEOUT,
.state = pSyncNode->state,
.state = pNode->state,
.seqNum = *pSeqNum,
.term = SYNC_TERM_INVALID,
.currentTerm = pSyncNode->pRaftStore->currentTerm,
.currentTerm = pNode->pRaftStore->currentTerm,
.flag = 0,
};
pStub->rpcMsg.pCont = NULL;
pStub->rpcMsg.contLen = 0;
// TODO: and make rpcMsg body, call commit cb
// pSyncNode->pFsm->FpCommitCb(pSyncNode->pFsm, &pStub->rpcMsg, cbMeta);
SRpcMsg rpcMsg = {.info = pStub->rpcMsg.info, .code = TSDB_CODE_SYN_TIMEOUT};
sInfo("vgId:%d, message handle:%p expired, type:%s ahandle:%p", pSyncNode->vgId, rpcMsg.info.handle,
sInfo("vgId:%d, message handle:%p expired, type:%s ahandle:%p", pNode->vgId, rpcMsg.info.handle,
TMSG_INFO(pStub->rpcMsg.msgType), rpcMsg.info.ahandle);
rpcSendResponse(&rpcMsg);
}
@ -162,12 +160,12 @@ static void syncRespCleanByTTL(SSyncRespMgr *pObj, int64_t ttl, bool rsp) {
}
int32_t arraySize = taosArrayGetSize(delIndexArray);
sDebug("vgId:%d, resp manager end clean by ttl, sum:%d, cnt:%d, array-size:%d", pSyncNode->vgId, sum, cnt, arraySize);
sDebug("vgId:%d, resp manager end clean by ttl, sum:%d, cnt:%d, array-size:%d", pNode->vgId, sum, cnt, arraySize);
for (int32_t i = 0; i < arraySize; ++i) {
uint64_t *pSeqNum = taosArrayGet(delIndexArray, i);
taosHashRemove(pObj->pRespHash, pSeqNum, sizeof(uint64_t));
sDebug("vgId:%d, resp manager clean by ttl, seq:%" PRId64, pSyncNode->vgId, *pSeqNum);
sDebug("vgId:%d, resp manager clean by ttl, seq:%" PRId64, pNode->vgId, *pSeqNum);
}
taosArrayDestroy(delIndexArray);
}

2
source/libs/sync/src/syncSnapshot.c
View File

@ -406,7 +406,7 @@ void snapshotReceiverStart(SSyncSnapshotReceiver *pReceiver, SyncSnapshotSend *p
}
// just set start = false
// FpSnapshotStopWrite should not be called, assert writer == NULL
// FpSnapshotStopWrite should not be called
void snapshotReceiverStop(SSyncSnapshotReceiver *pReceiver) {
sRInfo(pReceiver, "snapshot receiver stop, not apply, writer:%p", pReceiver->pWriter);

54
source/libs/sync/src/syncVoteMgr.c
View File

@ -23,21 +23,21 @@ static void voteGrantedClearVotes(SVotesGranted *pVotesGranted) {
pVotesGranted->votes = 0;
}
SVotesGranted *voteGrantedCreate(SSyncNode *pSyncNode) {
SVotesGranted *voteGrantedCreate(SSyncNode *pNode) {
SVotesGranted *pVotesGranted = taosMemoryCalloc(1, sizeof(SVotesGranted));
if (pVotesGranted == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
}
pVotesGranted->replicas = &(pSyncNode->replicasId);
pVotesGranted->replicaNum = pSyncNode->replicaNum;
pVotesGranted->replicas = &pNode->replicasId;
pVotesGranted->replicaNum = pNode->replicaNum;
voteGrantedClearVotes(pVotesGranted);
pVotesGranted->term = 0;
pVotesGranted->quorum = pSyncNode->quorum;
pVotesGranted->quorum = pNode->quorum;
pVotesGranted->toLeader = false;
pVotesGranted->pSyncNode = pSyncNode;
pVotesGranted->pNode = pNode;
return pVotesGranted;
}
@ -48,33 +48,33 @@ void voteGrantedDestroy(SVotesGranted *pVotesGranted) {
}
}
void voteGrantedUpdate(SVotesGranted *pVotesGranted, SSyncNode *pSyncNode) {
pVotesGranted->replicas = &(pSyncNode->replicasId);
pVotesGranted->replicaNum = pSyncNode->replicaNum;
void voteGrantedUpdate(SVotesGranted *pVotesGranted, SSyncNode *pNode) {
pVotesGranted->replicas = &pNode->replicasId;
pVotesGranted->replicaNum = pNode->replicaNum;
voteGrantedClearVotes(pVotesGranted);
pVotesGranted->term = 0;
pVotesGranted->quorum = pSyncNode->quorum;
pVotesGranted->quorum = pNode->quorum;
pVotesGranted->toLeader = false;
pVotesGranted->pSyncNode = pSyncNode;
pVotesGranted->pNode = pNode;
}
bool voteGrantedMajority(SVotesGranted *pVotesGranted) { return pVotesGranted->votes >= pVotesGranted->quorum; }
void voteGrantedVote(SVotesGranted *pVotesGranted, SyncRequestVoteReply *pMsg) {
if (!pMsg->voteGranted) {
sNFatal(pVotesGranted->pSyncNode, "vote granted should be true");
sNFatal(pVotesGranted->pNode, "vote granted should be true");
return;
}
if (pMsg->term != pVotesGranted->term) {
sNTrace(pVotesGranted->pSyncNode, "vote grant term:%" PRId64 " not matched with msg term:%" PRId64,
pVotesGranted->term, pMsg->term);
sNTrace(pVotesGranted->pNode, "vote grant term:%" PRId64 " not matched with msg term:%" PRId64, pVotesGranted->term,
pMsg->term);
return;
}
if (!syncUtilSameId(&pVotesGranted->pSyncNode->myRaftId, &pMsg->destId)) {
sNFatal(pVotesGranted->pSyncNode, "vote granted raftId not matched with msg");
if (!syncUtilSameId(&pVotesGranted->pNode->myRaftId, &pMsg->destId)) {
sNFatal(pVotesGranted->pNode, "vote granted raftId not matched with msg");
return;
}
@ -86,7 +86,7 @@ void voteGrantedVote(SVotesGranted *pVotesGranted, SyncRequestVoteReply *pMsg) {
}
}
if ((j == -1) || !(j >= 0 && j < pVotesGranted->replicaNum)) {
sNFatal(pVotesGranted->pSyncNode, "invalid msg srcId, index:%d", j);
sNFatal(pVotesGranted->pNode, "invalid msg srcId, index:%d", j);
return;
}
@ -96,7 +96,7 @@ void voteGrantedVote(SVotesGranted *pVotesGranted, SyncRequestVoteReply *pMsg) {
}
if (pVotesGranted->votes > pVotesGranted->replicaNum) {
sNFatal(pVotesGranted->pSyncNode, "votes:%d not matched with replicaNum:%d", pVotesGranted->votes,
sNFatal(pVotesGranted->pNode, "votes:%d not matched with replicaNum:%d", pVotesGranted->votes,
pVotesGranted->replicaNum);
return;
}
@ -108,17 +108,17 @@ void voteGrantedReset(SVotesGranted *pVotesGranted, SyncTerm term) {
pVotesGranted->toLeader = false;
}
SVotesRespond *votesRespondCreate(SSyncNode *pSyncNode) {
SVotesRespond *votesRespondCreate(SSyncNode *pNode) {
SVotesRespond *pVotesRespond = taosMemoryCalloc(1, sizeof(SVotesRespond));
if (pVotesRespond == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
}
pVotesRespond->replicas = &(pSyncNode->replicasId);
pVotesRespond->replicaNum = pSyncNode->replicaNum;
pVotesRespond->replicas = &pNode->replicasId;
pVotesRespond->replicaNum = pNode->replicaNum;
pVotesRespond->term = 0;
pVotesRespond->pSyncNode = pSyncNode;
pVotesRespond->pNode = pNode;
return pVotesRespond;
}
@ -129,11 +129,11 @@ void votesRespondDestory(SVotesRespond *pVotesRespond) {
}
}
void votesRespondUpdate(SVotesRespond *pVotesRespond, SSyncNode *pSyncNode) {
pVotesRespond->replicas = &(pSyncNode->replicasId);
pVotesRespond->replicaNum = pSyncNode->replicaNum;
void votesRespondUpdate(SVotesRespond *pVotesRespond, SSyncNode *pNode) {
pVotesRespond->replicas = &pNode->replicasId;
pVotesRespond->replicaNum = pNode->replicaNum;
pVotesRespond->term = 0;
pVotesRespond->pSyncNode = pSyncNode;
pVotesRespond->pNode = pNode;
}
bool votesResponded(SVotesRespond *pVotesRespond, const SRaftId *pRaftId) {
@ -149,7 +149,7 @@ bool votesResponded(SVotesRespond *pVotesRespond, const SRaftId *pRaftId) {
void votesRespondAdd(SVotesRespond *pVotesRespond, const SyncRequestVoteReply *pMsg) {
if (pVotesRespond->term != pMsg->term) {
sNTrace(pVotesRespond->pSyncNode, "vote respond add error");
sNTrace(pVotesRespond->pNode, "vote respond add error");
return;
}
@ -160,7 +160,7 @@ void votesRespondAdd(SVotesRespond *pVotesRespond, const SyncRequestVoteReply *p
}
}
sNFatal(pVotesRespond->pSyncNode, "votes respond not found");
sNFatal(pVotesRespond->pNode, "votes respond not found");
}
void votesRespondReset(SVotesRespond *pVotesRespond, SyncTerm term) {