mndSync.c 11.4 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16
/*
 * 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/>.
 */

#define _DEFAULT_SOURCE
S
Shengliang Guan 已提交
17
#include "mndSync.h"
18
#include "mndTrans.h"
19

20
static int32_t mndSyncEqMsg(const SMsgCb *msgcb, SRpcMsg *pMsg) {
M
Minghao Li 已提交
21 22 23 24
  SMsgHead *pHead = pMsg->pCont;
  pHead->contLen = htonl(pHead->contLen);
  pHead->vgId = htonl(pHead->vgId);

25 26 27 28 29 30
  int32_t code = tmsgPutToQueue(msgcb, SYNC_QUEUE, pMsg);
  if (code != 0) {
    rpcFreeCont(pMsg->pCont);
    pMsg->pCont = NULL;
  }
  return code;
M
Minghao Li 已提交
31
}
M
Minghao Li 已提交
32

33 34 35 36 37 38 39 40
static int32_t mndSyncSendMsg(const SEpSet *pEpSet, SRpcMsg *pMsg) {
  int32_t code = tmsgSendReq(pEpSet, pMsg);
  if (code != 0) {
    rpcFreeCont(pMsg->pCont);
    pMsg->pCont = NULL;
  }
  return code;
}
M
Minghao Li 已提交
41

42
void mndSyncCommitMsg(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cbMeta) {
S
Shengliang Guan 已提交
43 44 45 46
  SMnode    *pMnode = pFsm->data;
  SSyncMgmt *pMgmt = &pMnode->syncMgmt;
  SSdbRaw   *pRaw = pMsg->pCont;

47 48 49 50
  // delete msg handle
  SRpcMsg rpcMsg = {0};
  syncGetAndDelRespRpc(pMnode->syncMgmt.sync, cbMeta.seqNum, &rpcMsg.info);

S
Shengliang Guan 已提交
51
  int32_t transId = sdbGetIdFromRaw(pMnode->pSdb, pRaw);
S
Shengliang Guan 已提交
52
  pMgmt->errCode = cbMeta.code;
53
  mInfo("trans:%d, is proposed, saved:%d code:0x%x, apply index:%" PRId64 " term:%" PRIu64 " config:%" PRId64
H
Hongze Cheng 已提交
54 55 56
        " role:%s raw:%p",
        transId, pMgmt->transId, cbMeta.code, cbMeta.index, cbMeta.term, cbMeta.lastConfigIndex, syncStr(cbMeta.state),
        pRaw);
S
Shengliang Guan 已提交
57 58 59

  if (pMgmt->errCode == 0) {
    sdbWriteWithoutFree(pMnode->pSdb, pRaw);
60
    sdbSetApplyInfo(pMnode->pSdb, cbMeta.index, cbMeta.term, cbMeta.lastConfigIndex);
S
Shengliang Guan 已提交
61 62
  }

63
  taosWLockLatch(&pMgmt->lock);
64
  if (transId <= 0) {
65
    taosWUnLockLatch(&pMgmt->lock);
66 67
    mError("trans:%d, invalid commit msg", transId);
  } else if (transId == pMgmt->transId) {
S
Shengliang Guan 已提交
68
    if (pMgmt->errCode != 0) {
69 70
      mError("trans:%d, failed to propose since %s, post sem", transId, tstrerror(pMgmt->errCode));
    } else {
S
Shengliang Guan 已提交
71
      mInfo("trans:%d, is proposed and post sem", transId);
S
Shengliang Guan 已提交
72
    }
S
Shengliang Guan 已提交
73
    pMgmt->transId = 0;
74
    taosWUnLockLatch(&pMgmt->lock);
S
Shengliang Guan 已提交
75
    tsem_post(&pMgmt->syncSem);
76
  } else {
77
    taosWUnLockLatch(&pMgmt->lock);
78 79
    STrans *pTrans = mndAcquireTrans(pMnode, transId);
    if (pTrans != NULL) {
80
      mInfo("trans:%d, execute in mnode which not leader", transId);
81 82
      mndTransExecute(pMnode, pTrans);
      mndReleaseTrans(pMnode, pTrans);
83 84 85
      // sdbWriteFile(pMnode->pSdb, SDB_WRITE_DELTA);
    } else {
      mError("trans:%d, not found while execute in mnode since %s", transId, terrstr());
86
    }
M
Minghao Li 已提交
87 88 89
  }
}

