syncMain.c 98.9 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/>.
 */

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

M
Minghao Li 已提交
41 42 43 44 45
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);
46
static void    syncNodeEqPeerHeartbeatTimer(void* param, void* tmrId);
S
Shengliang Guan 已提交
47
static bool    syncIsConfigChanged(const SSyncCfg* pOldCfg, const SSyncCfg* pNewCfg);
S
Shengliang Guan 已提交
48 49 50
static int32_t syncHbTimerInit(SSyncNode* pSyncNode, SSyncTimer* pSyncTimer, SRaftId destId);
static int32_t syncHbTimerStart(SSyncNode* pSyncNode, SSyncTimer* pSyncTimer);
static int32_t syncHbTimerStop(SSyncNode* pSyncNode, SSyncTimer* pSyncTimer);
S
Shengliang Guan 已提交
51 52 53 54 55 56 57 58 59 60 61
static int32_t syncNodeUpdateNewConfigIndex(SSyncNode* ths, SSyncCfg* pNewCfg);
static bool    syncNodeInConfig(SSyncNode* pSyncNode, const SSyncCfg* config);
static void    syncNodeDoConfigChange(SSyncNode* pSyncNode, SSyncCfg* newConfig, SyncIndex lastConfigChangeIndex);
static bool    syncNodeIsOptimizedOneReplica(SSyncNode* ths, SRpcMsg* pMsg);

static bool    syncNodeCanChange(SSyncNode* pSyncNode);
static int32_t syncNodeLeaderTransfer(SSyncNode* pSyncNode);
static int32_t syncNodeLeaderTransferTo(SSyncNode* pSyncNode, SNodeInfo newLeader);
static int32_t syncDoLeaderTransfer(SSyncNode* ths, SRpcMsg* pRpcMsg, SSyncRaftEntry* pEntry);

static ESyncStrategy syncNodeStrategy(SSyncNode* pSyncNode);
M
Minghao Li 已提交
62

63
int64_t syncOpen(SSyncInfo* pSyncInfo) {
M
Minghao Li 已提交
64
  SSyncNode* pSyncNode = syncNodeOpen(pSyncInfo);
65
  if (pSyncNode == NULL) {
S
Shengliang Guan 已提交
66
    sError("vgId:%d, failed to open sync node", pSyncInfo->vgId);
67 68
    return -1;
  }
M
Minghao Li 已提交
69

S
Shengliang Guan 已提交
70
  pSyncNode->rid = syncNodeAdd(pSyncNode);
M
Minghao Li 已提交
71
  if (pSyncNode->rid < 0) {
72
    syncNodeClose(pSyncNode);
M
Minghao Li 已提交
73 74 75
    return -1;
  }

S
Shengliang Guan 已提交
76 77 78 79 80 81
  pSyncNode->pingBaseLine = pSyncInfo->pingMs;
  pSyncNode->pingTimerMS = pSyncInfo->pingMs;
  pSyncNode->electBaseLine = pSyncInfo->electMs;
  pSyncNode->hbBaseLine = pSyncInfo->heartbeatMs;
  pSyncNode->heartbeatTimerMS = pSyncInfo->heartbeatMs;
  pSyncNode->msgcb = pSyncInfo->msgcb;
M
Minghao Li 已提交
82
  return pSyncNode->rid;
M
Minghao Li 已提交
83
}
M
Minghao Li 已提交
84

B
Benguang Zhao 已提交
85
int32_t syncStart(int64_t rid) {
S
Shengliang Guan 已提交
86
  SSyncNode* pSyncNode = syncNodeAcquire(rid);
M
Minghao Li 已提交
87
  if (pSyncNode == NULL) {
B
Benguang Zhao 已提交
88 89 90 91 92
    sError("failed to acquire rid: %" PRId64 " of tsNodeReftId for pSyncNode", rid);
    return -1;
  }

  if (syncNodeRestore(pSyncNode) < 0) {
93
    sError("vgId:%d, failed to restore sync log buffer since %s", pSyncNode->vgId, terrstr());
94
    goto _err;
M
Minghao Li 已提交
95
  }
M
Minghao Li 已提交
96

B
Benguang Zhao 已提交
97 98 99 100
  if (syncNodeStart(pSyncNode) < 0) {
    sError("vgId:%d, failed to start sync node since %s", pSyncNode->vgId, terrstr());
    goto _err;
  }
M
Minghao Li 已提交
101

B
Benguang Zhao 已提交
102 103
  syncNodeRelease(pSyncNode);
  return 0;
M
Minghao Li 已提交
104

105 106 107
_err:
  syncNodeRelease(pSyncNode);
  return -1;
M
Minghao Li 已提交
108 109
}

M
Minghao Li 已提交
110
void syncStop(int64_t rid) {
S
Shengliang Guan 已提交
111
  SSyncNode* pSyncNode = syncNodeAcquire(rid);
S
Shengliang Guan 已提交
112
  if (pSyncNode != NULL) {
113
    pSyncNode->isStart = false;
S
Shengliang Guan 已提交
114
    syncNodeRelease(pSyncNode);
S
Shengliang Guan 已提交
115
    syncNodeRemove(rid);
M
Minghao Li 已提交
116 117 118
  }
}

M
Minghao Li 已提交
119 120
void syncPreStop(int64_t rid) {
  SSyncNode* pSyncNode = syncNodeAcquire(rid);
121 122 123
  if (pSyncNode != NULL) {
    syncNodePreClose(pSyncNode);
    syncNodeRelease(pSyncNode);
M
Minghao Li 已提交
124 125 126
  }
}

127 128 129 130 131 132 133 134
void syncPostStop(int64_t rid) {
  SSyncNode* pSyncNode = syncNodeAcquire(rid);
  if (pSyncNode != NULL) {
    syncNodePostClose(pSyncNode);
    syncNodeRelease(pSyncNode);
  }
}

S
Shengliang Guan 已提交
135 136 137
static bool syncNodeCheckNewConfig(SSyncNode* pSyncNode, const SSyncCfg* pCfg) {
  if (!syncNodeInConfig(pSyncNode, pCfg)) return false;
  return abs(pCfg->replicaNum - pSyncNode->replicaNum) <= 1;
M
Minghao Li 已提交
138 139
}

S
Shengliang Guan 已提交
140
int32_t syncReconfig(int64_t rid, SSyncCfg* pNewCfg) {
S
Shengliang Guan 已提交
141
  SSyncNode* pSyncNode = syncNodeAcquire(rid);
S
Shengliang Guan 已提交
142
  if (pSyncNode == NULL) return -1;
M
Minghao Li 已提交
143

M
Minghao Li 已提交
144
  if (!syncNodeCheckNewConfig(pSyncNode, pNewCfg)) {
S
Shengliang Guan 已提交
145
    syncNodeRelease(pSyncNode);
M
Minghao Li 已提交
146
    terrno = TSDB_CODE_SYN_NEW_CONFIG_ERROR;
S
Shengliang Guan 已提交
147
    sError("vgId:%d, failed to reconfig since invalid new config", pSyncNode->vgId);
M
Minghao Li 已提交
148
    return -1;
M
Minghao Li 已提交
149
  }
150

S
Shengliang Guan 已提交
151 152
  syncNodeUpdateNewConfigIndex(pSyncNode, pNewCfg);
  syncNodeDoConfigChange(pSyncNode, pNewCfg, SYNC_INDEX_INVALID);
S
Shengliang Guan 已提交
153

M
Minghao Li 已提交
154 155 156 157
  if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
    syncNodeStopHeartbeatTimer(pSyncNode);

    for (int32_t i = 0; i < TSDB_MAX_REPLICA; ++i) {
S
Shengliang Guan 已提交
158
      syncHbTimerInit(pSyncNode, &pSyncNode->peerHeartbeatTimerArr[i], pSyncNode->replicasId[i]);
M
Minghao Li 已提交
159 160 161
    }

    syncNodeStartHeartbeatTimer(pSyncNode);
S
Shengliang Guan 已提交
162
    // syncNodeReplicate(pSyncNode);
M
Minghao Li 已提交
163
  }
S
Shengliang Guan 已提交
164

S
Shengliang Guan 已提交
165
  syncNodeRelease(pSyncNode);
S
Shengliang Guan 已提交
166
  return 0;
M
Minghao Li 已提交
167
}
M
Minghao Li 已提交
168

S
Shengliang Guan 已提交
169 170 171 172
int32_t syncProcessMsg(int64_t rid, SRpcMsg* pMsg) {
  int32_t code = -1;
  if (!syncIsInit()) return code;

S
Shengliang Guan 已提交
173
  SSyncNode* pSyncNode = syncNodeAcquire(rid);
S
Shengliang Guan 已提交
174 175
  if (pSyncNode == NULL) return code;

S
Shengliang Guan 已提交
176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204
  switch (pMsg->msgType) {
    case TDMT_SYNC_HEARTBEAT:
      code = syncNodeOnHeartbeat(pSyncNode, pMsg);
      break;
    case TDMT_SYNC_HEARTBEAT_REPLY:
      code = syncNodeOnHeartbeatReply(pSyncNode, pMsg);
      break;
    case TDMT_SYNC_TIMEOUT:
      code = syncNodeOnTimeout(pSyncNode, pMsg);
      break;
    case TDMT_SYNC_CLIENT_REQUEST:
      code = syncNodeOnClientRequest(pSyncNode, pMsg, NULL);
      break;
    case TDMT_SYNC_REQUEST_VOTE:
      code = syncNodeOnRequestVote(pSyncNode, pMsg);
      break;
    case TDMT_SYNC_REQUEST_VOTE_REPLY:
      code = syncNodeOnRequestVoteReply(pSyncNode, pMsg);
      break;
    case TDMT_SYNC_APPEND_ENTRIES:
      code = syncNodeOnAppendEntries(pSyncNode, pMsg);
      break;
    case TDMT_SYNC_APPEND_ENTRIES_REPLY:
      code = syncNodeOnAppendEntriesReply(pSyncNode, pMsg);
      break;
    case TDMT_SYNC_SNAPSHOT_SEND:
      code = syncNodeOnSnapshot(pSyncNode, pMsg);
      break;
    case TDMT_SYNC_SNAPSHOT_RSP:
S
Shengliang Guan 已提交
205
      code = syncNodeOnSnapshotRsp(pSyncNode, pMsg);
S
Shengliang Guan 已提交
206 207 208 209 210
      break;
    case TDMT_SYNC_LOCAL_CMD:
      code = syncNodeOnLocalCmd(pSyncNode, pMsg);
      break;
    default:
211
      terrno = TSDB_CODE_MSG_NOT_PROCESSED;
S
Shengliang Guan 已提交
212
      code = -1;
M
Minghao Li 已提交
213 214
  }

S
Shengliang Guan 已提交
215
  syncNodeRelease(pSyncNode);
216 217 218 219
  if (code != 0) {
    sDebug("vgId:%d, failed to process sync msg:%p type:%s since 0x%x", pSyncNode->vgId, pMsg, TMSG_INFO(pMsg->msgType),
           terrno);
  }
S
Shengliang Guan 已提交
220
  return code;
221 222
}

S
Shengliang Guan 已提交
223
int32_t syncLeaderTransfer(int64_t rid) {
S
Shengliang Guan 已提交
224
  SSyncNode* pSyncNode = syncNodeAcquire(rid);
S
Shengliang Guan 已提交
225
  if (pSyncNode == NULL) return -1;
226

S
Shengliang Guan 已提交
227
  int32_t ret = syncNodeLeaderTransfer(pSyncNode);
S
Shengliang Guan 已提交
228
  syncNodeRelease(pSyncNode);
229 230 231
  return ret;
}

232
int32_t syncSendTimeoutRsp(int64_t rid, int64_t seq) {
S
Shengliang Guan 已提交
233
  SSyncNode* pNode = syncNodeAcquire(rid);
234
  if (pNode == NULL) return -1;
S
Shengliang Guan 已提交
235 236

  SRpcMsg rpcMsg = {0};
237
  int32_t ret = syncRespMgrGetAndDel(pNode->pSyncRespMgr, seq, &rpcMsg.info);
S
Shengliang Guan 已提交
238 239 240
  rpcMsg.code = TSDB_CODE_SYN_TIMEOUT;

  syncNodeRelease(pNode);
241
  if (ret == 1) {
242
    sInfo("send timeout response, seq:%" PRId64 " handle:%p ahandle:%p", seq, rpcMsg.info.handle, rpcMsg.info.ahandle);
243
    rpcSendResponse(&rpcMsg);
244 245
    return 0;
  } else {
246
    sError("no message handle to send timeout response, seq:%" PRId64, seq);
247
    return -1;
248
  }
S
Shengliang Guan 已提交
249 250
}

M
Minghao Li 已提交
251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266
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;
}

267
int32_t syncBeginSnapshot(int64_t rid, int64_t lastApplyIndex) {
S
Shengliang Guan 已提交
268
  SSyncNode* pSyncNode = syncNodeAcquire(rid);
269
  if (pSyncNode == NULL) {
270
    sError("sync begin snapshot error");
271 272
    return -1;
  }
273

274 275
  int32_t code = 0;

M
Minghao Li 已提交
276
  if (syncNodeIsMnode(pSyncNode)) {
M
Minghao Li 已提交
277 278 279
    // mnode
    int64_t logRetention = SYNC_MNODE_LOG_RETENTION;

M
Minghao Li 已提交
280 281 282
    SyncIndex beginIndex = pSyncNode->pLogStore->syncLogBeginIndex(pSyncNode->pLogStore);
    SyncIndex endIndex = pSyncNode->pLogStore->syncLogEndIndex(pSyncNode->pLogStore);
    int64_t   logNum = endIndex - beginIndex;
M
Minghao Li 已提交
283 284 285
    bool      isEmpty = pSyncNode->pLogStore->syncLogIsEmpty(pSyncNode->pLogStore);

    if (isEmpty || (!isEmpty && logNum < logRetention)) {
S
Shengliang Guan 已提交
286 287
      sNTrace(pSyncNode, "new-snapshot-index:%" PRId64 ", log-num:%" PRId64 ", empty:%d, do not delete wal",
              lastApplyIndex, logNum, isEmpty);
S
Shengliang Guan 已提交
288
      syncNodeRelease(pSyncNode);
M
Minghao Li 已提交
289 290 291
      return 0;
    }

M
Minghao Li 已提交
292 293 294
    goto _DEL_WAL;

  } else {
295 296 297 298 299 300 301 302 303 304 305 306
    lastApplyIndex -= SYNC_VNODE_LOG_RETENTION;

    SyncIndex beginIndex = pSyncNode->pLogStore->syncLogBeginIndex(pSyncNode->pLogStore);
    SyncIndex endIndex = pSyncNode->pLogStore->syncLogEndIndex(pSyncNode->pLogStore);
    bool      isEmpty = pSyncNode->pLogStore->syncLogIsEmpty(pSyncNode->pLogStore);

    if (isEmpty || !(lastApplyIndex >= beginIndex && lastApplyIndex <= endIndex)) {
      sNTrace(pSyncNode, "new-snapshot-index:%" PRId64 ", empty:%d, do not delete wal", lastApplyIndex, isEmpty);
      syncNodeRelease(pSyncNode);
      return 0;
    }

M
Minghao Li 已提交
307 308 309 310 311 312 313 314 315 316
    // 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) {
317 318 319 320
            sNTrace(pSyncNode,
                    "new-snapshot-index:%" PRId64 " is greater than match-index:%" PRId64
                    " of dnode:%d, do not delete wal",
                    lastApplyIndex, matchIndex, DID(&pSyncNode->peersId[i]));
M
Minghao Li 已提交
321

S
Shengliang Guan 已提交
322
            syncNodeRelease(pSyncNode);
M
Minghao Li 已提交
323 324 325 326 327 328
            return 0;
          }
        }

      } else if (pSyncNode->state == TAOS_SYNC_STATE_FOLLOWER) {
        if (lastApplyIndex > pSyncNode->minMatchIndex) {
S
Shengliang Guan 已提交
329 330 331
          sNTrace(pSyncNode,
                  "new-snapshot-index:%" PRId64 " is greater than min-match-index:%" PRId64 ", do not delete wal",
                  lastApplyIndex, pSyncNode->minMatchIndex);
S
Shengliang Guan 已提交
332
          syncNodeRelease(pSyncNode);
M
Minghao Li 已提交
333 334 335 336
          return 0;
        }

      } else if (pSyncNode->state == TAOS_SYNC_STATE_CANDIDATE) {
S
Shengliang Guan 已提交
337
        sNTrace(pSyncNode, "new-snapshot-index:%" PRId64 " candidate, do not delete wal", lastApplyIndex);
S
Shengliang Guan 已提交
338
        syncNodeRelease(pSyncNode);
M
Minghao Li 已提交
339 340 341
        return 0;

      } else {
S
Shengliang Guan 已提交
342
        sNTrace(pSyncNode, "new-snapshot-index:%" PRId64 " unknown state, do not delete wal", lastApplyIndex);
S
Shengliang Guan 已提交
343
        syncNodeRelease(pSyncNode);
M
Minghao Li 已提交
344 345 346 347 348 349 350 351 352
        return 0;
      }

      goto _DEL_WAL;

    } else {
      // one replica

      goto _DEL_WAL;
353 354 355
    }
  }

M
Minghao Li 已提交
356
_DEL_WAL:
357

M
Minghao Li 已提交
358
  do {
359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378
    SSyncLogStoreData* pData = pSyncNode->pLogStore->data;
    SyncIndex          snapshotVer = walGetSnapshotVer(pData->pWal);
    SyncIndex          walCommitVer = walGetCommittedVer(pData->pWal);
    SyncIndex          wallastVer = walGetLastVer(pData->pWal);
    if (lastApplyIndex <= walCommitVer) {
      SyncIndex snapshottingIndex = atomic_load_64(&pSyncNode->snapshottingIndex);

      if (snapshottingIndex == SYNC_INDEX_INVALID) {
        atomic_store_64(&pSyncNode->snapshottingIndex, lastApplyIndex);
        pSyncNode->snapshottingTime = taosGetTimestampMs();

        code = walBeginSnapshot(pData->pWal, lastApplyIndex);
        if (code == 0) {
          sNTrace(pSyncNode, "wal snapshot begin, index:%" PRId64 ", last apply index:%" PRId64,
                  pSyncNode->snapshottingIndex, lastApplyIndex);
        } else {
          sNError(pSyncNode, "wal snapshot begin error since:%s, index:%" PRId64 ", last apply index:%" PRId64,
                  terrstr(terrno), pSyncNode->snapshottingIndex, lastApplyIndex);
          atomic_store_64(&pSyncNode->snapshottingIndex, SYNC_INDEX_INVALID);
        }
379

M
Minghao Li 已提交
380
      } else {
381 382
        sNTrace(pSyncNode, "snapshotting for %" PRId64 ", do not delete wal for new-snapshot-index:%" PRId64,
                snapshottingIndex, lastApplyIndex);
M
Minghao Li 已提交
383
      }
384
    }
M
Minghao Li 已提交
385
  } while (0);
386

S
Shengliang Guan 已提交
387
  syncNodeRelease(pSyncNode);
388 389 390 391
  return code;
}

int32_t syncEndSnapshot(int64_t rid) {
S
Shengliang Guan 已提交
392
  SSyncNode* pSyncNode = syncNodeAcquire(rid);
393
  if (pSyncNode == NULL) {
394
    sError("sync end snapshot error");
395 396 397
    return -1;
  }

398 399 400 401
  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 已提交
402
    if (code != 0) {
403
      sNError(pSyncNode, "wal snapshot end error since:%s", terrstr());
S
Shengliang Guan 已提交
404
      syncNodeRelease(pSyncNode);
M
Minghao Li 已提交
405 406
      return -1;
    } else {
S
Shengliang Guan 已提交
407
      sNTrace(pSyncNode, "wal snapshot end, index:%" PRId64, atomic_load_64(&pSyncNode->snapshottingIndex));
M
Minghao Li 已提交
408 409
      atomic_store_64(&pSyncNode->snapshottingIndex, SYNC_INDEX_INVALID);
    }
410
  }
411

S
Shengliang Guan 已提交
412
  syncNodeRelease(pSyncNode);
413 414 415
  return code;
}

M
Minghao Li 已提交
416
int32_t syncStepDown(int64_t rid, SyncTerm newTerm) {
S
Shengliang Guan 已提交
417
  SSyncNode* pSyncNode = syncNodeAcquire(rid);
M
Minghao Li 已提交
418
  if (pSyncNode == NULL) {
419
    sError("sync step down error");
M
Minghao Li 已提交
420 421 422
    return -1;
  }

M
Minghao Li 已提交
423
  syncNodeStepDown(pSyncNode, newTerm);
S
Shengliang Guan 已提交
424
  syncNodeRelease(pSyncNode);
M
Minghao Li 已提交
425
  return 0;
M
Minghao Li 已提交
426 427
}

