Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
e060de70
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看板
提交
e060de70
编写于
11月 07, 2022
作者:
S
Shengliang Guan
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
refact: remove config change codes
上级
1ddbeb43
变更
7
隐藏空白更改
内联
并排
Showing
7 changed file
with
32 addition
and
474 deletion
+32
-474
source/libs/sync/inc/syncTools.h
source/libs/sync/inc/syncTools.h
+0
-39
source/libs/sync/src/syncMain.c
source/libs/sync/src/syncMain.c
+0
-136
source/libs/sync/src/syncMessage.c
source/libs/sync/src/syncMessage.c
+0
-159
source/libs/sync/src/syncUtil.c
source/libs/sync/src/syncUtil.c
+3
-6
source/libs/sync/test/CMakeLists.txt
source/libs/sync/test/CMakeLists.txt
+0
-1
source/libs/sync/test/syncLeaderTransferTest.cpp
source/libs/sync/test/syncLeaderTransferTest.cpp
+29
-0
source/libs/sync/test/syncReconfigFinishTest.cpp
source/libs/sync/test/syncReconfigFinishTest.cpp
+0
-133
未找到文件。
source/libs/sync/inc/syncTools.h
浏览文件 @
e060de70
...
...
@@ -688,45 +688,6 @@ SyncLeaderTransfer* syncLeaderTransferFromRpcMsg2(const SRpcMsg* pRpcMsg);
cJSON
*
syncLeaderTransfer2Json
(
const
SyncLeaderTransfer
*
pMsg
);
char
*
syncLeaderTransfer2Str
(
const
SyncLeaderTransfer
*
pMsg
);
// for debug ----------------------
void
syncLeaderTransferPrint
(
const
SyncLeaderTransfer
*
pMsg
);
void
syncLeaderTransferPrint2
(
char
*
s
,
const
SyncLeaderTransfer
*
pMsg
);
void
syncLeaderTransferLog
(
const
SyncLeaderTransfer
*
pMsg
);
void
syncLeaderTransferLog2
(
char
*
s
,
const
SyncLeaderTransfer
*
pMsg
);
// ---------------------------------------------
typedef
struct
SyncReconfigFinish
{
uint32_t
bytes
;
int32_t
vgId
;
uint32_t
msgType
;
SSyncCfg
oldCfg
;
SSyncCfg
newCfg
;
SyncIndex
newCfgIndex
;
SyncTerm
newCfgTerm
;
uint64_t
newCfgSeqNum
;
}
SyncReconfigFinish
;
SyncReconfigFinish
*
syncReconfigFinishBuild
(
int32_t
vgId
);
void
syncReconfigFinishDestroy
(
SyncReconfigFinish
*
pMsg
);
void
syncReconfigFinishSerialize
(
const
SyncReconfigFinish
*
pMsg
,
char
*
buf
,
uint32_t
bufLen
);
void
syncReconfigFinishDeserialize
(
const
char
*
buf
,
uint32_t
len
,
SyncReconfigFinish
*
pMsg
);
char
*
syncReconfigFinishSerialize2
(
const
SyncReconfigFinish
*
pMsg
,
uint32_t
*
len
);
SyncReconfigFinish
*
syncReconfigFinishDeserialize2
(
const
char
*
buf
,
uint32_t
len
);
void
syncReconfigFinish2RpcMsg
(
const
SyncReconfigFinish
*
pMsg
,
SRpcMsg
*
pRpcMsg
);
void
syncReconfigFinishFromRpcMsg
(
const
SRpcMsg
*
pRpcMsg
,
SyncReconfigFinish
*
pMsg
);
SyncReconfigFinish
*
syncReconfigFinishFromRpcMsg2
(
const
SRpcMsg
*
pRpcMsg
);
cJSON
*
syncReconfigFinish2Json
(
const
SyncReconfigFinish
*
pMsg
);
char
*
syncReconfigFinish2Str
(
const
SyncReconfigFinish
*
pMsg
);
// for debug ----------------------
void
syncReconfigFinishPrint
(
const
SyncReconfigFinish
*
pMsg
);
void
syncReconfigFinishPrint2
(
char
*
s
,
const
SyncReconfigFinish
*
pMsg
);
void
syncReconfigFinishLog
(
const
SyncReconfigFinish
*
pMsg
);
void
syncReconfigFinishLog2
(
char
*
s
,
const
SyncReconfigFinish
*
pMsg
);
// ---------------------------------------------
typedef
enum
{
SYNC_LOCAL_CMD_STEP_DOWN
=
100
,
SYNC_LOCAL_CMD_FOLLOWER_CMT
,
...
...
source/libs/sync/src/syncMain.c
浏览文件 @
e060de70
...
...
@@ -754,45 +754,11 @@ int32_t syncPropose(int64_t rid, SRpcMsg* pMsg, bool isWeak) {
return
ret
;
}
static
bool
syncNodeBatchOK
(
SRpcMsg
**
pMsgPArr
,
int32_t
arrSize
)
{
for
(
int32_t
i
=
0
;
i
<
arrSize
;
++
i
)
{
if
(
pMsgPArr
[
i
]
->
msgType
==
TDMT_SYNC_CONFIG_CHANGE
)
{
return
false
;
}
if
(
pMsgPArr
[
i
]
->
msgType
==
TDMT_SYNC_CONFIG_CHANGE_FINISH
)
{
return
false
;
}
}
return
true
;
}
int32_t
syncNodePropose
(
SSyncNode
*
pSyncNode
,
SRpcMsg
*
pMsg
,
bool
isWeak
)
{
int32_t
ret
=
0
;
sNTrace
(
pSyncNode
,
"propose message, type:%s"
,
TMSG_INFO
(
pMsg
->
msgType
));
if
(
pSyncNode
->
state
==
TAOS_SYNC_STATE_LEADER
)
{
if
(
pSyncNode
->
changing
&&
pMsg
->
msgType
!=
TDMT_SYNC_CONFIG_CHANGE_FINISH
)
{
ret
=
-
1
;
terrno
=
TSDB_CODE_SYN_PROPOSE_NOT_READY
;
sError
(
"vgId:%d, failed to sync propose since not ready, type:%s"
,
pSyncNode
->
vgId
,
TMSG_INFO
(
pMsg
->
msgType
));
goto
_END
;
}
// config change
if
(
pMsg
->
msgType
==
TDMT_SYNC_CONFIG_CHANGE
)
{
if
(
!
syncNodeCanChange
(
pSyncNode
))
{
ret
=
-
1
;
terrno
=
TSDB_CODE_SYN_RECONFIG_NOT_READY
;
sError
(
"vgId:%d, failed to sync reconfig since not ready, type:%s"
,
pSyncNode
->
vgId
,
TMSG_INFO
(
pMsg
->
msgType
));
goto
_END
;
}
ASSERT
(
!
pSyncNode
->
changing
);
pSyncNode
->
changing
=
true
;
}
// not restored, vnode enable
if
(
!
pSyncNode
->
restoreFinish
&&
pSyncNode
->
vgId
!=
1
)
{
ret
=
-
1
;
...
...
@@ -2800,86 +2766,6 @@ int32_t syncNodeUpdateNewConfigIndex(SSyncNode* ths, SSyncCfg* pNewCfg) {
return
-
1
;
}
static
int32_t
syncNodeConfigChangeFinish
(
SSyncNode
*
ths
,
SRpcMsg
*
pRpcMsg
,
SSyncRaftEntry
*
pEntry
)
{
SyncReconfigFinish
*
pFinish
=
syncReconfigFinishFromRpcMsg2
(
pRpcMsg
);
ASSERT
(
pFinish
);
if
(
ths
->
pFsm
->
FpReConfigCb
!=
NULL
)
{
SReConfigCbMeta
cbMeta
=
{
0
};
cbMeta
.
code
=
0
;
cbMeta
.
index
=
pEntry
->
index
;
cbMeta
.
term
=
pEntry
->
term
;
cbMeta
.
seqNum
=
pEntry
->
seqNum
;
cbMeta
.
lastConfigIndex
=
syncNodeGetSnapshotConfigIndex
(
ths
,
pEntry
->
index
);
cbMeta
.
state
=
ths
->
state
;
cbMeta
.
currentTerm
=
ths
->
pRaftStore
->
currentTerm
;
cbMeta
.
isWeak
=
pEntry
->
isWeak
;
cbMeta
.
flag
=
0
;
cbMeta
.
oldCfg
=
pFinish
->
oldCfg
;
cbMeta
.
newCfg
=
pFinish
->
newCfg
;
cbMeta
.
newCfgIndex
=
pFinish
->
newCfgIndex
;
cbMeta
.
newCfgTerm
=
pFinish
->
newCfgTerm
;
cbMeta
.
newCfgSeqNum
=
pFinish
->
newCfgSeqNum
;
ths
->
pFsm
->
FpReConfigCb
(
ths
->
pFsm
,
pRpcMsg
,
&
cbMeta
);
}
// clear changing
ths
->
changing
=
false
;
char
oldCfgStr
[
1024
]
=
{
0
};
char
newCfgStr
[
1024
]
=
{
0
};
syncCfg2SimpleStr
(
&
pFinish
->
oldCfg
,
oldCfgStr
,
sizeof
(
oldCfgStr
));
syncCfg2SimpleStr
(
&
pFinish
->
newCfg
,
oldCfgStr
,
sizeof
(
oldCfgStr
));
sNTrace
(
ths
,
"config change finish from %d to %d, index:%"
PRId64
", %s --> %s"
,
pFinish
->
oldCfg
.
replicaNum
,
pFinish
->
newCfg
.
replicaNum
,
pFinish
->
newCfgIndex
,
oldCfgStr
,
newCfgStr
);
syncReconfigFinishDestroy
(
pFinish
);
return
0
;
}
static
int32_t
syncNodeConfigChange
(
SSyncNode
*
ths
,
SRpcMsg
*
pRpcMsg
,
SSyncRaftEntry
*
pEntry
,
SyncReconfigFinish
*
pFinish
)
{
// set changing
ths
->
changing
=
true
;
// old config
SSyncCfg
oldSyncCfg
=
ths
->
pRaftCfg
->
cfg
;
// new config
SSyncCfg
newSyncCfg
;
int32_t
ret
=
syncCfgFromStr
(
pRpcMsg
->
pCont
,
&
newSyncCfg
);
ASSERT
(
ret
==
0
);
// update new config myIndex
syncNodeUpdateNewConfigIndex
(
ths
,
&
newSyncCfg
);
// do config change
syncNodeDoConfigChange
(
ths
,
&
newSyncCfg
,
pEntry
->
index
);
// set pFinish
pFinish
->
oldCfg
=
oldSyncCfg
;
pFinish
->
newCfg
=
newSyncCfg
;
pFinish
->
newCfgIndex
=
pEntry
->
index
;
pFinish
->
newCfgTerm
=
pEntry
->
term
;
pFinish
->
newCfgSeqNum
=
pEntry
->
seqNum
;
return
0
;
}
static
int32_t
syncNodeProposeConfigChangeFinish
(
SSyncNode
*
ths
,
SyncReconfigFinish
*
pFinish
)
{
SRpcMsg
rpcMsg
;
syncReconfigFinish2RpcMsg
(
pFinish
,
&
rpcMsg
);
int32_t
code
=
syncNodePropose
(
ths
,
&
rpcMsg
,
false
);
if
(
code
!=
0
)
{
sError
(
"syncNodeProposeConfigChangeFinish error"
);
ths
->
changing
=
false
;
}
return
0
;
}
bool
syncNodeIsOptimizedOneReplica
(
SSyncNode
*
ths
,
SRpcMsg
*
pMsg
)
{
return
(
ths
->
replicaNum
==
1
&&
syncUtilUserCommit
(
pMsg
->
msgType
)
&&
ths
->
vgId
!=
1
);
}
...
...
@@ -2961,28 +2847,6 @@ int32_t syncNodeDoCommit(SSyncNode* ths, SyncIndex beginIndex, SyncIndex endInde
}
}
// config change
if
(
pEntry
->
originalRpcType
==
TDMT_SYNC_CONFIG_CHANGE
)
{
SyncReconfigFinish
*
pFinish
=
syncReconfigFinishBuild
(
ths
->
vgId
);
ASSERT
(
pFinish
!=
NULL
);
code
=
syncNodeConfigChange
(
ths
,
&
rpcMsg
,
pEntry
,
pFinish
);
ASSERT
(
code
==
0
);
if
(
ths
->
state
==
TAOS_SYNC_STATE_LEADER
)
{
syncNodeProposeConfigChangeFinish
(
ths
,
pFinish
);
}
syncReconfigFinishDestroy
(
pFinish
);
}
// config change finish
if
(
pEntry
->
originalRpcType
==
TDMT_SYNC_CONFIG_CHANGE_FINISH
)
{
if
(
rpcMsg
.
pCont
!=
NULL
&&
rpcMsg
.
contLen
>
0
)
{
code
=
syncNodeConfigChangeFinish
(
ths
,
&
rpcMsg
,
pEntry
);
ASSERT
(
code
==
0
);
}
}
#if 0
// execute in pre-commit
// leader transfer
...
...
source/libs/sync/src/syncMessage.c
浏览文件 @
e060de70
...
...
@@ -3238,165 +3238,6 @@ char* syncLeaderTransfer2Str(const SyncLeaderTransfer* pMsg) {
return
serialized
;
}
// for debug ----------------------
void
syncLeaderTransferPrint
(
const
SyncLeaderTransfer
*
pMsg
)
{
char
*
serialized
=
syncLeaderTransfer2Str
(
pMsg
);
printf
(
"syncLeaderTransferPrint | len:%d | %s
\n
"
,
(
int32_t
)
strlen
(
serialized
),
serialized
);
fflush
(
NULL
);
taosMemoryFree
(
serialized
);
}
void
syncLeaderTransferPrint2
(
char
*
s
,
const
SyncLeaderTransfer
*
pMsg
)
{
char
*
serialized
=
syncLeaderTransfer2Str
(
pMsg
);
printf
(
"syncLeaderTransferPrint2 | len:%d | %s | %s
\n
"
,
(
int32_t
)
strlen
(
serialized
),
s
,
serialized
);
fflush
(
NULL
);
taosMemoryFree
(
serialized
);
}
void
syncLeaderTransferLog
(
const
SyncLeaderTransfer
*
pMsg
)
{
char
*
serialized
=
syncLeaderTransfer2Str
(
pMsg
);
sTrace
(
"syncLeaderTransferLog | len:%d | %s"
,
(
int32_t
)
strlen
(
serialized
),
serialized
);
taosMemoryFree
(
serialized
);
}
void
syncLeaderTransferLog2
(
char
*
s
,
const
SyncLeaderTransfer
*
pMsg
)
{
if
(
gRaftDetailLog
)
{
char
*
serialized
=
syncLeaderTransfer2Str
(
pMsg
);
sTrace
(
"syncLeaderTransferLog2 | len:%d | %s | %s"
,
(
int32_t
)
strlen
(
serialized
),
s
,
serialized
);
taosMemoryFree
(
serialized
);
}
}
// ---------------------------------------------
SyncReconfigFinish
*
syncReconfigFinishBuild
(
int32_t
vgId
)
{
uint32_t
bytes
=
sizeof
(
SyncReconfigFinish
);
SyncReconfigFinish
*
pMsg
=
taosMemoryMalloc
(
bytes
);
memset
(
pMsg
,
0
,
bytes
);
pMsg
->
bytes
=
bytes
;
pMsg
->
vgId
=
vgId
;
pMsg
->
msgType
=
TDMT_SYNC_CONFIG_CHANGE_FINISH
;
return
pMsg
;
}
void
syncReconfigFinishDestroy
(
SyncReconfigFinish
*
pMsg
)
{
if
(
pMsg
!=
NULL
)
{
taosMemoryFree
(
pMsg
);
}
}
void
syncReconfigFinishSerialize
(
const
SyncReconfigFinish
*
pMsg
,
char
*
buf
,
uint32_t
bufLen
)
{
ASSERT
(
pMsg
->
bytes
<=
bufLen
);
memcpy
(
buf
,
pMsg
,
pMsg
->
bytes
);
}
void
syncReconfigFinishDeserialize
(
const
char
*
buf
,
uint32_t
len
,
SyncReconfigFinish
*
pMsg
)
{
memcpy
(
pMsg
,
buf
,
len
);
ASSERT
(
len
==
pMsg
->
bytes
);
}
char
*
syncReconfigFinishSerialize2
(
const
SyncReconfigFinish
*
pMsg
,
uint32_t
*
len
)
{
char
*
buf
=
taosMemoryMalloc
(
pMsg
->
bytes
);
ASSERT
(
buf
!=
NULL
);
syncReconfigFinishSerialize
(
pMsg
,
buf
,
pMsg
->
bytes
);
if
(
len
!=
NULL
)
{
*
len
=
pMsg
->
bytes
;
}
return
buf
;
}
SyncReconfigFinish
*
syncReconfigFinishDeserialize2
(
const
char
*
buf
,
uint32_t
len
)
{
uint32_t
bytes
=
*
((
uint32_t
*
)
buf
);
SyncReconfigFinish
*
pMsg
=
taosMemoryMalloc
(
bytes
);
ASSERT
(
pMsg
!=
NULL
);
syncReconfigFinishDeserialize
(
buf
,
len
,
pMsg
);
ASSERT
(
len
==
pMsg
->
bytes
);
return
pMsg
;
}
void
syncReconfigFinish2RpcMsg
(
const
SyncReconfigFinish
*
pMsg
,
SRpcMsg
*
pRpcMsg
)
{
memset
(
pRpcMsg
,
0
,
sizeof
(
*
pRpcMsg
));
pRpcMsg
->
msgType
=
pMsg
->
msgType
;
pRpcMsg
->
contLen
=
pMsg
->
bytes
;
pRpcMsg
->
pCont
=
rpcMallocCont
(
pRpcMsg
->
contLen
);
syncReconfigFinishSerialize
(
pMsg
,
pRpcMsg
->
pCont
,
pRpcMsg
->
contLen
);
}
void
syncReconfigFinishFromRpcMsg
(
const
SRpcMsg
*
pRpcMsg
,
SyncReconfigFinish
*
pMsg
)
{
syncReconfigFinishDeserialize
(
pRpcMsg
->
pCont
,
pRpcMsg
->
contLen
,
pMsg
);
}
SyncReconfigFinish
*
syncReconfigFinishFromRpcMsg2
(
const
SRpcMsg
*
pRpcMsg
)
{
SyncReconfigFinish
*
pMsg
=
syncReconfigFinishDeserialize2
(
pRpcMsg
->
pCont
,
(
uint32_t
)(
pRpcMsg
->
contLen
));
ASSERT
(
pMsg
!=
NULL
);
return
pMsg
;
}
cJSON
*
syncReconfigFinish2Json
(
const
SyncReconfigFinish
*
pMsg
)
{
char
u64buf
[
128
];
cJSON
*
pRoot
=
cJSON_CreateObject
();
if
(
pMsg
!=
NULL
)
{
cJSON_AddNumberToObject
(
pRoot
,
"bytes"
,
pMsg
->
bytes
);
cJSON_AddNumberToObject
(
pRoot
,
"vgId"
,
pMsg
->
vgId
);
cJSON_AddNumberToObject
(
pRoot
,
"msgType"
,
pMsg
->
msgType
);
cJSON
*
pOldCfg
=
syncCfg2Json
((
SSyncCfg
*
)(
&
(
pMsg
->
oldCfg
)));
cJSON
*
pNewCfg
=
syncCfg2Json
((
SSyncCfg
*
)(
&
(
pMsg
->
newCfg
)));
cJSON_AddItemToObject
(
pRoot
,
"oldCfg"
,
pOldCfg
);
cJSON_AddItemToObject
(
pRoot
,
"newCfg"
,
pNewCfg
);
snprintf
(
u64buf
,
sizeof
(
u64buf
),
"%"
PRId64
,
pMsg
->
newCfgIndex
);
cJSON_AddStringToObject
(
pRoot
,
"newCfgIndex"
,
u64buf
);
snprintf
(
u64buf
,
sizeof
(
u64buf
),
"%"
PRIu64
,
pMsg
->
newCfgTerm
);
cJSON_AddStringToObject
(
pRoot
,
"newCfgTerm"
,
u64buf
);
snprintf
(
u64buf
,
sizeof
(
u64buf
),
"%"
PRIu64
,
pMsg
->
newCfgSeqNum
);
cJSON_AddStringToObject
(
pRoot
,
"newCfgSeqNum"
,
u64buf
);
}
cJSON
*
pJson
=
cJSON_CreateObject
();
cJSON_AddItemToObject
(
pJson
,
"SyncReconfigFinish"
,
pRoot
);
return
pJson
;
}
char
*
syncReconfigFinish2Str
(
const
SyncReconfigFinish
*
pMsg
)
{
cJSON
*
pJson
=
syncReconfigFinish2Json
(
pMsg
);
char
*
serialized
=
cJSON_Print
(
pJson
);
cJSON_Delete
(
pJson
);
return
serialized
;
}
// for debug ----------------------
void
syncReconfigFinishPrint
(
const
SyncReconfigFinish
*
pMsg
)
{
char
*
serialized
=
syncReconfigFinish2Str
(
pMsg
);
printf
(
"syncReconfigFinishPrint | len:%d | %s
\n
"
,
(
int32_t
)
strlen
(
serialized
),
serialized
);
fflush
(
NULL
);
taosMemoryFree
(
serialized
);
}
void
syncReconfigFinishPrint2
(
char
*
s
,
const
SyncReconfigFinish
*
pMsg
)
{
char
*
serialized
=
syncReconfigFinish2Str
(
pMsg
);
printf
(
"syncReconfigFinishPrint2 | len:%d | %s | %s
\n
"
,
(
int32_t
)
strlen
(
serialized
),
s
,
serialized
);
fflush
(
NULL
);
taosMemoryFree
(
serialized
);
}
void
syncReconfigFinishLog
(
const
SyncReconfigFinish
*
pMsg
)
{
char
*
serialized
=
syncReconfigFinish2Str
(
pMsg
);
sTrace
(
"syncReconfigFinishLog | len:%d | %s"
,
(
int32_t
)
strlen
(
serialized
),
serialized
);
taosMemoryFree
(
serialized
);
}
void
syncReconfigFinishLog2
(
char
*
s
,
const
SyncReconfigFinish
*
pMsg
)
{
if
(
gRaftDetailLog
)
{
char
*
serialized
=
syncReconfigFinish2Str
(
pMsg
);
sTrace
(
"syncReconfigFinishLog2 | len:%d | %s | %s"
,
(
int32_t
)
strlen
(
serialized
),
s
,
serialized
);
taosMemoryFree
(
serialized
);
}
}
// ---------------------------------------------
const
char
*
syncLocalCmdGetStr
(
int32_t
cmd
)
{
if
(
cmd
==
SYNC_LOCAL_CMD_STEP_DOWN
)
{
return
"step-down"
;
...
...
source/libs/sync/src/syncUtil.c
浏览文件 @
e060de70
...
...
@@ -179,8 +179,7 @@ void syncUtilMsgNtoH(void* msg) {
}
bool
syncUtilUserPreCommit
(
tmsg_t
msgType
)
{
if
(
msgType
!=
TDMT_SYNC_NOOP
&&
msgType
!=
TDMT_SYNC_CONFIG_CHANGE
&&
msgType
!=
TDMT_SYNC_CONFIG_CHANGE_FINISH
&&
msgType
!=
TDMT_SYNC_LEADER_TRANSFER
)
{
if
(
msgType
!=
TDMT_SYNC_NOOP
&&
msgType
!=
TDMT_SYNC_LEADER_TRANSFER
)
{
return
true
;
}
...
...
@@ -188,8 +187,7 @@ bool syncUtilUserPreCommit(tmsg_t msgType) {
}
bool
syncUtilUserCommit
(
tmsg_t
msgType
)
{
if
(
msgType
!=
TDMT_SYNC_NOOP
&&
msgType
!=
TDMT_SYNC_CONFIG_CHANGE
&&
msgType
!=
TDMT_SYNC_CONFIG_CHANGE_FINISH
&&
msgType
!=
TDMT_SYNC_LEADER_TRANSFER
)
{
if
(
msgType
!=
TDMT_SYNC_NOOP
&&
msgType
!=
TDMT_SYNC_LEADER_TRANSFER
)
{
return
true
;
}
...
...
@@ -197,8 +195,7 @@ bool syncUtilUserCommit(tmsg_t msgType) {
}
bool
syncUtilUserRollback
(
tmsg_t
msgType
)
{
if
(
msgType
!=
TDMT_SYNC_NOOP
&&
msgType
!=
TDMT_SYNC_CONFIG_CHANGE
&&
msgType
!=
TDMT_SYNC_CONFIG_CHANGE_FINISH
&&
msgType
!=
TDMT_SYNC_LEADER_TRANSFER
)
{
if
(
msgType
!=
TDMT_SYNC_NOOP
&&
msgType
!=
TDMT_SYNC_LEADER_TRANSFER
)
{
return
true
;
}
...
...
source/libs/sync/test/CMakeLists.txt
浏览文件 @
e060de70
...
...
@@ -54,7 +54,6 @@ add_executable(syncRaftLogTest "")
add_executable
(
syncRaftLogTest2
""
)
add_executable
(
syncRaftLogTest3
""
)
add_executable
(
syncLeaderTransferTest
""
)
add_executable
(
syncReconfigFinishTest
""
)
add_executable
(
syncRestoreFromSnapshot
""
)
add_executable
(
syncRaftCfgIndexTest
""
)
add_executable
(
syncHeartbeatTest
""
)
...
...
source/libs/sync/test/syncLeaderTransferTest.cpp
浏览文件 @
e060de70
...
...
@@ -27,6 +27,35 @@ SyncLeaderTransfer *createMsg() {
return
pMsg
;
}
// for debug ----------------------
void
syncLeaderTransferPrint
(
const
SyncLeaderTransfer
*
pMsg
)
{
char
*
serialized
=
syncLeaderTransfer2Str
(
pMsg
);
printf
(
"syncLeaderTransferPrint | len:%d | %s
\n
"
,
(
int32_t
)
strlen
(
serialized
),
serialized
);
fflush
(
NULL
);
taosMemoryFree
(
serialized
);
}
void
syncLeaderTransferPrint2
(
char
*
s
,
const
SyncLeaderTransfer
*
pMsg
)
{
char
*
serialized
=
syncLeaderTransfer2Str
(
pMsg
);
printf
(
"syncLeaderTransferPrint2 | len:%d | %s | %s
\n
"
,
(
int32_t
)
strlen
(
serialized
),
s
,
serialized
);
fflush
(
NULL
);
taosMemoryFree
(
serialized
);
}
void
syncLeaderTransferLog
(
const
SyncLeaderTransfer
*
pMsg
)
{
char
*
serialized
=
syncLeaderTransfer2Str
(
pMsg
);
sTrace
(
"syncLeaderTransferLog | len:%d | %s"
,
(
int32_t
)
strlen
(
serialized
),
serialized
);
taosMemoryFree
(
serialized
);
}
void
syncLeaderTransferLog2
(
char
*
s
,
const
SyncLeaderTransfer
*
pMsg
)
{
if
(
gRaftDetailLog
)
{
char
*
serialized
=
syncLeaderTransfer2Str
(
pMsg
);
sTrace
(
"syncLeaderTransferLog2 | len:%d | %s | %s"
,
(
int32_t
)
strlen
(
serialized
),
s
,
serialized
);
taosMemoryFree
(
serialized
);
}
}
void
test1
()
{
SyncLeaderTransfer
*
pMsg
=
createMsg
();
syncLeaderTransferLog2
((
char
*
)
"test1:"
,
pMsg
);
...
...
source/libs/sync/test/syncReconfigFinishTest.cpp
已删除
100644 → 0
浏览文件 @
1ddbeb43
#include <gtest/gtest.h>
#include <stdio.h>
#include "syncIO.h"
#include "syncInt.h"
#include "syncMessage.h"
#include "syncUtil.h"
void
logTest
()
{
sTrace
(
"--- sync log test: trace"
);
sDebug
(
"--- sync log test: debug"
);
sInfo
(
"--- sync log test: info"
);
sWarn
(
"--- sync log test: warn"
);
sError
(
"--- sync log test: error"
);
sFatal
(
"--- sync log test: fatal"
);
}
SSyncCfg
*
createSyncOldCfg
()
{
SSyncCfg
*
pCfg
=
(
SSyncCfg
*
)
taosMemoryMalloc
(
sizeof
(
SSyncCfg
));
memset
(
pCfg
,
0
,
sizeof
(
SSyncCfg
));
pCfg
->
replicaNum
=
3
;
pCfg
->
myIndex
=
1
;
for
(
int
i
=
0
;
i
<
pCfg
->
replicaNum
;
++
i
)
{
((
pCfg
->
nodeInfo
)[
i
]).
nodePort
=
i
*
100
;
snprintf
(((
pCfg
->
nodeInfo
)[
i
]).
nodeFqdn
,
sizeof
(((
pCfg
->
nodeInfo
)[
i
]).
nodeFqdn
),
"100.200.300.%d"
,
i
);
}
return
pCfg
;
}
SSyncCfg
*
createSyncNewCfg
()
{
SSyncCfg
*
pCfg
=
(
SSyncCfg
*
)
taosMemoryMalloc
(
sizeof
(
SSyncCfg
));
memset
(
pCfg
,
0
,
sizeof
(
SSyncCfg
));
pCfg
->
replicaNum
=
3
;
pCfg
->
myIndex
=
1
;
for
(
int
i
=
0
;
i
<
pCfg
->
replicaNum
;
++
i
)
{
((
pCfg
->
nodeInfo
)[
i
]).
nodePort
=
i
*
100
;
snprintf
(((
pCfg
->
nodeInfo
)[
i
]).
nodeFqdn
,
sizeof
(((
pCfg
->
nodeInfo
)[
i
]).
nodeFqdn
),
"500.600.700.%d"
,
i
);
}
return
pCfg
;
}
SyncReconfigFinish
*
createMsg
()
{
SyncReconfigFinish
*
pMsg
=
syncReconfigFinishBuild
(
1234
);
SSyncCfg
*
pOld
=
createSyncOldCfg
();
SSyncCfg
*
pNew
=
createSyncNewCfg
();
pMsg
->
oldCfg
=
*
pOld
;
pMsg
->
newCfg
=
*
pNew
;
pMsg
->
newCfgIndex
=
11
;
pMsg
->
newCfgTerm
=
22
;
pMsg
->
newCfgSeqNum
=
33
;
taosMemoryFree
(
pOld
);
taosMemoryFree
(
pNew
);
return
pMsg
;
}
void
test1
()
{
SyncReconfigFinish
*
pMsg
=
createMsg
();
syncReconfigFinishLog2
((
char
*
)
"test1:"
,
pMsg
);
syncReconfigFinishDestroy
(
pMsg
);
}
void
test2
()
{
SyncReconfigFinish
*
pMsg
=
createMsg
();
uint32_t
len
=
pMsg
->
bytes
;
char
*
serialized
=
(
char
*
)
taosMemoryMalloc
(
len
);
syncReconfigFinishSerialize
(
pMsg
,
serialized
,
len
);
SyncReconfigFinish
*
pMsg2
=
syncReconfigFinishBuild
(
1000
);
syncReconfigFinishDeserialize
(
serialized
,
len
,
pMsg2
);
syncReconfigFinishLog2
((
char
*
)
"test2: syncReconfigFinishSerialize -> syncReconfigFinishDeserialize "
,
pMsg2
);
taosMemoryFree
(
serialized
);
syncReconfigFinishDestroy
(
pMsg
);
syncReconfigFinishDestroy
(
pMsg2
);
}
void
test3
()
{
SyncReconfigFinish
*
pMsg
=
createMsg
();
uint32_t
len
;
char
*
serialized
=
syncReconfigFinishSerialize2
(
pMsg
,
&
len
);
SyncReconfigFinish
*
pMsg2
=
syncReconfigFinishDeserialize2
(
serialized
,
len
);
syncReconfigFinishLog2
((
char
*
)
"test3: SyncReconfigFinishSerialize2 -> syncReconfigFinishDeserialize2 "
,
pMsg2
);
taosMemoryFree
(
serialized
);
syncReconfigFinishDestroy
(
pMsg
);
syncReconfigFinishDestroy
(
pMsg2
);
}
void
test4
()
{
SyncReconfigFinish
*
pMsg
=
createMsg
();
SRpcMsg
rpcMsg
;
syncReconfigFinish2RpcMsg
(
pMsg
,
&
rpcMsg
);
SyncReconfigFinish
*
pMsg2
=
(
SyncReconfigFinish
*
)
taosMemoryMalloc
(
rpcMsg
.
contLen
);
syncReconfigFinishFromRpcMsg
(
&
rpcMsg
,
pMsg2
);
syncReconfigFinishLog2
((
char
*
)
"test4: syncReconfigFinish2RpcMsg -> syncReconfigFinishFromRpcMsg "
,
pMsg2
);
rpcFreeCont
(
rpcMsg
.
pCont
);
syncReconfigFinishDestroy
(
pMsg
);
syncReconfigFinishDestroy
(
pMsg2
);
}
void
test5
()
{
SyncReconfigFinish
*
pMsg
=
createMsg
();
SRpcMsg
rpcMsg
;
syncReconfigFinish2RpcMsg
(
pMsg
,
&
rpcMsg
);
SyncReconfigFinish
*
pMsg2
=
syncReconfigFinishFromRpcMsg2
(
&
rpcMsg
);
syncReconfigFinishLog2
((
char
*
)
"test5: syncReconfigFinish2RpcMsg -> syncReconfigFinishFromRpcMsg2 "
,
pMsg2
);
rpcFreeCont
(
rpcMsg
.
pCont
);
syncReconfigFinishDestroy
(
pMsg
);
syncReconfigFinishDestroy
(
pMsg2
);
}
int
main
()
{
gRaftDetailLog
=
true
;
tsAsyncLog
=
0
;
sDebugFlag
=
DEBUG_TRACE
+
DEBUG_SCREEN
+
DEBUG_FILE
;
logTest
();
test1
();
test2
();
test3
();
test4
();
test5
();
return
0
;
}
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录