90
int32_t mndSyncGetSnapshot(struct SSyncFSM *pFsm, SSnapshot *pSnapshot, void *pReaderParam, void **ppReader) {
91
  mInfo("start to read snapshot from sdb in atomic way");
92 93 94
  SMnode *pMnode = pFsm->data;
  return sdbStartRead(pMnode->pSdb, (SSdbIter **)ppReader, &pSnapshot->lastApplyIndex, &pSnapshot->lastApplyTerm,
                      &pSnapshot->lastConfigIndex);
95 96 97 98
  return 0;
}

int32_t mndSyncGetSnapshotInfo(struct SSyncFSM *pFsm, SSnapshot *pSnapshot) {
99
  SMnode *pMnode = pFsm->data;
100
  sdbGetCommitInfo(pMnode->pSdb, &pSnapshot->lastApplyIndex, &pSnapshot->lastApplyTerm, &pSnapshot->lastConfigIndex);
M
Minghao Li 已提交
101 102 103
  return 0;
}

104
void mndRestoreFinish(struct SSyncFSM *pFsm) {
105
  SMnode *pMnode = pFsm->data;
S
Shengliang Guan 已提交
106

S
Shengliang Guan 已提交
107
  if (!pMnode->deploy) {
108
    mInfo("vgId:1, sync restore finished, and will handle outstanding transactions");
S
Shengliang Guan 已提交
109
    mndTransPullup(pMnode);
110
    mndSetRestored(pMnode, true);
S
Shengliang Guan 已提交
111
  } else {
112
    mInfo("vgId:1, sync restore finished");
S
Shengliang Guan 已提交
113
  }
114 115
}

S
Shengliang Guan 已提交
116
void mndReConfig(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SReConfigCbMeta *cbMeta) {
S
Shengliang Guan 已提交
117 118 119
  SMnode    *pMnode = pFsm->data;
  SSyncMgmt *pMgmt = &pMnode->syncMgmt;

S
Shengliang Guan 已提交
120
  pMgmt->errCode = cbMeta->code;
121
  mInfo("trans:-1, sync reconfig is proposed, saved:%d code:0x%x, index:%" PRId64 " term:%" PRId64, pMgmt->transId,
S
Shengliang Guan 已提交
122
        cbMeta->code, cbMeta->index, cbMeta->term);
S
Shengliang Guan 已提交
123

124
  taosWLockLatch(&pMgmt->lock);
S
Shengliang Guan 已提交
125 126
  if (pMgmt->transId == -1) {
    if (pMgmt->errCode != 0) {
127 128
      mError("trans:-1, failed to propose sync reconfig since %s, post sem", tstrerror(pMgmt->errCode));
    } else {
129
      mInfo("trans:-1, sync reconfig is proposed, saved:%d code:0x%x, index:%" PRId64 " term:%" PRId64 " post sem",
S
Shengliang Guan 已提交
130
            pMgmt->transId, cbMeta->code, cbMeta->index, cbMeta->term);
S
Shengliang Guan 已提交
131
    }
S
Shengliang Guan 已提交
132
    pMgmt->transId = 0;
S
Shengliang Guan 已提交
133
    tsem_post(&pMgmt->syncSem);
S
Shengliang Guan 已提交
134
  }
135
  taosWUnLockLatch(&pMgmt->lock);
136 137
}

138
int32_t mndSnapshotStartRead(struct SSyncFSM *pFsm, void *pParam, void **ppReader) {
139
  mInfo("start to read snapshot from sdb");
S
Shengliang Guan 已提交
140
  SMnode *pMnode = pFsm->data;
141
  return sdbStartRead(pMnode->pSdb, (SSdbIter **)ppReader, NULL, NULL, NULL);
S
Shengliang Guan 已提交
142 143 144
}

