Skip to content

TDengine
TDengine

taosdata / TDengine
大约 2 年 前同步成功

通知 1193
Star 22018
Fork 4786
TDengine
TDengine
提交 afc6023a 编写于 作者: dengyihao's avatar dengyihao
浏览文件
操作

merge rocksdb inst

上级 c4f36d32
显示空白变更内容
内联 并排
Showing with 121 addition and 107 deletion
include/libs/stream/streamState.h
浏览文件 @ afc6023a
...@@ -56,6 +56,7 @@ typedef struct STdbState { ...@@ -56,6 +56,7 @@ typedef struct STdbState {
void* env; void* env;
SListNode* pComparNode; SListNode* pComparNode;
SBackendHandle* pBackendHandle; SBackendHandle* pBackendHandle;
char idstr[48];
TDB* db; TDB* db;
TTB* pStateDb; TTB* pStateDb;
......
source/libs/stream/src/streamStateRocksdb.c
浏览文件 @ afc6023a
...@@ -324,7 +324,7 @@ int32_t streaValueIsStale(void* k, int64_t ts) { ...@@ -324,7 +324,7 @@ int32_t streaValueIsStale(void* k, int64_t ts) {
typedef struct { typedef struct {
void* tableOpt; void* tableOpt;
} rocksdbCfParam; } RocksdbCfParam;
const char* cfName[] = {"default", "state", "fill", "sess", "func", "parname", "partag"}; const char* cfName[] = {"default", "state", "fill", "sess", "func", "parname", "partag"};
typedef int (*EncodeFunc)(void* key, char* buf); typedef int (*EncodeFunc)(void* key, char* buf);
...@@ -357,8 +357,7 @@ typedef struct { ...@@ -357,8 +357,7 @@ typedef struct {
} SCfInit; } SCfInit;
#define GEN_COLUMN_FAMILY_NAME(name, streamId, taskId, SUBFIX) \ #define GEN_COLUMN_FAMILY_NAME(name, idstr, SUBFIX) sprintf(name, "%s_%s", idstr, (SUBFIX));
sprintf(name, "%d_%d_%s", (streamId), (taskId), (SUBFIX));
SCfInit ginitDict[] = { SCfInit ginitDict[] = {
{"default", 7, 0, defaultKeyComp, defaultKeyEncode, defaultKeyDecode, defaultKeyToString, compareDefaultName, {"default", 7, 0, defaultKeyComp, defaultKeyEncode, defaultKeyDecode, defaultKeyToString, compareDefaultName,
...@@ -384,10 +383,11 @@ int streamStateOpenBackend(void* backend, SStreamState* pState) { ...@@ -384,10 +383,11 @@ int streamStateOpenBackend(void* backend, SStreamState* pState) {
qInfo("start to open backend, %p, %d-%d", pState, pState->streamId, pState->taskId); qInfo("start to open backend, %p, %d-%d", pState, pState->streamId, pState->taskId);
SBackendHandle* handle = backend; SBackendHandle* handle = backend;
sprintf(pState->pTdbState->idstr, "%d-%d", pState->streamId, pState->taskId);
char* err = NULL; char* err = NULL;
int cfLen = sizeof(ginitDict) / sizeof(ginitDict[0]); int cfLen = sizeof(ginitDict) / sizeof(ginitDict[0]);
rocksdbCfParam* param = taosMemoryCalloc(cfLen, sizeof(rocksdbCfParam)); RocksdbCfParam* param = taosMemoryCalloc(cfLen, sizeof(RocksdbCfParam));
const rocksdb_options_t** cfOpt = taosMemoryCalloc(cfLen, sizeof(rocksdb_options_t*)); const rocksdb_options_t** cfOpt = taosMemoryCalloc(cfLen, sizeof(rocksdb_options_t*));
for (int i = 0; i < cfLen; i++) { for (int i = 0; i < cfLen; i++) {
cfOpt[i] = rocksdb_options_create(); cfOpt[i] = rocksdb_options_create();
...@@ -414,12 +414,12 @@ int streamStateOpenBackend(void* backend, SStreamState* pState) { ...@@ -414,12 +414,12 @@ int streamStateOpenBackend(void* backend, SStreamState* pState) {
rocksdb_column_family_handle_t** cfHandle = taosMemoryMalloc(cfLen * sizeof(rocksdb_column_family_handle_t*)); rocksdb_column_family_handle_t** cfHandle = taosMemoryMalloc(cfLen * sizeof(rocksdb_column_family_handle_t*));
for (int i = 0; i < cfLen; i++) { for (int i = 0; i < cfLen; i++) {
char buf[64] = {0}; char buf[64] = {0};
GEN_COLUMN_FAMILY_NAME(buf, pState->streamId, pState->taskId, ginitDict[i].key); GEN_COLUMN_FAMILY_NAME(buf, pState->pTdbState->idstr, ginitDict[i].key);
cfHandle[i] = rocksdb_create_column_family(handle->db, cfOpt[i], buf, &err); cfHandle[i] = rocksdb_create_column_family(handle->db, cfOpt[i], buf, &err);
if (err != NULL) { if (err != NULL) {
qError("rocksdb create column family failed, reason:%s", err); qError("failed to create cf:%s_%s, reason:%s", pState->pTdbState->idstr, ginitDict[i].key, err);
taosMemoryFree(err); taosMemoryFreeClear(err);
return -1; // return -1;
} }
} }
...@@ -428,47 +428,54 @@ int streamStateOpenBackend(void* backend, SStreamState* pState) { ...@@ -428,47 +428,54 @@ int streamStateOpenBackend(void* backend, SStreamState* pState) {
pState->pTdbState->writeOpts = rocksdb_writeoptions_create(); pState->pTdbState->writeOpts = rocksdb_writeoptions_create();
pState->pTdbState->readOpts = rocksdb_readoptions_create(); pState->pTdbState->readOpts = rocksdb_readoptions_create();
pState->pTdbState->cfOpts = (rocksdb_options_t**)cfOpt; pState->pTdbState->cfOpts = (rocksdb_options_t**)cfOpt;
// pState->pTdbState->pCompare = pCompare;
pState->pTdbState->dbOpt = handle->dbOpt; pState->pTdbState->dbOpt = handle->dbOpt;
pState->pTdbState->param = param; pState->pTdbState->param = param;
pState->pTdbState->pBackendHandle = handle; pState->pTdbState->pBackendHandle = handle;
SCfComparator compare = {.comp = pCompare, .numOfComp = cfLen}; SCfComparator compare = {.comp = pCompare, .numOfComp = cfLen};
pState->pTdbState->pComparNode = streamBackendAddCompare(handle, &compare); pState->pTdbState->pComparNode = streamBackendAddCompare(handle, &compare);
rocksdb_writeoptions_disable_WAL(pState->pTdbState->writeOpts, 1); rocksdb_writeoptions_disable_WAL(pState->pTdbState->writeOpts, 1);
return 0; return 0;
} }
void streamStateCloseBackend(SStreamState* pState, bool remove) { void streamStateCloseBackend(SStreamState* pState, bool remove) {
char* status[] = {"remove", "drop"}; char* status[] = {"close", "drop"};
qInfo("start to %s backend, %p, %d-%d", status[remove == false ? 1 : 0], pState, pState->streamId, pState->taskId); qInfo("start to %s backend, %p, %d-%d", status[remove == false ? 0 : 1], pState, pState->streamId, pState->taskId);
if (pState->pTdbState->rocksdb == NULL) { if (pState->pTdbState->rocksdb == NULL) {
return; return;
} }
int cfLen = sizeof(ginitDict) / sizeof(ginitDict[0]); int cfLen = sizeof(ginitDict) / sizeof(ginitDict[0]);
rocksdbCfParam* param = pState->pTdbState->param;
char* err = NULL; char* err = NULL;
rocksdb_flushoptions_t* flushOpt = rocksdb_flushoptions_create();
for (int i = 0; i < cfLen; i++) {
if (remove) { if (remove) {
for (int i = 0; i < cfLen; i++) {
rocksdb_drop_column_family(pState->pTdbState->rocksdb, pState->pTdbState->pHandle[i], &err); rocksdb_drop_column_family(pState->pTdbState->rocksdb, pState->pTdbState->pHandle[i], &err);
if (err != NULL) {
qError("failed to create cf:%s_%s, reason:%s", pState->pTdbState->idstr, ginitDict[i].key, err);
taosMemoryFreeClear(err);
}
}
} else { } else {
rocksdb_flushoptions_t* flushOpt = rocksdb_flushoptions_create();
for (int i = 0; i < cfLen; i++) {
rocksdb_flush_cf(pState->pTdbState->rocksdb, flushOpt, pState->pTdbState->pHandle[i], &err); rocksdb_flush_cf(pState->pTdbState->rocksdb, flushOpt, pState->pTdbState->pHandle[i], &err);
if (err != NULL) {
qError("failed to create cf:%s_%s, reason:%s", pState->pTdbState->idstr, ginitDict[i].key, err);
taosMemoryFreeClear(err);
} }
} }
rocksdb_flushoptions_destroy(flushOpt); rocksdb_flushoptions_destroy(flushOpt);
}
for (int i = 0; i < cfLen; i++) { for (int i = 0; i < cfLen; i++) {
rocksdb_column_family_handle_destroy(pState->pTdbState->pHandle[i]); rocksdb_column_family_handle_destroy(pState->pTdbState->pHandle[i]);
} }
taosMemoryFreeClear(pState->pTdbState->pHandle); taosMemoryFreeClear(pState->pTdbState->pHandle);
for (int i = 0; i < cfLen; i++) { for (int i = 0; i < cfLen; i++) {
rocksdb_options_destroy(pState->pTdbState->cfOpts[i]); rocksdb_options_destroy(pState->pTdbState->cfOpts[i]);
rocksdb_block_based_options_destroy(param[i].tableOpt); rocksdb_block_based_options_destroy(((RocksdbCfParam*)pState->pTdbState->param)[i].tableOpt);
} }
if (remove) { if (remove) {
streamBackendDelCompare(pState->pTdbState->pBackendHandle, pState->pTdbState->pComparNode); streamBackendDelCompare(pState->pTdbState->pBackendHandle, pState->pTdbState->pComparNode);
} }
...@@ -518,7 +525,7 @@ rocksdb_iterator_t* streamStateIterCreate(SStreamState* pState, const char* cfNa ...@@ -518,7 +525,7 @@ rocksdb_iterator_t* streamStateIterCreate(SStreamState* pState, const char* cfNa
rocksdb_readoptions_t* rOpt = rocksdb_readoptions_create(); rocksdb_readoptions_t* rOpt = rocksdb_readoptions_create();
*readOpt = rOpt; *readOpt = rOpt;
// rocksdb_readoptions_set_snapshot(rOpt, *snapshot); rocksdb_readoptions_set_snapshot(rOpt, *snapshot);
rocksdb_readoptions_set_fill_cache(rOpt, 0); rocksdb_readoptions_set_fill_cache(rOpt, 0);
...@@ -532,7 +539,7 @@ rocksdb_iterator_t* streamStateIterCreate(SStreamState* pState, const char* cfNa ...@@ -532,7 +539,7 @@ rocksdb_iterator_t* streamStateIterCreate(SStreamState* pState, const char* cfNa
char* err = NULL; \ char* err = NULL; \
int i = streamGetInit(funcname); \ int i = streamGetInit(funcname); \
if (i < 0) { \ if (i < 0) { \
qWarn("streamState failed to get cf name: %s", funcname); \ qWarn("streamState failed to get cf name: %s_%s", pState->pTdbState->idstr, funcname); \
code = -1; \ code = -1; \
break; \ break; \
} \ } \
...@@ -545,10 +552,12 @@ rocksdb_iterator_t* streamStateIterCreate(SStreamState* pState, const char* cfNa ...@@ -545,10 +552,12 @@ rocksdb_iterator_t* streamStateIterCreate(SStreamState* pState, const char* cfNa
rocksdb_put_cf(db, opts, pHandle, (const char*)buf, klen, (const char*)value, (size_t)vLen, &err); \ rocksdb_put_cf(db, opts, pHandle, (const char*)buf, klen, (const char*)value, (size_t)vLen, &err); \
if (err != NULL) { \ if (err != NULL) { \
taosMemoryFree(err); \ taosMemoryFree(err); \
qDebug("streamState str: %s failed to write to %s, err: %s", toString, funcname, err); \ qDebug("streamState str: %s failed to write to %s_%s, err: %s", toString, pState->pTdbState->idstr, funcname, \
err); \
code = -1; \ code = -1; \
} else { \ } else { \
qDebug("streamState str:%s succ to write to %s, valLen:%d", toString, funcname, vLen); \ qDebug("streamState str:%s succ to write to %s_%s, valLen:%d", toString, pState->pTdbState->idstr, funcname, \
vLen); \
} \ } \
} while (0); } while (0);
...@@ -559,7 +568,7 @@ rocksdb_iterator_t* streamStateIterCreate(SStreamState* pState, const char* cfNa ...@@ -559,7 +568,7 @@ rocksdb_iterator_t* streamStateIterCreate(SStreamState* pState, const char* cfNa
char* err = NULL; \ char* err = NULL; \
int i = streamGetInit(funcname); \ int i = streamGetInit(funcname); \
if (i < 0) { \ if (i < 0) { \
qWarn("streamState failed to get cf name: %s", funcname); \ qWarn("streamState failed to get cf name: %s_%s", pState->pTdbState->idstr, funcname); \
code = -1; \ code = -1; \
break; \ break; \
} \ } \
...@@ -572,7 +581,8 @@ rocksdb_iterator_t* streamStateIterCreate(SStreamState* pState, const char* cfNa ...@@ -572,7 +581,8 @@ rocksdb_iterator_t* streamStateIterCreate(SStreamState* pState, const char* cfNa
size_t len = 0; \ size_t len = 0; \
char* val = rocksdb_get_cf(db, opts, pHandle, (const char*)buf, klen, (size_t*)&len, &err); \ char* val = rocksdb_get_cf(db, opts, pHandle, (const char*)buf, klen, (size_t*)&len, &err); \
if (val == NULL) { \ if (val == NULL) { \
qDebug("streamState str: %s failed to read from %s, err: not exist", toString, funcname); \ qDebug("streamState str: %s failed to read from %s_%s, err: not exist", toString, pState->pTdbState->idstr, \
funcname); \
if (err != NULL) taosMemoryFree(err); \ if (err != NULL) taosMemoryFree(err); \
code = -1; \ code = -1; \
} else { \ } else { \
...@@ -581,10 +591,12 @@ rocksdb_iterator_t* streamStateIterCreate(SStreamState* pState, const char* cfNa ...@@ -581,10 +591,12 @@ rocksdb_iterator_t* streamStateIterCreate(SStreamState* pState, const char* cfNa
} \ } \
if (err != NULL) { \ if (err != NULL) { \
taosMemoryFree(err); \ taosMemoryFree(err); \
qDebug("streamState str: %s failed to read from %s, err: %s", toString, funcname, err); \ qDebug("streamState str: %s failed to read from %s_%s, err: %s", toString, pState->pTdbState->idstr, funcname, \
err); \
code = -1; \ code = -1; \
} else { \ } else { \
if (code == 0) qDebug("streamState str: %s succ to read from %s", toString, funcname); \ if (code == 0) \
qDebug("streamState str: %s succ to read from %s_%s", toString, pState->pTdbState->idstr, funcname); \
} \ } \
} while (0); } while (0);
...@@ -595,7 +607,7 @@ rocksdb_iterator_t* streamStateIterCreate(SStreamState* pState, const char* cfNa ...@@ -595,7 +607,7 @@ rocksdb_iterator_t* streamStateIterCreate(SStreamState* pState, const char* cfNa
char* err = NULL; \ char* err = NULL; \
int i = streamGetInit(funcname); \ int i = streamGetInit(funcname); \
if (i < 0) { \ if (i < 0) { \
qWarn("streamState failed to get cf name: %s", funcname); \ qWarn("streamState failed to get cf name: %s_%s", pState->pTdbState->idstr, funcname); \
code = -1; \ code = -1; \
break; \ break; \
} \ } \
...@@ -607,11 +619,12 @@ rocksdb_iterator_t* streamStateIterCreate(SStreamState* pState, const char* cfNa ...@@ -607,11 +619,12 @@ rocksdb_iterator_t* streamStateIterCreate(SStreamState* pState, const char* cfNa
rocksdb_writeoptions_t* opts = pState->pTdbState->writeOpts; \ rocksdb_writeoptions_t* opts = pState->pTdbState->writeOpts; \
rocksdb_delete_cf(db, opts, pHandle, (const char*)buf, klen, &err); \ rocksdb_delete_cf(db, opts, pHandle, (const char*)buf, klen, &err); \
if (err != NULL) { \ if (err != NULL) { \
qDebug("streamState str: %s failed to del from %s, err: %s", toString, funcname, err); \ qError("streamState str: %s failed to del from %s_%s, err: %s", toString, pState->pTdbState->idstr, funcname, \
err); \
taosMemoryFree(err); \ taosMemoryFree(err); \
code = -1; \ code = -1; \
} else { \ } else { \
qDebug("streamState str: %s succ to del from %s", toString, funcname); \ qDebug("streamState str: %s succ to del from %s_%s", toString, pState->pTdbState->idstr, funcname); \
} \ } \
} while (0); } while (0);
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或