缘起

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

raft分布式一致性算法

  1. 分布式存储系统通常会通过维护多个副本来进行容错，
  2. 以提高系统的可用性。
  3. 这就引出了分布式存储系统的核心问题——如何保证多个副本的一致性？
  5. Raft算法把问题分解成了四个子问题：
  6. 1. 领袖选举（leader election）、
  7. 2. 日志复制（log replication）、
  8. 3. 安全性（safety）
  9. 4. 成员关系变化（membership changes）
  10. 这几个子问题。

目标

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

子目标（Day 6）

  • 大幅重构，提升代码的可理解/可管理性：

    • 基于事件驱动的逻辑编排，重构Follower和Candidate状态下的实现
    • 将字段状态的管理，实行读写分离。没看错，代码也是可以”读写分离”的 ^_^

      设计

  • random：为各种超时时间添加随机性

  • tFollowerState：基于事件驱动重构Follower状态的逻辑编排，各字段实施读写分离管理
  • tCandidateState：基于事件驱动重构Candidate状态的逻辑编排，各字段实施读写分离管理

random.go

为各种超时时间添加随机性

  1. package lsm
  3. import (
  4.     "math/rand"
  5.     "time"
  6. )
  8. // fnRandomizeInt64 returns int64 value from v to v*1.3
  9. func fnRandomizeInt64(v int64) int64 {
  10.     return v + v * gRand.Int63n(30) / 100
  11. }
  13. // fnRandomizeDuration returns duration value from v to v*1.3
  14. func fnRandomizeDuration(v time.Duration) time.Duration {
  15.     i := int64(v)
  16.     return time.Duration(fnRandomizeInt64(i))
  17. }
  19. var gRand = rand.New(rand.NewSource(time.Now().UnixNano()))

tFollowerState.go

