Skip to content

Commit

Permalink
Gossip Comm tests tweaks
Browse files Browse the repository at this point in the history 
This commit affects gossip comm tests:
0) Makes them run in parallel
1) Reduces gossip comm tests from 32s to 13s:
   Tweaked TestBlackListPKIid, TestAccept and TestBasic
2) Contains some refactoring work of gossip comm tests
3) Re-writes TestBlackListPKIid from concurrent to serial
   so it won't fail on occasions

But it also contains a tiny change to the flow of createConnection()
in gossip/comm/comm_impl.go that only affected wrong logging of messages:
If the remote peer was black-listed, it would have logged that
the remote peer has sent a nil pkiID, although it hadn't.

Change-Id: Iceecbe7b8fe44915c59ff8b28d22ba1853ca4e8a
Signed-off-by: Yacov Manevich <yacovm@il.ibm.com>
  • Loading branch information
@yacovm
yacovm committed Nov 20, 2016
1 parent 7129a55 commit 9d3abd1
Show file tree
Hide file tree
Showing 2 changed files with 102 additions and 74 deletions.
12 changes: 6 additions & 6 deletions gossip/comm/comm_impl.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -112,7 +112,7 @@ func NewCommInstanceWithServer(port int, sec SecurityProvider, pkID common.PKIid

// NewCommInstance creates a new comm instance that binds itself to the given gRPC server
func NewCommInstance(s *grpc.Server, sec SecurityProvider, PKIID common.PKIidType, dialOpts ...grpc.DialOption) (Comm, error) {
commInst, err := NewCommInstanceWithServer(-1, sec, PKIID, dialOpts ...)
commInst, err := NewCommInstanceWithServer(-1, sec, PKIID, dialOpts...)
if err != nil {
return nil, err
}
Expand Down Expand Up @@ -162,12 +162,12 @@ func (c *commImpl) createConnection(endpoint string, expectedPKIID common.PKIidT

if stream, err := cl.GossipStream(context.Background()); err == nil {
pkiID, err := c.authenticateRemotePeer(stream)
if expectedPKIID != nil && !bytes.Equal(pkiID, expectedPKIID) {
// PKIID is nil when we don't know the remote PKI id's
c.logger.Warning("Remote endpoint claims to be a different peer, expected", expectedPKIID, "but got", pkiID)
return nil, fmt.Errorf("Authentication failure")
}
if err == nil {
if expectedPKIID != nil && !bytes.Equal(pkiID, expectedPKIID) {
// PKIID is nil when we don't know the remote PKI id's
c.logger.Warning("Remote endpoint claims to be a different peer, expected", expectedPKIID, "but got", pkiID)
return nil, fmt.Errorf("Authentication failure")
}
conn := newConnection(cl, cc, stream, nil)
conn.pkiID = pkiID
conn.logger = c.logger
Expand Down
164 changes: 96 additions & 68 deletions gossip/comm/comm_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -165,28 +165,26 @@ func TestHandshake(t *testing.T) {
}

func TestBasic(t *testing.T) {
t.Parallel()
comm1, _ := newCommInstance(2000, naiveSec)
comm2, _ := newCommInstance(3000, naiveSec)
defer comm1.Stop()
defer comm2.Stop()
time.Sleep(time.Duration(3) * time.Second)
msgs := make(chan *proto.GossipMessage, 2)
go func() {
m := <-comm2.Accept(acceptAll)
msgs <- m.GetGossipMessage()
}()
go func() {
m := <-comm1.Accept(acceptAll)
msgs <- m.GetGossipMessage()
}()
comm1.Send(createGossipMsg(), &RemotePeer{PKIID: []byte("localhost:3000"), Endpoint: "localhost:3000"})
time.Sleep(time.Second)
comm2.Send(createGossipMsg(), &RemotePeer{PKIID: []byte("localhost:2000"), Endpoint: "localhost:2000"})
m1 := comm1.Accept(acceptAll)
m2 := comm2.Accept(acceptAll)
out := make(chan uint64, 2)
reader := func(ch <-chan ReceivedMessage) {
m := <- ch
out <- m.GetGossipMessage().Nonce
}
go reader(m1)
go reader(m2)
comm1.Send(createGossipMsg(), remotePeer(3000))
time.Sleep(time.Second)
assert.Equal(t, 2, len(msgs))
comm2.Send(createGossipMsg(), remotePeer(2000))
waitForMessages(t, out, 2, "Didn't receive 2 messages")
}

func TestBlackListPKIid(t *testing.T) {
t.Parallel()
comm1, _ := newCommInstance(1611, naiveSec)
comm2, _ := newCommInstance(1612, naiveSec)
comm3, _ := newCommInstance(1613, naiveSec)
Expand All @@ -196,7 +194,7 @@ func TestBlackListPKIid(t *testing.T) {
defer comm3.Stop()
defer comm4.Stop()

reader := func(out chan uint64, in <-chan ReceivedMessage) {
reader := func(instance string, out chan uint64, in <-chan ReceivedMessage) {
for {
msg := <-in
if msg == nil {
Expand All @@ -206,52 +204,48 @@ func TestBlackListPKIid(t *testing.T) {
}
}

sender := func(comm Comm, port int, n int) {
endpoint := fmt.Sprintf("localhost:%d", port)
for i := 0; i < n; i++ {
comm.Send(createGossipMsg(), &RemotePeer{Endpoint: endpoint, PKIID: []byte(endpoint)})
time.Sleep(time.Duration(1) * time.Second)
}
}
out1 := make(chan uint64, 4)
out2 := make(chan uint64, 4)
out3 := make(chan uint64, 4)
out4 := make(chan uint64, 4)

out1 := make(chan uint64, 5)
out2 := make(chan uint64, 5)
out3 := make(chan uint64, 10)
out4 := make(chan uint64, 10)

go reader(out1, comm1.Accept(acceptAll))
go reader(out2, comm2.Accept(acceptAll))
go reader(out3, comm3.Accept(acceptAll))
go reader(out4, comm4.Accept(acceptAll))
go reader("comm1", out1, comm1.Accept(acceptAll))
go reader("comm2", out2, comm2.Accept(acceptAll))
go reader("comm3", out3, comm3.Accept(acceptAll))
go reader("comm4", out4, comm4.Accept(acceptAll))

// have comm1 BL comm3
comm1.BlackListPKIid([]byte("localhost:1613"))

// make comm3 send to 1 and 2
go sender(comm3, 1611, 5)
go sender(comm3, 1612, 5)
comm3.Send(createGossipMsg(), remotePeer(1612)) // out2++
comm3.Send(createGossipMsg(), remotePeer(1611))

// make comm1 and comm2 send to comm3
go sender(comm1, 1613, 5)
go sender(comm2, 1613, 5)
waitForMessages(t, out2, 1, "comm2 should have received 1 message")

// make comm1 and comm2 send to comm4 which is not blacklisted
go sender(comm1, 1614, 5)
go sender(comm2, 1614, 5)
// make comm1 and comm2 send to comm3
comm1.Send(createGossipMsg(), remotePeer(1613))
comm2.Send(createGossipMsg(), remotePeer(1613)) // out3++
waitForMessages(t, out3, 1, "comm3 should have received 1 message")

time.Sleep(time.Duration(1) * time.Second)
// make comm1 and comm2 send to comm4 which is not blacklisted // out4 += 4
comm1.Send(createGossipMsg(), remotePeer(1614))
comm2.Send(createGossipMsg(), remotePeer(1614))
comm1.Send(createGossipMsg(), remotePeer(1614))
comm2.Send(createGossipMsg(), remotePeer(1614))

// blacklist comm3 mid-sending
// blacklist comm3 by comm2
comm2.BlackListPKIid([]byte("localhost:1613"))
time.Sleep(time.Duration(5) * time.Second)

assert.Equal(t, 0, len(out1), "Comm instance 1 received messages(%d) from comm3 although comm3 is black listed", len(out1))
assert.True(t, len(out2) < 2, "Comm instance 2 received too many messages(%d) from comm3 although comm3 is black listed", len(out2))
assert.True(t, len(out3) < 3, "Comm instance 3 received too many messages(%d) although black listed", len(out3))
assert.Equal(t, 10, len(out4), "Comm instance 4 didn't receive all messages sent to it")
// send from comm1 and comm2 to comm3 again
comm1.Send(createGossipMsg(), remotePeer(1613)) // shouldn't have an effect
comm2.Send(createGossipMsg(), remotePeer(1613)) // shouldn't have an effect

waitForMessages(t, out4, 4, "comm1 should have received 4 messages")
}

func TestParallelSend(t *testing.T) {
t.Parallel()
comm1, _ := newCommInstance(5611, naiveSec)
comm2, _ := newCommInstance(5612, naiveSec)
defer comm1.Stop()
Expand All @@ -266,7 +260,7 @@ func TestParallelSend(t *testing.T) {
emptyMsg := createGossipMsg()
go func() {
defer wg.Done()
comm1.Send(emptyMsg, &RemotePeer{Endpoint: "localhost:5612", PKIID: []byte("localhost:5612")})
comm1.Send(emptyMsg, remotePeer(5612))
}()
}
wg.Wait()
Expand All @@ -293,6 +287,7 @@ func TestParallelSend(t *testing.T) {
}

func TestResponses(t *testing.T) {
t.Parallel()
comm1, _ := newCommInstance(8611, naiveSec)
comm2, _ := newCommInstance(8612, naiveSec)

Expand All @@ -316,7 +311,7 @@ func TestResponses(t *testing.T) {
responsesFromComm1 := comm2.Accept(acceptAll)

ticker := time.NewTicker(time.Duration(6000) * time.Millisecond)
comm2.Send(msg, &RemotePeer{PKIID: []byte("localhost:8611"), Endpoint: "localhost:8611"})
comm2.Send(msg, remotePeer(8611))
time.Sleep(time.Duration(100) * time.Millisecond)

select {
Expand All @@ -331,6 +326,7 @@ func TestResponses(t *testing.T) {
}

func TestAccept(t *testing.T) {
t.Parallel()
comm1, _ := newCommInstance(7611, naiveSec)
comm2, _ := newCommInstance(7612, naiveSec)

Expand All @@ -348,11 +344,13 @@ func TestAccept(t *testing.T) {
var evenResults []uint64
var oddResults []uint64

out := make(chan uint64, defRecvBuffSize)
sem := make(chan struct{}, 0)

readIntoSlice := func(a *[]uint64, ch <-chan ReceivedMessage) {
for m := range ch {
*a = append(*a, m.GetGossipMessage().Nonce)
out <- m.GetGossipMessage().Nonce
}
sem <- struct{}{}
}
Expand All @@ -361,10 +359,10 @@ func TestAccept(t *testing.T) {
go readIntoSlice(&oddResults, oddNONCES)

for i := 0; i < defRecvBuffSize; i++ {
comm2.Send(createGossipMsg(), &RemotePeer{Endpoint: "localhost:7611", PKIID: []byte("localhost:7611")})
comm2.Send(createGossipMsg(), remotePeer(7611))
}

time.Sleep(time.Duration(5) * time.Second)
waitForMessages(t, out, defRecvBuffSize, "Didn't receive all messages sent")

comm1.Stop()
comm2.Stop()
Expand All @@ -386,6 +384,7 @@ func TestAccept(t *testing.T) {
}

func TestReConnections(t *testing.T) {
t.Parallel()
comm1, _ := newCommInstance(3611, naiveSec)
comm2, _ := newCommInstance(3612, naiveSec)

Expand All @@ -406,10 +405,10 @@ func TestReConnections(t *testing.T) {
go reader(out2, comm2.Accept(acceptAll))

// comm1 connects to comm2
comm1.Send(createGossipMsg(), &RemotePeer{Endpoint: "localhost:3612", PKIID: []byte("localhost:3612")})
comm1.Send(createGossipMsg(), remotePeer(3612))
time.Sleep(100 * time.Millisecond)
// comm2 sends to comm1
comm2.Send(createGossipMsg(), &RemotePeer{Endpoint: "localhost:3611", PKIID: []byte("localhost:3611")})
comm2.Send(createGossipMsg(), remotePeer(3611))
time.Sleep(100 * time.Millisecond)

assert.Equal(t, 1, len(out2))
Expand All @@ -419,39 +418,43 @@ func TestReConnections(t *testing.T) {
comm1, _ = newCommInstance(3611, naiveSec)
go reader(out1, comm1.Accept(acceptAll))
time.Sleep(300 * time.Millisecond)
comm2.Send(createGossipMsg(), &RemotePeer{Endpoint: "localhost:3611", PKIID: []byte("localhost:3611")})
comm2.Send(createGossipMsg(), remotePeer(3611))
time.Sleep(100 * time.Millisecond)
assert.Equal(t, 2, len(out1))
}

func TestProbe(t *testing.T) {
t.Parallel()
comm1, _ := newCommInstance(6611, naiveSec)
defer comm1.Stop()
comm2, _ := newCommInstance(6612, naiveSec)
time.Sleep(time.Duration(1) * time.Second)
assert.NoError(t, comm1.Probe(&RemotePeer{Endpoint: "localhost:6612", PKIID: []byte("localhost:6612")}))
assert.Error(t, comm1.Probe(&RemotePeer{Endpoint: "localhost:9012", PKIID: []byte("localhost:9012")}))
assert.NoError(t, comm1.Probe(remotePeer(6612)))
assert.Error(t, comm1.Probe(remotePeer(9012)))
comm2.Stop()
time.Sleep(time.Second)
assert.Error(t, comm1.Probe(&RemotePeer{Endpoint: "localhost:6612", PKIID: []byte("localhost:6612")}))
assert.Error(t, comm1.Probe(remotePeer(6612)))
comm2, _ = newCommInstance(6612, naiveSec)
defer comm2.Stop()
time.Sleep(time.Duration(1) * time.Second)
assert.NoError(t, comm2.Probe(&RemotePeer{Endpoint: "localhost:6611", PKIID: []byte("localhost:6611")}))
assert.NoError(t, comm1.Probe(&RemotePeer{Endpoint: "localhost:6612", PKIID: []byte("localhost:6612")}))
assert.NoError(t, comm2.Probe(remotePeer(6611)))
assert.NoError(t, comm1.Probe(remotePeer(6612)))
}

func TestPresumedDead(t *testing.T) {
comm1, _ := newCommInstance(7611, naiveSec)
defer comm1.Stop()
comm2, _ := newCommInstance(7612, naiveSec)
go comm1.Send(createGossipMsg(), &RemotePeer{PKIID: []byte("localhost:7612"), Endpoint: "localhost:7612"})
t.Parallel()
comm1, _ := newCommInstance(4611, naiveSec)
comm2, _ := newCommInstance(4612, naiveSec)
go comm1.Send(createGossipMsg(), remotePeer(4612))
<-comm2.Accept(acceptAll)
comm2.Stop()
for i := 0; i < 5; i++ {
comm1.Send(createGossipMsg(), &RemotePeer{Endpoint: "localhost:7612", PKIID: []byte("localhost:7612")})
time.Sleep(time.Second)
}
go func() {
for i := 0; i < 5; i++ {
comm1.Send(createGossipMsg(), remotePeer(4612))
time.Sleep(time.Millisecond * 200)
}
}()

ticker := time.NewTicker(time.Second * time.Duration(3))
select {
case <-ticker.C:
Expand All @@ -472,3 +475,28 @@ func createGossipMsg() *proto.GossipMessage {
},
}
}

func remotePeer(port int) *RemotePeer {
endpoint := fmt.Sprintf("localhost:%d", port)
return &RemotePeer{Endpoint: endpoint, PKIID: []byte(endpoint)}
}

func waitForMessages(t *testing.T, msgChan chan uint64, count int, errMsg string) {
c := 0
waiting := true
ticker := time.NewTicker(time.Duration(5) * time.Second)
for waiting {
select {
case <-msgChan:
c++
if c == count {
waiting = false
}
break
case <-ticker.C:
waiting = false
break
}
}
assert.Equal(t, count, c, errMsg)
}

0 comments on commit 9d3abd1

Please sign in to comment.