From acf13dc157993487d5f164e36fcf10ffa8cd838c Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Wed, 15 Jul 2015 12:07:56 +0100 Subject: [PATCH 1/3] Add sender to OnGossipBroadcast() --- ipam/allocator.go | 4 ++-- ipam/allocator_test.go | 14 +++++++------- nameserver/nameserver.go | 2 +- router/gossip.go | 2 +- router/gossip_channel.go | 2 +- router/gossip_test.go | 2 +- router/router.go | 2 +- router/surrogate_gossiper.go | 2 +- testing/gossip/mocks.go | 5 +++-- 9 files changed, 18 insertions(+), 17 deletions(-) diff --git a/ipam/allocator.go b/ipam/allocator.go index c4c4913d..ff0c7380 100644 --- a/ipam/allocator.go +++ b/ipam/allocator.go @@ -366,8 +366,8 @@ func (alloc *Allocator) OnGossipUnicast(sender router.PeerName, msg []byte) erro } // OnGossipBroadcast (Sync) -func (alloc *Allocator) OnGossipBroadcast(msg []byte) (router.GossipData, error) { - alloc.debugln("OnGossipBroadcast:", len(msg), "bytes") +func (alloc *Allocator) OnGossipBroadcast(sender router.PeerName, msg []byte) (router.GossipData, error) { + alloc.debugln("OnGossipBroadcast from", sender, ":", len(msg), "bytes") resultChan := make(chan error) alloc.actionChan <- func() { resultChan <- alloc.update(msg) diff --git a/ipam/allocator_test.go b/ipam/allocator_test.go index a2778248..67d162b8 100644 --- a/ipam/allocator_test.go +++ b/ipam/allocator_test.go @@ -86,7 +86,7 @@ func TestBootstrap(t *testing.T) { alloc2, _ := makeAllocatorWithMockGossip(t, peerNameString, testStart1+"/22", 2) defer alloc2.Stop() - alloc1.OnGossipBroadcast(alloc2.Encode()) + alloc1.OnGossipBroadcast(alloc2.ourName, alloc2.Encode()) alloc1.actionChan <- func() { alloc1.tryPendingOps() } @@ -108,19 +108,19 @@ func TestBootstrap(t *testing.T) { // alloc2 receives paxos update and broadcasts its reply ExpectBroadcastMessage(alloc2, nil) - alloc2.OnGossipBroadcast(alloc1.Encode()) + alloc2.OnGossipBroadcast(alloc1.ourName, alloc1.Encode()) ExpectBroadcastMessage(alloc1, nil) - alloc1.OnGossipBroadcast(alloc2.Encode()) + alloc1.OnGossipBroadcast(alloc2.ourName, 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.ourName, alloc1.Encode()) CheckAllExpectedMessagesSent(alloc1, alloc2) - alloc1.OnGossipBroadcast(alloc2.Encode()) + alloc1.OnGossipBroadcast(alloc2.ourName, alloc2.Encode()) // now alloc1 should have space AssertSent(t, done) @@ -181,7 +181,7 @@ func TestCancel(t *testing.T) { alloc2.Start() // tell peers about each other - alloc1.OnGossipBroadcast(alloc2.Encode()) + alloc1.OnGossipBroadcast(alloc2.ourName, alloc2.Encode()) // Get some IPs, so each allocator has some space res1, _ := alloc1.Allocate("foo", subnet, nil) @@ -476,7 +476,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.ourName, alloc2.Encode()); err == nil { t.Fail() } } diff --git a/nameserver/nameserver.go b/nameserver/nameserver.go index 43562ae5..e6e713c1 100644 --- a/nameserver/nameserver.go +++ b/nameserver/nameserver.go @@ -238,7 +238,7 @@ func (n *Nameserver) OnGossip(msg []byte) (router.GossipData, error) { // merge received data into state and return a representation of // the received data, for further propagation -func (n *Nameserver) OnGossipBroadcast(msg []byte) (router.GossipData, error) { +func (n *Nameserver) OnGossipBroadcast(_ router.PeerName, msg []byte) (router.GossipData, error) { _, entries, err := n.receiveGossip(msg) return entries, err } diff --git a/router/gossip.go b/router/gossip.go index a3af241b..45db0e83 100644 --- a/router/gossip.go +++ b/router/gossip.go @@ -21,7 +21,7 @@ type Gossiper interface { OnGossipUnicast(sender PeerName, msg []byte) error // merge received data into state and return a representation of // the received data, for further propagation - OnGossipBroadcast(update []byte) (GossipData, error) + OnGossipBroadcast(sender PeerName, update []byte) (GossipData, error) // return state of everything we know; gets called periodically Gossip() GossipData // merge received data into state and return "everything new I've diff --git a/router/gossip_channel.go b/router/gossip_channel.go index 795573f8..bb70bc20 100644 --- a/router/gossip_channel.go +++ b/router/gossip_channel.go @@ -79,7 +79,7 @@ func (c *GossipChannel) deliverBroadcast(srcName PeerName, _ []byte, dec *gob.De if err := dec.Decode(&payload); err != nil { return err } - data, err := c.gossiper.OnGossipBroadcast(payload) + data, err := c.gossiper.OnGossipBroadcast(srcName, payload) if err != nil || data == nil { return err } diff --git a/router/gossip_test.go b/router/gossip_test.go index b5d18f93..fce0391a 100644 --- a/router/gossip_test.go +++ b/router/gossip_test.go @@ -212,7 +212,7 @@ func (g *testGossiper) OnGossipUnicast(sender PeerName, msg []byte) error { return nil } -func (g *testGossiper) OnGossipBroadcast(update []byte) (GossipData, error) { +func (g *testGossiper) OnGossipBroadcast(_ PeerName, update []byte) (GossipData, error) { g.Lock() defer g.Unlock() for _, v := range update { diff --git a/router/router.go b/router/router.go index 8482ba5a..16f05dd6 100644 --- a/router/router.go +++ b/router/router.go @@ -389,7 +389,7 @@ func (router *Router) OnGossipUnicast(sender PeerName, msg []byte) error { return fmt.Errorf("unexpected topology gossip unicast: %v", msg) } -func (router *Router) OnGossipBroadcast(update []byte) (GossipData, error) { +func (router *Router) OnGossipBroadcast(_ PeerName, update []byte) (GossipData, error) { origUpdate, _, err := router.applyTopologyUpdate(update) if err != nil || len(origUpdate) == 0 { return nil, err diff --git a/router/surrogate_gossiper.go b/router/surrogate_gossiper.go index ea7f443d..04059eb7 100644 --- a/router/surrogate_gossiper.go +++ b/router/surrogate_gossiper.go @@ -24,7 +24,7 @@ func (*SurrogateGossiper) OnGossipUnicast(sender PeerName, msg []byte) error { return nil } -func (*SurrogateGossiper) OnGossipBroadcast(update []byte) (GossipData, error) { +func (*SurrogateGossiper) OnGossipBroadcast(_ PeerName, update []byte) (GossipData, error) { return NewSurrogateGossipData(update), nil } diff --git a/testing/gossip/mocks.go b/testing/gossip/mocks.go index d3aad9f3..64889d29 100644 --- a/testing/gossip/mocks.go +++ b/testing/gossip/mocks.go @@ -15,7 +15,8 @@ type unicastMessage struct { buf []byte } type broadcastMessage struct { - data router.GossipData + sender router.PeerName + data router.GossipData } type exitMessage struct { exitChan chan struct{} @@ -91,7 +92,7 @@ func (grouter *TestRouter) run(gossiper router.Gossiper, gossipChan chan interfa continue } for _, msg := range message.data.Encode() { - if _, err := gossiper.OnGossipBroadcast(msg); err != nil { + if _, err := gossiper.OnGossipBroadcast(message.sender, msg); err != nil { panic(fmt.Sprintf("Error doing gossip broadcast: %s", err)) } } From 273a0fba474a784137c20e7658b478b26a107953 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Wed, 15 Jul 2015 12:33:11 +0100 Subject: [PATCH 2/3] Add sender to mock broadcast messages --- ipam/allocator_test.go | 4 ++-- ipam/testutils_test.go | 2 +- testing/gossip/mocks.go | 12 ++++++------ 3 files changed, 9 insertions(+), 9 deletions(-) diff --git a/ipam/allocator_test.go b/ipam/allocator_test.go index 67d162b8..930dc437 100644 --- a/ipam/allocator_test.go +++ b/ipam/allocator_test.go @@ -254,9 +254,9 @@ func TestTransfer(t *testing.T) { _, err = alloc3.Allocate("bar", subnet, nil) require.True(t, err == nil, "Failed to get address") - router.GossipBroadcast(alloc2.Gossip()) + alloc2.gossip.GossipBroadcast(alloc2.Gossip()) router.Flush() - router.GossipBroadcast(alloc3.Gossip()) + alloc2.gossip.GossipBroadcast(alloc3.Gossip()) router.Flush() router.RemovePeer(alloc2.ourName) router.RemovePeer(alloc3.ourName) diff --git a/ipam/testutils_test.go b/ipam/testutils_test.go index 294444d6..95f6407a 100644 --- a/ipam/testutils_test.go +++ b/ipam/testutils_test.go @@ -208,7 +208,7 @@ func makeNetworkOfAllocators(size int, cidr string) ([]*Allocator, *gossip.TestR allocs[i] = alloc } - gossipRouter.GossipBroadcast(allocs[size-1].Gossip()) + allocs[size-1].gossip.GossipBroadcast(allocs[size-1].Gossip()) gossipRouter.Flush() return allocs, gossipRouter, subnet } diff --git a/testing/gossip/mocks.go b/testing/gossip/mocks.go index 64889d29..43f7ee5a 100644 --- a/testing/gossip/mocks.go +++ b/testing/gossip/mocks.go @@ -34,10 +34,10 @@ func NewTestRouter(loss float32) *TestRouter { return &TestRouter{make(map[router.PeerName]chan interface{}, 100), loss} } -func (grouter *TestRouter) GossipBroadcast(update router.GossipData) error { +func (grouter *TestRouter) gossipBroadcast(sender router.PeerName, update router.GossipData) error { for _, gossipChan := range grouter.gossipChans { select { - case gossipChan <- broadcastMessage{data: update}: + case gossipChan <- broadcastMessage{sender: sender, data: update}: default: // drop the message if we cannot send it common.Log.Errorf("Dropping message") } @@ -66,7 +66,7 @@ type TestRouterClient struct { sender router.PeerName } -func (grouter *TestRouter) run(gossiper router.Gossiper, gossipChan chan interface{}) { +func (grouter *TestRouter) run(sender router.PeerName, gossiper router.Gossiper, gossipChan chan interface{}) { gossipTimer := time.Tick(10 * time.Second) for { select { @@ -98,7 +98,7 @@ func (grouter *TestRouter) run(gossiper router.Gossiper, gossipChan chan interfa } } case <-gossipTimer: - grouter.GossipBroadcast(gossiper.Gossip()) + grouter.gossipBroadcast(sender, gossiper.Gossip()) } } } @@ -106,7 +106,7 @@ func (grouter *TestRouter) run(gossiper router.Gossiper, gossipChan chan interfa func (grouter *TestRouter) Connect(sender router.PeerName, gossiper router.Gossiper) router.Gossip { gossipChan := make(chan interface{}, 100) - go grouter.run(gossiper, gossipChan) + go grouter.run(sender, gossiper, gossipChan) grouter.gossipChans[sender] = gossipChan return TestRouterClient{grouter, sender} @@ -122,5 +122,5 @@ func (client TestRouterClient) GossipUnicast(dstPeerName router.PeerName, buf [] } func (client TestRouterClient) GossipBroadcast(update router.GossipData) error { - return client.router.GossipBroadcast(update) + return client.router.gossipBroadcast(client.sender, update) } From b3979c95051ec1aeb81059eec83e7f0559dfab9c Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Wed, 15 Jul 2015 12:39:47 +0100 Subject: [PATCH 3/3] Respond to paxos message if we already have a ring --- ipam/allocator.go | 30 ++++++++++++++++++------------ 1 file changed, 18 insertions(+), 12 deletions(-) diff --git a/ipam/allocator.go b/ipam/allocator.go index ff0c7380..8137cb33 100644 --- a/ipam/allocator.go +++ b/ipam/allocator.go @@ -359,7 +359,7 @@ func (alloc *Allocator) OnGossipUnicast(sender router.PeerName, msg []byte) erro alloc.donateSpace(r, sender) resultChan <- nil case msgRingUpdate: - resultChan <- alloc.update(msg[1:]) + resultChan <- alloc.update(sender, msg[1:]) } } return <-resultChan @@ -370,7 +370,7 @@ func (alloc *Allocator) OnGossipBroadcast(sender router.PeerName, msg []byte) (r alloc.debugln("OnGossipBroadcast from", sender, ":", len(msg), "bytes") resultChan := make(chan error) alloc.actionChan <- func() { - resultChan <- alloc.update(msg) + resultChan <- alloc.update(sender, msg) } return alloc.Gossip(), <-resultChan } @@ -419,7 +419,7 @@ func (alloc *Allocator) OnGossip(msg []byte) (router.GossipData, error) { alloc.debugln("Allocator.OnGossip:", len(msg), "bytes") resultChan := make(chan error) alloc.actionChan <- func() { - resultChan <- alloc.update(msg) + resultChan <- alloc.update(router.UnknownPeerName, msg) } return nil, <-resultChan // for now, we never propagate updates. TBD } @@ -589,7 +589,7 @@ func (alloc *Allocator) sendRingUpdate(dest router.PeerName) { alloc.gossip.GossipUnicast(dest, msg) } -func (alloc *Allocator) update(msg []byte) error { +func (alloc *Allocator) update(sender router.PeerName, msg []byte) error { reader := bytes.NewReader(msg) decoder := gob.NewDecoder(reader) var data gossipState @@ -621,16 +621,22 @@ func (alloc *Allocator) update(msg []byte) error { return err } - if data.Paxos != nil && alloc.ring.Empty() { - if alloc.paxos.Update(data.Paxos) { - if alloc.paxos.Think() { - // If something important changed, broadcast - alloc.gossip.GossipBroadcast(alloc.Gossip()) - } + if data.Paxos != nil { + if alloc.ring.Empty() { + if alloc.paxos.Update(data.Paxos) { + if alloc.paxos.Think() { + // If something important changed, broadcast + alloc.gossip.GossipBroadcast(alloc.Gossip()) + } - if ok, cons := alloc.paxos.Consensus(); ok { - alloc.createRing(cons.Value) + if ok, cons := alloc.paxos.Consensus(); ok { + alloc.createRing(cons.Value) + } } + } else if sender != router.UnknownPeerName { + // Sender is trying to initialize a ring, but we have one + // already - send it straight back + alloc.sendRingUpdate(sender) } }