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

fix(stream): update the check order.

This commit is contained in:
Haojun Liao 2024-01-25 15:17:14 +08:00
parent 2ecc202cb3
commit 49bc3924fb
1 changed files with 14 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -776,29 +776,29 @@ int32_t streamResumeTask(SStreamTask* pTask) {
const char* id = pTask->id.idStr;
while (1) {
/*int32_t code = */doStreamExecTask(pTask);
/*int32_t code = */ doStreamExecTask(pTask);
taosThreadMutexLock(&pTask->lock);
// check if this task needs to be idle for a while
if (pTask->status.schedIdleTime > 0) {
schedTaskInFuture(pTask);
int32_t numOfItems = streamQueueGetNumOfItems(pTask->inputq.queue);
if ((numOfItems == 0) || streamTaskShouldStop(pTask) || streamTaskShouldPause(pTask)) {
atomic_store_8(&pTask->status.schedStatus, TASK_SCHED_STATUS__INACTIVE);
clearTaskSchedInfo(pTask);
taosThreadMutexUnlock(&pTask->lock);
setLastExecTs(pTask, taosGetTimestampMs());
char* p = streamTaskGetStatus(pTask)->name;
stDebug("s-task:%s exec completed, status:%s, sched-status:%d, lastExecTs:%" PRId64, id, p,
pTask->status.schedStatus, pTask->status.lastExecTs);
return 0;
} else {
int32_t numOfItems = streamQueueGetNumOfItems(pTask->inputq.queue);
// check if this task needs to be idle for a while
if (pTask->status.schedIdleTime > 0) {
schedTaskInFuture(pTask);
if ((numOfItems == 0) || streamTaskShouldStop(pTask) || streamTaskShouldPause(pTask)) {
atomic_store_8(&pTask->status.schedStatus, TASK_SCHED_STATUS__INACTIVE);
taosThreadMutexUnlock(&pTask->lock);
setLastExecTs(pTask, taosGetTimestampMs());
char* p = streamTaskGetStatus(pTask)->name;
stDebug("s-task:%s exec completed, status:%s, sched-status:%d, lastExecTs:%" PRId64, id, p,
pTask->status.schedStatus, pTask->status.lastExecTs);
return 0;
}
}