From 94220f22b2b04a87f4404b8fe6549402027c33ad Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Sun, 14 Jun 2015 08:47:51 +0100 Subject: [PATCH] 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) }