Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
63aa7bac
T
TDengine
项目概览
taosdata
/
TDengine
12 个月 前同步成功
通知
1180
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,发现更多精彩内容 >>
提交
63aa7bac
编写于
5月 19, 2022
作者:
S
Shengliang Guan
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
refactor: redirect msg
上级
0161b645
变更
4
显示空白变更内容
内联
并排
Showing
4 changed file
with
55 addition
and
67 deletion
+55
-67
source/dnode/mgmt/mgmt_vnode/src/vmWorker.c
source/dnode/mgmt/mgmt_vnode/src/vmWorker.c
+0
-1
source/dnode/mgmt/node_mgmt/src/dmProc.c
source/dnode/mgmt/node_mgmt/src/dmProc.c
+1
-1
source/dnode/mgmt/node_mgmt/src/dmTransport.c
source/dnode/mgmt/node_mgmt/src/dmTransport.c
+50
-61
source/libs/qworker/src/qworkerMsg.c
source/libs/qworker/src/qworkerMsg.c
+4
-4
未找到文件。
source/dnode/mgmt/mgmt_vnode/src/vmWorker.c
浏览文件 @
63aa7bac
...
...
@@ -126,7 +126,6 @@ static void vmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
syncGetEpSet
(
vnodeGetSyncHandle
(
pVnode
->
pImpl
),
&
newEpSet
);
newEpSet
.
inUse
=
(
newEpSet
.
inUse
+
1
)
%
newEpSet
.
numOfEps
;
tmsgSendRedirectRsp
(
&
rsp
,
&
newEpSet
);
}
else
if
(
ret
==
TAOS_SYNC_PROPOSE_OTHER_ERROR
)
{
rsp
.
code
=
TSDB_CODE_SYN_INTERNAL_ERROR
;
tmsgSendRsp
(
&
rsp
);
...
...
source/dnode/mgmt/node_mgmt/src/dmProc.c
浏览文件 @
63aa7bac
...
...
@@ -103,7 +103,7 @@ static inline int32_t dmPushToProcQueue(SProc *proc, SProcQueue *queue, SRpcMsg
return
-
1
;
}
if
(
ftype
==
DND_FUNC_REQ
&&
IsReq
(
pMsg
)
&&
pMsg
->
code
==
0
&&
handle
!=
0
)
{
if
(
ftype
==
DND_FUNC_REQ
&&
IsReq
(
pMsg
)
&&
pMsg
->
code
==
0
&&
handle
!=
0
&&
pMsg
->
info
.
noResp
!=
0
)
{
if
(
taosHashPut
(
proc
->
hash
,
&
handle
,
sizeof
(
int64_t
),
&
pMsg
->
info
,
sizeof
(
SRpcConnInfo
))
!=
0
)
{
taosThreadMutexUnlock
(
&
queue
->
mutex
);
return
-
1
;
...
...
source/dnode/mgmt/node_mgmt/src/dmTransport.c
浏览文件 @
63aa7bac
...
...
@@ -200,47 +200,6 @@ int32_t dmInitMsgHandle(SDnode *pDnode) {
return
0
;
}
static
void
dmSendRpcRedirectRsp
(
const
SRpcMsg
*
pMsg
)
{
SDnode
*
pDnode
=
dmInstance
();
SEpSet
epSet
=
{
0
};
dmGetMnodeEpSet
(
&
pDnode
->
data
,
&
epSet
);
dDebug
(
"RPC %p, req is redirected, num:%d use:%d"
,
pMsg
->
info
.
handle
,
epSet
.
numOfEps
,
epSet
.
inUse
);
for
(
int32_t
i
=
0
;
i
<
epSet
.
numOfEps
;
++
i
)
{
dDebug
(
"mnode index:%d %s:%u"
,
i
,
epSet
.
eps
[
i
].
fqdn
,
epSet
.
eps
[
i
].
port
);
if
(
strcmp
(
epSet
.
eps
[
i
].
fqdn
,
tsLocalFqdn
)
==
0
&&
epSet
.
eps
[
i
].
port
==
tsServerPort
)
{
epSet
.
inUse
=
(
i
+
1
)
%
epSet
.
numOfEps
;
}
epSet
.
eps
[
i
].
port
=
htons
(
epSet
.
eps
[
i
].
port
);
}
SMEpSet
msg
=
{.
epSet
=
epSet
};
int32_t
len
=
tSerializeSMEpSet
(
NULL
,
0
,
&
msg
);
SRpcMsg
rsp
=
{
.
code
=
TSDB_CODE_RPC_REDIRECT
,
.
info
=
pMsg
->
info
,
.
contLen
=
len
,
};
rsp
.
pCont
=
rpcMallocCont
(
len
);
tSerializeSMEpSet
(
rsp
.
pCont
,
len
,
&
msg
);
rpcSendResponse
(
&
rsp
);
rpcFreeCont
(
pMsg
->
pCont
);
}
static
inline
void
dmSendRecv
(
SEpSet
*
pEpSet
,
SRpcMsg
*
pReq
,
SRpcMsg
*
pRsp
)
{
SDnode
*
pDnode
=
dmInstance
();
if
(
pDnode
->
status
!=
DND_STAT_RUNNING
)
{
pRsp
->
code
=
TSDB_CODE_NODE_OFFLINE
;
rpcFreeCont
(
pReq
->
pCont
);
pReq
->
pCont
=
NULL
;
}
else
{
rpcSendRecv
(
pDnode
->
trans
.
clientRpc
,
pEpSet
,
pReq
,
pRsp
);
}
}
static
inline
int32_t
dmSendReq
(
const
SEpSet
*
pEpSet
,
SRpcMsg
*
pMsg
)
{
SDnode
*
pDnode
=
dmInstance
();
if
(
pDnode
->
status
!=
DND_STAT_RUNNING
)
{
...
...
@@ -257,39 +216,38 @@ static inline int32_t dmSendReq(const SEpSet *pEpSet, SRpcMsg *pMsg) {
static
inline
void
dmSendRsp
(
SRpcMsg
*
pMsg
)
{
SMgmtWrapper
*
pWrapper
=
pMsg
->
info
.
wrapper
;
if
(
pMsg
->
code
==
TSDB_CODE_NODE_REDIRECT
)
{
dmSendRpcRedirectRsp
(
pMsg
);
}
else
{
if
(
InChildProc
(
pWrapper
))
{
dmPutToProcPQueue
(
&
pWrapper
->
proc
,
pMsg
,
DND_FUNC_RSP
);
rpcFreeCont
(
pMsg
->
pCont
);
pMsg
->
pCont
=
NULL
;
}
else
{
rpcSendResponse
(
pMsg
);
}
}
}
static
inline
void
dmSendRedirectRsp
(
SRpcMsg
*
pMsg
,
const
SEpSet
*
pNewEpSet
)
{
SMgmtWrapper
*
pWrapper
=
pMsg
->
info
.
wrapper
;
if
(
InChildProc
(
pWrapper
))
{
dmPutToProcPQueue
(
&
pWrapper
->
proc
,
pMsg
,
DND_FUNC_RSP
);
}
else
{
SRpcMsg
rsp
=
{
0
};
SRpcMsg
rsp
=
{.
code
=
TSDB_CODE_RPC_REDIRECT
,
.
info
=
pMsg
->
info
};
SMEpSet
msg
=
{.
epSet
=
*
pNewEpSet
};
int32_t
len
=
tSerializeSMEpSet
(
NULL
,
0
,
&
msg
);
rsp
.
pCont
=
rpcMallocCont
(
len
);
rsp
.
contLen
=
len
;
tSerializeSMEpSet
(
rsp
.
pCont
,
len
,
&
msg
);
int32_t
contLen
=
tSerializeSMEpSet
(
NULL
,
0
,
&
msg
);
rsp
.
code
=
TSDB_CODE_RPC_REDIRECT
;
rsp
.
info
=
pMsg
->
info
;
rpcSendResponse
(
&
rsp
);
rsp
.
pCont
=
rpcMallocCont
(
contLen
);
if
(
rsp
.
pCont
==
NULL
)
{
terrno
=
TSDB_CODE_OUT_OF_MEMORY
;
}
else
{
tSerializeSMEpSet
(
rsp
.
pCont
,
contLen
,
&
msg
);
rsp
.
contLen
=
contLen
;
}
dmSendRsp
(
&
rsp
);
rpcFreeCont
(
pMsg
->
pCont
);
pMsg
->
pCont
=
NULL
;
}
static
inline
void
dmRegisterBrokenLinkArg
(
SRpcMsg
*
pMsg
)
{
SMgmtWrapper
*
pWrapper
=
pMsg
->
info
.
wrapper
;
if
(
InChildProc
(
pWrapper
))
{
dmPutToProcPQueue
(
&
pWrapper
->
proc
,
pMsg
,
DND_FUNC_REGIST
);
rpcFreeCont
(
pMsg
->
pCont
);
pMsg
->
pCont
=
NULL
;
}
else
{
rpcRegisterBrokenLinkArg
(
pMsg
);
}
...
...
@@ -391,3 +349,34 @@ SMsgCb dmGetMsgcb(SDnode *pDnode) {
};
return
msgCb
;
}
static
void
dmSendMnodeRedirectRsp
(
SRpcMsg
*
pMsg
)
{
SDnode
*
pDnode
=
dmInstance
();
SEpSet
epSet
=
{
0
};
dmGetMnodeEpSet
(
&
pDnode
->
data
,
&
epSet
);
dDebug
(
"msg:%p, is redirected, num:%d use:%d"
,
pMsg
,
epSet
.
numOfEps
,
epSet
.
inUse
);
for
(
int32_t
i
=
0
;
i
<
epSet
.
numOfEps
;
++
i
)
{
dDebug
(
"mnode index:%d %s:%u"
,
i
,
epSet
.
eps
[
i
].
fqdn
,
epSet
.
eps
[
i
].
port
);
if
(
strcmp
(
epSet
.
eps
[
i
].
fqdn
,
tsLocalFqdn
)
==
0
&&
epSet
.
eps
[
i
].
port
==
tsServerPort
)
{
epSet
.
inUse
=
(
i
+
1
)
%
epSet
.
numOfEps
;
}
epSet
.
eps
[
i
].
port
=
htons
(
epSet
.
eps
[
i
].
port
);
}
SRpcMsg
rsp
=
{.
code
=
TSDB_CODE_RPC_REDIRECT
,
.
info
=
pMsg
->
info
};
SMEpSet
msg
=
{.
epSet
=
epSet
};
int32_t
contLen
=
tSerializeSMEpSet
(
NULL
,
0
,
&
msg
);
rsp
.
pCont
=
rpcMallocCont
(
contLen
);
if
(
rsp
.
pCont
==
NULL
)
{
terrno
=
TSDB_CODE_OUT_OF_MEMORY
;
}
else
{
tSerializeSMEpSet
(
rsp
.
pCont
,
contLen
,
&
msg
);
rsp
.
contLen
=
contLen
;
}
dmSendRsp
(
&
rsp
);
rpcFreeCont
(
pMsg
->
pCont
);
pMsg
->
pCont
=
NULL
;
}
source/libs/qworker/src/qworkerMsg.c
浏览文件 @
63aa7bac
...
...
@@ -287,7 +287,7 @@ int32_t qwRegisterQueryBrokenLinkArg(QW_FPARAMS_DEF, SRpcHandleInfo *pConn) {
req
->
taskId
=
htobe64
(
tId
);
req
->
refId
=
htobe64
(
rId
);
SRpcMsg
p
Msg
=
{
SRpcMsg
broken
Msg
=
{
.
msgType
=
TDMT_VND_DROP_TASK
,
.
pCont
=
req
,
.
contLen
=
sizeof
(
STaskDropReq
),
...
...
@@ -295,7 +295,7 @@ int32_t qwRegisterQueryBrokenLinkArg(QW_FPARAMS_DEF, SRpcHandleInfo *pConn) {
.
info
=
*
pConn
,
};
tmsgRegisterBrokenLinkArg
(
&
p
Msg
);
tmsgRegisterBrokenLinkArg
(
&
broken
Msg
);
return
TSDB_CODE_SUCCESS
;
}
...
...
@@ -321,7 +321,7 @@ int32_t qwRegisterHbBrokenLinkArg(SQWorker *mgmt, uint64_t sId, SRpcHandleInfo *
QW_ERR_RET
(
TSDB_CODE_QRY_OUT_OF_MEMORY
);
}
SRpcMsg
p
Msg
=
{
SRpcMsg
broken
Msg
=
{
.
msgType
=
TDMT_VND_QUERY_HEARTBEAT
,
.
pCont
=
msg
,
.
contLen
=
msgSize
,
...
...
@@ -329,7 +329,7 @@ int32_t qwRegisterHbBrokenLinkArg(SQWorker *mgmt, uint64_t sId, SRpcHandleInfo *
.
info
=
*
pConn
,
};
tmsgRegisterBrokenLinkArg
(
&
p
Msg
);
tmsgRegisterBrokenLinkArg
(
&
broken
Msg
);
return
TSDB_CODE_SUCCESS
;
}
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录