syncMain.c 123.5 KB
Newer Older
M
Minghao Li 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
/*
 * 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/>.
 */

M
Minghao Li 已提交
16
#include "sync.h"
M
Minghao Li 已提交
17 18
#include "syncAppendEntries.h"
#include "syncAppendEntriesReply.h"
M
Minghao Li 已提交
19
#include "syncCommit.h"
M
Minghao Li 已提交
20
#include "syncElection.h"
M
Minghao Li 已提交
21
#include "syncEnv.h"
M
Minghao Li 已提交
22
#include "syncIndexMgr.h"
M
Minghao Li 已提交
23
#include "syncInt.h"
M
Minghao Li 已提交
24
#include "syncMessage.h"
M
Minghao Li 已提交
25
#include "syncRaftCfg.h"
M
Minghao Li 已提交
26
#include "syncRaftLog.h"
M
Minghao Li 已提交
27
#include "syncRaftStore.h"
M
Minghao Li 已提交
28
#include "syncReplication.h"
M
Minghao Li 已提交
29 30
#include "syncRequestVote.h"
#include "syncRequestVoteReply.h"
M
Minghao Li 已提交
31
#include "syncRespMgr.h"
M
Minghao Li 已提交
32
#include "syncSnapshot.h"
M
Minghao Li 已提交
33
#include "syncTimeout.h"
M
Minghao Li 已提交
34
#include "syncUtil.h"
M
Minghao Li 已提交
35
#include "syncVoteMgr.h"
M
Minghao Li 已提交
36
#include "tref.h"
M
Minghao Li 已提交
37

38
bool gRaftDetailLog = false;
39

M
Minghao Li 已提交
40 41 42
static int32_t tsNodeRefId = -1;

// ------ local funciton ---------
M
Minghao Li 已提交
43
// enqueue message ----
M
Minghao Li 已提交
44 45 46 47 48
static void    syncNodeEqPingTimer(void* param, void* tmrId);
static void    syncNodeEqElectTimer(void* param, void* tmrId);
static void    syncNodeEqHeartbeatTimer(void* param, void* tmrId);
static int32_t syncNodeEqNoop(SSyncNode* ths);
static int32_t syncNodeAppendNoop(SSyncNode* ths);
49
static void    syncNodeEqPeerHeartbeatTimer(void* param, void* tmrId);
M
Minghao Li 已提交
50

M
Minghao Li 已提交
51
// process message ----
M
Minghao Li 已提交
52 53
int32_t syncNodeOnPingCb(SSyncNode* ths, SyncPing* pMsg);
int32_t syncNodeOnPingReplyCb(SSyncNode* ths, SyncPingReply* pMsg);
M
Minghao Li 已提交
54

M
Minghao Li 已提交
55
// ---------------------------------
H
Hongze Cheng 已提交
56 57 58
static void syncNodeFreeCb(void* param) {
  syncNodeClose(param);
  param = NULL;
59
}
M
Minghao Li 已提交
60 61

int32_t syncInit() {
M
Minghao Li 已提交
62 63 64
  int32_t ret = 0;

  if (!syncEnvIsStart()) {
65
    tsNodeRefId = taosOpenRef(200, syncNodeFreeCb);
M
Minghao Li 已提交
66 67 68 69 70
    if (tsNodeRefId < 0) {
      sError("failed to init node ref");
      syncCleanUp();
      ret = -1;
    } else {
M
Minghao Li 已提交
71
      sDebug("sync rsetId:%d is open", tsNodeRefId);
M
Minghao Li 已提交
72 73
      ret = syncEnvStart();
    }
M
Minghao Li 已提交
74 75
  }

M
Minghao Li 已提交
76
  return ret;
M
Minghao Li 已提交
77
}
M
Minghao Li 已提交
78

M
Minghao Li 已提交
79 80
void syncCleanUp() {
  int32_t ret = syncEnvStop();
M
Minghao Li 已提交
81
  ASSERT(ret == 0);
M
Minghao Li 已提交
82 83

  if (tsNodeRefId != -1) {
M
Minghao Li 已提交
84
    sDebug("sync rsetId:%d is closed", tsNodeRefId);
M
Minghao Li 已提交
85 86 87
    taosCloseRef(tsNodeRefId);
    tsNodeRefId = -1;
  }
M
Minghao Li 已提交
88
}
M
Minghao Li 已提交
89

90
int64_t syncOpen(SSyncInfo* pSyncInfo) {
M
Minghao Li 已提交
91
  SSyncNode* pSyncNode = syncNodeOpen(pSyncInfo);
92 93 94 95
  if (pSyncNode == NULL) {
    sError("failed to open sync node. vgId:%d", pSyncInfo->vgId);
    return -1;
  }
M
Minghao Li 已提交
96

M
Minghao Li 已提交
97 98
  pSyncNode->rid = taosAddRef(tsNodeRefId, pSyncNode);
  if (pSyncNode->rid < 0) {
99 100
    syncNodeClose(pSyncNode);
    pSyncNode = NULL;
M
Minghao Li 已提交
101 102 103
    return -1;
  }

M
Minghao Li 已提交
104
  sDebug("vgId:%d, sync rid:%" PRId64 " is added to rsetId:%d", pSyncInfo->vgId, pSyncNode->rid, tsNodeRefId);
M
Minghao Li 已提交
105
  return pSyncNode->rid;
M
Minghao Li 已提交
106
}
M
Minghao Li 已提交
107

M
Minghao Li 已提交
108 109 110 111 112
void syncStart(int64_t rid) {
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
    return;
  }
M
Minghao Li 已提交
113 114 115 116 117 118 119 120 121 122 123 124 125 126 127

  if (pSyncNode->pRaftCfg->isStandBy) {
    syncNodeStartStandBy(pSyncNode);
  } else {
    syncNodeStart(pSyncNode);
  }

  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
}

void syncStartNormal(int64_t rid) {
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
    return;
  }
M
Minghao Li 已提交
128 129 130 131 132
  syncNodeStart(pSyncNode);

  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
}

M
Minghao Li 已提交
133 134 135 136 137 138 139 140 141 142
void syncStartStandBy(int64_t rid) {
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
    return;
  }
  syncNodeStartStandBy(pSyncNode);

  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
}

M
Minghao Li 已提交
143
void syncStop(int64_t rid) {
M
Minghao Li 已提交
144
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
S
Shengliang Guan 已提交
145 146
  if (pSyncNode == NULL) return;
  int32_t vgId = pSyncNode->vgId;
M
Minghao Li 已提交
147
  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
148

M
Minghao Li 已提交
149
  taosRemoveRef(tsNodeRefId, rid);
M
Minghao Li 已提交
150
  sDebug("vgId:%d, sync rid:%" PRId64 " is removed from rsetId:%d", vgId, rid, tsNodeRefId);
M
Minghao Li 已提交
151
}
M
Minghao Li 已提交
152

M
Minghao Li 已提交
153 154 155
int32_t syncSetStandby(int64_t rid) {
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
M
Minghao Li 已提交
156
    terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
S
Shengliang Guan 已提交
157
    sError("failed to set standby since accquire ref error, rid:%" PRId64, rid);
M
Minghao Li 已提交
158
    return -1;
M
Minghao Li 已提交
159 160
  }

161
  if (pSyncNode->state != TAOS_SYNC_STATE_FOLLOWER) {
162 163 164 165 166 167
    if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
      terrno = TSDB_CODE_SYN_IS_LEADER;
    } else {
      terrno = TSDB_CODE_SYN_STANDBY_NOT_READY;
    }
    sError("failed to set standby since it is not follower, state:%s rid:%" PRId64, syncStr(pSyncNode->state), rid);
M
Minghao Li 已提交
168
    taosReleaseRef(tsNodeRefId, pSyncNode->rid);
M
Minghao Li 已提交
169
    return -1;
M
Minghao Li 已提交
170 171 172 173 174 175 176 177 178 179 180 181 182 183 184
  }

  // state change
  pSyncNode->state = TAOS_SYNC_STATE_FOLLOWER;
  syncNodeStopHeartbeatTimer(pSyncNode);

  // reset elect timer, long enough
  int32_t electMS = TIMER_MAX_MS;
  int32_t ret = syncNodeRestartElectTimer(pSyncNode, electMS);
  ASSERT(ret == 0);

  pSyncNode->pRaftCfg->isStandBy = 1;
  raftCfgPersist(pSyncNode->pRaftCfg);

  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
185
  sInfo("vgId:%d, set to standby", pSyncNode->vgId);
M
Minghao Li 已提交
186 187 188
  return 0;
}

M
Minghao Li 已提交
189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205
bool syncNodeCheckNewConfig(SSyncNode* pSyncNode, const SSyncCfg* pNewCfg) {
  bool IamInNew = syncNodeInConfig(pSyncNode, pNewCfg);
  if (!IamInNew) {
    return false;
  }

  if (pNewCfg->replicaNum > pSyncNode->replicaNum + 1) {
    return false;
  }

  if (pNewCfg->replicaNum < pSyncNode->replicaNum - 1) {
    return false;
  }

  return true;
}

M
Minghao Li 已提交
206 207 208
int32_t syncReconfigBuild(int64_t rid, const SSyncCfg* pNewCfg, SRpcMsg* pRpcMsg) {
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
M
Minghao Li 已提交
209 210
    terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
    return -1;
M
Minghao Li 已提交
211 212
  }
  ASSERT(rid == pSyncNode->rid);
M
Minghao Li 已提交
213
  int32_t ret = 0;
214

M
Minghao Li 已提交
215
  if (!syncNodeCheckNewConfig(pSyncNode, pNewCfg)) {
M
Minghao Li 已提交
216
    taosReleaseRef(tsNodeRefId, pSyncNode->rid);
M
Minghao Li 已提交
217
    terrno = TSDB_CODE_SYN_NEW_CONFIG_ERROR;
218
    sError("invalid new config. vgId:%d", pSyncNode->vgId);
M
Minghao Li 已提交
219
    return -1;
M
Minghao Li 已提交
220 221 222 223 224 225 226 227 228
  }

  char* newconfig = syncCfg2Str((SSyncCfg*)pNewCfg);
  pRpcMsg->msgType = TDMT_SYNC_CONFIG_CHANGE;
  pRpcMsg->info.noResp = 1;
  pRpcMsg->contLen = strlen(newconfig) + 1;
  pRpcMsg->pCont = rpcMallocCont(pRpcMsg->contLen);
  snprintf(pRpcMsg->pCont, pRpcMsg->contLen, "%s", newconfig);
  taosMemoryFree(newconfig);
229

M
Minghao Li 已提交
230 231 232 233
  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
  return ret;
}

S
Shengliang Guan 已提交
234
int32_t syncReconfig(int64_t rid, SSyncCfg* pNewCfg) {
M
Minghao Li 已提交
235 236
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
M
Minghao Li 已提交
237 238
    terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
    return -1;
M
Minghao Li 已提交
239 240 241
  }
  ASSERT(rid == pSyncNode->rid);

M
Minghao Li 已提交
242
  if (!syncNodeCheckNewConfig(pSyncNode, pNewCfg)) {
M
Minghao Li 已提交
243
    taosReleaseRef(tsNodeRefId, pSyncNode->rid);
M
Minghao Li 已提交
244
    terrno = TSDB_CODE_SYN_NEW_CONFIG_ERROR;
245
    sError("invalid new config. vgId:%d", pSyncNode->vgId);
M
Minghao Li 已提交
246
    return -1;
M
Minghao Li 已提交
247
  }
248

S
Shengliang Guan 已提交
249
#if 0
M
Minghao Li 已提交
250
  char*   newconfig = syncCfg2Str((SSyncCfg*)pNewCfg);
M
Minghao Li 已提交
251 252
  int32_t ret = 0;

M
Minghao Li 已提交
253
  SRpcMsg rpcMsg = {0};
254
  rpcMsg.msgType = TDMT_SYNC_CONFIG_CHANGE;
S
Shengliang Guan 已提交
255
  rpcMsg.info.noResp = 1;
256
  rpcMsg.contLen = strlen(newconfig) + 1;
M
Minghao Li 已提交
257
  rpcMsg.pCont = rpcMallocCont(rpcMsg.contLen);
258 259
  snprintf(rpcMsg.pCont, rpcMsg.contLen, "%s", newconfig);
  taosMemoryFree(newconfig);
M
Minghao Li 已提交
260 261 262
  ret = syncNodePropose(pSyncNode, &rpcMsg, false);

  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
M
Minghao Li 已提交
263
  return ret;
S
Shengliang Guan 已提交
264 265 266
#else
  syncNodeUpdateNewConfigIndex(pSyncNode, pNewCfg);
  syncNodeDoConfigChange(pSyncNode, pNewCfg, SYNC_INDEX_INVALID);
M
Minghao Li 已提交
267 268 269 270 271 272 273 274 275 276 277
  if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
    syncNodeStopHeartbeatTimer(pSyncNode);

    for (int32_t i = 0; i < TSDB_MAX_REPLICA; ++i) {
      syncHbTimerInit(pSyncNode, &(pSyncNode->peerHeartbeatTimerArr[i]), (pSyncNode->replicasId)[i]);
    }

    syncNodeStartHeartbeatTimer(pSyncNode);

    syncNodeReplicate(pSyncNode);
  }
S
Shengliang Guan 已提交
278 279 280
  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
  return 0;
#endif
M
Minghao Li 已提交
281
}
M
Minghao Li 已提交
282

283
int32_t syncLeaderTransfer(int64_t rid) {
M
Minghao Li 已提交
284 285
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
M
Minghao Li 已提交
286 287
    terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
    return -1;
M
Minghao Li 已提交
288 289 290
  }
  ASSERT(rid == pSyncNode->rid);

M
Minghao Li 已提交
291
  int32_t ret = syncNodeLeaderTransfer(pSyncNode);
M
Minghao Li 已提交
292
  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
293 294 295 296 297 298
  return ret;
}

int32_t syncLeaderTransferTo(int64_t rid, SNodeInfo newLeader) {
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
M
Minghao Li 已提交
299 300
    terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
    return -1;
301
  }
M
Minghao Li 已提交
302
  ASSERT(rid == pSyncNode->rid);
303

M
Minghao Li 已提交
304
  int32_t ret = syncNodeLeaderTransferTo(pSyncNode, newLeader);
305 306 307 308
  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
  return ret;
}

M
Minghao Li 已提交
309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324
SyncIndex syncMinMatchIndex(SSyncNode* pSyncNode) {
  SyncIndex minMatchIndex = SYNC_INDEX_INVALID;

  if (pSyncNode->peersNum > 0) {
    minMatchIndex = syncIndexMgrGetIndex(pSyncNode->pMatchIndex, &(pSyncNode->peersId[0]));
  }

  for (int32_t i = 1; i < pSyncNode->peersNum; ++i) {
    SyncIndex matchIndex = syncIndexMgrGetIndex(pSyncNode->pMatchIndex, &(pSyncNode->peersId[i]));
    if (matchIndex < minMatchIndex) {
      minMatchIndex = matchIndex;
    }
  }
  return minMatchIndex;
}

325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341
char* syncNodePeerState2Str(const SSyncNode* pSyncNode) {
  int32_t len = 128;
  int32_t useLen = 0;
  int32_t leftLen = len - useLen;
  char*   pStr = taosMemoryMalloc(len);
  memset(pStr, 0, len);

  char*   p = pStr;
  int32_t use = snprintf(p, leftLen, "{");
  useLen += use;
  leftLen -= use;

  for (int32_t i = 0; i < pSyncNode->replicaNum; ++i) {
    SPeerState* pState = syncNodeGetPeerState((SSyncNode*)pSyncNode, &(pSyncNode->replicasId[i]));
    ASSERT(pState != NULL);

    p = pStr + useLen;
S
Shengliang Guan 已提交
342
    use = snprintf(p, leftLen, "%d:%" PRId64 " ,%" PRId64, i, pState->lastSendIndex, pState->lastSendTime);
343 344 345 346 347 348 349 350 351 352 353 354 355 356
    useLen += use;
    leftLen -= use;
  }

  p = pStr + useLen;
  use = snprintf(p, leftLen, "}");
  useLen += use;
  leftLen -= use;

  // sTrace("vgId:%d, ------------------ syncNodePeerState2Str:%s", pSyncNode->vgId, pStr);

  return pStr;
}

357 358 359 360 361 362 363 364 365
int32_t syncBeginSnapshot(int64_t rid, int64_t lastApplyIndex) {
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
    terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
    return -1;
  }
  ASSERT(rid == pSyncNode->rid);
  int32_t code = 0;

M
Minghao Li 已提交
366
  if (syncNodeIsMnode(pSyncNode)) {
M
Minghao Li 已提交
367 368 369
    // mnode
    int64_t logRetention = SYNC_MNODE_LOG_RETENTION;

M
Minghao Li 已提交
370 371 372
    SyncIndex beginIndex = pSyncNode->pLogStore->syncLogBeginIndex(pSyncNode->pLogStore);
    SyncIndex endIndex = pSyncNode->pLogStore->syncLogEndIndex(pSyncNode->pLogStore);
    int64_t   logNum = endIndex - beginIndex;
M
Minghao Li 已提交
373 374 375
    bool      isEmpty = pSyncNode->pLogStore->syncLogIsEmpty(pSyncNode->pLogStore);

    if (isEmpty || (!isEmpty && logNum < logRetention)) {
M
Minghao Li 已提交
376
      char logBuf[256];
S
Shengliang Guan 已提交
377 378 379
      snprintf(logBuf, sizeof(logBuf),
               "new-snapshot-index:%" PRId64 ", log-num:%" PRId64 ", empty:%d, do not delete wal", lastApplyIndex,
               logNum, isEmpty);
M
Minghao Li 已提交
380 381 382 383 384 385
      syncNodeEventLog(pSyncNode, logBuf);

      taosReleaseRef(tsNodeRefId, pSyncNode->rid);
      return 0;
    }

M
Minghao Li 已提交
386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404
    goto _DEL_WAL;

  } else {
    // vnode
    if (pSyncNode->replicaNum > 1) {
      // multi replicas

      if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
        pSyncNode->minMatchIndex = syncMinMatchIndex(pSyncNode);

        for (int32_t i = 0; i < pSyncNode->peersNum; ++i) {
          int64_t matchIndex = syncIndexMgrGetIndex(pSyncNode->pMatchIndex, &(pSyncNode->peersId[i]));
          if (lastApplyIndex > matchIndex) {
            do {
              char     host[64];
              uint16_t port;
              syncUtilU642Addr(pSyncNode->peersId[i].addr, host, sizeof(host), &port);
              char logBuf[256];
              snprintf(logBuf, sizeof(logBuf),
S
Shengliang Guan 已提交
405 406
                       "new-snapshot-index:%" PRId64 " is greater than match-index:%" PRId64
                       " of %s:%d, do not delete wal",
M
Minghao Li 已提交
407 408 409 410 411 412 413 414 415 416 417 418 419
                       lastApplyIndex, matchIndex, host, port);
              syncNodeEventLog(pSyncNode, logBuf);
            } while (0);

            taosReleaseRef(tsNodeRefId, pSyncNode->rid);
            return 0;
          }
        }

      } else if (pSyncNode->state == TAOS_SYNC_STATE_FOLLOWER) {
        if (lastApplyIndex > pSyncNode->minMatchIndex) {
          char logBuf[256];
          snprintf(logBuf, sizeof(logBuf),
S
Shengliang Guan 已提交
420 421
                   "new-snapshot-index:%" PRId64 " is greater than min-match-index:%" PRId64 ", do not delete wal",
                   lastApplyIndex, pSyncNode->minMatchIndex);
M
Minghao Li 已提交
422 423 424 425 426 427 428
          syncNodeEventLog(pSyncNode, logBuf);

          taosReleaseRef(tsNodeRefId, pSyncNode->rid);
          return 0;
        }

      } else if (pSyncNode->state == TAOS_SYNC_STATE_CANDIDATE) {
429
        char logBuf[256];
S
Shengliang Guan 已提交
430
        snprintf(logBuf, sizeof(logBuf), "new-snapshot-index:%" PRId64 " candidate, do not delete wal", lastApplyIndex);
431 432
        syncNodeEventLog(pSyncNode, logBuf);

M
Minghao Li 已提交
433 434 435 436 437
        taosReleaseRef(tsNodeRefId, pSyncNode->rid);
        return 0;

      } else {
        char logBuf[256];
S
Shengliang Guan 已提交
438 439
        snprintf(logBuf, sizeof(logBuf), "new-snapshot-index:%" PRId64 " unknown state, do not delete wal",
                 lastApplyIndex);
M
Minghao Li 已提交
440 441 442 443 444 445 446 447 448 449 450 451
        syncNodeEventLog(pSyncNode, logBuf);

        taosReleaseRef(tsNodeRefId, pSyncNode->rid);
        return 0;
      }

      goto _DEL_WAL;

    } else {
      // one replica

      goto _DEL_WAL;
452 453 454
    }
  }

M
Minghao Li 已提交
455
_DEL_WAL:
456

M
Minghao Li 已提交
457
  do {
458 459 460 461
    SyncIndex snapshottingIndex = atomic_load_64(&pSyncNode->snapshottingIndex);

    if (snapshottingIndex == SYNC_INDEX_INVALID) {
      atomic_store_64(&pSyncNode->snapshottingIndex, lastApplyIndex);
M
Minghao Li 已提交
462
      pSyncNode->snapshottingTime = taosGetTimestampMs();
463

M
Minghao Li 已提交
464 465 466
      SSyncLogStoreData* pData = pSyncNode->pLogStore->data;
      code = walBeginSnapshot(pData->pWal, lastApplyIndex);
      if (code == 0) {
467
        char logBuf[256];
S
Shengliang Guan 已提交
468
        snprintf(logBuf, sizeof(logBuf), "wal snapshot begin, index:%" PRId64 ", last apply index:%" PRId64,
469 470 471
                 pSyncNode->snapshottingIndex, lastApplyIndex);
        syncNodeEventLog(pSyncNode, logBuf);

M
Minghao Li 已提交
472 473
      } else {
        char logBuf[256];
S
Shengliang Guan 已提交
474 475 476
        snprintf(logBuf, sizeof(logBuf),
                 "wal snapshot begin error since:%s, index:%" PRId64 ", last apply index:%" PRId64, terrstr(terrno),
                 pSyncNode->snapshottingIndex, lastApplyIndex);
M
Minghao Li 已提交
477 478 479 480
        syncNodeErrorLog(pSyncNode, logBuf);

        atomic_store_64(&pSyncNode->snapshottingIndex, SYNC_INDEX_INVALID);
      }
481 482

    } else {
483
      char logBuf[256];
S
Shengliang Guan 已提交
484 485 486
      snprintf(logBuf, sizeof(logBuf),
               "snapshotting for %" PRId64 ", do not delete wal for new-snapshot-index:%" PRId64, snapshottingIndex,
               lastApplyIndex);
487
      syncNodeEventLog(pSyncNode, logBuf);
488
    }
M
Minghao Li 已提交
489
  } while (0);
490 491 492 493 494 495 496 497 498 499 500 501 502

  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
  return code;
}

int32_t syncEndSnapshot(int64_t rid) {
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
    terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
    return -1;
  }
  ASSERT(rid == pSyncNode->rid);

503 504 505 506
  int32_t code = 0;
  if (atomic_load_64(&pSyncNode->snapshottingIndex) != SYNC_INDEX_INVALID) {
    SSyncLogStoreData* pData = pSyncNode->pLogStore->data;
    code = walEndSnapshot(pData->pWal);
M
Minghao Li 已提交
507
    if (code != 0) {
M
Minghao Li 已提交
508
      sError("vgId:%d, wal snapshot end error since:%s", pSyncNode->vgId, terrstr(terrno));
M
Minghao Li 已提交
509 510 511 512 513 514

      taosReleaseRef(tsNodeRefId, pSyncNode->rid);
      return -1;
    } else {
      do {
        char logBuf[256];
S
Shengliang Guan 已提交
515 516
        snprintf(logBuf, sizeof(logBuf), "wal snapshot end, index:%" PRId64,
                 atomic_load_64(&pSyncNode->snapshottingIndex));
M
Minghao Li 已提交
517 518
        syncNodeEventLog(pSyncNode, logBuf);
      } while (0);
519

M
Minghao Li 已提交
520 521
      atomic_store_64(&pSyncNode->snapshottingIndex, SYNC_INDEX_INVALID);
    }
522
  }
523 524 525 526 527

  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
  return code;
}

M
Minghao Li 已提交
528 529
int32_t syncNodeLeaderTransfer(SSyncNode* pSyncNode) {
  if (pSyncNode->peersNum == 0) {
530
    sDebug("only one replica, cannot leader transfer");
M
Minghao Li 已提交
531 532 533 534 535 536 537 538 539 540 541 542 543
    terrno = TSDB_CODE_SYN_ONE_REPLICA;
    return -1;
  }

  SNodeInfo newLeader = (pSyncNode->peersNodeInfo)[0];
  int32_t   ret = syncNodeLeaderTransferTo(pSyncNode, newLeader);
  return ret;
}

int32_t syncNodeLeaderTransferTo(SSyncNode* pSyncNode, SNodeInfo newLeader) {
  int32_t ret = 0;

  if (pSyncNode->replicaNum == 1) {
544
    sDebug("only one replica, cannot leader transfer");
M
Minghao Li 已提交
545 546 547 548
    terrno = TSDB_CODE_SYN_ONE_REPLICA;
    return -1;
  }

M
Minghao Li 已提交
549 550 551 552 553 554
  do {
    char logBuf[128];
    snprintf(logBuf, sizeof(logBuf), "begin leader transfer to %s:%u", newLeader.nodeFqdn, newLeader.nodePort);
    syncNodeEventLog(pSyncNode, logBuf);
  } while (0);

M
Minghao Li 已提交
555 556 557 558 559 560 561 562 563 564 565 566 567
  SyncLeaderTransfer* pMsg = syncLeaderTransferBuild(pSyncNode->vgId);
  pMsg->newLeaderId.addr = syncUtilAddr2U64(newLeader.nodeFqdn, newLeader.nodePort);
  pMsg->newLeaderId.vgId = pSyncNode->vgId;
  pMsg->newNodeInfo = newLeader;
  ASSERT(pMsg != NULL);
  SRpcMsg rpcMsg = {0};
  syncLeaderTransfer2RpcMsg(pMsg, &rpcMsg);
  syncLeaderTransferDestroy(pMsg);

  ret = syncNodePropose(pSyncNode, &rpcMsg, false);
  return ret;
}

568 569 570 571 572
bool syncCanLeaderTransfer(int64_t rid) {
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
    return false;
  }
M
Minghao Li 已提交
573
  ASSERT(rid == pSyncNode->rid);
574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599

  if (pSyncNode->replicaNum == 1) {
    taosReleaseRef(tsNodeRefId, pSyncNode->rid);
    return false;
  }

  if (pSyncNode->state == TAOS_SYNC_STATE_FOLLOWER) {
    taosReleaseRef(tsNodeRefId, pSyncNode->rid);
    return true;
  }

  bool matchOK = true;
  if (pSyncNode->state == TAOS_SYNC_STATE_CANDIDATE || pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
    SyncIndex myCommitIndex = pSyncNode->commitIndex;
    for (int i = 0; i < pSyncNode->peersNum; ++i) {
      SyncIndex peerMatchIndex = syncIndexMgrGetIndex(pSyncNode->pMatchIndex, &(pSyncNode->peersId)[i]);
      if (peerMatchIndex < myCommitIndex) {
        matchOK = false;
      }
    }
  }

  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
  return matchOK;
}

600
int32_t syncForwardToPeer(int64_t rid, SRpcMsg* pMsg, bool isWeak) {
M
Minghao Li 已提交
601 602 603
  int32_t ret = syncPropose(rid, pMsg, isWeak);
  return ret;
}
M
Minghao Li 已提交
604

M
Minghao Li 已提交
605 606 607 608 609
ESyncState syncGetMyRole(int64_t rid) {
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
    return TAOS_SYNC_STATE_ERROR;
  }
M
Minghao Li 已提交
610
  ASSERT(rid == pSyncNode->rid);
M
Minghao Li 已提交
611 612 613 614
  ESyncState state = pSyncNode->state;

  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
  return state;
M
Minghao Li 已提交
615 616
}

M
Minghao Li 已提交
617 618 619 620 621
bool syncIsReady(int64_t rid) {
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
    return false;
  }
M
Minghao Li 已提交
622
  ASSERT(rid == pSyncNode->rid);
M
Minghao Li 已提交
623 624
  bool b = (pSyncNode->state == TAOS_SYNC_STATE_LEADER) && pSyncNode->restoreFinish;
  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
625 626 627 628 629 630 631 632 633

  // if false, set error code
  if (false == b) {
    if (pSyncNode->state != TAOS_SYNC_STATE_LEADER) {
      terrno = TSDB_CODE_SYN_NOT_LEADER;
    } else {
      terrno = TSDB_CODE_APP_NOT_READY;
    }
  }
M
Minghao Li 已提交
634 635 636
  return b;
}

M
Minghao Li 已提交
637 638 639 640 641
bool syncIsRestoreFinish(int64_t rid) {
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
    return false;
  }
M
Minghao Li 已提交
642
  ASSERT(rid == pSyncNode->rid);
M
Minghao Li 已提交
643 644 645 646 647 648
  bool b = pSyncNode->restoreFinish;

  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
  return b;
}

649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680
int32_t syncGetSnapshotByIndex(int64_t rid, SyncIndex index, SSnapshot* pSnapshot) {
  if (index < SYNC_INDEX_BEGIN) {
    return -1;
  }

  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
    return -1;
  }
  ASSERT(rid == pSyncNode->rid);

  SSyncRaftEntry* pEntry = NULL;
  int32_t         code = pSyncNode->pLogStore->syncLogGetEntry(pSyncNode->pLogStore, index, &pEntry);
  if (code != 0) {
    if (pEntry != NULL) {
      syncEntryDestory(pEntry);
    }
    taosReleaseRef(tsNodeRefId, pSyncNode->rid);
    return -1;
  }
  ASSERT(pEntry != NULL);

  pSnapshot->data = NULL;
  pSnapshot->lastApplyIndex = index;
  pSnapshot->lastApplyTerm = pEntry->term;
  pSnapshot->lastConfigIndex = syncNodeGetSnapshotConfigIndex(pSyncNode, index);

  syncEntryDestory(pEntry);
  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
  return 0;
}

681 682 683 684 685
int32_t syncGetSnapshotMeta(int64_t rid, struct SSnapshotMeta* sMeta) {
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
    return -1;
  }
M
Minghao Li 已提交
686
  ASSERT(rid == pSyncNode->rid);
687 688
  sMeta->lastConfigIndex = pSyncNode->pRaftCfg->lastConfigIndex;

S
Shengliang Guan 已提交
689
  sTrace("vgId:%d, get snapshot meta, lastConfigIndex:%" PRId64, pSyncNode->vgId, pSyncNode->pRaftCfg->lastConfigIndex);
690 691 692 693 694

  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
  return 0;
}

695 696 697 698 699
int32_t syncGetSnapshotMetaByIndex(int64_t rid, SyncIndex snapshotIndex, struct SSnapshotMeta* sMeta) {
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
    return -1;
  }
M
Minghao Li 已提交
700
  ASSERT(rid == pSyncNode->rid);
701 702 703 704 705 706 707 708 709 710 711

  ASSERT(pSyncNode->pRaftCfg->configIndexCount >= 1);
  SyncIndex lastIndex = (pSyncNode->pRaftCfg->configIndexArr)[0];

  for (int i = 0; i < pSyncNode->pRaftCfg->configIndexCount; ++i) {
    if ((pSyncNode->pRaftCfg->configIndexArr)[i] > lastIndex &&
        (pSyncNode->pRaftCfg->configIndexArr)[i] <= snapshotIndex) {
      lastIndex = (pSyncNode->pRaftCfg->configIndexArr)[i];
    }
  }
  sMeta->lastConfigIndex = lastIndex;
712
  sTrace("vgId:%d, get snapshot meta by index:%" PRId64 " lcindex:%" PRId64, pSyncNode->vgId, snapshotIndex,
S
Shengliang Guan 已提交
713
         sMeta->lastConfigIndex);
714 715 716 717 718

  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
  return 0;
}

719 720 721 722 723 724 725 726 727 728
SyncIndex syncNodeGetSnapshotConfigIndex(SSyncNode* pSyncNode, SyncIndex snapshotLastApplyIndex) {
  ASSERT(pSyncNode->pRaftCfg->configIndexCount >= 1);
  SyncIndex lastIndex = (pSyncNode->pRaftCfg->configIndexArr)[0];

  for (int i = 0; i < pSyncNode->pRaftCfg->configIndexCount; ++i) {
    if ((pSyncNode->pRaftCfg->configIndexArr)[i] > lastIndex &&
        (pSyncNode->pRaftCfg->configIndexArr)[i] <= snapshotLastApplyIndex) {
      lastIndex = (pSyncNode->pRaftCfg->configIndexArr)[i];
    }
  }
S
Shengliang Guan 已提交
729
  sTrace("vgId:%d, sync get last config index, index:%" PRId64 " lcindex:%" PRId64, pSyncNode->vgId,
S
Shengliang Guan 已提交
730
         snapshotLastApplyIndex, lastIndex);
731 732 733 734

  return lastIndex;
}

M
Minghao Li 已提交
735 736 737 738 739
const char* syncGetMyRoleStr(int64_t rid) {
  const char* s = syncUtilState2String(syncGetMyRole(rid));
  return s;
}

M
Minghao Li 已提交
740 741 742 743 744 745 746 747 748 749 750 751
bool syncRestoreFinish(int64_t rid) {
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
    return false;
  }
  ASSERT(rid == pSyncNode->rid);
  bool restoreFinish = pSyncNode->restoreFinish;

  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
  return restoreFinish;
}

M
Minghao Li 已提交
752
SyncTerm syncGetMyTerm(int64_t rid) {
M
Minghao Li 已提交
753 754 755 756
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
    return TAOS_SYNC_STATE_ERROR;
  }
M
Minghao Li 已提交
757
  ASSERT(rid == pSyncNode->rid);
M
Minghao Li 已提交
758
  SyncTerm term = pSyncNode->pRaftStore->currentTerm;
M
Minghao Li 已提交
759 760

  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
M
Minghao Li 已提交
761
  return term;
M
Minghao Li 已提交
762 763
}

764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787
SyncIndex syncGetLastIndex(int64_t rid) {
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
    return SYNC_INDEX_INVALID;
  }
  ASSERT(rid == pSyncNode->rid);
  SyncIndex lastIndex = syncNodeGetLastIndex(pSyncNode);

  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
  return lastIndex;
}

SyncIndex syncGetCommitIndex(int64_t rid) {
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
    return SYNC_INDEX_INVALID;
  }
  ASSERT(rid == pSyncNode->rid);
  SyncIndex cmtIndex = pSyncNode->commitIndex;

  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
  return cmtIndex;
}

M
Minghao Li 已提交
788
SyncGroupId syncGetVgId(int64_t rid) {
M
Minghao Li 已提交
789 790
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
M
Minghao Li 已提交
791 792
    return TAOS_SYNC_STATE_ERROR;
  }
M
Minghao Li 已提交
793
  ASSERT(rid == pSyncNode->rid);
M
Minghao Li 已提交
794
  SyncGroupId vgId = pSyncNode->vgId;
M
Minghao Li 已提交
795 796

  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
M
Minghao Li 已提交
797
  return vgId;
M
Minghao Li 已提交
798 799
}

M
Minghao Li 已提交
800 801 802 803 804 805
void syncGetEpSet(int64_t rid, SEpSet* pEpSet) {
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
    memset(pEpSet, 0, sizeof(*pEpSet));
    return;
  }
M
Minghao Li 已提交
806
  ASSERT(rid == pSyncNode->rid);
M
Minghao Li 已提交
807 808
  pEpSet->numOfEps = 0;
  for (int i = 0; i < pSyncNode->pRaftCfg->cfg.replicaNum; ++i) {
M
Minghao Li 已提交
809 810
    snprintf(pEpSet->eps[i].fqdn, sizeof(pEpSet->eps[i].fqdn), "%s", (pSyncNode->pRaftCfg->cfg.nodeInfo)[i].nodeFqdn);
    pEpSet->eps[i].port = (pSyncNode->pRaftCfg->cfg.nodeInfo)[i].nodePort;
M
Minghao Li 已提交
811
    (pEpSet->numOfEps)++;
S
Shengliang Guan 已提交
812
    sInfo("vgId:%d, sync get epset: index:%d %s:%d", pSyncNode->vgId, i, pEpSet->eps[i].fqdn, pEpSet->eps[i].port);
M
Minghao Li 已提交
813 814
  }
  pEpSet->inUse = pSyncNode->pRaftCfg->cfg.myIndex;
S
Shengliang Guan 已提交
815
  sInfo("vgId:%d, sync get epset in-use:%d", pSyncNode->vgId, pEpSet->inUse);
816 817 818

  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
}
M
Minghao Li 已提交
819

820 821 822 823 824 825 826 827 828 829 830 831
void syncGetRetryEpSet(int64_t rid, SEpSet* pEpSet) {
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
    memset(pEpSet, 0, sizeof(*pEpSet));
    return;
  }
  ASSERT(rid == pSyncNode->rid);
  pEpSet->numOfEps = 0;
  for (int i = 0; i < pSyncNode->pRaftCfg->cfg.replicaNum; ++i) {
    snprintf(pEpSet->eps[i].fqdn, sizeof(pEpSet->eps[i].fqdn), "%s", (pSyncNode->pRaftCfg->cfg.nodeInfo)[i].nodeFqdn);
    pEpSet->eps[i].port = (pSyncNode->pRaftCfg->cfg.nodeInfo)[i].nodePort;
    (pEpSet->numOfEps)++;
M
Minghao Li 已提交
832 833
    sInfo("vgId:%d, sync get retry epset: index:%d %s:%d", pSyncNode->vgId, i, pEpSet->eps[i].fqdn,
          pEpSet->eps[i].port);
834 835
  }
  pEpSet->inUse = (pSyncNode->pRaftCfg->cfg.myIndex + 1) % pEpSet->numOfEps;
M
Minghao Li 已提交
836
  sInfo("vgId:%d, sync get retry epset in-use:%d", pSyncNode->vgId, pEpSet->inUse);
M
Minghao Li 已提交
837 838 839 840

  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
}

M
Minghao Li 已提交
841 842 843 844 845
int32_t syncGetRespRpc(int64_t rid, uint64_t index, SRpcMsg* msg) {
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
    return TAOS_SYNC_STATE_ERROR;
  }
M
Minghao Li 已提交
846
  ASSERT(rid == pSyncNode->rid);
M
Minghao Li 已提交
847 848 849 850 851 852 853 854 855 856 857

  SRespStub stub;
  int32_t   ret = syncRespMgrGet(pSyncNode->pSyncRespMgr, index, &stub);
  if (ret == 1) {
    memcpy(msg, &(stub.rpcMsg), sizeof(SRpcMsg));
  }

  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
  return ret;
}

S
Shengliang Guan 已提交
858
int32_t syncGetAndDelRespRpc(int64_t rid, uint64_t index, SRpcHandleInfo* pInfo) {
M
Minghao Li 已提交
859 860 861 862
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
    return TAOS_SYNC_STATE_ERROR;
  }
M
Minghao Li 已提交
863
  ASSERT(rid == pSyncNode->rid);
M
Minghao Li 已提交
864 865 866 867

  SRespStub stub;
  int32_t   ret = syncRespMgrGetAndDel(pSyncNode->pSyncRespMgr, index, &stub);
  if (ret == 1) {
S
Shengliang Guan 已提交
868
    *pInfo = stub.rpcMsg.info;
M
Minghao Li 已提交
869 870
  }

S
Shengliang Guan 已提交
871
  sTrace("vgId:%d, get seq:%" PRIu64 " rpc handle:%p", pSyncNode->vgId, index, pInfo->handle);
M
Minghao Li 已提交
872 873 874 875
  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
  return ret;
}

876
void syncSetMsgCb(int64_t rid, const SMsgCb* msgcb) {
M
Minghao Li 已提交
877 878
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
S
Shengliang Guan 已提交
879
    sTrace("syncSetQ get pSyncNode is NULL, rid:%" PRId64, rid);
M
Minghao Li 已提交
880 881
    return;
  }
M
Minghao Li 已提交
882
  ASSERT(rid == pSyncNode->rid);
S
Shengliang Guan 已提交
883
  pSyncNode->msgcb = msgcb;
M
Minghao Li 已提交
884 885 886 887 888 889 890

  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
}

char* sync2SimpleStr(int64_t rid) {
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
S
Shengliang Guan 已提交
891
    sTrace("syncSetRpc get pSyncNode is NULL, rid:%" PRId64, rid);
M
Minghao Li 已提交
892 893
    return NULL;
  }
M
Minghao Li 已提交
894
  ASSERT(rid == pSyncNode->rid);
M
Minghao Li 已提交
895 896 897 898 899 900 901 902 903 904 905
  char* s = syncNode2SimpleStr(pSyncNode);
  taosReleaseRef(tsNodeRefId, pSyncNode->rid);

  return s;
}

void setPingTimerMS(int64_t rid, int32_t pingTimerMS) {
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
    return;
  }
M
Minghao Li 已提交
906
  ASSERT(rid == pSyncNode->rid);
M
Minghao Li 已提交
907 908 909 910 911 912 913 914 915 916 917
  pSyncNode->pingBaseLine = pingTimerMS;
  pSyncNode->pingTimerMS = pingTimerMS;

  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
}

void setElectTimerMS(int64_t rid, int32_t electTimerMS) {
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
    return;
  }
M
Minghao Li 已提交
918
  ASSERT(rid == pSyncNode->rid);
M
Minghao Li 已提交
919 920 921 922 923 924 925 926 927 928
  pSyncNode->electBaseLine = electTimerMS;

  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
}

void setHeartbeatTimerMS(int64_t rid, int32_t hbTimerMS) {
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
    return;
  }
M
Minghao Li 已提交
929
  ASSERT(rid == pSyncNode->rid);
M
Minghao Li 已提交
930 931 932 933 934 935
  pSyncNode->hbBaseLine = hbTimerMS;
  pSyncNode->heartbeatTimerMS = hbTimerMS;

  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
}

M
Minghao Li 已提交
936
int32_t syncPropose(int64_t rid, SRpcMsg* pMsg, bool isWeak) {
937
  SSyncNode* pSyncNode = taosAcquireRef(tsNodeRefId, rid);
938
  if (pSyncNode == NULL) {
M
Minghao Li 已提交
939
    taosReleaseRef(tsNodeRefId, rid);
M
Minghao Li 已提交
940 941
    terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
    return -1;
942
  }
M
Minghao Li 已提交
943
  ASSERT(rid == pSyncNode->rid);
M
Minghao Li 已提交
944

945
  int32_t ret = syncNodePropose(pSyncNode, pMsg, isWeak);
M
Minghao Li 已提交
946 947 948 949
  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
  return ret;
}

950
static bool syncNodeBatchOK(SRpcMsg** pMsgPArr, int32_t arrSize) {
M
Minghao Li 已提交
951
  for (int32_t i = 0; i < arrSize; ++i) {
952
    if (pMsgPArr[i]->msgType == TDMT_SYNC_CONFIG_CHANGE) {
M
Minghao Li 已提交
953 954 955
      return false;
    }

956
    if (pMsgPArr[i]->msgType == TDMT_SYNC_CONFIG_CHANGE_FINISH) {
M
Minghao Li 已提交
957 958 959 960 961 962 963
      return false;
    }
  }

  return true;
}

964
int32_t syncNodePropose(SSyncNode* pSyncNode, SRpcMsg* pMsg, bool isWeak) {
M
Minghao Li 已提交
965
  int32_t ret = 0;
M
Minghao Li 已提交
966

M
Minghao Li 已提交
967 968
  do {
    char eventLog[128];
S
Shengliang Guan 已提交
969
    snprintf(eventLog, sizeof(eventLog), "propose message, type:%s", TMSG_INFO(pMsg->msgType));
M
Minghao Li 已提交
970 971
    syncNodeEventLog(pSyncNode, eventLog);
  } while (0);
M
Minghao Li 已提交
972

M
Minghao Li 已提交
973
  if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
M
Minghao Li 已提交
974 975 976
    if (pSyncNode->changing && pMsg->msgType != TDMT_SYNC_CONFIG_CHANGE_FINISH) {
      ret = -1;
      terrno = TSDB_CODE_SYN_PROPOSE_NOT_READY;
S
Shengliang Guan 已提交
977
      sError("vgId:%d, failed to sync propose since not ready, type:%s", pSyncNode->vgId, TMSG_INFO(pMsg->msgType));
M
Minghao Li 已提交
978 979 980 981 982 983 984 985
      goto _END;
    }

    // config change
    if (pMsg->msgType == TDMT_SYNC_CONFIG_CHANGE) {
      if (!syncNodeCanChange(pSyncNode)) {
        ret = -1;
        terrno = TSDB_CODE_SYN_RECONFIG_NOT_READY;
S
Shengliang Guan 已提交
986
        sError("vgId:%d, failed to sync reconfig since not ready, type:%s", pSyncNode->vgId, TMSG_INFO(pMsg->msgType));
M
Minghao Li 已提交
987 988 989 990 991 992 993
        goto _END;
      }

      ASSERT(!pSyncNode->changing);
      pSyncNode->changing = true;
    }

994 995
    // not restored, vnode enable
    if (!pSyncNode->restoreFinish && pSyncNode->vgId != 1) {
996 997
      ret = -1;
      terrno = TSDB_CODE_SYN_PROPOSE_NOT_READY;
S
Shengliang Guan 已提交
998 999
      sError("vgId:%d, failed to sync propose since not ready, type:%s, last:%" PRId64 ", cmt:%" PRId64,
             pSyncNode->vgId, TMSG_INFO(pMsg->msgType), syncNodeGetLastIndex(pSyncNode), pSyncNode->commitIndex);
1000 1001 1002
      goto _END;
    }

M
Minghao Li 已提交
1003 1004 1005 1006 1007 1008
    SRespStub stub;
    stub.createTime = taosGetTimestampMs();
    stub.rpcMsg = *pMsg;
    uint64_t seqNum = syncRespMgrAdd(pSyncNode->pSyncRespMgr, &stub);

    SyncClientRequest* pSyncMsg = syncClientRequestBuild2(pMsg, seqNum, isWeak, pSyncNode->vgId);
M
Minghao Li 已提交
1009 1010
    SRpcMsg            rpcMsg;
    syncClientRequest2RpcMsg(pSyncMsg, &rpcMsg);
1011

1012 1013 1014
    // optimized one replica
    if (syncNodeIsOptimizedOneReplica(pSyncNode, pMsg)) {
      SyncIndex retIndex;
M
Minghao Li 已提交
1015
      int32_t   code = syncNodeOnClientRequest(pSyncNode, pSyncMsg, &retIndex);
1016 1017
      if (code == 0) {
        pMsg->info.conn.applyIndex = retIndex;
M
Minghao Li 已提交
1018
        pMsg->info.conn.applyTerm = pSyncNode->pRaftStore->currentTerm;
1019 1020 1021
        rpcFreeCont(rpcMsg.pCont);
        syncRespMgrDel(pSyncNode->pSyncRespMgr, seqNum);
        ret = 1;
1022 1023
        sDebug("vgId:%d, sync optimize index:%" PRId64 ", type:%s", pSyncNode->vgId, retIndex,
               TMSG_INFO(pMsg->msgType));
1024 1025 1026
      } else {
        ret = -1;
        terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
S
Shengliang Guan 已提交
1027 1028
        sError("vgId:%d, failed to sync optimize index:%" PRId64 ", type:%s", pSyncNode->vgId, retIndex,
               TMSG_INFO(pMsg->msgType));
1029 1030
      }

M
Minghao Li 已提交
1031
    } else {
1032 1033 1034 1035 1036
      if (pSyncNode->FpEqMsg != NULL && (*pSyncNode->FpEqMsg)(pSyncNode->msgcb, &rpcMsg) == 0) {
        ret = 0;
      } else {
        ret = -1;
        terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
S
Shengliang Guan 已提交
1037
        sError("vgId:%d, failed to enqueue msg since its null", pSyncNode->vgId);
1038
      }
M
Minghao Li 已提交
1039
    }
1040

M
Minghao Li 已提交
1041
    syncClientRequestDestroy(pSyncMsg);
M
Minghao Li 已提交
1042 1043
    goto _END;

M
Minghao Li 已提交
1044
  } else {
M
Minghao Li 已提交
1045 1046
    ret = -1;
    terrno = TSDB_CODE_SYN_NOT_LEADER;
S
Shengliang Guan 已提交
1047 1048
    sError("vgId:%d, sync propose not leader, %s, type:%s", pSyncNode->vgId, syncUtilState2String(pSyncNode->state),
           TMSG_INFO(pMsg->msgType));
M
Minghao Li 已提交
1049
    goto _END;
M
Minghao Li 已提交
1050
  }
M
Minghao Li 已提交
1051

M
Minghao Li 已提交
1052
_END:
M
Minghao Li 已提交
1053 1054 1055
  return ret;
}

1056 1057 1058 1059 1060 1061 1062 1063 1064 1065 1066 1067 1068
int32_t syncHbTimerInit(SSyncNode* pSyncNode, SSyncTimer* pSyncTimer, SRaftId destId) {
  pSyncTimer->pTimer = NULL;
  pSyncTimer->counter = 0;
  pSyncTimer->timerMS = pSyncNode->hbBaseLine;
  pSyncTimer->timerCb = syncNodeEqPeerHeartbeatTimer;
  pSyncTimer->destId = destId;
  atomic_store_64(&pSyncTimer->logicClock, 0);
  return 0;
}

int32_t syncHbTimerStart(SSyncNode* pSyncNode, SSyncTimer* pSyncTimer) {
  int32_t ret = 0;
  if (syncEnvIsStart()) {
M
Minghao Li 已提交
1069
    SSyncHbTimerData* pData = taosMemoryMalloc(sizeof(SSyncHbTimerData));
1070 1071 1072 1073
    pData->pSyncNode = pSyncNode;
    pData->pTimer = pSyncTimer;
    pData->destId = pSyncTimer->destId;
    pData->logicClock = pSyncTimer->logicClock;
M
Minghao Li 已提交
1074

1075 1076 1077 1078 1079 1080 1081 1082 1083 1084 1085 1086 1087
    pSyncTimer->pData = pData;
    taosTmrReset(pSyncTimer->timerCb, pSyncTimer->timerMS, pData, gSyncEnv->pTimerManager, &pSyncTimer->pTimer);
  } else {
    sError("vgId:%d, start ctrl hb timer error, sync env is stop", pSyncNode->vgId);
  }
  return ret;
}

int32_t syncHbTimerStop(SSyncNode* pSyncNode, SSyncTimer* pSyncTimer) {
  int32_t ret = 0;
  atomic_add_fetch_64(&pSyncTimer->logicClock, 1);
  taosTmrStop(pSyncTimer->pTimer);
  pSyncTimer->pTimer = NULL;
M
Minghao Li 已提交
1088
  // taosMemoryFree(pSyncTimer->pData);
1089 1090 1091
  return ret;
}

