Merge branch '3.0' into 3.0_refact

include/libs/raft/raft.h

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ * Copyright (c) 2019 TAOS Data, Inc. <cli@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
@@ -13,15 +13,149 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#ifndef _TD_RAFT_H_
+#ifndef TD_RAFT_H
-#define _TD_RAFT_H_
+#define TD_RAFT_H
 
 #ifdef __cplusplus
 extern "C" {
 #endif
 
+#include <stdio.h>
+#include "taosdef.h"
+
+typedef unsigned int RaftId;
+typedef unsigned int RaftGroupId;
+
+// buffer holding data
+typedef struct RaftBuffer {
+  void* data;
+  size_t len;
+} RaftBuffer;
+
+// a single server information in a cluster
+typedef struct RaftServer {
+  RaftId    id;
+  char      fqdn[TSDB_FQDN_LEN];
+  uint16_t  port;
+} RaftServer;
+
+// all servers in a cluster
+typedef struct RaftConfiguration {
+  RaftServer *servers;
+  int nServer;
+} RaftConfiguration;
+
+// raft lib module
+struct Raft;
+typedef struct Raft Raft;
+
+struct RaftNode;
+typedef struct RaftNode RaftNode;
+
+// raft state machine
+struct RaftFSM;
+typedef struct RaftFSM {
+  // statemachine user data
+  void *data;
+
+  // apply buffer data, bufs will be free by raft module
+  int (*apply)(struct RaftFSM *fsm, const RaftBuffer *bufs[], int nBufs);
+
+  // configuration commit callback 
+  int (*onConfigurationCommit)(const RaftConfiguration* cluster);
+
+  // fsm return snapshot in ppBuf, bufs will be free by raft module
+  // TODO: getSnapshot SHOULD be async?
+  int (*getSnapshot)(struct RaftFSM *fsm, RaftBuffer **ppBuf);
+
+  // fsm restore with pBuf data
+  int (*restore)(struct RaftFSM *fsm, RaftBuffer *pBuf);
+
+  // fsm send data in buf to server,buf will be free by raft module
+  int (*send)(struct RaftFSM* fsm, const RaftServer* server, const RaftBuffer *buf);
+} RaftFSM;
+
+typedef struct RaftNodeOptions {
+  // user define state machine
+  RaftFSM* pFSM;
+
+  // election timeout(in ms)
+  // by default: 1000
+  int electionTimeoutMS;
+
+  // heart timeout(in ms)
+  // by default: 100
+  int heartbeatTimeoutMS;
+
+  // install snapshot timeout(in ms)
+  int installSnapshotTimeoutMS;
+
+  /** 
+   * number of log entries before starting a new snapshot. 
+   * by default: 1024
+   */
+  int snapshotThreshold;
+
+  /**
+   * Number of log entries to keep in the log after a snapshot has
+   * been taken. 
+   * by default: 128.
+   */
+  int snapshotTrailing;
+
+  /**
+   * Enable or disable pre-vote support.
+   * by default: false
+   */
+  bool preVote;
+
+} RaftNodeOptions;
+
+// create raft lib
+int RaftCreate(Raft** ppRaft);
+
+int RaftDestroy(Raft* pRaft);
+
+// start a raft node with options,node id,group id
+int RaftStart(Raft* pRaft, 
+              RaftId selfId,
+              RaftGroupId selfGroupId,
+              const RaftConfiguration* cluster,
+              const RaftNodeOptions* options,              
+              RaftNode **ppNode);
+
+// stop a raft node
+int RaftStop(RaftNode* pNode);
+
+// client apply a cmd in buf
+typedef void (*RaftApplyFp)(const RaftBuffer *pBuf, int result);
+
+int RaftApply(RaftNode *pNode,
+              const RaftBuffer *pBuf,
+              RaftApplyFp applyCb);
+
+// recv data from other servers in cluster,buf will be free in raft
+int RaftRecv(RaftNode *pNode, const RaftBuffer* pBuf);
+
+// change cluster servers API
+typedef void (*RaftChangeFp)(const RaftServer* pServer, int result);
+
+int RaftAddServer(RaftNode *pNode,
+                  const RaftServer* pServer,
+                  RaftChangeFp changeCb);
+
+int RaftRemoveServer(RaftNode *pNode, 
+                    const RaftServer* pServer,
+                    RaftChangeFp changeCb);
+
+// transfer leader to id
+typedef void (*RaftTransferFp)(RaftId id, int result);
+int RaftTransfer(RaftNode *pNode,
+                 RaftId id,
+                 RaftTransferFp transferCb);
+
 #ifdef __cplusplus
 }
 #endif
 
-#endif /*_TD_RAFT_H_*/
+#endif  /* TD_RAFT_H */ 

include/libs/wal/wal.h

@@ -60,8 +60,8 @@ void     walClose(twalh);
 //write
 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  walCommit(twalh, int64_t ver);
-//int32_t  walRollback(twalh, uint64_t ver);
+//int32_t  walRollback(twalh, int64_t ver);
 
 //read
 int32_t  walRead(twalh, SWalHead **, int64_t ver);

include/server/vnode/tq/tq.h

@@ -25,22 +25,13 @@ extern "C" {
 typedef struct STQ STQ;
 
 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*);
 
+//void* will be replace by a msg type
 int tqHandleMsg(STQ*, void* msg);
 
 #ifdef __cplusplus

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

@@ -25,6 +25,17 @@ extern "C" {
 //implement the array index
 //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
 }
 #endif

src/raft/CMakeLists.txt

@@ -0,0 +1,7 @@
+aux_source_directory(source RAFT_SRC)
+add_library(raft ${RAFT_SRC})
+target_include_directories(
+    raft
+    PUBLIC "${CMAKE_SOURCE_DIR}/include/raft"
+    PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/include"
+)