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

fix(stream): fix error.

This commit is contained in:
Haojun Liao 2023-08-12 16:32:05 +08:00
parent a89ce1a20b
commit ce721a0146
5 changed files with 25 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
include/libs/stream/tstream.h
View File

@ -272,6 +272,7 @@ typedef struct SStreamStatus {
int8_t schedStatus;
int8_t keepTaskStatus;
bool transferState;
bool appendTranstateBlock; // has append the transfer state data block already
int8_t timerActive; // timer is active
int8_t pauseAllowed; // allowed task status to be set to be paused
} SStreamStatus;

6
source/dnode/vnode/src/tq/tqRestore.c
View File

@ -214,10 +214,12 @@ static void checkForFillHistoryVerRange(SStreamTask* pTask, int64_t ver) {
qWarn("s-task:%s fill-history scan WAL, currentVer:%" PRId64 " reach the maximum ver:%" PRId64
", not scan wal anymore, set the transfer state flag",
pTask->id.idStr, ver, pTask->dataRange.range.maxVer);
pTask->status.transferState = true;
if (!pTask->status.appendTranstateBlock) {
pTask->status.appendTranstateBlock = true;
appendTranstateIntoInputQ(pTask);
/*int32_t code = */streamSchedExec(pTask);
}
}
}
int32_t createStreamTaskRunReq(SStreamMeta* pStreamMeta, bool* pScanIdle) {

3
source/libs/stream/src/stream.c
View File

@ -385,8 +385,9 @@ int32_t tAppendDataToInputQueue(SStreamTask* pTask, SStreamQueueItem* pItem) {
destroyStreamDataBlock((SStreamDataBlock*) pItem);
return code;
}
} else if (type == STREAM_INPUT__CHECKPOINT) {
} else if (type == STREAM_INPUT__CHECKPOINT || type == STREAM_INPUT__TRANS_STATE) {
taosWriteQitem(pTask->inputQueue->queue, pItem);
qDebug("s-task:%s trans-state blockdata enqueue, total in queue:%d, size:%.2fMiB", pTask->id.idStr, total, size);
} else if (type == STREAM_INPUT__GET_RES) {
// use the default memory limit, refactor later.
taosWriteQitem(pTask->inputQueue->queue, pItem);

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

@ -484,9 +484,10 @@ int32_t streamProcessTranstateBlock(SStreamTask* pTask, SStreamDataBlock* pBlock
pTask->status.transferState = true;
}
// dispatch the transtate block to downstream task immediately
if (level == TASK_LEVEL__SOURCE || level == TASK_LEVEL__AGG) {
// pBlock-> = pTask->id.taskId;
// dispatch the tran-state block to downstream task immediately
int32_t type = pTask->outputInfo.type;
if ((level == TASK_LEVEL__AGG || level == TASK_LEVEL__SOURCE) &&
(type == TASK_OUTPUT__FIXED_DISPATCH || type == TASK_OUTPUT__SHUFFLE_DISPATCH)) {
pBlock->srcVgId = pTask->pMeta->vgId;
code = taosWriteQitem(pTask->outputInfo.queue->queue, pBlock);
if (code == 0) {
@ -640,10 +641,10 @@ int32_t streamTryExec(SStreamTask* pTask) {
// the schedStatus == TASK_SCHED_STATUS__ACTIVE, streamSchedExec cannot be executed, so execute once again by
// call this function (streamExecForAll) directly.
code = streamExecForAll(pTask);
if (code < 0) {
// code = streamExecForAll(pTask);
// if (code < 0) {
// do nothing
}
// }
}
atomic_store_8(&pTask->status.schedStatus, TASK_SCHED_STATUS__INACTIVE);

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

@ -427,6 +427,8 @@ int32_t appendTranstateIntoInputQ(SStreamTask* pTask) {
return TSDB_CODE_OUT_OF_MEMORY;
}
pTranstate->type = STREAM_INPUT__TRANS_STATE;
pBlock->info.type = STREAM_TRANS_STATE;
pBlock->info.rows = 1;
pBlock->info.childId = pTask->info.selfChildId;
@ -440,7 +442,10 @@ int32_t appendTranstateIntoInputQ(SStreamTask* pTask) {
return TSDB_CODE_OUT_OF_MEMORY;
}
qDebug("s-task:%s set sched-status:%d, prev:%d", pTask->id.idStr, TASK_SCHED_STATUS__INACTIVE, pTask->status.schedStatus);
pTask->status.schedStatus = TASK_SCHED_STATUS__INACTIVE;
streamSchedExec(pTask);
return TSDB_CODE_SUCCESS;
}