From 7182e0948eb67a4d745a99bd270cff4cc6135aa9 Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Wed, 29 May 2024 10:49:36 +0800 Subject: [PATCH 1/3] fix:[TD-30306]error in converity scan --- source/client/src/clientSml.c | 2 +- source/client/src/clientSmlTelnet.c | 2 +- source/client/src/clientTmq.c | 15 ++++++--------- 3 files changed, 8 insertions(+), 11 deletions(-) diff --git a/source/client/src/clientSml.c b/source/client/src/clientSml.c index 6225cf703c..6eb7abe0eb 100644 --- a/source/client/src/clientSml.c +++ b/source/client/src/clientSml.c @@ -939,7 +939,7 @@ static int32_t smlSendMetaMsg(SSmlHandle *info, SName *pName, SArray *pColumns, pReq.pColumns = taosArrayInit(pReq.numOfColumns, sizeof(SFieldWithOptions)); for (int32_t i = 0; i < pReq.numOfColumns; ++i) { SField *pField = taosArrayGet(pColumns, i); - SFieldWithOptions fieldWithOption; + SFieldWithOptions fieldWithOption = {0}; setFieldWithOptions(&fieldWithOption, pField); setDefaultOptionsForField(&fieldWithOption); taosArrayPush(pReq.pColumns, &fieldWithOption); diff --git a/source/client/src/clientSmlTelnet.c b/source/client/src/clientSmlTelnet.c index f715f32556..bc0e560178 100644 --- a/source/client/src/clientSmlTelnet.c +++ b/source/client/src/clientSmlTelnet.c @@ -233,7 +233,7 @@ int32_t smlParseTelnetString(SSmlHandle *info, char *sql, char *sqlEnd, SSmlLine SSmlKv kvTs = {0}; smlBuildTsKv(&kvTs, ts); - if (needConverTime) { + if (needConverTime && info->currSTableMeta != NULL) { kvTs.i = convertTimePrecision(kvTs.i, TSDB_TIME_PRECISION_NANO, info->currSTableMeta->tableInfo.precision); } diff --git a/source/client/src/clientTmq.c b/source/client/src/clientTmq.c index 3141e21f26..0bc4214c09 100644 --- a/source/client/src/clientTmq.c +++ b/source/client/src/clientTmq.c @@ -1344,6 +1344,12 @@ int32_t tmqPollCb(void* param, SDataBuf* pMsg, int32_t code) { goto FAIL; } + if(pMsg->pData == NULL){ + tscError("consumer:0x%" PRIx64 " msg discard from vgId:%d, since msg is NULL", tmq->consumerId, vgId); + code = TSDB_CODE_TSC_INTERNAL_ERROR; + goto FAIL; + } + SMqPollRspWrapper* pRspWrapper = taosAllocateQitem(sizeof(SMqPollRspWrapper), DEF_QITEM, 0); if (pRspWrapper == NULL) { tscWarn("consumer:0x%" PRIx64 " msg discard from vgId:%d, since out of memory", tmq->consumerId, vgId); @@ -1356,11 +1362,6 @@ int32_t tmqPollCb(void* param, SDataBuf* pMsg, int32_t code) { goto END; } - if(pMsg->pData == NULL){ - tscError("consumer:0x%" PRIx64 " msg discard from vgId:%d, since msg is NULL", tmq->consumerId, vgId); - goto FAIL; - } - int32_t msgEpoch = ((SMqRspHead*)pMsg->pData)->epoch; int32_t clientEpoch = atomic_load_32(&tmq->epoch); if (msgEpoch < clientEpoch) { @@ -2809,8 +2810,6 @@ int64_t getCommittedFromServer(tmq_t* tmq, char* tname, int32_t vgId, SEpSet* ep int64_t transporterId = 0; code = asyncSendMsgToServer(tmq->pTscObj->pAppInfo->pTransporter, epSet, &transporterId, sendInfo); if (code != 0) { - taosMemoryFree(buf); - taosMemoryFree(sendInfo); tsem_destroy(&pParam->sem); taosMemoryFree(pParam); return code; @@ -3235,8 +3234,6 @@ int32_t tmq_offset_seek(tmq_t* tmq, const char* pTopicName, int32_t vgId, int64_ int64_t transporterId = 0; code = asyncSendMsgToServer(tmq->pTscObj->pAppInfo->pTransporter, &epSet, &transporterId, sendInfo); if (code != 0) { - taosMemoryFree(msg); - taosMemoryFree(sendInfo); tsem_destroy(&pParam->sem); taosMemoryFree(pParam); return code; From d85d3aa70e66e997074cd9b3b34255d52b77c4a4 Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Sun, 2 Jun 2024 22:06:52 +0800 Subject: [PATCH 2/3] fix:[TD-30306]error in converity scan --- source/client/src/clientTmq.c | 26 +++++++++++++------------- 1 file changed, 13 insertions(+), 13 deletions(-) diff --git a/source/client/src/clientTmq.c b/source/client/src/clientTmq.c index 0bc4214c09..7bc6a2f2e6 100644 --- a/source/client/src/clientTmq.c +++ b/source/client/src/clientTmq.c @@ -858,7 +858,10 @@ void tmqSendHbReq(void* param, void* tmrId) { SEpSet epSet = getEpSet_s(&tmq->pTscObj->pAppInfo->mgmtEp); int64_t transporterId = 0; - asyncSendMsgToServer(tmq->pTscObj->pAppInfo->pTransporter, &epSet, &transporterId, sendInfo); + int32_t code = asyncSendMsgToServer(tmq->pTscObj->pAppInfo->pTransporter, &epSet, &transporterId, sendInfo); + if (code != 0) { + tscError("tmqSendHbReq asyncSendMsgToServer failed"); + } OVER: tDestroySMqHbReq(&req); @@ -1220,12 +1223,15 @@ int32_t tmq_subscribe(tmq_t* tmq, const tmq_list_t* topic_list) { sendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo)); if (sendInfo == NULL) { code = TSDB_CODE_OUT_OF_MEMORY; + taosMemoryFree(buf); goto FAIL; } SMqSubscribeCbParam param = {.rspErr = 0}; if (tsem_init(¶m.rspSem, 0, 0) != 0) { code = TSDB_CODE_TSC_INTERNAL_ERROR; + taosMemoryFree(buf); + taosMemoryFree(sendInfo); goto FAIL; } @@ -1245,10 +1251,6 @@ int32_t tmq_subscribe(tmq_t* tmq, const tmq_list_t* topic_list) { goto FAIL; } - // avoid double free if msg is sent - buf = NULL; - sendInfo = NULL; - tsem_wait(¶m.rspSem); tsem_destroy(¶m.rspSem); @@ -1284,8 +1286,6 @@ int32_t tmq_subscribe(tmq_t* tmq, const tmq_list_t* topic_list) { FAIL: taosArrayDestroyP(req.topicNames, taosMemoryFree); - taosMemoryFree(buf); - taosMemoryFree(sendInfo); return code; } @@ -1745,6 +1745,7 @@ static int32_t doTmqPollImpl(tmq_t* pTmq, SMqClientTopic* pTopic, SMqClientVg* p pParam = taosMemoryMalloc(sizeof(SMqPollCbParam)); if (pParam == NULL) { code = TSDB_CODE_OUT_OF_MEMORY; + taosMemoryFreeClear(msg); goto FAIL; } @@ -1756,6 +1757,7 @@ static int32_t doTmqPollImpl(tmq_t* pTmq, SMqClientTopic* pTopic, SMqClientVg* p sendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo)); if (sendInfo == NULL) { code = TSDB_CODE_OUT_OF_MEMORY; + taosMemoryFreeClear(msg); goto FAIL; } @@ -1782,7 +1784,6 @@ static int32_t doTmqPollImpl(tmq_t* pTmq, SMqClientTopic* pTopic, SMqClientVg* p return 0; FAIL: - taosMemoryFreeClear(msg); return tmqPollCb(pParam, NULL, code); } @@ -2560,6 +2561,7 @@ void askEp(tmq_t* pTmq, void* param, bool sync, bool updateEpSet) { if (tSerializeSMqAskEpReq(pReq, tlen, &req) < 0) { tscError("consumer:0x%" PRIx64 ", tSerializeSMqAskEpReq %d failed", pTmq->consumerId, tlen); code = TSDB_CODE_INVALID_PARA; + taosMemoryFree(pReq); goto FAIL; } @@ -2567,6 +2569,7 @@ void askEp(tmq_t* pTmq, void* param, bool sync, bool updateEpSet) { if (pParam == NULL) { tscError("consumer:0x%" PRIx64 ", failed to malloc subscribe param", pTmq->consumerId); code = TSDB_CODE_OUT_OF_MEMORY; + taosMemoryFree(pReq); goto FAIL; } @@ -2577,6 +2580,7 @@ void askEp(tmq_t* pTmq, void* param, bool sync, bool updateEpSet) { SMsgSendInfo* sendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo)); if (sendInfo == NULL) { code = TSDB_CODE_OUT_OF_MEMORY; + taosMemoryFree(pReq); goto FAIL; } @@ -2598,8 +2602,6 @@ void askEp(tmq_t* pTmq, void* param, bool sync, bool updateEpSet) { } FAIL: - taosMemoryFreeClear(pParam); - taosMemoryFreeClear(pReq); askEpCb(pParam, NULL, code); } @@ -2706,7 +2708,6 @@ static int32_t tmqGetWalInfoCb(void* param, SDataBuf* pMsg, int32_t code) { taosMemoryFree(pMsg->pData); taosMemoryFree(pMsg->pEpSet); - taosMemoryFree(pParam); return 0; } @@ -3066,6 +3067,7 @@ int32_t tmq_get_topic_assignment(tmq_t* tmq, const char* pTopicName, tmq_topic_a sendInfo->requestId = req.reqId; sendInfo->requestObjRefId = 0; sendInfo->param = pParam; + sendInfo->paramFreeFp = taosMemoryFree; sendInfo->fp = tmqGetWalInfoCb; sendInfo->msgType = TDMT_VND_TMQ_VG_WALINFO; @@ -3077,8 +3079,6 @@ int32_t tmq_get_topic_assignment(tmq_t* tmq, const char* pTopicName, tmq_topic_a tmq->consumerId, pTopic->topicName, pClientVg->vgId, tmq->epoch, offsetFormatBuf, req.reqId); code = asyncSendMsgToServer(tmq->pTscObj->pAppInfo->pTransporter, &pClientVg->epSet, &transporterId, sendInfo); if (code != 0) { - taosMemoryFree(pParam); - taosMemoryFree(msg); goto end; } } From 5217131f500ccb65d2ca36150ce1904d91e536ad Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Mon, 3 Jun 2024 16:48:44 +0800 Subject: [PATCH 3/3] fix:[TD-30306]error in converity scan --- source/client/src/clientTmq.c | 20 +++++++++----------- 1 file changed, 9 insertions(+), 11 deletions(-) diff --git a/source/client/src/clientTmq.c b/source/client/src/clientTmq.c index 7bc6a2f2e6..13fc2ce16a 100644 --- a/source/client/src/clientTmq.c +++ b/source/client/src/clientTmq.c @@ -1344,12 +1344,6 @@ int32_t tmqPollCb(void* param, SDataBuf* pMsg, int32_t code) { goto FAIL; } - if(pMsg->pData == NULL){ - tscError("consumer:0x%" PRIx64 " msg discard from vgId:%d, since msg is NULL", tmq->consumerId, vgId); - code = TSDB_CODE_TSC_INTERNAL_ERROR; - goto FAIL; - } - SMqPollRspWrapper* pRspWrapper = taosAllocateQitem(sizeof(SMqPollRspWrapper), DEF_QITEM, 0); if (pRspWrapper == NULL) { tscWarn("consumer:0x%" PRIx64 " msg discard from vgId:%d, since out of memory", tmq->consumerId, vgId); @@ -1362,6 +1356,12 @@ int32_t tmqPollCb(void* param, SDataBuf* pMsg, int32_t code) { goto END; } + if(pMsg->pData == NULL){ + tscError("consumer:0x%" PRIx64 " msg discard from vgId:%d, since msg is NULL", tmq->consumerId, vgId); + code = TSDB_CODE_TSC_INTERNAL_ERROR; + goto END; + } + int32_t msgEpoch = ((SMqRspHead*)pMsg->pData)->epoch; int32_t clientEpoch = atomic_load_32(&tmq->epoch); if (msgEpoch < clientEpoch) { @@ -1388,7 +1388,7 @@ int32_t tmqPollCb(void* param, SDataBuf* pMsg, int32_t code) { tDecoderClear(&decoder); taosReleaseRef(tmqMgmt.rsetId, refId); code = TSDB_CODE_OUT_OF_MEMORY; - goto FAIL; + goto END; } tDecoderClear(&decoder); memcpy(&pRspWrapper->dataRsp, pMsg->pData, sizeof(SMqRspHead)); @@ -1404,7 +1404,7 @@ int32_t tmqPollCb(void* param, SDataBuf* pMsg, int32_t code) { tDecoderClear(&decoder); taosReleaseRef(tmqMgmt.rsetId, refId); code = TSDB_CODE_OUT_OF_MEMORY; - goto FAIL; + goto END; } tDecoderClear(&decoder); memcpy(&pRspWrapper->metaRsp, pMsg->pData, sizeof(SMqRspHead)); @@ -1415,7 +1415,7 @@ int32_t tmqPollCb(void* param, SDataBuf* pMsg, int32_t code) { tDecoderClear(&decoder); taosReleaseRef(tmqMgmt.rsetId, refId); code = TSDB_CODE_OUT_OF_MEMORY; - goto FAIL; + goto END; } tDecoderClear(&decoder); memcpy(&pRspWrapper->taosxRsp, pMsg->pData, sizeof(SMqRspHead)); @@ -1889,8 +1889,6 @@ static void* tmqHandleAllRsp(tmq_t* tmq, int64_t timeout) { terrno = pRspWrapper->code; tscError("consumer:0x%" PRIx64 " unexpected rsp from poll, code:%s", tmq->consumerId, tstrerror(pRspWrapper->code)); - taosFreeQitem(pRspWrapper); - return NULL; } else { if (pRspWrapper->code == TSDB_CODE_VND_INVALID_VGROUP_ID) { // for vnode transform askEp(tmq, NULL, false, true);