node_test.go 6.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293
  1. package raft
  2. import (
  3. "reflect"
  4. "testing"
  5. "time"
  6. "github.com/coreos/etcd/pkg"
  7. "github.com/coreos/etcd/raft/raftpb"
  8. "github.com/coreos/etcd/third_party/code.google.com/p/go.net/context"
  9. )
  10. // TestNodeStep ensures that node.Step sends msgProp to propc chan
  11. // and other kinds of messages to recvc chan.
  12. func TestNodeStep(t *testing.T) {
  13. for i := range mtmap {
  14. n := &node{
  15. propc: make(chan raftpb.Message, 1),
  16. recvc: make(chan raftpb.Message, 1),
  17. }
  18. msgt := int64(i)
  19. n.Step(context.TODO(), raftpb.Message{Type: msgt})
  20. // Proposal goes to proc chan. Others go to recvc chan.
  21. if int64(i) == msgProp {
  22. select {
  23. case <-n.propc:
  24. default:
  25. t.Errorf("%d: cannot receive %s on propc chan", i, mtmap[i])
  26. }
  27. } else {
  28. if msgt == msgBeat || msgt == msgHup {
  29. select {
  30. case <-n.recvc:
  31. t.Errorf("%d: step should ignore msgHub/msgBeat", i, mtmap[i])
  32. default:
  33. }
  34. } else {
  35. select {
  36. case <-n.recvc:
  37. default:
  38. t.Errorf("%d: cannot receive %s on recvc chan", i, mtmap[i])
  39. }
  40. }
  41. }
  42. }
  43. }
  44. // Cancel and Stop should unblock Step()
  45. func TestNodeStepUnblock(t *testing.T) {
  46. // a node without buffer to block step
  47. n := &node{
  48. propc: make(chan raftpb.Message),
  49. done: make(chan struct{}),
  50. }
  51. ctx, cancel := context.WithCancel(context.Background())
  52. stopFunc := func() { close(n.done) }
  53. tests := []struct {
  54. unblock func()
  55. werr error
  56. }{
  57. {stopFunc, ErrStopped},
  58. {cancel, context.Canceled},
  59. }
  60. for i, tt := range tests {
  61. errc := make(chan error, 1)
  62. go func() {
  63. err := n.Step(ctx, raftpb.Message{Type: msgProp})
  64. errc <- err
  65. }()
  66. tt.unblock()
  67. select {
  68. case err := <-errc:
  69. if err != tt.werr {
  70. t.Errorf("#%d: err = %v, want %v", err, tt.werr)
  71. }
  72. //clean up side-effect
  73. if ctx.Err() != nil {
  74. ctx = context.TODO()
  75. }
  76. select {
  77. case <-n.done:
  78. n.done = make(chan struct{})
  79. default:
  80. }
  81. case <-time.After(time.Millisecond * 100):
  82. t.Errorf("#%d: failed to unblock step", i)
  83. }
  84. }
  85. }
  86. // TestBlockProposal ensures that node will block proposal when it does not
  87. // know who is the current leader; node will accept proposal when it knows
  88. // who is the current leader.
  89. func TestBlockProposal(t *testing.T) {
  90. n := newNode()
  91. r := newRaft(1, []int64{1}, 10, 1)
  92. go n.run(r)
  93. defer n.Stop()
  94. errc := make(chan error, 1)
  95. go func() {
  96. errc <- n.Propose(context.TODO(), []byte("somedata"))
  97. }()
  98. pkg.ForceGosched()
  99. select {
  100. case err := <-errc:
  101. t.Errorf("err = %v, want blocking", err)
  102. default:
  103. }
  104. n.Campaign(context.TODO())
  105. pkg.ForceGosched()
  106. select {
  107. case err := <-errc:
  108. if err != nil {
  109. t.Errorf("err = %v, want %v", err, nil)
  110. }
  111. default:
  112. t.Errorf("blocking proposal, want unblocking")
  113. }
  114. }
  115. func TestReadyContainUpdates(t *testing.T) {
  116. tests := []struct {
  117. rd Ready
  118. wcontain bool
  119. }{
  120. {Ready{}, false},
  121. {Ready{SoftState: &SoftState{Lead: 1}}, true},
  122. {Ready{HardState: raftpb.HardState{Vote: 1}}, true},
  123. {Ready{Entries: make([]raftpb.Entry, 1, 1)}, true},
  124. {Ready{CommittedEntries: make([]raftpb.Entry, 1, 1)}, true},
  125. {Ready{Messages: make([]raftpb.Message, 1, 1)}, true},
  126. {Ready{Snapshot: raftpb.Snapshot{Index: 1}}, true},
  127. }
  128. for i, tt := range tests {
  129. if g := tt.rd.containsUpdates(); g != tt.wcontain {
  130. t.Errorf("#%d: containUpdates = %v, want %v", i, g, tt.wcontain)
  131. }
  132. }
  133. }
  134. func TestNode(t *testing.T) {
  135. ctx, cancel := context.WithCancel(context.Background())
  136. defer cancel()
  137. wants := []Ready{
  138. {
  139. SoftState: &SoftState{Lead: 1, RaftState: StateLeader},
  140. HardState: raftpb.HardState{Term: 1, Commit: 1},
  141. Entries: []raftpb.Entry{{}, {Term: 1, Index: 1}},
  142. CommittedEntries: []raftpb.Entry{{Term: 1, Index: 1}},
  143. },
  144. {
  145. HardState: raftpb.HardState{Term: 1, Commit: 2},
  146. Entries: []raftpb.Entry{{Term: 1, Index: 2, Data: []byte("foo")}},
  147. CommittedEntries: []raftpb.Entry{{Term: 1, Index: 2, Data: []byte("foo")}},
  148. },
  149. }
  150. n := StartNode(1, []int64{1}, 0, 0)
  151. n.Campaign(ctx)
  152. if g := <-n.Ready(); !reflect.DeepEqual(g, wants[0]) {
  153. t.Errorf("#%d: g = %+v,\n w %+v", 1, g, wants[0])
  154. }
  155. n.Propose(ctx, []byte("foo"))
  156. if g := <-n.Ready(); !reflect.DeepEqual(g, wants[1]) {
  157. t.Errorf("#%d: g = %+v,\n w %+v", 2, g, wants[1])
  158. }
  159. select {
  160. case rd := <-n.Ready():
  161. t.Errorf("unexpected Ready: %+v", rd)
  162. default:
  163. }
  164. }
  165. func TestNodeRestart(t *testing.T) {
  166. entries := []raftpb.Entry{
  167. {},
  168. {Term: 1, Index: 1},
  169. {Term: 1, Index: 2, Data: []byte("foo")},
  170. }
  171. st := raftpb.HardState{Term: 1, Commit: 1}
  172. want := Ready{
  173. HardState: emptyState,
  174. // commit upto index commit index in st
  175. CommittedEntries: entries[1 : st.Commit+1],
  176. }
  177. n := RestartNode(1, []int64{1}, 0, 0, nil, st, entries)
  178. if g := <-n.Ready(); !reflect.DeepEqual(g, want) {
  179. t.Errorf("g = %+v,\n w %+v", g, want)
  180. }
  181. select {
  182. case rd := <-n.Ready():
  183. t.Errorf("unexpected Ready: %+v", rd)
  184. default:
  185. }
  186. }
  187. // TestCompacts ensures Node.Compact creates a correct raft snapshot and compacts
  188. // the raft log (call raft.compact)
  189. func TestCompact(t *testing.T) {
  190. ctx := context.Background()
  191. n := newNode()
  192. r := newRaft(1, []int64{1}, 0, 0)
  193. go n.run(r)
  194. n.Campaign(ctx)
  195. n.Propose(ctx, []byte("foo"))
  196. w := raftpb.Snapshot{
  197. Term: 1,
  198. Index: 2, // one nop + one proposal
  199. Data: []byte("a snapshot"),
  200. Nodes: []int64{1},
  201. }
  202. pkg.ForceGosched()
  203. select {
  204. case <-n.Ready():
  205. default:
  206. t.Fatalf("unexpected proposal failure: unable to commit entry")
  207. }
  208. n.Compact(w.Data)
  209. pkg.ForceGosched()
  210. select {
  211. case rd := <-n.Ready():
  212. if !reflect.DeepEqual(rd.Snapshot, w) {
  213. t.Errorf("snap = %+v, want %+v", rd.Snapshot, w)
  214. }
  215. default:
  216. t.Fatalf("unexpected compact failure: unable to create a snapshot")
  217. }
  218. pkg.ForceGosched()
  219. // TODO: this test the run updates the snapi correctly... should be tested
  220. // separately with other kinds of updates
  221. select {
  222. case <-n.Ready():
  223. t.Fatalf("unexpected more ready")
  224. default:
  225. }
  226. n.Stop()
  227. if r.raftLog.offset != w.Index {
  228. t.Errorf("log.offset = %d, want %d", r.raftLog.offset, w.Index)
  229. }
  230. }
  231. func TestSoftStateEqual(t *testing.T) {
  232. tests := []struct {
  233. st *SoftState
  234. we bool
  235. }{
  236. {&SoftState{}, true},
  237. {&SoftState{Lead: 1}, false},
  238. {&SoftState{RaftState: StateLeader}, false},
  239. {&SoftState{ShouldStop: true}, false},
  240. }
  241. for i, tt := range tests {
  242. if g := tt.st.equal(&SoftState{}); g != tt.we {
  243. t.Errorf("#%d, equal = %v, want %v", i, g, tt.we)
  244. }
  245. }
  246. }
  247. func TestIsHardStateEqual(t *testing.T) {
  248. tests := []struct {
  249. st raftpb.HardState
  250. we bool
  251. }{
  252. {emptyState, true},
  253. {raftpb.HardState{Vote: 1}, false},
  254. {raftpb.HardState{Commit: 1}, false},
  255. {raftpb.HardState{Term: 1}, false},
  256. }
  257. for i, tt := range tests {
  258. if isHardStateEqual(tt.st, emptyState) != tt.we {
  259. t.Errorf("#%d, equal = %v, want %v", i, isHardStateEqual(tt.st, emptyState), tt.we)
  260. }
  261. }
  262. }