syncPipeline.c 40.8 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17
/*
 * 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/>.
 */

#define _DEFAULT_SOURCE

18
#include "syncPipeline.h"
19
#include "syncCommit.h"
20
#include "syncIndexMgr.h"
21 22 23 24
#include "syncInt.h"
#include "syncRaftEntry.h"
#include "syncRaftStore.h"
#include "syncReplication.h"
25 26
#include "syncRespMgr.h"
#include "syncSnapshot.h"
27 28
#include "syncUtil.h"

29 30 31 32 33
static bool syncIsMsgBlock(tmsg_t type) {
  return (type == TDMT_VND_CREATE_TABLE) || (type == TDMT_VND_ALTER_TABLE) || (type == TDMT_VND_DROP_TABLE) ||
         (type == TDMT_VND_UPDATE_TAG_VAL) || (type == TDMT_VND_ALTER_CONFIRM);
}

34 35 36 37
FORCE_INLINE static int64_t syncGetRetryMaxWaitMs() {
  return SYNC_LOG_REPL_RETRY_WAIT_MS * (1 << SYNC_MAX_RETRY_BACKOFF);
}

38 39 40 41 42 43 44 45 46 47 48 49 50
int64_t syncLogBufferGetEndIndex(SSyncLogBuffer* pBuf) {
  taosThreadMutexLock(&pBuf->mutex);
  int64_t index = pBuf->endIndex;
  taosThreadMutexUnlock(&pBuf->mutex);
  return index;
}

int32_t syncLogBufferAppend(SSyncLogBuffer* pBuf, SSyncNode* pNode, SSyncRaftEntry* pEntry) {
  taosThreadMutexLock(&pBuf->mutex);
  syncLogBufferValidate(pBuf);
  SyncIndex index = pEntry->index;

  if (index - pBuf->startIndex >= pBuf->size) {
51
    sError("vgId:%d, failed to append due to sync log buffer full. index:%" PRId64 "", pNode->vgId, index);
52
    goto _err;
53 54
  }

55
  ASSERT(index == pBuf->endIndex);
56 57

  SSyncRaftEntry* pExist = pBuf->entries[index % pBuf->size].pItem;
58
  ASSERT(pExist == NULL);
59 60 61

  // initial log buffer with at least one item, e.g. commitIndex
  SSyncRaftEntry* pMatch = pBuf->entries[(index - 1 + pBuf->size) % pBuf->size].pItem;
S
Shengliang Guan 已提交
62
  ASSERTS(pMatch != NULL, "no matched log entry");
63
  ASSERT(pMatch->index + 1 == index);
64 65 66 67 68 69 70 71 72

  SSyncLogBufEntry tmp = {.pItem = pEntry, .prevLogIndex = pMatch->index, .prevLogTerm = pMatch->term};
  pBuf->entries[index % pBuf->size] = tmp;
  pBuf->endIndex = index + 1;

  syncLogBufferValidate(pBuf);
  taosThreadMutexUnlock(&pBuf->mutex);
  return 0;

73
_err:
74 75 76 77 78 79 80 81 82 83 84 85
  syncLogBufferValidate(pBuf);
  taosThreadMutexUnlock(&pBuf->mutex);
  return -1;
}

SyncTerm syncLogReplMgrGetPrevLogTerm(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncIndex index) {
  SSyncLogBuffer* pBuf = pNode->pLogBuf;
  SSyncRaftEntry* pEntry = NULL;
  SyncIndex       prevIndex = index - 1;
  SyncTerm        prevLogTerm = -1;
  terrno = TSDB_CODE_SUCCESS;

86
  if (prevIndex == -1 && pNode->pLogStore->syncLogBeginIndex(pNode->pLogStore) == 0) return 0;
87

88
  if (prevIndex > pBuf->matchIndex) {
89 90 91 92
    terrno = TSDB_CODE_WAL_LOG_NOT_EXIST;
    return -1;
  }

93
  ASSERT(index - 1 == prevIndex);
94

95 96
  if (prevIndex >= pBuf->startIndex) {
    pEntry = pBuf->entries[(prevIndex + pBuf->size) % pBuf->size].pItem;
S
Shengliang Guan 已提交
97
    ASSERTS(pEntry != NULL, "no log entry found");
98
    prevLogTerm = pEntry->term;
99 100 101
    return prevLogTerm;
  }

102
  if (pMgr && pMgr->startIndex <= prevIndex && prevIndex < pMgr->endIndex) {
103
    int64_t timeMs = pMgr->states[(prevIndex + pMgr->size) % pMgr->size].timeMs;
S
Shengliang Guan 已提交
104
    ASSERTS(timeMs != 0, "no log entry found");
105
    prevLogTerm = pMgr->states[(prevIndex + pMgr->size) % pMgr->size].term;
106
    ASSERT(prevIndex == 0 || prevLogTerm != 0);
107 108 109
    return prevLogTerm;
  }

110 111 112
  SSnapshot snapshot = {0};
  pNode->pFsm->FpGetSnapshotInfo(pNode->pFsm, &snapshot);
  if (prevIndex == snapshot.lastApplyIndex) {
113 114 115 116 117 118 119 120 121 122
    return snapshot.lastApplyTerm;
  }

  if (pNode->pLogStore->syncLogGetEntry(pNode->pLogStore, prevIndex, &pEntry) == 0) {
    prevLogTerm = pEntry->term;
    syncEntryDestroy(pEntry);
    pEntry = NULL;
    return prevLogTerm;
  }

123
  sInfo("vgId:%d, failed to get log term since %s. index:%" PRId64, pNode->vgId, terrstr(), prevIndex);
124 125 126 127 128 129 130 131
  terrno = TSDB_CODE_WAL_LOG_NOT_EXIST;
  return -1;
}

SSyncRaftEntry* syncEntryBuildDummy(SyncTerm term, SyncIndex index, int32_t vgId) {
  return syncEntryBuildNoop(term, index, vgId);
}

132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151
int32_t syncLogValidateAlignmentOfCommit(SSyncNode* pNode, SyncIndex commitIndex) {
  SyncIndex firstVer = pNode->pLogStore->syncLogBeginIndex(pNode->pLogStore);
  if (firstVer > commitIndex + 1) {
    sError("vgId:%d, firstVer of WAL log greater than tsdb commit version + 1. firstVer: %" PRId64
           ", tsdb commit version: %" PRId64 "",
           pNode->vgId, firstVer, commitIndex);
    return -1;
  }

  SyncIndex lastVer = pNode->pLogStore->syncLogLastIndex(pNode->pLogStore);
  if (lastVer < commitIndex) {
    sError("vgId:%d, lastVer of WAL log less than tsdb commit version. lastVer: %" PRId64
           ", tsdb commit version: %" PRId64 "",
           pNode->vgId, lastVer, commitIndex);
    return -1;
  }

  return 0;
}

