手撸golang etcd raft协议之9

缘起

最近阅读 [云原生分布式存储基石：etcd深入解析] (杜军 , 2019.1)
本系列笔记拟采用golang练习之

raft分布式一致性算法

分布式存储系统通常会通过维护多个副本来进行容错，
以提高系统的可用性。
这就引出了分布式存储系统的核心问题——如何保证多个副本的一致性？

Raft算法把问题分解成了四个子问题：
1. 领袖选举（leader election）、
2. 日志复制（log replication）、
3. 安全性（safety）
4. 成员关系变化（membership changes）
这几个子问题。

目标

• 根据raft协议，实现高可用分布式强一致的kv存储

子目标（Day 9）

• Leader状态下的逻辑处理

设计

• tLeaderState: 实现Leader状态的raft状态机处理。事件驱动的逻辑编排，读写分离的字段管理。

tLeaderState.go

(未完成)实现Leader状态的raft状态机处理。事件驱动的逻辑编排，读写分离的字段管理。

// tLeaderState.go
// gitee: https://gitee.com/ioly/learning.gooop

package lsm

import (
    "learning/gooop/etcd/raft/roles"
    "learning/gooop/etcd/raft/rpc"
    "learning/gooop/etcd/raft/timeout"
    "sync"
    "time"
)

// tLeaderState presents a leader node
type tLeaderState struct {
    tEventDrivenModel

    context    iRaftStateContext
    mInitOnce  sync.Once
    mStartOnce sync.Once

    // update: leInit / leLeaderHeartbeat
    mTerm int64

    // update: leInit / leDisposing
    mDisposedFlag bool

    // update: leVoteToCandidate
    mVotedTerm int64
    mVotedCandidateID string
    mVotedTimestamp int64
}

// trigger: init()
// args: empty
const leInit = "leader.init"

// trigger: Start()
// args: empty
const leStart = "leader.Start"

// trigger: whenNewLeaderAnnouncedThenSwitchToFollower
// args: empty
const leDiposing = "leader.Disposing"

// trigger : Heartbeat() / AppendLog()
// args: term int64
const leNewLeaderAnnounced = "leader.NewLeaderAnnounced"


// trigger: RequestVote()
// args: *rpc.RequestVoteCmd
const leBeforeRequestVote = "leader.BeforeRequestVote"

// trigger:
// args: *rpc.RequestVoteCmd
const leVoteToCandidate = "leader.VoteToCandidate"


func newLeaderState(ctx iRaftStateContext, term int64) IRaftState {
    it := new(tLeaderState)
    it.init(ctx, term)
    return it
}

func (me *tLeaderState) init(ctx iRaftStateContext, term int64) {
    me.mInitOnce.Do(func() {
        me.context = ctx
        me.mTerm = term

        me.initEventHandlers()
        me.raise(leInit)
    })
}

func (me *tLeaderState) initEventHandlers() {
    // write only logic
    me.hookEventsForDisposedFlag()
    me.hookEventsForVotedTerm()

    // read only logic
    me.hook(leStart,
        me.whenStartThenBeginHeartbeatToOthers)
    me.hook(leNewLeaderAnnounced,
        me.whenNewLeaderAnnouncedThenSwitchToFollower)
}


func (me *tLeaderState) hookEventsForDisposedFlag() {
    me.hook(leInit, func(e string, args ...interface{}) {
        me.mDisposedFlag = false
    })

    me.hook(leDiposing, func(e string, args ...interface{}) {
        me.mDisposedFlag = true
    })
}

func (me *tLeaderState) hookEventsForVotedTerm() {
    me.hook(leBeforeRequestVote, func(e string, args ...interface{}) {
        // check last vote timeout
        if me.mVotedTerm == 0 {
            return
        }

        if time.Duration(time.Now().UnixNano() - me.mVotedTimestamp)*time.Nanosecond >= timeout.ElectionTimeout {
            me.mVotedTerm = 0
            me.mVotedTimestamp = 0
            me.mVotedCandidateID = ""
        }
    })

    me.hook(leVoteToCandidate, func(e string, args ...interface{}) {
        // after vote to candidate
        cmd := args[0].(*rpc.RequestVoteCmd)
        me.mVotedTerm = cmd.Term
        me.mVotedCandidateID = cmd.CandidateID
        me.mVotedTimestamp = time.Now().UnixNano()
    })
}



func (me *tLeaderState) Heartbeat(cmd *rpc.HeartbeatCmd, ret *rpc.HeartbeatRet) error {
    // check term
    if cmd.Term <= me.mTerm {
        ret.Code = rpc.HBTermMismatch
        return nil
    }

    // new leader
    me.raise(leNewLeaderAnnounced, cmd.Term)

    // return ok
    ret.Code = rpc.HBOk
    return nil
}

func (me *tLeaderState) AppendLog(cmd *rpc.AppendLogCmd, ret *rpc.AppendLogRet) error {
    // check term
    if cmd.Term <= me.mTerm {
        ret.Code = rpc.ALTermMismatch
        return nil
    }

    // new leader
    me.raise(leNewLeaderAnnounced, cmd.Term)

    // return ok
    ret.Code = rpc.ALInternalError
    return nil
}

func (me *tLeaderState) CommitLog(cmd *rpc.CommitLogCmd, ret *rpc.CommitLogRet) error {
    // just ignore
    ret.Code = rpc.CLInternalError
    return nil
}

func (me *tLeaderState) RequestVote(cmd *rpc.RequestVoteCmd, ret *rpc.RequestVoteRet) error {
    me.raise(leBeforeRequestVote, cmd)

    // check voted term
    if cmd.Term < me.mVotedTerm {
        ret.Code = rpc.RVTermMismatch
        return nil
    }

    if cmd.Term == me.mVotedTerm {
        if me.mVotedCandidateID != "" && me.mVotedCandidateID != cmd.CandidateID {
            // already vote another
            ret.Code = rpc.RVVotedAnother
            return nil
        } else {
            // already voted
            ret.Code = rpc.RVOk
            return nil
        }
    }

    if cmd.Term > me.mVotedTerm {
        // new term, check log
        if cmd.LastLogIndex >= me.context.Store().LastCommittedIndex() {
            // good log
            me.raise(leVoteToCandidate, cmd)
            ret.Code = rpc.RVOk

        } else {
            // bad log
            ret.Code = rpc.RVLogMismatch
        }

        return nil
    }

    // should not reach here
    ret.Code = rpc.RVTermMismatch
    return nil
}

func (me *tLeaderState) Role() roles.RaftRole {
    return roles.Leader
}

func (me *tLeaderState) Start() {
    me.mStartOnce.Do(func() {
        me.raise(leStart)
    })
}

func (me *tLeaderState) whenStartThenBeginHeartbeatToOthers(_ string, _ ...interface{}) {
    // todo: fixme
    panic("implements me")
}

func (me *tLeaderState) whenNewLeaderAnnouncedThenSwitchToFollower(_ string, args ...interface{}) {
    me.raise(leDiposing)

    term := args[0].(int64)
    me.context.HandleStateChanged(newFollowerState(me.context, term))
}

（未完待续）