428
bool syncNodeIsReadyForRead(SSyncNode* pSyncNode) {
429
  if (pSyncNode == NULL) {
430
    terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
431
    sError("sync ready for read error");
432 433
    return false;
  }
M
Minghao Li 已提交
434

435 436 437 438 439 440
  if (pSyncNode->state != TAOS_SYNC_STATE_LEADER) {
    terrno = TSDB_CODE_SYN_NOT_LEADER;
    return false;
  }

  if (pSyncNode->restoreFinish) {
441
    return true;
M
Minghao Li 已提交
442 443
  }

444
  bool ready = false;
445 446 447
  if (!pSyncNode->pFsm->FpApplyQueueEmptyCb(pSyncNode->pFsm)) {
    // apply queue not empty
    ready = false;
M
Minghao Li 已提交
448

449 450 451 452 453 454 455 456 457 458 459 460 461
  } else {
    if (!pSyncNode->pLogStore->syncLogIsEmpty(pSyncNode->pLogStore)) {
      SyncIndex       lastIndex = pSyncNode->pLogStore->syncLogLastIndex(pSyncNode->pLogStore);
      SSyncRaftEntry* pEntry = NULL;
      SLRUCache*      pCache = pSyncNode->pLogStore->pCache;
      LRUHandle*      h = taosLRUCacheLookup(pCache, &lastIndex, sizeof(lastIndex));
      int32_t         code = 0;
      if (h) {
        pEntry = (SSyncRaftEntry*)taosLRUCacheValue(pCache, h);
        code = 0;

        pSyncNode->pLogStore->cacheHit++;
        sNTrace(pSyncNode, "hit cache index:%" PRId64 ", bytes:%u, %p", lastIndex, pEntry->bytes, pEntry);
M
Minghao Li 已提交
462

463 464 465
      } else {
        pSyncNode->pLogStore->cacheMiss++;
        sNTrace(pSyncNode, "miss cache index:%" PRId64, lastIndex);
M
Minghao Li 已提交
466

467 468
        code = pSyncNode->pLogStore->syncLogGetEntry(pSyncNode->pLogStore, lastIndex, &pEntry);
      }
469

470
      if (code == 0 && pEntry != NULL) {
S
Shengliang Guan 已提交
471
        if (pEntry->originalRpcType == TDMT_SYNC_NOOP && pEntry->term == pSyncNode->raftStore.currentTerm) {
472
          ready = true;
473
        }
474

475 476 477 478
        if (h) {
          taosLRUCacheRelease(pCache, h, false);
        } else {
          syncEntryDestroy(pEntry);
479
        }
480 481 482 483
      }
    }
  }

484
  if (!ready) {
485
    terrno = TSDB_CODE_SYN_RESTORING;
486
  }
487

488 489 490 491 492 493 494 495 496 497 498 499
  return ready;
}

bool syncIsReadyForRead(int64_t rid) {
  SSyncNode* pSyncNode = syncNodeAcquire(rid);
  if (pSyncNode == NULL) {
    sError("sync ready for read error");
    return false;
  }

  bool ready = syncNodeIsReadyForRead(pSyncNode);

500 501
  syncNodeRelease(pSyncNode);
  return ready;
M
Minghao Li 已提交
502
}
M
Minghao Li 已提交
503

504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525
bool syncSnapshotSending(int64_t rid) {
  SSyncNode* pSyncNode = syncNodeAcquire(rid);
  if (pSyncNode == NULL) {
    return false;
  }

  bool b = syncNodeSnapshotSending(pSyncNode);
  syncNodeRelease(pSyncNode);
  return b;
}

bool syncSnapshotRecving(int64_t rid) {
  SSyncNode* pSyncNode = syncNodeAcquire(rid);
  if (pSyncNode == NULL) {
    return false;
  }

  bool b = syncNodeSnapshotRecving(pSyncNode);
  syncNodeRelease(pSyncNode);
  return b;
}

M
Minghao Li 已提交
526 527
int32_t syncNodeLeaderTransfer(SSyncNode* pSyncNode) {
  if (pSyncNode->peersNum == 0) {
S
Shengliang Guan 已提交
528
    sDebug("vgId:%d, only one replica, cannot leader transfer", pSyncNode->vgId);
M
Minghao Li 已提交
529 530
    terrno = TSDB_CODE_SYN_ONE_REPLICA;
    return -1;
M
Minghao Li 已提交
531
  }
M
Minghao Li 已提交
532

533
  int32_t ret = 0;
534
  if (pSyncNode->state == TAOS_SYNC_STATE_LEADER && pSyncNode->replicaNum > 1) {
535
    SNodeInfo newLeader = (pSyncNode->peersNodeInfo)[0];
536 537 538 539 540 541 542
    if (pSyncNode->peersNum == 2) {
      SyncIndex matchIndex0 = syncIndexMgrGetIndex(pSyncNode->pMatchIndex, &(pSyncNode->peersId[0]));
      SyncIndex matchIndex1 = syncIndexMgrGetIndex(pSyncNode->pMatchIndex, &(pSyncNode->peersId[1]));
      if (matchIndex1 > matchIndex0) {
        newLeader = (pSyncNode->peersNodeInfo)[1];
      }
    }
543 544 545
    ret = syncNodeLeaderTransferTo(pSyncNode, newLeader);
  }

M
Minghao Li 已提交
546
  return ret;
M
Minghao Li 已提交
547 548
}

M
Minghao Li 已提交
549 550
int32_t syncNodeLeaderTransferTo(SSyncNode* pSyncNode, SNodeInfo newLeader) {
  if (pSyncNode->replicaNum == 1) {
S
Shengliang Guan 已提交
551
    sDebug("vgId:%d, only one replica, cannot leader transfer", pSyncNode->vgId);
M
Minghao Li 已提交
552 553
    terrno = TSDB_CODE_SYN_ONE_REPLICA;
    return -1;
M
Minghao Li 已提交
554
  }
555

S
Shengliang Guan 已提交
556
  sNTrace(pSyncNode, "begin leader transfer to %s:%u", newLeader.nodeFqdn, newLeader.nodePort);
M
Minghao Li 已提交
557

558 559 560 561
  SRpcMsg rpcMsg = {0};
  (void)syncBuildLeaderTransfer(&rpcMsg, pSyncNode->vgId);

  SyncLeaderTransfer* pMsg = rpcMsg.pCont;
562
  pMsg->newLeaderId.addr = SYNC_ADDR(&newLeader);
M
Minghao Li 已提交
563 564 565
  pMsg->newLeaderId.vgId = pSyncNode->vgId;
  pMsg->newNodeInfo = newLeader;

S
Shengliang Guan 已提交
566
  int32_t ret = syncNodePropose(pSyncNode, &rpcMsg, false, NULL);
S
Shengliang Guan 已提交
567 568
  rpcFreeCont(rpcMsg.pCont);
  return ret;
M
Minghao Li 已提交
569 570
}

571 572
SSyncState syncGetState(int64_t rid) {
  SSyncState state = {.state = TAOS_SYNC_STATE_ERROR};
M
Minghao Li 已提交
573

S
Shengliang Guan 已提交
574
  SSyncNode* pSyncNode = syncNodeAcquire(rid);
575 576 577
  if (pSyncNode != NULL) {
    state.state = pSyncNode->state;
    state.restored = pSyncNode->restoreFinish;
578 579 580 581 582
    if (pSyncNode->vgId != 1) {
      state.canRead = syncNodeIsReadyForRead(pSyncNode);
    } else {
      state.canRead = state.restored;
    }
583
    syncNodeRelease(pSyncNode);
M
Minghao Li 已提交
584 585
  }

586
  return state;
M
Minghao Li 已提交
587 588
}

589
#if 0
590 591 592 593 594
int32_t syncGetSnapshotByIndex(int64_t rid, SyncIndex index, SSnapshot* pSnapshot) {
  if (index < SYNC_INDEX_BEGIN) {
    return -1;
  }

S
Shengliang Guan 已提交
595
  SSyncNode* pSyncNode = syncNodeAcquire(rid);
596 597 598
  if (pSyncNode == NULL) {
    return -1;
  }
599
  ASSERT(rid == pSyncNode->rid);
600 601 602 603 604

  SSyncRaftEntry* pEntry = NULL;
  int32_t         code = pSyncNode->pLogStore->syncLogGetEntry(pSyncNode->pLogStore, index, &pEntry);
  if (code != 0) {
    if (pEntry != NULL) {
B
Benguang Zhao 已提交
605
      syncEntryDestroy(pEntry);
606
    }
S
Shengliang Guan 已提交
607
    syncNodeRelease(pSyncNode);
608 609
    return -1;
  }
610
  ASSERT(pEntry != NULL);
611 612 613 614 615 616

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

617
  syncEntryDestroy(pEntry);
S
Shengliang Guan 已提交
618
  syncNodeRelease(pSyncNode);
619 620 621
  return 0;
}

622
int32_t syncGetSnapshotMeta(int64_t rid, struct SSnapshotMeta* sMeta) {
S
Shengliang Guan 已提交
623
  SSyncNode* pSyncNode = syncNodeAcquire(rid);
624 625 626
  if (pSyncNode == NULL) {
    return -1;
  }
627
  ASSERT(rid == pSyncNode->rid);
628
  sMeta->lastConfigIndex = pSyncNode->raftCfg.lastConfigIndex;
629

630
  sTrace("vgId:%d, get snapshot meta, lastConfigIndex:%" PRId64, pSyncNode->vgId, pSyncNode->raftCfg.lastConfigIndex);
631

S
Shengliang Guan 已提交
632
  syncNodeRelease(pSyncNode);
633 634 635
  return 0;
}

636
int32_t syncGetSnapshotMetaByIndex(int64_t rid, SyncIndex snapshotIndex, struct SSnapshotMeta* sMeta) {
S
Shengliang Guan 已提交
637
  SSyncNode* pSyncNode = syncNodeAcquire(rid);
638 639 640
  if (pSyncNode == NULL) {
    return -1;
  }
641
  ASSERT(rid == pSyncNode->rid);
642

643 644
  ASSERT(pSyncNode->raftCfg.configIndexCount >= 1);
  SyncIndex lastIndex = (pSyncNode->raftCfg.configIndexArr)[0];
645

646 647 648 649
  for (int32_t i = 0; i < pSyncNode->raftCfg.configIndexCount; ++i) {
    if ((pSyncNode->raftCfg.configIndexArr)[i] > lastIndex &&
        (pSyncNode->raftCfg.configIndexArr)[i] <= snapshotIndex) {
      lastIndex = (pSyncNode->raftCfg.configIndexArr)[i];
650 651 652
    }
  }
  sMeta->lastConfigIndex = lastIndex;
653
  sTrace("vgId:%d, get snapshot meta by index:%" PRId64 " lcindex:%" PRId64, pSyncNode->vgId, snapshotIndex,
S
Shengliang Guan 已提交
654
         sMeta->lastConfigIndex);
655

S
Shengliang Guan 已提交
656
  syncNodeRelease(pSyncNode);
657 658
  return 0;
}
659
#endif
660

661
SyncIndex syncNodeGetSnapshotConfigIndex(SSyncNode* pSyncNode, SyncIndex snapshotLastApplyIndex) {
662 663
  ASSERT(pSyncNode->raftCfg.configIndexCount >= 1);
  SyncIndex lastIndex = (pSyncNode->raftCfg.configIndexArr)[0];
664

665 666 667 668
  for (int32_t i = 0; i < pSyncNode->raftCfg.configIndexCount; ++i) {
    if ((pSyncNode->raftCfg.configIndexArr)[i] > lastIndex &&
        (pSyncNode->raftCfg.configIndexArr)[i] <= snapshotLastApplyIndex) {
      lastIndex = (pSyncNode->raftCfg.configIndexArr)[i];
669 670
    }
  }
S
Shengliang Guan 已提交
671
  sTrace("vgId:%d, sync get last config index, index:%" PRId64 " lcindex:%" PRId64, pSyncNode->vgId,
S
Shengliang Guan 已提交
672
         snapshotLastApplyIndex, lastIndex);
673 674 675 676

  return lastIndex;
}

677 678
void syncGetRetryEpSet(int64_t rid, SEpSet* pEpSet) {
  pEpSet->numOfEps = 0;
M
Minghao Li 已提交
679

S
Shengliang Guan 已提交
680
  SSyncNode* pSyncNode = syncNodeAcquire(rid);
S
Shengliang Guan 已提交
681
  if (pSyncNode == NULL) return;
M
Minghao Li 已提交
682

683
  for (int32_t i = 0; i < pSyncNode->raftCfg.cfg.replicaNum; ++i) {
S
Shengliang Guan 已提交
684
    SEp* pEp = &pEpSet->eps[i];
685 686
    tstrncpy(pEp->fqdn, pSyncNode->raftCfg.cfg.nodeInfo[i].nodeFqdn, TSDB_FQDN_LEN);
    pEp->port = (pSyncNode->raftCfg.cfg.nodeInfo)[i].nodePort;
S
Shengliang Guan 已提交
687
    pEpSet->numOfEps++;
688
    sDebug("vgId:%d, sync get retry epset, index:%d %s:%d", pSyncNode->vgId, i, pEp->fqdn, pEp->port);
M
Minghao Li 已提交
689
  }
M
Minghao Li 已提交
690
  if (pEpSet->numOfEps > 0) {
691
    pEpSet->inUse = (pSyncNode->raftCfg.cfg.myIndex + 1) % pEpSet->numOfEps;
M
Minghao Li 已提交
692 693
  }

S
Shengliang Guan 已提交
694
  sInfo("vgId:%d, sync get retry epset numOfEps:%d inUse:%d", pSyncNode->vgId, pEpSet->numOfEps, pEpSet->inUse);
S
Shengliang Guan 已提交
695
  syncNodeRelease(pSyncNode);
M
Minghao Li 已提交
696 697
}

S
Shengliang Guan 已提交
698
int32_t syncPropose(int64_t rid, SRpcMsg* pMsg, bool isWeak, int64_t* seq) {
S
Shengliang Guan 已提交
699
  SSyncNode* pSyncNode = syncNodeAcquire(rid);
700
  if (pSyncNode == NULL) {
701
    sError("sync propose error");
M
Minghao Li 已提交
702
    return -1;
703
  }
704

S
Shengliang Guan 已提交
705
  int32_t ret = syncNodePropose(pSyncNode, pMsg, isWeak, seq);
S
Shengliang Guan 已提交
706
  syncNodeRelease(pSyncNode);
M
Minghao Li 已提交
707 708
  return ret;
}
M
Minghao Li 已提交
709

S
Shengliang Guan 已提交
710
int32_t syncNodePropose(SSyncNode* pSyncNode, SRpcMsg* pMsg, bool isWeak, int64_t* seq) {
S
Shengliang Guan 已提交
711 712
  if (pSyncNode->state != TAOS_SYNC_STATE_LEADER) {
    terrno = TSDB_CODE_SYN_NOT_LEADER;
S
Shengliang Guan 已提交
713
    sNError(pSyncNode, "sync propose not leader, type:%s", TMSG_INFO(pMsg->msgType));
S
Shengliang Guan 已提交
714 715
    return -1;
  }
716

S
Shengliang Guan 已提交
717 718 719 720 721 722 723
  // not restored, vnode enable
  if (!pSyncNode->restoreFinish && pSyncNode->vgId != 1) {
    terrno = TSDB_CODE_SYN_PROPOSE_NOT_READY;
    sNError(pSyncNode, "failed to sync propose since not ready, type:%s, last:%" PRId64 ", cmt:%" PRId64,
            TMSG_INFO(pMsg->msgType), syncNodeGetLastIndex(pSyncNode), pSyncNode->commitIndex);
    return -1;
  }
724

725
  // heartbeat timeout
726
  if (syncNodeHeartbeatReplyTimeout(pSyncNode)) {
727 728 729 730 731 732
    terrno = TSDB_CODE_SYN_PROPOSE_NOT_READY;
    sNError(pSyncNode, "failed to sync propose since hearbeat timeout, type:%s, last:%" PRId64 ", cmt:%" PRId64,
            TMSG_INFO(pMsg->msgType), syncNodeGetLastIndex(pSyncNode), pSyncNode->commitIndex);
    return -1;
  }

S
Shengliang Guan 已提交
733 734 735
  // optimized one replica
  if (syncNodeIsOptimizedOneReplica(pSyncNode, pMsg)) {
    SyncIndex retIndex;
736
    int32_t   code = syncNodeOnClientRequest(pSyncNode, pMsg, &retIndex);
S
Shengliang Guan 已提交
737 738
    if (code == 0) {
      pMsg->info.conn.applyIndex = retIndex;
S
Shengliang Guan 已提交
739
      pMsg->info.conn.applyTerm = pSyncNode->raftStore.currentTerm;
740 741 742
      sTrace("vgId:%d, propose optimized msg, index:%" PRId64 " type:%s", pSyncNode->vgId, retIndex,
             TMSG_INFO(pMsg->msgType));
      return 1;
M
Minghao Li 已提交
743
    } else {
S
Shengliang Guan 已提交
744
      terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
745
      sError("vgId:%d, failed to propose optimized msg, index:%" PRId64 " type:%s", pSyncNode->vgId, retIndex,
S
Shengliang Guan 已提交
746
             TMSG_INFO(pMsg->msgType));
747
      return -1;
748
    }
S
Shengliang Guan 已提交
749
  } else {
S
Shengliang Guan 已提交
750 751
    SRespStub stub = {.createTime = taosGetTimestampMs(), .rpcMsg = *pMsg};
    uint64_t  seqNum = syncRespMgrAdd(pSyncNode->pSyncRespMgr, &stub);
752
    SRpcMsg   rpcMsg = {0};
S
Shengliang Guan 已提交
753
    int32_t   code = syncBuildClientRequest(&rpcMsg, pMsg, seqNum, isWeak, pSyncNode->vgId);
754 755 756 757
    if (code != 0) {
      sError("vgId:%d, failed to propose msg while serialize since %s", pSyncNode->vgId, terrstr());
      (void)syncRespMgrDel(pSyncNode->pSyncRespMgr, seqNum);
      return -1;
M
Minghao Li 已提交
758
    }
759

760 761 762 763 764
    sNTrace(pSyncNode, "propose msg, type:%s", TMSG_INFO(pMsg->msgType));
    code = (*pSyncNode->syncEqMsg)(pSyncNode->msgcb, &rpcMsg);
    if (code != 0) {
      sError("vgId:%d, failed to propose msg while enqueue since %s", pSyncNode->vgId, terrstr());
      (void)syncRespMgrDel(pSyncNode->pSyncRespMgr, seqNum);
M
Minghao Li 已提交
765
    }
M
Minghao Li 已提交
766

S
Shengliang Guan 已提交
767
    if (seq != NULL) *seq = seqNum;
768
    return code;
M
Minghao Li 已提交
769
  }
M
Minghao Li 已提交
770 771
}

S
Shengliang Guan 已提交
772
static int32_t syncHbTimerInit(SSyncNode* pSyncNode, SSyncTimer* pSyncTimer, SRaftId destId) {
773 774 775 776 777
  pSyncTimer->pTimer = NULL;
  pSyncTimer->counter = 0;
  pSyncTimer->timerMS = pSyncNode->hbBaseLine;
  pSyncTimer->timerCb = syncNodeEqPeerHeartbeatTimer;
  pSyncTimer->destId = destId;
M
Minghao Li 已提交
778
  pSyncTimer->timeStamp = taosGetTimestampMs();
779 780 781 782
  atomic_store_64(&pSyncTimer->logicClock, 0);
  return 0;
}

S
Shengliang Guan 已提交
783
static int32_t syncHbTimerStart(SSyncNode* pSyncNode, SSyncTimer* pSyncTimer) {
784
  int32_t ret = 0;
S
Shengliang Guan 已提交
785
  int64_t tsNow = taosGetTimestampMs();
S
Shengliang Guan 已提交
786
  if (syncIsInit()) {
787 788 789 790 791 792
    SSyncHbTimerData* pData = syncHbTimerDataAcquire(pSyncTimer->hbDataRid);
    if (pData == NULL) {
      pData = taosMemoryMalloc(sizeof(SSyncHbTimerData));
      pData->rid = syncHbTimerDataAdd(pData);
    }
    pSyncTimer->hbDataRid = pData->rid;
S
Shengliang Guan 已提交
793
    pSyncTimer->timeStamp = tsNow;
794 795

    pData->syncNodeRid = pSyncNode->rid;
796 797 798
    pData->pTimer = pSyncTimer;
    pData->destId = pSyncTimer->destId;
    pData->logicClock = pSyncTimer->logicClock;
S
Shengliang Guan 已提交
799
    pData->execTime = tsNow + pSyncTimer->timerMS;
M
Minghao Li 已提交
800

801 802
    taosTmrReset(pSyncTimer->timerCb, pSyncTimer->timerMS / HEARTBEAT_TICK_NUM, (void*)(pData->rid),
                 syncEnv()->pTimerManager, &pSyncTimer->pTimer);
803 804 805 806 807 808
  } else {
    sError("vgId:%d, start ctrl hb timer error, sync env is stop", pSyncNode->vgId);
  }
  return ret;
}

