Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
3d231831
T
TDengine
项目概览
taosdata
/
TDengine
1 年多 前同步成功
通知
1185
Star
22016
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看板
提交
3d231831
编写于
7月 29, 2023
作者:
H
Haojun Liao
浏览文件
操作
浏览文件
下载
差异文件
Merge branch '3.0' into fix/liaohj
上级
5730fac7
0c5727d1
变更
6
显示空白变更内容
内联
并排
Showing
6 changed file
with
112 addition
and
102 deletion
+112
-102
include/libs/stream/tstream.h
include/libs/stream/tstream.h
+0
-1
source/dnode/vnode/src/tq/tq.c
source/dnode/vnode/src/tq/tq.c
+71
-82
source/dnode/vnode/src/tq/tqRestore.c
source/dnode/vnode/src/tq/tqRestore.c
+4
-3
source/libs/executor/src/scanoperator.c
source/libs/executor/src/scanoperator.c
+14
-14
source/libs/stream/src/streamExec.c
source/libs/stream/src/streamExec.c
+10
-1
source/libs/stream/src/streamRecover.c
source/libs/stream/src/streamRecover.c
+13
-1
未找到文件。
include/libs/stream/tstream.h
浏览文件 @
3d231831
...
...
@@ -45,7 +45,6 @@ enum {
TASK_STATUS__FAIL
,
TASK_STATUS__STOP
,
TASK_STATUS__SCAN_HISTORY
,
// stream task scan history data by using tsdbread in the stream scanner
TASK_STATUS__SCAN_HISTORY_WAL
,
// scan history data in wal
TASK_STATUS__HALT
,
// pause, but not be manipulated by user command
TASK_STATUS__PAUSE
,
// pause
};
...
...
source/dnode/vnode/src/tq/tq.c
浏览文件 @
3d231831
...
...
@@ -1145,7 +1145,6 @@ int32_t tqProcessTaskDeployReq(STQ* pTq, int64_t sversion, char* msg, int32_t ms
// 3. It's an fill history task, do nothing. wait for the main task to start it
SStreamTask
*
p
=
streamMetaAcquireTask
(
pStreamMeta
,
taskId
);
if
(
p
!=
NULL
)
{
// reset the downstreamReady flag.
p
->
status
.
downstreamReady
=
0
;
streamTaskCheckDownstreamTasks
(
p
);
}
...
...
@@ -1154,12 +1153,10 @@ int32_t tqProcessTaskDeployReq(STQ* pTq, int64_t sversion, char* msg, int32_t ms
}
int32_t
tqProcessTaskScanHistory
(
STQ
*
pTq
,
SRpcMsg
*
pMsg
)
{
int32_t
code
=
TSDB_CODE_SUCCESS
;
char
*
msg
=
pMsg
->
pCont
;
SStreamScanHistoryReq
*
pReq
=
(
SStreamScanHistoryReq
*
)
pMsg
->
pCont
;
SStreamMeta
*
pMeta
=
pTq
->
pStreamMeta
;
SStreamScanHistoryReq
*
pReq
=
(
SStreamScanHistoryReq
*
)
msg
;
int32_t
code
=
TSDB_CODE_SUCCESS
;
SStreamTask
*
pTask
=
streamMetaAcquireTask
(
pMeta
,
pReq
->
taskId
);
if
(
pTask
==
NULL
)
{
tqError
(
"vgId:%d failed to acquire stream task:0x%x during stream recover, task may have been destroyed"
,
...
...
@@ -1167,12 +1164,20 @@ int32_t tqProcessTaskScanHistory(STQ* pTq, SRpcMsg* pMsg) {
return
-
1
;
}
// do recovery step
1
// do recovery step1
const
char
*
id
=
pTask
->
id
.
idStr
;
const
char
*
pStatus
=
streamGetTaskStatusStr
(
pTask
->
status
.
taskStatus
);
tqDebug
(
"s-task:%s start
history data scan
stage(step 1), status:%s"
,
id
,
pStatus
);
tqDebug
(
"s-task:%s start
scan-history
stage(step 1), status:%s"
,
id
,
pStatus
);
int64_t
st
=
taosGetTimestampMs
();
if
(
pTask
->
tsInfo
.
step1Start
==
0
)
{
ASSERT
(
pTask
->
status
.
pauseAllowed
==
false
);
pTask
->
tsInfo
.
step1Start
=
taosGetTimestampMs
();
if
(
pTask
->
info
.
fillHistory
==
1
)
{
streamTaskEnablePause
(
pTask
);
}
}
else
{
tqDebug
(
"s-task:%s resume from paused, start ts:%"
PRId64
,
pTask
->
id
.
idStr
,
pTask
->
tsInfo
.
step1Start
);
}
// we have to continue retrying to successfully execute the scan history task.
int8_t
schedStatus
=
atomic_val_compare_exchange_8
(
&
pTask
->
status
.
schedStatus
,
TASK_SCHED_STATUS__INACTIVE
,
...
...
@@ -1185,31 +1190,21 @@ int32_t tqProcessTaskScanHistory(STQ* pTq, SRpcMsg* pMsg) {
return
0
;
}
ASSERT
(
pTask
->
status
.
pauseAllowed
==
false
);
if
(
pTask
->
info
.
fillHistory
==
1
)
{
streamTaskEnablePause
(
pTask
);
ASSERT
(
pTask
->
status
.
pauseAllowed
==
true
);
}
if
(
!
streamTaskRecoverScanStep1Finished
(
pTask
))
{
streamSourceScanHistoryData
(
pTask
);
}
// disable the pause when handling the step2 scan of tsdb data.
// the whole next procedure cann't be stopped.
// todo fix it: the following procedure should be executed completed and then shutdown when trying to close vnode.
if
(
pTask
->
info
.
fillHistory
==
1
)
{
streamTaskDisablePause
(
pTask
);
}
if
(
streamTaskShouldStop
(
&
pTask
->
status
)
||
streamTaskShouldPause
(
&
pTask
->
status
))
{
tqDebug
(
"s-task:%s is dropped or paused, abort recover in step1"
,
id
);
if
(
pTask
->
status
.
taskStatus
==
TASK_STATUS__PAUSE
)
{
double
el
=
(
taosGetTimestampMs
()
-
pTask
->
tsInfo
.
step1Start
)
/
1000
.
0
;
tqDebug
(
"s-task:%s is paused in the step1, elapsed time:%.2fs, sched-status:%d"
,
pTask
->
id
.
idStr
,
el
,
TASK_SCHED_STATUS__INACTIVE
);
atomic_store_8
(
&
pTask
->
status
.
schedStatus
,
TASK_SCHED_STATUS__INACTIVE
);
streamMetaReleaseTask
(
pMeta
,
pTask
);
return
0
;
}
double
el
=
(
taosGetTimestampMs
()
-
st
)
/
1000
.
0
;
// the following procedure should be executed, no matter status is stop/pause or not
double
el
=
(
taosGetTimestampMs
()
-
pTask
->
tsInfo
.
step1Start
)
/
1000
.
0
;
tqDebug
(
"s-task:%s scan-history stage(step 1) ended, elapsed time:%.2fs"
,
id
,
el
);
if
(
pTask
->
info
.
fillHistory
)
{
...
...
@@ -1217,25 +1212,24 @@ int32_t tqProcessTaskScanHistory(STQ* pTq, SRpcMsg* pMsg) {
SStreamTask
*
pStreamTask
=
NULL
;
bool
done
=
false
;
if
(
!
pReq
->
igUntreated
&&
!
streamTaskRecoverScanStep1Finished
(
pTask
))
{
// 1. stop the related stream task, get the current scan wal version of stream task, ver.
// 1. get the related stream task
pStreamTask
=
streamMetaAcquireTask
(
pMeta
,
pTask
->
streamTaskId
.
taskId
);
if
(
pStreamTask
==
NULL
)
{
// todo delete this task, if the related stream task is dropped
qError
(
"failed to find s-task:0x%x, it may have been destroyed, drop fill-history task:%s"
,
pTask
->
streamTaskId
.
taskId
,
pTask
->
id
.
idStr
);
pTask
->
status
.
taskStatus
=
TASK_STATUS__DROPPING
;
tqDebug
(
"s-task:%s fill-history task set status to be dropping"
,
id
);
streamMetaSaveTask
(
pMeta
,
pTask
);
streamMetaUnregisterTask
(
pMeta
,
pTask
->
id
.
taskId
);
streamMetaReleaseTask
(
pMeta
,
pTask
);
return
-
1
;
}
ASSERT
(
pStreamTask
->
info
.
taskLevel
==
TASK_LEVEL__SOURCE
);
// stream task in TASK_STATUS__SCAN_HISTORY can not be paused.
// wait for the
stream task get ready for scan history data
// 2. it cannot be paused, when the stream task in TASK_STATUS__SCAN_HISTORY status. Let's wait for the
//
stream task get ready for scan history data
while
(
pStreamTask
->
status
.
taskStatus
==
TASK_STATUS__SCAN_HISTORY
)
{
tqDebug
(
"s-task:%s level:%d related stream task:%s(status:%s) not ready for halt, wait for it and recheck in 100ms"
,
...
...
@@ -1245,6 +1239,7 @@ int32_t tqProcessTaskScanHistory(STQ* pTq, SRpcMsg* pMsg) {
// now we can stop the stream task execution
streamTaskHalt
(
pStreamTask
);
tqDebug
(
"s-task:%s level:%d sched-status:%d is halt by fill-history task:%s"
,
pStreamTask
->
id
.
idStr
,
pStreamTask
->
info
.
taskLevel
,
pStreamTask
->
status
.
schedStatus
,
id
);
...
...
@@ -1252,42 +1247,36 @@ int32_t tqProcessTaskScanHistory(STQ* pTq, SRpcMsg* pMsg) {
pRange
=
&
pTask
->
dataRange
.
range
;
int64_t
latestVer
=
walReaderGetCurrentVer
(
pStreamTask
->
exec
.
pWalReader
);
done
=
streamHistoryTaskSetVerRangeStep2
(
pTask
,
latestVer
);
}
if
(
done
)
{
pTask
->
tsInfo
.
step2Start
=
taosGetTimestampMs
();
streamTaskEndScanWAL
(
pTask
);
streamMetaReleaseTask
(
pMeta
,
pTask
);
}
else
{
if
(
!
streamTaskRecoverScanStep1Finished
(
pTask
))
{
STimeWindow
*
pWindow
=
&
pTask
->
dataRange
.
window
;
tqDebug
(
"s-task:%s level:%d verRange:%"
PRId64
" - %"
PRId64
" window:%"
PRId64
"-%"
PRId64
", do secondary scan-history from WAL after halt the related stream task:%s"
,
id
,
pTask
->
info
.
taskLevel
,
pRange
->
minVer
,
pRange
->
maxVer
,
pWindow
->
skey
,
pWindow
->
ekey
,
id
);
id
,
pTask
->
info
.
taskLevel
,
pRange
->
minVer
,
pRange
->
maxVer
,
pWindow
->
skey
,
pWindow
->
ekey
,
pStreamTask
->
id
.
idStr
);
ASSERT
(
pTask
->
status
.
schedStatus
==
TASK_SCHED_STATUS__WAITING
);
pTask
->
tsInfo
.
step2Start
=
taosGetTimestampMs
();
streamSetParamForStreamScannerStep2
(
pTask
,
pRange
,
pWindow
);
}
if
(
!
streamTaskRecoverScanStep2Finished
(
pTask
))
{
pTask
->
status
.
taskStatus
=
TASK_STATUS__SCAN_HISTORY_WAL
;
if
(
streamTaskShouldStop
(
&
pTask
->
status
)
||
streamTaskShouldPause
(
&
pTask
->
status
))
{
tqDebug
(
"s-task:%s is dropped or paused, abort recover in step1"
,
id
);
streamMetaReleaseTask
(
pMeta
,
pTask
);
return
0
;
}
int64_t
dstVer
=
pTask
->
dataRange
.
range
.
minVer
-
1
;
pTask
->
chkInfo
.
currentVer
=
dstVer
;
walReaderSetSkipToVersion
(
pTask
->
exec
.
pWalReader
,
dstVer
);
tqDebug
(
"s-task:%s wal reader start scan from WAL ver:%"
PRId64
", set sched-status:%d"
,
id
,
dstVer
,
TASK_SCHED_STATUS__INACTIVE
);
}
tqDebug
(
"s-task:%s wal reader start scan WAL verRange:%"
PRId64
"-%"
PRId64
", set sched-status:%d"
,
id
,
dstVer
,
pTask
->
dataRange
.
range
.
maxVer
,
TASK_SCHED_STATUS__INACTIVE
);
atomic_store_8
(
&
pTask
->
status
.
schedStatus
,
TASK_SCHED_STATUS__INACTIVE
);
// set the fill-history task to be normal
if
(
pTask
->
info
.
fillHistory
==
1
)
{
streamSetStatusNormal
(
pTask
);
}
// 4. 1) transfer the ownership of executor state, 2) update the scan data range for source task.
// 5. resume the related stream task.
streamMetaReleaseTask
(
pMeta
,
pTask
);
...
...
@@ -1304,7 +1293,7 @@ int32_t tqProcessTaskScanHistory(STQ* pTq, SRpcMsg* pMsg) {
if
(
pTask
->
historyTaskId
.
taskId
==
0
)
{
*
pWindow
=
(
STimeWindow
){
INT64_MIN
,
INT64_MAX
};
tqDebug
(
"s-task:%s scanhistory in stream time window completed, no related fill-history task, reset the time "
"s-task:%s scan
-
history in stream time window completed, no related fill-history task, reset the time "
"window:%"
PRId64
" - %"
PRId64
,
id
,
pWindow
->
skey
,
pWindow
->
ekey
);
qResetStreamInfoTimeWindow
(
pTask
->
exec
.
pExecutor
);
...
...
@@ -1500,7 +1489,7 @@ int32_t tqProcessTaskRunReq(STQ* pTq, SRpcMsg* pMsg) {
if
(
pTask
!=
NULL
)
{
// even in halt status, the data in inputQ must be processed
int8_t
st
=
pTask
->
status
.
taskStatus
;
if
(
st
==
TASK_STATUS__NORMAL
||
st
==
TASK_STATUS__SCAN_HISTORY
||
st
==
TASK_STATUS__SCAN_HISTORY_WAL
)
{
if
(
st
==
TASK_STATUS__NORMAL
||
st
==
TASK_STATUS__SCAN_HISTORY
/* || st == TASK_STATUS__SCAN_HISTORY_WAL*/
)
{
tqDebug
(
"vgId:%d s-task:%s start to process block from inputQ, last chk point:%"
PRId64
,
vgId
,
pTask
->
id
.
idStr
,
pTask
->
chkInfo
.
version
);
streamProcessRunReq
(
pTask
);
...
...
@@ -1637,7 +1626,7 @@ int32_t tqProcessTaskResumeImpl(STQ* pTq, SStreamTask* pTask, int64_t sversion,
vgId
,
pTask
->
id
.
idStr
,
pTask
->
chkInfo
.
currentVer
,
sversion
,
pTask
->
status
.
schedStatus
);
}
if
(
level
==
TASK_LEVEL__SOURCE
&&
pTask
->
info
.
fillHistory
)
{
if
(
level
==
TASK_LEVEL__SOURCE
&&
pTask
->
info
.
fillHistory
&&
pTask
->
status
.
taskStatus
==
TASK_STATUS__SCAN_HISTORY
)
{
streamStartRecoverTask
(
pTask
,
igUntreated
);
}
else
if
(
level
==
TASK_LEVEL__SOURCE
&&
(
taosQueueItemSize
(
pTask
->
inputQueue
->
queue
)
==
0
))
{
tqStartStreamTasks
(
pTq
);
...
...
source/dnode/vnode/src/tq/tqRestore.c
浏览文件 @
3d231831
...
...
@@ -211,7 +211,7 @@ int32_t doSetOffsetForWalReader(SStreamTask *pTask, int32_t vgId) {
static
void
checkForFillHistoryVerRange
(
SStreamTask
*
pTask
,
int64_t
ver
)
{
if
((
pTask
->
info
.
fillHistory
==
1
)
&&
ver
>
pTask
->
dataRange
.
range
.
maxVer
)
{
qWarn
(
"s-task:%s fill-history scan WAL, currentVer:%"
PRId64
"reach the maximum ver:%"
PRId64
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
;
...
...
@@ -256,14 +256,15 @@ int32_t createStreamTaskRunReq(SStreamMeta* pStreamMeta, bool* pScanIdle) {
continue
;
}
if
(
status
!=
TASK_STATUS__NORMAL
&&
status
!=
TASK_STATUS__SCAN_HISTORY_WAL
)
{
if
(
status
!=
TASK_STATUS__NORMAL
/* && status != TASK_STATUS__SCAN_HISTORY_WAL*/
)
{
tqDebug
(
"s-task:%s not ready for new submit block from wal, status:%s"
,
pTask
->
id
.
idStr
,
streamGetTaskStatusStr
(
status
));
streamMetaReleaseTask
(
pStreamMeta
,
pTask
);
continue
;
}
if
((
pTask
->
info
.
fillHistory
==
1
)
&&
pTask
->
status
.
transferState
)
{
ASSERT
(
status
==
TASK_STATUS__SCAN_HISTORY_WAL
);
// ASSERT(status == TASK_STATUS__SCAN_HISTORY_WAL);
ASSERT
(
status
==
TASK_STATUS__NORMAL
);
// the maximum version of data in the WAL has reached already, the step2 is done
tqDebug
(
"s-task:%s fill-history reach the maximum ver:%"
PRId64
", not scan wal anymore"
,
pTask
->
id
.
idStr
,
pTask
->
dataRange
.
range
.
maxVer
);
...
...
source/libs/executor/src/scanoperator.c
浏览文件 @
3d231831
...
...
@@ -2121,8 +2121,7 @@ FETCH_NEXT_BLOCK:
return
pInfo
->
pUpdateRes
;
}
SSDataBlock
*
pBlock
=
pInfo
->
pRes
;
SDataBlockInfo
*
pBlockInfo
=
&
pBlock
->
info
;
SDataBlockInfo
*
pBlockInfo
=
&
pInfo
->
pRes
->
info
;
int32_t
totalBlocks
=
taosArrayGetSize
(
pInfo
->
pBlockLists
);
NEXT_SUBMIT_BLK:
...
...
@@ -2146,35 +2145,36 @@ FETCH_NEXT_BLOCK:
}
}
blockDataCleanup
(
p
Block
);
blockDataCleanup
(
p
Info
->
pRes
);
while
(
pAPI
->
tqReaderFn
.
tqNextBlockImpl
(
pInfo
->
tqReader
,
id
))
{
SSDataBlock
*
pRes
=
NULL
;
int32_t
code
=
pAPI
->
tqReaderFn
.
tqRetrieveBlock
(
pInfo
->
tqReader
,
&
pRes
,
id
);
qDebug
(
"retrieve data from submit completed code:%s, rows:%"
PRId64
" %s"
,
tstrerror
(
code
),
pRes
->
info
.
rows
,
id
);
qDebug
(
"retrieve data from submit completed code:%s rows:%"
PRId64
" %s"
,
tstrerror
(
code
),
pRes
->
info
.
rows
,
id
);
if
(
code
!=
TSDB_CODE_SUCCESS
||
pRes
->
info
.
rows
==
0
)
{
qDebug
(
"retrieve data failed, try next block in submit block, %s"
,
id
);
continue
;
}
setBlockIntoRes
(
pInfo
,
pRes
,
false
);
// filter the block extracted from WAL files, according to the time window
// apply additional time window filter
doBlockDataWindowFilter
(
pRes
,
pInfo
->
primaryTsIndex
,
&
pStreamInfo
->
fillHistoryWindow
,
id
);
blockDataUpdateTsWindow
(
pInfo
->
pRes
,
pInfo
->
primaryTsIndex
);
if
(
pRes
->
info
.
rows
==
0
)
{
continue
;
}
setBlockIntoRes
(
pInfo
,
pRes
,
false
);
if
(
pInfo
->
pCreateTbRes
->
info
.
rows
>
0
)
{
pInfo
->
scanMode
=
STREAM_SCAN_FROM_RES
;
qDebug
(
"create table res exists, rows:%"
PRId64
" return from stream scan, %s"
,
pInfo
->
pCreateTbRes
->
info
.
rows
,
id
);
return
pInfo
->
pCreateTbRes
;
}
// apply additional time window filter
doBlockDataWindowFilter
(
pBlock
,
pInfo
->
primaryTsIndex
,
&
pStreamInfo
->
fillHistoryWindow
,
id
);
pBlock
->
info
.
dataLoad
=
1
;
blockDataUpdateTsWindow
(
pBlock
,
pInfo
->
primaryTsIndex
);
doCheckUpdate
(
pInfo
,
pBlockInfo
->
window
.
ekey
,
pBlock
);
doFilter
(
pBlock
,
pOperator
->
exprSupp
.
pFilterInfo
,
NULL
);
doCheckUpdate
(
pInfo
,
pBlockInfo
->
window
.
ekey
,
pInfo
->
pRes
);
doFilter
(
pInfo
->
pRes
,
pOperator
->
exprSupp
.
pFilterInfo
,
NULL
);
int64_t
numOfUpdateRes
=
pInfo
->
pUpdateDataRes
->
info
.
rows
;
qDebug
(
"%s %"
PRId64
" rows in datablock, update res:%"
PRId64
,
id
,
pBlockInfo
->
rows
,
numOfUpdateRes
);
...
...
@@ -2196,7 +2196,7 @@ FETCH_NEXT_BLOCK:
qDebug
(
"stream scan completed, and return source rows:%"
PRId64
", %s"
,
pBlockInfo
->
rows
,
id
);
if
(
pBlockInfo
->
rows
>
0
)
{
return
p
Block
;
return
p
Info
->
pRes
;
}
if
(
pInfo
->
pUpdateDataRes
->
info
.
rows
>
0
)
{
...
...
source/libs/stream/src/streamExec.c
浏览文件 @
3d231831
...
...
@@ -172,6 +172,12 @@ int32_t streamScanExec(SStreamTask* pTask, int32_t batchSz) {
bool
finished
=
false
;
while
(
1
)
{
if
(
streamTaskShouldPause
(
&
pTask
->
status
))
{
double
el
=
(
taosGetTimestampMs
()
-
pTask
->
tsInfo
.
step1Start
)
/
1000
.
0
;
qDebug
(
"s-task:%s paused from the scan-history task, elapsed time:%.2fsec"
,
pTask
->
id
.
idStr
,
el
);
return
0
;
}
SArray
*
pRes
=
taosArrayInit
(
0
,
sizeof
(
SSDataBlock
));
if
(
pRes
==
NULL
)
{
terrno
=
TSDB_CODE_OUT_OF_MEMORY
;
...
...
@@ -404,6 +410,8 @@ static int32_t streamTransferStateToStreamTask(SStreamTask* pTask) {
streamTaskReleaseState
(
pTask
);
streamTaskReloadState
(
pStreamTask
);
// clear the link between fill-history task and stream task info
pStreamTask
->
historyTaskId
.
taskId
=
0
;
streamTaskResumeFromHalt
(
pStreamTask
);
qDebug
(
"s-task:%s fill-history task set status to be dropping, save the state into disk"
,
pTask
->
id
.
idStr
);
...
...
@@ -414,6 +422,7 @@ static int32_t streamTransferStateToStreamTask(SStreamTask* pTask) {
// save to disk
taosWLockLatch
(
&
pMeta
->
lock
);
streamMetaSaveTask
(
pMeta
,
pStreamTask
);
if
(
streamMetaCommit
(
pMeta
)
<
0
)
{
// persist to disk
...
...
@@ -615,7 +624,7 @@ int32_t streamTryExec(SStreamTask* pTask) {
// todo the task should be commit here
if
(
taosQueueEmpty
(
pTask
->
inputQueue
->
queue
))
{
// fill-history WAL scan has completed
if
(
pTask
->
status
.
taskStatus
==
TASK_STATUS__SCAN_HISTORY_WAL
&&
pTask
->
status
.
transferState
==
true
)
{
if
(
pTask
->
info
.
taskLevel
==
TASK_LEVEL__SOURCE
&&
pTask
->
status
.
transferState
==
true
)
{
streamTaskRecoverSetAllStepFinished
(
pTask
);
streamTaskEndScanWAL
(
pTask
);
}
else
{
...
...
source/libs/stream/src/streamRecover.c
浏览文件 @
3d231831
...
...
@@ -85,6 +85,7 @@ int32_t streamTaskLaunchScanHistory(SStreamTask* pTask) {
if
(
pTask
->
info
.
fillHistory
)
{
streamSetParamForScanHistory
(
pTask
);
}
streamTaskEnablePause
(
pTask
);
streamTaskScanHistoryPrepare
(
pTask
);
}
else
if
(
pTask
->
info
.
taskLevel
==
TASK_LEVEL__SINK
)
{
qDebug
(
"s-task:%s sink task do nothing to handle scan-history"
,
pTask
->
id
.
idStr
);
...
...
@@ -839,7 +840,7 @@ void streamTaskPause(SStreamTask* pTask) {
return
;
}
while
(
!
pTask
->
status
.
pauseAllowed
||
(
pTask
->
status
.
taskStatus
==
TASK_STATUS__HALT
))
{
while
(
!
pTask
->
status
.
pauseAllowed
||
(
pTask
->
status
.
taskStatus
==
TASK_STATUS__HALT
))
{
status
=
pTask
->
status
.
taskStatus
;
if
(
status
==
TASK_STATUS__DROPPING
)
{
qDebug
(
"vgId:%d s-task:%s task already dropped, do nothing"
,
pMeta
->
vgId
,
pTask
->
id
.
idStr
);
...
...
@@ -856,8 +857,19 @@ void streamTaskPause(SStreamTask* pTask) {
taosMsleep
(
100
);
}
// todo: use the lock of the task.
taosWLockLatch
(
&
pMeta
->
lock
);
status
=
pTask
->
status
.
taskStatus
;
if
(
status
==
TASK_STATUS__DROPPING
||
status
==
TASK_STATUS__STOP
)
{
taosWUnLockLatch
(
&
pMeta
->
lock
);
qDebug
(
"vgId:%d s-task:%s task already dropped/stopped/paused, do nothing"
,
pMeta
->
vgId
,
pTask
->
id
.
idStr
);
return
;
}
atomic_store_8
(
&
pTask
->
status
.
keepTaskStatus
,
pTask
->
status
.
taskStatus
);
atomic_store_8
(
&
pTask
->
status
.
taskStatus
,
TASK_STATUS__PAUSE
);
taosWUnLockLatch
(
&
pMeta
->
lock
);
int64_t
el
=
taosGetTimestampMs
()
-
st
;
qDebug
(
"vgId:%d s-task:%s set pause flag, prev:%s, elapsed time:%dms"
,
pMeta
->
vgId
,
pTask
->
id
.
idStr
,
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录