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

fix(sync): delete some assert temporary

This commit is contained in:
Minghao Li 2022-06-07 09:55:18 +08:00
parent 9baaf2cd0c
commit a89e455267
3 changed files with 23 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
source/libs/sync/inc/syncInt.h
View File

@ -225,6 +225,8 @@ int32_t syncNodeGetPreIndexTerm(SSyncNode* pSyncNode, SyncIndex index, SyncInd
int32_t syncNodeCommit(SSyncNode* ths, SyncIndex beginIndex, SyncIndex endIndex, uint64_t flag); int32_t syncNodeCommit(SSyncNode* ths, SyncIndex beginIndex, SyncIndex endIndex, uint64_t flag);
bool syncNodeInRaftGroup(SSyncNode* ths, SRaftId* pRaftId);
// for debug -------------- // for debug --------------
void syncNodePrint(SSyncNode* pObj); void syncNodePrint(SSyncNode* pObj);
void syncNodePrint2(char* s, SSyncNode* pObj); void syncNodePrint2(char* s, SSyncNode* pObj);

8
source/libs/sync/src/syncAppendEntriesReply.c
View File

@ -104,6 +104,12 @@ int32_t syncNodeOnAppendEntriesReplySnapshotCb(SSyncNode* ths, SyncAppendEntries
snprintf(logBuf, sizeof(logBuf), "recv SyncAppendEntriesReply, term:%lu", ths->pRaftStore->currentTerm); snprintf(logBuf, sizeof(logBuf), "recv SyncAppendEntriesReply, term:%lu", ths->pRaftStore->currentTerm);
syncAppendEntriesReplyLog2(logBuf, pMsg); syncAppendEntriesReplyLog2(logBuf, pMsg);
// if already drop replica, do not process
if (!syncNodeInRaftGroup(ths, &(pMsg->srcId))) {
sInfo("maybe already dropped");
return ret;
}
// drop stale response // drop stale response
if (pMsg->term < ths->pRaftStore->currentTerm) { if (pMsg->term < ths->pRaftStore->currentTerm) {
sTrace("recv SyncAppendEntriesReply, drop stale response, receive_term:%lu current_term:%lu", pMsg->term, sTrace("recv SyncAppendEntriesReply, drop stale response, receive_term:%lu current_term:%lu", pMsg->term,
@ -162,7 +168,7 @@ int32_t syncNodeOnAppendEntriesReplySnapshotCb(SSyncNode* ths, SyncAppendEntries
SSnapshot snapshot; SSnapshot snapshot;
ths->pFsm->FpGetSnapshot(ths->pFsm, &snapshot); ths->pFsm->FpGetSnapshot(ths->pFsm, &snapshot);
if (nextIndex <= snapshot.lastApplyIndex) { if (nextIndex <= snapshot.lastApplyIndex) {
ASSERT(nextIndex == snapshot.lastApplyIndex); // ASSERT(nextIndex == snapshot.lastApplyIndex);
nextIndex = snapshot.lastApplyIndex + 1; nextIndex = snapshot.lastApplyIndex + 1;
sInfo("reset new nextIndex %ld, snapshot.lastApplyIndex:%ld", nextIndex, snapshot.lastApplyIndex); sInfo("reset new nextIndex %ld, snapshot.lastApplyIndex:%ld", nextIndex, snapshot.lastApplyIndex);

17
source/libs/sync/src/syncMain.c
View File

@ -1347,7 +1347,7 @@ SyncIndex syncNodeGetPreIndex(SSyncNode* pSyncNode, SyncIndex index) {
pSyncNode->pFsm->FpGetSnapshot(pSyncNode->pFsm, &snapshot); pSyncNode->pFsm->FpGetSnapshot(pSyncNode->pFsm, &snapshot);
} }
ASSERT(index > snapshot.lastApplyIndex); // ASSERT(index > snapshot.lastApplyIndex);
preIndex = index - 1; preIndex = index - 1;
} else { } else {
@ -1375,7 +1375,7 @@ SyncTerm syncNodeGetPreTerm(SSyncNode* pSyncNode, SyncIndex index) {
pSyncNode->pFsm->FpGetSnapshot(pSyncNode->pFsm, &snapshot); pSyncNode->pFsm->FpGetSnapshot(pSyncNode->pFsm, &snapshot);
} }
ASSERT(index > snapshot.lastApplyIndex); // ASSERT(index > snapshot.lastApplyIndex);
if (index > snapshot.lastApplyIndex + 1) { if (index > snapshot.lastApplyIndex + 1) {
// should be log preTerm // should be log preTerm
SSyncRaftEntry* pPreEntry = NULL; SSyncRaftEntry* pPreEntry = NULL;
@ -1389,7 +1389,9 @@ SyncTerm syncNodeGetPreTerm(SSyncNode* pSyncNode, SyncIndex index) {
} else if (index == snapshot.lastApplyIndex + 1) { } else if (index == snapshot.lastApplyIndex + 1) {
preTerm = snapshot.lastApplyTerm; preTerm = snapshot.lastApplyTerm;
} else { } else {
ASSERT(0); // ASSERT(0);
// maybe snapshot change
preTerm = snapshot.lastApplyTerm;
} }
} else { } else {
@ -1799,4 +1801,13 @@ int32_t syncNodeCommit(SSyncNode* ths, SyncIndex beginIndex, SyncIndex endIndex,
} }
} }
return 0; return 0;
}
bool syncNodeInRaftGroup(SSyncNode* ths, SRaftId* pRaftId) {
for (int i = 0; i < ths->replicaNum; ++i) {
if (syncUtilSameId(&((ths->replicasId)[i]), pRaftId)) {
return true;
}
}
return false;
} }