diff --git a/src/client/inc/tsclient.h b/src/client/inc/tsclient.h index 68cd2747e9ee39ce21de87964cf26ad19d14af66..c9702ad1fc7e83e1d7473ee040bafc33e6277c87 100644 --- a/src/client/inc/tsclient.h +++ b/src/client/inc/tsclient.h @@ -297,6 +297,11 @@ typedef struct { struct SLocalMerger *pLocalMerger; } SSqlRes; +typedef struct { + char key[512]; + void *pDnodeConn; +} SRpcObj; + typedef struct STscObj { void * signature; void * pTimer; @@ -312,8 +317,8 @@ typedef struct STscObj { int64_t hbrid; struct SSqlObj * sqlList; struct SSqlStream *streamList; - SRpcCorEpSet *tscCorMgmtEpSet; - void* pDnodeConn; + SRpcObj *pRpcObj; + SRpcCorEpSet *tscCorMgmtEpSet; pthread_mutex_t mutex; int32_t numOfObj; // number of sqlObj from this tscObj } STscObj; @@ -390,8 +395,10 @@ typedef struct SSqlStream { void tscSetStreamDestTable(SSqlStream* pStream, const char* dstTable); -int32_t tscInitRpc(const char *user, const char *secret, void** pDnodeConn); -void tscInitMsgsFp(); + +int tscAcquireRpc(const char *key, const char *user, const char *secret,void **pRpcObj); +void tscReleaseRpc(void *param); +void tscInitMsgsFp(); int tsParseSql(SSqlObj *pSql, bool initial); diff --git a/src/client/src/tscServer.c b/src/client/src/tscServer.c index 6035a270c5494560744b12cfa77c7449ca2fcb06..a81e47f3ad6f0d040114961ad5030067df6624ee 100644 --- a/src/client/src/tscServer.c +++ b/src/client/src/tscServer.c @@ -157,13 +157,16 @@ void tscProcessHeartBeatRsp(void *param, TAOS_RES *tres, int code) { SRpcEpSet *epSet = &pRsp->epSet; if (epSet->numOfEps > 0) { tscEpSetHtons(epSet); - if (!tscEpSetIsEqual(&pSql->pTscObj->tscCorMgmtEpSet->epSet, epSet)) { - tscTrace("%p updating epset: numOfEps: %d, inUse: %d", pSql, epSet->numOfEps, epSet->inUse); - for (int8_t i = 0; i < epSet->numOfEps; i++) { - tscTrace("endpoint %d: fqdn=%s, port=%d", i, epSet->fqdn[i], epSet->port[i]); - } - tscUpdateMgmtEpSet(pSql, epSet); - } + + //SRpcCorEpSet *pCorEpSet = pSql->pTscObj->tscCorMgmtEpSet; + //if (!tscEpSetIsEqual(&pCorEpSet->epSet, epSet)) { + // tscTrace("%p updating epset: numOfEps: %d, inUse: %d", pSql, epSet->numOfEps, epSet->inUse); + // for (int8_t i = 0; i < epSet->numOfEps; i++) { + // tscTrace("endpoint %d: fqdn=%s, port=%d", i, epSet->fqdn[i], epSet->port[i]); + // } + //} + //concurrency problem, update mgmt epset anyway + tscUpdateMgmtEpSet(pSql, epSet); } pSql->pTscObj->connId = htonl(pRsp->connId); @@ -270,7 +273,8 @@ int tscSendMsgToServer(SSqlObj *pSql) { .code = 0 }; - rpcSendRequest(pObj->pDnodeConn, &pSql->epSet, &rpcMsg, &pSql->rpcRid); + + rpcSendRequest(pObj->pRpcObj->pDnodeConn, &pSql->epSet, &rpcMsg, &pSql->rpcRid); return TSDB_CODE_SUCCESS; } diff --git a/src/client/src/tscSql.c b/src/client/src/tscSql.c index 36146a49b9a2f1b9e59a1cc69ebeda4ef983ef64..448eea16bf5a529f489b59ced9a9f32c513338eb 100644 --- a/src/client/src/tscSql.c +++ b/src/client/src/tscSql.c @@ -90,9 +90,11 @@ 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}; + snprintf(rpcKey, sizeof(rpcKey), "%s:%s:%s:%d", user, pass, ip, port); - void *pDnodeConn = NULL; - if (tscInitRpc(user, secretEncrypt, &pDnodeConn) != 0) { + void *pRpcObj = NULL; + if (tscAcquireRpc(rpcKey, user, secretEncrypt, &pRpcObj) != 0) { terrno = TSDB_CODE_RPC_NETWORK_UNAVAIL; return NULL; } @@ -100,23 +102,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) { + + pObj->tscCorMgmtEpSet = malloc(sizeof(SRpcCorEpSet)); + if (pObj->tscCorMgmtEpSet == NULL) { terrno = TSDB_CODE_TSC_OUT_OF_MEMORY; - rpcClose(pDnodeConn); - free(pObj->tscCorMgmtEpSet); + tscReleaseRpc(pRpcObj); free(pObj); return NULL; } - memcpy(pObj->tscCorMgmtEpSet, &corMgmtEpSet, sizeof(SRpcCorEpSet)); - - pObj->signature = pObj; - pObj->pDnodeConn = pDnodeConn; + memcpy(pObj->tscCorMgmtEpSet, &corMgmtEpSet, sizeof(corMgmtEpSet)); + pObj->signature = pObj; + pObj->pRpcObj = (SRpcObj *)pRpcObj; tstrncpy(pObj->user, user, sizeof(pObj->user)); secretEncryptLen = MIN(secretEncryptLen, sizeof(pObj->pass)); memcpy(pObj->pass, secretEncrypt, secretEncryptLen); @@ -126,8 +126,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); - free(pObj->tscCorMgmtEpSet); + tscReleaseRpc(pRpcObj); free(pObj); return NULL; } @@ -144,8 +143,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); - free(pObj->tscCorMgmtEpSet); + tscReleaseRpc(pRpcObj); free(pObj); return NULL; } @@ -161,9 +159,8 @@ 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); return NULL; } @@ -202,7 +199,7 @@ TAOS *taos_connect_internal(const char *ip, const char *user, const char *pass, return NULL; } - tscDebug("%p DB connection is opening, dnodeConn:%p", pObj, pObj->pDnodeConn); + tscDebug("%p DB connection is opening, rpcObj: %p, dnodeConn:%p", pObj, pObj->pRpcObj, pObj->pRpcObj->pDnodeConn); taos_free_result(pSql); // version compare only requires the first 3 segments of the version string @@ -279,7 +276,7 @@ void taos_close(TAOS *taos) { return; } - tscDebug("%p try to free tscObj and close dnodeConn:%p", pObj, pObj->pDnodeConn); + tscDebug("%p try to free tscObj", pObj); if (pObj->signature != pObj) { tscDebug("%p already closed or invalid tscObj", pObj); return; @@ -303,7 +300,7 @@ void taos_close(TAOS *taos) { } } - tscDebug("%p all sqlObj are freed, free tscObj and close dnodeConn:%p", pObj, pObj->pDnodeConn); + tscDebug("%p all sqlObj are freed, free tscObj", pObj); taosRemoveRef(tscRefId, pObj->rid); } diff --git a/src/client/src/tscSystem.c b/src/client/src/tscSystem.c index dd0f248b85452342fb567357ba9346ab71754dfc..fe1c45bd393938cfd9a77db97169dde2ef6412de 100644 --- a/src/client/src/tscSystem.c +++ b/src/client/src/tscSystem.c @@ -43,41 +43,74 @@ void *tscTmr; void *tscQhandle; int32_t tscRefId = -1; int32_t tscNumOfObj = 0; // number of sqlObj in current process. - static void *tscCheckDiskUsageTmr; +void *tscRpcCache; // cache to keep rpc obj +int32_t tscNumOfThreads = 1; // num of rpc threads +static pthread_mutex_t rpcObjMutex; // mutex to protect open the rpc obj concurrently 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); + tscDebug("free rpcObj:%p and free pDnodeConn: %p", pRpcObj, pRpcObj->pDnodeConn); + rpcClose(pRpcObj->pDnodeConn); +} -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 tscReleaseRpc(void *param) { + if (param == NULL) { + return; } + pthread_mutex_lock(&rpcObjMutex); + taosCacheRelease(tscRpcCache, (void *)¶m, false); + pthread_mutex_unlock(&rpcObjMutex); +} +int32_t tscAcquireRpc(const char *key, const char *user, const char *secretEncrypt, void **ppRpcObj) { + pthread_mutex_lock(&rpcObjMutex); + + SRpcObj *pRpcObj = (SRpcObj *)taosCacheAcquireByKey(tscRpcCache, key, strlen(key)); + if (pRpcObj != NULL) { + *ppRpcObj = pRpcObj; + pthread_mutex_unlock(&rpcObjMutex); + return 0; + } + + SRpcInit rpcInit; + memset(&rpcInit, 0, sizeof(rpcInit)); + rpcInit.localPort = 0; + rpcInit.label = "TSC"; + rpcInit.numOfThreads = tscNumOfThreads * 2; + 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; + pthread_mutex_unlock(&rpcObjMutex); return 0; } @@ -118,7 +151,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; } @@ -140,6 +173,10 @@ void taos_init_imp(void) { tscTableMetaInfo = taosHashInit(1024, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK); tscDebug("TableMeta:%p", tscTableMetaInfo); } + + int refreshTime = 5; + tscRpcCache = taosCacheInit(TSDB_DATA_TYPE_BINARY, refreshTime, true, tscFreeRpcObj, "rpcObj"); + pthread_mutex_init(&rpcObjMutex, NULL); tscRefId = taosOpenRef(200, tscCloseTscObj); @@ -181,10 +218,16 @@ void taos_cleanup(void) { taosCleanupKeywordsTable(); taosCloseLog(); - if (tscEmbedded == 0) { - rpcCleanup(); + p = tscRpcCache; + tscRpcCache = NULL; + + if (p != NULL) { + taosCacheCleanup(p); + pthread_mutex_destroy(&rpcObjMutex); } + if (tscEmbedded == 0) rpcCleanup(); + p = tscTmr; tscTmr = NULL; taosTmrCleanUp(p); diff --git a/src/client/src/tscUtil.c b/src/client/src/tscUtil.c index 824fff574e2fd1160eaa2cef1d6edc0869ea2a67..0d7ac917aa45ac92f51d0844e165d306bfa8bb75 100644 --- a/src/client/src/tscUtil.c +++ b/src/client/src/tscUtil.c @@ -447,7 +447,6 @@ void tscFreeRegisteredSqlObj(void *pSql) { SSqlObj* p = *(SSqlObj**)pSql; STscObj* pTscObj = p->pTscObj; - assert(RID_VALID(p->self)); int32_t num = atomic_sub_fetch_32(&pTscObj->numOfObj, 1); @@ -898,16 +897,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; - } - tfree(pObj->tscCorMgmtEpSet); + tscReleaseRpc(pObj->pRpcObj); pthread_mutex_destroy(&pObj->mutex); - tscDebug("%p DB connection is closed, dnodeConn:%p", pObj, p); tfree(pObj); } diff --git a/src/rpc/src/rpcTcp.c b/src/rpc/src/rpcTcp.c index 4fe41b17099cadf37988716154f45c6dcccb3814..111b722f7666865e5d0c3a8b261bccc4030d025d 100644 --- a/src/rpc/src/rpcTcp.c +++ b/src/rpc/src/rpcTcp.c @@ -48,6 +48,13 @@ typedef struct SThreadObj { void *(*processData)(SRecvInfo *pPacket); } SThreadObj; +typedef struct { + char label[TSDB_LABEL_LEN]; + int32_t index; + int numOfThreads; + SThreadObj **pThreadObj; +} SClientObj; + typedef struct { SOCKET fd; uint32_t ip; @@ -116,6 +123,7 @@ void *taosInitTcpServer(uint32_t ip, uint16_t port, char *label, int numOfThread pThreadObj->processData = fp; tstrncpy(pThreadObj->label, label, sizeof(pThreadObj->label)); pThreadObj->shandle = shandle; + pThreadObj->stop = false; } // initialize mutex, thread, fd which may fail @@ -166,6 +174,7 @@ void *taosInitTcpServer(uint32_t ip, uint16_t port, char *label, int numOfThread } static void taosStopTcpThread(SThreadObj* pThreadObj) { + if (pThreadObj == NULL) { return;} // save thread into local variable and signal thread to stop pthread_t thread = pThreadObj->thread; if (!taosCheckPthreadValid(thread)) { @@ -282,49 +291,76 @@ static void *taosAcceptTcpConnection(void *arg) { return NULL; } -void *taosInitTcpClient(uint32_t ip, uint16_t port, char *label, int num, void *fp, void *shandle) { - SThreadObj *pThreadObj; - pthread_attr_t thattr; - - pThreadObj = (SThreadObj *)malloc(sizeof(SThreadObj)); - memset(pThreadObj, 0, sizeof(SThreadObj)); - tstrncpy(pThreadObj->label, label, sizeof(pThreadObj->label)); - pThreadObj->ip = ip; - pThreadObj->shandle = shandle; - - if (pthread_mutex_init(&(pThreadObj->mutex), NULL) < 0) { - tError("%s failed to init TCP client mutex(%s)", label, strerror(errno)); - free(pThreadObj); +void *taosInitTcpClient(uint32_t ip, uint16_t port, char *label, int numOfThreads, void *fp, void *shandle) { + SClientObj *pClientObj = (SClientObj *)calloc(1, sizeof(SClientObj)); + if (pClientObj == NULL) { + tError("TCP:%s no enough memory", label); terrno = TAOS_SYSTEM_ERROR(errno); return NULL; - } + } - pThreadObj->pollFd = (EpollFd)epoll_create(10); // size does not matter - if (pThreadObj->pollFd < 0) { - tError("%s failed to create TCP client epoll", label); - free(pThreadObj); + tstrncpy(pClientObj->label, label, sizeof(pClientObj->label)); + pClientObj->numOfThreads = numOfThreads; + pClientObj->pThreadObj = (SThreadObj **)calloc(numOfThreads, sizeof(SThreadObj*)); + if (pClientObj->pThreadObj == NULL) { + tError("TCP:%s no enough memory", label); + tfree(pClientObj); terrno = TAOS_SYSTEM_ERROR(errno); - return NULL; } - pThreadObj->processData = fp; - + int code = 0; + pthread_attr_t thattr; pthread_attr_init(&thattr); pthread_attr_setdetachstate(&thattr, PTHREAD_CREATE_JOINABLE); - int code = pthread_create(&(pThreadObj->thread), &thattr, taosProcessTcpData, (void *)(pThreadObj)); - pthread_attr_destroy(&thattr); - if (code != 0) { - EpollClose(pThreadObj->pollFd); - pThreadObj->pollFd = -1; - free(pThreadObj); - terrno = TAOS_SYSTEM_ERROR(errno); - tError("%s failed to create TCP read data thread(%s)", label, strerror(errno)); - return NULL; + + for (int i = 0; i < numOfThreads; ++i) { + SThreadObj *pThreadObj = (SThreadObj *)calloc(1, sizeof(SThreadObj)); + if (pThreadObj == NULL) { + tError("TCP:%s no enough memory", label); + terrno = TAOS_SYSTEM_ERROR(errno); + for (int j=0; jpThreadObj[j]); + free(pClientObj); + pthread_attr_destroy(&thattr); + return NULL; + } + pClientObj->pThreadObj[i] = pThreadObj; + taosResetPthread(&pThreadObj->thread); + pThreadObj->ip = ip; + pThreadObj->stop = false; + tstrncpy(pThreadObj->label, label, sizeof(pThreadObj->label)); + pThreadObj->shandle = shandle; + pThreadObj->processData = fp; } - tDebug("%s TCP client is initialized, ip:%u:%hu", label, ip, port); + // initialize mutex, thread, fd which may fail + for (int i = 0; i < numOfThreads; ++i) { + SThreadObj *pThreadObj = pClientObj->pThreadObj[i]; + code = pthread_mutex_init(&(pThreadObj->mutex), NULL); + if (code < 0) { + tError("%s failed to init TCP process data mutex(%s)", label, strerror(errno)); + break; + } - return pThreadObj; + pThreadObj->pollFd = (int64_t)epoll_create(10); // size does not matter + if (pThreadObj->pollFd < 0) { + tError("%s failed to create TCP epoll", label); + code = -1; + break; + } + + code = pthread_create(&(pThreadObj->thread), &thattr, taosProcessTcpData, (void *)(pThreadObj)); + if (code != 0) { + tError("%s failed to create TCP process data thread(%s)", label, strerror(errno)); + break; + } + pThreadObj->threadId = i; + } + if (code != 0) { + terrno = TAOS_SYSTEM_ERROR(errno); + taosCleanUpTcpClient(pClientObj); + pClientObj = NULL; + } + return pClientObj; } void taosStopTcpClient(void *chandle) { @@ -335,15 +371,23 @@ void taosStopTcpClient(void *chandle) { } void taosCleanUpTcpClient(void *chandle) { - SThreadObj *pThreadObj = chandle; - if (pThreadObj == NULL) return; - - tDebug ("%s TCP client will be cleaned up", pThreadObj->label); - taosStopTcpThread(pThreadObj); + SClientObj *pClientObj = chandle; + if (pClientObj == NULL) return; + for (int i = 0; i < pClientObj->numOfThreads; ++i) { + SThreadObj *pThreadObj= pClientObj->pThreadObj[i]; + taosStopTcpThread(pThreadObj); + } + + tDebug("%s TCP client is cleaned up", pClientObj->label); + tfree(pClientObj->pThreadObj); + tfree(pClientObj); } void *taosOpenTcpClientConnection(void *shandle, void *thandle, uint32_t ip, uint16_t port) { - SThreadObj * pThreadObj = shandle; + SClientObj * pClientObj = shandle; + int32_t index = atomic_load_32(&pClientObj->index) % pClientObj->numOfThreads; + atomic_store_32(&pClientObj->index, index + 1); + SThreadObj *pThreadObj = pClientObj->pThreadObj[index]; SOCKET fd = taosOpenTcpClientSocket(ip, port, pThreadObj->ip); if (fd <= 0) return NULL;