Merge pull request #14348 from taosdata/feature/3.0_mhli

refactor(sync): delete old functions

source/libs/sync/src/syncRaftLog.c

@@ -17,24 +17,27 @@
 #include "syncRaftCfg.h"
 #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);
+//-------------------------------
+// log[m .. n]
+
+// public function
+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);
 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);
 
+// private function
 static int32_t raftLogGetLastEntry(SSyncLogStore* pLogStore, SSyncRaftEntry** ppLastEntry);
 
 //-------------------------------
+// log[0 .. n]
 static SSyncRaftEntry* logStoreGetLastEntry(SSyncLogStore* pLogStore);
 static SyncIndex       logStoreLastIndex(SSyncLogStore* pLogStore);
 static SyncTerm        logStoreLastTerm(SSyncLogStore* pLogStore);
@@ -44,28 +47,86 @@ 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);
+//-------------------------------
+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;
-  SWal*              pWal = pData->pWal;
-  sTrace("vgId:%d, reset wal begin index:%ld", pData->pSyncNode->vgId, beginIndex);
+  pData->pSyncNode = pSyncNode;
+  pData->pWal = pSyncNode->pWal;
+  ASSERT(pData->pWal != NULL);
 
-  pData->beginIndex = beginIndex;
-  walRestoreFromSnapshot(pWal, beginIndex - 1);
-  return 0;
+  taosThreadMutexInit(&(pData->mutex), NULL);
+  pData->pWalHandle = walOpenReadHandle(pData->pWal);
+  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;
+
+  return pLogStore;
+}
+
+void logStoreDestory(SSyncLogStore* pLogStore) {
+  if (pLogStore != NULL) {
+    SSyncLogStoreData* pData = pLogStore->data;
+
+    taosThreadMutexLock(&(pData->mutex));
+    if (pData->pWalHandle != NULL) {
+      walCloseReadHandle(pData->pWalHandle);
+      pData->pWalHandle = NULL;
+    }
+    taosThreadMutexUnlock(&(pData->mutex));
+    taosThreadMutexDestroy(&(pData->mutex));
+
+    taosMemoryFree(pLogStore->data);
+    taosMemoryFree(pLogStore);
+  }
 }
-*/
 
+//-------------------------------
+// log[m .. n]
 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);
+
+    return -1;
+  }
+
   return 0;
 }
 
@@ -91,18 +152,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 +187,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 +228,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,91 +327,8 @@ 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));
-  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);
-  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;
-  pLogStore->getLastIndex = logStoreLastIndex;
-  pLogStore->getLastTerm = logStoreLastTerm;
-  pLogStore->updateCommitIndex = logStoreUpdateCommitIndex;
-  pLogStore->getCommitIndex = logStoreGetCommitIndex;
-
-  // pLogStore->syncLogSetBeginIndex = raftLogSetBeginIndex;
-  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->syncLogInRange = raftLogInRange;
-
-  return pLogStore;
-}
-
-void logStoreDestory(SSyncLogStore* pLogStore) {
-  if (pLogStore != NULL) {
-    SSyncLogStoreData* pData = pLogStore->data;
-
-    taosThreadMutexLock(&(pData->mutex));
-    if (pData->pWalHandle != NULL) {
-      walCloseReadHandle(pData->pWalHandle);
-      pData->pWalHandle = NULL;
-    }
-    taosThreadMutexUnlock(&(pData->mutex));
-    taosThreadMutexDestroy(&(pData->mutex));
-
-    taosMemoryFree(pLogStore->data);
-    taosMemoryFree(pLogStore);
-  }
-}
-
 //-------------------------------
+// log[0 .. n]
 int32_t logStoreAppendEntry(SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry) {
   SSyncLogStoreData* pData = pLogStore->data;
   SWal*              pWal = pData->pWal;
@@ -593,55 +494,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;

source/libs/sync/src/syncSnapshot.c

@@ -508,32 +508,51 @@ void snapshotReceiverStop(SSyncSnapshotReceiver *pReceiver) {
   } while (0);
 }
 
