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

Merge remote-tracking branch 'origin/feature/vnode' into feature/3.0_liaohj

This commit is contained in:
Haojun Liao 2022-01-13 16:54:09 +08:00
parent 97c66b5ee2 254e144f04
commit 23b3981415
169 changed files with 2012 additions and 3212 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
cmake/libuv_CMakeLists.txt.in
View File

@ -4,7 +4,7 @@ ExternalProject_Add(libuv
GIT_REPOSITORY https://github.com/libuv/libuv.git
GIT_TAG v1.42.0
SOURCE_DIR "${CMAKE_CONTRIB_DIR}/libuv"
BINARY_DIR ""
BINARY_DIR "${CMAKE_CONTRIB_DIR}/libuv"
CONFIGURE_COMMAND ""
BUILD_COMMAND ""
INSTALL_COMMAND ""

10
include/dnode/bnode/bnode.h
View File

@ -23,9 +23,9 @@ extern "C" {
/* ------------------------ TYPES EXPOSED ------------------------ */
typedef struct SDnode SDnode;
typedef struct SBnode SBnode;
typedef int32_t (*SendReqToDnodeFp)(SDnode *pDnode, struct SEpSet *epSet, struct SRpcMsg *rpcMsg);
typedef int32_t (*SendReqToMnodeFp)(SDnode *pDnode, struct SRpcMsg *rpcMsg);
typedef void (*SendRedirectRspFp)(SDnode *pDnode, struct SRpcMsg *rpcMsg);
typedef int32_t (*SendReqToDnodeFp)(SDnode *pDnode, struct SEpSet *epSet, struct SRpcMsg *pMsg);
typedef int32_t (*SendReqToMnodeFp)(SDnode *pDnode, struct SRpcMsg *pMsg);
typedef void (*SendRedirectRspFp)(SDnode *pDnode, struct SRpcMsg *pMsg);
typedef struct {
int64_t numOfErrors;
@ -33,12 +33,8 @@ typedef struct {
typedef struct {
int32_t sver;
} SBnodeCfg;
typedef struct {
int32_t dnodeId;
int64_t clusterId;
SBnodeCfg cfg;
SDnode *pDnode;
SendReqToDnodeFp sendReqToDnodeFp;
SendReqToMnodeFp sendReqToMnodeFp;

42
include/dnode/mgmt/dnode.h
View File

@ -22,15 +22,39 @@
extern "C" {
#endif
/* ------------------------ TYPES EXPOSED ------------------------ */
/* ------------------------ TYPES EXPOSED ---------------- */
typedef struct SDnode SDnode;
/* ------------------------ Environment ------------------ */
typedef struct {
int32_t sver;
int32_t numOfCores;
int32_t numOfSupportVnodes;
int16_t numOfCommitThreads;
int8_t enableTelem;
char timezone[TSDB_TIMEZONE_LEN];
char locale[TSDB_LOCALE_LEN];
char charset[TSDB_LOCALE_LEN];
char buildinfo[64];
char gitinfo[48];
} SDnodeEnvCfg;
/**
* @brief Initialize the environment
*
* @param pOption Option of the environment
* @return int32_t 0 for success and -1 for failure
*/
int32_t dndInit(const SDnodeEnvCfg *pCfg);
/**
* @brief clear the environment
*
*/
void dndCleanup();
/* ------------------------ SDnode ----------------------- */
typedef struct {
int32_t numOfSupportVnodes;
int32_t statusInterval;
float numOfThreadsPerCore;
float ratioOfQueryCores;
@ -41,28 +65,22 @@ typedef struct {
char localEp[TSDB_EP_LEN];
char localFqdn[TSDB_FQDN_LEN];
char firstEp[TSDB_EP_LEN];
char timezone[TSDB_TIMEZONE_LEN];
char locale[TSDB_LOCALE_LEN];
char charset[TSDB_LOCALE_LEN];
char buildinfo[64];
char gitinfo[48];
} SDnodeOpt;
} SDnodeObjCfg;
/* ------------------------ SDnode ------------------------ */
/**
* @brief Initialize and start the dnode.
*
* @param pOption Option of the dnode.
* @param pCfg Config of the dnode.
* @return SDnode* The dnode object.
*/
SDnode *dndInit(SDnodeOpt *pOption);
SDnode *dndCreate(SDnodeObjCfg *pCfg);
/**
* @brief Stop and cleanup the dnode.
*
* @param pDnode The dnode object to close.
*/
void dndCleanup(SDnode *pDnode);
void dndClose(SDnode *pDnode);
#ifdef __cplusplus
}

10
include/dnode/qnode/qnode.h
View File

@ -23,9 +23,9 @@ extern "C" {
/* ------------------------ TYPES EXPOSED ------------------------ */
typedef struct SDnode SDnode;
typedef struct SQnode SQnode;
typedef int32_t (*SendReqToDnodeFp)(SDnode *pDnode, struct SEpSet *epSet, struct SRpcMsg *rpcMsg);
typedef int32_t (*SendReqToMnodeFp)(SDnode *pDnode, struct SRpcMsg *rpcMsg);
typedef void (*SendRedirectRspFp)(SDnode *pDnode, struct SRpcMsg *rpcMsg);
typedef int32_t (*SendReqToDnodeFp)(SDnode *pDnode, struct SEpSet *epSet, struct SRpcMsg *pMsg);
typedef int32_t (*SendReqToMnodeFp)(SDnode *pDnode, struct SRpcMsg *pMsg);
typedef void (*SendRedirectRspFp)(SDnode *pDnode, struct SRpcMsg *pMsg);
typedef struct {
int64_t numOfStartTask;
@ -40,12 +40,8 @@ typedef struct {
typedef struct {
int32_t sver;
} SQnodeCfg;
typedef struct {
int32_t dnodeId;
int64_t clusterId;
SQnodeCfg cfg;
SDnode *pDnode;
SendReqToDnodeFp sendReqToDnodeFp;
SendReqToMnodeFp sendReqToMnodeFp;

10
include/dnode/snode/snode.h
View File

@ -23,9 +23,9 @@ extern "C" {
/* ------------------------ TYPES EXPOSED ------------------------ */
typedef struct SDnode SDnode;
typedef struct SSnode SSnode;
typedef int32_t (*SendReqToDnodeFp)(SDnode *pDnode, struct SEpSet *epSet, struct SRpcMsg *rpcMsg);
typedef int32_t (*SendReqToMnodeFp)(SDnode *pDnode, struct SRpcMsg *rpcMsg);
typedef void (*SendRedirectRspFp)(SDnode *pDnode, struct SRpcMsg *rpcMsg);
typedef int32_t (*SendReqToDnodeFp)(SDnode *pDnode, struct SEpSet *epSet, struct SRpcMsg *pMsg);
typedef int32_t (*SendReqToMnodeFp)(SDnode *pDnode, struct SRpcMsg *pMsg);
typedef void (*SendRedirectRspFp)(SDnode *pDnode, struct SRpcMsg *pMsg);
typedef struct {
int64_t numOfErrors;
@ -33,12 +33,8 @@ typedef struct {
typedef struct {
int32_t sver;
} SSnodeCfg;
typedef struct {
int32_t dnodeId;
int64_t clusterId;
SSnodeCfg cfg;
SDnode *pDnode;
SendReqToDnodeFp sendReqToDnodeFp;
SendReqToMnodeFp sendReqToMnodeFp;

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

@ -49,7 +49,7 @@ typedef struct {
} STierMeta;
int tfsInit(SDiskCfg *pDiskCfg, int ndisk);
void tfsDestroy();
void tfsCleanup();
void tfsUpdateInfo(SFSMeta *pFSMeta, STierMeta *tierMetas, int8_t numLevels);
void tfsGetMeta(SFSMeta *pMeta);
void tfsAllocDisk(int expLevel, int *level, int *id);

64
include/libs/tkv/tkv.h
View File

@ -1,64 +0,0 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef _TD_TKV_H_
#define _TD_TKV_H_
#if 0
#include "os.h"
#ifdef __cplusplus
extern "C" {
#endif
// Types exported
typedef struct STkvDb STkvDb;
typedef struct STkvOpts STkvOpts;
typedef struct STkvCache STkvCache;
typedef struct STkvReadOpts STkvReadOpts;
typedef struct STkvWriteOpts STkvWriteOpts;
// DB operations
STkvDb *tkvOpen(const STkvOpts *options, const char *path);
void tkvClose(STkvDb *db);
void tkvPut(STkvDb *db, const STkvWriteOpts *, const char *key, size_t keylen, const char *val, size_t vallen);
char * tkvGet(STkvDb *db, const STkvReadOpts *, const char *key, size_t keylen, size_t *vallen);
void tkvCommit(STkvDb *db);
// DB options
STkvOpts *tkvOptsCreate();
void tkvOptsDestroy(STkvOpts *);
void tkvOptionsSetCache(STkvOpts *, STkvCache *);
void tkvOptsSetCreateIfMissing(STkvOpts *, unsigned char);
// DB cache
typedef enum { TKV_LRU_CACHE = 0, TKV_LFU_CACHE = 1 } ETkvCacheType;
STkvCache *tkvCacheCreate(size_t capacity, ETkvCacheType type);
void tkvCacheDestroy(STkvCache *);
// STkvReadOpts
STkvReadOpts *tkvReadOptsCreate();
void tkvReadOptsDestroy(STkvReadOpts *);
// STkvWriteOpts
STkvWriteOpts *tkvWriteOptsCreate();
void tkvWriteOptsDestroy(STkvWriteOpts *);
#ifdef __cplusplus
}
#endif
#endif
#endif /*_TD_TKV_H_*/

2
include/util/taoserror.h
View File

@ -70,6 +70,8 @@ int32_t* taosGetErrno();
#define TSDB_CODE_INVALID_MSG TAOS_DEF_ERROR_CODE(0, 0x0108)
#define TSDB_CODE_MSG_NOT_PROCESSED TAOS_DEF_ERROR_CODE(0, 0x0109)
#define TSDB_CODE_INVALID_PARA TAOS_DEF_ERROR_CODE(0, 0x010A)
#define TSDB_CODE_REPEAT_INIT TAOS_DEF_ERROR_CODE(0, 0x010B)
#define TSDB_CODE_REF_NO_MEMORY TAOS_DEF_ERROR_CODE(0, 0x0110)
#define TSDB_CODE_REF_FULL TAOS_DEF_ERROR_CODE(0, 0x0111)
#define TSDB_CODE_REF_ID_REMOVED TAOS_DEF_ERROR_CODE(0, 0x0112)

14
include/util/tarray.h
View File

@ -23,6 +23,19 @@ extern "C" {
#include "os.h"
#include "talgo.h"
#if 0
#define TARRAY(TYPE) \
struct { \
int32_t tarray_size_; \
int32_t tarray_neles_; \
struct TYPE* td_array_data_; \
}
#define TARRAY_SIZE(ARRAY) (ARRAY)->tarray_size_
#define TARRAY_NELES(ARRAY) (ARRAY)->tarray_neles_
#define TARRAY_ELE_AT(ARRAY, IDX) ((ARRAY)->td_array_data_ + idx)
#endif
#define TARRAY_MIN_SIZE 8
#define TARRAY_GET_ELEM(array, index) ((void*)((char*)((array)->pData) + (index) * (array)->elemSize))
#define TARRAY_ELEM_IDX(array, ele) (POINTER_DISTANCE(ele, (array)->pData) / (array)->elemSize)
@ -242,7 +255,6 @@ int32_t taosArraySearchIdx(const SArray* pArray, const void* key, __compar_fn_t
*/
char* taosArraySearchString(const SArray* pArray, const char* key, __compar_fn_t comparFn, int flags);
/**
* sort the pointer data in the array
* @param pArray

7
source/dnode/bnode/inc/bndInt.h
View File

@ -30,12 +30,7 @@ extern "C" {
#endif
typedef struct SBnode {
int32_t dnodeId;
int64_t clusterId;
SBnodeCfg cfg;
SendReqToDnodeFp sendReqToDnodeFp;
SendReqToMnodeFp sendReqToMnodeFp;
SendRedirectRspFp sendRedirectRspFp;
SBnodeOpt opt;
} SBnode;
#ifdef __cplusplus

93
source/dnode/mgmt/daemon/src/daemon.c
View File

@ -28,11 +28,11 @@ static struct {
bool printAuth;
bool printVersion;
char configDir[PATH_MAX];
} global = {0};
} dmn = {0};
void dmnSigintHandle(int signum, void *info, void *ctx) {
uInfo("singal:%d is received", signum);
global.stop = true;
dmn.stop = true;
}
void dmnSetSignalHandle() {
@ -44,7 +44,7 @@ void dmnSetSignalHandle() {
}
int dmnParseOption(int argc, char const *argv[]) {
tstrncpy(global.configDir, "/etc/taos", PATH_MAX);
tstrncpy(dmn.configDir, "/etc/taos", PATH_MAX);
for (int i = 1; i < argc; ++i) {
if (strcmp(argv[i], "-c") == 0) {
@ -53,19 +53,19 @@ int dmnParseOption(int argc, char const *argv[]) {
printf("config file path overflow");
return -1;
}
tstrncpy(global.configDir, argv[i], PATH_MAX);
tstrncpy(dmn.configDir, argv[i], PATH_MAX);
} else {
printf("'-c' requires a parameter, default is %s\n", configDir);
return -1;
}
} else if (strcmp(argv[i], "-C") == 0) {
global.dumpConfig = true;
dmn.dumpConfig = true;
} else if (strcmp(argv[i], "-k") == 0) {
global.generateGrant = true;
dmn.generateGrant = true;
} else if (strcmp(argv[i], "-A") == 0) {
global.printAuth = true;
dmn.printAuth = true;
} else if (strcmp(argv[i], "-V") == 0) {
global.printVersion = true;
dmn.printVersion = true;
} else {
}
}
@ -92,7 +92,7 @@ void dmnPrintVersion() {
}
int dmnReadConfig(const char *path) {
tstrncpy(configDir, global.configDir, PATH_MAX);
tstrncpy(configDir, dmn.configDir, PATH_MAX);
taosInitGlobalCfg();
taosReadGlobalLogCfg();
@ -114,12 +114,12 @@ int dmnReadConfig(const char *path) {
}
if (taosReadCfgFromFile() != 0) {
uError("failed to read global config");
uError("failed to read config");
return -1;
}
if (taosCheckAndPrintCfg() != 0) {
uError("failed to check global config");
uError("failed to check config");
return -1;
}
@ -131,38 +131,50 @@ void dmnDumpConfig() { taosDumpGlobalCfg(); }
void dmnWaitSignal() {
dmnSetSignalHandle();
while (!global.stop) {
while (!dmn.stop) {
taosMsleep(100);
}
}
void dmnInitOption(SDnodeOpt *pOption) {
pOption->sver = 30000000; //3.0.0.0
pOption->numOfCores = tsNumOfCores;
pOption->numOfSupportVnodes = tsNumOfSupportVnodes;
pOption->numOfCommitThreads = tsNumOfCommitThreads;
pOption->statusInterval = tsStatusInterval;
pOption->numOfThreadsPerCore = tsNumOfThreadsPerCore;
pOption->ratioOfQueryCores = tsRatioOfQueryCores;
pOption->maxShellConns = tsMaxShellConns;
pOption->shellActivityTimer = tsShellActivityTimer;
pOption->serverPort = tsServerPort;
tstrncpy(pOption->dataDir, tsDataDir, TSDB_FILENAME_LEN);
tstrncpy(pOption->localEp, tsLocalEp, TSDB_EP_LEN);
tstrncpy(pOption->localFqdn, tsLocalFqdn, TSDB_FQDN_LEN);
tstrncpy(pOption->firstEp, tsFirst, TSDB_EP_LEN);
tstrncpy(pOption->timezone, tsTimezone, TSDB_TIMEZONE_LEN);
tstrncpy(pOption->locale, tsLocale, TSDB_LOCALE_LEN);
tstrncpy(pOption->charset, tsCharset, TSDB_LOCALE_LEN);
tstrncpy(pOption->buildinfo, buildinfo, 64);
tstrncpy(pOption->gitinfo, gitinfo, 48);
void dnmInitEnvCfg(SDnodeEnvCfg *pCfg) {
pCfg->sver = 30000000; // 3.0.0.0
pCfg->numOfCores = tsNumOfCores;
pCfg->numOfCommitThreads = tsNumOfCommitThreads;
pCfg->enableTelem = 0;
tstrncpy(pCfg->timezone, tsTimezone, TSDB_TIMEZONE_LEN);
tstrncpy(pCfg->locale, tsLocale, TSDB_LOCALE_LEN);
tstrncpy(pCfg->charset, tsCharset, TSDB_LOCALE_LEN);
tstrncpy(pCfg->buildinfo, buildinfo, 64);
tstrncpy(pCfg->gitinfo, gitinfo, 48);
}
void dmnInitObjCfg(SDnodeObjCfg *pCfg) {
pCfg->numOfSupportVnodes = tsNumOfSupportVnodes;
pCfg->statusInterval = tsStatusInterval;
pCfg->numOfThreadsPerCore = tsNumOfThreadsPerCore;
pCfg->ratioOfQueryCores = tsRatioOfQueryCores;
pCfg->maxShellConns = tsMaxShellConns;
pCfg->shellActivityTimer = tsShellActivityTimer;
pCfg->serverPort = tsServerPort;
tstrncpy(pCfg->dataDir, tsDataDir, TSDB_FILENAME_LEN);
tstrncpy(pCfg->localEp, tsLocalEp, TSDB_EP_LEN);
tstrncpy(pCfg->localFqdn, tsLocalFqdn, TSDB_FQDN_LEN);
tstrncpy(pCfg->firstEp, tsFirst, TSDB_EP_LEN);
}
int dmnRunDnode() {
SDnodeOpt option = {0};
dmnInitOption(&option);
SDnodeEnvCfg envCfg = {0};
SDnodeObjCfg objCfg = {0};
SDnode *pDnode = dndInit(&option);
dnmInitEnvCfg(&envCfg);
dmnInitObjCfg(&objCfg);
if (dndInit(&envCfg) != 0) {
uInfo("Failed to start TDengine, please check the log at %s", tsLogDir);
return -1;
}
SDnode *pDnode = dndCreate(&objCfg);
if (pDnode == NULL) {
uInfo("Failed to start TDengine, please check the log at %s", tsLogDir);
return -1;
@ -172,7 +184,8 @@ int dmnRunDnode() {
dmnWaitSignal();
uInfo("TDengine is shut down!");
dndCleanup(pDnode);
dndClose(pDnode);
dndCleanup();
taosCloseLog();
return 0;
}
@ -182,21 +195,21 @@ int main(int argc, char const *argv[]) {
return -1;
}
if (global.generateGrant) {
if (dmn.generateGrant) {
dmnGenerateGrant();
return 0;
}
if (global.printVersion) {
if (dmn.printVersion) {
dmnPrintVersion();
return 0;
}
if (dmnReadConfig(global.configDir) != 0) {
if (dmnReadConfig(dmn.configDir) != 0) {
return -1;
}
if (global.dumpConfig) {
if (dmn.dumpConfig) {
dmnDumpConfig();
return 0;
}

2
source/dnode/mgmt/impl/inc/dndBnode.h
View File

@ -19,7 +19,7 @@
#ifdef __cplusplus
extern "C" {
#endif
#include "dndInt.h"
#include "dndEnv.h"
int32_t dndInitBnode(SDnode *pDnode);
void dndCleanupBnode(SDnode *pDnode);

149
source/dnode/mgmt/impl/inc/dndEnv.h Normal file
View File

@ -0,0 +1,149 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef _TD_DND_ENV_H_
#define _TD_DND_ENV_H_
#ifdef __cplusplus
extern "C" {
#endif
#include "dndInt.h"
typedef struct {
EWorkerType type;
const char *name;
int32_t minNum;
int32_t maxNum;
void *queueFp;
SDnode *pDnode;
STaosQueue *queue;
union {
SWorkerPool pool;
SMWorkerPool mpool;
};
} SDnodeWorker;
typedef struct {
char *dnode;
char *mnode;
char *snode;
char *bnode;
char *vnodes;
} SDnodeDir;
typedef struct {
int32_t dnodeId;
int32_t dropped;
int64_t clusterId;
int64_t dver;
int64_t rebootTime;
int64_t updateTime;
int8_t statusSent;
SEpSet mnodeEpSet;
char *file;
SHashObj *dnodeHash;
SDnodeEps *dnodeEps;
pthread_t *threadId;
SRWLatch latch;
SDnodeWorker mgmtWorker;
SDnodeWorker statusWorker;
} SDnodeMgmt;
typedef struct {
int32_t refCount;
int8_t deployed;
int8_t dropped;
SMnode *pMnode;
SRWLatch latch;
SDnodeWorker readWorker;
SDnodeWorker writeWorker;
SDnodeWorker syncWorker;
int8_t replica;
int8_t selfIndex;
SReplica replicas[TSDB_MAX_REPLICA];
} SMnodeMgmt;
typedef struct {
int32_t refCount;
int8_t deployed;
int8_t dropped;
SQnode *pQnode;
SRWLatch latch;
SDnodeWorker queryWorker;
SDnodeWorker fetchWorker;
} SQnodeMgmt;
typedef struct {
int32_t refCount;
int8_t deployed;
int8_t dropped;
SSnode *pSnode;
SRWLatch latch;
SDnodeWorker writeWorker;
} SSnodeMgmt;
typedef struct {
int32_t refCount;
int8_t deployed;
int8_t dropped;
SBnode *pBnode;
SRWLatch latch;
SDnodeWorker writeWorker;
} SBnodeMgmt;
typedef struct {
SHashObj *hash;
int32_t openVnodes;
int32_t totalVnodes;
SRWLatch latch;
SWorkerPool queryPool;
SWorkerPool fetchPool;
SMWorkerPool syncPool;
SMWorkerPool writePool;
} SVnodesMgmt;
typedef struct {
void *serverRpc;
void *clientRpc;
DndMsgFp msgFp[TDMT_MAX];
} STransMgmt;
typedef struct SDnode {
EStat stat;
SDnodeObjCfg cfg;
SDnodeEnvCfg env;
SDnodeDir dir;
FileFd lockFd;
SDnodeMgmt dmgmt;
SMnodeMgmt mmgmt;
SQnodeMgmt qmgmt;
SSnodeMgmt smgmt;
SBnodeMgmt bmgmt;
SVnodesMgmt vmgmt;
STransMgmt tmgmt;
SStartupReq startup;
} SDnode;
typedef struct {
int8_t once;
SDnodeEnvCfg cfg;
} SDnodeEnv;
#ifdef __cplusplus
}
#endif
#endif /*_TD_DND_ENV_H_*/

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

@ -55,124 +55,12 @@ extern int32_t dDebugFlag;
typedef enum { DND_STAT_INIT, DND_STAT_RUNNING, DND_STAT_STOPPED } EStat;
typedef enum { DND_WORKER_SINGLE, DND_WORKER_MULTI } EWorkerType;
typedef enum { DND_ENV_INIT = 0, DND_ENV_READY = 1, DND_ENV_CLEANUP = 2 } EEnvStat;
typedef void (*DndMsgFp)(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEps);
typedef struct {
EWorkerType type;
const char *name;
int32_t minNum;
int32_t maxNum;
void *queueFp;
SDnode *pDnode;
STaosQueue *queue;
union {
SWorkerPool pool;
SMWorkerPool mpool;
};
} SDnodeWorker;
typedef struct {
char *dnode;
char *mnode;
char *snode;
char *bnode;
char *vnodes;
} SDnodeDir;
typedef struct {
int32_t dnodeId;
int32_t dropped;
int64_t clusterId;
int64_t dver;
int64_t rebootTime;
int64_t updateTime;
int8_t statusSent;
SEpSet mnodeEpSet;
char *file;
SHashObj *dnodeHash;
SDnodeEps *dnodeEps;
pthread_t *threadId;
SRWLatch latch;
SDnodeWorker mgmtWorker;
} SDnodeMgmt;
typedef struct {
int32_t refCount;
int8_t deployed;
int8_t dropped;
SMnode *pMnode;
SRWLatch latch;
SDnodeWorker readWorker;
SDnodeWorker writeWorker;
SDnodeWorker syncWorker;
int8_t replica;
int8_t selfIndex;
SReplica replicas[TSDB_MAX_REPLICA];
} SMnodeMgmt;
typedef struct {
int32_t refCount;
int8_t deployed;
int8_t dropped;
SQnode *pQnode;
SRWLatch latch;
SDnodeWorker queryWorker;
SDnodeWorker fetchWorker;
} SQnodeMgmt;
typedef struct {
int32_t refCount;
int8_t deployed;
int8_t dropped;
SSnode *pSnode;
SRWLatch latch;
SDnodeWorker writeWorker;
} SSnodeMgmt;
typedef struct {
int32_t refCount;
int8_t deployed;
int8_t dropped;
SBnode *pBnode;
SRWLatch latch;
SDnodeWorker writeWorker;
} SBnodeMgmt;
typedef struct {
SHashObj *hash;
int32_t openVnodes;
int32_t totalVnodes;
SRWLatch latch;
SWorkerPool queryPool;
SWorkerPool fetchPool;
SMWorkerPool syncPool;
SMWorkerPool writePool;
} SVnodesMgmt;
typedef struct {
void *serverRpc;
void *clientRpc;
DndMsgFp msgFp[TDMT_MAX];
} STransMgmt;
typedef struct SDnode {
EStat stat;
SDnodeOpt opt;
SDnodeDir dir;
FileFd lockFd;
SDnodeMgmt dmgmt;
SMnodeMgmt mmgmt;
SQnodeMgmt qmgmt;
SSnodeMgmt smgmt;
SBnodeMgmt bmgmt;
SVnodesMgmt vmgmt;
STransMgmt tmgmt;
SStartupReq startup;
} SDnode;
EStat dndGetStat(SDnode *pDnode);
void dndSetStat(SDnode *pDnode, EStat stat);
char *dndStatStr(EStat stat);
const char *dndStatStr(EStat stat);
void dndReportStartup(SDnode *pDnode, char *pName, char *pDesc);
void dndGetStartup(SDnode *pDnode, SStartupReq *pStartup);

2
source/dnode/mgmt/impl/inc/dndMgmt.h
View File

@ -19,7 +19,7 @@
#ifdef __cplusplus
extern "C" {
#endif
#include "dndInt.h"
#include "dndEnv.h"
int32_t dndInitMgmt(SDnode *pDnode);
void dndStopMgmt(SDnode *pDnode);

2
source/dnode/mgmt/impl/inc/dndMnode.h
View File

@ -19,7 +19,7 @@
#ifdef __cplusplus
extern "C" {
#endif
#include "dndInt.h"
#include "dndEnv.h"
int32_t dndInitMnode(SDnode *pDnode);
void dndCleanupMnode(SDnode *pDnode);

2
source/dnode/mgmt/impl/inc/dndQnode.h
View File

@ -19,7 +19,7 @@
#ifdef __cplusplus
extern "C" {
#endif
#include "dndInt.h"
#include "dndEnv.h"
int32_t dndInitQnode(SDnode *pDnode);
void dndCleanupQnode(SDnode *pDnode);

2
source/dnode/mgmt/impl/inc/dndSnode.h
View File

@ -19,7 +19,7 @@
#ifdef __cplusplus
extern "C" {
#endif
#include "dndInt.h"
#include "dndEnv.h"
int32_t dndInitSnode(SDnode *pDnode);
void dndCleanupSnode(SDnode *pDnode);

2
source/dnode/mgmt/impl/inc/dndTransport.h
View File

@ -19,7 +19,7 @@
#ifdef __cplusplus
extern "C" {
#endif
#include "dndInt.h"
#include "dndEnv.h"
int32_t dndInitTrans(SDnode *pDnode);
void dndCleanupTrans(SDnode *pDnode);

4
source/dnode/mgmt/impl/inc/dndVnodes.h
View File

@ -19,7 +19,7 @@
#ifdef __cplusplus
extern "C" {
#endif
#include "dndInt.h"
#include "dndEnv.h"
int32_t dndInitVnodes(SDnode *pDnode);
void dndCleanupVnodes(SDnode *pDnode);
@ -36,6 +36,8 @@ int32_t dndProcessAuthVnodeReq(SDnode *pDnode, SRpcMsg *pReq);
int32_t dndProcessSyncVnodeReq(SDnode *pDnode, SRpcMsg *pReq);
int32_t dndProcessCompactVnodeReq(SDnode *pDnode, SRpcMsg *pReq);
int32_t dndPutReqToVQueryQ(SDnode *pDnode, SRpcMsg *pReq);
#ifdef __cplusplus
}
#endif

2
source/dnode/mgmt/impl/inc/dndWorker.h
View File

@ -19,7 +19,7 @@
#ifdef __cplusplus
extern "C" {
#endif
#include "dndInt.h"
#include "dndEnv.h"
int32_t dndInitWorker(SDnode *pDnode, SDnodeWorker *pWorker, EWorkerType type, const char *name, int32_t minNum,
int32_t maxNum, void *queueFp);

2
source/dnode/mgmt/impl/src/dndBnode.c
View File

@ -179,7 +179,7 @@ static void dndBuildBnodeOption(SDnode *pDnode, SBnodeOpt *pOption) {
pOption->sendRedirectRspFp = dndSendRedirectRsp;
pOption->dnodeId = dndGetDnodeId(pDnode);
pOption->clusterId = dndGetClusterId(pDnode);
pOption->cfg.sver = pDnode->opt.sver;
pOption->sver = pDnode->env.sver;
}
static int32_t dndOpenBnode(SDnode *pDnode) {

169
source/dnode/mgmt/impl/src/dnode.c → source/dnode/mgmt/impl/src/dndEnv.c
View File

@ -22,8 +22,10 @@
#include "dndTransport.h"
#include "dndVnodes.h"
#include "sync.h"
#include "wal.h"
#include "tfs.h"
#include "wal.h"
static SDnodeEnv dndEnv = {0};
EStat dndGetStat(SDnode *pDnode) { return pDnode->stat; }
@ -32,7 +34,7 @@ void dndSetStat(SDnode *pDnode, EStat stat) {
pDnode->stat = stat;
}
char *dndStatStr(EStat stat) {
const char *dndStatStr(EStat stat) {
switch (stat) {
case DND_STAT_INIT:
return "init";
@ -79,25 +81,26 @@ static FileFd dndCheckRunning(char *dataDir) {
return fd;
}
static int32_t dndInitEnv(SDnode *pDnode, SDnodeOpt *pOption) {
pDnode->lockFd = dndCheckRunning(pOption->dataDir);
static int32_t dndCreateImp(SDnode *pDnode, SDnodeObjCfg *pCfg) {
pDnode->lockFd = dndCheckRunning(pCfg->dataDir);
if (pDnode->lockFd < 0) {
return -1;
}
char path[PATH_MAX + 100];
snprintf(path, sizeof(path), "%s%smnode", pOption->dataDir, TD_DIRSEP);
snprintf(path, sizeof(path), "%s%smnode", pCfg->dataDir, TD_DIRSEP);
pDnode->dir.mnode = tstrdup(path);
snprintf(path, sizeof(path), "%s%svnode", pOption->dataDir, TD_DIRSEP);
snprintf(path, sizeof(path), "%s%svnode", pCfg->dataDir, TD_DIRSEP);
pDnode->dir.vnodes = tstrdup(path);
snprintf(path, sizeof(path), "%s%sdnode", pOption->dataDir, TD_DIRSEP);
snprintf(path, sizeof(path), "%s%sdnode", pCfg->dataDir, TD_DIRSEP);
pDnode->dir.dnode = tstrdup(path);
snprintf(path, sizeof(path), "%s%ssnode", pOption->dataDir, TD_DIRSEP);
snprintf(path, sizeof(path), "%s%ssnode", pCfg->dataDir, TD_DIRSEP);
pDnode->dir.snode = tstrdup(path);
snprintf(path, sizeof(path), "%s%sbnode", pOption->dataDir, TD_DIRSEP);
snprintf(path, sizeof(path), "%s%sbnode", pCfg->dataDir, TD_DIRSEP);
pDnode->dir.bnode = tstrdup(path);
if (pDnode->dir.mnode == NULL || pDnode->dir.vnodes == NULL || pDnode->dir.dnode == NULL) {
if (pDnode->dir.mnode == NULL || pDnode->dir.vnodes == NULL || pDnode->dir.dnode == NULL ||
pDnode->dir.snode == NULL || pDnode->dir.bnode == NULL) {
dError("failed to malloc dir object");
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
@ -133,11 +136,12 @@ static int32_t dndInitEnv(SDnode *pDnode, SDnodeOpt *pOption) {
return -1;
}
memcpy(&pDnode->opt, pOption, sizeof(SDnodeOpt));
memcpy(&pDnode->cfg, pCfg, sizeof(SDnodeObjCfg));
memcpy(&pDnode->env, &dndEnv.cfg, sizeof(SDnodeEnvCfg));
return 0;
}
static void dndCleanupEnv(SDnode *pDnode) {
static void dndCloseImp(SDnode *pDnode) {
tfree(pDnode->dir.mnode);
tfree(pDnode->dir.vnodes);
tfree(pDnode->dir.dnode);
@ -149,118 +153,95 @@ static void dndCleanupEnv(SDnode *pDnode) {
taosCloseFile(pDnode->lockFd);
pDnode->lockFd = 0;
}
taosStopCacheRefreshWorker();
}
SDnode *dndInit(SDnodeOpt *pOption) {
taosIgnSIGPIPE();
taosBlockSIGPIPE();
taosResolveCRC();
SDnode *dndCreate(SDnodeObjCfg *pCfg) {
dInfo("start to create dnode object");
SDnode *pDnode = calloc(1, sizeof(SDnode));
if (pDnode == NULL) {
dError("failed to create dnode object");
terrno = TSDB_CODE_OUT_OF_MEMORY;
dError("failed to create dnode object since %s", terrstr());
return NULL;
}
dInfo("start to initialize TDengine");
dndSetStat(pDnode, DND_STAT_INIT);
if (dndInitEnv(pDnode, pOption) != 0) {
dError("failed to init env");
dndCleanup(pDnode);
return NULL;
}
if (rpcInit() != 0) {
dError("failed to init rpc env");
dndCleanup(pDnode);
return NULL;
}
if (walInit() != 0) {
dError("failed to init wal env");
dndCleanup(pDnode);
if (dndCreateImp(pDnode, pCfg) != 0) {
dError("failed to init dnode dir since %s", terrstr());
dndClose(pDnode);
return NULL;
}
SDiskCfg dCfg;
strcpy(dCfg.dir, pDnode->opt.dataDir);
strcpy(dCfg.dir, pDnode->cfg.dataDir);
dCfg.level = 0;
dCfg.primary = 1;
if (tfsInit(&dCfg, 1) != 0) {
dError("failed to init tfs env");
dndCleanup(pDnode);
return NULL;
}
if (vnodeInit(pDnode->opt.numOfCommitThreads) != 0) {
dError("failed to init vnode env");
dndCleanup(pDnode);
dError("failed to init tfs since %s", terrstr());
dndClose(pDnode);
return NULL;
}
if (dndInitMgmt(pDnode) != 0) {
dError("failed to init dnode");
dndCleanup(pDnode);
dError("failed to init mgmt since %s", terrstr());
dndClose(pDnode);
return NULL;
}
if (dndInitVnodes(pDnode) != 0) {
dError("failed to init vnodes");
dndCleanup(pDnode);
dError("failed to init vnodes since %s", terrstr());
dndClose(pDnode);
return NULL;
}
if (dndInitQnode(pDnode) != 0) {
dError("failed to init qnode");
dndCleanup(pDnode);
dError("failed to init qnode since %s", terrstr());
dndClose(pDnode);
return NULL;
}
if (dndInitSnode(pDnode) != 0) {
dError("failed to init snode");
dndCleanup(pDnode);
dError("failed to init snode since %s", terrstr());
dndClose(pDnode);
return NULL;
}
if (dndInitBnode(pDnode) != 0) {
dError("failed to init bnode");
dndCleanup(pDnode);
dError("failed to init bnode since %s", terrstr());
dndClose(pDnode);
return NULL;
}
if (dndInitMnode(pDnode) != 0) {
dError("failed to init mnode");
dndCleanup(pDnode);
dError("failed to init mnode since %s", terrstr());
dndClose(pDnode);
return NULL;
}
if (dndInitTrans(pDnode) != 0) {
dError("failed to init transport");
dndCleanup(pDnode);
dError("failed to init transport since %s", terrstr());
dndClose(pDnode);
return NULL;
}
dndSetStat(pDnode, DND_STAT_RUNNING);
dndSendStatusReq(pDnode);
dndReportStartup(pDnode, "TDengine", "initialized successfully");
dInfo("TDengine is initialized successfully, pDnode:%p", pDnode);
dInfo("dnode object is created, data:%p", pDnode);
return pDnode;
}
void dndCleanup(SDnode *pDnode) {
void dndClose(SDnode *pDnode) {
if (pDnode == NULL) return;
if (dndGetStat(pDnode) == DND_STAT_STOPPED) {
dError("dnode is shutting down");
dError("dnode is shutting down, data:%p", pDnode);
return;
}
dInfo("start to cleanup TDengine");
dInfo("start to close dnode, data:%p", pDnode);
dndSetStat(pDnode, DND_STAT_STOPPED);
dndCleanupTrans(pDnode);
dndStopMgmt(pDnode);
@ -270,12 +251,66 @@ void dndCleanup(SDnode *pDnode) {
dndCleanupQnode(pDnode);
dndCleanupVnodes(pDnode);
dndCleanupMgmt(pDnode);
vnodeClear();
tfsDestroy();
tfsCleanup();
dndCloseImp(pDnode);
free(pDnode);
dInfo("dnode object is closed, data:%p", pDnode);
}
int32_t dndInit(const SDnodeEnvCfg *pCfg) {
if (atomic_val_compare_exchange_8(&dndEnv.once, DND_ENV_INIT, DND_ENV_READY) != DND_ENV_INIT) {
terrno = TSDB_CODE_REPEAT_INIT;
dError("failed to init dnode env since %s", terrstr());
return -1;
}
taosIgnSIGPIPE();
taosBlockSIGPIPE();
taosResolveCRC();
if (rpcInit() != 0) {
dError("failed to init rpc since %s", terrstr());
dndCleanup();
return -1;
}
if (walInit() != 0) {
dError("failed to init wal since %s", terrstr());
dndCleanup();
return -1;
}
SVnodeOpt vnodeOpt = {
.sver = pCfg->sver,
.timezone = pCfg->timezone,
.locale = pCfg->locale,
.charset = pCfg->charset,
.nthreads = pCfg->numOfCommitThreads,
.putReqToVQueryQFp = dndPutReqToVQueryQ,
};
if (vnodeInit(&vnodeOpt) != 0) {
dError("failed to init vnode since %s", terrstr());
dndCleanup();
return -1;
}
memcpy(&dndEnv.cfg, pCfg, sizeof(SDnodeEnvCfg));
dInfo("dnode env is initialized");
return 0;
}
void dndCleanup() {
if (atomic_val_compare_exchange_8(&dndEnv.once, DND_ENV_READY, DND_ENV_CLEANUP) != DND_ENV_READY) {
dError("dnode env is already cleaned up");
return;
}
walCleanUp();
vnodeCleanup();
rpcCleanup();
dndCleanupEnv(pDnode);
free(pDnode);
dInfo("TDengine is cleaned up successfully");
taosStopCacheRefreshWorker();
dInfo("dnode env is cleaned up");
}

39
source/dnode/mgmt/impl/src/dndMgmt.c
View File

@ -86,7 +86,7 @@ void dndSendRedirectRsp(SDnode *pDnode, SRpcMsg *pReq) {
dDebug("RPC %p, req:%s is redirected, num:%d use:%d", pReq->handle, TMSG_INFO(msgType), epSet.numOfEps, epSet.inUse);
for (int32_t i = 0; i < epSet.numOfEps; ++i) {
dDebug("mnode index:%d %s:%u", i, epSet.fqdn[i], epSet.port[i]);
if (strcmp(epSet.fqdn[i], pDnode->opt.localFqdn) == 0 && epSet.port[i] == pDnode->opt.serverPort) {
if (strcmp(epSet.fqdn[i], pDnode->cfg.localFqdn) == 0 && epSet.port[i] == pDnode->cfg.serverPort) {
epSet.inUse = (i + 1) % epSet.numOfEps;
}
@ -289,8 +289,8 @@ PRASE_DNODE_OVER:
if (root != NULL) cJSON_Delete(root);
if (fp != NULL) fclose(fp);
if (dndIsEpChanged(pDnode, pMgmt->dnodeId, pDnode->opt.localEp)) {
dError("localEp %s different with %s and need reconfigured", pDnode->opt.localEp, pMgmt->file);
if (dndIsEpChanged(pDnode, pMgmt->dnodeId, pDnode->cfg.localEp)) {
dError("localEp %s different with %s and need reconfigured", pDnode->cfg.localEp, pMgmt->file);
return -1;
}
@ -298,7 +298,7 @@ PRASE_DNODE_OVER:
pMgmt->dnodeEps = calloc(1, sizeof(SDnodeEps) + sizeof(SDnodeEp));
pMgmt->dnodeEps->num = 1;
pMgmt->dnodeEps->eps[0].isMnode = 1;
taosGetFqdnPortFromEp(pDnode->opt.firstEp, pMgmt->dnodeEps->eps[0].fqdn, &pMgmt->dnodeEps->eps[0].port);
taosGetFqdnPortFromEp(pDnode->cfg.firstEp, pMgmt->dnodeEps->eps[0].fqdn, &pMgmt->dnodeEps->eps[0].port);
}
dndResetDnodes(pDnode, pMgmt->dnodeEps);
@ -362,24 +362,24 @@ void dndSendStatusReq(SDnode *pDnode) {
SDnodeMgmt *pMgmt = &pDnode->dmgmt;
taosRLockLatch(&pMgmt->latch);
pStatus->sver = htonl(pDnode->opt.sver);
pStatus->sver = htonl(pDnode->env.sver);
pStatus->dver = htobe64(pMgmt->dver);
pStatus->dnodeId = htonl(pMgmt->dnodeId);
pStatus->clusterId = htobe64(pMgmt->clusterId);
pStatus->rebootTime = htobe64(pMgmt->rebootTime);
pStatus->updateTime = htobe64(pMgmt->updateTime);
pStatus->numOfCores = htonl(pDnode->opt.numOfCores);
pStatus->numOfSupportVnodes = htonl(pDnode->opt.numOfSupportVnodes);
tstrncpy(pStatus->dnodeEp, pDnode->opt.localEp, TSDB_EP_LEN);
pStatus->numOfCores = htonl(pDnode->env.numOfCores);
pStatus->numOfSupportVnodes = htonl(pDnode->cfg.numOfSupportVnodes);
tstrncpy(pStatus->dnodeEp, pDnode->cfg.localEp, TSDB_EP_LEN);
pStatus->clusterCfg.statusInterval = htonl(pDnode->opt.statusInterval);
pStatus->clusterCfg.statusInterval = htonl(pDnode->cfg.statusInterval);
pStatus->clusterCfg.checkTime = 0;
char timestr[32] = "1970-01-01 00:00:00.00";
(void)taosParseTime(timestr, &pStatus->clusterCfg.checkTime, (int32_t)strlen(timestr), TSDB_TIME_PRECISION_MILLI, 0);
pStatus->clusterCfg.checkTime = htonl(pStatus->clusterCfg.checkTime);
tstrncpy(pStatus->clusterCfg.timezone, pDnode->opt.timezone, TSDB_TIMEZONE_LEN);
tstrncpy(pStatus->clusterCfg.locale, pDnode->opt.locale, TSDB_LOCALE_LEN);
tstrncpy(pStatus->clusterCfg.charset, pDnode->opt.charset, TSDB_LOCALE_LEN);
tstrncpy(pStatus->clusterCfg.timezone, pDnode->env.timezone, TSDB_TIMEZONE_LEN);
tstrncpy(pStatus->clusterCfg.locale, pDnode->env.locale, TSDB_LOCALE_LEN);
tstrncpy(pStatus->clusterCfg.charset, pDnode->env.charset, TSDB_LOCALE_LEN);
taosRUnLockLatch(&pMgmt->latch);
dndGetVnodeLoads(pDnode, &pStatus->vnodeLoads);
@ -485,7 +485,7 @@ void dndProcessStartupReq(SDnode *pDnode, SRpcMsg *pReq) {
static void *dnodeThreadRoutine(void *param) {
SDnode *pDnode = param;
SDnodeMgmt *pMgmt = &pDnode->dmgmt;
int32_t ms = pDnode->opt.statusInterval * 1000;
int32_t ms = pDnode->cfg.statusInterval * 1000;
while (true) {
pthread_testcancel();
@ -536,6 +536,11 @@ int32_t dndInitMgmt(SDnode *pDnode) {
return -1;
}
if (dndInitWorker(pDnode, &pMgmt->statusWorker, DND_WORKER_SINGLE, "dnode-status", 1, 1, dndProcessMgmtQueue) != 0) {
dError("failed to start dnode mgmt worker since %s", terrstr());
return -1;
}
pMgmt->threadId = taosCreateThread(dnodeThreadRoutine, pDnode);
if (pMgmt->threadId == NULL) {
dError("failed to init dnode thread");
@ -550,6 +555,7 @@ int32_t dndInitMgmt(SDnode *pDnode) {
void dndStopMgmt(SDnode *pDnode) {
SDnodeMgmt *pMgmt = &pDnode->dmgmt;
dndCleanupWorker(&pMgmt->mgmtWorker);
dndCleanupWorker(&pMgmt->statusWorker);
if (pMgmt->threadId != NULL) {
taosDestoryThread(pMgmt->threadId);
@ -587,7 +593,12 @@ void dndProcessMgmtMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
dndUpdateMnodeEpSet(pDnode, pEpSet);
}
if (dndWriteMsgToWorker(&pMgmt->mgmtWorker, pMsg, sizeof(SRpcMsg)) != 0) {
SDnodeWorker *pWorker = &pMgmt->mgmtWorker;
if (pMsg->msgType == TDMT_MND_STATUS_RSP) {
pWorker = &pMgmt->statusWorker;
}
if (dndWriteMsgToWorker(pWorker, pMsg, sizeof(SRpcMsg)) != 0) {
if (pMsg->msgType & 1u) {
SRpcMsg rsp = {.handle = pMsg->handle, .code = TSDB_CODE_OUT_OF_MEMORY};
rpcSendResponse(&rsp);

24
source/dnode/mgmt/impl/src/dndMnode.c
View File

@ -247,7 +247,7 @@ static bool dndNeedDeployMnode(SDnode *pDnode) {
return false;
}
if (strcmp(pDnode->opt.localEp, pDnode->opt.firstEp) != 0) {
if (strcmp(pDnode->cfg.localEp, pDnode->cfg.firstEp) != 0) {
return false;
}
@ -266,15 +266,15 @@ static void dndInitMnodeOption(SDnode *pDnode, SMnodeOpt *pOption) {
pOption->putReqToMWriteQFp = dndPutMsgToMWriteQ;
pOption->dnodeId = dndGetDnodeId(pDnode);
pOption->clusterId = dndGetClusterId(pDnode);
pOption->cfg.sver = pDnode->opt.sver;
pOption->cfg.enableTelem = pDnode->opt.enableTelem;
pOption->cfg.statusInterval = pDnode->opt.statusInterval;
pOption->cfg.shellActivityTimer = pDnode->opt.shellActivityTimer;
pOption->cfg.timezone = pDnode->opt.timezone;
pOption->cfg.charset = pDnode->opt.charset;
pOption->cfg.locale = pDnode->opt.locale;
pOption->cfg.gitinfo = pDnode->opt.gitinfo;
pOption->cfg.buildinfo = pDnode->opt.buildinfo;
pOption->cfg.sver = pDnode->env.sver;
pOption->cfg.enableTelem = pDnode->env.enableTelem;
pOption->cfg.statusInterval = pDnode->cfg.statusInterval;
pOption->cfg.shellActivityTimer = pDnode->cfg.shellActivityTimer;
pOption->cfg.timezone = pDnode->env.timezone;
pOption->cfg.charset = pDnode->env.charset;
pOption->cfg.locale = pDnode->env.locale;
pOption->cfg.gitinfo = pDnode->env.gitinfo;
pOption->cfg.buildinfo = pDnode->env.buildinfo;
}
static void dndBuildMnodeDeployOption(SDnode *pDnode, SMnodeOpt *pOption) {
@ -283,8 +283,8 @@ static void dndBuildMnodeDeployOption(SDnode *pDnode, SMnodeOpt *pOption) {
pOption->selfIndex = 0;
SReplica *pReplica = &pOption->replicas[0];
pReplica->id = 1;
pReplica->port = pDnode->opt.serverPort;
memcpy(pReplica->fqdn, pDnode->opt.localFqdn, TSDB_FQDN_LEN);
pReplica->port = pDnode->cfg.serverPort;
memcpy(pReplica->fqdn, pDnode->cfg.localFqdn, TSDB_FQDN_LEN);
SMnodeMgmt *pMgmt = &pDnode->mmgmt;
pMgmt->selfIndex = pOption->selfIndex;

2
source/dnode/mgmt/impl/src/dndQnode.c
View File

@ -185,7 +185,7 @@ static void dndBuildQnodeOption(SDnode *pDnode, SQnodeOpt *pOption) {
pOption->sendRedirectRspFp = dndSendRedirectRsp;
pOption->dnodeId = dndGetDnodeId(pDnode);
pOption->clusterId = dndGetClusterId(pDnode);
pOption->cfg.sver = pDnode->opt.sver;
pOption->sver = pDnode->env.sver;
}
static int32_t dndOpenQnode(SDnode *pDnode) {

2
source/dnode/mgmt/impl/src/dndSnode.c
View File

@ -179,7 +179,7 @@ static void dndBuildSnodeOption(SDnode *pDnode, SSnodeOpt *pOption) {
pOption->sendRedirectRspFp = dndSendRedirectRsp;
pOption->dnodeId = dndGetDnodeId(pDnode);
pOption->clusterId = dndGetClusterId(pDnode);
pOption->cfg.sver = pDnode->opt.sver;
pOption->sver = pDnode->env.sver;
}
static int32_t dndOpenSnode(SDnode *pDnode) {

10
source/dnode/mgmt/impl/src/dndTransport.c
View File

@ -176,7 +176,7 @@ static int32_t dndInitClient(SDnode *pDnode) {
rpcInit.cfp = dndProcessResponse;
rpcInit.sessions = 1024;
rpcInit.connType = TAOS_CONN_CLIENT;
rpcInit.idleTime = pDnode->opt.shellActivityTimer * 1000;
rpcInit.idleTime = pDnode->cfg.shellActivityTimer * 1000;
rpcInit.user = INTERNAL_USER;
rpcInit.ckey = INTERNAL_CKEY;
rpcInit.secret = INTERNAL_SECRET;
@ -325,20 +325,20 @@ static int32_t dndInitServer(SDnode *pDnode) {
STransMgmt *pMgmt = &pDnode->tmgmt;
dndInitMsgFp(pMgmt);
int32_t numOfThreads = (int32_t)((pDnode->opt.numOfCores * pDnode->opt.numOfThreadsPerCore) / 2.0);
int32_t numOfThreads = (int32_t)((pDnode->env.numOfCores * pDnode->cfg.numOfThreadsPerCore) / 2.0);
if (numOfThreads < 1) {
numOfThreads = 1;
}
SRpcInit rpcInit;
memset(&rpcInit, 0, sizeof(rpcInit));
rpcInit.localPort = pDnode->opt.serverPort;
rpcInit.localPort = pDnode->cfg.serverPort;
rpcInit.label = "DND-S";
rpcInit.numOfThreads = numOfThreads;
rpcInit.cfp = dndProcessRequest;
rpcInit.sessions = pDnode->opt.maxShellConns;
rpcInit.sessions = pDnode->cfg.maxShellConns;
rpcInit.connType = TAOS_CONN_SERVER;
rpcInit.idleTime = pDnode->opt.shellActivityTimer * 1000;
rpcInit.idleTime = pDnode->cfg.shellActivityTimer * 1000;
rpcInit.afp = dndRetrieveUserAuthInfo;
rpcInit.parent = pDnode;

42
source/dnode/mgmt/impl/src/dndVnodes.c
View File

@ -381,7 +381,8 @@ static void *dnodeOpenVnodeFunc(void *param) {
pMgmt->openVnodes, pMgmt->totalVnodes);
dndReportStartup(pDnode, "open-vnodes", stepDesc);
SVnode *pImpl = vnodeOpen(pCfg->path, NULL, pCfg->vgId);
SVnodeCfg cfg = {.pDnode = pDnode, .vgId = pCfg->vgId};
SVnode *pImpl = vnodeOpen(pCfg->path, &cfg);
if (pImpl == NULL) {
dError("vgId:%d, failed to open vnode by thread:%d", pCfg->vgId, pThread->threadIndex);
pThread->failed++;
@ -419,7 +420,7 @@ static int32_t dndOpenVnodes(SDnode *pDnode) {
pMgmt->totalVnodes = numOfVnodes;
int32_t threadNum = pDnode->opt.numOfCores;
int32_t threadNum = pDnode->env.numOfCores;
int32_t vnodesPerThread = numOfVnodes / threadNum + 1;
SVnodeThread *threads = calloc(threadNum, sizeof(SVnodeThread));
@ -581,7 +582,8 @@ int32_t dndProcessCreateVnodeReq(SDnode *pDnode, SRpcMsg *pReq) {
return -1;
}
SVnode *pImpl = vnodeOpen(wrapperCfg.path, NULL /*pCfg*/, pCreate->vgId);
vnodeCfg.pDnode = pDnode;
SVnode *pImpl = vnodeOpen(wrapperCfg.path, &vnodeCfg);
if (pImpl == NULL) {
dError("vgId:%d, failed to create vnode since %s", pCreate->vgId, terrstr());
return -1;
@ -800,7 +802,7 @@ static void dndProcessVnodeSyncQueue(SVnodeObj *pVnode, STaosQall *qall, int32_t
}
}
static int32_t dndWriteRpcMsgToVnodeQueue(STaosQueue *pQueue, SRpcMsg *pRpcMsg) {
static int32_t dndWriteRpcMsgToVnodeQueue(STaosQueue *pQueue, SRpcMsg *pRpcMsg, bool sendRsp) {
int32_t code = 0;
if (pQueue == NULL) {
@ -817,13 +819,15 @@ static int32_t dndWriteRpcMsgToVnodeQueue(STaosQueue *pQueue, SRpcMsg *pRpcMsg)
}
}
if (code != TSDB_CODE_SUCCESS) {
if (code != TSDB_CODE_SUCCESS && sendRsp) {
if (pRpcMsg->msgType & 1u) {
SRpcMsg rsp = {.handle = pRpcMsg->handle, .code = code};
rpcSendResponse(&rsp);
}
rpcFreeCont(pRpcMsg->pCont);
}
return code;
}
static SVnodeObj *dndAcquireVnodeFromMsg(SDnode *pDnode, SRpcMsg *pMsg) {
@ -846,7 +850,7 @@ static SVnodeObj *dndAcquireVnodeFromMsg(SDnode *pDnode, SRpcMsg *pMsg) {
void dndProcessVnodeWriteMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
SVnodeObj *pVnode = dndAcquireVnodeFromMsg(pDnode, pMsg);
if (pVnode != NULL) {
dndWriteRpcMsgToVnodeQueue(pVnode->pWriteQ, pMsg);
(void)dndWriteRpcMsgToVnodeQueue(pVnode->pWriteQ, pMsg, true);
dndReleaseVnode(pDnode, pVnode);
}
}
@ -854,7 +858,7 @@ void dndProcessVnodeWriteMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
void dndProcessVnodeSyncMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
SVnodeObj *pVnode = dndAcquireVnodeFromMsg(pDnode, pMsg);
if (pVnode != NULL) {
dndWriteRpcMsgToVnodeQueue(pVnode->pSyncQ, pMsg);
(void)dndWriteRpcMsgToVnodeQueue(pVnode->pSyncQ, pMsg, true);
dndReleaseVnode(pDnode, pVnode);
}
}
@ -862,7 +866,7 @@ void dndProcessVnodeSyncMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
void dndProcessVnodeQueryMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
SVnodeObj *pVnode = dndAcquireVnodeFromMsg(pDnode, pMsg);
if (pVnode != NULL) {
dndWriteRpcMsgToVnodeQueue(pVnode->pQueryQ, pMsg);
(void)dndWriteRpcMsgToVnodeQueue(pVnode->pQueryQ, pMsg, true);
dndReleaseVnode(pDnode, pVnode);
}
}
@ -870,11 +874,23 @@ void dndProcessVnodeQueryMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
void dndProcessVnodeFetchMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
SVnodeObj *pVnode = dndAcquireVnodeFromMsg(pDnode, pMsg);
if (pVnode != NULL) {
dndWriteRpcMsgToVnodeQueue(pVnode->pFetchQ, pMsg);
(void)dndWriteRpcMsgToVnodeQueue(pVnode->pFetchQ, pMsg, true);
dndReleaseVnode(pDnode, pVnode);
}
}
int32_t dndPutReqToVQueryQ(SDnode *pDnode, SRpcMsg *pMsg) {
SMsgHead *pHead = pMsg->pCont;
// pHead->vgId = htonl(pHead->vgId);
SVnodeObj *pVnode = dndAcquireVnode(pDnode, pHead->vgId);
if (pVnode == NULL) return -1;
int32_t code = dndWriteRpcMsgToVnodeQueue(pVnode->pFetchQ, pMsg, false);
dndReleaseVnode(pDnode, pVnode);
return code;
}
static int32_t dndPutMsgIntoVnodeApplyQueue(SDnode *pDnode, int32_t vgId, SRpcMsg *pMsg) {
SVnodeObj *pVnode = dndAcquireVnode(pDnode, vgId);
if (pVnode == NULL) return -1;
@ -888,11 +904,11 @@ static int32_t dndInitVnodeWorkers(SDnode *pDnode) {
SVnodesMgmt *pMgmt = &pDnode->vmgmt;
int32_t maxFetchThreads = 4;
int32_t minFetchThreads = MIN(maxFetchThreads, pDnode->opt.numOfCores);
int32_t minQueryThreads = MAX((int32_t)(pDnode->opt.numOfCores * pDnode->opt.ratioOfQueryCores), 1);
int32_t minFetchThreads = MIN(maxFetchThreads, pDnode->env.numOfCores);
int32_t minQueryThreads = MAX((int32_t)(pDnode->env.numOfCores * pDnode->cfg.ratioOfQueryCores), 1);
int32_t maxQueryThreads = minQueryThreads;
int32_t maxWriteThreads = MAX(pDnode->opt.numOfCores, 1);
int32_t maxSyncThreads = MAX(pDnode->opt.numOfCores / 2, 1);
int32_t maxWriteThreads = MAX(pDnode->env.numOfCores, 1);
int32_t maxSyncThreads = MAX(pDnode->env.numOfCores / 2, 1);
SWorkerPool *pPool = &pMgmt->queryPool;
pPool->name = "vnode-query";

2
source/dnode/mgmt/impl/test/sut/inc/server.h
View File

@ -24,7 +24,7 @@ class TestServer {
bool DoStart();
private:
SDnodeOpt BuildOption(const char* path, const char* fqdn, uint16_t port, const char* firstEp);
SDnodeObjCfg BuildOption(const char* path, const char* fqdn, uint16_t port, const char* firstEp);
private:
SDnode* pDnode;

37
source/dnode/mgmt/impl/test/sut/src/server.cpp
View File

@ -22,30 +22,27 @@ void* serverLoop(void* param) {
}
}
SDnodeOpt TestServer::BuildOption(const char* path, const char* fqdn, uint16_t port, const char* firstEp) {
SDnodeOpt option = {0};
option.sver = 1;
option.numOfCores = 1;
option.numOfSupportVnodes = 16;
option.numOfCommitThreads = 1;
option.statusInterval = 1;
option.numOfThreadsPerCore = 1;
option.ratioOfQueryCores = 1;
option.maxShellConns = 1000;
option.shellActivityTimer = 30;
option.serverPort = port;
strcpy(option.dataDir, path);
snprintf(option.localEp, TSDB_EP_LEN, "%s:%u", fqdn, port);
snprintf(option.localFqdn, TSDB_FQDN_LEN, "%s", fqdn);
snprintf(option.firstEp, TSDB_EP_LEN, "%s", firstEp);
return option;
SDnodeObjCfg TestServer::BuildOption(const char* path, const char* fqdn, uint16_t port, const char* firstEp) {
SDnodeObjCfg cfg = {0};
cfg.numOfSupportVnodes = 16;
cfg.statusInterval = 1;
cfg.numOfThreadsPerCore = 1;
cfg.ratioOfQueryCores = 1;
cfg.maxShellConns = 1000;
cfg.shellActivityTimer = 30;
cfg.serverPort = port;
strcpy(cfg.dataDir, path);
snprintf(cfg.localEp, TSDB_EP_LEN, "%s:%u", fqdn, port);
snprintf(cfg.localFqdn, TSDB_FQDN_LEN, "%s", fqdn);
snprintf(cfg.firstEp, TSDB_EP_LEN, "%s", firstEp);
return cfg;
}
bool TestServer::DoStart() {
SDnodeOpt option = BuildOption(path, fqdn, port, firstEp);
SDnodeObjCfg cfg = BuildOption(path, fqdn, port, firstEp);
taosMkDir(path);
pDnode = dndInit(&option);
pDnode = dndCreate(&cfg);
if (pDnode != NULL) {
return false;
}
@ -81,7 +78,7 @@ void TestServer::Stop() {
}
if (pDnode != NULL) {
dndCleanup(pDnode);
dndClose(pDnode);
pDnode = NULL;
}
}

6
source/dnode/mgmt/impl/test/sut/src/sut.cpp
View File

@ -43,6 +43,11 @@ void Testbase::InitLog(const char* path) {
}
void Testbase::Init(const char* path, int16_t port) {
SDnodeEnvCfg cfg = {0};
cfg.numOfCommitThreads = 1;
cfg.numOfCores = 1;
dndInit(&cfg);
char fqdn[] = "localhost";
char firstEp[TSDB_EP_LEN] = {0};
snprintf(firstEp, TSDB_EP_LEN, "%s:%u", fqdn, port);
@ -56,6 +61,7 @@ void Testbase::Init(const char* path, int16_t port) {
void Testbase::Cleanup() {
server.Stop();
client.Cleanup();
dndCleanup();
}
void Testbase::Restart() { server.Restart(); }

2
source/dnode/mnode/impl/src/mndDb.c
View File

@ -740,7 +740,7 @@ static int32_t mndBuildDropVgroupAction(SMnode *pMnode, STrans *pTrans, SDbObj *
if (pReq == NULL) return -1;
action.pCont = pReq;
action.contLen = sizeof(SCreateVnodeReq);
action.contLen = sizeof(SDropVnodeReq);
action.msgType = TDMT_DND_DROP_VNODE;
action.acceptableCode = TSDB_CODE_DND_VNODE_NOT_DEPLOYED;
if (mndTransAppendRedoAction(pTrans, &action) != 0) {

8
source/dnode/mnode/impl/src/mndVgroup.c
View File

@ -319,6 +319,14 @@ static int32_t mndGetAvailableDnode(SMnode *pMnode, SVgObj *pVgroup, SArray *pAr
taosArraySort(pArray, (__compar_fn_t)mndCompareDnodeVnodes);
int32_t size = taosArrayGetSize(pArray);
if (size < pVgroup->replica) {
mError("db:%s, vgId:%d, no enough online dnodes:%d to alloc %d replica", pVgroup->dbName, pVgroup->vgId, size,
pVgroup->replica);
terrno = TSDB_CODE_MND_NO_ENOUGH_DNODES;
return -1;
}
for (int32_t v = 0; v < pVgroup->replica; ++v) {
SVnodeGid *pVgid = &pVgroup->vnodeGid[v];
SDnodeObj *pDnode = taosArrayGet(pArray, v);

10
source/dnode/mnode/impl/test/profile/profile.cpp
View File

@ -13,7 +13,7 @@
class MndTestProfile : public ::testing::Test {
protected:
static void SetUpTestSuite() { test.Init("/tmp/mnode_test_profile", 9022); }
static void SetUpTestSuite() { test.Init("/tmp/mnode_test_profile", 9031); }
static void TearDownTestSuite() { test.Cleanup(); }
static Testbase test;
@ -53,7 +53,7 @@ TEST_F(MndTestProfile, 01_ConnectMsg) {
EXPECT_EQ(pRsp->epSet.inUse, 0);
EXPECT_EQ(pRsp->epSet.numOfEps, 1);
EXPECT_EQ(pRsp->epSet.port[0], 9022);
EXPECT_EQ(pRsp->epSet.port[0], 9031);
EXPECT_STREQ(pRsp->epSet.fqdn[0], "localhost");
connId = pRsp->connId;
@ -127,7 +127,7 @@ TEST_F(MndTestProfile, 04_HeartBeatMsg) {
EXPECT_EQ(pRsp->epSet.inUse, 0);
EXPECT_EQ(pRsp->epSet.numOfEps, 1);
EXPECT_EQ(pRsp->epSet.port[0], 9022);
EXPECT_EQ(pRsp->epSet.port[0], 9031);
EXPECT_STREQ(pRsp->epSet.fqdn[0], "localhost");
}
@ -185,7 +185,7 @@ TEST_F(MndTestProfile, 05_KillConnMsg) {
EXPECT_EQ(pRsp->epSet.inUse, 0);
EXPECT_EQ(pRsp->epSet.numOfEps, 1);
EXPECT_EQ(pRsp->epSet.port[0], 9022);
EXPECT_EQ(pRsp->epSet.port[0], 9031);
EXPECT_STREQ(pRsp->epSet.fqdn[0], "localhost");
connId = pRsp->connId;
@ -249,7 +249,7 @@ TEST_F(MndTestProfile, 07_KillQueryMsg) {
EXPECT_EQ(pRsp->epSet.inUse, 0);
EXPECT_EQ(pRsp->epSet.numOfEps, 1);
EXPECT_EQ(pRsp->epSet.port[0], 9022);
EXPECT_EQ(pRsp->epSet.port[0], 9031);
EXPECT_STREQ(pRsp->epSet.fqdn[0], "localhost");
}
}

7
source/dnode/qnode/inc/qndInt.h
View File

@ -29,12 +29,7 @@ extern "C" {
#endif
typedef struct SQnode {
int32_t dnodeId;
int64_t clusterId;
SQnodeCfg cfg;
SendReqToDnodeFp sendReqToDnodeFp;
SendReqToMnodeFp sendReqToMnodeFp;
SendRedirectRspFp sendRedirectRspFp;
SQnodeOpt opt;
} SQnode;
#ifdef __cplusplus

7
source/dnode/snode/inc/sndInt.h
View File

@ -29,12 +29,7 @@ extern "C" {
#endif
typedef struct SSnode {
int32_t dnodeId;
int64_t clusterId;
SSnodeCfg cfg;
SendReqToDnodeFp sendReqToDnodeFp;
SendReqToMnodeFp sendReqToMnodeFp;
SendRedirectRspFp sendRedirectRspFp;
SSnodeOpt cfg;
} SSnode;
#ifdef __cplusplus

37
source/dnode/vnode/CMakeLists.txt
View File

@ -1,4 +1,33 @@
add_subdirectory(meta)
add_subdirectory(tq)
add_subdirectory(tsdb)
add_subdirectory(impl)
aux_source_directory(src/meta META_SRC)
aux_source_directory(src/tq TQ_SRC)
aux_source_directory(src/tsdb TSDB_SRC)
aux_source_directory(src/vnd VND_SRC)
list(APPEND
VNODE_SRC
${META_SRC}
${TQ_SRC}
${TSDB_SRC}
${VND_SRC}
)
add_library(vnode STATIC ${VNODE_SRC})
target_include_directories(
vnode
PUBLIC inc
PRIVATE src/inc
)
target_link_libraries(
vnode
PUBLIC os
PUBLIC util
PUBLIC common
PUBLIC transport
PUBLIC bdb
PUBLIC tfs
PUBLIC wal
PUBLIC qworker
)
if(${BUILD_TEST})
# add_subdirectory(test)
endif(${BUILD_TEST})

24
source/dnode/vnode/impl/CMakeLists.txt
View File

@ -1,24 +0,0 @@
aux_source_directory(src VNODE_SRC)
add_library(vnode STATIC ${VNODE_SRC})
target_include_directories(
vnode
PUBLIC "${CMAKE_SOURCE_DIR}/include/dnode/vnode"
private "${CMAKE_CURRENT_SOURCE_DIR}/inc"
)
target_link_libraries(
vnode
PUBLIC os
PUBLIC transport
PUBLIC meta
PUBLIC tq
PUBLIC tsdb
PUBLIC wal
PUBLIC sync
PUBLIC cjson
PUBLIC qworker
)
# test
if(${BUILD_TEST})
# add_subdirectory(test)
endif(${BUILD_TEST})

30
source/dnode/vnode/impl/inc/vnodeRead.h
View File

@ -1,30 +0,0 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef _TD_VNODE_READ_H_
#define _TD_VNODE_READ_H_
#ifdef __cplusplus
extern "C" {
#endif
#include "vnodeInt.h"
void vnodeProcessReadMsg(SVnode *pVnode, SVnodeMsg *pMsg);
#ifdef __cplusplus
}
#endif
#endif /*_TD_VNODE_READ_H_*/

16
source/dnode/vnode/impl/src/vnodeFS.c
View File

@ -1,16 +0,0 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "vnodeDef.h"

16
source/dnode/vnode/impl/src/vnodeRead.c
View File

@ -1,16 +0,0 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "vnodeDef.h"

119
source/dnode/vnode/impl/src/vnodeRequest.c
View File

@ -1,119 +0,0 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "vnodeDef.h"
#if 0
static int vnodeBuildCreateTableReq(void **buf, const SVCreateTableReq *pReq);
static void *vnodeParseCreateTableReq(void *buf, SVCreateTableReq *pReq);
int vnodeBuildReq(void **buf, const SVnodeReq *pReq, tmsg_t type) {
int tsize = 0;
tsize += taosEncodeFixedU64(buf, pReq->ver);
switch (type) {
case TDMT_VND_CREATE_STB:
tsize += vnodeBuildCreateTableReq(buf, &(pReq->ctReq));
break;
case TDMT_VND_SUBMIT:
/* code */
break;
default:
break;
}
/* TODO */
return tsize;
}
void *vnodeParseReq(void *buf, SVnodeReq *pReq, tmsg_t type) {
buf = taosDecodeFixedU64(buf, &(pReq->ver));
switch (type) {
case TDMT_VND_CREATE_STB:
buf = vnodeParseCreateTableReq(buf, &(pReq->ctReq));
break;
default:
break;
}
// TODO
return buf;
}
static int vnodeBuildCreateTableReq(void **buf, const SVCreateTableReq *pReq) {
int tsize = 0;
tsize += taosEncodeString(buf, pReq->name);
tsize += taosEncodeFixedU32(buf, pReq->ttl);
tsize += taosEncodeFixedU32(buf, pReq->keep);
tsize += taosEncodeFixedU8(buf, pReq->type);
switch (pReq->type) {
case META_SUPER_TABLE:
tsize += taosEncodeFixedU64(buf, pReq->stbCfg.suid);
tsize += tdEncodeSchema(buf, pReq->stbCfg.pSchema);
tsize += tdEncodeSchema(buf, pReq->stbCfg.pTagSchema);
break;
case META_CHILD_TABLE:
tsize += taosEncodeFixedU64(buf, pReq->ctbCfg.suid);
tsize += tdEncodeKVRow(buf, pReq->ctbCfg.pTag);
break;
case META_NORMAL_TABLE:
tsize += tdEncodeSchema(buf, pReq->ntbCfg.pSchema);
break;
default:
break;
}
return tsize;
}
static void *vnodeParseCreateTableReq(void *buf, SVCreateTableReq *pReq) {
buf = taosDecodeString(buf, &(pReq->name));
buf = taosDecodeFixedU32(buf, &(pReq->ttl));
buf = taosDecodeFixedU32(buf, &(pReq->keep));
buf = taosDecodeFixedU8(buf, &(pReq->type));
switch (pReq->type) {
case META_SUPER_TABLE:
buf = taosDecodeFixedU64(buf, &(pReq->stbCfg.suid));
buf = tdDecodeSchema(buf, &(pReq->stbCfg.pSchema));
buf = tdDecodeSchema(buf, &(pReq->stbCfg.pTagSchema));
break;
case META_CHILD_TABLE:
buf = taosDecodeFixedU64(buf, &(pReq->ctbCfg.suid));
buf = tdDecodeKVRow(buf, &(pReq->ctbCfg.pTag));
break;
case META_NORMAL_TABLE:
buf = tdDecodeSchema(buf, &(pReq->ntbCfg.pSchema));
break;
default:
break;
}
return buf;
}
int vnodeBuildDropTableReq(void **buf, const SVDropTbReq *pReq) {
// TODO
return 0;
}
void *vnodeParseDropTableReq(void *buf, SVDropTbReq *pReq) {
// TODO
}
#endif

14
source/dnode/vnode/impl/src/vnodeSync.c
View File

@ -1,14 +0,0 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/

12
source/dnode/vnode/impl/test/CMakeLists.txt
View File

@ -1,12 +0,0 @@
# Vnode API test
add_executable(vnodeApiTests "")
target_sources(vnodeApiTests
PRIVATE
"vnodeApiTests.cpp"
)
target_link_libraries(vnodeApiTests vnode gtest gtest_main)
add_test(
NAME vnode_api_tests
COMMAND ${CMAKE_CURRENT_BINARY_DIR}/vnodeApiTests
)

2
source/dnode/vnode/impl/test/vBenchmarkTest.cpp
View File

@ -1,2 +0,0 @@
// https://stackoverflow.com/questions/8565666/benchmarking-with-googletest
// https://github.com/google/benchmark

285
source/dnode/vnode/impl/test/vnodeApiTests.cpp
View File

@ -1,285 +0,0 @@
/**
* @file vnodeApiTests.cpp
* @author hzcheng (hzcheng@taosdata.com)
* @brief VNODE module API tests
* @version 0.1
* @date 2021-12-13
*
* @copyright Copyright (c) 2021
*
*/
#include <gtest/gtest.h>
#include <iostream>
#include "vnode.h"
static STSchema *vtCreateBasicSchema() {
STSchemaBuilder sb;
STSchema * pSchema = NULL;
tdInitTSchemaBuilder(&sb, 0);
tdAddColToSchema(&sb, TSDB_DATA_TYPE_TIMESTAMP, 0, 0);
for (int i = 1; i < 10; i++) {
tdAddColToSchema(&sb, TSDB_DATA_TYPE_INT, i, 0);
}
pSchema = tdGetSchemaFromBuilder(&sb);
tdDestroyTSchemaBuilder(&sb);
return pSchema;
}
static STSchema *vtCreateBasicTagSchema() {
STSchemaBuilder sb;
STSchema * pSchema = NULL;
tdInitTSchemaBuilder(&sb, 0);
tdAddColToSchema(&sb, TSDB_DATA_TYPE_TIMESTAMP, 0, 0);
for (int i = 10; i < 12; i++) {
tdAddColToSchema(&sb, TSDB_DATA_TYPE_BINARY, i, 20);
}
pSchema = tdGetSchemaFromBuilder(&sb);
tdDestroyTSchemaBuilder(&sb);
return pSchema;
}
static SKVRow vtCreateBasicTag() {
SKVRowBuilder rb;
SKVRow pTag;
tdInitKVRowBuilder(&rb);
for (int i = 0; i < 2; i++) {
void *pVal = malloc(sizeof(VarDataLenT) + strlen("foo"));
varDataLen(pVal) = strlen("foo");
memcpy(varDataVal(pVal), "foo", strlen("foo"));
tdAddColToKVRow(&rb, i, TSDB_DATA_TYPE_BINARY, pVal);
free(pVal);
}
pTag = tdGetKVRowFromBuilder(&rb);
tdDestroyKVRowBuilder(&rb);
return pTag;
}
static void vtBuildCreateStbReq(tb_uid_t suid, char *tbname, SRpcMsg **ppMsg) {
SRpcMsg * pMsg;
STSchema *pSchema;
STSchema *pTagSchema;
int zs;
void * pBuf;
pSchema = vtCreateBasicSchema();
pTagSchema = vtCreateBasicTagSchema();
SVnodeReq vCreateSTbReq;
vnodeSetCreateStbReq(&vCreateSTbReq, tbname, UINT32_MAX, UINT32_MAX, suid, pSchema, pTagSchema);
zs = vnodeBuildReq(NULL, &vCreateSTbReq, TDMT_VND_CREATE_STB);
pMsg = (SRpcMsg *)malloc(sizeof(SRpcMsg) + zs);
pMsg->msgType = TDMT_VND_CREATE_STB;
pMsg->contLen = zs;
pMsg->pCont = POINTER_SHIFT(pMsg, sizeof(SRpcMsg));
pBuf = pMsg->pCont;
vnodeBuildReq(&pBuf, &vCreateSTbReq, TDMT_VND_CREATE_STB);
META_CLEAR_TB_CFG(&vCreateSTbReq);
tdFreeSchema(pSchema);
tdFreeSchema(pTagSchema);
*ppMsg = pMsg;
}
static void vtBuildCreateCtbReq(tb_uid_t suid, char *tbname, SRpcMsg **ppMsg) {
SRpcMsg *pMsg;
int tz;
SKVRow pTag = vtCreateBasicTag();
SVnodeReq vCreateCTbReq;
vnodeSetCreateCtbReq(&vCreateCTbReq, tbname, UINT32_MAX, UINT32_MAX, suid, pTag);
tz = vnodeBuildReq(NULL, &vCreateCTbReq, TDMT_VND_CREATE_TABLE);
pMsg = (SRpcMsg *)malloc(sizeof(SRpcMsg) + tz);
pMsg->msgType = TDMT_VND_CREATE_TABLE;
pMsg->contLen = tz;
pMsg->pCont = POINTER_SHIFT(pMsg, sizeof(*pMsg));
void *pBuf = pMsg->pCont;
vnodeBuildReq(&pBuf, &vCreateCTbReq, TDMT_VND_CREATE_TABLE);
META_CLEAR_TB_CFG(&vCreateCTbReq);
free(pTag);
*ppMsg = pMsg;
}
static void vtBuildCreateNtbReq(char *tbname, SRpcMsg **ppMsg) {
// TODO
}
static void vtBuildSubmitReq(SRpcMsg **ppMsg) {
SRpcMsg * pMsg;
SSubmitMsg *pSubmitMsg;
SSubmitBlk *pSubmitBlk;
int tz = 1024; // TODO
pMsg = (SRpcMsg *)malloc(sizeof(*pMsg) + tz);
pMsg->msgType = TDMT_VND_SUBMIT;
pMsg->contLen = tz;
pMsg->pCont = POINTER_SHIFT(pMsg, sizeof(*pMsg));
// For submit msg header
pSubmitMsg = (SSubmitMsg *)(pMsg->pCont);
// pSubmitMsg->header.contLen = 0;
// pSubmitMsg->header.vgId = 0;
// pSubmitMsg->length = 0;
pSubmitMsg->numOfBlocks = 1;
// For submit blk
pSubmitBlk = (SSubmitBlk *)(pSubmitMsg->blocks);
pSubmitBlk->uid = 0;
pSubmitBlk->tid = 0;
pSubmitBlk->padding = 0;
pSubmitBlk->sversion = 0;
pSubmitBlk->dataLen = 0;
pSubmitBlk->numOfRows = 0;
// For row batch
*ppMsg = pMsg;
}
static void vtClearMsgBatch(SArray *pMsgArr) {
SRpcMsg *pMsg;
for (size_t i = 0; i < taosArrayGetSize(pMsgArr); i++) {
pMsg = *(SRpcMsg **)taosArrayGet(pMsgArr, i);
free(pMsg);
}
taosArrayClear(pMsgArr);
}
static void vtProcessAndApplyReqs(SVnode *pVnode, SArray *pMsgArr) {
int rcode;
SRpcMsg *pReq;
SRpcMsg *pRsp;
rcode = vnodeProcessWMsgs(pVnode, pMsgArr);
GTEST_ASSERT_EQ(rcode, 0);
for (size_t i = 0; i < taosArrayGetSize(pMsgArr); i++) {
pReq = *(SRpcMsg **)taosArrayGet(pMsgArr, i);
rcode = vnodeApplyWMsg(pVnode, pReq, NULL);
GTEST_ASSERT_EQ(rcode, 0);
}
}
TEST(vnodeApiTest, vnode_simple_create_table_test) {
tb_uid_t suid = 1638166374163;
SRpcMsg *pMsg;
SArray * pMsgArr = NULL;
SVnode * pVnode;
int rcode;
int ntables = 1000000;
int batch = 10;
char tbname[128];
pMsgArr = (SArray *)taosArrayInit(batch, sizeof(pMsg));
vnodeDestroy("vnode1");
GTEST_ASSERT_GE(vnodeInit(2), 0);
// CREATE AND OPEN A VNODE
pVnode = vnodeOpen("vnode1", NULL);
ASSERT_NE(pVnode, nullptr);
// CREATE A SUPER TABLE
sprintf(tbname, "st");
vtBuildCreateStbReq(suid, tbname, &pMsg);
taosArrayPush(pMsgArr, &pMsg);
vtProcessAndApplyReqs(pVnode, pMsgArr);
vtClearMsgBatch(pMsgArr);
// CREATE A LOT OF CHILD TABLES
for (int i = 0; i < ntables / batch; i++) {
// Build request batch
for (int j = 0; j < batch; j++) {
sprintf(tbname, "ct%d", i * batch + j + 1);
vtBuildCreateCtbReq(suid, tbname, &pMsg);
taosArrayPush(pMsgArr, &pMsg);
}
// Process request batch
vtProcessAndApplyReqs(pVnode, pMsgArr);
// Clear request batch
vtClearMsgBatch(pMsgArr);
}
// CLOSE THE VNODE
vnodeClose(pVnode);
vnodeClear();
taosArrayDestroy(pMsgArr);
}
TEST(vnodeApiTest, vnode_simple_insert_test) {
const char *vname = "vnode2";
char tbname[128];
tb_uid_t suid = 1638166374163;
SRpcMsg * pMsg;
SArray * pMsgArr;
int rcode;
SVnode * pVnode;
int batch = 1;
int loop = 1000000;
pMsgArr = (SArray *)taosArrayInit(0, sizeof(pMsg));
vnodeDestroy(vname);
GTEST_ASSERT_GE(vnodeInit(2), 0);
// Open a vnode
pVnode = vnodeOpen(vname, NULL);
GTEST_ASSERT_NE(pVnode, nullptr);
// 1. CREATE A SUPER TABLE
sprintf(tbname, "st");
vtBuildCreateStbReq(suid, tbname, &pMsg);
taosArrayPush(pMsgArr, &pMsg);
vtProcessAndApplyReqs(pVnode, pMsgArr);
vtClearMsgBatch(pMsgArr);
// 2. CREATE A CHILD TABLE
sprintf(tbname, "t0");
vtBuildCreateCtbReq(suid, tbname, &pMsg);
taosArrayPush(pMsgArr, &pMsg);
vtProcessAndApplyReqs(pVnode, pMsgArr);
vtClearMsgBatch(pMsgArr);
// 3. WRITE A LOT OF TIME-SERIES DATA
for (int j = 0; j < loop; j++) {
for (int i = 0; i < batch; i++) {
vtBuildSubmitReq(&pMsg);
taosArrayPush(pMsgArr, &pMsg);
}
vtProcessAndApplyReqs(pVnode, pMsgArr);
vtClearMsgBatch(pMsgArr);
}
// Close the vnode
vnodeClose(pVnode);
vnodeClear();
taosArrayDestroy(pMsgArr);
}

0
include/dnode/vnode/meta/meta.h → source/dnode/vnode/inc/meta.h
View File

0
include/dnode/vnode/tq/tq.h → source/dnode/vnode/inc/tq.h
View File

0
include/dnode/vnode/tsdb/tsdb.h → source/dnode/vnode/inc/tsdb.h
View File

42
include/dnode/vnode/vnode.h → source/dnode/vnode/inc/vnode.h
View File

@ -31,56 +31,50 @@ extern "C" {
/* ------------------------ TYPES EXPOSED ------------------------ */
typedef struct SVnode SVnode;
typedef struct SDnode SDnode;
typedef int32_t (*PutReqToVQueryQFp)(SDnode *pDnode, struct SRpcMsg *pReq);
typedef struct SVnodeCfg {
int32_t vgId;
/** vnode buffer pool options */
SDnode *pDnode;
struct {
/** write buffer size */
uint64_t wsize;
uint64_t ssize;
uint64_t lsize;
/** use heap allocator or arena allocator */
bool isHeapAllocator;
};
/** time to live of tables in this vnode */
uint32_t ttl;
/** data to keep in this vnode */
uint32_t keep;
/** if TS data is eventually consistency */
bool isWeak;
/** TSDB config */
STsdbCfg tsdbCfg;
/** META config */
SMetaCfg metaCfg;
/** TQ config */
STqCfg tqCfg;
/** WAL config */
SWalCfg walCfg;
} SVnodeCfg;
typedef struct {
int32_t sver;
char *timezone;
char *locale;
char *charset;
uint16_t nthreads; // number of commit threads. 0 for no threads and a schedule queue should be given (TODO)
PutReqToVQueryQFp putReqToVQueryQFp;
} SVnodeOpt;
/* ------------------------ SVnode ------------------------ */
/**
* @brief Initialize the vnode module
*
* @param nthreads number of commit threads. 0 for no threads and
* a schedule queue should be given (TODO)
* @param pOption Option of the vnode mnodule
* @return int 0 for success and -1 for failure
*/
int vnodeInit(uint16_t nthreads);
int vnodeInit(const SVnodeOpt *pOption);
/**
* @brief clear a vnode
* @brief Cleanup the vnode module
*
*/
void vnodeClear();
void vnodeCleanup();
/**
* @brief Open a VNODE.
@ -89,7 +83,7 @@ void vnodeClear();
* @param pVnodeCfg options of the vnode
* @return SVnode* The vnode object
*/
SVnode *vnodeOpen(const char *path, const SVnodeCfg *pVnodeCfg, int32_t vid);
SVnode *vnodeOpen(const char *path, const SVnodeCfg *pVnodeCfg);
/**
* @brief Close a VNODE

46
source/dnode/vnode/meta/CMakeLists.txt
View File

@ -1,46 +0,0 @@
set(META_DB_IMPL_LIST "BDB" "SQLITE")
set(META_DB_IMPL "BDB" CACHE STRING "Use BDB as the default META implementation")
set_property(CACHE META_DB_IMPL PROPERTY STRINGS ${META_DB_IMPL_LIST})
if(META_DB_IMPL IN_LIST META_DB_IMPL_LIST)
message(STATUS "META DB Impl: ${META_DB_IMPL}==============")
else()
message(FATAL_ERROR "Invalid META DB IMPL: ${META_DB_IMPL}==============")
endif()
aux_source_directory(src META_SRC)
if(${META_DB_IMPL} STREQUAL "BDB")
list(REMOVE_ITEM META_SRC "src/metaSQLiteImpl.c")
elseif(${META_DB_IMPL} STREQUAL "SQLITE")
list(REMOVE_ITEM META_SRC "src/metaBDBImpl.c")
endif()
add_library(meta STATIC ${META_SRC})
target_include_directories(
meta
PUBLIC "${CMAKE_SOURCE_DIR}/include/dnode/vnode/meta"
PUBLIC "${CMAKE_SOURCE_DIR}/include/libs/index"
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
)
target_link_libraries(
meta
PUBLIC common
PUBLIC index
)
if(${META_DB_IMPL} STREQUAL "BDB")
target_link_libraries(
meta
PUBLIC bdb
)
elseif(${META_DB_IMPL} STREQUAL "SQLITE")
target_link_libraries(
meta
PUBLIC sqlite
)
endif()
if(${BUILD_TEST})
add_subdirectory(test)
endif(${BUILD_TEST})

212
source/dnode/vnode/meta/src/metaSQLiteImpl.c
View File

@ -1,212 +0,0 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "metaDef.h"
#include "sqlite3.h"
struct SMetaDB {
sqlite3 *pDB;
};
int metaOpenDB(SMeta *pMeta) {
char dir[128];
int rc;
char *err = NULL;
pMeta->pDB = (SMetaDB *)calloc(1, sizeof(SMetaDB));
if (pMeta->pDB == NULL) {
// TODO: handle error
return -1;
}
sprintf(dir, "%s/meta.db", pMeta->path);
rc = sqlite3_open(dir, &(pMeta->pDB->pDB));
if (rc != SQLITE_OK) {
// TODO: handle error
printf("failed to open meta.db\n");
}
// For all tables
rc = sqlite3_exec(pMeta->pDB->pDB,
"CREATE TABLE IF NOT EXISTS tb ("
" tbname VARCHAR(256) NOT NULL UNIQUE,"
" tb_uid INTEGER NOT NULL UNIQUE "
");",
NULL, NULL, &err);
if (rc != SQLITE_OK) {
// TODO: handle error
printf("failed to create meta table tb since %s\n", err);
}
// For super tables
rc = sqlite3_exec(pMeta->pDB->pDB,
"CREATE TABLE IF NOT EXISTS stb ("
" tb_uid INTEGER NOT NULL UNIQUE,"
" tbname VARCHAR(256) NOT NULL UNIQUE,"
" tb_schema BLOB NOT NULL,"
" tag_schema BLOB NOT NULL"
");",
NULL, NULL, &err);
if (rc != SQLITE_OK) {
// TODO: handle error
printf("failed to create meta table stb since %s\n", err);
}
// For normal tables
rc = sqlite3_exec(pMeta->pDB->pDB,
"CREATE TABLE IF NOT EXISTS ntb ("
" tb_uid INTEGER NOT NULL UNIQUE,"
" tbname VARCHAR(256) NOT NULL,"
" tb_schema BLOB NOT NULL"
");",
NULL, NULL, &err);
if (rc != SQLITE_OK) {
// TODO: handle error
printf("failed to create meta table ntb since %s\n", err);
}
sqlite3_exec(pMeta->pDB->pDB, "BEGIN;", NULL, NULL, &err);
tfree(err);
return 0;
}
void metaCloseDB(SMeta *pMeta) {
if (pMeta->pDB) {
sqlite3_exec(pMeta->pDB->pDB, "COMMIT;", NULL, NULL, NULL);
sqlite3_close(pMeta->pDB->pDB);
free(pMeta->pDB);
pMeta->pDB = NULL;
}
// TODO
}
int metaSaveTableToDB(SMeta *pMeta, const STbCfg *pTbCfg) {
char sql[256];
char * err = NULL;
int rc;
tb_uid_t uid;
sqlite3_stmt *stmt;
char buf[256];
void * pBuf;
switch (pTbCfg->type) {
case META_SUPER_TABLE:
uid = pTbCfg->stbCfg.suid;
sprintf(sql,
"INSERT INTO tb VALUES (\'%s\', %" PRIu64
");"
"CREATE TABLE IF NOT EXISTS stb_%" PRIu64
" ("
" tb_uid INTEGER NOT NULL UNIQUE,"
" tbname VARCHAR(256),"
" tag1 INTEGER);",
pTbCfg->name, uid, uid);
rc = sqlite3_exec(pMeta->pDB->pDB, sql, NULL, NULL, &err);
if (rc != SQLITE_OK) {
printf("failed to create normal table since %s\n", err);
}
sprintf(sql, "INSERT INTO stb VALUES (%" PRIu64 ", %s, ?, ?)", uid, pTbCfg->name);
sqlite3_prepare_v2(pMeta->pDB->pDB, sql, -1, &stmt, NULL);
pBuf = buf;
tdEncodeSchema(&pBuf, pTbCfg->stbCfg.pSchema);
sqlite3_bind_blob(stmt, 1, buf, POINTER_DISTANCE(pBuf, buf), NULL);
pBuf = buf;
tdEncodeSchema(&pBuf, pTbCfg->stbCfg.pTagSchema);
sqlite3_bind_blob(stmt, 2, buf, POINTER_DISTANCE(pBuf, buf), NULL);
sqlite3_step(stmt);
sqlite3_finalize(stmt);
#if 0
sprintf(sql,
"INSERT INTO tb VALUES (?, ?);"
// "INSERT INTO stb VALUES (?, ?, ?, ?);"
// "CREATE TABLE IF NOT EXISTS stb_%" PRIu64
// " ("
// " tb_uid INTEGER NOT NULL UNIQUE,"
// " tbname VARCHAR(256),"
// " tag1 INTEGER);"
,
uid);
rc = sqlite3_prepare_v2(pMeta->pDB->pDB, sql, -1, &stmt, NULL);
if (rc != SQLITE_OK) {
return -1;
}
sqlite3_bind_text(stmt, 1, pTbCfg->name, -1, SQLITE_TRANSIENT);
sqlite3_bind_int64(stmt, 2, uid);
sqlite3_step(stmt);
sqlite3_finalize(stmt);
// sqlite3_bind_int64(stmt, 3, uid);
// sqlite3_bind_text(stmt, 4, pTbCfg->name, -1, SQLITE_TRANSIENT);
// pBuf = buf;
// tdEncodeSchema(&pBuf, pTbCfg->stbCfg.pSchema);
// sqlite3_bind_blob(stmt, 5, buf, POINTER_DISTANCE(pBuf, buf), NULL);
// pBuf = buf;
// tdEncodeSchema(&pBuf, pTbCfg->stbCfg.pTagSchema);
// sqlite3_bind_blob(stmt, 6, buf, POINTER_DISTANCE(pBuf, buf), NULL);
rc = sqliteVjj3_step(stmt);
if (rc != SQLITE_OK) {
printf("failed to create normal table since %s\n", sqlite3_errmsg(pMeta->pDB->pDB));
}
sqlite3_finalize(stmt);
#endif
break;
case META_NORMAL_TABLE:
// uid = metaGenerateUid(pMeta);
// sprintf(sql,
// "INSERT INTO tb VALUES (\'%s\', %" PRIu64
// ");"
// "INSERT INTO ntb VALUES (%" PRIu64 ", \'%s\', );",
// pTbCfg->name, uid, uid, pTbCfg->name, );
// rc = sqlite3_exec(pMeta->pDB->pDB, sql, NULL, NULL, &err);
// if (rc != SQLITE_OK) {
// printf("failed to create normal table since %s\n", err);
// }
break;
case META_CHILD_TABLE:
#if 0
uid = metaGenerateUid(pMeta);
// sprintf(sql, "INSERT INTO tb VALUES (\'%s\', %" PRIu64
// ");"
// "INSERT INTO stb_%" PRIu64 " VALUES (%" PRIu64 ", \'%s\', );");
rc = sqlite3_exec(pMeta->pDB->pDB, sql, NULL, NULL, &err);
if (rc != SQLITE_OK) {
printf("failed to create child table since %s\n", err);
}
#endif
break;
default:
break;
}
tfree(err);
return 0;
}
int metaRemoveTableFromDb(SMeta *pMeta, tb_uid_t uid) {
/* TODO */
return 0;
}

24
source/dnode/vnode/meta/test/CMakeLists.txt
View File

@ -1,24 +0,0 @@
# add_executable(metaTest "")
# target_sources(metaTest
# PRIVATE
# "../src/metaMain.c"
# "../src/metaUid.c"
# "metaTests.cpp"
# )
# target_include_directories(metaTest
# PUBLIC
# "${CMAKE_SOURCE_DIR}/include/server/vnode/meta"
# "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
# )
# target_link_libraries(metaTest
# os
# util
# common
# gtest_main
# tkv
# )
# enable_testing()
# add_test(
# NAME meta_test
# COMMAND metaTest
# )

105
source/dnode/vnode/meta/test/metaAPITest.cpp
View File

@ -1,105 +0,0 @@
#if 0
#include <gtest/gtest.h>
#include <string.h>
#include <iostream>
#include "meta.h"
static STSchema *metaGetSimpleSchema() {
STSchema * pSchema = NULL;
STSchemaBuilder sb = {0};
tdInitTSchemaBuilder(&sb, 0);
tdAddColToSchema(&sb, TSDB_DATA_TYPE_TIMESTAMP, 0, 8);
tdAddColToSchema(&sb, TSDB_DATA_TYPE_INT, 1, 4);
pSchema = tdGetSchemaFromBuilder(&sb);
tdDestroyTSchemaBuilder(&sb);
return pSchema;
}
static SKVRow metaGetSimpleTags() {
SKVRowBuilder kvrb = {0};
SKVRow row;
tdInitKVRowBuilder(&kvrb);
int64_t ts = 1634287978000;
int32_t a = 10;
tdAddColToKVRow(&kvrb, 0, TSDB_DATA_TYPE_TIMESTAMP, (void *)(&ts));
tdAddColToKVRow(&kvrb, 0, TSDB_DATA_TYPE_INT, (void *)(&a));
row = tdGetKVRowFromBuilder(&kvrb);
tdDestroyKVRowBuilder(&kvrb);
return row;
}
TEST(MetaTest, DISABLED_meta_create_1m_normal_tables_test) {
// Open Meta
SMeta *meta = metaOpen(NULL, NULL);
std::cout << "Meta is opened!" << std::endl;
// Create 1000000 normal tables
META_TABLE_OPTS_DECLARE(tbOpts);
STSchema *pSchema = metaGetSimpleSchema();
char tbname[128];
for (size_t i = 0; i < 1000000; i++) {
sprintf(tbname, "ntb%ld", i);
metaNormalTableOptsInit(&tbOpts, tbname, pSchema);
metaCreateTable(meta, &tbOpts);
metaTableOptsClear(&tbOpts);
}
tdFreeSchema(pSchema);
// Close Meta
metaClose(meta);
std::cout << "Meta is closed!" << std::endl;
// Destroy Meta
metaDestroy("meta");
std::cout << "Meta is destroyed!" << std::endl;
}
TEST(MetaTest, meta_create_1m_child_tables_test) {
// Open Meta
SMeta *meta = metaOpen(NULL);
std::cout << "Meta is opened!" << std::endl;
// Create a super tables
tb_uid_t uid = 477529885843758ul;
META_TABLE_OPTS_DECLARE(tbOpts);
STSchema *pSchema = metaGetSimpleSchema();
STSchema *pTagSchema = metaGetSimpleSchema();
metaSuperTableOptsInit(&tbOpts, "st", uid, pSchema, pTagSchema);
metaCreateTable(meta, &tbOpts);
metaTableOptsClear(&tbOpts);
tdFreeSchema(pSchema);
tdFreeSchema(pTagSchema);
// Create 1000000 child tables
char name[128];
SKVRow row = metaGetSimpleTags();
for (size_t i = 0; i < 1000000; i++) {
sprintf(name, "ctb%ld", i);
metaChildTableOptsInit(&tbOpts, name, uid, row);
metaCreateTable(meta, &tbOpts);
metaTableOptsClear(&tbOpts);
}
kvRowFree(row);
// Close Meta
metaClose(meta);
std::cout << "Meta is closed!" << std::endl;
// Destroy Meta
metaDestroy("meta");
std::cout << "Meta is destroyed!" << std::endl;
}
#endif

0
source/dnode/vnode/meta/inc/metaCache.h → source/dnode/vnode/src/inc/metaCache.h
View File

0
source/dnode/vnode/meta/inc/metaCfg.h → source/dnode/vnode/src/inc/metaCfg.h
View File

0
source/dnode/vnode/meta/inc/metaDB.h → source/dnode/vnode/src/inc/metaDB.h
View File

0
source/dnode/vnode/meta/inc/metaDef.h → source/dnode/vnode/src/inc/metaDef.h
View File

0
source/dnode/vnode/meta/inc/metaIdx.h → source/dnode/vnode/src/inc/metaIdx.h
View File

0
source/dnode/vnode/meta/inc/metaQuery.h → source/dnode/vnode/src/inc/metaQuery.h
View File

0
source/dnode/vnode/meta/inc/metaTbCfg.h → source/dnode/vnode/src/inc/metaTbCfg.h
View File

0
source/dnode/vnode/meta/inc/metaTbTag.h → source/dnode/vnode/src/inc/metaTbTag.h
View File

0
source/dnode/vnode/meta/inc/metaTbUid.h → source/dnode/vnode/src/inc/metaTbUid.h
View File

0
source/dnode/vnode/tq/inc/tqCommit.h → source/dnode/vnode/src/inc/tqCommit.h
View File

0
source/dnode/vnode/tq/inc/tqInt.h → source/dnode/vnode/src/inc/tqInt.h
View File

0
source/dnode/vnode/tq/inc/tqMetaStore.h → source/dnode/vnode/src/inc/tqMetaStore.h
View File

0
source/dnode/vnode/tsdb/inc/tsdbCommit.h → source/dnode/vnode/src/inc/tsdbCommit.h
View File

0
source/dnode/vnode/tsdb/inc/tsdbCompact.h → source/dnode/vnode/src/inc/tsdbCompact.h
View File

0
source/dnode/vnode/tsdb/inc/tsdbDef.h → source/dnode/vnode/src/inc/tsdbDef.h
View File

0
source/dnode/vnode/tsdb/inc/tsdbFS.h → source/dnode/vnode/src/inc/tsdbFS.h
View File

0
source/dnode/vnode/tsdb/inc/tsdbFile.h → source/dnode/vnode/src/inc/tsdbFile.h
View File

0
source/dnode/vnode/tsdb/inc/tsdbHealth.h → source/dnode/vnode/src/inc/tsdbHealth.h
View File

0
source/dnode/vnode/tsdb/inc/tsdbLog.h → source/dnode/vnode/src/inc/tsdbLog.h
View File

0
source/dnode/vnode/tsdb/inc/tsdbMemTable.h → source/dnode/vnode/src/inc/tsdbMemTable.h
View File

0
source/dnode/vnode/tsdb/inc/tsdbMemory.h → source/dnode/vnode/src/inc/tsdbMemory.h
View File

0
source/dnode/vnode/tsdb/inc/tsdbOptions.h → source/dnode/vnode/src/inc/tsdbOptions.h
View File

0
source/dnode/vnode/tsdb/inc/tsdbReadImpl.h → source/dnode/vnode/src/inc/tsdbReadImpl.h
View File

0
source/dnode/vnode/tsdb/inc/tsdbRowMergeBuf.h → source/dnode/vnode/src/inc/tsdbRowMergeBuf.h
View File

0
source/dnode/vnode/impl/inc/vnodeBufferPool.h → source/dnode/vnode/src/inc/vnodeBufferPool.h
View File

0
source/dnode/vnode/impl/inc/vnodeCfg.h → source/dnode/vnode/src/inc/vnodeCfg.h
View File

0
source/dnode/vnode/impl/inc/vnodeCommit.h → source/dnode/vnode/src/inc/vnodeCommit.h
View File

22
source/dnode/vnode/impl/inc/vnodeDef.h → source/dnode/vnode/src/inc/vnodeDef.h
View File

@ -17,7 +17,7 @@
#define _TD_VNODE_DEF_H_
#include "mallocator.h"
#include "sync.h"
// #include "sync.h"
#include "tcoding.h"
#include "tlist.h"
#include "tlockfree.h"
@ -30,12 +30,9 @@
#include "vnodeBufferPool.h"
#include "vnodeCfg.h"
#include "vnodeCommit.h"
#include "vnodeFS.h"
#include "vnodeMemAllocator.h"
#include "vnodeQuery.h"
#include "vnodeRequest.h"
#include "vnodeStateMgr.h"
#include "vnodeSync.h"
#ifdef __cplusplus
extern "C" {
@ -57,6 +54,8 @@ typedef struct SVnodeMgr {
pthread_cond_t hasTask;
TD_DLIST(SVnodeTask) queue;
// For vnode Mgmt
SDnode* pDnode;
PutReqToVQueryQFp putReqToVQueryQFp;
} SVnodeMgr;
extern SVnodeMgr vnodeMgr;
@ -71,14 +70,25 @@ struct SVnode {
STsdb* pTsdb;
STQ* pTq;
SWal* pWal;
SVnodeSync* pSync;
SVnodeFS* pFs;
tsem_t canCommit;
SQHandle* pQuery;
SDnode* pDnode;
};
int vnodeScheduleTask(SVnodeTask* task);
int32_t vnodePutReqToVQueryQ(SVnode *pVnode, struct SRpcMsg *pReq);
// For Log
extern int32_t vDebugFlag;
#define vFatal(...) do { if (vDebugFlag & DEBUG_FATAL) { taosPrintLog("TDB FATAL ", 255, __VA_ARGS__); }} while(0)
#define vError(...) do { if (vDebugFlag & DEBUG_ERROR) { taosPrintLog("TDB ERROR ", 255, __VA_ARGS__); }} while(0)
#define vWarn(...) do { if (vDebugFlag & DEBUG_WARN) { taosPrintLog("TDB WARN ", 255, __VA_ARGS__); }} while(0)
#define vInfo(...) do { if (vDebugFlag & DEBUG_INFO) { taosPrintLog("TDB ", 255, __VA_ARGS__); }} while(0)
#define vDebug(...) do { if (vDebugFlag & DEBUG_DEBUG) { taosPrintLog("TDB ", tsdbDebugFlag, __VA_ARGS__); }} while(0)
#define vTrace(...) do { if (vDebugFlag & DEBUG_TRACE) { taosPrintLog("TDB ", tsdbDebugFlag, __VA_ARGS__); }} while(0)
#ifdef __cplusplus
}
#endif

2
source/dnode/vnode/impl/inc/vnodeInt.h → source/dnode/vnode/src/inc/vnodeInt.h
View File

@ -19,7 +19,7 @@
#include "vnode.h"
#include "meta.h"
#include "sync.h"
// #include "sync.h"
#include "tlog.h"
#include "tq.h"
#include "tsdb.h"

0
source/dnode/vnode/impl/inc/vnodeMemAllocator.h → source/dnode/vnode/src/inc/vnodeMemAllocator.h
View File

3
source/dnode/vnode/impl/inc/vnodeQuery.h → source/dnode/vnode/src/inc/vnodeQuery.h
View File

@ -19,12 +19,11 @@
#ifdef __cplusplus
extern "C" {
#endif
#include "vnodeInt.h"
#include "qworker.h"
#include "vnode.h"
typedef struct SQWorkerMgmt SQHandle;
int vnodeQueryOpen(SVnode *pVnode);
#ifdef __cplusplus

0
source/dnode/vnode/impl/inc/vnodeStateMgr.h → source/dnode/vnode/src/inc/vnodeStateMgr.h
View File

0
source/dnode/vnode/impl/inc/vnodeWrite.h → source/dnode/vnode/src/inc/vnodeWrite.h
View File

0
source/dnode/vnode/meta/src/metaBDBImpl.c → source/dnode/vnode/src/meta/metaBDBImpl.c
View File

0
source/dnode/vnode/meta/src/metaCache.c → source/dnode/vnode/src/meta/metaCache.c
View File

0
source/dnode/vnode/meta/src/metaCfg.c → source/dnode/vnode/src/meta/metaCfg.c
View File

0
source/dnode/vnode/meta/src/metaCommit.c → source/dnode/vnode/src/meta/metaCommit.c
View File

2
source/dnode/vnode/meta/src/metaIdx.c → source/dnode/vnode/src/meta/metaIdx.c
View File

@ -13,7 +13,9 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifdef USE_INVERTED_INDEX
#include "index.h"
#endif
#include "metaDef.h"
struct SMetaIdx {

0
source/dnode/vnode/meta/src/metaMain.c → source/dnode/vnode/src/meta/metaMain.c
View File

0
source/dnode/vnode/meta/src/metaQuery.c → source/dnode/vnode/src/meta/metaQuery.c
View File

0
source/dnode/vnode/meta/src/metaTable.c → source/dnode/vnode/src/meta/metaTable.c
View File

0
source/dnode/vnode/meta/src/metaTbCfg.c → source/dnode/vnode/src/meta/metaTbCfg.c
View File

0
source/dnode/vnode/meta/src/metaTbTag.c → source/dnode/vnode/src/meta/metaTbTag.c
View File

0
source/dnode/vnode/meta/src/metaTbUid.c → source/dnode/vnode/src/meta/metaTbUid.c
View File

0
source/dnode/vnode/tq/src/tq.c → source/dnode/vnode/src/tq/tq.c
View File

Some files were not shown because too many files have changed in this diff Show More