more progress

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

@@ -383,17 +383,7 @@ static int tdbBtreeZeroPage(SPage *pPage, void *arg) {
   return 0;
 }
 
-#ifndef TDB_BTREE_BALANCE
-typedef struct {
-  SBTree *pBt;
-  SPage  *pParent;
-  int     idx;
-  i8      nOld;
-  SPage  *pOldPages[3];
-  i8      nNewPages;
-  SPage  *pNewPages[5];
-} SBtreeBalanceHelper;
-
+// TDB_BTREE_BALANCE =====================
 static int tdbBtreeBalanceDeeper(SBTree *pBt, SPage *pRoot, SPage **ppChild) {
   SPager           *pPager;
   SPage            *pChild;
@@ -420,6 +410,13 @@ static int tdbBtreeBalanceDeeper(SBTree *pBt, SPage *pRoot, SPage **ppChild) {
     ((SIntHdr *)pChild->pData)->pgno = ((SIntHdr *)(pRoot->pData))->pgno;
   }
 
+  ret = tdbPagerWrite(pPager, pChild);
+  if (ret < 0) {
+    // TODO
+    ASSERT(0);
+    return 0;
+  }
+
   // Copy the root page content to the child page
   tdbPageCopy(pRoot, pChild);
 
@@ -497,6 +494,13 @@ static int tdbBtreeBalanceNonRoot(SBTree *pBt, SPage *pParent, int idx) {
         }
 
         if (i < nOlds - 1) {
+          ret = tdbPagerWrite(pBt->pPager, pOlds[i]);
+          if (ret < 0) {
+            // TODO
+            ASSERT(0);
+            return -1;
+          }
+
           ((SPgno *)pDivCell[i])[0] = ((SIntHdr *)pOlds[i]->pData)->pgno;
           ((SIntHdr *)pOlds[i]->pData)->pgno = 0;
           tdbPageInsertCell(pOlds[i], TDB_PAGE_TOTAL_CELLS(pOlds[i]), pDivCell[i], szDivCell[i], 1);
@@ -504,6 +508,14 @@ static int tdbBtreeBalanceNonRoot(SBTree *pBt, SPage *pParent, int idx) {
       }
       rPgno = ((SIntHdr *)pOlds[nOlds - 1]->pData)->pgno;
     }
+
+    ret = tdbPagerWrite(pBt->pPager, pParent);
+    if (ret < 0) {
+      // TODO
+      ASSERT(0);
+      return -1;
+    }
+
     // drop the cells on parent page
     for (int i = 0; i < nOlds; i++) {
       nCells = TDB_PAGE_TOTAL_CELLS(pParent);
@@ -631,6 +643,13 @@ static int tdbBtreeBalanceNonRoot(SBTree *pBt, SPage *pParent, int idx) {
         if (ret < 0) {
           ASSERT(0);
         }
+
+        ret = tdbPagerWrite(pBt->pPager, pNews[iNew]);
+        if (ret < 0) {
+          // TODO
+          ASSERT(0);
+          return -1;
+        }
       }
     }
 
@@ -749,9 +768,10 @@ static int tdbBtreeBalanceNonRoot(SBTree *pBt, SPage *pParent, int idx) {
 
 static int tdbBtreeBalance(SBTC *pBtc) {
   int    iPage;
+  int    ret;
+  int    nFree;
   SPage *pParent;
   SPage *pPage;
-  int    ret;
   u8     flags;
   u8     leaf;
   u8     root;
@@ -762,10 +782,11 @@ static int tdbBtreeBalance(SBTC *pBtc) {
     pPage = pBtc->pPage;
     leaf = TDB_BTREE_PAGE_IS_LEAF(pPage);
     root = TDB_BTREE_PAGE_IS_ROOT(pPage);
+    nFree = TDB_PAGE_FREE_SIZE(pPage);
 
     // when the page is not overflow and not too empty, the balance work
     // is finished. Just break out the balance loop.
-    if (pPage->nOverflow == 0 /* TODO: && pPage->nFree <= */) {
+    if (pPage->nOverflow == 0 && nFree < TDB_PAGE_USABLE_SIZE(pPage) * 2 / 3) {
       break;
     }
 
@@ -800,7 +821,7 @@ static int tdbBtreeBalance(SBTC *pBtc) {
 
   return 0;
 }
-#endif
+// TDB_BTREE_BALANCE
 
 // TDB_BTREE_CELL =====================
 static int tdbBtreeEncodePayload(SPage *pPage, SCell *pCell, int nHeader, const void *pKey, int kLen, const void *pVal,

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

@@ -118,13 +118,17 @@ int tdbPagerOpenDB(SPager *pPager, SPgno *ppgno, bool toCreate) {
 
 int tdbPagerWrite(SPager *pPager, SPage *pPage) {
   int     ret;
+  SPage **ppPage;
 
+  ASSERT(pPager->inTran);
+#if 0
   if (pPager->inTran == 0) {
     ret = tdbPagerBegin(pPager);
     if (ret < 0) {
       return -1;
     }
   }
+#endif
 
   if (pPage->isDirty) return 0;
 
@@ -132,11 +136,14 @@ int tdbPagerWrite(SPager *pPager, SPage *pPage) {
   pPage->isDirty = 1;
 
   // Add page to dirty list
-  // TODO: sort the list according to the page number
-  pPage->pDirtyNext = pPager->pDirty;
-  pPager->pDirty = pPage;
+  for (ppPage = &pPager->pDirty; (*ppPage) && TDB_PAGE_PGNO(*ppPage) < TDB_PAGE_PGNO(pPage);
+       ppPage = &((*ppPage)->pDirtyNext)) {
+  }
+  ASSERT(*ppPage == NULL || TDB_PAGE_PGNO(*ppPage) > TDB_PAGE_PGNO(pPage));
+  pPage->pDirtyNext = *ppPage;
+  *ppPage = pPage;
 
-  // Write page to journal
+  // Write page to journal if neccessary
   if (TDB_PAGE_PGNO(pPage) <= pPager->dbOrigSize) {
     ret = tdbPagerWritePageToJournal(pPager, pPage);
     if (ret < 0) {
@@ -170,30 +177,29 @@ int tdbPagerCommit(SPager *pPager) {
   SPage *pPage;
   int    ret;
 
-  // Begin commit
-  {
-    // TODO: Sync the journal file (Here or when write ?)
+  // sync the journal file
+  ret = tdbOsFSync(pPager->jfd);
+  if (ret < 0) {
+    // TODO
+    ASSERT(0);
+    return 0;
   }
 
-  for (;;) {
-    pPage = pPager->pDirty;
-
-    if (pPage == NULL) break;
-
+  // loop to write the dirty pages to file
+  for (pPage = pPager->pDirty; pPage; pPage = pPage->pDirtyNext) {
     ret = tdbPagerWritePageToDB(pPager, pPage);
     if (ret < 0) {
       ASSERT(0);
       return -1;
     }
-
-    pPager->pDirty = pPage->pDirtyNext;
-    pPage->pDirtyNext = NULL;
-
-    // TODO: release the page
   }
 
+  // TODO: loop to release the dirty pages
+
+  // sync the db file
   tdbOsFSync(pPager->fd);
 
+  // remote the journal file
   tdbOsClose(pPager->jfd);
   tdbOsRemove(pPager->jFileName);
 

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

@@ -100,6 +100,7 @@ struct SPage {
 // APIs
 #define TDB_PAGE_TOTAL_CELLS(pPage)        ((pPage)->nOverflow + (pPage)->pPageMethods->getCellNum(pPage))
 #define TDB_PAGE_USABLE_SIZE(pPage)        ((u8 *)(pPage)->pPageFtr - (pPage)->pCellIdx)
+#define TDB_PAGE_FREE_SIZE(pPage)          (*(pPage)->pPageMethods->getFreeBytes)(pPage)
 #define TDB_PAGE_PGNO(pPage)               ((pPage)->pgid.pgno)
 #define TDB_BYTES_CELL_TAKEN(pPage, pCell) ((*(pPage)->xCellSize)(pPage, pCell) + (pPage)->pPageMethods->szOffset)
 #define TDB_PAGE_OFFSET_SIZE(pPage)        ((pPage)->pPageMethods->szOffset)

source/libs/tdb/test/tdbTest.cpp

@@ -134,8 +134,9 @@ TEST(tdb_test, simple_test) {
     char val[64];
 
     {  // Insert some data
-      int i = 1;
+      tdbBegin(pEnv);
 
+      int i = 1;
       for (;;) {
         if (i > nData) break;