-static void snapshotReceiverFinish(SSyncSnapshotReceiver *pReceiver, SyncSnapshotSend *pMsg) {
+static int32_t snapshotReceiverFinish(SSyncSnapshotReceiver *pReceiver, SyncSnapshotSend *pMsg) {
   ASSERT(pMsg->seq == SYNC_SNAPSHOT_SEQ_END);
 
+  int32_t code = 0;
   if (pReceiver->pWriter != NULL) {
-    int32_t code = 0;
+    // write data
     if (pMsg->dataLen > 0) {
       code = pReceiver->pSyncNode->pFsm->FpSnapshotDoWrite(pReceiver->pSyncNode->pFsm, pReceiver->pWriter, pMsg->data,
                                                            pMsg->dataLen);
-      ASSERT(code == 0);
+      if (code != 0) {
+        syncNodeErrorLog(pReceiver->pSyncNode, "snapshot write error");
+        return -1;
+      }
     }
 
+    // reset wal
+    code =
+        pReceiver->pSyncNode->pLogStore->syncLogRestoreFromSnapshot(pReceiver->pSyncNode->pLogStore, pMsg->lastIndex);
+    if (code != 0) {
+      syncNodeErrorLog(pReceiver->pSyncNode, "wal restore from snapshot error");
+      return -1;
+    }
+
+    // update commit index
+    if (pReceiver->snapshot.lastApplyIndex > pReceiver->pSyncNode->commitIndex) {
+      pReceiver->pSyncNode->commitIndex = pReceiver->snapshot.lastApplyIndex;
+    }
+
+    // stop writer
     code = pReceiver->pSyncNode->pFsm->FpSnapshotStopWrite(pReceiver->pSyncNode->pFsm, pReceiver->pWriter, true);
-    ASSERT(code == 0);
+    if (code != 0) {
+      syncNodeErrorLog(pReceiver->pSyncNode, "snapshot stop writer true error");
+      ASSERT(0);
+      return -1;
+    }
     pReceiver->pWriter = NULL;
-  }
 
-  pReceiver->ack = SYNC_SNAPSHOT_SEQ_END;
+    // update progress
+    pReceiver->ack = SYNC_SNAPSHOT_SEQ_END;
 
-  // update commit index
-  if (pReceiver->snapshot.lastApplyIndex > pReceiver->pSyncNode->commitIndex) {
-    pReceiver->pSyncNode->commitIndex = pReceiver->snapshot.lastApplyIndex;
+  } else {
+    syncNodeErrorLog(pReceiver->pSyncNode, "snapshot stop writer true error");
+    return -1;
   }
 
-  // reset wal
-  pReceiver->pSyncNode->pLogStore->syncLogRestoreFromSnapshot(pReceiver->pSyncNode->pLogStore, pMsg->lastIndex);
-
   // event log
   do {
     SSnapshot snapshot;
@@ -542,6 +561,8 @@ static void snapshotReceiverFinish(SSyncSnapshotReceiver *pReceiver, SyncSnapsho
     syncNodeEventLog(pReceiver->pSyncNode, eventLog);
     taosMemoryFree(eventLog);
   } while (0);
+
+  return 0;
 }
 
 static void snapshotReceiverGotData(SSyncSnapshotReceiver *pReceiver, SyncSnapshotSend *pMsg) {
@@ -642,6 +663,7 @@ int32_t syncNodeOnSnapshotSendCb(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) {
   // get receiver
   SSyncSnapshotReceiver *pReceiver = pSyncNode->pNewNodeReceiver;
   bool                   needRsp = false;
+  int32_t                code = 0;
 
   // state, term, seq/ack
   if (pSyncNode->state == TAOS_SYNC_STATE_FOLLOWER) {
@@ -653,8 +675,10 @@ int32_t syncNodeOnSnapshotSendCb(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) {
 
       } else if (pMsg->seq == SYNC_SNAPSHOT_SEQ_END) {
         // end, finish FSM
-        snapshotReceiverFinish(pReceiver, pMsg);
-        snapshotReceiverStop(pReceiver);
+        code = snapshotReceiverFinish(pReceiver, pMsg);
+        if (code == 0) {
+          snapshotReceiverStop(pReceiver);
+        }
         needRsp = true;
 
         // maybe update lastconfig
@@ -772,6 +796,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);