int32_t mndSnapshotStopRead(struct SSyncFSM *pFsm, void *pReader) {
145
  mInfo("stop to read snapshot from sdb");
S
Shengliang Guan 已提交
146 147 148 149 150 151 152 153 154
  SMnode *pMnode = pFsm->data;
  return sdbStopRead(pMnode->pSdb, pReader);
}

int32_t mndSnapshotDoRead(struct SSyncFSM *pFsm, void *pReader, void **ppBuf, int32_t *len) {
  SMnode *pMnode = pFsm->data;
  return sdbDoRead(pMnode->pSdb, pReader, ppBuf, len);
}

155
int32_t mndSnapshotStartWrite(struct SSyncFSM *pFsm, void *pParam, void **ppWriter) {
S
Shengliang Guan 已提交
156 157 158 159 160
  mInfo("start to apply snapshot to sdb");
  SMnode *pMnode = pFsm->data;
  return sdbStartWrite(pMnode->pSdb, (SSdbIter **)ppWriter);
}

161
int32_t mndSnapshotStopWrite(struct SSyncFSM *pFsm, void *pWriter, bool isApply, SSnapshot *pSnapshot) {
162
  mInfo("stop to apply snapshot to sdb, apply:%d, index:%" PRId64 " term:%" PRIu64 " config:%" PRId64, isApply,
S
Shengliang Guan 已提交
163
        pSnapshot->lastApplyIndex, pSnapshot->lastApplyTerm, pSnapshot->lastConfigIndex);
S
Shengliang Guan 已提交
164
  SMnode *pMnode = pFsm->data;
165 166
  return sdbStopWrite(pMnode->pSdb, pWriter, isApply, pSnapshot->lastApplyIndex, pSnapshot->lastApplyTerm,
                      pSnapshot->lastConfigIndex);
S
Shengliang Guan 已提交
167 168 169 170 171 172 173
}

int32_t mndSnapshotDoWrite(struct SSyncFSM *pFsm, void *pWriter, void *pBuf, int32_t len) {
  SMnode *pMnode = pFsm->data;
  return sdbDoWrite(pMnode->pSdb, pWriter, pBuf, len);
}

174 175 176
void mndLeaderTransfer(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cbMeta) {
  SMnode *pMnode = pFsm->data;
  atomic_store_8(&(pMnode->syncMgmt.leaderTransferFinish), 1);
177
  mInfo("vgId:1, mnode leader transfer finish");
178 179
}

180 181
static void mndBecomeFollower(struct SSyncFSM *pFsm) {
  SMnode *pMnode = pFsm->data;
182
  mInfo("vgId:1, become follower and post sem");
183

184
  taosWLockLatch(&pMnode->syncMgmt.lock);
185 186 187 188
  if (pMnode->syncMgmt.transId != 0) {
    pMnode->syncMgmt.transId = 0;
    tsem_post(&pMnode->syncMgmt.syncSem);
  }
189
  taosWUnLockLatch(&pMnode->syncMgmt.lock);
190 191 192
}

static void mndBecomeLeader(struct SSyncFSM *pFsm) {
193
  mInfo("vgId:1, become leader");
194
  SMnode *pMnode = pFsm->data;
195 196
}

197 198
SSyncFSM *mndSyncMakeFsm(SMnode *pMnode) {
  SSyncFSM *pFsm = taosMemoryCalloc(1, sizeof(SSyncFSM));
M
Minghao Li 已提交
199
  pFsm->data = pMnode;
200
  pFsm->FpCommitCb = mndSyncCommitMsg;
201 202
  pFsm->FpPreCommitCb = NULL;
  pFsm->FpRollBackCb = NULL;
203
  pFsm->FpRestoreFinishCb = mndRestoreFinish;
204
  pFsm->FpLeaderTransferCb = mndLeaderTransfer;
205
  pFsm->FpReConfigCb = mndReConfig;
206 207
  pFsm->FpBecomeLeaderCb = mndBecomeLeader;
  pFsm->FpBecomeFollowerCb = mndBecomeFollower;
S
Shengliang Guan 已提交
208
  pFsm->FpGetSnapshot = mndSyncGetSnapshot;
209
  pFsm->FpGetSnapshotInfo = mndSyncGetSnapshotInfo;
S
Shengliang Guan 已提交
210 211 212 213 214 215
  pFsm->FpSnapshotStartRead = mndSnapshotStartRead;
  pFsm->FpSnapshotStopRead = mndSnapshotStopRead;
  pFsm->FpSnapshotDoRead = mndSnapshotDoRead;
  pFsm->FpSnapshotStartWrite = mndSnapshotStartWrite;
  pFsm->FpSnapshotStopWrite = mndSnapshotStopWrite;
  pFsm->FpSnapshotDoWrite = mndSnapshotDoWrite;
M
Minghao Li 已提交
216
  return pFsm;
217 218 219 220
}

