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

merge 3.0

This commit is contained in:
yihaoDeng 2023-11-08 10:04:46 +08:00
parent 0b4946eefb
commit 594da2d603
2 changed files with 60 additions and 51 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

60
source/libs/stream/inc/streamBackendRocksdb.h
View File

@ -66,7 +66,7 @@ typedef struct {
TdThreadMutex mutex; TdThreadMutex mutex;
char* idstr; char* idstr;
char* path; char* path;
int64_t refId; int64_t refId;
void* pTask; void* pTask;
int64_t streamId; int64_t streamId;
@ -79,6 +79,53 @@ typedef struct {
} STaskDbWrapper; } STaskDbWrapper;
typedef struct SDbChkp {
int8_t init;
char* pCurrent;
char* pManifest;
SArray* pSST;
int64_t preCkptId;
int64_t curChkpId;
char* path;
char* buf;
int32_t len;
// ping-pong buf
SHashObj* pSstTbl[2];
int8_t idx;
SArray* pAdd;
SArray* pDel;
int8_t update;
TdThreadRwlock rwLock;
} SDbChkp;
typedef struct {
int8_t init;
char* pCurrent;
char* pManifest;
SArray* pSST;
int64_t preCkptId;
int64_t curChkpId;
char* path;
char* buf;
int32_t len;
// ping-pong buf
SHashObj* pSstTbl[2];
int8_t idx;
SArray* pAdd;
SArray* pDel;
int8_t update;
SHashObj* pDbChkpTbl;
TdThreadRwlock rwLock;
} SBkdMgt;
void* streamBackendInit(const char* path, int64_t chkpId); void* streamBackendInit(const char* path, int64_t chkpId);
void streamBackendCleanup(void* arg); void streamBackendCleanup(void* arg);
void streamBackendHandleCleanup(void* arg); void streamBackendHandleCleanup(void* arg);
@ -194,4 +241,15 @@ int32_t streamBackendDelInUseChkp(void* arg, int64_t chkpId);
int32_t taskDbBuildSnap(void* arg, SArray* pSnap); int32_t taskDbBuildSnap(void* arg, SArray* pSnap);
// int32_t streamDefaultIter_rocksdb(SStreamState* pState, const void* start, const void* end, SArray* result); // int32_t streamDefaultIter_rocksdb(SStreamState* pState, const void* start, const void* end, SArray* result);
STaskDbWrapper* taskDbOpen(char* path, char* key, int64_t chkpId);
void taskDbDestroy(void* pDb);
int32_t taskDbDoCheckpoint(void* arg, int64_t chkpId);
SBkdMgt* bkdMgtCreate(char* path);
int32_t bkdMgtAddChkp(SBkdMgt* bm, char* task, char* path);
int32_t bkdMgtGetDelta(SBkdMgt* bm, char* taskId, int64_t chkpId, SArray* list);
int32_t bkdMgtDumpTo(SBkdMgt* bm, char* taskId, char* dname);
void bkdMgtDestroy(SBkdMgt* bm);
#endif #endif

51
source/libs/stream/src/streamBackendRocksdb.c
View File

@ -20,53 +20,6 @@
#include "tcommon.h" #include "tcommon.h"
#include "tref.h" #include "tref.h"
typedef struct SDbChkp {
int8_t init;
char* pCurrent;
char* pManifest;
SArray* pSST;
int64_t preCkptId;
int64_t curChkpId;
char* path;
char* buf;
int32_t len;
// ping-pong buf
SHashObj* pSstTbl[2];
int8_t idx;
SArray* pAdd;
SArray* pDel;
int8_t update;
TdThreadRwlock rwLock;
} SDbChkp;
typedef struct {
int8_t init;
char* pCurrent;
char* pManifest;
SArray* pSST;
int64_t preCkptId;
int64_t curChkpId;
char* path;
char* buf;
int32_t len;
// ping-pong buf
SHashObj* pSstTbl[2];
int8_t idx;
SArray* pAdd;
SArray* pDel;
int8_t update;
SHashObj* pDbChkpTbl;
TdThreadRwlock rwLock;
} SBkdMgt;
typedef struct SCompactFilteFactory { typedef struct SCompactFilteFactory {
void* status; void* status;
} SCompactFilteFactory; } SCompactFilteFactory;
@ -223,8 +176,6 @@ SCfInit ginitDict[] = {
valueEncode, valueDecode, compactFilteFactoryCreateFilter, destroyCompactFilteFactory, compactFilteFactoryName}, valueEncode, valueDecode, compactFilteFactoryCreateFilter, destroyCompactFilteFactory, compactFilteFactoryName},
}; };
const char* cfName[] = {"default", "state", "fill", "sess", "func", "parname", "partag"};
int32_t getCfIdx(const char* cfName) { int32_t getCfIdx(const char* cfName) {
int idx = -1; int idx = -1;
size_t len = strlen(cfName); size_t len = strlen(cfName);
@ -868,7 +819,6 @@ int32_t taskDbBuildSnap(void* arg, SArray* pSnap) {
code = taskDbDoCheckpoint(pTaskDb, pTaskDb->chkpId); code = taskDbDoCheckpoint(pTaskDb, pTaskDb->chkpId);
taskDbRemoveRef(pTaskDb); taskDbRemoveRef(pTaskDb);
pIter = taosHashIterate(pMeta->pTaskDbUnique, pIter);
SStreamTask* pTask = pTaskDb->pTask; SStreamTask* pTask = pTaskDb->pTask;
SStreamTaskSnap snap = {.streamId = pTask->id.streamId, SStreamTaskSnap snap = {.streamId = pTask->id.streamId,
@ -876,6 +826,7 @@ int32_t taskDbBuildSnap(void* arg, SArray* pSnap) {
.chkpId = pTaskDb->chkpId, .chkpId = pTaskDb->chkpId,
.dbPrefixPath = taosStrdup(pTaskDb->path)}; .dbPrefixPath = taosStrdup(pTaskDb->path)};
taosArrayPush(pSnap, &snap); taosArrayPush(pSnap, &snap);
pIter = taosHashIterate(pMeta->pTaskDbUnique, pIter);
} }
return code; return code;
} }