152
int32_t syncLogBufferInitWithoutLock(SSyncLogBuffer* pBuf, SSyncNode* pNode) {
S
Shengliang Guan 已提交
153 154 155
  ASSERTS(pNode->pLogStore != NULL, "log store not created");
  ASSERTS(pNode->pFsm != NULL, "pFsm not registered");
  ASSERTS(pNode->pFsm->FpGetSnapshotInfo != NULL, "FpGetSnapshotInfo not registered");
156

157 158 159
  SSnapshot snapshot = {0};
  pNode->pFsm->FpGetSnapshotInfo(pNode->pFsm, &snapshot);

160
  SyncIndex commitIndex = snapshot.lastApplyIndex;
161
  SyncTerm  commitTerm = TMAX(snapshot.lastApplyTerm, 0);
162
  if (syncLogValidateAlignmentOfCommit(pNode, commitIndex)) {
163 164 165 166
    terrno = TSDB_CODE_WAL_LOG_INCOMPLETE;
    goto _err;
  }

167
  SyncIndex lastVer = pNode->pLogStore->syncLogLastIndex(pNode->pLogStore);
168
  ASSERT(lastVer >= commitIndex);
169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192
  SyncIndex toIndex = lastVer;
  // update match index
  pBuf->commitIndex = commitIndex;
  pBuf->matchIndex = toIndex;
  pBuf->endIndex = toIndex + 1;

  // load log entries in reverse order
  SSyncLogStore*  pLogStore = pNode->pLogStore;
  SyncIndex       index = toIndex;
  SSyncRaftEntry* pEntry = NULL;
  bool            takeDummy = false;

  while (true) {
    if (index <= pBuf->commitIndex) {
      takeDummy = true;
      break;
    }

    if (pLogStore->syncLogGetEntry(pLogStore, index, &pEntry) < 0) {
      sError("vgId:%d, failed to get log entry since %s. index:%" PRId64 "", pNode->vgId, terrstr(), index);
      break;
    }

    bool taken = false;
193 194
    int  emptySize = 5;
    if (toIndex - index + 1 <= pBuf->size - emptySize) {
195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215
      SSyncLogBufEntry tmp = {.pItem = pEntry, .prevLogIndex = -1, .prevLogTerm = -1};
      pBuf->entries[index % pBuf->size] = tmp;
      taken = true;
    }

    if (index < toIndex) {
      pBuf->entries[(index + 1) % pBuf->size].prevLogIndex = pEntry->index;
      pBuf->entries[(index + 1) % pBuf->size].prevLogTerm = pEntry->term;
    }

    if (!taken) {
      syncEntryDestroy(pEntry);
      pEntry = NULL;
      break;
    }

    index--;
  }

  // put a dummy record at commitIndex if present in log buffer
  if (takeDummy) {
216
    ASSERT(index == pBuf->commitIndex);
217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234

    SSyncRaftEntry* pDummy = syncEntryBuildDummy(commitTerm, commitIndex, pNode->vgId);
    if (pDummy == NULL) {
      terrno = TSDB_CODE_OUT_OF_MEMORY;
      goto _err;
    }
    SSyncLogBufEntry tmp = {.pItem = pDummy, .prevLogIndex = commitIndex - 1, .prevLogTerm = commitTerm};
    pBuf->entries[(commitIndex + pBuf->size) % pBuf->size] = tmp;

    if (index < toIndex) {
      pBuf->entries[(index + 1) % pBuf->size].prevLogIndex = commitIndex;
      pBuf->entries[(index + 1) % pBuf->size].prevLogTerm = commitTerm;
    }
  }

  // update startIndex
  pBuf->startIndex = takeDummy ? index : index + 1;

235 236 237
  sInfo("vgId:%d, init sync log buffer. buffer: [%" PRId64 " %" PRId64 " %" PRId64 ", %" PRId64 ")", pNode->vgId,
        pBuf->startIndex, pBuf->commitIndex, pBuf->matchIndex, pBuf->endIndex);

238 239 240 241 242 243 244 245
  // validate
  syncLogBufferValidate(pBuf);
  return 0;

_err:
  return -1;
}

246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270
int32_t syncLogBufferInit(SSyncLogBuffer* pBuf, SSyncNode* pNode) {
  taosThreadMutexLock(&pBuf->mutex);
  int32_t ret = syncLogBufferInitWithoutLock(pBuf, pNode);
  taosThreadMutexUnlock(&pBuf->mutex);
  return ret;
}

int32_t syncLogBufferReInit(SSyncLogBuffer* pBuf, SSyncNode* pNode) {
  taosThreadMutexLock(&pBuf->mutex);
  for (SyncIndex index = pBuf->startIndex; index < pBuf->endIndex; index++) {
    SSyncRaftEntry* pEntry = pBuf->entries[(index + pBuf->size) % pBuf->size].pItem;
    if (pEntry == NULL) continue;
    syncEntryDestroy(pEntry);
    pEntry = NULL;
    memset(&pBuf->entries[(index + pBuf->size) % pBuf->size], 0, sizeof(pBuf->entries[0]));
  }
  pBuf->startIndex = pBuf->commitIndex = pBuf->matchIndex = pBuf->endIndex = 0;
  int32_t ret = syncLogBufferInitWithoutLock(pBuf, pNode);
  if (ret < 0) {
    sError("vgId:%d, failed to re-initialize sync log buffer since %s.", pNode->vgId, terrstr());
  }
  taosThreadMutexUnlock(&pBuf->mutex);
  return ret;
}

271
FORCE_INLINE SyncTerm syncLogBufferGetLastMatchTermWithoutLock(SSyncLogBuffer* pBuf) {
272 273
  SyncIndex       index = pBuf->matchIndex;
  SSyncRaftEntry* pEntry = pBuf->entries[(index + pBuf->size) % pBuf->size].pItem;
274
  ASSERT(pEntry != NULL);
275 276 277
  return pEntry->term;
}

278 279 280 281 282 283 284
SyncTerm syncLogBufferGetLastMatchTerm(SSyncLogBuffer* pBuf) {
  taosThreadMutexLock(&pBuf->mutex);
  SyncTerm term = syncLogBufferGetLastMatchTermWithoutLock(pBuf);
  taosThreadMutexUnlock(&pBuf->mutex);
  return term;
}

