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

fix(stream): execute the stream task directly, instead of executing it in a asynchronized way.

This commit is contained in:
Haojun Liao 2023-08-10 15:13:49 +08:00
parent 4393375e47
commit 8890fa5781
1 changed files with 12 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -592,14 +592,21 @@ int32_t streamTryExec(SStreamTask* pTask) {
if (pTask->status.transferState) { if (pTask->status.transferState) {
code = streamTransferStateToStreamTask(pTask); code = streamTransferStateToStreamTask(pTask);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
atomic_store_8(&pTask->status.schedStatus, TASK_SCHED_STATUS__INACTIVE);
return code; return code;
} }
streamSchedExec(pTask);
} else { // the schedStatus == TASK_SCHED_STATUS__ACTIVE, streamSchedExec cannot be executed, so execute once again by
atomic_store_8(&pTask->status.schedStatus, TASK_SCHED_STATUS__INACTIVE); // call this function (streamExecForAll) directly.
qDebug("s-task:%s exec completed, status:%s, sched-status:%d", id, streamGetTaskStatusStr(pTask->status.taskStatus), code = streamExecForAll(pTask);
pTask->status.schedStatus); if (code < 0) {
// do nothing
} }
}
atomic_store_8(&pTask->status.schedStatus, TASK_SCHED_STATUS__INACTIVE);
qDebug("s-task:%s exec completed, status:%s, sched-status:%d", id,
streamGetTaskStatusStr(pTask->status.taskStatus), pTask->status.schedStatus);
} else { } else {
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, status:%s, sched-status:%d", id, streamGetTaskStatusStr(pTask->status.taskStatus), qDebug("s-task:%s exec completed, status:%s, sched-status:%d", id, streamGetTaskStatusStr(pTask->status.taskStatus),