From 94220f22b2b04a87f4404b8fe6549402027c33ad Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Sun, 14 Jun 2015 08:47:51 +0100 Subject: [PATCH 1/5] allow dynamic addition of gossip channels --- router/gossip.go | 31 ++++++++++++++++++++++++++----- router/router.go | 6 ++++-- 2 files changed, 30 insertions(+), 7 deletions(-) diff --git a/router/gossip.go b/router/gossip.go index f8d026ae..eba13174 100644 --- a/router/gossip.go +++ b/router/gossip.go @@ -1,5 +1,9 @@ package router +import ( + "fmt" +) + type GossipData interface { Encode() []byte Merge(GossipData) @@ -88,17 +92,34 @@ type GossipChannels map[string]*GossipChannel func (router *Router) NewGossip(channelName string, g Gossiper) Gossip { channel := NewGossipChannel(channelName, router.Ourself, router.Routes, g) - router.GossipChannels[channelName] = channel + router.gossipLock.Lock() + defer router.gossipLock.Unlock() + if _, found := router.gossipChannels[channelName]; found { + checkFatal(fmt.Errorf("[gossip] duplicate channel %s", channelName)) + } + router.gossipChannels[channelName] = channel return channel } func (router *Router) gossipChannel(channelName string) (*GossipChannel, bool) { - channel, found := router.GossipChannels[channelName] + router.gossipLock.RLock() + defer router.gossipLock.RUnlock() + channel, found := router.gossipChannels[channelName] return channel, found } +func (router *Router) gossipChannelSet() map[*GossipChannel]struct{} { + channels := make(map[*GossipChannel]struct{}) + router.gossipLock.RLock() + defer router.gossipLock.RUnlock() + for _, channel := range router.gossipChannels { + channels[channel] = void + } + return channels +} + func (router *Router) SendAllGossip() { - for _, channel := range router.GossipChannels { + for channel := range router.gossipChannelSet() { if gossip := channel.gossiper.Gossip(); gossip != nil { channel.Send(router.Ourself.Name, gossip) } @@ -106,7 +127,7 @@ func (router *Router) SendAllGossip() { } func (router *Router) SendAllGossipDown(conn Connection) { - for _, channel := range router.GossipChannels { + for channel := range router.gossipChannelSet() { if gossip := channel.gossiper.Gossip(); gossip != nil { channel.SendDown(conn, channel.gossiper.Gossip()) } @@ -117,7 +138,7 @@ func (router *Router) SendAllGossipDown(conn Connection) { func (router *Router) sendPendingGossip() bool { sentSomething := false - for _, channel := range router.GossipChannels { + for channel := range router.gossipChannelSet() { channel.Lock() for _, sender := range channel.senders { sentSomething = sender.flush() || sentSomething diff --git a/router/router.go b/router/router.go index c0805d14..3473b05d 100644 --- a/router/router.go +++ b/router/router.go @@ -7,6 +7,7 @@ import ( "io" "log" "net" + "sync" "syscall" "time" @@ -45,7 +46,8 @@ type Router struct { Peers *Peers Routes *Routes ConnectionMaker *ConnectionMaker - GossipChannels GossipChannels + gossipLock sync.RWMutex + gossipChannels GossipChannels TopologyGossip Gossip UDPListener *net.UDPConn } @@ -64,7 +66,7 @@ type PacketSourceSink interface { } func NewRouter(config Config, name PeerName, nickName string) *Router { - router := &Router{Config: config, GossipChannels: make(GossipChannels)} + router := &Router{Config: config, gossipChannels: make(GossipChannels)} onMacExpiry := func(mac net.HardwareAddr, peer *Peer) { log.Println("Expired MAC", mac, "at", peer) } From 0b3584a3551e82748d2b906336d058da5063e4bb Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Sun, 14 Jun 2015 09:34:03 +0100 Subject: [PATCH 2/5] permit encoding of GossipData into multiple messages --- ipam/allocator.go | 4 ++-- ipam/testutils_test.go | 6 ++++-- router/gossip.go | 2 +- router/gossip_channel.go | 17 +++++++++++------ router/router.go | 4 ++-- 5 files changed, 20 insertions(+), 13 deletions(-) diff --git a/ipam/allocator.go b/ipam/allocator.go index 72ef3072..8b609b96 100644 --- a/ipam/allocator.go +++ b/ipam/allocator.go @@ -435,8 +435,8 @@ func (d *ipamGossipData) Merge(other router.GossipData) { // no-op } -func (d *ipamGossipData) Encode() []byte { - return d.alloc.Encode() +func (d *ipamGossipData) Encode() [][]byte { + return [][]byte{d.alloc.Encode()} } // Gossip returns a GossipData implementation, which in this case always diff --git a/ipam/testutils_test.go b/ipam/testutils_test.go index c685685b..6768a121 100644 --- a/ipam/testutils_test.go +++ b/ipam/testutils_test.go @@ -258,8 +258,10 @@ func (grouter *TestGossipRouter) connect(sender router.PeerName, gossiper router panic(fmt.Sprintf("Error doing gossip unicast to %s: %s", sender, err)) } } else { - if _, err := gossiper.OnGossipBroadcast(message.data.Encode()); err != nil { - panic(fmt.Sprintf("Error doing gossip broadcast to %s: %s", sender, err)) + for _, msg := range message.data.Encode() { + if _, err := gossiper.OnGossipBroadcast(msg); err != nil { + panic(fmt.Sprintf("Error doing gossip broadcast to %s: %s", sender, err)) + } } } case <-gossipTimer: diff --git a/router/gossip.go b/router/gossip.go index eba13174..d4293a4d 100644 --- a/router/gossip.go +++ b/router/gossip.go @@ -5,7 +5,7 @@ import ( ) type GossipData interface { - Encode() []byte + Encode() [][]byte Merge(GossipData) } diff --git a/router/gossip_channel.go b/router/gossip_channel.go index fdc7c5ed..0b1e808c 100644 --- a/router/gossip_channel.go +++ b/router/gossip_channel.go @@ -145,8 +145,10 @@ func (c *GossipChannel) sendDown(conn Connection, data GossipData) { sender, found := c.senders[conn] if !found { sender = NewGossipSender(func(pending GossipData) { - protocolMsg := ProtocolMsg{ProtocolGossip, GobEncode(c.name, c.ourself.Name, pending.Encode())} - conn.(ProtocolSender).SendProtocolMsg(protocolMsg) + for _, msg := range pending.Encode() { + protocolMsg := ProtocolMsg{ProtocolGossip, GobEncode(c.name, c.ourself.Name, msg)} + conn.(ProtocolSender).SendProtocolMsg(protocolMsg) + } }) c.senders[conn] = sender sender.Start() @@ -212,10 +214,13 @@ func (c *GossipChannel) sendBroadcast(srcName PeerName, update GossipData) { if len(nextHops) == 0 { return } - protocolMsg := ProtocolMsg{ProtocolGossipBroadcast, GobEncode(c.name, srcName, update.Encode())} - // FIXME a single blocked connection can stall us - for _, conn := range c.ourself.ConnectionsTo(nextHops) { - conn.(ProtocolSender).SendProtocolMsg(protocolMsg) + connections := c.ourself.ConnectionsTo(nextHops) + for _, msg := range update.Encode() { + protocolMsg := ProtocolMsg{ProtocolGossipBroadcast, GobEncode(c.name, srcName, msg)} + // FIXME a single blocked connection can stall us + for _, conn := range connections { + conn.(ProtocolSender).SendProtocolMsg(protocolMsg) + } } } diff --git a/router/router.go b/router/router.go index 3473b05d..ab671a0d 100644 --- a/router/router.go +++ b/router/router.go @@ -394,8 +394,8 @@ func (d *TopologyGossipData) Merge(other GossipData) { } } -func (d *TopologyGossipData) Encode() []byte { - return d.peers.EncodePeers(d.update) +func (d *TopologyGossipData) Encode() [][]byte { + return [][]byte{d.peers.EncodePeers(d.update)} } func (router *Router) OnGossipUnicast(sender PeerName, msg []byte) error { From dea5bd06392f4cff4e96e7f8e978dea5abb061a9 Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Mon, 15 Jun 2015 11:23:32 +0100 Subject: [PATCH 3/5] introduce SurrogateGossiper for dealing with unknown gossip channels Instead of erroring when we receive gossip for an unknown channel, we create a GossipChannel on the fly, with a SurrogateGossiper singleton as the gossiper. The SurrogateGossiper ignores unicasts and relays broadcasts and gossip. --- router/gossip.go | 17 ++++++++++++--- router/gossip_channel.go | 6 +----- router/surrogate_gossiper.go | 41 ++++++++++++++++++++++++++++++++++++ 3 files changed, 56 insertions(+), 8 deletions(-) create mode 100644 router/surrogate_gossiper.go diff --git a/router/gossip.go b/router/gossip.go index d4293a4d..f66ea9d0 100644 --- a/router/gossip.go +++ b/router/gossip.go @@ -101,11 +101,22 @@ func (router *Router) NewGossip(channelName string, g Gossiper) Gossip { return channel } -func (router *Router) gossipChannel(channelName string) (*GossipChannel, bool) { +func (router *Router) gossipChannel(channelName string) *GossipChannel { router.gossipLock.RLock() - defer router.gossipLock.RUnlock() channel, found := router.gossipChannels[channelName] - return channel, found + router.gossipLock.RUnlock() + if found { + return channel + } + router.gossipLock.Lock() + defer router.gossipLock.Unlock() + if channel, found = router.gossipChannels[channelName]; found { + return channel + } + channel = NewGossipChannel(channelName, router.Ourself, router.Routes, &surrogateGossiper) + channel.log("created surrogate channel") + router.gossipChannels[channelName] = channel + return channel } func (router *Router) gossipChannelSet() map[*GossipChannel]struct{} { diff --git a/router/gossip_channel.go b/router/gossip_channel.go index 0b1e808c..da7dc0ec 100644 --- a/router/gossip_channel.go +++ b/router/gossip_channel.go @@ -3,7 +3,6 @@ package router import ( "bytes" "encoding/gob" - "fmt" "log" "sync" ) @@ -37,10 +36,7 @@ func (router *Router) handleGossip(tag ProtocolTag, payload []byte) error { if err := decoder.Decode(&channelName); err != nil { return err } - channel, found := router.gossipChannel(channelName) - if !found { - return fmt.Errorf("[gossip] received unknown channel with name %s", channelName) - } + channel := router.gossipChannel(channelName) var srcName PeerName if err := decoder.Decode(&srcName); err != nil { return err diff --git a/router/surrogate_gossiper.go b/router/surrogate_gossiper.go new file mode 100644 index 00000000..ea7f443d --- /dev/null +++ b/router/surrogate_gossiper.go @@ -0,0 +1,41 @@ +package router + +type SurrogateGossipData struct { + messages [][]byte +} + +func NewSurrogateGossipData(msg []byte) *SurrogateGossipData { + return &SurrogateGossipData{messages: [][]byte{msg}} +} + +func (d *SurrogateGossipData) Encode() [][]byte { + return d.messages +} + +func (d *SurrogateGossipData) Merge(other GossipData) { + d.messages = append(d.messages, other.(*SurrogateGossipData).messages...) +} + +// SurrogateGossiper ignores unicasts and relays broadcasts and gossips. + +type SurrogateGossiper struct{} + +func (*SurrogateGossiper) OnGossipUnicast(sender PeerName, msg []byte) error { + return nil +} + +func (*SurrogateGossiper) OnGossipBroadcast(update []byte) (GossipData, error) { + return NewSurrogateGossipData(update), nil +} + +func (*SurrogateGossiper) Gossip() GossipData { + return nil +} + +func (*SurrogateGossiper) OnGossip(update []byte) (GossipData, error) { + return NewSurrogateGossipData(update), nil +} + +var ( + surrogateGossiper SurrogateGossiper +) From 36ba838f548ef3aa572b2e1582794ab1585f99c9 Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Sun, 14 Jun 2015 13:44:27 +0100 Subject: [PATCH 4/5] remove ipam.DummyAllocator and rely on new SurrogateGossiper As a result it is possible to mix nodes with and without IPAM. Note that the quorum only involves the former, and hence such mixed environments will typically require a custom `-initpeercount`. --- ipam/dummy.go | 32 -------------------------------- prog/weaver/main.go | 2 -- 2 files changed, 34 deletions(-) delete mode 100644 ipam/dummy.go diff --git a/ipam/dummy.go b/ipam/dummy.go deleted file mode 100644 index 35d6898b..00000000 --- a/ipam/dummy.go +++ /dev/null @@ -1,32 +0,0 @@ -package ipam - -import ( - "errors" - - "github.com/weaveworks/weave/router" -) - -// Exists solely for the purpose of detecting a network in which some -// peers are configured with IPAM on and some with IPAM off. -type DummyAllocator struct { -} - -var ( - ErrMixedConfig = errors.New("Mixed IPAM/non-IPAM configuration not supported") -) - -func (alloc *DummyAllocator) OnGossipUnicast(sender router.PeerName, msg []byte) error { - return ErrMixedConfig -} - -func (alloc *DummyAllocator) OnGossipBroadcast(msg []byte) (router.GossipData, error) { - return nil, ErrMixedConfig -} - -func (alloc *DummyAllocator) OnGossip(msg []byte) (router.GossipData, error) { - return nil, ErrMixedConfig -} - -func (alloc *DummyAllocator) Gossip() router.GossipData { - return nil -} diff --git a/prog/weaver/main.go b/prog/weaver/main.go index 25fdad75..d0ac8596 100644 --- a/prog/weaver/main.go +++ b/prog/weaver/main.go @@ -153,8 +153,6 @@ func main() { } } else if peerCount > 0 { log.Fatal("-initpeercount flag specified without -iprange") - } else { - router.NewGossip("IPallocation", &ipam.DummyAllocator{}) } router.Start() From e15ebd9118f5daa8e1d1c108579cfd764f00b1e6 Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Mon, 15 Jun 2015 11:38:01 +0100 Subject: [PATCH 5/5] add unit test for SurrogateGossiper --- router/gossip_test.go | 103 +++++++++++++++++++++++++++++++++++++++++- 1 file changed, 101 insertions(+), 2 deletions(-) diff --git a/router/gossip_test.go b/router/gossip_test.go index 8ff9feb5..15ae8195 100644 --- a/router/gossip_test.go +++ b/router/gossip_test.go @@ -1,14 +1,15 @@ package router import ( + "sync" "testing" "time" wt "github.com/weaveworks/weave/testing" ) -// TODO test gossip unicast; atm we only test topology gossip, which -// does not employ unicast. +// TODO test gossip unicast; atm we only test topology gossip and +// surrogates, neither of which employ unicast. type mockChannelConnection struct { RemoteConnection @@ -164,3 +165,101 @@ func implTestGossipTopology(t *testing.T) { sendPendingGossip(r1, r2, r3) checkTopology(t, r1, r1.tp(r2), r2.tp(r1), r3.tp(r1)) } + +func TestGossipSurrogate(t *testing.T) { + // create the topology r1 <-> r2 <-> r3 + r1 := NewTestRouter("01:00:00:01:00:00") + r2 := NewTestRouter("02:00:00:02:00:00") + r3 := NewTestRouter("03:00:00:03:00:00") + r1.AddTestChannelConnection(r2) + r2.AddTestChannelConnection(r1) + r3.AddTestChannelConnection(r2) + r2.AddTestChannelConnection(r3) + sendPendingGossip(r1, r2, r3) + checkTopology(t, r1, r1.tp(r2), r2.tp(r1, r3), r3.tp(r2)) + checkTopology(t, r2, r1.tp(r2), r2.tp(r1, r3), r3.tp(r2)) + checkTopology(t, r3, r1.tp(r2), r2.tp(r1, r3), r3.tp(r2)) + + // create a gossiper at either end, but not the middle + g1 := newTestGossiper() + g3 := newTestGossiper() + s1 := r1.NewGossip("Test", g1) + s3 := r3.NewGossip("Test", g3) + + // broadcast a message from each end, check it reaches the other + broadcast(s1, 1) + broadcast(s3, 2) + sendPendingGossip(r1, r2, r3) + g1.checkHas(t, 2) + g3.checkHas(t, 1) + + // check that each end gets their message back through periodic + // gossip + r1.SendAllGossip() + r3.SendAllGossip() + sendPendingGossip(r1, r2, r3) + g1.checkHas(t, 1, 2) + g3.checkHas(t, 1, 2) +} + +type testGossiper struct { + sync.RWMutex + state map[byte]struct{} +} + +func newTestGossiper() *testGossiper { + return &testGossiper{state: make(map[byte]struct{})} +} + +func (g *testGossiper) OnGossipUnicast(sender PeerName, msg []byte) error { + return nil +} + +func (g *testGossiper) OnGossipBroadcast(update []byte) (GossipData, error) { + g.Lock() + defer g.Unlock() + for _, v := range update { + g.state[v] = void + } + return NewSurrogateGossipData(update), nil +} + +func (g *testGossiper) Gossip() GossipData { + g.RLock() + defer g.RUnlock() + state := make([]byte, len(g.state)) + for v := range g.state { + state = append(state, v) + } + return NewSurrogateGossipData(state) +} + +func (g *testGossiper) OnGossip(update []byte) (GossipData, error) { + g.Lock() + defer g.Unlock() + var delta []byte + for _, v := range update { + if _, found := g.state[v]; !found { + delta = append(delta, v) + g.state[v] = void + } + } + if len(delta) > 0 { + return NewSurrogateGossipData(delta), nil + } + return nil, nil +} + +func (g *testGossiper) checkHas(t *testing.T, vs ...byte) { + g.RLock() + defer g.RUnlock() + for _, v := range vs { + if _, found := g.state[v]; !found { + wt.Fatalf(t, "%d is missing", v) + } + } +} + +func broadcast(s Gossip, v byte) { + s.GossipBroadcast(NewSurrogateGossipData([]byte{v})) +}