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

#include "syncRaftLog.h"
M
Minghao Li 已提交
17
#include "syncRaftCfg.h"
18
#include "syncRaftStore.h"
M
Minghao Li 已提交
19

20 21 22 23
//-------------------------------
// log[m .. n]

// public function
24
static int32_t   raftLogRestoreFromSnapshot(struct SSyncLogStore* pLogStore, SyncIndex snapshotIndex);
M
Minghao Li 已提交
25 26
static SyncIndex raftLogBeginIndex(struct SSyncLogStore* pLogStore);
static SyncIndex raftLogEndIndex(struct SSyncLogStore* pLogStore);
M
Minghao Li 已提交
27
static SyncIndex raftLogWriteIndex(struct SSyncLogStore* pLogStore);
M
Minghao Li 已提交
28 29 30 31 32 33 34
static bool      raftLogIsEmpty(struct SSyncLogStore* pLogStore);
static int32_t   raftLogEntryCount(struct SSyncLogStore* pLogStore);
static SyncIndex raftLogLastIndex(struct SSyncLogStore* pLogStore);
static SyncTerm  raftLogLastTerm(struct SSyncLogStore* pLogStore);
static int32_t   raftLogAppendEntry(struct SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry);
static int32_t   raftLogGetEntry(struct SSyncLogStore* pLogStore, SyncIndex index, SSyncRaftEntry** ppEntry);
static int32_t   raftLogTruncate(struct SSyncLogStore* pLogStore, SyncIndex fromIndex);
35
static bool      raftLogExist(struct SSyncLogStore* pLogStore, SyncIndex index);
M
Minghao Li 已提交
36

37
// private function
M
Minghao Li 已提交
38 39 40
static int32_t raftLogGetLastEntry(SSyncLogStore* pLogStore, SSyncRaftEntry** ppLastEntry);

//-------------------------------
41
// log[0 .. n]
42
static SSyncRaftEntry* logStoreGetLastEntry(SSyncLogStore* pLogStore);
M
Minghao Li 已提交
43 44 45 46 47 48 49
static SyncIndex       logStoreLastIndex(SSyncLogStore* pLogStore);
static SyncTerm        logStoreLastTerm(SSyncLogStore* pLogStore);
static SSyncRaftEntry* logStoreGetEntry(SSyncLogStore* pLogStore, SyncIndex index);
static int32_t         logStoreAppendEntry(SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry);
static int32_t         logStoreTruncate(SSyncLogStore* pLogStore, SyncIndex fromIndex);
static int32_t         logStoreUpdateCommitIndex(SSyncLogStore* pLogStore, SyncIndex index);
static SyncIndex       logStoreGetCommitIndex(SSyncLogStore* pLogStore);
50

51 52 53 54 55
//-------------------------------
SSyncLogStore* logStoreCreate(SSyncNode* pSyncNode) {
  SSyncLogStore* pLogStore = taosMemoryMalloc(sizeof(SSyncLogStore));
  ASSERT(pLogStore != NULL);

56 57 58 59 60 61 62 63 64
  pLogStore->pCache = taosLRUCacheInit(10 * 1024 * 1024, 1, .5);
  if (pLogStore->pCache == NULL) {
    terrno = TSDB_CODE_WAL_OUT_OF_MEMORY;
    taosMemoryFree(pLogStore);
    return NULL;
  }

  taosLRUCacheSetStrictCapacity(pLogStore->pCache, false);

65 66 67 68 69 70 71 72 73
  pLogStore->data = taosMemoryMalloc(sizeof(SSyncLogStoreData));
  ASSERT(pLogStore->data != NULL);

  SSyncLogStoreData* pData = pLogStore->data;
  pData->pSyncNode = pSyncNode;
  pData->pWal = pSyncNode->pWal;
  ASSERT(pData->pWal != NULL);

  taosThreadMutexInit(&(pData->mutex), NULL);
L
Liu Jicong 已提交
74
  pData->pWalHandle = walOpenReader(pData->pWal, NULL);
75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95
  ASSERT(pData->pWalHandle != NULL);

  pLogStore->appendEntry = logStoreAppendEntry;
  pLogStore->getEntry = logStoreGetEntry;
  pLogStore->truncate = logStoreTruncate;
  pLogStore->getLastIndex = logStoreLastIndex;
  pLogStore->getLastTerm = logStoreLastTerm;
  pLogStore->updateCommitIndex = logStoreUpdateCommitIndex;
  pLogStore->getCommitIndex = logStoreGetCommitIndex;

  pLogStore->syncLogRestoreFromSnapshot = raftLogRestoreFromSnapshot;
  pLogStore->syncLogBeginIndex = raftLogBeginIndex;
  pLogStore->syncLogEndIndex = raftLogEndIndex;
  pLogStore->syncLogIsEmpty = raftLogIsEmpty;
  pLogStore->syncLogEntryCount = raftLogEntryCount;
  pLogStore->syncLogLastIndex = raftLogLastIndex;
  pLogStore->syncLogLastTerm = raftLogLastTerm;
  pLogStore->syncLogAppendEntry = raftLogAppendEntry;
  pLogStore->syncLogGetEntry = raftLogGetEntry;
  pLogStore->syncLogTruncate = raftLogTruncate;
  pLogStore->syncLogWriteIndex = raftLogWriteIndex;
96
  pLogStore->syncLogExist = raftLogExist;
97 98 99 100 101 102 103 104 105 106

  return pLogStore;
}

