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

fix: send snapshot

This commit is contained in:
Minghao Li 2022-06-01 21:57:30 +08:00
parent 757a1248dc
commit c3bf3cc97c
3 changed files with 17 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -562,15 +562,18 @@ SSyncNode* syncNodeOpen(const SSyncInfo* pOldSyncInfo) {
pSyncNode->FpOnSnapshotRsp = syncNodeOnSnapshotRspCb; pSyncNode->FpOnSnapshotRsp = syncNodeOnSnapshotRspCb;
if (pSyncNode->pRaftCfg->snapshotEnable) { if (pSyncNode->pRaftCfg->snapshotEnable) {
pSyncNode->FpOnRequestVote = syncNodeOnRequestVoteCb; sInfo("sync node use snapshot");
pSyncNode->FpOnRequestVoteReply = syncNodeOnRequestVoteReplyCb;
pSyncNode->FpOnAppendEntries = syncNodeOnAppendEntriesCb;
pSyncNode->FpOnAppendEntriesReply = syncNodeOnAppendEntriesReplyCb;
} else {
pSyncNode->FpOnRequestVote = syncNodeOnRequestVoteSnapshotCb; pSyncNode->FpOnRequestVote = syncNodeOnRequestVoteSnapshotCb;
pSyncNode->FpOnRequestVoteReply = syncNodeOnRequestVoteReplySnapshotCb; pSyncNode->FpOnRequestVoteReply = syncNodeOnRequestVoteReplySnapshotCb;
pSyncNode->FpOnAppendEntries = syncNodeOnAppendEntriesSnapshotCb; pSyncNode->FpOnAppendEntries = syncNodeOnAppendEntriesSnapshotCb;
pSyncNode->FpOnAppendEntriesReply = syncNodeOnAppendEntriesReplySnapshotCb; pSyncNode->FpOnAppendEntriesReply = syncNodeOnAppendEntriesReplySnapshotCb;
} else {
sInfo("sync node do not use snapshot");
pSyncNode->FpOnRequestVote = syncNodeOnRequestVoteCb;
pSyncNode->FpOnRequestVoteReply = syncNodeOnRequestVoteReplyCb;
pSyncNode->FpOnAppendEntries = syncNodeOnAppendEntriesCb;
pSyncNode->FpOnAppendEntriesReply = syncNodeOnAppendEntriesReplyCb;
} }
// tools // tools

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

@ -432,6 +432,7 @@ int32_t syncNodeOnSnapshotSendCb(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) {
// end, finish FSM // end, finish FSM
pSyncNode->pFsm->FpSnapshotDoWrite(pSyncNode->pFsm, pReceiver->pWriter, pMsg->data, pMsg->dataLen); pSyncNode->pFsm->FpSnapshotDoWrite(pSyncNode->pFsm, pReceiver->pWriter, pMsg->data, pMsg->dataLen);
pSyncNode->pFsm->FpSnapshotStopWrite(pSyncNode->pFsm, pReceiver->pWriter, true); pSyncNode->pFsm->FpSnapshotStopWrite(pSyncNode->pFsm, pReceiver->pWriter, true);
pReceiver->pWriter = NULL;
snapshotReceiverStop(pReceiver); snapshotReceiverStop(pReceiver);
pReceiver->ack = pMsg->seq; pReceiver->ack = pMsg->seq;
needRsp = false; needRsp = false;

6
source/libs/sync/test/syncConfigChangeSnapshotTest.cpp
View File

@ -127,6 +127,12 @@ int32_t SnapshotStopWrite(struct SSyncFSM* pFsm, void* pWriter, bool isApply) {
snprintf(logBuf, sizeof(logBuf), "==callback== ==SnapshotStopWrite== pFsm:%p, pWriter:%p, isApply:%d", pFsm, pWriter, snprintf(logBuf, sizeof(logBuf), "==callback== ==SnapshotStopWrite== pFsm:%p, pWriter:%p, isApply:%d", pFsm, pWriter,
isApply); isApply);
sTrace("%s", logBuf); sTrace("%s", logBuf);
if (isApply) {
gSnapshotLastApplyIndex = 7;
gSnapshotLastApplyTerm = 1;
}
return 0; return 0;
} }