Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
1b9818e1
T
TDengine
项目概览
taosdata
/
TDengine
大约 1 年 前同步成功
通知
1184
Star
22015
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,发现更多精彩内容 >>
提交
1b9818e1
编写于
5月 16, 2022
作者:
M
Minghao Li
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
fix(sync): set handle when send response
上级
b6e3f9d1
变更
2
显示空白变更内容
内联
并排
Showing
2 changed file
with
31 addition
and
10 deletion
+31
-10
source/dnode/mgmt/mgmt_vnode/src/vmWorker.c
source/dnode/mgmt/mgmt_vnode/src/vmWorker.c
+17
-1
source/dnode/vnode/src/vnd/vnodeSvr.c
source/dnode/vnode/src/vnd/vnodeSvr.c
+14
-9
未找到文件。
source/dnode/mgmt/mgmt_vnode/src/vmWorker.c
浏览文件 @
1b9818e1
...
...
@@ -214,7 +214,23 @@ static void vmProcessSyncQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOf
// todo
SRpcMsg
*
pRsp
=
NULL
;
(
void
)
vnodeProcessSyncReq
(
pVnode
->
pImpl
,
&
pMsg
->
rpcMsg
,
&
pRsp
);
int32_t
ret
=
vnodeProcessSyncReq
(
pVnode
->
pImpl
,
&
pMsg
->
rpcMsg
,
&
pRsp
);
if
(
ret
!=
0
)
{
// if leader, send response
if
(
pMsg
->
rpcMsg
.
handle
!=
NULL
)
{
SRpcMsg
rsp
;
rsp
.
pCont
=
NULL
;
rsp
.
contLen
=
0
;
rsp
.
code
=
terrno
;
dTrace
(
"vmProcessSyncQueue error, code:%d"
,
terrno
);
rsp
.
ahandle
=
pMsg
->
rpcMsg
.
ahandle
;
rsp
.
handle
=
pMsg
->
rpcMsg
.
handle
;
rsp
.
refId
=
pMsg
->
rpcMsg
.
refId
;
tmsgSendRsp
(
&
rsp
);
}
}
rpcFreeCont
(
pMsg
->
rpcMsg
.
pCont
);
taosFreeQitem
(
pMsg
);
...
...
source/dnode/vnode/src/vnd/vnodeSvr.c
浏览文件 @
1b9818e1
...
...
@@ -199,6 +199,8 @@ void smaHandleRes(void *pVnode, int64_t smaId, const SArray *data) {
}
int
vnodeProcessSyncReq
(
SVnode
*
pVnode
,
SRpcMsg
*
pMsg
,
SRpcMsg
**
pRsp
)
{
int32_t
ret
=
TAOS_SYNC_PROPOSE_OTHER_ERROR
;
if
(
syncEnvIsStart
())
{
SSyncNode
*
pSyncNode
=
syncNodeAcquire
(
pVnode
->
sync
);
assert
(
pSyncNode
!=
NULL
);
...
...
@@ -220,67 +222,70 @@ int vnodeProcessSyncReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
SyncTimeout
*
pSyncMsg
=
syncTimeoutFromRpcMsg2
(
pRpcMsg
);
assert
(
pSyncMsg
!=
NULL
);
syncNodeOnTimeoutCb
(
pSyncNode
,
pSyncMsg
);
ret
=
syncNodeOnTimeoutCb
(
pSyncNode
,
pSyncMsg
);
syncTimeoutDestroy
(
pSyncMsg
);
}
else
if
(
pRpcMsg
->
msgType
==
TDMT_VND_SYNC_PING
)
{
SyncPing
*
pSyncMsg
=
syncPingFromRpcMsg2
(
pRpcMsg
);
assert
(
pSyncMsg
!=
NULL
);
syncNodeOnPingCb
(
pSyncNode
,
pSyncMsg
);
ret
=
syncNodeOnPingCb
(
pSyncNode
,
pSyncMsg
);
syncPingDestroy
(
pSyncMsg
);
}
else
if
(
pRpcMsg
->
msgType
==
TDMT_VND_SYNC_PING_REPLY
)
{
SyncPingReply
*
pSyncMsg
=
syncPingReplyFromRpcMsg2
(
pRpcMsg
);
assert
(
pSyncMsg
!=
NULL
);
syncNodeOnPingReplyCb
(
pSyncNode
,
pSyncMsg
);
ret
=
syncNodeOnPingReplyCb
(
pSyncNode
,
pSyncMsg
);
syncPingReplyDestroy
(
pSyncMsg
);
}
else
if
(
pRpcMsg
->
msgType
==
TDMT_VND_SYNC_CLIENT_REQUEST
)
{
SyncClientRequest
*
pSyncMsg
=
syncClientRequestFromRpcMsg2
(
pRpcMsg
);
assert
(
pSyncMsg
!=
NULL
);
syncNodeOnClientRequestCb
(
pSyncNode
,
pSyncMsg
);
ret
=
syncNodeOnClientRequestCb
(
pSyncNode
,
pSyncMsg
);
syncClientRequestDestroy
(
pSyncMsg
);
}
else
if
(
pRpcMsg
->
msgType
==
TDMT_VND_SYNC_REQUEST_VOTE
)
{
SyncRequestVote
*
pSyncMsg
=
syncRequestVoteFromRpcMsg2
(
pRpcMsg
);
assert
(
pSyncMsg
!=
NULL
);
syncNodeOnRequestVoteCb
(
pSyncNode
,
pSyncMsg
);
ret
=
syncNodeOnRequestVoteCb
(
pSyncNode
,
pSyncMsg
);
syncRequestVoteDestroy
(
pSyncMsg
);
}
else
if
(
pRpcMsg
->
msgType
==
TDMT_VND_SYNC_REQUEST_VOTE_REPLY
)
{
SyncRequestVoteReply
*
pSyncMsg
=
syncRequestVoteReplyFromRpcMsg2
(
pRpcMsg
);
assert
(
pSyncMsg
!=
NULL
);
syncNodeOnRequestVoteReplyCb
(
pSyncNode
,
pSyncMsg
);
ret
=
syncNodeOnRequestVoteReplyCb
(
pSyncNode
,
pSyncMsg
);
syncRequestVoteReplyDestroy
(
pSyncMsg
);
}
else
if
(
pRpcMsg
->
msgType
==
TDMT_VND_SYNC_APPEND_ENTRIES
)
{
SyncAppendEntries
*
pSyncMsg
=
syncAppendEntriesFromRpcMsg2
(
pRpcMsg
);
assert
(
pSyncMsg
!=
NULL
);
syncNodeOnAppendEntriesCb
(
pSyncNode
,
pSyncMsg
);
ret
=
syncNodeOnAppendEntriesCb
(
pSyncNode
,
pSyncMsg
);
syncAppendEntriesDestroy
(
pSyncMsg
);
}
else
if
(
pRpcMsg
->
msgType
==
TDMT_VND_SYNC_APPEND_ENTRIES_REPLY
)
{
SyncAppendEntriesReply
*
pSyncMsg
=
syncAppendEntriesReplyFromRpcMsg2
(
pRpcMsg
);
assert
(
pSyncMsg
!=
NULL
);
syncNodeOnAppendEntriesReplyCb
(
pSyncNode
,
pSyncMsg
);
ret
=
syncNodeOnAppendEntriesReplyCb
(
pSyncNode
,
pSyncMsg
);
syncAppendEntriesReplyDestroy
(
pSyncMsg
);
}
else
{
vError
(
"==vnodeProcessSyncReq== error msg type:%d"
,
pRpcMsg
->
msgType
);
ret
=
TAOS_SYNC_PROPOSE_OTHER_ERROR
;
}
syncNodeRelease
(
pSyncNode
);
}
else
{
vError
(
"==vnodeProcessSyncReq== error syncEnv stop"
);
ret
=
TAOS_SYNC_PROPOSE_OTHER_ERROR
;
}
return
0
;
return
ret
;
}
static
int
vnodeProcessCreateStbReq
(
SVnode
*
pVnode
,
int64_t
version
,
void
*
pReq
,
int
len
,
SRpcMsg
*
pRsp
)
{
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录