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

other: add some logs.

This commit is contained in:
Haojun Liao 2023-06-08 19:14:36 +08:00
parent c9aa59cbb9
commit 0fd11aaefe
1 changed files with 3 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -330,7 +330,8 @@ int32_t streamExecForAll(SStreamTask* pTask) {
while (1) { while (1) {
// downstream task's input queue is blocked, stop immediately // downstream task's input queue is blocked, stop immediately
if (streamTaskShouldPause(&pTask->status) || (pTask->outputStatus == TASK_INPUT_STATUS__BLOCKED)) { if (streamTaskShouldPause(&pTask->status) || (pTask->outputStatus == TASK_INPUT_STATUS__BLOCKED) ||
streamTaskShouldStop(&pTask->status)) {
if (batchSize > 1) { if (batchSize > 1) {
break; break;
} else { } else {
@ -461,7 +462,7 @@ int32_t streamTryExec(SStreamTask* pTask) {
// todo the task should be commit here // todo the task should be commit here
atomic_store_8(&pTask->status.schedStatus, TASK_SCHED_STATUS__INACTIVE); atomic_store_8(&pTask->status.schedStatus, TASK_SCHED_STATUS__INACTIVE);
qDebug("s-task:%s exec completed", pTask->id.idStr); qDebug("s-task:%s exec completed, status:%d", pTask->id.idStr, pTask->status.taskStatus);
if (!taosQueueEmpty(pTask->inputQueue->queue) && (!streamTaskShouldStop(&pTask->status)) && if (!taosQueueEmpty(pTask->inputQueue->queue) && (!streamTaskShouldStop(&pTask->status)) &&
(!streamTaskShouldPause(&pTask->status))) { (!streamTaskShouldPause(&pTask->status))) {