syncMain.c 18.8 KB
Newer Older
M
Minghao Li 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16
/*
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
 *
 * This program is free software: you can use, redistribute, and/or modify
 * it under the terms of the GNU Affero General Public License, version 3
 * or later ("AGPL"), as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

#include <stdint.h>
M
Minghao Li 已提交
17
#include "sync.h"
M
Minghao Li 已提交
18 19
#include "syncAppendEntries.h"
#include "syncAppendEntriesReply.h"
M
Minghao Li 已提交
20
#include "syncEnv.h"
M
Minghao Li 已提交
21
#include "syncInt.h"
M
Minghao Li 已提交
22
#include "syncRaft.h"
M
Minghao Li 已提交
23
#include "syncRaftStore.h"
M
Minghao Li 已提交
24 25
#include "syncRequestVote.h"
#include "syncRequestVoteReply.h"
M
Minghao Li 已提交
26
#include "syncTimeout.h"
M
Minghao Li 已提交
27
#include "syncUtil.h"
M
Minghao Li 已提交
28
#include "syncVoteMgr.h"
M
Minghao Li 已提交
29

M
Minghao Li 已提交
30 31 32
static int32_t tsNodeRefId = -1;

// ------ local funciton ---------
M
Minghao Li 已提交
33 34 35 36
static void syncNodeEqPingTimer(void* param, void* tmrId);
static void syncNodeEqElectTimer(void* param, void* tmrId);
static void syncNodeEqHeartbeatTimer(void* param, void* tmrId);

M
Minghao Li 已提交
37 38
static int32_t syncNodeOnPingCb(SSyncNode* ths, SyncPing* pMsg);
static int32_t syncNodeOnPingReplyCb(SSyncNode* ths, SyncPingReply* pMsg);
M
Minghao Li 已提交
39

M
Minghao Li 已提交
40
static void UpdateTerm(SSyncNode* pSyncNode, SyncTerm term);
M
Minghao Li 已提交
41 42 43 44 45 46
static void syncNodeBecomeFollower(SSyncNode* pSyncNode);
static void syncNodeBecomeLeader(SSyncNode* pSyncNode);
static void syncNodeFollower2Candidate(SSyncNode* pSyncNode);
static void syncNodeCandidate2Leader(SSyncNode* pSyncNode);
static void syncNodeLeader2Follower(SSyncNode* pSyncNode);
static void syncNodeCandidate2Follower(SSyncNode* pSyncNode);
M
Minghao Li 已提交
47 48 49 50 51 52
// ---------------------------------

int32_t syncInit() {
  sTrace("syncInit ok");
  return 0;
}
M
Minghao Li 已提交
53

M
Minghao Li 已提交
54
void syncCleanUp() { sTrace("syncCleanUp ok"); }
M
Minghao Li 已提交
55

M
Minghao Li 已提交
56 57 58 59 60
int64_t syncStart(const SSyncInfo* pSyncInfo) {
  SSyncNode* pSyncNode = syncNodeOpen(pSyncInfo);
  assert(pSyncNode != NULL);
  return 0;
}
M
Minghao Li 已提交
61 62 63 64 65

void syncStop(int64_t rid) {}

int32_t syncReconfig(int64_t rid, const SSyncCfg* pSyncCfg) { return 0; }

M
Minghao Li 已提交
66
int32_t syncForwardToPeer(int64_t rid, const SRpcMsg* pBuf, bool isWeak) { return 0; }
M
Minghao Li 已提交
67 68 69

ESyncState syncGetMyRole(int64_t rid) { return TAOS_SYNC_STATE_LEADER; }

M
Minghao Li 已提交
70 71 72 73 74
void syncGetNodesRole(int64_t rid, SNodesRole* pNodeRole) {}

SSyncNode* syncNodeOpen(const SSyncInfo* pSyncInfo) {
  SSyncNode* pSyncNode = (SSyncNode*)malloc(sizeof(SSyncNode));
  assert(pSyncNode != NULL);
M
Minghao Li 已提交
75
  memset(pSyncNode, 0, sizeof(SSyncNode));
M
Minghao Li 已提交
76

M
Minghao Li 已提交
77
  // init by SSyncInfo
M
Minghao Li 已提交
78 79 80
  pSyncNode->vgId = pSyncInfo->vgId;
  pSyncNode->syncCfg = pSyncInfo->syncCfg;
  memcpy(pSyncNode->path, pSyncInfo->path, sizeof(pSyncNode->path));
M
Minghao Li 已提交
81
  memcpy(pSyncNode->walPath, pSyncInfo->walPath, sizeof(pSyncNode->walPath));
M
Minghao Li 已提交
82 83
  pSyncNode->rpcClient = pSyncInfo->rpcClient;
  pSyncNode->FpSendMsg = pSyncInfo->FpSendMsg;
M
Minghao Li 已提交
84 85
  pSyncNode->queue = pSyncInfo->queue;
  pSyncNode->FpEqMsg = pSyncInfo->FpEqMsg;
M
Minghao Li 已提交
86

M
Minghao Li 已提交
87
  // init internal
M
Minghao Li 已提交
88 89
  pSyncNode->myNodeInfo = pSyncInfo->syncCfg.nodeInfo[pSyncInfo->syncCfg.myIndex];
  syncUtilnodeInfo2raftId(&pSyncNode->myNodeInfo, pSyncInfo->vgId, &pSyncNode->myRaftId);
M
Minghao Li 已提交
90

M
Minghao Li 已提交
91 92
  // init peersNum, peers, peersId
  pSyncNode->peersNum = pSyncInfo->syncCfg.replicaNum - 1;
M
Minghao Li 已提交
93 94 95
  int j = 0;
  for (int i = 0; i < pSyncInfo->syncCfg.replicaNum; ++i) {
    if (i != pSyncInfo->syncCfg.myIndex) {
M
Minghao Li 已提交
96
      pSyncNode->peersNodeInfo[j] = pSyncInfo->syncCfg.nodeInfo[i];
M
Minghao Li 已提交
97 98 99
      j++;
    }
  }
M
Minghao Li 已提交
100
  for (int i = 0; i < pSyncNode->peersNum; ++i) {
M
Minghao Li 已提交
101
    syncUtilnodeInfo2raftId(&pSyncNode->peersNodeInfo[i], pSyncInfo->vgId, &pSyncNode->peersId[i]);
M
Minghao Li 已提交
102
  }
M
Minghao Li 已提交
103

M
Minghao Li 已提交
104 105 106 107 108 109 110 111 112 113 114 115 116 117
  // init replicaNum, replicasId
  pSyncNode->replicaNum = pSyncInfo->syncCfg.replicaNum;
  for (int i = 0; i < pSyncInfo->syncCfg.replicaNum; ++i) {
    syncUtilnodeInfo2raftId(&pSyncInfo->syncCfg.nodeInfo[i], pSyncInfo->vgId, &pSyncNode->replicasId[i]);
  }

  // raft algorithm
  pSyncNode->pFsm = pSyncInfo->pFsm;
  pSyncNode->quorum = syncUtilQuorum(pSyncInfo->syncCfg.replicaNum);
  pSyncNode->leaderCache = EMPTY_RAFT_ID;

  // life cycle

  // init server vars
M
syncInt  
Minghao Li 已提交
118
  pSyncNode->state = TAOS_SYNC_STATE_FOLLOWER;
M
Minghao Li 已提交
119 120 121 122 123 124 125 126 127 128 129 130
  pSyncNode->pRaftStore = raftStoreOpen(pSyncInfo->walPath);
  assert(pSyncNode->pRaftStore != NULL);

  // init candidate vars
  pSyncNode->pVotesGranted = voteGrantedCreate(pSyncNode);
  assert(pSyncNode->pVotesGranted != NULL);
  pSyncNode->pVotesRespond = votesRespondCreate(pSyncNode);
  assert(pSyncNode->pVotesRespond != NULL);

  // init leader vars
  pSyncNode->pNextIndex = NULL;
  pSyncNode->pMatchIndex = NULL;
M
Minghao Li 已提交
131

M
Minghao Li 已提交
132
  // init ping timer
M
Minghao Li 已提交
133
  pSyncNode->pPingTimer = NULL;
M
Minghao Li 已提交
134 135 136
  pSyncNode->pingTimerMS = PING_TIMER_MS;
  atomic_store_64(&pSyncNode->pingTimerLogicClock, 0);
  atomic_store_64(&pSyncNode->pingTimerLogicClockUser, 0);
M
Minghao Li 已提交
137
  pSyncNode->FpPingTimer = syncNodeEqPingTimer;
M
Minghao Li 已提交
138
  pSyncNode->pingTimerCounter = 0;
M
Minghao Li 已提交
139

M
Minghao Li 已提交
140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155
  // 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;

M
Minghao Li 已提交
156
  // init callback
M
Minghao Li 已提交
157 158 159 160 161 162
  pSyncNode->FpOnPing = syncNodeOnPingCb;
  pSyncNode->FpOnPingReply = syncNodeOnPingReplyCb;
  pSyncNode->FpOnRequestVote = syncNodeOnRequestVoteCb;
  pSyncNode->FpOnRequestVoteReply = syncNodeOnRequestVoteReplyCb;
  pSyncNode->FpOnAppendEntries = syncNodeOnAppendEntriesCb;
  pSyncNode->FpOnAppendEntriesReply = syncNodeOnAppendEntriesReplyCb;
M
Minghao Li 已提交
163
  pSyncNode->FpOnTimeout = syncNodeOnTimeoutCb;
M
Minghao Li 已提交
164 165 166 167 168 169 170 171 172

  return pSyncNode;
}

void syncNodeClose(SSyncNode* pSyncNode) {
  assert(pSyncNode != NULL);
  free(pSyncNode);
}

M
Minghao Li 已提交
173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192
cJSON* syncNode2Json(const SSyncNode* pSyncNode) {
  char   u64buf[128];
  cJSON* pRoot = cJSON_CreateObject();

  // init by SSyncInfo
  cJSON_AddNumberToObject(pRoot, "vgId", pSyncNode->vgId);
  cJSON_AddStringToObject(pRoot, "path", pSyncNode->path);
  cJSON_AddStringToObject(pRoot, "walPath", pSyncNode->walPath);

  snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->rpcClient);
  cJSON_AddStringToObject(pRoot, "rpcClient", u64buf);
  snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpSendMsg);
  cJSON_AddStringToObject(pRoot, "FpSendMsg", u64buf);

  snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->queue);
  cJSON_AddStringToObject(pRoot, "queue", u64buf);
  snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpEqMsg);
  cJSON_AddStringToObject(pRoot, "FpEqMsg", u64buf);

  // init internal
M
Minghao Li 已提交
193 194 195 196
  cJSON* pMe = syncUtilNodeInfo2Json(&pSyncNode->myNodeInfo);
  cJSON_AddItemToObject(pRoot, "myNodeInfo", pMe);
  cJSON* pRaftId = syncUtilRaftId2Json(&pSyncNode->myRaftId);
  cJSON_AddItemToObject(pRoot, "myRaftId", pRaftId);
M
Minghao Li 已提交
197 198 199

  cJSON_AddNumberToObject(pRoot, "peersNum", pSyncNode->peersNum);
  cJSON* pPeers = cJSON_CreateArray();
M
Minghao Li 已提交
200
  cJSON_AddItemToObject(pRoot, "peersNodeInfo", pPeers);
M
Minghao Li 已提交
201
  for (int i = 0; i < pSyncNode->peersNum; ++i) {
M
Minghao Li 已提交
202
    cJSON_AddItemToArray(pPeers, syncUtilNodeInfo2Json(&pSyncNode->peersNodeInfo[i]));
M
Minghao Li 已提交
203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224
  }
  cJSON* pPeersId = cJSON_CreateArray();
  cJSON_AddItemToObject(pRoot, "peersId", pPeersId);
  for (int i = 0; i < pSyncNode->peersNum; ++i) {
    cJSON_AddItemToArray(pPeersId, syncUtilRaftId2Json(&pSyncNode->peersId[i]));
  }

  cJSON_AddNumberToObject(pRoot, "replicaNum", pSyncNode->replicaNum);
  cJSON* pReplicasId = cJSON_CreateArray();
  cJSON_AddItemToObject(pRoot, "replicasId", pReplicasId);
  for (int i = 0; i < pSyncNode->replicaNum; ++i) {
    cJSON_AddItemToArray(pReplicasId, syncUtilRaftId2Json(&pSyncNode->replicasId[i]));
  }

  // raft algorithm
  snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->pFsm);
  cJSON_AddStringToObject(pRoot, "pFsm", u64buf);
  cJSON_AddNumberToObject(pRoot, "quorum", pSyncNode->quorum);
  cJSON* pLaderCache = syncUtilRaftId2Json(&pSyncNode->leaderCache);
  cJSON_AddItemToObject(pRoot, "leaderCache", pLaderCache);

  // tla+ server vars
M
Minghao Li 已提交
225 226
  cJSON_AddNumberToObject(pRoot, "state", pSyncNode->state);
  cJSON_AddStringToObject(pRoot, "state_str", syncUtilState2String(pSyncNode->state));
M
Minghao Li 已提交
227 228 229 230 231 232 233 234 235 236 237 238

  // tla+ candidate vars

  // tla+ leader vars

  // tla+ log vars

  cJSON* pJson = cJSON_CreateObject();
  cJSON_AddItemToObject(pJson, "SSyncNode", pRoot);
  return pJson;
}

M
Minghao Li 已提交
239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280
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;
}

M
Minghao Li 已提交
281
int32_t syncNodePingAll(SSyncNode* pSyncNode) {
M
Minghao Li 已提交
282
  sTrace("syncNodePingAll pSyncNode:%p ", pSyncNode);
M
Minghao Li 已提交
283 284
  int32_t ret = 0;
  for (int i = 0; i < pSyncNode->syncCfg.replicaNum; ++i) {
M
Minghao Li 已提交
285 286
    SRaftId destId;
    syncUtilnodeInfo2raftId(&pSyncNode->syncCfg.nodeInfo[i], pSyncNode->vgId, &destId);
M
Minghao Li 已提交
287
    SyncPing* pMsg = syncPingBuild3(&pSyncNode->myRaftId, &destId);
M
Minghao Li 已提交
288
    ret = syncNodePing(pSyncNode, &destId, pMsg);
M
Minghao Li 已提交
289
    assert(ret == 0);
M
Minghao Li 已提交
290
    syncPingDestroy(pMsg);
M
Minghao Li 已提交
291
  }
M
Minghao Li 已提交
292
}
M
Minghao Li 已提交
293

M
Minghao Li 已提交
294
int32_t syncNodePingPeers(SSyncNode* pSyncNode) {
M
Minghao Li 已提交
295 296
  int32_t ret = 0;
  for (int i = 0; i < pSyncNode->peersNum; ++i) {
M
Minghao Li 已提交
297
    SRaftId destId;
M
Minghao Li 已提交
298 299
    syncUtilnodeInfo2raftId(&pSyncNode->peersNodeInfo[i], pSyncNode->vgId, &destId);
    SyncPing* pMsg = syncPingBuild3(&pSyncNode->myRaftId, &destId);
M
Minghao Li 已提交
300
    ret = syncNodePing(pSyncNode, &destId, pMsg);
M
Minghao Li 已提交
301
    assert(ret == 0);
M
Minghao Li 已提交
302
    syncPingDestroy(pMsg);
M
Minghao Li 已提交
303 304
  }
}
M
Minghao Li 已提交
305

M
Minghao Li 已提交
306
int32_t syncNodePingSelf(SSyncNode* pSyncNode) {
M
Minghao Li 已提交
307
  int32_t   ret;
M
Minghao Li 已提交
308
  SyncPing* pMsg = syncPingBuild3(&pSyncNode->myRaftId, &pSyncNode->myRaftId);
M
Minghao Li 已提交
309
  ret = syncNodePing(pSyncNode, &pMsg->destId, pMsg);
M
Minghao Li 已提交
310
  assert(ret == 0);
M
Minghao Li 已提交
311
  syncPingDestroy(pMsg);
M
Minghao Li 已提交
312
}
M
Minghao Li 已提交
313

M
Minghao Li 已提交
314
int32_t syncNodeStartPingTimer(SSyncNode* pSyncNode) {
M
Minghao Li 已提交
315 316
  atomic_store_64(&pSyncNode->pingTimerLogicClock, pSyncNode->pingTimerLogicClockUser);
  pSyncNode->pingTimerMS = PING_TIMER_MS;
M
Minghao Li 已提交
317 318
  if (pSyncNode->pPingTimer == NULL) {
    pSyncNode->pPingTimer =
M
Minghao Li 已提交
319
        taosTmrStart(pSyncNode->FpPingTimer, pSyncNode->pingTimerMS, pSyncNode, gSyncEnv->pTimerManager);
M
Minghao Li 已提交
320
  } else {
M
Minghao Li 已提交
321
    taosTmrReset(pSyncNode->FpPingTimer, pSyncNode->pingTimerMS, pSyncNode, gSyncEnv->pTimerManager,
M
Minghao Li 已提交
322 323 324 325 326 327
                 &pSyncNode->pPingTimer);
  }
  return 0;
}

int32_t syncNodeStopPingTimer(SSyncNode* pSyncNode) {
M
Minghao Li 已提交
328
  atomic_add_fetch_64(&pSyncNode->pingTimerLogicClockUser, 1);
M
Minghao Li 已提交
329 330 331 332
  pSyncNode->pingTimerMS = TIMER_MAX_MS;
  return 0;
}

M
Minghao Li 已提交
333 334 335
int32_t syncNodeStartElectTimer(SSyncNode* pSyncNode, int32_t ms) {
  pSyncNode->electTimerMS = ms;
  atomic_store_64(&pSyncNode->electTimerLogicClock, pSyncNode->electTimerLogicClockUser);
M
Minghao Li 已提交
336 337 338 339 340 341 342
  if (pSyncNode->pElectTimer == NULL) {
    pSyncNode->pElectTimer =
        taosTmrStart(pSyncNode->FpElectTimer, pSyncNode->electTimerMS, pSyncNode, gSyncEnv->pTimerManager);
  } else {
    taosTmrReset(pSyncNode->FpElectTimer, pSyncNode->electTimerMS, pSyncNode, gSyncEnv->pTimerManager,
                 &pSyncNode->pElectTimer);
  }
M
Minghao Li 已提交
343 344 345
  return 0;
}

M
Minghao Li 已提交
346
int32_t syncNodeStopElectTimer(SSyncNode* pSyncNode) {
M
Minghao Li 已提交
347
  atomic_add_fetch_64(&pSyncNode->electTimerLogicClockUser, 1);
M
Minghao Li 已提交
348 349 350 351
  pSyncNode->electTimerMS = TIMER_MAX_MS;
  return 0;
}

M
Minghao Li 已提交
352 353 354 355 356 357
int32_t syncNodeRestartElectTimer(SSyncNode* pSyncNode, int32_t ms) {
  syncNodeStopElectTimer(pSyncNode);
  syncNodeStartElectTimer(pSyncNode, ms);
  return 0;
}

M
Minghao Li 已提交
358
int32_t syncNodeStartHeartbeatTimer(SSyncNode* pSyncNode) {
M
Minghao Li 已提交
359
  atomic_store_64(&pSyncNode->heartbeatTimerLogicClock, pSyncNode->heartbeatTimerLogicClockUser);
M
Minghao Li 已提交
360 361 362 363 364 365 366 367 368 369 370
  if (pSyncNode->pHeartbeatTimer == NULL) {
    pSyncNode->pHeartbeatTimer =
        taosTmrStart(pSyncNode->FpHeartbeatTimer, pSyncNode->heartbeatTimerMS, pSyncNode, gSyncEnv->pTimerManager);
  } else {
    taosTmrReset(pSyncNode->FpHeartbeatTimer, pSyncNode->heartbeatTimerMS, pSyncNode, gSyncEnv->pTimerManager,
                 &pSyncNode->pHeartbeatTimer);
  }
  return 0;
}

int32_t syncNodeStopHeartbeatTimer(SSyncNode* pSyncNode) {
M
Minghao Li 已提交
371
  atomic_add_fetch_64(&pSyncNode->heartbeatTimerLogicClockUser, 1);
M
Minghao Li 已提交
372 373 374 375
  pSyncNode->heartbeatTimerMS = TIMER_MAX_MS;
  return 0;
}

M
Minghao Li 已提交
376
// ------ local funciton ---------
M
Minghao Li 已提交
377
static int32_t syncNodeOnPingCb(SSyncNode* ths, SyncPing* pMsg) {
M
Minghao Li 已提交
378
  int32_t ret = 0;
M
Minghao Li 已提交
379 380 381 382 383
  sTrace("<-- syncNodeOnPingCb -->");

  {
    cJSON* pJson = syncPing2Json(pMsg);
    char*  serialized = cJSON_Print(pJson);
M
Minghao Li 已提交
384
    sTrace("process syncMessage recv: syncNodeOnPingCb pMsg:%s ", serialized);
M
Minghao Li 已提交
385 386 387 388
    free(serialized);
    cJSON_Delete(pJson);
  }

M
Minghao Li 已提交
389
  SyncPingReply* pMsgReply = syncPingReplyBuild3(&ths->myRaftId, &pMsg->srcId);
M
Minghao Li 已提交
390 391 392 393
  SRpcMsg        rpcMsg;
  syncPingReply2RpcMsg(pMsgReply, &rpcMsg);
  syncNodeSendMsgById(&pMsgReply->destId, ths, &rpcMsg);

M
Minghao Li 已提交
394 395 396
  return ret;
}

M
Minghao Li 已提交
397
static int32_t syncNodeOnPingReplyCb(SSyncNode* ths, SyncPingReply* pMsg) {
M
Minghao Li 已提交
398
  int32_t ret = 0;
M
Minghao Li 已提交
399 400 401 402 403
  sTrace("<-- syncNodeOnPingReplyCb -->");

  {
    cJSON* pJson = syncPingReply2Json(pMsg);
    char*  serialized = cJSON_Print(pJson);
M
Minghao Li 已提交
404
    sTrace("process syncMessage recv: syncNodeOnPingReplyCb pMsg:%s ", serialized);
M
Minghao Li 已提交
405 406 407 408
    free(serialized);
    cJSON_Delete(pJson);
  }

M
Minghao Li 已提交
409 410 411
  return ret;
}

M
Minghao Li 已提交
412 413
static void syncNodeEqPingTimer(void* param, void* tmrId) {
  SSyncNode* pSyncNode = (SSyncNode*)param;
M
Minghao Li 已提交
414
  if (atomic_load_64(&pSyncNode->pingTimerLogicClockUser) <= atomic_load_64(&pSyncNode->pingTimerLogicClock)) {
M
Minghao Li 已提交
415 416 417 418 419 420 421 422
    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
M
Minghao Li 已提交
423 424
    // pSyncNode->pingTimerMS += 100;

M
Minghao Li 已提交
425 426 427 428 429 430 431 432 433 434 435 436 437
    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);
M
Minghao Li 已提交
438 439

    SRpcMsg rpcMsg;
M
Minghao Li 已提交
440 441 442 443
    syncTimeout2RpcMsg(pSyncMsg, &rpcMsg);
    pSyncNode->FpEqMsg(pSyncNode->queue, &rpcMsg);
    syncTimeoutDestroy(pSyncMsg);

M
Minghao Li 已提交
444 445 446
    // reset timer ms
    pSyncNode->electTimerMS = syncUtilElectRandomMS();

M
Minghao Li 已提交
447 448 449
    taosTmrReset(syncNodeEqPingTimer, pSyncNode->pingTimerMS, pSyncNode, &gSyncEnv->pTimerManager,
                 &pSyncNode->pPingTimer);
  } else {
M
Minghao Li 已提交
450 451
    sTrace("syncNodeEqElectTimer: electTimerLogicClock:%lu, electTimerLogicClockUser:%lu",
           pSyncNode->electTimerLogicClock, pSyncNode->electTimerLogicClockUser);
M
Minghao Li 已提交
452 453 454
  }
}

M
Minghao Li 已提交
455 456 457 458 459 460 461 462 463 464 465 466 467 468 469
static void syncNodeEqHeartbeatTimer(void* param, void* tmrId) {
  SSyncNode* pSyncNode = (SSyncNode*)param;
  if (atomic_load_64(&pSyncNode->heartbeatTimerLogicClockUser) <=
      atomic_load_64(&pSyncNode->heartbeatTimerLogicClock)) {
    SyncTimeout* pSyncMsg =
        syncTimeoutBuild2(SYNC_TIMEOUT_HEARTBEAT, atomic_load_64(&pSyncNode->heartbeatTimerLogicClock),
                          pSyncNode->heartbeatTimerMS, pSyncNode);

    SRpcMsg rpcMsg;
    syncTimeout2RpcMsg(pSyncMsg, &rpcMsg);
    pSyncNode->FpEqMsg(pSyncNode->queue, &rpcMsg);
    syncTimeoutDestroy(pSyncMsg);

    // reset timer ms
    // pSyncNode->heartbeatTimerMS += 100;
M
Minghao Li 已提交
470

M
Minghao Li 已提交
471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486
    taosTmrReset(syncNodeEqHeartbeatTimer, pSyncNode->heartbeatTimerMS, pSyncNode, &gSyncEnv->pTimerManager,
                 &pSyncNode->pHeartbeatTimer);
  } else {
    sTrace("syncNodeEqHeartbeatTimer: heartbeatTimerLogicClock:%lu, heartbeatTimerLogicClockUser:%lu",
           pSyncNode->heartbeatTimerLogicClock, pSyncNode->heartbeatTimerLogicClockUser);
  }
}

static void UpdateTerm(SSyncNode* pSyncNode, SyncTerm term) {
  if (term > pSyncNode->pRaftStore->currentTerm) {
    pSyncNode->pRaftStore->currentTerm = term;
    pSyncNode->pRaftStore->voteFor = EMPTY_RAFT_ID;
    raftStorePersist(pSyncNode->pRaftStore);
    syncNodeBecomeFollower(pSyncNode);
  }
}
M
Minghao Li 已提交
487

M
Minghao Li 已提交
488 489
static void syncNodeBecomeFollower(SSyncNode* pSyncNode) {
  if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
M
Minghao Li 已提交
490
    pSyncNode->leaderCache = EMPTY_RAFT_ID;
M
Minghao Li 已提交
491 492 493
  }

  syncNodeStopHeartbeatTimer(pSyncNode);
M
Minghao Li 已提交
494 495
  int32_t electMS = syncUtilElectRandomMS();
  syncNodeStartElectTimer(pSyncNode, electMS);
M
Minghao Li 已提交
496 497
}

M
Minghao Li 已提交
498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515
// TLA+ Spec
// \* Candidate i transitions to leader.
// BecomeLeader(i) ==
//     /\ state[i] = Candidate
//     /\ votesGranted[i] \in Quorum
//     /\ state'      = [state EXCEPT ![i] = Leader]
//     /\ nextIndex'  = [nextIndex EXCEPT ![i] =
//                          [j \in Server |-> Len(log[i]) + 1]]
//     /\ matchIndex' = [matchIndex EXCEPT ![i] =
//                          [j \in Server |-> 0]]
//     /\ elections'  = elections \cup
//                          {[eterm     |-> currentTerm[i],
//                            eleader   |-> i,
//                            elog      |-> log[i],
//                            evotes    |-> votesGranted[i],
//                            evoterLog |-> voterLog[i]]}
//     /\ UNCHANGED <<messages, currentTerm, votedFor, candidateVars, logVars>>
//
M
Minghao Li 已提交
516 517
static void syncNodeBecomeLeader(SSyncNode* pSyncNode) {
  pSyncNode->state = TAOS_SYNC_STATE_LEADER;
M
Minghao Li 已提交
518
  pSyncNode->leaderCache = pSyncNode->myRaftId;
M
Minghao Li 已提交
519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537

  // next Index +=1
  // match Index = 0;

  syncNodeStopElectTimer(pSyncNode);
  syncNodeStartHeartbeatTimer(pSyncNode);

  // appendEntries;
}

static void syncNodeFollower2Candidate(SSyncNode* pSyncNode) {
  assert(pSyncNode->state == TAOS_SYNC_STATE_FOLLOWER);
  pSyncNode->state = TAOS_SYNC_STATE_CANDIDATE;
}

static void syncNodeCandidate2Leader(SSyncNode* pSyncNode) {}

static void syncNodeLeader2Follower(SSyncNode* pSyncNode) {}

M
Minghao Li 已提交
538
static void syncNodeCandidate2Follower(SSyncNode* pSyncNode) {}