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

fix(stream): fix memory leak.

This commit is contained in:
Haojun Liao 2023-07-22 01:02:58 +08:00
parent bfc4a07207
commit f4864b8713
3 changed files with 15 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
source/libs/stream/inc/streamInt.h
View File

@ -31,6 +31,12 @@ typedef struct {
void* timer; void* timer;
} SStreamGlobalEnv; } SStreamGlobalEnv;
typedef struct {
SEpSet epset;
int32_t taskId;
SRpcMsg msg;
} SStreamContinueExecInfo;
extern SStreamGlobalEnv streamEnv; extern SStreamGlobalEnv streamEnv;
void streamRetryDispatchStreamBlock(SStreamTask* pTask, int64_t waitDuration); void streamRetryDispatchStreamBlock(SStreamTask* pTask, int64_t waitDuration);

6
source/libs/stream/src/streamDispatch.c
View File

@ -648,12 +648,6 @@ int32_t tDecodeCompleteHistoryDataMsg(SDecoder* pDecoder, SStreamCompleteHistory
return 0; return 0;
} }
typedef struct {
SEpSet epset;
int32_t taskId;
SRpcMsg msg;
} SStreamContinueExecInfo;
int32_t streamAddEndScanHistoryMsg(SStreamTask* pTask, SRpcHandleInfo* pRpcInfo, SStreamScanHistoryFinishReq* pReq) { int32_t streamAddEndScanHistoryMsg(SStreamTask* pTask, SRpcHandleInfo* pRpcInfo, SStreamScanHistoryFinishReq* pReq) {
int32_t len = 0; int32_t len = 0;
int32_t code = 0; int32_t code = 0;

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

@ -13,6 +13,8 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>. * along with this program. If not, see <http://www.gnu.org/licenses/>.
*/ */
#include <libs/transport/trpc.h>
#include <streamInt.h>
#include "executor.h" #include "executor.h"
#include "tstream.h" #include "tstream.h"
#include "wal.h" #include "wal.h"
@ -203,6 +205,11 @@ int32_t tDecodeStreamTask(SDecoder* pDecoder, SStreamTask* pTask) {
return 0; return 0;
} }
static void freeItem(void* p) {
SStreamContinueExecInfo* pInfo = p;
rpcFreeCont(pInfo->msg.pCont);
}
void tFreeStreamTask(SStreamTask* pTask) { void tFreeStreamTask(SStreamTask* pTask) {
qDebug("free s-task:%s", pTask->id.idStr); qDebug("free s-task:%s", pTask->id.idStr);
@ -252,7 +259,8 @@ void tFreeStreamTask(SStreamTask* pTask) {
} }
if (pTask->pRspMsgList != NULL) { if (pTask->pRspMsgList != NULL) {
pTask->pRspMsgList = taosArrayDestroy(pTask->pRspMsgList); taosArrayDestroyEx(pTask->pRspMsgList, freeItem);
pTask->pRspMsgList = NULL;
} }
taosThreadMutexDestroy(&pTask->lock); taosThreadMutexDestroy(&pTask->lock);