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

refactor: rsma restore

This commit is contained in:
Cary Xu 2022-06-27 21:35:17 +08:00
parent da8db399c3
commit 771801e2af
1 changed files with 14 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
source/dnode/vnode/src/sma/smaRollup.c
View File

@ -74,6 +74,12 @@ struct SRSmaQTaskFIter {
int32_t nBufPos; int32_t nBufPos;
}; };
static FORCE_INLINE int32_t tdRSmaQTaskInfoContLen(int32_t lenWithHead) {
return lenWithHead - sizeof(int32_t) - sizeof(int8_t) - sizeof(int64_t);
}
static FORCE_INLINE void tdRSmaQTaskInfoIterDestroy(SRSmaQTaskFIter *pIter) { taosMemoryFreeClear(pIter->buf); }
static FORCE_INLINE void tdFreeTaskHandle(qTaskInfo_t *taskHandle, int32_t vgId, int32_t level) { static FORCE_INLINE void tdFreeTaskHandle(qTaskInfo_t *taskHandle, int32_t vgId, int32_t level) {
// Note: free/kill may in RC // Note: free/kill may in RC
qTaskInfo_t otaskHandle = atomic_load_ptr(taskHandle); qTaskInfo_t otaskHandle = atomic_load_ptr(taskHandle);
@ -768,7 +774,7 @@ static void *tdRSmaPersistExec(void *param) {
isFileCreated = true; isFileCreated = true;
} }
len += (sizeof(len)+ sizeof(type) + sizeof(pRSmaInfo->suid)); len += (sizeof(len) + sizeof(type) + sizeof(pRSmaInfo->suid));
tdAppendTFile(&tFile, &len, sizeof(len), &toffset); tdAppendTFile(&tFile, &len, sizeof(len), &toffset);
tdAppendTFile(&tFile, &type, sizeof(type), &toffset); tdAppendTFile(&tFile, &type, sizeof(type), &toffset);
tdAppendTFile(&tFile, &pRSmaInfo->suid, sizeof(pRSmaInfo->suid), &toffset); tdAppendTFile(&tFile, &pRSmaInfo->suid, sizeof(pRSmaInfo->suid), &toffset);
@ -956,11 +962,15 @@ int32_t tdProcessRSmaRestoreImpl(SSma *pSma) {
if (isEnd) { if (isEnd) {
break; break;
} }
if((code = tdRSmaQTaskInfoItemRestore(pSma, &infoItem)) < 0){ if ((code = tdRSmaQTaskInfoItemRestore(pSma, &infoItem)) < 0) {
break; break;
} }
} }
if (code < 0) goto _err; tdRSmaQTaskInfoIterDestroy(&fIter);
if (code < 0) {
goto _err;
}
metaReaderClear(&mr); metaReaderClear(&mr);
taosArrayDestroy(suidList); taosArrayDestroy(suidList);
@ -1091,12 +1101,7 @@ static int32_t tdRSmaQTaskInfoIterNextBlock(SRSmaQTaskFIter *pIter, bool *isFini
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static FORCE_INLINE int32_t tdRSmaQTaskInfoContLen(int32_t lenWithHead) {
return lenWithHead - sizeof(int32_t) - sizeof(int8_t) - sizeof(int64_t);
}
static int32_t tdRSmaQTaskInfoIterNext(SRSmaQTaskFIter *pIter, SRSmaQTaskInfoItem *pItem, bool *isEnd) { static int32_t tdRSmaQTaskInfoIterNext(SRSmaQTaskFIter *pIter, SRSmaQTaskInfoItem *pItem, bool *isEnd) {
while (1) { while (1) {
// block iter // block iter
bool isFinish = false; bool isFinish = false;
@ -1112,7 +1117,7 @@ static int32_t tdRSmaQTaskInfoIterNext(SRSmaQTaskFIter *pIter, SRSmaQTaskInfoIte
// consume the block // consume the block
int32_t qTaskInfoLenWithHead = 0; int32_t qTaskInfoLenWithHead = 0;
pIter->buf = taosDecodeFixedI32(pIter->buf, &qTaskInfoLenWithHead); pIter->buf = taosDecodeFixedI32(pIter->buf, &qTaskInfoLenWithHead);
if(qTaskInfoLenWithHead < 0) { if (qTaskInfoLenWithHead < 0) {
terrno = TSDB_CODE_TDB_FILE_CORRUPTED; terrno = TSDB_CODE_TDB_FILE_CORRUPTED;
return TSDB_CODE_FAILED; return TSDB_CODE_FAILED;
} }