Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
9e8404f6
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看板
提交
9e8404f6
编写于
10月 25, 2022
作者:
M
Minghao Li
提交者:
Benguang Zhao
11月 15, 2022
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
refactor(sync): adjust timer
上级
bf634a84
变更
3
隐藏空白更改
内联
并排
Showing
3 changed file
with
29 addition
and
60 deletion
+29
-60
source/libs/sync/inc/syncInt.h
source/libs/sync/inc/syncInt.h
+0
-1
source/libs/sync/src/syncMain.c
source/libs/sync/src/syncMain.c
+28
-56
source/libs/sync/src/syncTimeout.c
source/libs/sync/src/syncTimeout.c
+1
-3
未找到文件。
source/libs/sync/inc/syncInt.h
浏览文件 @
9e8404f6
...
...
@@ -248,7 +248,6 @@ typedef struct SSyncNode {
tmr_h
pElectTimer
;
int32_t
electTimerMS
;
uint64_t
electTimerLogicClock
;
uint64_t
electTimerLogicClockUser
;
TAOS_TMR_CALLBACK
FpElectTimerCB
;
// Timer Fp
uint64_t
electTimerCounter
;
...
...
source/libs/sync/src/syncMain.c
浏览文件 @
9e8404f6
...
...
@@ -1631,7 +1631,6 @@ SSyncNode* syncNodeOpen(SSyncInfo* pOldSyncInfo) {
pSyncNode
->
pElectTimer
=
NULL
;
pSyncNode
->
electTimerMS
=
syncUtilElectRandomMS
(
pSyncNode
->
electBaseLine
,
2
*
pSyncNode
->
electBaseLine
);
atomic_store_64
(
&
pSyncNode
->
electTimerLogicClock
,
0
);
atomic_store_64
(
&
pSyncNode
->
electTimerLogicClockUser
,
0
);
pSyncNode
->
FpElectTimerCB
=
syncNodeEqElectTimer
;
pSyncNode
->
electTimerCounter
=
0
;
...
...
@@ -1937,15 +1936,6 @@ int32_t syncNodeStartElectTimer(SSyncNode* pSyncNode, int32_t ms) {
pSyncNode
->
electTimerMS
=
ms
;
taosTmrReset
(
pSyncNode
->
FpElectTimerCB
,
pSyncNode
->
electTimerMS
,
pSyncNode
,
gSyncEnv
->
pTimerManager
,
&
pSyncNode
->
pElectTimer
);
atomic_store_64
(
&
pSyncNode
->
electTimerLogicClock
,
pSyncNode
->
electTimerLogicClockUser
);
/*
do {
char logBuf[128];
snprintf(logBuf, sizeof(logBuf), "elect timer reset, ms:%d", ms);
syncNodeEventLog(pSyncNode, logBuf);
} while (0);
*/
}
else
{
sError
(
"vgId:%d, start elect timer error, sync env is stop"
,
pSyncNode
->
vgId
);
...
...
@@ -1955,11 +1945,10 @@ int32_t syncNodeStartElectTimer(SSyncNode* pSyncNode, int32_t ms) {
int32_t
syncNodeStopElectTimer
(
SSyncNode
*
pSyncNode
)
{
int32_t
ret
=
0
;
atomic_add_fetch_64
(
&
pSyncNode
->
electTimerLogicClock
User
,
1
);
atomic_add_fetch_64
(
&
pSyncNode
->
electTimerLogicClock
,
1
);
taosTmrStop
(
pSyncNode
->
pElectTimer
);
pSyncNode
->
pElectTimer
=
NULL
;
// sTrace("vgId:%d, sync %s stop elect timer", pSyncNode->vgId, syncUtilState2String(pSyncNode->state));
return
ret
;
}
...
...
@@ -2185,8 +2174,6 @@ cJSON* syncNode2Json(const SSyncNode* pSyncNode) {
cJSON_AddNumberToObject
(
pRoot
,
"electTimerMS"
,
pSyncNode
->
electTimerMS
);
snprintf
(
u64buf
,
sizeof
(
u64buf
),
"%"
PRIu64
,
pSyncNode
->
electTimerLogicClock
);
cJSON_AddStringToObject
(
pRoot
,
"electTimerLogicClock"
,
u64buf
);
snprintf
(
u64buf
,
sizeof
(
u64buf
),
"%"
PRIu64
,
pSyncNode
->
electTimerLogicClockUser
);
cJSON_AddStringToObject
(
pRoot
,
"electTimerLogicClockUser"
,
u64buf
);
snprintf
(
u64buf
,
sizeof
(
u64buf
),
"%p"
,
pSyncNode
->
FpElectTimerCB
);
cJSON_AddStringToObject
(
pRoot
,
"FpElectTimerCB"
,
u64buf
);
snprintf
(
u64buf
,
sizeof
(
u64buf
),
"%"
PRIu64
,
pSyncNode
->
electTimerCounter
);
...
...
@@ -2288,7 +2275,7 @@ inline void syncNodeEventLog(const SSyncNode* pSyncNode, char* str) {
snapshot
.
lastApplyTerm
,
pSyncNode
->
pRaftCfg
->
isStandBy
,
pSyncNode
->
pRaftCfg
->
snapshotStrategy
,
pSyncNode
->
pRaftCfg
->
batchSize
,
pSyncNode
->
replicaNum
,
pSyncNode
->
pRaftCfg
->
lastConfigIndex
,
pSyncNode
->
changing
,
pSyncNode
->
restoreFinish
,
syncNodeDynamicQuorum
(
pSyncNode
),
pSyncNode
->
electTimerLogicClock
User
,
pSyncNode
->
heartbeatTimerLogicClockUser
,
peerStateStr
,
printStr
);
pSyncNode
->
electTimerLogicClock
,
pSyncNode
->
heartbeatTimerLogicClockUser
,
peerStateStr
,
printStr
);
}
else
{
snprintf
(
logBuf
,
sizeof
(
logBuf
),
"%s"
,
str
);
}
...
...
@@ -2312,7 +2299,7 @@ inline void syncNodeEventLog(const SSyncNode* pSyncNode, char* str) {
snapshot
.
lastApplyTerm
,
pSyncNode
->
pRaftCfg
->
isStandBy
,
pSyncNode
->
pRaftCfg
->
snapshotStrategy
,
pSyncNode
->
pRaftCfg
->
batchSize
,
pSyncNode
->
replicaNum
,
pSyncNode
->
pRaftCfg
->
lastConfigIndex
,
pSyncNode
->
changing
,
pSyncNode
->
restoreFinish
,
syncNodeDynamicQuorum
(
pSyncNode
),
pSyncNode
->
electTimerLogicClock
User
,
pSyncNode
->
heartbeatTimerLogicClockUser
,
peerStateStr
,
printStr
);
pSyncNode
->
electTimerLogicClock
,
pSyncNode
->
heartbeatTimerLogicClockUser
,
peerStateStr
,
printStr
);
}
else
{
snprintf
(
s
,
len
,
"%s"
,
str
);
}
...
...
@@ -2362,7 +2349,7 @@ inline void syncNodeErrorLog(const SSyncNode* pSyncNode, char* str) {
snapshot
.
lastApplyTerm
,
pSyncNode
->
pRaftCfg
->
isStandBy
,
pSyncNode
->
pRaftCfg
->
snapshotStrategy
,
pSyncNode
->
pRaftCfg
->
batchSize
,
pSyncNode
->
replicaNum
,
pSyncNode
->
pRaftCfg
->
lastConfigIndex
,
pSyncNode
->
changing
,
pSyncNode
->
restoreFinish
,
syncNodeDynamicQuorum
(
pSyncNode
),
pSyncNode
->
electTimerLogicClock
User
,
pSyncNode
->
heartbeatTimerLogicClockUser
,
printStr
);
pSyncNode
->
electTimerLogicClock
,
pSyncNode
->
heartbeatTimerLogicClockUser
,
printStr
);
}
else
{
snprintf
(
logBuf
,
sizeof
(
logBuf
),
"%s"
,
str
);
}
...
...
@@ -2384,7 +2371,7 @@ inline void syncNodeErrorLog(const SSyncNode* pSyncNode, char* str) {
snapshot
.
lastApplyTerm
,
pSyncNode
->
pRaftCfg
->
isStandBy
,
pSyncNode
->
pRaftCfg
->
snapshotStrategy
,
pSyncNode
->
pRaftCfg
->
batchSize
,
pSyncNode
->
replicaNum
,
pSyncNode
->
pRaftCfg
->
lastConfigIndex
,
pSyncNode
->
changing
,
pSyncNode
->
restoreFinish
,
syncNodeDynamicQuorum
(
pSyncNode
),
pSyncNode
->
electTimerLogicClock
User
,
pSyncNode
->
heartbeatTimerLogicClockUser
,
printStr
);
pSyncNode
->
electTimerLogicClock
,
pSyncNode
->
heartbeatTimerLogicClockUser
,
printStr
);
}
else
{
snprintf
(
s
,
len
,
"%s"
,
str
);
}
...
...
@@ -3226,36 +3213,31 @@ static void syncNodeEqPingTimer(void* param, void* tmrId) {
static
void
syncNodeEqElectTimer
(
void
*
param
,
void
*
tmrId
)
{
SSyncNode
*
pSyncNode
=
(
SSyncNode
*
)
param
;
if
(
atomic_load_64
(
&
pSyncNode
->
electTimerLogicClockUser
)
<=
atomic_load_64
(
&
pSyncNode
->
electTimerLogicClock
))
{
SyncTimeout
*
pSyncMsg
=
syncTimeoutBuild2
(
SYNC_TIMEOUT_ELECTION
,
atomic_load_64
(
&
pSyncNode
->
electTimerLogicClock
),
pSyncNode
->
electTimerMS
,
pSyncNode
->
vgId
,
pSyncNode
);
SRpcMsg
rpcMsg
;
syncTimeout2RpcMsg
(
pSyncMsg
,
&
rpcMsg
);
syncRpcMsgLog2
((
char
*
)
"==syncNodeEqElectTimer=="
,
&
rpcMsg
);
if
(
pSyncNode
->
FpEqMsg
!=
NULL
)
{
int32_t
code
=
pSyncNode
->
FpEqMsg
(
pSyncNode
->
msgcb
,
&
rpcMsg
);
if
(
code
!=
0
)
{
sError
(
"vgId:%d, sync enqueue elect msg error, code:%d"
,
pSyncNode
->
vgId
,
code
);
rpcFreeCont
(
rpcMsg
.
pCont
);
syncTimeoutDestroy
(
pSyncMsg
);
return
;
}
}
else
{
sTrace
(
"syncNodeEqElectTimer FpEqMsg is NULL"
);
}
syncTimeoutDestroy
(
pSyncMsg
);
// reset timer ms
if
(
syncEnvIsStart
()
&&
pSyncNode
->
electBaseLine
>
0
)
{
pSyncNode
->
electTimerMS
=
syncUtilElectRandomMS
(
pSyncNode
->
electBaseLine
,
2
*
pSyncNode
->
electBaseLine
);
taosTmrReset
(
syncNodeEqElectTimer
,
pSyncNode
->
electTimerMS
,
pSyncNode
,
gSyncEnv
->
pTimerManager
,
&
pSyncNode
->
pElectTimer
);
}
else
{
sError
(
"sync env is stop, syncNodeEqElectTimer"
);
SyncTimeout
*
pSyncMsg
=
syncTimeoutBuild2
(
SYNC_TIMEOUT_ELECTION
,
atomic_load_64
(
&
pSyncNode
->
electTimerLogicClock
),
pSyncNode
->
electTimerMS
,
pSyncNode
->
vgId
,
pSyncNode
);
SRpcMsg
rpcMsg
;
syncTimeout2RpcMsg
(
pSyncMsg
,
&
rpcMsg
);
if
(
pSyncNode
->
FpEqMsg
!=
NULL
)
{
int32_t
code
=
pSyncNode
->
FpEqMsg
(
pSyncNode
->
msgcb
,
&
rpcMsg
);
if
(
code
!=
0
)
{
sError
(
"vgId:%d, sync enqueue elect msg error, code:%d"
,
pSyncNode
->
vgId
,
code
);
rpcFreeCont
(
rpcMsg
.
pCont
);
syncTimeoutDestroy
(
pSyncMsg
);
return
;
}
}
else
{
sTrace
(
"==syncNodeEqElectTimer== electTimerLogicClock:%"
PRIu64
", electTimerLogicClockUser:%"
PRIu64
,
pSyncNode
->
electTimerLogicClock
,
pSyncNode
->
electTimerLogicClockUser
);
sTrace
(
"syncNodeEqElectTimer FpEqMsg is NULL"
);
}
syncTimeoutDestroy
(
pSyncMsg
);
// reset timer ms
if
(
syncEnvIsStart
()
&&
pSyncNode
->
electBaseLine
>
0
)
{
pSyncNode
->
electTimerMS
=
syncUtilElectRandomMS
(
pSyncNode
->
electBaseLine
,
2
*
pSyncNode
->
electBaseLine
);
taosTmrReset
(
syncNodeEqElectTimer
,
pSyncNode
->
electTimerMS
,
pSyncNode
,
gSyncEnv
->
pTimerManager
,
&
pSyncNode
->
pElectTimer
);
}
else
{
sError
(
"sync env is stop, syncNodeEqElectTimer"
);
}
}
...
...
@@ -3480,16 +3462,6 @@ static int32_t syncNodeAppendNoopOld(SSyncNode* ths) {
// on message ----
int32_t
syncNodeOnPingCb
(
SSyncNode
*
ths
,
SyncPing
*
pMsg
)
{
// log state
char
logBuf
[
1024
]
=
{
0
};
snprintf
(
logBuf
,
sizeof
(
logBuf
),
"==syncNodeOnPingCb== vgId:%d, state: %d, %s, term:%"
PRIu64
" electTimerLogicClock:%"
PRIu64
", "
"electTimerLogicClockUser:%"
PRIu64
", electTimerMS:%d"
,
ths
->
vgId
,
ths
->
state
,
syncUtilState2String
(
ths
->
state
),
ths
->
pRaftStore
->
currentTerm
,
ths
->
electTimerLogicClock
,
ths
->
electTimerLogicClockUser
,
ths
->
electTimerMS
);
int32_t
ret
=
0
;
syncPingLog2
(
logBuf
,
pMsg
);
SyncPingReply
*
pMsgReply
=
syncPingReplyBuild3
(
&
ths
->
myRaftId
,
&
pMsg
->
srcId
,
ths
->
vgId
);
SRpcMsg
rpcMsg
;
syncPingReply2RpcMsg
(
pMsgReply
,
&
rpcMsg
);
...
...
@@ -3503,7 +3475,7 @@ int32_t syncNodeOnPingCb(SSyncNode* ths, SyncPing* pMsg) {
syncNodeSendMsgById
(
&
pMsgReply
->
destId
,
ths
,
&
rpcMsg
);
return
ret
;
return
0
;
}
int32_t
syncNodeOnPingReplyCb
(
SSyncNode
*
ths
,
SyncPingReply
*
pMsg
)
{
...
...
source/libs/sync/src/syncTimeout.c
浏览文件 @
9e8404f6
...
...
@@ -113,10 +113,8 @@ int32_t syncNodeOnTimer(SSyncNode* ths, SyncTimeout* pMsg) {
}
}
else
if
(
pMsg
->
timeoutType
==
SYNC_TIMEOUT_ELECTION
)
{
if
(
atomic_load_64
(
&
ths
->
electTimerLogicClock
User
)
<=
pMsg
->
logicClock
)
{
if
(
atomic_load_64
(
&
ths
->
electTimerLogicClock
)
<=
pMsg
->
logicClock
)
{
++
(
ths
->
electTimerCounter
);
sTrace
(
"vgId:%d, sync timer, type:election count:%"
PRIu64
", lc-user:%"
PRIu64
,
ths
->
vgId
,
ths
->
electTimerCounter
,
ths
->
electTimerLogicClockUser
);
syncNodeElect
(
ths
);
}
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录