Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
1061eef1
TDengine
项目概览
taosdata
/
TDengine
大约 1 年 前同步成功
通知
1181
Star
22014
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看板
体验新版 GitCode,发现更多精彩内容 >>
提交
1061eef1
编写于
4月 19, 2023
作者:
wmmhello
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
fix:move consumer msg from fetch thread to query thread
上级
25542c26
变更
2
显示空白变更内容
内联
并排
Showing
2 changed file
with
10 addition
and
10 deletion
+10
-10
source/dnode/mgmt/mgmt_vnode/src/vmHandle.c
source/dnode/mgmt/mgmt_vnode/src/vmHandle.c
+1
-1
source/dnode/vnode/src/vnd/vnodeSvr.c
source/dnode/vnode/src/vnd/vnodeSvr.c
+9
-9
未找到文件。
source/dnode/mgmt/mgmt_vnode/src/vmHandle.c
浏览文件 @
1061eef1
...
...
@@ -519,7 +519,7 @@ SArray *vmGetMsgHandles() {
if
(
dmSetMgmtHandle
(
pArray
,
TDMT_VND_TMQ_COMMIT_OFFSET
,
vmPutMsgToWriteQueue
,
0
)
==
NULL
)
goto
_OVER
;
if
(
dmSetMgmtHandle
(
pArray
,
TDMT_VND_TMQ_ADD_CHECKINFO
,
vmPutMsgToWriteQueue
,
0
)
==
NULL
)
goto
_OVER
;
if
(
dmSetMgmtHandle
(
pArray
,
TDMT_VND_TMQ_DEL_CHECKINFO
,
vmPutMsgToWriteQueue
,
0
)
==
NULL
)
goto
_OVER
;
if
(
dmSetMgmtHandle
(
pArray
,
TDMT_VND_TMQ_CONSUME
,
vmPutMsgTo
Fetch
Queue
,
0
)
==
NULL
)
goto
_OVER
;
if
(
dmSetMgmtHandle
(
pArray
,
TDMT_VND_TMQ_CONSUME
,
vmPutMsgTo
Query
Queue
,
0
)
==
NULL
)
goto
_OVER
;
if
(
dmSetMgmtHandle
(
pArray
,
TDMT_VND_DELETE
,
vmPutMsgToWriteQueue
,
0
)
==
NULL
)
goto
_OVER
;
if
(
dmSetMgmtHandle
(
pArray
,
TDMT_VND_BATCH_DEL
,
vmPutMsgToWriteQueue
,
0
)
==
NULL
)
goto
_OVER
;
if
(
dmSetMgmtHandle
(
pArray
,
TDMT_VND_COMMIT
,
vmPutMsgToWriteQueue
,
0
)
==
NULL
)
goto
_OVER
;
...
...
source/dnode/vnode/src/vnd/vnodeSvr.c
浏览文件 @
1061eef1
...
...
@@ -487,11 +487,16 @@ int32_t vnodePreprocessQueryMsg(SVnode *pVnode, SRpcMsg *pMsg) {
int32_t
vnodeProcessQueryMsg
(
SVnode
*
pVnode
,
SRpcMsg
*
pMsg
)
{
vTrace
(
"message in vnode query queue is processing"
);
if
((
pMsg
->
msgType
==
TDMT_SCH_QUERY
)
&&
!
syncIsReadyForRead
(
pVnode
->
sync
))
{
if
((
pMsg
->
msgType
==
TDMT_SCH_QUERY
||
pMsg
->
msgType
==
TDMT_VND_TMQ_CONSUME
)
&&
!
syncIsReadyForRead
(
pVnode
->
sync
))
{
vnodeRedirectRpcMsg
(
pVnode
,
pMsg
,
terrno
);
return
0
;
}
if
(
pMsg
->
msgType
==
TDMT_VND_TMQ_CONSUME
&&
!
pVnode
->
restored
)
{
vnodeRedirectRpcMsg
(
pVnode
,
pMsg
,
TSDB_CODE_SYN_RESTORING
);
return
0
;
}
SReadHandle
handle
=
{.
meta
=
pVnode
->
pMeta
,
.
config
=
&
pVnode
->
config
,
.
vnode
=
pVnode
,
.
pMsgCb
=
&
pVnode
->
msgCb
};
switch
(
pMsg
->
msgType
)
{
case
TDMT_SCH_QUERY
:
...
...
@@ -499,6 +504,8 @@ int32_t vnodeProcessQueryMsg(SVnode *pVnode, SRpcMsg *pMsg) {
return
qWorkerProcessQueryMsg
(
&
handle
,
pVnode
->
pQuery
,
pMsg
,
0
);
case
TDMT_SCH_QUERY_CONTINUE
:
return
qWorkerProcessCQueryMsg
(
&
handle
,
pVnode
->
pQuery
,
pMsg
,
0
);
case
TDMT_VND_TMQ_CONSUME
:
return
tqProcessPollReq
(
pVnode
->
pTq
,
pMsg
);
default:
vError
(
"unknown msg type:%d in query queue"
,
pMsg
->
msgType
);
return
TSDB_CODE_APP_ERROR
;
...
...
@@ -508,17 +515,12 @@ int32_t vnodeProcessQueryMsg(SVnode *pVnode, SRpcMsg *pMsg) {
int32_t
vnodeProcessFetchMsg
(
SVnode
*
pVnode
,
SRpcMsg
*
pMsg
,
SQueueInfo
*
pInfo
)
{
vTrace
(
"vgId:%d, msg:%p in fetch queue is processing"
,
pVnode
->
config
.
vgId
,
pMsg
);
if
((
pMsg
->
msgType
==
TDMT_SCH_FETCH
||
pMsg
->
msgType
==
TDMT_VND_TABLE_META
||
pMsg
->
msgType
==
TDMT_VND_TABLE_CFG
||
pMsg
->
msgType
==
TDMT_VND_BATCH_META
||
pMsg
->
msgType
==
TDMT_VND_TMQ_CONSUME
)
&&
pMsg
->
msgType
==
TDMT_VND_BATCH_META
)
&&
!
syncIsReadyForRead
(
pVnode
->
sync
))
{
vnodeRedirectRpcMsg
(
pVnode
,
pMsg
,
terrno
);
return
0
;
}
if
(
pMsg
->
msgType
==
TDMT_VND_TMQ_CONSUME
&&
!
pVnode
->
restored
)
{
vnodeRedirectRpcMsg
(
pVnode
,
pMsg
,
TSDB_CODE_SYN_RESTORING
);
return
0
;
}
switch
(
pMsg
->
msgType
)
{
case
TDMT_SCH_FETCH
:
case
TDMT_SCH_MERGE_FETCH
:
...
...
@@ -537,8 +539,6 @@ int32_t vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo) {
return
vnodeGetTableCfg
(
pVnode
,
pMsg
,
true
);
case
TDMT_VND_BATCH_META
:
return
vnodeGetBatchMeta
(
pVnode
,
pMsg
);
case
TDMT_VND_TMQ_CONSUME
:
return
tqProcessPollReq
(
pVnode
->
pTq
,
pMsg
);
case
TDMT_STREAM_TASK_RUN
:
return
tqProcessTaskRunReq
(
pVnode
->
pTq
,
pMsg
);
case
TDMT_STREAM_TASK_DISPATCH
:
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录