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

more code

This commit is contained in:
Hongze Cheng 2023-05-19 15:16:01 +08:00
parent e247870548
commit 1c4abc62c3
8 changed files with 215 additions and 281 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
include/util/tarray2.h
View File

@ -36,6 +36,8 @@ extern "C" {
TYPE *data; \ TYPE *data; \
} }
typedef void (*TArray2Cb)(void *);
#define TARRAY2_MIN_SIZE 16 #define TARRAY2_MIN_SIZE 16
#define TARRAY2_INITIALIZER \ #define TARRAY2_INITIALIZER \
@ -77,7 +79,7 @@ static FORCE_INLINE int32_t tarray2_make_room(void *arg, // array
#define TARRAY2_CLEAR(a, cb) \ #define TARRAY2_CLEAR(a, cb) \
do { \ do { \
if (cb) { \ if (cb) { \
void (*cb_)(void *) = (cb); \ TArray2Cb cb_ = (TArray2Cb)(cb); \
for (int32_t i = 0; i < (a)->size; ++i) { \ for (int32_t i = 0; i < (a)->size; ++i) { \
cb_((a)->data + i); \ cb_((a)->data + i); \
} \ } \
@ -98,7 +100,7 @@ static FORCE_INLINE int32_t tarray2_make_room(void *arg, // array
({ \ ({ \
int32_t __ret = 0; \ int32_t __ret = 0; \
if ((a)->size >= (a)->capacity) { \ if ((a)->size >= (a)->capacity) { \
__ret = tarray2_make_room(&(a), (a)->size + 1, sizeof(typeof((a)->data[0]))); \ __ret = tarray2_make_room((a), (a)->size + 1, sizeof(typeof((a)->data[0]))); \
} \ } \
if (!__ret) { \ if (!__ret) { \
if ((a)->size > (idx)) { \ if ((a)->size > (idx)) { \
@ -118,7 +120,7 @@ static FORCE_INLINE int32_t tarray2_make_room(void *arg, // array
do { \ do { \
if ((idx) < (a)->size) { \ if ((idx) < (a)->size) { \
if (cb) { \ if (cb) { \
void (*cb_)(void *) = cb; \ TArray2Cb cb_ = (TArray2Cb)(cb); \
cb_((a)->data + (idx)); \ cb_((a)->data + (idx)); \
} \ } \
if ((idx) < (a)->size - 1) { \ if ((idx) < (a)->size - 1) { \

22
source/dnode/vnode/src/tsdb/dev/inc/tsdbFSet.h
View File

@ -36,16 +36,16 @@ typedef enum {
TSDB_FOP_TRUNCATE, TSDB_FOP_TRUNCATE,
} tsdb_fop_t; } tsdb_fop_t;
int32_t tsdbFileSetInit(int32_t fid, STFileSet **fset); int32_t tsdbTFileSetInit(int32_t fid, STFileSet **fset);
int32_t tsdbFileSetInitEx(const STFileSet *fset1, STFileSet **fset2); int32_t tsdbTFileSetInitEx(const STFileSet *fset1, STFileSet **fset2);
int32_t tsdbFileSetClear(STFileSet **fset); int32_t tsdbTFileSetClear(STFileSet **fset);
int32_t tsdbFileSetToJson(const STFileSet *fset, cJSON *json); int32_t tsdbTFileSetToJson(const STFileSet *fset, cJSON *json);
int32_t tsdbJsonToFileSet(const cJSON *json, STFileSet **fset); int32_t tsdbJsonToTFileSet(const cJSON *json, STFileSet **fset);
int32_t tsdbFileSetEdit(STFileSet *fset, const STFileOp *op);
int32_t tsdbFSetCmprFn(const STFileSet *pSet1, const STFileSet *pSet2);
const SSttLvl *tsdbFileSetGetLvl(const STFileSet *fset, int32_t level); int32_t tsdbTFileSetEdit(STFileSet *fset, const STFileOp *op);
const SSttLvl *tsdbTFileSetGetLvl(const STFileSet *fset, int32_t level);
struct STFileOp { struct STFileOp {
tsdb_fop_t op; tsdb_fop_t op;
@ -55,10 +55,8 @@ struct STFileOp {
}; };
struct SSttLvl { struct SSttLvl {
int32_t level; // level int32_t level;
int32_t nstt; // number of .stt files on this level TFileObjArray farr;
SRBTree sttTree; // .stt file tree, sorted by cid
SRBTreeNode rbtn;
}; };
struct STFileSet { struct STFileSet {

13
source/dnode/vnode/src/tsdb/dev/inc/tsdbFile.h
View File

@ -24,6 +24,7 @@ extern "C" {
typedef struct STFile STFile; typedef struct STFile STFile;
typedef struct STFileObj STFileObj; typedef struct STFileObj STFileObj;
typedef TARRAY2(STFileObj *) TFileObjArray;
typedef enum { typedef enum {
TSDB_FTYPE_HEAD = 0, // .head TSDB_FTYPE_HEAD = 0, // .head
@ -37,17 +38,15 @@ typedef enum {
#define TSDB_FTYPE_MAX (TSDB_FTYPE_TOMB + 1) #define TSDB_FTYPE_MAX (TSDB_FTYPE_TOMB + 1)
// STFile // STFile
int32_t tsdbTFileInit(STsdb *pTsdb, STFile *pFile);
int32_t tsdbTFileClear(STFile *pFile);
int32_t tsdbTFileToJson(const STFile *f, cJSON *json); int32_t tsdbTFileToJson(const STFile *f, cJSON *json);
int32_t tsdbJsonToTFile(const cJSON *json, tsdb_ftype_t ftype, STFile *f); int32_t tsdbJsonToTFile(const cJSON *json, tsdb_ftype_t ftype, STFile *f);
// STFileObj // STFileObj
int32_t tsdbTFileObjCreate(const STFile *f, STFileObj **fobj); int32_t tsdbTFileObjInit(const STFile *f, STFileObj **fobj);
int32_t tsdbTFileObjDestroy(STFileObj *fobj); int32_t tsdbTFileObjRef(STFileObj *fobj);
int32_t tsdbTFileObjUnref(STFileObj *fobj);
struct STFile { struct STFile {
char fname[TSDB_FILENAME_LEN];
tsdb_ftype_t type; tsdb_ftype_t type;
SDiskID did; // disk id SDiskID did; // disk id
int32_t fid; // file id int32_t fid; // file id
@ -62,9 +61,9 @@ struct STFile {
}; };
struct STFileObj { struct STFileObj {
SRBTreeNode rbtn;
volatile int32_t ref;
STFile f; STFile f;
volatile int32_t ref;
char fname[TSDB_FILENAME_LEN];
}; };
#ifdef __cplusplus #ifdef __cplusplus

19
source/dnode/vnode/src/tsdb/dev/tsdbCommit.c
View File

@ -72,7 +72,7 @@ static int32_t open_writer_with_new_stt(SCommitter *pCommitter) {
config.file.size = 0; config.file.size = 0;
config.file.stt.level = 0; config.file.stt.level = 0;
config.file.stt.nseg = 0; config.file.stt.nseg = 0;
tsdbTFileInit(pTsdb, &config.file); // tsdbTFileInit(pTsdb, &config.file);
code = tsdbSttFWriterOpen(&config, &pCommitter->pWriter); code = tsdbSttFWriterOpen(&config, &pCommitter->pWriter);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
@ -120,21 +120,22 @@ static int32_t open_committer_writer(SCommitter *pCommitter) {
return open_writer_with_new_stt(pCommitter); return open_writer_with_new_stt(pCommitter);
} }
const SSttLvl *lvl0 = tsdbFileSetGetLvl(pCommitter->fset, 0); const SSttLvl *lvl0 = tsdbTFileSetGetLvl(pCommitter->fset, 0);
if (lvl0 == NULL) { if (lvl0 == NULL) {
return open_writer_with_new_stt(pCommitter); return open_writer_with_new_stt(pCommitter);
} }
SRBTreeNode *node = tRBTreeMax(&lvl0->sttTree); SRBTreeNode *node = NULL; // tRBTreeMax(&lvl0->sttTree);
if (node == NULL) { if (node == NULL) {
return open_writer_with_new_stt(pCommitter); return open_writer_with_new_stt(pCommitter);
} else { } else {
STFileObj *fobj = TCONTAINER_OF(node, STFileObj, rbtn); // STFileObj *fobj = TCONTAINER_OF(node, STFileObj, rbtn);
if (fobj->f.stt.nseg >= pCommitter->sttTrigger) { // if (fobj->f.stt.nseg >= pCommitter->sttTrigger) {
return open_writer_with_new_stt(pCommitter); // return open_writer_with_new_stt(pCommitter);
} else { // } else {
return open_writer_with_exist_stt(pCommitter, &fobj->f); // return open_writer_with_exist_stt(pCommitter, &fobj->f);
} // }
return 0;
} }
} }

6
source/dnode/vnode/src/tsdb/dev/tsdbFS.c
View File

@ -171,7 +171,7 @@ static int32_t save_fs(const TFileSetArray *arr, const char *fname) {
if (!item) TSDB_CHECK_CODE(code = TSDB_CODE_OUT_OF_MEMORY, lino, _exit); if (!item) TSDB_CHECK_CODE(code = TSDB_CODE_OUT_OF_MEMORY, lino, _exit);
cJSON_AddItemToArray(ajson, item); cJSON_AddItemToArray(ajson, item);
code = tsdbFileSetToJson(fset, item); code = tsdbTFileSetToJson(fset, item);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
} }
@ -214,7 +214,7 @@ static int32_t load_fs(const char *fname, TFileSetArray *arr) {
const cJSON *titem; const cJSON *titem;
cJSON_ArrayForEach(titem, item) { cJSON_ArrayForEach(titem, item) {
STFileSet *fset; STFileSet *fset;
code = tsdbJsonToFileSet(titem, &fset); code = tsdbJsonToTFileSet(titem, &fset);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
code = TARRAY2_APPEND(arr, fset); code = TARRAY2_APPEND(arr, fset);
@ -506,7 +506,7 @@ static int32_t edit_fs(STFileSystem *pFS, const SArray *aFileOp) {
// } // }
} }
code = tsdbFileSetEdit(pSet, op); code = tsdbTFileSetEdit(pSet, op);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
} }

339
source/dnode/vnode/src/tsdb/dev/tsdbFSet.c
View File

@ -15,18 +15,48 @@
#include "inc/tsdbFSet.h" #include "inc/tsdbFSet.h"
static int32_t stt_lvl_to_json(const SSttLvl *lvl, cJSON *json) { static int32_t tsdbSttLvlInit(int32_t level, SSttLvl **lvl) {
lvl[0] = taosMemoryMalloc(sizeof(SSttLvl));
if (lvl[0] == NULL) return TSDB_CODE_OUT_OF_MEMORY;
lvl[0]->level = level;
TARRAY2_INIT(&lvl[0]->farr);
return 0;
}
static void tsdbSttLvlClearFObj(void *data) { tsdbTFileObjUnref(*(STFileObj **)data); }
static int32_t tsdbSttLvlClear(SSttLvl **lvl) {
TARRAY2_CLEAR_FREE(&lvl[0]->farr, tsdbSttLvlClearFObj);
taosMemoryFree(lvl[0]);
lvl[0] = NULL;
return 0;
}
static int32_t tsdbSttLvlInitEx(const SSttLvl *lvl1, SSttLvl **lvl) {
int32_t code = tsdbSttLvlInit(lvl1->level, lvl);
if (code) return code;
const STFileObj *fobj1;
TARRAY2_FOREACH(&lvl1->farr, fobj1) {
STFileObj *fobj;
code = tsdbTFileObjInit(&fobj1->f, &fobj);
if (code) {
tsdbSttLvlClear(lvl);
return code;
}
TARRAY2_APPEND(&lvl[0]->farr, fobj);
}
return 0;
}
static int32_t tsdbSttLvlToJson(const SSttLvl *lvl, cJSON *json) {
if (cJSON_AddNumberToObject(json, "level", lvl->level) == NULL) { if (cJSON_AddNumberToObject(json, "level", lvl->level) == NULL) {
return TSDB_CODE_OUT_OF_MEMORY; return TSDB_CODE_OUT_OF_MEMORY;
} }
cJSON *ajson = cJSON_AddArrayToObject(json, "files"); cJSON *ajson = cJSON_AddArrayToObject(json, "files");
if (ajson == NULL) return TSDB_CODE_OUT_OF_MEMORY; if (ajson == NULL) return TSDB_CODE_OUT_OF_MEMORY;
const STFileObj *fobj;
SRBTreeIter iter = tRBTreeIterCreate(&lvl->sttTree, 1); TARRAY2_FOREACH(&lvl->farr, fobj) {
for (SRBTreeNode *node = tRBTreeIterNext(&iter); node; node = tRBTreeIterNext(&iter)) {
STFileObj *fobj = TCONTAINER_OF(node, STFileObj, rbtn);
cJSON *item = cJSON_CreateObject(); cJSON *item = cJSON_CreateObject();
if (item == NULL) return TSDB_CODE_OUT_OF_MEMORY; if (item == NULL) return TSDB_CODE_OUT_OF_MEMORY;
cJSON_AddItemToArray(ajson, item); cJSON_AddItemToArray(ajson, item);
@ -38,117 +68,47 @@ static int32_t stt_lvl_to_json(const SSttLvl *lvl, cJSON *json) {
return 0; return 0;
} }
static int32_t stt_file_cmpr(const SRBTreeNode *n1, const SRBTreeNode *n2) { static int32_t tsdbJsonToSttLvl(const cJSON *json, SSttLvl **lvl) {
STFileObj *f1 = TCONTAINER_OF(n1, STFileObj, rbtn);
STFileObj *f2 = TCONTAINER_OF(n2, STFileObj, rbtn);
if (f1->f.cid < f2->f.cid) {
return -1;
} else if (f1->f.cid > f2->f.cid) {
return 1;
}
return 0;
}
static int32_t stt_lvl_init(SSttLvl *lvl, int32_t level) {
lvl->level = level;
lvl->nstt = 0;
tRBTreeCreate(&lvl->sttTree, stt_file_cmpr);
return 0;
}
static int32_t add_file_to_stt_lvl(SSttLvl *lvl, STFileObj *fobj) {
lvl->nstt++;
tRBTreePut(&lvl->sttTree, &fobj->rbtn);
return 0;
}
static int32_t json_to_stt_lvl(const cJSON *json, SSttLvl *lvl) {
const cJSON *item1, *item2; const cJSON *item1, *item2;
int32_t level;
item1 = cJSON_GetObjectItem(json, "level"); item1 = cJSON_GetObjectItem(json, "level");
if (cJSON_IsNumber(item1)) { if (cJSON_IsNumber(item1)) {
lvl->level = item1->valuedouble; level = item1->valuedouble;
} else { } else {
return TSDB_CODE_FILE_CORRUPTED; return TSDB_CODE_FILE_CORRUPTED;
} }
stt_lvl_init(lvl, lvl->level); int32_t code = tsdbSttLvlInit(level, lvl);
if (code) return code;
item1 = cJSON_GetObjectItem(json, "files"); item1 = cJSON_GetObjectItem(json, "files");
if (cJSON_IsArray(item1)) { if (!cJSON_IsArray(item1)) {
cJSON_ArrayForEach(item2, item1) { tsdbSttLvlClear(lvl);
STFileObj *fobj;
// int32_t code = tsdbTFileObjCreate(&fobj);
// if (code) return code;
// code = tsdbJsonToTFile(item2, TSDB_FTYPE_STT, &fobj->f);
// if (code) return code;
add_file_to_stt_lvl(lvl, fobj);
}
} else {
return TSDB_CODE_FILE_CORRUPTED; return TSDB_CODE_FILE_CORRUPTED;
} }
return 0; cJSON_ArrayForEach(item2, item1) {
} STFile tf;
code = tsdbJsonToTFile(item2, TSDB_FTYPE_STT, &tf);
if (code) {
tsdbSttLvlClear(lvl);
return code;
}
static int32_t add_stt_lvl(STFileSet *fset, SSttLvl *lvl) { STFileObj *fobj;
// tRBTreePut(&fset->lvlTree, &lvl->rbtn); code = tsdbTFileObjInit(&tf, &fobj);
return 0; if (code) {
} tsdbSttLvlClear(lvl);
return code;
}
static int32_t add_file_to_fset(STFileSet *fset, STFileObj *fobj) { TARRAY2_APPEND(&lvl[0]->farr, fobj);
// if (fobj->f.type == TSDB_FTYPE_STT) {
// SSttLvl *lvl;
// SSttLvl tlvl = {.level = fobj->f.stt.level};
// SRBTreeNode *node = tRBTreeGet(&fset->lvlTree, &tlvl.rbtn);
// if (node) {
// lvl = TCONTAINER_OF(node, SSttLvl, rbtn);
// } else {
// lvl = taosMemoryMalloc(sizeof(*lvl));
// if (!lvl) return TSDB_CODE_OUT_OF_MEMORY;
// stt_lvl_init(lvl, fobj->f.stt.level);
// add_stt_lvl(fset, lvl);
// }
// add_file_to_stt_lvl(lvl, fobj);
// } else {
// fset->farr[fobj->f.type] = fobj;
// }
return 0;
}
static int32_t stt_lvl_cmpr(const SRBTreeNode *n1, const SRBTreeNode *n2) {
SSttLvl *lvl1 = TCONTAINER_OF(n1, SSttLvl, rbtn);
SSttLvl *lvl2 = TCONTAINER_OF(n2, SSttLvl, rbtn);
if (lvl1->level < lvl2->level) {
return -1;
} else if (lvl1->level > lvl2->level) {
return 1;
} }
return 0; return 0;
} }
// static int32_t fset_init(STFileSet *fset, int32_t fid) { int32_t tsdbTFileSetToJson(const STFileSet *fset, cJSON *json) {
// fset->fid = fid;
// for (int32_t ftype = TSDB_FTYPE_MIN; ftype < TSDB_FTYPE_MAX; ++ftype) {
// fset->farr[ftype] = NULL;
// }
// tRBTreeCreate(&fset->lvlTree, stt_lvl_cmpr);
// return 0;
// }
static int32_t fset_clear(STFileSet *fset) {
// TODO
return 0;
}
int32_t tsdbFileSetToJson(const STFileSet *fset, cJSON *json) {
int32_t code = 0; int32_t code = 0;
cJSON *item1, *item2; cJSON *item1, *item2;
@ -165,7 +125,7 @@ int32_t tsdbFileSetToJson(const STFileSet *fset, cJSON *json) {
} }
// each level // each level
item1 = cJSON_AddArrayToObject(json, "stt levels"); item1 = cJSON_AddArrayToObject(json, "stt lvl");
if (item1 == NULL) return TSDB_CODE_OUT_OF_MEMORY; if (item1 == NULL) return TSDB_CODE_OUT_OF_MEMORY;
const SSttLvl *lvl; const SSttLvl *lvl;
TARRAY2_FOREACH(&fset->lvlArr, lvl) { TARRAY2_FOREACH(&fset->lvlArr, lvl) {
@ -173,90 +133,82 @@ int32_t tsdbFileSetToJson(const STFileSet *fset, cJSON *json) {
if (!item2) return TSDB_CODE_OUT_OF_MEMORY; if (!item2) return TSDB_CODE_OUT_OF_MEMORY;
cJSON_AddItemToArray(item1, item2); cJSON_AddItemToArray(item1, item2);
code = stt_lvl_to_json(lvl, item2); code = tsdbSttLvlToJson(lvl, item2);
if (code) return code; if (code) return code;
} }
return 0; return 0;
} }
int32_t tsdbJsonToFileSet(const cJSON *json, STFileSet **fset) { int32_t tsdbJsonToTFileSet(const cJSON *json, STFileSet **fset) {
// const cJSON *item1, *item2; int32_t code;
// int32_t code; const cJSON *item1, *item2;
// STFile tf; int32_t fid;
STFile tf;
// /* fid */ // fid
// item1 = cJSON_GetObjectItem(json, "fid"); item1 = cJSON_GetObjectItem(json, "fid");
// if (cJSON_IsNumber(item1)) { if (cJSON_IsNumber(item1)) {
// fset->fid = item1->valueint; fid = item1->valuedouble;
// } else { } else {
// return TSDB_CODE_FILE_CORRUPTED; return TSDB_CODE_FILE_CORRUPTED;
// } }
// fset_init(fset, fset->fid); code = tsdbTFileSetInit(fid, fset);
// for (int32_t ftype = TSDB_FTYPE_MIN; ftype < TSDB_FTYPE_MAX; ++ftype) { if (code) return code;
// code = tsdbJsonToTFile(json, ftype, &tf);
// if (code == TSDB_CODE_NOT_FOUND) {
// continue;
// } else if (code) {
// return code;
// } else {
// code = tsdbTFileObjCreate(&fset->farr[ftype]);
// if (code) return code;
// fset->farr[ftype]->f = tf;
// }
// }
// // each level for (tsdb_ftype_t ftype = TSDB_FTYPE_MIN; ftype < TSDB_FTYPE_MAX; ++ftype) {
// item1 = cJSON_GetObjectItem(json, "stt"); code = tsdbJsonToTFile(json, ftype, &tf);
// if (cJSON_IsArray(item1)) { if (code == TSDB_CODE_NOT_FOUND) {
// cJSON_ArrayForEach(item2, item1) { continue;
// SSttLvl *lvl = taosMemoryCalloc(1, sizeof(*lvl)); } else if (code) {
// if (lvl == NULL) return TSDB_CODE_OUT_OF_MEMORY; tsdbTFileSetClear(fset);
return code;
}
}
// code = json_to_stt_lvl(item2, lvl); // each level
// if (code) { item1 = cJSON_GetObjectItem(json, "stt lvl");
// taosMemoryFree(lvl); if (cJSON_IsArray(item1)) {
// return code; cJSON_ArrayForEach(item2, item1) {
// } SSttLvl *lvl;
code = tsdbJsonToSttLvl(item2, &lvl);
if (code) {
tsdbTFileSetClear(fset);
return code;
}
// add_stt_lvl(fset, lvl); TARRAY2_APPEND(&(*fset)->lvlArr, lvl);
// } }
// } else { } else {
// return TSDB_CODE_FILE_CORRUPTED; return TSDB_CODE_FILE_CORRUPTED;
// } }
return 0; return 0;
} }
int32_t tsdbFSetCmprFn(const STFileSet *pSet1, const STFileSet *pSet2) { int32_t tsdbTFileSetEdit(STFileSet *fset, const STFileOp *op) {
if (pSet1->fid < pSet2->fid) return -1;
if (pSet1->fid > pSet2->fid) return 1;
return 0;
}
int32_t tsdbFileSetEdit(STFileSet *fset, const STFileOp *op) {
int32_t code = 0; int32_t code = 0;
if (op->oState.size == 0 // // if (op->oState.size == 0 //
|| 0 /* TODO*/ // || 0 /* TODO*/
) { // ) {
STFileObj *fobj; // STFileObj *fobj;
// code = tsdbTFileObjCreate(&fobj); // // code = tsdbTFileObjCreate(&fobj);
if (code) return code; // if (code) return code;
fobj->f = op->nState; // fobj->f = op->nState;
add_file_to_fset(fset, fobj); // add_file_to_fset(fset, fobj);
} else if (op->nState.size == 0) { // } else if (op->nState.size == 0) {
// delete // // delete
ASSERT(0); // ASSERT(0);
} else { // } else {
// modify // // modify
ASSERT(0); // ASSERT(0);
} // }
return 0; return 0;
} }
int32_t tsdbFileSetInit(int32_t fid, STFileSet **fset) { int32_t tsdbTFileSetInit(int32_t fid, STFileSet **fset) {
fset[0] = taosMemoryCalloc(1, sizeof(STFileSet)); fset[0] = taosMemoryCalloc(1, sizeof(STFileSet));
if (fset[0] == NULL) return TSDB_CODE_OUT_OF_MEMORY; if (fset[0] == NULL) return TSDB_CODE_OUT_OF_MEMORY;
@ -265,16 +217,16 @@ int32_t tsdbFileSetInit(int32_t fid, STFileSet **fset) {
return 0; return 0;
} }
int32_t tsdbFileSetInitEx(const STFileSet *fset1, STFileSet **fset) { int32_t tsdbTFileSetInitEx(const STFileSet *fset1, STFileSet **fset) {
int32_t code = tsdbFileSetInit(fset1->fid, fset); int32_t code = tsdbTFileSetInit(fset1->fid, fset);
if (code) return code; if (code) return code;
for (int32_t ftype = TSDB_FTYPE_MIN; ftype < TSDB_FTYPE_MAX; ++ftype) { for (int32_t ftype = TSDB_FTYPE_MIN; ftype < TSDB_FTYPE_MAX; ++ftype) {
if (fset1->farr[ftype] == NULL) continue; if (fset1->farr[ftype] == NULL) continue;
code = tsdbTFileObjCreate(&fset1->farr[ftype]->f, &fset[0]->farr[ftype]); code = tsdbTFileObjInit(&fset1->farr[ftype]->f, &fset[0]->farr[ftype]);
if (code) { if (code) {
tsdbFileSetClear(fset); tsdbTFileSetClear(fset);
return code; return code;
} }
} }
@ -282,56 +234,35 @@ int32_t tsdbFileSetInitEx(const STFileSet *fset1, STFileSet **fset) {
const SSttLvl *lvl1; const SSttLvl *lvl1;
TARRAY2_FOREACH(&fset1->lvlArr, lvl1) { TARRAY2_FOREACH(&fset1->lvlArr, lvl1) {
SSttLvl *lvl; SSttLvl *lvl;
// code = stt_lvl_init_ex(lvl1, &lvl); code = tsdbSttLvlInitEx(lvl1, &lvl);
if (code) { if (code) {
tsdbFileSetClear(fset); tsdbTFileSetClear(fset);
return code; return code;
} }
TARRAY2_APPEND(&fset[0]->lvlArr, lvl);
} }
// SRBTreeIter iter = tRBTreeIterCreate(&fset1->lvlTree, 1);
// for (SRBTreeNode *node = tRBTreeIterNext(&iter); node; node = tRBTreeIterNext(&iter)) {
// SSttLvl *lvl1 = TCONTAINER_OF(node, SSttLvl, rbtn);
// SSttLvl *lvl2 = taosMemoryCalloc(1, sizeof(*lvl2));
// if (lvl2 == NULL) return TSDB_CODE_OUT_OF_MEMORY;
// stt_lvl_init(lvl2, lvl1->level);
// add_stt_lvl(fset2, lvl2);
// SRBTreeIter iter2 = tRBTreeIterCreate(&lvl1->sttTree, 1);
// for (SRBTreeNode *node2 = tRBTreeIterNext(&iter2); node2; node2 = tRBTreeIterNext(&iter2)) {
// STFileObj *fobj1 = TCONTAINER_OF(node2, STFileObj, rbtn);
// STFileObj *fobj2;
// code = tsdbTFileObjCreate(&fobj2);
// if (code) return code;
// fobj2->f = fobj1->f;
// add_file_to_stt_lvl(lvl2, fobj2);
// }
// }
return 0; return 0;
} }
int32_t tsdbFileSetClear(STFileSet **fset) { int32_t tsdbTFileSetClear(STFileSet **fset) {
if (fset[0]) { if (!fset[0]) return 0;
for (tsdb_ftype_t ftype = TSDB_FTYPE_MIN; ftype < TSDB_FTYPE_MAX; ++ftype) {
// if (fset[0]->farr[ftype]) {
// tsdbTFileObjDestroy(&fset[0]->farr[ftype]);
// fset[0]->farr[ftype] = NULL;
// }
}
// TODO for (tsdb_ftype_t ftype = TSDB_FTYPE_MIN; ftype < TSDB_FTYPE_MAX; ++ftype) {
// SSttLvl *lvl; if (fset[0]->farr[ftype] == NULL) continue;
// TARRAY2_FOREACH(&fset[0]->lvlArr, lvl) { tsdbTFileObjUnref(fset[0]->farr[ftype]);
// // stt_lvl_clear(&lvl);
// }
taosMemoryFree(fset[0]);
fset[0] = NULL;
} }
TARRAY2_CLEAR_FREE(&fset[0]->lvlArr, tsdbSttLvlClear);
taosMemoryFree(fset[0]);
fset[0] = NULL;
return 0; return 0;
} }
const SSttLvl *tsdbFileSetGetLvl(const STFileSet *fset, int32_t level) { const SSttLvl *tsdbTFileSetGetLvl(const STFileSet *fset, int32_t level) {
// SSttLvl tlvl = {.level = level}; // SSttLvl tlvl = {.level = level};
// SRBTreeNode *node = tRBTreeGet(&fset->lvlTree, &tlvl.rbtn); // SRBTreeNode *node = tRBTreeGet(&fset->lvlTree, &tlvl.rbtn);
// return node ? TCONTAINER_OF(node, SSttLvl, rbtn) : NULL; // return node ? TCONTAINER_OF(node, SSttLvl, rbtn) : NULL;

85
source/dnode/vnode/src/tsdb/dev/tsdbFile.c
View File

@ -166,41 +166,35 @@ static int32_t stt_from_json(const cJSON *json, STFile *file) {
return 0; return 0;
} }
int32_t tsdbTFileInit(STsdb *pTsdb, STFile *pFile) { // int32_t tsdbTFileInit(STsdb *pTsdb, STFile *pFile) {
SVnode *pVnode = pTsdb->pVnode; // SVnode *pVnode = pTsdb->pVnode;
STfs *pTfs = pVnode->pTfs; // STfs *pTfs = pVnode->pTfs;
if (pTfs) { // if (pTfs) {
snprintf(pFile->fname, // // // snprintf(pFile->fname, //
TSDB_FILENAME_LEN, // // // TSDB_FILENAME_LEN, //
"%s%s%s%sv%df%dver%" PRId64 ".%s", // // // "%s%s%s%sv%df%dver%" PRId64 ".%s", //
tfsGetDiskPath(pTfs, pFile->did), // // // tfsGetDiskPath(pTfs, pFile->did), //
TD_DIRSEP, // // // TD_DIRSEP, //
pTsdb->path, // // // pTsdb->path, //
TD_DIRSEP, // // // TD_DIRSEP, //
TD_VID(pVnode), // // // TD_VID(pVnode), //
pFile->fid, // // // pFile->fid, //
pFile->cid, // // // pFile->cid, //
g_tfile_info[pFile->type].suffix); // // g_tfile_info[pFile->type].suffix);
} else { // } else {
snprintf(pFile->fname, // // // snprintf(pFile->fname, //
TSDB_FILENAME_LEN, // // // TSDB_FILENAME_LEN, //
"%s%sv%df%dver%" PRId64 ".%s", // // // "%s%sv%df%dver%" PRId64 ".%s", //
pTsdb->path, // // // pTsdb->path, //
TD_DIRSEP, // // // TD_DIRSEP, //
TD_VID(pVnode), // // // TD_VID(pVnode), //
pFile->fid, // // // pFile->fid, //
pFile->cid, // // // pFile->cid, //
g_tfile_info[pFile->type].suffix); // // g_tfile_info[pFile->type].suffix);
} // }
// pFile->ref = 1; // return 0;
return 0; // }
}
int32_t tsdbTFileClear(STFile *pFile) {
// TODO
return 0;
}
int32_t tsdbTFileToJson(const STFile *file, cJSON *json) { int32_t tsdbTFileToJson(const STFile *file, cJSON *json) {
if (file->type == TSDB_FTYPE_STT) { if (file->type == TSDB_FTYPE_STT) {
@ -228,21 +222,30 @@ int32_t tsdbJsonToTFile(const cJSON *json, tsdb_ftype_t ftype, STFile *f) {
} }
} }
// TODO: tsdbTFileInit(NULL, f);
return 0; return 0;
} }
int32_t tsdbTFileObjCreate(const STFile *f, STFileObj **fobj) { int32_t tsdbTFileObjInit(const STFile *f, STFileObj **fobj) {
fobj[0] = taosMemoryMalloc(sizeof(STFileObj)); fobj[0] = taosMemoryMalloc(sizeof(*fobj[0]));
if (!fobj[0]) return TSDB_CODE_OUT_OF_MEMORY; if (!fobj[0]) return TSDB_CODE_OUT_OF_MEMORY;
fobj[0]->f = *f;
fobj[0]->ref = 1; fobj[0]->ref = 1;
// TODO // TODO: generate the file name
return 0; return 0;
} }
int32_t tsdbTFileObjDestroy(STFileObj *fobj) { int32_t tsdbTFileObjRef(STFileObj *fobj) {
// TODO int32_t nRef = atomic_fetch_add_32(&fobj->ref, 1);
taosMemoryFree(fobj); ASSERT(nRef > 0);
return 0;
}
int32_t tsdbTFileObjUnref(STFileObj *fobj) {
int32_t nRef = atomic_sub_fetch_32(&fobj->ref, 1);
ASSERT(nRef >= 0);
if (nRef == 0) {
taosMemoryFree(fobj);
}
return 0; return 0;
} }

4
source/dnode/vnode/src/tsdb/dev/tsdbSttFReaderWriter.c
View File

@ -479,7 +479,7 @@ static int32_t open_stt_fwriter(SSttFileWriter *pWriter) {
flag |= (TD_FILE_CREATE | TD_FILE_TRUNC); flag |= (TD_FILE_CREATE | TD_FILE_TRUNC);
} }
code = tsdbOpenFile(pWriter->config.file.fname, pWriter->config.szPage, flag, &pWriter->pFd); code = tsdbOpenFile(NULL /*pWriter->config.file.fname*/, pWriter->config.szPage, flag, &pWriter->pFd);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
if (pWriter->tFile.size == 0) { if (pWriter->tFile.size == 0) {
@ -494,7 +494,7 @@ _exit:
if (pWriter->pFd) tsdbCloseFile(&pWriter->pFd); if (pWriter->pFd) tsdbCloseFile(&pWriter->pFd);
tsdbError("vgId:%d %s failed at line %d since %s", vid, __func__, lino, tstrerror(code)); tsdbError("vgId:%d %s failed at line %d since %s", vid, __func__, lino, tstrerror(code));
} else { } else {
tsdbDebug("vgId:%d %s done, fname:%s size:%" PRId64, vid, __func__, pWriter->config.file.fname, tsdbDebug("vgId:%d %s done, fname:%s size:%" PRId64, vid, __func__, "" /*pWriter->config.file.fname*/,
pWriter->config.file.size); pWriter->config.file.size);
} }
return code; return code;