From d3446a5c7aa1daeabba0bf3959d7ad9bc8da46c9 Mon Sep 17 00:00:00 2001 From: Hongze Cheng Date: Sun, 20 Mar 2022 09:32:21 +0000 Subject: [PATCH] more TDB --- source/libs/tdb/src/db/tdbBtree.c | 2 +- source/libs/tdb/src/db/tdbPage.c | 253 ---------------------------- source/libs/tdb/src/inc/tdbPage.h | 26 +-- source/libs/tdb/src/page/tdbPage.c | 244 +++++++++++++++++---------- source/libs/tdb/src/page/tdbPageL.c | 45 +++-- 5 files changed, 198 insertions(+), 372 deletions(-) delete mode 100644 source/libs/tdb/src/db/tdbPage.c diff --git a/source/libs/tdb/src/db/tdbBtree.c b/source/libs/tdb/src/db/tdbBtree.c index c6e8c9dca9..423de0b53c 100644 --- a/source/libs/tdb/src/db/tdbBtree.c +++ b/source/libs/tdb/src/db/tdbBtree.c @@ -414,7 +414,7 @@ static int tdbBtreeInitPage(SPage *pPage, void *arg) { } // TODO: need to update the SPage.nFree - pPage->nFree = pPage->pFreeEnd - pPage->pFreeStart; + // pPage->nFree = pPage->pFreeEnd - pPage->pFreeStart; pPage->nOverflow = 0; return 0; diff --git a/source/libs/tdb/src/db/tdbPage.c b/source/libs/tdb/src/db/tdbPage.c deleted file mode 100644 index df158de756..0000000000 --- a/source/libs/tdb/src/db/tdbPage.c +++ /dev/null @@ -1,253 +0,0 @@ -/* - * Copyright (c) 2019 TAOS Data, Inc. - * - * 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 . - */ - -#include "tdbInt.h" - -typedef struct __attribute__((__packed__)) { - u8 szCell[2]; - u8 nxOffset[2]; -} SFreeCell; - -typedef struct __attribute__((__packed__)) { - u8 szCell[3]; - u8 nxOffset[3]; -} SFreeCellL; - -/* For small page */ -#define TDB_SPAGE_FREE_CELL_SIZE_PTR(PCELL) (((SFreeCell *)(PCELL))->szCell) -#define TDB_SPAGE_FREE_CELL_NXOFFSET_PTR(PCELL) (((SFreeCell *)(PCELL))->nxOffset) - -#define TDB_SPAGE_FREE_CELL_SIZE(PCELL) ((u16 *)TDB_SPAGE_FREE_CELL_SIZE_PTR(PCELL))[0] -#define TDB_SPAGE_FREE_CELL_NXOFFSET(PCELL) ((u16 *)TDB_SPAGE_FREE_CELL_NXOFFSET_PTR(PCELL))[0] - -#define TDB_SPAGE_FREE_CELL_SIZE_SET(PCELL, SIZE) (TDB_SPAGE_FREE_CELL_SIZE(PCELL) = (SIZE)) -#define TDB_SPAGE_FREE_CELL_NXOFFSET_SET(PCELL, OFFSET) (TDB_SPAGE_FREE_CELL_NXOFFSET(PCELL) = (OFFSET)) - -/* For large page */ -#define TDB_LPAGE_FREE_CELL_SIZE_PTR(PCELL) (((SFreeCellL *)(PCELL))->szCell) -#define TDB_LPAGE_FREE_CELL_NXOFFSET_PTR(PCELL) (((SFreeCellL *)(PCELL))->nxOffset) - -#define TDB_LPAGE_FREE_CELL_SIZE(PCELL) TDB_GET_U24(TDB_LPAGE_FREE_CELL_SIZE_PTR(PCELL)) -#define TDB_LPAGE_FREE_CELL_NXOFFSET(PCELL) TDB_GET_U24(TDB_LPAGE_FREE_CELL_NXOFFSET_PTR(PCELL)) - -#define TDB_LPAGE_FREE_CELL_SIZE_SET(PCELL, SIZE) TDB_PUT_U24(TDB_LPAGE_FREE_CELL_SIZE_PTR(PCELL), SIZE) -#define TDB_LPAGE_FREE_CELL_NXOFFSET_SET(PCELL, OFFSET) TDB_PUT_U24(TDB_LPAGE_FREE_CELL_NXOFFSET_PTR(PCELL), OFFSET) - -/* For page */ -#define TDB_PAGE_FREE_CELL_SIZE_PTR(PPAGE, PCELL) \ - (TDB_IS_LARGE_PAGE(pPage) ? TDB_LPAGE_FREE_CELL_SIZE_PTR(PCELL) : TDB_SPAGE_FREE_CELL_SIZE_PTR(PCELL)) -#define TDB_PAGE_FREE_CELL_NXOFFSET_PTR(PPAGE, PCELL) \ - (TDB_IS_LARGE_PAGE(pPage) ? TDB_LPAGE_FREE_CELL_NXOFFSET_PTR(PCELL) : TDB_SPAGE_FREE_CELL_NXOFFSET_PTR(PCELL)) - -#define TDB_PAGE_FREE_CELL_SIZE(PPAGE, PCELL) \ - (TDB_IS_LARGE_PAGE(pPage) ? TDB_LPAGE_FREE_CELL_SIZE(PCELL) : TDB_SPAGE_FREE_CELL_SIZE(PCELL)) -#define TDB_PAGE_FREE_CELL_NXOFFSET(PPAGE, PCELL) \ - (TDB_IS_LARGE_PAGE(pPage) ? TDB_LPAGE_FREE_CELL_NXOFFSET(PCELL) : TDB_SPAGE_FREE_CELL_NXOFFSET(PCELL)) - -#define TDB_PAGE_FREE_CELL_SIZE_SET(PPAGE, PCELL, SIZE) \ - do { \ - if (TDB_IS_LARGE_PAGE(PPAGE)) { \ - TDB_LPAGE_FREE_CELL_SIZE_SET(PCELL, SIZE); \ - } else { \ - TDB_SPAGE_FREE_CELL_SIZE_SET(PCELL, SIZE); \ - } \ - } while (0) -#define TDB_PAGE_FREE_CELL_NXOFFSET_SET(PPAGE, PCELL, OFFSET) \ - do { \ - if (TDB_IS_LARGE_PAGE(PPAGE)) { \ - TDB_LPAGE_FREE_CELL_NXOFFSET_SET(PCELL, OFFSET); \ - } else { \ - TDB_SPAGE_FREE_CELL_NXOFFSET_SET(PCELL, OFFSET); \ - } \ - } while (0) - -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) { - 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; -} \ No newline at end of file diff --git a/source/libs/tdb/src/inc/tdbPage.h b/source/libs/tdb/src/inc/tdbPage.h index 6479574abe..418a1b6626 100644 --- a/source/libs/tdb/src/inc/tdbPage.h +++ b/source/libs/tdb/src/inc/tdbPage.h @@ -45,6 +45,9 @@ typedef struct { // cell offset at idx int (*getCellOffset)(SPage *, int); void (*setCellOffset)(SPage *, int, int); + // free cell info + void (*getFreeCellInfo)(SCell *pCell, int *szCell, int *nxOffset); + void (*setFreeCellInfo)(SCell *pCell, int szCell, int nxOffset); } SPageMethods; // Page footer @@ -59,20 +62,19 @@ struct SPage { SPageMethods *pPageMethods; // Fields below used by pager and am u8 szAmHdr; - u8 *pPageHdr; u8 *pAmHdr; + u8 *pPageHdr; 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]; + int kLen; // key length of the page, -1 for unknown + int vLen; // value length of the page, -1 for unknown + int maxLocal; + int minLocal; // Fields used by SPCache TDB_PCACHE_PAGE }; @@ -92,8 +94,6 @@ struct SPage { #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_OFFSET_SIZE(pPage) ((pPage)->pPageMethods->szOffset) - #define TDB_PAGE_CELL_AT(pPage, idx) ((pPage)->pData + TDB_PAGE_CELL_OFFSET_AT(pPage, idx)) // For page lock @@ -119,10 +119,12 @@ struct SPage { }) // 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); +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); +void tdbPageZero(SPage *pPage); +void tdbPageInit(SPage *pPage); +int tdbPageInsertCell(SPage *pPage, int idx, SCell *pCell, int szCell); +int tdbPageDropCell(SPage *pPage, int idx); #ifdef __cplusplus } diff --git a/source/libs/tdb/src/page/tdbPage.c b/source/libs/tdb/src/page/tdbPage.c index 4ec3a895e7..e198891338 100644 --- a/source/libs/tdb/src/page/tdbPage.c +++ b/source/libs/tdb/src/page/tdbPage.c @@ -18,10 +18,11 @@ extern SPageMethods pageMethods; extern SPageMethods pageLargeMethods; -typedef struct __attribute__((__packed__)) { - u16 szCell; - u16 nxOffset; -} SFreeCell; +#define TDB_PAGE_OFFSET_SIZE(pPage) ((pPage)->pPageMethods->szOffset) +#define TDB_PAGE_HDR_SIZE(pPage) ((pPage)->pPageMethods->szPageHdr) +#define TDB_PAGE_FREE_CELL_SIZE(pPage) ((pPage)->pPageMethods->szFreeCell) +#define TDB_PAGE_MAX_FREE_BLOCK(pPage) \ + ((pPage)->pageSize - (pPage)->szAmHdr - TDB_PAGE_HDR_SIZE(pPage) - sizeof(SPageFtr)) static int tdbPageAllocate(SPage *pPage, int size, SCell **ppCell); static int tdbPageDefragment(SPage *pPage); @@ -44,6 +45,7 @@ int tdbPageCreate(int pageSize, SPage **ppPage, void *(*xMalloc)(void *, size_t) memset(ptr, 0, size); pPage = (SPage *)(ptr + pageSize); + TDB_INIT_PAGE_LOCK(pPage); pPage->pData = ptr; pPage->pageSize = pageSize; if (pageSize < 65536) { @@ -51,9 +53,6 @@ int tdbPageCreate(int pageSize, SPage **ppPage, void *(*xMalloc)(void *, size_t) } else { pPage->pPageMethods = &pageLargeMethods; } - TDB_INIT_PAGE_LOCK(pPage); - - /* TODO */ *ppPage = pPage; return 0; @@ -68,29 +67,64 @@ int tdbPageDestroy(SPage *pPage, void (*xFree)(void *arg, void *ptr), void *arg) return 0; } +void tdbPageZero(SPage *pPage) { + TDB_PAGE_NCELLS_SET(pPage, 0); + TDB_PAGE_CCELLS_SET(pPage, pPage->pageSize - sizeof(SPageFtr)); + TDB_PAGE_FCELL_SET(pPage, 0); + TDB_PAGE_NFREE_SET(pPage, TDB_PAGE_MAX_FREE_BLOCK(pPage)); + tdbPageInit(pPage); +} + +void tdbPageInit(SPage *pPage) { + pPage->pAmHdr = pPage->pData; + pPage->pPageHdr = pPage->pAmHdr + pPage->szAmHdr; + pPage->pCellIdx = pPage->pPageHdr + TDB_PAGE_HDR_SIZE(pPage); + pPage->pFreeStart = pPage->pCellIdx + TDB_PAGE_OFFSET_SIZE(pPage) * TDB_PAGE_NCELLS(pPage); + pPage->pFreeEnd = pPage->pData + TDB_PAGE_CCELLS(pPage); + pPage->pPageFtr = (SPageFtr *)(pPage->pData + pPage->pageSize - sizeof(SPageFtr)); + pPage->nOverflow = 0; + + ASSERT(pPage->pFreeEnd >= pPage->pFreeStart); + ASSERT(pPage->pFreeEnd - pPage->pFreeStart <= TDB_PAGE_NFREE(pPage)); +} + int tdbPageInsertCell(SPage *pPage, int idx, SCell *pCell, int szCell) { + int nFree; int ret; - SCell *pTarget; - u8 *pTmp; - int j; + int nCells; + int lidx; // local idx + SCell *pNewCell; - 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 - j = pPage->nOverflow++; + ASSERT(szCell <= TDB_PAGE_MAX_FREE_BLOCK(pPage)); - pPage->apOvfl[j] = pCell; - pPage->aiOvfl[j] = idx; - } else { - ret = tdbPageAllocate(pPage, szCell, &pTarget); - if (ret < 0) { - return -1; + nFree = TDB_PAGE_NFREE(pPage); + nCells = TDB_PAGE_NCELLS(pPage); + + if (nFree >= szCell + TDB_PAGE_OFFSET_SIZE(pPage)) { + // page must has enough space to hold the cell locally + ret = tdbPageAllocate(pPage, szCell, &pNewCell); + ASSERT(ret == 0); + + memcpy(pNewCell, pCell, szCell); + + if (pPage->nOverflow == 0) { + lidx = idx; + } else { + // TODO + // lidx = ; } - memcpy(pTarget, pCell, szCell); - pTmp = pPage->pCellIdx + idx * TDB_PAGE_OFFSET_SIZE(pPage); - 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_NCELLS_SET(pPage, TDB_PAGE_NCELLS(pPage) + 1); + // no overflow cell exists in this page + u8 *src = pPage->pCellIdx + TDB_PAGE_OFFSET_SIZE(pPage) * lidx; + u8 *dest = src + TDB_PAGE_OFFSET_SIZE(pPage); + memmove(dest, src, pPage->pFreeStart - dest); + TDB_PAGE_CELL_OFFSET_AT_SET(pPage, lidx, TDB_PAGE_OFFSET_SIZE(pPage) * lidx); + TDB_PAGE_NCELLS_SET(pPage, nCells + 1); + } else { + // TODO: page not has enough space + pPage->apOvfl[pPage->nOverflow] = pCell; + pPage->aiOvfl[pPage->nOverflow] = idx; + pPage->nOverflow++; } return 0; @@ -101,73 +135,80 @@ int tdbPageDropCell(SPage *pPage, int idx) { return 0; } -static int tdbPageAllocate(SPage *pPage, int size, SCell **ppCell) { - SCell *pCell; - SFreeCell *pFreeCell; - u8 *pOffset; - int ret; +static int tdbPageAllocate(SPage *pPage, int szCell, SCell **ppCell) { + SCell *pFreeCell; + u8 *pOffset; + int nFree; + int ret; + int cellFree; - ASSERT(pPage->nFree > size + TDB_PAGE_OFFSET_SIZE(pPage)); - - pCell = NULL; *ppCell = NULL; + nFree = TDB_PAGE_NFREE(pPage); - // 1. Try to allocate from the free space area - if (pPage->pFreeEnd - pPage->pFreeStart > size + TDB_PAGE_OFFSET_SIZE(pPage)) { - pPage->pFreeEnd -= size; + ASSERT(nFree >= szCell + TDB_PAGE_OFFSET_SIZE(pPage)); + ASSERT(TDB_PAGE_CCELLS(pPage) == pPage->pFreeEnd - pPage->pData); + + // 1. Try to allocate from the free space block area + if (pPage->pFreeEnd - pPage->pFreeStart >= szCell + TDB_PAGE_OFFSET_SIZE(pPage)) { pPage->pFreeStart += TDB_PAGE_OFFSET_SIZE(pPage); - pCell = pPage->pFreeEnd; + pPage->pFreeEnd -= szCell; + + TDB_PAGE_CCELLS_SET(pPage, pPage->pFreeEnd - pPage->pData); + TDB_PAGE_NFREE_SET(pPage, nFree - szCell - TDB_PAGE_OFFSET_SIZE(pPage)); + + *ppCell = pPage->pFreeEnd; + return 0; } // 2. Try to allocate from the page free list - if ((pCell == NULL) && (pPage->pFreeEnd - pPage->pFreeStart >= TDB_PAGE_OFFSET_SIZE(pPage)) && - TDB_PAGE_FCELL(pPage)) { -#if 0 - 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); + cellFree = TDB_PAGE_FCELL(pPage); + ASSERT(cellFree == 0 || cellFree > pPage->pFreeEnd - pPage->pData); + if (cellFree && pPage->pFreeEnd - pPage->pFreeStart >= TDB_PAGE_OFFSET_SIZE(pPage)) { + SCell *pPrevFreeCell = NULL; + int szPrevFreeCell; + int szFreeCell; + int nxFreeCell; + int newSize; for (;;) { - // Find a cell - if (szCell >= size) { - if (szCell - size >= pPage->szFreeCell) { - SCell *pTmpCell = pCell + size; + if (cellFree == 0) break; - TDB_PAGE_FREE_CELL_SIZE_SET(pPage, pTmpCell, szCell - size); - TDB_PAGE_FREE_CELL_NXOFFSET_SET(pPage, pTmpCell, nxOffset); - // TODO: *pOffset = pTmpCell - pPage->pData; + pFreeCell = pPage->pData + cellFree; + pPage->pPageMethods->getFreeCellInfo(pFreeCell, &szFreeCell, &nxFreeCell); + + if (szFreeCell >= szCell) { + pPage->pFreeStart += TDB_PAGE_OFFSET_SIZE(pPage); + TDB_PAGE_NFREE_SET(pPage, nFree - TDB_PAGE_OFFSET_SIZE(pPage) - szCell); + *ppCell = pFreeCell; + + newSize = szFreeCell - szCell; + pFreeCell += szCell; + if (newSize >= TDB_PAGE_FREE_CELL_SIZE(pPage)) { + pPage->pPageMethods->setFreeCellInfo(pFreeCell, newSize, nxFreeCell); + if (pPrevFreeCell) { + pPage->pPageMethods->setFreeCellInfo(pPrevFreeCell, szPrevFreeCell, pFreeCell - pPage->pData); + } else { + TDB_PAGE_FCELL_SET(pPage, pFreeCell - pPage->pData); + } + + return 0; } else { - TDB_PAGE_NFREE_SET(pPage, TDB_PAGE_NFREE(pPage) + szCell - size); - // TODO: *pOffset = nxOffset; + if (pPrevFreeCell) { + pPage->pPageMethods->setFreeCellInfo(pPrevFreeCell, szPrevFreeCell, nxFreeCell); + } else { + TDB_PAGE_FCELL_SET(pPage, nxFreeCell); + } } - 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; + pPrevFreeCell = pFreeCell; + szPrevFreeCell = szFreeCell; + cellFree = nxFreeCell; } } - - if (pCell) { - pPage->pFreeStart = pPage->pFreeStart + pPage->szOffset; - } -#endif } - // 3. Try to dfragment and allocate again +// 3. Try to dfragment and allocate again +#if 0 if (pCell == NULL) { ret = tdbPageDefragment(pPage); if (ret < 0) { @@ -175,7 +216,7 @@ static int tdbPageAllocate(SPage *pPage, int size, SCell **ppCell) { } 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 pPage->pFreeEnd -= size; @@ -185,8 +226,9 @@ static int tdbPageAllocate(SPage *pPage, int size, SCell **ppCell) { ASSERT(pCell != NULL); - pPage->nFree = pPage->nFree - size - TDB_PAGE_OFFSET_SIZE(pPage); + // pPage->nFree = pPage->nFree - size - TDB_PAGE_OFFSET_SIZE(pPage); *ppCell = pCell; +#endif return 0; } @@ -210,6 +252,11 @@ typedef struct __attribute__((__packed__)) { u16 nFree; } SPageHdr; +typedef struct __attribute__((__packed__)) { + u16 szCell; + u16 nxOffset; +} SFreeCell; + // 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; } @@ -253,20 +300,35 @@ static inline void setPageCellOffset(SPage *pPage, int idx, int offset) { ((u16 *)pPage->pCellIdx)[idx] = (u16)offset; } +// free cell info +static inline void getPageFreeCellInfo(SCell *pCell, int *szCell, int *nxOffset) { + SFreeCell *pFreeCell = (SFreeCell *)pCell; + *szCell = pFreeCell->szCell; + *nxOffset = pFreeCell->nxOffset; +} + +static inline void setPageFreeCellInfo(SCell *pCell, int szCell, int nxOffset) { + SFreeCell *pFreeCell = (SFreeCell *)pCell; + pFreeCell->szCell = szCell; + pFreeCell->nxOffset = nxOffset; +} + SPageMethods pageMethods = { - 2, // szOffset - sizeof(SPageHdr), // szPageHdr - sizeof(SFreeCell), // szFreeCell - getPageFlags, // getPageFlags - setPageFlags, // setFlagsp - getPageCellNum, // getCellNum - setPageCellNum, // setCellNum - getPageCellBody, // getCellBody - setPageCellBody, // setCellBody - getPageCellFree, // getCellFree - setPageCellFree, // setCellFree - getPageNFree, // getFreeBytes - setPageNFree, // setFreeBytes - getPageCellOffset, // getCellOffset - setPageCellOffset // setCellOffset + 2, // szOffset + sizeof(SPageHdr), // szPageHdr + sizeof(SFreeCell), // szFreeCell + getPageFlags, // getPageFlags + setPageFlags, // setFlagsp + getPageCellNum, // getCellNum + setPageCellNum, // setCellNum + getPageCellBody, // getCellBody + setPageCellBody, // setCellBody + getPageCellFree, // getCellFree + setPageCellFree, // setCellFree + getPageNFree, // getFreeBytes + setPageNFree, // setFreeBytes + getPageCellOffset, // getCellOffset + setPageCellOffset, // setCellOffset + getPageFreeCellInfo, // getFreeCellInfo + setPageFreeCellInfo // setFreeCellInfo }; \ No newline at end of file diff --git a/source/libs/tdb/src/page/tdbPageL.c b/source/libs/tdb/src/page/tdbPageL.c index e7c60118d2..e3c1bb8d37 100644 --- a/source/libs/tdb/src/page/tdbPageL.c +++ b/source/libs/tdb/src/page/tdbPageL.c @@ -66,20 +66,35 @@ static inline void setPageCellOffset(SPage *pPage, int idx, int offset) { TDB_PUT_U24(pPage->pCellIdx + 3 * idx, offset); } +// free cell info +static inline void getPageFreeCellInfo(SCell *pCell, int *szCell, int *nxOffset) { + SFreeCellL *pFreeCell = (SFreeCellL *)pCell; + *szCell = TDB_GET_U24(pFreeCell->szCell); + *nxOffset = TDB_GET_U24(pFreeCell->nxOffset); +} + +static inline void setPageFreeCellInfo(SCell *pCell, int szCell, int nxOffset) { + SFreeCellL *pFreeCell = (SFreeCellL *)pCell; + TDB_PUT_U24(pFreeCell->szCell, szCell); + TDB_PUT_U24(pFreeCell->nxOffset, nxOffset); +} + SPageMethods pageLargeMethods = { - 3, // szOffset - sizeof(SPageHdrL), // szPageHdr - sizeof(SFreeCellL), // szFreeCell - getPageFlags, // getPageFlags - setPageFlags, // setFlagsp - getPageCellNum, // getCellNum - setPageCellNum, // setCellNum - getPageCellBody, // getCellBody - setPageCellBody, // setCellBody - getPageCellFree, // getCellFree - setPageCellFree, // setCellFree - getPageNFree, // getFreeBytes - setPageNFree, // setFreeBytes - getPageCellOffset, // getCellOffset - setPageCellOffset // setCellOffset + 3, // szOffset + sizeof(SPageHdrL), // szPageHdr + sizeof(SFreeCellL), // szFreeCell + getPageFlags, // getPageFlags + setPageFlags, // setFlagsp + getPageCellNum, // getCellNum + setPageCellNum, // setCellNum + getPageCellBody, // getCellBody + setPageCellBody, // setCellBody + getPageCellFree, // getCellFree + setPageCellFree, // setCellFree + getPageNFree, // getFreeBytes + setPageNFree, // setFreeBytes + getPageCellOffset, // getCellOffset + setPageCellOffset, // setCellOffset + getPageFreeCellInfo, // getFreeCellInfo + setPageFreeCellInfo // setFreeCellInfo }; \ No newline at end of file