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

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

38
bool gRaftDetailLog = false;
39

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

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

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

// life cycle
static void syncFreeNode(void* param);
M
Minghao Li 已提交
57 58 59
// ---------------------------------

int32_t syncInit() {
M
Minghao Li 已提交
60 61 62 63 64 65 66 67 68 69 70
  int32_t ret = 0;

  if (!syncEnvIsStart()) {
    tsNodeRefId = taosOpenRef(200, syncFreeNode);
    if (tsNodeRefId < 0) {
      sError("failed to init node ref");
      syncCleanUp();
      ret = -1;
    } else {
      ret = syncEnvStart();
    }
M
Minghao Li 已提交
71 72
  }

M
Minghao Li 已提交
73
  return ret;
M
Minghao Li 已提交
74
}
M
Minghao Li 已提交
75

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

  if (tsNodeRefId != -1) {
    taosCloseRef(tsNodeRefId);
    tsNodeRefId = -1;
  }
M
Minghao Li 已提交
84
}
M
Minghao Li 已提交
85

M
Minghao Li 已提交
86
int64_t syncOpen(const SSyncInfo* pSyncInfo) {
M
Minghao Li 已提交
87
  SSyncNode* pSyncNode = syncNodeOpen(pSyncInfo);
M
Minghao Li 已提交
88
  ASSERT(pSyncNode != NULL);
M
Minghao Li 已提交
89

90 91 92
  if (gRaftDetailLog) {
    syncNodeLog2("syncNodeOpen open success", pSyncNode);
  }
M
Minghao Li 已提交
93

M
Minghao Li 已提交
94 95 96 97 98 99 100
  pSyncNode->rid = taosAddRef(tsNodeRefId, pSyncNode);
  if (pSyncNode->rid < 0) {
    syncFreeNode(pSyncNode);
    return -1;
  }

  return pSyncNode->rid;
M
Minghao Li 已提交
101
}
M
Minghao Li 已提交
102

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

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

  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
}

void syncStartNormal(int64_t rid) {
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
    return;
  }
M
Minghao Li 已提交
123 124 125 126 127
  syncNodeStart(pSyncNode);

  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
}

M
Minghao Li 已提交
128 129 130 131 132 133 134 135 136 137
void syncStartStandBy(int64_t rid) {
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
    return;
  }
  syncNodeStartStandBy(pSyncNode);

  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
}

M
Minghao Li 已提交
138
void syncStop(int64_t rid) {
M
Minghao Li 已提交
139 140 141 142
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
    return;
  }
M
Minghao Li 已提交
143
  syncNodeClose(pSyncNode);
M
Minghao Li 已提交
144 145 146

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

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

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

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

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

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

  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
181
  sInfo("vgId:%d, set to standby", pSyncNode->vgId);
M
Minghao Li 已提交
182 183 184
  return 0;
}

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

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

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

  return true;
}

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

M
Minghao Li 已提交
211
  if (!syncNodeCheckNewConfig(pSyncNode, pNewCfg)) {
M
Minghao Li 已提交
212
    taosReleaseRef(tsNodeRefId, pSyncNode->rid);
M
Minghao Li 已提交
213 214
    terrno = TSDB_CODE_SYN_NEW_CONFIG_ERROR;
    sError("syncNodeCheckNewConfig error");
M
Minghao Li 已提交
215
    return -1;
M
Minghao Li 已提交
216 217 218 219 220 221 222 223 224
  }

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

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

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

M
Minghao Li 已提交
238
  if (!syncNodeCheckNewConfig(pSyncNode, pNewCfg)) {
M
Minghao Li 已提交
239
    taosReleaseRef(tsNodeRefId, pSyncNode->rid);
M
Minghao Li 已提交
240 241
    terrno = TSDB_CODE_SYN_NEW_CONFIG_ERROR;
    sError("syncNodeCheckNewConfig error");
M
Minghao Li 已提交
242
    return -1;
M
Minghao Li 已提交
243
  }
244

M
Minghao Li 已提交
245
  char* newconfig = syncCfg2Str((SSyncCfg*)pNewCfg);
246
  if (gRaftDetailLog) {
247
    sInfo("==syncReconfig== newconfig:%s", newconfig);
248
  }
M
Minghao Li 已提交
249

M
Minghao Li 已提交
250 251
  int32_t ret = 0;

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

  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
M
Minghao Li 已提交
262 263
  return ret;
}
M
Minghao Li 已提交
264

265
int32_t syncLeaderTransfer(int64_t rid) {
M
Minghao Li 已提交
266 267
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
M
Minghao Li 已提交
268 269
    terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
    return -1;
M
Minghao Li 已提交
270 271 272
  }
  ASSERT(rid == pSyncNode->rid);

M
Minghao Li 已提交
273
  if (pSyncNode->peersNum == 0) {
M
Minghao Li 已提交
274
    taosReleaseRef(tsNodeRefId, pSyncNode->rid);
M
Minghao Li 已提交
275 276
    terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
    return -1;
M
Minghao Li 已提交
277 278 279 280
  }

  SNodeInfo newLeader = (pSyncNode->peersNodeInfo)[0];
  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
281

M
Minghao Li 已提交
282
  int32_t ret = syncLeaderTransferTo(rid, newLeader);
283 284 285 286 287 288
  return ret;
}

int32_t syncLeaderTransferTo(int64_t rid, SNodeInfo newLeader) {
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
M
Minghao Li 已提交
289 290
    terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
    return -1;
291
  }
M
Minghao Li 已提交
292
  ASSERT(rid == pSyncNode->rid);
293 294 295 296
  int32_t ret = 0;

  if (pSyncNode->replicaNum == 1) {
    sError("only one replica, cannot drop leader");
M
Minghao Li 已提交
297
    taosReleaseRef(tsNodeRefId, pSyncNode->rid);
M
Minghao Li 已提交
298 299
    terrno = TSDB_CODE_SYN_ONE_REPLICA;
    return -1;
300 301 302 303 304
  }

  SyncLeaderTransfer* pMsg = syncLeaderTransferBuild(pSyncNode->vgId);
  pMsg->newLeaderId.addr = syncUtilAddr2U64(newLeader.nodeFqdn, newLeader.nodePort);
  pMsg->newLeaderId.vgId = pSyncNode->vgId;
M
Minghao Li 已提交
305
  pMsg->newNodeInfo = newLeader;
306 307 308 309 310
  ASSERT(pMsg != NULL);
  SRpcMsg rpcMsg = {0};
  syncLeaderTransfer2RpcMsg(pMsg, &rpcMsg);
  syncLeaderTransferDestroy(pMsg);

M
Minghao Li 已提交
311
  ret = syncNodePropose(pSyncNode, &rpcMsg, false);
312 313 314 315 316 317 318 319 320
  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
  return ret;
}

bool syncCanLeaderTransfer(int64_t rid) {
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
    return false;
  }
M
Minghao Li 已提交
321
  ASSERT(rid == pSyncNode->rid);
322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347

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

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

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

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

348
int32_t syncForwardToPeer(int64_t rid, SRpcMsg* pMsg, bool isWeak) {
M
Minghao Li 已提交
349 350 351
  int32_t ret = syncPropose(rid, pMsg, isWeak);
  return ret;
}
M
Minghao Li 已提交
352

M
Minghao Li 已提交
353 354 355 356 357
ESyncState syncGetMyRole(int64_t rid) {
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
    return TAOS_SYNC_STATE_ERROR;
  }
M
Minghao Li 已提交
358
  ASSERT(rid == pSyncNode->rid);
M
Minghao Li 已提交
359 360 361 362
  ESyncState state = pSyncNode->state;

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

M
Minghao Li 已提交
365 366 367 368 369
bool syncIsReady(int64_t rid) {
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
    return false;
  }
M
Minghao Li 已提交
370
  ASSERT(rid == pSyncNode->rid);
M
Minghao Li 已提交
371 372
  bool b = (pSyncNode->state == TAOS_SYNC_STATE_LEADER) && pSyncNode->restoreFinish;
  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
373 374 375 376 377 378 379 380 381

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

M
Minghao Li 已提交
385 386 387 388 389
bool syncIsRestoreFinish(int64_t rid) {
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
    return false;
  }
M
Minghao Li 已提交
390
  ASSERT(rid == pSyncNode->rid);
M
Minghao Li 已提交
391 392 393 394 395 396
  bool b = pSyncNode->restoreFinish;

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

397 398 399 400 401
int32_t syncGetSnapshotMeta(int64_t rid, struct SSnapshotMeta* sMeta) {
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
    return -1;
  }
M
Minghao Li 已提交
402
  ASSERT(rid == pSyncNode->rid);
403 404
  sMeta->lastConfigIndex = pSyncNode->pRaftCfg->lastConfigIndex;

S
Shengliang Guan 已提交
405
  sTrace("vgId:%d, get snapshot meta, lastConfigIndex:%" PRId64, pSyncNode->vgId, pSyncNode->pRaftCfg->lastConfigIndex);
406 407 408 409 410

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

411 412 413 414 415
int32_t syncGetSnapshotMetaByIndex(int64_t rid, SyncIndex snapshotIndex, struct SSnapshotMeta* sMeta) {
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
    return -1;
  }
M
Minghao Li 已提交
416
  ASSERT(rid == pSyncNode->rid);
417 418 419 420 421 422 423 424 425 426 427

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

  for (int i = 0; i < pSyncNode->pRaftCfg->configIndexCount; ++i) {
    if ((pSyncNode->pRaftCfg->configIndexArr)[i] > lastIndex &&
        (pSyncNode->pRaftCfg->configIndexArr)[i] <= snapshotIndex) {
      lastIndex = (pSyncNode->pRaftCfg->configIndexArr)[i];
    }
  }
  sMeta->lastConfigIndex = lastIndex;
428
  sTrace("vgId:%d, get snapshot meta by index:%" PRId64 " lcindex:%" PRId64, pSyncNode->vgId, snapshotIndex,
S
Shengliang Guan 已提交
429
         sMeta->lastConfigIndex);
430 431 432 433 434

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

435 436 437 438 439 440 441 442 443 444
SyncIndex syncNodeGetSnapshotConfigIndex(SSyncNode* pSyncNode, SyncIndex snapshotLastApplyIndex) {
  ASSERT(pSyncNode->pRaftCfg->configIndexCount >= 1);
  SyncIndex lastIndex = (pSyncNode->pRaftCfg->configIndexArr)[0];

  for (int i = 0; i < pSyncNode->pRaftCfg->configIndexCount; ++i) {
    if ((pSyncNode->pRaftCfg->configIndexArr)[i] > lastIndex &&
        (pSyncNode->pRaftCfg->configIndexArr)[i] <= snapshotLastApplyIndex) {
      lastIndex = (pSyncNode->pRaftCfg->configIndexArr)[i];
    }
  }
445 446
  sTrace("vgId:%d, sync get snapshot last config index, index:%ld lcindex:%ld", pSyncNode->vgId, snapshotLastApplyIndex,
         lastIndex);
447 448 449 450

  return lastIndex;
}

M
Minghao Li 已提交
451 452 453 454 455
const char* syncGetMyRoleStr(int64_t rid) {
  const char* s = syncUtilState2String(syncGetMyRole(rid));
  return s;
}

M
Minghao Li 已提交
456
SyncTerm syncGetMyTerm(int64_t rid) {
M
Minghao Li 已提交
457 458 459 460
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
    return TAOS_SYNC_STATE_ERROR;
  }
M
Minghao Li 已提交
461
  ASSERT(rid == pSyncNode->rid);
M
Minghao Li 已提交
462
  SyncTerm term = pSyncNode->pRaftStore->currentTerm;
M
Minghao Li 已提交
463 464

  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
M
Minghao Li 已提交
465
  return term;
M
Minghao Li 已提交
466 467
}

M
Minghao Li 已提交
468
SyncGroupId syncGetVgId(int64_t rid) {
M
Minghao Li 已提交
469 470
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
M
Minghao Li 已提交
471 472
    return TAOS_SYNC_STATE_ERROR;
  }
M
Minghao Li 已提交
473
  ASSERT(rid == pSyncNode->rid);
M
Minghao Li 已提交
474
  SyncGroupId vgId = pSyncNode->vgId;
M
Minghao Li 已提交
475 476

  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
M
Minghao Li 已提交
477
  return vgId;
M
Minghao Li 已提交
478 479
}

M
Minghao Li 已提交
480 481 482 483 484 485
void syncGetEpSet(int64_t rid, SEpSet* pEpSet) {
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
    memset(pEpSet, 0, sizeof(*pEpSet));
    return;
  }
M
Minghao Li 已提交
486
  ASSERT(rid == pSyncNode->rid);
M
Minghao Li 已提交
487 488
  pEpSet->numOfEps = 0;
  for (int i = 0; i < pSyncNode->pRaftCfg->cfg.replicaNum; ++i) {
M
Minghao Li 已提交
489 490
    snprintf(pEpSet->eps[i].fqdn, sizeof(pEpSet->eps[i].fqdn), "%s", (pSyncNode->pRaftCfg->cfg.nodeInfo)[i].nodeFqdn);
    pEpSet->eps[i].port = (pSyncNode->pRaftCfg->cfg.nodeInfo)[i].nodePort;
M
Minghao Li 已提交
491
    (pEpSet->numOfEps)++;
492
    sInfo("vgId:%d sync get epset: index:%d %s:%d", pSyncNode->vgId, i, pEpSet->eps[i].fqdn, pEpSet->eps[i].port);
M
Minghao Li 已提交
493 494
  }
  pEpSet->inUse = pSyncNode->pRaftCfg->cfg.myIndex;
495 496 497 498
  sInfo("vgId:%d sync get epset in-use:%d", pSyncNode->vgId, pEpSet->inUse);

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

500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515
void syncGetRetryEpSet(int64_t rid, SEpSet* pEpSet) {
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
    memset(pEpSet, 0, sizeof(*pEpSet));
    return;
  }
  ASSERT(rid == pSyncNode->rid);
  pEpSet->numOfEps = 0;
  for (int i = 0; i < pSyncNode->pRaftCfg->cfg.replicaNum; ++i) {
    snprintf(pEpSet->eps[i].fqdn, sizeof(pEpSet->eps[i].fqdn), "%s", (pSyncNode->pRaftCfg->cfg.nodeInfo)[i].nodeFqdn);
    pEpSet->eps[i].port = (pSyncNode->pRaftCfg->cfg.nodeInfo)[i].nodePort;
    (pEpSet->numOfEps)++;
    sInfo("vgId:%d sync get retry epset: index:%d %s:%d", pSyncNode->vgId, i, pEpSet->eps[i].fqdn, pEpSet->eps[i].port);
  }
  pEpSet->inUse = (pSyncNode->pRaftCfg->cfg.myIndex + 1) % pEpSet->numOfEps;
  sInfo("vgId:%d sync get retry epset in-use:%d", pSyncNode->vgId, pEpSet->inUse);
M
Minghao Li 已提交
516 517 518 519

  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
}

M
Minghao Li 已提交
520 521 522 523 524
int32_t syncGetRespRpc(int64_t rid, uint64_t index, SRpcMsg* msg) {
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
    return TAOS_SYNC_STATE_ERROR;
  }
M
Minghao Li 已提交
525
  ASSERT(rid == pSyncNode->rid);
M
Minghao Li 已提交
526 527 528 529 530 531 532 533 534 535 536

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

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

S
Shengliang Guan 已提交
537
int32_t syncGetAndDelRespRpc(int64_t rid, uint64_t index, SRpcHandleInfo* pInfo) {
M
Minghao Li 已提交
538 539 540 541
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
    return TAOS_SYNC_STATE_ERROR;
  }
M
Minghao Li 已提交
542
  ASSERT(rid == pSyncNode->rid);
M
Minghao Li 已提交
543 544 545 546

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

S
Shengliang Guan 已提交
550
  sTrace("vgId:%d, get seq:%" PRIu64 " rpc handle:%p", pSyncNode->vgId, index, pInfo->handle);
M
Minghao Li 已提交
551 552 553 554
  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
  return ret;
}

555
void syncSetMsgCb(int64_t rid, const SMsgCb* msgcb) {
M
Minghao Li 已提交
556 557 558 559 560
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
    sTrace("syncSetQ get pSyncNode is NULL, rid:%ld", rid);
    return;
  }
