From 88d7b970ae08d7bb5039736fde743cbea50e48af Mon Sep 17 00:00:00 2001 From: Shengliang Guan Date: Fri, 5 Nov 2021 15:42:19 +0800 Subject: [PATCH] refact dnode-mnode --- include/common/taosmsg.h | 56 +-- include/server/mnode/mnode.h | 22 +- include/server/vnode/vnode.h | 2 +- include/util/taoserror.h | 20 +- source/dnode/mgmt/inc/dnodeMnode.h | 6 +- source/dnode/mgmt/src/dnodeMnode.c | 609 ++++++++++++++++++------- source/dnode/mgmt/src/dnodeTransport.c | 100 ++-- source/dnode/mgmt/src/dnodeVnodes.c | 2 +- source/dnode/mnode/inc/mnodeWorker.h | 1 - source/dnode/mnode/src/mnodeWorker.c | 33 +- source/dnode/mnode/src/mondeInt.c | 8 +- source/util/src/terror.c | 8 +- 12 files changed, 601 insertions(+), 266 deletions(-) diff --git a/include/common/taosmsg.h b/include/common/taosmsg.h index 2ccab41f8f..86df19d04d 100644 --- a/include/common/taosmsg.h +++ b/include/common/taosmsg.h @@ -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_DROP_FUNCTION, "drop-function" ) 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_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_CONN, "kill-conn" ) 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_COMPACT_VNODE_IN, "compact-vnode" ) 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_CONFIG_DNODE_IN, "config-dnode" ) @@ -698,30 +699,30 @@ typedef struct { typedef struct { uint16_t port; char fqdn[TSDB_FQDN_LEN]; -} SVnodeDesc; +} SReplica; typedef struct { - char db[TSDB_FULL_DB_NAME_LEN]; - uint32_t vgId; - int32_t cacheBlockSize; - int32_t totalBlocks; - int32_t daysPerFile; - int32_t daysToKeep0; - int32_t daysToKeep1; - int32_t daysToKeep2; - int32_t minRowsPerFileBlock; - int32_t maxRowsPerFileBlock; - int32_t fsyncPeriod; - int8_t reserved[16]; - int8_t precision; - int8_t compression; - int8_t cacheLastRow; - int8_t update; - int8_t walLevel; - int8_t replica; - int8_t quorum; - int8_t selfIndex; - SVnodeDesc replicas[TSDB_MAX_REPLICA]; + char db[TSDB_FULL_DB_NAME_LEN]; + uint32_t vgId; + int32_t cacheBlockSize; + int32_t totalBlocks; + int32_t daysPerFile; + int32_t daysToKeep0; + int32_t daysToKeep1; + int32_t daysToKeep2; + int32_t minRowsPerFileBlock; + int32_t maxRowsPerFileBlock; + int32_t fsyncPeriod; + int8_t reserved[16]; + int8_t precision; + int8_t compression; + int8_t cacheLastRow; + int8_t update; + int8_t walLevel; + int8_t replica; + int8_t quorum; + int8_t selfIndex; + SReplica replicas[TSDB_MAX_REPLICA]; } SCreateVnodeMsg, SAlterVnodeMsg; typedef struct { @@ -825,8 +826,11 @@ typedef struct { } SCreateDnodeMsg, SDropDnodeMsg; typedef struct { - int32_t dnodeId; -} SCreateMnodeMsg, SDropMnodeMsg; + int32_t dnodeId; + int8_t replica; + int8_t reserved[3]; + SReplica replicas[TSDB_MAX_REPLICA]; +} SCreateMnodeMsg, SAlterMnodeMsg, SDropMnodeMsg; typedef struct { int32_t dnodeId; diff --git a/include/server/mnode/mnode.h b/include/server/mnode/mnode.h index d9674e2c8c..b4c98ceb85 100644 --- a/include/server/mnode/mnode.h +++ b/include/server/mnode/mnode.h @@ -20,6 +20,15 @@ extern "C" { #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 { int64_t numOfDnode; int64_t numOfMnode; @@ -39,20 +48,25 @@ typedef struct { void (*SendMsgToDnode)(struct SEpSet *epSet, struct SRpcMsg *rpcMsg); void (*SendMsgToMnode)(struct SRpcMsg *rpcMsg); void (*SendRedirectMsg)(struct SRpcMsg *rpcMsg, bool forShell); + int32_t (*PutMsgIntoApplyQueue)(SMnodeMsg *pMsg); } SMnodePara; int32_t mnodeInit(SMnodePara para); void mnodeCleanup(); -int32_t mnodeDeploy(); -void mnodeUnDeploy(); -int32_t mnodeStart(); +int32_t mnodeDeploy(char *path, SMnodeCfg *pCfg); +void mnodeUnDeploy(char *path); +int32_t mnodeStart(char *path, SMnodeCfg *pCfg); +int32_t mnodeAlter(SMnodeCfg *pCfg); void mnodeStop(); int32_t mnodeGetLoad(SMnodeLoad *pLoad); 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 } diff --git a/include/server/vnode/vnode.h b/include/server/vnode/vnode.h index e570cf4261..f7c4b594f7 100644 --- a/include/server/vnode/vnode.h +++ b/include/server/vnode/vnode.h @@ -44,7 +44,7 @@ typedef struct { int8_t replica; int8_t walLevel; int32_t fsyncPeriod; // millisecond - SVnodeDesc replicas[TSDB_MAX_REPLICA]; + SReplica replicas[TSDB_MAX_REPLICA]; } SVnodeCfg; typedef enum { diff --git a/include/util/taoserror.h b/include/util/taoserror.h index e2f9a58023..6c46a4c89f 100644 --- a/include/util/taoserror.h +++ b/include/util/taoserror.h @@ -216,14 +216,18 @@ int32_t* taosGetErrno(); // dnode #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_DNODE_ID_NOT_MATCHED TAOS_DEF_ERROR_CODE(0, 0x0402) //"Dnode Id not matched") -#define TSDB_CODE_DND_MNODE_ALREADY_DROPPED 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_INVALID_MSG_LEN TAOS_DEF_ERROR_CODE(0, 0x0405) //"Invalid message length") -#define TSDB_CODE_DND_ACTION_IN_PROGRESS TAOS_DEF_ERROR_CODE(0, 0x0406) //"Action in progress") -#define TSDB_CODE_DND_TOO_MANY_VNODES TAOS_DEF_ERROR_CODE(0, 0x0407) //"Too many vnode directories") -#define TSDB_CODE_DND_EXITING TAOS_DEF_ERROR_CODE(0, 0x0408) //"Dnode is exiting" -#define TSDB_CODE_DND_PARSE_VNODE_FILE_ERROR TAOS_DEF_ERROR_CODE(0, 0x0409) //"Parse vnodes.json error") +#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_DEPLOYED TAOS_DEF_ERROR_CODE(0, 0x0403) //"Mnode already deployed") +#define TSDB_CODE_DND_MNODE_NOT_DEPLOYED TAOS_DEF_ERROR_CODE(0, 0x0404) //"Mnode not deployed") +#define TSDB_CODE_DND_READ_MNODE_FILE_ERROR TAOS_DEF_ERROR_CODE(0, 0x0405) //"Read mnode.json error") +#define TSDB_CODE_DND_WRITE_MNODE_FILE_ERROR TAOS_DEF_ERROR_CODE(0, 0x0406) //"Write mnode.json error") +#define TSDB_CODE_DND_NO_WRITE_ACCESS TAOS_DEF_ERROR_CODE(0, 0x0407) //"No permission for disk files in dnode") +#define TSDB_CODE_DND_INVALID_MSG_LEN TAOS_DEF_ERROR_CODE(0, 0x0408) //"Invalid message length") +#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 #define TSDB_CODE_VND_ACTION_IN_PROGRESS TAOS_DEF_ERROR_CODE(0, 0x0500) //"Action in progress") diff --git a/source/dnode/mgmt/inc/dnodeMnode.h b/source/dnode/mgmt/inc/dnodeMnode.h index 28702dcb84..9f52e586fe 100644 --- a/source/dnode/mgmt/inc/dnodeMnode.h +++ b/source/dnode/mgmt/inc/dnodeMnode.h @@ -23,9 +23,13 @@ extern "C" { int32_t dnodeInitMnode(); void dnodeCleanupMnode(); -void dnodeProcessMnodeMsg(SRpcMsg *pMsg, SEpSet *pEpSet); 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 } #endif diff --git a/source/dnode/mgmt/src/dnodeMnode.c b/source/dnode/mgmt/src/dnodeMnode.c index be85abdc23..e5a758899e 100644 --- a/source/dnode/mgmt/src/dnodeMnode.c +++ b/source/dnode/mgmt/src/dnodeMnode.c @@ -15,73 +15,120 @@ #define _DEFAULT_SOURCE #include "dnodeMnode.h" +#include "cJSON.h" #include "dnodeDnode.h" #include "dnodeTransport.h" -#include "cJSON.h" #include "mnode.h" +#include "tlockfree.h" +#include "tqueue.h" +#include "tstep.h" +#include "tworker.h" static struct { - int8_t deployed; - int8_t dropped; - char file[PATH_MAX + 20]; - pthread_mutex_t mutex; + int32_t refCount; + int8_t deployed; + int8_t dropped; + SWorkerPool mgmtPool; + 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}; -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 maxLen = 300; char *content = calloc(1, maxLen + 1); cJSON *root = 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) { - dDebug("file %s not exist", tsMnode.file); + dDebug("file %s not exist", file); + code = 0; goto PRASE_MNODE_OVER; } len = (int32_t)fread(content, 1, maxLen, fp); 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; } content[len] = 0; root = cJSON_Parse(content); 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; } cJSON *deployed = cJSON_GetObjectItem(root, "deployed"); 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; } tsMnode.deployed = atoi(deployed->valuestring); cJSON *dropped = cJSON_GetObjectItem(root, "dropped"); 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; } 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: if (content != NULL) free(content); if (root != NULL) cJSON_Delete(root); if (fp != NULL) fclose(fp); - return 0; + return code; } -static int32_t dnodeWriteMnode() { - FILE *fp = fopen(tsMnode.file, "w"); +static int32_t dnodeWriteMnodeFile() { + 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) { - dError("failed to write %s since %s", tsMnode.file, strerror(errno)); - return -1; + dError("failed to write %s since %s", file, strerror(errno)); + return TSDB_CODE_DND_WRITE_MNODE_FILE_ERROR; } int32_t len = 0; @@ -97,119 +144,374 @@ static int32_t dnodeWriteMnode() { taosFsyncFile(fileno(fp)); fclose(fp); 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; } -static int32_t dnodeStartMnode(SCreateMnodeMsg *pCfg) { - int32_t code = 0; - - if (pCfg->dnodeId != dnodeGetDnodeId()) { - code = TSDB_CODE_DND_DNODE_ID_NOT_MATCHED; - dError("failed to start mnode since %s", tstrerror(code)); +static int32_t dnodeStartMnode() { + int32_t code = dnodeAllocMnodeReadQueue(); + if (code != 0) { return code; } - if (tsMnode.dropped) { - code = TSDB_CODE_DND_MNODE_ALREADY_DROPPED; - dError("failed to start mnode since %s", tstrerror(code)); + code = dnodeAllocMnodeWriteQueue(); + if (code != 0) { return code; } - if (tsMnode.deployed) { - dError("failed to start mnode since its already deployed"); - return 0; + code = dnodeAllocMnodeApplyQueue(); + if (code != 0) { + return code; } + code = dnodeAllocMnodeSyncQueue(); + if (code != 0) { + return code; + } + + taosWLockLatch(&tsMnode.latch); tsMnode.deployed = 1; - tsMnode.dropped = 0; + taosWUnLockLatch(&tsMnode.latch); - code = dnodeWriteMnode(); - if (code != 0) { - tsMnode.deployed = 0; - dError("failed to start mnode since %s", tstrerror(code)); - return code; - } - - code = mnodeDeploy(); - 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; + return code; } -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(); - +static void dnodeStopMnode() { + taosWLockLatch(&tsMnode.latch); tsMnode.deployed = 0; + taosWUnLockLatch(&tsMnode.latch); + + 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(); +} + +static int32_t dnodeUnDeployMnode() { tsMnode.dropped = 1; - - code = dnodeWriteMnode(); + int32_t code = dnodeWriteMnodeFile(); if (code != 0) { - tsMnode.deployed = 1; tsMnode.dropped = 0; - dError("failed to drop mnode since %s", tstrerror(code)); + dError("failed to undeploy mnode since %s", tstrerror(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; } -static void dnodeProcessCreateMnodeReq(SRpcMsg *pMsg) { - SCreateMnodeMsg *pCfg = pMsg->pCont; - pCfg->dnodeId = htonl(pCfg->dnodeId); - - int32_t code = dnodeStartMnode(pCfg); - SRpcMsg rspMsg = {.handle = pMsg->handle, .code = code}; - rpcSendResponse(&rspMsg); - rpcFreeCont(pMsg->pCont); +static void dnodeFreeMnodeMgmtQueue() { + tWorkerFreeQueue(&tsMnode.mgmtPool, tsMnode.pMgmtQ); + tsMnode.pMgmtQ = NULL; } -static void dnodeProcessDropMnodeReq(SRpcMsg *pMsg) { - SDropMnodeMsg *pCfg = pMsg->pCont; - pCfg->dnodeId = htonl(pCfg->dnodeId); - - int32_t code = dnodeDropMnode(pCfg); - SRpcMsg rspMsg = {.handle = pMsg->handle, .code = code}; - rpcSendResponse(&rspMsg); - rpcFreeCont(pMsg->pCont); +static int32_t dnodeInitMnodeMgmtWorker() { + SWorkerPool *pPool = &tsMnode.mgmtPool; + pPool->name = "mnode-mgmt"; + pPool->min = 1; + pPool->max = 1; + return tWorkerInit(pPool); } +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() { if (dnodeGetDnodeId() > 0) return false; if (dnodeGetClusterId() > 0) return false; @@ -217,76 +519,77 @@ static bool dnodeNeedDeployMnode() { return true; } -int32_t dnodeInitMnode() { - tsMnode.dropped = 0; - 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); +static int32_t dnodeOpenMnode() { + int32_t code = dnodeReadMnodeFile(); if (code != 0) { - dError("failed to init mnode module since %s", tstrerror(code)); - return code; - } - - code = dnodeReadMnode(); - if (code != 0) { - dError("failed to read file:%s since %s", tsMnode.file, tstrerror(code)); + dError("failed to read open mnode since %s", tstrerror(code)); return code; } if (tsMnode.dropped) { - dError("mnode already dropped, undeploy it"); - mnodeUnDeploy(); - return 0; + dInfo("mnode already dropped, undeploy it"); + return dnodeUnDeployMnode(); } if (!tsMnode.deployed) { bool needDeploy = dnodeNeedDeployMnode(); - if (needDeploy) { - code = mnodeDeploy(); - } else { - return 0; - } + if (!needDeploy) return 0; - if (code != 0) { - dError("failed to deploy mnode since %s", tstrerror(code)); - return code; - } - - tsMnode.deployed = 1; + 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 { + dInfo("start to open mnode"); + return dnodeStartMnode(); } +} - return mnodeStart(); +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) { + dError("dnode-mnode init failed since %s", tstrerror(code)); + } else { + dInfo("dnode-mnode is initialized"); + } } void dnodeCleanupMnode() { - if (tsMnode.deployed) { - mnodeStop(); - } - - mnodeCleanup(); -} - -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); + if (tsMnode.pSteps == NULL) { + dInfo("dnode-mnode start to clean up"); + taosStepCleanup(tsMnode.pSteps); + tsMnode.pSteps = NULL; + dInfo("dnode-mnode is cleaned up"); } } 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; } \ No newline at end of file diff --git a/source/dnode/mgmt/src/dnodeTransport.c b/source/dnode/mgmt/src/dnodeTransport.c index 265efe3070..475470b574 100644 --- a/source/dnode/mgmt/src/dnodeTransport.c +++ b/source/dnode/mgmt/src/dnodeTransport.c @@ -51,76 +51,78 @@ static void dnodeInitMsgFp() { tsTrans.msgFp[TSDB_MSG_TYPE_MQ_RESET] = dnodeProcessVnodeWriteMsg; // msg from client to mnode - tsTrans.msgFp[TSDB_MSG_TYPE_CONNECT] = dnodeProcessMnodeMsg; - tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_ACCT] = dnodeProcessMnodeMsg; - tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_ACCT] = dnodeProcessMnodeMsg; - tsTrans.msgFp[TSDB_MSG_TYPE_DROP_ACCT] = dnodeProcessMnodeMsg; - tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_USER] = dnodeProcessMnodeMsg; - tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_USER] = dnodeProcessMnodeMsg; - tsTrans.msgFp[TSDB_MSG_TYPE_DROP_USER] = dnodeProcessMnodeMsg; - tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_DNODE] = dnodeProcessMnodeMsg; - tsTrans.msgFp[TSDB_MSG_TYPE_CONFIG_DNODE] = dnodeProcessMnodeMsg; - tsTrans.msgFp[TSDB_MSG_TYPE_DROP_DNODE] = dnodeProcessMnodeMsg; - tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_DB] = dnodeProcessMnodeMsg; - tsTrans.msgFp[TSDB_MSG_TYPE_DROP_DB] = dnodeProcessMnodeMsg; - tsTrans.msgFp[TSDB_MSG_TYPE_USE_DB] = dnodeProcessMnodeMsg; - tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_DB] = dnodeProcessMnodeMsg; - tsTrans.msgFp[TSDB_MSG_TYPE_SYNC_DB] = dnodeProcessMnodeMsg; - tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_TOPIC] = dnodeProcessMnodeMsg; - tsTrans.msgFp[TSDB_MSG_TYPE_DROP_TOPIC] = dnodeProcessMnodeMsg; - tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_TOPIC] = dnodeProcessMnodeMsg; - tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_FUNCTION] = dnodeProcessMnodeMsg; - tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_FUNCTION] = dnodeProcessMnodeMsg; - tsTrans.msgFp[TSDB_MSG_TYPE_DROP_FUNCTION] = dnodeProcessMnodeMsg; - tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_STABLE] = dnodeProcessMnodeMsg; - tsTrans.msgFp[TSDB_MSG_TYPE_DROP_STABLE] = dnodeProcessMnodeMsg; - tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_STABLE] = dnodeProcessMnodeMsg; - tsTrans.msgFp[TSDB_MSG_TYPE_STABLE_VGROUP] = dnodeProcessMnodeMsg; - tsTrans.msgFp[TSDB_MSG_TYPE_KILL_QUERY] = dnodeProcessMnodeMsg; - tsTrans.msgFp[TSDB_MSG_TYPE_KILL_CONN] = dnodeProcessMnodeMsg; - tsTrans.msgFp[TSDB_MSG_TYPE_HEARTBEAT] = dnodeProcessMnodeMsg; - tsTrans.msgFp[TSDB_MSG_TYPE_SHOW] = dnodeProcessMnodeMsg; - tsTrans.msgFp[TSDB_MSG_TYPE_SHOW_RETRIEVE] = dnodeProcessMnodeMsg; - tsTrans.msgFp[TSDB_MSG_TYPE_SHOW_RETRIEVE_FUNC] = dnodeProcessMnodeMsg; - tsTrans.msgFp[TSDB_MSG_TYPE_COMPACT_VNODE] = dnodeProcessMnodeMsg; + tsTrans.msgFp[TSDB_MSG_TYPE_CONNECT] = dnodeProcessMnodeReadMsg; + tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_ACCT] = dnodeProcessMnodeWriteMsg; + tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_ACCT] = dnodeProcessMnodeWriteMsg; + tsTrans.msgFp[TSDB_MSG_TYPE_DROP_ACCT] = dnodeProcessMnodeWriteMsg; + tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_USER] = dnodeProcessMnodeWriteMsg; + tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_USER] = dnodeProcessMnodeWriteMsg; + tsTrans.msgFp[TSDB_MSG_TYPE_DROP_USER] = dnodeProcessMnodeWriteMsg; + tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_DNODE] = dnodeProcessMnodeWriteMsg; + tsTrans.msgFp[TSDB_MSG_TYPE_CONFIG_DNODE] = dnodeProcessMnodeWriteMsg; + tsTrans.msgFp[TSDB_MSG_TYPE_DROP_DNODE] = dnodeProcessMnodeWriteMsg; + tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_DB] = dnodeProcessMnodeWriteMsg; + tsTrans.msgFp[TSDB_MSG_TYPE_DROP_DB] = dnodeProcessMnodeWriteMsg; + tsTrans.msgFp[TSDB_MSG_TYPE_USE_DB] = dnodeProcessMnodeWriteMsg; + tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_DB] = dnodeProcessMnodeWriteMsg; + tsTrans.msgFp[TSDB_MSG_TYPE_SYNC_DB] = dnodeProcessMnodeWriteMsg; + tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_TOPIC] = dnodeProcessMnodeWriteMsg; + tsTrans.msgFp[TSDB_MSG_TYPE_DROP_TOPIC] = dnodeProcessMnodeWriteMsg; + tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_TOPIC] = dnodeProcessMnodeWriteMsg; + tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_FUNCTION] = dnodeProcessMnodeWriteMsg; + tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_FUNCTION] = dnodeProcessMnodeWriteMsg; + tsTrans.msgFp[TSDB_MSG_TYPE_DROP_FUNCTION] = dnodeProcessMnodeWriteMsg; + tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_STABLE] = dnodeProcessMnodeWriteMsg; + tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_STABLE] = dnodeProcessMnodeWriteMsg; + tsTrans.msgFp[TSDB_MSG_TYPE_DROP_STABLE] = dnodeProcessMnodeWriteMsg; + tsTrans.msgFp[TSDB_MSG_TYPE_STABLE_VGROUP] = dnodeProcessMnodeReadMsg; + tsTrans.msgFp[TSDB_MSG_TYPE_KILL_QUERY] = dnodeProcessMnodeWriteMsg; + tsTrans.msgFp[TSDB_MSG_TYPE_KILL_CONN] = dnodeProcessMnodeWriteMsg; + tsTrans.msgFp[TSDB_MSG_TYPE_HEARTBEAT] = dnodeProcessMnodeReadMsg; + tsTrans.msgFp[TSDB_MSG_TYPE_SHOW] = dnodeProcessMnodeReadMsg; + tsTrans.msgFp[TSDB_MSG_TYPE_SHOW_RETRIEVE] = dnodeProcessMnodeReadMsg; + tsTrans.msgFp[TSDB_MSG_TYPE_SHOW_RETRIEVE_FUNC] = dnodeProcessMnodeReadMsg; + tsTrans.msgFp[TSDB_MSG_TYPE_COMPACT_VNODE] = dnodeProcessMnodeWriteMsg; // message from client to dnode tsTrans.msgFp[TSDB_MSG_TYPE_NETWORK_TEST] = dnodeProcessDnodeMsg; // message from mnode to vnode 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_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] = dnodeProcessMnodeMsg; + tsTrans.msgFp[TSDB_MSG_TYPE_DROP_STABLE_IN] = dnodeProcessMnodeWriteMsg; // message from mnode to dnode 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_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_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_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_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_RSP] = dnodeProcessMnodeMsg; - tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_MNODE_IN] = dnodeProcessMnodeMsg; - tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_MNODE_IN_RSP] = dnodeProcessMnodeMsg; - tsTrans.msgFp[TSDB_MSG_TYPE_DROP_MNODE_IN] = dnodeProcessMnodeMsg; - tsTrans.msgFp[TSDB_MSG_TYPE_DROP_MNODE_IN_RSP] = dnodeProcessMnodeMsg; + tsTrans.msgFp[TSDB_MSG_TYPE_COMPACT_VNODE_IN_RSP] = dnodeProcessMnodeWriteMsg; + tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_MNODE_IN] = dnodeProcessMnodeMgmtMsg; + tsTrans.msgFp[TSDB_MSG_TYPE_CREATE_MNODE_IN_RSP] = dnodeProcessMnodeWriteMsg; + tsTrans.msgFp[TSDB_MSG_TYPE_ALTER_MNODE_IN] = dnodeProcessMnodeMgmtMsg; + 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_RSP] = dnodeProcessMnodeMsg; + tsTrans.msgFp[TSDB_MSG_TYPE_CONFIG_DNODE_IN_RSP] = dnodeProcessMnodeWriteMsg; // 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_GRANT] = dnodeProcessMnodeMsg; + tsTrans.msgFp[TSDB_MSG_TYPE_GRANT] = dnodeProcessMnodeWriteMsg; 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; } diff --git a/source/dnode/mgmt/src/dnodeVnodes.c b/source/dnode/mgmt/src/dnodeVnodes.c index 8bf80ccff8..c23773f92f 100644 --- a/source/dnode/mgmt/src/dnodeVnodes.c +++ b/source/dnode/mgmt/src/dnodeVnodes.c @@ -983,7 +983,7 @@ static int32_t dnodeInitVnodeModule() { int32_t dnodeInitVnodes() { 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-mgmt", dnodeInitVnodeMgmtWorker, dnodeCleanupVnodeMgmtWorker); taosStepAdd(pSteps, "dnode-vnode-read", dnodeInitVnodeReadWorker, dnodeCleanupVnodeReadWorker); diff --git a/source/dnode/mnode/inc/mnodeWorker.h b/source/dnode/mnode/inc/mnodeWorker.h index 7663df6559..8477af6b72 100644 --- a/source/dnode/mnode/inc/mnodeWorker.h +++ b/source/dnode/mnode/inc/mnodeWorker.h @@ -24,7 +24,6 @@ extern "C" { int32_t mnodeInitWorker(); void mnodeCleanupWorker(); -void mnodeProcessMsg(SRpcMsg *rpcMsg); void mnodeSendRsp(SMnMsg *pMsg, int32_t code); void mnodeReDispatchToWriteQueue(SMnMsg *pMsg); diff --git a/source/dnode/mnode/src/mnodeWorker.c b/source/dnode/mnode/src/mnodeWorker.c index e4b4c47ddf..bdf0e869fc 100644 --- a/source/dnode/mnode/src/mnodeWorker.c +++ b/source/dnode/mnode/src/mnodeWorker.c @@ -39,7 +39,7 @@ static struct { void (*msgFp[TSDB_MSG_TYPE_MAX])(SRpcMsg *pMsg); } tsMworker = {0}; -static SMnMsg *mnodeInitMsg(SRpcMsg *pRpcMsg) { +static SMnMsg *mnodeInitMsg2(SRpcMsg *pRpcMsg) { int32_t size = sizeof(SMnMsg) + pRpcMsg->contLen; SMnMsg *pMsg = taosAllocateQitem(size); @@ -62,7 +62,7 @@ static SMnMsg *mnodeInitMsg(SRpcMsg *pRpcMsg) { return pMsg; } -static void mnodeCleanupMsg(SMnMsg *pMsg) { +static void mnodeCleanupMsg2(SMnMsg *pMsg) { if (pMsg == NULL) return; if (pMsg->rpcMsg.pCont != pMsg->pCont) { tfree(pMsg->rpcMsg.pCont); @@ -75,7 +75,7 @@ static void mnodeDispatchToWriteQueue(SRpcMsg *pRpcMsg) { if (mnodeGetStatus() != MN_STATUS_READY || tsMworker.writeQ == NULL) { mnodeSendRedirectMsg(pRpcMsg, true); } else { - SMnMsg *pMsg = mnodeInitMsg(pRpcMsg); + SMnMsg *pMsg = mnodeInitMsg2(pRpcMsg); if (pMsg == NULL) { SRpcMsg rpcRsp = {.handle = pRpcMsg->handle, .code = TSDB_CODE_MND_INVALID_USER}; rpcSendResponse(&rpcRsp); @@ -91,7 +91,7 @@ static void mnodeDispatchToWriteQueue(SRpcMsg *pRpcMsg) { void mnodeReDispatchToWriteQueue(SMnMsg *pMsg) { if (mnodeGetStatus() != MN_STATUS_READY || tsMworker.writeQ == NULL) { mnodeSendRedirectMsg(&pMsg->rpcMsg, true); - mnodeCleanupMsg(pMsg); + mnodeCleanupMsg2(pMsg); } else { taosWriteQitem(tsMworker.writeQ, pMsg); } @@ -101,7 +101,7 @@ static void mnodeDispatchToReadQueue(SRpcMsg *pRpcMsg) { if (mnodeGetStatus() != MN_STATUS_READY || tsMworker.readQ == NULL) { mnodeSendRedirectMsg(pRpcMsg, true); } else { - SMnMsg *pMsg = mnodeInitMsg(pRpcMsg); + SMnMsg *pMsg = mnodeInitMsg2(pRpcMsg); if (pMsg == NULL) { SRpcMsg rpcRsp = {.handle = pRpcMsg->handle, .code = TSDB_CODE_MND_INVALID_USER}; rpcSendResponse(&rpcRsp); @@ -118,7 +118,7 @@ static void mnodeDispatchToPeerQueue(SRpcMsg *pRpcMsg) { if (mnodeGetStatus() != MN_STATUS_READY || tsMworker.peerReqQ == NULL) { mnodeSendRedirectMsg(pRpcMsg, false); } else { - SMnMsg *pMsg = mnodeInitMsg(pRpcMsg); + SMnMsg *pMsg = mnodeInitMsg2(pRpcMsg); if (pMsg == NULL) { SRpcMsg rpcRsp = {.handle = pRpcMsg->handle, .code = TSDB_CODE_MND_INVALID_USER}; rpcSendResponse(&rpcRsp); @@ -133,7 +133,7 @@ static void mnodeDispatchToPeerQueue(SRpcMsg *pRpcMsg) { } void mnodeDispatchToPeerRspQueue(SRpcMsg *pRpcMsg) { - SMnMsg *pMsg = mnodeInitMsg(pRpcMsg); + SMnMsg *pMsg = mnodeInitMsg2(pRpcMsg); if (pMsg == NULL) { SRpcMsg rpcRsp = {.handle = pRpcMsg->handle, .code = TSDB_CODE_MND_INVALID_USER}; rpcSendResponse(&rpcRsp); @@ -162,7 +162,7 @@ void mnodeSendRsp(SMnMsg *pMsg, int32_t code) { }; rpcSendResponse(&rpcRsp); - mnodeCleanupMsg(pMsg); + mnodeCleanupMsg2(pMsg); } static void mnodeInitMsgFp() { @@ -405,7 +405,7 @@ static void mnodeProcessPeerRsp(SMnMsg *pMsg, void *unused) { if (!mnodeIsMaster()) { 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]) { @@ -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]); } - mnodeCleanupMsg(pMsg); + mnodeCleanupMsg2(pMsg); } int32_t mnodeInitWorker() { @@ -486,10 +486,9 @@ void mnodeCleanupWorker() { mInfo("mnode worker is closed"); } -void mnodeProcessMsg(SRpcMsg *pMsg) { - if (tsMworker.msgFp[pMsg->msgType]) { - (*tsMworker.msgFp[pMsg->msgType])(pMsg); - } else { - assert(0); - } -} +SMnodeMsg *mnodeInitMsg(int32_t msgNum) { return NULL; } + +int32_t mnodeAppendMsg(SMnodeMsg *pMsg, SRpcMsg *pRpcMsg) { return 0; } + +void mnodeCleanupMsg(SMnodeMsg *pMsg) {} +void mnodeProcessMsg(SMnodeMsg *pMsg, EMnMsgType msgType) {} diff --git a/source/dnode/mnode/src/mondeInt.c b/source/dnode/mnode/src/mondeInt.c index 7f1f572432..eed710a1bd 100644 --- a/source/dnode/mnode/src/mondeInt.c +++ b/source/dnode/mnode/src/mondeInt.c @@ -70,6 +70,7 @@ static int32_t mnodeSetPara(SMnodePara para) { if (tsMint.para.SendMsgToDnode == NULL) return -1; if (tsMint.para.SendMsgToMnode == 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.clusterId < 0) return -1; @@ -141,7 +142,7 @@ static bool mnodeNeedDeploy() { return true; } -int32_t mnodeDeploy() { +int32_t mnodeDeploy(char *path, SMnodeCfg *pCfg) { if (tsMint.state != MN_STATUS_UNINIT) { mError("failed to deploy mnode since its deployed"); return 0; @@ -176,7 +177,7 @@ int32_t mnodeDeploy() { return 0; } -void mnodeUnDeploy() { +void mnodeUnDeploy(char *path) { sdbUnDeploy(); 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() {} \ No newline at end of file diff --git a/source/util/src/terror.c b/source/util/src/terror.c index 6838bab403..324f78ad79 100644 --- a/source/util/src/terror.c +++ b/source/util/src/terror.c @@ -228,14 +228,18 @@ TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOPIC_ALREADY_EXIST, "Topic already exists" // dnode 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_DNODE_ID_NOT_MATCHED, "Dnode Id not matched") -TAOS_DEFINE_ERROR(TSDB_CODE_DND_MNODE_ALREADY_DROPPED, "Mnode already deployed") +TAOS_DEFINE_ERROR(TSDB_CODE_DND_MNODE_ID_NOT_MATCH_DNODE, "Mnode Id not match Dnode") +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_INVALID_MSG_LEN, "Invalid message length") 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_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_DNODE_FILE_ERROR, "Parse dnodes.json error") // vnode TAOS_DEFINE_ERROR(TSDB_CODE_VND_ACTION_IN_PROGRESS, "Action in progress")