#4720

Committed 08 Sep 2025 08:43AM UTC coverage: 58.139% (-0.6%) from 58.762%

Build # #4720

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

Merge pull request #32881 from taosdata/enh/add-new-windows-ci

fix(ci): update workflow reference to use new Windows CI YAML

Run Details

133181 of 292179 branches covered (45.58%)

Branch coverage included in aggregate %.

201691 of 283811 relevant lines covered (71.07%)

5442780.71 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

57.99

/source/libs/sync/src/syncRaftLog.c

/*
 * 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
#include "syncRaftLog.h"
#include "syncRaftCfg.h"
#include "syncRaftStore.h"
#include "syncUtil.h"

// log[m .. n]

// public function
static int32_t   raftLogRestoreFromSnapshot(struct SSyncLogStore* pLogStore, SyncIndex snapshotIndex);
static int32_t   raftLogAppendEntry(struct SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry, bool forceSync);
static int32_t   raftLogTruncate(struct SSyncLogStore* pLogStore, SyncIndex fromIndex);
static bool      raftLogExist(struct SSyncLogStore* pLogStore, SyncIndex index);
static int32_t   raftLogUpdateCommitIndex(SSyncLogStore* pLogStore, SyncIndex index);
static SyncIndex raftlogCommitIndex(SSyncLogStore* pLogStore);
static int32_t   raftLogGetLastEntry(SSyncLogStore* pLogStore, SSyncRaftEntry** ppLastEntry);

SSyncLogStore* logStoreCreate(SSyncNode* pSyncNode) {
  SSyncLogStore* pLogStore = taosMemoryCalloc(1, sizeof(SSyncLogStore));
  if (pLogStore == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return NULL;
  }

  // pLogStore->pCache = taosLRUCacheInit(10 * 1024 * 1024, 1, .5);
  pLogStore->pCache = taosLRUCacheInit(30 * 1024 * 1024, 1, .5);
  if (pLogStore->pCache == NULL) {
    taosMemoryFree(pLogStore);
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return NULL;
  }

  pLogStore->cacheHit = 0;
  pLogStore->cacheMiss = 0;

  taosLRUCacheSetStrictCapacity(pLogStore->pCache, false);

  pLogStore->data = taosMemoryMalloc(sizeof(SSyncLogStoreData));
  if (!pLogStore->data) {
    taosMemoryFree(pLogStore);
    taosLRUCacheCleanup(pLogStore->pCache);
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return NULL;
  }

  SSyncLogStoreData* pData = pLogStore->data;
  pData->pSyncNode = pSyncNode;
  pData->pWal = pSyncNode->pWal;
  if (pData->pWal == NULL) {
    terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
    return NULL;
  }

  (void)taosThreadMutexInit(&(pData->mutex), NULL);
  pData->pWalHandle = walOpenReader(pData->pWal, 0);
  if (!pData->pWalHandle) {
    taosMemoryFree(pLogStore);
    taosLRUCacheCleanup(pLogStore->pCache);
    (void)taosThreadMutexDestroy(&(pData->mutex));
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return NULL;
  }

  pLogStore->syncLogUpdateCommitIndex = raftLogUpdateCommitIndex;
  pLogStore->syncLogCommitIndex = raftlogCommitIndex;
  pLogStore->syncLogRestoreFromSnapshot = raftLogRestoreFromSnapshot;
  pLogStore->syncLogBeginIndex = raftLogBeginIndex;
  pLogStore->syncLogEndIndex = raftLogEndIndex;
  pLogStore->syncLogIsEmpty = raftLogIsEmpty;
  pLogStore->syncLogEntryCount = raftLogEntryCount;
  pLogStore->syncLogLastIndex = raftLogLastIndex;
  pLogStore->syncLogIndexRetention = raftLogIndexRetention;
  pLogStore->syncLogLastTerm = raftLogLastTerm;
  pLogStore->syncLogAppendEntry = raftLogAppendEntry;
  pLogStore->syncLogGetEntry = raftLogGetEntry;
  pLogStore->syncLogTruncate = raftLogTruncate;
  pLogStore->syncLogWriteIndex = raftLogWriteIndex;
  pLogStore->syncLogExist = raftLogExist;

  return pLogStore;
}

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

    (void)taosThreadMutexLock(&(pData->mutex));
    if (pData->pWalHandle != NULL) {
      walCloseReader(pData->pWalHandle);
      pData->pWalHandle = NULL;
    }
    (void)taosThreadMutexUnlock(&(pData->mutex));
    (void)taosThreadMutexDestroy(&(pData->mutex));

    taosMemoryFree(pLogStore->data);

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

    taosMemoryFree(pLogStore);
  }
}

// log[m .. n]
static int32_t raftLogRestoreFromSnapshot(struct SSyncLogStore* pLogStore, SyncIndex snapshotIndex) {
  if (!(snapshotIndex >= 0)) return TSDB_CODE_SYN_INTERNAL_ERROR;

  SSyncLogStoreData* pData = pLogStore->data;
  SWal*              pWal = pData->pWal;

  sInfo("vgId:%d, before restore from snapshot, commitIndex:%" PRId64, pData->pSyncNode->vgId, pWal->vers.commitVer);

  int32_t            code = walRestoreFromSnapshot(pWal, snapshotIndex);
  if (code != 0) {
    int32_t     err = code;
    const char* errStr = tstrerror(err);
    int32_t     sysErr = ERRNO;
    const char* sysErrStr = strerror(ERRNO);

    sNError(pData->pSyncNode,
            "wal restore from snapshot error, index:%" PRId64 ", err:0x%x, msg:%s, syserr:%d, sysmsg:%s", snapshotIndex,
            err, errStr, sysErr, sysErrStr);
    TAOS_RETURN(err);
  }

  sInfo("vgId:%d, after restore from snapshot, commitIndex:%" PRId64, pData->pSyncNode->vgId, pWal->vers.commitVer);

  TAOS_RETURN(TSDB_CODE_SUCCESS);
}

SyncIndex raftLogBeginIndex(struct SSyncLogStore* pLogStore) {
  SSyncLogStoreData* pData = pLogStore->data;
  SWal*              pWal = pData->pWal;
  SyncIndex          firstVer = walGetFirstVer(pWal);
  return firstVer;
}

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

bool raftLogIsEmpty(struct SSyncLogStore* pLogStore) {
  SSyncLogStoreData* pData = pLogStore->data;
  SWal*              pWal = pData->pWal;
  return walIsEmpty(pWal);
}

int32_t raftLogEntryCount(struct SSyncLogStore* pLogStore) {
  SyncIndex beginIndex = raftLogBeginIndex(pLogStore);
  SyncIndex endIndex = raftLogEndIndex(pLogStore);
  int32_t   count = endIndex - beginIndex + 1;
  return count > 0 ? count : 0;
}

SyncIndex raftLogLastIndex(struct SSyncLogStore* pLogStore) {
  SyncIndex          lastIndex;
  SSyncLogStoreData* pData = pLogStore->data;
  SWal*              pWal = pData->pWal;
  SyncIndex          lastVer = walGetLastVer(pWal);

  return lastVer;
}

SyncIndex raftLogIndexRetention(struct SSyncLogStore* pLogStore, int64_t bytes) {
  SyncIndex          lastIndex;
  SSyncLogStoreData* pData = pLogStore->data;
  SWal*              pWal = pData->pWal;
  SyncIndex          lastVer = walGetVerRetention(pWal, bytes);

  return lastVer;
}

SyncIndex raftLogWriteIndex(struct SSyncLogStore* pLogStore) {
  SSyncLogStoreData* pData = pLogStore->data;
  SWal*              pWal = pData->pWal;
  SyncIndex          lastVer = walGetLastVer(pWal);
  return lastVer + 1;
}

static bool raftLogExist(struct SSyncLogStore* pLogStore, SyncIndex index) {
  SSyncLogStoreData* pData = pLogStore->data;
  SWal*              pWal = pData->pWal;
  bool               b = walLogExist(pWal, index);
  return b;
}

// if success, return last term
// if not log, return 0
// if error, return SYNC_TERM_INVALID
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);
    if (code == 0 && pLastEntry != NULL) {
      SyncTerm lastTerm = pLastEntry->term;
      taosMemoryFree(pLastEntry);
      return lastTerm;
    } else {
      return SYNC_TERM_INVALID;
    }
  }

  // can not be here!
  return SYNC_TERM_INVALID;
}

static int32_t rafeLogReBuildEntry(SSyncRaftEntry* pEntry) {
  int32_t code = 0;
  int32_t lino = 0;
  int32_t flags = 0;
  if (pEntry == NULL) {
    TAOS_RETURN(TSDB_CODE_SYN_INTERNAL_ERROR);
  }
  uint64_t nSubmitTbData = 0;

  if (pEntry->originalRpcType != TDMT_VND_SUBMIT) {
    return code;
  }
  int32_t len = pEntry->dataLen;

  SSubmitReq2Msg* pMsg = (SSubmitReq2Msg*)pEntry->data;

  void* pReq = POINTER_SHIFT(pReq, sizeof(SSubmitReq2Msg));
  len -= sizeof(SSubmitReq2Msg);

  SSubmitReq2* pSubmitReq = &(SSubmitReq2){0};
  SDecoder     dc = {0};
  tDecoderInit(&dc, pReq, len);
  if (tStartDecode(&dc) < 0) {
    code = TSDB_CODE_INVALID_MSG;
    goto _exit;
  }

_exit:

  return code;
}
static int32_t raftLogAppendEntry(struct SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry, bool forceSync) {
  SSyncLogStoreData* pData = pLogStore->data;
  SWal*              pWal = pData->pWal;

  SWalSyncInfo syncMeta = {0};
  syncMeta.isWeek = pEntry->isWeak;
  syncMeta.seqNum = pEntry->seqNum;
  syncMeta.term = pEntry->term;

  int32_t code = 0;
  METRICS_TIMING_BLOCK(pData->pSyncNode->wal_write_time, METRIC_LEVEL_HIGH, {
    code = walAppendLog(pWal, pEntry->index, pEntry->originalRpcType, syncMeta, pEntry->data, pEntry->dataLen,
                        &pEntry->originRpcTraceId);
  });
  METRICS_UPDATE(pData->pSyncNode->wal_write_bytes, METRIC_LEVEL_LOW, (int64_t)pEntry->bytes);

  if (code != 0) {
    int32_t     err = terrno;
    const char* errStr = tstrerror(err);
    int32_t     sysErr = ERRNO;
    const char* sysErrStr = strerror(ERRNO);

    sNError(pData->pSyncNode, "wal write error, index:%" PRId64 ", err:0x%x, msg:%s, syserr:%d, sysmsg:%s",
            pEntry->index, err, errStr, sysErr, sysErrStr);

    TAOS_RETURN(err);
  }

  code = walFsync(pWal, forceSync);
  if (TSDB_CODE_SUCCESS != code) {
    sNError(pData->pSyncNode, "wal fsync failed since %s", tstrerror(code));
    TAOS_RETURN(code);
  }

  sGDebug(&pEntry->originRpcTraceId, "vgId:%d, index:%" PRId64 ", persist raft entry, type:%s origin type:%s",
          pData->pSyncNode->vgId, pEntry->index, TMSG_INFO(pEntry->msgType), TMSG_INFO(pEntry->originalRpcType));
  TAOS_RETURN(TSDB_CODE_SUCCESS);
}

// entry found, return 0
// entry not found, return -1, terrno = TSDB_CODE_WAL_LOG_NOT_EXIST
// other error, return -1
int32_t raftLogGetEntry(struct SSyncLogStore* pLogStore, SyncIndex index, SSyncRaftEntry** ppEntry) {
  SSyncLogStoreData* pData = pLogStore->data;
  SWal*              pWal = pData->pWal;
  int32_t            code = 0;

  *ppEntry = NULL;

  int64_t ts1 = taosGetTimestampNs();
  (void)taosThreadMutexLock(&(pData->mutex));

  SWalReader* pWalHandle = pData->pWalHandle;
  if (pWalHandle == NULL) {
    sError("vgId:%d, wal handle is NULL", pData->pSyncNode->vgId);
    (void)taosThreadMutexUnlock(&(pData->mutex));

    TAOS_RETURN(TSDB_CODE_SYN_INTERNAL_ERROR);
  }

  int64_t ts2 = taosGetTimestampNs();
  code = walReadVer(pWalHandle, index);
  walReadReset(pWalHandle);
  int64_t ts3 = taosGetTimestampNs();

  // code = walReadVerCached(pWalHandle, index);
  if (code != 0) {
    int32_t     err = code;
    const char* errStr = tstrerror(err);
    int32_t     sysErr = ERRNO;
    const char* sysErrStr = strerror(ERRNO);

    if (terrno == TSDB_CODE_WAL_LOG_NOT_EXIST) {
      sNTrace(pData->pSyncNode, "wal read not exist, index:%" PRId64 ", err:0x%x, msg:%s, syserr:%d, sysmsg:%s", index,
              err, errStr, sysErr, sysErrStr);
    } else {
      sNTrace(pData->pSyncNode, "wal read error, index:%" PRId64 ", err:0x%x, msg:%s, syserr:%d, sysmsg:%s", index, err,
              errStr, sysErr, sysErrStr);
    }

    /*
        int32_t saveErr = terrno;
        walCloseReadHandle(pWalHandle);
        terrno = saveErr;
    */

    (void)taosThreadMutexUnlock(&(pData->mutex));

    TAOS_RETURN(code);
  }

  *ppEntry = syncEntryBuild(pWalHandle->pHead->head.bodyLen);
  if (*ppEntry == NULL) return TSDB_CODE_SYN_INTERNAL_ERROR;
  (*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;
  if ((*ppEntry)->dataLen != pWalHandle->pHead->head.bodyLen) return TSDB_CODE_SYN_INTERNAL_ERROR;
  (void)memcpy((*ppEntry)->data, pWalHandle->pHead->head.body, pWalHandle->pHead->head.bodyLen);

  /*
    int32_t saveErr = terrno;
    walCloseReadHandle(pWalHandle);
    terrno = saveErr;
  */

  (void)taosThreadMutexUnlock(&(pData->mutex));
  int64_t ts4 = taosGetTimestampNs();

  int64_t tsElapsed = ts4 - ts1;
  int64_t tsElapsedLock = ts2 - ts1;
  int64_t tsElapsedRead = ts3 - ts2;
  int64_t tsElapsedBuild = ts4 - ts3;

  sNTrace(pData->pSyncNode,
          "read index:%" PRId64 ", elapsed:%" PRId64 ", elapsed-lock:%" PRId64 ", elapsed-read:%" PRId64
          ", elapsed-build:%" PRId64,
          index, tsElapsed, tsElapsedLock, tsElapsedRead, tsElapsedBuild);

  TAOS_RETURN(code);
}