S
Shengliang Guan 已提交
809
static int32_t syncHbTimerStop(SSyncNode* pSyncNode, SSyncTimer* pSyncTimer) {
810 811 812 813
  int32_t ret = 0;
  atomic_add_fetch_64(&pSyncTimer->logicClock, 1);
  taosTmrStop(pSyncTimer->pTimer);
  pSyncTimer->pTimer = NULL;
814 815
  syncHbTimerDataRemove(pSyncTimer->hbDataRid);
  pSyncTimer->hbDataRid = -1;
816 817 818
  return ret;
}

819
int32_t syncNodeLogStoreRestoreOnNeed(SSyncNode* pNode) {
S
Shengliang Guan 已提交
820 821 822
  ASSERTS(pNode->pLogStore != NULL, "log store not created");
  ASSERTS(pNode->pFsm != NULL, "pFsm not registered");
  ASSERTS(pNode->pFsm->FpGetSnapshotInfo != NULL, "FpGetSnapshotInfo not registered");
823 824 825
  SSnapshot snapshot = {0};
  pNode->pFsm->FpGetSnapshotInfo(pNode->pFsm, &snapshot);

826 827 828 829 830 831 832 833 834 835 836 837 838
  SyncIndex commitIndex = snapshot.lastApplyIndex;
  SyncIndex firstVer = pNode->pLogStore->syncLogBeginIndex(pNode->pLogStore);
  SyncIndex lastVer = pNode->pLogStore->syncLogLastIndex(pNode->pLogStore);
  if (lastVer < commitIndex || firstVer > commitIndex + 1) {
    if (pNode->pLogStore->syncLogRestoreFromSnapshot(pNode->pLogStore, commitIndex)) {
      sError("vgId:%d, failed to restore log store from snapshot since %s. lastVer: %" PRId64 ", snapshotVer: %" PRId64,
             pNode->vgId, terrstr(), lastVer, commitIndex);
      return -1;
    }
  }
  return 0;
}

M
Minghao Li 已提交
839
// open/close --------------
S
Shengliang Guan 已提交
840 841
SSyncNode* syncNodeOpen(SSyncInfo* pSyncInfo) {
  SSyncNode* pSyncNode = taosMemoryCalloc(1, sizeof(SSyncNode));
842 843 844 845
  if (pSyncNode == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    goto _error;
  }
M
Minghao Li 已提交
846

M
Minghao Li 已提交
847 848 849 850
  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());
851
      goto _error;
M
Minghao Li 已提交
852
    }
853
  }
M
Minghao Li 已提交
854

855 856 857
  memcpy(pSyncNode->path, pSyncInfo->path, sizeof(pSyncNode->path));
  snprintf(pSyncNode->raftStorePath, sizeof(pSyncNode->raftStorePath), "%s%sraft_store.json", pSyncInfo->path,
           TD_DIRSEP);
S
Shengliang Guan 已提交
858
  snprintf(pSyncNode->configPath, sizeof(pSyncNode->configPath), "%s%sraft_config.json", pSyncInfo->path, TD_DIRSEP);
859

860
  if (!taosCheckExistFile(pSyncNode->configPath)) {
M
Minghao Li 已提交
861
    // create a new raft config file
862
    sInfo("vgId:%d, create a new raft config file", pSyncNode->vgId);
863 864 865 866 867 868 869 870 871 872
    pSyncNode->raftCfg.isStandBy = pSyncInfo->isStandBy;
    pSyncNode->raftCfg.snapshotStrategy = pSyncInfo->snapshotStrategy;
    pSyncNode->raftCfg.lastConfigIndex = SYNC_INDEX_INVALID;
    pSyncNode->raftCfg.batchSize = pSyncInfo->batchSize;
    pSyncNode->raftCfg.cfg = pSyncInfo->syncCfg;
    pSyncNode->raftCfg.configIndexCount = 1;
    pSyncNode->raftCfg.configIndexArr[0] = -1;

    if (syncWriteCfgFile(pSyncNode) != 0) {
      sError("vgId:%d, failed to create sync cfg file", pSyncNode->vgId);
H
Hongze Cheng 已提交
873
      goto _error;
874
    }
875 876
  } else {
    // update syncCfg by raft_config.json
877 878
    if (syncReadCfgFile(pSyncNode) != 0) {
      sError("vgId:%d, failed to read sync cfg file", pSyncNode->vgId);
H
Hongze Cheng 已提交
879
      goto _error;
880
    }
S
Shengliang Guan 已提交
881

882
    if (pSyncInfo->syncCfg.replicaNum > 0 && syncIsConfigChanged(&pSyncNode->raftCfg.cfg, &pSyncInfo->syncCfg)) {
S
Shengliang Guan 已提交
883
      sInfo("vgId:%d, use sync config from input options and write to cfg file", pSyncNode->vgId);
884 885 886
      pSyncNode->raftCfg.cfg = pSyncInfo->syncCfg;
      if (syncWriteCfgFile(pSyncNode) != 0) {
        sError("vgId:%d, failed to write sync cfg file", pSyncNode->vgId);
S
Shengliang Guan 已提交
887 888
        goto _error;
      }
S
Shengliang Guan 已提交
889
    } else {
890 891
      sInfo("vgId:%d, use sync config from sync cfg file", pSyncNode->vgId);
      pSyncInfo->syncCfg = pSyncNode->raftCfg.cfg;
S
Shengliang Guan 已提交
892
    }
M
Minghao Li 已提交
893 894
  }

M
Minghao Li 已提交
895
  // init by SSyncInfo
M
Minghao Li 已提交
896
  pSyncNode->vgId = pSyncInfo->vgId;
S
Shengliang Guan 已提交
897
  SSyncCfg* pCfg = &pSyncNode->raftCfg.cfg;
S
Shengliang Guan 已提交
898
  sInfo("vgId:%d, start to open sync node, replica:%d selfIndex:%d", pSyncNode->vgId, pCfg->replicaNum, pCfg->myIndex);
S
Shengliang Guan 已提交
899 900
  for (int32_t i = 0; i < pCfg->replicaNum; ++i) {
    SNodeInfo* pNode = &pCfg->nodeInfo[i];
S
Shengliang Guan 已提交
901
    (void)tmsgUpdateDnodeInfo(&pNode->nodeId, &pNode->clusterId, pNode->nodeFqdn, &pNode->nodePort);
902 903
    sInfo("vgId:%d, index:%d ep:%s:%u dnode:%d cluster:%" PRId64, pSyncNode->vgId, i, pNode->nodeFqdn, pNode->nodePort,
          pNode->nodeId, pNode->clusterId);
S
Shengliang Guan 已提交
904 905
  }

M
Minghao Li 已提交
906
  pSyncNode->pWal = pSyncInfo->pWal;
S
Shengliang Guan 已提交
907
  pSyncNode->msgcb = pSyncInfo->msgcb;
S
Shengliang Guan 已提交
908 909 910
  pSyncNode->syncSendMSg = pSyncInfo->syncSendMSg;
  pSyncNode->syncEqMsg = pSyncInfo->syncEqMsg;
  pSyncNode->syncEqCtrlMsg = pSyncInfo->syncEqCtrlMsg;
M
Minghao Li 已提交
911

B
Benguang Zhao 已提交
912 913 914
  // create raft log ring buffer
  pSyncNode->pLogBuf = syncLogBufferCreate();
  if (pSyncNode->pLogBuf == NULL) {
915
    sError("failed to init sync log buffer since %s. vgId:%d", terrstr(), pSyncNode->vgId);
B
Benguang Zhao 已提交
916 917 918
    goto _error;
  }

M
Minghao Li 已提交
919
  // init internal
920
  pSyncNode->myNodeInfo = pSyncNode->raftCfg.cfg.nodeInfo[pSyncNode->raftCfg.cfg.myIndex];
921
  if (!syncUtilNodeInfo2RaftId(&pSyncNode->myNodeInfo, pSyncNode->vgId, &pSyncNode->myRaftId)) {
S
Shengliang Guan 已提交
922
    sError("vgId:%d, failed to determine my raft member id", pSyncNode->vgId);
H
Hongze Cheng 已提交
923
    goto _error;
924
  }
M
Minghao Li 已提交
925

M
Minghao Li 已提交
926
  // init peersNum, peers, peersId
927
  pSyncNode->peersNum = pSyncNode->raftCfg.cfg.replicaNum - 1;
S
Shengliang Guan 已提交
928
  int32_t j = 0;
929 930 931 932
  for (int32_t i = 0; i < pSyncNode->raftCfg.cfg.replicaNum; ++i) {
    if (i != pSyncNode->raftCfg.cfg.myIndex) {
      pSyncNode->peersNodeInfo[j] = pSyncNode->raftCfg.cfg.nodeInfo[i];
      syncUtilNodeInfo2EpSet(&pSyncNode->peersNodeInfo[j], &pSyncNode->peersEpset[j]);
M
Minghao Li 已提交
933 934 935
      j++;
    }
  }
S
Shengliang Guan 已提交
936
  for (int32_t i = 0; i < pSyncNode->peersNum; ++i) {
937
    if (!syncUtilNodeInfo2RaftId(&pSyncNode->peersNodeInfo[i], pSyncNode->vgId, &pSyncNode->peersId[i])) {
S
Shengliang Guan 已提交
938
      sError("vgId:%d, failed to determine raft member id, peer:%d", pSyncNode->vgId, i);
H
Hongze Cheng 已提交
939
      goto _error;
940
    }
M
Minghao Li 已提交
941
  }
M
Minghao Li 已提交
942

M
Minghao Li 已提交
943
  // init replicaNum, replicasId
944 945 946
  pSyncNode->replicaNum = pSyncNode->raftCfg.cfg.replicaNum;
  for (int32_t i = 0; i < pSyncNode->raftCfg.cfg.replicaNum; ++i) {
    if (!syncUtilNodeInfo2RaftId(&pSyncNode->raftCfg.cfg.nodeInfo[i], pSyncNode->vgId, &pSyncNode->replicasId[i])) {
S
Shengliang Guan 已提交
947
      sError("vgId:%d, failed to determine raft member id, replica:%d", pSyncNode->vgId, i);
H
Hongze Cheng 已提交
948
      goto _error;
949
    }
M
Minghao Li 已提交
950 951
  }

M
Minghao Li 已提交
952
  // init raft algorithm
M
Minghao Li 已提交
953
  pSyncNode->pFsm = pSyncInfo->pFsm;
954
  pSyncInfo->pFsm = NULL;
955
  pSyncNode->quorum = syncUtilQuorum(pSyncNode->raftCfg.cfg.replicaNum);
M
Minghao Li 已提交
956 957
  pSyncNode->leaderCache = EMPTY_RAFT_ID;

M
Minghao Li 已提交
958
  // init life cycle outside
M
Minghao Li 已提交
959

M
Minghao Li 已提交
960 961 962 963 964 965 966 967 968 969 970 971 972 973 974 975 976 977 978 979 980 981 982 983
  // 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 已提交
984
  // init TLA+ server vars
M
syncInt  
Minghao Li 已提交
985
  pSyncNode->state = TAOS_SYNC_STATE_FOLLOWER;
S
Shengliang Guan 已提交
986
  if (raftStoreReadFile(pSyncNode) != 0) {
S
Shengliang Guan 已提交
987
    sError("vgId:%d, failed to open raft store at path %s", pSyncNode->vgId, pSyncNode->raftStorePath);
988 989
    goto _error;
  }
M
Minghao Li 已提交
990

M
Minghao Li 已提交
991
  // init TLA+ candidate vars
M
Minghao Li 已提交
992
  pSyncNode->pVotesGranted = voteGrantedCreate(pSyncNode);
993
  if (pSyncNode->pVotesGranted == NULL) {
S
Shengliang Guan 已提交
994
    sError("vgId:%d, failed to create VotesGranted", pSyncNode->vgId);
995 996
    goto _error;
  }
M
Minghao Li 已提交
997
  pSyncNode->pVotesRespond = votesRespondCreate(pSyncNode);
998
  if (pSyncNode->pVotesRespond == NULL) {
S
Shengliang Guan 已提交
999
    sError("vgId:%d, failed to create VotesRespond", pSyncNode->vgId);
1000 1001
    goto _error;
  }
M
Minghao Li 已提交
1002

M
Minghao Li 已提交
1003 1004
  // init TLA+ leader vars
  pSyncNode->pNextIndex = syncIndexMgrCreate(pSyncNode);
1005
  if (pSyncNode->pNextIndex == NULL) {
S
Shengliang Guan 已提交
1006
    sError("vgId:%d, failed to create SyncIndexMgr", pSyncNode->vgId);
1007 1008
    goto _error;
  }
M
Minghao Li 已提交
1009
  pSyncNode->pMatchIndex = syncIndexMgrCreate(pSyncNode);
1010
  if (pSyncNode->pMatchIndex == NULL) {
S
Shengliang Guan 已提交
1011
    sError("vgId:%d, failed to create SyncIndexMgr", pSyncNode->vgId);
1012 1013
    goto _error;
  }
M
Minghao Li 已提交
1014 1015 1016

  // init TLA+ log vars
  pSyncNode->pLogStore = logStoreCreate(pSyncNode);
1017
  if (pSyncNode->pLogStore == NULL) {
S
Shengliang Guan 已提交
1018
    sError("vgId:%d, failed to create SyncLogStore", pSyncNode->vgId);
1019 1020
    goto _error;
  }
1021 1022 1023 1024

  SyncIndex commitIndex = SYNC_INDEX_INVALID;
  if (pSyncNode->pFsm != NULL && pSyncNode->pFsm->FpGetSnapshotInfo != NULL) {
    SSnapshot snapshot = {0};
1025
    pSyncNode->pFsm->FpGetSnapshotInfo(pSyncNode->pFsm, &snapshot);
1026 1027
    if (snapshot.lastApplyIndex > commitIndex) {
      commitIndex = snapshot.lastApplyIndex;
S
Shengliang Guan 已提交
1028
      sNTrace(pSyncNode, "reset commit index by snapshot");
1029 1030 1031
    }
  }
  pSyncNode->commitIndex = commitIndex;
1032
  sInfo("vgId:%d, sync node commitIndex initialized as %" PRId64, pSyncNode->vgId, pSyncNode->commitIndex);
M
Minghao Li 已提交
1033

1034 1035 1036
  if (syncNodeLogStoreRestoreOnNeed(pSyncNode) < 0) {
    goto _error;
  }
M
Minghao Li 已提交
1037 1038
  // timer ms init
  pSyncNode->pingBaseLine = PING_TIMER_MS;
1039 1040
  pSyncNode->electBaseLine = tsElectInterval;
  pSyncNode->hbBaseLine = tsHeartbeatInterval;
M
Minghao Li 已提交
1041

M
Minghao Li 已提交
1042
  // init ping timer
M
Minghao Li 已提交
1043
  pSyncNode->pPingTimer = NULL;
M
Minghao Li 已提交
1044
  pSyncNode->pingTimerMS = pSyncNode->pingBaseLine;
M
Minghao Li 已提交
1045 1046
  atomic_store_64(&pSyncNode->pingTimerLogicClock, 0);
  atomic_store_64(&pSyncNode->pingTimerLogicClockUser, 0);
M
Minghao Li 已提交
1047
  pSyncNode->FpPingTimerCB = syncNodeEqPingTimer;
M
Minghao Li 已提交
1048
  pSyncNode->pingTimerCounter = 0;
M
Minghao Li 已提交
1049

M
Minghao Li 已提交
1050 1051
  // init elect timer
  pSyncNode->pElectTimer = NULL;
M
Minghao Li 已提交
1052
  pSyncNode->electTimerMS = syncUtilElectRandomMS(pSyncNode->electBaseLine, 2 * pSyncNode->electBaseLine);
M
Minghao Li 已提交
1053
  atomic_store_64(&pSyncNode->electTimerLogicClock, 0);
M
Minghao Li 已提交
1054
  pSyncNode->FpElectTimerCB = syncNodeEqElectTimer;
M
Minghao Li 已提交
1055 1056 1057 1058
  pSyncNode->electTimerCounter = 0;

  // init heartbeat timer
  pSyncNode->pHeartbeatTimer = NULL;
M
Minghao Li 已提交
1059
  pSyncNode->heartbeatTimerMS = pSyncNode->hbBaseLine;
M
Minghao Li 已提交
1060 1061
  atomic_store_64(&pSyncNode->heartbeatTimerLogicClock, 0);
  atomic_store_64(&pSyncNode->heartbeatTimerLogicClockUser, 0);
M
Minghao Li 已提交
1062
  pSyncNode->FpHeartbeatTimerCB = syncNodeEqHeartbeatTimer;
M
Minghao Li 已提交
1063 1064
  pSyncNode->heartbeatTimerCounter = 0;

1065 1066 1067 1068 1069
  // 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 已提交
1070
  // tools
M
Minghao Li 已提交
1071
  pSyncNode->pSyncRespMgr = syncRespMgrCreate(pSyncNode, SYNC_RESP_TTL_MS);
1072
  if (pSyncNode->pSyncRespMgr == NULL) {
S
Shengliang Guan 已提交
1073
    sError("vgId:%d, failed to create SyncRespMgr", pSyncNode->vgId);
1074 1075
    goto _error;
  }
M
Minghao Li 已提交
1076

1077 1078
  // restore state
  pSyncNode->restoreFinish = false;
1079

M
Minghao Li 已提交
1080
  // snapshot senders
S
Shengliang Guan 已提交
1081
  for (int32_t i = 0; i < TSDB_MAX_REPLICA; ++i) {
M
Minghao Li 已提交
1082
    SSyncSnapshotSender* pSender = snapshotSenderCreate(pSyncNode, i);
1083 1084 1085 1086
    if (pSender == NULL) return NULL;

    pSyncNode->senders[i] = pSender;
    sSDebug(pSender, "snapshot sender create while open sync node, data:%p", pSender);
M
Minghao Li 已提交
1087 1088 1089
  }

  // snapshot receivers
1090
  pSyncNode->pNewNodeReceiver = snapshotReceiverCreate(pSyncNode, EMPTY_RAFT_ID);
1091 1092 1093
  if (pSyncNode->pNewNodeReceiver == NULL) return NULL;
  sRDebug(pSyncNode->pNewNodeReceiver, "snapshot receiver create while open sync node, data:%p",
          pSyncNode->pNewNodeReceiver);
M
Minghao Li 已提交
1094

M
Minghao Li 已提交
1095 1096 1097
  // is config changing
  pSyncNode->changing = false;

B
Benguang Zhao 已提交
1098 1099 1100
  // replication mgr
  syncNodeLogReplMgrInit(pSyncNode);

M
Minghao Li 已提交
1101 1102 1103
  // peer state
  syncNodePeerStateInit(pSyncNode);

B
Benguang Zhao 已提交
1104
  //
M
Minghao Li 已提交
1105 1106 1107
  // min match index
  pSyncNode->minMatchIndex = SYNC_INDEX_INVALID;

M
Minghao Li 已提交
1108
  // start in syncNodeStart
M
Minghao Li 已提交
1109
  // start raft
M
Minghao Li 已提交
1110
  // syncNodeBecomeFollower(pSyncNode);
M
Minghao Li 已提交
1111

M
Minghao Li 已提交
1112 1113
  int64_t timeNow = taosGetTimestampMs();
  pSyncNode->startTime = timeNow;
1114
  pSyncNode->leaderTime = timeNow;
M
Minghao Li 已提交
1115 1116
  pSyncNode->lastReplicateTime = timeNow;

1117 1118 1119
  // snapshotting
  atomic_store_64(&pSyncNode->snapshottingIndex, SYNC_INDEX_INVALID);

B
Benguang Zhao 已提交
1120 1121
  // init log buffer
  if (syncLogBufferInit(pSyncNode->pLogBuf, pSyncNode) < 0) {
1122
    sError("vgId:%d, failed to init sync log buffer since %s", pSyncNode->vgId, terrstr());
1123
    goto _error;
B
Benguang Zhao 已提交
1124 1125
  }

1126
  pSyncNode->isStart = true;
1127 1128 1129
  pSyncNode->electNum = 0;
  pSyncNode->becomeLeaderNum = 0;
  pSyncNode->configChangeNum = 0;
1130 1131
  pSyncNode->hbSlowNum = 0;
  pSyncNode->hbrSlowNum = 0;
M
Minghao Li 已提交
1132
  pSyncNode->tmrRoutineNum = 0;
1133

1134 1135
  sNInfo(pSyncNode, "sync open, node:%p electInterval:%d heartbeatInterval:%d heartbeatTimeout:%d", pSyncNode,
         tsElectInterval, tsHeartbeatInterval, tsHeartbeatTimeout);