int32_t mndInitSync(SMnode *pMnode) {
  SSyncMgmt *pMgmt = &pMnode->syncMgmt;
221 222
  taosInitRWLatch(&pMgmt->lock);
  pMgmt->transId = 0;
223

224 225
  SSyncInfo syncInfo = {.vgId = 1, .FpSendMsg = mndSyncSendMsg, .FpEqMsg = mndSyncEqMsg};
  snprintf(syncInfo.path, sizeof(syncInfo.path), "%s%ssync", pMnode->path, TD_DIRSEP);
226
  syncInfo.pWal = pMnode->pWal;
227
  syncInfo.pFsm = mndSyncMakeFsm(pMnode);
M
Minghao Li 已提交
228
  syncInfo.snapshotStrategy = SYNC_STRATEGY_STANDARD_SNAPSHOT;
229

230 231 232 233 234 235 236 237 238
  mInfo("vgId:1, start to open sync, selfIndex:%d replica:%d", pMgmt->selfIndex, pMgmt->numOfReplicas);
  SSyncCfg *pCfg = &syncInfo.syncCfg;
  pCfg->replicaNum = pMgmt->numOfReplicas;
  pCfg->myIndex = pMgmt->selfIndex;
  for (int32_t i = 0; i < pMgmt->numOfReplicas; ++i) {
    SNodeInfo *pNode = &pCfg->nodeInfo[i];
    tstrncpy(pNode->nodeFqdn, pMgmt->replicas[i].fqdn, sizeof(pNode->nodeFqdn));
    pNode->nodePort = pMgmt->replicas[i].port;
    mInfo("vgId:1, index:%d ep:%s:%u", i, pNode->nodeFqdn, pNode->nodePort);
M
Minghao Li 已提交
239 240
  }

241
  tsem_init(&pMgmt->syncSem, 0, 0);
242 243 244 245 246
  pMgmt->sync = syncOpen(&syncInfo);
  if (pMgmt->sync <= 0) {
    mError("failed to open sync since %s", terrstr());
    return -1;
  }
M
Minghao Li 已提交
247

248
  // decrease election timer
M
Minghao Li 已提交
249
  setPingTimerMS(pMgmt->sync, 5000);
250
  setElectTimerMS(pMgmt->sync, 3000);
251
  setHeartbeatTimerMS(pMgmt->sync, 500);
252

253
  mInfo("mnode-sync is opened, id:%" PRId64, pMgmt->sync);
S
Shengliang Guan 已提交
254 255 256 257 258
  return 0;
}

void mndCleanupSync(SMnode *pMnode) {
  SSyncMgmt *pMgmt = &pMnode->syncMgmt;
259
  syncStop(pMgmt->sync);
260
  mInfo("mnode-sync is stopped, id:%" PRId64, pMgmt->sync);
261

262
  tsem_destroy(&pMgmt->syncSem);
263 264
  memset(pMgmt, 0, sizeof(SSyncMgmt));
}
M
Minghao Li 已提交
265

