前言

最近通关了 MIT 6.S081，写 Lab 和做一个新的需求的流程差不多，新手先从一个小需求开始熟悉代码逻辑。课程的难度坡度设计的很缓慢，帮我一点点把 OS 的代码串起来，通关这门课之后也对操作系统有了一个大致的了解。

接下来我想学习下分布式系统的经典入门课程 MIT 6.824 ，正好看到木鸟老师(www.zhihu.com/people/qtmu… —— 从零实现分布式 KV。这门课是基于 MIT 6.824 的lab，手把手带你看论文写代码。所以这篇文章的主要内容是我的课程心得，介绍什么是Raft，并实现一个基本的框架。

这篇文章只讨论怎么实现 Leader 选举部分，代码在这：github.com/Maricaya/ra…

什么是 Raft

要想写出一个共识性算法 Raft，我们就要先知道什么是 Raft，为什么要有 Raft。了解这个最好的方式就是直接看论文 Abstract部分 raft.github.io/raft.pdf。

Raft is a consensus algorithm for managing a replicated log.

Raft 是一种用于管理复制日志的共识算法。

那什么是共识算法？

顾名思义，共识算法的目的就是让一群节点有一致的想法。

我举个例子你就明白了，想象一下一群小朋友在一起玩一个游戏，但是每个人都有不同的想法。有些人认为应该做A，有些人认为应该做B，而另一些人认为应该做C。如果没有一个方法来做出最终的决定，大家可能会陷入争吵，游戏就无法进行下去了。而共识算法就是一个规则，能够帮我们在一群人中做出最终的决定，选择哪个游戏规则。

那什么是管理复制日志呢？

在共识算法中，有时候我们需要记录一系列事件或者操作，就像在一本日记中一样。这个记录被称为“日志”。但是，我们不只是要记录这些事件，我们还希望所有人都能看到相同的记录，就像大家在同一本日记上写字一样。我们继续看小朋友玩游戏的例子，在确定了游戏规则之后，每个人都有自己的游戏日志，记录游戏中有趣的事情。但是，我们希望看见所有人的游戏记录，而不仅仅是自己的。所以，我们需要「管理复制日志」，用这种方法来确保每个人都有相同的记录，就像每个人都在玩同一款游戏一样。

Raft 算法总览

我们把 Raft 算法分为正交的几个部分，分别来实现：

1. Leader选举：就像我们玩团队游戏时需要选择一个队长一样，在Raft中，每个节点（就像团队中的一个小朋友）都想成为队长，但只能有一个Leader。Raft规定了一种选举Leader的方式，通过这种方式，每个节点都有机会成为Leader，而其他节点会支持选举出的Leader。
2. 日志同步：在团队游戏中，大家需要记下一些重要的信息，比如完成任务的进度。在Raft中，每个节点都有一个日志，就像大家都有一张笔记本一样，用来记录重要的事情。当Leader选举完成后，大家要一起同步自己的日志，以确保所有节点都有相同的信息。
3. 状态持久化：有时候游戏要进行很长时间，但是突然停电了，大家的游戏进度就丢失了。在Raft中，为了防止这种情况发生，每个节点都会把自己的状态保存在一个安全的地方，就像是把游戏存档一样，这样即使断电了，之前的进度也不会丢失。
4. 日志压缩：有时候大家记了很多笔记，但是有些笔记已经不重要了，可以扔掉。在Raft中，为了节省空间，节点会定期把一些旧的日志删除，就像把不需要的笔记撕掉一样，这样就可以腾出更多的空间来记录新的重要信息。

这些就是Raft的几个主要部分，通过这些部分的配合，整个分布式系统就可以保持稳定和可靠地运行。

今天，我们只探讨 part A Leader选举怎么实现。

Part A Leader选举

首次选举的测试用例

我们来看第一个问题，如何选举Leader，我们可以先通过 TestInitialElectionPartA 来测试是否完成，再添加心跳的逻辑。

选举逻辑

  怎么选出唯一的leader？

• Raft中的初始状态是所有节点都是 Follower。当某个 Follower 在一段时间内没有收到Leader 的心跳信号时（超时），它会认为当前没有Leader，于是就会变成 Candidate。Candidate 会向其他节点发送选举请求 RequestVote，请求它们投票给自己。如果Candidate获得了大多数节点的投票，包括自己的一票，它就会成为新的Leader。

  Candidate依据什么选出投票？

• Candidate的任期号（Term）：Candidate 会在请求投票时携带自己的任期号，如果其他节点发现 Candidate 的任期号比自己的任期号大，则会将投票给 Candidate，以认可其领导地位。

  • 每个节点的初始 Term 都为 0

  • 当节点由 Follower 变为 Candidate 时，任期 Term ++

  • 当节点发现 Candidate 的任期号比自己的任期号大，投票 Candidate，并把自己的任期号变为

• 下面这张图是 Raft 一致性算法的代码摘要。我们接下来会按照这张图来构造代码。

  定义 Raft

  在正式开始写选举逻辑前，我们先把 Raft 类型定义出来。

  我们先定义了一个 Role 的类型，它是一个字符串类型，用于表示 Raft 一致性算法中的三种角色：跟随者（Follower）、候选者（Candidate）和领导者（Leader）。

  接下来，根据 Figure 2 定义了一个名为 Raft 的结构体，用于表示单个 Raft 节点。它包含了几个字段：

• role 表示节点当前的角色（跟随者、候选者或领导者）。
• currentTerm 表示节点当前所处的任期。
• votedFor 表示节点在当前任期内投票给了哪个候选者，如果没有投票则为 -1。
• electionStart 表示选举开始的时间。
• electionTimeout 表示选举超时的时间间隔。

type Role string

const (
        Follower  Role = "Follower"
        Candidate Role = "Candidate"
        Leader    Role = "Leader"
)

// A Go object implementing a single Raft peer.
type Raft struct {
        // Your data here (PartA, PartB, PartC).
        // Look at the paper's Figure 2 for a description of what
        // state a Raft server must maintain.
        role        Role
        currentTerm int
        votedFor    int // -1 means vote for none

        electionStart   time.Time
        electionTimeout time.Duration // random
}

1. 初始化状态：

• 当 Raft 节点启动时，所有节点的角色都是跟随者（Follower）。
• 节点初始化自己的任期（currentTerm）为 0，并设置 votedFor = -1 表示没有投票。

func Make(peers []*labrpc.ClientEnd, me int,
        persister *Persister, applyCh chan ApplyMsg) *Raft {
        rf := &Raft{}
        rf.peers = peers
        rf.persister = persister
        rf.me = me

        // Your initialization code here (PartA, PartB, PartC).
        rf.role = Follower
        rf.currentTerm = 0
        rf.votedFor = -1
        // initialize from state persisted before a crash
        rf.readPersist(persister.ReadRaftState())

        // start electionTicker goroutine to start elections
        go rf.electionTicker()

        return rf
}

1. 选举超时：

• 在 Raft 中，每个节点都有一个随机的选举超时时间，通常在几百毫秒到数秒之间。
• 当一个节点的选举计时器超时时，它会转变为候选者（Candidate）状态，并且开始新一轮的选举过程。

func (rf *Raft) isElectionTimeoutLocked() bool {
        return time.Since(rf.electionStart) > rf.electionTimeout
}

1. 候选者状态：

• 选举超时后，节点将自己的角色从跟随者（Follower）变为候选者（Candidate）。
• 节点增加自己的任期（currentTerm++）。
• 节点给自己投票。

func (rf *Raft) becomeCandidateLocked() {
        if rf.role == Leader {
                LOG(rf.me, rf.currentTerm, DVote, "Leadeer can't become Candidate")
                return
        }

        LOG(rf.me, rf.currentTerm, DVote, "%s->Candidate, For T%d", rf.role, rf.currentTerm+1)
        rf.currentTerm++
        rf.role = Candidate
        rf.votedFor = rf.me
}

func (rf *Raft) electionTicker() {
        for rf.killed() == false {
                // todo Your code here (PartA)
                // Check if a leader election should be started.
                rf.mu.Lock()
                // 为了线程安全 -> lock
                if rf.role != Leader && rf.isElectionTimeoutLocked() {
                        rf.becomeCandidateLocked()
                        go rf.startElection(rf.currentTerm)
                }
                rf.mu.Unlock()

                // pause for a random amount of time between 50 and 350
                // milliseconds.
                ms := 50 + (rand.Int63() % 300)
                time.Sleep(time.Duration(ms) * time.Millisecond)
        }
}

1. 请求投票：

• 候选者向其他节点发送请求投票的 RPC。
• 请求投票中包含了候选者的任期和候选者的 ID。

func (rf *Raft) startElection(term int) bool {
        votes := 0
        askVoteFromPeer := func(peer int, args *RequestVoteArgs) {
                // 6.投票计数 ...
        }

        
        rf.mu.Lock()
        defer rf.mu.Unlock()

        // every time locked
        if rf.contextLostLocked(Candidate, term) {
                LOG(rf.me, rf.currentTerm, DVote, "Lost Candidate to %s, abort RequestVote", rf.role)
                return false
        }

        for peer := 0; peer < len(rf.peers); peer++ {
                if peer == rf.me {
                        votes++ // 投票给自己
                        continue
                }

                args := &RequestVoteArgs{
                        Term:        rf.currentTerm,
                        CandidateId: rf.me,
                }

                go askVoteFromPeer(peer, args)
        }
        return true
}

1. 投票过程：

• 收到请求投票的节点首先检查候选者的任期，如果候选者的任期比自己的小，则拒绝投票。
• 如果候选者的任期比自己的大或者相同，则将自己的投票给候选者，并更新自己的任期为候选者的任期。
• 如果一个节点投票给了某个候选者，则会重置自己的选举超时计时器

// field names must start with capital letters!
type RequestVoteArgs struct {
        // Your data here (PartA, PartB).
        Term        int
        CandidateId int
}

// example RequestVote RPC reply structure.
// field names must start with capital letters!
type RequestVoteReply struct {
        // Your data here (PartA).
        Term        int
        VoteGranted bool // true means candidate received vote
}

func (rf *Raft) becomeFollowerLocked(term int) {
        if term < rf.currentTerm {
                LOG(rf.me, rf.currentTerm, DError, "Can't become Follower, lower term: T%d", term)
                return
        }
        LOG(rf.me, rf.currentTerm, DLog, "%s->Follower, For T%v->T%v", rf.role, rf.currentTerm, term)
        rf.role = Follower
        if term > rf.currentTerm {
                rf.votedFor = -1
        }
        rf.currentTerm = term
}

func (rf *Raft) RequestVote(args *RequestVoteArgs, reply *RequestVoteReply) {
        // Your code here (PartA, PartB).
        rf.mu.Lock()
        defer rf.mu.Unlock()

        // align the term
        reply.Term = rf.currentTerm

        if args.Term < rf.currentTerm {
                LOG(rf.me, rf.currentTerm, DVote, "-> S%d, Reject voted, Higher term, T%d>T%d", args, Candidate, rf.currentTerm, args.Term)
                reply.VoteGranted = false
                return
        }

        // 检查候选者的任期是否比自己的任期更新
        if args.Term > rf.currentTerm {
                rf.becomeFollowerLocked(args.Term)
        }

        // check for votedFor
        if rf.votedFor != -1 {
                LOG(rf.me, rf.currentTerm, DVote, "-> S%d, Reject, Already voted to S%d", args.CandidateId, rf.votedFor)
                reply.VoteGranted = false
                return
        }
        reply.VoteGranted = true
        rf.votedFor = args.CandidateId
        rf.resetElectionTimerLocked()
        LOG(rf.me, rf.currentTerm, DVote, "-> S%d, vote granted", args.CandidateId)
}

1. 投票计数：

• 候选者收到大多数节点的投票后，就会成为领导者（Leader）。
• 大多数节点是指超过半数的节点，因为 Raft 的核心思想是要保证大多数节点的一致性。

func (rf *Raft) becomeLeaderLocked() {
        if rf.role != Candidate {
                LOG(rf.me, rf.currentTerm, DError, "Only Candidate can become Leader")
                return
        }
        LOG(rf.me, rf.currentTerm, DLeader, "%s->Leader, For T%d", rf.role, rf.currentTerm)
        rf.role = Leader
}

 func (rf *Raft) contextLostLocked(role Role, term int) bool {
        return !(rf.currentTerm == term && rf.role == role)
}
 
 askVoteFromPeer := func(peer int, args *RequestVoteArgs) {
                reply := &RequestVoteReply{}
                ok := rf.sendRequestVote(peer, args, reply)

                // handle the response
                rf.mu.Lock()
                defer rf.mu.Unlock()
                if !ok {
                        LOG(rf.me, rf.currentTerm, DDebug, "Ask vote from S%d, Lost or error", peer)
                        return
                }

                // align term
                if reply.Term > rf.currentTerm {
                        rf.becomeFollowerLocked(reply.Term)
                        return
                }

                // check the context
                if rf.contextLostLocked(Candidate, term) {
                        LOG(rf.me, rf.currentTerm, DVote, "Lost context, abort RequestVoteReply for S%d", peer)
                        return
                }

                // count the votes
                if reply.VoteGranted {
                        votes++
                }
                // vote 数量大于一半
                if votes > len(rf.peers)/2 {
                        rf.becomeLeaderLocked()
                        // 心跳逻辑
                        go rf.replicationTicker(term)
                }
        }

1. 成为 Leader 之后，发送心跳消息给其他节点

• 当候选者收到超过半数的投票后，就会成为领导者（Leader）。
• 成为领导者后，节点开始发送心跳消息给其他节点，并开始进行日志复制和处理客户端请求的工作。

type AppendEntriesArgs struct {
       Term     int
       LeaderId int
}
type AppendEntriesReply struct {
       Term    int
       Success bool
}

const (
        electionTimeoutMin time.Duration = 250 * time.Millisecond
        electionTimeoutMax time.Duration = 400 * time.Millisecond

        replicateInterval time.Duration = 250 * time.Millisecond
)

func (rf *Raft) resetElectionTimerLocked() {
        rf.electionStart = time.Now()
        randRange := int64(electionTimeoutMax - electionTimeoutMin)
        rf.electionTimeout = electionTimeoutMin + time.Duration(rand.Int63()%randRange)
}

func (rf *Raft) AppendEntries(args *AppendEntriesArgs, reply *AppendEntriesReply) {
       rf.mu.Lock()
       defer rf.mu.Unlock()

       // align the term
if args.Term < rf.currentTerm {
               LOG(rf.me, rf.currentTerm, DLog2, "<- S%d, Reject log, Higher term, T%d<T%d", args.LeaderId, args.Term, rf.currentTerm)
               return
       }

       if args.Term >= rf.currentTerm {
               rf.becomeFollowerLocked(args.Term)
       }

       rf.resetElectionTimerLocked()
}

func (rf *Raft) sendAppendEntries(server int, args *AppendEntriesArgs, reply *AppendEntriesReply) bool {
       ok := rf.peers[server].Call("Raft.AppendEntries", args, reply)
       return ok
}

func (rf *Raft) startReplication(term int) bool {
       replicateToPeer := func(peer int, args *AppendEntriesArgs) {
               reply := &AppendEntriesReply{}
               ok := rf.sendAppendEntries(peer, args, reply)

               rf.mu.Lock()
               defer rf.mu.Unlock()

               if !ok {
                       LOG(rf.me, rf.currentTerm, DLog, "-> S%d, Lost or crashed", peer)
                       return
               }

               // align the term
if reply.Term > rf.currentTerm {
                       rf.becomeFollowerLocked(reply.Term)
                       return
               }
       }

       rf.mu.Lock()
       defer rf.mu.Unlock()

       if rf.contextLostLocked(Leader, term) {
               LOG(rf.me, rf.currentTerm, DLog, "Lost leader[%d] to %s[T%d]", rf.role, rf.currentTerm)
               return false
       }

       for peer := 0; peer < len(rf.peers); peer++ {
               if peer == rf.me {
                       continue
               }

               args := &AppendEntriesArgs{
                       Term:     rf.currentTerm,
                       LeaderId: rf.me,
               }

               go replicateToPeer(peer, args)

       }

       return true
}

// could only replicate in the given term
func (rf *Raft) replicationTicker(term int) {
       for rf.killed() == false {
               ok := rf.startReplication(term)
               if !ok {
                       break
               }

               rf.mu.Lock()

               rf.mu.Unlock()

               time.Sleep(replicateInterval)
       }
}

最后，完成 GetState

// 完成 GetState
func (rf *Raft) GetState() (int, bool) {
    rf.mu.Lock()
    defer rf.mu.Unlock()

    term := rf.currentTerm
    isleader := rf.role == Leader

return term, isleader
}

测试

由于本次代码涉及到多机日志混杂以及复杂的网络环境，因此使用单步调试就是一个难点。还好 TA 给我们提供了工具，参考这篇文章： blog.josejg.com/debugging-p… 和我的代码，可以进行直观的log打印。

测试结果

测试用例

基于缓冲区 channel 并发循环测试 100 遍通过图