Merge pull request #313 from bboreham/add_gossip

Gossip layer to communicate topology between peers

Note that I have squashed over 100 commits from @bboreham and @rade into the three for this PR; you can see the original history at https://github.com/bboreham/weave/compare/add_gossip_full

router/connection.go

@@ -48,7 +48,6 @@ type LocalConnection struct {
 	SessionKey        *[32]byte
 	heartbeatFrame    *ForwardedFrame
 	heartbeat         *time.Ticker
-	fetchAll          *time.Ticker
 	fragTest          *time.Ticker
 	forwardChan       chan<- *ForwardedFrame
 	forwardChanDF     chan<- *ForwardedFrame
@@ -285,8 +284,6 @@ func (conn *LocalConnection) queryLoop(queryChan <-chan *ConnectionInteraction)
 			}
 		case <-tickerChan(conn.heartbeat):
 			conn.Forward(true, conn.heartbeatFrame, nil)
-		case <-tickerChan(conn.fetchAll):
-			err = conn.handleSendSimpleProtocolMsg(ProtocolFetchAll)
 		case <-tickerChan(conn.fragTest):
 			conn.setStackFrag(false)
 			err = conn.handleSendSimpleProtocolMsg(ProtocolStartFragmentationTest)
@@ -343,13 +340,9 @@ func (conn *LocalConnection) handleSetEstablished() error {
 		frame:   PMTUDiscovery},
 		nil)
 	conn.heartbeat = time.NewTicker(SlowHeartbeat)
-	conn.fetchAll = time.NewTicker(FetchAllInterval)
 	conn.fragTest = time.NewTicker(FragTestInterval)
 	// avoid initial waits for timers to fire
 	conn.Forward(true, conn.heartbeatFrame, nil)
-	if err := conn.handleSendSimpleProtocolMsg(ProtocolFetchAll); err != nil {
-		return err
-	}
 	conn.setStackFrag(false)
 	if err := conn.handleSendSimpleProtocolMsg(ProtocolStartFragmentationTest); err != nil {
 		return err
@@ -376,7 +369,6 @@ func (conn *LocalConnection) handleShutdown() {
 	}
 
 	stopTicker(conn.heartbeat)
-	stopTicker(conn.fetchAll)
 	stopTicker(conn.fragTest)
 
 	// blank out the forwardChan so that the router processes don't
@@ -572,37 +564,14 @@ func (conn *LocalConnection) handleProtocolMsg(tag ProtocolTag, payload []byte)
 			return fmt.Errorf("unexpected nonce on unencrypted connection")
 		}
 		conn.Decryptor.ReceiveNonce(payload)
-	case ProtocolFetchAll:
-		// There are exactly two messages that relate to topology
-		// updates.
-		//
-		// 1. FetchAll. This carries no payload. The receiver responds
-		// with the entire topology model as the receiver has it.
-		//
-		// 2. Update. This carries a topology payload. The receiver
-		// merges it with its own topology model. If the payload is a
-		// subset of the receiver's topology, no further action is
-		// taken. Otherwise, the receiver sends out to all its
-		// connections an "improved" update.
-		conn.SendProtocolMsg(ProtocolMsg{ProtocolUpdate, conn.Router.Peers.EncodeAllPeers()})
-	case ProtocolUpdate:
-		newUpdate, err := conn.Router.Peers.ApplyUpdate(payload)
-		if _, ok := err.(UnknownPeersError); err != nil && ok {
-			// That update contained a peer we didn't know about;
-			// request full update
-			conn.SendProtocolMsg(ProtocolMsg{ProtocolFetchAll, nil})
-			break
-		}
-		if err != nil {
-			return err
-		}
-		if len(newUpdate) != 0 {
-			conn.Router.ConnectionMaker.Refresh()
-			conn.Router.Routes.Recalculate()
-			conn.Router.Ourself.SendProtocolMsg(ProtocolMsg{ProtocolUpdate, newUpdate})
-		}
 	case ProtocolPMTUVerified:
 		conn.verifyPMTU <- int(binary.BigEndian.Uint16(payload))
