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

refact tfs module

This commit is contained in:
Shengliang Guan 2022-01-18 21:39:32 -08:00
parent d6359f3ab1
commit ed9709c3f0
20 changed files with 581 additions and 447 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

255
include/libs/tfs/tfs.h
View File

@ -16,77 +16,226 @@
#ifndef _TD_TFS_H_ #ifndef _TD_TFS_H_
#define _TD_TFS_H_ #define _TD_TFS_H_
#include "tglobal.h" #include "tcfg.h"
#ifdef __cplusplus #ifdef __cplusplus
extern "C" { extern "C" {
#endif #endif
/* ------------------------ TYPES EXPOSED ------------------------ */
typedef struct STfs STfs;
typedef struct STfsDir STfsDir;
typedef struct { typedef struct {
int32_t level; int32_t level;
int32_t id; int32_t id;
} SDiskID; } SDiskID;
#define TFS_UNDECIDED_LEVEL -1
#define TFS_UNDECIDED_ID -1
#define TFS_PRIMARY_LEVEL 0
#define TFS_PRIMARY_ID 0
#define TFS_MIN_LEVEL 0
#define TFS_MAX_LEVEL (TSDB_MAX_TIERS - 1)
// FS APIs ====================================
typedef struct { typedef struct {
int64_t total; SDiskID did;
int64_t used;
int64_t avail;
} SFSMeta;
int32_t tfsInit(SDiskCfg *pDiskCfg, int32_t ndisk);
void tfsCleanup();
void tfsUpdateSize(SFSMeta *pFSMeta);
void tfsAllocDisk(int32_t expLevel, int32_t *level, int32_t *id);
const char *TFS_PRIMARY_PATH();
const char *TFS_DISK_PATH(int32_t level, int32_t id);
// TFILE APIs ====================================
typedef struct {
int32_t level;
int32_t id;
char rname[TSDB_FILENAME_LEN]; // REL name
char aname[TSDB_FILENAME_LEN]; // ABS name char aname[TSDB_FILENAME_LEN]; // ABS name
} TFILE; char rname[TSDB_FILENAME_LEN]; // REL name
STfs *pTfs;
} STfsFile;
#define TFILE_LEVEL(pf) ((pf)->level) /**
#define TFILE_ID(pf) ((pf)->id) * @brief Open a fs.
#define TFILE_NAME(pf) ((pf)->aname) *
#define TFILE_REL_NAME(pf) ((pf)->rname) * @param pCfg Config of the fs.
* @param ndisk Length of the config.
* @return STfs* The fs object.
*/
STfs *tfsOpen(SDiskCfg *pCfg, int32_t ndisk);
#define tfsopen(pf, flags) open(TFILE_NAME(pf), flags) /**
#define tfsclose(fd) close(fd) * @brief Close a fs.
#define tfsremove(pf) remove(TFILE_NAME(pf)) *
#define tfscopy(sf, df) taosCopyFile(TFILE_NAME(sf), TFILE_NAME(df)) * @param pTfs The fs object to close.
#define tfsrename(sf, df) taosRename(TFILE_NAME(sf), TFILE_NAME(df)) */
void tfsClose(STfs *pTfs);
void tfsInitFile(TFILE *pf, int32_t level, int32_t id, const char *bname); /**
bool tfsIsSameFile(const TFILE *pf1, const TFILE *pf2); * @brief Update the disk size.
int32_t tfsEncodeFile(void **buf, TFILE *pf); *
void *tfsDecodeFile(void *buf, TFILE *pf); * @param pTfs The fs object.
void tfsbasename(const TFILE *pf, char *dest); */
void tfsdirname(const TFILE *pf, char *dest); void tfsUpdateSize(STfs *pTfs);
// DIR APIs ==================================== /**
int32_t tfsMkdirAt(const char *rname, int32_t level, int32_t id); * @brief Get the disk size.
int32_t tfsMkdirRecurAt(const char *rname, int32_t level, int32_t id); *
int32_t tfsMkdir(const char *rname); * @param pTfs The fs object.
int32_t tfsRmdir(const char *rname); */
int32_t tfsRename(char *orname, char *nrname); SDiskSize tfsGetSize(STfs *pTfs);
typedef struct TDIR TDIR; /**
* @brief Allocate an existing available tier level from fs.
*
* @param pTfs The fs object.
* @param expLevel Disk level want to allocate.
* @param pDiskId The disk ID after allocation.
* @return int32_t 0 for success, -1 for failure.
*/
int32_t tfsAllocDisk(STfs *pTfs, int32_t expLevel, SDiskID *pDiskId);
TDIR *tfsOpendir(const char *rname); /**
const TFILE *tfsReaddir(TDIR *tdir); * @brief Get the primary path.
void tfsClosedir(TDIR *tdir); *
* @param pTfs The fs object.
* @return const char * The primary path.
*/
const char *tfsGetPrimaryPath(STfs *pTfs);
/**
* @brief Get the disk path.
*
* @param pTfs The fs object.
* @param diskId The diskId.
* @return const char * The primary path.
*/
const char *tfsGetDiskPath(STfs *pTfs, SDiskID diskId);
/**
* @brief Make directory at all levels in tfs.
*
* @param pTfs The fs object.
* @param rname The rel name of directory.
* @return int32_t 0 for success, -1 for failure.
*/
int32_t tfsMkdir(STfs *pTfs, const char *rname);
/**
* @brief Create directories in tfs.
*
* @param pTfs The fs object.
* @param rname The rel name of directory.
* @param diskId The disk ID.
* @return int32_t 0 for success, -1 for failure.
*/
int32_t tfsMkdirAt(STfs *pTfs, const char *rname, SDiskID diskId);
/**
* @brief Recursive create directories in tfs.
*
* @param pTfs The fs object.
* @param rname The rel name of directory.
* @param diskId The disk ID.
* @return int32_t 0 for success, -1 for failure.
*/
int32_t tfsMkdirRecurAt(STfs *pTfs, const char *rname, SDiskID diskId);
/**
* @brief Remove directory at all levels in tfs.
*
* @param pTfs The fs object.
* @param rname The rel name of directory.
* @return int32_t 0 for success, -1 for failure.
*/
int32_t tfsRmdir(STfs *pTfs, const char *rname);
/**
* @brief Rename file/directory in tfs.
*
* @param pTfs The fs object.
* @param orname The rel name of old file.
* @param nrname The rel name of new file.
* @return int32_t 0 for success, -1 for failure.
*/
int32_t tfsRename(STfs *pTfs, char *orname, char *nrname);
/**
* @brief Init file object in tfs.
*
* @param pTfs The fs object.
* @param pFile The file object.
* @param diskId The disk ID.
* @param rname The rel name of file.
*/
void tfsInitFile(STfs *pTfs, STfsFile *pFile, SDiskID diskId, const char *rname);
/**
* @brief Determine whether they are the same file.
*
* @param pFile1 The file object.
* @param pFile2 The file object.
* @param bool The compare result.
*/
bool tfsIsSameFile(const STfsFile *pFile1, const STfsFile *pFile2);
/**
* @brief Encode file name to a buffer.
*
* @param buf The buffer where file name are saved.
* @param pFile The file object.
* @return int32_t 0 for success, -1 for failure.
*/
int32_t tfsEncodeFile(void **buf, STfsFile *pFile);
/**
* @brief Decode file name from a buffer.
*
* @param pTfs The fs object.
* @param buf The buffer where file name are saved.
* @param pFile The file object.
* @return void * Buffer address after decode.
*/
void *tfsDecodeFile(STfs *pTfs, void *buf, STfsFile *pFile);
/**
* @brief Get the basename of the file.
*
* @param pFile The file object.
* @param dest The buffer where basename will be saved.
*/
void tfsBasename(const STfsFile *pFile, char *dest);
/**
* @brief Get the dirname of the file.
*
* @param pFile The file object.
* @param dest The buffer where dirname will be saved.
*/
void tfsDirname(const STfsFile *pFile, char *dest);
/**
* @brief Remove file in tfs.
*
* @param pFile The file to be removed.
* @return int32_t 0 for success, -1 for failure.
*/
int32_t tfsRemoveFile(const STfsFile *pFile);
/**
* @brief Copy file in tfs.
*
* @param pFile1 The src file.
* @param pFile2 The dest file.
* @return int32_t 0 for success, -1 for failure.
*/
int32_t tfsCopyFile(const STfsFile *pFile1, const STfsFile *pFile2);
/**
* @brief Open a directory for traversal.
*
* @param rname The rel name of file.
* @return STfsDir* The dir object.
*/
STfsDir *tfsOpendir(STfs *pTfs, const char *rname);
/**
* @brief Get a file from dir and move to next pos.
*
* @param pDir The dir object.
* @return STfsFile* The file in dir.
*/
const STfsFile *tfsReaddir(STfsDir *pDir);
/**
* @brief Close a directory.
*
* @param pDir The dir object.
*/
void tfsClosedir(STfsDir *pDir);
#ifdef __cplusplus #ifdef __cplusplus
} }

12
include/util/tdef.h
View File

@ -372,10 +372,14 @@ do { \
#define TSDB_ARB_DUMMY_TIME 4765104000000 // 2121-01-01 00:00:00.000, :P #define TSDB_ARB_DUMMY_TIME 4765104000000 // 2121-01-01 00:00:00.000, :P
#define TSDB_MAX_TIERS 3 #define TFS_MAX_TIERS 3
#define TSDB_MAX_DISKS_PER_TIER 16 #define TFS_MAX_DISKS_PER_TIER 16
#define TSDB_MAX_DISKS (TSDB_MAX_TIERS * TSDB_MAX_DISKS_PER_TIER) #define TFS_MAX_DISKS (TFS_MAX_TIERS * TFS_MAX_DISKS_PER_TIER)
#define TFS_MIN_LEVEL 0
#define TFS_MAX_LEVEL (TFS_MAX_TIERS - 1)
#define TFS_PRIMARY_LEVEL 0
#define TFS_PRIMARY_ID 0
#define TFS_MIN_DISK_FREE_SIZE 50 * 1024 * 1024
enum { TRANS_STAT_INIT = 0, TRANS_STAT_EXECUTING, TRANS_STAT_EXECUTED, TRANS_STAT_ROLLBACKING, TRANS_STAT_ROLLBACKED }; enum { TRANS_STAT_INIT = 0, TRANS_STAT_EXECUTING, TRANS_STAT_EXECUTED, TRANS_STAT_ROLLBACKING, TRANS_STAT_ROLLBACKED };
enum { TRANS_OPER_INIT = 0, TRANS_OPER_EXECUTE, TRANS_OPER_ROLLBACK }; enum { TRANS_OPER_INIT = 0, TRANS_OPER_EXECUTE, TRANS_OPER_ROLLBACK };

2
source/common/src/tglobal.c
View File

@ -137,7 +137,7 @@ int32_t tsDiskCfgNum = 0;
#ifndef _STORAGE #ifndef _STORAGE
SDiskCfg tsDiskCfg[1]; SDiskCfg tsDiskCfg[1];
#else #else
SDiskCfg tsDiskCfg[TSDB_MAX_DISKS]; SDiskCfg tsDiskCfg[TFS_MAX_DISKS];
#endif #endif
/* /*

1
source/dnode/mgmt/impl/inc/dndEnv.h
View File

@ -134,6 +134,7 @@ typedef struct SDnode {
SBnodeMgmt bmgmt; SBnodeMgmt bmgmt;
SVnodesMgmt vmgmt; SVnodesMgmt vmgmt;
STransMgmt tmgmt; STransMgmt tmgmt;
STfs *pTfs;
SStartupReq startup; SStartupReq startup;
} SDnode; } SDnode;

1
source/dnode/mgmt/impl/inc/dndInt.h
View File

@ -43,6 +43,7 @@ extern "C" {
#include "qnode.h" #include "qnode.h"
#include "snode.h" #include "snode.h"
#include "vnode.h" #include "vnode.h"
#include "tfs.h"
extern int32_t dDebugFlag; extern int32_t dDebugFlag;

33
source/dnode/mgmt/impl/src/dndEnv.c
View File

@ -173,11 +173,12 @@ SDnode *dndCreate(SDnodeObjCfg *pCfg) {
return NULL; return NULL;
} }
SDiskCfg dCfg; SDiskCfg dCfg = {0};
strcpy(dCfg.dir, pDnode->cfg.dataDir); tstrncpy(dCfg.dir, pDnode->cfg.dataDir, TSDB_FILENAME_LEN);
dCfg.level = 0; dCfg.level = 0;
dCfg.primary = 1; dCfg.primary = 1;
if (tfsInit(&dCfg, 1) != 0) { pDnode->pTfs = tfsOpen(&dCfg, 1);
if (pDnode->pTfs == NULL) {
dError("failed to init tfs since %s", terrstr()); dError("failed to init tfs since %s", terrstr());
dndClose(pDnode); dndClose(pDnode);
return NULL; return NULL;
@ -251,7 +252,7 @@ void dndClose(SDnode *pDnode) {
dndCleanupQnode(pDnode); dndCleanupQnode(pDnode);
dndCleanupVnodes(pDnode); dndCleanupVnodes(pDnode);
dndCleanupMgmt(pDnode); dndCleanupMgmt(pDnode);
tfsCleanup(); tfsClose(pDnode->pTfs);
dndCloseImp(pDnode); dndCloseImp(pDnode);
free(pDnode); free(pDnode);
@ -314,3 +315,27 @@ void dndCleanup() {
taosStopCacheRefreshWorker(); taosStopCacheRefreshWorker();
dInfo("dnode env is cleaned up"); dInfo("dnode env is cleaned up");
} }
// OTHER FUNCTIONS ===================================
void taosGetDisk() {
#if 0
const double unit = 1024 * 1024 * 1024;
SDiskSize diskSize = tfsGetSize(pTfs);
tfsUpdateSize(&fsMeta);
tsTotalDataDirGB = (float)(fsMeta.total / unit);
tsUsedDataDirGB = (float)(fsMeta.used / unit);
tsAvailDataDirGB = (float)(fsMeta.avail / unit);
if (taosGetDiskSize(tsLogDir, &diskSize) == 0) {
tsTotalLogDirGB = (float)(diskSize.total / unit);
tsAvailLogDirGB = (float)(diskSize.avail / unit);
}
if (taosGetDiskSize(tsTempDir, &diskSize) == 0) {
tsTotalTmpDirGB = (float)(diskSize.total / unit);
tsAvailTmpDirectorySpace = (float)(diskSize.avail / unit);
}
#endif
}

2
source/dnode/vnode/inc/tsdb.h
View File

@ -80,7 +80,7 @@ typedef struct {
} STableKeyInfo; } STableKeyInfo;
// STsdb // STsdb
STsdb *tsdbOpen(const char *path, int32_t vgId, const STsdbCfg *pTsdbCfg, SMemAllocatorFactory *pMAF, SMeta *pMeta); STsdb *tsdbOpen(const char *path, int32_t vgId, const STsdbCfg *pTsdbCfg, SMemAllocatorFactory *pMAF, SMeta *pMeta, STfs *pTfs);
void tsdbClose(STsdb *); void tsdbClose(STsdb *);
void tsdbRemove(const char *path); void tsdbRemove(const char *path);
int tsdbInsertData(STsdb *pTsdb, SSubmitMsg *pMsg, SSubmitRsp *pRsp); int tsdbInsertData(STsdb *pTsdb, SSubmitMsg *pMsg, SSubmitRsp *pRsp);

1
source/dnode/vnode/inc/vnode.h
View File

@ -37,6 +37,7 @@ typedef int32_t (*PutReqToVQueryQFp)(SDnode *pDnode, struct SRpcMsg *pReq);
typedef struct SVnodeCfg { typedef struct SVnodeCfg {
int32_t vgId; int32_t vgId;
SDnode *pDnode; SDnode *pDnode;
STfs *pTfs;
uint64_t wsize; uint64_t wsize;
uint64_t ssize; uint64_t ssize;
uint64_t lsize; uint64_t lsize;

1
source/dnode/vnode/src/inc/tsdbDef.h
View File

@ -50,6 +50,7 @@ struct STsdb {
SMemAllocatorFactory *pmaf; SMemAllocatorFactory *pmaf;
STsdbFS * fs; STsdbFS * fs;
SMeta * pMeta; SMeta * pMeta;
STfs * pTfs;
}; };
#define REPO_ID(r) ((r)->vgId) #define REPO_ID(r) ((r)->vgId)

4
source/dnode/vnode/src/inc/tsdbFile.h
View File

@ -55,7 +55,7 @@ typedef struct {
typedef struct { typedef struct {
SMFInfo info; SMFInfo info;
TFILE f; STfsFile f;
int fd; int fd;
uint8_t state; uint8_t state;
} SMFile; } SMFile;
@ -176,7 +176,7 @@ typedef struct {
typedef struct { typedef struct {
SDFInfo info; SDFInfo info;
TFILE f; STfsFile f;
int fd; int fd;
uint8_t state; uint8_t state;
} SDFile; } SDFile;

1
source/dnode/vnode/src/inc/vnd.h
View File

@ -78,6 +78,7 @@ struct SVnode {
tsem_t canCommit; tsem_t canCommit;
SQHandle* pQuery; SQHandle* pQuery;
SDnode* pDnode; SDnode* pDnode;
STfs* pTfs;
}; };
int vnodeScheduleTask(SVnodeTask* task); int vnodeScheduleTask(SVnodeTask* task);

6
source/dnode/vnode/src/tsdb/tsdbCommit.c
View File

@ -97,8 +97,7 @@ int tsdbApplyRtnOnFSet(STsdb *pRepo, SDFileSet *pSet, SRtn *pRtn) {
level = tsdbGetFidLevel(pSet->fid, pRtn); level = tsdbGetFidLevel(pSet->fid, pRtn);
tfsAllocDisk(level, &(did.level), &(did.id)); if (tfsAllocDisk(pRepo->pTfs, level, &did) < 0) {
if (did.level == TFS_UNDECIDED_LEVEL) {
terrno = TSDB_CODE_TDB_NO_AVAIL_DISK; terrno = TSDB_CODE_TDB_NO_AVAIL_DISK;
return -1; return -1;
} }
@ -456,8 +455,7 @@ static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid
STsdb * pRepo = TSDB_COMMIT_REPO(pCommith); STsdb * pRepo = TSDB_COMMIT_REPO(pCommith);
SDFileSet *pWSet = TSDB_COMMIT_WRITE_FSET(pCommith); SDFileSet *pWSet = TSDB_COMMIT_WRITE_FSET(pCommith);
tfsAllocDisk(tsdbGetFidLevel(fid, &(pCommith->rtn)), &(did.level), &(did.id)); if (tfsAllocDisk(pRepo->pTfs, tsdbGetFidLevel(fid, &(pCommith->rtn)), &did) < 0) {
if (did.level == TFS_UNDECIDED_LEVEL) {
terrno = TSDB_CODE_TDB_NO_AVAIL_DISK; terrno = TSDB_CODE_TDB_NO_AVAIL_DISK;
return -1; return -1;
} }

3
source/dnode/vnode/src/tsdb/tsdbCompact.c
View File

@ -186,8 +186,7 @@ static int tsdbCompactMeta(STsdbRepo *pRepo) {
} }
} else { } else {
// Create new fset as compacted fset // Create new fset as compacted fset
tfsAllocDisk(tsdbGetFidLevel(pSet->fid, &(pComph->rtn)), &(did.level), &(did.id)); if (tfsAllocDisk(pRepo->pTfs, tsdbGetFidLevel(pSet->fid, &(pComph->rtn)), &did) < 0) {
if (did.level == TFS_UNDECIDED_LEVEL) {
terrno = TSDB_CODE_TDB_NO_AVAIL_DISK; terrno = TSDB_CODE_TDB_NO_AVAIL_DISK;
tsdbError("vgId:%d failed to compact FSET %d since %s", REPO_ID(pRepo), pSet->fid, tstrerror(terrno)); tsdbError("vgId:%d failed to compact FSET %d since %s", REPO_ID(pRepo), pSet->fid, tstrerror(terrno));
tsdbCompactFSetEnd(pComph); tsdbCompactFSetEnd(pComph);

62
source/dnode/vnode/src/tsdb/tsdbFS.c
View File

@ -23,14 +23,14 @@ static const char *tsdbTxnFname[] = {"current.t", "current"};
static int tsdbComparFidFSet(const void *arg1, const void *arg2); static int tsdbComparFidFSet(const void *arg1, const void *arg2);
static void tsdbResetFSStatus(SFSStatus *pStatus); static void tsdbResetFSStatus(SFSStatus *pStatus);
static int tsdbSaveFSStatus(SFSStatus *pStatus, int vid); static int tsdbSaveFSStatus(STfs *pTfs, SFSStatus *pStatus, int vid);
static void tsdbApplyFSTxnOnDisk(SFSStatus *pFrom, SFSStatus *pTo); static void tsdbApplyFSTxnOnDisk(SFSStatus *pFrom, SFSStatus *pTo);
static void tsdbGetTxnFname(int repoid, TSDB_TXN_FILE_T ftype, char fname[]); static void tsdbGetTxnFname(STfs *pTfs, int repoid, TSDB_TXN_FILE_T ftype, char fname[]);
static int tsdbOpenFSFromCurrent(STsdb *pRepo); static int tsdbOpenFSFromCurrent(STsdb *pRepo);
static int tsdbScanAndTryFixFS(STsdb *pRepo); static int tsdbScanAndTryFixFS(STsdb *pRepo);
static int tsdbScanRootDir(STsdb *pRepo); static int tsdbScanRootDir(STsdb *pRepo);
static int tsdbScanDataDir(STsdb *pRepo); static int tsdbScanDataDir(STsdb *pRepo);
static bool tsdbIsTFileInFS(STsdbFS *pfs, const TFILE *pf); static bool tsdbIsTFileInFS(STsdbFS *pfs, const STfsFile *pf);
static int tsdbRestoreCurrent(STsdb *pRepo); static int tsdbRestoreCurrent(STsdb *pRepo);
static int tsdbComparTFILE(const void *arg1, const void *arg2); static int tsdbComparTFILE(const void *arg1, const void *arg2);
static void tsdbScanAndTryFixDFilesHeader(STsdb *pRepo, int32_t *nExpired); static void tsdbScanAndTryFixDFilesHeader(STsdb *pRepo, int32_t *nExpired);
@ -311,7 +311,7 @@ int tsdbOpenFS(STsdb *pRepo) {
ASSERT(pfs != NULL); ASSERT(pfs != NULL);
tsdbGetTxnFname(REPO_ID(pRepo), TSDB_TXN_CURR_FILE, current); tsdbGetTxnFname(pRepo->pTfs, REPO_ID(pRepo), TSDB_TXN_CURR_FILE, current);
tsdbGetRtnSnap(pRepo, &pRepo->rtn); tsdbGetRtnSnap(pRepo, &pRepo->rtn);
if (access(current, F_OK) == 0) { if (access(current, F_OK) == 0) {
@ -375,7 +375,7 @@ int tsdbEndFSTxn(STsdb *pRepo) {
SFSStatus *pStatus; SFSStatus *pStatus;
// Write current file system snapshot // Write current file system snapshot
if (tsdbSaveFSStatus(pfs->nstatus, REPO_ID(pRepo)) < 0) { if (tsdbSaveFSStatus(pRepo->pTfs, pfs->nstatus, REPO_ID(pRepo)) < 0) {
tsdbEndFSTxnWithError(pfs); tsdbEndFSTxnWithError(pfs);
return -1; return -1;
} }
@ -405,7 +405,7 @@ int tsdbEndFSTxnWithError(STsdbFS *pfs) {
int tsdbUpdateDFileSet(STsdbFS *pfs, const SDFileSet *pSet) { return tsdbAddDFileSetToStatus(pfs->nstatus, pSet); } int tsdbUpdateDFileSet(STsdbFS *pfs, const SDFileSet *pSet) { return tsdbAddDFileSetToStatus(pfs->nstatus, pSet); }
static int tsdbSaveFSStatus(SFSStatus *pStatus, int vid) { static int tsdbSaveFSStatus(STfs *pTfs, SFSStatus *pStatus, int vid) {
SFSHeader fsheader; SFSHeader fsheader;
void * pBuf = NULL; void * pBuf = NULL;
void * ptr; void * ptr;
@ -413,8 +413,8 @@ static int tsdbSaveFSStatus(SFSStatus *pStatus, int vid) {
char tfname[TSDB_FILENAME_LEN] = "\0"; char tfname[TSDB_FILENAME_LEN] = "\0";
char cfname[TSDB_FILENAME_LEN] = "\0"; char cfname[TSDB_FILENAME_LEN] = "\0";
tsdbGetTxnFname(vid, TSDB_TXN_TEMP_FILE, tfname); tsdbGetTxnFname(pTfs, vid, TSDB_TXN_TEMP_FILE, tfname);
tsdbGetTxnFname(vid, TSDB_TXN_CURR_FILE, cfname); tsdbGetTxnFname(pTfs, vid, TSDB_TXN_CURR_FILE, cfname);
int fd = open(tfname, O_WRONLY | O_CREAT | O_TRUNC | O_BINARY, 0755); int fd = open(tfname, O_WRONLY | O_CREAT | O_TRUNC | O_BINARY, 0755);
if (fd < 0) { if (fd < 0) {
@ -645,8 +645,8 @@ static int tsdbComparFidFSet(const void *arg1, const void *arg2) {
} }
} }
static void tsdbGetTxnFname(int repoid, TSDB_TXN_FILE_T ftype, char fname[]) { static void tsdbGetTxnFname(STfs *pTfs, int repoid, TSDB_TXN_FILE_T ftype, char fname[]) {
snprintf(fname, TSDB_FILENAME_LEN, "%s/vnode/vnode%d/tsdb/%s", TFS_PRIMARY_PATH(), repoid, tsdbTxnFname[ftype]); snprintf(fname, TSDB_FILENAME_LEN, "%s/vnode/vnode%d/tsdb/%s", tfsGetPrimaryPath(pTfs), repoid, tsdbTxnFname[ftype]);
} }
static int tsdbOpenFSFromCurrent(STsdb *pRepo) { static int tsdbOpenFSFromCurrent(STsdb *pRepo) {
@ -657,7 +657,7 @@ static int tsdbOpenFSFromCurrent(STsdb *pRepo) {
char current[TSDB_FILENAME_LEN] = "\0"; char current[TSDB_FILENAME_LEN] = "\0";
void * ptr; void * ptr;
tsdbGetTxnFname(REPO_ID(pRepo), TSDB_TXN_CURR_FILE, current); tsdbGetTxnFname(pRepo->pTfs, REPO_ID(pRepo), TSDB_TXN_CURR_FILE, current);
// current file exists, try to recover // current file exists, try to recover
fd = open(current, O_RDONLY | O_BINARY); fd = open(current, O_RDONLY | O_BINARY);
@ -910,17 +910,17 @@ static int tsdbScanRootDir(STsdb *pRepo) {
char rootDir[TSDB_FILENAME_LEN]; char rootDir[TSDB_FILENAME_LEN];
char bname[TSDB_FILENAME_LEN]; char bname[TSDB_FILENAME_LEN];
STsdbFS * pfs = REPO_FS(pRepo); STsdbFS * pfs = REPO_FS(pRepo);
const TFILE *pf; const STfsFile *pf;
tsdbGetRootDir(REPO_ID(pRepo), rootDir); tsdbGetRootDir(REPO_ID(pRepo), rootDir);
TDIR *tdir = tfsOpendir(rootDir); STfsDir *tdir = tfsOpendir(rootDir);
if (tdir == NULL) { if (tdir == NULL) {
tsdbError("vgId:%d failed to open directory %s since %s", REPO_ID(pRepo), rootDir, tstrerror(terrno)); tsdbError("vgId:%d failed to open directory %s since %s", REPO_ID(pRepo), rootDir, tstrerror(terrno));
return -1; return -1;
} }
while ((pf = tfsReaddir(tdir))) { while ((pf = tfsReaddir(tdir))) {
tfsbasename(pf, bname); tfsBasename(pf, bname);
if (strcmp(bname, tsdbTxnFname[TSDB_TXN_CURR_FILE]) == 0 || strcmp(bname, "data") == 0) { if (strcmp(bname, tsdbTxnFname[TSDB_TXN_CURR_FILE]) == 0 || strcmp(bname, "data") == 0) {
// Skip current file and data directory // Skip current file and data directory
@ -944,17 +944,17 @@ static int tsdbScanDataDir(STsdb *pRepo) {
char dataDir[TSDB_FILENAME_LEN]; char dataDir[TSDB_FILENAME_LEN];
char bname[TSDB_FILENAME_LEN]; char bname[TSDB_FILENAME_LEN];
STsdbFS * pfs = REPO_FS(pRepo); STsdbFS * pfs = REPO_FS(pRepo);
const TFILE *pf; const STfsFile *pf;
tsdbGetDataDir(REPO_ID(pRepo), dataDir); tsdbGetDataDir(REPO_ID(pRepo), dataDir);
TDIR *tdir = tfsOpendir(dataDir); STfsDir *tdir = tfsOpendir(dataDir);
if (tdir == NULL) { if (tdir == NULL) {
tsdbError("vgId:%d failed to open directory %s since %s", REPO_ID(pRepo), dataDir, tstrerror(terrno)); tsdbError("vgId:%d failed to open directory %s since %s", REPO_ID(pRepo), dataDir, tstrerror(terrno));
return -1; return -1;
} }
while ((pf = tfsReaddir(tdir))) { while ((pf = tfsReaddir(tdir))) {
tfsbasename(pf, bname); tfsBasename(pf, bname);
if (!tsdbIsTFileInFS(pfs, pf)) { if (!tsdbIsTFileInFS(pfs, pf)) {
(void)tfsremove(pf); (void)tfsremove(pf);
@ -967,7 +967,7 @@ static int tsdbScanDataDir(STsdb *pRepo) {
return 0; return 0;
} }
static bool tsdbIsTFileInFS(STsdbFS *pfs, const TFILE *pf) { static bool tsdbIsTFileInFS(STsdbFS *pfs, const STfsFile *pf) {
SFSIter fsiter; SFSIter fsiter;
tsdbFSIterInit(&fsiter, pfs, TSDB_FS_ITER_FORWARD); tsdbFSIterInit(&fsiter, pfs, TSDB_FS_ITER_FORWARD);
SDFileSet *pSet; SDFileSet *pSet;
@ -987,8 +987,8 @@ static bool tsdbIsTFileInFS(STsdbFS *pfs, const TFILE *pf) {
// static int tsdbRestoreMeta(STsdb *pRepo) { // static int tsdbRestoreMeta(STsdb *pRepo) {
// char rootDir[TSDB_FILENAME_LEN]; // char rootDir[TSDB_FILENAME_LEN];
// char bname[TSDB_FILENAME_LEN]; // char bname[TSDB_FILENAME_LEN];
// TDIR * tdir = NULL; // STfsDir * tdir = NULL;
// const TFILE *pf = NULL; // const STfsFile *pf = NULL;
// const char * pattern = "^meta(-ver[0-9]+)?$"; // const char * pattern = "^meta(-ver[0-9]+)?$";
// regex_t regex; // regex_t regex;
// STsdbFS * pfs = REPO_FS(pRepo); // STsdbFS * pfs = REPO_FS(pRepo);
@ -1007,7 +1007,7 @@ static bool tsdbIsTFileInFS(STsdbFS *pfs, const TFILE *pf) {
// } // }
// while ((pf = tfsReaddir(tdir))) { // while ((pf = tfsReaddir(tdir))) {
// tfsbasename(pf, bname); // tfsBasename(pf, bname);
// if (strcmp(bname, "data") == 0) { // if (strcmp(bname, "data") == 0) {
// // Skip the data/ directory // // Skip the data/ directory
@ -1108,8 +1108,8 @@ static bool tsdbIsTFileInFS(STsdbFS *pfs, const TFILE *pf) {
static int tsdbRestoreDFileSet(STsdb *pRepo) { static int tsdbRestoreDFileSet(STsdb *pRepo) {
char dataDir[TSDB_FILENAME_LEN]; char dataDir[TSDB_FILENAME_LEN];
char bname[TSDB_FILENAME_LEN]; char bname[TSDB_FILENAME_LEN];
TDIR * tdir = NULL; STfsDir * tdir = NULL;
const TFILE *pf = NULL; const STfsFile *pf = NULL;
const char * pattern = "^v[0-9]+f[0-9]+\\.(head|data|last)(-ver[0-9]+)?$"; const char * pattern = "^v[0-9]+f[0-9]+\\.(head|data|last)(-ver[0-9]+)?$";
SArray * fArray = NULL; SArray * fArray = NULL;
regex_t regex; regex_t regex;
@ -1120,7 +1120,7 @@ static int tsdbRestoreDFileSet(STsdb *pRepo) {
// Resource allocation and init // Resource allocation and init
regcomp(&regex, pattern, REG_EXTENDED); regcomp(&regex, pattern, REG_EXTENDED);
fArray = taosArrayInit(1024, sizeof(TFILE)); fArray = taosArrayInit(1024, sizeof(STfsFile));
if (fArray == NULL) { if (fArray == NULL) {
terrno = TSDB_CODE_TDB_OUT_OF_MEMORY; terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
tsdbError("vgId:%d failed to restore DFileSet while open directory %s since %s", REPO_ID(pRepo), dataDir, tsdbError("vgId:%d failed to restore DFileSet while open directory %s since %s", REPO_ID(pRepo), dataDir,
@ -1139,7 +1139,7 @@ static int tsdbRestoreDFileSet(STsdb *pRepo) {
} }
while ((pf = tfsReaddir(tdir))) { while ((pf = tfsReaddir(tdir))) {
tfsbasename(pf, bname); tfsBasename(pf, bname);
int code = regexec(&regex, bname, 0, NULL, 0); int code = regexec(&regex, bname, 0, NULL, 0);
if (code == 0) { if (code == 0) {
@ -1200,7 +1200,7 @@ static int tsdbRestoreDFileSet(STsdb *pRepo) {
uint32_t tversion; uint32_t tversion;
char _bname[TSDB_FILENAME_LEN]; char _bname[TSDB_FILENAME_LEN];
tfsbasename(pf, _bname); tfsBasename(pf, _bname);
tsdbParseDFilename(_bname, &tvid, &tfid, &ttype, &tversion); tsdbParseDFilename(_bname, &tvid, &tfid, &ttype, &tversion);
ASSERT(tvid == REPO_ID(pRepo)); ASSERT(tvid == REPO_ID(pRepo));
@ -1287,7 +1287,7 @@ static int tsdbRestoreCurrent(STsdb *pRepo) {
return -1; return -1;
} }
if (tsdbSaveFSStatus(pRepo->fs->cstatus, REPO_ID(pRepo)) < 0) { if (tsdbSaveFSStatus(pRepo->pTfs, pRepo->fs->cstatus, REPO_ID(pRepo)) < 0) {
tsdbError("vgId:%d failed to restore corrent since %s", REPO_ID(pRepo), tstrerror(terrno)); tsdbError("vgId:%d failed to restore corrent since %s", REPO_ID(pRepo), tstrerror(terrno));
return -1; return -1;
} }
@ -1296,8 +1296,8 @@ static int tsdbRestoreCurrent(STsdb *pRepo) {
} }
static int tsdbComparTFILE(const void *arg1, const void *arg2) { static int tsdbComparTFILE(const void *arg1, const void *arg2) {
TFILE *pf1 = (TFILE *)arg1; STfsFile *pf1 = (STfsFile *)arg1;
TFILE *pf2 = (TFILE *)arg2; STfsFile *pf2 = (STfsFile *)arg2;
int vid1, fid1, vid2, fid2; int vid1, fid1, vid2, fid2;
TSDB_FILE_T ftype1, ftype2; TSDB_FILE_T ftype1, ftype2;
@ -1305,8 +1305,8 @@ static int tsdbComparTFILE(const void *arg1, const void *arg2) {
char bname1[TSDB_FILENAME_LEN]; char bname1[TSDB_FILENAME_LEN];
char bname2[TSDB_FILENAME_LEN]; char bname2[TSDB_FILENAME_LEN];
tfsbasename(pf1, bname1); tfsBasename(pf1, bname1);
tfsbasename(pf2, bname2); tfsBasename(pf2, bname2);
tsdbParseDFilename(bname1, &vid1, &fid1, &ftype1, &version1); tsdbParseDFilename(bname1, &vid1, &fid1, &ftype1, &version1);
tsdbParseDFilename(bname2, &vid2, &fid2, &ftype2, &version2); tsdbParseDFilename(bname2, &vid2, &fid2, &ftype2, &version2);

11
source/dnode/vnode/src/tsdb/tsdbMain.c
View File

@ -16,12 +16,12 @@
#include "tsdbDef.h" #include "tsdbDef.h"
static STsdb *tsdbNew(const char *path, int32_t vgId, const STsdbCfg *pTsdbCfg, SMemAllocatorFactory *pMAF, static STsdb *tsdbNew(const char *path, int32_t vgId, const STsdbCfg *pTsdbCfg, SMemAllocatorFactory *pMAF,
SMeta *pMeta); SMeta *pMeta, STfs *pTfs);
static void tsdbFree(STsdb *pTsdb); static void tsdbFree(STsdb *pTsdb);
static int tsdbOpenImpl(STsdb *pTsdb); static int tsdbOpenImpl(STsdb *pTsdb);
static void tsdbCloseImpl(STsdb *pTsdb); static void tsdbCloseImpl(STsdb *pTsdb);
STsdb *tsdbOpen(const char *path, int32_t vgId, const STsdbCfg *pTsdbCfg, SMemAllocatorFactory *pMAF, SMeta *pMeta) { STsdb *tsdbOpen(const char *path, int32_t vgId, const STsdbCfg *pTsdbCfg, SMemAllocatorFactory *pMAF, SMeta *pMeta, STfs *pTfs) {
STsdb *pTsdb = NULL; STsdb *pTsdb = NULL;
// Set default TSDB Options // Set default TSDB Options
@ -36,7 +36,7 @@ STsdb *tsdbOpen(const char *path, int32_t vgId, const STsdbCfg *pTsdbCfg, SMemAl
} }
// Create the handle // Create the handle
pTsdb = tsdbNew(path, vgId, pTsdbCfg, pMAF, pMeta); pTsdb = tsdbNew(path, vgId, pTsdbCfg, pMAF, pMeta, pTfs);
if (pTsdb == NULL) { if (pTsdb == NULL) {
// TODO: handle error // TODO: handle error
return NULL; return NULL;
@ -64,7 +64,7 @@ void tsdbRemove(const char *path) { taosRemoveDir(path); }
/* ------------------------ STATIC METHODS ------------------------ */ /* ------------------------ STATIC METHODS ------------------------ */
static STsdb *tsdbNew(const char *path, int32_t vgId, const STsdbCfg *pTsdbCfg, SMemAllocatorFactory *pMAF, static STsdb *tsdbNew(const char *path, int32_t vgId, const STsdbCfg *pTsdbCfg, SMemAllocatorFactory *pMAF,
SMeta *pMeta) { SMeta *pMeta, STfs *pTfs) {
STsdb *pTsdb = NULL; STsdb *pTsdb = NULL;
pTsdb = (STsdb *)calloc(1, sizeof(STsdb)); pTsdb = (STsdb *)calloc(1, sizeof(STsdb));
@ -78,6 +78,7 @@ static STsdb *tsdbNew(const char *path, int32_t vgId, const STsdbCfg *pTsdbCfg,
tsdbOptionsCopy(&(pTsdb->config), pTsdbCfg); tsdbOptionsCopy(&(pTsdb->config), pTsdbCfg);
pTsdb->pmaf = pMAF; pTsdb->pmaf = pMAF;
pTsdb->pMeta = pMeta; pTsdb->pMeta = pMeta;
pTsdb->pTfs = pTfs;
pTsdb->fs = tsdbNewFS(pTsdbCfg); pTsdb->fs = tsdbNewFS(pTsdbCfg);
@ -494,7 +495,7 @@ uint32_t tsdbGetFileInfo(STsdbRepo *repo, char *name, uint32_t *index, uint32_t
} }
} else { // get the named file at the specified index. If not there, return 0 } else { // get the named file at the specified index. If not there, return 0
fname = malloc(256); fname = malloc(256);
sprintf(fname, "%s/vnode/vnode%d/%s", TFS_PRIMARY_PATH(), REPO_ID(pRepo), name); sprintf(fname, "%s/vnode/vnode%d/%s", tfsGetPrimaryPath(pRepo->pTfs), REPO_ID(pRepo), name);
if (access(fname, F_OK) != 0) { if (access(fname, F_OK) != 0) {
tfree(fname); tfree(fname);
return 0; return 0;

3
source/dnode/vnode/src/vnd/vnodeMain.c
View File

@ -75,6 +75,7 @@ static SVnode *vnodeNew(const char *path, const SVnodeCfg *pVnodeCfg) {
pVnode->vgId = pVnodeCfg->vgId; pVnode->vgId = pVnodeCfg->vgId;
pVnode->pDnode = pVnodeCfg->pDnode; pVnode->pDnode = pVnodeCfg->pDnode;
pVnode->pTfs = pVnodeCfg->pTfs;
pVnode->path = strdup(path); pVnode->path = strdup(path);
vnodeOptionsCopy(&(pVnode->config), pVnodeCfg); vnodeOptionsCopy(&(pVnode->config), pVnodeCfg);
@ -109,7 +110,7 @@ static int vnodeOpenImpl(SVnode *pVnode) {
// Open tsdb // Open tsdb
sprintf(dir, "%s/tsdb", pVnode->path); sprintf(dir, "%s/tsdb", pVnode->path);
pVnode->pTsdb = tsdbOpen(dir, pVnode->vgId, &(pVnode->config.tsdbCfg), vBufPoolGetMAF(pVnode), pVnode->pMeta); pVnode->pTsdb = tsdbOpen(dir, pVnode->vgId, &(pVnode->config.tsdbCfg), vBufPoolGetMAF(pVnode), pVnode->pMeta, pVnode->pTfs);
if (pVnode->pTsdb == NULL) { if (pVnode->pTsdb == NULL) {
// TODO: handle error // TODO: handle error
return -1; return -1;

82
source/libs/tfs/inc/tfsInt.h
View File

@ -33,28 +33,73 @@ extern int32_t fsDebugFlag;
#define fError(...) { if (fsDebugFlag & DEBUG_ERROR) { taosPrintLog("TFS ERROR ", 255, __VA_ARGS__); }} #define fError(...) { if (fsDebugFlag & DEBUG_ERROR) { taosPrintLog("TFS ERROR ", 255, __VA_ARGS__); }}
#define fWarn(...) { if (fsDebugFlag & DEBUG_WARN) { taosPrintLog("TFS WARN ", 255, __VA_ARGS__); }} #define fWarn(...) { if (fsDebugFlag & DEBUG_WARN) { taosPrintLog("TFS WARN ", 255, __VA_ARGS__); }}
#define fInfo(...) { if (fsDebugFlag & DEBUG_INFO) { taosPrintLog("TFS ", 255, __VA_ARGS__); }} #define fInfo(...) { if (fsDebugFlag & DEBUG_INFO) { taosPrintLog("TFS ", 255, __VA_ARGS__); }}
#define fDebug(...) { if (fsDebugFlag & DEBUG_DEBUG) { taosPrintLog("TFS ", cqDebugFlag, __VA_ARGS__); }} #define fDebug(...) { if (fsDebugFlag & DEBUG_DEBUG) { taosPrintLog("TFS ", fsDebugFlag, __VA_ARGS__); }}
#define fTrace(...) { if (fsDebugFlag & DEBUG_TRACE) { taosPrintLog("TFS ", cqDebugFlag, __VA_ARGS__); }} #define fTrace(...) { if (fsDebugFlag & DEBUG_TRACE) { taosPrintLog("TFS ", fsDebugFlag, __VA_ARGS__); }}
// Global Definitions typedef struct {
#define TFS_MIN_DISK_FREE_SIZE 50 * 1024 * 1024
typedef struct SDisk {
int32_t level; int32_t level;
int32_t id; int32_t id;
char *path; char *path;
SDiskSize size; SDiskSize size;
} SDisk; } STfsDisk;
typedef struct STier { typedef struct {
pthread_spinlock_t lock; pthread_spinlock_t lock;
int32_t level; int32_t level;
int16_t nextid; // next disk id to allocate int32_t nextid; // next disk id to allocate
int16_t ndisk; // # of disks mounted to this tier int32_t ndisk; // # of disks mounted to this tier
int16_t nAvailDisks; // # of Available disks int32_t nAvailDisks; // # of Available disks
SDisk *disks[TSDB_MAX_DISKS_PER_TIER]; STfsDisk *disks[TFS_MAX_DISKS_PER_TIER];
SDiskSize size; SDiskSize size;
} STier; } STfsTier;
typedef struct {
STfsDisk *pDisk;
} SDiskIter;
typedef struct STfsDir {
SDiskIter *iter;
SDiskID did;
char dirname[TSDB_FILENAME_LEN];
STfsFile tfile;
DIR *dir;
STfs *pTfs;
} STfsDir;
typedef struct STfs {
pthread_spinlock_t lock;
SDiskSize size;
int32_t nlevel;
STfsTier tiers[TFS_MAX_TIERS];
SHashObj *hash; // name to did map
} STfs;
STfsDisk *tfsNewDisk(int32_t level, int32_t id, const char *dir);
STfsDisk *tfsFreeDisk(STfsDisk *pDisk);
int32_t tfsUpdateDiskSize(STfsDisk *pDisk);
int32_t tfsInitTier(STfsTier *pTier, int32_t level);
void tfsDestroyTier(STfsTier *pTier);
STfsDisk *tfsMountDiskToTier(STfsTier *pTier, SDiskCfg *pCfg);
void tfsUpdateTierSize(STfsTier *pTier);
int32_t tfsAllocDiskOnTier(STfsTier *pTier);
void tfsPosNextId(STfsTier *pTier);
#define tfsLockTier(pTier) pthread_spin_lock(&(pTier)->lock)
#define tfsUnLockTier(pTier) pthread_spin_unlock(&(pTier)->lock)
#define TMPNAME_LEN (TSDB_FILENAME_LEN * 2 + 32)
#define tfsLock(pTfs) pthread_spin_lock(&(pTfs)->lock)
#define tfsUnLock(pTfs) pthread_spin_unlock(&(pTfs)->lock)
#define TFS_TIER_AT(pTfs, level) (&pTfs->tiers[level])
#define TFS_DISK_AT(pTfs, did) (pTfs->tiers[(did).level].disks[(did).id])
#define TFS_PRIMARY_DISK(pTfs) (pTfs->tiers[0].disks[0])
#define TFS_IS_VALID_LEVEL(level) (((level) >= 0) && ((level) < TFS_NLEVEL()))
#define TFS_IS_VALID_ID(level, id) (((id) >= 0) && ((id) < TIER_NDISKS(TFS_TIER_AT(level))))
#define TFS_IS_VALID_DISK(level, id) (TFS_IS_VALID_LEVEL(level) && TFS_IS_VALID_ID(level, id))
#define TIER_LEVEL(pt) ((pt)->level) #define TIER_LEVEL(pt) ((pt)->level)
#define TIER_NDISKS(pt) ((pt)->ndisk) #define TIER_NDISKS(pt) ((pt)->ndisk)
@ -64,17 +109,6 @@ typedef struct STier {
#define DISK_AT_TIER(pt, id) ((pt)->disks[id]) #define DISK_AT_TIER(pt, id) ((pt)->disks[id])
#define DISK_DIR(pd) ((pd)->path) #define DISK_DIR(pd) ((pd)->path)
SDisk *tfsNewDisk(int32_t level, int32_t id, const char *dir);
SDisk *tfsFreeDisk(SDisk *pDisk);
int32_t tfsUpdateDiskSize(SDisk *pDisk);
int32_t tfsInitTier(STier *pTier, int32_t level);
void tfsDestroyTier(STier *pTier);
SDisk *tfsMountDiskToTier(STier *pTier, SDiskCfg *pCfg);
void tfsUpdateTierSize(STier *pTier);
int32_t tfsAllocDiskOnTier(STier *pTier);
void tfsPosNextId(STier *pTier);
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif

527
source/libs/tfs/src/tfs.c
View File

@ -16,227 +16,193 @@
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE
#include "tfsInt.h" #include "tfsInt.h"
#define TMPNAME_LEN (TSDB_FILENAME_LEN * 2 + 32) static int32_t tfsMount(STfs *pTfs, SDiskCfg *pCfg);
static int32_t tfsCheck(STfs *pTfs);
typedef struct { static int32_t tfsCheckAndFormatCfg(STfs *pTfs, SDiskCfg *pCfg);
pthread_spinlock_t lock;
SFSMeta meta;
int32_t nlevel;
STier tiers[TSDB_MAX_TIERS];
SHashObj *map; // name to did map
} SFS;
typedef struct {
SDisk *pDisk;
} SDiskIter;
#define TFS_META() (pfs->meta)
#define TFS_NLEVEL() (pfs->nlevel)
#define TFS_TIERS() (pfs->tiers)
#define TFS_TIER_AT(level) (TFS_TIERS() + (level))
#define TFS_DISK_AT(level, id) DISK_AT_TIER(TFS_TIER_AT(level), id)
#define TFS_PRIMARY_DISK() TFS_DISK_AT(TFS_PRIMARY_LEVEL, TFS_PRIMARY_ID)
#define TFS_IS_VALID_LEVEL(level) (((level) >= 0) && ((level) < TFS_NLEVEL()))
#define TFS_IS_VALID_ID(level, id) (((id) >= 0) && ((id) < TIER_NDISKS(TFS_TIER_AT(level))))
#define TFS_IS_VALID_DISK(level, id) (TFS_IS_VALID_LEVEL(level) && TFS_IS_VALID_ID(level, id))
#define tfsLock() pthread_spin_lock(&(pfs->lock))
#define tfsUnLock() pthread_spin_unlock(&(pfs->lock))
static SFS tfs = {0};
static SFS *pfs = &tfs;
// STATIC DECLARATION
static int32_t tfsMount(SDiskCfg *pCfg);
static int32_t tfsCheck();
static int32_t tfsCheckAndFormatCfg(SDiskCfg *pCfg);
static int32_t tfsFormatDir(char *idir, char *odir); static int32_t tfsFormatDir(char *idir, char *odir);
static SDisk *tfsGetDiskByID(SDiskID did); static STfsDisk *tfsGetDiskByName(STfs *pTfs, const char *dir);
static SDisk *tfsGetDiskByName(const char *dir); static int32_t tfsOpendirImpl(STfs *pTfs, STfsDir *pDir);
static int32_t tfsOpendirImpl(TDIR *tdir); static STfsDisk *tfsNextDisk(STfs *pTfs, SDiskIter *pIter);
static void tfsInitDiskIter(SDiskIter *pIter);
static SDisk *tfsNextDisk(SDiskIter *pIter);
// FS APIs ==================================== STfs *tfsOpen(SDiskCfg *pCfg, int32_t ndisk) {
int32_t tfsInit(SDiskCfg *pDiskCfg, int32_t ndisk) { if (ndisk < 0 || ndisk > TFS_MAX_DISKS) {
if (ndisk < 0) {
terrno = TSDB_CODE_INVALID_PARA; terrno = TSDB_CODE_INVALID_PARA;
return -1; return NULL;
} }
for (int32_t level = 0; level < TSDB_MAX_TIERS; level++) { STfs *pTfs = calloc(1, sizeof(STfs));
if (tfsInitTier(TFS_TIER_AT(level), level) < 0) { if (pTfs == NULL) {
while (true) {
level--;
if (level < 0) break;
tfsDestroyTier(TFS_TIER_AT(level));
}
return -1;
}
}
pthread_spin_init(&(pfs->lock), 0);
pfs->map = taosHashInit(TSDB_MAX_TIERS * TSDB_MAX_DISKS_PER_TIER * 2,
taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
if (pfs->map == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
tfsCleanup(); return NULL;
return -1; }
if (pthread_spin_init(&pTfs->lock, 0) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
tfsClose(pTfs);
return NULL;
}
for (int32_t level = 0; level < TFS_MAX_TIERS; level++) {
STfsTier *pTier = &pTfs->tiers[level];
if (tfsInitTier(pTier, level) < 0) {
tfsClose(pTfs);
return NULL;
}
}
pTfs->hash = taosHashInit(TFS_MAX_DISKS * 2, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
if (pTfs->hash == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
tfsClose(pTfs);
return NULL;
} }
for (int32_t idisk = 0; idisk < ndisk; idisk++) { for (int32_t idisk = 0; idisk < ndisk; idisk++) {
if (tfsMount(pDiskCfg + idisk) < 0) { if (tfsMount(pTfs, &pCfg[idisk]) < 0) {
tfsCleanup(); tfsClose(pTfs);
return -1; return NULL;
} }
} }
if (tfsCheck() < 0) { if (tfsCheck(pTfs) < 0) {
tfsCleanup(); tfsClose(pTfs);
return -1; return NULL;
} }
tfsUpdateSize(NULL); tfsUpdateSize(pTfs);
for (int32_t level = 0; level < TFS_NLEVEL(); level++) { for (int32_t level = 0; level < pTfs->nlevel; level++) {
tfsPosNextId(TFS_TIER_AT(level)); tfsPosNextId(&pTfs->tiers[level]);
}
return pTfs;
}
void tfsClose(STfs *pTfs) {
if (pTfs == NULL) return;
for (int32_t level = 0; level < TFS_MAX_LEVEL; level++) {
tfsDestroyTier(&pTfs->tiers[level]);
}
taosHashCleanup(pTfs->hash);
pthread_spin_destroy(&pTfs->lock);
free(pTfs);
}
void tfsUpdateSize(STfs *pTfs) {
SDiskSize size = {0};
for (int32_t level = 0; level < pTfs->nlevel; level++) {
STfsTier *pTier = &pTfs->tiers[level];
tfsUpdateTierSize(pTier);
size.total += pTier->size.total;
size.avail += pTier->size.avail;
size.used += pTier->size.used;
}
tfsLock(pTfs);
pTfs->size = size;
tfsUnLock(pTfs);
}
SDiskSize tfsGetSize(STfs *pTfs) {
tfsLock(pTfs);
SDiskSize size = pTfs->size;
tfsUnLock(pTfs);
return size;
}
int32_t tfsAllocDisk(STfs *pTfs, int32_t expLevel, SDiskID *pDiskId) {
pDiskId->level = expLevel;
pDiskId->id = -1;
if (pDiskId->level >= pTfs->nlevel) {
pDiskId->level--;
}
while (pDiskId->level >= 0) {
pDiskId->id = tfsAllocDiskOnTier(&pTfs->tiers[pDiskId->level]);
if (pDiskId->id < 0) {
pDiskId->level--;
continue;
} }
return 0; return 0;
} }
void tfsCleanup() { terrno = TSDB_CODE_FS_NO_VALID_DISK;
taosHashCleanup(pfs->map); return -1;
pfs->map = NULL;
pthread_spin_destroy(&(pfs->lock));
for (int32_t level = 0; level < TFS_NLEVEL(); level++) {
tfsDestroyTier(TFS_TIER_AT(level));
}
} }
void tfsUpdateSize(SFSMeta *pFSMeta) { const char *tfsGetPrimaryPath(STfs *pTfs) { return TFS_PRIMARY_DISK(pTfs)->path; }
SFSMeta fsMeta = {0};
SDiskSize size = {0};
if (pFSMeta == NULL) { const char *tfsGetDiskPath(STfs *pTfs, SDiskID diskId) { return TFS_DISK_AT(pTfs, diskId)->path; }
pFSMeta = &fsMeta;
}
memset(pFSMeta, 0, sizeof(SFSMeta)); void tfsInitFile(STfs *pTfs, STfsFile *pFile, SDiskID diskId, const char *rname) {
STfsDisk *pDisk = TFS_DISK_AT(pTfs, diskId);
for (int32_t level = 0; level < TFS_NLEVEL(); level++) { pFile->did = diskId;
STier *pTier = TFS_TIER_AT(level); tstrncpy(pFile->rname, rname, TSDB_FILENAME_LEN);
tfsUpdateTierSize(pTier);
pFSMeta->total += pTier->size.total;
pFSMeta->avail += pTier->size.avail;
pFSMeta->used += pTier->size.used;
}
tfsLock();
pfs->meta = *pFSMeta;
tfsUnLock();
}
/* Allocate an existing available tier level
*/
void tfsAllocDisk(int32_t expLevel, int32_t *level, int32_t *id) {
ASSERT(expLevel >= 0);
*level = expLevel;
*id = TFS_UNDECIDED_ID;
if (*level >= TFS_NLEVEL()) {
*level = TFS_NLEVEL() - 1;
}
while (*level >= 0) {
*id = tfsAllocDiskOnTier(TFS_TIER_AT(*level));
if (*id == TFS_UNDECIDED_ID) {
(*level)--;
continue;
}
return;
}
*level = TFS_UNDECIDED_LEVEL;
*id = TFS_UNDECIDED_ID;
}
const char *TFS_PRIMARY_PATH() { return DISK_DIR(TFS_PRIMARY_DISK()); }
const char *TFS_DISK_PATH(int32_t level, int32_t id) { return DISK_DIR(TFS_DISK_AT(level, id)); }
// TFILE APIs ====================================
void tfsInitFile(TFILE *pf, int32_t level, int32_t id, const char *bname) {
ASSERT(TFS_IS_VALID_DISK(level, id));
SDisk *pDisk = TFS_DISK_AT(level, id);
pf->level = level;
pf->id = id;
tstrncpy(pf->rname, bname, TSDB_FILENAME_LEN);
char tmpName[TMPNAME_LEN] = {0}; char tmpName[TMPNAME_LEN] = {0};
snprintf(tmpName, TMPNAME_LEN, "%s/%s", DISK_DIR(pDisk), bname); snprintf(tmpName, TMPNAME_LEN, "%s%s%s", DISK_DIR(pDisk), TD_DIRSEP, rname);
tstrncpy(pf->aname, tmpName, TSDB_FILENAME_LEN); tstrncpy(pFile->aname, tmpName, TSDB_FILENAME_LEN);
pFile->pTfs = pTfs;
} }
bool tfsIsSameFile(const TFILE *pf1, const TFILE *pf2) { bool tfsIsSameFile(const STfsFile *pFile1, const STfsFile *pFile2) {
ASSERT(pf1 != NULL || pf2 != NULL); ASSERT(pFile1 != NULL || pFile2 != NULL);
if (pf1 == NULL || pf2 == NULL) return false; if (pFile1 == NULL || pFile2 == NULL || pFile1->pTfs != pFile2->pTfs) return false;
if (pf1->level != pf2->level) return false; if (pFile1->did.level != pFile2->did.level) return false;
if (pf1->id != pf2->id) return false; if (pFile1->did.id != pFile2->did.id) return false;
if (strncmp(pf1->rname, pf2->rname, TSDB_FILENAME_LEN) != 0) return false; if (strncmp(pFile1->rname, pFile2->rname, TSDB_FILENAME_LEN) != 0) return false;
return true; return true;
} }
int32_t tfsEncodeFile(void **buf, TFILE *pf) { int32_t tfsEncodeFile(void **buf, STfsFile *pFile) {
int32_t tlen = 0; int32_t tlen = 0;
tlen += taosEncodeVariantI32(buf, pf->level); tlen += taosEncodeVariantI32(buf, pFile->did.level);
tlen += taosEncodeVariantI32(buf, pf->id); tlen += taosEncodeVariantI32(buf, pFile->did.id);
tlen += taosEncodeString(buf, pf->rname); tlen += taosEncodeString(buf, pFile->rname);
return tlen; return tlen;
} }
void *tfsDecodeFile(void *buf, TFILE *pf) { void *tfsDecodeFile(STfs *pTfs, void *buf, STfsFile *pFile) {
int32_t level, id; SDiskID diskId = {0};
char *rname; char *rname = NULL;
buf = taosDecodeVariantI32(buf, &(level)); buf = taosDecodeVariantI32(buf, &diskId.level);
buf = taosDecodeVariantI32(buf, &(id)); buf = taosDecodeVariantI32(buf, &diskId.id);
buf = taosDecodeString(buf, &rname); buf = taosDecodeString(buf, &rname);
tfsInitFile(pf, level, id, rname); tfsInitFile(pTfs, pFile, diskId, rname);
tfree(rname); tfree(rname);
return buf; return buf;
} }
void tfsbasename(const TFILE *pf, char *dest) { void tfsBasename(const STfsFile *pFile, char *dest) {
char tname[TSDB_FILENAME_LEN] = "\0"; char tname[TSDB_FILENAME_LEN] = "\0";
tstrncpy(tname, pf->aname, TSDB_FILENAME_LEN); tstrncpy(tname, pFile->aname, TSDB_FILENAME_LEN);
tstrncpy(dest, basename(tname), TSDB_FILENAME_LEN); tstrncpy(dest, basename(tname), TSDB_FILENAME_LEN);
} }
void tfsdirname(const TFILE *pf, char *dest) { void tfsDirname(const STfsFile *pFile, char *dest) {
char tname[TSDB_FILENAME_LEN] = "\0"; char tname[TSDB_FILENAME_LEN] = "\0";
tstrncpy(tname, pf->aname, TSDB_FILENAME_LEN); tstrncpy(tname, pFile->aname, TSDB_FILENAME_LEN);
tstrncpy(dest, dirname(tname), TSDB_FILENAME_LEN); tstrncpy(dest, dirname(tname), TSDB_FILENAME_LEN);
} }
// DIR APIs ==================================== int32_t tfsMkdirAt(STfs *pTfs, const char *rname, SDiskID diskId) {
int32_t tfsMkdirAt(const char *rname, int32_t level, int32_t id) { STfsDisk *pDisk = TFS_DISK_AT(pTfs, diskId);
SDisk *pDisk = TFS_DISK_AT(level, id);
char aname[TMPNAME_LEN]; char aname[TMPNAME_LEN];
snprintf(aname, TMPNAME_LEN, "%s/%s", DISK_DIR(pDisk), rname); snprintf(aname, TMPNAME_LEN, "%s%s%s", pDisk->path, TD_DIRSEP, rname);
if (taosMkDir(aname) != 0) { if (taosMkDir(aname) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno); terrno = TAOS_SYSTEM_ERROR(errno);
return -1; return -1;
@ -245,8 +211,8 @@ int32_t tfsMkdirAt(const char *rname, int32_t level, int32_t id) {
return 0; return 0;
} }
int32_t tfsMkdirRecurAt(const char *rname, int32_t level, int32_t id) { int32_t tfsMkdirRecurAt(STfs *pTfs, const char *rname, SDiskID diskId) {
if (tfsMkdirAt(rname, level, id) < 0) { if (tfsMkdirAt(pTfs, rname, diskId) < 0) {
if (errno == ENOENT) { if (errno == ENOENT) {
// Try to create upper // Try to create upper
char *s = strdup(rname); char *s = strdup(rname);
@ -259,7 +225,7 @@ int32_t tfsMkdirRecurAt(const char *rname, int32_t level, int32_t id) {
// https://developer.apple.com/library/archive/documentation/System/Conceptual/ManPages_iPhoneOS/man3/dirname.3.html // https://developer.apple.com/library/archive/documentation/System/Conceptual/ManPages_iPhoneOS/man3/dirname.3.html
char *dir = strdup(dirname(s)); char *dir = strdup(dirname(s));
if (tfsMkdirRecurAt(dir, level, id) < 0) { if (tfsMkdirRecurAt(pTfs, dir, diskId) < 0) {
free(s); free(s);
free(dir); free(dir);
return -1; return -1;
@ -267,7 +233,7 @@ int32_t tfsMkdirRecurAt(const char *rname, int32_t level, int32_t id) {
free(s); free(s);
free(dir); free(dir);
if (tfsMkdirAt(rname, level, id) < 0) { if (tfsMkdirAt(pTfs, rname, diskId) < 0) {
return -1; return -1;
} }
} else { } else {
@ -278,11 +244,12 @@ int32_t tfsMkdirRecurAt(const char *rname, int32_t level, int32_t id) {
return 0; return 0;
} }
int32_t tfsMkdir(const char *rname) { int32_t tfsMkdir(STfs *pTfs, const char *rname) {
for (int32_t level = 0; level < TFS_NLEVEL(); level++) { for (int32_t level = 0; level < pTfs->nlevel; level++) {
STier *pTier = TFS_TIER_AT(level); STfsTier *pTier = TFS_TIER_AT(pTfs, level);
for (int32_t id = 0; id < TIER_NDISKS(pTier); id++) { for (int32_t id = 0; id < pTier->ndisk; id++) {
if (tfsMkdirAt(rname, level, id) < 0) { SDiskID did = {.id = id, .level = level};
if (tfsMkdirAt(pTfs, rname, did) < 0) {
return -1; return -1;
} }
} }
@ -291,16 +258,14 @@ int32_t tfsMkdir(const char *rname) {
return 0; return 0;
} }
int32_t tfsRmdir(const char *rname) { int32_t tfsRmdir(STfs *pTfs, const char *rname) {
char aname[TMPNAME_LEN] = "\0"; char aname[TMPNAME_LEN] = "\0";
for (int32_t level = 0; level < TFS_NLEVEL(); level++) { for (int32_t level = 0; level < pTfs->nlevel; level++) {
STier *pTier = TFS_TIER_AT(level); STfsTier *pTier = TFS_TIER_AT(pTfs, level);
for (int32_t id = 0; id < TIER_NDISKS(pTier); id++) { for (int32_t id = 0; id < pTier->ndisk; id++) {
SDisk *pDisk = pTier->disks[id]; STfsDisk *pDisk = pTier->disks[id];
snprintf(aname, TMPNAME_LEN, "%s%s%s", pDisk->path, TD_DIRSEP, rname);
snprintf(aname, TMPNAME_LEN, "%s%s%s", DISK_DIR(pDisk), TS_PATH_DELIMITER, rname);
taosRemoveDir(aname); taosRemoveDir(aname);
} }
} }
@ -308,117 +273,108 @@ int32_t tfsRmdir(const char *rname) {
return 0; return 0;
} }
#if 0 int32_t tfsRename(STfs *pTfs, char *orname, char *nrname) {
int32_t tfsRename(char *orname, char *nrname) {
char oaname[TMPNAME_LEN] = "\0"; char oaname[TMPNAME_LEN] = "\0";
char naname[TMPNAME_LEN] = "\0"; char naname[TMPNAME_LEN] = "\0";
for (int32_t level = 0; level < pfs->nlevel; level++) { for (int32_t level = 0; level < pTfs->nlevel; level++) {
STier *pTier = TFS_TIER_AT(level); STfsTier *pTier = TFS_TIER_AT(pTfs, level);
for (int32_t id = 0; id < TIER_NDISKS(pTier); id++) { for (int32_t id = 0; id < pTier->ndisk; id++) {
SDisk *pDisk = DISK_AT_TIER(pTier, id); STfsDisk *pDisk = pTier->disks[id];
snprintf(oaname, TMPNAME_LEN, "%s%s%s", DISK_DIR(pDisk), TD_DIRSEP, orname);
snprintf(oaname, TMPNAME_LEN, "%s/%s", DISK_DIR(pDisk), orname); snprintf(naname, TMPNAME_LEN, "%s%s%s", DISK_DIR(pDisk), TD_DIRSEP, nrname);
snprintf(naname, TMPNAME_LEN, "%s/%s", DISK_DIR(pDisk), nrname); if (taosRenameFile(oaname, naname) != 0) {
return -1;
taosRenameFile(oaname, naname); }
} }
} }
return 0; return 0;
} }
#endif
struct TDIR { STfsDir *tfsOpendir(STfs *pTfs, const char *rname) {
SDiskIter iter; STfsDir *pDir = calloc(1, sizeof(STfsDir));
int32_t level; if (pDir == NULL) {
int32_t id;
char dirname[TSDB_FILENAME_LEN];
TFILE tfile;
DIR *dir;
};
TDIR *tfsOpendir(const char *rname) {
TDIR *tdir = (TDIR *)calloc(1, sizeof(*tdir));
if (tdir == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL; return NULL;
} }
tfsInitDiskIter(&(tdir->iter)); SDiskID diskId = {.id = 0, .level = 0};
tstrncpy(tdir->dirname, rname, TSDB_FILENAME_LEN); pDir->iter->pDisk = TFS_DISK_AT(pTfs, diskId);
pDir->pTfs = pTfs;
tstrncpy(pDir->dirname, rname, TSDB_FILENAME_LEN);
if (tfsOpendirImpl(tdir) < 0) { if (tfsOpendirImpl(pTfs, pDir) < 0) {
free(tdir); free(pDir);
return NULL; return NULL;
} }
return tdir; return pDir;
} }
const TFILE *tfsReaddir(TDIR *tdir) { const STfsFile *tfsReaddir(STfsDir *pDir) {
if (tdir == NULL || tdir->dir == NULL) return NULL; if (pDir == NULL || pDir->dir == NULL) return NULL;
char bname[TMPNAME_LEN * 2] = "\0"; char bname[TMPNAME_LEN * 2] = "\0";
while (true) { while (true) {
struct dirent *dp = NULL; struct dirent *dp = NULL;
dp = readdir(tdir->dir); dp = readdir(pDir->dir);
if (dp != NULL) { if (dp != NULL) {
// Skip . and .. // Skip . and ..
if (strcmp(dp->d_name, ".") == 0 || strcmp(dp->d_name, "..") == 0) continue; if (strcmp(dp->d_name, ".") == 0 || strcmp(dp->d_name, "..") == 0) continue;
snprintf(bname, TMPNAME_LEN * 2, "%s/%s", tdir->dirname, dp->d_name); snprintf(bname, TMPNAME_LEN * 2, "%s/%s", pDir->dirname, dp->d_name);
tfsInitFile(&(tdir->tfile), tdir->level, tdir->id, bname); tfsInitFile(pDir->pTfs, &pDir->tfile, pDir->did, bname);
return &(tdir->tfile); return &pDir->tfile;
} }
if (tfsOpendirImpl(tdir) < 0) { if (tfsOpendirImpl(pDir->pTfs, pDir) < 0) {
return NULL; return NULL;
} }
if (tdir->dir == NULL) { if (pDir->dir == NULL) {
terrno = TSDB_CODE_SUCCESS; terrno = TSDB_CODE_SUCCESS;
return NULL; return NULL;
} }
} }
} }
void tfsClosedir(TDIR *tdir) { void tfsClosedir(STfsDir *pDir) {
if (tdir) { if (pDir) {
if (tdir->dir != NULL) { if (pDir->dir != NULL) {
closedir(tdir->dir); closedir(pDir->dir);
tdir->dir = NULL; pDir->dir = NULL;
} }
free(tdir); free(pDir);
} }
} }
// private static int32_t tfsMount(STfs *pTfs, SDiskCfg *pCfg) {
static int32_t tfsMount(SDiskCfg *pCfg) { if (tfsCheckAndFormatCfg(pTfs, pCfg) < 0) {
SDiskID did; return -1;
SDisk *pDisk = NULL; }
if (tfsCheckAndFormatCfg(pCfg) < 0) return -1; SDiskID did = {.level = pCfg->level};
STfsDisk *pDisk = tfsMountDiskToTier(TFS_TIER_AT(pTfs, did.level), pCfg);
did.level = pCfg->level;
pDisk = tfsMountDiskToTier(TFS_TIER_AT(did.level), pCfg);
if (pDisk == NULL) { if (pDisk == NULL) {
fError("failed to mount disk %s to level %d since %s", pCfg->dir, pCfg->level, tstrerror(terrno)); fError("failed to mount disk %s to level %d since %s", pCfg->dir, pCfg->level, terrstr());
return -1; return -1;
} }
did.id = pDisk->id; did.id = pDisk->id;
taosHashPut(pfs->map, (void *)(pCfg->dir), strnlen(pCfg->dir, TSDB_FILENAME_LEN), (void *)(&did), sizeof(did)); taosHashPut(pTfs->hash, (void *)(pCfg->dir), strnlen(pCfg->dir, TSDB_FILENAME_LEN), (void *)(&did), sizeof(did));
if (pfs->nlevel < pCfg->level + 1) pfs->nlevel = pCfg->level + 1; if (pTfs->nlevel < pCfg->level + 1) {
pTfs->nlevel = pCfg->level + 1;
}
return 0; return 0;
} }
static int32_t tfsCheckAndFormatCfg(SDiskCfg *pCfg) { static int32_t tfsCheckAndFormatCfg(STfs *pTfs, SDiskCfg *pCfg) {
char dirName[TSDB_FILENAME_LEN] = "\0"; char dirName[TSDB_FILENAME_LEN] = "\0";
struct stat pstat; struct stat pstat;
if (pCfg->level < 0 || pCfg->level >= TSDB_MAX_TIERS) { if (pCfg->level < 0 || pCfg->level >= TFS_MAX_TIERS) {
fError("failed to mount %s to FS since invalid level %d", pCfg->dir, pCfg->level); fError("failed to mount %s to FS since invalid level %d", pCfg->dir, pCfg->level);
terrno = TSDB_CODE_FS_INVLD_CFG; terrno = TSDB_CODE_FS_INVLD_CFG;
return -1; return -1;
@ -431,7 +387,7 @@ static int32_t tfsCheckAndFormatCfg(SDiskCfg *pCfg) {
return -1; return -1;
} }
if (TFS_PRIMARY_DISK() != NULL) { if (TFS_PRIMARY_DISK(pTfs) != NULL) {
fError("failed to mount %s to FS since duplicate primary mount", pCfg->dir); fError("failed to mount %s to FS since duplicate primary mount", pCfg->dir);
terrno = TSDB_CODE_FS_DUP_PRIMARY; terrno = TSDB_CODE_FS_DUP_PRIMARY;
return -1; return -1;
@ -444,7 +400,7 @@ static int32_t tfsCheckAndFormatCfg(SDiskCfg *pCfg) {
return -1; return -1;
} }
if (tfsGetDiskByName(dirName) != NULL) { if (tfsGetDiskByName(pTfs, dirName) != NULL) {
fError("failed to mount %s to FS since duplicate mount", pCfg->dir); fError("failed to mount %s to FS since duplicate mount", pCfg->dir);
terrno = TSDB_CODE_FS_INVLD_CFG; terrno = TSDB_CODE_FS_INVLD_CFG;
return -1; return -1;
@ -494,15 +450,15 @@ static int32_t tfsFormatDir(char *idir, char *odir) {
return 0; return 0;
} }
static int32_t tfsCheck() { static int32_t tfsCheck(STfs *pTfs) {
if (TFS_PRIMARY_DISK() == NULL) { if (TFS_PRIMARY_DISK(pTfs) == NULL) {
fError("no primary disk is set"); fError("no primary disk is set");
terrno = TSDB_CODE_FS_NO_PRIMARY_DISK; terrno = TSDB_CODE_FS_NO_PRIMARY_DISK;
return -1; return -1;
} }
for (int32_t level = 0; level < TFS_NLEVEL(); level++) { for (int32_t level = 0; level < pTfs->nlevel; level++) {
if (TIER_NDISKS(TFS_TIER_AT(level)) == 0) { if (TFS_TIER_AT(pTfs, level)->ndisk == 0) {
fError("no disk at level %d", level); fError("no disk at level %d", level);
terrno = TSDB_CODE_FS_NO_MOUNT_AT_TIER; terrno = TSDB_CODE_FS_NO_MOUNT_AT_TIER;
return -1; return -1;
@ -512,66 +468,53 @@ static int32_t tfsCheck() {
return 0; return 0;
} }
static SDisk *tfsGetDiskByID(SDiskID did) { return TFS_DISK_AT(did.level, did.id); } static STfsDisk *tfsGetDiskByName(STfs *pTfs, const char *dir) {
static SDisk *tfsGetDiskByName(const char *dir) { void *pr = taosHashGet(pTfs->hash, (void *)dir, strnlen(dir, TSDB_FILENAME_LEN));
SDiskID did;
SDisk *pDisk = NULL;
void *pr = NULL;
pr = taosHashGet(pfs->map, (void *)dir, strnlen(dir, TSDB_FILENAME_LEN));
if (pr == NULL) return NULL; if (pr == NULL) return NULL;
did = *(SDiskID *)pr; SDiskID did = *(SDiskID *)pr;
pDisk = tfsGetDiskByID(did); STfsDisk *pDisk = TFS_DISK_AT(pTfs, did);
ASSERT(pDisk != NULL);
return pDisk; return pDisk;
} }
static int32_t tfsOpendirImpl(TDIR *tdir) { static int32_t tfsOpendirImpl(STfs *pTfs, STfsDir *pDir) {
SDisk *pDisk = NULL; STfsDisk *pDisk = NULL;
char adir[TMPNAME_LEN * 2] = "\0"; char adir[TMPNAME_LEN * 2] = "\0";
if (tdir->dir != NULL) { if (pDir->dir != NULL) {
closedir(tdir->dir); closedir(pDir->dir);
tdir->dir = NULL; pDir->dir = NULL;
} }
while (true) { while (true) {
pDisk = tfsNextDisk(&(tdir->iter)); pDisk = tfsNextDisk(pTfs, pDir->iter);
if (pDisk == NULL) return 0; if (pDisk == NULL) return 0;
tdir->level = pDisk->level; pDir->did.level = pDisk->level;
tdir->id = pDisk->id; pDir->did.id = pDisk->id;
snprintf(adir, TMPNAME_LEN * 2, "%s%s%s", pDisk->path, TS_PATH_DELIMITER,tdir->dirname); snprintf(adir, TMPNAME_LEN * 2, "%s%s%s", pDisk->path, TD_DIRSEP, pDir->dirname);
tdir->dir = opendir(adir); pDir->dir = opendir(adir);
if (tdir->dir != NULL) break; if (pDir->dir != NULL) break;
} }
return 0; return 0;
} }
static void tfsInitDiskIter(SDiskIter *pIter) { pIter->pDisk = TFS_DISK_AT(0, 0); } static STfsDisk *tfsNextDisk(STfs *pTfs, SDiskIter *pIter) {
if (pIter == NULL) return NULL;
static SDisk *tfsNextDisk(SDiskIter *pIter) { STfsDisk *pDisk = pIter->pDisk;
SDisk *pDisk = pIter->pDisk; SDiskID did = {.level = pDisk->level, .id = pDisk->id + 1};
if (pDisk == NULL) return NULL; if (did.id < TFS_TIER_AT(pTfs, did.level)->ndisk) {
pIter->pDisk = TFS_DISK_AT(pTfs, did);
int32_t level = pDisk->level;
int32_t id = pDisk->id;
id++;
if (id < TIER_NDISKS(TFS_TIER_AT(level))) {
pIter->pDisk = TFS_DISK_AT(level, id);
ASSERT(pIter->pDisk != NULL);
} else { } else {
level++; did.level++;
id = 0; did.id = 0;
if (level < TFS_NLEVEL()) { if (did.level < pTfs->nlevel) {
pIter->pDisk = TFS_DISK_AT(level, id); pIter->pDisk = TFS_DISK_AT(pTfs, did);
ASSERT(pIter->pDisk != NULL);
} else { } else {
pIter->pDisk = NULL; pIter->pDisk = NULL;
} }
@ -579,25 +522,3 @@ static SDisk *tfsNextDisk(SDiskIter *pIter) {
return pDisk; return pDisk;
} }
// OTHER FUNCTIONS ===================================
void taosGetDisk() {
const double unit = 1024 * 1024 * 1024;
SDiskSize diskSize;
SFSMeta fsMeta;
tfsUpdateSize(&fsMeta);
tsTotalDataDirGB = (float)(fsMeta.total / unit);
tsUsedDataDirGB = (float)(fsMeta.used / unit);
tsAvailDataDirGB = (float)(fsMeta.avail / unit);
if (taosGetDiskSize(tsLogDir, &diskSize) == 0) {
tsTotalLogDirGB = (float)(diskSize.total / unit);
tsAvailLogDirGB = (float)(diskSize.avail / unit);
}
if (taosGetDiskSize(tsTempDir, &diskSize) == 0) {
tsTotalTmpDirGB = (float)(diskSize.total / unit);
tsAvailTmpDirectorySpace = (float)(diskSize.avail / unit);
}
}

8
source/libs/tfs/src/tfsDisk.c
View File

@ -16,8 +16,8 @@
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE
#include "tfsInt.h" #include "tfsInt.h"
SDisk *tfsNewDisk(int32_t level, int32_t id, const char *path) { STfsDisk *tfsNewDisk(int32_t level, int32_t id, const char *path) {
SDisk *pDisk = calloc(1, sizeof(SDisk)); STfsDisk *pDisk = calloc(1, sizeof(STfsDisk));
if (pDisk == NULL) { if (pDisk == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL; return NULL;
@ -36,7 +36,7 @@ SDisk *tfsNewDisk(int32_t level, int32_t id, const char *path) {
return pDisk; return pDisk;
} }
SDisk *tfsFreeDisk(SDisk *pDisk) { STfsDisk *tfsFreeDisk(STfsDisk *pDisk) {
if (pDisk != NULL) { if (pDisk != NULL) {
free(pDisk->path); free(pDisk->path);
free(pDisk); free(pDisk);
@ -45,7 +45,7 @@ SDisk *tfsFreeDisk(SDisk *pDisk) {
return NULL; return NULL;
} }
int32_t tfsUpdateDiskSize(SDisk *pDisk) { int32_t tfsUpdateDiskSize(STfsDisk *pDisk) {
if (taosGetDiskSize(pDisk->path, &pDisk->size) != 0) { if (taosGetDiskSize(pDisk->path, &pDisk->size) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno); terrno = TAOS_SYSTEM_ERROR(errno);
fError("failed to get disk:%s size, level:%d id:%d since %s", pDisk->path, pDisk->level, pDisk->id, terrstr()); fError("failed to get disk:%s size, level:%d id:%d since %s", pDisk->path, pDisk->level, pDisk->id, terrstr());

39
source/libs/tfs/src/tfsTier.c
View File

@ -16,11 +16,8 @@
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE
#include "tfsInt.h" #include "tfsInt.h"
#define tfsLockTier(pTier) pthread_spin_lock(&(pTier)->lock) int32_t tfsInitTier(STfsTier *pTier, int32_t level) {
#define tfsUnLockTier(pTier) pthread_spin_unlock(&(pTier)->lock) memset(pTier, 0, sizeof(STfsTier));
int32_t tfsInitTier(STier *pTier, int32_t level) {
memset(pTier, 0, sizeof(STier));
if (pthread_spin_init(&pTier->lock, 0) != 0) { if (pthread_spin_init(&pTier->lock, 0) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno); terrno = TAOS_SYSTEM_ERROR(errno);
@ -31,17 +28,17 @@ int32_t tfsInitTier(STier *pTier, int32_t level) {
return 0; return 0;
} }
void tfsDestroyTier(STier *pTier) { void tfsDestroyTier(STfsTier *pTier) {
for (int32_t id = 0; id < TSDB_MAX_DISKS_PER_TIER; id++) { for (int32_t id = 0; id < TFS_MAX_DISKS_PER_TIER; id++) {
pTier->disks[id] = tfsFreeDisk(pTier->disks[id]); pTier->disks[id] = tfsFreeDisk(pTier->disks[id]);
} }
pTier->ndisk = 0; pTier->ndisk = 0;
pthread_spin_destroy(&(pTier->lock)); pthread_spin_destroy(&pTier->lock);
} }
SDisk *tfsMountDiskToTier(STier *pTier, SDiskCfg *pCfg) { STfsDisk *tfsMountDiskToTier(STfsTier *pTier, SDiskCfg *pCfg) {
if (pTier->ndisk >= TSDB_MAX_DISKS_PER_TIER) { if (pTier->ndisk >= TFS_MAX_DISKS_PER_TIER) {
terrno = TSDB_CODE_FS_TOO_MANY_MOUNT; terrno = TSDB_CODE_FS_TOO_MANY_MOUNT;
return NULL; return NULL;
} }
@ -61,12 +58,12 @@ SDisk *tfsMountDiskToTier(STier *pTier, SDiskCfg *pCfg) {
id = pTier->ndisk; id = pTier->ndisk;
} }
if (id >= TSDB_MAX_DISKS_PER_TIER) { if (id >= TFS_MAX_DISKS_PER_TIER) {
terrno = TSDB_CODE_FS_TOO_MANY_MOUNT; terrno = TSDB_CODE_FS_TOO_MANY_MOUNT;
return NULL; return NULL;
} }
SDisk *pDisk = tfsNewDisk(pCfg->level, id, pCfg->dir); STfsDisk *pDisk = tfsNewDisk(pCfg->level, id, pCfg->dir);
if (pDisk == NULL) return NULL; if (pDisk == NULL) return NULL;
pTier->disks[id] = pDisk; pTier->disks[id] = pDisk;
@ -76,14 +73,14 @@ SDisk *tfsMountDiskToTier(STier *pTier, SDiskCfg *pCfg) {
return pTier->disks[id]; return pTier->disks[id];
} }
void tfsUpdateTierSize(STier *pTier) { void tfsUpdateTierSize(STfsTier *pTier) {
SDiskSize size = {0}; SDiskSize size = {0};
int16_t nAvailDisks = 0; int32_t nAvailDisks = 0;
tfsLockTier(pTier); tfsLockTier(pTier);
for (int32_t id = 0; id < pTier->ndisk; id++) { for (int32_t id = 0; id < pTier->ndisk; id++) {
SDisk *pDisk = pTier->disks[id]; STfsDisk *pDisk = pTier->disks[id];
if (pDisk == NULL) continue; if (pDisk == NULL) continue;
size.total += pDisk->size.total; size.total += pDisk->size.total;
@ -99,7 +96,7 @@ void tfsUpdateTierSize(STier *pTier) {
} }
// Round-Robin to allocate disk on a tier // Round-Robin to allocate disk on a tier
int32_t tfsAllocDiskOnTier(STier *pTier) { int32_t tfsAllocDiskOnTier(STfsTier *pTier) {
terrno = TSDB_CODE_FS_NO_VALID_DISK; terrno = TSDB_CODE_FS_NO_VALID_DISK;
tfsLockTier(pTier); tfsLockTier(pTier);
@ -110,9 +107,9 @@ int32_t tfsAllocDiskOnTier(STier *pTier) {
} }
int32_t retId = -1; int32_t retId = -1;
for (int32_t id = 0; id < TSDB_MAX_DISKS_PER_TIER; ++id) { for (int32_t id = 0; id < TFS_MAX_DISKS_PER_TIER; ++id) {
int32_t diskId = (pTier->nextid + id) % pTier->ndisk; int32_t diskId = (pTier->nextid + id) % pTier->ndisk;
SDisk *pDisk = pTier->disks[diskId]; STfsDisk *pDisk = pTier->disks[diskId];
if (pDisk == NULL) continue; if (pDisk == NULL) continue;
@ -128,12 +125,12 @@ int32_t tfsAllocDiskOnTier(STier *pTier) {
return retId; return retId;
} }
void tfsPosNextId(STier *pTier) { void tfsPosNextId(STfsTier *pTier) {
int32_t nextid = 0; int32_t nextid = 0;
for (int32_t id = 1; id < pTier->ndisk; id++) { for (int32_t id = 1; id < pTier->ndisk; id++) {
SDisk *pLDisk = pTier->disks[nextid]; STfsDisk *pLDisk = pTier->disks[nextid];
SDisk *pDisk = pTier->disks[id]; STfsDisk *pDisk = pTier->disks[id];
if (pDisk->size.avail > TFS_MIN_DISK_FREE_SIZE && pDisk->size.avail > pLDisk->size.avail) { if (pDisk->size.avail > TFS_MIN_DISK_FREE_SIZE && pDisk->size.avail > pLDisk->size.avail) {
nextid = id; nextid = id;
} }