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

enh: add lrucache overwriter

This commit is contained in:
Shungang Li 2024-09-27 17:17:37 +08:00
parent 435af1e486
commit 10bd0fed21
6 changed files with 48 additions and 71 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
include/util/tlrucache.h
View File

@ -25,6 +25,7 @@ extern "C" {
typedef struct SLRUCache SLRUCache;
typedef void (*_taos_lru_deleter_t)(const void *key, size_t keyLen, void *value, void *ud);
typedef void (*_taos_lru_overwriter_t)(const void *key, size_t keyLen, void *value, void *ud);
typedef int (*_taos_lru_functor_t)(const void *key, size_t keyLen, void *value, void *ud);
typedef struct LRUHandle LRUHandle;
@ -42,7 +43,8 @@ SLRUCache *taosLRUCacheInit(size_t capacity, int numShardBits, double highPriPoo
void taosLRUCacheCleanup(SLRUCache *cache);
LRUStatus taosLRUCacheInsert(SLRUCache *cache, const void *key, size_t keyLen, void *value, size_t charge,
_taos_lru_deleter_t deleter, LRUHandle **handle, LRUPriority priority, void *ud);
_taos_lru_deleter_t deleter, _taos_lru_overwriter_t overwriter, LRUHandle **handle,
LRUPriority priority, void *ud);
LRUHandle *taosLRUCacheLookup(SLRUCache *cache, const void *key, size_t keyLen);
void taosLRUCacheErase(SLRUCache *cache, const void *key, size_t keyLen);

4
source/dnode/vnode/src/meta/metaCache.c
View File

@ -646,7 +646,7 @@ int32_t metaUidFilterCachePut(void* pVnode, uint64_t suid, const void* pKey, int
}
// add to cache.
(void)taosLRUCacheInsert(pCache, key, TAG_FILTER_RES_KEY_LEN, pPayload, payloadLen, freeUidCachePayload, NULL,
(void)taosLRUCacheInsert(pCache, key, TAG_FILTER_RES_KEY_LEN, pPayload, payloadLen, freeUidCachePayload, NULL, NULL,
TAOS_LRU_PRIORITY_LOW, NULL);
_end:
(void)taosThreadMutexUnlock(pLock);
@ -804,7 +804,7 @@ int32_t metaPutTbGroupToCache(void* pVnode, uint64_t suid, const void* pKey, int
}
// add to cache.
(void)taosLRUCacheInsert(pCache, key, TAG_FILTER_RES_KEY_LEN, pPayload, payloadLen, freeTbGroupCachePayload, NULL,
(void)taosLRUCacheInsert(pCache, key, TAG_FILTER_RES_KEY_LEN, pPayload, payloadLen, freeTbGroupCachePayload, NULL, NULL,
TAOS_LRU_PRIORITY_LOW, NULL);
_end:
(void)taosThreadMutexUnlock(pLock);

64
source/dnode/vnode/src/tsdb/tsdbCache.c
View File

@ -597,6 +597,11 @@ static void tsdbCacheDeleter(const void *key, size_t klen, void *value, void *ud
taosMemoryFree(value);
}
static void tsdbCacheOverWriter(const void *key, size_t klen, void *value, void *ud) {
SLastCol *pLastCol = (SLastCol *)value;
pLastCol->dirty = 0;
}
static int32_t tsdbCacheNewTableColumn(STsdb *pTsdb, int64_t uid, int16_t cid, int8_t col_type, int8_t lflag) {
int32_t code = 0, lino = 0;
@ -616,8 +621,8 @@ static int32_t tsdbCacheNewTableColumn(STsdb *pTsdb, int64_t uid, int16_t cid, i
TAOS_CHECK_EXIT(tsdbCacheReallocSLastCol(pLastCol, &charge));
SLastKey *pLastKey = &(SLastKey){.lflag = lflag, .uid = uid, .cid = cid};
LRUStatus status = taosLRUCacheInsert(pCache, pLastKey, ROCKS_KEY_LEN, pLastCol, charge, tsdbCacheDeleter, NULL,
TAOS_LRU_PRIORITY_LOW, pTsdb);
LRUStatus status = taosLRUCacheInsert(pCache, pLastKey, ROCKS_KEY_LEN, pLastCol, charge, tsdbCacheDeleter,
tsdbCacheOverWriter, NULL, TAOS_LRU_PRIORITY_LOW, pTsdb);
if (status != TAOS_LRU_STATUS_OK) {
tsdbError("vgId:%d, %s failed at line %d status %d.", TD_VID(pTsdb->pVnode), __func__, __LINE__, status);
code = TSDB_CODE_FAILED;
@ -1071,40 +1076,6 @@ typedef struct {
SLastKey key;
} SIdxKey;
static int32_t tsdbCacheUpdateValue(SValue *pOld, SValue *pNew) {
uint8_t *pFree = NULL;
int nData = 0;
if (IS_VAR_DATA_TYPE(pOld->type)) {
pFree = pOld->pData;
nData = pOld->nData;
}
*pOld = *pNew;
if (IS_VAR_DATA_TYPE(pNew->type)) {
if (nData < pNew->nData) {
pOld->pData = taosMemoryCalloc(1, pNew->nData);
if (!pOld->pData) {
return terrno;
}
} else {
pOld->pData = pFree;
pFree = NULL;
}
if (pNew->nData) {
memcpy(pOld->pData, pNew->pData, pNew->nData);
} else {
pFree = pOld->pData;
pOld->pData = NULL;
}
}
taosMemoryFreeClear(pFree);
TAOS_RETURN(TSDB_CODE_SUCCESS);
}
static void tsdbCacheUpdateLastColToNone(SLastCol *pLastCol, ELastCacheStatus cacheStatus) {
// update rowkey
pLastCol->rowKey.ts = TSKEY_MIN;
@ -1128,11 +1099,7 @@ static void tsdbCacheUpdateLastColToNone(SLastCol *pLastCol, ELastCacheStatus ca
}
pLastCol->colVal = COL_VAL_NONE(pLastCol->colVal.cid, pLastCol->colVal.value.type);
if (!pLastCol->dirty) {
pLastCol->dirty = 1;
}
pLastCol->dirty = 1;
pLastCol->cacheStatus = cacheStatus;
}
@ -1168,7 +1135,7 @@ static int32_t tsdbCachePutToLRU(STsdb *pTsdb, SLastKey *pLastKey, SLastCol *pLa
TAOS_CHECK_EXIT(tsdbCacheReallocSLastCol(pLRULastCol, &charge));
LRUStatus status = taosLRUCacheInsert(pTsdb->lruCache, pLastKey, ROCKS_KEY_LEN, pLRULastCol, charge, tsdbCacheDeleter,
NULL, TAOS_LRU_PRIORITY_LOW, pTsdb);
tsdbCacheOverWriter, NULL, TAOS_LRU_PRIORITY_LOW, pTsdb);
if (TAOS_LRU_STATUS_OK != status && TAOS_LRU_STATUS_OK_OVERWRITTEN != status) {
tsdbError("vgId:%d, %s failed at line %d status %d.", TD_VID(pTsdb->pVnode), __func__, __LINE__, status);
code = TSDB_CODE_FAILED;
@ -1215,7 +1182,8 @@ static int32_t tsdbCacheUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, SArray
if (pLastCol->cacheStatus != TSDB_LAST_CACHE_NO_CACHE) {
int32_t cmp_res = tRowKeyCompare(&pLastCol->rowKey, pRowKey);
if (cmp_res < 0 || (cmp_res == 0 && !COL_VAL_IS_NONE(pColVal))) {
SLastCol newLastCol = {.rowKey = *pRowKey, .colVal = *pColVal, .dirty = 1, .cacheStatus = TSDB_LAST_CACHE_VALID};
SLastCol newLastCol = {
.rowKey = *pRowKey, .colVal = *pColVal, .dirty = 1, .cacheStatus = TSDB_LAST_CACHE_VALID};
code = tsdbCachePutToLRU(pTsdb, key, &newLastCol);
}
}
@ -1694,8 +1662,8 @@ static int32_t tsdbCacheLoadFromRaw(STsdb *pTsdb, tb_uid_t uid, SArray *pLastArr
TAOS_CHECK_EXIT(code);
}
LRUStatus status = taosLRUCacheInsert(pCache, &idxKey->key, ROCKS_KEY_LEN, pLastCol, charge, tsdbCacheDeleter, NULL,
TAOS_LRU_PRIORITY_LOW, pTsdb);
LRUStatus status = taosLRUCacheInsert(pCache, &idxKey->key, ROCKS_KEY_LEN, pLastCol, charge, tsdbCacheDeleter,
tsdbCacheOverWriter, NULL, TAOS_LRU_PRIORITY_LOW, pTsdb);
if (TAOS_LRU_STATUS_OK != status && TAOS_LRU_STATUS_OK_OVERWRITTEN != status) {
tsdbError("vgId:%d, %s failed at line %d status %d.", TD_VID(pTsdb->pVnode), __func__, __LINE__, status);
pLastCol = NULL;
@ -1804,7 +1772,7 @@ static int32_t tsdbCacheLoadFromRocks(STsdb *pTsdb, tb_uid_t uid, SArray *pLastA
}
LRUStatus status = taosLRUCacheInsert(pCache, &idxKey->key, ROCKS_KEY_LEN, pLastCol, charge, tsdbCacheDeleter,
NULL, TAOS_LRU_PRIORITY_LOW, pTsdb);
tsdbCacheOverWriter, NULL, TAOS_LRU_PRIORITY_LOW, pTsdb);
if (TAOS_LRU_STATUS_OK != status && TAOS_LRU_STATUS_OK_OVERWRITTEN != status) {
tsdbError("vgId:%d, %s failed at line %d status %d.", TD_VID(pTsdb->pVnode), __func__, __LINE__, status);
taosMemoryFreeClear(pToFree);
@ -3659,7 +3627,7 @@ int32_t tsdbCacheGetBlockS3(SLRUCache *pCache, STsdbFD *pFD, LRUHandle **handle)
size_t charge = tsS3BlockSize * pFD->szPage;
_taos_lru_deleter_t deleter = deleteBCache;
LRUStatus status =
taosLRUCacheInsert(pCache, key, keyLen, pBlock, charge, deleter, &h, TAOS_LRU_PRIORITY_LOW, NULL);
taosLRUCacheInsert(pCache, key, keyLen, pBlock, charge, deleter, NULL, &h, TAOS_LRU_PRIORITY_LOW, NULL);
if (status != TAOS_LRU_STATUS_OK) {
// code = -1;
}
@ -3702,7 +3670,7 @@ void tsdbCacheSetPageS3(SLRUCache *pCache, STsdbFD *pFD, int64_t pgno, uint8_t *
memcpy(pPg, pPage, charge);
LRUStatus status =
taosLRUCacheInsert(pCache, key, keyLen, pPg, charge, deleter, &handle, TAOS_LRU_PRIORITY_LOW, NULL);
taosLRUCacheInsert(pCache, key, keyLen, pPg, charge, deleter, NULL, &handle, TAOS_LRU_PRIORITY_LOW, NULL);
if (status != TAOS_LRU_STATUS_OK) {
// ignore cache updating if not ok
// code = TSDB_CODE_OUT_OF_MEMORY;

2
source/libs/executor/src/scanoperator.c
View File

@ -737,7 +737,7 @@ _end:
if (NULL != pVal) {
insertRet = taosLRUCacheInsert(pCache->pTableMetaEntryCache, &pBlock->info.id.uid, sizeof(uint64_t), pVal,
sizeof(STableCachedVal), freeCachedMetaItem, NULL, TAOS_LRU_PRIORITY_LOW, NULL);
sizeof(STableCachedVal), freeCachedMetaItem, NULL, NULL, TAOS_LRU_PRIORITY_LOW, NULL);
if (insertRet != TAOS_LRU_STATUS_OK) {
qWarn("failed to put meta into lru cache, code:%d, %s", insertRet, idStr);
}

2
source/libs/index/src/indexFstFile.c
View File

@ -148,7 +148,7 @@ static int idxFileCtxDoReadFrom(IFileCtx* ctx, uint8_t* buf, int len, int32_t of
memcpy(buf + total, blk->buf + blkOffset, nread);
LRUStatus s = taosLRUCacheInsert(ctx->lru, key, strlen(key), blk, cacheMemSize, deleteDataBlockFromLRU, NULL,
TAOS_LRU_PRIORITY_LOW, NULL);
NULL, TAOS_LRU_PRIORITY_LOW, NULL);
if (s != TAOS_LRU_STATUS_OK) {
return -1;
}

43
source/util/src/tlrucache.c
View File

@ -38,18 +38,19 @@ enum {
};
struct SLRUEntry {
void *value;
_taos_lru_deleter_t deleter;
void *ud;
SLRUEntry *nextHash;
SLRUEntry *next;
SLRUEntry *prev;
size_t totalCharge;
size_t keyLength;
uint32_t hash;
uint32_t refs;
uint8_t flags;
char keyData[1];
void *value;
_taos_lru_deleter_t deleter;
_taos_lru_overwriter_t overwriter;
void *ud;
SLRUEntry *nextHash;
SLRUEntry *next;
SLRUEntry *prev;
size_t totalCharge;
size_t keyLength;
uint32_t hash;
uint32_t refs;
uint8_t flags;
char keyData[1];
};
#define TAOS_LRU_ENTRY_IN_CACHE(h) ((h)->flags & TAOS_LRU_IN_CACHE)
@ -403,6 +404,10 @@ static LRUStatus taosLRUCacheShardInsertEntry(SLRUCacheShard *shard, SLRUEntry *
if (old != NULL) {
status = TAOS_LRU_STATUS_OK_OVERWRITTEN;
if (old->overwriter) {
(*old->overwriter)(old->keyData, old->keyLength, old->value, old->ud);
}
TAOS_LRU_ENTRY_SET_IN_CACHE(old, false);
if (!TAOS_LRU_ENTRY_HAS_REFS(old)) {
taosLRUCacheShardLRURemove(shard, old);
@ -440,8 +445,9 @@ _exit:
}
static LRUStatus taosLRUCacheShardInsert(SLRUCacheShard *shard, const void *key, size_t keyLen, uint32_t hash,
void *value, size_t charge, _taos_lru_deleter_t deleter, LRUHandle **handle,
LRUPriority priority, void *ud) {
void *value, size_t charge, _taos_lru_deleter_t deleter,
_taos_lru_overwriter_t overwriter, LRUHandle **handle, LRUPriority priority,
void *ud) {
SLRUEntry *e = taosMemoryCalloc(1, sizeof(SLRUEntry) - 1 + keyLen);
if (!e) {
if (deleter) {
@ -453,6 +459,7 @@ static LRUStatus taosLRUCacheShardInsert(SLRUCacheShard *shard, const void *key,
e->value = value;
e->flags = 0;
e->deleter = deleter;
e->overwriter = overwriter;
e->ud = ud;
e->keyLength = keyLen;
e->hash = hash;
@ -726,12 +733,12 @@ void taosLRUCacheCleanup(SLRUCache *cache) {
}
LRUStatus taosLRUCacheInsert(SLRUCache *cache, const void *key, size_t keyLen, void *value, size_t charge,
_taos_lru_deleter_t deleter, LRUHandle **handle, LRUPriority priority, void *ud) {
_taos_lru_deleter_t deleter, _taos_lru_overwriter_t overwriter, LRUHandle **handle, LRUPriority priority, void *ud) {
uint32_t hash = TAOS_LRU_CACHE_SHARD_HASH32(key, keyLen);
uint32_t shardIndex = hash & cache->shardedCache.shardMask;
return taosLRUCacheShardInsert(&cache->shards[shardIndex], key, keyLen, hash, value, charge, deleter, handle,
priority, ud);
return taosLRUCacheShardInsert(&cache->shards[shardIndex], key, keyLen, hash, value, charge, deleter, overwriter,
handle, priority, ud);
}
LRUHandle *taosLRUCacheLookup(SLRUCache *cache, const void *key, size_t keyLen) {
@ -869,4 +876,4 @@ bool taosLRUCacheIsStrictCapacity(SLRUCache *cache) {
(void)taosThreadMutexUnlock(&cache->shardedCache.capacityMutex);
return strict;
}
}