sync refactor

source/libs/sync/inc/syncEnv.h

@@ -30,12 +30,17 @@ extern "C" {
 
 #define TIMER_MAX_MS 0x7FFFFFFF
 #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 {
   tmr_h pEnvTickTimer;
   tmr_h pTimerManager;
   char  name[128];
-
 } SSyncEnv;
 
 extern SSyncEnv* gSyncEnv;

source/libs/sync/inc/syncInt.h

@@ -154,9 +154,8 @@ typedef struct SSyncNode {
   SyncIndex      commitIndex;
 
   // timer
-  tmr_h   pPingTimer;
-  int32_t pingTimerMS;
-  // uint8_t           pingTimerEnable;
+  tmr_h             pPingTimer;
+  int32_t           pingTimerMS;
   uint64_t          pingTimerLogicClock;
   uint64_t          pingTimerLogicClockUser;
   TAOS_TMR_CALLBACK FpPingTimer;  // Timer Fp
@@ -164,13 +163,15 @@ typedef struct SSyncNode {
 
   tmr_h             pElectTimer;
   int32_t           electTimerMS;
-  uint8_t           electTimerEnable;
+  uint64_t          electTimerLogicClock;
+  uint64_t          electTimerLogicClockUser;
   TAOS_TMR_CALLBACK FpElectTimer;  // Timer Fp
   uint64_t          electTimerCounter;
 
   tmr_h             pHeartbeatTimer;
   int32_t           heartbeatTimerMS;
-  uint8_t           heartbeatTimerEnable;
+  uint64_t          heartbeatTimerLogicClock;
+  uint64_t          heartbeatTimerLogicClockUser;
   TAOS_TMR_CALLBACK FpHeartbeatTimer;  // Timer Fp
   uint64_t          heartbeatTimerCounter;
 
@@ -194,8 +195,9 @@ void       syncNodePingSelf(SSyncNode* pSyncNode);
 int32_t syncNodeStartPingTimer(SSyncNode* pSyncNode);
 int32_t syncNodeStopPingTimer(SSyncNode* pSyncNode);
 
-int32_t syncNodeStartElectTimer(SSyncNode* pSyncNode);
+int32_t syncNodeStartElectTimer(SSyncNode* pSyncNode, int32_t ms);
 int32_t syncNodeStopElectTimer(SSyncNode* pSyncNode);
+int32_t syncNodeRestartElectTimer(SSyncNode* pSyncNode, int32_t ms);
 
 int32_t syncNodeStartHeartbeatTimer(SSyncNode* pSyncNode);
 int32_t syncNodeStopHeartbeatTimer(SSyncNode* pSyncNode);

source/libs/sync/inc/syncMessage.h

@@ -59,6 +59,7 @@ typedef struct SyncTimeout {
   uint32_t         msgType;
   ESyncTimeoutType timeoutType;
   uint64_t         logicClock;
+  int32_t          timerMS;
   void*            data;
 } SyncTimeout;
 
@@ -69,7 +70,7 @@ void         syncTimeoutDeserialize(const char* buf, uint32_t len, SyncTimeout*
 void         syncTimeout2RpcMsg(const SyncTimeout* pMsg, SRpcMsg* pRpcMsg);
 void         syncTimeoutFromRpcMsg(const SRpcMsg* pRpcMsg, 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 {

source/libs/sync/inc/syncUtil.h

@@ -28,28 +28,23 @@ extern "C" {
 #include "taosdef.h"
 
 // ---- encode / decode
-
 uint64_t syncUtilAddr2U64(const char* host, 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 syncUtilnodeInfo2raftId(const SNodeInfo* pNodeInfo, SyncGroupId vgId, SRaftId* raftId);
-
-bool syncUtilSameId(const SRaftId* pId1, const SRaftId* pId2);
+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     syncUtilnodeInfo2raftId(const SNodeInfo* pNodeInfo, SyncGroupId vgId, SRaftId* raftId);
+bool     syncUtilSameId(const SRaftId* pId1, const SRaftId* pId2);
 
 // ---- SSyncBuffer ----
 void syncUtilbufBuild(SSyncBuffer* syncBuf, size_t len);
-
 void syncUtilbufDestroy(SSyncBuffer* syncBuf);
-
 void syncUtilbufCopy(const SSyncBuffer* src, SSyncBuffer* dest);
-
 void syncUtilbufCopyDeep(const SSyncBuffer* src, SSyncBuffer* dest);
 
+// ---- misc ----
+int32_t syncUtilRand(int32_t max);
+int32_t syncUtilElectRandomMS();
+
 #ifdef __cplusplus
 }
 #endif

source/libs/sync/src/syncEnv.c

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

source/libs/sync/src/syncIO.c

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

source/libs/sync/src/syncMain.c

@@ -98,6 +98,7 @@ SSyncNode* syncNodeOpen(const SSyncInfo* pSyncInfo) {
   pSyncNode->state = TAOS_SYNC_STATE_FOLLOWER;
   syncUtilnodeInfo2raftId(&pSyncNode->me, pSyncNode->vgId, &pSyncNode->raftId);
 
+  // init ping timer
   pSyncNode->pPingTimer = NULL;
   pSyncNode->pingTimerMS = PING_TIMER_MS;
   atomic_store_64(&pSyncNode->pingTimerLogicClock, 0);
@@ -105,6 +106,22 @@ SSyncNode* syncNodeOpen(const SSyncInfo* pSyncInfo) {
   pSyncNode->FpPingTimer = syncNodeEqPingTimer;
   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->FpOnPingReply = syncNodeOnPingReplyCb;
   pSyncNode->FpOnRequestVote = syncNodeOnRequestVoteCb;
@@ -157,7 +174,6 @@ void syncNodePingSelf(SSyncNode* pSyncNode) {
 int32_t syncNodeStartPingTimer(SSyncNode* pSyncNode) {
   atomic_store_64(&pSyncNode->pingTimerLogicClock, pSyncNode->pingTimerLogicClockUser);
   pSyncNode->pingTimerMS = PING_TIMER_MS;
-
   if (pSyncNode->pPingTimer == NULL) {
     pSyncNode->pPingTimer =
         taosTmrStart(pSyncNode->FpPingTimer, pSyncNode->pingTimerMS, pSyncNode, gSyncEnv->pTimerManager);
@@ -165,7 +181,6 @@ int32_t syncNodeStartPingTimer(SSyncNode* pSyncNode) {
     taosTmrReset(pSyncNode->FpPingTimer, pSyncNode->pingTimerMS, pSyncNode, gSyncEnv->pTimerManager,
                  &pSyncNode->pPingTimer);
   }
-
   return 0;
 }
 
@@ -175,7 +190,9 @@ int32_t syncNodeStopPingTimer(SSyncNode* pSyncNode) {
   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) {
     pSyncNode->pElectTimer =
         taosTmrStart(pSyncNode->FpElectTimer, pSyncNode->electTimerMS, pSyncNode, gSyncEnv->pTimerManager);
@@ -183,18 +200,23 @@ int32_t syncNodeStartElectTimer(SSyncNode* pSyncNode) {
     taosTmrReset(pSyncNode->FpElectTimer, pSyncNode->electTimerMS, pSyncNode, gSyncEnv->pTimerManager,
                  &pSyncNode->pElectTimer);
   }
-
-  atomic_store_8(&pSyncNode->electTimerEnable, 1);
   return 0;
 }
 
 int32_t syncNodeStopElectTimer(SSyncNode* pSyncNode) {
-  atomic_store_8(&pSyncNode->electTimerEnable, 0);
+  atomic_add_fetch_64(&pSyncNode->electTimerLogicClockUser, 1);
   pSyncNode->electTimerMS = TIMER_MAX_MS;
   return 0;
 }
 
+int32_t syncNodeRestartElectTimer(SSyncNode* pSyncNode, int32_t ms) {
+  syncNodeStopElectTimer(pSyncNode);
+  syncNodeStartElectTimer(pSyncNode, ms);
+  return 0;
+}
+
 int32_t syncNodeStartHeartbeatTimer(SSyncNode* pSyncNode) {
+  atomic_store_64(&pSyncNode->heartbeatTimerLogicClock, pSyncNode->heartbeatTimerLogicClockUser);
   if (pSyncNode->pHeartbeatTimer == NULL) {
     pSyncNode->pHeartbeatTimer =
         taosTmrStart(pSyncNode->FpHeartbeatTimer, pSyncNode->heartbeatTimerMS, pSyncNode, gSyncEnv->pTimerManager);
@@ -202,13 +224,11 @@ int32_t syncNodeStartHeartbeatTimer(SSyncNode* pSyncNode) {
     taosTmrReset(pSyncNode->FpHeartbeatTimer, pSyncNode->heartbeatTimerMS, pSyncNode, gSyncEnv->pTimerManager,
                  &pSyncNode->pHeartbeatTimer);
   }
-
-  atomic_store_8(&pSyncNode->heartbeatTimerEnable, 1);
   return 0;
 }
 
 int32_t syncNodeStopHeartbeatTimer(SSyncNode* pSyncNode) {
-  atomic_store_8(&pSyncNode->heartbeatTimerEnable, 0);
+  atomic_add_fetch_64(&pSyncNode->heartbeatTimerLogicClockUser, 1);
   pSyncNode->heartbeatTimerMS = TIMER_MAX_MS;
   return 0;
 }
@@ -320,36 +340,56 @@ static int32_t syncNodeOnTimeoutCb(SSyncNode* ths, SyncTimeout* pMsg) {
 static void syncNodeEqPingTimer(void* param, void* tmrId) {
   SSyncNode* pSyncNode = (SSyncNode*)param;
   if (atomic_load_64(&pSyncNode->pingTimerLogicClockUser) <= atomic_load_64(&pSyncNode->pingTimerLogicClock)) {
+    SyncTimeout* pSyncMsg = syncTimeoutBuild2(SYNC_TIMEOUT_PING, atomic_load_64(&pSyncNode->pingTimerLogicClock),
+                                              pSyncNode->pingTimerMS, pSyncNode);
+    SRpcMsg      rpcMsg;
+    syncTimeout2RpcMsg(pSyncMsg, &rpcMsg);
+    pSyncNode->FpEqMsg(pSyncNode->queue, &rpcMsg);
+    syncTimeoutDestroy(pSyncMsg);
+
+    // reset timer ms
     // pSyncNode->pingTimerMS += 100;
 
-    SyncTimeout* pSyncMsg =
-        syncTimeoutBuild2(SYNC_TIMEOUT_PING, atomic_load_64(&pSyncNode->pingTimerLogicClock), pSyncNode);
+    taosTmrReset(syncNodeEqPingTimer, pSyncNode->pingTimerMS, pSyncNode, &gSyncEnv->pTimerManager,
+                 &pSyncNode->pPingTimer);
+  } else {
+    sTrace("syncNodeEqPingTimer: pingTimerLogicClock:%lu, pingTimerLogicClockUser:%lu", pSyncNode->pingTimerLogicClock,
+           pSyncNode->pingTimerLogicClockUser);
+  }
+}
+
+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("syncNodeEqPingTimer: pingTimerLogicClock:%lu, pingTimerLogicClockUser:%lu", pSyncNode->pingTimerLogicClock,
-           pSyncNode->pingTimerLogicClockUser);
+    sTrace("syncNodeEqElectTimer: electTimerLogicClock:%lu, electTimerLogicClockUser:%lu",
+           pSyncNode->electTimerLogicClock, pSyncNode->electTimerLogicClockUser);
   }
 }
 
-static void syncNodeEqElectTimer(void* param, void* tmrId) {}
-
 static void syncNodeEqHeartbeatTimer(void* param, void* tmrId) {}
 
 static void syncNodeBecomeFollower(SSyncNode* pSyncNode) {
   if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
-    pSyncNode->leaderCache.addr = 0;
-    pSyncNode->leaderCache.vgId = 0;
+    pSyncNode->leaderCache = EMPTY_RAFT_ID;
   }
 
   syncNodeStopHeartbeatTimer(pSyncNode);
-  syncNodeStartElectTimer(pSyncNode);
+  int32_t electMS = syncUtilElectRandomMS();
+  syncNodeStartElectTimer(pSyncNode, electMS);
 }
 
 static void syncNodeBecomeLeader(SSyncNode* pSyncNode) {

source/libs/sync/src/syncMessage.c

@@ -125,6 +125,7 @@ cJSON* syncTimeout2Json(const SyncTimeout* pMsg) {
   cJSON_AddNumberToObject(pRoot, "timeoutType", pMsg->timeoutType);
   snprintf(u64buf, sizeof(u64buf), "%lu", pMsg->logicClock);
   cJSON_AddStringToObject(pRoot, "logicClock", u64buf);
+  cJSON_AddNumberToObject(pRoot, "timerMS", pMsg->timerMS);
   snprintf(u64buf, sizeof(u64buf), "%p", pMsg->data);
   cJSON_AddStringToObject(pRoot, "data", u64buf);
 
@@ -133,10 +134,11 @@ cJSON* syncTimeout2Json(const SyncTimeout* pMsg) {
   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();
   pMsg->timeoutType = timeoutType;
   pMsg->logicClock = logicClock;
+  pMsg->timerMS = timerMS;
   pMsg->data = data;
   return pMsg;
 }

source/libs/sync/src/syncUtil.c

@@ -17,6 +17,7 @@
 #include <arpa/inet.h>
 #include <netinet/in.h>
 #include <sys/socket.h>
+#include "syncEnv.h"
 
 // ---- encode / decode
 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);
   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); }
\ No newline at end of file

source/libs/sync/test/syncIndexTest.cpp

@@ -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() {
   // taosInitLog((char *)"syncTest.log", 100000, 10);
   tsAsyncLog = 0;
   sDebugFlag = 143 + 64;
 
-  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");
+  logTest();
 
   SRaftId me;
   SRaftId peer1;

source/libs/sync/test/syncRaftStoreTest.cpp

@@ -4,14 +4,13 @@
 #include "syncIO.h"
 #include "syncInt.h"
 
-void *pingFunc(void *param) {
-  SSyncIO *io = (SSyncIO *)param;
-  while (1) {
-    sDebug("io->ping");
-    // io->ping(io);
-    sleep(1);
-  }
-  return NULL;
+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() {
@@ -19,12 +18,7 @@ int main() {
   tsAsyncLog = 0;
   sDebugFlag = 143 + 64;
 
-  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");
+  logTest();
 
   SRaftStore *pRaftStore = raftStoreOpen("./raft_store.json");
   assert(pRaftStore != NULL);

source/libs/sync/test/syncTest.cpp

@@ -4,55 +4,20 @@
 #include "syncInt.h"
 #include "syncRaftStore.h"
 
-void *pingFunc(void *param) {
-  SSyncIO *io = (SSyncIO *)param;
-  while (1) {
-    sDebug("io->ping");
-    // io->ping(io);
-    sleep(1);
-  }
-  return NULL;
+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() {
   // taosInitLog((char *)"syncTest.log", 100000, 10);
   tsAsyncLog = 0;
   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;
 }