vnodeSync.c 17.8 KB
Newer Older
M
Minghao Li 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
/*
 * 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/>.
 */

S
Shengliang Guan 已提交
16
#define _DEFAULT_SOURCE
H
Hongze Cheng 已提交
17
#include "vnd.h"
M
Minghao Li 已提交
18

19 20
#define BATCH_DISABLE 1

21
static inline bool vnodeIsMsgBlock(tmsg_t type) {
S
Shengliang Guan 已提交
22
  return (type == TDMT_VND_CREATE_TABLE) || (type == TDMT_VND_ALTER_TABLE) || (type == TDMT_VND_DROP_TABLE) ||
S
Shengliang Guan 已提交
23
         (type == TDMT_VND_UPDATE_TAG_VAL);
24
}
M
Minghao Li 已提交
25

26
static inline bool vnodeIsMsgWeak(tmsg_t type) { return false; }
S
Shengliang Guan 已提交
27

28 29
static inline void vnodeWaitBlockMsg(SVnode *pVnode, const SRpcMsg *pMsg) {
  if (vnodeIsMsgBlock(pMsg->msgType)) {
30
    const STraceId *trace = &pMsg->info.traceId;
31
    taosThreadMutexLock(&pVnode->lock);
32 33 34
    if (!pVnode->blocked) {
      vGTrace("vgId:%d, msg:%p wait block, type:%s", pVnode->config.vgId, pMsg, TMSG_INFO(pMsg->msgType));
      pVnode->blocked = true;
35
      taosThreadMutexUnlock(&pVnode->lock);
36 37
      tsem_wait(&pVnode->syncSem);
    } else {
38
      taosThreadMutexUnlock(&pVnode->lock);
39
    }
40
  }
M
Minghao Li 已提交
41 42
}

43 44
static inline void vnodePostBlockMsg(SVnode *pVnode, const SRpcMsg *pMsg) {
  if (vnodeIsMsgBlock(pMsg->msgType)) {
45
    const STraceId *trace = &pMsg->info.traceId;
46
    taosThreadMutexLock(&pVnode->lock);
47
    if (pVnode->blocked) {
48
      vGTrace("vgId:%d, msg:%p post block, type:%s", pVnode->config.vgId, pMsg, TMSG_INFO(pMsg->msgType));
49
      pVnode->blocked = false;
50 51
      tsem_post(&pVnode->syncSem);
    }
52
    taosThreadMutexUnlock(&pVnode->lock);
53 54 55
  }
}

56 57 58 59 60 61 62 63 64 65 66 67 68
void vnodeRedirectRpcMsg(SVnode *pVnode, SRpcMsg *pMsg) {
  SEpSet newEpSet = {0};
  syncGetRetryEpSet(pVnode->sync, &newEpSet);

  const STraceId *trace = &pMsg->info.traceId;
  vGTrace("vgId:%d, msg:%p is redirect since not leader, numOfEps:%d inUse:%d", pVnode->config.vgId, pMsg,
          newEpSet.numOfEps, newEpSet.inUse);
  for (int32_t i = 0; i < newEpSet.numOfEps; ++i) {
    vGTrace("vgId:%d, msg:%p redirect:%d ep:%s:%u", pVnode->config.vgId, pMsg, i, newEpSet.eps[i].fqdn,
            newEpSet.eps[i].port);
  }
  pMsg->info.hasEpSet = 1;

D
dapan1121 已提交
69
  SRpcMsg rsp = {.code = TSDB_CODE_RPC_REDIRECT, .info = pMsg->info, .msgType = pMsg->msgType + 1};
70 71 72
  tmsgSendRedirectRsp(&rsp, &newEpSet);
}

