second draft

src/client/inc/tsclient.h

@@ -298,7 +298,6 @@ typedef struct STscObj {
   char               sversion[TSDB_VERSION_LEN];
   char               writeAuth : 1;
   char               superAuth : 1;
-  void*              pMgmtConn;
   struct SSqlObj *   pSql;
   struct SSqlObj *   pHb;
   struct SSqlObj *   sqlList;
@@ -358,7 +357,7 @@ typedef struct SSqlStream {
   struct SSqlStream *prev, *next;
 } SSqlStream;
 
-int32_t tscInitRpc(const char *user, const char *secret, void** pMgmtConn);
+int32_t tscInitRpc(const char *user, const char *secret);
 void    tscInitMsgsFp();
 
 int tsParseSql(SSqlObj *pSql, bool multiVnodeInsertion);
@@ -425,7 +424,7 @@ void    tscQueueAsyncFreeResult(SSqlObj *pSql);
 int32_t tscToSQLCmd(SSqlObj *pSql, struct SSqlInfo *pInfo);
 char *  tscGetResultColumnChr(SSqlRes *pRes, SQueryInfo *pQueryInfo, int32_t column);
 
-extern void *    pVnodeConn;
+extern void *    pDnodeConn;
 extern void *    tscCacheHandle;
 extern void *    tscTmr;
 extern void *    tscQhandle;

src/client/src/tscServer.c

@@ -191,7 +191,6 @@ void tscProcessActivityTimer(void *handle, void *tmrId) {
 }
 
 int tscSendMsgToServer(SSqlObj *pSql) {
-  STscObj* pObj = pSql->pTscObj;
   SSqlCmd* pCmd = &pSql->cmd;
   
   char *pMsg = rpcMallocCont(pCmd->payloadLen);
@@ -201,30 +200,22 @@ int tscSendMsgToServer(SSqlObj *pSql) {
   }
 
   if (pSql->cmd.command < TSDB_SQL_MGMT) {
-    tscTrace("%p msg:%s is sent to server %d", pSql, taosMsg[pSql->cmd.msgType], pSql->ipList.port);
     memcpy(pMsg, pSql->cmd.payload + tsRpcHeadSize, pSql->cmd.payloadLen);
+  } else {
+    pSql->ipList = tscMgmtIpSet;
+    memcpy(pMsg, pSql->cmd.payload, pSql->cmd.payloadLen);
+  }
 
-    SRpcMsg rpcMsg = {
+  tscTrace("%p msg:%s is sent to server %d", pSql, taosMsg[pSql->cmd.msgType], pSql->ipList.port);
+
+  SRpcMsg rpcMsg = {
       .msgType = pSql->cmd.msgType,
       .pCont   = pMsg,
       .contLen = pSql->cmd.payloadLen,
       .handle  = pSql,
       .code    = 0
-    };
+  };
-    rpcSendRequest(pVnodeConn, &pSql->ipList, &rpcMsg);
+  rpcSendRequest(pDnodeConn, &pSql->ipList, &rpcMsg);
-  } else {
-    pSql->ipList = tscMgmtIpSet;
-    memcpy(pMsg, pSql->cmd.payload, pSql->cmd.payloadLen);
-    SRpcMsg rpcMsg = {
-        .msgType = pSql->cmd.msgType,
-        .pCont   = pMsg,
-        .contLen = pSql->cmd.payloadLen,
-        .handle  = pSql,
-        .code   = 0
-    };
-    tscTrace("%p msg:%s is sent to server", pSql, taosMsg[pSql->cmd.msgType]);
-    rpcSendRequest(pObj->pMgmtConn, &pSql->ipList, &rpcMsg);
-  }
 
   return TSDB_CODE_SUCCESS;
 }

src/client/src/tscSql.c

@@ -66,8 +66,7 @@ STscObj *taosConnectImpl(const char *ip, const char *user, const char *pass, con
     return NULL;
   }
 