M
Minghao Li 已提交
561
  ASSERT(rid == pSyncNode->rid);
S
Shengliang Guan 已提交
562
  pSyncNode->msgcb = msgcb;
M
Minghao Li 已提交
563 564 565 566 567 568 569 570 571 572

  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
}

char* sync2SimpleStr(int64_t rid) {
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
    sTrace("syncSetRpc get pSyncNode is NULL, rid:%ld", rid);
    return NULL;
  }
M
Minghao Li 已提交
573
  ASSERT(rid == pSyncNode->rid);
M
Minghao Li 已提交
574 575 576 577 578 579 580 581 582 583 584
  char* s = syncNode2SimpleStr(pSyncNode);
  taosReleaseRef(tsNodeRefId, pSyncNode->rid);

  return s;
}

void setPingTimerMS(int64_t rid, int32_t pingTimerMS) {
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
    return;
  }
M
Minghao Li 已提交
585
  ASSERT(rid == pSyncNode->rid);
M
Minghao Li 已提交
586 587 588 589 590 591 592 593 594 595 596
  pSyncNode->pingBaseLine = pingTimerMS;
  pSyncNode->pingTimerMS = pingTimerMS;

  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
}

void setElectTimerMS(int64_t rid, int32_t electTimerMS) {
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
    return;
  }
M
Minghao Li 已提交
597
  ASSERT(rid == pSyncNode->rid);
M
Minghao Li 已提交
598 599 600 601 602 603 604 605 606 607
  pSyncNode->electBaseLine = electTimerMS;

  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
}

void setHeartbeatTimerMS(int64_t rid, int32_t hbTimerMS) {
  SSyncNode* pSyncNode = (SSyncNode*)taosAcquireRef(tsNodeRefId, rid);
  if (pSyncNode == NULL) {
    return;
  }
M
Minghao Li 已提交
608
  ASSERT(rid == pSyncNode->rid);
M
Minghao Li 已提交
609 610 611 612 613 614
  pSyncNode->hbBaseLine = hbTimerMS;
  pSyncNode->heartbeatTimerMS = hbTimerMS;

  taosReleaseRef(tsNodeRefId, pSyncNode->rid);
}

