From 94ab74cc7f3d1738381aecc125347439744b65e6 Mon Sep 17 00:00:00 2001 From: 54liuyao <54liuyao> Date: Wed, 17 Jan 2024 14:47:54 +0800 Subject: [PATCH 1/3] clone task id --- source/libs/executor/src/exchangeoperator.c | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/source/libs/executor/src/exchangeoperator.c b/source/libs/executor/src/exchangeoperator.c index 2797cb2d82..140b390913 100644 --- a/source/libs/executor/src/exchangeoperator.c +++ b/source/libs/executor/src/exchangeoperator.c @@ -37,7 +37,7 @@ typedef struct SSourceDataInfo { int64_t startTime; int32_t code; EX_SOURCE_STATUS status; - const char* taskId; + char* taskId; SArray* pSrcUidList; int32_t srcOpType; bool tableSeq; @@ -260,14 +260,16 @@ static int32_t initDataSource(int32_t numOfSources, SExchangeInfo* pInfo, const return TSDB_CODE_SUCCESS; } + int32_t len = strlen(id); for (int32_t i = 0; i < numOfSources; ++i) { SSourceDataInfo dataInfo = {0}; dataInfo.status = EX_SOURCE_DATA_NOT_READY; - dataInfo.taskId = id; + dataInfo.taskId = taosMemoryCalloc(1, len); + memcpy(dataInfo.taskId, id, len); dataInfo.index = i; SSourceDataInfo* pDs = taosArrayPush(pInfo->pSourceDataInfo, &dataInfo); if (pDs == NULL) { - taosArrayDestroy(pInfo->pSourceDataInfo); + taosArrayDestroyEx(pInfo->pSourceDataInfo, freeSourceDataInfo); return TSDB_CODE_OUT_OF_MEMORY; } } @@ -368,6 +370,7 @@ void freeBlock(void* pParam) { void freeSourceDataInfo(void* p) { SSourceDataInfo* pInfo = (SSourceDataInfo*)p; taosMemoryFreeClear(pInfo->pRsp); + taosMemoryFreeClear(pInfo->taskId); } void doDestroyExchangeOperatorInfo(void* param) { @@ -782,7 +785,10 @@ int32_t addSingleExchangeSource(SOperatorInfo* pOperator, SExchangeOperatorBasic if (pIdx->inUseIdx < 0) { SSourceDataInfo dataInfo = {0}; dataInfo.status = EX_SOURCE_DATA_NOT_READY; - dataInfo.taskId = GET_TASKID(pOperator->pTaskInfo); + char* pTaskId = GET_TASKID(pOperator->pTaskInfo); + int32_t len = strlen(pTaskId); + dataInfo.taskId = taosMemoryCalloc(1, len); + memcpy(dataInfo.taskId, pTaskId, len); dataInfo.index = pIdx->srcIdx; dataInfo.pSrcUidList = taosArrayDup(pBasicParam->uidList, NULL); dataInfo.srcOpType = pBasicParam->srcOpType; From 1309e56a95eb0a38cd69254276da608c89d7cfe1 Mon Sep 17 00:00:00 2001 From: 54liuyao <54liuyao> Date: Wed, 17 Jan 2024 16:52:47 +0800 Subject: [PATCH 2/3] use strncpy --- source/libs/executor/src/exchangeoperator.c | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/source/libs/executor/src/exchangeoperator.c b/source/libs/executor/src/exchangeoperator.c index 140b390913..ac103ebbc1 100644 --- a/source/libs/executor/src/exchangeoperator.c +++ b/source/libs/executor/src/exchangeoperator.c @@ -260,12 +260,12 @@ static int32_t initDataSource(int32_t numOfSources, SExchangeInfo* pInfo, const return TSDB_CODE_SUCCESS; } - int32_t len = strlen(id); + int32_t len = strlen(id) + 1; for (int32_t i = 0; i < numOfSources; ++i) { SSourceDataInfo dataInfo = {0}; dataInfo.status = EX_SOURCE_DATA_NOT_READY; dataInfo.taskId = taosMemoryCalloc(1, len); - memcpy(dataInfo.taskId, id, len); + strncpy(dataInfo.taskId, id, len); dataInfo.index = i; SSourceDataInfo* pDs = taosArrayPush(pInfo->pSourceDataInfo, &dataInfo); if (pDs == NULL) { @@ -786,9 +786,9 @@ int32_t addSingleExchangeSource(SOperatorInfo* pOperator, SExchangeOperatorBasic SSourceDataInfo dataInfo = {0}; dataInfo.status = EX_SOURCE_DATA_NOT_READY; char* pTaskId = GET_TASKID(pOperator->pTaskInfo); - int32_t len = strlen(pTaskId); + int32_t len = strlen(pTaskId) + 1; dataInfo.taskId = taosMemoryCalloc(1, len); - memcpy(dataInfo.taskId, pTaskId, len); + strncpy(dataInfo.taskId, pTaskId, len); dataInfo.index = pIdx->srcIdx; dataInfo.pSrcUidList = taosArrayDup(pBasicParam->uidList, NULL); dataInfo.srcOpType = pBasicParam->srcOpType; From 7986951f440e9cbf585d1238ab077881b1b24e7f Mon Sep 17 00:00:00 2001 From: 54liuyao <54liuyao> Date: Thu, 18 Jan 2024 09:22:39 +0800 Subject: [PATCH 3/3] use taskid of operator --- source/libs/executor/inc/executorInt.h | 1 + source/libs/executor/src/exchangeoperator.c | 15 +++++++-------- 2 files changed, 8 insertions(+), 8 deletions(-) diff --git a/source/libs/executor/inc/executorInt.h b/source/libs/executor/inc/executorInt.h index c3f47cde9d..72da249f50 100644 --- a/source/libs/executor/inc/executorInt.h +++ b/source/libs/executor/inc/executorInt.h @@ -200,6 +200,7 @@ typedef struct SExchangeInfo { uint64_t self; SLimitInfo limitInfo; int64_t openedTs; // start exec time stamp, todo: move to SLoadRemoteDataInfo + char* pTaskId; } SExchangeInfo; typedef struct SScanInfo { diff --git a/source/libs/executor/src/exchangeoperator.c b/source/libs/executor/src/exchangeoperator.c index ac103ebbc1..06dd43e170 100644 --- a/source/libs/executor/src/exchangeoperator.c +++ b/source/libs/executor/src/exchangeoperator.c @@ -37,7 +37,7 @@ typedef struct SSourceDataInfo { int64_t startTime; int32_t code; EX_SOURCE_STATUS status; - char* taskId; + const char* taskId; SArray* pSrcUidList; int32_t srcOpType; bool tableSeq; @@ -261,11 +261,12 @@ static int32_t initDataSource(int32_t numOfSources, SExchangeInfo* pInfo, const } int32_t len = strlen(id) + 1; + pInfo->pTaskId = taosMemoryCalloc(1, len); + strncpy(pInfo->pTaskId, id, len); for (int32_t i = 0; i < numOfSources; ++i) { SSourceDataInfo dataInfo = {0}; dataInfo.status = EX_SOURCE_DATA_NOT_READY; - dataInfo.taskId = taosMemoryCalloc(1, len); - strncpy(dataInfo.taskId, id, len); + dataInfo.taskId = pInfo->pTaskId; dataInfo.index = i; SSourceDataInfo* pDs = taosArrayPush(pInfo->pSourceDataInfo, &dataInfo); if (pDs == NULL) { @@ -370,7 +371,6 @@ void freeBlock(void* pParam) { void freeSourceDataInfo(void* p) { SSourceDataInfo* pInfo = (SSourceDataInfo*)p; taosMemoryFreeClear(pInfo->pRsp); - taosMemoryFreeClear(pInfo->taskId); } void doDestroyExchangeOperatorInfo(void* param) { @@ -386,6 +386,8 @@ void doDestroyExchangeOperatorInfo(void* param) { tSimpleHashCleanup(pExInfo->pHashSources); tsem_destroy(&pExInfo->ready); + taosMemoryFreeClear(pExInfo->pTaskId); + taosMemoryFreeClear(param); } @@ -785,10 +787,7 @@ int32_t addSingleExchangeSource(SOperatorInfo* pOperator, SExchangeOperatorBasic if (pIdx->inUseIdx < 0) { SSourceDataInfo dataInfo = {0}; dataInfo.status = EX_SOURCE_DATA_NOT_READY; - char* pTaskId = GET_TASKID(pOperator->pTaskInfo); - int32_t len = strlen(pTaskId) + 1; - dataInfo.taskId = taosMemoryCalloc(1, len); - strncpy(dataInfo.taskId, pTaskId, len); + dataInfo.taskId = pExchangeInfo->pTaskId; dataInfo.index = pIdx->srcIdx; dataInfo.pSrcUidList = taosArrayDup(pBasicParam->uidList, NULL); dataInfo.srcOpType = pBasicParam->srcOpType;