diff --git a/include/common/tmsg.h b/include/common/tmsg.h index 3281bca96a5c9007747b78e691c0c80c6df6adc0..afe54bce0b1accf5c035398ca981c64911c8f60e 100644 --- a/include/common/tmsg.h +++ b/include/common/tmsg.h @@ -2627,6 +2627,7 @@ typedef struct { int32_t tEncodeSTqCheckInfo(SEncoder* pEncoder, const STqCheckInfo* pInfo); int32_t tDecodeSTqCheckInfo(SDecoder* pDecoder, STqCheckInfo* pInfo); +void tDeleteSTqCheckInfo(STqCheckInfo* pInfo); typedef struct { char topic[TSDB_TOPIC_FNAME_LEN]; diff --git a/source/common/src/tmsg.c b/source/common/src/tmsg.c index bd8e34a395f94e93873123eba46a62c14fbe0da4..fd445ff2aec71654fa5893cd0d847a0a6d73cc3e 100644 --- a/source/common/src/tmsg.c +++ b/source/common/src/tmsg.c @@ -5903,6 +5903,7 @@ int32_t tDecodeSTqCheckInfo(SDecoder *pDecoder, STqCheckInfo *pInfo) { } return 0; } +void tDeleteSTqCheckInfo(STqCheckInfo *pInfo) { taosArrayDestroy(pInfo->colIdList); } int32_t tEncodeDeleteRes(SEncoder *pCoder, const SDeleteRes *pRes) { int32_t nUid = taosArrayGetSize(pRes->uidList); diff --git a/source/dnode/mnode/impl/src/mndSubscribe.c b/source/dnode/mnode/impl/src/mndSubscribe.c index 71e0b09e022d73a4c3e6a0ea3dd871e2d1de4e93..14adeb00808234e4afd454e4a1a6fc48cad4300e 100644 --- a/source/dnode/mnode/impl/src/mndSubscribe.c +++ b/source/dnode/mnode/impl/src/mndSubscribe.c @@ -235,6 +235,7 @@ static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqR taosHashPut(pHash, &pVgEp->vgId, sizeof(int32_t), &outputVg, sizeof(SMqRebOutputVg)); mInfo("mq rebalance: remove vgId:%d from consumer:%" PRId64, pVgEp->vgId, consumerId); } + taosArrayDestroy(pConsumerEp->vgs); taosHashRemove(pOutput->pSub->consumerHash, &consumerId, sizeof(int64_t)); // put into removed taosArrayPush(pOutput->removedConsumers, &consumerId); diff --git a/source/dnode/vnode/src/inc/tq.h b/source/dnode/vnode/src/inc/tq.h index ef54adf0d6502abe2817d6331f70b0f2a2f932e7..8ae1c8720ca6a37ba46e942febc0433dd091d282 100644 --- a/source/dnode/vnode/src/inc/tq.h +++ b/source/dnode/vnode/src/inc/tq.h @@ -88,8 +88,7 @@ typedef struct { STqExecTb execTb; STqExecDb execDb; }; - int32_t numOfCols; // number of out pout column, temporarily used - SSchemaWrapper* pSchemaWrapper; // columns that are involved in query + int32_t numOfCols; // number of out pout column, temporarily used } STqExecHandle; typedef struct { diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c index f75dce8231e5b3fd41758371fabe2c5f3af9e6b4..eafbd8464daab011d1209bb68453a8882b6ed903 100644 --- a/source/dnode/vnode/src/tq/tq.c +++ b/source/dnode/vnode/src/tq/tq.c @@ -55,6 +55,7 @@ static void destroySTqHandle(void* data) { STqHandle* pData = (STqHandle*)data; qDestroyTask(pData->execHandle.task); if (pData->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) { + taosMemoryFreeClear(pData->execHandle.execCol.qmsg); } else if (pData->execHandle.subType == TOPIC_SUB_TYPE__DB) { tqCloseReader(pData->execHandle.pExecReader); walCloseReader(pData->pWalReader); @@ -80,7 +81,6 @@ STQ* tqOpen(const char* path, SVnode* pVnode) { pTq->pVnode = pVnode; pTq->pHandle = taosHashInit(64, MurmurHash3_32, true, HASH_ENTRY_LOCK); - taosHashSetFreeFp(pTq->pHandle, destroySTqHandle); taosInitRWLatch(&pTq->pushLock); @@ -88,6 +88,7 @@ STQ* tqOpen(const char* path, SVnode* pVnode) { taosHashSetFreeFp(pTq->pPushMgr, tqPushEntryFree); pTq->pCheckInfo = taosHashInit(64, MurmurHash3_32, true, HASH_ENTRY_LOCK); + taosHashSetFreeFp(pTq->pCheckInfo, (FDelete)tDeleteSTqCheckInfo); if (tqMetaOpen(pTq) < 0) { ASSERT(0); @@ -779,6 +780,7 @@ int32_t tqProcessSubscribeReq(STQ* pTq, int64_t version, char* msg, int32_t msgL } if (req.newConsumerId == -1) { tqError("vgId:%d, tq invalid rebalance request, new consumerId %" PRId64 "", req.vgId, req.newConsumerId); + taosMemoryFree(req.qmsg); return 0; } STqHandle tqHandle = {0}; @@ -815,8 +817,7 @@ int32_t tqProcessSubscribeReq(STQ* pTq, int64_t version, char* msg, int32_t msgL req.qmsg = NULL; pHandle->execHandle.task = - qCreateQueueExecTaskInfo(pHandle->execHandle.execCol.qmsg, &handle, &pHandle->execHandle.numOfCols, - &pHandle->execHandle.pSchemaWrapper); + qCreateQueueExecTaskInfo(pHandle->execHandle.execCol.qmsg, &handle, &pHandle->execHandle.numOfCols, NULL); ASSERT(pHandle->execHandle.task); void* scanner = NULL; qExtractStreamScanner(pHandle->execHandle.task, &scanner); @@ -864,6 +865,7 @@ int32_t tqProcessSubscribeReq(STQ* pTq, int64_t version, char* msg, int32_t msgL atomic_store_32(&pHandle->epoch, -1); atomic_store_64(&pHandle->consumerId, req.newConsumerId); atomic_add_fetch_32(&pHandle->epoch, 1); + taosMemoryFree(req.qmsg); if (tqMetaSaveHandle(pTq, req.subKey, pHandle) < 0) { // TODO ASSERT(0); diff --git a/source/dnode/vnode/src/tq/tqMeta.c b/source/dnode/vnode/src/tq/tqMeta.c index 27c491c86b432477b4174edc6f5f842e1a5a843b..b021c5ee7fb1abc762898810c3eb7f192da8235d 100644 --- a/source/dnode/vnode/src/tq/tqMeta.c +++ b/source/dnode/vnode/src/tq/tqMeta.c @@ -305,8 +305,8 @@ int32_t tqMetaRestoreHandle(STQ* pTq) { }; if (handle.execHandle.subType == TOPIC_SUB_TYPE__COLUMN) { - handle.execHandle.task = qCreateQueueExecTaskInfo( - handle.execHandle.execCol.qmsg, &reader, &handle.execHandle.numOfCols, &handle.execHandle.pSchemaWrapper); + handle.execHandle.task = + qCreateQueueExecTaskInfo(handle.execHandle.execCol.qmsg, &reader, &handle.execHandle.numOfCols, NULL); ASSERT(handle.execHandle.task); void* scanner = NULL; qExtractStreamScanner(handle.execHandle.task, &scanner); diff --git a/source/libs/executor/src/executor.c b/source/libs/executor/src/executor.c index 428af19a6c01e5e6fe4058ea48bd6347ad83233c..e561b6e12486970abbcb79671df239cc1952146d 100644 --- a/source/libs/executor/src/executor.c +++ b/source/libs/executor/src/executor.c @@ -246,7 +246,9 @@ qTaskInfo_t qCreateQueueExecTaskInfo(void* msg, SReadHandle* readers, int32_t* n } } - *pSchema = tCloneSSchemaWrapper(((SExecTaskInfo*)pTaskInfo)->schemaInfo.qsw); + if (pSchema) { + *pSchema = tCloneSSchemaWrapper(((SExecTaskInfo*)pTaskInfo)->schemaInfo.qsw); + } return pTaskInfo; } @@ -659,7 +661,7 @@ int32_t qExecTask(qTaskInfo_t tinfo, SSDataBlock** pRes, uint64_t* useconds) { return pTaskInfo->code; } -int32_t qAppendTaskStopInfo(SExecTaskInfo* pTaskInfo, SExchangeOpStopInfo *pInfo) { +int32_t qAppendTaskStopInfo(SExecTaskInfo* pTaskInfo, SExchangeOpStopInfo* pInfo) { taosWLockLatch(&pTaskInfo->stopInfo.lock); taosArrayPush(pTaskInfo->stopInfo.pStopInfo, pInfo); taosWUnLockLatch(&pTaskInfo->stopInfo.lock); @@ -680,7 +682,7 @@ int32_t stopInfoComp(void const* lp, void const* rp) { return 0; } -void qRemoveTaskStopInfo(SExecTaskInfo* pTaskInfo, SExchangeOpStopInfo *pInfo) { +void qRemoveTaskStopInfo(SExecTaskInfo* pTaskInfo, SExchangeOpStopInfo* pInfo) { taosWLockLatch(&pTaskInfo->stopInfo.lock); int32_t idx = taosArraySearchIdx(pTaskInfo->stopInfo.pStopInfo, pInfo, stopInfoComp, TD_EQ); if (idx >= 0) { @@ -696,8 +698,8 @@ void qStopTaskOperators(SExecTaskInfo* pTaskInfo) { int32_t num = taosArrayGetSize(pTaskInfo->stopInfo.pStopInfo); for (int32_t i = 0; i < num; ++i) { - SExchangeOpStopInfo *pStop = taosArrayGet(pTaskInfo->stopInfo.pStopInfo, i); - SExchangeInfo* pExchangeInfo = taosAcquireRef(exchangeObjRefPool, pStop->refId); + SExchangeOpStopInfo* pStop = taosArrayGet(pTaskInfo->stopInfo.pStopInfo, i); + SExchangeInfo* pExchangeInfo = taosAcquireRef(exchangeObjRefPool, pStop->refId); if (pExchangeInfo) { tsem_post(&pExchangeInfo->ready); taosReleaseRef(exchangeObjRefPool, pStop->refId); @@ -715,11 +717,11 @@ int32_t qAsyncKillTask(qTaskInfo_t qinfo) { } qDebug("%s execTask async killed", GET_TASKID(pTaskInfo)); - + setTaskKilled(pTaskInfo); qStopTaskOperators(pTaskInfo); - + return TSDB_CODE_SUCCESS; } @@ -1178,4 +1180,4 @@ void qProcessRspMsg(void* parent, SRpcMsg* pMsg, SEpSet* pEpSet) { pSendInfo->fp(pSendInfo->param, &buf, pMsg->code); rpcFreeCont(pMsg->pCont); destroySendMsgInfo(pSendInfo); -} \ No newline at end of file +} diff --git a/source/libs/wal/src/walMgmt.c b/source/libs/wal/src/walMgmt.c index 1a70a3038f4351016cf4098f9497253e7e75fb9f..b78d5ca6c1c0f4573e473d78ed7d784dc2297d6b 100644 --- a/source/libs/wal/src/walMgmt.c +++ b/source/libs/wal/src/walMgmt.c @@ -203,6 +203,14 @@ void walClose(SWal *pWal) { pWal->pIdxFile = NULL; taosArrayDestroy(pWal->fileInfoSet); pWal->fileInfoSet = NULL; + + void *pIter = NULL; + while (1) { + pIter = taosHashIterate(pWal->pRefHash, pIter); + if (pIter == NULL) break; + SWalRef *pRef = *(SWalRef **)pIter; + taosMemoryFree(pRef); + } taosHashCleanup(pWal->pRefHash); taosThreadMutexUnlock(&pWal->mutex); diff --git a/source/libs/wal/src/walWrite.c b/source/libs/wal/src/walWrite.c index 216dd5fcb149b50c17ef44f2315b6018d215de74..7c3b4cba30ce08948a6a0f0b5a2ef43a238b02d0 100644 --- a/source/libs/wal/src/walWrite.c +++ b/source/libs/wal/src/walWrite.c @@ -26,7 +26,7 @@ int32_t walRestoreFromSnapshot(SWal *pWal, int64_t ver) { while (1) { pIter = taosHashIterate(pWal->pRefHash, pIter); if (pIter == NULL) break; - SWalRef *pRef = (SWalRef *)pIter; + SWalRef *pRef = *(SWalRef **)pIter; if (pRef->refVer != -1 && pRef->refVer <= ver) { taosHashCancelIterate(pWal->pRefHash, pIter); taosThreadMutexUnlock(&pWal->mutex);