Merge pull request #28199 from taosdata/fix/TD-32338-3.0
enh: (last) performance issue
This commit is contained in:
commit
318ae7fd5e
|
@ -342,7 +342,10 @@ typedef struct {
|
|||
rocksdb_writeoptions_t *writeoptions;
|
||||
rocksdb_readoptions_t *readoptions;
|
||||
rocksdb_writebatch_t *writebatch;
|
||||
TdThreadMutex writeBatchMutex;
|
||||
TdThreadMutex writeBatchMutex;
|
||||
int32_t sver;
|
||||
tb_uid_t suid;
|
||||
tb_uid_t uid;
|
||||
STSchema *pTSchema;
|
||||
} SRocksCache;
|
||||
|
||||
|
|
|
@ -222,6 +222,7 @@ int32_t tsdbCacheNewSTableColumn(STsdb* pTsdb, SArray* uids, int16_t cid, int8_t
|
|||
int32_t tsdbCacheDropSTableColumn(STsdb* pTsdb, SArray* uids, int16_t cid, bool hasPrimayKey);
|
||||
int32_t tsdbCacheNewNTableColumn(STsdb* pTsdb, int64_t uid, int16_t cid, int8_t col_type);
|
||||
int32_t tsdbCacheDropNTableColumn(STsdb* pTsdb, int64_t uid, int16_t cid, bool hasPrimayKey);
|
||||
void tsdbCacheInvalidateSchema(STsdb* pTsdb, tb_uid_t suid, tb_uid_t uid, int32_t sver);
|
||||
int tsdbScanAndConvertSubmitMsg(STsdb* pTsdb, SSubmitReq2* pMsg);
|
||||
int tsdbInsertData(STsdb* pTsdb, int64_t version, SSubmitReq2* pMsg, SSubmitRsp2* pRsp);
|
||||
int32_t tsdbInsertTableData(STsdb* pTsdb, int64_t version, SSubmitTbData* pSubmitTbData, int32_t* affectedRows);
|
||||
|
|
|
@ -620,6 +620,8 @@ int metaAlterSTable(SMeta *pMeta, int64_t version, SVCreateStbReq *pReq) {
|
|||
}
|
||||
}
|
||||
if (uids) taosArrayDestroy(uids);
|
||||
|
||||
tsdbCacheInvalidateSchema(pTsdb, pReq->suid, -1, pReq->schemaRow.version);
|
||||
}
|
||||
|
||||
metaWLock(pMeta);
|
||||
|
@ -1945,6 +1947,10 @@ static int metaAlterTableColumn(SMeta *pMeta, int64_t version, SVAlterTbReq *pAl
|
|||
break;
|
||||
}
|
||||
|
||||
if (!TSDB_CACHE_NO(pMeta->pVnode->config)) {
|
||||
tsdbCacheInvalidateSchema(pMeta->pVnode->pTsdb, 0, entry.uid, pSchema->version);
|
||||
}
|
||||
|
||||
entry.version = version;
|
||||
|
||||
// do actual write
|
||||
|
|
|
@ -221,7 +221,7 @@ static int32_t tsdbOpenRocksCache(STsdb *pTsdb) {
|
|||
|
||||
rocksdb_writebatch_t *writebatch = rocksdb_writebatch_create();
|
||||
|
||||
TAOS_CHECK_GOTO(taosThreadMutexInit(&pTsdb->rCache.writeBatchMutex, NULL), &lino, _err6) ;
|
||||
TAOS_CHECK_GOTO(taosThreadMutexInit(&pTsdb->rCache.writeBatchMutex, NULL), &lino, _err6);
|
||||
|
||||
pTsdb->rCache.writebatch = writebatch;
|
||||
pTsdb->rCache.my_comparator = cmp;
|
||||
|
@ -230,6 +230,9 @@ static int32_t tsdbOpenRocksCache(STsdb *pTsdb) {
|
|||
pTsdb->rCache.readoptions = readoptions;
|
||||
pTsdb->rCache.flushoptions = flushoptions;
|
||||
pTsdb->rCache.db = db;
|
||||
pTsdb->rCache.sver = -1;
|
||||
pTsdb->rCache.suid = -1;
|
||||
pTsdb->rCache.uid = -1;
|
||||
pTsdb->rCache.pTSchema = NULL;
|
||||
|
||||
TAOS_RETURN(code);
|
||||
|
@ -1132,19 +1135,17 @@ static int32_t tsdbCacheUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, SArray
|
|||
|
||||
(void)taosThreadMutexLock(&pTsdb->lruMutex);
|
||||
for (int i = 0; i < num_keys; ++i) {
|
||||
SLastUpdateCtx *updCtx = (SLastUpdateCtx *)taosArrayGet(updCtxArray, i);
|
||||
|
||||
int8_t lflag = updCtx->lflag;
|
||||
SRowKey *pRowKey = &updCtx->tsdbRowKey.key;
|
||||
SColVal *pColVal = &updCtx->colVal;
|
||||
SLastUpdateCtx *updCtx = &((SLastUpdateCtx *)TARRAY_DATA(updCtxArray))[i];
|
||||
int8_t lflag = updCtx->lflag;
|
||||
SRowKey *pRowKey = &updCtx->tsdbRowKey.key;
|
||||
SColVal *pColVal = &updCtx->colVal;
|
||||
|
||||
if (lflag == LFLAG_LAST && !COL_VAL_IS_VALUE(pColVal)) {
|
||||
continue;
|
||||
}
|
||||
|
||||
SLastKey *key = &(SLastKey){.lflag = lflag, .uid = uid, .cid = pColVal->cid};
|
||||
size_t klen = ROCKS_KEY_LEN;
|
||||
LRUHandle *h = taosLRUCacheLookup(pCache, key, klen);
|
||||
LRUHandle *h = taosLRUCacheLookup(pCache, key, ROCKS_KEY_LEN);
|
||||
if (h) {
|
||||
SLastCol *pLastCol = (SLastCol *)taosLRUCacheValue(pCache, h);
|
||||
if (pLastCol->cacheStatus != TSDB_LAST_CACHE_NO_CACHE) {
|
||||
|
@ -1299,53 +1300,94 @@ _exit:
|
|||
TAOS_RETURN(code);
|
||||
}
|
||||
|
||||
void tsdbCacheInvalidateSchema(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, int32_t sver) {
|
||||
SRocksCache *pRCache = &pTsdb->rCache;
|
||||
if (!pRCache->pTSchema || sver <= pTsdb->rCache.sver) return;
|
||||
|
||||
if (suid > 0 && suid == pRCache->suid) {
|
||||
pRCache->sver = -1;
|
||||
pRCache->suid = -1;
|
||||
}
|
||||
if (suid == 0 && uid == pRCache->uid) {
|
||||
pRCache->sver = -1;
|
||||
pRCache->uid = -1;
|
||||
}
|
||||
}
|
||||
|
||||
static int32_t tsdbUpdateSkm(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, int32_t sver) {
|
||||
SRocksCache *pRCache = &pTsdb->rCache;
|
||||
if (pRCache->pTSchema && sver == pRCache->sver) {
|
||||
if (suid > 0 && suid == pRCache->suid) {
|
||||
return 0;
|
||||
}
|
||||
if (suid == 0 && uid == pRCache->uid) {
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
|
||||
pRCache->suid = suid;
|
||||
pRCache->uid = uid;
|
||||
pRCache->sver = sver;
|
||||
tDestroyTSchema(pRCache->pTSchema);
|
||||
return metaGetTbTSchemaEx(pTsdb->pVnode->pMeta, suid, uid, -1, &pRCache->pTSchema);
|
||||
}
|
||||
|
||||
int32_t tsdbCacheRowFormatUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, int64_t version, int32_t nRow,
|
||||
SRow **aRow) {
|
||||
int32_t code = 0, lino = 0;
|
||||
|
||||
// 1. prepare last
|
||||
TSDBROW lRow = {.type = TSDBROW_ROW_FMT, .pTSRow = aRow[nRow - 1], .version = version};
|
||||
|
||||
TSDBROW lRow = {.type = TSDBROW_ROW_FMT, .pTSRow = aRow[nRow - 1], .version = version};
|
||||
STSchema *pTSchema = NULL;
|
||||
int32_t sver = TSDBROW_SVERSION(&lRow);
|
||||
SArray *ctxArray = NULL;
|
||||
SSHashObj *iColHash = NULL;
|
||||
|
||||
TAOS_CHECK_GOTO(metaGetTbTSchemaEx(pTsdb->pVnode->pMeta, suid, uid, sver, &pTSchema), &lino, _exit);
|
||||
TAOS_CHECK_GOTO(tsdbUpdateSkm(pTsdb, suid, uid, sver), &lino, _exit);
|
||||
pTSchema = pTsdb->rCache.pTSchema;
|
||||
|
||||
TSDBROW tRow = {.type = TSDBROW_ROW_FMT, .version = version};
|
||||
int32_t nCol = pTSchema->numOfCols;
|
||||
|
||||
ctxArray = taosArrayInit(nCol, sizeof(SLastUpdateCtx));
|
||||
iColHash = tSimpleHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT));
|
||||
ctxArray = taosArrayInit(nCol * 2, sizeof(SLastUpdateCtx));
|
||||
if (ctxArray == NULL) {
|
||||
TAOS_CHECK_GOTO(TSDB_CODE_OUT_OF_MEMORY, &lino, _exit);
|
||||
}
|
||||
|
||||
// 1. prepare by lrow
|
||||
STsdbRowKey tsdbRowKey = {0};
|
||||
tsdbRowGetKey(&lRow, &tsdbRowKey);
|
||||
|
||||
STSDBRowIter iter = {0};
|
||||
code = tsdbRowIterOpen(&iter, &lRow, pTSchema);
|
||||
if (code != TSDB_CODE_SUCCESS) {
|
||||
tsdbError("vgId:%d, %s tsdbRowIterOpen failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
|
||||
tstrerror(code));
|
||||
TAOS_CHECK_GOTO(code, &lino, _exit);
|
||||
}
|
||||
TAOS_CHECK_GOTO(tsdbRowIterOpen(&iter, &lRow, pTSchema), &lino, _exit);
|
||||
|
||||
int32_t iCol = 0;
|
||||
for (SColVal *pColVal = tsdbRowIterNext(&iter); pColVal && iCol < nCol; pColVal = tsdbRowIterNext(&iter), iCol++) {
|
||||
SLastUpdateCtx updateCtx = {.lflag = LFLAG_LAST_ROW, .tsdbRowKey = tsdbRowKey, .colVal = *pColVal};
|
||||
if (!taosArrayPush(ctxArray, &updateCtx)) {
|
||||
tsdbRowClose(&iter);
|
||||
TAOS_CHECK_GOTO(terrno, &lino, _exit);
|
||||
}
|
||||
|
||||
if (!COL_VAL_IS_VALUE(pColVal)) {
|
||||
if (COL_VAL_IS_VALUE(pColVal)) {
|
||||
updateCtx.lflag = LFLAG_LAST;
|
||||
if (!taosArrayPush(ctxArray, &updateCtx)) {
|
||||
tsdbRowClose(&iter);
|
||||
TAOS_CHECK_GOTO(terrno, &lino, _exit);
|
||||
}
|
||||
} else {
|
||||
if (!iColHash) {
|
||||
iColHash = tSimpleHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT));
|
||||
if (iColHash == NULL) {
|
||||
tsdbRowClose(&iter);
|
||||
TAOS_CHECK_GOTO(TSDB_CODE_OUT_OF_MEMORY, &lino, _exit);
|
||||
}
|
||||
}
|
||||
|
||||
if (tSimpleHashPut(iColHash, &iCol, sizeof(iCol), NULL, 0)) {
|
||||
tsdbRowClose(&iter);
|
||||
TAOS_CHECK_GOTO(TSDB_CODE_OUT_OF_MEMORY, &lino, _exit);
|
||||
}
|
||||
continue;
|
||||
}
|
||||
updateCtx.lflag = LFLAG_LAST;
|
||||
if (!taosArrayPush(ctxArray, &updateCtx)) {
|
||||
TAOS_CHECK_GOTO(terrno, &lino, _exit);
|
||||
}
|
||||
}
|
||||
tsdbRowClose(&iter);
|
||||
|
@ -1390,7 +1432,10 @@ int32_t tsdbCacheRowFormatUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, int6
|
|||
}
|
||||
|
||||
_exit:
|
||||
taosMemoryFreeClear(pTSchema);
|
||||
if (code) {
|
||||
tsdbError("vgId:%d, %s failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__, tstrerror(code));
|
||||
}
|
||||
|
||||
taosArrayDestroy(ctxArray);
|
||||
tSimpleHashCleanup(iColHash);
|
||||
|
||||
|
|
|
@ -14,12 +14,12 @@
|
|||
*/
|
||||
|
||||
#define _DEFAULT_SOURCE
|
||||
#include "tlrucache.h"
|
||||
#include "os.h"
|
||||
#include "taoserror.h"
|
||||
#include "tarray.h"
|
||||
#include "tdef.h"
|
||||
#include "tlog.h"
|
||||
#include "tlrucache.h"
|
||||
#include "tutil.h"
|
||||
|
||||
typedef struct SLRUEntry SLRUEntry;
|
||||
|
@ -110,7 +110,7 @@ struct SLRUEntryTable {
|
|||
};
|
||||
|
||||
static int taosLRUEntryTableInit(SLRUEntryTable *table, int maxUpperHashBits) {
|
||||
table->lengthBits = 4;
|
||||
table->lengthBits = 16;
|
||||
table->list = taosMemoryCalloc(1 << table->lengthBits, sizeof(SLRUEntry *));
|
||||
if (!table->list) {
|
||||
TAOS_RETURN(terrno);
|
||||
|
@ -371,24 +371,35 @@ static void taosLRUCacheShardCleanup(SLRUCacheShard *shard) {
|
|||
|
||||
static LRUStatus taosLRUCacheShardInsertEntry(SLRUCacheShard *shard, SLRUEntry *e, LRUHandle **handle,
|
||||
bool freeOnFail) {
|
||||
LRUStatus status = TAOS_LRU_STATUS_OK;
|
||||
SArray *lastReferenceList = taosArrayInit(16, POINTER_BYTES);
|
||||
if (!lastReferenceList) {
|
||||
taosLRUEntryFree(e);
|
||||
return TAOS_LRU_STATUS_FAIL;
|
||||
LRUStatus status = TAOS_LRU_STATUS_OK;
|
||||
SLRUEntry *toFree = NULL;
|
||||
SArray *lastReferenceList = NULL;
|
||||
if (shard->usage + e->totalCharge > shard->capacity) {
|
||||
lastReferenceList = taosArrayInit(16, POINTER_BYTES);
|
||||
if (!lastReferenceList) {
|
||||
taosLRUEntryFree(e);
|
||||
return TAOS_LRU_STATUS_FAIL;
|
||||
}
|
||||
}
|
||||
|
||||
(void)taosThreadMutexLock(&shard->mutex);
|
||||
|
||||
taosLRUCacheShardEvictLRU(shard, e->totalCharge, lastReferenceList);
|
||||
if (shard->usage + e->totalCharge > shard->capacity && shard->lru.next != &shard->lru) {
|
||||
if (!lastReferenceList) {
|
||||
lastReferenceList = taosArrayInit(16, POINTER_BYTES);
|
||||
if (!lastReferenceList) {
|
||||
taosLRUEntryFree(e);
|
||||
(void)taosThreadMutexUnlock(&shard->mutex);
|
||||
return TAOS_LRU_STATUS_FAIL;
|
||||
}
|
||||
}
|
||||
taosLRUCacheShardEvictLRU(shard, e->totalCharge, lastReferenceList);
|
||||
}
|
||||
|
||||
if (shard->usage + e->totalCharge > shard->capacity && (shard->strictCapacity || handle == NULL)) {
|
||||
TAOS_LRU_ENTRY_SET_IN_CACHE(e, false);
|
||||
if (handle == NULL) {
|
||||
if (!taosArrayPush(lastReferenceList, &e)) {
|
||||
taosLRUEntryFree(e);
|
||||
goto _exit;
|
||||
}
|
||||
toFree = e;
|
||||
} else {
|
||||
if (freeOnFail) {
|
||||
taosLRUEntryFree(e);
|
||||
|
@ -413,11 +424,7 @@ static LRUStatus taosLRUCacheShardInsertEntry(SLRUCacheShard *shard, SLRUEntry *
|
|||
taosLRUCacheShardLRURemove(shard, old);
|
||||
shard->usage -= old->totalCharge;
|
||||
|
||||
if (!taosArrayPush(lastReferenceList, &old)) {
|
||||
taosLRUEntryFree(e);
|
||||
taosLRUEntryFree(old);
|
||||
goto _exit;
|
||||
}
|
||||
toFree = old;
|
||||
}
|
||||
}
|
||||
if (handle == NULL) {
|
||||
|
@ -434,6 +441,10 @@ static LRUStatus taosLRUCacheShardInsertEntry(SLRUCacheShard *shard, SLRUEntry *
|
|||
_exit:
|
||||
(void)taosThreadMutexUnlock(&shard->mutex);
|
||||
|
||||
if (toFree) {
|
||||
taosLRUEntryFree(toFree);
|
||||
}
|
||||
|
||||
for (int i = 0; i < taosArrayGetSize(lastReferenceList); ++i) {
|
||||
SLRUEntry *entry = taosArrayGetP(lastReferenceList, i);
|
||||
|
||||
|
@ -733,7 +744,8 @@ void taosLRUCacheCleanup(SLRUCache *cache) {
|
|||
}
|
||||
|
||||
LRUStatus taosLRUCacheInsert(SLRUCache *cache, const void *key, size_t keyLen, void *value, size_t charge,
|
||||
_taos_lru_deleter_t deleter, _taos_lru_overwriter_t overwriter, 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;
|
||||
|
||||
|
|
Loading…
Reference in New Issue