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

massage the code to make the system structure more clear

This commit is contained in:
jtao1735 2020-05-08 02:03:46 +00:00
parent c7a0669318
commit c50321969d
20 changed files with 126 additions and 129 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
src/common/inc/tglobal.h
View File

@ -139,7 +139,8 @@ extern int32_t tsMonitorInterval;
extern int32_t tsAsyncLog; extern int32_t tsAsyncLog;
extern int32_t tsNumOfLogLines; extern int32_t tsNumOfLogLines;
extern int32_t ddebugFlag; extern int32_t dDebugFlag;
extern int32_t vDebugFlag;
extern int32_t mdebugFlag; extern int32_t mdebugFlag;
extern int32_t cdebugFlag; extern int32_t cdebugFlag;
extern int32_t jnidebugFlag; extern int32_t jnidebugFlag;

11
src/common/src/tglobal.c
View File

@ -129,17 +129,18 @@ int32_t tsMaxSQLStringLen = TSDB_MAX_SQL_LEN;
int32_t tsNumOfLogLines = 10000000; int32_t tsNumOfLogLines = 10000000;
int32_t mdebugFlag = 135; int32_t mdebugFlag = 135;
int32_t sdbDebugFlag = 135; int32_t sdbDebugFlag = 135;
int32_t ddebugFlag = 131; int32_t dDebugFlag = 135;
int32_t cdebugFlag = 131; int32_t vDebugFlag = 135;
int32_t cdebugFlag = 135;
int32_t jnidebugFlag = 131; int32_t jnidebugFlag = 131;
int32_t odbcdebugFlag = 131; int32_t odbcdebugFlag = 131;
int32_t httpDebugFlag = 131; int32_t httpDebugFlag = 131;
int32_t monitorDebugFlag = 131; int32_t monitorDebugFlag = 131;
int32_t qdebugFlag = 131; int32_t qdebugFlag = 131;
int32_t rpcDebugFlag = 131; int32_t rpcDebugFlag = 135;
int32_t uDebugFlag = 131; int32_t uDebugFlag = 131;
int32_t debugFlag = 131; int32_t debugFlag = 131;
int32_t sDebugFlag = 131; int32_t sDebugFlag = 135;
// the maximum number of results for projection query on super table that are returned from // the maximum number of results for projection query on super table that are returned from
// one virtual node, to order according to timestamp // one virtual node, to order according to timestamp
@ -1002,7 +1003,7 @@ static void doInitGlobalConfig() {
taosInitConfigOption(cfg); taosInitConfigOption(cfg);
cfg.option = "dDebugFlag"; cfg.option = "dDebugFlag";
cfg.ptr = &ddebugFlag; cfg.ptr = &dDebugFlag;
cfg.valType = TAOS_CFG_VTYPE_INT32; cfg.valType = TAOS_CFG_VTYPE_INT32;
cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_LOG; cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_LOG;
cfg.minValue = 0; cfg.minValue = 0;

4
src/cq/test/cqtest.c
View File

@ -33,13 +33,13 @@ int main(int argc, char *argv[]) {
for (int i=1; i<argc; ++i) { for (int i=1; i<argc; ++i) {
if (strcmp(argv[i], "-d")==0 && i < argc-1) { if (strcmp(argv[i], "-d")==0 && i < argc-1) {
ddebugFlag = atoi(argv[++i]); dDebugFlag = atoi(argv[++i]);
} else if (strcmp(argv[i], "-n") == 0 && i <argc-1) { } else if (strcmp(argv[i], "-n") == 0 && i <argc-1) {
num = atoi(argv[++i]); num = atoi(argv[++i]);
} else { } else {
printf("\nusage: %s [options] \n", argv[0]); printf("\nusage: %s [options] \n", argv[0]);
printf(" [-n num]: number of streams, default:%d\n", num); printf(" [-n num]: number of streams, default:%d\n", num);
printf(" [-d debugFlag]: debug flag, default:%d\n", ddebugFlag); printf(" [-d debugFlag]: debug flag, default:%d\n", dDebugFlag);
printf(" [-h help]: print out this help\n\n"); printf(" [-h help]: print out this help\n\n");
exit(0); exit(0);
} }

20
src/dnode/inc/dnodeLog.h → src/dnode/inc/dnodeInt.h
View File

@ -22,22 +22,12 @@ extern "C" {
#include "tlog.h" #include "tlog.h"
extern int32_t ddebugFlag; extern int32_t dDebugFlag;
#define dError(...) \ #define dError(...) if (dDebugFlag & DEBUG_ERROR) {taosPrintLog("ERROR DND ", 255, __VA_ARGS__); }
if (ddebugFlag & DEBUG_ERROR) { \ #define dWarn(...) if (dDebugFlag & DEBUG_WARN) {taosPrintLog("WARN DND ", dDebugFlag, __VA_ARGS__); }
taosPrintLog("ERROR DND ", 255, __VA_ARGS__); \ #define dTrace(...) if (dDebugFlag & DEBUG_TRACE) {taosPrintLog("DND ", dDebugFlag, __VA_ARGS__); }
} #define dPrint(...) {taosPrintLog("DND ", 255, __VA_ARGS__); }
#define dWarn(...) \
if (ddebugFlag & DEBUG_WARN) { \
taosPrintLog("WARN DND ", ddebugFlag, __VA_ARGS__); \
}
#define dTrace(...) \
if (ddebugFlag & DEBUG_TRACE) { \
taosPrintLog("DND ", ddebugFlag, __VA_ARGS__); \
}
#define dPrint(...) \
{ taosPrintLog("DND ", 255, __VA_ARGS__); }
#ifdef __cplusplus #ifdef __cplusplus
} }

2
src/dnode/inc/dnodeMgmt.h
View File

@ -22,7 +22,7 @@ extern "C" {
int32_t dnodeInitMgmt(); int32_t dnodeInitMgmt();
void dnodeCleanupMgmt(); void dnodeCleanupMgmt();
void dnodeMgmt(SRpcMsg *rpcMsg); void dnodeDispatchToDnodeMgmt(SRpcMsg *rpcMsg);
void* dnodeGetVnode(int32_t vgId); void* dnodeGetVnode(int32_t vgId);
int32_t dnodeGetVnodeStatus(void *pVnode); int32_t dnodeGetVnodeStatus(void *pVnode);

2
src/dnode/inc/dnodeRead.h → src/dnode/inc/dnodeVRead.h
View File

@ -22,7 +22,7 @@ extern "C" {
int32_t dnodeInitRead(); int32_t dnodeInitRead();
void dnodeCleanupRead(); void dnodeCleanupRead();
void dnodeRead(SRpcMsg *pMsg); void dnodeDispatchToVnodeReadQueue(SRpcMsg *pMsg);
#ifdef __cplusplus #ifdef __cplusplus
} }

2
src/dnode/inc/dnodeWrite.h → src/dnode/inc/dnodeVWrite.h
View File

@ -22,7 +22,7 @@ extern "C" {
int32_t dnodeInitWrite(); int32_t dnodeInitWrite();
void dnodeCleanupWrite(); void dnodeCleanupWrite();
void dnodeWrite(SRpcMsg *pMsg); void dnodeDispatchToVnodeWriteQueue(SRpcMsg *pMsg);
void dnodeSendWriteResponse(void *pVnode, void *param, int32_t code); void dnodeSendWriteResponse(void *pVnode, void *param, int32_t code);
#ifdef __cplusplus #ifdef __cplusplus

6
src/dnode/src/dnodeMain.c
View File

@ -22,13 +22,13 @@
#include "tconfig.h" #include "tconfig.h"
#include "tglobal.h" #include "tglobal.h"
#include "dnode.h" #include "dnode.h"
#include "dnodeLog.h" #include "dnodeInt.h"
#include "dnodeMgmt.h" #include "dnodeMgmt.h"
#include "dnodePeer.h" #include "dnodePeer.h"
#include "dnodeModule.h" #include "dnodeModule.h"
#include "dnodeRead.h" #include "dnodeVRead.h"
#include "dnodeShell.h" #include "dnodeShell.h"
#include "dnodeWrite.h" #include "dnodeVWrite.h"
#include "tgrant.h" #include "tgrant.h"
static int32_t dnodeInitSystem(); static int32_t dnodeInitSystem();

8
src/dnode/src/dnodeMgmt.c
View File

@ -32,10 +32,10 @@
#include "dnode.h" #include "dnode.h"
#include "vnode.h" #include "vnode.h"
#include "mnode.h" #include "mnode.h"
#include "dnodeLog.h" #include "dnodeInt.h"
#include "dnodeMgmt.h" #include "dnodeMgmt.h"
#include "dnodeRead.h" #include "dnodeVRead.h"
#include "dnodeWrite.h" #include "dnodeVWrite.h"
#include "dnodeModule.h" #include "dnodeModule.h"
#define MPEER_CONTENT_LEN 2000 #define MPEER_CONTENT_LEN 2000
@ -127,7 +127,7 @@ void dnodeCleanupMgmt() {
dnodeCloseVnodes(); dnodeCloseVnodes();
} }
void dnodeMgmt(SRpcMsg *pMsg) { void dnodeDispatchToDnodeMgmt(SRpcMsg *pMsg) {
SRpcMsg rsp; SRpcMsg rsp;
if (dnodeProcessMgmtMsgFp[pMsg->msgType]) { if (dnodeProcessMgmtMsgFp[pMsg->msgType]) {

2
src/dnode/src/dnodeModule.c
View File

@ -21,7 +21,7 @@
#include "mnode.h" #include "mnode.h"
#include "http.h" #include "http.h"
#include "monitor.h" #include "monitor.h"
#include "dnodeLog.h" #include "dnodeInt.h"
#include "dnodeModule.h" #include "dnodeModule.h"
typedef struct { typedef struct {

20
src/dnode/src/dnodePeer.c
View File

@ -24,9 +24,9 @@
#include "tglobal.h" #include "tglobal.h"
#include "trpc.h" #include "trpc.h"
#include "dnode.h" #include "dnode.h"
#include "dnodeLog.h" #include "dnodeInt.h"
#include "dnodeMgmt.h" #include "dnodeMgmt.h"
#include "dnodeWrite.h" #include "dnodeVWrite.h"
#include "mnode.h" #include "mnode.h"
extern void dnodeUpdateIpSet(void *ahandle, SRpcIpSet *pIpSet); extern void dnodeUpdateIpSet(void *ahandle, SRpcIpSet *pIpSet);
@ -38,15 +38,15 @@ static void *tsDnodeServerRpc = NULL;
static void *tsDnodeClientRpc = NULL; static void *tsDnodeClientRpc = NULL;
int32_t dnodeInitServer() { int32_t dnodeInitServer() {
dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_CREATE_TABLE] = dnodeWrite; dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_CREATE_TABLE] = dnodeDispatchToVnodeWriteQueue;
dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_DROP_TABLE] = dnodeWrite; dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_DROP_TABLE] = dnodeDispatchToVnodeWriteQueue;
dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_ALTER_TABLE] = dnodeWrite; dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_ALTER_TABLE] = dnodeDispatchToVnodeWriteQueue;
dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_DROP_STABLE] = dnodeWrite; dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_DROP_STABLE] = dnodeDispatchToVnodeWriteQueue;
dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_CREATE_VNODE] = dnodeMgmt; dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_CREATE_VNODE] = dnodeDispatchToDnodeMgmt;
dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_DROP_VNODE] = dnodeMgmt; dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_DROP_VNODE] = dnodeDispatchToDnodeMgmt;
dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_ALTER_STREAM] = dnodeMgmt; dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_ALTER_STREAM] = dnodeDispatchToDnodeMgmt;
dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_CONFIG_DNODE] = dnodeMgmt; dnodeProcessReqMsgFp[TSDB_MSG_TYPE_MD_CONFIG_DNODE] = dnodeDispatchToDnodeMgmt;
dnodeProcessReqMsgFp[TSDB_MSG_TYPE_DM_CONFIG_TABLE] = mgmtProcessReqMsgFromDnode; dnodeProcessReqMsgFp[TSDB_MSG_TYPE_DM_CONFIG_TABLE] = mgmtProcessReqMsgFromDnode;
dnodeProcessReqMsgFp[TSDB_MSG_TYPE_DM_CONFIG_VNODE] = mgmtProcessReqMsgFromDnode; dnodeProcessReqMsgFp[TSDB_MSG_TYPE_DM_CONFIG_VNODE] = mgmtProcessReqMsgFromDnode;

