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

fix(stream): limit the batch size.

This commit is contained in:
Haojun Liao 2023-05-11 15:07:07 +08:00
parent a76e47ef54
commit b2d141fe40
1 changed files with 3 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
source/libs/stream/src/streamExec.c
View File

@ -15,7 +15,8 @@
#include "streamInc.h" #include "streamInc.h"
#define MAX_STREAM_EXEC_BATCH_NUM 10240 // maximum allowed processed block batches. One block may include several submit blocks
#define MAX_STREAM_EXEC_BATCH_NUM 128
#define MIN_STREAM_EXEC_BATCH_NUM 16 #define MIN_STREAM_EXEC_BATCH_NUM 16
bool streamTaskShouldStop(const SStreamStatus* pStatus) { bool streamTaskShouldStop(const SStreamStatus* pStatus) {
@ -297,6 +298,7 @@ int32_t streamExecForAll(SStreamTask* pTask) {
pInput = newRet; pInput = newRet;
streamQueueProcessSuccess(pTask->inputQueue); streamQueueProcessSuccess(pTask->inputQueue);
if (batchSize > MAX_STREAM_EXEC_BATCH_NUM) { if (batchSize > MAX_STREAM_EXEC_BATCH_NUM) {
qDebug("maximum batch limit:%d reached, processing, %s", MAX_STREAM_EXEC_BATCH_NUM, pTask->id.idStr);
break; break;
} }
} }