基于事件驱动重构Follower状态的逻辑编排，各字段实施读写分离管理

  1. package lsm
  3. import (
  4.     "learning/gooop/etcd/raft/roles"
  5.     "learning/gooop/etcd/raft/rpc"
  6.     "learning/gooop/etcd/raft/timeout"
  7.     "sync"
  8.     "time"
  9. )
  11. // tFollowerState presents a follower node
  12. type tFollowerState struct {
  13.     tEventDrivenModel
  15.     context IRaftLSM
  16.     mInitOnce    sync.Once
  17.     mStartOnce   sync.Once
  19.     // update: feInit / feLeaderHeartbeat
  20.     mTerm int64
  22.     // update: feInit / feLeaderHeartbeat
  23.     mLeaderHeartbeatTimestamp int64
  25.     // update: feLeaderHeartbeat
  26.     mLeaderID string
  28.     // update: feCandidateRequestVote / feVoteToCandidate
  29.     mLastVotedTerm int64
  31.     // update: feCandidateRequestVote / feVoteToCandidate
  32.     mLastVotedCandidateID string
  34.     // update: feCandidateRequestVote / feVoteToCandidate
  35.     mLastVotedTimestamp int64
  37.     // update: feInit / feDisposing
  38.     mDiseposedFlag bool
  39. }
  41. // trigger: init()
  42. // args: empty
  43. const feInit = "follower.init"
  45. // trigger: Start()
  46. // args: empty
  47. const feStart = "follower.Start"
  49. // trigger: Heartbeat()
  50. // args: rpc.HeartbeatCmd
  51. const feLeaderHeartbeat = "follower.LeaderHeartbeat"
  53. // trigger: whenStartThenBeginWatchLeaderTimeout()
  54. // args: empty
  55. const feLeaderHeartbeatTimeout = "follower.LeaderHeartbeatTimeout"
  57. // trigger: RequestVote()
  58. // args: rpc.RequestVoteCmd
  59. const feCandidateRequestVote = "candidate.RequestVote"
  61. // trigger: RequestVote()
  62. // args: rpc.RequestVoteCmd
  63. const feVoteToCandidate = "follower.CandidateRequestVote"
  65. // trigger: whenLeaderHeartbeatTimeoutThenSwitchToCandidateState
  66. const feDisposing = "follower.Disposing"
  68. func newFollowerState(ctx IRaftLSM) IRaftState {
  69.     it := new(tFollowerState)
  70.     it.init(ctx)
  71.     return it
  72. }
  74. func (me *tFollowerState) init(ctx IRaftLSM) {
  75.     me.mInitOnce.Do(func() {
  76.         me.context = ctx
  77.         me.initEventHandlers()
  78.     })
  79. }
  81. func (me *tFollowerState) initEventHandlers() {
  82.     // write only logic
  83.     me.hookEventsForTerm()
  84.     me.hookEventsForLeaderHeartbeatTimestamp()
  85.     me.hookEventsForLeaderID()
  86.     me.hookEventsForLastVotedTerm()
  87.     me.hookEventsForLastVotedCandicateID()
  88.     me.hookEventsForLastVotedTimestamp()
  89.     me.hookEventsForDisposedFlag()
  91.     // read only logic
  92.     me.hook(feStart,
  93.         me.whenStartThenBeginWatchLeaderTimeout)
  94.     me.hook(feLeaderHeartbeatTimeout,
  95.         me.whenLeaderHeartbeatTimeoutThenSwitchToCandidateState)
  96. }
  98. // hookEventsForTerm maintains field: mTerm
  99. // update : feInit / feLeaderHeartbeat
  100. func (me *tFollowerState) hookEventsForTerm() {
  101.     me.hook(feInit, func(e string, args ...interface{}) {
  102.         me.mTerm = me.context.store().LastCommittedTerm()
  103.     })
  105.     me.hook(feLeaderHeartbeat, func(e string, args ...interface{}) {
  106.         cmd := args[0].(*rpc.HeartbeatCmd)
  107.         me.mTerm = cmd.Term
  108.     })
  109. }
  111. // hookEventsForLeaderHeartbeatClock maintains field: mLeaderHeartbeatClock
  112. // update : feLeaderHeartbeat / feLeaderHeartbeatTimeout
  113. func (me *tFollowerState) hookEventsForLeaderHeartbeatTimestamp() {
  114.     me.hook(feInit, func(e string, args ...interface{}) {
  115.         me.mLeaderHeartbeatTimestamp = time.Now().UnixNano()
  116.     })
  118.     me.hook(feLeaderHeartbeat, func(e string, args ...interface{}) {
  119.         me.mLeaderHeartbeatTimestamp = time.Now().UnixNano()
  120.     })
  122.     me.hook(feLeaderHeartbeatTimeout, func(e string, args ...interface{}) {
  123.         me.mLeaderHeartbeatTimestamp = 0
  124.     })
  125. }
  127. // hookEventsForLeaderID maintains field: mLeaderID
  128. // update : feLeaderHeartbeat / feLeaderHeartbeatTimeout
  129. func (me *tFollowerState) hookEventsForLeaderID() {
  130.     me.hook(feLeaderHeartbeat, func(e string, args ...interface{}) {
  131.         cmd := args[0].(*rpc.HeartbeatCmd)
  132.         me.mLeaderID = cmd.LeaderID
  133.     })
  135.     me.hook(feLeaderHeartbeatTimeout, func(e string, args ...interface{}) {
  136.         me.mLeaderID = ""
  137.     })
  138. }
  141. // hookEventsForLastVotedTerm maintains field: mLastVotedTerm
  142. // update : feCandidateRequestVote / feVoteToCandidate
  143. func (me *tFollowerState) hookEventsForLastVotedTerm() {
  144.     me.hook(feCandidateRequestVote, func(e string, args ...interface{}) {
  145.         // before voting, check whether last vote timeout
  146.         now := time.Now().UnixNano()
  147.         if time.Duration(now - me.mLastVotedTimestamp) * time.Nanosecond >= fnRandomizeDuration(timeout.ElectionTimeout) {
  148.             // timeout, reset to empty
  149.             me.mLastVotedTerm = 0
  150.             me.mLastVotedCandidateID = ""
  151.             me.mLastVotedTimestamp = 0
  152.         }
  153.     })
  155.     me.hook(feVoteToCandidate, func(e string, args ...interface{}) {
  156.         cmd := args[0].(*rpc.RequestVoteCmd)
  157.         me.mLastVotedTerm = cmd.Term
  158.     })
  159. }
  161. // hookEventsForLastVotedCandicateID maintains field: mLastVotedCandidateID
  162. // update : feCandidateRequestVote / feVoteToCandidate
  163. func (me *tFollowerState) hookEventsForLastVotedCandicateID() {
  164.     me.hook(feCandidateRequestVote, func(e string, args ...interface{}) {
  165.         // before voting, check whether last vote timeout
  166.         now := time.Now().UnixNano()
  167.         if time.Duration(now - me.mLastVotedTimestamp) * time.Nanosecond >= fnRandomizeDuration(timeout.ElectionTimeout) {
  168.             // timeout, reset to empty
  169.             me.mLastVotedTerm = 0
  170.             me.mLastVotedCandidateID = ""
  171.             me.mLastVotedTimestamp = 0
  172.         }
  173.     })
  175.     me.hook(feVoteToCandidate, func(e string, args ...interface{}) {
  176.         cmd := args[0].(*rpc.RequestVoteCmd)
  177.         me.mLastVotedCandidateID = cmd.CandidateID
  178.     })
  179. }
  181. // hookEventsForLastVotedTimestamp maintains field: mLastVotedTimestamp
  182. // update : feCandidateRequestVote / feVoteToCandidate
  183. func (me *tFollowerState) hookEventsForLastVotedTimestamp() {
  184.     me.hook(feCandidateRequestVote, func(e string, args ...interface{}) {
  185.         // before voting, check whether last vote timeout
  186.         now := time.Now().UnixNano()
  187.         if time.Duration(now - me.mLastVotedTimestamp) * time.Nanosecond >= fnRandomizeDuration(timeout.ElectionTimeout) {
  188.             // timeout, reset to empty
  189.             me.mLastVotedTerm = 0
  190.             me.mLastVotedCandidateID = ""
  191.             me.mLastVotedTimestamp = 0
  192.         }
  193.     })
  195.     me.hook(feVoteToCandidate, func(e string, args ...interface{}) {
  196.         me.mLastVotedTimestamp = time.Now().UnixNano()
  197.     })
  198. }
  200. // hookEventsForDisposedFlag maintains field: mDisposedFlag
  201. // update: feInit / feDisposing
  202. func (me *tFollowerState) hookEventsForDisposedFlag() {
  203.     me.hook(feInit, func(e string, args ...interface{}) {
  204.         me.mDiseposedFlag = false
  205.     })
  207.     me.hook(feDisposing, func(e string, args ...interface{}) {
  208.         me.mDiseposedFlag = true
  209.     })
  210. }
  213. func (me *tFollowerState) Start() {
  214.     me.mStartOnce.Do(func() {
  215.         me.raise(feStart)
  216.     })
  217. }
  219. func (me *tFollowerState) whenStartThenBeginWatchLeaderTimeout(e string, args ...interface{}) {
  220.     go func() {
  221.         iCheckingTimeoutInterval := fnRandomizeDuration(timeout.HeartbeatTimeout / 3)
  222.         for range time.Tick(iCheckingTimeoutInterval) {
  223.             if me.mDiseposedFlag {
  224.                 return
  225.             }
  227.             now := time.Now().UnixNano()
  228.             iHeartbeatTimeoutNanos := fnRandomizeInt64(int64(timeout.HeartbeatTimeout / time.Nanosecond))
  230.             if now - me.mLeaderHeartbeatTimestamp >= iHeartbeatTimeoutNanos {
  231.                 me.raise(feLeaderHeartbeatTimeout)
  232.                 return
  233.             }
  234.         }
  235.     }()
  236. }
  238. func (me *tFollowerState) whenLeaderHeartbeatTimeoutThenSwitchToCandidateState(_ string, args ...interface{}) {
  239.     me.raise(feDisposing)
  240.     me.context.handleStateChanged(newCandidateState(me.context, me.mTerm + 1))
  241. }
  243. func (me *tFollowerState) Role() roles.RaftRole {
  244.     return roles.Follower
  245. }
  247. // Heartbeat leader to follower
  248. func (me *tFollowerState) Heartbeat(cmd *rpc.HeartbeatCmd, ret *rpc.HeartbeatRet) error {
  249.     // check term
  250.     if cmd.Term < me.mTerm {
  251.         // invalid leader
  252.         ret.Code = rpc.HBTermMismatch
  253.         ret.Term = me.mTerm
  254.         return nil
  255.     }
  257.     // raise LeaderHeartbeat
  258.     me.raise(feLeaderHeartbeat, cmd)
  260.     // return
  261.     ret.Code = rpc.HBOk
  262.     return nil
  263. }
  265. // AppendLog leader to follower
  266. func (me *tFollowerState) AppendLog(cmd *rpc.AppendLogCmd, ret *rpc.AppendLogRet) error {
  267.     ret.Term = me.mTerm
  269.     if cmd.Term < me.mTerm {
  270.         // invalid leader
  271.         ret.Code = rpc.ALTermMismatch
  272.         return nil
  273.     }
  275.     store := me.context.store()
  276.     entry := cmd.Entry
  278.     // check log: expecting appending action follows previous committing action
  279.     if entry.PrevIndex != store.LastCommittedIndex() || entry.PrevTerm != store.LastCommittedTerm() {
  280.         // check log
  281.         e, log := store.GetLog(entry.Index)
  282.         if e != nil {
  283.             ret.Code = rpc.ALInternalError
  284.             return nil
  285.         }
  287.         if log == nil || log.PrevIndex != entry.PrevIndex || log.PrevTerm != entry.PrevTerm {
  288.             // bad log
  289.             ret.Code = rpc.ALIndexMismatch
  290.             ret.PrevLogIndex = store.LastCommittedIndex()
  291.             ret.PrevLogTerm = store.LastCommittedTerm()
  292.             return nil
  293.         }
  295.         // good log, but old, just ignore it
  296.         ret.Code = rpc.ALOk
  297.         return nil
  298.     }
  300.     // good log
  301.     e := store.Append(entry)
  302.     if e != nil {
  303.         ret.Code = rpc.ALInternalError
  304.         return nil
  305.     } else {
  306.         ret.Code = rpc.ALOk
  307.         return nil
  308.     }
  309. }
  311. // CommitLog leader to follower
  312. func (me *tFollowerState) CommitLog(cmd *rpc.CommitLogCmd, ret *rpc.CommitLogRet) error {
  313.     store := me.context.store()
  314.     if cmd.Index != store.LastAppendedIndex() || cmd.Term != store.LastAppendedTerm() {
  315.         // bad index
  316.         ret.Code = rpc.CLLogNotFound
  317.         return nil
  318.     }
  320.     e := store.Commit(cmd.Index)
  321.     if e != nil {
  322.         ret.Code = rpc.CLInternalError
  323.         return nil
  324.     }
  326.     ret.Code = rpc.CLOk
  327.     return nil
  328. }
  330. // RequestVote candidate to follower
  331. func (me *tFollowerState) RequestVote(cmd *rpc.RequestVoteCmd, ret *rpc.RequestVoteRet) error {
  332.     // before voting
  333.     me.raise(feCandidateRequestVote, cmd)
  335.     // check term
  336.     if cmd.Term <= me.mTerm {
  337.         ret.Term = me.mTerm
  338.         ret.Code = rpc.RVTermMismatch
  339.         return nil
  340.     }
  342.     // check if already voted another
  343.     if me.mLastVotedTerm >= cmd.Term && me.mLastVotedCandidateID != "" && me.mLastVotedCandidateID != cmd.CandidateID {
  344.         ret.Code = rpc.RVVotedAnother
  345.         return nil
  346.     }
  348.     // check log index
  349.     if cmd.LastLogIndex < me.context.store().LastCommittedIndex() {
  350.         ret.Code = rpc.RVLogMismatch
  351.         return nil
  352.     }
  354.     // vote ok
  355.     me.raise(feVoteToCandidate, cmd)
  356.     ret.Term = cmd.Term
  357.     ret.Code = rpc.RVOk
  359.     return nil
  360. }

