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

connection.go

@@ -221,6 +221,7 @@ 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
@@ -345,8 +346,8 @@ func (conn *LocalConnection) handshake(acceptNewPeer bool) error {
 		return err
 	}
 	if !acceptNewPeer {
-	    if _, found := conn.Router.Peers.Fetch(name); !found {
-		    return fmt.Errorf("Found unknown remote name: %s at %s", name, conn.remoteTCPAddr)
+		if _, found := conn.Router.Peers.Fetch(name); !found {
+			return fmt.Errorf("Found unknown remote name: %s at %s", name, conn.remoteTCPAddr)
 		}
 	}
 
@@ -447,7 +448,6 @@ func (conn *LocalConnection) receiveTCP(decoder *gob.Decoder, usingPassword bool
 			// them and thus has informed us via TCP that it considers
 			// the connection is now up. We now do a fetchAll on it.
 			conn.SetEstablished()
-			conn.Router.ConnectionMaker.ConnectionEstablished(conn.remoteTCPAddr)
 			conn.SendTCP(ProtocolFetchAllByte)
 		} else if msg[0] == ProtocolStartFragmentationTest {
 			conn.Forward(false, &ForwardedFrame{

connection_maker.go

@@ -10,29 +10,34 @@ import (
 )
 
 const (
-	InitialInterval = 1 * time.Second
+	InitialInterval = 5 * time.Second
 	MaxInterval     = 10 * time.Minute
 	MaxAttemptCount = 100
 	CMEnsure        = iota
 	CMStatus        = iota
 	CMEstablished   = iota
-	CMConnSucceeded = iota
 	CMConnFailed    = iota
 )
 
+const (
+	CMUnconnected      ConnectionState = iota
+	CMAttempting                       = iota
+	CMEstablishedState                 = iota
+)
+
 func StartConnectionMaker(router *Router) *ConnectionMaker {
 	queryChan := make(chan *ConnectionMakerInteraction, ChannelSize)
 	state := &ConnectionMaker{
-		router:            router,
-		queryChan:         queryChan,
-		targets: 		   make(map[string]*Target)}
+		router:    router,
+		queryChan: queryChan,
+		targets:   make(map[string]*Target)}
 	go state.queryLoop(queryChan)
 	return state
 }
 
 func (cm *ConnectionMaker) InitiateConnection(address string) {
 	cm.queryChan <- &ConnectionMakerInteraction{
-		Interaction: Interaction{code: CMEnsure},
+		Interaction:   Interaction{code: CMEnsure},
 		acceptAnyPeer: true,
 		address:       address}
 }
@@ -43,9 +48,9 @@ func (cm *ConnectionMaker) EnsureConnection(address string) {
 		address = addrHost
 	}
 	cm.queryChan <- &ConnectionMakerInteraction{
-		Interaction: Interaction{code: CMEnsure},
+		Interaction:   Interaction{code: CMEnsure},
 		acceptAnyPeer: false,
-		address:     address}
+		address:       address}
 }
 
 func (cm *ConnectionMaker) ConnectionEstablished(address string) {
@@ -74,6 +79,11 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio
 			tick = time.After(5 * time.Second)
 		}
 	}