M
Minghao Li 已提交
615
int32_t syncPropose(int64_t rid, SRpcMsg* pMsg, bool isWeak) {
M
Minghao Li 已提交
616
  int32_t ret = 0;
M
Minghao Li 已提交
617

618
  SSyncNode* pSyncNode = taosAcquireRef(tsNodeRefId, rid);
619
  if (pSyncNode == NULL) {
M
Minghao Li 已提交
620 621
    terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
    return -1;
622
  }
M
Minghao Li 已提交
623
  ASSERT(rid == pSyncNode->rid);
M
Minghao Li 已提交
624 625 626 627 628 629 630
  ret = syncNodePropose(pSyncNode, pMsg, isWeak);

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

int32_t syncNodePropose(SSyncNode* pSyncNode, const SRpcMsg* pMsg, bool isWeak) {
M
Minghao Li 已提交
631
  int32_t ret = 0;
M
Minghao Li 已提交
632 633 634 635

  char eventLog[128];
  snprintf(eventLog, sizeof(eventLog), "propose type:%s,%d", TMSG_INFO(pMsg->msgType), pMsg->msgType);
  syncNodeEventLog(pSyncNode, eventLog);
M
Minghao Li 已提交
636

M
Minghao Li 已提交
637
  if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
M
Minghao Li 已提交
638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657
    if (pSyncNode->changing && pMsg->msgType != TDMT_SYNC_CONFIG_CHANGE_FINISH) {
      ret = -1;
      terrno = TSDB_CODE_SYN_PROPOSE_NOT_READY;
      sError("sync propose not ready, type:%s,%d", TMSG_INFO(pMsg->msgType), pMsg->msgType);
      goto _END;
    }

    // config change
    if (pMsg->msgType == TDMT_SYNC_CONFIG_CHANGE) {
      if (!syncNodeCanChange(pSyncNode)) {
        ret = -1;
        terrno = TSDB_CODE_SYN_RECONFIG_NOT_READY;
        sError("sync reconfig not ready, type:%s,%d", TMSG_INFO(pMsg->msgType), pMsg->msgType);
        goto _END;
      }

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

M
Minghao Li 已提交
658 659 660 661 662 663
    SRespStub stub;
    stub.createTime = taosGetTimestampMs();
    stub.rpcMsg = *pMsg;
    uint64_t seqNum = syncRespMgrAdd(pSyncNode->pSyncRespMgr, &stub);

    SyncClientRequest* pSyncMsg = syncClientRequestBuild2(pMsg, seqNum, isWeak, pSyncNode->vgId);
M
Minghao Li 已提交
664 665
    SRpcMsg            rpcMsg;
    syncClientRequest2RpcMsg(pSyncMsg, &rpcMsg);
666 667

    if (pSyncNode->FpEqMsg != NULL && (*pSyncNode->FpEqMsg)(pSyncNode->msgcb, &rpcMsg) == 0) {
M
Minghao Li 已提交
668
      ret = 0;
M
Minghao Li 已提交
669
    } else {
M
Minghao Li 已提交
670 671
      ret = -1;
      terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
M
Minghao Li 已提交
672
      sError("syncPropose pSyncNode->FpEqMsg is NULL");
M
Minghao Li 已提交
673
    }
M
Minghao Li 已提交
674
    syncClientRequestDestroy(pSyncMsg);
M
Minghao Li 已提交
675 676
    goto _END;

M
Minghao Li 已提交
677
  } else {
M
Minghao Li 已提交
678 679
    ret = -1;
    terrno = TSDB_CODE_SYN_NOT_LEADER;
M
Minghao Li 已提交
680
    sError("syncPropose not leader, %s", syncUtilState2String(pSyncNode->state));
M
Minghao Li 已提交
681
    goto _END;
M
Minghao Li 已提交
682
  }
M
Minghao Li 已提交
683

M
Minghao Li 已提交
684
_END:
M
Minghao Li 已提交
685 686 687
  return ret;
}

M
Minghao Li 已提交
688
// open/close --------------
689 690 691
SSyncNode* syncNodeOpen(const SSyncInfo* pOldSyncInfo) {
  SSyncInfo* pSyncInfo = (SSyncInfo*)pOldSyncInfo;

wafwerar's avatar
wafwerar 已提交
692
  SSyncNode* pSyncNode = (SSyncNode*)taosMemoryMalloc(sizeof(SSyncNode));
M
Minghao Li 已提交
693
  ASSERT(pSyncNode != NULL);
M
Minghao Li 已提交
694
  memset(pSyncNode, 0, sizeof(SSyncNode));
M
Minghao Li 已提交
695

M
Minghao Li 已提交
696 697 698 699 700 701 702
  int32_t ret = 0;
  if (!taosDirExist((char*)(pSyncInfo->path))) {
    if (taosMkDir(pSyncInfo->path) != 0) {
      terrno = TAOS_SYSTEM_ERROR(errno);
      sError("failed to create dir:%s since %s", pSyncInfo->path, terrstr());
      return NULL;
    }
703
  }
M
Minghao Li 已提交
704

705 706
  snprintf(pSyncNode->configPath, sizeof(pSyncNode->configPath), "%s/raft_config.json", pSyncInfo->path);
  if (!taosCheckExistFile(pSyncNode->configPath)) {
M
Minghao Li 已提交
707 708 709 710
    // create a new raft config file
    SRaftCfgMeta meta;
    meta.isStandBy = pSyncInfo->isStandBy;
    meta.snapshotEnable = pSyncInfo->snapshotEnable;
711
    meta.lastConfigIndex = SYNC_INDEX_INVALID;
M
Minghao Li 已提交
712
    ret = raftCfgCreateFile((SSyncCfg*)&(pSyncInfo->syncCfg), meta, pSyncNode->configPath);
M
Minghao Li 已提交
713
    ASSERT(ret == 0);
714 715 716 717

  } else {
    // update syncCfg by raft_config.json
    pSyncNode->pRaftCfg = raftCfgOpen(pSyncNode->configPath);
M
Minghao Li 已提交
718
    ASSERT(pSyncNode->pRaftCfg != NULL);
719 720
    pSyncInfo->syncCfg = pSyncNode->pRaftCfg->cfg;

721 722 723 724 725
    if (gRaftDetailLog) {
      char* seralized = raftCfg2Str(pSyncNode->pRaftCfg);
      sInfo("syncNodeOpen update config :%s", seralized);
      taosMemoryFree(seralized);
    }
726 727

    raftCfgClose(pSyncNode->pRaftCfg);
M
Minghao Li 已提交
728 729
  }

M
Minghao Li 已提交
730
  // init by SSyncInfo
M
Minghao Li 已提交
731 732
  pSyncNode->vgId = pSyncInfo->vgId;
  memcpy(pSyncNode->path, pSyncInfo->path, sizeof(pSyncNode->path));
M
Minghao Li 已提交
733
  snprintf(pSyncNode->raftStorePath, sizeof(pSyncNode->raftStorePath), "%s/raft_store.json", pSyncInfo->path);
M
Minghao Li 已提交
734 735
  snprintf(pSyncNode->configPath, sizeof(pSyncNode->configPath), "%s/raft_config.json", pSyncInfo->path);

M
Minghao Li 已提交
736
  pSyncNode->pWal = pSyncInfo->pWal;
S
Shengliang Guan 已提交
737
  pSyncNode->msgcb = pSyncInfo->msgcb;
M
Minghao Li 已提交
738
  pSyncNode->FpSendMsg = pSyncInfo->FpSendMsg;
M
Minghao Li 已提交
739
  pSyncNode->FpEqMsg = pSyncInfo->FpEqMsg;
M
Minghao Li 已提交
740

M
Minghao Li 已提交
741 742
  // init raft config
  pSyncNode->pRaftCfg = raftCfgOpen(pSyncNode->configPath);
M
Minghao Li 已提交
743
  ASSERT(pSyncNode->pRaftCfg != NULL);
M
Minghao Li 已提交
744

M
Minghao Li 已提交
745
  // init internal
M
Minghao Li 已提交
746 747
  pSyncNode->myNodeInfo = pSyncNode->pRaftCfg->cfg.nodeInfo[pSyncNode->pRaftCfg->cfg.myIndex];
  syncUtilnodeInfo2raftId(&pSyncNode->myNodeInfo, pSyncNode->vgId, &pSyncNode->myRaftId);
M
Minghao Li 已提交
748

M
Minghao Li 已提交
749
  // init peersNum, peers, peersId
M
Minghao Li 已提交
750
  pSyncNode->peersNum = pSyncNode->pRaftCfg->cfg.replicaNum - 1;
M
Minghao Li 已提交
751
  int j = 0;
M
Minghao Li 已提交
752 753 754
  for (int i = 0; i < pSyncNode->pRaftCfg->cfg.replicaNum; ++i) {
    if (i != pSyncNode->pRaftCfg->cfg.myIndex) {
      pSyncNode->peersNodeInfo[j] = pSyncNode->pRaftCfg->cfg.nodeInfo[i];
M
Minghao Li 已提交
755 756 757
      j++;
    }
  }
M
Minghao Li 已提交
758
  for (int i = 0; i < pSyncNode->peersNum; ++i) {
M
Minghao Li 已提交
759
    syncUtilnodeInfo2raftId(&pSyncNode->peersNodeInfo[i], pSyncNode->vgId, &pSyncNode->peersId[i]);
M
Minghao Li 已提交
760
  }
M
Minghao Li 已提交
761

M
Minghao Li 已提交
762
  // init replicaNum, replicasId
M
Minghao Li 已提交
763 764 765
  pSyncNode->replicaNum = pSyncNode->pRaftCfg->cfg.replicaNum;
  for (int i = 0; i < pSyncNode->pRaftCfg->cfg.replicaNum; ++i) {
    syncUtilnodeInfo2raftId(&pSyncNode->pRaftCfg->cfg.nodeInfo[i], pSyncNode->vgId, &pSyncNode->replicasId[i]);
M
Minghao Li 已提交
766 767
  }

M
Minghao Li 已提交
768
  // init raft algorithm
M
Minghao Li 已提交
769
  pSyncNode->pFsm = pSyncInfo->pFsm;
M
Minghao Li 已提交
770
  pSyncNode->quorum = syncUtilQuorum(pSyncNode->pRaftCfg->cfg.replicaNum);
M
Minghao Li 已提交
771 772
  pSyncNode->leaderCache = EMPTY_RAFT_ID;

M
Minghao Li 已提交
773
  // init life cycle outside
M
Minghao Li 已提交
774

M
Minghao Li 已提交
775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798
  // 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 已提交
799
  // init TLA+ server vars
M
syncInt  
Minghao Li 已提交
800
  pSyncNode->state = TAOS_SYNC_STATE_FOLLOWER;
M
Minghao Li 已提交
801
  pSyncNode->pRaftStore = raftStoreOpen(pSyncNode->raftStorePath);
M
Minghao Li 已提交
802
  ASSERT(pSyncNode->pRaftStore != NULL);
M
Minghao Li 已提交
803

M
Minghao Li 已提交
804
  // init TLA+ candidate vars
M
Minghao Li 已提交
805
  pSyncNode->pVotesGranted = voteGrantedCreate(pSyncNode);
M
Minghao Li 已提交
806
  ASSERT(pSyncNode->pVotesGranted != NULL);
M
Minghao Li 已提交
807
  pSyncNode->pVotesRespond = votesRespondCreate(pSyncNode);
M
Minghao Li 已提交
808
  ASSERT(pSyncNode->pVotesRespond != NULL);
M
Minghao Li 已提交
809

M
Minghao Li 已提交
810 811
  // init TLA+ leader vars
  pSyncNode->pNextIndex = syncIndexMgrCreate(pSyncNode);
M
Minghao Li 已提交
812
  ASSERT(pSyncNode->pNextIndex != NULL);
M
Minghao Li 已提交
813
  pSyncNode->pMatchIndex = syncIndexMgrCreate(pSyncNode);
M
Minghao Li 已提交
814
  ASSERT(pSyncNode->pMatchIndex != NULL);
M
Minghao Li 已提交
815 816 817

  // init TLA+ log vars
  pSyncNode->pLogStore = logStoreCreate(pSyncNode);
M
Minghao Li 已提交
818
  ASSERT(pSyncNode->pLogStore != NULL);
M
Minghao Li 已提交
819
  pSyncNode->commitIndex = SYNC_INDEX_INVALID;
M
Minghao Li 已提交
820

M
Minghao Li 已提交
821 822 823 824 825
  // timer ms init
  pSyncNode->pingBaseLine = PING_TIMER_MS;
  pSyncNode->electBaseLine = ELECT_TIMER_MS_MIN;
  pSyncNode->hbBaseLine = HEARTBEAT_TIMER_MS;

M
Minghao Li 已提交
826
  // init ping timer
M
Minghao Li 已提交
827
  pSyncNode->pPingTimer = NULL;
M
Minghao Li 已提交
828
  pSyncNode->pingTimerMS = pSyncNode->pingBaseLine;
M
Minghao Li 已提交
829 830
  atomic_store_64(&pSyncNode->pingTimerLogicClock, 0);
  atomic_store_64(&pSyncNode->pingTimerLogicClockUser, 0);
M
Minghao Li 已提交
831
  pSyncNode->FpPingTimerCB = syncNodeEqPingTimer;
M
Minghao Li 已提交
832
  pSyncNode->pingTimerCounter = 0;
M
Minghao Li 已提交
833

M
Minghao Li 已提交
834 835
  // init elect timer
  pSyncNode->pElectTimer = NULL;
M
Minghao Li 已提交
836
  pSyncNode->electTimerMS = syncUtilElectRandomMS(pSyncNode->electBaseLine, 2 * pSyncNode->electBaseLine);
M
Minghao Li 已提交
837 838
  atomic_store_64(&pSyncNode->electTimerLogicClock, 0);
  atomic_store_64(&pSyncNode->electTimerLogicClockUser, 0);
M
Minghao Li 已提交
839
  pSyncNode->FpElectTimerCB = syncNodeEqElectTimer;
M
Minghao Li 已提交
840 841 842 843
  pSyncNode->electTimerCounter = 0;

  // init heartbeat timer
  pSyncNode->pHeartbeatTimer = NULL;
M
Minghao Li 已提交
844
  pSyncNode->heartbeatTimerMS = pSyncNode->hbBaseLine;
M
Minghao Li 已提交
845 846
  atomic_store_64(&pSyncNode->heartbeatTimerLogicClock, 0);
  atomic_store_64(&pSyncNode->heartbeatTimerLogicClockUser, 0);
M
Minghao Li 已提交
847
  pSyncNode->FpHeartbeatTimerCB = syncNodeEqHeartbeatTimer;
M
Minghao Li 已提交
848 849
  pSyncNode->heartbeatTimerCounter = 0;

M
Minghao Li 已提交
850
  // init callback
M
Minghao Li 已提交
851 852
  pSyncNode->FpOnPing = syncNodeOnPingCb;
  pSyncNode->FpOnPingReply = syncNodeOnPingReplyCb;
M
Minghao Li 已提交
853
  pSyncNode->FpOnClientRequest = syncNodeOnClientRequestCb;
M
Minghao Li 已提交
854
  pSyncNode->FpOnTimeout = syncNodeOnTimeoutCb;
M
Minghao Li 已提交
855

M
Minghao Li 已提交
856 857 858
  pSyncNode->FpOnSnapshotSend = syncNodeOnSnapshotSendCb;
  pSyncNode->FpOnSnapshotRsp = syncNodeOnSnapshotRspCb;

M
Minghao Li 已提交
859
  if (pSyncNode->pRaftCfg->snapshotEnable) {
M
Minghao Li 已提交
860
    sInfo("sync node use snapshot");
M
Minghao Li 已提交
861 862 863 864
    pSyncNode->FpOnRequestVote = syncNodeOnRequestVoteSnapshotCb;
    pSyncNode->FpOnRequestVoteReply = syncNodeOnRequestVoteReplySnapshotCb;
    pSyncNode->FpOnAppendEntries = syncNodeOnAppendEntriesSnapshotCb;
    pSyncNode->FpOnAppendEntriesReply = syncNodeOnAppendEntriesReplySnapshotCb;
M
Minghao Li 已提交
865 866 867 868 869 870 871

  } else {
    sInfo("sync node do not use snapshot");
    pSyncNode->FpOnRequestVote = syncNodeOnRequestVoteCb;
    pSyncNode->FpOnRequestVoteReply = syncNodeOnRequestVoteReplyCb;
    pSyncNode->FpOnAppendEntries = syncNodeOnAppendEntriesCb;
    pSyncNode->FpOnAppendEntriesReply = syncNodeOnAppendEntriesReplyCb;
M
Minghao Li 已提交
872 873
  }

M
Minghao Li 已提交
874
  // tools
M
Minghao Li 已提交
875
  pSyncNode->pSyncRespMgr = syncRespMgrCreate(pSyncNode, 0);
M
Minghao Li 已提交
876
  ASSERT(pSyncNode->pSyncRespMgr != NULL);
M
Minghao Li 已提交
877

878 879
  // restore state
  pSyncNode->restoreFinish = false;
880 881

  // pSyncNode->pSnapshot = NULL;
882
  // if (pSyncNode->pFsm->FpGetSnapshotInfo != NULL) {
883
  //   pSyncNode->pSnapshot = taosMemoryMalloc(sizeof(SSnapshot));
884
  //   pSyncNode->pFsm->FpGetSnapshotInfo(pSyncNode->pFsm, pSyncNode->pSnapshot);
885
  // }
886
  // tsem_init(&(pSyncNode->restoreSem), 0, 0);
887

M
Minghao Li 已提交
888 889 890 891 892 893 894 895
  // snapshot senders
  for (int i = 0; i < TSDB_MAX_REPLICA; ++i) {
    SSyncSnapshotSender* pSender = snapshotSenderCreate(pSyncNode, i);
    // ASSERT(pSender != NULL);
    (pSyncNode->senders)[i] = pSender;
  }

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

M
Minghao Li 已提交
898 899 900
  // is config changing
  pSyncNode->changing = false;

M
Minghao Li 已提交
901
  // start in syncNodeStart
M
Minghao Li 已提交
902
  // start raft
M
Minghao Li 已提交
903
  // syncNodeBecomeFollower(pSyncNode);
M
Minghao Li 已提交
904

M
Minghao Li 已提交
905
  syncNodeEventLog(pSyncNode, "sync open");
906

M
Minghao Li 已提交
907 908 909
  return pSyncNode;
}

M
Minghao Li 已提交
910 911
void syncNodeStart(SSyncNode* pSyncNode) {
  // start raft
912
  if (pSyncNode->replicaNum == 1) {
M
Minghao Li 已提交
913
    raftStoreNextTerm(pSyncNode->pRaftStore);
914
    syncNodeBecomeLeader(pSyncNode, "one replica start");
M
format  
Minghao Li 已提交
915

916
    // Raft 3.6.2 Committing entries from previous terms
917 918 919
    syncNodeAppendNoop(pSyncNode);
    syncMaybeAdvanceCommitIndex(pSyncNode);

920 921 922
    return;
  }

923
  syncNodeBecomeFollower(pSyncNode, "first start");
M
Minghao Li 已提交
924

925
  // int32_t ret = 0;
M
Minghao Li 已提交
926
  // ret = syncNodeStartPingTimer(pSyncNode);
M
Minghao Li 已提交
927
  // ASSERT(ret == 0);
928

929 930
  if (gRaftDetailLog) {
    syncNodeLog2("==state change become leader immediately==", pSyncNode);
931
  }
M
Minghao Li 已提交
932 933
}

M
Minghao Li 已提交
934 935 936 937 938 939 940 941 942 943 944
void syncNodeStartStandBy(SSyncNode* pSyncNode) {
  // state change
  pSyncNode->state = TAOS_SYNC_STATE_FOLLOWER;
  syncNodeStopHeartbeatTimer(pSyncNode);

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

M
Minghao Li 已提交
945
void syncNodeClose(SSyncNode* pSyncNode) {
M
Minghao Li 已提交
946
  syncNodeEventLog(pSyncNode, "sync close");
947

M
Minghao Li 已提交
948
  int32_t ret;
M
Minghao Li 已提交
949
  ASSERT(pSyncNode != NULL);
M
Minghao Li 已提交
950 951

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

M
Minghao Li 已提交
954
  syncRespMgrDestroy(pSyncNode->pSyncRespMgr);
M
Minghao Li 已提交
955 956 957 958 959
  voteGrantedDestroy(pSyncNode->pVotesGranted);
  votesRespondDestory(pSyncNode->pVotesRespond);
  syncIndexMgrDestroy(pSyncNode->pNextIndex);
  syncIndexMgrDestroy(pSyncNode->pMatchIndex);
  logStoreDestory(pSyncNode->pLogStore);
M
Minghao Li 已提交
960
  raftCfgClose(pSyncNode->pRaftCfg);
M
Minghao Li 已提交
961 962 963 964 965

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

M
Minghao Li 已提交
966 967 968 969
  if (pSyncNode->pFsm != NULL) {
    taosMemoryFree(pSyncNode->pFsm);
  }

M
Minghao Li 已提交
970 971 972 973 974 975 976
  for (int i = 0; i < TSDB_MAX_REPLICA; ++i) {
    if ((pSyncNode->senders)[i] != NULL) {
      snapshotSenderDestroy((pSyncNode->senders)[i]);
      (pSyncNode->senders)[i] = NULL;
    }
  }

M
Minghao Li 已提交
977 978 979 980 981
  if (pSyncNode->pNewNodeReceiver != NULL) {
    snapshotReceiverDestroy(pSyncNode->pNewNodeReceiver);
    pSyncNode->pNewNodeReceiver = NULL;
  }

982
  /*
983 984 985
  if (pSyncNode->pSnapshot != NULL) {
    taosMemoryFree(pSyncNode->pSnapshot);
  }
986
  */
987

988
  // tsem_destroy(&pSyncNode->restoreSem);
989

M
Minghao Li 已提交
990 991
  // free memory in syncFreeNode
  // taosMemoryFree(pSyncNode);
M
Minghao Li 已提交
992 993
}

M
Minghao Li 已提交
994 995 996
// option
bool syncNodeSnapshotEnable(SSyncNode* pSyncNode) { return pSyncNode->pRaftCfg->snapshotEnable; }

M
Minghao Li 已提交
997 998 999 1000 1001 1002 1003 1004 1005 1006 1007 1008 1009 1010 1011
// ping --------------
int32_t syncNodePing(SSyncNode* pSyncNode, const SRaftId* destRaftId, SyncPing* pMsg) {
  syncPingLog2((char*)"==syncNodePing==", pMsg);
  int32_t ret = 0;

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

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

int32_t syncNodePingSelf(SSyncNode* pSyncNode) {
  int32_t   ret = 0;
M
Minghao Li 已提交
1012
  SyncPing* pMsg = syncPingBuild3(&pSyncNode->myRaftId, &pSyncNode->myRaftId, pSyncNode->vgId);
M
Minghao Li 已提交
1013
  ret = syncNodePing(pSyncNode, &pMsg->destId, pMsg);
M
Minghao Li 已提交
1014
  ASSERT(ret == 0);
M
Minghao Li 已提交
1015 1016 1017 1018 1019 1020 1021 1022

  syncPingDestroy(pMsg);
  return ret;
}

int32_t syncNodePingPeers(SSyncNode* pSyncNode) {
  int32_t ret = 0;
  for (int i = 0; i < pSyncNode->peersNum; ++i) {
M
Minghao Li 已提交
1023 1024 1025
    SRaftId*  destId = &(pSyncNode->peersId[i]);
    SyncPing* pMsg = syncPingBuild3(&pSyncNode->myRaftId, destId, pSyncNode->vgId);
    ret = syncNodePing(pSyncNode, destId, pMsg);
M
Minghao Li 已提交
1026
    ASSERT(ret == 0);
M
Minghao Li 已提交
1027 1028 1029 1030 1031 1032 1033
    syncPingDestroy(pMsg);
  }
  return ret;
}

int32_t syncNodePingAll(SSyncNode* pSyncNode) {
  int32_t ret = 0;
M
Minghao Li 已提交
1034 1035 1036 1037
  for (int i = 0; i < pSyncNode->pRaftCfg->cfg.replicaNum; ++i) {
    SRaftId*  destId = &(pSyncNode->replicasId[i]);
    SyncPing* pMsg = syncPingBuild3(&pSyncNode->myRaftId, destId, pSyncNode->vgId);
    ret = syncNodePing(pSyncNode, destId, pMsg);
M
Minghao Li 已提交
1038
    ASSERT(ret == 0);
M
Minghao Li 已提交
1039 1040 1041 1042 1043 1044 1045 1046
    syncPingDestroy(pMsg);
  }
  return ret;
}

// timer control --------------
int32_t syncNodeStartPingTimer(SSyncNode* pSyncNode) {
  int32_t ret = 0;
1047 1048 1049 1050 1051 1052 1053
  if (syncEnvIsStart()) {
    taosTmrReset(pSyncNode->FpPingTimerCB, pSyncNode->pingTimerMS, pSyncNode, gSyncEnv->pTimerManager,
                 &pSyncNode->pPingTimer);
    atomic_store_64(&pSyncNode->pingTimerLogicClock, pSyncNode->pingTimerLogicClockUser);
  } else {
    sError("sync env is stop, syncNodeStartPingTimer");
  }
M
Minghao Li 已提交
1054 1055 1056 1057 1058 1059 1060 1061 1062 1063 1064 1065 1066
  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;
1067 1068 1069 1070 1071 1072 1073 1074
  if (syncEnvIsStart()) {
    pSyncNode->electTimerMS = ms;
    taosTmrReset(pSyncNode->FpElectTimerCB, pSyncNode->electTimerMS, pSyncNode, gSyncEnv->pTimerManager,
                 &pSyncNode->pElectTimer);
    atomic_store_64(&pSyncNode->electTimerLogicClock, pSyncNode->electTimerLogicClockUser);
  } else {
    sError("sync env is stop, syncNodeStartElectTimer");
  }
M
Minghao Li 已提交
1075 1076 1077 1078 1079 1080 1081 1082 1083 1084 1085 1086 1087 1088 1089 1090 1091 1092
  return ret;
}

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

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

M
Minghao Li 已提交
1093 1094
int32_t syncNodeResetElectTimer(SSyncNode* pSyncNode) {
  int32_t ret = 0;
M
Minghao Li 已提交
1095 1096 1097 1098 1099 1100 1101
  int32_t electMS;

  if (pSyncNode->pRaftCfg->isStandBy) {
    electMS = TIMER_MAX_MS;
  } else {
    electMS = syncUtilElectRandomMS(pSyncNode->electBaseLine, 2 * pSyncNode->electBaseLine);
  }
M
Minghao Li 已提交
1102 1103 1104 1105
  ret = syncNodeRestartElectTimer(pSyncNode, electMS);
  return ret;
}

M
Minghao Li 已提交
1106 1107
int32_t syncNodeStartHeartbeatTimer(SSyncNode* pSyncNode) {
  int32_t ret = 0;
1108 1109 1110 1111 1112 1113 1114
  if (syncEnvIsStart()) {
    taosTmrReset(pSyncNode->FpHeartbeatTimerCB, pSyncNode->heartbeatTimerMS, pSyncNode, gSyncEnv->pTimerManager,
                 &pSyncNode->pHeartbeatTimer);
    atomic_store_64(&pSyncNode->heartbeatTimerLogicClock, pSyncNode->heartbeatTimerLogicClockUser);
  } else {
    sError("sync env is stop, syncNodeStartHeartbeatTimer");
  }
M
Minghao Li 已提交
1115 1116 1117 1118 1119 1120 1121 1122 1123 1124 1125 1126 1127 1128 1129
  return ret;
}

int32_t syncNodeStopHeartbeatTimer(SSyncNode* pSyncNode) {
  int32_t ret = 0;
  atomic_add_fetch_64(&pSyncNode->heartbeatTimerLogicClockUser, 1);
  taosTmrStop(pSyncNode->pHeartbeatTimer);
  pSyncNode->pHeartbeatTimer = NULL;
  return ret;
}

// utils --------------
int32_t syncNodeSendMsgById(const SRaftId* destRaftId, SSyncNode* pSyncNode, SRpcMsg* pMsg) {
  SEpSet epSet;
  syncUtilraftId2EpSet(destRaftId, &epSet);
M
Minghao Li 已提交
1130
  if (pSyncNode->FpSendMsg != NULL) {
M
Minghao Li 已提交
1131 1132 1133 1134 1135 1136
    if (gRaftDetailLog) {
      char* JsonStr = syncRpcMsg2Str(pMsg);
      syncUtilJson2Line(JsonStr);
      sTrace("sync send msg, vgId:%d, type:%d, msg:%s", pSyncNode->vgId, pMsg->msgType, JsonStr);
      taosMemoryFree(JsonStr);
    }
M
Minghao Li 已提交
1137

M
Minghao Li 已提交
1138 1139 1140
    // htonl
    syncUtilMsgHtoN(pMsg->pCont);

1141
    pMsg->info.noResp = 1;
S
Shengliang Guan 已提交
1142
    pSyncNode->FpSendMsg(&epSet, pMsg);
M
Minghao Li 已提交
1143 1144 1145
  } else {
    sTrace("syncNodeSendMsgById pSyncNode->FpSendMsg is NULL");
  }
M
Minghao Li 已提交
1146 1147 1148 1149 1150 1151
  return 0;
}

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

1156
    pMsg->info.noResp = 1;
S
Shengliang Guan 已提交
1157
    pSyncNode->FpSendMsg(&epSet, pMsg);
M
Minghao Li 已提交
1158 1159 1160
  } else {
    sTrace("syncNodeSendMsgByInfo pSyncNode->FpSendMsg is NULL");
  }
M
Minghao Li 已提交
1161 1162 1163
  return 0;
}

M
Minghao Li 已提交
1164
cJSON* syncNode2Json(const SSyncNode* pSyncNode) {
C
Cary Xu 已提交
1165
  char   u64buf[128] = {0};
M
Minghao Li 已提交
1166 1167
  cJSON* pRoot = cJSON_CreateObject();

M
Minghao Li 已提交
1168 1169 1170
  if (pSyncNode != NULL) {
    // init by SSyncInfo
    cJSON_AddNumberToObject(pRoot, "vgId", pSyncNode->vgId);
M
Minghao Li 已提交
1171
    cJSON_AddItemToObject(pRoot, "SRaftCfg", raftCfg2Json(pSyncNode->pRaftCfg));
M
Minghao Li 已提交
1172
    cJSON_AddStringToObject(pRoot, "path", pSyncNode->path);
M
Minghao Li 已提交
1173 1174 1175
    cJSON_AddStringToObject(pRoot, "raftStorePath", pSyncNode->raftStorePath);
    cJSON_AddStringToObject(pRoot, "configPath", pSyncNode->configPath);

M
Minghao Li 已提交
1176 1177 1178
    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->pWal);
    cJSON_AddStringToObject(pRoot, "pWal", u64buf);

S
Shengliang Guan 已提交
1179
    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->msgcb);
M
Minghao Li 已提交
1180 1181 1182 1183
    cJSON_AddStringToObject(pRoot, "rpcClient", u64buf);
    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpSendMsg);
    cJSON_AddStringToObject(pRoot, "FpSendMsg", u64buf);