tCandidateState.go

基于事件驱动重构Candidate状态的逻辑编排，各字段实施读写分离管理

  1. package lsm
  3. import (
  4.     "learning/gooop/etcd/raft/roles"
  5.     "learning/gooop/etcd/raft/rpc"
  6.     "sync"
  7.     "time"
  8. )
  10. // tCandidateState presents a candidate node
  11. type tCandidateState struct {
  12.     tEventDrivenModel
  14.     context IRaftLSM
  15.     mInitOnce    sync.Once
  16.     mStartOnce   sync.Once
  18.     // update: init / ceElectionTimeout
  19.     mTerm int64
  21.     // update: ceInit / ceElectionTimeout / ceVoteToCandidate
  22.     mVotedTerm int64
  24.     // update: ceInit / ceElectionTimeout / ceVoteToCandidate
  25.     mVotedCandidateID string
  27.     // update: ceInit / ceElectionTimeout / ceVoteToCandidate
  28.     mVotedTimestamp int64
  29. }
  31. // trigger: init()
  32. // args: empty
  33. const ceInit = "candidate.init"
  35. // trigger: Start()
  36. // args: empty
  37. const ceStart = "candidate.Start"
  39. // trigger: whenStartThenWatchElectionTimeout()
  40. // args: empty
  41. const ceElectionTimeout = "candidate.ElectionTimeout"
  43. // trigger: Heartbeat() / AppendLog() / CommitLog()
  44. // args: empty
  45. const ceLeaderAnnounced = "candidate.LeaderAnnounced"
  47. // trigger: RequestVote()
  48. // args: *rpc.RequestVoteCmd
  49. const ceVoteToCandidate = "candidate.VoteToCandidate"
  51. // trigger: whenLeaderHeartbeatThenSwitchToFollower()
  52. // args: empty
  53. const ceDisposing = "candidate.Disposing"
  55. func newCandidateState(ctx IRaftLSM, term int64) IRaftState {
  56.     it := new(tCandidateState)
  57.     it.init(ctx, term)
  58.     return it
  59. }
  61. func (me *tCandidateState) init(ctx IRaftLSM, term int64) {
  62.     me.mInitOnce.Do(func() {
  63.         me.context = ctx
  64.         me.mTerm = term
  65.         me.initEventHandlers()
  67.         me.raise(ceInit)
  68.     })
  69. }
  72. func (me *tCandidateState) initEventHandlers() {
  73.     // write only logic
  74.     me.hookEventsForTerm()
  75.     me.hookEventsForVotedTerm()
  76.     me.hookEventsForVotedCandidateID()
  77.     me.hookEventsForVotedTimestamp()
  79.     // read only logic
  80.     me.hook(ceLeaderAnnounced,
  81.         me.whenLeaderAnnouncedThenSwitchToFollower)
  82.     me.hook(ceElectionTimeout,
  83.         me.whenElectionTimeoutThenRequestVoteAgain)
  84. }
  86. // hookEventsForTerm maintains field: mTerm
  87. // update: ceElectionTimeout
  88. func (me *tCandidateState) hookEventsForTerm() {
  89.     me.hook(ceElectionTimeout, func(e string, args ...interface{}) {
  90.         // when election timeout, term++ and request vote again
  91.         me.mTerm++
  92.     })
  93. }
  95. // hookEventsForVotedTerm maintains field: mVotedTerm
  96. // update: ceInit / ceElectionTimeout / ceVoteToCandidate
  97. func (me *tCandidateState) hookEventsForVotedTerm() {
  98.     me.hook(ceInit, func(e string, args ...interface{}) {
  99.         // initially, vote to itself
  100.         me.mVotedTerm = me.mTerm
  101.     })
  103.     me.hook(ceElectionTimeout, func(e string, args ...interface{}) {
  104.         // when timeout, reset to itself
  105.         me.mVotedTerm = me.mTerm
  106.     })
  108.     me.hook(ceVoteToCandidate, func(e string, args ...interface{}) {
  109.         // after vote to candidate
  110.         cmd := args[0].(*rpc.RequestVoteCmd)
  111.         me.mVotedTerm = cmd.Term
  112.     })
  113. }
  115. // hookEventsForVotedCandidateID maintains field: mVotedCandidateID
  116. // update: ceInit / ceElectionTimeout / ceVoteToCandidate
  117. func (me *tCandidateState) hookEventsForVotedCandidateID() {
  118.     me.hook(ceInit, func(e string, args ...interface{}) {
  119.         // initially, vote to itself
  120.         me.mVotedCandidateID = me.context.config().ID()
  121.     })
  123.     me.hook(ceElectionTimeout, func(e string, args ...interface{}) {
  124.         // when timeout, reset to itself
  125.         me.mVotedCandidateID = me.context.config().ID()
  126.     })
  128.     me.hook(ceVoteToCandidate, func(e string, args ...interface{}) {
  129.         // after vote to candidate
  130.         cmd := args[0].(*rpc.RequestVoteCmd)
  131.         me.mVotedCandidateID = cmd.CandidateID
  132.     })
  133. }
  135. func (me *tCandidateState) hookEventsForVotedTimestamp() {
  136.     me.hook(ceInit, func(e string, args ...interface{}) {
  137.         // initially, vote to itself
  138.         me.mVotedTimestamp = time.Now().UnixNano()
  139.     })
  141.     me.hook(ceElectionTimeout, func(e string, args ...interface{}) {
  142.         // when timeout, reset to itself
  143.         me.mVotedTimestamp = time.Now().UnixNano()
  144.     })
  146.     me.hook(ceVoteToCandidate, func(e string, args ...interface{}) {
  147.         // after vote to candidate
  148.         me.mVotedTimestamp = time.Now().UnixNano()
  149.     })
  150. }
  152. func (me *tCandidateState) Heartbeat(cmd *rpc.HeartbeatCmd, ret *rpc.HeartbeatRet) error {
  153.     // check term
  154.     if cmd.Term <= me.mTerm {
  155.         // bad leader
  156.         ret.Code = rpc.HBTermMismatch
  157.         return nil
  158.     }
  160.     // new leader
  161.     me.raise(ceLeaderAnnounced)
  163.     // return ok
  164.     ret.Code = rpc.HBOk
  165.     return nil
  166. }
  168. func (me *tCandidateState) AppendLog(cmd *rpc.AppendLogCmd, ret *rpc.AppendLogRet) error {
  169.     // check term
  170.     if cmd.Term <= me.mTerm {
  171.         // bad leader
  172.         ret.Code = rpc.ALTermMismatch
  173.         return nil
  174.     }
  176.     // new leader
  177.     me.raise(ceLeaderAnnounced)
  179.     // ignore and return
  180.     ret.Code = rpc.ALInternalError
  181.     return nil
  182. }
  184. func (me *tCandidateState) CommitLog(cmd *rpc.CommitLogCmd, ret *rpc.CommitLogRet) error {
  185.     // ignore and return
  186.     ret.Code = rpc.CLInternalError
  187.     return nil
  188. }
  190. func (me *tCandidateState) RequestVote(cmd *rpc.RequestVoteCmd, ret *rpc.RequestVoteRet) error {
  191.     // todo: fixme
  192.     panic("implements me")
  193. }
  195. func (me *tCandidateState) Role() roles.RaftRole {
  196.     return roles.Candidate
  197. }
  199. func (me *tCandidateState) Start() {
  200.     me.mStartOnce.Do(func() {
  201.         me.raise(feStart)
  202.     })
  203. }
  206. func (me *tCandidateState) whenLeaderAnnouncedThenSwitchToFollower(_ string, _ ...interface{}) {
  207.     me.raise(ceDisposing)
  208.     me.context.handleStateChanged(newFollowerState(me.context))
  209. }
  211. func (me *tCandidateState) whenElectionTimeoutThenRequestVoteAgain(_ string, _ ...interface{}) {
  212.     // todo: fixme
  213.     panic("implements me")
  214. }

(未完待续)