modify transport

include/libs/transport/trpc.h

@@ -64,7 +64,6 @@ typedef struct SRpcInit {
   int8_t   connType;      // TAOS_CONN_UDP, TAOS_CONN_TCPC, TAOS_CONN_TCPS
   int      idleTime;      // milliseconds, 0 means idle timer is disabled
 
-  bool noPool;  //  create conn pool or not
   // the following is for client app ecurity only
   char *user;     // user name
   char  spi;      // security parameter index

source/client/src/clientEnv.c

@@ -72,8 +72,6 @@ static void deregisterRequest(SRequestObj *pRequest) {
   taosReleaseRef(clientConnRefPool, pTscObj->id);
 }
 
-
-
 // todo close the transporter properly
 void closeTransporter(STscObj *pTscObj) {
   if (pTscObj == NULL || pTscObj->pAppInfo->pTransporter == NULL) {
@@ -241,6 +239,7 @@ void taos_init_imp(void) {
   clientConnRefPool = taosOpenRef(200, destroyTscObj);
   clientReqRefPool = taosOpenRef(40960, doDestroyRequest);
 
+  // transDestroyBuffer(&conn->readBuf);
   taosGetAppName(appInfo.appName, NULL);
   pthread_mutex_init(&appInfo.mutex, NULL);
 

source/dnode/mgmt/impl/src/dndTransport.c

@@ -25,8 +25,8 @@
 #include "dndMnode.h"
 #include "dndVnodes.h"
 
-#define INTERNAL_USER   "_dnd"
-#define INTERNAL_CKEY   "_key"
+#define INTERNAL_USER "_dnd"
+#define INTERNAL_CKEY "_key"
 #define INTERNAL_SECRET "_pwd"
 
 static void dndInitMsgFp(STransMgmt *pMgmt) {
@@ -155,7 +155,7 @@ static void dndInitMsgFp(STransMgmt *pMgmt) {
 }
 
 static void dndProcessResponse(void *parent, SRpcMsg *pRsp, SEpSet *pEpSet) {
-  SDnode     *pDnode = parent;
+  SDnode *    pDnode = parent;
   STransMgmt *pMgmt = &pDnode->tmgmt;
 
   tmsg_t msgType = pRsp->msgType;
@@ -193,7 +193,6 @@ static int32_t dndInitClient(SDnode *pDnode) {
   rpcInit.ckey = INTERNAL_CKEY;
   rpcInit.spi = 1;
   rpcInit.parent = pDnode;
-  rpcInit.noPool = true;
 
   char pass[TSDB_PASSWORD_LEN + 1] = {0};
   taosEncryptPass_c((uint8_t *)(INTERNAL_SECRET), strlen(INTERNAL_SECRET), pass);
@@ -219,7 +218,7 @@ static void dndCleanupClient(SDnode *pDnode) {
 }
 
 static void dndProcessRequest(void *param, SRpcMsg *pReq, SEpSet *pEpSet) {
-  SDnode     *pDnode = param;
+  SDnode *    pDnode = param;
   STransMgmt *pMgmt = &pDnode->tmgmt;
 
   tmsg_t msgType = pReq->msgType;
@@ -313,7 +312,7 @@ static int32_t dndRetrieveUserAuthInfo(void *parent, char *user, char *spi, char
   SAuthReq authReq = {0};
   tstrncpy(authReq.user, user, TSDB_USER_LEN);
   int32_t contLen = tSerializeSAuthReq(NULL, 0, &authReq);
-  void   *pReq = rpcMallocCont(contLen);
+  void *  pReq = rpcMallocCont(contLen);
   tSerializeSAuthReq(pReq, contLen, &authReq);
 
   SRpcMsg rpcMsg = {.pCont = pReq, .contLen = contLen, .msgType = TDMT_MND_AUTH, .ahandle = (void *)9528};

source/libs/transport/inc/transportInt.h

@@ -54,7 +54,6 @@ typedef struct {
   int8_t   connType;
   int64_t  index;
   char     label[TSDB_LABEL_LEN];
-  bool     noPool;  // pool or not
 
   char user[TSDB_UNI_LEN];         // meter ID
   char spi;                        // security parameter index

source/libs/transport/src/rpcMain.c

@@ -64,7 +64,6 @@ typedef struct {
   void (*cfp)(void *parent, SRpcMsg *, SEpSet *);
   int (*afp)(void *parent, char *user, char *spi, char *encrypt, char *secret, char *ckey);
 
-  bool             noPool;
   int32_t          refCount;
   void *           parent;
   void *           idPool;     // handle to ID pool

source/libs/transport/src/trans.c

@@ -41,7 +41,6 @@ void* rpcOpen(const SRpcInit* pInit) {
     pRpc->numOfThreads = pInit->numOfThreads;
   }
 
-  pRpc->noPool = pInit->noPool;
   pRpc->connType = pInit->connType;
   pRpc->idleTime = pInit->idleTime;
   pRpc->tcphandle = (*taosInitHandle[pRpc->connType])(0, pInit->localPort, pRpc->label, pRpc->numOfThreads, NULL, pRpc);

source/libs/transport/src/transCli.c

@@ -170,13 +170,7 @@ static void  clientHandleResp(SCliConn* conn) {
 
   // user owns conn->persist = 1
   if (conn->persist == 0) {
-    if (pTransInst->noPool == true) {
-      destroyCmsg(conn->data);
-      clientConnDestroy(conn, true);
-      return;
-    } else {
-      addConnToPool(pThrd->pool, pCtx->ip, pCtx->port, conn);
-    }
+    addConnToPool(pThrd->pool, pCtx->ip, pCtx->port, conn);
   }
   destroyCmsg(conn->data);
   conn->data = NULL;
@@ -463,10 +457,8 @@ static void clientHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd) {
       tTrace("%s client conn %p reused", CONN_GET_INST_LABEL(conn), conn);
     }
   } else {
-    if (pTransInst->noPool == false) {
-      conn = getConnFromPool(pThrd->pool, pCtx->ip, pCtx->port);
-      if (conn != NULL) tTrace("%s client conn %p get from conn pool", CONN_GET_INST_LABEL(conn), conn);
-    }
+    conn = getConnFromPool(pThrd->pool, pCtx->ip, pCtx->port);
+    if (conn != NULL) tTrace("%s client conn %p get from conn pool", CONN_GET_INST_LABEL(conn), conn);
   }
 
   if (conn != NULL) {