From ec4ea9cf5d6f3456de2a2bb77b3bafa810d766d4 Mon Sep 17 00:00:00 2001 From: Minglei Jin Date: Fri, 14 Apr 2023 18:28:15 +0800 Subject: [PATCH] enh(cache/rocks): base for rocks put/get --- contrib/CMakeLists.txt | 1 + contrib/test/CMakeLists.txt | 2 + contrib/test/rocksdb/CMakeLists.txt | 4 +- contrib/test/rocksdb/main.c | 8 +- source/dnode/vnode/CMakeLists.txt | 2 + source/dnode/vnode/src/inc/tsdb.h | 10 ++ source/dnode/vnode/src/inc/vnodeInt.h | 1 + source/dnode/vnode/src/tsdb/tsdbCache.c | 111 +++++++++++++++++++++ source/dnode/vnode/src/tsdb/tsdbMemTable.c | 26 +++-- 9 files changed, 150 insertions(+), 15 deletions(-) diff --git a/contrib/CMakeLists.txt b/contrib/CMakeLists.txt index 8f94313382..6cd6f79d8b 100644 --- a/contrib/CMakeLists.txt +++ b/contrib/CMakeLists.txt @@ -438,6 +438,7 @@ endif(${BUILD_ADDR2LINE}) # ================================================================================================ # Build test # ================================================================================================ +message("contrib tests:" ${BUILD_DEPENDENCY_TESTS}) if(${BUILD_DEPENDENCY_TESTS}) add_subdirectory(test EXCLUDE_FROM_ALL) endif(${BUILD_DEPENDENCY_TESTS}) diff --git a/contrib/test/CMakeLists.txt b/contrib/test/CMakeLists.txt index f35cf0d13d..911eca50a7 100644 --- a/contrib/test/CMakeLists.txt +++ b/contrib/test/CMakeLists.txt @@ -1,4 +1,6 @@ # rocksdb +message("contrib test dir:" ${BUILD_WITH_ROCKSDB}) + if(${BUILD_WITH_ROCKSDB}) add_subdirectory(rocksdb) endif(${BUILD_WITH_ROCKSDB}) diff --git a/contrib/test/rocksdb/CMakeLists.txt b/contrib/test/rocksdb/CMakeLists.txt index b500e0a661..5ae1b0395b 100644 --- a/contrib/test/rocksdb/CMakeLists.txt +++ b/contrib/test/rocksdb/CMakeLists.txt @@ -1,6 +1,8 @@ +message("contrib test/rocksdb:" ${BUILD_DEPENDENCY_TESTS}) + add_executable(rocksdbTest "") target_sources(rocksdbTest PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/main.c" ) -target_link_libraries(rocksdbTest rocksdb) \ No newline at end of file +target_link_libraries(rocksdbTest rocksdb) diff --git a/contrib/test/rocksdb/main.c b/contrib/test/rocksdb/main.c index d1cbd373da..09435790cc 100644 --- a/contrib/test/rocksdb/main.c +++ b/contrib/test/rocksdb/main.c @@ -25,10 +25,12 @@ int main(int argc, char const *argv[]) { // Read rocksdb_readoptions_t *readoptions = rocksdb_readoptions_create(); - rocksdb_readoptions_set_snapshot(readoptions, rocksdb_create_snapshot(db)); +//rocksdb_readoptions_set_snapshot(readoptions, rocksdb_create_snapshot(db)); + char buf[256] = {0}; size_t vallen = 0; char * val = rocksdb_get(db, readoptions, "key", 3, &vallen, &err); - printf("val:%s\n", val); + snprintf(buf, vallen+5, "val:%s", val); + printf("%ld %ld %s\n", strlen(val), vallen, buf); // Update // rocksdb_put(db, writeoptions, "key", 3, "eulav", 5, &err); @@ -43,4 +45,4 @@ int main(int argc, char const *argv[]) { rocksdb_close(db); return 0; -} \ No newline at end of file +} diff --git a/source/dnode/vnode/CMakeLists.txt b/source/dnode/vnode/CMakeLists.txt index 9911752f8e..52df20877a 100644 --- a/source/dnode/vnode/CMakeLists.txt +++ b/source/dnode/vnode/CMakeLists.txt @@ -82,6 +82,7 @@ target_include_directories( PUBLIC "inc" PUBLIC "src/inc" PUBLIC "${TD_SOURCE_DIR}/include/libs/scalar" + PUBLIC "${TD_SOURCE_DIR}/contrib/rocksdb/include" ) target_link_libraries( vnode @@ -98,6 +99,7 @@ target_link_libraries( # PUBLIC bdb # PUBLIC scalar + PUBLIC rocksdb PUBLIC transport PUBLIC stream PUBLIC index diff --git a/source/dnode/vnode/src/inc/tsdb.h b/source/dnode/vnode/src/inc/tsdb.h index 2a85b191a4..b698909af1 100644 --- a/source/dnode/vnode/src/inc/tsdb.h +++ b/source/dnode/vnode/src/inc/tsdb.h @@ -343,6 +343,14 @@ struct STsdbFS { SArray *aDFileSet; // SArray }; +typedef struct { + rocksdb_t *db; + rocksdb_options_t *options; + rocksdb_writeoptions_t *writeoptions; + rocksdb_readoptions_t *readoptions; + TdThreadMutex rMutex; +} SRocksCache; + struct STsdb { char *path; SVnode *pVnode; @@ -355,6 +363,7 @@ struct STsdb { TdThreadMutex lruMutex; SLRUCache *biCache; TdThreadMutex biMutex; + SRocksCache rCache; }; struct TSDBKEY { @@ -796,6 +805,7 @@ typedef struct { int32_t tsdbOpenCache(STsdb *pTsdb); void tsdbCloseCache(STsdb *pTsdb); +int32_t tsdbCacheUpdate(STsdb *pTsdb, tb_uid_t uid, TSDBROW *row); int32_t tsdbCacheInsertLast(SLRUCache *pCache, tb_uid_t uid, TSDBROW *row, STsdb *pTsdb); int32_t tsdbCacheInsertLastrow(SLRUCache *pCache, STsdb *pTsdb, tb_uid_t uid, TSDBROW *row, bool dup); int32_t tsdbCacheGetLastH(SLRUCache *pCache, tb_uid_t uid, SCacheRowsReader *pr, LRUHandle **h); diff --git a/source/dnode/vnode/src/inc/vnodeInt.h b/source/dnode/vnode/src/inc/vnodeInt.h index 253d5aebce..a3f2c4559f 100644 --- a/source/dnode/vnode/src/inc/vnodeInt.h +++ b/source/dnode/vnode/src/inc/vnodeInt.h @@ -19,6 +19,7 @@ #include "executor.h" #include "filter.h" #include "qworker.h" +#include "rocksdb/c.h" #include "sync.h" #include "tRealloc.h" #include "tchecksum.h" diff --git a/source/dnode/vnode/src/tsdb/tsdbCache.c b/source/dnode/vnode/src/tsdb/tsdbCache.c index 48d3371284..171a63aa33 100644 --- a/source/dnode/vnode/src/tsdb/tsdbCache.c +++ b/source/dnode/vnode/src/tsdb/tsdbCache.c @@ -43,6 +43,110 @@ static void tsdbCloseBICache(STsdb *pTsdb) { } } +static void tsdbGetRocksPath(STsdb *pTsdb, char *path) { + SVnode *pVnode = pTsdb->pVnode; + if (pVnode->pTfs) { + if (path) { + snprintf(path, TSDB_FILENAME_LEN, "%s%s%s%scache.rdb", tfsGetPrimaryPath(pTsdb->pVnode->pTfs), TD_DIRSEP, + pTsdb->path, TD_DIRSEP); + } + } else { + if (path) { + snprintf(path, TSDB_FILENAME_LEN, "%s%scache.rdb", pTsdb->path, TD_DIRSEP); + } + } +} + +static int32_t tsdbOpenRocksCache(STsdb *pTsdb) { + int32_t code = 0; + + rocksdb_options_t *options = rocksdb_options_create(); + if (NULL == options) { + code = TSDB_CODE_OUT_OF_MEMORY; + goto _err; + } + + rocksdb_options_set_create_if_missing(options, 1); + rocksdb_options_set_inplace_update_support(options, 1); + rocksdb_options_set_allow_concurrent_memtable_write(options, 0); + + rocksdb_writeoptions_t *writeoptions = rocksdb_writeoptions_create(); + if (NULL == writeoptions) { + code = TSDB_CODE_OUT_OF_MEMORY; + goto _err2; + } + + rocksdb_readoptions_t *readoptions = rocksdb_readoptions_create(); + if (NULL == readoptions) { + code = TSDB_CODE_OUT_OF_MEMORY; + goto _err2; + } + + char *err = NULL; + char cachePath[TSDB_FILENAME_LEN] = {0}; + tsdbGetRocksPath(pTsdb, cachePath); + + rocksdb_t *db = rocksdb_open(options, cachePath, &err); + if (NULL == db) { + code = -1; + goto _err3; + } + + taosThreadMutexInit(&pTsdb->rCache.rMutex, NULL); + + pTsdb->rCache.options = options; + pTsdb->rCache.writeoptions = writeoptions; + pTsdb->rCache.readoptions = readoptions; + pTsdb->rCache.db = db; + + return code; + +_err4: + rocksdb_readoptions_destroy(readoptions); +_err3: + rocksdb_writeoptions_destroy(writeoptions); +_err2: + rocksdb_options_destroy(options); +_err: + return code; +} + +static void tsdbCloseRocksCache(STsdb *pTsdb) { + rocksdb_close(pTsdb->rCache.db); + rocksdb_readoptions_destroy(pTsdb->rCache.readoptions); + rocksdb_writeoptions_destroy(pTsdb->rCache.writeoptions); + rocksdb_options_destroy(pTsdb->rCache.options); +} + +int32_t tsdbCacheUpdate(STsdb *pTsdb, tb_uid_t uid, TSDBROW *pRow) { + int32_t code = 0; + + STSDBRowIter iter = {0}; + tsdbRowIterOpen(&iter, pRow, pTSchema); + + for (SColVal *pColVal = tsdbRowIterNext(&iter); pColVal; pColVal = tsdbRowIterNext(&iter)) { + SColVal *pRColVal = tsdbCacheGetRColVal(pTsdb); + if (pRColVal) { + // merge pColVal with pRColVal + } + + tsdbCachePutRColVal(pColVal); + } + + tsdbRowClose(&iter); + + char *err = NULL; + char buf[256] = {0}; + size_t vallen = 0; + char *val = rocksdb_get(pTsdb->rCache.db, pTsdb->rCache.readoptions, "key", 3, &vallen, &err); + if (val) { + } else { + } + rocksdb_put(pTsdb->rCache.db, pTsdb->rCache.writeoptions, "key", 3, "value", 5, &err); + + return code; +} + int32_t tsdbOpenCache(STsdb *pTsdb) { int32_t code = 0; SLRUCache *pCache = NULL; @@ -60,6 +164,12 @@ int32_t tsdbOpenCache(STsdb *pTsdb) { goto _err; } + code = tsdbOpenRocksCache(pTsdb); + if (code != TSDB_CODE_SUCCESS) { + code = TSDB_CODE_OUT_OF_MEMORY; + goto _err; + } + taosLRUCacheSetStrictCapacity(pCache, false); taosThreadMutexInit(&pTsdb->lruMutex, NULL); @@ -80,6 +190,7 @@ void tsdbCloseCache(STsdb *pTsdb) { } tsdbCloseBICache(pTsdb); + tsdbCloseRocksCache(pTsdb); } static void getTableCacheKey(tb_uid_t uid, int cacheType, char *key, int *len) { diff --git a/source/dnode/vnode/src/tsdb/tsdbMemTable.c b/source/dnode/vnode/src/tsdb/tsdbMemTable.c index d0ff403bf7..f480fb6081 100644 --- a/source/dnode/vnode/src/tsdb/tsdbMemTable.c +++ b/source/dnode/vnode/src/tsdb/tsdbMemTable.c @@ -284,8 +284,8 @@ bool tsdbTbDataIterNext(STbDataIter *pIter) { int64_t tsdbCountTbDataRows(STbData *pTbData) { SMemSkipListNode *pNode = pTbData->sl.pHead; - int64_t rowsNum = 0; - + int64_t rowsNum = 0; + while (NULL != pNode) { pNode = SL_GET_NODE_FORWARD(pNode, 0); if (pNode == pTbData->sl.pTail) { @@ -298,17 +298,17 @@ int64_t tsdbCountTbDataRows(STbData *pTbData) { return rowsNum; } -void tsdbMemTableCountRows(SMemTable *pMemTable, SHashObj* pTableMap, int64_t *rowsNum) { +void tsdbMemTableCountRows(SMemTable *pMemTable, SHashObj *pTableMap, int64_t *rowsNum) { taosRLockLatch(&pMemTable->latch); for (int32_t i = 0; i < pMemTable->nBucket; ++i) { STbData *pTbData = pMemTable->aBucket[i]; while (pTbData) { - void* p = taosHashGet(pTableMap, &pTbData->uid, sizeof(pTbData->uid)); + void *p = taosHashGet(pTableMap, &pTbData->uid, sizeof(pTbData->uid)); if (p == NULL) { pTbData = pTbData->next; continue; } - + *rowsNum += tsdbCountTbDataRows(pTbData); pTbData = pTbData->next; } @@ -668,15 +668,17 @@ static int32_t tsdbInsertColDataToTable(SMemTable *pMemTable, STbData *pTbData, if (key.ts >= pTbData->maxKey) { pTbData->maxKey = key.ts; - + /* if (TSDB_CACHE_LAST_ROW(pMemTable->pTsdb->pVnode->config)) { tsdbCacheInsertLastrow(pMemTable->pTsdb->lruCache, pMemTable->pTsdb, pTbData->uid, &lRow, true); - } + }*/ } - + /* if (TSDB_CACHE_LAST(pMemTable->pTsdb->pVnode->config)) { tsdbCacheInsertLast(pMemTable->pTsdb->lruCache, pTbData->uid, &lRow, pMemTable->pTsdb); } + */ + tsdbCacheUpdate(pMemTable->pTsdb, pTbData->uid, &lRow); // SMemTable pMemTable->minKey = TMIN(pMemTable->minKey, pTbData->minKey); @@ -736,15 +738,17 @@ static int32_t tsdbInsertRowDataToTable(SMemTable *pMemTable, STbData *pTbData, if (key.ts >= pTbData->maxKey) { pTbData->maxKey = key.ts; - + /* if (TSDB_CACHE_LAST_ROW(pMemTable->pTsdb->pVnode->config)) { tsdbCacheInsertLastrow(pMemTable->pTsdb->lruCache, pMemTable->pTsdb, pTbData->uid, &lRow, true); - } + }*/ } - + /* if (TSDB_CACHE_LAST(pMemTable->pTsdb->pVnode->config)) { tsdbCacheInsertLast(pMemTable->pTsdb->lruCache, pTbData->uid, &lRow, pMemTable->pTsdb); } + */ + tsdbCacheUpdate(pMemTable->pTsdb, pTbData->uid, &lRow); // SMemTable pMemTable->minKey = TMIN(pMemTable->minKey, pTbData->minKey); -- GitLab