xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

fix(stream): fix invalid read and memory leak.

This commit is contained in:
Haojun Liao 2023-09-25 23:10:33 +08:00
parent 65f66f0986
commit 6a98b11bef
2 changed files with 7 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
source/dnode/vnode/src/tq/tqSink.c
View File

@ -824,6 +824,8 @@ void tqSinkDataIntoDstTable(SStreamTask* pTask, void* vnode, void* data) {
bool hasSubmit = false; bool hasSubmit = false;
for (int32_t i = 0; i < numOfBlocks; i++) { for (int32_t i = 0; i < numOfBlocks; i++) {
if (streamTaskShouldStop(&pTask->status)) { if (streamTaskShouldStop(&pTask->status)) {
taosHashCleanup(pTableIndexMap);
tDestroySubmitReq(&submitReq, TSDB_MSG_FLG_ENCODE);
return; return;
} }

10
source/libs/stream/src/streamData.c
View File

@ -165,17 +165,17 @@ SStreamQueueItem* streamMergeQueueItem(SStreamQueueItem* dst, SStreamQueueItem*
SStreamDataBlock* pBlockSrc = (SStreamDataBlock*)pElem; SStreamDataBlock* pBlockSrc = (SStreamDataBlock*)pElem;
taosArrayAddAll(pBlock->blocks, pBlockSrc->blocks); taosArrayAddAll(pBlock->blocks, pBlockSrc->blocks);
taosArrayDestroy(pBlockSrc->blocks); taosArrayDestroy(pBlockSrc->blocks);
taosFreeQitem(pElem);
streamQueueItemIncSize(dst, streamQueueItemGetSize(pElem)); streamQueueItemIncSize(dst, streamQueueItemGetSize(pElem));
taosFreeQitem(pElem);
return dst; return dst;
} else if (dst->type == STREAM_INPUT__MERGED_SUBMIT && pElem->type == STREAM_INPUT__DATA_SUBMIT) { } else if (dst->type == STREAM_INPUT__MERGED_SUBMIT && pElem->type == STREAM_INPUT__DATA_SUBMIT) {
SStreamMergedSubmit* pMerged = (SStreamMergedSubmit*)dst; SStreamMergedSubmit* pMerged = (SStreamMergedSubmit*)dst;
SStreamDataSubmit* pBlockSrc = (SStreamDataSubmit*)pElem; SStreamDataSubmit* pBlockSrc = (SStreamDataSubmit*)pElem;
streamMergeSubmit(pMerged, pBlockSrc); streamMergeSubmit(pMerged, pBlockSrc);
taosFreeQitem(pElem);
streamQueueItemIncSize(dst, streamQueueItemGetSize(pElem)); streamQueueItemIncSize(dst, streamQueueItemGetSize(pElem));
taosFreeQitem(pElem);
return dst; return dst;
} else if (dst->type == STREAM_INPUT__DATA_SUBMIT && pElem->type == STREAM_INPUT__DATA_SUBMIT) { } else if (dst->type == STREAM_INPUT__DATA_SUBMIT && pElem->type == STREAM_INPUT__DATA_SUBMIT) {
SStreamMergedSubmit* pMerged = streamMergedSubmitNew(); SStreamMergedSubmit* pMerged = streamMergedSubmitNew();
@ -188,9 +188,9 @@ SStreamQueueItem* streamMergeQueueItem(SStreamQueueItem* dst, SStreamQueueItem*
streamMergeSubmit(pMerged, (SStreamDataSubmit*)dst); streamMergeSubmit(pMerged, (SStreamDataSubmit*)dst);
streamMergeSubmit(pMerged, (SStreamDataSubmit*)pElem); streamMergeSubmit(pMerged, (SStreamDataSubmit*)pElem);
taosFreeQitem(dst); taosFreeQitem(dst);
taosFreeQitem(pElem); taosFreeQitem(pElem);
return (SStreamQueueItem*)pMerged; return (SStreamQueueItem*)pMerged;
} else { } else {
stDebug("block type:%s not merged with existed blocks list, type:%d", streamQueueItemGetTypeStr(pElem->type), dst->type); stDebug("block type:%s not merged with existed blocks list, type:%d", streamQueueItemGetTypeStr(pElem->type), dst->type);