From f78c61aac64547eb0aedf68857ca132a5786b8b8 Mon Sep 17 00:00:00 2001 From: Cary Xu Date: Sun, 4 Sep 2022 14:54:33 +0800 Subject: [PATCH] refactor: rsma fetch trigger refactor and use ref to manage qtaskf --- source/common/src/trow.c | 4 +- source/dnode/vnode/src/inc/sma.h | 33 ++-- source/dnode/vnode/src/inc/vnodeInt.h | 20 +-- source/dnode/vnode/src/sma/smaCommit.c | 34 ++--- source/dnode/vnode/src/sma/smaEnv.c | 44 ++++-- source/dnode/vnode/src/sma/smaFS.c | 75 +++++++-- source/dnode/vnode/src/sma/smaRollup.c | 109 ++++++++------ source/dnode/vnode/src/sma/smaSnapshot.c | 167 ++++++++++++++------- source/dnode/vnode/src/vnd/vnodeSnapshot.c | 4 +- 9 files changed, 321 insertions(+), 169 deletions(-) diff --git a/source/common/src/trow.c b/source/common/src/trow.c index 565498a47b..9880fe362e 100644 --- a/source/common/src/trow.c +++ b/source/common/src/trow.c @@ -538,12 +538,12 @@ bool tdSTSRowIterGetTpVal(STSRowIter *pIter, col_type_t colType, int32_t offset, } else { pVal->val = POINTER_SHIFT(TD_ROW_DATA(pRow), offset); } - return TSDB_CODE_SUCCESS; + return true; } if (tdGetBitmapValType(pIter->pBitmap, pIter->colIdx - 1, &pVal->valType, 0) != TSDB_CODE_SUCCESS) { pVal->valType = TD_VTYPE_NONE; - return terrno; + return true; } if (pVal->valType == TD_VTYPE_NORM) { diff --git a/source/dnode/vnode/src/inc/sma.h b/source/dnode/vnode/src/inc/sma.h index 85720ccef6..e303faf7de 100644 --- a/source/dnode/vnode/src/inc/sma.h +++ b/source/dnode/vnode/src/inc/sma.h @@ -38,7 +38,7 @@ typedef struct SSmaEnv SSmaEnv; typedef struct SSmaStat SSmaStat; typedef struct STSmaStat STSmaStat; typedef struct SRSmaStat SRSmaStat; -typedef struct SSmaKey SSmaKey; +typedef struct SRSmaRef SRSmaRef; typedef struct SRSmaInfo SRSmaInfo; typedef struct SRSmaInfoItem SRSmaInfoItem; typedef struct SRSmaFS SRSmaFS; @@ -55,10 +55,21 @@ struct SSmaEnv { #define SMA_ENV_FLG_CLOSE ((int8_t)0x1) +struct SRSmaRef { + int64_t refId; // for SRSmaStat + int64_t suid; +}; + typedef struct { int8_t inited; int32_t rsetId; void *tmrHandle; // shared by all fetch tasks + /** + * @brief key: void* of SRSmaInfoItem, value: SRSmaRef + * N.B. Although there is a very small possibility that "void*" point to different objects while with the same + * address after release/renew, the functionality is not affected as it just used to fetch the rsma results. + */ + SHashObj *refHash; // shared by all vgroups } SSmaMgmt; #define SMA_ENV_LOCK(env) (&(env)->lock) @@ -80,15 +91,15 @@ struct SQTaskFile { }; struct SQTaskFReader { - SSma *pSma; - SQTaskFile fTask; - TdFilePtr pReadH; + SSma *pSma; + int64_t version; + TdFilePtr pReadH; }; struct SQTaskFWriter { - SSma *pSma; - SQTaskFile fTask; - TdFilePtr pWriteH; - char *fname; + SSma *pSma; + int64_t version; + TdFilePtr pWriteH; + char *fname; }; struct SRSmaFS { @@ -124,6 +135,7 @@ struct SSmaStat { #define RSMA_TRIGGER_STAT(r) (&(r)->triggerStat) #define RSMA_COMMIT_STAT(r) (&(r)->commitStat) #define RSMA_REF_ID(r) ((r)->refId) +#define RSMA_FS(r) (&(r)->fs) #define RSMA_FS_LOCK(r) (&(r)->lock) struct SRSmaInfoItem { @@ -138,7 +150,6 @@ struct SRSmaInfoItem { struct SRSmaInfo { STSchema *pTSchema; int64_t suid; - int64_t refId; // refId of SRSmaStat int64_t lastRecv; // ms int8_t assigned; // 0 idle, 1 assgined for exec int8_t delFlag; @@ -197,7 +208,9 @@ int32_t tdUnRefRSmaInfo(SSma *pSma, SRSmaInfo *pRSmaInfo); void *tdFreeRSmaInfo(SSma *pSma, SRSmaInfo *pInfo, bool isDeepFree); int32_t tdRSmaFSOpen(SSma *pSma, int64_t version); void tdRSmaFSClose(SRSmaFS *fs); -int32_t tdRSmaFSUpsertQFile(SRSmaFS *fs, SQTaskFile *pTaskF); +int32_t tdRSmaFSRef(SSma *pSma, SRSmaStat *pStat, int64_t version); +void tdRSmaFSUnRef(SSma *pSma, SRSmaStat *pStat, int64_t version); +int32_t tdRSmaFSUpsertQTaskFile(SRSmaFS *pFS, SQTaskFile *qTaskFile); int32_t tdRSmaRestore(SSma *pSma, int8_t type, int64_t committedVer); int32_t tdRSmaProcessCreateImpl(SSma *pSma, SRSmaParam *param, int64_t suid, const char *tbName); int32_t tdRSmaProcessExecImpl(SSma *pSma, ERsmaExecType type); diff --git a/source/dnode/vnode/src/inc/vnodeInt.h b/source/dnode/vnode/src/inc/vnodeInt.h index 9b252df58b..6423c1aa69 100644 --- a/source/dnode/vnode/src/inc/vnodeInt.h +++ b/source/dnode/vnode/src/inc/vnodeInt.h @@ -70,8 +70,8 @@ typedef struct SStreamTaskReader SStreamTaskReader; typedef struct SStreamTaskWriter SStreamTaskWriter; typedef struct SStreamStateReader SStreamStateReader; typedef struct SStreamStateWriter SStreamStateWriter; -typedef struct SRsmaSnapReader SRsmaSnapReader; -typedef struct SRsmaSnapWriter SRsmaSnapWriter; +typedef struct SRSmaSnapReader SRSmaSnapReader; +typedef struct SRSmaSnapWriter SRSmaSnapWriter; typedef struct SSnapDataHdr SSnapDataHdr; #define VNODE_META_DIR "meta" @@ -246,14 +246,14 @@ int32_t tqOffsetSnapWrite(STqOffsetWriter* pWriter, uint8_t* pData, uint32_t nDa // SStreamTaskReader ====================================== // SStreamStateWriter ===================================== // SStreamStateReader ===================================== -// SRsmaSnapReader ======================================== -int32_t rsmaSnapReaderOpen(SSma* pSma, int64_t sver, int64_t ever, SRsmaSnapReader** ppReader); -int32_t rsmaSnapReaderClose(SRsmaSnapReader** ppReader); -int32_t rsmaSnapRead(SRsmaSnapReader* pReader, uint8_t** ppData); -// SRsmaSnapWriter ======================================== -int32_t rsmaSnapWriterOpen(SSma* pSma, int64_t sver, int64_t ever, SRsmaSnapWriter** ppWriter); -int32_t rsmaSnapWrite(SRsmaSnapWriter* pWriter, uint8_t* pData, uint32_t nData); -int32_t rsmaSnapWriterClose(SRsmaSnapWriter** ppWriter, int8_t rollback); +// SRSmaSnapReader ======================================== +int32_t rsmaSnapReaderOpen(SSma* pSma, int64_t sver, int64_t ever, SRSmaSnapReader** ppReader); +int32_t rsmaSnapReaderClose(SRSmaSnapReader** ppReader); +int32_t rsmaSnapRead(SRSmaSnapReader* pReader, uint8_t** ppData); +// SRSmaSnapWriter ======================================== +int32_t rsmaSnapWriterOpen(SSma* pSma, int64_t sver, int64_t ever, SRSmaSnapWriter** ppWriter); +int32_t rsmaSnapWrite(SRSmaSnapWriter* pWriter, uint8_t* pData, uint32_t nData); +int32_t rsmaSnapWriterClose(SRSmaSnapWriter** ppWriter, int8_t rollback); typedef struct { int8_t streamType; // sma or other diff --git a/source/dnode/vnode/src/sma/smaCommit.c b/source/dnode/vnode/src/sma/smaCommit.c index 77b5399139..6d3db47b18 100644 --- a/source/dnode/vnode/src/sma/smaCommit.c +++ b/source/dnode/vnode/src/sma/smaCommit.c @@ -15,6 +15,8 @@ #include "sma.h" +extern SSmaMgmt smaMgmt; + static int32_t tdProcessRSmaSyncPreCommitImpl(SSma *pSma); static int32_t tdProcessRSmaSyncCommitImpl(SSma *pSma); static int32_t tdProcessRSmaSyncPostCommitImpl(SSma *pSma); @@ -178,10 +180,12 @@ static int32_t tdProcessRSmaSyncCommitImpl(SSma *pSma) { */ static int32_t tdUpdateQTaskInfoFiles(SSma *pSma, SRSmaStat *pStat) { SVnode *pVnode = pSma->pVnode; - SRSmaFS *pFS = &pStat->fs; + SRSmaFS *pFS = RSMA_FS(pStat); int64_t committed = pStat->commitAppliedVer; char qTaskInfoFullName[TSDB_FILENAME_LEN]; + taosWLockLatch(RSMA_FS_LOCK(pStat)); + for (int32_t i = 0; i < taosArrayGetSize(pFS->aQTaskInf);) { SQTaskFile *pTaskF = taosArrayGet(pFS->aQTaskInf, i); if (atomic_sub_fetch_32(&pTaskF->nRef, 1) <= 0) { @@ -199,10 +203,12 @@ static int32_t tdUpdateQTaskInfoFiles(SSma *pSma, SRSmaStat *pStat) { } SQTaskFile qFile = {.nRef = 1, .padding = 0, .version = committed, .size = 0}; - if (tdRSmaFSUpsertQFile(pFS, &qFile) < 0) { + if (tdRSmaFSUpsertQTaskFile(pFS, &qFile) < 0) { + taosWUnLockLatch(RSMA_FS_LOCK(pStat)); return TSDB_CODE_FAILED; } + taosWUnLockLatch(RSMA_FS_LOCK(pStat)); return TSDB_CODE_SUCCESS; } @@ -357,8 +363,8 @@ static int32_t tdProcessRSmaAsyncPostCommitImpl(SSma *pSma) { return TSDB_CODE_SUCCESS; } - SRSmaStat *pRSmaStat = (SRSmaStat *)SMA_ENV_STAT(pEnv); - SArray *rsmaDeleted = NULL; + SRSmaStat *pRSmaStat = (SRSmaStat *)SMA_ENV_STAT(pEnv); + SRSmaInfoItem *pItem = NULL; // step 1: merge qTaskInfo and iQTaskInfo // lock @@ -371,11 +377,7 @@ static int32_t tdProcessRSmaAsyncPostCommitImpl(SSma *pSma) { if (RSMA_INFO_IS_DEL(pRSmaInfo)) { int32_t refVal = T_REF_VAL_GET(pRSmaInfo); if (refVal == 0) { - if (!rsmaDeleted) { - if ((rsmaDeleted = taosArrayInit(1, sizeof(tb_uid_t)))) { - taosArrayPush(rsmaDeleted, pSuid); - } - } + taosHashRemove(RSMA_INFO_HASH(pRSmaStat), pSuid, sizeof(*pSuid)); } else { smaDebug( "vgId:%d, rsma async post commit, not free rsma info since ref is %d although already deleted for " @@ -401,20 +403,6 @@ static int32_t tdProcessRSmaAsyncPostCommitImpl(SSma *pSma) { #endif } - for (int32_t i = 0; i < taosArrayGetSize(rsmaDeleted); ++i) { - tb_uid_t *pSuid = taosArrayGet(rsmaDeleted, i); - void *pRSmaInfo = taosHashGet(RSMA_INFO_HASH(pRSmaStat), pSuid, sizeof(tb_uid_t)); - if ((pRSmaInfo = *(SRSmaInfo **)pRSmaInfo)) { - tdFreeRSmaInfo(pSma, pRSmaInfo, true); - smaDebug( - "vgId:%d, rsma async post commit, free rsma info since already deleted and ref is 0 for " - "table:%" PRIi64, - SMA_VID(pSma), *pSuid); - } - taosHashRemove(RSMA_INFO_HASH(pRSmaStat), pSuid, sizeof(tb_uid_t)); - } - taosArrayDestroy(rsmaDeleted); - // unlock // taosWUnLockLatch(SMA_ENV_LOCK(pEnv)); diff --git a/source/dnode/vnode/src/sma/smaEnv.c b/source/dnode/vnode/src/sma/smaEnv.c index 21d64238a1..8bf79ca30e 100644 --- a/source/dnode/vnode/src/sma/smaEnv.c +++ b/source/dnode/vnode/src/sma/smaEnv.c @@ -61,12 +61,23 @@ int32_t smaInit() { return TSDB_CODE_FAILED; } + int32_t type = (8 == POINTER_BYTES) ? TSDB_DATA_TYPE_UBIGINT : TSDB_DATA_TYPE_UINT; + smaMgmt.refHash = taosHashInit(1, taosGetDefaultHashFunction(type), true, HASH_ENTRY_LOCK); + if (!smaMgmt.refHash) { + taosCloseRef(smaMgmt.rsetId); + atomic_store_8(&smaMgmt.inited, 0); + smaError("failed to init sma tmr hanle since %s", terrstr()); + return TSDB_CODE_FAILED; + } + // init fetch timer handle smaMgmt.tmrHandle = taosTmrInit(10000, 100, 10000, "RSMA"); if (!smaMgmt.tmrHandle) { taosCloseRef(smaMgmt.rsetId); + taosHashCleanup(smaMgmt.refHash); + smaMgmt.refHash = NULL; atomic_store_8(&smaMgmt.inited, 0); - smaError("failed to init sma tmr hanle since %s", terrstr()); + smaError("failed to init sma tmr handle since %s", terrstr()); return TSDB_CODE_FAILED; } @@ -95,6 +106,7 @@ void smaCleanUp() { if (old == 1) { taosCloseRef(smaMgmt.rsetId); + taosHashCleanup(smaMgmt.refHash); taosTmrCleanUp(smaMgmt.tmrHandle); smaInfo("sma mgmt env is cleaned up, rsetId:%d, tmrHandle:%p", smaMgmt.rsetId, smaMgmt.tmrHandle); atomic_store_8(&smaMgmt.inited, 0); @@ -197,6 +209,21 @@ int32_t tdUnRefRSmaInfo(SSma *pSma, SRSmaInfo *pRSmaInfo) { return 0; } +static void tRSmaInfoHashFreeNode(void *data) { + SRSmaInfo *pRSmaInfo = NULL; + SRSmaInfoItem *pItem = NULL; + + if ((pRSmaInfo = *(SRSmaInfo **)data)) { + if ((pItem = RSMA_INFO_ITEM((SRSmaInfo *)pRSmaInfo, 0)) && pItem->level) { + taosHashRemove(smaMgmt.refHash, &pItem, POINTER_BYTES); + } + if ((pItem = RSMA_INFO_ITEM((SRSmaInfo *)pRSmaInfo, 1)) && pItem->level) { + taosHashRemove(smaMgmt.refHash, &pItem, POINTER_BYTES); + } + tdFreeRSmaInfo(NULL, pRSmaInfo, true); + } +} + static int32_t tdInitSmaStat(SSmaStat **pSmaStat, int8_t smaType, const SSma *pSma) { ASSERT(pSmaStat != NULL); @@ -242,12 +269,13 @@ static int32_t tdInitSmaStat(SSmaStat **pSmaStat, int8_t smaType, const SSma *pS if (!RSMA_INFO_HASH(pRSmaStat)) { return TSDB_CODE_FAILED; } + taosHashSetFreeFp(RSMA_INFO_HASH(pRSmaStat), tRSmaInfoHashFreeNode); if (tdRsmaStartExecutor(pSma) < 0) { return TSDB_CODE_FAILED; } - if (!(pRSmaStat->fs.aQTaskInf = taosArrayInit(1, sizeof(SQTaskFile)))) { + if (!(RSMA_FS(pRSmaStat)->aQTaskInf = taosArrayInit(1, sizeof(SQTaskFile)))) { terrno = TSDB_CODE_OUT_OF_MEMORY; return TSDB_CODE_FAILED; } @@ -285,14 +313,6 @@ static void tdDestroyRSmaStat(void *pRSmaStat) { tsem_destroy(&(pStat->notEmpty)); // step 2: destroy the rsma info and associated fetch tasks - if (taosHashGetSize(RSMA_INFO_HASH(pStat)) > 0) { - void *infoHash = taosHashIterate(RSMA_INFO_HASH(pStat), NULL); - while (infoHash) { - SRSmaInfo *pSmaInfo = *(SRSmaInfo **)infoHash; - tdFreeRSmaInfo(pSma, pSmaInfo, true); - infoHash = taosHashIterate(RSMA_INFO_HASH(pStat), infoHash); - } - } taosHashCleanup(RSMA_INFO_HASH(pStat)); // step 3: wait for all triggered fetch tasks to finish @@ -315,7 +335,7 @@ static void tdDestroyRSmaStat(void *pRSmaStat) { tdRsmaStopExecutor(pSma); // step 5: - tdRSmaFSClose(&pStat->fs); + tdRSmaFSClose(RSMA_FS(pStat)); // step 6: free pStat taosMemoryFreeClear(pStat); @@ -347,7 +367,7 @@ static int32_t tdDestroySmaState(SSmaStat *pSmaStat, int8_t smaType) { SRSmaStat *pRSmaStat = &pSmaStat->rsmaStat; int32_t vid = SMA_VID(pRSmaStat->pSma); int64_t refId = RSMA_REF_ID(pRSmaStat); - if (taosRemoveRef(smaMgmt.rsetId, RSMA_REF_ID(pRSmaStat)) < 0) { + if (taosRemoveRef(smaMgmt.rsetId, refId) < 0) { smaError("vgId:%d, remove refId:%" PRIi64 " from rsmaRef:%" PRIi32 " failed since %s", vid, refId, smaMgmt.rsetId, terrstr()); } else { diff --git a/source/dnode/vnode/src/sma/smaFS.c b/source/dnode/vnode/src/sma/smaFS.c index 2ab189bc8c..acb529bb9e 100644 --- a/source/dnode/vnode/src/sma/smaFS.c +++ b/source/dnode/vnode/src/sma/smaFS.c @@ -18,7 +18,8 @@ // ================================================================================================= static int32_t tdFetchQTaskInfoFiles(SSma *pSma, int64_t version, SArray **output); - +static int32_t tdQTaskInfCmprFn1(const void *p1, const void *p2); +static int32_t tdQTaskInfCmprFn2(const void *p1, const void *p2); /** * @brief Open RSma FS from qTaskInfo files * @@ -32,7 +33,6 @@ int32_t tdRSmaFSOpen(SSma *pSma, int64_t version) { SSmaEnv *pEnv = SMA_RSMA_ENV(pSma); SRSmaStat *pStat = NULL; SArray *output = NULL; - SRSmaFS *fs = NULL; if (!pEnv) { return TSDB_CODE_SUCCESS; @@ -43,14 +43,13 @@ int32_t tdRSmaFSOpen(SSma *pSma, int64_t version) { } pStat = (SRSmaStat *)SMA_ENV_STAT(pEnv); - fs = &pStat->fs; for (int32_t i = 0; i < taosArrayGetSize(output); ++i) { int32_t vid = 0; int64_t version = -1; sscanf((const char *)taosArrayGetP(output, i), "v%dqinfo.v%" PRIi64, &vid, &version); SQTaskFile qTaskFile = {.version = version, .nRef = 1}; - if ((terrno = tdRSmaFSUpsertQFile(fs, &qTaskFile)) < 0) { + if ((terrno = tdRSmaFSUpsertQTaskFile(RSMA_FS(pStat), &qTaskFile)) < 0) { goto _end; } smaInfo("vgId:%d, open fs, version:%" PRIi64 ", ref:%" PRIi64, TD_VID(pVnode), qTaskFile.version, qTaskFile.nRef); @@ -62,7 +61,7 @@ _end: taosMemoryFreeClear(ptr); } taosArrayDestroy(output); - + if (terrno != 0) { smaError("vgId:%d, open rsma fs failed since %s", TD_VID(pVnode), terrstr()); return TSDB_CODE_FAILED; @@ -72,8 +71,56 @@ _end: void tdRSmaFSClose(SRSmaFS *fs) { taosArrayDestroy(fs->aQTaskInf); } +static int32_t tdQTaskInfCmprFn1(const void *p1, const void *p2) { + if (*(int64_t *)p1 < ((SQTaskFile *)p2)->version) { + return -1; + } else if (*(int64_t *)p1 > ((SQTaskFile *)p2)->version) { + return 1; + } + return 0; +} + +int32_t tdRSmaFSRef(SSma *pSma, SRSmaStat *pStat, int64_t version) { + SArray *aQTaskInf = RSMA_FS(pStat)->aQTaskInf; + SQTaskFile *pTaskF = NULL; + int32_t oldVal = 0; + + taosRLockLatch(RSMA_FS_LOCK(pStat)); + if ((pTaskF = taosArraySearch(aQTaskInf, &version, tdQTaskInfCmprFn1, TD_EQ))) { + oldVal = atomic_fetch_add_32(&pTaskF->nRef, 1); + ASSERT(oldVal > 0); + } + taosRUnLockLatch(RSMA_FS_LOCK(pStat)); + return oldVal; +} + +void tdRSmaFSUnRef(SSma *pSma, SRSmaStat *pStat, int64_t version) { + SVnode *pVnode = pSma->pVnode; + SArray *aQTaskInf = RSMA_FS(pStat)->aQTaskInf; + char qTaskFullName[TSDB_FILENAME_LEN]; + SQTaskFile *pTaskF = NULL; + int32_t idx = -1; + + taosWLockLatch(RSMA_FS_LOCK(pStat)); + if ((idx = taosArraySearchIdx(aQTaskInf, &version, tdQTaskInfCmprFn1, TD_EQ)) >= 0) { + ASSERT(idx < taosArrayGetSize(aQTaskInf)); + pTaskF = taosArrayGet(aQTaskInf, idx); + if (atomic_sub_fetch_32(&pTaskF->nRef, 1) <= 0) { + tdRSmaQTaskInfoGetFullName(TD_VID(pVnode), pTaskF->version, tfsGetPrimaryPath(pVnode->pTfs), qTaskFullName); + if (taosRemoveFile(qTaskFullName) < 0) { + smaWarn("vgId:%d, failed to remove %s since %s", TD_VID(pVnode), qTaskFullName, + tstrerror(TAOS_SYSTEM_ERROR(errno))); + } else { + smaDebug("vgId:%d, success to remove %s", TD_VID(pVnode), qTaskFullName); + } + taosArrayRemove(aQTaskInf, idx); + } + } + taosWUnLockLatch(RSMA_FS_LOCK(pStat)); +} + /** - * @brief Fetch qtaskfiles no more than version + * @brief Fetch qtaskfiles LE than version * * @param pSma * @param version @@ -100,7 +147,7 @@ static int32_t tdFetchQTaskInfoFiles(SSma *pSma, int64_t version, SArray **outpu return TSDB_CODE_FAILED; } - if ((pDir = taosOpenDir(dir)) == NULL) { + if (!(pDir = taosOpenDir(dir))) { regfree(®ex); terrno = TAOS_SYSTEM_ERROR(errno); smaDebug("vgId:%d, fetch qtask files, open dir %s failed since %s", TD_VID(pVnode), dir, terrstr()); @@ -110,7 +157,7 @@ static int32_t tdFetchQTaskInfoFiles(SSma *pSma, int64_t version, SArray **outpu int32_t dirLen = strlen(dir); char *dirEnd = POINTER_SHIFT(dir, dirLen); regmatch_t regMatch[2]; - while ((pDirEntry = taosReadDir(pDir)) != NULL) { + while ((pDirEntry = taosReadDir(pDir))) { char *entryName = taosGetDirEntryName(pDirEntry); if (!entryName) { continue; @@ -140,6 +187,7 @@ static int32_t tdFetchQTaskInfoFiles(SSma *pSma, int64_t version, SArray **outpu terrno = TSDB_CODE_OUT_OF_MEMORY; goto _end; } + } else { } } else if (code == REG_NOMATCH) { // not match @@ -154,14 +202,13 @@ static int32_t tdFetchQTaskInfoFiles(SSma *pSma, int64_t version, SArray **outpu goto _end; } } - _end: taosCloseDir(&pDir); regfree(®ex); return terrno == 0 ? TSDB_CODE_SUCCESS : TSDB_CODE_FAILED; } -static int32_t tdQTaskFileCmprFn(const void *p1, const void *p2) { +static int32_t tdQTaskFileCmprFn2(const void *p1, const void *p2) { if (((SQTaskFile *)p1)->version < ((SQTaskFile *)p2)->version) { return -1; } else if (((SQTaskFile *)p1)->version > ((SQTaskFile *)p2)->version) { @@ -171,15 +218,15 @@ static int32_t tdQTaskFileCmprFn(const void *p1, const void *p2) { return 0; } -int32_t tdRSmaFSUpsertQFile(SRSmaFS *pFS, SQTaskFile *qTaskFile) { +int32_t tdRSmaFSUpsertQTaskFile(SRSmaFS *pFS, SQTaskFile *qTaskFile) { int32_t code = 0; - int32_t idx = taosArraySearchIdx(pFS->aQTaskInf, qTaskFile, tdQTaskFileCmprFn, TD_GE); + int32_t idx = taosArraySearchIdx(pFS->aQTaskInf, qTaskFile, tdQTaskFileCmprFn2, TD_GE); if (idx < 0) { idx = taosArrayGetSize(pFS->aQTaskInf); } else { SQTaskFile *pTaskF = (SQTaskFile *)taosArrayGet(pFS->aQTaskInf, idx); - int32_t c = tdQTaskFileCmprFn(pTaskF, qTaskFile); + int32_t c = tdQTaskFileCmprFn2(pTaskF, qTaskFile); if (c == 0) { pTaskF->nRef = qTaskFile->nRef; pTaskF->version = qTaskFile->version; @@ -195,4 +242,4 @@ int32_t tdRSmaFSUpsertQFile(SRSmaFS *pFS, SQTaskFile *qTaskFile) { _exit: return code; -} +} \ No newline at end of file diff --git a/source/dnode/vnode/src/sma/smaRollup.c b/source/dnode/vnode/src/sma/smaRollup.c index 977f7dda07..8fcc371153 100644 --- a/source/dnode/vnode/src/sma/smaRollup.c +++ b/source/dnode/vnode/src/sma/smaRollup.c @@ -20,7 +20,7 @@ #define RSMA_QTASKEXEC_SMOOTH_SIZE (100) // cnt #define RSMA_SUBMIT_BATCH_SIZE (1024) // cnt #define RSMA_FETCH_DELAY_MAX (900000) // ms -#define RSMA_FETCH_ACTIVE_MAX (1800) // ms +#define RSMA_FETCH_ACTIVE_MAX (1000) // ms #define RSMA_FETCH_INTERVAL (5000) // ms SSmaMgmt smaMgmt = { @@ -42,7 +42,7 @@ static int32_t tdExecuteRSmaImpl(SSma *pSma, const void *pMsg, int32_t msgSiz static SRSmaInfo *tdAcquireRSmaInfoBySuid(SSma *pSma, int64_t suid); static void tdReleaseRSmaInfo(SSma *pSma, SRSmaInfo *pInfo); static void tdFreeRSmaSubmitItems(SArray *pItems); -static int32_t tdRSmaFetchAllResult(SSma *pSma, SRSmaInfo *pInfo, SArray *pSubmitArr); +static int32_t tdRSmaFetchAllResult(SSma *pSma, SRSmaInfo *pInfo); static int32_t tdRSmaExecAndSubmitResult(SSma *pSma, qTaskInfo_t taskInfo, SRSmaInfoItem *pItem, STSchema *pTSchema, int64_t suid); static void tdRSmaFetchTrigger(void *param, void *tmrId); @@ -121,27 +121,27 @@ static void tdRSmaQTaskInfoFree(qTaskInfo_t *taskHandle, int32_t vgId, int32_t l */ void *tdFreeRSmaInfo(SSma *pSma, SRSmaInfo *pInfo, bool isDeepFree) { if (pInfo) { + int32_t vid = pSma ? SMA_VID(pSma) : -1; for (int32_t i = 0; i < TSDB_RETENTION_L2; ++i) { SRSmaInfoItem *pItem = &pInfo->items[i]; if (isDeepFree && pItem->tmrId) { - smaDebug("vgId:%d, stop fetch timer %p for table %" PRIi64 " level %d", SMA_VID(pSma), pInfo->suid, - pItem->tmrId, i + 1); + smaDebug("vgId:%d, stop fetch timer %p for table %" PRIi64 " level %d", vid, pItem->tmrId, pInfo->suid, i + 1); taosTmrStopA(&pItem->tmrId); } if (isDeepFree && pInfo->taskInfo[i]) { - tdRSmaQTaskInfoFree(&pInfo->taskInfo[i], SMA_VID(pSma), i + 1); + tdRSmaQTaskInfoFree(&pInfo->taskInfo[i], vid, i + 1); } else { - smaDebug("vgId:%d, table %" PRIi64 " no need to destroy rsma info level %d since empty taskInfo", SMA_VID(pSma), + smaDebug("vgId:%d, table %" PRIi64 " no need to destroy rsma info level %d since empty taskInfo", vid, pInfo->suid, i + 1); } if (pInfo->iTaskInfo[i]) { - tdRSmaQTaskInfoFree(&pInfo->iTaskInfo[i], SMA_VID(pSma), i + 1); + tdRSmaQTaskInfoFree(&pInfo->iTaskInfo[i], vid, i + 1); } else { - smaDebug("vgId:%d, table %" PRIi64 " no need to destroy rsma info level %d since empty iTaskInfo", - SMA_VID(pSma), pInfo->suid, i + 1); + smaDebug("vgId:%d, table %" PRIi64 " no need to destroy rsma info level %d since empty iTaskInfo", vid, + pInfo->suid, i + 1); } } if (isDeepFree) { @@ -315,10 +315,17 @@ static int32_t tdSetRSmaInfoItemParams(SSma *pSma, SRSmaParam *param, SRSmaStat } pItem->level = idx == 0 ? TSDB_RETENTION_L1 : TSDB_RETENTION_L2; + ASSERT(pItem->level > 0); + + SRSmaRef rsmaRef = {.refId = pStat->refId, .suid = pRSmaInfo->suid}; + taosHashPut(smaMgmt.refHash, &pItem, POINTER_BYTES, &rsmaRef, sizeof(rsmaRef)); + taosTmrReset(tdRSmaFetchTrigger, pItem->maxDelay, pItem, smaMgmt.tmrHandle, &pItem->tmrId); - smaInfo("vgId:%d, table:%" PRIi64 " level:%" PRIi8 " maxdelay:%" PRIi64 " watermark:%" PRIi64 + + smaInfo("vgId:%d, item:%p table:%" PRIi64 " level:%" PRIi8 " maxdelay:%" PRIi64 " watermark:%" PRIi64 ", finally maxdelay:%" PRIi32, - TD_VID(pVnode), pRSmaInfo->suid, idx + 1, param->maxdelay[idx], param->watermark[idx], pItem->maxDelay); + TD_VID(pVnode), pItem, pRSmaInfo->suid, idx + 1, param->maxdelay[idx], param->watermark[idx], + pItem->maxDelay); } return TSDB_CODE_SUCCESS; } @@ -385,7 +392,6 @@ int32_t tdRSmaProcessCreateImpl(SSma *pSma, SRSmaParam *param, int64_t suid, con goto _err; } pRSmaInfo->suid = suid; - pRSmaInfo->refId = RSMA_REF_ID(pStat); T_REF_INIT_VAL(pRSmaInfo, 1); if (tdSetRSmaInfoItemParams(pSma, param, pStat, pRSmaInfo, 0) < 0) { @@ -656,15 +662,15 @@ static int32_t tdRSmaExecAndSubmitResult(SSma *pSma, qTaskInfo_t taskInfo, SRSma if (taosArrayGetSize(pResList) == 0) { if (terrno == 0) { - // smaDebug("vgId:%d, no rsma %" PRIi8 " data fetched yet", SMA_VID(pSma), pItem->level); + // smaDebug("vgId:%d, no rsma level %" PRIi8 " data fetched yet", SMA_VID(pSma), pItem->level); } else { - smaDebug("vgId:%d, no rsma %" PRIi8 " data fetched since %s", SMA_VID(pSma), pItem->level, terrstr()); + smaDebug("vgId:%d, no rsma level %" PRIi8 " data fetched since %s", SMA_VID(pSma), pItem->level, terrstr()); goto _err; } break; } else { - smaDebug("vgId:%d, rsma %" PRIi8 " data fetched", SMA_VID(pSma), pItem->level); + smaDebug("vgId:%d, rsma level %" PRIi8 " data fetched", SMA_VID(pSma), pItem->level); } #if 0 char flag[10] = {0}; @@ -678,21 +684,22 @@ static int32_t tdRSmaExecAndSubmitResult(SSma *pSma, qTaskInfo_t taskInfo, SRSma // TODO: the schema update should be handled later(TD-17965) if (buildSubmitReqFromDataBlock(&pReq, output, pTSchema, SMA_VID(pSma), suid) < 0) { - smaError("vgId:%d, build submit req for rsma stable %" PRIi64 " level %" PRIi8 " failed since %s", - SMA_VID(pSma), suid, pItem->level, terrstr()); + smaError("vgId:%d, build submit req for rsma table %" PRIi64 " level %" PRIi8 " failed since %s", SMA_VID(pSma), + suid, pItem->level, terrstr()); goto _err; } if (pReq && tdProcessSubmitReq(sinkTsdb, output->info.version, pReq) < 0) { taosMemoryFreeClear(pReq); - smaError("vgId:%d, process submit req for rsma stable %" PRIi64 " level %" PRIi8 " failed since %s", + smaError("vgId:%d, process submit req for rsma table %" PRIi64 " level %" PRIi8 " failed since %s", SMA_VID(pSma), suid, pItem->level, terrstr()); goto _err; } - taosMemoryFreeClear(pReq); - smaDebug("vgId:%d, process submit req for rsma table %" PRIi64 " level %" PRIi8 " version:%" PRIi64, - SMA_VID(pSma), suid, pItem->level, output->info.version); + smaDebug("vgId:%d, process submit req for rsma table %" PRIi64 " level %" PRIi8 " ver %" PRIi64 " len %" PRIu32, + SMA_VID(pSma), suid, pItem->level, output->info.version, htonl(pReq->header.contLen)); + + taosMemoryFreeClear(pReq); } } @@ -1565,26 +1572,46 @@ _err: * @param tmrId */ static void tdRSmaFetchTrigger(void *param, void *tmrId) { - SRSmaInfoItem *pItem = param; + SRSmaRef *pRSmaRef = NULL; SSma *pSma = NULL; - SRSmaInfo *pRSmaInfo = tdGetRSmaInfoByItem(pItem); + SRSmaStat *pStat = NULL; + SRSmaInfo *pRSmaInfo = NULL; + SRSmaInfoItem *pItem = NULL; - if (RSMA_INFO_IS_DEL(pRSmaInfo)) { - smaDebug("rsma fetch task not start since rsma info already deleted, rsetId:%" PRIi64 " refId:%d)", smaMgmt.rsetId, - pRSmaInfo->refId); + if (!(pRSmaRef = taosHashGet(smaMgmt.refHash, ¶m, POINTER_BYTES))) { + smaDebug("rsma fetch task not start since rsma info item:%p not exist in refHash:%p, rsetId:%" PRIi64, param, + *(int64_t *)¶m, smaMgmt.refHash, smaMgmt.rsetId); return; } - SRSmaStat *pStat = (SRSmaStat *)tdAcquireSmaRef(smaMgmt.rsetId, pRSmaInfo->refId); - - if (!pStat) { + if (!(pStat = (SRSmaStat *)tdAcquireSmaRef(smaMgmt.rsetId, pRSmaRef->refId))) { smaDebug("rsma fetch task not start since rsma stat already destroyed, rsetId:%" PRIi64 " refId:%d)", - smaMgmt.rsetId, pRSmaInfo->refId); + smaMgmt.rsetId, pRSmaRef->refId); // pRSmaRef freed in taosHashRemove + taosHashRemove(smaMgmt.refHash, ¶m, POINTER_BYTES); return; } pSma = pStat->pSma; + if (!(pRSmaInfo = tdAcquireRSmaInfoBySuid(pSma, pRSmaRef->suid))) { + smaDebug("rsma fetch task not start since rsma info not exist, rsetId:%" PRIi64 " refId:%d)", smaMgmt.rsetId, + pRSmaRef->refId); // pRSmaRef freed in taosHashRemove + tdReleaseSmaRef(smaMgmt.rsetId, pRSmaRef->refId); + taosHashRemove(smaMgmt.refHash, ¶m, POINTER_BYTES); + return; + } + + if (RSMA_INFO_IS_DEL(pRSmaInfo)) { + smaDebug("rsma fetch task not start since rsma info already deleted, rsetId:%" PRIi64 " refId:%d)", smaMgmt.rsetId, + pRSmaRef->refId); // pRSmaRef freed in taosHashRemove + tdReleaseRSmaInfo(pSma, pRSmaInfo); + tdReleaseSmaRef(smaMgmt.rsetId, pRSmaRef->refId); + taosHashRemove(smaMgmt.refHash, ¶m, POINTER_BYTES); + return; + } + + pItem = *(SRSmaInfoItem **)¶m; + // if rsma trigger stat in paused, cancelled or finished, not start fetch task int8_t rsmaTriggerStat = atomic_load_8(RSMA_TRIGGER_STAT(pStat)); switch (rsmaTriggerStat) { @@ -1592,11 +1619,12 @@ static void tdRSmaFetchTrigger(void *param, void *tmrId) { case TASK_TRIGGER_STAT_CANCELLED: { smaDebug("vgId:%d, rsma fetch task not start for level %" PRIi8 " since stat is %" PRIi8 ", rsetId rsetId:%" PRIi64 " refId:%d", - SMA_VID(pSma), pItem->level, rsmaTriggerStat, smaMgmt.rsetId, pRSmaInfo->refId); + SMA_VID(pSma), pItem->level, rsmaTriggerStat, smaMgmt.rsetId, pRSmaRef->refId); if (rsmaTriggerStat == TASK_TRIGGER_STAT_PAUSED) { taosTmrReset(tdRSmaFetchTrigger, RSMA_FETCH_INTERVAL, pItem, smaMgmt.tmrHandle, &pItem->tmrId); } - tdReleaseSmaRef(smaMgmt.rsetId, pRSmaInfo->refId); + tdReleaseRSmaInfo(pSma, pRSmaInfo); + tdReleaseSmaRef(smaMgmt.rsetId, pRSmaRef->refId); return; } default: @@ -1607,8 +1635,8 @@ static void tdRSmaFetchTrigger(void *param, void *tmrId) { atomic_val_compare_exchange_8(&pItem->triggerStat, TASK_TRIGGER_STAT_ACTIVE, TASK_TRIGGER_STAT_INACTIVE); switch (fetchTriggerStat) { case TASK_TRIGGER_STAT_ACTIVE: { - smaDebug("vgId:%d, rsma fetch task planned for level:%" PRIi8 " suid:%" PRIi64 " since stat is active", - SMA_VID(pSma), pItem->level, pRSmaInfo->suid); + smaInfo("vgId:%d, rsma fetch task planned for level:%" PRIi8 " suid:%" PRIi64 " since stat is active", + SMA_VID(pSma), pItem->level, pRSmaInfo->suid); // async process pItem->fetchLevel = pItem->level; #if 0 @@ -1641,11 +1669,11 @@ static void tdRSmaFetchTrigger(void *param, void *tmrId) { _end: taosTmrReset(tdRSmaFetchTrigger, pItem->maxDelay, pItem, smaMgmt.tmrHandle, &pItem->tmrId); - tdReleaseSmaRef(smaMgmt.rsetId, pRSmaInfo->refId); + tdReleaseRSmaInfo(pSma, pRSmaInfo); + tdReleaseSmaRef(smaMgmt.rsetId, pRSmaRef->refId); } static void tdFreeRSmaSubmitItems(SArray *pItems) { - ASSERT(taosArrayGetSize(pItems) > 0); for (int32_t i = 0; i < taosArrayGetSize(pItems); ++i) { taosFreeQitem(*(void **)taosArrayGet(pItems, i)); } @@ -1657,10 +1685,9 @@ static void tdFreeRSmaSubmitItems(SArray *pItems) { * * @param pSma * @param pInfo - * @param pSubmitArr * @return int32_t */ -static int32_t tdRSmaFetchAllResult(SSma *pSma, SRSmaInfo *pInfo, SArray *pSubmitArr) { +static int32_t tdRSmaFetchAllResult(SSma *pSma, SRSmaInfo *pInfo) { SSDataBlock dataBlock = {.info.type = STREAM_GET_ALL}; for (int8_t i = 1; i <= TSDB_RETENTION_L2; ++i) { SRSmaInfoItem *pItem = RSMA_INFO_ITEM(pInfo, i - 1); @@ -1677,7 +1704,7 @@ static int32_t tdRSmaFetchAllResult(SSma *pSma, SRSmaInfo *pInfo, SArray *pSubmi SMA_VID(pSma), pInfo->suid, i, pItem->nSkipped, pItem->maxDelay); } else if (((curMs - pInfo->lastRecv) < RSMA_FETCH_ACTIVE_MAX)) { ++pItem->nSkipped; - smaDebug("vgId:%d, suid:%" PRIi64 " level:%" PRIi8 " curMs:%" PRIi64 " lastRecv:%" PRIi64 ", fetch skipped ", + smaInfo("vgId:%d, suid:%" PRIi64 " level:%" PRIi8 " curMs:%" PRIi64 " lastRecv:%" PRIi64 ", fetch skipped ", SMA_VID(pSma), pInfo->suid, i, curMs, pInfo->lastRecv); continue; } else { @@ -1706,10 +1733,8 @@ static int32_t tdRSmaFetchAllResult(SSma *pSma, SRSmaInfo *pInfo, SArray *pSubmi } _end: - tdReleaseRSmaInfo(pSma, pInfo); return TSDB_CODE_SUCCESS; _err: - tdReleaseRSmaInfo(pSma, pInfo); return TSDB_CODE_FAILED; } @@ -1806,7 +1831,7 @@ int32_t tdRSmaProcessExecImpl(SSma *pSma, ERsmaExecType type) { if (oldStat == 0 || ((oldStat == 2) && atomic_load_8(RSMA_TRIGGER_STAT(pRSmaStat)) < TASK_TRIGGER_STAT_PAUSED)) { atomic_fetch_add_32(&pRSmaStat->nFetchAll, 1); - tdRSmaFetchAllResult(pSma, pInfo, pSubmitArr); + tdRSmaFetchAllResult(pSma, pInfo); if (0 == atomic_sub_fetch_32(&pRSmaStat->nFetchAll, 1)) { atomic_store_8(RSMA_COMMIT_STAT(pRSmaStat), 0); } diff --git a/source/dnode/vnode/src/sma/smaSnapshot.c b/source/dnode/vnode/src/sma/smaSnapshot.c index 75ec26bef2..5a0167a75f 100644 --- a/source/dnode/vnode/src/sma/smaSnapshot.c +++ b/source/dnode/vnode/src/sma/smaSnapshot.c @@ -15,11 +15,13 @@ #include "sma.h" -static int32_t rsmaSnapReadQTaskInfo(SRsmaSnapReader* pReader, uint8_t** ppData); -static int32_t rsmaSnapWriteQTaskInfo(SRsmaSnapWriter* pWriter, uint8_t* pData, uint32_t nData); +static int32_t rsmaSnapReadQTaskInfo(SRSmaSnapReader* pReader, uint8_t** ppData); +static int32_t rsmaSnapWriteQTaskInfo(SRSmaSnapWriter* pWriter, uint8_t* pData, uint32_t nData); +static int32_t rsmaQTaskInfSnapReaderOpen(SRSmaSnapReader* pReader, int64_t version); +static int32_t rsmaQTaskInfSnapReaderClose(SQTaskFReader** ppReader); -// SRsmaSnapReader ======================================== -struct SRsmaSnapReader { +// SRSmaSnapReader ======================================== +struct SRSmaSnapReader { SSma* pSma; int64_t sver; int64_t ever; @@ -33,13 +35,13 @@ struct SRsmaSnapReader { SQTaskFReader* pQTaskFReader; }; -int32_t rsmaSnapReaderOpen(SSma* pSma, int64_t sver, int64_t ever, SRsmaSnapReader** ppReader) { +int32_t rsmaSnapReaderOpen(SSma* pSma, int64_t sver, int64_t ever, SRSmaSnapReader** ppReader) { int32_t code = 0; SVnode* pVnode = pSma->pVnode; - SRsmaSnapReader* pReader = NULL; + SRSmaSnapReader* pReader = NULL; // alloc - pReader = (SRsmaSnapReader*)taosMemoryCalloc(1, sizeof(*pReader)); + pReader = (SRSmaSnapReader*)taosMemoryCalloc(1, sizeof(*pReader)); if (pReader == NULL) { code = TSDB_CODE_OUT_OF_MEMORY; goto _err; @@ -48,7 +50,7 @@ int32_t rsmaSnapReaderOpen(SSma* pSma, int64_t sver, int64_t ever, SRsmaSnapRead pReader->sver = sver; pReader->ever = ever; - // rsma1/rsma2 + // open rsma1/rsma2 for (int32_t i = 0; i < TSDB_RETENTION_L2; ++i) { if (pSma->pRSmaTsdb[i]) { code = tsdbSnapReaderOpen(pSma->pRSmaTsdb[i], sver, ever, i == 0 ? SNAP_DATA_RSMA1 : SNAP_DATA_RSMA2, @@ -59,51 +61,112 @@ int32_t rsmaSnapReaderOpen(SSma* pSma, int64_t sver, int64_t ever, SRsmaSnapRead } } - // qtaskinfo - // 1. add ref to qtaskinfo.v${ever} if exists and then start to replicate - char qTaskInfoFullName[TSDB_FILENAME_LEN]; - tdRSmaQTaskInfoGetFullName(TD_VID(pVnode), ever, tfsGetPrimaryPath(pVnode->pTfs), qTaskInfoFullName); - - if (!taosCheckExistFile(qTaskInfoFullName)) { - smaInfo("vgId:%d, vnode snapshot rsma reader for qtaskinfo not need as %s not exists", TD_VID(pVnode), - qTaskInfoFullName); - } else { - pReader->pQTaskFReader = taosMemoryCalloc(1, sizeof(SQTaskFReader)); - if (!pReader->pQTaskFReader) { - code = TSDB_CODE_OUT_OF_MEMORY; - goto _err; - } - - TdFilePtr qTaskF = taosOpenFile(qTaskInfoFullName, TD_FILE_READ); - if (!qTaskF) { - code = TAOS_SYSTEM_ERROR(errno); - goto _err; - } - pReader->pQTaskFReader->pReadH = qTaskF; -#if 0 - SQTaskFile* pQTaskF = &pReader->pQTaskFReader->fTask; - pQTaskF->nRef = 1; -#endif + // open qtaskinfo + if ((code = rsmaQTaskInfSnapReaderOpen(pReader, ever)) < 0) { + goto _err; } *ppReader = pReader; - smaInfo("vgId:%d, vnode snapshot rsma reader opened %s succeed", TD_VID(pVnode), qTaskInfoFullName); + return TSDB_CODE_SUCCESS; _err: - smaError("vgId:%d, vnode snapshot rsma reader opened failed since %s", TD_VID(pVnode), tstrerror(code)); + smaError("vgId:%d, vnode snapshot rsma reader open failed since %s", TD_VID(pVnode), tstrerror(code)); return TSDB_CODE_FAILED; } -static int32_t rsmaSnapReadQTaskInfo(SRsmaSnapReader* pReader, uint8_t** ppBuf) { +static int32_t rsmaQTaskInfSnapReaderOpen(SRSmaSnapReader* pReader, int64_t version) { + int32_t code = 0; + SSma* pSma = pReader->pSma; + SVnode* pVnode = pSma->pVnode; + SSmaEnv* pEnv = NULL; + SRSmaStat* pStat = NULL; + + if (!(pEnv = SMA_RSMA_ENV(pSma))) { + smaInfo("vgId:%d, vnode snapshot rsma reader for qtaskinfo version %" PRIi64 " not need as env is NULL", + TD_VID(pVnode), version); + return TSDB_CODE_SUCCESS; + } + + pStat = (SRSmaStat*)SMA_ENV_STAT(pEnv); + + int32_t ref = tdRSmaFSRef(pReader->pSma, pStat, version); + if (ref < 1) { + smaInfo("vgId:%d, vnode snapshot rsma reader for qtaskinfo version %" PRIi64 " not need as ref is %d", + TD_VID(pVnode), version, ref); + return TSDB_CODE_SUCCESS; + } + + char qTaskInfoFullName[TSDB_FILENAME_LEN]; + tdRSmaQTaskInfoGetFullName(TD_VID(pVnode), version, tfsGetPrimaryPath(pVnode->pTfs), qTaskInfoFullName); + + if (!taosCheckExistFile(qTaskInfoFullName)) { + tdRSmaFSUnRef(pSma, pStat, version); + smaInfo("vgId:%d, vnode snapshot rsma reader for qtaskinfo version %" PRIi64 " not need as %s not exists", + TD_VID(pVnode), qTaskInfoFullName); + return TSDB_CODE_SUCCESS; + } + + pReader->pQTaskFReader = taosMemoryCalloc(1, sizeof(SQTaskFReader)); + if (!pReader->pQTaskFReader) { + code = TSDB_CODE_OUT_OF_MEMORY; + goto _end; + } + + TdFilePtr fp = taosOpenFile(qTaskInfoFullName, TD_FILE_READ); + if (!fp) { + code = TAOS_SYSTEM_ERROR(errno); + taosMemoryFreeClear(pReader->pQTaskFReader); + goto _end; + } + + pReader->pQTaskFReader->pReadH = fp; + pReader->pQTaskFReader->pSma = pSma; + pReader->pQTaskFReader->version = pReader->ever; + +_end: + if (code < 0) { + tdRSmaFSUnRef(pSma, pStat, version); + smaError("vgId:%d, vnode snapshot rsma reader open %s succeed", TD_VID(pVnode), qTaskInfoFullName); + return TSDB_CODE_FAILED; + } + + smaInfo("vgId:%d, vnode snapshot rsma reader open %s succeed", TD_VID(pVnode), qTaskInfoFullName); + return TSDB_CODE_SUCCESS; +} + +static int32_t rsmaQTaskInfSnapReaderClose(SQTaskFReader** ppReader) { + if (!(*ppReader)) { + return TSDB_CODE_SUCCESS; + } + + SSma* pSma = (*ppReader)->pSma; + SRSmaStat* pStat = SMA_RSMA_STAT(pSma); + int64_t version = (*ppReader)->version; + + taosCloseFile(&(*ppReader)->pReadH); + tdRSmaFSUnRef(pSma, pStat, version); + taosMemoryFreeClear(*ppReader); + smaInfo("vgId:%d, vnode snapshot rsma reader closed for qTaskInfo version %" PRIi64, SMA_VID(pSma), version); + + return TSDB_CODE_SUCCESS; +} + +static int32_t rsmaSnapReadQTaskInfo(SRSmaSnapReader* pReader, uint8_t** ppBuf) { int32_t code = 0; SSma* pSma = pReader->pSma; int64_t n = 0; uint8_t* pBuf = NULL; SQTaskFReader* qReader = pReader->pQTaskFReader; + if (!qReader) { + *ppBuf = NULL; + smaInfo("vgId:%d, vnode snapshot rsma reader qtaskinfo, qTaskReader is NULL", SMA_VID(pSma)); + return 0; + } + if (!qReader->pReadH) { *ppBuf = NULL; - smaInfo("vgId:%d, vnode snapshot rsma reader qtaskinfo, readh is empty", SMA_VID(pSma)); + smaInfo("vgId:%d, vnode snapshot rsma reader qtaskinfo, readh is NULL", SMA_VID(pSma)); return 0; } @@ -153,7 +216,7 @@ _err: return code; } -int32_t rsmaSnapRead(SRsmaSnapReader* pReader, uint8_t** ppData) { +int32_t rsmaSnapRead(SRSmaSnapReader* pReader, uint8_t** ppData) { int32_t code = 0; *ppData = NULL; @@ -205,9 +268,9 @@ _err: return code; } -int32_t rsmaSnapReaderClose(SRsmaSnapReader** ppReader) { +int32_t rsmaSnapReaderClose(SRSmaSnapReader** ppReader) { int32_t code = 0; - SRsmaSnapReader* pReader = *ppReader; + SRSmaSnapReader* pReader = *ppReader; for (int32_t i = 0; i < TSDB_RETENTION_L2; ++i) { if (pReader->pDataReader[i]) { @@ -215,11 +278,7 @@ int32_t rsmaSnapReaderClose(SRsmaSnapReader** ppReader) { } } - if (pReader->pQTaskFReader) { - taosCloseFile(&pReader->pQTaskFReader->pReadH); - taosMemoryFreeClear(pReader->pQTaskFReader); - smaInfo("vgId:%d, vnode snapshot rsma reader closed for qTaskInfo", SMA_VID(pReader->pSma)); - } + rsmaQTaskInfSnapReaderClose(&pReader->pQTaskFReader); smaInfo("vgId:%d, vnode snapshot rsma reader closed", SMA_VID(pReader->pSma)); @@ -227,8 +286,8 @@ int32_t rsmaSnapReaderClose(SRsmaSnapReader** ppReader) { return code; } -// SRsmaSnapWriter ======================================== -struct SRsmaSnapWriter { +// SRSmaSnapWriter ======================================== +struct SRSmaSnapWriter { SSma* pSma; int64_t sver; int64_t ever; @@ -244,13 +303,13 @@ struct SRsmaSnapWriter { SQTaskFWriter* pQTaskFWriter; }; -int32_t rsmaSnapWriterOpen(SSma* pSma, int64_t sver, int64_t ever, SRsmaSnapWriter** ppWriter) { +int32_t rsmaSnapWriterOpen(SSma* pSma, int64_t sver, int64_t ever, SRSmaSnapWriter** ppWriter) { int32_t code = 0; - SRsmaSnapWriter* pWriter = NULL; + SRSmaSnapWriter* pWriter = NULL; SVnode* pVnode = pSma->pVnode; // alloc - pWriter = (SRsmaSnapWriter*)taosMemoryCalloc(1, sizeof(*pWriter)); + pWriter = (SRSmaSnapWriter*)taosMemoryCalloc(1, sizeof(*pWriter)); if (pWriter == NULL) { code = TSDB_CODE_OUT_OF_MEMORY; goto _err; @@ -301,9 +360,9 @@ _err: return code; } -int32_t rsmaSnapWriterClose(SRsmaSnapWriter** ppWriter, int8_t rollback) { +int32_t rsmaSnapWriterClose(SRSmaSnapWriter** ppWriter, int8_t rollback) { int32_t code = 0; - SRsmaSnapWriter* pWriter = *ppWriter; + SRSmaSnapWriter* pWriter = *ppWriter; SVnode* pVnode = pWriter->pSma->pVnode; if (rollback) { @@ -349,7 +408,7 @@ _err: return code; } -int32_t rsmaSnapWrite(SRsmaSnapWriter* pWriter, uint8_t* pData, uint32_t nData) { +int32_t rsmaSnapWrite(SRSmaSnapWriter* pWriter, uint8_t* pData, uint32_t nData) { int32_t code = 0; SSnapDataHdr* pHdr = (SSnapDataHdr*)pData; @@ -377,7 +436,7 @@ _err: return code; } -static int32_t rsmaSnapWriteQTaskInfo(SRsmaSnapWriter* pWriter, uint8_t* pData, uint32_t nData) { +static int32_t rsmaSnapWriteQTaskInfo(SRSmaSnapWriter* pWriter, uint8_t* pData, uint32_t nData) { int32_t code = 0; SQTaskFWriter* qWriter = pWriter->pQTaskFWriter; diff --git a/source/dnode/vnode/src/vnd/vnodeSnapshot.c b/source/dnode/vnode/src/vnd/vnodeSnapshot.c index 5a81f91919..03ab5e8285 100644 --- a/source/dnode/vnode/src/vnd/vnodeSnapshot.c +++ b/source/dnode/vnode/src/vnd/vnodeSnapshot.c @@ -39,7 +39,7 @@ struct SVSnapReader { SStreamStateReader *pStreamStateReader; // rsma int8_t rsmaDone; - SRsmaSnapReader *pRsmaReader; + SRSmaSnapReader *pRsmaReader; }; int32_t vnodeSnapReaderOpen(SVnode *pVnode, int64_t sver, int64_t ever, SVSnapReader **ppReader) { @@ -241,7 +241,7 @@ struct SVSnapWriter { SStreamTaskWriter *pStreamTaskWriter; SStreamStateWriter *pStreamStateWriter; // rsma - SRsmaSnapWriter *pRsmaSnapWriter; + SRSmaSnapWriter *pRsmaSnapWriter; }; int32_t vnodeSnapWriterOpen(SVnode *pVnode, int64_t sver, int64_t ever, SVSnapWriter **ppWriter) {