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

refact dnode-mnode

This commit is contained in:
Shengliang Guan 2021-11-05 15:42:19 +08:00
parent 99cb481b6b
commit 88d7b970ae
12 changed files with 601 additions and 266 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
include/common/taosmsg.h
View File

@ -75,9 +75,9 @@ TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CREATE_FUNCTION, "create-function" )
TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_ALTER_FUNCTION, "alter-function" ) TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_ALTER_FUNCTION, "alter-function" )
TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_DROP_FUNCTION, "drop-function" ) TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_DROP_FUNCTION, "drop-function" )
TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CREATE_STABLE, "create-stable" ) TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CREATE_STABLE, "create-stable" )
TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_STABLE_VGROUP, "stable-vgroup" )
TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_DROP_STABLE, "drop-stable" )
TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_ALTER_STABLE, "alter-stable" ) TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_ALTER_STABLE, "alter-stable" )
TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_DROP_STABLE, "drop-stable" )
TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_STABLE_VGROUP, "stable-vgroup" )
TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_KILL_QUERY, "kill-query" ) TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_KILL_QUERY, "kill-query" )
TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_KILL_CONN, "kill-conn" ) TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_KILL_CONN, "kill-conn" )
TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_HEARTBEAT, "heartbeat" ) TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_HEARTBEAT, "heartbeat" )
@ -108,6 +108,7 @@ TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_AUTH_VNODE_IN, "auth-vnode" )
TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_SYNC_VNODE_IN, "sync-vnode" ) TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_SYNC_VNODE_IN, "sync-vnode" )
TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_COMPACT_VNODE_IN, "compact-vnode" ) TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_COMPACT_VNODE_IN, "compact-vnode" )
TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CREATE_MNODE_IN, "create-mnode" ) TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CREATE_MNODE_IN, "create-mnode" )
TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_ALTER_MNODE_IN, "alter-mnode" )
TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_DROP_MNODE_IN, "drop-mnode" ) TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_DROP_MNODE_IN, "drop-mnode" )
TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CONFIG_DNODE_IN, "config-dnode" ) TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CONFIG_DNODE_IN, "config-dnode" )
@ -698,7 +699,7 @@ typedef struct {
typedef struct { typedef struct {
uint16_t port; uint16_t port;
char fqdn[TSDB_FQDN_LEN]; char fqdn[TSDB_FQDN_LEN];
} SVnodeDesc; } SReplica;
typedef struct { typedef struct {
char db[TSDB_FULL_DB_NAME_LEN]; char db[TSDB_FULL_DB_NAME_LEN];
@ -721,7 +722,7 @@ typedef struct {
int8_t replica; int8_t replica;
int8_t quorum; int8_t quorum;
int8_t selfIndex; int8_t selfIndex;
SVnodeDesc replicas[TSDB_MAX_REPLICA]; SReplica replicas[TSDB_MAX_REPLICA];
} SCreateVnodeMsg, SAlterVnodeMsg; } SCreateVnodeMsg, SAlterVnodeMsg;
typedef struct { typedef struct {
@ -826,7 +827,10 @@ typedef struct {
typedef struct { typedef struct {
int32_t dnodeId; int32_t dnodeId;
} SCreateMnodeMsg, SDropMnodeMsg; int8_t replica;
int8_t reserved[3];
SReplica replicas[TSDB_MAX_REPLICA];
} SCreateMnodeMsg, SAlterMnodeMsg, SDropMnodeMsg;
typedef struct { typedef struct {
int32_t dnodeId; int32_t dnodeId;

22
include/server/mnode/mnode.h
View File

@ -20,6 +20,15 @@
extern "C" { extern "C" {
#endif #endif
typedef enum { MN_MSG_TYPE_WRITE = 1, MN_MSG_TYPE_APPLY, MN_MSG_TYPE_SYNC, MN_MSG_TYPE_READ } EMnMsgType;
typedef struct SMnodeMsg SMnodeMsg;
typedef struct {
int8_t replica;
SReplica replicas[TSDB_MAX_REPLICA];
} SMnodeCfg;
typedef struct { typedef struct {
int64_t numOfDnode; int64_t numOfDnode;
int64_t numOfMnode; int64_t numOfMnode;
@ -39,20 +48,25 @@ typedef struct {
void (*SendMsgToDnode)(struct SEpSet *epSet, struct SRpcMsg *rpcMsg); void (*SendMsgToDnode)(struct SEpSet *epSet, struct SRpcMsg *rpcMsg);
void (*SendMsgToMnode)(struct SRpcMsg *rpcMsg); void (*SendMsgToMnode)(struct SRpcMsg *rpcMsg);
void (*SendRedirectMsg)(struct SRpcMsg *rpcMsg, bool forShell); void (*SendRedirectMsg)(struct SRpcMsg *rpcMsg, bool forShell);
int32_t (*PutMsgIntoApplyQueue)(SMnodeMsg *pMsg);
} SMnodePara; } SMnodePara;
int32_t mnodeInit(SMnodePara para); int32_t mnodeInit(SMnodePara para);
void mnodeCleanup(); void mnodeCleanup();
int32_t mnodeDeploy(); int32_t mnodeDeploy(char *path, SMnodeCfg *pCfg);
void mnodeUnDeploy(); void mnodeUnDeploy(char *path);
int32_t mnodeStart(); int32_t mnodeStart(char *path, SMnodeCfg *pCfg);
int32_t mnodeAlter(SMnodeCfg *pCfg);
void mnodeStop(); void mnodeStop();
int32_t mnodeGetLoad(SMnodeLoad *pLoad); int32_t mnodeGetLoad(SMnodeLoad *pLoad);
int32_t mnodeRetriveAuth(char *user, char *spi, char *encrypt, char *secret, char *ckey); int32_t mnodeRetriveAuth(char *user, char *spi, char *encrypt, char *secret, char *ckey);
void mnodeProcessMsg(SRpcMsg *rpcMsg); SMnodeMsg *mnodeInitMsg(int32_t msgNum);
int32_t mnodeAppendMsg(SMnodeMsg *pMsg, SRpcMsg *pRpcMsg);
void mnodeCleanupMsg(SMnodeMsg *pMsg);
void mnodeProcessMsg(SMnodeMsg *pMsg, EMnMsgType msgType);
#ifdef __cplusplus #ifdef __cplusplus
} }

2
include/server/vnode/vnode.h
View File

@ -44,7 +44,7 @@ typedef struct {
int8_t replica; int8_t replica;
int8_t walLevel; int8_t walLevel;
int32_t fsyncPeriod; // millisecond int32_t fsyncPeriod; // millisecond
SVnodeDesc replicas[TSDB_MAX_REPLICA]; SReplica replicas[TSDB_MAX_REPLICA];
} SVnodeCfg; } SVnodeCfg;
typedef enum { typedef enum {

20
include/util/taoserror.h
View File

@ -216,14 +216,18 @@ int32_t* taosGetErrno();
// dnode // dnode
#define TSDB_CODE_DND_MSG_NOT_PROCESSED TAOS_DEF_ERROR_CODE(0, 0x0400) //"Message not processed") #define TSDB_CODE_DND_MSG_NOT_PROCESSED TAOS_DEF_ERROR_CODE(0, 0x0400) //"Message not processed")
#define TSDB_CODE_DND_OUT_OF_MEMORY TAOS_DEF_ERROR_CODE(0, 0x0401) //"Dnode out of memory") #define TSDB_CODE_DND_OUT_OF_MEMORY TAOS_DEF_ERROR_CODE(0, 0x0401) //"Dnode out of memory")
#define TSDB_CODE_DND_DNODE_ID_NOT_MATCHED TAOS_DEF_ERROR_CODE(0, 0x0402) //"Dnode Id not matched") #define TSDB_CODE_DND_MNODE_ID_NOT_MATCH_DNODE TAOS_DEF_ERROR_CODE(0, 0x0402) //"Mnode Id not match Dnode")
#define TSDB_CODE_DND_MNODE_ALREADY_DROPPED TAOS_DEF_ERROR_CODE(0, 0x0403) //"Mnode already deployed") #define TSDB_CODE_DND_MNODE_ALREADY_DEPLOYED TAOS_DEF_ERROR_CODE(0, 0x0403) //"Mnode already deployed")
#define TSDB_CODE_DND_NO_WRITE_ACCESS TAOS_DEF_ERROR_CODE(0, 0x0404) //"No permission for disk files in dnode") #define TSDB_CODE_DND_MNODE_NOT_DEPLOYED TAOS_DEF_ERROR_CODE(0, 0x0404) //"Mnode not deployed")
#define TSDB_CODE_DND_INVALID_MSG_LEN TAOS_DEF_ERROR_CODE(0, 0x0405) //"Invalid message length") #define TSDB_CODE_DND_READ_MNODE_FILE_ERROR TAOS_DEF_ERROR_CODE(0, 0x0405) //"Read mnode.json error")
#define TSDB_CODE_DND_ACTION_IN_PROGRESS TAOS_DEF_ERROR_CODE(0, 0x0406) //"Action in progress") #define TSDB_CODE_DND_WRITE_MNODE_FILE_ERROR TAOS_DEF_ERROR_CODE(0, 0x0406) //"Write mnode.json error")
#define TSDB_CODE_DND_TOO_MANY_VNODES TAOS_DEF_ERROR_CODE(0, 0x0407) //"Too many vnode directories") #define TSDB_CODE_DND_NO_WRITE_ACCESS TAOS_DEF_ERROR_CODE(0, 0x0407) //"No permission for disk files in dnode")
#define TSDB_CODE_DND_EXITING TAOS_DEF_ERROR_CODE(0, 0x0408) //"Dnode is exiting" #define TSDB_CODE_DND_INVALID_MSG_LEN TAOS_DEF_ERROR_CODE(0, 0x0408) //"Invalid message length")
#define TSDB_CODE_DND_PARSE_VNODE_FILE_ERROR TAOS_DEF_ERROR_CODE(0, 0x0409) //"Parse vnodes.json error") #define TSDB_CODE_DND_ACTION_IN_PROGRESS TAOS_DEF_ERROR_CODE(0, 0x0409) //"Action in progress")
#define TSDB_CODE_DND_TOO_MANY_VNODES TAOS_DEF_ERROR_CODE(0, 0x040A) //"Too many vnode directories")
#define TSDB_CODE_DND_EXITING TAOS_DEF_ERROR_CODE(0, 0x040B) //"Dnode is exiting"
#define TSDB_CODE_DND_PARSE_VNODE_FILE_ERROR TAOS_DEF_ERROR_CODE(0, 0x040C) //"Parse vnodes.json error")
#define TSDB_CODE_DND_PARSE_DNODE_FILE_ERROR TAOS_DEF_ERROR_CODE(0, 0x040D) //"Parse dnodes.json error")
// vnode // vnode
#define TSDB_CODE_VND_ACTION_IN_PROGRESS TAOS_DEF_ERROR_CODE(0, 0x0500) //"Action in progress") #define TSDB_CODE_VND_ACTION_IN_PROGRESS TAOS_DEF_ERROR_CODE(0, 0x0500) //"Action in progress")

6
source/dnode/mgmt/inc/dnodeMnode.h
View File

@ -23,9 +23,13 @@ extern "C" {
int32_t dnodeInitMnode(); int32_t dnodeInitMnode();
void dnodeCleanupMnode(); void dnodeCleanupMnode();
void dnodeProcessMnodeMsg(SRpcMsg *pMsg, SEpSet *pEpSet);
int32_t dnodeGetUserAuthFromMnode(char *user, char *spi, char *encrypt, char *secret, char *ckey); int32_t dnodeGetUserAuthFromMnode(char *user, char *spi, char *encrypt, char *secret, char *ckey);
void dnodeProcessMnodeMgmtMsg(SRpcMsg *pMsg, SEpSet *pEpSet);
void dnodeProcessMnodeReadMsg(SRpcMsg *pMsg, SEpSet *pEpSet);
void dnodeProcessMnodeWriteMsg(SRpcMsg *pMsg, SEpSet *pEpSet);
void dnodeProcessMnodeSyncMsg(SRpcMsg *pMsg, SEpSet *pEpSet);
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif

601
source/dnode/mgmt/src/dnodeMnode.c
View File

@ -15,73 +15,120 @@
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE
#include "dnodeMnode.h" #include "dnodeMnode.h"
#include "cJSON.h"
#include "dnodeDnode.h" #include "dnodeDnode.h"
#include "dnodeTransport.h" #include "dnodeTransport.h"
#include "cJSON.h"
#include "mnode.h" #include "mnode.h"
#include "tlockfree.h"
#include "tqueue.h"
#include "tstep.h"
#include "tworker.h"
static struct { static struct {
int32_t refCount;
int8_t deployed; int8_t deployed;
int8_t dropped; int8_t dropped;
char file[PATH_MAX + 20]; SWorkerPool mgmtPool;
pthread_mutex_t mutex; SWorkerPool readPool;
SWorkerPool writePool;
SWorkerPool syncPool;
taos_queue pReadQ;
taos_queue pWriteQ;
taos_queue pApplyQ;
taos_queue pSyncQ;
taos_queue pMgmtQ;
SSteps *pSteps;
SRWLatch latch;
} tsMnode = {0}; } tsMnode = {0};
static int32_t dnodeReadMnode() { static int32_t dnodeAllocMnodeReadQueue();
static void dnodeFreeMnodeReadQueue();
static int32_t dnodeAllocMnodeWriteQueue();
static void dnodeFreeMnodeWriteQueue();
static int32_t dnodeAllocMnodeApplyQueue();
static void dnodeFreeMnodeApplyQueue();
static int32_t dnodeAllocMnodeSyncQueue();
static void dnodeFreeMnodeSyncQueue();
static int32_t dnodeAcquireMnode() {
taosRLockLatch(&tsMnode.latch);
int32_t code = tsMnode.deployed ? 0 : TSDB_CODE_DND_MNODE_NOT_DEPLOYED;
if (code == 0) {
atomic_add_fetch_32(&tsMnode.refCount, 1);
}
taosRUnLockLatch(&tsMnode.latch);
return code;
}
static void dnodeReleaseMnode() { atomic_sub_fetch_32(&tsMnode.refCount, 1); }
static int32_t dnodeReadMnodeFile() {
int32_t code = TSDB_CODE_DND_READ_MNODE_FILE_ERROR;
int32_t len = 0; int32_t len = 0;
int32_t maxLen = 300; int32_t maxLen = 300;
char *content = calloc(1, maxLen + 1); char *content = calloc(1, maxLen + 1);
cJSON *root = NULL; cJSON *root = NULL;
FILE *fp = NULL; FILE *fp = NULL;
char file[PATH_MAX + 20] = {0};
fp = fopen(tsMnode.file, "r"); snprintf(file, sizeof(file), "%s/mnode.json", tsDnodeDir);
fp = fopen(file, "r");
if (!fp) { if (!fp) {
dDebug("file %s not exist", tsMnode.file); dDebug("file %s not exist", file);
code = 0;
goto PRASE_MNODE_OVER; goto PRASE_MNODE_OVER;
} }
len = (int32_t)fread(content, 1, maxLen, fp); len = (int32_t)fread(content, 1, maxLen, fp);
if (len <= 0) { if (len <= 0) {
dError("failed to read %s since content is null", tsMnode.file); dError("failed to read %s since content is null", file);
goto PRASE_MNODE_OVER; goto PRASE_MNODE_OVER;
} }
content[len] = 0; content[len] = 0;
root = cJSON_Parse(content); root = cJSON_Parse(content);
if (root == NULL) { if (root == NULL) {
dError("failed to read %s since invalid json format", tsMnode.file); dError("failed to read %s since invalid json format", file);
goto PRASE_MNODE_OVER; goto PRASE_MNODE_OVER;
} }
cJSON *deployed = cJSON_GetObjectItem(root, "deployed"); cJSON *deployed = cJSON_GetObjectItem(root, "deployed");
if (!deployed || deployed->type != cJSON_String) { if (!deployed || deployed->type != cJSON_String) {
dError("failed to read %s since deployed not found", tsMnode.file); dError("failed to read %s since deployed not found", file);
goto PRASE_MNODE_OVER; goto PRASE_MNODE_OVER;
} }
tsMnode.deployed = atoi(deployed->valuestring); tsMnode.deployed = atoi(deployed->valuestring);
cJSON *dropped = cJSON_GetObjectItem(root, "dropped"); cJSON *dropped = cJSON_GetObjectItem(root, "dropped");
if (!dropped || dropped->type != cJSON_String) { if (!dropped || dropped->type != cJSON_String) {
dError("failed to read %s since dropped not found", tsMnode.file); dError("failed to read %s since dropped not found", file);
goto PRASE_MNODE_OVER; goto PRASE_MNODE_OVER;
} }
tsMnode.dropped = atoi(dropped->valuestring); tsMnode.dropped = atoi(dropped->valuestring);
dInfo("succcessed to read file %s", tsMnode.file); code = 0;
dInfo("succcessed to read file %s", file);
PRASE_MNODE_OVER: PRASE_MNODE_OVER:
if (content != NULL) free(content); if (content != NULL) free(content);
if (root != NULL) cJSON_Delete(root); if (root != NULL) cJSON_Delete(root);
if (fp != NULL) fclose(fp); if (fp != NULL) fclose(fp);
return 0; return code;
} }
static int32_t dnodeWriteMnode() { static int32_t dnodeWriteMnodeFile() {
FILE *fp = fopen(tsMnode.file, "w"); char file[PATH_MAX + 20] = {0};
char realfile[PATH_MAX + 20] = {0};
snprintf(file, sizeof(file), "%s/mnode.json.bak", tsDnodeDir);
snprintf(realfile, sizeof(realfile), "%s/mnode.json", tsDnodeDir);
FILE *fp = fopen(file, "w");
if (!fp) { if (!fp) {
dError("failed to write %s since %s", tsMnode.file, strerror(errno)); dError("failed to write %s since %s", file, strerror(errno));
return -1; return TSDB_CODE_DND_WRITE_MNODE_FILE_ERROR;
} }
int32_t len = 0; int32_t len = 0;
@ -97,119 +144,374 @@ static int32_t dnodeWriteMnode() {
taosFsyncFile(fileno(fp)); taosFsyncFile(fileno(fp));
fclose(fp); fclose(fp);
free(content); free(content);
terrno = 0;
dInfo("successed to write %s", tsMnode.file); int32_t code = taosRenameFile(file, realfile);
if (code != 0) {
dError("failed to rename %s since %s", file, tstrerror(code));
return TSDB_CODE_DND_WRITE_MNODE_FILE_ERROR;
}
dInfo("successed to write %s", realfile);
return 0; return 0;
} }
static int32_t dnodeStartMnode(SCreateMnodeMsg *pCfg) { static int32_t dnodeStartMnode() {
int32_t code = 0; int32_t code = dnodeAllocMnodeReadQueue();
if (code != 0) {
if (pCfg->dnodeId != dnodeGetDnodeId()) {
code = TSDB_CODE_DND_DNODE_ID_NOT_MATCHED;
dError("failed to start mnode since %s", tstrerror(code));
return code; return code;
} }
if (tsMnode.dropped) { code = dnodeAllocMnodeWriteQueue();
code = TSDB_CODE_DND_MNODE_ALREADY_DROPPED; if (code != 0) {
dError("failed to start mnode since %s", tstrerror(code));
return code; return code;
} }
if (tsMnode.deployed) { code = dnodeAllocMnodeApplyQueue();
dError("failed to start mnode since its already deployed"); if (code != 0) {
return 0; return code;
} }
code = dnodeAllocMnodeSyncQueue();
if (code != 0) {
return code;
}
taosWLockLatch(&tsMnode.latch);
tsMnode.deployed = 1; tsMnode.deployed = 1;
tsMnode.dropped = 0; taosWUnLockLatch(&tsMnode.latch);
code = dnodeWriteMnode(); return code;
if (code != 0) { }
static void dnodeStopMnode() {
taosWLockLatch(&tsMnode.latch);
tsMnode.deployed = 0; tsMnode.deployed = 0;
dError("failed to start mnode since %s", tstrerror(code)); taosWUnLockLatch(&tsMnode.latch);
return code;
dnodeReleaseMnode();
while (tsMnode.refCount > 0) taosMsleep(10);
while (!taosQueueEmpty(tsMnode.pReadQ)) taosMsleep(10);
while (!taosQueueEmpty(tsMnode.pApplyQ)) taosMsleep(10);
while (!taosQueueEmpty(tsMnode.pWriteQ)) taosMsleep(10);
while (!taosQueueEmpty(tsMnode.pSyncQ)) taosMsleep(10);
dnodeFreeMnodeReadQueue();
dnodeFreeMnodeWriteQueue();
dnodeFreeMnodeApplyQueue();
dnodeFreeMnodeSyncQueue();
} }
code = mnodeDeploy(); static int32_t dnodeUnDeployMnode() {
if (code != 0) {
tsMnode.deployed = 0;
dError("failed to start mnode since %s", tstrerror(code));
return code;
}
code = mnodeStart();
if (code != 0) {
tsMnode.deployed = 0;
dError("failed to start mnode since %s", tstrerror(code));
return code;
}
tsMnode.deployed = 1;
return 0;
}
static int32_t dnodeDropMnode(SDropMnodeMsg *pCfg) {
int32_t code = 0;
if (pCfg->dnodeId != dnodeGetDnodeId()) {
code = TSDB_CODE_DND_DNODE_ID_NOT_MATCHED;
dError("failed to drop mnode since %s", tstrerror(code));
return code;
}
if (tsMnode.dropped) {
code = TSDB_CODE_DND_MNODE_ALREADY_DROPPED;
dError("failed to drop mnode since %s", tstrerror(code));
return code;
}
if (!tsMnode.deployed) {
dError("failed to drop mnode since not deployed");
return 0;
}
mnodeStop();
tsMnode.deployed = 0;
tsMnode.dropped = 1; tsMnode.dropped = 1;
int32_t code = dnodeWriteMnodeFile();
code = dnodeWriteMnode();
if (code != 0) { if (code != 0) {
tsMnode.deployed = 1;
tsMnode.dropped = 0; tsMnode.dropped = 0;
dError("failed to drop mnode since %s", tstrerror(code)); dError("failed to undeploy mnode since %s", tstrerror(code));
return code; return code;
} }
mnodeUnDeploy(); dnodeStopMnode();
mnodeUnDeploy(tsMnodeDir);
dnodeWriteMnodeFile();
tsMnode.deployed = 0; return code;
}
static int32_t dnodeDeployMnode(SMnodeCfg *pCfg) {
int32_t code = mnodeDeploy(tsMnodeDir, pCfg);
if (code != 0) {
dError("failed to deploy mnode since %s", tstrerror(code));
return code;
}
code = dnodeStartMnode();
if (code != 0) {
dnodeUnDeployMnode();
dError("failed to deploy mnode since %s", tstrerror(code));
return code;
}
code = dnodeWriteMnodeFile();
if (code != 0) {
dnodeUnDeployMnode();
dError("failed to deploy mnode since %s", tstrerror(code));
return code;
}
dInfo("deploy mnode success");
return code;
}
static int32_t dnodeAlterMnode(SMnodeCfg *pCfg) {
int32_t code = dnodeAcquireMnode();
if (code == 0) {
code = mnodeAlter(pCfg);
dnodeReleaseMnode();
}
return code;
}
static SCreateMnodeMsg *dnodeParseCreateMnodeMsg(SRpcMsg *pRpcMsg) {
SCreateMnodeMsg *pMsg = pRpcMsg->pCont;
pMsg->dnodeId = htonl(pMsg->dnodeId);
for (int32_t i = 0; i < pMsg->replica; ++i) {
pMsg->replicas[i].port = htons(pMsg->replicas[i].port);
}
return pMsg;
}
static int32_t dnodeProcessCreateMnodeReq(SRpcMsg *pRpcMsg) {
SAlterMnodeMsg *pMsg = (SAlterMnodeMsg *)dnodeParseCreateMnodeMsg(pRpcMsg->pCont);
if (pMsg->dnodeId != dnodeGetDnodeId()) {
return TSDB_CODE_DND_MNODE_ID_NOT_MATCH_DNODE;
} else {
SMnodeCfg cfg = {0};
cfg.replica = pMsg->replica;
memcpy(cfg.replicas, pMsg->replicas, sizeof(SReplica) * sizeof(TSDB_MAX_REPLICA));
return dnodeDeployMnode(&cfg);
}
}
static int32_t dnodeProcessAlterMnodeReq(SRpcMsg *pRpcMsg) {
SAlterMnodeMsg *pMsg = (SAlterMnodeMsg *)dnodeParseCreateMnodeMsg(pRpcMsg->pCont);
if (pMsg->dnodeId != dnodeGetDnodeId()) {
return TSDB_CODE_DND_MNODE_ID_NOT_MATCH_DNODE;
} else {
SMnodeCfg cfg = {0};
cfg.replica = pMsg->replica;
memcpy(cfg.replicas, pMsg->replicas, sizeof(SReplica) * sizeof(TSDB_MAX_REPLICA));
return dnodeAlterMnode(&cfg);
}
}
static int32_t dnodeProcessDropMnodeReq(SRpcMsg *pMsg) {
SAlterMnodeMsg *pCfg = pMsg->pCont;
pCfg->dnodeId = htonl(pCfg->dnodeId);
if (pCfg->dnodeId != dnodeGetDnodeId()) {
return TSDB_CODE_DND_MNODE_ID_NOT_MATCH_DNODE;
} else {
return dnodeUnDeployMnode();
}
}
static void dnodeProcessMnodeMgmtQueue(void *unused, SRpcMsg *pMsg) {
int32_t code = 0;
switch (pMsg->msgType) {
case TSDB_MSG_TYPE_CREATE_MNODE_IN:
code = dnodeProcessCreateMnodeReq(pMsg);
break;
case TSDB_MSG_TYPE_ALTER_MNODE_IN:
code = dnodeProcessAlterMnodeReq(pMsg);
break;
case TSDB_MSG_TYPE_DROP_MNODE_IN:
code = dnodeProcessDropMnodeReq(pMsg);
break;
default:
code = TSDB_CODE_DND_MSG_NOT_PROCESSED;
break;
}
SRpcMsg rsp = {.code = code, .handle = pMsg->handle};
rpcSendResponse(&rsp);
rpcFreeCont(pMsg->pCont);
taosFreeQitem(pMsg);
}
static void dnodeProcessMnodeReadQueue(void *unused, SMnodeMsg *pMsg) { mnodeProcessMsg(pMsg, MN_MSG_TYPE_READ); }
static void dnodeProcessMnodeWriteQueue(void *unused, SMnodeMsg *pMsg) { mnodeProcessMsg(pMsg, MN_MSG_TYPE_WRITE); }
static void dnodeProcessMnodeApplyQueue(void *unused, SMnodeMsg *pMsg) { mnodeProcessMsg(pMsg, MN_MSG_TYPE_APPLY); }
static void dnodeProcessMnodeSyncQueue(void *unused, SMnodeMsg *pMsg) { mnodeProcessMsg(pMsg, MN_MSG_TYPE_SYNC); }
static int32_t dnodeWriteMnodeMsgToQueue(taos_queue pQueue, SRpcMsg *pRpcMsg) {
int32_t code = 0;
if (pQueue == NULL) {
code = TSDB_CODE_DND_MSG_NOT_PROCESSED;
} else {
SMnodeMsg *pMsg = mnodeInitMsg(1);
if (pMsg == NULL) {
code = TSDB_CODE_DND_OUT_OF_MEMORY;
} else {
mnodeAppendMsg(pMsg, pRpcMsg);
code = taosWriteQitem(pQueue, pMsg);
}
}
if (code != TSDB_CODE_SUCCESS) {
SRpcMsg rsp = {.handle = pRpcMsg->handle, .code = code};
rpcSendResponse(&rsp);
rpcFreeCont(pRpcMsg->pCont);
}
}
void dnodeProcessMnodeMgmtMsg(SRpcMsg *pMsg, SEpSet *pEpSet) { dnodeWriteMnodeMsgToQueue(tsMnode.pMgmtQ, pMsg); }
void dnodeProcessMnodeWriteMsg(SRpcMsg *pMsg, SEpSet *pEpSet) {
if (dnodeAcquireMnode() == 0) {
dnodeWriteMnodeMsgToQueue(tsMnode.pWriteQ, pMsg);
dnodeReleaseMnode();
} else {
dnodeSendRedirectMsg(pMsg, 0);
}
}
void dnodeProcessMnodeSyncMsg(SRpcMsg *pMsg, SEpSet *pEpSet) {
int32_t code = dnodeAcquireMnode();
if (code == 0) {
dnodeWriteMnodeMsgToQueue(tsMnode.pSyncQ, pMsg);
dnodeReleaseMnode();
} else {
SRpcMsg rsp = {.handle = pMsg->handle, .code = code};
rpcSendResponse(&rsp);
rpcFreeCont(pMsg->pCont);
}
}
void dnodeProcessMnodeReadMsg(SRpcMsg *pMsg, SEpSet *pEpSet) {
if (dnodeAcquireMnode() == 0) {
dnodeWriteMnodeMsgToQueue(tsMnode.pReadQ, pMsg);
dnodeReleaseMnode();
} else {
dnodeSendRedirectMsg(pMsg, 0);
}
}
static int32_t dnodePutMsgIntoMnodeApplyQueue(SMnodeMsg *pMsg) {
int32_t code = dnodeAcquireMnode();
if (code != 0) return code;
code = taosWriteQitem(tsMnode.pApplyQ, pMsg);
dnodeReleaseMnode();
return code;
}
static int32_t dnodeAllocMnodeMgmtQueue() {
tsMnode.pMgmtQ = tWorkerAllocQueue(&tsMnode.mgmtPool, NULL, (FProcessItem)dnodeProcessMnodeMgmtQueue);
if (tsMnode.pMgmtQ == NULL) {
return TSDB_CODE_DND_OUT_OF_MEMORY;
}
return 0; return 0;
} }
static void dnodeProcessCreateMnodeReq(SRpcMsg *pMsg) { static void dnodeFreeMnodeMgmtQueue() {
SCreateMnodeMsg *pCfg = pMsg->pCont; tWorkerFreeQueue(&tsMnode.mgmtPool, tsMnode.pMgmtQ);
pCfg->dnodeId = htonl(pCfg->dnodeId); tsMnode.pMgmtQ = NULL;
int32_t code = dnodeStartMnode(pCfg);
SRpcMsg rspMsg = {.handle = pMsg->handle, .code = code};
rpcSendResponse(&rspMsg);
rpcFreeCont(pMsg->pCont);
} }
static void dnodeProcessDropMnodeReq(SRpcMsg *pMsg) { static int32_t dnodeInitMnodeMgmtWorker() {
SDropMnodeMsg *pCfg = pMsg->pCont; SWorkerPool *pPool = &tsMnode.mgmtPool;
pCfg->dnodeId = htonl(pCfg->dnodeId); pPool->name = "mnode-mgmt";
pPool->min = 1;
int32_t code = dnodeDropMnode(pCfg); pPool->max = 1;
SRpcMsg rspMsg = {.handle = pMsg->handle, .code = code}; return tWorkerInit(pPool);
rpcSendResponse(&rspMsg);
rpcFreeCont(pMsg->pCont);
} }
static void dnodeCleanupMnodeMgmtWorker() { tWorkerCleanup(&tsMnode.mgmtPool); }
static int32_t dnodeAllocMnodeReadQueue() {
tsMnode.pReadQ = tWorkerAllocQueue(&tsMnode.readPool, NULL, (FProcessItem)dnodeProcessMnodeReadQueue);
if (tsMnode.pReadQ == NULL) {
return TSDB_CODE_DND_OUT_OF_MEMORY;
}
return 0;
}
static void dnodeFreeMnodeReadQueue() {
tWorkerFreeQueue(&tsMnode.readPool, tsMnode.pReadQ);
tsMnode.pReadQ = NULL;
}
static int32_t dnodeInitMnodeReadWorker() {
SWorkerPool *pPool = &tsMnode.readPool;
pPool->name = "mnode-read";
pPool->min = 0;
pPool->max = 1;
return tWorkerInit(pPool);
}
static void dnodeCleanupMnodeReadWorker() { tWorkerCleanup(&tsMnode.readPool); }
static int32_t dnodeAllocMnodeWriteQueue() {
tsMnode.pWriteQ = tWorkerAllocQueue(&tsMnode.writePool, NULL, (FProcessItem)dnodeProcessMnodeWriteQueue);
if (tsMnode.pWriteQ == NULL) {
return TSDB_CODE_DND_OUT_OF_MEMORY;
}
return 0;
}
static void dnodeFreeMnodeWriteQueue() {
tWorkerFreeQueue(&tsMnode.writePool, tsMnode.pWriteQ);
tsMnode.pWriteQ = NULL;
}
static int32_t dnodeAllocMnodeApplyQueue() {
tsMnode.pApplyQ = tWorkerAllocQueue(&tsMnode.writePool, NULL, (FProcessItem)dnodeProcessMnodeApplyQueue);
if (tsMnode.pApplyQ == NULL) {
return TSDB_CODE_DND_OUT_OF_MEMORY;
}
return 0;
}
static void dnodeFreeMnodeApplyQueue() {
tWorkerFreeQueue(&tsMnode.writePool, tsMnode.pApplyQ);
tsMnode.pApplyQ = NULL;
}
static int32_t dnodeInitMnodeWriteWorker() {
SWorkerPool *pPool = &tsMnode.writePool;
pPool->name = "mnode-write";
pPool->min = 0;
pPool->max = 1;
return tWorkerInit(pPool);
}
static void dnodeCleanupMnodeWriteWorker() { tWorkerCleanup(&tsMnode.writePool); }
static int32_t dnodeAllocMnodeSyncQueue() {
tsMnode.pSyncQ = tWorkerAllocQueue(&tsMnode.syncPool, NULL, (FProcessItem)dnodeProcessMnodeSyncQueue);
if (tsMnode.pSyncQ == NULL) {
return TSDB_CODE_DND_OUT_OF_MEMORY;
}
return 0;
}
static void dnodeFreeMnodeSyncQueue() {
tWorkerFreeQueue(&tsMnode.syncPool, tsMnode.pSyncQ);
tsMnode.pSyncQ = NULL;
}
static int32_t dnodeInitMnodeSyncWorker() {
SWorkerPool *pPool = &tsMnode.syncPool;
pPool->name = "mnode-sync";
pPool->min = 0;
pPool->max = 1;
return tWorkerInit(pPool);
}
static void dnodeCleanupMnodeSyncWorker() { tWorkerCleanup(&tsMnode.syncPool); }
static int32_t dnodeInitMnodeModule() {
taosInitRWLatch(&tsMnode.latch);
SMnodePara para;
para.dnodeId = dnodeGetDnodeId();
para.clusterId = dnodeGetClusterId();
para.SendMsgToDnode = dnodeSendMsgToDnode;
para.SendMsgToMnode = dnodeSendMsgToMnode;
para.SendRedirectMsg = dnodeSendRedirectMsg;
return mnodeInit(para);
}
static void dnodeCleanupMnodeModule() { mnodeCleanup(); }
static bool dnodeNeedDeployMnode() { static bool dnodeNeedDeployMnode() {
if (dnodeGetDnodeId() > 0) return false; if (dnodeGetDnodeId() > 0) return false;
if (dnodeGetClusterId() > 0) return false; if (dnodeGetClusterId() > 0) return false;
@ -217,76 +519,77 @@ static bool dnodeNeedDeployMnode() {
return true; return true;
} }
int32_t dnodeInitMnode() { static int32_t dnodeOpenMnode() {
tsMnode.dropped = 0; int32_t code = dnodeReadMnodeFile();
tsMnode.deployed = 0;
snprintf(tsMnode.file, sizeof(tsMnode.file), "%s/mnode.json", tsDnodeDir);
SMnodePara para;
para.SendMsgToDnode = dnodeSendMsgToDnode;
para.SendMsgToMnode = dnodeSendMsgToMnode;
para.SendRedirectMsg = dnodeSendRedirectMsg;
para.dnodeId = dnodeGetDnodeId();
para.clusterId = dnodeGetClusterId();
int32_t code = mnodeInit(para);
if (code != 0) { if (code != 0) {
dError("failed to init mnode module since %s", tstrerror(code)); dError("failed to read open mnode since %s", tstrerror(code));
return code;
}
code = dnodeReadMnode();
if (code != 0) {
dError("failed to read file:%s since %s", tsMnode.file, tstrerror(code));
return code; return code;
} }
if (tsMnode.dropped) { if (tsMnode.dropped) {
dError("mnode already dropped, undeploy it"); dInfo("mnode already dropped, undeploy it");
mnodeUnDeploy(); return dnodeUnDeployMnode();
return 0;
} }
if (!tsMnode.deployed) { if (!tsMnode.deployed) {
bool needDeploy = dnodeNeedDeployMnode(); bool needDeploy = dnodeNeedDeployMnode();
if (needDeploy) { if (!needDeploy) return 0;
code = mnodeDeploy();
dInfo("start to deploy mnode");
SMnodeCfg cfg = {.replica = 1};
cfg.replicas[0].port = tsServerPort;
tstrncpy(cfg.replicas[0].fqdn, tsLocalFqdn, TSDB_FQDN_LEN);
return dnodeDeployMnode(&cfg);
} else { } else {
return 0; dInfo("start to open mnode");
return dnodeStartMnode();
} }
}
static void dnodeCloseMnode() {
if (dnodeAcquireMnode() == 0) {
dnodeStopMnode();
}
}
int32_t dnodeInitMnode() {
dInfo("dnode-mnode start to init");
SSteps *pSteps = taosStepInit(6, dnodeReportStartup);
taosStepAdd(pSteps, "dnode-mnode-env", dnodeInitMnodeModule, dnodeCleanupMnodeModule);
taosStepAdd(pSteps, "dnode-mnode-mgmt", dnodeInitMnodeMgmtWorker, dnodeCleanupMnodeMgmtWorker);
taosStepAdd(pSteps, "dnode-mnode-read", dnodeInitMnodeReadWorker, dnodeCleanupMnodeReadWorker);
taosStepAdd(pSteps, "dnode-mnode-write", dnodeInitMnodeWriteWorker, dnodeCleanupMnodeWriteWorker);
taosStepAdd(pSteps, "dnode-mnode-sync", dnodeInitMnodeSyncWorker, dnodeCleanupMnodeSyncWorker);
taosStepAdd(pSteps, "dnode-mnode", dnodeOpenMnode, dnodeCloseMnode);
tsMnode.pSteps = pSteps;
int32_t code = taosStepExec(pSteps);
if (code != 0) { if (code != 0) {
dError("failed to deploy mnode since %s", tstrerror(code)); dError("dnode-mnode init failed since %s", tstrerror(code));
return code; } else {
dInfo("dnode-mnode is initialized");
} }
tsMnode.deployed = 1;
}
return mnodeStart();
} }
void dnodeCleanupMnode() { void dnodeCleanupMnode() {
if (tsMnode.deployed) { if (tsMnode.pSteps == NULL) {
mnodeStop(); dInfo("dnode-mnode start to clean up");
} taosStepCleanup(tsMnode.pSteps);
tsMnode.pSteps = NULL;
mnodeCleanup(); dInfo("dnode-mnode is cleaned up");
}
void dnodeProcessMnodeMsg(SRpcMsg *pMsg, SEpSet *pEpSet) {
switch (pMsg->msgType) {
case TSDB_MSG_TYPE_CREATE_MNODE_IN:
dnodeProcessCreateMnodeReq(pMsg);
break;
case TSDB_MSG_TYPE_DROP_MNODE_IN:
dnodeProcessDropMnodeReq(pMsg);
break;
default:
mnodeProcessMsg(pMsg);
} }
} }
int32_t dnodeGetUserAuthFromMnode(char *user, char *spi, char *encrypt, char *secret, char *ckey) { int32_t dnodeGetUserAuthFromMnode(char *user, char *spi, char *encrypt, char *secret, char *ckey) {
return mnodeRetriveAuth(user, spi, encrypt, secret, ckey); int32_t code = dnodeAcquireMnode();
if (code != 0) {
dTrace("failed to get user auth since mnode not deployed");
return code;
}
code = mnodeRetriveAuth(user, spi, encrypt, secret, ckey);
dnodeReleaseMnode();
return code;
} }

100
source/dnode/mgmt/src/dnodeTransport.c
View File

@ -51,76 +51,78 @@ static void dnodeInitMsgFp() {
tsTrans.msgFp[TSDB_MSG_TYPE_MQ_RESET] = dnodeProcessVnodeWriteMsg; tsTrans.msgFp[TSDB_MSG_TYPE_MQ_RESET] = dnodeProcessVnodeWriteMsg;
// msg from client to mnode // msg from client to mnode
tsTrans.msgFp[TSDB_MSG_TYPE_CONNECT] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_CONNECT] = dnodeProcessMnodeReadMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_ACCT] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_ACCT] = dnodeProcessMnodeWriteMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_ACCT] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_ACCT] = dnodeProcessMnodeWriteMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_DROP_ACCT] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_DROP_ACCT] = dnodeProcessMnodeWriteMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_USER] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_USER] = dnodeProcessMnodeWriteMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_USER] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_USER] = dnodeProcessMnodeWriteMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_DROP_USER] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_DROP_USER] = dnodeProcessMnodeWriteMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_DNODE] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_DNODE] = dnodeProcessMnodeWriteMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_CONFIG_DNODE] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_CONFIG_DNODE] = dnodeProcessMnodeWriteMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_DROP_DNODE] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_DROP_DNODE] = dnodeProcessMnodeWriteMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_DB] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_DB] = dnodeProcessMnodeWriteMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_DROP_DB] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_DROP_DB] = dnodeProcessMnodeWriteMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_USE_DB] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_USE_DB] = dnodeProcessMnodeWriteMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_DB] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_DB] = dnodeProcessMnodeWriteMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_SYNC_DB] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_SYNC_DB] = dnodeProcessMnodeWriteMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_TOPIC] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_TOPIC] = dnodeProcessMnodeWriteMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_DROP_TOPIC] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_DROP_TOPIC] = dnodeProcessMnodeWriteMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_TOPIC] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_TOPIC] = dnodeProcessMnodeWriteMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_FUNCTION] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_FUNCTION] = dnodeProcessMnodeWriteMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_FUNCTION] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_FUNCTION] = dnodeProcessMnodeWriteMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_DROP_FUNCTION] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_DROP_FUNCTION] = dnodeProcessMnodeWriteMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_STABLE] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_STABLE] = dnodeProcessMnodeWriteMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_DROP_STABLE] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_STABLE] = dnodeProcessMnodeWriteMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_STABLE] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_DROP_STABLE] = dnodeProcessMnodeWriteMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_STABLE_VGROUP] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_STABLE_VGROUP] = dnodeProcessMnodeReadMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_KILL_QUERY] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_KILL_QUERY] = dnodeProcessMnodeWriteMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_KILL_CONN] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_KILL_CONN] = dnodeProcessMnodeWriteMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_HEARTBEAT] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_HEARTBEAT] = dnodeProcessMnodeReadMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_SHOW] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_SHOW] = dnodeProcessMnodeReadMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_SHOW_RETRIEVE] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_SHOW_RETRIEVE] = dnodeProcessMnodeReadMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_SHOW_RETRIEVE_FUNC] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_SHOW_RETRIEVE_FUNC] = dnodeProcessMnodeReadMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_COMPACT_VNODE] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_COMPACT_VNODE] = dnodeProcessMnodeWriteMsg;
// message from client to dnode // message from client to dnode
tsTrans.msgFp[TSDB_MSG_TYPE_NETWORK_TEST] = dnodeProcessDnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_NETWORK_TEST] = dnodeProcessDnodeMsg;
// message from mnode to vnode // message from mnode to vnode
tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_STABLE_IN] = dnodeProcessVnodeWriteMsg; tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_STABLE_IN] = dnodeProcessVnodeWriteMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_STABLE_IN_RSP] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_STABLE_IN_RSP] = dnodeProcessMnodeWriteMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_STABLE_IN] = dnodeProcessVnodeWriteMsg; tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_STABLE_IN] = dnodeProcessVnodeWriteMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_STABLE_IN_RSP] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_STABLE_IN_RSP] = dnodeProcessMnodeWriteMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_DROP_STABLE_IN] = dnodeProcessVnodeWriteMsg; tsTrans.msgFp[TSDB_MSG_TYPE_DROP_STABLE_IN] = dnodeProcessVnodeWriteMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_DROP_STABLE_IN] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_DROP_STABLE_IN] = dnodeProcessMnodeWriteMsg;
// message from mnode to dnode // message from mnode to dnode
tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_VNODE_IN] = dnodeProcessVnodeMgmtMsg; tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_VNODE_IN] = dnodeProcessVnodeMgmtMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_VNODE_IN_RSP] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_VNODE_IN_RSP] = dnodeProcessMnodeWriteMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_VNODE_IN] = dnodeProcessVnodeMgmtMsg; tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_VNODE_IN] = dnodeProcessVnodeMgmtMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_VNODE_IN_RSP] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_VNODE_IN_RSP] = dnodeProcessMnodeWriteMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_DROP_VNODE_IN] = dnodeProcessVnodeMgmtMsg; tsTrans.msgFp[TSDB_MSG_TYPE_DROP_VNODE_IN] = dnodeProcessVnodeMgmtMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_DROP_VNODE_IN_RSP] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_DROP_VNODE_IN_RSP] = dnodeProcessMnodeWriteMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_SYNC_VNODE_IN] = dnodeProcessVnodeMgmtMsg; tsTrans.msgFp[TSDB_MSG_TYPE_SYNC_VNODE_IN] = dnodeProcessVnodeMgmtMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_SYNC_VNODE_IN_RSP] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_SYNC_VNODE_IN_RSP] = dnodeProcessMnodeWriteMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_AUTH_VNODE_IN] = dnodeProcessVnodeMgmtMsg; tsTrans.msgFp[TSDB_MSG_TYPE_AUTH_VNODE_IN] = dnodeProcessVnodeMgmtMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_AUTH_VNODE_IN_RSP] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_AUTH_VNODE_IN_RSP] = dnodeProcessMnodeWriteMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_COMPACT_VNODE_IN] = dnodeProcessVnodeMgmtMsg; tsTrans.msgFp[TSDB_MSG_TYPE_COMPACT_VNODE_IN] = dnodeProcessVnodeMgmtMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_COMPACT_VNODE_IN_RSP] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_COMPACT_VNODE_IN_RSP] = dnodeProcessMnodeWriteMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_MNODE_IN] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_MNODE_IN] = dnodeProcessMnodeMgmtMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_MNODE_IN_RSP] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_MNODE_IN_RSP] = dnodeProcessMnodeWriteMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_DROP_MNODE_IN] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_MNODE_IN] = dnodeProcessMnodeMgmtMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_DROP_MNODE_IN_RSP] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_MNODE_IN_RSP] = dnodeProcessMnodeWriteMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_DROP_MNODE_IN] = dnodeProcessMnodeMgmtMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_DROP_MNODE_IN_RSP] = dnodeProcessMnodeWriteMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_CONFIG_DNODE_IN] = dnodeProcessDnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_CONFIG_DNODE_IN] = dnodeProcessDnodeMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_CONFIG_DNODE_IN_RSP] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_CONFIG_DNODE_IN_RSP] = dnodeProcessMnodeWriteMsg;
// message from dnode to mnode // message from dnode to mnode
tsTrans.msgFp[TSDB_MSG_TYPE_AUTH] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_AUTH] = dnodeProcessMnodeReadMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_AUTH_RSP] = dnodeProcessDnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_AUTH_RSP] = dnodeProcessDnodeMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_GRANT] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_GRANT] = dnodeProcessMnodeWriteMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_GRANT_RSP] = dnodeProcessDnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_GRANT_RSP] = dnodeProcessDnodeMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_STATUS] = dnodeProcessMnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_STATUS] = dnodeProcessMnodeWriteMsg;
tsTrans.msgFp[TSDB_MSG_TYPE_STATUS_RSP] = dnodeProcessDnodeMsg; tsTrans.msgFp[TSDB_MSG_TYPE_STATUS_RSP] = dnodeProcessDnodeMsg;
} }

