Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
9bc8307d
T
TDengine
项目概览
taosdata
/
TDengine
11 个月 前同步成功
通知
1179
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,发现更多精彩内容 >>
未验证
提交
9bc8307d
编写于
5月 16, 2022
作者:
L
Li Minghao
提交者:
GitHub
5月 16, 2022
浏览文件
操作
浏览文件
下载
差异文件
Merge pull request #12530 from taosdata/feature/3.0_mhli
enh(sync): raft config change
上级
f199ce48
956ab45a
变更
12
隐藏空白更改
内联
并排
Showing
12 changed file
with
372 addition
and
50 deletion
+372
-50
include/libs/sync/sync.h
include/libs/sync/sync.h
+1
-0
source/libs/sync/inc/syncIndexMgr.h
source/libs/sync/inc/syncIndexMgr.h
+1
-0
source/libs/sync/inc/syncInt.h
source/libs/sync/inc/syncInt.h
+2
-1
source/libs/sync/inc/syncUtil.h
source/libs/sync/inc/syncUtil.h
+0
-1
source/libs/sync/src/syncAppendEntries.c
source/libs/sync/src/syncAppendEntries.c
+11
-6
source/libs/sync/src/syncCommit.c
source/libs/sync/src/syncCommit.c
+13
-8
source/libs/sync/src/syncIO.c
source/libs/sync/src/syncIO.c
+2
-0
source/libs/sync/src/syncIndexMgr.c
source/libs/sync/src/syncIndexMgr.c
+7
-0
source/libs/sync/src/syncMain.c
source/libs/sync/src/syncMain.c
+31
-6
source/libs/sync/src/syncRaftLog.c
source/libs/sync/src/syncRaftLog.c
+31
-28
source/libs/sync/test/CMakeLists.txt
source/libs/sync/test/CMakeLists.txt
+14
-0
source/libs/sync/test/syncConfigChangeTest.cpp
source/libs/sync/test/syncConfigChangeTest.cpp
+259
-0
未找到文件。
include/libs/sync/sync.h
浏览文件 @
9bc8307d
...
...
@@ -144,6 +144,7 @@ int32_t syncInit();
void
syncCleanUp
();
int64_t
syncOpen
(
const
SSyncInfo
*
pSyncInfo
);
void
syncStart
(
int64_t
rid
);
void
syncStartStandBy
(
int64_t
rid
);
void
syncStop
(
int64_t
rid
);
int32_t
syncReconfig
(
int64_t
rid
,
const
SSyncCfg
*
pSyncCfg
);
ESyncState
syncGetMyRole
(
int64_t
rid
);
...
...
source/libs/sync/inc/syncIndexMgr.h
浏览文件 @
9bc8307d
...
...
@@ -35,6 +35,7 @@ typedef struct SSyncIndexMgr {
}
SSyncIndexMgr
;
SSyncIndexMgr
*
syncIndexMgrCreate
(
SSyncNode
*
pSyncNode
);
void
syncIndexMgrUpdate
(
SSyncIndexMgr
*
pSyncIndexMgr
,
SSyncNode
*
pSyncNode
);
void
syncIndexMgrDestroy
(
SSyncIndexMgr
*
pSyncIndexMgr
);
void
syncIndexMgrClear
(
SSyncIndexMgr
*
pSyncIndexMgr
);
void
syncIndexMgrSetIndex
(
SSyncIndexMgr
*
pSyncIndexMgr
,
const
SRaftId
*
pRaftId
,
SyncIndex
index
);
...
...
source/libs/sync/inc/syncInt.h
浏览文件 @
9bc8307d
...
...
@@ -247,6 +247,7 @@ typedef struct SSyncNode {
// open/close --------------
SSyncNode
*
syncNodeOpen
(
const
SSyncInfo
*
pSyncInfo
);
void
syncNodeStart
(
SSyncNode
*
pSyncNode
);
void
syncNodeStartStandBy
(
SSyncNode
*
pSyncNode
);
void
syncNodeClose
(
SSyncNode
*
pSyncNode
);
// ping --------------
...
...
@@ -271,7 +272,7 @@ int32_t syncNodeSendMsgByInfo(const SNodeInfo* nodeInfo, SSyncNode* pSyncNode, S
cJSON
*
syncNode2Json
(
const
SSyncNode
*
pSyncNode
);
char
*
syncNode2Str
(
const
SSyncNode
*
pSyncNode
);
char
*
syncNode2SimpleStr
(
const
SSyncNode
*
pSyncNode
);
void
syncNodeUpdateConfig
(
SSyncNode
*
pSyncNode
,
SSyncCfg
*
newConfig
);
void
syncNodeUpdateConfig
(
SSyncNode
*
pSyncNode
,
SSyncCfg
*
newConfig
);
SSyncNode
*
syncNodeAcquire
(
int64_t
rid
);
void
syncNodeRelease
(
SSyncNode
*
pNode
);
...
...
source/libs/sync/inc/syncUtil.h
浏览文件 @
9bc8307d
...
...
@@ -62,7 +62,6 @@ bool syncUtilUserPreCommit(tmsg_t msgType);
bool
syncUtilUserCommit
(
tmsg_t
msgType
);
bool
syncUtilUserRollback
(
tmsg_t
msgType
);
#ifdef __cplusplus
}
#endif
...
...
source/libs/sync/src/syncAppendEntries.c
浏览文件 @
9bc8307d
...
...
@@ -15,11 +15,11 @@
#include "syncAppendEntries.h"
#include "syncInt.h"
#include "syncRaftCfg.h"
#include "syncRaftLog.h"
#include "syncRaftStore.h"
#include "syncUtil.h"
#include "syncVoteMgr.h"
#include "syncRaftCfg.h"
// TLA+ Spec
// HandleAppendEntriesRequest(i, j, m) ==
...
...
@@ -200,7 +200,7 @@ int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) {
SSyncRaftEntry
*
pRollBackEntry
=
logStoreGetEntry
(
ths
->
pLogStore
,
index
);
assert
(
pRollBackEntry
!=
NULL
);
//if (pRollBackEntry->msgType != TDMT_VND_SYNC_NOOP) {
//
if (pRollBackEntry->msgType != TDMT_VND_SYNC_NOOP) {
if
(
syncUtilUserRollback
(
pRollBackEntry
->
msgType
))
{
SRpcMsg
rpcMsg
;
syncEntry2OriginalRpc
(
pRollBackEntry
,
&
rpcMsg
);
...
...
@@ -229,7 +229,7 @@ int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) {
SRpcMsg
rpcMsg
;
syncEntry2OriginalRpc
(
pAppendEntry
,
&
rpcMsg
);
if
(
ths
->
pFsm
!=
NULL
)
{
//if (ths->pFsm->FpPreCommitCb != NULL && pAppendEntry->originalRpcType != TDMT_VND_SYNC_NOOP) {
//
if (ths->pFsm->FpPreCommitCb != NULL && pAppendEntry->originalRpcType != TDMT_VND_SYNC_NOOP) {
if
(
ths
->
pFsm
->
FpPreCommitCb
!=
NULL
&&
syncUtilUserPreCommit
(
pAppendEntry
->
originalRpcType
))
{
SFsmCbMeta
cbMeta
;
cbMeta
.
index
=
pAppendEntry
->
index
;
...
...
@@ -261,7 +261,7 @@ int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) {
SRpcMsg
rpcMsg
;
syncEntry2OriginalRpc
(
pAppendEntry
,
&
rpcMsg
);
if
(
ths
->
pFsm
!=
NULL
)
{
//if (ths->pFsm->FpPreCommitCb != NULL && pAppendEntry->originalRpcType != TDMT_VND_SYNC_NOOP) {
//
if (ths->pFsm->FpPreCommitCb != NULL && pAppendEntry->originalRpcType != TDMT_VND_SYNC_NOOP) {
if
(
ths
->
pFsm
->
FpPreCommitCb
!=
NULL
&&
syncUtilUserPreCommit
(
pAppendEntry
->
originalRpcType
))
{
SFsmCbMeta
cbMeta
;
cbMeta
.
index
=
pAppendEntry
->
index
;
...
...
@@ -324,7 +324,7 @@ int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) {
SRpcMsg
rpcMsg
;
syncEntry2OriginalRpc
(
pEntry
,
&
rpcMsg
);
//if (ths->pFsm->FpCommitCb != NULL && pEntry->originalRpcType != TDMT_VND_SYNC_NOOP) {
//
if (ths->pFsm->FpCommitCb != NULL && pEntry->originalRpcType != TDMT_VND_SYNC_NOOP) {
if
(
ths
->
pFsm
->
FpCommitCb
!=
NULL
&&
syncUtilUserCommit
(
pEntry
->
originalRpcType
))
{
SFsmCbMeta
cbMeta
;
cbMeta
.
index
=
pEntry
->
index
;
...
...
@@ -338,10 +338,15 @@ int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) {
// config change
if
(
pEntry
->
originalRpcType
==
TDMT_VND_SYNC_CONFIG_CHANGE
)
{
SSyncCfg
newSyncCfg
;
int32_t
ret
=
syncCfgFromStr
(
rpcMsg
.
pCont
,
&
newSyncCfg
);
int32_t
ret
=
syncCfgFromStr
(
rpcMsg
.
pCont
,
&
newSyncCfg
);
ASSERT
(
ret
==
0
);
syncNodeUpdateConfig
(
ths
,
&
newSyncCfg
);
if
(
ths
->
state
==
TAOS_SYNC_STATE_LEADER
)
{
syncNodeBecomeLeader
(
ths
);
}
else
{
syncNodeBecomeFollower
(
ths
);
}
}
rpcFreeCont
(
rpcMsg
.
pCont
);
...
...
source/libs/sync/src/syncCommit.c
浏览文件 @
9bc8307d
...
...
@@ -16,10 +16,10 @@
#include "syncCommit.h"
#include "syncIndexMgr.h"
#include "syncInt.h"
#include "syncRaftCfg.h"
#include "syncRaftLog.h"
#include "syncRaftStore.h"
#include "syncUtil.h"
#include "syncRaftCfg.h"
// \* Leader i advances its commitIndex.
// \* This is done as a separate step from handling AppendEntries responses,
...
...
@@ -102,7 +102,7 @@ void syncMaybeAdvanceCommitIndex(SSyncNode* pSyncNode) {
SRpcMsg
rpcMsg
;
syncEntry2OriginalRpc
(
pEntry
,
&
rpcMsg
);
//if (pSyncNode->pFsm->FpCommitCb != NULL && pEntry->originalRpcType != TDMT_VND_SYNC_NOOP) {
//
if (pSyncNode->pFsm->FpCommitCb != NULL && pEntry->originalRpcType != TDMT_VND_SYNC_NOOP) {
if
(
pSyncNode
->
pFsm
->
FpCommitCb
!=
NULL
&&
syncUtilUserCommit
(
pEntry
->
originalRpcType
))
{
SFsmCbMeta
cbMeta
;
cbMeta
.
index
=
pEntry
->
index
;
...
...
@@ -114,12 +114,17 @@ void syncMaybeAdvanceCommitIndex(SSyncNode* pSyncNode) {
}
// config change
if
(
pEntry
->
originalRpcType
==
TDMT_VND_SYNC_CONFIG_CHANGE
)
{
SSyncCfg
newSyncCfg
;
int32_t
ret
=
syncCfgFromStr
(
rpcMsg
.
pCont
,
&
newSyncCfg
);
ASSERT
(
ret
==
0
);
syncNodeUpdateConfig
(
pSyncNode
,
&
newSyncCfg
);
if
(
pEntry
->
originalRpcType
==
TDMT_VND_SYNC_CONFIG_CHANGE
)
{
SSyncCfg
newSyncCfg
;
int32_t
ret
=
syncCfgFromStr
(
rpcMsg
.
pCont
,
&
newSyncCfg
);
ASSERT
(
ret
==
0
);
syncNodeUpdateConfig
(
pSyncNode
,
&
newSyncCfg
);
if
(
pSyncNode
->
state
==
TAOS_SYNC_STATE_LEADER
)
{
syncNodeBecomeLeader
(
pSyncNode
);
}
else
{
syncNodeBecomeFollower
(
pSyncNode
);
}
}
rpcFreeCont
(
rpcMsg
.
pCont
);
...
...
source/libs/sync/src/syncIO.c
浏览文件 @
9bc8307d
...
...
@@ -15,6 +15,7 @@
#include "syncIO.h"
#include <tdatablock.h>
#include "os.h"
#include "syncMessage.h"
#include "syncUtil.h"
#include "tglobal.h"
...
...
@@ -198,6 +199,7 @@ static int32_t syncIOStartInternal(SSyncIO *io) {
{
SRpcInit
rpcInit
;
memset
(
&
rpcInit
,
0
,
sizeof
(
rpcInit
));
snprintf
(
rpcInit
.
localFqdn
,
sizeof
(
rpcInit
.
localFqdn
),
"%s"
,
"127.0.0.1"
);
rpcInit
.
localPort
=
io
->
myAddr
.
eps
[
0
].
port
;
rpcInit
.
label
=
"SYNC-IO-SERVER"
;
rpcInit
.
numOfThreads
=
1
;
...
...
source/libs/sync/src/syncIndexMgr.c
浏览文件 @
9bc8307d
...
...
@@ -31,6 +31,13 @@ SSyncIndexMgr *syncIndexMgrCreate(SSyncNode *pSyncNode) {
return
pSyncIndexMgr
;
}
void
syncIndexMgrUpdate
(
SSyncIndexMgr
*
pSyncIndexMgr
,
SSyncNode
*
pSyncNode
)
{
pSyncIndexMgr
->
replicas
=
&
(
pSyncNode
->
replicasId
);
pSyncIndexMgr
->
replicaNum
=
pSyncNode
->
replicaNum
;
pSyncIndexMgr
->
pSyncNode
=
pSyncNode
;
syncIndexMgrClear
(
pSyncIndexMgr
);
}
void
syncIndexMgrDestroy
(
SSyncIndexMgr
*
pSyncIndexMgr
)
{
if
(
pSyncIndexMgr
!=
NULL
)
{
taosMemoryFree
(
pSyncIndexMgr
);
...
...
source/libs/sync/src/syncMain.c
浏览文件 @
9bc8307d
...
...
@@ -103,6 +103,16 @@ void syncStart(int64_t rid) {
taosReleaseRef
(
tsNodeRefId
,
pSyncNode
->
rid
);
}
void
syncStartStandBy
(
int64_t
rid
)
{
SSyncNode
*
pSyncNode
=
(
SSyncNode
*
)
taosAcquireRef
(
tsNodeRefId
,
rid
);
if
(
pSyncNode
==
NULL
)
{
return
;
}
syncNodeStartStandBy
(
pSyncNode
);
taosReleaseRef
(
tsNodeRefId
,
pSyncNode
->
rid
);
}
void
syncStop
(
int64_t
rid
)
{
SSyncNode
*
pSyncNode
=
(
SSyncNode
*
)
taosAcquireRef
(
tsNodeRefId
,
rid
);
if
(
pSyncNode
==
NULL
)
{
...
...
@@ -116,7 +126,7 @@ void syncStop(int64_t rid) {
int32_t
syncReconfig
(
int64_t
rid
,
const
SSyncCfg
*
pSyncCfg
)
{
int32_t
ret
=
0
;
char
*
configChange
=
syncCfg2Str
((
SSyncCfg
*
)
pSyncCfg
);
char
*
configChange
=
syncCfg2Str
((
SSyncCfg
*
)
pSyncCfg
);
SRpcMsg
rpcMsg
=
{
0
};
rpcMsg
.
msgType
=
TDMT_VND_SYNC_CONFIG_CHANGE
;
rpcMsg
.
noResp
=
1
;
...
...
@@ -188,10 +198,9 @@ void syncGetEpSet(int64_t rid, SEpSet* pEpSet) {
(
pEpSet
->
numOfEps
)
++
;
sInfo
(
"syncGetEpSet index:%d %s:%d"
,
i
,
pEpSet
->
eps
[
i
].
fqdn
,
pEpSet
->
eps
[
i
].
port
);
}
pEpSet
->
inUse
=
pSyncNode
->
pRaftCfg
->
cfg
.
myIndex
;
sInfo
(
"syncGetEpSet pEpSet->inUse:%d "
,
pEpSet
->
inUse
);
taosReleaseRef
(
tsNodeRefId
,
pSyncNode
->
rid
);
...
...
@@ -524,6 +533,17 @@ void syncNodeStart(SSyncNode* pSyncNode) {
assert
(
ret
==
0
);
}
void
syncNodeStartStandBy
(
SSyncNode
*
pSyncNode
)
{
// state change
pSyncNode
->
state
=
TAOS_SYNC_STATE_FOLLOWER
;
syncNodeStopHeartbeatTimer
(
pSyncNode
);
// reset elect timer, long enough
int32_t
electMS
=
TIMER_MAX_MS
;
int32_t
ret
=
syncNodeRestartElectTimer
(
pSyncNode
,
electMS
);
ASSERT
(
ret
==
0
);
}
void
syncNodeClose
(
SSyncNode
*
pSyncNode
)
{
int32_t
ret
;
assert
(
pSyncNode
!=
NULL
);
...
...
@@ -858,7 +878,7 @@ char* syncNode2SimpleStr(const SSyncNode* pSyncNode) {
return
s
;
}
void
syncNodeUpdateConfig
(
SSyncNode
*
pSyncNode
,
SSyncCfg
*
newConfig
)
{
void
syncNodeUpdateConfig
(
SSyncNode
*
pSyncNode
,
SSyncCfg
*
newConfig
)
{
pSyncNode
->
pRaftCfg
->
cfg
=
*
newConfig
;
int32_t
ret
=
raftCfgPersist
(
pSyncNode
->
pRaftCfg
);
ASSERT
(
ret
==
0
);
...
...
@@ -885,6 +905,11 @@ void syncNodeUpdateConfig(SSyncNode* pSyncNode, SSyncCfg *newConfig) {
for
(
int
i
=
0
;
i
<
pSyncNode
->
pRaftCfg
->
cfg
.
replicaNum
;
++
i
)
{
syncUtilnodeInfo2raftId
(
&
pSyncNode
->
pRaftCfg
->
cfg
.
nodeInfo
[
i
],
pSyncNode
->
vgId
,
&
pSyncNode
->
replicasId
[
i
]);
}
syncIndexMgrUpdate
(
pSyncNode
->
pNextIndex
,
pSyncNode
);
syncIndexMgrUpdate
(
pSyncNode
->
pMatchIndex
,
pSyncNode
);
syncNodeLog2
(
"==syncNodeUpdateConfig=="
,
pSyncNode
);
}
SSyncNode
*
syncNodeAcquire
(
int64_t
rid
)
{
...
...
@@ -1245,7 +1270,7 @@ int32_t syncNodeOnClientRequestCb(SSyncNode* ths, SyncClientRequest* pMsg) {
syncEntry2OriginalRpc
(
pEntry
,
&
rpcMsg
);
if
(
ths
->
pFsm
!=
NULL
)
{
//if (ths->pFsm->FpPreCommitCb != NULL && pEntry->originalRpcType != TDMT_VND_SYNC_NOOP) {
//
if (ths->pFsm->FpPreCommitCb != NULL && pEntry->originalRpcType != TDMT_VND_SYNC_NOOP) {
if
(
ths
->
pFsm
->
FpPreCommitCb
!=
NULL
&&
syncUtilUserPreCommit
(
pEntry
->
originalRpcType
))
{
SFsmCbMeta
cbMeta
;
cbMeta
.
index
=
pEntry
->
index
;
...
...
@@ -1267,7 +1292,7 @@ int32_t syncNodeOnClientRequestCb(SSyncNode* ths, SyncClientRequest* pMsg) {
syncEntry2OriginalRpc
(
pEntry
,
&
rpcMsg
);
if
(
ths
->
pFsm
!=
NULL
)
{
//if (ths->pFsm->FpPreCommitCb != NULL && pEntry->originalRpcType != TDMT_VND_SYNC_NOOP) {
//
if (ths->pFsm->FpPreCommitCb != NULL && pEntry->originalRpcType != TDMT_VND_SYNC_NOOP) {
if
(
ths
->
pFsm
->
FpPreCommitCb
!=
NULL
&&
syncUtilUserPreCommit
(
pEntry
->
originalRpcType
))
{
SFsmCbMeta
cbMeta
;
cbMeta
.
index
=
pEntry
->
index
;
...
...
source/libs/sync/src/syncRaftLog.c
浏览文件 @
9bc8307d
...
...
@@ -58,14 +58,15 @@ int32_t logStoreAppendEntry(SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry) {
syncMeta
.
term
=
pEntry
->
term
;
code
=
walWriteWithSyncInfo
(
pWal
,
pEntry
->
index
,
pEntry
->
originalRpcType
,
syncMeta
,
pEntry
->
data
,
pEntry
->
dataLen
);
if
(
code
!=
0
)
{
int32_t
err
=
terrno
;
const
char
*
errStr
=
tstrerror
(
err
);
int32_t
linuxErr
=
errno
;
const
char
*
linuxErrMsg
=
strerror
(
errno
);
sError
(
"walWriteWithSyncInfo error, err:%d %X, msg:%s, linuxErr:%d, linuxErrMsg:%s"
,
err
,
err
,
errStr
,
linuxErr
,
linuxErrMsg
);
int32_t
err
=
terrno
;
const
char
*
errStr
=
tstrerror
(
err
);
int32_t
linuxErr
=
errno
;
const
char
*
linuxErrMsg
=
strerror
(
errno
);
sError
(
"walWriteWithSyncInfo error, err:%d %X, msg:%s, linuxErr:%d, linuxErrMsg:%s"
,
err
,
err
,
errStr
,
linuxErr
,
linuxErrMsg
);
ASSERT
(
0
);
}
//assert(code == 0);
}
//
assert(code == 0);
walFsync
(
pWal
,
true
);
return
code
;
...
...
@@ -77,16 +78,17 @@ SSyncRaftEntry* logStoreGetEntry(SSyncLogStore* pLogStore, SyncIndex index) {
if
(
index
>=
SYNC_INDEX_BEGIN
&&
index
<=
logStoreLastIndex
(
pLogStore
))
{
SWalReadHandle
*
pWalHandle
=
walOpenReadHandle
(
pWal
);
int32_t
code
=
walReadWithHandle
(
pWalHandle
,
index
);
int32_t
code
=
walReadWithHandle
(
pWalHandle
,
index
);
if
(
code
!=
0
)
{
int32_t
err
=
terrno
;
const
char
*
errStr
=
tstrerror
(
err
);
int32_t
linuxErr
=
errno
;
const
char
*
linuxErrMsg
=
strerror
(
errno
);
sError
(
"walReadWithHandle error, err:%d %X, msg:%s, linuxErr:%d, linuxErrMsg:%s"
,
err
,
err
,
errStr
,
linuxErr
,
linuxErrMsg
);
int32_t
err
=
terrno
;
const
char
*
errStr
=
tstrerror
(
err
);
int32_t
linuxErr
=
errno
;
const
char
*
linuxErrMsg
=
strerror
(
errno
);
sError
(
"walReadWithHandle error, err:%d %X, msg:%s, linuxErr:%d, linuxErrMsg:%s"
,
err
,
err
,
errStr
,
linuxErr
,
linuxErrMsg
);
ASSERT
(
0
);
}
//assert(walReadWithHandle(pWalHandle, index) == 0);
}
//
assert(walReadWithHandle(pWalHandle, index) == 0);
SSyncRaftEntry
*
pEntry
=
syncEntryBuild
(
pWalHandle
->
pHead
->
head
.
bodyLen
);
assert
(
pEntry
!=
NULL
);
...
...
@@ -112,16 +114,17 @@ SSyncRaftEntry* logStoreGetEntry(SSyncLogStore* pLogStore, SyncIndex index) {
int32_t
logStoreTruncate
(
SSyncLogStore
*
pLogStore
,
SyncIndex
fromIndex
)
{
SSyncLogStoreData
*
pData
=
pLogStore
->
data
;
SWal
*
pWal
=
pData
->
pWal
;
//assert(walRollback(pWal, fromIndex) == 0);
//
assert(walRollback(pWal, fromIndex) == 0);
int32_t
code
=
walRollback
(
pWal
,
fromIndex
);
if
(
code
!=
0
)
{
int32_t
err
=
terrno
;
const
char
*
errStr
=
tstrerror
(
err
);
int32_t
linuxErr
=
errno
;
const
char
*
linuxErrMsg
=
strerror
(
errno
);
sError
(
"walRollback error, err:%d %X, msg:%s, linuxErr:%d, linuxErrMsg:%s"
,
err
,
err
,
errStr
,
linuxErr
,
linuxErrMsg
);
int32_t
err
=
terrno
;
const
char
*
errStr
=
tstrerror
(
err
);
int32_t
linuxErr
=
errno
;
const
char
*
linuxErrMsg
=
strerror
(
errno
);
sError
(
"walRollback error, err:%d %X, msg:%s, linuxErr:%d, linuxErrMsg:%s"
,
err
,
err
,
errStr
,
linuxErr
,
linuxErrMsg
);
ASSERT
(
0
);
}
}
return
0
;
// to avoid compiler error
}
...
...
@@ -145,16 +148,16 @@ SyncTerm logStoreLastTerm(SSyncLogStore* pLogStore) {
int32_t
logStoreUpdateCommitIndex
(
SSyncLogStore
*
pLogStore
,
SyncIndex
index
)
{
SSyncLogStoreData
*
pData
=
pLogStore
->
data
;
SWal
*
pWal
=
pData
->
pWal
;
//assert(walCommit(pWal, index) == 0);
//
assert(walCommit(pWal, index) == 0);
int32_t
code
=
walCommit
(
pWal
,
index
);
if
(
code
!=
0
)
{
int32_t
err
=
terrno
;
const
char
*
errStr
=
tstrerror
(
err
);
int32_t
linuxErr
=
errno
;
const
char
*
linuxErrMsg
=
strerror
(
errno
);
int32_t
err
=
terrno
;
const
char
*
errStr
=
tstrerror
(
err
);
int32_t
linuxErr
=
errno
;
const
char
*
linuxErrMsg
=
strerror
(
errno
);
sError
(
"walCommit error, err:%d %X, msg:%s, linuxErr:%d, linuxErrMsg:%s"
,
err
,
err
,
errStr
,
linuxErr
,
linuxErrMsg
);
ASSERT
(
0
);
}
}
return
0
;
// to avoid compiler error
}
...
...
source/libs/sync/test/CMakeLists.txt
浏览文件 @
9bc8307d
...
...
@@ -37,6 +37,7 @@ add_executable(syncRaftCfgTest "")
add_executable
(
syncRespMgrTest
""
)
add_executable
(
syncSnapshotTest
""
)
add_executable
(
syncApplyMsgTest
""
)
add_executable
(
syncConfigChangeTest
""
)
target_sources
(
syncTest
...
...
@@ -195,6 +196,10 @@ target_sources(syncApplyMsgTest
PRIVATE
"syncApplyMsgTest.cpp"
)
target_sources
(
syncConfigChangeTest
PRIVATE
"syncConfigChangeTest.cpp"
)
target_include_directories
(
syncTest
...
...
@@ -392,6 +397,11 @@ target_include_directories(syncApplyMsgTest
"
${
TD_SOURCE_DIR
}
/include/libs/sync"
"
${
CMAKE_CURRENT_SOURCE_DIR
}
/../inc"
)
target_include_directories
(
syncConfigChangeTest
PUBLIC
"
${
TD_SOURCE_DIR
}
/include/libs/sync"
"
${
CMAKE_CURRENT_SOURCE_DIR
}
/../inc"
)
target_link_libraries
(
syncTest
...
...
@@ -550,6 +560,10 @@ target_link_libraries(syncApplyMsgTest
sync
gtest_main
)
target_link_libraries
(
syncConfigChangeTest
sync
gtest_main
)
enable_testing
()
...
...
source/libs/sync/test/syncConfigChangeTest.cpp
0 → 100644
浏览文件 @
9bc8307d
#include <gtest/gtest.h>
#include <stdio.h>
#include "os.h"
#include "syncEnv.h"
#include "syncIO.h"
#include "syncInt.h"
#include "syncUtil.h"
#include "wal.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
gPorts
[]
=
{
7010
,
7110
,
7210
,
7310
,
7410
};
const
char
*
gDir
=
"./syncReplicateTest"
;
int32_t
gVgId
=
1234
;
SyncIndex
gSnapshotLastApplyIndex
;
void
init
()
{
int
code
=
walInit
();
assert
(
code
==
0
);
code
=
syncInit
();
assert
(
code
==
0
);
sprintf
(
tsTempDir
,
"%s"
,
"."
);
}
void
cleanup
()
{
walCleanUp
();
}
void
CommitCb
(
struct
SSyncFSM
*
pFsm
,
const
SRpcMsg
*
pMsg
,
SFsmCbMeta
cbMeta
)
{
SyncIndex
beginIndex
=
SYNC_INDEX_INVALID
;
if
(
pFsm
->
FpGetSnapshot
!=
NULL
)
{
SSnapshot
snapshot
;
pFsm
->
FpGetSnapshot
(
pFsm
,
&
snapshot
);
beginIndex
=
snapshot
.
lastApplyIndex
;
}
if
(
cbMeta
.
index
>
beginIndex
)
{
char
logBuf
[
256
];
snprintf
(
logBuf
,
sizeof
(
logBuf
),
"==callback== ==CommitCb== pFsm:%p, index:%ld, isWeak:%d, code:%d, state:%d %s
\n
"
,
pFsm
,
cbMeta
.
index
,
cbMeta
.
isWeak
,
cbMeta
.
code
,
cbMeta
.
state
,
syncUtilState2String
(
cbMeta
.
state
));
syncRpcMsgLog2
(
logBuf
,
(
SRpcMsg
*
)
pMsg
);
}
else
{
sTrace
(
"==callback== ==CommitCb== do not apply again %ld"
,
cbMeta
.
index
);
}
}
void
PreCommitCb
(
struct
SSyncFSM
*
pFsm
,
const
SRpcMsg
*
pMsg
,
SFsmCbMeta
cbMeta
)
{
char
logBuf
[
256
];
snprintf
(
logBuf
,
sizeof
(
logBuf
),
"==callback== ==PreCommitCb== pFsm:%p, index:%ld, isWeak:%d, code:%d, state:%d %s
\n
"
,
pFsm
,
cbMeta
.
index
,
cbMeta
.
isWeak
,
cbMeta
.
code
,
cbMeta
.
state
,
syncUtilState2String
(
cbMeta
.
state
));
syncRpcMsgLog2
(
logBuf
,
(
SRpcMsg
*
)
pMsg
);
}
void
RollBackCb
(
struct
SSyncFSM
*
pFsm
,
const
SRpcMsg
*
pMsg
,
SFsmCbMeta
cbMeta
)
{
char
logBuf
[
256
];
snprintf
(
logBuf
,
sizeof
(
logBuf
),
"==callback== ==RollBackCb== pFsm:%p, index:%ld, isWeak:%d, code:%d, state:%d %s
\n
"
,
pFsm
,
cbMeta
.
index
,
cbMeta
.
isWeak
,
cbMeta
.
code
,
cbMeta
.
state
,
syncUtilState2String
(
cbMeta
.
state
));
syncRpcMsgLog2
(
logBuf
,
(
SRpcMsg
*
)
pMsg
);
}
int32_t
GetSnapshotCb
(
struct
SSyncFSM
*
pFsm
,
SSnapshot
*
pSnapshot
)
{
pSnapshot
->
data
=
NULL
;
pSnapshot
->
lastApplyIndex
=
gSnapshotLastApplyIndex
;
pSnapshot
->
lastApplyTerm
=
100
;
return
0
;
}
SSyncFSM
*
createFsm
()
{
SSyncFSM
*
pFsm
=
(
SSyncFSM
*
)
taosMemoryMalloc
(
sizeof
(
SSyncFSM
));
pFsm
->
FpCommitCb
=
CommitCb
;
pFsm
->
FpPreCommitCb
=
PreCommitCb
;
pFsm
->
FpRollBackCb
=
RollBackCb
;
pFsm
->
FpGetSnapshot
=
GetSnapshotCb
;
return
pFsm
;
}
SWal
*
createWal
(
char
*
path
,
int32_t
vgId
)
{
SWalCfg
walCfg
;
memset
(
&
walCfg
,
0
,
sizeof
(
SWalCfg
));
walCfg
.
vgId
=
vgId
;
walCfg
.
fsyncPeriod
=
1000
;
walCfg
.
retentionPeriod
=
1000
;
walCfg
.
rollPeriod
=
1000
;
walCfg
.
retentionSize
=
1000
;
walCfg
.
segSize
=
1000
;
walCfg
.
level
=
TAOS_WAL_FSYNC
;
SWal
*
pWal
=
walOpen
(
path
,
&
walCfg
);
assert
(
pWal
!=
NULL
);
return
pWal
;
}
int64_t
createSyncNode
(
int32_t
replicaNum
,
int32_t
myIndex
,
int32_t
vgId
,
SWal
*
pWal
,
char
*
path
,
bool
isStandBy
)
{
SSyncInfo
syncInfo
;
syncInfo
.
vgId
=
vgId
;
syncInfo
.
rpcClient
=
gSyncIO
->
clientRpc
;
syncInfo
.
FpSendMsg
=
syncIOSendMsg
;
syncInfo
.
queue
=
gSyncIO
->
pMsgQ
;
syncInfo
.
FpEqMsg
=
syncIOEqMsg
;
syncInfo
.
pFsm
=
createFsm
();
snprintf
(
syncInfo
.
path
,
sizeof
(
syncInfo
.
path
),
"%s_sync_replica%d_index%d"
,
path
,
replicaNum
,
myIndex
);
syncInfo
.
pWal
=
pWal
;
SSyncCfg
*
pCfg
=
&
syncInfo
.
syncCfg
;
if
(
isStandBy
)
{
pCfg
->
myIndex
=
0
;
pCfg
->
replicaNum
=
1
;
pCfg
->
nodeInfo
[
0
].
nodePort
=
gPorts
[
myIndex
];
taosGetFqdn
(
pCfg
->
nodeInfo
[
0
].
nodeFqdn
);
}
else
{
pCfg
->
myIndex
=
myIndex
;
pCfg
->
replicaNum
=
replicaNum
;
for
(
int
i
=
0
;
i
<
replicaNum
;
++
i
)
{
pCfg
->
nodeInfo
[
i
].
nodePort
=
gPorts
[
i
];
taosGetFqdn
(
pCfg
->
nodeInfo
[
i
].
nodeFqdn
);
// snprintf(pCfg->nodeInfo[i].nodeFqdn, sizeof(pCfg->nodeInfo[i].nodeFqdn), "%s", "127.0.0.1");
}
}
int64_t
rid
=
syncOpen
(
&
syncInfo
);
assert
(
rid
>
0
);
SSyncNode
*
pSyncNode
=
(
SSyncNode
*
)
syncNodeAcquire
(
rid
);
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
->
FpOnSyncTimeout
=
pSyncNode
->
FpOnTimeout
;
gSyncIO
->
FpOnSyncClientRequest
=
pSyncNode
->
FpOnClientRequest
;
gSyncIO
->
pSyncNode
=
pSyncNode
;
syncNodeRelease
(
pSyncNode
);
return
rid
;
}
void
configChange
(
int64_t
rid
,
int32_t
replicaNum
,
int32_t
myIndex
)
{
SSyncCfg
syncCfg
;
syncCfg
.
myIndex
=
myIndex
;
syncCfg
.
replicaNum
=
replicaNum
;
for
(
int
i
=
0
;
i
<
replicaNum
;
++
i
)
{
syncCfg
.
nodeInfo
[
i
].
nodePort
=
gPorts
[
i
];
taosGetFqdn
(
syncCfg
.
nodeInfo
[
i
].
nodeFqdn
);
}
syncReconfig
(
rid
,
&
syncCfg
);
}
void
usage
(
char
*
exe
)
{
printf
(
"usage: %s replicaNum myIndex lastApplyIndex writeRecordNum isStandBy isConfigChange
\n
"
,
exe
);
}
SRpcMsg
*
createRpcMsg
(
int
i
,
int
count
,
int
myIndex
)
{
SRpcMsg
*
pMsg
=
(
SRpcMsg
*
)
taosMemoryMalloc
(
sizeof
(
SRpcMsg
));
memset
(
pMsg
,
0
,
sizeof
(
SRpcMsg
));
pMsg
->
msgType
=
9999
;
pMsg
->
contLen
=
256
;
pMsg
->
pCont
=
rpcMallocCont
(
pMsg
->
contLen
);
snprintf
((
char
*
)(
pMsg
->
pCont
),
pMsg
->
contLen
,
"value-myIndex:%u-%d-%d-%ld"
,
myIndex
,
i
,
count
,
taosGetTimestampMs
());
return
pMsg
;
}
int
main
(
int
argc
,
char
**
argv
)
{
tsAsyncLog
=
0
;
sDebugFlag
=
DEBUG_TRACE
+
DEBUG_SCREEN
+
DEBUG_FILE
;
if
(
argc
!=
7
)
{
usage
(
argv
[
0
]);
exit
(
-
1
);
}
int32_t
replicaNum
=
atoi
(
argv
[
1
]);
int32_t
myIndex
=
atoi
(
argv
[
2
]);
int32_t
lastApplyIndex
=
atoi
(
argv
[
3
]);
int32_t
writeRecordNum
=
atoi
(
argv
[
4
]);
bool
isStandBy
=
atoi
(
argv
[
5
]);
bool
isConfigChange
=
atoi
(
argv
[
6
]);
gSnapshotLastApplyIndex
=
lastApplyIndex
;
if
(
!
isStandBy
)
{
assert
(
replicaNum
>=
1
&&
replicaNum
<=
5
);
assert
(
myIndex
>=
0
&&
myIndex
<
replicaNum
);
assert
(
lastApplyIndex
>=
-
1
);
assert
(
writeRecordNum
>=
0
);
}
init
();
int32_t
ret
=
syncIOStart
((
char
*
)
"127.0.0.1"
,
gPorts
[
myIndex
]);
assert
(
ret
==
0
);
char
walPath
[
128
];
snprintf
(
walPath
,
sizeof
(
walPath
),
"%s_wal_replica%d_index%d"
,
gDir
,
replicaNum
,
myIndex
);
SWal
*
pWal
=
createWal
(
walPath
,
gVgId
);
int64_t
rid
=
createSyncNode
(
replicaNum
,
myIndex
,
gVgId
,
pWal
,
(
char
*
)
gDir
,
isStandBy
);
assert
(
rid
>
0
);
if
(
isStandBy
)
{
syncStartStandBy
(
rid
);
}
else
{
syncStart
(
rid
);
}
SSyncNode
*
pSyncNode
=
(
SSyncNode
*
)
syncNodeAcquire
(
rid
);
assert
(
pSyncNode
!=
NULL
);
if
(
isConfigChange
)
{
configChange
(
rid
,
3
,
myIndex
);
}
//---------------------------
int32_t
alreadySend
=
0
;
while
(
1
)
{
char
*
s
=
syncNode2SimpleStr
(
pSyncNode
);
if
(
alreadySend
<
writeRecordNum
)
{
SRpcMsg
*
pRpcMsg
=
createRpcMsg
(
alreadySend
,
writeRecordNum
,
myIndex
);
int32_t
ret
=
syncPropose
(
rid
,
pRpcMsg
,
false
);
if
(
ret
==
TAOS_SYNC_PROPOSE_NOT_LEADER
)
{
sTrace
(
"%s value%d write not leader"
,
s
,
alreadySend
);
}
else
{
assert
(
ret
==
0
);
sTrace
(
"%s value%d write ok"
,
s
,
alreadySend
);
}
alreadySend
++
;
rpcFreeCont
(
pRpcMsg
->
pCont
);
taosMemoryFree
(
pRpcMsg
);
}
else
{
sTrace
(
"%s"
,
s
);
}
taosMsleep
(
1000
);
taosMemoryFree
(
s
);
taosMsleep
(
1000
);
}
syncNodeRelease
(
pSyncNode
);
syncStop
(
rid
);
walClose
(
pWal
);
syncIOStop
();
cleanup
();
return
0
;
}
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录