Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
1e7cbee2
T
TDengine
项目概览
taosdata
/
TDengine
12 个月 前同步成功
通知
1180
Star
22014
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,发现更多精彩内容 >>
未验证
提交
1e7cbee2
编写于
3月 09, 2022
作者:
L
Li Minghao
提交者:
GitHub
3月 09, 2022
浏览文件
操作
浏览文件
下载
差异文件
Merge pull request #10640 from taosdata/feature/3.0_mhli
Feature/3.0 mhli
上级
290c04d7
57b4e7ce
变更
8
隐藏空白更改
内联
并排
Showing
8 changed file
with
530 addition
and
8 deletion
+530
-8
source/libs/sync/inc/syncIndexMgr.h
source/libs/sync/inc/syncIndexMgr.h
+49
-0
source/libs/sync/inc/syncInt.h
source/libs/sync/inc/syncInt.h
+5
-2
source/libs/sync/inc/syncVoteMgr.h
source/libs/sync/inc/syncVoteMgr.h
+4
-2
source/libs/sync/src/syncIndexMgr.c
source/libs/sync/src/syncIndexMgr.c
+100
-0
source/libs/sync/src/syncVoteMgr.c
source/libs/sync/src/syncVoteMgr.c
+47
-4
source/libs/sync/test/CMakeLists.txt
source/libs/sync/test/CMakeLists.txt
+28
-0
source/libs/sync/test/syncIndexMgrTest.cpp
source/libs/sync/test/syncIndexMgrTest.cpp
+141
-0
source/libs/sync/test/syncVotesRespondTest.cpp
source/libs/sync/test/syncVotesRespondTest.cpp
+156
-0
未找到文件。
source/libs/sync/inc/syncIndexMgr.h
0 → 100644
浏览文件 @
1e7cbee2
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef _TD_LIBS_SYNC_INDEX_MGR_H
#define _TD_LIBS_SYNC_INDEX_MGR_H
#ifdef __cplusplus
extern
"C"
{
#endif
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include "syncInt.h"
#include "taosdef.h"
// SIndexMgr -----------------------------
typedef
struct
SSyncIndexMgr
{
SRaftId
(
*
replicas
)[
TSDB_MAX_REPLICA
];
SyncIndex
index
[
TSDB_MAX_REPLICA
];
int32_t
replicaNum
;
SSyncNode
*
pSyncNode
;
}
SSyncIndexMgr
;
SSyncIndexMgr
*
syncIndexMgrCreate
(
SSyncNode
*
pSyncNode
);
void
syncIndexMgrDestroy
(
SSyncIndexMgr
*
pSyncIndexMgr
);
void
syncIndexMgrClear
(
SSyncIndexMgr
*
pSyncIndexMgr
);
void
syncIndexMgrSetIndex
(
SSyncIndexMgr
*
pSyncIndexMgr
,
const
SRaftId
*
pRaftId
,
SyncIndex
index
);
SyncIndex
syncIndexMgrGetIndex
(
SSyncIndexMgr
*
pSyncIndexMgr
,
const
SRaftId
*
pRaftId
);
cJSON
*
syncIndexMgr2Json
(
SSyncIndexMgr
*
pSyncIndexMgr
);
char
*
syncIndexMgr2Str
(
SSyncIndexMgr
*
pSyncIndexMgr
);
#ifdef __cplusplus
}
#endif
#endif
/*_TD_LIBS_SYNC_INDEX_MGR_H*/
source/libs/sync/inc/syncInt.h
浏览文件 @
1e7cbee2
...
...
@@ -103,6 +103,9 @@ typedef struct SVotesGranted SVotesGranted;
struct
SVotesRespond
;
typedef
struct
SVotesRespond
SVotesRespond
;
struct
SSyncIndexMgr
;
typedef
struct
SSyncIndexMgr
SSyncIndexMgr
;
typedef
struct
SRaftId
{
SyncNodeId
addr
;
// typedef uint64_t SyncNodeId;
SyncGroupId
vgId
;
// typedef int32_t SyncGroupId;
...
...
@@ -148,8 +151,8 @@ typedef struct SSyncNode {
SVotesRespond
*
pVotesRespond
;
// tla+ leader vars
S
HashObj
*
pNextIndex
;
S
HashObj
*
pMatchIndex
;
S
SyncIndexMgr
*
pNextIndex
;
S
SyncIndexMgr
*
pMatchIndex
;
// tla+ log vars
SSyncLogStore
*
pLogStore
;
...
...
source/libs/sync/inc/syncVoteMgr.h
浏览文件 @
1e7cbee2
...
...
@@ -31,8 +31,8 @@ extern "C" {
// SVotesGranted -----------------------------
typedef
struct
SVotesGranted
{
SRaftId
(
*
replicas
)[
TSDB_MAX_REPLICA
];
int32_t
replicaNum
;
bool
isGranted
[
TSDB_MAX_REPLICA
];
int32_t
replicaNum
;
int32_t
votes
;
SyncTerm
term
;
int32_t
quorum
;
...
...
@@ -61,7 +61,9 @@ SVotesRespond *votesRespondCreate(SSyncNode *pSyncNode);
void
votesRespondDestory
(
SVotesRespond
*
pVotesRespond
);
bool
votesResponded
(
SVotesRespond
*
pVotesRespond
,
const
SRaftId
*
pRaftId
);
void
votesRespondAdd
(
SVotesRespond
*
pVotesRespond
,
const
SyncRequestVoteReply
*
pMsg
);
void
Reset
(
SVotesRespond
*
pVotesRespond
,
SyncTerm
term
);
void
votesRespondReset
(
SVotesRespond
*
pVotesRespond
,
SyncTerm
term
);
cJSON
*
votesRespond2Json
(
SVotesRespond
*
pVotesRespond
);
char
*
votesRespond2Str
(
SVotesRespond
*
pVotesRespond
);
#ifdef __cplusplus
}
...
...
source/libs/sync/src/syncIndexMgr.c
0 → 100644
浏览文件 @
1e7cbee2
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "syncIndexMgr.h"
#include "syncUtil.h"
// SMatchIndex -----------------------------
SSyncIndexMgr
*
syncIndexMgrCreate
(
SSyncNode
*
pSyncNode
)
{
SSyncIndexMgr
*
pSyncIndexMgr
=
malloc
(
sizeof
(
SSyncIndexMgr
));
assert
(
pSyncIndexMgr
!=
NULL
);
memset
(
pSyncIndexMgr
,
0
,
sizeof
(
SSyncIndexMgr
));
pSyncIndexMgr
->
replicas
=
&
(
pSyncNode
->
replicasId
);
pSyncIndexMgr
->
replicaNum
=
pSyncNode
->
replicaNum
;
pSyncIndexMgr
->
pSyncNode
=
pSyncNode
;
syncIndexMgrClear
(
pSyncIndexMgr
);
return
pSyncIndexMgr
;
}
void
syncIndexMgrDestroy
(
SSyncIndexMgr
*
pSyncIndexMgr
)
{
if
(
pSyncIndexMgr
!=
NULL
)
{
free
(
pSyncIndexMgr
);
}
}
void
syncIndexMgrClear
(
SSyncIndexMgr
*
pSyncIndexMgr
)
{
memset
(
pSyncIndexMgr
->
index
,
0
,
sizeof
(
pSyncIndexMgr
->
index
));
/*
for (int i = 0; i < pSyncIndexMgr->replicaNum; ++i) {
pSyncIndexMgr->index[i] = 0;
}
*/
}
void
syncIndexMgrSetIndex
(
SSyncIndexMgr
*
pSyncIndexMgr
,
const
SRaftId
*
pRaftId
,
SyncIndex
index
)
{
for
(
int
i
=
0
;
i
<
pSyncIndexMgr
->
replicaNum
;
++
i
)
{
if
(
syncUtilSameId
(
&
((
*
(
pSyncIndexMgr
->
replicas
))[
i
]),
pRaftId
))
{
(
pSyncIndexMgr
->
index
)[
i
]
=
index
;
return
;
}
}
assert
(
0
);
}
SyncIndex
syncIndexMgrGetIndex
(
SSyncIndexMgr
*
pSyncIndexMgr
,
const
SRaftId
*
pRaftId
)
{
for
(
int
i
=
0
;
i
<
pSyncIndexMgr
->
replicaNum
;
++
i
)
{
if
(
syncUtilSameId
(
&
((
*
(
pSyncIndexMgr
->
replicas
))[
i
]),
pRaftId
))
{
SyncIndex
idx
=
(
pSyncIndexMgr
->
index
)[
i
];
return
idx
;
}
}
assert
(
0
);
}
cJSON
*
syncIndexMgr2Json
(
SSyncIndexMgr
*
pSyncIndexMgr
)
{
char
u64buf
[
128
];
cJSON
*
pRoot
=
cJSON_CreateObject
();
cJSON_AddNumberToObject
(
pRoot
,
"replicaNum"
,
pSyncIndexMgr
->
replicaNum
);
cJSON
*
pReplicas
=
cJSON_CreateArray
();
cJSON_AddItemToObject
(
pRoot
,
"replicas"
,
pReplicas
);
for
(
int
i
=
0
;
i
<
pSyncIndexMgr
->
replicaNum
;
++
i
)
{
cJSON_AddItemToArray
(
pReplicas
,
syncUtilRaftId2Json
(
&
(
*
(
pSyncIndexMgr
->
replicas
))[
i
]));
}
int
respondNum
=
0
;
int
*
arr
=
(
int
*
)
malloc
(
sizeof
(
int
)
*
pSyncIndexMgr
->
replicaNum
);
for
(
int
i
=
0
;
i
<
pSyncIndexMgr
->
replicaNum
;
++
i
)
{
arr
[
i
]
=
pSyncIndexMgr
->
index
[
i
];
}
cJSON
*
pIndex
=
cJSON_CreateIntArray
(
arr
,
pSyncIndexMgr
->
replicaNum
);
free
(
arr
);
cJSON_AddItemToObject
(
pRoot
,
"index"
,
pIndex
);
snprintf
(
u64buf
,
sizeof
(
u64buf
),
"%p"
,
pSyncIndexMgr
->
pSyncNode
);
cJSON_AddStringToObject
(
pRoot
,
"pSyncNode"
,
u64buf
);
cJSON
*
pJson
=
cJSON_CreateObject
();
cJSON_AddItemToObject
(
pJson
,
"pSyncIndexMgr"
,
pRoot
);
return
pJson
;
}
char
*
syncIndexMgr2Str
(
SSyncIndexMgr
*
pSyncIndexMgr
)
{
cJSON
*
pJson
=
syncIndexMgr2Json
(
pSyncIndexMgr
);
char
*
serialized
=
cJSON_Print
(
pJson
);
cJSON_Delete
(
pJson
);
return
serialized
;
}
\ No newline at end of file
source/libs/sync/src/syncVoteMgr.c
浏览文件 @
1e7cbee2
...
...
@@ -153,8 +153,8 @@ bool votesResponded(SVotesRespond *pVotesRespond, const SRaftId *pRaftId) {
void
votesRespondAdd
(
SVotesRespond
*
pVotesRespond
,
const
SyncRequestVoteReply
*
pMsg
)
{
assert
(
pVotesRespond
->
term
==
pMsg
->
term
);
for
(
int
i
=
0
;
i
<
pVotesRespond
->
replicaNum
;
++
i
)
{
if
(
syncUtilSameId
(
&
(
*
pVotesRespond
->
replicas
)[
i
]
,
&
pMsg
->
srcId
))
{
assert
(
pVotesRespond
->
isRespond
[
i
]
==
false
);
if
(
syncUtilSameId
(
&
(
(
*
(
pVotesRespond
->
replicas
))[
i
])
,
&
pMsg
->
srcId
))
{
//
assert(pVotesRespond->isRespond[i] == false);
pVotesRespond
->
isRespond
[
i
]
=
true
;
return
;
}
...
...
@@ -162,9 +162,52 @@ void votesRespondAdd(SVotesRespond *pVotesRespond, const SyncRequestVoteReply *p
assert
(
0
);
}
void
Reset
(
SVotesRespond
*
pVotesRespond
,
SyncTerm
term
)
{
void
votesRespond
Reset
(
SVotesRespond
*
pVotesRespond
,
SyncTerm
term
)
{
pVotesRespond
->
term
=
term
;
memset
(
pVotesRespond
->
isRespond
,
0
,
sizeof
(
pVotesRespond
->
isRespond
));
/*
for (int i = 0; i < pVotesRespond->replicaNum; ++i) {
pVotesRespond->isRespond[i] = false;
}
*/
}
cJSON
*
votesRespond2Json
(
SVotesRespond
*
pVotesRespond
)
{
char
u64buf
[
128
];
cJSON
*
pRoot
=
cJSON_CreateObject
();
cJSON_AddNumberToObject
(
pRoot
,
"replicaNum"
,
pVotesRespond
->
replicaNum
);
cJSON
*
pReplicas
=
cJSON_CreateArray
();
cJSON_AddItemToObject
(
pRoot
,
"replicas"
,
pReplicas
);
for
(
int
i
=
0
;
i
<
pVotesRespond
->
replicaNum
;
++
i
)
{
cJSON_AddItemToArray
(
pReplicas
,
syncUtilRaftId2Json
(
&
(
*
(
pVotesRespond
->
replicas
))[
i
]));
}
int
respondNum
=
0
;
int
*
arr
=
(
int
*
)
malloc
(
sizeof
(
int
)
*
pVotesRespond
->
replicaNum
);
for
(
int
i
=
0
;
i
<
pVotesRespond
->
replicaNum
;
++
i
)
{
pVotesRespond
->
isRespond
[
i
]
=
false
;
arr
[
i
]
=
pVotesRespond
->
isRespond
[
i
];
if
(
pVotesRespond
->
isRespond
[
i
])
{
respondNum
++
;
}
}
cJSON
*
pIsRespond
=
cJSON_CreateIntArray
(
arr
,
pVotesRespond
->
replicaNum
);
free
(
arr
);
cJSON_AddItemToObject
(
pRoot
,
"isRespond"
,
pIsRespond
);
cJSON_AddNumberToObject
(
pRoot
,
"respondNum"
,
respondNum
);
snprintf
(
u64buf
,
sizeof
(
u64buf
),
"%lu"
,
pVotesRespond
->
term
);
cJSON_AddStringToObject
(
pRoot
,
"term"
,
u64buf
);
snprintf
(
u64buf
,
sizeof
(
u64buf
),
"%p"
,
pVotesRespond
->
pSyncNode
);
cJSON_AddStringToObject
(
pRoot
,
"pSyncNode"
,
u64buf
);
cJSON
*
pJson
=
cJSON_CreateObject
();
cJSON_AddItemToObject
(
pJson
,
"SVotesRespond"
,
pRoot
);
return
pJson
;
}
char
*
votesRespond2Str
(
SVotesRespond
*
pVotesRespond
)
{
cJSON
*
pJson
=
votesRespond2Json
(
pVotesRespond
);
char
*
serialized
=
cJSON_Print
(
pJson
);
cJSON_Delete
(
pJson
);
return
serialized
;
}
\ No newline at end of file
source/libs/sync/test/CMakeLists.txt
浏览文件 @
1e7cbee2
...
...
@@ -13,6 +13,8 @@ add_executable(syncIndexTest "")
add_executable
(
syncInitTest
""
)
add_executable
(
syncUtilTest
""
)
add_executable
(
syncVotesGrantedTest
""
)
add_executable
(
syncVotesRespondTest
""
)
add_executable
(
syncIndexMgrTest
""
)
target_sources
(
syncTest
...
...
@@ -75,6 +77,14 @@ target_sources(syncVotesGrantedTest
PRIVATE
"syncVotesGrantedTest.cpp"
)
target_sources
(
syncVotesRespondTest
PRIVATE
"syncVotesRespondTest.cpp"
)
target_sources
(
syncIndexMgrTest
PRIVATE
"syncIndexMgrTest.cpp"
)
target_include_directories
(
syncTest
...
...
@@ -152,6 +162,16 @@ target_include_directories(syncVotesGrantedTest
"
${
CMAKE_SOURCE_DIR
}
/include/libs/sync"
"
${
CMAKE_CURRENT_SOURCE_DIR
}
/../inc"
)
target_include_directories
(
syncVotesRespondTest
PUBLIC
"
${
CMAKE_SOURCE_DIR
}
/include/libs/sync"
"
${
CMAKE_CURRENT_SOURCE_DIR
}
/../inc"
)
target_include_directories
(
syncIndexMgrTest
PUBLIC
"
${
CMAKE_SOURCE_DIR
}
/include/libs/sync"
"
${
CMAKE_CURRENT_SOURCE_DIR
}
/../inc"
)
target_link_libraries
(
syncTest
...
...
@@ -214,6 +234,14 @@ target_link_libraries(syncVotesGrantedTest
sync
gtest_main
)
target_link_libraries
(
syncVotesRespondTest
sync
gtest_main
)
target_link_libraries
(
syncIndexMgrTest
sync
gtest_main
)
enable_testing
()
...
...
source/libs/sync/test/syncIndexMgrTest.cpp
0 → 100644
浏览文件 @
1e7cbee2
#include "syncIndexMgr.h"
//#include <gtest/gtest.h>
#include <stdio.h>
#include "syncEnv.h"
#include "syncIO.h"
#include "syncInt.h"
#include "syncRaftStore.h"
#include "syncUtil.h"
#include "syncVoteMgr.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"
);
}
uint16_t
ports
[]
=
{
7010
,
7110
,
7210
,
7310
,
7410
};
int32_t
replicaNum
=
3
;
int32_t
myIndex
=
0
;
SRaftId
ids
[
TSDB_MAX_REPLICA
];
SSyncInfo
syncInfo
;
SSyncFSM
*
pFsm
;
SSyncNode
*
pSyncNode
;
SSyncNode
*
syncNodeInit
()
{
syncInfo
.
vgId
=
1234
;
syncInfo
.
rpcClient
=
gSyncIO
->
clientRpc
;
syncInfo
.
FpSendMsg
=
syncIOSendMsg
;
syncInfo
.
queue
=
gSyncIO
->
pMsgQ
;
syncInfo
.
FpEqMsg
=
syncIOEqMsg
;
syncInfo
.
pFsm
=
pFsm
;
snprintf
(
syncInfo
.
path
,
sizeof
(
syncInfo
.
path
),
"%s"
,
"./test_path"
);
snprintf
(
syncInfo
.
walPath
,
sizeof
(
syncInfo
.
walPath
),
"%s"
,
"./test_wal_path"
);
SSyncCfg
*
pCfg
=
&
syncInfo
.
syncCfg
;
pCfg
->
myIndex
=
myIndex
;
pCfg
->
replicaNum
=
replicaNum
;
for
(
int
i
=
0
;
i
<
replicaNum
;
++
i
)
{
pCfg
->
nodeInfo
[
i
].
nodePort
=
ports
[
i
];
snprintf
(
pCfg
->
nodeInfo
[
i
].
nodeFqdn
,
sizeof
(
pCfg
->
nodeInfo
[
i
].
nodeFqdn
),
"%s"
,
"127.0.0.1"
);
// taosGetFqdn(pCfg->nodeInfo[0].nodeFqdn);
}
pSyncNode
=
syncNodeOpen
(
&
syncInfo
);
assert
(
pSyncNode
!=
NULL
);
gSyncIO
->
FpOnSyncPing
=
pSyncNode
->
FpOnPing
;
gSyncIO
->
FpOnSyncPingReply
=
pSyncNode
->
FpOnPingReply
;
gSyncIO
->
FpOnSyncRequestVote
=
pSyncNode
->
FpOnRequestVote
;
gSyncIO
->
FpOnSyncRequestVoteReply
=
pSyncNode
->
FpOnRequestVoteReply
;
gSyncIO
->
FpOnSyncAppendEntries
=
pSyncNode
->
FpOnAppendEntries
;
gSyncIO
->
FpOnSyncAppendEntriesReply
=
pSyncNode
->
FpOnAppendEntriesReply
;
gSyncIO
->
FpOnSyncPing
=
pSyncNode
->
FpOnPing
;
gSyncIO
->
FpOnSyncPingReply
=
pSyncNode
->
FpOnPingReply
;
gSyncIO
->
pSyncNode
=
pSyncNode
;
return
pSyncNode
;
}
SSyncNode
*
syncInitTest
()
{
return
syncNodeInit
();
}
void
initRaftId
(
SSyncNode
*
pSyncNode
)
{
for
(
int
i
=
0
;
i
<
replicaNum
;
++
i
)
{
ids
[
i
]
=
pSyncNode
->
replicasId
[
i
];
char
*
s
=
syncUtilRaftId2Str
(
&
ids
[
i
]);
printf
(
"raftId[%d] : %s
\n
"
,
i
,
s
);
free
(
s
);
}
}
int
main
(
int
argc
,
char
**
argv
)
{
// taosInitLog((char *)"syncTest.log", 100000, 10);
tsAsyncLog
=
0
;
sDebugFlag
=
143
+
64
;
myIndex
=
0
;
if
(
argc
>=
2
)
{
myIndex
=
atoi
(
argv
[
1
]);
}
int32_t
ret
=
syncIOStart
((
char
*
)
"127.0.0.1"
,
ports
[
myIndex
]);
assert
(
ret
==
0
);
ret
=
syncEnvStart
();
assert
(
ret
==
0
);
SSyncNode
*
pSyncNode
=
syncInitTest
();
assert
(
pSyncNode
!=
NULL
);
char
*
serialized
=
syncNode2Str
(
pSyncNode
);
printf
(
"%s
\n
"
,
serialized
);
free
(
serialized
);
initRaftId
(
pSyncNode
);
SSyncIndexMgr
*
pSyncIndexMgr
=
syncIndexMgrCreate
(
pSyncNode
);
assert
(
pSyncIndexMgr
!=
NULL
);
printf
(
"---------------------------------------
\n
"
);
{
char
*
serialized
=
syncIndexMgr2Str
(
pSyncIndexMgr
);
assert
(
serialized
!=
NULL
);
printf
(
"%s
\n
"
,
serialized
);
free
(
serialized
);
}
syncIndexMgrSetIndex
(
pSyncIndexMgr
,
&
ids
[
0
],
100
);
syncIndexMgrSetIndex
(
pSyncIndexMgr
,
&
ids
[
1
],
200
);
syncIndexMgrSetIndex
(
pSyncIndexMgr
,
&
ids
[
2
],
300
);
printf
(
"---------------------------------------
\n
"
);
{
char
*
serialized
=
syncIndexMgr2Str
(
pSyncIndexMgr
);
assert
(
serialized
!=
NULL
);
printf
(
"%s
\n
"
,
serialized
);
free
(
serialized
);
}
printf
(
"---------------------------------------
\n
"
);
for
(
int
i
=
0
;
i
<
pSyncIndexMgr
->
replicaNum
;
++
i
)
{
SyncIndex
idx
=
syncIndexMgrGetIndex
(
pSyncIndexMgr
,
&
ids
[
i
]);
printf
(
"index %d : %lu
\n
"
,
i
,
idx
);
}
syncIndexMgrClear
(
pSyncIndexMgr
);
printf
(
"---------------------------------------
\n
"
);
{
char
*
serialized
=
syncIndexMgr2Str
(
pSyncIndexMgr
);
assert
(
serialized
!=
NULL
);
printf
(
"%s
\n
"
,
serialized
);
free
(
serialized
);
}
syncIndexMgrDestroy
(
pSyncIndexMgr
);
return
0
;
}
source/libs/sync/test/syncVotesRespondTest.cpp
0 → 100644
浏览文件 @
1e7cbee2
#include <gtest/gtest.h>
#include <stdio.h>
#include "syncEnv.h"
#include "syncIO.h"
#include "syncInt.h"
#include "syncRaftStore.h"
#include "syncUtil.h"
#include "syncVoteMgr.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"
);
}
uint16_t
ports
[]
=
{
7010
,
7110
,
7210
,
7310
,
7410
};
int32_t
replicaNum
=
3
;
int32_t
myIndex
=
0
;
SRaftId
ids
[
TSDB_MAX_REPLICA
];
SSyncInfo
syncInfo
;
SSyncFSM
*
pFsm
;
SSyncNode
*
pSyncNode
;
SSyncNode
*
syncNodeInit
()
{
syncInfo
.
vgId
=
1234
;
syncInfo
.
rpcClient
=
gSyncIO
->
clientRpc
;
syncInfo
.
FpSendMsg
=
syncIOSendMsg
;
syncInfo
.
queue
=
gSyncIO
->
pMsgQ
;
syncInfo
.
FpEqMsg
=
syncIOEqMsg
;
syncInfo
.
pFsm
=
pFsm
;
snprintf
(
syncInfo
.
path
,
sizeof
(
syncInfo
.
path
),
"%s"
,
"./test_path"
);
snprintf
(
syncInfo
.
walPath
,
sizeof
(
syncInfo
.
walPath
),
"%s"
,
"./test_wal_path"
);
SSyncCfg
*
pCfg
=
&
syncInfo
.
syncCfg
;
pCfg
->
myIndex
=
myIndex
;
pCfg
->
replicaNum
=
replicaNum
;
for
(
int
i
=
0
;
i
<
replicaNum
;
++
i
)
{
pCfg
->
nodeInfo
[
i
].
nodePort
=
ports
[
i
];
snprintf
(
pCfg
->
nodeInfo
[
i
].
nodeFqdn
,
sizeof
(
pCfg
->
nodeInfo
[
i
].
nodeFqdn
),
"%s"
,
"127.0.0.1"
);
// taosGetFqdn(pCfg->nodeInfo[0].nodeFqdn);
}
pSyncNode
=
syncNodeOpen
(
&
syncInfo
);
assert
(
pSyncNode
!=
NULL
);
gSyncIO
->
FpOnSyncPing
=
pSyncNode
->
FpOnPing
;
gSyncIO
->
FpOnSyncPingReply
=
pSyncNode
->
FpOnPingReply
;
gSyncIO
->
FpOnSyncRequestVote
=
pSyncNode
->
FpOnRequestVote
;
gSyncIO
->
FpOnSyncRequestVoteReply
=
pSyncNode
->
FpOnRequestVoteReply
;
gSyncIO
->
FpOnSyncAppendEntries
=
pSyncNode
->
FpOnAppendEntries
;
gSyncIO
->
FpOnSyncAppendEntriesReply
=
pSyncNode
->
FpOnAppendEntriesReply
;
gSyncIO
->
FpOnSyncPing
=
pSyncNode
->
FpOnPing
;
gSyncIO
->
FpOnSyncPingReply
=
pSyncNode
->
FpOnPingReply
;
gSyncIO
->
pSyncNode
=
pSyncNode
;
return
pSyncNode
;
}
SSyncNode
*
syncInitTest
()
{
return
syncNodeInit
();
}
void
initRaftId
(
SSyncNode
*
pSyncNode
)
{
for
(
int
i
=
0
;
i
<
replicaNum
;
++
i
)
{
ids
[
i
]
=
pSyncNode
->
replicasId
[
i
];
char
*
s
=
syncUtilRaftId2Str
(
&
ids
[
i
]);
printf
(
"raftId[%d] : %s
\n
"
,
i
,
s
);
free
(
s
);
}
}
int
main
(
int
argc
,
char
**
argv
)
{
// taosInitLog((char *)"syncTest.log", 100000, 10);
tsAsyncLog
=
0
;
sDebugFlag
=
143
+
64
;
myIndex
=
0
;
if
(
argc
>=
2
)
{
myIndex
=
atoi
(
argv
[
1
]);
}
int32_t
ret
=
syncIOStart
((
char
*
)
"127.0.0.1"
,
ports
[
myIndex
]);
assert
(
ret
==
0
);
ret
=
syncEnvStart
();
assert
(
ret
==
0
);
SSyncNode
*
pSyncNode
=
syncInitTest
();
assert
(
pSyncNode
!=
NULL
);
char
*
serialized
=
syncNode2Str
(
pSyncNode
);
printf
(
"%s
\n
"
,
serialized
);
free
(
serialized
);
initRaftId
(
pSyncNode
);
SVotesRespond
*
pVotesRespond
=
votesRespondCreate
(
pSyncNode
);
assert
(
pVotesRespond
!=
NULL
);
printf
(
"---------------------------------------
\n
"
);
{
char
*
serialized
=
votesRespond2Str
(
pVotesRespond
);
assert
(
serialized
!=
NULL
);
printf
(
"%s
\n
"
,
serialized
);
free
(
serialized
);
}
SyncTerm
term
=
1234
;
printf
(
"---------------------------------------
\n
"
);
votesRespondReset
(
pVotesRespond
,
term
);
{
char
*
serialized
=
votesRespond2Str
(
pVotesRespond
);
assert
(
serialized
!=
NULL
);
printf
(
"%s
\n
"
,
serialized
);
free
(
serialized
);
}
for
(
int
i
=
0
;
i
<
replicaNum
;
++
i
)
{
SyncRequestVoteReply
*
reply
=
SyncRequestVoteReplyBuild
();
reply
->
destId
=
pSyncNode
->
myRaftId
;
reply
->
srcId
=
ids
[
i
];
reply
->
term
=
term
;
reply
->
voteGranted
=
true
;
votesRespondAdd
(
pVotesRespond
,
reply
);
{
char
*
serialized
=
votesRespond2Str
(
pVotesRespond
);
assert
(
serialized
!=
NULL
);
printf
(
"%s
\n
"
,
serialized
);
free
(
serialized
);
}
votesRespondAdd
(
pVotesRespond
,
reply
);
{
char
*
serialized
=
votesRespond2Str
(
pVotesRespond
);
assert
(
serialized
!=
NULL
);
printf
(
"%s
\n
"
,
serialized
);
free
(
serialized
);
}
}
printf
(
"---------------------------------------
\n
"
);
votesRespondReset
(
pVotesRespond
,
123456789
);
{
char
*
serialized
=
votesRespond2Str
(
pVotesRespond
);
assert
(
serialized
!=
NULL
);
printf
(
"%s
\n
"
,
serialized
);
free
(
serialized
);
}
votesRespondDestory
(
pVotesRespond
);
return
0
;
}
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录