Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
7aa911b4
T
TDengine
项目概览
taosdata
/
TDengine
大约 1 年 前同步成功
通知
1181
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,发现更多精彩内容 >>
未验证
提交
7aa911b4
编写于
11月 13, 2022
作者:
S
Shengliang Guan
提交者:
GitHub
11月 13, 2022
浏览文件
操作
浏览文件
下载
差异文件
Merge pull request #18104 from taosdata/fix/TD-20052
refact: adjust sync code
上级
84fa23c9
11c34638
变更
8
隐藏空白更改
内联
并排
Showing
8 changed file
with
96 addition
and
85 deletion
+96
-85
source/libs/sync/inc/syncInt.h
source/libs/sync/inc/syncInt.h
+0
-57
source/libs/sync/inc/syncUtil.h
source/libs/sync/inc/syncUtil.h
+40
-0
source/libs/sync/src/syncEnv.c
source/libs/sync/src/syncEnv.c
+1
-1
source/libs/sync/src/syncMain.c
source/libs/sync/src/syncMain.c
+50
-27
source/libs/sync/src/syncRaftCfg.c
source/libs/sync/src/syncRaftCfg.c
+1
-0
source/libs/sync/test/sync_test_lib/inc/syncIO.h
source/libs/sync/test/sync_test_lib/inc/syncIO.h
+2
-0
source/libs/sync/test/sync_test_lib/inc/syncTest.h
source/libs/sync/test/sync_test_lib/inc/syncTest.h
+1
-0
source/libs/sync/test/sync_test_lib/src/syncIO.c
source/libs/sync/test/sync_test_lib/src/syncIO.c
+1
-0
未找到文件。
source/libs/sync/inc/syncInt.h
浏览文件 @
7aa911b4
...
...
@@ -22,49 +22,9 @@ extern "C" {
#include "sync.h"
#include "taosdef.h"
#include "tlog.h"
#include "trpc.h"
#include "ttimer.h"
// clang-format off
#define sFatal(...) if (sDebugFlag & DEBUG_FATAL) { taosPrintLog("SYN FATAL ", DEBUG_FATAL, 255, __VA_ARGS__); }
#define sError(...) if (sDebugFlag & DEBUG_ERROR) { taosPrintLog("SYN ERROR ", DEBUG_ERROR, 255, __VA_ARGS__); }
#define sWarn(...) if (sDebugFlag & DEBUG_WARN) { taosPrintLog("SYN WARN ", DEBUG_WARN, 255, __VA_ARGS__); }
#define sInfo(...) if (sDebugFlag & DEBUG_INFO) { taosPrintLog("SYN ", DEBUG_INFO, 255, __VA_ARGS__); }
#define sDebug(...) if (sDebugFlag & DEBUG_DEBUG) { taosPrintLog("SYN ", DEBUG_DEBUG, sDebugFlag, __VA_ARGS__); }
#define sTrace(...) if (sDebugFlag & DEBUG_TRACE) { taosPrintLog("SYN ", DEBUG_TRACE, sDebugFlag, __VA_ARGS__); }
#define sLFatal(...) if (sDebugFlag & DEBUG_FATAL) { taosPrintLongString("SYN FATAL ", DEBUG_FATAL, 255, __VA_ARGS__); }
#define sLError(...) if (sDebugFlag & DEBUG_ERROR) { taosPrintLongString("SYN ERROR ", DEBUG_ERROR, 255, __VA_ARGS__); }
#define sLWarn(...) if (sDebugFlag & DEBUG_WARN) { taosPrintLongString("SYN WARN ", DEBUG_WARN, 255, __VA_ARGS__); }
#define sLInfo(...) if (sDebugFlag & DEBUG_INFO) { taosPrintLongString("SYN ", DEBUG_INFO, 255, __VA_ARGS__); }
#define sLDebug(...) if (sDebugFlag & DEBUG_DEBUG) { taosPrintLongString("SYN ", DEBUG_DEBUG, sDebugFlag, __VA_ARGS__); }
#define sLTrace(...) if (sDebugFlag & DEBUG_TRACE) { taosPrintLongString("SYN ", DEBUG_TRACE, sDebugFlag, __VA_ARGS__); }
#define sNFatal(pNode, ...) if (sDebugFlag & DEBUG_FATAL) { syncPrintNodeLog("SYN FATAL ", DEBUG_FATAL, 255, pNode, __VA_ARGS__); }
#define sNError(pNode, ...) if (sDebugFlag & DEBUG_ERROR) { syncPrintNodeLog("SYN ERROR ", DEBUG_ERROR, 255, pNode, __VA_ARGS__); }
#define sNWarn(pNode, ...) if (sDebugFlag & DEBUG_WARN) { syncPrintNodeLog("SYN WARN ", DEBUG_WARN, 255, pNode, __VA_ARGS__); }
#define sNInfo(pNode, ...) if (sDebugFlag & DEBUG_INFO) { syncPrintNodeLog("SYN ", DEBUG_INFO, 255, pNode, __VA_ARGS__); }
#define sNDebug(pNode, ...) if (sDebugFlag & DEBUG_DEBUG) { syncPrintNodeLog("SYN ", DEBUG_DEBUG, sDebugFlag, pNode, __VA_ARGS__); }
#define sNTrace(pNode, ...) if (sDebugFlag & DEBUG_TRACE) { syncPrintNodeLog("SYN ", DEBUG_TRACE, sDebugFlag, pNode, __VA_ARGS__); }
#define sSFatal(pSender, ...) if (sDebugFlag & DEBUG_FATAL) { syncPrintSnapshotSenderLog("SYN FATAL ", DEBUG_FATAL, 255, pSender, __VA_ARGS__); }
#define sSError(pSender, ...) if (sDebugFlag & DEBUG_ERROR) { syncPrintSnapshotSenderLog("SYN ERROR ", DEBUG_ERROR, 255, pSender, __VA_ARGS__); }
#define sSWarn(pSender, ...) if (sDebugFlag & DEBUG_WARN) { syncPrintSnapshotSenderLog("SYN WARN ", DEBUG_WARN, 255, pSender, __VA_ARGS__); }
#define sSInfo(pSender, ...) if (sDebugFlag & DEBUG_INFO) { syncPrintSnapshotSenderLog("SYN ", DEBUG_INFO, 255, pSender, __VA_ARGS__); }
#define sSDebug(pSender, ...) if (sDebugFlag & DEBUG_DEBUG) { syncPrintSnapshotSenderLog("SYN ", DEBUG_DEBUG, sDebugFlag, pSender, __VA_ARGS__); }
#define sSTrace(pSender, ...) if (sDebugFlag & DEBUG_TRACE) { syncPrintSnapshotSenderLog("SYN ", DEBUG_TRACE, sDebugFlag, pSender, __VA_ARGS__); }
#define sRFatal(pReceiver, ...) if (sDebugFlag & DEBUG_FATAL) { syncPrintSnapshotReceiverLog("SYN FATAL ", DEBUG_FATAL, 255, pReceiver, __VA_ARGS__); }
#define sRError(pReceiver, ...) if (sDebugFlag & DEBUG_ERROR) { syncPrintSnapshotReceiverLog("SYN ERROR ", DEBUG_ERROR, 255, pReceiver, __VA_ARGS__); }
#define sRWarn(pReceiver, ...) if (sDebugFlag & DEBUG_WARN) { syncPrintSnapshotReceiverLog("SYN WARN ", DEBUG_WARN, 255, pReceiver, __VA_ARGS__); }
#define sRInfo(pReceiver, ...) if (sDebugFlag & DEBUG_INFO) { syncPrintSnapshotReceiverLog("SYN ", DEBUG_INFO, 255, pReceiver, __VA_ARGS__); }
#define sRDebug(pReceiver, ...) if (sDebugFlag & DEBUG_DEBUG) { syncPrintSnapshotReceiverLog("SYN ", DEBUG_DEBUG, sDebugFlag, pReceiver, __VA_ARGS__); }
#define sRTrace(pReceiver, ...) if (sDebugFlag & DEBUG_TRACE) { syncPrintSnapshotReceiverLog("SYN ", DEBUG_TRACE, sDebugFlag, pReceiver, __VA_ARGS__); }
// clang-format on
typedef
struct
SyncTimeout
SyncTimeout
;
typedef
struct
SyncClientRequest
SyncClientRequest
;
typedef
struct
SyncRequestVote
SyncRequestVote
;
...
...
@@ -91,8 +51,6 @@ typedef struct SyncHeartbeatReply SyncHeartbeatReply;
typedef
struct
SyncHeartbeat
SyncHeartbeat
;
typedef
struct
SyncPreSnapshot
SyncPreSnapshot
;
extern
bool
gRaftDetailLog
;
typedef
struct
SRaftId
{
SyncNodeId
addr
;
SyncGroupId
vgId
;
...
...
@@ -239,9 +197,6 @@ void syncNodeClose(SSyncNode* pSyncNode);
void
syncNodePreClose
(
SSyncNode
*
pSyncNode
);
int32_t
syncNodePropose
(
SSyncNode
*
pSyncNode
,
SRpcMsg
*
pMsg
,
bool
isWeak
);
ESyncStrategy
syncNodeStrategy
(
SSyncNode
*
pSyncNode
);
SyncIndex
syncNodeGetSnapshotConfigIndex
(
SSyncNode
*
pSyncNode
,
SyncIndex
snapshotLastApplyIndex
);
// on message ---------------------
int32_t
syncNodeOnTimeout
(
SSyncNode
*
ths
,
const
SRpcMsg
*
pMsg
);
int32_t
syncNodeOnClientRequest
(
SSyncNode
*
ths
,
SRpcMsg
*
pMsg
,
SyncIndex
*
pRetIndex
);
...
...
@@ -269,9 +224,6 @@ int32_t syncNodeRestartHeartbeatTimer(SSyncNode* pSyncNode);
// utils --------------
int32_t
syncNodeSendMsgById
(
const
SRaftId
*
destRaftId
,
SSyncNode
*
pSyncNode
,
SRpcMsg
*
pMsg
);
int32_t
syncNodeSendMsgByInfo
(
const
SNodeInfo
*
nodeInfo
,
SSyncNode
*
pSyncNode
,
SRpcMsg
*
pMsg
);
char
*
syncNode2SimpleStr
(
const
SSyncNode
*
pSyncNode
);
bool
syncNodeInConfig
(
SSyncNode
*
pSyncNode
,
const
SSyncCfg
*
config
);
void
syncNodeDoConfigChange
(
SSyncNode
*
pSyncNode
,
SSyncCfg
*
newConfig
,
SyncIndex
lastConfigChangeIndex
);
SyncIndex
syncMinMatchIndex
(
SSyncNode
*
pSyncNode
);
// raft state change --------------
...
...
@@ -302,13 +254,10 @@ SyncIndex syncNodeGetPreIndex(SSyncNode* pSyncNode, SyncIndex index);
SyncTerm
syncNodeGetPreTerm
(
SSyncNode
*
pSyncNode
,
SyncIndex
index
);
int32_t
syncNodeGetPreIndexTerm
(
SSyncNode
*
pSyncNode
,
SyncIndex
index
,
SyncIndex
*
pPreIndex
,
SyncTerm
*
pPreTerm
);
bool
syncNodeIsOptimizedOneReplica
(
SSyncNode
*
ths
,
SRpcMsg
*
pMsg
);
int32_t
syncNodeDoCommit
(
SSyncNode
*
ths
,
SyncIndex
beginIndex
,
SyncIndex
endIndex
,
uint64_t
flag
);
int32_t
syncNodeFollowerCommit
(
SSyncNode
*
ths
,
SyncIndex
newCommitIndex
);
int32_t
syncNodePreCommit
(
SSyncNode
*
ths
,
SSyncRaftEntry
*
pEntry
,
int32_t
code
);
int32_t
syncNodeUpdateNewConfigIndex
(
SSyncNode
*
ths
,
SSyncCfg
*
pNewCfg
);
bool
syncNodeInRaftGroup
(
SSyncNode
*
ths
,
SRaftId
*
pRaftId
);
SSyncSnapshotSender
*
syncNodeGetSnapshotSender
(
SSyncNode
*
ths
,
SRaftId
*
pDestId
);
SSyncTimer
*
syncNodeGetHbTimer
(
SSyncNode
*
ths
,
SRaftId
*
pDestId
);
...
...
@@ -318,13 +267,7 @@ bool syncNodeNeedSendAppendEntries(SSyncNode* ths, const SRaftId* pDestId, const
int32_t
syncGetSnapshotMeta
(
int64_t
rid
,
struct
SSnapshotMeta
*
sMeta
);
int32_t
syncGetSnapshotMetaByIndex
(
int64_t
rid
,
SyncIndex
snapshotIndex
,
struct
SSnapshotMeta
*
sMeta
);
bool
syncNodeCanChange
(
SSyncNode
*
pSyncNode
);
int32_t
syncNodeLeaderTransfer
(
SSyncNode
*
pSyncNode
);
int32_t
syncNodeLeaderTransferTo
(
SSyncNode
*
pSyncNode
,
SNodeInfo
newLeader
);
int32_t
syncDoLeaderTransfer
(
SSyncNode
*
ths
,
SRpcMsg
*
pRpcMsg
,
SSyncRaftEntry
*
pEntry
);
int32_t
syncNodeDynamicQuorum
(
const
SSyncNode
*
pSyncNode
);
bool
syncNodeIsMnode
(
SSyncNode
*
pSyncNode
);
int32_t
syncNodePeerStateInit
(
SSyncNode
*
pSyncNode
);
...
...
source/libs/sync/inc/syncUtil.h
浏览文件 @
7aa911b4
...
...
@@ -21,6 +21,46 @@ extern "C" {
#endif
#include "syncInt.h"
#include "tlog.h"
// clang-format off
#define sFatal(...) if (sDebugFlag & DEBUG_FATAL) { taosPrintLog("SYN FATAL ", DEBUG_FATAL, 255, __VA_ARGS__); }
#define sError(...) if (sDebugFlag & DEBUG_ERROR) { taosPrintLog("SYN ERROR ", DEBUG_ERROR, 255, __VA_ARGS__); }
#define sWarn(...) if (sDebugFlag & DEBUG_WARN) { taosPrintLog("SYN WARN ", DEBUG_WARN, 255, __VA_ARGS__); }
#define sInfo(...) if (sDebugFlag & DEBUG_INFO) { taosPrintLog("SYN ", DEBUG_INFO, 255, __VA_ARGS__); }
#define sDebug(...) if (sDebugFlag & DEBUG_DEBUG) { taosPrintLog("SYN ", DEBUG_DEBUG, sDebugFlag, __VA_ARGS__); }
#define sTrace(...) if (sDebugFlag & DEBUG_TRACE) { taosPrintLog("SYN ", DEBUG_TRACE, sDebugFlag, __VA_ARGS__); }
#define sLFatal(...) if (sDebugFlag & DEBUG_FATAL) { taosPrintLongString("SYN FATAL ", DEBUG_FATAL, 255, __VA_ARGS__); }
#define sLError(...) if (sDebugFlag & DEBUG_ERROR) { taosPrintLongString("SYN ERROR ", DEBUG_ERROR, 255, __VA_ARGS__); }
#define sLWarn(...) if (sDebugFlag & DEBUG_WARN) { taosPrintLongString("SYN WARN ", DEBUG_WARN, 255, __VA_ARGS__); }
#define sLInfo(...) if (sDebugFlag & DEBUG_INFO) { taosPrintLongString("SYN ", DEBUG_INFO, 255, __VA_ARGS__); }
#define sLDebug(...) if (sDebugFlag & DEBUG_DEBUG) { taosPrintLongString("SYN ", DEBUG_DEBUG, sDebugFlag, __VA_ARGS__); }
#define sLTrace(...) if (sDebugFlag & DEBUG_TRACE) { taosPrintLongString("SYN ", DEBUG_TRACE, sDebugFlag, __VA_ARGS__); }
#define sNFatal(pNode, ...) if (sDebugFlag & DEBUG_FATAL) { syncPrintNodeLog("SYN FATAL ", DEBUG_FATAL, 255, pNode, __VA_ARGS__); }
#define sNError(pNode, ...) if (sDebugFlag & DEBUG_ERROR) { syncPrintNodeLog("SYN ERROR ", DEBUG_ERROR, 255, pNode, __VA_ARGS__); }
#define sNWarn(pNode, ...) if (sDebugFlag & DEBUG_WARN) { syncPrintNodeLog("SYN WARN ", DEBUG_WARN, 255, pNode, __VA_ARGS__); }
#define sNInfo(pNode, ...) if (sDebugFlag & DEBUG_INFO) { syncPrintNodeLog("SYN ", DEBUG_INFO, 255, pNode, __VA_ARGS__); }
#define sNDebug(pNode, ...) if (sDebugFlag & DEBUG_DEBUG) { syncPrintNodeLog("SYN ", DEBUG_DEBUG, sDebugFlag, pNode, __VA_ARGS__); }
#define sNTrace(pNode, ...) if (sDebugFlag & DEBUG_TRACE) { syncPrintNodeLog("SYN ", DEBUG_TRACE, sDebugFlag, pNode, __VA_ARGS__); }
#define sSFatal(pSender, ...) if (sDebugFlag & DEBUG_FATAL) { syncPrintSnapshotSenderLog("SYN FATAL ", DEBUG_FATAL, 255, pSender, __VA_ARGS__); }
#define sSError(pSender, ...) if (sDebugFlag & DEBUG_ERROR) { syncPrintSnapshotSenderLog("SYN ERROR ", DEBUG_ERROR, 255, pSender, __VA_ARGS__); }
#define sSWarn(pSender, ...) if (sDebugFlag & DEBUG_WARN) { syncPrintSnapshotSenderLog("SYN WARN ", DEBUG_WARN, 255, pSender, __VA_ARGS__); }
#define sSInfo(pSender, ...) if (sDebugFlag & DEBUG_INFO) { syncPrintSnapshotSenderLog("SYN ", DEBUG_INFO, 255, pSender, __VA_ARGS__); }
#define sSDebug(pSender, ...) if (sDebugFlag & DEBUG_DEBUG) { syncPrintSnapshotSenderLog("SYN ", DEBUG_DEBUG, sDebugFlag, pSender, __VA_ARGS__); }
#define sSTrace(pSender, ...) if (sDebugFlag & DEBUG_TRACE) { syncPrintSnapshotSenderLog("SYN ", DEBUG_TRACE, sDebugFlag, pSender, __VA_ARGS__); }
#define sRFatal(pReceiver, ...) if (sDebugFlag & DEBUG_FATAL) { syncPrintSnapshotReceiverLog("SYN FATAL ", DEBUG_FATAL, 255, pReceiver, __VA_ARGS__); }
#define sRError(pReceiver, ...) if (sDebugFlag & DEBUG_ERROR) { syncPrintSnapshotReceiverLog("SYN ERROR ", DEBUG_ERROR, 255, pReceiver, __VA_ARGS__); }
#define sRWarn(pReceiver, ...) if (sDebugFlag & DEBUG_WARN) { syncPrintSnapshotReceiverLog("SYN WARN ", DEBUG_WARN, 255, pReceiver, __VA_ARGS__); }
#define sRInfo(pReceiver, ...) if (sDebugFlag & DEBUG_INFO) { syncPrintSnapshotReceiverLog("SYN ", DEBUG_INFO, 255, pReceiver, __VA_ARGS__); }
#define sRDebug(pReceiver, ...) if (sDebugFlag & DEBUG_DEBUG) { syncPrintSnapshotReceiverLog("SYN ", DEBUG_DEBUG, sDebugFlag, pReceiver, __VA_ARGS__); }
#define sRTrace(pReceiver, ...) if (sDebugFlag & DEBUG_TRACE) { syncPrintSnapshotReceiverLog("SYN ", DEBUG_TRACE, sDebugFlag, pReceiver, __VA_ARGS__); }
// clang-format on
uint64_t
syncUtilAddr2U64
(
const
char
*
host
,
uint16_t
port
);
void
syncUtilU642Addr
(
uint64_t
u64
,
char
*
host
,
int64_t
len
,
uint16_t
*
port
);
...
...
source/libs/sync/src/syncEnv.c
浏览文件 @
7aa911b4
...
...
@@ -15,11 +15,11 @@
#define _DEFAULT_SOURCE
#include "syncEnv.h"
#include "syncUtil.h"
#include "tref.h"
static
SSyncEnv
gSyncEnv
=
{
0
};
static
int32_t
gNodeRefId
=
-
1
;
bool
gRaftDetailLog
=
false
;
static
void
syncEnvTick
(
void
*
param
,
void
*
tmrId
);
SSyncEnv
*
syncEnv
()
{
return
&
gSyncEnv
;
}
...
...
source/libs/sync/src/syncMain.c
浏览文件 @
7aa911b4
...
...
@@ -45,6 +45,18 @@ static bool syncIsConfigChanged(const SSyncCfg* pOldCfg, const SSyncCfg* pNew
static
int32_t
syncHbTimerInit
(
SSyncNode
*
pSyncNode
,
SSyncTimer
*
pSyncTimer
,
SRaftId
destId
);
static
int32_t
syncHbTimerStart
(
SSyncNode
*
pSyncNode
,
SSyncTimer
*
pSyncTimer
);
static
int32_t
syncHbTimerStop
(
SSyncNode
*
pSyncNode
,
SSyncTimer
*
pSyncTimer
);
static
int32_t
syncNodeUpdateNewConfigIndex
(
SSyncNode
*
ths
,
SSyncCfg
*
pNewCfg
);
static
bool
syncNodeInConfig
(
SSyncNode
*
pSyncNode
,
const
SSyncCfg
*
config
);
static
void
syncNodeDoConfigChange
(
SSyncNode
*
pSyncNode
,
SSyncCfg
*
newConfig
,
SyncIndex
lastConfigChangeIndex
);
static
bool
syncNodeIsOptimizedOneReplica
(
SSyncNode
*
ths
,
SRpcMsg
*
pMsg
);
static
bool
syncNodeCanChange
(
SSyncNode
*
pSyncNode
);
static
int32_t
syncNodeLeaderTransfer
(
SSyncNode
*
pSyncNode
);
static
int32_t
syncNodeLeaderTransferTo
(
SSyncNode
*
pSyncNode
,
SNodeInfo
newLeader
);
static
int32_t
syncDoLeaderTransfer
(
SSyncNode
*
ths
,
SRpcMsg
*
pRpcMsg
,
SSyncRaftEntry
*
pEntry
);
static
ESyncStrategy
syncNodeStrategy
(
SSyncNode
*
pSyncNode
);
static
SyncIndex
syncNodeGetSnapshotConfigIndex
(
SSyncNode
*
pSyncNode
,
SyncIndex
snapshotLastApplyIndex
);
int64_t
syncOpen
(
SSyncInfo
*
pSyncInfo
)
{
SSyncNode
*
pSyncNode
=
syncNodeOpen
(
pSyncInfo
);
...
...
@@ -133,31 +145,44 @@ int32_t syncProcessMsg(int64_t rid, SRpcMsg* pMsg) {
SSyncNode
*
pSyncNode
=
syncNodeAcquire
(
rid
);
if
(
pSyncNode
==
NULL
)
return
code
;
if
(
pMsg
->
msgType
==
TDMT_SYNC_HEARTBEAT
)
{
code
=
syncNodeOnHeartbeat
(
pSyncNode
,
pMsg
);
}
else
if
(
pMsg
->
msgType
==
TDMT_SYNC_HEARTBEAT_REPLY
)
{
code
=
syncNodeOnHeartbeatReply
(
pSyncNode
,
pMsg
);
}
else
if
(
pMsg
->
msgType
==
TDMT_SYNC_TIMEOUT
)
{
code
=
syncNodeOnTimeout
(
pSyncNode
,
pMsg
);
}
else
if
(
pMsg
->
msgType
==
TDMT_SYNC_CLIENT_REQUEST
)
{
code
=
syncNodeOnClientRequest
(
pSyncNode
,
pMsg
,
NULL
);
}
else
if
(
pMsg
->
msgType
==
TDMT_SYNC_REQUEST_VOTE
)
{
code
=
syncNodeOnRequestVote
(
pSyncNode
,
pMsg
);
}
else
if
(
pMsg
->
msgType
==
TDMT_SYNC_REQUEST_VOTE_REPLY
)
{
code
=
syncNodeOnRequestVoteReply
(
pSyncNode
,
pMsg
);
}
else
if
(
pMsg
->
msgType
==
TDMT_SYNC_APPEND_ENTRIES
)
{
code
=
syncNodeOnAppendEntries
(
pSyncNode
,
pMsg
);
}
else
if
(
pMsg
->
msgType
==
TDMT_SYNC_APPEND_ENTRIES_REPLY
)
{
code
=
syncNodeOnAppendEntriesReply
(
pSyncNode
,
pMsg
);
}
else
if
(
pMsg
->
msgType
==
TDMT_SYNC_SNAPSHOT_SEND
)
{
code
=
syncNodeOnSnapshot
(
pSyncNode
,
pMsg
);
}
else
if
(
pMsg
->
msgType
==
TDMT_SYNC_SNAPSHOT_RSP
)
{
code
=
syncNodeOnSnapshotReply
(
pSyncNode
,
pMsg
);
}
else
if
(
pMsg
->
msgType
==
TDMT_SYNC_LOCAL_CMD
)
{
code
=
syncNodeOnLocalCmd
(
pSyncNode
,
pMsg
);
}
else
{
sError
(
"vgId:%d, failed to process msg:%p since invalid type:%s"
,
pSyncNode
->
vgId
,
pMsg
,
TMSG_INFO
(
pMsg
->
msgType
));
code
=
-
1
;
switch
(
pMsg
->
msgType
)
{
case
TDMT_SYNC_HEARTBEAT
:
code
=
syncNodeOnHeartbeat
(
pSyncNode
,
pMsg
);
break
;
case
TDMT_SYNC_HEARTBEAT_REPLY
:
code
=
syncNodeOnHeartbeatReply
(
pSyncNode
,
pMsg
);
break
;
case
TDMT_SYNC_TIMEOUT
:
code
=
syncNodeOnTimeout
(
pSyncNode
,
pMsg
);
break
;
case
TDMT_SYNC_CLIENT_REQUEST
:
code
=
syncNodeOnClientRequest
(
pSyncNode
,
pMsg
,
NULL
);
break
;
case
TDMT_SYNC_REQUEST_VOTE
:
code
=
syncNodeOnRequestVote
(
pSyncNode
,
pMsg
);
break
;
case
TDMT_SYNC_REQUEST_VOTE_REPLY
:
code
=
syncNodeOnRequestVoteReply
(
pSyncNode
,
pMsg
);
break
;
case
TDMT_SYNC_APPEND_ENTRIES
:
code
=
syncNodeOnAppendEntries
(
pSyncNode
,
pMsg
);
break
;
case
TDMT_SYNC_APPEND_ENTRIES_REPLY
:
code
=
syncNodeOnAppendEntriesReply
(
pSyncNode
,
pMsg
);
break
;
case
TDMT_SYNC_SNAPSHOT_SEND
:
code
=
syncNodeOnSnapshot
(
pSyncNode
,
pMsg
);
break
;
case
TDMT_SYNC_SNAPSHOT_RSP
:
code
=
syncNodeOnSnapshotReply
(
pSyncNode
,
pMsg
);
break
;
case
TDMT_SYNC_LOCAL_CMD
:
code
=
syncNodeOnLocalCmd
(
pSyncNode
,
pMsg
);
break
;
default:
sError
(
"vgId:%d, failed to process msg:%p since invalid type:%s"
,
pSyncNode
->
vgId
,
pMsg
,
TMSG_INFO
(
pMsg
->
msgType
));
code
=
-
1
;
}
syncNodeRelease
(
pSyncNode
);
...
...
@@ -1619,8 +1644,6 @@ void syncNodeCandidate2Follower(SSyncNode* pSyncNode) {
sNTrace
(
pSyncNode
,
"candidate to follower"
);
}
// raft vote --------------
// just called by syncNodeVoteForSelf
// need assert
void
syncNodeVoteForTerm
(
SSyncNode
*
pSyncNode
,
SyncTerm
term
,
SRaftId
*
pRaftId
)
{
...
...
source/libs/sync/src/syncRaftCfg.c
浏览文件 @
7aa911b4
...
...
@@ -15,6 +15,7 @@
#define _DEFAULT_SOURCE
#include "syncRaftCfg.h"
#include "syncUtil.h"
// file must already exist!
SRaftCfgIndex
*
raftCfgIndexOpen
(
const
char
*
path
)
{
...
...
source/libs/sync/test/sync_test_lib/inc/syncIO.h
浏览文件 @
7aa911b4
...
...
@@ -36,6 +36,8 @@ extern "C" {
typedef
struct
SyncPing
SyncPing
;
typedef
struct
SyncPingReply
SyncPingReply
;
extern
bool
gRaftDetailLog
;
typedef
struct
SSyncIO
{
STaosQueue
*
pMsgQ
;
STaosQset
*
pQset
;
...
...
source/libs/sync/test/sync_test_lib/inc/syncTest.h
浏览文件 @
7aa911b4
...
...
@@ -110,6 +110,7 @@ void logStoreSimpleLog2(char* s, SSyncLogStore* pLogStore);
cJSON
*
syncNode2Json
(
const
SSyncNode
*
pSyncNode
);
char
*
syncNode2Str
(
const
SSyncNode
*
pSyncNode
);
char
*
syncNode2SimpleStr
(
const
SSyncNode
*
pSyncNode
);
cJSON
*
voteGranted2Json
(
SVotesGranted
*
pVotesGranted
);
char
*
voteGranted2Str
(
SVotesGranted
*
pVotesGranted
);
...
...
source/libs/sync/test/sync_test_lib/src/syncIO.c
浏览文件 @
7aa911b4
...
...
@@ -22,6 +22,7 @@
#include "ttimer.h"
#include "tutil.h"
bool
gRaftDetailLog
=
false
;
SSyncIO
*
gSyncIO
=
NULL
;
// local function ------------
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录