void logStoreDestory(SSyncLogStore* pLogStore) {
  if (pLogStore != NULL) {
    SSyncLogStoreData* pData = pLogStore->data;

    taosThreadMutexLock(&(pData->mutex));
    if (pData->pWalHandle != NULL) {
L
Liu Jicong 已提交
107
      walCloseReader(pData->pWalHandle);
108 109 110 111 112 113
      pData->pWalHandle = NULL;
    }
    taosThreadMutexUnlock(&(pData->mutex));
    taosThreadMutexDestroy(&(pData->mutex));

    taosMemoryFree(pLogStore->data);
114 115 116 117

    taosLRUCacheEraseUnrefEntries(pLogStore->pCache);
    taosLRUCacheCleanup(pLogStore->pCache);

118 119 120 121 122 123
    taosMemoryFree(pLogStore);
  }
}

//-------------------------------
// log[m .. n]
124 125
static int32_t raftLogRestoreFromSnapshot(struct SSyncLogStore* pLogStore, SyncIndex snapshotIndex) {
  ASSERT(snapshotIndex >= 0);
M
Minghao Li 已提交
126 127 128

  SSyncLogStoreData* pData = pLogStore->data;
  SWal*              pWal = pData->pWal;
129 130 131 132 133 134
  int32_t            code = walRestoreFromSnapshot(pWal, snapshotIndex);
  if (code != 0) {
    int32_t     err = terrno;
    const char* errStr = tstrerror(err);
    int32_t     sysErr = errno;
    const char* sysErrStr = strerror(errno);
135

136 137
    char logBuf[128];
    snprintf(logBuf, sizeof(logBuf),
L
Liu Jicong 已提交
138 139
             "wal restore from snapshot error, index:%" PRId64 ", err:%d %X, msg:%s, syserr:%d, sysmsg:%s",
             snapshotIndex, err, err, errStr, sysErr, sysErrStr);
140
    syncNodeErrorLog(pData->pSyncNode, logBuf);
141

142
    return -1;
143
  }
144 145 146

  return 0;
}
M
Minghao Li 已提交
147 148 149 150

static SyncIndex raftLogBeginIndex(struct SSyncLogStore* pLogStore) {
  SSyncLogStoreData* pData = pLogStore->data;
  SWal*              pWal = pData->pWal;
151 152
  SyncIndex          firstVer = walGetFirstVer(pWal);
  return firstVer;
M
Minghao Li 已提交
153 154 155 156 157
}

static SyncIndex raftLogEndIndex(struct SSyncLogStore* pLogStore) { return raftLogLastIndex(pLogStore); }

static bool raftLogIsEmpty(struct SSyncLogStore* pLogStore) {
158 159 160
  SSyncLogStoreData* pData = pLogStore->data;
  SWal*              pWal = pData->pWal;
  return walIsEmpty(pWal);
M
Minghao Li 已提交
161 162 163 164 165
}

static int32_t raftLogEntryCount(struct SSyncLogStore* pLogStore) {
  SyncIndex beginIndex = raftLogBeginIndex(pLogStore);
  SyncIndex endIndex = raftLogEndIndex(pLogStore);
M
Minghao Li 已提交
166
  int32_t   count = endIndex - beginIndex + 1;
M
Minghao Li 已提交
167 168 169 170
  return count > 0 ? count : 0;
}

static SyncIndex raftLogLastIndex(struct SSyncLogStore* pLogStore) {
M
Minghao Li 已提交
171
  SyncIndex          lastIndex;
M
Minghao Li 已提交
172 173 174
  SSyncLogStoreData* pData = pLogStore->data;
  SWal*              pWal = pData->pWal;
  SyncIndex          lastVer = walGetLastVer(pWal);
M
Minghao Li 已提交
175

176
  return lastVer;
M
Minghao Li 已提交
177 178 179 180 181 182 183
}

static SyncIndex raftLogWriteIndex(struct SSyncLogStore* pLogStore) {
  SSyncLogStoreData* pData = pLogStore->data;
  SWal*              pWal = pData->pWal;
  SyncIndex          lastVer = walGetLastVer(pWal);
  return lastVer + 1;
M
Minghao Li 已提交
184 185
}

186 187 188 189 190 191 192
static bool raftLogExist(struct SSyncLogStore* pLogStore, SyncIndex index) {
  SSyncLogStoreData* pData = pLogStore->data;
  SWal*              pWal = pData->pWal;
  bool               b = walLogExist(pWal, index);
  return b;
}

