xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge pull request #10585 from taosdata/feature/3.0_mhli 

Feature/3.0 mhli
This commit is contained in:
Li Minghao 2022-03-07 16:28:58 +08:00 committed by GitHub
parent 342ef936be fa8284af73
commit 2f792722c4
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
29 changed files with 422 additions and 517 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
source/libs/sync/inc/syncAppendEntries.h
View File

@ -28,9 +28,7 @@ extern "C" {
#include "syncRaft.h" #include "syncRaft.h"
#include "taosdef.h" #include "taosdef.h"
void appendEntries(SRaft *pRaft, const SyncAppendEntries *pMsg); int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg);
void onAppendEntries(SRaft *pRaft, const SyncAppendEntries *pMsg);
#ifdef __cplusplus #ifdef __cplusplus
} }

2
source/libs/sync/inc/syncAppendEntriesReply.h
View File

@ -28,7 +28,7 @@ extern "C" {
#include "syncRaft.h" #include "syncRaft.h"
#include "taosdef.h" #include "taosdef.h"
void onAppendEntriesReply(SRaft *pRaft, const SyncAppendEntriesReply *pMsg); int32_t syncNodeOnAppendEntriesReplyCb(SSyncNode* ths, SyncAppendEntriesReply* pMsg);
#ifdef __cplusplus #ifdef __cplusplus
} }

7
source/libs/sync/inc/syncElection.h
View File

@ -26,8 +26,11 @@ extern "C" {
#include "syncInt.h" #include "syncInt.h"
#include "taosdef.h" #include "taosdef.h"
void syncNodeElect(SSyncNode* pSyncNode); int32_t syncNodeElect(SSyncNode* pSyncNode);
void syncNodeRequestVotePeers(SSyncNode* pSyncNode);
int32_t syncNodeRequestVotePeers(SSyncNode* pSyncNode);
int32_t syncNodeRequestVote(SSyncNode* pSyncNode, const SRaftId* destRaftId, const SyncRequestVote* pMsg);
#ifdef __cplusplus #ifdef __cplusplus
} }

14
source/libs/sync/inc/syncEnv.h
View File

@ -30,23 +30,25 @@ extern "C" {
#define TIMER_MAX_MS 0x7FFFFFFF #define TIMER_MAX_MS 0x7FFFFFFF
#define PING_TIMER_MS 1000 #define PING_TIMER_MS 1000
#define ELECT_TIMER_MS_MIN 150
#define ELECT_TIMER_MS_MAX 300
#define ELECT_TIMER_MS_RANGE (ELECT_TIMER_MS_MAX - ELECT_TIMER_MS_MIN)
#define HEARTBEAT_TIMER_MS 30
#define EMPTY_RAFT_ID ((SRaftId){.addr = 0, .vgId = 0})
typedef struct SSyncEnv { typedef struct SSyncEnv {
tmr_h pEnvTickTimer; tmr_h pEnvTickTimer;
tmr_h pTimerManager; tmr_h pTimerManager;
char name[128]; char name[128];
} SSyncEnv; } SSyncEnv;
extern SSyncEnv* gSyncEnv; extern SSyncEnv* gSyncEnv;
int32_t syncEnvStart(); int32_t syncEnvStart();
int32_t syncEnvStop(); int32_t syncEnvStop();
tmr_h syncEnvStartTimer(TAOS_TMR_CALLBACK fp, int mseconds, void* param);
tmr_h syncEnvStartTimer(TAOS_TMR_CALLBACK fp, int mseconds, void* param); void syncEnvStopTimer(tmr_h* pTimer);
void syncEnvStopTimer(tmr_h* pTimer);
#ifdef __cplusplus #ifdef __cplusplus
} }

33
source/libs/sync/inc/syncInt.h
View File