285 286 287
int32_t syncLogBufferAccept(SSyncLogBuffer* pBuf, SSyncNode* pNode, SSyncRaftEntry* pEntry, SyncTerm prevTerm) {
  taosThreadMutexLock(&pBuf->mutex);
  syncLogBufferValidate(pBuf);
S
Shengliang Guan 已提交
288 289 290
  int32_t         ret = -1;
  SyncIndex       index = pEntry->index;
  SyncIndex       prevIndex = pEntry->index - 1;
291
  SyncTerm        lastMatchTerm = syncLogBufferGetLastMatchTermWithoutLock(pBuf);
292 293
  SSyncRaftEntry* pExist = NULL;
  bool            inBuf = true;
294 295

  if (index <= pBuf->commitIndex) {
296 297
    sTrace("vgId:%d, already committed. index: %" PRId64 ", term: %" PRId64 ". log buffer: [%" PRId64 " %" PRId64
           " %" PRId64 ", %" PRId64 ")",
298 299
           pNode->vgId, pEntry->index, pEntry->term, pBuf->startIndex, pBuf->commitIndex, pBuf->matchIndex,
           pBuf->endIndex);
300
    SyncTerm term = syncLogReplMgrGetPrevLogTerm(NULL, pNode, index + 1);
301
    ASSERT(pEntry->term >= 0);
302 303 304
    if (term == pEntry->term) {
      ret = 0;
    }
305 306 307 308
    goto _out;
  }

  if (index - pBuf->startIndex >= pBuf->size) {
309
    sWarn("vgId:%d, out of buffer range. index: %" PRId64 ", term: %" PRId64 ". log buffer: [%" PRId64 " %" PRId64
310
          " %" PRId64 ", %" PRId64 ")",
311 312 313 314 315 316
          pNode->vgId, pEntry->index, pEntry->term, pBuf->startIndex, pBuf->commitIndex, pBuf->matchIndex,
          pBuf->endIndex);
    goto _out;
  }

  if (index > pBuf->matchIndex && lastMatchTerm != prevTerm) {
317
    sWarn("vgId:%d, not ready to accept. index: %" PRId64 ", term: %" PRId64 ": prevterm: %" PRId64
318 319 320 321 322 323 324
          " != lastmatch: %" PRId64 ". log buffer: [%" PRId64 " %" PRId64 " %" PRId64 ", %" PRId64 ")",
          pNode->vgId, pEntry->index, pEntry->term, prevTerm, lastMatchTerm, pBuf->startIndex, pBuf->commitIndex,
          pBuf->matchIndex, pBuf->endIndex);
    goto _out;
  }

  // check current in buffer
325
  pExist = syncLogBufferGetOneEntry(pBuf, pNode, index, &inBuf);
326
  if (pExist != NULL) {
327
    ASSERT(pEntry->index == pExist->index);
328
    if (pEntry->term != pExist->term) {
329
      (void)syncLogBufferRollback(pBuf, pNode, index);
330
    } else {
331
      sTrace("vgId:%d, duplicate log entry received. index: %" PRId64 ", term: %" PRId64 ". log buffer: [%" PRId64
332 333 334
             " %" PRId64 " %" PRId64 ", %" PRId64 ")",
             pNode->vgId, pEntry->index, pEntry->term, pBuf->startIndex, pBuf->commitIndex, pBuf->matchIndex,
             pBuf->endIndex);
335 336
      SyncTerm existPrevTerm = syncLogReplMgrGetPrevLogTerm(NULL, pNode, index);
      ASSERT(pEntry->term == pExist->term && (pEntry->index > pBuf->matchIndex || prevTerm == existPrevTerm));
337 338 339 340 341 342
      ret = 0;
      goto _out;
    }
  }

  // update
B
Benguang Zhao 已提交
343 344
  ASSERT(pBuf->startIndex < index);
  ASSERT(index - pBuf->startIndex < pBuf->size);
345
  ASSERT(pBuf->entries[index % pBuf->size].pItem == NULL);
346 347 348 349 350 351 352 353 354 355 356 357
  SSyncLogBufEntry tmp = {.pItem = pEntry, .prevLogIndex = prevIndex, .prevLogTerm = prevTerm};
  pEntry = NULL;
  pBuf->entries[index % pBuf->size] = tmp;

  // update end index
  pBuf->endIndex = TMAX(index + 1, pBuf->endIndex);

  // success
  ret = 0;

_out:
  syncEntryDestroy(pEntry);
358 359 360 361
  if (!inBuf) {
    syncEntryDestroy(pExist);
    pExist = NULL;
  }
362 363 364 365 366
  syncLogBufferValidate(pBuf);
  taosThreadMutexUnlock(&pBuf->mutex);
  return ret;
}

367 368 369 370 371
static inline bool syncLogStoreNeedFlush(SSyncRaftEntry* pEntry, int32_t replicaNum) {
  return (replicaNum > 1) && (pEntry->originalRpcType == TDMT_VND_COMMIT);
}

int32_t syncLogStorePersist(SSyncLogStore* pLogStore, SSyncNode* pNode, SSyncRaftEntry* pEntry) {
372
  ASSERT(pEntry->index >= 0);
373 374 375 376 377 378
  SyncIndex lastVer = pLogStore->syncLogLastIndex(pLogStore);
  if (lastVer >= pEntry->index && pLogStore->syncLogTruncate(pLogStore, pEntry->index) < 0) {
    sError("failed to truncate log store since %s. from index:%" PRId64 "", terrstr(), pEntry->index);
    return -1;
  }
  lastVer = pLogStore->syncLogLastIndex(pLogStore);
379
  ASSERT(pEntry->index == lastVer + 1);
380

381 382
  bool doFsync = syncLogStoreNeedFlush(pEntry, pNode->replicaNum);
  if (pLogStore->syncLogAppendEntry(pLogStore, pEntry, doFsync) < 0) {
383
    sError("failed to append sync log entry since %s. index:%" PRId64 ", term:%" PRId64 "", terrstr(), pEntry->index,
384 385 386 387 388
           pEntry->term);
    return -1;
  }

  lastVer = pLogStore->syncLogLastIndex(pLogStore);
389
  ASSERT(pEntry->index == lastVer);
390 391 392
  return 0;
}

393
int64_t syncLogBufferProceed(SSyncLogBuffer* pBuf, SSyncNode* pNode, SyncTerm* pMatchTerm) {
394 395 396 397 398 399 400 401
  taosThreadMutexLock(&pBuf->mutex);
  syncLogBufferValidate(pBuf);

  SSyncLogStore* pLogStore = pNode->pLogStore;
  int64_t        matchIndex = pBuf->matchIndex;

  while (pBuf->matchIndex + 1 < pBuf->endIndex) {
    int64_t index = pBuf->matchIndex + 1;
402
    ASSERT(index >= 0);
403 404 405 406 407 408 409

    // try to proceed
    SSyncLogBufEntry* pBufEntry = &pBuf->entries[index % pBuf->size];
    SyncIndex         prevLogIndex = pBufEntry->prevLogIndex;
    SyncTerm          prevLogTerm = pBufEntry->prevLogTerm;
    SSyncRaftEntry*   pEntry = pBufEntry->pItem;
    if (pEntry == NULL) {
410
      sTrace("vgId:%d, cannot proceed match index in log buffer. no raft entry at next pos of matchIndex:%" PRId64,
411 412 413 414
             pNode->vgId, pBuf->matchIndex);
      goto _out;
    }

415
    ASSERT(index == pEntry->index);
416 417 418

    // match
    SSyncRaftEntry* pMatch = pBuf->entries[(pBuf->matchIndex + pBuf->size) % pBuf->size].pItem;
419 420 421 422
    ASSERT(pMatch != NULL);
    ASSERT(pMatch->index == pBuf->matchIndex);
    ASSERT(pMatch->index + 1 == pEntry->index);
    ASSERT(prevLogIndex == pMatch->index);
423 424 425

    if (pMatch->term != prevLogTerm) {
      sInfo(
426
          "vgId:%d, mismatching sync log entries encountered. "
427 428 429 430 431 432 433 434 435 436
          "{ index:%" PRId64 ", term:%" PRId64
          " } "
          "{ index:%" PRId64 ", term:%" PRId64 ", prevLogIndex:%" PRId64 ", prevLogTerm:%" PRId64 " } ",
          pNode->vgId, pMatch->index, pMatch->term, pEntry->index, pEntry->term, prevLogIndex, prevLogTerm);
      goto _out;
    }

    // increase match index
    pBuf->matchIndex = index;

437 438
    sTrace("vgId:%d, log buffer proceed. start index: %" PRId64 ", match index: %" PRId64 ", end index: %" PRId64,
           pNode->vgId, pBuf->startIndex, pBuf->matchIndex, pBuf->endIndex);
439 440

    // replicate on demand
441
    (void)syncNodeReplicateWithoutLock(pNode);
442 443

    // persist
444
    if (syncLogStorePersist(pLogStore, pNode, pEntry) < 0) {
445 446
      sError("vgId:%d, failed to persist sync log entry from buffer since %s. index:%" PRId64, pNode->vgId, terrstr(),
             pEntry->index);
447 448
      goto _out;
    }
449
    ASSERT(pEntry->index == pBuf->matchIndex);
450 451 452 453 454 455 456 457

    // update my match index
    matchIndex = pBuf->matchIndex;
    syncIndexMgrSetIndex(pNode->pMatchIndex, &pNode->myRaftId, pBuf->matchIndex);
  }  // end of while

_out:
  pBuf->matchIndex = matchIndex;
458 459 460
  if (pMatchTerm) {
    *pMatchTerm = pBuf->entries[(matchIndex + pBuf->size) % pBuf->size].pItem->term;
  }
461 462 463 464 465
  syncLogBufferValidate(pBuf);
  taosThreadMutexUnlock(&pBuf->mutex);
  return matchIndex;
}