193 194 195
// if success, return last term
// if not log, return 0
// if error, return SYNC_TERM_INVALID
196 197 198 199 200 201 202 203
static SyncTerm raftLogLastTerm(struct SSyncLogStore* pLogStore) {
  SSyncLogStoreData* pData = pLogStore->data;
  SWal*              pWal = pData->pWal;
  if (walIsEmpty(pWal)) {
    return 0;
  } else {
    SSyncRaftEntry* pLastEntry;
    int32_t         code = raftLogGetLastEntry(pLogStore, &pLastEntry);
204 205 206 207 208 209 210
    if (code == 0 && pLastEntry != NULL) {
      SyncTerm lastTerm = pLastEntry->term;
      taosMemoryFree(pLastEntry);
      return lastTerm;
    } else {
      return SYNC_TERM_INVALID;
    }
211 212
  }

213 214
  // can not be here!
  return SYNC_TERM_INVALID;
215 216
}

M
Minghao Li 已提交
217 218 219 220 221
static int32_t raftLogAppendEntry(struct SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry) {
  SSyncLogStoreData* pData = pLogStore->data;
  SWal*              pWal = pData->pWal;

  SyncIndex    index = 0;
S
Shengliang Guan 已提交
222
  SWalSyncInfo syncMeta = {0};
M
Minghao Li 已提交
223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244
  syncMeta.isWeek = pEntry->isWeak;
  syncMeta.seqNum = pEntry->seqNum;
  syncMeta.term = pEntry->term;
  index = walAppendLog(pWal, pEntry->originalRpcType, syncMeta, pEntry->data, pEntry->dataLen);
  if (index < 0) {
    int32_t     err = terrno;
    const char* errStr = tstrerror(err);
    int32_t     sysErr = errno;
    const char* sysErrStr = strerror(errno);

    char logBuf[128];
    snprintf(logBuf, sizeof(logBuf), "wal write error, index:%" PRId64 ", err:%d %X, msg:%s, syserr:%d, sysmsg:%s",
             pEntry->index, err, err, errStr, sysErr, sysErrStr);
    syncNodeErrorLog(pData->pSyncNode, logBuf);

    ASSERT(0);
    return -1;
  }
  pEntry->index = index;

  do {
    char eventLog[128];
S
Shengliang Guan 已提交
245 246
    snprintf(eventLog, sizeof(eventLog), "write index:%" PRId64 ", type:%s, origin type:%s", pEntry->index,
             TMSG_INFO(pEntry->msgType), TMSG_INFO(pEntry->originalRpcType));
M
Minghao Li 已提交
247 248 249 250 251 252
    syncNodeEventLog(pData->pSyncNode, eventLog);
  } while (0);

  return 0;
}

253 254 255
// entry found, return 0
// entry not found, return -1, terrno = TSDB_CODE_WAL_LOG_NOT_EXIST
// other error, return -1
256 257 258
static int32_t raftLogGetEntry(struct SSyncLogStore* pLogStore, SyncIndex index, SSyncRaftEntry** ppEntry) {
  SSyncLogStoreData* pData = pLogStore->data;
  SWal*              pWal = pData->pWal;
259
  int32_t            code = 0;
260 261 262

  *ppEntry = NULL;

263
  // SWalReadHandle* pWalHandle = walOpenReadHandle(pWal);
L
Liu Jicong 已提交
264
  SWalReader* pWalHandle = pData->pWalHandle;
265
  if (pWalHandle == NULL) {
266
    terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
267 268 269
    return -1;
  }

270 271
  taosThreadMutexLock(&(pData->mutex));

L
Liu Jicong 已提交
272
  code = walReadVer(pWalHandle, index);
273
  // code = walReadVerCached(pWalHandle, index);
274 275 276
  if (code != 0) {
    int32_t     err = terrno;
    const char* errStr = tstrerror(err);
277 278
    int32_t     sysErr = errno;
    const char* sysErrStr = strerror(errno);
279 280 281

    do {
      char logBuf[128];
L
Liu Jicong 已提交
282 283
      snprintf(logBuf, sizeof(logBuf), "wal read error, index:%" PRId64 ", err:%d %X, msg:%s, syserr:%d, sysmsg:%s",
               index, err, err, errStr, sysErr, sysErrStr);
284
      if (terrno == TSDB_CODE_WAL_LOG_NOT_EXIST) {
285
        // syncNodeEventLog(pData->pSyncNode, logBuf);
286 287 288 289
      } else {
        syncNodeErrorLog(pData->pSyncNode, logBuf);
      }
    } while (0);
290

291 292 293 294 295
    /*
        int32_t saveErr = terrno;
        walCloseReadHandle(pWalHandle);
        terrno = saveErr;
    */
M
Minghao Li 已提交
296

297
    taosThreadMutexUnlock(&(pData->mutex));
298 299 300 301 302 303 304 305 306 307 308 309 310 311
    return code;
  }

  *ppEntry = syncEntryBuild(pWalHandle->pHead->head.bodyLen);
  ASSERT(*ppEntry != NULL);
  (*ppEntry)->msgType = TDMT_SYNC_CLIENT_REQUEST;
  (*ppEntry)->originalRpcType = pWalHandle->pHead->head.msgType;
  (*ppEntry)->seqNum = pWalHandle->pHead->head.syncMeta.seqNum;
  (*ppEntry)->isWeak = pWalHandle->pHead->head.syncMeta.isWeek;
  (*ppEntry)->term = pWalHandle->pHead->head.syncMeta.term;
  (*ppEntry)->index = index;
  ASSERT((*ppEntry)->dataLen == pWalHandle->pHead->head.bodyLen);
  memcpy((*ppEntry)->data, pWalHandle->pHead->head.body, pWalHandle->pHead->head.bodyLen);

312 313 314 315 316
  /*
    int32_t saveErr = terrno;
    walCloseReadHandle(pWalHandle);
    terrno = saveErr;
  */
317

318
  taosThreadMutexUnlock(&(pData->mutex));
319 320
  return code;
}
M
Minghao Li 已提交
321

