sync.h 7.4 KB
Newer Older
S
Shengliang Guan 已提交
1
/*
M
Minghao Li 已提交
2
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
S
Shengliang Guan 已提交
3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22
 *
 * 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_LIBS_SYNC_H
#define _TD_LIBS_SYNC_H

#ifdef __cplusplus
extern "C" {
#endif

M
Minghao Li 已提交
23
#include "cJSON.h"
M
Minghao Li 已提交
24
#include "tdef.h"
S
Shengliang Guan 已提交
25
#include "tmsgcb.h"
S
Shengliang Guan 已提交
26

M
Minghao Li 已提交
27 28
extern bool gRaftDetailLog;

29
#define SYNC_RESP_TTL_MS 10000000
M
Minghao Li 已提交
30

M
Minghao Li 已提交
31
#define SYNC_MAX_BATCH_SIZE 500
32
#define SYNC_INDEX_BEGIN 0
33
#define SYNC_INDEX_INVALID -1
34
#define SYNC_TERM_INVALID 0xFFFFFFFFFFFFFFFF
35

M
Minghao Li 已提交
36 37 38 39 40 41
typedef enum {
  SYNC_STRATEGY_NO_SNAPSHOT = 0,
  SYNC_STRATEGY_STANDARD_SNAPSHOT = 1,
  SYNC_STRATEGY_WAL_FIRST = 2,
} ESyncStrategy;

M
Minghao Li 已提交
42
typedef uint64_t SyncNodeId;
S
Shengliang Guan 已提交
43 44 45 46
typedef int32_t  SyncGroupId;
typedef int64_t  SyncIndex;
typedef uint64_t SyncTerm;

47 48 49 50 51
typedef struct SSyncNode      SSyncNode;
typedef struct SSyncBuffer    SSyncBuffer;
typedef struct SWal           SWal;
typedef struct SSyncRaftEntry SSyncRaftEntry;

S
Shengliang Guan 已提交
52
typedef enum {
M
Minghao Li 已提交
53 54 55
  TAOS_SYNC_STATE_FOLLOWER = 100,
  TAOS_SYNC_STATE_CANDIDATE = 101,
  TAOS_SYNC_STATE_LEADER = 102,
M
Minghao Li 已提交
56
  TAOS_SYNC_STATE_ERROR = 103,
M
syncInt  
Minghao Li 已提交
57
} ESyncState;
S
Shengliang Guan 已提交
58

M
Minghao Li 已提交
59
typedef struct SNodeInfo {
M
Minghao Li 已提交
60 61
  uint16_t nodePort;
  char     nodeFqdn[TSDB_FQDN_LEN];
S
Shengliang Guan 已提交
62 63
} SNodeInfo;

M
Minghao Li 已提交
64
typedef struct SSyncCfg {
M
Minghao Li 已提交
65
  int32_t   replicaNum;
M
Minghao Li 已提交
66
  int32_t   myIndex;
S
Shengliang Guan 已提交
67
  SNodeInfo nodeInfo[TSDB_MAX_REPLICA];
M
Minghao Li 已提交
68
} SSyncCfg;
S
Shengliang Guan 已提交
69

M
Minghao Li 已提交
70
typedef struct SFsmCbMeta {
M
Minghao Li 已提交
71
  int32_t    code;
M
Minghao Li 已提交
72
  SyncIndex  index;
M
Minghao Li 已提交
73 74
  SyncTerm   term;
  uint64_t   seqNum;
75
  SyncIndex  lastConfigIndex;
M
Minghao Li 已提交
76
  ESyncState state;
77
  SyncTerm   currentTerm;
M
Minghao Li 已提交
78
  bool       isWeak;
79
  uint64_t   flag;
M
Minghao Li 已提交
80 81
} SFsmCbMeta;

82
typedef struct SReConfigCbMeta {
M
Minghao Li 已提交
83 84 85 86 87 88 89 90 91 92 93
  int32_t    code;
  SyncIndex  index;
  SyncTerm   term;
  uint64_t   seqNum;
  SyncIndex  lastConfigIndex;
  ESyncState state;
  SyncTerm   currentTerm;
  bool       isWeak;
  uint64_t   flag;

  // config info
M
Minghao Li 已提交
94
  SSyncCfg  oldCfg;
95
  SSyncCfg  newCfg;
M
Minghao Li 已提交
96 97 98 99
  SyncIndex newCfgIndex;
  SyncTerm  newCfgTerm;
  uint64_t  newCfgSeqNum;

100 101
} SReConfigCbMeta;

102 103 104 105 106
typedef struct SSnapshotParam {
  SyncIndex start;
  SyncIndex end;
} SSnapshotParam;

107
typedef struct SSnapshot {
M
Minghao Li 已提交
108
  void*     data;
109 110
  SyncIndex lastApplyIndex;
  SyncTerm  lastApplyTerm;
M
Minghao Li 已提交
111
  SyncIndex lastConfigIndex;
112 113
} SSnapshot;

114 115 116 117
typedef struct SSnapshotMeta {
  SyncIndex lastConfigIndex;
} SSnapshotMeta;

M
Minghao Li 已提交
118 119
typedef struct SSyncFSM {
  void* data;
120

M
Minghao Li 已提交
121 122 123
  void (*FpCommitCb)(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SFsmCbMeta cbMeta);
  void (*FpPreCommitCb)(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SFsmCbMeta cbMeta);
  void (*FpRollBackCb)(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SFsmCbMeta cbMeta);
124

125
  void (*FpRestoreFinishCb)(struct SSyncFSM* pFsm);
126
  void (*FpReConfigCb)(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SReConfigCbMeta cbMeta);
M
Minghao Li 已提交
127
  void (*FpLeaderTransferCb)(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SFsmCbMeta cbMeta);
M
Minghao Li 已提交
128

M
Minghao Li 已提交
129
  int32_t (*FpGetSnapshot)(struct SSyncFSM* pFsm, SSnapshot* pSnapshot, void* pReaderParam, void** ppReader);
130
  int32_t (*FpGetSnapshotInfo)(struct SSyncFSM* pFsm, SSnapshot* pSnapshot);
131

132
  int32_t (*FpSnapshotStartRead)(struct SSyncFSM* pFsm, void* pReaderParam, void** ppReader);
133 134
  int32_t (*FpSnapshotStopRead)(struct SSyncFSM* pFsm, void* pReader);
  int32_t (*FpSnapshotDoRead)(struct SSyncFSM* pFsm, void* pReader, void** ppBuf, int32_t* len);
135

136
  int32_t (*FpSnapshotStartWrite)(struct SSyncFSM* pFsm, void* pWriterParam, void** ppWriter);
137
  int32_t (*FpSnapshotStopWrite)(struct SSyncFSM* pFsm, void* pWriter, bool isApply, SSnapshot* pSnapshot);
138
  int32_t (*FpSnapshotDoWrite)(struct SSyncFSM* pFsm, void* pWriter, void* pBuf, int32_t len);
139

S
Shengliang Guan 已提交
140 141
} SSyncFSM;

M
Minghao Li 已提交
142 143
// abstract definition of log store in raft
// SWal implements it
S
Shengliang Guan 已提交
144
typedef struct SSyncLogStore {
M
Minghao Li 已提交
145 146 147
  void* data;

  // append one log entry
M
Minghao Li 已提交
148
  int32_t (*appendEntry)(struct SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry);
S
Shengliang Guan 已提交
149

M
Minghao Li 已提交
150
  // get one log entry, user need to free pEntry->pCont
M
Minghao Li 已提交
151
  SSyncRaftEntry* (*getEntry)(struct SSyncLogStore* pLogStore, SyncIndex index);
S
Shengliang Guan 已提交
152

M
Minghao Li 已提交
153 154
  // truncate log with index, entries after the given index (>=index) will be deleted
  int32_t (*truncate)(struct SSyncLogStore* pLogStore, SyncIndex fromIndex);
S
Shengliang Guan 已提交
155

M
Minghao Li 已提交
156 157 158 159 160
  // return index of last entry
  SyncIndex (*getLastIndex)(struct SSyncLogStore* pLogStore);

  // return term of last entry
  SyncTerm (*getLastTerm)(struct SSyncLogStore* pLogStore);
S
Shengliang Guan 已提交
161

M
Minghao Li 已提交
162 163 164 165 166 167
  // update log store commit index with "index"
  int32_t (*updateCommitIndex)(struct SSyncLogStore* pLogStore, SyncIndex index);

  // return commit index of log
  SyncIndex (*getCommitIndex)(struct SSyncLogStore* pLogStore);

M
Minghao Li 已提交
168 169 170 171
  SyncIndex (*syncLogBeginIndex)(struct SSyncLogStore* pLogStore);
  SyncIndex (*syncLogEndIndex)(struct SSyncLogStore* pLogStore);
  bool (*syncLogIsEmpty)(struct SSyncLogStore* pLogStore);
  int32_t (*syncLogEntryCount)(struct SSyncLogStore* pLogStore);
172
  int32_t (*syncLogRestoreFromSnapshot)(struct SSyncLogStore* pLogStore, SyncIndex index);
173
  bool (*syncLogExist)(struct SSyncLogStore* pLogStore, SyncIndex index);
M
Minghao Li 已提交
174

M
Minghao Li 已提交
175
  SyncIndex (*syncLogWriteIndex)(struct SSyncLogStore* pLogStore);
M
Minghao Li 已提交
176 177 178 179 180 181 182
  SyncIndex (*syncLogLastIndex)(struct SSyncLogStore* pLogStore);
  SyncTerm (*syncLogLastTerm)(struct SSyncLogStore* pLogStore);

  int32_t (*syncLogAppendEntry)(struct SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry);
  int32_t (*syncLogGetEntry)(struct SSyncLogStore* pLogStore, SyncIndex index, SSyncRaftEntry** ppEntry);
  int32_t (*syncLogTruncate)(struct SSyncLogStore* pLogStore, SyncIndex fromIndex);

S
Shengliang Guan 已提交
183 184
} SSyncLogStore;

M
Minghao Li 已提交
185
typedef struct SSyncInfo {
M
Minghao Li 已提交
186 187 188 189 190 191 192 193 194
  bool          isStandBy;
  ESyncStrategy snapshotStrategy;
  SyncGroupId   vgId;
  int32_t       batchSize;
  SSyncCfg      syncCfg;
  char          path[TSDB_FILENAME_LEN];
  SWal*         pWal;
  SSyncFSM*     pFsm;
  SMsgCb*       msgcb;
S
Shengliang Guan 已提交
195 196
  int32_t (*FpSendMsg)(const SEpSet* pEpSet, SRpcMsg* pMsg);
  int32_t (*FpEqMsg)(const SMsgCb* msgcb, SRpcMsg* pMsg);
S
Shengliang Guan 已提交
197 198
} SSyncInfo;

M
Minghao Li 已提交
199 200 201 202 203
int32_t     syncInit();
void        syncCleanUp();
int64_t     syncOpen(const SSyncInfo* pSyncInfo);
void        syncStart(int64_t rid);
void        syncStop(int64_t rid);
M
Minghao Li 已提交
204
int32_t     syncSetStandby(int64_t rid);
M
Minghao Li 已提交
205
ESyncState  syncGetMyRole(int64_t rid);
M
Minghao Li 已提交
206
bool        syncIsReady(int64_t rid);
M
Minghao Li 已提交
207 208
const char* syncGetMyRoleStr(int64_t rid);
SyncTerm    syncGetMyTerm(int64_t rid);
M
Minghao Li 已提交
209
SyncGroupId syncGetVgId(int64_t rid);
M
Minghao Li 已提交
210
void        syncGetEpSet(int64_t rid, SEpSet* pEpSet);
211
void        syncGetRetryEpSet(int64_t rid, SEpSet* pEpSet);
M
Minghao Li 已提交
212
int32_t     syncPropose(int64_t rid, SRpcMsg* pMsg, bool isWeak);
213
int32_t     syncProposeBatch(int64_t rid, SRpcMsg** pMsgPArr, bool* pIsWeakArr, int32_t arrSize);
214
bool        syncEnvIsStart();
M
Minghao Li 已提交
215
const char* syncStr(ESyncState state);
M
Minghao Li 已提交
216
bool        syncIsRestoreFinish(int64_t rid);
217
int32_t     syncGetSnapshotByIndex(int64_t rid, SyncIndex index, SSnapshot* pSnapshot);
218

219
int32_t syncReconfig(int64_t rid, const SSyncCfg* pNewCfg);
M
Minghao Li 已提交
220 221 222

// build SRpcMsg, need to call syncPropose with SRpcMsg
int32_t syncReconfigBuild(int64_t rid, const SSyncCfg* pNewCfg, SRpcMsg* pRpcMsg);
223

224 225 226
int32_t syncLeaderTransfer(int64_t rid);
int32_t syncLeaderTransferTo(int64_t rid, SNodeInfo newLeader);

S
Shengliang Guan 已提交
227 228 229 230 231
#ifdef __cplusplus
}
#endif

#endif /*_TD_LIBS_SYNC_H*/