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

fix(sync): wal write from middle

This commit is contained in:
Minghao Li 2022-06-02 13:53:19 +08:00
parent e4f04491df
commit ac90f61b63
3 changed files with 17 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
source/libs/sync/src/syncAppendEntries.c
View File

@ -20,6 +20,7 @@
#include "syncRaftStore.h" #include "syncRaftStore.h"
#include "syncUtil.h" #include "syncUtil.h"
#include "syncVoteMgr.h" #include "syncVoteMgr.h"
#include "wal.h"
// TLA+ Spec // TLA+ Spec
// HandleAppendEntriesRequest(i, j, m) == // HandleAppendEntriesRequest(i, j, m) ==
@ -687,7 +688,10 @@ int32_t syncNodeOnAppendEntriesSnapshotCb(SSyncNode* ths, SyncAppendEntries* pMs
// execute fsm // execute fsm
if (ths->pFsm != NULL) { if (ths->pFsm != NULL) {
for (SyncIndex i = beginIndex; i <= endIndex; ++i) { for (SyncIndex i = beginIndex; i <= endIndex; ++i) {
if (i != SYNC_INDEX_INVALID) { // notice! wal maybe deleted, update firstVer
SyncIndex walFirstVer = walGetFirstVer(ths->pWal);
if (i != SYNC_INDEX_INVALID && i >= walFirstVer) {
SSyncRaftEntry* pEntry = ths->pLogStore->getEntry(ths->pLogStore, i); SSyncRaftEntry* pEntry = ths->pLogStore->getEntry(ths->pLogStore, i);
assert(pEntry != NULL); assert(pEntry != NULL);

10
source/libs/sync/src/syncRaftLog.c
View File

@ -208,9 +208,13 @@ cJSON* logStore2Json(SSyncLogStore* pLogStore) {
cJSON_AddItemToObject(pRoot, "pEntries", pEntries); cJSON_AddItemToObject(pRoot, "pEntries", pEntries);
SyncIndex lastIndex = logStoreLastIndex(pLogStore); SyncIndex lastIndex = logStoreLastIndex(pLogStore);
for (SyncIndex i = 0; i <= lastIndex; ++i) { for (SyncIndex i = 0; i <= lastIndex; ++i) {
SSyncRaftEntry* pEntry = logStoreGetEntry(pLogStore, i); SyncIndex walFirstVer = walGetFirstVer(pData->pWal);
cJSON_AddItemToArray(pEntries, syncEntry2Json(pEntry));
syncEntryDestory(pEntry); if (i != SYNC_INDEX_INVALID && i >= walFirstVer) {
SSyncRaftEntry* pEntry = logStoreGetEntry(pLogStore, i);
cJSON_AddItemToArray(pEntries, syncEntry2Json(pEntry));
syncEntryDestory(pEntry);
}
} }
} }

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

@ -16,6 +16,7 @@
#include "syncSnapshot.h" #include "syncSnapshot.h"
#include "syncRaftStore.h" #include "syncRaftStore.h"
#include "syncUtil.h" #include "syncUtil.h"
#include "wal.h"
static void snapshotSenderDoStart(SSyncSnapshotSender *pSender); static void snapshotSenderDoStart(SSyncSnapshotSender *pSender);
static void snapshotReceiverDoStart(SSyncSnapshotReceiver *pReceiver); static void snapshotReceiverDoStart(SSyncSnapshotReceiver *pReceiver);
@ -434,6 +435,10 @@ 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);
walRestoreFromSnapshot(pSyncNode->pWal, pMsg->lastIndex);
sInfo("walRestoreFromSnapshot lastIndex:%ld", pMsg->lastIndex);
pReceiver->pWriter = NULL; pReceiver->pWriter = NULL;
snapshotReceiverStop(pReceiver); snapshotReceiverStop(pReceiver);
pReceiver->ack = pMsg->seq; pReceiver->ack = pMsg->seq;