+	case ProtocolGossipUnicast:
+		return conn.Router.handleGossip(payload, deliverGossipUnicast)
+	case ProtocolGossipBroadcast:
+		return conn.Router.handleGossip(payload, deliverGossipBroadcast)
+	case ProtocolGossip:
+		return conn.Router.handleGossip(payload, deliverGossip)
 	default:
 		conn.log("ignoring unknown protocol tag:", tag)
 	}

router/consts.go

@@ -18,9 +18,8 @@ const (
 	PMTUDiscoverySize  = 60000
 	FastHeartbeat      = 500 * time.Millisecond
 	SlowHeartbeat      = 10 * time.Second
-	FetchAllInterval   = 30 * time.Second
 	FragTestInterval   = 5 * time.Minute
-	ReadTimeout        = 2 * FetchAllInterval
+	ReadTimeout        = 1 * time.Minute
 	PMTUVerifyAttempts = 8
 	PMTUVerifyTimeout  = 10 * time.Millisecond // gets doubled with every attempt
 	MaxDuration        = time.Duration(math.MaxInt64)

router/gossip.go

@@ -0,0 +1,162 @@
+package router
+
+import (
+	"bytes"
+	"encoding/gob"
+	"fmt"
+	"log"
+	"time"
+)
+
+const GossipInterval = 30 * time.Second
+
+type Gossip interface {
+	// specific message from one peer to another
+	// intermediate peers relay it using unicast topology.
+	GossipUnicast(dstPeerName PeerName, buf []byte) error
+	// send a message to every peer, relayed using broadcast topology.
+	GossipBroadcast(buf []byte) error
+}
+
+type Gossiper interface {
+	OnGossipUnicast(sender PeerName, msg []byte) error
+	OnGossipBroadcast(msg []byte) error
+	// Return state of everything we know; gets called periodically
+	Gossip() []byte
+	// merge in state and return "everything new I've just learnt",
+	// or nil if nothing in the received message was new
+	OnGossip(buf []byte) ([]byte, error)
+}
+
+type GossipChannel struct {
+	ourself  *LocalPeer
+	name     string
+	hash     uint32
+	gossiper Gossiper
+}
+
+func (router *Router) NewGossip(channelName string, g Gossiper) Gossip {
+	channelHash := hash(channelName)
+	channel := &GossipChannel{router.Ourself, channelName, channelHash, g}
+	router.GossipChannels[channelHash] = channel
+	return channel
+}
+
+func (router *Router) SendAllGossip() {
+	for _, channel := range router.GossipChannels {
+		channel.SendGossipMsg(channel.gossiper.Gossip())
+	}
+}
+
+func (router *Router) SendAllGossipDown(conn Connection) {
+	for _, channel := range router.GossipChannels {
+		protocolMsg := channel.gossipMsg(channel.gossiper.Gossip())
+		conn.(ProtocolSender).SendProtocolMsg(protocolMsg)
+	}
+}
+
+func (router *Router) handleGossip(payload []byte, onok func(*GossipChannel, PeerName, []byte, *gob.Decoder) error) error {
+	decoder := gob.NewDecoder(bytes.NewReader(payload))
+	var channelHash uint32
+	if err := decoder.Decode(&channelHash); err != nil {
+		return err
+	}
+	channel, found := router.GossipChannels[channelHash]
+	if !found {
+		return fmt.Errorf("[gossip] received unknown channel with hash %v", channelHash)
+	}
+	var srcName PeerName
+	if err := decoder.Decode(&srcName); err != nil {
+		return err
+	}
+	if err := onok(channel, srcName, payload, decoder); err != nil {
+		return err
+	}
+	return nil
+}
+
+func deliverGossipUnicast(channel *GossipChannel, srcName PeerName, origPayload []byte, dec *gob.Decoder) error {
+	var destName PeerName
+	if err := dec.Decode(&destName); err != nil {
+		return err
+	}
+	if channel.ourself.Name == destName {
+		var payload []byte
+		if err := dec.Decode(&payload); err != nil {
+			return err
+		}
+		return channel.gossiper.OnGossipUnicast(srcName, payload)
+	} else {
+		return channel.relayGossipUnicast(destName, origPayload)
+	}
+}
+
+func deliverGossipBroadcast(channel *GossipChannel, srcName PeerName, origPayload []byte, dec *gob.Decoder) error {
+	var payload []byte
+	if err := dec.Decode(&payload); err != nil {
+		return err
+	}
+	if err := channel.gossiper.OnGossipBroadcast(payload); err != nil {
+		return err
+	}
+	return channel.relayGossipBroadcast(srcName, origPayload)
+}
+
+func deliverGossip(channel *GossipChannel, srcName PeerName, _ []byte, dec *gob.Decoder) error {
+	var payload []byte
+	if err := dec.Decode(&payload); err != nil {
+		return err
+	}
+	if newBuf, err := channel.gossiper.OnGossip(payload); err != nil {
+		return err
+	} else if newBuf != nil {
+		channel.SendGossipMsg(newBuf)
+	}
+	return nil
+}
+
+func (c *GossipChannel) SendGossipMsg(buf []byte) {
+	protocolMsg := c.gossipMsg(buf)
+	c.ourself.ForEachConnection(func(_ PeerName, conn Connection) {
+		conn.(ProtocolSender).SendProtocolMsg(protocolMsg)
+	})
+}
+
+func (c *GossipChannel) gossipMsg(buf []byte) ProtocolMsg {
+	return ProtocolMsg{ProtocolGossip, GobEncode(c.hash, c.ourself.Name, buf)}
+}
+
+func (c *GossipChannel) GossipUnicast(dstPeerName PeerName, buf []byte) error {
+	return c.relayGossipUnicast(dstPeerName, GobEncode(c.hash, c.ourself.Name, dstPeerName, buf))
+}
+
+func (c *GossipChannel) GossipBroadcast(buf []byte) error {
+	return c.relayGossipBroadcast(c.ourself.Name, GobEncode(c.hash, c.ourself.Name, buf))
+}
+
+func (c *GossipChannel) relayGossipUnicast(dstPeerName PeerName, msg []byte) error {
+	if relayPeerName, found := c.ourself.Router.Routes.Unicast(dstPeerName); !found {
+		c.log("unknown relay destination:", dstPeerName)
+	} else if conn, found := c.ourself.ConnectionTo(relayPeerName); !found {
+		c.log("unable to find connection to relay peer", relayPeerName)
+	} else {
+		conn.(ProtocolSender).SendProtocolMsg(ProtocolMsg{ProtocolGossipUnicast, msg})
+	}
+	return nil
+}
+
+func (c *GossipChannel) relayGossipBroadcast(srcName PeerName, msg []byte) error {
+	if srcPeer, found := c.ourself.Router.Peers.Fetch(srcName); !found {
+		c.log("unable to relay broadcast from unknown peer", srcName)
+	} else {
+		protocolMsg := ProtocolMsg{ProtocolGossipBroadcast, msg}
+		for _, conn := range c.ourself.NextBroadcastHops(srcPeer) {
+			conn.SendProtocolMsg(protocolMsg)
+		}
+	}
+	return nil
+}
+
+func (c *GossipChannel) log(args ...interface{}) {
+	log.Println(append(append([]interface{}{}, "[gossip "+c.name+"]:"), args...)...)
+}

router/gossip_test.go

@@ -0,0 +1,146 @@
+package router
+
+import (
+	wt "github.com/zettio/weave/testing"
+	"testing"
+	"time"
+)
+
+// TODO test gossip unicast and broadcast; atm we only test topology
+// gossip, which does not employ unicast or broadcast.
+
+type mockChannelConnection struct {
+	RemoteConnection
+	dest *Router
+}
+
+// Construct a "passive" Router, i.e. without any goroutines.
+//
+// We need to create some dummy channels otherwise tests hang on nil
+// channels when Router.OnGossip() calls async methods.
+func NewTestRouter(name PeerName) *Router {
+	router := NewRouter(nil, name, nil, 10, 1024, nil)
+	router.ConnectionMaker.queryChan = make(chan *ConnectionMakerInteraction, ChannelSize)
+	router.Routes.queryChan = make(chan *Interaction, ChannelSize)
+	return router
+}
+
+func (conn *mockChannelConnection) SendProtocolMsg(protocolMsg ProtocolMsg) {
+	if err := conn.dest.handleGossip(protocolMsg.msg, deliverGossip); err != nil {
+		panic(err)
+	}
+}
+
+func (router *Router) AddTestChannelConnection(r *Router) {
+	fromName := router.Ourself.Peer.Name
+	toName := r.Ourself.Peer.Name
+
+	fromPeer := NewPeer(fromName, router.Ourself.Peer.UID, 0)
+	toPeer := NewPeer(toName, r.Ourself.Peer.UID, 0)
+
+	r.Peers.FetchWithDefault(fromPeer)    // Has side-effect of incrementing refcount
+	router.Peers.FetchWithDefault(toPeer) //
+
+	conn := &mockChannelConnection{RemoteConnection{router.Ourself.Peer, toPeer, ""}, r}
+	router.Ourself.handleAddConnection(conn)
+	router.Ourself.handleConnectionEstablished(conn)
+}
+
+func (router *Router) DeleteTestChannelConnection(r *Router) {
+	fromName := router.Ourself.Peer.Name
+	toName := r.Ourself.Peer.Name
+
+	fromPeer, _ := r.Peers.Fetch(fromName)
+	toPeer, _ := router.Peers.Fetch(toName)
+
+	fromPeer.DecrementLocalRefCount()
+	toPeer.DecrementLocalRefCount()
+
+	conn, _ := router.Ourself.ConnectionTo(toName)
+	router.Ourself.handleDeleteConnection(conn)
+}
+
+func TestGossipTopology(t *testing.T) {
+	wt.RunWithTimeout(t, 1*time.Second, func() {
+		implTestGossipTopology(t)
+	})
+}
+
+// Create a Peer representing the receiver router, with connections to
+// the routers supplied as arguments, carrying across all UID and
+// version information.
+func (router *Router) tp(routers ...*Router) *Peer {
+	peer := NewPeer(router.Ourself.Peer.Name, router.Ourself.Peer.UID, 0)
+	connections := make(map[PeerName]Connection)
+	for _, r := range routers {
+		p := NewPeer(r.Ourself.Peer.Name, r.Ourself.Peer.UID, r.Ourself.Peer.version)
+		connections[r.Ourself.Peer.Name] = newMockConnection(peer, p)
+	}
+	peer.SetVersionAndConnections(router.Ourself.Peer.version, connections)
+	return peer
+}
+
+// Check that the topology of router matches the peers and all of their connections
+func checkTopology(t *testing.T, router *Router, wantedPeers ...*Peer) {
+	checkTopologyPeers(t, true, router.Peers.allPeers(), wantedPeers...)
+}
+
+func implTestGossipTopology(t *testing.T) {
+	// Create some peers that will talk to each other
+	peer1Name, _ := PeerNameFromString("01:00:00:01:00:00")
+	peer2Name, _ := PeerNameFromString("02:00:00:02:00:00")
+	peer3Name, _ := PeerNameFromString("03:00:00:03:00:00")
+	r1 := NewTestRouter(peer1Name)
+	r2 := NewTestRouter(peer2Name)
+	r3 := NewTestRouter(peer3Name)
+
+	// Check state when they have no connections
+	checkTopology(t, r1, r1.tp())
+	checkTopology(t, r2, r2.tp())
+
+	// Now try adding some connections
+	r1.AddTestChannelConnection(r2)
+	checkTopology(t, r1, r1.tp(r2), r2.tp())
+	checkTopology(t, r2, r1.tp(r2), r2.tp())
+	r2.AddTestChannelConnection(r1)
+	checkTopology(t, r1, r1.tp(r2), r2.tp(r1))
+	checkTopology(t, r2, r1.tp(r2), r2.tp(r1))
+
+	// Currently, the connection from 2 to 3 is one-way only
+	r2.AddTestChannelConnection(r3)
+	checkTopology(t, r1, r1.tp(r2), r2.tp(r1, r3), r3.tp())
+	checkTopology(t, r2, r1.tp(r2), r2.tp(r1, r3), r3.tp())
+	// When r2 gossiped to r3, 1 was unreachable from r3 so it got removed from the
+	// list of peers, but remains referenced in the connection from 1 to 3.
+	checkTopology(t, r3, r2.tp(r1, r3), r3.tp())
+
+	// Add a connection from 3 to 1 and now r1 is reachable.
+	r3.AddTestChannelConnection(r1)
+	checkTopology(t, r1, r1.tp(r2), r2.tp(r1, r3), r3.tp(r1))
+	checkTopology(t, r2, r1.tp(r2), r2.tp(r1, r3), r3.tp(r1))
+	checkTopology(t, r3, r1.tp(), r2.tp(r1, r3), r3.tp(r1))
+
+	r1.AddTestChannelConnection(r3)
+	checkTopology(t, r1, r1.tp(r2, r3), r2.tp(r1, r3), r3.tp(r1))
+	checkTopology(t, r2, r1.tp(r2, r3), r2.tp(r1, r3), r3.tp(r1))
+	checkTopology(t, r3, r1.tp(r2, r3), r2.tp(r1, r3), r3.tp(r1))
+
+	// Drop the connection from 2 to 3
+	r2.DeleteTestChannelConnection(r3)
+	checkTopology(t, r1, r1.tp(r2, r3), r2.tp(r1), r3.tp(r1))
+	checkTopology(t, r2, r1.tp(r2, r3), r2.tp(r1))
+	checkTopology(t, r3, r1.tp(r2, r3), r2.tp(r1), r3.tp(r1))
+
+	// Drop the connection from 1 to 3
+	r1.DeleteTestChannelConnection(r3)
+	checkTopology(t, r1, r1.tp(r2), r2.tp(r1), r3.tp(r1))
+
+	checkTopology(t, r1, r1.tp(r2), r2.tp(r1), r3.tp(r1))
+	checkTopology(t, r2, r1.tp(r2), r2.tp(r1))
+	// r3 still thinks r1 has a connection to it
+	checkTopology(t, r3, r1.tp(r2, r3), r2.tp(r1), r3.tp(r1))
+
+	// On a timer, r3 will gossip to r1
+	r3.SendAllGossip()
+	checkTopology(t, r1, r1.tp(r2), r2.tp(r1), r3.tp(r1))
+}

router/local_peer.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"log"
 	"net"
+	"time"
 )
 
 type LocalPeer struct {
@@ -156,30 +157,35 @@ func (peer *LocalPeer) SendProtocolMsg(m ProtocolMsg) {
 // ACTOR server
 
 func (peer *LocalPeer) queryLoop(queryChan <-chan *PeerInteraction) {
+	gossipTimer := time.Tick(GossipInterval)
 	for {
-		query, ok := <-queryChan
-		if !ok {
-			return
-		}
-		switch query.code {
-		case PAddConnection:
-			conn := query.payload.(*LocalConnection)
-			if peer.handleAddConnection(conn) {
-				conn.log("connection added")
+		select {
+		case query, ok := <-queryChan:
+			if !ok {
+				return
 			}
-		case PDeleteConnection:
-			conn := query.payload.(*LocalConnection)
-			if peer.handleDeleteConnection(conn) {
-				conn.log("connection deleted")
+			switch query.code {
+			case PAddConnection:
+				conn := query.payload.(*LocalConnection)
+				if peer.handleAddConnection(conn) {
+					conn.log("connection added")
+				}
+			case PDeleteConnection:
+				conn := query.payload.(*LocalConnection)
+				if peer.handleDeleteConnection(conn) {
+					conn.log("connection deleted")
+				}
+				query.resultChan <- nil
+			case PConnectionEstablished:
+				conn := query.payload.(*LocalConnection)
+				if peer.handleConnectionEstablished(conn) {
+					conn.log("connection fully established")
+				}
+			case PSendProtocolMsg:
+				peer.handleSendProtocolMsg(query.payload.(ProtocolMsg))
 			}
-			query.resultChan <- nil
-		case PConnectionEstablished:
-			conn := query.payload.(*LocalConnection)
-			if peer.handleConnectionEstablished(conn) {
-				conn.log("connection fully established")
-			}
-		case PSendProtocolMsg:
-			peer.handleSendProtocolMsg(query.payload.(ProtocolMsg))
+		case <-gossipTimer:
+			peer.Router.SendAllGossip()
 		}
 	}
 }
@@ -251,6 +257,7 @@ func (peer *LocalPeer) handleConnectionEstablished(conn Connection) bool {
 		return false
 	}
 	peer.connectionEstablished(conn)
+	peer.Router.SendAllGossipDown(conn)
 	peer.broadcastPeerUpdate(conn.Remote())
 	return true
 }
@@ -263,7 +270,14 @@ func (peer *LocalPeer) handleSendProtocolMsg(m ProtocolMsg) {
 
 func (peer *LocalPeer) broadcastPeerUpdate(peers ...*Peer) {
 	peer.Router.Routes.Recalculate()
-	peer.handleSendProtocolMsg(ProtocolMsg{ProtocolUpdate, EncodePeers(append(peers, peer.Peer)...)})
+	// TODO We should just be invoking TopologyGossip.GossipBroadcast
+	// here, but route calculation is asynchronous and in this
+	// particular case would likely result in the broadcast not
+	// reaching all peers. So instead we slightly break the Gossip
+	// abstraction (hence the cast) and send a regular update. This is
+	// less efficient though since it will almost certainly reach
+	// peers more than once.
+	peer.Router.TopologyGossip.(*GossipChannel).SendGossipMsg(EncodePeers(append(peers, peer.Peer)...))
 }
 
 func (peer *LocalPeer) checkConnectionLimit() error {

router/protocol.go

@@ -2,7 +2,7 @@ package router
 
 const (
 	Protocol        = "weave"
-	ProtocolVersion = 10
+	ProtocolVersion = 11
 )
 
 type ProtocolTag byte
@@ -12,9 +12,10 @@ const (
 	ProtocolFragmentationReceived
 	ProtocolStartFragmentationTest
 	ProtocolNonce
-	ProtocolFetchAll
-	ProtocolUpdate
 	ProtocolPMTUVerified
+	ProtocolGossip
+	ProtocolGossipUnicast
+	ProtocolGossipBroadcast
 )
 
 type ProtocolMsg struct {

router/router.go

@@ -24,6 +24,8 @@ type Router struct {
 	Peers           *Peers
 	Routes          *Routes
 	ConnectionMaker *ConnectionMaker
+	GossipChannels  map[uint32]*GossipChannel
+	TopologyGossip  Gossip
 	UDPListener     *net.UDPConn
 	Password        *[]byte
 	ConnLimit       int
@@ -46,10 +48,11 @@ type PacketSourceSink interface {
 
 func NewRouter(iface *net.Interface, name PeerName, password []byte, connLimit int, bufSz int, logFrame func(string, []byte, *layers.Ethernet)) *Router {
 	router := &Router{
-		Iface:     iface,
-		ConnLimit: connLimit,
-		BufSz:     bufSz,
-		LogFrame:  logFrame}
+		Iface:          iface,
+		GossipChannels: make(map[uint32]*GossipChannel),
+		ConnLimit:      connLimit,
+		BufSz:          bufSz,
+		LogFrame:       logFrame}
 	if len(password) > 0 {
 		router.Password = &password
 	}
@@ -66,6 +69,7 @@ func NewRouter(iface *net.Interface, name PeerName, password []byte, connLimit i
 	router.Peers.FetchWithDefault(router.Ourself.Peer)
 	router.Routes = NewRoutes(router.Ourself.Peer, router.Peers)
 	router.ConnectionMaker = NewConnectionMaker(router.Ourself, router.Peers)
+	router.TopologyGossip = router.NewGossip("topology", router)
 	return router
 }
 
@@ -348,3 +352,39 @@ func (router *Router) handleUDPPacketFunc(dec *EthernetDecoder, po PacketSink) F
 		return nil
 	}
 }
+
+// Gossiper methods - the Router is the topology Gossiper
+
+func (router *Router) OnGossipUnicast(sender PeerName, msg []byte) error {
+	return fmt.Errorf("unexpected topology gossip unicast: %v", msg)
+}
+
+func (router *Router) OnGossipBroadcast(msg []byte) error {
+	return fmt.Errorf("unexpected topology gossip broadcast: %v", msg)
+}
+
+// Return state of everything we know; intended to be called periodically
+func (router *Router) Gossip() []byte {
+	return router.Peers.EncodeAllPeers()
+}
+
+// merge in state and return "everything new I've just learnt",
+// or nil if nothing in the received message was new
+func (router *Router) OnGossip(buf []byte) ([]byte, error) {
+	newUpdate, err := router.Peers.ApplyUpdate(buf)
+	if _, ok := err.(UnknownPeersError); err != nil && ok {
+		// That update contained a peer we didn't know about; we
+		// ignore this; eventually we should receive an update
+		// containing a complete topology.
+		return nil, nil
+	}
+	if err != nil {
+		return nil, err
+	}
+	if len(newUpdate) == 0 {
+		return nil, nil
+	}
+	router.ConnectionMaker.Refresh()
+	router.Routes.Recalculate()
+	return newUpdate, nil
+}

router/utils.go

@@ -1,8 +1,11 @@
 package router
 
 import (
+	"bytes"
 	"crypto/rand"
+	"encoding/gob"
 	"fmt"
+	"hash/fnv"
 	"log"
 	"net"
 )
@@ -74,6 +77,21 @@ func randUint64() (r uint64) {
 	return
 }
 
+func hash(s string) uint32 {
+	h := fnv.New32a()
+	h.Write([]byte(s))
+	return h.Sum32()
+}
+
+func GobEncode(items ...interface{}) []byte {
+	buf := new(bytes.Buffer)
+	enc := gob.NewEncoder(buf)
+	for _, i := range items {
+		checkFatal(enc.Encode(i))
+	}
+	return buf.Bytes()
+}
+
 func macint(mac net.HardwareAddr) (r uint64) {
 	for _, b := range mac {
 		r <<= 8

site/how-it-works.md

@@ -105,38 +105,40 @@ other peers. Weave peers communicate their knowledge of the topology
 information on to their own neighbours, and so on, until the entire
 network knows about any change.
 
-Topology is communicated over the TCP links between peers. There are
-two message types:
+Topology is communicated over the TCP links between peers, using a
+Gossip mechanism.  Topology messages are sent by a peer...
 
-##### FetchAll
-This carries no payload. The receiver responds with the entire
-topology model as the receiver has it. FetchAll is sent:
+- when a connection has been established; the entire topology is sent
+  to the remote peer, and an incremental update, containing
+  information on just the two peers at the ends of the connection, is
+  sent to all neighbours
+- when a connection has been torn down; an update containing just
+  information about the local peer is sent to all neighbours
+- periodically, on a timer, in case someone has missed an update.
 
- * when an update is received that references a peer that the receiver
-   does not know
- * periodically, on a timer, in case any updates have been missed
-
-##### Update
-This carries a topology payload. This is sent:
-
-  * upon receipt of a FetchAll message, as above,
-  * when a connection is established - the update contains the two
-peers that have just connected
-  * when a connection is deleted - the update contains just the peer
-that lost the connection
-
-The receiver merges the update with its own topology model, adding
-peers hitherto unknown to it, and updating peers for which the update
-contains a more recent version than known to it. If there were any
-such new/updated peers, then an improved update containing them is
-sent out on all connections.
+The receiver of a topology update merges that update with its own
+topology model, adding peers hitherto unknown to it, and updating
+peers for which the update contains a more recent version than known
+to it. If there were any such new/updated peers, then an improved
+update containing them is sent out on all connections.
 
 If the update mentions a peer that the receiver does not know, then
-the entire update is rejected and the receiver sends a FetchAll
-message back to the sender.
+the entire update is ignored.
 
 #### Message details
-A topology update message is laid out like this:
+Every gossip message is structured as follows:
+
+    +-----------------------------------+
+    | 1-byte message type - Gossip      |
+    +-----------------------------------+
+    | 4-byte Gossip channel - Topology  |
+    +-----------------------------------+
+    | Peer Name of source               |
+    +-----------------------------------+
+    | Gossip payload (topology update)  |
+    +-----------------------------------+
+
+The topology update payload is laid out like this:
 
     +-----------------------------------+
     | Peer 1: Name                      |
@@ -183,7 +185,7 @@ no longer has any connections within the network, it drops all
 knowledge of that second peer.
 
 #### Out-of-date topology
-The peer-to-peer passing of updates is not instantaneous, so it is
+The peer-to-peer gossiping of updates is not instantaneous, so it is
 very possible for a node elsewhere in the network to have an
 out-of-date view.