466 467
int32_t syncLogFsmExecute(SSyncNode* pNode, SSyncFSM* pFsm, ESyncState role, SyncTerm term, SSyncRaftEntry* pEntry,
                          int32_t applyCode) {
468 469 470 471
  if ((pNode->replicaNum == 1) && pNode->restoreFinish && pNode->vgId != 1) {
    return 0;
  }

472 473 474
  if (pNode->vgId != 1 && syncIsMsgBlock(pEntry->originalRpcType)) {
    sTrace("vgId:%d, blocking msg ready to execute, index:%" PRId64 ", term:%" PRId64 ", type:%s code:0x%x",
           pNode->vgId, pEntry->index, pEntry->term, TMSG_INFO(pEntry->originalRpcType), applyCode);
475 476
  }

B
Benguang Zhao 已提交
477 478 479 480 481
  if (pEntry->originalRpcType == TDMT_VND_COMMIT) {
    sInfo("vgId:%d, fsm execute vnode commit. index: %" PRId64 ", term: %" PRId64 "", pNode->vgId, pEntry->index,
          pEntry->term);
  }

482
  SRpcMsg rpcMsg = {.code = applyCode};
483 484 485 486
  syncEntry2OriginalRpc(pEntry, &rpcMsg);

  SFsmCbMeta cbMeta = {0};
  cbMeta.index = pEntry->index;
487
  cbMeta.lastConfigIndex = syncNodeGetSnapshotConfigIndex(pNode, pEntry->index);
488
  cbMeta.isWeak = pEntry->isWeak;
489
  cbMeta.code = applyCode;
490 491 492 493 494 495
  cbMeta.state = role;
  cbMeta.seqNum = pEntry->seqNum;
  cbMeta.term = pEntry->term;
  cbMeta.currentTerm = term;
  cbMeta.flag = -1;

496
  (void)syncRespMgrGetAndDel(pNode->pSyncRespMgr, cbMeta.seqNum, &rpcMsg.info);
497 498
  int32_t code = pFsm->FpCommitCb(pFsm, &rpcMsg, &cbMeta);
  return code;
499 500 501
}

int32_t syncLogBufferValidate(SSyncLogBuffer* pBuf) {
502 503 504 505 506
  ASSERT(pBuf->startIndex <= pBuf->matchIndex);
  ASSERT(pBuf->commitIndex <= pBuf->matchIndex);
  ASSERT(pBuf->matchIndex < pBuf->endIndex);
  ASSERT(pBuf->endIndex - pBuf->startIndex <= pBuf->size);
  ASSERT(pBuf->entries[(pBuf->matchIndex + pBuf->size) % pBuf->size].pItem);
507 508 509 510 511 512 513 514 515 516
  return 0;
}

int32_t syncLogBufferCommit(SSyncLogBuffer* pBuf, SSyncNode* pNode, int64_t commitIndex) {
  taosThreadMutexLock(&pBuf->mutex);
  syncLogBufferValidate(pBuf);

  SSyncLogStore*  pLogStore = pNode->pLogStore;
  SSyncFSM*       pFsm = pNode->pFsm;
  ESyncState      role = pNode->state;
S
Shengliang Guan 已提交
517
  SyncTerm        term = pNode->raftStore.currentTerm;
518
  SyncGroupId     vgId = pNode->vgId;
519
  int32_t         ret = -1;
520 521 522 523 524
  int64_t         upperIndex = TMIN(commitIndex, pBuf->matchIndex);
  SSyncRaftEntry* pEntry = NULL;
  bool            inBuf = false;

  if (commitIndex <= pBuf->commitIndex) {
525
    sDebug("vgId:%d, stale commit index. current:%" PRId64 ", notified:%" PRId64 "", vgId, pBuf->commitIndex,
526 527 528 529 530
           commitIndex);
    ret = 0;
    goto _out;
  }

531 532
  sTrace("vgId:%d, commit. log buffer: [%" PRId64 " %" PRId64 " %" PRId64 ", %" PRId64 "), role: %d, term: %" PRId64,
         pNode->vgId, pBuf->startIndex, pBuf->commitIndex, pBuf->matchIndex, pBuf->endIndex, role, term);
533 534 535 536 537 538 539 540 541 542 543

  // execute in fsm
  for (int64_t index = pBuf->commitIndex + 1; index <= upperIndex; index++) {
    // get a log entry
    pEntry = syncLogBufferGetOneEntry(pBuf, pNode, index, &inBuf);
    if (pEntry == NULL) {
      goto _out;
    }

    // execute it
    if (!syncUtilUserCommit(pEntry->originalRpcType)) {
544
      sInfo("vgId:%d, commit sync barrier. index: %" PRId64 ", term:%" PRId64 ", type: %s", vgId, pEntry->index,
545
            pEntry->term, TMSG_INFO(pEntry->originalRpcType));
546
    }
547

548
    if (syncLogFsmExecute(pNode, pFsm, role, term, pEntry, 0) != 0) {
549
      sError("vgId:%d, failed to execute sync log entry. index:%" PRId64 ", term:%" PRId64
550 551
             ", role: %d, current term: %" PRId64,
             vgId, pEntry->index, pEntry->term, role, term);
552 553 554 555
      goto _out;
    }
    pBuf->commitIndex = index;

556
    sTrace("vgId:%d, committed index: %" PRId64 ", term: %" PRId64 ", role: %d, current term: %" PRId64 "", pNode->vgId,
557 558 559 560 561 562 563 564 565
           pEntry->index, pEntry->term, role, term);

    if (!inBuf) {
      syncEntryDestroy(pEntry);
      pEntry = NULL;
    }
  }

  // recycle
566
  SyncIndex until = pBuf->commitIndex - (pBuf->size >> 4);
567 568
  for (SyncIndex index = pBuf->startIndex; index < until; index++) {
    SSyncRaftEntry* pEntry = pBuf->entries[(index + pBuf->size) % pBuf->size].pItem;
569
    ASSERT(pEntry != NULL);
570 571 572 573 574
    syncEntryDestroy(pEntry);
    memset(&pBuf->entries[(index + pBuf->size) % pBuf->size], 0, sizeof(pBuf->entries[0]));
    pBuf->startIndex = index + 1;
  }

575
  ret = 0;
576 577
_out:
  // mark as restored if needed
578
  if (!pNode->restoreFinish && pBuf->commitIndex >= pNode->commitIndex && pEntry != NULL &&
S
Shengliang Guan 已提交
579
      pNode->raftStore.currentTerm <= pEntry->term) {
580 581
    pNode->pFsm->FpRestoreFinishCb(pNode->pFsm);
    pNode->restoreFinish = true;
582
    sInfo("vgId:%d, restore finished. log buffer: [%" PRId64 " %" PRId64 " %" PRId64 ", %" PRId64 ")", pNode->vgId,
583 584 585 586 587 588 589 590 591 592 593 594
          pBuf->startIndex, pBuf->commitIndex, pBuf->matchIndex, pBuf->endIndex);
  }

  if (!inBuf) {
    syncEntryDestroy(pEntry);
    pEntry = NULL;
  }
  syncLogBufferValidate(pBuf);
  taosThreadMutexUnlock(&pBuf->mutex);
  return ret;
}

595 596 597
void syncLogReplMgrReset(SSyncLogReplMgr* pMgr) {
  if (pMgr == NULL) return;

598
  ASSERT(pMgr->startIndex >= 0);
599 600 601 602 603 604 605 606 607 608 609 610 611 612 613
  for (SyncIndex index = pMgr->startIndex; index < pMgr->endIndex; index++) {
    memset(&pMgr->states[index % pMgr->size], 0, sizeof(pMgr->states[0]));
  }
  pMgr->startIndex = 0;
  pMgr->matchIndex = 0;
  pMgr->endIndex = 0;
  pMgr->restored = false;
  pMgr->retryBackoff = 0;
}