// truncate semantic
static int32_t raftLogTruncate(struct SSyncLogStore* pLogStore, SyncIndex fromIndex) {
  SSyncLogStoreData* pData = pLogStore->data;
  SWal*              pWal = pData->pWal;

  int32_t code = walRollback(pWal, fromIndex);
  if (code != 0) {
    int32_t     err = code;
    const char* errStr = tstrerror(err);
    int32_t     sysErr = ERRNO;
    const char* sysErrStr = strerror(ERRNO);
    sError("vgId:%d, wal truncate error, from-index:%" PRId64 ", err:0x%x, msg:%s, syserr:%d, sysmsg:%s",
           pData->pSyncNode->vgId, fromIndex, err, errStr, sysErr, sysErrStr);
  }

  // event log
  sNTrace(pData->pSyncNode, "log truncate, from-index:%" PRId64, fromIndex);

  TAOS_RETURN(code);
}

// entry found, return 0
// entry not found, return -1, terrno = TSDB_CODE_WAL_LOG_NOT_EXIST
// other error, return -1
static int32_t raftLogGetLastEntry(SSyncLogStore* pLogStore, SSyncRaftEntry** ppLastEntry) {
  SSyncLogStoreData* pData = pLogStore->data;
  SWal*              pWal = pData->pWal;
  if (ppLastEntry == NULL) return TSDB_CODE_SYN_INTERNAL_ERROR;

  *ppLastEntry = NULL;
  if (walIsEmpty(pWal)) {
    TAOS_RETURN(TSDB_CODE_WAL_LOG_NOT_EXIST);
  } else {
    SyncIndex lastIndex = raftLogLastIndex(pLogStore);
    if (!(lastIndex >= SYNC_INDEX_BEGIN)) return TSDB_CODE_SYN_INTERNAL_ERROR;
    int32_t code = raftLogGetEntry(pLogStore, lastIndex, ppLastEntry);

    TAOS_RETURN(code);
  }

  TAOS_RETURN(TSDB_CODE_FAILED);
}

