raft_test.go 47 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708
  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. "bytes"
  17. "fmt"
  18. "math"
  19. "math/rand"
  20. "reflect"
  21. "testing"
  22. pb "github.com/coreos/etcd/raft/raftpb"
  23. )
  24. // nextEnts returns the appliable entries and updates the applied index
  25. func nextEnts(r *raft, s *MemoryStorage) (ents []pb.Entry) {
  26. // Transfer all unstable entries to "stable" storage.
  27. s.Append(r.raftLog.unstableEntries())
  28. r.raftLog.stableTo(r.raftLog.lastIndex(), r.raftLog.lastTerm())
  29. ents = r.raftLog.nextEnts()
  30. r.raftLog.appliedTo(r.raftLog.committed)
  31. return ents
  32. }
  33. type Interface interface {
  34. Step(m pb.Message) error
  35. readMessages() []pb.Message
  36. }
  37. func (r *raft) readMessages() []pb.Message {
  38. msgs := r.msgs
  39. r.msgs = make([]pb.Message, 0)
  40. return msgs
  41. }
  42. func TestProgressUpdate(t *testing.T) {
  43. prevM, prevN := uint64(3), uint64(5)
  44. tests := []struct {
  45. update uint64
  46. wm uint64
  47. wn uint64
  48. }{
  49. {prevM - 1, prevM, prevN}, // do not decrease match, next
  50. {prevM, prevM, prevN}, // do not decrease next
  51. {prevM + 1, prevM + 1, prevN}, // increase match, do not decrease next
  52. {prevM + 2, prevM + 2, prevN + 1}, // increase match, next
  53. }
  54. for i, tt := range tests {
  55. p := &Progress{
  56. Match: prevM,
  57. Next: prevN,
  58. }
  59. p.update(tt.update)
  60. if p.Match != tt.wm {
  61. t.Errorf("#%d: match= %d, want %d", i, p.Match, tt.wm)
  62. }
  63. if p.Next != tt.wn {
  64. t.Errorf("#%d: next= %d, want %d", i, p.Next, tt.wn)
  65. }
  66. }
  67. }
  68. func TestProgressMaybeDecr(t *testing.T) {
  69. tests := []struct {
  70. m uint64
  71. n uint64
  72. rejected uint64
  73. last uint64
  74. w bool
  75. wn uint64
  76. }{
  77. {
  78. // match != 0 is always false
  79. 1, 0, 0, 0, false, 0,
  80. },
  81. {
  82. // match != 0 and to is greater than match
  83. // directly decrease to match+1
  84. 5, 10, 5, 5, false, 10,
  85. },
  86. {
  87. // match != 0 and to is greater than match
  88. // directly decrease to match+1
  89. 5, 10, 4, 4, false, 10,
  90. },
  91. {
  92. // match != 0 and to is not greater than match
  93. 5, 10, 9, 9, true, 6,
  94. },
  95. {
  96. // next-1 != rejected is always false
  97. 0, 0, 0, 0, false, 0,
  98. },
  99. {
  100. // next-1 != rejected is always false
  101. 0, 10, 5, 5, false, 10,
  102. },
  103. {
  104. // next>1 = decremented by 1
  105. 0, 10, 9, 9, true, 9,
  106. },
  107. {
  108. // next>1 = decremented by 1
  109. 0, 2, 1, 1, true, 1,
  110. },
  111. {
  112. // next<=1 = reset to 1
  113. 0, 1, 0, 0, true, 1,
  114. },
  115. {
  116. // decrease to min(rejected, last+1)
  117. 0, 10, 9, 2, true, 3,
  118. },
  119. {
  120. // rejected < 1, reset to 1
  121. 0, 10, 9, 0, true, 1,
  122. },
  123. }
  124. for i, tt := range tests {
  125. p := &Progress{
  126. Match: tt.m,
  127. Next: tt.n,
  128. }
  129. if g := p.maybeDecrTo(tt.rejected, tt.last); g != tt.w {
  130. t.Errorf("#%d: maybeDecrTo= %t, want %t", i, g, tt.w)
  131. }
  132. if gm := p.Match; gm != tt.m {
  133. t.Errorf("#%d: match= %d, want %d", i, gm, tt.m)
  134. }
  135. if gn := p.Next; gn != tt.wn {
  136. t.Errorf("#%d: next= %d, want %d", i, gn, tt.wn)
  137. }
  138. }
  139. }
  140. func TestProgressShouldWait(t *testing.T) {
  141. tests := []struct {
  142. m uint64
  143. wait int
  144. w bool
  145. }{
  146. // match != 0 is always not wait
  147. {1, 0, false},
  148. {1, 1, false},
  149. {0, 1, true},
  150. {0, 0, false},
  151. }
  152. for i, tt := range tests {
  153. p := &Progress{
  154. Match: tt.m,
  155. Wait: tt.wait,
  156. }
  157. if g := p.shouldWait(); g != tt.w {
  158. t.Errorf("#%d: shouldwait = %t, want %t", i, g, tt.w)
  159. }
  160. }
  161. }
  162. // TestProgressWaitReset ensures that progress.Update and progress.DercTo
  163. // will reset progress.wait.
  164. func TestProgressWaitReset(t *testing.T) {
  165. p := &Progress{
  166. Wait: 1,
  167. }
  168. p.maybeDecrTo(1, 1)
  169. if p.Wait != 0 {
  170. t.Errorf("wait= %d, want 0", p.Wait)
  171. }
  172. p.Wait = 1
  173. p.update(2)
  174. if p.Wait != 0 {
  175. t.Errorf("wait= %d, want 0", p.Wait)
  176. }
  177. }
  178. // TestProgressDecr ensures raft.heartbeat decreases progress.wait by heartbeat.
  179. func TestProgressDecr(t *testing.T) {
  180. r := newRaft(1, []uint64{1, 2}, 5, 1, NewMemoryStorage(), 0)
  181. r.becomeCandidate()
  182. r.becomeLeader()
  183. r.prs[2].Wait = r.heartbeatTimeout * 2
  184. r.Step(pb.Message{From: 1, To: 1, Type: pb.MsgBeat})
  185. if r.prs[2].Wait != r.heartbeatTimeout*(2-1) {
  186. t.Errorf("wait = %d, want %d", r.prs[2].Wait, r.heartbeatTimeout*(2-1))
  187. }
  188. }
  189. func TestProgressWait(t *testing.T) {
  190. r := newRaft(1, []uint64{1, 2}, 5, 1, NewMemoryStorage(), 0)
  191. r.becomeCandidate()
  192. r.becomeLeader()
  193. r.Step(pb.Message{From: 1, To: 1, Type: pb.MsgProp, Entries: []pb.Entry{{Data: []byte("somedata")}}})
  194. r.Step(pb.Message{From: 1, To: 1, Type: pb.MsgProp, Entries: []pb.Entry{{Data: []byte("somedata")}}})
  195. r.Step(pb.Message{From: 1, To: 1, Type: pb.MsgProp, Entries: []pb.Entry{{Data: []byte("somedata")}}})
  196. ms := r.readMessages()
  197. if len(ms) != 1 {
  198. t.Errorf("len(ms) = %d, want 1", len(ms))
  199. }
  200. }
  201. func TestLeaderElection(t *testing.T) {
  202. tests := []struct {
  203. *network
  204. state StateType
  205. }{
  206. {newNetwork(nil, nil, nil), StateLeader},
  207. {newNetwork(nil, nil, nopStepper), StateLeader},
  208. {newNetwork(nil, nopStepper, nopStepper), StateCandidate},
  209. {newNetwork(nil, nopStepper, nopStepper, nil), StateCandidate},
  210. {newNetwork(nil, nopStepper, nopStepper, nil, nil), StateLeader},
  211. // three logs further along than 0
  212. {newNetwork(nil, ents(1), ents(2), ents(1, 3), nil), StateFollower},
  213. // logs converge
  214. {newNetwork(ents(1), nil, ents(2), ents(1), nil), StateLeader},
  215. }
  216. for i, tt := range tests {
  217. tt.send(pb.Message{From: 1, To: 1, Type: pb.MsgHup})
  218. sm := tt.network.peers[1].(*raft)
  219. if sm.state != tt.state {
  220. t.Errorf("#%d: state = %s, want %s", i, sm.state, tt.state)
  221. }
  222. if g := sm.Term; g != 1 {
  223. t.Errorf("#%d: term = %d, want %d", i, g, 1)
  224. }
  225. }
  226. }
  227. func TestLogReplication(t *testing.T) {
  228. tests := []struct {
  229. *network
  230. msgs []pb.Message
  231. wcommitted uint64
  232. }{
  233. {
  234. newNetwork(nil, nil, nil),
  235. []pb.Message{
  236. {From: 1, To: 1, Type: pb.MsgProp, Entries: []pb.Entry{{Data: []byte("somedata")}}},
  237. },
  238. 2,
  239. },
  240. {
  241. newNetwork(nil, nil, nil),
  242. []pb.Message{
  243. {From: 1, To: 1, Type: pb.MsgProp, Entries: []pb.Entry{{Data: []byte("somedata")}}},
  244. {From: 1, To: 2, Type: pb.MsgHup},
  245. {From: 1, To: 2, Type: pb.MsgProp, Entries: []pb.Entry{{Data: []byte("somedata")}}},
  246. },
  247. 4,
  248. },
  249. }
  250. for i, tt := range tests {
  251. tt.send(pb.Message{From: 1, To: 1, Type: pb.MsgHup})
  252. for _, m := range tt.msgs {
  253. tt.send(m)
  254. }
  255. for j, x := range tt.network.peers {
  256. sm := x.(*raft)
  257. if sm.raftLog.committed != tt.wcommitted {
  258. t.Errorf("#%d.%d: committed = %d, want %d", i, j, sm.raftLog.committed, tt.wcommitted)
  259. }
  260. ents := []pb.Entry{}
  261. for _, e := range nextEnts(sm, tt.network.storage[j]) {
  262. if e.Data != nil {
  263. ents = append(ents, e)
  264. }
  265. }
  266. props := []pb.Message{}
  267. for _, m := range tt.msgs {
  268. if m.Type == pb.MsgProp {
  269. props = append(props, m)
  270. }
  271. }
  272. for k, m := range props {
  273. if !bytes.Equal(ents[k].Data, m.Entries[0].Data) {
  274. t.Errorf("#%d.%d: data = %d, want %d", i, j, ents[k].Data, m.Entries[0].Data)
  275. }
  276. }
  277. }
  278. }
  279. }
  280. func TestSingleNodeCommit(t *testing.T) {
  281. tt := newNetwork(nil)
  282. tt.send(pb.Message{From: 1, To: 1, Type: pb.MsgHup})
  283. tt.send(pb.Message{From: 1, To: 1, Type: pb.MsgProp, Entries: []pb.Entry{{Data: []byte("some data")}}})
  284. tt.send(pb.Message{From: 1, To: 1, Type: pb.MsgProp, Entries: []pb.Entry{{Data: []byte("some data")}}})
  285. sm := tt.peers[1].(*raft)
  286. if sm.raftLog.committed != 3 {
  287. t.Errorf("committed = %d, want %d", sm.raftLog.committed, 3)
  288. }
  289. }
  290. // TestCannotCommitWithoutNewTermEntry tests the entries cannot be committed
  291. // when leader changes, no new proposal comes in and ChangeTerm proposal is
  292. // filtered.
  293. func TestCannotCommitWithoutNewTermEntry(t *testing.T) {
  294. tt := newNetwork(nil, nil, nil, nil, nil)
  295. tt.send(pb.Message{From: 1, To: 1, Type: pb.MsgHup})
  296. // 0 cannot reach 2,3,4
  297. tt.cut(1, 3)
  298. tt.cut(1, 4)
  299. tt.cut(1, 5)
  300. tt.send(pb.Message{From: 1, To: 1, Type: pb.MsgProp, Entries: []pb.Entry{{Data: []byte("some data")}}})
  301. tt.send(pb.Message{From: 1, To: 1, Type: pb.MsgProp, Entries: []pb.Entry{{Data: []byte("some data")}}})
  302. sm := tt.peers[1].(*raft)
  303. if sm.raftLog.committed != 1 {
  304. t.Errorf("committed = %d, want %d", sm.raftLog.committed, 1)
  305. }
  306. // network recovery
  307. tt.recover()
  308. // avoid committing ChangeTerm proposal
  309. tt.ignore(pb.MsgApp)
  310. // elect 2 as the new leader with term 2
  311. tt.send(pb.Message{From: 2, To: 2, Type: pb.MsgHup})
  312. // no log entries from previous term should be committed
  313. sm = tt.peers[2].(*raft)
  314. if sm.raftLog.committed != 1 {
  315. t.Errorf("committed = %d, want %d", sm.raftLog.committed, 1)
  316. }
  317. tt.recover()
  318. // send heartbeat; reset wait
  319. tt.send(pb.Message{From: 2, To: 2, Type: pb.MsgBeat})
  320. // append an entry at current term
  321. tt.send(pb.Message{From: 2, To: 2, Type: pb.MsgProp, Entries: []pb.Entry{{Data: []byte("some data")}}})
  322. // expect the committed to be advanced
  323. if sm.raftLog.committed != 5 {
  324. t.Errorf("committed = %d, want %d", sm.raftLog.committed, 5)
  325. }
  326. }
  327. // TestCommitWithoutNewTermEntry tests the entries could be committed
  328. // when leader changes, no new proposal comes in.
  329. func TestCommitWithoutNewTermEntry(t *testing.T) {
  330. tt := newNetwork(nil, nil, nil, nil, nil)
  331. tt.send(pb.Message{From: 1, To: 1, Type: pb.MsgHup})
  332. // 0 cannot reach 2,3,4
  333. tt.cut(1, 3)
  334. tt.cut(1, 4)
  335. tt.cut(1, 5)
  336. tt.send(pb.Message{From: 1, To: 1, Type: pb.MsgProp, Entries: []pb.Entry{{Data: []byte("some data")}}})
  337. tt.send(pb.Message{From: 1, To: 1, Type: pb.MsgProp, Entries: []pb.Entry{{Data: []byte("some data")}}})
  338. sm := tt.peers[1].(*raft)
  339. if sm.raftLog.committed != 1 {
  340. t.Errorf("committed = %d, want %d", sm.raftLog.committed, 1)
  341. }
  342. // network recovery
  343. tt.recover()
  344. // elect 1 as the new leader with term 2
  345. // after append a ChangeTerm entry from the current term, all entries
  346. // should be committed
  347. tt.send(pb.Message{From: 2, To: 2, Type: pb.MsgHup})
  348. if sm.raftLog.committed != 4 {
  349. t.Errorf("committed = %d, want %d", sm.raftLog.committed, 4)
  350. }
  351. }
  352. func TestDuelingCandidates(t *testing.T) {
  353. a := newRaft(1, []uint64{1, 2, 3}, 10, 1, NewMemoryStorage(), 0)
  354. b := newRaft(2, []uint64{1, 2, 3}, 10, 1, NewMemoryStorage(), 0)
  355. c := newRaft(3, []uint64{1, 2, 3}, 10, 1, NewMemoryStorage(), 0)
  356. nt := newNetwork(a, b, c)
  357. nt.cut(1, 3)
  358. nt.send(pb.Message{From: 1, To: 1, Type: pb.MsgHup})
  359. nt.send(pb.Message{From: 3, To: 3, Type: pb.MsgHup})
  360. nt.recover()
  361. nt.send(pb.Message{From: 3, To: 3, Type: pb.MsgHup})
  362. wlog := &raftLog{
  363. storage: &MemoryStorage{ents: []pb.Entry{{}, pb.Entry{Data: nil, Term: 1, Index: 1}}},
  364. committed: 1,
  365. unstable: unstable{offset: 2},
  366. }
  367. tests := []struct {
  368. sm *raft
  369. state StateType
  370. term uint64
  371. raftLog *raftLog
  372. }{
  373. {a, StateFollower, 2, wlog},
  374. {b, StateFollower, 2, wlog},
  375. {c, StateFollower, 2, newLog(NewMemoryStorage())},
  376. }
  377. for i, tt := range tests {
  378. if g := tt.sm.state; g != tt.state {
  379. t.Errorf("#%d: state = %s, want %s", i, g, tt.state)
  380. }
  381. if g := tt.sm.Term; g != tt.term {
  382. t.Errorf("#%d: term = %d, want %d", i, g, tt.term)
  383. }
  384. base := ltoa(tt.raftLog)
  385. if sm, ok := nt.peers[1+uint64(i)].(*raft); ok {
  386. l := ltoa(sm.raftLog)
  387. if g := diffu(base, l); g != "" {
  388. t.Errorf("#%d: diff:\n%s", i, g)
  389. }
  390. } else {
  391. t.Logf("#%d: empty log", i)
  392. }
  393. }
  394. }
  395. func TestCandidateConcede(t *testing.T) {
  396. tt := newNetwork(nil, nil, nil)
  397. tt.isolate(1)
  398. tt.send(pb.Message{From: 1, To: 1, Type: pb.MsgHup})
  399. tt.send(pb.Message{From: 3, To: 3, Type: pb.MsgHup})
  400. // heal the partition
  401. tt.recover()
  402. // send heartbeat; reset wait
  403. tt.send(pb.Message{From: 3, To: 3, Type: pb.MsgBeat})
  404. data := []byte("force follower")
  405. // send a proposal to 3 to flush out a MsgApp to 1
  406. tt.send(pb.Message{From: 3, To: 3, Type: pb.MsgProp, Entries: []pb.Entry{{Data: data}}})
  407. // send heartbeat; flush out commit
  408. tt.send(pb.Message{From: 3, To: 3, Type: pb.MsgBeat})
  409. a := tt.peers[1].(*raft)
  410. if g := a.state; g != StateFollower {
  411. t.Errorf("state = %s, want %s", g, StateFollower)
  412. }
  413. if g := a.Term; g != 1 {
  414. t.Errorf("term = %d, want %d", g, 1)
  415. }
  416. wantLog := ltoa(&raftLog{
  417. storage: &MemoryStorage{
  418. ents: []pb.Entry{{}, {Data: nil, Term: 1, Index: 1}, {Term: 1, Index: 2, Data: data}},
  419. },
  420. unstable: unstable{offset: 3},
  421. committed: 2,
  422. })
  423. for i, p := range tt.peers {
  424. if sm, ok := p.(*raft); ok {
  425. l := ltoa(sm.raftLog)
  426. if g := diffu(wantLog, l); g != "" {
  427. t.Errorf("#%d: diff:\n%s", i, g)
  428. }
  429. } else {
  430. t.Logf("#%d: empty log", i)
  431. }
  432. }
  433. }
  434. func TestSingleNodeCandidate(t *testing.T) {
  435. tt := newNetwork(nil)
  436. tt.send(pb.Message{From: 1, To: 1, Type: pb.MsgHup})
  437. sm := tt.peers[1].(*raft)
  438. if sm.state != StateLeader {
  439. t.Errorf("state = %d, want %d", sm.state, StateLeader)
  440. }
  441. }
  442. func TestOldMessages(t *testing.T) {
  443. tt := newNetwork(nil, nil, nil)
  444. // make 0 leader @ term 3
  445. tt.send(pb.Message{From: 1, To: 1, Type: pb.MsgHup})
  446. tt.send(pb.Message{From: 2, To: 2, Type: pb.MsgHup})
  447. tt.send(pb.Message{From: 1, To: 1, Type: pb.MsgHup})
  448. // pretend we're an old leader trying to make progress; this entry is expected to be ignored.
  449. tt.send(pb.Message{From: 2, To: 1, Type: pb.MsgApp, Term: 2, Entries: []pb.Entry{{Index: 3, Term: 2}}})
  450. // commit a new entry
  451. tt.send(pb.Message{From: 1, To: 1, Type: pb.MsgProp, Entries: []pb.Entry{{Data: []byte("somedata")}}})
  452. ilog := &raftLog{
  453. storage: &MemoryStorage{
  454. ents: []pb.Entry{
  455. {}, {Data: nil, Term: 1, Index: 1},
  456. {Data: nil, Term: 2, Index: 2}, {Data: nil, Term: 3, Index: 3},
  457. {Data: []byte("somedata"), Term: 3, Index: 4},
  458. },
  459. },
  460. unstable: unstable{offset: 5},
  461. committed: 4,
  462. }
  463. base := ltoa(ilog)
  464. for i, p := range tt.peers {
  465. if sm, ok := p.(*raft); ok {
  466. l := ltoa(sm.raftLog)
  467. if g := diffu(base, l); g != "" {
  468. t.Errorf("#%d: diff:\n%s", i, g)
  469. }
  470. } else {
  471. t.Logf("#%d: empty log", i)
  472. }
  473. }
  474. }
  475. // TestOldMessagesReply - optimization - reply with new term.
  476. func TestProposal(t *testing.T) {
  477. tests := []struct {
  478. *network
  479. success bool
  480. }{
  481. {newNetwork(nil, nil, nil), true},
  482. {newNetwork(nil, nil, nopStepper), true},
  483. {newNetwork(nil, nopStepper, nopStepper), false},
  484. {newNetwork(nil, nopStepper, nopStepper, nil), false},
  485. {newNetwork(nil, nopStepper, nopStepper, nil, nil), true},
  486. }
  487. for j, tt := range tests {
  488. send := func(m pb.Message) {
  489. defer func() {
  490. // only recover is we expect it to panic so
  491. // panics we don't expect go up.
  492. if !tt.success {
  493. e := recover()
  494. if e != nil {
  495. t.Logf("#%d: err: %s", j, e)
  496. }
  497. }
  498. }()
  499. tt.send(m)
  500. }
  501. data := []byte("somedata")
  502. // promote 0 the leader
  503. send(pb.Message{From: 1, To: 1, Type: pb.MsgHup})
  504. send(pb.Message{From: 1, To: 1, Type: pb.MsgProp, Entries: []pb.Entry{{Data: data}}})
  505. wantLog := newLog(NewMemoryStorage())
  506. if tt.success {
  507. wantLog = &raftLog{
  508. storage: &MemoryStorage{
  509. ents: []pb.Entry{{}, {Data: nil, Term: 1, Index: 1}, {Term: 1, Index: 2, Data: data}},
  510. },
  511. unstable: unstable{offset: 3},
  512. committed: 2}
  513. }
  514. base := ltoa(wantLog)
  515. for i, p := range tt.peers {
  516. if sm, ok := p.(*raft); ok {
  517. l := ltoa(sm.raftLog)
  518. if g := diffu(base, l); g != "" {
  519. t.Errorf("#%d: diff:\n%s", i, g)
  520. }
  521. } else {
  522. t.Logf("#%d: empty log", i)
  523. }
  524. }
  525. sm := tt.network.peers[1].(*raft)
  526. if g := sm.Term; g != 1 {
  527. t.Errorf("#%d: term = %d, want %d", j, g, 1)
  528. }
  529. }
  530. }
  531. func TestProposalByProxy(t *testing.T) {
  532. data := []byte("somedata")
  533. tests := []*network{
  534. newNetwork(nil, nil, nil),
  535. newNetwork(nil, nil, nopStepper),
  536. }
  537. for j, tt := range tests {
  538. // promote 0 the leader
  539. tt.send(pb.Message{From: 1, To: 1, Type: pb.MsgHup})
  540. // propose via follower
  541. tt.send(pb.Message{From: 2, To: 2, Type: pb.MsgProp, Entries: []pb.Entry{{Data: []byte("somedata")}}})
  542. wantLog := &raftLog{
  543. storage: &MemoryStorage{
  544. ents: []pb.Entry{{}, {Data: nil, Term: 1, Index: 1}, {Term: 1, Data: data, Index: 2}},
  545. },
  546. unstable: unstable{offset: 3},
  547. committed: 2}
  548. base := ltoa(wantLog)
  549. for i, p := range tt.peers {
  550. if sm, ok := p.(*raft); ok {
  551. l := ltoa(sm.raftLog)
  552. if g := diffu(base, l); g != "" {
  553. t.Errorf("#%d: diff:\n%s", i, g)
  554. }
  555. } else {
  556. t.Logf("#%d: empty log", i)
  557. }
  558. }
  559. sm := tt.peers[1].(*raft)
  560. if g := sm.Term; g != 1 {
  561. t.Errorf("#%d: term = %d, want %d", j, g, 1)
  562. }
  563. }
  564. }
  565. func TestCommit(t *testing.T) {
  566. tests := []struct {
  567. matches []uint64
  568. logs []pb.Entry
  569. smTerm uint64
  570. w uint64
  571. }{
  572. // single
  573. {[]uint64{1}, []pb.Entry{{Index: 1, Term: 1}}, 1, 1},
  574. {[]uint64{1}, []pb.Entry{{Index: 1, Term: 1}}, 2, 0},
  575. {[]uint64{2}, []pb.Entry{{Index: 1, Term: 1}, {Index: 2, Term: 2}}, 2, 2},
  576. {[]uint64{1}, []pb.Entry{{Index: 1, Term: 2}}, 2, 1},
  577. // odd
  578. {[]uint64{2, 1, 1}, []pb.Entry{{Index: 1, Term: 1}, {Index: 2, Term: 2}}, 1, 1},
  579. {[]uint64{2, 1, 1}, []pb.Entry{{Index: 1, Term: 1}, {Index: 2, Term: 1}}, 2, 0},
  580. {[]uint64{2, 1, 2}, []pb.Entry{{Index: 1, Term: 1}, {Index: 2, Term: 2}}, 2, 2},
  581. {[]uint64{2, 1, 2}, []pb.Entry{{Index: 1, Term: 1}, {Index: 2, Term: 1}}, 2, 0},
  582. // even
  583. {[]uint64{2, 1, 1, 1}, []pb.Entry{{Index: 1, Term: 1}, {Index: 2, Term: 2}}, 1, 1},
  584. {[]uint64{2, 1, 1, 1}, []pb.Entry{{Index: 1, Term: 1}, {Index: 2, Term: 1}}, 2, 0},
  585. {[]uint64{2, 1, 1, 2}, []pb.Entry{{Index: 1, Term: 1}, {Index: 2, Term: 2}}, 1, 1},
  586. {[]uint64{2, 1, 1, 2}, []pb.Entry{{Index: 1, Term: 1}, {Index: 2, Term: 1}}, 2, 0},
  587. {[]uint64{2, 1, 2, 2}, []pb.Entry{{Index: 1, Term: 1}, {Index: 2, Term: 2}}, 2, 2},
  588. {[]uint64{2, 1, 2, 2}, []pb.Entry{{Index: 1, Term: 1}, {Index: 2, Term: 1}}, 2, 0},
  589. }
  590. for i, tt := range tests {
  591. storage := NewMemoryStorage()
  592. storage.Append(tt.logs)
  593. storage.hardState = pb.HardState{Term: tt.smTerm}
  594. sm := newRaft(1, []uint64{1}, 5, 1, storage, 0)
  595. for j := 0; j < len(tt.matches); j++ {
  596. sm.setProgress(uint64(j)+1, tt.matches[j], tt.matches[j]+1)
  597. }
  598. sm.maybeCommit()
  599. if g := sm.raftLog.committed; g != tt.w {
  600. t.Errorf("#%d: committed = %d, want %d", i, g, tt.w)
  601. }
  602. }
  603. }
  604. func TestIsElectionTimeout(t *testing.T) {
  605. tests := []struct {
  606. elapse int
  607. wprobability float64
  608. round bool
  609. }{
  610. {5, 0, false},
  611. {13, 0.3, true},
  612. {15, 0.5, true},
  613. {18, 0.8, true},
  614. {20, 1, false},
  615. }
  616. for i, tt := range tests {
  617. sm := newRaft(1, []uint64{1}, 10, 1, NewMemoryStorage(), 0)
  618. sm.elapsed = tt.elapse
  619. c := 0
  620. for j := 0; j < 10000; j++ {
  621. if sm.isElectionTimeout() {
  622. c++
  623. }
  624. }
  625. got := float64(c) / 10000.0
  626. if tt.round {
  627. got = math.Floor(got*10+0.5) / 10.0
  628. }
  629. if got != tt.wprobability {
  630. t.Errorf("#%d: possibility = %v, want %v", i, got, tt.wprobability)
  631. }
  632. }
  633. }
  634. // ensure that the Step function ignores the message from old term and does not pass it to the
  635. // acutal stepX function.
  636. func TestStepIgnoreOldTermMsg(t *testing.T) {
  637. called := false
  638. fakeStep := func(r *raft, m pb.Message) {
  639. called = true
  640. }
  641. sm := newRaft(1, []uint64{1}, 10, 1, NewMemoryStorage(), 0)
  642. sm.step = fakeStep
  643. sm.Term = 2
  644. sm.Step(pb.Message{Type: pb.MsgApp, Term: sm.Term - 1})
  645. if called == true {
  646. t.Errorf("stepFunc called = %v , want %v", called, false)
  647. }
  648. }
  649. // TestHandleMsgApp ensures:
  650. // 1. Reply false if log doesn’t contain an entry at prevLogIndex whose term matches prevLogTerm.
  651. // 2. If an existing entry conflicts with a new one (same index but different terms),
  652. // delete the existing entry and all that follow it; append any new entries not already in the log.
  653. // 3. If leaderCommit > commitIndex, set commitIndex = min(leaderCommit, index of last new entry).
  654. func TestHandleMsgApp(t *testing.T) {
  655. tests := []struct {
  656. m pb.Message
  657. wIndex uint64
  658. wCommit uint64
  659. wReject bool
  660. }{
  661. // Ensure 1
  662. {pb.Message{Type: pb.MsgApp, Term: 2, LogTerm: 3, Index: 2, Commit: 3}, 2, 0, true}, // previous log mismatch
  663. {pb.Message{Type: pb.MsgApp, Term: 2, LogTerm: 3, Index: 3, Commit: 3}, 2, 0, true}, // previous log non-exist
  664. // Ensure 2
  665. {pb.Message{Type: pb.MsgApp, Term: 2, LogTerm: 1, Index: 1, Commit: 1}, 2, 1, false},
  666. {pb.Message{Type: pb.MsgApp, Term: 2, LogTerm: 0, Index: 0, Commit: 1, Entries: []pb.Entry{{Index: 1, Term: 2}}}, 1, 1, false},
  667. {pb.Message{Type: pb.MsgApp, Term: 2, LogTerm: 2, Index: 2, Commit: 3, Entries: []pb.Entry{{Index: 3, Term: 2}, {Index: 4, Term: 2}}}, 4, 3, false},
  668. {pb.Message{Type: pb.MsgApp, Term: 2, LogTerm: 2, Index: 2, Commit: 4, Entries: []pb.Entry{{Index: 3, Term: 2}}}, 3, 3, false},
  669. {pb.Message{Type: pb.MsgApp, Term: 2, LogTerm: 1, Index: 1, Commit: 4, Entries: []pb.Entry{{Index: 2, Term: 2}}}, 2, 2, false},
  670. // Ensure 3
  671. {pb.Message{Type: pb.MsgApp, Term: 1, LogTerm: 1, Index: 1, Commit: 3}, 2, 1, false}, // match entry 1, commit up to last new entry 1
  672. {pb.Message{Type: pb.MsgApp, Term: 1, LogTerm: 1, Index: 1, Commit: 3, Entries: []pb.Entry{{Index: 2, Term: 2}}}, 2, 2, false}, // match entry 1, commit up to last new entry 2
  673. {pb.Message{Type: pb.MsgApp, Term: 2, LogTerm: 2, Index: 2, Commit: 3}, 2, 2, false}, // match entry 2, commit up to last new entry 2
  674. {pb.Message{Type: pb.MsgApp, Term: 2, LogTerm: 2, Index: 2, Commit: 4}, 2, 2, false}, // commit up to log.last()
  675. }
  676. for i, tt := range tests {
  677. storage := NewMemoryStorage()
  678. storage.Append([]pb.Entry{{Index: 1, Term: 1}, {Index: 2, Term: 2}})
  679. sm := newRaft(1, []uint64{1}, 10, 1, storage, 0)
  680. sm.becomeFollower(2, None)
  681. sm.handleAppendEntries(tt.m)
  682. if sm.raftLog.lastIndex() != tt.wIndex {
  683. t.Errorf("#%d: lastIndex = %d, want %d", i, sm.raftLog.lastIndex(), tt.wIndex)
  684. }
  685. if sm.raftLog.committed != tt.wCommit {
  686. t.Errorf("#%d: committed = %d, want %d", i, sm.raftLog.committed, tt.wCommit)
  687. }
  688. m := sm.readMessages()
  689. if len(m) != 1 {
  690. t.Fatalf("#%d: msg = nil, want 1", i)
  691. }
  692. if m[0].Reject != tt.wReject {
  693. t.Errorf("#%d: reject = %v, want %v", i, m[0].Reject, tt.wReject)
  694. }
  695. }
  696. }
  697. // TestHandleHeartbeat ensures that the follower commits to the commit in the message.
  698. func TestHandleHeartbeat(t *testing.T) {
  699. commit := uint64(2)
  700. tests := []struct {
  701. m pb.Message
  702. wCommit uint64
  703. }{
  704. {pb.Message{From: 2, To: 1, Type: pb.MsgApp, Term: 2, Commit: commit + 1}, commit + 1},
  705. {pb.Message{From: 2, To: 1, Type: pb.MsgApp, Term: 2, Commit: commit - 1}, commit}, // do not decrease commit
  706. }
  707. for i, tt := range tests {
  708. storage := NewMemoryStorage()
  709. storage.Append([]pb.Entry{{Index: 1, Term: 1}, {Index: 2, Term: 2}, {Index: 3, Term: 3}})
  710. sm := newRaft(1, []uint64{1, 2}, 5, 1, storage, 0)
  711. sm.becomeFollower(2, 2)
  712. sm.raftLog.commitTo(commit)
  713. sm.handleHeartbeat(tt.m)
  714. if sm.raftLog.committed != tt.wCommit {
  715. t.Errorf("#%d: committed = %d, want %d", i, sm.raftLog.committed, tt.wCommit)
  716. }
  717. m := sm.readMessages()
  718. if len(m) != 1 {
  719. t.Fatalf("#%d: msg = nil, want 1", i)
  720. }
  721. if m[0].Type != pb.MsgHeartbeatResp {
  722. t.Errorf("#%d: type = %v, want MsgHeartbeatResp", i, m[0].Type)
  723. }
  724. }
  725. }
  726. // TestHandleHeartbeatResp ensures that we re-send log entries when we get a heartbeat response.
  727. func TestHandleHeartbeatResp(t *testing.T) {
  728. storage := NewMemoryStorage()
  729. storage.Append([]pb.Entry{{Index: 1, Term: 1}, {Index: 2, Term: 2}, {Index: 3, Term: 3}})
  730. sm := newRaft(1, []uint64{1, 2}, 5, 1, storage, 0)
  731. sm.becomeCandidate()
  732. sm.becomeLeader()
  733. sm.raftLog.commitTo(sm.raftLog.lastIndex())
  734. // A heartbeat response from a node that is behind; re-send MsgApp
  735. sm.Step(pb.Message{From: 2, Type: pb.MsgHeartbeatResp})
  736. msgs := sm.readMessages()
  737. if len(msgs) != 1 {
  738. t.Fatalf("len(msgs) = %d, want 1", len(msgs))
  739. }
  740. if msgs[0].Type != pb.MsgApp {
  741. t.Errorf("type = %v, want MsgApp", msgs[0].Type)
  742. }
  743. // A second heartbeat response with no AppResp does not re-send because we are in the wait state.
  744. sm.Step(pb.Message{From: 2, Type: pb.MsgHeartbeatResp})
  745. msgs = sm.readMessages()
  746. if len(msgs) != 0 {
  747. t.Fatalf("len(msgs) = %d, want 0", len(msgs))
  748. }
  749. // Send a heartbeat to reset the wait state; next heartbeat will re-send MsgApp.
  750. sm.bcastHeartbeat()
  751. sm.Step(pb.Message{From: 2, Type: pb.MsgHeartbeatResp})
  752. msgs = sm.readMessages()
  753. if len(msgs) != 2 {
  754. t.Fatalf("len(msgs) = %d, want 2", len(msgs))
  755. }
  756. if msgs[0].Type != pb.MsgHeartbeat {
  757. t.Errorf("type = %v, want MsgHeartbeat", msgs[0].Type)
  758. }
  759. if msgs[1].Type != pb.MsgApp {
  760. t.Errorf("type = %v, want MsgApp", msgs[1].Type)
  761. }
  762. // Once we have an MsgAppResp, heartbeats no longer send MsgApp.
  763. sm.Step(pb.Message{
  764. From: 2,
  765. Type: pb.MsgAppResp,
  766. Index: msgs[1].Index + uint64(len(msgs[1].Entries)),
  767. })
  768. // Consume the message sent in response to MsgAppResp
  769. sm.readMessages()
  770. sm.bcastHeartbeat() // reset wait state
  771. sm.Step(pb.Message{From: 2, Type: pb.MsgHeartbeatResp})
  772. msgs = sm.readMessages()
  773. if len(msgs) != 1 {
  774. t.Fatalf("len(msgs) = %d, want 1: %+v", len(msgs), msgs)
  775. }
  776. if msgs[0].Type != pb.MsgHeartbeat {
  777. t.Errorf("type = %v, want MsgHeartbeat", msgs[0].Type)
  778. }
  779. }
  780. // TestMsgAppRespWaitReset verifies the waitReset behavior of a leader
  781. // MsgAppResp.
  782. func TestMsgAppRespWaitReset(t *testing.T) {
  783. sm := newRaft(1, []uint64{1, 2, 3}, 5, 1, NewMemoryStorage(), 0)
  784. sm.becomeCandidate()
  785. sm.becomeLeader()
  786. // The new leader has just emitted a new Term 4 entry; consume those messages
  787. // from the outgoing queue.
  788. sm.bcastAppend()
  789. sm.readMessages()
  790. // Node 2 acks the first entry, making it committed.
  791. sm.Step(pb.Message{
  792. From: 2,
  793. Type: pb.MsgAppResp,
  794. Index: 1,
  795. })
  796. if sm.Commit != 1 {
  797. t.Fatalf("expected Commit to be 1, got %d", sm.Commit)
  798. }
  799. // Also consume the MsgApp messages that update Commit on the followers.
  800. sm.readMessages()
  801. // A new command is now proposed on node 1.
  802. sm.Step(pb.Message{
  803. From: 1,
  804. Type: pb.MsgProp,
  805. Entries: []pb.Entry{{}},
  806. })
  807. // The command is broadcast to all nodes not in the wait state.
  808. // Node 2 left the wait state due to its MsgAppResp, but node 3 is still waiting.
  809. msgs := sm.readMessages()
  810. if len(msgs) != 1 {
  811. t.Fatalf("expected 1 message, got %d: %+v", len(msgs), msgs)
  812. }
  813. if msgs[0].Type != pb.MsgApp || msgs[0].To != 2 {
  814. t.Errorf("expected MsgApp to node 2, got %s to %d", msgs[0].Type, msgs[0].To)
  815. }
  816. if len(msgs[0].Entries) != 1 || msgs[0].Entries[0].Index != 2 {
  817. t.Errorf("expected to send entry 2, but got %v", msgs[0].Entries)
  818. }
  819. // Now Node 3 acks the first entry. This releases the wait and entry 2 is sent.
  820. sm.Step(pb.Message{
  821. From: 3,
  822. Type: pb.MsgAppResp,
  823. Index: 1,
  824. })
  825. msgs = sm.readMessages()
  826. if len(msgs) != 1 {
  827. t.Fatalf("expected 1 message, got %d: %+v", len(msgs), msgs)
  828. }
  829. if msgs[0].Type != pb.MsgApp || msgs[0].To != 3 {
  830. t.Errorf("expected MsgApp to node 3, got %s to %d", msgs[0].Type, msgs[0].To)
  831. }
  832. if len(msgs[0].Entries) != 1 || msgs[0].Entries[0].Index != 2 {
  833. t.Errorf("expected to send entry 2, but got %v", msgs[0].Entries)
  834. }
  835. }
  836. func TestRecvMsgVote(t *testing.T) {
  837. tests := []struct {
  838. state StateType
  839. i, term uint64
  840. voteFor uint64
  841. wreject bool
  842. }{
  843. {StateFollower, 0, 0, None, true},
  844. {StateFollower, 0, 1, None, true},
  845. {StateFollower, 0, 2, None, true},
  846. {StateFollower, 0, 3, None, false},
  847. {StateFollower, 1, 0, None, true},
  848. {StateFollower, 1, 1, None, true},
  849. {StateFollower, 1, 2, None, true},
  850. {StateFollower, 1, 3, None, false},
  851. {StateFollower, 2, 0, None, true},
  852. {StateFollower, 2, 1, None, true},
  853. {StateFollower, 2, 2, None, false},
  854. {StateFollower, 2, 3, None, false},
  855. {StateFollower, 3, 0, None, true},
  856. {StateFollower, 3, 1, None, true},
  857. {StateFollower, 3, 2, None, false},
  858. {StateFollower, 3, 3, None, false},
  859. {StateFollower, 3, 2, 2, false},
  860. {StateFollower, 3, 2, 1, true},
  861. {StateLeader, 3, 3, 1, true},
  862. {StateCandidate, 3, 3, 1, true},
  863. }
  864. for i, tt := range tests {
  865. sm := newRaft(1, []uint64{1}, 10, 1, NewMemoryStorage(), 0)
  866. sm.state = tt.state
  867. switch tt.state {
  868. case StateFollower:
  869. sm.step = stepFollower
  870. case StateCandidate:
  871. sm.step = stepCandidate
  872. case StateLeader:
  873. sm.step = stepLeader
  874. }
  875. sm.HardState = pb.HardState{Vote: tt.voteFor}
  876. sm.raftLog = &raftLog{
  877. storage: &MemoryStorage{ents: []pb.Entry{{}, {Index: 1, Term: 2}, {Index: 2, Term: 2}}},
  878. unstable: unstable{offset: 3},
  879. }
  880. sm.Step(pb.Message{Type: pb.MsgVote, From: 2, Index: tt.i, LogTerm: tt.term})
  881. msgs := sm.readMessages()
  882. if g := len(msgs); g != 1 {
  883. t.Fatalf("#%d: len(msgs) = %d, want 1", i, g)
  884. continue
  885. }
  886. if g := msgs[0].Reject; g != tt.wreject {
  887. t.Errorf("#%d, m.Reject = %v, want %v", i, g, tt.wreject)
  888. }
  889. }
  890. }
  891. func TestStateTransition(t *testing.T) {
  892. tests := []struct {
  893. from StateType
  894. to StateType
  895. wallow bool
  896. wterm uint64
  897. wlead uint64
  898. }{
  899. {StateFollower, StateFollower, true, 1, None},
  900. {StateFollower, StateCandidate, true, 1, None},
  901. {StateFollower, StateLeader, false, 0, None},
  902. {StateCandidate, StateFollower, true, 0, None},
  903. {StateCandidate, StateCandidate, true, 1, None},
  904. {StateCandidate, StateLeader, true, 0, 1},
  905. {StateLeader, StateFollower, true, 1, None},
  906. {StateLeader, StateCandidate, false, 1, None},
  907. {StateLeader, StateLeader, true, 0, 1},
  908. }
  909. for i, tt := range tests {
  910. func() {
  911. defer func() {
  912. if r := recover(); r != nil {
  913. if tt.wallow == true {
  914. t.Errorf("%d: allow = %v, want %v", i, false, true)
  915. }
  916. }
  917. }()
  918. sm := newRaft(1, []uint64{1}, 10, 1, NewMemoryStorage(), 0)
  919. sm.state = tt.from
  920. switch tt.to {
  921. case StateFollower:
  922. sm.becomeFollower(tt.wterm, tt.wlead)
  923. case StateCandidate:
  924. sm.becomeCandidate()
  925. case StateLeader:
  926. sm.becomeLeader()
  927. }
  928. if sm.Term != tt.wterm {
  929. t.Errorf("%d: term = %d, want %d", i, sm.Term, tt.wterm)
  930. }
  931. if sm.lead != tt.wlead {
  932. t.Errorf("%d: lead = %d, want %d", i, sm.lead, tt.wlead)
  933. }
  934. }()
  935. }
  936. }
  937. func TestAllServerStepdown(t *testing.T) {
  938. tests := []struct {
  939. state StateType
  940. wstate StateType
  941. wterm uint64
  942. windex uint64
  943. }{
  944. {StateFollower, StateFollower, 3, 0},
  945. {StateCandidate, StateFollower, 3, 0},
  946. {StateLeader, StateFollower, 3, 1},
  947. }
  948. tmsgTypes := [...]pb.MessageType{pb.MsgVote, pb.MsgApp}
  949. tterm := uint64(3)
  950. for i, tt := range tests {
  951. sm := newRaft(1, []uint64{1, 2, 3}, 10, 1, NewMemoryStorage(), 0)
  952. switch tt.state {
  953. case StateFollower:
  954. sm.becomeFollower(1, None)
  955. case StateCandidate:
  956. sm.becomeCandidate()
  957. case StateLeader:
  958. sm.becomeCandidate()
  959. sm.becomeLeader()
  960. }
  961. for j, msgType := range tmsgTypes {
  962. sm.Step(pb.Message{From: 2, Type: msgType, Term: tterm, LogTerm: tterm})
  963. if sm.state != tt.wstate {
  964. t.Errorf("#%d.%d state = %v , want %v", i, j, sm.state, tt.wstate)
  965. }
  966. if sm.Term != tt.wterm {
  967. t.Errorf("#%d.%d term = %v , want %v", i, j, sm.Term, tt.wterm)
  968. }
  969. if uint64(sm.raftLog.lastIndex()) != tt.windex {
  970. t.Errorf("#%d.%d index = %v , want %v", i, j, sm.raftLog.lastIndex(), tt.windex)
  971. }
  972. if uint64(len(sm.raftLog.allEntries())) != tt.windex {
  973. t.Errorf("#%d.%d len(ents) = %v , want %v", i, j, len(sm.raftLog.allEntries()), tt.windex)
  974. }
  975. wlead := uint64(2)
  976. if msgType == pb.MsgVote {
  977. wlead = None
  978. }
  979. if sm.lead != wlead {
  980. t.Errorf("#%d, sm.lead = %d, want %d", i, sm.lead, None)
  981. }
  982. }
  983. }
  984. }
  985. func TestLeaderAppResp(t *testing.T) {
  986. // initial progress: match = 0; next = 3
  987. tests := []struct {
  988. index uint64
  989. reject bool
  990. // progress
  991. wmatch uint64
  992. wnext uint64
  993. // message
  994. wmsgNum int
  995. windex uint64
  996. wcommitted uint64
  997. }{
  998. {3, true, 0, 3, 0, 0, 0}, // stale resp; no replies
  999. {2, true, 0, 2, 1, 1, 0}, // denied resp; leader does not commit; decrease next and send probing msg
  1000. {2, false, 2, 4, 2, 2, 2}, // accept resp; leader commits; broadcast with commit index
  1001. {0, false, 0, 3, 0, 0, 0}, // ignore heartbeat replies
  1002. }
  1003. for i, tt := range tests {
  1004. // sm term is 1 after it becomes the leader.
  1005. // thus the last log term must be 1 to be committed.
  1006. sm := newRaft(1, []uint64{1, 2, 3}, 10, 1, NewMemoryStorage(), 0)
  1007. sm.raftLog = &raftLog{
  1008. storage: &MemoryStorage{ents: []pb.Entry{{}, {Index: 1, Term: 0}, {Index: 2, Term: 1}}},
  1009. unstable: unstable{offset: 3},
  1010. }
  1011. sm.becomeCandidate()
  1012. sm.becomeLeader()
  1013. sm.readMessages()
  1014. sm.Step(pb.Message{From: 2, Type: pb.MsgAppResp, Index: tt.index, Term: sm.Term, Reject: tt.reject, RejectHint: tt.index})
  1015. p := sm.prs[2]
  1016. if p.Match != tt.wmatch {
  1017. t.Errorf("#%d match = %d, want %d", i, p.Match, tt.wmatch)
  1018. }
  1019. if p.Next != tt.wnext {
  1020. t.Errorf("#%d next = %d, want %d", i, p.Next, tt.wnext)
  1021. }
  1022. msgs := sm.readMessages()
  1023. if len(msgs) != tt.wmsgNum {
  1024. t.Errorf("#%d msgNum = %d, want %d", i, len(msgs), tt.wmsgNum)
  1025. }
  1026. for j, msg := range msgs {
  1027. if msg.Index != tt.windex {
  1028. t.Errorf("#%d.%d index = %d, want %d", i, j, msg.Index, tt.windex)
  1029. }
  1030. if msg.Commit != tt.wcommitted {
  1031. t.Errorf("#%d.%d commit = %d, want %d", i, j, msg.Commit, tt.wcommitted)
  1032. }
  1033. }
  1034. }
  1035. }
  1036. // When the leader receives a heartbeat tick, it should
  1037. // send a MsgApp with m.Index = 0, m.LogTerm=0 and empty entries.
  1038. func TestBcastBeat(t *testing.T) {
  1039. offset := uint64(1000)
  1040. // make a state machine with log.offset = 1000
  1041. s := pb.Snapshot{
  1042. Metadata: pb.SnapshotMetadata{
  1043. Index: offset,
  1044. Term: 1,
  1045. ConfState: pb.ConfState{Nodes: []uint64{1, 2, 3}},
  1046. },
  1047. }
  1048. storage := NewMemoryStorage()
  1049. storage.ApplySnapshot(s)
  1050. sm := newRaft(1, nil, 10, 1, storage, 0)
  1051. sm.Term = 1
  1052. sm.becomeCandidate()
  1053. sm.becomeLeader()
  1054. for i := 0; i < 10; i++ {
  1055. sm.appendEntry(pb.Entry{Index: uint64(i) + 1})
  1056. }
  1057. // slow follower
  1058. sm.prs[2].Match, sm.prs[2].Next = 5, 6
  1059. // normal follower
  1060. sm.prs[3].Match, sm.prs[3].Next = sm.raftLog.lastIndex(), sm.raftLog.lastIndex()+1
  1061. sm.Step(pb.Message{Type: pb.MsgBeat})
  1062. msgs := sm.readMessages()
  1063. if len(msgs) != 2 {
  1064. t.Fatalf("len(msgs) = %v, want 2", len(msgs))
  1065. }
  1066. wantCommitMap := map[uint64]uint64{
  1067. 2: min(sm.raftLog.committed, sm.prs[2].Match),
  1068. 3: min(sm.raftLog.committed, sm.prs[3].Match),
  1069. }
  1070. for i, m := range msgs {
  1071. if m.Type != pb.MsgHeartbeat {
  1072. t.Fatalf("#%d: type = %v, want = %v", i, m.Type, pb.MsgHeartbeat)
  1073. }
  1074. if m.Index != 0 {
  1075. t.Fatalf("#%d: prevIndex = %d, want %d", i, m.Index, 0)
  1076. }
  1077. if m.LogTerm != 0 {
  1078. t.Fatalf("#%d: prevTerm = %d, want %d", i, m.LogTerm, 0)
  1079. }
  1080. if wantCommitMap[m.To] == 0 {
  1081. t.Fatalf("#%d: unexpected to %d", i, m.To)
  1082. } else {
  1083. if m.Commit != wantCommitMap[m.To] {
  1084. t.Fatalf("#%d: commit = %d, want %d", i, m.Commit, wantCommitMap[m.To])
  1085. }
  1086. delete(wantCommitMap, m.To)
  1087. }
  1088. if len(m.Entries) != 0 {
  1089. t.Fatalf("#%d: len(entries) = %d, want 0", i, len(m.Entries))
  1090. }
  1091. }
  1092. }
  1093. // tests the output of the statemachine when receiving MsgBeat
  1094. func TestRecvMsgBeat(t *testing.T) {
  1095. tests := []struct {
  1096. state StateType
  1097. wMsg int
  1098. }{
  1099. {StateLeader, 2},
  1100. // candidate and follower should ignore MsgBeat
  1101. {StateCandidate, 0},
  1102. {StateFollower, 0},
  1103. }
  1104. for i, tt := range tests {
  1105. sm := newRaft(1, []uint64{1, 2, 3}, 10, 1, NewMemoryStorage(), 0)
  1106. sm.raftLog = &raftLog{storage: &MemoryStorage{ents: []pb.Entry{{}, {Index: 1, Term: 0}, {Index: 2, Term: 1}}}}
  1107. sm.Term = 1
  1108. sm.state = tt.state
  1109. switch tt.state {
  1110. case StateFollower:
  1111. sm.step = stepFollower
  1112. case StateCandidate:
  1113. sm.step = stepCandidate
  1114. case StateLeader:
  1115. sm.step = stepLeader
  1116. }
  1117. sm.Step(pb.Message{From: 1, To: 1, Type: pb.MsgBeat})
  1118. msgs := sm.readMessages()
  1119. if len(msgs) != tt.wMsg {
  1120. t.Errorf("%d: len(msgs) = %d, want %d", i, len(msgs), tt.wMsg)
  1121. }
  1122. for _, m := range msgs {
  1123. if m.Type != pb.MsgHeartbeat {
  1124. t.Errorf("%d: msg.type = %v, want %v", i, m.Type, pb.MsgHeartbeat)
  1125. }
  1126. }
  1127. }
  1128. }
  1129. func TestLeaderIncreaseNext(t *testing.T) {
  1130. previousEnts := []pb.Entry{{Term: 1, Index: 1}, {Term: 1, Index: 2}, {Term: 1, Index: 3}}
  1131. tests := []struct {
  1132. // progress
  1133. match uint64
  1134. next uint64
  1135. wnext uint64
  1136. }{
  1137. // match is not zero, optimistically increase next
  1138. // previous entries + noop entry + propose + 1
  1139. {1, 2, uint64(len(previousEnts) + 1 + 1 + 1)},
  1140. // match is zero, not optimistically increase next
  1141. {0, 2, 2},
  1142. }
  1143. for i, tt := range tests {
  1144. sm := newRaft(1, []uint64{1, 2}, 10, 1, NewMemoryStorage(), 0)
  1145. sm.raftLog.append(previousEnts...)
  1146. sm.becomeCandidate()
  1147. sm.becomeLeader()
  1148. sm.prs[2].Match, sm.prs[2].Next = tt.match, tt.next
  1149. sm.Step(pb.Message{From: 1, To: 1, Type: pb.MsgProp, Entries: []pb.Entry{{Data: []byte("somedata")}}})
  1150. p := sm.prs[2]
  1151. if p.Next != tt.wnext {
  1152. t.Errorf("#%d next = %d, want %d", i, p.Next, tt.wnext)
  1153. }
  1154. }
  1155. }
  1156. func TestRestore(t *testing.T) {
  1157. s := pb.Snapshot{
  1158. Metadata: pb.SnapshotMetadata{
  1159. Index: 11, // magic number
  1160. Term: 11, // magic number
  1161. ConfState: pb.ConfState{Nodes: []uint64{1, 2, 3}},
  1162. },
  1163. }
  1164. storage := NewMemoryStorage()
  1165. sm := newRaft(1, []uint64{1, 2}, 10, 1, storage, 0)
  1166. if ok := sm.restore(s); !ok {
  1167. t.Fatal("restore fail, want succeed")
  1168. }
  1169. if sm.raftLog.lastIndex() != s.Metadata.Index {
  1170. t.Errorf("log.lastIndex = %d, want %d", sm.raftLog.lastIndex(), s.Metadata.Index)
  1171. }
  1172. if sm.raftLog.term(s.Metadata.Index) != s.Metadata.Term {
  1173. t.Errorf("log.lastTerm = %d, want %d", sm.raftLog.term(s.Metadata.Index), s.Metadata.Term)
  1174. }
  1175. sg := sm.nodes()
  1176. if !reflect.DeepEqual(sg, s.Metadata.ConfState.Nodes) {
  1177. t.Errorf("sm.Nodes = %+v, want %+v", sg, s.Metadata.ConfState.Nodes)
  1178. }
  1179. if ok := sm.restore(s); ok {
  1180. t.Fatal("restore succeed, want fail")
  1181. }
  1182. }
  1183. func TestRestoreIgnoreSnapshot(t *testing.T) {
  1184. previousEnts := []pb.Entry{{Term: 1, Index: 1}, {Term: 1, Index: 2}, {Term: 1, Index: 3}}
  1185. commit := uint64(1)
  1186. storage := NewMemoryStorage()
  1187. sm := newRaft(1, []uint64{1, 2}, 10, 1, storage, 0)
  1188. sm.raftLog.append(previousEnts...)
  1189. sm.raftLog.commitTo(commit)
  1190. s := pb.Snapshot{
  1191. Metadata: pb.SnapshotMetadata{
  1192. Index: commit,
  1193. Term: 1,
  1194. ConfState: pb.ConfState{Nodes: []uint64{1, 2}},
  1195. },
  1196. }
  1197. // ignore snapshot
  1198. if ok := sm.restore(s); ok {
  1199. t.Errorf("restore = %t, want %t", ok, false)
  1200. }
  1201. if sm.raftLog.committed != commit {
  1202. t.Errorf("commit = %d, want %d", sm.raftLog.committed, commit)
  1203. }
  1204. // ignore snapshot and fast forward commit
  1205. s.Metadata.Index = commit + 1
  1206. if ok := sm.restore(s); ok {
  1207. t.Errorf("restore = %t, want %t", ok, false)
  1208. }
  1209. if sm.raftLog.committed != commit+1 {
  1210. t.Errorf("commit = %d, want %d", sm.raftLog.committed, commit+1)
  1211. }
  1212. }
  1213. func TestProvideSnap(t *testing.T) {
  1214. // restore the statemachin from a snapshot
  1215. // so it has a compacted log and a snapshot
  1216. s := pb.Snapshot{
  1217. Metadata: pb.SnapshotMetadata{
  1218. Index: 11, // magic number
  1219. Term: 11, // magic number
  1220. ConfState: pb.ConfState{Nodes: []uint64{1, 2}},
  1221. },
  1222. }
  1223. storage := NewMemoryStorage()
  1224. sm := newRaft(1, []uint64{1}, 10, 1, storage, 0)
  1225. sm.restore(s)
  1226. sm.becomeCandidate()
  1227. sm.becomeLeader()
  1228. // force set the next of node 1, so that
  1229. // node 1 needs a snapshot
  1230. sm.prs[2].Next = sm.raftLog.firstIndex()
  1231. sm.Step(pb.Message{From: 2, To: 1, Type: pb.MsgAppResp, Index: sm.prs[2].Next - 1, Reject: true})
  1232. msgs := sm.readMessages()
  1233. if len(msgs) != 1 {
  1234. t.Fatalf("len(msgs) = %d, want 1", len(msgs))
  1235. }
  1236. m := msgs[0]
  1237. if m.Type != pb.MsgSnap {
  1238. t.Errorf("m.Type = %v, want %v", m.Type, pb.MsgSnap)
  1239. }
  1240. }
  1241. func TestRestoreFromSnapMsg(t *testing.T) {
  1242. s := pb.Snapshot{
  1243. Metadata: pb.SnapshotMetadata{
  1244. Index: 11, // magic number
  1245. Term: 11, // magic number
  1246. ConfState: pb.ConfState{Nodes: []uint64{1, 2}},
  1247. },
  1248. }
  1249. m := pb.Message{Type: pb.MsgSnap, From: 1, Term: 2, Snapshot: s}
  1250. sm := newRaft(2, []uint64{1, 2}, 10, 1, NewMemoryStorage(), 0)
  1251. sm.Step(m)
  1252. // TODO(bdarnell): what should this test?
  1253. }
  1254. func TestSlowNodeRestore(t *testing.T) {
  1255. nt := newNetwork(nil, nil, nil)
  1256. nt.send(pb.Message{From: 1, To: 1, Type: pb.MsgHup})
  1257. nt.isolate(3)
  1258. for j := 0; j <= 100; j++ {
  1259. nt.send(pb.Message{From: 1, To: 1, Type: pb.MsgProp, Entries: []pb.Entry{{}}})
  1260. }
  1261. lead := nt.peers[1].(*raft)
  1262. nextEnts(lead, nt.storage[1])
  1263. nt.storage[1].Compact(lead.raftLog.applied, &pb.ConfState{Nodes: lead.nodes()}, nil)
  1264. nt.recover()
  1265. // trigger a snapshot
  1266. nt.send(pb.Message{From: 1, To: 1, Type: pb.MsgProp, Entries: []pb.Entry{{}}})
  1267. follower := nt.peers[3].(*raft)
  1268. // trigger a commit
  1269. nt.send(pb.Message{From: 1, To: 1, Type: pb.MsgProp, Entries: []pb.Entry{{}}})
  1270. if follower.raftLog.committed != lead.raftLog.committed {
  1271. t.Errorf("follower.committed = %d, want %d", follower.raftLog.committed, lead.raftLog.committed)
  1272. }
  1273. }
  1274. // TestStepConfig tests that when raft step msgProp in EntryConfChange type,
  1275. // it appends the entry to log and sets pendingConf to be true.
  1276. func TestStepConfig(t *testing.T) {
  1277. // a raft that cannot make progress
  1278. r := newRaft(1, []uint64{1, 2}, 10, 1, NewMemoryStorage(), 0)
  1279. r.becomeCandidate()
  1280. r.becomeLeader()
  1281. index := r.raftLog.lastIndex()
  1282. r.Step(pb.Message{From: 1, To: 1, Type: pb.MsgProp, Entries: []pb.Entry{{Type: pb.EntryConfChange}}})
  1283. if g := r.raftLog.lastIndex(); g != index+1 {
  1284. t.Errorf("index = %d, want %d", g, index+1)
  1285. }
  1286. if r.pendingConf != true {
  1287. t.Errorf("pendingConf = %v, want true", r.pendingConf)
  1288. }
  1289. }
  1290. // TestStepIgnoreConfig tests that if raft step the second msgProp in
  1291. // EntryConfChange type when the first one is uncommitted, the node will set
  1292. // the proposal to noop and keep its original state.
  1293. func TestStepIgnoreConfig(t *testing.T) {
  1294. // a raft that cannot make progress
  1295. r := newRaft(1, []uint64{1, 2}, 10, 1, NewMemoryStorage(), 0)
  1296. r.becomeCandidate()
  1297. r.becomeLeader()
  1298. r.Step(pb.Message{From: 1, To: 1, Type: pb.MsgProp, Entries: []pb.Entry{{Type: pb.EntryConfChange}}})
  1299. index := r.raftLog.lastIndex()
  1300. pendingConf := r.pendingConf
  1301. r.Step(pb.Message{From: 1, To: 1, Type: pb.MsgProp, Entries: []pb.Entry{{Type: pb.EntryConfChange}}})
  1302. wents := []pb.Entry{{Type: pb.EntryNormal, Term: 1, Index: 3, Data: nil}}
  1303. if ents := r.raftLog.entries(index + 1); !reflect.DeepEqual(ents, wents) {
  1304. t.Errorf("ents = %+v, want %+v", ents, wents)
  1305. }
  1306. if r.pendingConf != pendingConf {
  1307. t.Errorf("pendingConf = %v, want %v", r.pendingConf, pendingConf)
  1308. }
  1309. }
  1310. // TestRecoverPendingConfig tests that new leader recovers its pendingConf flag
  1311. // based on uncommitted entries.
  1312. func TestRecoverPendingConfig(t *testing.T) {
  1313. tests := []struct {
  1314. entType pb.EntryType
  1315. wpending bool
  1316. }{
  1317. {pb.EntryNormal, false},
  1318. {pb.EntryConfChange, true},
  1319. }
  1320. for i, tt := range tests {
  1321. r := newRaft(1, []uint64{1, 2}, 10, 1, NewMemoryStorage(), 0)
  1322. r.appendEntry(pb.Entry{Type: tt.entType})
  1323. r.becomeCandidate()
  1324. r.becomeLeader()
  1325. if r.pendingConf != tt.wpending {
  1326. t.Errorf("#%d: pendingConf = %v, want %v", i, r.pendingConf, tt.wpending)
  1327. }
  1328. }
  1329. }
  1330. // TestRecoverDoublePendingConfig tests that new leader will panic if
  1331. // there exist two uncommitted config entries.
  1332. func TestRecoverDoublePendingConfig(t *testing.T) {
  1333. func() {
  1334. defer func() {
  1335. if err := recover(); err == nil {
  1336. t.Errorf("expect panic, but nothing happens")
  1337. }
  1338. }()
  1339. r := newRaft(1, []uint64{1, 2}, 10, 1, NewMemoryStorage(), 0)
  1340. r.appendEntry(pb.Entry{Type: pb.EntryConfChange})
  1341. r.appendEntry(pb.Entry{Type: pb.EntryConfChange})
  1342. r.becomeCandidate()
  1343. r.becomeLeader()
  1344. }()
  1345. }
  1346. // TestAddNode tests that addNode could update pendingConf and nodes correctly.
  1347. func TestAddNode(t *testing.T) {
  1348. r := newRaft(1, []uint64{1}, 10, 1, NewMemoryStorage(), 0)
  1349. r.pendingConf = true
  1350. r.addNode(2)
  1351. if r.pendingConf != false {
  1352. t.Errorf("pendingConf = %v, want false", r.pendingConf)
  1353. }
  1354. nodes := r.nodes()
  1355. wnodes := []uint64{1, 2}
  1356. if !reflect.DeepEqual(nodes, wnodes) {
  1357. t.Errorf("nodes = %v, want %v", nodes, wnodes)
  1358. }
  1359. }
  1360. // TestRemoveNode tests that removeNode could update pendingConf, nodes and
  1361. // and removed list correctly.
  1362. func TestRemoveNode(t *testing.T) {
  1363. r := newRaft(1, []uint64{1, 2}, 10, 1, NewMemoryStorage(), 0)
  1364. r.pendingConf = true
  1365. r.removeNode(2)
  1366. if r.pendingConf != false {
  1367. t.Errorf("pendingConf = %v, want false", r.pendingConf)
  1368. }
  1369. w := []uint64{1}
  1370. if g := r.nodes(); !reflect.DeepEqual(g, w) {
  1371. t.Errorf("nodes = %v, want %v", g, w)
  1372. }
  1373. }
  1374. func TestPromotable(t *testing.T) {
  1375. id := uint64(1)
  1376. tests := []struct {
  1377. peers []uint64
  1378. wp bool
  1379. }{
  1380. {[]uint64{1}, true},
  1381. {[]uint64{1, 2, 3}, true},
  1382. {[]uint64{}, false},
  1383. {[]uint64{2, 3}, false},
  1384. }
  1385. for i, tt := range tests {
  1386. r := newRaft(id, tt.peers, 5, 1, NewMemoryStorage(), 0)
  1387. if g := r.promotable(); g != tt.wp {
  1388. t.Errorf("#%d: promotable = %v, want %v", i, g, tt.wp)
  1389. }
  1390. }
  1391. }
  1392. func TestRaftNodes(t *testing.T) {
  1393. tests := []struct {
  1394. ids []uint64
  1395. wids []uint64
  1396. }{
  1397. {
  1398. []uint64{1, 2, 3},
  1399. []uint64{1, 2, 3},
  1400. },
  1401. {
  1402. []uint64{3, 2, 1},
  1403. []uint64{1, 2, 3},
  1404. },
  1405. }
  1406. for i, tt := range tests {
  1407. r := newRaft(1, tt.ids, 10, 1, NewMemoryStorage(), 0)
  1408. if !reflect.DeepEqual(r.nodes(), tt.wids) {
  1409. t.Errorf("#%d: nodes = %+v, want %+v", i, r.nodes(), tt.wids)
  1410. }
  1411. }
  1412. }
  1413. func ents(terms ...uint64) *raft {
  1414. storage := NewMemoryStorage()
  1415. for i, term := range terms {
  1416. storage.Append([]pb.Entry{{Index: uint64(i + 1), Term: term}})
  1417. }
  1418. sm := newRaft(1, []uint64{}, 5, 1, storage, 0)
  1419. sm.reset(0)
  1420. return sm
  1421. }
  1422. type network struct {
  1423. peers map[uint64]Interface
  1424. storage map[uint64]*MemoryStorage
  1425. dropm map[connem]float64
  1426. ignorem map[pb.MessageType]bool
  1427. }
  1428. // newNetwork initializes a network from peers.
  1429. // A nil node will be replaced with a new *stateMachine.
  1430. // A *stateMachine will get its k, id.
  1431. // When using stateMachine, the address list is always [1, n].
  1432. func newNetwork(peers ...Interface) *network {
  1433. size := len(peers)
  1434. peerAddrs := idsBySize(size)
  1435. npeers := make(map[uint64]Interface, size)
  1436. nstorage := make(map[uint64]*MemoryStorage, size)
  1437. for j, p := range peers {
  1438. id := peerAddrs[j]
  1439. switch v := p.(type) {
  1440. case nil:
  1441. nstorage[id] = NewMemoryStorage()
  1442. sm := newRaft(id, peerAddrs, 10, 1, nstorage[id], 0)
  1443. npeers[id] = sm
  1444. case *raft:
  1445. v.id = id
  1446. v.prs = make(map[uint64]*Progress)
  1447. for i := 0; i < size; i++ {
  1448. v.prs[peerAddrs[i]] = &Progress{}
  1449. }
  1450. v.reset(0)
  1451. npeers[id] = v
  1452. case *blackHole:
  1453. npeers[id] = v
  1454. default:
  1455. panic(fmt.Sprintf("unexpected state machine type: %T", p))
  1456. }
  1457. }
  1458. return &network{
  1459. peers: npeers,
  1460. storage: nstorage,
  1461. dropm: make(map[connem]float64),
  1462. ignorem: make(map[pb.MessageType]bool),
  1463. }
  1464. }
  1465. func (nw *network) send(msgs ...pb.Message) {
  1466. for len(msgs) > 0 {
  1467. m := msgs[0]
  1468. p := nw.peers[m.To]
  1469. p.Step(m)
  1470. msgs = append(msgs[1:], nw.filter(p.readMessages())...)
  1471. }
  1472. }
  1473. func (nw *network) drop(from, to uint64, perc float64) {
  1474. nw.dropm[connem{from, to}] = perc
  1475. }
  1476. func (nw *network) cut(one, other uint64) {
  1477. nw.drop(one, other, 1)
  1478. nw.drop(other, one, 1)
  1479. }
  1480. func (nw *network) isolate(id uint64) {
  1481. for i := 0; i < len(nw.peers); i++ {
  1482. nid := uint64(i) + 1
  1483. if nid != id {
  1484. nw.drop(id, nid, 1.0)
  1485. nw.drop(nid, id, 1.0)
  1486. }
  1487. }
  1488. }
  1489. func (nw *network) ignore(t pb.MessageType) {
  1490. nw.ignorem[t] = true
  1491. }
  1492. func (nw *network) recover() {
  1493. nw.dropm = make(map[connem]float64)
  1494. nw.ignorem = make(map[pb.MessageType]bool)
  1495. }
  1496. func (nw *network) filter(msgs []pb.Message) []pb.Message {
  1497. mm := []pb.Message{}
  1498. for _, m := range msgs {
  1499. if nw.ignorem[m.Type] {
  1500. continue
  1501. }
  1502. switch m.Type {
  1503. case pb.MsgHup:
  1504. // hups never go over the network, so don't drop them but panic
  1505. panic("unexpected msgHup")
  1506. default:
  1507. perc := nw.dropm[connem{m.From, m.To}]
  1508. if n := rand.Float64(); n < perc {
  1509. continue
  1510. }
  1511. }
  1512. mm = append(mm, m)
  1513. }
  1514. return mm
  1515. }
  1516. type connem struct {
  1517. from, to uint64
  1518. }
  1519. type blackHole struct{}
  1520. func (blackHole) Step(pb.Message) error { return nil }
  1521. func (blackHole) readMessages() []pb.Message { return nil }
  1522. var nopStepper = &blackHole{}
  1523. func idsBySize(size int) []uint64 {
  1524. ids := make([]uint64, size)
  1525. for i := 0; i < size; i++ {
  1526. ids[i] = 1 + uint64(i)
  1527. }
  1528. return ids
  1529. }