-  void* pMgmtConn = NULL;
+  if (tscInitRpc(user, pass) != 0) {
-  if (tscInitRpc(user, pass, &pMgmtConn) != 0) {
     terrno = TSDB_CODE_NETWORK_UNAVAIL;
     return NULL;
   }
@@ -119,7 +118,6 @@ STscObj *taosConnectImpl(const char *ip, const char *user, const char *pass, con
     strtolower(pObj->db, tmp);
   }
 
-  pObj->pMgmtConn = pMgmtConn;
   pthread_mutex_init(&pObj->mutex, NULL);
 
   SSqlObj *pSql = (SSqlObj *)calloc(1, sizeof(SSqlObj));

src/client/src/tscSystem.c

@@ -30,7 +30,7 @@
 #include "tlocale.h"
 
 // global, not configurable
-void *  pVnodeConn;
+void *  pDnodeConn;
 void *  tscCacheHandle;
 void *  tscTmr;
 void *  tscQhandle;
@@ -48,12 +48,12 @@ void tscCheckDiskUsage(void *UNUSED_PARAM(para), void* UNUSED_PARAM(param)) {
   taosTmrReset(tscCheckDiskUsage, 1000, NULL, tscTmr, &tscCheckDiskUsageTmr);
 }
 
-int32_t tscInitRpc(const char *user, const char *secret, void** pMgmtConn) {
+int32_t tscInitRpc(const char *user, const char *secret) {
   SRpcInit rpcInit;
   char secretEncrypt[32] = {0};
   taosEncryptPass((uint8_t *)secret, strlen(secret), secretEncrypt);
 
-  if (pVnodeConn == NULL) {
+  if (pDnodeConn == NULL) {
     memset(&rpcInit, 0, sizeof(rpcInit));
     rpcInit.localPort = 0;
     rpcInit.label = "TSC-vnode";
@@ -66,35 +66,13 @@ int32_t tscInitRpc(const char *user, const char *secret, void** pMgmtConn) {
     rpcInit.ckey = "key";
     rpcInit.secret = secretEncrypt;
 
-    pVnodeConn = rpcOpen(&rpcInit);
+    pDnodeConn = rpcOpen(&rpcInit);
-    if (pVnodeConn == NULL) {
+    if (pDnodeConn == NULL) {
       tscError("failed to init connection to vnode");
       return -1;
     }
   }
 
-  if (*pMgmtConn == NULL) {
-    memset(&rpcInit, 0, sizeof(rpcInit));
-    rpcInit.localPort = 0;
-    rpcInit.label = "TSC-mgmt";
-    rpcInit.numOfThreads = 1;
-    rpcInit.cfp = tscProcessMsgFromServer;
-    rpcInit.ufp = tscUpdateIpSet;
-    rpcInit.sessions = tsMaxMgmtConnections;
-    rpcInit.connType = TAOS_CONN_CLIENT;
-    rpcInit.idleTime = 2000;
-    rpcInit.user = (char*)user;
-    rpcInit.ckey = "key";
-    rpcInit.spi = 1;
-    rpcInit.secret = secretEncrypt;
-
-    *pMgmtConn = rpcOpen(&rpcInit);
-    if (*pMgmtConn == NULL) {
-      tscError("failed to init connection to mgmt");
-      return -1;
-    }
-  }
-
   return 0;
 }
 
@@ -190,9 +168,9 @@ void taos_cleanup() {
   
   taosCloseLog();
   
-  if (pVnodeConn != NULL) {
+  if (pDnodeConn != NULL) {
-    rpcClose(pVnodeConn);
+    rpcClose(pDnodeConn);
-    pVnodeConn = NULL;
+    pDnodeConn = NULL;
   }
   
   taosTmrCleanUp(tscTmr);

src/client/src/tscUtil.c

@@ -760,7 +760,6 @@ void tscCloseTscObj(STscObj* pObj) {
   if (pSql) {
     sem_destroy(&pSql->rspSem);
   }
-  rpcClose(pObj->pMgmtConn);
   
   pthread_mutex_destroy(&pObj->mutex);
   

src/common/src/tglobal.c

@@ -1244,7 +1244,6 @@ bool taosCheckGlobalCfg() {
   
   tsVersion = 10 * tsVersion;
 
-  tsMnodeShellPort = tsServerPort + TSDB_PORT_MNODESHELL;   // udp[6030-6034] tcp[6030]
   tsDnodeShellPort = tsServerPort + TSDB_PORT_DNODESHELL;  // udp[6035-6039] tcp[6035]
   tsMnodeDnodePort = tsServerPort + TSDB_PORT_MNODEDNODE;   // udp/tcp
   tsDnodeMnodePort = tsServerPort + TSDB_PORT_DNODEMNODE;   // udp/tcp

src/dnode/src/dnodeModule.c

@@ -17,6 +17,7 @@
 #include "os.h"
 #include "taosdef.h"
 #include "tglobal.h"
+#include "trpc.h"
 #include "mnode.h"
 #include "http.h"
 #include "monitor.h"

src/dnode/src/dnodeShell.c

@@ -34,6 +34,8 @@ static void  * tsDnodeShellRpc = NULL;
 static int32_t tsDnodeQueryReqNum  = 0;
 static int32_t tsDnodeSubmitReqNum = 0;
 
+void mgmtProcessMsgFromShell(SRpcMsg *rpcMsg);
+
 int32_t dnodeInitShell() {
   dnodeProcessShellMsgFp[TSDB_MSG_TYPE_SUBMIT] = dnodeWrite;
   dnodeProcessShellMsgFp[TSDB_MSG_TYPE_QUERY]  = dnodeRead;
@@ -47,8 +49,8 @@ int32_t dnodeInitShell() {
 
   SRpcInit rpcInit;
   memset(&rpcInit, 0, sizeof(rpcInit));
-  rpcInit.localPort    = tsDnodeShellPort;
+  rpcInit.localPort    = tsMnodeShellPort;
-  rpcInit.label        = "DND-shell";
+  rpcInit.label        = "SHELL";
   rpcInit.numOfThreads = numOfThreads;
   rpcInit.cfp          = dnodeProcessMsgFromShell;
   rpcInit.sessions     = TSDB_SESSIONS_PER_DNODE;
@@ -96,13 +98,11 @@ void dnodeProcessMsgFromShell(SRpcMsg *pMsg) {
   if ( dnodeProcessShellMsgFp[pMsg->msgType] ) {
     (*dnodeProcessShellMsgFp[pMsg->msgType])(pMsg);
   } else {
-    dError("RPC %p, msg:%s from shell is not handled", pMsg->handle, taosMsg[pMsg->msgType]);
+    mgmtProcessMsgFromShell(pMsg);
-    rpcMsg.code = TSDB_CODE_MSG_NOT_PROCESSED;
-    rpcSendResponse(&rpcMsg);
-    rpcFreeCont(pMsg->pCont);
   }
 }
 
+
 static int dnodeRetrieveUserAuthInfo(char *user, char *spi, char *encrypt, char *secret, char *ckey) {
   return TSDB_CODE_SUCCESS;
 }

src/inc/mnode.h

@@ -26,6 +26,8 @@ void    mgmtCleanUpSystem();
 void    mgmtStopSystem();
 void    sdbUpdateSync();
 
+void    mgmtProcessMsgFromShell(SRpcMsg *rpcMsg);
+
 #ifdef __cplusplus
 }
 #endif

src/inc/taosdef.h

@@ -329,8 +329,7 @@ void tsDataSwap(void *pLeft, void *pRight, int32_t type, int32_t size);
 #define TSDB_MAX_NORMAL_TABLES 1000
 #define TSDB_MAX_CHILD_TABLES  100000
 
-#define TSDB_PORT_MNODESHELL 0
+#define TSDB_PORT_DNODESHELL 0 
-#define TSDB_PORT_DNODESHELL 5
 #define TSDB_PORT_DNODEMNODE 10
 #define TSDB_PORT_MNODEDNODE 15 
 #define TSDB_PORT_SYNC 20

src/mnode/src/mgmtShell.c

@@ -41,9 +41,9 @@
 typedef int32_t (*SShowMetaFp)(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn);
 typedef int32_t (*SShowRetrieveFp)(SShowObj *pShow, char *data, int32_t rows, void *pConn);
 
-static int  mgmtShellRetriveAuth(char *user, char *spi, char *encrypt, char *secret, char *ckey);
+//static int  mgmtShellRetriveAuth(char *user, char *spi, char *encrypt, char *secret, char *ckey);
 static bool mgmtCheckMsgReadOnly(SQueuedMsg *pMsg);
-static void mgmtProcessMsgFromShell(SRpcMsg *pMsg);
+//static void mgmtProcessMsgFromShell(SRpcMsg *pMsg);
 static void mgmtProcessUnSupportMsg(SRpcMsg *rpcMsg);
 static void mgmtProcessShowMsg(SQueuedMsg *queuedMsg);
 static void mgmtProcessRetrieveMsg(SQueuedMsg *queuedMsg);
@@ -52,7 +52,7 @@ static void mgmtProcessConnectMsg(SQueuedMsg *queuedMsg);
 static void mgmtProcessUseMsg(SQueuedMsg *queuedMsg);
 
 void *tsMgmtTmr;
-static void *tsMgmtShellRpc = NULL;
+//static void *tsMgmtShellRpc = NULL;
 static void *tsMgmtTranQhandle = NULL;
 static void (*tsMgmtProcessShellMsgFp[TSDB_MSG_TYPE_MAX])(SQueuedMsg *) = {0};
 static void *tsQhandleCache = NULL;
@@ -70,28 +70,6 @@ int32_t mgmtInitShell() {
   tsMgmtTranQhandle = taosInitScheduler(tsMaxShellConns, 1, "mnodeT");
   tsQhandleCache = taosCacheInit(tsMgmtTmr, 2);
 
-  int32_t numOfThreads = tsNumOfCores * tsNumOfThreadsPerCore / 4.0;
-  if (numOfThreads < 1) {
-    numOfThreads = 1;
-  }
-
-  SRpcInit rpcInit = {0};
-  rpcInit.localPort    = tsMnodeShellPort;
-  rpcInit.label        = "MND-shell";
-  rpcInit.numOfThreads = numOfThreads;
-  rpcInit.cfp          = mgmtProcessMsgFromShell;
-  rpcInit.sessions     = tsMaxShellConns;
-  rpcInit.connType     = TAOS_CONN_SERVER;
-  rpcInit.idleTime     = tsShellActivityTimer * 1000;
-  rpcInit.afp          = mgmtShellRetriveAuth;
-
-  tsMgmtShellRpc = rpcOpen(&rpcInit);
-  if (tsMgmtShellRpc == NULL) {
-    mError("failed to init server connection to shell");
-    return -1;
-  }
-
-  mPrint("server connection to shell is opened");
   return 0;
 }
 
@@ -101,12 +79,6 @@ void mgmtCleanUpShell() {
     tsMgmtTranQhandle = NULL;
   }
 
-  if (tsMgmtShellRpc) {
-    rpcClose(tsMgmtShellRpc);
-    tsMgmtShellRpc = NULL;
-    mPrint("server connection to shell is closed");
-  }
-
   if (tsQhandleCache) {
     taosCacheEmpty(tsQhandleCache);
     taosCacheCleanup(tsQhandleCache);
@@ -148,7 +120,7 @@ void mgmtDealyedAddToShellQueue(SQueuedMsg *queuedMsg) {
   taosTmrReset(mgmtDoDealyedAddToShellQueue, 1000, queuedMsg, tsMgmtTmr, &unUsed);
 }
 
-static void mgmtProcessMsgFromShell(SRpcMsg *rpcMsg) {
+void mgmtProcessMsgFromShell(SRpcMsg *rpcMsg) {
   assert(rpcMsg);
 
   if (rpcMsg->pCont == NULL) {
@@ -370,6 +342,7 @@ static void mgmtProcessHeartBeatMsg(SQueuedMsg *pMsg) {
   rpcSendResponse(&rpcRsp);
 }
 
+/*
 static int mgmtShellRetriveAuth(char *user, char *spi, char *encrypt, char *secret, char *ckey) {
   *spi = 1;
   *encrypt = 0;
@@ -390,6 +363,7 @@ static int mgmtShellRetriveAuth(char *user, char *spi, char *encrypt, char *secr
     return TSDB_CODE_SUCCESS;
   }
 }
+*/
 
 static void mgmtProcessConnectMsg(SQueuedMsg *pMsg) {
   SRpcMsg rpcRsp = {.handle = pMsg->thandle, .pCont = NULL, .contLen = 0, .code = 0, .msgType = 0};