raft.go 22 KB

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