diff --git a/connection.go b/connection.go index 68c337ee..a88d7d42 100644 --- a/connection.go +++ b/connection.go @@ -51,7 +51,7 @@ func (conn *RemoteConnection) String() string { // Async. Does not return anything. If the connection is successful, // it will end up in the local peer's connections map. -func NewLocalConnection(connRemote *RemoteConnection, expectedName PeerName, tcpConn *net.TCPConn, udpAddr *net.UDPAddr, router *Router) { +func NewLocalConnection(connRemote *RemoteConnection, acceptNewPeer bool, tcpConn *net.TCPConn, udpAddr *net.UDPAddr, router *Router) { if connRemote.local != router.Ourself { log.Fatal("Attempt to create local connection from a peer which is not ourself") } @@ -65,7 +65,7 @@ func NewLocalConnection(connRemote *RemoteConnection, expectedName PeerName, tcp remoteUDPAddr: udpAddr, effectivePMTU: DefaultPMTU, queryChan: queryChan} - go connLocal.queryLoop(queryChan, expectedName) + go connLocal.queryLoop(queryChan, acceptNewPeer) } func (conn *LocalConnection) Established() bool { @@ -161,8 +161,8 @@ func (conn *LocalConnection) SendTCP(msg []byte) { // ACTOR server -func (conn *LocalConnection) queryLoop(queryChan <-chan *ConnectionInteraction, expectedName PeerName) { - err := conn.handshake(expectedName) +func (conn *LocalConnection) queryLoop(queryChan <-chan *ConnectionInteraction, acceptNewPeer bool) { + err := conn.handshake(acceptNewPeer) if err != nil { log.Printf("->[%s] encountered error during handshake: %v\n", conn.remoteTCPAddr, err) conn.handleShutdown() @@ -279,11 +279,11 @@ func (conn *LocalConnection) handleShutdown() { conn.stopForwarders() if conn.remote != nil { - conn.Router.ConnectionMaker.EnsureConnection(conn.remote.Name, conn.remoteTCPAddr) + conn.Router.ConnectionMaker.ShutdownConnection(conn.remoteTCPAddr) } } -func (conn *LocalConnection) handshake(expectedName PeerName) error { +func (conn *LocalConnection) handshake(acceptNewPeer bool) error { // We do not need to worry about locking in here as at this point, // the connection is not reachable by any go-routine other than // ourself. Only when we add this connection to the conn.local @@ -344,8 +344,10 @@ func (conn *LocalConnection) handshake(expectedName PeerName) error { if err != nil { return err } - if expectedName != UnknownPeerName && expectedName != name { - return fmt.Errorf("Unexpected remote name: expected %s, received %s", expectedName, name) + if !acceptNewPeer { + if _, found := conn.Router.Peers.Fetch(name); !found { + return fmt.Errorf("Found unknown remote name: %s at %s", name, conn.remoteTCPAddr) + } } uidStr, err := checkHandshakeStringField("UID", "", handshakeRecv) diff --git a/connection_maker.go b/connection_maker.go index 94d1cf3c..babcee3b 100644 --- a/connection_maker.go +++ b/connection_maker.go @@ -5,8 +5,6 @@ import ( "fmt" "log" "math/rand" - "net" - "strconv" "time" ) @@ -14,25 +12,37 @@ const ( InitialInterval = 5 * time.Second MaxInterval = 10 * time.Minute MaxAttemptCount = 100 - CMEnsure = iota + CMInitiate = iota CMStatus = iota + CMFailed = iota +) + +const ( + CSUnconnected ConnectionState = iota + CSAttempting = iota ) func StartConnectionMaker(router *Router) *ConnectionMaker { queryChan := make(chan *ConnectionMakerInteraction, ChannelSize) state := &ConnectionMaker{ - router: router, - queryChan: queryChan, - failedConnections: make(map[PeerName]*FailedConnection)} + router: router, + queryChan: queryChan, + cmdLineAddress: make(map[string]bool), + targets: make(map[string]*Target)} go state.queryLoop(queryChan) return state } -func (cm *ConnectionMaker) EnsureConnection(name PeerName, foundAt string) { +func (cm *ConnectionMaker) InitiateConnection(address string) { cm.queryChan <- &ConnectionMakerInteraction{ - Interaction: Interaction{code: CMEnsure}, - name: name, - foundAt: foundAt} + Interaction: Interaction{code: CMInitiate}, + address: address} +} + +func (cm *ConnectionMaker) ShutdownConnection(address string) { + cm.queryChan <- &ConnectionMakerInteraction{ + Interaction: Interaction{code: CMFailed}, + address: address} } func (cm *ConnectionMaker) String() string { @@ -46,7 +56,8 @@ func (cm *ConnectionMaker) String() string { func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteraction) { var tick <-chan time.Time maybeTick := func() { - if tick == nil && len(cm.failedConnections) > 0 { + // would be nice to optimise this to stop ticking when there is nothing worth trying + if tick == nil { tick = time.After(5 * time.Second) } } @@ -57,82 +68,128 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio return } switch { - case query.code == CMEnsure: - if query.name != cm.router.Ourself.Name { - cm.addToFailedConnection(query.name, query.foundAt) - maybeTick() - } + case query.code == CMInitiate: + cm.cmdLineAddress[NormalisePeerAddr(query.address)] = true + cm.checkStateAndAttemptConnections(time.Now()) + maybeTick() case query.code == CMStatus: query.resultChan <- cm.status() + case query.code == CMFailed: + if target, found := cm.targets[query.address]; found { + target.state = CSUnconnected + target.tryAfter, target.tryInterval = tryAfter(target.tryInterval) + maybeTick() + } default: log.Fatal("Unexpected connection maker query:", query) } case now := <-tick: - for name, failedConn := range cm.failedConnections { - if now.After(failedConn.tryAfter) { - if _, found := cm.router.Ourself.ConnectionTo(name); found { - delete(cm.failedConnections, name) - continue - } else if failedConn.attemptCount == MaxAttemptCount { - delete(cm.failedConnections, name) - continue - } - after, interval := tryAfter(failedConn.tryInterval) - failedConn.tryInterval = interval - failedConn.tryAfter = after - failedConn.attemptCount += 1 - for target := range failedConn.foundAt { - go cm.attemptConnection(target, name) - } - } - } + cm.checkStateAndAttemptConnections(now) tick = nil maybeTick() } } } -func (cm *ConnectionMaker) addToFailedConnection(name PeerName, foundAt string) { - failed := cm.failedConnections[name] - if failed == nil { - after, interval := tryAfter(InitialInterval) - failed = &FailedConnection{ - foundAt: make(map[string]bool), - tryInterval: interval, - tryAfter: after} - } - foundAtHost, foundAtPortStr, err := net.SplitHostPort(foundAt) - if err == nil { - // ensure port-less version is there - failed.foundAt[foundAtHost] = true - if foundAtPort, err := strconv.Atoi(foundAtPortStr); err == nil && foundAtPort != Port { - failed.foundAt[foundAt] = true +func (cm *ConnectionMaker) checkStateAndAttemptConnections(now time.Time) { + ourself := cm.router.Ourself + validTarget := make(map[string]bool) + + // copy the set of things we are connected to, so we can access them without locking + our_connected_peers := make(map[PeerName]bool) + our_connected_targets := make(map[string]bool) + ourself.ForEachConnection(func(peer PeerName, conn Connection) { + //log.Println("Connected peer:", peer, conn.RemoteTCPAddr()) + our_connected_peers[peer] = true + our_connected_targets[conn.RemoteTCPAddr()] = true + }) + + // Add command-line targets that are not connected + for address, _ := range cm.cmdLineAddress { + if !our_connected_targets[NormalisePeerAddr(address)] { + //log.Println("Unconnected cmdline:", address) + validTarget[address] = true } - } else { - // can't split it, assume it must not have port on it - failed.foundAt[foundAt] = true } - cm.failedConnections[name] = failed + + // Add peers that someone else is connected to, but we aren't + cm.router.Peers.ForEach(func(name PeerName, peer *Peer) { + peer.ForEachConnection(func(peer2 PeerName, conn Connection) { + if peer2 != ourself.Name && !our_connected_peers[peer2] { + address := conn.RemoteTCPAddr() + //log.Println("Unconnected peer:", peer2, address) + // try both portnumber of connection and standart port + if host, port, err := ExtractHostPort(address); err == nil { + if port != Port { + // This is an address with port, that is not the standard port + validTarget[address] = true + } + // Add the address with standard port + validTarget[NormalisePeerAddr(host)] = true + } + } + }) + }) + + for address, _ := range validTarget { + cm.addToTargets(address) + } + + now = time.Now() // make sure we catch items just added + for address, target := range cm.targets { + if target.state == CSUnconnected { + if our_connected_targets[address] || !validTarget[address] { + //log.Println("Deleting target no longer valid:", address) + delete(cm.targets, address) + } else if now.After(target.tryAfter) { + target.attemptCount += 1 + target.state = CSAttempting + go cm.attemptConnection(address, cm.cmdLineAddress[address]) + } + } else if our_connected_targets[address] { + //log.Println("Deleting target now connected:", address) + delete(cm.targets, address) + } + } +} + +func (cm *ConnectionMaker) addToTargets(address string) { + address = NormalisePeerAddr(address) + target := cm.targets[address] + if target == nil { + target = &Target{ + state: CSUnconnected, + } + target.tryAfter, target.tryInterval = tryImmediately() + cm.targets[address] = target + } } func (cm *ConnectionMaker) status() string { var buf bytes.Buffer - for name, failedConn := range cm.failedConnections { - foundAt := make([]string, 0, len(failedConn.foundAt)) - for target := range failedConn.foundAt { - foundAt = append(foundAt, target) + for address, target := range cm.targets { + if target.state == CSAttempting { + buf.WriteString(fmt.Sprintf("%s (%v attempts, trying since %v)\n", address, target.attemptCount, target.tryAfter)) + } else { + buf.WriteString(fmt.Sprintf("%s (%v attempts, next at %v)\n", address, target.attemptCount, target.tryAfter)) } - buf.WriteString(fmt.Sprintf("%s (%v attempts, next at %v): %v\n", name, failedConn.attemptCount, failedConn.tryAfter, foundAt)) } return buf.String() } -func (cm *ConnectionMaker) attemptConnection(foundAt string, targetName PeerName) { - if err := cm.router.Ourself.CreateConnection(foundAt, targetName); err != nil { +func (cm *ConnectionMaker) attemptConnection(address string, acceptNewPeer bool) { + log.Println("Attempting connection to", address) + if err := cm.router.Ourself.CreateConnection(address, acceptNewPeer); err != nil { log.Println(err) + cm.ShutdownConnection(address) } } +func tryImmediately() (time.Time, time.Duration) { + interval := time.Duration(rand.Int63n(int64(InitialInterval))) + return time.Now(), interval +} + func tryAfter(interval time.Duration) (time.Time, time.Duration) { interval += time.Duration(rand.Int63n(int64(interval))) if interval > MaxInterval { diff --git a/peer.go b/peer.go index b01a3053..b62e8b27 100644 --- a/peer.go +++ b/peer.go @@ -109,9 +109,6 @@ func (peer *Peer) SetVersionAndConnections(version uint64, connections map[PeerN defer peer.Unlock() peer.version = version peer.connections = connections - for remoteName, conn := range connections { - peer.Router.ConnectionMaker.EnsureConnection(remoteName, conn.RemoteTCPAddr()) - } } func (peer *Peer) Forward(dstPeer *Peer, df bool, frame []byte, dec *EthernetDecoder) error { @@ -190,18 +187,12 @@ func (peer *Peer) StartLocalPeer() { go peer.queryLoop(queryChan) } -func (peer *Peer) CreateConnection(peerAddr string, expectedName PeerName) error { +func (peer *Peer) CreateConnection(peerAddr string, acceptNewPeer bool) error { if err := peer.checkConnectionLimit(); err != nil { return err } // We're dialing the remote so that means connections will come from random ports - _, _, err := net.SplitHostPort(peerAddr) - var addrStr string - if err == nil { - addrStr = peerAddr - } else { - addrStr = fmt.Sprintf("%s:%d", peerAddr, Port) - } + addrStr := NormalisePeerAddr(peerAddr) tcpAddr, tcpErr := net.ResolveTCPAddr("tcp4", addrStr) udpAddr, udpErr := net.ResolveUDPAddr("udp4", addrStr) if tcpErr != nil || udpErr != nil { @@ -216,7 +207,7 @@ func (peer *Peer) CreateConnection(peerAddr string, expectedName PeerName) error return err } connRemote := NewRemoteConnection(peer, nil, tcpConn.RemoteAddr().String()) - NewLocalConnection(connRemote, expectedName, tcpConn, udpAddr, peer.Router) + NewLocalConnection(connRemote, acceptNewPeer, tcpConn, udpAddr, peer.Router) return nil } diff --git a/router.go b/router.go index 39a7bac6..ec2d9de1 100644 --- a/router.go +++ b/router.go @@ -154,7 +154,7 @@ func (router *Router) acceptTCP(tcpConn *net.TCPConn) { // on Port and we wait for them to send us something on UDP to // start. connRemote := NewRemoteConnection(router.Ourself, nil, tcpConn.RemoteAddr().String()) - NewLocalConnection(connRemote, UnknownPeerName, tcpConn, nil, router) + NewLocalConnection(connRemote, true, tcpConn, nil, router) } func (router *Router) listenUDP(localPort int, po PacketSink) *net.UDPConn { diff --git a/types.go b/types.go index 020fba18..afff09d7 100644 --- a/types.go +++ b/types.go @@ -155,21 +155,23 @@ type Topology struct { type ConnectionMakerInteraction struct { Interaction - foundAt string - name PeerName + address string } type ConnectionMaker struct { - router *Router - queryChan chan<- *ConnectionMakerInteraction - activeConnections map[string]string - failedConnections map[PeerName]*FailedConnection + router *Router + queryChan chan<- *ConnectionMakerInteraction + targets map[string]*Target + cmdLineAddress map[string]bool } -type FailedConnection struct { - foundAt map[string]bool - tryAfter time.Time - tryInterval time.Duration +type ConnectionState int + +// Information about an address where we may find a peer +type Target struct { + state ConnectionState + tryAfter time.Time // next time to try this address + tryInterval time.Duration // backoff time on next failure attemptCount int } diff --git a/utils.go b/utils.go index 22d7235b..74dce894 100644 --- a/utils.go +++ b/utils.go @@ -5,6 +5,7 @@ import ( "fmt" "log" "net" + "strconv" ) func checkFatal(e error) { @@ -97,3 +98,36 @@ func (lop ListOfPeers) Swap(i, j int) { func (lop ListOfPeers) Less(i, j int) bool { return lop[i].Name < lop[i].Name } + +func StripPortFromAddr(address string) string { + // If we've been given a port number, take it off + if addrHost, _, err := net.SplitHostPort(address); err == nil { + return addrHost + } + 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 { + _, _, err := net.SplitHostPort(peerAddr) + if err == nil { + return peerAddr + } else { + return fmt.Sprintf("%s:%d", peerAddr, Port) + } +} diff --git a/weaver/main.go b/weaver/main.go index e11f1c49..63bb60d1 100644 --- a/weaver/main.go +++ b/weaver/main.go @@ -124,12 +124,7 @@ func main() { router := weave.NewRouter(iface, ourName, []byte(password), connLimit, bufSz*1024*1024, logFrame) router.Start() for _, peer := range peers { - targetPeer := peer - go func() { - if err := router.Ourself.CreateConnection(targetPeer, weave.UnknownPeerName); err != nil { - log.Println(err) - } - }() + router.ConnectionMaker.InitiateConnection(peer) } go handleHttp(router) handleSignals(router)