syncCommit.c 6.9 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/>.
 */

M
Minghao Li 已提交
16
#include "syncCommit.h"
M
Minghao Li 已提交
17
#include "syncIndexMgr.h"
M
Minghao Li 已提交
18
#include "syncInt.h"
M
Minghao Li 已提交
19
#include "syncRaftCfg.h"
M
Minghao Li 已提交
20
#include "syncRaftLog.h"
M
Minghao Li 已提交
21
#include "syncRaftStore.h"
M
Minghao Li 已提交
22
#include "syncUtil.h"
M
Minghao Li 已提交
23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46

// \* Leader i advances its commitIndex.
// \* This is done as a separate step from handling AppendEntries responses,
// \* in part to minimize atomic regions, and in part so that leaders of
// \* single-server clusters are able to mark entries committed.
// AdvanceCommitIndex(i) ==
//     /\ state[i] = Leader
//     /\ LET \* The set of servers that agree up through index.
//            Agree(index) == {i} \cup {k \in Server :
//                                          matchIndex[i][k] >= index}
//            \* The maximum indexes for which a quorum agrees
//            agreeIndexes == {index \in 1..Len(log[i]) :
//                                 Agree(index) \in Quorum}
//            \* New value for commitIndex'[i]
//            newCommitIndex ==
//               IF /\ agreeIndexes /= {}
//                  /\ log[i][Max(agreeIndexes)].term = currentTerm[i]
//               THEN
//                   Max(agreeIndexes)
//               ELSE
//                   commitIndex[i]
//        IN commitIndex' = [commitIndex EXCEPT ![i] = newCommitIndex]
//     /\ UNCHANGED <<messages, serverVars, candidateVars, leaderVars, log>>
//
M
Minghao Li 已提交
47
void syncMaybeAdvanceCommitIndex(SSyncNode* pSyncNode) {
M
Minghao Li 已提交
48 49
  syncIndexMgrLog2("==syncNodeMaybeAdvanceCommitIndex== pNextIndex", pSyncNode->pNextIndex);
  syncIndexMgrLog2("==syncNodeMaybeAdvanceCommitIndex== pMatchIndex", pSyncNode->pMatchIndex);
M
Minghao Li 已提交
50 51

  // update commit index
M
Minghao Li 已提交
52 53
  SyncIndex newCommitIndex = pSyncNode->commitIndex;
  for (SyncIndex index = pSyncNode->pLogStore->getLastIndex(pSyncNode->pLogStore); index > pSyncNode->commitIndex;
M
Minghao Li 已提交
54
       --index) {
M
Minghao Li 已提交
55
    bool agree = syncAgree(pSyncNode, index);
M
Minghao Li 已提交
56 57
    sTrace("syncMaybeAdvanceCommitIndex syncAgree:%d, index:%ld, pSyncNode->commitIndex:%ld", agree, index,
           pSyncNode->commitIndex);
M
Minghao Li 已提交
58
    if (agree) {
M
Minghao Li 已提交
59 60 61 62 63 64 65 66
      // term
      SSyncRaftEntry* pEntry = pSyncNode->pLogStore->getEntry(pSyncNode->pLogStore, index);
      assert(pEntry != NULL);

      // cannot commit, even if quorum agree. need check term!
      if (pEntry->term == pSyncNode->pRaftStore->currentTerm) {
        // update commit index
        newCommitIndex = index;
M
Minghao Li 已提交
67 68
        sTrace("syncMaybeAdvanceCommitIndex maybe to update, newCommitIndex:%ld commit, pSyncNode->commitIndex:%ld",
               newCommitIndex, pSyncNode->commitIndex);
69 70

        syncEntryDestory(pEntry);
M
Minghao Li 已提交
71
        break;
M
Minghao Li 已提交
72 73 74 75 76
      } else {
        sTrace(
            "syncMaybeAdvanceCommitIndex can not commit due to term not equal, pEntry->term:%lu, "
            "pSyncNode->pRaftStore->currentTerm:%lu",
            pEntry->term, pSyncNode->pRaftStore->currentTerm);
M
Minghao Li 已提交
77
      }
M
Minghao Li 已提交
78 79

      syncEntryDestory(pEntry);
M
Minghao Li 已提交
80 81
    }
  }
M
Minghao Li 已提交
82

M
Minghao Li 已提交
83 84 85
  if (newCommitIndex > pSyncNode->commitIndex) {
    SyncIndex beginIndex = pSyncNode->commitIndex + 1;
    SyncIndex endIndex = newCommitIndex;
M
Minghao Li 已提交
86

M
Minghao Li 已提交
87 88
    sTrace("syncMaybeAdvanceCommitIndex sync commit %ld", newCommitIndex);

M
Minghao Li 已提交
89
    // update commit index
M
Minghao Li 已提交
90
    pSyncNode->commitIndex = newCommitIndex;
M
Minghao Li 已提交
91

M
Minghao Li 已提交
92 93 94 95
    // call back Wal
    pSyncNode->pLogStore->updateCommitIndex(pSyncNode->pLogStore, pSyncNode->commitIndex);

    // execute fsm
M
Minghao Li 已提交
96 97 98 99 100
    if (pSyncNode->pFsm != NULL) {
      for (SyncIndex i = beginIndex; i <= endIndex; ++i) {
        if (i != SYNC_INDEX_INVALID) {
          SSyncRaftEntry* pEntry = pSyncNode->pLogStore->getEntry(pSyncNode->pLogStore, i);
          assert(pEntry != NULL);
M
Minghao Li 已提交
101

M
Minghao Li 已提交
102 103
          SRpcMsg rpcMsg;
          syncEntry2OriginalRpc(pEntry, &rpcMsg);
M
Minghao Li 已提交
104

M
Minghao Li 已提交
105
          if (pSyncNode->pFsm->FpCommitCb != NULL && syncUtilUserCommit(pEntry->originalRpcType)) {
M
Minghao Li 已提交
106 107 108 109 110 111
            SFsmCbMeta cbMeta;
            cbMeta.index = pEntry->index;
            cbMeta.isWeak = pEntry->isWeak;
            cbMeta.code = 0;
            cbMeta.state = pSyncNode->state;
            cbMeta.seqNum = pEntry->seqNum;
112 113 114 115 116 117 118 119 120 121 122
            cbMeta.term = pEntry->term;
            cbMeta.currentTerm = pSyncNode->pRaftStore->currentTerm;

            bool needExecute = true;
            if (pSyncNode->pSnapshot != NULL && cbMeta.index <= pSyncNode->pSnapshot->lastApplyIndex) {
              needExecute = false;
            }

            if (needExecute) {
              pSyncNode->pFsm->FpCommitCb(pSyncNode->pFsm, &rpcMsg, cbMeta);
            }
M
Minghao Li 已提交
123 124
          }

M
Minghao Li 已提交
125
          // config change
M
Minghao Li 已提交
126 127 128 129
          if (pEntry->originalRpcType == TDMT_VND_SYNC_CONFIG_CHANGE) {
            SSyncCfg newSyncCfg;
            int32_t  ret = syncCfgFromStr(rpcMsg.pCont, &newSyncCfg);
            ASSERT(ret == 0);
M
Minghao Li 已提交
130

M
Minghao Li 已提交
131
            syncNodeUpdateConfig(pSyncNode, &newSyncCfg);
M
Minghao Li 已提交
132 133 134 135 136
            if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
              syncNodeBecomeLeader(pSyncNode);
            } else {
              syncNodeBecomeFollower(pSyncNode);
            }
M
Minghao Li 已提交
137 138
          }

139 140 141
          // restore finish
          if (pEntry->index == pSyncNode->pLogStore->getLastIndex(pSyncNode->pLogStore)) {
            if (pSyncNode->restoreFinish == false) {
142 143 144
              if (pSyncNode->pFsm->FpRestoreFinish != NULL) {
                pSyncNode->pFsm->FpRestoreFinish(pSyncNode->pFsm);
              }
145
              pSyncNode->restoreFinish = true;
146
              sInfo("==syncMaybeAdvanceCommitIndex== restoreFinish set true %p vgId:%d", pSyncNode, pSyncNode->vgId);
147

148
              /*
149
              tsem_post(&pSyncNode->restoreSem);
150 151
              sInfo("==syncMaybeAdvanceCommitIndex== RestoreFinish tsem_post %p", pSyncNode);
              */
152 153 154
            }
          }

M
Minghao Li 已提交
155 156 157
          rpcFreeCont(rpcMsg.pCont);
          syncEntryDestory(pEntry);
        }
M
Minghao Li 已提交
158
      }
M
Minghao Li 已提交
159 160
    }
  }