M
Minghao Li 已提交
1136
  return pSyncNode;
1137 1138 1139

_error:
  if (pSyncInfo->pFsm) {
H
Hongze Cheng 已提交
1140 1141
    taosMemoryFree(pSyncInfo->pFsm);
    pSyncInfo->pFsm = NULL;
1142 1143 1144 1145
  }
  syncNodeClose(pSyncNode);
  pSyncNode = NULL;
  return NULL;
M
Minghao Li 已提交
1146 1147
}

M
Minghao Li 已提交
1148 1149
void syncNodeMaybeUpdateCommitBySnapshot(SSyncNode* pSyncNode) {
  if (pSyncNode->pFsm != NULL && pSyncNode->pFsm->FpGetSnapshotInfo != NULL) {
1150 1151
    SSnapshot snapshot = {0};
    pSyncNode->pFsm->FpGetSnapshotInfo(pSyncNode->pFsm, &snapshot);
M
Minghao Li 已提交
1152 1153 1154 1155 1156 1157
    if (snapshot.lastApplyIndex > pSyncNode->commitIndex) {
      pSyncNode->commitIndex = snapshot.lastApplyIndex;
    }
  }
}

B
Benguang Zhao 已提交
1158
int32_t syncNodeRestore(SSyncNode* pSyncNode) {
S
Shengliang Guan 已提交
1159 1160
  ASSERTS(pSyncNode->pLogStore != NULL, "log store not created");
  ASSERTS(pSyncNode->pLogBuf != NULL, "ring log buffer not created");
B
Benguang Zhao 已提交
1161 1162 1163 1164

  SyncIndex lastVer = pSyncNode->pLogStore->syncLogLastIndex(pSyncNode->pLogStore);
  SyncIndex commitIndex = pSyncNode->pLogStore->syncLogCommitIndex(pSyncNode->pLogStore);
  SyncIndex endIndex = pSyncNode->pLogBuf->endIndex;
1165 1166 1167 1168 1169 1170
  if (lastVer != -1 && endIndex != lastVer + 1) {
    terrno = TSDB_CODE_WAL_LOG_INCOMPLETE;
    sError("vgId:%d, failed to restore sync node since %s. expected lastLogIndex: %" PRId64 ", lastVer: %" PRId64 "",
           pSyncNode->vgId, terrstr(), endIndex - 1, lastVer);
    return -1;
  }
B
Benguang Zhao 已提交
1171

1172
  ASSERT(endIndex == lastVer + 1);
1173 1174
  pSyncNode->commitIndex = TMAX(pSyncNode->commitIndex, commitIndex);
  sInfo("vgId:%d, restore sync until commitIndex:%" PRId64, pSyncNode->vgId, pSyncNode->commitIndex);
B
Benguang Zhao 已提交
1175

1176
  if (syncLogBufferCommit(pSyncNode->pLogBuf, pSyncNode, pSyncNode->commitIndex) < 0) {
B
Benguang Zhao 已提交
1177 1178 1179 1180 1181 1182 1183 1184 1185
    return -1;
  }

  return 0;
}

int32_t syncNodeStart(SSyncNode* pSyncNode) {
  // start raft
  if (pSyncNode->replicaNum == 1) {
S
Shengliang Guan 已提交
1186
    raftStoreNextTerm(pSyncNode);
B
Benguang Zhao 已提交
1187 1188 1189 1190 1191 1192 1193 1194 1195 1196
    syncNodeBecomeLeader(pSyncNode, "one replica start");

    // Raft 3.6.2 Committing entries from previous terms
    syncNodeAppendNoop(pSyncNode);
  } else {
    syncNodeBecomeFollower(pSyncNode, "first start");
  }

  int32_t ret = 0;
  ret = syncNodeStartPingTimer(pSyncNode);
1197
  ASSERT(ret == 0);
B
Benguang Zhao 已提交
1198 1199 1200 1201
  return ret;
}

void syncNodeStartOld(SSyncNode* pSyncNode) {
M
Minghao Li 已提交
1202
  // start raft
1203
  if (pSyncNode->replicaNum == 1) {
S
Shengliang Guan 已提交
1204
    raftStoreNextTerm(pSyncNode);
1205
    syncNodeBecomeLeader(pSyncNode, "one replica start");
M
format  
Minghao Li 已提交
1206

1207
    // Raft 3.6.2 Committing entries from previous terms
1208 1209
    syncNodeAppendNoop(pSyncNode);
    syncMaybeAdvanceCommitIndex(pSyncNode);
M
Minghao Li 已提交
1210

M
Minghao Li 已提交
1211 1212
  } else {
    syncNodeBecomeFollower(pSyncNode, "first start");
1213 1214
  }

1215 1216
  int32_t ret = 0;
  ret = syncNodeStartPingTimer(pSyncNode);
1217
  ASSERT(ret == 0);
M
Minghao Li 已提交
1218 1219
}

B
Benguang Zhao 已提交
1220
int32_t syncNodeStartStandBy(SSyncNode* pSyncNode) {
M
Minghao Li 已提交
1221 1222 1223 1224 1225 1226 1227
  // 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);
1228
  ASSERT(ret == 0);
1229

1230 1231
  ret = 0;
  ret = syncNodeStartPingTimer(pSyncNode);
1232
  ASSERT(ret == 0);
B
Benguang Zhao 已提交
1233
  return ret;
M
Minghao Li 已提交
1234 1235
}

M
Minghao Li 已提交
1236
void syncNodePreClose(SSyncNode* pSyncNode) {
1237 1238 1239 1240
  if (pSyncNode != NULL && pSyncNode->pFsm != NULL && pSyncNode->pFsm->FpApplyQueueItems != NULL) {
    while (1) {
      int32_t aqItems = pSyncNode->pFsm->FpApplyQueueItems(pSyncNode->pFsm);
      sTrace("vgId:%d, pre close, %d items in apply queue", pSyncNode->vgId, aqItems);
1241
      if (aqItems == 0 || aqItems == -1) {
1242 1243 1244 1245 1246 1247
        break;
      }
      taosMsleep(20);
    }
  }

1248
#if 0
1249 1250
  if (pSyncNode->pNewNodeReceiver != NULL) {
    if (snapshotReceiverIsStart(pSyncNode->pNewNodeReceiver)) {
S
Shengliang Guan 已提交
1251
      snapshotReceiverStop(pSyncNode->pNewNodeReceiver);
1252 1253
    }

1254 1255
    sDebug("vgId:%d, snapshot receiver destroy while preclose sync node, data:%p", pSyncNode->vgId,
           pSyncNode->pNewNodeReceiver);
1256 1257 1258
    snapshotReceiverDestroy(pSyncNode->pNewNodeReceiver);
    pSyncNode->pNewNodeReceiver = NULL;
  }
1259
#endif
1260

M
Minghao Li 已提交
1261 1262 1263 1264 1265
  // stop elect timer
  syncNodeStopElectTimer(pSyncNode);

  // stop heartbeat timer
  syncNodeStopHeartbeatTimer(pSyncNode);
1266 1267 1268

  // clean rsp
  syncRespCleanRsp(pSyncNode->pSyncRespMgr);
M
Minghao Li 已提交
1269 1270
}

1271 1272 1273
void syncNodePostClose(SSyncNode* pSyncNode) {
  if (pSyncNode->pNewNodeReceiver != NULL) {
    if (snapshotReceiverIsStart(pSyncNode->pNewNodeReceiver)) {
S
Shengliang Guan 已提交
1274
      snapshotReceiverStop(pSyncNode->pNewNodeReceiver);
1275 1276 1277 1278 1279 1280 1281
    }

    sDebug("vgId:%d, snapshot receiver destroy while preclose sync node, data:%p", pSyncNode->vgId,
           pSyncNode->pNewNodeReceiver);
    snapshotReceiverDestroy(pSyncNode->pNewNodeReceiver);
    pSyncNode->pNewNodeReceiver = NULL;
  }
M
Minghao Li 已提交
1282 1283
}

1284
void syncHbTimerDataFree(SSyncHbTimerData* pData) { taosMemoryFree(pData); }
M
Minghao Li 已提交
1285

M
Minghao Li 已提交
1286
void syncNodeClose(SSyncNode* pSyncNode) {
S
Shengliang Guan 已提交
1287
  if (pSyncNode == NULL) return;
1288
  sNInfo(pSyncNode, "sync close, node:%p", pSyncNode);
M
Minghao Li 已提交
1289

B
Benguang Zhao 已提交
1290
  syncNodeLogReplMgrDestroy(pSyncNode);
M
Minghao Li 已提交
1291
  syncRespMgrDestroy(pSyncNode->pSyncRespMgr);
1292
  pSyncNode->pSyncRespMgr = NULL;
M
Minghao Li 已提交
1293
  voteGrantedDestroy(pSyncNode->pVotesGranted);
1294
  pSyncNode->pVotesGranted = NULL;
M
Minghao Li 已提交
1295
  votesRespondDestory(pSyncNode->pVotesRespond);
1296
  pSyncNode->pVotesRespond = NULL;
M
Minghao Li 已提交
1297
  syncIndexMgrDestroy(pSyncNode->pNextIndex);
1298
  pSyncNode->pNextIndex = NULL;
M
Minghao Li 已提交
1299
  syncIndexMgrDestroy(pSyncNode->pMatchIndex);
1300
  pSyncNode->pMatchIndex = NULL;
M
Minghao Li 已提交
1301
  logStoreDestory(pSyncNode->pLogStore);
1302
  pSyncNode->pLogStore = NULL;
B
Benguang Zhao 已提交
1303 1304
  syncLogBufferDestroy(pSyncNode->pLogBuf);
  pSyncNode->pLogBuf = NULL;
M
Minghao Li 已提交
1305 1306 1307 1308 1309

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

S
Shengliang Guan 已提交
1310
  for (int32_t i = 0; i < TSDB_MAX_REPLICA; ++i) {
1311 1312
    if (pSyncNode->senders[i] != NULL) {
      sDebug("vgId:%d, snapshot sender destroy while close, data:%p", pSyncNode->vgId, pSyncNode->senders[i]);
1313

1314 1315
      if (snapshotSenderIsStart(pSyncNode->senders[i])) {
        snapshotSenderStop(pSyncNode->senders[i], false);
1316 1317
      }

1318 1319
      snapshotSenderDestroy(pSyncNode->senders[i]);
      pSyncNode->senders[i] = NULL;
M
Minghao Li 已提交
1320 1321 1322
    }
  }

M
Minghao Li 已提交
1323
  if (pSyncNode->pNewNodeReceiver != NULL) {
1324
    if (snapshotReceiverIsStart(pSyncNode->pNewNodeReceiver)) {
S
Shengliang Guan 已提交
1325
      snapshotReceiverStop(pSyncNode->pNewNodeReceiver);
1326 1327
    }

1328
    sDebug("vgId:%d, snapshot receiver destroy while close, data:%p", pSyncNode->vgId, pSyncNode->pNewNodeReceiver);
M
Minghao Li 已提交
1329 1330 1331 1332
    snapshotReceiverDestroy(pSyncNode->pNewNodeReceiver);
    pSyncNode->pNewNodeReceiver = NULL;
  }

1333 1334 1335 1336
  if (pSyncNode->pFsm != NULL) {
    taosMemoryFree(pSyncNode->pFsm);
  }

1337
  taosMemoryFree(pSyncNode);
M
Minghao Li 已提交
1338 1339
}

1340
ESyncStrategy syncNodeStrategy(SSyncNode* pSyncNode) { return pSyncNode->raftCfg.snapshotStrategy; }
M
Minghao Li 已提交
1341

M
Minghao Li 已提交
1342 1343 1344
// timer control --------------
int32_t syncNodeStartPingTimer(SSyncNode* pSyncNode) {
  int32_t ret = 0;
S
Shengliang Guan 已提交
1345 1346
  if (syncIsInit()) {
    taosTmrReset(pSyncNode->FpPingTimerCB, pSyncNode->pingTimerMS, pSyncNode, syncEnv()->pTimerManager,
1347 1348 1349
                 &pSyncNode->pPingTimer);
    atomic_store_64(&pSyncNode->pingTimerLogicClock, pSyncNode->pingTimerLogicClockUser);
  } else {
M
Minghao Li 已提交
1350
    sError("vgId:%d, start ping timer error, sync env is stop", pSyncNode->vgId);
1351
  }
M
Minghao Li 已提交
1352 1353 1354 1355 1356 1357 1358 1359 1360 1361 1362 1363 1364
  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;
S
Shengliang Guan 已提交
1365
  if (syncIsInit()) {
1366
    pSyncNode->electTimerMS = ms;
S
Shengliang Guan 已提交
1367

1368 1369 1370 1371 1372
    int64_t execTime = taosGetTimestampMs() + ms;
    atomic_store_64(&(pSyncNode->electTimerParam.executeTime), execTime);
    atomic_store_64(&(pSyncNode->electTimerParam.logicClock), pSyncNode->electTimerLogicClock);
    pSyncNode->electTimerParam.pSyncNode = pSyncNode;
    pSyncNode->electTimerParam.pData = NULL;
S
Shengliang Guan 已提交
1373

M
Minghao Li 已提交
1374
    taosTmrReset(pSyncNode->FpElectTimerCB, pSyncNode->electTimerMS, (void*)(pSyncNode->rid), syncEnv()->pTimerManager,
1375
                 &pSyncNode->pElectTimer);
1376

1377
  } else {
M
Minghao Li 已提交
1378
    sError("vgId:%d, start elect timer error, sync env is stop", pSyncNode->vgId);
1379
  }
M
Minghao Li 已提交
1380 1381 1382 1383 1384
  return ret;
}

int32_t syncNodeStopElectTimer(SSyncNode* pSyncNode) {
  int32_t ret = 0;
M
Minghao Li 已提交
1385
  atomic_add_fetch_64(&pSyncNode->electTimerLogicClock, 1);
M
Minghao Li 已提交
1386 1387
  taosTmrStop(pSyncNode->pElectTimer);
  pSyncNode->pElectTimer = NULL;
1388

M
Minghao Li 已提交
1389 1390 1391 1392 1393 1394 1395 1396 1397 1398
  return ret;
}

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

1399
void syncNodeResetElectTimer(SSyncNode* pSyncNode) {
M
Minghao Li 已提交
1400 1401
  int32_t electMS;

1402
  if (pSyncNode->raftCfg.isStandBy) {
M
Minghao Li 已提交
1403 1404 1405 1406
    electMS = TIMER_MAX_MS;
  } else {
    electMS = syncUtilElectRandomMS(pSyncNode->electBaseLine, 2 * pSyncNode->electBaseLine);
  }
1407 1408

  (void)syncNodeRestartElectTimer(pSyncNode, electMS);
1409

S
Shengliang Guan 已提交
1410 1411
  sNTrace(pSyncNode, "reset elect timer, min:%d, max:%d, ms:%d", pSyncNode->electBaseLine, 2 * pSyncNode->electBaseLine,
          electMS);
M
Minghao Li 已提交
1412 1413
}

M
Minghao Li 已提交
1414
static int32_t syncNodeDoStartHeartbeatTimer(SSyncNode* pSyncNode) {
M
Minghao Li 已提交
1415
  int32_t ret = 0;
S
Shengliang Guan 已提交
1416 1417
  if (syncIsInit()) {
    taosTmrReset(pSyncNode->FpHeartbeatTimerCB, pSyncNode->heartbeatTimerMS, pSyncNode, syncEnv()->pTimerManager,
1418 1419 1420
                 &pSyncNode->pHeartbeatTimer);
    atomic_store_64(&pSyncNode->heartbeatTimerLogicClock, pSyncNode->heartbeatTimerLogicClockUser);
  } else {
M
Minghao Li 已提交
1421
    sError("vgId:%d, start heartbeat timer error, sync env is stop", pSyncNode->vgId);
1422
  }
1423

S
Shengliang Guan 已提交
1424
  sNTrace(pSyncNode, "start heartbeat timer, ms:%d", pSyncNode->heartbeatTimerMS);
M
Minghao Li 已提交
1425 1426 1427
  return ret;
}

M
Minghao Li 已提交
1428
int32_t syncNodeStartHeartbeatTimer(SSyncNode* pSyncNode) {
1429
  int32_t ret = 0;
M
Minghao Li 已提交
1430

1431
#if 0
M
Minghao Li 已提交
1432
  pSyncNode->heartbeatTimerMS = pSyncNode->hbBaseLine;
1433 1434
  ret = syncNodeDoStartHeartbeatTimer(pSyncNode);
#endif
1435

S
Shengliang Guan 已提交
1436
  for (int32_t i = 0; i < pSyncNode->peersNum; ++i) {
1437
    SSyncTimer* pSyncTimer = syncNodeGetHbTimer(pSyncNode, &(pSyncNode->peersId[i]));
M
Minghao Li 已提交
1438 1439 1440
    if (pSyncTimer != NULL) {
      syncHbTimerStart(pSyncNode, pSyncTimer);
    }
1441
  }
1442

M
Minghao Li 已提交
1443 1444 1445
  return ret;
}

M
Minghao Li 已提交
1446 1447
int32_t syncNodeStopHeartbeatTimer(SSyncNode* pSyncNode) {
  int32_t ret = 0;
1448 1449

#if 0
M
Minghao Li 已提交
1450 1451 1452
  atomic_add_fetch_64(&pSyncNode->heartbeatTimerLogicClockUser, 1);
  taosTmrStop(pSyncNode->pHeartbeatTimer);
  pSyncNode->pHeartbeatTimer = NULL;
1453
#endif
1454

S
Shengliang Guan 已提交
1455
  for (int32_t i = 0; i < pSyncNode->peersNum; ++i) {
1456
    SSyncTimer* pSyncTimer = syncNodeGetHbTimer(pSyncNode, &(pSyncNode->peersId[i]));
M
Minghao Li 已提交
1457 1458 1459
    if (pSyncTimer != NULL) {
      syncHbTimerStop(pSyncNode, pSyncTimer);
    }
1460
  }
1461

M
Minghao Li 已提交
1462 1463 1464
  return ret;
}

1465 1466 1467 1468 1469 1470
int32_t syncNodeRestartHeartbeatTimer(SSyncNode* pSyncNode) {
  syncNodeStopHeartbeatTimer(pSyncNode);
  syncNodeStartHeartbeatTimer(pSyncNode);
  return 0;
}

1471 1472 1473 1474 1475 1476 1477 1478
int32_t syncNodeSendMsgById(const SRaftId* destRaftId, SSyncNode* pNode, SRpcMsg* pMsg) {
  SEpSet* epSet = NULL;
  for (int32_t i = 0; i < pNode->peersNum; ++i) {
    if (destRaftId->addr == pNode->peersId[i].addr) {
      epSet = &pNode->peersEpset[i];
      break;
    }
  }
1479

S
Shengliang Guan 已提交
1480
  int32_t code = -1;
1481
  if (pNode->syncSendMSg != NULL && epSet != NULL) {
M
Minghao Li 已提交
1482
    syncUtilMsgHtoN(pMsg->pCont);
1483
    pMsg->info.noResp = 1;
S
Shengliang Guan 已提交
1484 1485 1486 1487 1488 1489
    code = pNode->syncSendMSg(epSet, pMsg);
  }

  if (code < 0) {
    sError("vgId:%d, sync send msg by id error, epset:%p dnode:%d addr:%" PRId64 " err:0x%x", pNode->vgId, epSet,
           DID(destRaftId), destRaftId->addr, terrno);
S
Shengliang Guan 已提交
1490
    rpcFreeCont(pMsg->pCont);
1491
    terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
M
Minghao Li 已提交
1492
  }
S
Shengliang Guan 已提交
1493 1494

  return code;
M
Minghao Li 已提交
1495 1496
}

1497
inline bool syncNodeInConfig(SSyncNode* pNode, const SSyncCfg* pCfg) {
1498 1499 1500
  bool b1 = false;
  bool b2 = false;

1501 1502 1503
  for (int32_t i = 0; i < pCfg->replicaNum; ++i) {
    if (strcmp(pCfg->nodeInfo[i].nodeFqdn, pNode->myNodeInfo.nodeFqdn) == 0 &&
        pCfg->nodeInfo[i].nodePort == pNode->myNodeInfo.nodePort) {
1504 1505 1506 1507 1508
      b1 = true;
      break;
    }
  }

1509 1510 1511 1512 1513
  for (int32_t i = 0; i < pCfg->replicaNum; ++i) {
    SRaftId raftId = {
        .addr = SYNC_ADDR(&pCfg->nodeInfo[i]),
        .vgId = pNode->vgId,
    };
1514

1515
    if (syncUtilSameId(&raftId, &pNode->myRaftId)) {
1516 1517 1518 1519 1520
      b2 = true;
      break;
    }
  }

1521
  ASSERT(b1 == b2);
1522 1523 1524
  return b1;
}

