From 40b6ce676142816ac74be97ada419e7179abdd7e Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Tue, 22 Sep 2020 15:33:08 +0800 Subject: [PATCH 1/5] [td-225] fix coverity problem --- src/client/src/tscUtil.c | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/client/src/tscUtil.c b/src/client/src/tscUtil.c index 20c3bc2cb6..7387c23e1f 100644 --- a/src/client/src/tscUtil.c +++ b/src/client/src/tscUtil.c @@ -402,6 +402,8 @@ void tscFreeSqlObj(SSqlObj* pSql) { return; } + void *p = pSql; + tscDebug("%p start to free sqlObj", pSql); STscObj* pTscObj = pSql->pTscObj; @@ -421,7 +423,8 @@ void tscFreeSqlObj(SSqlObj* pSql) { tsem_destroy(&pSql->rspSem); free(pSql); - tscDebug("%p free sqlObj completed", pSql); + + tscDebug("%p free sqlObj completed", p); int32_t ref = T_REF_DEC(pTscObj); assert(ref >= 0); From 8ba206115a18239e43453abd11186648b04d16ef Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Wed, 23 Sep 2020 14:39:14 +0800 Subject: [PATCH 2/5] [td-1575] --- src/util/inc/tskiplist.h | 1 + src/util/src/tskiplist.c | 35 ++++++++++++++++++++++++++++++----- 2 files changed, 31 insertions(+), 5 deletions(-) diff --git a/src/util/inc/tskiplist.h b/src/util/inc/tskiplist.h index 4ba620dce0..a14a856561 100644 --- a/src/util/inc/tskiplist.h +++ b/src/util/inc/tskiplist.h @@ -136,6 +136,7 @@ typedef struct SSkipListIterator { SSkipListNode *cur; int32_t step; // the number of nodes that have been checked already int32_t order; // order of the iterator + SSkipListNode *next; // next points to the true qualified node in skip list } SSkipListIterator; /** diff --git a/src/util/src/tskiplist.c b/src/util/src/tskiplist.c index 303c2440bf..bacdaef6c8 100644 --- a/src/util/src/tskiplist.c +++ b/src/util/src/tskiplist.c @@ -79,9 +79,12 @@ static SSkipListIterator* doCreateSkipListIterator(SSkipList *pSkipList, int32_t // when order is TSDB_ORDER_ASC, return the last node with key less than val // when order is TSDB_ORDER_DESC, return the first node with key large than val -static SSkipListNode* getPriorNode(SSkipList* pSkipList, const char* val, int32_t order) { +static SSkipListNode* getPriorNode(SSkipList* pSkipList, const char* val, int32_t order, SSkipListNode** pCur) { __compar_fn_t comparFn = pSkipList->comparFn; SSkipListNode *pNode = NULL; + if (pCur != NULL) { + *pCur = NULL; + } if (order == TSDB_ORDER_ASC) { pNode = pSkipList->pHead; @@ -93,6 +96,9 @@ static SSkipListNode* getPriorNode(SSkipList* pSkipList, const char* val, int32_ pNode = p; p = SL_GET_FORWARD_POINTER(p, i); } else { + if (pCur != NULL) { + *pCur = p; + } break; } } @@ -107,6 +113,9 @@ static SSkipListNode* getPriorNode(SSkipList* pSkipList, const char* val, int32_ pNode = p; p = SL_GET_BACKWARD_POINTER(p, i); } else { + if (pCur != NULL) { + *pCur = p; + } break; } } @@ -295,7 +304,7 @@ SArray* tSkipListGet(SSkipList *pSkipList, SSkipListKey key) { pthread_rwlock_wrlock(pSkipList->lock); } - SSkipListNode* pNode = getPriorNode(pSkipList, key, TSDB_ORDER_ASC); + SSkipListNode* pNode = getPriorNode(pSkipList, key, TSDB_ORDER_ASC, NULL); while (1) { SSkipListNode *p = SL_GET_FORWARD_POINTER(pNode, 0); if (p == pSkipList->pTail) { @@ -452,7 +461,7 @@ uint32_t tSkipListRemove(SSkipList *pSkipList, SSkipListKey key) { pthread_rwlock_wrlock(pSkipList->lock); } - SSkipListNode* pNode = getPriorNode(pSkipList, key, TSDB_ORDER_ASC); + SSkipListNode* pNode = getPriorNode(pSkipList, key, TSDB_ORDER_ASC, NULL); while (1) { SSkipListNode *p = SL_GET_FORWARD_POINTER(pNode, 0); if (p == pSkipList->pTail) { @@ -545,7 +554,7 @@ SSkipListIterator *tSkipListCreateIterFromVal(SSkipList* pSkipList, const char* pthread_rwlock_rdlock(pSkipList->lock); } - iter->cur = getPriorNode(pSkipList, val, order); + iter->cur = getPriorNode(pSkipList, val, order, &iter->next); if (pSkipList->lock) { pthread_rwlock_unlock(pSkipList->lock); @@ -567,8 +576,22 @@ bool tSkipListIterNext(SSkipListIterator *iter) { if (iter->order == TSDB_ORDER_ASC) { // ascending order iterate iter->cur = SL_GET_FORWARD_POINTER(iter->cur, 0); + + // a new node is inserted into between iter->cur and iter->next, ignore it + if (iter->cur != iter->next && (iter->next != NULL)) { + iter->cur = iter->next; + } + + iter->next = SL_GET_FORWARD_POINTER(iter->cur, 0); } else { // descending order iterate iter->cur = SL_GET_BACKWARD_POINTER(iter->cur, 0); + + // a new node is inserted into between iter->cur and iter->next, ignore it + if (iter->cur != iter->next && (iter->next != NULL)) { + iter->cur = iter->next; + } + + iter->next = SL_GET_BACKWARD_POINTER(iter->cur, 0); } if (pSkipList->lock) { @@ -715,9 +738,11 @@ SSkipListIterator* doCreateSkipListIterator(SSkipList *pSkipList, int32_t order) iter->order = order; if(order == TSDB_ORDER_ASC) { iter->cur = pSkipList->pHead; + iter->next = SL_GET_FORWARD_POINTER(iter->cur, 0); } else { iter->cur = pSkipList->pTail; + iter->next = SL_GET_BACKWARD_POINTER(iter->cur, 0); } - + return iter; } \ No newline at end of file From 02d1583725496bce0937ff305a5d21b431830155 Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Thu, 24 Sep 2020 11:09:16 +0800 Subject: [PATCH 3/5] [td-1523] fix memory leaks and refactor some functions. --- src/client/inc/tscUtil.h | 2 ++ src/client/src/tscAsync.c | 5 +---- src/client/src/tscPrepare.c | 6 ++---- src/client/src/tscServer.c | 23 ++++++----------------- src/client/src/tscSql.c | 7 ++----- src/client/src/tscStream.c | 4 +--- src/client/src/tscSub.c | 4 +--- src/client/src/tscSystem.c | 12 +++--------- src/client/src/tscUtil.c | 24 ++++++++++++++---------- src/util/src/tcache.c | 2 +- 10 files changed, 33 insertions(+), 56 deletions(-) diff --git a/src/client/inc/tscUtil.h b/src/client/inc/tscUtil.h index 79a792ab65..0323434a99 100644 --- a/src/client/inc/tscUtil.h +++ b/src/client/inc/tscUtil.h @@ -258,6 +258,8 @@ void tscDoQuery(SSqlObj* pSql); */ SSqlObj* createSimpleSubObj(SSqlObj* pSql, void (*fp)(), void* param, int32_t cmd); +void registerSqlObj(SSqlObj* pSql); + SSqlObj* createSubqueryObj(SSqlObj* pSql, int16_t tableIndex, void (*fp)(), void* param, int32_t cmd, SSqlObj* pPrevSql); void addGroupInfoForSubquery(SSqlObj* pParentObj, SSqlObj* pSql, int32_t subClauseIndex, int32_t tableIndex); diff --git a/src/client/src/tscAsync.c b/src/client/src/tscAsync.c index 09610575f6..c5d622e245 100644 --- a/src/client/src/tscAsync.c +++ b/src/client/src/tscAsync.c @@ -51,10 +51,7 @@ void doAsyncQuery(STscObj* pObj, SSqlObj* pSql, void (*fp)(), void* param, const pSql->fp = fp; pSql->fetchFp = fp; - uint64_t handle = (uint64_t) pSql; - pSql->self = taosCachePut(tscObjCache, &handle, sizeof(uint64_t), &pSql, sizeof(uint64_t), 2*3600*1000); - - T_REF_INC(pSql->pTscObj); + registerSqlObj(pSql); pSql->sqlstr = calloc(1, sqlLen + 1); if (pSql->sqlstr == NULL) { diff --git a/src/client/src/tscPrepare.c b/src/client/src/tscPrepare.c index c4ca6793ff..cdbd8685df 100644 --- a/src/client/src/tscPrepare.c +++ b/src/client/src/tscPrepare.c @@ -545,10 +545,8 @@ int taos_stmt_prepare(TAOS_STMT* stmt, const char* sql, unsigned long length) { pSql->cmd.numOfParams = 0; pSql->cmd.batchSize = 0; - - uint64_t handle = (uint64_t) pSql; - pSql->self = taosCachePut(tscObjCache, &handle, sizeof(uint64_t), &pSql, sizeof(uint64_t), 2*3600*1000); - T_REF_INC(pSql->pTscObj); + + registerSqlObj(pSql); int32_t code = tsParseSql(pSql, true); if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) { diff --git a/src/client/src/tscServer.c b/src/client/src/tscServer.c index 8eaa406bce..b26abf2145 100644 --- a/src/client/src/tscServer.c +++ b/src/client/src/tscServer.c @@ -1987,15 +1987,11 @@ static void createHBObj(STscObj* pObj) { pSql->param = pObj; pSql->pTscObj = pObj; pSql->signature = pSql; + + registerSqlObj(pSql); + tscDebug("%p HB is allocated, pObj:%p", pSql, pObj); + pObj->pHb = pSql; - - tscAddSubqueryInfo(&pObj->pHb->cmd); - - int64_t ad = (int64_t) pSql; - pSql->self = taosCachePut(tscObjCache, &ad, sizeof(int64_t), &pSql, sizeof(int64_t), 2 * 60 * 1000); - T_REF_INC(pObj); - - tscDebug("%p HB is allocated, pObj:%p", pObj->pHb, pObj); } int tscProcessConnectRsp(SSqlObj *pSql) { @@ -2170,11 +2166,7 @@ static int32_t getTableMetaFromMgmt(SSqlObj *pSql, STableMetaInfo *pTableMetaInf pNew->signature = pNew; pNew->cmd.command = TSDB_SQL_META; - T_REF_INC(pNew->pTscObj); - - // TODO add test case on x86 platform - uint64_t adr = (uint64_t) pNew; - pNew->self = taosCachePut(tscObjCache, &adr, sizeof(uint64_t), &pNew, sizeof(uint64_t), 2*60*1000); + registerSqlObj(pNew); tscAddSubqueryInfo(&pNew->cmd); @@ -2301,10 +2293,7 @@ int tscGetSTableVgroupInfo(SSqlObj *pSql, int32_t clauseIndex) { } pNewQueryInfo->numOfTables = pQueryInfo->numOfTables; - T_REF_INC(pNew->pTscObj); - - uint64_t p = (uint64_t) pNew; - pNew->self = taosCachePut(tscObjCache, &p, sizeof(uint64_t), &pNew, sizeof(uint64_t), 2 * 600 * 1000); + registerSqlObj(pNew); tscDebug("%p new sqlObj:%p to get vgroupInfo, numOfTables:%d", pSql, pNew, pNewQueryInfo->numOfTables); pNew->fp = tscTableMetaCallBack; diff --git a/src/client/src/tscSql.c b/src/client/src/tscSql.c index e1a07fdcfe..347e3cb508 100644 --- a/src/client/src/tscSql.c +++ b/src/client/src/tscSql.c @@ -156,10 +156,7 @@ SSqlObj *taosConnectImpl(const char *ip, const char *user, const char *pass, con *taos = pObj; } - T_REF_INC(pSql->pTscObj); - - uint64_t key = (uint64_t) pSql; - pSql->self = taosCachePut(tscObjCache, &key, sizeof(uint64_t), &pSql, sizeof(uint64_t), 2*3600*1000); + registerSqlObj(pSql); tsInsertHeadSize = sizeof(SMsgDesc) + sizeof(SSubmitMsg); return pSql; @@ -270,7 +267,7 @@ void taos_close(TAOS *taos) { pHb->pRpcCtx = NULL; } - tscDebug("%p, HB is freed", pHb); + tscDebug("%p HB is freed", pHb); taos_free_result(pHb); } diff --git a/src/client/src/tscStream.c b/src/client/src/tscStream.c index 35cd09a033..81b8cf7359 100644 --- a/src/client/src/tscStream.c +++ b/src/client/src/tscStream.c @@ -510,9 +510,7 @@ static void tscCreateStream(void *param, TAOS_RES *res, int code) { return; } - uint64_t handle = (uint64_t) pSql; - pSql->self = taosCachePut(tscObjCache, &handle, sizeof(uint64_t), &pSql, sizeof(uint64_t), 2*3600*1000); - T_REF_INC(pSql->pTscObj); + registerSqlObj(pSql); SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, 0); STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0); diff --git a/src/client/src/tscSub.c b/src/client/src/tscSub.c index 760c5f5a51..7913e0fa03 100644 --- a/src/client/src/tscSub.c +++ b/src/client/src/tscSub.c @@ -152,9 +152,7 @@ static SSub* tscCreateSubscription(STscObj* pObj, const char* topic, const char* goto fail; } - uint64_t handle = (uint64_t) pSql; - pSql->self = taosCachePut(tscObjCache, &handle, sizeof(uint64_t), &pSql, sizeof(uint64_t), 2*3600*1000); - T_REF_INC(pSql->pTscObj); + registerSqlObj(pSql); code = tsParseSql(pSql, false); if (code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) { diff --git a/src/client/src/tscSystem.c b/src/client/src/tscSystem.c index 82ce1d3679..620fe13a9f 100644 --- a/src/client/src/tscSystem.c +++ b/src/client/src/tscSystem.c @@ -122,11 +122,8 @@ void taos_init_imp(void) { tscInitMsgsFp(); int queueSize = tsMaxConnections*2; - if (tscEmbedded == 0) { - tscNumOfThreads = (int)(tsNumOfCores * tsNumOfThreadsPerCore / 2.0); - } else { - tscNumOfThreads = (int)(tsNumOfCores * tsNumOfThreadsPerCore / 4.0); - } + double factor = (tscEmbedded == 0)? 2.0:4.0; + tscNumOfThreads = (int)(tsNumOfCores * tsNumOfThreadsPerCore / factor); if (tscNumOfThreads < 2) tscNumOfThreads = 2; @@ -140,11 +137,8 @@ void taos_init_imp(void) { if(0 == tscEmbedded){ taosTmrReset(tscCheckDiskUsage, 10, NULL, tscTmr, &tscCheckDiskUsageTmr); } - - int64_t refreshTime = tsTableMetaKeepTimer; - refreshTime = refreshTime > 10 ? 10 : refreshTime; - refreshTime = refreshTime < 10 ? 10 : refreshTime; + int64_t refreshTime = 10; // 10 seconds by default if (tscMetaCache == NULL) { tscMetaCache = taosCacheInit(TSDB_DATA_TYPE_BINARY, refreshTime, false, NULL, "tableMeta"); tscObjCache = taosCacheInit(TSDB_DATA_TYPE_BIGINT, refreshTime/2, false, tscFreeSqlObjInCache, "sqlObj"); diff --git a/src/client/src/tscUtil.c b/src/client/src/tscUtil.c index 7f445344b1..5d439927e2 100644 --- a/src/client/src/tscUtil.c +++ b/src/client/src/tscUtil.c @@ -424,12 +424,12 @@ void tscFreeSqlObj(SSqlObj* pSql) { free(pSql); - tscDebug("%p free sqlObj completed", p); - int32_t ref = T_REF_DEC(pTscObj); assert(ref >= 0); + tscDebug("%p free sqlObj completed, tscObj:%p ref:%d", p, pTscObj, ref); if (ref == 0) { + tscDebug("%p all sqlObj freed, free tscObj:%p", p, pTscObj); tscCloseTscObj(pTscObj); } } @@ -1783,6 +1783,16 @@ void tscResetForNextRetrieve(SSqlRes* pRes) { pRes->numOfRows = 0; } +void registerSqlObj(SSqlObj* pSql) { + int64_t DEFAULT_LIFE_TIME = 2 * 600 * 1000; // 1200 sec + + int32_t ref = T_REF_INC(pSql->pTscObj); + tscDebug("%p add to tscObj:%p, ref:%d", pSql, pSql->pTscObj, ref); + + uint64_t p = (uint64_t) pSql; + pSql->self = taosCachePut(tscObjCache, &p, sizeof(uint64_t), &p, sizeof(uint64_t), DEFAULT_LIFE_TIME); +} + SSqlObj* createSimpleSubObj(SSqlObj* pSql, void (*fp)(), void* param, int32_t cmd) { SSqlObj* pNew = (SSqlObj*)calloc(1, sizeof(SSqlObj)); if (pNew == NULL) { @@ -1822,10 +1832,7 @@ SSqlObj* createSimpleSubObj(SSqlObj* pSql, void (*fp)(), void* param, int32_t cm STableMetaInfo* pMasterTableMetaInfo = tscGetTableMetaInfoFromCmd(&pSql->cmd, pSql->cmd.clauseIndex, 0); tscAddTableMetaInfo(pQueryInfo, pMasterTableMetaInfo->name, NULL, NULL, NULL); - - T_REF_INC(pNew->pTscObj); - uint64_t p = (uint64_t) pNew; - pNew->self = taosCachePut(tscObjCache, &p, sizeof(uint64_t), &pNew, sizeof(uint64_t), 2 * 600 * 1000); + registerSqlObj(pNew); return pNew; } @@ -2063,10 +2070,7 @@ SSqlObj* createSubqueryObj(SSqlObj* pSql, int16_t tableIndex, void (*fp)(), void tscDebug("%p new sub insertion: %p, vnodeIdx:%d", pSql, pNew, pTableMetaInfo->vgroupIndex); } - T_REF_INC(pNew->pTscObj); - - uint64_t p = (uint64_t) pNew; - pNew->self = taosCachePut(tscObjCache, &p, sizeof(uint64_t), &pNew, sizeof(uint64_t), 2 * 600 * 10); + registerSqlObj(pNew); return pNew; _error: diff --git a/src/util/src/tcache.c b/src/util/src/tcache.c index 49b9996cf4..2637699adb 100644 --- a/src/util/src/tcache.c +++ b/src/util/src/tcache.c @@ -97,7 +97,7 @@ static FORCE_INLINE void taosCacheReleaseNode(SCacheObj *pCacheObj, SCacheDataNo int32_t size = (int32_t)taosHashGetSize(pCacheObj->pHashTable); assert(size > 0); - uDebug("cache:%s, key:%p, %p is destroyed from cache, size:%dbytes, num:%d size:%" PRId64 "bytes", + uDebug("cache:%s, key:%p, %p is destroyed from cache, size:%dbytes, totalNum:%d size:%" PRId64 "bytes", pCacheObj->name, pNode->key, pNode->data, pNode->size, size - 1, pCacheObj->totalSize); if (pCacheObj->freeFp) { From 3a26dd3bfcd019f5ba45164a220c997c91c69139 Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Thu, 24 Sep 2020 11:14:35 +0800 Subject: [PATCH 4/5] [td-225]fix compiler error. --- src/client/src/tscUtil.c | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/client/src/tscUtil.c b/src/client/src/tscUtil.c index 5d439927e2..21524f8fd2 100644 --- a/src/client/src/tscUtil.c +++ b/src/client/src/tscUtil.c @@ -1784,7 +1784,7 @@ void tscResetForNextRetrieve(SSqlRes* pRes) { } void registerSqlObj(SSqlObj* pSql) { - int64_t DEFAULT_LIFE_TIME = 2 * 600 * 1000; // 1200 sec + int32_t DEFAULT_LIFE_TIME = 2 * 600 * 1000; // 1200 sec int32_t ref = T_REF_INC(pSql->pTscObj); tscDebug("%p add to tscObj:%p, ref:%d", pSql, pSql->pTscObj, ref); From 4aaf2bd99c7252f3a3b520a7de70e35888e3dd68 Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Thu, 24 Sep 2020 13:52:30 +0800 Subject: [PATCH 5/5] [td-225] refactor codes. --- src/client/inc/tsclient.h | 4 ++-- src/client/src/tscServer.c | 3 ++- src/client/src/tscSql.c | 34 ++++++++++++++++++---------------- src/client/src/tscSub.c | 6 ++++++ src/client/src/tscUtil.c | 27 +++++++++++++-------------- 5 files changed, 41 insertions(+), 33 deletions(-) diff --git a/src/client/inc/tsclient.h b/src/client/inc/tsclient.h index 55ca02dfb5..ea42b0f6a3 100644 --- a/src/client/inc/tsclient.h +++ b/src/client/inc/tsclient.h @@ -89,7 +89,7 @@ typedef struct STableComInfo { typedef struct SCMCorVgroupInfo { int32_t version; - int8_t inUse; + int8_t inUse; int8_t numOfEps; SEpAddr epAddr[TSDB_MAX_REPLICA]; } SCMCorVgroupInfo; @@ -107,7 +107,7 @@ typedef struct STableMeta { } STableMeta; typedef struct STableMetaInfo { - STableMeta * pTableMeta; // table meta, cached in client side and acquired by name + STableMeta *pTableMeta; // table meta, cached in client side and acquired by name SVgroupsInfo *vgroupList; SArray *pVgroupTables; // SArray diff --git a/src/client/src/tscServer.c b/src/client/src/tscServer.c index b26abf2145..3cbb0d936e 100644 --- a/src/client/src/tscServer.c +++ b/src/client/src/tscServer.c @@ -128,6 +128,7 @@ static void tscUpdateVgroupInfo(SSqlObj *pObj, SRpcEpSet *pEpSet) { tscDebug("after: EndPoint in use: %d", pVgroupInfo->inUse); taosCorEndWrite(&pVgroupInfo->version); } + void tscPrintMgmtEp() { SRpcEpSet dump; tscDumpMgmtEpSet(&dump); @@ -745,7 +746,6 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) { SSqlExpr *pExpr = tscSqlExprGet(pQueryInfo, i); if (!tscValidateColumnId(pTableMetaInfo, pExpr->colInfo.colId, pExpr->numOfParams)) { - /* column id is not valid according to the cached table meta, the table meta is expired */ tscError("%p table schema is not matched with parsed sql", pSql); return TSDB_CODE_TSC_INVALID_SQL; } @@ -2294,6 +2294,7 @@ int tscGetSTableVgroupInfo(SSqlObj *pSql, int32_t clauseIndex) { pNewQueryInfo->numOfTables = pQueryInfo->numOfTables; registerSqlObj(pNew); + tscDebug("%p new sqlObj:%p to get vgroupInfo, numOfTables:%d", pSql, pNew, pNewQueryInfo->numOfTables); pNew->fp = tscTableMetaCallBack; diff --git a/src/client/src/tscSql.c b/src/client/src/tscSql.c index 347e3cb508..430a762321 100644 --- a/src/client/src/tscSql.c +++ b/src/client/src/tscSql.c @@ -786,14 +786,17 @@ int taos_validate_sql(TAOS *taos, const char *sql) { } SSqlObj* pSql = calloc(1, sizeof(SSqlObj)); + pSql->pTscObj = taos; pSql->signature = pSql; + SSqlRes *pRes = &pSql->res; SSqlCmd *pCmd = &pSql->cmd; pRes->numOfTotal = 0; pRes->numOfClauseTotal = 0; + tscDebug("%p Valid SQL: %s pObj:%p", pSql, sql, pObj); int32_t sqlLen = (int32_t)strlen(sql); @@ -829,11 +832,12 @@ int taos_validate_sql(TAOS *taos, const char *sql) { tsem_wait(&pSql->rspSem); code = pSql->res.code; } + if (code != TSDB_CODE_SUCCESS) { tscDebug("%p Valid SQL result:%d, %s pObj:%p", pSql, code, taos_errstr(taos), pObj); } - taos_free_result(pSql); + taos_free_result(pSql); return code; } @@ -932,34 +936,32 @@ int taos_load_table_info(TAOS *taos, const char *tableNameList) { SSqlObj* pSql = calloc(1, sizeof(SSqlObj)); pSql->pTscObj = taos; pSql->signature = pSql; + SSqlRes *pRes = &pSql->res; + pRes->code = 0; pRes->numOfTotal = 0; // the number of getting table meta from server pRes->numOfClauseTotal = 0; - pRes->code = 0; - assert(pSql->fp == NULL); tscDebug("%p tableNameList: %s pObj:%p", pSql, tableNameList, pObj); int32_t tblListLen = (int32_t)strlen(tableNameList); if (tblListLen > MAX_TABLE_NAME_LENGTH) { tscError("%p tableNameList too long, length:%d, maximum allowed:%d", pSql, tblListLen, MAX_TABLE_NAME_LENGTH); - pRes->code = TSDB_CODE_TSC_INVALID_SQL; - taosTFree(pSql); - return pRes->code; + tscFreeSqlObj(pSql); + return TSDB_CODE_TSC_INVALID_SQL; } char *str = calloc(1, tblListLen + 1); if (str == NULL) { - pRes->code = TSDB_CODE_TSC_OUT_OF_MEMORY; tscError("%p failed to malloc sql string buffer", pSql); - taosTFree(pSql); - return pRes->code; + tscFreeSqlObj(pSql); + return TSDB_CODE_TSC_OUT_OF_MEMORY; } strtolower(str, tableNameList); - pRes->code = (uint8_t)tscParseTblNameList(pSql, str, tblListLen); + int32_t code = (uint8_t) tscParseTblNameList(pSql, str, tblListLen); /* * set the qhandle to 0 before return in order to erase the qhandle value assigned in the previous successful query. @@ -969,17 +971,17 @@ int taos_load_table_info(TAOS *taos, const char *tableNameList) { pRes->qhandle = 0; free(str); - if (pRes->code != TSDB_CODE_SUCCESS) { + if (code != TSDB_CODE_SUCCESS) { tscFreeSqlObj(pSql); - return pRes->code; + return code; } tscDoQuery(pSql); - tscDebug("%p load multi metermeta result:%d %s pObj:%p", pSql, pRes->code, taos_errstr(taos), pObj); - if (pRes->code != TSDB_CODE_SUCCESS) { - tscPartiallyFreeSqlObj(pSql); + tscDebug("%p load multi table meta result:%d %s pObj:%p", pSql, pRes->code, taos_errstr(taos), pObj); + if ((code = pRes->code) != TSDB_CODE_SUCCESS) { + tscFreeSqlObj(pSql); } - return pRes->code; + return code; } diff --git a/src/client/src/tscSub.c b/src/client/src/tscSub.c index 7913e0fa03..2c81bd7c7c 100644 --- a/src/client/src/tscSub.c +++ b/src/client/src/tscSub.c @@ -105,6 +105,7 @@ static SSub* tscCreateSubscription(STscObj* pObj, const char* topic, const char* code = TAOS_SYSTEM_ERROR(errno); goto fail; } + tstrncpy(pSub->topic, topic, sizeof(pSub->topic)); pSub->progress = taosArrayInit(32, sizeof(SSubscriptionProgress)); if (pSub->progress == NULL) { @@ -119,6 +120,7 @@ static SSub* tscCreateSubscription(STscObj* pObj, const char* topic, const char* code = TSDB_CODE_TSC_OUT_OF_MEMORY; goto fail; } + pSql->signature = pSql; pSql->pTscObj = pObj; pSql->pSubscription = pSub; @@ -142,6 +144,7 @@ static SSub* tscCreateSubscription(STscObj* pObj, const char* topic, const char* code = TSDB_CODE_TSC_OUT_OF_MEMORY; goto fail; } + strtolower(pSql->sqlstr, pSql->sqlstr); pRes->qhandle = 0; pRes->numOfRows = 1; @@ -159,6 +162,7 @@ static SSub* tscCreateSubscription(STscObj* pObj, const char* topic, const char* tsem_wait(&pSub->sem); code = pSql->res.code; } + if (code != TSDB_CODE_SUCCESS) { line = __LINE__; goto fail; @@ -180,8 +184,10 @@ fail: } else { tscFreeSqlObj(pSql); } + pSql = NULL; } + if (pSub != NULL) { taosArrayDestroy(pSub->progress); tsem_destroy(&pSub->sem); diff --git a/src/client/src/tscUtil.c b/src/client/src/tscUtil.c index 21524f8fd2..33362409cf 100644 --- a/src/client/src/tscUtil.c +++ b/src/client/src/tscUtil.c @@ -391,10 +391,21 @@ static UNUSED_FUNC void tscFreeSubobj(SSqlObj* pSql) { */ void tscFreeSqlObjInCache(void *pSql) { assert(pSql != NULL); + SSqlObj** p = (SSqlObj**)pSql; + STscObj* pTscObj = (*p)->pTscObj; assert((*p)->self != 0 && (*p)->self == (p)); tscFreeSqlObj(*p); + + int32_t ref = T_REF_DEC(pTscObj); + assert(ref >= 0); + + tscDebug("%p free sqlObj completed, tscObj:%p ref:%d", *p, pTscObj, ref); + if (ref == 0) { + tscDebug("%p all sqlObj freed, free tscObj:%p", *p, pTscObj); + tscCloseTscObj(pTscObj); + } } void tscFreeSqlObj(SSqlObj* pSql) { @@ -402,10 +413,7 @@ void tscFreeSqlObj(SSqlObj* pSql) { return; } - void *p = pSql; - tscDebug("%p start to free sqlObj", pSql); - STscObj* pTscObj = pSql->pTscObj; tscFreeSubobj(pSql); tscPartiallyFreeSqlObj(pSql); @@ -423,15 +431,6 @@ void tscFreeSqlObj(SSqlObj* pSql) { tsem_destroy(&pSql->rspSem); free(pSql); - - int32_t ref = T_REF_DEC(pTscObj); - assert(ref >= 0); - - tscDebug("%p free sqlObj completed, tscObj:%p ref:%d", p, pTscObj, ref); - if (ref == 0) { - tscDebug("%p all sqlObj freed, free tscObj:%p", p, pTscObj); - tscCloseTscObj(pTscObj); - } } void tscDestroyDataBlock(STableDataBlocks* pDataBlock) { @@ -1821,8 +1820,7 @@ SSqlObj* createSimpleSubObj(SSqlObj* pSql, void (*fp)(), void* param, int32_t cm pNew->sqlstr = strdup(pSql->sqlstr); if (pNew->sqlstr == NULL) { tscError("%p new subquery failed", pSql); - - free(pNew); + tscFreeSqlObj(pNew); return NULL; } @@ -1832,6 +1830,7 @@ SSqlObj* createSimpleSubObj(SSqlObj* pSql, void (*fp)(), void* param, int32_t cm STableMetaInfo* pMasterTableMetaInfo = tscGetTableMetaInfoFromCmd(&pSql->cmd, pSql->cmd.clauseIndex, 0); tscAddTableMetaInfo(pQueryInfo, pMasterTableMetaInfo->name, NULL, NULL, NULL); + registerSqlObj(pNew); return pNew; }