int32_t syncLogReplMgrRetryOnNeed(SSyncLogReplMgr* pMgr, SSyncNode* pNode) {
  if (pMgr->endIndex <= pMgr->startIndex) {
    return 0;
  }

614 615 616 617 618 619 620 621
  SRaftId* pDestId = &pNode->replicasId[pMgr->peerId];
  if (pMgr->retryBackoff == SYNC_MAX_RETRY_BACKOFF) {
    syncLogReplMgrReset(pMgr);
    sWarn("vgId:%d, reset sync log repl mgr since retry backoff exceeding limit. peer: %" PRIx64, pNode->vgId,
          pDestId->addr);
    return -1;
  }

S
Shengliang Guan 已提交
622 623 624
  int32_t  ret = -1;
  bool     retried = false;
  int64_t  retryWaitMs = syncLogGetRetryBackoffTimeMs(pMgr);
625 626 627 628
  int64_t  nowMs = taosGetMonoTimestampMs();
  int      count = 0;
  int64_t  firstIndex = -1;
  SyncTerm term = -1;
629
  int64_t  batchSize = TMAX(1, pMgr->size >> (4 + pMgr->retryBackoff));
630 631 632

  for (SyncIndex index = pMgr->startIndex; index < pMgr->endIndex; index++) {
    int64_t pos = index % pMgr->size;
633
    ASSERT(!pMgr->states[pos].barrier || (index == pMgr->startIndex || index + 1 == pMgr->endIndex));
634

635 636 637 638
    if (nowMs < pMgr->states[pos].timeMs + retryWaitMs) {
      break;
    }

639
    if (pMgr->states[pos].acked) {
640
      if (pMgr->matchIndex < index && pMgr->states[pos].timeMs + (syncGetRetryMaxWaitMs() << 3) < nowMs) {
641
        syncLogReplMgrReset(pMgr);
642 643
        sWarn("vgId:%d, reset sync log repl mgr since stagnation. index: %" PRId64 ", peer: %" PRIx64, pNode->vgId,
              index, pDestId->addr);
644 645
        goto _out;
      }
646 647 648 649
      continue;
    }

    bool barrier = false;
650
    if (syncLogReplMgrReplicateOneTo(pMgr, pNode, index, &term, pDestId, &barrier) < 0) {
651
      sError("vgId:%d, failed to replicate sync log entry since %s. index: %" PRId64 ", dest: %" PRIx64 "", pNode->vgId,
652 653 654
             terrstr(), index, pDestId->addr);
      goto _out;
    }
655
    ASSERT(barrier == pMgr->states[pos].barrier);
656 657 658
    pMgr->states[pos].timeMs = nowMs;
    pMgr->states[pos].term = term;
    pMgr->states[pos].acked = false;
659

660
    retried = true;
661
    if (firstIndex == -1) firstIndex = index;
662 663 664 665

    if (batchSize <= count++) {
      break;
    }
666 667 668 669 670 671
  }

  ret = 0;
_out:
  if (retried) {
    pMgr->retryBackoff = syncLogGetNextRetryBackoff(pMgr);
672 673 674 675
    SSyncLogBuffer* pBuf = pNode->pLogBuf;
    sInfo("vgId:%d, resend %d sync log entries. dest: %" PRIx64 ", indexes: %" PRId64 " ..., terms: ... %" PRId64
          ", retryWaitMs: %" PRId64 ", mgr: [%" PRId64 " %" PRId64 ", %" PRId64 "), buffer: [%" PRId64 " %" PRId64
          " %" PRId64 ", %" PRId64 ")",
676
          pNode->vgId, count, pDestId->addr, firstIndex, term, retryWaitMs, pMgr->startIndex, pMgr->matchIndex,
677
          pMgr->endIndex, pBuf->startIndex, pBuf->commitIndex, pBuf->matchIndex, pBuf->endIndex);
678 679 680 681
  }
  return ret;
}

682
int32_t syncLogReplMgrProcessReplyAsRecovery(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncAppendEntriesReply* pMsg) {
683 684
  SSyncLogBuffer* pBuf = pNode->pLogBuf;
  SRaftId         destId = pMsg->srcId;
685
  ASSERT(pMgr->restored == false);
686 687

  if (pMgr->endIndex == 0) {
688 689
    ASSERT(pMgr->startIndex == 0);
    ASSERT(pMgr->matchIndex == 0);
690 691
    if (pMsg->matchIndex < 0) {
      pMgr->restored = true;
692
      sInfo("vgId:%d, sync log repl mgr restored. peer: dnode:%d (%" PRIx64 "), mgr: rs(%d) [%" PRId64 " %" PRId64
693
            ", %" PRId64 "), buffer: [%" PRId64 " %" PRId64 " %" PRId64 ", %" PRId64 ")",
694
            pNode->vgId, DID(&destId), destId.addr, pMgr->restored, pMgr->startIndex, pMgr->matchIndex, pMgr->endIndex,
695 696 697 698 699 700 701 702 703 704 705
            pBuf->startIndex, pBuf->commitIndex, pBuf->matchIndex, pBuf->endIndex);
      return 0;
    }
  } else {
    if (pMsg->lastSendIndex < pMgr->startIndex || pMsg->lastSendIndex >= pMgr->endIndex) {
      syncLogReplMgrRetryOnNeed(pMgr, pNode);
      return 0;
    }

    pMgr->states[pMsg->lastSendIndex % pMgr->size].acked = true;

706
    if (pMsg->success && pMsg->matchIndex == pMsg->lastSendIndex) {
707
      pMgr->matchIndex = pMsg->matchIndex;
708
      pMgr->restored = true;
709
      sInfo("vgId:%d, sync log repl mgr restored. peer: dnode:%d (%" PRIx64 "), mgr: rs(%d) [%" PRId64 " %" PRId64
710
            ", %" PRId64 "), buffer: [%" PRId64 " %" PRId64 " %" PRId64 ", %" PRId64 ")",
711
            pNode->vgId, DID(&destId), destId.addr, pMgr->restored, pMgr->startIndex, pMgr->matchIndex, pMgr->endIndex,
712 713 714 715
            pBuf->startIndex, pBuf->commitIndex, pBuf->matchIndex, pBuf->endIndex);
      return 0;
    }

716
    if (pMsg->success == false && pMsg->matchIndex >= pMsg->lastSendIndex) {
717 718
      sWarn("vgId:%d, failed to rollback match index. peer: dnode:%d, match index: %" PRId64 ", last sent: %" PRId64,
            pNode->vgId, DID(&destId), pMsg->matchIndex, pMsg->lastSendIndex);
719
      if (syncNodeStartSnapshot(pNode, &destId) < 0) {
720
        sError("vgId:%d, failed to start snapshot for peer dnode:%d", pNode->vgId, DID(&destId));
721 722
        return -1;
      }
723
      sInfo("vgId:%d, snapshot replication to peer dnode:%d", pNode->vgId, DID(&destId));
724 725
      return 0;
    }
726 727
  }

728 729
  // check last match term
  SyncTerm  term = -1;
730
  SyncIndex firstVer = pNode->pLogStore->syncLogBeginIndex(pNode->pLogStore);
731 732 733 734 735
  SyncIndex index = TMIN(pMsg->matchIndex, pNode->pLogBuf->matchIndex);

  if (pMsg->matchIndex < pNode->pLogBuf->matchIndex) {
    term = syncLogReplMgrGetPrevLogTerm(pMgr, pNode, index + 1);
    if (term < 0 || (term != pMsg->lastMatchTerm && (index + 1 == firstVer || index == firstVer))) {
736
      ASSERT(term >= 0 || terrno == TSDB_CODE_WAL_LOG_NOT_EXIST);
737
      if (syncNodeStartSnapshot(pNode, &destId) < 0) {
S
Shengliang Guan 已提交
738
        sError("vgId:%d, failed to start snapshot for peer dnode:%d", pNode->vgId, DID(&destId));
739
        return -1;
740
      }
S
Shengliang Guan 已提交
741
      sInfo("vgId:%d, snapshot replication to peer dnode:%d", pNode->vgId, DID(&destId));
742 743 744
      return 0;
    }

745
    ASSERT(index + 1 >= firstVer);
746 747 748

    if (term == pMsg->lastMatchTerm) {
      index = index + 1;
749
      ASSERT(index <= pNode->pLogBuf->matchIndex);
750
    } else {
751
      ASSERT(index > firstVer);
752 753 754
    }
  }

755
  // attempt to replicate the raft log at index
756
  (void)syncLogReplMgrReset(pMgr);
757
  return syncLogReplMgrReplicateProbe(pMgr, pNode, index);
758 759 760 761 762 763
}

