From 11c3463814e50e764be454b72f070036c9dfea4b Mon Sep 17 00:00:00 2001 From: Shengliang Guan Date: Sun, 13 Nov 2022 18:18:46 +0800 Subject: [PATCH] refact: adjust sync code struct --- source/libs/sync/inc/syncInt.h | 57 -------------- source/libs/sync/inc/syncUtil.h | 40 ++++++++++ source/libs/sync/src/syncEnv.c | 2 +- source/libs/sync/src/syncMain.c | 77 ++++++++++++------- source/libs/sync/src/syncRaftCfg.c | 1 + .../libs/sync/test/sync_test_lib/inc/syncIO.h | 2 + .../sync/test/sync_test_lib/inc/syncTest.h | 1 + .../libs/sync/test/sync_test_lib/src/syncIO.c | 1 + 8 files changed, 96 insertions(+), 85 deletions(-) diff --git a/source/libs/sync/inc/syncInt.h b/source/libs/sync/inc/syncInt.h index 304d1bb695..e81f63d871 100644 --- a/source/libs/sync/inc/syncInt.h +++ b/source/libs/sync/inc/syncInt.h @@ -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); diff --git a/source/libs/sync/inc/syncUtil.h b/source/libs/sync/inc/syncUtil.h index d920e98ab0..b7ee320aa5 100644 --- a/source/libs/sync/inc/syncUtil.h +++ b/source/libs/sync/inc/syncUtil.h @@ -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); diff --git a/source/libs/sync/src/syncEnv.c b/source/libs/sync/src/syncEnv.c index a0e0a5a2c2..2abc8c0b7f 100644 --- a/source/libs/sync/src/syncEnv.c +++ b/source/libs/sync/src/syncEnv.c @@ -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; } diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c index f4cc8627f3..7ed90fb140 100644 --- a/source/libs/sync/src/syncMain.c +++ b/source/libs/sync/src/syncMain.c @@ -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) { diff --git a/source/libs/sync/src/syncRaftCfg.c b/source/libs/sync/src/syncRaftCfg.c index b6b2616973..b425b0dba7 100644 --- a/source/libs/sync/src/syncRaftCfg.c +++ b/source/libs/sync/src/syncRaftCfg.c @@ -15,6 +15,7 @@ #define _DEFAULT_SOURCE #include "syncRaftCfg.h" +#include "syncUtil.h" // file must already exist! SRaftCfgIndex *raftCfgIndexOpen(const char *path) { diff --git a/source/libs/sync/test/sync_test_lib/inc/syncIO.h b/source/libs/sync/test/sync_test_lib/inc/syncIO.h index 2c5e19d863..19f896182e 100644 --- a/source/libs/sync/test/sync_test_lib/inc/syncIO.h +++ b/source/libs/sync/test/sync_test_lib/inc/syncIO.h @@ -36,6 +36,8 @@ extern "C" { typedef struct SyncPing SyncPing; typedef struct SyncPingReply SyncPingReply; +extern bool gRaftDetailLog; + typedef struct SSyncIO { STaosQueue *pMsgQ; STaosQset *pQset; diff --git a/source/libs/sync/test/sync_test_lib/inc/syncTest.h b/source/libs/sync/test/sync_test_lib/inc/syncTest.h index dd67b6148e..443d3a45d6 100644 --- a/source/libs/sync/test/sync_test_lib/inc/syncTest.h +++ b/source/libs/sync/test/sync_test_lib/inc/syncTest.h @@ -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); diff --git a/source/libs/sync/test/sync_test_lib/src/syncIO.c b/source/libs/sync/test/sync_test_lib/src/syncIO.c index 14adc18c66..2c24451713 100644 --- a/source/libs/sync/test/sync_test_lib/src/syncIO.c +++ b/source/libs/sync/test/sync_test_lib/src/syncIO.c @@ -22,6 +22,7 @@ #include "ttimer.h" #include "tutil.h" +bool gRaftDetailLog = false; SSyncIO *gSyncIO = NULL; // local function ------------ -- GitLab