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

705 lines
24 KiB
C
Raw Normal View History

2022-02-22 03:28:15 +00:00
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "syncRaftLog.h"
2022-06-10 05:15:43 +00:00
#include "syncRaftCfg.h"
#include "syncRaftStore.h"
2022-02-22 03:28:15 +00:00
//-------------------------------
// log[m .. n]
// public function
2022-06-29 06:02:40 +00:00
static int32_t raftLogRestoreFromSnapshot(struct SSyncLogStore* pLogStore, SyncIndex snapshotIndex);
2022-06-05 11:47:54 +00:00
static SyncIndex raftLogBeginIndex(struct SSyncLogStore* pLogStore);
static SyncIndex raftLogEndIndex(struct SSyncLogStore* pLogStore);
2022-06-05 13:48:50 +00:00
static SyncIndex raftLogWriteIndex(struct SSyncLogStore* pLogStore);
2022-06-05 11:47:54 +00:00
static bool raftLogIsEmpty(struct SSyncLogStore* pLogStore);
static int32_t raftLogEntryCount(struct SSyncLogStore* pLogStore);
static SyncIndex raftLogLastIndex(struct SSyncLogStore* pLogStore);
static SyncTerm raftLogLastTerm(struct SSyncLogStore* pLogStore);
static int32_t raftLogAppendEntry(struct SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry);
static int32_t raftLogGetEntry(struct SSyncLogStore* pLogStore, SyncIndex index, SSyncRaftEntry** ppEntry);
static int32_t raftLogTruncate(struct SSyncLogStore* pLogStore, SyncIndex fromIndex);
static bool raftLogExist(struct SSyncLogStore* pLogStore, SyncIndex index);
2022-06-05 11:47:54 +00:00
// private function
2022-06-05 11:47:54 +00:00
static int32_t raftLogGetLastEntry(SSyncLogStore* pLogStore, SSyncRaftEntry** ppLastEntry);
//-------------------------------
// log[0 .. n]
static SSyncRaftEntry* logStoreGetLastEntry(SSyncLogStore* pLogStore);
2022-05-27 08:36:44 +00:00
static SyncIndex logStoreLastIndex(SSyncLogStore* pLogStore);
static SyncTerm logStoreLastTerm(SSyncLogStore* pLogStore);
static SSyncRaftEntry* logStoreGetEntry(SSyncLogStore* pLogStore, SyncIndex index);
static int32_t logStoreAppendEntry(SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry);
static int32_t logStoreTruncate(SSyncLogStore* pLogStore, SyncIndex fromIndex);
static int32_t logStoreUpdateCommitIndex(SSyncLogStore* pLogStore, SyncIndex index);
static SyncIndex logStoreGetCommitIndex(SSyncLogStore* pLogStore);
//-------------------------------
SSyncLogStore* logStoreCreate(SSyncNode* pSyncNode) {
SSyncLogStore* pLogStore = taosMemoryMalloc(sizeof(SSyncLogStore));
ASSERT(pLogStore != NULL);
pLogStore->data = taosMemoryMalloc(sizeof(SSyncLogStoreData));
ASSERT(pLogStore->data != NULL);
SSyncLogStoreData* pData = pLogStore->data;
pData->pSyncNode = pSyncNode;
pData->pWal = pSyncNode->pWal;
ASSERT(pData->pWal != NULL);
taosThreadMutexInit(&(pData->mutex), NULL);
2022-07-07 03:42:45 +00:00
pData->pWalHandle = walOpenReader(pData->pWal, NULL);
ASSERT(pData->pWalHandle != NULL);
pLogStore->appendEntry = logStoreAppendEntry;
pLogStore->getEntry = logStoreGetEntry;
pLogStore->truncate = logStoreTruncate;
pLogStore->getLastIndex = logStoreLastIndex;
pLogStore->getLastTerm = logStoreLastTerm;
pLogStore->updateCommitIndex = logStoreUpdateCommitIndex;
pLogStore->getCommitIndex = logStoreGetCommitIndex;
pLogStore->syncLogRestoreFromSnapshot = raftLogRestoreFromSnapshot;
pLogStore->syncLogBeginIndex = raftLogBeginIndex;
pLogStore->syncLogEndIndex = raftLogEndIndex;
pLogStore->syncLogIsEmpty = raftLogIsEmpty;
pLogStore->syncLogEntryCount = raftLogEntryCount;
pLogStore->syncLogLastIndex = raftLogLastIndex;
pLogStore->syncLogLastTerm = raftLogLastTerm;
pLogStore->syncLogAppendEntry = raftLogAppendEntry;
pLogStore->syncLogGetEntry = raftLogGetEntry;
pLogStore->syncLogTruncate = raftLogTruncate;
pLogStore->syncLogWriteIndex = raftLogWriteIndex;
pLogStore->syncLogExist = raftLogExist;
return pLogStore;
}
void logStoreDestory(SSyncLogStore* pLogStore) {
if (pLogStore != NULL) {
SSyncLogStoreData* pData = pLogStore->data;
taosThreadMutexLock(&(pData->mutex));
if (pData->pWalHandle != NULL) {
2022-07-07 03:42:45 +00:00
walCloseReader(pData->pWalHandle);
pData->pWalHandle = NULL;
}
taosThreadMutexUnlock(&(pData->mutex));
taosThreadMutexDestroy(&(pData->mutex));
taosMemoryFree(pLogStore->data);
taosMemoryFree(pLogStore);
}
}
//-------------------------------
// log[m .. n]
2022-06-29 06:02:40 +00:00
static int32_t raftLogRestoreFromSnapshot(struct SSyncLogStore* pLogStore, SyncIndex snapshotIndex) {
ASSERT(snapshotIndex >= 0);
2022-06-05 11:47:54 +00:00
SSyncLogStoreData* pData = pLogStore->data;
SWal* pWal = pData->pWal;
2022-06-29 06:02:40 +00:00
int32_t code = walRestoreFromSnapshot(pWal, snapshotIndex);
if (code != 0) {
int32_t err = terrno;
const char* errStr = tstrerror(err);
int32_t sysErr = errno;
const char* sysErrStr = strerror(errno);
2022-06-23 06:07:18 +00:00
2022-06-29 06:02:40 +00:00
char logBuf[128];
snprintf(logBuf, sizeof(logBuf),
"wal restore from snapshot error, index:%ld, err:%d %X, msg:%s, syserr:%d, sysmsg:%s", snapshotIndex, err,
err, errStr, sysErr, sysErrStr);
syncNodeErrorLog(pData->pSyncNode, logBuf);
return -1;
2022-06-29 06:02:40 +00:00
}
return 0;
}
2022-06-05 11:47:54 +00:00
static SyncIndex raftLogBeginIndex(struct SSyncLogStore* pLogStore) {
SSyncLogStoreData* pData = pLogStore->data;
SWal* pWal = pData->pWal;
SyncIndex firstVer = walGetFirstVer(pWal);
return firstVer;
2022-06-05 11:47:54 +00:00
}
static SyncIndex raftLogEndIndex(struct SSyncLogStore* pLogStore) { return raftLogLastIndex(pLogStore); }
static bool raftLogIsEmpty(struct SSyncLogStore* pLogStore) {
SSyncLogStoreData* pData = pLogStore->data;
SWal* pWal = pData->pWal;
return walIsEmpty(pWal);
2022-06-05 11:47:54 +00:00
}
static int32_t raftLogEntryCount(struct SSyncLogStore* pLogStore) {
SyncIndex beginIndex = raftLogBeginIndex(pLogStore);
SyncIndex endIndex = raftLogEndIndex(pLogStore);
2022-06-05 13:48:50 +00:00
int32_t count = endIndex - beginIndex + 1;
2022-06-05 11:47:54 +00:00
return count > 0 ? count : 0;
}
static SyncIndex raftLogLastIndex(struct SSyncLogStore* pLogStore) {
2022-06-05 13:48:50 +00:00
SyncIndex lastIndex;
2022-06-05 11:47:54 +00:00
SSyncLogStoreData* pData = pLogStore->data;
SWal* pWal = pData->pWal;
SyncIndex lastVer = walGetLastVer(pWal);
2022-06-05 13:48:50 +00:00
return lastVer;
2022-06-05 13:48:50 +00:00
}
static SyncIndex raftLogWriteIndex(struct SSyncLogStore* pLogStore) {
SSyncLogStoreData* pData = pLogStore->data;
SWal* pWal = pData->pWal;
SyncIndex lastVer = walGetLastVer(pWal);
return lastVer + 1;
2022-06-05 11:47:54 +00:00
}
static bool raftLogExist(struct SSyncLogStore* pLogStore, SyncIndex index) {
SSyncLogStoreData* pData = pLogStore->data;
SWal* pWal = pData->pWal;
bool b = walLogExist(pWal, index);
return b;
}
2022-06-29 07:44:30 +00:00
// if success, return last term
// if not log, return 0
// if error, return SYNC_TERM_INVALID
static SyncTerm raftLogLastTerm(struct SSyncLogStore* pLogStore) {
SSyncLogStoreData* pData = pLogStore->data;
SWal* pWal = pData->pWal;
if (walIsEmpty(pWal)) {
return 0;
} else {
SSyncRaftEntry* pLastEntry;
int32_t code = raftLogGetLastEntry(pLogStore, &pLastEntry);
2022-06-29 07:44:30 +00:00
if (code == 0 && pLastEntry != NULL) {
SyncTerm lastTerm = pLastEntry->term;
taosMemoryFree(pLastEntry);
return lastTerm;
} else {
return SYNC_TERM_INVALID;
}
}
2022-06-29 07:44:30 +00:00
// can not be here!
return SYNC_TERM_INVALID;
}
2022-06-05 11:47:54 +00:00
static int32_t raftLogAppendEntry(struct SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry) {
SSyncLogStoreData* pData = pLogStore->data;
SWal* pWal = pData->pWal;
2022-06-05 13:48:50 +00:00
SyncIndex writeIndex = raftLogWriteIndex(pLogStore);
if (pEntry->index != writeIndex) {
2022-06-23 03:59:28 +00:00
sError("vgId:%d wal write index error, entry-index:%ld update to %ld", pData->pSyncNode->vgId, pEntry->index,
writeIndex);
pEntry->index = writeIndex;
}
2022-06-05 11:47:54 +00:00
int code = 0;
SSyncLogMeta syncMeta;
syncMeta.isWeek = pEntry->isWeak;
syncMeta.seqNum = pEntry->seqNum;
syncMeta.term = pEntry->term;
code = walWriteWithSyncInfo(pWal, pEntry->index, pEntry->originalRpcType, syncMeta, pEntry->data, pEntry->dataLen);
if (code != 0) {
int32_t err = terrno;
const char* errStr = tstrerror(err);
2022-06-23 03:59:28 +00:00
int32_t sysErr = errno;
const char* sysErrStr = strerror(errno);
2022-06-23 06:07:18 +00:00
char logBuf[128];
snprintf(logBuf, sizeof(logBuf), "wal write error, index:%ld, err:%d %X, msg:%s, syserr:%d, sysmsg:%s",
pEntry->index, err, err, errStr, sysErr, sysErrStr);
syncNodeErrorLog(pData->pSyncNode, logBuf);
2022-06-05 11:47:54 +00:00
ASSERT(0);
}
2022-06-29 07:44:30 +00:00
// walFsync(pWal, true);
2022-06-29 07:44:30 +00:00
do {
char eventLog[128];
snprintf(eventLog, sizeof(eventLog), "write index:%ld, type:%s,%d, type2:%s,%d", pEntry->index,
TMSG_INFO(pEntry->msgType), pEntry->msgType, TMSG_INFO(pEntry->originalRpcType), pEntry->originalRpcType);
syncNodeEventLog(pData->pSyncNode, eventLog);
} while (0);
2022-06-05 11:47:54 +00:00
return code;
}
2022-06-29 07:44:30 +00:00
// entry found, return 0
// entry not found, return -1, terrno = TSDB_CODE_WAL_LOG_NOT_EXIST
// other error, return -1
2022-06-21 09:45:08 +00:00
static int32_t raftLogGetEntry(struct SSyncLogStore* pLogStore, SyncIndex index, SSyncRaftEntry** ppEntry) {
SSyncLogStoreData* pData = pLogStore->data;
SWal* pWal = pData->pWal;
int32_t code;
*ppEntry = NULL;
// SWalReadHandle* pWalHandle = walOpenReadHandle(pWal);
2022-07-07 03:42:45 +00:00
SWalReader* pWalHandle = pData->pWalHandle;
2022-06-21 09:45:08 +00:00
if (pWalHandle == NULL) {
2022-06-29 07:44:30 +00:00
terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
2022-06-21 09:45:08 +00:00
return -1;
}
2022-06-27 02:36:28 +00:00
taosThreadMutexLock(&(pData->mutex));
2022-07-07 03:42:45 +00:00
code = walReadVer(pWalHandle, index);
2022-06-21 09:45:08 +00:00
if (code != 0) {
int32_t err = terrno;
const char* errStr = tstrerror(err);
2022-06-23 03:59:28 +00:00
int32_t sysErr = errno;
const char* sysErrStr = strerror(errno);
2022-06-23 06:07:18 +00:00
do {
char logBuf[128];
snprintf(logBuf, sizeof(logBuf), "wal read error, index:%ld, err:%d %X, msg:%s, syserr:%d, sysmsg:%s", index, err,
err, errStr, sysErr, sysErrStr);
if (terrno == TSDB_CODE_WAL_LOG_NOT_EXIST) {
syncNodeEventLog(pData->pSyncNode, logBuf);
} else {
syncNodeErrorLog(pData->pSyncNode, logBuf);
}
} while (0);
2022-06-21 09:45:08 +00:00
/*
int32_t saveErr = terrno;
walCloseReadHandle(pWalHandle);
terrno = saveErr;
*/
2022-06-22 12:23:49 +00:00
2022-06-27 02:36:28 +00:00
taosThreadMutexUnlock(&(pData->mutex));
2022-06-21 09:45:08 +00:00
return code;
}
*ppEntry = syncEntryBuild(pWalHandle->pHead->head.bodyLen);
ASSERT(*ppEntry != NULL);
(*ppEntry)->msgType = TDMT_SYNC_CLIENT_REQUEST;
(*ppEntry)->originalRpcType = pWalHandle->pHead->head.msgType;
(*ppEntry)->seqNum = pWalHandle->pHead->head.syncMeta.seqNum;
(*ppEntry)->isWeak = pWalHandle->pHead->head.syncMeta.isWeek;
(*ppEntry)->term = pWalHandle->pHead->head.syncMeta.term;
(*ppEntry)->index = index;
ASSERT((*ppEntry)->dataLen == pWalHandle->pHead->head.bodyLen);
memcpy((*ppEntry)->data, pWalHandle->pHead->head.body, pWalHandle->pHead->head.bodyLen);
/*
int32_t saveErr = terrno;
walCloseReadHandle(pWalHandle);
terrno = saveErr;
*/
2022-06-21 09:45:08 +00:00
2022-06-27 02:36:28 +00:00
taosThreadMutexUnlock(&(pData->mutex));
2022-06-21 09:45:08 +00:00
return code;
}
2022-06-05 11:47:54 +00:00
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 = terrno;
const char* errStr = tstrerror(err);
2022-06-23 03:59:28 +00:00
int32_t sysErr = errno;
const char* sysErrStr = strerror(errno);
sError("vgId:%d wal truncate error, from-index:%ld, err:%d %X, msg:%s, syserr:%d, sysmsg:%s",
pData->pSyncNode->vgId, fromIndex, err, err, errStr, sysErr, sysErrStr);
2022-06-05 11:47:54 +00:00
ASSERT(0);
}
2022-07-08 06:47:00 +00:00
// event log
do {
char logBuf[128];
snprintf(logBuf, sizeof(logBuf), "wal truncate, from-index:%ld", fromIndex);
syncNodeEventLog(pData->pSyncNode, logBuf);
} while (0);
2022-06-05 11:47:54 +00:00
return code;
}
2022-06-29 07:44:30 +00:00
// 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;
ASSERT(ppLastEntry != NULL);
*ppLastEntry = NULL;
if (walIsEmpty(pWal)) {
terrno = TSDB_CODE_WAL_LOG_NOT_EXIST;
return -1;
} else {
SyncIndex lastIndex = raftLogLastIndex(pLogStore);
2022-06-29 07:44:30 +00:00
ASSERT(lastIndex >= SYNC_INDEX_BEGIN);
int32_t code = raftLogGetEntry(pLogStore, lastIndex, ppLastEntry);
return code;
}
return -1;
}
2022-06-05 11:47:54 +00:00
//-------------------------------
// log[0 .. n]
2022-03-09 12:04:39 +00:00
int32_t logStoreAppendEntry(SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry) {
SSyncLogStoreData* pData = pLogStore->data;
SWal* pWal = pData->pWal;
2022-03-24 09:30:50 +00:00
SyncIndex lastIndex = logStoreLastIndex(pLogStore);
2022-06-21 08:02:36 +00:00
ASSERT(pEntry->index == lastIndex + 1);
2022-03-10 06:34:02 +00:00
2022-04-18 13:50:56 +00:00
int code = 0;
SSyncLogMeta syncMeta;
syncMeta.isWeek = pEntry->isWeak;
syncMeta.seqNum = pEntry->seqNum;
syncMeta.term = pEntry->term;
code = walWriteWithSyncInfo(pWal, pEntry->index, pEntry->originalRpcType, syncMeta, pEntry->data, pEntry->dataLen);
2022-05-06 15:40:34 +00:00
if (code != 0) {
2022-05-14 10:12:53 +00:00
int32_t err = terrno;
const char* errStr = tstrerror(err);
2022-06-23 03:59:28 +00:00
int32_t sysErr = errno;
const char* sysErrStr = strerror(errno);
2022-06-23 06:07:18 +00:00
char logBuf[128];
snprintf(logBuf, sizeof(logBuf), "wal write error, index:%ld, err:%d %X, msg:%s, syserr:%d, sysmsg:%s",
pEntry->index, err, err, errStr, sysErr, sysErrStr);
syncNodeErrorLog(pData->pSyncNode, logBuf);
2022-06-23 03:59:28 +00:00
2022-05-06 15:40:34 +00:00
ASSERT(0);
2022-05-14 10:12:53 +00:00
}
2022-03-09 12:04:39 +00:00
2022-06-29 07:44:30 +00:00
// walFsync(pWal, true);
2022-06-18 07:17:58 +00:00
char eventLog[128];
snprintf(eventLog, sizeof(eventLog), "old write index:%ld, type:%s,%d, type2:%s,%d", pEntry->index,
TMSG_INFO(pEntry->msgType), pEntry->msgType, TMSG_INFO(pEntry->originalRpcType), pEntry->originalRpcType);
syncNodeEventLog(pData->pSyncNode, eventLog);
2022-03-24 03:40:36 +00:00
return code;
2022-03-09 12:04:39 +00:00
}
2022-03-09 06:51:02 +00:00
2022-03-09 12:04:39 +00:00
SSyncRaftEntry* logStoreGetEntry(SSyncLogStore* pLogStore, SyncIndex index) {
SSyncLogStoreData* pData = pLogStore->data;
SWal* pWal = pData->pWal;
2022-03-16 12:03:22 +00:00
if (index >= SYNC_INDEX_BEGIN && index <= logStoreLastIndex(pLogStore)) {
2022-06-27 02:36:28 +00:00
taosThreadMutexLock(&(pData->mutex));
// SWalReadHandle* pWalHandle = walOpenReadHandle(pWal);
2022-07-07 03:42:45 +00:00
SWalReader* pWalHandle = pData->pWalHandle;
2022-05-27 10:00:09 +00:00
ASSERT(pWalHandle != NULL);
2022-07-07 03:42:45 +00:00
int32_t code = walReadVer(pWalHandle, index);
2022-05-06 15:40:34 +00:00
if (code != 0) {
2022-05-14 10:12:53 +00:00
int32_t err = terrno;
const char* errStr = tstrerror(err);
2022-06-23 03:59:28 +00:00
int32_t sysErr = errno;
const char* sysErrStr = strerror(errno);
2022-06-23 06:07:18 +00:00
do {
char logBuf[128];
snprintf(logBuf, sizeof(logBuf), "wal read error, index:%ld, err:%d %X, msg:%s, syserr:%d, sysmsg:%s", index,
err, err, errStr, sysErr, sysErrStr);
if (terrno == TSDB_CODE_WAL_LOG_NOT_EXIST) {
syncNodeEventLog(pData->pSyncNode, logBuf);
} else {
syncNodeErrorLog(pData->pSyncNode, logBuf);
}
} while (0);
2022-06-23 03:59:28 +00:00
2022-05-06 15:40:34 +00:00
ASSERT(0);
2022-05-14 10:12:53 +00:00
}
2022-04-18 13:50:56 +00:00
2022-04-24 05:42:54 +00:00
SSyncRaftEntry* pEntry = syncEntryBuild(pWalHandle->pHead->head.bodyLen);
2022-06-21 08:02:36 +00:00
ASSERT(pEntry != NULL);
2022-03-16 12:03:22 +00:00
pEntry->msgType = TDMT_SYNC_CLIENT_REQUEST;
2022-04-18 13:50:56 +00:00
pEntry->originalRpcType = pWalHandle->pHead->head.msgType;
pEntry->seqNum = pWalHandle->pHead->head.syncMeta.seqNum;
pEntry->isWeak = pWalHandle->pHead->head.syncMeta.isWeek;
pEntry->term = pWalHandle->pHead->head.syncMeta.term;
pEntry->index = index;
2022-06-21 08:02:36 +00:00
ASSERT(pEntry->dataLen == pWalHandle->pHead->head.bodyLen);
2022-04-24 05:42:54 +00:00
memcpy(pEntry->data, pWalHandle->pHead->head.body, pWalHandle->pHead->head.bodyLen);
2022-04-18 13:50:56 +00:00
/*
int32_t saveErr = terrno;
walCloseReadHandle(pWalHandle);
terrno = saveErr;
*/
2022-06-22 12:23:49 +00:00
2022-06-27 02:36:28 +00:00
taosThreadMutexUnlock(&(pData->mutex));
2022-04-18 13:50:56 +00:00
return pEntry;
2022-03-09 12:04:39 +00:00
2022-04-18 13:50:56 +00:00
} else {
return NULL;
}
2022-03-09 12:04:39 +00:00
}
2022-03-09 06:51:02 +00:00
2022-03-09 12:04:39 +00:00
int32_t logStoreTruncate(SSyncLogStore* pLogStore, SyncIndex fromIndex) {
SSyncLogStoreData* pData = pLogStore->data;
SWal* pWal = pData->pWal;
2022-06-21 08:02:36 +00:00
// ASSERT(walRollback(pWal, fromIndex) == 0);
2022-05-10 04:20:13 +00:00
int32_t code = walRollback(pWal, fromIndex);
if (code != 0) {
2022-05-14 10:12:53 +00:00
int32_t err = terrno;
const char* errStr = tstrerror(err);
2022-06-23 03:59:28 +00:00
int32_t sysErr = errno;
const char* sysErrStr = strerror(errno);
sError("vgId:%d wal truncate error, from-index:%ld, err:%d %X, msg:%s, syserr:%d, sysmsg:%s",
pData->pSyncNode->vgId, fromIndex, err, err, errStr, sysErr, sysErrStr);
2022-05-10 04:20:13 +00:00
ASSERT(0);
2022-05-14 10:12:53 +00:00
}
2022-07-08 06:47:00 +00:00
// event log
do {
char logBuf[128];
snprintf(logBuf, sizeof(logBuf), "wal truncate, from-index:%ld", fromIndex);
syncNodeEventLog(pData->pSyncNode, logBuf);
} while (0);
2022-06-05 11:47:54 +00:00
return 0;
2022-03-09 12:04:39 +00:00
}
2022-02-22 03:28:15 +00:00
2022-03-09 12:04:39 +00:00
SyncIndex logStoreLastIndex(SSyncLogStore* pLogStore) {
2022-03-10 06:34:02 +00:00
SSyncLogStoreData* pData = pLogStore->data;
SWal* pWal = pData->pWal;
2022-03-10 08:18:16 +00:00
SyncIndex lastIndex = walGetLastVer(pWal);
2022-03-09 12:04:39 +00:00
return lastIndex;
}
2022-02-22 03:28:15 +00:00
2022-03-09 12:04:39 +00:00
SyncTerm logStoreLastTerm(SSyncLogStore* pLogStore) {
2022-03-10 08:18:16 +00:00
SyncTerm lastTerm = 0;
2022-03-09 12:04:39 +00:00
SSyncRaftEntry* pLastEntry = logStoreGetLastEntry(pLogStore);
2022-03-10 08:18:16 +00:00
if (pLastEntry != NULL) {
lastTerm = pLastEntry->term;
2022-03-25 16:29:53 +00:00
taosMemoryFree(pLastEntry);
2022-03-10 08:18:16 +00:00
}
2022-03-09 12:04:39 +00:00
return lastTerm;
}
2022-03-09 06:51:02 +00:00
2022-03-09 12:04:39 +00:00
int32_t logStoreUpdateCommitIndex(SSyncLogStore* pLogStore, SyncIndex index) {
2022-06-17 03:37:10 +00:00
SSyncLogStoreData* pData = pLogStore->data;
SWal* pWal = pData->pWal;
2022-06-21 08:02:36 +00:00
// ASSERT(walCommit(pWal, index) == 0);
2022-06-17 03:37:10 +00:00
int32_t code = walCommit(pWal, index);
if (code != 0) {
int32_t err = terrno;
const char* errStr = tstrerror(err);
2022-06-23 03:59:28 +00:00
int32_t sysErr = errno;
const char* sysErrStr = strerror(errno);
sError("vgId:%d wal update commit index error, index:%ld, err:%d %X, msg:%s, syserr:%d, sysmsg:%s",
pData->pSyncNode->vgId, index, err, err, errStr, sysErr, sysErrStr);
2022-06-17 03:37:10 +00:00
ASSERT(0);
}
2022-06-05 11:47:54 +00:00
return 0;
2022-03-09 12:04:39 +00:00
}
2022-03-09 06:51:02 +00:00
2022-03-09 12:04:39 +00:00
SyncIndex logStoreGetCommitIndex(SSyncLogStore* pLogStore) {
SSyncLogStoreData* pData = pLogStore->data;
return pData->pSyncNode->commitIndex;
}
SSyncRaftEntry* logStoreGetLastEntry(SSyncLogStore* pLogStore) {
SSyncLogStoreData* pData = pLogStore->data;
SWal* pWal = pData->pWal;
SyncIndex lastIndex = walGetLastVer(pWal);
2022-03-10 08:18:16 +00:00
SSyncRaftEntry* pEntry = NULL;
if (lastIndex > 0) {
pEntry = logStoreGetEntry(pLogStore, lastIndex);
}
2022-03-09 12:04:39 +00:00
return pEntry;
}
2022-03-09 06:51:02 +00:00
cJSON* logStore2Json(SSyncLogStore* pLogStore) {
char u64buf[128] = {0};
SSyncLogStoreData* pData = (SSyncLogStoreData*)pLogStore->data;
cJSON* pRoot = cJSON_CreateObject();
if (pData != NULL && pData->pWal != NULL) {
snprintf(u64buf, sizeof(u64buf), "%p", pData->pSyncNode);
cJSON_AddStringToObject(pRoot, "pSyncNode", u64buf);
snprintf(u64buf, sizeof(u64buf), "%p", pData->pWal);
cJSON_AddStringToObject(pRoot, "pWal", u64buf);
SyncIndex beginIndex = raftLogBeginIndex(pLogStore);
snprintf(u64buf, sizeof(u64buf), "%ld", beginIndex);
cJSON_AddStringToObject(pRoot, "beginIndex", u64buf);
SyncIndex endIndex = raftLogEndIndex(pLogStore);
snprintf(u64buf, sizeof(u64buf), "%ld", endIndex);
cJSON_AddStringToObject(pRoot, "endIndex", u64buf);
int32_t count = raftLogEntryCount(pLogStore);
cJSON_AddNumberToObject(pRoot, "entryCount", count);
snprintf(u64buf, sizeof(u64buf), "%ld", raftLogWriteIndex(pLogStore));
cJSON_AddStringToObject(pRoot, "WriteIndex", u64buf);
snprintf(u64buf, sizeof(u64buf), "%d", raftLogIsEmpty(pLogStore));
cJSON_AddStringToObject(pRoot, "IsEmpty", u64buf);
snprintf(u64buf, sizeof(u64buf), "%ld", raftLogLastIndex(pLogStore));
cJSON_AddStringToObject(pRoot, "LastIndex", u64buf);
snprintf(u64buf, sizeof(u64buf), "%lu", raftLogLastTerm(pLogStore));
cJSON_AddStringToObject(pRoot, "LastTerm", u64buf);
cJSON* pEntries = cJSON_CreateArray();
cJSON_AddItemToObject(pRoot, "pEntries", pEntries);
if (!raftLogIsEmpty(pLogStore)) {
for (SyncIndex i = beginIndex; i <= endIndex; ++i) {
SSyncRaftEntry* pEntry = logStoreGetEntry(pLogStore, i);
cJSON_AddItemToArray(pEntries, syncEntry2Json(pEntry));
syncEntryDestory(pEntry);
}
}
}
cJSON* pJson = cJSON_CreateObject();
cJSON_AddItemToObject(pJson, "SSyncLogStore", pRoot);
return pJson;
}
2022-03-09 06:51:02 +00:00
char* logStore2Str(SSyncLogStore* pLogStore) {
cJSON* pJson = logStore2Json(pLogStore);
char* serialized = cJSON_Print(pJson);
cJSON_Delete(pJson);
return serialized;
2022-03-10 03:21:04 +00:00
}
2022-03-23 09:08:07 +00:00
cJSON* logStoreSimple2Json(SSyncLogStore* pLogStore) {
char u64buf[128] = {0};
2022-03-23 09:08:07 +00:00
SSyncLogStoreData* pData = (SSyncLogStoreData*)pLogStore->data;
cJSON* pRoot = cJSON_CreateObject();
if (pData != NULL && pData->pWal != NULL) {
snprintf(u64buf, sizeof(u64buf), "%p", pData->pSyncNode);
cJSON_AddStringToObject(pRoot, "pSyncNode", u64buf);
snprintf(u64buf, sizeof(u64buf), "%p", pData->pWal);
cJSON_AddStringToObject(pRoot, "pWal", u64buf);
2022-06-05 11:47:54 +00:00
SyncIndex beginIndex = raftLogBeginIndex(pLogStore);
snprintf(u64buf, sizeof(u64buf), "%ld", beginIndex);
2022-06-05 11:47:54 +00:00
cJSON_AddStringToObject(pRoot, "beginIndex", u64buf);
2022-06-06 12:02:27 +00:00
SyncIndex endIndex = raftLogEndIndex(pLogStore);
snprintf(u64buf, sizeof(u64buf), "%ld", endIndex);
cJSON_AddStringToObject(pRoot, "endIndex", u64buf);
int32_t count = raftLogEntryCount(pLogStore);
cJSON_AddNumberToObject(pRoot, "entryCount", count);
2022-06-08 08:45:40 +00:00
snprintf(u64buf, sizeof(u64buf), "%ld", raftLogWriteIndex(pLogStore));
cJSON_AddStringToObject(pRoot, "WriteIndex", u64buf);
snprintf(u64buf, sizeof(u64buf), "%d", raftLogIsEmpty(pLogStore));
cJSON_AddStringToObject(pRoot, "IsEmpty", u64buf);
snprintf(u64buf, sizeof(u64buf), "%ld", raftLogLastIndex(pLogStore));
cJSON_AddStringToObject(pRoot, "LastIndex", u64buf);
snprintf(u64buf, sizeof(u64buf), "%lu", raftLogLastTerm(pLogStore));
cJSON_AddStringToObject(pRoot, "LastTerm", u64buf);
2022-03-23 09:08:07 +00:00
}
cJSON* pJson = cJSON_CreateObject();
cJSON_AddItemToObject(pJson, "SSyncLogStoreSimple", pRoot);
return pJson;
}
char* logStoreSimple2Str(SSyncLogStore* pLogStore) {
cJSON* pJson = logStoreSimple2Json(pLogStore);
char* serialized = cJSON_Print(pJson);
cJSON_Delete(pJson);
return serialized;
}
2022-06-02 11:47:06 +00:00
SyncIndex logStoreFirstIndex(SSyncLogStore* pLogStore) {
SSyncLogStoreData* pData = pLogStore->data;
SWal* pWal = pData->pWal;
return walGetFirstVer(pWal);
}
2022-03-10 08:54:26 +00:00
// for debug -----------------
2022-03-10 03:21:04 +00:00
void logStorePrint(SSyncLogStore* pLogStore) {
2022-03-10 08:54:26 +00:00
char* serialized = logStore2Str(pLogStore);
printf("logStorePrint | len:%lu | %s \n", strlen(serialized), serialized);
fflush(NULL);
2022-03-25 16:29:53 +00:00
taosMemoryFree(serialized);
2022-03-10 08:54:26 +00:00
}
void logStorePrint2(char* s, SSyncLogStore* pLogStore) {
char* serialized = logStore2Str(pLogStore);
2022-04-18 13:50:56 +00:00
printf("logStorePrint2 | len:%lu | %s | %s \n", strlen(serialized), s, serialized);
2022-03-10 08:54:26 +00:00
fflush(NULL);
2022-03-25 16:29:53 +00:00
taosMemoryFree(serialized);
2022-03-10 08:54:26 +00:00
}
2022-03-10 08:18:16 +00:00
2022-03-10 08:54:26 +00:00
void logStoreLog(SSyncLogStore* pLogStore) {
2022-06-11 04:44:58 +00:00
if (gRaftDetailLog) {
char* serialized = logStore2Str(pLogStore);
sTraceLong("logStoreLog | len:%lu | %s", strlen(serialized), serialized);
taosMemoryFree(serialized);
}
2022-03-10 08:54:26 +00:00
}
void logStoreLog2(char* s, SSyncLogStore* pLogStore) {
2022-06-11 04:44:58 +00:00
if (gRaftDetailLog) {
char* serialized = logStore2Str(pLogStore);
sTraceLong("logStoreLog2 | len:%lu | %s | %s", strlen(serialized), s, serialized);
taosMemoryFree(serialized);
}
}
2022-03-23 09:08:07 +00:00
// for debug -----------------
void logStoreSimplePrint(SSyncLogStore* pLogStore) {
char* serialized = logStoreSimple2Str(pLogStore);
printf("logStoreSimplePrint | len:%lu | %s \n", strlen(serialized), serialized);
fflush(NULL);
2022-03-25 16:29:53 +00:00
taosMemoryFree(serialized);
2022-03-23 09:08:07 +00:00
}
void logStoreSimplePrint2(char* s, SSyncLogStore* pLogStore) {
char* serialized = logStoreSimple2Str(pLogStore);
printf("logStoreSimplePrint2 | len:%lu | %s | %s \n", strlen(serialized), s, serialized);
fflush(NULL);
2022-03-25 16:29:53 +00:00
taosMemoryFree(serialized);
2022-03-23 09:08:07 +00:00
}
void logStoreSimpleLog(SSyncLogStore* pLogStore) {
char* serialized = logStoreSimple2Str(pLogStore);
sTrace("logStoreSimpleLog | len:%lu | %s", strlen(serialized), serialized);
2022-03-25 16:29:53 +00:00
taosMemoryFree(serialized);
2022-03-23 09:08:07 +00:00
}
void logStoreSimpleLog2(char* s, SSyncLogStore* pLogStore) {
2022-06-09 03:18:48 +00:00
if (gRaftDetailLog) {
char* serialized = logStoreSimple2Str(pLogStore);
sTrace("logStoreSimpleLog2 | len:%lu | %s | %s", strlen(serialized), s, serialized);
taosMemoryFree(serialized);
}
2022-04-24 05:42:54 +00:00
}