Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
c056e345
TDengine
项目概览
taosdata
/
TDengine
大约 2 年 前同步成功
通知
1192
Star
22018
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看板
提交
c056e345
编写于
7月 11, 2023
作者:
Y
yihaoDeng
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
fix compile error
上级
184b2d64
变更
1
隐藏空白更改
内联
并排
Showing
1 changed file
with
17 addition
and
16 deletion
+17
-16
source/libs/stream/src/streamExec.c
source/libs/stream/src/streamExec.c
+17
-16
未找到文件。
source/libs/stream/src/streamExec.c
浏览文件 @
c056e345
...
@@ -16,7 +16,7 @@
...
@@ -16,7 +16,7 @@
#include "streamInt.h"
#include "streamInt.h"
// maximum allowed processed block batches. One block may include several submit blocks
// maximum allowed processed block batches. One block may include several submit blocks
#define MAX_STREAM_RESULT_DUMP_THRESHOLD
100
#define MAX_STREAM_RESULT_DUMP_THRESHOLD 100
static
int32_t
updateCheckPointInfo
(
SStreamTask
*
pTask
,
int64_t
checkpointId
);
static
int32_t
updateCheckPointInfo
(
SStreamTask
*
pTask
,
int64_t
checkpointId
);
...
@@ -104,8 +104,8 @@ static int32_t streamTaskExecImpl(SStreamTask* pTask, SStreamQueueItem* pItem, i
...
@@ -104,8 +104,8 @@ static int32_t streamTaskExecImpl(SStreamTask* pTask, SStreamQueueItem* pItem, i
taosArrayPush
(
pRes
,
&
block
);
taosArrayPush
(
pRes
,
&
block
);
numOfBlocks
+=
1
;
numOfBlocks
+=
1
;
qDebug
(
"s-task:%s(child %d) retrieve process completed, reqId:0x%"
PRIx64
" dump results"
,
pTask
->
id
.
idStr
,
pTask
->
info
.
selfChildId
,
qDebug
(
"s-task:%s(child %d) retrieve process completed, reqId:0x%"
PRIx64
" dump results"
,
pTask
->
id
.
idStr
,
pRetrieveBlock
->
reqId
);
p
Task
->
info
.
selfChildId
,
p
RetrieveBlock
->
reqId
);
}
}
break
;
break
;
...
@@ -329,7 +329,8 @@ static void waitForTaskIdle(SStreamTask* pTask, SStreamTask* pStreamTask) {
...
@@ -329,7 +329,8 @@ static void waitForTaskIdle(SStreamTask* pTask, SStreamTask* pStreamTask) {
static
int32_t
streamTransferStateToStreamTask
(
SStreamTask
*
pTask
)
{
static
int32_t
streamTransferStateToStreamTask
(
SStreamTask
*
pTask
)
{
SStreamTask
*
pStreamTask
=
streamMetaAcquireTask
(
pTask
->
pMeta
,
pTask
->
streamTaskId
.
taskId
);
SStreamTask
*
pStreamTask
=
streamMetaAcquireTask
(
pTask
->
pMeta
,
pTask
->
streamTaskId
.
taskId
);
qDebug
(
"s-task:%s scan history task end, update stream task:%s info, transfer exec state"
,
pTask
->
id
.
idStr
,
pStreamTask
->
id
.
idStr
);
qDebug
(
"s-task:%s scan history task end, update stream task:%s info, transfer exec state"
,
pTask
->
id
.
idStr
,
pStreamTask
->
id
.
idStr
);
// todo handle stream task is dropped here
// todo handle stream task is dropped here
...
@@ -355,7 +356,7 @@ static int32_t streamTransferStateToStreamTask(SStreamTask* pTask) {
...
@@ -355,7 +356,7 @@ static int32_t streamTransferStateToStreamTask(SStreamTask* pTask) {
if
(
pStreamTask
->
info
.
taskLevel
==
TASK_LEVEL__SOURCE
)
{
if
(
pStreamTask
->
info
.
taskLevel
==
TASK_LEVEL__SOURCE
)
{
// update the scan data range for source task.
// update the scan data range for source task.
qDebug
(
"s-task:%s level:%d stream task window %"
PRId64
" - %"
PRId64
" update to %"
PRId64
" - %"
PRId64
qDebug
(
"s-task:%s level:%d stream task window %"
PRId64
" - %"
PRId64
" update to %"
PRId64
" - %"
PRId64
", status:%s, sched-status:%d"
,
", status:%s, sched-status:%d"
,
pStreamTask
->
id
.
idStr
,
TASK_LEVEL__SOURCE
,
pTimeWindow
->
skey
,
pTimeWindow
->
ekey
,
INT64_MIN
,
pStreamTask
->
id
.
idStr
,
TASK_LEVEL__SOURCE
,
pTimeWindow
->
skey
,
pTimeWindow
->
ekey
,
INT64_MIN
,
pTimeWindow
->
ekey
,
streamGetTaskStatusStr
(
TASK_STATUS__NORMAL
),
pStreamTask
->
status
.
schedStatus
);
pTimeWindow
->
ekey
,
streamGetTaskStatusStr
(
TASK_STATUS__NORMAL
),
pStreamTask
->
status
.
schedStatus
);
}
else
{
}
else
{
...
@@ -385,13 +386,13 @@ int32_t streamExecForAll(SStreamTask* pTask) {
...
@@ -385,13 +386,13 @@ int32_t streamExecForAll(SStreamTask* pTask) {
const
char
*
id
=
pTask
->
id
.
idStr
;
const
char
*
id
=
pTask
->
id
.
idStr
;
while
(
1
)
{
while
(
1
)
{
int32_t
batchSize
=
0
;
int32_t
batchSize
=
0
;
SStreamQueueItem
*
pInput
=
NULL
;
SStreamQueueItem
*
pInput
=
NULL
;
// merge multiple input data if possible in the input queue.
// merge multiple input data if possible in the input queue.
qDebug
(
"s-task:%s start to extract data block from inputQ"
,
id
);
qDebug
(
"s-task:%s start to extract data block from inputQ"
,
id
);
/*int32_t code = */
extractBlocksFromInputQ
(
pTask
,
&
pInput
,
&
batchSize
,
id
);
/*int32_t code = */
extractBlocksFromInputQ
(
pTask
,
&
pInput
,
&
batchSize
,
id
);
if
(
pInput
==
NULL
)
{
if
(
pInput
==
NULL
)
{
ASSERT
(
batchSize
==
0
);
ASSERT
(
batchSize
==
0
);
if
(
pTask
->
info
.
fillHistory
&&
pTask
->
status
.
transferState
)
{
if
(
pTask
->
info
.
fillHistory
&&
pTask
->
status
.
transferState
)
{
...
@@ -451,7 +452,7 @@ int32_t streamExecForAll(SStreamTask* pTask) {
...
@@ -451,7 +452,7 @@ int32_t streamExecForAll(SStreamTask* pTask) {
const
SStreamRefDataBlock
*
pRefBlock
=
(
const
SStreamRefDataBlock
*
)
pInput
;
const
SStreamRefDataBlock
*
pRefBlock
=
(
const
SStreamRefDataBlock
*
)
pInput
;
qSetMultiStreamInput
(
pExecutor
,
pRefBlock
->
pBlock
,
1
,
STREAM_INPUT__DATA_BLOCK
);
qSetMultiStreamInput
(
pExecutor
,
pRefBlock
->
pBlock
,
1
,
STREAM_INPUT__DATA_BLOCK
);
}
else
if
(
pItem
->
type
==
STREAM_INPUT__CHECKPOINT
)
{
}
else
if
(
pItem
->
type
==
STREAM_INPUT__CHECKPOINT
)
{
const
SStreamCheckpoint
*
pCheckpoint
=
(
const
SStreamCheckpoint
*
)
pInput
;
const
SStreamCheckpoint
*
pCheckpoint
=
(
const
SStreamCheckpoint
*
)
pInput
;
qSetMultiStreamInput
(
pExecutor
,
pCheckpoint
->
pBlock
,
1
,
STREAM_INPUT__CHECKPOINT
);
qSetMultiStreamInput
(
pExecutor
,
pCheckpoint
->
pBlock
,
1
,
STREAM_INPUT__CHECKPOINT
);
}
else
{
}
else
{
ASSERT
(
0
);
ASSERT
(
0
);
...
@@ -462,9 +463,9 @@ int32_t streamExecForAll(SStreamTask* pTask) {
...
@@ -462,9 +463,9 @@ int32_t streamExecForAll(SStreamTask* pTask) {
int32_t
totalBlocks
=
0
;
int32_t
totalBlocks
=
0
;
streamTaskExecImpl
(
pTask
,
pInput
,
&
resSize
,
&
totalBlocks
);
streamTaskExecImpl
(
pTask
,
pInput
,
&
resSize
,
&
totalBlocks
);
double
el
=
(
taosGetTimestampMs
()
-
st
)
/
1000
.
0
;
double
el
=
(
taosGetTimestampMs
()
-
st
)
/
1000
.
0
;
qDebug
(
"s-task:%s batch of input blocks exec end, elapsed time:%.2fs, result size:%.2fMiB, numOfBlocks:%d"
,
qDebug
(
"s-task:%s batch of input blocks exec end, elapsed time:%.2fs, result size:%.2fMiB, numOfBlocks:%d"
,
id
,
el
,
id
,
el
,
resSize
/
1048576
.
0
,
totalBlocks
);
resSize
/
1048576
.
0
,
totalBlocks
);
int32_t
type
=
pInput
->
type
;
int32_t
type
=
pInput
->
type
;
streamFreeQitem
(
pInput
);
streamFreeQitem
(
pInput
);
...
@@ -475,7 +476,6 @@ int32_t streamExecForAll(SStreamTask* pTask) {
...
@@ -475,7 +476,6 @@ int32_t streamExecForAll(SStreamTask* pTask) {
pTask
->
status
.
taskStatus
=
TASK_STATUS__CK_READY
;
pTask
->
status
.
taskStatus
=
TASK_STATUS__CK_READY
;
return
0
;
return
0
;
}
}
}
}
}
}
...
@@ -512,19 +512,20 @@ int32_t streamTryExec(SStreamTask* pTask) {
...
@@ -512,19 +512,20 @@ int32_t streamTryExec(SStreamTask* pTask) {
// todo the task should be commit here
// todo the task should be commit here
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"
,
pTask
->
id
.
idStr
,
streamGetTaskStatusStr
(
pTask
->
status
.
taskStatus
),
qDebug
(
"s-task:%s exec completed, status:%s, sched-status:%d"
,
pTask
->
id
.
idStr
,
pTask
->
status
.
schedStatus
);
streamGetTaskStatusStr
(
pTask
->
status
.
taskStatus
),
pTask
->
status
.
schedStatus
);
if
(
pTask
->
status
.
taskStatus
==
TASK_STATUS__CK_READY
)
{
if
(
pTask
->
status
.
taskStatus
==
TASK_STATUS__CK_READY
)
{
// check for all tasks, and do generate the vnode-wide checkpoint data.
// check for all tasks, and do generate the vnode-wide checkpoint data.
// todo extract method
// todo extract method
SStreamMeta
*
pMeta
=
pTask
->
pMeta
;
SStreamMeta
*
pMeta
=
pTask
->
pMeta
;
int32_t
remain
=
atomic_sub_fetch_32
(
&
pMeta
->
chkptNotReadyTasks
,
1
);
int32_t
remain
=
atomic_sub_fetch_32
(
&
pMeta
->
chkptNotReadyTasks
,
1
);
ASSERT
(
remain
>=
0
);
ASSERT
(
remain
>=
0
);
if
(
remain
==
0
)
{
// all tasks are in TASK_STATUS__CK_READY state
if
(
remain
==
0
)
{
// all tasks are in TASK_STATUS__CK_READY state
streamBackendDoCheckpoint
(
pMeta
,
pTask
->
checkpointingId
);
streamBackendDoCheckpoint
(
pMeta
,
pTask
->
checkpointingId
);
qDebug
(
"vgId:%d do vnode wide checkpoint completed, checkpoint id:%"
PRId64
,
pMeta
->
vgId
);
qDebug
(
"vgId:%d do vnode wide checkpoint completed, checkpoint id:%"
PRId64
""
,
pMeta
->
vgId
,
pTask
->
checkpointingId
);
}
}
if
(
pTask
->
info
.
taskLevel
!=
TASK_LEVEL__SINK
)
{
if
(
pTask
->
info
.
taskLevel
!=
TASK_LEVEL__SINK
)
{
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录