+	tickNow := func() {
+		if tick == nil {
+			tick = time.After(0 * time.Second)
+		}
+	}
 	for {
 		select {
 		case query, ok := <-queryChan:
@@ -83,32 +93,30 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio
 			switch {
 			case query.code == CMEnsure:
 				cm.addToTargets(query.acceptAnyPeer, query.address)
-				maybeTick()
+				cm.resetTarget(query.address)
+				tickNow()
 			case query.code == CMStatus:
 				query.resultChan <- cm.status()
 			case query.code == CMEstablished:
 				target := cm.targets[query.address]
-				log.Println("Connection established to", query.address, target)
 				if target != nil {
-					target.established = true
+					target.state = CMEstablishedState
+				} else {
+					log.Fatal("Connection established to unknown address", query.address)
 				}
-			case query.code == CMConnSucceeded:
-				cm.targets[query.address].attempting = false
 			case query.code == CMConnFailed:
 				target := cm.targets[query.address]
-				target.attempting = false
-				after, interval := tryAfter(target.tryInterval)
-				target.tryInterval = interval
-				target.tryAfter = after
+				target.state = CMUnconnected
+				target.tryAfter, target.tryInterval = tryAfter(target.tryInterval)
 				maybeTick()
 			default:
 				log.Fatal("Unexpected connection maker query:", query)
 			}
 		case now := <-tick:
 			for address, target := range cm.targets {
-				if !target.established && !target.attempting && now.After(target.tryAfter) {
+				if target.state == CMUnconnected && now.After(target.tryAfter) {
 					target.attemptCount += 1
-					target.attempting = true;
+					target.state = CMAttempting
 					go cm.attemptConnection(address, target.acceptAnyPeer)
 				}
 			}
@@ -121,23 +129,31 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio
 func (cm *ConnectionMaker) addToTargets(acceptAnyPeer bool, address string) {
 	target := cm.targets[address]
 	if target == nil {
-		after, interval := tryAfter(InitialInterval)
 		target = &Target{
+			state:         CMUnconnected,
 			acceptAnyPeer: acceptAnyPeer,
-			tryInterval: interval,
-			tryAfter:    after}
+		}
+		cm.targets[address] = target
 	}
-	target.established = false  // this seems a crap place to do this
-	// FIXME: what does it mean if an address is added twice?
-    cm.targets[address] = target
+}
+
+func (cm *ConnectionMaker) resetTarget(address string) {
+	target := cm.targets[address]
+	// If we were in the middle of a connection attempt then back-off again
+	if target.state == CMAttempting {
+		target.tryAfter, target.tryInterval = tryAfter(target.tryInterval)
+	} else {
+		target.tryAfter, target.tryInterval = tryImmediately()
+	}
+	target.state = CMUnconnected
 }
 
 func (cm *ConnectionMaker) status() string {
 	var buf bytes.Buffer
 	for address, target := range cm.targets {
-		if (target.established) {
+		if target.state == CMEstablishedState {
 			buf.WriteString(fmt.Sprintf("%s connected\n", address))
-		} else if target.attempting {
+		} else if target.state == CMAttempting {
 			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))
@@ -148,15 +164,17 @@ func (cm *ConnectionMaker) status() string {
 
 func (cm *ConnectionMaker) attemptConnection(address string, acceptNewPeer bool) {
 	log.Println("Attempting connection to", address)
-	var conncode = CMConnSucceeded
 	if err := cm.router.Ourself.CreateConnection(address, acceptNewPeer); err != nil {
 		log.Println(err)
-		conncode = CMConnFailed
+		cm.queryChan <- &ConnectionMakerInteraction{
+			Interaction: Interaction{code: CMConnFailed},
+			address:     address}
 	}
-	// Tell the query loop we've finished this attempt
-	cm.queryChan <- &ConnectionMakerInteraction{
-		Interaction: Interaction{code: conncode},
-		address:     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) {

peer.go

@@ -115,6 +115,7 @@ func (peer *Peer) SetVersionAndConnections(version uint64, connections map[PeerN
 		}
 	}
 }
+
 func (peer *Peer) Forward(dstPeer *Peer, df bool, frame []byte, dec *EthernetDecoder) error {
 	return peer.Relay(peer, dstPeer, df, frame, dec)
 }

types.go

@@ -160,19 +160,20 @@ type ConnectionMakerInteraction struct {
 }
 
 type ConnectionMaker struct {
-	router            *Router
-	queryChan         chan<- *ConnectionMakerInteraction
-    targets           map[string]*Target
+	router    *Router
+	queryChan chan<- *ConnectionMakerInteraction
+	targets   map[string]*Target
 }
 
+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?
-	attempting	 bool		// are we currently attempting to connect to this address?
-	established  bool       // has the connection been established
-	tryAfter     time.Time				   // next time to try this address
-	tryInterval  time.Duration			   // backoff time on next failure
-	attemptCount int
+	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
 }
 
 // UDPSender interface and implementations

weaver/main.go

@@ -1,13 +1,13 @@
 package main
 
 import (
-	"github.com/zettio/weave"
 	"code.google.com/p/gopacket/layers"
 	"crypto/sha256"
 	"flag"
 	"fmt"
-	"io"
 	"github.com/davecheney/profile"
+	"github.com/zettio/weave"
+	"io"
 	"log"
 	"net"
 	"net/http"
@@ -141,7 +141,6 @@ func handleHttp(router *weave.Router) {
 	}
 }
 
-
 func handleSignals(router *weave.Router) {
 	sigs := make(chan os.Signal, 1)
 	signal.Notify(sigs, syscall.SIGQUIT, syscall.SIGUSR1)