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

fix mem leak

This commit is contained in:
liuyao 2023-09-20 17:35:11 +08:00
parent 1cebda368a
commit b4be5fe895
3 changed files with 12 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
source/libs/executor/src/streamtimewindowoperator.c
View File

@ -1520,6 +1520,7 @@ void destroyStreamAggSupporter(SStreamAggSupporter* pSup) {
tSimpleHashCleanup(pSup->pResultRows); tSimpleHashCleanup(pSup->pResultRows);
destroyDiskbasedBuf(pSup->pResultBuf); destroyDiskbasedBuf(pSup->pResultBuf);
blockDataDestroy(pSup->pScanBlock); blockDataDestroy(pSup->pScanBlock);
pSup->stateStore.streamFileStateDestroy(pSup->pState->pFileState);
taosMemoryFreeClear(pSup->pState); taosMemoryFreeClear(pSup->pState);
taosMemoryFreeClear(pSup->pDummyCtx); taosMemoryFreeClear(pSup->pDummyCtx);
} }

14
source/libs/stream/src/streamSessionState.c
View File

@ -203,7 +203,7 @@ void sessionWinStateCleanup(void* pBuff) {
size_t keyLen = 0; size_t keyLen = 0;
int32_t iter = 0; int32_t iter = 0;
while ((pIte = tSimpleHashIterate(pBuff, pIte, &iter)) != NULL) { while ((pIte = tSimpleHashIterate(pBuff, pIte, &iter)) != NULL) {
SArray* pWinStates = (SArray*) pIte; SArray* pWinStates = (SArray*) (*(void**)pIte);
taosArrayDestroy(pWinStates); taosArrayDestroy(pWinStates);
} }
tSimpleHashCleanup(pBuff); tSimpleHashCleanup(pBuff);
@ -360,19 +360,23 @@ int32_t sessionWinStateGetKeyByRange(SStreamFileState* pFileState, const SSessio
int32_t code = sessionWinStateGetKVByCur(pCur, &tmpKey, NULL, NULL); int32_t code = sessionWinStateGetKVByCur(pCur, &tmpKey, NULL, NULL);
bool hasCurrentPrev = true; bool hasCurrentPrev = true;
if (code == TSDB_CODE_FAILED) { if (code == TSDB_CODE_FAILED) {
streamStateFreeCur(pCur);
pCur = sessionWinStateSeekKeyNext(pFileState, key); pCur = sessionWinStateSeekKeyNext(pFileState, key);
code = sessionWinStateGetKVByCur(pCur, &tmpKey, NULL, NULL); code = sessionWinStateGetKVByCur(pCur, &tmpKey, NULL, NULL);
hasCurrentPrev = false; hasCurrentPrev = false;
} }
if (code == TSDB_CODE_FAILED) { if (code == TSDB_CODE_FAILED) {
return TSDB_CODE_FAILED; code = TSDB_CODE_FAILED;
goto _end;
} }
if (sessionRangeKeyCmpr(key, &tmpKey) == 0) { if (sessionRangeKeyCmpr(key, &tmpKey) == 0) {
*curKey = tmpKey; *curKey = tmpKey;
return code; goto _end;
} else if (!hasCurrentPrev) { } else if (!hasCurrentPrev) {
return TSDB_CODE_FAILED; code = TSDB_CODE_FAILED;
goto _end;
} }
sessionWinStateMoveToNext(pCur); sessionWinStateMoveToNext(pCur);
@ -383,6 +387,8 @@ int32_t sessionWinStateGetKeyByRange(SStreamFileState* pFileState, const SSessio
code = TSDB_CODE_FAILED; code = TSDB_CODE_FAILED;
} }
_end:
streamStateFreeCur(pCur);
return code; return code;
} }

1
source/libs/stream/src/streamState.c
View File

@ -683,6 +683,7 @@ int32_t streamStateCurPrev(SStreamState* pState, SStreamStateCur* pCur) {
} }
void streamStateFreeCur(SStreamStateCur* pCur) { void streamStateFreeCur(SStreamStateCur* pCur) {
if (!pCur || pCur->buffIndex >= 0) { if (!pCur || pCur->buffIndex >= 0) {
taosMemoryFree(pCur);
return; return;
} }
qDebug("streamStateFreeCur"); qDebug("streamStateFreeCur");