Merge branch '23_retry_peers' of https://github.com/bboreham/weave into bboreham-23_retry_peers

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)

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 {

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
 }
 

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 {

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
 }
 

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)
+	}
+}

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)