From eb5b10d14d9b6ab2b7d4e0e5a149c705cb533e90 Mon Sep 17 00:00:00 2001
From: Hongze Cheng <hzcheng@taosdata.com>
Date: Fri, 4 Mar 2022 09:53:26 +0000
Subject: [PATCH] more TDB

---
 source/libs/tdb/src/db/tdbPCache.c | 34 ++++++++++++++++++++++++------
 1 file changed, 27 insertions(+), 7 deletions(-)

diff --git a/source/libs/tdb/src/db/tdbPCache.c b/source/libs/tdb/src/db/tdbPCache.c
index fcacb40819..9f7c314f7c 100644
--- a/source/libs/tdb/src/db/tdbPCache.c
+++ b/source/libs/tdb/src/db/tdbPCache.c
@@ -86,8 +86,12 @@ SPage *tdbPCacheFetch(SPCache *pCache, const SPgid *pPgid, bool alcNewPage) {
 }
 
 void tdbPCacheRelease(SPage *pPage) {
-  pPage->nRef--;
-  if (pPage->nRef == 0) {
+  i32 nRef;
+
+  nRef = TDB_UNREF_PAGE(pPage);
+  ASSERT(nRef >= 0);
+
+  if (nRef == 0) {
     if (1 /*TODO: page still clean*/) {
       tdbPCacheUnpinPage(pPage);
     } else {
@@ -125,6 +129,7 @@ static SPage *tdbPCacheFetchImpl(SPCache *pCache, const SPgid *pPgid, bool alcNe
     if (pPage) {
       tdbPCachePinPage(pPage);
     }
+    TDB_REF_PAGE(pPage);
     return pPage;
   }
 
@@ -153,6 +158,7 @@ static SPage *tdbPCacheFetchImpl(SPCache *pCache, const SPgid *pPgid, bool alcNe
     pPage->pLruNext = NULL;
     pPage->pPager = NULL;
     tdbPCacheAddPageToHash(pPage);
+    TDB_REF_PAGE(pPage);
   }
 
   return pPage;
@@ -172,15 +178,28 @@ static void tdbPCachePinPage(SPage *pPage) {
 }
 
 static void tdbPCacheUnpinPage(SPage *pPage) {
-  // Add current page to the LRU list
   SPCache *pCache;
+  i32      nRef;
 
-  pPage->pLruPrev = &(pCache->lru);
-  pPage->pLruNext = pCache->lru.pLruNext;
-  pCache->lru.pLruNext->pLruPrev = pPage;
-  pCache->lru.pLruNext = pPage;
+  pCache = pPage->pCache;
+
+  tdbPCacheLock(pCache);
+
+  nRef = TDB_PAGE_REF(pPage);
+  ASSERT(nRef >= 0);
+  if (nRef == 0) {
+    // Add the page to LRU list
+    ASSERT(pPage->pLruNext == NULL);
+
+    pPage->pLruPrev = &(pCache->lru);
+    pPage->pLruNext = pCache->lru.pLruNext;
+    pCache->lru.pLruNext->pLruPrev = pPage;
+    pCache->lru.pLruNext = pPage;
+  }
 
   pCache->nRecyclable++;
+
+  tdbPCacheUnlock(pCache);
 }
 
 static void tdbPCacheRemovePageFromHash(SPage *pPage) {
@@ -236,6 +255,7 @@ static int tdbPCacheOpenImpl(SPCache *pCache) {
     pPage->isAnchor = 0;
     pPage->isLocalPage = 1;
     pPage->pCache = pCache;
+    TDB_INIT_PAGE_REF(pPage);
     pPage->pHashNext = NULL;
     pPage->pLruNext = NULL;
     pPage->pLruPrev = NULL;