add tq header

include/libs/wal/wal.h

@@ -15,13 +15,16 @@
 #ifndef _TD_WAL_H_
 #define _TD_WAL_H_
 
+#include "os.h"
+
 #ifdef __cplusplus
 extern "C" {
 #endif
 
 typedef enum {
   TAOS_WAL_NOLOG = 0,
-  TAOS_WAL_WRITE = 1
+  TAOS_WAL_WRITE = 1,
+  TAOS_WAL_FSYNC = 2
 } EWalType;
 
 typedef struct {
@@ -55,8 +58,8 @@ void     walStop(twalh);
 void     walClose(twalh);
 
 //write
-int32_t  walWrite(twalh, int8_t msgType, void* body, uint32_t bodyLen);
-void     walWaitFsync(twalh, bool forceHint);
+int64_t  walWrite(twalh, int8_t msgType, void* body, uint32_t bodyLen);
+void     walFsync(twalh, bool forceHint);
 //int32_t  walCommit(twalh, uint64_t ver);
 //int32_t  walRollback(twalh, uint64_t ver);
 
@@ -67,7 +70,7 @@ int32_t  walReadWithFp(twalh, FWalWrite writeFp, int64_t verStart, int readNum);
 //life cycle
 int32_t  walDataPersisted(twalh, int64_t ver);
 int32_t  walFirstVer(twalh);
-int32_t  walLastVer(twal);
+int32_t  walLastVer(twalh);
 //int32_t  walDataCorrupted(twalh);
 
 #ifdef __cplusplus

include/server/vnode/tq/tq.h

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

source/libs/wal/CMakeLists.txt

@@ -4,4 +4,9 @@ target_include_directories(
     wal
     PUBLIC "${CMAKE_SOURCE_DIR}/include/libs/wal"
     PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
-)
\ No newline at end of file
+    PRIVATE "${CMAKE_SOURCE_DIR}/include/os"
+)
+
+target_link_libraries(
+    os
+)

source/server/vnode/tq/CMakeLists.txt

@@ -3,8 +3,8 @@ add_library(tq ${TQ_SRC})
 target_include_directories(
     tq
     PUBLIC "${CMAKE_SOURCE_DIR}/include/server/vnode/tq"
-    PRIVATE "${CMAKE_SOURCE_DIR}/include/os"
     PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
+    PRIVATE "${CMAKE_SOURCE_DIR}/include/os"
 )
 
 target_link_libraries(

source/server/vnode/tq/inc/tqInt.h

@@ -16,6 +16,8 @@
 #ifndef _TD_TQ_INT_H_
 #define _TD_TQ_INT_H_
 
+#include "tq.h"
+
 #ifdef __cplusplus
 extern "C" {
 #endif

source/server/vnode/tq/src/tq.c

@@ -13,14 +13,21 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#include "tq.h"
+#include "tqInt.h"
 
-int tqPushMsg(STQ *pTQ, void * p, int64_t version) {
+//static
+//read next version data
+//
+//send to fetch queue
+//
+//handle management message
+
+int tqPushMsg(STQ* pTq , void* p, int64_t version) {
   //add reference
   //
   return 0;
 }
 
-int tqCommit(STQ *pTQ) {
+int tqCommit(STQ* pTq) {
   return 0;
 }