M
Minghao Li 已提交
1092
// open/close --------------
M
Minghao Li 已提交
1093
SSyncNode* syncNodeOpen(SSyncInfo* pOldSyncInfo) {
1094 1095
  SSyncInfo* pSyncInfo = (SSyncInfo*)pOldSyncInfo;

1096 1097 1098 1099 1100
  SSyncNode* pSyncNode = (SSyncNode*)taosMemoryCalloc(1, sizeof(SSyncNode));
  if (pSyncNode == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    goto _error;
  }
M
Minghao Li 已提交
1101

M
Minghao Li 已提交
1102 1103 1104 1105 1106
  int32_t ret = 0;
  if (!taosDirExist((char*)(pSyncInfo->path))) {
    if (taosMkDir(pSyncInfo->path) != 0) {
      terrno = TAOS_SYSTEM_ERROR(errno);
      sError("failed to create dir:%s since %s", pSyncInfo->path, terrstr());
1107
      goto _error;
M
Minghao Li 已提交
1108
    }
1109
  }
M
Minghao Li 已提交
1110

S
Shengliang Guan 已提交
1111
  snprintf(pSyncNode->configPath, sizeof(pSyncNode->configPath), "%s%sraft_config.json", pSyncInfo->path, TD_DIRSEP);
1112
  if (!taosCheckExistFile(pSyncNode->configPath)) {
M
Minghao Li 已提交
1113 1114 1115
    // create a new raft config file
    SRaftCfgMeta meta;
    meta.isStandBy = pSyncInfo->isStandBy;
M
Minghao Li 已提交
1116
    meta.snapshotStrategy = pSyncInfo->snapshotStrategy;
1117
    meta.lastConfigIndex = SYNC_INDEX_INVALID;
M
Minghao Li 已提交
1118
    meta.batchSize = pSyncInfo->batchSize;
M
Minghao Li 已提交
1119
    ret = raftCfgCreateFile((SSyncCfg*)&(pSyncInfo->syncCfg), meta, pSyncNode->configPath);
1120
    if (ret != 0) {
H
Hongze Cheng 已提交
1121 1122
      sError("failed to create raft cfg file. configPath: %s", pSyncNode->configPath);
      goto _error;
1123
    }
1124 1125 1126
    if (pSyncInfo->syncCfg.replicaNum == 0) {
      pSyncInfo->syncCfg = pSyncNode->pRaftCfg->cfg;
    }
1127 1128 1129
  } else {
    // update syncCfg by raft_config.json
    pSyncNode->pRaftCfg = raftCfgOpen(pSyncNode->configPath);
1130
    if (pSyncNode->pRaftCfg == NULL) {
H
Hongze Cheng 已提交
1131 1132
      sError("failed to open raft cfg file. path:%s", pSyncNode->configPath);
      goto _error;
1133
    }
1134 1135 1136
    pSyncInfo->syncCfg = pSyncNode->pRaftCfg->cfg;

    raftCfgClose(pSyncNode->pRaftCfg);
1137
    pSyncNode->pRaftCfg = NULL;
M
Minghao Li 已提交
1138 1139
  }

M
Minghao Li 已提交
1140
  // init by SSyncInfo
M
Minghao Li 已提交
1141 1142
  pSyncNode->vgId = pSyncInfo->vgId;
  memcpy(pSyncNode->path, pSyncInfo->path, sizeof(pSyncNode->path));
S
Shengliang Guan 已提交
1143 1144 1145
  snprintf(pSyncNode->raftStorePath, sizeof(pSyncNode->raftStorePath), "%s%sraft_store.json", pSyncInfo->path,
           TD_DIRSEP);
  snprintf(pSyncNode->configPath, sizeof(pSyncNode->configPath), "%s%sraft_config.json", pSyncInfo->path, TD_DIRSEP);
M
Minghao Li 已提交
1146

M
Minghao Li 已提交
1147
  pSyncNode->pWal = pSyncInfo->pWal;
S
Shengliang Guan 已提交
1148
  pSyncNode->msgcb = pSyncInfo->msgcb;
M
Minghao Li 已提交
1149
  pSyncNode->FpSendMsg = pSyncInfo->FpSendMsg;
M
Minghao Li 已提交
1150
  pSyncNode->FpEqMsg = pSyncInfo->FpEqMsg;
1151
  pSyncNode->FpEqCtrlMsg = pSyncInfo->FpEqCtrlMsg;
M
Minghao Li 已提交
1152

M
Minghao Li 已提交
1153 1154
  // init raft config
  pSyncNode->pRaftCfg = raftCfgOpen(pSyncNode->configPath);
1155 1156 1157 1158
  if (pSyncNode->pRaftCfg == NULL) {
    sError("failed to open raft cfg file. path:%s", pSyncNode->configPath);
    goto _error;
  }
M
Minghao Li 已提交
1159

M
Minghao Li 已提交
1160
  // init internal
M
Minghao Li 已提交
1161
  pSyncNode->myNodeInfo = pSyncNode->pRaftCfg->cfg.nodeInfo[pSyncNode->pRaftCfg->cfg.myIndex];
1162
  if (!syncUtilnodeInfo2raftId(&pSyncNode->myNodeInfo, pSyncNode->vgId, &pSyncNode->myRaftId)) {
H
Hongze Cheng 已提交
1163 1164
    sError("failed to determine my raft member id. vgId:%d", pSyncNode->vgId);
    goto _error;
1165
  }
M
Minghao Li 已提交
1166

M
Minghao Li 已提交
1167
  // init peersNum, peers, peersId
M
Minghao Li 已提交
1168
  pSyncNode->peersNum = pSyncNode->pRaftCfg->cfg.replicaNum - 1;
M
Minghao Li 已提交
1169
  int j = 0;
M
Minghao Li 已提交
1170 1171 1172
  for (int i = 0; i < pSyncNode->pRaftCfg->cfg.replicaNum; ++i) {
    if (i != pSyncNode->pRaftCfg->cfg.myIndex) {
      pSyncNode->peersNodeInfo[j] = pSyncNode->pRaftCfg->cfg.nodeInfo[i];
M
Minghao Li 已提交
1173 1174 1175
      j++;
    }
  }
M
Minghao Li 已提交
1176
  for (int i = 0; i < pSyncNode->peersNum; ++i) {
1177
    if (!syncUtilnodeInfo2raftId(&pSyncNode->peersNodeInfo[i], pSyncNode->vgId, &pSyncNode->peersId[i])) {
H
Hongze Cheng 已提交
1178 1179
      sError("failed to determine raft member id. vgId:%d, peer:%d", pSyncNode->vgId, i);
      goto _error;
1180
    }
M
Minghao Li 已提交
1181
  }
M
Minghao Li 已提交
1182

M
Minghao Li 已提交
1183
  // init replicaNum, replicasId
M
Minghao Li 已提交
1184 1185
  pSyncNode->replicaNum = pSyncNode->pRaftCfg->cfg.replicaNum;
  for (int i = 0; i < pSyncNode->pRaftCfg->cfg.replicaNum; ++i) {
H
Hongze Cheng 已提交
1186 1187 1188
    if (!syncUtilnodeInfo2raftId(&pSyncNode->pRaftCfg->cfg.nodeInfo[i], pSyncNode->vgId, &pSyncNode->replicasId[i])) {
      sError("failed to determine raft member id. vgId:%d, replica:%d", pSyncNode->vgId, i);
      goto _error;
1189
    }
M
Minghao Li 已提交
1190 1191
  }

M
Minghao Li 已提交
1192
  // init raft algorithm
M
Minghao Li 已提交
1193
  pSyncNode->pFsm = pSyncInfo->pFsm;
1194
  pSyncInfo->pFsm = NULL;
M
Minghao Li 已提交
1195
  pSyncNode->quorum = syncUtilQuorum(pSyncNode->pRaftCfg->cfg.replicaNum);
M
Minghao Li 已提交
1196 1197
  pSyncNode->leaderCache = EMPTY_RAFT_ID;

M
Minghao Li 已提交
1198
  // init life cycle outside
M
Minghao Li 已提交
1199

M
Minghao Li 已提交
1200 1201 1202 1203 1204 1205 1206 1207 1208 1209 1210 1211 1212 1213 1214 1215 1216 1217 1218 1219 1220 1221 1222 1223
  // TLA+ Spec
  // InitHistoryVars == /\ elections = {}
  //                    /\ allLogs   = {}
  //                    /\ voterLog  = [i \in Server |-> [j \in {} |-> <<>>]]
  // InitServerVars == /\ currentTerm = [i \in Server |-> 1]
  //                   /\ state       = [i \in Server |-> Follower]
  //                   /\ votedFor    = [i \in Server |-> Nil]
  // InitCandidateVars == /\ votesResponded = [i \in Server |-> {}]
  //                      /\ votesGranted   = [i \in Server |-> {}]
  // \* The values nextIndex[i][i] and matchIndex[i][i] are never read, since the
  // \* leader does not send itself messages. It's still easier to include these
  // \* in the functions.
  // InitLeaderVars == /\ nextIndex  = [i \in Server |-> [j \in Server |-> 1]]
  //                   /\ matchIndex = [i \in Server |-> [j \in Server |-> 0]]
  // InitLogVars == /\ log          = [i \in Server |-> << >>]
  //                /\ commitIndex  = [i \in Server |-> 0]
  // Init == /\ messages = [m \in {} |-> 0]
  //         /\ InitHistoryVars
  //         /\ InitServerVars
  //         /\ InitCandidateVars
  //         /\ InitLeaderVars
  //         /\ InitLogVars
  //

M
Minghao Li 已提交
1224
  // init TLA+ server vars
M
syncInt  
Minghao Li 已提交
1225
  pSyncNode->state = TAOS_SYNC_STATE_FOLLOWER;
M
Minghao Li 已提交
1226
  pSyncNode->pRaftStore = raftStoreOpen(pSyncNode->raftStorePath);
1227 1228 1229 1230
  if (pSyncNode->pRaftStore == NULL) {
    sError("failed to open raft store. path: %s", pSyncNode->raftStorePath);
    goto _error;
  }
M
Minghao Li 已提交
1231

M
Minghao Li 已提交
1232
  // init TLA+ candidate vars
M
Minghao Li 已提交
1233
  pSyncNode->pVotesGranted = voteGrantedCreate(pSyncNode);
1234 1235 1236 1237
  if (pSyncNode->pVotesGranted == NULL) {
    sError("failed to create VotesGranted. vgId:%d", pSyncNode->vgId);
    goto _error;
  }
M
Minghao Li 已提交
1238
  pSyncNode->pVotesRespond = votesRespondCreate(pSyncNode);
1239 1240 1241 1242
  if (pSyncNode->pVotesRespond == NULL) {
    sError("failed to create VotesRespond. vgId:%d", pSyncNode->vgId);
    goto _error;
  }
M
Minghao Li 已提交
1243

M
Minghao Li 已提交
1244 1245
  // init TLA+ leader vars
  pSyncNode->pNextIndex = syncIndexMgrCreate(pSyncNode);
1246 1247 1248 1249
  if (pSyncNode->pNextIndex == NULL) {
    sError("failed to create SyncIndexMgr. vgId:%d", pSyncNode->vgId);
    goto _error;
  }
M
Minghao Li 已提交
1250
  pSyncNode->pMatchIndex = syncIndexMgrCreate(pSyncNode);
1251 1252 1253 1254
  if (pSyncNode->pMatchIndex == NULL) {
    sError("failed to create SyncIndexMgr. vgId:%d", pSyncNode->vgId);
    goto _error;
  }
M
Minghao Li 已提交
1255 1256 1257

  // init TLA+ log vars
  pSyncNode->pLogStore = logStoreCreate(pSyncNode);
1258 1259 1260 1261
  if (pSyncNode->pLogStore == NULL) {
    sError("failed to create SyncLogStore. vgId:%d", pSyncNode->vgId);
    goto _error;
  }
1262 1263 1264 1265 1266

  SyncIndex commitIndex = SYNC_INDEX_INVALID;
  if (pSyncNode->pFsm != NULL && pSyncNode->pFsm->FpGetSnapshotInfo != NULL) {
    SSnapshot snapshot = {0};
    int32_t   code = pSyncNode->pFsm->FpGetSnapshotInfo(pSyncNode->pFsm, &snapshot);
1267
    if (code != 0) {
H
Hongze Cheng 已提交
1268 1269
      sError("failed to get snapshot info. vgId:%d, code:%d", pSyncNode->vgId, code);
      goto _error;
1270
    }
1271 1272 1273 1274 1275 1276
    if (snapshot.lastApplyIndex > commitIndex) {
      commitIndex = snapshot.lastApplyIndex;
      syncNodeEventLog(pSyncNode, "reset commit index by snapshot");
    }
  }
  pSyncNode->commitIndex = commitIndex;
M
Minghao Li 已提交
1277

M
Minghao Li 已提交
1278 1279 1280 1281 1282
  // timer ms init
  pSyncNode->pingBaseLine = PING_TIMER_MS;
  pSyncNode->electBaseLine = ELECT_TIMER_MS_MIN;
  pSyncNode->hbBaseLine = HEARTBEAT_TIMER_MS;

M
Minghao Li 已提交
1283
  // init ping timer
M
Minghao Li 已提交
1284
  pSyncNode->pPingTimer = NULL;
M
Minghao Li 已提交
1285
  pSyncNode->pingTimerMS = pSyncNode->pingBaseLine;
M
Minghao Li 已提交
1286 1287
  atomic_store_64(&pSyncNode->pingTimerLogicClock, 0);
  atomic_store_64(&pSyncNode->pingTimerLogicClockUser, 0);
M
Minghao Li 已提交
1288
  pSyncNode->FpPingTimerCB = syncNodeEqPingTimer;
M
Minghao Li 已提交
1289
  pSyncNode->pingTimerCounter = 0;
M
Minghao Li 已提交
1290

M
Minghao Li 已提交
1291 1292
  // init elect timer
  pSyncNode->pElectTimer = NULL;
M
Minghao Li 已提交
1293
  pSyncNode->electTimerMS = syncUtilElectRandomMS(pSyncNode->electBaseLine, 2 * pSyncNode->electBaseLine);
M
Minghao Li 已提交
1294 1295
  atomic_store_64(&pSyncNode->electTimerLogicClock, 0);
  atomic_store_64(&pSyncNode->electTimerLogicClockUser, 0);
M
Minghao Li 已提交
1296
  pSyncNode->FpElectTimerCB = syncNodeEqElectTimer;
M
Minghao Li 已提交
1297 1298 1299 1300
  pSyncNode->electTimerCounter = 0;

  // init heartbeat timer
  pSyncNode->pHeartbeatTimer = NULL;
M
Minghao Li 已提交
1301
  pSyncNode->heartbeatTimerMS = pSyncNode->hbBaseLine;
M
Minghao Li 已提交
1302 1303
  atomic_store_64(&pSyncNode->heartbeatTimerLogicClock, 0);
  atomic_store_64(&pSyncNode->heartbeatTimerLogicClockUser, 0);
M
Minghao Li 已提交
1304
  pSyncNode->FpHeartbeatTimerCB = syncNodeEqHeartbeatTimer;
M
Minghao Li 已提交
1305 1306
  pSyncNode->heartbeatTimerCounter = 0;

1307 1308 1309 1310 1311
  // init peer heartbeat timer
  for (int32_t i = 0; i < TSDB_MAX_REPLICA; ++i) {
    syncHbTimerInit(pSyncNode, &(pSyncNode->peerHeartbeatTimerArr[i]), (pSyncNode->replicasId)[i]);
  }

M
Minghao Li 已提交
1312
  // init callback
M
Minghao Li 已提交
1313 1314
  pSyncNode->FpOnPing = syncNodeOnPingCb;
  pSyncNode->FpOnPingReply = syncNodeOnPingReplyCb;
M
Minghao Li 已提交
1315
  pSyncNode->FpOnClientRequest = syncNodeOnClientRequest;
1316
  pSyncNode->FpOnTimeout = syncNodeOnTimer;
M
Minghao Li 已提交
1317 1318 1319 1320 1321 1322
  pSyncNode->FpOnSnapshot = syncNodeOnSnapshot;
  pSyncNode->FpOnSnapshotReply = syncNodeOnSnapshotReply;
  pSyncNode->FpOnRequestVote = syncNodeOnRequestVote;
  pSyncNode->FpOnRequestVoteReply = syncNodeOnRequestVoteReply;
  pSyncNode->FpOnAppendEntries = syncNodeOnAppendEntries;
  pSyncNode->FpOnAppendEntriesReply = syncNodeOnAppendEntriesReply;
M
Minghao Li 已提交
1323

M
Minghao Li 已提交
1324
  // tools
M
Minghao Li 已提交
1325
  pSyncNode->pSyncRespMgr = syncRespMgrCreate(pSyncNode, SYNC_RESP_TTL_MS);
1326 1327 1328 1329
  if (pSyncNode->pSyncRespMgr == NULL) {
    sError("failed to create SyncRespMgr. vgId:%d", pSyncNode->vgId);
    goto _error;
  }
M
Minghao Li 已提交
1330

1331 1332
  // restore state
  pSyncNode->restoreFinish = false;
1333

M
Minghao Li 已提交
1334 1335 1336 1337 1338 1339 1340 1341
  // snapshot senders
  for (int i = 0; i < TSDB_MAX_REPLICA; ++i) {
    SSyncSnapshotSender* pSender = snapshotSenderCreate(pSyncNode, i);
    // ASSERT(pSender != NULL);
    (pSyncNode->senders)[i] = pSender;
  }

  // snapshot receivers
1342
  pSyncNode->pNewNodeReceiver = snapshotReceiverCreate(pSyncNode, EMPTY_RAFT_ID);
M
Minghao Li 已提交
1343

M
Minghao Li 已提交
1344 1345 1346
  // is config changing
  pSyncNode->changing = false;

M
Minghao Li 已提交
1347 1348 1349
  // peer state
  syncNodePeerStateInit(pSyncNode);

M
Minghao Li 已提交
1350 1351 1352
  // min match index
  pSyncNode->minMatchIndex = SYNC_INDEX_INVALID;

M
Minghao Li 已提交
1353
  // start in syncNodeStart
M
Minghao Li 已提交
1354
  // start raft
M
Minghao Li 已提交
1355
  // syncNodeBecomeFollower(pSyncNode);
M
Minghao Li 已提交
1356

M
Minghao Li 已提交
1357 1358
  int64_t timeNow = taosGetTimestampMs();
  pSyncNode->startTime = timeNow;
1359
  pSyncNode->leaderTime = timeNow;
M
Minghao Li 已提交
1360 1361
  pSyncNode->lastReplicateTime = timeNow;

1362 1363 1364
  // snapshotting
  atomic_store_64(&pSyncNode->snapshottingIndex, SYNC_INDEX_INVALID);

M
Minghao Li 已提交
1365
  syncNodeEventLog(pSyncNode, "sync open");
1366

M
Minghao Li 已提交
1367
  return pSyncNode;
1368 1369 1370

_error:
  if (pSyncInfo->pFsm) {
H
Hongze Cheng 已提交
1371 1372
    taosMemoryFree(pSyncInfo->pFsm);
    pSyncInfo->pFsm = NULL;
1373 1374 1375 1376
  }
  syncNodeClose(pSyncNode);
  pSyncNode = NULL;
  return NULL;
M
Minghao Li 已提交
1377 1378
}

M
Minghao Li 已提交
1379 1380 1381 1382 1383 1384 1385 1386 1387 1388 1389
void syncNodeMaybeUpdateCommitBySnapshot(SSyncNode* pSyncNode) {
  if (pSyncNode->pFsm != NULL && pSyncNode->pFsm->FpGetSnapshotInfo != NULL) {
    SSnapshot snapshot;
    int32_t   code = pSyncNode->pFsm->FpGetSnapshotInfo(pSyncNode->pFsm, &snapshot);
    ASSERT(code == 0);
    if (snapshot.lastApplyIndex > pSyncNode->commitIndex) {
      pSyncNode->commitIndex = snapshot.lastApplyIndex;
    }
  }
}

M
Minghao Li 已提交
1390 1391
void syncNodeStart(SSyncNode* pSyncNode) {
  // start raft
1392
  if (pSyncNode->replicaNum == 1) {
M
Minghao Li 已提交
1393
    raftStoreNextTerm(pSyncNode->pRaftStore);
1394
    syncNodeBecomeLeader(pSyncNode, "one replica start");
M
format  
Minghao Li 已提交
1395

1396
    // Raft 3.6.2 Committing entries from previous terms
1397 1398
    syncNodeAppendNoop(pSyncNode);
    syncMaybeAdvanceCommitIndex(pSyncNode);
M
Minghao Li 已提交
1399

M
Minghao Li 已提交
1400 1401
  } else {
    syncNodeBecomeFollower(pSyncNode, "first start");
1402 1403
  }

1404 1405 1406
  int32_t ret = 0;
  ret = syncNodeStartPingTimer(pSyncNode);
  ASSERT(ret == 0);
M
Minghao Li 已提交
1407 1408
}

M
Minghao Li 已提交
1409 1410 1411 1412 1413 1414 1415 1416 1417
void syncNodeStartStandBy(SSyncNode* pSyncNode) {
  // state change
  pSyncNode->state = TAOS_SYNC_STATE_FOLLOWER;
  syncNodeStopHeartbeatTimer(pSyncNode);

  // reset elect timer, long enough
  int32_t electMS = TIMER_MAX_MS;
  int32_t ret = syncNodeRestartElectTimer(pSyncNode, electMS);
  ASSERT(ret == 0);
1418

1419 1420 1421
  ret = 0;
  ret = syncNodeStartPingTimer(pSyncNode);
  ASSERT(ret == 0);
M
Minghao Li 已提交
1422 1423
}

M
Minghao Li 已提交
1424
void syncNodeClose(SSyncNode* pSyncNode) {
M
Minghao Li 已提交
1425
  syncNodeEventLog(pSyncNode, "sync close");
1426 1427 1428
  if (pSyncNode == NULL) {
    return;
  }
M
Minghao Li 已提交
1429 1430 1431
  int32_t ret;

  ret = raftStoreClose(pSyncNode->pRaftStore);
M
Minghao Li 已提交
1432
  ASSERT(ret == 0);
M
Minghao Li 已提交
1433

M
Minghao Li 已提交
1434
  syncRespMgrDestroy(pSyncNode->pSyncRespMgr);
1435
  pSyncNode->pSyncRespMgr = NULL;
M
Minghao Li 已提交
1436
  voteGrantedDestroy(pSyncNode->pVotesGranted);
1437
  pSyncNode->pVotesGranted = NULL;
M
Minghao Li 已提交
1438
  votesRespondDestory(pSyncNode->pVotesRespond);
1439
  pSyncNode->pVotesRespond = NULL;
M
Minghao Li 已提交
1440
  syncIndexMgrDestroy(pSyncNode->pNextIndex);
1441
  pSyncNode->pNextIndex = NULL;
M
Minghao Li 已提交
1442
  syncIndexMgrDestroy(pSyncNode->pMatchIndex);
1443
  pSyncNode->pMatchIndex = NULL;
M
Minghao Li 已提交
1444
  logStoreDestory(pSyncNode->pLogStore);
1445
  pSyncNode->pLogStore = NULL;
M
Minghao Li 已提交
1446
  raftCfgClose(pSyncNode->pRaftCfg);
1447
  pSyncNode->pRaftCfg = NULL;
M
Minghao Li 已提交
1448 1449 1450 1451 1452

  syncNodeStopPingTimer(pSyncNode);
  syncNodeStopElectTimer(pSyncNode);
  syncNodeStopHeartbeatTimer(pSyncNode);

M
Minghao Li 已提交
1453 1454 1455 1456
  if (pSyncNode->pFsm != NULL) {
    taosMemoryFree(pSyncNode->pFsm);
  }

M
Minghao Li 已提交
1457 1458 1459 1460 1461 1462 1463
  for (int i = 0; i < TSDB_MAX_REPLICA; ++i) {
    if ((pSyncNode->senders)[i] != NULL) {
      snapshotSenderDestroy((pSyncNode->senders)[i]);
      (pSyncNode->senders)[i] = NULL;
    }
  }

M
Minghao Li 已提交
1464 1465 1466 1467 1468
  if (pSyncNode->pNewNodeReceiver != NULL) {
    snapshotReceiverDestroy(pSyncNode->pNewNodeReceiver);
    pSyncNode->pNewNodeReceiver = NULL;
  }

1469
  taosMemoryFree(pSyncNode);
M
Minghao Li 已提交
1470 1471
}

M
Minghao Li 已提交
1472
// option
M
Minghao Li 已提交
1473 1474
// bool syncNodeSnapshotEnable(SSyncNode* pSyncNode) { return pSyncNode->pRaftCfg->snapshotEnable; }

M
Minghao Li 已提交
1475
ESyncStrategy syncNodeStrategy(SSyncNode* pSyncNode) { return pSyncNode->pRaftCfg->snapshotStrategy; }
M
Minghao Li 已提交
1476

M
Minghao Li 已提交
1477 1478 1479 1480 1481 1482 1483 1484 1485 1486 1487 1488 1489 1490 1491
// ping --------------
int32_t syncNodePing(SSyncNode* pSyncNode, const SRaftId* destRaftId, SyncPing* pMsg) {
  syncPingLog2((char*)"==syncNodePing==", pMsg);
  int32_t ret = 0;

  SRpcMsg rpcMsg;
  syncPing2RpcMsg(pMsg, &rpcMsg);
  syncRpcMsgLog2((char*)"==syncNodePing==", &rpcMsg);

  ret = syncNodeSendMsgById(destRaftId, pSyncNode, &rpcMsg);
  return ret;
}

int32_t syncNodePingSelf(SSyncNode* pSyncNode) {
  int32_t   ret = 0;
M
Minghao Li 已提交
1492
  SyncPing* pMsg = syncPingBuild3(&pSyncNode->myRaftId, &pSyncNode->myRaftId, pSyncNode->vgId);
M
Minghao Li 已提交
1493
  ret = syncNodePing(pSyncNode, &pMsg->destId, pMsg);
M
Minghao Li 已提交
1494
  ASSERT(ret == 0);
M
Minghao Li 已提交
1495 1496 1497 1498 1499 1500 1501 1502

  syncPingDestroy(pMsg);
  return ret;
}

int32_t syncNodePingPeers(SSyncNode* pSyncNode) {
  int32_t ret = 0;
  for (int i = 0; i < pSyncNode->peersNum; ++i) {
M
Minghao Li 已提交
1503 1504 1505
    SRaftId*  destId = &(pSyncNode->peersId[i]);
    SyncPing* pMsg = syncPingBuild3(&pSyncNode->myRaftId, destId, pSyncNode->vgId);
    ret = syncNodePing(pSyncNode, destId, pMsg);
M
Minghao Li 已提交
1506
    ASSERT(ret == 0);
M
Minghao Li 已提交
1507 1508 1509 1510 1511 1512 1513
    syncPingDestroy(pMsg);
  }
  return ret;
}