int32_t syncLogReplMgrProcessHeartbeatReply(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncHeartbeatReply* pMsg) {
  SSyncLogBuffer* pBuf = pNode->pLogBuf;
  taosThreadMutexLock(&pBuf->mutex);
  if (pMsg->startTime != 0 && pMsg->startTime != pMgr->peerStartTime) {
764 765
    sInfo("vgId:%d, reset sync log repl mgr in heartbeat. peer: %" PRIx64 ", start time:%" PRId64 ", old:%" PRId64 "",
          pNode->vgId, pMsg->srcId.addr, pMsg->startTime, pMgr->peerStartTime);
766
    syncLogReplMgrReset(pMgr);
767 768 769 770 771 772 773 774 775 776
    pMgr->peerStartTime = pMsg->startTime;
  }
  taosThreadMutexUnlock(&pBuf->mutex);
  return 0;
}

int32_t syncLogReplMgrProcessReply(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncAppendEntriesReply* pMsg) {
  SSyncLogBuffer* pBuf = pNode->pLogBuf;
  taosThreadMutexLock(&pBuf->mutex);
  if (pMsg->startTime != pMgr->peerStartTime) {
777 778
    sInfo("vgId:%d, reset sync log repl mgr in appendlog reply. peer: %" PRIx64 ", start time:%" PRId64
          ", old:%" PRId64,
779
          pNode->vgId, pMsg->srcId.addr, pMsg->startTime, pMgr->peerStartTime);
780
    syncLogReplMgrReset(pMgr);
781 782 783 784
    pMgr->peerStartTime = pMsg->startTime;
  }

  if (pMgr->restored) {
785
    (void)syncLogReplMgrProcessReplyAsNormal(pMgr, pNode, pMsg);
786
  } else {
787
    (void)syncLogReplMgrProcessReplyAsRecovery(pMgr, pNode, pMsg);
788 789 790 791 792
  }
  taosThreadMutexUnlock(&pBuf->mutex);
  return 0;
}

793
int32_t syncLogReplMgrReplicateOnce(SSyncLogReplMgr* pMgr, SSyncNode* pNode) {
794
  if (pMgr->restored) {
795
    (void)syncLogReplMgrReplicateAttempt(pMgr, pNode);
796
  } else {
797
    (void)syncLogReplMgrReplicateProbe(pMgr, pNode, pNode->pLogBuf->matchIndex);
798 799 800 801
  }
  return 0;
}

802
int32_t syncLogReplMgrReplicateProbe(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncIndex index) {
803 804
  ASSERT(!pMgr->restored);
  ASSERT(pMgr->startIndex >= 0);
805
  int64_t retryMaxWaitMs = syncGetRetryMaxWaitMs();
806 807 808 809 810 811 812 813
  int64_t nowMs = taosGetMonoTimestampMs();

  if (pMgr->endIndex > pMgr->startIndex &&
      nowMs < pMgr->states[pMgr->startIndex % pMgr->size].timeMs + retryMaxWaitMs) {
    return 0;
  }
  (void)syncLogReplMgrReset(pMgr);

S
Shengliang Guan 已提交
814 815 816
  SRaftId* pDestId = &pNode->replicasId[pMgr->peerId];
  bool     barrier = false;
  SyncTerm term = -1;
817
  if (syncLogReplMgrReplicateOneTo(pMgr, pNode, index, &term, pDestId, &barrier) < 0) {
818 819 820 821 822
    sError("vgId:%d, failed to replicate log entry since %s. index: %" PRId64 ", dest: 0x%016" PRIx64 "", pNode->vgId,
           terrstr(), index, pDestId->addr);
    return -1;
  }

823
  ASSERT(index >= 0);
824 825 826 827 828 829 830 831
  pMgr->states[index % pMgr->size].barrier = barrier;
  pMgr->states[index % pMgr->size].timeMs = nowMs;
  pMgr->states[index % pMgr->size].term = term;
  pMgr->states[index % pMgr->size].acked = false;

  pMgr->startIndex = index;
  pMgr->endIndex = index + 1;

832
  SSyncLogBuffer* pBuf = pNode->pLogBuf;
833 834 835 836
  sInfo("vgId:%d, probe peer:%" PRIx64 " with msg of index:%" PRId64 " term: %" PRId64 ". mgr (rs:%d): [%" PRId64
        " %" PRId64 ", %" PRId64 "), buffer: [%" PRId64 " %" PRId64 " %" PRId64 ", %" PRId64 ")",
        pNode->vgId, pDestId->addr, index, term, pMgr->restored, pMgr->startIndex, pMgr->matchIndex, pMgr->endIndex,
        pBuf->startIndex, pBuf->commitIndex, pBuf->matchIndex, pBuf->endIndex);
837 838 839
  return 0;
}

840
int32_t syncLogReplMgrReplicateAttempt(SSyncLogReplMgr* pMgr, SSyncNode* pNode) {
841
  ASSERT(pMgr->restored);
842

S
Shengliang Guan 已提交
843 844 845 846 847
  SRaftId*  pDestId = &pNode->replicasId[pMgr->peerId];
  int32_t   batchSize = TMAX(1, pMgr->size >> (4 + pMgr->retryBackoff));
  int32_t   count = 0;
  int64_t   nowMs = taosGetMonoTimestampMs();
  int64_t   limit = pMgr->size >> 1;
848 849
  SyncTerm  term = -1;
  SyncIndex firstIndex = -1;
850 851

  for (SyncIndex index = pMgr->endIndex; index <= pNode->pLogBuf->matchIndex; index++) {
852
    if (batchSize < count || limit <= index - pMgr->startIndex) {
853 854 855 856 857 858 859 860 861
      break;
    }
    if (pMgr->startIndex + 1 < index && pMgr->states[(index - 1) % pMgr->size].barrier) {
      break;
    }
    int64_t  pos = index % pMgr->size;
    SRaftId* pDestId = &pNode->replicasId[pMgr->peerId];
    bool     barrier = false;
    SyncTerm term = -1;
862
    if (syncLogReplMgrReplicateOneTo(pMgr, pNode, index, &term, pDestId, &barrier) < 0) {
863 864 865 866 867 868 869 870 871
      sError("vgId:%d, failed to replicate log entry since %s. index: %" PRId64 ", dest: 0x%016" PRIx64 "", pNode->vgId,
             terrstr(), index, pDestId->addr);
      return -1;
    }
    pMgr->states[pos].barrier = barrier;
    pMgr->states[pos].timeMs = nowMs;
    pMgr->states[pos].term = term;
    pMgr->states[pos].acked = false;

872 873 874
    if (firstIndex == -1) firstIndex = index;
    count++;

875 876
    pMgr->endIndex = index + 1;
    if (barrier) {
877 878 879 880
      sInfo("vgId:%d, replicated sync barrier to dest: %" PRIx64 ". index: %" PRId64 ", term: %" PRId64
            ", repl mgr: rs(%d) [%" PRId64 " %" PRId64 ", %" PRId64 ")",
            pNode->vgId, pDestId->addr, index, term, pMgr->restored, pMgr->startIndex, pMgr->matchIndex,
            pMgr->endIndex);
881 882 883 884
      break;
    }
  }

885 886
  syncLogReplMgrRetryOnNeed(pMgr, pNode);

887
  SSyncLogBuffer* pBuf = pNode->pLogBuf;
B
Benguang Zhao 已提交
888
  sTrace("vgId:%d, replicated %d msgs to peer: %" PRIx64 ". indexes: %" PRId64 "..., terms: ...%" PRId64
889 890 891 892
         ", mgr: (rs:%d) [%" PRId64 " %" PRId64 ", %" PRId64 "), buffer: [%" PRId64 " %" PRId64 " %" PRId64 ", %" PRId64
         ")",
         pNode->vgId, count, pDestId->addr, firstIndex, term, pMgr->restored, pMgr->startIndex, pMgr->matchIndex,
         pMgr->endIndex, pBuf->startIndex, pBuf->commitIndex, pBuf->matchIndex, pBuf->endIndex);
893 894 895
  return 0;
}

