Merge branch '2.0' into feature/2.0tsdb

src/client/src/tscSQLParser.c

@@ -106,7 +106,7 @@ static int32_t optrToString(tSQLExpr* pExpr, char** exprString);
 static int32_t getMeterIndex(SSQLToken* pTableToken, SQueryInfo* pQueryInfo, SColumnIndex* pIndex);
 static int32_t doFunctionsCompatibleCheck(SSqlCmd* pCmd, SQueryInfo* pQueryInfo);
 static int32_t doLocalQueryProcess(SQueryInfo* pQueryInfo, SQuerySQL* pQuerySql);
-static int32_t tscCheckCreateDbParams(SSqlCmd* pCmd, SCreateDbMsg* pCreate);
+static int32_t tscCheckCreateDbParams(SSqlCmd* pCmd, SCMCreateDbMsg* pCreate);
 
 static SColumnList getColumnList(int32_t num, int16_t tableIndex, int32_t columnIndex);
 
@@ -4689,7 +4689,7 @@ int32_t parseLimitClause(SQueryInfo* pQueryInfo, int32_t clauseIndex, SQuerySQL*
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t setKeepOption(SSqlCmd* pCmd, SCreateDbMsg* pMsg, SCreateDBInfo* pCreateDb) {
+static int32_t setKeepOption(SSqlCmd* pCmd, SCMCreateDbMsg* pMsg, SCreateDBInfo* pCreateDb) {
   const char* msg = "invalid number of options";
 
   pMsg->daysToKeep = htonl(-1);
@@ -4720,7 +4720,7 @@ static int32_t setKeepOption(SSqlCmd* pCmd, SCreateDbMsg* pMsg, SCreateDBInfo* p
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t setTimePrecisionOption(SSqlCmd* pCmd, SCreateDbMsg* pMsg, SCreateDBInfo* pCreateDbInfo) {
+static int32_t setTimePrecisionOption(SSqlCmd* pCmd, SCMCreateDbMsg* pMsg, SCreateDBInfo* pCreateDbInfo) {
   const char* msg = "invalid time precision";
 
   pMsg->precision = TSDB_TIME_PRECISION_MILLI;  // millisecond by default
@@ -4744,7 +4744,7 @@ static int32_t setTimePrecisionOption(SSqlCmd* pCmd, SCreateDbMsg* pMsg, SCreate
   return TSDB_CODE_SUCCESS;
 }
 
-static void setCreateDBOption(SCreateDbMsg* pMsg, SCreateDBInfo* pCreateDb) {
+static void setCreateDBOption(SCMCreateDbMsg* pMsg, SCreateDBInfo* pCreateDb) {
   pMsg->blocksPerTable = htons(pCreateDb->numOfBlocksPerTable);
   pMsg->compression = pCreateDb->compressionLevel;
 
@@ -4759,7 +4759,7 @@ static void setCreateDBOption(SCreateDbMsg* pMsg, SCreateDBInfo* pCreateDb) {
 }
 
 int32_t parseCreateDBOptions(SSqlCmd* pCmd, SCreateDBInfo* pCreateDbSql) {
-  SCreateDbMsg* pMsg = (SCreateDbMsg*)(pCmd->payload);
+  SCMCreateDbMsg* pMsg = (SCMCreateDbMsg*)(pCmd->payload);
   setCreateDBOption(pMsg, pCreateDbSql);
 
   if (setKeepOption(pCmd, pMsg, pCreateDbSql) != TSDB_CODE_SUCCESS) {
@@ -5251,7 +5251,7 @@ int32_t doLocalQueryProcess(SQueryInfo* pQueryInfo, SQuerySQL* pQuerySql) {
 }
 
 // can only perform the parameters based on the macro definitation
-int32_t tscCheckCreateDbParams(SSqlCmd* pCmd, SCreateDbMsg* pCreate) {
+int32_t tscCheckCreateDbParams(SSqlCmd* pCmd, SCMCreateDbMsg* pCreate) {
   char msg[512] = {0};
 
   if (pCreate->commitLog != -1 && (pCreate->commitLog < 0 || pCreate->commitLog > 1)) {

src/client/src/tscSql.c

@@ -67,7 +67,7 @@ TAOS *taos_connect_imp(const char *ip, const char *user, const char *pass, const
 
   if (ip && ip[0]) {
     tscMgmtIpList.inUse = 0;
-    tscMgmtIpList.port = tsMgmtShellPort;
+    tscMgmtIpList.port = tsMnodeShellPort;
     tscMgmtIpList.numOfIps = 1;
     tscMgmtIpList.ip[0] = inet_addr(ip);
 
@@ -82,7 +82,7 @@ TAOS *taos_connect_imp(const char *ip, const char *user, const char *pass, const
     }
   }
 
-  tscMgmtIpList.port = port ? port : tsMgmtShellPort;
+  tscMgmtIpList.port = port ? port : tsMnodeShellPort;
 
   pObj = (STscObj *)malloc(sizeof(STscObj));
   if (NULL == pObj) {
@@ -95,7 +95,7 @@ TAOS *taos_connect_imp(const char *ip, const char *user, const char *pass, const
 
   strncpy(pObj->user, user, TSDB_USER_LEN);
   taosEncryptPass((uint8_t *)pass, strlen(pass), pObj->pass);
-  pObj->mgmtPort = port ? port : tsMgmtShellPort;
+  pObj->mgmtPort = port ? port : tsMnodeShellPort;
 
   if (db) {
     int32_t len = strlen(db);

src/client/src/tscSystem.c

@@ -151,7 +151,7 @@ void taos_init_imp() {
   }
 
   tscMgmtIpList.inUse = 0;
-  tscMgmtIpList.port = tsMgmtShellPort;
+  tscMgmtIpList.port = tsMnodeShellPort;
   tscMgmtIpList.numOfIps = 1;
   tscMgmtIpList.ip[0] = inet_addr(tsMasterIp);
 

src/client/src/tscUtil.c

@@ -769,7 +769,9 @@ int32_t tscMergeTableDataBlocks(SSqlObj* pSql, SDataBlockList* pTableDataBlockLi
 void tscCloseTscObj(STscObj* pObj) {
   pObj->signature = NULL;
   SSqlObj* pSql = pObj->pSql;
+  if (pSql) {
     globalCode = pSql->res.code;
+  }
   
   taosTmrStopA(&(pObj->pTimer));
   tscFreeSqlObj(pSql);

src/dnode/inc/dnodeMgmt.h

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

src/dnode/inc/dnodeModule.h

@@ -20,9 +20,9 @@
 extern "C" {
 #endif
 
-void    dnodeAllocModules();
 int32_t dnodeInitModules();
 void dnodeCleanUpModules();
+void dnodeStartModules();
 
 #ifdef __cplusplus
 }

src/dnode/inc/dnodeRead.h

@@ -22,7 +22,7 @@ extern "C" {
 
 int32_t dnodeInitRead();
 void    dnodeCleanupRead();
-void    dnodeRead(void *pMsg);
+void    dnodeRead(SRpcMsg *pMsg);
 void *  dnodeAllocateReadWorker();
 void    dnodeFreeReadWorker(void *rqueue);
 

src/dnode/inc/dnodeSystem.h

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

src/dnode/inc/dnodeWrite.h

@@ -22,7 +22,7 @@ extern "C" {
 
 int32_t dnodeInitWrite();
 void    dnodeCleanupWrite();
-void    dnodeWrite(void *pMsg);
+void    dnodeWrite(SRpcMsg *pMsg);
 void *  dnodeAllocateWriteWorker();
 void    dnodeFreeWriteWorker(void *worker);
 

src/dnode/src/dnodeMClient.c

@@ -18,7 +18,7 @@
 #include "taosmsg.h"
 #include "tlog.h"
 #include "trpc.h"
-#include "dnodeSystem.h"
+#include "dnode.h"
 
 static void (*dnodeProcessMgmtRspFp[TSDB_MSG_TYPE_MAX])(SRpcMsg *);
 static void   dnodeProcessRspFromMnode(SRpcMsg *pMsg);
@@ -26,7 +26,7 @@ static void   dnodeProcessStatusRsp(SRpcMsg *pMsg);
 static void  *tsDnodeMClientRpc;
 
 int32_t dnodeInitMClient() {
-  dnodeProcessMgmtRspFp[TSDB_MSG_TYPE_STATUS_RSP] = dnodeProcessStatusRsp;
+  dnodeProcessMgmtRspFp[TSDB_MSG_TYPE_DM_STATUS_RSP] = dnodeProcessStatusRsp;
 
   SRpcInit rpcInit;
   memset(&rpcInit, 0, sizeof(rpcInit));
@@ -68,6 +68,4 @@ static void dnodeProcessRspFromMnode(SRpcMsg *pMsg) {
 }
 
 static void dnodeProcessStatusRsp(SRpcMsg *pMsg) {
-
-
 }

src/dnode/src/dnodeService.c → src/dnode/src/dnodeMain.c

@@ -15,41 +15,46 @@
 
 #define _DEFAULT_SOURCE
 #include "os.h"
-#include "tlog.h"
 #include "tglobalcfg.h"
-#include "dnodeSystem.h"
+#include "tlog.h"
+#include "tmodule.h"
+#include "trpc.h"
+#include "tutil.h"
+#include "dnode.h"
+#include "dnodeMClient.h"
+#include "dnodeMgmt.h"
+#include "dnodeMnode.h"
+#include "dnodeModule.h"
+#include "dnodeRead.h"
+#include "dnodeShell.h"
+#include "dnodeWrite.h"
+#ifdef CLUSTER
+#include "account.h"
+#include "admin.h"
+#include "balance.h"
+#include "cluster.h"
+#include "grant.h"
+#include "mpeer.h"
+#include "storage.h"
+#include "vpeer.h"
+#endif
 
+static int32_t dnodeInitSystem();
+static int32_t dnodeInitStorage();
+static void dnodeInitPlugins();
+static void dnodeCleanupStorage();
+static void dnodeCleanUpSystem();
+static void dnodeSetRunStatus(SDnodeRunStatus status);
+static void signal_handler(int32_t signum, siginfo_t *sigInfo, void *context);
+static void dnodeCheckDataDirOpenned(char *dir);
+static SDnodeRunStatus tsDnodeRunStatus = TSDB_DNODE_RUN_STATUS_STOPPED;
 void (*dnodeParseParameterKFp)() = NULL;
 
-/*
- * Termination handler
- */
-void signal_handler(int signum, siginfo_t *sigInfo, void *context) {
-  if (signum == SIGUSR1) {
-    tsCfgDynamicOptions("debugFlag 135");
-    return;
-  }
-  if (signum == SIGUSR2) {
-    tsCfgDynamicOptions("resetlog");
-    return;
-  }
-  syslog(LOG_INFO, "Shut down signal is %d", signum);
-  syslog(LOG_INFO, "Shutting down TDengine service...");
-  // clean the system.
-  dPrint("shut down signal is %d, sender PID:%d", signum, sigInfo->si_pid);
-  dnodeCleanUpSystem();
-  // close the syslog
-  syslog(LOG_INFO, "Shut down TDengine service successfully");
-  dPrint("TDengine is shut down!");
-  closelog();
-  exit(EXIT_SUCCESS);
-}
-
-int main(int argc, char *argv[]) {
+int32_t main(int32_t argc, char *argv[]) {
   dnodeInitPlugins();
 
   // Set global configuration file
-  for (int i = 1; i < argc; ++i) {
+  for (int32_t i = 1; i < argc; ++i) {
     if (strcmp(argv[i], "-c") == 0) {
       if (i < argc - 1) {
         strcpy(configDir, argv[++i]);
@@ -63,10 +68,11 @@ int main(int argc, char *argv[]) {
       printf("gitinfo: %s\n", gitinfo);
       printf("gitinfoI: %s\n", gitinfoOfInternal);
       printf("buildinfo: %s\n", buildinfo);
-      return 0;
+      exit(EXIT_SUCCESS);
     } else if (strcmp(argv[i], "-k") == 0) {
       if (dnodeParseParameterKFp) {
         dnodeParseParameterKFp();
+        exit(EXIT_SUCCESS);
       }
 #ifdef TAOS_MEM_CHECK
     } else if (strcmp(argv[i], "--alloc-random-fail") == 0) {
@@ -94,7 +100,6 @@ int main(int argc, char *argv[]) {
   sigaction(SIGINT, &act, NULL);
   sigaction(SIGUSR1, &act, NULL);
   sigaction(SIGUSR2, &act, NULL);
-  // sigaction(SIGABRT, &act, NULL);
 
   // Open /var/log/syslog file to record information.
   openlog("TDengine:", LOG_PID | LOG_CONS | LOG_NDELAY, LOG_LOCAL1);
@@ -116,4 +121,140 @@ int main(int argc, char *argv[]) {
   }
 }
 
+static void signal_handler(int32_t signum, siginfo_t *sigInfo, void *context) {
+  if (signum == SIGUSR1) {
+    tsCfgDynamicOptions("debugFlag 135");
+    return;
+  }
+  if (signum == SIGUSR2) {
+    tsCfgDynamicOptions("resetlog");
+    return;
+  }
+  syslog(LOG_INFO, "Shut down signal is %d", signum);
+  syslog(LOG_INFO, "Shutting down TDengine service...");
+  // clean the system.
+  dPrint("shut down signal is %d, sender PID:%d", signum, sigInfo->si_pid);
+  dnodeCleanUpSystem();
+  // close the syslog
+  syslog(LOG_INFO, "Shut down TDengine service successfully");
+  dPrint("TDengine is shut down!");
+  closelog();
+  exit(EXIT_SUCCESS);
+}
+
+static int32_t dnodeInitSystem() {
+  dnodeSetRunStatus(TSDB_DNODE_RUN_STATUS_INITIALIZE);
+  tscEmbedded  = 1;
+  taosResolveCRC();
+  tsReadGlobalLogConfig();
+  taosSetCoreDump();
+  signal(SIGPIPE, SIG_IGN);
+
+  struct stat dirstat;
+  if (stat(logDir, &dirstat) < 0) {
+    mkdir(logDir, 0755);
+  }
+
+  char temp[TSDB_FILENAME_LEN];
+  sprintf(temp, "%s/taosdlog", logDir);
+  if (taosInitLog(temp, tsNumOfLogLines, 1) < 0) {
+    printf("failed to init log file\n");
+  }
+
+  if (!tsReadGlobalConfig()) {
+    tsPrintGlobalConfig();
+    dError("TDengine read global config failed");
+    return -1;
+  }
+  tsPrintGlobalConfig();
+
+  dPrint("Server IP address is:%s", tsPrivateIp);
+  dPrint("starting to initialize TDengine ...");
+
+  if (dnodeInitStorage() != 0) return -1;
+  if (dnodeInitModules() != 0) return -1;
+  if (dnodeInitRead() != 0) return -1;
+  if (dnodeInitWrite() != 0) return -1;
+  if (dnodeInitMgmt() != 0) return -1;
+  if (dnodeInitMnode() != 0) return -1;
+  if (dnodeInitMClient() != 0) return -1;
+  if (dnodeInitShell() != 0) return -1;
+
+  dnodeStartModules();
+  dnodeSetRunStatus(TSDB_DNODE_RUN_STATUS_RUNING);
+
+  dPrint("TDengine is initialized successfully");
+
+  return 0;
+}
+
+static void dnodeCleanUpSystem() {
+  if (dnodeGetRunStatus() != TSDB_DNODE_RUN_STATUS_STOPPED) {
+    tclearModuleStatus(TSDB_MOD_MGMT);
+    dnodeSetRunStatus(TSDB_DNODE_RUN_STATUS_STOPPED);
+    dnodeCleanupShell();
+    dnodeCleanupMClient();
+    dnodeCleanupMnode();
+    dnodeCleanupMgmt();
+    dnodeCleanupWrite();
+    dnodeCleanupRead();
+    dnodeCleanUpModules();
+    dnodeCleanupStorage();
+    taosCloseLogger();
+  }
+}
+
+SDnodeRunStatus dnodeGetRunStatus() {
+  return tsDnodeRunStatus;
+}
+
+static void dnodeSetRunStatus(SDnodeRunStatus status) {
+  tsDnodeRunStatus = status;
+}
+
+static void dnodeCheckDataDirOpenned(char *dir) {
+  char filepath[256] = {0};
+  sprintf(filepath, "%s/.running", dir);
+
+  int32_t fd  = open(filepath, O_WRONLY | O_CREAT | O_TRUNC, S_IRWXU | S_IRWXG | S_IRWXO);
+  int32_t ret = flock(fd, LOCK_EX | LOCK_NB);
+  if (ret != 0) {
+    dError("failed to lock file:%s ret:%d, database may be running, quit", filepath, ret);
+    exit(0);
+  }
+}
 
+static int32_t dnodeInitStorage() {
+  struct stat dirstat;
+  strcpy(tsDirectory, dataDir);
+  if (stat(dataDir, &dirstat) < 0) {
+    mkdir(dataDir, 0755);
+  }
+
+  char fileName[128];
+  sprintf(fileName, "%s/tsdb", tsDirectory);
+  mkdir(fileName, 0755);
+  sprintf(fileName, "%s/data", tsDirectory);
+  mkdir(fileName, 0755);
+  sprintf(tsMgmtDirectory, "%s/mgmt", tsDirectory);
+  sprintf(tsDirectory, "%s/tsdb", dataDir);
+
+  dnodeCheckDataDirOpenned(dataDir);
+
+  dPrint("storage directory is initialized");
+  return 0;
+}
+
+static void dnodeCleanupStorage() {}
+
+static void dnodeInitPlugins() {
+#ifdef CLUSTER
+//  acctInit();
+//  adminInit();
+//  balanceInit();
+//  clusterInit();
+//  grantInit();
+//  mpeerInit();
+//  storageInit();
+#endif
+}

src/dnode/src/dnodeMgmt.c

@@ -22,6 +22,7 @@
 #include "trpc.h"
 #include "tstatus.h"
 #include "tsdb.h"
+#include "ttimer.h"
 #include "dnodeMgmt.h"
 #include "dnodeRead.h"
 #include "dnodeWrite.h"
@@ -46,33 +47,52 @@ static int32_t  dnodeOpenVnode(int32_t vgId);
 static void     dnodeCleanupVnode(SVnodeObj *pVnode);
 static int32_t  dnodeCreateVnode(SMDCreateVnodeMsg *cfg);
 static void     dnodeDropVnode(SVnodeObj *pVnode);
-static void     dnodeProcesSMDCreateVnodeMsg(SRpcMsg *pMsg);
-static void     dnodeProcesSMDDropVnodeMsg(SRpcMsg *pMsg);
+static void     dnodeProcessCreateVnodeMsg(SRpcMsg *pMsg);
+static void     dnodeProcessDropVnodeMsg(SRpcMsg *pMsg);
 static void     dnodeProcessAlterVnodeMsg(SRpcMsg *pMsg);
+static void     dnodeProcessAlterStreamMsg(SRpcMsg *pMsg);
+static void     dnodeProcessConfigDnodeMsg(SRpcMsg *pMsg);
 static void   (*dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MAX])(SRpcMsg *pMsg);
+static void     dnodeSendStatusMsg(void *handle, void *tmrId);
 
 static void * tsDnodeVnodesHash = NULL;
+static void  *tsDnodeTmr = NULL;
+static void  *tsStatusTimer = NULL;
 
 int32_t dnodeInitMgmt() {
-  dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_CREATE_VNODE] = dnodeProcesSMDCreateVnodeMsg;
-  dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_DROP_VNODE]   = dnodeProcesSMDDropVnodeMsg;
+  dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_CREATE_VNODE] = dnodeProcessCreateVnodeMsg;
+  dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_DROP_VNODE]   = dnodeProcessDropVnodeMsg;
   dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_ALTER_VNODE]  = dnodeProcessAlterVnodeMsg;
+  dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_ALTER_STREAM]  = dnodeProcessAlterStreamMsg;
+  dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_CONFIG_DNODE]  = dnodeProcessConfigDnodeMsg;
 
   tsDnodeVnodesHash = taosInitIntHash(TSDB_MAX_VNODES, sizeof(SVnodeObj), taosHashInt);
   if (tsDnodeVnodesHash == NULL) {
-    return TSDB_CODE_SERV_OUT_OF_MEMORY;
+    dError("failed to init vnode list");
+    return -1;
   }
 
+  tsDnodeTmr = taosTmrInit(100, 200, 60000, "DND-DM");
+  if (tsDnodeTmr == NULL) {
+    dError("failed to init dnode timer");
+    return -1;
+  }
+  taosTmrReset(dnodeSendStatusMsg, 500, NULL, tsDnodeTmr, &tsStatusTimer);
+
   return dnodeOpenVnodes();
 }
 
 void dnodeCleanupMgmt() {
+  if (tsStatusTimer != NULL) {
+    taosTmrStopA(&tsStatusTimer);
+    tsStatusTimer = NULL;
+  }
+
   dnodeCleanupVnodes();
   taosCleanUpIntHash(tsDnodeVnodesHash);
 }
 
-void dnodeMgmt(void *rpcMsg) {
-  SRpcMsg *pMsg = rpcMsg;
+void dnodeMgmt(SRpcMsg *pMsg) {
   terrno = 0;
 
   if (dnodeProcessMgmtMsgFp[pMsg->msgType]) {
@@ -90,7 +110,7 @@ void dnodeMgmt(void *rpcMsg) {
 }
 
 void *dnodeGetVnode(int32_t vgId) {
-  SVnodeObj *pVnode = taosGetIntHashData(tsDnodeVnodesHash, vgId);
+  SVnodeObj *pVnode = (SVnodeObj *) taosGetIntHashData(tsDnodeVnodesHash, vgId);
   if (pVnode == NULL) {
     terrno = TSDB_CODE_INVALID_VGROUP_ID;
     return NULL;
@@ -160,7 +180,7 @@ static int32_t dnodeOpenVnode(int32_t vgId) {
   vnodeObj.events   = NULL;
   vnodeObj.cq       = NULL;
 
-  taosAddIntHash(tsDnodeVnodesHash, vnodeObj.vgId, &vnodeObj);
+  taosAddIntHash(tsDnodeVnodesHash, vnodeObj.vgId, (char *) (&vnodeObj));
 
   return TSDB_CODE_SUCCESS;
 }
@@ -225,7 +245,7 @@ static int32_t dnodeCreateVnode(SMDCreateVnodeMsg *pVnodeCfg) {
   vnodeObj.events   = NULL;
   vnodeObj.cq       = NULL;
 
-  taosAddIntHash(tsDnodeVnodesHash, vnodeObj.vgId, &vnodeObj);
+  taosAddIntHash(tsDnodeVnodesHash, vnodeObj.vgId, (char *) (&vnodeObj));
 
   return TSDB_CODE_SUCCESS;
 }
@@ -246,16 +266,16 @@ static void dnodeDropVnode(SVnodeObj *pVnode) {
   dnodeCleanupVnode(pVnode);
 }
 
-static void dnodeProcesSMDCreateVnodeMsg(SRpcMsg *rpcMsg) {
+static void dnodeProcessCreateVnodeMsg(SRpcMsg *rpcMsg) {
   SRpcMsg rpcRsp = {.handle = rpcMsg->handle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0};
 
-  SMDCreateVnodeMsg *pCreate = (SMDCreateVnodeMsg *) rpcMsg->pCont;
+  SMDCreateVnodeMsg *pCreate = rpcMsg->pCont;
   pCreate->vnode           = htonl(pCreate->vnode);
   pCreate->cfg.vgId        = htonl(pCreate->cfg.vgId);
   pCreate->cfg.maxSessions = htonl(pCreate->cfg.maxSessions);
   pCreate->cfg.daysPerFile = htonl(pCreate->cfg.daysPerFile);
 
-  SVnodeObj *pVnodeObj = taosGetIntHashData(tsDnodeVnodesHash, pCreate->cfg.vgId);
+  SVnodeObj *pVnodeObj = (SVnodeObj *) taosGetIntHashData(tsDnodeVnodesHash, pCreate->cfg.vgId);
   if (pVnodeObj != NULL) {
     rpcRsp.code = TSDB_CODE_SUCCESS;
   } else {
@@ -266,13 +286,13 @@ static void dnodeProcesSMDCreateVnodeMsg(SRpcMsg *rpcMsg) {
   rpcFreeCont(rpcMsg->pCont);
 }
 
-static void dnodeProcesSMDDropVnodeMsg(SRpcMsg *rpcMsg) {
+static void dnodeProcessDropVnodeMsg(SRpcMsg *rpcMsg) {
   SRpcMsg rpcRsp = {.handle = rpcMsg->handle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0};
 
-  SMDDropVnodeMsg *pDrop = (SMDCreateVnodeMsg *) rpcMsg->pCont;
+  SMDDropVnodeMsg *pDrop = rpcMsg->pCont;
   pDrop->vgId = htonl(pDrop->vgId);
 
-  SVnodeObj *pVnodeObj = taosGetIntHashData(tsDnodeVnodesHash, pDrop->vgId);
+  SVnodeObj *pVnodeObj = (SVnodeObj *) taosGetIntHashData(tsDnodeVnodesHash, pDrop->vgId);
   if (pVnodeObj != NULL) {
     dnodeDropVnode(pVnodeObj);
     rpcRsp.code = TSDB_CODE_SUCCESS;
@@ -287,13 +307,13 @@ static void dnodeProcesSMDDropVnodeMsg(SRpcMsg *rpcMsg) {
 static void dnodeProcessAlterVnodeMsg(SRpcMsg *rpcMsg) {
   SRpcMsg rpcRsp = {.handle = rpcMsg->handle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0};
 
-  SMDCreateVnodeMsg *pCreate = (SMDCreateVnodeMsg *) rpcMsg->pCont;
+  SMDCreateVnodeMsg *pCreate = rpcMsg->pCont;
   pCreate->vnode           = htonl(pCreate->vnode);
   pCreate->cfg.vgId        = htonl(pCreate->cfg.vgId);
   pCreate->cfg.maxSessions = htonl(pCreate->cfg.maxSessions);
   pCreate->cfg.daysPerFile = htonl(pCreate->cfg.daysPerFile);
 
-  SVnodeObj *pVnodeObj = taosGetIntHashData(tsDnodeVnodesHash, pCreate->cfg.vgId);
+  SVnodeObj *pVnodeObj = (SVnodeObj *) taosGetIntHashData(tsDnodeVnodesHash, pCreate->cfg.vgId);
   if (pVnodeObj != NULL) {
     rpcRsp.code = TSDB_CODE_SUCCESS;
   } else {
@@ -303,3 +323,80 @@ static void dnodeProcessAlterVnodeMsg(SRpcMsg *rpcMsg) {
   rpcSendResponse(&rpcRsp);
   rpcFreeCont(rpcMsg->pCont);
 }
+
+static void dnodeProcessAlterStreamMsg(SRpcMsg *pMsg) {
+//  SMDAlterStreamMsg *pStream = pCont;
+//  pStream->uid    = htobe64(pStream->uid);
+//  pStream->stime  = htobe64(pStream->stime);
+//  pStream->vnode  = htonl(pStream->vnode);
+//  pStream->sid    = htonl(pStream->sid);
+//  pStream->status = htonl(pStream->status);
+//
+//  int32_t code = dnodeCreateStream(pStream);
+}
+
+static void dnodeProcessConfigDnodeMsg(SRpcMsg *pMsg) {
+//  SCfgDnodeMsg *pCfg = (SCfgDnodeMsg *)pCont;
+//
+//  int32_t code = tsCfgDynamicOptions(pCfg->config);
+//  dnodeSendRspToMnode(pConn, msgType + 1, code, NULL, 0);
+}
+
+
+static void dnodeSendStatusMsg(void *handle, void *tmrId) {
+  taosTmrReset(dnodeSendStatusMsg, tsStatusInterval * 1000, NULL, tsDnodeTmr, &tsStatusTimer);
+  if (tsStatusTimer == NULL) {
+    dError("failed to start status timer");
+    return;
+  }
+
+//  int32_t contLen = sizeof(SDMStatusMsg) + dnodeGetVnodesNum() * sizeof(SVnodeLoad);
+//  SDMStatusMsg *pStatus = rpcMallocCont(contLen);
+//  if (pStatus == NULL) {
+//    dError("Failed to malloc status message");
+//    return;
+//  }
+//
+//  int32_t totalVnodes = dnodeGetVnodesNum();
+//
+//  pStatus->version          = htonl(tsVersion);
+//  pStatus->privateIp        = htonl(inet_addr(tsPrivateIp));
+//  pStatus->publicIp         = htonl(inet_addr(tsPublicIp));
+//  pStatus->lastReboot       = htonl(tsRebootTime);
+//  pStatus->numOfTotalVnodes = htons((uint16_t) tsNumOfTotalVnodes);
+//  pStatus->openVnodes       = htons((uint16_t) totalVnodes);
+//  pStatus->numOfCores       = htons((uint16_t) tsNumOfCores);
+//  pStatus->diskAvailable    = tsAvailDataDirGB;
+//  pStatus->alternativeRole  = (uint8_t) tsAlternativeRole;
+//
+//  SVnodeLoad *pLoad = (SVnodeLoad *)pStatus->load;
+
+  //TODO loop all vnodes
+  //  for (int32_t vnode = 0, count = 0; vnode <= totalVnodes; ++vnode) {
+  //    if (vnodeList[vnode].cfg.maxSessions <= 0) continue;
+  //
+  //    SVnodeObj *pVnode = vnodeList + vnode;
+  //    pLoad->vnode = htonl(vnode);
+  //    pLoad->vgId = htonl(pVnode->cfg.vgId);
+  //    pLoad->status = (uint8_t)vnodeList[vnode].vnodeStatus;
+  //    pLoad->syncStatus =(uint8_t)vnodeList[vnode].syncStatus;
+  //    pLoad->accessState = (uint8_t)(pVnode->accessState);
+  //    pLoad->totalStorage = htobe64(pVnode->vnodeStatistic.totalStorage);
+  //    pLoad->compStorage = htobe64(pVnode->vnodeStatistic.compStorage);
+  //    if (pVnode->vnodeStatus == TSDB_VN_STATUS_MASTER) {
+  //      pLoad->pointsWritten = htobe64(pVnode->vnodeStatistic.pointsWritten);
+  //    } else {
+  //      pLoad->pointsWritten = htobe64(0);
+  //    }
+  //    pLoad++;
+  //
+  //    if (++count >= tsOpenVnodes) {
+  //      break;
+  //    }
+  //  }
+
+//  dnodeSendMsgToMnode(TSDB_MSG_TYPE_STATUS, pStatus, contLen);
+}
+
+
+

src/dnode/src/dnodeMnode.c

@@ -17,7 +17,7 @@
 #include "taosmsg.h"
 #include "tlog.h"
 #include "trpc.h"
-#include "dnodeSystem.h"
+#include "dnode.h"
 #include "dnodeMgmt.h"
 #include "dnodeWrite.h"
 
@@ -28,14 +28,18 @@ static void  *tsDnodeMnodeRpc = NULL;
 int32_t dnodeInitMnode() {
   dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_CREATE_TABLE] = dnodeWrite;
   dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_DROP_TABLE]   = dnodeWrite;
+  dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_ALTER_TABLE]  = dnodeWrite;
+  dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_DROP_STABLE]  = dnodeWrite;
+  dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_CREATE_VNODE] = dnodeMgmt;
   dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_DROP_VNODE]   = dnodeMgmt;
+  dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_ALTER_VNODE]  = dnodeMgmt;
+  dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_ALTER_STREAM] = dnodeMgmt;
+  dnodeProcessMgmtMsgFp[TSDB_MSG_TYPE_MD_CONFIG_DNODE] = dnodeMgmt;
 
   SRpcInit rpcInit;
   memset(&rpcInit, 0, sizeof(rpcInit));
   rpcInit.localIp      = tsAnyIp ? "0.0.0.0" : tsPrivateIp;
-
-  // note: a new port shall be assigned
-  // rpcInit.localPort    = tsDnodeMnodePort;   
+  rpcInit.localPort    = tsDnodeMnodePort;
   rpcInit.label        = "DND-mgmt";
   rpcInit.numOfThreads = 1;
   rpcInit.cfp          = dnodeProcessMsgFromMnode;

src/dnode/src/dnodeModule.c

@@ -22,9 +22,9 @@
 #include "http.h"
 #include "monitor.h"
 #include "dnodeModule.h"
-#include "dnodeSystem.h"
+#include "dnode.h"
 
-void dnodeAllocModules() {
+static void dnodeAllocModules() {
   tsModule[TSDB_MOD_MGMT].name          = "mgmt";
   tsModule[TSDB_MOD_MGMT].initFp        = mgmtInitSystem;
   tsModule[TSDB_MOD_MGMT].cleanUpFp     = mgmtCleanUpSystem;
@@ -69,6 +69,8 @@ void dnodeCleanUpModules() {
 }
 
 int32_t dnodeInitModules() {
+  dnodeAllocModules();
+
   for (int mod = 0; mod < TSDB_MOD_MAX; ++mod) {
     if (tsModule[mod].num != 0 && tsModule[mod].initFp) {
       if ((*tsModule[mod].initFp)() != 0) {
@@ -81,7 +83,7 @@ int32_t dnodeInitModules() {
   return TSDB_CODE_SUCCESS;
 }
 
-void dnodeStartModulesImp() {
+void dnodeStartModules() {
   for (int mod = 1; mod < TSDB_MOD_MAX; ++mod) {
     if (tsModule[mod].num != 0 && tsModule[mod].startFp) {
       if ((*tsModule[mod].startFp)() != 0) {
@@ -90,5 +92,3 @@ void dnodeStartModulesImp() {
     }
   }
 }
-
-void (*dnodeStartModules)() = dnodeStartModulesImp;

src/dnode/src/dnodeRead.c

@@ -67,9 +67,7 @@ void dnodeCleanupRead() {
   taosCloseQset(readQset);
 }
 
-void dnodeRead(void *rpcMsg) {
-  SRpcMsg *pMsg = rpcMsg;
-
+void dnodeRead(SRpcMsg *pMsg) {
   int32_t     leftLen      = pMsg->contLen;
   char        *pCont       = (char *) pMsg->pCont;
   int32_t     contLen      = 0;
@@ -205,9 +203,43 @@ static void dnodeProcessReadResult(SReadMsg *pRead) {
 }
 
 static void dnodeProcessQueryMsg(SReadMsg *pMsg) {
-
+  void *pQInfo = (void*)100;
+  dTrace("query msg is disposed, qInfo:%p", pQInfo);
+
+  SQueryTableRsp *pRsp = (SQueryTableRsp *) rpcMallocCont(sizeof(SQueryTableRsp));
+  pRsp->code    = 0;
+  pRsp->qhandle = htobe64((uint64_t) (pQInfo));
+
+  SRpcMsg rpcRsp = {
+      .handle = pMsg->rpcMsg.handle,
+      .pCont = pRsp,
+      .contLen = sizeof(SQueryTableRsp),
+      .code = 0,
+      .msgType = 0
+  };
+  rpcSendResponse(&rpcRsp);
 }
 
 static void dnodeProcessRetrieveMsg(SReadMsg *pMsg) {
-
+  SRetrieveTableMsg *pRetrieve = pMsg->pCont;
+  void *pQInfo = htobe64(pRetrieve->qhandle);
+
+  dTrace("retrieve msg is disposed, qInfo:%p", pQInfo);
+
+  assert(pQInfo != NULL);
+  int32_t contLen = 100;
+  SRetrieveTableRsp *pRsp = (SRetrieveTableRsp *) rpcMallocCont(contLen);
+  pRsp->numOfRows = 0;
+  pRsp->precision = 0;
+  pRsp->offset    = 0;
+  pRsp->useconds  = 0;
+
+  SRpcMsg rpcRsp = {
+      .handle = pMsg->rpcMsg.handle,
+      .pCont = pRsp,
+      .contLen = contLen,
+      .code = 0,
+      .msgType = 0
+  };
+  rpcSendResponse(&rpcRsp);
 }

src/dnode/src/dnodeShell.c

@@ -20,14 +20,17 @@
 #include "taosmsg.h"
 #include "tlog.h"
 #include "trpc.h"
-#include "dnodeSystem.h"
+#include "dnode.h"
 #include "dnodeRead.h"
 #include "dnodeWrite.h"
 #include "dnodeShell.h"
 
 static void (*dnodeProcessShellMsgFp[TSDB_MSG_TYPE_MAX])(SRpcMsg *);
 static void   dnodeProcessMsgFromShell(SRpcMsg *pMsg);
+static int    dnodeRetrieveUserAuthInfo(char *user, char *spi, char *encrypt, char *secret, char *ckey);
 static void  *tsDnodeShellRpc = NULL;
+static int32_t tsDnodeQueryReqNum  = 0;
+static int32_t tsDnodeSubmitReqNum = 0;
 
 int32_t dnodeInitShell() {
   dnodeProcessShellMsgFp[TSDB_MSG_TYPE_SUBMIT]   = dnodeWrite;
@@ -43,13 +46,14 @@ int32_t dnodeInitShell() {
   SRpcInit rpcInit;
   memset(&rpcInit, 0, sizeof(rpcInit));
   rpcInit.localIp      = tsAnyIp ? "0.0.0.0" : tsPrivateIp;
-  rpcInit.localPort    = tsVnodeShellPort;
+  rpcInit.localPort    = tsDnodeShellPort;
   rpcInit.label        = "DND-shell";
   rpcInit.numOfThreads = numOfThreads;
   rpcInit.cfp          = dnodeProcessMsgFromShell;
   rpcInit.sessions     = TSDB_SESSIONS_PER_DNODE;
   rpcInit.connType     = TAOS_CONN_SERVER;
   rpcInit.idleTime     = tsShellActivityTimer * 1500;
+  rpcInit.afp          =
 
   tsDnodeShellRpc = rpcOpen(&rpcInit);
   if (tsDnodeShellRpc == NULL) {
@@ -64,12 +68,12 @@ int32_t dnodeInitShell() {
 void dnodeCleanupShell() {
   if (tsDnodeShellRpc) {
     rpcClose(tsDnodeShellRpc);
+    tsDnodeShellRpc = NULL;
   }
 }
 
 void dnodeProcessMsgFromShell(SRpcMsg *pMsg) {
   SRpcMsg rpcMsg;
-
   rpcMsg.handle = pMsg->handle;
   rpcMsg.pCont = NULL;
   rpcMsg.contLen = 0;
@@ -82,6 +86,12 @@ void dnodeProcessMsgFromShell(SRpcMsg *pMsg) {
     return;
   }
 
+  if (pMsg->msgType == TSDB_MSG_TYPE_QUERY) {
+    atomic_fetch_add_32(&tsDnodeQueryReqNum, 1);
+  } else if (pMsg->msgType == TSDB_MSG_TYPE_SUBMIT) {
+    atomic_fetch_add_32(&tsDnodeSubmitReqNum, 1);
+  } else {}
+
   if ( dnodeProcessShellMsgFp[pMsg->msgType] ) {
     (*dnodeProcessShellMsgFp[pMsg->msgType])(pMsg);
   } else {
@@ -92,4 +102,17 @@ void dnodeProcessMsgFromShell(SRpcMsg *pMsg) {
   }
 }
 
+static int dnodeRetrieveUserAuthInfo(char *user, char *spi, char *encrypt, char *secret, char *ckey) {
+  return TSDB_CODE_SUCCESS;
+}
+
+SDnodeStatisInfo dnodeGetStatisInfo() {
+  SDnodeStatisInfo info = {0};
+  if (dnodeGetRunStatus() == TSDB_DNODE_RUN_STATUS_RUNING) {
+    //info.httpReqNum   = httpGetReqCount();
+    info.queryReqNum  = atomic_exchange_32(&tsDnodeQueryReqNum, 0);
+    info.submitReqNum = atomic_exchange_32(&tsDnodeSubmitReqNum, 0);
+  }
 
+  return info;
+}

src/dnode/src/dnodeSystem.c

-/*
- * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
- *
- * This program is free software: you can use, redistribute, and/or modify
- * it under the terms of the GNU Affero General Public License, version 3
- * or later ("AGPL"), as published by the Free Software Foundation.
- *
- * This program is distributed in the hope that it will be useful, but WITHOUT
- * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
- * FITNESS FOR A PARTICULAR PURPOSE.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program. If not, see <http://www.gnu.org/licenses/>.
- */
-
-#define _DEFAULT_SOURCE
-#include "os.h"
-#include "taosdef.h"
-#include "taoserror.h"
-#include "tcrc32c.h"
-#include "tlog.h"
-#include "tmodule.h"
-#include "tsched.h"
-#include "ttime.h"
-#include "ttimer.h"
-#include "tutil.h"
-#include "http.h"
-#include "trpc.h"
-#include "dnode.h"
-#include "dnodeMgmt.h"
-#include "dnodeModule.h"
-#include "dnodeShell.h"
-#include "dnodeSystem.h"
-
-#ifdef CLUSTER
-#include "account.h"
-#include "admin.h"
-#include "balance.h"
-#include "cluster.h"
-#include "grant.h"
-#include "mpeer.h"
-#include "storage.h"
-#include "vpeer.h"
-#endif
-
-static pthread_mutex_t tsDnodeMutex;
-static SDnodeRunStatus tsDnodeRunStatus = TSDB_DNODE_RUN_STATUS_STOPPED;
-
-static int32_t dnodeInitRpcQHandle();
-static int32_t dnodeInitQueryQHandle();
-static int32_t dnodeInitTmrCtl();
-
-
-int32_t (*dnodeInitStorage)() = NULL;
-void    (*dnodeCleanupStorage)() = NULL;
-int32_t (*dnodeInitPeers)(int32_t numOfThreads) = NULL;
-
-void     *tsDnodeTmr;
-void     **tsRpcQhandle;
-void     *tsDnodeMgmtQhandle;
-void     *tsQueryQhandle;
-int32_t  tsVnodePeers   = TSDB_VNODES_SUPPORT - 1;
-int32_t  tsMaxQueues;
-uint32_t tsRebootTime;
-
-static void dnodeInitVnodesLock() {
-  pthread_mutex_init(&tsDnodeMutex, NULL);
-}
-
-void dnodeLockVnodes() {
-  pthread_mutex_lock(&tsDnodeMutex);
-}
-
-void dnodeUnLockVnodes() {
-  pthread_mutex_unlock(&tsDnodeMutex);
-}
-
-static void dnodeCleanVnodesLock() {
-  pthread_mutex_destroy(&tsDnodeMutex);
-}
-
-SDnodeRunStatus dnodeGetRunStatus() {
-  return tsDnodeRunStatus;
-}
-
-void dnodeSetRunStatus(SDnodeRunStatus status) {
-  tsDnodeRunStatus = status;
-}
-
-void dnodeCleanUpSystem() {
-  tclearModuleStatus(TSDB_MOD_MGMT);
-
-  if (dnodeGetRunStatus() == TSDB_DNODE_RUN_STATUS_STOPPED) {
-    return;
-  } else {
-    dnodeSetRunStatus(TSDB_DNODE_RUN_STATUS_STOPPED);
-  }
-
-  dnodeCleanupShell();
-  dnodeCleanUpModules();
-  dnodeCleanupMgmt();
-  taosCloseLogger();
-  dnodeCleanupStorage();
-  dnodeCleanVnodesLock();
-}
-
-void dnodeCheckDataDirOpenned(const char *dir) {
-  char filepath[256] = {0};
-  sprintf(filepath, "%s/.running", dir);
-  int32_t fd  = open(filepath, O_WRONLY | O_CREAT | O_TRUNC, S_IRWXU | S_IRWXG | S_IRWXO);
-  int32_t ret = flock(fd, LOCK_EX | LOCK_NB);
-  if (ret != 0) {
-    dError("failed to lock file:%s ret:%d, database may be running, quit", filepath, ret);
-    exit(0);
-  }
-}
-
-void dnodeInitPlugins() {
-#ifdef CLUSTER
-//  acctInit();
-//  adminInit();
-//  balanceInit();
-//  clusterInit();
-//  grantInit();
-//  mpeerInit();
-//  storageInit();
-#endif
-}
-
-int32_t dnodeInitSystem() {
-  tsRebootTime = taosGetTimestampSec();
-  tscEmbedded  = 1;
-
-  dnodeSetRunStatus(TSDB_DNODE_RUN_STATUS_INITIALIZE);
-  taosResolveCRC();
-
-  // Read global configuration.
-  tsReadGlobalLogConfig();
-
-  struct stat dirstat;
-  if (stat(logDir, &dirstat) < 0) {
-    mkdir(logDir, 0755);
-  }
-
-  char temp[128];
-  sprintf(temp, "%s/taosdlog", logDir);
-  if (taosInitLog(temp, tsNumOfLogLines, 1) < 0) {
-    printf("failed to init log file\n");
-  }
-
-  if (!tsReadGlobalConfig()) {
-    tsPrintGlobalConfig();
-    dError("TDengine read global config failed");
-    return -1;
-  }
-
-  if (dnodeInitStorage() != 0) {
-    dError("TDengine init tier directory failed");
-    return -1;
-  }
-
-//  dnodeInitMgmtIp();
-
-  tsPrintGlobalConfig();
-
-  dPrint("Server IP address is:%s", tsPrivateIp);
-
-  taosSetCoreDump();
-
-  signal(SIGPIPE, SIG_IGN);
-
-  dnodeAllocModules();
-
-  dnodeInitVnodesLock();
-
-  dPrint("starting to initialize TDengine ...");
-
-  if (dnodeInitRpcQHandle() < 0) {
-    dError("failed to init query qhandle, exit");
-    return -1;
-  }
-
-  if (dnodeCheckSystem() < 0) {
-    return -1;
-  }
-
-  if (dnodeInitModules() < 0) {
-    return -1;
-  }
-
-  if (dnodeInitTmrCtl() < 0) {
-    dError("failed to init timer, exit");
-    return -1;
-  }
-
-  if (dnodeInitQueryQHandle() < 0) {
-    dError("failed to init query qhandle, exit");
-    return -1;
-  }
-
-  if (dnodeInitMgmt() < 0) {
-    dError("failed to init vnode storage");
-    return -1;
-  }
-
-  int32_t numOfThreads = (1.0 - tsRatioOfQueryThreads) * tsNumOfCores * tsNumOfThreadsPerCore / 2.0;
-  if (numOfThreads < 1) numOfThreads = 1;
-  if (dnodeInitPeers(numOfThreads) < 0) {
-    dError("failed to init vnode peer communication");
-    return -1;
-  }
-
-  if (dnodeInitMgmt() < 0) {
-    dError("failed to init communication to mgmt");
-    return -1;
-  }
-
-  if (dnodeInitShell() < 0) {
-    dError("failed to init communication to shell");
-    return -1;
-  }
-
-  dnodeStartModules();
-
-  dnodeSetRunStatus(TSDB_DNODE_RUN_STATUS_RUNING);
-
-  dPrint("TDengine is initialized successfully");
-
-  return 0;
-}
-
-int32_t dnodeInitStorageImp() {
-  struct stat dirstat;
-  strcpy(tsDirectory, dataDir);
-  if (stat(dataDir, &dirstat) < 0) {
-    mkdir(dataDir, 0755);
-  }
-
-  char fileName[128];
-
-  sprintf(fileName, "%s/tsdb", tsDirectory);
-  mkdir(fileName, 0755);
-
-  sprintf(fileName, "%s/data", tsDirectory);
-  mkdir(fileName, 0755);
-
-  sprintf(tsMgmtDirectory, "%s/mgmt", tsDirectory);
-  sprintf(tsDirectory, "%s/tsdb", dataDir);
-  dnodeCheckDataDirOpenned(dataDir);
-
-  return 0;
-}
-
-static int32_t dnodeInitQueryQHandle() {
-  int32_t numOfThreads = tsRatioOfQueryThreads * tsNumOfCores * tsNumOfThreadsPerCore;
-  if (numOfThreads < 1) {
-    numOfThreads = 1;
-  }
-
-  int32_t maxQueueSize = tsNumOfVnodesPerCore * tsNumOfCores * tsSessionsPerVnode;
-  dTrace("query task queue initialized, max slot:%d, task threads:%d", maxQueueSize, numOfThreads);
-
-  tsQueryQhandle = taosInitSchedulerWithInfo(maxQueueSize, numOfThreads, "query", tsDnodeTmr);
-
-  return 0;
-}
-
-static int32_t dnodeInitTmrCtl() {
-  tsDnodeTmr = taosTmrInit(TSDB_MAX_VNODES * (tsVnodePeers + 10) + tsSessionsPerVnode + 1000, 200, 60000,
-                             "DND-vnode");
-  if (tsDnodeTmr == NULL) {
-    dError("failed to init timer, exit");
-    return -1;
-  }
-
-  return 0;
-}
-
-static int32_t dnodeInitRpcQHandle() {
-  tsMaxQueues = (1.0 - tsRatioOfQueryThreads) * tsNumOfCores * tsNumOfThreadsPerCore / 2.0;
-  if (tsMaxQueues < 1) {
-    tsMaxQueues = 1;
-  }
-
-  tsRpcQhandle = malloc(tsMaxQueues * sizeof(void *));
-
-  for (int32_t i = 0; i < tsMaxQueues; ++i) {
-    tsRpcQhandle[i] = taosInitScheduler(tsSessionsPerVnode, 1, "dnode");
-  }
-
-  tsDnodeMgmtQhandle = taosInitScheduler(tsSessionsPerVnode, 1, "mgmt");
-
-  return 0;
-}
-
-int32_t dnodeCheckSystemImp() {
-  return 0;
-}
-
-int32_t (*dnodeCheckSystem)() = dnodeCheckSystemImp;
-
-int32_t dnodeInitPeersImp(int32_t numOfThreads) {
-  return 0;
-}

src/dnode/src/dnodeWrite.c

@@ -56,6 +56,8 @@ static void   dnodeProcessWriteResult(SWriteMsg *pWrite);
 static void   dnodeProcessSubmitMsg(SWriteMsg *pMsg);
 static void   dnodeProcessCreateTableMsg(SWriteMsg *pMsg);
 static void   dnodeProcessDropTableMsg(SWriteMsg *pMsg);
+static void   dnodeProcessAlterTableMsg(SWriteMsg *pMsg);
+static void   dnodeProcessDropStableMsg(SWriteMsg *pMsg);
 
 SWriteWorkerPool wWorkerPool;
 
@@ -63,6 +65,8 @@ int32_t dnodeInitWrite() {
   dnodeProcessWriteMsgFp[TSDB_MSG_TYPE_SUBMIT]          = dnodeProcessSubmitMsg;
   dnodeProcessWriteMsgFp[TSDB_MSG_TYPE_MD_CREATE_TABLE] = dnodeProcessCreateTableMsg;
   dnodeProcessWriteMsgFp[TSDB_MSG_TYPE_MD_DROP_TABLE]   = dnodeProcessDropTableMsg;
+  dnodeProcessWriteMsgFp[TSDB_MSG_TYPE_MD_ALTER_TABLE]  = dnodeProcessAlterTableMsg;
+  dnodeProcessWriteMsgFp[TSDB_MSG_TYPE_MD_DROP_STABLE]  = dnodeProcessDropStableMsg;
 
   wWorkerPool.max = tsNumOfCores;
   wWorkerPool.writeWorker = (SWriteWorker *)calloc(sizeof(SWriteWorker), wWorkerPool.max);
@@ -79,9 +83,7 @@ void dnodeCleanupWrite() {
   free(wWorkerPool.writeWorker);
 }
 
-void dnodeWrite(void *rpcMsg) {
-  SRpcMsg *pMsg = rpcMsg;
-
+void dnodeWrite(SRpcMsg *pMsg) {
   int32_t     leftLen      = pMsg->contLen;
   char        *pCont       = (char *) pMsg->pCont;
   int32_t     contLen      = 0;
@@ -245,13 +247,98 @@ static void dnodeHandleIdleWorker(SWriteWorker *pWorker) {
 }
 
 static void dnodeProcessSubmitMsg(SWriteMsg *pMsg) {
-
+  dTrace("submit msg is disposed");
+
+  SShellSubmitRspMsg *pRsp = rpcMallocCont(sizeof(SShellSubmitRspMsg));
+  pRsp->code              = 0;
+  pRsp->numOfRows         = htonl(1);
+  pRsp->affectedRows      = htonl(1);
+  pRsp->numOfFailedBlocks = 0;
+
+  SRpcMsg rpcRsp = {
+    .handle = pMsg->rpcMsg.handle,
+    .pCont = pRsp,
+    .contLen = sizeof(SShellSubmitRspMsg),
+    .code = 0,
+    .msgType = 0
+  };
+  rpcSendResponse(&rpcRsp);
 }
 
 static void dnodeProcessCreateTableMsg(SWriteMsg *pMsg) {
+  SMDCreateTableMsg *pTable = pMsg->rpcMsg.pCont;
+  if (pTable->tableType == TSDB_TABLE_TYPE_CHILD_TABLE) {
+    dTrace("table:%s, start to create child table, stable:%s", pTable->tableId, pTable->superTableId);
+  } else if (pTable->tableType == TSDB_TABLE_TYPE_NORMAL_TABLE){
+    dTrace("table:%s, start to create normal table", pTable->tableId);
+  } else if (pTable->tableType == TSDB_TABLE_TYPE_STREAM_TABLE){
+    dTrace("table:%s, start to create stream table", pTable->tableId);
+  } else {
+    dError("table:%s, invalid table type:%d", pTable->tableType);
+  }
 
+//  pTable->numOfColumns  = htons(pTable->numOfColumns);
+//  pTable->numOfTags     = htons(pTable->numOfTags);
+//  pTable->sid           = htonl(pTable->sid);
+//  pTable->sversion      = htonl(pTable->sversion);
+//  pTable->tagDataLen    = htonl(pTable->tagDataLen);
+//  pTable->sqlDataLen    = htonl(pTable->sqlDataLen);
+//  pTable->contLen       = htonl(pTable->contLen);
+//  pTable->numOfVPeers   = htonl(pTable->numOfVPeers);
+//  pTable->uid           = htobe64(pTable->uid);
+//  pTable->superTableUid = htobe64(pTable->superTableUid);
+//  pTable->createdTime   = htobe64(pTable->createdTime);
+//
+//  for (int i = 0; i < pTable->numOfVPeers; ++i) {
+//    pTable->vpeerDesc[i].ip    = htonl(pTable->vpeerDesc[i].ip);
+//    pTable->vpeerDesc[i].vnode = htonl(pTable->vpeerDesc[i].vnode);
+//  }
+//
+//  int32_t totalCols = pTable->numOfColumns + pTable->numOfTags;
+//  SSchema *pSchema = (SSchema *) pTable->data;
+//  for (int32_t col = 0; col < totalCols; ++col) {
+//    pSchema->bytes = htons(pSchema->bytes);
+//    pSchema->colId = htons(pSchema->colId);
+//    pSchema++;
+//  }
+//
+//  int32_t code = dnodeCreateTable(pTable);
+
+  SRpcMsg rpcRsp = {.handle = pMsg->rpcMsg.handle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0};
+  rpcSendResponse(&rpcRsp);
 }
 
 static void dnodeProcessDropTableMsg(SWriteMsg *pMsg) {
+  SMDDropTableMsg *pTable = pMsg->rpcMsg.pCont;
+  dPrint("table:%s, sid:%d is dropped", pTable->tableId, pTable->sid);
+
+//  pTable->sid         = htonl(pTable->sid);
+//  pTable->numOfVPeers = htonl(pTable->numOfVPeers);
+//  pTable->uid         = htobe64(pTable->uid);
+//
+//  for (int i = 0; i < pTable->numOfVPeers; ++i) {
+//    pTable->vpeerDesc[i].ip    = htonl(pTable->vpeerDesc[i].ip);
+//    pTable->vpeerDesc[i].vnode = htonl(pTable->vpeerDesc[i].vnode);
+//  }
+//
+//  int32_t code = dnodeDropTable(pTable);
+//
+  SRpcMsg rpcRsp = {.handle = pMsg->rpcMsg.handle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0};
+  rpcSendResponse(&rpcRsp);
+}
+
+static void dnodeProcessAlterTableMsg(SWriteMsg *pMsg) {
+  SMDCreateTableMsg *pTable = pMsg->rpcMsg.pCont;
+  dPrint("table:%s, sid:%d is alterd", pTable->tableId, pTable->sid);
+
+  SRpcMsg rpcRsp = {.handle = pMsg->rpcMsg.handle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0};
+  rpcSendResponse(&rpcRsp);
+}
+
+static void dnodeProcessDropStableMsg(SWriteMsg *pMsg) {
+  SMDDropSTableMsg *pTable = pMsg->rpcMsg.pCont;
+  dPrint("stable:%s, is dropped", pTable->tableId);
 
+  SRpcMsg rpcRsp = {.handle = pMsg->rpcMsg.handle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0};
+  rpcSendResponse(&rpcRsp);
 }

src/inc/dnode.h

@@ -21,7 +21,7 @@ extern "C" {
 #endif
 
 #include <stdint.h>
-#include <pthread.h>
+#include <stdbool.h>
 
 typedef struct {
   int32_t queryReqNum;
@@ -29,33 +29,13 @@ typedef struct {
   int32_t httpReqNum;
 } SDnodeStatisInfo;
 
-typedef struct {
-  char  id[20];
-  char  sid;
-  void *thandle;
-  int   mgmtIndex;
-  char  status;  // 0:offline, 1:online
-} SMgmtObj;
-
-// global variables
-extern uint32_t tsRebootTime;
-
-// dnodeCluster
-extern void (*dnodeStartModules)();
-extern int32_t (*dnodeCheckSystem)();
-
-
-// dnodeSystem
-extern void *tsDnodeMgmtQhandle;
-void dnodeCheckDataDirOpenned(const char* dir);
-
-// dnodeModule
-extern void (*dnodeStartModules)();
-
-
+typedef enum {
+  TSDB_DNODE_RUN_STATUS_INITIALIZE,
+  TSDB_DNODE_RUN_STATUS_RUNING,
+  TSDB_DNODE_RUN_STATUS_STOPPED
+} SDnodeRunStatus;
 
-void dnodeLockVnodes();
-void dnodeUnLockVnodes();
+SDnodeRunStatus dnodeGetRunStatus();
 SDnodeStatisInfo dnodeGetStatisInfo();
 
 #ifdef __cplusplus

src/inc/mnode.h

@@ -25,8 +25,6 @@ extern "C" {
 #include "taosdef.h"
 #include "taosmsg.h"
 #include "taoserror.h"
-
-
 #include "sdb.h"
 #include "tglobalcfg.h"
 #include "thash.h"
@@ -44,7 +42,6 @@ extern "C" {
 // internal globals
 extern char  version[];
 extern void *tsMgmtTmr;
-extern void *tsMgmtTranQhandle;
 extern char  tsMgmtDirectory[];
 
 typedef struct {

src/inc/taosmsg.h

@@ -38,7 +38,7 @@ extern "C" {
 #define TSDB_MSG_TYPE_RETRIEVE               7
 #define TSDB_MSG_TYPE_RETRIEVE_RSP           8
 
-// message from mgmt to dnode
+// message from mnode to dnode
 #define TSDB_MSG_TYPE_MD_CREATE_TABLE        9
 #define TSDB_MSG_TYPE_MD_CREATE_TABLE_RSP    10
 #define TSDB_MSG_TYPE_MD_DROP_TABLE          11
@@ -58,84 +58,76 @@ extern "C" {
 #define TSDB_MSG_TYPE_MD_CONFIG_DNODE        25
 #define TSDB_MSG_TYPE_MD_CONFIG_DNODE_RSP    26
 
-
+// message from client to mnode
+#define TSDB_MSG_TYPE_CM_CONNECT             31
+#define TSDB_MSG_TYPE_CM_CONNECT_RSP         32
+#define TSDB_MSG_TYPE_CM_CREATE_ACCT         33
+#define TSDB_MSG_TYPE_CM_CREATE_ACCT_RSP     34
+#define TSDB_MSG_TYPE_CM_ALTER_ACCT          35
+#define TSDB_MSG_TYPE_CM_ALTER_ACCT_RSP      36
+#define TSDB_MSG_TYPE_CM_DROP_ACCT           37
+#define TSDB_MSG_TYPE_CM_DROP_ACCT_RSP       38
+#define TSDB_MSG_TYPE_CM_CREATE_USER         39
+#define TSDB_MSG_TYPE_CM_CREATE_USER_RSP     40
+#define TSDB_MSG_TYPE_CM_ALTER_USER          41
+#define TSDB_MSG_TYPE_CM_ALTER_USER_RSP      42
+#define TSDB_MSG_TYPE_CM_DROP_USER           43
+#define TSDB_MSG_TYPE_CM_DROP_USER_RSP       44
+#define TSDB_MSG_TYPE_CM_CREATE_DNODE        45
+#define TSDB_MSG_TYPE_CM_CREATE_DNODE_RSP    46
+#define TSDB_MSG_TYPE_CM_DROP_DNODE          47
+#define TSDB_MSG_TYPE_CM_DROP_DNODE_RSP      48
 #define TSDB_MSG_TYPE_CM_CONFIG_DNODE        TSDB_MSG_TYPE_MD_CONFIG_DNODE
 #define TSDB_MSG_TYPE_CM_CONFIG_DNODE_RSP    TSDB_MSG_TYPE_MD_CONFIG_DNODE_RSP
-
-#define TSDB_MSG_TYPE_DM_CONFIG_VNODE           19
-#define TSDB_MSG_TYPE_DM_CONFIG_VNODE_RSP       20
-
-
-#define TSDB_MSG_TYPE_SDB_SYNC               21
-#define TSDB_MSG_TYPE_SDB_SYNC_RSP           22
-#define TSDB_MSG_TYPE_SDB_FORWARD            23
-#define TSDB_MSG_TYPE_SDB_FORWARD_RSP        24
-#define TSDB_MSG_TYPE_CONNECT                31
-#define TSDB_MSG_TYPE_CONNECT_RSP            32
-#define TSDB_MSG_TYPE_CREATE_ACCT            33
-#define TSDB_MSG_TYPE_CREATE_ACCT_RSP        34
-#define TSDB_MSG_TYPE_ALTER_ACCT             35
-#define TSDB_MSG_TYPE_ALTER_ACCT_RSP         36
-#define TSDB_MSG_TYPE_DROP_ACCT              37
-#define TSDB_MSG_TYPE_DROP_ACCT_RSP          38
-#define TSDB_MSG_TYPE_CREATE_USER            39
-#define TSDB_MSG_TYPE_CREATE_USER_RSP        40
-#define TSDB_MSG_TYPE_ALTER_USER             41
-#define TSDB_MSG_TYPE_ALTER_USER_RSP         42
-#define TSDB_MSG_TYPE_DROP_USER              43
-#define TSDB_MSG_TYPE_DROP_USER_RSP          44
-#define TSDB_MSG_TYPE_CREATE_MNODE           45
-#define TSDB_MSG_TYPE_CREATE_MNODE_RSP       46
-#define TSDB_MSG_TYPE_DROP_MNODE             47
-#define TSDB_MSG_TYPE_DROP_MNODE_RSP         48
-#define TSDB_MSG_TYPE_CREATE_DNODE           49
-#define TSDB_MSG_TYPE_CREATE_DNODE_RSP       50
-#define TSDB_MSG_TYPE_DROP_DNODE             51
-#define TSDB_MSG_TYPE_DROP_DNODE_RSP         52
-#define TSDB_MSG_TYPE_ALTER_DNODE            53
-#define TSDB_MSG_TYPE_ALTER_DNODE_RSP        54
-#define TSDB_MSG_TYPE_CREATE_DB              55
-#define TSDB_MSG_TYPE_CREATE_DB_RSP          56
-#define TSDB_MSG_TYPE_DROP_DB                57
-#define TSDB_MSG_TYPE_DROP_DB_RSP            58
-#define TSDB_MSG_TYPE_USE_DB                 59
-#define TSDB_MSG_TYPE_USE_DB_RSP             60
-#define TSDB_MSG_TYPE_ALTER_DB               61
-#define TSDB_MSG_TYPE_ALTER_DB_RSP           62
-#define TSDB_MSG_TYPE_CREATE_TABLE           63
-#define TSDB_MSG_TYPE_CREATE_TABLE_RSP       64
-#define TSDB_MSG_TYPE_DROP_TABLE             65
-#define TSDB_MSG_TYPE_DROP_TABLE_RSP         66
-#define TSDB_MSG_TYPE_ALTER_TABLE            67
-#define TSDB_MSG_TYPE_ALTER_TABLE_RSP        68
-
-#define TSDB_MSG_TYPE_TABLE_CFG              71
-#define TSDB_MSG_TYPE_TABLE_CFG_RSP          72
-#define TSDB_MSG_TYPE_TABLE_META             73
-#define TSDB_MSG_TYPE_TABLE_META_RSP         74
-#define TSDB_MSG_TYPE_STABLE_META            75
-#define TSDB_MSG_TYPE_STABLE_META_RSP        76
-#define TSDB_MSG_TYPE_MULTI_TABLE_META       77
-#define TSDB_MSG_TYPE_MULTI_TABLE_META_RSP   78
-#define TSDB_MSG_TYPE_ALTER_STREAM           79
-#define TSDB_MSG_TYPE_ALTER_STREAM_RSP       80
-#define TSDB_MSG_TYPE_SHOW                   81
-#define TSDB_MSG_TYPE_SHOW_RSP               82
-#define TSDB_MSG_TYPE_CFG_MNODE              83
-#define TSDB_MSG_TYPE_CFG_MNODE_RSP          84
-#define TSDB_MSG_TYPE_KILL_QUERY             85
-#define TSDB_MSG_TYPE_KILL_QUERY_RSP         86
-#define TSDB_MSG_TYPE_KILL_STREAM            87
-#define TSDB_MSG_TYPE_KILL_STREAM_RSP        88
-#define TSDB_MSG_TYPE_KILL_CONNECTION        89
-#define TSDB_MSG_TYPE_KILL_CONNECTION_RSP    90
-#define TSDB_MSG_TYPE_HEARTBEAT              91
-#define TSDB_MSG_TYPE_HEARTBEAT_RSP          92
-#define TSDB_MSG_TYPE_STATUS                 93
-#define TSDB_MSG_TYPE_STATUS_RSP             94
-#define TSDB_MSG_TYPE_GRANT                  95
-#define TSDB_MSG_TYPE_GRANT_RSP              96
-#define TSDB_MSG_TYPE_MAX                    97
+#define TSDB_MSG_TYPE_CM_CREATE_DB           49
+#define TSDB_MSG_TYPE_CM_CREATE_DB_RSP       50
+#define TSDB_MSG_TYPE_CM_DROP_DB             51
+#define TSDB_MSG_TYPE_CM_DROP_DB_RSP         52
+#define TSDB_MSG_TYPE_CM_USE_DB              53
+#define TSDB_MSG_TYPE_CM_USE_DB_RSP          54
+#define TSDB_MSG_TYPE_CM_ALTER_DB            55
+#define TSDB_MSG_TYPE_CM_ALTER_DB_RSP        56
+#define TSDB_MSG_TYPE_CM_CREATE_TABLE        57
+#define TSDB_MSG_TYPE_CM_CREATE_TABLE_RSP    58
+#define TSDB_MSG_TYPE_CM_DROP_TABLE          59
+#define TSDB_MSG_TYPE_CM_DROP_TABLE_RSP      60
+#define TSDB_MSG_TYPE_CM_ALTER_TABLE         61
+#define TSDB_MSG_TYPE_CM_ALTER_TABLE_RSP     62
+#define TSDB_MSG_TYPE_CM_TABLE_META          63
+#define TSDB_MSG_TYPE_CM_TABLE_META_RSP      64
+#define TSDB_MSG_TYPE_CM_STABLE_META         65
+#define TSDB_MSG_TYPE_CM_STABLE_META_RSP     66
+#define TSDB_MSG_TYPE_CM_TABLES_META         67
+#define TSDB_MSG_TYPE_CM_TABLES_META_RSP     68
+#define TSDB_MSG_TYPE_CM_ALTER_STREAM        69
+#define TSDB_MSG_TYPE_CM_ALTER_STREAM_RSP    70
+#define TSDB_MSG_TYPE_CM_SHOW                71
+#define TSDB_MSG_TYPE_CM_SHOW_RSP            72
+#define TSDB_MSG_TYPE_CM_KILL_QUERY          73
+#define TSDB_MSG_TYPE_CM_KILL_QUERY_RSP      74
+#define TSDB_MSG_TYPE_CM_KILL_STREAM         75
+#define TSDB_MSG_TYPE_CM_KILL_STREAM_RSP     76
+#define TSDB_MSG_TYPE_CM_KILL_CONN           77
+#define TSDB_MSG_TYPE_CM_KILL_CONN_RSP       78
+#define TSDB_MSG_TYPE_CM_HEARTBEAT           79
+#define TSDB_MSG_TYPE_CM_HEARTBEAT_RSP       80
+
+// message from dnode to mnode
+#define TSDB_MSG_TYPE_DM_CONFIG_TABLE        91
+#define TSDB_MSG_TYPE_DM_CONFIG_TABLE_RSP    92
+#define TSDB_MSG_TYPE_DM_CONFIG_VNODE        93
+#define TSDB_MSG_TYPE_DM_CONFIG_VNODE_RSP    94
+#define TSDB_MSG_TYPE_DM_STATUS              95
+#define TSDB_MSG_TYPE_DM_STATUS_RSP          96
+#define TSDB_MSG_TYPE_DM_GRANT               97
+#define TSDB_MSG_TYPE_DM_GRANT_RSP           98
+
+#define TSDB_MSG_TYPE_SDB_SYNC               101
+#define TSDB_MSG_TYPE_SDB_SYNC_RSP           102
+#define TSDB_MSG_TYPE_SDB_FORWARD            103
+#define TSDB_MSG_TYPE_SDB_FORWARD_RSP        104
+
+#define TSDB_MSG_TYPE_MAX                    105
 
 // IE type
 #define TSDB_IE_TYPE_SEC               1
@@ -248,7 +240,7 @@ typedef struct SSchema {
 typedef struct {
   int32_t  vnode;  //the index of vnode
   uint32_t ip;
-} SVPeerDesc;
+} SVnodeDesc;
 
 typedef struct {
   int8_t     tableType;
@@ -263,11 +255,11 @@ typedef struct {
   uint64_t   uid;
   uint64_t   superTableUid;
   uint64_t   createdTime;
-  SVPeerDesc vpeerDesc[TSDB_MAX_MPEERS];
+  SVnodeDesc vpeerDesc[TSDB_MAX_MPEERS];
   char       tableId[TSDB_TABLE_ID_LEN + 1];
   char       superTableId[TSDB_TABLE_ID_LEN + 1];
   char       data[];
-} SDMCreateTableMsg;
+} SMDCreateTableMsg;
 
 typedef struct {
   char      tableId[TSDB_TABLE_ID_LEN + 1];
@@ -278,12 +270,12 @@ typedef struct {
   int16_t   sqlLen;  // the length of SQL, it starts after schema , sql is a null-terminated string
   int16_t   reserved[16];
   char      schema[];
-} SCreateTableMsg;
+} SCMCreateTableMsg;
 
 typedef struct {
   char   tableId[TSDB_TABLE_ID_LEN + 1];
   int8_t igNotExists;
-} SDropTableMsg;
+} SCMDropTableMsg;
 
 typedef struct {
   char    tableId[TSDB_TABLE_ID_LEN + 1];
@@ -292,13 +284,13 @@ typedef struct {
   char    tagVal[TSDB_MAX_BYTES_PER_ROW];
   int8_t  numOfCols; /* number of schema */
   SSchema schema[];
-} SAlterTableMsg;
+} SCMAlterTableMsg;
 
 typedef struct {
   char clientVersion[TSDB_VERSION_LEN];
   char msgVersion[TSDB_VERSION_LEN];
   char db[TSDB_TABLE_ID_LEN + 1];
-} SConnectMsg;
+} SCMConnectMsg;
 
 typedef struct {
   char      acctId[TSDB_ACCT_LEN + 1];
@@ -306,7 +298,7 @@ typedef struct {
   int8_t    writeAuth;
   int8_t    superAuth;
   SRpcIpSet ipList;
-} SConnectRsp;
+} SCMConnectRsp;
 
 typedef struct {
   int32_t maxUsers;
@@ -326,18 +318,18 @@ typedef struct {
   char     user[TSDB_USER_LEN + 1];
   char     pass[TSDB_KEY_LEN + 1];
   SAcctCfg cfg;
-} SCreateAcctMsg, SAlterAcctMsg;
+} SCMCreateAcctMsg, SCMAlterAcctMsg;
 
 typedef struct {
   char user[TSDB_USER_LEN + 1];
-} SDropUserMsg, SDropAcctMsg;
+} SCMDropUserMsg, SCMDropAcctMsg;
 
 typedef struct {
   char   user[TSDB_USER_LEN + 1];
   char   pass[TSDB_KEY_LEN + 1];
   int8_t privilege;
   int8_t flag;
-} SCreateUserMsg, SAlterUserMsg;
+} SCMCreateUserMsg, SCMAlterUserMsg;
 
 typedef struct {
   char db[TSDB_TABLE_ID_LEN + 1];
@@ -347,14 +339,14 @@ typedef struct {
   int32_t    sid;
   int32_t    numOfVPeers;
   uint64_t   uid;
-  SVPeerDesc vpeerDesc[TSDB_MAX_MPEERS];
+  SVnodeDesc vpeerDesc[TSDB_MAX_MPEERS];
   char       tableId[TSDB_TABLE_ID_LEN + 1];
 } SMDDropTableMsg;
 
 typedef struct {
   char    tableId[TSDB_TABLE_ID_LEN + 1];
   int64_t uid;
-} SDRemoveSuperTableMsg;
+} SMDDropSTableMsg;
 
 typedef struct {
   int32_t vgId;
@@ -570,12 +562,12 @@ typedef struct {
   int8_t  loadLatest;  // load into mem or not
   uint8_t precision;   // time resolution
   int8_t  reserved[16];
-} SVnodeCfg, SCreateDbMsg, SDbCfg, SAlterDbMsg;
+} SVnodeCfg, SDbCfg, SCMCreateDbMsg, SCMAlterDbMsg;
 
 typedef struct {
   char    db[TSDB_TABLE_ID_LEN + 1];
   uint8_t ignoreNotExists;
-} SDropDbMsg, SUseDbMsg;
+} SCMDropDbMsg, SCMUseDbMsg;
 
 // IMPORTANT: sizeof(SVnodeStatisticInfo) should not exceed
 // TSDB_FILE_HEADER_LEN/4 - TSDB_FILE_HEADER_VERSION_SIZE
@@ -606,52 +598,40 @@ typedef struct {
   uint8_t    alternativeRole;
   uint8_t    reserve[15];
   SVnodeLoad load[];
-} SStatusMsg;
+} SDMStatusMsg;
 
 typedef struct {
   int32_t      code;
   SDnodeState  dnodeState;
   SRpcIpSet    ipList;
   SVnodeAccess vnodeAccess[];
-} SStatusRsp;
-
-typedef struct {
-  char spi;
-  char encrypt;
-  char secret[TSDB_KEY_LEN];  // key is changed if updated
-  char cipheringKey[TSDB_KEY_LEN];
-} SSecIe;
-
-typedef struct {
-  int32_t numOfVPeers;
-  SVPeerDesc vpeerDesc[];
-} SVpeerDescArray;
+} SDMStatusRsp;
 
 typedef struct {
   int32_t    vnode;
   SVnodeCfg  cfg;
-  SVPeerDesc vpeerDesc[TSDB_MAX_MPEERS];
+  SVnodeDesc vpeerDesc[TSDB_MAX_MPEERS];
 } SMDCreateVnodeMsg;
 
 typedef struct {
   char    tableId[TSDB_TABLE_ID_LEN + 1];
   int16_t createFlag;
   char    tags[];
-} STableInfoMsg;
+} SCMTableInfoMsg;
 
 typedef struct {
   int32_t numOfTables;
   char    tableIds[];
-} SMultiTableInfoMsg;
+} SCMMultiTableInfoMsg;
 
 typedef struct {
   char    tableId[TSDB_TABLE_ID_LEN + 1];
-} SSuperTableInfoMsg;
+} SCMSuperTableInfoMsg;
 
 typedef struct {
   int32_t  numOfDnodes;
   uint32_t dnodeIps[];
-} SSuperTableInfoRsp;
+} SCMSuperTableInfoRsp;
 
 typedef struct {
   int16_t elemLen;
@@ -683,7 +663,7 @@ typedef struct {
 } SSuperTableMetaMsg;
 
 typedef struct {
-  SVPeerDesc vpeerDesc[TSDB_VNODES_SUPPORT];
+  SVnodeDesc vpeerDesc[TSDB_VNODES_SUPPORT];
   int16_t    index;  // used locally
   int32_t    numOfSids;
   int32_t    pSidExtInfoList[];  // offset value of STableSidExtInfo
@@ -707,7 +687,7 @@ typedef struct STableMeta {
   int16_t rowSize;  // used locally, calculated in client
   int16_t sversion;
   int8_t  numOfVpeers;
-  SVPeerDesc vpeerDesc[TSDB_VNODES_SUPPORT];
+  SVnodeDesc vpeerDesc[TSDB_VNODES_SUPPORT];
   int32_t  sid;
   int32_t  vgid;
   uint64_t uid;
@@ -735,27 +715,27 @@ typedef struct {
   char     db[TSDB_DB_NAME_LEN + 1];
   uint16_t payloadLen;
   char     payload[];
-} SShowMsg;
+} SCMShowMsg;
 
 typedef struct {
   uint64_t   qhandle;
   STableMeta tableMeta;
-} SShowRsp;
+} SCMShowRsp;
 
 typedef struct {
   char ip[32];
-} SCreateMnodeMsg, SDropMnodeMsg, SCreateDnodeMsg, SDropDnodeMsg;
+} SCMCreateDnodeMsg, SCMDropDnodeMsg;
 
 typedef struct {
   uint32_t dnode;
   int32_t  vnode;
   int32_t  sid;
-} STableCfgMsg;
+} SDMConfigTableMsg;
 
 typedef struct {
   uint32_t dnode;
   int32_t vnode;
-} SVpeerCfgMsg;
+} SDMConfigVnodeMsg;
 
 typedef struct {
   char ip[32];
@@ -793,18 +773,18 @@ typedef struct {
 typedef struct {
   SQqueryList qlist;
   SStreamList slist;
-} SHeartBeatMsg;
+} SCMHeartBeatMsg;
 
 typedef struct {
   uint32_t  queryId;
   uint32_t  streamId;
   int8_t    killConnection;
   SRpcIpSet ipList;
-} SHeartBeatRsp;
+} SCMHeartBeatRsp;
 
 typedef struct {
   char queryId[TSDB_KILL_MSG_LEN + 1];
-} SKillQueryMsg, SKillStreamMsg, SKillConnectionMsg;
+} SCMKillQueryMsg, SCMKillStreamMsg, SCMKillConnMsg;
 
 typedef struct {
   int32_t  vnode;
@@ -813,7 +793,7 @@ typedef struct {
   uint64_t stime;  // stream starting time
   int32_t  status;
   char     tableId[TSDB_TABLE_ID_LEN + 1];
-} SDAlterStreamMsg;
+} SMDAlterStreamMsg;
 
 #pragma pack(pop)
 

src/inc/trpc.h

@@ -84,7 +84,7 @@ void *rpcReallocCont(void *ptr, int contLen);
 void  rpcSendRequest(void *thandle, SRpcIpSet *pIpSet, SRpcMsg *pMsg);
 void  rpcSendResponse(SRpcMsg *pMsg);
 void  rpcSendRedirectRsp(void *pConn, SRpcIpSet *pIpSet); 
-void  rpcGetConnInfo(void *thandle, SRpcConnInfo *pInfo);
+int   rpcGetConnInfo(void *thandle, SRpcConnInfo *pInfo);
 
 #ifdef __cplusplus
 }

src/kit/shell/src/shellDarwin.c

@@ -81,7 +81,7 @@ void shellParseArgument(int argc, char *argv[], struct arguments *arguments) {
       // for management port
     else if (strcmp(argv[i], "-P") == 0) {
       if (i < argc - 1) {
-        tsMgmtShellPort = atoi(argv[++i]);
+        tsMnodeShellPort = atoi(argv[++i]);
       } else {
         fprintf(stderr, "option -P requires an argument\n");
         exit(EXIT_FAILURE);

src/kit/shell/src/shellEngine.c

@@ -68,7 +68,7 @@ TAOS *shellInit(struct arguments *args) {
   tsMeterMetaKeepTimer = 3000;
 
   // Connect to the database.
-  TAOS *con = taos_connect(args->host, args->user, args->password, args->database, tsMgmtShellPort);
+  TAOS *con = taos_connect(args->host, args->user, args->password, args->database, tsMnodeShellPort);
   if (con == NULL) {
     return con;
   }

src/kit/shell/src/shellImport.c

@@ -227,7 +227,7 @@ static void shellRunImportThreads(struct arguments* args)
     ShellThreadObj *pThread = threadObj + t;
     pThread->threadIndex = t;
     pThread->totalThreads = args->threadNum;
-    pThread->taos = taos_connect(args->host, args->user, args->password, args->database, tsMgmtShellPort);
+    pThread->taos = taos_connect(args->host, args->user, args->password, args->database, tsMnodeShellPort);
     if (pThread->taos == NULL) {
       fprintf(stderr, "ERROR: thread:%d failed connect to TDengine, error:%s\n", pThread->threadIndex, taos_errstr(pThread->taos));
       exit(0);

src/kit/shell/src/shellLinux.c

@@ -62,7 +62,7 @@ static error_t parse_opt(int key, char *arg, struct argp_state *state) {
       if (arg) arguments->password = arg;
       break;
     case 'P':
-      tsMgmtShellPort = atoi(arg);
+      tsMnodeShellPort = atoi(arg);
       break;
     case 't':
       arguments->timezone = arg;

src/kit/shell/src/shellMain.c

@@ -81,17 +81,6 @@ struct arguments args = {
  */
 int main(int argc, char* argv[]) {
   /*setlocale(LC_ALL, "en_US.UTF-8"); */
-  //
-  if (argc == 1)
-  {
-    printf("=== this a test for debug usage\n");
-    void *taos = taos_connect(NULL, "root", "taosdata", NULL, 0);
-    taos_query(taos, "select * from d1.t6");
-    while (1) {
-      sleep(1000);
-    }
-  }
-  //
 
   if (!checkVersion()) {
     exit(EXIT_FAILURE);

src/kit/shell/src/shellWindows.c

@@ -61,7 +61,7 @@ void shellParseArgument(int argc, char *argv[], struct arguments *arguments) {
     // for management port
     else if (strcmp(argv[i], "-P") == 0) {
       if (i < argc - 1) {
-        tsMgmtShellPort = atoi(argv[++i]);
+        tsMnodeShellPort = atoi(argv[++i]);
       } else {
         fprintf(stderr, "option -P requires an argument\n");
         exit(EXIT_FAILURE);

src/mnode/inc/mgmtChildTable.h

@@ -30,8 +30,8 @@ int32_t mgmtInitChildTables();
 void    mgmtCleanUpChildTables();
 void *  mgmtGetChildTable(char *tableId);
 
-int32_t mgmtCreateChildTable(SCreateTableMsg *pCreate, int32_t contLen, SVgObj *pVgroup, int32_t sid,
-                             SDMCreateTableMsg **pDCreateOut, STableInfo **pTableOut);
+int32_t mgmtCreateChildTable(SCMCreateTableMsg *pCreate, int32_t contLen, SVgObj *pVgroup, int32_t sid,
+                             SMDCreateTableMsg **pDCreateOut, STableInfo **pTableOut);
 int32_t mgmtDropChildTable(SDbObj *pDb, SChildTableObj *pTable);
 int32_t mgmtModifyChildTableTagValueByName(SChildTableObj *pTable, char *tagName, char *nContent);
 

src/mnode/inc/mgmtDServer.h

@@ -27,7 +27,7 @@ void    mgmtAddDServerMsgHandle(uint8_t msgType, void (*fp)(SRpcMsg *rpcMsg));
 
 //extern void *mgmtStatusTimer;
 //
-//void mgmtSendCreateTableMsg(SDMCreateTableMsg *pCreate, SRpcIpSet *ipSet, void *ahandle);
+//void mgmtSendCreateTableMsg(SMDCreateTableMsg *pCreate, SRpcIpSet *ipSet, void *ahandle);
 //void mgmtSendDropTableMsg(SMDDropTableMsg *pRemove, SRpcIpSet *ipSet, void *ahandle);
 //void mgmtSendAlterStreamMsg(STableInfo *pTable, SRpcIpSet *ipSet, void *ahandle);
 //void mgmtSendDropVnodeMsg(int32_t vnode, SRpcIpSet *ipSet, void *ahandle);

src/mnode/inc/mgmtNormalTable.h

@@ -28,8 +28,8 @@ int32_t mgmtInitNormalTables();
 void    mgmtCleanUpNormalTables();
 void *  mgmtGetNormalTable(char *tableId);
 
-int32_t mgmtCreateNormalTable(SCreateTableMsg *pCreate, int32_t contLen, SVgObj *pVgroup, int32_t sid,
-                              SDMCreateTableMsg **pDCreateOut, STableInfo **pTableOut);
+int32_t mgmtCreateNormalTable(SCMCreateTableMsg *pCreate, int32_t contLen, SVgObj *pVgroup, int32_t sid,
+                              SMDCreateTableMsg **pDCreateOut, STableInfo **pTableOut);
 int32_t mgmtDropNormalTable(SDbObj *pDb, SNormalTableObj *pTable);
 int32_t mgmtAddNormalTableColumn(SNormalTableObj *pTable, SSchema schema[], int32_t ncols);
 int32_t mgmtDropNormalTableColumnByName(SNormalTableObj *pTable, char *colName);

src/mnode/inc/mgmtSuperTable.h

@@ -31,7 +31,7 @@ void    mgmtCleanUpSuperTables();
 
 void *  mgmtGetSuperTable(char *tableId);
 
-int32_t mgmtCreateSuperTable(SDbObj *pDb, SCreateTableMsg *pCreate);
+int32_t mgmtCreateSuperTable(SDbObj *pDb, SCMCreateTableMsg *pCreate);
 int32_t mgmtDropSuperTable(SDbObj *pDb, SSuperTableObj *pTable);
 int32_t mgmtAddSuperTableTag(SSuperTableObj *pTable, SSchema schema[], int32_t ntags);
 int32_t mgmtDropSuperTableTag(SSuperTableObj *pTable, char *tagName);

src/mnode/inc/mgmtSystem.h

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

src/mnode/inc/mgmtTable.h

@@ -33,20 +33,20 @@ STableInfo* mgmtGetTableByPos(uint32_t dnodeIp, int32_t vnode, int32_t sid);
 int32_t     mgmtGetTableMeta(SDbObj *pDb, STableInfo *pTable, STableMeta *pMeta, bool usePublicIp);
 
 int32_t  mgmtRetrieveMetricMeta(void *pConn, char **pStart, SSuperTableMetaMsg *pInfo);
-int32_t  mgmtCreateTable(SCreateTableMsg *pCreate, int32_t contLen, void *thandle, bool isGetMeta);
+int32_t  mgmtCreateTable(SCMCreateTableMsg *pCreate, int32_t contLen, void *thandle, bool isGetMeta);
 int32_t  mgmtDropTable(SDbObj *pDb, char *tableId, int32_t ignore);
-int32_t  mgmtAlterTable(SDbObj *pDb, SAlterTableMsg *pAlter);
+int32_t  mgmtAlterTable(SDbObj *pDb, SCMAlterTableMsg *pAlter);
 
 void    mgmtAddTableIntoSuperTable(SSuperTableObj *pStable);
 void    mgmtRemoveTableFromSuperTable(SSuperTableObj *pStable);
 void    mgmtSetTableDirty(STableInfo *pTable, bool isDirty);
 
 SMDDropTableMsg *mgmtBuildRemoveTableMsg(STableInfo *pTable);
-SDRemoveSuperTableMsg *mgmtBuildRemoveSuperTableMsg(STableInfo *pTable);
+SMDDropSTableMsg *mgmtBuildRemoveSuperTableMsg(STableInfo *pTable);
 
 void mgmtProcessGetTableMeta(STableInfo *pTable, void *thandle);
-void mgmtProcessCreateTable(SVgObj *pVgroup, SCreateTableMsg *pCreate, int32_t contLen, void *thandle, bool isGetMeta);
-void mgmtProcessCreateVgroup(SCreateTableMsg *pCreate, int32_t contLen, void *thandle, bool isGetMeta);
+void mgmtProcessCreateTable(SVgObj *pVgroup, SCMCreateTableMsg *pCreate, int32_t contLen, void *thandle, bool isGetMeta);
+void mgmtProcessCreateVgroup(SCMCreateTableMsg *pCreate, int32_t contLen, void *thandle, bool isGetMeta);
 
 #ifdef __cplusplus
 }

src/mnode/src/mgmtChildTable.c

@@ -274,9 +274,9 @@ void mgmtCleanUpChildTables() {
 
 static void *mgmtBuildCreateChildTableMsg(SChildTableObj *pTable, SVgObj *pVgroup, void *pTagData, int32_t tagDataLen) {
   int32_t totalCols = pTable->superTable->numOfColumns + pTable->superTable->numOfTags;
-  int32_t contLen   = sizeof(SDMCreateTableMsg) + totalCols * sizeof(SSchema) + tagDataLen;
+  int32_t contLen   = sizeof(SMDCreateTableMsg) + totalCols * sizeof(SSchema) + tagDataLen;
 
-  SDMCreateTableMsg *pCreateTable = rpcMallocCont(contLen);
+  SMDCreateTableMsg *pCreateTable = rpcMallocCont(contLen);
   if (pCreateTable == NULL) {
     return NULL;
   }
@@ -309,13 +309,13 @@ static void *mgmtBuildCreateChildTableMsg(SChildTableObj *pTable, SVgObj *pVgrou
     pSchema++;
   }
 
-  memcpy(pCreateTable + sizeof(SDMCreateTableMsg) + totalCols * sizeof(SSchema), pTagData, tagDataLen);
+  memcpy(pCreateTable + sizeof(SMDCreateTableMsg) + totalCols * sizeof(SSchema), pTagData, tagDataLen);
 
   return pCreateTable;
 }
 
-int32_t mgmtCreateChildTable(SCreateTableMsg *pCreate, int32_t contLen, SVgObj *pVgroup, int32_t sid,
-                             SDMCreateTableMsg **pDCreateOut, STableInfo **pTableOut) {
+int32_t mgmtCreateChildTable(SCMCreateTableMsg *pCreate, int32_t contLen, SVgObj *pVgroup, int32_t sid,
+                             SMDCreateTableMsg **pDCreateOut, STableInfo **pTableOut) {
   int32_t numOfTables = sdbGetNumOfRows(tsChildTableSdb);
   if (numOfTables >= tsMaxTables) {
     mError("table:%s, numOfTables:%d exceed maxTables:%d", pCreate->tableId, numOfTables, tsMaxTables);
@@ -350,7 +350,7 @@ int32_t mgmtCreateChildTable(SCreateTableMsg *pCreate, int32_t contLen, SVgObj *
   }
 
   pTagData += (TSDB_TABLE_ID_LEN + 1);
-  int32_t tagDataLen = contLen - sizeof(SCreateTableMsg) - TSDB_TABLE_ID_LEN - 1;
+  int32_t tagDataLen = contLen - sizeof(SCMCreateTableMsg) - TSDB_TABLE_ID_LEN - 1;
   *pDCreateOut = mgmtBuildCreateChildTableMsg(pTable, pVgroup, pTagData, tagDataLen);
   if (*pDCreateOut == NULL) {
     mError("table:%s, failed to build create table message", pCreate->tableId);

src/mnode/src/mgmtDServer.c

@@ -41,7 +41,7 @@ static void  *tsMgmtDServerRpc;
 int32_t mgmtInitDServer() {
   SRpcInit rpcInit = {0};
   rpcInit.localIp = tsAnyIp ? "0.0.0.0" : tsPrivateIp;;
-  rpcInit.localPort    = tsMgmtDnodePort;
+  rpcInit.localPort    = tsMnodeDnodePort;
   rpcInit.label        = "MND-DS";
   rpcInit.numOfThreads = 1;
   rpcInit.cfp          = mgmtProcessMsgFromDnode;
@@ -89,7 +89,7 @@ static int mgmtDServerRetrieveAuth(char *user, char *spi, char *encrypt, char *s
 //
 //
 //static void mgmtProcessTableCfgMsg(int8_t msgType, int8_t *pCont, int32_t contLen, void *thandle) {
-//  STableCfgMsg *pCfg = (STableCfgMsg *) pCont;
+//  SDMConfigTableMsg *pCfg = (SDMConfigTableMsg *) pCont;
 //  pCfg->dnode = htonl(pCfg->dnode);
 //  pCfg->vnode = htonl(pCfg->vnode);
 //  pCfg->sid   = htonl(pCfg->sid);
@@ -121,7 +121,7 @@ static int mgmtDServerRetrieveAuth(char *user, char *spi, char *encrypt, char *s
 //    return;
 //  }
 //
-//  SVpeerCfgMsg *pCfg = (SVpeerCfgMsg *) pCont;
+//  SDMConfigVnodeMsg *pCfg = (SDMConfigVnodeMsg *) pCont;
 //  pCfg->dnode = htonl(pCfg->dnode);
 //  pCfg->vnode = htonl(pCfg->vnode);
 //
@@ -234,7 +234,7 @@ static int mgmtDServerRetrieveAuth(char *user, char *spi, char *encrypt, char *s
 //
 //  mTrace("msg:%d:%s is received from dnode, pConn:%p", msgType, taosMsg[(int8_t)msgType], pConn);
 //
-//  if (msgType == TSDB_MSG_TYPE_TABLE_CFG) {
+//  if (msgType == TSDB_MSG_TYPE_DM_CONFIG_TABLE) {
 //    mgmtProcessTableCfgMsg(msgType, pCont, contLen, pConn);
 //  } else if (msgType == TSDB_MSG_TYPE_DM_CONFIG_VNODE) {
 //    mgmtProcessVnodeCfgMsg(msgType, pCont, contLen, pConn);
@@ -249,8 +249,8 @@ static int mgmtDServerRetrieveAuth(char *user, char *spi, char *encrypt, char *s
 //  } else if (msgType == TSDB_MSG_TYPE_MD_DROP_STABLE) {
 //    mgmtProcessDropStableRsp(msgType, pCont, contLen, pConn, code);
 //  } else if (msgType == TSDB_MSG_TYPE_MD_CONFIG_DNODE_RSP) {
-//  } else if (msgType == TSDB_MSG_TYPE_ALTER_STREAM_RSP) {
-//  } else if (msgType == TSDB_MSG_TYPE_STATUS) {
+//  } else if (msgType == TSDB_MSG_TYPE_CM_ALTER_STREAM_RSP) {
+//  } else if (msgType == TSDB_MSG_TYPE_DM_STATUS) {
 //    mgmtProcessDnodeStatus(msgType, pCont, contLen, pConn, code);
 //  }  else {
 //    mError("%s from dnode is not processed", taosMsg[(int8_t)msgType]);
@@ -317,7 +317,7 @@ static int mgmtDServerRetrieveAuth(char *user, char *spi, char *encrypt, char *s
 //}
 //
 //void mgmtProcessDnodeStatus(int8_t msgType, void *pCont, int32_t contLen, void *pConn, int32_t code) {
-//  SStatusMsg *pStatus = (SStatusMsg *)pCont;
+//  SDMStatusMsg *pStatus = (SDMStatusMsg *)pCont;
 //
 //  SDnodeObj *pObj = mgmtGetDnode(htonl(pStatus->privateIp));
 //  if (pObj == NULL) {

src/mnode/src/mgmtDb.c

@@ -35,7 +35,7 @@ static void *tsDbSdb = NULL;
 static int32_t tsDbUpdateSize;
 
 static int32_t mgmtUpdateDb(SDbObj *pDb);
-static int32_t mgmtCreateDb(SAcctObj *pAcct, SCreateDbMsg *pCreate);
+static int32_t mgmtCreateDb(SAcctObj *pAcct, SCMCreateDbMsg *pCreate);
 static int32_t mgmtDropDbByName(SAcctObj *pAcct, char *name, short ignoreNotExists);
 static int32_t mgmtDropDb(SDbObj *pDb);
 
@@ -81,7 +81,7 @@ int32_t mgmtInitDbs() {
   SDbObj tObj;
   tsDbUpdateSize = tObj.updateEnd - (char *)&tObj;
 
-  tsDbSdb = sdbOpenTable(tsMaxDbs, tsDbUpdateSize, "db", SDB_KEYTYPE_STRING, tsMgmtDirectory, mgmtDbAction);
+  tsDbSdb = sdbOpenTable(tsMaxDbs, tsDbUpdateSize, "dbs", SDB_KEYTYPE_STRING, tsMgmtDirectory, mgmtDbAction);
   if (tsDbSdb == NULL) {
     mError("failed to init db data");
     return -1;
@@ -108,9 +108,9 @@ int32_t mgmtInitDbs() {
     }
   }
 
-  mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CREATE_DB, mgmtProcessCreateDbMsg);
-  mgmtAddShellMsgHandle(TSDB_MSG_TYPE_ALTER_DB, mgmtProcessAlterDbMsg);
-  mgmtAddShellMsgHandle(TSDB_MSG_TYPE_DROP_DB, mgmtProcessDropDbMsg);
+  mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_CREATE_DB, mgmtProcessCreateDbMsg);
+  mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_ALTER_DB, mgmtProcessAlterDbMsg);
+  mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_DROP_DB, mgmtProcessDropDbMsg);
   mgmtAddShellShowMetaHandle(TSDB_MGMT_TABLE_DB, mgmtGetDbMeta);
   mgmtAddShellShowRetrieveHandle(TSDB_MGMT_TABLE_DB, mgmtRetrieveDbs);
 
@@ -133,7 +133,7 @@ SDbObj *mgmtGetDbByTableId(char *tableId) {
   return (SDbObj *)sdbGetRow(tsDbSdb, db);
 }
 
-static int32_t mgmtCheckDBParams(SCreateDbMsg *pCreate) {
+static int32_t mgmtCheckDBParams(SCMCreateDbMsg *pCreate) {
   if (pCreate->commitLog < 0 || pCreate->commitLog > 1) {
     mError("invalid db option commitLog: %d, only 0 or 1 allowed", pCreate->commitLog);
     return TSDB_CODE_INVALID_OPTION;
@@ -206,7 +206,7 @@ static int32_t mgmtCheckDBParams(SCreateDbMsg *pCreate) {
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mgmtCheckDbParams(SCreateDbMsg *pCreate) {
+static int32_t mgmtCheckDbParams(SCMCreateDbMsg *pCreate) {
   // assign default parameters
   if (pCreate->maxSessions < 0) pCreate->maxSessions = tsSessionsPerVnode;      //
   if (pCreate->cacheBlockSize < 0) pCreate->cacheBlockSize = tsCacheBlockSize;  //
@@ -251,7 +251,7 @@ static int32_t mgmtCheckDbParams(SCreateDbMsg *pCreate) {
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t mgmtCreateDb(SAcctObj *pAcct, SCreateDbMsg *pCreate) {
+static int32_t mgmtCreateDb(SAcctObj *pAcct, SCMCreateDbMsg *pCreate) {
   int32_t numOfDbs = sdbGetNumOfRows(tsDbSdb);
   if (numOfDbs >= tsMaxDbs) {
     mWarn("numOfDbs:%d, exceed tsMaxDbs:%d", numOfDbs, tsMaxDbs);
@@ -442,7 +442,7 @@ static void mgmtMonitorDbDrop(void *unused, void *unusedt) {
   }
 }
 
-static int32_t mgmtAlterDb(SAcctObj *pAcct, SAlterDbMsg *pAlter) {
+static int32_t mgmtAlterDb(SAcctObj *pAcct, SCMAlterDbMsg *pAlter) {
   return 0;
 //  int32_t code = TSDB_CODE_SUCCESS;
 //
@@ -915,7 +915,7 @@ static void mgmtProcessCreateDbMsg(SRpcMsg *rpcMsg) {
     return;
   }
 
-  SCreateDbMsg *pCreate = (SCreateDbMsg *) rpcMsg->pCont;
+  SCMCreateDbMsg *pCreate = (SCMCreateDbMsg *) rpcMsg->pCont;
 
   pCreate->maxSessions     = htonl(pCreate->maxSessions);
   pCreate->cacheBlockSize  = htonl(pCreate->cacheBlockSize);
@@ -953,7 +953,7 @@ static void mgmtProcessAlterDbMsg(SRpcMsg *rpcMsg) {
     return;
   }
 
-  SAlterDbMsg *pAlter = (SAlterDbMsg *) rpcMsg->pCont;
+  SCMAlterDbMsg *pAlter = (SCMAlterDbMsg *) rpcMsg->pCont;
   pAlter->daysPerFile = htonl(pAlter->daysPerFile);
   pAlter->daysToKeep  = htonl(pAlter->daysToKeep);
   pAlter->maxSessions = htonl(pAlter->maxSessions) + 1;
@@ -983,7 +983,7 @@ static void mgmtProcessDropDbMsg(SRpcMsg *rpcMsg) {
   }
 
   if (pUser->superAuth) {
-    SDropDbMsg *pDrop = rpcMsg->pCont;
+    SCMDropDbMsg *pDrop = rpcMsg->pCont;
     rpcRsp.code = mgmtDropDbByName(pUser->pAcct, pDrop->db, pDrop->ignoreNotExists);
     if (rpcRsp.code == TSDB_CODE_SUCCESS) {
       mLPrint("DB:%s is dropped by %s", pDrop->db, pUser->user);

src/mnode/src/mgmtSystem.c → src/mnode/src/mgmtMain.c

@@ -27,41 +27,28 @@
 #include "mgmtDServer.h"
 #include "mgmtVgroup.h"
 #include "mgmtUser.h"
-#include "mgmtSystem.h"
 #include "mgmtTable.h"
 #include "mgmtShell.h"
 
+static int32_t mgmtCheckMgmtRunning();
 char tsMgmtDirectory[128] = {0};
 void *tsMgmtTmr = NULL;
-void *tsMgmtTranQhandle   = NULL;
 
+int32_t mgmtInitSystem() {
+  if (mgmtInitShell() != 0) {
+    mError("failed to init shell");
+    return -1;
+  }
 
-void mgmtCleanUpSystem() {
-  mPrint("starting to clean up mgmt");
-
-  sdbCleanUpPeers();
-  mgmtCleanupBalance();
-  mgmtCleanupDClient();
-  mgmtCleanupDServer();
-  mgmtCleanUpShell();
-  mgmtCleanUpTables();
-  mgmtCleanUpVgroups();
-  mgmtCleanUpDbs();
-  mgmtCleanUpDnodes();
-  mgmtCleanUpUsers();
-  mgmtCleanUpAccts();
-  taosTmrCleanUp(tsMgmtTmr);
-  taosCleanUpScheduler(tsMgmtTranQhandle);
-
-  mPrint("mgmt is cleaned up");
-}
+  struct stat dirstat;
+  bool fileExist  = (stat(tsMgmtDirectory, &dirstat) == 0);
+  bool asMaster = (strcmp(tsMasterIp, tsPrivateIp) == 0);
 
-int32_t mgmtCheckMgmtRunning() {
-  if (tsModuleStatus & (1 << TSDB_MOD_MGMT)) {
+  if (asMaster || fileExist) {
+    if (mgmtStartSystem() != 0) {
       return -1;
     }
-
-  tsetModuleStatus(TSDB_MOD_MGMT);
+  }
 
   return 0;
 }
@@ -79,11 +66,9 @@ int32_t mgmtStartSystem() {
     return 0;
   }
 
-  tsMgmtTranQhandle = taosInitScheduler(tsMaxDnodes + tsMaxShellConns, 1, "mnodeT");
-
   tsMgmtTmr = taosTmrInit((tsMaxDnodes + tsMaxShellConns) * 3, 200, 3600000, "MND");
   if (tsMgmtTmr == NULL) {
-    mError("failed to init timer, exit");
+    mError("failed to init timer");
     return -1;
   }
 
@@ -125,11 +110,6 @@ int32_t mgmtStartSystem() {
     return -1;
   }
 
-  if (mgmtInitShell() < 0) {
-    mError("failed to init shell");
-    return -1;
-  }
-
   if (sdbInitPeers(tsMgmtDirectory) < 0) {
     mError("failed to init peers");
     return -1;
@@ -139,30 +119,11 @@ int32_t mgmtStartSystem() {
     mError("failed to init dnode balance")
   }
 
-
   mPrint("TDengine mgmt is initialized successfully");
 
   return 0;
 }
 
-int32_t mgmtInitSystem() {
-  struct stat dirstat;
-  bool directoryExist  = (stat(tsMgmtDirectory, &dirstat) == 0);
-  bool equalWithMaster = (strcmp(tsMasterIp, tsPrivateIp) == 0);
-
-  if (equalWithMaster || directoryExist) {
-    if (mgmtStartSystem() != 0) {
-      return -1;
-    }
-  }
-
-  if (mgmtInitShell() < 0) {
-    mError("failed to init shell");
-    return -1;
-  }
-
-  return 0;
-}
 
 void mgmtStopSystem() {
   if (sdbMaster) {
@@ -172,5 +133,30 @@ void mgmtStopSystem() {
 
   mgmtCleanUpSystem();
   remove(tsMgmtDirectory);
-//  mgmtInitRedirect();
+}
+
+void mgmtCleanUpSystem() {
+  mPrint("starting to clean up mgmt");
+  sdbCleanUpPeers();
+  mgmtCleanupBalance();
+  mgmtCleanUpShell();
+  mgmtCleanupDClient();
+  mgmtCleanupDServer();
+  mgmtCleanUpTables();
+  mgmtCleanUpVgroups();
+  mgmtCleanUpDbs();
+  mgmtCleanUpDnodes();
+  mgmtCleanUpUsers();
+  mgmtCleanUpAccts();
+  taosTmrCleanUp(tsMgmtTmr);
+  mPrint("mgmt is cleaned up");
+}
+
+static int32_t mgmtCheckMgmtRunning() {
+  if (tsModuleStatus & (1 << TSDB_MOD_MGMT)) {
+    return -1;
+  }
+
+  tsetModuleStatus(TSDB_MOD_MGMT);
+  return 0;
 }
\ No newline at end of file

src/mnode/src/mgmtMnode.c

@@ -18,5 +18,5 @@
 #include "mgmtMnode.h"
 
 bool mgmtCheckRedirect(void *handle) {
-  return true;
+  return false;
 }
\ No newline at end of file

src/mnode/src/mgmtNormalTable.c

@@ -289,9 +289,9 @@ void mgmtCleanUpNormalTables() {
 
 static void *mgmtBuildCreateNormalTableMsg(SNormalTableObj *pTable, SVgObj *pVgroup) {
   int32_t totalCols = pTable->numOfColumns;
-  int32_t contLen   = sizeof(SDMCreateTableMsg) + totalCols * sizeof(SSchema) + pTable->sqlLen;
+  int32_t contLen   = sizeof(SMDCreateTableMsg) + totalCols * sizeof(SSchema) + pTable->sqlLen;
 
-  SDMCreateTableMsg *pCreateTable = rpcMallocCont(contLen);
+  SMDCreateTableMsg *pCreateTable = rpcMallocCont(contLen);
   if (pCreateTable == NULL) {
     return NULL;
   }
@@ -323,13 +323,13 @@ static void *mgmtBuildCreateNormalTableMsg(SNormalTableObj *pTable, SVgObj *pVgr
     pSchema++;
   }
 
-  memcpy(pCreateTable + sizeof(SDMCreateTableMsg) + totalCols * sizeof(SSchema), pTable->sql, pTable->sqlLen);
+  memcpy(pCreateTable + sizeof(SMDCreateTableMsg) + totalCols * sizeof(SSchema), pTable->sql, pTable->sqlLen);
 
   return pCreateTable;
 }
 
-int32_t mgmtCreateNormalTable(SCreateTableMsg *pCreate, int32_t contLen, SVgObj *pVgroup, int32_t sid,
-                              SDMCreateTableMsg **pDCreateOut, STableInfo **pTableOut) {
+int32_t mgmtCreateNormalTable(SCMCreateTableMsg *pCreate, int32_t contLen, SVgObj *pVgroup, int32_t sid,
+                              SMDCreateTableMsg **pDCreateOut, STableInfo **pTableOut) {
   int32_t numOfTables = sdbGetNumOfRows(tsNormalTableSdb);
   if (numOfTables >= TSDB_MAX_NORMAL_TABLES) {
     mError("table:%s, numOfTables:%d exceed maxTables:%d", pCreate->tableId, numOfTables, TSDB_MAX_NORMAL_TABLES);

src/mnode/src/mgmtProfile.c

@@ -22,7 +22,7 @@
 #include "mgmtShell.h"
 #include "mgmtUser.h"
 
-int32_t mgmtSaveQueryStreamList(SHeartBeatMsg *pHBMsg);
+int32_t mgmtSaveQueryStreamList(SCMHeartBeatMsg *pHBMsg);
 
 int32_t mgmtKillQuery(char *qidstr, void *pConn);
 int32_t mgmtKillStream(char *qidstr, void *pConn);
@@ -63,7 +63,7 @@ typedef struct {
   SStreamDesc   sdesc[];
 } SStreamShow;
 
-int32_t  mgmtSaveQueryStreamList(SHeartBeatMsg *pHBMsg) {
+int32_t  mgmtSaveQueryStreamList(SCMHeartBeatMsg *pHBMsg) {
 //  SAcctObj *pAcct = pConn->pAcct;
 //
 //  if (contLen <= 0 || pAcct == NULL) {
@@ -684,7 +684,7 @@ void mgmtProcessKillQueryMsg(SRpcMsg *rpcMsg) {
     return;
   }
 
-  SKillQueryMsg *pKill = (SKillQueryMsg *) rpcMsg->pCont;
+  SCMKillQueryMsg *pKill = (SCMKillQueryMsg *) rpcMsg->pCont;
   int32_t code;
 
   if (!pUser->writeAuth) {
@@ -708,7 +708,7 @@ void mgmtProcessKillStreamMsg(SRpcMsg *rpcMsg) {
     return;
   }
 
-  SKillStreamMsg *pKill = (SKillStreamMsg *) rpcMsg->pCont;
+  SCMKillStreamMsg *pKill = (SCMKillStreamMsg *) rpcMsg->pCont;
   int32_t code;
 
   if (!pUser->writeAuth) {
@@ -732,7 +732,7 @@ void mgmtProcessKillConnectionMsg(SRpcMsg *rpcMsg) {
     return;
   }
 
-  SKillConnectionMsg *pKill = (SKillConnectionMsg *) rpcMsg->pCont;
+  SCMKillConnMsg *pKill = (SCMKillConnMsg *) rpcMsg->pCont;
   int32_t code;
 
   if (!pUser->writeAuth) {
@@ -752,9 +752,9 @@ int32_t mgmtInitProfile() {
   mgmtAddShellShowRetrieveHandle(TSDB_MGMT_TABLE_CONNS, mgmtRetrieveConns);
   mgmtAddShellShowMetaHandle(TSDB_MGMT_TABLE_STREAMS, mgmtGetStreamMeta);
   mgmtAddShellShowRetrieveHandle(TSDB_MGMT_TABLE_STREAMS, mgmtRetrieveStreams);
-  mgmtAddShellMsgHandle(TSDB_MSG_TYPE_KILL_QUERY, mgmtProcessKillQueryMsg);
-  mgmtAddShellMsgHandle(TSDB_MSG_TYPE_KILL_STREAM, mgmtProcessKillStreamMsg);
-  mgmtAddShellMsgHandle(TSDB_MSG_TYPE_KILL_CONNECTION, mgmtProcessKillConnectionMsg);
+  mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_KILL_QUERY, mgmtProcessKillQueryMsg);
+  mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_KILL_STREAM, mgmtProcessKillStreamMsg);
+  mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_KILL_CONN, mgmtProcessKillConnectionMsg);
 
   return 0;
 }

src/mnode/src/mgmtShell.c

@@ -21,7 +21,7 @@
 #include "trpc.h"
 #include "tstatus.h"
 #include "tsched.h"
-#include "dnodeSystem.h"
+#include "dnode.h"
 #include "mnode.h"
 #include "mgmtAcct.h"
 #include "mgmtBalance.h"
@@ -51,14 +51,17 @@ static void mgmtProcessHeartBeatMsg(SRpcMsg *rpcMsg);
 static void mgmtProcessConnectMsg(SRpcMsg *rpcMsg);
 
 static void *tsMgmtShellRpc = NULL;
+static void *tsMgmtTranQhandle = NULL;
 static void (*tsMgmtProcessShellMsgFp[TSDB_MSG_TYPE_MAX])(SRpcMsg *) = {0};
 static SShowMetaFp     tsMgmtShowMetaFp[TSDB_MGMT_TABLE_MAX]     = {0};
 static SShowRetrieveFp tsMgmtShowRetrieveFp[TSDB_MGMT_TABLE_MAX] = {0};
 
 int32_t mgmtInitShell() {
-  mgmtAddShellMsgHandle(TSDB_MSG_TYPE_SHOW, mgmtProcessShowMsg);
+  mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_SHOW, mgmtProcessShowMsg);
   mgmtAddShellMsgHandle(TSDB_MSG_TYPE_RETRIEVE, mgmtProcessRetrieveMsg);
 
+  tsMgmtTranQhandle = taosInitScheduler(tsMaxDnodes + tsMaxShellConns, 1, "mnodeT");
+
   int32_t numOfThreads = tsNumOfCores * tsNumOfThreadsPerCore / 4.0;
   if (numOfThreads < 1) {
     numOfThreads = 1;
@@ -66,7 +69,7 @@ int32_t mgmtInitShell() {
 
   SRpcInit rpcInit = {0};
   rpcInit.localIp      = tsAnyIp ? "0.0.0.0" : tsPrivateIp;
-  rpcInit.localPort    = tsMgmtShellPort;
+  rpcInit.localPort    = tsMnodeShellPort;
   rpcInit.label        = "MND-shell";
   rpcInit.numOfThreads = numOfThreads;
   rpcInit.cfp          = mgmtProcessMsgFromShell;
@@ -81,14 +84,19 @@ int32_t mgmtInitShell() {
     return -1;
   }
 
-  mgmtAddShellMsgHandle(TSDB_MSG_TYPE_HEARTBEAT, mgmtProcessHeartBeatMsg);
-  mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CONNECT, mgmtProcessConnectMsg);
+  mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_HEARTBEAT, mgmtProcessHeartBeatMsg);
+  mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_CONNECT, mgmtProcessConnectMsg);
 
   mPrint("server connection to shell is opened");
   return 0;
 }
 
 void mgmtCleanUpShell() {
+  if (tsMgmtTranQhandle) {
+    taosCleanUpScheduler(tsMgmtTranQhandle);
+    tsMgmtTranQhandle = NULL;
+  }
+
   if (tsMgmtShellRpc) {
     rpcClose(tsMgmtShellRpc);
     tsMgmtShellRpc = NULL;
@@ -112,11 +120,15 @@ void mgmtProcessTranRequest(SSchedMsg *sched) {
   SRpcMsg *rpcMsg = sched->msg;
   (*tsMgmtProcessShellMsgFp[rpcMsg->msgType])(rpcMsg);
   rpcFreeCont(rpcMsg->pCont);
+  free(rpcMsg);
 }
 
 void mgmtAddToTranRequest(SRpcMsg *rpcMsg) {
+  SRpcMsg *queuedRpcMsg = malloc(sizeof(SRpcMsg));
+  memcpy(queuedRpcMsg, rpcMsg, sizeof(SRpcMsg));
+
   SSchedMsg schedMsg;
-  schedMsg.msg     = rpcMsg;
+  schedMsg.msg = queuedRpcMsg;
   schedMsg.fp  = mgmtProcessTranRequest;
   taosScheduleTask(tsMgmtTranQhandle, &schedMsg);
 }
@@ -130,6 +142,7 @@ static void mgmtProcessMsgFromShell(SRpcMsg *rpcMsg) {
     return;
   }
 
+  mTrace("%s is received", taosMsg[rpcMsg->msgType]);
   if (tsMgmtProcessShellMsgFp[rpcMsg->msgType]) {
     if (mgmtCheckMsgReadOnly(rpcMsg->msgType, rpcMsg->pCont)) {
       (*tsMgmtProcessShellMsgFp[rpcMsg->msgType])(rpcMsg);
@@ -147,15 +160,15 @@ static void mgmtProcessMsgFromShell(SRpcMsg *rpcMsg) {
 static void mgmtProcessShowMsg(SRpcMsg *rpcMsg) {
   SRpcMsg rpcRsp = {.handle = rpcMsg->handle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0};
 
-  SShowMsg *pShowMsg = rpcMsg->pCont;
+  SCMShowMsg *pShowMsg = rpcMsg->pCont;
   if (pShowMsg->type == TSDB_MGMT_TABLE_DNODE || TSDB_MGMT_TABLE_GRANTS || TSDB_MGMT_TABLE_SCORES) {
     if (mgmtCheckRedirect(rpcMsg->handle) != TSDB_CODE_SUCCESS) {
       return;
     }
   }
 
-  int32_t  size = sizeof(SShowRsp) + sizeof(SSchema) * TSDB_MAX_COLUMNS + TSDB_EXTRA_PAYLOAD_SIZE;
-  SShowRsp *pShowRsp = rpcMallocCont(size);
+  int32_t  size = sizeof(SCMShowRsp) + sizeof(SSchema) * TSDB_MAX_COLUMNS + TSDB_EXTRA_PAYLOAD_SIZE;
+  SCMShowRsp *pShowRsp = rpcMallocCont(size);
   if (pShowRsp == NULL) {
     rpcRsp.code = TSDB_CODE_SERV_OUT_OF_MEMORY;
     rpcSendResponse(&rpcRsp);
@@ -179,9 +192,9 @@ static void mgmtProcessShowMsg(SRpcMsg *rpcMsg) {
     mgmtSaveQhandle(pShow);
     pShowRsp->qhandle = htobe64((uint64_t) pShow);
     if (tsMgmtShowMetaFp[pShowMsg->type]) {
-      code = (*tsMgmtShowMetaFp[(uint8_t) pShowMsg->type])(&pShowRsp->tableMeta, pShow, rpcMsg->handle);
+      code = (*tsMgmtShowMetaFp[pShowMsg->type])(&pShowRsp->tableMeta, pShow, rpcMsg->handle);
       if (code == 0) {
-        size = sizeof(SShowRsp) + sizeof(SSchema) * pShow->numOfColumns;
+        size = sizeof(SCMShowRsp) + sizeof(SSchema) * pShow->numOfColumns;
       } else {
         mError("pShow:%p, type:%d %s, failed to get Meta, code:%d", pShow, pShowMsg->type,
                taosMsg[(uint8_t) pShowMsg->type], code);
@@ -245,7 +258,7 @@ static void mgmtProcessRetrieveMsg(SRpcMsg *rpcMsg) {
 
   // if free flag is set, client wants to clean the resources
   if ((pRetrieve->free & TSDB_QUERY_TYPE_FREE_RESOURCE) != TSDB_QUERY_TYPE_FREE_RESOURCE)
-    rowsRead = (*tsMgmtShowRetrieveFp[(uint8_t) pShow->type])(pShow, pRsp->data, rowsToRead, rpcMsg->handle);
+    rowsRead = (*tsMgmtShowRetrieveFp[pShow->type])(pShow, pRsp->data, rowsToRead, rpcMsg->handle);
 
   if (rowsRead < 0) {
     rowsRead = 0;  // TSDB_CODE_ACTION_IN_PROGRESS;
@@ -267,10 +280,10 @@ static void mgmtProcessRetrieveMsg(SRpcMsg *rpcMsg) {
 
 static void mgmtProcessHeartBeatMsg(SRpcMsg *rpcMsg) {
   SRpcMsg rpcRsp = {.handle = rpcMsg->handle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0};
-  //SHeartBeatMsg *pHBMsg = (SHeartBeatMsg *) rpcMsg->pCont;
+  //SCMHeartBeatMsg *pHBMsg = (SCMHeartBeatMsg *) rpcMsg->pCont;
   //mgmtSaveQueryStreamList(pHBMsg);
 
-  SHeartBeatRsp *pHBRsp = (SHeartBeatRsp *) rpcMallocCont(sizeof(SHeartBeatRsp));
+  SCMHeartBeatRsp *pHBRsp = (SCMHeartBeatRsp *) rpcMallocCont(sizeof(SCMHeartBeatRsp));
   if (pHBRsp == NULL) {
     rpcRsp.code = TSDB_CODE_SERV_OUT_OF_MEMORY;
     rpcSendResponse(&rpcRsp);
@@ -278,10 +291,13 @@ static void mgmtProcessHeartBeatMsg(SRpcMsg *rpcMsg) {
   }
 
   SRpcConnInfo connInfo;
-  rpcGetConnInfo(rpcMsg->handle, &connInfo);
+  if (rpcGetConnInfo(rpcMsg->handle, &connInfo) != 0) {
+    mError("conn:%p is already released while process heart beat msg", rpcMsg->handle);
+    return;
+  }
 
   pHBRsp->ipList.inUse = 0;
-  pHBRsp->ipList.port = htons(tsMgmtShellPort);
+  pHBRsp->ipList.port = htons(tsMnodeShellPort);
   pHBRsp->ipList.numOfIps = 0;
   if (pSdbPublicIpList != NULL && pSdbIpList != NULL) {
     pHBRsp->ipList.numOfIps = htons(pSdbPublicIpList->numOfIps);
@@ -305,7 +321,7 @@ static void mgmtProcessHeartBeatMsg(SRpcMsg *rpcMsg) {
   pHBRsp->killConnection = 0;
 
   rpcRsp.pCont = pHBRsp;
-  rpcRsp.contLen = sizeof(SHeartBeatRsp);
+  rpcRsp.contLen = sizeof(SCMHeartBeatRsp);
   rpcSendResponse(&rpcRsp);
 }
 
@@ -326,12 +342,15 @@ static int mgmtShellRetriveAuth(char *user, char *spi, char *encrypt, char *secr
 
 static void mgmtProcessConnectMsg(SRpcMsg *rpcMsg) {
   SRpcMsg rpcRsp = {.handle = rpcMsg->handle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0};
-  SConnectMsg *pConnectMsg = (SConnectMsg *) rpcMsg->pCont;
+  SCMConnectMsg *pConnectMsg = (SCMConnectMsg *) rpcMsg->pCont;
 
   SRpcConnInfo connInfo;
-  rpcGetConnInfo(rpcMsg->handle, &connInfo);
-  int32_t code;
+  if (rpcGetConnInfo(rpcMsg->handle, &connInfo) != 0) {
+    mError("conn:%p is already released while process connect msg", rpcMsg->handle);
+    return;
+  }
 
+  int32_t code;
   SUserObj *pUser = mgmtGetUser(connInfo.user);
   if (pUser == NULL) {
     code = TSDB_CODE_INVALID_USER;
@@ -364,7 +383,7 @@ static void mgmtProcessConnectMsg(SRpcMsg *rpcMsg) {
     }
   }
 
-  SConnectRsp *pConnectRsp = rpcMallocCont(sizeof(SConnectRsp));
+  SCMConnectRsp *pConnectRsp = rpcMallocCont(sizeof(SCMConnectRsp));
   if (pConnectRsp == NULL) {
     code = TSDB_CODE_SERV_OUT_OF_MEMORY;
     goto connect_over;
@@ -375,7 +394,7 @@ static void mgmtProcessConnectMsg(SRpcMsg *rpcMsg) {
   pConnectRsp->writeAuth = pUser->writeAuth;
   pConnectRsp->superAuth = pUser->superAuth;
   pConnectRsp->ipList.inUse = 0;
-  pConnectRsp->ipList.port = htons(tsMgmtShellPort);
+  pConnectRsp->ipList.port = htons(tsMnodeShellPort);
   pConnectRsp->ipList.numOfIps = 0;
   if (pSdbPublicIpList != NULL && pSdbIpList != NULL) {
     pConnectRsp->ipList.numOfIps = htons(pSdbPublicIpList->numOfIps);
@@ -397,7 +416,7 @@ connect_over:
   } else {
     mLPrint("user:%s login from %s, code:%d", connInfo.user, taosIpStr(connInfo.clientIp), code);
     rpcRsp.pCont   = pConnectRsp;
-    rpcRsp.contLen = sizeof(SConnectRsp);
+    rpcRsp.contLen = sizeof(SCMConnectRsp);
   }
   rpcSendResponse(&rpcRsp);
 }
@@ -406,7 +425,7 @@ connect_over:
  * check if we need to add mgmtProcessTableMetaMsg into tranQueue, which will be executed one-by-one.
  */
 static bool mgmtCheckMeterMetaMsgType(void *pMsg) {
-  STableInfoMsg *pInfo = (STableInfoMsg *) pMsg;
+  SCMTableInfoMsg *pInfo = (SCMTableInfoMsg *) pMsg;
   int16_t autoCreate = htons(pInfo->createFlag);
   STableInfo *pTable = mgmtGetTable(pInfo->tableId);
 
@@ -420,10 +439,10 @@ static bool mgmtCheckMeterMetaMsgType(void *pMsg) {
 }
 
 static bool mgmtCheckMsgReadOnly(int8_t type, void *pCont) {
-  if ((type == TSDB_MSG_TYPE_TABLE_META && (!mgmtCheckMeterMetaMsgType(pCont)))  ||
-       type == TSDB_MSG_TYPE_STABLE_META || type == TSDB_MSG_TYPE_RETRIEVE ||
-       type == TSDB_MSG_TYPE_SHOW || type == TSDB_MSG_TYPE_MULTI_TABLE_META      ||
-       type == TSDB_MSG_TYPE_CONNECT) {
+  if ((type == TSDB_MSG_TYPE_CM_TABLE_META && (!mgmtCheckMeterMetaMsgType(pCont)))  ||
+       type == TSDB_MSG_TYPE_CM_STABLE_META || type == TSDB_MSG_TYPE_RETRIEVE ||
+       type == TSDB_MSG_TYPE_CM_SHOW || type == TSDB_MSG_TYPE_CM_TABLES_META      ||
+       type == TSDB_MSG_TYPE_CM_CONNECT) {
     return true;
   }
 

src/mnode/src/mgmtSuperTable.c

@@ -200,7 +200,7 @@ void mgmtCleanUpSuperTables() {
   sdbCloseTable(tsSuperTableSdb);
 }
 
-int32_t mgmtCreateSuperTable(SDbObj *pDb, SCreateTableMsg *pCreate) {
+int32_t mgmtCreateSuperTable(SDbObj *pDb, SCMCreateTableMsg *pCreate) {
   int32_t numOfTables = sdbGetNumOfRows(tsSuperTableSdb);
   if (numOfTables >= TSDB_MAX_SUPER_TABLES) {
     mError("stable:%s, numOfTables:%d exceed maxTables:%d", pCreate->tableId, numOfTables, TSDB_MAX_SUPER_TABLES);
@@ -259,7 +259,7 @@ void* mgmtGetSuperTable(char *tableId) {
 
 void *mgmtGetSuperTableVgroup(SSuperTableObj *pStable) {
   //TODO get vgroup of dnodes
-  SSuperTableInfoRsp *rsp = rpcMallocCont(sizeof(SSuperTableInfoRsp) + sizeof(uint32_t) * mgmtGetDnodesNum());
+  SCMSuperTableInfoRsp *rsp = rpcMallocCont(sizeof(SCMSuperTableInfoRsp) + sizeof(uint32_t) * mgmtGetDnodesNum());
   rsp->numOfDnodes = 1;
   rsp->dnodeIps[0] = 0;
   return rsp;

src/mnode/src/mgmtTable.c

@@ -71,12 +71,12 @@ int32_t mgmtInitTables() {
 
   mgmtSetVgroupIdPool();
 
-  mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CREATE_TABLE, mgmtProcessCreateTableMsg);
-  mgmtAddShellMsgHandle(TSDB_MSG_TYPE_DROP_TABLE, mgmtProcessDropTableMsg);
-  mgmtAddShellMsgHandle(TSDB_MSG_TYPE_ALTER_TABLE, mgmtProcessAlterTableMsg);
-  mgmtAddShellMsgHandle(TSDB_MSG_TYPE_TABLE_META, mgmtProcessTableMetaMsg);
-  mgmtAddShellMsgHandle(TSDB_MSG_TYPE_MULTI_TABLE_META, mgmtProcessMultiTableMetaMsg);
-  mgmtAddShellMsgHandle(TSDB_MSG_TYPE_STABLE_META, mgmtProcessSuperTableMetaMsg);
+  mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_CREATE_TABLE, mgmtProcessCreateTableMsg);
+  mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_DROP_TABLE, mgmtProcessDropTableMsg);
+  mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_ALTER_TABLE, mgmtProcessAlterTableMsg);
+  mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_TABLE_META, mgmtProcessTableMetaMsg);
+  mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_TABLES_META, mgmtProcessMultiTableMetaMsg);
+  mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_STABLE_META, mgmtProcessSuperTableMetaMsg);
   mgmtAddShellShowMetaHandle(TSDB_MGMT_TABLE_TABLE, mgmtGetShowTableMeta);
   mgmtAddShellShowRetrieveHandle(TSDB_MGMT_TABLE_TABLE, mgmtRetrieveShowTables);
 
@@ -133,7 +133,7 @@ int32_t mgmtGetTableMeta(SDbObj *pDb, STableInfo *pTable, STableMeta *pMeta, boo
 
 
 
-void mgmtProcessCreateVgroup(SCreateTableMsg *pCreate, int32_t contLen, void *thandle, bool isGetMeta) {
+void mgmtProcessCreateVgroup(SCMCreateTableMsg *pCreate, int32_t contLen, void *thandle, bool isGetMeta) {
   SRpcMsg rpcRsp = {.handle = thandle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0};
   SDbObj *pDb = mgmtGetDb(pCreate->db);
   if (pDb == NULL) {
@@ -175,7 +175,7 @@ void mgmtProcessCreateVgroup(SCreateTableMsg *pCreate, int32_t contLen, void *th
   mgmtSendCreateVgroupMsg(pVgroup, info);
 }
 
-//void mgmtSendCreateTableMsg(SDMCreateTableMsg *pCreate, SRpcIpSet *ipSet, void *ahandle) {
+//void mgmtSendCreateTableMsg(SMDCreateTableMsg *pCreate, SRpcIpSet *ipSet, void *ahandle) {
 //  mTrace("table:%s, send create table msg, ahandle:%p", pCreate->tableId, ahandle);
 //  SRpcMsg rpcMsg = {
 //    .handle  = ahandle,
@@ -189,7 +189,7 @@ void mgmtProcessCreateVgroup(SCreateTableMsg *pCreate, int32_t contLen, void *th
 //
 
 
-void mgmtProcessCreateTable(SVgObj *pVgroup, SCreateTableMsg *pCreate, int32_t contLen, void *thandle, bool isGetMeta) {
+void mgmtProcessCreateTable(SVgObj *pVgroup, SCMCreateTableMsg *pCreate, int32_t contLen, void *thandle, bool isGetMeta) {
   assert(pVgroup != NULL);
   SRpcMsg rpcRsp = {.handle = thandle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0};
 
@@ -201,7 +201,7 @@ void mgmtProcessCreateTable(SVgObj *pVgroup, SCreateTableMsg *pCreate, int32_t c
   }
 
   STableInfo *pTable;
-  SDMCreateTableMsg *pDCreate = NULL;
+  SMDCreateTableMsg *pDCreate = NULL;
 
   if (pCreate->numOfColumns == 0) {
     mTrace("table:%s, start to create child table, vgroup:%d sid:%d", pCreate->tableId, pVgroup->vgId, sid);
@@ -239,7 +239,7 @@ void mgmtProcessCreateTable(SVgObj *pVgroup, SCreateTableMsg *pCreate, int32_t c
   mgmtSendMsgToDnode(&ipSet, &rpcMsg);
 }
 
-int32_t mgmtCreateTable(SCreateTableMsg *pCreate, int32_t contLen, void *thandle, bool isGetMeta) {
+int32_t mgmtCreateTable(SCMCreateTableMsg *pCreate, int32_t contLen, void *thandle, bool isGetMeta) {
   SDbObj *pDb = mgmtGetDb(pCreate->db);
   if (pDb == NULL) {
     mError("table:%s, failed to create table, db not selected", pCreate->tableId);
@@ -331,7 +331,7 @@ int32_t mgmtDropTable(SDbObj *pDb, char *tableId, int32_t ignore) {
   }
 }
 
-int32_t mgmtAlterTable(SDbObj *pDb, SAlterTableMsg *pAlter) {
+int32_t mgmtAlterTable(SDbObj *pDb, SCMAlterTableMsg *pAlter) {
   STableInfo *pTable = mgmtGetTable(pAlter->tableId);
   if (pTable == NULL) {
     return TSDB_CODE_INVALID_TABLE;
@@ -553,7 +553,7 @@ void mgmtSetTableDirty(STableInfo *pTable, bool isDirty) {
 void mgmtProcessCreateTableMsg(SRpcMsg *rpcMsg) {
   SRpcMsg rpcRsp = {.handle = rpcMsg->handle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0};
 
-  SCreateTableMsg *pCreate = (SCreateTableMsg *) rpcMsg->pCont;
+  SCMCreateTableMsg *pCreate = (SCMCreateTableMsg *) rpcMsg->pCont;
   pCreate->numOfColumns = htons(pCreate->numOfColumns);
   pCreate->numOfTags    = htons(pCreate->numOfTags);
   pCreate->sqlLen       = htons(pCreate->sqlLen);
@@ -594,7 +594,7 @@ void mgmtProcessCreateTableMsg(SRpcMsg *rpcMsg) {
 
 void mgmtProcessDropTableMsg(SRpcMsg *rpcMsg) {
   SRpcMsg rpcRsp = {.handle = rpcMsg->handle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0};
-  SDropTableMsg *pDrop = (SDropTableMsg *) rpcMsg->pCont;
+  SCMDropTableMsg *pDrop = (SCMDropTableMsg *) rpcMsg->pCont;
 
   if (mgmtCheckRedirect(rpcMsg->handle) != TSDB_CODE_SUCCESS) {
     mError("table:%s, failed to drop table, need redirect message", pDrop->tableId);
@@ -644,7 +644,7 @@ void mgmtProcessAlterTableMsg(SRpcMsg *rpcMsg) {
     return;
   }
 
-  SAlterTableMsg *pAlter = (SAlterTableMsg *) rpcMsg->pCont;
+  SCMAlterTableMsg *pAlter = (SCMAlterTableMsg *) rpcMsg->pCont;
 
   if (!pUser->writeAuth) {
     rpcRsp.code = TSDB_CODE_NO_RIGHTS;
@@ -686,7 +686,11 @@ void mgmtProcessGetTableMeta(STableInfo *pTable, void *thandle) {
   }
 
   SRpcConnInfo connInfo;
-  rpcGetConnInfo(thandle, &connInfo);
+  if (rpcGetConnInfo(thandle, &connInfo) != 0) {
+    mError("conn:%p is already released while get table meta", thandle);
+    return;
+  }
+
   bool usePublicIp = (connInfo.serverIp == tsPublicIpInt);
 
   STableMeta *pMeta = rpcMallocCont(sizeof(STableMeta) + sizeof(SSchema) * TSDB_MAX_COLUMNS);
@@ -710,7 +714,7 @@ void mgmtProcessTableMetaMsg(SRpcMsg *rpcMsg) {
   rpcRsp.pCont   = NULL;
   rpcRsp.contLen = 0;
 
-  STableInfoMsg *pInfo = rpcMsg->pCont;
+  SCMTableInfoMsg *pInfo = rpcMsg->pCont;
   pInfo->createFlag = htons(pInfo->createFlag);
 
   SUserObj *pUser = mgmtGetUserFromConn(rpcMsg->handle);
@@ -735,8 +739,8 @@ void mgmtProcessTableMetaMsg(SRpcMsg *rpcMsg) {
         return;
       }
 
-      int32_t contLen = sizeof(SCreateTableMsg) + sizeof(STagData);
-      SCreateTableMsg *pCreateMsg = rpcMallocCont(contLen);
+      int32_t contLen = sizeof(SCMCreateTableMsg) + sizeof(STagData);
+      SCMCreateTableMsg *pCreateMsg = rpcMallocCont(contLen);
       if (pCreateMsg == NULL) {
         mError("table:%s, failed to create table while get meta info, no enough memory", pInfo->tableId);
         rpcRsp.code = TSDB_CODE_SERV_OUT_OF_MEMORY;
@@ -762,7 +766,10 @@ void mgmtProcessMultiTableMetaMsg(SRpcMsg *rpcMsg) {
   rpcRsp.contLen = 0;
 
   SRpcConnInfo connInfo;
-  rpcGetConnInfo(rpcMsg->handle, &connInfo);
+  if (rpcGetConnInfo(rpcMsg->handle, &connInfo) != 0) {
+    mError("conn:%p is already released while get mulit table meta", rpcMsg->handle);
+    return;
+  }
 
   bool usePublicIp = (connInfo.serverIp == tsPublicIpInt);
   SUserObj *pUser = mgmtGetUser(connInfo.user);
@@ -772,7 +779,7 @@ void mgmtProcessMultiTableMetaMsg(SRpcMsg *rpcMsg) {
     return;
   }
 
-  SMultiTableInfoMsg *pInfo = rpcMsg->pCont;
+  SCMMultiTableInfoMsg *pInfo = rpcMsg->pCont;
   pInfo->numOfTables = htonl(pInfo->numOfTables);
 
   int32_t totalMallocLen = 4*1024*1024; // first malloc 4 MB, subsequent reallocation as twice
@@ -823,7 +830,7 @@ void mgmtProcessMultiTableMetaMsg(SRpcMsg *rpcMsg) {
 
 void mgmtProcessSuperTableMetaMsg(SRpcMsg *rpcMsg) {
   SRpcMsg rpcRsp = {.handle = rpcMsg->handle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0};
-  SSuperTableInfoMsg *pInfo = rpcMsg->pCont;
+  SCMSuperTableInfoMsg *pInfo = rpcMsg->pCont;
   STableInfo *pTable = mgmtGetSuperTable(pInfo->tableId);
   if (pTable == NULL) {
     rpcRsp.code = TSDB_CODE_INVALID_TABLE;
@@ -831,7 +838,7 @@ void mgmtProcessSuperTableMetaMsg(SRpcMsg *rpcMsg) {
     return;
   }
 
-  SSuperTableInfoRsp *pRsp = mgmtGetSuperTableVgroup((SSuperTableObj *) pTable);
+  SCMSuperTableInfoRsp *pRsp = mgmtGetSuperTableVgroup((SSuperTableObj *) pTable);
   if (pRsp != NULL) {
     int32_t msgLen = sizeof(SSuperTableObj) + htonl(pRsp->numOfDnodes) * sizeof(int32_t);
     rpcRsp.pCont = pRsp;

src/mnode/src/mgmtUser.c

@@ -59,7 +59,7 @@ int32_t mgmtInitUsers() {
   SUserObj tObj;
   tsUserUpdateSize = tObj.updateEnd - (int8_t *)&tObj;
 
-  tsUserSdb = sdbOpenTable(tsMaxUsers, tsUserUpdateSize, "user", SDB_KEYTYPE_STRING, tsMgmtDirectory, mgmtUserAction);
+  tsUserSdb = sdbOpenTable(tsMaxUsers, tsUserUpdateSize, "users", SDB_KEYTYPE_STRING, tsMgmtDirectory, mgmtUserAction);
   if (tsUserSdb == NULL) {
     mError("failed to init user data");
     return -1;
@@ -83,9 +83,9 @@ int32_t mgmtInitUsers() {
   mgmtCreateUser(pAcct, "monitor", tsInternalPass);
   mgmtCreateUser(pAcct, "_root", tsInternalPass);
 
-  mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CREATE_USER, mgmtProcessCreateUserMsg);
-  mgmtAddShellMsgHandle(TSDB_MSG_TYPE_ALTER_USER, mgmtProcessAlterUserMsg);
-  mgmtAddShellMsgHandle(TSDB_MSG_TYPE_DROP_USER, mgmtProcessDropUserMsg);
+  mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_CREATE_USER, mgmtProcessCreateUserMsg);
+  mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_ALTER_USER, mgmtProcessAlterUserMsg);
+  mgmtAddShellMsgHandle(TSDB_MSG_TYPE_CM_DROP_USER, mgmtProcessDropUserMsg);
   mgmtAddShellShowMetaHandle(TSDB_MGMT_TABLE_USER, mgmtGetUserMeta);
   mgmtAddShellShowRetrieveHandle(TSDB_MGMT_TABLE_USER, mgmtRetrieveUsers);
   
@@ -123,7 +123,7 @@ static int32_t mgmtCreateUser(SAcctObj *pAcct, char *name, char *pass) {
 
   SUserObj *pUser = (SUserObj *)sdbGetRow(tsUserSdb, name);
   if (pUser != NULL) {
-    mWarn("user:%s is already there", name);
+    mTrace("user:%s is already there", name);
     return TSDB_CODE_USER_ALREADY_EXIST;
   }
 
@@ -330,9 +330,11 @@ static void *mgmtUserActionDestroy(void *row, char *str, int32_t size, int32_t *
 
 SUserObj *mgmtGetUserFromConn(void *pConn) {
   SRpcConnInfo connInfo;
-  rpcGetConnInfo(pConn, &connInfo);
-
+  if (rpcGetConnInfo(pConn, &connInfo) == 0) {
     return mgmtGetUser(connInfo.user);
+  }
+
+  return NULL;
 }
 
 static void mgmtProcessCreateUserMsg(SRpcMsg *rpcMsg) {
@@ -347,7 +349,7 @@ static void mgmtProcessCreateUserMsg(SRpcMsg *rpcMsg) {
   }
 
   if (pUser->superAuth) {
-    SCreateUserMsg *pCreate = rpcMsg->pCont;
+    SCMCreateUserMsg *pCreate = rpcMsg->pCont;
     rpcRsp.code = mgmtCreateUser(pUser->pAcct, pCreate->user, pCreate->pass);
     if (rpcRsp.code == TSDB_CODE_SUCCESS) {
       mLPrint("user:%s is created by %s", pCreate->user, pUser->user);
@@ -370,7 +372,7 @@ static void mgmtProcessAlterUserMsg(SRpcMsg *rpcMsg) {
     return;
   }
 
-  SAlterUserMsg *pAlter = rpcMsg->pCont;
+  SCMAlterUserMsg *pAlter = rpcMsg->pCont;
   SUserObj *pUser = mgmtGetUser(pAlter->user);
   if (pUser == NULL) {
     rpcRsp.code = TSDB_CODE_INVALID_USER;
@@ -477,7 +479,7 @@ static void mgmtProcessDropUserMsg(SRpcMsg *rpcMsg) {
     return ;
   }
 
-  SDropUserMsg *pDrop = rpcMsg->pCont;
+  SCMDropUserMsg *pDrop = rpcMsg->pCont;
   SUserObj *pUser = mgmtGetUser(pDrop->user);
   if (pUser == NULL) {
     rpcRsp.code = TSDB_CODE_INVALID_USER;

src/mnode/src/mgmtVgroup.c

@@ -534,7 +534,7 @@ SMDCreateVnodeMsg *mgmtBuildCreateVnodeMsg(SVgObj *pVgroup, int32_t vnode) {
   pCfg->replications                 = (char) pVgroup->numOfVnodes;
   pCfg->rowsInFileBlock              = htonl(pCfg->rowsInFileBlock);
 
-  SVPeerDesc *vpeerDesc = pVPeers->vpeerDesc;
+  SVnodeDesc *vpeerDesc = pVPeers->vpeerDesc;
   for (int32_t j = 0; j < pVgroup->numOfVnodes; ++j) {
     vpeerDesc[j].ip = htonl(pVgroup->vnodeGid[j].ip);
     vpeerDesc[j].vnode = htonl(pVgroup->vnodeGid[j].vnode);
@@ -558,7 +558,7 @@ SVgObj *mgmtGetVgroupByVnode(uint32_t dnode, int32_t vnode) {
 }
 
 SRpcIpSet mgmtGetIpSetFromVgroup(SVgObj *pVgroup) {
-  SRpcIpSet ipSet = {.numOfIps = pVgroup->numOfVnodes, .inUse = 0, .port = tsMgmtDnodePort + 1};
+  SRpcIpSet ipSet = {.numOfIps = pVgroup->numOfVnodes, .inUse = 0, .port = tsMnodeDnodePort + 1};
   for (int i = 0; i < pVgroup->numOfVnodes; ++i) {
     ipSet.ip[i] = pVgroup->vnodeGid[i].ip;
   }
@@ -566,7 +566,7 @@ SRpcIpSet mgmtGetIpSetFromVgroup(SVgObj *pVgroup) {
 }
 
 SRpcIpSet mgmtGetIpSetFromIp(uint32_t ip) {
-  SRpcIpSet ipSet = {.ip[0] = ip, .numOfIps = 1, .inUse = 0, .port = tsMgmtDnodePort + 1};
+  SRpcIpSet ipSet = {.ip[0] = ip, .numOfIps = 1, .inUse = 0, .port = tsMnodeDnodePort + 1};
   return ipSet;
 }
 

src/rpc/src/rpcMain.c

@@ -360,8 +360,8 @@ void rpcSendRequest(void *shandle, SRpcIpSet *pIpSet, SRpcMsg *pMsg) {
   // for TDengine, all the query, show commands shall have TCP connection
   char type = pMsg->msgType;
   if (type == TSDB_MSG_TYPE_QUERY || type == TSDB_MSG_TYPE_RETRIEVE ||
-      type == TSDB_MSG_TYPE_STABLE_META || type == TSDB_MSG_TYPE_MULTI_TABLE_META ||
-      type == TSDB_MSG_TYPE_SHOW ) 
+      type == TSDB_MSG_TYPE_CM_STABLE_META || type == TSDB_MSG_TYPE_CM_TABLES_META ||
+      type == TSDB_MSG_TYPE_CM_SHOW )
     pContext->connType = RPC_CONN_TCPC;
   
   rpcSendReqToServer(pRpc, pContext);
@@ -441,15 +441,16 @@ void rpcSendRedirectRsp(void *thandle, SRpcIpSet *pIpSet) {
   return;
 }
 
-void rpcGetConnInfo(void *thandle, SRpcConnInfo *pInfo) {
+int rpcGetConnInfo(void *thandle, SRpcConnInfo *pInfo) {
   SRpcConn  *pConn = (SRpcConn *)thandle;
+  if (pConn->user[0] == 0) return -1;
 
   pInfo->clientIp = pConn->peerIp;
   pInfo->clientPort = pConn->peerPort;
   pInfo->serverIp = pConn->destIp;
 
-  assert(pConn->user[0]);
   strcpy(pInfo->user, pConn->user);
+  return 0;
 }
 
 static void rpcFreeMsg(void *msg) {
@@ -814,7 +815,7 @@ static void *rpcProcessMsgFromPeer(SRecvInfo *pRecv) {
   terrno = 0;
   pConn = rpcProcessMsgHead(pRpc, pRecv);
 
-  if (pHead->msgType < TSDB_MSG_TYPE_HEARTBEAT || (rpcDebugFlag & 16)) {
+  if (pHead->msgType < TSDB_MSG_TYPE_CM_HEARTBEAT || (rpcDebugFlag & 16)) {
     tTrace("%s %p, %s received from 0x%x:%hu, parse code:%x len:%d sig:0x%08x:0x%08x:%d",
         pRpc->label, pConn, taosMsg[pHead->msgType], pRecv->ip, pRecv->port, terrno, 
         pRecv->msgLen, pHead->sourceId, pHead->destId, pHead->tranId, pHead->port);
@@ -983,12 +984,12 @@ static void rpcSendMsgToPeer(SRpcConn *pConn, void *msg, int msgLen) {
   msgLen = rpcAddAuthPart(pConn, msg, msgLen);
 
   if ( rpcIsReq(pHead->msgType)) {
-    if (pHead->msgType < TSDB_MSG_TYPE_HEARTBEAT || (rpcDebugFlag & 16))
+    if (pHead->msgType < TSDB_MSG_TYPE_CM_HEARTBEAT || (rpcDebugFlag & 16))
       tTrace("%s %p, %s is sent to %s:%hu, len:%d sig:0x%08x:0x%08x:%d",
              pRpc->label, pConn, taosMsg[pHead->msgType], pConn->peerIpstr,
              pConn->peerPort, msgLen, pHead->sourceId, pHead->destId, pHead->tranId);
   } else {
-    if (pHead->msgType < TSDB_MSG_TYPE_HEARTBEAT || (rpcDebugFlag & 16))
+    if (pHead->msgType < TSDB_MSG_TYPE_CM_HEARTBEAT || (rpcDebugFlag & 16))
       tTrace( "%s %p, %s is sent to %s:%hu, code:%u len:%d sig:0x%08x:0x%08x:%d",
           pRpc->label, pConn, taosMsg[pHead->msgType], pConn->peerIpstr, pConn->peerPort, 
           (uint8_t)pHead->content[0], msgLen, pHead->sourceId, pHead->destId, pHead->tranId);

src/util/inc/tglobalcfg.h

@@ -58,9 +58,10 @@ extern char osName[];
 
 extern char  tsMasterIp[];
 extern char  tsSecondIp[];
-extern uint16_t tsMgmtDnodePort;
-extern uint16_t tsMgmtShellPort;
-extern uint16_t tsVnodeShellPort;
+extern uint16_t tsMnodeDnodePort;
+extern uint16_t tsMnodeShellPort;
+extern uint16_t tsDnodeShellPort;
+extern uint16_t tsDnodeMnodePort;
 extern uint16_t tsVnodeVnodePort;
 extern uint16_t tsMgmtMgmtPort;
 extern uint16_t tsMgmtSyncPort;

src/util/src/tglobalcfg.c

@@ -61,9 +61,10 @@ int64_t tsMsPerDay[] = {86400000L, 86400000000L};
 
 char  tsMasterIp[TSDB_IPv4ADDR_LEN] = {0};
 char  tsSecondIp[TSDB_IPv4ADDR_LEN] = {0};
-uint16_t tsMgmtShellPort = 6030;   // udp[6030-6034] tcp[6030]
-uint16_t tsVnodeShellPort = 6035;  // udp[6035-6039] tcp[6035]
-uint16_t tsMgmtDnodePort = 6040;   // udp[6040-6044] tcp[6040]
+uint16_t tsMnodeShellPort = 6030;   // udp[6030-6034] tcp[6030]
+uint16_t tsDnodeShellPort = 6035;  // udp[6035-6039] tcp[6035]
+uint16_t tsMnodeDnodePort = 6040;   // udp/tcp
+uint16_t tsDnodeMnodePort = 6041;   // udp/tcp
 uint16_t tsVnodeVnodePort = 6045;  // tcp[6045]
 uint16_t tsMgmtMgmtPort = 6050;    // udp, numOfVnodes fixed to 1, range udp[6050]
 uint16_t tsMgmtSyncPort = 6050;    // tcp, range tcp[6050]
@@ -492,13 +493,13 @@ static void doInitGlobalConfig() {
   tsInitConfigOption(cfg++, "httpPort", &tsHttpPort, TSDB_CFG_VTYPE_SHORT,
                      TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_SHOW,
                      1, 65535, 0, TSDB_CFG_UTYPE_NONE);
-  tsInitConfigOption(cfg++, "mgmtShellPort", &tsMgmtShellPort, TSDB_CFG_VTYPE_SHORT,
+  tsInitConfigOption(cfg++, "mgmtShellPort", &tsMnodeShellPort, TSDB_CFG_VTYPE_SHORT,
                      TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_SHOW | TSDB_CFG_CTYPE_B_CLIENT,
                      1, 65535, 0, TSDB_CFG_UTYPE_NONE);
-  tsInitConfigOption(cfg++, "vnodeShellPort", &tsVnodeShellPort, TSDB_CFG_VTYPE_SHORT,
+  tsInitConfigOption(cfg++, "vnodeShellPort", &tsDnodeShellPort, TSDB_CFG_VTYPE_SHORT,
                      TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_SHOW | TSDB_CFG_CTYPE_B_CLIENT,
                      1, 65535, 0, TSDB_CFG_UTYPE_NONE);
-  tsInitConfigOption(cfg++, "mgmtVnodePort", &tsMgmtDnodePort, TSDB_CFG_VTYPE_SHORT,
+  tsInitConfigOption(cfg++, "mgmtVnodePort", &tsMnodeDnodePort, TSDB_CFG_VTYPE_SHORT,
                      TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_SHOW | TSDB_CFG_CTYPE_B_CLUSTER,
                      1, 65535, 0, TSDB_CFG_UTYPE_NONE);
   tsInitConfigOption(cfg++, "vnodeVnodePort", &tsVnodeVnodePort, TSDB_CFG_VTYPE_SHORT,

src/util/src/tstring.c

@@ -26,23 +26,23 @@ char *taosMsg[] = {
     "create-table",
     "create-table-rsp",        //10
 
-    "remove-table",
-    "remove-table-rsp",
+    "drop-table",
+    "drop-table-rsp",
+    "alter-table",
+    "alter-table-rsp",
     "create-vnode",
     "create-vnode-rsp",
-    "free-vnode",
-    "free-vnode-rsp",
-    "cfg-dnode",
-    "cfg-dnode-rsp",
-    "alter-stream",
-    "alter-stream-rsp",        //20
+    "drop-vnode",
+    "drop-vnode-rsp",
+    "alter-vnode",
+    "alter-vnode-rsp",        //20
 
-    "sync",
-    "sync-rsp",
-    "forward",
-    "forward-rsp",
     "drop-stable",
     "drop-stable-rsp",
+    "alter-stream",
+    "alter-stream-rsp",
+    "config-dnode",
+    "config-dnode-rsp",
     "",
     "",
     "",
@@ -63,37 +63,26 @@ char *taosMsg[] = {
     "alter-user-rsp",
     "drop-user",
     "drop-user-rsp",
-    "create-mnode",
-    "create-mnode-rsp",
-    "drop-mnode",
-    "drop-mnode-rsp",
     "create-dnode",
-    "create-dnode-rsp",        //50
-
+    "create-dnode-rsp",
     "drop-dnode",
     "drop-dnode-rsp",
-    "alter-dnode",
-    "alter-dnode-rsp",
     "create-db",
-    "create-db-rsp",
+    "create-db-rsp",          //50
+
     "drop-db",
     "drop-db-rsp",
     "use-db",
-    "use-db-rsp",              //60
-
+    "use-db-rsp",
     "alter-db",
     "alter-db-rsp",
     "create-table",
     "create-table-rsp",
     "drop-table",
-    "drop-table-rsp",
+    "drop-table-rsp",        //60
+
     "alter-table",
     "alter-table-rsp",
-    "cfg-vnode",
-    "cfg-vnode-rsp",           //70
-
-    "cfg-table",
-    "cfg-table-rsp",
     "table-meta",
     "table-meta-rsp",
     "super-table-meta",
@@ -101,24 +90,42 @@ char *taosMsg[] = {
     "multi-table-meta",
     "multi-table-meta-rsp",
     "alter-stream",
-    "alter-stream-rsp",        //80
+    "alter-stream-rsp",     //70
 
     "show",
     "show-rsp",
-    "cfg-mnode",
-    "cfg-mnode-rsp",
     "kill-query",
     "kill-query-rsp",
     "kill-stream",
     "kill-stream-rsp",
     "kill-connection",
-    "kill-connectoin-rsp",     //90
-
+    "kill-connectoin-rsp",
     "heart-beat",
-    "heart-beat-rsp",
+    "heart-beat-rsp",      //80
+
+    "",
+    "",
+    "",
+    "",
+    "",
+    "",
+    "",
+    "",                    //90
+
+    "config-table",
+    "config-table-rsp",
+    "config-vnode",
+    "config-vnode-rsp",
     "status",
     "status-rsp",
     "grant",
     "grant-rsp",
+    "",
+    "",                   //100
+
+    "sdb-sync",
+    "sdb-sync-rsp",
+    "sdb-forward",
+    "sdb-forward-rsp",
     "max"
 };
\ No newline at end of file

src/vnode/tsdb/CMakeLists.txt

@@ -15,5 +15,5 @@ IF ((TD_LINUX_64) OR (TD_LINUX_32 AND TD_ARM))
   TARGET_LINK_LIBRARIES(tsdb common tutil)
 
   # Someone has no gtest directory, so comment it
-  ADD_SUBDIRECTORY(tests)
+  #ADD_SUBDIRECTORY(tests)
 ENDIF ()