int32_t syncNodePingAll(SSyncNode* pSyncNode) {
  int32_t ret = 0;
M
Minghao Li 已提交
1514 1515 1516 1517
  for (int i = 0; i < pSyncNode->pRaftCfg->cfg.replicaNum; ++i) {
    SRaftId*  destId = &(pSyncNode->replicasId[i]);
    SyncPing* pMsg = syncPingBuild3(&pSyncNode->myRaftId, destId, pSyncNode->vgId);
    ret = syncNodePing(pSyncNode, destId, pMsg);
M
Minghao Li 已提交
1518
    ASSERT(ret == 0);
M
Minghao Li 已提交
1519 1520 1521 1522 1523 1524 1525 1526
    syncPingDestroy(pMsg);
  }
  return ret;
}

// timer control --------------
int32_t syncNodeStartPingTimer(SSyncNode* pSyncNode) {
  int32_t ret = 0;
1527 1528 1529 1530 1531
  if (syncEnvIsStart()) {
    taosTmrReset(pSyncNode->FpPingTimerCB, pSyncNode->pingTimerMS, pSyncNode, gSyncEnv->pTimerManager,
                 &pSyncNode->pPingTimer);
    atomic_store_64(&pSyncNode->pingTimerLogicClock, pSyncNode->pingTimerLogicClockUser);
  } else {
M
Minghao Li 已提交
1532
    sError("vgId:%d, start ping timer error, sync env is stop", pSyncNode->vgId);
1533
  }
M
Minghao Li 已提交
1534 1535 1536 1537 1538 1539 1540 1541 1542 1543 1544 1545 1546
  return ret;
}

int32_t syncNodeStopPingTimer(SSyncNode* pSyncNode) {
  int32_t ret = 0;
  atomic_add_fetch_64(&pSyncNode->pingTimerLogicClockUser, 1);
  taosTmrStop(pSyncNode->pPingTimer);
  pSyncNode->pPingTimer = NULL;
  return ret;
}

int32_t syncNodeStartElectTimer(SSyncNode* pSyncNode, int32_t ms) {
  int32_t ret = 0;
1547 1548 1549 1550 1551
  if (syncEnvIsStart()) {
    pSyncNode->electTimerMS = ms;
    taosTmrReset(pSyncNode->FpElectTimerCB, pSyncNode->electTimerMS, pSyncNode, gSyncEnv->pTimerManager,
                 &pSyncNode->pElectTimer);
    atomic_store_64(&pSyncNode->electTimerLogicClock, pSyncNode->electTimerLogicClockUser);
1552

1553 1554 1555 1556 1557 1558 1559
    /*
        do {
          char logBuf[128];
          snprintf(logBuf, sizeof(logBuf), "elect timer reset, ms:%d", ms);
          syncNodeEventLog(pSyncNode, logBuf);
        } while (0);
    */
1560

1561
  } else {
M
Minghao Li 已提交
1562
    sError("vgId:%d, start elect timer error, sync env is stop", pSyncNode->vgId);
1563
  }
M
Minghao Li 已提交
1564 1565 1566 1567 1568 1569 1570 1571
  return ret;
}

int32_t syncNodeStopElectTimer(SSyncNode* pSyncNode) {
  int32_t ret = 0;
  atomic_add_fetch_64(&pSyncNode->electTimerLogicClockUser, 1);
  taosTmrStop(pSyncNode->pElectTimer);
  pSyncNode->pElectTimer = NULL;
1572

1573
  // sTrace("vgId:%d, sync %s stop elect timer", pSyncNode->vgId, syncUtilState2String(pSyncNode->state));
M
Minghao Li 已提交
1574 1575 1576 1577 1578 1579 1580 1581 1582 1583
  return ret;
}

int32_t syncNodeRestartElectTimer(SSyncNode* pSyncNode, int32_t ms) {
  int32_t ret = 0;
  syncNodeStopElectTimer(pSyncNode);
  syncNodeStartElectTimer(pSyncNode, ms);
  return ret;
}

M
Minghao Li 已提交
1584 1585
int32_t syncNodeResetElectTimer(SSyncNode* pSyncNode) {
  int32_t ret = 0;
M
Minghao Li 已提交
1586 1587 1588 1589 1590 1591 1592
  int32_t electMS;

  if (pSyncNode->pRaftCfg->isStandBy) {
    electMS = TIMER_MAX_MS;
  } else {
    electMS = syncUtilElectRandomMS(pSyncNode->electBaseLine, 2 * pSyncNode->electBaseLine);
  }
M
Minghao Li 已提交
1593
  ret = syncNodeRestartElectTimer(pSyncNode, electMS);
1594 1595 1596 1597 1598 1599 1600 1601

  do {
    char logBuf[128];
    snprintf(logBuf, sizeof(logBuf), "reset elect timer, min:%d, max:%d, ms:%d", pSyncNode->electBaseLine,
             2 * pSyncNode->electBaseLine, electMS);
    syncNodeEventLog(pSyncNode, logBuf);
  } while (0);

M
Minghao Li 已提交
1602 1603 1604
  return ret;
}

M
Minghao Li 已提交
1605
static int32_t syncNodeDoStartHeartbeatTimer(SSyncNode* pSyncNode) {
M
Minghao Li 已提交
1606
  int32_t ret = 0;
1607 1608 1609 1610 1611
  if (syncEnvIsStart()) {
    taosTmrReset(pSyncNode->FpHeartbeatTimerCB, pSyncNode->heartbeatTimerMS, pSyncNode, gSyncEnv->pTimerManager,
                 &pSyncNode->pHeartbeatTimer);
    atomic_store_64(&pSyncNode->heartbeatTimerLogicClock, pSyncNode->heartbeatTimerLogicClockUser);
  } else {
M
Minghao Li 已提交
1612
    sError("vgId:%d, start heartbeat timer error, sync env is stop", pSyncNode->vgId);
1613
  }
1614 1615 1616 1617 1618 1619 1620

  do {
    char logBuf[128];
    snprintf(logBuf, sizeof(logBuf), "start heartbeat timer, ms:%d", pSyncNode->heartbeatTimerMS);
    syncNodeEventLog(pSyncNode, logBuf);
  } while (0);

M
Minghao Li 已提交
1621 1622 1623
  return ret;
}

M
Minghao Li 已提交
1624
int32_t syncNodeStartHeartbeatTimer(SSyncNode* pSyncNode) {
1625
  int32_t ret = 0;
M
Minghao Li 已提交
1626

1627
#if 0
M
Minghao Li 已提交
1628
  pSyncNode->heartbeatTimerMS = pSyncNode->hbBaseLine;
1629 1630
  ret = syncNodeDoStartHeartbeatTimer(pSyncNode);
#endif
1631

1632 1633 1634 1635
  for (int i = 0; i < pSyncNode->peersNum; ++i) {
    SSyncTimer* pSyncTimer = syncNodeGetHbTimer(pSyncNode, &(pSyncNode->peersId[i]));
    syncHbTimerStart(pSyncNode, pSyncTimer);
  }
1636

M
Minghao Li 已提交
1637 1638 1639
  return ret;
}

M
Minghao Li 已提交
1640 1641
int32_t syncNodeStopHeartbeatTimer(SSyncNode* pSyncNode) {
  int32_t ret = 0;
1642 1643

#if 0
M
Minghao Li 已提交
1644 1645 1646
  atomic_add_fetch_64(&pSyncNode->heartbeatTimerLogicClockUser, 1);
  taosTmrStop(pSyncNode->pHeartbeatTimer);
  pSyncNode->pHeartbeatTimer = NULL;
1647
#endif
1648

1649 1650 1651 1652
  for (int i = 0; i < pSyncNode->peersNum; ++i) {
    SSyncTimer* pSyncTimer = syncNodeGetHbTimer(pSyncNode, &(pSyncNode->peersId[i]));
    syncHbTimerStop(pSyncNode, pSyncTimer);
  }
1653

M
Minghao Li 已提交
1654 1655 1656
  return ret;
}

1657 1658 1659 1660 1661 1662
int32_t syncNodeRestartHeartbeatTimer(SSyncNode* pSyncNode) {
  syncNodeStopHeartbeatTimer(pSyncNode);
  syncNodeStartHeartbeatTimer(pSyncNode);
  return 0;
}

M
Minghao Li 已提交
1663 1664 1665 1666
// utils --------------
int32_t syncNodeSendMsgById(const SRaftId* destRaftId, SSyncNode* pSyncNode, SRpcMsg* pMsg) {
  SEpSet epSet;
  syncUtilraftId2EpSet(destRaftId, &epSet);
M
Minghao Li 已提交
1667 1668 1669 1670
  if (pSyncNode->FpSendMsg != NULL) {
    // htonl
    syncUtilMsgHtoN(pMsg->pCont);

1671
    pMsg->info.noResp = 1;
S
Shengliang Guan 已提交
1672
    pSyncNode->FpSendMsg(&epSet, pMsg);
M
Minghao Li 已提交
1673
  } else {
M
Minghao Li 已提交
1674 1675
    sError("vgId:%d, sync send msg by id error, fp-send-msg is null", pSyncNode->vgId);
    return -1;
M
Minghao Li 已提交
1676
  }
M
Minghao Li 已提交
1677

M
Minghao Li 已提交
1678 1679 1680 1681 1682 1683
  return 0;
}

int32_t syncNodeSendMsgByInfo(const SNodeInfo* nodeInfo, SSyncNode* pSyncNode, SRpcMsg* pMsg) {
  SEpSet epSet;
  syncUtilnodeInfo2EpSet(nodeInfo, &epSet);
M
Minghao Li 已提交
1684 1685 1686 1687
  if (pSyncNode->FpSendMsg != NULL) {
    // htonl
    syncUtilMsgHtoN(pMsg->pCont);

1688
    pMsg->info.noResp = 1;
S
Shengliang Guan 已提交
1689
    pSyncNode->FpSendMsg(&epSet, pMsg);
M
Minghao Li 已提交
1690
  } else {
M
Minghao Li 已提交
1691
    sError("vgId:%d, sync send msg by info error, fp-send-msg is null", pSyncNode->vgId);
M
Minghao Li 已提交
1692
  }
M
Minghao Li 已提交
1693 1694 1695
  return 0;
}

M
Minghao Li 已提交
1696
cJSON* syncNode2Json(const SSyncNode* pSyncNode) {
C
Cary Xu 已提交
1697
  char   u64buf[128] = {0};
M
Minghao Li 已提交
1698 1699
  cJSON* pRoot = cJSON_CreateObject();

M
Minghao Li 已提交
1700 1701 1702
  if (pSyncNode != NULL) {
    // init by SSyncInfo
    cJSON_AddNumberToObject(pRoot, "vgId", pSyncNode->vgId);
M
Minghao Li 已提交
1703
    cJSON_AddItemToObject(pRoot, "SRaftCfg", raftCfg2Json(pSyncNode->pRaftCfg));
M
Minghao Li 已提交
1704
    cJSON_AddStringToObject(pRoot, "path", pSyncNode->path);
M
Minghao Li 已提交
1705 1706 1707
    cJSON_AddStringToObject(pRoot, "raftStorePath", pSyncNode->raftStorePath);
    cJSON_AddStringToObject(pRoot, "configPath", pSyncNode->configPath);

M
Minghao Li 已提交
1708 1709 1710
    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->pWal);
    cJSON_AddStringToObject(pRoot, "pWal", u64buf);

S
Shengliang Guan 已提交
1711
    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->msgcb);
M
Minghao Li 已提交
1712 1713 1714 1715
    cJSON_AddStringToObject(pRoot, "rpcClient", u64buf);
    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpSendMsg);
    cJSON_AddStringToObject(pRoot, "FpSendMsg", u64buf);

S
Shengliang Guan 已提交
1716
    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->msgcb);
M
Minghao Li 已提交
1717 1718 1719 1720 1721 1722 1723 1724 1725 1726 1727 1728 1729 1730 1731 1732 1733 1734 1735 1736 1737
    cJSON_AddStringToObject(pRoot, "queue", u64buf);
    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpEqMsg);
    cJSON_AddStringToObject(pRoot, "FpEqMsg", u64buf);

    // init internal
    cJSON* pMe = syncUtilNodeInfo2Json(&pSyncNode->myNodeInfo);
    cJSON_AddItemToObject(pRoot, "myNodeInfo", pMe);
    cJSON* pRaftId = syncUtilRaftId2Json(&pSyncNode->myRaftId);
    cJSON_AddItemToObject(pRoot, "myRaftId", pRaftId);

    cJSON_AddNumberToObject(pRoot, "peersNum", pSyncNode->peersNum);
    cJSON* pPeers = cJSON_CreateArray();
    cJSON_AddItemToObject(pRoot, "peersNodeInfo", pPeers);
    for (int i = 0; i < pSyncNode->peersNum; ++i) {
      cJSON_AddItemToArray(pPeers, syncUtilNodeInfo2Json(&pSyncNode->peersNodeInfo[i]));
    }
    cJSON* pPeersId = cJSON_CreateArray();
    cJSON_AddItemToObject(pRoot, "peersId", pPeersId);
    for (int i = 0; i < pSyncNode->peersNum; ++i) {
      cJSON_AddItemToArray(pPeersId, syncUtilRaftId2Json(&pSyncNode->peersId[i]));
    }
M
Minghao Li 已提交
1738

M
Minghao Li 已提交
1739 1740 1741 1742 1743 1744
    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]));
    }
M
Minghao Li 已提交
1745

M
Minghao Li 已提交
1746 1747 1748 1749 1750 1751 1752
    // 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);

M
Minghao Li 已提交
1753
    // life cycle
S
Shengliang Guan 已提交
1754
    snprintf(u64buf, sizeof(u64buf), "%" PRId64, pSyncNode->rid);
M
Minghao Li 已提交
1755 1756
    cJSON_AddStringToObject(pRoot, "rid", u64buf);

M
Minghao Li 已提交
1757 1758 1759
    // tla+ server vars
    cJSON_AddNumberToObject(pRoot, "state", pSyncNode->state);
    cJSON_AddStringToObject(pRoot, "state_str", syncUtilState2String(pSyncNode->state));
M
Minghao Li 已提交
1760
    cJSON_AddItemToObject(pRoot, "pRaftStore", raftStore2Json(pSyncNode->pRaftStore));
M
Minghao Li 已提交
1761 1762 1763 1764 1765 1766 1767 1768 1769 1770 1771

    // tla+ candidate vars
    cJSON_AddItemToObject(pRoot, "pVotesGranted", voteGranted2Json(pSyncNode->pVotesGranted));
    cJSON_AddItemToObject(pRoot, "pVotesRespond", votesRespond2Json(pSyncNode->pVotesRespond));

    // tla+ leader vars
    cJSON_AddItemToObject(pRoot, "pNextIndex", syncIndexMgr2Json(pSyncNode->pNextIndex));
    cJSON_AddItemToObject(pRoot, "pMatchIndex", syncIndexMgr2Json(pSyncNode->pMatchIndex));

    // tla+ log vars
    cJSON_AddItemToObject(pRoot, "pLogStore", logStore2Json(pSyncNode->pLogStore));
S
Shengliang Guan 已提交
1772
    snprintf(u64buf, sizeof(u64buf), "%" PRId64, pSyncNode->commitIndex);
M
Minghao Li 已提交
1773 1774
    cJSON_AddStringToObject(pRoot, "commitIndex", u64buf);

M
Minghao Li 已提交
1775 1776 1777 1778 1779
    // timer ms init
    cJSON_AddNumberToObject(pRoot, "pingBaseLine", pSyncNode->pingBaseLine);
    cJSON_AddNumberToObject(pRoot, "electBaseLine", pSyncNode->electBaseLine);
    cJSON_AddNumberToObject(pRoot, "hbBaseLine", pSyncNode->hbBaseLine);

M
Minghao Li 已提交
1780 1781 1782 1783
    // ping timer
    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->pPingTimer);
    cJSON_AddStringToObject(pRoot, "pPingTimer", u64buf);
    cJSON_AddNumberToObject(pRoot, "pingTimerMS", pSyncNode->pingTimerMS);
S
Shengliang Guan 已提交
1784
    snprintf(u64buf, sizeof(u64buf), "%" PRIu64, pSyncNode->pingTimerLogicClock);
M
Minghao Li 已提交
1785
    cJSON_AddStringToObject(pRoot, "pingTimerLogicClock", u64buf);
S
Shengliang Guan 已提交
1786
    snprintf(u64buf, sizeof(u64buf), "%" PRIu64, pSyncNode->pingTimerLogicClockUser);
M
Minghao Li 已提交
1787 1788 1789
    cJSON_AddStringToObject(pRoot, "pingTimerLogicClockUser", u64buf);
    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpPingTimerCB);
    cJSON_AddStringToObject(pRoot, "FpPingTimerCB", u64buf);
S
Shengliang Guan 已提交
1790
    snprintf(u64buf, sizeof(u64buf), "%" PRIu64, pSyncNode->pingTimerCounter);
M
Minghao Li 已提交
1791 1792 1793 1794 1795 1796
    cJSON_AddStringToObject(pRoot, "pingTimerCounter", u64buf);

    // elect timer
    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->pElectTimer);
    cJSON_AddStringToObject(pRoot, "pElectTimer", u64buf);
    cJSON_AddNumberToObject(pRoot, "electTimerMS", pSyncNode->electTimerMS);
S
Shengliang Guan 已提交
1797
    snprintf(u64buf, sizeof(u64buf), "%" PRIu64, pSyncNode->electTimerLogicClock);
M
Minghao Li 已提交
1798
    cJSON_AddStringToObject(pRoot, "electTimerLogicClock", u64buf);
S
Shengliang Guan 已提交
1799
    snprintf(u64buf, sizeof(u64buf), "%" PRIu64, pSyncNode->electTimerLogicClockUser);
M
Minghao Li 已提交
1800 1801 1802
    cJSON_AddStringToObject(pRoot, "electTimerLogicClockUser", u64buf);
    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpElectTimerCB);
    cJSON_AddStringToObject(pRoot, "FpElectTimerCB", u64buf);
S
Shengliang Guan 已提交
1803
    snprintf(u64buf, sizeof(u64buf), "%" PRIu64, pSyncNode->electTimerCounter);
M
Minghao Li 已提交
1804 1805 1806 1807 1808 1809
    cJSON_AddStringToObject(pRoot, "electTimerCounter", u64buf);

    // heartbeat timer
    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->pHeartbeatTimer);
    cJSON_AddStringToObject(pRoot, "pHeartbeatTimer", u64buf);
    cJSON_AddNumberToObject(pRoot, "heartbeatTimerMS", pSyncNode->heartbeatTimerMS);
S
Shengliang Guan 已提交
1810
    snprintf(u64buf, sizeof(u64buf), "%" PRIu64, pSyncNode->heartbeatTimerLogicClock);
M
Minghao Li 已提交
1811
    cJSON_AddStringToObject(pRoot, "heartbeatTimerLogicClock", u64buf);
S
Shengliang Guan 已提交
1812
    snprintf(u64buf, sizeof(u64buf), "%" PRIu64, pSyncNode->heartbeatTimerLogicClockUser);
M
Minghao Li 已提交
1813 1814 1815
    cJSON_AddStringToObject(pRoot, "heartbeatTimerLogicClockUser", u64buf);
    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpHeartbeatTimerCB);
    cJSON_AddStringToObject(pRoot, "FpHeartbeatTimerCB", u64buf);
S
Shengliang Guan 已提交
1816
    snprintf(u64buf, sizeof(u64buf), "%" PRIu64, pSyncNode->heartbeatTimerCounter);
M
Minghao Li 已提交
1817 1818 1819 1820 1821 1822 1823 1824 1825 1826 1827 1828 1829 1830 1831 1832 1833
    cJSON_AddStringToObject(pRoot, "heartbeatTimerCounter", u64buf);

    // callback
    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpOnPing);
    cJSON_AddStringToObject(pRoot, "FpOnPing", u64buf);
    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpOnPingReply);
    cJSON_AddStringToObject(pRoot, "FpOnPingReply", u64buf);
    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpOnRequestVote);
    cJSON_AddStringToObject(pRoot, "FpOnRequestVote", u64buf);
    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpOnRequestVoteReply);
    cJSON_AddStringToObject(pRoot, "FpOnRequestVoteReply", u64buf);
    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpOnAppendEntries);
    cJSON_AddStringToObject(pRoot, "FpOnAppendEntries", u64buf);
    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpOnAppendEntriesReply);
    cJSON_AddStringToObject(pRoot, "FpOnAppendEntriesReply", u64buf);
    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpOnTimeout);
    cJSON_AddStringToObject(pRoot, "FpOnTimeout", u64buf);
M
Minghao Li 已提交
1834 1835 1836 1837 1838 1839 1840 1841 1842 1843 1844 1845 1846

    // restoreFinish
    cJSON_AddNumberToObject(pRoot, "restoreFinish", pSyncNode->restoreFinish);

    // snapshot senders
    cJSON* pSenders = cJSON_CreateArray();
    cJSON_AddItemToObject(pRoot, "senders", pSenders);
    for (int i = 0; i < TSDB_MAX_REPLICA; ++i) {
      cJSON_AddItemToArray(pSenders, snapshotSender2Json((pSyncNode->senders)[i]));
    }

    // snapshot receivers
    cJSON* pReceivers = cJSON_CreateArray();
1847
    cJSON_AddItemToObject(pRoot, "receiver", snapshotReceiver2Json(pSyncNode->pNewNodeReceiver));
M
Minghao Li 已提交
1848 1849 1850

    // changing
    cJSON_AddNumberToObject(pRoot, "changing", pSyncNode->changing);
M
Minghao Li 已提交
1851 1852 1853 1854 1855 1856 1857
  }

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

M
Minghao Li 已提交
1858 1859 1860 1861 1862 1863 1864
char* syncNode2Str(const SSyncNode* pSyncNode) {
  cJSON* pJson = syncNode2Json(pSyncNode);
  char*  serialized = cJSON_Print(pJson);
  cJSON_Delete(pJson);
  return serialized;
}

1865
inline void syncNodeEventLog(const SSyncNode* pSyncNode, char* str) {
M
Minghao Li 已提交
1866
  SSnapshot snapshot = {.data = NULL, .lastApplyIndex = -1, .lastApplyTerm = 0};
M
Minghao Li 已提交
1867
  if (pSyncNode->pFsm != NULL && pSyncNode->pFsm->FpGetSnapshotInfo != NULL) {
M
Minghao Li 已提交
1868 1869
    pSyncNode->pFsm->FpGetSnapshotInfo(pSyncNode->pFsm, &snapshot);
  }
M
Minghao Li 已提交
1870 1871 1872 1873 1874 1875 1876

  SyncIndex logLastIndex = SYNC_INDEX_INVALID;
  SyncIndex logBeginIndex = SYNC_INDEX_INVALID;
  if (pSyncNode->pLogStore != NULL) {
    logLastIndex = pSyncNode->pLogStore->syncLogLastIndex(pSyncNode->pLogStore);
    logBeginIndex = pSyncNode->pLogStore->syncLogBeginIndex(pSyncNode->pLogStore);
  }
M
Minghao Li 已提交
1877

M
Minghao Li 已提交
1878
  char* pCfgStr = syncCfg2SimpleStr(&(pSyncNode->pRaftCfg->cfg));
1879 1880 1881 1882
  char* printStr = "";
  if (pCfgStr != NULL) {
    printStr = pCfgStr;
  }
M
Minghao Li 已提交
1883

1884 1885 1886
  char*   peerStateStr = syncNodePeerState2Str(pSyncNode);
  int32_t userStrLen = strlen(str) + strlen(peerStateStr);

M
Minghao Li 已提交
1887
  if (userStrLen < 256) {
M
Minghao Li 已提交
1888
    char logBuf[256 + 256];
1889 1890
    if (pSyncNode != NULL && pSyncNode->pRaftCfg != NULL && pSyncNode->pRaftStore != NULL) {
      snprintf(logBuf, sizeof(logBuf),
M
Minghao Li 已提交
1891 1892
               "vgId:%d, sync %s %s, tm:%" PRIu64 ", cmt:%" PRId64 ", fst:%" PRId64 ", lst:%" PRId64 ", min:%" PRId64
               ", snap:%" PRId64 ", snap-tm:%" PRIu64
M
Minghao Li 已提交
1893 1894 1895
               ", sby:%d, "
               "stgy:%d, bch:%d, "
               "r-num:%d, "
1896
               "lcfg:%" PRId64 ", chging:%d, rsto:%d, dquorum:%d, elt:%" PRId64 ", hb:%" PRId64 ", %s, %s",
1897
               pSyncNode->vgId, syncUtilState2String(pSyncNode->state), str, pSyncNode->pRaftStore->currentTerm,
M
Minghao Li 已提交
1898 1899 1900 1901
               pSyncNode->commitIndex, logBeginIndex, logLastIndex, pSyncNode->minMatchIndex, snapshot.lastApplyIndex,
               snapshot.lastApplyTerm, pSyncNode->pRaftCfg->isStandBy, pSyncNode->pRaftCfg->snapshotStrategy,
               pSyncNode->pRaftCfg->batchSize, pSyncNode->replicaNum, pSyncNode->pRaftCfg->lastConfigIndex,
               pSyncNode->changing, pSyncNode->restoreFinish, syncNodeDynamicQuorum(pSyncNode),
1902
               pSyncNode->electTimerLogicClockUser, pSyncNode->heartbeatTimerLogicClockUser, peerStateStr, printStr);
1903 1904 1905
    } else {
      snprintf(logBuf, sizeof(logBuf), "%s", str);
    }
1906
    // sDebug("%s", logBuf);
M
Minghao Li 已提交
1907 1908
    // sInfo("%s", logBuf);
    sTrace("%s", logBuf);
M
Minghao Li 已提交
1909

M
Minghao Li 已提交
1910
  } else {
M
Minghao Li 已提交
1911
    int   len = 256 + userStrLen;
M
Minghao Li 已提交
1912
    char* s = (char*)taosMemoryMalloc(len);
1913 1914
    if (pSyncNode != NULL && pSyncNode->pRaftCfg != NULL && pSyncNode->pRaftStore != NULL) {
      snprintf(s, len,
M
Minghao Li 已提交
1915 1916
               "vgId:%d, sync %s %s, tm:%" PRIu64 ", cmt:%" PRId64 ", fst:%" PRId64 ", lst:%" PRId64 ", min:%" PRId64
               ", snap:%" PRId64 ", snap-tm:%" PRIu64
M
Minghao Li 已提交
1917 1918 1919
               ", sby:%d, "
               "stgy:%d, bch:%d, "
               "r-num:%d, "
1920
               "lcfg:%" PRId64 ", chging:%d, rsto:%d, dquorum:%d, elt:%" PRId64 ", hb:%" PRId64 ", %s, %s",
1921
               pSyncNode->vgId, syncUtilState2String(pSyncNode->state), str, pSyncNode->pRaftStore->currentTerm,
M
Minghao Li 已提交
1922 1923 1924 1925
               pSyncNode->commitIndex, logBeginIndex, logLastIndex, pSyncNode->minMatchIndex, snapshot.lastApplyIndex,
               snapshot.lastApplyTerm, pSyncNode->pRaftCfg->isStandBy, pSyncNode->pRaftCfg->snapshotStrategy,
               pSyncNode->pRaftCfg->batchSize, pSyncNode->replicaNum, pSyncNode->pRaftCfg->lastConfigIndex,
               pSyncNode->changing, pSyncNode->restoreFinish, syncNodeDynamicQuorum(pSyncNode),
1926
               pSyncNode->electTimerLogicClockUser, pSyncNode->heartbeatTimerLogicClockUser, peerStateStr, printStr);
1927 1928 1929
    } else {
      snprintf(s, len, "%s", str);
    }
1930
    // sDebug("%s", s);
M
Minghao Li 已提交
1931 1932
    // sInfo("%s", s);
    sTrace("%s", s);
M
Minghao Li 已提交
1933 1934
    taosMemoryFree(s);
  }
M
Minghao Li 已提交
1935

M
Minghao Li 已提交
1936
  taosMemoryFree(peerStateStr);
M
Minghao Li 已提交
1937
  taosMemoryFree(pCfgStr);
M
Minghao Li 已提交
1938 1939
}

