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

enh: unify handling of vnode primary dir for vnd and tsdb

This commit is contained in:
Benguang Zhao 2023-07-12 15:17:52 +08:00
parent 4c3bf41178
commit 05e8af4837
7 changed files with 51 additions and 76 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
source/dnode/vnode/src/tsdb/tsdbCache.c
View File

@ -13,6 +13,7 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>. * along with this program. If not, see <http://www.gnu.org/licenses/>.
*/ */
#include "tsdb.h" #include "tsdb.h"
#include "vnd.h"
#define ROCKS_BATCH_SIZE (4096) #define ROCKS_BATCH_SIZE (4096)
@ -58,16 +59,10 @@ typedef struct {
static void tsdbGetRocksPath(STsdb *pTsdb, char *path) { static void tsdbGetRocksPath(STsdb *pTsdb, char *path) {
SVnode *pVnode = pTsdb->pVnode; SVnode *pVnode = pTsdb->pVnode;
if (pVnode->pTfs) { vnodeGetAbsDir(pTsdb->path, pVnode->pTfs, path, TSDB_FILENAME_LEN);
if (path) {
snprintf(path, TSDB_FILENAME_LEN, "%s%s%s%scache.rdb", tfsGetPrimaryPath(pTsdb->pVnode->pTfs), TD_DIRSEP, int32_t offset = strlen(path);
pTsdb->path, TD_DIRSEP); snprintf(path + offset, TSDB_FILENAME_LEN - offset - 1, "%scache.rdb", TD_DIRSEP);
}
} else {
if (path) {
snprintf(path, TSDB_FILENAME_LEN, "%s%scache.rdb", pTsdb->path, TD_DIRSEP);
}
}
} }
static const char *myCmpName(void *state) { static const char *myCmpName(void *state) {

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

@ -14,6 +14,7 @@
*/ */
#include "tsdb.h" #include "tsdb.h"
#include "vnd.h"
// ================================================================================================= // =================================================================================================
static int32_t tsdbFSToBinary(uint8_t *p, STsdbFS *pFS) { static int32_t tsdbFSToBinary(uint8_t *p, STsdbFS *pFS) {
@ -271,22 +272,20 @@ int32_t tDFileSetCmprFn(const void *p1, const void *p2) {
static void tsdbGetCurrentFName(STsdb *pTsdb, char *current, char *current_t) { static void tsdbGetCurrentFName(STsdb *pTsdb, char *current, char *current_t) {
SVnode *pVnode = pTsdb->pVnode; SVnode *pVnode = pTsdb->pVnode;
if (pVnode->pTfs) { int32_t offset = 0;
if (current) {
snprintf(current, TSDB_FILENAME_LEN - 1, "%s%s%s%sCURRENT", tfsGetPrimaryPath(pTsdb->pVnode->pTfs), TD_DIRSEP, // CURRENT
pTsdb->path, TD_DIRSEP); if (current) {
} vnodeGetAbsDir(pTsdb->path, pVnode->pTfs, current, TSDB_FILENAME_LEN);
if (current_t) { offset = strlen(current);
snprintf(current_t, TSDB_FILENAME_LEN - 1, "%s%s%s%sCURRENT.t", tfsGetPrimaryPath(pTsdb->pVnode->pTfs), TD_DIRSEP, snprintf(current + offset, TSDB_FILENAME_LEN - offset - 1, "%sCURRENT", TD_DIRSEP);
pTsdb->path, TD_DIRSEP); }
}
} else { // CURRENT.t
if (current) { if (current_t) {
snprintf(current, TSDB_FILENAME_LEN - 1, "%s%sCURRENT", pTsdb->path, TD_DIRSEP); vnodeGetAbsDir(pTsdb->path, pVnode->pTfs, current_t, TSDB_FILENAME_LEN);
} offset = strlen(current_t);
if (current_t) { snprintf(current_t + offset, TSDB_FILENAME_LEN - offset - 1, "%sCURRENT.t", TD_DIRSEP);
snprintf(current_t, TSDB_FILENAME_LEN - 1, "%s%sCURRENT.t", pTsdb->path, TD_DIRSEP);
}
} }
} }
@ -1142,4 +1141,4 @@ void tsdbFSUnref(STsdb *pTsdb, STsdbFS *pFS) {
} }
taosArrayDestroy(pFS->aDFileSet); taosArrayDestroy(pFS->aDFileSet);
} }

9
source/dnode/vnode/src/tsdb/tsdbFile.c
View File

@ -14,6 +14,7 @@
*/ */
#include "tsdb.h" #include "tsdb.h"
#include "vnd.h"
int32_t tPutHeadFile(uint8_t *p, SHeadFile *pHeadFile) { int32_t tPutHeadFile(uint8_t *p, SHeadFile *pHeadFile) {
int32_t n = 0; int32_t n = 0;
@ -282,8 +283,12 @@ int32_t tGetDFileSet(uint8_t *p, SDFileSet *pSet) {
// SDelFile =============================================== // SDelFile ===============================================
void tsdbDelFileName(STsdb *pTsdb, SDelFile *pFile, char fname[]) { void tsdbDelFileName(STsdb *pTsdb, SDelFile *pFile, char fname[]) {
snprintf(fname, TSDB_FILENAME_LEN - 1, "%s%s%s%sv%dver%" PRId64 "%s", tfsGetPrimaryPath(pTsdb->pVnode->pTfs), int32_t offset = 0;
TD_DIRSEP, pTsdb->path, TD_DIRSEP, TD_VID(pTsdb->pVnode), pFile->commitID, ".del");
vnodeGetAbsDir(pTsdb->path, pTsdb->pVnode->pTfs, fname, TSDB_FILENAME_LEN);
offset = strlen(fname);
snprintf((char *)fname + offset, TSDB_FILENAME_LEN - offset - 1, "%sv%dver%" PRId64 ".del", TD_DIRSEP,
TD_VID(pTsdb->pVnode), pFile->commitID);
} }
int32_t tPutDelFile(uint8_t *p, SDelFile *pDelFile) { int32_t tPutDelFile(uint8_t *p, SDelFile *pDelFile) {

26
source/dnode/vnode/src/vnd/vnodeCommit.c
View File

@ -290,11 +290,7 @@ static int32_t vnodePrepareCommit(SVnode *pVnode, SCommitInfo *pInfo) {
pInfo->txn = metaGetTxn(pVnode->pMeta); pInfo->txn = metaGetTxn(pVnode->pMeta);
// save info // save info
if (pVnode->pTfs) { vnodeGetAbsDir(pVnode->path, pVnode->pTfs, dir, TSDB_FILENAME_LEN);
snprintf(dir, TSDB_FILENAME_LEN, "%s%s%s", tfsGetPrimaryPath(pVnode->pTfs), TD_DIRSEP, pVnode->path);
} else {
snprintf(dir, TSDB_FILENAME_LEN, "%s", pVnode->path);
}
vDebug("vgId:%d, save config while prepare commit", TD_VID(pVnode)); vDebug("vgId:%d, save config while prepare commit", TD_VID(pVnode));
if (vnodeSaveInfo(dir, &pInfo->info) < 0) { if (vnodeSaveInfo(dir, &pInfo->info) < 0) {
@ -427,11 +423,7 @@ static int vnodeCommitImpl(SCommitInfo *pInfo) {
return -1; return -1;
} }
if (pVnode->pTfs) { vnodeGetAbsDir(pVnode->path, pVnode->pTfs, dir, TSDB_FILENAME_LEN);
snprintf(dir, TSDB_FILENAME_LEN, "%s%s%s", tfsGetPrimaryPath(pVnode->pTfs), TD_DIRSEP, pVnode->path);
} else {
snprintf(dir, TSDB_FILENAME_LEN, "%s", pVnode->path);
}
syncBeginSnapshot(pVnode->sync, pInfo->info.state.committed); syncBeginSnapshot(pVnode->sync, pInfo->info.state.committed);
@ -493,16 +485,22 @@ _exit:
bool vnodeShouldRollback(SVnode *pVnode) { bool vnodeShouldRollback(SVnode *pVnode) {
char tFName[TSDB_FILENAME_LEN] = {0}; char tFName[TSDB_FILENAME_LEN] = {0};
snprintf(tFName, TSDB_FILENAME_LEN, "%s%s%s%s%s", tfsGetPrimaryPath(pVnode->pTfs), TD_DIRSEP, pVnode->path, TD_DIRSEP, int32_t offset = 0;
VND_INFO_FNAME_TMP);
vnodeGetAbsDir(pVnode->path, pVnode->pTfs, tFName, TSDB_FILENAME_LEN);
offset = strlen(tFName);
snprintf(tFName + offset, TSDB_FILENAME_LEN - offset - 1, "%s%s", TD_DIRSEP, VND_INFO_FNAME_TMP);
return taosCheckExistFile(tFName); return taosCheckExistFile(tFName);
} }
void vnodeRollback(SVnode *pVnode) { void vnodeRollback(SVnode *pVnode) {
char tFName[TSDB_FILENAME_LEN] = {0}; char tFName[TSDB_FILENAME_LEN] = {0};
snprintf(tFName, TSDB_FILENAME_LEN, "%s%s%s%s%s", tfsGetPrimaryPath(pVnode->pTfs), TD_DIRSEP, pVnode->path, TD_DIRSEP, int32_t offset = 0;
VND_INFO_FNAME_TMP);
vnodeGetAbsDir(pVnode->path, pVnode->pTfs, tFName, TSDB_FILENAME_LEN);
offset = strlen(tFName);
snprintf(tFName + offset, TSDB_FILENAME_LEN - offset - 1, "%s%s", TD_DIRSEP, VND_INFO_FNAME_TMP);
(void)taosRemoveFile(tFName); (void)taosRemoveFile(tFName);
} }

6
source/dnode/vnode/src/vnd/vnodeOpen.c
View File

@ -291,11 +291,7 @@ SVnode *vnodeOpen(const char *path, STfs *pTfs, SMsgCb msgCb) {
char tdir[TSDB_FILENAME_LEN * 2] = {0}; char tdir[TSDB_FILENAME_LEN * 2] = {0};
int32_t ret = 0; int32_t ret = 0;
if (pTfs) { vnodeGetAbsDir(path, pTfs, dir, TSDB_FILENAME_LEN);
snprintf(dir, TSDB_FILENAME_LEN, "%s%s%s", tfsGetPrimaryPath(pTfs), TD_DIRSEP, path);
} else {
snprintf(dir, TSDB_FILENAME_LEN, "%s", path);
}
info.config = vnodeCfgDefault; info.config = vnodeCfgDefault;

14
source/dnode/vnode/src/vnd/vnodeRetention.c
View File

@ -35,11 +35,7 @@ static int32_t vnodePrepareRentention(SVnode *pVnode, SRetentionInfo *pInfo) {
pInfo->commitID = ++pVnode->state.commitID; pInfo->commitID = ++pVnode->state.commitID;
char dir[TSDB_FILENAME_LEN] = {0}; char dir[TSDB_FILENAME_LEN] = {0};
if (pVnode->pTfs) { vnodeGetAbsDir(pVnode->path, pVnode->pTfs, dir, TSDB_FILENAME_LEN);
snprintf(dir, TSDB_FILENAME_LEN, "%s%s%s", tfsGetPrimaryPath(pVnode->pTfs), TD_DIRSEP, pVnode->path);
} else {
snprintf(dir, TSDB_FILENAME_LEN, "%s", pVnode->path);
}
if (vnodeLoadInfo(dir, &pInfo->info) < 0) { if (vnodeLoadInfo(dir, &pInfo->info) < 0) {
code = terrno; code = terrno;
@ -64,11 +60,7 @@ static int32_t vnodeRetentionTask(void *param) {
SVnode *pVnode = pInfo->pVnode; SVnode *pVnode = pInfo->pVnode;
char dir[TSDB_FILENAME_LEN] = {0}; char dir[TSDB_FILENAME_LEN] = {0};
if (pVnode->pTfs) { vnodeGetAbsDir(pVnode->path, pVnode->pTfs, dir, TSDB_FILENAME_LEN);
snprintf(dir, TSDB_FILENAME_LEN, "%s%s%s", tfsGetPrimaryPath(pVnode->pTfs), TD_DIRSEP, pVnode->path);
} else {
snprintf(dir, TSDB_FILENAME_LEN, "%s", pVnode->path);
}
// save info // save info
pInfo->info.state.commitID = pInfo->commitID; pInfo->info.state.commitID = pInfo->commitID;
@ -127,4 +119,4 @@ _exit:
vInfo("vgId:%d %s done", TD_VID(pInfo->pVnode), __func__); vInfo("vgId:%d %s done", TD_VID(pInfo->pVnode), __func__);
} }
return 0; return 0;
} }

24
source/dnode/vnode/src/vnd/vnodeSnapshot.c
View File

@ -91,12 +91,11 @@ int32_t vnodeSnapRead(SVSnapReader *pReader, uint8_t **ppData, uint32_t *nData)
// FIXME: if commit multiple times and the config changed? // FIXME: if commit multiple times and the config changed?
if (!pReader->cfgDone) { if (!pReader->cfgDone) {
char fName[TSDB_FILENAME_LEN]; char fName[TSDB_FILENAME_LEN];
if (pReader->pVnode->pTfs) { int32_t offset = 0;
snprintf(fName, TSDB_FILENAME_LEN, "%s%s%s%s%s", tfsGetPrimaryPath(pReader->pVnode->pTfs), TD_DIRSEP,
pReader->pVnode->path, TD_DIRSEP, VND_INFO_FNAME); vnodeGetAbsDir(pReader->pVnode->path, pReader->pVnode->pTfs, fName, TSDB_FILENAME_LEN);
} else { offset = strlen(fName);
snprintf(fName, TSDB_FILENAME_LEN, "%s%s%s", pReader->pVnode->path, TD_DIRSEP, VND_INFO_FNAME); snprintf(fName + offset, TSDB_FILENAME_LEN - offset - 1, "%s%s", TD_DIRSEP, VND_INFO_FNAME);
}
TdFilePtr pFile = taosOpenFile(fName, TD_FILE_READ); TdFilePtr pFile = taosOpenFile(fName, TD_FILE_READ);
if (NULL == pFile) { if (NULL == pFile) {
@ -344,11 +343,7 @@ int32_t vnodeSnapWriterClose(SVSnapWriter *pWriter, int8_t rollback, SSnapshot *
.applyTerm = pWriter->info.state.commitTerm}; .applyTerm = pWriter->info.state.commitTerm};
pVnode->statis = pWriter->info.statis; pVnode->statis = pWriter->info.statis;
char dir[TSDB_FILENAME_LEN] = {0}; char dir[TSDB_FILENAME_LEN] = {0};
if (pWriter->pVnode->pTfs) { vnodeGetAbsDir(pVnode->path, pVnode->pTfs, dir, TSDB_FILENAME_LEN);
snprintf(dir, TSDB_FILENAME_LEN, "%s%s%s", tfsGetPrimaryPath(pVnode->pTfs), TD_DIRSEP, pVnode->path);
} else {
snprintf(dir, TSDB_FILENAME_LEN, "%s", pWriter->pVnode->path);
}
vnodeCommitInfo(dir); vnodeCommitInfo(dir);
} else { } else {
@ -400,12 +395,7 @@ static int32_t vnodeSnapWriteInfo(SVSnapWriter *pWriter, uint8_t *pData, uint32_
// modify info as needed // modify info as needed
char dir[TSDB_FILENAME_LEN] = {0}; char dir[TSDB_FILENAME_LEN] = {0};
if (pWriter->pVnode->pTfs) { vnodeGetAbsDir(pWriter->pVnode->path, pWriter->pVnode->pTfs, dir, TSDB_FILENAME_LEN);
snprintf(dir, TSDB_FILENAME_LEN, "%s%s%s", tfsGetPrimaryPath(pWriter->pVnode->pTfs), TD_DIRSEP,
pWriter->pVnode->path);
} else {
snprintf(dir, TSDB_FILENAME_LEN, "%s", pWriter->pVnode->path);
}
SVnodeStats vndStats = pWriter->info.config.vndStats; SVnodeStats vndStats = pWriter->info.config.vndStats;
SVnode *pVnode = pWriter->pVnode; SVnode *pVnode = pWriter->pVnode;