Merge pull request #914 from rade/911_different_gossip_channels

LGTM. Fixes #911

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

ipam/dummy.go

@@ -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
-}

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:

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()

router/gossip.go

@@ -1,7 +1,11 @@
 package router
 
+import (
+	"fmt"
+)
+
 type GossipData interface {
-	Encode() []byte
+	Encode() [][]byte
 	Merge(GossipData)
 }
 
@@ -88,17 +92,45 @@ 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]
-	return channel, found
+func (router *Router) gossipChannel(channelName string) *GossipChannel {
+	router.gossipLock.RLock()
+	channel, found := router.gossipChannels[channelName]
+	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{} {
+	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 +138,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 +149,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

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
@@ -145,8 +141,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 +210,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)
+		}
 	}
 }
 

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}))
+}

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)
 	}
@@ -392,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 {

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
+)