6.5840 実験 2a —— リーダー選出

type Raft struct {
    ...

    // 状態
    Status ServerStatus
    // コミット済みログを外部に渡すチャネル
    ApplyCh chan ApplyMsg

    /***** 全サーバーが持つ永続状態 *****/
    // CurrentTerm 現在の最大任期。起動時は 0 で単調増加
    CurrentTerm int
    // VotedFor 現任期で投票した候補者 ID。未投票は -1
    VotedFor int
    // Logs ログエントリ。各エントリは状態機械のコマンドとリーダーが受け取った任期を含む。インデックスは 1 始まり
    Logs []*LogEntry

    /***** 全サーバーが持つ可変状態 *****/
    // CommitIndex 既知の最大のコミット予定ログインデックス。起動時は 0 で単調増加
    CommitIndex uint64
    // LastApplied 最大の適用済みログインデックス。起動時は 0 で単調増加
    LastApplied uint64

    /******* リーダーが持つ可変状態。選挙後に初期化 *******/
    // NextIndex 各サーバーに次に送信するログエントリのインデックス。リーダーの最後のログインデックス +1 で初期化
    NextIndex []uint64
    // MatchIndex 各サーバーが複製済みの最高ログエントリ。0 で初期化し単調増加
    MatchIndex []uint64

    // タイマー
    electionTimer  *time.Timer
    heartbeatTimer *time.Timer

    // RPC リクエスト処理用チャネル
    requestVoteChan   chan RequestVoteMsg
    appendEntriesChan chan AppendEntriesMsg
}

func (rf *Raft) startElection() {
    if rf.Status == Leader {
        // リーダーは新たな選挙を開始しない
        return
    }
    rf.CurrentTerm += 1
    // fmt.Printf("server %d start election for term %d\n", rf.me, rf.CurrentTerm)
    rf.Status = Candidate
    rf.VotedFor = rf.me
    args := RequestVoteArgs{
        Term:         rf.CurrentTerm,
        CandidateId:  rf.me,
        LastLogIndex: len(rf.Logs) - 1,
    }
    if len(rf.Logs) != 0 {
        args.LastLogTerm = rf.Logs[len(rf.Logs)-1].Term
    }
    meta := ElectionMeta{
        term: rf.CurrentTerm,
        yeas: 1,
        nays: 0,
    }
    for peer := range rf.peers {
        if peer == rf.me {
            continue
        }
        go rf.sendRequestVoteRoutine(peer, args, &meta)
    }
}

/********* 投票依頼受信側関連メソッド *********/
// 投票依頼 RPC 受信入口
func (rf *Raft) RequestVote(args *RequestVoteArgs, reply *RequestVoteReply) {
    msg := RequestVoteMsg{
        req: args,
        ok:  make(chan RequestVoteReply),
    }
    rf.requestVoteChan <- msg
    resp := <-msg.ok
    *reply = resp
}

// メインゴルーチンで投票依頼を処理
func (rf *Raft) handleRequestVote(msg RequestVoteMsg) {
    req := msg.req
    if req.Term < rf.CurrentTerm ||
        (req.Term == rf.CurrentTerm && rf.VotedFor != -1 && rf.VotedFor != req.CandidateId) {
        msg.ok <- RequestVoteReply{
            Term:        rf.CurrentTerm,
            VoteGranted: false,
        }
        return
    }
    rf.rpcTermCheck(req.Term)
    rf.VotedFor = req.CandidateId
    resetTimer(rf.electionTimer, RandomizedElectionTimeout())
    // fmt.Printf("server %d vote for server %d for term %d\n", rf.me, msg.req.CandidateId, req.Term)
    msg.ok <- RequestVoteReply{
        Term:        rf.CurrentTerm,
        VoteGranted: true,
    }
}

// メインゴルーチンで投票依頼の返答を処理
func (rf *Raft) handleRequestVoteRes(msg RequestVoteResMsg) {
    meta := msg.meta
    if rf.Status != Candidate {
        return
    }
    if rf.CurrentTerm != meta.term {
        return
    }
    if msg.resp.VoteGranted {
        meta.yeas++
        if meta.yeas > len(rf.peers)/2 {
            // fmt.Printf("server %d become leader for term %d\n", rf.me, rf.CurrentTerm)
            rf.Status = Leader
            resetTimer(rf.heartbeatTimer, FixedHeartbeatTimeout())
            rf.broadcastHeartbeat()
        }
    } else {
        meta.nays++
        rf.rpcTermCheck(msg.resp.Term)
        if meta.nays > len(rf.peers)/2 {
            // 反対票が過半数を超えた場合、その任期の選挙は失敗。ほかの候補者に投票可能
            rf.VotedFor = -1
        }
    }
}

// 追加 RPC リクエスト
type AppendEntriesArgs struct {
    // Term リーダーの任期
    Term int
    // LeaderID Follower はクライアントリクエストをリーダーにリダイレクト可能
    LeaderID int
    // PrevLogIndex 新しいログエントリの直前のログインデックス
    PrevLogIndex int
    // PrevLogTerm 直前のログエントリの任期
    PrevLogTerm int
    // Entries 保存すべきログエントリ。ハートビートは空
    Entries []*LogEntry
    // LeaderCommit リーダーの CommitIndex
    LeaderCommit int
}

// 追加 RPC レスポンス
type AppendEntriesReply struct {
    // Term Follower の現在任期
    Term int
    // Success PrevLogIndex と PrevLogTerm が一致すれば true
    Success bool
}

// 追加要求 RPC 送信入口
func (rf *Raft) sendAppendEntries(server int, args *AppendEntriesArgs, reply *AppendEntriesReply) bool {
    ok := rf.peers[server].Call("Raft.AppendEntries", args, reply)
    return ok
}