1940
inline void syncNodeErrorLog(const SSyncNode* pSyncNode, char* str) {
M
Minghao Li 已提交
1941 1942 1943
  int32_t userStrLen = strlen(str);

  SSnapshot snapshot = {.data = NULL, .lastApplyIndex = -1, .lastApplyTerm = 0};
M
Minghao Li 已提交
1944
  if (pSyncNode->pFsm != NULL && pSyncNode->pFsm->FpGetSnapshotInfo != NULL) {
M
Minghao Li 已提交
1945 1946
    pSyncNode->pFsm->FpGetSnapshotInfo(pSyncNode->pFsm, &snapshot);
  }
M
Minghao Li 已提交
1947 1948 1949 1950 1951 1952 1953 1954 1955 1956 1957 1958 1959

  SyncIndex logLastIndex = SYNC_INDEX_INVALID;
  SyncIndex logBeginIndex = SYNC_INDEX_INVALID;
  if (pSyncNode->pLogStore != NULL) {
    logLastIndex = pSyncNode->pLogStore->syncLogLastIndex(pSyncNode->pLogStore);
    logBeginIndex = pSyncNode->pLogStore->syncLogBeginIndex(pSyncNode->pLogStore);
  }

  char* pCfgStr = syncCfg2SimpleStr(&(pSyncNode->pRaftCfg->cfg));
  char* printStr = "";
  if (pCfgStr != NULL) {
    printStr = pCfgStr;
  }
M
Minghao Li 已提交
1960 1961

  if (userStrLen < 256) {
M
Minghao Li 已提交
1962
    char logBuf[256 + 256];
1963 1964
    if (pSyncNode != NULL && pSyncNode->pRaftCfg != NULL && pSyncNode->pRaftStore != NULL) {
      snprintf(logBuf, sizeof(logBuf),
M
Minghao Li 已提交
1965 1966
               "vgId:%d, sync %s %s, tm:%" PRIu64 ", cmt:%" PRId64 ", fst:%" PRId64 ", lst:%" PRId64 ", min:%" PRId64
               ", snap:%" PRId64 ", snap-tm:%" PRIu64
M
Minghao Li 已提交
1967 1968 1969
               ", sby:%d, "
               "stgy:%d, bch:%d, "
               "r-num:%d, "
M
Minghao Li 已提交
1970
               "lcfg:%" PRId64 ", chging:%d, rsto:%d, dquorum:%d, elt:%" PRId64 ", hb:%" PRId64 ", %s",
1971
               pSyncNode->vgId, syncUtilState2String(pSyncNode->state), str, pSyncNode->pRaftStore->currentTerm,
M
Minghao Li 已提交
1972 1973 1974 1975 1976
               pSyncNode->commitIndex, logBeginIndex, logLastIndex, pSyncNode->minMatchIndex, snapshot.lastApplyIndex,
               snapshot.lastApplyTerm, pSyncNode->pRaftCfg->isStandBy, pSyncNode->pRaftCfg->snapshotStrategy,
               pSyncNode->pRaftCfg->batchSize, pSyncNode->replicaNum, pSyncNode->pRaftCfg->lastConfigIndex,
               pSyncNode->changing, pSyncNode->restoreFinish, syncNodeDynamicQuorum(pSyncNode),
               pSyncNode->electTimerLogicClockUser, pSyncNode->heartbeatTimerLogicClockUser, printStr);
1977 1978 1979
    } else {
      snprintf(logBuf, sizeof(logBuf), "%s", str);
    }
M
Minghao Li 已提交
1980 1981 1982
    sError("%s", logBuf);

  } else {
M
Minghao Li 已提交
1983
    int   len = 256 + userStrLen;
M
Minghao Li 已提交
1984
    char* s = (char*)taosMemoryMalloc(len);
1985 1986
    if (pSyncNode != NULL && pSyncNode->pRaftCfg != NULL && pSyncNode->pRaftStore != NULL) {
      snprintf(s, len,
M
Minghao Li 已提交
1987 1988
               "vgId:%d, sync %s %s, tm:%" PRIu64 ", cmt:%" PRId64 ", fst:%" PRId64 ", lst:%" PRId64 ", min:%" PRId64
               ", snap:%" PRId64 ", snap-tm:%" PRIu64
M
Minghao Li 已提交
1989 1990 1991
               ", sby:%d, "
               "stgy:%d, bch:%d, "
               "r-num:%d, "
M
Minghao Li 已提交
1992
               "lcfg:%" PRId64 ", chging:%d, rsto:%d, dquorum:%d, elt:%" PRId64 ", hb:%" PRId64 ", %s",
1993
               pSyncNode->vgId, syncUtilState2String(pSyncNode->state), str, pSyncNode->pRaftStore->currentTerm,
M
Minghao Li 已提交
1994 1995 1996 1997 1998
               pSyncNode->commitIndex, logBeginIndex, logLastIndex, pSyncNode->minMatchIndex, snapshot.lastApplyIndex,
               snapshot.lastApplyTerm, pSyncNode->pRaftCfg->isStandBy, pSyncNode->pRaftCfg->snapshotStrategy,
               pSyncNode->pRaftCfg->batchSize, pSyncNode->replicaNum, pSyncNode->pRaftCfg->lastConfigIndex,
               pSyncNode->changing, pSyncNode->restoreFinish, syncNodeDynamicQuorum(pSyncNode),
               pSyncNode->electTimerLogicClockUser, pSyncNode->heartbeatTimerLogicClockUser, printStr);
1999 2000 2001
    } else {
      snprintf(s, len, "%s", str);
    }
M
Minghao Li 已提交
2002 2003 2004
    sError("%s", s);
    taosMemoryFree(s);
  }
M
Minghao Li 已提交
2005 2006

  taosMemoryFree(pCfgStr);
M
Minghao Li 已提交
2007 2008
}

2009
inline char* syncNode2SimpleStr(const SSyncNode* pSyncNode) {
M
Minghao Li 已提交
2010 2011
  int   len = 256;
  char* s = (char*)taosMemoryMalloc(len);
M
Minghao Li 已提交
2012 2013 2014 2015 2016 2017 2018 2019

  SSnapshot snapshot = {.data = NULL, .lastApplyIndex = -1, .lastApplyTerm = 0};
  if (pSyncNode->pFsm->FpGetSnapshotInfo != NULL) {
    pSyncNode->pFsm->FpGetSnapshotInfo(pSyncNode->pFsm, &snapshot);
  }
  SyncIndex logLastIndex = pSyncNode->pLogStore->syncLogLastIndex(pSyncNode->pLogStore);
  SyncIndex logBeginIndex = pSyncNode->pLogStore->syncLogBeginIndex(pSyncNode->pLogStore);

M
Minghao Li 已提交
2020
  snprintf(s, len,
M
Minghao Li 已提交
2021 2022 2023 2024
           "vgId:%d, sync %s, tm:%" PRIu64 ", cmt:%" PRId64 ", fst:%" PRId64 ", lst:%" PRId64 ", snap:%" PRId64
           ", sby:%d, "
           "r-num:%d, "
           "lcfg:%" PRId64 ", chging:%d, rsto:%d",
M
Minghao Li 已提交
2025 2026 2027 2028
           pSyncNode->vgId, syncUtilState2String(pSyncNode->state), pSyncNode->pRaftStore->currentTerm,
           pSyncNode->commitIndex, logBeginIndex, logLastIndex, snapshot.lastApplyIndex, pSyncNode->pRaftCfg->isStandBy,
           pSyncNode->replicaNum, pSyncNode->pRaftCfg->lastConfigIndex, pSyncNode->changing, pSyncNode->restoreFinish);

M
Minghao Li 已提交
2029 2030 2031
  return s;
}

2032
inline bool syncNodeInConfig(SSyncNode* pSyncNode, const SSyncCfg* config) {
2033 2034 2035 2036 2037 2038 2039 2040 2041 2042 2043 2044 2045 2046 2047 2048 2049 2050 2051 2052 2053 2054 2055 2056 2057 2058
  bool b1 = false;
  bool b2 = false;

  for (int i = 0; i < config->replicaNum; ++i) {
    if (strcmp((config->nodeInfo)[i].nodeFqdn, pSyncNode->myNodeInfo.nodeFqdn) == 0 &&
        (config->nodeInfo)[i].nodePort == pSyncNode->myNodeInfo.nodePort) {
      b1 = true;
      break;
    }
  }

  for (int i = 0; i < config->replicaNum; ++i) {
    SRaftId raftId;
    raftId.addr = syncUtilAddr2U64((config->nodeInfo)[i].nodeFqdn, (config->nodeInfo)[i].nodePort);
    raftId.vgId = pSyncNode->vgId;

    if (syncUtilSameId(&raftId, &(pSyncNode->myRaftId))) {
      b2 = true;
      break;
    }
  }

  ASSERT(b1 == b2);
  return b1;
}

2059 2060 2061 2062 2063 2064 2065 2066 2067 2068 2069 2070 2071
static bool syncIsConfigChanged(const SSyncCfg* pOldCfg, const SSyncCfg* pNewCfg) {
  if (pOldCfg->replicaNum != pNewCfg->replicaNum) return true;
  if (pOldCfg->myIndex != pNewCfg->myIndex) return true;
  for (int32_t i = 0; i < pOldCfg->replicaNum; ++i) {
    const SNodeInfo* pOldInfo = &pOldCfg->nodeInfo[i];
    const SNodeInfo* pNewInfo = &pNewCfg->nodeInfo[i];
    if (strcmp(pOldInfo->nodeFqdn, pNewInfo->nodeFqdn) != 0) return true;
    if (pOldInfo->nodePort != pNewInfo->nodePort) return true;
  }

  return false;
}

M
Minghao Li 已提交
2072
void syncNodeDoConfigChange(SSyncNode* pSyncNode, SSyncCfg* pNewConfig, SyncIndex lastConfigChangeIndex) {
2073
  SSyncCfg oldConfig = pSyncNode->pRaftCfg->cfg;
S
Shengliang Guan 已提交
2074
#if 1
2075 2076 2077 2078 2079
  if (!syncIsConfigChanged(&oldConfig, pNewConfig)) {
    sInfo("vgId:1, sync not reconfig since not changed");
    return;
  }
#endif
2080
  pSyncNode->pRaftCfg->cfg = *pNewConfig;
2081 2082
  pSyncNode->pRaftCfg->lastConfigIndex = lastConfigChangeIndex;

M
Minghao Li 已提交
2083 2084
  bool IamInOld = syncNodeInConfig(pSyncNode, &oldConfig);
  bool IamInNew = syncNodeInConfig(pSyncNode, pNewConfig);
M
Minghao Li 已提交
2085

M
Minghao Li 已提交
2086 2087
  bool isDrop = false;
  bool isAdd = false;
M
Minghao Li 已提交
2088

M
Minghao Li 已提交
2089 2090 2091 2092
  if (IamInOld && !IamInNew) {
    isDrop = true;
  } else {
    isDrop = false;
2093
  }
2094

M
Minghao Li 已提交
2095 2096 2097 2098 2099
  if (!IamInOld && IamInNew) {
    isAdd = true;
  } else {
    isAdd = false;
  }
M
Minghao Li 已提交
2100

M
Minghao Li 已提交
2101 2102 2103 2104 2105 2106 2107 2108 2109 2110 2111
  // log begin config change
  do {
    char  eventLog[256];
    char* pOldCfgStr = syncCfg2SimpleStr(&oldConfig);
    char* pNewCfgStr = syncCfg2SimpleStr(pNewConfig);
    snprintf(eventLog, sizeof(eventLog), "begin do config change, from %s to %s", pOldCfgStr, pNewCfgStr);
    syncNodeEventLog(pSyncNode, eventLog);
    taosMemoryFree(pOldCfgStr);
    taosMemoryFree(pNewCfgStr);
  } while (0);

M
Minghao Li 已提交
2112 2113
  if (IamInNew) {
    pSyncNode->pRaftCfg->isStandBy = 0;  // change isStandBy to normal
M
Minghao Li 已提交
2114
  }
M
Minghao Li 已提交
2115 2116
  if (isDrop) {
    pSyncNode->pRaftCfg->isStandBy = 1;  // set standby
M
Minghao Li 已提交
2117 2118
  }

M
Minghao Li 已提交
2119
  // add last config index
M
Minghao Li 已提交
2120
  raftCfgAddConfigIndex(pSyncNode->pRaftCfg, lastConfigChangeIndex);
M
Minghao Li 已提交
2121

M
Minghao Li 已提交
2122 2123 2124 2125 2126 2127 2128 2129 2130 2131 2132
  if (IamInNew) {
    //-----------------------------------------
    int32_t ret = 0;

    // save snapshot senders
    int32_t oldReplicaNum = pSyncNode->replicaNum;
    SRaftId oldReplicasId[TSDB_MAX_REPLICA];
    memcpy(oldReplicasId, pSyncNode->replicasId, sizeof(oldReplicasId));
    SSyncSnapshotSender* oldSenders[TSDB_MAX_REPLICA];
    for (int i = 0; i < TSDB_MAX_REPLICA; ++i) {
      oldSenders[i] = (pSyncNode->senders)[i];
M
Minghao Li 已提交
2133

M
Minghao Li 已提交
2134 2135 2136 2137
      char* eventLog = snapshotSender2SimpleStr(oldSenders[i], "snapshot sender save old");
      syncNodeEventLog(pSyncNode, eventLog);
      taosMemoryFree(eventLog);
    }
2138

M
Minghao Li 已提交
2139 2140 2141 2142 2143 2144 2145 2146 2147 2148 2149 2150 2151 2152 2153 2154
    // init internal
    pSyncNode->myNodeInfo = pSyncNode->pRaftCfg->cfg.nodeInfo[pSyncNode->pRaftCfg->cfg.myIndex];
    syncUtilnodeInfo2raftId(&pSyncNode->myNodeInfo, pSyncNode->vgId, &pSyncNode->myRaftId);

    // init peersNum, peers, peersId
    pSyncNode->peersNum = pSyncNode->pRaftCfg->cfg.replicaNum - 1;
    int j = 0;
    for (int i = 0; i < pSyncNode->pRaftCfg->cfg.replicaNum; ++i) {
      if (i != pSyncNode->pRaftCfg->cfg.myIndex) {
        pSyncNode->peersNodeInfo[j] = pSyncNode->pRaftCfg->cfg.nodeInfo[i];
        j++;
      }
    }
    for (int i = 0; i < pSyncNode->peersNum; ++i) {
      syncUtilnodeInfo2raftId(&pSyncNode->peersNodeInfo[i], pSyncNode->vgId, &pSyncNode->peersId[i]);
    }
2155

M
Minghao Li 已提交
2156 2157 2158 2159 2160
    // init replicaNum, replicasId
    pSyncNode->replicaNum = pSyncNode->pRaftCfg->cfg.replicaNum;
    for (int i = 0; i < pSyncNode->pRaftCfg->cfg.replicaNum; ++i) {
      syncUtilnodeInfo2raftId(&pSyncNode->pRaftCfg->cfg.nodeInfo[i], pSyncNode->vgId, &pSyncNode->replicasId[i]);
    }
2161

2162 2163 2164
    // update quorum first
    pSyncNode->quorum = syncUtilQuorum(pSyncNode->pRaftCfg->cfg.replicaNum);

M
Minghao Li 已提交
2165 2166 2167 2168
    syncIndexMgrUpdate(pSyncNode->pNextIndex, pSyncNode);
    syncIndexMgrUpdate(pSyncNode->pMatchIndex, pSyncNode);
    voteGrantedUpdate(pSyncNode->pVotesGranted, pSyncNode);
    votesRespondUpdate(pSyncNode->pVotesRespond, pSyncNode);
M
Minghao Li 已提交
2169

M
Minghao Li 已提交
2170
    // reset snapshot senders
2171

M
Minghao Li 已提交
2172 2173 2174 2175
    // clear new
    for (int i = 0; i < TSDB_MAX_REPLICA; ++i) {
      (pSyncNode->senders)[i] = NULL;
    }
M
Minghao Li 已提交
2176

M
Minghao Li 已提交
2177 2178 2179 2180 2181 2182 2183 2184 2185 2186 2187 2188
    // reset new
    for (int i = 0; i < pSyncNode->replicaNum; ++i) {
      // reset sender
      bool reset = false;
      for (int j = 0; j < TSDB_MAX_REPLICA; ++j) {
        if (syncUtilSameId(&(pSyncNode->replicasId)[i], &oldReplicasId[j])) {
          char     host[128];
          uint16_t port;
          syncUtilU642Addr((pSyncNode->replicasId)[i].addr, host, sizeof(host), &port);

          do {
            char eventLog[256];
S
Shengliang Guan 已提交
2189
            snprintf(eventLog, sizeof(eventLog), "snapshot sender reset for: %" PRIu64 ", newIndex:%d, %s:%d, %p",
M
Minghao Li 已提交
2190 2191 2192 2193 2194 2195 2196 2197 2198 2199 2200 2201 2202 2203 2204 2205 2206 2207 2208 2209
                     (pSyncNode->replicasId)[i].addr, i, host, port, oldSenders[j]);
            syncNodeEventLog(pSyncNode, eventLog);
          } while (0);

          (pSyncNode->senders)[i] = oldSenders[j];
          oldSenders[j] = NULL;
          reset = true;

          // reset replicaIndex
          int32_t oldreplicaIndex = (pSyncNode->senders)[i]->replicaIndex;
          (pSyncNode->senders)[i]->replicaIndex = i;

          do {
            char eventLog[256];
            snprintf(eventLog, sizeof(eventLog),
                     "snapshot sender udpate replicaIndex from %d to %d, %s:%d, %p, reset:%d", oldreplicaIndex, i, host,
                     port, (pSyncNode->senders)[i], reset);
            syncNodeEventLog(pSyncNode, eventLog);
          } while (0);
        }
2210 2211
      }
    }
2212

M
Minghao Li 已提交
2213 2214 2215 2216
    // create new
    for (int i = 0; i < TSDB_MAX_REPLICA; ++i) {
      if ((pSyncNode->senders)[i] == NULL) {
        (pSyncNode->senders)[i] = snapshotSenderCreate(pSyncNode, i);
M
Minghao Li 已提交
2217

M
Minghao Li 已提交
2218 2219 2220 2221
        char* eventLog = snapshotSender2SimpleStr((pSyncNode->senders)[i], "snapshot sender create new");
        syncNodeEventLog(pSyncNode, eventLog);
        taosMemoryFree(eventLog);
      }
2222 2223
    }

M
Minghao Li 已提交
2224 2225 2226 2227
    // free old
    for (int i = 0; i < TSDB_MAX_REPLICA; ++i) {
      if (oldSenders[i] != NULL) {
        snapshotSenderDestroy(oldSenders[i]);
M
Minghao Li 已提交
2228

M
Minghao Li 已提交
2229 2230 2231 2232 2233
        do {
          char eventLog[128];
          snprintf(eventLog, sizeof(eventLog), "snapshot sender delete old %p replica-index:%d", oldSenders[i], i);
          syncNodeEventLog(pSyncNode, eventLog);
        } while (0);
M
Minghao Li 已提交
2234

M
Minghao Li 已提交
2235 2236
        oldSenders[i] = NULL;
      }
2237 2238
    }

2239
    // persist cfg
M
Minghao Li 已提交
2240
    raftCfgPersist(pSyncNode->pRaftCfg);
2241

M
Minghao Li 已提交
2242 2243 2244
    char  tmpbuf[512];
    char* oldStr = syncCfg2SimpleStr(&oldConfig);
    char* newStr = syncCfg2SimpleStr(pNewConfig);
2245 2246
    snprintf(tmpbuf, sizeof(tmpbuf), "config change from %d to %d, index:%" PRId64 ", %s  -->  %s",
             oldConfig.replicaNum, pNewConfig->replicaNum, lastConfigChangeIndex, oldStr, newStr);
M
Minghao Li 已提交
2247 2248
    taosMemoryFree(oldStr);
    taosMemoryFree(newStr);
M
Minghao Li 已提交
2249

M
Minghao Li 已提交
2250 2251 2252
    // change isStandBy to normal (election timeout)
    if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
      syncNodeBecomeLeader(pSyncNode, tmpbuf);
2253 2254 2255 2256 2257

      // Raft 3.6.2 Committing entries from previous terms
      syncNodeAppendNoop(pSyncNode);
      syncMaybeAdvanceCommitIndex(pSyncNode);

M
Minghao Li 已提交
2258 2259 2260 2261
    } else {
      syncNodeBecomeFollower(pSyncNode, tmpbuf);
    }
  } else {
2262
    // persist cfg
M
Minghao Li 已提交
2263
    raftCfgPersist(pSyncNode->pRaftCfg);
2264

M
Minghao Li 已提交
2265 2266 2267
    char  tmpbuf[512];
    char* oldStr = syncCfg2SimpleStr(&oldConfig);
    char* newStr = syncCfg2SimpleStr(pNewConfig);
2268 2269
    snprintf(tmpbuf, sizeof(tmpbuf), "do not config change from %d to %d, index:%" PRId64 ", %s  -->  %s",
             oldConfig.replicaNum, pNewConfig->replicaNum, lastConfigChangeIndex, oldStr, newStr);
M
Minghao Li 已提交
2270 2271 2272
    taosMemoryFree(oldStr);
    taosMemoryFree(newStr);
    syncNodeEventLog(pSyncNode, tmpbuf);
2273
  }
2274

M
Minghao Li 已提交
2275
_END:
M
Minghao Li 已提交
2276 2277 2278 2279 2280 2281 2282 2283 2284 2285 2286

  // log end config change
  do {
    char  eventLog[256];
    char* pOldCfgStr = syncCfg2SimpleStr(&oldConfig);
    char* pNewCfgStr = syncCfg2SimpleStr(pNewConfig);
    snprintf(eventLog, sizeof(eventLog), "end do config change, from %s to %s", pOldCfgStr, pNewCfgStr);
    syncNodeEventLog(pSyncNode, eventLog);
    taosMemoryFree(pOldCfgStr);
    taosMemoryFree(pNewCfgStr);
  } while (0);
M
Minghao Li 已提交
2287
  return;
M
Minghao Li 已提交
2288 2289
}

M
Minghao Li 已提交
2290 2291 2292 2293 2294 2295 2296 2297 2298 2299 2300
SSyncNode* syncNodeAcquire(int64_t rid) {
  SSyncNode* pNode = taosAcquireRef(tsNodeRefId, rid);
  if (pNode == NULL) {
    sTrace("failed to acquire node from refId:%" PRId64, rid);
  }

  return pNode;
}

void syncNodeRelease(SSyncNode* pNode) { taosReleaseRef(tsNodeRefId, pNode->rid); }

M
Minghao Li 已提交
2301 2302 2303 2304
// raft state change --------------
void syncNodeUpdateTerm(SSyncNode* pSyncNode, SyncTerm term) {
  if (term > pSyncNode->pRaftStore->currentTerm) {
    raftStoreSetTerm(pSyncNode->pRaftStore, term);
2305
    char tmpBuf[64];
S
Shengliang Guan 已提交
2306
    snprintf(tmpBuf, sizeof(tmpBuf), "update term to %" PRIu64, term);
2307
    syncNodeBecomeFollower(pSyncNode, tmpBuf);
M
Minghao Li 已提交
2308 2309 2310 2311
    raftStoreClearVote(pSyncNode->pRaftStore);
  }
}

2312 2313 2314 2315 2316 2317
void syncNodeUpdateTermWithoutStepDown(SSyncNode* pSyncNode, SyncTerm term) {
  if (term > pSyncNode->pRaftStore->currentTerm) {
    raftStoreSetTerm(pSyncNode->pRaftStore, term);
  }
}

M
Minghao Li 已提交
2318 2319 2320 2321 2322
void syncNodeStepDown(SSyncNode* pSyncNode, SyncTerm newTerm) {
  ASSERT(pSyncNode->pRaftStore->currentTerm <= newTerm);

  do {
    char logBuf[128];
S
Shengliang Guan 已提交
2323
    snprintf(logBuf, sizeof(logBuf), "step down, new-term:%" PRIu64 ", current-term:%" PRIu64, newTerm,
M
Minghao Li 已提交
2324 2325 2326 2327 2328 2329 2330 2331 2332 2333 2334 2335 2336 2337 2338 2339 2340 2341
             pSyncNode->pRaftStore->currentTerm);
    syncNodeEventLog(pSyncNode, logBuf);
  } while (0);

  if (pSyncNode->pRaftStore->currentTerm < newTerm) {
    raftStoreSetTerm(pSyncNode->pRaftStore, newTerm);
    char tmpBuf[64];
    snprintf(tmpBuf, sizeof(tmpBuf), "step down, update term to %" PRIu64, newTerm);
    syncNodeBecomeFollower(pSyncNode, tmpBuf);
    raftStoreClearVote(pSyncNode->pRaftStore);

  } else {
    if (pSyncNode->state != TAOS_SYNC_STATE_FOLLOWER) {
      syncNodeBecomeFollower(pSyncNode, "step down");
    }
  }
}

