Raft 算法

func (rf *Raft) leaderElection() {
    // 增加 Term
    rf.currentTerm++
    // 变为 Candidate 状态
    rf.state = Candidate
    rf.votedFor = rf.me
    rf.persist()
    rf.resetElectionTimer()
    term := rf.currentTerm
    // 获得选票总数，首先投给自己
    voteCounter := 1
    // 当前节点最后一条 log
    lastLog := rf.log.lastLog()
    args := RequestVoteArgs{
        Term:         term,
        CandidateId:  rf.me,
        LastLogIndex: lastLog.Index,
        LastLogTerm:  lastLog.Term,
    }

    var becomeLeader sync.Once
    for serverId, _ := range rf.peers {
        if serverId != rf.me {
            // 发送 RequestVote
            go rf.candidateRequestVote(serverId, &args, &voteCounter, &becomeLeader)
        }
    }
}

func (rf *Raft) candidateRequestVote(serverId int, args *RequestVoteArgs,    voteCounter *int, becomeLeader *sync.Once) {
    reply := RequestVoteReply{}
    // 并发的发起 rpc 投票请求
    ok := rf.sendRequestVote(serverId, args, &reply)
    if !ok {
        return
    }
    // 同步的处理投票结果
    rf.mu.Lock()
    defer rf.mu.Unlock()
    // 存在更新的 Term，选举无效，更新 Term，转换为 Follower
    if reply.Term > args.Term {
        rf.setNewTerm(reply.Term)
        rf.setState(Follower)
        return
    }

    if !reply.VoteGranted {
        return
    }

    // 收到 Follower 投票
    *voteCounter++
    if *voteCounter > len(rf.peers)/2 &&
        rf.currentTerm == args.Term &&
        rf.state == Candidate {

        // 获得半数以上选票，且节点状态没变，转换为 Leader
        becomeLeader.Do(func() {
            rf.state = Leader
            lastLogIndex := rf.log.lastLog().Index
            for i, _ := range rf.peers {
                rf.nextIndex[i] = lastLogIndex + 1
                rf.matchIndex[i] = 0
            }
            rf.appendEntries(true)
        })
    }
}

func (rf *Raft) RequestVote(args *RequestVoteArgs, reply *RequestVoteReply) {
    rf.mu.Lock()
    defer rf.mu.Unlock()

    // 候选节点任期大于当前节点，变为 Follower
    if args.Term > rf.currentTerm {
        rf.setNewTerm(args.Term)
        rf.setState(Follower)
    }

    // 候选节点任期比当前节点小，忽略
    if args.Term < rf.currentTerm {
        reply.Term = rf.currentTerm
        reply.VoteGranted = false
        return
    }

    myLastLog := rf.log.lastLog()
    // 候选节点最后一条 log 任期比当前节点大
    // 或者任期相同，但是候选节点最后一条 logIndex 比当前节点大，说明候选节点比本节点日志新
    upToDate := args.LastLogTerm > myLastLog.Term ||
        (args.LastLogTerm == myLastLog.Term && args.LastLogIndex >= myLastLog.Index)
    // 每个节点每一轮选举只能投给一个节点
    if (rf.votedFor == -1 || rf.votedFor == args.CandidateId) && upToDate {
        reply.VoteGranted = true
        rf.votedFor = args.CandidateId
        rf.persist()
        rf.resetElectionTimer()
    } else {
        reply.VoteGranted = false
    }
    reply.Term = rf.currentTerm
}

func (rf *Raft) appendEntries() {
    lastLog := rf.log.lastLog()
    for peer, _ := range rf.peers {
        if peer == rf.me {
            rf.resetElectionTimer()
            continue
        }
        // 从 nextIndex 开始发送
        nextIndex := rf.nextIndex[peer]
        if nextIndex <= 0 {
            nextIndex = 1
        }
        // 上图（Raft 论文 Figure7）中的情况 cdf，包含额外的未提交的条目
        // 直接回退到 lastLog.Index
        if lastLog.Index+1 < nextIndex {
            nextIndex = lastLog.Index
        }
        // 前一条 log 的 index 和 Term，用于 Follower 校验日志是否一致
        prevLog := rf.log.at(nextIndex - 1)
        args := AppendEntriesArgs{
            Term:         rf.currentTerm,
            LeaderId:     rf.me,
            PrevLogIndex: prevLog.Index,
            PrevLogTerm:  prevLog.Term,
            Entries:      make([]Entry, lastLog.Index-nextIndex+1),
            // 目前 Leader 提交到状态机的 logIndex
            LeaderCommit: rf.commitIndex,
        }
        // 发送 [nextIndex, lastLog.Index]
        copy(args.Entries, rf.log.slice(nextIndex))
        go rf.leaderSendEntries(peer, &args)
    }
}

