diff --git a/connection.go b/connection.go index eba52584..4545f0a1 100644 --- a/connection.go +++ b/connection.go @@ -221,7 +221,6 @@ func (conn *LocalConnection) handleSetEstablished() error { conn.established = true conn.Unlock() if !old { - conn.Router.ConnectionMaker.ConnectionEstablished(conn.remoteTCPAddr) conn.local.ConnectionEstablished(conn) if err := conn.ensureHeartbeat(false); err != nil { return err diff --git a/connection_maker.go b/connection_maker.go index 77845ce8..b15c7a5d 100644 --- a/connection_maker.go +++ b/connection_maker.go @@ -12,7 +12,7 @@ const ( InitialInterval = 5 * time.Second MaxInterval = 10 * time.Minute MaxAttemptCount = 100 - CMEnsure = iota + CMInitiate = iota CMStatus = iota CMEstablished = iota CMConnFailed = iota @@ -36,24 +36,13 @@ func StartConnectionMaker(router *Router) *ConnectionMaker { func (cm *ConnectionMaker) InitiateConnection(address string) { cm.queryChan <- &ConnectionMakerInteraction{ - Interaction: Interaction{code: CMEnsure}, - acceptAnyPeer: true, - address: address} + Interaction: Interaction{code: CMInitiate}, + isCmdLine: true, + address: address} } func (cm *ConnectionMaker) EnsureConnection(address string) { - address = StripPortFromAddr(address) - cm.queryChan <- &ConnectionMakerInteraction{ - Interaction: Interaction{code: CMEnsure}, - acceptAnyPeer: false, - address: address} -} - -func (cm *ConnectionMaker) ConnectionEstablished(address string) { - address = StripPortFromAddr(address) - cm.queryChan <- &ConnectionMakerInteraction{ - Interaction: Interaction{code: CMEstablished}, - address: address} + // Do nothing - we will figure out what to do later } func (cm *ConnectionMaker) String() string { @@ -84,65 +73,99 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio return } switch { - case query.code == CMEnsure: - cm.addToTargets(query.acceptAnyPeer, query.address) - cm.resetTarget(query.address) + case query.code == CMInitiate: + cm.addToTargets(query.isCmdLine, query.address) tickNow() case query.code == CMStatus: query.resultChan <- cm.status() - case query.code == CMEstablished: - target := cm.targets[query.address] - if target != nil { - target.state = CSEstablished - } else { - log.Fatal("Connection established to unknown address", query.address) - } case query.code == CMConnFailed: target := cm.targets[query.address] - target.state = CSUnconnected - target.tryAfter, target.tryInterval = tryAfter(target.tryInterval) - maybeTick() + if target != nil { + target.state = CSUnconnected + target.tryAfter, target.tryInterval = tryAfter(target.tryInterval) + maybeTick() + } else { + log.Fatal("CMConnFailed unknown address", query.address) + } default: log.Fatal("Unexpected connection maker query:", query) } case now := <-tick: - for address, target := range cm.targets { - if target.state == CSUnconnected && now.After(target.tryAfter) { - if _, found := cm.router.Ourself.ConnectionOn(address); found { - target.state = CSEstablished - } else { - target.attemptCount += 1 - target.state = CSAttempting - go cm.attemptConnection(address, target.acceptAnyPeer) - } - } - } + cm.checkStateAndAttemptConnections(now) tick = nil maybeTick() } } } -func (cm *ConnectionMaker) addToTargets(acceptAnyPeer bool, address string) { - target := cm.targets[address] - if target == nil { - target = &Target{ - state: CSUnconnected, - acceptAnyPeer: acceptAnyPeer, +func (cm *ConnectionMaker) checkStateAndAttemptConnections(now time.Time) { + ourself := cm.router.Ourself + count_unconnected_commandline := 0 + // Any targets that are now connected, we don't need to attempt any more + for address, target := range cm.targets { + if _, found := ourself.ConnectionOn(address); found { + if target.isCmdLine { + // We need to keep hold of targets supplied on the + // command-line, because that info isn't + // tracked anywhere else + target.state = CSEstablished + target.tryInterval = InitialInterval + } else { + delete(cm.targets, address) + } + } else if target.isCmdLine { + if target.state != CSAttempting { + target.state = CSUnconnected + } + count_unconnected_commandline += 1 + } + } + + // Look for peers that we don't have a connection to. + count_unconnected_peers := 0 + cm.router.Peers.ForEach(func(name PeerName, peer *Peer) { + for peer2, conn := range peer.connections { + if _, found := ourself.ConnectionTo(peer2); !found && + peer2 != ourself.Name { + log.Println("Unconnected peer:", peer2) + count_unconnected_peers += 1 + // peer2 is a peer that someone else knows about, but we don't have a connection to. + address := conn.RemoteTCPAddr() + if host, port, err := ExtractHostPort(address); err == nil { + if port != Port { + cm.addToTargets(false, address) + } + cm.addToTargets(false, host) + } + } + } + }) + + if count_unconnected_commandline == 0 && count_unconnected_peers == 0 { + // We are already connected to everyone we could possibly be connected to; nothing further to do. + return + } + + // Now connect to any targets in scope + for address, target := range cm.targets { + if target.state == CSUnconnected && now.After(target.tryAfter) { + target.attemptCount += 1 + target.state = CSAttempting + go cm.attemptConnection(address, target.isCmdLine) } - cm.targets[address] = target } } -func (cm *ConnectionMaker) resetTarget(address string) { +func (cm *ConnectionMaker) addToTargets(isCmdLine bool, address string) { target := cm.targets[address] - // If we were in the middle of a connection attempt then back-off again - if target.state == CSAttempting { - target.tryAfter, target.tryInterval = tryAfter(target.tryInterval) - } else { + if target == nil { + target = &Target{ + state: CSUnconnected, + isCmdLine: isCmdLine, + } target.tryAfter, target.tryInterval = tryImmediately() + cm.targets[address] = target } - target.state = CSUnconnected } func (cm *ConnectionMaker) status() string { diff --git a/types.go b/types.go index 53178072..e5effe91 100644 --- a/types.go +++ b/types.go @@ -155,8 +155,8 @@ type Topology struct { type ConnectionMakerInteraction struct { Interaction - acceptAnyPeer bool - address string + isCmdLine bool + address string } type ConnectionMaker struct { @@ -169,11 +169,11 @@ type ConnectionState int // Information about an address where we may find a peer type Target struct { - acceptAnyPeer bool // was this address given to us on the command line? - state ConnectionState - tryAfter time.Time // next time to try this address - tryInterval time.Duration // backoff time on next failure - attemptCount int + isCmdLine bool // was this address given to us on the command line? + state ConnectionState + tryAfter time.Time // next time to try this address + tryInterval time.Duration // backoff time on next failure + attemptCount int } // UDPSender interface and implementations diff --git a/utils.go b/utils.go index 8d9a54bc..74dce894 100644 --- a/utils.go +++ b/utils.go @@ -5,6 +5,7 @@ import ( "fmt" "log" "net" + "strconv" ) func checkFatal(e error) { @@ -106,6 +107,20 @@ func StripPortFromAddr(address string) string { return address } +// given an address like '1.2.3.4:567', return the host and port number, +// given an address like '1.2.3.4', return the host and weave's standard port number +func ExtractHostPort(address string) (host string, port int, err error) { + if host, port, err := net.SplitHostPort(address); err == nil { + if portnum, err := strconv.Atoi(port); err == nil { + return host, portnum, nil + } else { + return host, 0, err + } + } else { + return host, Port, nil + } +} + // given an address like '1.2.3.4:567', return the address if it has a port, // otherwise return the address with weave's standard port number func NormalisePeerAddr(peerAddr string) string {