S
Shengliang Guan 已提交
1184
    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->msgcb);
M
Minghao Li 已提交
1185 1186 1187 1188 1189 1190 1191 1192 1193 1194 1195 1196 1197 1198 1199 1200 1201 1202 1203 1204 1205
    cJSON_AddStringToObject(pRoot, "queue", u64buf);
    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpEqMsg);
    cJSON_AddStringToObject(pRoot, "FpEqMsg", u64buf);

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

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

M
Minghao Li 已提交
1207 1208 1209 1210 1211 1212
    cJSON_AddNumberToObject(pRoot, "replicaNum", pSyncNode->replicaNum);
    cJSON* pReplicasId = cJSON_CreateArray();
    cJSON_AddItemToObject(pRoot, "replicasId", pReplicasId);
    for (int i = 0; i < pSyncNode->replicaNum; ++i) {
      cJSON_AddItemToArray(pReplicasId, syncUtilRaftId2Json(&pSyncNode->replicasId[i]));
    }
M
Minghao Li 已提交
1213

M
Minghao Li 已提交
1214 1215 1216 1217 1218 1219 1220
    // raft algorithm
    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->pFsm);
    cJSON_AddStringToObject(pRoot, "pFsm", u64buf);
    cJSON_AddNumberToObject(pRoot, "quorum", pSyncNode->quorum);
    cJSON* pLaderCache = syncUtilRaftId2Json(&pSyncNode->leaderCache);
    cJSON_AddItemToObject(pRoot, "leaderCache", pLaderCache);

M
Minghao Li 已提交
1221 1222 1223 1224
    // life cycle
    snprintf(u64buf, sizeof(u64buf), "%ld", pSyncNode->rid);
    cJSON_AddStringToObject(pRoot, "rid", u64buf);

M
Minghao Li 已提交
1225 1226 1227
    // tla+ server vars
    cJSON_AddNumberToObject(pRoot, "state", pSyncNode->state);
    cJSON_AddStringToObject(pRoot, "state_str", syncUtilState2String(pSyncNode->state));
M
Minghao Li 已提交
1228
    cJSON_AddItemToObject(pRoot, "pRaftStore", raftStore2Json(pSyncNode->pRaftStore));
M
Minghao Li 已提交
1229 1230 1231 1232 1233 1234 1235 1236 1237 1238 1239

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

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

    // tla+ log vars
    cJSON_AddItemToObject(pRoot, "pLogStore", logStore2Json(pSyncNode->pLogStore));
M
Minghao Li 已提交
1240
    snprintf(u64buf, sizeof(u64buf), "%" PRId64 "", pSyncNode->commitIndex);
M
Minghao Li 已提交
1241 1242
    cJSON_AddStringToObject(pRoot, "commitIndex", u64buf);

M
Minghao Li 已提交
1243 1244 1245 1246 1247
    // timer ms init
    cJSON_AddNumberToObject(pRoot, "pingBaseLine", pSyncNode->pingBaseLine);
    cJSON_AddNumberToObject(pRoot, "electBaseLine", pSyncNode->electBaseLine);
    cJSON_AddNumberToObject(pRoot, "hbBaseLine", pSyncNode->hbBaseLine);

M
Minghao Li 已提交
1248 1249 1250 1251
    // ping timer
    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->pPingTimer);
    cJSON_AddStringToObject(pRoot, "pPingTimer", u64buf);
    cJSON_AddNumberToObject(pRoot, "pingTimerMS", pSyncNode->pingTimerMS);
M
Minghao Li 已提交
1252
    snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pSyncNode->pingTimerLogicClock);
M
Minghao Li 已提交
1253
    cJSON_AddStringToObject(pRoot, "pingTimerLogicClock", u64buf);
M
Minghao Li 已提交
1254
    snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pSyncNode->pingTimerLogicClockUser);
M
Minghao Li 已提交
1255 1256 1257
    cJSON_AddStringToObject(pRoot, "pingTimerLogicClockUser", u64buf);
    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpPingTimerCB);
    cJSON_AddStringToObject(pRoot, "FpPingTimerCB", u64buf);
M
Minghao Li 已提交
1258
    snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pSyncNode->pingTimerCounter);
M
Minghao Li 已提交
1259 1260 1261 1262 1263 1264
    cJSON_AddStringToObject(pRoot, "pingTimerCounter", u64buf);

    // elect timer
    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->pElectTimer);
    cJSON_AddStringToObject(pRoot, "pElectTimer", u64buf);
    cJSON_AddNumberToObject(pRoot, "electTimerMS", pSyncNode->electTimerMS);
M
Minghao Li 已提交
1265
    snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pSyncNode->electTimerLogicClock);
M
Minghao Li 已提交
1266
    cJSON_AddStringToObject(pRoot, "electTimerLogicClock", u64buf);
M
Minghao Li 已提交
1267
    snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pSyncNode->electTimerLogicClockUser);
M
Minghao Li 已提交
1268 1269 1270
    cJSON_AddStringToObject(pRoot, "electTimerLogicClockUser", u64buf);
    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpElectTimerCB);
    cJSON_AddStringToObject(pRoot, "FpElectTimerCB", u64buf);
M
Minghao Li 已提交
1271
    snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pSyncNode->electTimerCounter);
M
Minghao Li 已提交
1272 1273 1274 1275 1276 1277
    cJSON_AddStringToObject(pRoot, "electTimerCounter", u64buf);

    // heartbeat timer
    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->pHeartbeatTimer);
    cJSON_AddStringToObject(pRoot, "pHeartbeatTimer", u64buf);
    cJSON_AddNumberToObject(pRoot, "heartbeatTimerMS", pSyncNode->heartbeatTimerMS);
M
Minghao Li 已提交
1278
    snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pSyncNode->heartbeatTimerLogicClock);
M
Minghao Li 已提交
1279
    cJSON_AddStringToObject(pRoot, "heartbeatTimerLogicClock", u64buf);
M
Minghao Li 已提交
1280
    snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pSyncNode->heartbeatTimerLogicClockUser);
M
Minghao Li 已提交
1281 1282 1283
    cJSON_AddStringToObject(pRoot, "heartbeatTimerLogicClockUser", u64buf);
    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpHeartbeatTimerCB);
    cJSON_AddStringToObject(pRoot, "FpHeartbeatTimerCB", u64buf);
M
Minghao Li 已提交
1284
    snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pSyncNode->heartbeatTimerCounter);
M
Minghao Li 已提交
1285 1286 1287 1288 1289 1290 1291 1292 1293 1294 1295 1296 1297 1298 1299 1300 1301
    cJSON_AddStringToObject(pRoot, "heartbeatTimerCounter", u64buf);

    // callback
    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpOnPing);
    cJSON_AddStringToObject(pRoot, "FpOnPing", u64buf);
    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpOnPingReply);
    cJSON_AddStringToObject(pRoot, "FpOnPingReply", u64buf);
    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpOnRequestVote);
    cJSON_AddStringToObject(pRoot, "FpOnRequestVote", u64buf);
    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpOnRequestVoteReply);
    cJSON_AddStringToObject(pRoot, "FpOnRequestVoteReply", u64buf);
    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpOnAppendEntries);
    cJSON_AddStringToObject(pRoot, "FpOnAppendEntries", u64buf);
    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpOnAppendEntriesReply);
    cJSON_AddStringToObject(pRoot, "FpOnAppendEntriesReply", u64buf);
    snprintf(u64buf, sizeof(u64buf), "%p", pSyncNode->FpOnTimeout);
    cJSON_AddStringToObject(pRoot, "FpOnTimeout", u64buf);
M
Minghao Li 已提交
1302 1303 1304 1305 1306 1307 1308 1309 1310 1311 1312 1313 1314

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

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

    // snapshot receivers
    cJSON* pReceivers = cJSON_CreateArray();
1315
    cJSON_AddItemToObject(pRoot, "receiver", snapshotReceiver2Json(pSyncNode->pNewNodeReceiver));
M
Minghao Li 已提交
1316 1317 1318

    // changing
    cJSON_AddNumberToObject(pRoot, "changing", pSyncNode->changing);
M
Minghao Li 已提交
1319 1320 1321 1322 1323 1324 1325
  }

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

M
Minghao Li 已提交
1326 1327 1328 1329 1330 1331 1332
char* syncNode2Str(const SSyncNode* pSyncNode) {
  cJSON* pJson = syncNode2Json(pSyncNode);
  char*  serialized = cJSON_Print(pJson);
  cJSON_Delete(pJson);
  return serialized;
}

M
Minghao Li 已提交
1333 1334
void syncNodeEventLog(const SSyncNode* pSyncNode, char* str) {
  int32_t userStrLen = strlen(str);
M
Minghao Li 已提交
1335 1336 1337 1338 1339 1340

  SSnapshot snapshot = {.data = NULL, .lastApplyIndex = -1, .lastApplyTerm = 0};
  if (pSyncNode->pFsm->FpGetSnapshotInfo != NULL) {
    pSyncNode->pFsm->FpGetSnapshotInfo(pSyncNode->pFsm, &snapshot);
  }
  SyncIndex logLastIndex = pSyncNode->pLogStore->syncLogLastIndex(pSyncNode->pLogStore);
M
Minghao Li 已提交
1341
  SyncIndex logBeginIndex = pSyncNode->pLogStore->syncLogBeginIndex(pSyncNode->pLogStore);
M
Minghao Li 已提交
1342

M
Minghao Li 已提交
1343
  char* pCfgStr = syncCfg2SimpleStr(&(pSyncNode->pRaftCfg->cfg));
1344 1345 1346 1347
  char* printStr = "";
  if (pCfgStr != NULL) {
    printStr = pCfgStr;
  }
M
Minghao Li 已提交
1348

M
Minghao Li 已提交
1349
  if (userStrLen < 256) {
M
Minghao Li 已提交
1350
    char logBuf[256 + 256];
1351 1352 1353 1354
    if (pSyncNode != NULL && pSyncNode->pRaftCfg != NULL && pSyncNode->pRaftStore != NULL) {
      snprintf(logBuf, sizeof(logBuf),
               "vgId:%d, sync %s %s, term:%lu, commit:%ld, beginlog:%ld, lastlog:%ld, lastsnapshot:%ld, standby:%d, "
               "replica-num:%d, "
M
Minghao Li 已提交
1355
               "lconfig:%ld, changing:%d, restore:%d, %s",
1356 1357 1358
               pSyncNode->vgId, syncUtilState2String(pSyncNode->state), str, pSyncNode->pRaftStore->currentTerm,
               pSyncNode->commitIndex, logBeginIndex, logLastIndex, snapshot.lastApplyIndex,
               pSyncNode->pRaftCfg->isStandBy, pSyncNode->replicaNum, pSyncNode->pRaftCfg->lastConfigIndex,
M
Minghao Li 已提交
1359
               pSyncNode->changing, pSyncNode->restoreFinish, printStr);
1360 1361 1362
    } else {
      snprintf(logBuf, sizeof(logBuf), "%s", str);
    }
M
Minghao Li 已提交
1363
    sDebug("%s", logBuf);
M
Minghao Li 已提交
1364

M
Minghao Li 已提交
1365
  } else {
M
Minghao Li 已提交
1366
    int   len = 256 + userStrLen;
M
Minghao Li 已提交
1367
    char* s = (char*)taosMemoryMalloc(len);
1368 1369 1370 1371
    if (pSyncNode != NULL && pSyncNode->pRaftCfg != NULL && pSyncNode->pRaftStore != NULL) {
      snprintf(s, len,
               "vgId:%d, sync %s %s, term:%lu, commit:%ld, beginlog:%ld, lastlog:%ld, lastsnapshot:%ld, standby:%d, "
               "replica-num:%d, "
M
Minghao Li 已提交
1372
               "lconfig:%ld, changing:%d, restore:%d, %s",
1373 1374 1375
               pSyncNode->vgId, syncUtilState2String(pSyncNode->state), str, pSyncNode->pRaftStore->currentTerm,
               pSyncNode->commitIndex, logBeginIndex, logLastIndex, snapshot.lastApplyIndex,
               pSyncNode->pRaftCfg->isStandBy, pSyncNode->replicaNum, pSyncNode->pRaftCfg->lastConfigIndex,
M
Minghao Li 已提交
1376
               pSyncNode->changing, pSyncNode->restoreFinish, printStr);
1377 1378 1379
    } else {
      snprintf(s, len, "%s", str);
    }
M
Minghao Li 已提交
1380 1381 1382
    sDebug("%s", s);
    taosMemoryFree(s);
  }
M
Minghao Li 已提交
1383 1384

  taosMemoryFree(pCfgStr);
M
Minghao Li 已提交
1385 1386
}

M
Minghao Li 已提交
1387 1388 1389 1390 1391 1392 1393 1394 1395 1396 1397 1398
void syncNodeErrorLog(const SSyncNode* pSyncNode, char* str) {
  int32_t userStrLen = strlen(str);

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

  if (userStrLen < 256) {
    char logBuf[128 + 256];
1399 1400 1401 1402 1403 1404 1405 1406 1407 1408 1409 1410
    if (pSyncNode != NULL && pSyncNode->pRaftCfg != NULL && pSyncNode->pRaftStore != NULL) {
      snprintf(logBuf, sizeof(logBuf),
               "vgId:%d, sync %s %s, term:%lu, commit:%ld, beginlog:%ld, lastlog:%ld, lastsnapshot:%ld, standby:%d, "
               "replica-num:%d, "
               "lconfig:%ld, changing:%d",
               pSyncNode->vgId, syncUtilState2String(pSyncNode->state), str, pSyncNode->pRaftStore->currentTerm,
               pSyncNode->commitIndex, logBeginIndex, logLastIndex, snapshot.lastApplyIndex,
               pSyncNode->pRaftCfg->isStandBy, pSyncNode->replicaNum, pSyncNode->pRaftCfg->lastConfigIndex,
               pSyncNode->changing);
    } else {
      snprintf(logBuf, sizeof(logBuf), "%s", str);
    }
M
Minghao Li 已提交
1411 1412 1413 1414 1415
    sError("%s", logBuf);

  } else {
    int   len = 128 + userStrLen;
    char* s = (char*)taosMemoryMalloc(len);
1416 1417 1418 1419 1420 1421 1422 1423 1424 1425 1426 1427
    if (pSyncNode != NULL && pSyncNode->pRaftCfg != NULL && pSyncNode->pRaftStore != NULL) {
      snprintf(s, len,
               "vgId:%d, sync %s %s, term:%lu, commit:%ld, beginlog:%ld, lastlog:%ld, lastsnapshot:%ld, standby:%d, "
               "replica-num:%d, "
               "lconfig:%ld, changing:%d",
               pSyncNode->vgId, syncUtilState2String(pSyncNode->state), str, pSyncNode->pRaftStore->currentTerm,
               pSyncNode->commitIndex, logBeginIndex, logLastIndex, snapshot.lastApplyIndex,
               pSyncNode->pRaftCfg->isStandBy, pSyncNode->replicaNum, pSyncNode->pRaftCfg->lastConfigIndex,
               pSyncNode->changing);
    } else {
      snprintf(s, len, "%s", str);
    }
M
Minghao Li 已提交
1428 1429 1430 1431 1432
    sError("%s", s);
    taosMemoryFree(s);
  }
}