2342 2343
void syncNodeLeaderChangeRsp(SSyncNode* pSyncNode) { syncRespCleanRsp(pSyncNode->pSyncRespMgr); }

2344
void syncNodeBecomeFollower(SSyncNode* pSyncNode, const char* debugStr) {
M
Minghao Li 已提交
2345
  // maybe clear leader cache
M
Minghao Li 已提交
2346 2347 2348 2349
  if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
    pSyncNode->leaderCache = EMPTY_RAFT_ID;
  }

M
Minghao Li 已提交
2350
  // state change
M
Minghao Li 已提交
2351 2352 2353
  pSyncNode->state = TAOS_SYNC_STATE_FOLLOWER;
  syncNodeStopHeartbeatTimer(pSyncNode);

M
Minghao Li 已提交
2354 2355
  // reset elect timer
  syncNodeResetElectTimer(pSyncNode);
M
Minghao Li 已提交
2356

2357 2358 2359
  // send rsp to client
  syncNodeLeaderChangeRsp(pSyncNode);

2360 2361 2362 2363 2364
  // call back
  if (pSyncNode->pFsm != NULL && pSyncNode->pFsm->FpBecomeFollowerCb != NULL) {
    pSyncNode->pFsm->FpBecomeFollowerCb(pSyncNode->pFsm);
  }

M
Minghao Li 已提交
2365 2366 2367
  // min match index
  pSyncNode->minMatchIndex = SYNC_INDEX_INVALID;

M
Minghao Li 已提交
2368 2369 2370 2371 2372 2373 2374 2375 2376 2377 2378 2379 2380 2381
  // trace log
  do {
    int32_t debugStrLen = strlen(debugStr);
    if (debugStrLen < 256) {
      char eventLog[256 + 64];
      snprintf(eventLog, sizeof(eventLog), "become follower %s", debugStr);
      syncNodeEventLog(pSyncNode, eventLog);
    } else {
      char* eventLog = taosMemoryMalloc(debugStrLen + 64);
      snprintf(eventLog, debugStrLen, "become follower %s", debugStr);
      syncNodeEventLog(pSyncNode, eventLog);
      taosMemoryFree(eventLog);
    }
  } while (0);
M
Minghao Li 已提交
2382 2383 2384 2385 2386 2387 2388 2389 2390 2391 2392 2393 2394 2395 2396 2397 2398 2399 2400 2401
}

// 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>>
//
2402
void syncNodeBecomeLeader(SSyncNode* pSyncNode, const char* debugStr) {
2403 2404
  pSyncNode->leaderTime = taosGetTimestampMs();

2405 2406 2407
  // reset restoreFinish
  pSyncNode->restoreFinish = false;

M
Minghao Li 已提交
2408
  // state change
M
Minghao Li 已提交
2409
  pSyncNode->state = TAOS_SYNC_STATE_LEADER;
M
Minghao Li 已提交
2410 2411

  // set leader cache
M
Minghao Li 已提交
2412 2413 2414
  pSyncNode->leaderCache = pSyncNode->myRaftId;

  for (int i = 0; i < pSyncNode->pNextIndex->replicaNum; ++i) {
M
Minghao Li 已提交
2415 2416
    // maybe overwrite myself, no harm
    // just do it!
2417 2418 2419 2420 2421 2422 2423 2424 2425

    // pSyncNode->pNextIndex->index[i] = pSyncNode->pLogStore->getLastIndex(pSyncNode->pLogStore) + 1;

    // maybe wal is deleted
    SyncIndex lastIndex;
    SyncTerm  lastTerm;
    int32_t   code = syncNodeGetLastIndexTerm(pSyncNode, &lastIndex, &lastTerm);
    ASSERT(code == 0);
    pSyncNode->pNextIndex->index[i] = lastIndex + 1;
M
Minghao Li 已提交
2426 2427 2428
  }

  for (int i = 0; i < pSyncNode->pMatchIndex->replicaNum; ++i) {
M
Minghao Li 已提交
2429 2430
    // maybe overwrite myself, no harm
    // just do it!
M
Minghao Li 已提交
2431 2432 2433
    pSyncNode->pMatchIndex->index[i] = SYNC_INDEX_INVALID;
  }

M
Minghao Li 已提交
2434 2435 2436
  // init peer mgr
  syncNodePeerStateInit(pSyncNode);

2437 2438
  // update sender private term
  SSyncSnapshotSender* pMySender = syncNodeGetSnapshotSender(pSyncNode, &(pSyncNode->myRaftId));
2439 2440 2441 2442 2443
  if (pMySender != NULL) {
    for (int i = 0; i < pSyncNode->pMatchIndex->replicaNum; ++i) {
      if ((pSyncNode->senders)[i]->privateTerm > pMySender->privateTerm) {
        pMySender->privateTerm = (pSyncNode->senders)[i]->privateTerm;
      }
2444
    }
2445
    (pMySender->privateTerm) += 100;
2446 2447
  }

2448 2449 2450 2451 2452
  // close receiver
  if (snapshotReceiverIsStart(pSyncNode->pNewNodeReceiver)) {
    snapshotReceiverForceStop(pSyncNode->pNewNodeReceiver);
  }

M
Minghao Li 已提交
2453
  // stop elect timer
M
Minghao Li 已提交
2454
  syncNodeStopElectTimer(pSyncNode);
M
Minghao Li 已提交
2455

M
Minghao Li 已提交
2456 2457
  // start heartbeat timer
  syncNodeStartHeartbeatTimer(pSyncNode);
M
Minghao Li 已提交
2458

M
Minghao Li 已提交
2459 2460
  // send heartbeat right now
  syncNodeHeartbeatPeers(pSyncNode);
M
Minghao Li 已提交
2461

2462 2463 2464 2465 2466
  // call back
  if (pSyncNode->pFsm != NULL && pSyncNode->pFsm->FpBecomeLeaderCb != NULL) {
    pSyncNode->pFsm->FpBecomeLeaderCb(pSyncNode->pFsm);
  }

M
Minghao Li 已提交
2467 2468 2469
  // min match index
  pSyncNode->minMatchIndex = SYNC_INDEX_INVALID;

M
Minghao Li 已提交
2470 2471 2472 2473 2474 2475 2476 2477 2478 2479 2480 2481 2482 2483
  // trace log
  do {
    int32_t debugStrLen = strlen(debugStr);
    if (debugStrLen < 256) {
      char eventLog[256 + 64];
      snprintf(eventLog, sizeof(eventLog), "become leader %s", debugStr);
      syncNodeEventLog(pSyncNode, eventLog);
    } else {
      char* eventLog = taosMemoryMalloc(debugStrLen + 64);
      snprintf(eventLog, debugStrLen, "become leader %s", debugStr);
      syncNodeEventLog(pSyncNode, eventLog);
      taosMemoryFree(eventLog);
    }
  } while (0);
M
Minghao Li 已提交
2484 2485 2486
}

void syncNodeCandidate2Leader(SSyncNode* pSyncNode) {
M
Minghao Li 已提交
2487 2488
  ASSERT(pSyncNode->state == TAOS_SYNC_STATE_CANDIDATE);
  ASSERT(voteGrantedMajority(pSyncNode->pVotesGranted));
2489
  syncNodeBecomeLeader(pSyncNode, "candidate to leader");
M
Minghao Li 已提交
2490

M
Minghao Li 已提交
2491 2492
  syncNodeLog2("==state change syncNodeCandidate2Leader==", pSyncNode);

M
Minghao Li 已提交
2493
  // Raft 3.6.2 Committing entries from previous terms
2494 2495
  syncNodeAppendNoop(pSyncNode);
  syncMaybeAdvanceCommitIndex(pSyncNode);
2496 2497

  if (pSyncNode->replicaNum > 1) {
M
Minghao Li 已提交
2498
    syncNodeReplicate(pSyncNode);
2499
  }
M
Minghao Li 已提交
2500 2501
}

M
Minghao Li 已提交
2502 2503
bool syncNodeIsMnode(SSyncNode* pSyncNode) { return (pSyncNode->vgId == 1); }

M
Minghao Li 已提交
2504 2505 2506 2507 2508 2509 2510
int32_t syncNodePeerStateInit(SSyncNode* pSyncNode) {
  for (int i = 0; i < TSDB_MAX_REPLICA; ++i) {
    pSyncNode->peerStates[i].lastSendIndex = SYNC_INDEX_INVALID;
    pSyncNode->peerStates[i].lastSendTime = 0;
  }

  return 0;
M
Minghao Li 已提交
2511 2512 2513
}

void syncNodeFollower2Candidate(SSyncNode* pSyncNode) {
M
Minghao Li 已提交
2514
  ASSERT(pSyncNode->state == TAOS_SYNC_STATE_FOLLOWER);
M
Minghao Li 已提交
2515
  pSyncNode->state = TAOS_SYNC_STATE_CANDIDATE;
M
Minghao Li 已提交
2516

M
Minghao Li 已提交
2517
  syncNodeEventLog(pSyncNode, "follower to candidate");
M
Minghao Li 已提交
2518 2519 2520
}

void syncNodeLeader2Follower(SSyncNode* pSyncNode) {
M
Minghao Li 已提交
2521
  ASSERT(pSyncNode->state == TAOS_SYNC_STATE_LEADER);
2522
  syncNodeBecomeFollower(pSyncNode, "leader to follower");
M
Minghao Li 已提交
2523

M
Minghao Li 已提交
2524
  syncNodeEventLog(pSyncNode, "leader to follower");
M
Minghao Li 已提交
2525 2526 2527
}

void syncNodeCandidate2Follower(SSyncNode* pSyncNode) {
M
Minghao Li 已提交
2528
  ASSERT(pSyncNode->state == TAOS_SYNC_STATE_CANDIDATE);
2529
  syncNodeBecomeFollower(pSyncNode, "candidate to follower");
M
Minghao Li 已提交
2530

M
Minghao Li 已提交
2531
  syncNodeEventLog(pSyncNode, "candidate to follower");
M
Minghao Li 已提交
2532 2533 2534
}

// raft vote --------------
M
Minghao Li 已提交
2535 2536 2537

// just called by syncNodeVoteForSelf
// need assert
M
Minghao Li 已提交
2538
void syncNodeVoteForTerm(SSyncNode* pSyncNode, SyncTerm term, SRaftId* pRaftId) {
M
Minghao Li 已提交
2539 2540
  ASSERT(term == pSyncNode->pRaftStore->currentTerm);
  ASSERT(!raftStoreHasVoted(pSyncNode->pRaftStore));
M
Minghao Li 已提交
2541 2542 2543 2544

  raftStoreVote(pSyncNode->pRaftStore, pRaftId);
}

M
Minghao Li 已提交
2545
// simulate get vote from outside
M
Minghao Li 已提交
2546 2547 2548
void syncNodeVoteForSelf(SSyncNode* pSyncNode) {
  syncNodeVoteForTerm(pSyncNode, pSyncNode->pRaftStore->currentTerm, &(pSyncNode->myRaftId));

M
Minghao Li 已提交
2549
  SyncRequestVoteReply* pMsg = syncRequestVoteReplyBuild(pSyncNode->vgId);
M
Minghao Li 已提交
2550 2551 2552 2553 2554 2555 2556 2557 2558 2559
  pMsg->srcId = pSyncNode->myRaftId;
  pMsg->destId = pSyncNode->myRaftId;
  pMsg->term = pSyncNode->pRaftStore->currentTerm;
  pMsg->voteGranted = true;

  voteGrantedVote(pSyncNode->pVotesGranted, pMsg);
  votesRespondAdd(pSyncNode->pVotesRespond, pMsg);
  syncRequestVoteReplyDestroy(pMsg);
}

M
Minghao Li 已提交
2560
// snapshot --------------
M
Minghao Li 已提交
2561 2562

// return if has a snapshot
M
Minghao Li 已提交
2563 2564
bool syncNodeHasSnapshot(SSyncNode* pSyncNode) {
  bool      ret = false;
2565
  SSnapshot snapshot = {.data = NULL, .lastApplyIndex = -1, .lastApplyTerm = 0, .lastConfigIndex = -1};
2566 2567
  if (pSyncNode->pFsm->FpGetSnapshotInfo != NULL) {
    pSyncNode->pFsm->FpGetSnapshotInfo(pSyncNode->pFsm, &snapshot);
M
Minghao Li 已提交
2568 2569 2570 2571 2572 2573 2574
    if (snapshot.lastApplyIndex >= SYNC_INDEX_BEGIN) {
      ret = true;
    }
  }
  return ret;
}

M
Minghao Li 已提交
2575 2576
// return max(logLastIndex, snapshotLastIndex)
// if no snapshot and log, return -1
2577
SyncIndex syncNodeGetLastIndex(const SSyncNode* pSyncNode) {
M
Minghao Li 已提交
2578
  SSnapshot snapshot = {.data = NULL, .lastApplyIndex = -1, .lastApplyTerm = 0, .lastConfigIndex = -1};
2579 2580
  if (pSyncNode->pFsm->FpGetSnapshotInfo != NULL) {
    pSyncNode->pFsm->FpGetSnapshotInfo(pSyncNode->pFsm, &snapshot);
M
Minghao Li 已提交
2581 2582 2583 2584 2585 2586 2587
  }
  SyncIndex logLastIndex = pSyncNode->pLogStore->syncLogLastIndex(pSyncNode->pLogStore);

  SyncIndex lastIndex = logLastIndex > snapshot.lastApplyIndex ? logLastIndex : snapshot.lastApplyIndex;
  return lastIndex;
}

M
Minghao Li 已提交
2588 2589
// return the last term of snapshot and log
// if error, return SYNC_TERM_INVALID (by syncLogLastTerm)
M
Minghao Li 已提交
2590 2591
SyncTerm syncNodeGetLastTerm(SSyncNode* pSyncNode) {
  SyncTerm lastTerm = 0;
M
Minghao Li 已提交
2592 2593
  if (syncNodeHasSnapshot(pSyncNode)) {
    // has snapshot
2594
    SSnapshot snapshot = {.data = NULL, .lastApplyIndex = -1, .lastApplyTerm = 0, .lastConfigIndex = -1};
2595 2596
    if (pSyncNode->pFsm->FpGetSnapshotInfo != NULL) {
      pSyncNode->pFsm->FpGetSnapshotInfo(pSyncNode->pFsm, &snapshot);
M
Minghao Li 已提交
2597 2598
    }

M
Minghao Li 已提交
2599 2600 2601
    SyncIndex logLastIndex = pSyncNode->pLogStore->syncLogLastIndex(pSyncNode->pLogStore);
    if (logLastIndex > snapshot.lastApplyIndex) {
      lastTerm = pSyncNode->pLogStore->syncLogLastTerm(pSyncNode->pLogStore);
M
Minghao Li 已提交
2602 2603 2604 2605
    } else {
      lastTerm = snapshot.lastApplyTerm;
    }

M
Minghao Li 已提交
2606
  } else {
M
Minghao Li 已提交
2607 2608
    // no snapshot
    lastTerm = pSyncNode->pLogStore->syncLogLastTerm(pSyncNode->pLogStore);
2609
  }
M
Minghao Li 已提交
2610

M
Minghao Li 已提交
2611 2612 2613 2614 2615 2616 2617
  return lastTerm;
}

// get last index and term along with snapshot
int32_t syncNodeGetLastIndexTerm(SSyncNode* pSyncNode, SyncIndex* pLastIndex, SyncTerm* pLastTerm) {
  *pLastIndex = syncNodeGetLastIndex(pSyncNode);
  *pLastTerm = syncNodeGetLastTerm(pSyncNode);
2618 2619
  return 0;
}
M
Minghao Li 已提交
2620

M
Minghao Li 已提交
2621
// return append-entries first try index
M
Minghao Li 已提交
2622 2623 2624 2625 2626
SyncIndex syncNodeSyncStartIndex(SSyncNode* pSyncNode) {
  SyncIndex syncStartIndex = syncNodeGetLastIndex(pSyncNode) + 1;
  return syncStartIndex;
}

M
Minghao Li 已提交
2627 2628
// if index > 0, return index - 1
// else, return -1
2629 2630 2631 2632 2633 2634 2635 2636 2637
SyncIndex syncNodeGetPreIndex(SSyncNode* pSyncNode, SyncIndex index) {
  SyncIndex preIndex = index - 1;
  if (preIndex < SYNC_INDEX_INVALID) {
    preIndex = SYNC_INDEX_INVALID;
  }

  return preIndex;
}

M
Minghao Li 已提交
2638 2639 2640 2641
// if index < 0, return SYNC_TERM_INVALID
// if index == 0, return 0
// if index > 0, return preTerm
// if error, return SYNC_TERM_INVALID
2642 2643 2644 2645 2646 2647 2648 2649 2650 2651 2652 2653 2654
SyncTerm syncNodeGetPreTerm(SSyncNode* pSyncNode, SyncIndex index) {
  if (index < SYNC_INDEX_BEGIN) {
    return SYNC_TERM_INVALID;
  }

  if (index == SYNC_INDEX_BEGIN) {
    return 0;
  }

  SyncTerm        preTerm = 0;
  SyncIndex       preIndex = index - 1;
  SSyncRaftEntry* pPreEntry = NULL;
  int32_t         code = pSyncNode->pLogStore->syncLogGetEntry(pSyncNode->pLogStore, preIndex, &pPreEntry);
M
Minghao Li 已提交
2655 2656 2657 2658 2659 2660

  SSnapshot snapshot = {.data = NULL,
                        .lastApplyIndex = SYNC_INDEX_INVALID,
                        .lastApplyTerm = SYNC_TERM_INVALID,
                        .lastConfigIndex = SYNC_INDEX_INVALID};

2661 2662 2663 2664 2665 2666
  if (code == 0) {
    ASSERT(pPreEntry != NULL);
    preTerm = pPreEntry->term;
    taosMemoryFree(pPreEntry);
    return preTerm;
  } else {
2667 2668 2669 2670
    if (pSyncNode->pFsm->FpGetSnapshotInfo != NULL) {
      pSyncNode->pFsm->FpGetSnapshotInfo(pSyncNode->pFsm, &snapshot);
      if (snapshot.lastApplyIndex == preIndex) {
        return snapshot.lastApplyTerm;
2671 2672 2673 2674
      }
    }
  }

2675 2676
  do {
    char logBuf[128];
S
Shengliang Guan 已提交
2677 2678
    snprintf(logBuf, sizeof(logBuf),
             "sync node get pre term error, index:%" PRId64 ", snap-index:%" PRId64 ", snap-term:%" PRIu64, index,
M
Minghao Li 已提交
2679
             snapshot.lastApplyIndex, snapshot.lastApplyTerm);
2680 2681 2682
    syncNodeErrorLog(pSyncNode, logBuf);
  } while (0);

2683 2684
  return SYNC_TERM_INVALID;
}
M
Minghao Li 已提交
2685 2686 2687 2688

// get pre index and term of "index"
int32_t syncNodeGetPreIndexTerm(SSyncNode* pSyncNode, SyncIndex index, SyncIndex* pPreIndex, SyncTerm* pPreTerm) {
  *pPreIndex = syncNodeGetPreIndex(pSyncNode, index);
M
Minghao Li 已提交
2689
  *pPreTerm = syncNodeGetPreTerm(pSyncNode, index);
M
Minghao Li 已提交
2690 2691 2692
  return 0;
}

M
Minghao Li 已提交
2693 2694 2695
// for debug --------------
void syncNodePrint(SSyncNode* pObj) {
  char* serialized = syncNode2Str(pObj);
S
Shengliang Guan 已提交
2696
  printf("syncNodePrint | len:%" PRIu64 " | %s \n", strlen(serialized), serialized);
M
Minghao Li 已提交
2697
  fflush(NULL);
wafwerar's avatar
wafwerar 已提交
2698
  taosMemoryFree(serialized);
M
Minghao Li 已提交
2699 2700 2701 2702
}

void syncNodePrint2(char* s, SSyncNode* pObj) {
  char* serialized = syncNode2Str(pObj);
S
Shengliang Guan 已提交
2703
  printf("syncNodePrint2 | len:%" PRIu64 " | %s | %s \n", strlen(serialized), s, serialized);
M
Minghao Li 已提交
2704
  fflush(NULL);
wafwerar's avatar
wafwerar 已提交
2705
  taosMemoryFree(serialized);
M
Minghao Li 已提交
2706 2707 2708 2709
}

void syncNodeLog(SSyncNode* pObj) {
  char* serialized = syncNode2Str(pObj);
S
Shengliang Guan 已提交
2710
  sTraceLong("syncNodeLog | len:%" PRIu64 " | %s", strlen(serialized), serialized);
wafwerar's avatar
wafwerar 已提交
2711
  taosMemoryFree(serialized);
M
Minghao Li 已提交
2712 2713 2714
}

void syncNodeLog2(char* s, SSyncNode* pObj) {
2715 2716
  if (gRaftDetailLog) {
    char* serialized = syncNode2Str(pObj);
S
Shengliang Guan 已提交
2717
    sTraceLong("syncNodeLog2 | len:%" PRIu64 " | %s | %s", strlen(serialized), s, serialized);
2718 2719
    taosMemoryFree(serialized);
  }
M
Minghao Li 已提交
2720 2721
}

M
Minghao Li 已提交
2722 2723
void syncNodeLog3(char* s, SSyncNode* pObj) {
  char* serialized = syncNode2Str(pObj);
S
Shengliang Guan 已提交
2724
  sTraceLong("syncNodeLog3 | len:%" PRIu64 " | %s | %s", strlen(serialized), s, serialized);
M
Minghao Li 已提交
2725 2726 2727
  taosMemoryFree(serialized);
}

M
Minghao Li 已提交
2728
// ------ local funciton ---------
M
Minghao Li 已提交
2729
// enqueue message ----
M
Minghao Li 已提交
2730 2731
static void syncNodeEqPingTimer(void* param, void* tmrId) {
  SSyncNode* pSyncNode = (SSyncNode*)param;
M
Minghao Li 已提交
2732
  if (atomic_load_64(&pSyncNode->pingTimerLogicClockUser) <= atomic_load_64(&pSyncNode->pingTimerLogicClock)) {
M
Minghao Li 已提交
2733
    SyncTimeout* pSyncMsg = syncTimeoutBuild2(SYNC_TIMEOUT_PING, atomic_load_64(&pSyncNode->pingTimerLogicClock),
M
Minghao Li 已提交
2734
                                              pSyncNode->pingTimerMS, pSyncNode->vgId, pSyncNode);
M
Minghao Li 已提交
2735 2736
    SRpcMsg      rpcMsg;
    syncTimeout2RpcMsg(pSyncMsg, &rpcMsg);
M
Minghao Li 已提交
2737
    syncRpcMsgLog2((char*)"==syncNodeEqPingTimer==", &rpcMsg);
M
Minghao Li 已提交
2738
    if (pSyncNode->FpEqMsg != NULL) {
2739 2740
      int32_t code = pSyncNode->FpEqMsg(pSyncNode->msgcb, &rpcMsg);
      if (code != 0) {
S
Shengliang Guan 已提交
2741
        sError("vgId:%d, sync enqueue ping msg error, code:%d", pSyncNode->vgId, code);
2742 2743 2744 2745
        rpcFreeCont(rpcMsg.pCont);
        syncTimeoutDestroy(pSyncMsg);
        return;
      }
M
Minghao Li 已提交
2746 2747 2748
    } else {
      sTrace("syncNodeEqPingTimer pSyncNode->FpEqMsg is NULL");
    }
M
Minghao Li 已提交
2749 2750
    syncTimeoutDestroy(pSyncMsg);

2751 2752 2753 2754 2755 2756 2757
    if (syncEnvIsStart()) {
      taosTmrReset(syncNodeEqPingTimer, pSyncNode->pingTimerMS, pSyncNode, gSyncEnv->pTimerManager,
                   &pSyncNode->pPingTimer);
    } else {
      sError("sync env is stop, syncNodeEqPingTimer");
    }

M
Minghao Li 已提交
2758
  } else {
S
Shengliang Guan 已提交
2759
    sTrace("==syncNodeEqPingTimer== pingTimerLogicClock:%" PRIu64 ", pingTimerLogicClockUser:%" PRIu64,
M
Minghao Li 已提交
2760
           pSyncNode->pingTimerLogicClock, pSyncNode->pingTimerLogicClockUser);
M
Minghao Li 已提交
2761 2762 2763 2764 2765 2766 2767
  }
}

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),
M
Minghao Li 已提交
2768
                                              pSyncNode->electTimerMS, pSyncNode->vgId, pSyncNode);
M
Minghao Li 已提交
2769
    SRpcMsg      rpcMsg;
M
Minghao Li 已提交
2770
    syncTimeout2RpcMsg(pSyncMsg, &rpcMsg);
M
Minghao Li 已提交
2771
    syncRpcMsgLog2((char*)"==syncNodeEqElectTimer==", &rpcMsg);
M
Minghao Li 已提交
2772
    if (pSyncNode->FpEqMsg != NULL) {
2773 2774
      int32_t code = pSyncNode->FpEqMsg(pSyncNode->msgcb, &rpcMsg);
      if (code != 0) {
S
Shengliang Guan 已提交
2775
        sError("vgId:%d, sync enqueue elect msg error, code:%d", pSyncNode->vgId, code);
2776 2777 2778 2779
        rpcFreeCont(rpcMsg.pCont);
        syncTimeoutDestroy(pSyncMsg);
        return;
      }
M
Minghao Li 已提交
2780
    } else {
2781
      sTrace("syncNodeEqElectTimer FpEqMsg is NULL");
M
Minghao Li 已提交
2782
    }
M
Minghao Li 已提交
2783 2784
    syncTimeoutDestroy(pSyncMsg);

M
Minghao Li 已提交
2785
    // reset timer ms
