Mirroristas/etcd
2
0
Fork 0
mirror of https://github.com/etcd-io/etcd.git synced 2024-09-27 06:25:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

raft: use [1,n] as address list for raft state machines

Browse Source
This commit is contained in:
Yicheng Qin 2014-09-11 19:23:05 -07:00
parent 84c2bd0b7d
commit 0e0fc2bd24
2 changed files with 91 additions and 103 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
etcdserver/server_test.go
View File

@ -16,11 +16,6 @@ import (
func TestClusterOf1(t *testing.T) { testServer(t, 1) } func TestClusterOf1(t *testing.T) { testServer(t, 1) }
func TestClusterOf3(t *testing.T) { testServer(t, 3) } func TestClusterOf3(t *testing.T) { testServer(t, 3) }
// firstId is the id of the first raft machine in the array.
// It implies the way to set id for raft machines:
// The id of n-th machine is firstId+n, and machine with machineId is at machineId-firstId place in the array.
const firstId int64 = 0x1000
func testServer(t *testing.T, ns int64) { func testServer(t *testing.T, ns int64) {
ctx, cancel := context.WithCancel(context.Background()) ctx, cancel := context.WithCancel(context.Background())
defer cancel() defer cancel()
@ -30,17 +25,17 @@ func testServer(t *testing.T, ns int64) {
send := func(msgs []raftpb.Message) { send := func(msgs []raftpb.Message) {
for _, m := range msgs { for _, m := range msgs {
t.Logf("m = %+v\n", m) t.Logf("m = %+v\n", m)
ss[m.To-firstId].Node.Step(ctx, m) ss[m.To-1].Node.Step(ctx, m)
} }
} }
peers := make([]int64, ns) peers := make([]int64, ns)
for i := int64(0); i < ns; i++ { for i := int64(0); i < ns; i++ {
peers[i] = firstId + i peers[i] = i + 1
} }
for i := int64(0); i < ns; i++ { for i := int64(0); i < ns; i++ {
id := firstId + i id := i + 1
n := raft.Start(id, peers, 10, 1) n := raft.Start(id, peers, 10, 1)
tk := time.NewTicker(10 * time.Millisecond) tk := time.NewTicker(10 * time.Millisecond)
defer tk.Stop() defer tk.Stop()

183
raft/raft_test.go
View File

@ -10,13 +10,6 @@ import (
pb "github.com/coreos/etcd/raft/raftpb" pb "github.com/coreos/etcd/raft/raftpb"
) )
// firstId is the id of the first raft machine in the array.
// It implies the way to set id for raft machines:
// The id of n-th machine is firstId+n, and machine with machineId is at machineId-firstId place in the array.
// This is how newNetwork sets ids for raft machines.
// TODO: use more flexible ways to set it
const firstId int64 = 0x1000
// nextEnts returns the appliable entries and updates the applied index // nextEnts returns the appliable entries and updates the applied index
func (r *raft) nextEnts() (ents []pb.Entry) { func (r *raft) nextEnts() (ents []pb.Entry) {
ents = r.raftLog.nextEnts() ents = r.raftLog.nextEnts()
@ -48,8 +41,8 @@ func TestLeaderElection(t *testing.T) {
} }
for i, tt := range tests { for i, tt := range tests {
tt.send(pb.Message{From: firstId, To: firstId, Type: msgHup}) tt.send(pb.Message{From: 1, To: 1, Type: msgHup})
sm := tt.network.peers[firstId].(*raft) sm := tt.network.peers[1].(*raft)
if sm.state != tt.state { if sm.state != tt.state {
t.Errorf("#%d: state = %s, want %s", i, sm.state, tt.state) t.Errorf("#%d: state = %s, want %s", i, sm.state, tt.state)
} }
@ -68,23 +61,23 @@ func TestLogReplication(t *testing.T) {
{ {
newNetwork(nil, nil, nil), newNetwork(nil, nil, nil),
[]pb.Message{ []pb.Message{
{From: firstId, To: firstId, Type: msgProp, Entries: []pb.Entry{{Data: []byte("somedata")}}}, {From: 1, To: 1, Type: msgProp, Entries: []pb.Entry{{Data: []byte("somedata")}}},
}, },
2, 2,
}, },
{ {
newNetwork(nil, nil, nil), newNetwork(nil, nil, nil),
[]pb.Message{ []pb.Message{
{From: firstId, To: firstId, Type: msgProp, Entries: []pb.Entry{{Data: []byte("somedata")}}}, {From: 1, To: 1, Type: msgProp, Entries: []pb.Entry{{Data: []byte("somedata")}}},
{From: firstId, To: firstId + 1, Type: msgHup}, {From: 1, To: 2, Type: msgHup},
{From: firstId, To: firstId + 1, Type: msgProp, Entries: []pb.Entry{{Data: []byte("somedata")}}}, {From: 1, To: 2, Type: msgProp, Entries: []pb.Entry{{Data: []byte("somedata")}}},
}, },
4, 4,
}, },
} }
for i, tt := range tests { for i, tt := range tests {
tt.send(pb.Message{From: firstId, To: firstId, Type: msgHup}) tt.send(pb.Message{From: 1, To: 1, Type: msgHup})
for _, m := range tt.msgs { for _, m := range tt.msgs {
tt.send(m) tt.send(m)
@ -120,11 +113,11 @@ func TestLogReplication(t *testing.T) {
func TestSingleNodeCommit(t *testing.T) { func TestSingleNodeCommit(t *testing.T) {
tt := newNetwork(nil) tt := newNetwork(nil)
tt.send(pb.Message{From: firstId, To: firstId, Type: msgHup}) tt.send(pb.Message{From: 1, To: 1, Type: msgHup})
tt.send(pb.Message{From: firstId, To: firstId, Type: msgProp, Entries: []pb.Entry{{Data: []byte("some data")}}}) tt.send(pb.Message{From: 1, To: 1, Type: msgProp, Entries: []pb.Entry{{Data: []byte("some data")}}})
tt.send(pb.Message{From: firstId, To: firstId, Type: msgProp, Entries: []pb.Entry{{Data: []byte("some data")}}}) tt.send(pb.Message{From: 1, To: 1, Type: msgProp, Entries: []pb.Entry{{Data: []byte("some data")}}})
sm := tt.peers[firstId].(*raft) sm := tt.peers[1].(*raft)
if sm.raftLog.committed != 3 { if sm.raftLog.committed != 3 {
t.Errorf("committed = %d, want %d", sm.raftLog.committed, 3) t.Errorf("committed = %d, want %d", sm.raftLog.committed, 3)
} }
@ -135,17 +128,17 @@ func TestSingleNodeCommit(t *testing.T) {
// filtered. // filtered.
func TestCannotCommitWithoutNewTermEntry(t *testing.T) { func TestCannotCommitWithoutNewTermEntry(t *testing.T) {
tt := newNetwork(nil, nil, nil, nil, nil) tt := newNetwork(nil, nil, nil, nil, nil)
tt.send(pb.Message{From: firstId, To: firstId, Type: msgHup}) tt.send(pb.Message{From: 1, To: 1, Type: msgHup})
// 0 cannot reach 2,3,4 // 0 cannot reach 2,3,4
tt.cut(firstId, firstId+2) tt.cut(1, 3)
tt.cut(firstId, firstId+3) tt.cut(1, 4)
tt.cut(firstId, firstId+4) tt.cut(1, 5)
tt.send(pb.Message{From: firstId, To: firstId, Type: msgProp, Entries: []pb.Entry{{Data: []byte("some data")}}}) tt.send(pb.Message{From: 1, To: 1, Type: msgProp, Entries: []pb.Entry{{Data: []byte("some data")}}})
tt.send(pb.Message{From: firstId, To: firstId, Type: msgProp, Entries: []pb.Entry{{Data: []byte("some data")}}}) tt.send(pb.Message{From: 1, To: 1, Type: msgProp, Entries: []pb.Entry{{Data: []byte("some data")}}})
sm := tt.peers[firstId].(*raft) sm := tt.peers[1].(*raft)
if sm.raftLog.committed != 1 { if sm.raftLog.committed != 1 {
t.Errorf("committed = %d, want %d", sm.raftLog.committed, 1) t.Errorf("committed = %d, want %d", sm.raftLog.committed, 1)
} }
@ -156,10 +149,10 @@ func TestCannotCommitWithoutNewTermEntry(t *testing.T) {
tt.ignore(msgApp) tt.ignore(msgApp)
// elect 1 as the new leader with term 2 // elect 1 as the new leader with term 2
tt.send(pb.Message{From: firstId + 1, To: firstId + 1, Type: msgHup}) tt.send(pb.Message{From: 2, To: 2, Type: msgHup})
// no log entries from previous term should be committed // no log entries from previous term should be committed
sm = tt.peers[firstId+1].(*raft) sm = tt.peers[2].(*raft)
if sm.raftLog.committed != 1 { if sm.raftLog.committed != 1 {
t.Errorf("committed = %d, want %d", sm.raftLog.committed, 1) t.Errorf("committed = %d, want %d", sm.raftLog.committed, 1)
} }
@ -169,14 +162,14 @@ func TestCannotCommitWithoutNewTermEntry(t *testing.T) {
// send out a heartbeat // send out a heartbeat
// after append a ChangeTerm entry from the current term, all entries // after append a ChangeTerm entry from the current term, all entries
// should be committed // should be committed
tt.send(pb.Message{From: firstId + 1, To: firstId + 1, Type: msgBeat}) tt.send(pb.Message{From: 2, To: 2, Type: msgBeat})
if sm.raftLog.committed != 4 { if sm.raftLog.committed != 4 {
t.Errorf("committed = %d, want %d", sm.raftLog.committed, 4) t.Errorf("committed = %d, want %d", sm.raftLog.committed, 4)
} }
// still be able to append a entry // still be able to append a entry
tt.send(pb.Message{From: firstId + 1, To: firstId + 1, Type: msgProp, Entries: []pb.Entry{{Data: []byte("some data")}}}) tt.send(pb.Message{From: 2, To: 2, Type: msgProp, Entries: []pb.Entry{{Data: []byte("some data")}}})
if sm.raftLog.committed != 5 { if sm.raftLog.committed != 5 {
t.Errorf("committed = %d, want %d", sm.raftLog.committed, 5) t.Errorf("committed = %d, want %d", sm.raftLog.committed, 5)
@ -187,17 +180,17 @@ func TestCannotCommitWithoutNewTermEntry(t *testing.T) {
// when leader changes, no new proposal comes in. // when leader changes, no new proposal comes in.
func TestCommitWithoutNewTermEntry(t *testing.T) { func TestCommitWithoutNewTermEntry(t *testing.T) {
tt := newNetwork(nil, nil, nil, nil, nil) tt := newNetwork(nil, nil, nil, nil, nil)
tt.send(pb.Message{From: firstId, To: firstId, Type: msgHup}) tt.send(pb.Message{From: 1, To: 1, Type: msgHup})
// 0 cannot reach 2,3,4 // 0 cannot reach 2,3,4
tt.cut(firstId, firstId+2) tt.cut(1, 3)
tt.cut(firstId, firstId+3) tt.cut(1, 4)
tt.cut(firstId, firstId+4) tt.cut(1, 5)
tt.send(pb.Message{From: firstId, To: firstId, Type: msgProp, Entries: []pb.Entry{{Data: []byte("some data")}}}) tt.send(pb.Message{From: 1, To: 1, Type: msgProp, Entries: []pb.Entry{{Data: []byte("some data")}}})
tt.send(pb.Message{From: firstId, To: firstId, Type: msgProp, Entries: []pb.Entry{{Data: []byte("some data")}}}) tt.send(pb.Message{From: 1, To: 1, Type: msgProp, Entries: []pb.Entry{{Data: []byte("some data")}}})
sm := tt.peers[firstId].(*raft) sm := tt.peers[1].(*raft)
if sm.raftLog.committed != 1 { if sm.raftLog.committed != 1 {
t.Errorf("committed = %d, want %d", sm.raftLog.committed, 1) t.Errorf("committed = %d, want %d", sm.raftLog.committed, 1)
} }
@ -208,7 +201,7 @@ func TestCommitWithoutNewTermEntry(t *testing.T) {
// elect 1 as the new leader with term 2 // elect 1 as the new leader with term 2
// after append a ChangeTerm entry from the current term, all entries // after append a ChangeTerm entry from the current term, all entries
// should be committed // should be committed
tt.send(pb.Message{From: firstId + 1, To: firstId + 1, Type: msgHup}) tt.send(pb.Message{From: 2, To: 2, Type: msgHup})
if sm.raftLog.committed != 4 { if sm.raftLog.committed != 4 {
t.Errorf("committed = %d, want %d", sm.raftLog.committed, 4) t.Errorf("committed = %d, want %d", sm.raftLog.committed, 4)
@ -221,13 +214,13 @@ func TestDuelingCandidates(t *testing.T) {
c := newRaft(-1, nil, 0, 0) c := newRaft(-1, nil, 0, 0)
nt := newNetwork(a, b, c) nt := newNetwork(a, b, c)
nt.cut(firstId, firstId+2) nt.cut(1, 3)
nt.send(pb.Message{From: firstId, To: firstId, Type: msgHup}) nt.send(pb.Message{From: 1, To: 1, Type: msgHup})
nt.send(pb.Message{From: firstId + 2, To: firstId + 2, Type: msgHup}) nt.send(pb.Message{From: 3, To: 3, Type: msgHup})
nt.recover() nt.recover()
nt.send(pb.Message{From: firstId + 2, To: firstId + 2, Type: msgHup}) nt.send(pb.Message{From: 3, To: 3, Type: msgHup})
wlog := &raftLog{ents: []pb.Entry{{}, pb.Entry{Data: nil, Term: 1, Index: 1}}, committed: 1} wlog := &raftLog{ents: []pb.Entry{{}, pb.Entry{Data: nil, Term: 1, Index: 1}}, committed: 1}
tests := []struct { tests := []struct {
@ -249,7 +242,7 @@ func TestDuelingCandidates(t *testing.T) {
t.Errorf("#%d: term = %d, want %d", i, g, tt.term) t.Errorf("#%d: term = %d, want %d", i, g, tt.term)
} }
base := ltoa(tt.raftLog) base := ltoa(tt.raftLog)
if sm, ok := nt.peers[firstId+int64(i)].(*raft); ok { if sm, ok := nt.peers[1+int64(i)].(*raft); ok {
l := ltoa(sm.raftLog) l := ltoa(sm.raftLog)
if g := diffu(base, l); g != "" { if g := diffu(base, l); g != "" {
t.Errorf("#%d: diff:\n%s", i, g) t.Errorf("#%d: diff:\n%s", i, g)
@ -262,19 +255,19 @@ func TestDuelingCandidates(t *testing.T) {
func TestCandidateConcede(t *testing.T) { func TestCandidateConcede(t *testing.T) {
tt := newNetwork(nil, nil, nil) tt := newNetwork(nil, nil, nil)
tt.isolate(firstId) tt.isolate(1)
tt.send(pb.Message{From: firstId, To: firstId, Type: msgHup}) tt.send(pb.Message{From: 1, To: 1, Type: msgHup})
tt.send(pb.Message{From: firstId + 2, To: firstId + 2, Type: msgHup}) tt.send(pb.Message{From: 3, To: 3, Type: msgHup})
// heal the partition // heal the partition
tt.recover() tt.recover()
data := []byte("force follower") data := []byte("force follower")
// send a proposal to 2 to flush out a msgApp to 0 // send a proposal to 2 to flush out a msgApp to 0
tt.send(pb.Message{From: firstId + 2, To: firstId + 2, Type: msgProp, Entries: []pb.Entry{{Data: data}}}) tt.send(pb.Message{From: 3, To: 3, Type: msgProp, Entries: []pb.Entry{{Data: data}}})
a := tt.peers[firstId].(*raft) a := tt.peers[1].(*raft)
if g := a.state; g != stateFollower { if g := a.state; g != stateFollower {
t.Errorf("state = %s, want %s", g, stateFollower) t.Errorf("state = %s, want %s", g, stateFollower)
} }
@ -296,9 +289,9 @@ func TestCandidateConcede(t *testing.T) {
func TestSingleNodeCandidate(t *testing.T) { func TestSingleNodeCandidate(t *testing.T) {
tt := newNetwork(nil) tt := newNetwork(nil)
tt.send(pb.Message{From: firstId, To: firstId, Type: msgHup}) tt.send(pb.Message{From: 1, To: 1, Type: msgHup})
sm := tt.peers[firstId].(*raft) sm := tt.peers[1].(*raft)
if sm.state != stateLeader { if sm.state != stateLeader {
t.Errorf("state = %d, want %d", sm.state, stateLeader) t.Errorf("state = %d, want %d", sm.state, stateLeader)
} }
@ -307,11 +300,11 @@ func TestSingleNodeCandidate(t *testing.T) {
func TestOldMessages(t *testing.T) { func TestOldMessages(t *testing.T) {
tt := newNetwork(nil, nil, nil) tt := newNetwork(nil, nil, nil)
// make 0 leader @ term 3 // make 0 leader @ term 3
tt.send(pb.Message{From: firstId, To: firstId, Type: msgHup}) tt.send(pb.Message{From: 1, To: 1, Type: msgHup})
tt.send(pb.Message{From: firstId + 1, To: firstId + 1, Type: msgHup}) tt.send(pb.Message{From: 2, To: 2, Type: msgHup})
tt.send(pb.Message{From: firstId, To: firstId, Type: msgHup}) tt.send(pb.Message{From: 1, To: 1, Type: msgHup})
// pretend we're an old leader trying to make progress // pretend we're an old leader trying to make progress
tt.send(pb.Message{From: firstId, To: firstId, Type: msgApp, Term: 1, Entries: []pb.Entry{{Term: 1}}}) tt.send(pb.Message{From: 1, To: 1, Type: msgApp, Term: 1, Entries: []pb.Entry{{Term: 1}}})
l := &raftLog{ l := &raftLog{
ents: []pb.Entry{ ents: []pb.Entry{
@ -365,8 +358,8 @@ func TestProposal(t *testing.T) {
data := []byte("somedata") data := []byte("somedata")
// promote 0 the leader // promote 0 the leader
send(pb.Message{From: firstId, To: firstId, Type: msgHup}) send(pb.Message{From: 1, To: 1, Type: msgHup})
send(pb.Message{From: firstId, To: firstId, Type: msgProp, Entries: []pb.Entry{{Data: data}}}) send(pb.Message{From: 1, To: 1, Type: msgProp, Entries: []pb.Entry{{Data: data}}})
wantLog := newLog() wantLog := newLog()
if tt.success { if tt.success {
@ -383,7 +376,7 @@ func TestProposal(t *testing.T) {
t.Logf("#%d: empty log", i) t.Logf("#%d: empty log", i)
} }
} }
sm := tt.network.peers[firstId].(*raft) sm := tt.network.peers[1].(*raft)
if g := sm.Term; g != 1 { if g := sm.Term; g != 1 {
t.Errorf("#%d: term = %d, want %d", i, g, 1) t.Errorf("#%d: term = %d, want %d", i, g, 1)
} }
@ -399,10 +392,10 @@ func TestProposalByProxy(t *testing.T) {
for i, tt := range tests { for i, tt := range tests {
// promote 0 the leader // promote 0 the leader
tt.send(pb.Message{From: firstId, To: firstId, Type: msgHup}) tt.send(pb.Message{From: 1, To: 1, Type: msgHup})
// propose via follower // propose via follower
tt.send(pb.Message{From: firstId + 1, To: firstId + 1, Type: msgProp, Entries: []pb.Entry{{Data: []byte("somedata")}}}) tt.send(pb.Message{From: 2, To: 2, Type: msgProp, Entries: []pb.Entry{{Data: []byte("somedata")}}})
wantLog := &raftLog{ents: []pb.Entry{{}, {Data: nil, Term: 1, Index: 1}, {Term: 1, Data: data, Index: 2}}, committed: 2} wantLog := &raftLog{ents: []pb.Entry{{}, {Data: nil, Term: 1, Index: 1}, {Term: 1, Data: data, Index: 2}}, committed: 2}
base := ltoa(wantLog) base := ltoa(wantLog)
@ -416,7 +409,7 @@ func TestProposalByProxy(t *testing.T) {
t.Logf("#%d: empty log", i) t.Logf("#%d: empty log", i)
} }
} }
sm := tt.peers[firstId].(*raft) sm := tt.peers[1].(*raft)
if g := sm.Term; g != 1 { if g := sm.Term; g != 1 {
t.Errorf("#%d: term = %d, want %d", i, g, 1) t.Errorf("#%d: term = %d, want %d", i, g, 1)
} }
@ -471,7 +464,7 @@ func TestStepIgnoreOldTermMsg(t *testing.T) {
fakeStep := func(r *raft, m pb.Message) { fakeStep := func(r *raft, m pb.Message) {
called = true called = true
} }
sm := newRaft(firstId, []int64{firstId}, 0, 0) sm := newRaft(1, []int64{1}, 0, 0)
sm.step = fakeStep sm.step = fakeStep
sm.Term = 2 sm.Term = 2
sm.Step(pb.Message{Type: msgApp, Term: sm.Term - 1}) sm.Step(pb.Message{Type: msgApp, Term: sm.Term - 1})
@ -563,15 +556,15 @@ func TestRecvMsgVote(t *testing.T) {
{stateFollower, 3, 2, none, 2}, {stateFollower, 3, 2, none, 2},
{stateFollower, 3, 3, none, 2}, {stateFollower, 3, 3, none, 2},
{stateFollower, 3, 2, firstId + 1, 2}, {stateFollower, 3, 2, 2, 2},
{stateFollower, 3, 2, firstId, -1}, {stateFollower, 3, 2, 1, -1},
{stateLeader, 3, 3, firstId, -1}, {stateLeader, 3, 3, 1, -1},
{stateCandidate, 3, 3, firstId, -1}, {stateCandidate, 3, 3, 1, -1},
} }
for i, tt := range tests { for i, tt := range tests {
sm := newRaft(firstId, []int64{firstId}, 0, 0) sm := newRaft(1, []int64{1}, 0, 0)
sm.state = tt.state sm.state = tt.state
switch tt.state { switch tt.state {
case stateFollower: case stateFollower:
@ -584,7 +577,7 @@ func TestRecvMsgVote(t *testing.T) {
sm.State = pb.State{Vote: tt.voteFor} sm.State = pb.State{Vote: tt.voteFor}
sm.raftLog = &raftLog{ents: []pb.Entry{{}, {Term: 2}, {Term: 2}}} sm.raftLog = &raftLog{ents: []pb.Entry{{}, {Term: 2}, {Term: 2}}}
sm.Step(pb.Message{Type: msgVote, From: firstId + 1, Index: tt.i, LogTerm: tt.term}) sm.Step(pb.Message{Type: msgVote, From: 2, Index: tt.i, LogTerm: tt.term})
msgs := sm.ReadMessages() msgs := sm.ReadMessages()
if g := len(msgs); g != 1 { if g := len(msgs); g != 1 {
@ -611,11 +604,11 @@ func TestStateTransition(t *testing.T) {
{stateCandidate, stateFollower, true, 0, none}, {stateCandidate, stateFollower, true, 0, none},
{stateCandidate, stateCandidate, true, 1, none}, {stateCandidate, stateCandidate, true, 1, none},
{stateCandidate, stateLeader, true, 0, firstId}, {stateCandidate, stateLeader, true, 0, 1},
{stateLeader, stateFollower, true, 1, none}, {stateLeader, stateFollower, true, 1, none},
{stateLeader, stateCandidate, false, 1, none}, {stateLeader, stateCandidate, false, 1, none},
{stateLeader, stateLeader, true, 0, firstId}, {stateLeader, stateLeader, true, 0, 1},
} }
for i, tt := range tests { for i, tt := range tests {
@ -628,7 +621,7 @@ func TestStateTransition(t *testing.T) {
} }
}() }()
sm := newRaft(firstId, []int64{firstId}, 0, 0) sm := newRaft(1, []int64{1}, 0, 0)
sm.state = tt.from sm.state = tt.from
switch tt.to { switch tt.to {
@ -667,7 +660,7 @@ func TestAllServerStepdown(t *testing.T) {
tterm := int64(3) tterm := int64(3)
for i, tt := range tests { for i, tt := range tests {
sm := newRaft(firstId, []int64{firstId, firstId + 1, firstId + 2}, 0, 0) sm := newRaft(1, []int64{1, 2, 3}, 0, 0)
switch tt.state { switch tt.state {
case stateFollower: case stateFollower:
sm.becomeFollower(1, none) sm.becomeFollower(1, none)
@ -679,7 +672,7 @@ func TestAllServerStepdown(t *testing.T) {
} }
for j, msgType := range tmsgTypes { for j, msgType := range tmsgTypes {
sm.Step(pb.Message{From: firstId + 1, Type: msgType, Term: tterm, LogTerm: tterm}) sm.Step(pb.Message{From: 2, Type: msgType, Term: tterm, LogTerm: tterm})
if sm.state != tt.wstate { if sm.state != tt.wstate {
t.Errorf("#%d.%d state = %v , want %v", i, j, sm.state, tt.wstate) t.Errorf("#%d.%d state = %v , want %v", i, j, sm.state, tt.wstate)
@ -690,7 +683,7 @@ func TestAllServerStepdown(t *testing.T) {
if int64(len(sm.raftLog.ents)) != tt.windex { if int64(len(sm.raftLog.ents)) != tt.windex {
t.Errorf("#%d.%d index = %v , want %v", i, j, len(sm.raftLog.ents), tt.windex) t.Errorf("#%d.%d index = %v , want %v", i, j, len(sm.raftLog.ents), tt.windex)
} }
wlead := int64(firstId + 1) wlead := int64(2)
if msgType == msgVote { if msgType == msgVote {
wlead = none wlead = none
} }
@ -715,12 +708,12 @@ func TestLeaderAppResp(t *testing.T) {
for i, tt := range tests { for i, tt := range tests {
// sm term is 1 after it becomes the leader. // sm term is 1 after it becomes the leader.
// thus the last log term must be 1 to be committed. // thus the last log term must be 1 to be committed.
sm := newRaft(firstId, []int64{firstId, firstId + 1, firstId + 2}, 0, 0) sm := newRaft(1, []int64{1, 2, 3}, 0, 0)
sm.raftLog = &raftLog{ents: []pb.Entry{{}, {Term: 0}, {Term: 1}}} sm.raftLog = &raftLog{ents: []pb.Entry{{}, {Term: 0}, {Term: 1}}}
sm.becomeCandidate() sm.becomeCandidate()
sm.becomeLeader() sm.becomeLeader()
sm.ReadMessages() sm.ReadMessages()
sm.Step(pb.Message{From: firstId + 1, Type: msgAppResp, Index: tt.index, Term: sm.Term}) sm.Step(pb.Message{From: 2, Type: msgAppResp, Index: tt.index, Term: sm.Term})
msgs := sm.ReadMessages() msgs := sm.ReadMessages()
if len(msgs) != tt.wmsgNum { if len(msgs) != tt.wmsgNum {
@ -750,7 +743,7 @@ func TestRecvMsgBeat(t *testing.T) {
} }
for i, tt := range tests { for i, tt := range tests {
sm := newRaft(firstId, []int64{firstId, firstId + 1, firstId + 2}, 0, 0) sm := newRaft(1, []int64{1, 2, 3}, 0, 0)
sm.raftLog = &raftLog{ents: []pb.Entry{{}, {Term: 0}, {Term: 1}}} sm.raftLog = &raftLog{ents: []pb.Entry{{}, {Term: 0}, {Term: 1}}}
sm.Term = 1 sm.Term = 1
sm.state = tt.state sm.state = tt.state
@ -762,7 +755,7 @@ func TestRecvMsgBeat(t *testing.T) {
case stateLeader: case stateLeader:
sm.step = stepLeader sm.step = stepLeader
} }
sm.Step(pb.Message{From: firstId, To: firstId, Type: msgBeat}) sm.Step(pb.Message{From: 1, To: 1, Type: msgBeat})
msgs := sm.ReadMessages() msgs := sm.ReadMessages()
if len(msgs) != tt.wMsg { if len(msgs) != tt.wMsg {
@ -780,10 +773,10 @@ func TestRestore(t *testing.T) {
s := pb.Snapshot{ s := pb.Snapshot{
Index: defaultCompactThreshold + 1, Index: defaultCompactThreshold + 1,
Term: defaultCompactThreshold + 1, Term: defaultCompactThreshold + 1,
Nodes: []int64{firstId, firstId + 1, firstId + 2}, Nodes: []int64{1, 2, 3},
} }
sm := newRaft(firstId, []int64{firstId, firstId + 1}, 0, 0) sm := newRaft(1, []int64{1, 2}, 0, 0)
if ok := sm.restore(s); !ok { if ok := sm.restore(s); !ok {
t.Fatal("restore fail, want succeed") t.Fatal("restore fail, want succeed")
} }
@ -814,9 +807,9 @@ func TestProvideSnap(t *testing.T) {
s := pb.Snapshot{ s := pb.Snapshot{
Index: defaultCompactThreshold + 1, Index: defaultCompactThreshold + 1,
Term: defaultCompactThreshold + 1, Term: defaultCompactThreshold + 1,
Nodes: []int64{firstId, firstId + 1}, Nodes: []int64{1, 2},
} }
sm := newRaft(firstId, []int64{firstId}, 0, 0) sm := newRaft(1, []int64{1}, 0, 0)
// restore the statemachin from a snapshot // restore the statemachin from a snapshot
// so it has a compacted log and a snapshot // so it has a compacted log and a snapshot
sm.restore(s) sm.restore(s)
@ -824,7 +817,7 @@ func TestProvideSnap(t *testing.T) {
sm.becomeCandidate() sm.becomeCandidate()
sm.becomeLeader() sm.becomeLeader()
sm.Step(pb.Message{From: firstId, To: firstId, Type: msgBeat}) sm.Step(pb.Message{From: 1, To: 1, Type: msgBeat})
msgs := sm.ReadMessages() msgs := sm.ReadMessages()
if len(msgs) != 1 { if len(msgs) != 1 {
t.Errorf("len(msgs) = %d, want 1", len(msgs)) t.Errorf("len(msgs) = %d, want 1", len(msgs))
@ -836,9 +829,9 @@ func TestProvideSnap(t *testing.T) {
// force set the next of node 1, so that // force set the next of node 1, so that
// node 1 needs a snapshot // node 1 needs a snapshot
sm.prs[firstId+1].next = sm.raftLog.offset sm.prs[2].next = sm.raftLog.offset
sm.Step(pb.Message{From: firstId + 1, To: firstId, Type: msgAppResp, Index: -1}) sm.Step(pb.Message{From: 2, To: 1, Type: msgAppResp, Index: -1})
msgs = sm.ReadMessages() msgs = sm.ReadMessages()
if len(msgs) != 1 { if len(msgs) != 1 {
t.Errorf("len(msgs) = %d, want 1", len(msgs)) t.Errorf("len(msgs) = %d, want 1", len(msgs))
@ -853,11 +846,11 @@ func TestRestoreFromSnapMsg(t *testing.T) {
s := pb.Snapshot{ s := pb.Snapshot{
Index: defaultCompactThreshold + 1, Index: defaultCompactThreshold + 1,
Term: defaultCompactThreshold + 1, Term: defaultCompactThreshold + 1,
Nodes: []int64{firstId, firstId + 1}, Nodes: []int64{1, 2},
} }
m := pb.Message{Type: msgSnap, From: firstId, Term: firstId + 1, Snapshot: s} m := pb.Message{Type: msgSnap, From: 1, Term: 2, Snapshot: s}
sm := newRaft(firstId+1, []int64{firstId, firstId + 1}, 0, 0) sm := newRaft(2, []int64{1, 2}, 0, 0)
sm.Step(m) sm.Step(m)
if !reflect.DeepEqual(sm.raftLog.snapshot, s) { if !reflect.DeepEqual(sm.raftLog.snapshot, s) {
@ -867,26 +860,26 @@ func TestRestoreFromSnapMsg(t *testing.T) {
func TestSlowNodeRestore(t *testing.T) { func TestSlowNodeRestore(t *testing.T) {
nt := newNetwork(nil, nil, nil) nt := newNetwork(nil, nil, nil)
nt.send(pb.Message{From: firstId, To: firstId, Type: msgHup}) nt.send(pb.Message{From: 1, To: 1, Type: msgHup})
nt.isolate(firstId + 2) nt.isolate(3)
for j := 0; j < defaultCompactThreshold+1; j++ { for j := 0; j < defaultCompactThreshold+1; j++ {
nt.send(pb.Message{From: firstId, To: firstId, Type: msgProp, Entries: []pb.Entry{{}}}) nt.send(pb.Message{From: 1, To: 1, Type: msgProp, Entries: []pb.Entry{{}}})
} }
lead := nt.peers[firstId].(*raft) lead := nt.peers[1].(*raft)
lead.nextEnts() lead.nextEnts()
lead.compact(nil) lead.compact(nil)
nt.recover() nt.recover()
nt.send(pb.Message{From: firstId, To: firstId, Type: msgBeat}) nt.send(pb.Message{From: 1, To: 1, Type: msgBeat})
follower := nt.peers[firstId+2].(*raft) follower := nt.peers[3].(*raft)
if !reflect.DeepEqual(follower.raftLog.snapshot, lead.raftLog.snapshot) { if !reflect.DeepEqual(follower.raftLog.snapshot, lead.raftLog.snapshot) {
t.Errorf("follower.snap = %+v, want %+v", follower.raftLog.snapshot, lead.raftLog.snapshot) t.Errorf("follower.snap = %+v, want %+v", follower.raftLog.snapshot, lead.raftLog.snapshot)
} }
committed := follower.raftLog.lastIndex() committed := follower.raftLog.lastIndex()
nt.send(pb.Message{From: firstId, To: firstId, Type: msgProp, Entries: []pb.Entry{{}}}) nt.send(pb.Message{From: 1, To: 1, Type: msgProp, Entries: []pb.Entry{{}}})
if follower.raftLog.committed != committed+1 { if follower.raftLog.committed != committed+1 {
t.Errorf("follower.comitted = %d, want %d", follower.raftLog.committed, committed+1) t.Errorf("follower.comitted = %d, want %d", follower.raftLog.committed, committed+1)
} }
@ -917,7 +910,7 @@ func newNetwork(peers ...Interface) *network {
size := len(peers) size := len(peers)
peerAddrs := make([]int64, size) peerAddrs := make([]int64, size)
for i := 0; i < size; i++ { for i := 0; i < size; i++ {
peerAddrs[i] = firstId + int64(i) peerAddrs[i] = 1 + int64(i)
} }
npeers := make(map[int64]Interface, size) npeers := make(map[int64]Interface, size)
@ -969,7 +962,7 @@ func (nw *network) cut(one, other int64) {
func (nw *network) isolate(id int64) { func (nw *network) isolate(id int64) {
for i := 0; i < len(nw.peers); i++ { for i := 0; i < len(nw.peers); i++ {
nid := firstId + int64(i) nid := 1 + int64(i)
if nid != id { if nid != id {
nw.drop(id, nid, 1.0) nw.drop(id, nid, 1.0)
nw.drop(nid, id, 1.0) nw.drop(nid, id, 1.0)