Merge pull request #8394 from taosdata/feature/raft-interface

[feeature][raft]add raft interface

include/libs/raft/raft.h

-/*
- * 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
- * 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_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 */ 
\ No newline at end of file

include/libs/sync/sync.h

+/*
+ * 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
+ * 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 TDENGINE_RAFT_SYNC_H
+#define TDENGINE_RAFT_SYNC_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include <stdint.h>
+#include "taosdef.h"
+#include "wal.h"
+
+typedef uint32_t SyncNodeId;
+typedef int32_t SyncGroupId;
+typedef int64_t   SyncIndex;
+typedef uint64_t SSyncTerm;
+
+typedef enum {
+  TAOS_SYNC_ROLE_FOLLOWER  = 0,
+  TAOS_SYNC_ROLE_CANDIDATE = 1,
+  TAOS_SYNC_ROLE_LEADER  = 2,
+} ESyncRole;
+
+typedef struct {
+  void* data;
+  size_t len;
+} SSyncBuffer;
+
+typedef struct {
+  SyncNodeId  nodeId;    // node ID assigned by TDengine
+  uint16_t  nodePort;  // node sync Port
+  char      nodeFqdn[TSDB_FQDN_LEN]; // node FQDN  
+} SNodeInfo;
+
+typedef struct {
+  int selfIndex;
+  int nNode;
+  SNodeInfo*  nodeInfo;
+} SSyncCluster;
+
+typedef struct {
+  int32_t  selfIndex;
+  int nNode;
+  SyncNodeId* nodeId;
+  ESyncRole*  role;
+} SNodesRole;
+
+struct SSyncFSM;
+typedef struct SSyncFSM {
+  void* pData;
+
+  // apply committed log, bufs will be free by raft module
+  int (*applyLog)(struct SSyncFSM *fsm, SyncIndex index, const SSyncBuffer *buf, void *pData);
+
+  // cluster commit callback 
+  int (*onClusterChanged)(struct SSyncFSM *fsm, const SSyncCluster* cluster, void *pData);
+
+  // fsm return snapshot in ppBuf, bufs will be free by raft module
+  // TODO: getSnapshot SHOULD be async?
+  int (*getSnapshot)(struct SSyncFSM *fsm, SSyncBuffer **ppBuf, int* objId, bool *isLast);
+
+  // fsm apply snapshot with pBuf data
+  int (*applySnapshot)(struct SSyncFSM *fsm, SSyncBuffer *pBuf, int objId, bool isLast);
+
+  // call when restore snapshot and log done
+  int (*onRestoreDone)(struct SSyncFSM *fsm);
+
+  void (*onRollback)(struct SSyncFSM *fsm, SyncIndex index, const SSyncBuffer *buf);
+
+  void (*onRoleChanged)(struct SSyncFSM *fsm, const SNodesRole* pRole);
+
+} SSyncFSM;
+
+typedef struct SSyncServerState {
+  SyncNodeId voteFor;
+  SSyncTerm term;  
+} SSyncServerState;
+
+typedef struct SStateManager {
+  void* pData;
+
+  void (*saveServerState)(struct SStateManager* stateMng, const SSyncServerState* state);
+
+  const SSyncServerState* (*readServerState)(struct SStateManager* stateMng);
+
+  void (*saveCluster)(struct SStateManager* stateMng, const SSyncCluster* cluster);
+
+  const SSyncCluster* (*readCluster)(struct SStateManager* stateMng);
+} SStateManager;
+
+typedef struct {
+  SyncGroupId  vgId;
+
+  twalh walHandle;
+
+  SyncIndex snapshotIndex;    // initial version
+  SSyncCluster syncCfg;    // configuration from mgmt
+
+  SSyncFSM fsm;
+
+  SStateManager stateManager;
+} SSyncInfo;
+
+int32_t syncInit();
+void    syncCleanUp();
+
+SyncNodeId syncStart(const SSyncInfo *);
+void    syncStop(SyncNodeId);
+
+int32_t syncPropose(SyncNodeId nodeId, SSyncBuffer buffer, void *pData, bool isWeak);
+
+extern int32_t  raftDebugFlag;
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif  // TDENGINE_RAFT_SYNC_H

source/libs/CMakeLists.txt

 add_subdirectory(transport)
-add_subdirectory(raft)
+add_subdirectory(sync)
 add_subdirectory(tkv)
 add_subdirectory(index)
 add_subdirectory(wal)

source/libs/raft/CMakeLists.txt

-aux_source_directory(src RAFT_SRC)
-add_library(raft ${RAFT_SRC})
-target_include_directories(
-    raft
-    PUBLIC "${CMAKE_SOURCE_DIR}/include/libs/raft"
-    PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
-)
\ No newline at end of file

source/libs/sync/CMakeLists.txt

+aux_source_directory(src SYNC_SRC)
+add_library(sync ${SYNC_SRC})
+
+target_link_libraries(
+    sync
+    PUBLIC common
+    PUBLIC util
+    PUBLIC wal
+)
+
+target_include_directories(
+    sync
+    PUBLIC "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
+)
\ No newline at end of file

source/libs/raft/src/raft.c → source/libs/sync/src/sync.c

@@ -11,4 +11,6 @@
  *
  * 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/>.
- */
\ No newline at end of file
+ */
+
+#include "sync.h"
\ No newline at end of file