From 66f14099edee4dead78d678872e3a975381747af Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Tue, 29 Sep 2015 18:10:35 +0100 Subject: [PATCH 1/5] prevent ConnectionMaker.ConnectionTerminated race Get the ConnectionMaker to maintain its own knowledge of the local connections, based on events emitted in the connection lifecycle. Fixes #1478. --- router/connection.go | 3 +- router/connection_maker.go | 56 +++++++++++++++++++++++++------------- 2 files changed, 39 insertions(+), 20 deletions(-) diff --git a/router/connection.go b/router/connection.go index e3967ec6..33758539 100644 --- a/router/connection.go +++ b/router/connection.go @@ -212,6 +212,7 @@ func (conn *LocalConnection) run(actionChan <-chan ConnectionAction, finished ch if err = conn.Router.Ourself.AddConnection(conn); err != nil { return } + conn.Router.ConnectionMaker.ConnectionCreated(conn) // SetListener has the side-effect of telling the forwarder // that the connection is confirmed. This comes after @@ -360,7 +361,7 @@ func (conn *LocalConnection) shutdown(err error) { conn.forwarder.Stop() } - conn.Router.ConnectionMaker.ConnectionTerminated(conn.remoteTCPAddr, err) + conn.Router.ConnectionMaker.ConnectionTerminated(conn, err) } func (conn *LocalConnection) forwarderCrypto() *OverlayCrypto { diff --git a/router/connection_maker.go b/router/connection_maker.go index 9ade66ae..5cb11771 100644 --- a/router/connection_maker.go +++ b/router/connection_maker.go @@ -20,6 +20,7 @@ type ConnectionMaker struct { port int discovery bool targets map[string]*Target + connections map[string]Connection directPeers peerAddrs actionChan chan<- ConnectionMakerAction } @@ -43,6 +44,7 @@ func NewConnectionMaker(ourself *LocalPeer, peers *Peers, port int, discovery bo discovery: discovery, directPeers: peerAddrs{}, targets: make(map[string]*Target), + connections: make(map[string]Connection), actionChan: actionChan} go cm.queryLoop(actionChan) return cm @@ -88,12 +90,26 @@ func (cm *ConnectionMaker) ForgetConnections(peers []string) { } } -func (cm *ConnectionMaker) ConnectionTerminated(address string, err error) { +func (cm *ConnectionMaker) ConnectionAborted(address string, err error) { cm.actionChan <- func() bool { - if target, found := cm.targets[address]; found { - target.attempting = false - target.lastError = err - target.retry() + cm.retry(address, err) + return true + } +} + +func (cm *ConnectionMaker) ConnectionCreated(conn Connection) { + cm.actionChan <- func() bool { + cm.connections[conn.RemoteTCPAddr()] = conn + return false + } +} + +func (cm *ConnectionMaker) ConnectionTerminated(conn Connection, err error) { + cm.actionChan <- func() bool { + address := conn.RemoteTCPAddr() + delete(cm.connections, address) + if conn.Outbound() { + cm.retry(address, err) } return true } @@ -123,13 +139,10 @@ func (cm *ConnectionMaker) checkStateAndAttemptConnections() time.Duration { validTarget = make(map[string]struct{}) directTarget = make(map[string]struct{}) ) - // Copy the set of things we are connected to, so we can access - // them without locking. Also clear out any entries in cm.targets - // for existing connections. - ourConnectedPeers, ourConnectedTargets, ourInboundIPs := cm.ourConnections() + ourConnectedPeers, ourInboundIPs := cm.ourConnections() addTarget := func(address string) { - if _, connected := ourConnectedTargets[address]; connected { + if _, connected := cm.connections[address]; connected { return } validTarget[address] = void @@ -170,17 +183,14 @@ func (cm *ConnectionMaker) checkStateAndAttemptConnections() time.Duration { return cm.connectToTargets(validTarget, directTarget) } -func (cm *ConnectionMaker) ourConnections() (PeerNameSet, map[string]struct{}, map[string]struct{}) { +func (cm *ConnectionMaker) ourConnections() (PeerNameSet, map[string]struct{}) { var ( - ourConnectedPeers = make(PeerNameSet) - ourConnectedTargets = make(map[string]struct{}) - ourInboundIPs = make(map[string]struct{}) + ourConnectedPeers = make(PeerNameSet) + ourInboundIPs = make(map[string]struct{}) ) - for conn := range cm.ourself.Connections() { - address := conn.RemoteTCPAddr() + for address, conn := range cm.connections { delete(cm.targets, address) ourConnectedPeers[conn.Remote().Name] = void - ourConnectedTargets[address] = void if conn.Outbound() { continue } @@ -188,7 +198,7 @@ func (cm *ConnectionMaker) ourConnections() (PeerNameSet, map[string]struct{}, m ourInboundIPs[ip] = void } } - return ourConnectedPeers, ourConnectedTargets, ourInboundIPs + return ourConnectedPeers, ourInboundIPs } func (cm *ConnectionMaker) addPeerTargets(ourConnectedPeers PeerNameSet, addTarget func(string)) { @@ -250,7 +260,15 @@ func (cm *ConnectionMaker) attemptConnection(address string, acceptNewPeer bool) log.Printf("->[%s] attempting connection", address) if err := cm.ourself.CreateConnection(address, acceptNewPeer); err != nil { log.Errorf("->[%s] error during connection attempt: %v", address, err) - cm.ConnectionTerminated(address, err) + cm.ConnectionAborted(address, err) + } +} + +func (cm *ConnectionMaker) retry(address string, err error) { + if target, found := cm.targets[address]; found { + target.attempting = false + target.lastError = err + target.retry() } } From f8b074f69694cf746ec1fdb2e35fe133fb651711 Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Wed, 30 Sep 2015 11:03:20 +0100 Subject: [PATCH 2/5] a better way to clean up after established connections ...than 3db4e86 --- router/connection_maker.go | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/router/connection_maker.go b/router/connection_maker.go index 5cb11771..1514b5e0 100644 --- a/router/connection_maker.go +++ b/router/connection_maker.go @@ -99,7 +99,11 @@ func (cm *ConnectionMaker) ConnectionAborted(address string, err error) { func (cm *ConnectionMaker) ConnectionCreated(conn Connection) { cm.actionChan <- func() bool { - cm.connections[conn.RemoteTCPAddr()] = conn + address := conn.RemoteTCPAddr() + cm.connections[address] = conn + if conn.Outbound() { + delete(cm.targets, address) + } return false } } @@ -189,7 +193,6 @@ func (cm *ConnectionMaker) ourConnections() (PeerNameSet, map[string]struct{}) { ourInboundIPs = make(map[string]struct{}) ) for address, conn := range cm.connections { - delete(cm.targets, address) ourConnectedPeers[conn.Remote().Name] = void if conn.Outbound() { continue From fc6f3993bda1cd835fc2ad807990559ebf90980e Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Wed, 30 Sep 2015 11:21:57 +0100 Subject: [PATCH 3/5] refactor(ish): obtain local connection status from CM alone --- router/status.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/router/status.go b/router/status.go index c1486c77..40193963 100644 --- a/router/status.go +++ b/router/status.go @@ -171,12 +171,12 @@ func NewLocalConnectionStatusSlice(cm *ConnectionMaker) []LocalConnectionStatus resultChan := make(chan []LocalConnectionStatus, 0) cm.actionChan <- func() bool { var slice []LocalConnectionStatus - for conn := range cm.ourself.Connections() { + for address, conn := range cm.connections { state := "pending" if conn.Established() { state = "established" } - slice = append(slice, LocalConnectionStatus{conn.RemoteTCPAddr(), conn.Outbound(), state, conn.Remote().String()}) + slice = append(slice, LocalConnectionStatus{address, conn.Outbound(), state, conn.Remote().String()}) } for address, target := range cm.targets { var state, info string From e68ca761a9835c305424c165225c28494ba86444 Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Wed, 30 Sep 2015 13:58:08 +0100 Subject: [PATCH 4/5] we no longer need to Refresh() in CM status since CM now gets an event - ConnectionCreated - for new connections. --- router/status.go | 6 ------ 1 file changed, 6 deletions(-) diff --git a/router/status.go b/router/status.go index 40193963..e750efe8 100644 --- a/router/status.go +++ b/router/status.go @@ -162,12 +162,6 @@ func NewBroadcastRouteStatusSlice(routes *Routes) []BroadcastRouteStatus { } func NewLocalConnectionStatusSlice(cm *ConnectionMaker) []LocalConnectionStatus { - // We need to Refresh first in order to clear out any 'attempting' - // connections from cm.targets that have been established since - // the last run of cm.checkStateAndAttemptConnections. These - // entries are harmless but do represent stale state that we do - // not want to report. - cm.Refresh() resultChan := make(chan []LocalConnectionStatus, 0) cm.actionChan <- func() bool { var slice []LocalConnectionStatus From 0159e266313da02faf06d841d2f7d85dabf94ab0 Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Wed, 30 Sep 2015 17:29:13 +0100 Subject: [PATCH 5/5] store connections as a set instead of map-of-remote-addresses ...since it is theoretically possible that the lifecycle events from two consecutive inbound connections from the same remote address overlap. --- router/connection_maker.go | 31 ++++++++++++++++--------------- router/status.go | 4 ++-- 2 files changed, 18 insertions(+), 17 deletions(-) diff --git a/router/connection_maker.go b/router/connection_maker.go index 1514b5e0..4bef62db 100644 --- a/router/connection_maker.go +++ b/router/connection_maker.go @@ -20,7 +20,7 @@ type ConnectionMaker struct { port int discovery bool targets map[string]*Target - connections map[string]Connection + connections map[Connection]struct{} directPeers peerAddrs actionChan chan<- ConnectionMakerAction } @@ -44,7 +44,7 @@ func NewConnectionMaker(ourself *LocalPeer, peers *Peers, port int, discovery bo discovery: discovery, directPeers: peerAddrs{}, targets: make(map[string]*Target), - connections: make(map[string]Connection), + connections: make(map[Connection]struct{}), actionChan: actionChan} go cm.queryLoop(actionChan) return cm @@ -99,10 +99,9 @@ func (cm *ConnectionMaker) ConnectionAborted(address string, err error) { func (cm *ConnectionMaker) ConnectionCreated(conn Connection) { cm.actionChan <- func() bool { - address := conn.RemoteTCPAddr() - cm.connections[address] = conn + cm.connections[conn] = void if conn.Outbound() { - delete(cm.targets, address) + delete(cm.targets, conn.RemoteTCPAddr()) } return false } @@ -110,10 +109,9 @@ func (cm *ConnectionMaker) ConnectionCreated(conn Connection) { func (cm *ConnectionMaker) ConnectionTerminated(conn Connection, err error) { cm.actionChan <- func() bool { - address := conn.RemoteTCPAddr() - delete(cm.connections, address) + delete(cm.connections, conn) if conn.Outbound() { - cm.retry(address, err) + cm.retry(conn.RemoteTCPAddr(), err) } return true } @@ -143,10 +141,10 @@ func (cm *ConnectionMaker) checkStateAndAttemptConnections() time.Duration { validTarget = make(map[string]struct{}) directTarget = make(map[string]struct{}) ) - ourConnectedPeers, ourInboundIPs := cm.ourConnections() + ourConnectedPeers, ourConnectedTargets, ourInboundIPs := cm.ourConnections() addTarget := func(address string) { - if _, connected := cm.connections[address]; connected { + if _, connected := ourConnectedTargets[address]; connected { return } validTarget[address] = void @@ -187,13 +185,16 @@ func (cm *ConnectionMaker) checkStateAndAttemptConnections() time.Duration { return cm.connectToTargets(validTarget, directTarget) } -func (cm *ConnectionMaker) ourConnections() (PeerNameSet, map[string]struct{}) { +func (cm *ConnectionMaker) ourConnections() (PeerNameSet, map[string]struct{}, map[string]struct{}) { var ( - ourConnectedPeers = make(PeerNameSet) - ourInboundIPs = make(map[string]struct{}) + ourConnectedPeers = make(PeerNameSet) + ourConnectedTargets = make(map[string]struct{}) + ourInboundIPs = make(map[string]struct{}) ) - for address, conn := range cm.connections { + for conn := range cm.connections { + address := conn.RemoteTCPAddr() ourConnectedPeers[conn.Remote().Name] = void + ourConnectedTargets[address] = void if conn.Outbound() { continue } @@ -201,7 +202,7 @@ func (cm *ConnectionMaker) ourConnections() (PeerNameSet, map[string]struct{}) { ourInboundIPs[ip] = void } } - return ourConnectedPeers, ourInboundIPs + return ourConnectedPeers, ourConnectedTargets, ourInboundIPs } func (cm *ConnectionMaker) addPeerTargets(ourConnectedPeers PeerNameSet, addTarget func(string)) { diff --git a/router/status.go b/router/status.go index e750efe8..a2e5ad56 100644 --- a/router/status.go +++ b/router/status.go @@ -165,12 +165,12 @@ func NewLocalConnectionStatusSlice(cm *ConnectionMaker) []LocalConnectionStatus resultChan := make(chan []LocalConnectionStatus, 0) cm.actionChan <- func() bool { var slice []LocalConnectionStatus - for address, conn := range cm.connections { + for conn := range cm.connections { state := "pending" if conn.Established() { state = "established" } - slice = append(slice, LocalConnectionStatus{address, conn.Outbound(), state, conn.Remote().String()}) + slice = append(slice, LocalConnectionStatus{conn.RemoteTCPAddr(), conn.Outbound(), state, conn.Remote().String()}) } for address, target := range cm.targets { var state, info string