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

refact tier and disk in tfs module

This commit is contained in:
Shengliang Guan 2022-01-17 23:20:11 -08:00
parent f7c0c4745f
commit d6359f3ab1
7 changed files with 109 additions and 182 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -36,22 +36,14 @@ typedef struct {
// FS APIs ====================================
typedef struct {
int64_t tsize;
int64_t total;
int64_t used;
int64_t avail;
} SFSMeta;
typedef struct {
int64_t size;
int64_t used;
int64_t free;
int16_t nAvailDisks; // # of Available disks
} STierMeta;
int32_t tfsInit(SDiskCfg *pDiskCfg, int32_t ndisk);
void tfsCleanup();
void tfsUpdateInfo(SFSMeta *pFSMeta, STierMeta *tierMetas, int8_t numLevels);
void tfsGetMeta(SFSMeta *pMeta);
void tfsUpdateSize(SFSMeta *pFSMeta);
void tfsAllocDisk(int32_t expLevel, int32_t *level, int32_t *id);
const char *TFS_PRIMARY_PATH();

6
include/os/osSysinfo.h
View File

@ -35,12 +35,12 @@ extern char tsLocale[];
extern char tsCharset[]; // default encode string
typedef struct {
int64_t tsize;
int64_t total;
int64_t used;
int64_t avail;
} SysDiskSize;
} SDiskSize;
int32_t taosGetDiskSize(char *dataDir, SysDiskSize *diskSize);
int32_t taosGetDiskSize(char *dataDir, SDiskSize *diskSize);
int32_t taosGetCpuCores();
void taosGetSystemInfo();
bool taosReadProcIO(int64_t *rchars, int64_t *wchars);

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

@ -13,8 +13,8 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef _TD_TFSINT_H_
#define _TD_TFSINT_H_
#ifndef _TD_TFS_INT_H_
#define _TD_TFS_INT_H_
#include "os.h"
@ -39,54 +39,39 @@ extern int32_t fsDebugFlag;
// Global Definitions
#define TFS_MIN_DISK_FREE_SIZE 50 * 1024 * 1024
// tdisk.c ======================================================
typedef struct {
int64_t size;
int64_t used;
int64_t free;
} SDiskMeta;
typedef struct SDisk {
int32_t level;
int32_t id;
char dir[TSDB_FILENAME_LEN];
SDiskMeta dmeta;
char *path;
SDiskSize size;
} SDisk;
#define DISK_LEVEL(pd) ((pd)->level)
#define DISK_ID(pd) ((pd)->id)
#define DISK_DIR(pd) ((pd)->dir)
#define DISK_META(pd) ((pd)->dmeta)
#define DISK_SIZE(pd) ((pd)->dmeta.size)
#define DISK_USED_SIZE(pd) ((pd)->dmeta.used)
#define DISK_FREE_SIZE(pd) ((pd)->dmeta.free)
SDisk *tfsNewDisk(int32_t level, int32_t id, const char *dir);
SDisk *tfsFreeDisk(SDisk *pDisk);
int32_t tfsUpdateDiskInfo(SDisk *pDisk);
// ttier.c ======================================================
typedef struct STier {
pthread_spinlock_t lock;
int32_t level;
int16_t ndisk; // # of disks mounted to this tier
int16_t nextid; // next disk id to allocate
STierMeta tmeta;
int16_t nextid; // next disk id to allocate
int16_t ndisk; // # of disks mounted to this tier
int16_t nAvailDisks; // # of Available disks
SDisk *disks[TSDB_MAX_DISKS_PER_TIER];
SDiskSize size;
} STier;
#define TIER_LEVEL(pt) ((pt)->level)
#define TIER_NDISKS(pt) ((pt)->ndisk)
#define TIER_SIZE(pt) ((pt)->tmeta.size)
#define TIER_FREE_SIZE(pt) ((pt)->tmeta.free)
#define TIER_AVAIL_DISKS(pt) ((pt)->tmeta.nAvailDisks)
#define DISK_AT_TIER(pt, id) ((pt)->disks[id])
#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 tfsUpdateTierInfo(STier *pTier, STierMeta *pTierMeta);
void tfsUpdateTierSize(STier *pTier);
int32_t tfsAllocDiskOnTier(STier *pTier);
void tfsPosNextId(STier *pTier);
@ -94,4 +79,4 @@ void tfsPosNextId(STier *pTier);
}
#endif
#endif /*_TD_TFSINT_H_*/
#endif /*_TD_TFS_INT_H_*/

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

