raft.go 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771
  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. if r.needSnapshot(pr.Next) {
  208. m.Type = pb.MsgSnap
  209. snapshot, err := r.raftLog.snapshot()
  210. if err != nil {
  211. panic(err) // TODO(bdarnell)
  212. }
  213. if IsEmptySnap(snapshot) {
  214. panic("need non-empty snapshot")
  215. }
  216. m.Snapshot = snapshot
  217. sindex, sterm := snapshot.Metadata.Index, snapshot.Metadata.Term
  218. raftLogger.Infof("%x [firstindex: %d, commit: %d] sent snapshot[index: %d, term: %d] to %x [%s]",
  219. r.id, r.raftLog.firstIndex(), r.Commit, sindex, sterm, to, pr)
  220. pr.becomeSnapshot(sindex)
  221. raftLogger.Infof("%x paused sending replication messages to %x [%s]", r.id, to, pr)
  222. } else {
  223. m.Type = pb.MsgApp
  224. m.Index = pr.Next - 1
  225. m.LogTerm = r.raftLog.term(pr.Next - 1)
  226. m.Entries = r.raftLog.entries(pr.Next, r.maxMsgSize)
  227. m.Commit = r.raftLog.committed
  228. if n := len(m.Entries); n != 0 {
  229. switch pr.State {
  230. // optimistically increase the next when in ProgressStateReplicate
  231. case ProgressStateReplicate:
  232. last := m.Entries[n-1].Index
  233. pr.optimisticUpdate(last)
  234. pr.ins.add(last)
  235. case ProgressStateProbe:
  236. pr.pause()
  237. default:
  238. raftLogger.Panicf("%x is sending append in unhandled state %s", r.id, pr.State)
  239. }
  240. }
  241. }
  242. r.send(m)
  243. }
  244. // sendHeartbeat sends an empty MsgApp
  245. func (r *raft) sendHeartbeat(to uint64) {
  246. // Attach the commit as min(to.matched, r.committed).
  247. // When the leader sends out heartbeat message,
  248. // the receiver(follower) might not be matched with the leader
  249. // or it might not have all the committed entries.
  250. // The leader MUST NOT forward the follower's commit to
  251. // an unmatched index.
  252. commit := min(r.prs[to].Match, r.raftLog.committed)
  253. m := pb.Message{
  254. To: to,
  255. Type: pb.MsgHeartbeat,
  256. Commit: commit,
  257. }
  258. r.send(m)
  259. }
  260. // bcastAppend sends RRPC, with entries to all peers that are not up-to-date
  261. // according to the progress recorded in r.prs.
  262. func (r *raft) bcastAppend() {
  263. for i := range r.prs {
  264. if i == r.id {
  265. continue
  266. }
  267. r.sendAppend(i)
  268. }
  269. }
  270. // bcastHeartbeat sends RRPC, without entries to all the peers.
  271. func (r *raft) bcastHeartbeat() {
  272. for i := range r.prs {
  273. if i == r.id {
  274. continue
  275. }
  276. r.sendHeartbeat(i)
  277. r.prs[i].resume()
  278. }
  279. }
  280. func (r *raft) maybeCommit() bool {
  281. // TODO(bmizerany): optimize.. Currently naive
  282. mis := make(uint64Slice, 0, len(r.prs))
  283. for i := range r.prs {
  284. mis = append(mis, r.prs[i].Match)
  285. }
  286. sort.Sort(sort.Reverse(mis))
  287. mci := mis[r.q()-1]
  288. return r.raftLog.maybeCommit(mci, r.Term)
  289. }
  290. func (r *raft) reset(term uint64) {
  291. if r.Term != term {
  292. r.Term = term
  293. r.Vote = None
  294. }
  295. r.lead = None
  296. r.elapsed = 0
  297. r.votes = make(map[uint64]bool)
  298. for i := range r.prs {
  299. r.prs[i] = &Progress{Next: r.raftLog.lastIndex() + 1, ins: newInflights(r.maxInflight)}
  300. if i == r.id {
  301. r.prs[i].Match = r.raftLog.lastIndex()
  302. }
  303. }
  304. r.pendingConf = false
  305. }
  306. func (r *raft) appendEntry(es ...pb.Entry) {
  307. li := r.raftLog.lastIndex()
  308. for i := range es {
  309. es[i].Term = r.Term
  310. es[i].Index = li + 1 + uint64(i)
  311. }
  312. r.raftLog.append(es...)
  313. r.prs[r.id].maybeUpdate(r.raftLog.lastIndex())
  314. r.maybeCommit()
  315. }
  316. // tickElection is run by followers and candidates after r.electionTimeout.
  317. func (r *raft) tickElection() {
  318. if !r.promotable() {
  319. r.elapsed = 0
  320. return
  321. }
  322. r.elapsed++
  323. if r.isElectionTimeout() {
  324. r.elapsed = 0
  325. r.Step(pb.Message{From: r.id, Type: pb.MsgHup})
  326. }
  327. }
  328. // tickHeartbeat is run by leaders to send a MsgBeat after r.heartbeatTimeout.
  329. func (r *raft) tickHeartbeat() {
  330. r.elapsed++
  331. if r.elapsed >= r.heartbeatTimeout {
  332. r.elapsed = 0
  333. r.Step(pb.Message{From: r.id, Type: pb.MsgBeat})
  334. }
  335. }
  336. func (r *raft) becomeFollower(term uint64, lead uint64) {
  337. r.step = stepFollower
  338. r.reset(term)
  339. r.tick = r.tickElection
  340. r.lead = lead
  341. r.state = StateFollower
  342. raftLogger.Infof("%x became follower at term %d", r.id, r.Term)
  343. }
  344. func (r *raft) becomeCandidate() {
  345. // TODO(xiangli) remove the panic when the raft implementation is stable
  346. if r.state == StateLeader {
  347. panic("invalid transition [leader -> candidate]")
  348. }
  349. r.step = stepCandidate
  350. r.reset(r.Term + 1)
  351. r.tick = r.tickElection
  352. r.Vote = r.id
  353. r.state = StateCandidate
  354. raftLogger.Infof("%x became candidate at term %d", r.id, r.Term)
  355. }
  356. func (r *raft) becomeLeader() {
  357. // TODO(xiangli) remove the panic when the raft implementation is stable
  358. if r.state == StateFollower {
  359. panic("invalid transition [follower -> leader]")
  360. }
  361. r.step = stepLeader
  362. r.reset(r.Term)
  363. r.tick = r.tickHeartbeat
  364. r.lead = r.id
  365. r.state = StateLeader
  366. for _, e := range r.raftLog.entries(r.raftLog.committed+1, noLimit) {
  367. if e.Type != pb.EntryConfChange {
  368. continue
  369. }
  370. if r.pendingConf {
  371. panic("unexpected double uncommitted config entry")
  372. }
  373. r.pendingConf = true
  374. }
  375. r.appendEntry(pb.Entry{Data: nil})
  376. raftLogger.Infof("%x became leader at term %d", r.id, r.Term)
  377. }
  378. func (r *raft) campaign() {
  379. r.becomeCandidate()
  380. if r.q() == r.poll(r.id, true) {
  381. r.becomeLeader()
  382. return
  383. }
  384. for i := range r.prs {
  385. if i == r.id {
  386. continue
  387. }
  388. raftLogger.Infof("%x [logterm: %d, index: %d] sent vote request to %x at term %d",
  389. r.id, r.raftLog.lastTerm(), r.raftLog.lastIndex(), i, r.Term)
  390. r.send(pb.Message{To: i, Type: pb.MsgVote, Index: r.raftLog.lastIndex(), LogTerm: r.raftLog.lastTerm()})
  391. }
  392. }
  393. func (r *raft) poll(id uint64, v bool) (granted int) {
  394. if v {
  395. raftLogger.Infof("%x received vote from %x at term %d", r.id, id, r.Term)
  396. } else {
  397. raftLogger.Infof("%x received vote rejection from %x at term %d", r.id, id, r.Term)
  398. }
  399. if _, ok := r.votes[id]; !ok {
  400. r.votes[id] = v
  401. }
  402. for _, vv := range r.votes {
  403. if vv {
  404. granted++
  405. }
  406. }
  407. return granted
  408. }
  409. func (r *raft) Step(m pb.Message) error {
  410. if m.Type == pb.MsgHup {
  411. raftLogger.Infof("%x is starting a new election at term %d", r.id, r.Term)
  412. r.campaign()
  413. r.Commit = r.raftLog.committed
  414. return nil
  415. }
  416. switch {
  417. case m.Term == 0:
  418. // local message
  419. case m.Term > r.Term:
  420. lead := m.From
  421. if m.Type == pb.MsgVote {
  422. lead = None
  423. }
  424. raftLogger.Infof("%x [term: %d] received a %s message with higher term from %x [term: %d]",
  425. r.id, r.Term, m.Type, m.From, m.Term)
  426. r.becomeFollower(m.Term, lead)
  427. case m.Term < r.Term:
  428. // ignore
  429. raftLogger.Infof("%x [term: %d] ignored a %s message with lower term from %x [term: %d]",
  430. r.id, r.Term, m.Type, m.From, m.Term)
  431. return nil
  432. }
  433. r.step(r, m)
  434. r.Commit = r.raftLog.committed
  435. return nil
  436. }
  437. type stepFunc func(r *raft, m pb.Message)
  438. func stepLeader(r *raft, m pb.Message) {
  439. pr := r.prs[m.From]
  440. switch m.Type {
  441. case pb.MsgBeat:
  442. r.bcastHeartbeat()
  443. case pb.MsgProp:
  444. if len(m.Entries) == 0 {
  445. raftLogger.Panicf("%x stepped empty MsgProp", r.id)
  446. }
  447. for i, e := range m.Entries {
  448. if e.Type == pb.EntryConfChange {
  449. if r.pendingConf {
  450. m.Entries[i] = pb.Entry{Type: pb.EntryNormal}
  451. }
  452. r.pendingConf = true
  453. }
  454. }
  455. r.appendEntry(m.Entries...)
  456. r.bcastAppend()
  457. case pb.MsgAppResp:
  458. if m.Reject {
  459. raftLogger.Debugf("%x received msgApp rejection(lastindex: %d) from %x for index %d",
  460. r.id, m.RejectHint, m.From, m.Index)
  461. if pr.maybeDecrTo(m.Index, m.RejectHint) {
  462. raftLogger.Debugf("%x decreased progress of %x to [%s]", r.id, m.From, pr)
  463. if pr.State == ProgressStateReplicate {
  464. pr.becomeProbe()
  465. }
  466. r.sendAppend(m.From)
  467. }
  468. } else {
  469. oldPaused := pr.isPaused()
  470. if pr.maybeUpdate(m.Index) {
  471. switch {
  472. case pr.State == ProgressStateProbe:
  473. pr.becomeReplicate()
  474. case pr.State == ProgressStateSnapshot && pr.maybeSnapshotAbort():
  475. raftLogger.Infof("%x snapshot aborted, resumed sending replication messages to %x [%s]", r.id, m.From, pr)
  476. pr.becomeProbe()
  477. case pr.State == ProgressStateReplicate:
  478. pr.ins.freeTo(m.Index)
  479. }
  480. if r.maybeCommit() {
  481. r.bcastAppend()
  482. } else if oldPaused {
  483. // update() reset the wait state on this node. If we had delayed sending
  484. // an update before, send it now.
  485. r.sendAppend(m.From)
  486. }
  487. }
  488. }
  489. case pb.MsgHeartbeatResp:
  490. // free one slot for the full inflights window to allow progress.
  491. if pr.State == ProgressStateReplicate && pr.ins.full() {
  492. pr.ins.freeFirstOne()
  493. }
  494. if pr.Match < r.raftLog.lastIndex() {
  495. r.sendAppend(m.From)
  496. }
  497. case pb.MsgVote:
  498. raftLogger.Infof("%x [logterm: %d, index: %d, vote: %x] rejected vote from %x [logterm: %d, index: %d] at term %d",
  499. r.id, r.raftLog.lastTerm(), r.raftLog.lastIndex(), r.Vote, m.From, m.LogTerm, m.Index, r.Term)
  500. r.send(pb.Message{To: m.From, Type: pb.MsgVoteResp, Reject: true})
  501. case pb.MsgSnapStatus:
  502. if pr.State != ProgressStateSnapshot {
  503. return
  504. }
  505. if !m.Reject {
  506. pr.becomeProbe()
  507. raftLogger.Infof("%x snapshot succeeded, resumed sending replication messages to %x [%s]", r.id, m.From, pr)
  508. } else {
  509. pr.snapshotFailure()
  510. pr.becomeProbe()
  511. raftLogger.Infof("%x snapshot failed, resumed sending replication messages to %x [%s]", r.id, m.From, pr)
  512. }
  513. // If snapshot finish, wait for the msgAppResp from the remote node before sending
  514. // out the next msgApp.
  515. // If snapshot failure, wait for a heartbeat interval before next try
  516. pr.pause()
  517. case pb.MsgUnreachable:
  518. // During optimistic replication, if the remote becomes unreachable,
  519. // there is huge probability that a MsgApp is lost.
  520. if pr.State == ProgressStateReplicate {
  521. pr.becomeProbe()
  522. }
  523. raftLogger.Debugf("%x failed to send message to %x because it is unreachable [%s]", r.id, m.From, pr)
  524. }
  525. }
  526. func stepCandidate(r *raft, m pb.Message) {
  527. switch m.Type {
  528. case pb.MsgProp:
  529. raftLogger.Infof("%x no leader at term %d; dropping proposal", r.id, r.Term)
  530. return
  531. case pb.MsgApp:
  532. r.becomeFollower(r.Term, m.From)
  533. r.handleAppendEntries(m)
  534. case pb.MsgHeartbeat:
  535. r.becomeFollower(r.Term, m.From)
  536. r.handleHeartbeat(m)
  537. case pb.MsgSnap:
  538. r.becomeFollower(m.Term, m.From)
  539. r.handleSnapshot(m)
  540. case pb.MsgVote:
  541. raftLogger.Infof("%x [logterm: %d, index: %d, vote: %x] rejected vote from %x [logterm: %d, index: %d] at term %x",
  542. r.id, r.raftLog.lastTerm(), r.raftLog.lastIndex(), r.Vote, m.From, m.LogTerm, m.Index, r.Term)
  543. r.send(pb.Message{To: m.From, Type: pb.MsgVoteResp, Reject: true})
  544. case pb.MsgVoteResp:
  545. gr := r.poll(m.From, !m.Reject)
  546. raftLogger.Infof("%x [q:%d] has received %d votes and %d vote rejections", r.id, r.q(), gr, len(r.votes)-gr)
  547. switch r.q() {
  548. case gr:
  549. r.becomeLeader()
  550. r.bcastAppend()
  551. case len(r.votes) - gr:
  552. r.becomeFollower(r.Term, None)
  553. }
  554. }
  555. }
  556. func stepFollower(r *raft, m pb.Message) {
  557. switch m.Type {
  558. case pb.MsgProp:
  559. if r.lead == None {
  560. raftLogger.Infof("%x no leader at term %d; dropping proposal", r.id, r.Term)
  561. return
  562. }
  563. m.To = r.lead
  564. r.send(m)
  565. case pb.MsgApp:
  566. r.elapsed = 0
  567. r.lead = m.From
  568. r.handleAppendEntries(m)
  569. case pb.MsgHeartbeat:
  570. r.elapsed = 0
  571. r.lead = m.From
  572. r.handleHeartbeat(m)
  573. case pb.MsgSnap:
  574. r.elapsed = 0
  575. r.handleSnapshot(m)
  576. case pb.MsgVote:
  577. if (r.Vote == None || r.Vote == m.From) && r.raftLog.isUpToDate(m.Index, m.LogTerm) {
  578. r.elapsed = 0
  579. raftLogger.Infof("%x [logterm: %d, index: %d, vote: %x] voted for %x [logterm: %d, index: %d] at term %d",
  580. r.id, r.raftLog.lastTerm(), r.raftLog.lastIndex(), r.Vote, m.From, m.LogTerm, m.Index, r.Term)
  581. r.Vote = m.From
  582. r.send(pb.Message{To: m.From, Type: pb.MsgVoteResp})
  583. } else {
  584. raftLogger.Infof("%x [logterm: %d, index: %d, vote: %x] rejected vote from %x [logterm: %d, index: %d] at term %d",
  585. r.id, r.raftLog.lastTerm(), r.raftLog.lastIndex(), r.Vote, m.From, m.LogTerm, m.Index, r.Term)
  586. r.send(pb.Message{To: m.From, Type: pb.MsgVoteResp, Reject: true})
  587. }
  588. }
  589. }
  590. func (r *raft) handleAppendEntries(m pb.Message) {
  591. if m.Index < r.Commit {
  592. r.send(pb.Message{To: m.From, Type: pb.MsgAppResp, Index: r.Commit})
  593. return
  594. }
  595. if mlastIndex, ok := r.raftLog.maybeAppend(m.Index, m.LogTerm, m.Commit, m.Entries...); ok {
  596. r.send(pb.Message{To: m.From, Type: pb.MsgAppResp, Index: mlastIndex})
  597. } else {
  598. raftLogger.Debugf("%x [logterm: %d, index: %d] rejected msgApp [logterm: %d, index: %d] from %x",
  599. r.id, r.raftLog.term(m.Index), m.Index, m.LogTerm, m.Index, m.From)
  600. r.send(pb.Message{To: m.From, Type: pb.MsgAppResp, Index: m.Index, Reject: true, RejectHint: r.raftLog.lastIndex()})
  601. }
  602. }
  603. func (r *raft) handleHeartbeat(m pb.Message) {
  604. r.raftLog.commitTo(m.Commit)
  605. r.send(pb.Message{To: m.From, Type: pb.MsgHeartbeatResp})
  606. }
  607. func (r *raft) handleSnapshot(m pb.Message) {
  608. sindex, sterm := m.Snapshot.Metadata.Index, m.Snapshot.Metadata.Term
  609. if r.restore(m.Snapshot) {
  610. raftLogger.Infof("%x [commit: %d] restored snapshot [index: %d, term: %d]",
  611. r.id, r.Commit, sindex, sterm)
  612. r.send(pb.Message{To: m.From, Type: pb.MsgAppResp, Index: r.raftLog.lastIndex()})
  613. } else {
  614. raftLogger.Infof("%x [commit: %d] ignored snapshot [index: %d, term: %d]",
  615. r.id, r.Commit, sindex, sterm)
  616. r.send(pb.Message{To: m.From, Type: pb.MsgAppResp, Index: r.raftLog.committed})
  617. }
  618. }
  619. // restore recovers the state machine from a snapshot. It restores the log and the
  620. // configuration of state machine.
  621. func (r *raft) restore(s pb.Snapshot) bool {
  622. if s.Metadata.Index <= r.raftLog.committed {
  623. return false
  624. }
  625. if r.raftLog.matchTerm(s.Metadata.Index, s.Metadata.Term) {
  626. raftLogger.Infof("%x [commit: %d, lastindex: %d, lastterm: %d] fast-forwarded commit to snapshot [index: %d, term: %d]",
  627. r.id, r.Commit, r.raftLog.lastIndex(), r.raftLog.lastTerm(), s.Metadata.Index, s.Metadata.Term)
  628. r.raftLog.commitTo(s.Metadata.Index)
  629. return false
  630. }
  631. raftLogger.Infof("%x [commit: %d, lastindex: %d, lastterm: %d] starts to restore snapshot [index: %d, term: %d]",
  632. r.id, r.Commit, r.raftLog.lastIndex(), r.raftLog.lastTerm(), s.Metadata.Index, s.Metadata.Term)
  633. r.raftLog.restore(s)
  634. r.prs = make(map[uint64]*Progress)
  635. for _, n := range s.Metadata.ConfState.Nodes {
  636. match, next := uint64(0), uint64(r.raftLog.lastIndex())+1
  637. if n == r.id {
  638. match = next - 1
  639. } else {
  640. match = 0
  641. }
  642. r.setProgress(n, match, next)
  643. raftLogger.Infof("%x restored progress of %x [%s]", r.id, n, r.prs[n])
  644. }
  645. return true
  646. }
  647. func (r *raft) needSnapshot(i uint64) bool {
  648. return i < r.raftLog.firstIndex()
  649. }
  650. // promotable indicates whether state machine can be promoted to leader,
  651. // which is true when its own id is in progress list.
  652. func (r *raft) promotable() bool {
  653. _, ok := r.prs[r.id]
  654. return ok
  655. }
  656. func (r *raft) addNode(id uint64) {
  657. if _, ok := r.prs[id]; ok {
  658. // Ignore any redundant addNode calls (which can happen because the
  659. // initial bootstrapping entries are applied twice).
  660. return
  661. }
  662. r.setProgress(id, 0, r.raftLog.lastIndex()+1)
  663. r.pendingConf = false
  664. }
  665. func (r *raft) removeNode(id uint64) {
  666. r.delProgress(id)
  667. r.pendingConf = false
  668. }
  669. func (r *raft) resetPendingConf() { r.pendingConf = false }
  670. func (r *raft) setProgress(id, match, next uint64) {
  671. r.prs[id] = &Progress{Next: next, Match: match, ins: newInflights(r.maxInflight)}
  672. }
  673. func (r *raft) delProgress(id uint64) {
  674. delete(r.prs, id)
  675. }
  676. func (r *raft) loadState(state pb.HardState) {
  677. if state.Commit < r.raftLog.committed || state.Commit > r.raftLog.lastIndex() {
  678. raftLogger.Panicf("%x state.commit %d is out of range [%d, %d]", r.id, state.Commit, r.raftLog.committed, r.raftLog.lastIndex())
  679. }
  680. r.raftLog.committed = state.Commit
  681. r.Term = state.Term
  682. r.Vote = state.Vote
  683. r.Commit = state.Commit
  684. }
  685. // isElectionTimeout returns true if r.elapsed is greater than the
  686. // randomized election timeout in (electiontimeout, 2 * electiontimeout - 1).
  687. // Otherwise, it returns false.
  688. func (r *raft) isElectionTimeout() bool {
  689. d := r.elapsed - r.electionTimeout
  690. if d < 0 {
  691. return false
  692. }
  693. return d > r.rand.Int()%r.electionTimeout
  694. }