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

Merge pull request #19666 from taosdata/fix/nodisk 

refactor:do some internal refactor.
This commit is contained in:
Haojun Liao 2023-01-25 22:26:14 +08:00 committed by GitHub
parent ab1f87d19f fc80c3d373
commit 9615504024
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 83 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

42
source/util/src/tarray.c
View File

@ -20,7 +20,10 @@
// todo refactor API // todo refactor API
SArray* taosArrayInit(size_t size, size_t elemSize) { SArray* taosArrayInit(size_t size, size_t elemSize) {
assert(elemSize > 0); if (elemSize == 0) {
terrno = TSDB_CODE_INVALID_PARA;
return NULL;
}
if (size < TARRAY_MIN_SIZE) { if (size < TARRAY_MIN_SIZE) {
size = TARRAY_MIN_SIZE; size = TARRAY_MIN_SIZE;
@ -96,8 +99,6 @@ void* taosArrayAddBatch(SArray* pArray, const void* pData, int32_t nEles) {
} }
void taosArrayRemoveDuplicate(SArray* pArray, __compar_fn_t comparFn, void (*fp)(void*)) { void taosArrayRemoveDuplicate(SArray* pArray, __compar_fn_t comparFn, void (*fp)(void*)) {
assert(pArray);
size_t size = pArray->size; size_t size = pArray->size;
if (size <= 1) { if (size <= 1) {
return; return;
@ -136,8 +137,6 @@ void taosArrayRemoveDuplicate(SArray* pArray, __compar_fn_t comparFn, void (*fp)
} }
void taosArrayRemoveDuplicateP(SArray* pArray, __compar_fn_t comparFn, void (*fp)(void*)) { void taosArrayRemoveDuplicateP(SArray* pArray, __compar_fn_t comparFn, void (*fp)(void*)) {
assert(pArray);
size_t size = pArray->size; size_t size = pArray->size;
if (size <= 1) { if (size <= 1) {
return; return;
@ -195,11 +194,10 @@ void* taosArrayReserve(SArray* pArray, int32_t num) {
} }
void* taosArrayPop(SArray* pArray) { void* taosArrayPop(SArray* pArray) {
assert(pArray != NULL);
if (pArray->size == 0) { if (pArray->size == 0) {
return NULL; return NULL;
} }
pArray->size -= 1; pArray->size -= 1;
return TARRAY_GET_ELEM(pArray, pArray->size); return TARRAY_GET_ELEM(pArray, pArray->size);
} }
@ -208,16 +206,21 @@ void* taosArrayGet(const SArray* pArray, size_t index) {
if (NULL == pArray) { if (NULL == pArray) {
return NULL; return NULL;
} }
assert(index < pArray->size);
if (index >= pArray->size) {
uError("index is out of range, current:%"PRIzu" max:%d", index, pArray->capacity);
return NULL;
}
return TARRAY_GET_ELEM(pArray, index); return TARRAY_GET_ELEM(pArray, index);
} }
void* taosArrayGetP(const SArray* pArray, size_t index) { void* taosArrayGetP(const SArray* pArray, size_t index) {
assert(index < pArray->size); void** p = taosArrayGet(pArray, index);
if (p == NULL) {
void* d = TARRAY_GET_ELEM(pArray, index); return NULL;
}
return *(void**)d; return *p;
} }
void* taosArrayGetLast(const SArray* pArray) { return TARRAY_GET_ELEM(pArray, pArray->size - 1); } void* taosArrayGetLast(const SArray* pArray) { return TARRAY_GET_ELEM(pArray, pArray->size - 1); }
@ -296,9 +299,12 @@ void taosArrayRemove(SArray* pArray, size_t index) {
} }
SArray* taosArrayFromList(const void* src, size_t size, size_t elemSize) { SArray* taosArrayFromList(const void* src, size_t size, size_t elemSize) {
assert(src != NULL && elemSize > 0); if (elemSize <= 0) {
SArray* pDst = taosArrayInit(size, elemSize); terrno = TSDB_CODE_INVALID_PARA;
return NULL;
}
SArray* pDst = taosArrayInit(size, elemSize);
memcpy(pDst->pData, src, elemSize * size); memcpy(pDst->pData, src, elemSize * size);
pDst->size = size; pDst->size = size;
@ -306,8 +312,6 @@ SArray* taosArrayFromList(const void* src, size_t size, size_t elemSize) {
} }
SArray* taosArrayDup(const SArray* pSrc, __array_item_dup_fn_t fn) { SArray* taosArrayDup(const SArray* pSrc, __array_item_dup_fn_t fn) {
assert(pSrc != NULL);
if (pSrc->size == 0) { // empty array list if (pSrc->size == 0) { // empty array list
return taosArrayInit(8, pSrc->elemSize); return taosArrayInit(8, pSrc->elemSize);
} }
@ -399,14 +403,10 @@ void taosArrayDestroyEx(SArray* pArray, FDelete fp) {
} }
void taosArraySort(SArray* pArray, __compar_fn_t compar) { void taosArraySort(SArray* pArray, __compar_fn_t compar) {
ASSERT(pArray != NULL && compar != NULL);
taosSort(pArray->pData, pArray->size, pArray->elemSize, compar); taosSort(pArray->pData, pArray->size, pArray->elemSize, compar);
} }
void* taosArraySearch(const SArray* pArray, const void* key, __compar_fn_t comparFn, int32_t flags) { void* taosArraySearch(const SArray* pArray, const void* key, __compar_fn_t comparFn, int32_t flags) {
assert(pArray != NULL && comparFn != NULL);
assert(key != NULL);
return taosbsearch(key, pArray->pData, pArray->size, pArray->elemSize, comparFn, flags); return taosbsearch(key, pArray->pData, pArray->size, pArray->elemSize, comparFn, flags);
} }

88
source/util/src/tpagedbuf.c
View File

@ -6,6 +6,9 @@
#include "tlog.h" #include "tlog.h"
#define GET_PAYLOAD_DATA(_p) ((char*)(_p)->pData + POINTER_BYTES) #define GET_PAYLOAD_DATA(_p) ((char*)(_p)->pData + POINTER_BYTES)
#define BUF_PAGE_IN_MEM(_p) ((_p)->pData != NULL)
#define CLEAR_BUF_PAGE_IN_MEM_FLAG(_p) ((_p)->pData = NULL)
#define HAS_DATA_IN_DISK(_p) ((_p)->offset >= 0)
#define NO_IN_MEM_AVAILABLE_PAGES(_b) (listNEles((_b)->lruList) >= (_b)->inMemPages) #define NO_IN_MEM_AVAILABLE_PAGES(_b) (listNEles((_b)->lruList) >= (_b)->inMemPages)
typedef struct SPageDiskInfo { typedef struct SPageDiskInfo {
@ -14,7 +17,7 @@ typedef struct SPageDiskInfo {
} SPageDiskInfo, SFreeListItem; } SPageDiskInfo, SFreeListItem;
struct SPageInfo { struct SPageInfo {
SListNode* pn; // point to list node struct SListNode* pn; // point to list node struct. it is NULL when the page is evicted from the in-memory buffer
void* pData; void* pData;
int64_t offset; int64_t offset;
int32_t pageId; int32_t pageId;
@ -112,8 +115,6 @@ static uint64_t allocateNewPositionInFile(SDiskbasedBuf* pBuf, size_t size) {
} }
} }
static void setPageNotInBuf(SPageInfo* pPageInfo) { pPageInfo->pData = NULL; }
/** /**
* +--------------------------+-------------------+--------------+ * +--------------------------+-------------------+--------------+
* | PTR to SPageInfo (8bytes)| Payload (PageSize)| 2 Extra Bytes| * | PTR to SPageInfo (8bytes)| Payload (PageSize)| 2 Extra Bytes|
@ -126,21 +127,26 @@ static void setPageNotInBuf(SPageInfo* pPageInfo) { pPageInfo->pData = NULL; }
static FORCE_INLINE size_t getAllocPageSize(int32_t pageSize) { return pageSize + POINTER_BYTES + sizeof(SFilePage); } static FORCE_INLINE size_t getAllocPageSize(int32_t pageSize) { return pageSize + POINTER_BYTES + sizeof(SFilePage); }
static char* doFlushBufPage(SDiskbasedBuf* pBuf, SPageInfo* pg) { static char* doFlushBufPage(SDiskbasedBuf* pBuf, SPageInfo* pg) {
ASSERT(!pg->used && pg->pData != NULL); if (pg->pData == NULL || pg->used) {
uError("invalid params in paged buffer process when flushing buf to disk, %s", pBuf->id);
terrno = TSDB_CODE_INVALID_PARA;
return NULL;
}
int32_t size = pBuf->pageSize; int32_t size = pBuf->pageSize;
char* t = NULL; char* t = NULL;
if (pg->offset == -1 || pg->dirty) { if ((!HAS_DATA_IN_DISK(pg)) || pg->dirty) {
void* payload = GET_PAYLOAD_DATA(pg); void* payload = GET_PAYLOAD_DATA(pg);
t = doCompressData(payload, pBuf->pageSize, &size, pBuf); t = doCompressData(payload, pBuf->pageSize, &size, pBuf);
ASSERTS(size >= 0, "size is negative"); if (size < 0) {
uError("failed to compress data when flushing data to disk, %s", pBuf->id);
return NULL;
}
} }
// this page is flushed to disk for the first time // this page is flushed to disk for the first time
if (pg->dirty) { if (pg->dirty) {
if (pg->offset == -1) { if (!HAS_DATA_IN_DISK(pg)) {
ASSERTS(pg->dirty == true, "pg->dirty is false");
pg->offset = allocateNewPositionInFile(pBuf, size); pg->offset = allocateNewPositionInFile(pBuf, size);
pBuf->nextPos += size; pBuf->nextPos += size;
@ -156,6 +162,7 @@ static char* doFlushBufPage(SDiskbasedBuf* pBuf, SPageInfo* pg) {
return NULL; return NULL;
} }
// extend the file size
if (pBuf->fileSize < pg->offset + size) { if (pBuf->fileSize < pg->offset + size) {
pBuf->fileSize = pg->offset + size; pBuf->fileSize = pg->offset + size;
} }
@ -198,13 +205,13 @@ static char* doFlushBufPage(SDiskbasedBuf* pBuf, SPageInfo* pg) {
size = pg->length; size = pg->length;
} }
ASSERT(size > 0 || (pg->offset == -1 && pg->length == -1));
char* pDataBuf = pg->pData; char* pDataBuf = pg->pData;
memset(pDataBuf, 0, getAllocPageSize(pBuf->pageSize)); memset(pDataBuf, 0, getAllocPageSize(pBuf->pageSize));
#ifdef BUF_PAGE_DEBUG #ifdef BUF_PAGE_DEBUG
uDebug("page_flush %p, pageId:%d, offset:%d", pDataBuf, pg->pageId, pg->offset); uDebug("page_flush %p, pageId:%d, offset:%d", pDataBuf, pg->pageId, pg->offset);
#endif #endif
pg->length = size; // on disk size pg->length = size; // on disk size
return pDataBuf; return pDataBuf;
} }
@ -220,13 +227,19 @@ static char* flushBufPage(SDiskbasedBuf* pBuf, SPageInfo* pg) {
} }
char* p = doFlushBufPage(pBuf, pg); char* p = doFlushBufPage(pBuf, pg);
setPageNotInBuf(pg); CLEAR_BUF_PAGE_IN_MEM_FLAG(pg);
pg->dirty = false; pg->dirty = false;
return p; return p;
} }
// load file block data in disk // load file block data in disk
static int32_t loadPageFromDisk(SDiskbasedBuf* pBuf, SPageInfo* pg) { static int32_t loadPageFromDisk(SDiskbasedBuf* pBuf, SPageInfo* pg) {
if (pg->offset < 0 || pg->length <= 0) {
uError("failed to load buf page from disk, offset:%"PRId64", length:%d, %s", pg->offset, pg->length, pBuf->id);
return TSDB_CODE_INVALID_PARA;
}
int32_t ret = taosLSeekFile(pBuf->pFile, pg->offset, SEEK_SET); int32_t ret = taosLSeekFile(pBuf->pFile, pg->offset, SEEK_SET);
if (ret == -1) { if (ret == -1) {
ret = TAOS_SYSTEM_ERROR(errno); ret = TAOS_SYSTEM_ERROR(errno);
@ -248,7 +261,7 @@ static int32_t loadPageFromDisk(SDiskbasedBuf* pBuf, SPageInfo* pg) {
return 0; return 0;
} }
static SPageInfo* registerPage(SDiskbasedBuf* pBuf, int32_t pageId) { static SPageInfo* registerNewPageInfo(SDiskbasedBuf* pBuf, int32_t pageId) {
pBuf->numOfPages += 1; pBuf->numOfPages += 1;
SPageInfo* ppi = taosMemoryMalloc(sizeof(SPageInfo)); SPageInfo* ppi = taosMemoryMalloc(sizeof(SPageInfo));
@ -275,7 +288,9 @@ static SListNode* getEldestUnrefedPage(SDiskbasedBuf* pBuf) {
SListNode* pn = NULL; SListNode* pn = NULL;
while ((pn = tdListNext(&iter)) != NULL) { while ((pn = tdListNext(&iter)) != NULL) {
SPageInfo* pageInfo = *(SPageInfo**)pn->data; SPageInfo* pageInfo = *(SPageInfo**)pn->data;
ASSERT(pageInfo->pageId >= 0 && pageInfo->pn == pn);
SPageInfo* p = *(SPageInfo**)(pageInfo->pData);
ASSERT(pageInfo->pageId >= 0 && pageInfo->pn == pn && p == pageInfo);
if (!pageInfo->used) { if (!pageInfo->used) {
break; break;
@ -295,7 +310,6 @@ static char* evictBufPage(SDiskbasedBuf* pBuf) {
tdListPopNode(pBuf->lruList, pn); tdListPopNode(pBuf->lruList, pn);
SPageInfo* d = *(SPageInfo**)pn->data; SPageInfo* d = *(SPageInfo**)pn->data;
ASSERTS(d->pn == pn, "d->pn not equal pn");
d->pn = NULL; d->pn = NULL;
taosMemoryFreeClear(pn); taosMemoryFreeClear(pn);
@ -333,7 +347,6 @@ int32_t createDiskbasedBuf(SDiskbasedBuf** pBuf, int32_t pagesize, int32_t inMem
pPBuf->pageSize = pagesize; pPBuf->pageSize = pagesize;
pPBuf->numOfPages = 0; // all pages are in buffer in the first place pPBuf->numOfPages = 0; // all pages are in buffer in the first place
pPBuf->totalBufSize = 0; pPBuf->totalBufSize = 0;
pPBuf->inMemPages = inMemBufSize / pagesize; // maximum allowed pages, it is a soft limit.
pPBuf->allocateId = -1; pPBuf->allocateId = -1;
pPBuf->pFile = NULL; pPBuf->pFile = NULL;
pPBuf->id = strdup(id); pPBuf->id = strdup(id);
@ -342,13 +355,22 @@ int32_t createDiskbasedBuf(SDiskbasedBuf** pBuf, int32_t pagesize, int32_t inMem
pPBuf->freePgList = tdListNew(POINTER_BYTES); pPBuf->freePgList = tdListNew(POINTER_BYTES);
// at least more than 2 pages must be in memory // at least more than 2 pages must be in memory
ASSERT(inMemBufSize >= pagesize * 2); if (inMemBufSize < pagesize * 2) {
inMemBufSize = pagesize * 2;
}
pPBuf->inMemPages = inMemBufSize / pagesize; // maximum allowed pages, it is a soft limit.
pPBuf->lruList = tdListNew(POINTER_BYTES); pPBuf->lruList = tdListNew(POINTER_BYTES);
if (pPBuf->lruList == NULL) {
goto _error;
}
// init id hash table // init id hash table
_hash_fn_t fn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT); _hash_fn_t fn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT);
pPBuf->pIdList = taosArrayInit(4, POINTER_BYTES); pPBuf->pIdList = taosArrayInit(4, POINTER_BYTES);
if (pPBuf->pIdList == NULL) {
goto _error;
}
pPBuf->assistBuf = taosMemoryMalloc(pPBuf->pageSize + 2); // EXTRA BYTES pPBuf->assistBuf = taosMemoryMalloc(pPBuf->pageSize + 2); // EXTRA BYTES
if (pPBuf->assistBuf == NULL) { if (pPBuf->assistBuf == NULL) {
@ -410,7 +432,7 @@ void* getNewBufPage(SDiskbasedBuf* pBuf, int32_t* pageId) {
*pageId = (++pBuf->allocateId); *pageId = (++pBuf->allocateId);
// register page id info // register page id info
pi = registerPage(pBuf, *pageId); pi = registerNewPageInfo(pBuf, *pageId);
if (pi == NULL) { if (pi == NULL) {
return NULL; return NULL;
} }
@ -434,15 +456,21 @@ void* getNewBufPage(SDiskbasedBuf* pBuf, int32_t* pageId) {
void* getBufPage(SDiskbasedBuf* pBuf, int32_t id) { void* getBufPage(SDiskbasedBuf* pBuf, int32_t id) {
if (id < 0) { if (id < 0) {
terrno = TSDB_CODE_INVALID_PARA;
uError("invalid page id:%d, %s", id, pBuf->id);
return NULL; return NULL;
} }
pBuf->statis.getPages += 1; pBuf->statis.getPages += 1;
SPageInfo** pi = taosHashGet(pBuf->all, &id, sizeof(int32_t)); SPageInfo** pi = taosHashGet(pBuf->all, &id, sizeof(int32_t));
ASSERT(pi != NULL && *pi != NULL); if (pi == NULL || *pi == NULL) {
uError("failed to locate the buffer page:%d, %s", id, pBuf->id);
terrno = TSDB_CODE_INVALID_PARA;
return NULL;
}
if ((*pi)->pData != NULL) { // it is in memory if (BUF_PAGE_IN_MEM(*pi)) { // it is in memory
// no need to update the LRU list if only one page exists // no need to update the LRU list if only one page exists
if (pBuf->numOfPages == 1) { if (pBuf->numOfPages == 1) {
(*pi)->used = true; (*pi)->used = true;
@ -450,7 +478,10 @@ void* getBufPage(SDiskbasedBuf* pBuf, int32_t id) {
} }
SPageInfo** pInfo = (SPageInfo**)((*pi)->pn->data); SPageInfo** pInfo = (SPageInfo**)((*pi)->pn->data);
ASSERT(*pInfo == *pi); if (*pInfo != *pi) {
uError("inconsistently data in paged buffer, pInfo:%p, pi:%p, %s", *pInfo, *pi, pBuf->id);
return NULL;
}
lruListMoveToFront(pBuf->lruList, (*pi)); lruListMoveToFront(pBuf->lruList, (*pi));
(*pi)->used = true; (*pi)->used = true;
@ -460,10 +491,12 @@ void* getBufPage(SDiskbasedBuf* pBuf, int32_t id) {
#endif #endif
return (void*)(GET_PAYLOAD_DATA(*pi)); return (void*)(GET_PAYLOAD_DATA(*pi));
} else { // not in memory } else { // not in memory
ASSERT((*pi)->pData == NULL && (*pi)->pn == NULL && ASSERT((!BUF_PAGE_IN_MEM(*pi)) && (*pi)->pn == NULL &&
(((*pi)->length >= 0 && (*pi)->offset >= 0) || ((*pi)->length == -1 && (*pi)->offset == -1))); (((*pi)->length >= 0 && (*pi)->offset >= 0) || ((*pi)->length == -1 && (*pi)->offset == -1)));
(*pi)->pData = doExtractPage(pBuf); (*pi)->pData = doExtractPage(pBuf);
// failed to evict buffer page, return with error code.
if ((*pi)->pData == NULL) { if ((*pi)->pData == NULL) {
return NULL; return NULL;
} }
@ -475,7 +508,7 @@ void* getBufPage(SDiskbasedBuf* pBuf, int32_t id) {
(*pi)->used = true; (*pi)->used = true;
// some data has been flushed to disk, and needs to be loaded into buffer again. // some data has been flushed to disk, and needs to be loaded into buffer again.
if ((*pi)->length > 0 && (*pi)->offset >= 0) { if (HAS_DATA_IN_DISK(*pi)) {
int32_t code = loadPageFromDisk(pBuf, *pi); int32_t code = loadPageFromDisk(pBuf, *pi);
if (code != 0) { if (code != 0) {
terrno = code; terrno = code;
@ -503,7 +536,12 @@ void releaseBufPageInfo(SDiskbasedBuf* pBuf, SPageInfo* pi) {
uDebug("page_releaseBufPageInfo pageId:%d, used:%d, offset:%" PRId64, pi->pageId, pi->used, pi->offset); uDebug("page_releaseBufPageInfo pageId:%d, used:%d, offset:%" PRId64, pi->pageId, pi->used, pi->offset);
#endif #endif
if (ASSERTS(pi->pData != NULL, "pi->pData is NULL")) { if (pi == NULL) {
return;
}
if (pi->pData == NULL) {
uError("pi->pData (page data) is null");
return; return;
} }
@ -514,7 +552,6 @@ void releaseBufPageInfo(SDiskbasedBuf* pBuf, SPageInfo* pi) {
size_t getTotalBufSize(const SDiskbasedBuf* pBuf) { return (size_t)pBuf->totalBufSize; } size_t getTotalBufSize(const SDiskbasedBuf* pBuf) { return (size_t)pBuf->totalBufSize; }
SArray* getDataBufPagesIdList(SDiskbasedBuf* pBuf) { SArray* getDataBufPagesIdList(SDiskbasedBuf* pBuf) {
ASSERT(pBuf != NULL);
return pBuf->pIdList; return pBuf->pIdList;
} }
@ -592,7 +629,6 @@ SPageInfo* getLastPageInfo(SArray* pList) {
} }
int32_t getPageId(const SPageInfo* pPgInfo) { int32_t getPageId(const SPageInfo* pPgInfo) {
ASSERT(pPgInfo != NULL);
return pPgInfo->pageId; return pPgInfo->pageId;
} }