#4756

Committed 25 Sep 2025 05:58AM UTC coverage: 58.829% (+0.2%) from 58.63%

Build # #4756

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

enh: taos command line support '-uroot' on windows (#33055)

Coverage Stats

135574 of 293169 branches covered (46.24%)

Branch coverage included in aggregate %.

204395 of 284720 relevant lines covered (71.79%)

18747092.16 hits per line

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

57.74

/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,786✔
34	SSyncLogStore* pLogStore = taosMemoryCalloc(1, sizeof(SSyncLogStore));	14,786!
35	if (pLogStore == NULL) {	14,793!
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,793✔
42	if (pLogStore->pCache == NULL) {	14,793!
43	taosMemoryFree(pLogStore);	×
44	terrno = TSDB_CODE_OUT_OF_MEMORY;	×
45	return NULL;	×
46	}
47
48	pLogStore->cacheHit = 0;	14,793✔
49	pLogStore->cacheMiss = 0;	14,793✔
50
51	taosLRUCacheSetStrictCapacity(pLogStore->pCache, false);	14,793✔
52
53	pLogStore->data = taosMemoryMalloc(sizeof(SSyncLogStoreData));	14,793!
54	if (!pLogStore->data) {	14,793!
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,793✔
62	pData->pSyncNode = pSyncNode;	14,793✔
63	pData->pWal = pSyncNode->pWal;	14,793✔
64	if (pData->pWal == NULL) {	14,793!
65	terrno = TSDB_CODE_SYN_INTERNAL_ERROR;	×
66	return NULL;	×
67	}
68
69	(void)taosThreadMutexInit(&(pData->mutex), NULL);	14,793✔
70	pData->pWalHandle = walOpenReader(pData->pWal, 0);	14,793✔
71	if (!pData->pWalHandle) {	14,793!
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,793✔
80	pLogStore->syncLogCommitIndex = raftlogCommitIndex;	14,793✔
81	pLogStore->syncLogRestoreFromSnapshot = raftLogRestoreFromSnapshot;	14,793✔
82	pLogStore->syncLogBeginIndex = raftLogBeginIndex;	14,793✔
83	pLogStore->syncLogEndIndex = raftLogEndIndex;	14,793✔
84	pLogStore->syncLogIsEmpty = raftLogIsEmpty;	14,793✔
85	pLogStore->syncLogEntryCount = raftLogEntryCount;	14,793✔
86	pLogStore->syncLogLastIndex = raftLogLastIndex;	14,793✔
87	pLogStore->syncLogIndexRetention = raftLogIndexRetention;	14,793✔
88	pLogStore->syncLogLastTerm = raftLogLastTerm;	14,793✔
89	pLogStore->syncLogAppendEntry = raftLogAppendEntry;	14,793✔
90	pLogStore->syncLogGetEntry = raftLogGetEntry;	14,793✔
91	pLogStore->syncLogTruncate = raftLogTruncate;	14,793✔
92	pLogStore->syncLogWriteIndex = raftLogWriteIndex;	14,793✔
93	pLogStore->syncLogExist = raftLogExist;	14,793✔
94
95	return pLogStore;	14,793✔
96	}
97
98	void logStoreDestory(SSyncLogStore* pLogStore) {	14,790✔
99	if (pLogStore != NULL) {	14,790!
100	SSyncLogStoreData* pData = pLogStore->data;	14,790✔
101
102	(void)taosThreadMutexLock(&(pData->mutex));	14,790✔
103	if (pData->pWalHandle != NULL) {	14,792!
104	walCloseReader(pData->pWalHandle);	14,792✔
105	pData->pWalHandle = NULL;	14,791✔
106	}
107	(void)taosThreadMutexUnlock(&(pData->mutex));	14,791✔
108	(void)taosThreadMutexDestroy(&(pData->mutex));	14,791✔
109
110	taosMemoryFree(pLogStore->data);	14,791!
111
112	taosLRUCacheEraseUnrefEntries(pLogStore->pCache);	14,790✔
113	taosLRUCacheCleanup(pLogStore->pCache);	14,792✔
114
115	taosMemoryFree(pLogStore);	14,792!
116	}
117	}	14,792✔
118
119	// log[m .. n]
120	static int32_t raftLogRestoreFromSnapshot(struct SSyncLogStore* pLogStore, SyncIndex snapshotIndex) {	116✔
121	if (!(snapshotIndex >= 0)) return TSDB_CODE_SYN_INTERNAL_ERROR;	116!
122
123	SSyncLogStoreData* pData = pLogStore->data;	116✔
124	SWal* pWal = pData->pWal;	116✔
125
126	sInfo("vgId:%d, before restore from snapshot, commitIndex:%" PRId64, pData->pSyncNode->vgId, pWal->vers.commitVer);	116!
127
128	int32_t code = walRestoreFromSnapshot(pWal, snapshotIndex);	116✔
129	if (code != 0) {	116!
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);	116!
142
143	TAOS_RETURN(TSDB_CODE_SUCCESS);	116✔
144	}
145
146	SyncIndex raftLogBeginIndex(struct SSyncLogStore* pLogStore) {	129,188✔
147	SSyncLogStoreData* pData = pLogStore->data;	129,188✔
148	SWal* pWal = pData->pWal;	129,188✔
149	SyncIndex firstVer = walGetFirstVer(pWal);	129,188✔
150	return firstVer;	129,187✔
151	}
152
153	SyncIndex raftLogEndIndex(struct SSyncLogStore* pLogStore) { return raftLogLastIndex(pLogStore); }	31,238✔
154
155	bool raftLogIsEmpty(struct SSyncLogStore* pLogStore) {	31,238✔
156	SSyncLogStoreData* pData = pLogStore->data;	31,238✔
157	SWal* pWal = pData->pWal;	31,238✔
158	return walIsEmpty(pWal);	31,238✔
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) {	54,783,981✔
169	SyncIndex lastIndex;
170	SSyncLogStoreData* pData = pLogStore->data;	54,783,981✔
171	SWal* pWal = pData->pWal;	54,783,981✔
172	SyncIndex lastVer = walGetLastVer(pWal);	54,783,981✔
173
174	return lastVer;	54,784,152✔
175	}
176
177	SyncIndex raftLogIndexRetention(struct SSyncLogStore* pLogStore, int64_t bytes) {	1,391✔
178	SyncIndex lastIndex;
179	SSyncLogStoreData* pData = pLogStore->data;	1,391✔
180	SWal* pWal = pData->pWal;	1,391✔
181	SyncIndex lastVer = walGetVerRetention(pWal, bytes);	1,391✔
182
183	return lastVer;	1,391✔
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,797✔
204	SSyncLogStoreData* pData = pLogStore->data;	18,797✔
205	SWal* pWal = pData->pWal;	18,797✔
206	if (walIsEmpty(pWal)) {	18,797✔
207	return 0;	13,081✔
208	} else {
209	SSyncRaftEntry* pLastEntry;
210	int32_t code = raftLogGetLastEntry(pLogStore, &pLastEntry);	5,716✔
211	if (code == 0 && pLastEntry != NULL) {	5,716!
212	SyncTerm lastTerm = pLastEntry->term;	5,716✔
213	taosMemoryFree(pLastEntry);	5,716!
214	return lastTerm;	5,716✔
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) {	13,709,249✔
256	SSyncLogStoreData* pData = pLogStore->data;	13,709,249✔
257	SWal* pWal = pData->pWal;	13,709,249✔
258
259	SWalSyncInfo syncMeta = {0};	13,709,249✔
260	syncMeta.isWeek = pEntry->isWeak;	13,709,249✔
261	syncMeta.seqNum = pEntry->seqNum;	13,709,249✔
262	syncMeta.term = pEntry->term;	13,709,249✔
263
264	int32_t code = 0;	13,709,249✔
265	METRICS_TIMING_BLOCK(pData->pSyncNode->wal_write_time, METRIC_LEVEL_HIGH, {	13,709,249!
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);	13,709,357!
270
271	if (code != 0) {	13,709,302!
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);	13,709,302✔
284	if (TSDB_CODE_SUCCESS != code) {	13,709,469!
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",	13,709,469!
290	pData->pSyncNode->vgId, pEntry->index, TMSG_INFO(pEntry->msgType), TMSG_INFO(pEntry->originalRpcType));
291	TAOS_RETURN(TSDB_CODE_SUCCESS);	13,709,469✔
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) {	3,473,053✔
298	SSyncLogStoreData* pData = pLogStore->data;	3,473,053✔
299	SWal* pWal = pData->pWal;	3,473,053✔
300	int32_t code = 0;	3,473,053✔
301
302	*ppEntry = NULL;	3,473,053✔
303
304	int64_t ts1 = taosGetTimestampNs();	3,473,055✔
305	(void)taosThreadMutexLock(&(pData->mutex));	3,473,055✔
306
307	SWalReader* pWalHandle = pData->pWalHandle;	3,473,055✔
308	if (pWalHandle == NULL) {	3,473,055!
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();	3,473,050✔
316	code = walReadVer(pWalHandle, index);	3,473,050✔
317	walReadReset(pWalHandle);	3,473,053✔
318	int64_t ts3 = taosGetTimestampNs();	3,473,047✔
319
320	// code = walReadVerCached(pWalHandle, index);
321	if (code != 0) {	3,473,047✔
322	int32_t err = code;	924✔
323	const char* errStr = tstrerror(err);	924✔
324	int32_t sysErr = ERRNO;	924✔
325	const char* sysErrStr = strerror(ERRNO);	924✔
326
327	if (terrno == TSDB_CODE_WAL_LOG_NOT_EXIST) {	924!
328	sNTrace(pData->pSyncNode, "wal read not exist, index:%" PRId64 ", err:0x%x, msg:%s, syserr:%d, sysmsg:%s", index,	924!
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));	924✔
342
343	TAOS_RETURN(code);	924✔
344	}
345
346	*ppEntry = syncEntryBuild(pWalHandle->pHead->head.bodyLen);	3,472,123✔
347	if (*ppEntry == NULL) return TSDB_CODE_SYN_INTERNAL_ERROR;	3,472,121!
348	(*ppEntry)->msgType = TDMT_SYNC_CLIENT_REQUEST;	3,472,121✔
349	(*ppEntry)->originalRpcType = pWalHandle->pHead->head.msgType;	3,472,121✔
350	(*ppEntry)->seqNum = pWalHandle->pHead->head.syncMeta.seqNum;	3,472,121✔
351	(*ppEntry)->isWeak = pWalHandle->pHead->head.syncMeta.isWeek;	3,472,121✔
352	(*ppEntry)->term = pWalHandle->pHead->head.syncMeta.term;	3,472,121✔
353	(*ppEntry)->index = index;	3,472,121✔
354	if ((*ppEntry)->dataLen != pWalHandle->pHead->head.bodyLen) return TSDB_CODE_SYN_INTERNAL_ERROR;	3,472,121!
355	(void)memcpy((*ppEntry)->data, pWalHandle->pHead->head.body, pWalHandle->pHead->head.bodyLen);	3,472,121✔
356
357	/*
358	int32_t saveErr = terrno;
359	walCloseReadHandle(pWalHandle);
360	terrno = saveErr;
361	*/
362
363	(void)taosThreadMutexUnlock(&(pData->mutex));	3,472,121✔
364	int64_t ts4 = taosGetTimestampNs();	3,472,116✔
365
366	int64_t tsElapsed = ts4 - ts1;	3,472,116✔
367	int64_t tsElapsedLock = ts2 - ts1;	3,472,116✔
368	int64_t tsElapsedRead = ts3 - ts2;	3,472,116✔
369	int64_t tsElapsedBuild = ts4 - ts3;	3,472,116✔
370
371	sNTrace(pData->pSyncNode,	3,472,116✔
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);	3,472,116✔
377	}
378
379	// truncate semantic
380	static int32_t raftLogTruncate(struct SSyncLogStore* pLogStore, SyncIndex fromIndex) {	5✔
381	SSyncLogStoreData* pData = pLogStore->data;	5✔
382	SWal* pWal = pData->pWal;	5✔
383
384	int32_t code = walRollback(pWal, fromIndex);	5✔
385	if (code != 0) {	5!
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);	5!
396
397	TAOS_RETURN(code);	5✔
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,716✔
404	SSyncLogStoreData* pData = pLogStore->data;	5,716✔
405	SWal* pWal = pData->pWal;	5,716✔
406	if (ppLastEntry == NULL) return TSDB_CODE_SYN_INTERNAL_ERROR;	5,716!
407
408	*ppLastEntry = NULL;	5,716✔
409	if (walIsEmpty(pWal)) {	5,716!
410	TAOS_RETURN(TSDB_CODE_WAL_LOG_NOT_EXIST);	×
411	} else {
412	SyncIndex lastIndex = raftLogLastIndex(pLogStore);	5,716✔
413	if (!(lastIndex >= SYNC_INDEX_BEGIN)) return TSDB_CODE_SYN_INTERNAL_ERROR;	5,716!
414	int32_t code = raftLogGetEntry(pLogStore, lastIndex, ppLastEntry);	5,716✔
415
416	TAOS_RETURN(code);	5,716✔
417	}
418
419	TAOS_RETURN(TSDB_CODE_FAILED);
420	}
421
422	int32_t raftLogUpdateCommitIndex(SSyncLogStore* pLogStore, SyncIndex index) {	13,452,150✔
423	SSyncLogStoreData* pData = pLogStore->data;	13,452,150✔
424	SWal* pWal = pData->pWal;	13,452,150✔
425
426	// need not update
427	SyncIndex snapshotVer = walGetSnapshotVer(pWal);	13,452,150✔
428	SyncIndex walCommitVer = walGetCommittedVer(pWal);	13,452,244✔
429	SyncIndex wallastVer = walGetLastVer(pWal);	13,452,296✔
430
431	if (index < snapshotVer \|\| index > wallastVer) {	13,452,306!
432	// ignore
433	TAOS_RETURN(TSDB_CODE_SUCCESS);	×
434	}
435
436	int32_t code = walCommit(pWal, index);	13,452,311✔
437	if (code != 0) {	13,452,316!
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);	13,452,316✔
449	}
450
451	SyncIndex raftlogCommitIndex(SSyncLogStore* pLogStore) {	14,792✔
452	SSyncLogStoreData* pData = pLogStore->data;	14,792✔
453	return pData->pSyncNode->commitIndex;	14,792✔
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 / #4756

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