From 0ee872def4c73e870977801fa8d2884bca70c8db Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Thu, 28 May 2015 16:19:05 +0100 Subject: [PATCH 1/2] Lock round mockGossipComms to avoid race conditions --- ipam/testutils_test.go | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/ipam/testutils_test.go b/ipam/testutils_test.go index ee72d8af..cc86abca 100644 --- a/ipam/testutils_test.go +++ b/ipam/testutils_test.go @@ -3,6 +3,7 @@ package ipam import ( "fmt" "math/rand" + "sync" "testing" "time" @@ -30,12 +31,15 @@ func toStringArray(messages []mockMessage) []string { } type mockGossipComms struct { + sync.RWMutex t *testing.T name string messages []mockMessage } func (m *mockGossipComms) String() string { + m.RLock() + defer m.RUnlock() return fmt.Sprintf("[mockGossipComms %s]", m.name) } @@ -44,6 +48,8 @@ func (m *mockGossipComms) String() string { // requires they are not based off iterating through a map. func (m *mockGossipComms) GossipBroadcast(update router.GossipData) error { + m.Lock() + defer m.Unlock() buf := []byte{} if len(m.messages) == 0 { m.Fatalf("%s: Gossip broadcast message unexpected: \n%x", m.name, buf) @@ -76,6 +82,8 @@ func (m *mockGossipComms) Fatalf(format string, args ...interface{}) { } func (m *mockGossipComms) GossipUnicast(dstPeerName router.PeerName, buf []byte) error { + m.Lock() + defer m.Unlock() if len(m.messages) == 0 { m.Fatalf("%s: Gossip message to %s unexpected: \n%s", m.name, dstPeerName, buf) } else if msg := m.messages[0]; msg.dst == router.UnknownPeerName { @@ -96,20 +104,26 @@ func (m *mockGossipComms) GossipUnicast(dstPeerName router.PeerName, buf []byte) func ExpectMessage(alloc *Allocator, dst string, msgType byte, buf []byte) { m := alloc.gossip.(*mockGossipComms) dstPeerName, _ := router.PeerNameFromString(dst) + m.Lock() m.messages = append(m.messages, mockMessage{dstPeerName, msgType, buf}) + m.Unlock() } func ExpectBroadcastMessage(alloc *Allocator, buf []byte) { m := alloc.gossip.(*mockGossipComms) + m.Lock() m.messages = append(m.messages, mockMessage{router.UnknownPeerName, 0, buf}) + m.Unlock() } func CheckAllExpectedMessagesSent(allocs ...*Allocator) { for _, alloc := range allocs { m := alloc.gossip.(*mockGossipComms) + m.RLock() if len(m.messages) > 0 { wt.Fatalf(m.t, "%s: Gossip message(s) not sent as expected: \n%x", m.name, m.messages) } + m.RUnlock() } } From adeb0d3934c617c63d408387b775fd9e6c48f66f Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Thu, 28 May 2015 16:21:57 +0100 Subject: [PATCH 2/2] Don't bypass the actor mechanism in unit tests, to avoid race conditions --- ipam/allocator_test.go | 24 ++++++++++++------------ ipam/testutils_test.go | 16 +++++++++++++--- 2 files changed, 25 insertions(+), 15 deletions(-) diff --git a/ipam/allocator_test.go b/ipam/allocator_test.go index b3c550c8..51c825c7 100644 --- a/ipam/allocator_test.go +++ b/ipam/allocator_test.go @@ -54,7 +54,7 @@ func TestAllocFree(t *testing.T) { alloc.ContainerDied(container2) alloc.ContainerDied(container3) alloc.String() // force sync-up after async call - wt.AssertEquals(t, alloc.space.NumFreeAddresses(), address.Offset(spaceSize)) + wt.AssertEquals(t, alloc.NumFreeAddresses(), address.Offset(spaceSize)) } func TestBootstrap(t *testing.T) { @@ -73,9 +73,9 @@ func TestBootstrap(t *testing.T) { alloc2 := makeAllocatorWithMockGossip(t, peerNameString, testStart1+"/22", 2) defer alloc2.Stop() - alloc1.OnGossipBroadcast(alloc2.encode()) + alloc1.OnGossipBroadcast(alloc2.Encode()) - alloc1.tryPendingOps() + alloc1.actionChan <- func() { alloc1.tryPendingOps() } ExpectBroadcastMessage(alloc1, nil) // alloc1 will try to form consensus done := make(chan bool) @@ -88,26 +88,26 @@ func TestBootstrap(t *testing.T) { CheckAllExpectedMessagesSent(alloc1, alloc2) - alloc1.tryPendingOps() + alloc1.actionChan <- func() { alloc1.tryPendingOps() } AssertNothingSent(t, done) CheckAllExpectedMessagesSent(alloc1, alloc2) // alloc2 receives paxos update and broadcasts its reply ExpectBroadcastMessage(alloc2, nil) - alloc2.OnGossipBroadcast(alloc1.encode()) + alloc2.OnGossipBroadcast(alloc1.Encode()) ExpectBroadcastMessage(alloc1, nil) - alloc1.OnGossipBroadcast(alloc2.encode()) + alloc1.OnGossipBroadcast(alloc2.Encode()) // both nodes will get consensus now so initialize the ring ExpectBroadcastMessage(alloc2, nil) ExpectBroadcastMessage(alloc2, nil) - alloc2.OnGossipBroadcast(alloc1.encode()) + alloc2.OnGossipBroadcast(alloc1.Encode()) CheckAllExpectedMessagesSent(alloc1, alloc2) - alloc1.OnGossipBroadcast(alloc2.encode()) + alloc1.OnGossipBroadcast(alloc2.Encode()) // now alloc1 should have space AssertSent(t, done) @@ -172,7 +172,7 @@ func TestCancel(t *testing.T) { alloc2.Start() // tell peers about each other - alloc1.OnGossipBroadcast(alloc2.encode()) + alloc1.OnGossipBroadcast(alloc2.Encode()) // Get some IPs, so each allocator has some space res1, _ := alloc1.Allocate("foo", nil) @@ -188,7 +188,7 @@ func TestCancel(t *testing.T) { unpause := alloc2.pause() // Use up all the IPs that alloc1 owns, so the allocation after this will prompt a request to alloc2 - for i := 0; alloc1.space.NumFreeAddresses() > 0; i++ { + for i := 0; alloc1.NumFreeAddresses() > 0; i++ { alloc1.Allocate(fmt.Sprintf("tmp%d", i), nil) } cancelChan := make(chan bool, 1) @@ -256,7 +256,7 @@ func TestTransfer(t *testing.T) { wt.AssertSuccess(t, alloc1.AdminTakeoverRanges(alloc2.ourName.String())) wt.AssertSuccess(t, alloc1.AdminTakeoverRanges(alloc3.ourName.String())) - wt.AssertEquals(t, alloc1.space.NumFreeAddresses(), address.Offset(1022)) + wt.AssertEquals(t, alloc1.NumFreeAddresses(), address.Offset(1022)) _, err = alloc1.Allocate("foo", nil) wt.AssertTrue(t, err == nil, "Failed to get address") @@ -447,7 +447,7 @@ func TestGossipSkew(t *testing.T) { alloc2.now = func() time.Time { return time.Now().Add(time.Hour * 2) } defer alloc2.Stop() - if _, err := alloc1.OnGossipBroadcast(alloc2.encode()); err == nil { + if _, err := alloc1.OnGossipBroadcast(alloc2.Encode()); err == nil { t.Fail() } } diff --git a/ipam/testutils_test.go b/ipam/testutils_test.go index cc86abca..c47273cb 100644 --- a/ipam/testutils_test.go +++ b/ipam/testutils_test.go @@ -8,6 +8,7 @@ import ( "time" "github.com/weaveworks/weave/common" + "github.com/weaveworks/weave/ipam/address" "github.com/weaveworks/weave/router" wt "github.com/weaveworks/weave/testing" ) @@ -159,6 +160,14 @@ func (alloc *Allocator) claimRingForTesting(allocs ...*Allocator) { alloc.space.AddRanges(alloc.ring.OwnedRanges()) } +func (alloc *Allocator) NumFreeAddresses() address.Offset { + resultChan := make(chan address.Offset) + alloc.actionChan <- func() { + resultChan <- alloc.space.NumFreeAddresses() + } + return <-resultChan +} + // Check whether or not something was sent on a channel func AssertSent(t *testing.T, ch <-chan bool) { timeout := time.After(10 * time.Second) @@ -192,7 +201,8 @@ func AssertNothingSentErr(t *testing.T, ch <-chan error) { type gossipMessage struct { isUnicast bool sender *router.PeerName - buf []byte + buf []byte // for unicast + data router.GossipData // for broadcast exitChan chan bool } @@ -204,7 +214,7 @@ type TestGossipRouter struct { func (grouter *TestGossipRouter) GossipBroadcast(update router.GossipData) error { for _, gossipChan := range grouter.gossipChans { select { - case gossipChan <- gossipMessage{buf: update.(*ipamGossipData).alloc.encode()}: + case gossipChan <- gossipMessage{data: update}: default: // drop the message if we cannot send it } } @@ -247,7 +257,7 @@ func (grouter *TestGossipRouter) connect(sender router.PeerName, gossiper router panic(fmt.Sprintf("Error doing gossip unicast to %s: %s", sender, err)) } } else { - _, err := gossiper.OnGossipBroadcast(message.buf) + _, err := gossiper.OnGossipBroadcast(message.data.Encode()) if err != nil { panic(fmt.Sprintf("Error doing gossip broadcast to %s: %s", sender, err)) }