Skip to content

T
TDengine

taosdata / TDengine
1 年多 前同步成功

通知 1185
Star 22017
Fork 4786
T
TDengine
未验证 提交 d4972b0d 编写于 作者: H Hongze Cheng 提交者： GitHub
浏览文件
操作

Merge pull request #11048 from taosdata/feature/tdb 

Feature/tdb
上级 4bd42a2d 249c3376
隐藏空白更改
内联 并排
Showing with 620 addition and 331 deletion
source/libs/CMakeLists.txt
浏览文件 @ d4972b0d
add_subdirectory(transport)
add_subdirectory(sync)
# add_subdirectory(tdb)
add_subdirectory(tdb)
add_subdirectory(index)
add_subdirectory(wal)
add_subdirectory(parser)
......
source/libs/tdb/CMakeLists.txt
浏览文件 @ d4972b0d
......@@ -8,6 +8,8 @@ target_sources(tdb
"src/db/tdbBtree.c"
"src/db/tdbDb.c"
"src/db/tdbEnv.c"
"src/db/tdbTxn.c"
"src/db/tdbOs.c"
"src/page/tdbPage.c"
"src/page/tdbPageL.c"
)
......
source/libs/tdb/inc/tdb.h
浏览文件 @ d4972b0d
......@@ -22,43 +22,6 @@
extern "C" {
#endif
// typedef struct STDb TDB;
// typedef struct STDbEnv TENV;
// typedef struct STDbCurosr TDBC;
// typedef int32_t pgsz_t;
// typedef int32_t cachesz_t;
// typedef int (*TdbKeyCmprFn)(int keyLen1, const void *pKey1, int keyLen2, const void *pKey2);
// // TEVN
// int tdbEnvCreate(TENV **ppEnv, const char *rootDir);
// int tdbEnvOpen(TENV *ppEnv);
// int tdbEnvClose(TENV *pEnv);
// int tdbEnvSetCache(TENV *pEnv, pgsz_t pgSize, cachesz_t cacheSize);
// pgsz_t tdbEnvGetPageSize(TENV *pEnv);
// cachesz_t tdbEnvGetCacheSize(TENV *pEnv);
// int tdbEnvBeginTxn(TENV *pEnv);
// int tdbEnvCommit(TENV *pEnv);
// // TDB
// int tdbCreate(TDB **ppDb);
// int tdbOpen(TDB *pDb, const char *fname, const char *dbname, TENV *pEnv);
// int tdbClose(TDB *pDb);
// int tdbDrop(TDB *pDb);
// int tdbSetKeyLen(TDB *pDb, int klen);
// int tdbSetValLen(TDB *pDb, int vlen);
// int tdbSetDup(TDB *pDb, int dup);
// int tdbSetCmprFunc(TDB *pDb, TdbKeyCmprFn fn);
// int tdbGetKeyLen(TDB *pDb);
// int tdbGetValLen(TDB *pDb);
// int tdbGetDup(TDB *pDb);
// int tdbInsert(TDB *pDb, const void *pKey, int nKey, const void *pData, int nData);
#ifdef __cplusplus
}
#endif
......
source/libs/tdb/src/db/tdbBtree.c
浏览文件 @ d4972b0d
......@@ -67,7 +67,7 @@ typedef struct {
u8 *pTmpSpace;
} SCellDecoder;
static int tdbBtCursorMoveTo(SBTC *pCur, const void *pKey, int kLen, int *pCRst);
static int tdbBtCursorMoveTo(SBTC *pBtc, const void *pKey, int kLen, int *pCRst);
static int tdbDefaultKeyCmprFn(const void *pKey1, int keyLen1, const void *pKey2, int keyLen2);
static int tdbBtreeOpenImpl(SBTree *pBt);
static int tdbBtreeZeroPage(SPage *pPage, void *arg);
......@@ -75,10 +75,10 @@ static int tdbBtreeInitPage(SPage *pPage, void *arg);
static int tdbBtreeEncodeCell(SPage *pPage, const void *pKey, int kLen, const void *pVal, int vLen, SCell *pCell,
int *szCell);
static int tdbBtreeDecodeCell(SPage *pPage, const SCell *pCell, SCellDecoder *pDecoder);
static int tdbBtreeBalance(SBTC *pCur);
static int tdbBtreeBalance(SBTC *pBtc);
static int tdbBtreeCellSize(const SPage *pPage, SCell *pCell);
static int tdbBtcMoveToNext(SBTC *pBtc);
static int tdbBtcMoveDownward(SBTC *pCur, SPgno pgno);
static int tdbBtcMoveDownward(SBTC *pBtc, SPgno pgno);
static int tdbBtcMoveUpward(SBTC *pBtc);
int tdbBtreeOpen(int keyLen, int valLen, SPager *pPager, FKeyComparator kcmpr, SBTree **ppBt) {
......@@ -87,7 +87,7 @@ int tdbBtreeOpen(int keyLen, int valLen, SPager *pPager, FKeyComparator kcmpr, S
*ppBt = NULL;
pBt = (SBTree *)calloc(1, sizeof(*pBt));
pBt = (SBTree *)tdbOsCalloc(1, sizeof(*pBt));
if (pBt == NULL) {
return -1;
}
......@@ -121,7 +121,7 @@ int tdbBtreeOpen(int keyLen, int valLen, SPager *pPager, FKeyComparator kcmpr, S
// TODO: pBt->root
ret = tdbBtreeOpenImpl(pBt);
if (ret < 0) {
free(pBt);
tdbOsFree(pBt);
return -1;
}
......@@ -134,7 +134,7 @@ int tdbBtreeClose(SBTree *pBt) {
return 0;
}
int tdbBtCursorInsert(SBTC *pCur, const void *pKey, int kLen, const void *pVal, int vLen) {
int tdbBtCursorInsert(SBTC *pBtc, const void *pKey, int kLen, const void *pVal, int vLen) {
int ret;
int idx;
SPager *pPager;
......@@ -143,20 +143,20 @@ int tdbBtCursorInsert(SBTC *pCur, const void *pKey, int kLen, const void *pVal,
int cret;
SBTree *pBt;
ret = tdbBtCursorMoveTo(pCur, pKey, kLen, &cret);
ret = tdbBtCursorMoveTo(pBtc, pKey, kLen, &cret);
if (ret < 0) {
// TODO: handle error
return -1;
}
if (pCur->idx == -1) {
ASSERT(TDB_PAGE_TOTAL_CELLS(pCur->pPage) == 0);
if (pBtc->idx == -1) {
ASSERT(TDB_PAGE_TOTAL_CELLS(pBtc->pPage) == 0);
idx = 0;
} else {
if (cret > 0) {
idx = pCur->idx + 1;
idx = pBtc->idx + 1;
} else if (cret < 0) {
idx = pCur->idx;
idx = pBtc->idx;
} else {
/* TODO */
ASSERT(0);
......@@ -164,9 +164,9 @@ int tdbBtCursorInsert(SBTC *pCur, const void *pKey, int kLen, const void *pVal,
}
// TODO: refact code here
pBt = pCur->pBt;
pBt = pBtc->pBt;
if (!pBt->pTmp) {
pBt->pTmp = (u8 *)malloc(pBt->pageSize);
pBt->pTmp = (u8 *)tdbOsMalloc(pBt->pageSize);
if (pBt->pTmp == NULL) {
return -1;
}
......@@ -175,20 +175,20 @@ int tdbBtCursorInsert(SBTC *pCur, const void *pKey, int kLen, const void *pVal,
pCell = pBt->pTmp;
// Encode the cell
ret = tdbBtreeEncodeCell(pCur->pPage, pKey, kLen, pVal, vLen, pCell, &szCell);
ret = tdbBtreeEncodeCell(pBtc->pPage, pKey, kLen, pVal, vLen, pCell, &szCell);
if (ret < 0) {
return -1;
}
// Insert the cell to the index
ret = tdbPageInsertCell(pCur->pPage, idx, pCell, szCell, 0);
ret = tdbPageInsertCell(pBtc->pPage, idx, pCell, szCell, 0);
if (ret < 0) {
return -1;
}
// If page is overflow, balance the tree
if (pCur->pPage->nOverflow > 0) {
ret = tdbBtreeBalance(pCur);
if (pBtc->pPage->nOverflow > 0) {
ret = tdbBtreeBalance(pBtc);
if (ret < 0) {
return -1;
}
......@@ -226,30 +226,30 @@ int tdbBtreeGet(SBTree *pBt, const void *pKey, int kLen, void **ppVal, int *vLen
return 0;
}
static int tdbBtCursorMoveTo(SBTC *pCur, const void *pKey, int kLen, int *pCRst) {
static int tdbBtCursorMoveTo(SBTC *pBtc, const void *pKey, int kLen, int *pCRst) {
int ret;
SBTree *pBt;
SPager *pPager;
pBt = pCur->pBt;
pBt = pBtc->pBt;
pPager = pBt->pPager;
if (pCur->iPage < 0) {
ASSERT(pCur->iPage == -1);
ASSERT(pCur->idx == -1);
if (pBtc->iPage < 0) {
ASSERT(pBtc->iPage == -1);
ASSERT(pBtc->idx == -1);
// Move from the root
ret = tdbPagerFetchPage(pPager, pBt->root, &(pCur->pPage), tdbBtreeInitPage, pBt);
ret = tdbPagerFetchPage(pPager, pBt->root, &(pBtc->pPage), tdbBtreeInitPage, pBt);
if (ret < 0) {
ASSERT(0);
return -1;
}
pCur->iPage = 0;
pBtc->iPage = 0;
if (TDB_PAGE_TOTAL_CELLS(pCur->pPage) == 0) {
if (TDB_PAGE_TOTAL_CELLS(pBtc->pPage) == 0) {
// Current page is empty
// ASSERT(TDB_FLAG_IS(TDB_PAGE_FLAGS(pCur->pPage), TDB_BTREE_ROOT | TDB_BTREE_LEAF));
// ASSERT(TDB_FLAG_IS(TDB_PAGE_FLAGS(pBtc->pPage), TDB_BTREE_ROOT | TDB_BTREE_LEAF));
return 0;
}
......@@ -259,7 +259,7 @@ static int tdbBtCursorMoveTo(SBTC *pCur, const void *pKey, int kLen, int *pCRst)
SPage *pPage;
SCellDecoder cd = {0};
pPage = pCur->pPage;
pPage = pBtc->pPage;
nCells = TDB_PAGE_TOTAL_CELLS(pPage);
lidx = 0;
ridx = nCells - 1;
......@@ -297,22 +297,22 @@ static int tdbBtCursorMoveTo(SBTC *pCur, const void *pKey, int kLen, int *pCRst)
u8 flags = TDB_BTREE_PAGE_GET_FLAGS(pPage);
u8 leaf = TDB_BTREE_PAGE_IS_LEAF(flags);
if (leaf) {
pCur->idx = midx;
pBtc->idx = midx;
*pCRst = c;
break;
} else {
if (c <= 0) {
pCur->idx = midx;
tdbBtcMoveDownward(pCur, cd.pgno);
pBtc->idx = midx;
tdbBtcMoveDownward(pBtc, cd.pgno);
} else {
pCur->idx = midx + 1;
pBtc->idx = midx + 1;
if (midx == nCells - 1) {
/* Move to right-most child */
tdbBtcMoveDownward(pCur, ((SIntHdr *)pCur->pPage->pData)->pgno);
tdbBtcMoveDownward(pBtc, ((SIntHdr *)pBtc->pPage->pData)->pgno);
} else {
pCell = tdbPageGetCell(pPage, pCur->idx);
pCell = tdbPageGetCell(pPage, pBtc->idx);
tdbBtreeDecodeCell(pPage, pCell, &cd);
tdbBtcMoveDownward(pCur, cd.pgno);
tdbBtcMoveDownward(pBtc, cd.pgno);
}
}
}
......@@ -550,7 +550,7 @@ static int tdbBtreeBalanceNonRoot(SBTree *pBt, SPage *pParent, int idx) {
if (sIdx + i < TDB_PAGE_TOTAL_CELLS(pParent)) {
pCell = tdbPageGetCell(pParent, sIdx + i);
szDivCell[i] = tdbBtreeCellSize(pParent, pCell);
pDivCell[i] = malloc(szDivCell[i]);
pDivCell[i] = tdbOsMalloc(szDivCell[i]);
memcpy(pDivCell[i], pCell, szDivCell[i]);
}
......@@ -740,13 +740,13 @@ static int tdbBtreeBalanceNonRoot(SBTree *pBt, SPage *pParent, int idx) {
tdbBtreeDecodeCell(pPage, pCell, &cd);
// TODO: pCell here may be inserted as an overflow cell, handle it
SCell *pNewCell = malloc(cd.kLen + 9);
SCell *pNewCell = tdbOsMalloc(cd.kLen + 9);
int szNewCell;
SPgno pgno;
pgno = TDB_PAGE_PGNO(pNews[iNew]);
tdbBtreeEncodeCell(pParent, cd.pKey, cd.kLen, (void *)&pgno, sizeof(SPgno), pNewCell, &szNewCell);
tdbPageInsertCell(pParent, sIdx++, pNewCell, szNewCell, 0);
free(pNewCell);
tdbOsFree(pNewCell);
}
// move to next new page
......@@ -798,14 +798,14 @@ static int tdbBtreeBalanceNonRoot(SBTree *pBt, SPage *pParent, int idx) {
for (int i = 0; i < 3; i++) {
if (pDivCell[i]) {
free(pDivCell[i]);
tdbOsFree(pDivCell[i]);
}
}
return 0;
}
static int tdbBtreeBalance(SBTC *pCur) {
static int tdbBtreeBalance(SBTC *pBtc) {
int iPage;
SPage *pParent;
SPage *pPage;
......@@ -816,8 +816,8 @@ static int tdbBtreeBalance(SBTC *pCur) {
// Main loop to balance the BTree
for (;;) {
iPage = pCur->iPage;
pPage = pCur->pPage;
iPage = pBtc->iPage;
pPage = pBtc->pPage;
flags = TDB_BTREE_PAGE_GET_FLAGS(pPage);
leaf = TDB_BTREE_PAGE_IS_LEAF(flags);
root = TDB_BTREE_PAGE_IS_ROOT(flags);
......@@ -833,27 +833,27 @@ static int tdbBtreeBalance(SBTC *pCur) {
// ignore the case of empty
if (pPage->nOverflow == 0) break;
ret = tdbBtreeBalanceDeeper(pCur->pBt, pPage, &(pCur->pgStack[1]));
ret = tdbBtreeBalanceDeeper(pBtc->pBt, pPage, &(pBtc->pgStack[1]));
if (ret < 0) {
return -1;
}
pCur->idx = 0;
pCur->idxStack[0] = 0;
pCur->pgStack[0] = pCur->pPage;
pCur->iPage = 1;
pCur->pPage = pCur->pgStack[1];
pBtc->idx = 0;
pBtc->idxStack[0] = 0;
pBtc->pgStack[0] = pBtc->pPage;
pBtc->iPage = 1;
pBtc->pPage = pBtc->pgStack[1];
} else {
// Generalized balance step
pParent = pCur->pgStack[iPage - 1];
pParent = pBtc->pgStack[iPage - 1];
ret = tdbBtreeBalanceNonRoot(pCur->pBt, pParent, pCur->idxStack[pCur->iPage - 1]);
ret = tdbBtreeBalanceNonRoot(pBtc->pBt, pParent, pBtc->idxStack[pBtc->iPage - 1]);
if (ret < 0) {
return -1;
}
pCur->iPage--;
pCur->pPage = pCur->pgStack[pCur->iPage];
pBtc->iPage--;
pBtc->pPage = pBtc->pgStack[pBtc->iPage];
}
}
......@@ -1050,11 +1050,11 @@ static int tdbBtreeCellSize(const SPage *pPage, SCell *pCell) {
#endif
int tdbBtcOpen(SBTC *pCur, SBTree *pBt) {
pCur->pBt = pBt;
pCur->iPage = -1;
pCur->pPage = NULL;
pCur->idx = -1;
int tdbBtcOpen(SBTC *pBtc, SBTree *pBt) {
pBtc->pBt = pBt;
pBtc->iPage = -1;
pBtc->pPage = NULL;
pBtc->idx = -1;
return 0;
}
......@@ -1262,16 +1262,16 @@ int tdbBtcClose(SBTC *pBtc) {
return 0;
}
static int tdbBtcMoveDownward(SBTC *pCur, SPgno pgno) {
static int tdbBtcMoveDownward(SBTC *pBtc, SPgno pgno) {
int ret;
pCur->pgStack[pCur->iPage] = pCur->pPage;
pCur->idxStack[pCur->iPage] = pCur->idx;
pCur->iPage++;
pCur->pPage = NULL;
pCur->idx = -1;
pBtc->pgStack[pBtc->iPage] = pBtc->pPage;
pBtc->idxStack[pBtc->iPage] = pBtc->idx;
pBtc->iPage++;
pBtc->pPage = NULL;
pBtc->idx = -1;
ret = tdbPagerFetchPage(pCur->pBt->pPager, pgno, &pCur->pPage, tdbBtreeInitPage, pCur->pBt);
ret = tdbPagerFetchPage(pBtc->pBt->pPager, pgno, &pBtc->pPage, tdbBtreeInitPage, pBtc->pBt);
if (ret < 0) {
ASSERT(0);
}
......
source/libs/tdb/src/db/tdbDb.c
浏览文件 @ d4972b0d
......@@ -16,7 +16,7 @@
#include "tdbInt.h"
struct STDB {
STEnv *pEnv;
TENV *pEnv;
SBTree *pBt;
};
......@@ -24,8 +24,8 @@ struct STDBC {
SBTC btc;
};
int tdbDbOpen(const char *fname, int keyLen, int valLen, FKeyComparator keyCmprFn, STEnv *pEnv, STDB **ppDb) {
STDB *pDb;
int tdbDbOpen(const char *fname, int keyLen, int valLen, FKeyComparator keyCmprFn, TENV *pEnv, TDB **ppDb) {
TDB *pDb;
SPager *pPager;
int ret;
char fFullName[TDB_FILENAME_LEN];
......@@ -34,7 +34,7 @@ int tdbDbOpen(const char *fname, int keyLen, int valLen, FKeyComparator keyCmprF
*ppDb = NULL;
pDb = (STDB *)calloc(1, sizeof(*pDb));
pDb = (TDB *)tdbOsCalloc(1, sizeof(*pDb));
if (pDb == NULL) {
return -1;
}
......@@ -63,17 +63,17 @@ int tdbDbOpen(const char *fname, int keyLen, int valLen, FKeyComparator keyCmprF
return 0;
}
int tdbDbClose(STDB *pDb) {
int tdbDbClose(TDB *pDb) {
// TODO
return 0;
}
int tdbDbDrop(STDB *pDb) {
int tdbDbDrop(TDB *pDb) {
// TODO
return 0;
}
int tdbDbInsert(STDB *pDb, const void *pKey, int keyLen, const void *pVal, int valLen) {
int tdbDbInsert(TDB *pDb, const void *pKey, int keyLen, const void *pVal, int valLen) {
SBTC btc;
SBTC *pCur;
int ret;
......@@ -92,16 +92,16 @@ int tdbDbInsert(STDB *pDb, const void *pKey, int keyLen, const void *pVal, int v
return 0;
}
int tdbDbGet(STDB *pDb, const void *pKey, int kLen, void **ppVal, int *vLen) {
int tdbDbGet(TDB *pDb, const void *pKey, int kLen, void **ppVal, int *vLen) {
return tdbBtreeGet(pDb->pBt, pKey, kLen, ppVal, vLen);
}
int tdbDbcOpen(STDB *pDb, STDBC **ppDbc) {
int ret;
STDBC *pDbc = NULL;
int tdbDbcOpen(TDB *pDb, TDBC **ppDbc) {
int ret;
TDBC *pDbc = NULL;
*ppDbc = NULL;
pDbc = malloc(sizeof(*pDbc));
pDbc = (TDBC *)tdbOsMalloc(sizeof(*pDbc));
if (pDbc == NULL) {
return -1;
}
......@@ -120,13 +120,13 @@ int tdbDbcOpen(STDB *pDb, STDBC **ppDbc) {
return 0;
}
int tdbDbNext(STDBC *pDbc, void **ppKey, int *kLen, void **ppVal, int *vLen) {
int tdbDbNext(TDBC *pDbc, void **ppKey, int *kLen, void **ppVal, int *vLen) {
return tdbBtreeNext(&pDbc->btc, ppKey, kLen, ppVal, vLen);
}
int tdbDbcClose(STDBC *pDbc) {
int tdbDbcClose(TDBC *pDbc) {
if (pDbc) {
free(pDbc);
tdbOsFree(pDbc);
}
return 0;
......
source/libs/tdb/src/db/tdbEnv.c
浏览文件 @ d4972b0d
......@@ -15,24 +15,24 @@
#include "tdbInt.h"
int tdbEnvOpen(const char *rootDir, int pageSize, int cacheSize, STEnv **ppEnv) {
STEnv *pEnv;
int dsize;
int zsize;
u8 *pPtr;
int ret;
int tdbEnvOpen(const char *rootDir, int pageSize, int cacheSize, TENV **ppEnv) {
TENV *pEnv;
int dsize;
int zsize;
u8 *pPtr;
int ret;
*ppEnv = NULL;
dsize = strlen(rootDir);
zsize = sizeof(*pEnv) + dsize * 2 + strlen(TDB_JOURNAL_NAME) + 3;
pPtr = (uint8_t *)calloc(1, zsize);
pPtr = (uint8_t *)tdbOsCalloc(1, zsize);
if (pPtr == NULL) {
return -1;
}
pEnv = (STEnv *)pPtr;
pEnv = (TENV *)pPtr;
pPtr += sizeof(*pEnv);
// pEnv->rootDir
pEnv->rootDir = pPtr;
......@@ -59,12 +59,12 @@ int tdbEnvOpen(const char *rootDir, int pageSize, int cacheSize, STEnv **ppEnv)
return 0;
}
int tdbEnvClose(STEnv *pEnv) {
int tdbEnvClose(TENV *pEnv) {
// TODO
return 0;
}
SPager *tdbEnvGetPager(STEnv *pEnv, const char *fname) {
SPager *tdbEnvGetPager(TENV *pEnv, const char *fname) {
// TODO
return NULL;
}
\ No newline at end of file
source/libs/tdb/src/db/tdbOs.c 0 → 100644
浏览文件 @ d4972b0d
/*
* 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"
#ifndef TDB_FOR_TDENGINE
// tdbOsRead
i64 tdbOsRead(tdb_fd_t fd, void *pData, i64 nBytes) {
i64 nRead = 0;
i64 iRead = 0;
u8 *pBuf = (u8 *)pData;
while (nBytes > 0) {
iRead = read(fd, pBuf, nBytes);
if (iRead < 0) {
if (errno == EINTR) {
continue;
} else {
return -1;
}
} else if (iRead == 0) {
break;
}
nRead += iRead;
pBuf += iRead;
nBytes -= iRead;
}
return nRead;
}
// tdbOsPRead
i64 tdbOsPRead(tdb_fd_t fd, void *pData, i64 nBytes, i64 offset) {
i64 nRead = 0;
i64 iRead = 0;
i64 iOffset = offset;
u8 *pBuf = (u8 *)pData;
while (nBytes > 0) {
iRead = pread(fd, pBuf, nBytes, iOffset);
if (iRead < 0) {
if (errno == EINTR) {
continue;
} else {
return -1;
}
} else if (iRead == 0) {
break;
}
nRead += iRead;
pBuf += iRead;
iOffset += iRead;
nBytes -= iRead;
}
return nRead;
}
// tdbOsWrite
i64 tdbOsWrite(tdb_fd_t fd, const void *pData, i64 nBytes) {
i64 nWrite = 0;
i64 iWrite = 0;
u8 *pBuf = (u8 *)pData;
while (nBytes > 0) {
iWrite = write(fd, pBuf, nBytes);
if (iWrite < 0) {
if (errno == EINTR) {
continue;
}
return -1;
}
nWrite += iWrite;
pBuf += iWrite;
nBytes -= iWrite;
}
return nWrite;
}
#endif
\ No newline at end of file
source/libs/tdb/src/db/tdbPCache.c
浏览文件 @ d4972b0d
......@@ -15,16 +15,16 @@
#include "tdbInt.h"
struct SPCache {
int pageSize;
int cacheSize;
pthread_mutex_t mutex;
int nFree;
SPage *pFree;
int nPage;
int nHash;
SPage **pgHash;
int nRecyclable;
SPage lru;
int pageSize;
int cacheSize;
tdb_mutex_t mutex;
int nFree;
SPage *pFree;
int nPage;
int nHash;
SPage **pgHash;
int nRecyclable;
SPage lru;
};
#define PCACHE_PAGE_HASH(pPgid) \
......@@ -63,7 +63,7 @@ int tdbPCacheOpen(int pageSize, int cacheSize, SPCache **ppCache) {
void *pPtr;
SPage *pPgHdr;
pCache = (SPCache *)calloc(1, sizeof(*pCache));
pCache = (SPCache *)tdbOsCalloc(1, sizeof(*pCache));
if (pCache == NULL) {
return -1;
}
......@@ -72,7 +72,7 @@ int tdbPCacheOpen(int pageSize, int cacheSize, SPCache **ppCache) {
pCache->cacheSize = cacheSize;
if (tdbPCacheOpenImpl(pCache) < 0) {
free(pCache);
tdbOsFree(pCache);
return -1;
}
......@@ -116,13 +116,13 @@ void tdbPCacheRelease(SPCache *pCache, SPage *pPage) {
}
}
static void tdbPCacheInitLock(SPCache *pCache) { pthread_mutex_init(&(pCache->mutex), NULL); }
static void tdbPCacheInitLock(SPCache *pCache) { tdbMutexInit(&(pCache->mutex), NULL); }
static void tdbPCacheClearLock(SPCache *pCache) { pthread_mutex_destroy(&(pCache->mutex)); }
static void tdbPCacheClearLock(SPCache *pCache) { tdbMutexDestroy(&(pCache->mutex)); }
static void tdbPCacheLock(SPCache *pCache) { pthread_mutex_lock(&(pCache->mutex)); }
static void tdbPCacheLock(SPCache *pCache) { tdbMutexLock(&(pCache->mutex)); }
static void tdbPCacheUnlock(SPCache *pCache) { pthread_mutex_unlock(&(pCache->mutex)); }
static void tdbPCacheUnlock(SPCache *pCache) { tdbMutexDestroy(&(pCache->mutex)); }
static bool tdbPCacheLocked(SPCache *pCache) {
assert(0);
......@@ -268,7 +268,7 @@ static int tdbPCacheOpenImpl(SPCache *pCache) {
// Open the hash table
pCache->nPage = 0;
pCache->nHash = pCache->cacheSize;
pCache->pgHash = (SPage **)calloc(pCache->nHash, sizeof(SPage *));
pCache->pgHash = (SPage **)tdbOsCalloc(pCache->nHash, sizeof(SPage *));
if (pCache->pgHash == NULL) {
// TODO
return -1;
......
source/libs/tdb/src/db/tdbPager.c
浏览文件 @ d4972b0d
......@@ -20,14 +20,12 @@ struct SPager {
char *jFileName;
int pageSize;
uint8_t fid[TDB_FILE_ID_LEN];
int fd;
int jfd;
tdb_fd_t fd;
tdb_fd_t jfd;
SPCache *pCache;
SPgno dbFileSize;
SPgno dbOrigSize;
int nDirty;
SPage *pDirty;
SPage *pDirtyTail;
u8 inTran;
};
......@@ -46,6 +44,8 @@ TDB_STATIC_ASSERT(sizeof(SFileHdr) == 128, "Size of file header is not correct")
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);
static int tdbPagerWritePageToJournal(SPager *pPager, SPage *pPage);
static int tdbPagerWritePageToDB(SPager *pPager, SPage *pPage);
int tdbPagerOpen(SPCache *pCache, const char *fileName, SPager **ppPager) {
uint8_t *pPtr;
......@@ -60,7 +60,7 @@ int tdbPagerOpen(SPCache *pCache, const char *fileName, SPager **ppPager) {
zsize = sizeof(*pPager) /* SPager */
+ fsize + 1 /* dbFileName */
+ fsize + 8 + 1; /* jFileName */
pPtr = (uint8_t *)calloc(1, zsize);
pPtr = (uint8_t *)tdbOsCalloc(1, zsize);
if (pPtr == NULL) {
return -1;
}
......@@ -80,7 +80,7 @@ int tdbPagerOpen(SPCache *pCache, const char *fileName, SPager **ppPager) {
// pPager->pCache
pPager->pCache = pCache;
pPager->fd = open(pPager->dbFileName, O_RDWR | O_CREAT, 0755);
pPager->fd = tdbOsOpen(pPager->dbFileName, TDB_O_CREAT | TDB_O_RDWR, 0755);
if (pPager->fd < 0) {
return -1;
}
......@@ -90,7 +90,7 @@ int tdbPagerOpen(SPCache *pCache, const char *fileName, SPager **ppPager) {
return -1;
}
pPager->jfd = -1;
// pPager->jfd = -1;
pPager->pageSize = tdbPCacheGetPageSize(pCache);
*ppPager = pPager;
......@@ -140,14 +140,25 @@ int tdbPagerWrite(SPager *pPager, SPage *pPage) {
}
}
if (pPage->isDirty == 0) {
pPage->isDirty = 1;
// TODO: add the page to the dirty list
if (pPage->isDirty) return 0;
// TODO: write the page to the journal
if (1 /*actually load from the file*/) {
// Set page as dirty
pPage->isDirty = 1;
// Add page to dirty list
// TODO: sort the list according to the page number
pPage->pDirtyNext = pPager->pDirty;
pPager->pDirty = pPage;
// Write page to journal
if (TDB_PAGE_PGNO(pPage) <= pPager->dbOrigSize) {
ret = tdbPagerWritePageToJournal(pPager, pPage);
if (ret < 0) {
ASSERT(0);
return -1;
}
}
return 0;
}
......@@ -157,7 +168,7 @@ int tdbPagerBegin(SPager *pPager) {
}
// Open the journal
pPager->jfd = open(pPager->jFileName, O_RDWR | O_CREAT, 0755);
pPager->jfd = tdbOsOpen(pPager->jFileName, TDB_O_CREAT | TDB_O_RDWR, 0755);
if (pPager->jfd < 0) {
return -1;
}
......@@ -170,7 +181,37 @@ int tdbPagerBegin(SPager *pPager) {
}
int tdbPagerCommit(SPager *pPager) {
// TODO
SPage *pPage;
int ret;
// Begin commit
{
// TODO: Sync the journal file (Here or when write ?)
}
for (;;) {
pPage = pPager->pDirty;
if (pPage == NULL) break;
ret = tdbPagerWritePageToDB(pPager, pPage);
if (ret < 0) {
ASSERT(0);
return -1;
}
pPager->pDirty = pPage->pDirtyNext;
pPage->pDirtyNext = NULL;
// TODO: release the page
}
tdbOsFSync(pPager->fd);
tdbOsClose(pPager->jfd);
tdbOsRemove(pPager->jFileName);
// pPager->jfd = -1;
return 0;
}
......@@ -181,7 +222,7 @@ static int tdbPagerReadPage(SPager *pPager, SPage *pPage) {
ASSERT(memcmp(pPager->fid, pPage->pgid.fileid, TDB_FILE_ID_LEN) == 0);
offset = (pPage->pgid.pgno - 1) * (i64)(pPager->pageSize);
ret = tdbPRead(pPager->fd, pPage->pData, pPager->pageSize, offset);
ret = tdbOsPRead(pPager->fd, pPage->pData, pPager->pageSize, offset);
if (ret < 0) {
// TODO: handle error
return -1;
......@@ -255,9 +296,7 @@ int tdbPagerNewPage(SPager *pPager, SPgno *ppgno, SPage **ppPage, int (*initPage
return 0;
}
void tdbPagerReturnPage(SPager *pPager, SPage *pPage) {
tdbPCacheRelease(pPager->pCache, pPage);
}
void tdbPagerReturnPage(SPager *pPager, SPage *pPage) { tdbPCacheRelease(pPager->pCache, pPage); }
static int tdbPagerAllocFreePage(SPager *pPager, SPgno *ppgno) {
// TODO: Allocate a page from the free list
......@@ -328,5 +367,44 @@ static int tdbPagerInitPage(SPager *pPager, SPage *pPage, int (*initPage)(SPage
return -1;
}
return 0;
}
// ---------------------------- Journal manipulation
static int tdbPagerWritePageToJournal(SPager *pPager, SPage *pPage) {
int ret;
SPgno pgno;
pgno = TDB_PAGE_PGNO(pPage);
ret = tdbOsWrite(pPager->jfd, &pgno, sizeof(pgno));
if (ret < 0) {
return -1;
}
ret = tdbOsWrite(pPager->jfd, pPage->pData, pPage->pageSize);
if (ret < 0) {
return -1;
}
return 0;
}
static int tdbPagerWritePageToDB(SPager *pPager, SPage *pPage) {
i64 offset;
int ret;
offset = pPage->pageSize * TDB_PAGE_PGNO(pPage);
if (tdbOsLSeek(pPager->fd, offset, SEEK_SET) < 0) {
ASSERT(0);
return -1;
}
ret = tdbOsWrite(pPager->fd, pPage->pData, pPage->pageSize);
if (ret < 0) {
ASSERT(0);
return -1;
}
return 0;
}
\ No newline at end of file
source/libs/tdb/src/db/tdbTxn.c 0 → 100644
浏览文件 @ d4972b0d
/*
* 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"
int tdbTxnBegin(TENV *pEnv) {
// TODO
return 0;
}
int tdbTxnCommit(TENV *pEnv) {
// TODO
return 0;
}
int tdbTxnRollback(TENV *pEnv) {
// TODO
return 0;
}
\ No newline at end of file
source/libs/tdb/src/db/tdbUtil.c
浏览文件 @ d4972b0d
......@@ -31,62 +31,4 @@ int tdbGnrtFileID(const char *fname, uint8_t *fileid, bool unique) {
}
return 0;
}
// int tdbCheckFileAccess(const char *pathname, int mode) {
// int flags = 0;
// if (mode & TDB_F_OK) {
// flags |= F_OK;
// }
// if (mode & TDB_R_OK) {
// flags |= R_OK;
// }
// if (mode & TDB_W_OK) {
// flags |= W_OK;
// }
// return access(pathname, flags);
// }
int tdbGetFileSize(const char *fname, int pgSize, SPgno *pSize) {
struct stat st;
int ret;
int64_t file_size = 0;
ret = taosStatFile(fname, &file_size, NULL);
if (ret != 0) {
return -1;
}
ASSERT(file_size % pgSize == 0);
*pSize = file_size / pgSize;
return 0;
}
int tdbPRead(int fd, void *pData, int count, i64 offset) {
void *pBuf;
int nbytes;
i64 ioffset;
int iread;
pBuf = pData;
nbytes = count;
ioffset = offset;
while (nbytes > 0) {
iread = pread(fd, pBuf, nbytes, ioffset);
if (iread < 0) {
/* TODO */
} else if (iread == 0) {
return (count - iread);
}
nbytes = nbytes - iread;
pBuf = (void *)((u8 *)pBuf + iread);
ioffset += iread;
}
return count;
}
\ No newline at end of file
source/libs/tdb/src/inc/tdbDb.h
浏览文件 @ d4972b0d
......@@ -20,20 +20,20 @@
extern "C" {
#endif
typedef struct STDB STDB;
typedef struct STDBC STDBC;
typedef struct STDB TDB;
typedef struct STDBC TDBC;
// STDB
int tdbDbOpen(const char *fname, int keyLen, int valLen, FKeyComparator keyCmprFn, STEnv *pEnv, STDB **ppDb);
int tdbDbClose(STDB *pDb);
int tdbDbDrop(STDB *pDb);
int tdbDbInsert(STDB *pDb, const void *pKey, int keyLen, const void *pVal, int valLen);
int tdbDbGet(STDB *pDb, const void *pKey, int kLen, void **ppVal, int *vLen);
// TDB
int tdbDbOpen(const char *fname, int keyLen, int valLen, FKeyComparator keyCmprFn, TENV *pEnv, TDB **ppDb);
int tdbDbClose(TDB *pDb);
int tdbDbDrop(TDB *pDb);
int tdbDbInsert(TDB *pDb, const void *pKey, int keyLen, const void *pVal, int valLen);
int tdbDbGet(TDB *pDb, const void *pKey, int kLen, void **ppVal, int *vLen);
// STDBC
int tdbDbcOpen(STDB *pDb, STDBC **ppDbc);
int tdbDbNext(STDBC *pDbc, void **ppKey, int *kLen, void **ppVal, int *vLen);
int tdbDbcClose(STDBC *pDbc);
// TDBC
int tdbDbcOpen(TDB *pDb, TDBC **ppDbc);
int tdbDbNext(TDBC *pDbc, void **ppKey, int *kLen, void **ppVal, int *vLen);
int tdbDbcClose(TDBC *pDbc);
#ifdef __cplusplus
}
......
source/libs/tdb/src/inc/tdbEnv.h
浏览文件 @ d4972b0d
......@@ -21,16 +21,16 @@ extern "C" {
#endif
typedef struct STEnv {
char * rootDir;
char * jfname;
char *rootDir;
char *jfname;
int jfd;
SPCache *pCache;
} STEnv;
} TENV;
int tdbEnvOpen(const char *rootDir, int pageSize, int cacheSize, STEnv **ppEnv);
int tdbEnvClose(STEnv *pEnv);
int tdbEnvOpen(const char *rootDir, int pageSize, int cacheSize, TENV **ppEnv);
int tdbEnvClose(TENV *pEnv);
SPager *tdbEnvGetPager(STEnv *pEnv, const char *fname);
SPager *tdbEnvGetPager(TENV *pEnv, const char *fname);
#ifdef __cplusplus
}
......
source/libs/tdb/src/inc/tdbInt.h
浏览文件 @ d4972b0d
......@@ -16,10 +16,9 @@
#ifndef _TD_TDB_INTERNAL_H_
#define _TD_TDB_INTERNAL_H_
#include "tlist.h"
#include "tlockfree.h"
#include "os.h"
// #include "tdb.h"
#include "tdb.h"
#ifdef __cplusplus
extern "C" {
......@@ -51,18 +50,18 @@ typedef u32 SPgno;
// fileid
#define TDB_FILE_ID_LEN 24
// pgid_t
// SPgid
typedef struct {
uint8_t fileid[TDB_FILE_ID_LEN];
SPgno pgno;
} pgid_t, SPgid;
} SPgid;
#define TDB_IVLD_PGID (pgid_t){0, TDB_IVLD_PGNO};
#define TDB_IVLD_PGID (SPgid){0, TDB_IVLD_PGNO};
static FORCE_INLINE int tdbCmprPgId(const void *p1, const void *p2) {
pgid_t *pgid1 = (pgid_t *)p1;
pgid_t *pgid2 = (pgid_t *)p2;
int rcode;
SPgid *pgid1 = (SPgid *)p1;
SPgid *pgid2 = (SPgid *)p2;
int rcode;
rcode = memcmp(pgid1->fileid, pgid2->fileid, TDB_FILE_ID_LEN);
if (rcode) {
......@@ -95,10 +94,6 @@ static FORCE_INLINE int tdbCmprPgId(const void *p1, const void *p2) {
// tdb_log
#define tdbError(var)
typedef TD_DLIST(STDB) STDbList;
typedef TD_DLIST(SPgFile) SPgFileList;
typedef TD_DLIST_NODE(SPgFile) SPgFileListNode;
#define TERR_A(val, op, flag) \
do { \
if (((val) = (op)) != 0) { \
......@@ -115,19 +110,6 @@ typedef TD_DLIST_NODE(SPgFile) SPgFileListNode;
#define TDB_VARIANT_LEN ((int)-1)
// page payload format
// <keyLen> + <valLen> + [key] + [value]
#define TDB_DECODE_PAYLOAD(pPayload, keyLen, pKey, valLen, pVal) \
do { \
if ((keyLen) == TDB_VARIANT_LEN) { \
/* TODO: decode the keyLen */ \
} \
if ((valLen) == TDB_VARIANT_LEN) { \
/* TODO: decode the valLen */ \
} \
/* TODO */ \
} while (0)
typedef int (*FKeyComparator)(const void *pKey1, int kLen1, const void *pKey2, int kLen2);
#define TDB_JOURNAL_NAME "tdb.journal"
......@@ -148,6 +130,8 @@ typedef struct SPager SPager;
typedef struct SPCache SPCache;
typedef struct SPage SPage;
#include "tdbOs.h"
#include "tdbUtil.h"
#include "tdbPCache.h"
......@@ -162,6 +146,8 @@ typedef struct SPage SPage;
#include "tdbPage.h"
#include "tdbTxn.h"
#ifdef __cplusplus
}
#endif
......
source/libs/tdb/src/inc/tdbOs.h 0 → 100644
浏览文件 @ d4972b0d
/*
* 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_OS_H_
#define _TDB_OS_H_
#ifdef __cplusplus
extern "C" {
#endif
// TODO: use cmake to control the option
#define TDB_FOR_TDENGINE
#ifdef TDB_FOR_TDENGINE
// For memory -----------------
#define tdbOsMalloc taosMemoryMalloc
#define tdbOsCalloc taosMemoryCalloc
#define tdbOsRealloc taosMemoryRealloc
#define tdbOsFree taosMemoryFree
// For file and directory -----------------
/* file */
typedef TdFilePtr tdb_fd_t;
#define TDB_O_CREAT TD_FILE_CTEATE
#define TDB_O_WRITE TD_FILE_WRITE
#define TDB_O_READ TD_FILE_READ
#define TDB_O_TRUNC TD_FILE_TRUNC
#define TDB_O_APPEND TD_FILE_APPEND
#define TDB_O_RDWR (TD_FILE_WRITE) | (TD_FILE_READ)
#define tdbOsOpen(PATH, OPTION, MODE) taosOpenFile((PATH), (OPTION))
#define tdbOsClose(FD) taosCloseFile(&(FD))
#define tdbOsRead taosReadFile
#define tdbOsPRead taosPReadFile
#define tdbOsWrite taosWriteFile
#define tdbOsFSync taosFsyncFile
#define tdbOsLSeek taosLSeekFile
#define tdbOsRemove remove
/* directory */
#define tdbOsMkdir taosMkDir
#define tdbOsRmdir taosRemoveDir
// For threads and lock -----------------
/* spin lock */
typedef TdThreadSpinlock tdb_spinlock_t;
#define tdbSpinlockInit taosThreadSpinInit
#define tdbSpinlockDestroy taosThreadSpinDestroy
#define tdbSpinlockLock taosThreadSpinLock
#define tdbSpinlockUnlock taosThreadSpinUnlock
#define tdbSpinlockTrylock pthread_spin_trylock
/* mutex lock */
typedef TdThreadMutex tdb_mutex_t;
#define tdbMutexInit taosThreadMutexInit
#define tdbMutexDestroy taosThreadMutexDestroy
#define tdbMutexLock taosThreadMutexLock
#define tdbMutexUnlock taosThreadMutexUnlock
#else
// For memory -----------------
#define tdbOsMalloc malloc
#define tdbOsCalloc calloc
#define tdbOsRealloc realloc
#define tdbOsFree free
// For file and directory -----------------
/* file */
typedef int tdb_fd_t;
#define TDB_O_CREAT O_CREAT
#define TDB_O_WRITE O_WRONLY
#define TDB_O_READ O_RDONLY
#define TDB_O_TRUNC O_TRUNC
#define TDB_O_APPEND O_APPEND
#define TDB_O_RDWR O_RDWR
#define tdbOsOpen(PATH, OPTION, MODE) open((PATH), (OPTION), (MODE))
#define tdbOsClose close
i64 tdbOsRead(tdb_fd_t fd, void *pData, i64 nBytes);
i64 tdbOsPRead(tdb_fd_t fd, void *pData, i64 nBytes, i64 offset);
i64 tdbOsWrite(tdb_fd_t fd, const void *pData, i64 nBytes);
#define tdbOsFSync fsync
#define tdbOsLSeek lseek
#define tdbOsRemove remove
/* directory */
#define tdbOsMkdir mkdir
#define tdbOsRmdir rmdir
// For threads and lock -----------------
/* spin lock */
typedef pthread_spinlock_t tdb_spinlock_t;
#define tdbSpinlockInit pthread_spin_init
#define tdbSpinlockDestroy pthread_spin_destroy
#define tdbSpinlockLock pthread_spin_lock
#define tdbSpinlockUnlock pthread_spin_unlock
#define tdbSpinlockTrylock pthread_spin_trylock
/* mutex lock */
typedef pthread_mutex_t tdb_mutex_t;
#define tdbMutexInit pthread_mutex_init
#define tdbMutexDestroy pthread_mutex_destroy
#define tdbMutexLock pthread_mutex_lock
#define tdbMutexUnlock pthread_mutex_unlock
#endif
#ifdef __cplusplus
}
#endif
#endif /*_TDB_OS_H_*/
\ No newline at end of file
source/libs/tdb/src/inc/tdbPage.h
浏览文件 @ d4972b0d
......@@ -53,10 +53,10 @@ typedef struct __attribute__((__packed__)) {
} SPageFtr;
struct SPage {
pthread_spinlock_t lock;
int pageSize;
u8 *pData;
SPageMethods *pPageMethods;
tdb_spinlock_t lock;
int pageSize;
u8 *pData;
SPageMethods *pPageMethods;
// Fields below used by pager and am
u8 *pPageHdr;
u8 *pCellIdx;
......@@ -80,21 +80,21 @@ struct SPage {
#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; \
#define TDB_INIT_PAGE_LOCK(pPage) tdbSpinlockInit(&((pPage)->lock), 0)
#define TDB_DESTROY_PAGE_LOCK(pPage) tdbSpinlockDestroy(&((pPage)->lock))
#define TDB_LOCK_PAGE(pPage) tdbSpinlockLock(&((pPage)->lock))
#define TDB_UNLOCK_PAGE(pPage) tdbSpinlockUnlock(&((pPage)->lock))
#define TDB_TRY_LOCK_PAGE(pPage) \
({ \
int ret; \
if (tdbSpinlockTrylock(&((pPage)->lock)) == 0) { \
ret = P_LOCK_SUCC; \
} else if (errno == EBUSY) { \
ret = P_LOCK_BUSY; \
} else { \
ret = P_LOCK_FAIL; \
} \
ret; \
})
// APIs
......
source/libs/tdb/src/inc/tdbTxn.h 0 → 100644
浏览文件 @ d4972b0d
/*
* 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_TXN_H_
#define _TDB_TXN_H_
#ifdef __cplusplus
extern "C" {
#endif
typedef struct STxn STXN;
struct STxn {
u64 txnId;
void *(*xMalloc)(void *, int);
void *xArg;
};
int tdbTxnBegin(TENV *pEnv);
int tdbTxnCommit(TENV *pEnv);
int tdbTxnRollback(TENV *pEnv);
#ifdef __cplusplus
}
#endif
#endif /*_TDB_TXN_H_*/
\ No newline at end of file
source/libs/tdb/src/inc/tdbUtil.h
浏览文件 @ d4972b0d
......@@ -30,46 +30,37 @@ extern "C" {
int tdbGnrtFileID(const char *fname, uint8_t *fileid, bool unique);
// #define TDB_F_OK 0x1
// #define TDB_R_OK 0x2
// #define TDB_W_OK 0x4
// int tdbCheckFileAccess(const char *pathname, int mode);
int tdbGetFileSize(const char *fname, int pgSize, SPgno *pSize);
int tdbPRead(int fd, void *pData, int count, i64 offset);
#define TDB_REALLOC(PTR, SIZE) \
({ \
void *nPtr; \
if ((PTR) == NULL || ((int *)(PTR))[-1] < (SIZE)) { \
nPtr = realloc((PTR) ? (char *)(PTR) - sizeof(int) : NULL, (SIZE) + sizeof(int)); \
if (nPtr) { \
((int *)nPtr)[0] = (SIZE); \
nPtr = (char *)nPtr + sizeof(int); \
} \
} else { \
nPtr = (PTR); \
} \
nPtr; \
#define TDB_REALLOC(PTR, SIZE) \
({ \
void *nPtr; \
if ((PTR) == NULL || ((int *)(PTR))[-1] < (SIZE)) { \
nPtr = tdbOsRealloc((PTR) ? (char *)(PTR) - sizeof(int) : NULL, (SIZE) + sizeof(int)); \
if (nPtr) { \
((int *)nPtr)[0] = (SIZE); \
nPtr = (char *)nPtr + sizeof(int); \
} \
} else { \
nPtr = (PTR); \
} \
nPtr; \
})
#define TDB_FREE(PTR) \
do { \
if (PTR) { \
free((char *)(PTR) - sizeof(int)); \
} \
#define TDB_FREE(PTR) \
do { \
if (PTR) { \
tdbOsFree((char *)(PTR) - sizeof(int)); \
} \
} while (0)
static inline void *tdbOsMalloc(void *arg, size_t size) {
static inline void *tdbDefaultMalloc(void *arg, size_t size) {
void *ptr;
ptr = malloc(size);
ptr = tdbOsMalloc(size);
return ptr;
}
static inline void tdbOsFree(void *arg, void *ptr) { free(ptr); }
static inline void tdbDefaultFree(void *arg, void *ptr) { tdbOsFree(ptr); }
static inline int tdbPutVarInt(u8 *p, int v) {
int n = 0;
......
source/libs/tdb/src/page/tdbPage.c
浏览文件 @ d4972b0d
......@@ -48,7 +48,7 @@ int tdbPageCreate(int pageSize, SPage **ppPage, void *(*xMalloc)(void *, size_t)
*ppPage = NULL;
size = pageSize + sizeof(*pPage);
if (xMalloc == NULL) {
xMalloc = tdbOsMalloc;
xMalloc = tdbDefaultMalloc;
}
ptr = (u8 *)((*xMalloc)(arg, size));
......@@ -76,7 +76,7 @@ int tdbPageDestroy(SPage *pPage, void (*xFree)(void *arg, void *ptr), void *arg)
u8 *ptr;
if (!xFree) {
xFree = tdbOsFree;
xFree = tdbDefaultFree;
}
ptr = pPage->pData;
......@@ -144,7 +144,7 @@ int tdbPageInsertCell(SPage *pPage, int idx, SCell *pCell, int szCell, u8 asOvfl
}
// TODO: here has memory leak
pNewCell = (SCell *)malloc(szCell);
pNewCell = (SCell *)tdbOsMalloc(szCell);
memcpy(pNewCell, pCell, szCell);
pPage->apOvfl[iOvfl] = pNewCell;
......@@ -372,11 +372,11 @@ static int tdbPageDefragment(SPage *pPage) {
int idx;
int iCell;
ASSERT(pPage->pFreeEnd - pPage->pFreeStart < nFree);
nFree = TDB_PAGE_NFREE(pPage);
nCells = TDB_PAGE_NCELLS(pPage);
ASSERT(pPage->pFreeEnd - pPage->pFreeStart < nFree);
// Loop to compact the page content
// Here we use an O(n^2) algorithm to do the job since
// this is a low frequency job.
......
source/libs/tdb/test/tdbTest.cpp
浏览文件 @ d4972b0d
......@@ -11,7 +11,7 @@ typedef struct SPoolMem {
} SPoolMem;
static SPoolMem *openPool() {
SPoolMem *pPool = (SPoolMem *)malloc(sizeof(*pPool));
SPoolMem *pPool = (SPoolMem *)tdbOsMalloc(sizeof(*pPool));
pPool->prev = pPool->next = pPool;
pPool->size = 0;
......@@ -31,20 +31,22 @@ static void closePool(SPoolMem *pPool) {
pMem->prev->next = pMem->next;
pPool->size -= pMem->size;
free(pMem);
tdbOsFree(pMem);
} while (1);
assert(pPool->size == 0);
free(pPool);
tdbOsFree(pPool);
}
#define clearPool closePool
static void *poolMalloc(void *arg, int size) {
void *ptr = NULL;
SPoolMem *pPool = (SPoolMem *)arg;
SPoolMem *pMem;
pMem = (SPoolMem *)malloc(sizeof(*pMem) + size);
pMem = (SPoolMem *)tdbOsMalloc(sizeof(*pMem) + size);
if (pMem == NULL) {
assert(0);
}
......@@ -71,7 +73,7 @@ static void poolFree(void *arg, void *ptr) {
pMem->prev->next = pMem->next;
pPool->size -= pMem->size;
free(pMem);
tdbOsFree(pMem);
}
static int tKeyCmpr(const void *pKey1, int kLen1, const void *pKey2, int kLen2) {
......@@ -113,10 +115,10 @@ static int tDefaultKeyCmpr(const void *pKey1, int keyLen1, const void *pKey2, in
TEST(tdb_test, simple_test) {
int ret;
STEnv *pEnv;
STDB *pDb;
TENV *pEnv;
TDB *pDb;
FKeyComparator compFunc;
int nData = 10000000;
int nData = 1000000;
// Open Env
ret = tdbEnvOpen("tdb", 4096, 256000, &pEnv);
......@@ -132,13 +134,34 @@ TEST(tdb_test, simple_test) {
char val[64];
{ // Insert some data
int i = 1;
SPoolMem *pPool;
int memPoolCapacity = 16 * 1024;
pPool = openPool();
tdbTxnBegin(pEnv);
for (;;) {
if (i > nData) break;
for (int i = 1; i <= nData; i++) {
sprintf(key, "key%d", i);
sprintf(val, "value%d", i);
ret = tdbDbInsert(pDb, key, strlen(key), val, strlen(val));
GTEST_ASSERT_EQ(ret, 0);
if (pPool->size >= memPoolCapacity) {
tdbTxnCommit(pEnv);
clearPool(pPool);
tdbTxnBegin(pEnv);
}
i++;
}
closePool(pPool);
}
{ // Query the data
......@@ -160,11 +183,11 @@ TEST(tdb_test, simple_test) {
}
{ // Iterate to query the DB data
STDBC *pDBC;
void *pKey = NULL;
void *pVal = NULL;
int vLen, kLen;
int count = 0;
TDBC *pDBC;
void *pKey = NULL;
void *pVal = NULL;
int vLen, kLen;
int count = 0;
ret = tdbDbcOpen(pDb, &pDBC);
GTEST_ASSERT_EQ(ret, 0);
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或