322
// truncate semantic
M
Minghao Li 已提交
323 324 325
static int32_t raftLogTruncate(struct SSyncLogStore* pLogStore, SyncIndex fromIndex) {
  SSyncLogStoreData* pData = pLogStore->data;
  SWal*              pWal = pData->pWal;
326 327 328 329 330 331 332 333

  // need not truncate
  SyncIndex wallastVer = walGetLastVer(pWal);
  if (fromIndex > wallastVer) {
    return 0;
  }

  int32_t code = walRollback(pWal, fromIndex);
M
Minghao Li 已提交
334 335 336
  if (code != 0) {
    int32_t     err = terrno;
    const char* errStr = tstrerror(err);
337 338
    int32_t     sysErr = errno;
    const char* sysErrStr = strerror(errno);
S
Shengliang Guan 已提交
339
    sError("vgId:%d, wal truncate error, from-index:%" PRId64 ", err:%d %X, msg:%s, syserr:%d, sysmsg:%s",
340 341
           pData->pSyncNode->vgId, fromIndex, err, err, errStr, sysErr, sysErrStr);

M
Minghao Li 已提交
342 343
    ASSERT(0);
  }
M
Minghao Li 已提交
344 345 346 347

  // event log
  do {
    char logBuf[128];
348
    snprintf(logBuf, sizeof(logBuf), "log truncate, from-index:%" PRId64, fromIndex);
M
Minghao Li 已提交
349 350 351
    syncNodeEventLog(pData->pSyncNode, logBuf);
  } while (0);

M
Minghao Li 已提交
352 353 354
  return code;
}

355 356 357
// entry found, return 0
// entry not found, return -1, terrno = TSDB_CODE_WAL_LOG_NOT_EXIST
// other error, return -1
358 359 360 361 362 363 364 365 366 367 368
static int32_t raftLogGetLastEntry(SSyncLogStore* pLogStore, SSyncRaftEntry** ppLastEntry) {
  SSyncLogStoreData* pData = pLogStore->data;
  SWal*              pWal = pData->pWal;
  ASSERT(ppLastEntry != NULL);

  *ppLastEntry = NULL;
  if (walIsEmpty(pWal)) {
    terrno = TSDB_CODE_WAL_LOG_NOT_EXIST;
    return -1;
  } else {
    SyncIndex lastIndex = raftLogLastIndex(pLogStore);
369 370
    ASSERT(lastIndex >= SYNC_INDEX_BEGIN);
    int32_t code = raftLogGetEntry(pLogStore, lastIndex, ppLastEntry);
371 372 373 374 375 376
    return code;
  }

  return -1;
}

M
Minghao Li 已提交
377
//-------------------------------
378
// log[0 .. n]
M
Minghao Li 已提交
379 380 381 382 383 384

int32_t logStoreAppendEntry(SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry) {
  SSyncLogStoreData* pData = pLogStore->data;
  SWal*              pWal = pData->pWal;

  SyncIndex    index = 0;
S
Shengliang Guan 已提交
385
  SWalSyncInfo syncMeta = {0};
M
Minghao Li 已提交
386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408
  syncMeta.isWeek = pEntry->isWeak;
  syncMeta.seqNum = pEntry->seqNum;
  syncMeta.term = pEntry->term;

  index = walAppendLog(pWal, pEntry->originalRpcType, syncMeta, pEntry->data, pEntry->dataLen);
  if (index < 0) {
    int32_t     err = terrno;
    const char* errStr = tstrerror(err);
    int32_t     sysErr = errno;
    const char* sysErrStr = strerror(errno);

    char logBuf[128];
    snprintf(logBuf, sizeof(logBuf), "wal write error, index:%" PRId64 ", err:%d %X, msg:%s, syserr:%d, sysmsg:%s",
             pEntry->index, err, err, errStr, sysErr, sysErrStr);
    syncNodeErrorLog(pData->pSyncNode, logBuf);

    ASSERT(0);
    return -1;
  }
  pEntry->index = index;

  do {
    char eventLog[128];
S
Shengliang Guan 已提交
409 410
    snprintf(eventLog, sizeof(eventLog), "write2 index:%" PRId64 ", type:%s, origin type:%s", pEntry->index,
             TMSG_INFO(pEntry->msgType), TMSG_INFO(pEntry->originalRpcType));
M
Minghao Li 已提交
411 412 413 414 415
    syncNodeEventLog(pData->pSyncNode, eventLog);
  } while (0);

  return 0;
}
M
Minghao Li 已提交
416