12
src/dnode/src/dnodeShell.c
View File

@ -22,9 +22,9 @@
#include "tglobal.h" #include "tglobal.h"
#include "http.h" #include "http.h"
#include "dnode.h" #include "dnode.h"
#include "dnodeLog.h" #include "dnodeInt.h"
#include "dnodeRead.h" #include "dnodeVRead.h"
#include "dnodeWrite.h" #include "dnodeVWrite.h"
#include "dnodeShell.h" #include "dnodeShell.h"
static void (*dnodeProcessShellMsgFp[TSDB_MSG_TYPE_MAX])(SRpcMsg *); static void (*dnodeProcessShellMsgFp[TSDB_MSG_TYPE_MAX])(SRpcMsg *);
@ -37,9 +37,9 @@ static int32_t tsDnodeSubmitReqNum = 0;
void mgmtProcessMsgFromShell(SRpcMsg *rpcMsg); void mgmtProcessMsgFromShell(SRpcMsg *rpcMsg);
int32_t dnodeInitShell() { int32_t dnodeInitShell() {
dnodeProcessShellMsgFp[TSDB_MSG_TYPE_SUBMIT] = dnodeWrite; dnodeProcessShellMsgFp[TSDB_MSG_TYPE_SUBMIT] = dnodeDispatchToVnodeWriteQueue;
dnodeProcessShellMsgFp[TSDB_MSG_TYPE_QUERY] = dnodeRead; dnodeProcessShellMsgFp[TSDB_MSG_TYPE_QUERY] = dnodeDispatchToVnodeReadQueue;
dnodeProcessShellMsgFp[TSDB_MSG_TYPE_FETCH] = dnodeRead; dnodeProcessShellMsgFp[TSDB_MSG_TYPE_FETCH] = dnodeDispatchToVnodeReadQueue;
// the following message shall be treated as mnode write // the following message shall be treated as mnode write
dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CONNECT] = mgmtProcessMsgFromShell; dnodeProcessShellMsgFp[TSDB_MSG_TYPE_CM_CONNECT] = mgmtProcessMsgFromShell;

6
src/dnode/src/dnodeRead.c → src/dnode/src/dnodeVRead.c
View File

@ -21,9 +21,9 @@
#include "trpc.h" #include "trpc.h"
#include "twal.h" #include "twal.h"
#include "tglobal.h" #include "tglobal.h"
#include "dnodeLog.h" #include "dnodeInt.h"
#include "dnodeMgmt.h" #include "dnodeMgmt.h"
#include "dnodeRead.h" #include "dnodeVRead.h"
#include "vnode.h" #include "vnode.h"
typedef struct { typedef struct {
@ -84,7 +84,7 @@ void dnodeCleanupRead() {
dPrint("dnode read is closed"); dPrint("dnode read is closed");
} }
void dnodeRead(SRpcMsg *pMsg) { void dnodeDispatchToVnodeReadQueue(SRpcMsg *pMsg) {
int32_t queuedMsgNum = 0; int32_t queuedMsgNum = 0;
int32_t leftLen = pMsg->contLen; int32_t leftLen = pMsg->contLen;
char *pCont = (char *) pMsg->pCont; char *pCont = (char *) pMsg->pCont;

6
src/dnode/src/dnodeWrite.c → src/dnode/src/dnodeVWrite.c
View File

@ -24,8 +24,8 @@
#include "tglobal.h" #include "tglobal.h"
#include "vnode.h" #include "vnode.h"
#include "tdataformat.h" #include "tdataformat.h"
#include "dnodeLog.h" #include "dnodeInt.h"
#include "dnodeWrite.h" #include "dnodeVWrite.h"
#include "dnodeMgmt.h" #include "dnodeMgmt.h"
typedef struct { typedef struct {
@ -82,7 +82,7 @@ void dnodeCleanupWrite() {
dPrint("dnode write is closed"); dPrint("dnode write is closed");
} }
void dnodeWrite(SRpcMsg *pMsg) { void dnodeDispatchToVnodeWriteQueue(SRpcMsg *pMsg) {
char *pCont = (char *)pMsg->pCont; char *pCont = (char *)pMsg->pCont;
if (pMsg->msgType == TSDB_MSG_TYPE_SUBMIT || pMsg->msgType == TSDB_MSG_TYPE_MD_DROP_STABLE) { if (pMsg->msgType == TSDB_MSG_TYPE_SUBMIT || pMsg->msgType == TSDB_MSG_TYPE_MD_DROP_STABLE) {

2
src/rpc/test/rserver.c
View File

@ -151,7 +151,7 @@ int main(int argc, char *argv[]) {
commit = atoi(argv[++i]); commit = atoi(argv[++i]);
} else if (strcmp(argv[i], "-d")==0 && i < argc-1) { } else if (strcmp(argv[i], "-d")==0 && i < argc-1) {
rpcDebugFlag = atoi(argv[++i]); rpcDebugFlag = atoi(argv[++i]);
ddebugFlag = rpcDebugFlag; dDebugFlag = rpcDebugFlag;
uDebugFlag = rpcDebugFlag; uDebugFlag = rpcDebugFlag;
} else { } else {
printf("\nusage: %s [options] \n", argv[0]); printf("\nusage: %s [options] \n", argv[0]);

9
src/vnode/inc/vnodeInt.h
View File

@ -20,8 +20,17 @@
extern "C" { extern "C" {
#endif #endif
#include "tlog.h"
#include "tsync.h" #include "tsync.h"
#include "twal.h" #include "twal.h"
#include "tcq.h"
extern int32_t vDebugFlag;
#define vError(...) if (vDebugFlag & DEBUG_ERROR) {taosPrintLog("ERROR VND ", 255, __VA_ARGS__); }
#define vWarn(...) if (vDebugFlag & DEBUG_WARN) {taosPrintLog("WARN VND ", vDebugFlag, __VA_ARGS__); }
#define vTrace(...) if (vDebugFlag & DEBUG_TRACE) {taosPrintLog("VND ", vDebugFlag, __VA_ARGS__); }
#define vPrint(...) {taosPrintLog("VND ", 255, __VA_ARGS__); }
typedef struct { typedef struct {
int32_t vgId; // global vnode group ID int32_t vgId; // global vnode group ID

114
src/vnode/src/vnodeMain.c
View File

@ -24,14 +24,10 @@
#include "ttime.h" #include "ttime.h"
#include "ttimer.h" #include "ttimer.h"
#include "cJSON.h" #include "cJSON.h"
#include "twal.h"
#include "tglobal.h" #include "tglobal.h"
#include "dnode.h" #include "dnode.h"
#include "vnode.h" #include "vnode.h"
#include "vnodeInt.h" #include "vnodeInt.h"
#include "vnodeLog.h"
#include "tcq.h"
//#include "tsync.h"
static int32_t tsOpennedVnodes; static int32_t tsOpennedVnodes;
static void *tsDnodeVnodesHash; static void *tsDnodeVnodesHash;
@ -64,7 +60,7 @@ static void vnodeInit() {
tsDnodeVnodesHash = taosInitIntHash(TSDB_MAX_VNODES, sizeof(SVnodeObj *), taosHashInt); tsDnodeVnodesHash = taosInitIntHash(TSDB_MAX_VNODES, sizeof(SVnodeObj *), taosHashInt);
if (tsDnodeVnodesHash == NULL) { if (tsDnodeVnodesHash == NULL) {
dError("failed to init vnode list"); vError("failed to init vnode list");
} }
} }
@ -74,7 +70,7 @@ int32_t vnodeCreate(SMDCreateVnodeMsg *pVnodeCfg) {
SVnodeObj *pTemp = (SVnodeObj *)taosGetIntHashData(tsDnodeVnodesHash, pVnodeCfg->cfg.vgId); SVnodeObj *pTemp = (SVnodeObj *)taosGetIntHashData(tsDnodeVnodesHash, pVnodeCfg->cfg.vgId);
if (pTemp != NULL) { if (pTemp != NULL) {
dPrint("vgId:%d, vnode already exist, pVnode:%p", pVnodeCfg->cfg.vgId, pTemp); vPrint("vgId:%d, vnode already exist, pVnode:%p", pVnodeCfg->cfg.vgId, pTemp);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -93,7 +89,7 @@ int32_t vnodeCreate(SMDCreateVnodeMsg *pVnodeCfg) {
code = vnodeSaveCfg(pVnodeCfg); code = vnodeSaveCfg(pVnodeCfg);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
dError("vgId:%d, failed to save vnode cfg, reason:%s", pVnodeCfg->cfg.vgId, tstrerror(code)); vError("vgId:%d, failed to save vnode cfg, reason:%s", pVnodeCfg->cfg.vgId, tstrerror(code));
return code; return code;
} }
@ -113,11 +109,11 @@ int32_t vnodeCreate(SMDCreateVnodeMsg *pVnodeCfg) {
sprintf(tsdbDir, "%s/vnode%d/tsdb", tsVnodeDir, pVnodeCfg->cfg.vgId); sprintf(tsdbDir, "%s/vnode%d/tsdb", tsVnodeDir, pVnodeCfg->cfg.vgId);
code = tsdbCreateRepo(tsdbDir, &tsdbCfg, NULL); code = tsdbCreateRepo(tsdbDir, &tsdbCfg, NULL);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
dError("vgId:%d, failed to create tsdb in vnode, reason:%s", pVnodeCfg->cfg.vgId, tstrerror(code)); vError("vgId:%d, failed to create tsdb in vnode, reason:%s", pVnodeCfg->cfg.vgId, tstrerror(code));
return TSDB_CODE_VG_INIT_FAILED; return TSDB_CODE_VG_INIT_FAILED;
} }
dPrint("vgId:%d, vnode is created, clog:%d", pVnodeCfg->cfg.vgId, pVnodeCfg->cfg.walLevel); vPrint("vgId:%d, vnode is created, clog:%d", pVnodeCfg->cfg.vgId, pVnodeCfg->cfg.walLevel);
code = vnodeOpen(pVnodeCfg->cfg.vgId, rootDir); code = vnodeOpen(pVnodeCfg->cfg.vgId, rootDir);
return code; return code;
@ -126,12 +122,12 @@ int32_t vnodeCreate(SMDCreateVnodeMsg *pVnodeCfg) {
int32_t vnodeDrop(int32_t vgId) { int32_t vnodeDrop(int32_t vgId) {
SVnodeObj **ppVnode = (SVnodeObj **)taosGetIntHashData(tsDnodeVnodesHash, vgId); SVnodeObj **ppVnode = (SVnodeObj **)taosGetIntHashData(tsDnodeVnodesHash, vgId);
if (ppVnode == NULL || *ppVnode == NULL) { if (ppVnode == NULL || *ppVnode == NULL) {
dTrace("vgId:%d, failed to drop, vgId not exist", vgId); vTrace("vgId:%d, failed to drop, vgId not exist", vgId);
return TSDB_CODE_INVALID_VGROUP_ID; return TSDB_CODE_INVALID_VGROUP_ID;
} }
SVnodeObj *pVnode = *ppVnode; SVnodeObj *pVnode = *ppVnode;
dTrace("pVnode:%p vgId:%d, vnode will be dropped", pVnode, pVnode->vgId); vTrace("vgId:%d, vnode will be dropped", pVnode->vgId);
pVnode->status = TAOS_VN_STATUS_DELETING; pVnode->status = TAOS_VN_STATUS_DELETING;
vnodeCleanUp(pVnode); vnodeCleanUp(pVnode);
@ -144,34 +140,34 @@ int32_t vnodeAlter(void *param, SMDCreateVnodeMsg *pVnodeCfg) {
int32_t code = vnodeSaveCfg(pVnodeCfg); int32_t code = vnodeSaveCfg(pVnodeCfg);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
dError("vgId:%d, failed to save vnode cfg, reason:%s", pVnodeCfg->cfg.vgId, tstrerror(code)); vError("vgId:%d, failed to save vnode cfg, reason:%s", pVnodeCfg->cfg.vgId, tstrerror(code));
return code; return code;
} }
code = vnodeReadCfg(pVnode); code = vnodeReadCfg(pVnode);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
dError("pVnode:%p vgId:%d, failed to read cfg file", pVnode, pVnode->vgId); vError("vgId:%d, failed to read cfg file", pVnode->vgId);
taosDeleteIntHash(tsDnodeVnodesHash, pVnode->vgId); taosDeleteIntHash(tsDnodeVnodesHash, pVnode->vgId);
return code; return code;
} }
code = syncReconfig(pVnode->sync, &pVnode->syncCfg); code = syncReconfig(pVnode->sync, &pVnode->syncCfg);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
dTrace("pVnode:%p vgId:%d, failed to alter vnode, canot reconfig sync, result:%s", pVnode, pVnode->vgId, vTrace("vgId:%d, failed to alter vnode, canot reconfig sync, result:%s", pVnode->vgId,
tstrerror(code)); tstrerror(code));
return code; return code;
} }
code = tsdbConfigRepo(pVnode->tsdb, &pVnode->tsdbCfg); code = tsdbConfigRepo(pVnode->tsdb, &pVnode->tsdbCfg);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
dTrace("pVnode:%p vgId:%d, failed to alter vnode, canot reconfig tsdb, result:%s", pVnode, pVnode->vgId, vTrace("vgId:%d, failed to alter vnode, canot reconfig tsdb, result:%s", pVnode->vgId,
tstrerror(code)); tstrerror(code));
return code; return code;
} }
pVnode->status = TAOS_VN_STATUS_READY; pVnode->status = TAOS_VN_STATUS_READY;
dTrace("pVnode:%p vgId:%d, vnode is altered", pVnode, pVnode->vgId); vTrace("vgId:%d, vnode is altered", pVnode->vgId);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -189,7 +185,7 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
int32_t code = vnodeReadCfg(pVnode); int32_t code = vnodeReadCfg(pVnode);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
dError("pVnode:%p vgId:%d, failed to read cfg file", pVnode, pVnode->vgId); vError("vgId:%d, failed to read cfg file", pVnode->vgId);
taosDeleteIntHash(tsDnodeVnodesHash, pVnode->vgId); taosDeleteIntHash(tsDnodeVnodesHash, pVnode->vgId);
return code; return code;
} }
@ -214,7 +210,7 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
sprintf(temp, "%s/tsdb", rootDir); sprintf(temp, "%s/tsdb", rootDir);
pVnode->tsdb = tsdbOpenRepo(temp, &appH); pVnode->tsdb = tsdbOpenRepo(temp, &appH);
if (pVnode->tsdb == NULL) { if (pVnode->tsdb == NULL) {
dError("pVnode:%p vgId:%d, failed to open tsdb at %s(%s)", pVnode, pVnode->vgId, temp, tstrerror(terrno)); vError("vgId:%d, failed to open tsdb at %s(%s)", pVnode->vgId, temp, tstrerror(terrno));
taosDeleteIntHash(tsDnodeVnodesHash, pVnode->vgId); taosDeleteIntHash(tsDnodeVnodesHash, pVnode->vgId);
return terrno; return terrno;
} }
@ -243,7 +239,7 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
pVnode->events = NULL; pVnode->events = NULL;
pVnode->status = TAOS_VN_STATUS_READY; pVnode->status = TAOS_VN_STATUS_READY;
dTrace("pVnode:%p vgId:%d, vnode is opened in %s", pVnode, pVnode->vgId, rootDir); vTrace("vgId:%d, vnode is opened in %s, pVnode:%p", pVnode->vgId, rootDir, pVnode);
atomic_add_fetch_32(&tsOpennedVnodes, 1); atomic_add_fetch_32(&tsOpennedVnodes, 1);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -254,7 +250,7 @@ int32_t vnodeClose(int32_t vgId) {
if (ppVnode == NULL || *ppVnode == NULL) return 0; if (ppVnode == NULL || *ppVnode == NULL) return 0;
SVnodeObj *pVnode = *ppVnode; SVnodeObj *pVnode = *ppVnode;
dTrace("pVnode:%p vgId:%d, vnode will be closed", pVnode, pVnode->vgId); vTrace("vgId:%d, vnode will be closed", pVnode->vgId);
pVnode->status = TAOS_VN_STATUS_CLOSING; pVnode->status = TAOS_VN_STATUS_CLOSING;
vnodeCleanUp(pVnode); vnodeCleanUp(pVnode);
@ -269,7 +265,7 @@ void vnodeRelease(void *pVnodeRaw) {
assert(refCount >= 0); assert(refCount >= 0);
if (refCount > 0) { if (refCount > 0) {
dTrace("pVnode:%p vgId:%d, release vnode, refCount:%d", pVnode, vgId, refCount); vTrace("vgId:%d, release vnode, refCount:%d", pVnode, vgId, refCount);
return; return;
} }
@ -290,7 +286,7 @@ void vnodeRelease(void *pVnodeRaw) {
free(pVnode); free(pVnode);
int32_t count = atomic_sub_fetch_32(&tsOpennedVnodes, 1); int32_t count = atomic_sub_fetch_32(&tsOpennedVnodes, 1);
dTrace("pVnode:%p vgId:%d, vnode is released, vnodes:%d", pVnode, vgId, count); vTrace("vgId:%d, vnode is released, vnodes:%d", pVnode, vgId, count);
if (count <= 0) { if (count <= 0) {
taosCleanUpIntHash(tsDnodeVnodesHash); taosCleanUpIntHash(tsDnodeVnodesHash);
@ -303,7 +299,7 @@ void *vnodeGetVnode(int32_t vgId) {
SVnodeObj **ppVnode = (SVnodeObj **)taosGetIntHashData(tsDnodeVnodesHash, vgId); SVnodeObj **ppVnode = (SVnodeObj **)taosGetIntHashData(tsDnodeVnodesHash, vgId);
if (ppVnode == NULL || *ppVnode == NULL) { if (ppVnode == NULL || *ppVnode == NULL) {
terrno = TSDB_CODE_INVALID_VGROUP_ID; terrno = TSDB_CODE_INVALID_VGROUP_ID;
dPrint("vgId:%d not exist", vgId); vPrint("vgId:%d not exist", vgId);
return NULL; return NULL;
} }
@ -315,7 +311,7 @@ void *vnodeAccquireVnode(int32_t vgId) {
if (pVnode == NULL) return pVnode; if (pVnode == NULL) return pVnode;
atomic_add_fetch_32(&pVnode->refCount, 1); atomic_add_fetch_32(&pVnode->refCount, 1);
dTrace("pVnode:%p vgId:%d, get vnode, refCount:%d", pVnode, pVnode->vgId, pVnode->refCount); vTrace("vgId:%d, get vnode, refCount:%d", pVnode->vgId, pVnode->refCount);
return pVnode; return pVnode;
} }
@ -410,7 +406,7 @@ static int32_t vnodeSaveCfg(SMDCreateVnodeMsg *pVnodeCfg) {
sprintf(cfgFile, "%s/vnode%d/config.json", tsVnodeDir, pVnodeCfg->cfg.vgId); sprintf(cfgFile, "%s/vnode%d/config.json", tsVnodeDir, pVnodeCfg->cfg.vgId);
FILE *fp = fopen(cfgFile, "w"); FILE *fp = fopen(cfgFile, "w");
if (!fp) { if (!fp) {
dError("vgId:%d, failed to open vnode cfg file for write, file:%s error:%s", pVnodeCfg->cfg.vgId, cfgFile, vError("vgId:%d, failed to open vnode cfg file for write, file:%s error:%s", pVnodeCfg->cfg.vgId, cfgFile,
strerror(errno)); strerror(errno));
return errno; return errno;
} }
@ -456,7 +452,7 @@ static int32_t vnodeSaveCfg(SMDCreateVnodeMsg *pVnodeCfg) {
fclose(fp); fclose(fp);
free(content); free(content);
dPrint("vgId:%d, save vnode cfg successed", pVnodeCfg->cfg.vgId); vPrint("vgId:%d, save vnode cfg successed", pVnodeCfg->cfg.vgId);
return 0; return 0;
} }
@ -466,7 +462,7 @@ static int32_t vnodeReadCfg(SVnodeObj *pVnode) {
sprintf(cfgFile, "%s/vnode%d/config.json", tsVnodeDir, pVnode->vgId); sprintf(cfgFile, "%s/vnode%d/config.json", tsVnodeDir, pVnode->vgId);
FILE *fp = fopen(cfgFile, "r"); FILE *fp = fopen(cfgFile, "r");
if (!fp) { if (!fp) {
dError("pVnode:%p vgId:%d, failed to open vnode cfg file for read, file:%s, error:%s", pVnode, pVnode->vgId, vError("vgId:%d, failed to open vnode cfg file for read, file:%s, error:%s", pVnode->vgId,
cfgFile, strerror(errno)); cfgFile, strerror(errno));
return errno; return errno;
} }
@ -478,117 +474,117 @@ static int32_t vnodeReadCfg(SVnodeObj *pVnode) {
if (len <= 0) { if (len <= 0) {
free(content); free(content);
fclose(fp); fclose(fp);
dError("pVnode:%p vgId:%d, failed to read vnode cfg, content is null", pVnode, pVnode->vgId); vError("vgId:%d, failed to read vnode cfg, content is null", pVnode->vgId);
return false; return false;
} }
cJSON *root = cJSON_Parse(content); cJSON *root = cJSON_Parse(content);
if (root == NULL) { if (root == NULL) {
dError("pVnode:%p vgId:%d, failed to read vnode cfg, invalid json format", pVnode, pVnode->vgId); vError("vgId:%d, failed to read vnode cfg, invalid json format", pVnode->vgId);
goto PARSE_OVER; goto PARSE_OVER;
} }
cJSON *cfgVersion = cJSON_GetObjectItem(root, "cfgVersion"); cJSON *cfgVersion = cJSON_GetObjectItem(root, "cfgVersion");
if (!cfgVersion || cfgVersion->type != cJSON_Number) { if (!cfgVersion || cfgVersion->type != cJSON_Number) {
dError("pVnode:%p vgId:%d, failed to read vnode cfg, cfgVersion not found", pVnode, pVnode->vgId); vError("vgId:%d, failed to read vnode cfg, cfgVersion not found", pVnode->vgId);
goto PARSE_OVER; goto PARSE_OVER;
} }
pVnode->cfgVersion = cfgVersion->valueint; pVnode->cfgVersion = cfgVersion->valueint;
cJSON *cacheBlockSize = cJSON_GetObjectItem(root, "cacheBlockSize"); cJSON *cacheBlockSize = cJSON_GetObjectItem(root, "cacheBlockSize");
if (!cacheBlockSize || cacheBlockSize->type != cJSON_Number) { if (!cacheBlockSize || cacheBlockSize->type != cJSON_Number) {
dError("pVnode:%p vgId:%d, failed to read vnode cfg, cacheBlockSize not found", pVnode, pVnode->vgId); vError("vgId:%d, failed to read vnode cfg, cacheBlockSize not found", pVnode->vgId);
goto PARSE_OVER; goto PARSE_OVER;
} }
pVnode->tsdbCfg.cacheBlockSize = cacheBlockSize->valueint; pVnode->tsdbCfg.cacheBlockSize = cacheBlockSize->valueint;
cJSON *totalBlocks = cJSON_GetObjectItem(root, "totalBlocks"); cJSON *totalBlocks = cJSON_GetObjectItem(root, "totalBlocks");
if (!totalBlocks || totalBlocks->type != cJSON_Number) { if (!totalBlocks || totalBlocks->type != cJSON_Number) {
dError("pVnode:%p vgId:%d, failed to read vnode cfg, totalBlocks not found", pVnode, pVnode->vgId); vError("vgId:%d, failed to read vnode cfg, totalBlocks not found", pVnode->vgId);
goto PARSE_OVER; goto PARSE_OVER;
} }
pVnode->tsdbCfg.totalBlocks = totalBlocks->valueint; pVnode->tsdbCfg.totalBlocks = totalBlocks->valueint;
cJSON *maxTables = cJSON_GetObjectItem(root, "maxTables"); cJSON *maxTables = cJSON_GetObjectItem(root, "maxTables");
if (!maxTables || maxTables->type != cJSON_Number) { if (!maxTables || maxTables->type != cJSON_Number) {
dError("pVnode:%p vgId:%d, failed to read vnode cfg, maxTables not found", pVnode, pVnode->vgId); vError("vgId:%d, failed to read vnode cfg, maxTables not found", pVnode->vgId);
goto PARSE_OVER; goto PARSE_OVER;
} }
pVnode->tsdbCfg.maxTables = maxTables->valueint; pVnode->tsdbCfg.maxTables = maxTables->valueint;
cJSON *daysPerFile = cJSON_GetObjectItem(root, "daysPerFile"); cJSON *daysPerFile = cJSON_GetObjectItem(root, "daysPerFile");
if (!daysPerFile || daysPerFile->type != cJSON_Number) { if (!daysPerFile || daysPerFile->type != cJSON_Number) {
dError("pVnode:%p vgId:%d, failed to read vnode cfg, daysPerFile not found", pVnode, pVnode->vgId); vError("vgId:%d, failed to read vnode cfg, daysPerFile not found", pVnode->vgId);
goto PARSE_OVER; goto PARSE_OVER;
} }
pVnode->tsdbCfg.daysPerFile = daysPerFile->valueint; pVnode->tsdbCfg.daysPerFile = daysPerFile->valueint;
cJSON *daysToKeep = cJSON_GetObjectItem(root, "daysToKeep"); cJSON *daysToKeep = cJSON_GetObjectItem(root, "daysToKeep");
if (!daysToKeep || daysToKeep->type != cJSON_Number) { if (!daysToKeep || daysToKeep->type != cJSON_Number) {
dError("pVnode:%p vgId:%d, failed to read vnode cfg, daysToKeep not found", pVnode, pVnode->vgId); vError("vgId:%d, failed to read vnode cfg, daysToKeep not found", pVnode->vgId);
goto PARSE_OVER; goto PARSE_OVER;
} }
pVnode->tsdbCfg.keep = daysToKeep->valueint; pVnode->tsdbCfg.keep = daysToKeep->valueint;
cJSON *daysToKeep1 = cJSON_GetObjectItem(root, "daysToKeep1"); cJSON *daysToKeep1 = cJSON_GetObjectItem(root, "daysToKeep1");
if (!daysToKeep1 || daysToKeep1->type != cJSON_Number) { if (!daysToKeep1 || daysToKeep1->type != cJSON_Number) {
dError("pVnode:%p vgId:%d, failed to read vnode cfg, daysToKeep1 not found", pVnode, pVnode->vgId); vError("vgId:%d, failed to read vnode cfg, daysToKeep1 not found", pVnode->vgId);
goto PARSE_OVER; goto PARSE_OVER;
} }
pVnode->tsdbCfg.keep1 = daysToKeep1->valueint; pVnode->tsdbCfg.keep1 = daysToKeep1->valueint;
cJSON *daysToKeep2 = cJSON_GetObjectItem(root, "daysToKeep2"); cJSON *daysToKeep2 = cJSON_GetObjectItem(root, "daysToKeep2");
if (!daysToKeep2 || daysToKeep2->type != cJSON_Number) { if (!daysToKeep2 || daysToKeep2->type != cJSON_Number) {
dError("pVnode:%p vgId:%d, failed to read vnode cfg, daysToKeep2 not found", pVnode, pVnode->vgId); vError("vgId:%d, failed to read vnode cfg, daysToKeep2 not found", pVnode->vgId);
goto PARSE_OVER; goto PARSE_OVER;
} }
pVnode->tsdbCfg.keep2 = daysToKeep2->valueint; pVnode->tsdbCfg.keep2 = daysToKeep2->valueint;
cJSON *minRowsPerFileBlock = cJSON_GetObjectItem(root, "minRowsPerFileBlock"); cJSON *minRowsPerFileBlock = cJSON_GetObjectItem(root, "minRowsPerFileBlock");
if (!minRowsPerFileBlock || minRowsPerFileBlock->type != cJSON_Number) { if (!minRowsPerFileBlock || minRowsPerFileBlock->type != cJSON_Number) {
dError("pVnode:%p vgId:%d, failed to read vnode cfg, minRowsPerFileBlock not found", pVnode, pVnode->vgId); vError("vgId:%d, failed to read vnode cfg, minRowsPerFileBlock not found", pVnode->vgId);
goto PARSE_OVER; goto PARSE_OVER;
} }
pVnode->tsdbCfg.minRowsPerFileBlock = minRowsPerFileBlock->valueint; pVnode->tsdbCfg.minRowsPerFileBlock = minRowsPerFileBlock->valueint;
cJSON *maxRowsPerFileBlock = cJSON_GetObjectItem(root, "maxRowsPerFileBlock"); cJSON *maxRowsPerFileBlock = cJSON_GetObjectItem(root, "maxRowsPerFileBlock");
if (!maxRowsPerFileBlock || maxRowsPerFileBlock->type != cJSON_Number) { if (!maxRowsPerFileBlock || maxRowsPerFileBlock->type != cJSON_Number) {
dError("pVnode:%p vgId:%d, failed to read vnode cfg, maxRowsPerFileBlock not found", pVnode, pVnode->vgId); vError("vgId:%d, failed to read vnode cfg, maxRowsPerFileBlock not found", pVnode->vgId);
goto PARSE_OVER; goto PARSE_OVER;
} }
pVnode->tsdbCfg.maxRowsPerFileBlock = maxRowsPerFileBlock->valueint; pVnode->tsdbCfg.maxRowsPerFileBlock = maxRowsPerFileBlock->valueint;
cJSON *commitTime = cJSON_GetObjectItem(root, "commitTime"); cJSON *commitTime = cJSON_GetObjectItem(root, "commitTime");
if (!commitTime || commitTime->type != cJSON_Number) { if (!commitTime || commitTime->type != cJSON_Number) {
dError("pVnode:%p vgId:%d, failed to read vnode cfg, commitTime not found", pVnode, pVnode->vgId); vError("vgId:%d, failed to read vnode cfg, commitTime not found", pVnode->vgId);
goto PARSE_OVER; goto PARSE_OVER;
} }
pVnode->tsdbCfg.commitTime = (int8_t)commitTime->valueint; pVnode->tsdbCfg.commitTime = (int8_t)commitTime->valueint;
cJSON *precision = cJSON_GetObjectItem(root, "precision"); cJSON *precision = cJSON_GetObjectItem(root, "precision");
if (!precision || precision->type != cJSON_Number) { if (!precision || precision->type != cJSON_Number) {
dError("pVnode:%p vgId:%d, failed to read vnode cfg, precision not found", pVnode, pVnode->vgId); vError("vgId:%d, failed to read vnode cfg, precision not found", pVnode->vgId);
goto PARSE_OVER; goto PARSE_OVER;
} }
pVnode->tsdbCfg.precision = (int8_t)precision->valueint; pVnode->tsdbCfg.precision = (int8_t)precision->valueint;
cJSON *compression = cJSON_GetObjectItem(root, "compression"); cJSON *compression = cJSON_GetObjectItem(root, "compression");
if (!compression || compression->type != cJSON_Number) { if (!compression || compression->type != cJSON_Number) {
dError("pVnode:%p vgId:%d, failed to read vnode cfg, compression not found", pVnode, pVnode->vgId); vError("vgId:%d, failed to read vnode cfg, compression not found", pVnode->vgId);
goto PARSE_OVER; goto PARSE_OVER;
} }
pVnode->tsdbCfg.compression = (int8_t)compression->valueint; pVnode->tsdbCfg.compression = (int8_t)compression->valueint;
cJSON *walLevel = cJSON_GetObjectItem(root, "walLevel"); cJSON *walLevel = cJSON_GetObjectItem(root, "walLevel");
if (!walLevel || walLevel->type != cJSON_Number) { if (!walLevel || walLevel->type != cJSON_Number) {
dError("pVnode:%p vgId:%d, failed to read vnode cfg, walLevel not found", pVnode, pVnode->vgId); vError("vgId:%d, failed to read vnode cfg, walLevel not found", pVnode->vgId);
goto PARSE_OVER; goto PARSE_OVER;
} }
pVnode->walCfg.walLevel = (int8_t) walLevel->valueint; pVnode->walCfg.walLevel = (int8_t) walLevel->valueint;
cJSON *wals = cJSON_GetObjectItem(root, "wals"); cJSON *wals = cJSON_GetObjectItem(root, "wals");
if (!wals || wals->type != cJSON_Number) { if (!wals || wals->type != cJSON_Number) {
dError("pVnode:%p vgId:%d, failed to read vnode cfg, wals not found", pVnode, pVnode->vgId); vError("vgId:%d, failed to read vnode cfg, wals not found", pVnode->vgId);
goto PARSE_OVER; goto PARSE_OVER;
} }
pVnode->walCfg.wals = (int8_t)wals->valueint; pVnode->walCfg.wals = (int8_t)wals->valueint;
@ -596,27 +592,27 @@ static int32_t vnodeReadCfg(SVnodeObj *pVnode) {
cJSON *replica = cJSON_GetObjectItem(root, "replica"); cJSON *replica = cJSON_GetObjectItem(root, "replica");
if (!replica || replica->type != cJSON_Number) { if (!replica || replica->type != cJSON_Number) {
dError("pVnode:%p vgId:%d, failed to read vnode cfg, replica not found", pVnode, pVnode->vgId); vError("vgId:%d, failed to read vnode cfg, replica not found", pVnode->vgId);
goto PARSE_OVER; goto PARSE_OVER;
} }
pVnode->syncCfg.replica = (int8_t)replica->valueint; pVnode->syncCfg.replica = (int8_t)replica->valueint;
cJSON *quorum = cJSON_GetObjectItem(root, "quorum"); cJSON *quorum = cJSON_GetObjectItem(root, "quorum");
if (!quorum || quorum->type != cJSON_Number) { if (!quorum || quorum->type != cJSON_Number) {
dError("failed to read vnode cfg, quorum not found", pVnode, pVnode->vgId); vError("failed to read vnode cfg, quorum not found", pVnode->vgId);
goto PARSE_OVER; goto PARSE_OVER;
} }
pVnode->syncCfg.quorum = (int8_t)quorum->valueint; pVnode->syncCfg.quorum = (int8_t)quorum->valueint;
cJSON *nodeInfos = cJSON_GetObjectItem(root, "nodeInfos"); cJSON *nodeInfos = cJSON_GetObjectItem(root, "nodeInfos");
if (!nodeInfos || nodeInfos->type != cJSON_Array) { if (!nodeInfos || nodeInfos->type != cJSON_Array) {
dError("pVnode:%p vgId:%d, failed to read vnode cfg, nodeInfos not found", pVnode, pVnode->vgId); vError("vgId:%d, failed to read vnode cfg, nodeInfos not found", pVnode->vgId);
goto PARSE_OVER; goto PARSE_OVER;
} }
int size = cJSON_GetArraySize(nodeInfos); int size = cJSON_GetArraySize(nodeInfos);
if (size != pVnode->syncCfg.replica) { if (size != pVnode->syncCfg.replica) {
dError("pVnode:%p vgId:%d, failed to read vnode cfg, nodeInfos size not matched", pVnode, pVnode->vgId); vError("vgId:%d, failed to read vnode cfg, nodeInfos size not matched", pVnode->vgId);
goto PARSE_OVER; goto PARSE_OVER;
} }
@ -626,14 +622,14 @@ static int32_t vnodeReadCfg(SVnodeObj *pVnode) {
cJSON *nodeId = cJSON_GetObjectItem(nodeInfo, "nodeId"); cJSON *nodeId = cJSON_GetObjectItem(nodeInfo, "nodeId");
if (!nodeId || nodeId->type != cJSON_Number) { if (!nodeId || nodeId->type != cJSON_Number) {
dError("pVnode:%p vgId:%d, failed to read vnode cfg, nodeId not found", pVnode, pVnode->vgId); vError("vgId:%d, failed to read vnode cfg, nodeId not found", pVnode->vgId);
goto PARSE_OVER; goto PARSE_OVER;
} }
pVnode->syncCfg.nodeInfo[i].nodeId = nodeId->valueint; pVnode->syncCfg.nodeInfo[i].nodeId = nodeId->valueint;
cJSON *nodeEp = cJSON_GetObjectItem(nodeInfo, "nodeEp"); cJSON *nodeEp = cJSON_GetObjectItem(nodeInfo, "nodeEp");
if (!nodeEp || nodeEp->type != cJSON_String || nodeEp->valuestring == NULL) { if (!nodeEp || nodeEp->type != cJSON_String || nodeEp->valuestring == NULL) {
dError("pVnode:%p vgId:%d, failed to read vnode cfg, nodeFqdn not found", pVnode, pVnode->vgId); vError("vgId:%d, failed to read vnode cfg, nodeFqdn not found", pVnode->vgId);
goto PARSE_OVER; goto PARSE_OVER;
} }
@ -643,9 +639,9 @@ static int32_t vnodeReadCfg(SVnodeObj *pVnode) {
ret = 0; ret = 0;
dPrint("pVnode:%p vgId:%d, read vnode cfg successed, replcia:%d", pVnode, pVnode->vgId, pVnode->syncCfg.replica); vPrint("vgId:%d, read vnode cfg successed, replcia:%d", pVnode->vgId, pVnode->syncCfg.replica);
for (int32_t i = 0; i < pVnode->syncCfg.replica; i++) { for (int32_t i = 0; i < pVnode->syncCfg.replica; i++) {
dPrint("pVnode:%p vgId:%d, dnode:%d, %s:%d", pVnode, pVnode->vgId, pVnode->syncCfg.nodeInfo[i].nodeId, vPrint("vgId:%d, dnode:%d, %s:%d", pVnode->vgId, pVnode->syncCfg.nodeInfo[i].nodeId,
pVnode->syncCfg.nodeInfo[i].nodeFqdn, pVnode->syncCfg.nodeInfo[i].nodePort); pVnode->syncCfg.nodeInfo[i].nodeFqdn, pVnode->syncCfg.nodeInfo[i].nodePort);
} }
@ -661,7 +657,7 @@ static int32_t vnodeSaveVersion(SVnodeObj *pVnode) {
sprintf(versionFile, "%s/vnode%d/version.json", tsVnodeDir, pVnode->vgId); sprintf(versionFile, "%s/vnode%d/version.json", tsVnodeDir, pVnode->vgId);
FILE *fp = fopen(versionFile, "w"); FILE *fp = fopen(versionFile, "w");
if (!fp) { if (!fp) {
dError("pVnode:%p vgId:%d, failed to open vnode version file for write, file:%s error:%s", pVnode, pVnode->vgId, vError("vgId:%d, failed to open vnode version file for write, file:%s error:%s", pVnode->vgId,
versionFile, strerror(errno)); versionFile, strerror(errno));
return errno; return errno;
} }
@ -678,7 +674,7 @@ static int32_t vnodeSaveVersion(SVnodeObj *pVnode) {
fclose(fp); fclose(fp);
free(content); free(content);
dPrint("pVnode:%p vgId:%d, save vnode version:%" PRId64 " successed", pVnode, pVnode->vgId, pVnode->version); vPrint("vgId:%d, save vnode version:%" PRId64 " successed", pVnode->vgId, pVnode->version);
return 0; return 0;
} }
@ -688,7 +684,7 @@ static bool vnodeReadVersion(SVnodeObj *pVnode) {
sprintf(versionFile, "%s/vnode%d/version.json", tsVnodeDir, pVnode->vgId); sprintf(versionFile, "%s/vnode%d/version.json", tsVnodeDir, pVnode->vgId);
FILE *fp = fopen(versionFile, "r"); FILE *fp = fopen(versionFile, "r");
if (!fp) { if (!fp) {
dTrace("pVnode:%p vgId:%d, failed to open version file:%s error:%s", pVnode, pVnode->vgId, vTrace("vgId:%d, failed to open version file:%s error:%s", pVnode->vgId,
versionFile, strerror(errno)); versionFile, strerror(errno));
return false; return false;
} }
@ -700,26 +696,26 @@ static bool vnodeReadVersion(SVnodeObj *pVnode) {
if (len <= 0) { if (len <= 0) {
free(content); free(content);
fclose(fp); fclose(fp);
dPrint("pVnode:%p vgId:%d, failed to read vnode version, content is null", pVnode, pVnode->vgId); vPrint("vgId:%d, failed to read vnode version, content is null", pVnode->vgId);
return false; return false;
} }
cJSON *root = cJSON_Parse(content); cJSON *root = cJSON_Parse(content);
if (root == NULL) { if (root == NULL) {
dError("pVnode:%p vgId:%d, failed to read vnode version, invalid json format", pVnode, pVnode->vgId); vError("vgId:%d, failed to read vnode version, invalid json format", pVnode->vgId);
goto PARSE_OVER; goto PARSE_OVER;
} }
cJSON *version = cJSON_GetObjectItem(root, "version"); cJSON *version = cJSON_GetObjectItem(root, "version");
if (!version || version->type != cJSON_Number) { if (!version || version->type != cJSON_Number) {
dError("pVnode:%p vgId:%d, failed to read vnode version, version not found", pVnode, pVnode->vgId); vError("vgId:%d, failed to read vnode version, version not found", pVnode->vgId);
goto PARSE_OVER; goto PARSE_OVER;
} }
pVnode->version = version->valueint; pVnode->version = version->valueint;
ret = true; ret = true;
dPrint("pVnode:%p vgId:%d, read vnode version successed, version:%%" PRId64, pVnode, pVnode->vgId, pVnode->version); vPrint("vgId:%d, read vnode version successed, version:%%" PRId64, pVnode->vgId, pVnode->version);
PARSE_OVER: PARSE_OVER:
free(content); free(content);

6
src/vnode/src/vnodeRead.c
View File

@ -65,7 +65,7 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, void *pCont, int32_t cont
pRet->len = sizeof(SQueryTableRsp); pRet->len = sizeof(SQueryTableRsp);
pRet->rsp = pRsp; pRet->rsp = pRsp;
dTrace("pVnode:%p vgId:%d QInfo:%p, dnode query msg disposed", pVnode, pVnode->vgId, pQInfo); vTrace("vgId:%d QInfo:%p, dnode query msg disposed", pVnode->vgId, pQInfo);
} else { } else {
pQInfo = pCont; pQInfo = pCont;
code = TSDB_CODE_ACTION_IN_PROGRESS; code = TSDB_CODE_ACTION_IN_PROGRESS;
@ -83,7 +83,7 @@ static int32_t vnodeProcessFetchMsg(SVnodeObj *pVnode, void *pCont, int32_t cont
int32_t code = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;
dTrace("pVnode:%p vgId:%d QInfo:%p, retrieve msg is received", pVnode, pVnode->vgId, pQInfo); vTrace("vgId:%d QInfo:%p, retrieve msg is received", pVnode->vgId, pQInfo);
pRet->code = qRetrieveQueryResultInfo(pQInfo); pRet->code = qRetrieveQueryResultInfo(pQInfo);
if (pRet->code != TSDB_CODE_SUCCESS) { if (pRet->code != TSDB_CODE_SUCCESS) {
@ -104,6 +104,6 @@ static int32_t vnodeProcessFetchMsg(SVnodeObj *pVnode, void *pCont, int32_t cont
} }
} }
dTrace("pVnode:%p vgId:%d QInfo:%p, retrieve msg is disposed", pVnode, pVnode->vgId, pQInfo); vTrace("vgId:%d QInfo:%p, retrieve msg is disposed", pVnode->vgId, pQInfo);
return code; return code;
} }

16
src/vnode/src/vnodeWrite.c
View File

@ -91,7 +91,7 @@ static int32_t vnodeProcessSubmitMsg(SVnodeObj *pVnode, void *pCont, SRspRet *pR
// save insert result into item // save insert result into item
dTrace("pVnode:%p vgId:%d, submit msg is processed", pVnode, pVnode->vgId); vTrace("vgId:%d, submit msg is processed", pVnode->vgId);
code = tsdbInsertData(pVnode->tsdb, pCont); code = tsdbInsertData(pVnode->tsdb, pCont);
pRet->len = sizeof(SShellSubmitRspMsg); pRet->len = sizeof(SShellSubmitRspMsg);
@ -110,7 +110,7 @@ static int32_t vnodeProcessCreateTableMsg(SVnodeObj *pVnode, void *pCont, SRspRe
SMDCreateTableMsg *pTable = pCont; SMDCreateTableMsg *pTable = pCont;
int32_t code = 0; int32_t code = 0;
dTrace("pVnode:%p vgId:%d, table:%s, start to create", pVnode, pVnode->vgId, pTable->tableId); vTrace("vgId:%d, table:%s, start to create", pVnode->vgId, pTable->tableId);
int16_t numOfColumns = htons(pTable->numOfColumns); int16_t numOfColumns = htons(pTable->numOfColumns);
int16_t numOfTags = htons(pTable->numOfTags); int16_t numOfTags = htons(pTable->numOfTags);
int32_t sid = htonl(pTable->sid); int32_t sid = htonl(pTable->sid);
@ -157,7 +157,7 @@ static int32_t vnodeProcessCreateTableMsg(SVnodeObj *pVnode, void *pCont, SRspRe
tfree(pDestTagSchema); tfree(pDestTagSchema);
tfree(pDestSchema); tfree(pDestSchema);
dTrace("pVnode:%p vgId:%d, table:%s is created, result:%x", pVnode, pVnode->vgId, pTable->tableId, code); vTrace("vgId:%d, table:%s is created, result:%x", pVnode->vgId, pTable->tableId, code);
return code; return code;
} }
@ -165,7 +165,7 @@ static int32_t vnodeProcessDropTableMsg(SVnodeObj *pVnode, void *pCont, SRspRet
SMDDropTableMsg *pTable = pCont; SMDDropTableMsg *pTable = pCont;
int32_t code = 0; int32_t code = 0;
dTrace("pVnode:%p vgId:%d, table:%s, start to drop", pVnode, pVnode->vgId, pTable->tableId); vTrace("vgId:%d, table:%s, start to drop", pVnode->vgId, pTable->tableId);
STableId tableId = { STableId tableId = {
.uid = htobe64(pTable->uid), .uid = htobe64(pTable->uid),
.tid = htonl(pTable->sid) .tid = htonl(pTable->sid)
@ -180,7 +180,7 @@ static int32_t vnodeProcessAlterTableMsg(SVnodeObj *pVnode, void *pCont, SRspRet
SMDCreateTableMsg *pTable = pCont; SMDCreateTableMsg *pTable = pCont;
int32_t code = 0; int32_t code = 0;
dTrace("pVnode:%p vgId:%d, table:%s, start to alter", pVnode, pVnode->vgId, pTable->tableId); vTrace("vgId:%d, table:%s, start to alter", pVnode->vgId, pTable->tableId);
int16_t numOfColumns = htons(pTable->numOfColumns); int16_t numOfColumns = htons(pTable->numOfColumns);
int16_t numOfTags = htons(pTable->numOfTags); int16_t numOfTags = htons(pTable->numOfTags);
int32_t sid = htonl(pTable->sid); int32_t sid = htonl(pTable->sid);
@ -221,7 +221,7 @@ static int32_t vnodeProcessAlterTableMsg(SVnodeObj *pVnode, void *pCont, SRspRet
tfree(pDestSchema); tfree(pDestSchema);
dTrace("pVnode:%p vgId:%d, table:%s, alter table result:%d", pVnode, pVnode->vgId, pTable->tableId, code); vTrace("vgId:%d, table:%s, alter table result:%d", pVnode->vgId, pTable->tableId, code);
return code; return code;
} }
@ -230,14 +230,14 @@ static int32_t vnodeProcessDropStableMsg(SVnodeObj *pVnode, void *pCont, SRspRet
SMDDropSTableMsg *pTable = pCont; SMDDropSTableMsg *pTable = pCont;
int32_t code = 0; int32_t code = 0;
dTrace("pVnode:%p vgId:%d, stable:%s, start to drop", pVnode, pVnode->vgId, pTable->tableId); vTrace("vgId:%d, stable:%s, start to drop", pVnode->vgId, pTable->tableId);
// TODO: drop stable in vvnode // TODO: drop stable in vvnode
//int64_t uid = htobe64(pTable->uid); //int64_t uid = htobe64(pTable->uid);
//void *pTsdb = dnodeGetVnodeTsdb(pMsg->pVnode); //void *pTsdb = dnodeGetVnodeTsdb(pMsg->pVnode);
//rpcRsp.code = tsdbDropTable(pTsdb, pTable->uid); //rpcRsp.code = tsdbDropTable(pTsdb, pTable->uid);
code = TSDB_CODE_SUCCESS; code = TSDB_CODE_SUCCESS;
dTrace("pVnode:%p vgId:%d, stable:%s, drop stable result:%x", pVnode, pTable->tableId, code); vTrace("vgId:%d, stable:%s, drop stable result:%x", pVnode, pTable->tableId, code);
return code; return code;
} }

4
src/wal/test/waltest.c
View File

@ -61,7 +61,7 @@ int main(int argc, char *argv[]) {
} else if (strcmp(argv[i], "-v")==0 && i < argc-1) { } else if (strcmp(argv[i], "-v")==0 && i < argc-1) {
ver = atoll(argv[++i]); ver = atoll(argv[++i]);
} else if (strcmp(argv[i], "-d")==0 && i < argc-1) { } else if (strcmp(argv[i], "-d")==0 && i < argc-1) {
ddebugFlag = atoi(argv[++i]); dDebugFlag = atoi(argv[++i]);
} else { } else {
printf("\nusage: %s [options] \n", argv[0]); printf("\nusage: %s [options] \n", argv[0]);
printf(" [-p path]: wal file path default is:%s\n", path); printf(" [-p path]: wal file path default is:%s\n", path);
@ -71,7 +71,7 @@ int main(int argc, char *argv[]) {
printf(" [-r rows]: rows of records per wal file, default is:%d\n", rows); printf(" [-r rows]: rows of records per wal file, default is:%d\n", rows);
printf(" [-k keep]: keep the wal after closing, default is:%d\n", keep); printf(" [-k keep]: keep the wal after closing, default is:%d\n", keep);
printf(" [-v version]: initial version, default is:%ld\n", ver); printf(" [-v version]: initial version, default is:%ld\n", ver);
printf(" [-d debugFlag]: debug flag, default:%d\n", ddebugFlag); printf(" [-d debugFlag]: debug flag, default:%d\n", dDebugFlag);
printf(" [-h help]: print out this help\n\n"); printf(" [-h help]: print out this help\n\n");
exit(0); exit(0);
} }