1525 1526 1527 1528 1529 1530 1531 1532 1533 1534 1535 1536 1537
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 已提交
1538
void syncNodeDoConfigChange(SSyncNode* pSyncNode, SSyncCfg* pNewConfig, SyncIndex lastConfigChangeIndex) {
1539
  SSyncCfg oldConfig = pSyncNode->raftCfg.cfg;
1540 1541 1542 1543
  if (!syncIsConfigChanged(&oldConfig, pNewConfig)) {
    sInfo("vgId:1, sync not reconfig since not changed");
    return;
  }
S
Shengliang Guan 已提交
1544

1545 1546
  pSyncNode->raftCfg.cfg = *pNewConfig;
  pSyncNode->raftCfg.lastConfigIndex = lastConfigChangeIndex;
1547

1548 1549
  pSyncNode->configChangeNum++;

M
Minghao Li 已提交
1550 1551
  bool IamInOld = syncNodeInConfig(pSyncNode, &oldConfig);
  bool IamInNew = syncNodeInConfig(pSyncNode, pNewConfig);
M
Minghao Li 已提交
1552

M
Minghao Li 已提交
1553 1554
  bool isDrop = false;
  bool isAdd = false;
M
Minghao Li 已提交
1555

M
Minghao Li 已提交
1556 1557 1558 1559
  if (IamInOld && !IamInNew) {
    isDrop = true;
  } else {
    isDrop = false;
1560
  }
1561

M
Minghao Li 已提交
1562 1563 1564 1565 1566
  if (!IamInOld && IamInNew) {
    isAdd = true;
  } else {
    isAdd = false;
  }
M
Minghao Li 已提交
1567

M
Minghao Li 已提交
1568
  // log begin config change
1569 1570
  sNInfo(pSyncNode, "begin do config change, from %d to %d", pSyncNode->vgId, oldConfig.replicaNum,
         pNewConfig->replicaNum);
M
Minghao Li 已提交
1571

M
Minghao Li 已提交
1572
  if (IamInNew) {
1573
    pSyncNode->raftCfg.isStandBy = 0;  // change isStandBy to normal
M
Minghao Li 已提交
1574
  }
M
Minghao Li 已提交
1575
  if (isDrop) {
1576
    pSyncNode->raftCfg.isStandBy = 1;  // set standby
M
Minghao Li 已提交
1577 1578
  }

M
Minghao Li 已提交
1579
  // add last config index
1580
  syncAddCfgIndex(pSyncNode, lastConfigChangeIndex);
M
Minghao Li 已提交
1581

M
Minghao Li 已提交
1582 1583 1584 1585 1586 1587 1588 1589 1590
  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];
S
Shengliang Guan 已提交
1591
    for (int32_t i = 0; i < TSDB_MAX_REPLICA; ++i) {
1592
      oldSenders[i] = pSyncNode->senders[i];
S
Shengliang Guan 已提交
1593
      sSTrace(oldSenders[i], "snapshot sender save old");
M
Minghao Li 已提交
1594
    }
1595

M
Minghao Li 已提交
1596
    // init internal
1597
    pSyncNode->myNodeInfo = pSyncNode->raftCfg.cfg.nodeInfo[pSyncNode->raftCfg.cfg.myIndex];
1598
    syncUtilNodeInfo2RaftId(&pSyncNode->myNodeInfo, pSyncNode->vgId, &pSyncNode->myRaftId);
M
Minghao Li 已提交
1599 1600

    // init peersNum, peers, peersId
1601
    pSyncNode->peersNum = pSyncNode->raftCfg.cfg.replicaNum - 1;
S
Shengliang Guan 已提交
1602
    int32_t j = 0;
1603 1604 1605 1606
    for (int32_t i = 0; i < pSyncNode->raftCfg.cfg.replicaNum; ++i) {
      if (i != pSyncNode->raftCfg.cfg.myIndex) {
        pSyncNode->peersNodeInfo[j] = pSyncNode->raftCfg.cfg.nodeInfo[i];
        syncUtilNodeInfo2EpSet(&pSyncNode->peersNodeInfo[j], &pSyncNode->peersEpset[j]);
M
Minghao Li 已提交
1607 1608 1609
        j++;
      }
    }
S
Shengliang Guan 已提交
1610
    for (int32_t i = 0; i < pSyncNode->peersNum; ++i) {
1611
      syncUtilNodeInfo2RaftId(&pSyncNode->peersNodeInfo[i], pSyncNode->vgId, &pSyncNode->peersId[i]);
M
Minghao Li 已提交
1612
    }
1613

M
Minghao Li 已提交
1614
    // init replicaNum, replicasId
1615 1616 1617
    pSyncNode->replicaNum = pSyncNode->raftCfg.cfg.replicaNum;
    for (int32_t i = 0; i < pSyncNode->raftCfg.cfg.replicaNum; ++i) {
      syncUtilNodeInfo2RaftId(&pSyncNode->raftCfg.cfg.nodeInfo[i], pSyncNode->vgId, &pSyncNode->replicasId[i]);
M
Minghao Li 已提交
1618
    }
1619

1620
    // update quorum first
1621
    pSyncNode->quorum = syncUtilQuorum(pSyncNode->raftCfg.cfg.replicaNum);
1622

M
Minghao Li 已提交
1623 1624 1625 1626
    syncIndexMgrUpdate(pSyncNode->pNextIndex, pSyncNode);
    syncIndexMgrUpdate(pSyncNode->pMatchIndex, pSyncNode);
    voteGrantedUpdate(pSyncNode->pVotesGranted, pSyncNode);
    votesRespondUpdate(pSyncNode->pVotesRespond, pSyncNode);
M
Minghao Li 已提交
1627

M
Minghao Li 已提交
1628
    // reset snapshot senders
1629

M
Minghao Li 已提交
1630
    // clear new
S
Shengliang Guan 已提交
1631
    for (int32_t i = 0; i < TSDB_MAX_REPLICA; ++i) {
1632
      pSyncNode->senders[i] = NULL;
M
Minghao Li 已提交
1633
    }
M
Minghao Li 已提交
1634

M
Minghao Li 已提交
1635
    // reset new
S
Shengliang Guan 已提交
1636
    for (int32_t i = 0; i < pSyncNode->replicaNum; ++i) {
M
Minghao Li 已提交
1637 1638
      // reset sender
      bool reset = false;
S
Shengliang Guan 已提交
1639
      for (int32_t j = 0; j < TSDB_MAX_REPLICA; ++j) {
M
Minghao Li 已提交
1640
        if (syncUtilSameId(&(pSyncNode->replicasId)[i], &oldReplicasId[j]) && oldSenders[j] != NULL) {
1641 1642
          sNTrace(pSyncNode, "snapshot sender reset for:%" PRId64 ", newIndex:%d, dnode:%d, %p",
                  (pSyncNode->replicasId)[i].addr, i, DID(&pSyncNode->replicasId[i]), oldSenders[j]);
M
Minghao Li 已提交
1643

1644
          pSyncNode->senders[i] = oldSenders[j];
M
Minghao Li 已提交
1645 1646 1647 1648
          oldSenders[j] = NULL;
          reset = true;

          // reset replicaIndex
1649 1650
          int32_t oldreplicaIndex = pSyncNode->senders[i]->replicaIndex;
          pSyncNode->senders[i]->replicaIndex = i;
M
Minghao Li 已提交
1651

1652 1653
          sNTrace(pSyncNode, "snapshot sender udpate replicaIndex from %d to %d, dnode:%d, %p, reset:%d",
                  oldreplicaIndex, i, DID(&pSyncNode->replicasId[i]), pSyncNode->senders[i], reset);
M
Minghao Li 已提交
1654 1655

          break;
M
Minghao Li 已提交
1656
        }
1657 1658
      }
    }
1659

M
Minghao Li 已提交
1660
    // create new
S
Shengliang Guan 已提交
1661
    for (int32_t i = 0; i < TSDB_MAX_REPLICA; ++i) {
1662 1663 1664 1665 1666 1667 1668 1669
      if (pSyncNode->senders[i] == NULL) {
        pSyncNode->senders[i] = snapshotSenderCreate(pSyncNode, i);
        if (pSyncNode->senders[i] == NULL) {
          // will be created later while send snapshot
          sSError(pSyncNode->senders[i], "snapshot sender create failed while reconfig");
        } else {
          sSDebug(pSyncNode->senders[i], "snapshot sender create while reconfig, data:%p", pSyncNode->senders[i]);
        }
S
Shengliang Guan 已提交
1670
      } else {
1671
        sSDebug(pSyncNode->senders[i], "snapshot sender already exist, data:%p", pSyncNode->senders[i]);
M
Minghao Li 已提交
1672
      }
1673 1674
    }

M
Minghao Li 已提交
1675
    // free old
S
Shengliang Guan 已提交
1676
    for (int32_t i = 0; i < TSDB_MAX_REPLICA; ++i) {
M
Minghao Li 已提交
1677
      if (oldSenders[i] != NULL) {
1678
        sSDebug(oldSenders[i], "snapshot sender destroy old, data:%p replica-index:%d", oldSenders[i], i);
M
Minghao Li 已提交
1679 1680 1681
        snapshotSenderDestroy(oldSenders[i]);
        oldSenders[i] = NULL;
      }
1682 1683
    }

1684
    // persist cfg
1685
    syncWriteCfgFile(pSyncNode);
1686

M
Minghao Li 已提交
1687 1688
    // change isStandBy to normal (election timeout)
    if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
1689
      syncNodeBecomeLeader(pSyncNode, "");
1690 1691 1692

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

M
Minghao Li 已提交
1695
    } else {
1696
      syncNodeBecomeFollower(pSyncNode, "");
M
Minghao Li 已提交
1697 1698
    }
  } else {
1699
    // persist cfg
1700 1701
    syncWriteCfgFile(pSyncNode);
    sNInfo(pSyncNode, "do not config change from %d to %d", oldConfig.replicaNum, pNewConfig->replicaNum);
1702
  }
1703

M
Minghao Li 已提交
1704
_END:
M
Minghao Li 已提交
1705
  // log end config change
1706 1707
  sNInfo(pSyncNode, "end do config change, from %d to %d", pSyncNode->vgId, oldConfig.replicaNum,
         pNewConfig->replicaNum);
M
Minghao Li 已提交
1708 1709
}

M
Minghao Li 已提交
1710 1711
// raft state change --------------
void syncNodeUpdateTerm(SSyncNode* pSyncNode, SyncTerm term) {
S
Shengliang Guan 已提交
1712 1713
  if (term > pSyncNode->raftStore.currentTerm) {
    raftStoreSetTerm(pSyncNode, term);
1714
    char tmpBuf[64];
1715
    snprintf(tmpBuf, sizeof(tmpBuf), "update term to %" PRId64, term);
1716
    syncNodeBecomeFollower(pSyncNode, tmpBuf);
S
Shengliang Guan 已提交
1717
    raftStoreClearVote(pSyncNode);
M
Minghao Li 已提交
1718 1719 1720
  }
}

1721
void syncNodeUpdateTermWithoutStepDown(SSyncNode* pSyncNode, SyncTerm term) {
S
Shengliang Guan 已提交
1722 1723
  if (term > pSyncNode->raftStore.currentTerm) {
    raftStoreSetTerm(pSyncNode, term);
1724 1725 1726
  }
}

M
Minghao Li 已提交
1727
void syncNodeStepDown(SSyncNode* pSyncNode, SyncTerm newTerm) {
S
Shengliang Guan 已提交
1728
  if (pSyncNode->raftStore.currentTerm > newTerm) {
1729
    sNTrace(pSyncNode, "step down, ignore, new-term:%" PRId64 ", current-term:%" PRId64, newTerm,
S
Shengliang Guan 已提交
1730
            pSyncNode->raftStore.currentTerm);
M
Minghao Li 已提交
1731 1732
    return;
  }
M
Minghao Li 已提交
1733 1734

  do {
1735
    sNTrace(pSyncNode, "step down, new-term:%" PRId64 ", current-term:%" PRId64, newTerm,
S
Shengliang Guan 已提交
1736
            pSyncNode->raftStore.currentTerm);
M
Minghao Li 已提交
1737 1738
  } while (0);

S
Shengliang Guan 已提交
1739 1740
  if (pSyncNode->raftStore.currentTerm < newTerm) {
    raftStoreSetTerm(pSyncNode, newTerm);
M
Minghao Li 已提交
1741
    char tmpBuf[64];
1742
    snprintf(tmpBuf, sizeof(tmpBuf), "step down, update term to %" PRId64, newTerm);
M
Minghao Li 已提交
1743
    syncNodeBecomeFollower(pSyncNode, tmpBuf);
S
Shengliang Guan 已提交
1744
    raftStoreClearVote(pSyncNode);
M
Minghao Li 已提交
1745 1746 1747 1748 1749 1750 1751 1752

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

1753 1754
void syncNodeLeaderChangeRsp(SSyncNode* pSyncNode) { syncRespCleanRsp(pSyncNode->pSyncRespMgr); }

1755
void syncNodeBecomeFollower(SSyncNode* pSyncNode, const char* debugStr) {
M
Minghao Li 已提交
1756
  // maybe clear leader cache
M
Minghao Li 已提交
1757 1758 1759 1760
  if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
    pSyncNode->leaderCache = EMPTY_RAFT_ID;
  }

1761 1762
  pSyncNode->hbSlowNum = 0;

M
Minghao Li 已提交
1763
  // state change
M
Minghao Li 已提交
1764 1765 1766
  pSyncNode->state = TAOS_SYNC_STATE_FOLLOWER;
  syncNodeStopHeartbeatTimer(pSyncNode);

M
Minghao Li 已提交
1767 1768
  // reset elect timer
  syncNodeResetElectTimer(pSyncNode);
M
Minghao Li 已提交
1769

1770 1771 1772
  // send rsp to client
  syncNodeLeaderChangeRsp(pSyncNode);

1773 1774 1775 1776 1777
  // call back
  if (pSyncNode->pFsm != NULL && pSyncNode->pFsm->FpBecomeFollowerCb != NULL) {
    pSyncNode->pFsm->FpBecomeFollowerCb(pSyncNode->pFsm);
  }

M
Minghao Li 已提交
1778 1779 1780
  // min match index
  pSyncNode->minMatchIndex = SYNC_INDEX_INVALID;

B
Benguang Zhao 已提交
1781 1782 1783
  // reset log buffer
  syncLogBufferReset(pSyncNode->pLogBuf, pSyncNode);

M
Minghao Li 已提交
1784
  // trace log
S
Shengliang Guan 已提交
1785
  sNTrace(pSyncNode, "become follower %s", debugStr);
M
Minghao Li 已提交
1786 1787 1788 1789 1790 1791 1792 1793 1794 1795 1796 1797 1798 1799 1800 1801 1802 1803 1804 1805
}

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

1809
  pSyncNode->becomeLeaderNum++;
1810
  pSyncNode->hbrSlowNum = 0;
1811

1812 1813 1814
  // reset restoreFinish
  pSyncNode->restoreFinish = false;

M
Minghao Li 已提交
1815
  // state change
M
Minghao Li 已提交
1816
  pSyncNode->state = TAOS_SYNC_STATE_LEADER;
M
Minghao Li 已提交
1817 1818

  // set leader cache
M
Minghao Li 已提交
1819 1820
  pSyncNode->leaderCache = pSyncNode->myRaftId;

S
Shengliang Guan 已提交
1821
  for (int32_t i = 0; i < pSyncNode->pNextIndex->replicaNum; ++i) {
M
Minghao Li 已提交
1822 1823
    // maybe overwrite myself, no harm
    // just do it!
1824 1825 1826 1827 1828 1829 1830

    // 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);
1831
    ASSERT(code == 0);
1832
    pSyncNode->pNextIndex->index[i] = lastIndex + 1;
M
Minghao Li 已提交
1833 1834
  }

S
Shengliang Guan 已提交
1835
  for (int32_t i = 0; i < pSyncNode->pMatchIndex->replicaNum; ++i) {
M
Minghao Li 已提交
1836 1837
    // maybe overwrite myself, no harm
    // just do it!
M
Minghao Li 已提交
1838 1839 1840
    pSyncNode->pMatchIndex->index[i] = SYNC_INDEX_INVALID;
  }

M
Minghao Li 已提交
1841 1842 1843
  // init peer mgr
  syncNodePeerStateInit(pSyncNode);

M
Minghao Li 已提交
1844
#if 0
1845 1846
  // update sender private term
  SSyncSnapshotSender* pMySender = syncNodeGetSnapshotSender(pSyncNode, &(pSyncNode->myRaftId));
1847
  if (pMySender != NULL) {
S
Shengliang Guan 已提交
1848
    for (int32_t i = 0; i < pSyncNode->pMatchIndex->replicaNum; ++i) {
1849 1850
      if (pSyncNode->senders[i]->privateTerm > pMySender->privateTerm) {
        pMySender->privateTerm = pSyncNode->senders[i]->privateTerm;
1851
      }
1852
    }
1853
    (pMySender->privateTerm) += 100;
1854
  }
M
Minghao Li 已提交
1855
#endif
1856

1857
  // close receiver
M
Minghao Li 已提交
1858 1859
  if (pSyncNode != NULL && pSyncNode->pNewNodeReceiver != NULL &&
      snapshotReceiverIsStart(pSyncNode->pNewNodeReceiver)) {
S
Shengliang Guan 已提交
1860
    snapshotReceiverStop(pSyncNode->pNewNodeReceiver);
1861 1862
  }

M
Minghao Li 已提交
1863
  // stop elect timer
M
Minghao Li 已提交
1864
  syncNodeStopElectTimer(pSyncNode);
M
Minghao Li 已提交
1865

M
Minghao Li 已提交
1866 1867
  // start heartbeat timer
  syncNodeStartHeartbeatTimer(pSyncNode);
M
Minghao Li 已提交
1868

M
Minghao Li 已提交
1869 1870
  // send heartbeat right now
  syncNodeHeartbeatPeers(pSyncNode);
M
Minghao Li 已提交
1871

1872 1873 1874 1875 1876
  // call back
  if (pSyncNode->pFsm != NULL && pSyncNode->pFsm->FpBecomeLeaderCb != NULL) {
    pSyncNode->pFsm->FpBecomeLeaderCb(pSyncNode->pFsm);
  }

M
Minghao Li 已提交
1877 1878 1879
  // min match index
  pSyncNode->minMatchIndex = SYNC_INDEX_INVALID;

B
Benguang Zhao 已提交
1880 1881 1882
  // reset log buffer
  syncLogBufferReset(pSyncNode->pLogBuf, pSyncNode);

M
Minghao Li 已提交
1883
  // trace log
1884
  sNInfo(pSyncNode, "become leader %s", debugStr);
M
Minghao Li 已提交
1885 1886 1887
}

void syncNodeCandidate2Leader(SSyncNode* pSyncNode) {
1888 1889
  ASSERT(pSyncNode->state == TAOS_SYNC_STATE_CANDIDATE);
  ASSERT(voteGrantedMajority(pSyncNode->pVotesGranted));
1890
  syncNodeBecomeLeader(pSyncNode, "candidate to leader");
M
Minghao Li 已提交
1891

S
Shengliang Guan 已提交
1892
  sNTrace(pSyncNode, "state change syncNodeCandidate2Leader");
M
Minghao Li 已提交
1893

B
Benguang Zhao 已提交
1894
  int32_t ret = syncNodeAppendNoop(pSyncNode);
1895 1896 1897 1898
  if (ret < 0) {
    sError("vgId:%d, failed to append noop entry since %s", pSyncNode->vgId, terrstr());
  }

B
Benguang Zhao 已提交
1899
  SyncIndex lastIndex = pSyncNode->pLogStore->syncLogLastIndex(pSyncNode->pLogStore);
1900
  ASSERT(lastIndex >= 0);
B
Benguang Zhao 已提交
1901
  sInfo("vgId:%d, become leader. term: %" PRId64 ", commit index: %" PRId64 ", last index: %" PRId64 "",
S
Shengliang Guan 已提交
1902
        pSyncNode->vgId, pSyncNode->raftStore.currentTerm, pSyncNode->commitIndex, lastIndex);
B
Benguang Zhao 已提交
1903 1904 1905
}

void syncNodeCandidate2LeaderOld(SSyncNode* pSyncNode) {
1906 1907
  ASSERT(pSyncNode->state == TAOS_SYNC_STATE_CANDIDATE);
  ASSERT(voteGrantedMajority(pSyncNode->pVotesGranted));
B
Benguang Zhao 已提交
1908 1909
  syncNodeBecomeLeader(pSyncNode, "candidate to leader");

M
Minghao Li 已提交
1910
  // Raft 3.6.2 Committing entries from previous terms
1911 1912
  syncNodeAppendNoop(pSyncNode);
  syncMaybeAdvanceCommitIndex(pSyncNode);
1913 1914

  if (pSyncNode->replicaNum > 1) {
M
Minghao Li 已提交
1915
    syncNodeReplicate(pSyncNode);
1916
  }
M
Minghao Li 已提交
1917 1918
}

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