M
Minghao Li 已提交
417 418 419 420
SSyncRaftEntry* logStoreGetEntry(SSyncLogStore* pLogStore, SyncIndex index) {
  SSyncLogStoreData* pData = pLogStore->data;
  SWal*              pWal = pData->pWal;

M
Minghao Li 已提交
421
  if (index >= SYNC_INDEX_BEGIN && index <= logStoreLastIndex(pLogStore)) {
422 423
    taosThreadMutexLock(&(pData->mutex));

424
    // SWalReadHandle* pWalHandle = walOpenReadHandle(pWal);
L
Liu Jicong 已提交
425
    SWalReader* pWalHandle = pData->pWalHandle;
M
Minghao Li 已提交
426 427
    ASSERT(pWalHandle != NULL);

L
Liu Jicong 已提交
428
    int32_t code = walReadVer(pWalHandle, index);
429
    // int32_t code = walReadVerCached(pWalHandle, index);
M
Minghao Li 已提交
430
    if (code != 0) {
M
Minghao Li 已提交
431 432
      int32_t     err = terrno;
      const char* errStr = tstrerror(err);
433 434
      int32_t     sysErr = errno;
      const char* sysErrStr = strerror(errno);
435 436 437

      do {
        char logBuf[128];
L
Liu Jicong 已提交
438 439
        snprintf(logBuf, sizeof(logBuf), "wal read error, index:%" PRId64 ", err:%d %X, msg:%s, syserr:%d, sysmsg:%s",
                 index, err, err, errStr, sysErr, sysErrStr);
440
        if (terrno == TSDB_CODE_WAL_LOG_NOT_EXIST) {
441
          // syncNodeEventLog(pData->pSyncNode, logBuf);
442 443 444 445
        } else {
          syncNodeErrorLog(pData->pSyncNode, logBuf);
        }
      } while (0);
446

M
Minghao Li 已提交
447
      ASSERT(0);
M
Minghao Li 已提交
448
    }
M
Minghao Li 已提交
449

L
Liu Jicong 已提交
450
    SSyncRaftEntry* pEntry = syncEntryBuild(pWalHandle->pHead->head.bodyLen);
M
Minghao Li 已提交
451
    ASSERT(pEntry != NULL);
M
Minghao Li 已提交
452

453
    pEntry->msgType = TDMT_SYNC_CLIENT_REQUEST;
M
Minghao Li 已提交
454 455 456 457 458
    pEntry->originalRpcType = pWalHandle->pHead->head.msgType;
    pEntry->seqNum = pWalHandle->pHead->head.syncMeta.seqNum;
    pEntry->isWeak = pWalHandle->pHead->head.syncMeta.isWeek;
    pEntry->term = pWalHandle->pHead->head.syncMeta.term;
    pEntry->index = index;
M
Minghao Li 已提交
459
    ASSERT(pEntry->dataLen == pWalHandle->pHead->head.bodyLen);
L
Liu Jicong 已提交
460
    memcpy(pEntry->data, pWalHandle->pHead->head.body, pWalHandle->pHead->head.bodyLen);
M
Minghao Li 已提交
461

462 463 464 465 466
    /*
        int32_t saveErr = terrno;
        walCloseReadHandle(pWalHandle);
        terrno = saveErr;
    */
M
Minghao Li 已提交
467

468
    taosThreadMutexUnlock(&(pData->mutex));
M
Minghao Li 已提交
469
    return pEntry;
M
Minghao Li 已提交
470

M
Minghao Li 已提交
471 472 473
  } else {
    return NULL;
  }
M
Minghao Li 已提交
474
}
M
Minghao Li 已提交
475

M
Minghao Li 已提交
476 477 478
int32_t logStoreTruncate(SSyncLogStore* pLogStore, SyncIndex fromIndex) {
  SSyncLogStoreData* pData = pLogStore->data;
  SWal*              pWal = pData->pWal;
M
Minghao Li 已提交
479
  // ASSERT(walRollback(pWal, fromIndex) == 0);
M
Minghao Li 已提交
480 481
  int32_t code = walRollback(pWal, fromIndex);
  if (code != 0) {
M
Minghao Li 已提交
482 483
    int32_t     err = terrno;
    const char* errStr = tstrerror(err);
484 485
    int32_t     sysErr = errno;
    const char* sysErrStr = strerror(errno);
S
Shengliang Guan 已提交
486
    sError("vgId:%d, wal truncate error, from-index:%" PRId64 ", err:%d %X, msg:%s, syserr:%d, sysmsg:%s",
487 488
           pData->pSyncNode->vgId, fromIndex, err, err, errStr, sysErr, sysErrStr);

M
Minghao Li 已提交
489
    ASSERT(0);
M
Minghao Li 已提交
490
  }
M
Minghao Li 已提交
491 492 493 494

  // event log
  do {
    char logBuf[128];
S
Shengliang Guan 已提交
495
    snprintf(logBuf, sizeof(logBuf), "wal truncate, from-index:%" PRId64, fromIndex);
M
Minghao Li 已提交
496 497 498
    syncNodeEventLog(pData->pSyncNode, logBuf);
  } while (0);

M
Minghao Li 已提交
499
  return 0;
M
Minghao Li 已提交
500
}
M
Minghao Li 已提交
501

