more TDB

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

@@ -299,8 +299,8 @@ static int tdbBtreeOpenImpl(SBTree *pBt) {
   //   return -1;
   // }
 
-  ASSERT(pgno != 0);
-  pBt->root = pgno;
+  // ASSERT(pgno != 0);
+  // pBt->root = pgno;
 
   return 0;
 }

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

@@ -41,8 +41,11 @@ typedef struct __attribute__((__packed__)) {
 
 TDB_STATIC_ASSERT(sizeof(SFileHdr) == 128, "Size of file header is not correct");
 
+#define TDB_PAGE_INITIALIZED(pPage) ((pPage)->pPager != NULL)
+
 static int tdbPagerReadPage(SPager *pPager, SPage *pPage);
 static int tdbPagerAllocPage(SPager *pPager, SPgno *ppgno);
+static int tdbPagerInitPage(SPager *pPager, SPage *pPage, int (*initPage)(SPage *, void *), void *arg);
 
 int tdbPagerOpen(SPCache *pCache, const char *fileName, SPager **ppPager) {
   uint8_t *pPtr;
@@ -240,30 +243,17 @@ int tdbPagerFetchPage(SPager *pPager, SPgno pgno, SPage **ppPage, int (*initPage
     return -1;
   }
 
-  if (pPage->pPager == NULL) {
-    ASSERT(pgno < pPager->dbOrigSize);
-
-    // tdbWLockPage(pPage);
-
-    if (pPage->pPager == NULL) {
-      ret = tdbPagerReadPage(pPager, pPage);
-      if (ret < 0) {
-        return -1;
-      }
-
-      ret = (*initPage)(pPage, arg);
-      if (ret < 0) {
-        return -1;
-      }
-
-      pPage->pPager = pPager;
+  // Initialize the page if need
+  if (!TDB_PAGE_INITIALIZED(pPage)) {
+    ret = tdbPagerInitPage(pPager, pPage, initPage, arg);
+    if (ret < 0) {
+      return -1;
     }
-
-    // tdbWUnlockPage(pPage);
-  } else {
-    ASSERT(pPage->pPager == pPager);
   }
 
+  ASSERT(TDB_PAGE_INITIALIZED(pPage));
+  ASSERT(pPage->pPager == pPager);
+
   *ppPage = pPage;
   return 0;
 }
@@ -289,18 +279,16 @@ int tdbPagerNewPage(SPager *pPager, SPgno *ppgno, SPage **ppPage, int (*initPage
     return -1;
   }
 
-  ASSERT(pPage->pPager == NULL);
-
-  // TODO: a race condition problem may occur here
-
-  // tdbWLockPage(pPage);
-
-  // TODO: zero init the new page
-  (*initPage)(pPage, arg);
+  ASSERT(!TDB_PAGE_INITIALIZED(pPage));
 
-  pPage->pPager = NULL;
+  // Initialize the page if need
+  ret = tdbPagerInitPage(pPager, pPage, initPage, arg);
+  if (ret < 0) {
+    return -1;
+  }
 
-  // tdbWunlockPage(pPage);
+  ASSERT(TDB_PAGE_INITIALIZED(pPage));
+  ASSERT(pPage->pPager == pPager);
 
   *ppPage = pPage;
   return 0;
@@ -337,5 +325,32 @@ static int tdbPagerAllocPage(SPager *pPager, SPgno *ppgno) {
 
   ASSERT(*ppgno != 0);
 
+  return 0;
+}
+
+static int tdbPagerInitPage(SPager *pPager, SPage *pPage, int (*initPage)(SPage *, void *), void *arg) {
+  int ret;
+
+  ret = TDB_TRY_LOCK_PAGE(pPage);
+  if (ret == 0) {
+    if (TDB_PAGE_INITIALIZED(pPage)) {
+      TDB_UNLOCK_PAGE(pPage);
+      return 0;
+    }
+
+    ret = (*initPage)(pPage, arg);
+    if (ret < 0) {
+      TDB_UNLOCK_PAGE(pPage);
+      return -1;
+    }
+
+    pPage->pPager = pPager;
+
+    TDB_UNLOCK_PAGE(pPage);
+  } else {
+    while (!TDB_PAGE_INITIALIZED(pPage))
+      ;
+  }
+
   return 0;
 }
\ No newline at end of file

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

@@ -30,6 +30,7 @@ typedef struct __attribute__((__packed__)) {
 
 typedef struct SPage SPage;
 struct SPage {
+  pthread_spinlock_t lock;
   // Fields below used by page cache
   void *   pData;
   SPgid    pgid;
@@ -53,6 +54,12 @@ struct SPage {
   int       minLocal;
 };
 
+#define TDB_INIT_PAGE_LOCK(pPage)    pthread_spin_init(&((pPage)->lock), 0) // TODO: use the macros
+#define TDB_DESTROY_PAGE_LOCK(pPage) pthread_spin_destroy(&((pPage)->lock))
+#define TDB_LOCK_PAGE(pPage)         pthread_spin_lock(&((pPage)->lock))
+#define TDB_TRY_LOCK_PAGE(pPage)     pthread_spin_trylock(&((pPage)->lock))
+#define TDB_UNLOCK_PAGE(pPage)       pthread_spin_unlock(&((pPage)->lock))
+
 #ifdef __cplusplus
 }
 #endif