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

fix(stream): fix error while fill history exists.

This commit is contained in:
Haojun Liao 2023-06-26 18:44:59 +08:00
parent daafe240ed
commit 437eb93a4d
4 changed files with 19 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
source/dnode/mnode/impl/src/mndScheduler.c
View File

@ -554,13 +554,15 @@ static int32_t addSourceTasksForMultiLevelStream(SMnode* pMnode, SQueryPlan* pPl
sdbRelease(pSdb, pVgroup);
return code;
}
setHTasksId(pSourceTaskList, pHSourceTaskList);
}
sdbRelease(pSdb, pVgroup);
}
if (pStream->conf.fillHistory) {
setHTasksId(pSourceTaskList, pHSourceTaskList);
}
return TSDB_CODE_SUCCESS;
}

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

@ -343,7 +343,7 @@ static void waitForTaskTobeIdle(SStreamTask* pTask, SStreamTask* pStreamTask) {
double el = (taosGetTimestampMs() - st) / 1000.0;
if (el > 0) {
qDebug("s-task:%s wait for stream task:%s for %.2fs to execute all data in inputQ", pTask->id.idStr,
qDebug("s-task:%s wait for stream task:%s for %.2fs to handle all data in inputQ", pTask->id.idStr,
pStreamTask->id.idStr, el);
}
}

20
source/libs/stream/src/streamRecover.c
View File

@ -479,22 +479,28 @@ static void tryLaunchHistoryTask(void* param, void* tmrId) {
// abort the timer if intend to stop task
SStreamTask* pHTask = streamMetaAcquireTask(pMeta, pTask->historyTaskId.taskId);
if (pHTask == NULL && pTask->status.taskStatus == TASK_STATUS__NORMAL) {
qWarn("s-task:%s vgId:%d failed to launch history task:0x%x, since it may not be built or have been destroyed",
pTask->id.idStr, pMeta->vgId, pTask->historyTaskId.taskId);
if (pHTask == NULL && (!streamTaskShouldStop(&pTask->status))) {
const char* pStatus = streamGetTaskStatusStr(pTask->status.taskStatus);
qWarn(
"s-task:%s vgId:%d status:%s failed to launch history task:0x%x, since it may not be built, or have been "
"destroyed, or should stop exec",
pTask->id.idStr, pMeta->vgId, pStatus, pTask->historyTaskId.taskId);
taosTmrReset(tryLaunchHistoryTask, 100, pInfo, streamEnv.timer, &pTask->timer);
streamMetaReleaseTask(pMeta, pTask);
return;
}
doCheckDownstreamStatus(pTask, pHTask);
if (pHTask != NULL) {
doCheckDownstreamStatus(pTask, pHTask);
streamMetaReleaseTask(pMeta, pHTask);
}
// not in timer anymore
pTask->status.timerActive = 0;
streamMetaReleaseTask(pMeta, pHTask);
streamMetaReleaseTask(pMeta, pTask);
} else {
qError("s-task:0x%x failed to load task", pInfo->taskId);
qError("s-task:0x%x failed to load task, it may have been destoryed", pInfo->taskId);
}
taosMemoryFree(pInfo);
@ -664,7 +670,7 @@ void streamPrepareNdoCheckDownstream(SStreamTask* pTask) {
// launch current task
SHistDataRange* pRange = &pTask->dataRange;
int64_t ekey = pRange->window.ekey;
int64_t ekey = pRange->window.ekey + 1;
int64_t ver = pRange->range.minVer;
pRange->window.skey = ekey;

6
source/libs/stream/src/streamTask.c
View File

@ -42,11 +42,7 @@ SStreamTask* tNewStreamTask(int64_t streamId, int8_t taskLevel, int8_t fillHisto
pTask->id.idStr = taosStrdup(buf);
pTask->status.schedStatus = TASK_SCHED_STATUS__INACTIVE;
if (fillHistory) {
pTask->status.taskStatus = TASK_STATUS__SCAN_HISTORY;
} else {
pTask->status.taskStatus = TASK_STATUS__NORMAL;
}
pTask->status.taskStatus = TASK_STATUS__SCAN_HISTORY;
pTask->inputStatus = TASK_INPUT_STATUS__NORMAL;
pTask->outputStatus = TASK_OUTPUT_STATUS__NORMAL;