From 437d4947f6914a7c7491b5f2ae087a58c33ecc8e Mon Sep 17 00:00:00 2001
From: Minglei Jin <mljin@taosdata.com>
Date: Wed, 24 Aug 2022 20:13:18 +0800
Subject: [PATCH] fix: tdb page cache hash page removing fix

---
 source/libs/tdb/src/db/tdbBtree.c  |  2 +-
 source/libs/tdb/src/db/tdbPCache.c | 35 ++++++++++++++++--------------
 source/libs/tdb/src/db/tdbPage.c   |  3 +++
 3 files changed, 23 insertions(+), 17 deletions(-)

diff --git a/source/libs/tdb/src/db/tdbBtree.c b/source/libs/tdb/src/db/tdbBtree.c
index 7a44edb12c..5430acb972 100644
--- a/source/libs/tdb/src/db/tdbBtree.c
+++ b/source/libs/tdb/src/db/tdbBtree.c
@@ -1401,7 +1401,7 @@ static int tdbBtreeDecodeCell(SPage *pPage, const SCell *pCell, SCellDecoder *pD
   pDecoder->pgno = 0;
   TDB_CELLDECODER_SET_FREE_NIL(pDecoder);
 
-  tdbDebug("tdb btc decoder set nil: %p/0x%x ", pDecoder, pDecoder->freeKV);
+  // tdbTrace("tdb btc decoder set nil: %p/0x%x ", pDecoder, pDecoder->freeKV);
 
   // 1. Decode header part
   if (!leaf) {
diff --git a/source/libs/tdb/src/db/tdbPCache.c b/source/libs/tdb/src/db/tdbPCache.c
index ab9b21dc3f..527ad200d4 100644
--- a/source/libs/tdb/src/db/tdbPCache.c
+++ b/source/libs/tdb/src/db/tdbPCache.c
@@ -145,7 +145,7 @@ static SPage *tdbPCacheFetchImpl(SPCache *pCache, const SPgid *pPgid, TXN *pTxn)
   // 1. Search the hash table
   pPage = pCache->pgHash[tdbPCachePageHash(pPgid) % pCache->nHash];
   while (pPage) {
-    if (memcmp(pPage->pgid.fileid, pPgid->fileid, TDB_FILE_ID_LEN) == 0 && pPage->pgid.pgno == pPgid->pgno) break;
+    if (pPage->pgid.pgno == pPgid->pgno && memcmp(pPage->pgid.fileid, pPgid->fileid, TDB_FILE_ID_LEN) == 0) break;
     pPage = pPage->pHashNext;
   }
 
@@ -243,7 +243,7 @@ static void tdbPCachePinPage(SPCache *pCache, SPage *pPage) {
     pCache->nRecyclable--;
 
     // printf("pin page %d pgno %d pPage %p\n", pPage->id, TDB_PAGE_PGNO(pPage), pPage);
-    tdbTrace("pin page %d", pPage->id);
+    tdbDebug("pin page %d", pPage->id);
   }
 }
 
@@ -264,30 +264,33 @@ static void tdbPCacheUnpinPage(SPCache *pCache, SPage *pPage) {
   pCache->nRecyclable++;
 
   // printf("unpin page %d pgno %d pPage %p\n", pPage->id, TDB_PAGE_PGNO(pPage), pPage);
-  tdbTrace("unpin page %d", pPage->id);
+  tdbDebug("unpin page %d", pPage->id);
 }
 
 static void tdbPCacheRemovePageFromHash(SPCache *pCache, SPage *pPage) {
-  SPage  **ppPage;
-  uint32_t h;
-
-  h = tdbPCachePageHash(&(pPage->pgid));
-  for (ppPage = &(pCache->pgHash[h % pCache->nHash]); (*ppPage) && *ppPage != pPage; ppPage = &((*ppPage)->pHashNext))
-    ;
+  uint32_t h = tdbPCachePageHash(&(pPage->pgid)) % pCache->nHash;
 
+  SPage **ppPage = &(pCache->pgHash[h]);
+  if (*ppPage == pPage) {
+    pCache->pgHash[h] = pPage->pHashNext;
+  } else {
+    for (; (*ppPage) && (*ppPage)->pHashNext != pPage; ppPage = &((*ppPage)->pHashNext))
+      ;
+    if (*ppPage) {
+      (*ppPage)->pHashNext = pPage->pHashNext;
+    }
+  }
   if (*ppPage) {
-    *ppPage = pPage->pHashNext;
-    pCache->nPage--;
+    pPage->pHashNext = NULL;
+    --pCache->nPage;
     // printf("rmv page %d to hash, pgno %d, pPage %p\n", pPage->id, TDB_PAGE_PGNO(pPage), pPage);
   }
 
-  tdbTrace("remove page %d to hash", pPage->id);
+  tdbDebug("remove page %p/%d from hash", pPage, pPage->id);
 }
 
 static void tdbPCacheAddPageToHash(SPCache *pCache, SPage *pPage) {
-  int h;
-
-  h = tdbPCachePageHash(&(pPage->pgid)) % pCache->nHash;
+  uint32_t h = tdbPCachePageHash(&(pPage->pgid)) % pCache->nHash;
 
   pPage->pHashNext = pCache->pgHash[h];
   pCache->pgHash[h] = pPage;
@@ -295,7 +298,7 @@ static void tdbPCacheAddPageToHash(SPCache *pCache, SPage *pPage) {
   pCache->nPage++;
 
   // printf("add page %d to hash, pgno %d, pPage %p\n", pPage->id, TDB_PAGE_PGNO(pPage), pPage);
-  tdbTrace("add page %d to hash", pPage->id);
+  tdbDebug("add page %p/%d to hash", pPage, pPage->id);
 }
 
 static int tdbPCacheOpenImpl(SPCache *pCache) {
diff --git a/source/libs/tdb/src/db/tdbPage.c b/source/libs/tdb/src/db/tdbPage.c
index 276b06b147..9e0cd76573 100644
--- a/source/libs/tdb/src/db/tdbPage.c
+++ b/source/libs/tdb/src/db/tdbPage.c
@@ -74,6 +74,7 @@ int tdbPageCreate(int pageSize, SPage **ppPage, void *(*xMalloc)(void *, size_t)
 int tdbPageDestroy(SPage *pPage, void (*xFree)(void *arg, void *ptr), void *arg) {
   u8 *ptr;
 
+  tdbDebug("page/destroy: %p %p", pPage, xFree);
   ASSERT(xFree);
 
   for (int iOvfl = 0; iOvfl < pPage->nOverflow; iOvfl++) {
@@ -87,6 +88,7 @@ int tdbPageDestroy(SPage *pPage, void (*xFree)(void *arg, void *ptr), void *arg)
 }
 
 void tdbPageZero(SPage *pPage, u8 szAmHdr, int (*xCellSize)(const SPage *, SCell *, int, TXN *, SBTree *pBt)) {
+  tdbDebug("page/zero: %p %" PRIu8 " %p", pPage, szAmHdr, xCellSize);
   pPage->pPageHdr = pPage->pData + szAmHdr;
   TDB_PAGE_NCELLS_SET(pPage, 0);
   TDB_PAGE_CCELLS_SET(pPage, pPage->pageSize - sizeof(SPageFtr));
@@ -103,6 +105,7 @@ void tdbPageZero(SPage *pPage, u8 szAmHdr, int (*xCellSize)(const SPage *, SCell
 }
 
 void tdbPageInit(SPage *pPage, u8 szAmHdr, int (*xCellSize)(const SPage *, SCell *, int, TXN *, SBTree *pBt)) {
+  tdbDebug("page/init: %p %" PRIu8 " %p", pPage, szAmHdr, xCellSize);
   pPage->pPageHdr = pPage->pData + szAmHdr;
   pPage->pCellIdx = pPage->pPageHdr + TDB_PAGE_HDR_SIZE(pPage);
   pPage->pFreeStart = pPage->pCellIdx + TDB_PAGE_OFFSET_SIZE(pPage) * TDB_PAGE_NCELLS(pPage);