M
Minghao Li 已提交
502
SyncIndex logStoreLastIndex(SSyncLogStore* pLogStore) {
M
Minghao Li 已提交
503 504
  SSyncLogStoreData* pData = pLogStore->data;
  SWal*              pWal = pData->pWal;
M
Minghao Li 已提交
505
  SyncIndex          lastIndex = walGetLastVer(pWal);
M
Minghao Li 已提交
506 507
  return lastIndex;
}
M
Minghao Li 已提交
508

M
Minghao Li 已提交
509
SyncTerm logStoreLastTerm(SSyncLogStore* pLogStore) {
M
Minghao Li 已提交
510
  SyncTerm        lastTerm = 0;
M
Minghao Li 已提交
511
  SSyncRaftEntry* pLastEntry = logStoreGetLastEntry(pLogStore);
M
Minghao Li 已提交
512 513
  if (pLastEntry != NULL) {
    lastTerm = pLastEntry->term;
wafwerar's avatar
wafwerar 已提交
514
    taosMemoryFree(pLastEntry);
M
Minghao Li 已提交
515
  }
M
Minghao Li 已提交
516 517
  return lastTerm;
}
M
Minghao Li 已提交
518

M
Minghao Li 已提交
519
int32_t logStoreUpdateCommitIndex(SSyncLogStore* pLogStore, SyncIndex index) {
520 521
  SSyncLogStoreData* pData = pLogStore->data;
  SWal*              pWal = pData->pWal;
M
Minghao Li 已提交
522
  // ASSERT(walCommit(pWal, index) == 0);
523 524 525 526
  int32_t code = walCommit(pWal, index);
  if (code != 0) {
    int32_t     err = terrno;
    const char* errStr = tstrerror(err);
527 528
    int32_t     sysErr = errno;
    const char* sysErrStr = strerror(errno);
S
Shengliang Guan 已提交
529
    sError("vgId:%d, wal update commit index error, index:%" PRId64 ", err:%d %X, msg:%s, syserr:%d, sysmsg:%s",
530 531
           pData->pSyncNode->vgId, index, err, err, errStr, sysErr, sysErrStr);

532 533
    ASSERT(0);
  }
M
Minghao Li 已提交
534
  return 0;
M
Minghao Li 已提交
535
}
M
Minghao Li 已提交
536

M
Minghao Li 已提交
537 538 539 540 541 542 543 544 545
SyncIndex logStoreGetCommitIndex(SSyncLogStore* pLogStore) {
  SSyncLogStoreData* pData = pLogStore->data;
  return pData->pSyncNode->commitIndex;
}

SSyncRaftEntry* logStoreGetLastEntry(SSyncLogStore* pLogStore) {
  SSyncLogStoreData* pData = pLogStore->data;
  SWal*              pWal = pData->pWal;
  SyncIndex          lastIndex = walGetLastVer(pWal);
M
Minghao Li 已提交
546 547 548 549 550

  SSyncRaftEntry* pEntry = NULL;
  if (lastIndex > 0) {
    pEntry = logStoreGetEntry(pLogStore, lastIndex);
  }
M
Minghao Li 已提交
551 552
  return pEntry;
}
M
Minghao Li 已提交
553

