more TDB

source/libs/tdb/src/page/tdbPage.c

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "tdbInt.h"
+
+typedef struct __attribute__((__packed__)) {
+  u16 flags;
+  u16 cellNum;
+  u16 cellBody;
+  u16 cellFree;
+  u16 nFree;
+} SPageHdr;
+
+typedef struct __attribute__((__packed__)) {
+  u8 szCell[2];
+  u8 nxOffset[2];
+} SFreeCell;
+
+int tdbPageCreate(int pageSize, SPage **ppPage, void *(*xMalloc)(void *, size_t), void *arg) {
+  SPage *pPage;
+  u8    *ptr;
+  int    size;
+
+  ASSERT(TDB_IS_PGSIZE_VLD(pageSize));
+
+  *ppPage = NULL;
+  size = pageSize + sizeof(*pPage);
+
+  ptr = (u8 *)((*xMalloc)(arg, size));
+  if (pPage == NULL) {
+    return -1;
+  }
+
+  memset(ptr, 0, size);
+  pPage = (SPage *)(ptr + pageSize);
+
+  pPage->pData = ptr;
+  pPage->pageSize = pageSize;
+  if (pageSize < 65536) {
+    pPage->szOffset = 2;
+    pPage->szPageHdr = sizeof(SPageHdr);
+    pPage->szFreeCell = sizeof(SFreeCell);
+  } else {
+    pPage->szOffset = 3;
+    pPage->szPageHdr = sizeof(SPageHdrL);
+    pPage->szFreeCell = sizeof(SFreeCellL);
+  }
+  TDB_INIT_PAGE_LOCK(pPage);
+
+  /* TODO */
+
+  *ppPage = pPage;
+  return 0;
+}
+
+int tdbPageDestroy(SPage *pPage, void (*xFree)(void *arg, void *ptr), void *arg) {
+  u8 *ptr;
+
+  ptr = pPage->pData;
+  (*xFree)(arg, ptr);
+
+  return 0;
+}
+
+int tdbPageInsertCell(SPage *pPage, int idx, SCell *pCell, int szCell) {
+  int    ret;
+  SCell *pTarget;
+  u8    *pTmp;
+  int    j;
+
+  if (pPage->nOverflow || szCell + pPage->szOffset > pPage->nFree) {
+    // TODO: need to figure out if pCell may be used by outside of this function
+    j = pPage->nOverflow++;
+
+    pPage->apOvfl[j] = pCell;
+    pPage->aiOvfl[j] = idx;
+  } else {
+    ret = tdbPageAllocate(pPage, szCell, &pTarget);
+    if (ret < 0) {
+      return -1;
+    }
+
+    memcpy(pTarget, pCell, szCell);
+    pTmp = pPage->pCellIdx + idx * pPage->szOffset;
+    memmove(pTmp + pPage->szOffset, pTmp, pPage->pFreeStart - pTmp - pPage->szOffset);
+    TDB_PAGE_CELL_OFFSET_AT_SET(pPage, idx, pTarget - pPage->pData);
+    TDB_PAGE_NCELLS_SET(pPage, TDB_PAGE_NCELLS(pPage) + 1);
+  }
+
+  return 0;
+}
+
+int tdbPageDropCell(SPage *pPage, int idx) {
+  // TODO
+  return 0;
+}
+
+static int tdbPageAllocate(SPage *pPage, int size, SCell **ppCell) {
+  SCell     *pCell;
+  SFreeCell *pFreeCell;
+  u8        *pOffset;
+  int        ret;
+
+  ASSERT(pPage->nFree > size + pPage->szOffset);
+
+  pCell = NULL;
+  *ppCell = NULL;
+
+  // 1. Try to allocate from the free space area
+  if (pPage->pFreeEnd - pPage->pFreeStart > size + pPage->szOffset) {
+    pPage->pFreeEnd -= size;
+    pPage->pFreeStart += pPage->szOffset;
+    pCell = pPage->pFreeEnd;
+  }
+
+  // 2. Try to allocate from the page free list
+  if ((pCell == NULL) && (pPage->pFreeEnd - pPage->pFreeStart >= pPage->szOffset) && TDB_PAGE_FCELL(pPage)) {
+    int szCell;
+    int nxOffset;
+
+    pCell = pPage->pData + TDB_PAGE_FCELL(pPage);
+    pOffset = TDB_IS_LARGE_PAGE(pPage) ? ((SPageHdrL *)(pPage->pPageHdr))[0].fCell
+                                       : (u8 *)&(((SPageHdr *)(pPage->pPageHdr))[0].fCell);
+    szCell = TDB_PAGE_FREE_CELL_SIZE(pPage, pCell);
+    nxOffset = TDB_PAGE_FREE_CELL_NXOFFSET(pPage, pCell);
+
+    for (;;) {
+      // Find a cell
+      if (szCell >= size) {
+        if (szCell - size >= pPage->szFreeCell) {
+          SCell *pTmpCell = pCell + size;
+
+          TDB_PAGE_FREE_CELL_SIZE_SET(pPage, pTmpCell, szCell - size);
+          TDB_PAGE_FREE_CELL_NXOFFSET_SET(pPage, pTmpCell, nxOffset);
+          // TODO: *pOffset = pTmpCell - pPage->pData;
+        } else {
+          TDB_PAGE_NFREE_SET(pPage, TDB_PAGE_NFREE(pPage) + szCell - size);
+          // TODO: *pOffset = nxOffset;
+        }
+        break;
+      }
+
+      // Not find a cell yet
+      if (nxOffset > 0) {
+        pCell = pPage->pData + nxOffset;
+        pOffset = TDB_PAGE_FREE_CELL_NXOFFSET_PTR(pPage, pCell);
+        szCell = TDB_PAGE_FREE_CELL_SIZE(pPage, pCell);
+        nxOffset = TDB_PAGE_FREE_CELL_NXOFFSET(pPage, pCell);
+        continue;
+      } else {
+        pCell = NULL;
+        break;
+      }
+    }
+
+    if (pCell) {
+      pPage->pFreeStart = pPage->pFreeStart + pPage->szOffset;
+    }
+  }
+
+  // 3. Try to dfragment and allocate again
+  if (pCell == NULL) {
+    ret = tdbPageDefragment(pPage);
+    if (ret < 0) {
+      return -1;
+    }
+
+    ASSERT(pPage->pFreeEnd - pPage->pFreeStart > size + pPage->szOffset);
+    ASSERT(pPage->nFree == pPage->pFreeEnd - pPage->pFreeStart);
+
+    // Allocate from the free space area again
+    pPage->pFreeEnd -= size;
+    pPage->pFreeStart += pPage->szOffset;
+    pCell = pPage->pFreeEnd;
+  }
+
+  ASSERT(pCell != NULL);
+
+  pPage->nFree = pPage->nFree - size - pPage->szOffset;
+  *ppCell = pCell;
+  return 0;
+}
+
+static int tdbPageFree(SPage *pPage, int idx, SCell *pCell, int size) {
+  // TODO
+  return 0;
+}
+
+static int tdbPageDefragment(SPage *pPage) {
+  // TODO
+  ASSERT(0);
+  return 0;
+}
+
+/* ---------------------------------------------------------------------------------------------------------- */
+// flags
+static inline u16  getPageFlags(SPage *pPage) { return ((SPageHdr *)(pPage->pPageHdr))[0].flags; }
+static inline void setPageFlags(SPage *pPage, u16 flags) { ((SPageHdr *)(pPage->pPageHdr))[0].flags = flags; }
+
+// cellNum
+static inline int  getPageCellNum(SPage *pPage) { return ((SPageHdr *)(pPage->pPageHdr))[0].cellNum; }
+static inline void setPageCellNum(SPage *pPage, int cellNum) {
+  ASSERT(cellNum < 65536);
+  ((SPageHdr *)(pPage->pPageHdr))[0].cellNum = (u16)cellNum;
+}
+
+// cellBody
+static inline int  getPageCellBody(SPage *pPage) { return ((SPageHdr *)(pPage->pPageHdr))[0].cellBody; }
+static inline void setPageCellBody(SPage *pPage, int cellBody) {
+  ASSERT(cellBody < 65536);
+  ((SPageHdr *)(pPage->pPageHdr))[0].cellBody = (u16)cellBody;
+}
+
+// cellFree
+static inline int  getPageCellFree(SPage *pPage) { return ((SPageHdr *)(pPage->pPageHdr))[0].cellFree; }
+static inline void setPageCellFree(SPage *pPage, int cellFree) {
+  ASSERT(cellFree < 65536);
+  ((SPageHdr *)(pPage->pPageHdr))[0].cellFree = (u16)cellFree;
+}
+
+// nFree
+static inline int  getPageNFree(SPage *pPage) { return ((SPageHdr *)(pPage->pPageHdr))[0].nFree; }
+static inline void setPageNFree(SPage *pPage, int nFree) {
+  ASSERT(cellFree < 65536);
+  ((SPageHdr *)(pPage->pPageHdr))[0].nFree = (u16)nFree;
+}
+
+// cell offset
+static inline int getPageCellOffset(SPage *pPage, int idx) {
+  ASSERT(idx >= 0 && idx < getPageCellNum(pPage));
+  return ((u16 *)pPage->pCellIdx)[idx];
+}
+
+static inline void setPageCellOffset(SPage *pPage, int idx, int offset) {
+  ASSERT(offset < 65536);
+  ((u16 *)pPage->pCellIdx)[idx] = (u16)offset;
+}
+
+SPageMethods pageMethods = {
+    getPageFlags,       // getPageFlags
+    setPageFlags,       // setFlagsp
+    getPageCellNum,     // getCellNum
+    setPageCellNum,     // setCellNum
+    getPageCellBody,    // getCellBody
+    setPageCellBody,    // setCellBody
+    getPageCellFree,    // getCellFree
+    setPageCellFree,    // setCellFree
+    getPageNFree,       // getFreeBytes
+    setPageNFree,       // setFreeBytes
+    getPageCellOffset,  // getCellOffset
+    setPageCellOffset   // setCellOffset
+};
\ No newline at end of file

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

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef _TDB_PAGE_H_
+#define _TDB_PAGE_H_
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+typedef u8 SCell;
+
+// PAGE APIS implemented
+typedef struct {
+  // flags
+  u16 (*getFlags)(SPage *);
+  void (*setFlags)(SPage *, u16);
+  // cell number
+  int (*getCellNum)(SPage *);
+  void (*setCellNum)(SPage *, int);
+  // cell content offset
+  int (*getCellBody)(SPage *);
+  void (*setCellBody)(SPage *, int);
+  // first free cell offset (0 means no free cells)
+  int (*getCellFree)(SPage *);
+  void (*setCellFree)(SPage *, int);
+  // total free bytes
+  int (*getFreeBytes)(SPage *);
+  void (*setFreeBytes)(SPage *, int);
+  // cell offset at idx
+  int (*getCellOffset)(SPage *, int);
+  void (*setCellOffset)(SPage *, int, int);
+} SPageMethods;
+
+extern SPageMethods pageMethods;
+extern SPageMethods pageLargeMethods;
+
+// Page footer
+typedef struct __attribute__((__packed__)) {
+  u8 cksm[4];
+} SPageFtr;
+
+struct SPage {
+  pthread_spinlock_t lock;
+  u8                *pData;
+  int                pageSize;
+  u8                 szOffset;
+  u8                 szPageHdr;
+  u8                 szFreeCell;
+  // Fields below used by pager and am
+  u8            szAmHdr;
+  u8           *pPageHdr;
+  u8           *pAmHdr;
+  u8           *pCellIdx;
+  u8           *pFreeStart;
+  u8           *pFreeEnd;
+  SPageFtr     *pPageFtr;
+  int           kLen;  // key length of the page, -1 for unknown
+  int           vLen;  // value length of the page, -1 for unknown
+  int           nFree;
+  int           maxLocal;
+  int           minLocal;
+  int           nOverflow;
+  SCell        *apOvfl[4];
+  int           aiOvfl[4];
+  SPageMethods *pPageMethods;
+  // Fields used by SPCache
+  TDB_PCACHE_PAGE
+};
+
+/* For page */
+#define TDB_PAGE_FLAGS(pPage)               (*(pPage)->pPageMethods->getFlags)(pPage)
+#define TDB_PAGE_NCELLS(pPage)              (*(pPage)->pPageMethods->getCellNum)(pPage)
+#define TDB_PAGE_CCELLS(pPage)              (*(pPage)->pPageMethods->getCellBody)(pPage)
+#define TDB_PAGE_FCELL(pPage)               (*(pPage)->pPageMethods->getCellFree)(pPage)
+#define TDB_PAGE_NFREE(pPage)               (*(pPage)->pPageMethods->getFreeBytes)(pPage)
+#define TDB_PAGE_CELL_OFFSET_AT(pPage, idx) (*(pPage)->pPageMethods->getCellOffset)(pPage, idx)
+
+#define TDB_PAGE_FLAGS_SET(pPage, FLAGS)                (*(pPage)->pPageMethods->setFlags)(pPage, FLAGS)
+#define TDB_PAGE_NCELLS_SET(pPage, NCELLS)              (*(pPage)->pPageMethods->setCellNum)(pPage, NCELLS)
+#define TDB_PAGE_CCELLS_SET(pPage, CCELLS)              (*(pPage)->pPageMethods->setCellBody)(pPage, CCELLS)
+#define TDB_PAGE_FCELL_SET(pPage, FCELL)                (*(pPage)->pPageMethods->setCellFree)(pPage, FCELL)
+#define TDB_PAGE_NFREE_SET(pPage, NFREE)                (*(pPage)->pPageMethods->setFreeBytes)(pPage, NFREE)
+#define TDB_PAGE_CELL_OFFSET_AT_SET(pPage, idx, OFFSET) (*(pPage)->pPageMethods->setCellOffset)(pPage, idx, OFFSET)
+
+#define TDB_PAGE_CELL_AT(pPage, idx) ((pPage)->pData + TDB_PAGE_CELL_OFFSET_AT(pPage, idx))
+
+// For page lock
+#define P_LOCK_SUCC 0
+#define P_LOCK_BUSY 1
+#define P_LOCK_FAIL -1
+
+#define TDB_INIT_PAGE_LOCK(pPage)    pthread_spin_init(&((pPage)->lock), 0)
+#define TDB_DESTROY_PAGE_LOCK(pPage) pthread_spin_destroy(&((pPage)->lock))
+#define TDB_LOCK_PAGE(pPage)         pthread_spin_lock(&((pPage)->lock))
+#define TDB_UNLOCK_PAGE(pPage)       pthread_spin_unlock(&((pPage)->lock))
+#define TDB_TRY_LOCK_PAGE(pPage)                       \
+  ({                                                   \
+    int ret;                                           \
+    if (pthread_spin_trylock(&((pPage)->lock)) == 0) { \
+      ret = P_LOCK_SUCC;                               \
+    } else if (errno == EBUSY) {                       \
+      ret = P_LOCK_BUSY;                               \
+    } else {                                           \
+      ret = P_LOCK_FAIL;                               \
+    }                                                  \
+    ret;                                               \
+  })
+
+// For page ref
+#define TDB_INIT_PAGE_REF(pPage) ((pPage)->nRef = 0)
+#if 0
+#define TDB_REF_PAGE(pPage)     (++(pPage)->nRef)
+#define TDB_UNREF_PAGE(pPage)   (--(pPage)->nRef)
+#define TDB_GET_PAGE_REF(pPage) ((pPage)->nRef)
+#else
+#define TDB_REF_PAGE(pPage)     atomic_add_fetch_32(&((pPage)->nRef), 1)
+#define TDB_UNREF_PAGE(pPage)   atomic_sub_fetch_32(&((pPage)->nRef), 1)
+#define TDB_GET_PAGE_REF(pPage) atomic_load_32(&((pPage)->nRef))
+#endif
+
+// APIs
+int tdbPageCreate(int pageSize, SPage **ppPage, void *(*xMalloc)(void *, size_t), void *arg);
+int tdbPageDestroy(SPage *pPage, void (*xFree)(void *arg, void *ptr), void *arg);
+int tdbPageInsertCell(SPage *pPage, int idx, SCell *pCell, int szCell);
+int tdbPageDropCell(SPage *pPage, int idx);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /*_TDB_PAGE_H_*/
\ No newline at end of file