M
Minghao Li 已提交
161 162 163
}

bool syncAgreeIndex(SSyncNode* pSyncNode, SRaftId* pRaftId, SyncIndex index) {
M
Minghao Li 已提交
164 165 166 167
  // I am leader, I agree
  if (syncUtilSameId(pRaftId, &(pSyncNode->myRaftId)) && pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
    return true;
  }
M
Minghao Li 已提交
168

M
Minghao Li 已提交
169 170
  // follower agree
  SyncIndex matchIndex = syncIndexMgrGetIndex(pSyncNode->pMatchIndex, pRaftId);
M
Minghao Li 已提交
171
  if (matchIndex >= index) {
M
Minghao Li 已提交
172
    return true;
M
Minghao Li 已提交
173
  }
M
Minghao Li 已提交
174

M
Minghao Li 已提交
175
  // not agree
M
Minghao Li 已提交
176
  return false;
M
Minghao Li 已提交
177 178 179 180 181 182 183 184 185 186 187 188 189
}

bool syncAgree(SSyncNode* pSyncNode, SyncIndex index) {
  int agreeCount = 0;
  for (int i = 0; i < pSyncNode->replicaNum; ++i) {
    if (syncAgreeIndex(pSyncNode, &(pSyncNode->replicasId[i]), index)) {
      ++agreeCount;
    }
    if (agreeCount >= pSyncNode->quorum) {
      return true;
    }
  }
  return false;
190
}