Basic gossip implementation

router/connection.go

@@ -603,6 +603,12 @@ func (conn *LocalConnection) handleProtocolMsg(tag ProtocolTag, payload []byte)
 		}
 	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/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/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
 }

router/protocol.go

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

router/router.go

@@ -24,6 +24,7 @@ type Router struct {
 	Peers           *Peers
 	Routes          *Routes
 	ConnectionMaker *ConnectionMaker
+	GossipChannels  map[uint32]*GossipChannel
 	UDPListener     *net.UDPConn
 	Password        *[]byte
 	ConnLimit       int
@@ -46,10 +47,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
 	}

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