M
Minghao Li 已提交
1921
int32_t syncNodePeerStateInit(SSyncNode* pSyncNode) {
S
Shengliang Guan 已提交
1922
  for (int32_t i = 0; i < TSDB_MAX_REPLICA; ++i) {
M
Minghao Li 已提交
1923 1924 1925 1926 1927
    pSyncNode->peerStates[i].lastSendIndex = SYNC_INDEX_INVALID;
    pSyncNode->peerStates[i].lastSendTime = 0;
  }

  return 0;
M
Minghao Li 已提交
1928 1929 1930
}

void syncNodeFollower2Candidate(SSyncNode* pSyncNode) {
1931
  ASSERT(pSyncNode->state == TAOS_SYNC_STATE_FOLLOWER);
M
Minghao Li 已提交
1932
  pSyncNode->state = TAOS_SYNC_STATE_CANDIDATE;
B
Benguang Zhao 已提交
1933 1934
  SyncIndex lastIndex = pSyncNode->pLogStore->syncLogLastIndex(pSyncNode->pLogStore);
  sInfo("vgId:%d, become candidate from follower. term: %" PRId64 ", commit index: %" PRId64 ", last index: %" PRId64,
S
Shengliang Guan 已提交
1935
        pSyncNode->vgId, pSyncNode->raftStore.currentTerm, pSyncNode->commitIndex, lastIndex);
M
Minghao Li 已提交
1936

S
Shengliang Guan 已提交
1937
  sNTrace(pSyncNode, "follower to candidate");
M
Minghao Li 已提交
1938 1939 1940
}

void syncNodeLeader2Follower(SSyncNode* pSyncNode) {
1941
  ASSERT(pSyncNode->state == TAOS_SYNC_STATE_LEADER);
1942
  syncNodeBecomeFollower(pSyncNode, "leader to follower");
B
Benguang Zhao 已提交
1943 1944
  SyncIndex lastIndex = pSyncNode->pLogStore->syncLogLastIndex(pSyncNode->pLogStore);
  sInfo("vgId:%d, become follower from leader. term: %" PRId64 ", commit index: %" PRId64 ", last index: %" PRId64,
S
Shengliang Guan 已提交
1945
        pSyncNode->vgId, pSyncNode->raftStore.currentTerm, pSyncNode->commitIndex, lastIndex);
B
Benguang Zhao 已提交
1946

S
Shengliang Guan 已提交
1947
  sNTrace(pSyncNode, "leader to follower");
M
Minghao Li 已提交
1948 1949 1950
}

void syncNodeCandidate2Follower(SSyncNode* pSyncNode) {
1951
  ASSERT(pSyncNode->state == TAOS_SYNC_STATE_CANDIDATE);
1952
  syncNodeBecomeFollower(pSyncNode, "candidate to follower");
B
Benguang Zhao 已提交
1953 1954
  SyncIndex lastIndex = pSyncNode->pLogStore->syncLogLastIndex(pSyncNode->pLogStore);
  sInfo("vgId:%d, become follower from candidate. term: %" PRId64 ", commit index: %" PRId64 ", last index: %" PRId64,
S
Shengliang Guan 已提交
1955
        pSyncNode->vgId, pSyncNode->raftStore.currentTerm, pSyncNode->commitIndex, lastIndex);
B
Benguang Zhao 已提交
1956

S
Shengliang Guan 已提交
1957
  sNTrace(pSyncNode, "candidate to follower");
M
Minghao Li 已提交
1958 1959
}

M
Minghao Li 已提交
1960 1961
// just called by syncNodeVoteForSelf
// need assert
M
Minghao Li 已提交
1962
void syncNodeVoteForTerm(SSyncNode* pSyncNode, SyncTerm term, SRaftId* pRaftId) {
S
Shengliang Guan 已提交
1963 1964
  ASSERT(term == pSyncNode->raftStore.currentTerm);
  ASSERT(!raftStoreHasVoted(pSyncNode));
M
Minghao Li 已提交
1965

S
Shengliang Guan 已提交
1966
  raftStoreVote(pSyncNode, pRaftId);
M
Minghao Li 已提交
1967 1968
}

M
Minghao Li 已提交
1969
// simulate get vote from outside
M
Minghao Li 已提交
1970
void syncNodeVoteForSelf(SSyncNode* pSyncNode) {
S
Shengliang Guan 已提交
1971
  syncNodeVoteForTerm(pSyncNode, pSyncNode->raftStore.currentTerm, &pSyncNode->myRaftId);
M
Minghao Li 已提交
1972

S
Shengliang Guan 已提交
1973 1974
  SRpcMsg rpcMsg = {0};
  int32_t ret = syncBuildRequestVoteReply(&rpcMsg, pSyncNode->vgId);
S
Shengliang Guan 已提交
1975
  if (ret != 0) return;
M
Minghao Li 已提交
1976

S
Shengliang Guan 已提交
1977
  SyncRequestVoteReply* pMsg = rpcMsg.pCont;
M
Minghao Li 已提交
1978 1979
  pMsg->srcId = pSyncNode->myRaftId;
  pMsg->destId = pSyncNode->myRaftId;
S
Shengliang Guan 已提交
1980
  pMsg->term = pSyncNode->raftStore.currentTerm;
M
Minghao Li 已提交
1981 1982 1983 1984
  pMsg->voteGranted = true;

  voteGrantedVote(pSyncNode->pVotesGranted, pMsg);
  votesRespondAdd(pSyncNode->pVotesRespond, pMsg);
S
Shengliang Guan 已提交
1985
  rpcFreeCont(rpcMsg.pCont);
M
Minghao Li 已提交
1986 1987
}

M
Minghao Li 已提交
1988
// return if has a snapshot
M
Minghao Li 已提交
1989 1990
bool syncNodeHasSnapshot(SSyncNode* pSyncNode) {
  bool      ret = false;
1991
  SSnapshot snapshot = {.data = NULL, .lastApplyIndex = -1, .lastApplyTerm = 0, .lastConfigIndex = -1};
1992 1993
  if (pSyncNode->pFsm->FpGetSnapshotInfo != NULL) {
    pSyncNode->pFsm->FpGetSnapshotInfo(pSyncNode->pFsm, &snapshot);
M
Minghao Li 已提交
1994 1995 1996 1997 1998 1999 2000
    if (snapshot.lastApplyIndex >= SYNC_INDEX_BEGIN) {
      ret = true;
    }
  }
  return ret;
}

M
Minghao Li 已提交
2001 2002
// return max(logLastIndex, snapshotLastIndex)
// if no snapshot and log, return -1
2003
SyncIndex syncNodeGetLastIndex(const SSyncNode* pSyncNode) {
M
Minghao Li 已提交
2004
  SSnapshot snapshot = {.data = NULL, .lastApplyIndex = -1, .lastApplyTerm = 0, .lastConfigIndex = -1};
2005 2006
  if (pSyncNode->pFsm->FpGetSnapshotInfo != NULL) {
    pSyncNode->pFsm->FpGetSnapshotInfo(pSyncNode->pFsm, &snapshot);
M
Minghao Li 已提交
2007 2008 2009 2010 2011 2012 2013
  }
  SyncIndex logLastIndex = pSyncNode->pLogStore->syncLogLastIndex(pSyncNode->pLogStore);

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

M
Minghao Li 已提交
2014 2015
// return the last term of snapshot and log
// if error, return SYNC_TERM_INVALID (by syncLogLastTerm)
M
Minghao Li 已提交
2016 2017
SyncTerm syncNodeGetLastTerm(SSyncNode* pSyncNode) {
  SyncTerm lastTerm = 0;
M
Minghao Li 已提交
2018 2019
  if (syncNodeHasSnapshot(pSyncNode)) {
    // has snapshot
2020
    SSnapshot snapshot = {.data = NULL, .lastApplyIndex = -1, .lastApplyTerm = 0, .lastConfigIndex = -1};
2021 2022
    if (pSyncNode->pFsm->FpGetSnapshotInfo != NULL) {
      pSyncNode->pFsm->FpGetSnapshotInfo(pSyncNode->pFsm, &snapshot);
M
Minghao Li 已提交
2023 2024
    }

M
Minghao Li 已提交
2025 2026 2027
    SyncIndex logLastIndex = pSyncNode->pLogStore->syncLogLastIndex(pSyncNode->pLogStore);
    if (logLastIndex > snapshot.lastApplyIndex) {
      lastTerm = pSyncNode->pLogStore->syncLogLastTerm(pSyncNode->pLogStore);
M
Minghao Li 已提交
2028 2029 2030 2031
    } else {
      lastTerm = snapshot.lastApplyTerm;
    }

M
Minghao Li 已提交
2032
  } else {
M
Minghao Li 已提交
2033 2034
    // no snapshot
    lastTerm = pSyncNode->pLogStore->syncLogLastTerm(pSyncNode->pLogStore);
2035
  }
M
Minghao Li 已提交
2036

M
Minghao Li 已提交
2037 2038 2039 2040 2041 2042 2043
  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);
2044 2045
  return 0;
}
M
Minghao Li 已提交
2046

M
Minghao Li 已提交
2047
// return append-entries first try index
M
Minghao Li 已提交
2048 2049 2050 2051 2052
SyncIndex syncNodeSyncStartIndex(SSyncNode* pSyncNode) {
  SyncIndex syncStartIndex = syncNodeGetLastIndex(pSyncNode) + 1;
  return syncStartIndex;
}

M
Minghao Li 已提交
2053 2054
// if index > 0, return index - 1
// else, return -1
2055 2056 2057 2058 2059 2060 2061 2062 2063
SyncIndex syncNodeGetPreIndex(SSyncNode* pSyncNode, SyncIndex index) {
  SyncIndex preIndex = index - 1;
  if (preIndex < SYNC_INDEX_INVALID) {
    preIndex = SYNC_INDEX_INVALID;
  }

  return preIndex;
}

M
Minghao Li 已提交
2064 2065 2066 2067
// if index < 0, return SYNC_TERM_INVALID
// if index == 0, return 0
// if index > 0, return preTerm
// if error, return SYNC_TERM_INVALID
2068 2069 2070 2071 2072 2073 2074 2075 2076
SyncTerm syncNodeGetPreTerm(SSyncNode* pSyncNode, SyncIndex index) {
  if (index < SYNC_INDEX_BEGIN) {
    return SYNC_TERM_INVALID;
  }

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

2077 2078 2079
  SyncTerm  preTerm = 0;
  SyncIndex preIndex = index - 1;

2080
  SSyncRaftEntry* pPreEntry = NULL;
2081 2082 2083 2084 2085 2086 2087
  SLRUCache*      pCache = pSyncNode->pLogStore->pCache;
  LRUHandle*      h = taosLRUCacheLookup(pCache, &preIndex, sizeof(preIndex));
  int32_t         code = 0;
  if (h) {
    pPreEntry = (SSyncRaftEntry*)taosLRUCacheValue(pCache, h);
    code = 0;

2088
    pSyncNode->pLogStore->cacheHit++;
2089 2090 2091
    sNTrace(pSyncNode, "hit cache index:%" PRId64 ", bytes:%u, %p", preIndex, pPreEntry->bytes, pPreEntry);

  } else {
2092
    pSyncNode->pLogStore->cacheMiss++;
2093 2094 2095 2096
    sNTrace(pSyncNode, "miss cache index:%" PRId64, preIndex);

    code = pSyncNode->pLogStore->syncLogGetEntry(pSyncNode->pLogStore, preIndex, &pPreEntry);
  }
M
Minghao Li 已提交
2097 2098 2099 2100 2101 2102

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

2103
  if (code == 0) {
2104
    ASSERT(pPreEntry != NULL);
2105
    preTerm = pPreEntry->term;
2106 2107 2108 2109

    if (h) {
      taosLRUCacheRelease(pCache, h, false);
    } else {
2110
      syncEntryDestroy(pPreEntry);
2111 2112
    }

2113 2114
    return preTerm;
  } else {
2115 2116 2117 2118
    if (pSyncNode->pFsm->FpGetSnapshotInfo != NULL) {
      pSyncNode->pFsm->FpGetSnapshotInfo(pSyncNode->pFsm, &snapshot);
      if (snapshot.lastApplyIndex == preIndex) {
        return snapshot.lastApplyTerm;
2119 2120 2121 2122
      }
    }
  }

2123
  sNError(pSyncNode, "sync node get pre term error, index:%" PRId64 ", snap-index:%" PRId64 ", snap-term:%" PRId64,
S
Shengliang Guan 已提交
2124
          index, snapshot.lastApplyIndex, snapshot.lastApplyTerm);
2125 2126
  return SYNC_TERM_INVALID;
}
M
Minghao Li 已提交
2127 2128 2129 2130

// 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 已提交
2131
  *pPreTerm = syncNodeGetPreTerm(pSyncNode, index);
M
Minghao Li 已提交
2132 2133 2134
  return 0;
}

M
Minghao Li 已提交
2135
static void syncNodeEqPingTimer(void* param, void* tmrId) {
S
Shengliang Guan 已提交
2136
  if (!syncIsInit()) return;
M
Minghao Li 已提交
2137

S
Shengliang Guan 已提交
2138 2139 2140
  SSyncNode* pNode = param;
  if (atomic_load_64(&pNode->pingTimerLogicClockUser) <= atomic_load_64(&pNode->pingTimerLogicClock)) {
    SRpcMsg rpcMsg = {0};
S
Shengliang Guan 已提交
2141
    int32_t code = syncBuildTimeout(&rpcMsg, SYNC_TIMEOUT_PING, atomic_load_64(&pNode->pingTimerLogicClock),
S
Shengliang Guan 已提交
2142 2143
                                    pNode->pingTimerMS, pNode);
    if (code != 0) {
M
Minghao Li 已提交
2144
      sError("failed to build ping msg");
S
Shengliang Guan 已提交
2145 2146
      rpcFreeCont(rpcMsg.pCont);
      return;
M
Minghao Li 已提交
2147
    }
M
Minghao Li 已提交
2148

M
Minghao Li 已提交
2149
    // sTrace("enqueue ping msg");
S
Shengliang Guan 已提交
2150 2151
    code = pNode->syncEqMsg(pNode->msgcb, &rpcMsg);
    if (code != 0) {
M
Minghao Li 已提交
2152
      sError("failed to sync enqueue ping msg since %s", terrstr());
S
Shengliang Guan 已提交
2153 2154
      rpcFreeCont(rpcMsg.pCont);
      return;
2155
    }
M
Minghao Li 已提交
2156

S
Shengliang Guan 已提交
2157
    taosTmrReset(syncNodeEqPingTimer, pNode->pingTimerMS, pNode, syncEnv()->pTimerManager, &pNode->pPingTimer);
2158
  }
M
Minghao Li 已提交
2159 2160
}

M
Minghao Li 已提交
2161
static void syncNodeEqElectTimer(void* param, void* tmrId) {
S
Shengliang Guan 已提交
2162
  if (!syncIsInit()) return;
M
Minghao Li 已提交
2163

M
Minghao Li 已提交
2164 2165
  int64_t    rid = (int64_t)param;
  SSyncNode* pNode = syncNodeAcquire(rid);
M
Minghao Li 已提交
2166

2167
  if (pNode == NULL) return;
M
Minghao Li 已提交
2168 2169 2170 2171 2172

  if (pNode->syncEqMsg == NULL) {
    syncNodeRelease(pNode);
    return;
  }
2173

2174
  int64_t tsNow = taosGetTimestampMs();
M
Minghao Li 已提交
2175 2176 2177 2178
  if (tsNow < pNode->electTimerParam.executeTime) {
    syncNodeRelease(pNode);
    return;
  }
M
Minghao Li 已提交
2179

S
Shengliang Guan 已提交
2180
  SRpcMsg rpcMsg = {0};
2181 2182
  int32_t code =
      syncBuildTimeout(&rpcMsg, SYNC_TIMEOUT_ELECTION, pNode->electTimerParam.logicClock, pNode->electTimerMS, pNode);
S
Shengliang Guan 已提交
2183

S
Shengliang Guan 已提交
2184
  if (code != 0) {
M
Minghao Li 已提交
2185
    sError("failed to build elect msg");
M
Minghao Li 已提交
2186
    syncNodeRelease(pNode);
S
Shengliang Guan 已提交
2187
    return;
M
Minghao Li 已提交
2188 2189
  }

S
Shengliang Guan 已提交
2190
  SyncTimeout* pTimeout = rpcMsg.pCont;
S
Shengliang Guan 已提交
2191
  sNTrace(pNode, "enqueue elect msg lc:%" PRId64, pTimeout->logicClock);
S
Shengliang Guan 已提交
2192 2193 2194

  code = pNode->syncEqMsg(pNode->msgcb, &rpcMsg);
  if (code != 0) {
M
Minghao Li 已提交
2195
    sError("failed to sync enqueue elect msg since %s", terrstr());
S
Shengliang Guan 已提交
2196
    rpcFreeCont(rpcMsg.pCont);
M
Minghao Li 已提交
2197
    syncNodeRelease(pNode);
2198
    return;
M
Minghao Li 已提交
2199
  }
M
Minghao Li 已提交
2200 2201

  syncNodeRelease(pNode);
M
Minghao Li 已提交
2202 2203
}

M
Minghao Li 已提交
2204
static void syncNodeEqHeartbeatTimer(void* param, void* tmrId) {
S
Shengliang Guan 已提交
2205
  if (!syncIsInit()) return;
2206

S
Shengliang Guan 已提交
2207 2208 2209 2210
  SSyncNode* pNode = param;
  if (pNode->replicaNum > 1) {
    if (atomic_load_64(&pNode->heartbeatTimerLogicClockUser) <= atomic_load_64(&pNode->heartbeatTimerLogicClock)) {
      SRpcMsg rpcMsg = {0};
S
Shengliang Guan 已提交
2211
      int32_t code = syncBuildTimeout(&rpcMsg, SYNC_TIMEOUT_HEARTBEAT, atomic_load_64(&pNode->heartbeatTimerLogicClock),
S
Shengliang Guan 已提交
2212 2213 2214
                                      pNode->heartbeatTimerMS, pNode);

      if (code != 0) {
M
Minghao Li 已提交
2215
        sError("failed to build heartbeat msg");
S
Shengliang Guan 已提交
2216
        return;
2217
      }
M
Minghao Li 已提交
2218

2219
      sTrace("vgId:%d, enqueue heartbeat timer", pNode->vgId);
S
Shengliang Guan 已提交
2220 2221
      code = pNode->syncEqMsg(pNode->msgcb, &rpcMsg);
      if (code != 0) {
M
Minghao Li 已提交
2222
        sError("failed to enqueue heartbeat msg since %s", terrstr());
S
Shengliang Guan 已提交
2223 2224
        rpcFreeCont(rpcMsg.pCont);
        return;
2225
      }
S
Shengliang Guan 已提交
2226 2227 2228 2229

      taosTmrReset(syncNodeEqHeartbeatTimer, pNode->heartbeatTimerMS, pNode, syncEnv()->pTimerManager,
                   &pNode->pHeartbeatTimer);

2230
    } else {
S
Shengliang Guan 已提交
2231 2232
      sTrace("==syncNodeEqHeartbeatTimer== heartbeatTimerLogicClock:%" PRId64 ", heartbeatTimerLogicClockUser:%" PRId64,
             pNode->heartbeatTimerLogicClock, pNode->heartbeatTimerLogicClockUser);
2233
    }
M
Minghao Li 已提交
2234 2235 2236
  }
}