896
int32_t syncLogReplMgrProcessReplyAsNormal(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncAppendEntriesReply* pMsg) {
897
  ASSERT(pMgr->restored == true);
898
  if (pMgr->startIndex <= pMsg->lastSendIndex && pMsg->lastSendIndex < pMgr->endIndex) {
S
Shengliang Guan 已提交
899 900 901 902 903 904 905
    if (pMgr->startIndex < pMgr->matchIndex && pMgr->retryBackoff > 0) {
      int64_t firstSentMs = pMgr->states[pMgr->startIndex % pMgr->size].timeMs;
      int64_t lastSentMs = pMgr->states[(pMgr->endIndex - 1) % pMgr->size].timeMs;
      int64_t timeDiffMs = lastSentMs - firstSentMs;
      if (timeDiffMs > 0 && timeDiffMs < (SYNC_LOG_REPL_RETRY_WAIT_MS << (pMgr->retryBackoff - 1))) {
        pMgr->retryBackoff -= 1;
      }
906
    }
907 908 909 910 911 912 913 914
    pMgr->states[pMsg->lastSendIndex % pMgr->size].acked = true;
    pMgr->matchIndex = TMAX(pMgr->matchIndex, pMsg->matchIndex);
    for (SyncIndex index = pMgr->startIndex; index < pMgr->matchIndex; index++) {
      memset(&pMgr->states[index % pMgr->size], 0, sizeof(pMgr->states[0]));
    }
    pMgr->startIndex = pMgr->matchIndex;
  }

915
  return syncLogReplMgrReplicateAttempt(pMgr, pNode);
916 917 918 919 920 921 922 923 924 925 926
}

SSyncLogReplMgr* syncLogReplMgrCreate() {
  SSyncLogReplMgr* pMgr = taosMemoryCalloc(1, sizeof(SSyncLogReplMgr));
  if (pMgr == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return NULL;
  }

  pMgr->size = sizeof(pMgr->states) / sizeof(pMgr->states[0]);

927
  ASSERT(pMgr->size == TSDB_SYNC_LOG_BUFFER_SIZE);
928 929 930 931 932 933 934 935 936 937 938 939 940 941 942 943 944 945

  return pMgr;

_err:
  taosMemoryFree(pMgr);
  return NULL;
}

void syncLogReplMgrDestroy(SSyncLogReplMgr* pMgr) {
  if (pMgr == NULL) {
    return;
  }
  (void)taosMemoryFree(pMgr);
  return;
}

int32_t syncNodeLogReplMgrInit(SSyncNode* pNode) {
  for (int i = 0; i < TSDB_MAX_REPLICA; i++) {
946
    ASSERT(pNode->logReplMgrs[i] == NULL);
947
    pNode->logReplMgrs[i] = syncLogReplMgrCreate();
948 949 950 951
    if (pNode->logReplMgrs[i] == NULL) {
      terrno = TSDB_CODE_OUT_OF_MEMORY;
      return -1;
    }
952 953 954 955 956 957 958 959 960 961 962 963 964 965 966 967 968 969 970 971 972
    pNode->logReplMgrs[i]->peerId = i;
  }
  return 0;
}

void syncNodeLogReplMgrDestroy(SSyncNode* pNode) {
  for (int i = 0; i < TSDB_MAX_REPLICA; i++) {
    syncLogReplMgrDestroy(pNode->logReplMgrs[i]);
    pNode->logReplMgrs[i] = NULL;
  }
}

