Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
99359912
TDengine
项目概览
taosdata
/
TDengine
1 年多 前同步成功
通知
1185
Star
22016
Fork
4786
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
1
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
1
Issue
1
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
未验证
提交
99359912
编写于
5月 31, 2023
作者:
H
Haojun Liao
提交者:
GitHub
5月 31, 2023
浏览文件
操作
浏览文件
下载
差异文件
Merge pull request #21539 from taosdata/fix/resume_mem
fix(stream): set the correct rps flag as the dispatch rsp.
上级
92ecdda4
fc20ca84
变更
1
显示空白变更内容
内联
并排
Showing
1 changed file
with
19 addition
and
10 deletion
+19
-10
source/libs/stream/src/stream.c
source/libs/stream/src/stream.c
+19
-10
未找到文件。
source/libs/stream/src/stream.c
浏览文件 @
99359912
...
...
@@ -132,8 +132,6 @@ int32_t streamTaskEnqueueBlocks(SStreamTask* pTask, const SStreamDispatchReq* pR
int32_t
code
=
tAppendDataToInputQueue
(
pTask
,
(
SStreamQueueItem
*
)
pBlock
);
// input queue is full, upstream is blocked now
status
=
(
code
==
TSDB_CODE_SUCCESS
)
?
TASK_INPUT_STATUS__NORMAL
:
TASK_INPUT_STATUS__BLOCKED
;
}
// rsp by input status
...
...
@@ -235,8 +233,7 @@ int32_t streamProcessDispatchMsg(SStreamTask* pTask, SStreamDispatchReq* pReq, S
}
int32_t
streamProcessDispatchRsp
(
SStreamTask
*
pTask
,
SStreamDispatchRsp
*
pRsp
,
int32_t
code
)
{
ASSERT
(
pRsp
->
inputStatus
==
TASK_OUTPUT_STATUS__NORMAL
||
pRsp
->
inputStatus
==
TASK_OUTPUT_STATUS__BLOCKED
);
qDebug
(
"s-task:%s receive dispatch rsp, code: %x"
,
pTask
->
id
.
idStr
,
code
);
qDebug
(
"s-task:%s receive dispatch rsp, status:%d code:%d"
,
pTask
->
id
.
idStr
,
pRsp
->
inputStatus
,
code
);
if
(
pTask
->
outputType
==
TASK_OUTPUT__SHUFFLE_DISPATCH
)
{
int32_t
leftRsp
=
atomic_sub_fetch_32
(
&
pTask
->
shuffleDispatcher
.
waitingRspCnt
,
1
);
...
...
@@ -248,13 +245,16 @@ int32_t streamProcessDispatchRsp(SStreamTask* pTask, SStreamDispatchRsp* pRsp, i
int8_t
old
=
atomic_exchange_8
(
&
pTask
->
outputStatus
,
pRsp
->
inputStatus
);
ASSERT
(
old
==
TASK_OUTPUT_STATUS__WAIT
);
// the input queue of the (down stream) task that receive the output data is full, so the TASK_INPUT_STATUS_BLOCKED is rsp
// todo we need to send EMPTY PACKAGE to detect if the input queue is available for output of upstream task, every 50 ms.
if
(
pRsp
->
inputStatus
==
TASK_INPUT_STATUS__BLOCKED
)
{
// TODO: init recover timer
ASSERT
(
0
);
qError
(
"s-task:%s inputQ of downstream task:0x%x is full, need to block output"
,
pTask
->
id
.
idStr
,
pRsp
->
downstreamTaskId
);
return
0
;
}
//
continue dispatch one block to down stream
in pipeline
//
otherwise, continue dispatch the first block to down stream task
in pipeline
streamDispatchStreamBlock
(
pTask
);
return
0
;
}
...
...
@@ -304,23 +304,32 @@ int32_t tAppendDataToInputQueue(SStreamTask* pTask, SStreamQueueItem* pItem) {
return
-
1
;
}
taosWriteQitem
(
pTask
->
inputQueue
->
queue
,
pItem
);
int32_t
code
=
taosWriteQitem
(
pTask
->
inputQueue
->
queue
,
pItem
);
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
streamDataSubmitDestroy
(
px
);
taosFreeQitem
(
pItem
);
return
code
;
}
}
else
if
(
type
==
STREAM_INPUT__DATA_BLOCK
||
type
==
STREAM_INPUT__DATA_RETRIEVE
||
type
==
STREAM_INPUT__REF_DATA_BLOCK
)
{
if
(
(
pTask
->
taskLevel
==
TASK_LEVEL__SOURCE
)
&&
(
tInputQueueIsFull
(
pTask
)))
{
if
(
/*(pTask->taskLevel == TASK_LEVEL__SOURCE) && */
(
tInputQueueIsFull
(
pTask
)))
{
qError
(
"s-task:%s input queue is full, capacity:%d size:%d MiB, current(blocks:%d, size:%.2fMiB) abort"
,
pTask
->
id
.
idStr
,
STREAM_TASK_INPUT_QUEUEU_CAPACITY
,
STREAM_TASK_INPUT_QUEUEU_CAPACITY_IN_SIZE
,
total
,
size
);
destroyStreamDataBlock
((
SStreamDataBlock
*
)
pItem
);
taosFreeQitem
(
pItem
);
return
-
1
;
}
qDebug
(
"s-task:%s data block enqueue, current(blocks:%d, size:%.2fMiB)"
,
pTask
->
id
.
idStr
,
total
,
size
);
taosWriteQitem
(
pTask
->
inputQueue
->
queue
,
pItem
);
int32_t
code
=
taosWriteQitem
(
pTask
->
inputQueue
->
queue
,
pItem
);
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
destroyStreamDataBlock
((
SStreamDataBlock
*
)
pItem
);
return
code
;
}
}
else
if
(
type
==
STREAM_INPUT__CHECKPOINT
)
{
taosWriteQitem
(
pTask
->
inputQueue
->
queue
,
pItem
);
}
else
if
(
type
==
STREAM_INPUT__GET_RES
)
{
// use the default memory limit, refactor later.
taosWriteQitem
(
pTask
->
inputQueue
->
queue
,
pItem
);
qDebug
(
"s-task:%s data res enqueue, current(blocks:%d, size:%.2fMiB)"
,
pTask
->
id
.
idStr
,
total
,
size
);
}
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录