2786
    if (syncEnvIsStart()) {
2787 2788 2789 2790
      pSyncNode->electTimerMS = syncUtilElectRandomMS(pSyncNode->electBaseLine, 2 * pSyncNode->electBaseLine);
      taosTmrReset(syncNodeEqElectTimer, pSyncNode->electTimerMS, pSyncNode, gSyncEnv->pTimerManager,
                   &pSyncNode->pElectTimer);
    } else {
2791
      sError("sync env is stop, syncNodeEqElectTimer");
2792
    }
M
Minghao Li 已提交
2793
  } else {
S
Shengliang Guan 已提交
2794
    sTrace("==syncNodeEqElectTimer== electTimerLogicClock:%" PRIu64 ", electTimerLogicClockUser:%" PRIu64,
M
Minghao Li 已提交
2795
           pSyncNode->electTimerLogicClock, pSyncNode->electTimerLogicClockUser);
M
Minghao Li 已提交
2796 2797 2798
  }
}

M
Minghao Li 已提交
2799 2800
static void syncNodeEqHeartbeatTimer(void* param, void* tmrId) {
  SSyncNode* pSyncNode = (SSyncNode*)param;
2801 2802 2803

  syncNodeEventLog(pSyncNode, "eq hb timer");

2804 2805 2806 2807 2808 2809 2810 2811 2812 2813 2814 2815
  if (pSyncNode->replicaNum > 1) {
    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->vgId, pSyncNode);
      SRpcMsg rpcMsg;
      syncTimeout2RpcMsg(pSyncMsg, &rpcMsg);
      syncRpcMsgLog2((char*)"==syncNodeEqHeartbeatTimer==", &rpcMsg);
      if (pSyncNode->FpEqMsg != NULL) {
        int32_t code = pSyncNode->FpEqMsg(pSyncNode->msgcb, &rpcMsg);
        if (code != 0) {
S
Shengliang Guan 已提交
2816
          sError("vgId:%d, sync enqueue timer msg error, code:%d", pSyncNode->vgId, code);
2817 2818 2819 2820 2821
          rpcFreeCont(rpcMsg.pCont);
          syncTimeoutDestroy(pSyncMsg);
          return;
        }
      } else {
2822
        sError("vgId:%d, enqueue msg cb ptr (i.e. FpEqMsg) not set.", pSyncNode->vgId);
2823
      }
2824
      syncTimeoutDestroy(pSyncMsg);
M
Minghao Li 已提交
2825

2826 2827 2828 2829 2830 2831
      if (syncEnvIsStart()) {
        taosTmrReset(syncNodeEqHeartbeatTimer, pSyncNode->heartbeatTimerMS, pSyncNode, gSyncEnv->pTimerManager,
                     &pSyncNode->pHeartbeatTimer);
      } else {
        sError("sync env is stop, syncNodeEqHeartbeatTimer");
      }
2832
    } else {
2833 2834 2835
      sTrace("==syncNodeEqHeartbeatTimer== heartbeatTimerLogicClock:%" PRIu64 ", heartbeatTimerLogicClockUser:%" PRIu64
             "",
             pSyncNode->heartbeatTimerLogicClock, pSyncNode->heartbeatTimerLogicClockUser);
2836
    }
M
Minghao Li 已提交
2837 2838 2839
  }
}

2840 2841 2842 2843 2844
static void syncNodeEqPeerHeartbeatTimer(void* param, void* tmrId) {
  SSyncHbTimerData* pData = (SSyncHbTimerData*)param;
  SSyncNode*        pSyncNode = pData->pSyncNode;
  SSyncTimer*       pSyncTimer = pData->pTimer;

M
Minghao Li 已提交
2845 2846 2847 2848
  if (pSyncNode->state != TAOS_SYNC_STATE_LEADER) {
    return;
  }

2849 2850 2851 2852 2853 2854 2855 2856 2857 2858 2859 2860
  syncNodeEventLog(pSyncNode, "eq peer hb timer");

  int64_t timerLogicClock = atomic_load_64(&pSyncTimer->logicClock);
  int64_t msgLogicClock = atomic_load_64(&pData->logicClock);

  if (pSyncNode->replicaNum > 1) {
    if (timerLogicClock == msgLogicClock) {
      SyncHeartbeat* pSyncMsg = syncHeartbeatBuild(pSyncNode->vgId);
      pSyncMsg->srcId = pSyncNode->myRaftId;
      pSyncMsg->destId = pData->destId;
      pSyncMsg->term = pSyncNode->pRaftStore->currentTerm;
      pSyncMsg->commitIndex = pSyncNode->commitIndex;
M
Minghao Li 已提交
2861
      pSyncMsg->minMatchIndex = syncMinMatchIndex(pSyncNode);
2862 2863 2864 2865 2866 2867 2868 2869 2870 2871 2872 2873 2874 2875 2876 2877 2878 2879 2880 2881 2882
      pSyncMsg->privateTerm = 0;

      SRpcMsg rpcMsg;
      syncHeartbeat2RpcMsg(pSyncMsg, &rpcMsg);

// eq msg
#if 0
      if (pSyncNode->FpEqCtrlMsg != NULL) {
        int32_t code = pSyncNode->FpEqCtrlMsg(pSyncNode->msgcb, &rpcMsg);
        if (code != 0) {
          sError("vgId:%d, sync ctrl enqueue timer msg error, code:%d", pSyncNode->vgId, code);
          rpcFreeCont(rpcMsg.pCont);
          syncHeartbeatDestroy(pSyncMsg);
          return;
        }
      } else {
        sError("vgId:%d, enqueue ctrl msg cb ptr (i.e. FpEqMsg) not set.", pSyncNode->vgId);
      }
#endif

      // send msg
M
Minghao Li 已提交
2883
      syncNodeSendHeartbeat(pSyncNode, &(pSyncMsg->destId), pSyncMsg);
2884 2885 2886 2887 2888 2889 2890 2891 2892 2893 2894 2895 2896 2897 2898 2899 2900

      syncHeartbeatDestroy(pSyncMsg);

      if (syncEnvIsStart()) {
        taosTmrReset(syncNodeEqPeerHeartbeatTimer, pSyncTimer->timerMS, pData, gSyncEnv->pTimerManager,
                     &pSyncTimer->pTimer);
      } else {
        sError("sync env is stop, syncNodeEqHeartbeatTimer");
      }

    } else {
      sTrace("==syncNodeEqPeerHeartbeatTimer== timerLogicClock:%" PRIu64 ", msgLogicClock:%" PRIu64 "", timerLogicClock,
             msgLogicClock);
    }
  }
}

M
Minghao Li 已提交
2901 2902
static int32_t syncNodeEqNoop(SSyncNode* ths) {
  int32_t ret = 0;
M
Minghao Li 已提交
2903
  ASSERT(ths->state == TAOS_SYNC_STATE_LEADER);
M
Minghao Li 已提交
2904

2905
  SyncIndex       index = ths->pLogStore->syncLogWriteIndex(ths->pLogStore);
M
Minghao Li 已提交
2906
  SyncTerm        term = ths->pRaftStore->currentTerm;
M
Minghao Li 已提交
2907
  SSyncRaftEntry* pEntry = syncEntryBuildNoop(term, index, ths->vgId);
M
Minghao Li 已提交
2908
  ASSERT(pEntry != NULL);
M
Minghao Li 已提交
2909 2910 2911 2912

  uint32_t           entryLen;
  char*              serialized = syncEntrySerialize(pEntry, &entryLen);
  SyncClientRequest* pSyncMsg = syncClientRequestBuild(entryLen);
M
Minghao Li 已提交
2913
  ASSERT(pSyncMsg->dataLen == entryLen);
M
Minghao Li 已提交
2914 2915
  memcpy(pSyncMsg->data, serialized, entryLen);

S
Shengliang Guan 已提交
2916
  SRpcMsg rpcMsg = {0};
M
Minghao Li 已提交
2917
  syncClientRequest2RpcMsg(pSyncMsg, &rpcMsg);
M
Minghao Li 已提交
2918
  if (ths->FpEqMsg != NULL) {
S
Shengliang Guan 已提交
2919
    ths->FpEqMsg(ths->msgcb, &rpcMsg);
M
Minghao Li 已提交
2920 2921 2922
  } else {
    sTrace("syncNodeEqNoop pSyncNode->FpEqMsg is NULL");
  }
M
Minghao Li 已提交
2923

wafwerar's avatar
wafwerar 已提交
2924
  taosMemoryFree(serialized);
M
Minghao Li 已提交
2925 2926 2927 2928 2929
  syncClientRequestDestroy(pSyncMsg);

  return ret;
}

2930 2931 2932 2933 2934 2935 2936 2937 2938 2939 2940 2941 2942 2943
static void deleteCacheEntry(const void* key, size_t keyLen, void* value) { taosMemoryFree(value); }

static int32_t syncCacheEntry(SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry, LRUHandle** h) {
  int       code = 0;
  int       entryLen = sizeof(*pEntry) + pEntry->dataLen;
  LRUStatus status = taosLRUCacheInsert(pLogStore->pCache, &pEntry->index, sizeof(pEntry->index), pEntry, entryLen,
                                        deleteCacheEntry, h, TAOS_LRU_PRIORITY_LOW);
  if (status != TAOS_LRU_STATUS_OK) {
    code = -1;
  }

  return code;
}

M
Minghao Li 已提交
2944 2945 2946
static int32_t syncNodeAppendNoop(SSyncNode* ths) {
  int32_t ret = 0;

2947
  SyncIndex       index = ths->pLogStore->syncLogWriteIndex(ths->pLogStore);
M
Minghao Li 已提交
2948
  SyncTerm        term = ths->pRaftStore->currentTerm;
M
Minghao Li 已提交
2949
  SSyncRaftEntry* pEntry = syncEntryBuildNoop(term, index, ths->vgId);
M
Minghao Li 已提交
2950
  ASSERT(pEntry != NULL);
M
Minghao Li 已提交
2951

2952 2953 2954
  LRUHandle* h = NULL;
  syncCacheEntry(ths->pLogStore, pEntry, &h);

M
Minghao Li 已提交
2955
  if (ths->state == TAOS_SYNC_STATE_LEADER) {
2956
    int32_t code = ths->pLogStore->syncLogAppendEntry(ths->pLogStore, pEntry);
2957
    if (code != 0) {
M
Minghao Li 已提交
2958
      syncNodeErrorLog(ths, "append noop error");
2959 2960
      return -1;
    }
M
Minghao Li 已提交
2961 2962
  }

2963 2964 2965 2966 2967 2968
  if (h) {
    taosLRUCacheRelease(ths->pLogStore->pCache, h, false);
  } else {
    syncEntryDestory(pEntry);
  }

M
Minghao Li 已提交
2969 2970 2971
  return ret;
}

M
Minghao Li 已提交
2972
// on message ----
M
Minghao Li 已提交
2973 2974
int32_t syncNodeOnPingCb(SSyncNode* ths, SyncPing* pMsg) {
  // log state
2975
  char logBuf[1024] = {0};
M
Minghao Li 已提交
2976
  snprintf(logBuf, sizeof(logBuf),
2977 2978
           "==syncNodeOnPingCb== vgId:%d, state: %d, %s, term:%" PRIu64 " electTimerLogicClock:%" PRIu64
           ", "
S
Shengliang Guan 已提交
2979
           "electTimerLogicClockUser:%" PRIu64 ", electTimerMS:%d",
M
Minghao Li 已提交
2980 2981 2982
           ths->vgId, ths->state, syncUtilState2String(ths->state), ths->pRaftStore->currentTerm,
           ths->electTimerLogicClock, ths->electTimerLogicClockUser, ths->electTimerMS);

M
Minghao Li 已提交
2983
  int32_t ret = 0;
M
Minghao Li 已提交
2984 2985
  syncPingLog2(logBuf, pMsg);
  SyncPingReply* pMsgReply = syncPingReplyBuild3(&ths->myRaftId, &pMsg->srcId, ths->vgId);
M
Minghao Li 已提交
2986 2987
  SRpcMsg        rpcMsg;
  syncPingReply2RpcMsg(pMsgReply, &rpcMsg);
M
Minghao Li 已提交
2988 2989 2990 2991 2992 2993 2994 2995

  /*
    // htonl
    SMsgHead* pHead = rpcMsg.pCont;
    pHead->contLen = htonl(pHead->contLen);
    pHead->vgId = htonl(pHead->vgId);
  */

M
Minghao Li 已提交
2996 2997 2998 2999 3000
  syncNodeSendMsgById(&pMsgReply->destId, ths, &rpcMsg);

  return ret;
}

M
Minghao Li 已提交
3001
int32_t syncNodeOnPingReplyCb(SSyncNode* ths, SyncPingReply* pMsg) {
M
Minghao Li 已提交
3002 3003 3004 3005
  int32_t ret = 0;
  syncPingReplyLog2("==syncNodeOnPingReplyCb==", pMsg);
  return ret;
}
M
Minghao Li 已提交
3006

3007 3008 3009 3010 3011 3012 3013 3014 3015 3016 3017 3018
int32_t syncNodeOnHeartbeat(SSyncNode* ths, SyncHeartbeat* pMsg) {
  syncLogRecvHeartbeat(ths, pMsg, "");

  SyncHeartbeatReply* pMsgReply = syncHeartbeatReplyBuild(ths->vgId);
  pMsgReply->destId = pMsg->srcId;
  pMsgReply->srcId = ths->myRaftId;
  pMsgReply->term = ths->pRaftStore->currentTerm;
  pMsgReply->privateTerm = 8864;  // magic number

  SRpcMsg rpcMsg;
  syncHeartbeatReply2RpcMsg(pMsgReply, &rpcMsg);

M
Minghao Li 已提交
3019
#if 1
3020
  if (pMsg->term >= ths->pRaftStore->currentTerm && ths->state != TAOS_SYNC_STATE_FOLLOWER) {
M
Minghao Li 已提交
3021
    syncNodeStepDown(ths, pMsg->term);
3022
  }
M
Minghao Li 已提交
3023
#endif
3024

M
Minghao Li 已提交
3025
  if (pMsg->term == ths->pRaftStore->currentTerm && ths->state != TAOS_SYNC_STATE_LEADER) {
3026
    syncNodeResetElectTimer(ths);
M
Minghao Li 已提交
3027
    ths->minMatchIndex = pMsg->minMatchIndex;
3028 3029 3030 3031 3032 3033 3034 3035 3036 3037 3038 3039 3040 3041 3042 3043 3044 3045 3046 3047 3048 3049 3050 3051 3052 3053 3054 3055 3056 3057

#if 0
    if (ths->state == TAOS_SYNC_STATE_FOLLOWER) {
      syncNodeFollowerCommit(ths, pMsg->commitIndex);
    }
#endif
  }

  /*
    // htonl
    SMsgHead* pHead = rpcMsg.pCont;
    pHead->contLen = htonl(pHead->contLen);
    pHead->vgId = htonl(pHead->vgId);
  */

  // reply
  syncNodeSendMsgById(&pMsgReply->destId, ths, &rpcMsg);

  return 0;
}

int32_t syncNodeOnHeartbeatReply(SSyncNode* ths, SyncHeartbeatReply* pMsg) {
  syncLogRecvHeartbeatReply(ths, pMsg, "");

  // update last reply time, make decision whether the other node is alive or not
  syncIndexMgrSetRecvTime(ths->pMatchIndex, &(pMsg->destId), pMsg->startTime);

  return 0;
}

M
Minghao Li 已提交
3058 3059 3060 3061 3062 3063 3064 3065 3066 3067
// TLA+ Spec
// ClientRequest(i, v) ==
//     /\ state[i] = Leader
//     /\ LET entry == [term  |-> currentTerm[i],
//                      value |-> v]
//            newLog == Append(log[i], entry)
//        IN  log' = [log EXCEPT ![i] = newLog]
//     /\ UNCHANGED <<messages, serverVars, candidateVars,
//                    leaderVars, commitIndex>>
//
M
Minghao Li 已提交
3068

M
Minghao Li 已提交
3069
int32_t syncNodeOnClientRequest(SSyncNode* ths, SyncClientRequest* pMsg, SyncIndex* pRetIndex) {
3070 3071
  syncNodeEventLog(ths, "on client request");

M
Minghao Li 已提交
3072
  int32_t ret = 0;
3073
  int32_t code = 0;
M
Minghao Li 已提交
3074

M
Minghao Li 已提交
3075
  SyncIndex       index = ths->pLogStore->syncLogWriteIndex(ths->pLogStore);
M
Minghao Li 已提交
3076 3077
  SyncTerm        term = ths->pRaftStore->currentTerm;
  SSyncRaftEntry* pEntry = syncEntryBuild2((SyncClientRequest*)pMsg, term, index);
M
Minghao Li 已提交
3078
  ASSERT(pEntry != NULL);
M
Minghao Li 已提交
3079

3080 3081 3082
  LRUHandle* h = NULL;
  syncCacheEntry(ths->pLogStore, pEntry, &h);

M
Minghao Li 已提交
3083
  if (ths->state == TAOS_SYNC_STATE_LEADER) {
3084 3085 3086 3087
    // append entry
    code = ths->pLogStore->syncLogAppendEntry(ths->pLogStore, pEntry);
    if (code != 0) {
      // del resp mgr, call FpCommitCb
M
Minghao Li 已提交
3088
      ASSERT(0);
3089 3090
      return -1;
    }
M
Minghao Li 已提交
3091

3092 3093
    // if mulit replica, start replicate right now
    if (ths->replicaNum > 1) {
M
Minghao Li 已提交
3094
      syncNodeReplicate(ths);
3095
    }
3096

3097 3098 3099 3100
    // if only myself, maybe commit right now
    if (ths->replicaNum == 1) {
      syncMaybeAdvanceCommitIndex(ths);
    }
M
Minghao Li 已提交
3101 3102
  }

3103 3104 3105 3106 3107 3108 3109 3110
  if (pRetIndex != NULL) {
    if (ret == 0 && pEntry != NULL) {
      *pRetIndex = pEntry->index;
    } else {
      *pRetIndex = SYNC_INDEX_INVALID;
    }
  }

3111 3112 3113 3114 3115 3116
  if (h) {
    taosLRUCacheRelease(ths->pLogStore->pCache, h, false);
  } else {
    syncEntryDestory(pEntry);
  }

M
Minghao Li 已提交
3117
  return ret;
3118
}
M
Minghao Li 已提交
3119

S
Shengliang Guan 已提交
3120 3121 3122
const char* syncStr(ESyncState state) {
  switch (state) {
    case TAOS_SYNC_STATE_FOLLOWER:
3123
      return "follower";
S
Shengliang Guan 已提交
3124
    case TAOS_SYNC_STATE_CANDIDATE:
3125
      return "candidate";
S
Shengliang Guan 已提交
3126
    case TAOS_SYNC_STATE_LEADER:
3127
      return "leader";
S
Shengliang Guan 已提交
3128
    default:
3129
      return "error";
S
Shengliang Guan 已提交
3130
  }
M
Minghao Li 已提交
3131
}
3132

3133
int32_t syncDoLeaderTransfer(SSyncNode* ths, SRpcMsg* pRpcMsg, SSyncRaftEntry* pEntry) {
3134 3135 3136 3137
  if (ths->state != TAOS_SYNC_STATE_FOLLOWER) {
    syncNodeEventLog(ths, "I am not follower, can not do leader transfer");
    return 0;
  }
3138 3139 3140 3141 3142 3143

  if (!ths->restoreFinish) {
    syncNodeEventLog(ths, "restore not finish, can not do leader transfer");
    return 0;
  }

3144 3145
  if (pEntry->term < ths->pRaftStore->currentTerm) {
    char logBuf[128];
S
Shengliang Guan 已提交
3146
    snprintf(logBuf, sizeof(logBuf), "little term:%" PRIu64 ", can not do leader transfer", pEntry->term);
3147 3148 3149 3150 3151 3152
    syncNodeEventLog(ths, logBuf);
    return 0;
  }

  if (pEntry->index < syncNodeGetLastIndex(ths)) {
    char logBuf[128];
S
Shengliang Guan 已提交
3153
    snprintf(logBuf, sizeof(logBuf), "little index:%" PRId64 ", can not do leader transfer", pEntry->index);
3154
    syncNodeEventLog(ths, logBuf);
3155 3156 3157
    return 0;
  }

3158 3159 3160 3161 3162 3163 3164
  /*
    if (ths->vgId > 1) {
      syncNodeEventLog(ths, "I am vnode, can not do leader transfer");
      return 0;
    }
  */

M
Minghao Li 已提交
3165 3166
  SyncLeaderTransfer* pSyncLeaderTransfer = syncLeaderTransferFromRpcMsg2(pRpcMsg);

3167 3168
  do {
    char logBuf[128];
S
Shengliang Guan 已提交
3169
    snprintf(logBuf, sizeof(logBuf), "do leader transfer, index:%" PRId64, pEntry->index);
3170 3171
    syncNodeEventLog(ths, logBuf);
  } while (0);
M
Minghao Li 已提交
3172

M
Minghao Li 已提交
3173 3174 3175
  bool sameId = syncUtilSameId(&(pSyncLeaderTransfer->newLeaderId), &(ths->myRaftId));
  bool sameNodeInfo = strcmp(pSyncLeaderTransfer->newNodeInfo.nodeFqdn, ths->myNodeInfo.nodeFqdn) == 0 &&
                      pSyncLeaderTransfer->newNodeInfo.nodePort == ths->myNodeInfo.nodePort;
M
Minghao Li 已提交
3176

M
Minghao Li 已提交
3177 3178
  bool same = sameId || sameNodeInfo;
  if (same) {
M
Minghao Li 已提交
3179 3180 3181 3182
    // reset elect timer now!
    int32_t electMS = 1;
    int32_t ret = syncNodeRestartElectTimer(ths, electMS);
    ASSERT(ret == 0);
M
Minghao Li 已提交
3183 3184

    char eventLog[256];
S
Shengliang Guan 已提交
3185
    snprintf(eventLog, sizeof(eventLog), "maybe leader transfer to %s:%d %" PRIu64,
M
Minghao Li 已提交
3186 3187 3188
             pSyncLeaderTransfer->newNodeInfo.nodeFqdn, pSyncLeaderTransfer->newNodeInfo.nodePort,
             pSyncLeaderTransfer->newLeaderId.addr);
    syncNodeEventLog(ths, eventLog);
3189 3190
  }

M
Minghao Li 已提交
3191
  if (ths->pFsm->FpLeaderTransferCb != NULL) {
3192
    SFsmCbMeta cbMeta = {0};
M
Minghao Li 已提交
3193 3194 3195 3196
    cbMeta.code = 0;
    cbMeta.currentTerm = ths->pRaftStore->currentTerm;
    cbMeta.flag = 0;
    cbMeta.index = pEntry->index;
3197
    cbMeta.lastConfigIndex = syncNodeGetSnapshotConfigIndex(ths, cbMeta.index);
M
Minghao Li 已提交
3198 3199 3200 3201 3202
    cbMeta.isWeak = pEntry->isWeak;
    cbMeta.seqNum = pEntry->seqNum;
    cbMeta.state = ths->state;
    cbMeta.term = pEntry->term;
    ths->pFsm->FpLeaderTransferCb(ths->pFsm, pRpcMsg, cbMeta);
3203 3204
  }

M
Minghao Li 已提交
3205
  syncLeaderTransferDestroy(pSyncLeaderTransfer);
3206 3207 3208
  return 0;
}

3209 3210 3211 3212 3213 3214 3215 3216 3217 3218 3219 3220 3221 3222 3223
int32_t syncNodeUpdateNewConfigIndex(SSyncNode* ths, SSyncCfg* pNewCfg) {
  for (int i = 0; i < pNewCfg->replicaNum; ++i) {
    SRaftId raftId;
    raftId.addr = syncUtilAddr2U64((pNewCfg->nodeInfo)[i].nodeFqdn, (pNewCfg->nodeInfo)[i].nodePort);
    raftId.vgId = ths->vgId;

    if (syncUtilSameId(&(ths->myRaftId), &raftId)) {
      pNewCfg->myIndex = i;
      return 0;
    }
  }

  return -1;
}

M
Minghao Li 已提交
3224 3225 3226 3227 3228 3229 3230 3231 3232 3233 3234 3235 3236 3237 3238 3239 3240 3241 3242 3243 3244 3245
static int32_t syncNodeConfigChangeFinish(SSyncNode* ths, SRpcMsg* pRpcMsg, SSyncRaftEntry* pEntry) {
  SyncReconfigFinish* pFinish = syncReconfigFinishFromRpcMsg2(pRpcMsg);
  ASSERT(pFinish);

  if (ths->pFsm->FpReConfigCb != NULL) {
    SReConfigCbMeta cbMeta = {0};
    cbMeta.code = 0;
    cbMeta.index = pEntry->index;
    cbMeta.term = pEntry->term;
    cbMeta.seqNum = pEntry->seqNum;
    cbMeta.lastConfigIndex = syncNodeGetSnapshotConfigIndex(ths, pEntry->index);
    cbMeta.state = ths->state;
    cbMeta.currentTerm = ths->pRaftStore->currentTerm;
    cbMeta.isWeak = pEntry->isWeak;
    cbMeta.flag = 0;

    cbMeta.oldCfg = pFinish->oldCfg;
    cbMeta.newCfg = pFinish->newCfg;
    cbMeta.newCfgIndex = pFinish->newCfgIndex;
    cbMeta.newCfgTerm = pFinish->newCfgTerm;
    cbMeta.newCfgSeqNum = pFinish->newCfgSeqNum;

S
Shengliang Guan 已提交
3246
    ths->pFsm->FpReConfigCb(ths->pFsm, pRpcMsg, &cbMeta);
M
Minghao Li 已提交
3247 3248
  }

3249
  // clear changing
M
Minghao Li 已提交
3250 3251 3252 3253 3254
  ths->changing = false;

  char  tmpbuf[512];
  char* oldStr = syncCfg2SimpleStr(&(pFinish->oldCfg));
  char* newStr = syncCfg2SimpleStr(&(pFinish->newCfg));
S
Shengliang Guan 已提交
3255
  snprintf(tmpbuf, sizeof(tmpbuf), "config change finish from %d to %d, index:%" PRId64 ", %s  -->  %s",
M
Minghao Li 已提交
3256 3257 3258 3259 3260 3261 3262 3263 3264 3265 3266 3267
           pFinish->oldCfg.replicaNum, pFinish->newCfg.replicaNum, pFinish->newCfgIndex, oldStr, newStr);
  taosMemoryFree(oldStr);
  taosMemoryFree(newStr);
  syncNodeEventLog(ths, tmpbuf);

  syncReconfigFinishDestroy(pFinish);

  return 0;
}

