xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

more TDB

This commit is contained in:
Hongze Cheng 2022-03-18 09:35:09 +00:00
parent 2a597151c9
commit 465dac8234
5 changed files with 65 additions and 272 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
source/libs/tdb/CMakeLists.txt
View File

@ -8,7 +8,9 @@ target_sources(tdb
"src/db/tdbBtree.c" "src/db/tdbBtree.c"
"src/db/tdbDb.c" "src/db/tdbDb.c"
"src/db/tdbEnv.c" "src/db/tdbEnv.c"
"src/db/tdbPage.c" # "src/db/tdbPage.c"
"src/page/tdbPage.c"
"src/page/tdbPageL.c"
) )
target_include_directories( target_include_directories(

14
source/libs/tdb/src/db/tdbBtree.c
View File

@ -96,11 +96,11 @@ int tdbBtreeOpen(int keyLen, int valLen, SPager *pPager, FKeyComparator kcmpr, S
// pBt->pageSize // pBt->pageSize
pBt->pageSize = tdbPagerGetPageSize(pPager); pBt->pageSize = tdbPagerGetPageSize(pPager);
// pBt->maxLocal // pBt->maxLocal
pBt->maxLocal = (pBt->pageSize - sizeof(SPageHdr)) / pBt->fanout; pBt->maxLocal = (pBt->pageSize - 14) / pBt->fanout;
// pBt->minLocal: Should not be allowed smaller than 15, which is [nPayload][nKey][nData] // pBt->minLocal: Should not be allowed smaller than 15, which is [nPayload][nKey][nData]
pBt->minLocal = (pBt->pageSize - sizeof(SPageHdr)) / pBt->fanout / 2; pBt->minLocal = (pBt->pageSize - 14) / pBt->fanout / 2;
// pBt->maxLeaf // pBt->maxLeaf
pBt->maxLeaf = pBt->pageSize - sizeof(SPageHdr); pBt->maxLeaf = pBt->pageSize - 14;
// pBt->minLeaf // pBt->minLeaf
pBt->minLeaf = pBt->minLocal; pBt->minLeaf = pBt->minLocal;
@ -392,9 +392,9 @@ static int tdbBtreeInitPage(SPage *pPage, void *arg) {
pPage->szAmHdr = sizeof(SBtPageHdr); pPage->szAmHdr = sizeof(SBtPageHdr);
} }
pPage->pPageHdr = pPage->pData; pPage->pPageHdr = pPage->pData;
pPage->pAmHdr = pPage->pPageHdr + pPage->szPageHdr; pPage->pAmHdr = pPage->pPageHdr + pPage->pPageMethods->szPageHdr;
pPage->pCellIdx = pPage->pAmHdr + pPage->szAmHdr; pPage->pCellIdx = pPage->pAmHdr + pPage->szAmHdr;
pPage->pFreeStart = pPage->pCellIdx + pPage->szOffset * TDB_PAGE_NCELLS(pPage); pPage->pFreeStart = pPage->pCellIdx + pPage->pPageMethods->szOffset * TDB_PAGE_NCELLS(pPage);
pPage->pFreeEnd = pPage->pData + TDB_PAGE_CCELLS(pPage); pPage->pFreeEnd = pPage->pData + TDB_PAGE_CCELLS(pPage);
pPage->pPageFtr = (SPageFtr *)(pPage->pData + pPage->pageSize - sizeof(SPageFtr)); pPage->pPageFtr = (SPageFtr *)(pPage->pData + pPage->pageSize - sizeof(SPageFtr));
@ -458,8 +458,8 @@ static int tdbBtreeCopyPageContent(SPage *pFrom, SPage *pTo) {
int fCell = TDB_PAGE_FCELL(pFrom); int fCell = TDB_PAGE_FCELL(pFrom);
int nFree = TDB_PAGE_NFREE(pFrom); int nFree = TDB_PAGE_NFREE(pFrom);
pTo->pFreeStart = pTo->pCellIdx + nCells * pFrom->szOffset; pTo->pFreeStart = pTo->pCellIdx + nCells * pFrom->pPageMethods->szOffset;
memcpy(pTo->pCellIdx, pFrom->pCellIdx, nCells * pFrom->szOffset); memcpy(pTo->pCellIdx, pFrom->pCellIdx, nCells * pFrom->pPageMethods->szOffset);
pTo->pFreeEnd = (u8 *)pTo->pPageFtr - (u8 *)(pFrom->pPageFtr) + pFrom->pFreeEnd; pTo->pFreeEnd = (u8 *)pTo->pPageFtr - (u8 *)(pFrom->pPageFtr) + pFrom->pFreeEnd;
memcpy(pTo->pFreeEnd, pFrom->pFreeEnd, (u8 *)pFrom->pPageFtr - pFrom->pFreeEnd); memcpy(pTo->pFreeEnd, pFrom->pFreeEnd, (u8 *)pFrom->pPageFtr - pFrom->pFreeEnd);

150
source/libs/tdb/src/inc/tdbPage.h
View File

@ -22,23 +22,30 @@ extern "C" {
typedef u8 SCell; typedef u8 SCell;
// Page header (pageSize < 65536 (64K)) // PAGE APIS implemented
typedef struct __attribute__((__packed__)) { typedef struct {
u16 flags; int szOffset;
u16 nCells; // number of cells int szPageHdr;
u16 cCells; // cell content offset int szFreeCell;
u16 fCell; // first free cell offset // flags
u16 nFree; // total fragment bytes in this page u16 (*getFlags)(SPage *);
} SPageHdr; void (*setFlags)(SPage *, u16);
// cell number
// Large page header (pageSize >= 65536 (64K)) int (*getCellNum)(SPage *);
typedef struct __attribute__((__packed__)) { void (*setCellNum)(SPage *, int);
u16 flags; // cell content offset
u8 nCells[3]; int (*getCellBody)(SPage *);
u8 cCells[3]; void (*setCellBody)(SPage *, int);
u8 fCell[3]; // first free cell offset (0 means no free cells)
u8 nFree[3]; int (*getCellFree)(SPage *);
} SPageHdrL; 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;
// Page footer // Page footer
typedef struct __attribute__((__packed__)) { typedef struct __attribute__((__packed__)) {
@ -49,9 +56,7 @@ struct SPage {
pthread_spinlock_t lock; pthread_spinlock_t lock;
u8 *pData; u8 *pData;
int pageSize; int pageSize;
u8 szOffset; SPageMethods *pPageMethods;
u8 szPageHdr;
u8 szFreeCell;
// Fields below used by pager and am // Fields below used by pager and am
u8 szAmHdr; u8 szAmHdr;
u8 *pPageHdr; u8 *pPageHdr;
@ -72,101 +77,22 @@ struct SPage {
TDB_PCACHE_PAGE TDB_PCACHE_PAGE
}; };
// Macros
#define TDB_IS_LARGE_PAGE(pPage) ((pPage)->szOffset == 3)
/* For small page */
#define TDB_SPAGE_FLAGS(pPage) (((SPageHdr *)(pPage)->pPageHdr)->flags)
#define TDB_SPAGE_NCELLS(pPage) (((SPageHdr *)(pPage)->pPageHdr)->nCells)
#define TDB_SPAGE_CCELLS(pPage) (((SPageHdr *)(pPage)->pPageHdr)->cCells)
#define TDB_SPAGE_FCELL(pPage) (((SPageHdr *)(pPage)->pPageHdr)->fCell)
#define TDB_SPAGE_NFREE(pPage) (((SPageHdr *)(pPage)->pPageHdr)->nFree)
#define TDB_SPAGE_CELL_OFFSET_AT(pPage, idx) ((u16 *)((pPage)->pCellIdx))[idx]
#define TDB_SPAGE_FLAGS_SET(pPage, FLAGS) TDB_SPAGE_FLAGS(pPage) = (FLAGS)
#define TDB_SPAGE_NCELLS_SET(pPage, NCELLS) TDB_SPAGE_NCELLS(pPage) = (NCELLS)
#define TDB_SPAGE_CCELLS_SET(pPage, CCELLS) TDB_SPAGE_CCELLS(pPage) = (CCELLS)
#define TDB_SPAGE_FCELL_SET(pPage, FCELL) TDB_SPAGE_FCELL(pPage) = (FCELL)
#define TDB_SPAGE_NFREE_SET(pPage, NFREE) TDB_SPAGE_NFREE(pPage) = (NFREE)
#define TDB_SPAGE_CELL_OFFSET_AT_SET(pPage, idx, OFFSET) TDB_SPAGE_CELL_OFFSET_AT(pPage, idx) = (OFFSET)
/* For large page */
#define TDB_LPAGE_FLAGS(pPage) (((SPageHdrL *)(pPage)->pPageHdr)->flags)
#define TDB_LPAGE_NCELLS(pPage) TDB_GET_U24(((SPageHdrL *)(pPage)->pPageHdr)->nCells)
#define TDB_LPAGE_CCELLS(pPage) TDB_GET_U24(((SPageHdrL *)(pPage)->pPageHdr)->cCells)
#define TDB_LPAGE_FCELL(pPage) TDB_GET_U24(((SPageHdrL *)(pPage)->pPageHdr)->fCell)
#define TDB_LPAGE_NFREE(pPage) TDB_GET_U24(((SPageHdrL *)(pPage)->pPageHdr)->nFree)
#define TDB_LPAGE_CELL_OFFSET_AT(pPage, idx) TDB_GET_U24((pPage)->pCellIdx + idx * 3)
#define TDB_LPAGE_FLAGS_SET(pPage, FLAGS) TDB_LPAGE_FLAGS(pPage) = (flags)
#define TDB_LPAGE_NCELLS_SET(pPage, NCELLS) TDB_PUT_U24(((SPageHdrL *)(pPage)->pPageHdr)->nCells, NCELLS)
#define TDB_LPAGE_CCELLS_SET(pPage, CCELLS) TDB_PUT_U24(((SPageHdrL *)(pPage)->pPageHdr)->cCells, CCELLS)
#define TDB_LPAGE_FCELL_SET(pPage, FCELL) TDB_PUT_U24(((SPageHdrL *)(pPage)->pPageHdr)->fCell, FCELL)
#define TDB_LPAGE_NFREE_SET(pPage, NFREE) TDB_PUT_U24(((SPageHdrL *)(pPage)->pPageHdr)->nFree, NFREE)
#define TDB_LPAGE_CELL_OFFSET_AT_SET(pPage, idx, OFFSET) TDB_PUT_U24((pPage)->pCellIdx + idx * 3, OFFSET)
/* For page */ /* For page */
#define TDB_PAGE_FLAGS(pPage) (TDB_IS_LARGE_PAGE(pPage) ? TDB_LPAGE_FLAGS(pPage) : TDB_SPAGE_FLAGS(pPage)) #define TDB_PAGE_FLAGS(pPage) (*(pPage)->pPageMethods->getFlags)(pPage)
#define TDB_PAGE_NCELLS(pPage) (TDB_IS_LARGE_PAGE(pPage) ? TDB_LPAGE_NCELLS(pPage) : TDB_SPAGE_NCELLS(pPage)) #define TDB_PAGE_NCELLS(pPage) (*(pPage)->pPageMethods->getCellNum)(pPage)
#define TDB_PAGE_CCELLS(pPage) (TDB_IS_LARGE_PAGE(pPage) ? TDB_LPAGE_CCELLS(pPage) : TDB_SPAGE_CCELLS(pPage)) #define TDB_PAGE_CCELLS(pPage) (*(pPage)->pPageMethods->getCellBody)(pPage)
#define TDB_PAGE_FCELL(pPage) (TDB_IS_LARGE_PAGE(pPage) ? TDB_LPAGE_FCELL(pPage) : TDB_SPAGE_FCELL(pPage)) #define TDB_PAGE_FCELL(pPage) (*(pPage)->pPageMethods->getCellFree)(pPage)
#define TDB_PAGE_NFREE(pPage) (TDB_IS_LARGE_PAGE(pPage) ? TDB_LPAGE_NFREE(pPage) : TDB_SPAGE_NFREE(pPage)) #define TDB_PAGE_NFREE(pPage) (*(pPage)->pPageMethods->getFreeBytes)(pPage)
#define TDB_PAGE_CELL_OFFSET_AT(pPage, idx) \ #define TDB_PAGE_CELL_OFFSET_AT(pPage, idx) (*(pPage)->pPageMethods->getCellOffset)(pPage, idx)
(TDB_IS_LARGE_PAGE(pPage) ? TDB_LPAGE_CELL_OFFSET_AT(pPage, idx) : TDB_SPAGE_CELL_OFFSET_AT(pPage, idx))
#define TDB_PAGE_FLAGS_SET(pPage, FLAGS) \ #define TDB_PAGE_FLAGS_SET(pPage, FLAGS) (*(pPage)->pPageMethods->setFlags)(pPage, FLAGS)
do { \ #define TDB_PAGE_NCELLS_SET(pPage, NCELLS) (*(pPage)->pPageMethods->setCellNum)(pPage, NCELLS)
if (TDB_IS_LARGE_PAGE(pPage)) { \ #define TDB_PAGE_CCELLS_SET(pPage, CCELLS) (*(pPage)->pPageMethods->setCellBody)(pPage, CCELLS)
TDB_LPAGE_FLAGS_SET(pPage, FLAGS); \ #define TDB_PAGE_FCELL_SET(pPage, FCELL) (*(pPage)->pPageMethods->setCellFree)(pPage, FCELL)
} else { \ #define TDB_PAGE_NFREE_SET(pPage, NFREE) (*(pPage)->pPageMethods->setFreeBytes)(pPage, NFREE)
TDB_SPAGE_FLAGS_SET(pPage, FLAGS); \ #define TDB_PAGE_CELL_OFFSET_AT_SET(pPage, idx, OFFSET) (*(pPage)->pPageMethods->setCellOffset)(pPage, idx, OFFSET)
} \
} while (0)
#define TDB_PAGE_NCELLS_SET(pPage, NCELLS) \ #define TDB_PAGE_OFFSET_SIZE(pPage) ((pPage)->pPageMethods->szOffset)
do { \
if (TDB_IS_LARGE_PAGE(pPage)) { \
TDB_LPAGE_NCELLS_SET(pPage, NCELLS); \
} else { \
TDB_SPAGE_NCELLS_SET(pPage, NCELLS); \
} \
} while (0)
#define TDB_PAGE_CCELLS_SET(pPage, CCELLS) \
do { \
if (TDB_IS_LARGE_PAGE(pPage)) { \
TDB_LPAGE_CCELLS_SET(pPage, CCELLS); \
} else { \
TDB_SPAGE_CCELLS_SET(pPage, CCELLS); \
} \
} while (0)
#define TDB_PAGE_FCELL_SET(pPage, FCELL) \
do { \
if (TDB_IS_LARGE_PAGE(pPage)) { \
TDB_LPAGE_FCELL_SET(pPage, FCELL); \
} else { \
TDB_SPAGE_FCELL_SET(pPage, FCELL); \
} \
} while (0)
#define TDB_PAGE_NFREE_SET(pPage, NFREE) \
do { \
if (TDB_IS_LARGE_PAGE(pPage)) { \
TDB_LPAGE_NFREE_SET(pPage, NFREE); \
} else { \
TDB_SPAGE_NFREE_SET(pPage, NFREE); \
} \
} while (0)
#define TDB_PAGE_CELL_OFFSET_AT_SET(pPage, idx, OFFSET) \
do { \
if (TDB_IS_LARGE_PAGE(pPage)) { \
TDB_LPAGE_CELL_OFFSET_AT_SET(pPage, idx, OFFSET); \
} else { \
TDB_SPAGE_CELL_OFFSET_AT_SET(pPage, idx, OFFSET); \
} \
} while (0)
#define TDB_PAGE_CELL_AT(pPage, idx) ((pPage)->pData + TDB_PAGE_CELL_OFFSET_AT(pPage, idx)) #define TDB_PAGE_CELL_AT(pPage, idx) ((pPage)->pData + TDB_PAGE_CELL_OFFSET_AT(pPage, idx))

28
source/libs/tdb/src/page/tdbPage.c
View File

@ -31,6 +31,9 @@ typedef struct __attribute__((__packed__)) {
u8 nxOffset[2]; u8 nxOffset[2];
} SFreeCell; } SFreeCell;
static int tdbPageAllocate(SPage *pPage, int size, SCell **ppCell);
static int tdbPageDefragment(SPage *pPage);
int tdbPageCreate(int pageSize, SPage **ppPage, void *(*xMalloc)(void *, size_t), void *arg) { int tdbPageCreate(int pageSize, SPage **ppPage, void *(*xMalloc)(void *, size_t), void *arg) {
SPage *pPage; SPage *pPage;
u8 *ptr; u8 *ptr;
@ -79,7 +82,7 @@ int tdbPageInsertCell(SPage *pPage, int idx, SCell *pCell, int szCell) {
u8 *pTmp; u8 *pTmp;
int j; int j;
if (pPage->nOverflow || szCell + pPage->szOffset > pPage->nFree) { if (pPage->nOverflow || szCell + TDB_PAGE_OFFSET_SIZE(pPage) > pPage->nFree) {
// TODO: need to figure out if pCell may be used by outside of this function // TODO: need to figure out if pCell may be used by outside of this function
j = pPage->nOverflow++; j = pPage->nOverflow++;
@ -92,8 +95,8 @@ int tdbPageInsertCell(SPage *pPage, int idx, SCell *pCell, int szCell) {
} }
memcpy(pTarget, pCell, szCell); memcpy(pTarget, pCell, szCell);
pTmp = pPage->pCellIdx + idx * pPage->szOffset; pTmp = pPage->pCellIdx + idx * TDB_PAGE_OFFSET_SIZE(pPage);
memmove(pTmp + pPage->szOffset, pTmp, pPage->pFreeStart - pTmp - pPage->szOffset); memmove(pTmp + TDB_PAGE_OFFSET_SIZE(pPage), pTmp, pPage->pFreeStart - pTmp - TDB_PAGE_OFFSET_SIZE(pPage));
TDB_PAGE_CELL_OFFSET_AT_SET(pPage, idx, pTarget - pPage->pData); TDB_PAGE_CELL_OFFSET_AT_SET(pPage, idx, pTarget - pPage->pData);
TDB_PAGE_NCELLS_SET(pPage, TDB_PAGE_NCELLS(pPage) + 1); TDB_PAGE_NCELLS_SET(pPage, TDB_PAGE_NCELLS(pPage) + 1);
} }
@ -112,20 +115,22 @@ static int tdbPageAllocate(SPage *pPage, int size, SCell **ppCell) {
u8 *pOffset; u8 *pOffset;
int ret; int ret;
ASSERT(pPage->nFree > size + pPage->szOffset); ASSERT(pPage->nFree > size + TDB_PAGE_OFFSET_SIZE(pPage));
pCell = NULL; pCell = NULL;
*ppCell = NULL; *ppCell = NULL;
// 1. Try to allocate from the free space area // 1. Try to allocate from the free space area
if (pPage->pFreeEnd - pPage->pFreeStart > size + pPage->szOffset) { if (pPage->pFreeEnd - pPage->pFreeStart > size + TDB_PAGE_OFFSET_SIZE(pPage)) {
pPage->pFreeEnd -= size; pPage->pFreeEnd -= size;
pPage->pFreeStart += pPage->szOffset; pPage->pFreeStart += TDB_PAGE_OFFSET_SIZE(pPage);
pCell = pPage->pFreeEnd; pCell = pPage->pFreeEnd;
} }
// 2. Try to allocate from the page free list // 2. Try to allocate from the page free list
if ((pCell == NULL) && (pPage->pFreeEnd - pPage->pFreeStart >= pPage->szOffset) && TDB_PAGE_FCELL(pPage)) { if ((pCell == NULL) && (pPage->pFreeEnd - pPage->pFreeStart >= TDB_PAGE_OFFSET_SIZE(pPage)) &&
TDB_PAGE_FCELL(pPage)) {
#if 0
int szCell; int szCell;
int nxOffset; int nxOffset;
@ -167,6 +172,7 @@ static int tdbPageAllocate(SPage *pPage, int size, SCell **ppCell) {
if (pCell) { if (pCell) {
pPage->pFreeStart = pPage->pFreeStart + pPage->szOffset; pPage->pFreeStart = pPage->pFreeStart + pPage->szOffset;
} }
#endif
} }
// 3. Try to dfragment and allocate again // 3. Try to dfragment and allocate again
@ -176,18 +182,18 @@ static int tdbPageAllocate(SPage *pPage, int size, SCell **ppCell) {
return -1; return -1;
} }
ASSERT(pPage->pFreeEnd - pPage->pFreeStart > size + pPage->szOffset); ASSERT(pPage->pFreeEnd - pPage->pFreeStart > size + TDB_PAGE_OFFSET_SIZE(pPage));
ASSERT(pPage->nFree == pPage->pFreeEnd - pPage->pFreeStart); ASSERT(pPage->nFree == pPage->pFreeEnd - pPage->pFreeStart);
// Allocate from the free space area again // Allocate from the free space area again
pPage->pFreeEnd -= size; pPage->pFreeEnd -= size;
pPage->pFreeStart += pPage->szOffset; pPage->pFreeStart += TDB_PAGE_OFFSET_SIZE(pPage);
pCell = pPage->pFreeEnd; pCell = pPage->pFreeEnd;
} }
ASSERT(pCell != NULL); ASSERT(pCell != NULL);
pPage->nFree = pPage->nFree - size - pPage->szOffset; pPage->nFree = pPage->nFree - size - TDB_PAGE_OFFSET_SIZE(pPage);
*ppCell = pCell; *ppCell = pCell;
return 0; return 0;
} }
@ -232,7 +238,7 @@ static inline void setPageCellFree(SPage *pPage, int cellFree) {
// nFree // nFree
static inline int getPageNFree(SPage *pPage) { return ((SPageHdr *)(pPage->pPageHdr))[0].nFree; } static inline int getPageNFree(SPage *pPage) { return ((SPageHdr *)(pPage->pPageHdr))[0].nFree; }
static inline void setPageNFree(SPage *pPage, int nFree) { static inline void setPageNFree(SPage *pPage, int nFree) {
ASSERT(cellFree < 65536); ASSERT(nFree < 65536);
((SPageHdr *)(pPage->pPageHdr))[0].nFree = (u16)nFree; ((SPageHdr *)(pPage->pPageHdr))[0].nFree = (u16)nFree;
} }

141
source/libs/tdb/src/page/tdbPage.h
View File

@ -1,141 +0,0 @@
/*
* 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 {
int szOffset;
int szPageHdr;
int szFreeCell;
// 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;
// Page footer
typedef struct __attribute__((__packed__)) {
u8 cksm[4];
} SPageFtr;
struct SPage {
pthread_spinlock_t lock;
u8 *pData;
int pageSize;
SPageMethods *pPageMethods;
// 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];
// 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_*/