diff --git a/router/connection.go b/router/connection.go index a42f04ce..a17a6fea 100644 --- a/router/connection.go +++ b/router/connection.go @@ -47,6 +47,7 @@ type LocalConnection struct { stackFrag bool effectivePMTU int SessionKey *[32]byte + heartbeatTCP *time.Ticker heartbeatTimeout *time.Timer heartbeatFrame *ForwardedFrame heartbeat *time.Ticker @@ -339,6 +340,7 @@ func (conn *LocalConnection) run(actionChan <-chan ConnectionAction, finished ch // goroutine, at least not without some synchronisation. func (conn *LocalConnection) initHeartbeats() error { + conn.heartbeatTCP = time.NewTicker(TCPHeartbeat) conn.heartbeatTimeout = time.NewTimer(HeartbeatTimeout) heartbeatFrameBytes := make([]byte, EthernetOverhead+8) binary.BigEndian.PutUint64(heartbeatFrameBytes[EthernetOverhead:], conn.uid) @@ -360,6 +362,8 @@ func (conn *LocalConnection) actorLoop(actionChan <-chan ConnectionAction) (err break } err = action() + case <-conn.heartbeatTCP.C: + err = conn.sendSimpleProtocolMsg(ProtocolHeartbeat) case <-conn.heartbeatTimeout.C: err = fmt.Errorf("timed out waiting for UDP heartbeat") case <-tickerChan(conn.heartbeat): @@ -386,6 +390,7 @@ func (conn *LocalConnection) shutdown() { conn.heartbeatTimeout.Stop() } + stopTicker(conn.heartbeatTCP) stopTicker(conn.heartbeat) stopTicker(conn.fragTest) @@ -439,6 +444,7 @@ func (conn *LocalConnection) receiveTCP(decoder *gob.Decoder) { func (conn *LocalConnection) handleProtocolMsg(tag ProtocolTag, payload []byte) error { switch tag { + case ProtocolHeartbeat: case ProtocolConnectionEstablished: // We sent fast heartbeats to the remote peer, which has now // received at least one of them and told us via this message. @@ -469,7 +475,7 @@ func (conn *LocalConnection) handleProtocolMsg(tag ProtocolTag, payload []byte) } func (conn *LocalConnection) extendReadDeadline() { - conn.TCPConn.SetReadDeadline(time.Now().Add(ReadTimeout)) + conn.TCPConn.SetReadDeadline(time.Now().Add(TCPHeartbeat * 2)) } func (conn *LocalConnection) sendFastHeartbeats() error { diff --git a/router/consts.go b/router/consts.go index 4d8f312a..0aa13fd6 100644 --- a/router/consts.go +++ b/router/consts.go @@ -16,10 +16,10 @@ const ( UDPNonceSendAt = 8192 FragTestSize = 60001 PMTUDiscoverySize = 60000 + TCPHeartbeat = 30 * time.Second FastHeartbeat = 500 * time.Millisecond SlowHeartbeat = 10 * time.Second FragTestInterval = 5 * time.Minute - ReadTimeout = 1 * time.Minute 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 96f4abe8..c0cfcde6 100644 --- a/router/gossip.go +++ b/router/gossip.go @@ -105,7 +105,7 @@ func (router *Router) NewGossip(channelName string, g Gossiper) Gossip { func (router *Router) SendAllGossip() { for _, channel := range router.GossipChannels { - channel.Send(channel.gossiper.Gossip()) + channel.Send(router.Ourself.Name, channel.gossiper.Gossip()) } } @@ -167,7 +167,7 @@ func (c *GossipChannel) deliverBroadcast(srcName PeerName, _ []byte, dec *gob.De return c.relayBroadcast(srcName, data) } -func (c *GossipChannel) deliver(_ PeerName, _ []byte, dec *gob.Decoder) error { +func (c *GossipChannel) deliver(srcName PeerName, _ []byte, dec *gob.Decoder) error { var payload []byte if err := dec.Decode(&payload); err != nil { return err @@ -175,26 +175,48 @@ func (c *GossipChannel) deliver(_ PeerName, _ []byte, dec *gob.Decoder) error { if data, err := c.gossiper.OnGossip(payload); err != nil { return err } else if data != nil { - c.Send(data) + c.Send(srcName, data) } return nil } -func (c *GossipChannel) Send(data GossipData) { - connections := c.ourself.Connections() // do this outside the lock so they don't nest - retainedSenders := make(connectionSenders) +func (c *GossipChannel) Send(srcName PeerName, data GossipData) { + // do this outside the lock below so we avoid lock nesting + c.ourself.Router.Routes.EnsureRecalculated() + selectedConnections := make(ConnectionSet) + for name := range c.ourself.Router.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() - for conn := range connections { + // 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) - retainedSenders[conn] = c.senders[conn] - delete(c.senders, conn) } - // stop any senders for connections that are gone - for _, sender := range c.senders { - sender.Stop() - } - c.senders = retainedSenders } func (c *GossipChannel) SendDown(conn Connection, data GossipData) { diff --git a/router/protocol.go b/router/protocol.go index b9bb9f29..0fc548c6 100644 --- a/router/protocol.go +++ b/router/protocol.go @@ -8,7 +8,8 @@ const ( type ProtocolTag byte const ( - ProtocolConnectionEstablished ProtocolTag = iota + ProtocolHeartbeat ProtocolTag = iota + ProtocolConnectionEstablished ProtocolFragmentationReceived ProtocolStartFragmentationTest ProtocolNonce diff --git a/router/routes.go b/router/routes.go index 3da3d1f2..1e854223 100644 --- a/router/routes.go +++ b/router/routes.go @@ -3,6 +3,7 @@ package router import ( "bytes" "fmt" + "math" "sync" ) @@ -77,6 +78,36 @@ func (routes *Routes) BroadcastAll(name PeerName) []PeerName { return hops } +// Choose min(log2(n_peers), n_neighbouring_peers) neighbours, with a +// random distribution that is topology-sensitive, favouring +// neighbours at the end of "bottleneck links". We determine the +// latter based on the unicast routing table. If a neighbour appears +// as the value more frequently than others - meaning that we reach a +// higher proportion of peers via that neighbour than other neighbours +// - then it is chosen with a higher probability. +// +// Note that we choose log2(n_peers) *neighbours*, not +// peers. Consequently, on sparsely connected peers this function +// returns a higher proportion of neighbours than elsewhere. In +// extremis, on peers with fewer than log2(n_peers) neighbours, all +// neighbours are returned. +func (routes *Routes) RandomNeighbours(except PeerName) PeerNameSet { + res := make(PeerNameSet) + routes.RLock() + defer routes.RUnlock() + count := int(math.Log2(float64(len(routes.unicastAll)))) + // depends on go's random map iteration + for _, dst := range routes.unicastAll { + if dst != UnknownPeerName && dst != except { + res[dst] = void + if len(res) >= count { + break + } + } + } + return res +} + func (routes *Routes) String() string { var buf bytes.Buffer routes.RLock()