From 7ffafb5f169d03a641e9090ed38e3f7fcba4326c Mon Sep 17 00:00:00 2001 From: Hongze Cheng Date: Mon, 1 Feb 2021 13:59:42 +0800 Subject: [PATCH] refact --- src/tsdb/src/tsdbCommit.c | 58 +++++++++++++++++++++++++++++---------- src/tsdb/src/tsdbFS.c | 8 ++++++ 2 files changed, 51 insertions(+), 15 deletions(-) diff --git a/src/tsdb/src/tsdbCommit.c b/src/tsdb/src/tsdbCommit.c index 6c5d3ae015..13c0d568a5 100644 --- a/src/tsdb/src/tsdbCommit.c +++ b/src/tsdb/src/tsdbCommit.c @@ -52,7 +52,7 @@ static int tsdbCommitMeta(STsdbRepo *pRepo); static int tsdbUpdateMetaRecord(STsdbFS *pfs, SMFile *pMFile, uint64_t uid, void *cont, int contLen); static int tsdbDropMetaRecord(STsdbFS *pfs, SMFile *pMFile, uint64_t uid); static int tsdbCommitTSData(STsdbRepo *pRepo); -static int tsdbStartCommit(STsdbRepo *pRepo); +static void tsdbStartCommit(STsdbRepo *pRepo); static void tsdbEndCommit(STsdbRepo *pRepo, int eno); static int tsdbCommitToFile(SCommitH *pCommith, SDFileSet *pSet, int fid); static int tsdbCreateCommitIters(SCommitH *pCommith); @@ -84,10 +84,7 @@ static int tsdbApplyRtn(STsdbRepo *pRepo); static int tsdbApplyRtnOnFSet(STsdbRepo *pRepo, SDFileSet *pSet, SRtn *pRtn); void *tsdbCommitData(STsdbRepo *pRepo) { - if (tsdbStartCommit(pRepo) < 0) { - tsdbError("vgId:%d failed to commit data while startting to commit since %s", REPO_ID(pRepo), tstrerror(terrno)); - goto _err; - } + tsdbStartCommit(pRepo); // Commit to update meta file if (tsdbCommitMeta(pRepo) < 0) { @@ -138,11 +135,15 @@ static int tsdbCommitMeta(STsdbRepo *pRepo) { tsdbInitMFile(&mf, did, REPO_ID(pRepo), FS_TXN_VERSION(REPO_FS(pRepo))); if (tsdbCreateMFile(&mf, true) < 0) { + tsdbError("vgId:%d failed to create META file since %s", REPO_ID(pRepo), tstrerror(terrno)); return -1; } + + tsdbInfo("vgId:%d meta file %s is created to commit", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(&mf)); } else { tsdbInitMFileEx(&mf, pOMFile); if (tsdbOpenMFile(&mf, O_WRONLY) < 0) { + tsdbError("vgId:%d failed to open META file since %s", REPO_ID(pRepo), tstrerror(terrno)); return -1; } } @@ -154,12 +155,20 @@ static int tsdbCommitMeta(STsdbRepo *pRepo) { if (pAct->act == TSDB_UPDATE_META) { pCont = (SActCont *)POINTER_SHIFT(pAct, sizeof(SActObj)); if (tsdbUpdateMetaRecord(pfs, &mf, pAct->uid, (void *)(pCont->cont), pCont->len) < 0) { + tsdbError("vgId:%d failed to update META record, uid %" PRIu64 " since %s", REPO_ID(pRepo), pAct->uid, + tstrerror(terrno)); tsdbCloseMFile(&mf); + tsdbApplyMFileChange(&mf, pOMFile); + // TODO: need to reload metaCache return -1; } } else if (pAct->act == TSDB_DROP_META) { if (tsdbDropMetaRecord(pfs, &mf, pAct->uid) < 0) { + tsdbError("vgId:%d failed to drop META record, uid %" PRIu64 " since %s", REPO_ID(pRepo), pAct->uid, + tstrerror(terrno)); tsdbCloseMFile(&mf); + tsdbApplyMFileChange(&mf, pOMFile); + // TODO: need to reload metaCache return -1; } } else { @@ -168,6 +177,9 @@ static int tsdbCommitMeta(STsdbRepo *pRepo) { } if (tsdbUpdateMFileHeader(&mf) < 0) { + tsdbError("vgId:%d failed to update META file header since %s, revert it", REPO_ID(pRepo), tstrerror(terrno)); + tsdbApplyMFileChange(&mf, pOMFile); + // TODO: need to reload metaCache return -1; } @@ -238,7 +250,7 @@ static int tsdbUpdateMetaRecord(STsdbFS *pfs, SMFile *pMFile, uint64_t uid, void tsdbUpdateMFileMagic(pMFile, POINTER_SHIFT(cont, contLen - sizeof(TSCKSUM))); SKVRecord *pRecord = taosHashGet(pfs->metaCache, (void *)&uid, sizeof(uid)); if (pRecord != NULL) { - pMFile->info.tombSize += pRecord->size; + pMFile->info.tombSize += (pRecord->size + sizeof(SKVRecord)); } else { pMFile->info.nRecords++; } @@ -253,7 +265,7 @@ static int tsdbDropMetaRecord(STsdbFS *pfs, SMFile *pMFile, uint64_t uid) { SKVRecord *pRecord = taosHashGet(pfs->metaCache, (void *)(&uid), sizeof(uid)); if (pRecord == NULL) { - tsdbError("failed to drop KV store record with key %" PRIu64 " since not find", uid); + tsdbError("failed to drop META record with key %" PRIu64 " since not find", uid); return -1; } @@ -264,11 +276,11 @@ static int tsdbDropMetaRecord(STsdbFS *pfs, SMFile *pMFile, uint64_t uid) { void *pBuf = buf; tsdbEncodeKVRecord(&pBuf, &rInfo); - if (tsdbAppendMFile(pMFile, buf, POINTER_DISTANCE(pBuf, buf), NULL) < 0) { + if (tsdbAppendMFile(pMFile, buf, sizeof(SKVRecord), NULL) < 0) { return -1; } - pMFile->info.magic = taosCalcChecksum(pMFile->info.magic, (uint8_t *)buf, (uint32_t)POINTER_DISTANCE(pBuf, buf)); + pMFile->info.magic = taosCalcChecksum(pMFile->info.magic, (uint8_t *)buf, sizeof(SKVRecord)); pMFile->info.nDels++; pMFile->info.nRecords--; pMFile->info.tombSize += (rInfo.size + sizeof(SKVRecord) * 2); @@ -300,7 +312,12 @@ static int tsdbCommitTSData(STsdbRepo *pRepo) { // Skip expired memory data and expired FSET tsdbSeekCommitIter(&commith, commith.rtn.minKey); while ((pSet = tsdbFSIterNext(&(commith.fsIter)))) { - if (pSet->fid >= commith.rtn.minFid) break; + if (pSet->fid >= commith.rtn.minFid) { + break; + } else { + tsdbInfo("vgId:%d FSET %d on level %d disk id %d expires, remove it", REPO_ID(pRepo), pSet->fid, + TSDB_FSET_LEVEL(pSet), TSDB_FSET_ID(pSet)); + } } // Loop to commit to each file @@ -347,7 +364,7 @@ static int tsdbCommitTSData(STsdbRepo *pRepo) { return 0; } -static int tsdbStartCommit(STsdbRepo *pRepo) { +static void tsdbStartCommit(STsdbRepo *pRepo) { SMemTable *pMem = pRepo->imem; ASSERT(pMem->numOfRows > 0 || listNEles(pMem->actList) > 0); @@ -358,7 +375,6 @@ static int tsdbStartCommit(STsdbRepo *pRepo) { tsdbStartFSTxn(pRepo, pMem->pointsAdd, pMem->storageAdd); pRepo->code = TSDB_CODE_SUCCESS; - return 0; } static void tsdbEndCommit(STsdbRepo *pRepo, int eno) { @@ -1371,12 +1387,16 @@ static int tsdbApplyRtn(STsdbRepo *pRepo) { STsdbFS * pfs = REPO_FS(pRepo); SDFileSet *pSet; - // Get retentioni snapshot + // Get retention snapshot tsdbGetRtnSnap(pRepo, &rtn); tsdbFSIterInit(&fsiter, pfs, TSDB_FS_ITER_FORWARD); while ((pSet = tsdbFSIterNext(&fsiter))) { - if (pSet->fid < rtn.minFid) continue; + if (pSet->fid < rtn.minFid) { + tsdbInfo("vgId:%d FSET %d at level %d disk id %d expires, remove it", REPO_ID(pRepo), pSet->fid, + TSDB_FSET_LEVEL(pSet), TSDB_FSET_ID(pSet)); + continue; + } if (tsdbApplyRtnOnFSet(pRepo, pSet, &rtn) < 0) { return -1; @@ -1390,10 +1410,13 @@ static int tsdbApplyRtnOnFSet(STsdbRepo *pRepo, SDFileSet *pSet, SRtn *pRtn) { SDiskID did; SDFileSet nSet; STsdbFS * pfs = REPO_FS(pRepo); + int level; ASSERT(pSet->fid >= pRtn->minFid); - tfsAllocDisk(tsdbGetFidLevel(pSet->fid, pRtn), &(did.level), &(did.id)); + level = tsdbGetFidLevel(pSet->fid, pRtn); + + tfsAllocDisk(level, &(did.level), &(did.id)); if (did.level == TFS_UNDECIDED_LEVEL) { terrno = TSDB_CODE_TDB_NO_AVAIL_DISK; return -1; @@ -1404,12 +1427,17 @@ static int tsdbApplyRtnOnFSet(STsdbRepo *pRepo, SDFileSet *pSet, SRtn *pRtn) { tsdbInitDFileSet(&nSet, did, REPO_ID(pRepo), pSet->fid, FS_TXN_VERSION(pfs)); if (tsdbCopyDFileSet(pSet, &nSet) < 0) { + tsdbError("vgId:%d failed to copy FSET %d from level %d to level %d since %s", REPO_ID(pRepo), pSet->fid, + TSDB_FSET_LEVEL(pSet), did.level, tstrerror(terrno)); return -1; } if (tsdbUpdateDFileSet(pfs, &nSet) < 0) { return -1; } + + tsdbInfo("vgId:%d FSET %d is copied from level %d disk id %d to level %d disk id %d", REPO_ID(pRepo), pSet->fid, + TSDB_FSET_LEVEL(pSet), TSDB_FSET_ID(pSet) did.level, did.id); } else { // On a correct level if (tsdbUpdateDFileSet(pfs, pSet) < 0) { diff --git a/src/tsdb/src/tsdbFS.c b/src/tsdb/src/tsdbFS.c index 9196fecfcd..5be44c79a0 100644 --- a/src/tsdb/src/tsdbFS.c +++ b/src/tsdb/src/tsdbFS.c @@ -699,6 +699,8 @@ int tsdbLoadMetaCache(STsdbRepo *pRepo, bool recoverMeta) { int64_t maxBufSize = 0; SMFInfo minfo; + taosHashEmpty(pfs->metaCache); + // No meta file, just return if (pfs->cstatus->pmf == NULL) return 0; @@ -716,6 +718,12 @@ int tsdbLoadMetaCache(STsdbRepo *pRepo, bool recoverMeta) { while (true) { int64_t tsize = tsdbReadMFile(pMFile, tbuf, sizeof(SKVRecord)); if (tsize == 0) break; + + if (tsize < 0) { + tsdbError("vgId:%d failed to read META file since %s", REPO_ID(pRepo), tstrerror(terrno)); + return -1; + } + if (tsize < sizeof(SKVRecord)) { tsdbError("vgId:%d failed to read %" PRIzu " bytes from file %s", REPO_ID(pRepo), sizeof(SKVRecord), TSDB_FILE_FULL_NAME(pMFile)); -- GitLab