@ -154,9 +154,8 @@ typedef struct SSyncNode {
SyncIndex commitIndex; SyncIndex commitIndex;
// timer // timer
tmr_h pPingTimer; tmr_h pPingTimer;
int32_t pingTimerMS; int32_t pingTimerMS;
// uint8_t pingTimerEnable;
uint64_t pingTimerLogicClock; uint64_t pingTimerLogicClock;
uint64_t pingTimerLogicClockUser; uint64_t pingTimerLogicClockUser;
TAOS_TMR_CALLBACK FpPingTimer; // Timer Fp TAOS_TMR_CALLBACK FpPingTimer; // Timer Fp
@ -164,13 +163,15 @@ typedef struct SSyncNode {
tmr_h pElectTimer; tmr_h pElectTimer;
int32_t electTimerMS; int32_t electTimerMS;
uint8_t electTimerEnable; uint64_t electTimerLogicClock;
uint64_t electTimerLogicClockUser;
TAOS_TMR_CALLBACK FpElectTimer; // Timer Fp TAOS_TMR_CALLBACK FpElectTimer; // Timer Fp
uint64_t electTimerCounter; uint64_t electTimerCounter;
tmr_h pHeartbeatTimer; tmr_h pHeartbeatTimer;
int32_t heartbeatTimerMS; int32_t heartbeatTimerMS;
uint8_t heartbeatTimerEnable; uint64_t heartbeatTimerLogicClock;
uint64_t heartbeatTimerLogicClockUser;
TAOS_TMR_CALLBACK FpHeartbeatTimer; // Timer Fp TAOS_TMR_CALLBACK FpHeartbeatTimer; // Timer Fp
uint64_t heartbeatTimerCounter; uint64_t heartbeatTimerCounter;
@ -187,26 +188,22 @@ typedef struct SSyncNode {
SSyncNode* syncNodeOpen(const SSyncInfo* pSyncInfo); SSyncNode* syncNodeOpen(const SSyncInfo* pSyncInfo);
void syncNodeClose(SSyncNode* pSyncNode); void syncNodeClose(SSyncNode* pSyncNode);
void syncNodePingAll(SSyncNode* pSyncNode);
void syncNodePingPeers(SSyncNode* pSyncNode); int32_t syncNodeSendMsgById(const SRaftId* destRaftId, SSyncNode* pSyncNode, SRpcMsg* pMsg);
void syncNodePingSelf(SSyncNode* pSyncNode); int32_t syncNodeSendMsgByInfo(const SNodeInfo* nodeInfo, SSyncNode* pSyncNode, SRpcMsg* pMsg);
int32_t syncNodePing(SSyncNode* pSyncNode, const SRaftId* destRaftId, SyncPing* pMsg);
void syncNodePingAll(SSyncNode* pSyncNode);
void syncNodePingPeers(SSyncNode* pSyncNode);
void syncNodePingSelf(SSyncNode* pSyncNode);
int32_t syncNodeStartPingTimer(SSyncNode* pSyncNode); int32_t syncNodeStartPingTimer(SSyncNode* pSyncNode);
int32_t syncNodeStopPingTimer(SSyncNode* pSyncNode); int32_t syncNodeStopPingTimer(SSyncNode* pSyncNode);
int32_t syncNodeStartElectTimer(SSyncNode* pSyncNode, int32_t ms);
int32_t syncNodeStartElectTimer(SSyncNode* pSyncNode);
int32_t syncNodeStopElectTimer(SSyncNode* pSyncNode); int32_t syncNodeStopElectTimer(SSyncNode* pSyncNode);
int32_t syncNodeRestartElectTimer(SSyncNode* pSyncNode, int32_t ms);
int32_t syncNodeStartHeartbeatTimer(SSyncNode* pSyncNode); int32_t syncNodeStartHeartbeatTimer(SSyncNode* pSyncNode);
int32_t syncNodeStopHeartbeatTimer(SSyncNode* pSyncNode); int32_t syncNodeStopHeartbeatTimer(SSyncNode* pSyncNode);
int32_t syncNodeRequestVote(SSyncNode* ths, const SyncRequestVote* pMsg);
int32_t syncNodeOnRequestVoteCb(SSyncNode* ths, SyncRequestVote* pMsg);
int32_t syncNodeOnRequestVoteReplyCb(SSyncNode* ths, SyncRequestVoteReply* pMsg);
int32_t syncNodeAppendEntries(SSyncNode* ths, const SyncAppendEntries* pMsg);
int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg);
int32_t syncNodeOnAppendEntriesReplyCb(SSyncNode* ths, SyncAppendEntriesReply* pMsg);
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif

3
source/libs/sync/inc/syncMessage.h
View File

@ -59,6 +59,7 @@ typedef struct SyncTimeout {
uint32_t msgType; uint32_t msgType;
ESyncTimeoutType timeoutType; ESyncTimeoutType timeoutType;
uint64_t logicClock; uint64_t logicClock;
int32_t timerMS;
void* data; void* data;
} SyncTimeout; } SyncTimeout;
@ -69,7 +70,7 @@ void syncTimeoutDeserialize(const char* buf, uint32_t len, SyncTimeout*
void syncTimeout2RpcMsg(const SyncTimeout* pMsg, SRpcMsg* pRpcMsg); void syncTimeout2RpcMsg(const SyncTimeout* pMsg, SRpcMsg* pRpcMsg);
void syncTimeoutFromRpcMsg(const SRpcMsg* pRpcMsg, SyncTimeout* pMsg); void syncTimeoutFromRpcMsg(const SRpcMsg* pRpcMsg, SyncTimeout* pMsg);
cJSON* syncTimeout2Json(const SyncTimeout* pMsg); cJSON* syncTimeout2Json(const SyncTimeout* pMsg);
SyncTimeout* syncTimeoutBuild2(ESyncTimeoutType timeoutType, uint64_t logicClock, void* data); SyncTimeout* syncTimeoutBuild2(ESyncTimeoutType timeoutType, uint64_t logicClock, int32_t timerMS, void* data);
// --------------------------------------------- // ---------------------------------------------
typedef struct SyncPing { typedef struct SyncPing {

2
source/libs/sync/inc/syncOnMessage.h
View File

@ -26,8 +26,6 @@ extern "C" {
#include "syncRaft.h" #include "syncRaft.h"
#include "taosdef.h" #include "taosdef.h"
void onMessage(SRaft *pRaft, void *pMsg);
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif

24
source/libs/sync/inc/syncRaftStore.h
View File

@ -32,28 +32,18 @@ extern "C" {
#define RAFT_STORE_PATH_LEN 128 #define RAFT_STORE_PATH_LEN 128
typedef struct SRaftStore { typedef struct SRaftStore {
SyncTerm currentTerm; SyncTerm currentTerm;
SRaftId voteFor; SRaftId voteFor;
// FileFd fd;
TdFilePtr pFile; TdFilePtr pFile;
char path[RAFT_STORE_PATH_LEN]; char path[RAFT_STORE_PATH_LEN];
} SRaftStore; } SRaftStore;
SRaftStore *raftStoreOpen(const char *path); SRaftStore *raftStoreOpen(const char *path);
int32_t raftStoreClose(SRaftStore *pRaftStore);
static int32_t raftStoreInit(SRaftStore *pRaftStore); int32_t raftStorePersist(SRaftStore *pRaftStore);
int32_t raftStoreSerialize(SRaftStore *pRaftStore, char *buf, size_t len);
int32_t raftStoreClose(SRaftStore *pRaftStore); int32_t raftStoreDeserialize(SRaftStore *pRaftStore, char *buf, size_t len);
void raftStorePrint(SRaftStore *pRaftStore);
int32_t raftStorePersist(SRaftStore *pRaftStore);
static bool raftStoreFileExist(char *path);
int32_t raftStoreSerialize(SRaftStore *pRaftStore, char *buf, size_t len);
int32_t raftStoreDeserialize(SRaftStore *pRaftStore, char *buf, size_t len);
void raftStorePrint(SRaftStore *pRaftStore);
#ifdef __cplusplus #ifdef __cplusplus
} }

4
source/libs/sync/inc/syncReplication.h
View File

@ -26,7 +26,9 @@ extern "C" {
#include "syncInt.h" #include "syncInt.h"
#include "taosdef.h" #include "taosdef.h"
void syncNodeAppendEntriesPeers(SSyncNode* pSyncNode); int32_t syncNodeAppendEntriesPeers(SSyncNode* pSyncNode);
int32_t syncNodeAppendEntries(SSyncNode* pSyncNode, const SRaftId* destRaftId, const SyncAppendEntries* pMsg);
#ifdef __cplusplus #ifdef __cplusplus
} }

2
source/libs/sync/inc/syncRequestVote.h
View File

@ -28,6 +28,8 @@ extern "C" {
#include "syncRaft.h" #include "syncRaft.h"
#include "taosdef.h" #include "taosdef.h"
int32_t syncNodeOnRequestVoteCb(SSyncNode* ths, SyncRequestVote* pMsg);
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif

2
source/libs/sync/inc/syncRequestVoteReply.h
View File

@ -28,6 +28,8 @@ extern "C" {
#include "syncRaft.h" #include "syncRaft.h"
#include "taosdef.h" #include "taosdef.h"
int32_t syncNodeOnRequestVoteReplyCb(SSyncNode* ths, SyncRequestVoteReply* pMsg);
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif

2
source/libs/sync/inc/syncTimeout.h
View File

@ -28,7 +28,7 @@ extern "C" {
#include "syncRaft.h" #include "syncRaft.h"
#include "taosdef.h" #include "taosdef.h"
void onTimeout(SRaft *pRaft, void *pMsg); int32_t syncNodeOnTimeoutCb(SSyncNode* ths, SyncTimeout* pMsg);
#ifdef __cplusplus #ifdef __cplusplus
} }

23
source/libs/sync/inc/syncUtil.h
View File

@ -28,28 +28,23 @@ extern "C" {
#include "taosdef.h" #include "taosdef.h"
// ---- encode / decode // ---- encode / decode
uint64_t syncUtilAddr2U64(const char* host, uint16_t port); uint64_t syncUtilAddr2U64(const char* host, uint16_t port);
void syncUtilU642Addr(uint64_t u64, char* host, size_t len, uint16_t* port);
void syncUtilU642Addr(uint64_t u64, char* host, size_t len, uint16_t* port); void syncUtilnodeInfo2EpSet(const SNodeInfo* pNodeInfo, SEpSet* pEpSet);
void syncUtilraftId2EpSet(const SRaftId* raftId, SEpSet* pEpSet);
void syncUtilnodeInfo2EpSet(const SNodeInfo* pNodeInfo, SEpSet* pEpSet); void syncUtilnodeInfo2raftId(const SNodeInfo* pNodeInfo, SyncGroupId vgId, SRaftId* raftId);
bool syncUtilSameId(const SRaftId* pId1, const SRaftId* pId2);
void syncUtilraftId2EpSet(const SRaftId* raftId, SEpSet* pEpSet);
void syncUtilnodeInfo2raftId(const SNodeInfo* pNodeInfo, SyncGroupId vgId, SRaftId* raftId);
bool syncUtilSameId(const SRaftId* pId1, const SRaftId* pId2);
// ---- SSyncBuffer ---- // ---- SSyncBuffer ----
void syncUtilbufBuild(SSyncBuffer* syncBuf, size_t len); void syncUtilbufBuild(SSyncBuffer* syncBuf, size_t len);
void syncUtilbufDestroy(SSyncBuffer* syncBuf); void syncUtilbufDestroy(SSyncBuffer* syncBuf);
void syncUtilbufCopy(const SSyncBuffer* src, SSyncBuffer* dest); void syncUtilbufCopy(const SSyncBuffer* src, SSyncBuffer* dest);
void syncUtilbufCopyDeep(const SSyncBuffer* src, SSyncBuffer* dest); void syncUtilbufCopyDeep(const SSyncBuffer* src, SSyncBuffer* dest);
// ---- misc ----
int32_t syncUtilRand(int32_t max);
int32_t syncUtilElectRandomMS();
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif

160
source/libs/sync/src/syncAppendEntries.c
View File

@ -15,97 +15,69 @@
#include "syncAppendEntries.h" #include "syncAppendEntries.h"
int32_t syncNodeAppendEntries(SSyncNode* ths, const SyncAppendEntries* pMsg) { // TLA+ Spec
// TLA+ Spec // HandleAppendEntriesRequest(i, j, m) ==
// AppendEntries(i, j) == // LET logOk == \/ m.mprevLogIndex = 0
// /\ i /= j // \/ /\ m.mprevLogIndex > 0
// /\ state[i] = Leader // /\ m.mprevLogIndex <= Len(log[i])
// /\ LET prevLogIndex == nextIndex[i][j] - 1 // /\ m.mprevLogTerm = log[i][m.mprevLogIndex].term
// prevLogTerm == IF prevLogIndex > 0 THEN // IN /\ m.mterm <= currentTerm[i]
// log[i][prevLogIndex].term // /\ \/ /\ \* reject request
// ELSE // \/ m.mterm < currentTerm[i]
// 0 // \/ /\ m.mterm = currentTerm[i]
// \* Send up to 1 entry, constrained by the end of the log. // /\ state[i] = Follower
// lastEntry == Min({Len(log[i]), nextIndex[i][j]}) // /\ \lnot logOk
// entries == SubSeq(log[i], nextIndex[i][j], lastEntry) // /\ Reply([mtype |-> AppendEntriesResponse,
// IN Send([mtype |-> AppendEntriesRequest, // mterm |-> currentTerm[i],
// mterm |-> currentTerm[i], // msuccess |-> FALSE,
// mprevLogIndex |-> prevLogIndex, // mmatchIndex |-> 0,
// mprevLogTerm |-> prevLogTerm, // msource |-> i,
// mentries |-> entries, // mdest |-> j],
// \* mlog is used as a history variable for the proof. // m)
// \* It would not exist in a real implementation. // /\ UNCHANGED <<serverVars, logVars>>
// mlog |-> log[i], // \/ \* return to follower state
// mcommitIndex |-> Min({commitIndex[i], lastEntry}), // /\ m.mterm = currentTerm[i]
// msource |-> i, // /\ state[i] = Candidate
// mdest |-> j]) // /\ state' = [state EXCEPT ![i] = Follower]
// /\ UNCHANGED <<serverVars, candidateVars, leaderVars, logVars>> // /\ UNCHANGED <<currentTerm, votedFor, logVars, messages>>
} // \/ \* accept request
// /\ m.mterm = currentTerm[i]
int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) { // /\ state[i] = Follower
// TLA+ Spec // /\ logOk
// HandleAppendEntriesRequest(i, j, m) == // /\ LET index == m.mprevLogIndex + 1
// LET logOk == \/ m.mprevLogIndex = 0 // IN \/ \* already done with request
// \/ /\ m.mprevLogIndex > 0 // /\ \/ m.mentries = << >>
// /\ m.mprevLogIndex <= Len(log[i]) // \/ /\ m.mentries /= << >>
// /\ m.mprevLogTerm = log[i][m.mprevLogIndex].term // /\ Len(log[i]) >= index
// IN /\ m.mterm <= currentTerm[i] // /\ log[i][index].term = m.mentries[1].term
// /\ \/ /\ \* reject request // \* This could make our commitIndex decrease (for
// \/ m.mterm < currentTerm[i] // \* example if we process an old, duplicated request),
// \/ /\ m.mterm = currentTerm[i] // \* but that doesn't really affect anything.
// /\ state[i] = Follower // /\ commitIndex' = [commitIndex EXCEPT ![i] =
// /\ \lnot logOk // m.mcommitIndex]
// /\ Reply([mtype |-> AppendEntriesResponse, // /\ Reply([mtype |-> AppendEntriesResponse,
// mterm |-> currentTerm[i], // mterm |-> currentTerm[i],
// msuccess |-> FALSE, // msuccess |-> TRUE,
// mmatchIndex |-> 0, // mmatchIndex |-> m.mprevLogIndex +
// msource |-> i, // Len(m.mentries),
// mdest |-> j], // msource |-> i,
// m) // mdest |-> j],
// /\ UNCHANGED <<serverVars, logVars>> // m)
// \/ \* return to follower state // /\ UNCHANGED <<serverVars, log>>
// /\ m.mterm = currentTerm[i] // \/ \* conflict: remove 1 entry
// /\ state[i] = Candidate // /\ m.mentries /= << >>
// /\ state' = [state EXCEPT ![i] = Follower] // /\ Len(log[i]) >= index
// /\ UNCHANGED <<currentTerm, votedFor, logVars, messages>> // /\ log[i][index].term /= m.mentries[1].term
// \/ \* accept request // /\ LET new == [index2 \in 1..(Len(log[i]) - 1) |->
// /\ m.mterm = currentTerm[i] // log[i][index2]]
// /\ state[i] = Follower // IN log' = [log EXCEPT ![i] = new]
// /\ logOk // /\ UNCHANGED <<serverVars, commitIndex, messages>>
// /\ LET index == m.mprevLogIndex + 1 // \/ \* no conflict: append entry
// IN \/ \* already done with request // /\ m.mentries /= << >>
// /\ \/ m.mentries = << >> // /\ Len(log[i]) = m.mprevLogIndex
// \/ /\ m.mentries /= << >> // /\ log' = [log EXCEPT ![i] =
// /\ Len(log[i]) >= index // Append(log[i], m.mentries[1])]
// /\ log[i][index].term = m.mentries[1].term // /\ UNCHANGED <<serverVars, commitIndex, messages>>
// \* This could make our commitIndex decrease (for // /\ UNCHANGED <<candidateVars, leaderVars>>
// \* example if we process an old, duplicated request), //
// \* but that doesn't really affect anything. int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) {}
// /\ commitIndex' = [commitIndex EXCEPT ![i] =
// m.mcommitIndex]
// /\ Reply([mtype |-> AppendEntriesResponse,
// mterm |-> currentTerm[i],
// msuccess |-> TRUE,
// mmatchIndex |-> m.mprevLogIndex +
// Len(m.mentries),
// msource |-> i,
// mdest |-> j],
// m)
// /\ UNCHANGED <<serverVars, log>>
// \/ \* conflict: remove 1 entry
// /\ m.mentries /= << >>
// /\ Len(log[i]) >= index
// /\ log[i][index].term /= m.mentries[1].term
// /\ LET new == [index2 \in 1..(Len(log[i]) - 1) |->
// log[i][index2]]
// IN log' = [log EXCEPT ![i] = new]
// /\ UNCHANGED <<serverVars, commitIndex, messages>>
// \/ \* no conflict: append entry
// /\ m.mentries /= << >>
// /\ Len(log[i]) = m.mprevLogIndex
// /\ log' = [log EXCEPT ![i] =
// Append(log[i], m.mentries[1])]
// /\ UNCHANGED <<serverVars, commitIndex, messages>>
// /\ UNCHANGED <<candidateVars, leaderVars>>
//
}

27
source/libs/sync/src/syncAppendEntriesReply.c
View File

@ -15,17 +15,16 @@
#include "syncAppendEntriesReply.h" #include "syncAppendEntriesReply.h"
int32_t syncNodeOnAppendEntriesReplyCb(SSyncNode* ths, SyncAppendEntriesReply* pMsg) { // TLA+ Spec
// TLA+ Spec // HandleAppendEntriesResponse(i, j, m) ==
// HandleAppendEntriesResponse(i, j, m) == // /\ m.mterm = currentTerm[i]
// /\ m.mterm = currentTerm[i] // /\ \/ /\ m.msuccess \* successful
// /\ \/ /\ m.msuccess \* successful // /\ nextIndex' = [nextIndex EXCEPT ![i][j] = m.mmatchIndex + 1]
// /\ nextIndex' = [nextIndex EXCEPT ![i][j] = m.mmatchIndex + 1] // /\ matchIndex' = [matchIndex EXCEPT ![i][j] = m.mmatchIndex]
// /\ matchIndex' = [matchIndex EXCEPT ![i][j] = m.mmatchIndex] // \/ /\ \lnot m.msuccess \* not successful
// \/ /\ \lnot m.msuccess \* not successful // /\ nextIndex' = [nextIndex EXCEPT ![i][j] =
// /\ nextIndex' = [nextIndex EXCEPT ![i][j] = // Max({nextIndex[i][j] - 1, 1})]
// Max({nextIndex[i][j] - 1, 1})] // /\ UNCHANGED <<matchIndex>>
// /\ UNCHANGED <<matchIndex>> // /\ Discard(m)
// /\ Discard(m) // /\ UNCHANGED <<serverVars, candidateVars, logVars, elections>>
// /\ UNCHANGED <<serverVars, candidateVars, logVars, elections>> int32_t syncNodeOnAppendEntriesReplyCb(SSyncNode* ths, SyncAppendEntriesReply* pMsg) {}
}

29
source/libs/sync/src/syncElection.c
View File

@ -14,7 +14,32 @@
*/ */
#include "syncElection.h" #include "syncElection.h"
#include "syncMessage.h"
void syncNodeElect(SSyncNode* pSyncNode) {} int32_t syncNodeElect(SSyncNode* pSyncNode) {
// start election
syncNodeRequestVotePeers(pSyncNode);
}
void syncNodeRequestVotePeers(SSyncNode* pSyncNode) {} // TLA+ Spec
// RequestVote(i, j) ==
// /\ state[i] = Candidate
// /\ j \notin votesResponded[i]
// /\ Send([mtype |-> RequestVoteRequest,
// mterm |-> currentTerm[i],
// mlastLogTerm |-> LastTerm(log[i]),
// mlastLogIndex |-> Len(log[i]),
// msource |-> i,
// mdest |-> j])
// /\ UNCHANGED <<serverVars, candidateVars, leaderVars, logVars>>
int32_t syncNodeRequestVotePeers(SSyncNode* pSyncNode) {}
int32_t syncNodeRequestVote(SSyncNode* pSyncNode, const SRaftId* destRaftId, const SyncRequestVote* pMsg) {
sTrace("syncNodeRequestVote pSyncNode:%p ", pSyncNode);
int32_t ret = 0;
SRpcMsg rpcMsg;
syncRequestVote2RpcMsg(pMsg, &rpcMsg);
syncNodeSendMsgById(destRaftId, pSyncNode, &rpcMsg);
return ret;
}

1
source/libs/sync/src/syncEnv.c
View File

@ -28,6 +28,7 @@ static void doSyncEnvStopTimer(SSyncEnv *pSyncEnv, tmr_h *pTimer);
int32_t syncEnvStart() { int32_t syncEnvStart() {
int32_t ret; int32_t ret;
srand(time(NULL));
gSyncEnv = (SSyncEnv *)malloc(sizeof(SSyncEnv)); gSyncEnv = (SSyncEnv *)malloc(sizeof(SSyncEnv));
assert(gSyncEnv != NULL); assert(gSyncEnv != NULL);
ret = doSyncEnvStart(gSyncEnv); ret = doSyncEnvStart(gSyncEnv);

1
source/libs/sync/src/syncIO.c
View File

@ -44,6 +44,7 @@ int32_t syncIOStart(char *host, uint16_t port) {
gSyncIO = syncIOCreate(host, port); gSyncIO = syncIOCreate(host, port);
assert(gSyncIO != NULL); assert(gSyncIO != NULL);
srand(time(NULL));
int32_t ret = syncIOStartInternal(gSyncIO); int32_t ret = syncIOStartInternal(gSyncIO);
assert(ret == 0); assert(ret == 0);

205
source/libs/sync/src/syncMain.c
View File

@ -15,25 +15,25 @@
#include <stdint.h> #include <stdint.h>
#include "sync.h" #include "sync.h"
#include "syncAppendEntries.h"
#include "syncAppendEntriesReply.h"
#include "syncEnv.h" #include "syncEnv.h"
#include "syncInt.h" #include "syncInt.h"
#include "syncRaft.h" #include "syncRaft.h"
#include "syncRequestVote.h"
#include "syncRequestVoteReply.h"
#include "syncTimeout.h"
#include "syncUtil.h" #include "syncUtil.h"
static int32_t tsNodeRefId = -1; static int32_t tsNodeRefId = -1;
// ------ local funciton --------- // ------ local funciton ---------
static int32_t syncNodeSendMsgById(const SRaftId* destRaftId, SSyncNode* pSyncNode, SRpcMsg* pMsg);
static int32_t syncNodeSendMsgByInfo(const SNodeInfo* nodeInfo, SSyncNode* pSyncNode, SRpcMsg* pMsg);
static void syncNodeEqPingTimer(void* param, void* tmrId); static void syncNodeEqPingTimer(void* param, void* tmrId);
static void syncNodeEqElectTimer(void* param, void* tmrId); static void syncNodeEqElectTimer(void* param, void* tmrId);
static void syncNodeEqHeartbeatTimer(void* param, void* tmrId); static void syncNodeEqHeartbeatTimer(void* param, void* tmrId);
static int32_t syncNodePing(SSyncNode* pSyncNode, const SRaftId* destRaftId, SyncPing* pMsg);
static int32_t syncNodeOnPingCb(SSyncNode* ths, SyncPing* pMsg); static int32_t syncNodeOnPingCb(SSyncNode* ths, SyncPing* pMsg);
static int32_t syncNodeOnPingReplyCb(SSyncNode* ths, SyncPingReply* pMsg); static int32_t syncNodeOnPingReplyCb(SSyncNode* ths, SyncPingReply* pMsg);
static int32_t syncNodeOnTimeoutCb(SSyncNode* ths, SyncTimeout* pMsg);
static void syncNodeBecomeFollower(SSyncNode* pSyncNode); static void syncNodeBecomeFollower(SSyncNode* pSyncNode);
static void syncNodeBecomeLeader(SSyncNode* pSyncNode); static void syncNodeBecomeLeader(SSyncNode* pSyncNode);
@ -41,9 +41,6 @@ static void syncNodeFollower2Candidate(SSyncNode* pSyncNode);
static void syncNodeCandidate2Leader(SSyncNode* pSyncNode); static void syncNodeCandidate2Leader(SSyncNode* pSyncNode);
static void syncNodeLeader2Follower(SSyncNode* pSyncNode); static void syncNodeLeader2Follower(SSyncNode* pSyncNode);
static void syncNodeCandidate2Follower(SSyncNode* pSyncNode); static void syncNodeCandidate2Follower(SSyncNode* pSyncNode);
void syncNodeRequestVotePeers(SSyncNode* pSyncNode);
void syncNodeAppendEntriesPeers(SSyncNode* pSyncNode);
// --------------------------------- // ---------------------------------
int32_t syncInit() { int32_t syncInit() {
@ -98,6 +95,7 @@ SSyncNode* syncNodeOpen(const SSyncInfo* pSyncInfo) {
pSyncNode->state = TAOS_SYNC_STATE_FOLLOWER; pSyncNode->state = TAOS_SYNC_STATE_FOLLOWER;
syncUtilnodeInfo2raftId(&pSyncNode->me, pSyncNode->vgId, &pSyncNode->raftId); syncUtilnodeInfo2raftId(&pSyncNode->me, pSyncNode->vgId, &pSyncNode->raftId);
// init ping timer
pSyncNode->pPingTimer = NULL; pSyncNode->pPingTimer = NULL;
pSyncNode->pingTimerMS = PING_TIMER_MS; pSyncNode->pingTimerMS = PING_TIMER_MS;
atomic_store_64(&pSyncNode->pingTimerLogicClock, 0); atomic_store_64(&pSyncNode->pingTimerLogicClock, 0);
@ -105,6 +103,22 @@ SSyncNode* syncNodeOpen(const SSyncInfo* pSyncInfo) {
pSyncNode->FpPingTimer = syncNodeEqPingTimer; pSyncNode->FpPingTimer = syncNodeEqPingTimer;
pSyncNode->pingTimerCounter = 0; pSyncNode->pingTimerCounter = 0;
// init elect timer
pSyncNode->pElectTimer = NULL;
pSyncNode->electTimerMS = syncUtilElectRandomMS();
atomic_store_64(&pSyncNode->electTimerLogicClock, 0);
atomic_store_64(&pSyncNode->electTimerLogicClockUser, 0);
pSyncNode->FpElectTimer = syncNodeEqElectTimer;
pSyncNode->electTimerCounter = 0;
// init heartbeat timer
pSyncNode->pHeartbeatTimer = NULL;
pSyncNode->heartbeatTimerMS = HEARTBEAT_TIMER_MS;
atomic_store_64(&pSyncNode->heartbeatTimerLogicClock, 0);
atomic_store_64(&pSyncNode->heartbeatTimerLogicClockUser, 0);
pSyncNode->FpHeartbeatTimer = syncNodeEqHeartbeatTimer;
pSyncNode->heartbeatTimerCounter = 0;
pSyncNode->FpOnPing = syncNodeOnPingCb; pSyncNode->FpOnPing = syncNodeOnPingCb;
pSyncNode->FpOnPingReply = syncNodeOnPingReplyCb; pSyncNode->FpOnPingReply = syncNodeOnPingReplyCb;
pSyncNode->FpOnRequestVote = syncNodeOnRequestVoteCb; pSyncNode->FpOnRequestVote = syncNodeOnRequestVoteCb;
@ -121,6 +135,48 @@ void syncNodeClose(SSyncNode* pSyncNode) {
free(pSyncNode); free(pSyncNode);
} }
int32_t syncNodeSendMsgById(const SRaftId* destRaftId, SSyncNode* pSyncNode, SRpcMsg* pMsg) {
SEpSet epSet;
syncUtilraftId2EpSet(destRaftId, &epSet);
pSyncNode->FpSendMsg(pSyncNode->rpcClient, &epSet, pMsg);
return 0;
}
int32_t syncNodeSendMsgByInfo(const SNodeInfo* nodeInfo, SSyncNode* pSyncNode, SRpcMsg* pMsg) {
SEpSet epSet;
syncUtilnodeInfo2EpSet(nodeInfo, &epSet);
pSyncNode->FpSendMsg(pSyncNode->rpcClient, &epSet, pMsg);
return 0;
}
int32_t syncNodePing(SSyncNode* pSyncNode, const SRaftId* destRaftId, SyncPing* pMsg) {
sTrace("syncNodePing pSyncNode:%p ", pSyncNode);
int32_t ret = 0;
SRpcMsg rpcMsg;
syncPing2RpcMsg(pMsg, &rpcMsg);
syncNodeSendMsgById(destRaftId, pSyncNode, &rpcMsg);
{
cJSON* pJson = syncPing2Json(pMsg);
char* serialized = cJSON_Print(pJson);
sTrace("syncNodePing pMsg:%s ", serialized);
free(serialized);
cJSON_Delete(pJson);
}
{
SyncPing* pMsg2 = rpcMsg.pCont;
cJSON* pJson = syncPing2Json(pMsg2);
char* serialized = cJSON_Print(pJson);
sTrace("syncNodePing rpcMsg.pCont:%s ", serialized);
free(serialized);
cJSON_Delete(pJson);
}
return ret;
}
void syncNodePingAll(SSyncNode* pSyncNode) { void syncNodePingAll(SSyncNode* pSyncNode) {
sTrace("syncNodePingAll pSyncNode:%p ", pSyncNode); sTrace("syncNodePingAll pSyncNode:%p ", pSyncNode);
int32_t ret = 0; int32_t ret = 0;
@ -157,7 +213,6 @@ void syncNodePingSelf(SSyncNode* pSyncNode) {
int32_t syncNodeStartPingTimer(SSyncNode* pSyncNode) { int32_t syncNodeStartPingTimer(SSyncNode* pSyncNode) {
atomic_store_64(&pSyncNode->pingTimerLogicClock, pSyncNode->pingTimerLogicClockUser); atomic_store_64(&pSyncNode->pingTimerLogicClock, pSyncNode->pingTimerLogicClockUser);
pSyncNode->pingTimerMS = PING_TIMER_MS; pSyncNode->pingTimerMS = PING_TIMER_MS;
if (pSyncNode->pPingTimer == NULL) { if (pSyncNode->pPingTimer == NULL) {
pSyncNode->pPingTimer = pSyncNode->pPingTimer =
taosTmrStart(pSyncNode->FpPingTimer, pSyncNode->pingTimerMS, pSyncNode, gSyncEnv->pTimerManager); taosTmrStart(pSyncNode->FpPingTimer, pSyncNode->pingTimerMS, pSyncNode, gSyncEnv->pTimerManager);
@ -165,7 +220,6 @@ int32_t syncNodeStartPingTimer(SSyncNode* pSyncNode) {
taosTmrReset(pSyncNode->FpPingTimer, pSyncNode->pingTimerMS, pSyncNode, gSyncEnv->pTimerManager, taosTmrReset(pSyncNode->FpPingTimer, pSyncNode->pingTimerMS, pSyncNode, gSyncEnv->pTimerManager,
&pSyncNode->pPingTimer); &pSyncNode->pPingTimer);
} }
return 0; return 0;
} }
@ -175,7 +229,9 @@ int32_t syncNodeStopPingTimer(SSyncNode* pSyncNode) {
return 0; return 0;
} }
int32_t syncNodeStartElectTimer(SSyncNode* pSyncNode) { int32_t syncNodeStartElectTimer(SSyncNode* pSyncNode, int32_t ms) {
pSyncNode->electTimerMS = ms;
atomic_store_64(&pSyncNode->electTimerLogicClock, pSyncNode->electTimerLogicClockUser);
if (pSyncNode->pElectTimer == NULL) { if (pSyncNode->pElectTimer == NULL) {
pSyncNode->pElectTimer = pSyncNode->pElectTimer =
taosTmrStart(pSyncNode->FpElectTimer, pSyncNode->electTimerMS, pSyncNode, gSyncEnv->pTimerManager); taosTmrStart(pSyncNode->FpElectTimer, pSyncNode->electTimerMS, pSyncNode, gSyncEnv->pTimerManager);
@ -183,18 +239,23 @@ int32_t syncNodeStartElectTimer(SSyncNode* pSyncNode) {
taosTmrReset(pSyncNode->FpElectTimer, pSyncNode->electTimerMS, pSyncNode, gSyncEnv->pTimerManager, taosTmrReset(pSyncNode->FpElectTimer, pSyncNode->electTimerMS, pSyncNode, gSyncEnv->pTimerManager,
&pSyncNode->pElectTimer); &pSyncNode->pElectTimer);
} }
atomic_store_8(&pSyncNode->electTimerEnable, 1);
return 0; return 0;
} }
int32_t syncNodeStopElectTimer(SSyncNode* pSyncNode) { int32_t syncNodeStopElectTimer(SSyncNode* pSyncNode) {
atomic_store_8(&pSyncNode->electTimerEnable, 0); atomic_add_fetch_64(&pSyncNode->electTimerLogicClockUser, 1);
pSyncNode->electTimerMS = TIMER_MAX_MS; pSyncNode->electTimerMS = TIMER_MAX_MS;
return 0; return 0;
} }
int32_t syncNodeRestartElectTimer(SSyncNode* pSyncNode, int32_t ms) {
syncNodeStopElectTimer(pSyncNode);
syncNodeStartElectTimer(pSyncNode, ms);
return 0;
}
int32_t syncNodeStartHeartbeatTimer(SSyncNode* pSyncNode) { int32_t syncNodeStartHeartbeatTimer(SSyncNode* pSyncNode) {
atomic_store_64(&pSyncNode->heartbeatTimerLogicClock, pSyncNode->heartbeatTimerLogicClockUser);
if (pSyncNode->pHeartbeatTimer == NULL) { if (pSyncNode->pHeartbeatTimer == NULL) {
pSyncNode->pHeartbeatTimer = pSyncNode->pHeartbeatTimer =
taosTmrStart(pSyncNode->FpHeartbeatTimer, pSyncNode->heartbeatTimerMS, pSyncNode, gSyncEnv->pTimerManager); taosTmrStart(pSyncNode->FpHeartbeatTimer, pSyncNode->heartbeatTimerMS, pSyncNode, gSyncEnv->pTimerManager);
@ -202,60 +263,16 @@ int32_t syncNodeStartHeartbeatTimer(SSyncNode* pSyncNode) {
taosTmrReset(pSyncNode->FpHeartbeatTimer, pSyncNode->heartbeatTimerMS, pSyncNode, gSyncEnv->pTimerManager, taosTmrReset(pSyncNode->FpHeartbeatTimer, pSyncNode->heartbeatTimerMS, pSyncNode, gSyncEnv->pTimerManager,
&pSyncNode->pHeartbeatTimer); &pSyncNode->pHeartbeatTimer);
} }
atomic_store_8(&pSyncNode->heartbeatTimerEnable, 1);
return 0; return 0;
} }
int32_t syncNodeStopHeartbeatTimer(SSyncNode* pSyncNode) { int32_t syncNodeStopHeartbeatTimer(SSyncNode* pSyncNode) {
atomic_store_8(&pSyncNode->heartbeatTimerEnable, 0); atomic_add_fetch_64(&pSyncNode->heartbeatTimerLogicClockUser, 1);
pSyncNode->heartbeatTimerMS = TIMER_MAX_MS; pSyncNode->heartbeatTimerMS = TIMER_MAX_MS;
return 0; return 0;
} }
// ------ local funciton --------- // ------ local funciton ---------
static int32_t syncNodePing(SSyncNode* pSyncNode, const SRaftId* destRaftId, SyncPing* pMsg) {
sTrace("syncNodePing pSyncNode:%p ", pSyncNode);
int32_t ret = 0;
SRpcMsg rpcMsg;
syncPing2RpcMsg(pMsg, &rpcMsg);
syncNodeSendMsgById(destRaftId, pSyncNode, &rpcMsg);
{
cJSON* pJson = syncPing2Json(pMsg);
char* serialized = cJSON_Print(pJson);
sTrace("syncNodePing pMsg:%s ", serialized);
free(serialized);
cJSON_Delete(pJson);
}
{
SyncPing* pMsg2 = rpcMsg.pCont;
cJSON* pJson = syncPing2Json(pMsg2);
char* serialized = cJSON_Print(pJson);
sTrace("syncNodePing rpcMsg.pCont:%s ", serialized);
free(serialized);
cJSON_Delete(pJson);
}
return ret;
}
static int32_t syncNodeSendMsgById(const SRaftId* destRaftId, SSyncNode* pSyncNode, SRpcMsg* pMsg) {
SEpSet epSet;
syncUtilraftId2EpSet(destRaftId, &epSet);
pSyncNode->FpSendMsg(pSyncNode->rpcClient, &epSet, pMsg);
return 0;
}
static int32_t syncNodeSendMsgByInfo(const SNodeInfo* nodeInfo, SSyncNode* pSyncNode, SRpcMsg* pMsg) {
SEpSet epSet;
syncUtilnodeInfo2EpSet(nodeInfo, &epSet);
pSyncNode->FpSendMsg(pSyncNode->rpcClient, &epSet, pMsg);
return 0;
}
static int32_t syncNodeOnPingCb(SSyncNode* ths, SyncPing* pMsg) { static int32_t syncNodeOnPingCb(SSyncNode* ths, SyncPing* pMsg) {
int32_t ret = 0; int32_t ret = 0;
sTrace("<-- syncNodeOnPingCb -->"); sTrace("<-- syncNodeOnPingCb -->");
@ -291,45 +308,19 @@ static int32_t syncNodeOnPingReplyCb(SSyncNode* ths, SyncPingReply* pMsg) {
return ret; return ret;
} }
static int32_t syncNodeOnTimeoutCb(SSyncNode* ths, SyncTimeout* pMsg) {
int32_t ret = 0;
sTrace("<-- syncNodeOnTimeoutCb -->");
{
cJSON* pJson = syncTimeout2Json(pMsg);
char* serialized = cJSON_Print(pJson);
sTrace("process syncMessage recv: syncNodeOnTimeoutCb pMsg:%s ", serialized);
free(serialized);
cJSON_Delete(pJson);
}
if (pMsg->timeoutType == SYNC_TIMEOUT_PING) {
if (atomic_load_64(&ths->pingTimerLogicClockUser) <= pMsg->logicClock) {
++(ths->pingTimerCounter);
syncNodePingAll(ths);
}
} else if (pMsg->timeoutType == SYNC_TIMEOUT_ELECTION) {
} else if (pMsg->timeoutType == SYNC_TIMEOUT_HEARTBEAT) {
} else {
}
return ret;
}
static void syncNodeEqPingTimer(void* param, void* tmrId) { static void syncNodeEqPingTimer(void* param, void* tmrId) {
SSyncNode* pSyncNode = (SSyncNode*)param; SSyncNode* pSyncNode = (SSyncNode*)param;
if (atomic_load_64(&pSyncNode->pingTimerLogicClockUser) <= atomic_load_64(&pSyncNode->pingTimerLogicClock)) { if (atomic_load_64(&pSyncNode->pingTimerLogicClockUser) <= atomic_load_64(&pSyncNode->pingTimerLogicClock)) {
// pSyncNode->pingTimerMS += 100; SyncTimeout* pSyncMsg = syncTimeoutBuild2(SYNC_TIMEOUT_PING, atomic_load_64(&pSyncNode->pingTimerLogicClock),
pSyncNode->pingTimerMS, pSyncNode);
SyncTimeout* pSyncMsg = SRpcMsg rpcMsg;
syncTimeoutBuild2(SYNC_TIMEOUT_PING, atomic_load_64(&pSyncNode->pingTimerLogicClock), pSyncNode);
SRpcMsg rpcMsg;
syncTimeout2RpcMsg(pSyncMsg, &rpcMsg); syncTimeout2RpcMsg(pSyncMsg, &rpcMsg);
pSyncNode->FpEqMsg(pSyncNode->queue, &rpcMsg); pSyncNode->FpEqMsg(pSyncNode->queue, &rpcMsg);
syncTimeoutDestroy(pSyncMsg); syncTimeoutDestroy(pSyncMsg);
// reset timer ms
// pSyncNode->pingTimerMS += 100;
taosTmrReset(syncNodeEqPingTimer, pSyncNode->pingTimerMS, pSyncNode, &gSyncEnv->pTimerManager, taosTmrReset(syncNodeEqPingTimer, pSyncNode->pingTimerMS, pSyncNode, &gSyncEnv->pTimerManager,
&pSyncNode->pPingTimer); &pSyncNode->pPingTimer);
} else { } else {
@ -338,18 +329,38 @@ static void syncNodeEqPingTimer(void* param, void* tmrId) {
} }
} }
static void syncNodeEqElectTimer(void* param, void* tmrId) {} static void syncNodeEqElectTimer(void* param, void* tmrId) {
SSyncNode* pSyncNode = (SSyncNode*)param;
if (atomic_load_64(&pSyncNode->electTimerLogicClockUser) <= atomic_load_64(&pSyncNode->electTimerLogicClock)) {
SyncTimeout* pSyncMsg = syncTimeoutBuild2(SYNC_TIMEOUT_ELECTION, atomic_load_64(&pSyncNode->electTimerLogicClock),
pSyncNode->electTimerMS, pSyncNode);
SRpcMsg rpcMsg;
syncTimeout2RpcMsg(pSyncMsg, &rpcMsg);
pSyncNode->FpEqMsg(pSyncNode->queue, &rpcMsg);
syncTimeoutDestroy(pSyncMsg);
// reset timer ms
pSyncNode->electTimerMS = syncUtilElectRandomMS();
taosTmrReset(syncNodeEqPingTimer, pSyncNode->pingTimerMS, pSyncNode, &gSyncEnv->pTimerManager,
&pSyncNode->pPingTimer);
} else {
sTrace("syncNodeEqElectTimer: electTimerLogicClock:%lu, electTimerLogicClockUser:%lu",
pSyncNode->electTimerLogicClock, pSyncNode->electTimerLogicClockUser);
}
}
static void syncNodeEqHeartbeatTimer(void* param, void* tmrId) {} static void syncNodeEqHeartbeatTimer(void* param, void* tmrId) {}
static void syncNodeBecomeFollower(SSyncNode* pSyncNode) { static void syncNodeBecomeFollower(SSyncNode* pSyncNode) {
if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) { if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
pSyncNode->leaderCache.addr = 0; pSyncNode->leaderCache = EMPTY_RAFT_ID;
pSyncNode->leaderCache.vgId = 0;
} }
syncNodeStopHeartbeatTimer(pSyncNode); syncNodeStopHeartbeatTimer(pSyncNode);
syncNodeStartElectTimer(pSyncNode); int32_t electMS = syncUtilElectRandomMS();
syncNodeStartElectTimer(pSyncNode, electMS);
} }
static void syncNodeBecomeLeader(SSyncNode* pSyncNode) { static void syncNodeBecomeLeader(SSyncNode* pSyncNode) {
@ -375,7 +386,3 @@ static void syncNodeCandidate2Leader(SSyncNode* pSyncNode) {}
static void syncNodeLeader2Follower(SSyncNode* pSyncNode) {} static void syncNodeLeader2Follower(SSyncNode* pSyncNode) {}
static void syncNodeCandidate2Follower(SSyncNode* pSyncNode) {} static void syncNodeCandidate2Follower(SSyncNode* pSyncNode) {}
void syncNodeRequestVotePeers(SSyncNode* pSyncNode) {}
void syncNodeAppendEntriesPeers(SSyncNode* pSyncNode) {}

6
source/libs/sync/src/syncMessage.c
View File

@ -18,8 +18,6 @@
#include "syncUtil.h" #include "syncUtil.h"
#include "tcoding.h" #include "tcoding.h"
void onMessage(SRaft* pRaft, void* pMsg) {}
// --------------------------------------------- // ---------------------------------------------
cJSON* syncRpcMsg2Json(SRpcMsg* pRpcMsg) { cJSON* syncRpcMsg2Json(SRpcMsg* pRpcMsg) {
cJSON* pRoot; cJSON* pRoot;
@ -125,6 +123,7 @@ cJSON* syncTimeout2Json(const SyncTimeout* pMsg) {
cJSON_AddNumberToObject(pRoot, "timeoutType", pMsg->timeoutType); cJSON_AddNumberToObject(pRoot, "timeoutType", pMsg->timeoutType);
snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->logicClock); snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->logicClock);
cJSON_AddStringToObject(pRoot, "logicClock", u64buf); cJSON_AddStringToObject(pRoot, "logicClock", u64buf);
cJSON_AddNumberToObject(pRoot, "timerMS", pMsg->timerMS);
snprintf(u64buf, sizeof(u64buf), "%p", pMsg->data); snprintf(u64buf, sizeof(u64buf), "%p", pMsg->data);
cJSON_AddStringToObject(pRoot, "data", u64buf); cJSON_AddStringToObject(pRoot, "data", u64buf);
@ -133,10 +132,11 @@ cJSON* syncTimeout2Json(const SyncTimeout* pMsg) {
return pJson; return pJson;
} }
SyncTimeout* syncTimeoutBuild2(ESyncTimeoutType timeoutType, uint64_t logicClock, void* data) { SyncTimeout* syncTimeoutBuild2(ESyncTimeoutType timeoutType, uint64_t logicClock, int32_t timerMS, void* data) {
SyncTimeout* pMsg = syncTimeoutBuild(); SyncTimeout* pMsg = syncTimeoutBuild();
pMsg->timeoutType = timeoutType; pMsg->timeoutType = timeoutType;
pMsg->logicClock = logicClock; pMsg->logicClock = logicClock;
pMsg->timerMS = timerMS;
pMsg->data = data; pMsg->data = data;
return pMsg; return pMsg;
} }

123
source/libs/sync/src/syncRaftStore.c
View File

@ -16,8 +16,11 @@
#include "syncRaftStore.h" #include "syncRaftStore.h"
#include "cJSON.h" #include "cJSON.h"
// to complie success: FileIO interface is modified // private function
static int32_t raftStoreInit(SRaftStore *pRaftStore);
static bool raftStoreFileExist(char *path);
// public function
SRaftStore *raftStoreOpen(const char *path) { SRaftStore *raftStoreOpen(const char *path) {
int32_t ret; int32_t ret;
@ -137,121 +140,3 @@ void raftStorePrint(SRaftStore *pRaftStore) {
raftStoreSerialize(pRaftStore, storeBuf, sizeof(storeBuf)); raftStoreSerialize(pRaftStore, storeBuf, sizeof(storeBuf));
printf("%s\n", storeBuf); printf("%s\n", storeBuf);
} }
#if 0
SRaftStore *raftStoreOpen(const char *path) {
int32_t ret;
SRaftStore *pRaftStore = malloc(sizeof(SRaftStore));
if (pRaftStore == NULL) {
sError("raftStoreOpen malloc error");
return NULL;
}
memset(pRaftStore, 0, sizeof(*pRaftStore));
snprintf(pRaftStore->path, sizeof(pRaftStore->path), "%s", path);
char storeBuf[RAFT_STORE_BLOCK_SIZE];
memset(storeBuf, 0, sizeof(storeBuf));
if (!raftStoreFileExist(pRaftStore->path)) {
ret = raftStoreInit(pRaftStore);
assert(ret == 0);
}
pRaftStore->fd = taosOpenFileReadWrite(pRaftStore->path);
if (pRaftStore->fd < 0) {
return NULL;
}
int len = taosReadFile(pRaftStore->fd, storeBuf, sizeof(storeBuf));
assert(len == RAFT_STORE_BLOCK_SIZE);
ret = raftStoreDeserialize(pRaftStore, storeBuf, len);
assert(ret == 0);
return pRaftStore;
}
static int32_t raftStoreInit(SRaftStore *pRaftStore) {
pRaftStore->fd = taosOpenFileCreateWrite(pRaftStore->path);
if (pRaftStore->fd < 0) {
return -1;
}
pRaftStore->currentTerm = 0;
pRaftStore->voteFor.addr = 0;
pRaftStore->voteFor.vgId = 0;
int32_t ret = raftStorePersist(pRaftStore);
assert(ret == 0);
taosCloseFile(pRaftStore->fd);
return 0;
}
int32_t raftStoreClose(SRaftStore *pRaftStore) {
taosCloseFile(pRaftStore->fd);
free(pRaftStore);
return 0;
}
int32_t raftStorePersist(SRaftStore *pRaftStore) {
int32_t ret;
char storeBuf[RAFT_STORE_BLOCK_SIZE];
ret = raftStoreSerialize(pRaftStore, storeBuf, sizeof(storeBuf));
assert(ret == 0);
taosLSeekFile(pRaftStore->fd, 0, SEEK_SET);
ret = taosWriteFile(pRaftStore->fd, storeBuf, sizeof(storeBuf));
assert(ret == RAFT_STORE_BLOCK_SIZE);
fsync(pRaftStore->fd);
return 0;
}
static bool raftStoreFileExist(char *path) { return taosStatFile(path, NULL, NULL) >= 0; }
int32_t raftStoreSerialize(SRaftStore *pRaftStore, char *buf, size_t len) {
cJSON *pRoot = cJSON_CreateObject();
cJSON_AddNumberToObject(pRoot, "current_term", pRaftStore->currentTerm);
cJSON_AddNumberToObject(pRoot, "vote_for_addr", pRaftStore->voteFor.addr);
cJSON_AddNumberToObject(pRoot, "vote_for_vgid", pRaftStore->voteFor.vgId);
char *serialized = cJSON_Print(pRoot);
int len2 = strlen(serialized);
assert(len2 < len);
memset(buf, 0, len);
snprintf(buf, len, "%s", serialized);
free(serialized);
cJSON_Delete(pRoot);
return 0;
}
int32_t raftStoreDeserialize(SRaftStore *pRaftStore, char *buf, size_t len) {
assert(len > 0 && len <= RAFT_STORE_BLOCK_SIZE);
cJSON *pRoot = cJSON_Parse(buf);
cJSON *pCurrentTerm = cJSON_GetObjectItem(pRoot, "current_term");
pRaftStore->currentTerm = pCurrentTerm->valueint;
cJSON *pVoteForAddr = cJSON_GetObjectItem(pRoot, "vote_for_addr");
pRaftStore->voteFor.addr = pVoteForAddr->valueint;
cJSON *pVoteForVgid = cJSON_GetObjectItem(pRoot, "vote_for_vgid");
pRaftStore->voteFor.vgId = pVoteForVgid->valueint;
cJSON_Delete(pRoot);
return 0;
}
void raftStorePrint(SRaftStore *pRaftStore) {
char storeBuf[RAFT_STORE_BLOCK_SIZE];
raftStoreSerialize(pRaftStore, storeBuf, sizeof(storeBuf));
printf("%s\n", storeBuf);
}
#endif

37
source/libs/sync/src/syncReplication.c
View File

@ -14,5 +14,40 @@
*/ */
#include "syncReplication.h" #include "syncReplication.h"
#include "syncMessage.h"
void syncNodeAppendEntriesPeers(SSyncNode* pSyncNode) {} // TLA+ Spec
// AppendEntries(i, j) ==
// /\ i /= j
// /\ state[i] = Leader
// /\ LET prevLogIndex == nextIndex[i][j] - 1
// prevLogTerm == IF prevLogIndex > 0 THEN
// log[i][prevLogIndex].term
// ELSE
// 0
// \* Send up to 1 entry, constrained by the end of the log.
// lastEntry == Min({Len(log[i]), nextIndex[i][j]})
// entries == SubSeq(log[i], nextIndex[i][j], lastEntry)
// IN Send([mtype |-> AppendEntriesRequest,
// mterm |-> currentTerm[i],
// mprevLogIndex |-> prevLogIndex,
// mprevLogTerm |-> prevLogTerm,
// mentries |-> entries,
// \* mlog is used as a history variable for the proof.
// \* It would not exist in a real implementation.
// mlog |-> log[i],
// mcommitIndex |-> Min({commitIndex[i], lastEntry}),
// msource |-> i,
// mdest |-> j])
// /\ UNCHANGED <<serverVars, candidateVars, leaderVars, logVars>>
int32_t syncNodeAppendEntriesPeers(SSyncNode* pSyncNode) {}
int32_t syncNodeAppendEntries(SSyncNode* pSyncNode, const SRaftId* destRaftId, const SyncAppendEntries* pMsg) {
sTrace("syncNodeAppendEntries pSyncNode:%p ", pSyncNode);
int32_t ret = 0;
SRpcMsg rpcMsg;
syncAppendEntries2RpcMsg(pMsg, &rpcMsg);
syncNodeSendMsgById(destRaftId, pSyncNode, &rpcMsg);
return ret;
}

59
source/libs/sync/src/syncRequestVote.c
View File

@ -15,40 +15,25 @@
#include "syncRequestVote.h" #include "syncRequestVote.h"
int32_t syncNodeRequestVote(SSyncNode* ths, const SyncRequestVote* pMsg) { // TLA+ Spec
// TLA+ Spec // HandleRequestVoteRequest(i, j, m) ==
// RequestVote(i, j) == // LET logOk == \/ m.mlastLogTerm > LastTerm(log[i])
// /\ state[i] = Candidate // \/ /\ m.mlastLogTerm = LastTerm(log[i])
// /\ j \notin votesResponded[i] // /\ m.mlastLogIndex >= Len(log[i])
// /\ Send([mtype |-> RequestVoteRequest, // grant == /\ m.mterm = currentTerm[i]
// mterm |-> currentTerm[i], // /\ logOk
// mlastLogTerm |-> LastTerm(log[i]), // /\ votedFor[i] \in {Nil, j}
// mlastLogIndex |-> Len(log[i]), // IN /\ m.mterm <= currentTerm[i]
// msource |-> i, // /\ \/ grant /\ votedFor' = [votedFor EXCEPT ![i] = j]
// mdest |-> j]) // \/ ~grant /\ UNCHANGED votedFor
// /\ UNCHANGED <<serverVars, candidateVars, leaderVars, logVars>> // /\ Reply([mtype |-> RequestVoteResponse,
} // mterm |-> currentTerm[i],
// mvoteGranted |-> grant,
int32_t syncNodeOnRequestVoteCb(SSyncNode* ths, SyncRequestVote* pMsg) { // \* mlog is used just for the `elections' history variable for
// TLA+ Spec // \* the proof. It would not exist in a real implementation.
// HandleRequestVoteRequest(i, j, m) == // mlog |-> log[i],
// LET logOk == \/ m.mlastLogTerm > LastTerm(log[i]) // msource |-> i,
// \/ /\ m.mlastLogTerm = LastTerm(log[i]) // mdest |-> j],
// /\ m.mlastLogIndex >= Len(log[i]) // m)
// grant == /\ m.mterm = currentTerm[i] // /\ UNCHANGED <<state, currentTerm, candidateVars, leaderVars, logVars>>
// /\ logOk int32_t syncNodeOnRequestVoteCb(SSyncNode* ths, SyncRequestVote* pMsg) {}
// /\ votedFor[i] \in {Nil, j}
// IN /\ m.mterm <= currentTerm[i]
// /\ \/ grant /\ votedFor' = [votedFor EXCEPT ![i] = j]
// \/ ~grant /\ UNCHANGED votedFor
// /\ Reply([mtype |-> RequestVoteResponse,
// mterm |-> currentTerm[i],
// mvoteGranted |-> grant,
// \* mlog is used just for the `elections' history variable for
// \* the proof. It would not exist in a real implementation.
// mlog |-> log[i],
// msource |-> i,
// mdest |-> j],
// m)
// /\ UNCHANGED <<state, currentTerm, candidateVars, leaderVars, logVars>>
}

35
source/libs/sync/src/syncRequestVoteReply.c
View File

@ -15,21 +15,20 @@
#include "syncRequestVoteReply.h" #include "syncRequestVoteReply.h"
int32_t syncNodeOnRequestVoteReplyCb(SSyncNode* ths, SyncRequestVoteReply* pMsg) { // TLA+ Spec
// TLA+ Spec // HandleRequestVoteResponse(i, j, m) ==
// HandleRequestVoteResponse(i, j, m) == // \* This tallies votes even when the current state is not Candidate, but
// \* This tallies votes even when the current state is not Candidate, but // \* they won't be looked at, so it doesn't matter.
// \* they won't be looked at, so it doesn't matter. // /\ m.mterm = currentTerm[i]
// /\ m.mterm = currentTerm[i] // /\ votesResponded' = [votesResponded EXCEPT ![i] =
// /\ votesResponded' = [votesResponded EXCEPT ![i] = // votesResponded[i] \cup {j}]
// votesResponded[i] \cup {j}] // /\ \/ /\ m.mvoteGranted
// /\ \/ /\ m.mvoteGranted // /\ votesGranted' = [votesGranted EXCEPT ![i] =
// /\ votesGranted' = [votesGranted EXCEPT ![i] = // votesGranted[i] \cup {j}]
// votesGranted[i] \cup {j}] // /\ voterLog' = [voterLog EXCEPT ![i] =
// /\ voterLog' = [voterLog EXCEPT ![i] = // voterLog[i] @@ (j :> m.mlog)]
// voterLog[i] @@ (j :> m.mlog)] // \/ /\ ~m.mvoteGranted
// \/ /\ ~m.mvoteGranted // /\ UNCHANGED <<votesGranted, voterLog>>
// /\ UNCHANGED <<votesGranted, voterLog>> // /\ Discard(m)
// /\ Discard(m) // /\ UNCHANGED <<serverVars, votedFor, leaderVars, logVars>>
// /\ UNCHANGED <<serverVars, votedFor, leaderVars, logVars>> int32_t syncNodeOnRequestVoteReplyCb(SSyncNode* ths, SyncRequestVoteReply* pMsg) {}
}

38
source/libs/sync/src/syncTimeout.c
View File

@ -14,5 +14,41 @@
*/ */
#include "syncTimeout.h" #include "syncTimeout.h"
#include "syncElection.h"
#include "syncReplication.h"
void onTimeout(SRaft *pRaft, void *pMsg) {} int32_t syncNodeOnTimeoutCb(SSyncNode* ths, SyncTimeout* pMsg) {
int32_t ret = 0;
sTrace("<-- syncNodeOnTimeoutCb -->");
{
cJSON* pJson = syncTimeout2Json(pMsg);
char* serialized = cJSON_Print(pJson);
sTrace("process syncMessage recv: syncNodeOnTimeoutCb pMsg:%s ", serialized);
free(serialized);
cJSON_Delete(pJson);
}
if (pMsg->timeoutType == SYNC_TIMEOUT_PING) {
if (atomic_load_64(&ths->pingTimerLogicClockUser) <= pMsg->logicClock) {
++(ths->pingTimerCounter);
syncNodePingAll(ths);
}
} else if (pMsg->timeoutType == SYNC_TIMEOUT_ELECTION) {
if (atomic_load_64(&ths->electTimerLogicClockUser) <= pMsg->logicClock) {
++(ths->electTimerCounter);
syncNodeElect(ths);
}
} else if (pMsg->timeoutType == SYNC_TIMEOUT_HEARTBEAT) {
if (atomic_load_64(&ths->heartbeatTimerLogicClockUser) <= pMsg->logicClock) {
++(ths->heartbeatTimerCounter);
syncNodeAppendEntriesPeers(ths);
}
} else {
sTrace("unknown timeoutType:%d", pMsg->timeoutType);
}
return ret;
}

7
source/libs/sync/src/syncUtil.c
View File

@ -17,6 +17,7 @@
#include <arpa/inet.h> #include <arpa/inet.h>
#include <netinet/in.h> #include <netinet/in.h>
#include <sys/socket.h> #include <sys/socket.h>
#include "syncEnv.h"
// ---- encode / decode // ---- encode / decode
uint64_t syncUtilAddr2U64(const char* host, uint16_t port) { uint64_t syncUtilAddr2U64(const char* host, uint16_t port) {
@ -91,3 +92,9 @@ void syncUtilbufCopyDeep(const SSyncBuffer* src, SSyncBuffer* dest) {
dest->data = malloc(dest->len); dest->data = malloc(dest->len);
memcpy(dest->data, src->data, dest->len); memcpy(dest->data, src->data, dest->len);
} }
// ---- misc ----
int32_t syncUtilRand(int32_t max) { return rand() % max; }
int32_t syncUtilElectRandomMS() { ELECT_TIMER_MS_MIN + syncUtilRand(ELECT_TIMER_MS_RANGE); }

16
source/libs/sync/test/syncIndexTest.cpp
View File

@ -13,17 +13,21 @@ void print(SHashObj *pNextIndex) {
} }
} }
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");
}
int main() { int main() {
// taosInitLog((char *)"syncTest.log", 100000, 10); // taosInitLog((char *)"syncTest.log", 100000, 10);
tsAsyncLog = 0; tsAsyncLog = 0;
sDebugFlag = 143 + 64; sDebugFlag = 143 + 64;
sTrace("sync log test: trace"); logTest();
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");
SRaftId me; SRaftId me;
SRaftId peer1; SRaftId peer1;

22
source/libs/sync/test/syncRaftStoreTest.cpp
View File

@ -4,14 +4,13 @@
#include "syncIO.h" #include "syncIO.h"
#include "syncInt.h" #include "syncInt.h"
void *pingFunc(void *param) { void logTest() {
SSyncIO *io = (SSyncIO *)param; sTrace("--- sync log test: trace");
while (1) { sDebug("--- sync log test: debug");
sDebug("io->ping"); sInfo("--- sync log test: info");
// io->ping(io); sWarn("--- sync log test: warn");
sleep(1); sError("--- sync log test: error");
} sFatal("--- sync log test: fatal");
return NULL;
} }
int main() { int main() {
@ -19,12 +18,7 @@ int main() {
tsAsyncLog = 0; tsAsyncLog = 0;
sDebugFlag = 143 + 64; sDebugFlag = 143 + 64;
sTrace("sync log test: trace"); logTest();
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");
SRaftStore *pRaftStore = raftStoreOpen("./raft_store.json"); SRaftStore *pRaftStore = raftStoreOpen("./raft_store.json");
assert(pRaftStore != NULL); assert(pRaftStore != NULL);

51
source/libs/sync/test/syncTest.cpp
View File

@ -4,55 +4,20 @@
#include "syncInt.h" #include "syncInt.h"
#include "syncRaftStore.h" #include "syncRaftStore.h"
void *pingFunc(void *param) { void logTest() {
SSyncIO *io = (SSyncIO *)param; sTrace("--- sync log test: trace");
while (1) { sDebug("--- sync log test: debug");
sDebug("io->ping"); sInfo("--- sync log test: info");
// io->ping(io); sWarn("--- sync log test: warn");
sleep(1); sError("--- sync log test: error");
} sFatal("--- sync log test: fatal");
return NULL;
} }
int main() { int main() {
// taosInitLog((char *)"syncTest.log", 100000, 10); // taosInitLog((char *)"syncTest.log", 100000, 10);
tsAsyncLog = 0; tsAsyncLog = 0;
sDebugFlag = 143 + 64; sDebugFlag = 143 + 64;
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");
SRaftStore *pRaftStore = raftStoreOpen("./raft_store.json");
// assert(pRaftStore != NULL);
// raftStorePrint(pRaftStore);
// pRaftStore->currentTerm = 100;
// pRaftStore->voteFor.addr = 200;
// pRaftStore->voteFor.vgId = 300;
// raftStorePrint(pRaftStore);
// raftStorePersist(pRaftStore);
// sDebug("sync test");
// SSyncIO *syncIO = syncIOCreate();
// assert(syncIO != NULL);
// syncIO->start(syncIO);
// sleep(2);
// pthread_t tid;
// pthread_create(&tid, NULL, pingFunc, syncIO);
// while (1) {
// sleep(1);
// }
return 0; return 0;
} }