static int32_t syncNodeConfigChange(SSyncNode* ths, SRpcMsg* pRpcMsg, SSyncRaftEntry* pEntry,
                                    SyncReconfigFinish* pFinish) {
3268 3269 3270
  // set changing
  ths->changing = true;

M
Minghao Li 已提交
3271
  // old config
3272 3273
  SSyncCfg oldSyncCfg = ths->pRaftCfg->cfg;

M
Minghao Li 已提交
3274
  // new config
3275 3276 3277 3278 3279
  SSyncCfg newSyncCfg;
  int32_t  ret = syncCfgFromStr(pRpcMsg->pCont, &newSyncCfg);
  ASSERT(ret == 0);

  // update new config myIndex
3280 3281
  syncNodeUpdateNewConfigIndex(ths, &newSyncCfg);

M
Minghao Li 已提交
3282 3283
  // do config change
  syncNodeDoConfigChange(ths, &newSyncCfg, pEntry->index);
3284

M
Minghao Li 已提交
3285 3286 3287 3288 3289 3290
  // set pFinish
  pFinish->oldCfg = oldSyncCfg;
  pFinish->newCfg = newSyncCfg;
  pFinish->newCfgIndex = pEntry->index;
  pFinish->newCfgTerm = pEntry->term;
  pFinish->newCfgSeqNum = pEntry->seqNum;
3291

M
Minghao Li 已提交
3292 3293
  return 0;
}
3294

M
Minghao Li 已提交
3295 3296 3297 3298 3299 3300 3301 3302 3303
static int32_t syncNodeProposeConfigChangeFinish(SSyncNode* ths, SyncReconfigFinish* pFinish) {
  SRpcMsg rpcMsg;
  syncReconfigFinish2RpcMsg(pFinish, &rpcMsg);

  int32_t code = syncNodePropose(ths, &rpcMsg, false);
  if (code != 0) {
    sError("syncNodeProposeConfigChangeFinish error");
    ths->changing = false;
  }
3304 3305 3306
  return 0;
}

3307 3308 3309 3310
bool syncNodeIsOptimizedOneReplica(SSyncNode* ths, SRpcMsg* pMsg) {
  return (ths->replicaNum == 1 && syncUtilUserCommit(pMsg->msgType) && ths->vgId != 1);
}

M
Minghao Li 已提交
3311
int32_t syncNodeDoCommit(SSyncNode* ths, SyncIndex beginIndex, SyncIndex endIndex, uint64_t flag) {
3312 3313 3314 3315 3316 3317 3318 3319 3320 3321 3322 3323 3324 3325 3326 3327 3328
  if (beginIndex > endIndex) {
    return 0;
  }

  // advance commit index to sanpshot first
  SSnapshot snapshot = {0};
  ths->pFsm->FpGetSnapshotInfo(ths->pFsm, &snapshot);
  if (snapshot.lastApplyIndex >= 0 && snapshot.lastApplyIndex >= beginIndex) {
    char eventLog[128];
    snprintf(eventLog, sizeof(eventLog), "commit by snapshot from index:%" PRId64 " to index:%" PRId64, beginIndex,
             snapshot.lastApplyIndex);
    syncNodeEventLog(ths, eventLog);

    // update begin index
    beginIndex = snapshot.lastApplyIndex + 1;
  }

3329 3330
  int32_t    code = 0;
  ESyncState state = flag;
M
Minghao Li 已提交
3331 3332

  char eventLog[128];
3333
  snprintf(eventLog, sizeof(eventLog), "commit by wal from index:%" PRId64 " to index:%" PRId64, beginIndex, endIndex);
M
Minghao Li 已提交
3334
  syncNodeEventLog(ths, eventLog);
3335 3336 3337 3338 3339 3340

  // execute fsm
  if (ths->pFsm != NULL) {
    for (SyncIndex i = beginIndex; i <= endIndex; ++i) {
      if (i != SYNC_INDEX_INVALID) {
        SSyncRaftEntry* pEntry;
3341 3342 3343 3344 3345 3346 3347 3348 3349
        SLRUCache*      pCache = ths->pLogStore->pCache;
        LRUHandle*      h = taosLRUCacheLookup(pCache, &i, sizeof(i));
        if (h) {
          pEntry = (SSyncRaftEntry*)taosLRUCacheValue(pCache, h);
        } else {
          code = ths->pLogStore->syncLogGetEntry(ths->pLogStore, i, &pEntry);
          ASSERT(code == 0);
          ASSERT(pEntry != NULL);
        }
3350 3351 3352 3353

        SRpcMsg rpcMsg;
        syncEntry2OriginalRpc(pEntry, &rpcMsg);

3354
        // user commit
3355 3356
        if ((ths->pFsm->FpCommitCb != NULL) && syncUtilUserCommit(pEntry->originalRpcType)) {
          bool internalExecute = true;
S
Shengliang Guan 已提交
3357
          if ((ths->replicaNum == 1) && ths->restoreFinish && ths->vgId != 1) {
3358 3359 3360 3361 3362
            internalExecute = false;
          }

          do {
            char logBuf[128];
S
Shengliang Guan 已提交
3363
            snprintf(logBuf, sizeof(logBuf), "commit index:%" PRId64 ", internal:%d", i, internalExecute);
3364 3365
            syncNodeEventLog(ths, logBuf);
          } while (0);
3366

3367 3368 3369 3370 3371 3372 3373 3374 3375 3376 3377 3378 3379 3380 3381
          // execute fsm in apply thread, or execute outside syncPropose
          if (internalExecute) {
            SFsmCbMeta cbMeta = {0};
            cbMeta.index = pEntry->index;
            cbMeta.lastConfigIndex = syncNodeGetSnapshotConfigIndex(ths, cbMeta.index);
            cbMeta.isWeak = pEntry->isWeak;
            cbMeta.code = 0;
            cbMeta.state = ths->state;
            cbMeta.seqNum = pEntry->seqNum;
            cbMeta.term = pEntry->term;
            cbMeta.currentTerm = ths->pRaftStore->currentTerm;
            cbMeta.flag = flag;

            ths->pFsm->FpCommitCb(ths->pFsm, &rpcMsg, cbMeta);
          }
3382 3383 3384
        }

        // config change
3385
        if (pEntry->originalRpcType == TDMT_SYNC_CONFIG_CHANGE) {
M
Minghao Li 已提交
3386 3387 3388 3389 3390 3391 3392 3393 3394 3395 3396 3397 3398 3399 3400
          SyncReconfigFinish* pFinish = syncReconfigFinishBuild(ths->vgId);
          ASSERT(pFinish != NULL);

          code = syncNodeConfigChange(ths, &rpcMsg, pEntry, pFinish);
          ASSERT(code == 0);

          if (ths->state == TAOS_SYNC_STATE_LEADER) {
            syncNodeProposeConfigChangeFinish(ths, pFinish);
          }
          syncReconfigFinishDestroy(pFinish);
        }

        // config change finish
        if (pEntry->originalRpcType == TDMT_SYNC_CONFIG_CHANGE_FINISH) {
          code = syncNodeConfigChangeFinish(ths, &rpcMsg, pEntry);
3401 3402
          ASSERT(code == 0);
        }
3403

3404 3405
#if 0
        // execute in pre-commit
M
Minghao Li 已提交
3406
        // leader transfer
3407 3408 3409
        if (pEntry->originalRpcType == TDMT_SYNC_LEADER_TRANSFER) {
          code = syncDoLeaderTransfer(ths, &rpcMsg, pEntry);
          ASSERT(code == 0);
3410
        }
3411
#endif
3412 3413

        // restore finish
3414
        // if only snapshot, a noop entry will be append, so syncLogLastIndex is always ok
3415 3416 3417 3418 3419 3420
        if (pEntry->index == ths->pLogStore->syncLogLastIndex(ths->pLogStore)) {
          if (ths->restoreFinish == false) {
            if (ths->pFsm->FpRestoreFinishCb != NULL) {
              ths->pFsm->FpRestoreFinishCb(ths->pFsm);
            }
            ths->restoreFinish = true;
M
Minghao Li 已提交
3421

3422 3423
            int64_t restoreDelay = taosGetTimestampMs() - ths->leaderTime;

M
Minghao Li 已提交
3424
            char eventLog[128];
S
Shengliang Guan 已提交
3425 3426
            snprintf(eventLog, sizeof(eventLog), "restore finish, index:%" PRId64 ", elapsed:%" PRId64 " ms, ",
                     pEntry->index, restoreDelay);
M
Minghao Li 已提交
3427
            syncNodeEventLog(ths, eventLog);
3428 3429 3430 3431
          }
        }

        rpcFreeCont(rpcMsg.pCont);
3432 3433 3434 3435 3436
        if (h) {
          taosLRUCacheRelease(pCache, h, false);
        } else {
          syncEntryDestory(pEntry);
        }
3437 3438 3439 3440
      }
    }
  }
  return 0;
3441 3442 3443 3444 3445 3446 3447 3448 3449
}

bool syncNodeInRaftGroup(SSyncNode* ths, SRaftId* pRaftId) {
  for (int i = 0; i < ths->replicaNum; ++i) {
    if (syncUtilSameId(&((ths->replicasId)[i]), pRaftId)) {
      return true;
    }
  }
  return false;
M
Minghao Li 已提交
3450 3451 3452 3453 3454 3455 3456 3457 3458 3459
}

SSyncSnapshotSender* syncNodeGetSnapshotSender(SSyncNode* ths, SRaftId* pDestId) {
  SSyncSnapshotSender* pSender = NULL;
  for (int i = 0; i < ths->replicaNum; ++i) {
    if (syncUtilSameId(pDestId, &((ths->replicasId)[i]))) {
      pSender = (ths->senders)[i];
    }
  }
  return pSender;
M
Minghao Li 已提交
3460
}
M
Minghao Li 已提交
3461

3462 3463 3464 3465 3466 3467 3468 3469 3470 3471
SSyncTimer* syncNodeGetHbTimer(SSyncNode* ths, SRaftId* pDestId) {
  SSyncTimer* pTimer = NULL;
  for (int i = 0; i < ths->replicaNum; ++i) {
    if (syncUtilSameId(pDestId, &((ths->replicasId)[i]))) {
      pTimer = &((ths->peerHeartbeatTimerArr)[i]);
    }
  }
  return pTimer;
}

M
Minghao Li 已提交
3472 3473 3474 3475 3476 3477 3478 3479 3480 3481 3482 3483
SPeerState* syncNodeGetPeerState(SSyncNode* ths, const SRaftId* pDestId) {
  SPeerState* pState = NULL;
  for (int i = 0; i < ths->replicaNum; ++i) {
    if (syncUtilSameId(pDestId, &((ths->replicasId)[i]))) {
      pState = &((ths->peerStates)[i]);
    }
  }
  return pState;
}

bool syncNodeNeedSendAppendEntries(SSyncNode* ths, const SRaftId* pDestId, const SyncAppendEntries* pMsg) {
  SPeerState* pState = syncNodeGetPeerState(ths, pDestId);
M
Minghao Li 已提交
3484 3485 3486
  if (pState == NULL) {
    return false;
  }
M
Minghao Li 已提交
3487 3488 3489 3490 3491 3492 3493 3494 3495 3496 3497

  SyncIndex sendIndex = pMsg->prevLogIndex + 1;
  int64_t   tsNow = taosGetTimestampMs();

  if (pState->lastSendIndex == sendIndex && tsNow - pState->lastSendTime < SYNC_APPEND_ENTRIES_TIMEOUT_MS) {
    return false;
  }

  return true;
}

M
Minghao Li 已提交
3498 3499 3500 3501 3502 3503 3504 3505 3506 3507 3508 3509 3510 3511 3512 3513 3514 3515 3516 3517 3518 3519 3520
bool syncNodeCanChange(SSyncNode* pSyncNode) {
  if (pSyncNode->changing) {
    sError("sync cannot change");
    return false;
  }

  if ((pSyncNode->commitIndex >= SYNC_INDEX_BEGIN)) {
    SyncIndex lastIndex = syncNodeGetLastIndex(pSyncNode);
    if (pSyncNode->commitIndex != lastIndex) {
      sError("sync cannot change2");
      return false;
    }
  }

  for (int i = 0; i < pSyncNode->peersNum; ++i) {
    SSyncSnapshotSender* pSender = syncNodeGetSnapshotSender(pSyncNode, &(pSyncNode->peersId)[i]);
    if (pSender->start) {
      sError("sync cannot change3");
      return false;
    }
  }

  return true;
M
Minghao Li 已提交
3521 3522
}

3523 3524 3525 3526 3527 3528 3529 3530 3531 3532 3533 3534 3535 3536
const char* syncTimerTypeStr(enum ESyncTimeoutType timerType) {
  if (timerType == SYNC_TIMEOUT_PING) {
    return "ping";
  } else if (timerType == SYNC_TIMEOUT_ELECTION) {
    return "elect";
  } else if (timerType == SYNC_TIMEOUT_HEARTBEAT) {
    return "heartbeat";
  } else {
    return "unknown";
  }
}

void syncLogRecvTimer(SSyncNode* pSyncNode, const SyncTimeout* pMsg, const char* s) {
  char logBuf[256];
S
Shengliang Guan 已提交
3537
  snprintf(logBuf, sizeof(logBuf), "recv sync-timer {type:%s, lc:%" PRIu64 ", ms:%d, data:%p}, %s",
M
Minghao Li 已提交
3538
           syncTimerTypeStr(pMsg->timeoutType), pMsg->logicClock, pMsg->timerMS, pMsg->data, s);
3539 3540 3541
  syncNodeEventLog(pSyncNode, logBuf);
}

wafwerar's avatar
wafwerar 已提交
3542
void syncLogSendRequestVote(SSyncNode* pSyncNode, const SyncRequestVote* pMsg, const char* s) {
M
Minghao Li 已提交
3543 3544 3545 3546 3547 3548 3549 3550 3551 3552
  char     host[64];
  uint16_t port;
  syncUtilU642Addr(pMsg->destId.addr, host, sizeof(host), &port);
  char logBuf[256];
  snprintf(logBuf, sizeof(logBuf),
           "send sync-request-vote to %s:%d {term:%" PRIu64 ", lindex:%" PRId64 ", lterm:%" PRIu64 "}, %s", host, port,
           pMsg->term, pMsg->lastLogIndex, pMsg->lastLogTerm, s);
  syncNodeEventLog(pSyncNode, logBuf);
}

wafwerar's avatar
wafwerar 已提交
3553
void syncLogRecvRequestVote(SSyncNode* pSyncNode, const SyncRequestVote* pMsg, const char* s) {
M
Minghao Li 已提交
3554 3555 3556 3557 3558 3559 3560 3561 3562 3563
  char     logBuf[256];
  char     host[64];
  uint16_t port;
  syncUtilU642Addr(pMsg->srcId.addr, host, sizeof(host), &port);
  snprintf(logBuf, sizeof(logBuf),
           "recv sync-request-vote from %s:%d, {term:%" PRIu64 ", lindex:%" PRId64 ", lterm:%" PRIu64 "}, %s", host,
           port, pMsg->term, pMsg->lastLogIndex, pMsg->lastLogTerm, s);
  syncNodeEventLog(pSyncNode, logBuf);
}

wafwerar's avatar
wafwerar 已提交
3564
void syncLogSendRequestVoteReply(SSyncNode* pSyncNode, const SyncRequestVoteReply* pMsg, const char* s) {
M
Minghao Li 已提交
3565 3566 3567 3568 3569 3570 3571 3572 3573
  char     host[64];
  uint16_t port;
  syncUtilU642Addr(pMsg->destId.addr, host, sizeof(host), &port);
  char logBuf[256];
  snprintf(logBuf, sizeof(logBuf), "send sync-request-vote-reply to %s:%d {term:%" PRIu64 ", grant:%d}, %s", host, port,
           pMsg->term, pMsg->voteGranted, s);
  syncNodeEventLog(pSyncNode, logBuf);
}

wafwerar's avatar
wafwerar 已提交
3574
void syncLogRecvRequestVoteReply(SSyncNode* pSyncNode, const SyncRequestVoteReply* pMsg, const char* s) {
M
Minghao Li 已提交
3575 3576 3577 3578 3579 3580 3581 3582 3583
  char     host[64];
  uint16_t port;
  syncUtilU642Addr(pMsg->srcId.addr, host, sizeof(host), &port);
  char logBuf[256];
  snprintf(logBuf, sizeof(logBuf), "recv sync-request-vote-reply from %s:%d {term:%" PRIu64 ", grant:%d}, %s", host,
           port, pMsg->term, pMsg->voteGranted, s);
  syncNodeEventLog(pSyncNode, logBuf);
}

wafwerar's avatar
wafwerar 已提交
3584
void syncLogSendAppendEntries(SSyncNode* pSyncNode, const SyncAppendEntries* pMsg, const char* s) {
M
Minghao Li 已提交
3585 3586 3587 3588 3589 3590
  char     host[64];
  uint16_t port;
  syncUtilU642Addr(pMsg->destId.addr, host, sizeof(host), &port);
  char logBuf[256];
  snprintf(logBuf, sizeof(logBuf),
           "send sync-append-entries to %s:%d, {term:%" PRIu64 ", pre-index:%" PRId64 ", pre-term:%" PRIu64
M
Minghao Li 已提交
3591
           ", pterm:%" PRIu64 ", cmt:%" PRId64
M
Minghao Li 已提交
3592 3593 3594 3595 3596 3597 3598
           ", "
           "datalen:%d}, %s",
           host, port, pMsg->term, pMsg->prevLogIndex, pMsg->prevLogTerm, pMsg->privateTerm, pMsg->commitIndex,
           pMsg->dataLen, s);
  syncNodeEventLog(pSyncNode, logBuf);
}

wafwerar's avatar
wafwerar 已提交
3599
void syncLogRecvAppendEntries(SSyncNode* pSyncNode, const SyncAppendEntries* pMsg, const char* s) {
M
Minghao Li 已提交
3600 3601 3602 3603 3604
  char     host[64];
  uint16_t port;
  syncUtilU642Addr(pMsg->srcId.addr, host, sizeof(host), &port);
  char logBuf[256];
  snprintf(logBuf, sizeof(logBuf),
M
Minghao Li 已提交
3605
           "recv sync-append-entries from %s:%d {term:%" PRIu64 ", pre-index:%" PRIu64 ", pre-term:%" PRIu64
M
Minghao Li 已提交
3606
           ", cmt:%" PRIu64 ", pterm:%" PRIu64
M
Minghao Li 已提交
3607 3608 3609 3610
           ", "
           "datalen:%d}, %s",
           host, port, pMsg->term, pMsg->prevLogIndex, pMsg->prevLogTerm, pMsg->commitIndex, pMsg->privateTerm,
           pMsg->dataLen, s);
M
Minghao Li 已提交
3611
  syncNodeEventLog(pSyncNode, logBuf);
M
Minghao Li 已提交
3612 3613
}

wafwerar's avatar
wafwerar 已提交
3614
void syncLogSendAppendEntriesBatch(SSyncNode* pSyncNode, const SyncAppendEntriesBatch* pMsg, const char* s) {
M
Minghao Li 已提交
3615 3616 3617 3618 3619 3620
  char     host[64];
  uint16_t port;
  syncUtilU642Addr(pMsg->destId.addr, host, sizeof(host), &port);
  char logBuf[256];
  snprintf(logBuf, sizeof(logBuf),
           "send sync-append-entries-batch to %s:%d, {term:%" PRIu64 ", pre-index:%" PRId64 ", pre-term:%" PRIu64
M
Minghao Li 已提交
3621
           ", pterm:%" PRIu64 ", cmt:%" PRId64 ", datalen:%d, count:%d}, %s",
M
Minghao Li 已提交
3622 3623
           host, port, pMsg->term, pMsg->prevLogIndex, pMsg->prevLogTerm, pMsg->privateTerm, pMsg->commitIndex,
           pMsg->dataLen, pMsg->dataCount, s);
M
Minghao Li 已提交
3624 3625 3626
  syncNodeEventLog(pSyncNode, logBuf);
}

wafwerar's avatar
wafwerar 已提交
3627
void syncLogRecvAppendEntriesBatch(SSyncNode* pSyncNode, const SyncAppendEntriesBatch* pMsg, const char* s) {
M
Minghao Li 已提交
3628 3629 3630 3631 3632 3633
  char     host[64];
  uint16_t port;
  syncUtilU642Addr(pMsg->srcId.addr, host, sizeof(host), &port);
  char logBuf[256];
  snprintf(logBuf, sizeof(logBuf),
           "recv sync-append-entries-batch from %s:%d, {term:%" PRIu64 ", pre-index:%" PRId64 ", pre-term:%" PRIu64
M
Minghao Li 已提交
3634
           ", pterm:%" PRIu64 ", cmt:%" PRId64 ", datalen:%d, count:%d}, %s",
M
Minghao Li 已提交
3635 3636
           host, port, pMsg->term, pMsg->prevLogIndex, pMsg->prevLogTerm, pMsg->privateTerm, pMsg->commitIndex,
           pMsg->dataLen, pMsg->dataCount, s);
M
Minghao Li 已提交
3637
  syncNodeEventLog(pSyncNode, logBuf);
M
Minghao Li 已提交
3638 3639
}

3640
void syncLogSendAppendEntriesReply(SSyncNode* pSyncNode, const SyncAppendEntriesReply* pMsg, const char* s) {
M
Minghao Li 已提交
3641 3642 3643 3644 3645 3646 3647 3648 3649 3650 3651
  char     host[64];
  uint16_t port;
  syncUtilU642Addr(pMsg->destId.addr, host, sizeof(host), &port);
  char logBuf[256];
  snprintf(logBuf, sizeof(logBuf),
           "send sync-append-entries-reply to %s:%d, {term:%" PRIu64 ", pterm:%" PRIu64 ", success:%d, match:%" PRId64
           "}, %s",
           host, port, pMsg->term, pMsg->privateTerm, pMsg->success, pMsg->matchIndex, s);
  syncNodeEventLog(pSyncNode, logBuf);
}

wafwerar's avatar
wafwerar 已提交
3652
void syncLogRecvAppendEntriesReply(SSyncNode* pSyncNode, const SyncAppendEntriesReply* pMsg, const char* s) {
M
Minghao Li 已提交
3653 3654 3655 3656 3657
  char     host[64];
  uint16_t port;
  syncUtilU642Addr(pMsg->srcId.addr, host, sizeof(host), &port);
  char logBuf[256];
  snprintf(logBuf, sizeof(logBuf),
M
Minghao Li 已提交
3658
           "recv sync-append-entries-reply from %s:%d {term:%" PRIu64 ", pterm:%" PRIu64 ", success:%d, match:%" PRId64
M
Minghao Li 已提交
3659 3660
           "}, %s",
           host, port, pMsg->term, pMsg->privateTerm, pMsg->success, pMsg->matchIndex, s);
M
Minghao Li 已提交
3661
  syncNodeEventLog(pSyncNode, logBuf);
M
Minghao Li 已提交
3662
}
3663 3664 3665 3666 3667 3668 3669

void syncLogSendHeartbeat(SSyncNode* pSyncNode, const SyncHeartbeat* pMsg, const char* s) {
  char     host[64];
  uint16_t port;
  syncUtilU642Addr(pMsg->destId.addr, host, sizeof(host), &port);
  char logBuf[256];
  snprintf(logBuf, sizeof(logBuf),
M
Minghao Li 已提交
3670
           "send sync-heartbeat to %s:%d {term:%" PRIu64 ", cmt:%" PRId64 ", min-match:%" PRId64 ", pterm:%" PRIu64
M
Minghao Li 已提交
3671 3672
           "}, %s",
           host, port, pMsg->term, pMsg->commitIndex, pMsg->minMatchIndex, pMsg->privateTerm, s);
3673 3674 3675 3676 3677 3678 3679 3680 3681
  syncNodeEventLog(pSyncNode, logBuf);
}

void syncLogRecvHeartbeat(SSyncNode* pSyncNode, const SyncHeartbeat* pMsg, const char* s) {
  char     host[64];
  uint16_t port;
  syncUtilU642Addr(pMsg->srcId.addr, host, sizeof(host), &port);
  char logBuf[256];
  snprintf(logBuf, sizeof(logBuf),
M
Minghao Li 已提交
3682 3683 3684
           "recv sync-heartbeat from %s:%d {term:%" PRIu64 ", cmt:%" PRId64 ", min-match:%" PRId64 ", pterm:%" PRIu64
           "}, %s",
           host, port, pMsg->term, pMsg->commitIndex, pMsg->minMatchIndex, pMsg->privateTerm, s);
3685 3686 3687 3688 3689 3690 3691 3692 3693 3694 3695 3696 3697 3698 3699 3700 3701 3702 3703 3704 3705
  syncNodeEventLog(pSyncNode, logBuf);
}

void syncLogSendHeartbeatReply(SSyncNode* pSyncNode, const SyncHeartbeatReply* pMsg, const char* s) {
  char     host[64];
  uint16_t port;
  syncUtilU642Addr(pMsg->destId.addr, host, sizeof(host), &port);
  char logBuf[256];
  snprintf(logBuf, sizeof(logBuf), "send sync-heartbeat-reply from %s:%d {term:%" PRIu64 ", pterm:%" PRIu64 "}, %s",
           host, port, pMsg->term, pMsg->privateTerm, s);
  syncNodeEventLog(pSyncNode, logBuf);
}

void syncLogRecvHeartbeatReply(SSyncNode* pSyncNode, const SyncHeartbeatReply* pMsg, const char* s) {
  char     host[64];
  uint16_t port;
  syncUtilU642Addr(pMsg->srcId.addr, host, sizeof(host), &port);
  char logBuf[256];
  snprintf(logBuf, sizeof(logBuf), "recv sync-heartbeat-reply from %s:%d {term:%" PRIu64 ", pterm:%" PRIu64 "}, %s",
           host, port, pMsg->term, pMsg->privateTerm, s);
  syncNodeEventLog(pSyncNode, logBuf);
3706
}