From cea85eb05b473ba71b594d13bb9ca0863ce54876 Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Sun, 14 Jun 2015 10:35:51 +0100 Subject: [PATCH] cosmetic: move GossipChannel code into separate file --- router/consts.go | 1 + router/gossip.go | 226 --------------------------------------- router/gossip_channel.go | 224 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 225 insertions(+), 226 deletions(-) create mode 100644 router/gossip_channel.go diff --git a/router/consts.go b/router/consts.go index af6ad0fe..1acfc18b 100644 --- a/router/consts.go +++ b/router/consts.go @@ -19,6 +19,7 @@ const ( FastHeartbeat = 500 * time.Millisecond SlowHeartbeat = 10 * time.Second FragTestInterval = 5 * time.Minute + GossipInterval = 30 * time.Second PMTUVerifyAttempts = 8 PMTUVerifyTimeout = 10 * time.Millisecond // gets doubled with every attempt MaxDuration = time.Duration(math.MaxInt64) diff --git a/router/gossip.go b/router/gossip.go index 5e9ab203..baaeb1b5 100644 --- a/router/gossip.go +++ b/router/gossip.go @@ -1,16 +1,5 @@ package router -import ( - "bytes" - "encoding/gob" - "fmt" - "log" - "sync" - "time" -) - -const GossipInterval = 30 * time.Second - type GossipData interface { Encode() []byte Merge(GossipData) @@ -95,31 +84,8 @@ func (sender *GossipSender) Stop() { close(sender.cell) } -type connectionSenders map[Connection]*GossipSender -type peerSenders map[PeerName]*GossipSender - -type GossipChannel struct { - sync.Mutex - name string - ourself *LocalPeer - routes *Routes - gossiper Gossiper - senders connectionSenders - broadcasters peerSenders -} - type GossipChannels map[string]*GossipChannel -func NewGossipChannel(channelName string, ourself *LocalPeer, routes *Routes, g Gossiper) *GossipChannel { - return &GossipChannel{ - name: channelName, - ourself: ourself, - routes: routes, - gossiper: g, - senders: make(connectionSenders), - broadcasters: make(peerSenders)} -} - func (router *Router) NewGossip(channelName string, g Gossiper) Gossip { channel := NewGossipChannel(channelName, router.Ourself, router.Routes, g) router.GossipChannels[channelName] = channel @@ -142,198 +108,6 @@ func (router *Router) SendAllGossipDown(conn Connection) { } } -func (router *Router) handleGossip(tag ProtocolTag, payload []byte) error { - decoder := gob.NewDecoder(bytes.NewReader(payload)) - var channelName string - if err := decoder.Decode(&channelName); err != nil { - return err - } - channel, found := router.GossipChannels[channelName] - if !found { - return fmt.Errorf("[gossip] received unknown channel with name %s", channelName) - } - var srcName PeerName - if err := decoder.Decode(&srcName); err != nil { - return err - } - switch tag { - case ProtocolGossipUnicast: - return channel.deliverUnicast(srcName, payload, decoder) - case ProtocolGossipBroadcast: - return channel.deliverBroadcast(srcName, payload, decoder) - case ProtocolGossip: - return channel.deliver(srcName, payload, decoder) - } - return nil -} - -func (c *GossipChannel) deliverUnicast(srcName PeerName, origPayload []byte, dec *gob.Decoder) error { - var destName PeerName - if err := dec.Decode(&destName); err != nil { - return err - } - if c.ourself.Name != destName { - return c.relayUnicast(destName, origPayload) - } - var payload []byte - if err := dec.Decode(&payload); err != nil { - return err - } - return c.gossiper.OnGossipUnicast(srcName, payload) -} - -func (c *GossipChannel) deliverBroadcast(srcName PeerName, _ []byte, dec *gob.Decoder) error { - var payload []byte - if err := dec.Decode(&payload); err != nil { - return err - } - data, err := c.gossiper.OnGossipBroadcast(payload) - if err != nil || data == nil { - return err - } - return c.relayBroadcast(srcName, data) -} - -func (c *GossipChannel) deliver(srcName PeerName, _ []byte, dec *gob.Decoder) error { - var payload []byte - if err := dec.Decode(&payload); err != nil { - return err - } - if data, err := c.gossiper.OnGossip(payload); err != nil { - return err - } else if data != nil { - c.Send(srcName, data) - } - return nil -} - -func (c *GossipChannel) Send(srcName PeerName, data GossipData) { - // do this outside the lock below so we avoid lock nesting - c.routes.EnsureRecalculated() - selectedConnections := make(ConnectionSet) - for name := range c.routes.RandomNeighbours(srcName) { - if conn, found := c.ourself.ConnectionTo(name); found { - selectedConnections[conn] = void - } - } - if len(selectedConnections) == 0 { - return - } - connections := c.ourself.Connections() - c.Lock() - defer c.Unlock() - // GC - randomly (courtesy of go's map iterator) pick some - // existing entries and stop&remove them if the associated - // connection is no longer active. We stop as soon as we - // encounter a valid entry; the idea being that when there is - // little or no garbage then this executes close to O(1)[1], - // whereas when there is lots of garbage we remove it quickly. - // - // [1] TODO Unfortunately, due to the desire to avoid nested - // locks, instead of simply invoking Peer.ConnectionTo(name) - // below, we have that Peer.Connections() invocation above. That - // is O(n_our_connections) at best. - for conn, sender := range c.senders { - if _, found := connections[conn]; !found { - delete(c.senders, conn) - sender.Stop() - } else { - break - } - } - for conn := range selectedConnections { - c.sendDown(conn, data) - } -} - -func (c *GossipChannel) SendDown(conn Connection, data GossipData) { - c.Lock() - c.sendDown(conn, data) - c.Unlock() -} - -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) - }) - c.senders[conn] = sender - sender.Start() - } - sender.Send(data) -} - -func (c *GossipChannel) GossipUnicast(dstPeerName PeerName, msg []byte) error { - return c.relayUnicast(dstPeerName, GobEncode(c.name, c.ourself.Name, dstPeerName, msg)) -} - -func (c *GossipChannel) GossipBroadcast(update GossipData) error { - return c.relayBroadcast(c.ourself.Name, update) -} - -func (c *GossipChannel) relayUnicast(dstPeerName PeerName, buf []byte) error { - if relayPeerName, found := c.routes.UnicastAll(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, buf}) - } - return nil -} - -func (c *GossipChannel) relayBroadcast(srcName PeerName, update GossipData) error { - names := c.routes.PeerNames() // do this outside the lock so they don't nest - c.Lock() - defer c.Unlock() - // GC - randomly (courtesy of go's map iterator) pick some - // existing broadcasters and stop&remove them if their source peer - // is unknown. We stop as soon as we encounter a valid entry; the - // idea being that when there is little or no garbage then this - // executes close to O(1)[1], whereas when there is lots of - // garbage we remove it quickly. - // - // [1] TODO Unfortunately, due to the desire to avoid nested - // locks, instead of simply invoking Peers.Fetch(name) below, we - // have that Peers.Names() invocation above. That is O(n_peers) at - // best. - for name, broadcaster := range c.broadcasters { - if _, found := names[name]; !found { - delete(c.broadcasters, name) - broadcaster.Stop() - } else { - break - } - } - broadcaster, found := c.broadcasters[srcName] - if !found { - broadcaster = NewGossipSender(func(pending GossipData) { c.sendBroadcast(srcName, pending) }) - c.broadcasters[srcName] = broadcaster - broadcaster.Start() - } - broadcaster.Send(update) - return nil -} - -func (c *GossipChannel) sendBroadcast(srcName PeerName, update GossipData) { - c.routes.EnsureRecalculated() - nextHops := c.routes.BroadcastAll(srcName) - 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) - } -} - -func (c *GossipChannel) log(args ...interface{}) { - log.Println(append(append([]interface{}{}, "[gossip "+c.name+"]:"), args...)...) -} - // for testing func (router *Router) sendPendingGossip() bool { diff --git a/router/gossip_channel.go b/router/gossip_channel.go new file mode 100644 index 00000000..31fa06ed --- /dev/null +++ b/router/gossip_channel.go @@ -0,0 +1,224 @@ +package router + +import ( + "bytes" + "encoding/gob" + "fmt" + "log" + "sync" +) + +type GossipChannel struct { + sync.Mutex + name string + ourself *LocalPeer + routes *Routes + gossiper Gossiper + senders connectionSenders + broadcasters peerSenders +} + +type connectionSenders map[Connection]*GossipSender +type peerSenders map[PeerName]*GossipSender + +func NewGossipChannel(channelName string, ourself *LocalPeer, routes *Routes, g Gossiper) *GossipChannel { + return &GossipChannel{ + name: channelName, + ourself: ourself, + routes: routes, + gossiper: g, + senders: make(connectionSenders), + broadcasters: make(peerSenders)} +} + +func (router *Router) handleGossip(tag ProtocolTag, payload []byte) error { + decoder := gob.NewDecoder(bytes.NewReader(payload)) + var channelName string + if err := decoder.Decode(&channelName); err != nil { + return err + } + channel, found := router.GossipChannels[channelName] + if !found { + return fmt.Errorf("[gossip] received unknown channel with name %s", channelName) + } + var srcName PeerName + if err := decoder.Decode(&srcName); err != nil { + return err + } + switch tag { + case ProtocolGossipUnicast: + return channel.deliverUnicast(srcName, payload, decoder) + case ProtocolGossipBroadcast: + return channel.deliverBroadcast(srcName, payload, decoder) + case ProtocolGossip: + return channel.deliver(srcName, payload, decoder) + } + return nil +} + +func (c *GossipChannel) deliverUnicast(srcName PeerName, origPayload []byte, dec *gob.Decoder) error { + var destName PeerName + if err := dec.Decode(&destName); err != nil { + return err + } + if c.ourself.Name != destName { + return c.relayUnicast(destName, origPayload) + } + var payload []byte + if err := dec.Decode(&payload); err != nil { + return err + } + return c.gossiper.OnGossipUnicast(srcName, payload) +} + +func (c *GossipChannel) deliverBroadcast(srcName PeerName, _ []byte, dec *gob.Decoder) error { + var payload []byte + if err := dec.Decode(&payload); err != nil { + return err + } + data, err := c.gossiper.OnGossipBroadcast(payload) + if err != nil || data == nil { + return err + } + return c.relayBroadcast(srcName, data) +} + +func (c *GossipChannel) deliver(srcName PeerName, _ []byte, dec *gob.Decoder) error { + var payload []byte + if err := dec.Decode(&payload); err != nil { + return err + } + if data, err := c.gossiper.OnGossip(payload); err != nil { + return err + } else if data != nil { + c.Send(srcName, data) + } + return nil +} + +func (c *GossipChannel) Send(srcName PeerName, data GossipData) { + // do this outside the lock below so we avoid lock nesting + c.routes.EnsureRecalculated() + selectedConnections := make(ConnectionSet) + for name := range c.routes.RandomNeighbours(srcName) { + if conn, found := c.ourself.ConnectionTo(name); found { + selectedConnections[conn] = void + } + } + if len(selectedConnections) == 0 { + return + } + connections := c.ourself.Connections() + c.Lock() + defer c.Unlock() + // GC - randomly (courtesy of go's map iterator) pick some + // existing entries and stop&remove them if the associated + // connection is no longer active. We stop as soon as we + // encounter a valid entry; the idea being that when there is + // little or no garbage then this executes close to O(1)[1], + // whereas when there is lots of garbage we remove it quickly. + // + // [1] TODO Unfortunately, due to the desire to avoid nested + // locks, instead of simply invoking Peer.ConnectionTo(name) + // below, we have that Peer.Connections() invocation above. That + // is O(n_our_connections) at best. + for conn, sender := range c.senders { + if _, found := connections[conn]; !found { + delete(c.senders, conn) + sender.Stop() + } else { + break + } + } + for conn := range selectedConnections { + c.sendDown(conn, data) + } +} + +func (c *GossipChannel) SendDown(conn Connection, data GossipData) { + c.Lock() + c.sendDown(conn, data) + c.Unlock() +} + +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) + }) + c.senders[conn] = sender + sender.Start() + } + sender.Send(data) +} + +func (c *GossipChannel) GossipUnicast(dstPeerName PeerName, msg []byte) error { + return c.relayUnicast(dstPeerName, GobEncode(c.name, c.ourself.Name, dstPeerName, msg)) +} + +func (c *GossipChannel) GossipBroadcast(update GossipData) error { + return c.relayBroadcast(c.ourself.Name, update) +} + +func (c *GossipChannel) relayUnicast(dstPeerName PeerName, buf []byte) error { + if relayPeerName, found := c.routes.UnicastAll(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, buf}) + } + return nil +} + +func (c *GossipChannel) relayBroadcast(srcName PeerName, update GossipData) error { + names := c.routes.PeerNames() // do this outside the lock so they don't nest + c.Lock() + defer c.Unlock() + // GC - randomly (courtesy of go's map iterator) pick some + // existing broadcasters and stop&remove them if their source peer + // is unknown. We stop as soon as we encounter a valid entry; the + // idea being that when there is little or no garbage then this + // executes close to O(1)[1], whereas when there is lots of + // garbage we remove it quickly. + // + // [1] TODO Unfortunately, due to the desire to avoid nested + // locks, instead of simply invoking Peers.Fetch(name) below, we + // have that Peers.Names() invocation above. That is O(n_peers) at + // best. + for name, broadcaster := range c.broadcasters { + if _, found := names[name]; !found { + delete(c.broadcasters, name) + broadcaster.Stop() + } else { + break + } + } + broadcaster, found := c.broadcasters[srcName] + if !found { + broadcaster = NewGossipSender(func(pending GossipData) { c.sendBroadcast(srcName, pending) }) + c.broadcasters[srcName] = broadcaster + broadcaster.Start() + } + broadcaster.Send(update) + return nil +} + +func (c *GossipChannel) sendBroadcast(srcName PeerName, update GossipData) { + c.routes.EnsureRecalculated() + nextHops := c.routes.BroadcastAll(srcName) + 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) + } +} + +func (c *GossipChannel) log(args ...interface{}) { + log.Println(append(append([]interface{}{}, "[gossip "+c.name+"]:"), args...)...) +}