From a3fb4064d00ece5144faee8110b32e710e395c0a Mon Sep 17 00:00:00 2001 From: Minglei Jin Date: Sat, 13 Aug 2022 13:34:02 +0800 Subject: [PATCH] fix: destroy pResList array in error cases --- source/libs/qworker/src/qworker.c | 109 +++++++++++++++--------------- 1 file changed, 55 insertions(+), 54 deletions(-) diff --git a/source/libs/qworker/src/qworker.c b/source/libs/qworker/src/qworker.c index e06b752862..862d142100 100644 --- a/source/libs/qworker/src/qworker.c +++ b/source/libs/qworker/src/qworker.c @@ -1,3 +1,5 @@ +#include "qworker.h" + #include "dataSinkMgt.h" #include "executor.h" #include "planner.h" @@ -7,7 +9,6 @@ #include "tcommon.h" #include "tmsg.h" #include "tname.h" -#include "qworker.h" SQWorkerMgmt gQwMgmt = { .lock = 0, @@ -15,7 +16,6 @@ SQWorkerMgmt gQwMgmt = { .qwNum = 0, }; - int32_t qwProcessHbLinkBroken(SQWorker *mgmt, SQWMsg *qwMsg, SSchedulerHbReq *req) { int32_t code = 0; SSchedulerHbRsp rsp = {0}; @@ -26,7 +26,7 @@ int32_t qwProcessHbLinkBroken(SQWorker *mgmt, SQWMsg *qwMsg, SSchedulerHbReq *re QW_LOCK(QW_WRITE, &sch->hbConnLock); sch->hbBrokenTs = taosGetTimestampMs(); - + if (qwMsg->connInfo.handle == sch->hbConnInfo.handle) { tmsgReleaseHandle(&sch->hbConnInfo, TAOS_CONN_SERVER); sch->hbConnInfo.handle = NULL; @@ -44,8 +44,8 @@ int32_t qwProcessHbLinkBroken(SQWorker *mgmt, SQWMsg *qwMsg, SSchedulerHbReq *re QW_RET(TSDB_CODE_SUCCESS); } -static void freeItem(void* param) { - SExplainExecInfo* pInfo = param; +static void freeItem(void *param) { + SExplainExecInfo *pInfo = param; taosMemoryFree(pInfo->verboseInfo); } @@ -54,7 +54,7 @@ int32_t qwHandleTaskComplete(QW_FPARAMS_DEF, SQWTaskCtx *ctx) { if (TASK_TYPE_TEMP == ctx->taskType && taskHandle) { if (ctx->explain) { - SArray* execInfoList = taosArrayInit(4, sizeof(SExplainExecInfo)); + SArray *execInfoList = taosArrayInit(4, sizeof(SExplainExecInfo)); QW_ERR_RET(qGetExplainExecInfo(taskHandle, execInfoList)); SRpcHandleInfo connInfo = ctx->ctrlConnInfo; @@ -81,7 +81,7 @@ int32_t qwExecTask(QW_FPARAMS_DEF, SQWTaskCtx *ctx, bool *queryStop) { qTaskInfo_t taskHandle = ctx->taskHandle; DataSinkHandle sinkHandle = ctx->sinkHandle; - SArray* pResList = taosArrayInit(4, POINTER_BYTES); + SArray *pResList = taosArrayInit(4, POINTER_BYTES); while (true) { QW_TASK_DLOG("start to execTask, loopIdx:%d", i++); @@ -95,7 +95,7 @@ int32_t qwExecTask(QW_FPARAMS_DEF, SQWTaskCtx *ctx, bool *queryStop) { } else { QW_TASK_DLOG("qExecTask failed, code:%x - %s", code, tstrerror(code)); } - QW_ERR_RET(code); + QW_ERR_JRET(code); } } @@ -105,7 +105,7 @@ int32_t qwExecTask(QW_FPARAMS_DEF, SQWTaskCtx *ctx, bool *queryStop) { QW_TASK_DLOG("qExecTask end with empty res, useconds:%" PRIu64, useconds); dsEndPut(sinkHandle, useconds); - QW_ERR_RET(qwHandleTaskComplete(QW_FPARAMS(), ctx)); + QW_ERR_JRET(qwHandleTaskComplete(QW_FPARAMS(), ctx)); if (queryStop) { *queryStop = true; @@ -114,7 +114,7 @@ int32_t qwExecTask(QW_FPARAMS_DEF, SQWTaskCtx *ctx, bool *queryStop) { break; } - for(int32_t j = 0; j < taosArrayGetSize(pResList); ++j) { + for (int32_t j = 0; j < taosArrayGetSize(pResList); ++j) { SSDataBlock *pRes = taosArrayGetP(pResList, j); ASSERT(pRes->info.rows > 0); @@ -122,7 +122,7 @@ int32_t qwExecTask(QW_FPARAMS_DEF, SQWTaskCtx *ctx, bool *queryStop) { code = dsPutDataBlock(sinkHandle, &inputData, &qcontinue); if (code) { QW_TASK_ELOG("dsPutDataBlock failed, code:%x - %s", code, tstrerror(code)); - QW_ERR_RET(code); + QW_ERR_JRET(code); } QW_TASK_DLOG("data put into sink, rows:%d, continueExecTask:%d", pRes->info.rows, qcontinue); @@ -132,7 +132,7 @@ int32_t qwExecTask(QW_FPARAMS_DEF, SQWTaskCtx *ctx, bool *queryStop) { if (queryStop) { *queryStop = true; } - + break; } @@ -151,6 +151,11 @@ int32_t qwExecTask(QW_FPARAMS_DEF, SQWTaskCtx *ctx, bool *queryStop) { taosArrayDestroy(pResList); QW_RET(code); + +_return: + taosArrayDestroy(pResList); + + return code; } int32_t qwGenerateSchHbRsp(SQWorker *mgmt, SQWSchStatus *sch, SQWHbInfo *hbInfo) { @@ -222,7 +227,8 @@ int32_t qwGetQueryResFromSink(QW_FPARAMS_DEF, SQWTaskCtx *ctx, int32_t *dataLen, QW_ERR_RET(code); } - QW_TASK_DLOG("no more data in sink and query end, fetched blocks %d rows %d", pOutput->numOfBlocks, pOutput->numOfRows); + QW_TASK_DLOG("no more data in sink and query end, fetched blocks %d rows %d", pOutput->numOfBlocks, + pOutput->numOfRows); qwUpdateTaskStatus(QW_FPARAMS(), JOB_TASK_STATUS_SUCC); if (NULL == rsp) { @@ -266,7 +272,8 @@ int32_t qwGetQueryResFromSink(QW_FPARAMS_DEF, SQWTaskCtx *ctx, int32_t *dataLen, pOutput->numOfBlocks++; if (DS_BUF_EMPTY == pOutput->bufStatus && pOutput->queryEnd) { - QW_TASK_DLOG("task all data fetched and done, fetched blocks %d rows %d", pOutput->numOfBlocks, pOutput->numOfRows); + QW_TASK_DLOG("task all data fetched and done, fetched blocks %d rows %d", pOutput->numOfBlocks, + pOutput->numOfRows); qwUpdateTaskStatus(QW_FPARAMS(), JOB_TASK_STATUS_SUCC); break; } @@ -288,10 +295,10 @@ int32_t qwGetQueryResFromSink(QW_FPARAMS_DEF, SQWTaskCtx *ctx, int32_t *dataLen, } int32_t qwGetDeleteResFromSink(QW_FPARAMS_DEF, SQWTaskCtx *ctx, SDeleteRes *pRes) { - int64_t len = 0; - bool queryEnd = false; - int32_t code = 0; - SOutputData output = {0}; + int64_t len = 0; + bool queryEnd = false; + int32_t code = 0; + SOutputData output = {0}; dsGetDataLength(ctx->sinkHandle, &len, &queryEnd); @@ -304,7 +311,7 @@ int32_t qwGetDeleteResFromSink(QW_FPARAMS_DEF, SQWTaskCtx *ctx, SDeleteRes *pRes if (NULL == output.pData) { QW_ERR_RET(TSDB_CODE_OUT_OF_MEMORY); } - + code = dsGetDataBlock(ctx->sinkHandle, &output); if (code) { QW_TASK_ELOG("dsGetDataBlock failed, code:%x - %s", code, tstrerror(code)); @@ -312,8 +319,8 @@ int32_t qwGetDeleteResFromSink(QW_FPARAMS_DEF, SQWTaskCtx *ctx, SDeleteRes *pRes QW_ERR_RET(code); } - SDeleterRes* pDelRes = (SDeleterRes*)output.pData; - + SDeleterRes *pDelRes = (SDeleterRes *)output.pData; + pRes->suid = pDelRes->suid; pRes->uidList = pDelRes->uidList; pRes->skey = pDelRes->skey; @@ -322,14 +329,13 @@ int32_t qwGetDeleteResFromSink(QW_FPARAMS_DEF, SQWTaskCtx *ctx, SDeleteRes *pRes strcpy(pRes->tableFName, pDelRes->tableName); strcpy(pRes->tsColName, pDelRes->tsColName); taosMemoryFree(output.pData); - + return TSDB_CODE_SUCCESS; } - int32_t qwHandlePrePhaseEvents(QW_FPARAMS_DEF, int8_t phase, SQWPhaseInput *input, SQWPhaseOutput *output) { - int32_t code = 0; - SQWTaskCtx *ctx = NULL; + int32_t code = 0; + SQWTaskCtx *ctx = NULL; QW_TASK_DLOG("start to handle event at phase %s", qwPhaseStr(phase)); @@ -355,8 +361,8 @@ int32_t qwHandlePrePhaseEvents(QW_FPARAMS_DEF, int8_t phase, SQWPhaseInput *inpu if (QW_EVENT_RECEIVED(ctx, QW_EVENT_DROP)) { QW_ERR_JRET(qwDropTask(QW_FPARAMS())); - //qwBuildAndSendDropRsp(&ctx->ctrlConnInfo, code); - //QW_TASK_DLOG("drop rsp send, handle:%p, code:%x - %s", ctx->ctrlConnInfo.handle, code, tstrerror(code)); + // qwBuildAndSendDropRsp(&ctx->ctrlConnInfo, code); + // QW_TASK_DLOG("drop rsp send, handle:%p, code:%x - %s", ctx->ctrlConnInfo.handle, code, tstrerror(code)); QW_ERR_JRET(TSDB_CODE_QRY_TASK_DROPPED); break; @@ -391,8 +397,8 @@ int32_t qwHandlePrePhaseEvents(QW_FPARAMS_DEF, int8_t phase, SQWPhaseInput *inpu if (QW_EVENT_RECEIVED(ctx, QW_EVENT_DROP)) { QW_ERR_JRET(qwDropTask(QW_FPARAMS())); - //qwBuildAndSendDropRsp(&ctx->ctrlConnInfo, code); - //QW_TASK_DLOG("drop rsp send, handle:%p, code:%x - %s", ctx->ctrlConnInfo.handle, code, tstrerror(code)); + // qwBuildAndSendDropRsp(&ctx->ctrlConnInfo, code); + // QW_TASK_DLOG("drop rsp send, handle:%p, code:%x - %s", ctx->ctrlConnInfo.handle, code, tstrerror(code)); QW_ERR_JRET(TSDB_CODE_QRY_TASK_DROPPED); } @@ -428,9 +434,9 @@ _return: } int32_t qwHandlePostPhaseEvents(QW_FPARAMS_DEF, int8_t phase, SQWPhaseInput *input, SQWPhaseOutput *output) { - int32_t code = 0; - SQWTaskCtx *ctx = NULL; - SRpcHandleInfo connInfo = {0}; + int32_t code = 0; + SQWTaskCtx *ctx = NULL; + SRpcHandleInfo connInfo = {0}; QW_TASK_DLOG("start to handle event at phase %s", qwPhaseStr(phase)); @@ -449,8 +455,8 @@ int32_t qwHandlePostPhaseEvents(QW_FPARAMS_DEF, int8_t phase, SQWPhaseInput *inp QW_ERR_JRET(TSDB_CODE_QRY_APP_ERROR); } - //qwBuildAndSendDropRsp(&ctx->ctrlConnInfo, code); - //QW_TASK_DLOG("drop rsp send, handle:%p, code:%x - %s", ctx->ctrlConnInfo.handle, code, tstrerror(code)); + // qwBuildAndSendDropRsp(&ctx->ctrlConnInfo, code); + // QW_TASK_DLOG("drop rsp send, handle:%p, code:%x - %s", ctx->ctrlConnInfo.handle, code, tstrerror(code)); QW_ERR_JRET(qwDropTask(QW_FPARAMS())); QW_ERR_JRET(TSDB_CODE_QRY_TASK_DROPPED); @@ -473,14 +479,14 @@ _return: if (QW_PHASE_POST_QUERY == phase && ctx) { ctx->queryRsped = true; - bool rsped = false; + bool rsped = false; SQWMsg qwMsg = {.msgType = ctx->msgType, .connInfo = ctx->ctrlConnInfo}; qwDbgSimulateRedirect(&qwMsg, ctx, &rsped); qwDbgSimulateDead(QW_FPARAMS(), ctx, &rsped); if (!rsped) { qwBuildAndSendQueryRsp(input->msgType + 1, &ctx->ctrlConnInfo, code, ctx); QW_TASK_DLOG("query msg rsped, handle:%p, code:%x - %s", ctx->ctrlConnInfo.handle, code, tstrerror(code)); - } + } } if (ctx) { @@ -507,7 +513,6 @@ int32_t qwAbortPrerocessQuery(QW_FPARAMS_DEF) { QW_RET(TSDB_CODE_SUCCESS); } - int32_t qwPreprocessQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg) { int32_t code = 0; bool queryRsped = false; @@ -537,8 +542,7 @@ _return: QW_RET(TSDB_CODE_SUCCESS); } - -int32_t qwProcessQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg, char* sql) { +int32_t qwProcessQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg, char *sql) { int32_t code = 0; bool queryRsped = false; SSubplan *plan = NULL; @@ -556,7 +560,7 @@ int32_t qwProcessQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg, char* sql) { ctx->needFetch = qwMsg->msgInfo.needFetch; ctx->msgType = qwMsg->msgType; - //QW_TASK_DLOGL("subplan json string, len:%d, %s", qwMsg->msgLen, qwMsg->msg); + // QW_TASK_DLOGL("subplan json string, len:%d, %s", qwMsg->msgLen, qwMsg->msg); code = qStringToSubplan(qwMsg->msg, &plan); if (TSDB_CODE_SUCCESS != code) { @@ -594,7 +598,7 @@ int32_t qwProcessQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg, char* sql) { _return: taosMemoryFree(sql); - + input.code = code; input.msgType = qwMsg->msgType; code = qwHandlePostPhaseEvents(QW_FPARAMS(), QW_PHASE_POST_QUERY, &input, NULL); @@ -648,7 +652,7 @@ int32_t qwProcessCQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg) { qwBuildAndSendFetchRsp(ctx->fetchType, &qwMsg->connInfo, rsp, dataLen, code); rsp = NULL; - + QW_TASK_DLOG("fetch rsp send, handle:%p, code:%x - %s, dataLen:%d", qwMsg->connInfo.handle, code, tstrerror(code), dataLen); } else { @@ -754,13 +758,13 @@ _return: if (code || rsp) { bool rsped = false; if (ctx) { - qwDbgSimulateRedirect(qwMsg, ctx, &rsped); + qwDbgSimulateRedirect(qwMsg, ctx, &rsped); qwDbgSimulateDead(QW_FPARAMS(), ctx, &rsped); } if (!rsped) { qwBuildAndSendFetchRsp(qwMsg->msgType + 1, &qwMsg->connInfo, rsp, dataLen, code); - QW_TASK_DLOG("%s send, handle:%p, code:%x - %s, dataLen:%d", TMSG_INFO(qwMsg->msgType + 1), qwMsg->connInfo.handle, code, tstrerror(code), - dataLen); + QW_TASK_DLOG("%s send, handle:%p, code:%x - %s, dataLen:%d", TMSG_INFO(qwMsg->msgType + 1), + qwMsg->connInfo.handle, code, tstrerror(code), dataLen); } } @@ -919,10 +923,11 @@ void qwProcessHbTimerEvent(void *param, void *tmrId) { uint64_t *sId = taosHashGetKey(pIter, NULL); QW_TLOG("cancel send hb to sch %" PRIx64 " cause of no connection handle", *sId); - if (sch->hbBrokenTs > 0 && ((currentMs - sch->hbBrokenTs) > QW_SCH_TIMEOUT_MSEC) && taosHashGetSize(sch->tasksHash) <= 0) { + if (sch->hbBrokenTs > 0 && ((currentMs - sch->hbBrokenTs) > QW_SCH_TIMEOUT_MSEC) && + taosHashGetSize(sch->tasksHash) <= 0) { taosArrayPush(pExpiredSch, sId); } - + pIter = taosHashIterate(mgmt->schHash, pIter); continue; } @@ -998,7 +1003,6 @@ _return: QW_RET(TSDB_CODE_SUCCESS); } - int32_t qWorkerInit(int8_t nodeType, int32_t nodeId, SQWorkerCfg *cfg, void **qWorkerMgmt, const SMsgCb *pMsgCb) { if (NULL == qWorkerMgmt || pMsgCb->mgmt == NULL) { qError("invalid param to init qworker"); @@ -1119,12 +1123,12 @@ int32_t qWorkerGetStat(SReadHandle *handle, void *qWorkerMgmt, SQWorkerStat *pSt QW_RET(TSDB_CODE_QRY_INVALID_INPUT); } - SQWorker *mgmt = (SQWorker *)qWorkerMgmt; + SQWorker *mgmt = (SQWorker *)qWorkerMgmt; SDataSinkStat sinkStat = {0}; - + dsDataSinkGetCacheSize(&sinkStat); pStat->cacheDataSize = sinkStat.cachedSize; - + pStat->queryProcessed = QW_STAT_GET(mgmt->stat.msgStat.queryProcessed); pStat->cqueryProcessed = QW_STAT_GET(mgmt->stat.msgStat.cqueryProcessed); pStat->fetchProcessed = QW_STAT_GET(mgmt->stat.msgStat.fetchProcessed); @@ -1139,6 +1143,3 @@ int32_t qWorkerGetStat(SReadHandle *handle, void *qWorkerMgmt, SQWorkerStat *pSt return TSDB_CODE_SUCCESS; } - - -