73 74 75 76 77 78 79 80 81
static void inline vnodeHandleWriteMsg(SVnode *pVnode, SRpcMsg *pMsg) {
  SRpcMsg rsp = {.code = pMsg->code, .info = pMsg->info};
  if (vnodeProcessWriteMsg(pVnode, pMsg, pMsg->info.conn.applyIndex, &rsp) < 0) {
    rsp.code = terrno;
    const STraceId *trace = &pMsg->info.traceId;
    vGError("vgId:%d, msg:%p failed to apply right now since %s", pVnode->config.vgId, pMsg, terrstr());
  }
  if (rsp.info.handle != NULL) {
    tmsgSendRsp(&rsp);
L
Liu Jicong 已提交
82 83 84 85
  } else {
    if (rsp.pCont) {
      rpcFreeCont(rsp.pCont);
    }
86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101
  }
}

static void vnodeHandleProposeError(SVnode *pVnode, SRpcMsg *pMsg, int32_t code) {
  if (code == TSDB_CODE_SYN_NOT_LEADER) {
    vnodeRedirectRpcMsg(pVnode, pMsg);
  } else {
    const STraceId *trace = &pMsg->info.traceId;
    vGError("vgId:%d, msg:%p failed to propose since %s, code:0x%x", pVnode->config.vgId, pMsg, tstrerror(code), code);
    SRpcMsg rsp = {.code = code, .info = pMsg->info};
    if (rsp.info.handle != NULL) {
      tmsgSendRsp(&rsp);
    }
  }
}

102
static void inline vnodeProposeBatchMsg(SVnode *pVnode, SRpcMsg **pMsgArr, bool *pIsWeakArr, int32_t *arrSize) {
103 104 105
  if (*arrSize <= 0) return;

#if BATCH_DISABLE
106
  int32_t code = syncPropose(pVnode->sync, pMsgArr[0], pIsWeakArr[0]);
107 108 109 110 111 112
#else
  int32_t code = syncProposeBatch(pVnode->sync, pMsgArr, pIsWeakArr, *arrSize);
#endif

  if (code > 0) {
    for (int32_t i = 0; i < *arrSize; ++i) {
113
      vnodeHandleWriteMsg(pVnode, pMsgArr[i]);
114 115
    }
  } else if (code == 0) {
116
    vnodeWaitBlockMsg(pVnode, pMsgArr[*arrSize - 1]);
117 118 119
  } else {
    if (terrno != 0) code = terrno;
    for (int32_t i = 0; i < *arrSize; ++i) {
120
      vnodeHandleProposeError(pVnode, pMsgArr[i], code);
121 122 123 124
    }
  }

  for (int32_t i = 0; i < *arrSize; ++i) {
125 126 127
    SRpcMsg        *pMsg = pMsgArr[i];
    const STraceId *trace = &pMsg->info.traceId;
    vGTrace("vgId:%d, msg:%p is freed, code:0x%x", pVnode->config.vgId, pMsg, code);
128
    rpcFreeCont(pMsg->pCont);
129
    taosFreeQitem(pMsg);
130 131 132 133 134
  }

  *arrSize = 0;
}

