Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
53d8c2d7
T
TDengine
项目概览
taosdata
/
TDengine
大约 1 年 前同步成功
通知
1183
Star
22014
Fork
4786
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
1
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
T
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
1
Issue
1
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
未验证
提交
53d8c2d7
编写于
7月 16, 2022
作者:
L
Liu Jicong
提交者:
GitHub
7月 16, 2022
浏览文件
操作
浏览文件
下载
差异文件
Merge pull request #14972 from taosdata/feature/stream
feat(stream): pipelined exec for recovering
上级
f41b7a68
280ed4f6
变更
6
隐藏空白更改
内联
并排
Showing
6 changed file
with
156 addition
and
13 deletion
+156
-13
include/libs/stream/tstream.h
include/libs/stream/tstream.h
+5
-0
source/dnode/mnode/impl/src/mndStream.c
source/dnode/mnode/impl/src/mndStream.c
+87
-3
source/dnode/snode/inc/sndInt.h
source/dnode/snode/inc/sndInt.h
+0
-9
source/libs/stream/inc/streamInc.h
source/libs/stream/inc/streamInc.h
+2
-0
source/libs/stream/src/streamExec.c
source/libs/stream/src/streamExec.c
+57
-0
source/libs/stream/src/streamRecover.c
source/libs/stream/src/streamRecover.c
+5
-1
未找到文件。
include/libs/stream/tstream.h
浏览文件 @
53d8c2d7
...
...
@@ -31,6 +31,11 @@ extern "C" {
typedef
struct
SStreamTask
SStreamTask
;
enum
{
STREAM_STATUS__NORMAL
=
0
,
STREAM_STATUS__RECOVER
,
};
enum
{
TASK_STATUS__NORMAL
=
0
,
TASK_STATUS__DROPPING
,
...
...
source/dnode/mnode/impl/src/mndStream.c
浏览文件 @
53d8c2d7
...
...
@@ -177,7 +177,7 @@ static int32_t mndStreamActionUpdate(SSdb *pSdb, SStreamObj *pOldStream, SStream
taosWLockLatch
(
&
pOldStream
->
lock
);
// TODO handle update
pOldStream
->
status
=
pNewStream
->
status
;
taosWUnLockLatch
(
&
pOldStream
->
lock
);
return
0
;
...
...
@@ -395,6 +395,20 @@ int32_t mndPersistDropStreamLog(SMnode *pMnode, STrans *pTrans, SStreamObj *pStr
return
0
;
}
static
int32_t
mndSetStreamRecover
(
SMnode
*
pMnode
,
STrans
*
pTrans
,
const
SStreamObj
*
pStream
)
{
SStreamObj
streamObj
=
{
0
};
memcpy
(
streamObj
.
name
,
pStream
->
name
,
TSDB_STREAM_FNAME_LEN
);
streamObj
.
status
=
STREAM_STATUS__RECOVER
;
SSdbRaw
*
pCommitRaw
=
mndStreamActionEncode
(
&
streamObj
);
if
(
pCommitRaw
==
NULL
||
mndTransAppendCommitlog
(
pTrans
,
pCommitRaw
)
!=
0
)
{
mError
(
"stream trans:%d, failed to append commit log since %s"
,
pTrans
->
id
,
terrstr
());
mndTransDrop
(
pTrans
);
return
-
1
;
}
sdbSetRawStatus
(
pCommitRaw
,
SDB_STATUS_READY
);
return
0
;
}
static
int32_t
mndCreateStbForStream
(
SMnode
*
pMnode
,
STrans
*
pTrans
,
const
SStreamObj
*
pStream
,
const
char
*
user
)
{
SStbObj
*
pStb
=
NULL
;
SDbObj
*
pDb
=
NULL
;
...
...
@@ -492,6 +506,76 @@ static int32_t mndPersistTaskDropReq(STrans *pTrans, SStreamTask *pTask) {
return
0
;
}
static
int32_t
mndPersistTaskRecoverReq
(
STrans
*
pTrans
,
SStreamTask
*
pTask
)
{
SMStreamTaskRecoverReq
*
pReq
=
taosMemoryCalloc
(
1
,
sizeof
(
SMStreamTaskRecoverReq
));
if
(
pReq
==
NULL
)
{
terrno
=
TSDB_CODE_OUT_OF_MEMORY
;
return
-
1
;
}
pReq
->
streamId
=
pTask
->
streamId
;
pReq
->
taskId
=
pTask
->
taskId
;
int32_t
len
;
int32_t
code
;
tEncodeSize
(
tEncodeSMStreamTaskRecoverReq
,
pReq
,
len
,
code
);
if
(
code
!=
0
)
{
return
-
1
;
}
void
*
buf
=
taosMemoryCalloc
(
1
,
sizeof
(
SMsgHead
)
+
len
);
if
(
buf
==
NULL
)
{
return
-
1
;
}
void
*
abuf
=
POINTER_SHIFT
(
buf
,
sizeof
(
SMsgHead
));
SEncoder
encoder
;
tEncoderInit
(
&
encoder
,
abuf
,
len
);
tEncodeSMStreamTaskRecoverReq
(
&
encoder
,
pReq
);
((
SMsgHead
*
)
buf
)
->
vgId
=
pTask
->
nodeId
;
STransAction
action
=
{
0
};
memcpy
(
&
action
.
epSet
,
&
pTask
->
epSet
,
sizeof
(
SEpSet
));
action
.
pCont
=
buf
;
action
.
contLen
=
sizeof
(
SMsgHead
)
+
len
;
action
.
msgType
=
TDMT_STREAM_TASK_RECOVER
;
if
(
mndTransAppendRedoAction
(
pTrans
,
&
action
)
!=
0
)
{
taosMemoryFree
(
buf
);
return
-
1
;
}
return
0
;
}
int32_t
mndRecoverStreamTasks
(
SMnode
*
pMnode
,
STrans
*
pTrans
,
SStreamObj
*
pStream
)
{
if
(
pStream
->
isDistributed
)
{
int32_t
lv
=
taosArrayGetSize
(
pStream
->
tasks
);
for
(
int32_t
i
=
0
;
i
<
lv
;
i
++
)
{
SArray
*
pTasks
=
taosArrayGetP
(
pStream
->
tasks
,
i
);
int32_t
sz
=
taosArrayGetSize
(
pTasks
);
SStreamTask
*
pTask
=
taosArrayGetP
(
pTasks
,
0
);
if
(
!
pTask
->
isDataScan
&&
pTask
->
execType
!=
TASK_EXEC__NONE
)
{
ASSERT
(
sz
==
1
);
if
(
mndPersistTaskRecoverReq
(
pTrans
,
pTask
)
<
0
)
{
return
-
1
;
}
}
else
{
continue
;
}
}
}
else
{
int32_t
lv
=
taosArrayGetSize
(
pStream
->
tasks
);
for
(
int32_t
i
=
0
;
i
<
lv
;
i
++
)
{
SArray
*
pTasks
=
taosArrayGetP
(
pStream
->
tasks
,
i
);
int32_t
sz
=
taosArrayGetSize
(
pTasks
);
for
(
int32_t
j
=
0
;
j
<
sz
;
j
++
)
{
SStreamTask
*
pTask
=
taosArrayGetP
(
pTasks
,
j
);
if
(
!
pTask
->
isDataScan
)
break
;
ASSERT
(
pTask
->
execType
!=
TASK_EXEC__NONE
);
if
(
mndPersistTaskRecoverReq
(
pTrans
,
pTask
)
<
0
)
{
return
-
1
;
}
}
}
}
return
0
;
}
int32_t
mndDropStreamTasks
(
SMnode
*
pMnode
,
STrans
*
pTrans
,
SStreamObj
*
pStream
)
{
int32_t
lv
=
taosArrayGetSize
(
pStream
->
tasks
);
for
(
int32_t
i
=
0
;
i
<
lv
;
i
++
)
{
...
...
@@ -712,14 +796,14 @@ static int32_t mndProcessRecoverStreamReq(SRpcMsg *pReq) {
mDebug
(
"trans:%d, used to drop stream:%s"
,
pTrans
->
id
,
recoverReq
.
name
);
// broadcast to recover all tasks
if
(
mnd
Drop
StreamTasks
(
pMnode
,
pTrans
,
pStream
)
<
0
)
{
if
(
mnd
Recover
StreamTasks
(
pMnode
,
pTrans
,
pStream
)
<
0
)
{
mError
(
"stream:%s, failed to recover task since %s"
,
recoverReq
.
name
,
terrstr
());
sdbRelease
(
pMnode
->
pSdb
,
pStream
);
return
-
1
;
}
// update stream status
if
(
mnd
PersistDropStreamLog
(
pMnode
,
pTrans
,
pStream
)
<
0
)
{
if
(
mnd
SetStreamRecover
(
pMnode
,
pTrans
,
pStream
)
<
0
)
{
sdbRelease
(
pMnode
->
pSdb
,
pStream
);
return
-
1
;
}
...
...
source/dnode/snode/inc/sndInt.h
浏览文件 @
53d8c2d7
...
...
@@ -30,15 +30,6 @@
extern
"C"
{
#endif
enum
{
STREAM_STATUS__RUNNING
=
1
,
STREAM_STATUS__STOPPED
,
STREAM_STATUS__CREATING
,
STREAM_STATUS__STOPING
,
STREAM_STATUS__RESTORING
,
STREAM_STATUS__DELETING
,
};
typedef
struct
{
SHashObj
*
pHash
;
// taskId -> SStreamTask
}
SStreamMeta
;
...
...
source/libs/stream/inc/streamInc.h
浏览文件 @
53d8c2d7
...
...
@@ -33,6 +33,8 @@ typedef struct {
static
SStreamGlobalEnv
streamEnv
;
int32_t
streamExec
(
SStreamTask
*
pTask
,
SMsgCb
*
pMsgCb
);
int32_t
streamPipelineExec
(
SStreamTask
*
pTask
,
int32_t
batchNum
);
int32_t
streamDispatch
(
SStreamTask
*
pTask
,
SMsgCb
*
pMsgCb
);
int32_t
streamDispatchReqToData
(
const
SStreamDispatchReq
*
pReq
,
SStreamDataBlock
*
pData
);
int32_t
streamRetrieveReqToData
(
const
SStreamRetrieveReq
*
pReq
,
SStreamDataBlock
*
pData
);
...
...
source/libs/stream/src/streamExec.c
浏览文件 @
53d8c2d7
...
...
@@ -82,6 +82,63 @@ static FORCE_INLINE int32_t streamUpdateVer(SStreamTask* pTask, SStreamDataBlock
return
0
;
}
int32_t
streamPipelineExec
(
SStreamTask
*
pTask
,
int32_t
batchNum
)
{
ASSERT
(
pTask
->
execType
!=
TASK_EXEC__NONE
);
void
*
exec
=
pTask
->
exec
.
executor
;
while
(
1
)
{
SArray
*
pRes
=
taosArrayInit
(
0
,
sizeof
(
SSDataBlock
));
if
(
pRes
==
NULL
)
{
terrno
=
TSDB_CODE_OUT_OF_MEMORY
;
return
-
1
;
}
int32_t
batchCnt
=
0
;
while
(
1
)
{
SSDataBlock
*
output
=
NULL
;
uint64_t
ts
=
0
;
if
(
qExecTask
(
exec
,
&
output
,
&
ts
)
<
0
)
{
ASSERT
(
0
);
}
if
(
output
==
NULL
)
break
;
SSDataBlock
block
=
{
0
};
assignOneDataBlock
(
&
block
,
output
);
block
.
info
.
childId
=
pTask
->
selfChildId
;
taosArrayPush
(
pRes
,
&
block
);
if
(
++
batchCnt
>=
batchNum
)
break
;
}
if
(
taosArrayGetSize
(
pRes
)
==
0
)
{
taosArrayDestroy
(
pRes
);
break
;
}
SStreamDataBlock
*
qRes
=
taosAllocateQitem
(
sizeof
(
SStreamDataBlock
),
DEF_QITEM
);
if
(
qRes
==
NULL
)
{
taosArrayDestroyEx
(
pRes
,
(
FDelete
)
blockDataFreeRes
);
return
-
1
;
}
qRes
->
type
=
STREAM_INPUT__DATA_BLOCK
;
qRes
->
blocks
=
pRes
;
qRes
->
childId
=
pTask
->
selfChildId
;
if
(
streamTaskOutput
(
pTask
,
qRes
)
<
0
)
{
taosArrayDestroyEx
(
pRes
,
(
FDelete
)
blockDataFreeRes
);
taosFreeQitem
(
qRes
);
return
-
1
;
}
if
(
pTask
->
dispatchType
!=
TASK_DISPATCH__NONE
)
{
ASSERT
(
pTask
->
sinkType
==
TASK_SINK__NONE
);
streamDispatch
(
pTask
,
pTask
->
pMsgCb
);
}
}
return
0
;
}
static
SArray
*
streamExecForQall
(
SStreamTask
*
pTask
,
SArray
*
pRes
)
{
int32_t
cnt
=
0
;
void
*
data
=
NULL
;
...
...
source/libs/stream/src/streamRecover.c
浏览文件 @
53d8c2d7
...
...
@@ -125,7 +125,11 @@ int32_t streamProcessFailRecoverReq(SStreamTask* pTask, SMStreamTaskRecoverReq*
}
if
(
pTask
->
taskStatus
==
TASK_STATUS__RECOVERING
)
{
streamProcessRunReq
(
pTask
);
if
(
streamPipelineExec
(
pTask
,
10
)
<
0
)
{
// set fail
return
-
1
;
}
}
return
0
;
}
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录