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

enh: restore wal log from snapshot after finish

This commit is contained in:
Benguang Zhao 2023-10-24 20:03:48 +08:00
parent 374217ceff
commit 4e3a8b893d
1 changed files with 12 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -539,8 +539,6 @@ void snapshotReceiverStop(SSyncSnapshotReceiver *pReceiver) {
}
syncSnapBufferReset(pReceiver->pRcvBuf);
sRInfo(pReceiver, "snapshot receiver stop, from dnode:%d.", DID(&pReceiver->fromId));
}
// when recv last snapshot block, apply data into snapshot
@ -548,7 +546,7 @@ static int32_t snapshotReceiverFinish(SSyncSnapshotReceiver *pReceiver, SyncSnap
int32_t code = 0;
if (pReceiver->pWriter != NULL) {
// write data
sRInfo(pReceiver, "snapshot receiver write finish, blockLen:%d seq:%d", pMsg->dataLen, pMsg->seq);
sRInfo(pReceiver, "snapshot receiver write about to finish, blockLen:%d seq:%d", pMsg->dataLen, pMsg->seq);
if (pMsg->dataLen > 0) {
code = pReceiver->pSyncNode->pFsm->FpSnapshotDoWrite(pReceiver->pSyncNode->pFsm, pReceiver->pWriter, pMsg->data,
pMsg->dataLen);
@ -558,15 +556,6 @@ static int32_t snapshotReceiverFinish(SSyncSnapshotReceiver *pReceiver, SyncSnap
}
}
// reset wal
sRInfo(pReceiver, "snapshot receiver log restore");
code =
pReceiver->pSyncNode->pLogStore->syncLogRestoreFromSnapshot(pReceiver->pSyncNode->pLogStore, pMsg->lastIndex);
if (code != 0) {
sRError(pReceiver, "failed to snapshot receiver log restore since %s", terrstr());
return -1;
}
// update commit index
if (pReceiver->snapshot.lastApplyIndex > pReceiver->pSyncNode->commitIndex) {
pReceiver->pSyncNode->commitIndex = pReceiver->snapshot.lastApplyIndex;
@ -578,7 +567,6 @@ static int32_t snapshotReceiverFinish(SSyncSnapshotReceiver *pReceiver, SyncSnap
}
// stop writer, apply data
sRInfo(pReceiver, "snapshot receiver apply write");
code = pReceiver->pSyncNode->pFsm->FpSnapshotStopWrite(pReceiver->pSyncNode->pFsm, pReceiver->pWriter, true,
&pReceiver->snapshot);
if (code != 0) {
@ -586,10 +574,21 @@ static int32_t snapshotReceiverFinish(SSyncSnapshotReceiver *pReceiver, SyncSnap
return -1;
}
pReceiver->pWriter = NULL;
sRInfo(pReceiver, "snapshot receiver write stopped");
// update progress
pReceiver->ack = SYNC_SNAPSHOT_SEQ_END;
// reset wal
code =
pReceiver->pSyncNode->pLogStore->syncLogRestoreFromSnapshot(pReceiver->pSyncNode->pLogStore, pMsg->lastIndex);
if (code != 0) {
sRError(pReceiver, "failed to snapshot receiver log restore since %s", terrstr());
return -1;
}
sRInfo(pReceiver, "wal log restored from snapshot");
// get fsmState
SSnapshot snapshot = {0};
pReceiver->pSyncNode->pFsm->FpGetSnapshotInfo(pReceiver->pSyncNode->pFsm, &snapshot);
pReceiver->pSyncNode->fsmState = snapshot.state;
@ -599,8 +598,6 @@ static int32_t snapshotReceiverFinish(SSyncSnapshotReceiver *pReceiver, SyncSnap
return -1;
}
// event log
sRInfo(pReceiver, "snapshot receiver got last data and apply snapshot finished");
return 0;
}