Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
5268b523
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,发现更多精彩内容 >>
提交
5268b523
编写于
6月 14, 2022
作者:
M
Minghao Li
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
refactor(sync): update replica index in snapshot sender
上级
9cd54c5c
变更
2
隐藏空白更改
内联
并排
Showing
2 changed file
with
37 addition
and
21 deletion
+37
-21
source/libs/sync/src/syncMain.c
source/libs/sync/src/syncMain.c
+31
-19
source/libs/sync/src/syncSnapshot.c
source/libs/sync/src/syncSnapshot.c
+6
-2
未找到文件。
source/libs/sync/src/syncMain.c
浏览文件 @
5268b523
...
...
@@ -561,7 +561,7 @@ int32_t syncNodePropose(SSyncNode* pSyncNode, const SRpcMsg* pMsg, bool isWeak)
stub
.
createTime
=
taosGetTimestampMs
();
stub
.
rpcMsg
=
*
pMsg
;
uint64_t
seqNum
=
syncRespMgrAdd
(
pSyncNode
->
pSyncRespMgr
,
&
stub
);
sDebug
(
"vgId:%d
,
sync event propose, type:%s seq:%"
PRIu64
" handle:%p"
,
pSyncNode
->
vgId
,
TMSG_INFO
(
pMsg
->
msgType
),
sDebug
(
"vgId:%d sync event propose, type:%s seq:%"
PRIu64
" handle:%p"
,
pSyncNode
->
vgId
,
TMSG_INFO
(
pMsg
->
msgType
),
seqNum
,
pMsg
->
info
.
handle
);
SyncClientRequest
*
pSyncMsg
=
syncClientRequestBuild2
(
pMsg
,
seqNum
,
isWeak
,
pSyncNode
->
vgId
);
...
...
@@ -1244,7 +1244,13 @@ void syncNodeUpdateConfig(SSyncNode* pSyncNode, SSyncCfg* newConfig, SyncIndex l
SRaftId
oldReplicasId
[
TSDB_MAX_REPLICA
];
memcpy
(
oldReplicasId
,
pSyncNode
->
replicasId
,
sizeof
(
oldReplicasId
));
SSyncSnapshotSender
*
oldSenders
[
TSDB_MAX_REPLICA
];
memcpy
(
oldSenders
,
pSyncNode
->
senders
,
sizeof
(
oldSenders
));
for
(
int
i
=
0
;
i
<
TSDB_MAX_REPLICA
;
++
i
)
{
oldSenders
[
i
]
=
(
pSyncNode
->
senders
)[
i
];
sDebug
(
"vgId:%d sync event save senders %d, %p"
,
pSyncNode
->
vgId
,
i
,
oldSenders
[
i
]);
if
(
gRaftDetailLog
)
{
;
}
}
// init internal
pSyncNode
->
myNodeInfo
=
pSyncNode
->
pRaftCfg
->
cfg
.
nodeInfo
[
pSyncNode
->
pRaftCfg
->
cfg
.
myIndex
];
...
...
@@ -1286,22 +1292,24 @@ void syncNodeUpdateConfig(SSyncNode* pSyncNode, SSyncCfg* newConfig, SyncIndex l
// reset new
for
(
int
i
=
0
;
i
<
pSyncNode
->
replicaNum
;
++
i
)
{
// reset sender
bool
reset
=
false
;
for
(
int
j
=
0
;
j
<
TSDB_MAX_REPLICA
;
++
j
)
{
if
(
syncUtilSameId
(
&
(
pSyncNode
->
replicasId
)[
i
],
&
oldReplicasId
[
j
]))
{
char
host
[
128
];
uint16_t
port
;
syncUtilU642Addr
((
pSyncNode
->
replicasId
)[
i
].
addr
,
host
,
sizeof
(
host
),
&
port
);
sDebug
(
"vgId:%d sync event reset sender for %lu,
%s:%d"
,
pSyncNode
->
vgId
,
(
pSyncNode
->
replicasId
)[
i
].
addr
,
host
,
port
);
sDebug
(
"vgId:%d sync event reset sender for %lu,
newIndex:%d, %s:%d, %p"
,
pSyncNode
->
vgId
,
(
pSyncNode
->
replicasId
)[
i
].
addr
,
i
,
host
,
port
,
oldSenders
[
j
]
);
(
pSyncNode
->
senders
)[
i
]
=
oldSenders
[
j
];
oldSenders
[
j
]
=
NULL
;
}
// reset replicaIndex
int32_t
oldreplicaIndex
=
(
pSyncNode
->
senders
)[
i
]
->
replicaIndex
;
(
pSyncNode
->
senders
)[
i
]
->
replicaIndex
=
i
;
reset
=
true
;
sDebug
(
"vgId:%d sync event udpate replicaIndex from %d to %d"
,
pSyncNode
->
vgId
,
oldreplicaIndex
,
i
);
// reset replicaIndex
int32_t
oldreplicaIndex
=
(
pSyncNode
->
senders
)[
i
]
->
replicaIndex
;
(
pSyncNode
->
senders
)[
i
]
->
replicaIndex
=
i
;
sDebug
(
"vgId:%d sync event udpate replicaIndex from %d to %d, %s:%d, %p, reset:%d"
,
pSyncNode
->
vgId
,
oldreplicaIndex
,
i
,
host
,
port
,
(
pSyncNode
->
senders
)[
i
],
reset
);
}
}
}
...
...
@@ -1309,7 +1317,7 @@ void syncNodeUpdateConfig(SSyncNode* pSyncNode, SSyncCfg* newConfig, SyncIndex l
for
(
int
i
=
0
;
i
<
TSDB_MAX_REPLICA
;
++
i
)
{
if
((
pSyncNode
->
senders
)[
i
]
==
NULL
)
{
(
pSyncNode
->
senders
)[
i
]
=
snapshotSenderCreate
(
pSyncNode
,
i
);
sDebug
(
"vgId:%d sync event create new sender
replicaIndex:%d"
,
pSyncNode
->
vgId
,
i
);
sDebug
(
"vgId:%d sync event create new sender
%p replicaIndex:%d"
,
pSyncNode
->
vgId
,
(
pSyncNode
->
senders
)[
i
]
,
i
);
}
}
...
...
@@ -1317,8 +1325,8 @@ void syncNodeUpdateConfig(SSyncNode* pSyncNode, SSyncCfg* newConfig, SyncIndex l
for
(
int
i
=
0
;
i
<
TSDB_MAX_REPLICA
;
++
i
)
{
if
(
oldSenders
[
i
]
!=
NULL
)
{
snapshotSenderDestroy
(
oldSenders
[
i
]);
sDebug
(
"vgId:%d sync event delete old sender %p replicaIndex:%d"
,
pSyncNode
->
vgId
,
oldSenders
[
i
],
i
);
oldSenders
[
i
]
=
NULL
;
sDebug
(
"vgId:%d sync event delete old sender replicaIndex:%d"
,
pSyncNode
->
vgId
,
i
);
}
}
...
...
@@ -1378,8 +1386,8 @@ void syncNodeUpdateTerm(SSyncNode* pSyncNode, SyncTerm term) {
}
void
syncNodeBecomeFollower
(
SSyncNode
*
pSyncNode
,
const
char
*
debugStr
)
{
sDebug
(
"vgId:%d sync event become follower, isStandBy:%d,
%s"
,
pSyncNode
->
vgId
,
pSyncNode
->
pRaftCfg
->
isStandBy
,
debugStr
);
sDebug
(
"vgId:%d sync event become follower, isStandBy:%d,
replicaNum:%d, %s"
,
pSyncNode
->
vgId
,
pSyncNode
->
pRaftCfg
->
isStandBy
,
pSyncNode
->
replicaNum
,
debugStr
);
// maybe clear leader cache
if
(
pSyncNode
->
state
==
TAOS_SYNC_STATE_LEADER
)
{
...
...
@@ -1413,8 +1421,8 @@ void syncNodeBecomeFollower(SSyncNode* pSyncNode, const char* debugStr) {
// /\ UNCHANGED <<messages, currentTerm, votedFor, candidateVars, logVars>>
//
void
syncNodeBecomeLeader
(
SSyncNode
*
pSyncNode
,
const
char
*
debugStr
)
{
sDebug
(
"vgId:%d sync event become leader, isStandBy:%d,
%s"
,
pSyncNode
->
vgId
,
pSyncNode
->
pRaftCfg
->
isStandBy
,
debugStr
);
sDebug
(
"vgId:%d sync event become leader, isStandBy:%d,
replicaNum:%d %s"
,
pSyncNode
->
vgId
,
pSyncNode
->
pRaftCfg
->
isStandBy
,
pSyncNode
->
replicaNum
,
debugStr
);
// state change
pSyncNode
->
state
=
TAOS_SYNC_STATE_LEADER
;
...
...
@@ -2028,14 +2036,18 @@ static int32_t syncNodeConfigChange(SSyncNode* ths, SRpcMsg* pRpcMsg, SSyncRaftE
// change isStandBy to normal
if
(
!
isDrop
)
{
char
tmpbuf
[
128
];
snprintf
(
tmpbuf
,
sizeof
(
tmpbuf
),
"config change from %d to %d"
,
oldSyncCfg
.
replicaNum
,
newSyncCfg
.
replicaNum
);
if
(
ths
->
state
==
TAOS_SYNC_STATE_LEADER
)
{
syncNodeBecomeLeader
(
ths
,
"config change"
);
syncNodeBecomeLeader
(
ths
,
tmpbuf
);
}
else
{
syncNodeBecomeFollower
(
ths
,
"config change"
);
syncNodeBecomeFollower
(
ths
,
tmpbuf
);
}
}
}
else
{
syncNodeBecomeFollower
(
ths
,
"config change2"
);
char
tmpbuf
[
128
];
snprintf
(
tmpbuf
,
sizeof
(
tmpbuf
),
"config change2 from %d to %d"
,
oldSyncCfg
.
replicaNum
,
newSyncCfg
.
replicaNum
);
syncNodeBecomeFollower
(
ths
,
tmpbuf
);
}
if
(
gRaftDetailLog
)
{
...
...
source/libs/sync/src/syncSnapshot.c
浏览文件 @
5268b523
...
...
@@ -588,6 +588,8 @@ int32_t syncNodeOnSnapshotSendCb(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) {
// maybe update lastconfig
if
(
pMsg
->
lastConfigIndex
>=
SYNC_INDEX_BEGIN
)
{
int32_t
oldReplicaNum
=
pSyncNode
->
replicaNum
;
// update new config myIndex
bool
IamInNew
=
false
;
SSyncCfg
newSyncCfg
=
pMsg
->
lastConfig
;
...
...
@@ -614,10 +616,12 @@ int32_t syncNodeOnSnapshotSendCb(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) {
// change isStandBy to normal
if
(
!
isDrop
)
{
char
tmpbuf
[
128
];
snprintf
(
tmpbuf
,
sizeof
(
tmpbuf
),
"config change3 from %d to %d"
,
oldReplicaNum
,
newSyncCfg
.
replicaNum
);
if
(
pSyncNode
->
state
==
TAOS_SYNC_STATE_LEADER
)
{
syncNodeBecomeLeader
(
pSyncNode
,
"config change3"
);
syncNodeBecomeLeader
(
pSyncNode
,
tmpbuf
);
}
else
{
syncNodeBecomeFollower
(
pSyncNode
,
"config change3"
);
syncNodeBecomeFollower
(
pSyncNode
,
tmpbuf
);
}
}
}
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录