From a22814a5a7434df7f271a2a35e03afd083078997 Mon Sep 17 00:00:00 2001
From: Minghao Li <castermode@gmail.com>
Date: Mon, 27 Jun 2022 10:36:28 +0800
Subject: [PATCH] refactor(sync): optimized one replica

---
 source/libs/sync/inc/syncRaftLog.h |  7 +++++--
 source/libs/sync/src/syncRaftLog.c | 13 +++++++++++++
 2 files changed, 18 insertions(+), 2 deletions(-)

diff --git a/source/libs/sync/inc/syncRaftLog.h b/source/libs/sync/inc/syncRaftLog.h
index f124867097..f3ed9e302b 100644
--- a/source/libs/sync/inc/syncRaftLog.h
+++ b/source/libs/sync/inc/syncRaftLog.h
@@ -29,9 +29,12 @@ extern "C" {
 #include "wal.h"
 
 typedef struct SSyncLogStoreData {
-  SSyncNode*      pSyncNode;
-  SWal*           pWal;
+  SSyncNode* pSyncNode;
+  SWal*      pWal;
+
+  TdThreadMutex   mutex;
   SWalReadHandle* pWalHandle;
+
   // SyncIndex       beginIndex;  // valid begin index, default 0, may be set beginIndex > 0
 } SSyncLogStoreData;
 
diff --git a/source/libs/sync/src/syncRaftLog.c b/source/libs/sync/src/syncRaftLog.c
index cc606f4df8..d84c576714 100644
--- a/source/libs/sync/src/syncRaftLog.c
+++ b/source/libs/sync/src/syncRaftLog.c
@@ -257,6 +257,8 @@ static int32_t raftLogGetEntry(struct SSyncLogStore* pLogStore, SyncIndex index,
     return -1;
   }
 
+  taosThreadMutexLock(&(pData->mutex));
+
   code = walReadWithHandle(pWalHandle, index);
   if (code != 0) {
     int32_t     err = terrno;
@@ -281,6 +283,7 @@ static int32_t raftLogGetEntry(struct SSyncLogStore* pLogStore, SyncIndex index,
         terrno = saveErr;
     */
 
+    taosThreadMutexUnlock(&(pData->mutex));
     return code;
   }
 
@@ -301,6 +304,7 @@ static int32_t raftLogGetEntry(struct SSyncLogStore* pLogStore, SyncIndex index,
     terrno = saveErr;
   */
 
+  taosThreadMutexUnlock(&(pData->mutex));
   return code;
 }
 
@@ -364,6 +368,7 @@ SSyncLogStore* logStoreCreate(SSyncNode* pSyncNode) {
   pData->pWal = pSyncNode->pWal;
   ASSERT(pData->pWal != NULL);
 
+  taosThreadMutexInit(&(pData->mutex), NULL);
   pData->pWalHandle = walOpenReadHandle(pData->pWal);
   ASSERT(pData->pWalHandle != NULL);
 
@@ -408,9 +413,14 @@ SSyncLogStore* logStoreCreate(SSyncNode* pSyncNode) {
 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);
@@ -460,6 +470,8 @@ SSyncRaftEntry* logStoreGetEntry(SSyncLogStore* pLogStore, SyncIndex index) {
   SWal*              pWal = pData->pWal;
 
   if (index >= SYNC_INDEX_BEGIN && index <= logStoreLastIndex(pLogStore)) {
+    taosThreadMutexLock(&(pData->mutex));
+
     // SWalReadHandle* pWalHandle = walOpenReadHandle(pWal);
     SWalReadHandle* pWalHandle = pData->pWalHandle;
     ASSERT(pWalHandle != NULL);
@@ -503,6 +515,7 @@ SSyncRaftEntry* logStoreGetEntry(SSyncLogStore* pLogStore, SyncIndex index) {
         terrno = saveErr;
     */
 
+    taosThreadMutexUnlock(&(pData->mutex));
     return pEntry;
 
   } else {