/* * Copyright (c) 2019 TAOS Data, Inc. * * This program is free software: you can use, redistribute, and/or modify * it under the terms of the GNU Affero General Public License, version 3 * or later ("AGPL"), as published by the Free Software Foundation. * * This program is distributed in the hope that it will be useful, but WITHOUT * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or * FITNESS FOR A PARTICULAR PURPOSE. * * You should have received a copy of the GNU Affero General Public License * along with this program. If not, see . */ #include "tsdb.h" extern const char *TSDB_LEVEL_DNAME[]; typedef enum { TSDB_TXN_TEMP_FILE = 0, TSDB_TXN_CURR_FILE } TSDB_TXN_FILE_T; static const char *tsdbTxnFname[] = {"current.t", "current"}; #define TSDB_MAX_FSETS(keep, days) ((keep) / (days) + 3) static int tsdbComparFidFSet(const void *arg1, const void *arg2); static void tsdbResetFSStatus(SFSStatus *pStatus); static int tsdbSaveFSStatus(STsdb *pRepo, SFSStatus *pStatus); static void tsdbApplyFSTxnOnDisk(SFSStatus *pFrom, SFSStatus *pTo); static void tsdbGetTxnFname(STsdb *pRepo, TSDB_TXN_FILE_T ftype, char fname[]); static int tsdbOpenFSFromCurrent(STsdb *pRepo); static int tsdbScanAndTryFixFS(STsdb *pRepo); static int tsdbScanRootDir(STsdb *pRepo); static int tsdbScanDataDir(STsdb *pRepo); static bool tsdbIsTFileInFS(STsdbFS *pfs, const STfsFile *pf); static int tsdbRestoreCurrent(STsdb *pRepo); static int tsdbComparTFILE(const void *arg1, const void *arg2); static void tsdbScanAndTryFixDFilesHeader(STsdb *pRepo, int32_t *nExpired); // static int tsdbProcessExpiredFS(STsdb *pRepo); // static int tsdbCreateMeta(STsdb *pRepo); static void tsdbGetRootDir(int repoid, int8_t level, char dirName[]) { snprintf(dirName, TSDB_FILENAME_LEN, "vnode/vnode%d/%s", repoid, TSDB_LEVEL_DNAME[level]); } static void tsdbGetDataDir(int repoid, int8_t level, char dirName[]) { snprintf(dirName, TSDB_FILENAME_LEN, "vnode/vnode%d/%s/data", repoid, TSDB_LEVEL_DNAME[level]); } // For backward compatibility // ================== CURRENT file header info static int tsdbEncodeFSHeader(void **buf, SFSHeader *pHeader) { int tlen = 0; tlen += taosEncodeFixedU32(buf, pHeader->version); tlen += taosEncodeFixedU32(buf, pHeader->len); return tlen; } static void *tsdbDecodeFSHeader(void *buf, SFSHeader *pHeader) { buf = taosDecodeFixedU32(buf, &(pHeader->version)); buf = taosDecodeFixedU32(buf, &(pHeader->len)); return buf; } // ================== STsdbFSMeta static int tsdbEncodeFSMeta(void **buf, STsdbFSMeta *pMeta) { int tlen = 0; tlen += taosEncodeFixedU32(buf, pMeta->version); tlen += taosEncodeFixedI64(buf, pMeta->totalPoints); tlen += taosEncodeFixedI64(buf, pMeta->totalStorage); return tlen; } static void *tsdbDecodeFSMeta(void *buf, STsdbFSMeta *pMeta) { buf = taosDecodeFixedU32(buf, &(pMeta->version)); buf = taosDecodeFixedI64(buf, &(pMeta->totalPoints)); buf = taosDecodeFixedI64(buf, &(pMeta->totalStorage)); return buf; } // ================== SFSStatus static int tsdbEncodeDFileSetArray(void **buf, SArray *pArray) { int tlen = 0; uint64_t nset = taosArrayGetSize(pArray); tlen += taosEncodeFixedU64(buf, nset); for (size_t i = 0; i < nset; i++) { SDFileSet *pSet = taosArrayGet(pArray, i); tlen += tsdbEncodeDFileSet(buf, pSet); } return tlen; } static void *tsdbDecodeDFileSetArray(STsdb *pRepo, void *buf, SArray *pArray) { uint64_t nset = 0; taosArrayClear(pArray); buf = taosDecodeFixedU64(buf, &nset); for (size_t i = 0; i < nset; i++) { SDFileSet dset = {0}; buf = tsdbDecodeDFileSet(pRepo, buf, &dset); taosArrayPush(pArray, (void *)(&dset)); } return buf; } static int tsdbEncodeFSStatus(void **buf, SFSStatus *pStatus) { // ASSERT(pStatus->pmf); int tlen = 0; // tlen += tsdbEncodeSMFile(buf, pStatus->pmf); tlen += tsdbEncodeDFileSetArray(buf, pStatus->df); return tlen; } static void *tsdbDecodeFSStatus(STsdb *pRepo, void *buf, SFSStatus *pStatus) { tsdbResetFSStatus(pStatus); // pStatus->pmf = &(pStatus->mf); // buf = tsdbDecodeSMFile(buf, pStatus->pmf); buf = tsdbDecodeDFileSetArray(pRepo, buf, pStatus->df); return buf; } static SFSStatus *tsdbNewFSStatus(int maxFSet) { SFSStatus *pStatus = (SFSStatus *)taosMemoryCalloc(1, sizeof(*pStatus)); if (pStatus == NULL) { terrno = TSDB_CODE_TDB_OUT_OF_MEMORY; return NULL; } // TSDB_FILE_SET_CLOSED(&(pStatus->mf)); pStatus->df = taosArrayInit(maxFSet, sizeof(SDFileSet)); if (pStatus->df == NULL) { terrno = TSDB_CODE_TDB_OUT_OF_MEMORY; taosMemoryFree(pStatus); return NULL; } return pStatus; } static SFSStatus *tsdbFreeFSStatus(SFSStatus *pStatus) { if (pStatus) { pStatus->df = taosArrayDestroy(pStatus->df); taosMemoryFree(pStatus); } return NULL; } static void tsdbResetFSStatus(SFSStatus *pStatus) { if (pStatus == NULL) { return; } // TSDB_FILE_SET_CLOSED(&(pStatus->mf)); // pStatus->pmf = NULL; taosArrayClear(pStatus->df); } // static void tsdbSetStatusMFile(SFSStatus *pStatus, const SMFile *pMFile) { // ASSERT(pStatus->pmf == NULL); // pStatus->pmf = &(pStatus->mf); // tsdbInitMFileEx(pStatus->pmf, (SMFile *)pMFile); // } static int tsdbAddDFileSetToStatus(SFSStatus *pStatus, const SDFileSet *pSet) { if (taosArrayPush(pStatus->df, (void *)pSet) == NULL) { terrno = TSDB_CODE_TDB_OUT_OF_MEMORY; return -1; } TSDB_FSET_SET_CLOSED(((SDFileSet *)taosArrayGetLast(pStatus->df))); return 0; } // ================== STsdbFS STsdbFS *tsdbNewFS(const STsdbKeepCfg *pCfg) { int keep = pCfg->keep2; int days = pCfg->days; int maxFSet = TSDB_MAX_FSETS(keep, days); STsdbFS *pfs; pfs = (STsdbFS *)taosMemoryCalloc(1, sizeof(*pfs)); if (pfs == NULL) { terrno = TSDB_CODE_TDB_OUT_OF_MEMORY; return NULL; } int code = taosThreadRwlockInit(&(pfs->lock), NULL); if (code) { terrno = TAOS_SYSTEM_ERROR(code); taosMemoryFree(pfs); return NULL; } pfs->cstatus = tsdbNewFSStatus(maxFSet); if (pfs->cstatus == NULL) { tsdbFreeFS(pfs); return NULL; } pfs->metaCache = taosHashInit(4096, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_NO_LOCK); if (pfs->metaCache == NULL) { terrno = TSDB_CODE_TDB_OUT_OF_MEMORY; tsdbFreeFS(pfs); return NULL; } pfs->intxn = false; pfs->metaCacheComp = NULL; pfs->nstatus = tsdbNewFSStatus(maxFSet); if (pfs->nstatus == NULL) { tsdbFreeFS(pfs); return NULL; } return pfs; } void *tsdbFreeFS(STsdbFS *pfs) { if (pfs) { pfs->nstatus = tsdbFreeFSStatus(pfs->nstatus); taosHashCleanup(pfs->metaCache); pfs->metaCache = NULL; pfs->cstatus = tsdbFreeFSStatus(pfs->cstatus); taosThreadRwlockDestroy(&(pfs->lock)); taosMemoryFree(pfs); } return NULL; } int tsdbOpenFS(STsdb *pRepo) { STsdbFS *pfs = REPO_FS(pRepo); char current[TSDB_FILENAME_LEN] = "\0"; int nExpired = 0; ASSERT(pfs != NULL); tsdbGetTxnFname(pRepo, TSDB_TXN_CURR_FILE, current); tsdbGetRtnSnap(pRepo, &pRepo->rtn); if (taosCheckExistFile(current)) { if (tsdbOpenFSFromCurrent(pRepo) < 0) { tsdbError("vgId:%d failed to open FS since %s", REPO_ID(pRepo), tstrerror(terrno)); return -1; } tsdbScanAndTryFixDFilesHeader(pRepo, &nExpired); // if (nExpired > 0) { // tsdbProcessExpiredFS(pRepo); // } } else { // should skip expired fileset inside of the function if (tsdbRestoreCurrent(pRepo) < 0) { tsdbError("vgId:%d failed to restore current file since %s", REPO_ID(pRepo), tstrerror(terrno)); return -1; } } if (tsdbScanAndTryFixFS(pRepo) < 0) { tsdbError("vgId:%d failed to scan and fix FS since %s", REPO_ID(pRepo), tstrerror(terrno)); return -1; } // // Load meta cache if has meta file // if ((!(pRepo->state & TSDB_STATE_BAD_META)) && tsdbLoadMetaCache(pRepo, true) < 0) { // tsdbError("vgId:%d failed to open FS while loading meta cache since %s", REPO_ID(pRepo), tstrerror(terrno)); // return -1; // } return 0; } void tsdbCloseFS(STsdb *pRepo) { // Do nothing } // Start a new transaction to modify the file system void tsdbStartFSTxn(STsdb *pRepo, int64_t pointsAdd, int64_t storageAdd) { STsdbFS *pfs = REPO_FS(pRepo); ASSERT(pfs->intxn == false); pfs->intxn = true; tsdbResetFSStatus(pfs->nstatus); pfs->nstatus->meta = pfs->cstatus->meta; // if (pfs->cstatus->pmf == NULL) { pfs->nstatus->meta.version += 1; // } else { // pfs->nstatus->meta.version = pfs->cstatus->meta.version + 1; // } pfs->nstatus->meta.totalPoints = pfs->cstatus->meta.totalPoints + pointsAdd; pfs->nstatus->meta.totalStorage = pfs->cstatus->meta.totalStorage += storageAdd; } void tsdbUpdateFSTxnMeta(STsdbFS *pfs, STsdbFSMeta *pMeta) { pfs->nstatus->meta = *pMeta; } int tsdbEndFSTxn(STsdb *pRepo) { STsdbFS *pfs = REPO_FS(pRepo); ASSERT(FS_IN_TXN(pfs)); SFSStatus *pStatus; // Write current file system snapshot if (tsdbSaveFSStatus(pRepo, pfs->nstatus) < 0) { tsdbEndFSTxnWithError(pfs); return -1; } // Make new tsdbWLockFS(pfs); pStatus = pfs->cstatus; pfs->cstatus = pfs->nstatus; pfs->nstatus = pStatus; tsdbUnLockFS(pfs); // Apply actual change to each file and SDFileSet tsdbApplyFSTxnOnDisk(pfs->nstatus, pfs->cstatus); pfs->intxn = false; return 0; } int tsdbEndFSTxnWithError(STsdbFS *pfs) { tsdbApplyFSTxnOnDisk(pfs->nstatus, pfs->cstatus); // TODO: if mf change, reload pfs->metaCache pfs->intxn = false; return 0; } // void tsdbUpdateMFile(STsdbFS *pfs, const SMFile *pMFile) { tsdbSetStatusMFile(pfs->nstatus, pMFile); } int tsdbUpdateDFileSet(STsdbFS *pfs, const SDFileSet *pSet) { return tsdbAddDFileSetToStatus(pfs->nstatus, pSet); } static int tsdbSaveFSStatus(STsdb *pRepo, SFSStatus *pStatus) { SFSHeader fsheader; void *pBuf = NULL; void *ptr; char hbuf[TSDB_FILE_HEAD_SIZE] = "\0"; char tfname[TSDB_FILENAME_LEN] = "\0"; char cfname[TSDB_FILENAME_LEN] = "\0"; tsdbGetTxnFname(pRepo, TSDB_TXN_TEMP_FILE, tfname); tsdbGetTxnFname(pRepo, TSDB_TXN_CURR_FILE, cfname); TdFilePtr pFile = taosOpenFile(tfname, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC); if (pFile == NULL) { terrno = TAOS_SYSTEM_ERROR(errno); return -1; } fsheader.version = TSDB_LATEST_SFS_VER; if (taosArrayGetSize(pStatus->df) == 0) { fsheader.len = 0; } else { fsheader.len = tsdbEncodeFSStatus(NULL, pStatus) + sizeof(TSCKSUM); } // Encode header part and write ptr = hbuf; tsdbEncodeFSHeader(&ptr, &fsheader); tsdbEncodeFSMeta(&ptr, &(pStatus->meta)); taosCalcChecksumAppend(0, (uint8_t *)hbuf, TSDB_FILE_HEAD_SIZE); if (taosWriteFile(pFile, hbuf, TSDB_FILE_HEAD_SIZE) < TSDB_FILE_HEAD_SIZE) { terrno = TAOS_SYSTEM_ERROR(errno); taosCloseFile(&pFile); taosRemoveFile(tfname); return -1; } // Encode file status and write to file if (fsheader.len > 0) { if (tsdbMakeRoom(&(pBuf), fsheader.len) < 0) { taosCloseFile(&pFile); taosRemoveFile(tfname); return -1; } ptr = pBuf; tsdbEncodeFSStatus(&ptr, pStatus); taosCalcChecksumAppend(0, (uint8_t *)pBuf, fsheader.len); if (taosWriteFile(pFile, pBuf, fsheader.len) < fsheader.len) { terrno = TAOS_SYSTEM_ERROR(errno); taosCloseFile(&pFile); (void)taosRemoveFile(tfname); taosTZfree(pBuf); return -1; } } // fsync, close and rename if (taosFsyncFile(pFile) < 0) { terrno = TAOS_SYSTEM_ERROR(errno); taosCloseFile(&pFile); taosRemoveFile(tfname); taosTZfree(pBuf); return -1; } (void)taosCloseFile(&pFile); (void)taosRenameFile(tfname, cfname); taosTZfree(pBuf); return 0; } static void tsdbApplyFSTxnOnDisk(SFSStatus *pFrom, SFSStatus *pTo) { int ifrom = 0; int ito = 0; size_t sizeFrom, sizeTo; SDFileSet *pSetFrom; SDFileSet *pSetTo; sizeFrom = taosArrayGetSize(pFrom->df); sizeTo = taosArrayGetSize(pTo->df); // Apply meta file change // (void)tsdbApplyMFileChange(pFrom->pmf, pTo->pmf); // Apply SDFileSet change if (ifrom >= sizeFrom) { pSetFrom = NULL; } else { pSetFrom = taosArrayGet(pFrom->df, ifrom); } if (ito >= sizeTo) { pSetTo = NULL; } else { pSetTo = taosArrayGet(pTo->df, ito); } while (true) { if ((pSetTo == NULL) && (pSetFrom == NULL)) break; if (pSetTo == NULL || (pSetFrom && pSetFrom->fid < pSetTo->fid)) { tsdbApplyDFileSetChange(pSetFrom, NULL); ifrom++; if (ifrom >= sizeFrom) { pSetFrom = NULL; } else { pSetFrom = taosArrayGet(pFrom->df, ifrom); } } else if (pSetFrom == NULL || pSetFrom->fid > pSetTo->fid) { // Do nothing ito++; if (ito >= sizeTo) { pSetTo = NULL; } else { pSetTo = taosArrayGet(pTo->df, ito); } } else { tsdbApplyDFileSetChange(pSetFrom, pSetTo); ifrom++; if (ifrom >= sizeFrom) { pSetFrom = NULL; } else { pSetFrom = taosArrayGet(pFrom->df, ifrom); } ito++; if (ito >= sizeTo) { pSetTo = NULL; } else { pSetTo = taosArrayGet(pTo->df, ito); } } } } // ================== SFSIter // ASSUMPTIONS: the FS Should be read locked when calling these functions void tsdbFSIterInit(SFSIter *pIter, STsdbFS *pfs, int direction) { pIter->pfs = pfs; pIter->direction = direction; size_t size = taosArrayGetSize(pfs->cstatus->df); pIter->version = pfs->cstatus->meta.version; if (size == 0) { pIter->index = -1; pIter->fid = TSDB_IVLD_FID; } else { if (direction == TSDB_FS_ITER_FORWARD) { pIter->index = 0; } else { pIter->index = (int)(size - 1); } pIter->fid = ((SDFileSet *)taosArrayGet(pfs->cstatus->df, pIter->index))->fid; } } void tsdbFSIterSeek(SFSIter *pIter, int fid) { STsdbFS *pfs = pIter->pfs; size_t size = taosArrayGetSize(pfs->cstatus->df); int flags; if (pIter->direction == TSDB_FS_ITER_FORWARD) { flags = TD_GE; } else { flags = TD_LE; } void *ptr = taosbsearch(&fid, pfs->cstatus->df->pData, size, sizeof(SDFileSet), tsdbComparFidFSet, flags); if (ptr == NULL) { pIter->index = -1; pIter->fid = TSDB_IVLD_FID; } else { pIter->index = (int)(TARRAY_ELEM_IDX(pfs->cstatus->df, ptr)); pIter->fid = ((SDFileSet *)ptr)->fid; } } SDFileSet *tsdbFSIterNext(SFSIter *pIter) { STsdbFS *pfs = pIter->pfs; SDFileSet *pSet; if (pIter->index < 0) { ASSERT(pIter->fid == TSDB_IVLD_FID); return NULL; } ASSERT(pIter->fid != TSDB_IVLD_FID); if (pIter->version != pfs->cstatus->meta.version) { pIter->version = pfs->cstatus->meta.version; tsdbFSIterSeek(pIter, pIter->fid); } if (pIter->index < 0) { return NULL; } pSet = (SDFileSet *)taosArrayGet(pfs->cstatus->df, pIter->index); ASSERT(pSet->fid == pIter->fid); if (pIter->direction == TSDB_FS_ITER_FORWARD) { pIter->index++; if (pIter->index >= taosArrayGetSize(pfs->cstatus->df)) { pIter->index = -1; } } else { pIter->index--; } if (pIter->index >= 0) { pIter->fid = ((SDFileSet *)taosArrayGet(pfs->cstatus->df, pIter->index))->fid; } else { pIter->fid = TSDB_IVLD_FID; } return pSet; } static int tsdbComparFidFSet(const void *arg1, const void *arg2) { int fid = *(int *)arg1; SDFileSet *pSet = (SDFileSet *)arg2; if (fid < pSet->fid) { return -1; } else if (fid == pSet->fid) { return 0; } else { return 1; } } static void tsdbGetTxnFname(STsdb *pRepo, TSDB_TXN_FILE_T ftype, char fname[]) { snprintf(fname, TSDB_FILENAME_LEN, "%s/vnode/vnode%d/%s/%s", tfsGetPrimaryPath(REPO_TFS(pRepo)), REPO_ID(pRepo), TSDB_LEVEL_DNAME[REPO_LEVEL(pRepo)], tsdbTxnFname[ftype]); } static int tsdbOpenFSFromCurrent(STsdb *pRepo) { STsdbFS *pfs = REPO_FS(pRepo); TdFilePtr pFile = NULL; void *buffer = NULL; SFSHeader fsheader; char current[TSDB_FILENAME_LEN] = "\0"; void *ptr; tsdbGetTxnFname(pRepo, TSDB_TXN_CURR_FILE, current); // current file exists, try to recover pFile = taosOpenFile(current, TD_FILE_READ); if (pFile == NULL) { tsdbError("vgId:%d failed to open file %s since %s", REPO_ID(pRepo), current, strerror(errno)); terrno = TAOS_SYSTEM_ERROR(errno); goto _err; } if (tsdbMakeRoom(&buffer, TSDB_FILE_HEAD_SIZE) < 0) { goto _err; } int nread = (int)taosReadFile(pFile, buffer, TSDB_FILE_HEAD_SIZE); if (nread < 0) { tsdbError("vgId:%d failed to read %d bytes from file %s since %s", REPO_ID(pRepo), TSDB_FILENAME_LEN, current, strerror(errno)); terrno = TAOS_SYSTEM_ERROR(errno); goto _err; } if (nread < TSDB_FILE_HEAD_SIZE) { tsdbError("vgId:%d failed to read header of file %s, read bytes:%d", REPO_ID(pRepo), current, nread); terrno = TSDB_CODE_TDB_FILE_CORRUPTED; goto _err; } if (!taosCheckChecksumWhole((uint8_t *)buffer, TSDB_FILE_HEAD_SIZE)) { tsdbError("vgId:%d header of file %s failed checksum check", REPO_ID(pRepo), current); terrno = TSDB_CODE_TDB_FILE_CORRUPTED; goto _err; } SFSStatus *pStatus = pfs->cstatus; ptr = buffer; ptr = tsdbDecodeFSHeader(ptr, &fsheader); ptr = tsdbDecodeFSMeta(ptr, &(pStatus->meta)); if (fsheader.version != TSDB_LATEST_SFS_VER) { // TODO: handle file version change } if (fsheader.len > 0) { if (tsdbMakeRoom(&buffer, fsheader.len) < 0) { goto _err; } nread = (int)taosReadFile(pFile, buffer, fsheader.len); if (nread < 0) { tsdbError("vgId:%d failed to read file %s since %s", REPO_ID(pRepo), current, strerror(errno)); terrno = TAOS_SYSTEM_ERROR(errno); goto _err; } if (nread < fsheader.len) { tsdbError("vgId:%d failed to read %d bytes from file %s", REPO_ID(pRepo), fsheader.len, current); terrno = TSDB_CODE_TDB_FILE_CORRUPTED; goto _err; } if (!taosCheckChecksumWhole((uint8_t *)buffer, fsheader.len)) { tsdbError("vgId:%d file %s is corrupted since wrong checksum", REPO_ID(pRepo), current); terrno = TSDB_CODE_TDB_FILE_CORRUPTED; goto _err; } ptr = buffer; ptr = tsdbDecodeFSStatus(pRepo, ptr, pStatus); } else { tsdbResetFSStatus(pStatus); } taosTZfree(buffer); taosCloseFile(&pFile); return 0; _err: if (pFile != NULL) { taosCloseFile(&pFile); } taosTZfree(buffer); return -1; } // Scan and try to fix incorrect files static int tsdbScanAndTryFixFS(STsdb *pRepo) { STsdbFS *pfs = REPO_FS(pRepo); SFSStatus *pStatus = pfs->cstatus; // if (tsdbScanAndTryFixMFile(pRepo) < 0) { // tsdbError("vgId:%d failed to fix MFile since %s", REPO_ID(pRepo), tstrerror(terrno)); // return -1; // } size_t size = taosArrayGetSize(pStatus->df); for (size_t i = 0; i < size; i++) { SDFileSet *pSet = (SDFileSet *)taosArrayGet(pStatus->df, i); if (tsdbScanAndTryFixDFileSet(pRepo, pSet) < 0) { tsdbError("vgId:%d failed to fix MFile since %s", REPO_ID(pRepo), tstrerror(terrno)); return -1; } } // remove those unused files tsdbScanRootDir(pRepo); tsdbScanDataDir(pRepo); return 0; } static int tsdbScanRootDir(STsdb *pRepo) { char rootDir[TSDB_FILENAME_LEN]; char bname[TSDB_FILENAME_LEN]; STsdbFS *pfs = REPO_FS(pRepo); const STfsFile *pf; tsdbGetRootDir(REPO_ID(pRepo), REPO_LEVEL(pRepo), rootDir); STfsDir *tdir = tfsOpendir(REPO_TFS(pRepo), rootDir); if (tdir == NULL) { tsdbError("vgId:%d failed to open directory %s since %s", REPO_ID(pRepo), rootDir, tstrerror(terrno)); return -1; } while ((pf = tfsReaddir(tdir))) { tfsBasename(pf, bname); if (strcmp(bname, tsdbTxnFname[TSDB_TXN_CURR_FILE]) == 0 || strcmp(bname, "data") == 0) { // Skip current file and data directory continue; } // if (/*pfs->cstatus->pmf && */ tfsIsSameFile(pf, &(pfs->cstatus->pmf->f))) { // continue; // } (void)tfsRemoveFile(pf); tsdbDebug("vgId:%d invalid file %s is removed", REPO_ID(pRepo), pf->aname); } tfsClosedir(tdir); return 0; } static int tsdbScanDataDir(STsdb *pRepo) { char dataDir[TSDB_FILENAME_LEN]; char bname[TSDB_FILENAME_LEN]; STsdbFS *pfs = REPO_FS(pRepo); const STfsFile *pf; tsdbGetDataDir(REPO_ID(pRepo), REPO_LEVEL(pRepo), dataDir); STfsDir *tdir = tfsOpendir(REPO_TFS(pRepo), dataDir); if (tdir == NULL) { tsdbError("vgId:%d failed to open directory %s since %s", REPO_ID(pRepo), dataDir, tstrerror(terrno)); return -1; } while ((pf = tfsReaddir(tdir))) { tfsBasename(pf, bname); if (!tsdbIsTFileInFS(pfs, pf)) { (void)tfsRemoveFile(pf); tsdbDebug("vgId:%d invalid file %s is removed", REPO_ID(pRepo), pf->aname); } } tfsClosedir(tdir); return 0; } static bool tsdbIsTFileInFS(STsdbFS *pfs, const STfsFile *pf) { SFSIter fsiter; tsdbFSIterInit(&fsiter, pfs, TSDB_FS_ITER_FORWARD); SDFileSet *pSet; while ((pSet = tsdbFSIterNext(&fsiter))) { for (TSDB_FILE_T ftype = 0; ftype < TSDB_FILE_MAX; ftype++) { SDFile *pDFile = TSDB_DFILE_IN_SET(pSet, ftype); if (tfsIsSameFile(pf, TSDB_FILE_F(pDFile))) { return true; } } } return false; } static int tsdbRestoreDFileSet(STsdb *pRepo) { char dataDir[TSDB_FILENAME_LEN]; char bname[TSDB_FILENAME_LEN]; STfsDir *tdir = NULL; const STfsFile *pf = NULL; const char *pattern = "^v[0-9]+f[0-9]+\\.(head|data|last|smad|smal)(-ver[0-9]+)?$"; SArray *fArray = NULL; regex_t regex; STsdbFS *pfs = REPO_FS(pRepo); tsdbGetDataDir(REPO_ID(pRepo), REPO_LEVEL(pRepo), dataDir); // Resource allocation and init regcomp(®ex, pattern, REG_EXTENDED); fArray = taosArrayInit(1024, sizeof(STfsFile)); if (fArray == NULL) { terrno = TSDB_CODE_TDB_OUT_OF_MEMORY; tsdbError("vgId:%d failed to restore DFileSet while open directory %s since %s", REPO_ID(pRepo), dataDir, tstrerror(terrno)); regfree(®ex); return -1; } tdir = tfsOpendir(REPO_TFS(pRepo), dataDir); if (tdir == NULL) { tsdbError("vgId:%d failed to restore DFileSet while open directory %s since %s", REPO_ID(pRepo), dataDir, tstrerror(terrno)); taosArrayDestroy(fArray); regfree(®ex); return -1; } while ((pf = tfsReaddir(tdir))) { tfsBasename(pf, bname); int code = regexec(®ex, bname, 0, NULL, 0); if (code == 0) { if (taosArrayPush(fArray, (void *)pf) == NULL) { terrno = TSDB_CODE_TDB_OUT_OF_MEMORY; tfsClosedir(tdir); taosArrayDestroy(fArray); regfree(®ex); return -1; } } else if (code == REG_NOMATCH) { // Not match tsdbInfo("vgId:%d invalid file %s exists, remove it", REPO_ID(pRepo), pf->aname); (void)tfsRemoveFile(pf); continue; } else { // Has other error tsdbError("vgId:%d failed to restore DFileSet Array while run regexec since %s", REPO_ID(pRepo), strerror(code)); terrno = TAOS_SYSTEM_ERROR(code); tfsClosedir(tdir); taosArrayDestroy(fArray); regfree(®ex); return -1; } } tfsClosedir(tdir); regfree(®ex); // Sort the array according to file name taosArraySort(fArray, tsdbComparTFILE); size_t index = 0; // Loop to recover each file set for (;;) { if (index >= taosArrayGetSize(fArray)) { break; } SDFileSet fset = {0}; TSDB_FSET_SET_CLOSED(&fset); // Loop to recover ONE fset for (TSDB_FILE_T ftype = 0; ftype < TSDB_FILE_MAX; ftype++) { SDFile *pDFile = TSDB_DFILE_IN_SET(&fset, ftype); if (index >= taosArrayGetSize(fArray)) { tsdbError("vgId:%d incomplete DFileSet, fid:%d", REPO_ID(pRepo), fset.fid); taosArrayDestroy(fArray); return -1; } pf = taosArrayGet(fArray, index); int tvid, tfid; TSDB_FILE_T ttype; uint32_t tversion; char _bname[TSDB_FILENAME_LEN]; tfsBasename(pf, _bname); tsdbParseDFilename(_bname, &tvid, &tfid, &ttype, &tversion); ASSERT(tvid == REPO_ID(pRepo)); if (tfid < pRepo->rtn.minFid) { // skip file expired ++index; continue; } if (ftype == 0) { fset.fid = tfid; } else { if (tfid != fset.fid) { tsdbError("vgId:%d incomplete dFileSet, fid:%d", REPO_ID(pRepo), fset.fid); taosArrayDestroy(fArray); return -1; } } if (ttype != ftype) { tsdbError("vgId:%d incomplete dFileSet, fid:%d", REPO_ID(pRepo), fset.fid); taosArrayDestroy(fArray); return -1; } pDFile->f = *pf; // if (tsdbOpenDFile(pDFile, O_RDONLY) < 0) { if (tsdbOpenDFile(pDFile, TD_FILE_READ) < 0) { tsdbError("vgId:%d failed to open DFile %s since %s", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pDFile), tstrerror(terrno)); taosArrayDestroy(fArray); return -1; } if (tsdbLoadDFileHeader(pDFile, &(pDFile->info)) < 0) { tsdbError("vgId:%d failed to load DFile %s header since %s", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pDFile), tstrerror(terrno)); taosArrayDestroy(fArray); return -1; } if (tsdbForceKeepFile) { int64_t file_size; // Get real file size if (taosFStatFile(pDFile->pFile, &file_size, NULL) < 0) { terrno = TAOS_SYSTEM_ERROR(errno); taosArrayDestroy(fArray); return -1; } if (pDFile->info.size != file_size) { int64_t tfsize = pDFile->info.size; pDFile->info.size = file_size; tsdbInfo("vgId:%d file %s header size is changed from %" PRId64 " to %" PRId64, REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pDFile), tfsize, pDFile->info.size); } } tsdbCloseDFile(pDFile); index++; } tsdbInfo("vgId:%d FSET %d is restored", REPO_ID(pRepo), fset.fid); taosArrayPush(pfs->cstatus->df, &fset); } // Resource release taosArrayDestroy(fArray); return 0; } static int tsdbRestoreCurrent(STsdb *pRepo) { // // Loop to recover mfile // if (tsdbRestoreMeta(pRepo) < 0) { // tsdbError("vgId:%d failed to restore current since %s", REPO_ID(pRepo), tstrerror(terrno)); // return -1; // } // Loop to recover dfile set if (tsdbRestoreDFileSet(pRepo) < 0) { tsdbError("vgId:%d failed to restore DFileSet since %s", REPO_ID(pRepo), tstrerror(terrno)); return -1; } if (tsdbSaveFSStatus(pRepo, pRepo->fs->cstatus) < 0) { tsdbError("vgId:%d failed to restore current since %s", REPO_ID(pRepo), tstrerror(terrno)); return -1; } return 0; } static int tsdbComparTFILE(const void *arg1, const void *arg2) { STfsFile *pf1 = (STfsFile *)arg1; STfsFile *pf2 = (STfsFile *)arg2; int vid1, fid1, vid2, fid2; TSDB_FILE_T ftype1, ftype2; uint32_t version1, version2; char bname1[TSDB_FILENAME_LEN]; char bname2[TSDB_FILENAME_LEN]; tfsBasename(pf1, bname1); tfsBasename(pf2, bname2); tsdbParseDFilename(bname1, &vid1, &fid1, &ftype1, &version1); tsdbParseDFilename(bname2, &vid2, &fid2, &ftype2, &version2); if (fid1 < fid2) { return -1; } else if (fid1 > fid2) { return 1; } else { if (ftype1 < ftype2) { return -1; } else if (ftype1 > ftype2) { return 1; } else { return 0; } } } static void tsdbScanAndTryFixDFilesHeader(STsdb *pRepo, int32_t *nExpired) { STsdbFS *pfs = REPO_FS(pRepo); SFSStatus *pStatus = pfs->cstatus; SDFInfo info; for (size_t i = 0; i < taosArrayGetSize(pStatus->df); i++) { SDFileSet fset; tsdbInitDFileSetEx(&fset, (SDFileSet *)taosArrayGet(pStatus->df, i)); if (fset.fid < pRepo->rtn.minFid) { ++*nExpired; } tsdbDebug("vgId:%d scan DFileSet %d header", REPO_ID(pRepo), fset.fid); // if (tsdbOpenDFileSet(&fset, O_RDWR) < 0) { if (tsdbOpenDFileSet(&fset, TD_FILE_WRITE | TD_FILE_READ) < 0) { tsdbError("vgId:%d failed to open DFileSet %d since %s, continue", REPO_ID(pRepo), fset.fid, tstrerror(terrno)); continue; } for (TSDB_FILE_T ftype = 0; ftype < TSDB_FILE_MAX; ftype++) { SDFile *pDFile = TSDB_DFILE_IN_SET(&fset, ftype); if ((tsdbLoadDFileHeader(pDFile, &info) < 0) || pDFile->info.size != info.size || pDFile->info.magic != info.magic) { if (tsdbUpdateDFileHeader(pDFile) < 0) { tsdbError("vgId:%d failed to update DFile header of %s since %s, continue", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pDFile), tstrerror(terrno)); } else { tsdbInfo("vgId:%d DFile header of %s is updated", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pDFile)); TSDB_FILE_FSYNC(pDFile); } } else { tsdbDebug("vgId:%d DFile header of %s is correct", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pDFile)); } } tsdbCloseDFileSet(&fset); } }