func (rf *Raft) AppendEntries(args *AppendEntriesArgs, reply *AppendEntriesReply) {
    rf.mu.Lock()
    defer rf.mu.Unlock()
    reply.Success = false
    reply.Term = rf.currentTerm

    // 忽略，Leader 的 Term 比自己小
    if args.Term < rf.currentTerm {
        return
    }

    // 比自己任期大，更新任期
    if args.Term > rf.currentTerm {
        rf.setNewTerm(args.Term)
        rf.setState(Follower)
        reply.Term = args.Term
    }

    // 重置 Election 定时器
    rf.resetElectionTimer()

    // Candidate 收到了 AppendEntries 请求，转换为 Follower
    if rf.state == Candidate {
        rf.setState(Follower)
    }

    // Follower 日志较少，Pre 处没有 log
    if rf.log.lastLog().Index < args.PrevLogIndex {
        reply.XTerm = -1
        reply.XIndex = -1
        reply.XLen = rf.log.len()
        return
    }

    // Pre 处有 log 但是 Term 不一样
    if rf.log.at(args.PrevLogIndex).Term != args.PrevLogTerm {
        // 获取 Follower 冲突位置的 Term
        xTerm := rf.log.at(args.PrevLogIndex).Term
        for xIndex := args.PrevLogIndex; xIndex > 0; xIndex-- {
            // XTerm 第一条 log 的 index
            if rf.log.at(xIndex-1).Term != xTerm {
                reply.XIndex = xIndex
                break
            }
        }
        reply.XTerm = xTerm
        reply.XLen = rf.log.len()
        return
    }

    // 匹配上了
    for idx, entry := range args.Entries {
        // 正常情况下 entry.Index 肯定是大于 lastLog.Index 的
        // 这里小于等于，且任期不一致，说明 Follower 有额外的未提交的日志，需要覆盖掉
        if entry.Index <= rf.log.lastLog().Index && rf.log.at(entry.Index).Term != entry.Term {
            // 截断 Index 后的日志
            rf.log.truncate(entry.Index)
            rf.persist()
        }
        // Follower 复制 log
        if entry.Index > rf.log.lastLog().Index {
            rf.log.append(args.Entries[idx:]...)
            rf.persist()
            break
        }
    }

    // 设置当前节点提交进度
    if args.LeaderCommit > rf.commitIndex {
        rf.commitIndex = min(args.LeaderCommit, rf.log.lastLog().Index)
        // 应用到状态机
        rf.apply()
    }
    reply.Success = true
}

func (rf *Raft) leaderSendEntries(serverId int, args *AppendEntriesArgs) {
    var reply AppendEntriesReply
    ok := rf.sendAppendEntries(serverId, args, &reply)
    if !ok {
        return
    }
    // 加锁，同步处理结果
    rf.mu.Lock()
    defer rf.mu.Unlock()
    // 返回的 Term 比 Leader 大，当前 Leader 节点过时了
    // 更新 Term 转换为 Follower
    if reply.Term > rf.currentTerm {
        rf.setNewTerm(reply.Term)
        rf.setState(Follower)
        return
    }
    // 发送到接收需要一段时间，确认状态，确认还是在自己的任期
    if args.Term == rf.currentTerm {
        // 没有冲突，校验成功
        if reply.Success {
            // 已知的 Follower 复制到的位置
            match := args.PrevLogIndex + len(args.Entries)
            next := match + 1
            rf.nextIndex[serverId] = max(rf.nextIndex[serverId], next)
            rf.matchIndex[serverId] = max(rf.matchIndex[serverId], match)
        } else {
            if reply.XTerm == -1 {
                // 有冲突，且 Follower 日志少，下一次从 XLen 开始
                // i: 1    2    3   4   5
                //        XLen
                // F: 4  
                // L: 4    6    6   6
                //             Pre     Next
                rf.nextIndex[serverId] = reply.XLen
            } else {
                // 找到 Leader 中 XTerm 中后一条日志的 Index，从这条再开始
                lastLogInXTerm := rf.findNextLogInTerm(reply.XTerm)
                if lastLogInXTerm > 0 {
                    // i: 1    2    3    4    5
                    // F: 4    4    4
                    // L: 4    6    6    6
                    //       Last  Pre       Next
                    rf.nextIndex[serverId] = lastLogInXTerm
                } else {
                    // Leader 没有 XTerm 直接从 XIndex 开始
                    // i: 1    2    3    4    5
                    //      XIndex
                    // F: 4    5    5
                    // L: 4    6    6   6
                    //             Pre      Next
                    rf.nextIndex[serverId] = reply.XIndex
                }
            }
        }
        // 提交 log
        rf.leaderCommit()
    }
}

func (rf *Raft) leaderCommit() {
    // 确认状态
    if rf.state != Leader {
        return
    }

    // 从上次 commitIndex 开始提交
    for n := rf.commitIndex + 1; n <= rf.log.lastLog().Index; n++ {
        // 只关心自己任期内的 log 是否复制到多数节点
        // 当前任期内日志的提交后，上一任的间接的也就提交了
        if rf.log.at(n).Term != rf.currentTerm {
            continue
        }
        counter := 1
        for serverId := 0; serverId < len(rf.peers); serverId++ {
            // follower 已经复制的 Index >= n
            if serverId != rf.me && rf.matchIndex[serverId] >= n {
                counter++
            }
            // 超过半数已经复制了 log[n]
            if counter > len(rf.peers)/2 {
                // 提交日志，并且应用到状态机
                rf.commitIndex = n
                rf.apply()
                break
            }
        }
    }
}

func (rf *Raft) Start(command interface{}) (int, int, bool) {
    rf.mu.Lock()
    defer rf.mu.Unlock()
    if rf.state != Leader {
        return -1, rf.currentTerm, false
    }
    index := rf.log.lastLog().Index + 1
    term := rf.currentTerm

    log := Entry{
        Command: command,
        Index:   index,
        Term:    term,
    }
    rf.log.append(log)
    rf.persist()

    return index, term, true
}