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

refactor tq

This commit is contained in:
Liu Jicong 2021-10-09 17:14:37 +08:00
parent ada087b002
commit 4a883c23c5
3 changed files with 18 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
include/libs/wal/wal.h
View File

@ -60,8 +60,8 @@ void walClose(twalh);
//write //write
int64_t walWrite(twalh, int8_t msgType, void* body, uint32_t bodyLen); int64_t walWrite(twalh, int8_t msgType, void* body, uint32_t bodyLen);
void walFsync(twalh, bool forceHint); void walFsync(twalh, bool forceHint);
//int32_t walCommit(twalh, uint64_t ver); //int32_t walCommit(twalh, int64_t ver);
//int32_t walRollback(twalh, uint64_t ver); //int32_t walRollback(twalh, int64_t ver);
//read //read
int32_t walRead(twalh, SWalHead **, int64_t ver); int32_t walRead(twalh, SWalHead **, int64_t ver);

19
include/server/vnode/tq/tq.h
View File

@ -25,23 +25,14 @@ extern "C" {
typedef struct STQ STQ; typedef struct STQ STQ;
STQ* tqInit(void* ref_func(void*), void* unref_func(void*)); STQ* tqInit(void* ref_func(void*), void* unref_func(void*));
void tqCleanUp(STQ* pTq); void tqCleanUp(STQ*);
//create persistent storage for meta info such as consuming offset //void* will be replace by a msg type
//return value > 0: cgId int tqPushMsg(STQ*, void* msg, int64_t version);
//return value <= 0: error code
int tqCreateGroup(STQ*);
//create ring buffer in memory and load consuming offset
int tqOpenGroup(STQ*, int cgId);
//destroy ring buffer and persist consuming offset
int tqCloseGroup(STQ*, int cgId);
//delete persistent storage for meta info
int tqDropGroup(STQ*, int cgId);
int tqPushMsg(STQ*, void *, int64_t version);
int tqCommit(STQ*); int tqCommit(STQ*);
int tqHandleMsg(STQ*, void *msg); //void* will be replace by a msg type
int tqHandleMsg(STQ*, void* msg);
#ifdef __cplusplus #ifdef __cplusplus
} }

11
source/server/vnode/tq/inc/tqInt.h
View File

@ -25,6 +25,17 @@ extern "C" {
//implement the array index //implement the array index
//implement the ring buffer //implement the ring buffer
//create persistent storage for meta info such as consuming offset
//return value > 0: cgId
//return value <= 0: error code
int tqCreateGroup(STQ*);
//create ring buffer in memory and load consuming offset
int tqOpenGroup(STQ*, int cgId);
//destroy ring buffer and persist consuming offset
int tqCloseGroup(STQ*, int cgId);
//delete persistent storage for meta info
int tqDropGroup(STQ*, int cgId);
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif