diff --git a/source/client/inc/clientInt.h b/source/client/inc/clientInt.h index 0dc0a31afe..32f79599f8 100644 --- a/source/client/inc/clientInt.h +++ b/source/client/inc/clientInt.h @@ -205,8 +205,7 @@ typedef struct SRequestSendRecvBody { __taos_async_fn_t queryFp; __taos_async_fn_t fetchFp; EQueryExecMode execMode; - void* param; - bool paramCreatedInternal; + void* interParam; SDataBuf requestMsg; int64_t queryJob; // query job, created according to sql query DAG. int32_t subplanNum; @@ -285,6 +284,7 @@ typedef struct SRequestObj { typedef struct SSyncQueryParam { tsem_t sem; SRequestObj* pRequest; + void* userParam; } SSyncQueryParam; void* doAsyncFetchRows(SRequestObj* pRequest, bool setupOneRowPtr, bool convertUcs4); diff --git a/source/client/src/clientEnv.c b/source/client/src/clientEnv.c index 16e90fc9c3..fb665aa96b 100644 --- a/source/client/src/clientEnv.c +++ b/source/client/src/clientEnv.c @@ -316,6 +316,15 @@ void *createRequest(uint64_t connId, int32_t type, int64_t reqid) { terrno = TSDB_CODE_TSC_DISCONNECTED; return NULL; } + SSyncQueryParam *interParam = taosMemoryCalloc(1, sizeof(SSyncQueryParam)); + if (interParam == NULL) { + doDestroyRequest(pRequest); + terrno = TSDB_CODE_OUT_OF_MEMORY; + return NULL; + } + tsem_init(&interParam->sem, 0, 0); + interParam->pRequest = pRequest; + pRequest->body.interParam = interParam; pRequest->resType = RES_TYPE__QUERY; pRequest->requestId = reqid == 0 ? generateRequestId() : reqid; @@ -437,12 +446,10 @@ void doDestroyRequest(void *p) { deregisterRequest(pRequest); } - if (pRequest->syncQuery || pRequest->body.paramCreatedInternal) { - if (pRequest->body.param) { - tsem_destroy(&((SSyncQueryParam *)pRequest->body.param)->sem); - } - taosMemoryFree(pRequest->body.param); + if (pRequest->body.interParam) { + tsem_destroy(&((SSyncQueryParam *)pRequest->body.interParam)->sem); } + taosMemoryFree(pRequest->body.interParam); qDestroyQuery(pRequest->pQuery); nodesDestroyAllocator(pRequest->allocatorRefId); diff --git a/source/client/src/clientImpl.c b/source/client/src/clientImpl.c index a0eebe019d..fe8180fc31 100644 --- a/source/client/src/clientImpl.c +++ b/source/client/src/clientImpl.c @@ -196,22 +196,7 @@ int32_t buildRequest(uint64_t connId, const char* sql, int sqlLen, void* param, (*pRequest)->sqlLen = sqlLen; (*pRequest)->validateOnly = validateSql; - SSyncQueryParam* newpParam = NULL; - if (param == NULL) { - newpParam = taosMemoryCalloc(1, sizeof(SSyncQueryParam)); - if (newpParam == NULL) { - destroyRequest(*pRequest); - *pRequest = NULL; - return TSDB_CODE_OUT_OF_MEMORY; - } - - tsem_init(&newpParam->sem, 0, 0); - newpParam->pRequest = (*pRequest); - param = newpParam; - (*pRequest)->body.paramCreatedInternal = true; - } - - (*pRequest)->body.param = param; + ((SSyncQueryParam*)(*pRequest)->body.interParam)->userParam = param; STscObj* pTscObj = (*pRequest)->pTscObj; int32_t err = taosHashPut(pTscObj->pRequests, &(*pRequest)->self, sizeof((*pRequest)->self), &(*pRequest)->self, @@ -219,7 +204,6 @@ int32_t buildRequest(uint64_t connId, const char* sql, int sqlLen, void* param, if (err) { tscError("%" PRId64 " failed to add to request container, reqId:0x%" PRIx64 ", conn:%" PRId64 ", %s", (*pRequest)->self, (*pRequest)->requestId, pTscObj->id, sql); - freeQueryParam(newpParam); destroyRequest(*pRequest); *pRequest = NULL; return TSDB_CODE_OUT_OF_MEMORY; @@ -231,7 +215,6 @@ int32_t buildRequest(uint64_t connId, const char* sql, int sqlLen, void* param, nodesCreateAllocator((*pRequest)->requestId, tsQueryNodeChunkSize, &((*pRequest)->allocatorRefId))) { tscError("%" PRId64 " failed to create node allocator, reqId:0x%" PRIx64 ", conn:%" PRId64 ", %s", (*pRequest)->self, (*pRequest)->requestId, pTscObj->id, sql); - freeQueryParam(newpParam); destroyRequest(*pRequest); *pRequest = NULL; return TSDB_CODE_OUT_OF_MEMORY; @@ -1684,8 +1667,8 @@ void* doFetchRows(SRequestObj* pRequest, bool setupOneRowPtr, bool convertUcs4) } static void syncFetchFn(void* param, TAOS_RES* res, int32_t numOfRows) { - SSyncQueryParam* pParam = param; - tsem_post(&pParam->sem); + tsem_t* sem = param; + tsem_post(sem); } void* doAsyncFetchRows(SRequestObj* pRequest, bool setupOneRowPtr, bool convertUcs4) { @@ -1703,10 +1686,11 @@ void* doAsyncFetchRows(SRequestObj* pRequest, bool setupOneRowPtr, bool convertU // convert ucs4 to native multi-bytes string pResultInfo->convertUcs4 = convertUcs4; - - SSyncQueryParam* pParam = pRequest->body.param; - taos_fetch_rows_a(pRequest, syncFetchFn, pParam); - tsem_wait(&pParam->sem); + tsem_t sem; + tsem_init(&sem, 0, 0); + taos_fetch_rows_a(pRequest, syncFetchFn, &sem); + tsem_wait(&sem); + tsem_destroy(&sem); } if (pResultInfo->numOfRows == 0 || pRequest->code != TSDB_CODE_SUCCESS) { @@ -2473,6 +2457,10 @@ TAOS_RES* taosQueryImpl(TAOS* taos, const char* sql, bool validateOnly) { tscDebug("taos_query start with sql:%s", sql); SSyncQueryParam* param = taosMemoryCalloc(1, sizeof(SSyncQueryParam)); + if (NULL == param) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + return NULL; + } tsem_init(¶m->sem, 0, 0); taosAsyncQueryImpl(*(int64_t*)taos, sql, syncQueryFn, param, validateOnly); @@ -2482,9 +2470,8 @@ TAOS_RES* taosQueryImpl(TAOS* taos, const char* sql, bool validateOnly) { if (param->pRequest != NULL) { param->pRequest->syncQuery = true; pRequest = param->pRequest; - } else { - taosMemoryFree(param); } + taosMemoryFree(param); tscDebug("taos_query end with sql:%s", sql); @@ -2498,19 +2485,20 @@ TAOS_RES* taosQueryImplWithReqid(TAOS* taos, const char* sql, bool validateOnly, } SSyncQueryParam* param = taosMemoryCalloc(1, sizeof(SSyncQueryParam)); + if (param == NULL) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + return NULL; + } tsem_init(¶m->sem, 0, 0); taosAsyncQueryImplWithReqid(*(int64_t*)taos, sql, syncQueryFn, param, validateOnly, reqid); tsem_wait(¶m->sem); - SRequestObj* pRequest = NULL; if (param->pRequest != NULL) { param->pRequest->syncQuery = true; pRequest = param->pRequest; - } else { - taosMemoryFree(param); } - + taosMemoryFree(param); return pRequest; } @@ -2528,13 +2516,13 @@ static void fetchCallback(void* pResult, void* param, int32_t code) { if (code != TSDB_CODE_SUCCESS) { pRequest->code = code; taosMemoryFreeClear(pResultInfo->pData); - pRequest->body.fetchFp(pRequest->body.param, pRequest, 0); + pRequest->body.fetchFp(((SSyncQueryParam *)pRequest->body.interParam)->userParam, pRequest, 0); return; } if (pRequest->code != TSDB_CODE_SUCCESS) { taosMemoryFreeClear(pResultInfo->pData); - pRequest->body.fetchFp(pRequest->body.param, pRequest, 0); + pRequest->body.fetchFp(((SSyncQueryParam *)pRequest->body.interParam)->userParam, pRequest, 0); return; } @@ -2555,21 +2543,12 @@ static void fetchCallback(void* pResult, void* param, int32_t code) { atomic_add_fetch_64((int64_t*)&pActivity->fetchBytes, pRequest->body.resInfo.payloadLen); } - pRequest->body.fetchFp(pRequest->body.param, pRequest, pResultInfo->numOfRows); + pRequest->body.fetchFp(((SSyncQueryParam *)pRequest->body.interParam)->userParam, pRequest, pResultInfo->numOfRows); } void taosAsyncFetchImpl(SRequestObj* pRequest, __taos_async_fn_t fp, void* param) { - if ((pRequest->syncQuery || pRequest->body.paramCreatedInternal) && pRequest->body.param != param) { - if (pRequest->body.param) { - tsem_destroy(&((SSyncQueryParam *)pRequest->body.param)->sem); - } - taosMemoryFree(pRequest->body.param); - pRequest->syncQuery = false; - pRequest->body.paramCreatedInternal = false; - } - pRequest->body.fetchFp = fp; - pRequest->body.param = param; + ((SSyncQueryParam *)pRequest->body.interParam)->userParam = param; SReqResultInfo* pResultInfo = &pRequest->body.resInfo; @@ -2608,9 +2587,5 @@ void taosAsyncFetchImpl(SRequestObj* pRequest, __taos_async_fn_t fp, void* param } void doRequestCallback(SRequestObj* pRequest, int32_t code) { - if (pRequest->body.paramCreatedInternal) { - pRequest->body.queryFp(NULL, pRequest, code); - } else { - pRequest->body.queryFp(pRequest->body.param, pRequest, code); - } + pRequest->body.queryFp(((SSyncQueryParam *)pRequest->body.interParam)->userParam, pRequest, code); } diff --git a/source/client/src/clientMain.c b/source/client/src/clientMain.c index 46bdc4e2ad..c8e8eaa602 100644 --- a/source/client/src/clientMain.c +++ b/source/client/src/clientMain.c @@ -1561,12 +1561,12 @@ int taos_load_table_info(TAOS *taos, const char *tableNameList) { conn.mgmtEps = getEpSet_s(&pTscObj->pAppInfo->mgmtEp); - code = catalogAsyncGetAllMeta(pCtg, &conn, &catalogReq, syncCatalogFn, pRequest->body.param, NULL); + code = catalogAsyncGetAllMeta(pCtg, &conn, &catalogReq, syncCatalogFn, pRequest->body.interParam, NULL); if (code) { goto _return; } - SSyncQueryParam *pParam = pRequest->body.param; + SSyncQueryParam *pParam = pRequest->body.interParam; tsem_wait(&pParam->sem); _return: diff --git a/source/libs/scheduler/src/schRemote.c b/source/libs/scheduler/src/schRemote.c index 7b8decc007..f987420ece 100644 --- a/source/libs/scheduler/src/schRemote.c +++ b/source/libs/scheduler/src/schRemote.c @@ -455,7 +455,6 @@ int32_t schHandleCallback(void *param, SDataBuf *pMsg, int32_t rspCode) { tstrerror(rspCode)); SCH_ERR_JRET(schProcessOnCbBegin(&pJob, &pTask, pParam->queryId, pParam->refId, pParam->taskId)); - code = schHandleResponseMsg(pJob, pTask, pParam->execId, pMsg, rspCode); pMsg->pData = NULL; diff --git a/tests/taosc_test/taoscTest.cpp b/tests/taosc_test/taoscTest.cpp index 137f607a32..fbdb152ab4 100644 --- a/tests/taosc_test/taoscTest.cpp +++ b/tests/taosc_test/taoscTest.cpp @@ -188,10 +188,9 @@ TEST_F(taoscTest, taos_query_test) { void queryCallback2(void* param, void* res, int32_t code) { ASSERT_TRUE(code == 0); ASSERT_TRUE(param == pUserParam); - // After using taos_query_a to query, using taos_fetch_row - // in the callback will cause blocking. - /* - TAOS_ROW row; + // After using taos_query_a to query, using taos_fetch_row in the callback will cause blocking. + // Reason: schProcessOnCbBegin SCH_LOCK_TASK(pTask) + /* TAOS_ROW row; while ((row = taos_fetch_row(res))) { getRecordCounts++; } */ @@ -251,14 +250,14 @@ TEST_F(taoscTest, taos_query_a_fetch_row) { getRecordCounts = 0; TAOS_ROW row; printf("taos_query_a_fetch_row taos_fetch_row start...\n"); - // will cause heap-buffer-overfow - // while ((row = taos_fetch_row(*pres))) { - // getRecordCounts++; - // } + + while ((row = taos_fetch_row(*pres))) { + getRecordCounts++; + } printf("taos_query_a_fetch_row taos_fetch_row end. %p record count:%d.\n", *pres, getRecordCounts); taos_free_result(*pres); - ASSERT_NE(getRecordCounts, insertCounts); + ASSERT_EQ(getRecordCounts, insertCounts); taos_close(taos); printf("taos_query_a_fetch_row test finished.\n");