Merge branch '3.0' into feature/3.0_liaohj

include/client/consumer/consumer.h

@@ -16,6 +16,10 @@
 #ifndef _TD_CONSUMER_H_
 #define _TD_CONSUMER_H_
 
+#include "tlist.h"
+#include "tarray.h"
+#include "hash.h"
+
 #ifdef __cplusplus
 extern "C" {
 #endif
@@ -32,16 +36,15 @@ extern "C" {
   struct tmq_resp_err_t;
   typedef struct tmq_resp_err_t tmq_resp_err_t;
 
-  //topic list
+  struct tmq_message_t;
-  //resouces are supposed to be free by users by calling tmq_list_destroy
+  typedef struct tmq_message_t tmq_message_t;
-  struct tmq_topic_list_t;
+
-  typedef struct tmq_topic_list_t tmq_topic_list_t;
+  struct tmq_col_batch_t;
-  int32_t tmq_topic_list_add(tmq_topic_list_t*, const char*);
+  typedef struct tmq_col_batch_t tmq_col_batch_t;
-  void    tmq_topic_list_destroy(tmq_topic_list_t*);
 
   //get content of message
-  tmq_col_batch_t *tmq_get_msg_col_by_idx(tmq_message_t*, int32_t);
+  tmq_col_batch_t* tmq_get_msg_col_by_idx(tmq_message_t*, int32_t col_id);
-  tmq_col_batch_t *tmq_get_msg_col_by_name(tmq_message_t*, const char*);
+  tmq_col_batch_t* tmq_get_msg_col_by_name(tmq_message_t*, const char*);
 
   //consumer config
   int32_t tmq_conf_set(tmq_consumer_config_t* , const char* config_key, const char* config_value, char* errstr, int32_t errstr_cap);
@@ -51,11 +54,12 @@ extern "C" {
   tmq_consumer_t* tmq_consumer_new(tmq_consumer_config_t* , char* errstr, int32_t errstr_cap);
 
   //subscribe
-  tmq_resp_err_t tmq_subscribe(tmq_consumer_t*, const tmq_topic_list_t*);
+  tmq_resp_err_t tmq_subscribe(tmq_consumer_t*, const SList*);
+  tmq_resp_err_t tmq_unsubscribe(tmq_consumer_t*);
 
   //consume
   //resouces are supposed to be free by users by calling tmq_message_destroy
-  tmq_message_t tmq_consume_poll(tmq_consumer_t*, int64_t blocking_time);
+  tmq_message_t* tmq_consume_poll(tmq_consumer_t*, int64_t blocking_time);
 
   //destroy message and free memory
   void tmq_message_destroy(tmq_message_t*);

include/common/taosMsg.h

@@ -1,55 +0,0 @@
-/*
- * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
- *
- * This program is free software: you can use, redistribute, and/or modify
- * it under the terms of the GNU Affero General Public License, version 3
- * or later ("AGPL"), as published by the Free Software Foundation.
- *
- * This program is distributed in the hope that it will be useful, but WITHOUT
- * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
- * FITNESS FOR A PARTICULAR PURPOSE.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program. If not, see <http://www.gnu.org/licenses/>.
- */
-
-#ifndef _TD_TAOS_MSG_H_
-#define _TD_TAOS_MSG_H_
-
-typedef struct {
-  /* data */
-} SSubmitReq;
-
-typedef struct {
-  /* data */
-} SSubmitRsp;
-
-typedef struct {
-  /* data */
-} SSubmitReqReader;
-
-typedef struct {
-  /* data */
-} SCreateTableReq;
-
-typedef struct {
-  /* data */
-} SCreateTableRsp;
-
-typedef struct {
-  /* data */
-} SDropTableReq;
-
-typedef struct {
-  /* data */
-} SDropTableRsp;
-
-typedef struct {
-  /* data */
-} SAlterTableReq;
-
-typedef struct {
-  /* data */
-} SAlterTableRsp;
-
-#endif /*_TD_TAOS_MSG_H_*/

include/common/taosmsg.h

@@ -952,6 +952,42 @@ typedef struct {
   char    reserved2[64];
 } SStartupStep;
 
+typedef struct {
+  /* data */
+} SSubmitReq;
+
+typedef struct {
+  /* data */
+} SSubmitRsp;
+
+typedef struct {
+  /* data */
+} SSubmitReqReader;
+
+typedef struct {
+  /* data */
+} SCreateTableReq;
+
+typedef struct {
+  /* data */
+} SCreateTableRsp;
+
+typedef struct {
+  /* data */
+} SDropTableReq;
+
+typedef struct {
+  /* data */
+} SDropTableRsp;
+
+typedef struct {
+  /* data */
+} SAlterTableReq;
+
+typedef struct {
+  /* data */
+} SAlterTableRsp;
+
 #pragma pack(pop)
 
 #ifdef __cplusplus

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);
+int64_t  walWrite(twalh, int8_t msgType, void* body, uint32_t bodyLen);
-void     walWaitFsync(twalh, bool forceHint);
+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/qnode/qnode.h

@@ -20,6 +20,94 @@
 extern "C" {
 #endif
 
+
+typedef struct {
+  uint64_t numOfStartTask;
+  uint64_t numOfStopTask;
+  uint64_t numOfRecvedFetch;
+  uint64_t numOfSentHb;
+  uint64_t numOfSentFetch;
+  uint64_t numOfTaskInQueue;
+  uint64_t numOfFetchInQueue;
+  uint64_t numOfErrors;
+} SQnodeStat;
+
+/* start Task msg */
+typedef struct {
+  uint32_t  schedulerIp;
+  uint16_t  schedulerPort;
+  int64_t   taskId;
+  int64_t   queryId;
+  uint32_t  srcIp;
+  uint16_t  srcPort;
+} SQnodeStartTaskMsg;
+
+/* stop Task msg */
+typedef struct {
+  int64_t   taskId;
+} SQnodeStopTaskMsg;
+
+/* start/stop Task msg response */
+typedef struct {
+  int64_t   taskId;
+  int32_t code;
+} SQnodeTaskRespMsg;
+
+/* Task status msg */
+typedef struct {
+  int64_t   taskId;
+  int32_t   status;
+  int64_t   queryId;
+} SQnodeTaskStatusMsg;
+
+/* Qnode/Scheduler heartbeat msg */
+typedef struct {
+  int32_t status;
+  int32_t load;
+  
+} SQnodeHeartbeatMsg;
+
+/* Qnode sent/received msg */
+typedef struct {
+  int8_t   msgType;
+  int32_t  msgLen;
+  char     msg[];
+} SQnodeMsg;
+
+
+/**
+ * Start one Qnode in Dnode.
+ * @return Error Code.
+ */
+int32_t qnodeStart();
+
+/**
+ * Stop Qnode in Dnode.
+ *
+ * @param qnodeId Qnode ID to stop, -1 for all Qnodes.
+ */
+void qnodeStop(int64_t qnodeId);
+
+ 
+/**
+ * Get the statistical information of Qnode
+ *
+ * @param qnodeId Qnode ID to get statistics, -1 for all 
+ * @param stat Statistical information.
+ * @return Error Code.
+ */
+int32_t qnodeGetStatistics(int64_t qnodeId, SQnodeStat *stat);
+
+/**
+ * Interface for processing Qnode messages.
+ * 
+ * @param pMsg Message to be processed.
+ * @return Error code
+ */
+void qnodeProcessReq(SRpcMsg *pMsg);
+
+
+
 #ifdef __cplusplus
 }
 #endif

include/server/vnode/meta/meta.h

@@ -16,7 +16,7 @@
 #ifndef _TD_META_H_
 #define _TD_META_H_
 
-#include "taosMsg.h"
+#include "taosmsg.h"
 
 #ifdef __cplusplus
 extern "C" {

include/server/vnode/tq/tq.h

@@ -16,14 +16,32 @@
 #ifndef _TD_TQ_H_
 #define _TD_TQ_H_
 
+#include "os.h"
+
 #ifdef __cplusplus
 extern "C" {
 #endif
 
 typedef struct STQ STQ;
 
-int tqPushMsg(void *);
+STQ* tqInit(void* ref_func(void*), void* unref_func(void*));
-int tqCommit(STQ *pTQ);
+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*);
+//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 tqHandleMsg(STQ*, void *msg);
 
 #ifdef __cplusplus
 }

include/server/vnode/tsdb/tsdb.h

@@ -17,7 +17,7 @@
 #define _TD_TSDB_H_
 
 #include "os.h"
-#include "taosMsg.h"
+#include "taosmsg.h"
 
 #ifdef __cplusplus
 extern "C" {

source/client/consumer/consumer.c

@@ -12,3 +12,5 @@
  * You should have received a copy of the GNU Affero General Public License
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
+
+#include "consumer.h"

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"
+    PRIVATE "${CMAKE_SOURCE_DIR}/include/os"
+)
+
+target_link_libraries(
+    os
 )

source/server/vnode/tq/CMakeLists.txt

@@ -4,4 +4,9 @@ target_include_directories(
     tq
     PUBLIC "${CMAKE_SOURCE_DIR}/include/server/vnode/tq"
     PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
+    PRIVATE "${CMAKE_SOURCE_DIR}/include/os"
+)
+
+target_link_libraries(
+    os
 )

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

@@ -16,10 +16,15 @@
 #ifndef _TD_TQ_INT_H_
 #define _TD_TQ_INT_H_
 
+#include "tq.h"
+
 #ifdef __cplusplus
 extern "C" {
 #endif
 
+//implement the array index
+//implement the ring buffer
+
 #ifdef __cplusplus
 }
 #endif

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

@@ -13,7 +13,21 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#include "tq.h"
+#include "tqInt.h"
 
-int tqPushMsg(void * p) {return 0;}
+//static
-int tqCommit(STQ *pTQ) {return 0;}
+//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) {
+  return 0;
+}