@ -59,7 +59,10 @@ static SDisk *tfsNextDisk(SDiskIter *pIter);
// FS APIs ====================================
int32_t tfsInit(SDiskCfg *pDiskCfg, int32_t ndisk) {
ASSERT(ndisk > 0);
if (ndisk < 0) {
terrno = TSDB_CODE_INVALID_PARA;
return -1;
}
for (int32_t level = 0; level < TSDB_MAX_TIERS; level++) {
if (tfsInitTier(TFS_TIER_AT(level), level) < 0) {
@ -96,7 +99,7 @@ int32_t tfsInit(SDiskCfg *pDiskCfg, int32_t ndisk) {
return -1;
}
tfsUpdateInfo(NULL, NULL, 0);
tfsUpdateSize(NULL);
for (int32_t level = 0; level < TFS_NLEVEL(); level++) {
tfsPosNextId(TFS_TIER_AT(level));
}
@ -114,27 +117,22 @@ void tfsCleanup() {
}
}
void tfsUpdateInfo(SFSMeta *pFSMeta, STierMeta *tierMetas, int8_t numTiers) {
SFSMeta fsMeta;
STierMeta tierMeta;
void tfsUpdateSize(SFSMeta *pFSMeta) {
SFSMeta fsMeta = {0};
SDiskSize size = {0};
if (pFSMeta == NULL) {
pFSMeta = &fsMeta;
}
memset(pFSMeta, 0, sizeof(*pFSMeta));
memset(pFSMeta, 0, sizeof(SFSMeta));
for (int32_t level = 0; level < TFS_NLEVEL(); level++) {
STierMeta *pTierMeta = &tierMeta;
if (tierMetas && level < numTiers) {
pTierMeta = tierMetas + level;
}
STier *pTier = TFS_TIER_AT(level);
tfsUpdateTierInfo(pTier, pTierMeta);
pFSMeta->tsize += pTierMeta->size;
pFSMeta->avail += pTierMeta->free;
pFSMeta->used += pTierMeta->used;
tfsUpdateTierSize(pTier);
pFSMeta->total += pTier->size.total;
pFSMeta->avail += pTier->size.avail;
pFSMeta->used += pTier->size.used;
}
tfsLock();
@ -142,14 +140,6 @@ void tfsUpdateInfo(SFSMeta *pFSMeta, STierMeta *tierMetas, int8_t numTiers) {
tfsUnLock();
}
void tfsGetMeta(SFSMeta *pMeta) {
ASSERT(pMeta);
tfsLock();
*pMeta = pfs->meta;
tfsUnLock();
}
/* Allocate an existing available tier level
*/
void tfsAllocDisk(int32_t expLevel, int32_t *level, int32_t *id) {
@ -307,9 +297,9 @@ int32_t tfsRmdir(const char *rname) {
for (int32_t level = 0; level < TFS_NLEVEL(); level++) {
STier *pTier = TFS_TIER_AT(level);
for (int32_t id = 0; id < TIER_NDISKS(pTier); id++) {
SDisk *pDisk = DISK_AT_TIER(pTier, id);
SDisk *pDisk = pTier->disks[id];
snprintf(aname, TMPNAME_LEN, "%s/%s", DISK_DIR(pDisk), rname);
snprintf(aname, TMPNAME_LEN, "%s%s%s", DISK_DIR(pDisk), TS_PATH_DELIMITER, rname);
taosRemoveDir(aname);
}
@ -351,7 +341,7 @@ struct TDIR {
TDIR *tfsOpendir(const char *rname) {
TDIR *tdir = (TDIR *)calloc(1, sizeof(*tdir));
if (tdir == NULL) {
terrno = TSDB_CODE_FS_OUT_OF_MEMORY;
terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
}
@ -416,7 +406,7 @@ static int32_t tfsMount(SDiskCfg *pCfg) {
fError("failed to mount disk %s to level %d since %s", pCfg->dir, pCfg->level, tstrerror(terrno));
return -1;
}
did.id = DISK_ID(pDisk);
did.id = pDisk->id;
taosHashPut(pfs->map, (void *)(pCfg->dir), strnlen(pCfg->dir, TSDB_FILENAME_LEN), (void *)(&did), sizeof(did));
if (pfs->nlevel < pCfg->level + 1) pfs->nlevel = pCfg->level + 1;
@ -551,10 +541,10 @@ static int32_t tfsOpendirImpl(TDIR *tdir) {
pDisk = tfsNextDisk(&(tdir->iter));
if (pDisk == NULL) return 0;
tdir->level = DISK_LEVEL(pDisk);
tdir->id = DISK_ID(pDisk);
tdir->level = pDisk->level;
tdir->id = pDisk->id;
snprintf(adir, TMPNAME_LEN * 2, "%s/%s", DISK_DIR(pDisk), tdir->dirname);
snprintf(adir, TMPNAME_LEN * 2, "%s%s%s", pDisk->path, TS_PATH_DELIMITER,tdir->dirname);
tdir->dir = opendir(adir);
if (tdir->dir != NULL) break;
}
@ -569,8 +559,8 @@ static SDisk *tfsNextDisk(SDiskIter *pIter) {
if (pDisk == NULL) return NULL;
int32_t level = DISK_LEVEL(pDisk);
int32_t id = DISK_ID(pDisk);
int32_t level = pDisk->level;
int32_t id = pDisk->id;
id++;
if (id < TIER_NDISKS(TFS_TIER_AT(level))) {
@ -593,21 +583,21 @@ static SDisk *tfsNextDisk(SDiskIter *pIter) {
// OTHER FUNCTIONS ===================================
void taosGetDisk() {
const double unit = 1024 * 1024 * 1024;
SysDiskSize diskSize;
SDiskSize diskSize;
SFSMeta fsMeta;
tfsUpdateInfo(&fsMeta, NULL, 0);
tsTotalDataDirGB = (float)(fsMeta.tsize / unit);
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.tsize / unit);
tsTotalLogDirGB = (float)(diskSize.total / unit);
tsAvailLogDirGB = (float)(diskSize.avail / unit);
}
if (taosGetDiskSize(tsTempDir, &diskSize) == 0) {
tsTotalTmpDirGB = (float)(diskSize.tsize / unit);
tsTotalTmpDirGB = (float)(diskSize.total / unit);
tsAvailTmpDirectorySpace = (float)(diskSize.avail / unit);
}
}

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

@ -16,44 +16,41 @@
#define _DEFAULT_SOURCE
#include "tfsInt.h"
SDisk *tfsNewDisk(int32_t level, int32_t id, const char *dir) {
SDisk *tfsNewDisk(int32_t level, int32_t id, const char *path) {
SDisk *pDisk = calloc(1, sizeof(SDisk));
if (pDisk == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
}
pDisk->path = strdup(path);
if (pDisk->path == NULL) {
free(pDisk);
terrno = TSDB_CODE_OUT_OF_MEMORY;
return NULL;
}
pDisk->level = level;
pDisk->id = id;
tstrncpy(pDisk->dir, dir, TSDB_FILENAME_LEN);
taosGetDiskSize(pDisk->path, &pDisk->size);
return pDisk;
}
SDisk *tfsFreeDisk(SDisk *pDisk) {
if (pDisk != NULL) {
free(pDisk->path);
free(pDisk);
}
return NULL;
}
int32_t tfsUpdateDiskInfo(SDisk *pDisk) {
if (pDisk == NULL) {
terrno = TSDB_CODE_INVALID_PARA;
int32_t tfsUpdateDiskSize(SDisk *pDisk) {
if (taosGetDiskSize(pDisk->path, &pDisk->size) != 0) {
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());
return -1;
}
SysDiskSize diskSize = {0};
if (taosGetDiskSize(pDisk->dir, &diskSize) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
fError("failed to update disk information at level %d id %d dir %s since %s", pDisk->level, pDisk->id, pDisk->dir,
strerror(errno));
return -1
}
pDisk->dmeta.size = diskSize.tsize;
pDisk->dmeta.used = diskSize.used;
pDisk->dmeta.free = diskSize.avail;
return 0;
}

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

@ -20,16 +20,10 @@
#define tfsUnLockTier(pTier) pthread_spin_unlock(&(pTier)->lock)
int32_t tfsInitTier(STier *pTier, int32_t level) {
if (pTier == NULL) {
terrno = TSDB_CODE_INVALID_PARA;
return -1;
}
memset(pTier, 0, sizeof(STier));
int32_t code = pthread_spin_init(&pTier->lock, 0);
if (code != 0) {
terrno = TAOS_SYSTEM_ERROR(code);
if (pthread_spin_init(&pTier->lock, 0) != 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
}
@ -38,10 +32,8 @@ int32_t tfsInitTier(STier *pTier, int32_t level) {
}
void tfsDestroyTier(STier *pTier) {
if (pTier == NULL) return;
for (int32_t id = 0; id < TSDB_MAX_DISKS_PER_TIER; id++) {
DISK_AT_TIER(pTier, id) = tfsFreeDisk(DISK_AT_TIER(pTier, id));
pTier->disks[id] = tfsFreeDisk(pTier->disks[id]);
}
pTier->ndisk = 0;
@ -49,19 +41,14 @@ void tfsDestroyTier(STier *pTier) {
}
SDisk *tfsMountDiskToTier(STier *pTier, SDiskCfg *pCfg) {
if (pTier == NULL || pCfg == NULL || pTier->level != pCfg->level) {
terrno = TSDB_CODE_INVALID_PARA;
return -1;
}
if (TIER_NDISKS(pTier) >= TSDB_MAX_DISKS_PER_TIER) {
if (pTier->ndisk >= TSDB_MAX_DISKS_PER_TIER) {
terrno = TSDB_CODE_FS_TOO_MANY_MOUNT;
return NULL;
}
int32_t id = 0;
if (pTier->level == 0) {
if (DISK_AT_TIER(pTier, 0) != NULL) {
if (pTier->disks[0] != NULL) {
id = pTier->ndisk;
} else {
if (pCfg->primary) {
@ -69,108 +56,85 @@ SDisk *tfsMountDiskToTier(STier *pTier, SDiskCfg *pCfg) {
} else {
id = pTier->ndisk + 1;
}
if (id >= TSDB_MAX_DISKS_PER_TIER) {
terrno = TSDB_CODE_FS_TOO_MANY_MOUNT;
return NULL;
}
}
} else {
id = pTier->ndisk;
}
if (id >= TSDB_MAX_DISKS_PER_TIER) {
terrno = TSDB_CODE_FS_TOO_MANY_MOUNT;
return NULL;
}
SDisk *pDisk = tfsNewDisk(pCfg->level, id, pCfg->dir);
if (pDisk == NULL) return NULL;
DISK_AT_TIER(pTier, id) = pDisk;
pTier->disks[id] = pDisk;
pTier->ndisk++;
fInfo("disk %s is mounted to tier level %d id %d", pCfg->dir, pCfg->level, id);
return DISK_AT_TIER(pTier, id);
return pTier->disks[id];
}
void tfsUpdateTierInfo(STier *pTier, STierMeta *pTierMeta) {
STierMeta tmeta = {0};
if (pTierMeta == NULL) {
pTierMeta = &tmeta;
}
memset(pTierMeta, 0, sizeof(STierMeta));
void tfsUpdateTierSize(STier *pTier) {
SDiskSize size = {0};
int16_t nAvailDisks = 0;
tfsLockTier(pTier);
for (int32_t id = 0; id < pTier->ndisk; id++) {
if (tfsUpdateDiskInfo(DISK_AT_TIER(pTier, id)) != 0) {
continue;
}
SDisk *pDisk = pTier->disks[id];
if (pDisk == NULL) continue;
pTierMeta->size += DISK_SIZE(DISK_AT_TIER(pTier, id));
pTierMeta->used += DISK_USED_SIZE(DISK_AT_TIER(pTier, id));
pTierMeta->free += DISK_FREE_SIZE(DISK_AT_TIER(pTier, id));
pTierMeta->nAvailDisks++;
size.total += pDisk->size.total;
size.used += pDisk->size.used;
size.avail += pDisk->size.avail;
nAvailDisks++;
}
pTier->tmeta = *pTierMeta;
pTier->size = size;
pTier->nAvailDisks = nAvailDisks;
tfsUnLockTier(pTier);
}
// Round-Robin to allocate disk on a tier
int32_t tfsAllocDiskOnTier(STier *pTier) {
if (pTier == NULL || pTier->ndisk <= 0) {
terrno = TSDB_CODE_INVALID_PARA;
terrno = TSDB_CODE_FS_NO_VALID_DISK;
tfsLockTier(pTier);
if (pTier->ndisk <= 0 || pTier->nAvailDisks <= 0) {
tfsUnLockTier(pTier);
return -1;
}
tfsLockTier(pTier);
int32_t retId = -1;
for (int32_t id = 0; id < TSDB_MAX_DISKS_PER_TIER; ++id) {
int32_t diskId = (pTier->nextid + id) % pTier->ndisk;
SDisk *pDisk = pTier->disks[diskId];
if (TIER_AVAIL_DISKS(pTier) <= 0) {
tfsUnLockTier(pTier);
return TFS_UNDECIDED_ID;
}
if (pDisk == NULL) continue;
int32_t id = pTier->nextid;
while (true) {
SDisk *pDisk = DISK_AT_TIER(pTier, id);
if (pDisk == NULL) {
tfsUnLockTier(pTier);
return TFS_UNDECIDED_ID;
}
if (pDisk->size.avail < TFS_MIN_DISK_FREE_SIZE) continue;
if (DISK_FREE_SIZE(pDisk) < TFS_MIN_DISK_FREE_SIZE) {
id = (id + 1) % pTier->ndisk;
if (id == pTier->nextid) {
tfsUnLockTier(pTier);
return TFS_UNDECIDED_ID;
} else {
continue;
}
} else {
pTier->nextid = (id + 1) % pTier->ndisk;
break;
}
retId = diskId;
terrno = 0;
pTier->nextid = (diskId + 1) % pTier->ndisk;
break;
}
tfsUnLockTier(pTier);
return id;
}
void tfsGetTierMeta(STier *pTier, STierMeta *pTierMeta) {
if (pTierMeta == NULL || pTierMeta == NULL) return;
tfsLockTier(pTier);
*pTierMeta = pTier->tmeta;
tfsUnLockTier(pTier);
return retId;
}
void tfsPosNextId(STier *pTier) {
if (pTier == NULL || pTier->ndisk <= 0) return;
int32_t nextid = 0;
for (int32_t id = 1; id < pTier->ndisk; id++) {
SDisk *pLDisk = DISK_AT_TIER(pTier, nextid);
SDisk *pDisk = DISK_AT_TIER(pTier, id);
if (DISK_FREE_SIZE(pDisk) > TFS_MIN_DISK_FREE_SIZE && DISK_FREE_SIZE(pDisk) > DISK_FREE_SIZE(pLDisk)) {
SDisk *pLDisk = pTier->disks[nextid];
SDisk *pDisk = pTier->disks[id];
if (pDisk->size.avail > TFS_MIN_DISK_FREE_SIZE && pDisk->size.avail > pLDisk->size.avail) {
nextid = id;
}
}

9
source/os/src/osSysinfo.c
View File

@ -121,7 +121,7 @@ bool taosGetCpuUsage(float *sysCpuUsage, float *procCpuUsage) {
return true;
}
int32_t taosGetDiskSize(char *dataDir, SysDiskSize *diskSize) {
int32_t taosGetDiskSize(char *dataDir, SDiskSize *diskSize) {
unsigned _int64 i64FreeBytesToCaller;
unsigned _int64 i64TotalBytes;
unsigned _int64 i64FreeBytes;
@ -438,7 +438,7 @@ int taosSystem(const char *cmd) {
void taosSetCoreDump() {}
int32_t taosGetDiskSize(char *dataDir, SysDiskSize *diskSize) {
int32_t taosGetDiskSize(char *dataDir, SDiskSize *diskSize) {
struct statvfs info;
if (statvfs(dataDir, &info)) {
//printf("failed to get disk size, dataDir:%s errno:%s", tsDataDir, strerror(errno));
@ -771,13 +771,12 @@ bool taosGetCpuUsage(float *sysCpuUsage, float *procCpuUsage) {
return true;
}
int32_t taosGetDiskSize(char *dataDir, SysDiskSize *diskSize) {
int32_t taosGetDiskSize(char *dataDir, SDiskSize *diskSize) {
struct statvfs info;
if (statvfs(dataDir, &info)) {
//printf("failed to get disk size, dataDir:%s errno:%s", dataDir, strerror(errno));
return -1;
} else {
diskSize->tsize = info.f_blocks * info.f_frsize;
diskSize->total = info.f_blocks * info.f_frsize;
diskSize->avail = info.f_bavail * info.f_frsize;
diskSize->used = (info.f_blocks - info.f_bfree) * info.f_frsize;
return 0;