rafttest: add network delay
parent
c94db98177
commit
65cd0051fe
|
@ -33,6 +33,7 @@ type raftNetwork struct {
|
|||
mu sync.Mutex
|
||||
disconnected map[uint64]bool
|
||||
dropmap map[conn]float64
|
||||
delaymap map[conn]delay
|
||||
recvQueues map[uint64]chan raftpb.Message
|
||||
}
|
||||
|
||||
|
@ -40,10 +41,16 @@ type conn struct {
|
|||
from, to uint64
|
||||
}
|
||||
|
||||
type delay struct {
|
||||
d time.Duration
|
||||
rate float64
|
||||
}
|
||||
|
||||
func newRaftNetwork(nodes ...uint64) *raftNetwork {
|
||||
pn := &raftNetwork{
|
||||
recvQueues: make(map[uint64]chan raftpb.Message),
|
||||
dropmap: make(map[conn]float64),
|
||||
delaymap: make(map[conn]delay),
|
||||
disconnected: make(map[uint64]bool),
|
||||
}
|
||||
|
||||
|
@ -64,6 +71,7 @@ func (rn *raftNetwork) send(m raftpb.Message) {
|
|||
to = nil
|
||||
}
|
||||
drop := rn.dropmap[conn{m.From, m.To}]
|
||||
delay := rn.delaymap[conn{m.From, m.To}]
|
||||
rn.mu.Unlock()
|
||||
|
||||
if to == nil {
|
||||
|
@ -72,6 +80,11 @@ func (rn *raftNetwork) send(m raftpb.Message) {
|
|||
if drop != 0 && rand.Float64() < drop {
|
||||
return
|
||||
}
|
||||
// TODO: shall we delay without blocking the send call?
|
||||
if delay.d != 0 && rand.Float64() < delay.rate {
|
||||
rd := rand.Int63n(int64(delay.d))
|
||||
time.Sleep(time.Duration(rd))
|
||||
}
|
||||
|
||||
to <- m
|
||||
}
|
||||
|
@ -94,13 +107,16 @@ func (rn *raftNetwork) drop(from, to uint64, rate float64) {
|
|||
}
|
||||
|
||||
func (rn *raftNetwork) delay(from, to uint64, d time.Duration, rate float64) {
|
||||
panic("unimplemented")
|
||||
rn.mu.Lock()
|
||||
defer rn.mu.Unlock()
|
||||
rn.delaymap[conn{from, to}] = delay{d, rate}
|
||||
}
|
||||
|
||||
func (rn *raftNetwork) heal() {
|
||||
rn.mu.Lock()
|
||||
defer rn.mu.Unlock()
|
||||
rn.dropmap = make(map[conn]float64)
|
||||
rn.delaymap = make(map[conn]delay)
|
||||
}
|
||||
|
||||
func (rn *raftNetwork) disconnect(id uint64) {
|
||||
|
|
|
@ -2,6 +2,7 @@ package rafttest
|
|||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/coreos/etcd/raft/raftpb"
|
||||
)
|
||||
|
@ -34,3 +35,24 @@ func TestNetworkDrop(t *testing.T) {
|
|||
t.Errorf("drop = %d, want around %d", drop, droprate*float64(sent))
|
||||
}
|
||||
}
|
||||
|
||||
func TestNetworkDelay(t *testing.T) {
|
||||
sent := 1000
|
||||
delay := time.Millisecond
|
||||
delayrate := 0.1
|
||||
nt := newRaftNetwork(1, 2)
|
||||
|
||||
nt.delay(1, 2, delay, delayrate)
|
||||
var total time.Duration
|
||||
for i := 0; i < sent; i++ {
|
||||
s := time.Now()
|
||||
nt.send(raftpb.Message{From: 1, To: 2})
|
||||
total += time.Since(s)
|
||||
}
|
||||
|
||||
w := time.Duration(float64(sent)*delayrate/2) * delay
|
||||
// there are pretty overhead in the send call, since it genarete random numbers.
|
||||
if total < w+10*delay {
|
||||
t.Errorf("total = %v, want > %v", total, w)
|
||||
}
|
||||
}
|
||||
|
|
Loading…
Reference in New Issue