diff --git a/router/connection.go b/router/connection.go index 994874b7..0469c06c 100644 --- a/router/connection.go +++ b/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) } diff --git a/router/gossip.go b/router/gossip.go new file mode 100644 index 00000000..5c374221 --- /dev/null +++ b/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...)...) +} diff --git a/router/local_peer.go b/router/local_peer.go index 93f13ca9..6df816aa 100644 --- a/router/local_peer.go +++ b/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 } diff --git a/router/protocol.go b/router/protocol.go index 2218d813..7291f35c 100644 --- a/router/protocol.go +++ b/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 { diff --git a/router/router.go b/router/router.go index 8ed59c28..d54a24d0 100644 --- a/router/router.go +++ b/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 } diff --git a/router/utils.go b/router/utils.go index 7f0a2b3b..0a6fe57c 100644 --- a/router/utils.go +++ b/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