554 555 556 557 558 559 560 561 562 563 564 565
cJSON* logStore2Json(SSyncLogStore* pLogStore) {
  char               u64buf[128] = {0};
  SSyncLogStoreData* pData = (SSyncLogStoreData*)pLogStore->data;
  cJSON*             pRoot = cJSON_CreateObject();

  if (pData != NULL && pData->pWal != NULL) {
    snprintf(u64buf, sizeof(u64buf), "%p", pData->pSyncNode);
    cJSON_AddStringToObject(pRoot, "pSyncNode", u64buf);
    snprintf(u64buf, sizeof(u64buf), "%p", pData->pWal);
    cJSON_AddStringToObject(pRoot, "pWal", u64buf);

    SyncIndex beginIndex = raftLogBeginIndex(pLogStore);
S
Shengliang Guan 已提交
566
    snprintf(u64buf, sizeof(u64buf), "%" PRId64, beginIndex);
567 568 569
    cJSON_AddStringToObject(pRoot, "beginIndex", u64buf);

    SyncIndex endIndex = raftLogEndIndex(pLogStore);
S
Shengliang Guan 已提交
570
    snprintf(u64buf, sizeof(u64buf), "%" PRId64, endIndex);
571 572 573 574 575
    cJSON_AddStringToObject(pRoot, "endIndex", u64buf);

    int32_t count = raftLogEntryCount(pLogStore);
    cJSON_AddNumberToObject(pRoot, "entryCount", count);

S
Shengliang Guan 已提交
576
    snprintf(u64buf, sizeof(u64buf), "%" PRId64, raftLogWriteIndex(pLogStore));
577 578 579 580 581
    cJSON_AddStringToObject(pRoot, "WriteIndex", u64buf);

    snprintf(u64buf, sizeof(u64buf), "%d", raftLogIsEmpty(pLogStore));
    cJSON_AddStringToObject(pRoot, "IsEmpty", u64buf);

S
Shengliang Guan 已提交
582
    snprintf(u64buf, sizeof(u64buf), "%" PRId64, raftLogLastIndex(pLogStore));
583
    cJSON_AddStringToObject(pRoot, "LastIndex", u64buf);
S
Shengliang Guan 已提交
584
    snprintf(u64buf, sizeof(u64buf), "%" PRIu64, raftLogLastTerm(pLogStore));
585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602
    cJSON_AddStringToObject(pRoot, "LastTerm", u64buf);

    cJSON* pEntries = cJSON_CreateArray();
    cJSON_AddItemToObject(pRoot, "pEntries", pEntries);

    if (!raftLogIsEmpty(pLogStore)) {
      for (SyncIndex i = beginIndex; i <= endIndex; ++i) {
        SSyncRaftEntry* pEntry = logStoreGetEntry(pLogStore, i);
        cJSON_AddItemToArray(pEntries, syncEntry2Json(pEntry));
        syncEntryDestory(pEntry);
      }
    }
  }

  cJSON* pJson = cJSON_CreateObject();
  cJSON_AddItemToObject(pJson, "SSyncLogStore", pRoot);
  return pJson;
}
M
Minghao Li 已提交
603 604 605 606 607 608

char* logStore2Str(SSyncLogStore* pLogStore) {
  cJSON* pJson = logStore2Json(pLogStore);
  char*  serialized = cJSON_Print(pJson);
  cJSON_Delete(pJson);
  return serialized;
M
Minghao Li 已提交
609 610
}