int32_t raftLogUpdateCommitIndex(SSyncLogStore* pLogStore, SyncIndex index) {
  SSyncLogStoreData* pData = pLogStore->data;
  SWal*              pWal = pData->pWal;

  // need not update
  SyncIndex snapshotVer = walGetSnapshotVer(pWal);
  SyncIndex walCommitVer = walGetCommittedVer(pWal);
  SyncIndex wallastVer = walGetLastVer(pWal);

  if (index < snapshotVer || index > wallastVer) {
    // ignore
    TAOS_RETURN(TSDB_CODE_SUCCESS);
  }

  int32_t code = walCommit(pWal, index);
  if (code != 0) {
    int32_t     err = code;
    const char* errStr = tstrerror(err);
    int32_t     sysErr = ERRNO;
    const char* sysErrStr = strerror(ERRNO);
    sError("vgId:%d, index:%" PRId64 ", raft entry update commit index error, code:0x%x msg:%s syserr:%d sysmsg:%s",
           pData->pSyncNode->vgId, index, err, errStr, sysErr, sysErrStr);

    TAOS_RETURN(code);
  }

  TAOS_RETURN(TSDB_CODE_SUCCESS);
}

SyncIndex raftlogCommitIndex(SSyncLogStore* pLogStore) {
  SSyncLogStoreData* pData = pLogStore->data;
  return pData->pSyncNode->commitIndex;
}

