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

fix(stream): fix obj.

This commit is contained in:
Haojun Liao 2023-09-20 14:50:43 +08:00
parent a997fa5552
commit 29e1efb757
1 changed files with 7 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
source/dnode/vnode/src/tq/tq.c
View File

@ -1128,7 +1128,7 @@ int32_t tqProcessTaskScanHistory(STQ* pTq, SRpcMsg* pMsg) {
while (1) { while (1) {
taosThreadMutexLock(&pStreamTask->lock); taosThreadMutexLock(&pStreamTask->lock);
int8_t status = pTask->status.taskStatus; int8_t status = pStreamTask->status.taskStatus;
if (status == TASK_STATUS__DROPPING || status == TASK_STATUS__STOP) { if (status == TASK_STATUS__DROPPING || status == TASK_STATUS__STOP) {
// return; // return;
// do nothing // do nothing
@ -1139,9 +1139,9 @@ int32_t tqProcessTaskScanHistory(STQ* pTq, SRpcMsg* pMsg) {
// do nothing // do nothing
} }
if (pTask->status.taskStatus == TASK_STATUS__CK) { if (pStreamTask->status.taskStatus == TASK_STATUS__CK) {
qDebug("s-task:%s status:%s during generating checkpoint, wait for 1sec and retry set status:halt", qDebug("s-task:%s status:%s during generating checkpoint, wait for 1sec and retry set status:halt",
pTask->id.idStr, streamGetTaskStatusStr(TASK_STATUS__CK)); pStreamTask->id.idStr, streamGetTaskStatusStr(TASK_STATUS__CK));
taosThreadMutexUnlock(&pStreamTask->lock); taosThreadMutexUnlock(&pStreamTask->lock);
taosMsleep(1000); taosMsleep(1000);
continue; continue;
@ -1149,14 +1149,14 @@ int32_t tqProcessTaskScanHistory(STQ* pTq, SRpcMsg* pMsg) {
// upgrade to halt status // upgrade to halt status
if (status == TASK_STATUS__PAUSE) { if (status == TASK_STATUS__PAUSE) {
qDebug("s-task:%s upgrade status to %s from %s", pTask->id.idStr, streamGetTaskStatusStr(TASK_STATUS__HALT), qDebug("s-task:%s upgrade status to %s from %s", pStreamTask->id.idStr, streamGetTaskStatusStr(TASK_STATUS__HALT),
streamGetTaskStatusStr(TASK_STATUS__PAUSE)); streamGetTaskStatusStr(TASK_STATUS__PAUSE));
} else { } else {
qDebug("s-task:%s halt task", pTask->id.idStr); qDebug("s-task:%s halt task", pStreamTask->id.idStr);
} }
pTask->status.keepTaskStatus = status; pStreamTask->status.keepTaskStatus = status;
pTask->status.taskStatus = TASK_STATUS__HALT; pStreamTask->status.taskStatus = TASK_STATUS__HALT;
tqDebug("s-task:%s level:%d sched-status:%d is halt by fill-history task:%s", pStreamTask->id.idStr, tqDebug("s-task:%s level:%d sched-status:%d is halt by fill-history task:%s", pStreamTask->id.idStr,
pStreamTask->info.taskLevel, pStreamTask->status.schedStatus, id); pStreamTask->info.taskLevel, pStreamTask->status.schedStatus, id);