2237
static void syncNodeEqPeerHeartbeatTimer(void* param, void* tmrId) {
2238
  int64_t hbDataRid = (int64_t)param;
2239
  int64_t tsNow = taosGetTimestampMs();
2240

2241 2242
  SSyncHbTimerData* pData = syncHbTimerDataAcquire(hbDataRid);
  if (pData == NULL) {
M
Minghao Li 已提交
2243
    sError("hb timer get pData NULL, %" PRId64, hbDataRid);
2244 2245
    return;
  }
2246

2247
  SSyncNode* pSyncNode = syncNodeAcquire(pData->syncNodeRid);
M
Minghao Li 已提交
2248
  if (pSyncNode == NULL) {
2249
    syncHbTimerDataRelease(pData);
M
Minghao Li 已提交
2250
    sError("hb timer get pSyncNode NULL");
2251 2252 2253 2254 2255 2256 2257 2258
    return;
  }

  SSyncTimer* pSyncTimer = pData->pTimer;

  if (!pSyncNode->isStart) {
    syncNodeRelease(pSyncNode);
    syncHbTimerDataRelease(pData);
M
Minghao Li 已提交
2259
    sError("vgId:%d, hb timer sync node already stop", pSyncNode->vgId);
M
Minghao Li 已提交
2260 2261 2262
    return;
  }

M
Minghao Li 已提交
2263
  if (pSyncNode->state != TAOS_SYNC_STATE_LEADER) {
2264 2265
    syncNodeRelease(pSyncNode);
    syncHbTimerDataRelease(pData);
M
Minghao Li 已提交
2266
    sError("vgId:%d, hb timer sync node not leader", pSyncNode->vgId);
M
Minghao Li 已提交
2267 2268 2269
    return;
  }

M
Minghao Li 已提交
2270
  // sTrace("vgId:%d, eq peer hb timer", pSyncNode->vgId);
2271 2272

  if (pSyncNode->replicaNum > 1) {
M
Minghao Li 已提交
2273 2274 2275
    int64_t timerLogicClock = atomic_load_64(&pSyncTimer->logicClock);
    int64_t msgLogicClock = atomic_load_64(&pData->logicClock);

2276
    if (timerLogicClock == msgLogicClock) {
2277 2278 2279 2280 2281 2282 2283 2284 2285 2286 2287 2288 2289 2290 2291 2292
      if (tsNow > pData->execTime) {
#if 0        
        sTrace(
            "vgId:%d, hbDataRid:%ld,  EXECUTE this step-------- heartbeat tsNow:%ld, exec:%ld, tsNow-exec:%ld, "
            "---------",
            pSyncNode->vgId, hbDataRid, tsNow, pData->execTime, tsNow - pData->execTime);
#endif

        pData->execTime += pSyncTimer->timerMS;

        SRpcMsg rpcMsg = {0};
        (void)syncBuildHeartbeat(&rpcMsg, pSyncNode->vgId);

        SyncHeartbeat* pSyncMsg = rpcMsg.pCont;
        pSyncMsg->srcId = pSyncNode->myRaftId;
        pSyncMsg->destId = pData->destId;
S
Shengliang Guan 已提交
2293
        pSyncMsg->term = pSyncNode->raftStore.currentTerm;
2294 2295 2296
        pSyncMsg->commitIndex = pSyncNode->commitIndex;
        pSyncMsg->minMatchIndex = syncMinMatchIndex(pSyncNode);
        pSyncMsg->privateTerm = 0;
2297
        pSyncMsg->timeStamp = tsNow;
2298 2299 2300 2301 2302 2303

        // update reset time
        int64_t timerElapsed = tsNow - pSyncTimer->timeStamp;
        pSyncTimer->timeStamp = tsNow;

        // send msg
2304 2305
        syncLogSendHeartbeat(pSyncNode, pSyncMsg, false, timerElapsed, pData->execTime);
        syncNodeSendHeartbeat(pSyncNode, &pSyncMsg->destId, &rpcMsg);
2306 2307 2308 2309 2310 2311 2312 2313
      } else {
#if 0        
        sTrace(
            "vgId:%d, hbDataRid:%ld,  pass this step-------- heartbeat tsNow:%ld, exec:%ld, tsNow-exec:%ld, ---------",
            pSyncNode->vgId, hbDataRid, tsNow, pData->execTime, tsNow - pData->execTime);
#endif
      }

M
Minghao Li 已提交
2314 2315
      if (syncIsInit()) {
        // sTrace("vgId:%d, reset peer hb timer", pSyncNode->vgId);
2316 2317
        taosTmrReset(syncNodeEqPeerHeartbeatTimer, pSyncTimer->timerMS / HEARTBEAT_TICK_NUM, (void*)hbDataRid,
                     syncEnv()->pTimerManager, &pSyncTimer->pTimer);
M
Minghao Li 已提交
2318 2319 2320 2321
      } else {
        sError("sync env is stop, reset peer hb timer error");
      }

2322
    } else {
M
Minghao Li 已提交
2323 2324
      sTrace("vgId:%d, do not send hb, timerLogicClock:%" PRId64 ", msgLogicClock:%" PRId64 "", pSyncNode->vgId,
             timerLogicClock, msgLogicClock);
2325 2326
    }
  }
2327 2328 2329

  syncHbTimerDataRelease(pData);
  syncNodeRelease(pSyncNode);
2330 2331
}

2332 2333 2334 2335 2336
static int32_t syncNodeEqNoop(SSyncNode* pNode) {
  if (pNode->state == TAOS_SYNC_STATE_LEADER) {
    terrno = TSDB_CODE_SYN_NOT_LEADER;
    return -1;
  }
M
Minghao Li 已提交
2337

2338
  SyncIndex       index = pNode->pLogStore->syncLogWriteIndex(pNode->pLogStore);
S
Shengliang Guan 已提交
2339
  SyncTerm        term = pNode->raftStore.currentTerm;
2340 2341
  SSyncRaftEntry* pEntry = syncEntryBuildNoop(term, index, pNode->vgId);
  if (pEntry == NULL) return -1;
M
Minghao Li 已提交
2342

S
Shengliang Guan 已提交
2343
  SRpcMsg rpcMsg = {0};
S
Shengliang Guan 已提交
2344
  int32_t code = syncBuildClientRequestFromNoopEntry(&rpcMsg, pEntry, pNode->vgId);
2345
  syncEntryDestroy(pEntry);
M
Minghao Li 已提交
2346

2347 2348 2349
  sNTrace(pNode, "propose msg, type:noop");
  code = (*pNode->syncEqMsg)(pNode->msgcb, &rpcMsg);
  if (code != 0) {
M
Minghao Li 已提交
2350
    sError("failed to propose noop msg while enqueue since %s", terrstr());
2351
  }
M
Minghao Li 已提交
2352

2353
  return code;
M
Minghao Li 已提交
2354 2355
}

2356 2357
static void deleteCacheEntry(const void* key, size_t keyLen, void* value) { taosMemoryFree(value); }

2358 2359 2360 2361
int32_t syncCacheEntry(SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry, LRUHandle** h) {
  SSyncLogStoreData* pData = pLogStore->data;
  sNTrace(pData->pSyncNode, "in cache index:%" PRId64 ", bytes:%u, %p", pEntry->index, pEntry->bytes, pEntry);

S
Shengliang Guan 已提交
2362 2363
  int32_t   code = 0;
  int32_t   entryLen = sizeof(*pEntry) + pEntry->dataLen;
2364 2365 2366 2367 2368 2369 2370 2371 2372
  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;
}

B
Benguang Zhao 已提交
2373
int32_t syncNodeAppend(SSyncNode* ths, SSyncRaftEntry* pEntry) {
2374 2375 2376 2377 2378 2379 2380
  if (pEntry->dataLen < sizeof(SMsgHead)) {
    sError("vgId:%d, cannot append an invalid client request with no msg head. type:%s, dataLen:%d", ths->vgId,
           TMSG_INFO(pEntry->originalRpcType), pEntry->dataLen);
    syncEntryDestroy(pEntry);
    return -1;
  }

B
Benguang Zhao 已提交
2381 2382
  // append to log buffer
  if (syncLogBufferAppend(ths->pLogBuf, ths, pEntry) < 0) {
2383 2384
    sError("vgId:%d, failed to enqueue sync log buffer, index:%" PRId64, ths->vgId, pEntry->index);
    terrno = TSDB_CODE_SYN_BUFFER_FULL;
S
Shengliang Guan 已提交
2385
    (void)syncLogFsmExecute(ths, ths->pFsm, ths->state, ths->raftStore.currentTerm, pEntry, TSDB_CODE_SYN_BUFFER_FULL);
2386
    syncEntryDestroy(pEntry);
B
Benguang Zhao 已提交
2387 2388 2389 2390
    return -1;
  }

  // proceed match index, with replicating on needed
2391
  SyncIndex matchIndex = syncLogBufferProceed(ths->pLogBuf, ths, NULL);
B
Benguang Zhao 已提交
2392

2393
  sTrace("vgId:%d, append raft entry. index: %" PRId64 ", term: %" PRId64 " pBuf: [%" PRId64 " %" PRId64 " %" PRId64
2394 2395 2396
         ", %" PRId64 ")",
         ths->vgId, pEntry->index, pEntry->term, ths->pLogBuf->startIndex, ths->pLogBuf->commitIndex,
         ths->pLogBuf->matchIndex, ths->pLogBuf->endIndex);
B
Benguang Zhao 已提交
2397

B
Benguang Zhao 已提交
2398 2399 2400 2401 2402 2403 2404 2405 2406 2407 2408 2409 2410 2411 2412 2413
  // multi replica
  if (ths->replicaNum > 1) {
    return 0;
  }

  // single replica
  (void)syncNodeUpdateCommitIndex(ths, matchIndex);

  if (syncLogBufferCommit(ths->pLogBuf, ths, ths->commitIndex) < 0) {
    sError("vgId:%d, failed to commit until commitIndex:%" PRId64 "", ths->vgId, ths->commitIndex);
    return -1;
  }

  return 0;
}

2414
bool syncNodeHeartbeatReplyTimeout(SSyncNode* pSyncNode) {
2415 2416 2417 2418 2419 2420 2421 2422 2423 2424 2425 2426
  if (pSyncNode->replicaNum == 1) {
    return false;
  }

  int32_t toCount = 0;
  int64_t tsNow = taosGetTimestampMs();
  for (int32_t i = 0; i < pSyncNode->peersNum; ++i) {
    int64_t recvTime = syncIndexMgrGetRecvTime(pSyncNode->pMatchIndex, &(pSyncNode->peersId[i]));
    if (recvTime == 0 || recvTime == -1) {
      continue;
    }

2427
    if (tsNow - recvTime > tsHeartbeatTimeout) {
2428 2429 2430 2431 2432 2433 2434 2435 2436
      toCount++;
    }
  }

  bool b = (toCount >= pSyncNode->quorum ? true : false);

  return b;
}

2437 2438 2439 2440 2441 2442 2443 2444 2445 2446 2447 2448 2449 2450 2451 2452 2453 2454 2455
bool syncNodeSnapshotSending(SSyncNode* pSyncNode) {
  if (pSyncNode == NULL) return false;
  bool b = false;
  for (int32_t i = 0; i < pSyncNode->replicaNum; ++i) {
    if (pSyncNode->senders[i] != NULL && pSyncNode->senders[i]->start) {
      b = true;
      break;
    }
  }
  return b;
}

bool syncNodeSnapshotRecving(SSyncNode* pSyncNode) {
  if (pSyncNode == NULL) return false;
  if (pSyncNode->pNewNodeReceiver == NULL) return false;
  if (pSyncNode->pNewNodeReceiver->start) return true;
  return false;
}

M
Minghao Li 已提交
2456
static int32_t syncNodeAppendNoop(SSyncNode* ths) {
B
Benguang Zhao 已提交
2457
  SyncIndex index = syncLogBufferGetEndIndex(ths->pLogBuf);
S
Shengliang Guan 已提交
2458
  SyncTerm  term = ths->raftStore.currentTerm;
B
Benguang Zhao 已提交
2459 2460 2461 2462 2463 2464 2465

  SSyncRaftEntry* pEntry = syncEntryBuildNoop(term, index, ths->vgId);
  if (pEntry == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return -1;
  }

B
Benguang Zhao 已提交
2466 2467
  int32_t ret = syncNodeAppend(ths, pEntry);
  return 0;
B
Benguang Zhao 已提交
2468 2469 2470
}

static int32_t syncNodeAppendNoopOld(SSyncNode* ths) {
M
Minghao Li 已提交
2471 2472
  int32_t ret = 0;

2473
  SyncIndex       index = ths->pLogStore->syncLogWriteIndex(ths->pLogStore);
S
Shengliang Guan 已提交
2474
  SyncTerm        term = ths->raftStore.currentTerm;
M
Minghao Li 已提交
2475
  SSyncRaftEntry* pEntry = syncEntryBuildNoop(term, index, ths->vgId);
2476
  ASSERT(pEntry != NULL);
M
Minghao Li 已提交
2477

2478 2479
  LRUHandle* h = NULL;

M
Minghao Li 已提交
2480
  if (ths->state == TAOS_SYNC_STATE_LEADER) {
2481
    int32_t code = ths->pLogStore->syncLogAppendEntry(ths->pLogStore, pEntry);
2482
    if (code != 0) {
M
Minghao Li 已提交
2483
      sError("append noop error");
2484 2485
      return -1;
    }
2486 2487

    syncCacheEntry(ths->pLogStore, pEntry, &h);
M
Minghao Li 已提交
2488 2489
  }

2490 2491 2492
  if (h) {
    taosLRUCacheRelease(ths->pLogStore->pCache, h, false);
  } else {
B
Benguang Zhao 已提交
2493
    syncEntryDestroy(pEntry);
2494 2495
  }

M
Minghao Li 已提交
2496 2497 2498
  return ret;
}