135
void vnodeProposeWriteMsg(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {
136 137 138 139 140 141 142
  SVnode   *pVnode = pInfo->ahandle;
  int32_t   vgId = pVnode->config.vgId;
  int32_t   code = 0;
  SRpcMsg  *pMsg = NULL;
  int32_t   arrayPos = 0;
  SRpcMsg **pMsgArr = taosMemoryCalloc(numOfMsgs, sizeof(SRpcMsg *));
  bool     *pIsWeakArr = taosMemoryCalloc(numOfMsgs, sizeof(bool));
143 144
  vTrace("vgId:%d, get %d msgs from vnode-write queue", vgId, numOfMsgs);

145
  for (int32_t msg = 0; msg < numOfMsgs; msg++) {
146
    if (taosGetQitem(qall, (void **)&pMsg) == 0) continue;
147 148 149
    bool isWeak = vnodeIsMsgWeak(pMsg->msgType);
    bool isBlock = vnodeIsMsgBlock(pMsg->msgType);

150
    const STraceId *trace = &pMsg->info.traceId;
151 152 153
    vGTrace("vgId:%d, msg:%p get from vnode-write queue, weak:%d block:%d msg:%d:%d pos:%d, handle:%p", vgId, pMsg,
            isWeak, isBlock, msg, numOfMsgs, arrayPos, pMsg->info.handle);

154
    if (!pVnode->restored) {
155
      vGError("vgId:%d, msg:%p failed to process since restore not finished", vgId, pMsg);
156 157 158 159 160 161 162
      terrno = TSDB_CODE_APP_NOT_READY;
      vnodeHandleProposeError(pVnode, pMsg, TSDB_CODE_APP_NOT_READY);
      rpcFreeCont(pMsg->pCont);
      taosFreeQitem(pMsg);
      continue;
    }

163 164
    if (pMsgArr == NULL || pIsWeakArr == NULL) {
      vGError("vgId:%d, msg:%p failed to process since out of memory", vgId, pMsg);
165 166
      terrno = TSDB_CODE_OUT_OF_MEMORY;
      vnodeHandleProposeError(pVnode, pMsg, terrno);
167 168 169 170
      rpcFreeCont(pMsg->pCont);
      taosFreeQitem(pMsg);
      continue;
    }
171

172
    code = vnodePreProcessWriteMsg(pVnode, pMsg);
173
    if (code != 0) {
174 175 176 177
      vGError("vgId:%d, msg:%p failed to pre-process since %s", vgId, pMsg, terrstr());
      rpcFreeCont(pMsg->pCont);
      taosFreeQitem(pMsg);
      continue;
178 179
    }

180 181 182 183
    if (isBlock || BATCH_DISABLE) {
      vnodeProposeBatchMsg(pVnode, pMsgArr, pIsWeakArr, &arrayPos);
    }

184
    pMsgArr[arrayPos] = pMsg;
185 186 187 188 189
    pIsWeakArr[arrayPos] = isWeak;
    arrayPos++;

    if (isBlock || msg == numOfMsgs - 1 || BATCH_DISABLE) {
      vnodeProposeBatchMsg(pVnode, pMsgArr, pIsWeakArr, &arrayPos);
190
    }
191
  }
192 193 194

  taosMemoryFree(pMsgArr);
  taosMemoryFree(pIsWeakArr);
195 196
}

197
void vnodeApplyWriteMsg(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {
198
  SVnode  *pVnode = pInfo->ahandle;
199 200 201 202 203 204
  int32_t  vgId = pVnode->config.vgId;
  int32_t  code = 0;
  SRpcMsg *pMsg = NULL;

  for (int32_t i = 0; i < numOfMsgs; ++i) {
    if (taosGetQitem(qall, (void **)&pMsg) == 0) continue;
205
    const STraceId *trace = &pMsg->info.traceId;
206 207
    vGTrace("vgId:%d, msg:%p get from vnode-apply queue, type:%s handle:%p index:%" PRId64, vgId, pMsg,
            TMSG_INFO(pMsg->msgType), pMsg->info.handle, pMsg->info.conn.applyIndex);
208

S
Shengliang Guan 已提交
209 210
    SRpcMsg rsp = {.code = pMsg->code, .info = pMsg->info};
    if (rsp.code == 0) {
211
      if (vnodeProcessWriteMsg(pVnode, pMsg, pMsg->info.conn.applyIndex, &rsp) < 0) {
S
Shengliang Guan 已提交
212
        rsp.code = terrno;
213 214
        vGError("vgId:%d, msg:%p failed to apply since %s, index:%" PRId64, vgId, pMsg, terrstr(),
                pMsg->info.conn.applyIndex);
S
Shengliang Guan 已提交
215 216
      }
    }
217

218
    vnodePostBlockMsg(pVnode, pMsg);
S
Shengliang Guan 已提交
219
    if (rsp.info.handle != NULL) {
220
      tmsgSendRsp(&rsp);
L
Liu Jicong 已提交
221 222 223 224
    } else {
      if (rsp.pCont) {
        rpcFreeCont(rsp.pCont);
      }
225 226
    }

227
    vGTrace("vgId:%d, msg:%p is freed, code:0x%x index:%" PRId64, vgId, pMsg, rsp.code, pMsg->info.conn.applyIndex);
228 229
    rpcFreeCont(pMsg->pCont);
    taosFreeQitem(pMsg);
230
  }
M
Minghao Li 已提交
231 232
}

233
int32_t vnodeProcessSyncMsg(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
234
  const STraceId *trace = &pMsg->info.traceId;
235
  vGTrace("vgId:%d, sync msg:%p will be processed, type:%s", pVnode->config.vgId, pMsg, TMSG_INFO(pMsg->msgType));
236

S
Shengliang Guan 已提交
237 238 239
  int32_t code = syncProcessMsg(pVnode->sync, pMsg);
  if (code != 0) {
    vGError("vgId:%d, failed to process sync msg:%p type:%s since %s", pVnode->config.vgId, pMsg,
S
Shengliang Guan 已提交
240
            TMSG_INFO(pMsg->msgType), terrstr());
S
Shengliang Guan 已提交
241 242
  }

243
  return code;
S
Shengliang Guan 已提交
244 245
}

246 247 248 249 250 251 252 253 254
static int32_t vnodeSyncEqCtrlMsg(const SMsgCb *msgcb, SRpcMsg *pMsg) {
  int32_t code = tmsgPutToQueue(msgcb, SYNC_CTRL_QUEUE, pMsg);
  if (code != 0) {
    rpcFreeCont(pMsg->pCont);
    pMsg->pCont = NULL;
  }
  return code;
}

255
static int32_t vnodeSyncEqMsg(const SMsgCb *msgcb, SRpcMsg *pMsg) {
M
Minghao Li 已提交
256 257 258
  int32_t code = tmsgPutToQueue(msgcb, SYNC_QUEUE, pMsg);
  if (code != 0) {
    rpcFreeCont(pMsg->pCont);
259
    pMsg->pCont = NULL;
M
Minghao Li 已提交
260 261 262
  }
  return code;
}
M
Minghao Li 已提交
263

264
static int32_t vnodeSyncSendMsg(const SEpSet *pEpSet, SRpcMsg *pMsg) {
265 266 267 268 269 270 271
  int32_t code = tmsgSendReq(pEpSet, pMsg);
  if (code != 0) {
    rpcFreeCont(pMsg->pCont);
    pMsg->pCont = NULL;
  }
  return code;
}
M
Minghao Li 已提交
272

S
Shengliang Guan 已提交
273
static int32_t vnodeSyncGetSnapshot(const SSyncFSM *pFsm, SSnapshot *pSnapshot) {
S
Shengliang Guan 已提交
274
  vnodeGetSnapshot(pFsm->data, pSnapshot);
M
Minghao Li 已提交
275 276 277
  return 0;
}

S
Shengliang Guan 已提交
278 279
static void vnodeSyncApplyMsg(const SSyncFSM *pFsm, const SRpcMsg *pMsg, const SFsmCbMeta *pMeta) {
  SVnode *pVnode = pFsm->data;
280

S
Shengliang Guan 已提交
281 282 283 284 285 286 287
  if (pMeta->code == 0) {
    SRpcMsg rpcMsg = {.msgType = pMsg->msgType, .contLen = pMsg->contLen};
    rpcMsg.pCont = rpcMallocCont(rpcMsg.contLen);
    memcpy(rpcMsg.pCont, pMsg->pCont, pMsg->contLen);
    rpcMsg.info = pMsg->info;
    rpcMsg.info.conn.applyIndex = pMeta->index;
    rpcMsg.info.conn.applyTerm = pMeta->term;
M
Minghao Li 已提交
288

S
Shengliang Guan 已提交
289 290
    const STraceId *trace = &pMsg->info.traceId;
    vGTrace("vgId:%d, commit-cb is excuted, fsm:%p, index:%" PRId64 ", term:%" PRIu64 ", msg-index:%" PRId64
S
Shengliang Guan 已提交
291
            ", weak:%d, code:%d, state:%d %s, type:%s",
292 293
            pVnode->config.vgId, pFsm, pMeta->index, pMeta->term, rpcMsg.info.conn.applyIndex, pMeta->isWeak,
            pMeta->code, pMeta->state, syncStr(pMeta->state), TMSG_INFO(pMsg->msgType));
M
Minghao Li 已提交
294

S
Shengliang Guan 已提交
295 296 297
    tmsgPutToQueue(&pVnode->msgCb, APPLY_QUEUE, &rpcMsg);
  } else {
    SRpcMsg rsp = {.code = pMeta->code, .info = pMsg->info};
298
    vError("vgId:%d, commit-cb execute error, type:%s, index:%" PRId64 ", error:0x%x %s", pVnode->config.vgId,
S
Shengliang Guan 已提交
299 300 301
           TMSG_INFO(pMsg->msgType), pMeta->index, pMeta->code, tstrerror(pMeta->code));
    if (rsp.info.handle != NULL) {
      tmsgSendRsp(&rsp);
M
Minghao Li 已提交
302 303
    }
  }
M
Minghao Li 已提交
304 305
}

S
Shengliang Guan 已提交
306 307
static void vnodeSyncCommitMsg(const SSyncFSM *pFsm, const SRpcMsg *pMsg, const SFsmCbMeta *pMeta) {
  if (pMeta->isWeak == 0) {
S
Shengliang Guan 已提交
308
    vnodeSyncApplyMsg(pFsm, pMsg, pMeta);
309
  }
M
Minghao Li 已提交
310 311
}

S
Shengliang Guan 已提交
312 313
static void vnodeSyncPreCommitMsg(const SSyncFSM *pFsm, const SRpcMsg *pMsg, const SFsmCbMeta *pMeta) {
  if (pMeta->isWeak == 1) {
S
Shengliang Guan 已提交
314
    vnodeSyncApplyMsg(pFsm, pMsg, pMeta);
S
Shengliang Guan 已提交
315 316 317 318
  }
}

static void vnodeSyncRollBackMsg(const SSyncFSM *pFsm, const SRpcMsg *pMsg, const SFsmCbMeta *pMeta) {
319
  SVnode *pVnode = pFsm->data;
S
Shengliang Guan 已提交
320
  vTrace("vgId:%d, rollback-cb is excuted, fsm:%p, index:%" PRId64 ", weak:%d, code:%d, state:%d %s, type:%s",
321 322
         pVnode->config.vgId, pFsm, pMeta->index, pMeta->isWeak, pMeta->code, pMeta->state, syncStr(pMeta->state),
         TMSG_INFO(pMsg->msgType));
M
Minghao Li 已提交
323 324
}

S
Shengliang Guan 已提交
325
static int32_t vnodeSnapshotStartRead(const SSyncFSM *pFsm, void *pParam, void **ppReader) {
326 327
  SVnode         *pVnode = pFsm->data;
  SSnapshotParam *pSnapshotParam = pParam;
H
Hongze Cheng 已提交
328
  int32_t code = vnodeSnapReaderOpen(pVnode, pSnapshotParam->start, pSnapshotParam->end, (SVSnapReader **)ppReader);
329 330
  return code;
}
S
Shengliang Guan 已提交
331

S
Shengliang Guan 已提交
332
static int32_t vnodeSnapshotStopRead(const SSyncFSM *pFsm, void *pReader) {
333
  SVnode *pVnode = pFsm->data;
H
Hongze Cheng 已提交
334
  int32_t code = vnodeSnapReaderClose(pReader);
335 336
  return code;
}
S
Shengliang Guan 已提交
337

S
Shengliang Guan 已提交
338
static int32_t vnodeSnapshotDoRead(const SSyncFSM *pFsm, void *pReader, void **ppBuf, int32_t *len) {
339
  SVnode *pVnode = pFsm->data;
H
Hongze Cheng 已提交
340
  int32_t code = vnodeSnapRead(pReader, (uint8_t **)ppBuf, len);
341 342
  return code;
}
S
Shengliang Guan 已提交
343

S
Shengliang Guan 已提交
344
static int32_t vnodeSnapshotStartWrite(const SSyncFSM *pFsm, void *pParam, void **ppWriter) {
M
Minghao Li 已提交
345 346
  SVnode         *pVnode = pFsm->data;
  SSnapshotParam *pSnapshotParam = pParam;
347 348 349

  do {
    int32_t itemSize = tmsgGetQueueSize(&pVnode->msgCb, pVnode->config.vgId, APPLY_QUEUE);
S
Shengliang Guan 已提交
350
    if (itemSize == 0) {
S
Shengliang Guan 已提交
351
      vInfo("vgId:%d, start write vnode snapshot since apply queue is empty", pVnode->config.vgId);
352 353
      break;
    } else {
S
Shengliang Guan 已提交
354
      vInfo("vgId:%d, write vnode snapshot later since %d items in apply queue", pVnode->config.vgId, itemSize);
355 356 357 358
      taosMsleep(10);
    }
  } while (true);

M
Minghao Li 已提交
359 360 361
  int32_t code = vnodeSnapWriterOpen(pVnode, pSnapshotParam->start, pSnapshotParam->end, (SVSnapWriter **)ppWriter);
  return code;
}
S
Shengliang Guan 已提交
362

S
Shengliang Guan 已提交
363
static int32_t vnodeSnapshotStopWrite(const SSyncFSM *pFsm, void *pWriter, bool isApply, SSnapshot *pSnapshot) {
M
Minghao Li 已提交
364
  SVnode *pVnode = pFsm->data;
365
  vInfo("vgId:%d, stop write vnode snapshot, apply:%d, index:%" PRId64 " term:%" PRIu64 " config:%" PRId64,
366
        pVnode->config.vgId, isApply, pSnapshot->lastApplyIndex, pSnapshot->lastApplyTerm, pSnapshot->lastConfigIndex);
367

368
  int32_t code = vnodeSnapWriterClose(pWriter, !isApply, pSnapshot);
S
Shengliang Guan 已提交
369
  vInfo("vgId:%d, apply vnode snapshot finished, code:0x%x", pVnode->config.vgId, code);
M
Minghao Li 已提交
370 371
  return code;
}
S
Shengliang Guan 已提交
372

S
Shengliang Guan 已提交
373
static int32_t vnodeSnapshotDoWrite(const SSyncFSM *pFsm, void *pWriter, void *pBuf, int32_t len) {
M
Minghao Li 已提交
374
  SVnode *pVnode = pFsm->data;
S
Shengliang Guan 已提交
375
  vDebug("vgId:%d, continue write vnode snapshot, len:%d", pVnode->config.vgId, len);
M
Minghao Li 已提交
376
  int32_t code = vnodeSnapWrite(pWriter, pBuf, len);
S
Shengliang Guan 已提交
377
  vDebug("vgId:%d, continue write vnode snapshot finished, len:%d", pVnode->config.vgId, len);
M
Minghao Li 已提交
378 379
  return code;
}
S
Shengliang Guan 已提交
380

S
Shengliang Guan 已提交
381
static void vnodeRestoreFinish(const SSyncFSM *pFsm) {
382
  SVnode *pVnode = pFsm->data;
383 384 385

  do {
    int32_t itemSize = tmsgGetQueueSize(&pVnode->msgCb, pVnode->config.vgId, APPLY_QUEUE);
S
Shengliang Guan 已提交
386
    if (itemSize == 0) {
387 388 389
      vInfo("vgId:%d, apply queue is empty, restore finish", pVnode->config.vgId);
      break;
    } else {
S
Shengliang Guan 已提交
390
      vInfo("vgId:%d, restore not finish since %d items in apply queue", pVnode->config.vgId, itemSize);
391 392 393 394
      taosMsleep(10);
    }
  } while (true);

395 396
  walApplyVer(pVnode->pWal, pVnode->state.applied);

397 398 399 400
  pVnode->restored = true;
  vDebug("vgId:%d, sync restore finished", pVnode->config.vgId);
}

S
Shengliang Guan 已提交
401
static void vnodeBecomeFollower(const SSyncFSM *pFsm) {
402 403 404
  SVnode *pVnode = pFsm->data;
  vDebug("vgId:%d, become follower", pVnode->config.vgId);

405
  taosThreadMutexLock(&pVnode->lock);
406 407
  if (pVnode->blocked) {
    pVnode->blocked = false;
408
    vDebug("vgId:%d, become follower and post block", pVnode->config.vgId);
409 410
    tsem_post(&pVnode->syncSem);
  }
411
  taosThreadMutexUnlock(&pVnode->lock);
412 413
}

S
Shengliang Guan 已提交
414
static void vnodeBecomeLeader(const SSyncFSM *pFsm) {
415 416 417 418
  SVnode *pVnode = pFsm->data;
  vDebug("vgId:%d, become leader", pVnode->config.vgId);
}

419
static SSyncFSM *vnodeSyncMakeFsm(SVnode *pVnode) {
S
Shengliang Guan 已提交
420
  SSyncFSM *pFsm = taosMemoryCalloc(1, sizeof(SSyncFSM));
M
Minghao Li 已提交
421
  pFsm->data = pVnode;
S
Shengliang Guan 已提交
422 423 424
  pFsm->FpCommitCb = vnodeSyncCommitMsg;
  pFsm->FpPreCommitCb = vnodeSyncPreCommitMsg;
  pFsm->FpRollBackCb = vnodeSyncRollBackMsg;
425
  pFsm->FpGetSnapshotInfo = vnodeSyncGetSnapshot;
426
  pFsm->FpRestoreFinishCb = vnodeRestoreFinish;
S
Shengliang Guan 已提交
427
  pFsm->FpLeaderTransferCb = NULL;
428 429
  pFsm->FpBecomeLeaderCb = vnodeBecomeLeader;
  pFsm->FpBecomeFollowerCb = vnodeBecomeFollower;
S
Shengliang Guan 已提交
430
  pFsm->FpReConfigCb = NULL;
S
Shengliang Guan 已提交
431 432 433 434 435 436 437
  pFsm->FpSnapshotStartRead = vnodeSnapshotStartRead;
  pFsm->FpSnapshotStopRead = vnodeSnapshotStopRead;
  pFsm->FpSnapshotDoRead = vnodeSnapshotDoRead;
  pFsm->FpSnapshotStartWrite = vnodeSnapshotStartWrite;
  pFsm->FpSnapshotStopWrite = vnodeSnapshotStopWrite;
  pFsm->FpSnapshotDoWrite = vnodeSnapshotDoWrite;

M
Minghao Li 已提交
438
  return pFsm;
439 440 441 442
}

int32_t vnodeSyncOpen(SVnode *pVnode, char *path) {
  SSyncInfo syncInfo = {
443
      .snapshotStrategy = SYNC_STRATEGY_WAL_FIRST,
M
Minghao Li 已提交
444
      .batchSize = 1,
445 446 447
      .vgId = pVnode->config.vgId,
      .syncCfg = pVnode->config.syncCfg,
      .pWal = pVnode->pWal,
S
Shengliang Guan 已提交
448
      .msgcb = &pVnode->msgCb,
S
Shengliang Guan 已提交
449 450 451 452 453 454
      .syncSendMSg = vnodeSyncSendMsg,
      .syncEqMsg = vnodeSyncEqMsg,
      .syncEqCtrlMsg = vnodeSyncEqCtrlMsg,
      .pingMs = 5000,
      .electMs = 4000,
      .heartbeatMs = 700,
455 456 457 458 459
  };

  snprintf(syncInfo.path, sizeof(syncInfo.path), "%s%ssync", path, TD_DIRSEP);
  syncInfo.pFsm = vnodeSyncMakeFsm(pVnode);

S
Shengliang Guan 已提交
460 461 462 463 464 465 466
  SSyncCfg *pCfg = &syncInfo.syncCfg;
  vInfo("vgId:%d, start to open sync, replica:%d selfIndex:%d", pVnode->config.vgId, pCfg->replicaNum, pCfg->myIndex);
  for (int32_t i = 0; i < pCfg->replicaNum; ++i) {
    SNodeInfo *pNode = &pCfg->nodeInfo[i];
    vInfo("vgId:%d, index:%d ep:%s:%u", pVnode->config.vgId, i, pNode->nodeFqdn, pNode->nodePort);
  }

467 468 469 470 471 472 473 474 475 476
  pVnode->sync = syncOpen(&syncInfo);
  if (pVnode->sync <= 0) {
    vError("vgId:%d, failed to open sync since %s", pVnode->config.vgId, terrstr());
    return -1;
  }

  return 0;
}

void vnodeSyncStart(SVnode *pVnode) {
S
Shengliang Guan 已提交
477
  vInfo("vgId:%d, start sync", pVnode->config.vgId);
M
Minghao Li 已提交
478
  syncStart(pVnode->sync);
479 480
}

S
Shengliang Guan 已提交
481 482 483 484 485 486 487 488 489 490 491 492 493
void vnodeSyncPreClose(SVnode *pVnode) {
  vInfo("vgId:%d, pre close sync", pVnode->config.vgId);
  syncLeaderTransfer(pVnode->sync);
  syncPreStop(pVnode->sync);
  taosThreadMutexLock(&pVnode->lock);
  if (pVnode->blocked) {
    vInfo("vgId:%d, post block after close sync", pVnode->config.vgId);
    pVnode->blocked = false;
    tsem_post(&pVnode->syncSem);
  }
  taosThreadMutexUnlock(&pVnode->lock);
}

S
Shengliang Guan 已提交
494
void vnodeSyncClose(SVnode *pVnode) {
S
Shengliang Guan 已提交
495
  vInfo("vgId:%d, close sync", pVnode->config.vgId);
S
Shengliang Guan 已提交
496 497
  syncStop(pVnode->sync);
}
498

499 500 501 502
bool vnodeIsRoleLeader(SVnode *pVnode) {
  SSyncState state = syncGetState(pVnode->sync);
  return state.state == TAOS_SYNC_STATE_LEADER;
}
503

504
bool vnodeIsLeader(SVnode *pVnode) {
505 506 507 508 509 510 511 512 513
  SSyncState state = syncGetState(pVnode->sync);

  if (state.state != TAOS_SYNC_STATE_LEADER || !state.restored) {
    if (state.state != TAOS_SYNC_STATE_LEADER) {
      terrno = TSDB_CODE_SYN_NOT_LEADER;
    } else {
      terrno = TSDB_CODE_APP_NOT_READY;
    }
    vDebug("vgId:%d, vnode not ready, state:%s, restore:%d", pVnode->config.vgId, syncStr(state.state), state.restored);
514 515 516
    return false;
  }

517
  if (!pVnode->restored) {
518
    vDebug("vgId:%d, vnode not restored", pVnode->config.vgId);
519 520 521
    terrno = TSDB_CODE_APP_NOT_READY;
    return false;
  }
522 523

  return true;
L
Liu Jicong 已提交
524
}