S
Shengliang Guan 已提交
266
int32_t mndSyncPropose(SMnode *pMnode, SSdbRaw *pRaw, int32_t transId) {
S
Shengliang Guan 已提交
267
  SSyncMgmt *pMgmt = &pMnode->syncMgmt;
268
  SRpcMsg    req = {.msgType = TDMT_MND_APPLY_MSG, .contLen = sdbGetRawTotalSize(pRaw)};
S
Shengliang Guan 已提交
269 270 271 272 273
  if (req.contLen <= 0) {
    terrno = TSDB_CODE_APP_ERROR;
    return -1;
  }

274 275 276
  req.pCont = rpcMallocCont(req.contLen);
  if (req.pCont == NULL) return -1;
  memcpy(req.pCont, pRaw, req.contLen);
S
Shengliang Guan 已提交
277

S
Shengliang Guan 已提交
278
  pMgmt->errCode = 0;
279
  taosWLockLatch(&pMgmt->lock);
280
  if (pMgmt->transId != 0) {
S
Shengliang Guan 已提交
281
    mError("trans:%d, can't be proposed since trans:%d alrady waiting for confirm", transId, pMgmt->transId);
282 283 284 285 286
    taosWUnLockLatch(&pMgmt->lock);
    terrno = TSDB_CODE_APP_NOT_READY;
    return -1;
  } else {
    pMgmt->transId = transId;
287
    mInfo("trans:%d, will be proposed", pMgmt->transId);
288 289
    taosWUnLockLatch(&pMgmt->lock);
  }
S
Shengliang Guan 已提交
290

291
  const bool isWeak = false;
292
  int32_t    code = syncPropose(pMgmt->sync, &req, isWeak);
293

294 295
  if (code == 0) {
    tsem_wait(&pMgmt->syncSem);
M
Minghao Li 已提交
296
  } else if (code == -1 && terrno == TSDB_CODE_SYN_NOT_LEADER) {
297
    terrno = TSDB_CODE_APP_NOT_READY;
M
Minghao Li 已提交
298
  } else if (code == -1 && terrno == TSDB_CODE_SYN_INTERNAL_ERROR) {
299 300
    terrno = TSDB_CODE_SYN_INTERNAL_ERROR;
  } else {
301
    terrno = TSDB_CODE_APP_ERROR;
302
  }
303

304
  rpcFreeCont(req.pCont);
S
Shengliang Guan 已提交
305 306 307 308 309
  if (code != 0) {
    mError("trans:%d, failed to propose, code:0x%x", pMgmt->transId, code);
    return code;
  }

S
Shengliang Guan 已提交
310
  return pMgmt->errCode;
311 312
}

313
void mndSyncStart(SMnode *pMnode) {
314 315
  SSyncMgmt *pMgmt = &pMnode->syncMgmt;
  syncSetMsgCb(pMgmt->sync, &pMnode->msgCb);
M
Minghao Li 已提交
316
  syncStart(pMgmt->sync);
317
  mInfo("vgId:1, sync started, id:%" PRId64, pMgmt->sync);
318 319
}

S
Shengliang Guan 已提交
320
void mndSyncStop(SMnode *pMnode) {
321
  taosWLockLatch(&pMnode->syncMgmt.lock);
S
Shengliang Guan 已提交
322
  if (pMnode->syncMgmt.transId != 0) {
S
Shengliang Guan 已提交
323
    pMnode->syncMgmt.transId = 0;
S
Shengliang Guan 已提交
324 325
    tsem_post(&pMnode->syncMgmt.syncSem);
  }
326
  taosWUnLockLatch(&pMnode->syncMgmt.lock);
S
Shengliang Guan 已提交
327
}
328

329
bool mndIsLeader(SMnode *pMnode) {
S
Shengliang Guan 已提交
330
  SSyncMgmt *pMgmt = &pMnode->syncMgmt;
331

S
Shengliang Guan 已提交
332
  if (!syncIsReady(pMgmt->sync)) {
333 334
    // get terrno from syncIsReady
    // terrno = TSDB_CODE_SYN_NOT_LEADER;
335 336 337
    return false;
  }

338
  if (!mndGetRestored(pMnode)) {
339 340 341 342 343
    terrno = TSDB_CODE_APP_NOT_READY;
    return false;
  }

  return true;
L
Liu Jicong 已提交
344
}