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

fix(stream): adjust the stop condition for stream tasks.

This commit is contained in:
Haojun Liao 2023-10-31 11:51:47 +08:00
parent fc70fa053c
commit 762cfef498
1 changed files with 2 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -24,7 +24,7 @@ static int32_t streamDoTransferStateToStreamTask(SStreamTask* pTask);
bool streamTaskShouldStop(const SStreamTask* pTask) { bool streamTaskShouldStop(const SStreamTask* pTask) {
ETaskStatus s = streamTaskGetStatus(pTask, NULL); ETaskStatus s = streamTaskGetStatus(pTask, NULL);
return (s == TASK_STATUS__STOP) || (s == TASK_STATUS__DROPPING) || (s == TASK_STATUS__UNINIT); return (s == TASK_STATUS__STOP) || (s == TASK_STATUS__DROPPING);
} }
bool streamTaskShouldPause(const SStreamTask* pTask) { bool streamTaskShouldPause(const SStreamTask* pTask) {
@ -525,7 +525,7 @@ int32_t streamExecForAll(SStreamTask* pTask) {
int32_t blockSize = 0; int32_t blockSize = 0;
int32_t numOfBlocks = 0; int32_t numOfBlocks = 0;
SStreamQueueItem* pInput = NULL; SStreamQueueItem* pInput = NULL;
if (streamTaskShouldStop(pTask)) { if (streamTaskShouldStop(pTask) || (streamTaskGetStatus(pTask, NULL) == TASK_STATUS__UNINIT)) {
stDebug("s-task:%s stream task is stopped", id); stDebug("s-task:%s stream task is stopped", id);
break; break;
} }