share rpc obj

src/client/inc/tsclient.h

@@ -307,6 +307,7 @@ typedef struct STscObj {
   struct SSqlStream *streamList;
   SRpcCorEpSet       *tscCorMgmtEpSet;
   void*              pDnodeConn;
+  void*              pRpcObj;
   pthread_mutex_t    mutex;
   int32_t            numOfObj; // number of sqlObj from this tscObj
 } STscObj;
@@ -377,7 +378,14 @@ typedef struct SSqlStream {
 
 void tscSetStreamDestTable(SSqlStream* pStream, const char* dstTable);
 
-int32_t tscInitRpc(const char *user, const char *secret, void** pDnodeConn);
+typedef struct {
+  char    key[512]; 
+  void    *pDnodeConn; 
+} SRpcObj;
+
+void *tscAcquireRpc(const char *insKey);
+void tscReleaseRpc(void *param);
+int32_t tscInitRpc(const char *key, const char *user, const char *secret, void **pRpcObj, void** pDnodeConn);
 void    tscInitMsgsFp();
 
 int tsParseSql(SSqlObj *pSql, bool initial);

src/client/src/tscSql.c

@@ -90,9 +90,12 @@ static SSqlObj *taosConnectImpl(const char *ip, const char *user, const char *pa
   } else {
     if (tscSetMgmtEpSetFromCfg(tsFirst, tsSecond, &corMgmtEpSet) < 0) return NULL;
   }
+  char rpcKey[512] = {0};
+  sprintf(rpcKey, "%s:%s:%s:%d", user, pass, ip, port);
  
+  void *pRpcObj = NULL;
   void *pDnodeConn = NULL;
-  if (tscInitRpc(user, secretEncrypt, &pDnodeConn) != 0) {
+  if (tscInitRpc(rpcKey, user, secretEncrypt, &pRpcObj, &pDnodeConn) != 0) {
     terrno = TSDB_CODE_RPC_NETWORK_UNAVAIL;
     return NULL;
   }
@@ -100,20 +103,21 @@ static SSqlObj *taosConnectImpl(const char *ip, const char *user, const char *pa
   STscObj *pObj = (STscObj *)calloc(1, sizeof(STscObj));
   if (NULL == pObj) {
     terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
-    rpcClose(pDnodeConn);
+    tscReleaseRpc(pRpcObj);
     return NULL;
   }
   // set up tscObj's mgmtEpSet
   pObj->tscCorMgmtEpSet = (SRpcCorEpSet *)malloc(sizeof(SRpcCorEpSet));
   if (NULL == pObj->tscCorMgmtEpSet) {
     terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
-    rpcClose(pDnodeConn);
+    tscReleaseRpc(pRpcObj);
     free(pObj->tscCorMgmtEpSet);
     free(pObj);
   }
   memcpy(pObj->tscCorMgmtEpSet, &corMgmtEpSet, sizeof(SRpcCorEpSet));
 
   pObj->signature = pObj;
+  pObj->pRpcObj = pRpcObj;
   pObj->pDnodeConn = pDnodeConn;
   
   tstrncpy(pObj->user, user, sizeof(pObj->user));
@@ -125,7 +129,7 @@ static SSqlObj *taosConnectImpl(const char *ip, const char *user, const char *pa
     /* db name is too long */
     if (len >= TSDB_DB_NAME_LEN) {
       terrno = TSDB_CODE_TSC_INVALID_DB_LENGTH;
-      rpcClose(pDnodeConn);
+      tscReleaseRpc(pRpcObj);
       free(pObj->tscCorMgmtEpSet);
       free(pObj);
       return NULL;
@@ -143,7 +147,7 @@ static SSqlObj *taosConnectImpl(const char *ip, const char *user, const char *pa
   SSqlObj *pSql = (SSqlObj *)calloc(1, sizeof(SSqlObj));
   if (NULL == pSql) {
     terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
-    rpcClose(pDnodeConn);
+    tscReleaseRpc(pRpcObj);
     free(pObj->tscCorMgmtEpSet);
     free(pObj);
     return NULL;
@@ -160,7 +164,7 @@ static SSqlObj *taosConnectImpl(const char *ip, const char *user, const char *pa
 
   if (TSDB_CODE_SUCCESS != tscAllocPayload(&pSql->cmd, TSDB_DEFAULT_PAYLOAD_SIZE)) {
     terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
-    rpcClose(pDnodeConn);
+    tscReleaseRpc(pRpcObj);
     free(pSql);
     free(pObj->tscCorMgmtEpSet);
     free(pObj);

src/client/src/tscSystem.c

@@ -40,21 +40,53 @@ void   *tscCheckDiskUsageTmr;
 int     tscRefId = -1;
 int     tscNumOfObj = 0;       // number of sqlObj in current process.
 
+int32_t tscNumOfThreads = 1;
+void *  tscRpcCache;
+static pthread_mutex_t rpcObjMutex;
 static pthread_once_t tscinit = PTHREAD_ONCE_INIT;
 
 void tscCheckDiskUsage(void *UNUSED_PARAM(para), void* UNUSED_PARAM(param)) {
   taosGetDisk();
   taosTmrReset(tscCheckDiskUsage, 1000, NULL, tscTmr, &tscCheckDiskUsageTmr);
 }
+void tscFreeRpcObj(void *param) {
+  assert(param);
+  SRpcObj *pRpcObj = (SRpcObj *)(param);
+  rpcClose(pRpcObj->pDnodeConn);
+}
+void *tscAcquireRpc(const char *key) {
+  SRpcObj *pRpcObj = taosCacheAcquireByKey(tscRpcCache, key, strlen(key)); 
+  if (pRpcObj == NULL) {
+    return NULL;
+  }
+  return pRpcObj; 
+}
+
+void tscReleaseRpc(void *param)  {
+  if (param == NULL) {
+    return;
+  }
+  pthread_mutex_lock(&rpcObjMutex);
+  taosCacheRelease(tscRpcCache, (void *)&param, false); 
+  pthread_mutex_unlock(&rpcObjMutex);
+} 
+
+int32_t tscInitRpc(const char *key, const char *user, const char *secretEncrypt, void **ppRpcObj, void **pDnodeConn) {
+  pthread_mutex_lock(&rpcObjMutex);
+
+  SRpcObj *pRpcObj = (SRpcObj *)tscAcquireRpc(key); 
+  if (pRpcObj != NULL) {
+    *ppRpcObj = pRpcObj;   
+    *pDnodeConn = pRpcObj->pDnodeConn; 
+    pthread_mutex_unlock(&rpcObjMutex);
+    return 0;
+  } 
 
-int32_t tscInitRpc(const char *user, const char *secretEncrypt, void **pDnodeConn) {
   SRpcInit rpcInit;
-
-  if (*pDnodeConn == NULL) {
   memset(&rpcInit, 0, sizeof(rpcInit));
   rpcInit.localPort = 0;
   rpcInit.label = "TSC";
-    rpcInit.numOfThreads = 1;  // every DB connection has only one thread
+  rpcInit.numOfThreads = 1;    
   rpcInit.cfp = tscProcessMsgFromServer;
   rpcInit.sessions = tsMaxConnections;
   rpcInit.connType = TAOS_CONN_CLIENT;
@@ -64,15 +96,25 @@ int32_t tscInitRpc(const char *user, const char *secretEncrypt, void **pDnodeCon
   rpcInit.spi = 1; 
   rpcInit.secret = (char *)secretEncrypt;
 
-    *pDnodeConn = rpcOpen(&rpcInit);
+  SRpcObj rpcObj;
-    if (*pDnodeConn == NULL) {
+  memset(&rpcObj, 0, sizeof(rpcObj));
+  strncpy(rpcObj.key, key, strlen(key));
+  rpcObj.pDnodeConn = rpcOpen(&rpcInit);
+  if (rpcObj.pDnodeConn == NULL) {
+    pthread_mutex_unlock(&rpcObjMutex);
     tscError("failed to init connection to TDengine");
     return -1;
-    } else {
-      tscDebug("dnodeConn:%p is created, user:%s", *pDnodeConn, user);
   } 
+  pRpcObj = taosCachePut(tscRpcCache, rpcObj.key, strlen(rpcObj.key), &rpcObj, sizeof(rpcObj), 1000*10);   
+  if (pRpcObj == NULL) {
+    rpcClose(rpcObj.pDnodeConn);
+    pthread_mutex_unlock(&rpcObjMutex);
+    return -1;
   } 
 
+  *ppRpcObj  = pRpcObj;
+  *pDnodeConn = pRpcObj->pDnodeConn;
+  pthread_mutex_unlock(&rpcObjMutex);
   return 0;
 }
 
@@ -113,7 +155,7 @@ void taos_init_imp(void) {
   int queueSize = tsMaxConnections*2;
 
   double factor = (tscEmbedded == 0)? 2.0:4.0;
-  int32_t tscNumOfThreads = (int)(tsNumOfCores * tsNumOfThreadsPerCore / factor);
+  tscNumOfThreads = (int)(tsNumOfCores * tsNumOfThreadsPerCore / factor);
   if (tscNumOfThreads < 2) {
     tscNumOfThreads = 2;
   }
@@ -135,6 +177,8 @@ void taos_init_imp(void) {
     tscObjRef  = taosOpenRef(40960, tscFreeRegisteredSqlObj);
     tscHashMap = taosHashInit(1024, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK);
   }
+  tscRpcCache = taosCacheInit(TSDB_DATA_TYPE_BINARY, refreshTime, true, tscFreeRpcObj, "rpcObj");
+  pthread_mutex_init(&rpcObjMutex, NULL);
 
   tscRefId = taosOpenRef(200, tscCloseTscObj);
 
@@ -169,6 +213,16 @@ void taos_cleanup(void) {
   taosCloseRef(tscRefId);
   taosCleanupKeywordsTable();
   taosCloseLog();
+
+  m = tscRpcCache; 
+  if (m != NULL && atomic_val_compare_exchange_ptr(&tscRpcCache, m, 0) == m) {
+    pthread_mutex_lock(&rpcObjMutex);
+    taosCacheCleanup(tscRpcCache); 
+    tscRpcCache = NULL;
+    pthread_mutex_unlock(&rpcObjMutex);
+    pthread_mutex_destroy(&rpcObjMutex);
+  }
+
   if (tscEmbedded == 0) rpcCleanup();
 
   m = tscTmr;

src/client/src/tscUtil.c

@@ -895,15 +895,10 @@ void tscCloseTscObj(void *param) {
   pObj->signature = NULL;
   taosTmrStopA(&(pObj->pTimer));
 
-  void* p = pObj->pDnodeConn;
+  tscReleaseRpc(pObj->pRpcObj);
-  if (pObj->pDnodeConn != NULL) {
-    rpcClose(pObj->pDnodeConn);
-    pObj->pDnodeConn = NULL;
-  }
   tfree(pObj->tscCorMgmtEpSet);
   pthread_mutex_destroy(&pObj->mutex);
 
-  tscDebug("%p DB connection is closed, dnodeConn:%p", pObj, p);
   tfree(pObj);
 }