Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
559f570e
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,发现更多精彩内容 >>
提交
559f570e
编写于
3月 22, 2022
作者:
M
Minghao Li
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
sync refactor
上级
0c9e64f0
变更
5
隐藏空白更改
内联
并排
Showing
5 changed file
with
223 addition
and
17 deletion
+223
-17
source/libs/sync/src/syncIO.c
source/libs/sync/src/syncIO.c
+3
-10
source/libs/sync/src/syncMain.c
source/libs/sync/src/syncMain.c
+1
-1
source/libs/sync/src/syncMessage.c
source/libs/sync/src/syncMessage.c
+6
-6
source/libs/sync/test/CMakeLists.txt
source/libs/sync/test/CMakeLists.txt
+14
-0
source/libs/sync/test/syncReplicateTest2.cpp
source/libs/sync/test/syncReplicateTest2.cpp
+199
-0
未找到文件。
source/libs/sync/src/syncIO.c
浏览文件 @
559f570e
...
...
@@ -29,7 +29,7 @@ static int32_t syncIODestroy(SSyncIO *io);
static
int32_t
syncIOStartInternal
(
SSyncIO
*
io
);
static
int32_t
syncIOStopInternal
(
SSyncIO
*
io
);
static
void
*
syncIOConsumerFunc
(
void
*
param
);
static
void
*
syncIOConsumerFunc
(
void
*
param
);
static
void
syncIOProcessRequest
(
void
*
pParent
,
SRpcMsg
*
pMsg
,
SEpSet
*
pEpSet
);
static
void
syncIOProcessReply
(
void
*
pParent
,
SRpcMsg
*
pMsg
,
SEpSet
*
pEpSet
);
static
int32_t
syncIOAuth
(
void
*
parent
,
char
*
meterId
,
char
*
spi
,
char
*
encrypt
,
char
*
secret
,
char
*
ckey
);
...
...
@@ -234,9 +234,9 @@ static int32_t syncIOStopInternal(SSyncIO *io) {
}
static
void
*
syncIOConsumerFunc
(
void
*
param
)
{
SSyncIO
*
io
=
param
;
SSyncIO
*
io
=
param
;
STaosQall
*
qall
;
SRpcMsg
*
pRpcMsg
,
rpcMsg
;
SRpcMsg
*
pRpcMsg
,
rpcMsg
;
qall
=
taosAllocateQall
();
while
(
1
)
{
...
...
@@ -257,13 +257,6 @@ static void *syncIOConsumerFunc(void *param) {
assert
(
pSyncMsg
!=
NULL
);
io
->
FpOnSyncPing
(
io
->
pSyncNode
,
pSyncMsg
);
syncPingDestroy
(
pSyncMsg
);
/*
pSyncMsg = syncPingBuild(pRpcMsg->contLen);
syncPingFromRpcMsg(pRpcMsg, pSyncMsg);
// memcpy(pSyncMsg, tmpRpcMsg.pCont, tmpRpcMsg.contLen);
io->FpOnSyncPing(io->pSyncNode, pSyncMsg);
syncPingDestroy(pSyncMsg);
*/
}
}
else
if
(
pRpcMsg
->
msgType
==
SYNC_PING_REPLY
)
{
...
...
source/libs/sync/src/syncMain.c
浏览文件 @
559f570e
...
...
@@ -122,7 +122,7 @@ int32_t syncPropose(int64_t rid, const SRpcMsg* pMsg, bool isWeak) {
ret
=
0
;
}
else
{
sTrace
(
"sync
ForwardToPeer
not leader, %s"
,
syncUtilState2String
(
pSyncNode
->
state
));
sTrace
(
"sync
Propose
not leader, %s"
,
syncUtilState2String
(
pSyncNode
->
state
));
ret
=
-
1
;
// todo : need define err code !!
}
...
...
source/libs/sync/src/syncMessage.c
浏览文件 @
559f570e
...
...
@@ -834,7 +834,7 @@ cJSON* syncRequestVote2Json(const SyncRequestVote* pMsg) {
snprintf
(
u64buf
,
sizeof
(
u64buf
),
"%lu"
,
pMsg
->
term
);
cJSON_AddStringToObject
(
pRoot
,
"term"
,
u64buf
);
snprintf
(
u64buf
,
sizeof
(
u64buf
),
"%l
u
"
,
pMsg
->
lastLogIndex
);
snprintf
(
u64buf
,
sizeof
(
u64buf
),
"%l
d
"
,
pMsg
->
lastLogIndex
);
cJSON_AddStringToObject
(
pRoot
,
"lastLogIndex"
,
u64buf
);
snprintf
(
u64buf
,
sizeof
(
u64buf
),
"%lu"
,
pMsg
->
lastLogTerm
);
cJSON_AddStringToObject
(
pRoot
,
"lastLogTerm"
,
u64buf
);
...
...
@@ -1127,14 +1127,14 @@ cJSON* syncAppendEntries2Json(const SyncAppendEntries* pMsg) {
snprintf
(
u64buf
,
sizeof
(
u64buf
),
"%lu"
,
pMsg
->
term
);
cJSON_AddStringToObject
(
pRoot
,
"term"
,
u64buf
);
snprintf
(
u64buf
,
sizeof
(
u64buf
),
"%l
u
"
,
pMsg
->
prevLogIndex
);
cJSON_AddStringToObject
(
pRoot
,
"pre
_log_i
ndex"
,
u64buf
);
snprintf
(
u64buf
,
sizeof
(
u64buf
),
"%l
d
"
,
pMsg
->
prevLogIndex
);
cJSON_AddStringToObject
(
pRoot
,
"pre
vLogI
ndex"
,
u64buf
);
snprintf
(
u64buf
,
sizeof
(
u64buf
),
"%lu"
,
pMsg
->
prevLogTerm
);
cJSON_AddStringToObject
(
pRoot
,
"pre_log_term"
,
u64buf
);
snprintf
(
u64buf
,
sizeof
(
u64buf
),
"%l
u
"
,
pMsg
->
commitIndex
);
cJSON_AddStringToObject
(
pRoot
,
"commit
_i
ndex"
,
u64buf
);
snprintf
(
u64buf
,
sizeof
(
u64buf
),
"%l
d
"
,
pMsg
->
commitIndex
);
cJSON_AddStringToObject
(
pRoot
,
"commit
I
ndex"
,
u64buf
);
cJSON_AddNumberToObject
(
pRoot
,
"dataLen"
,
pMsg
->
dataLen
);
char
*
s
;
...
...
@@ -1288,7 +1288,7 @@ cJSON* syncAppendEntriesReply2Json(const SyncAppendEntriesReply* pMsg) {
snprintf
(
u64buf
,
sizeof
(
u64buf
),
"%lu"
,
pMsg
->
term
);
cJSON_AddStringToObject
(
pRoot
,
"term"
,
u64buf
);
cJSON_AddNumberToObject
(
pRoot
,
"success"
,
pMsg
->
success
);
snprintf
(
u64buf
,
sizeof
(
u64buf
),
"%l
u
"
,
pMsg
->
matchIndex
);
snprintf
(
u64buf
,
sizeof
(
u64buf
),
"%l
d
"
,
pMsg
->
matchIndex
);
cJSON_AddStringToObject
(
pRoot
,
"matchIndex"
,
u64buf
);
}
...
...
source/libs/sync/test/CMakeLists.txt
浏览文件 @
559f570e
...
...
@@ -33,6 +33,7 @@ add_executable(syncElectTest3 "")
add_executable
(
syncEncodeTest
""
)
add_executable
(
syncWriteTest
""
)
add_executable
(
syncReplicateTest
""
)
add_executable
(
syncReplicateTest2
""
)
add_executable
(
syncRefTest
""
)
...
...
@@ -176,6 +177,10 @@ target_sources(syncReplicateTest
PRIVATE
"syncReplicateTest.cpp"
)
target_sources
(
syncReplicateTest2
PRIVATE
"syncReplicateTest2.cpp"
)
target_sources
(
syncRefTest
PRIVATE
"syncRefTest.cpp"
...
...
@@ -357,6 +362,11 @@ target_include_directories(syncReplicateTest
"
${
CMAKE_SOURCE_DIR
}
/include/libs/sync"
"
${
CMAKE_CURRENT_SOURCE_DIR
}
/../inc"
)
target_include_directories
(
syncReplicateTest2
PUBLIC
"
${
CMAKE_SOURCE_DIR
}
/include/libs/sync"
"
${
CMAKE_CURRENT_SOURCE_DIR
}
/../inc"
)
target_include_directories
(
syncRefTest
PUBLIC
"
${
CMAKE_SOURCE_DIR
}
/include/libs/sync"
...
...
@@ -504,6 +514,10 @@ target_link_libraries(syncReplicateTest
sync
gtest_main
)
target_link_libraries
(
syncReplicateTest2
sync
gtest_main
)
target_link_libraries
(
syncRefTest
sync
gtest_main
...
...
source/libs/sync/test/syncReplicateTest2.cpp
0 → 100644
浏览文件 @
559f570e
#include <gtest/gtest.h>
#include <stdio.h>
#include "syncEnv.h"
#include "syncIO.h"
#include "syncInt.h"
#include "syncMessage.h"
#include "syncRaftEntry.h"
#include "syncRaftLog.h"
#include "syncRaftStore.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"
);
}
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
;
SWal
*
pWal
;
void
CommitCb
(
struct
SSyncFSM
*
pFsm
,
const
SRpcMsg
*
pMsg
,
SyncIndex
index
,
bool
isWeak
,
int32_t
code
)
{
char
logBuf
[
256
];
snprintf
(
logBuf
,
sizeof
(
logBuf
),
"==callback== ==CommitCb== pFsm:%p, index:%ld, isWeak:%d, code:%d
\n
"
,
pFsm
,
index
,
isWeak
,
code
);
syncRpcMsgPrint2
(
logBuf
,
(
SRpcMsg
*
)
pMsg
);
}
void
PreCommitCb
(
struct
SSyncFSM
*
pFsm
,
const
SRpcMsg
*
pMsg
,
SyncIndex
index
,
bool
isWeak
,
int32_t
code
)
{
char
logBuf
[
256
];
snprintf
(
logBuf
,
sizeof
(
logBuf
),
"==callback== ==PreCommitCb== pFsm:%p, index:%ld, isWeak:%d, code:%d
\n
"
,
pFsm
,
index
,
isWeak
,
code
);
syncRpcMsgPrint2
(
logBuf
,
(
SRpcMsg
*
)
pMsg
);
}
void
RollBackCb
(
struct
SSyncFSM
*
pFsm
,
const
SRpcMsg
*
pMsg
,
SyncIndex
index
,
bool
isWeak
,
int32_t
code
)
{
char
logBuf
[
256
];
snprintf
(
logBuf
,
sizeof
(
logBuf
),
"==callback== ==RollBackCb== pFsm:%p, index:%ld, isWeak:%d, code:%d
\n
"
,
pFsm
,
index
,
isWeak
,
code
);
syncRpcMsgPrint2
(
logBuf
,
(
SRpcMsg
*
)
pMsg
);
}
void
initFsm
()
{
pFsm
=
(
SSyncFSM
*
)
malloc
(
sizeof
(
SSyncFSM
));
pFsm
->
FpCommitCb
=
CommitCb
;
pFsm
->
FpPreCommitCb
=
PreCommitCb
;
pFsm
->
FpRollBackCb
=
RollBackCb
;
}
int64_t
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
),
"./replicate2_test_%d"
,
myIndex
);
int
code
=
walInit
();
assert
(
code
==
0
);
SWalCfg
walCfg
;
memset
(
&
walCfg
,
0
,
sizeof
(
SWalCfg
));
walCfg
.
vgId
=
syncInfo
.
vgId
;
walCfg
.
fsyncPeriod
=
1000
;
walCfg
.
retentionPeriod
=
1000
;
walCfg
.
rollPeriod
=
1000
;
walCfg
.
retentionSize
=
1000
;
walCfg
.
segSize
=
1000
;
walCfg
.
level
=
TAOS_WAL_FSYNC
;
char
tmpdir
[
128
];
snprintf
(
tmpdir
,
sizeof
(
tmpdir
),
"./replicate2_test_wal_%d"
,
myIndex
);
pWal
=
walOpen
(
tmpdir
,
&
walCfg
);
assert
(
pWal
!=
NULL
);
syncInfo
.
pWal
=
pWal
;
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);
}
int64_t
rid
=
syncStart
(
&
syncInfo
);
assert
(
rid
>
0
);
SSyncNode
*
pSyncNode
=
(
SSyncNode
*
)
syncNodeAcquire
(
rid
);
assert
(
pSyncNode
!=
NULL
);
// pSyncNode->hbBaseLine = 500;
// pSyncNode->electBaseLine = 1500;
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
->
FpOnSyncTimeout
=
pSyncNode
->
FpOnTimeout
;
gSyncIO
->
FpOnSyncClientRequest
=
pSyncNode
->
FpOnClientRequest
;
gSyncIO
->
pSyncNode
=
pSyncNode
;
syncNodeRelease
(
pSyncNode
);
return
rid
;
}
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
);
}
}
SRpcMsg
*
step0
(
int
i
)
{
SRpcMsg
*
pMsg
=
(
SRpcMsg
*
)
malloc
(
sizeof
(
SRpcMsg
));
memset
(
pMsg
,
0
,
sizeof
(
SRpcMsg
));
pMsg
->
msgType
=
9999
;
pMsg
->
contLen
=
128
;
pMsg
->
pCont
=
malloc
(
pMsg
->
contLen
);
snprintf
((
char
*
)(
pMsg
->
pCont
),
pMsg
->
contLen
,
"value-%u-%d"
,
ports
[
myIndex
],
i
);
return
pMsg
;
}
SyncClientRequest
*
step1
(
const
SRpcMsg
*
pMsg
)
{
SyncClientRequest
*
pRetMsg
=
syncClientRequestBuild2
(
pMsg
,
123
,
true
);
return
pRetMsg
;
}
int
main
(
int
argc
,
char
**
argv
)
{
// taosInitLog((char *)"syncTest.log", 100000, 10);
tsAsyncLog
=
0
;
sDebugFlag
=
143
+
64
;
void
logTest
();
myIndex
=
0
;
if
(
argc
>=
2
)
{
myIndex
=
atoi
(
argv
[
1
]);
}
int32_t
ret
=
syncIOStart
((
char
*
)
"127.0.0.1"
,
ports
[
myIndex
]);
assert
(
ret
==
0
);
initFsm
();
ret
=
syncInit
();
assert
(
ret
==
0
);
int64_t
rid
=
syncNodeInit
();
assert
(
rid
>
0
);
SSyncNode
*
pSyncNode
=
(
SSyncNode
*
)
syncNodeAcquire
(
rid
);
assert
(
pSyncNode
!=
NULL
);
syncNodePrint2
((
char
*
)
""
,
pSyncNode
);
initRaftId
(
pSyncNode
);
for
(
int
i
=
0
;
i
<
30
;
++
i
)
{
// step0
SRpcMsg
*
pMsg0
=
step0
(
i
);
syncRpcMsgPrint2
((
char
*
)
"==step0=="
,
pMsg0
);
syncPropose
(
rid
,
pMsg0
,
true
);
taosMsleep
(
1000
);
free
(
pMsg0
);
sTrace
(
"syncPropose sleep, state: %d, %s, term:%lu electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, "
"electTimerMS:%d"
,
pSyncNode
->
state
,
syncUtilState2String
(
pSyncNode
->
state
),
pSyncNode
->
pRaftStore
->
currentTerm
,
pSyncNode
->
electTimerLogicClock
,
pSyncNode
->
electTimerLogicClockUser
,
pSyncNode
->
electTimerMS
);
}
while
(
1
)
{
sTrace
(
"replicate sleep, state: %d, %s, term:%lu electTimerLogicClock:%lu, electTimerLogicClockUser:%lu, "
"electTimerMS:%d"
,
pSyncNode
->
state
,
syncUtilState2String
(
pSyncNode
->
state
),
pSyncNode
->
pRaftStore
->
currentTerm
,
pSyncNode
->
electTimerLogicClock
,
pSyncNode
->
electTimerLogicClockUser
,
pSyncNode
->
electTimerMS
);
taosMsleep
(
1000
);
}
return
0
;
}
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录