raft_test.go 34 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269
  1. package raft
  2. import (
  3. "bytes"
  4. "fmt"
  5. "math/rand"
  6. "reflect"
  7. "sort"
  8. "testing"
  9. pb "github.com/coreos/etcd/raft/raftpb"
  10. )
  11. // nextEnts returns the appliable entries and updates the applied index
  12. func nextEnts(r *raft) (ents []pb.Entry) {
  13. ents = r.raftLog.nextEnts()
  14. r.raftLog.resetNextEnts()
  15. return ents
  16. }
  17. type Interface interface {
  18. Step(m pb.Message) error
  19. ReadMessages() []pb.Message
  20. }
  21. func TestLeaderElection(t *testing.T) {
  22. tests := []struct {
  23. *network
  24. state StateType
  25. }{
  26. {newNetwork(nil, nil, nil), StateLeader},
  27. {newNetwork(nil, nil, nopStepper), StateLeader},
  28. {newNetwork(nil, nopStepper, nopStepper), StateCandidate},
  29. {newNetwork(nil, nopStepper, nopStepper, nil), StateCandidate},
  30. {newNetwork(nil, nopStepper, nopStepper, nil, nil), StateLeader},
  31. // three logs further along than 0
  32. {newNetwork(nil, ents(1), ents(2), ents(1, 3), nil), StateFollower},
  33. // logs converge
  34. {newNetwork(ents(1), nil, ents(2), ents(1), nil), StateLeader},
  35. }
  36. for i, tt := range tests {
  37. tt.send(pb.Message{From: 1, To: 1, Type: msgHup})
  38. sm := tt.network.peers[1].(*raft)
  39. if sm.state != tt.state {
  40. t.Errorf("#%d: state = %s, want %s", i, sm.state, tt.state)
  41. }
  42. if g := sm.Term; g != 1 {
  43. t.Errorf("#%d: term = %d, want %d", i, g, 1)
  44. }
  45. }
  46. }
  47. func TestLogReplication(t *testing.T) {
  48. tests := []struct {
  49. *network
  50. msgs []pb.Message
  51. wcommitted int64
  52. }{
  53. {
  54. newNetwork(nil, nil, nil),
  55. []pb.Message{
  56. {From: 1, To: 1, Type: msgProp, Entries: []pb.Entry{{Data: []byte("somedata")}}},
  57. },
  58. 2,
  59. },
  60. {
  61. newNetwork(nil, nil, nil),
  62. []pb.Message{
  63. {From: 1, To: 1, Type: msgProp, Entries: []pb.Entry{{Data: []byte("somedata")}}},
  64. {From: 1, To: 2, Type: msgHup},
  65. {From: 1, To: 2, Type: msgProp, Entries: []pb.Entry{{Data: []byte("somedata")}}},
  66. },
  67. 4,
  68. },
  69. }
  70. for i, tt := range tests {
  71. tt.send(pb.Message{From: 1, To: 1, Type: msgHup})
  72. for _, m := range tt.msgs {
  73. tt.send(m)
  74. }
  75. for j, x := range tt.network.peers {
  76. sm := x.(*raft)
  77. if sm.raftLog.committed != tt.wcommitted {
  78. t.Errorf("#%d.%d: committed = %d, want %d", i, j, sm.raftLog.committed, tt.wcommitted)
  79. }
  80. ents := []pb.Entry{}
  81. for _, e := range nextEnts(sm) {
  82. if e.Data != nil {
  83. ents = append(ents, e)
  84. }
  85. }
  86. props := []pb.Message{}
  87. for _, m := range tt.msgs {
  88. if m.Type == msgProp {
  89. props = append(props, m)
  90. }
  91. }
  92. for k, m := range props {
  93. if !bytes.Equal(ents[k].Data, m.Entries[0].Data) {
  94. t.Errorf("#%d.%d: data = %d, want %d", i, j, ents[k].Data, m.Entries[0].Data)
  95. }
  96. }
  97. }
  98. }
  99. }
  100. func TestSingleNodeCommit(t *testing.T) {
  101. tt := newNetwork(nil)
  102. tt.send(pb.Message{From: 1, To: 1, Type: msgHup})
  103. tt.send(pb.Message{From: 1, To: 1, Type: msgProp, Entries: []pb.Entry{{Data: []byte("some data")}}})
  104. tt.send(pb.Message{From: 1, To: 1, Type: msgProp, Entries: []pb.Entry{{Data: []byte("some data")}}})
  105. sm := tt.peers[1].(*raft)
  106. if sm.raftLog.committed != 3 {
  107. t.Errorf("committed = %d, want %d", sm.raftLog.committed, 3)
  108. }
  109. }
  110. // TestCannotCommitWithoutNewTermEntry tests the entries cannot be committed
  111. // when leader changes, no new proposal comes in and ChangeTerm proposal is
  112. // filtered.
  113. func TestCannotCommitWithoutNewTermEntry(t *testing.T) {
  114. tt := newNetwork(nil, nil, nil, nil, nil)
  115. tt.send(pb.Message{From: 1, To: 1, Type: msgHup})
  116. // 0 cannot reach 2,3,4
  117. tt.cut(1, 3)
  118. tt.cut(1, 4)
  119. tt.cut(1, 5)
  120. tt.send(pb.Message{From: 1, To: 1, Type: msgProp, Entries: []pb.Entry{{Data: []byte("some data")}}})
  121. tt.send(pb.Message{From: 1, To: 1, Type: msgProp, Entries: []pb.Entry{{Data: []byte("some data")}}})
  122. sm := tt.peers[1].(*raft)
  123. if sm.raftLog.committed != 1 {
  124. t.Errorf("committed = %d, want %d", sm.raftLog.committed, 1)
  125. }
  126. // network recovery
  127. tt.recover()
  128. // avoid committing ChangeTerm proposal
  129. tt.ignore(msgApp)
  130. // elect 1 as the new leader with term 2
  131. tt.send(pb.Message{From: 2, To: 2, Type: msgHup})
  132. // no log entries from previous term should be committed
  133. sm = tt.peers[2].(*raft)
  134. if sm.raftLog.committed != 1 {
  135. t.Errorf("committed = %d, want %d", sm.raftLog.committed, 1)
  136. }
  137. tt.recover()
  138. // still be able to append a entry
  139. tt.send(pb.Message{From: 2, To: 2, Type: msgProp, Entries: []pb.Entry{{Data: []byte("some data")}}})
  140. if sm.raftLog.committed != 5 {
  141. t.Errorf("committed = %d, want %d", sm.raftLog.committed, 5)
  142. }
  143. }
  144. // TestCommitWithoutNewTermEntry tests the entries could be committed
  145. // when leader changes, no new proposal comes in.
  146. func TestCommitWithoutNewTermEntry(t *testing.T) {
  147. tt := newNetwork(nil, nil, nil, nil, nil)
  148. tt.send(pb.Message{From: 1, To: 1, Type: msgHup})
  149. // 0 cannot reach 2,3,4
  150. tt.cut(1, 3)
  151. tt.cut(1, 4)
  152. tt.cut(1, 5)
  153. tt.send(pb.Message{From: 1, To: 1, Type: msgProp, Entries: []pb.Entry{{Data: []byte("some data")}}})
  154. tt.send(pb.Message{From: 1, To: 1, Type: msgProp, Entries: []pb.Entry{{Data: []byte("some data")}}})
  155. sm := tt.peers[1].(*raft)
  156. if sm.raftLog.committed != 1 {
  157. t.Errorf("committed = %d, want %d", sm.raftLog.committed, 1)
  158. }
  159. // network recovery
  160. tt.recover()
  161. // elect 1 as the new leader with term 2
  162. // after append a ChangeTerm entry from the current term, all entries
  163. // should be committed
  164. tt.send(pb.Message{From: 2, To: 2, Type: msgHup})
  165. if sm.raftLog.committed != 4 {
  166. t.Errorf("committed = %d, want %d", sm.raftLog.committed, 4)
  167. }
  168. }
  169. func TestDuelingCandidates(t *testing.T) {
  170. a := newRaft(-1, nil, 0, 0) // k, id are set later
  171. b := newRaft(-1, nil, 0, 0)
  172. c := newRaft(-1, nil, 0, 0)
  173. nt := newNetwork(a, b, c)
  174. nt.cut(1, 3)
  175. nt.send(pb.Message{From: 1, To: 1, Type: msgHup})
  176. nt.send(pb.Message{From: 3, To: 3, Type: msgHup})
  177. nt.recover()
  178. nt.send(pb.Message{From: 3, To: 3, Type: msgHup})
  179. wlog := &raftLog{ents: []pb.Entry{{}, pb.Entry{Data: nil, Term: 1, Index: 1}}, committed: 1}
  180. tests := []struct {
  181. sm *raft
  182. state StateType
  183. term int64
  184. raftLog *raftLog
  185. }{
  186. {a, StateFollower, 2, wlog},
  187. {b, StateFollower, 2, wlog},
  188. {c, StateFollower, 2, newLog()},
  189. }
  190. for i, tt := range tests {
  191. if g := tt.sm.state; g != tt.state {
  192. t.Errorf("#%d: state = %s, want %s", i, g, tt.state)
  193. }
  194. if g := tt.sm.Term; g != tt.term {
  195. t.Errorf("#%d: term = %d, want %d", i, g, tt.term)
  196. }
  197. base := ltoa(tt.raftLog)
  198. if sm, ok := nt.peers[1+int64(i)].(*raft); ok {
  199. l := ltoa(sm.raftLog)
  200. if g := diffu(base, l); g != "" {
  201. t.Errorf("#%d: diff:\n%s", i, g)
  202. }
  203. } else {
  204. t.Logf("#%d: empty log", i)
  205. }
  206. }
  207. }
  208. func TestCandidateConcede(t *testing.T) {
  209. tt := newNetwork(nil, nil, nil)
  210. tt.isolate(1)
  211. tt.send(pb.Message{From: 1, To: 1, Type: msgHup})
  212. tt.send(pb.Message{From: 3, To: 3, Type: msgHup})
  213. // heal the partition
  214. tt.recover()
  215. data := []byte("force follower")
  216. // send a proposal to 2 to flush out a msgApp to 0
  217. tt.send(pb.Message{From: 3, To: 3, Type: msgProp, Entries: []pb.Entry{{Data: data}}})
  218. a := tt.peers[1].(*raft)
  219. if g := a.state; g != StateFollower {
  220. t.Errorf("state = %s, want %s", g, StateFollower)
  221. }
  222. if g := a.Term; g != 1 {
  223. t.Errorf("term = %d, want %d", g, 1)
  224. }
  225. wantLog := ltoa(&raftLog{ents: []pb.Entry{{}, {Data: nil, Term: 1, Index: 1}, {Term: 1, Index: 2, Data: data}}, committed: 2})
  226. for i, p := range tt.peers {
  227. if sm, ok := p.(*raft); ok {
  228. l := ltoa(sm.raftLog)
  229. if g := diffu(wantLog, l); g != "" {
  230. t.Errorf("#%d: diff:\n%s", i, g)
  231. }
  232. } else {
  233. t.Logf("#%d: empty log", i)
  234. }
  235. }
  236. }
  237. func TestSingleNodeCandidate(t *testing.T) {
  238. tt := newNetwork(nil)
  239. tt.send(pb.Message{From: 1, To: 1, Type: msgHup})
  240. sm := tt.peers[1].(*raft)
  241. if sm.state != StateLeader {
  242. t.Errorf("state = %d, want %d", sm.state, StateLeader)
  243. }
  244. }
  245. func TestOldMessages(t *testing.T) {
  246. tt := newNetwork(nil, nil, nil)
  247. // make 0 leader @ term 3
  248. tt.send(pb.Message{From: 1, To: 1, Type: msgHup})
  249. tt.send(pb.Message{From: 2, To: 2, Type: msgHup})
  250. tt.send(pb.Message{From: 1, To: 1, Type: msgHup})
  251. // pretend we're an old leader trying to make progress
  252. tt.send(pb.Message{From: 1, To: 1, Type: msgApp, Term: 1, Entries: []pb.Entry{{Term: 1}}})
  253. l := &raftLog{
  254. ents: []pb.Entry{
  255. {}, {Data: nil, Term: 1, Index: 1},
  256. {Data: nil, Term: 2, Index: 2}, {Data: nil, Term: 3, Index: 3},
  257. },
  258. committed: 3,
  259. }
  260. base := ltoa(l)
  261. for i, p := range tt.peers {
  262. if sm, ok := p.(*raft); ok {
  263. l := ltoa(sm.raftLog)
  264. if g := diffu(base, l); g != "" {
  265. t.Errorf("#%d: diff:\n%s", i, g)
  266. }
  267. } else {
  268. t.Logf("#%d: empty log", i)
  269. }
  270. }
  271. }
  272. // TestOldMessagesReply - optimization - reply with new term.
  273. func TestProposal(t *testing.T) {
  274. tests := []struct {
  275. *network
  276. success bool
  277. }{
  278. {newNetwork(nil, nil, nil), true},
  279. {newNetwork(nil, nil, nopStepper), true},
  280. {newNetwork(nil, nopStepper, nopStepper), false},
  281. {newNetwork(nil, nopStepper, nopStepper, nil), false},
  282. {newNetwork(nil, nopStepper, nopStepper, nil, nil), true},
  283. }
  284. for i, tt := range tests {
  285. send := func(m pb.Message) {
  286. defer func() {
  287. // only recover is we expect it to panic so
  288. // panics we don't expect go up.
  289. if !tt.success {
  290. e := recover()
  291. if e != nil {
  292. t.Logf("#%d: err: %s", i, e)
  293. }
  294. }
  295. }()
  296. tt.send(m)
  297. }
  298. data := []byte("somedata")
  299. // promote 0 the leader
  300. send(pb.Message{From: 1, To: 1, Type: msgHup})
  301. send(pb.Message{From: 1, To: 1, Type: msgProp, Entries: []pb.Entry{{Data: data}}})
  302. wantLog := newLog()
  303. if tt.success {
  304. wantLog = &raftLog{ents: []pb.Entry{{}, {Data: nil, Term: 1, Index: 1}, {Term: 1, Index: 2, Data: data}}, committed: 2}
  305. }
  306. base := ltoa(wantLog)
  307. for i, p := range tt.peers {
  308. if sm, ok := p.(*raft); ok {
  309. l := ltoa(sm.raftLog)
  310. if g := diffu(base, l); g != "" {
  311. t.Errorf("#%d: diff:\n%s", i, g)
  312. }
  313. } else {
  314. t.Logf("#%d: empty log", i)
  315. }
  316. }
  317. sm := tt.network.peers[1].(*raft)
  318. if g := sm.Term; g != 1 {
  319. t.Errorf("#%d: term = %d, want %d", i, g, 1)
  320. }
  321. }
  322. }
  323. func TestProposalByProxy(t *testing.T) {
  324. data := []byte("somedata")
  325. tests := []*network{
  326. newNetwork(nil, nil, nil),
  327. newNetwork(nil, nil, nopStepper),
  328. }
  329. for i, tt := range tests {
  330. // promote 0 the leader
  331. tt.send(pb.Message{From: 1, To: 1, Type: msgHup})
  332. // propose via follower
  333. tt.send(pb.Message{From: 2, To: 2, Type: msgProp, Entries: []pb.Entry{{Data: []byte("somedata")}}})
  334. wantLog := &raftLog{ents: []pb.Entry{{}, {Data: nil, Term: 1, Index: 1}, {Term: 1, Data: data, Index: 2}}, committed: 2}
  335. base := ltoa(wantLog)
  336. for i, p := range tt.peers {
  337. if sm, ok := p.(*raft); ok {
  338. l := ltoa(sm.raftLog)
  339. if g := diffu(base, l); g != "" {
  340. t.Errorf("#%d: diff:\n%s", i, g)
  341. }
  342. } else {
  343. t.Logf("#%d: empty log", i)
  344. }
  345. }
  346. sm := tt.peers[1].(*raft)
  347. if g := sm.Term; g != 1 {
  348. t.Errorf("#%d: term = %d, want %d", i, g, 1)
  349. }
  350. }
  351. }
  352. func TestCompact(t *testing.T) {
  353. tests := []struct {
  354. compacti int64
  355. nodes []int64
  356. snapd []byte
  357. wpanic bool
  358. }{
  359. {1, []int64{1, 2, 3}, []byte("some data"), false},
  360. {2, []int64{1, 2, 3}, []byte("some data"), false},
  361. {4, []int64{1, 2, 3}, []byte("some data"), true}, // compact out of range
  362. }
  363. for i, tt := range tests {
  364. func() {
  365. defer func() {
  366. if r := recover(); r != nil {
  367. if tt.wpanic != true {
  368. t.Errorf("%d: panic = %v, want %v", i, false, true)
  369. }
  370. }
  371. }()
  372. sm := &raft{
  373. state: StateLeader,
  374. raftLog: &raftLog{
  375. committed: 2,
  376. applied: 2,
  377. ents: []pb.Entry{{}, {Term: 1}, {Term: 1}, {Term: 1}},
  378. },
  379. }
  380. sm.compact(tt.compacti, tt.nodes, tt.snapd)
  381. if sm.raftLog.offset != tt.compacti {
  382. t.Errorf("%d: log.offset = %d, want %d", i, sm.raftLog.offset, tt.compacti)
  383. }
  384. if !reflect.DeepEqual(sm.raftLog.snapshot.Nodes, tt.nodes) {
  385. t.Errorf("%d: snap.nodes = %v, want %v", i, sm.raftLog.snapshot.Nodes, tt.nodes)
  386. }
  387. if !reflect.DeepEqual(sm.raftLog.snapshot.Data, tt.snapd) {
  388. t.Errorf("%d: snap.data = %v, want %v", i, sm.raftLog.snapshot.Data, tt.snapd)
  389. }
  390. }()
  391. }
  392. }
  393. func TestCommit(t *testing.T) {
  394. tests := []struct {
  395. matches []int64
  396. logs []pb.Entry
  397. smTerm int64
  398. w int64
  399. }{
  400. // single
  401. {[]int64{1}, []pb.Entry{{}, {Term: 1}}, 1, 1},
  402. {[]int64{1}, []pb.Entry{{}, {Term: 1}}, 2, 0},
  403. {[]int64{2}, []pb.Entry{{}, {Term: 1}, {Term: 2}}, 2, 2},
  404. {[]int64{1}, []pb.Entry{{}, {Term: 2}}, 2, 1},
  405. // odd
  406. {[]int64{2, 1, 1}, []pb.Entry{{}, {Term: 1}, {Term: 2}}, 1, 1},
  407. {[]int64{2, 1, 1}, []pb.Entry{{}, {Term: 1}, {Term: 1}}, 2, 0},
  408. {[]int64{2, 1, 2}, []pb.Entry{{}, {Term: 1}, {Term: 2}}, 2, 2},
  409. {[]int64{2, 1, 2}, []pb.Entry{{}, {Term: 1}, {Term: 1}}, 2, 0},
  410. // even
  411. {[]int64{2, 1, 1, 1}, []pb.Entry{{}, {Term: 1}, {Term: 2}}, 1, 1},
  412. {[]int64{2, 1, 1, 1}, []pb.Entry{{}, {Term: 1}, {Term: 1}}, 2, 0},
  413. {[]int64{2, 1, 1, 2}, []pb.Entry{{}, {Term: 1}, {Term: 2}}, 1, 1},
  414. {[]int64{2, 1, 1, 2}, []pb.Entry{{}, {Term: 1}, {Term: 1}}, 2, 0},
  415. {[]int64{2, 1, 2, 2}, []pb.Entry{{}, {Term: 1}, {Term: 2}}, 2, 2},
  416. {[]int64{2, 1, 2, 2}, []pb.Entry{{}, {Term: 1}, {Term: 1}}, 2, 0},
  417. }
  418. for i, tt := range tests {
  419. prs := make(map[int64]*progress)
  420. for j := 0; j < len(tt.matches); j++ {
  421. prs[int64(j)] = &progress{tt.matches[j], tt.matches[j] + 1}
  422. }
  423. sm := &raft{raftLog: &raftLog{ents: tt.logs}, prs: prs, HardState: pb.HardState{Term: tt.smTerm}}
  424. sm.maybeCommit()
  425. if g := sm.raftLog.committed; g != tt.w {
  426. t.Errorf("#%d: committed = %d, want %d", i, g, tt.w)
  427. }
  428. }
  429. }
  430. // ensure that the Step function ignores the message from old term and does not pass it to the
  431. // acutal stepX function.
  432. func TestStepIgnoreOldTermMsg(t *testing.T) {
  433. called := false
  434. fakeStep := func(r *raft, m pb.Message) {
  435. called = true
  436. }
  437. sm := newRaft(1, []int64{1}, 0, 0)
  438. sm.step = fakeStep
  439. sm.Term = 2
  440. sm.Step(pb.Message{Type: msgApp, Term: sm.Term - 1})
  441. if called == true {
  442. t.Errorf("stepFunc called = %v , want %v", called, false)
  443. }
  444. }
  445. // TestHandleMsgApp ensures:
  446. // 1. Reply false if log doesn’t contain an entry at prevLogIndex whose term matches prevLogTerm.
  447. // 2. If an existing entry conflicts with a new one (same index but different terms),
  448. // delete the existing entry and all that follow it; append any new entries not already in the log.
  449. // 3. If leaderCommit > commitIndex, set commitIndex = min(leaderCommit, index of last new entry).
  450. func TestHandleMsgApp(t *testing.T) {
  451. tests := []struct {
  452. m pb.Message
  453. wIndex int64
  454. wCommit int64
  455. wReject bool
  456. }{
  457. // Ensure 1
  458. {pb.Message{Type: msgApp, Term: 2, LogTerm: 3, Index: 2, Commit: 3}, 2, 0, true}, // previous log mismatch
  459. {pb.Message{Type: msgApp, Term: 2, LogTerm: 3, Index: 3, Commit: 3}, 2, 0, true}, // previous log non-exist
  460. // Ensure 2
  461. {pb.Message{Type: msgApp, Term: 2, LogTerm: 1, Index: 1, Commit: 1}, 2, 1, false},
  462. {pb.Message{Type: msgApp, Term: 2, LogTerm: 0, Index: 0, Commit: 1, Entries: []pb.Entry{{Term: 2}}}, 1, 1, false},
  463. {pb.Message{Type: msgApp, Term: 2, LogTerm: 2, Index: 2, Commit: 3, Entries: []pb.Entry{{Term: 2}, {Term: 2}}}, 4, 3, false},
  464. {pb.Message{Type: msgApp, Term: 2, LogTerm: 2, Index: 2, Commit: 4, Entries: []pb.Entry{{Term: 2}}}, 3, 3, false},
  465. {pb.Message{Type: msgApp, Term: 2, LogTerm: 1, Index: 1, Commit: 4, Entries: []pb.Entry{{Term: 2}}}, 2, 2, false},
  466. // Ensure 3
  467. {pb.Message{Type: msgApp, Term: 1, LogTerm: 1, Index: 1, Commit: 3}, 2, 1, false}, // match entry 1, commit upto last new entry 1
  468. {pb.Message{Type: msgApp, Term: 1, LogTerm: 1, Index: 1, Commit: 3, Entries: []pb.Entry{{Term: 2}}}, 2, 2, false}, // match entry 1, commit upto last new entry 2
  469. {pb.Message{Type: msgApp, Term: 2, LogTerm: 2, Index: 2, Commit: 3}, 2, 2, false}, // match entry 2, commit upto last new entry 2
  470. {pb.Message{Type: msgApp, Term: 2, LogTerm: 2, Index: 2, Commit: 4}, 2, 2, false}, // commit upto log.last()
  471. }
  472. for i, tt := range tests {
  473. sm := &raft{
  474. state: StateFollower,
  475. HardState: pb.HardState{Term: 2},
  476. raftLog: &raftLog{committed: 0, ents: []pb.Entry{{}, {Term: 1}, {Term: 2}}},
  477. }
  478. sm.handleAppendEntries(tt.m)
  479. if sm.raftLog.lastIndex() != tt.wIndex {
  480. t.Errorf("#%d: lastIndex = %d, want %d", i, sm.raftLog.lastIndex(), tt.wIndex)
  481. }
  482. if sm.raftLog.committed != tt.wCommit {
  483. t.Errorf("#%d: committed = %d, want %d", i, sm.raftLog.committed, tt.wCommit)
  484. }
  485. m := sm.ReadMessages()
  486. if len(m) != 1 {
  487. t.Fatalf("#%d: msg = nil, want 1", i)
  488. }
  489. if m[0].Reject != tt.wReject {
  490. t.Errorf("#%d: reject = %v, want %v", i, m[0].Reject, tt.wReject)
  491. }
  492. }
  493. }
  494. func TestRecvMsgVote(t *testing.T) {
  495. tests := []struct {
  496. state StateType
  497. i, term int64
  498. voteFor int64
  499. wreject bool
  500. }{
  501. {StateFollower, 0, 0, None, true},
  502. {StateFollower, 0, 1, None, true},
  503. {StateFollower, 0, 2, None, true},
  504. {StateFollower, 0, 3, None, false},
  505. {StateFollower, 1, 0, None, true},
  506. {StateFollower, 1, 1, None, true},
  507. {StateFollower, 1, 2, None, true},
  508. {StateFollower, 1, 3, None, false},
  509. {StateFollower, 2, 0, None, true},
  510. {StateFollower, 2, 1, None, true},
  511. {StateFollower, 2, 2, None, false},
  512. {StateFollower, 2, 3, None, false},
  513. {StateFollower, 3, 0, None, true},
  514. {StateFollower, 3, 1, None, true},
  515. {StateFollower, 3, 2, None, false},
  516. {StateFollower, 3, 3, None, false},
  517. {StateFollower, 3, 2, 2, false},
  518. {StateFollower, 3, 2, 1, true},
  519. {StateLeader, 3, 3, 1, true},
  520. {StateCandidate, 3, 3, 1, true},
  521. }
  522. for i, tt := range tests {
  523. sm := newRaft(1, []int64{1}, 0, 0)
  524. sm.state = tt.state
  525. switch tt.state {
  526. case StateFollower:
  527. sm.step = stepFollower
  528. case StateCandidate:
  529. sm.step = stepCandidate
  530. case StateLeader:
  531. sm.step = stepLeader
  532. }
  533. sm.HardState = pb.HardState{Vote: tt.voteFor}
  534. sm.raftLog = &raftLog{ents: []pb.Entry{{}, {Term: 2}, {Term: 2}}}
  535. sm.Step(pb.Message{Type: msgVote, From: 2, Index: tt.i, LogTerm: tt.term})
  536. msgs := sm.ReadMessages()
  537. if g := len(msgs); g != 1 {
  538. t.Fatalf("#%d: len(msgs) = %d, want 1", i, g)
  539. continue
  540. }
  541. if g := msgs[0].Reject; g != tt.wreject {
  542. t.Errorf("#%d, m.Reject = %d, want %v", i, g, tt.wreject)
  543. }
  544. }
  545. }
  546. func TestStateTransition(t *testing.T) {
  547. tests := []struct {
  548. from StateType
  549. to StateType
  550. wallow bool
  551. wterm int64
  552. wlead int64
  553. }{
  554. {StateFollower, StateFollower, true, 1, None},
  555. {StateFollower, StateCandidate, true, 1, None},
  556. {StateFollower, StateLeader, false, -1, None},
  557. {StateCandidate, StateFollower, true, 0, None},
  558. {StateCandidate, StateCandidate, true, 1, None},
  559. {StateCandidate, StateLeader, true, 0, 1},
  560. {StateLeader, StateFollower, true, 1, None},
  561. {StateLeader, StateCandidate, false, 1, None},
  562. {StateLeader, StateLeader, true, 0, 1},
  563. }
  564. for i, tt := range tests {
  565. func() {
  566. defer func() {
  567. if r := recover(); r != nil {
  568. if tt.wallow == true {
  569. t.Errorf("%d: allow = %v, want %v", i, false, true)
  570. }
  571. }
  572. }()
  573. sm := newRaft(1, []int64{1}, 0, 0)
  574. sm.state = tt.from
  575. switch tt.to {
  576. case StateFollower:
  577. sm.becomeFollower(tt.wterm, tt.wlead)
  578. case StateCandidate:
  579. sm.becomeCandidate()
  580. case StateLeader:
  581. sm.becomeLeader()
  582. }
  583. if sm.Term != tt.wterm {
  584. t.Errorf("%d: term = %d, want %d", i, sm.Term, tt.wterm)
  585. }
  586. if sm.lead != tt.wlead {
  587. t.Errorf("%d: lead = %d, want %d", i, sm.lead, tt.wlead)
  588. }
  589. }()
  590. }
  591. }
  592. func TestAllServerStepdown(t *testing.T) {
  593. tests := []struct {
  594. state StateType
  595. wstate StateType
  596. wterm int64
  597. windex int64
  598. }{
  599. {StateFollower, StateFollower, 3, 1},
  600. {StateCandidate, StateFollower, 3, 1},
  601. {StateLeader, StateFollower, 3, 2},
  602. }
  603. tmsgTypes := [...]int64{msgVote, msgApp}
  604. tterm := int64(3)
  605. for i, tt := range tests {
  606. sm := newRaft(1, []int64{1, 2, 3}, 0, 0)
  607. switch tt.state {
  608. case StateFollower:
  609. sm.becomeFollower(1, None)
  610. case StateCandidate:
  611. sm.becomeCandidate()
  612. case StateLeader:
  613. sm.becomeCandidate()
  614. sm.becomeLeader()
  615. }
  616. for j, msgType := range tmsgTypes {
  617. sm.Step(pb.Message{From: 2, Type: msgType, Term: tterm, LogTerm: tterm})
  618. if sm.state != tt.wstate {
  619. t.Errorf("#%d.%d state = %v , want %v", i, j, sm.state, tt.wstate)
  620. }
  621. if sm.Term != tt.wterm {
  622. t.Errorf("#%d.%d term = %v , want %v", i, j, sm.Term, tt.wterm)
  623. }
  624. if int64(len(sm.raftLog.ents)) != tt.windex {
  625. t.Errorf("#%d.%d index = %v , want %v", i, j, len(sm.raftLog.ents), tt.windex)
  626. }
  627. wlead := int64(2)
  628. if msgType == msgVote {
  629. wlead = None
  630. }
  631. if sm.lead != wlead {
  632. t.Errorf("#%d, sm.lead = %d, want %d", i, sm.lead, None)
  633. }
  634. }
  635. }
  636. }
  637. func TestLeaderAppResp(t *testing.T) {
  638. tests := []struct {
  639. index int64
  640. reject bool
  641. wmsgNum int
  642. windex int64
  643. wcommitted int64
  644. }{
  645. {3, true, 0, 0, 0}, // stale resp; no replies
  646. {2, true, 1, 1, 0}, // denied resp; leader does not commit; decrese next and send probing msg
  647. {2, false, 2, 2, 2}, // accept resp; leader commits; broadcast with commit index
  648. }
  649. for i, tt := range tests {
  650. // sm term is 1 after it becomes the leader.
  651. // thus the last log term must be 1 to be committed.
  652. sm := newRaft(1, []int64{1, 2, 3}, 0, 0)
  653. sm.raftLog = &raftLog{ents: []pb.Entry{{}, {Term: 0}, {Term: 1}}}
  654. sm.becomeCandidate()
  655. sm.becomeLeader()
  656. sm.ReadMessages()
  657. sm.Step(pb.Message{From: 2, Type: msgAppResp, Index: tt.index, Term: sm.Term, Reject: tt.reject})
  658. msgs := sm.ReadMessages()
  659. if len(msgs) != tt.wmsgNum {
  660. t.Errorf("#%d msgNum = %d, want %d", i, len(msgs), tt.wmsgNum)
  661. }
  662. for j, msg := range msgs {
  663. if msg.Index != tt.windex {
  664. t.Errorf("#%d.%d index = %d, want %d", i, j, msg.Index, tt.windex)
  665. }
  666. if msg.Commit != tt.wcommitted {
  667. t.Errorf("#%d.%d commit = %d, want %d", i, j, msg.Commit, tt.wcommitted)
  668. }
  669. }
  670. }
  671. }
  672. // When the leader receives a heartbeat tick, it should
  673. // send a msgApp with m.Index = 0, m.LogTerm=0 and empty entries.
  674. func TestBcastBeat(t *testing.T) {
  675. offset := int64(1000)
  676. // make a state machine with log.offset = 1000
  677. s := pb.Snapshot{
  678. Index: offset,
  679. Term: 1,
  680. Nodes: []int64{1, 2, 3},
  681. }
  682. sm := newRaft(1, []int64{1, 2, 3}, 0, 0)
  683. sm.Term = 1
  684. sm.restore(s)
  685. sm.becomeCandidate()
  686. sm.becomeLeader()
  687. for i := 0; i < 10; i++ {
  688. sm.appendEntry(pb.Entry{})
  689. }
  690. sm.Step(pb.Message{Type: msgBeat})
  691. msgs := sm.ReadMessages()
  692. if len(msgs) != 2 {
  693. t.Fatalf("len(msgs) = %v, want 1", len(msgs))
  694. }
  695. tomap := map[int64]bool{2: true, 3: true}
  696. for i, m := range msgs {
  697. if m.Type != msgApp {
  698. t.Fatalf("#%d: type = %v, want = %v", i, m.Type, msgApp)
  699. }
  700. if m.Index != 0 {
  701. t.Fatalf("#%d: prevIndex = %d, want %d", i, m.Index, 0)
  702. }
  703. if m.LogTerm != 0 {
  704. t.Fatalf("#%d: prevTerm = %d, want %d", i, m.LogTerm, 0)
  705. }
  706. if !tomap[m.To] {
  707. t.Fatalf("#%d: unexpected to %d", i, m.To)
  708. } else {
  709. delete(tomap, m.To)
  710. }
  711. if len(m.Entries) != 0 {
  712. t.Fatalf("#%d: len(entries) = %d, want 0", i, len(m.Entries))
  713. }
  714. }
  715. }
  716. // tests the output of the statemachine when receiving msgBeat
  717. func TestRecvMsgBeat(t *testing.T) {
  718. tests := []struct {
  719. state StateType
  720. wMsg int
  721. }{
  722. {StateLeader, 2},
  723. // candidate and follower should ignore msgBeat
  724. {StateCandidate, 0},
  725. {StateFollower, 0},
  726. }
  727. for i, tt := range tests {
  728. sm := newRaft(1, []int64{1, 2, 3}, 0, 0)
  729. sm.raftLog = &raftLog{ents: []pb.Entry{{}, {Term: 0}, {Term: 1}}}
  730. sm.Term = 1
  731. sm.state = tt.state
  732. switch tt.state {
  733. case StateFollower:
  734. sm.step = stepFollower
  735. case StateCandidate:
  736. sm.step = stepCandidate
  737. case StateLeader:
  738. sm.step = stepLeader
  739. }
  740. sm.Step(pb.Message{From: 1, To: 1, Type: msgBeat})
  741. msgs := sm.ReadMessages()
  742. if len(msgs) != tt.wMsg {
  743. t.Errorf("%d: len(msgs) = %d, want %d", i, len(msgs), tt.wMsg)
  744. }
  745. for _, m := range msgs {
  746. if m.Type != msgApp {
  747. t.Errorf("%d: msg.type = %v, want %v", i, m.Type, msgApp)
  748. }
  749. }
  750. }
  751. }
  752. func TestRestore(t *testing.T) {
  753. s := pb.Snapshot{
  754. Index: defaultCompactThreshold + 1,
  755. Term: defaultCompactThreshold + 1,
  756. Nodes: []int64{1, 2, 3},
  757. }
  758. sm := newRaft(1, []int64{1, 2}, 0, 0)
  759. if ok := sm.restore(s); !ok {
  760. t.Fatal("restore fail, want succeed")
  761. }
  762. if sm.raftLog.lastIndex() != s.Index {
  763. t.Errorf("log.lastIndex = %d, want %d", sm.raftLog.lastIndex(), s.Index)
  764. }
  765. if sm.raftLog.term(s.Index) != s.Term {
  766. t.Errorf("log.lastTerm = %d, want %d", sm.raftLog.term(s.Index), s.Term)
  767. }
  768. sg := int64Slice(sm.nodes())
  769. sw := int64Slice(s.Nodes)
  770. sort.Sort(sg)
  771. sort.Sort(sw)
  772. if !reflect.DeepEqual(sg, sw) {
  773. t.Errorf("sm.Nodes = %+v, want %+v", sg, sw)
  774. }
  775. if !reflect.DeepEqual(sm.raftLog.snapshot, s) {
  776. t.Errorf("snapshot = %+v, want %+v", sm.raftLog.snapshot, s)
  777. }
  778. if ok := sm.restore(s); ok {
  779. t.Fatal("restore succeed, want fail")
  780. }
  781. }
  782. func TestProvideSnap(t *testing.T) {
  783. s := pb.Snapshot{
  784. Index: defaultCompactThreshold + 1,
  785. Term: defaultCompactThreshold + 1,
  786. Nodes: []int64{1, 2},
  787. }
  788. sm := newRaft(1, []int64{1}, 0, 0)
  789. // restore the statemachin from a snapshot
  790. // so it has a compacted log and a snapshot
  791. sm.restore(s)
  792. sm.becomeCandidate()
  793. sm.becomeLeader()
  794. // force set the next of node 1, so that
  795. // node 1 needs a snapshot
  796. sm.prs[2].next = sm.raftLog.offset
  797. sm.Step(pb.Message{From: 2, To: 1, Type: msgAppResp, Index: sm.prs[2].next - 1, Reject: true})
  798. msgs := sm.ReadMessages()
  799. if len(msgs) != 1 {
  800. t.Fatalf("len(msgs) = %d, want 1", len(msgs))
  801. }
  802. m := msgs[0]
  803. if m.Type != msgSnap {
  804. t.Errorf("m.Type = %v, want %v", m.Type, msgSnap)
  805. }
  806. }
  807. func TestRestoreFromSnapMsg(t *testing.T) {
  808. s := pb.Snapshot{
  809. Index: defaultCompactThreshold + 1,
  810. Term: defaultCompactThreshold + 1,
  811. Nodes: []int64{1, 2},
  812. }
  813. m := pb.Message{Type: msgSnap, From: 1, Term: 2, Snapshot: s}
  814. sm := newRaft(2, []int64{1, 2}, 0, 0)
  815. sm.Step(m)
  816. if !reflect.DeepEqual(sm.raftLog.snapshot, s) {
  817. t.Errorf("snapshot = %+v, want %+v", sm.raftLog.snapshot, s)
  818. }
  819. }
  820. func TestSlowNodeRestore(t *testing.T) {
  821. nt := newNetwork(nil, nil, nil)
  822. nt.send(pb.Message{From: 1, To: 1, Type: msgHup})
  823. nt.isolate(3)
  824. for j := 0; j < defaultCompactThreshold+1; j++ {
  825. nt.send(pb.Message{From: 1, To: 1, Type: msgProp, Entries: []pb.Entry{{}}})
  826. }
  827. lead := nt.peers[1].(*raft)
  828. nextEnts(lead)
  829. lead.compact(lead.raftLog.applied, lead.nodes(), nil)
  830. nt.recover()
  831. // trigger a snapshot
  832. nt.send(pb.Message{From: 1, To: 1, Type: msgProp, Entries: []pb.Entry{{}}})
  833. follower := nt.peers[3].(*raft)
  834. if !reflect.DeepEqual(follower.raftLog.snapshot, lead.raftLog.snapshot) {
  835. t.Errorf("follower.snap = %+v, want %+v", follower.raftLog.snapshot, lead.raftLog.snapshot)
  836. }
  837. // trigger a commit
  838. nt.send(pb.Message{From: 1, To: 1, Type: msgProp, Entries: []pb.Entry{{}}})
  839. if follower.raftLog.committed != lead.raftLog.committed {
  840. t.Errorf("follower.comitted = %d, want %d", follower.raftLog.committed, lead.raftLog.committed)
  841. }
  842. }
  843. // TestStepConfig tests that when raft step msgProp in EntryConfChange type,
  844. // it appends the entry to log and sets pendingConf to be true.
  845. func TestStepConfig(t *testing.T) {
  846. // a raft that cannot make progress
  847. r := newRaft(1, []int64{1, 2}, 0, 0)
  848. r.becomeCandidate()
  849. r.becomeLeader()
  850. index := r.raftLog.lastIndex()
  851. r.Step(pb.Message{From: 1, To: 1, Type: msgProp, Entries: []pb.Entry{{Type: pb.EntryConfChange}}})
  852. if g := r.raftLog.lastIndex(); g != index+1 {
  853. t.Errorf("index = %d, want %d", g, index+1)
  854. }
  855. if r.pendingConf != true {
  856. t.Errorf("pendingConf = %v, want true", r.pendingConf)
  857. }
  858. }
  859. // TestStepIgnoreConfig tests that if raft step the second msgProp in
  860. // EntryConfChange type when the first one is uncommitted, the node will deny
  861. // the proposal and keep its original state.
  862. func TestStepIgnoreConfig(t *testing.T) {
  863. // a raft that cannot make progress
  864. r := newRaft(1, []int64{1, 2}, 0, 0)
  865. r.becomeCandidate()
  866. r.becomeLeader()
  867. r.Step(pb.Message{From: 1, To: 1, Type: msgProp, Entries: []pb.Entry{{Type: pb.EntryConfChange}}})
  868. index := r.raftLog.lastIndex()
  869. pendingConf := r.pendingConf
  870. r.Step(pb.Message{From: 1, To: 1, Type: msgProp, Entries: []pb.Entry{{Type: pb.EntryConfChange}}})
  871. if g := r.raftLog.lastIndex(); g != index {
  872. t.Errorf("index = %d, want %d", g, index)
  873. }
  874. if r.pendingConf != pendingConf {
  875. t.Errorf("pendingConf = %v, want %v", r.pendingConf, pendingConf)
  876. }
  877. }
  878. // TestRecoverPendingConfig tests that new leader recovers its pendingConf flag
  879. // based on uncommitted entries.
  880. func TestRecoverPendingConfig(t *testing.T) {
  881. tests := []struct {
  882. entType pb.EntryType
  883. wpending bool
  884. }{
  885. {pb.EntryNormal, false},
  886. {pb.EntryConfChange, true},
  887. }
  888. for i, tt := range tests {
  889. r := newRaft(1, []int64{1, 2}, 0, 0)
  890. r.appendEntry(pb.Entry{Type: tt.entType})
  891. r.becomeCandidate()
  892. r.becomeLeader()
  893. if r.pendingConf != tt.wpending {
  894. t.Errorf("#%d: pendingConf = %v, want %v", i, r.pendingConf, tt.wpending)
  895. }
  896. }
  897. }
  898. // TestRecoverDoublePendingConfig tests that new leader will panic if
  899. // there exist two uncommitted config entries.
  900. func TestRecoverDoublePendingConfig(t *testing.T) {
  901. func() {
  902. defer func() {
  903. if err := recover(); err == nil {
  904. t.Errorf("expect panic, but nothing happens")
  905. }
  906. }()
  907. r := newRaft(1, []int64{1, 2}, 0, 0)
  908. r.appendEntry(pb.Entry{Type: pb.EntryConfChange})
  909. r.appendEntry(pb.Entry{Type: pb.EntryConfChange})
  910. r.becomeCandidate()
  911. r.becomeLeader()
  912. }()
  913. }
  914. // TestAddNode tests that addNode could update pendingConf and nodes correctly.
  915. func TestAddNode(t *testing.T) {
  916. r := newRaft(1, []int64{1}, 0, 0)
  917. r.pendingConf = true
  918. r.addNode(2)
  919. if r.pendingConf != false {
  920. t.Errorf("pendingConf = %v, want false", r.pendingConf)
  921. }
  922. nodes := r.nodes()
  923. sort.Sort(int64Slice(nodes))
  924. wnodes := []int64{1, 2}
  925. if !reflect.DeepEqual(nodes, wnodes) {
  926. t.Errorf("nodes = %v, want %v", nodes, wnodes)
  927. }
  928. }
  929. // TestRemoveNode tests that removeNode could update pendingConf, nodes and
  930. // and removed list correctly.
  931. func TestRemoveNode(t *testing.T) {
  932. r := newRaft(1, []int64{1, 2}, 0, 0)
  933. r.pendingConf = true
  934. r.removeNode(2)
  935. if r.pendingConf != false {
  936. t.Errorf("pendingConf = %v, want false", r.pendingConf)
  937. }
  938. w := []int64{1}
  939. if g := r.nodes(); !reflect.DeepEqual(g, w) {
  940. t.Errorf("nodes = %v, want %v", g, w)
  941. }
  942. wremoved := map[int64]bool{2: true}
  943. if !reflect.DeepEqual(r.removed, wremoved) {
  944. t.Errorf("rmNodes = %v, want %v", r.removed, wremoved)
  945. }
  946. }
  947. // TestRecvMsgDenied tests that state machine sets the removed list when
  948. // handling msgDenied, and does not pass it to the actual stepX function.
  949. func TestRecvMsgDenied(t *testing.T) {
  950. called := false
  951. fakeStep := func(r *raft, m pb.Message) {
  952. called = true
  953. }
  954. r := newRaft(1, []int64{1, 2}, 0, 0)
  955. r.step = fakeStep
  956. r.Step(pb.Message{From: 2, Type: msgDenied})
  957. if called != false {
  958. t.Errorf("stepFunc called = %v , want %v", called, false)
  959. }
  960. wremoved := map[int64]bool{1: true}
  961. if !reflect.DeepEqual(r.removed, wremoved) {
  962. t.Errorf("rmNodes = %v, want %v", r.removed, wremoved)
  963. }
  964. }
  965. // TestRecvMsgFromRemovedNode tests that state machine sends correct
  966. // messages out when handling message from removed node, and does not
  967. // pass it to the actual stepX function.
  968. func TestRecvMsgFromRemovedNode(t *testing.T) {
  969. tests := []struct {
  970. from int64
  971. wmsgNum int
  972. }{
  973. {1, 0},
  974. {2, 1},
  975. }
  976. for i, tt := range tests {
  977. called := false
  978. fakeStep := func(r *raft, m pb.Message) {
  979. called = true
  980. }
  981. r := newRaft(1, []int64{1}, 0, 0)
  982. r.step = fakeStep
  983. r.removeNode(tt.from)
  984. r.Step(pb.Message{From: tt.from, Type: msgVote})
  985. if called != false {
  986. t.Errorf("#%d: stepFunc called = %v , want %v", i, called, false)
  987. }
  988. if len(r.msgs) != tt.wmsgNum {
  989. t.Errorf("#%d: len(msgs) = %d, want %d", i, len(r.msgs), tt.wmsgNum)
  990. }
  991. for j, msg := range r.msgs {
  992. if msg.Type != msgDenied {
  993. t.Errorf("#%d.%d: msgType = %d, want %d", i, j, msg.Type, msgDenied)
  994. }
  995. }
  996. }
  997. }
  998. func TestPromotable(t *testing.T) {
  999. id := int64(1)
  1000. tests := []struct {
  1001. peers []int64
  1002. wp bool
  1003. }{
  1004. {[]int64{1}, true},
  1005. {[]int64{1, 2, 3}, true},
  1006. {[]int64{}, false},
  1007. {[]int64{2, 3}, false},
  1008. }
  1009. for i, tt := range tests {
  1010. r := &raft{id: id, prs: make(map[int64]*progress)}
  1011. for _, id := range tt.peers {
  1012. r.prs[id] = &progress{}
  1013. }
  1014. if g := r.promotable(); g != tt.wp {
  1015. t.Errorf("#%d: promotable = %v, want %v", i, g, tt.wp)
  1016. }
  1017. }
  1018. }
  1019. func ents(terms ...int64) *raft {
  1020. ents := []pb.Entry{{}}
  1021. for _, term := range terms {
  1022. ents = append(ents, pb.Entry{Term: term})
  1023. }
  1024. sm := &raft{raftLog: &raftLog{ents: ents}}
  1025. sm.reset(0)
  1026. return sm
  1027. }
  1028. type network struct {
  1029. peers map[int64]Interface
  1030. dropm map[connem]float64
  1031. ignorem map[int64]bool
  1032. }
  1033. // newNetwork initializes a network from peers.
  1034. // A nil node will be replaced with a new *stateMachine.
  1035. // A *stateMachine will get its k, id.
  1036. // When using stateMachine, the address list is always [0, n).
  1037. func newNetwork(peers ...Interface) *network {
  1038. size := len(peers)
  1039. peerAddrs := make([]int64, size)
  1040. for i := 0; i < size; i++ {
  1041. peerAddrs[i] = 1 + int64(i)
  1042. }
  1043. npeers := make(map[int64]Interface, size)
  1044. for i, p := range peers {
  1045. id := peerAddrs[i]
  1046. switch v := p.(type) {
  1047. case nil:
  1048. sm := newRaft(id, peerAddrs, 0, 0)
  1049. npeers[id] = sm
  1050. case *raft:
  1051. v.id = id
  1052. v.prs = make(map[int64]*progress)
  1053. for i := 0; i < size; i++ {
  1054. v.prs[peerAddrs[i]] = &progress{}
  1055. }
  1056. v.reset(0)
  1057. npeers[id] = v
  1058. case *blackHole:
  1059. npeers[id] = v
  1060. default:
  1061. panic(fmt.Sprintf("unexpected state machine type: %T", p))
  1062. }
  1063. }
  1064. return &network{
  1065. peers: npeers,
  1066. dropm: make(map[connem]float64),
  1067. ignorem: make(map[int64]bool),
  1068. }
  1069. }
  1070. func (nw *network) send(msgs ...pb.Message) {
  1071. for len(msgs) > 0 {
  1072. m := msgs[0]
  1073. p := nw.peers[m.To]
  1074. p.Step(m)
  1075. msgs = append(msgs[1:], nw.filter(p.ReadMessages())...)
  1076. }
  1077. }
  1078. func (nw *network) drop(from, to int64, perc float64) {
  1079. nw.dropm[connem{from, to}] = perc
  1080. }
  1081. func (nw *network) cut(one, other int64) {
  1082. nw.drop(one, other, 1)
  1083. nw.drop(other, one, 1)
  1084. }
  1085. func (nw *network) isolate(id int64) {
  1086. for i := 0; i < len(nw.peers); i++ {
  1087. nid := int64(i) + 1
  1088. if nid != id {
  1089. nw.drop(id, nid, 1.0)
  1090. nw.drop(nid, id, 1.0)
  1091. }
  1092. }
  1093. }
  1094. func (nw *network) ignore(t int64) {
  1095. nw.ignorem[t] = true
  1096. }
  1097. func (nw *network) recover() {
  1098. nw.dropm = make(map[connem]float64)
  1099. nw.ignorem = make(map[int64]bool)
  1100. }
  1101. func (nw *network) filter(msgs []pb.Message) []pb.Message {
  1102. mm := []pb.Message{}
  1103. for _, m := range msgs {
  1104. if nw.ignorem[m.Type] {
  1105. continue
  1106. }
  1107. switch m.Type {
  1108. case msgHup:
  1109. // hups never go over the network, so don't drop them but panic
  1110. panic("unexpected msgHup")
  1111. default:
  1112. perc := nw.dropm[connem{m.From, m.To}]
  1113. if n := rand.Float64(); n < perc {
  1114. continue
  1115. }
  1116. }
  1117. mm = append(mm, m)
  1118. }
  1119. return mm
  1120. }
  1121. type connem struct {
  1122. from, to int64
  1123. }
  1124. type blackHole struct{}
  1125. func (blackHole) Step(pb.Message) error { return nil }
  1126. func (blackHole) ReadMessages() []pb.Message { return nil }
  1127. var nopStepper = &blackHole{}