From ddc13d294cef52d81451604498fe8d87accbcedc Mon Sep 17 00:00:00 2001 From: yihaoDeng Date: Sat, 2 Jan 2021 02:57:58 +0000 Subject: [PATCH] share rpc obj --- src/client/inc/tsclient.h | 10 +++- src/client/src/tscSql.c | 16 +++--- src/client/src/tscSystem.c | 106 ++++++++++++++++++++++++++++--------- src/client/src/tscUtil.c | 7 +-- 4 files changed, 100 insertions(+), 39 deletions(-) diff --git a/src/client/inc/tsclient.h b/src/client/inc/tsclient.h index 6879470145..0f2f0513a3 100644 --- a/src/client/inc/tsclient.h +++ b/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); diff --git a/src/client/src/tscSql.c b/src/client/src/tscSql.c index fa7bc99a9f..e70db12a7b 100644 --- a/src/client/src/tscSql.c +++ b/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); diff --git a/src/client/src/tscSystem.c b/src/client/src/tscSystem.c index ff605dad72..4e23bd8a03 100644 --- a/src/client/src/tscSystem.c +++ b/src/client/src/tscSystem.c @@ -40,39 +40,81 @@ 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); } - -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.cfp = tscProcessMsgFromServer; - rpcInit.sessions = tsMaxConnections; - rpcInit.connType = TAOS_CONN_CLIENT; - rpcInit.user = (char *)user; - rpcInit.idleTime = 2000; - rpcInit.ckey = "key"; - rpcInit.spi = 1; - rpcInit.secret = (char *)secretEncrypt; - - *pDnodeConn = rpcOpen(&rpcInit); - if (*pDnodeConn == NULL) { - tscError("failed to init connection to TDengine"); - return -1; - } else { - tscDebug("dnodeConn:%p is created, user:%s", *pDnodeConn, user); - } +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 *)¶m, 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; + } + + SRpcInit rpcInit; + memset(&rpcInit, 0, sizeof(rpcInit)); + rpcInit.localPort = 0; + rpcInit.label = "TSC"; + rpcInit.numOfThreads = 1; + rpcInit.cfp = tscProcessMsgFromServer; + rpcInit.sessions = tsMaxConnections; + rpcInit.connType = TAOS_CONN_CLIENT; + rpcInit.user = (char *)user; + rpcInit.idleTime = 2000; + rpcInit.ckey = "key"; + rpcInit.spi = 1; + rpcInit.secret = (char *)secretEncrypt; + + SRpcObj rpcObj; + 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; + } + 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; diff --git a/src/client/src/tscUtil.c b/src/client/src/tscUtil.c index 52e26fe95a..7936c31634 100644 --- a/src/client/src/tscUtil.c +++ b/src/client/src/tscUtil.c @@ -895,15 +895,10 @@ void tscCloseTscObj(void *param) { pObj->signature = NULL; taosTmrStopA(&(pObj->pTimer)); - void* p = pObj->pDnodeConn; - if (pObj->pDnodeConn != NULL) { - rpcClose(pObj->pDnodeConn); - pObj->pDnodeConn = NULL; - } + tscReleaseRpc(pObj->pRpcObj); tfree(pObj->tscCorMgmtEpSet); pthread_mutex_destroy(&pObj->mutex); - tscDebug("%p DB connection is closed, dnodeConn:%p", pObj, p); tfree(pObj); }