SyncIndex logStoreFirstIndex(SSyncLogStore* pLogStore) {
  SSyncLogStoreData* pData = pLogStore->data;
  SWal*              pWal = pData->pWal;
  return walGetFirstVer(pWal);
}

SyncIndex logStoreWalCommitVer(SSyncLogStore* pLogStore) {
  SSyncLogStoreData* pData = pLogStore->data;
  SWal*              pWal = pData->pWal;

  return walGetCommittedVer(pWal);
}

1	/*
2	* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
3	*
4	* This program is free software: you can use, redistribute, and/or modify
5	* it under the terms of the GNU Affero General Public License, version 3
6	* or later ("AGPL"), as published by the Free Software Foundation.
7	*
8	* This program is distributed in the hope that it will be useful, but WITHOUT
9	* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10	* FITNESS FOR A PARTICULAR PURPOSE.
11	*
12	* You should have received a copy of the GNU Affero General Public License
13	* along with this program. If not, see <http://www.gnu.org/licenses/>.
14	*/
15
16	#define _DEFAULT_SOURCE
17	#include "syncRaftLog.h"
18	#include "syncRaftCfg.h"
19	#include "syncRaftStore.h"
20	#include "syncUtil.h"
21
22	// log[m .. n]
23
24	// public function
25	static int32_t raftLogRestoreFromSnapshot(struct SSyncLogStore* pLogStore, SyncIndex snapshotIndex);
26	static int32_t raftLogAppendEntry(struct SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry, bool forceSync);
27	static int32_t raftLogTruncate(struct SSyncLogStore* pLogStore, SyncIndex fromIndex);
28	static bool raftLogExist(struct SSyncLogStore* pLogStore, SyncIndex index);
29	static int32_t raftLogUpdateCommitIndex(SSyncLogStore* pLogStore, SyncIndex index);
30	static SyncIndex raftlogCommitIndex(SSyncLogStore* pLogStore);
31	static int32_t raftLogGetLastEntry(SSyncLogStore* pLogStore, SSyncRaftEntry** ppLastEntry);
32
33	SSyncLogStore* logStoreCreate(SSyncNode* pSyncNode) {	14,331✔
34	SSyncLogStore* pLogStore = taosMemoryCalloc(1, sizeof(SSyncLogStore));	14,331!
35	if (pLogStore == NULL) {	14,338!
36	terrno = TSDB_CODE_OUT_OF_MEMORY;	×
37	return NULL;	×
38	}
39
40	// pLogStore->pCache = taosLRUCacheInit(10 * 1024 * 1024, 1, .5);
41	pLogStore->pCache = taosLRUCacheInit(30 * 1024 * 1024, 1, .5);	14,338✔
42	if (pLogStore->pCache == NULL) {	14,338!
43	taosMemoryFree(pLogStore);	×
44	terrno = TSDB_CODE_OUT_OF_MEMORY;	×
45	return NULL;	×
46	}
47
48	pLogStore->cacheHit = 0;	14,338✔
49	pLogStore->cacheMiss = 0;	14,338✔
50
51	taosLRUCacheSetStrictCapacity(pLogStore->pCache, false);	14,338✔
52
53	pLogStore->data = taosMemoryMalloc(sizeof(SSyncLogStoreData));	14,338!
54	if (!pLogStore->data) {	14,338!
55	taosMemoryFree(pLogStore);	×
56	taosLRUCacheCleanup(pLogStore->pCache);	×
57	terrno = TSDB_CODE_OUT_OF_MEMORY;	×
58	return NULL;	×
59	}
60
61	SSyncLogStoreData* pData = pLogStore->data;	14,338✔
62	pData->pSyncNode = pSyncNode;	14,338✔
63	pData->pWal = pSyncNode->pWal;	14,338✔
64	if (pData->pWal == NULL) {	14,338!
65	terrno = TSDB_CODE_SYN_INTERNAL_ERROR;	×
66	return NULL;	×
67	}
68
69	(void)taosThreadMutexInit(&(pData->mutex), NULL);	14,338✔
70	pData->pWalHandle = walOpenReader(pData->pWal, 0);	14,338✔
71	if (!pData->pWalHandle) {	14,338!
72	taosMemoryFree(pLogStore);	×
73	taosLRUCacheCleanup(pLogStore->pCache);	×
74	(void)taosThreadMutexDestroy(&(pData->mutex));	×
75	terrno = TSDB_CODE_OUT_OF_MEMORY;	×
76	return NULL;	×
77	}
78
79	pLogStore->syncLogUpdateCommitIndex = raftLogUpdateCommitIndex;	14,338✔
80	pLogStore->syncLogCommitIndex = raftlogCommitIndex;	14,338✔
81	pLogStore->syncLogRestoreFromSnapshot = raftLogRestoreFromSnapshot;	14,338✔
82	pLogStore->syncLogBeginIndex = raftLogBeginIndex;	14,338✔
83	pLogStore->syncLogEndIndex = raftLogEndIndex;	14,338✔
84	pLogStore->syncLogIsEmpty = raftLogIsEmpty;	14,338✔
85	pLogStore->syncLogEntryCount = raftLogEntryCount;	14,338✔
86	pLogStore->syncLogLastIndex = raftLogLastIndex;	14,338✔
87	pLogStore->syncLogIndexRetention = raftLogIndexRetention;	14,338✔
88	pLogStore->syncLogLastTerm = raftLogLastTerm;	14,338✔
89	pLogStore->syncLogAppendEntry = raftLogAppendEntry;	14,338✔
90	pLogStore->syncLogGetEntry = raftLogGetEntry;	14,338✔
91	pLogStore->syncLogTruncate = raftLogTruncate;	14,338✔
92	pLogStore->syncLogWriteIndex = raftLogWriteIndex;	14,338✔
93	pLogStore->syncLogExist = raftLogExist;	14,338✔
94
95	return pLogStore;	14,338✔
96	}
97
98	void logStoreDestory(SSyncLogStore* pLogStore) {	14,338✔
99	if (pLogStore != NULL) {	14,338!
100	SSyncLogStoreData* pData = pLogStore->data;	14,338✔
101
102	(void)taosThreadMutexLock(&(pData->mutex));	14,338✔
103	if (pData->pWalHandle != NULL) {	14,338!
104	walCloseReader(pData->pWalHandle);	14,338✔
105	pData->pWalHandle = NULL;	14,336✔
106	}
107	(void)taosThreadMutexUnlock(&(pData->mutex));	14,336✔
108	(void)taosThreadMutexDestroy(&(pData->mutex));	14,338✔
109
110	taosMemoryFree(pLogStore->data);	14,338!
111
112	taosLRUCacheEraseUnrefEntries(pLogStore->pCache);	14,338✔
113	taosLRUCacheCleanup(pLogStore->pCache);	14,338✔
114
115	taosMemoryFree(pLogStore);	14,338!
116	}
117	}	14,338✔
118
119	// log[m .. n]
120	static int32_t raftLogRestoreFromSnapshot(struct SSyncLogStore* pLogStore, SyncIndex snapshotIndex) {	94✔
121	if (!(snapshotIndex >= 0)) return TSDB_CODE_SYN_INTERNAL_ERROR;	94!
122
123	SSyncLogStoreData* pData = pLogStore->data;	94✔
124	SWal* pWal = pData->pWal;	94✔
125
126	sInfo("vgId:%d, before restore from snapshot, commitIndex:%" PRId64, pData->pSyncNode->vgId, pWal->vers.commitVer);	94!
127
128	int32_t code = walRestoreFromSnapshot(pWal, snapshotIndex);	94✔
129	if (code != 0) {	94!
130	int32_t err = code;	×
131	const char* errStr = tstrerror(err);	×
132	int32_t sysErr = ERRNO;	×
133	const char* sysErrStr = strerror(ERRNO);	×
134
135	sNError(pData->pSyncNode,	×
136	"wal restore from snapshot error, index:%" PRId64 ", err:0x%x, msg:%s, syserr:%d, sysmsg:%s", snapshotIndex,
137	err, errStr, sysErr, sysErrStr);
138	TAOS_RETURN(err);	×
139	}
140
141	sInfo("vgId:%d, after restore from snapshot, commitIndex:%" PRId64, pData->pSyncNode->vgId, pWal->vers.commitVer);	94!
142
143	TAOS_RETURN(TSDB_CODE_SUCCESS);	94✔
144	}
145
146	SyncIndex raftLogBeginIndex(struct SSyncLogStore* pLogStore) {	115,853✔
147	SSyncLogStoreData* pData = pLogStore->data;	115,853✔
148	SWal* pWal = pData->pWal;	115,853✔
149	SyncIndex firstVer = walGetFirstVer(pWal);	115,853✔
150	return firstVer;	115,853✔
151	}
152
153	SyncIndex raftLogEndIndex(struct SSyncLogStore* pLogStore) { return raftLogLastIndex(pLogStore); }	25,986✔
154
155	bool raftLogIsEmpty(struct SSyncLogStore* pLogStore) {	25,986✔
156	SSyncLogStoreData* pData = pLogStore->data;	25,986✔
157	SWal* pWal = pData->pWal;	25,986✔
158	return walIsEmpty(pWal);	25,986✔
159	}
160
161	int32_t raftLogEntryCount(struct SSyncLogStore* pLogStore) {	×
162	SyncIndex beginIndex = raftLogBeginIndex(pLogStore);	×
163	SyncIndex endIndex = raftLogEndIndex(pLogStore);	×
164	int32_t count = endIndex - beginIndex + 1;	×
165	return count > 0 ? count : 0;	×
166	}
167
168	SyncIndex raftLogLastIndex(struct SSyncLogStore* pLogStore) {	9,129,788✔
169	SyncIndex lastIndex;
170	SSyncLogStoreData* pData = pLogStore->data;	9,129,788✔
171	SWal* pWal = pData->pWal;	9,129,788✔
172	SyncIndex lastVer = walGetLastVer(pWal);	9,129,788✔
173
174	return lastVer;	9,129,820✔
175	}
176
177	SyncIndex raftLogIndexRetention(struct SSyncLogStore* pLogStore, int64_t bytes) {	1,316✔
178	SyncIndex lastIndex;
179	SSyncLogStoreData* pData = pLogStore->data;	1,316✔
180	SWal* pWal = pData->pWal;	1,316✔
181	SyncIndex lastVer = walGetVerRetention(pWal, bytes);	1,316✔
182
183	return lastVer;	1,316✔
184	}
185
186	SyncIndex raftLogWriteIndex(struct SSyncLogStore* pLogStore) {	×
187	SSyncLogStoreData* pData = pLogStore->data;	×
188	SWal* pWal = pData->pWal;	×
189	SyncIndex lastVer = walGetLastVer(pWal);	×
190	return lastVer + 1;	×
191	}
192
193	static bool raftLogExist(struct SSyncLogStore* pLogStore, SyncIndex index) {	×
194	SSyncLogStoreData* pData = pLogStore->data;	×
195	SWal* pWal = pData->pWal;	×
196	bool b = walLogExist(pWal, index);	×
197	return b;	×
198	}
199
200	// if success, return last term
201	// if not log, return 0
202	// if error, return SYNC_TERM_INVALID
203	SyncTerm raftLogLastTerm(struct SSyncLogStore* pLogStore) {	18,238✔
204	SSyncLogStoreData* pData = pLogStore->data;	18,238✔
205	SWal* pWal = pData->pWal;	18,238✔
206	if (walIsEmpty(pWal)) {	18,238✔
207	return 0;	12,740✔
208	} else {
209	SSyncRaftEntry* pLastEntry;
210	int32_t code = raftLogGetLastEntry(pLogStore, &pLastEntry);	5,498✔
211	if (code == 0 && pLastEntry != NULL) {	5,498!
212	SyncTerm lastTerm = pLastEntry->term;	5,498✔
213	taosMemoryFree(pLastEntry);	5,498!
214	return lastTerm;	5,498✔
215	} else {
216	return SYNC_TERM_INVALID;	×
217	}
218	}
219
220	// can not be here!
221	return SYNC_TERM_INVALID;
222	}
223
224	static int32_t rafeLogReBuildEntry(SSyncRaftEntry* pEntry) {	×
225	int32_t code = 0;	×
226	int32_t lino = 0;	×
227	int32_t flags = 0;	×
228	if (pEntry == NULL) {	×
229	TAOS_RETURN(TSDB_CODE_SYN_INTERNAL_ERROR);	×
230	}
231	uint64_t nSubmitTbData = 0;	×
232
233	if (pEntry->originalRpcType != TDMT_VND_SUBMIT) {	×
234	return code;	×
235	}
236	int32_t len = pEntry->dataLen;	×
237
238	SSubmitReq2Msg* pMsg = (SSubmitReq2Msg*)pEntry->data;	×
239
240	void* pReq = POINTER_SHIFT(pReq, sizeof(SSubmitReq2Msg));	×
241	len -= sizeof(SSubmitReq2Msg);	×
242
243	SSubmitReq2* pSubmitReq = &(SSubmitReq2){0};	×
244	SDecoder dc = {0};	×
245	tDecoderInit(&dc, pReq, len);	×
246	if (tStartDecode(&dc) < 0) {	×
247	code = TSDB_CODE_INVALID_MSG;	×
248	goto _exit;	×
249	}
250
251	_exit:	×
252
253	return code;	×
254	}
255	static int32_t raftLogAppendEntry(struct SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry, bool forceSync) {	2,223,451✔
256	SSyncLogStoreData* pData = pLogStore->data;	2,223,451✔
257	SWal* pWal = pData->pWal;	2,223,451✔
258
259	SWalSyncInfo syncMeta = {0};	2,223,451✔
260	syncMeta.isWeek = pEntry->isWeak;	2,223,451✔
261	syncMeta.seqNum = pEntry->seqNum;	2,223,451✔
262	syncMeta.term = pEntry->term;	2,223,451✔
263
264	int32_t code = 0;	2,223,451✔
265	METRICS_TIMING_BLOCK(pData->pSyncNode->wal_write_time, METRIC_LEVEL_HIGH, {	2,223,451!
266	code = walAppendLog(pWal, pEntry->index, pEntry->originalRpcType, syncMeta, pEntry->data, pEntry->dataLen,
267	&pEntry->originRpcTraceId);
268	});
269	METRICS_UPDATE(pData->pSyncNode->wal_write_bytes, METRIC_LEVEL_LOW, (int64_t)pEntry->bytes);	2,223,423!
270
271	if (code != 0) {	2,223,427!
272	int32_t err = terrno;	×
273	const char* errStr = tstrerror(err);	×
274	int32_t sysErr = ERRNO;	×
275	const char* sysErrStr = strerror(ERRNO);	×
276
277	sNError(pData->pSyncNode, "wal write error, index:%" PRId64 ", err:0x%x, msg:%s, syserr:%d, sysmsg:%s",	×
278	pEntry->index, err, errStr, sysErr, sysErrStr);
279
280	TAOS_RETURN(err);	×
281	}
282
283	code = walFsync(pWal, forceSync);	2,223,427✔
284	if (TSDB_CODE_SUCCESS != code) {	2,223,445!
285	sNError(pData->pSyncNode, "wal fsync failed since %s", tstrerror(code));	×
286	TAOS_RETURN(code);	×
287	}
288
289	sGDebug(&pEntry->originRpcTraceId, "vgId:%d, index:%" PRId64 ", persist raft entry, type:%s origin type:%s",	2,223,445!
290	pData->pSyncNode->vgId, pEntry->index, TMSG_INFO(pEntry->msgType), TMSG_INFO(pEntry->originalRpcType));
291	TAOS_RETURN(TSDB_CODE_SUCCESS);	2,223,445✔
292	}
293
294	// entry found, return 0
295	// entry not found, return -1, terrno = TSDB_CODE_WAL_LOG_NOT_EXIST
296	// other error, return -1
297	int32_t raftLogGetEntry(struct SSyncLogStore* pLogStore, SyncIndex index, SSyncRaftEntry** ppEntry) {	172,451✔
298	SSyncLogStoreData* pData = pLogStore->data;	172,451✔
299	SWal* pWal = pData->pWal;	172,451✔
300	int32_t code = 0;	172,451✔
301
302	*ppEntry = NULL;	172,451✔
303
304	int64_t ts1 = taosGetTimestampNs();	172,449✔
305	(void)taosThreadMutexLock(&(pData->mutex));	172,449✔
306
307	SWalReader* pWalHandle = pData->pWalHandle;	172,457✔
308	if (pWalHandle == NULL) {	172,457!
309	sError("vgId:%d, wal handle is NULL", pData->pSyncNode->vgId);	×
310	(void)taosThreadMutexUnlock(&(pData->mutex));	×
311
312	TAOS_RETURN(TSDB_CODE_SYN_INTERNAL_ERROR);	×
313	}
314
315	int64_t ts2 = taosGetTimestampNs();	172,457✔
316	code = walReadVer(pWalHandle, index);	172,457✔
317	walReadReset(pWalHandle);	172,456✔
318	int64_t ts3 = taosGetTimestampNs();	172,446✔
319
320	// code = walReadVerCached(pWalHandle, index);
321	if (code != 0) {	172,446✔
322	int32_t err = code;	749✔
323	const char* errStr = tstrerror(err);	749✔
324	int32_t sysErr = ERRNO;	749✔
325	const char* sysErrStr = strerror(ERRNO);	749✔
326
327	if (terrno == TSDB_CODE_WAL_LOG_NOT_EXIST) {	749!
328	sNTrace(pData->pSyncNode, "wal read not exist, index:%" PRId64 ", err:0x%x, msg:%s, syserr:%d, sysmsg:%s", index,	749!
329	err, errStr, sysErr, sysErrStr);
330	} else {
331	sNTrace(pData->pSyncNode, "wal read error, index:%" PRId64 ", err:0x%x, msg:%s, syserr:%d, sysmsg:%s", index, err,	×
332	errStr, sysErr, sysErrStr);
333	}
334
335	/*
336	int32_t saveErr = terrno;
337	walCloseReadHandle(pWalHandle);
338	terrno = saveErr;
339	*/
340
341	(void)taosThreadMutexUnlock(&(pData->mutex));	749✔
342
343	TAOS_RETURN(code);	749✔
344	}
345
346	*ppEntry = syncEntryBuild(pWalHandle->pHead->head.bodyLen);	171,697✔
347	if (*ppEntry == NULL) return TSDB_CODE_SYN_INTERNAL_ERROR;	171,693!
348	(*ppEntry)->msgType = TDMT_SYNC_CLIENT_REQUEST;	171,693✔
349	(*ppEntry)->originalRpcType = pWalHandle->pHead->head.msgType;	171,693✔
350	(*ppEntry)->seqNum = pWalHandle->pHead->head.syncMeta.seqNum;	171,693✔
351	(*ppEntry)->isWeak = pWalHandle->pHead->head.syncMeta.isWeek;	171,693✔
352	(*ppEntry)->term = pWalHandle->pHead->head.syncMeta.term;	171,693✔
353	(*ppEntry)->index = index;	171,693✔
354	if ((*ppEntry)->dataLen != pWalHandle->pHead->head.bodyLen) return TSDB_CODE_SYN_INTERNAL_ERROR;	171,693!
355	(void)memcpy((*ppEntry)->data, pWalHandle->pHead->head.body, pWalHandle->pHead->head.bodyLen);	171,693✔
356
357	/*
358	int32_t saveErr = terrno;
359	walCloseReadHandle(pWalHandle);
360	terrno = saveErr;
361	*/
362
363	(void)taosThreadMutexUnlock(&(pData->mutex));	171,693✔
364	int64_t ts4 = taosGetTimestampNs();	171,700✔
365
366	int64_t tsElapsed = ts4 - ts1;	171,700✔
367	int64_t tsElapsedLock = ts2 - ts1;	171,700✔
368	int64_t tsElapsedRead = ts3 - ts2;	171,700✔
369	int64_t tsElapsedBuild = ts4 - ts3;	171,700✔
370
371	sNTrace(pData->pSyncNode,	171,700✔
372	"read index:%" PRId64 ", elapsed:%" PRId64 ", elapsed-lock:%" PRId64 ", elapsed-read:%" PRId64
373	", elapsed-build:%" PRId64,
374	index, tsElapsed, tsElapsedLock, tsElapsedRead, tsElapsedBuild);
375
376	TAOS_RETURN(code);	171,700✔
377	}
378
379	// truncate semantic
380	static int32_t raftLogTruncate(struct SSyncLogStore* pLogStore, SyncIndex fromIndex) {	3✔
381	SSyncLogStoreData* pData = pLogStore->data;	3✔
382	SWal* pWal = pData->pWal;	3✔
383
384	int32_t code = walRollback(pWal, fromIndex);	3✔
385	if (code != 0) {	3!
386	int32_t err = code;	×
387	const char* errStr = tstrerror(err);	×
388	int32_t sysErr = ERRNO;	×
389	const char* sysErrStr = strerror(ERRNO);	×
390	sError("vgId:%d, wal truncate error, from-index:%" PRId64 ", err:0x%x, msg:%s, syserr:%d, sysmsg:%s",	×
391	pData->pSyncNode->vgId, fromIndex, err, errStr, sysErr, sysErrStr);
392	}
393
394	// event log
395	sNTrace(pData->pSyncNode, "log truncate, from-index:%" PRId64, fromIndex);	3!
396
397	TAOS_RETURN(code);	3✔
398	}
399
400	// entry found, return 0
401	// entry not found, return -1, terrno = TSDB_CODE_WAL_LOG_NOT_EXIST
402	// other error, return -1
403	static int32_t raftLogGetLastEntry(SSyncLogStore* pLogStore, SSyncRaftEntry** ppLastEntry) {	5,498✔
404	SSyncLogStoreData* pData = pLogStore->data;	5,498✔
405	SWal* pWal = pData->pWal;	5,498✔
406	if (ppLastEntry == NULL) return TSDB_CODE_SYN_INTERNAL_ERROR;	5,498!
407
408	*ppLastEntry = NULL;	5,498✔
409	if (walIsEmpty(pWal)) {	5,498!
410	TAOS_RETURN(TSDB_CODE_WAL_LOG_NOT_EXIST);	×
411	} else {
412	SyncIndex lastIndex = raftLogLastIndex(pLogStore);	5,498✔
413	if (!(lastIndex >= SYNC_INDEX_BEGIN)) return TSDB_CODE_SYN_INTERNAL_ERROR;	5,498!
414	int32_t code = raftLogGetEntry(pLogStore, lastIndex, ppLastEntry);	5,498✔
415
416	TAOS_RETURN(code);	5,498✔
417	}
418
419	TAOS_RETURN(TSDB_CODE_FAILED);
420	}
421
422	int32_t raftLogUpdateCommitIndex(SSyncLogStore* pLogStore, SyncIndex index) {	2,271,071✔
423	SSyncLogStoreData* pData = pLogStore->data;	2,271,071✔
424	SWal* pWal = pData->pWal;	2,271,071✔
425
426	// need not update
427	SyncIndex snapshotVer = walGetSnapshotVer(pWal);	2,271,071✔
428	SyncIndex walCommitVer = walGetCommittedVer(pWal);	2,271,077✔
429	SyncIndex wallastVer = walGetLastVer(pWal);	2,271,081✔
430
431	if (index < snapshotVer \|\| index > wallastVer) {	2,271,104!
432	// ignore
433	TAOS_RETURN(TSDB_CODE_SUCCESS);	5✔
434	}
435
436	int32_t code = walCommit(pWal, index);	2,271,099✔
437	if (code != 0) {	2,271,108!
438	int32_t err = code;	×
439	const char* errStr = tstrerror(err);	×
440	int32_t sysErr = ERRNO;	×
441	const char* sysErrStr = strerror(ERRNO);	×
442	sError("vgId:%d, index:%" PRId64 ", raft entry update commit index error, code:0x%x msg:%s syserr:%d sysmsg:%s",	×
443	pData->pSyncNode->vgId, index, err, errStr, sysErr, sysErrStr);
444
445	TAOS_RETURN(code);	×
446	}
447
448	TAOS_RETURN(TSDB_CODE_SUCCESS);	2,271,108✔
449	}
450
451	SyncIndex raftlogCommitIndex(SSyncLogStore* pLogStore) {	14,338✔
452	SSyncLogStoreData* pData = pLogStore->data;	14,338✔
453	return pData->pSyncNode->commitIndex;	14,338✔
454	}
455
456	SyncIndex logStoreFirstIndex(SSyncLogStore* pLogStore) {	×
457	SSyncLogStoreData* pData = pLogStore->data;	×
458	SWal* pWal = pData->pWal;	×
459	return walGetFirstVer(pWal);	×
460	}
461
462	SyncIndex logStoreWalCommitVer(SSyncLogStore* pLogStore) {	×
463	SSyncLogStoreData* pData = pLogStore->data;	×
464	SWal* pWal = pData->pWal;	×
465
466	return walGetCommittedVer(pWal);	×
467	}

taosdata / TDengine / #4720

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous