Merge pull request #44 from bboreham/23_retry_peers

Rewrite of ConnectionMaker

Fixes #23. Fixes #26.

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()
@@ -278,12 +278,10 @@ func (conn *LocalConnection) handleShutdown() {
 	// try to send any more
 	conn.stopForwarders()
 
-	if conn.remote != nil {
-		conn.Router.ConnectionMaker.EnsureConnection(conn.remote.Name, conn.remoteTCPAddr)
-	}
+	conn.Router.ConnectionMaker.ConnectionTerminated(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 +342,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

@@ -6,33 +6,41 @@ import (
 	"log"
 	"math/rand"
 	"net"
-	"strconv"
 	"time"
 )
 
 const (
 	InitialInterval = 5 * time.Second
 	MaxInterval     = 10 * time.Minute
-	MaxAttemptCount = 100
-	CMEnsure        = iota
-	CMStatus        = iota
+)
+
+const (
+	CMInitiate   = iota
+	CMStatus     = iota
+	CMTerminated = 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) ConnectionTerminated(address string) {
+	cm.queryChan <- &ConnectionMakerInteraction{
+		Interaction: Interaction{code: CMTerminated},
+		address:     address}
 }
 
 func (cm *ConnectionMaker) String() string {
@@ -46,7 +54,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 +66,124 @@ 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 == CMTerminated:
+				if target, found := cm.targets[query.address]; found {
+					target.attempting = false
+					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) {
+		our_connected_peers[peer] = true
+		our_connected_targets[conn.RemoteTCPAddr()] = true
+	})
+
+	addTarget := func(address string) {
+		if !our_connected_targets[address] {
+			validTarget[address] = true
+			cm.addTarget(address)
 		}
-	} else {
-		// can't split it, assume it must not have port on it
-		failed.foundAt[foundAt] = true
 	}
-	cm.failedConnections[name] = failed
+
+	// Add command-line targets that are not connected
+	for address, _ := range cm.cmdLineAddress {
+		addTarget(address)
+	}
+
+	// Add targets for peers that someone else is connected to, but we
+	// aren't
+	cm.router.Peers.ForEach(func(name PeerName, peer *Peer) {
+		peer.ForEachConnection(func(otherPeer PeerName, conn Connection) {
+			if otherPeer == ourself.Name || our_connected_peers[otherPeer] {
+				return
+			}
+			address := conn.RemoteTCPAddr()
+			// try both portnumber of connection and standard port
+			addTarget(address)
+			if host, _, err := net.SplitHostPort(address); err == nil {
+				addTarget(NormalisePeerAddr(host))
+			}
+		})
+	})
+
+	now = time.Now() // make sure we catch items just added
+	for address, target := range cm.targets {
+		if our_connected_targets[address] {
+			delete(cm.targets, address)
+			continue
+		}
+		if target.attempting {
+			continue
+		}
+		if !validTarget[address] {
+			delete(cm.targets, address)
+			continue
+		}
+		if now.After(target.tryAfter) {
+			target.attempting = true
+			go cm.attemptConnection(address, cm.cmdLineAddress[address])
+		}
+	}
+}
+
+func (cm *ConnectionMaker) addTarget(address string) {
+	if _, found := cm.targets[address]; !found {
+		target := &Target{}
+		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 {
+		var fmtStr string
+		if target.attempting {
+			fmtStr = "%s (trying since %v)\n"
+		} else {
+			fmtStr = "%s (next try at %v)\n"
 		}
-		buf.WriteString(fmt.Sprintf("%s (%v attempts, next at %v): %v\n", name, failedConn.attemptCount, failedConn.tryAfter, foundAt))
+		buf.WriteString(fmt.Sprintf(fmtStr, address, target.tryAfter))
 	}
 	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.ConnectionTerminated(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,22 +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
-	attemptCount int
+type ConnectionState int
+
+// Information about an address where we may find a peer
+type Target struct {
+	attempting   bool          // are we currently attempting to connect there?
+	tryAfter     time.Time     // next time to try this address
+	tryInterval  time.Duration // backoff time on next failure
 }
 
 // UDPSender interface and implementations

utils.go

@@ -97,3 +97,14 @@ func (lop ListOfPeers) Swap(i, j int) {
 func (lop ListOfPeers) Less(i, j int) bool {
 	return lop[i].Name < lop[i].Name
 }
+
+// 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)