From 4bd650609de50e88d752e3eeb648496423ef8d32 Mon Sep 17 00:00:00 2001 From: Minghao Li Date: Wed, 29 Jun 2022 14:02:40 +0800 Subject: [PATCH] refactor(sync): delete old functions --- source/libs/sync/src/syncRaftLog.c | 189 +++------------------------- source/libs/sync/src/syncSnapshot.c | 1 + 2 files changed, 18 insertions(+), 172 deletions(-) diff --git a/source/libs/sync/src/syncRaftLog.c b/source/libs/sync/src/syncRaftLog.c index d84c576714..9c4d527aed 100644 --- a/source/libs/sync/src/syncRaftLog.c +++ b/source/libs/sync/src/syncRaftLog.c @@ -18,8 +18,7 @@ #include "syncRaftStore.h" // refactor, log[0 .. n] ==> log[m .. n] -static int32_t raftLogRestoreFromSnapshot(struct SSyncLogStore* pLogStore, SyncIndex snapshotIndex); -// static int32_t raftLogSetBeginIndex(struct SSyncLogStore* pLogStore, SyncIndex beginIndex); +static int32_t raftLogRestoreFromSnapshot(struct SSyncLogStore* pLogStore, SyncIndex snapshotIndex); static SyncIndex raftLogBeginIndex(struct SSyncLogStore* pLogStore); static SyncIndex raftLogEndIndex(struct SSyncLogStore* pLogStore); static SyncIndex raftLogWriteIndex(struct SSyncLogStore* pLogStore); @@ -44,28 +43,27 @@ static int32_t logStoreTruncate(SSyncLogStore* pLogStore, SyncIndex from static int32_t logStoreUpdateCommitIndex(SSyncLogStore* pLogStore, SyncIndex index); static SyncIndex logStoreGetCommitIndex(SSyncLogStore* pLogStore); -// refactor, log[0 .. n] ==> log[m .. n] -/* -static int32_t raftLogSetBeginIndex(struct SSyncLogStore* pLogStore, SyncIndex beginIndex) { - // if beginIndex == 0, donot need call this funciton - ASSERT(beginIndex > 0); - - SSyncLogStoreData* pData = pLogStore->data; - SWal* pWal = pData->pWal; - sTrace("vgId:%d, reset wal begin index:%ld", pData->pSyncNode->vgId, beginIndex); - - pData->beginIndex = beginIndex; - walRestoreFromSnapshot(pWal, beginIndex - 1); - return 0; -} -*/ - static int32_t raftLogRestoreFromSnapshot(struct SSyncLogStore* pLogStore, SyncIndex snapshotIndex) { ASSERT(snapshotIndex >= 0); SSyncLogStoreData* pData = pLogStore->data; SWal* pWal = pData->pWal; - walRestoreFromSnapshot(pWal, snapshotIndex); + 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); + + 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); + + ASSERT(0); + } + return 0; } @@ -91,18 +89,6 @@ static int32_t raftLogEntryCount(struct SSyncLogStore* pLogStore) { return count > 0 ? count : 0; } -#if 0 -static bool raftLogInRange(struct SSyncLogStore* pLogStore, SyncIndex index) { - SyncIndex beginIndex = raftLogBeginIndex(pLogStore); - SyncIndex endIndex = raftLogEndIndex(pLogStore); - if (index >= beginIndex && index <= endIndex) { - return true; - } else { - return false; - } -} -#endif - static SyncIndex raftLogLastIndex(struct SSyncLogStore* pLogStore) { SyncIndex lastIndex; SSyncLogStoreData* pData = pLogStore->data; @@ -138,24 +124,6 @@ static SyncTerm raftLogLastTerm(struct SSyncLogStore* pLogStore) { return 0; } -/* -static SyncTerm raftLogLastTerm(struct SSyncLogStore* pLogStore) { - SyncTerm lastTerm = 0; - if (raftLogEntryCount(pLogStore) == 0) { - lastTerm = 0; - } else { - SSyncRaftEntry* pLastEntry; - int32_t code = raftLogGetLastEntry(pLogStore, &pLastEntry); - ASSERT(code == 0); - if (pLastEntry != NULL) { - lastTerm = pLastEntry->term; - taosMemoryFree(pLastEntry); - } - } - return lastTerm; -} -*/ - static int32_t raftLogAppendEntry(struct SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry) { SSyncLogStoreData* pData = pLogStore->data; SWal* pWal = pData->pWal; @@ -197,53 +165,6 @@ static int32_t raftLogAppendEntry(struct SSyncLogStore* pLogStore, SSyncRaftEntr return code; } -#if 0 -static int32_t raftLogGetEntry(struct SSyncLogStore* pLogStore, SyncIndex index, SSyncRaftEntry** ppEntry) { - SSyncLogStoreData* pData = pLogStore->data; - SWal* pWal = pData->pWal; - int32_t code; - - *ppEntry = NULL; - if (raftLogInRange(pLogStore, index)) { - SWalReadHandle* pWalHandle = walOpenReadHandle(pWal); - ASSERT(pWalHandle != NULL); - - code = walReadWithHandle(pWalHandle, index); - if (code != 0) { - int32_t err = terrno; - const char* errStr = tstrerror(err); - int32_t linuxErr = errno; - const char* linuxErrMsg = strerror(errno); - sError("raftLogGetEntry error, err:%d %X, msg:%s, linuxErr:%d, linuxErrMsg:%s", err, err, errStr, linuxErr, - linuxErrMsg); - ASSERT(0); - walCloseReadHandle(pWalHandle); - 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); - - // need to hold, do not new every time!! - walCloseReadHandle(pWalHandle); - - } else { - // index not in range - code = 0; - } - - return code; -} -#endif - static int32_t raftLogGetEntry(struct SSyncLogStore* pLogStore, SyncIndex index, SSyncRaftEntry** ppEntry) { SSyncLogStoreData* pData = pLogStore->data; SWal* pWal = pData->pWal; @@ -343,18 +264,6 @@ static int32_t raftLogGetLastEntry(SSyncLogStore* pLogStore, SSyncRaftEntry** pp return -1; } -/* -static int32_t raftLogGetLastEntry(SSyncLogStore* pLogStore, SSyncRaftEntry** ppLastEntry) { - *ppLastEntry = NULL; - if (raftLogEntryCount(pLogStore) == 0) { - return 0; - } - SyncIndex lastIndex = raftLogLastIndex(pLogStore); - int32_t code = raftLogGetEntry(pLogStore, lastIndex, ppLastEntry); - return code; -} -*/ - //------------------------------- SSyncLogStore* logStoreCreate(SSyncNode* pSyncNode) { SSyncLogStore* pLogStore = taosMemoryMalloc(sizeof(SSyncLogStore)); @@ -372,18 +281,6 @@ SSyncLogStore* logStoreCreate(SSyncNode* pSyncNode) { pData->pWalHandle = walOpenReadHandle(pData->pWal); ASSERT(pData->pWalHandle != NULL); - /* - SyncIndex firstVer = walGetFirstVer(pData->pWal); - SyncIndex lastVer = walGetLastVer(pData->pWal); - if (firstVer >= 0) { - pData->beginIndex = firstVer; - } else if (firstVer == -1) { - pData->beginIndex = lastVer + 1; - } else { - ASSERT(0); - } - */ - pLogStore->appendEntry = logStoreAppendEntry; pLogStore->getEntry = logStoreGetEntry; pLogStore->truncate = logStoreTruncate; @@ -392,7 +289,6 @@ SSyncLogStore* logStoreCreate(SSyncNode* pSyncNode) { pLogStore->updateCommitIndex = logStoreUpdateCommitIndex; pLogStore->getCommitIndex = logStoreGetCommitIndex; - // pLogStore->syncLogSetBeginIndex = raftLogSetBeginIndex; pLogStore->syncLogRestoreFromSnapshot = raftLogRestoreFromSnapshot; pLogStore->syncLogBeginIndex = raftLogBeginIndex; pLogStore->syncLogEndIndex = raftLogEndIndex; @@ -405,8 +301,6 @@ SSyncLogStore* logStoreCreate(SSyncNode* pSyncNode) { pLogStore->syncLogTruncate = raftLogTruncate; pLogStore->syncLogWriteIndex = raftLogWriteIndex; - // pLogStore->syncLogInRange = raftLogInRange; - return pLogStore; } @@ -593,55 +487,6 @@ SSyncRaftEntry* logStoreGetLastEntry(SSyncLogStore* pLogStore) { return pEntry; } -/* -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); - - snprintf(u64buf, sizeof(u64buf), "%ld", pData->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); - - for (SyncIndex i = pData->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; -} -*/ - cJSON* logStore2Json(SSyncLogStore* pLogStore) { char u64buf[128] = {0}; SSyncLogStoreData* pData = (SSyncLogStoreData*)pLogStore->data; diff --git a/source/libs/sync/src/syncSnapshot.c b/source/libs/sync/src/syncSnapshot.c index 0e77876b97..24fe8564ea 100644 --- a/source/libs/sync/src/syncSnapshot.c +++ b/source/libs/sync/src/syncSnapshot.c @@ -772,6 +772,7 @@ int32_t syncNodeOnSnapshotRspCb(SSyncNode *pSyncNode, SyncSnapshotRsp *pMsg) { snapshotReSend(pSender); } else { + // error log do { char logBuf[96]; snprintf(logBuf, sizeof(logBuf), "snapshot sender recv error ack:%d, my seq:%d", pMsg->ack, pSender->seq);