Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
10478112
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看板
提交
10478112
编写于
4月 23, 2023
作者:
B
Benguang Zhao
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
fix: update commit index from heartbeat on learner in the same way as on follower
上级
03ffbdd4
变更
1
隐藏空白更改
内联
并排
Showing
1 changed file
with
7 addition
and
39 deletion
+7
-39
source/libs/sync/src/syncMain.c
source/libs/sync/src/syncMain.c
+7
-39
未找到文件。
source/libs/sync/src/syncMain.c
浏览文件 @
10478112
...
...
@@ -2414,18 +2414,19 @@ int32_t syncNodeOnHeartbeat(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
(
void
)
syncBuildLocalCmd
(
&
rpcMsgLocalCmd
,
ths
->
vgId
);
SyncLocalCmd
*
pSyncMsg
=
rpcMsgLocalCmd
.
pCont
;
pSyncMsg
->
cmd
=
SYNC_LOCAL_CMD_FOLLOWER_CMT
;
pSyncMsg
->
cmd
=
(
ths
->
state
==
TAOS_SYNC_STATE_LEARNER
)
?
SYNC_LOCAL_CMD_LEARNER_CMT
:
SYNC_LOCAL_CMD_FOLLOWER_CMT
;
pSyncMsg
->
commitIndex
=
pMsg
->
commitIndex
;
pSyncMsg
->
currentTerm
=
pMsg
->
term
;
SyncIndex
fcIndex
=
pSyncMsg
->
commitIndex
;
if
(
ths
->
syncEqMsg
!=
NULL
&&
ths
->
msgcb
!=
NULL
)
{
int32_t
code
=
ths
->
syncEqMsg
(
ths
->
msgcb
,
&
rpcMsgLocalCmd
);
if
(
code
!=
0
)
{
sError
(
"vgId:%d,
sync enqueue fc-commit msg error, code:%d"
,
ths
->
vgId
,
code
);
sError
(
"vgId:%d,
failed to enqueue commit msg from heartbeat since %s, code:%d"
,
ths
->
vgId
,
terrstr
()
,
code
);
rpcFreeCont
(
rpcMsgLocalCmd
.
pCont
);
}
else
{
sTrace
(
"vgId:%d, sync enqueue fc-commit msg, fc-index:%"
PRId64
,
ths
->
vgId
,
fcIndex
);
sTrace
(
"vgId:%d, enqueue commit msg from heartbeat, commit-index:%"
PRId64
", term:%"
PRId64
,
ths
->
vgId
,
pMsg
->
commitIndex
,
pMsg
->
term
);
}
}
}
...
...
@@ -2451,26 +2452,6 @@ int32_t syncNodeOnHeartbeat(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
}
}
if
(
pMsg
->
term
>=
currentTerm
&&
ths
->
state
==
TAOS_SYNC_STATE_LEARNER
)
{
SRpcMsg
rpcMsgLocalCmd
=
{
0
};
(
void
)
syncBuildLocalCmd
(
&
rpcMsgLocalCmd
,
ths
->
vgId
);
SyncLocalCmd
*
pSyncMsg
=
rpcMsgLocalCmd
.
pCont
;
pSyncMsg
->
cmd
=
SYNC_LOCAL_CMD_LEARNER_CMT
;
pSyncMsg
->
currentTerm
=
pMsg
->
term
;
pSyncMsg
->
commitIndex
=
pMsg
->
commitIndex
;
if
(
ths
->
syncEqMsg
!=
NULL
&&
ths
->
msgcb
!=
NULL
)
{
int32_t
code
=
ths
->
syncEqMsg
(
ths
->
msgcb
,
&
rpcMsgLocalCmd
);
if
(
code
!=
0
)
{
sError
(
"vgId:%d, sync enqueue step-down msg error, code:%d"
,
ths
->
vgId
,
code
);
rpcFreeCont
(
rpcMsgLocalCmd
.
pCont
);
}
else
{
sTrace
(
"vgId:%d, sync enqueue step-down msg, new-term:%"
PRId64
,
ths
->
vgId
,
pMsg
->
term
);
}
}
}
// reply
syncNodeSendMsgById
(
&
pMsgReply
->
destId
,
ths
,
&
rpcMsg
);
...
...
@@ -2521,7 +2502,7 @@ int32_t syncNodeOnLocalCmd(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
if
(
pMsg
->
cmd
==
SYNC_LOCAL_CMD_STEP_DOWN
)
{
syncNodeStepDown
(
ths
,
pMsg
->
currentTerm
);
}
else
if
(
pMsg
->
cmd
==
SYNC_LOCAL_CMD_FOLLOWER_CMT
)
{
}
else
if
(
pMsg
->
cmd
==
SYNC_LOCAL_CMD_FOLLOWER_CMT
||
pMsg
->
cmd
==
SYNC_LOCAL_CMD_LEARNER_CMT
)
{
if
(
syncLogBufferIsEmpty
(
ths
->
pLogBuf
))
{
sError
(
"vgId:%d, sync log buffer is empty."
,
ths
->
vgId
);
return
0
;
...
...
@@ -2534,20 +2515,7 @@ int32_t syncNodeOnLocalCmd(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
sError
(
"vgId:%d, failed to commit raft log since %s. commit index:%"
PRId64
""
,
ths
->
vgId
,
terrstr
(),
ths
->
commitIndex
);
}
}
else
if
(
pMsg
->
cmd
==
SYNC_LOCAL_CMD_LEARNER_CMT
){
if
(
syncLogBufferIsEmpty
(
ths
->
pLogBuf
))
{
sError
(
"vgId:%d, sync log buffer is empty."
,
ths
->
vgId
);
return
0
;
}
raftStoreSetTerm
(
ths
,
pMsg
->
currentTerm
);
(
void
)
syncNodeUpdateCommitIndex
(
ths
,
pMsg
->
commitIndex
);
sTrace
(
"vgId:%d, start to commit raft log in heartbeat. commit index:%"
PRId64
""
,
ths
->
vgId
,
ths
->
commitIndex
);
if
(
syncLogBufferCommit
(
ths
->
pLogBuf
,
ths
,
ths
->
commitIndex
)
<
0
)
{
sError
(
"vgId:%d, failed to commit raft log since %s. commit index:%"
PRId64
""
,
ths
->
vgId
,
terrstr
(),
ths
->
commitIndex
);
}
}
else
{
}
else
{
sError
(
"error local cmd"
);
}
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录