M
Minghao Li 已提交
611
cJSON* logStoreSimple2Json(SSyncLogStore* pLogStore) {
612
  char               u64buf[128] = {0};
M
Minghao Li 已提交
613 614 615 616 617 618 619 620
  SSyncLogStoreData* pData = (SSyncLogStoreData*)pLogStore->data;
  cJSON*             pRoot = cJSON_CreateObject();

  if (pData != NULL && pData->pWal != NULL) {
    snprintf(u64buf, sizeof(u64buf), "%p", pData->pSyncNode);
    cJSON_AddStringToObject(pRoot, "pSyncNode", u64buf);
    snprintf(u64buf, sizeof(u64buf), "%p", pData->pWal);
    cJSON_AddStringToObject(pRoot, "pWal", u64buf);
M
Minghao Li 已提交
621

622
    SyncIndex beginIndex = raftLogBeginIndex(pLogStore);
S
Shengliang Guan 已提交
623
    snprintf(u64buf, sizeof(u64buf), "%" PRId64, beginIndex);
M
Minghao Li 已提交
624
    cJSON_AddStringToObject(pRoot, "beginIndex", u64buf);
M
Minghao Li 已提交
625 626

    SyncIndex endIndex = raftLogEndIndex(pLogStore);
S
Shengliang Guan 已提交
627
    snprintf(u64buf, sizeof(u64buf), "%" PRId64, endIndex);
M
Minghao Li 已提交
628 629 630 631
    cJSON_AddStringToObject(pRoot, "endIndex", u64buf);

    int32_t count = raftLogEntryCount(pLogStore);
    cJSON_AddNumberToObject(pRoot, "entryCount", count);
M
Minghao Li 已提交
632

S
Shengliang Guan 已提交
633
    snprintf(u64buf, sizeof(u64buf), "%" PRId64, raftLogWriteIndex(pLogStore));
M
Minghao Li 已提交
634 635 636 637 638
    cJSON_AddStringToObject(pRoot, "WriteIndex", u64buf);

    snprintf(u64buf, sizeof(u64buf), "%d", raftLogIsEmpty(pLogStore));
    cJSON_AddStringToObject(pRoot, "IsEmpty", u64buf);

S
Shengliang Guan 已提交
639
    snprintf(u64buf, sizeof(u64buf), "%" PRId64, raftLogLastIndex(pLogStore));
M
Minghao Li 已提交
640
    cJSON_AddStringToObject(pRoot, "LastIndex", u64buf);
S
Shengliang Guan 已提交
641
    snprintf(u64buf, sizeof(u64buf), "%" PRIu64, raftLogLastTerm(pLogStore));
M
Minghao Li 已提交
642
    cJSON_AddStringToObject(pRoot, "LastTerm", u64buf);
M
Minghao Li 已提交
643 644 645 646 647 648 649 650 651 652 653 654 655 656
  }

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

char* logStoreSimple2Str(SSyncLogStore* pLogStore) {
  cJSON* pJson = logStoreSimple2Json(pLogStore);
  char*  serialized = cJSON_Print(pJson);
  cJSON_Delete(pJson);
  return serialized;
}

M
Minghao Li 已提交
657 658 659 660
SyncIndex logStoreFirstIndex(SSyncLogStore* pLogStore) {
  SSyncLogStoreData* pData = pLogStore->data;
  SWal*              pWal = pData->pWal;
  return walGetFirstVer(pWal);
661 662 663 664 665 666
}

SyncIndex logStoreWalCommitVer(SSyncLogStore* pLogStore) {
  SSyncLogStoreData* pData = pLogStore->data;
  SWal*              pWal = pData->pWal;
  return walGetCommittedVer(pWal);
M
Minghao Li 已提交
667 668
}

M
Minghao Li 已提交
669
// for debug -----------------
M
Minghao Li 已提交
670
void logStorePrint(SSyncLogStore* pLogStore) {
M
Minghao Li 已提交
671
  char* serialized = logStore2Str(pLogStore);
S
Shengliang Guan 已提交
672
  printf("logStorePrint | len:%" PRIu64 " | %s \n", strlen(serialized), serialized);
M
Minghao Li 已提交
673
  fflush(NULL);
wafwerar's avatar
wafwerar 已提交
674
  taosMemoryFree(serialized);
M
Minghao Li 已提交
675 676 677 678
}

void logStorePrint2(char* s, SSyncLogStore* pLogStore) {
  char* serialized = logStore2Str(pLogStore);
S
Shengliang Guan 已提交
679
  printf("logStorePrint2 | len:%" PRIu64 " | %s | %s \n", strlen(serialized), s, serialized);
M
Minghao Li 已提交
680
  fflush(NULL);
wafwerar's avatar
wafwerar 已提交
681
  taosMemoryFree(serialized);
M
Minghao Li 已提交
682
}
M
Minghao Li 已提交
683

M
Minghao Li 已提交
684
void logStoreLog(SSyncLogStore* pLogStore) {
685 686
  if (gRaftDetailLog) {
    char* serialized = logStore2Str(pLogStore);
S
Shengliang Guan 已提交
687
    sTraceLong("logStoreLog | len:%" PRIu64 " | %s", strlen(serialized), serialized);
688 689
    taosMemoryFree(serialized);
  }
M
Minghao Li 已提交
690 691 692
}

void logStoreLog2(char* s, SSyncLogStore* pLogStore) {
693 694
  if (gRaftDetailLog) {
    char* serialized = logStore2Str(pLogStore);
S
Shengliang Guan 已提交
695
    sTraceLong("logStoreLog2 | len:%" PRIu64 " | %s | %s", strlen(serialized), s, serialized);
696 697
    taosMemoryFree(serialized);
  }
698
}
M
Minghao Li 已提交
699 700 701 702

// for debug -----------------
void logStoreSimplePrint(SSyncLogStore* pLogStore) {
  char* serialized = logStoreSimple2Str(pLogStore);
S
Shengliang Guan 已提交
703
  printf("logStoreSimplePrint | len:%" PRIu64 " | %s \n", strlen(serialized), serialized);
M
Minghao Li 已提交
704
  fflush(NULL);
wafwerar's avatar
wafwerar 已提交
705
  taosMemoryFree(serialized);
M
Minghao Li 已提交
706 707 708 709
}

void logStoreSimplePrint2(char* s, SSyncLogStore* pLogStore) {
  char* serialized = logStoreSimple2Str(pLogStore);
S
Shengliang Guan 已提交
710
  printf("logStoreSimplePrint2 | len:%" PRIu64 " | %s | %s \n", strlen(serialized), s, serialized);
M
Minghao Li 已提交
711
  fflush(NULL);
wafwerar's avatar
wafwerar 已提交
712
  taosMemoryFree(serialized);
M
Minghao Li 已提交
713 714 715 716
}

void logStoreSimpleLog(SSyncLogStore* pLogStore) {
  char* serialized = logStoreSimple2Str(pLogStore);
S
Shengliang Guan 已提交
717
  sTrace("logStoreSimpleLog | len:%" PRIu64 " | %s", strlen(serialized), serialized);
wafwerar's avatar
wafwerar 已提交
718
  taosMemoryFree(serialized);
M
Minghao Li 已提交
719 720 721
}

void logStoreSimpleLog2(char* s, SSyncLogStore* pLogStore) {
M
Minghao Li 已提交
722 723
  if (gRaftDetailLog) {
    char* serialized = logStoreSimple2Str(pLogStore);
S
Shengliang Guan 已提交
724
    sTrace("logStoreSimpleLog2 | len:%" PRIu64 " | %s | %s", strlen(serialized), s, serialized);
M
Minghao Li 已提交
725 726
    taosMemoryFree(serialized);
  }
L
Liu Jicong 已提交
727
}