2
source/dnode/mgmt/src/dnodeVnodes.c
View File

@ -983,7 +983,7 @@ static int32_t dnodeInitVnodeModule() {
int32_t dnodeInitVnodes() { int32_t dnodeInitVnodes() {
dInfo("dnode-vnodes start to init"); dInfo("dnode-vnodes start to init");
SSteps *pSteps = taosStepInit(3, dnodeReportStartup); SSteps *pSteps = taosStepInit(6, dnodeReportStartup);
taosStepAdd(pSteps, "dnode-vnode-env", dnodeInitVnodeModule, vnodeCleanup); taosStepAdd(pSteps, "dnode-vnode-env", dnodeInitVnodeModule, vnodeCleanup);
taosStepAdd(pSteps, "dnode-vnode-mgmt", dnodeInitVnodeMgmtWorker, dnodeCleanupVnodeMgmtWorker); taosStepAdd(pSteps, "dnode-vnode-mgmt", dnodeInitVnodeMgmtWorker, dnodeCleanupVnodeMgmtWorker);
taosStepAdd(pSteps, "dnode-vnode-read", dnodeInitVnodeReadWorker, dnodeCleanupVnodeReadWorker); taosStepAdd(pSteps, "dnode-vnode-read", dnodeInitVnodeReadWorker, dnodeCleanupVnodeReadWorker);

1
source/dnode/mnode/inc/mnodeWorker.h
View File

@ -24,7 +24,6 @@ extern "C" {
int32_t mnodeInitWorker(); int32_t mnodeInitWorker();
void mnodeCleanupWorker(); void mnodeCleanupWorker();
void mnodeProcessMsg(SRpcMsg *rpcMsg);
void mnodeSendRsp(SMnMsg *pMsg, int32_t code); void mnodeSendRsp(SMnMsg *pMsg, int32_t code);
void mnodeReDispatchToWriteQueue(SMnMsg *pMsg); void mnodeReDispatchToWriteQueue(SMnMsg *pMsg);

33
source/dnode/mnode/src/mnodeWorker.c
View File

@ -39,7 +39,7 @@ static struct {
void (*msgFp[TSDB_MSG_TYPE_MAX])(SRpcMsg *pMsg); void (*msgFp[TSDB_MSG_TYPE_MAX])(SRpcMsg *pMsg);
} tsMworker = {0}; } tsMworker = {0};
static SMnMsg *mnodeInitMsg(SRpcMsg *pRpcMsg) { static SMnMsg *mnodeInitMsg2(SRpcMsg *pRpcMsg) {
int32_t size = sizeof(SMnMsg) + pRpcMsg->contLen; int32_t size = sizeof(SMnMsg) + pRpcMsg->contLen;
SMnMsg *pMsg = taosAllocateQitem(size); SMnMsg *pMsg = taosAllocateQitem(size);
@ -62,7 +62,7 @@ static SMnMsg *mnodeInitMsg(SRpcMsg *pRpcMsg) {
return pMsg; return pMsg;
} }
static void mnodeCleanupMsg(SMnMsg *pMsg) { static void mnodeCleanupMsg2(SMnMsg *pMsg) {
if (pMsg == NULL) return; if (pMsg == NULL) return;
if (pMsg->rpcMsg.pCont != pMsg->pCont) { if (pMsg->rpcMsg.pCont != pMsg->pCont) {
tfree(pMsg->rpcMsg.pCont); tfree(pMsg->rpcMsg.pCont);
@ -75,7 +75,7 @@ static void mnodeDispatchToWriteQueue(SRpcMsg *pRpcMsg) {
if (mnodeGetStatus() != MN_STATUS_READY || tsMworker.writeQ == NULL) { if (mnodeGetStatus() != MN_STATUS_READY || tsMworker.writeQ == NULL) {
mnodeSendRedirectMsg(pRpcMsg, true); mnodeSendRedirectMsg(pRpcMsg, true);
} else { } else {
SMnMsg *pMsg = mnodeInitMsg(pRpcMsg); SMnMsg *pMsg = mnodeInitMsg2(pRpcMsg);
if (pMsg == NULL) { if (pMsg == NULL) {
SRpcMsg rpcRsp = {.handle = pRpcMsg->handle, .code = TSDB_CODE_MND_INVALID_USER}; SRpcMsg rpcRsp = {.handle = pRpcMsg->handle, .code = TSDB_CODE_MND_INVALID_USER};
rpcSendResponse(&rpcRsp); rpcSendResponse(&rpcRsp);
@ -91,7 +91,7 @@ static void mnodeDispatchToWriteQueue(SRpcMsg *pRpcMsg) {
void mnodeReDispatchToWriteQueue(SMnMsg *pMsg) { void mnodeReDispatchToWriteQueue(SMnMsg *pMsg) {
if (mnodeGetStatus() != MN_STATUS_READY || tsMworker.writeQ == NULL) { if (mnodeGetStatus() != MN_STATUS_READY || tsMworker.writeQ == NULL) {
mnodeSendRedirectMsg(&pMsg->rpcMsg, true); mnodeSendRedirectMsg(&pMsg->rpcMsg, true);
mnodeCleanupMsg(pMsg); mnodeCleanupMsg2(pMsg);
} else { } else {
taosWriteQitem(tsMworker.writeQ, pMsg); taosWriteQitem(tsMworker.writeQ, pMsg);
} }
@ -101,7 +101,7 @@ static void mnodeDispatchToReadQueue(SRpcMsg *pRpcMsg) {
if (mnodeGetStatus() != MN_STATUS_READY || tsMworker.readQ == NULL) { if (mnodeGetStatus() != MN_STATUS_READY || tsMworker.readQ == NULL) {
mnodeSendRedirectMsg(pRpcMsg, true); mnodeSendRedirectMsg(pRpcMsg, true);
} else { } else {
SMnMsg *pMsg = mnodeInitMsg(pRpcMsg); SMnMsg *pMsg = mnodeInitMsg2(pRpcMsg);
if (pMsg == NULL) { if (pMsg == NULL) {
SRpcMsg rpcRsp = {.handle = pRpcMsg->handle, .code = TSDB_CODE_MND_INVALID_USER}; SRpcMsg rpcRsp = {.handle = pRpcMsg->handle, .code = TSDB_CODE_MND_INVALID_USER};
rpcSendResponse(&rpcRsp); rpcSendResponse(&rpcRsp);
@ -118,7 +118,7 @@ static void mnodeDispatchToPeerQueue(SRpcMsg *pRpcMsg) {
if (mnodeGetStatus() != MN_STATUS_READY || tsMworker.peerReqQ == NULL) { if (mnodeGetStatus() != MN_STATUS_READY || tsMworker.peerReqQ == NULL) {
mnodeSendRedirectMsg(pRpcMsg, false); mnodeSendRedirectMsg(pRpcMsg, false);
} else { } else {
SMnMsg *pMsg = mnodeInitMsg(pRpcMsg); SMnMsg *pMsg = mnodeInitMsg2(pRpcMsg);
if (pMsg == NULL) { if (pMsg == NULL) {
SRpcMsg rpcRsp = {.handle = pRpcMsg->handle, .code = TSDB_CODE_MND_INVALID_USER}; SRpcMsg rpcRsp = {.handle = pRpcMsg->handle, .code = TSDB_CODE_MND_INVALID_USER};
rpcSendResponse(&rpcRsp); rpcSendResponse(&rpcRsp);
@ -133,7 +133,7 @@ static void mnodeDispatchToPeerQueue(SRpcMsg *pRpcMsg) {
} }
void mnodeDispatchToPeerRspQueue(SRpcMsg *pRpcMsg) { void mnodeDispatchToPeerRspQueue(SRpcMsg *pRpcMsg) {
SMnMsg *pMsg = mnodeInitMsg(pRpcMsg); SMnMsg *pMsg = mnodeInitMsg2(pRpcMsg);
if (pMsg == NULL) { if (pMsg == NULL) {
SRpcMsg rpcRsp = {.handle = pRpcMsg->handle, .code = TSDB_CODE_MND_INVALID_USER}; SRpcMsg rpcRsp = {.handle = pRpcMsg->handle, .code = TSDB_CODE_MND_INVALID_USER};
rpcSendResponse(&rpcRsp); rpcSendResponse(&rpcRsp);
@ -162,7 +162,7 @@ void mnodeSendRsp(SMnMsg *pMsg, int32_t code) {
}; };
rpcSendResponse(&rpcRsp); rpcSendResponse(&rpcRsp);
mnodeCleanupMsg(pMsg); mnodeCleanupMsg2(pMsg);
} }
static void mnodeInitMsgFp() { static void mnodeInitMsgFp() {
@ -405,7 +405,7 @@ static void mnodeProcessPeerRsp(SMnMsg *pMsg, void *unused) {
if (!mnodeIsMaster()) { if (!mnodeIsMaster()) {
mError("msg:%p, ahandle:%p type:%s not processed for not master", pRpcMsg, pRpcMsg->ahandle, taosMsg[msgType]); mError("msg:%p, ahandle:%p type:%s not processed for not master", pRpcMsg, pRpcMsg->ahandle, taosMsg[msgType]);
mnodeCleanupMsg(pMsg); mnodeCleanupMsg2(pMsg);
} }
if (tsMworker.peerRspFp[msgType]) { if (tsMworker.peerRspFp[msgType]) {
@ -414,7 +414,7 @@ static void mnodeProcessPeerRsp(SMnMsg *pMsg, void *unused) {
mError("msg:%p, ahandle:%p type:%s is not processed", pRpcMsg, pRpcMsg->ahandle, taosMsg[msgType]); mError("msg:%p, ahandle:%p type:%s is not processed", pRpcMsg, pRpcMsg->ahandle, taosMsg[msgType]);
} }
mnodeCleanupMsg(pMsg); mnodeCleanupMsg2(pMsg);
} }
int32_t mnodeInitWorker() { int32_t mnodeInitWorker() {
@ -486,10 +486,9 @@ void mnodeCleanupWorker() {
mInfo("mnode worker is closed"); mInfo("mnode worker is closed");
} }
void mnodeProcessMsg(SRpcMsg *pMsg) { SMnodeMsg *mnodeInitMsg(int32_t msgNum) { return NULL; }
if (tsMworker.msgFp[pMsg->msgType]) {
(*tsMworker.msgFp[pMsg->msgType])(pMsg); int32_t mnodeAppendMsg(SMnodeMsg *pMsg, SRpcMsg *pRpcMsg) { return 0; }
} else {
assert(0); void mnodeCleanupMsg(SMnodeMsg *pMsg) {}
} void mnodeProcessMsg(SMnodeMsg *pMsg, EMnMsgType msgType) {}
}

8
source/dnode/mnode/src/mondeInt.c
View File

@ -70,6 +70,7 @@ static int32_t mnodeSetPara(SMnodePara para) {
if (tsMint.para.SendMsgToDnode == NULL) return -1; if (tsMint.para.SendMsgToDnode == NULL) return -1;
if (tsMint.para.SendMsgToMnode == NULL) return -1; if (tsMint.para.SendMsgToMnode == NULL) return -1;
if (tsMint.para.SendRedirectMsg == NULL) return -1; if (tsMint.para.SendRedirectMsg == NULL) return -1;
if (tsMint.para.PutMsgIntoApplyQueue == NULL) return -1;
if (tsMint.para.dnodeId < 0) return -1; if (tsMint.para.dnodeId < 0) return -1;
if (tsMint.para.clusterId < 0) return -1; if (tsMint.para.clusterId < 0) return -1;
@ -141,7 +142,7 @@ static bool mnodeNeedDeploy() {
return true; return true;
} }
int32_t mnodeDeploy() { int32_t mnodeDeploy(char *path, SMnodeCfg *pCfg) {
if (tsMint.state != MN_STATUS_UNINIT) { if (tsMint.state != MN_STATUS_UNINIT) {
mError("failed to deploy mnode since its deployed"); mError("failed to deploy mnode since its deployed");
return 0; return 0;
@ -176,7 +177,7 @@ int32_t mnodeDeploy() {
return 0; return 0;
} }
void mnodeUnDeploy() { void mnodeUnDeploy(char *path) {
sdbUnDeploy(); sdbUnDeploy();
mnodeCleanup(); mnodeCleanup();
} }
@ -244,5 +245,6 @@ void mnodeCleanup() {
} }
} }
int32_t mnodeStart() { return 0; } int32_t mnodeStart(char *path, SMnodeCfg *pCfg) { return 0; }
int32_t mnodeAlter(SMnodeCfg *pCfg) { return 0; }
void mnodeStop() {} void mnodeStop() {}

8
source/util/src/terror.c
View File

@ -228,14 +228,18 @@ TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOPIC_ALREADY_EXIST, "Topic already exists"
// dnode // dnode
TAOS_DEFINE_ERROR(TSDB_CODE_DND_MSG_NOT_PROCESSED, "Message not processed") TAOS_DEFINE_ERROR(TSDB_CODE_DND_MSG_NOT_PROCESSED, "Message not processed")
TAOS_DEFINE_ERROR(TSDB_CODE_DND_OUT_OF_MEMORY, "Dnode out of memory") TAOS_DEFINE_ERROR(TSDB_CODE_DND_OUT_OF_MEMORY, "Dnode out of memory")
TAOS_DEFINE_ERROR(TSDB_CODE_DND_DNODE_ID_NOT_MATCHED, "Dnode Id not matched") TAOS_DEFINE_ERROR(TSDB_CODE_DND_MNODE_ID_NOT_MATCH_DNODE, "Mnode Id not match Dnode")
TAOS_DEFINE_ERROR(TSDB_CODE_DND_MNODE_ALREADY_DROPPED, "Mnode already deployed") TAOS_DEFINE_ERROR(TSDB_CODE_DND_MNODE_ALREADY_DEPLOYED, "Mnode already deployed")
TAOS_DEFINE_ERROR(TSDB_CODE_DND_MNODE_NOT_DEPLOYED, "Mnode not deployed")
TAOS_DEFINE_ERROR(TSDB_CODE_DND_READ_MNODE_FILE_ERROR, "Read mnode.json error")
TAOS_DEFINE_ERROR(TSDB_CODE_DND_WRITE_MNODE_FILE_ERROR, "Write mnode.json error")
TAOS_DEFINE_ERROR(TSDB_CODE_DND_NO_WRITE_ACCESS, "No permission for disk files in dnode") TAOS_DEFINE_ERROR(TSDB_CODE_DND_NO_WRITE_ACCESS, "No permission for disk files in dnode")
TAOS_DEFINE_ERROR(TSDB_CODE_DND_INVALID_MSG_LEN, "Invalid message length") TAOS_DEFINE_ERROR(TSDB_CODE_DND_INVALID_MSG_LEN, "Invalid message length")
TAOS_DEFINE_ERROR(TSDB_CODE_DND_ACTION_IN_PROGRESS, "Action in progress") TAOS_DEFINE_ERROR(TSDB_CODE_DND_ACTION_IN_PROGRESS, "Action in progress")
TAOS_DEFINE_ERROR(TSDB_CODE_DND_TOO_MANY_VNODES, "Too many vnode directories") TAOS_DEFINE_ERROR(TSDB_CODE_DND_TOO_MANY_VNODES, "Too many vnode directories")
TAOS_DEFINE_ERROR(TSDB_CODE_DND_EXITING, "Dnode is exiting") TAOS_DEFINE_ERROR(TSDB_CODE_DND_EXITING, "Dnode is exiting")
TAOS_DEFINE_ERROR(TSDB_CODE_DND_PARSE_VNODE_FILE_ERROR, "Parse vnodes.json error") TAOS_DEFINE_ERROR(TSDB_CODE_DND_PARSE_VNODE_FILE_ERROR, "Parse vnodes.json error")
TAOS_DEFINE_ERROR(TSDB_CODE_DND_PARSE_DNODE_FILE_ERROR, "Parse dnodes.json error")
// vnode // vnode
TAOS_DEFINE_ERROR(TSDB_CODE_VND_ACTION_IN_PROGRESS, "Action in progress") TAOS_DEFINE_ERROR(TSDB_CODE_VND_ACTION_IN_PROGRESS, "Action in progress")