SSyncLogBuffer* syncLogBufferCreate() {
  SSyncLogBuffer* pBuf = taosMemoryCalloc(1, sizeof(SSyncLogBuffer));
  if (pBuf == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return NULL;
  }

  pBuf->size = sizeof(pBuf->entries) / sizeof(pBuf->entries[0]);

973
  ASSERT(pBuf->size == TSDB_SYNC_LOG_BUFFER_SIZE);
974

975 976 977 978 979 980 981 982 983 984 985 986 987
  if (taosThreadMutexAttrInit(&pBuf->attr) < 0) {
    sError("failed to init log buffer mutexattr due to %s", strerror(errno));
    terrno = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  if (taosThreadMutexAttrSetType(&pBuf->attr, PTHREAD_MUTEX_RECURSIVE) < 0) {
    sError("failed to set log buffer mutexattr type due to %s", strerror(errno));
    terrno = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }

  if (taosThreadMutexInit(&pBuf->mutex, &pBuf->attr) < 0) {
988 989 990 991
    sError("failed to init log buffer mutex due to %s", strerror(errno));
    terrno = TAOS_SYSTEM_ERROR(errno);
    goto _err;
  }
992

993 994 995 996 997 998 999
  return pBuf;

_err:
  taosMemoryFree(pBuf);
  return NULL;
}

1000 1001 1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 1012
void syncLogBufferClear(SSyncLogBuffer* pBuf) {
  taosThreadMutexLock(&pBuf->mutex);
  for (SyncIndex index = pBuf->startIndex; index < pBuf->endIndex; index++) {
    SSyncRaftEntry* pEntry = pBuf->entries[(index + pBuf->size) % pBuf->size].pItem;
    if (pEntry == NULL) continue;
    syncEntryDestroy(pEntry);
    pEntry = NULL;
    memset(&pBuf->entries[(index + pBuf->size) % pBuf->size], 0, sizeof(pBuf->entries[0]));
  }
  pBuf->startIndex = pBuf->commitIndex = pBuf->matchIndex = pBuf->endIndex = 0;
  taosThreadMutexUnlock(&pBuf->mutex);
}

1013 1014 1015 1016
void syncLogBufferDestroy(SSyncLogBuffer* pBuf) {
  if (pBuf == NULL) {
    return;
  }
1017
  syncLogBufferClear(pBuf);
1018
  (void)taosThreadMutexDestroy(&pBuf->mutex);
1019
  (void)taosThreadMutexAttrDestroy(&pBuf->attr);
1020 1021 1022 1023
  (void)taosMemoryFree(pBuf);
  return;
}

1024
int32_t syncLogBufferRollback(SSyncLogBuffer* pBuf, SSyncNode* pNode, SyncIndex toIndex) {
1025
  ASSERT(pBuf->commitIndex < toIndex && toIndex <= pBuf->endIndex);
1026

1027 1028 1029 1030
  if (toIndex == pBuf->endIndex) {
    return 0;
  }

1031 1032 1033 1034
  sInfo("vgId:%d, rollback sync log buffer. toindex: %" PRId64 ", buffer: [%" PRId64 " %" PRId64 " %" PRId64
        ", %" PRId64 ")",
        pNode->vgId, toIndex, pBuf->startIndex, pBuf->commitIndex, pBuf->matchIndex, pBuf->endIndex);

1035
  // trunc buffer
1036 1037 1038 1039
  SyncIndex index = pBuf->endIndex - 1;
  while (index >= toIndex) {
    SSyncRaftEntry* pEntry = pBuf->entries[index % pBuf->size].pItem;
    if (pEntry != NULL) {
1040
      (void)syncEntryDestroy(pEntry);
1041 1042 1043 1044 1045 1046 1047
      pEntry = NULL;
      memset(&pBuf->entries[index % pBuf->size], 0, sizeof(pBuf->entries[0]));
    }
    index--;
  }
  pBuf->endIndex = toIndex;
  pBuf->matchIndex = TMIN(pBuf->matchIndex, index);
1048
  ASSERT(index + 1 == toIndex);
1049 1050 1051 1052 1053 1054 1055 1056

  // trunc wal
  SyncIndex lastVer = pNode->pLogStore->syncLogLastIndex(pNode->pLogStore);
  if (lastVer >= toIndex && pNode->pLogStore->syncLogTruncate(pNode->pLogStore, toIndex) < 0) {
    sError("vgId:%d, failed to truncate log store since %s. from index:%" PRId64 "", pNode->vgId, terrstr(), toIndex);
    return -1;
  }
  lastVer = pNode->pLogStore->syncLogLastIndex(pNode->pLogStore);
1057
  ASSERT(toIndex == lastVer + 1);
1058

1059 1060 1061 1062 1063 1064 1065 1066 1067 1068
  // refill buffer on need
  if (toIndex <= pBuf->startIndex) {
    int32_t ret = syncLogBufferInitWithoutLock(pBuf, pNode);
    if (ret < 0) {
      sError("vgId:%d, failed to refill sync log buffer since %s", pNode->vgId, terrstr());
      return -1;
    }
  }

  ASSERT(pBuf->endIndex == toIndex);
1069
  syncLogBufferValidate(pBuf);
1070 1071 1072 1073 1074 1075
  return 0;
}

int32_t syncLogBufferReset(SSyncLogBuffer* pBuf, SSyncNode* pNode) {
  taosThreadMutexLock(&pBuf->mutex);
  SyncIndex lastVer = pNode->pLogStore->syncLogLastIndex(pNode->pLogStore);
1076
  ASSERT(lastVer == pBuf->matchIndex);
1077 1078
  SyncIndex index = pBuf->endIndex - 1;

1079
  (void)syncLogBufferRollback(pBuf, pNode, pBuf->matchIndex + 1);
1080

1081
  sInfo("vgId:%d, reset sync log buffer. buffer: [%" PRId64 " %" PRId64 " %" PRId64 ", %" PRId64 ")", pNode->vgId,
1082 1083 1084 1085 1086 1087 1088
        pBuf->startIndex, pBuf->commitIndex, pBuf->matchIndex, pBuf->endIndex);

  pBuf->endIndex = pBuf->matchIndex + 1;

  // reset repl mgr
  for (int i = 0; i < pNode->replicaNum; i++) {
    SSyncLogReplMgr* pMgr = pNode->logReplMgrs[i];
1089
    syncLogReplMgrReset(pMgr);
1090 1091 1092 1093 1094 1095 1096 1097 1098 1099 1100 1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111
  }
  taosThreadMutexUnlock(&pBuf->mutex);
  return 0;
}

SSyncRaftEntry* syncLogBufferGetOneEntry(SSyncLogBuffer* pBuf, SSyncNode* pNode, SyncIndex index, bool* pInBuf) {
  SSyncRaftEntry* pEntry = NULL;
  if (index >= pBuf->endIndex) {
    return NULL;
  }
  if (index > pBuf->startIndex) {  // startIndex might be dummy
    *pInBuf = true;
    pEntry = pBuf->entries[index % pBuf->size].pItem;
  } else {
    *pInBuf = false;
    if (pNode->pLogStore->syncLogGetEntry(pNode->pLogStore, index, &pEntry) < 0) {
      sError("vgId:%d, failed to get log entry since %s. index:%" PRId64 "", pNode->vgId, terrstr(), index);
    }
  }
  return pEntry;
}

1112 1113
int32_t syncLogReplMgrReplicateOneTo(SSyncLogReplMgr* pMgr, SSyncNode* pNode, SyncIndex index, SyncTerm* pTerm,
                                     SRaftId* pDestId, bool* pBarrier) {
1114 1115 1116 1117 1118 1119 1120 1121 1122
  SSyncRaftEntry* pEntry = NULL;
  SRpcMsg         msgOut = {0};
  bool            inBuf = false;
  SyncTerm        prevLogTerm = -1;
  SSyncLogBuffer* pBuf = pNode->pLogBuf;

  pEntry = syncLogBufferGetOneEntry(pBuf, pNode, index, &inBuf);
  if (pEntry == NULL) {
    sError("vgId:%d, failed to get raft entry for index: %" PRId64 "", pNode->vgId, index);
1123 1124 1125
    if (terrno == TSDB_CODE_WAL_LOG_NOT_EXIST) {
      SSyncLogReplMgr* pMgr = syncNodeGetLogReplMgr(pNode, pDestId);
      if (pMgr) {
1126
        sInfo("vgId:%d, reset sync log repl mgr of peer: %" PRIx64 " since %s. index: %" PRId64, pNode->vgId,
1127
              pDestId->addr, terrstr(), index);
1128 1129 1130
        (void)syncLogReplMgrReset(pMgr);
      }
    }
1131 1132 1133 1134 1135 1136 1137 1138 1139 1140 1141
    goto _err;
  }
  *pBarrier = syncLogIsReplicationBarrier(pEntry);

  prevLogTerm = syncLogReplMgrGetPrevLogTerm(pMgr, pNode, index);
  if (prevLogTerm < 0) {
    sError("vgId:%d, failed to get prev log term since %s. index: %" PRId64 "", pNode->vgId, terrstr(), index);
    goto _err;
  }
  if (pTerm) *pTerm = pEntry->term;

1142
  int32_t code = syncBuildAppendEntriesFromRaftEntry(pNode, pEntry, prevLogTerm, &msgOut);
1143 1144 1145 1146 1147 1148 1149
  if (code < 0) {
    sError("vgId:%d, failed to get append entries for index:%" PRId64 "", pNode->vgId, index);
    goto _err;
  }

  (void)syncNodeSendAppendEntries(pNode, pDestId, &msgOut);

1150 1151
  sTrace("vgId:%d, replicate one msg index: %" PRId64 " term: %" PRId64 " prevterm: %" PRId64 " to dest: 0x%016" PRIx64,
         pNode->vgId, pEntry->index, pEntry->term, prevLogTerm, pDestId->addr);
1152 1153 1154 1155 1156 1157 1158 1159 1160 1161 1162 1163 1164 1165 1166 1167

  if (!inBuf) {
    syncEntryDestroy(pEntry);
    pEntry = NULL;
  }
  return 0;

_err:
  rpcFreeCont(msgOut.pCont);
  msgOut.pCont = NULL;
  if (!inBuf) {
    syncEntryDestroy(pEntry);
    pEntry = NULL;
  }
  return -1;
}