diff --git a/source/libs/sync/inc/syncVoteMgr.h b/source/libs/sync/inc/syncVoteMgr.h index 066a4dd76f..6e63a2b396 100644 --- a/source/libs/sync/inc/syncVoteMgr.h +++ b/source/libs/sync/inc/syncVoteMgr.h @@ -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); diff --git a/source/libs/sync/src/syncRespMgr.c b/source/libs/sync/src/syncRespMgr.c index 503c56b1e7..b55aae4c76 100644 --- a/source/libs/sync/src/syncRespMgr.c +++ b/source/libs/sync/src/syncRespMgr.c @@ -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); } diff --git a/source/libs/sync/src/syncSnapshot.c b/source/libs/sync/src/syncSnapshot.c index f2da2fb2ce..b68b735f46 100644 --- a/source/libs/sync/src/syncSnapshot.c +++ b/source/libs/sync/src/syncSnapshot.c @@ -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); diff --git a/source/libs/sync/src/syncVoteMgr.c b/source/libs/sync/src/syncVoteMgr.c index 8a0a35ce33..6bd9625276 100644 --- a/source/libs/sync/src/syncVoteMgr.c +++ b/source/libs/sync/src/syncVoteMgr.c @@ -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) {