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

<feeature>[raft]refactor raft interface,add log store methods

This commit is contained in:
lichuang 2021-10-28 09:56:20 +08:00
parent eddbd36c9e
commit 4845ca7f99
1 changed files with 40 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

50
include/libs/sync/sync.h
View File

@ -22,7 +22,6 @@ extern "C" {
#include <stdint.h> #include <stdint.h>
#include "taosdef.h" #include "taosdef.h"
#include "wal.h"
typedef int64_t SyncNodeId; typedef int64_t SyncNodeId;
typedef int32_t SyncGroupId; typedef int32_t SyncGroupId;
@ -41,6 +40,7 @@ typedef struct {
} SSyncBuffer; } SSyncBuffer;
typedef struct { typedef struct {
SyncNodeId nodeId;
uint16_t nodePort; // node sync Port uint16_t nodePort; // node sync Port
char nodeFqdn[TSDB_FQDN_LEN]; // node FQDN char nodeFqdn[TSDB_FQDN_LEN]; // node FQDN
} SNodeInfo; } SNodeInfo;
@ -83,11 +83,38 @@ typedef struct SSyncFSM {
} SSyncFSM; } SSyncFSM;
typedef struct SSyncLogStore {
void* pData;
// write log with given index
int32_t (*logWrite)(struct SSyncLogStore* logStore, SyncIndex index, SSyncBuffer* pBuf);
// mark log with given index has been commtted
int32_t (*logCommit)(struct SSyncLogStore* logStore, SyncIndex index);
// prune log before given index
int32_t (*logPrune)(struct SSyncLogStore* logStore, SyncIndex index);
// rollback log after given index
int32_t (*logRollback)(struct SSyncLogStore* logStore, SyncIndex index);
} SSyncLogStore;
typedef struct SSyncServerState { typedef struct SSyncServerState {
SNodeInfo voteFor; SyncNodeId voteFor;
SSyncTerm term; SSyncTerm term;
} SSyncServerState; } SSyncServerState;
typedef struct SSyncClusterConfig {
// Log index number of current cluster config.
SyncIndex index;
// Log index number of previous cluster config.
SyncIndex prevIndex;
// current cluster
const SSyncCluster* cluster;
} SSyncClusterConfig;
typedef struct SStateManager { typedef struct SStateManager {
void* pData; void* pData;
@ -95,35 +122,38 @@ typedef struct SStateManager {
const SSyncServerState* (*readServerState)(struct SStateManager* stateMng); const SSyncServerState* (*readServerState)(struct SStateManager* stateMng);
void (*saveCluster)(struct SStateManager* stateMng, const SSyncCluster* cluster); void (*saveCluster)(struct SStateManager* stateMng, const SSyncClusterConfig* cluster);
const SSyncCluster* (*readCluster)(struct SStateManager* stateMng); const SSyncClusterConfig* (*readCluster)(struct SStateManager* stateMng);
} SStateManager; } SStateManager;
typedef struct { typedef struct {
SyncGroupId vgId; SyncGroupId vgId;
twalh walHandle;
SyncIndex snapshotIndex; SyncIndex snapshotIndex;
SSyncCluster syncCfg; SSyncCluster syncCfg;
SSyncFSM fsm; SSyncFSM fsm;
SSyncLogStore logStore;
SStateManager stateManager; SStateManager stateManager;
} SSyncInfo; } SSyncInfo;
struct SSyncNode;
typedef struct SSyncNode SSyncNode;
int32_t syncInit(); int32_t syncInit();
void syncCleanUp(); void syncCleanUp();
SyncNodeId syncStart(const SSyncInfo*); SSyncNode syncStart(const SSyncInfo*);
void syncStop(SyncNodeId); void syncStop(SyncNodeId);
int32_t syncPropose(SyncNodeId nodeId, SSyncBuffer buffer, void* pData, bool isWeak); int32_t syncPropose(SSyncNode syncNode, SSyncBuffer buffer, void* pData, bool isWeak);
int32_t syncAddNode(SyncNodeId nodeId, const SNodeInfo *pNode); int32_t syncAddNode(SSyncNode syncNode, const SNodeInfo *pNode);
int32_t syncRemoveNode(SyncNodeId nodeId, const SNodeInfo *pNode); int32_t syncRemoveNode(SSyncNode syncNode, const SNodeInfo *pNode);
extern int32_t syncDebugFlag; extern int32_t syncDebugFlag;