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

refactor(sync): modify strategy for syncBeginSnapshot

This commit is contained in:
Minghao Li 2022-10-18 17:15:57 +08:00
parent e85954276f
commit 1df2703c41
1 changed files with 14 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -301,7 +301,7 @@ int32_t syncBeginSnapshot(int64_t rid, int64_t lastApplyIndex) {
SyncIndex beginIndex = pSyncNode->pLogStore->syncLogBeginIndex(pSyncNode->pLogStore); SyncIndex beginIndex = pSyncNode->pLogStore->syncLogBeginIndex(pSyncNode->pLogStore);
SyncIndex endIndex = pSyncNode->pLogStore->syncLogEndIndex(pSyncNode->pLogStore); SyncIndex endIndex = pSyncNode->pLogStore->syncLogEndIndex(pSyncNode->pLogStore);
int64_t logNum = endIndex - beginIndex; int64_t logNum = endIndex - beginIndex;
if (logNum > SYNC_MNODE_MAX_LOG_NUM) { if (logNum < SYNC_MNODE_MAX_LOG_NUM) {
char logBuf[256]; char logBuf[256];
snprintf(logBuf, sizeof(logBuf), "mnode log num:%ld, do not delete", logNum); snprintf(logBuf, sizeof(logBuf), "mnode log num:%ld, do not delete", logNum);
syncNodeEventLog(pSyncNode, logBuf); syncNodeEventLog(pSyncNode, logBuf);
@ -341,6 +341,13 @@ int32_t syncBeginSnapshot(int64_t rid, int64_t lastApplyIndex) {
if (snapshottingIndex == SYNC_INDEX_INVALID) { if (snapshottingIndex == SYNC_INDEX_INVALID) {
atomic_store_64(&pSyncNode->snapshottingIndex, lastApplyIndex); atomic_store_64(&pSyncNode->snapshottingIndex, lastApplyIndex);
do {
char logBuf[256];
snprintf(logBuf, sizeof(logBuf), "wal snapshot begin, index:%ld, last apply index:%ld",
pSyncNode->snapshottingIndex, lastApplyIndex);
syncNodeEventLog(pSyncNode, logBuf);
} while (0);
SSyncLogStoreData* pData = pSyncNode->pLogStore->data; SSyncLogStoreData* pData = pSyncNode->pLogStore->data;
code = walBeginSnapshot(pData->pWal, lastApplyIndex); code = walBeginSnapshot(pData->pWal, lastApplyIndex);
@ -366,6 +373,12 @@ int32_t syncEndSnapshot(int64_t rid) {
int32_t code = 0; int32_t code = 0;
if (atomic_load_64(&pSyncNode->snapshottingIndex) != SYNC_INDEX_INVALID) { if (atomic_load_64(&pSyncNode->snapshottingIndex) != SYNC_INDEX_INVALID) {
do {
char logBuf[256];
snprintf(logBuf, sizeof(logBuf), "wal snapshot end, index:%ld", atomic_load_64(&pSyncNode->snapshottingIndex));
syncNodeEventLog(pSyncNode, logBuf);
} while (0);
SSyncLogStoreData* pData = pSyncNode->pLogStore->data; SSyncLogStoreData* pData = pSyncNode->pLogStore->data;
code = walEndSnapshot(pData->pWal); code = walEndSnapshot(pData->pWal);