M
Minghao Li 已提交
1433 1434 1435
char* syncNode2SimpleStr(const SSyncNode* pSyncNode) {
  int   len = 256;
  char* s = (char*)taosMemoryMalloc(len);
M
Minghao Li 已提交
1436 1437 1438 1439 1440 1441 1442 1443

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

M
Minghao Li 已提交
1444
  snprintf(s, len,
M
Minghao Li 已提交
1445 1446 1447 1448 1449 1450 1451 1452 1453 1454 1455 1456 1457 1458 1459 1460 1461 1462 1463
           "vgId:%d, sync %s, term:%lu, commit:%ld, beginlog:%ld, lastlog:%ld, lastsnapshot:%ld, standby:%d, "
           "replica-num:%d, "
           "lconfig:%ld, changing:%d, restore:%d",
           pSyncNode->vgId, syncUtilState2String(pSyncNode->state), pSyncNode->pRaftStore->currentTerm,
           pSyncNode->commitIndex, logBeginIndex, logLastIndex, snapshot.lastApplyIndex, pSyncNode->pRaftCfg->isStandBy,
           pSyncNode->replicaNum, pSyncNode->pRaftCfg->lastConfigIndex, pSyncNode->changing, pSyncNode->restoreFinish);

  /*
    snprintf(s, len,
             "syncNode: vgId:%d, term:%lu, commit:%ld, state:%d %s, standby:%d, "
             "lc:%lu, "
             "lc-user:%lu, "
             "ems:%d, replica-num:%d, restore:%d, changing:%d",
             pSyncNode->vgId, pSyncNode->pRaftStore->currentTerm, pSyncNode->commitIndex, pSyncNode->state,
             syncUtilState2String(pSyncNode->state), pSyncNode->pRaftCfg->isStandBy, pSyncNode->electTimerLogicClock,
             pSyncNode->electTimerLogicClockUser, pSyncNode->electTimerMS, pSyncNode->replicaNum,
             pSyncNode->restoreFinish, pSyncNode->changing);
  */

M
Minghao Li 已提交
1464 1465 1466
  return s;
}