S
Shengliang Guan 已提交
2499 2500
int32_t syncNodeOnHeartbeat(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
  SyncHeartbeat* pMsg = pRpcMsg->pCont;
2501

M
Minghao Li 已提交
2502 2503 2504 2505
  const STraceId* trace = &pRpcMsg->info.traceId;
  char            tbuf[40] = {0};
  TRACE_TO_STR(trace, tbuf);

2506
  int64_t tsMs = taosGetTimestampMs();
M
Minghao Li 已提交
2507
  int64_t timeDiff = tsMs - pMsg->timeStamp;
M
Minghao Li 已提交
2508
  syncLogRecvHeartbeat(ths, pMsg, timeDiff, tbuf);
2509

2510 2511 2512 2513
  SRpcMsg rpcMsg = {0};
  (void)syncBuildHeartbeatReply(&rpcMsg, ths->vgId);

  SyncHeartbeatReply* pMsgReply = rpcMsg.pCont;
2514 2515
  pMsgReply->destId = pMsg->srcId;
  pMsgReply->srcId = ths->myRaftId;
S
Shengliang Guan 已提交
2516
  pMsgReply->term = ths->raftStore.currentTerm;
2517
  pMsgReply->privateTerm = 8864;  // magic number
2518
  pMsgReply->startTime = ths->startTime;
2519
  pMsgReply->timeStamp = tsMs;
2520

S
Shengliang Guan 已提交
2521
  if (pMsg->term == ths->raftStore.currentTerm && ths->state != TAOS_SYNC_STATE_LEADER) {
2522 2523
    syncIndexMgrSetRecvTime(ths->pNextIndex, &(pMsg->srcId), tsMs);

2524
    syncNodeResetElectTimer(ths);
M
Minghao Li 已提交
2525
    ths->minMatchIndex = pMsg->minMatchIndex;
2526 2527

    if (ths->state == TAOS_SYNC_STATE_FOLLOWER) {
2528
      // syncNodeFollowerCommit(ths, pMsg->commitIndex);
S
Shengliang Guan 已提交
2529 2530 2531 2532
      SRpcMsg rpcMsgLocalCmd = {0};
      (void)syncBuildLocalCmd(&rpcMsgLocalCmd, ths->vgId);

      SyncLocalCmd* pSyncMsg = rpcMsgLocalCmd.pCont;
2533
      pSyncMsg->cmd = SYNC_LOCAL_CMD_FOLLOWER_CMT;
2534 2535 2536
      pSyncMsg->commitIndex = pMsg->commitIndex;
      pSyncMsg->currentTerm = pMsg->term;
      SyncIndex fcIndex = pSyncMsg->commitIndex;
2537 2538 2539 2540 2541 2542 2543

      if (ths->syncEqMsg != NULL && ths->msgcb != NULL) {
        int32_t code = ths->syncEqMsg(ths->msgcb, &rpcMsgLocalCmd);
        if (code != 0) {
          sError("vgId:%d, sync enqueue fc-commit msg error, code:%d", ths->vgId, code);
          rpcFreeCont(rpcMsgLocalCmd.pCont);
        } else {
2544
          sTrace("vgId:%d, sync enqueue fc-commit msg, fc-index:%" PRId64, ths->vgId, fcIndex);
2545 2546
        }
      }
2547 2548 2549
    }
  }

S
Shengliang Guan 已提交
2550
  if (pMsg->term >= ths->raftStore.currentTerm && ths->state != TAOS_SYNC_STATE_FOLLOWER) {
2551
    // syncNodeStepDown(ths, pMsg->term);
S
Shengliang Guan 已提交
2552 2553 2554 2555
    SRpcMsg rpcMsgLocalCmd = {0};
    (void)syncBuildLocalCmd(&rpcMsgLocalCmd, ths->vgId);

    SyncLocalCmd* pSyncMsg = rpcMsgLocalCmd.pCont;
2556
    pSyncMsg->cmd = SYNC_LOCAL_CMD_STEP_DOWN;
2557 2558
    pSyncMsg->currentTerm = pMsg->term;
    pSyncMsg->commitIndex = pMsg->commitIndex;
2559

S
Shengliang Guan 已提交
2560 2561
    if (ths->syncEqMsg != NULL && ths->msgcb != NULL) {
      int32_t code = ths->syncEqMsg(ths->msgcb, &rpcMsgLocalCmd);
2562 2563 2564 2565
      if (code != 0) {
        sError("vgId:%d, sync enqueue step-down msg error, code:%d", ths->vgId, code);
        rpcFreeCont(rpcMsgLocalCmd.pCont);
      } else {
2566
        sTrace("vgId:%d, sync enqueue step-down msg, new-term: %" PRId64, ths->vgId, pSyncMsg->currentTerm);
2567
      }
2568 2569 2570 2571 2572 2573 2574 2575 2576 2577 2578 2579 2580 2581 2582
    }
  }

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

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

2583
int32_t syncNodeOnHeartbeatReply(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
S
Shengliang Guan 已提交
2584 2585 2586 2587
  const STraceId* trace = &pRpcMsg->info.traceId;
  char            tbuf[40] = {0};
  TRACE_TO_STR(trace, tbuf);

2588
  SyncHeartbeatReply* pMsg = pRpcMsg->pCont;
B
Benguang Zhao 已提交
2589
  SSyncLogReplMgr*    pMgr = syncNodeGetLogReplMgr(ths, &pMsg->srcId);
2590 2591 2592 2593
  if (pMgr == NULL) {
    sError("vgId:%d, failed to get log repl mgr for the peer at addr 0x016%" PRIx64 "", ths->vgId, pMsg->srcId.addr);
    return -1;
  }
2594 2595

  int64_t tsMs = taosGetTimestampMs();
S
Shengliang Guan 已提交
2596
  syncLogRecvHeartbeatReply(ths, pMsg, tsMs - pMsg->timeStamp, tbuf);
2597

2598 2599
  syncIndexMgrSetRecvTime(ths->pMatchIndex, &pMsg->srcId, tsMs);

2600 2601 2602
  return syncLogReplMgrProcessHeartbeatReply(pMgr, ths, pMsg);
}

2603
int32_t syncNodeOnHeartbeatReplyOld(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
2604
  SyncHeartbeatReply* pMsg = pRpcMsg->pCont;
2605

M
Minghao Li 已提交
2606 2607 2608 2609
  const STraceId* trace = &pRpcMsg->info.traceId;
  char            tbuf[40] = {0};
  TRACE_TO_STR(trace, tbuf);

M
Minghao Li 已提交
2610
  int64_t tsMs = taosGetTimestampMs();
M
Minghao Li 已提交
2611
  int64_t timeDiff = tsMs - pMsg->timeStamp;
M
Minghao Li 已提交
2612
  syncLogRecvHeartbeatReply(ths, pMsg, timeDiff, tbuf);
M
Minghao Li 已提交
2613

2614
  // update last reply time, make decision whether the other node is alive or not
M
Minghao Li 已提交
2615
  syncIndexMgrSetRecvTime(ths->pMatchIndex, &pMsg->srcId, tsMs);
2616 2617 2618
  return 0;
}

S
Shengliang Guan 已提交
2619 2620
int32_t syncNodeOnLocalCmd(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
  SyncLocalCmd* pMsg = pRpcMsg->pCont;
2621 2622
  syncLogRecvLocalCmd(ths, pMsg, "");

2623
  if (pMsg->cmd == SYNC_LOCAL_CMD_STEP_DOWN) {
2624
    syncNodeStepDown(ths, pMsg->currentTerm);
2625 2626

  } else if (pMsg->cmd == SYNC_LOCAL_CMD_FOLLOWER_CMT) {
2627 2628 2629 2630
    SyncTerm matchTerm = syncLogBufferGetLastMatchTerm(ths->pLogBuf);
    if (pMsg->currentTerm == matchTerm) {
      (void)syncNodeUpdateCommitIndex(ths, pMsg->commitIndex);
    }
2631 2632 2633 2634 2635 2636 2637 2638 2639 2640 2641 2642
    if (syncLogBufferCommit(ths->pLogBuf, ths, ths->commitIndex) < 0) {
      sError("vgId:%d, failed to commit raft log since %s. commit index: %" PRId64 "", ths->vgId, terrstr(),
             ths->commitIndex);
    }
  } else {
    sError("error local cmd");
  }

  return 0;
}

int32_t syncNodeOnLocalCmdOld(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
2643
  ASSERT(false && "deprecated");
2644 2645 2646
  SyncLocalCmd* pMsg = pRpcMsg->pCont;
  syncLogRecvLocalCmd(ths, pMsg, "");

M
Minghao Li 已提交
2647
  if (pMsg->cmd == SYNC_LOCAL_CMD_STEP_DOWN) {
2648
    syncNodeStepDown(ths, pMsg->currentTerm);
M
Minghao Li 已提交
2649

2650
  } else if (pMsg->cmd == SYNC_LOCAL_CMD_FOLLOWER_CMT) {
2651
    syncNodeFollowerCommit(ths, pMsg->commitIndex);
2652

M
Minghao Li 已提交
2653
  } else {
M
Minghao Li 已提交
2654
    sError("error local cmd");
M
Minghao Li 已提交
2655
  }
2656 2657 2658 2659

  return 0;
}

M
Minghao Li 已提交
2660 2661 2662 2663 2664 2665 2666 2667 2668 2669
// 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 已提交
2670

2671
int32_t syncNodeOnClientRequest(SSyncNode* ths, SRpcMsg* pMsg, SyncIndex* pRetIndex) {
S
Shengliang Guan 已提交
2672
  sNTrace(ths, "on client request");
2673

B
Benguang Zhao 已提交
2674 2675
  int32_t code = 0;

B
Benguang Zhao 已提交
2676
  SyncIndex       index = syncLogBufferGetEndIndex(ths->pLogBuf);
S
Shengliang Guan 已提交
2677
  SyncTerm        term = ths->raftStore.currentTerm;
B
Benguang Zhao 已提交
2678
  SSyncRaftEntry* pEntry = NULL;
2679 2680 2681 2682
  if (pMsg->msgType == TDMT_SYNC_CLIENT_REQUEST) {
    pEntry = syncEntryBuildFromClientRequest(pMsg->pCont, term, index);
  } else {
    pEntry = syncEntryBuildFromRpcMsg(pMsg, term, index);
B
Benguang Zhao 已提交
2683 2684
  }

2685 2686 2687 2688 2689
  if (pEntry == NULL) {
    sError("vgId:%d, failed to process client request since %s.", ths->vgId, terrstr());
    return -1;
  }

B
Benguang Zhao 已提交
2690 2691 2692 2693 2694
  if (ths->state == TAOS_SYNC_STATE_LEADER) {
    if (pRetIndex) {
      (*pRetIndex) = index;
    }

2695 2696
    int32_t code = syncNodeAppend(ths, pEntry);
    return code;
2697 2698 2699
  } else {
    syncEntryDestroy(pEntry);
    pEntry = NULL;
2700
    return -1;
B
Benguang Zhao 已提交
2701 2702 2703
  }
}

2704 2705
int32_t syncNodeOnClientRequestOld(SSyncNode* ths, SRpcMsg* pMsg, SyncIndex* pRetIndex) {
  sNTrace(ths, "on client request");
B
Benguang Zhao 已提交
2706

M
Minghao Li 已提交
2707
  int32_t ret = 0;
2708
  int32_t code = 0;
M
Minghao Li 已提交
2709

M
Minghao Li 已提交
2710
  SyncIndex       index = ths->pLogStore->syncLogWriteIndex(ths->pLogStore);
S
Shengliang Guan 已提交
2711
  SyncTerm        term = ths->raftStore.currentTerm;
2712 2713 2714 2715 2716 2717 2718
  SSyncRaftEntry* pEntry;

  if (pMsg->msgType == TDMT_SYNC_CLIENT_REQUEST) {
    pEntry = syncEntryBuildFromClientRequest(pMsg->pCont, term, index);
  } else {
    pEntry = syncEntryBuildFromRpcMsg(pMsg, term, index);
  }
M
Minghao Li 已提交
2719

2720 2721
  LRUHandle* h = NULL;

M
Minghao Li 已提交
2722
  if (ths->state == TAOS_SYNC_STATE_LEADER) {
2723 2724 2725
    // append entry
    code = ths->pLogStore->syncLogAppendEntry(ths->pLogStore, pEntry);
    if (code != 0) {
2726 2727 2728 2729
      if (ths->replicaNum == 1) {
        if (h) {
          taosLRUCacheRelease(ths->pLogStore->pCache, h, false);
        } else {
2730
          syncEntryDestroy(pEntry);
2731
        }
2732

2733 2734 2735 2736
        return -1;

      } else {
        // del resp mgr, call FpCommitCb
2737 2738 2739 2740 2741 2742 2743 2744
        SFsmCbMeta cbMeta = {
            .index = pEntry->index,
            .lastConfigIndex = SYNC_INDEX_INVALID,
            .isWeak = pEntry->isWeak,
            .code = -1,
            .state = ths->state,
            .seqNum = pEntry->seqNum,
            .term = pEntry->term,
S
Shengliang Guan 已提交
2745
            .currentTerm = ths->raftStore.currentTerm,
2746 2747
            .flag = 0,
        };
2748
        ths->pFsm->FpCommitCb(ths->pFsm, pMsg, &cbMeta);
2749 2750 2751 2752

        if (h) {
          taosLRUCacheRelease(ths->pLogStore->pCache, h, false);
        } else {
2753
          syncEntryDestroy(pEntry);
2754 2755
        }

2756 2757
        return -1;
      }
2758
    }
M
Minghao Li 已提交
2759

2760 2761
    syncCacheEntry(ths->pLogStore, pEntry, &h);

2762 2763
    // if mulit replica, start replicate right now
    if (ths->replicaNum > 1) {
M
Minghao Li 已提交
2764
      syncNodeReplicate(ths);
2765
    }
2766

2767 2768
    // if only myself, maybe commit right now
    if (ths->replicaNum == 1) {
2769 2770 2771 2772 2773
      if (syncNodeIsMnode(ths)) {
        syncMaybeAdvanceCommitIndex(ths);
      } else {
        syncOneReplicaAdvance(ths);
      }
2774
    }
M
Minghao Li 已提交
2775 2776
  }

2777 2778 2779 2780 2781 2782 2783 2784
  if (pRetIndex != NULL) {
    if (ret == 0 && pEntry != NULL) {
      *pRetIndex = pEntry->index;
    } else {
      *pRetIndex = SYNC_INDEX_INVALID;
    }
  }

2785 2786 2787
  if (h) {
    taosLRUCacheRelease(ths->pLogStore->pCache, h, false);
  } else {
B
Benguang Zhao 已提交
2788
    syncEntryDestroy(pEntry);
2789 2790
  }

M
Minghao Li 已提交
2791
  return ret;
2792
}
M
Minghao Li 已提交
2793

S
Shengliang Guan 已提交
2794 2795 2796
const char* syncStr(ESyncState state) {
  switch (state) {
    case TAOS_SYNC_STATE_FOLLOWER:
2797
      return "follower";
S
Shengliang Guan 已提交
2798
    case TAOS_SYNC_STATE_CANDIDATE:
2799
      return "candidate";
S
Shengliang Guan 已提交
2800
    case TAOS_SYNC_STATE_LEADER:
2801
      return "leader";
S
Shengliang Guan 已提交
2802
    case TAOS_SYNC_STATE_ERROR:
2803
      return "error";
S
Shengliang Guan 已提交
2804 2805 2806 2807
    case TAOS_SYNC_STATE_OFFLINE:
      return "offline";
    default:
      return "unknown";
S
Shengliang Guan 已提交
2808
  }
M
Minghao Li 已提交
2809
}
2810

2811
#if 0
2812
int32_t syncDoLeaderTransfer(SSyncNode* ths, SRpcMsg* pRpcMsg, SSyncRaftEntry* pEntry) {
2813
  if (ths->state != TAOS_SYNC_STATE_FOLLOWER) {
S
Shengliang Guan 已提交
2814
    sNTrace(ths, "I am not follower, can not do leader transfer");
2815 2816
    return 0;
  }
2817 2818

  if (!ths->restoreFinish) {
S
Shengliang Guan 已提交
2819
    sNTrace(ths, "restore not finish, can not do leader transfer");
2820 2821 2822
    return 0;
  }

S
Shengliang Guan 已提交
2823
  if (pEntry->term < ths->raftStore.currentTerm) {
2824
    sNTrace(ths, "little term:%" PRId64 ", can not do leader transfer", pEntry->term);
2825 2826 2827 2828
    return 0;
  }

  if (pEntry->index < syncNodeGetLastIndex(ths)) {
S
Shengliang Guan 已提交
2829
    sNTrace(ths, "little index:%" PRId64 ", can not do leader transfer", pEntry->index);
2830 2831 2832
    return 0;
  }

2833 2834
  /*
    if (ths->vgId > 1) {
S
Shengliang Guan 已提交
2835
      sNTrace(ths, "I am vnode, can not do leader transfer");
2836 2837 2838 2839
      return 0;
    }
  */

2840
  SyncLeaderTransfer* pSyncLeaderTransfer = pRpcMsg->pCont;
S
Shengliang Guan 已提交
2841
  sNTrace(ths, "do leader transfer, index:%" PRId64, pEntry->index);
M
Minghao Li 已提交
2842

M
Minghao Li 已提交
2843 2844 2845
  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 已提交
2846

M
Minghao Li 已提交
2847 2848
  bool same = sameId || sameNodeInfo;
  if (same) {
M
Minghao Li 已提交
2849 2850 2851
    // reset elect timer now!
    int32_t electMS = 1;
    int32_t ret = syncNodeRestartElectTimer(ths, electMS);
2852
    ASSERT(ret == 0);
M
Minghao Li 已提交
2853

2854
    sNTrace(ths, "maybe leader transfer to %s:%d %" PRId64, pSyncLeaderTransfer->newNodeInfo.nodeFqdn,
S
Shengliang Guan 已提交
2855
            pSyncLeaderTransfer->newNodeInfo.nodePort, pSyncLeaderTransfer->newLeaderId.addr);
2856 2857
  }

M
Minghao Li 已提交
2858
  if (ths->pFsm->FpLeaderTransferCb != NULL) {
S
Shengliang Guan 已提交
2859
    SFsmCbMeta cbMeta = {
S
Shengliang Guan 已提交
2860
        .code = 0,
S
Shengliang Guan 已提交
2861
        .currentTerm = ths->raftStore.currentTerm,
S
Shengliang Guan 已提交
2862 2863 2864 2865 2866 2867 2868
        .flag = 0,
        .index = pEntry->index,
        .lastConfigIndex = syncNodeGetSnapshotConfigIndex(ths, pEntry->index),
        .isWeak = pEntry->isWeak,
        .seqNum = pEntry->seqNum,
        .state = ths->state,
        .term = pEntry->term,
S
Shengliang Guan 已提交
2869 2870
    };
    ths->pFsm->FpLeaderTransferCb(ths->pFsm, pRpcMsg, &cbMeta);
2871 2872
  }

2873 2874 2875
  return 0;
}

2876 2877
#endif

2878
int32_t syncNodeUpdateNewConfigIndex(SSyncNode* ths, SSyncCfg* pNewCfg) {
S
Shengliang Guan 已提交
2879
  for (int32_t i = 0; i < pNewCfg->replicaNum; ++i) {
2880 2881 2882 2883
    SRaftId raftId = {
        .addr = SYNC_ADDR(&pNewCfg->nodeInfo[i]),
        .vgId = ths->vgId,
    };
2884 2885 2886 2887 2888 2889 2890 2891 2892 2893

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

  return -1;
}

2894 2895 2896 2897
bool syncNodeIsOptimizedOneReplica(SSyncNode* ths, SRpcMsg* pMsg) {
  return (ths->replicaNum == 1 && syncUtilUserCommit(pMsg->msgType) && ths->vgId != 1);
}

M
Minghao Li 已提交
2898
int32_t syncNodeDoCommit(SSyncNode* ths, SyncIndex beginIndex, SyncIndex endIndex, uint64_t flag) {
2899
  ASSERT(false);
2900 2901 2902 2903
  if (beginIndex > endIndex) {
    return 0;
  }

M
Minghao Li 已提交
2904 2905 2906 2907 2908 2909 2910 2911 2912
  if (ths == NULL) {
    return -1;
  }

  if (ths->pFsm != NULL && ths->pFsm->FpGetSnapshotInfo != NULL) {
    // advance commit index to sanpshot first
    SSnapshot snapshot = {0};
    ths->pFsm->FpGetSnapshotInfo(ths->pFsm, &snapshot);
    if (snapshot.lastApplyIndex >= 0 && snapshot.lastApplyIndex >= beginIndex) {
S
Shengliang Guan 已提交
2913
      sNTrace(ths, "commit by snapshot from index:%" PRId64 " to index:%" PRId64, beginIndex, snapshot.lastApplyIndex);
2914

M
Minghao Li 已提交
2915 2916 2917
      // update begin index
      beginIndex = snapshot.lastApplyIndex + 1;
    }
2918 2919
  }

2920 2921
  int32_t    code = 0;
  ESyncState state = flag;
M
Minghao Li 已提交
2922

S
Shengliang Guan 已提交
2923
  sNTrace(ths, "commit by wal from index:%" PRId64 " to index:%" PRId64, beginIndex, endIndex);
2924 2925 2926 2927 2928 2929

  // execute fsm
  if (ths->pFsm != NULL) {
    for (SyncIndex i = beginIndex; i <= endIndex; ++i) {
      if (i != SYNC_INDEX_INVALID) {
        SSyncRaftEntry* pEntry;
2930 2931 2932 2933
        SLRUCache*      pCache = ths->pLogStore->pCache;
        LRUHandle*      h = taosLRUCacheLookup(pCache, &i, sizeof(i));
        if (h) {
          pEntry = (SSyncRaftEntry*)taosLRUCacheValue(pCache, h);
2934

2935
          ths->pLogStore->cacheHit++;
2936 2937
          sNTrace(ths, "hit cache index:%" PRId64 ", bytes:%u, %p", i, pEntry->bytes, pEntry);

2938
        } else {
2939
          ths->pLogStore->cacheMiss++;
2940 2941
          sNTrace(ths, "miss cache index:%" PRId64, i);

2942
          code = ths->pLogStore->syncLogGetEntry(ths->pLogStore, i, &pEntry);
2943 2944
          // ASSERT(code == 0);
          // ASSERT(pEntry != NULL);
M
Minghao Li 已提交
2945
          if (code != 0 || pEntry == NULL) {
S
Shengliang Guan 已提交
2946
            sNError(ths, "get log entry error");
2947
            sFatal("vgId:%d, get log entry %" PRId64 " error when commit since %s", ths->vgId, i, terrstr());
M
Minghao Li 已提交
2948 2949
            continue;
          }
2950
        }
2951

2952
        SRpcMsg rpcMsg = {0};
2953 2954
        syncEntry2OriginalRpc(pEntry, &rpcMsg);

2955
        sTrace("do commit index:%" PRId64 ", type:%s", i, TMSG_INFO(pEntry->msgType));
M
Minghao Li 已提交
2956

2957
        // user commit
2958 2959
        if ((ths->pFsm->FpCommitCb != NULL) && syncUtilUserCommit(pEntry->originalRpcType)) {
          bool internalExecute = true;
S
Shengliang Guan 已提交
2960
          if ((ths->replicaNum == 1) && ths->restoreFinish && ths->vgId != 1) {
2961 2962 2963
            internalExecute = false;
          }

M
Minghao Li 已提交
2964 2965
          sNTrace(ths, "user commit index:%" PRId64 ", internal:%d, type:%s", i, internalExecute,
                  TMSG_INFO(pEntry->msgType));
2966

2967 2968
          // execute fsm in apply thread, or execute outside syncPropose
          if (internalExecute) {
S
Shengliang Guan 已提交
2969 2970 2971 2972 2973 2974 2975 2976
            SFsmCbMeta cbMeta = {
                .index = pEntry->index,
                .lastConfigIndex = syncNodeGetSnapshotConfigIndex(ths, pEntry->index),
                .isWeak = pEntry->isWeak,
                .code = 0,
                .state = ths->state,
                .seqNum = pEntry->seqNum,
                .term = pEntry->term,
S
Shengliang Guan 已提交
2977
                .currentTerm = ths->raftStore.currentTerm,
S
Shengliang Guan 已提交
2978 2979 2980
                .flag = flag,
            };

S
Shengliang Guan 已提交
2981
            syncRespMgrGetAndDel(ths->pSyncRespMgr, cbMeta.seqNum, &rpcMsg.info);
S
Shengliang Guan 已提交
2982
            ths->pFsm->FpCommitCb(ths->pFsm, &rpcMsg, &cbMeta);
M
Minghao Li 已提交
2983
          }
2984
        }
2985

2986 2987
#if 0
        // execute in pre-commit
M
Minghao Li 已提交
2988
        // leader transfer
2989 2990
        if (pEntry->originalRpcType == TDMT_SYNC_LEADER_TRANSFER) {
          code = syncDoLeaderTransfer(ths, &rpcMsg, pEntry);
2991
          ASSERT(code == 0);
2992
        }
2993
#endif
2994 2995

        // restore finish
2996
        // if only snapshot, a noop entry will be append, so syncLogLastIndex is always ok
2997 2998 2999 3000 3001 3002
        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 已提交
3003

3004
            int64_t restoreDelay = taosGetTimestampMs() - ths->leaderTime;
S
Shengliang Guan 已提交
3005
            sNTrace(ths, "restore finish, index:%" PRId64 ", elapsed:%" PRId64 " ms", pEntry->index, restoreDelay);
3006 3007 3008 3009
          }
        }

        rpcFreeCont(rpcMsg.pCont);
3010 3011 3012
        if (h) {
          taosLRUCacheRelease(pCache, h, false);
        } else {
B
Benguang Zhao 已提交
3013
          syncEntryDestroy(pEntry);
3014
        }
3015 3016 3017 3018
      }
    }
  }
  return 0;
3019 3020 3021
}

bool syncNodeInRaftGroup(SSyncNode* ths, SRaftId* pRaftId) {
S
Shengliang Guan 已提交
3022
  for (int32_t i = 0; i < ths->replicaNum; ++i) {
3023 3024 3025 3026 3027
    if (syncUtilSameId(&((ths->replicasId)[i]), pRaftId)) {
      return true;
    }
  }
  return false;
M
Minghao Li 已提交
3028 3029 3030 3031
}

SSyncSnapshotSender* syncNodeGetSnapshotSender(SSyncNode* ths, SRaftId* pDestId) {
  SSyncSnapshotSender* pSender = NULL;
S
Shengliang Guan 已提交
3032
  for (int32_t i = 0; i < ths->replicaNum; ++i) {
M
Minghao Li 已提交
3033 3034 3035 3036 3037
    if (syncUtilSameId(pDestId, &((ths->replicasId)[i]))) {
      pSender = (ths->senders)[i];
    }
  }
  return pSender;
M
Minghao Li 已提交
3038
}
M
Minghao Li 已提交
3039

3040 3041
SSyncTimer* syncNodeGetHbTimer(SSyncNode* ths, SRaftId* pDestId) {
  SSyncTimer* pTimer = NULL;
S
Shengliang Guan 已提交
3042
  for (int32_t i = 0; i < ths->replicaNum; ++i) {
3043 3044 3045 3046 3047 3048 3049
    if (syncUtilSameId(pDestId, &((ths->replicasId)[i]))) {
      pTimer = &((ths->peerHeartbeatTimerArr)[i]);
    }
  }
  return pTimer;
}

M
Minghao Li 已提交
3050 3051
SPeerState* syncNodeGetPeerState(SSyncNode* ths, const SRaftId* pDestId) {
  SPeerState* pState = NULL;
S
Shengliang Guan 已提交
3052
  for (int32_t i = 0; i < ths->replicaNum; ++i) {
M
Minghao Li 已提交
3053 3054 3055 3056 3057 3058 3059 3060 3061
    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 已提交
3062
  if (pState == NULL) {
3063
    sError("vgId:%d, replica maybe dropped", ths->vgId);
M
Minghao Li 已提交
3064 3065
    return false;
  }
M
Minghao Li 已提交
3066 3067 3068 3069 3070 3071 3072 3073 3074 3075 3076

  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 已提交
3077 3078 3079 3080 3081 3082 3083 3084 3085 3086 3087 3088 3089 3090
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;
    }
  }

S
Shengliang Guan 已提交
3091
  for (int32_t i = 0; i < pSyncNode->peersNum; ++i) {
M
Minghao Li 已提交
3092
    SSyncSnapshotSender* pSender = syncNodeGetSnapshotSender(pSyncNode, &(pSyncNode->peersId)[i]);
M
Minghao Li 已提交
3093
    if (pSender != NULL && pSender->start) {
M
Minghao Li 已提交
3094 3095 3096 3097 3098 3099
      sError("sync cannot change3");
      return false;
    }
  }

  return true;
M
Minghao Li 已提交
3100
}