diff --git a/src/client/inc/tsclient.h b/src/client/inc/tsclient.h index e8b03c0db9665ad9ddd8d92af9f2237b9dc7e7cf..c9702ad1fc7e83e1d7473ee040bafc33e6277c87 100644 --- a/src/client/inc/tsclient.h +++ b/src/client/inc/tsclient.h @@ -299,7 +299,6 @@ typedef struct { typedef struct { char key[512]; - SRpcCorEpSet *tscCorMgmtEpSet; void *pDnodeConn; } SRpcObj; @@ -319,6 +318,7 @@ typedef struct STscObj { struct SSqlObj * sqlList; struct SSqlStream *streamList; SRpcObj *pRpcObj; + SRpcCorEpSet *tscCorMgmtEpSet; pthread_mutex_t mutex; int32_t numOfObj; // number of sqlObj from this tscObj } STscObj; @@ -396,7 +396,7 @@ typedef struct SSqlStream { void tscSetStreamDestTable(SSqlStream* pStream, const char* dstTable); -int tscAcquireRpc(const char *key, const char *user, const char *secret, SRpcCorEpSet *corMgmtEpSet, void **pRpcObj); +int tscAcquireRpc(const char *key, const char *user, const char *secret,void **pRpcObj); void tscReleaseRpc(void *param); void tscInitMsgsFp(); diff --git a/src/client/src/tscServer.c b/src/client/src/tscServer.c index 381ee159124c11c9e4f3bfd3312e60320ce67a05..a81e47f3ad6f0d040114961ad5030067df6624ee 100644 --- a/src/client/src/tscServer.c +++ b/src/client/src/tscServer.c @@ -69,7 +69,7 @@ static void tscSetDnodeEpSet(SRpcEpSet* pEpSet, SVgroupInfo* pVgroupInfo) { } static void tscDumpMgmtEpSet(SSqlObj *pSql) { - SRpcCorEpSet *pCorEpSet = pSql->pTscObj->pRpcObj->tscCorMgmtEpSet; + SRpcCorEpSet *pCorEpSet = pSql->pTscObj->tscCorMgmtEpSet; taosCorBeginRead(&pCorEpSet->version); pSql->epSet = pCorEpSet->epSet; taosCorEndRead(&pCorEpSet->version); @@ -94,7 +94,7 @@ bool tscEpSetIsEqual(SRpcEpSet *s1, SRpcEpSet *s2) { void tscUpdateMgmtEpSet(SSqlObj *pSql, SRpcEpSet *pEpSet) { // no need to update if equal - SRpcCorEpSet *pCorEpSet = pSql->pTscObj->pRpcObj->tscCorMgmtEpSet; + SRpcCorEpSet *pCorEpSet = pSql->pTscObj->tscCorMgmtEpSet; taosCorBeginWrite(&pCorEpSet->version); pCorEpSet->epSet = *pEpSet; taosCorEndWrite(&pCorEpSet->version); @@ -158,7 +158,7 @@ void tscProcessHeartBeatRsp(void *param, TAOS_RES *tres, int code) { if (epSet->numOfEps > 0) { tscEpSetHtons(epSet); - //SRpcCorEpSet *pCorEpSet = pSql->pTscObj->pRpcObj->tscCorMgmtEpSet; + //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++) { diff --git a/src/client/src/tscSql.c b/src/client/src/tscSql.c index 04e7e764f40efcfcfe78bcedfa2e7b4a2f653166..448eea16bf5a529f489b59ced9a9f32c513338eb 100644 --- a/src/client/src/tscSql.c +++ b/src/client/src/tscSql.c @@ -94,7 +94,7 @@ static SSqlObj *taosConnectImpl(const char *ip, const char *user, const char *pa snprintf(rpcKey, sizeof(rpcKey), "%s:%s:%s:%d", user, pass, ip, port); void *pRpcObj = NULL; - if (tscAcquireRpc(rpcKey, user, secretEncrypt,&corMgmtEpSet, &pRpcObj) != 0) { + if (tscAcquireRpc(rpcKey, user, secretEncrypt, &pRpcObj) != 0) { terrno = TSDB_CODE_RPC_NETWORK_UNAVAIL; return NULL; } @@ -105,6 +105,16 @@ static SSqlObj *taosConnectImpl(const char *ip, const char *user, const char *pa tscReleaseRpc(pRpcObj); return NULL; } + + pObj->tscCorMgmtEpSet = malloc(sizeof(SRpcCorEpSet)); + if (pObj->tscCorMgmtEpSet == NULL) { + terrno = TSDB_CODE_TSC_OUT_OF_MEMORY; + tscReleaseRpc(pRpcObj); + free(pObj); + return NULL; + } + memcpy(pObj->tscCorMgmtEpSet, &corMgmtEpSet, sizeof(corMgmtEpSet)); + pObj->signature = pObj; pObj->pRpcObj = (SRpcObj *)pRpcObj; tstrncpy(pObj->user, user, sizeof(pObj->user)); diff --git a/src/client/src/tscSystem.c b/src/client/src/tscSystem.c index bb10f9fae9284ddbc45a9b7c57b777197be6f881..fe1c45bd393938cfd9a77db97169dde2ef6412de 100644 --- a/src/client/src/tscSystem.c +++ b/src/client/src/tscSystem.c @@ -58,7 +58,6 @@ void tscFreeRpcObj(void *param) { SRpcObj *pRpcObj = (SRpcObj *)(param); tscDebug("free rpcObj:%p and free pDnodeConn: %p", pRpcObj, pRpcObj->pDnodeConn); rpcClose(pRpcObj->pDnodeConn); - tfree(pRpcObj->tscCorMgmtEpSet); } void tscReleaseRpc(void *param) { @@ -70,7 +69,7 @@ void tscReleaseRpc(void *param) { pthread_mutex_unlock(&rpcObjMutex); } -int32_t tscAcquireRpc(const char *key, const char *user, const char *secretEncrypt, SRpcCorEpSet *corMgmtEpSet, void **ppRpcObj) { +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)); @@ -109,13 +108,6 @@ int32_t tscAcquireRpc(const char *key, const char *user, const char *secretEncry pthread_mutex_unlock(&rpcObjMutex); return -1; } - pRpcObj->tscCorMgmtEpSet = malloc(sizeof(SRpcCorEpSet)); - if (pRpcObj->tscCorMgmtEpSet == NULL) { - rpcClose(rpcObj.pDnodeConn); - pthread_mutex_unlock(&rpcObjMutex); - return -1; - } - memcpy(pRpcObj->tscCorMgmtEpSet, corMgmtEpSet, sizeof(*corMgmtEpSet)); *ppRpcObj = pRpcObj; pthread_mutex_unlock(&rpcObjMutex); diff --git a/src/client/src/tscUtil.c b/src/client/src/tscUtil.c index 756dc93098f770004cf6791a9f6480a4b2247416..c77d5e5764575f190457248a0f6d4b188915e4bd 100644 --- a/src/client/src/tscUtil.c +++ b/src/client/src/tscUtil.c @@ -897,6 +897,7 @@ void tscCloseTscObj(void *param) { pObj->signature = NULL; taosTmrStopA(&(pObj->pTimer)); + tfree(pObj->tscCorMgmtEpSet); tscReleaseRpc(pObj->pRpcObj); pthread_mutex_destroy(&pObj->mutex);