1467 1468 1469 1470 1471 1472 1473 1474 1475 1476 1477 1478 1479 1480 1481 1482 1483 1484 1485 1486 1487 1488 1489 1490 1491 1492 1493
bool syncNodeInConfig(SSyncNode* pSyncNode, const SSyncCfg* config) {
  bool b1 = false;
  bool b2 = false;

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

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

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

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

M
Minghao Li 已提交
1494
void syncNodeDoConfigChange(SSyncNode* pSyncNode, SSyncCfg* pNewConfig, SyncIndex lastConfigChangeIndex) {
1495
  SSyncCfg oldConfig = pSyncNode->pRaftCfg->cfg;
1496
  pSyncNode->pRaftCfg->cfg = *pNewConfig;
1497 1498
  pSyncNode->pRaftCfg->lastConfigIndex = lastConfigChangeIndex;

M
Minghao Li 已提交
1499 1500
  bool IamInOld = syncNodeInConfig(pSyncNode, &oldConfig);
  bool IamInNew = syncNodeInConfig(pSyncNode, pNewConfig);
M
Minghao Li 已提交
1501

M
Minghao Li 已提交
1502 1503
  bool isDrop = false;
  bool isAdd = false;
M
Minghao Li 已提交
1504

M
Minghao Li 已提交
1505 1506 1507 1508
  if (IamInOld && !IamInNew) {
    isDrop = true;
  } else {
    isDrop = false;
1509
  }
1510

M
Minghao Li 已提交
1511 1512 1513 1514 1515
  if (!IamInOld && IamInNew) {
    isAdd = true;
  } else {
    isAdd = false;
  }
M
Minghao Li 已提交
1516

M
Minghao Li 已提交
1517 1518 1519 1520 1521 1522 1523 1524 1525 1526 1527
  // log begin config change
  do {
    char  eventLog[256];
    char* pOldCfgStr = syncCfg2SimpleStr(&oldConfig);
    char* pNewCfgStr = syncCfg2SimpleStr(pNewConfig);
    snprintf(eventLog, sizeof(eventLog), "begin do config change, from %s to %s", pOldCfgStr, pNewCfgStr);
    syncNodeEventLog(pSyncNode, eventLog);
    taosMemoryFree(pOldCfgStr);
    taosMemoryFree(pNewCfgStr);
  } while (0);

M
Minghao Li 已提交
1528 1529
  if (IamInNew) {
    pSyncNode->pRaftCfg->isStandBy = 0;  // change isStandBy to normal
M
Minghao Li 已提交
1530
  }
M
Minghao Li 已提交
1531 1532
  if (isDrop) {
    pSyncNode->pRaftCfg->isStandBy = 1;  // set standby
M
Minghao Li 已提交
1533 1534
  }

M
Minghao Li 已提交
1535
  // add last config index
M
Minghao Li 已提交
1536
  raftCfgAddConfigIndex(pSyncNode->pRaftCfg, lastConfigChangeIndex);
M
Minghao Li 已提交
1537

M
Minghao Li 已提交
1538 1539 1540 1541 1542 1543 1544 1545 1546 1547 1548
  if (IamInNew) {
    //-----------------------------------------
    int32_t ret = 0;

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

M
Minghao Li 已提交
1550 1551 1552 1553
      char* eventLog = snapshotSender2SimpleStr(oldSenders[i], "snapshot sender save old");
      syncNodeEventLog(pSyncNode, eventLog);
      taosMemoryFree(eventLog);
    }
1554

M
Minghao Li 已提交
1555 1556 1557 1558 1559 1560 1561 1562 1563 1564 1565 1566 1567 1568 1569 1570
    // init internal
    pSyncNode->myNodeInfo = pSyncNode->pRaftCfg->cfg.nodeInfo[pSyncNode->pRaftCfg->cfg.myIndex];
    syncUtilnodeInfo2raftId(&pSyncNode->myNodeInfo, pSyncNode->vgId, &pSyncNode->myRaftId);

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

M
Minghao Li 已提交
1572 1573 1574 1575 1576
    // init replicaNum, replicasId
    pSyncNode->replicaNum = pSyncNode->pRaftCfg->cfg.replicaNum;
    for (int i = 0; i < pSyncNode->pRaftCfg->cfg.replicaNum; ++i) {
      syncUtilnodeInfo2raftId(&pSyncNode->pRaftCfg->cfg.nodeInfo[i], pSyncNode->vgId, &pSyncNode->replicasId[i]);
    }
1577

M
Minghao Li 已提交
1578 1579 1580 1581
    syncIndexMgrUpdate(pSyncNode->pNextIndex, pSyncNode);
    syncIndexMgrUpdate(pSyncNode->pMatchIndex, pSyncNode);
    voteGrantedUpdate(pSyncNode->pVotesGranted, pSyncNode);
    votesRespondUpdate(pSyncNode->pVotesRespond, pSyncNode);
M
Minghao Li 已提交
1582

M
Minghao Li 已提交
1583
    pSyncNode->quorum = syncUtilQuorum(pSyncNode->pRaftCfg->cfg.replicaNum);
M
Minghao Li 已提交
1584

M
Minghao Li 已提交
1585
    // reset snapshot senders
1586

M
Minghao Li 已提交
1587 1588 1589 1590
    // clear new
    for (int i = 0; i < TSDB_MAX_REPLICA; ++i) {
      (pSyncNode->senders)[i] = NULL;
    }
M
Minghao Li 已提交
1591

M
Minghao Li 已提交
1592 1593 1594 1595 1596 1597 1598 1599 1600 1601 1602 1603 1604 1605 1606 1607 1608 1609 1610 1611 1612 1613 1614 1615 1616 1617 1618 1619 1620 1621 1622 1623 1624
    // reset new
    for (int i = 0; i < pSyncNode->replicaNum; ++i) {
      // reset sender
      bool reset = false;
      for (int j = 0; j < TSDB_MAX_REPLICA; ++j) {
        if (syncUtilSameId(&(pSyncNode->replicasId)[i], &oldReplicasId[j])) {
          char     host[128];
          uint16_t port;
          syncUtilU642Addr((pSyncNode->replicasId)[i].addr, host, sizeof(host), &port);

          do {
            char eventLog[256];
            snprintf(eventLog, sizeof(eventLog), "snapshot sender reset for %lu, newIndex:%d, %s:%d, %p",
                     (pSyncNode->replicasId)[i].addr, i, host, port, oldSenders[j]);
            syncNodeEventLog(pSyncNode, eventLog);
          } while (0);

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

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

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

M
Minghao Li 已提交
1628 1629 1630 1631
    // create new
    for (int i = 0; i < TSDB_MAX_REPLICA; ++i) {
      if ((pSyncNode->senders)[i] == NULL) {
        (pSyncNode->senders)[i] = snapshotSenderCreate(pSyncNode, i);
M
Minghao Li 已提交
1632

M
Minghao Li 已提交
1633 1634 1635 1636
        char* eventLog = snapshotSender2SimpleStr((pSyncNode->senders)[i], "snapshot sender create new");
        syncNodeEventLog(pSyncNode, eventLog);
        taosMemoryFree(eventLog);
      }
1637 1638
    }

M
Minghao Li 已提交
1639 1640 1641 1642
    // free old
    for (int i = 0; i < TSDB_MAX_REPLICA; ++i) {
      if (oldSenders[i] != NULL) {
        snapshotSenderDestroy(oldSenders[i]);
M
Minghao Li 已提交
1643

M
Minghao Li 已提交
1644 1645 1646 1647 1648
        do {
          char eventLog[128];
          snprintf(eventLog, sizeof(eventLog), "snapshot sender delete old %p replica-index:%d", oldSenders[i], i);
          syncNodeEventLog(pSyncNode, eventLog);
        } while (0);
M
Minghao Li 已提交
1649

M
Minghao Li 已提交
1650 1651
        oldSenders[i] = NULL;
      }
1652 1653
    }

1654
    // persist cfg
M
Minghao Li 已提交
1655
    raftCfgPersist(pSyncNode->pRaftCfg);
1656

M
Minghao Li 已提交
1657 1658 1659 1660 1661 1662 1663
    char  tmpbuf[512];
    char* oldStr = syncCfg2SimpleStr(&oldConfig);
    char* newStr = syncCfg2SimpleStr(pNewConfig);
    snprintf(tmpbuf, sizeof(tmpbuf), "config change from %d to %d, index:%ld, %s  -->  %s", oldConfig.replicaNum,
             pNewConfig->replicaNum, lastConfigChangeIndex, oldStr, newStr);
    taosMemoryFree(oldStr);
    taosMemoryFree(newStr);
M
Minghao Li 已提交
1664

M
Minghao Li 已提交
1665 1666 1667
    // change isStandBy to normal (election timeout)
    if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
      syncNodeBecomeLeader(pSyncNode, tmpbuf);
1668 1669 1670 1671 1672 1673

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

M
Minghao Li 已提交
1674 1675 1676 1677
    } else {
      syncNodeBecomeFollower(pSyncNode, tmpbuf);
    }
  } else {
1678
    // persist cfg
M
Minghao Li 已提交
1679
    raftCfgPersist(pSyncNode->pRaftCfg);
1680

M
Minghao Li 已提交
1681 1682 1683 1684 1685 1686 1687 1688
    char  tmpbuf[512];
    char* oldStr = syncCfg2SimpleStr(&oldConfig);
    char* newStr = syncCfg2SimpleStr(pNewConfig);
    snprintf(tmpbuf, sizeof(tmpbuf), "do not config change from %d to %d, index:%ld, %s  -->  %s", oldConfig.replicaNum,
             pNewConfig->replicaNum, lastConfigChangeIndex, oldStr, newStr);
    taosMemoryFree(oldStr);
    taosMemoryFree(newStr);
    syncNodeEventLog(pSyncNode, tmpbuf);
1689
  }
1690

M
Minghao Li 已提交
1691
_END:
M
Minghao Li 已提交
1692 1693 1694 1695 1696 1697 1698 1699 1700 1701 1702

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

M
Minghao Li 已提交
1706 1707 1708 1709 1710 1711 1712 1713 1714 1715 1716
SSyncNode* syncNodeAcquire(int64_t rid) {
  SSyncNode* pNode = taosAcquireRef(tsNodeRefId, rid);
  if (pNode == NULL) {
    sTrace("failed to acquire node from refId:%" PRId64, rid);
  }

  return pNode;
}

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

M
Minghao Li 已提交
1717 1718 1719 1720
// raft state change --------------
void syncNodeUpdateTerm(SSyncNode* pSyncNode, SyncTerm term) {
  if (term > pSyncNode->pRaftStore->currentTerm) {
    raftStoreSetTerm(pSyncNode->pRaftStore, term);
1721 1722 1723
    char tmpBuf[64];
    snprintf(tmpBuf, sizeof(tmpBuf), "update term to %lu", term);
    syncNodeBecomeFollower(pSyncNode, tmpBuf);
M
Minghao Li 已提交
1724 1725 1726 1727
    raftStoreClearVote(pSyncNode->pRaftStore);
  }
}

1728
void syncNodeBecomeFollower(SSyncNode* pSyncNode, const char* debugStr) {
M
Minghao Li 已提交
1729
  // maybe clear leader cache
M
Minghao Li 已提交
1730 1731 1732 1733
  if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
    pSyncNode->leaderCache = EMPTY_RAFT_ID;
  }

M
Minghao Li 已提交
1734
  // state change
M
Minghao Li 已提交
1735 1736 1737
  pSyncNode->state = TAOS_SYNC_STATE_FOLLOWER;
  syncNodeStopHeartbeatTimer(pSyncNode);

M
Minghao Li 已提交
1738 1739
  // reset elect timer
  syncNodeResetElectTimer(pSyncNode);
M
Minghao Li 已提交
1740 1741 1742 1743 1744 1745 1746 1747 1748 1749 1750 1751 1752 1753 1754

  // trace log
  do {
    int32_t debugStrLen = strlen(debugStr);
    if (debugStrLen < 256) {
      char eventLog[256 + 64];
      snprintf(eventLog, sizeof(eventLog), "become follower %s", debugStr);
      syncNodeEventLog(pSyncNode, eventLog);
    } else {
      char* eventLog = taosMemoryMalloc(debugStrLen + 64);
      snprintf(eventLog, debugStrLen, "become follower %s", debugStr);
      syncNodeEventLog(pSyncNode, eventLog);
      taosMemoryFree(eventLog);
    }
  } while (0);
M
Minghao Li 已提交
1755 1756 1757 1758 1759 1760 1761 1762 1763 1764 1765 1766 1767 1768 1769 1770 1771 1772 1773 1774
}

// 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>>
//
1775
void syncNodeBecomeLeader(SSyncNode* pSyncNode, const char* debugStr) {
1776 1777 1778
  // reset restoreFinish
  pSyncNode->restoreFinish = false;

M
Minghao Li 已提交
1779
  // state change
M
Minghao Li 已提交
1780
  pSyncNode->state = TAOS_SYNC_STATE_LEADER;
M
Minghao Li 已提交
1781 1782

  // set leader cache
M
Minghao Li 已提交
1783 1784 1785
  pSyncNode->leaderCache = pSyncNode->myRaftId;

  for (int i = 0; i < pSyncNode->pNextIndex->replicaNum; ++i) {
M
Minghao Li 已提交
1786 1787
    // maybe overwrite myself, no harm
    // just do it!
1788 1789 1790 1791 1792 1793 1794 1795 1796

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

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

  for (int i = 0; i < pSyncNode->pMatchIndex->replicaNum; ++i) {
M
Minghao Li 已提交
1800 1801
    // maybe overwrite myself, no harm
    // just do it!
M
Minghao Li 已提交
1802 1803 1804
    pSyncNode->pMatchIndex->index[i] = SYNC_INDEX_INVALID;
  }

1805 1806
  // update sender private term
  SSyncSnapshotSender* pMySender = syncNodeGetSnapshotSender(pSyncNode, &(pSyncNode->myRaftId));
1807 1808 1809 1810 1811
  if (pMySender != NULL) {
    for (int i = 0; i < pSyncNode->pMatchIndex->replicaNum; ++i) {
      if ((pSyncNode->senders)[i]->privateTerm > pMySender->privateTerm) {
        pMySender->privateTerm = (pSyncNode->senders)[i]->privateTerm;
      }
1812
    }
1813
    (pMySender->privateTerm) += 100;
1814 1815
  }

M
Minghao Li 已提交
1816
  // stop elect timer
M
Minghao Li 已提交
1817
  syncNodeStopElectTimer(pSyncNode);
M
Minghao Li 已提交
1818

M
Minghao Li 已提交
1819 1820
  // start heartbeat timer
  syncNodeStartHeartbeatTimer(pSyncNode);
M
Minghao Li 已提交
1821 1822 1823 1824 1825 1826 1827 1828 1829 1830 1831 1832 1833 1834 1835

  // trace log
  do {
    int32_t debugStrLen = strlen(debugStr);
    if (debugStrLen < 256) {
      char eventLog[256 + 64];
      snprintf(eventLog, sizeof(eventLog), "become leader %s", debugStr);
      syncNodeEventLog(pSyncNode, eventLog);
    } else {
      char* eventLog = taosMemoryMalloc(debugStrLen + 64);
      snprintf(eventLog, debugStrLen, "become leader %s", debugStr);
      syncNodeEventLog(pSyncNode, eventLog);
      taosMemoryFree(eventLog);
    }
  } while (0);
M
Minghao Li 已提交
1836 1837 1838
}

void syncNodeCandidate2Leader(SSyncNode* pSyncNode) {
M
Minghao Li 已提交
1839 1840
  ASSERT(pSyncNode->state == TAOS_SYNC_STATE_CANDIDATE);
  ASSERT(voteGrantedMajority(pSyncNode->pVotesGranted));
1841
  syncNodeBecomeLeader(pSyncNode, "candidate to leader");
M
Minghao Li 已提交
1842

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

M
Minghao Li 已提交
1845
  // Raft 3.6.2 Committing entries from previous terms
1846 1847 1848
  syncNodeReplicate(pSyncNode);
  syncNodeAppendNoop(pSyncNode);
  syncMaybeAdvanceCommitIndex(pSyncNode);
M
Minghao Li 已提交
1849 1850 1851
}

void syncNodeFollower2Candidate(SSyncNode* pSyncNode) {
M
Minghao Li 已提交
1852
  ASSERT(pSyncNode->state == TAOS_SYNC_STATE_FOLLOWER);
M
Minghao Li 已提交
1853
  pSyncNode->state = TAOS_SYNC_STATE_CANDIDATE;
M
Minghao Li 已提交
1854 1855

  syncNodeLog2("==state change syncNodeFollower2Candidate==", pSyncNode);
M
Minghao Li 已提交
1856 1857 1858
}

void syncNodeLeader2Follower(SSyncNode* pSyncNode) {
M
Minghao Li 已提交
1859
  ASSERT(pSyncNode->state == TAOS_SYNC_STATE_LEADER);
1860
  syncNodeBecomeFollower(pSyncNode, "leader to follower");
M
Minghao Li 已提交
1861 1862

  syncNodeLog2("==state change syncNodeLeader2Follower==", pSyncNode);
M
Minghao Li 已提交
1863 1864 1865
}

void syncNodeCandidate2Follower(SSyncNode* pSyncNode) {
M
Minghao Li 已提交
1866
  ASSERT(pSyncNode->state == TAOS_SYNC_STATE_CANDIDATE);
1867
  syncNodeBecomeFollower(pSyncNode, "candidate to follower");
M
Minghao Li 已提交
1868 1869

  syncNodeLog2("==state change syncNodeCandidate2Follower==", pSyncNode);
M
Minghao Li 已提交
1870 1871 1872
}

// raft vote --------------
M
Minghao Li 已提交
1873 1874 1875

// just called by syncNodeVoteForSelf
// need assert
M
Minghao Li 已提交
1876
void syncNodeVoteForTerm(SSyncNode* pSyncNode, SyncTerm term, SRaftId* pRaftId) {
M
Minghao Li 已提交
1877 1878
  ASSERT(term == pSyncNode->pRaftStore->currentTerm);
  ASSERT(!raftStoreHasVoted(pSyncNode->pRaftStore));
M
Minghao Li 已提交
1879 1880 1881 1882

  raftStoreVote(pSyncNode->pRaftStore, pRaftId);
}

M
Minghao Li 已提交
1883
// simulate get vote from outside
M
Minghao Li 已提交
1884 1885 1886
void syncNodeVoteForSelf(SSyncNode* pSyncNode) {
  syncNodeVoteForTerm(pSyncNode, pSyncNode->pRaftStore->currentTerm, &(pSyncNode->myRaftId));

M
Minghao Li 已提交
1887
  SyncRequestVoteReply* pMsg = syncRequestVoteReplyBuild(pSyncNode->vgId);
M
Minghao Li 已提交
1888 1889 1890 1891 1892 1893 1894 1895 1896 1897
  pMsg->srcId = pSyncNode->myRaftId;
  pMsg->destId = pSyncNode->myRaftId;
  pMsg->term = pSyncNode->pRaftStore->currentTerm;
  pMsg->voteGranted = true;

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

M
Minghao Li 已提交
1898
// snapshot --------------
M
Minghao Li 已提交
1899 1900
bool syncNodeHasSnapshot(SSyncNode* pSyncNode) {
  bool      ret = false;
1901
  SSnapshot snapshot = {.data = NULL, .lastApplyIndex = -1, .lastApplyTerm = 0, .lastConfigIndex = -1};
1902 1903
  if (pSyncNode->pFsm->FpGetSnapshotInfo != NULL) {
    pSyncNode->pFsm->FpGetSnapshotInfo(pSyncNode->pFsm, &snapshot);
M
Minghao Li 已提交
1904 1905 1906 1907 1908 1909 1910
    if (snapshot.lastApplyIndex >= SYNC_INDEX_BEGIN) {
      ret = true;
    }
  }
  return ret;
}

1911
#if 0
1912
bool syncNodeIsIndexInSnapshot(SSyncNode* pSyncNode, SyncIndex index) {
M
Minghao Li 已提交
1913
  ASSERT(syncNodeHasSnapshot(pSyncNode));
1914
  ASSERT(pSyncNode->pFsm->FpGetSnapshotInfo != NULL);
M
Minghao Li 已提交
1915 1916
  ASSERT(index >= SYNC_INDEX_BEGIN);

1917
  SSnapshot snapshot;
1918
  pSyncNode->pFsm->FpGetSnapshotInfo(pSyncNode->pFsm, &snapshot);
M
Minghao Li 已提交
1919
  bool b = (index <= snapshot.lastApplyIndex);
1920 1921
  return b;
}
1922
#endif
1923

M
Minghao Li 已提交
1924 1925
SyncIndex syncNodeGetLastIndex(SSyncNode* pSyncNode) {
  SSnapshot snapshot = {.data = NULL, .lastApplyIndex = -1, .lastApplyTerm = 0};
1926 1927
  if (pSyncNode->pFsm->FpGetSnapshotInfo != NULL) {
    pSyncNode->pFsm->FpGetSnapshotInfo(pSyncNode->pFsm, &snapshot);
M
Minghao Li 已提交
1928 1929 1930 1931 1932 1933 1934 1935 1936
  }
  SyncIndex logLastIndex = pSyncNode->pLogStore->syncLogLastIndex(pSyncNode->pLogStore);

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

SyncTerm syncNodeGetLastTerm(SSyncNode* pSyncNode) {
  SyncTerm lastTerm = 0;
M
Minghao Li 已提交
1937 1938
  if (syncNodeHasSnapshot(pSyncNode)) {
    // has snapshot
1939
    SSnapshot snapshot = {.data = NULL, .lastApplyIndex = -1, .lastApplyTerm = 0, .lastConfigIndex = -1};
1940 1941
    if (pSyncNode->pFsm->FpGetSnapshotInfo != NULL) {
      pSyncNode->pFsm->FpGetSnapshotInfo(pSyncNode->pFsm, &snapshot);
M
Minghao Li 已提交
1942 1943
    }

M
Minghao Li 已提交
1944 1945 1946
    SyncIndex logLastIndex = pSyncNode->pLogStore->syncLogLastIndex(pSyncNode->pLogStore);
    if (logLastIndex > snapshot.lastApplyIndex) {
      lastTerm = pSyncNode->pLogStore->syncLogLastTerm(pSyncNode->pLogStore);
M
Minghao Li 已提交
1947 1948 1949 1950
    } else {
      lastTerm = snapshot.lastApplyTerm;
    }

M
Minghao Li 已提交
1951
  } else {
M
Minghao Li 已提交
1952 1953
    // no snapshot
    lastTerm = pSyncNode->pLogStore->syncLogLastTerm(pSyncNode->pLogStore);
1954
  }
M
Minghao Li 已提交
1955

M
Minghao Li 已提交
1956 1957 1958 1959 1960 1961 1962
  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);
1963 1964
  return 0;
}
M
Minghao Li 已提交
1965

M
Minghao Li 已提交
1966 1967 1968 1969 1970
SyncIndex syncNodeSyncStartIndex(SSyncNode* pSyncNode) {
  SyncIndex syncStartIndex = syncNodeGetLastIndex(pSyncNode) + 1;
  return syncStartIndex;
}

1971 1972 1973 1974 1975 1976 1977 1978 1979 1980
SyncIndex syncNodeGetPreIndex(SSyncNode* pSyncNode, SyncIndex index) {
  SyncIndex preIndex = index - 1;
  if (preIndex < SYNC_INDEX_INVALID) {
    preIndex = SYNC_INDEX_INVALID;
  }

  return preIndex;
}

/*
M
Minghao Li 已提交
1981
SyncIndex syncNodeGetPreIndex(SSyncNode* pSyncNode, SyncIndex index) {
1982
  ASSERT(index >= SYNC_INDEX_BEGIN);
M
Minghao Li 已提交
1983

1984
  SyncIndex syncStartIndex = syncNodeSyncStartIndex(pSyncNode);
M
Minghao Li 已提交
1985 1986 1987 1988
  if (index > syncStartIndex) {
    syncNodeLog3("syncNodeGetPreIndex", pSyncNode);
    ASSERT(0);
  }
M
Minghao Li 已提交
1989

M
Minghao Li 已提交
1990
  SyncIndex preIndex = index - 1;
M
Minghao Li 已提交
1991 1992
  return preIndex;
}
1993 1994 1995 1996 1997 1998 1999 2000 2001 2002 2003 2004 2005 2006 2007 2008 2009 2010 2011 2012 2013
*/

SyncTerm syncNodeGetPreTerm(SSyncNode* pSyncNode, SyncIndex index) {
  if (index < SYNC_INDEX_BEGIN) {
    return SYNC_TERM_INVALID;
  }

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

  SyncTerm        preTerm = 0;
  SyncIndex       preIndex = index - 1;
  SSyncRaftEntry* pPreEntry = NULL;
  int32_t         code = pSyncNode->pLogStore->syncLogGetEntry(pSyncNode->pLogStore, preIndex, &pPreEntry);
  if (code == 0) {
    ASSERT(pPreEntry != NULL);
    preTerm = pPreEntry->term;
    taosMemoryFree(pPreEntry);
    return preTerm;
  } else {
2014 2015 2016 2017 2018
    SSnapshot snapshot = {.data = NULL, .lastApplyIndex = -1, .lastApplyTerm = 0, .lastConfigIndex = -1};
    if (pSyncNode->pFsm->FpGetSnapshotInfo != NULL) {
      pSyncNode->pFsm->FpGetSnapshotInfo(pSyncNode->pFsm, &snapshot);
      if (snapshot.lastApplyIndex == preIndex) {
        return snapshot.lastApplyTerm;
2019 2020 2021 2022
      }
    }
  }

2023 2024 2025 2026 2027 2028
  do {
    char logBuf[128];
    snprintf(logBuf, sizeof(logBuf), "sync node get pre term error, index:%ld", index);
    syncNodeErrorLog(pSyncNode, logBuf);
  } while (0);

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

2032
#if 0
M
Minghao Li 已提交
2033 2034
SyncTerm syncNodeGetPreTerm(SSyncNode* pSyncNode, SyncIndex index) {
  ASSERT(index >= SYNC_INDEX_BEGIN);
2035

M
Minghao Li 已提交
2036
  SyncIndex syncStartIndex = syncNodeSyncStartIndex(pSyncNode);
2037 2038 2039 2040
  if (index > syncStartIndex) {
    syncNodeLog3("syncNodeGetPreTerm", pSyncNode);
    ASSERT(0);
  }
M
Minghao Li 已提交
2041

2042 2043 2044 2045 2046 2047 2048 2049 2050 2051 2052 2053 2054 2055 2056 2057 2058 2059 2060 2061 2062 2063 2064 2065 2066 2067 2068 2069
  if (index == SYNC_INDEX_BEGIN) {
    return 0;
  }

  SyncTerm        preTerm = 0;
  SyncIndex       preIndex = index - 1;
  SSyncRaftEntry* pPreEntry = NULL;
  int32_t         code = pSyncNode->pLogStore->syncLogGetEntry(pSyncNode->pLogStore, preIndex, &pPreEntry);
  if (code == 0) {
    ASSERT(pPreEntry != NULL);
    preTerm = pPreEntry->term;
    taosMemoryFree(pPreEntry);
    return preTerm;
  } else {
    if (terrno == TSDB_CODE_WAL_LOG_NOT_EXIST) {
      SSnapshot snapshot = {.data = NULL, .lastApplyIndex = -1, .lastApplyTerm = 0, .lastConfigIndex = -1};
      if (pSyncNode->pFsm->FpGetSnapshotInfo != NULL) {
        pSyncNode->pFsm->FpGetSnapshotInfo(pSyncNode->pFsm, &snapshot);
        if (snapshot.lastApplyIndex == preIndex) {
          return snapshot.lastApplyTerm;
        }
      }
    }
  }

  ASSERT(0);
  return -1;
}
2070
#endif
2071 2072 2073 2074 2075 2076

#if 0
SyncTerm syncNodeGetPreTerm(SSyncNode* pSyncNode, SyncIndex index) {
  ASSERT(index >= SYNC_INDEX_BEGIN);

  SyncIndex syncStartIndex = syncNodeSyncStartIndex(pSyncNode);
M
Minghao Li 已提交
2077 2078 2079 2080
  if (index > syncStartIndex) {
    syncNodeLog3("syncNodeGetPreTerm", pSyncNode);
    ASSERT(0);
  }
M
Minghao Li 已提交
2081

M
Minghao Li 已提交
2082 2083
  if (index == SYNC_INDEX_BEGIN) {
    return 0;
M
Minghao Li 已提交
2084
  }
M
Minghao Li 已提交
2085

M
Minghao Li 已提交
2086 2087 2088
  SyncTerm preTerm = 0;
  if (syncNodeHasSnapshot(pSyncNode)) {
    // has snapshot
2089
    SSnapshot snapshot = {.data = NULL, .lastApplyIndex = -1, .lastApplyTerm = 0, .lastConfigIndex = -1};
2090 2091
    if (pSyncNode->pFsm->FpGetSnapshotInfo != NULL) {
      pSyncNode->pFsm->FpGetSnapshotInfo(pSyncNode->pFsm, &snapshot);
M
Minghao Li 已提交
2092 2093 2094 2095 2096 2097 2098 2099 2100
    }

    if (index > snapshot.lastApplyIndex + 1) {
      // should be log preTerm
      SSyncRaftEntry* pPreEntry = NULL;
      int32_t         code = pSyncNode->pLogStore->syncLogGetEntry(pSyncNode->pLogStore, index - 1, &pPreEntry);
      ASSERT(code == 0);
      ASSERT(pPreEntry != NULL);

M
Minghao Li 已提交
2101 2102
      preTerm = pPreEntry->term;
      taosMemoryFree(pPreEntry);
M
Minghao Li 已提交
2103 2104 2105

    } else if (index == snapshot.lastApplyIndex + 1) {
      preTerm = snapshot.lastApplyTerm;
M
Minghao Li 已提交
2106

M
Minghao Li 已提交
2107
    } else {
2108
      // maybe snapshot change
M
Minghao Li 已提交
2109 2110 2111 2112 2113 2114 2115 2116 2117 2118
      sError("sync get pre term, bad scene. index:%ld", index);
      logStoreLog2("sync get pre term, bad scene", pSyncNode->pLogStore);

      SSyncRaftEntry* pPreEntry = NULL;
      int32_t         code = pSyncNode->pLogStore->syncLogGetEntry(pSyncNode->pLogStore, index - 1, &pPreEntry);
      ASSERT(code == 0);
      ASSERT(pPreEntry != NULL);

      preTerm = pPreEntry->term;
      taosMemoryFree(pPreEntry);
2119
    }
M
Minghao Li 已提交
2120 2121 2122 2123 2124 2125 2126 2127 2128 2129 2130 2131

  } else {
    // no snapshot
    ASSERT(index > SYNC_INDEX_BEGIN);

    SSyncRaftEntry* pPreEntry = NULL;
    int32_t         code = pSyncNode->pLogStore->syncLogGetEntry(pSyncNode->pLogStore, index - 1, &pPreEntry);
    ASSERT(code == 0);
    ASSERT(pPreEntry != NULL);

    preTerm = pPreEntry->term;
    taosMemoryFree(pPreEntry);
2132
  }
M
Minghao Li 已提交
2133

M
Minghao Li 已提交
2134 2135
  return preTerm;
}
2136
#endif
2137

M
Minghao Li 已提交
2138 2139 2140
// 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 已提交
2141
  *pPreTerm = syncNodeGetPreTerm(pSyncNode, index);
M
Minghao Li 已提交
2142 2143 2144
  return 0;
}

M
Minghao Li 已提交
2145 2146 2147 2148 2149
// for debug --------------
void syncNodePrint(SSyncNode* pObj) {
  char* serialized = syncNode2Str(pObj);
  printf("syncNodePrint | len:%lu | %s \n", strlen(serialized), serialized);
  fflush(NULL);
wafwerar's avatar
wafwerar 已提交
2150
  taosMemoryFree(serialized);
M
Minghao Li 已提交
2151 2152 2153 2154 2155 2156
}

void syncNodePrint2(char* s, SSyncNode* pObj) {
  char* serialized = syncNode2Str(pObj);
  printf("syncNodePrint2 | len:%lu | %s | %s \n", strlen(serialized), s, serialized);
  fflush(NULL);
wafwerar's avatar
wafwerar 已提交
2157
  taosMemoryFree(serialized);
M
Minghao Li 已提交
2158 2159 2160 2161
}

void syncNodeLog(SSyncNode* pObj) {
  char* serialized = syncNode2Str(pObj);
M
Minghao Li 已提交
2162
  sTraceLong("syncNodeLog | len:%lu | %s", strlen(serialized), serialized);
wafwerar's avatar
wafwerar 已提交
2163
  taosMemoryFree(serialized);
M
Minghao Li 已提交
2164 2165 2166
}

void syncNodeLog2(char* s, SSyncNode* pObj) {
2167 2168 2169 2170 2171
  if (gRaftDetailLog) {
    char* serialized = syncNode2Str(pObj);
    sTraceLong("syncNodeLog2 | len:%lu | %s | %s", strlen(serialized), s, serialized);
    taosMemoryFree(serialized);
  }
M
Minghao Li 已提交
2172 2173
}

M
Minghao Li 已提交
2174 2175 2176 2177 2178 2179
void syncNodeLog3(char* s, SSyncNode* pObj) {
  char* serialized = syncNode2Str(pObj);
  sTraceLong("syncNodeLog3 | len:%lu | %s | %s", strlen(serialized), s, serialized);
  taosMemoryFree(serialized);
}

M
Minghao Li 已提交
2180
// ------ local funciton ---------
M
Minghao Li 已提交
2181
// enqueue message ----
M
Minghao Li 已提交
2182 2183
static void syncNodeEqPingTimer(void* param, void* tmrId) {
  SSyncNode* pSyncNode = (SSyncNode*)param;
M
Minghao Li 已提交
2184
  if (atomic_load_64(&pSyncNode->pingTimerLogicClockUser) <= atomic_load_64(&pSyncNode->pingTimerLogicClock)) {
M
Minghao Li 已提交
2185
    SyncTimeout* pSyncMsg = syncTimeoutBuild2(SYNC_TIMEOUT_PING, atomic_load_64(&pSyncNode->pingTimerLogicClock),
M
Minghao Li 已提交
2186
                                              pSyncNode->pingTimerMS, pSyncNode->vgId, pSyncNode);
M
Minghao Li 已提交
2187 2188
    SRpcMsg      rpcMsg;
    syncTimeout2RpcMsg(pSyncMsg, &rpcMsg);
M
Minghao Li 已提交
2189
    syncRpcMsgLog2((char*)"==syncNodeEqPingTimer==", &rpcMsg);
M
Minghao Li 已提交
2190
    if (pSyncNode->FpEqMsg != NULL) {
2191 2192
      int32_t code = pSyncNode->FpEqMsg(pSyncNode->msgcb, &rpcMsg);
      if (code != 0) {
S
Shengliang Guan 已提交
2193
        sError("vgId:%d, sync enqueue ping msg error, code:%d", pSyncNode->vgId, code);
2194 2195 2196 2197
        rpcFreeCont(rpcMsg.pCont);
        syncTimeoutDestroy(pSyncMsg);
        return;
      }
M
Minghao Li 已提交
2198 2199 2200
    } else {
      sTrace("syncNodeEqPingTimer pSyncNode->FpEqMsg is NULL");
    }
M
Minghao Li 已提交
2201 2202
    syncTimeoutDestroy(pSyncMsg);

2203 2204 2205 2206 2207 2208 2209
    if (syncEnvIsStart()) {
      taosTmrReset(syncNodeEqPingTimer, pSyncNode->pingTimerMS, pSyncNode, gSyncEnv->pTimerManager,
                   &pSyncNode->pPingTimer);
    } else {
      sError("sync env is stop, syncNodeEqPingTimer");
    }

M
Minghao Li 已提交
2210
  } else {
2211
    sTrace("==syncNodeEqPingTimer== pingTimerLogicClock:%" PRIu64 ", pingTimerLogicClockUser:%" PRIu64 "",
M
Minghao Li 已提交
2212
           pSyncNode->pingTimerLogicClock, pSyncNode->pingTimerLogicClockUser);
M
Minghao Li 已提交
2213 2214 2215 2216 2217 2218 2219
  }
}

static void syncNodeEqElectTimer(void* param, void* tmrId) {
  SSyncNode* pSyncNode = (SSyncNode*)param;
  if (atomic_load_64(&pSyncNode->electTimerLogicClockUser) <= atomic_load_64(&pSyncNode->electTimerLogicClock)) {
    SyncTimeout* pSyncMsg = syncTimeoutBuild2(SYNC_TIMEOUT_ELECTION, atomic_load_64(&pSyncNode->electTimerLogicClock),
M
Minghao Li 已提交
2220
                                              pSyncNode->electTimerMS, pSyncNode->vgId, pSyncNode);
M
Minghao Li 已提交
2221
    SRpcMsg      rpcMsg;
M
Minghao Li 已提交
2222
    syncTimeout2RpcMsg(pSyncMsg, &rpcMsg);
M
Minghao Li 已提交
2223
    syncRpcMsgLog2((char*)"==syncNodeEqElectTimer==", &rpcMsg);
M
Minghao Li 已提交
2224
    if (pSyncNode->FpEqMsg != NULL) {
2225 2226
      int32_t code = pSyncNode->FpEqMsg(pSyncNode->msgcb, &rpcMsg);
      if (code != 0) {
S
Shengliang Guan 已提交
2227
        sError("vgId:%d, sync enqueue elect msg error, code:%d", pSyncNode->vgId, code);
2228 2229 2230 2231
        rpcFreeCont(rpcMsg.pCont);
        syncTimeoutDestroy(pSyncMsg);
        return;
      }
M
Minghao Li 已提交
2232
    } else {
2233
      sTrace("syncNodeEqElectTimer FpEqMsg is NULL");
M
Minghao Li 已提交
2234
    }
M
Minghao Li 已提交
2235 2236
    syncTimeoutDestroy(pSyncMsg);

M
Minghao Li 已提交
2237
    // reset timer ms
2238
    if (syncEnvIsStart()) {
2239 2240 2241 2242
      pSyncNode->electTimerMS = syncUtilElectRandomMS(pSyncNode->electBaseLine, 2 * pSyncNode->electBaseLine);
      taosTmrReset(syncNodeEqElectTimer, pSyncNode->electTimerMS, pSyncNode, gSyncEnv->pTimerManager,
                   &pSyncNode->pElectTimer);
    } else {
2243
      sError("sync env is stop, syncNodeEqElectTimer");
2244
    }
M
Minghao Li 已提交
2245
  } else {
2246
    sTrace("==syncNodeEqElectTimer== electTimerLogicClock:%" PRIu64 ", electTimerLogicClockUser:%" PRIu64 "",
M
Minghao Li 已提交
2247
           pSyncNode->electTimerLogicClock, pSyncNode->electTimerLogicClockUser);
M
Minghao Li 已提交
2248 2249 2250
  }
}

M
Minghao Li 已提交
2251 2252
static void syncNodeEqHeartbeatTimer(void* param, void* tmrId) {
  SSyncNode* pSyncNode = (SSyncNode*)param;
2253 2254 2255 2256 2257 2258 2259 2260 2261 2262 2263 2264
  if (pSyncNode->replicaNum > 1) {
    if (atomic_load_64(&pSyncNode->heartbeatTimerLogicClockUser) <=
        atomic_load_64(&pSyncNode->heartbeatTimerLogicClock)) {
      SyncTimeout* pSyncMsg =
          syncTimeoutBuild2(SYNC_TIMEOUT_HEARTBEAT, atomic_load_64(&pSyncNode->heartbeatTimerLogicClock),
                            pSyncNode->heartbeatTimerMS, pSyncNode->vgId, pSyncNode);
      SRpcMsg rpcMsg;
      syncTimeout2RpcMsg(pSyncMsg, &rpcMsg);
      syncRpcMsgLog2((char*)"==syncNodeEqHeartbeatTimer==", &rpcMsg);
      if (pSyncNode->FpEqMsg != NULL) {
        int32_t code = pSyncNode->FpEqMsg(pSyncNode->msgcb, &rpcMsg);
        if (code != 0) {
S
Shengliang Guan 已提交
2265
          sError("vgId:%d, sync enqueue timer msg error, code:%d", pSyncNode->vgId, code);
2266 2267 2268 2269 2270 2271
          rpcFreeCont(rpcMsg.pCont);
          syncTimeoutDestroy(pSyncMsg);
          return;
        }
      } else {
        sError("syncNodeEqHeartbeatTimer FpEqMsg is NULL");
2272
      }
2273
      syncTimeoutDestroy(pSyncMsg);
M
Minghao Li 已提交
2274

2275 2276 2277 2278 2279 2280
      if (syncEnvIsStart()) {
        taosTmrReset(syncNodeEqHeartbeatTimer, pSyncNode->heartbeatTimerMS, pSyncNode, gSyncEnv->pTimerManager,
                     &pSyncNode->pHeartbeatTimer);
      } else {
        sError("sync env is stop, syncNodeEqHeartbeatTimer");
      }
2281
    } else {
2282 2283 2284
      sTrace("==syncNodeEqHeartbeatTimer== heartbeatTimerLogicClock:%" PRIu64 ", heartbeatTimerLogicClockUser:%" PRIu64
             "",
             pSyncNode->heartbeatTimerLogicClock, pSyncNode->heartbeatTimerLogicClockUser);
2285
    }
M
Minghao Li 已提交
2286 2287 2288
  }
}

M
Minghao Li 已提交
2289 2290
static int32_t syncNodeEqNoop(SSyncNode* ths) {
  int32_t ret = 0;
M
Minghao Li 已提交
2291
  ASSERT(ths->state == TAOS_SYNC_STATE_LEADER);
M
Minghao Li 已提交
2292

2293
  SyncIndex       index = ths->pLogStore->syncLogWriteIndex(ths->pLogStore);
M
Minghao Li 已提交
2294
  SyncTerm        term = ths->pRaftStore->currentTerm;
M
Minghao Li 已提交
2295
  SSyncRaftEntry* pEntry = syncEntryBuildNoop(term, index, ths->vgId);
M
Minghao Li 已提交
2296
  ASSERT(pEntry != NULL);
M
Minghao Li 已提交
2297 2298 2299 2300

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

S
Shengliang Guan 已提交
2304
  SRpcMsg rpcMsg = {0};
M
Minghao Li 已提交
2305
  syncClientRequest2RpcMsg(pSyncMsg, &rpcMsg);
M
Minghao Li 已提交
2306
  if (ths->FpEqMsg != NULL) {
S
Shengliang Guan 已提交
2307
    ths->FpEqMsg(ths->msgcb, &rpcMsg);
M
Minghao Li 已提交
2308 2309 2310
  } else {
    sTrace("syncNodeEqNoop pSyncNode->FpEqMsg is NULL");
  }
M
Minghao Li 已提交
2311

wafwerar's avatar
wafwerar 已提交
2312
  taosMemoryFree(serialized);
M
Minghao Li 已提交
2313 2314 2315 2316 2317 2318 2319 2320
  syncClientRequestDestroy(pSyncMsg);

  return ret;
}

static int32_t syncNodeAppendNoop(SSyncNode* ths) {
  int32_t ret = 0;

2321
  SyncIndex       index = ths->pLogStore->syncLogWriteIndex(ths->pLogStore);
M
Minghao Li 已提交
2322
  SyncTerm        term = ths->pRaftStore->currentTerm;
M
Minghao Li 已提交
2323
  SSyncRaftEntry* pEntry = syncEntryBuildNoop(term, index, ths->vgId);
M
Minghao Li 已提交
2324
  ASSERT(pEntry != NULL);
M
Minghao Li 已提交
2325 2326

  if (ths->state == TAOS_SYNC_STATE_LEADER) {
2327 2328
    // ths->pLogStore->appendEntry(ths->pLogStore, pEntry);
    ths->pLogStore->syncLogAppendEntry(ths->pLogStore, pEntry);
M
Minghao Li 已提交
2329 2330 2331
    syncNodeReplicate(ths);
  }

M
Minghao Li 已提交
2332
  syncEntryDestory(pEntry);
M
Minghao Li 已提交
2333 2334 2335
  return ret;
}

M
Minghao Li 已提交
2336
// on message ----
M
Minghao Li 已提交
2337 2338
int32_t syncNodeOnPingCb(SSyncNode* ths, SyncPing* pMsg) {
  // log state
2339
  char logBuf[1024] = {0};
M
Minghao Li 已提交
2340 2341 2342 2343 2344 2345
  snprintf(logBuf, sizeof(logBuf),
           "==syncNodeOnPingCb== vgId:%d, state: %d, %s, term:%lu electTimerLogicClock:%lu, "
           "electTimerLogicClockUser:%lu, electTimerMS:%d",
           ths->vgId, ths->state, syncUtilState2String(ths->state), ths->pRaftStore->currentTerm,
           ths->electTimerLogicClock, ths->electTimerLogicClockUser, ths->electTimerMS);

M
Minghao Li 已提交
2346
  int32_t ret = 0;
M
Minghao Li 已提交
2347 2348
  syncPingLog2(logBuf, pMsg);
  SyncPingReply* pMsgReply = syncPingReplyBuild3(&ths->myRaftId, &pMsg->srcId, ths->vgId);
M
Minghao Li 已提交
2349 2350
  SRpcMsg        rpcMsg;
  syncPingReply2RpcMsg(pMsgReply, &rpcMsg);
M
Minghao Li 已提交
2351 2352 2353 2354 2355 2356 2357 2358

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

M
Minghao Li 已提交
2359 2360 2361 2362 2363
  syncNodeSendMsgById(&pMsgReply->destId, ths, &rpcMsg);

  return ret;
}

M
Minghao Li 已提交
2364
int32_t syncNodeOnPingReplyCb(SSyncNode* ths, SyncPingReply* pMsg) {
M
Minghao Li 已提交
2365 2366 2367 2368
  int32_t ret = 0;
  syncPingReplyLog2("==syncNodeOnPingReplyCb==", pMsg);
  return ret;
}
M
Minghao Li 已提交
2369

M
Minghao Li 已提交
2370 2371 2372 2373 2374 2375 2376 2377 2378 2379
// 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 已提交
2380
int32_t syncNodeOnClientRequestCb(SSyncNode* ths, SyncClientRequest* pMsg) {
M
Minghao Li 已提交
2381 2382 2383
  int32_t ret = 0;
  syncClientRequestLog2("==syncNodeOnClientRequestCb==", pMsg);

M
Minghao Li 已提交
2384
  SyncIndex       index = ths->pLogStore->syncLogWriteIndex(ths->pLogStore);
M
Minghao Li 已提交
2385 2386
  SyncTerm        term = ths->pRaftStore->currentTerm;
  SSyncRaftEntry* pEntry = syncEntryBuild2((SyncClientRequest*)pMsg, term, index);
M
Minghao Li 已提交
2387
  ASSERT(pEntry != NULL);
M
Minghao Li 已提交
2388

M
Minghao Li 已提交
2389
  if (ths->state == TAOS_SYNC_STATE_LEADER) {
2390 2391
    // ths->pLogStore->appendEntry(ths->pLogStore, pEntry);
    ths->pLogStore->syncLogAppendEntry(ths->pLogStore, pEntry);
M
Minghao Li 已提交
2392 2393

    // start replicate right now!
M
Minghao Li 已提交
2394
    syncNodeReplicate(ths);
M
Minghao Li 已提交
2395

M
Minghao Li 已提交
2396 2397 2398 2399
    // pre commit
    SRpcMsg rpcMsg;
    syncEntry2OriginalRpc(pEntry, &rpcMsg);

M
Minghao Li 已提交
2400
    if (ths->pFsm != NULL) {
2401
      // if (ths->pFsm->FpPreCommitCb != NULL && pEntry->originalRpcType != TDMT_SYNC_NOOP) {
M
Minghao Li 已提交
2402
      if (ths->pFsm->FpPreCommitCb != NULL && syncUtilUserPreCommit(pEntry->originalRpcType)) {
2403
        SFsmCbMeta cbMeta = {0};
M
Minghao Li 已提交
2404
        cbMeta.index = pEntry->index;
2405
        cbMeta.lastConfigIndex = syncNodeGetSnapshotConfigIndex(ths, cbMeta.index);
M
Minghao Li 已提交
2406 2407 2408 2409 2410
        cbMeta.isWeak = pEntry->isWeak;
        cbMeta.code = 0;
        cbMeta.state = ths->state;
        cbMeta.seqNum = pEntry->seqNum;
        ths->pFsm->FpPreCommitCb(ths->pFsm, &rpcMsg, cbMeta);
M
Minghao Li 已提交
2411
      }
M
Minghao Li 已提交
2412 2413 2414
    }
    rpcFreeCont(rpcMsg.pCont);

M
Minghao Li 已提交
2415 2416 2417
    // only myself, maybe commit
    syncMaybeAdvanceCommitIndex(ths);

M
Minghao Li 已提交
2418
  } else {
M
Minghao Li 已提交
2419 2420 2421 2422
    // pre commit
    SRpcMsg rpcMsg;
    syncEntry2OriginalRpc(pEntry, &rpcMsg);

M
Minghao Li 已提交
2423
    if (ths->pFsm != NULL) {
2424
      // if (ths->pFsm->FpPreCommitCb != NULL && pEntry->originalRpcType != TDMT_SYNC_NOOP) {
M
Minghao Li 已提交
2425
      if (ths->pFsm->FpPreCommitCb != NULL && syncUtilUserPreCommit(pEntry->originalRpcType)) {
2426
        SFsmCbMeta cbMeta = {0};
M
Minghao Li 已提交
2427
        cbMeta.index = pEntry->index;
2428
        cbMeta.lastConfigIndex = syncNodeGetSnapshotConfigIndex(ths, cbMeta.index);
M
Minghao Li 已提交
2429 2430 2431 2432 2433
        cbMeta.isWeak = pEntry->isWeak;
        cbMeta.code = 1;
        cbMeta.state = ths->state;
        cbMeta.seqNum = pEntry->seqNum;
        ths->pFsm->FpPreCommitCb(ths->pFsm, &rpcMsg, cbMeta);
M
Minghao Li 已提交
2434
      }
M
Minghao Li 已提交
2435 2436
    }
    rpcFreeCont(rpcMsg.pCont);
M
Minghao Li 已提交
2437 2438
  }

M
Minghao Li 已提交
2439
  syncEntryDestory(pEntry);
M
Minghao Li 已提交
2440
  return ret;
2441
}
M
Minghao Li 已提交
2442 2443 2444

static void syncFreeNode(void* param) {
  SSyncNode* pNode = param;
M
Minghao Li 已提交
2445 2446
  // inner object already free
  // syncNodePrint2((char*)"==syncFreeNode==", pNode);
M
Minghao Li 已提交
2447

wafwerar's avatar
wafwerar 已提交
2448
  taosMemoryFree(pNode);
M
Minghao Li 已提交
2449
}
S
Shengliang Guan 已提交
2450 2451 2452 2453

const char* syncStr(ESyncState state) {
  switch (state) {
    case TAOS_SYNC_STATE_FOLLOWER:
2454
      return "follower";
S
Shengliang Guan 已提交
2455
    case TAOS_SYNC_STATE_CANDIDATE:
2456
      return "candidate";
S
Shengliang Guan 已提交
2457
    case TAOS_SYNC_STATE_LEADER:
2458
      return "leader";
S
Shengliang Guan 已提交
2459
    default:
2460
      return "error";
S
Shengliang Guan 已提交
2461
  }
M
Minghao Li 已提交
2462
}
2463

2464
static int32_t syncDoLeaderTransfer(SSyncNode* ths, SRpcMsg* pRpcMsg, SSyncRaftEntry* pEntry) {
M
Minghao Li 已提交
2465 2466
  SyncLeaderTransfer* pSyncLeaderTransfer = syncLeaderTransferFromRpcMsg2(pRpcMsg);

M
Minghao Li 已提交
2467
  syncNodeEventLog(ths, "begin leader transfer");
M
Minghao Li 已提交
2468

M
Minghao Li 已提交
2469 2470 2471
  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 已提交
2472

M
Minghao Li 已提交
2473 2474
  bool same = sameId || sameNodeInfo;
  if (same) {
M
Minghao Li 已提交
2475 2476 2477 2478
    // reset elect timer now!
    int32_t electMS = 1;
    int32_t ret = syncNodeRestartElectTimer(ths, electMS);
    ASSERT(ret == 0);
M
Minghao Li 已提交
2479 2480 2481 2482 2483 2484

    char eventLog[256];
    snprintf(eventLog, sizeof(eventLog), "maybe leader transfer to %s:%d %lu",
             pSyncLeaderTransfer->newNodeInfo.nodeFqdn, pSyncLeaderTransfer->newNodeInfo.nodePort,
             pSyncLeaderTransfer->newLeaderId.addr);
    syncNodeEventLog(ths, eventLog);
2485 2486
  }

M
Minghao Li 已提交
2487
  if (ths->pFsm->FpLeaderTransferCb != NULL) {
2488
    SFsmCbMeta cbMeta = {0};
M
Minghao Li 已提交
2489 2490 2491 2492
    cbMeta.code = 0;
    cbMeta.currentTerm = ths->pRaftStore->currentTerm;
    cbMeta.flag = 0;
    cbMeta.index = pEntry->index;
2493
    cbMeta.lastConfigIndex = syncNodeGetSnapshotConfigIndex(ths, cbMeta.index);
M
Minghao Li 已提交
2494 2495 2496 2497 2498
    cbMeta.isWeak = pEntry->isWeak;
    cbMeta.seqNum = pEntry->seqNum;
    cbMeta.state = ths->state;
    cbMeta.term = pEntry->term;
    ths->pFsm->FpLeaderTransferCb(ths->pFsm, pRpcMsg, cbMeta);
2499 2500
  }

M
Minghao Li 已提交
2501
  syncLeaderTransferDestroy(pSyncLeaderTransfer);
2502 2503 2504
  return 0;
}

2505 2506 2507 2508 2509 2510 2511 2512 2513 2514 2515 2516 2517 2518 2519
int32_t syncNodeUpdateNewConfigIndex(SSyncNode* ths, SSyncCfg* pNewCfg) {
  for (int i = 0; i < pNewCfg->replicaNum; ++i) {
    SRaftId raftId;
    raftId.addr = syncUtilAddr2U64((pNewCfg->nodeInfo)[i].nodeFqdn, (pNewCfg->nodeInfo)[i].nodePort);
    raftId.vgId = ths->vgId;

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

  return -1;
}

M
Minghao Li 已提交
2520 2521 2522 2523 2524 2525 2526 2527 2528 2529 2530 2531 2532 2533 2534 2535 2536 2537 2538 2539 2540 2541 2542 2543 2544
static int32_t syncNodeConfigChangeFinish(SSyncNode* ths, SRpcMsg* pRpcMsg, SSyncRaftEntry* pEntry) {
  SyncReconfigFinish* pFinish = syncReconfigFinishFromRpcMsg2(pRpcMsg);
  ASSERT(pFinish);

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

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

    ths->pFsm->FpReConfigCb(ths->pFsm, pRpcMsg, cbMeta);
  }

2545
  // clear changing
M
Minghao Li 已提交
2546 2547 2548 2549 2550 2551 2552 2553 2554 2555 2556 2557 2558 2559 2560 2561 2562 2563
  ths->changing = false;

  char  tmpbuf[512];
  char* oldStr = syncCfg2SimpleStr(&(pFinish->oldCfg));
  char* newStr = syncCfg2SimpleStr(&(pFinish->newCfg));
  snprintf(tmpbuf, sizeof(tmpbuf), "config change finish from %d to %d, index:%ld, %s  -->  %s",
           pFinish->oldCfg.replicaNum, pFinish->newCfg.replicaNum, pFinish->newCfgIndex, oldStr, newStr);
  taosMemoryFree(oldStr);
  taosMemoryFree(newStr);
  syncNodeEventLog(ths, tmpbuf);

  syncReconfigFinishDestroy(pFinish);

  return 0;
}

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

M
Minghao Li 已提交
2567
  // old config
2568 2569
  SSyncCfg oldSyncCfg = ths->pRaftCfg->cfg;

M
Minghao Li 已提交
2570
  // new config
2571 2572 2573 2574 2575
  SSyncCfg newSyncCfg;
  int32_t  ret = syncCfgFromStr(pRpcMsg->pCont, &newSyncCfg);
  ASSERT(ret == 0);

  // update new config myIndex
2576 2577
  syncNodeUpdateNewConfigIndex(ths, &newSyncCfg);

M
Minghao Li 已提交
2578 2579
  // do config change
  syncNodeDoConfigChange(ths, &newSyncCfg, pEntry->index);
2580

M
Minghao Li 已提交
2581 2582 2583 2584 2585 2586
  // set pFinish
  pFinish->oldCfg = oldSyncCfg;
  pFinish->newCfg = newSyncCfg;
  pFinish->newCfgIndex = pEntry->index;
  pFinish->newCfgTerm = pEntry->term;
  pFinish->newCfgSeqNum = pEntry->seqNum;
2587

M
Minghao Li 已提交
2588 2589
  return 0;
}
2590

M
Minghao Li 已提交
2591 2592 2593 2594 2595 2596 2597 2598 2599
static int32_t syncNodeProposeConfigChangeFinish(SSyncNode* ths, SyncReconfigFinish* pFinish) {
  SRpcMsg rpcMsg;
  syncReconfigFinish2RpcMsg(pFinish, &rpcMsg);

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

2603
int32_t syncNodeCommit(SSyncNode* ths, SyncIndex beginIndex, SyncIndex endIndex, uint64_t flag) {
2604 2605
  int32_t    code = 0;
  ESyncState state = flag;
M
Minghao Li 已提交
2606 2607 2608 2609

  char eventLog[128];
  snprintf(eventLog, sizeof(eventLog), "commit by wal from index:%ld to index:%ld", beginIndex, endIndex);
  syncNodeEventLog(ths, eventLog);
2610 2611 2612 2613 2614 2615 2616 2617 2618 2619 2620 2621 2622

  // execute fsm
  if (ths->pFsm != NULL) {
    for (SyncIndex i = beginIndex; i <= endIndex; ++i) {
      if (i != SYNC_INDEX_INVALID) {
        SSyncRaftEntry* pEntry;
        code = ths->pLogStore->syncLogGetEntry(ths->pLogStore, i, &pEntry);
        ASSERT(code == 0);
        ASSERT(pEntry != NULL);

        SRpcMsg rpcMsg;
        syncEntry2OriginalRpc(pEntry, &rpcMsg);

2623
        // user commit
2624
        if (ths->pFsm->FpCommitCb != NULL && syncUtilUserCommit(pEntry->originalRpcType)) {
2625
          SFsmCbMeta cbMeta = {0};
2626
          cbMeta.index = pEntry->index;
2627
          cbMeta.lastConfigIndex = syncNodeGetSnapshotConfigIndex(ths, cbMeta.index);
2628 2629 2630 2631 2632 2633 2634 2635 2636 2637 2638 2639
          cbMeta.isWeak = pEntry->isWeak;
          cbMeta.code = 0;
          cbMeta.state = ths->state;
          cbMeta.seqNum = pEntry->seqNum;
          cbMeta.term = pEntry->term;
          cbMeta.currentTerm = ths->pRaftStore->currentTerm;
          cbMeta.flag = flag;

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

        // config change
2640
        if (pEntry->originalRpcType == TDMT_SYNC_CONFIG_CHANGE) {
M
Minghao Li 已提交
2641 2642 2643 2644 2645 2646 2647 2648 2649 2650 2651 2652 2653 2654 2655
          SyncReconfigFinish* pFinish = syncReconfigFinishBuild(ths->vgId);
          ASSERT(pFinish != NULL);

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

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

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

M
Minghao Li 已提交
2659
        // leader transfer
2660 2661 2662
        if (pEntry->originalRpcType == TDMT_SYNC_LEADER_TRANSFER) {
          code = syncDoLeaderTransfer(ths, &rpcMsg, pEntry);
          ASSERT(code == 0);
2663 2664 2665
        }

        // restore finish
2666
        // if only snapshot, a noop entry will be append, so syncLogLastIndex is always ok
2667 2668 2669 2670 2671 2672
        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 已提交
2673 2674 2675 2676

            char eventLog[128];
            snprintf(eventLog, sizeof(eventLog), "restore finish, index:%ld", pEntry->index);
            syncNodeEventLog(ths, eventLog);
2677 2678 2679 2680 2681 2682 2683 2684 2685
          }
        }

        rpcFreeCont(rpcMsg.pCont);
        syncEntryDestory(pEntry);
      }
    }
  }
  return 0;
2686 2687 2688 2689 2690 2691 2692 2693 2694
}

bool syncNodeInRaftGroup(SSyncNode* ths, SRaftId* pRaftId) {
  for (int i = 0; i < ths->replicaNum; ++i) {
    if (syncUtilSameId(&((ths->replicasId)[i]), pRaftId)) {
      return true;
    }
  }
  return false;
M
Minghao Li 已提交
2695 2696 2697 2698 2699 2700 2701 2702 2703 2704
}

SSyncSnapshotSender* syncNodeGetSnapshotSender(SSyncNode* ths, SRaftId* pDestId) {
  SSyncSnapshotSender* pSender = NULL;
  for (int i = 0; i < ths->replicaNum; ++i) {
    if (syncUtilSameId(pDestId, &((ths->replicasId)[i]))) {
      pSender = (ths->senders)[i];
    }
  }
  return pSender;
M
Minghao Li 已提交
2705
}
M
Minghao Li 已提交
2706 2707 2708 2709 2710 2711 2712 2713 2714 2715 2716 2717 2718 2719 2720 2721 2722 2723 2724 2725 2726 2727 2728 2729 2730

bool syncNodeCanChange(SSyncNode* pSyncNode) {
  if (pSyncNode->changing) {
    sError("sync cannot change");
    return false;
  }

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

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

  return true;
}