raft.go 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776
  1. // Copyright 2015 CoreOS, Inc.
  2. //
  3. // Licensed under the Apache License, Version 2.0 (the "License");
  4. // you may not use this file except in compliance with the License.
  5. // You may obtain a copy of the License at
  6. //
  7. // http://www.apache.org/licenses/LICENSE-2.0
  8. //
  9. // Unless required by applicable law or agreed to in writing, software
  10. // distributed under the License is distributed on an "AS IS" BASIS,
  11. // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12. // See the License for the specific language governing permissions and
  13. // limitations under the License.
  14. package raft
  15. import (
  16. "errors"
  17. "fmt"
  18. "math"
  19. "math/rand"
  20. "sort"
  21. "strings"
  22. pb "github.com/coreos/etcd/raft/raftpb"
  23. )
  24. // None is a placeholder node ID used when there is no leader.
  25. const None uint64 = 0
  26. const noLimit = math.MaxUint64
  27. var errNoLeader = errors.New("no leader")
  28. // Possible values for StateType.
  29. const (
  30. StateFollower StateType = iota
  31. StateCandidate
  32. StateLeader
  33. )
  34. // StateType represents the role of a node in a cluster.
  35. type StateType uint64
  36. var stmap = [...]string{
  37. "StateFollower",
  38. "StateCandidate",
  39. "StateLeader",
  40. }
  41. func (st StateType) String() string {
  42. return stmap[uint64(st)]
  43. }
  44. // Config contains the parameters to start a raft.
  45. type Config struct {
  46. // ID is the identity of the local raft. ID cannot be 0.
  47. ID uint64
  48. // peers contains the IDs of all nodes (including self) in
  49. // the raft cluster. It should only be set when starting a new
  50. // raft cluster.
  51. // Restarting raft from previous configuration will panic if
  52. // peers is set.
  53. // peer is private and only used for testing right now.
  54. peers []uint64
  55. // ElectionTick is the election timeout. If a follower does not
  56. // receive any message from the leader of current term during
  57. // ElectionTick, it will become candidate and start an election.
  58. // ElectionTick must be greater than HeartbeatTick. We suggest
  59. // to use ElectionTick = 10 * HeartbeatTick to avoid unnecessary
  60. // leader switching.
  61. ElectionTick int
  62. // HeartbeatTick is the heartbeat interval. A leader sends heartbeat
  63. // message to maintain the leadership every heartbeat interval.
  64. HeartbeatTick int
  65. // Storage is the storage for raft. raft generates entires and
  66. // states to be stored in storage. raft reads the persisted entires
  67. // and states out of Storage when it needs. raft reads out the previous
  68. // state and configuration out of storage when restarting.
  69. Storage Storage
  70. // Applied is the last applied index. It should only be set when restarting
  71. // raft. raft will not return entries to the application smaller or equal to Applied.
  72. // If Applied is unset when restarting, raft might return previous applied entries.
  73. // This is a very application dependent configuration.
  74. Applied uint64
  75. // MaxSizePerMsg limits the max size of each append message. Smaller value lowers
  76. // the raft recovery cost(initial probing and message lost during normal operation).
  77. // On the other side, it might affect the throughput during normal replication.
  78. // Note: math.MaxUint64 for unlimited, 0 for at most one entry per message.
  79. MaxSizePerMsg uint64
  80. // MaxInflightMsgs limits the max number of in-flight append messages during optimistic
  81. // replication phase. The application transportation layer usually has its own sending
  82. // buffer over TCP/UDP. Setting MaxInflightMsgs to avoid overflowing that sending buffer.
  83. // TODO (xiangli): feedback to application to limit the proposal rate?
  84. MaxInflightMsgs int
  85. }
  86. func (c *Config) validate() error {
  87. if c.ID == None {
  88. return errors.New("cannot use none as id")
  89. }
  90. if c.HeartbeatTick <= 0 {
  91. return errors.New("heartbeat tick must be greater than 0")
  92. }
  93. if c.ElectionTick <= c.HeartbeatTick {
  94. return errors.New("election tick must be greater than heartbeat tick")
  95. }
  96. if c.Storage == nil {
  97. return errors.New("storage cannot be nil")
  98. }
  99. if c.MaxInflightMsgs <= 0 {
  100. return errors.New("max inflight messages must be greater than 0")
  101. }
  102. return nil
  103. }
  104. type raft struct {
  105. pb.HardState
  106. id uint64
  107. // the log
  108. raftLog *raftLog
  109. maxInflight int
  110. maxMsgSize uint64
  111. prs map[uint64]*Progress
  112. state StateType
  113. votes map[uint64]bool
  114. msgs []pb.Message
  115. // the leader id
  116. lead uint64
  117. // New configuration is ignored if there exists unapplied configuration.
  118. pendingConf bool
  119. elapsed int // number of ticks since the last msg
  120. heartbeatTimeout int
  121. electionTimeout int
  122. rand *rand.Rand
  123. tick func()
  124. step stepFunc
  125. }
  126. func newRaft(c *Config) *raft {
  127. if err := c.validate(); err != nil {
  128. panic(err.Error())
  129. }
  130. raftlog := newLog(c.Storage)
  131. hs, cs, err := c.Storage.InitialState()
  132. if err != nil {
  133. panic(err) // TODO(bdarnell)
  134. }
  135. peers := c.peers
  136. if len(cs.Nodes) > 0 {
  137. if len(peers) > 0 {
  138. // TODO(bdarnell): the peers argument is always nil except in
  139. // tests; the argument should be removed and these tests should be
  140. // updated to specify their nodes through a snapshot.
  141. panic("cannot specify both newRaft(peers) and ConfState.Nodes)")
  142. }
  143. peers = cs.Nodes
  144. }
  145. r := &raft{
  146. id: c.ID,
  147. lead: None,
  148. raftLog: raftlog,
  149. // 4MB for now and hard code it
  150. // TODO(xiang): add a config arguement into newRaft after we add
  151. // the max inflight message field.
  152. maxMsgSize: c.MaxSizePerMsg,
  153. maxInflight: c.MaxInflightMsgs,
  154. prs: make(map[uint64]*Progress),
  155. electionTimeout: c.ElectionTick,
  156. heartbeatTimeout: c.HeartbeatTick,
  157. }
  158. r.rand = rand.New(rand.NewSource(int64(c.ID)))
  159. for _, p := range peers {
  160. r.prs[p] = &Progress{Next: 1, ins: newInflights(r.maxInflight)}
  161. }
  162. if !isHardStateEqual(hs, emptyState) {
  163. r.loadState(hs)
  164. }
  165. if c.Applied > 0 {
  166. raftlog.appliedTo(c.Applied)
  167. }
  168. r.becomeFollower(r.Term, None)
  169. nodesStrs := make([]string, 0)
  170. for _, n := range r.nodes() {
  171. nodesStrs = append(nodesStrs, fmt.Sprintf("%x", n))
  172. }
  173. raftLogger.Infof("newRaft %x [peers: [%s], term: %d, commit: %d, applied: %d, lastindex: %d, lastterm: %d]",
  174. r.id, strings.Join(nodesStrs, ","), r.Term, r.raftLog.committed, r.raftLog.applied, r.raftLog.lastIndex(), r.raftLog.lastTerm())
  175. return r
  176. }
  177. func (r *raft) hasLeader() bool { return r.lead != None }
  178. func (r *raft) softState() *SoftState { return &SoftState{Lead: r.lead, RaftState: r.state} }
  179. func (r *raft) q() int { return len(r.prs)/2 + 1 }
  180. func (r *raft) nodes() []uint64 {
  181. nodes := make([]uint64, 0, len(r.prs))
  182. for k := range r.prs {
  183. nodes = append(nodes, k)
  184. }
  185. sort.Sort(uint64Slice(nodes))
  186. return nodes
  187. }
  188. // send persists state to stable storage and then sends to its mailbox.
  189. func (r *raft) send(m pb.Message) {
  190. m.From = r.id
  191. // do not attach term to MsgProp
  192. // proposals are a way to forward to the leader and
  193. // should be treated as local message.
  194. if m.Type != pb.MsgProp {
  195. m.Term = r.Term
  196. }
  197. r.msgs = append(r.msgs, m)
  198. }
  199. // sendAppend sends RRPC, with entries to the given peer.
  200. func (r *raft) sendAppend(to uint64) {
  201. pr := r.prs[to]
  202. if pr.isPaused() {
  203. return
  204. }
  205. m := pb.Message{}
  206. m.To = to
  207. term, errt := r.raftLog.term(pr.Next - 1)
  208. ents, erre := r.raftLog.entries(pr.Next, r.maxMsgSize)
  209. if errt != nil || erre != nil { // send snapshot if we failed to get term or entries
  210. m.Type = pb.MsgSnap
  211. snapshot, err := r.raftLog.snapshot()
  212. if err != nil {
  213. panic(err) // TODO(bdarnell)
  214. }
  215. if IsEmptySnap(snapshot) {
  216. panic("need non-empty snapshot")
  217. }
  218. m.Snapshot = snapshot
  219. sindex, sterm := snapshot.Metadata.Index, snapshot.Metadata.Term
  220. raftLogger.Debugf("%x [firstindex: %d, commit: %d] sent snapshot[index: %d, term: %d] to %x [%s]",
  221. r.id, r.raftLog.firstIndex(), r.Commit, sindex, sterm, to, pr)
  222. pr.becomeSnapshot(sindex)
  223. raftLogger.Debugf("%x paused sending replication messages to %x [%s]", r.id, to, pr)
  224. } else {
  225. m.Type = pb.MsgApp
  226. m.Index = pr.Next - 1
  227. m.LogTerm = term
  228. m.Entries = ents
  229. m.Commit = r.raftLog.committed
  230. if n := len(m.Entries); n != 0 {
  231. switch pr.State {
  232. // optimistically increase the next when in ProgressStateReplicate
  233. case ProgressStateReplicate:
  234. last := m.Entries[n-1].Index
  235. pr.optimisticUpdate(last)
  236. pr.ins.add(last)
  237. case ProgressStateProbe:
  238. pr.pause()
  239. default:
  240. raftLogger.Panicf("%x is sending append in unhandled state %s", r.id, pr.State)
  241. }
  242. }
  243. }
  244. r.send(m)
  245. }
  246. // sendHeartbeat sends an empty MsgApp
  247. func (r *raft) sendHeartbeat(to uint64) {
  248. // Attach the commit as min(to.matched, r.committed).
  249. // When the leader sends out heartbeat message,
  250. // the receiver(follower) might not be matched with the leader
  251. // or it might not have all the committed entries.
  252. // The leader MUST NOT forward the follower's commit to
  253. // an unmatched index.
  254. commit := min(r.prs[to].Match, r.raftLog.committed)
  255. m := pb.Message{
  256. To: to,
  257. Type: pb.MsgHeartbeat,
  258. Commit: commit,
  259. }
  260. r.send(m)
  261. }
  262. // bcastAppend sends RRPC, with entries to all peers that are not up-to-date
  263. // according to the progress recorded in r.prs.
  264. func (r *raft) bcastAppend() {
  265. for i := range r.prs {
  266. if i == r.id {
  267. continue
  268. }
  269. r.sendAppend(i)
  270. }
  271. }
  272. // bcastHeartbeat sends RRPC, without entries to all the peers.
  273. func (r *raft) bcastHeartbeat() {
  274. for i := range r.prs {
  275. if i == r.id {
  276. continue
  277. }
  278. r.sendHeartbeat(i)
  279. r.prs[i].resume()
  280. }
  281. }
  282. func (r *raft) maybeCommit() bool {
  283. // TODO(bmizerany): optimize.. Currently naive
  284. mis := make(uint64Slice, 0, len(r.prs))
  285. for i := range r.prs {
  286. mis = append(mis, r.prs[i].Match)
  287. }
  288. sort.Sort(sort.Reverse(mis))
  289. mci := mis[r.q()-1]
  290. return r.raftLog.maybeCommit(mci, r.Term)
  291. }
  292. func (r *raft) reset(term uint64) {
  293. if r.Term != term {
  294. r.Term = term
  295. r.Vote = None
  296. }
  297. r.lead = None
  298. r.elapsed = 0
  299. r.votes = make(map[uint64]bool)
  300. for i := range r.prs {
  301. r.prs[i] = &Progress{Next: r.raftLog.lastIndex() + 1, ins: newInflights(r.maxInflight)}
  302. if i == r.id {
  303. r.prs[i].Match = r.raftLog.lastIndex()
  304. }
  305. }
  306. r.pendingConf = false
  307. }
  308. func (r *raft) appendEntry(es ...pb.Entry) {
  309. li := r.raftLog.lastIndex()
  310. for i := range es {
  311. es[i].Term = r.Term
  312. es[i].Index = li + 1 + uint64(i)
  313. }
  314. r.raftLog.append(es...)
  315. r.prs[r.id].maybeUpdate(r.raftLog.lastIndex())
  316. r.maybeCommit()
  317. }
  318. // tickElection is run by followers and candidates after r.electionTimeout.
  319. func (r *raft) tickElection() {
  320. if !r.promotable() {
  321. r.elapsed = 0
  322. return
  323. }
  324. r.elapsed++
  325. if r.isElectionTimeout() {
  326. r.elapsed = 0
  327. r.Step(pb.Message{From: r.id, Type: pb.MsgHup})
  328. }
  329. }
  330. // tickHeartbeat is run by leaders to send a MsgBeat after r.heartbeatTimeout.
  331. func (r *raft) tickHeartbeat() {
  332. r.elapsed++
  333. if r.elapsed >= r.heartbeatTimeout {
  334. r.elapsed = 0
  335. r.Step(pb.Message{From: r.id, Type: pb.MsgBeat})
  336. }
  337. }
  338. func (r *raft) becomeFollower(term uint64, lead uint64) {
  339. r.step = stepFollower
  340. r.reset(term)
  341. r.tick = r.tickElection
  342. r.lead = lead
  343. r.state = StateFollower
  344. raftLogger.Infof("%x became follower at term %d", r.id, r.Term)
  345. }
  346. func (r *raft) becomeCandidate() {
  347. // TODO(xiangli) remove the panic when the raft implementation is stable
  348. if r.state == StateLeader {
  349. panic("invalid transition [leader -> candidate]")
  350. }
  351. r.step = stepCandidate
  352. r.reset(r.Term + 1)
  353. r.tick = r.tickElection
  354. r.Vote = r.id
  355. r.state = StateCandidate
  356. raftLogger.Infof("%x became candidate at term %d", r.id, r.Term)
  357. }
  358. func (r *raft) becomeLeader() {
  359. // TODO(xiangli) remove the panic when the raft implementation is stable
  360. if r.state == StateFollower {
  361. panic("invalid transition [follower -> leader]")
  362. }
  363. r.step = stepLeader
  364. r.reset(r.Term)
  365. r.tick = r.tickHeartbeat
  366. r.lead = r.id
  367. r.state = StateLeader
  368. ents, err := r.raftLog.entries(r.raftLog.committed+1, noLimit)
  369. if err != nil {
  370. raftLogger.Panicf("unexpected error getting uncommitted entries (%v)", err)
  371. }
  372. for _, e := range ents {
  373. if e.Type != pb.EntryConfChange {
  374. continue
  375. }
  376. if r.pendingConf {
  377. panic("unexpected double uncommitted config entry")
  378. }
  379. r.pendingConf = true
  380. }
  381. r.appendEntry(pb.Entry{Data: nil})
  382. raftLogger.Infof("%x became leader at term %d", r.id, r.Term)
  383. }
  384. func (r *raft) campaign() {
  385. r.becomeCandidate()
  386. if r.q() == r.poll(r.id, true) {
  387. r.becomeLeader()
  388. return
  389. }
  390. for i := range r.prs {
  391. if i == r.id {
  392. continue
  393. }
  394. raftLogger.Infof("%x [logterm: %d, index: %d] sent vote request to %x at term %d",
  395. r.id, r.raftLog.lastTerm(), r.raftLog.lastIndex(), i, r.Term)
  396. r.send(pb.Message{To: i, Type: pb.MsgVote, Index: r.raftLog.lastIndex(), LogTerm: r.raftLog.lastTerm()})
  397. }
  398. }
  399. func (r *raft) poll(id uint64, v bool) (granted int) {
  400. if v {
  401. raftLogger.Infof("%x received vote from %x at term %d", r.id, id, r.Term)
  402. } else {
  403. raftLogger.Infof("%x received vote rejection from %x at term %d", r.id, id, r.Term)
  404. }
  405. if _, ok := r.votes[id]; !ok {
  406. r.votes[id] = v
  407. }
  408. for _, vv := range r.votes {
  409. if vv {
  410. granted++
  411. }
  412. }
  413. return granted
  414. }
  415. func (r *raft) Step(m pb.Message) error {
  416. if m.Type == pb.MsgHup {
  417. raftLogger.Infof("%x is starting a new election at term %d", r.id, r.Term)
  418. r.campaign()
  419. r.Commit = r.raftLog.committed
  420. return nil
  421. }
  422. switch {
  423. case m.Term == 0:
  424. // local message
  425. case m.Term > r.Term:
  426. lead := m.From
  427. if m.Type == pb.MsgVote {
  428. lead = None
  429. }
  430. raftLogger.Infof("%x [term: %d] received a %s message with higher term from %x [term: %d]",
  431. r.id, r.Term, m.Type, m.From, m.Term)
  432. r.becomeFollower(m.Term, lead)
  433. case m.Term < r.Term:
  434. // ignore
  435. raftLogger.Infof("%x [term: %d] ignored a %s message with lower term from %x [term: %d]",
  436. r.id, r.Term, m.Type, m.From, m.Term)
  437. return nil
  438. }
  439. r.step(r, m)
  440. r.Commit = r.raftLog.committed
  441. return nil
  442. }
  443. type stepFunc func(r *raft, m pb.Message)
  444. func stepLeader(r *raft, m pb.Message) {
  445. pr := r.prs[m.From]
  446. switch m.Type {
  447. case pb.MsgBeat:
  448. r.bcastHeartbeat()
  449. case pb.MsgProp:
  450. if len(m.Entries) == 0 {
  451. raftLogger.Panicf("%x stepped empty MsgProp", r.id)
  452. }
  453. for i, e := range m.Entries {
  454. if e.Type == pb.EntryConfChange {
  455. if r.pendingConf {
  456. m.Entries[i] = pb.Entry{Type: pb.EntryNormal}
  457. }
  458. r.pendingConf = true
  459. }
  460. }
  461. r.appendEntry(m.Entries...)
  462. r.bcastAppend()
  463. case pb.MsgAppResp:
  464. if m.Reject {
  465. raftLogger.Debugf("%x received msgApp rejection(lastindex: %d) from %x for index %d",
  466. r.id, m.RejectHint, m.From, m.Index)
  467. if pr.maybeDecrTo(m.Index, m.RejectHint) {
  468. raftLogger.Debugf("%x decreased progress of %x to [%s]", r.id, m.From, pr)
  469. if pr.State == ProgressStateReplicate {
  470. pr.becomeProbe()
  471. }
  472. r.sendAppend(m.From)
  473. }
  474. } else {
  475. oldPaused := pr.isPaused()
  476. if pr.maybeUpdate(m.Index) {
  477. switch {
  478. case pr.State == ProgressStateProbe:
  479. pr.becomeReplicate()
  480. case pr.State == ProgressStateSnapshot && pr.maybeSnapshotAbort():
  481. raftLogger.Debugf("%x snapshot aborted, resumed sending replication messages to %x [%s]", r.id, m.From, pr)
  482. pr.becomeProbe()
  483. case pr.State == ProgressStateReplicate:
  484. pr.ins.freeTo(m.Index)
  485. }
  486. if r.maybeCommit() {
  487. r.bcastAppend()
  488. } else if oldPaused {
  489. // update() reset the wait state on this node. If we had delayed sending
  490. // an update before, send it now.
  491. r.sendAppend(m.From)
  492. }
  493. }
  494. }
  495. case pb.MsgHeartbeatResp:
  496. // free one slot for the full inflights window to allow progress.
  497. if pr.State == ProgressStateReplicate && pr.ins.full() {
  498. pr.ins.freeFirstOne()
  499. }
  500. if pr.Match < r.raftLog.lastIndex() {
  501. r.sendAppend(m.From)
  502. }
  503. case pb.MsgVote:
  504. raftLogger.Infof("%x [logterm: %d, index: %d, vote: %x] rejected vote from %x [logterm: %d, index: %d] at term %d",
  505. r.id, r.raftLog.lastTerm(), r.raftLog.lastIndex(), r.Vote, m.From, m.LogTerm, m.Index, r.Term)
  506. r.send(pb.Message{To: m.From, Type: pb.MsgVoteResp, Reject: true})
  507. case pb.MsgSnapStatus:
  508. if pr.State != ProgressStateSnapshot {
  509. return
  510. }
  511. if !m.Reject {
  512. pr.becomeProbe()
  513. raftLogger.Debugf("%x snapshot succeeded, resumed sending replication messages to %x [%s]", r.id, m.From, pr)
  514. } else {
  515. pr.snapshotFailure()
  516. pr.becomeProbe()
  517. raftLogger.Debugf("%x snapshot failed, resumed sending replication messages to %x [%s]", r.id, m.From, pr)
  518. }
  519. // If snapshot finish, wait for the msgAppResp from the remote node before sending
  520. // out the next msgApp.
  521. // If snapshot failure, wait for a heartbeat interval before next try
  522. pr.pause()
  523. case pb.MsgUnreachable:
  524. // During optimistic replication, if the remote becomes unreachable,
  525. // there is huge probability that a MsgApp is lost.
  526. if pr.State == ProgressStateReplicate {
  527. pr.becomeProbe()
  528. }
  529. raftLogger.Debugf("%x failed to send message to %x because it is unreachable [%s]", r.id, m.From, pr)
  530. }
  531. }
  532. func stepCandidate(r *raft, m pb.Message) {
  533. switch m.Type {
  534. case pb.MsgProp:
  535. raftLogger.Infof("%x no leader at term %d; dropping proposal", r.id, r.Term)
  536. return
  537. case pb.MsgApp:
  538. r.becomeFollower(r.Term, m.From)
  539. r.handleAppendEntries(m)
  540. case pb.MsgHeartbeat:
  541. r.becomeFollower(r.Term, m.From)
  542. r.handleHeartbeat(m)
  543. case pb.MsgSnap:
  544. r.becomeFollower(m.Term, m.From)
  545. r.handleSnapshot(m)
  546. case pb.MsgVote:
  547. raftLogger.Infof("%x [logterm: %d, index: %d, vote: %x] rejected vote from %x [logterm: %d, index: %d] at term %x",
  548. r.id, r.raftLog.lastTerm(), r.raftLog.lastIndex(), r.Vote, m.From, m.LogTerm, m.Index, r.Term)
  549. r.send(pb.Message{To: m.From, Type: pb.MsgVoteResp, Reject: true})
  550. case pb.MsgVoteResp:
  551. gr := r.poll(m.From, !m.Reject)
  552. raftLogger.Infof("%x [q:%d] has received %d votes and %d vote rejections", r.id, r.q(), gr, len(r.votes)-gr)
  553. switch r.q() {
  554. case gr:
  555. r.becomeLeader()
  556. r.bcastAppend()
  557. case len(r.votes) - gr:
  558. r.becomeFollower(r.Term, None)
  559. }
  560. }
  561. }
  562. func stepFollower(r *raft, m pb.Message) {
  563. switch m.Type {
  564. case pb.MsgProp:
  565. if r.lead == None {
  566. raftLogger.Infof("%x no leader at term %d; dropping proposal", r.id, r.Term)
  567. return
  568. }
  569. m.To = r.lead
  570. r.send(m)
  571. case pb.MsgApp:
  572. r.elapsed = 0
  573. r.lead = m.From
  574. r.handleAppendEntries(m)
  575. case pb.MsgHeartbeat:
  576. r.elapsed = 0
  577. r.lead = m.From
  578. r.handleHeartbeat(m)
  579. case pb.MsgSnap:
  580. r.elapsed = 0
  581. r.handleSnapshot(m)
  582. case pb.MsgVote:
  583. if (r.Vote == None || r.Vote == m.From) && r.raftLog.isUpToDate(m.Index, m.LogTerm) {
  584. r.elapsed = 0
  585. raftLogger.Infof("%x [logterm: %d, index: %d, vote: %x] voted for %x [logterm: %d, index: %d] at term %d",
  586. r.id, r.raftLog.lastTerm(), r.raftLog.lastIndex(), r.Vote, m.From, m.LogTerm, m.Index, r.Term)
  587. r.Vote = m.From
  588. r.send(pb.Message{To: m.From, Type: pb.MsgVoteResp})
  589. } else {
  590. raftLogger.Infof("%x [logterm: %d, index: %d, vote: %x] rejected vote from %x [logterm: %d, index: %d] at term %d",
  591. r.id, r.raftLog.lastTerm(), r.raftLog.lastIndex(), r.Vote, m.From, m.LogTerm, m.Index, r.Term)
  592. r.send(pb.Message{To: m.From, Type: pb.MsgVoteResp, Reject: true})
  593. }
  594. }
  595. }
  596. func (r *raft) handleAppendEntries(m pb.Message) {
  597. if m.Index < r.Commit {
  598. r.send(pb.Message{To: m.From, Type: pb.MsgAppResp, Index: r.Commit})
  599. return
  600. }
  601. if mlastIndex, ok := r.raftLog.maybeAppend(m.Index, m.LogTerm, m.Commit, m.Entries...); ok {
  602. r.send(pb.Message{To: m.From, Type: pb.MsgAppResp, Index: mlastIndex})
  603. } else {
  604. raftLogger.Debugf("%x [logterm: %d, index: %d] rejected msgApp [logterm: %d, index: %d] from %x",
  605. r.id, zeroTermOnErrCompacted(r.raftLog.term(m.Index)), m.Index, m.LogTerm, m.Index, m.From)
  606. r.send(pb.Message{To: m.From, Type: pb.MsgAppResp, Index: m.Index, Reject: true, RejectHint: r.raftLog.lastIndex()})
  607. }
  608. }
  609. func (r *raft) handleHeartbeat(m pb.Message) {
  610. r.raftLog.commitTo(m.Commit)
  611. r.send(pb.Message{To: m.From, Type: pb.MsgHeartbeatResp})
  612. }
  613. func (r *raft) handleSnapshot(m pb.Message) {
  614. sindex, sterm := m.Snapshot.Metadata.Index, m.Snapshot.Metadata.Term
  615. if r.restore(m.Snapshot) {
  616. raftLogger.Infof("%x [commit: %d] restored snapshot [index: %d, term: %d]",
  617. r.id, r.Commit, sindex, sterm)
  618. r.send(pb.Message{To: m.From, Type: pb.MsgAppResp, Index: r.raftLog.lastIndex()})
  619. } else {
  620. raftLogger.Infof("%x [commit: %d] ignored snapshot [index: %d, term: %d]",
  621. r.id, r.Commit, sindex, sterm)
  622. r.send(pb.Message{To: m.From, Type: pb.MsgAppResp, Index: r.raftLog.committed})
  623. }
  624. }
  625. // restore recovers the state machine from a snapshot. It restores the log and the
  626. // configuration of state machine.
  627. func (r *raft) restore(s pb.Snapshot) bool {
  628. if s.Metadata.Index <= r.raftLog.committed {
  629. return false
  630. }
  631. if r.raftLog.matchTerm(s.Metadata.Index, s.Metadata.Term) {
  632. raftLogger.Infof("%x [commit: %d, lastindex: %d, lastterm: %d] fast-forwarded commit to snapshot [index: %d, term: %d]",
  633. r.id, r.Commit, r.raftLog.lastIndex(), r.raftLog.lastTerm(), s.Metadata.Index, s.Metadata.Term)
  634. r.raftLog.commitTo(s.Metadata.Index)
  635. return false
  636. }
  637. raftLogger.Infof("%x [commit: %d, lastindex: %d, lastterm: %d] starts to restore snapshot [index: %d, term: %d]",
  638. r.id, r.Commit, r.raftLog.lastIndex(), r.raftLog.lastTerm(), s.Metadata.Index, s.Metadata.Term)
  639. r.raftLog.restore(s)
  640. r.prs = make(map[uint64]*Progress)
  641. for _, n := range s.Metadata.ConfState.Nodes {
  642. match, next := uint64(0), uint64(r.raftLog.lastIndex())+1
  643. if n == r.id {
  644. match = next - 1
  645. } else {
  646. match = 0
  647. }
  648. r.setProgress(n, match, next)
  649. raftLogger.Infof("%x restored progress of %x [%s]", r.id, n, r.prs[n])
  650. }
  651. return true
  652. }
  653. // promotable indicates whether state machine can be promoted to leader,
  654. // which is true when its own id is in progress list.
  655. func (r *raft) promotable() bool {
  656. _, ok := r.prs[r.id]
  657. return ok
  658. }
  659. func (r *raft) addNode(id uint64) {
  660. if _, ok := r.prs[id]; ok {
  661. // Ignore any redundant addNode calls (which can happen because the
  662. // initial bootstrapping entries are applied twice).
  663. return
  664. }
  665. r.setProgress(id, 0, r.raftLog.lastIndex()+1)
  666. r.pendingConf = false
  667. }
  668. func (r *raft) removeNode(id uint64) {
  669. r.delProgress(id)
  670. r.pendingConf = false
  671. }
  672. func (r *raft) resetPendingConf() { r.pendingConf = false }
  673. func (r *raft) setProgress(id, match, next uint64) {
  674. r.prs[id] = &Progress{Next: next, Match: match, ins: newInflights(r.maxInflight)}
  675. }
  676. func (r *raft) delProgress(id uint64) {
  677. delete(r.prs, id)
  678. }
  679. func (r *raft) loadState(state pb.HardState) {
  680. if state.Commit < r.raftLog.committed || state.Commit > r.raftLog.lastIndex() {
  681. raftLogger.Panicf("%x state.commit %d is out of range [%d, %d]", r.id, state.Commit, r.raftLog.committed, r.raftLog.lastIndex())
  682. }
  683. r.raftLog.committed = state.Commit
  684. r.Term = state.Term
  685. r.Vote = state.Vote
  686. r.Commit = state.Commit
  687. }
  688. // isElectionTimeout returns true if r.elapsed is greater than the
  689. // randomized election timeout in (electiontimeout, 2 * electiontimeout - 1).
  690. // Otherwise, it returns false.
  691. func (r *raft) isElectionTimeout() bool {
  692. d := r.elapsed - r.electionTimeout
  693. if d < 0 {
  694. return false
  695. }
  696. return d > r.rand.Int()%r.electionTimeout
  697. }