source/libs/tdb/src/page/tdbPageL.c

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "tdbInt.h"
+
+typedef struct __attribute__((__packed__)) {
+  u16 flags;
+  u8  cellNum[3];
+  u8  cellBody[3];
+  u8  cellFree[3];
+  u8  nFree[3];
+} SPageHdrL;
+
+typedef struct __attribute__((__packed__)) {
+  u8 szCell[3];
+  u8 nxOffset[3];
+} SFreeCellL;
+
+// flags
+static inline u16  getPageFlags(SPage *pPage) { return ((SPageHdrL *)(pPage->pPageHdr))[0].flags; }
+static inline void setPageFlags(SPage *pPage, u16 flags) { ((SPageHdrL *)(pPage->pPageHdr))[0].flags = flags; }
+
+// cellNum
+static inline int  getPageCellNum(SPage *pPage) { return TDB_GET_U24(((SPageHdrL *)(pPage->pPageHdr))[0].cellNum); }
+static inline void setPageCellNum(SPage *pPage, int cellNum) {
+  TDB_PUT_U24(((SPageHdrL *)(pPage->pPageHdr))[0].cellNum, cellNum);
+}
+
+// cellBody
+static inline int  getPageCellBody(SPage *pPage) { return TDB_GET_U24(((SPageHdrL *)(pPage->pPageHdr))[0].cellBody); }
+static inline void setPageCellBody(SPage *pPage, int cellBody) {
+  TDB_PUT_U24(((SPageHdrL *)(pPage->pPageHdr))[0].cellBody, cellBody);
+}
+
+// cellFree
+static inline int  getPageCellFree(SPage *pPage) { return TDB_GET_U24(((SPageHdrL *)(pPage->pPageHdr))[0].cellFree); }
+static inline void setPageCellFree(SPage *pPage, int cellFree) {
+  TDB_PUT_U24(((SPageHdrL *)(pPage->pPageHdr))[0].cellFree, cellFree);
+}
+
+// nFree
+static inline int  getPageNFree(SPage *pPage) { return TDB_GET_U24(((SPageHdrL *)(pPage->pPageHdr))[0].nFree); }
+static inline void setPageNFree(SPage *pPage, int nFree) {
+  TDB_PUT_U24(((SPageHdrL *)(pPage->pPageHdr))[0].nFree, nFree);
+}
+
+// cell offset
+static inline int getPageCellOffset(SPage *pPage, int idx) {
+  ASSERT(idx >= 0 && idx < getPageCellNum(pPage));
+  return TDB_GET_U24(pPage->pCellIdx + 3 * idx);
+}
+
+static inline void setPageCellOffset(SPage *pPage, int idx, int offset) {
+  TDB_PUT_U24(pPage->pCellIdx + 3 * idx, offset);
+}
+
+SPageMethods pageLargeMethods = {
+    getPageFlags,       // getPageFlags
+    setPageFlags,       // setFlagsp
+    getPageCellNum,     // getCellNum
+    setPageCellNum,     // setCellNum
+    getPageCellBody,    // getCellBody
+    setPageCellBody,    // setCellBody
+    getPageCellFree,    // getCellFree
+    setPageCellFree,    // setCellFree
+    getPageNFree,       // getFreeBytes
+    setPageNFree,       // setFreeBytes
+    getPageCellOffset,  // getCellOffset
+    setPageCellOffset   // setCellOffset
+};
\ No newline at end of file