more TDB

source/libs/tdb/src/db/tdbBtree.c

@@ -78,6 +78,8 @@ static int tdbBtreeDecodeCell(SPage *pPage, const SCell *pCell, SCellDecoder *pD
 static int tdbBtreeBalance(SBTC *pCur);
 static int tdbBtreeCellSize(const SPage *pPage, SCell *pCell);
 static int tdbBtcMoveToNext(SBTC *pBtc);
+static int tdbBtcMoveDownward(SBTC *pCur, SPgno pgno);
+static int tdbBtcMoveUpward(SBTC *pBtc);
 
 int tdbBtreeOpen(int keyLen, int valLen, SPager *pPager, FKeyComparator kcmpr, SBTree **ppBt) {
   SBTree *pBt;
@@ -224,23 +226,6 @@ int tdbBtreeGet(SBTree *pBt, const void *pKey, int kLen, void **ppVal, int *vLen
   return 0;
 }
 
-static int tdbBtCursorMoveToChild(SBTC *pCur, SPgno pgno) {
-  int ret;
-
-  pCur->pgStack[pCur->iPage] = pCur->pPage;
-  pCur->idxStack[pCur->iPage] = pCur->idx;
-  pCur->iPage++;
-  pCur->pPage = NULL;
-  pCur->idx = -1;
-
-  ret = tdbPagerFetchPage(pCur->pBt->pPager, pgno, &pCur->pPage, tdbBtreeInitPage, pCur->pBt);
-  if (ret < 0) {
-    ASSERT(0);
-  }
-
-  return 0;
-}
-
 static int tdbBtCursorMoveTo(SBTC *pCur, const void *pKey, int kLen, int *pCRst) {
   int     ret;
   SBTree *pBt;
@@ -318,16 +303,16 @@ static int tdbBtCursorMoveTo(SBTC *pCur, const void *pKey, int kLen, int *pCRst)
       } else {
         if (c <= 0) {
           pCur->idx = midx;
-          tdbBtCursorMoveToChild(pCur, cd.pgno);
+          tdbBtcMoveDownward(pCur, cd.pgno);
         } else {
           pCur->idx = midx + 1;
           if (midx == nCells - 1) {
             /* Move to right-most child */
-            tdbBtCursorMoveToChild(pCur, ((SIntHdr *)pCur->pPage->pData)->pgno);
+            tdbBtcMoveDownward(pCur, ((SIntHdr *)pCur->pPage->pData)->pgno);
           } else {
             pCell = tdbPageGetCell(pPage, pCur->idx);
             tdbBtreeDecodeCell(pPage, pCell, &cd);
-            tdbBtCursorMoveToChild(pCur, cd.pgno);
+            tdbBtcMoveDownward(pCur, cd.pgno);
           }
         }
       }
@@ -1109,7 +1094,7 @@ int tdbBtcMoveToFirst(SBTC *pBtc) {
     pCell = tdbPageGetCell(pBtc->pPage, 0);
     pgno = *(SPgno *)pCell;
 
-    ret = tdbBtCursorMoveToChild(pBtc, pgno);
+    ret = tdbBtcMoveDownward(pBtc, pgno);
     if (ret < 0) {
       ASSERT(0);
       return -1;
@@ -1158,7 +1143,7 @@ int tdbBtcMoveToLast(SBTC *pBtc) {
       pBtc->idx = TDB_PAGE_TOTAL_CELLS(pBtc->pPage);
       pgno = ((SIntHdr *)pBtc->pPage->pData)->pgno;
 
-      ret = tdbBtCursorMoveToChild(pBtc, pgno);
+      ret = tdbBtcMoveDownward(pBtc, pgno);
       if (ret < 0) {
         ASSERT(0);
         return -1;
@@ -1242,4 +1227,26 @@ static int tdbBtcMoveToNext(SBTC *pBtc) {
 int tdbBtcClose(SBTC *pBtc) {
   // TODO
   return 0;
+}
+
+static int tdbBtcMoveDownward(SBTC *pCur, SPgno pgno) {
+  int ret;
+
+  pCur->pgStack[pCur->iPage] = pCur->pPage;
+  pCur->idxStack[pCur->iPage] = pCur->idx;
+  pCur->iPage++;
+  pCur->pPage = NULL;
+  pCur->idx = -1;
+
+  ret = tdbPagerFetchPage(pCur->pBt->pPager, pgno, &pCur->pPage, tdbBtreeInitPage, pCur->pBt);
+  if (ret < 0) {
+    ASSERT(0);
+  }
+
+  return 0;
+}
+
+static int tdbBtcMoveUpward(SBTC *pBtc) {
+  // TODO
+  return 0;
 }
\ No newline at end of file

source/libs/tdb/src/db/tdbPCache.c

@@ -53,10 +53,10 @@ static void   tdbPCacheLock(SPCache *pCache);
 static void   tdbPCacheUnlock(SPCache *pCache);
 static bool   tdbPCacheLocked(SPCache *pCache);
 static SPage *tdbPCacheFetchImpl(SPCache *pCache, const SPgid *pPgid, bool alcNewPage);
-static void   tdbPCachePinPage(SPage *pPage);
-static void   tdbPCacheRemovePageFromHash(SPage *pPage);
-static void   tdbPCacheAddPageToHash(SPage *pPage);
-static void   tdbPCacheUnpinPage(SPage *pPage);
+static void   tdbPCachePinPage(SPCache *pCache, SPage *pPage);
+static void   tdbPCacheRemovePageFromHash(SPCache *pCache, SPage *pPage);
+static void   tdbPCacheAddPageToHash(SPCache *pCache, SPage *pPage);
+static void   tdbPCacheUnpinPage(SPCache *pCache, SPage *pPage);
 
 int tdbPCacheOpen(int pageSize, int cacheSize, SPCache **ppCache) {
   SPCache *pCache;
@@ -100,7 +100,7 @@ SPage *tdbPCacheFetch(SPCache *pCache, const SPgid *pPgid, bool alcNewPage) {
   return pPage;
 }
 
-void tdbPCacheRelease(SPage *pPage) {
+void tdbPCacheRelease(SPCache *pCache, SPage *pPage) {
   i32 nRef;
 
   nRef = TDB_UNREF_PAGE(pPage);
@@ -108,7 +108,7 @@ void tdbPCacheRelease(SPage *pPage) {
 
   if (nRef == 0) {
     if (1 /*TODO: page still clean*/) {
-      tdbPCacheUnpinPage(pPage);
+      tdbPCacheUnpinPage(pCache, pPage);
     } else {
       // TODO
       ASSERT(0);
@@ -142,7 +142,7 @@ static SPage *tdbPCacheFetchImpl(SPCache *pCache, const SPgid *pPgid, bool alcNe
 
   if (pPage || !alcNewPage) {
     if (pPage) {
-      tdbPCachePinPage(pPage);
+      tdbPCachePinPage(pCache, pPage);
     }
     return pPage;
   }
@@ -158,8 +158,8 @@ static SPage *tdbPCacheFetchImpl(SPCache *pCache, const SPgid *pPgid, bool alcNe
   // 3. Try to Recycle a page
   if (!pPage && !pCache->lru.pLruPrev->isAnchor) {
     pPage = pCache->lru.pLruPrev;
-    tdbPCacheRemovePageFromHash(pPage);
-    tdbPCachePinPage(pPage);
+    tdbPCacheRemovePageFromHash(pCache, pPage);
+    tdbPCachePinPage(pCache, pPage);
   }
 
   // 4. Try a stress allocation (TODO)
@@ -171,16 +171,13 @@ static SPage *tdbPCacheFetchImpl(SPCache *pCache, const SPgid *pPgid, bool alcNe
     memcpy(&(pPage->pgid), pPgid, sizeof(*pPgid));
     pPage->pLruNext = NULL;
     pPage->pPager = NULL;
-    tdbPCacheAddPageToHash(pPage);
+    tdbPCacheAddPageToHash(pCache, pPage);
   }
 
   return pPage;
 }
 
-static void tdbPCachePinPage(SPage *pPage) {
-  SPCache *pCache;
-
-  pCache = pPage->pCache;
+static void tdbPCachePinPage(SPCache *pCache, SPage *pPage) {
   if (!PAGE_IS_PINNED(pPage)) {
     pPage->pLruPrev->pLruNext = pPage->pLruNext;
     pPage->pLruNext->pLruPrev = pPage->pLruPrev;
@@ -190,11 +187,8 @@ static void tdbPCachePinPage(SPage *pPage) {
   }
 }
 
-static void tdbPCacheUnpinPage(SPage *pPage) {
-  SPCache *pCache;
-  i32      nRef;
-
-  pCache = pPage->pCache;
+static void tdbPCacheUnpinPage(SPCache *pCache, SPage *pPage) {
+  i32 nRef;
 
   tdbPCacheLock(pCache);
 
@@ -215,12 +209,10 @@ static void tdbPCacheUnpinPage(SPage *pPage) {
   tdbPCacheUnlock(pCache);
 }
 
-static void tdbPCacheRemovePageFromHash(SPage *pPage) {
-  SPCache *pCache;
-  SPage  **ppPage;
-  int      h;
+static void tdbPCacheRemovePageFromHash(SPCache *pCache, SPage *pPage) {
+  SPage **ppPage;
+  int     h;
 
-  pCache = pPage->pCache;
   h = PCACHE_PAGE_HASH(&(pPage->pgid));
   for (ppPage = &(pCache->pgHash[h % pCache->nHash]); *ppPage != pPage; ppPage = &((*ppPage)->pHashNext))
     ;
@@ -230,11 +222,9 @@ static void tdbPCacheRemovePageFromHash(SPage *pPage) {
   pCache->nPage--;
 }
 
-static void tdbPCacheAddPageToHash(SPage *pPage) {
-  SPCache *pCache;
-  int      h;
+static void tdbPCacheAddPageToHash(SPCache *pCache, SPage *pPage) {
+  int h;
 
-  pCache = pPage->pCache;
   h = PCACHE_PAGE_HASH(&(pPage->pgid)) % pCache->nHash;
 
   pPage->pHashNext = pCache->pgHash[h];
@@ -264,7 +254,6 @@ static int tdbPCacheOpenImpl(SPCache *pCache) {
     // pPage->pgid = 0;
     pPage->isAnchor = 0;
     pPage->isLocalPage = 1;
-    pPage->pCache = pCache;
     TDB_INIT_PAGE_REF(pPage);
     pPage->pHashNext = NULL;
     pPage->pLruNext = NULL;

source/libs/tdb/src/db/tdbPager.c

@@ -255,6 +255,10 @@ int tdbPagerNewPage(SPager *pPager, SPgno *ppgno, SPage **ppPage, int (*initPage
   return 0;
 }
 
+void tdbPagerReturnPage(SPager *pPager, SPage *pPage) {
+  tdbPCacheRelease(pPager->pCache, pPage);
+}
+
 static int tdbPagerAllocFreePage(SPager *pPager, SPgno *ppgno) {
   // TODO: Allocate a page from the free list
   return 0;

source/libs/tdb/src/inc/tdbPCache.h

@@ -21,23 +21,22 @@ extern "C" {
 #endif
 
 #define TDB_PCACHE_PAGE \
-  u8       isAnchor;    \
-  u8       isLocalPage; \
-  u8       isDirty;     \
-  i32      nRef;        \
-  SPCache *pCache;      \
-  SPage   *pFreeNext;   \
-  SPage   *pHashNext;   \
-  SPage   *pLruNext;    \
-  SPage   *pLruPrev;    \
-  SPage   *pDirtyNext;  \
-  SPager  *pPager;      \
-  SPgid    pgid;
+  u8      isAnchor;     \
+  u8      isLocalPage;  \
+  u8      isDirty;      \
+  i32     nRef;         \
+  SPage  *pFreeNext;    \
+  SPage  *pHashNext;    \
+  SPage  *pLruNext;     \
+  SPage  *pLruPrev;     \
+  SPage  *pDirtyNext;   \
+  SPager *pPager;       \
+  SPgid   pgid;
 
 int    tdbPCacheOpen(int pageSize, int cacheSize, SPCache **ppCache);
 int    tdbPCacheClose(SPCache *pCache);
 SPage *tdbPCacheFetch(SPCache *pCache, const SPgid *pPgid, bool alcNewPage);
-void   tdbPCacheRelease(SPage *pPage);
+void   tdbPCacheRelease(SPCache *pCache, SPage *pPage);
 int    tdbPCacheGetPageSize(SPCache *pCache);
 
 #ifdef __cplusplus

source/libs/tdb/src/inc/tdbPager.h

@@ -20,15 +20,16 @@
 extern "C" {
 #endif
 
-int    tdbPagerOpen(SPCache *pCache, const char *fileName, SPager **ppPager);
-int    tdbPagerClose(SPager *pPager);
-int    tdbPagerOpenDB(SPager *pPager, SPgno *ppgno, bool toCreate);
-int    tdbPagerWrite(SPager *pPager, SPage *pPage);
-int    tdbPagerBegin(SPager *pPager);
-int    tdbPagerCommit(SPager *pPager);
-int    tdbPagerGetPageSize(SPager *pPager);
-int    tdbPagerFetchPage(SPager *pPager, SPgno pgno, SPage **ppPage, int (*initPage)(SPage *, void *), void *arg);
-int    tdbPagerNewPage(SPager *pPager, SPgno *ppgno, SPage **ppPage, int (*initPage)(SPage *, void *), void *arg);
+int  tdbPagerOpen(SPCache *pCache, const char *fileName, SPager **ppPager);
+int  tdbPagerClose(SPager *pPager);
+int  tdbPagerOpenDB(SPager *pPager, SPgno *ppgno, bool toCreate);
+int  tdbPagerWrite(SPager *pPager, SPage *pPage);
+int  tdbPagerBegin(SPager *pPager);
+int  tdbPagerCommit(SPager *pPager);
+int  tdbPagerGetPageSize(SPager *pPager);
+int  tdbPagerFetchPage(SPager *pPager, SPgno pgno, SPage **ppPage, int (*initPage)(SPage *, void *), void *arg);
+int  tdbPagerNewPage(SPager *pPager, SPgno *ppgno, SPage **ppPage, int (*initPage)(SPage *, void *), void *arg);
+void tdbPagerReturnPage(SPager *pPager, SPage *pPage);
 
 #ifdef __cplusplus
 }