mirror of
https://github.com/etcd-io/etcd.git
synced 2024-09-27 06:25:44 +00:00
018fb8e6d9
ForceGosched() performs bad when GOMAXPROCS>1. When GOMAXPROCS=1, it
could promise that other goroutines run long enough
because it always yield the processor to other goroutines. But it cannot
yield processor to goroutine running on other processors. So when
GOMAXPROCS>1, the yield may finish when goroutine on the other
processor just runs for little time.
Here is a test to confirm the case:
```
package main
import (
"fmt"
"runtime"
"testing"
)
func ForceGosched() {
// possibility enough to sched up to 10 go routines.
for i := 0; i < 10000; i++ {
runtime.Gosched()
}
}
var d int
func loop(c chan struct{}) {
for {
select {
case <-c:
for i := 0; i < 1000; i++ {
fmt.Sprintf("come to time %d", i)
}
d++
}
}
}
func TestLoop(t *testing.T) {
c := make(chan struct{}, 1)
go loop(c)
c <- struct{}{}
ForceGosched()
if d != 1 {
t.Fatal("d is not incremented")
}
}
```
`go test -v -race` runs well, but `GOMAXPROCS=2 go test -v -race` fails.
Change the functionality to waiting for schedule to happen.
504 lines
12 KiB
Go
504 lines
12 KiB
Go
// Copyright 2015 CoreOS, Inc.
|
|
//
|
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
|
// you may not use this file except in compliance with the License.
|
|
// You may obtain a copy of the License at
|
|
//
|
|
// http://www.apache.org/licenses/LICENSE-2.0
|
|
//
|
|
// Unless required by applicable law or agreed to in writing, software
|
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
// See the License for the specific language governing permissions and
|
|
// limitations under the License.
|
|
|
|
package raft
|
|
|
|
import (
|
|
"reflect"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/coreos/etcd/Godeps/_workspace/src/golang.org/x/net/context"
|
|
"github.com/coreos/etcd/pkg/testutil"
|
|
"github.com/coreos/etcd/raft/raftpb"
|
|
)
|
|
|
|
// TestNodeStep ensures that node.Step sends msgProp to propc chan
|
|
// and other kinds of messages to recvc chan.
|
|
func TestNodeStep(t *testing.T) {
|
|
for i, msgn := range raftpb.MessageType_name {
|
|
n := &node{
|
|
propc: make(chan raftpb.Message, 1),
|
|
recvc: make(chan raftpb.Message, 1),
|
|
}
|
|
msgt := raftpb.MessageType(i)
|
|
n.Step(context.TODO(), raftpb.Message{Type: msgt})
|
|
// Proposal goes to proc chan. Others go to recvc chan.
|
|
if msgt == raftpb.MsgProp {
|
|
select {
|
|
case <-n.propc:
|
|
default:
|
|
t.Errorf("%d: cannot receive %s on propc chan", msgt, msgn)
|
|
}
|
|
} else {
|
|
if msgt == raftpb.MsgBeat || msgt == raftpb.MsgHup || msgt == raftpb.MsgUnreachable || msgt == raftpb.MsgSnapStatus {
|
|
select {
|
|
case <-n.recvc:
|
|
t.Errorf("%d: step should ignore %s", msgt, msgn)
|
|
default:
|
|
}
|
|
} else {
|
|
select {
|
|
case <-n.recvc:
|
|
default:
|
|
t.Errorf("%d: cannot receive %s on recvc chan", msgt, msgn)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// Cancel and Stop should unblock Step()
|
|
func TestNodeStepUnblock(t *testing.T) {
|
|
// a node without buffer to block step
|
|
n := &node{
|
|
propc: make(chan raftpb.Message),
|
|
done: make(chan struct{}),
|
|
}
|
|
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
stopFunc := func() { close(n.done) }
|
|
|
|
tests := []struct {
|
|
unblock func()
|
|
werr error
|
|
}{
|
|
{stopFunc, ErrStopped},
|
|
{cancel, context.Canceled},
|
|
}
|
|
|
|
for i, tt := range tests {
|
|
errc := make(chan error, 1)
|
|
go func() {
|
|
err := n.Step(ctx, raftpb.Message{Type: raftpb.MsgProp})
|
|
errc <- err
|
|
}()
|
|
tt.unblock()
|
|
select {
|
|
case err := <-errc:
|
|
if err != tt.werr {
|
|
t.Errorf("#%d: err = %v, want %v", i, err, tt.werr)
|
|
}
|
|
//clean up side-effect
|
|
if ctx.Err() != nil {
|
|
ctx = context.TODO()
|
|
}
|
|
select {
|
|
case <-n.done:
|
|
n.done = make(chan struct{})
|
|
default:
|
|
}
|
|
case <-time.After(time.Millisecond * 100):
|
|
t.Errorf("#%d: failed to unblock step", i)
|
|
}
|
|
}
|
|
}
|
|
|
|
// TestNodePropose ensures that node.Propose sends the given proposal to the underlying raft.
|
|
func TestNodePropose(t *testing.T) {
|
|
msgs := []raftpb.Message{}
|
|
appendStep := func(r *raft, m raftpb.Message) {
|
|
msgs = append(msgs, m)
|
|
}
|
|
|
|
n := newNode()
|
|
s := NewMemoryStorage()
|
|
r := newTestRaft(1, []uint64{1}, 10, 1, s)
|
|
go n.run(r)
|
|
n.Campaign(context.TODO())
|
|
for {
|
|
rd := <-n.Ready()
|
|
s.Append(rd.Entries)
|
|
// change the step function to appendStep until this raft becomes leader
|
|
if rd.SoftState.Lead == r.id {
|
|
r.step = appendStep
|
|
n.Advance()
|
|
break
|
|
}
|
|
n.Advance()
|
|
}
|
|
n.Propose(context.TODO(), []byte("somedata"))
|
|
n.Stop()
|
|
|
|
if len(msgs) != 1 {
|
|
t.Fatalf("len(msgs) = %d, want %d", len(msgs), 1)
|
|
}
|
|
if msgs[0].Type != raftpb.MsgProp {
|
|
t.Errorf("msg type = %d, want %d", msgs[0].Type, raftpb.MsgProp)
|
|
}
|
|
if !reflect.DeepEqual(msgs[0].Entries[0].Data, []byte("somedata")) {
|
|
t.Errorf("data = %v, want %v", msgs[0].Entries[0].Data, []byte("somedata"))
|
|
}
|
|
}
|
|
|
|
// TestNodeProposeConfig ensures that node.ProposeConfChange sends the given configuration proposal
|
|
// to the underlying raft.
|
|
func TestNodeProposeConfig(t *testing.T) {
|
|
msgs := []raftpb.Message{}
|
|
appendStep := func(r *raft, m raftpb.Message) {
|
|
msgs = append(msgs, m)
|
|
}
|
|
|
|
n := newNode()
|
|
s := NewMemoryStorage()
|
|
r := newTestRaft(1, []uint64{1}, 10, 1, s)
|
|
go n.run(r)
|
|
n.Campaign(context.TODO())
|
|
for {
|
|
rd := <-n.Ready()
|
|
s.Append(rd.Entries)
|
|
// change the step function to appendStep until this raft becomes leader
|
|
if rd.SoftState.Lead == r.id {
|
|
r.step = appendStep
|
|
n.Advance()
|
|
break
|
|
}
|
|
n.Advance()
|
|
}
|
|
cc := raftpb.ConfChange{Type: raftpb.ConfChangeAddNode, NodeID: 1}
|
|
ccdata, err := cc.Marshal()
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
n.ProposeConfChange(context.TODO(), cc)
|
|
n.Stop()
|
|
|
|
if len(msgs) != 1 {
|
|
t.Fatalf("len(msgs) = %d, want %d", len(msgs), 1)
|
|
}
|
|
if msgs[0].Type != raftpb.MsgProp {
|
|
t.Errorf("msg type = %d, want %d", msgs[0].Type, raftpb.MsgProp)
|
|
}
|
|
if !reflect.DeepEqual(msgs[0].Entries[0].Data, ccdata) {
|
|
t.Errorf("data = %v, want %v", msgs[0].Entries[0].Data, ccdata)
|
|
}
|
|
}
|
|
|
|
// TestBlockProposal ensures that node will block proposal when it does not
|
|
// know who is the current leader; node will accept proposal when it knows
|
|
// who is the current leader.
|
|
func TestBlockProposal(t *testing.T) {
|
|
n := newNode()
|
|
r := newTestRaft(1, []uint64{1}, 10, 1, NewMemoryStorage())
|
|
go n.run(r)
|
|
defer n.Stop()
|
|
|
|
errc := make(chan error, 1)
|
|
go func() {
|
|
errc <- n.Propose(context.TODO(), []byte("somedata"))
|
|
}()
|
|
|
|
testutil.WaitSchedule()
|
|
select {
|
|
case err := <-errc:
|
|
t.Errorf("err = %v, want blocking", err)
|
|
default:
|
|
}
|
|
|
|
n.Campaign(context.TODO())
|
|
testutil.WaitSchedule()
|
|
select {
|
|
case err := <-errc:
|
|
if err != nil {
|
|
t.Errorf("err = %v, want %v", err, nil)
|
|
}
|
|
default:
|
|
t.Errorf("blocking proposal, want unblocking")
|
|
}
|
|
}
|
|
|
|
// TestNodeTick ensures that node.Tick() will increase the
|
|
// elapsed of the underlying raft state machine.
|
|
func TestNodeTick(t *testing.T) {
|
|
n := newNode()
|
|
s := NewMemoryStorage()
|
|
r := newTestRaft(1, []uint64{1}, 10, 1, s)
|
|
go n.run(r)
|
|
elapsed := r.elapsed
|
|
n.Tick()
|
|
n.Stop()
|
|
if r.elapsed != elapsed+1 {
|
|
t.Errorf("elapsed = %d, want %d", r.elapsed, elapsed+1)
|
|
}
|
|
}
|
|
|
|
// TestNodeStop ensures that node.Stop() blocks until the node has stopped
|
|
// processing, and that it is idempotent
|
|
func TestNodeStop(t *testing.T) {
|
|
n := newNode()
|
|
s := NewMemoryStorage()
|
|
r := newTestRaft(1, []uint64{1}, 10, 1, s)
|
|
donec := make(chan struct{})
|
|
|
|
go func() {
|
|
n.run(r)
|
|
close(donec)
|
|
}()
|
|
|
|
elapsed := r.elapsed
|
|
n.Tick()
|
|
n.Stop()
|
|
|
|
select {
|
|
case <-donec:
|
|
case <-time.After(time.Second):
|
|
t.Fatalf("timed out waiting for node to stop!")
|
|
}
|
|
|
|
if r.elapsed != elapsed+1 {
|
|
t.Errorf("elapsed = %d, want %d", r.elapsed, elapsed+1)
|
|
}
|
|
// Further ticks should have no effect, the node is stopped.
|
|
n.Tick()
|
|
if r.elapsed != elapsed+1 {
|
|
t.Errorf("elapsed = %d, want %d", r.elapsed, elapsed+1)
|
|
}
|
|
// Subsequent Stops should have no effect.
|
|
n.Stop()
|
|
}
|
|
|
|
func TestReadyContainUpdates(t *testing.T) {
|
|
tests := []struct {
|
|
rd Ready
|
|
wcontain bool
|
|
}{
|
|
{Ready{}, false},
|
|
{Ready{SoftState: &SoftState{Lead: 1}}, true},
|
|
{Ready{HardState: raftpb.HardState{Vote: 1}}, true},
|
|
{Ready{Entries: make([]raftpb.Entry, 1, 1)}, true},
|
|
{Ready{CommittedEntries: make([]raftpb.Entry, 1, 1)}, true},
|
|
{Ready{Messages: make([]raftpb.Message, 1, 1)}, true},
|
|
{Ready{Snapshot: raftpb.Snapshot{Metadata: raftpb.SnapshotMetadata{Index: 1}}}, true},
|
|
}
|
|
|
|
for i, tt := range tests {
|
|
if g := tt.rd.containsUpdates(); g != tt.wcontain {
|
|
t.Errorf("#%d: containUpdates = %v, want %v", i, g, tt.wcontain)
|
|
}
|
|
}
|
|
}
|
|
|
|
// TestNodeStart ensures that a node can be started correctly. The node should
|
|
// start with correct configuration change entries, and can accept and commit
|
|
// proposals.
|
|
func TestNodeStart(t *testing.T) {
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
|
|
cc := raftpb.ConfChange{Type: raftpb.ConfChangeAddNode, NodeID: 1}
|
|
ccdata, err := cc.Marshal()
|
|
if err != nil {
|
|
t.Fatalf("unexpected marshal error: %v", err)
|
|
}
|
|
wants := []Ready{
|
|
{
|
|
SoftState: &SoftState{Lead: 1, RaftState: StateLeader},
|
|
HardState: raftpb.HardState{Term: 2, Commit: 2, Vote: 1},
|
|
Entries: []raftpb.Entry{
|
|
{Type: raftpb.EntryConfChange, Term: 1, Index: 1, Data: ccdata},
|
|
{Term: 2, Index: 2},
|
|
},
|
|
CommittedEntries: []raftpb.Entry{
|
|
{Type: raftpb.EntryConfChange, Term: 1, Index: 1, Data: ccdata},
|
|
{Term: 2, Index: 2},
|
|
},
|
|
},
|
|
{
|
|
HardState: raftpb.HardState{Term: 2, Commit: 3, Vote: 1},
|
|
Entries: []raftpb.Entry{{Term: 2, Index: 3, Data: []byte("foo")}},
|
|
CommittedEntries: []raftpb.Entry{{Term: 2, Index: 3, Data: []byte("foo")}},
|
|
},
|
|
}
|
|
storage := NewMemoryStorage()
|
|
c := &Config{
|
|
ID: 1,
|
|
ElectionTick: 10,
|
|
HeartbeatTick: 1,
|
|
Storage: storage,
|
|
MaxSizePerMsg: noLimit,
|
|
MaxInflightMsgs: 256,
|
|
}
|
|
n := StartNode(c, []Peer{{ID: 1}})
|
|
n.Campaign(ctx)
|
|
g := <-n.Ready()
|
|
if !reflect.DeepEqual(g, wants[0]) {
|
|
t.Fatalf("#%d: g = %+v,\n w %+v", 1, g, wants[0])
|
|
} else {
|
|
storage.Append(g.Entries)
|
|
n.Advance()
|
|
}
|
|
|
|
n.Propose(ctx, []byte("foo"))
|
|
if g2 := <-n.Ready(); !reflect.DeepEqual(g2, wants[1]) {
|
|
t.Errorf("#%d: g = %+v,\n w %+v", 2, g2, wants[1])
|
|
} else {
|
|
storage.Append(g2.Entries)
|
|
n.Advance()
|
|
}
|
|
|
|
select {
|
|
case rd := <-n.Ready():
|
|
t.Errorf("unexpected Ready: %+v", rd)
|
|
case <-time.After(time.Millisecond):
|
|
}
|
|
}
|
|
|
|
func TestNodeRestart(t *testing.T) {
|
|
entries := []raftpb.Entry{
|
|
{Term: 1, Index: 1},
|
|
{Term: 1, Index: 2, Data: []byte("foo")},
|
|
}
|
|
st := raftpb.HardState{Term: 1, Commit: 1}
|
|
|
|
want := Ready{
|
|
HardState: st,
|
|
// commit up to index commit index in st
|
|
CommittedEntries: entries[:st.Commit],
|
|
}
|
|
|
|
storage := NewMemoryStorage()
|
|
storage.SetHardState(st)
|
|
storage.Append(entries)
|
|
c := &Config{
|
|
ID: 1,
|
|
ElectionTick: 10,
|
|
HeartbeatTick: 1,
|
|
Storage: storage,
|
|
MaxSizePerMsg: noLimit,
|
|
MaxInflightMsgs: 256,
|
|
}
|
|
n := RestartNode(c)
|
|
if g := <-n.Ready(); !reflect.DeepEqual(g, want) {
|
|
t.Errorf("g = %+v,\n w %+v", g, want)
|
|
}
|
|
n.Advance()
|
|
|
|
select {
|
|
case rd := <-n.Ready():
|
|
t.Errorf("unexpected Ready: %+v", rd)
|
|
case <-time.After(time.Millisecond):
|
|
}
|
|
}
|
|
|
|
func TestNodeRestartFromSnapshot(t *testing.T) {
|
|
snap := raftpb.Snapshot{
|
|
Metadata: raftpb.SnapshotMetadata{
|
|
ConfState: raftpb.ConfState{Nodes: []uint64{1, 2}},
|
|
Index: 2,
|
|
Term: 1,
|
|
},
|
|
}
|
|
entries := []raftpb.Entry{
|
|
{Term: 1, Index: 3, Data: []byte("foo")},
|
|
}
|
|
st := raftpb.HardState{Term: 1, Commit: 3}
|
|
|
|
want := Ready{
|
|
HardState: st,
|
|
// commit up to index commit index in st
|
|
CommittedEntries: entries,
|
|
}
|
|
|
|
s := NewMemoryStorage()
|
|
s.SetHardState(st)
|
|
s.ApplySnapshot(snap)
|
|
s.Append(entries)
|
|
c := &Config{
|
|
ID: 1,
|
|
ElectionTick: 10,
|
|
HeartbeatTick: 1,
|
|
Storage: s,
|
|
MaxSizePerMsg: noLimit,
|
|
MaxInflightMsgs: 256,
|
|
}
|
|
n := RestartNode(c)
|
|
if g := <-n.Ready(); !reflect.DeepEqual(g, want) {
|
|
t.Errorf("g = %+v,\n w %+v", g, want)
|
|
} else {
|
|
n.Advance()
|
|
}
|
|
|
|
select {
|
|
case rd := <-n.Ready():
|
|
t.Errorf("unexpected Ready: %+v", rd)
|
|
case <-time.After(time.Millisecond):
|
|
}
|
|
}
|
|
|
|
func TestNodeAdvance(t *testing.T) {
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
|
|
storage := NewMemoryStorage()
|
|
c := &Config{
|
|
ID: 1,
|
|
ElectionTick: 10,
|
|
HeartbeatTick: 1,
|
|
Storage: storage,
|
|
MaxSizePerMsg: noLimit,
|
|
MaxInflightMsgs: 256,
|
|
}
|
|
n := StartNode(c, []Peer{{ID: 1}})
|
|
n.Campaign(ctx)
|
|
<-n.Ready()
|
|
n.Propose(ctx, []byte("foo"))
|
|
var rd Ready
|
|
select {
|
|
case rd = <-n.Ready():
|
|
t.Fatalf("unexpected Ready before Advance: %+v", rd)
|
|
case <-time.After(time.Millisecond):
|
|
}
|
|
storage.Append(rd.Entries)
|
|
n.Advance()
|
|
select {
|
|
case <-n.Ready():
|
|
case <-time.After(time.Millisecond):
|
|
t.Errorf("expect Ready after Advance, but there is no Ready available")
|
|
}
|
|
}
|
|
|
|
func TestSoftStateEqual(t *testing.T) {
|
|
tests := []struct {
|
|
st *SoftState
|
|
we bool
|
|
}{
|
|
{&SoftState{}, true},
|
|
{&SoftState{Lead: 1}, false},
|
|
{&SoftState{RaftState: StateLeader}, false},
|
|
}
|
|
for i, tt := range tests {
|
|
if g := tt.st.equal(&SoftState{}); g != tt.we {
|
|
t.Errorf("#%d, equal = %v, want %v", i, g, tt.we)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestIsHardStateEqual(t *testing.T) {
|
|
tests := []struct {
|
|
st raftpb.HardState
|
|
we bool
|
|
}{
|
|
{emptyState, true},
|
|
{raftpb.HardState{Vote: 1}, false},
|
|
{raftpb.HardState{Commit: 1}, false},
|
|
{raftpb.HardState{Term: 1}, false},
|
|
}
|
|
|
|
for i, tt := range tests {
|
|
if isHardStateEqual(tt.st, emptyState) != tt.we {
|
|
t.Errorf("#%d, equal = %v, want %v", i, isHardStateEqual(tt.st, emptyState), tt.we)
|
|
}
|
|
}
|
|
}
|