#3942

Committed 25 Apr 2025 11:21AM UTC coverage: 62.853% (+0.3%) from 62.507%

Build # #3942

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

docs: jdbc tmq supports database subscription. [TS-6222] (#30819)

* docs: jdbc tmq supports database subscription. [TS-6222]

* Update docs/zh/07-develop/07-tmq.md

Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>

* Update 07-tmq.md

---------

Co-authored-by: haoranchen <haoran920c@163.com>
Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>

Run Details

156603 of 317531 branches covered (49.32%)

Branch coverage included in aggregate %.

241895 of 316485 relevant lines covered (76.43%)

6664240.48 hits per line

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

61.73

/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, NULL, 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;
  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);
  }

  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 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;

  int64_t tsWriteBegin = taosGetTimestampNs();
  int32_t code = walAppendLog(pWal, pEntry->index, pEntry->originalRpcType, syncMeta, pEntry->data, pEntry->dataLen,
                              &pEntry->originRpcTraceId);
  int64_t tsWriteEnd = taosGetTimestampNs();
  int64_t tsElapsed = tsWriteEnd - tsWriteBegin;

  if (TSDB_CODE_SUCCESS != code) {
    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 elapsed:%" PRId64,
          pData->pSyncNode->vgId, pEntry->index, TMSG_INFO(pEntry->msgType), TMSG_INFO(pEntry->originalRpcType),
          tsElapsed);
  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) {	15,598✔
34	SSyncLogStore* pLogStore = taosMemoryCalloc(1, sizeof(SSyncLogStore));	15,598!
35	if (pLogStore == NULL) {	15,605!
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);	15,605✔
42	if (pLogStore->pCache == NULL) {	15,605!
43	taosMemoryFree(pLogStore);	×
44	terrno = TSDB_CODE_OUT_OF_MEMORY;	×
45	return NULL;	×
46	}
47
48	pLogStore->cacheHit = 0;	15,605✔
49	pLogStore->cacheMiss = 0;	15,605✔
50
51	taosLRUCacheSetStrictCapacity(pLogStore->pCache, false);	15,605✔
52
53	pLogStore->data = taosMemoryMalloc(sizeof(SSyncLogStoreData));	15,605!
54	if (!pLogStore->data) {	15,605!
55	taosMemoryFree(pLogStore);	×
56	taosLRUCacheCleanup(pLogStore->pCache);	×
57	terrno = TSDB_CODE_OUT_OF_MEMORY;	×
58	return NULL;	×
59	}
60
61	SSyncLogStoreData* pData = pLogStore->data;	15,605✔
62	pData->pSyncNode = pSyncNode;	15,605✔
63	pData->pWal = pSyncNode->pWal;	15,605✔
64	if (pData->pWal == NULL) {	15,605!
65	terrno = TSDB_CODE_SYN_INTERNAL_ERROR;	×
66	return NULL;	×
67	}
68
69	(void)taosThreadMutexInit(&(pData->mutex), NULL);	15,605✔
70	pData->pWalHandle = walOpenReader(pData->pWal, NULL, 0);	15,605✔
71	if (!pData->pWalHandle) {	15,604!
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;	15,604✔
80	pLogStore->syncLogCommitIndex = raftlogCommitIndex;	15,604✔
81	pLogStore->syncLogRestoreFromSnapshot = raftLogRestoreFromSnapshot;	15,604✔
82	pLogStore->syncLogBeginIndex = raftLogBeginIndex;	15,604✔
83	pLogStore->syncLogEndIndex = raftLogEndIndex;	15,604✔
84	pLogStore->syncLogIsEmpty = raftLogIsEmpty;	15,604✔
85	pLogStore->syncLogEntryCount = raftLogEntryCount;	15,604✔
86	pLogStore->syncLogLastIndex = raftLogLastIndex;	15,604✔
87	pLogStore->syncLogIndexRetention = raftLogIndexRetention;	15,604✔
88	pLogStore->syncLogLastTerm = raftLogLastTerm;	15,604✔
89	pLogStore->syncLogAppendEntry = raftLogAppendEntry;	15,604✔
90	pLogStore->syncLogGetEntry = raftLogGetEntry;	15,604✔
91	pLogStore->syncLogTruncate = raftLogTruncate;	15,604✔
92	pLogStore->syncLogWriteIndex = raftLogWriteIndex;	15,604✔
93	pLogStore->syncLogExist = raftLogExist;	15,604✔
94
95	return pLogStore;	15,604✔
96	}
97
98	void logStoreDestory(SSyncLogStore* pLogStore) {	15,600✔
99	if (pLogStore != NULL) {	15,600!
100	SSyncLogStoreData* pData = pLogStore->data;	15,600✔
101
102	(void)taosThreadMutexLock(&(pData->mutex));	15,600✔
103	if (pData->pWalHandle != NULL) {	15,602!
104	walCloseReader(pData->pWalHandle);	15,602✔
105	pData->pWalHandle = NULL;	15,600✔
106	}
107	(void)taosThreadMutexUnlock(&(pData->mutex));	15,600✔
108	(void)taosThreadMutexDestroy(&(pData->mutex));	15,601✔
109
110	taosMemoryFree(pLogStore->data);	15,601!
111
112	taosLRUCacheEraseUnrefEntries(pLogStore->pCache);	15,599✔
113	taosLRUCacheCleanup(pLogStore->pCache);	15,602✔
114
115	taosMemoryFree(pLogStore);	15,600!
116	}
117	}	15,599✔
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	int32_t code = walRestoreFromSnapshot(pWal, snapshotIndex);	94✔
126	if (code != 0) {	94!
127	int32_t err = code;	×
128	const char* errStr = tstrerror(err);	×
129	int32_t sysErr = ERRNO;	×
130	const char* sysErrStr = strerror(ERRNO);	×
131
132	sNError(pData->pSyncNode,	×
133	"wal restore from snapshot error, index:%" PRId64 ", err:0x%x, msg:%s, syserr:%d, sysmsg:%s", snapshotIndex,
134	err, errStr, sysErr, sysErrStr);
135	TAOS_RETURN(err);	×
136	}
137
138	TAOS_RETURN(TSDB_CODE_SUCCESS);	94✔
139	}
140
141	SyncIndex raftLogBeginIndex(struct SSyncLogStore* pLogStore) {	259,472✔
142	SSyncLogStoreData* pData = pLogStore->data;	259,472✔
143	SWal* pWal = pData->pWal;	259,472✔
144	SyncIndex firstVer = walGetFirstVer(pWal);	259,472✔
145	return firstVer;	259,472✔
146	}
147
148	SyncIndex raftLogEndIndex(struct SSyncLogStore* pLogStore) { return raftLogLastIndex(pLogStore); }	28,612✔
149
150	bool raftLogIsEmpty(struct SSyncLogStore* pLogStore) {	28,612✔
151	SSyncLogStoreData* pData = pLogStore->data;	28,612✔
152	SWal* pWal = pData->pWal;	28,612✔
153	return walIsEmpty(pWal);	28,612✔
154	}
155
156	int32_t raftLogEntryCount(struct SSyncLogStore* pLogStore) {	×
157	SyncIndex beginIndex = raftLogBeginIndex(pLogStore);	×
158	SyncIndex endIndex = raftLogEndIndex(pLogStore);	×
159	int32_t count = endIndex - beginIndex + 1;	×
160	return count > 0 ? count : 0;	×
161	}
162
163	SyncIndex raftLogLastIndex(struct SSyncLogStore* pLogStore) {	10,517,869✔
164	SyncIndex lastIndex;
165	SSyncLogStoreData* pData = pLogStore->data;	10,517,869✔
166	SWal* pWal = pData->pWal;	10,517,869✔
167	SyncIndex lastVer = walGetLastVer(pWal);	10,517,869✔
168
169	return lastVer;	10,518,060✔
170	}
171
172	SyncIndex raftLogIndexRetention(struct SSyncLogStore* pLogStore, int64_t bytes) {	1,893✔
173	SyncIndex lastIndex;
174	SSyncLogStoreData* pData = pLogStore->data;	1,893✔
175	SWal* pWal = pData->pWal;	1,893✔
176	SyncIndex lastVer = walGetVerRetention(pWal, bytes);	1,893✔
177
178	return lastVer;	1,893✔
179	}
180
181	SyncIndex raftLogWriteIndex(struct SSyncLogStore* pLogStore) {	×
182	SSyncLogStoreData* pData = pLogStore->data;	×
183	SWal* pWal = pData->pWal;	×
184	SyncIndex lastVer = walGetLastVer(pWal);	×
185	return lastVer + 1;	×
186	}
187
188	static bool raftLogExist(struct SSyncLogStore* pLogStore, SyncIndex index) {	×
189	SSyncLogStoreData* pData = pLogStore->data;	×
190	SWal* pWal = pData->pWal;	×
191	bool b = walLogExist(pWal, index);	×
192	return b;	×
193	}
194
195	// if success, return last term
196	// if not log, return 0
197	// if error, return SYNC_TERM_INVALID
198	SyncTerm raftLogLastTerm(struct SSyncLogStore* pLogStore) {	20,547✔
199	SSyncLogStoreData* pData = pLogStore->data;	20,547✔
200	SWal* pWal = pData->pWal;	20,547✔
201	if (walIsEmpty(pWal)) {	20,547✔
202	return 0;	14,687✔
203	} else {
204	SSyncRaftEntry* pLastEntry;
205	int32_t code = raftLogGetLastEntry(pLogStore, &pLastEntry);	5,860✔
206	if (code == 0 && pLastEntry != NULL) {	5,860!
207	SyncTerm lastTerm = pLastEntry->term;	5,860✔
208	taosMemoryFree(pLastEntry);	5,860!
209	return lastTerm;	5,860✔
210	} else {
211	return SYNC_TERM_INVALID;	×
212	}
213	}
214
215	// can not be here!
216	return SYNC_TERM_INVALID;
217	}
218
219	static int32_t raftLogAppendEntry(struct SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry, bool forceSync) {	2,533,672✔
220	SSyncLogStoreData* pData = pLogStore->data;	2,533,672✔
221	SWal* pWal = pData->pWal;	2,533,672✔
222
223	SWalSyncInfo syncMeta = {0};	2,533,672✔
224	syncMeta.isWeek = pEntry->isWeak;	2,533,672✔
225	syncMeta.seqNum = pEntry->seqNum;	2,533,672✔
226	syncMeta.term = pEntry->term;	2,533,672✔
227
228	int64_t tsWriteBegin = taosGetTimestampNs();	2,533,748✔
229	int32_t code = walAppendLog(pWal, pEntry->index, pEntry->originalRpcType, syncMeta, pEntry->data, pEntry->dataLen,	2,533,748✔
230	&pEntry->originRpcTraceId);	2,533,748✔
231	int64_t tsWriteEnd = taosGetTimestampNs();	2,533,781✔
232	int64_t tsElapsed = tsWriteEnd - tsWriteBegin;	2,533,781✔
233
234	if (TSDB_CODE_SUCCESS != code) {	2,533,781!
235	int32_t err = terrno;	×
236	const char* errStr = tstrerror(err);	×
237	int32_t sysErr = ERRNO;	×
238	const char* sysErrStr = strerror(ERRNO);	×
239
240	sNError(pData->pSyncNode, "wal write error, index:%" PRId64 ", err:0x%x, msg:%s, syserr:%d, sysmsg:%s",	×
241	pEntry->index, err, errStr, sysErr, sysErrStr);
242
243	TAOS_RETURN(err);	×
244	}
245
246	code = walFsync(pWal, forceSync);	2,533,781✔
247	if (TSDB_CODE_SUCCESS != code) {	2,533,662!
248	sNError(pData->pSyncNode, "wal fsync failed since %s", tstrerror(code));	×
249	TAOS_RETURN(code);	×
250	}
251
252	sGDebug(&pEntry->originRpcTraceId,	2,533,662!
253	"vgId:%d, index:%" PRId64 ", persist raft entry, type:%s origin type:%s elapsed:%" PRId64,
254	pData->pSyncNode->vgId, pEntry->index, TMSG_INFO(pEntry->msgType), TMSG_INFO(pEntry->originalRpcType),
255	tsElapsed);
256	TAOS_RETURN(TSDB_CODE_SUCCESS);	2,533,688✔
257	}
258
259	// entry found, return 0
260	// entry not found, return -1, terrno = TSDB_CODE_WAL_LOG_NOT_EXIST
261	// other error, return -1
262	int32_t raftLogGetEntry(struct SSyncLogStore* pLogStore, SyncIndex index, SSyncRaftEntry** ppEntry) {	519,247✔
263	SSyncLogStoreData* pData = pLogStore->data;	519,247✔
264	SWal* pWal = pData->pWal;	519,247✔
265	int32_t code = 0;	519,247✔
266
267	*ppEntry = NULL;	519,247✔
268
269	int64_t ts1 = taosGetTimestampNs();	519,253✔
270	(void)taosThreadMutexLock(&(pData->mutex));	519,253✔
271
272	SWalReader* pWalHandle = pData->pWalHandle;	519,261✔
273	if (pWalHandle == NULL) {	519,261!
274	sError("vgId:%d, wal handle is NULL", pData->pSyncNode->vgId);	×
275	(void)taosThreadMutexUnlock(&(pData->mutex));	×
276
277	TAOS_RETURN(TSDB_CODE_SYN_INTERNAL_ERROR);	×
278	}
279
280	int64_t ts2 = taosGetTimestampNs();	519,256✔
281	code = walReadVer(pWalHandle, index);	519,256✔
282	walReadReset(pWalHandle);	519,280✔
283	int64_t ts3 = taosGetTimestampNs();	519,260✔
284
285	// code = walReadVerCached(pWalHandle, index);
286	if (code != 0) {	519,260✔
287	int32_t err = code;	775✔
288	const char* errStr = tstrerror(err);	775✔
289	int32_t sysErr = ERRNO;	775✔
290	const char* sysErrStr = strerror(ERRNO);	775✔
291
292	if (terrno == TSDB_CODE_WAL_LOG_NOT_EXIST) {	775!
293	sNTrace(pData->pSyncNode, "wal read not exist, index:%" PRId64 ", err:0x%x, msg:%s, syserr:%d, sysmsg:%s", index,	775!
294	err, errStr, sysErr, sysErrStr);
295	} else {
296	sNTrace(pData->pSyncNode, "wal read error, index:%" PRId64 ", err:0x%x, msg:%s, syserr:%d, sysmsg:%s", index, err,	×
297	errStr, sysErr, sysErrStr);
298	}
299
300	/*
301	int32_t saveErr = terrno;
302	walCloseReadHandle(pWalHandle);
303	terrno = saveErr;
304	*/
305
306	(void)taosThreadMutexUnlock(&(pData->mutex));	775✔
307
308	TAOS_RETURN(code);	775✔
309	}
310
311	*ppEntry = syncEntryBuild(pWalHandle->pHead->head.bodyLen);	518,485✔
312	if (*ppEntry == NULL) return TSDB_CODE_SYN_INTERNAL_ERROR;	518,439!
313	(*ppEntry)->msgType = TDMT_SYNC_CLIENT_REQUEST;	518,439✔
314	(*ppEntry)->originalRpcType = pWalHandle->pHead->head.msgType;	518,439✔
315	(*ppEntry)->seqNum = pWalHandle->pHead->head.syncMeta.seqNum;	518,439✔
316	(*ppEntry)->isWeak = pWalHandle->pHead->head.syncMeta.isWeek;	518,439✔
317	(*ppEntry)->term = pWalHandle->pHead->head.syncMeta.term;	518,439✔
318	(*ppEntry)->index = index;	518,439✔
319	if ((*ppEntry)->dataLen != pWalHandle->pHead->head.bodyLen) return TSDB_CODE_SYN_INTERNAL_ERROR;	518,439!
320	(void)memcpy((*ppEntry)->data, pWalHandle->pHead->head.body, pWalHandle->pHead->head.bodyLen);	518,439✔
321
322	/*
323	int32_t saveErr = terrno;
324	walCloseReadHandle(pWalHandle);
325	terrno = saveErr;
326	*/
327
328	(void)taosThreadMutexUnlock(&(pData->mutex));	518,439✔
329	int64_t ts4 = taosGetTimestampNs();	518,486✔
330
331	int64_t tsElapsed = ts4 - ts1;	518,486✔
332	int64_t tsElapsedLock = ts2 - ts1;	518,486✔
333	int64_t tsElapsedRead = ts3 - ts2;	518,486✔
334	int64_t tsElapsedBuild = ts4 - ts3;	518,486✔
335
336	sNTrace(pData->pSyncNode,	518,486✔
337	"read index:%" PRId64 ", elapsed:%" PRId64 ", elapsed-lock:%" PRId64 ", elapsed-read:%" PRId64
338	", elapsed-build:%" PRId64,
339	index, tsElapsed, tsElapsedLock, tsElapsedRead, tsElapsedBuild);
340
341	TAOS_RETURN(code);	518,486✔
342	}
343
344	// truncate semantic
345	static int32_t raftLogTruncate(struct SSyncLogStore* pLogStore, SyncIndex fromIndex) {	4✔
346	SSyncLogStoreData* pData = pLogStore->data;	4✔
347	SWal* pWal = pData->pWal;	4✔
348
349	int32_t code = walRollback(pWal, fromIndex);	4✔
350	if (code != 0) {	4!
351	int32_t err = code;	×
352	const char* errStr = tstrerror(err);	×
353	int32_t sysErr = ERRNO;	×
354	const char* sysErrStr = strerror(ERRNO);	×
355	sError("vgId:%d, wal truncate error, from-index:%" PRId64 ", err:0x%x, msg:%s, syserr:%d, sysmsg:%s",	×
356	pData->pSyncNode->vgId, fromIndex, err, errStr, sysErr, sysErrStr);
357	}
358
359	// event log
360	sNTrace(pData->pSyncNode, "log truncate, from-index:%" PRId64, fromIndex);	4!
361
362	TAOS_RETURN(code);	4✔
363	}
364
365	// entry found, return 0
366	// entry not found, return -1, terrno = TSDB_CODE_WAL_LOG_NOT_EXIST
367	// other error, return -1
368	static int32_t raftLogGetLastEntry(SSyncLogStore* pLogStore, SSyncRaftEntry** ppLastEntry) {	5,860✔
369	SSyncLogStoreData* pData = pLogStore->data;	5,860✔
370	SWal* pWal = pData->pWal;	5,860✔
371	if (ppLastEntry == NULL) return TSDB_CODE_SYN_INTERNAL_ERROR;	5,860!
372
373	*ppLastEntry = NULL;	5,860✔
374	if (walIsEmpty(pWal)) {	5,860!
375	TAOS_RETURN(TSDB_CODE_WAL_LOG_NOT_EXIST);	×
376	} else {
377	SyncIndex lastIndex = raftLogLastIndex(pLogStore);	5,860✔
378	if (!(lastIndex >= SYNC_INDEX_BEGIN)) return TSDB_CODE_SYN_INTERNAL_ERROR;	5,860!
379	int32_t code = raftLogGetEntry(pLogStore, lastIndex, ppLastEntry);	5,860✔
380
381	TAOS_RETURN(code);	5,860✔
382	}
383
384	TAOS_RETURN(TSDB_CODE_FAILED);
385	}
386
387	int32_t raftLogUpdateCommitIndex(SSyncLogStore* pLogStore, SyncIndex index) {	2,577,327✔
388	SSyncLogStoreData* pData = pLogStore->data;	2,577,327✔
389	SWal* pWal = pData->pWal;	2,577,327✔
390
391	// need not update
392	SyncIndex snapshotVer = walGetSnapshotVer(pWal);	2,577,327✔
393	SyncIndex walCommitVer = walGetCommittedVer(pWal);	2,577,390✔
394	SyncIndex wallastVer = walGetLastVer(pWal);	2,577,455✔
395
396	if (index < snapshotVer \|\| index > wallastVer) {	2,577,493!
397	// ignore
398	TAOS_RETURN(TSDB_CODE_SUCCESS);	×
399	}
400
401	int32_t code = walCommit(pWal, index);	2,577,499✔
402	if (code != 0) {	2,577,462!
403	int32_t err = code;	×
404	const char* errStr = tstrerror(err);	×
405	int32_t sysErr = ERRNO;	×
406	const char* sysErrStr = strerror(ERRNO);	×
407	sError("vgId:%d, index:%" PRId64 ", raft entry update commit index error, code:0x%x msg:%s syserr:%d sysmsg:%s",	×
408	pData->pSyncNode->vgId, index, err, errStr, sysErr, sysErrStr);
409
410	TAOS_RETURN(code);	×
411	}
412
413	TAOS_RETURN(TSDB_CODE_SUCCESS);	2,577,462✔
414	}
415
416	SyncIndex raftlogCommitIndex(SSyncLogStore* pLogStore) {	15,602✔
417	SSyncLogStoreData* pData = pLogStore->data;	15,602✔
418	return pData->pSyncNode->commitIndex;	15,602✔
419	}
420
421	SyncIndex logStoreFirstIndex(SSyncLogStore* pLogStore) {	×
422	SSyncLogStoreData* pData = pLogStore->data;	×
423	SWal* pWal = pData->pWal;	×
424	return walGetFirstVer(pWal);	×
425	}
426
427	SyncIndex logStoreWalCommitVer(SSyncLogStore* pLogStore) {	×
428	SSyncLogStoreData* pData = pLogStore->data;	×
429	SWal* pWal = pData->pWal;	×
430
431	return walGetCommittedVer(pWal);	×
432	}

taosdata / TDengine / #3942

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