From a5be96d35fd8525b6382fc0a207d8629446df498 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Fri, 5 Sep 2014 11:11:38 +0100 Subject: [PATCH 01/38] Kick off the first connection attempt to each peer using ConnectionMaker instead of calling CreateConnection directly. --- weaver/main.go | 7 +------ 1 file changed, 1 insertion(+), 6 deletions(-) diff --git a/weaver/main.go b/weaver/main.go index c0be4ea5..96dbaf21 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.EnsureConnection(weave.UnknownPeerName, peer) } go handleHttp(router) handleSignals(router) From 09271f48486ef920d4b64cb280f7826e964ee097 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Fri, 5 Sep 2014 11:16:13 +0100 Subject: [PATCH 02/38] =?UTF-8?q?Track=20which=20connections=20we=20are=20?= =?UTF-8?q?currently=20attempting=20a=20connect=20on,=20so=20we=20don?= =?UTF-8?q?=E2=80=99t=20do=20multiple=20attempts=20at=20the=20same=20time?= =?UTF-8?q?=20to=20the=20same=20endpoint?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- connection_maker.go | 25 ++++++++++++++++++++++--- types.go | 6 ++++++ 2 files changed, 28 insertions(+), 3 deletions(-) diff --git a/connection_maker.go b/connection_maker.go index 94d1cf3c..b2b7e00c 100644 --- a/connection_maker.go +++ b/connection_maker.go @@ -11,11 +11,12 @@ import ( ) const ( - InitialInterval = 5 * time.Second + InitialInterval = 1 * time.Second MaxInterval = 10 * time.Minute MaxAttemptCount = 100 CMEnsure = iota CMStatus = iota + CMConnAttemptFinished = iota ) func StartConnectionMaker(router *Router) *ConnectionMaker { @@ -23,7 +24,8 @@ func StartConnectionMaker(router *Router) *ConnectionMaker { state := &ConnectionMaker{ router: router, queryChan: queryChan, - failedConnections: make(map[PeerName]*FailedConnection)} + failedConnections: make(map[PeerName]*FailedConnection), + attemptingConnections: make(map[ConnectionMakerPair]bool)} go state.queryLoop(queryChan) return state } @@ -64,6 +66,8 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio } case query.code == CMStatus: query.resultChan <- cm.status() + case query.code == CMConnAttemptFinished: + cm.attemptingConnections[ConnectionMakerPair{query.foundAt, query.name}] = false; default: log.Fatal("Unexpected connection maker query:", query) } @@ -82,6 +86,10 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio failedConn.tryAfter = after failedConn.attemptCount += 1 for target := range failedConn.foundAt { + if cm.attemptingConnections[ConnectionMakerPair{target, name}] { + continue + } + cm.attemptingConnections[ConnectionMakerPair{target, name}] = true go cm.attemptConnection(target, name) } } @@ -118,11 +126,17 @@ func (cm *ConnectionMaker) addToFailedConnection(name PeerName, foundAt string) func (cm *ConnectionMaker) status() string { var buf bytes.Buffer for name, failedConn := range cm.failedConnections { + tryingNow := false foundAt := make([]string, 0, len(failedConn.foundAt)) for target := range failedConn.foundAt { foundAt = append(foundAt, target) + tryingNow = tryingNow || cm.attemptingConnections[ConnectionMakerPair{target, name}] + } + if tryingNow { + buf.WriteString(fmt.Sprintf("%s (%v attempts, trying again now): %v\n", name, failedConn.attemptCount, foundAt)) + } else { + buf.WriteString(fmt.Sprintf("%s (%v attempts, next at %v): %v\n", name, failedConn.attemptCount, failedConn.tryAfter, foundAt)) } - buf.WriteString(fmt.Sprintf("%s (%v attempts, next at %v): %v\n", name, failedConn.attemptCount, failedConn.tryAfter, foundAt)) } return buf.String() } @@ -131,6 +145,11 @@ func (cm *ConnectionMaker) attemptConnection(foundAt string, targetName PeerName if err := cm.router.Ourself.CreateConnection(foundAt, targetName); err != nil { log.Println(err) } + // Tell the query loop we've finished this attempt + cm.queryChan <- &ConnectionMakerInteraction{ + Interaction: Interaction{code: CMConnAttemptFinished}, + name: targetName, + foundAt: foundAt} } func tryAfter(interval time.Duration) (time.Time, time.Duration) { diff --git a/types.go b/types.go index 020fba18..231deb42 100644 --- a/types.go +++ b/types.go @@ -159,11 +159,17 @@ type ConnectionMakerInteraction struct { name PeerName } +type ConnectionMakerPair struct { + foundAt string + name PeerName +} + type ConnectionMaker struct { router *Router queryChan chan<- *ConnectionMakerInteraction activeConnections map[string]string failedConnections map[PeerName]*FailedConnection + attemptingConnections map[ConnectionMakerPair]bool } type FailedConnection struct { From 54d7972b5e0a9262e0720194e75c495dad370d27 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Fri, 5 Sep 2014 14:26:42 +0100 Subject: [PATCH 03/38] Don't start a new connection attempt until all in-flight attempts for this peer have finished Also cleaned up the naming slightly --- connection_maker.go | 26 +++++++++++++++++++------- types.go | 2 +- 2 files changed, 20 insertions(+), 8 deletions(-) diff --git a/connection_maker.go b/connection_maker.go index b2b7e00c..964949af 100644 --- a/connection_maker.go +++ b/connection_maker.go @@ -1,5 +1,13 @@ package weave +/* ConnectionMaker is responsible for making connections between peers, +and retrying them when they break. It sits in a loop waiting for requests +to be sent over queryChan, or for timer ticks. +Every connection has a PeerName, which is a unique identifier for every weave +router (e.g. a MAC address), and an IP address which it was foundAt. + +*/ + import ( "bytes" "fmt" @@ -25,7 +33,7 @@ func StartConnectionMaker(router *Router) *ConnectionMaker { router: router, queryChan: queryChan, failedConnections: make(map[PeerName]*FailedConnection), - attemptingConnections: make(map[ConnectionMakerPair]bool)} + attempting: make(map[ConnectionMakerPair]bool)} go state.queryLoop(queryChan) return state } @@ -67,11 +75,12 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio case query.code == CMStatus: query.resultChan <- cm.status() case query.code == CMConnAttemptFinished: - cm.attemptingConnections[ConnectionMakerPair{query.foundAt, query.name}] = false; + delete(cm.attempting, ConnectionMakerPair{query.foundAt, query.name}) default: log.Fatal("Unexpected connection maker query:", query) } case now := <-tick: + ConnectionLoop: for name, failedConn := range cm.failedConnections { if now.After(failedConn.tryAfter) { if _, found := cm.router.Ourself.ConnectionTo(name); found { @@ -81,15 +90,18 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio delete(cm.failedConnections, name) continue } + // wait until all current attempts to contact this peer have failed before starting a new batch + for target := range failedConn.foundAt { + if cm.attempting[ConnectionMakerPair{target, name}] { + continue ConnectionLoop + } + } after, interval := tryAfter(failedConn.tryInterval) failedConn.tryInterval = interval failedConn.tryAfter = after failedConn.attemptCount += 1 for target := range failedConn.foundAt { - if cm.attemptingConnections[ConnectionMakerPair{target, name}] { - continue - } - cm.attemptingConnections[ConnectionMakerPair{target, name}] = true + cm.attempting[ConnectionMakerPair{target, name}] = true go cm.attemptConnection(target, name) } } @@ -130,7 +142,7 @@ func (cm *ConnectionMaker) status() string { foundAt := make([]string, 0, len(failedConn.foundAt)) for target := range failedConn.foundAt { foundAt = append(foundAt, target) - tryingNow = tryingNow || cm.attemptingConnections[ConnectionMakerPair{target, name}] + tryingNow = tryingNow || cm.attempting[ConnectionMakerPair{target, name}] } if tryingNow { buf.WriteString(fmt.Sprintf("%s (%v attempts, trying again now): %v\n", name, failedConn.attemptCount, foundAt)) diff --git a/types.go b/types.go index 231deb42..bfd434b8 100644 --- a/types.go +++ b/types.go @@ -169,7 +169,7 @@ type ConnectionMaker struct { queryChan chan<- *ConnectionMakerInteraction activeConnections map[string]string failedConnections map[PeerName]*FailedConnection - attemptingConnections map[ConnectionMakerPair]bool + attempting map[ConnectionMakerPair]bool } type FailedConnection struct { From 2d387e3847cbbc8d805f5cfd80c045fe9750b05c Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Fri, 5 Sep 2014 14:28:47 +0100 Subject: [PATCH 04/38] Removed ConnectionMaker.activeConnections which is never used --- types.go | 1 - 1 file changed, 1 deletion(-) diff --git a/types.go b/types.go index bfd434b8..e41b73c0 100644 --- a/types.go +++ b/types.go @@ -167,7 +167,6 @@ type ConnectionMakerPair struct { type ConnectionMaker struct { router *Router queryChan chan<- *ConnectionMakerInteraction - activeConnections map[string]string failedConnections map[PeerName]*FailedConnection attempting map[ConnectionMakerPair]bool } From 5c7010625a2bc4e0d20e1aa45cb457195cd95728 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Fri, 5 Sep 2014 15:45:32 +0100 Subject: [PATCH 05/38] Remove entries from the failedConnection list when they succeed. Avoids unsightly error messages where both ends make the same connection and we then have to drop one of them. --- connection_maker.go | 14 +++++++++++--- 1 file changed, 11 insertions(+), 3 deletions(-) diff --git a/connection_maker.go b/connection_maker.go index 964949af..57aa6ef8 100644 --- a/connection_maker.go +++ b/connection_maker.go @@ -24,7 +24,8 @@ const ( MaxAttemptCount = 100 CMEnsure = iota CMStatus = iota - CMConnAttemptFinished = iota + CMConnSucceeded = iota + CMConnFailed = iota ) func StartConnectionMaker(router *Router) *ConnectionMaker { @@ -74,7 +75,10 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio } case query.code == CMStatus: query.resultChan <- cm.status() - case query.code == CMConnAttemptFinished: + case query.code == CMConnSucceeded: + delete(cm.attempting, ConnectionMakerPair{query.foundAt, query.name}) + delete(cm.failedConnections[query.name].foundAt, query.foundAt) + case query.code == CMConnFailed: delete(cm.attempting, ConnectionMakerPair{query.foundAt, query.name}) default: log.Fatal("Unexpected connection maker query:", query) @@ -86,6 +90,8 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio if _, found := cm.router.Ourself.ConnectionTo(name); found { delete(cm.failedConnections, name) continue + } else if len(failedConn.foundAt) == 0 { + delete(cm.failedConnections, name) } else if failedConn.attemptCount == MaxAttemptCount { delete(cm.failedConnections, name) continue @@ -154,12 +160,14 @@ func (cm *ConnectionMaker) status() string { } func (cm *ConnectionMaker) attemptConnection(foundAt string, targetName PeerName) { + var conncode = CMConnSucceeded if err := cm.router.Ourself.CreateConnection(foundAt, targetName); err != nil { log.Println(err) + conncode = CMConnFailed } // Tell the query loop we've finished this attempt cm.queryChan <- &ConnectionMakerInteraction{ - Interaction: Interaction{code: CMConnAttemptFinished}, + Interaction: Interaction{code: conncode}, name: targetName, foundAt: foundAt} } From 783ee2146170f10a980a1a1243b01bad1e64c3d6 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Fri, 5 Sep 2014 18:28:42 +0100 Subject: [PATCH 06/38] Continue after we establish that there is nothing left to do for failedCOnnection --- connection_maker.go | 1 + 1 file changed, 1 insertion(+) diff --git a/connection_maker.go b/connection_maker.go index 57aa6ef8..f597bb17 100644 --- a/connection_maker.go +++ b/connection_maker.go @@ -92,6 +92,7 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio continue } else if len(failedConn.foundAt) == 0 { delete(cm.failedConnections, name) + continue } else if failedConn.attemptCount == MaxAttemptCount { delete(cm.failedConnections, name) continue From 0a94388f0a0c589febfa4193d206b2dff4e6bcc9 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Mon, 8 Sep 2014 12:42:06 +0100 Subject: [PATCH 07/38] Change ConnectionMaker from being focused on PeerName to being focused on address --- connection_maker.go | 89 +++++++++++++++++---------------------------- types.go | 19 ++++------ 2 files changed, 42 insertions(+), 66 deletions(-) diff --git a/connection_maker.go b/connection_maker.go index f597bb17..49254ed8 100644 --- a/connection_maker.go +++ b/connection_maker.go @@ -33,8 +33,7 @@ func StartConnectionMaker(router *Router) *ConnectionMaker { state := &ConnectionMaker{ router: router, queryChan: queryChan, - failedConnections: make(map[PeerName]*FailedConnection), - attempting: make(map[ConnectionMakerPair]bool)} + targets: make(map[string]*Target)} go state.queryLoop(queryChan) return state } @@ -57,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) } } @@ -70,47 +70,31 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio switch { case query.code == CMEnsure: if query.name != cm.router.Ourself.Name { - cm.addToFailedConnection(query.name, query.foundAt) + cm.addToTargets(query.name, query.foundAt) maybeTick() } case query.code == CMStatus: query.resultChan <- cm.status() case query.code == CMConnSucceeded: - delete(cm.attempting, ConnectionMakerPair{query.foundAt, query.name}) - delete(cm.failedConnections[query.name].foundAt, query.foundAt) + cm.targets[query.foundAt].attempting = false case query.code == CMConnFailed: - delete(cm.attempting, ConnectionMakerPair{query.foundAt, query.name}) + cm.targets[query.foundAt].attempting = false default: log.Fatal("Unexpected connection maker query:", query) } case now := <-tick: - ConnectionLoop: - 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 len(failedConn.foundAt) == 0 { - delete(cm.failedConnections, name) - continue - } else if failedConn.attemptCount == MaxAttemptCount { - delete(cm.failedConnections, name) + for address, target := range cm.targets { + if target.conn == nil && !target.attempting && now.After(target.tryAfter) { + if target.attemptCount == MaxAttemptCount { + // FIXME continue } - // wait until all current attempts to contact this peer have failed before starting a new batch - for target := range failedConn.foundAt { - if cm.attempting[ConnectionMakerPair{target, name}] { - continue ConnectionLoop - } - } - after, interval := tryAfter(failedConn.tryInterval) - failedConn.tryInterval = interval - failedConn.tryAfter = after - failedConn.attemptCount += 1 - for target := range failedConn.foundAt { - cm.attempting[ConnectionMakerPair{target, name}] = true - go cm.attemptConnection(target, name) - } + after, interval := tryAfter(target.tryInterval) + target.tryInterval = interval + target.tryAfter = after + target.attemptCount += 1 + target.attempting = true; + go cm.attemptConnection(address, target.commandLine) } } tick = nil @@ -119,58 +103,53 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio } } -func (cm *ConnectionMaker) addToFailedConnection(name PeerName, foundAt string) { - failed := cm.failedConnections[name] - if failed == nil { +func (cm *ConnectionMaker) addToTargets(name PeerName, foundAt string) { + target := cm.targets[foundAt] + if target == nil { after, interval := tryAfter(InitialInterval) - failed = &FailedConnection{ - foundAt: make(map[string]bool), + target = &Target{ tryInterval: interval, tryAfter: after} } foundAtHost, foundAtPortStr, err := net.SplitHostPort(foundAt) if err == nil { // ensure port-less version is there - failed.foundAt[foundAtHost] = true + cm.targets[foundAtHost] = target if foundAtPort, err := strconv.Atoi(foundAtPortStr); err == nil && foundAtPort != Port { - failed.foundAt[foundAt] = true + cm.targets[foundAt] = target } } else { // can't split it, assume it must not have port on it - failed.foundAt[foundAt] = true + cm.targets[foundAt] = target } - cm.failedConnections[name] = failed } func (cm *ConnectionMaker) status() string { var buf bytes.Buffer - for name, failedConn := range cm.failedConnections { - tryingNow := false - foundAt := make([]string, 0, len(failedConn.foundAt)) - for target := range failedConn.foundAt { - foundAt = append(foundAt, target) - tryingNow = tryingNow || cm.attempting[ConnectionMakerPair{target, name}] - } - if tryingNow { - buf.WriteString(fmt.Sprintf("%s (%v attempts, trying again now): %v\n", name, failedConn.attemptCount, foundAt)) + for address, target := range cm.targets { + if (target.conn != nil) { + buf.WriteString(fmt.Sprintf("%s connected to: %v\n", address, target.conn)) + } else if target.attempting { + buf.WriteString(fmt.Sprintf("%s (%v attempts, trying again now)\n", address, target.attemptCount)) } else { - buf.WriteString(fmt.Sprintf("%s (%v attempts, next at %v): %v\n", name, failedConn.attemptCount, failedConn.tryAfter, foundAt)) + buf.WriteString(fmt.Sprintf("%s (%v attempts, next at %v)\n", address, target.attemptCount, target.tryAfter)) } } return buf.String() } -func (cm *ConnectionMaker) attemptConnection(foundAt string, targetName PeerName) { +func (cm *ConnectionMaker) attemptConnection(address string, acceptNewPeer bool) { + log.Println("Attempting connection to ", address) var conncode = CMConnSucceeded - if err := cm.router.Ourself.CreateConnection(foundAt, targetName); err != nil { + if err := cm.router.Ourself.CreateConnection(address, UnknownPeerName); err != nil { log.Println(err) conncode = CMConnFailed } // Tell the query loop we've finished this attempt cm.queryChan <- &ConnectionMakerInteraction{ Interaction: Interaction{code: conncode}, - name: targetName, - foundAt: foundAt} + name: UnknownPeerName, + foundAt: address} } func tryAfter(interval time.Duration) (time.Time, time.Duration) { diff --git a/types.go b/types.go index e41b73c0..d7ce9020 100644 --- a/types.go +++ b/types.go @@ -159,22 +159,19 @@ type ConnectionMakerInteraction struct { name PeerName } -type ConnectionMakerPair struct { - foundAt string - name PeerName -} - type ConnectionMaker struct { router *Router queryChan chan<- *ConnectionMakerInteraction - failedConnections map[PeerName]*FailedConnection - attempting map[ConnectionMakerPair]bool + targets map[string]*Target } -type FailedConnection struct { - foundAt map[string]bool - tryAfter time.Time - tryInterval time.Duration +// Information about an address where we may find a peer +type Target struct { + commandLine bool // was this address given to us on the command line? + attempting bool // are we currently attempting to connect to this address? + conn *Connection + tryAfter time.Time // next time to try this address + tryInterval time.Duration // backoff time on next failure attemptCount int } From 8b3e97b96a63e899e6fd7c1cc9298b26daba8de1 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Mon, 8 Sep 2014 14:07:42 +0100 Subject: [PATCH 08/38] Change from stating which peer we expect to be on the end of a connection to accepting any known peer --- connection.go | 16 +++++++++------- connection_maker.go | 20 +++++++++----------- peer.go | 4 ++-- router.go | 2 +- 4 files changed, 21 insertions(+), 21 deletions(-) diff --git a/connection.go b/connection.go index 0eb1bd63..138df0c6 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() @@ -283,7 +283,7 @@ func (conn *LocalConnection) handleShutdown() { } } -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 49254ed8..df7346df 100644 --- a/connection_maker.go +++ b/connection_maker.go @@ -78,20 +78,18 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio case query.code == CMConnSucceeded: cm.targets[query.foundAt].attempting = false case query.code == CMConnFailed: - cm.targets[query.foundAt].attempting = false + target := cm.targets[query.foundAt] + target.attempting = false + after, interval := tryAfter(target.tryInterval) + target.tryInterval = interval + target.tryAfter = after + maybeTick() default: log.Fatal("Unexpected connection maker query:", query) } case now := <-tick: for address, target := range cm.targets { if target.conn == nil && !target.attempting && now.After(target.tryAfter) { - if target.attemptCount == MaxAttemptCount { - // FIXME - continue - } - after, interval := tryAfter(target.tryInterval) - target.tryInterval = interval - target.tryAfter = after target.attemptCount += 1 target.attempting = true; go cm.attemptConnection(address, target.commandLine) @@ -130,7 +128,7 @@ func (cm *ConnectionMaker) status() string { if (target.conn != nil) { buf.WriteString(fmt.Sprintf("%s connected to: %v\n", address, target.conn)) } else if target.attempting { - buf.WriteString(fmt.Sprintf("%s (%v attempts, trying again now)\n", address, target.attemptCount)) + 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)) } @@ -139,9 +137,9 @@ func (cm *ConnectionMaker) status() string { } func (cm *ConnectionMaker) attemptConnection(address string, acceptNewPeer bool) { - log.Println("Attempting connection to ", address) + log.Println("Attempting connection to", address) var conncode = CMConnSucceeded - if err := cm.router.Ourself.CreateConnection(address, UnknownPeerName); err != nil { + if err := cm.router.Ourself.CreateConnection(address, acceptNewPeer); err != nil { log.Println(err) conncode = CMConnFailed } diff --git a/peer.go b/peer.go index b01a3053..b818a7b5 100644 --- a/peer.go +++ b/peer.go @@ -190,7 +190,7 @@ 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 } @@ -216,7 +216,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 { From dec794eb72cb72f985013ac429adaf253a4f8d57 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Mon, 8 Sep 2014 15:05:50 +0100 Subject: [PATCH 09/38] Remove all references to PeerName in ConnectionMaker --- connection.go | 2 +- connection_maker.go | 52 ++++++++++++++++++++++----------------------- peer.go | 5 +++-- types.go | 6 +++--- weaver/main.go | 2 +- 5 files changed, 33 insertions(+), 34 deletions(-) diff --git a/connection.go b/connection.go index 138df0c6..37363b19 100644 --- a/connection.go +++ b/connection.go @@ -279,7 +279,7 @@ func (conn *LocalConnection) handleShutdown() { conn.stopForwarders() if conn.remote != nil { - conn.Router.ConnectionMaker.EnsureConnection(conn.remote.Name, conn.remoteTCPAddr) + conn.Router.ConnectionMaker.EnsureConnection(conn.remoteTCPAddr) } } diff --git a/connection_maker.go b/connection_maker.go index df7346df..73530b46 100644 --- a/connection_maker.go +++ b/connection_maker.go @@ -1,13 +1,5 @@ package weave -/* ConnectionMaker is responsible for making connections between peers, -and retrying them when they break. It sits in a loop waiting for requests -to be sent over queryChan, or for timer ticks. -Every connection has a PeerName, which is a unique identifier for every weave -router (e.g. a MAC address), and an IP address which it was foundAt. - -*/ - import ( "bytes" "fmt" @@ -38,11 +30,18 @@ func StartConnectionMaker(router *Router) *ConnectionMaker { 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} + acceptAnyPeer: true, + address: address} +} + +func (cm *ConnectionMaker) EnsureConnection(address string) { + cm.queryChan <- &ConnectionMakerInteraction{ + Interaction: Interaction{code: CMEnsure}, + acceptAnyPeer: false, + address: address} } func (cm *ConnectionMaker) String() string { @@ -69,16 +68,14 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio } switch { case query.code == CMEnsure: - if query.name != cm.router.Ourself.Name { - cm.addToTargets(query.name, query.foundAt) - maybeTick() - } + cm.addToTargets(query.acceptAnyPeer, query.address) + maybeTick() case query.code == CMStatus: query.resultChan <- cm.status() case query.code == CMConnSucceeded: - cm.targets[query.foundAt].attempting = false + cm.targets[query.address].attempting = false case query.code == CMConnFailed: - target := cm.targets[query.foundAt] + target := cm.targets[query.address] target.attempting = false after, interval := tryAfter(target.tryInterval) target.tryInterval = interval @@ -92,7 +89,7 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio if target.conn == nil && !target.attempting && now.After(target.tryAfter) { target.attemptCount += 1 target.attempting = true; - go cm.attemptConnection(address, target.commandLine) + go cm.attemptConnection(address, target.acceptAnyPeer) } } tick = nil @@ -101,24 +98,26 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio } } -func (cm *ConnectionMaker) addToTargets(name PeerName, foundAt string) { - target := cm.targets[foundAt] +func (cm *ConnectionMaker) addToTargets(acceptAnyPeer bool, address string) { + target := cm.targets[address] if target == nil { after, interval := tryAfter(InitialInterval) target = &Target{ + acceptAnyPeer: acceptAnyPeer, tryInterval: interval, tryAfter: after} } - foundAtHost, foundAtPortStr, err := net.SplitHostPort(foundAt) + // FIXME: what does it mean if an address is added twice? + addrHost, addrPortStr, err := net.SplitHostPort(address) if err == nil { // ensure port-less version is there - cm.targets[foundAtHost] = target - if foundAtPort, err := strconv.Atoi(foundAtPortStr); err == nil && foundAtPort != Port { - cm.targets[foundAt] = target + cm.targets[addrHost] = target + if addrPort, err := strconv.Atoi(addrPortStr); err == nil && addrPort != Port { + cm.targets[address] = target } } else { // can't split it, assume it must not have port on it - cm.targets[foundAt] = target + cm.targets[address] = target } } @@ -146,8 +145,7 @@ func (cm *ConnectionMaker) attemptConnection(address string, acceptNewPeer bool) // Tell the query loop we've finished this attempt cm.queryChan <- &ConnectionMakerInteraction{ Interaction: Interaction{code: conncode}, - name: UnknownPeerName, - foundAt: address} + address: address} } func tryAfter(interval time.Duration) (time.Time, time.Duration) { diff --git a/peer.go b/peer.go index b818a7b5..9eef00dd 100644 --- a/peer.go +++ b/peer.go @@ -110,10 +110,11 @@ func (peer *Peer) SetVersionAndConnections(version uint64, connections map[PeerN peer.version = version peer.connections = connections for remoteName, conn := range connections { - peer.Router.ConnectionMaker.EnsureConnection(remoteName, conn.RemoteTCPAddr()) + if remoteName != peer.Router.Ourself.Name { + peer.Router.ConnectionMaker.EnsureConnection(conn.RemoteTCPAddr()) + } } } - func (peer *Peer) Forward(dstPeer *Peer, df bool, frame []byte, dec *EthernetDecoder) error { return peer.Relay(peer, dstPeer, df, frame, dec) } diff --git a/types.go b/types.go index d7ce9020..c13ab96c 100644 --- a/types.go +++ b/types.go @@ -155,8 +155,8 @@ type Topology struct { type ConnectionMakerInteraction struct { Interaction - foundAt string - name PeerName + acceptAnyPeer bool + address string } type ConnectionMaker struct { @@ -167,7 +167,7 @@ type ConnectionMaker struct { // Information about an address where we may find a peer type Target struct { - commandLine bool // was this address given to us on the command line? + acceptAnyPeer bool // was this address given to us on the command line? attempting bool // are we currently attempting to connect to this address? conn *Connection tryAfter time.Time // next time to try this address diff --git a/weaver/main.go b/weaver/main.go index 96dbaf21..1b2c2361 100644 --- a/weaver/main.go +++ b/weaver/main.go @@ -124,7 +124,7 @@ func main() { router := weave.NewRouter(iface, ourName, []byte(password), connLimit, bufSz*1024*1024, logFrame) router.Start() for _, peer := range peers { - router.ConnectionMaker.EnsureConnection(weave.UnknownPeerName, peer) + router.ConnectionMaker.InitiateConnection(peer) } go handleHttp(router) handleSignals(router) From 5cf2cb0c5009b248d3bf25968eb4e1075e8ff390 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Mon, 8 Sep 2014 15:36:19 +0100 Subject: [PATCH 10/38] Keep track of which connections have been established --- connection.go | 1 + connection_maker.go | 41 ++++++++++++++++++++++++++--------------- types.go | 2 +- 3 files changed, 28 insertions(+), 16 deletions(-) diff --git a/connection.go b/connection.go index 37363b19..6567a091 100644 --- a/connection.go +++ b/connection.go @@ -447,6 +447,7 @@ 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{ diff --git a/connection_maker.go b/connection_maker.go index 73530b46..4049cb6e 100644 --- a/connection_maker.go +++ b/connection_maker.go @@ -6,7 +6,6 @@ import ( "log" "math/rand" "net" - "strconv" "time" ) @@ -16,6 +15,7 @@ const ( MaxAttemptCount = 100 CMEnsure = iota CMStatus = iota + CMEstablished = iota CMConnSucceeded = iota CMConnFailed = iota ) @@ -38,12 +38,26 @@ func (cm *ConnectionMaker) InitiateConnection(address string) { } func (cm *ConnectionMaker) EnsureConnection(address string) { + // If we've been given a port number, take it off + if addrHost, _, err := net.SplitHostPort(address); err == nil { + address = addrHost + } cm.queryChan <- &ConnectionMakerInteraction{ Interaction: Interaction{code: CMEnsure}, acceptAnyPeer: false, address: address} } +func (cm *ConnectionMaker) ConnectionEstablished(address string) { + // If we've been given a port number, take it off + if addrHost, _, err := net.SplitHostPort(address); err == nil { + address = addrHost + } + cm.queryChan <- &ConnectionMakerInteraction{ + Interaction: Interaction{code: CMEstablished}, + address: address} +} + func (cm *ConnectionMaker) String() string { resultChan := make(chan interface{}, 0) cm.queryChan <- &ConnectionMakerInteraction{ @@ -72,6 +86,12 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio maybeTick() 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 + } case query.code == CMConnSucceeded: cm.targets[query.address].attempting = false case query.code == CMConnFailed: @@ -86,7 +106,7 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio } case now := <-tick: for address, target := range cm.targets { - if target.conn == nil && !target.attempting && now.After(target.tryAfter) { + if !target.established && !target.attempting && now.After(target.tryAfter) { target.attemptCount += 1 target.attempting = true; go cm.attemptConnection(address, target.acceptAnyPeer) @@ -107,25 +127,16 @@ func (cm *ConnectionMaker) addToTargets(acceptAnyPeer bool, address string) { tryInterval: interval, tryAfter: after} } + target.established = false // this seems a crap place to do this // FIXME: what does it mean if an address is added twice? - addrHost, addrPortStr, err := net.SplitHostPort(address) - if err == nil { - // ensure port-less version is there - cm.targets[addrHost] = target - if addrPort, err := strconv.Atoi(addrPortStr); err == nil && addrPort != Port { - cm.targets[address] = target - } - } else { - // can't split it, assume it must not have port on it - cm.targets[address] = target - } + cm.targets[address] = target } func (cm *ConnectionMaker) status() string { var buf bytes.Buffer for address, target := range cm.targets { - if (target.conn != nil) { - buf.WriteString(fmt.Sprintf("%s connected to: %v\n", address, target.conn)) + if (target.established) { + buf.WriteString(fmt.Sprintf("%s connected\n", address)) } else if target.attempting { buf.WriteString(fmt.Sprintf("%s (%v attempts, trying since %v)\n", address, target.attemptCount, target.tryAfter)) } else { diff --git a/types.go b/types.go index c13ab96c..5a2baa36 100644 --- a/types.go +++ b/types.go @@ -169,7 +169,7 @@ type ConnectionMaker struct { 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? - conn *Connection + 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 From a3312c822c40b197a5b8f5de3c225de4a92935b8 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Mon, 8 Sep 2014 16:05:54 +0100 Subject: [PATCH 11/38] Attempt to connect immediately when given a new connection --- connection_maker.go | 34 +++++++++++++++++++++++++--------- 1 file changed, 25 insertions(+), 9 deletions(-) diff --git a/connection_maker.go b/connection_maker.go index 4049cb6e..97476115 100644 --- a/connection_maker.go +++ b/connection_maker.go @@ -10,7 +10,7 @@ import ( ) const ( - InitialInterval = 1 * time.Second + InitialInterval = 5 * time.Second MaxInterval = 10 * time.Minute MaxAttemptCount = 100 CMEnsure = iota @@ -74,6 +74,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,14 +88,16 @@ 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 + } else { + log.Fatal("Connection established to unknown address", query.address) } case query.code == CMConnSucceeded: cm.targets[query.address].attempting = false @@ -121,15 +128,19 @@ 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{ 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] + target.established = false + after, interval := tryImmediately() + target.tryInterval = interval + target.tryAfter = after } func (cm *ConnectionMaker) status() string { @@ -159,6 +170,11 @@ func (cm *ConnectionMaker) attemptConnection(address string, acceptNewPeer bool) 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) { interval += time.Duration(rand.Int63n(int64(interval))) if interval > MaxInterval { From 16e9daf057e95f9070964d957dc48a14327d28cf Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Mon, 8 Sep 2014 16:25:55 +0100 Subject: [PATCH 12/38] Reformatted with 'go fmt' --- connection.go | 6 +++--- connection_maker.go | 18 +++++++++--------- peer_cache.go | 2 +- types.go | 18 +++++++++--------- weaver/main.go | 5 ++--- 5 files changed, 24 insertions(+), 25 deletions(-) diff --git a/connection.go b/connection.go index 6567a091..0b8a289f 100644 --- a/connection.go +++ b/connection.go @@ -99,7 +99,7 @@ func (conn *LocalConnection) CheckFatal(err error) error { func (conn *LocalConnection) setEffectivePMTU(pmtu int) { conn.Lock() defer conn.Unlock() - if (conn.effectivePMTU != pmtu) { + if conn.effectivePMTU != pmtu { conn.effectivePMTU = pmtu conn.log("Effective PMTU set to", pmtu) } @@ -345,8 +345,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) } } diff --git a/connection_maker.go b/connection_maker.go index 97476115..d3d5822b 100644 --- a/connection_maker.go +++ b/connection_maker.go @@ -23,16 +23,16 @@ const ( 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 +43,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) { @@ -115,7 +115,7 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio for address, target := range cm.targets { if !target.established && !target.attempting && now.After(target.tryAfter) { target.attemptCount += 1 - target.attempting = true; + target.attempting = true go cm.attemptConnection(address, target.acceptAnyPeer) } } @@ -140,13 +140,13 @@ func (cm *ConnectionMaker) resetTarget(address string) { target.established = false after, interval := tryImmediately() target.tryInterval = interval - target.tryAfter = after + target.tryAfter = after } func (cm *ConnectionMaker) status() string { var buf bytes.Buffer for address, target := range cm.targets { - if (target.established) { + if target.established { buf.WriteString(fmt.Sprintf("%s connected\n", address)) } else if target.attempting { buf.WriteString(fmt.Sprintf("%s (%v attempts, trying since %v)\n", address, target.attemptCount, target.tryAfter)) diff --git a/peer_cache.go b/peer_cache.go index fc0c2b2a..390a164c 100644 --- a/peer_cache.go +++ b/peer_cache.go @@ -11,7 +11,7 @@ import ( type PeerCache struct { sync.RWMutex table map[PeerName]*Peer - onGC func(*Peer) + onGC func(*Peer) } func NewPeerCache(onGC func(*Peer)) *PeerCache { diff --git a/types.go b/types.go index 5a2baa36..e9a8462b 100644 --- a/types.go +++ b/types.go @@ -160,19 +160,19 @@ type ConnectionMakerInteraction struct { } type ConnectionMaker struct { - router *Router - queryChan chan<- *ConnectionMakerInteraction - targets map[string]*Target + router *Router + queryChan chan<- *ConnectionMakerInteraction + targets map[string]*Target } // 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? + 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 } // UDPSender interface and implementations diff --git a/weaver/main.go b/weaver/main.go index 1b2c2361..63bb60d1 100644 --- a/weaver/main.go +++ b/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) From f3cf2b598de577a6642384e08431f08d1ed514c3 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Mon, 8 Sep 2014 16:33:04 +0100 Subject: [PATCH 13/38] Add back blank line that got spuriously deleted --- peer.go | 1 + 1 file changed, 1 insertion(+) diff --git a/peer.go b/peer.go index 9eef00dd..fb43ccef 100644 --- a/peer.go +++ b/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) } From 64999d830d35319f2ecbc3f5ec1179831ddbf36a Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Mon, 8 Sep 2014 17:30:54 +0100 Subject: [PATCH 14/38] State of connection is now three-way choice rather than two booleans --- connection_maker.go | 46 +++++++++++++++++++++++---------------------- types.go | 7 ++++--- 2 files changed, 28 insertions(+), 25 deletions(-) diff --git a/connection_maker.go b/connection_maker.go index d3d5822b..ebdd8b1b 100644 --- a/connection_maker.go +++ b/connection_maker.go @@ -16,10 +16,15 @@ const ( 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{ @@ -95,27 +100,23 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio case query.code == CMEstablished: target := cm.targets[query.address] 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) } } @@ -129,6 +130,7 @@ func (cm *ConnectionMaker) addToTargets(acceptAnyPeer bool, address string) { target := cm.targets[address] if target == nil { target = &Target{ + state: CMUnconnected, acceptAnyPeer: acceptAnyPeer, } cm.targets[address] = target @@ -137,18 +139,21 @@ func (cm *ConnectionMaker) addToTargets(acceptAnyPeer bool, address string) { func (cm *ConnectionMaker) resetTarget(address string) { target := cm.targets[address] - target.established = false - after, interval := tryImmediately() - target.tryInterval = interval - target.tryAfter = after + // 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)) @@ -159,15 +164,12 @@ 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) { diff --git a/types.go b/types.go index e9a8462b..53178072 100644 --- a/types.go +++ b/types.go @@ -165,11 +165,12 @@ type ConnectionMaker struct { 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 + 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 From f8f32a60ba9fbd6a786a0ff5e47a169485170424 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Mon, 8 Sep 2014 17:31:38 +0100 Subject: [PATCH 15/38] Move call to tell ConnectionMaker that connection is established onto the queryLoop goroutine to avoid races with shutdown events --- connection.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/connection.go b/connection.go index 0b8a289f..eba52584 100644 --- a/connection.go +++ b/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 @@ -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{ From 2f53df8a97878bb64851ab1a9cb7b806a125d2dd Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Tue, 9 Sep 2014 11:27:46 +0100 Subject: [PATCH 16/38] Check if we already have a connection on an address before attempting to connect to it --- connection_maker.go | 10 +++++++--- peer.go | 31 ++++++++++++++++++++++++------- 2 files changed, 31 insertions(+), 10 deletions(-) diff --git a/connection_maker.go b/connection_maker.go index ebdd8b1b..e7822d26 100644 --- a/connection_maker.go +++ b/connection_maker.go @@ -115,9 +115,13 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio case now := <-tick: for address, target := range cm.targets { if target.state == CMUnconnected && now.After(target.tryAfter) { - target.attemptCount += 1 - target.state = CMAttempting - go cm.attemptConnection(address, target.acceptAnyPeer) + if _, found := cm.router.Ourself.ConnectionOn(address); found { + target.state = CMEstablished + } else { + target.attemptCount += 1 + target.state = CMAttempting + go cm.attemptConnection(address, target.acceptAnyPeer) + } } } tick = nil diff --git a/peer.go b/peer.go index fb43ccef..9bdd3968 100644 --- a/peer.go +++ b/peer.go @@ -57,6 +57,29 @@ func (peer *Peer) ConnectionTo(name PeerName) (Connection, bool) { return conn, found // yes, you really can't inline that. FFS. } +// 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) + } +} + +func (peer *Peer) ConnectionOn(peerAddr string) (Connection, bool) { + peerAddr = normalisePeerAddr(peerAddr) + peer.RLock() + defer peer.RUnlock() + for _, conn := range peer.connections { + if conn.RemoteTCPAddr() == peerAddr { + return conn, true + } + } + return nil, false +} + func (peer *Peer) ForEachConnection(fun func(PeerName, Connection)) { peer.RLock() defer peer.RUnlock() @@ -197,13 +220,7 @@ func (peer *Peer) CreateConnection(peerAddr string, acceptNewPeer bool) error { 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 { From 8271cbf0077115ecae9b647db8b533a7ec28487a Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Tue, 9 Sep 2014 11:45:09 +0100 Subject: [PATCH 17/38] In status message, change "Reconnects" to "Connections" to match what it now does --- router.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/router.go b/router.go index ec2d9de1..bdaa9d3d 100644 --- a/router.go +++ b/router.go @@ -63,7 +63,7 @@ func (router *Router) Status() string { buf.WriteString(fmt.Sprintf("MACs:\n%s", router.Macs)) buf.WriteString(fmt.Sprintf("Peers:\n%s", router.Peers)) buf.WriteString(fmt.Sprintf("Topology:\n%s", router.Topology)) - buf.WriteString(fmt.Sprintf("Reconnects:\n%s", router.ConnectionMaker)) + buf.WriteString(fmt.Sprintf("Connections:\n%s", router.ConnectionMaker)) return buf.String() } From 527ddfca0d2a50b6b1f63cdb2dbc7c8a3c1fd6f9 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Tue, 9 Sep 2014 12:04:05 +0100 Subject: [PATCH 18/38] Clean up naming of commands and states --- connection_maker.go | 26 +++++++++++++------------- 1 file changed, 13 insertions(+), 13 deletions(-) diff --git a/connection_maker.go b/connection_maker.go index e7822d26..a9d6edf4 100644 --- a/connection_maker.go +++ b/connection_maker.go @@ -20,9 +20,9 @@ const ( ) const ( - CMUnconnected ConnectionState = iota - CMAttempting = iota - CMEstablishedState = iota + CSUnconnected ConnectionState = iota + CSAttempting = iota + CSEstablished = iota ) func StartConnectionMaker(router *Router) *ConnectionMaker { @@ -100,13 +100,13 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio case query.code == CMEstablished: target := cm.targets[query.address] if target != nil { - target.state = CMEstablishedState + target.state = CSEstablished } else { log.Fatal("Connection established to unknown address", query.address) } case query.code == CMConnFailed: target := cm.targets[query.address] - target.state = CMUnconnected + target.state = CSUnconnected target.tryAfter, target.tryInterval = tryAfter(target.tryInterval) maybeTick() default: @@ -114,12 +114,12 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio } case now := <-tick: for address, target := range cm.targets { - if target.state == CMUnconnected && now.After(target.tryAfter) { + if target.state == CSUnconnected && now.After(target.tryAfter) { if _, found := cm.router.Ourself.ConnectionOn(address); found { - target.state = CMEstablished + target.state = CSEstablished } else { target.attemptCount += 1 - target.state = CMAttempting + target.state = CSAttempting go cm.attemptConnection(address, target.acceptAnyPeer) } } @@ -134,7 +134,7 @@ func (cm *ConnectionMaker) addToTargets(acceptAnyPeer bool, address string) { target := cm.targets[address] if target == nil { target = &Target{ - state: CMUnconnected, + state: CSUnconnected, acceptAnyPeer: acceptAnyPeer, } cm.targets[address] = target @@ -144,20 +144,20 @@ func (cm *ConnectionMaker) addToTargets(acceptAnyPeer bool, address string) { 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 { + if target.state == CSAttempting { target.tryAfter, target.tryInterval = tryAfter(target.tryInterval) } else { target.tryAfter, target.tryInterval = tryImmediately() } - target.state = CMUnconnected + target.state = CSUnconnected } func (cm *ConnectionMaker) status() string { var buf bytes.Buffer for address, target := range cm.targets { - if target.state == CMEstablishedState { + if target.state == CSEstablished { buf.WriteString(fmt.Sprintf("%s connected\n", address)) - } else if target.state == CMAttempting { + } else 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)) From c074c2991d28127b8cfbf8fce3ae90b8d93e932b Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Tue, 9 Sep 2014 13:03:14 +0100 Subject: [PATCH 19/38] Pull a couple of common bits of code out into utils.go --- connection_maker.go | 17 ++++++----------- peer.go | 15 ++------------- utils.go | 19 +++++++++++++++++++ 3 files changed, 27 insertions(+), 24 deletions(-) diff --git a/connection_maker.go b/connection_maker.go index a9d6edf4..0e4c04e7 100644 --- a/connection_maker.go +++ b/connection_maker.go @@ -5,7 +5,6 @@ import ( "fmt" "log" "math/rand" - "net" "time" ) @@ -43,10 +42,7 @@ func (cm *ConnectionMaker) InitiateConnection(address string) { } func (cm *ConnectionMaker) EnsureConnection(address string) { - // If we've been given a port number, take it off - if addrHost, _, err := net.SplitHostPort(address); err == nil { - address = addrHost - } + address = StripPortFromAddr(address) cm.queryChan <- &ConnectionMakerInteraction{ Interaction: Interaction{code: CMEnsure}, acceptAnyPeer: false, @@ -54,10 +50,7 @@ func (cm *ConnectionMaker) EnsureConnection(address string) { } func (cm *ConnectionMaker) ConnectionEstablished(address string) { - // If we've been given a port number, take it off - if addrHost, _, err := net.SplitHostPort(address); err == nil { - address = addrHost - } + address = StripPortFromAddr(address) cm.queryChan <- &ConnectionMakerInteraction{ Interaction: Interaction{code: CMEstablished}, address: address} @@ -92,6 +85,7 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio } switch { case query.code == CMEnsure: + log.Println("Connection ensure", query.address) cm.addToTargets(query.acceptAnyPeer, query.address) cm.resetTarget(query.address) tickNow() @@ -100,6 +94,7 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio case query.code == CMEstablished: target := cm.targets[query.address] if target != nil { + log.Println("Connection established to", query.address, target) target.state = CSEstablished } else { log.Fatal("Connection established to unknown address", query.address) @@ -115,9 +110,9 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio 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 { + /*if _, found := cm.router.Ourself.ConnectionOn(address); found { target.state = CSEstablished - } else { + } else*/ { target.attemptCount += 1 target.state = CSAttempting go cm.attemptConnection(address, target.acceptAnyPeer) diff --git a/peer.go b/peer.go index 9bdd3968..b09771ee 100644 --- a/peer.go +++ b/peer.go @@ -57,19 +57,8 @@ func (peer *Peer) ConnectionTo(name PeerName) (Connection, bool) { return conn, found // yes, you really can't inline that. FFS. } -// 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) - } -} - func (peer *Peer) ConnectionOn(peerAddr string) (Connection, bool) { - peerAddr = normalisePeerAddr(peerAddr) + peerAddr = NormalisePeerAddr(peerAddr) peer.RLock() defer peer.RUnlock() for _, conn := range peer.connections { @@ -220,7 +209,7 @@ func (peer *Peer) CreateConnection(peerAddr string, acceptNewPeer bool) error { return err } // We're dialing the remote so that means connections will come from random ports - addrStr := normalisePeerAddr(peerAddr) + addrStr := NormalisePeerAddr(peerAddr) tcpAddr, tcpErr := net.ResolveTCPAddr("tcp4", addrStr) udpAddr, udpErr := net.ResolveUDPAddr("udp4", addrStr) if tcpErr != nil || udpErr != nil { diff --git a/utils.go b/utils.go index 22d7235b..8d9a54bc 100644 --- a/utils.go +++ b/utils.go @@ -97,3 +97,22 @@ 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 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) + } +} From c1856e1ff16d5f100d2fb3a445c921656872a7ee Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Tue, 9 Sep 2014 13:06:37 +0100 Subject: [PATCH 20/38] Move a couple of commmon bits of code out to utils.go --- connection_maker.go | 11 ++--------- peer.go | 15 ++------------- utils.go | 19 +++++++++++++++++++ 3 files changed, 23 insertions(+), 22 deletions(-) diff --git a/connection_maker.go b/connection_maker.go index a9d6edf4..77845ce8 100644 --- a/connection_maker.go +++ b/connection_maker.go @@ -5,7 +5,6 @@ import ( "fmt" "log" "math/rand" - "net" "time" ) @@ -43,10 +42,7 @@ func (cm *ConnectionMaker) InitiateConnection(address string) { } func (cm *ConnectionMaker) EnsureConnection(address string) { - // If we've been given a port number, take it off - if addrHost, _, err := net.SplitHostPort(address); err == nil { - address = addrHost - } + address = StripPortFromAddr(address) cm.queryChan <- &ConnectionMakerInteraction{ Interaction: Interaction{code: CMEnsure}, acceptAnyPeer: false, @@ -54,10 +50,7 @@ func (cm *ConnectionMaker) EnsureConnection(address string) { } func (cm *ConnectionMaker) ConnectionEstablished(address string) { - // If we've been given a port number, take it off - if addrHost, _, err := net.SplitHostPort(address); err == nil { - address = addrHost - } + address = StripPortFromAddr(address) cm.queryChan <- &ConnectionMakerInteraction{ Interaction: Interaction{code: CMEstablished}, address: address} diff --git a/peer.go b/peer.go index 9bdd3968..b09771ee 100644 --- a/peer.go +++ b/peer.go @@ -57,19 +57,8 @@ func (peer *Peer) ConnectionTo(name PeerName) (Connection, bool) { return conn, found // yes, you really can't inline that. FFS. } -// 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) - } -} - func (peer *Peer) ConnectionOn(peerAddr string) (Connection, bool) { - peerAddr = normalisePeerAddr(peerAddr) + peerAddr = NormalisePeerAddr(peerAddr) peer.RLock() defer peer.RUnlock() for _, conn := range peer.connections { @@ -220,7 +209,7 @@ func (peer *Peer) CreateConnection(peerAddr string, acceptNewPeer bool) error { return err } // We're dialing the remote so that means connections will come from random ports - addrStr := normalisePeerAddr(peerAddr) + addrStr := NormalisePeerAddr(peerAddr) tcpAddr, tcpErr := net.ResolveTCPAddr("tcp4", addrStr) udpAddr, udpErr := net.ResolveUDPAddr("udp4", addrStr) if tcpErr != nil || udpErr != nil { diff --git a/utils.go b/utils.go index 22d7235b..8d9a54bc 100644 --- a/utils.go +++ b/utils.go @@ -97,3 +97,22 @@ 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 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) + } +} From dc927314ba21d99aab676d46787b74f08bf9254d Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Tue, 9 Sep 2014 13:07:55 +0100 Subject: [PATCH 21/38] Revert "Move a couple of commmon bits of code out to utils.go" This reverts commit c1856e1ff16d5f100d2fb3a445c921656872a7ee. --- connection_maker.go | 11 +++++++++-- peer.go | 15 +++++++++++++-- utils.go | 19 ------------------- 3 files changed, 22 insertions(+), 23 deletions(-) diff --git a/connection_maker.go b/connection_maker.go index 77845ce8..a9d6edf4 100644 --- a/connection_maker.go +++ b/connection_maker.go @@ -5,6 +5,7 @@ import ( "fmt" "log" "math/rand" + "net" "time" ) @@ -42,7 +43,10 @@ func (cm *ConnectionMaker) InitiateConnection(address string) { } func (cm *ConnectionMaker) EnsureConnection(address string) { - address = StripPortFromAddr(address) + // If we've been given a port number, take it off + if addrHost, _, err := net.SplitHostPort(address); err == nil { + address = addrHost + } cm.queryChan <- &ConnectionMakerInteraction{ Interaction: Interaction{code: CMEnsure}, acceptAnyPeer: false, @@ -50,7 +54,10 @@ func (cm *ConnectionMaker) EnsureConnection(address string) { } func (cm *ConnectionMaker) ConnectionEstablished(address string) { - address = StripPortFromAddr(address) + // If we've been given a port number, take it off + if addrHost, _, err := net.SplitHostPort(address); err == nil { + address = addrHost + } cm.queryChan <- &ConnectionMakerInteraction{ Interaction: Interaction{code: CMEstablished}, address: address} diff --git a/peer.go b/peer.go index b09771ee..9bdd3968 100644 --- a/peer.go +++ b/peer.go @@ -57,8 +57,19 @@ func (peer *Peer) ConnectionTo(name PeerName) (Connection, bool) { return conn, found // yes, you really can't inline that. FFS. } +// 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) + } +} + func (peer *Peer) ConnectionOn(peerAddr string) (Connection, bool) { - peerAddr = NormalisePeerAddr(peerAddr) + peerAddr = normalisePeerAddr(peerAddr) peer.RLock() defer peer.RUnlock() for _, conn := range peer.connections { @@ -209,7 +220,7 @@ func (peer *Peer) CreateConnection(peerAddr string, acceptNewPeer bool) error { return err } // We're dialing the remote so that means connections will come from random ports - addrStr := NormalisePeerAddr(peerAddr) + addrStr := normalisePeerAddr(peerAddr) tcpAddr, tcpErr := net.ResolveTCPAddr("tcp4", addrStr) udpAddr, udpErr := net.ResolveUDPAddr("udp4", addrStr) if tcpErr != nil || udpErr != nil { diff --git a/utils.go b/utils.go index 8d9a54bc..22d7235b 100644 --- a/utils.go +++ b/utils.go @@ -97,22 +97,3 @@ 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 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) - } -} From 42ce45dd551b86371e81fb25ab9f3f03b9df41f5 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Tue, 9 Sep 2014 13:10:37 +0100 Subject: [PATCH 22/38] remove debugging code added by accident --- connection_maker.go | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/connection_maker.go b/connection_maker.go index 0e4c04e7..77845ce8 100644 --- a/connection_maker.go +++ b/connection_maker.go @@ -85,7 +85,6 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio } switch { case query.code == CMEnsure: - log.Println("Connection ensure", query.address) cm.addToTargets(query.acceptAnyPeer, query.address) cm.resetTarget(query.address) tickNow() @@ -94,7 +93,6 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio case query.code == CMEstablished: target := cm.targets[query.address] if target != nil { - log.Println("Connection established to", query.address, target) target.state = CSEstablished } else { log.Fatal("Connection established to unknown address", query.address) @@ -110,9 +108,9 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio 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 { + if _, found := cm.router.Ourself.ConnectionOn(address); found { target.state = CSEstablished - } else*/ { + } else { target.attemptCount += 1 target.state = CSAttempting go cm.attemptConnection(address, target.acceptAnyPeer) From 84aeaf340a2816e498faa8e1858f9f432e825d04 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Wed, 10 Sep 2014 12:08:49 +0100 Subject: [PATCH 23/38] New implementation of ConnectionMaker just periodically scanning who we need to connect to --- connection.go | 1 - connection_maker.go | 129 ++++++++++++++++++++++++++------------------ types.go | 14 ++--- utils.go | 15 ++++++ 4 files changed, 98 insertions(+), 61 deletions(-) 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 { From 26f8a68a82c8c0a9e63d1dd2263f1e4af4c8bb96 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Wed, 10 Sep 2014 12:29:21 +0100 Subject: [PATCH 24/38] Remove stuff that is not needed with new ConnectionMaker approach --- connection.go | 4 ---- connection_maker.go | 7 +------ peer.go | 5 ----- types.go | 1 - 4 files changed, 1 insertion(+), 16 deletions(-) diff --git a/connection.go b/connection.go index 4545f0a1..f0aed6f9 100644 --- a/connection.go +++ b/connection.go @@ -277,10 +277,6 @@ func (conn *LocalConnection) handleShutdown() { // blank out the forwardChan so that the router processes don't // try to send any more conn.stopForwarders() - - if conn.remote != nil { - conn.Router.ConnectionMaker.EnsureConnection(conn.remoteTCPAddr) - } } func (conn *LocalConnection) handshake(acceptNewPeer bool) error { diff --git a/connection_maker.go b/connection_maker.go index b15c7a5d..01107664 100644 --- a/connection_maker.go +++ b/connection_maker.go @@ -37,14 +37,9 @@ func StartConnectionMaker(router *Router) *ConnectionMaker { func (cm *ConnectionMaker) InitiateConnection(address string) { cm.queryChan <- &ConnectionMakerInteraction{ Interaction: Interaction{code: CMInitiate}, - isCmdLine: true, address: address} } -func (cm *ConnectionMaker) EnsureConnection(address string) { - // Do nothing - we will figure out what to do later -} - func (cm *ConnectionMaker) String() string { resultChan := make(chan interface{}, 0) cm.queryChan <- &ConnectionMakerInteraction{ @@ -74,7 +69,7 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio } switch { case query.code == CMInitiate: - cm.addToTargets(query.isCmdLine, query.address) + cm.addToTargets(true, query.address) tickNow() case query.code == CMStatus: query.resultChan <- cm.status() diff --git a/peer.go b/peer.go index b09771ee..4d3dd0c6 100644 --- a/peer.go +++ b/peer.go @@ -121,11 +121,6 @@ func (peer *Peer) SetVersionAndConnections(version uint64, connections map[PeerN defer peer.Unlock() peer.version = version peer.connections = connections - for remoteName, conn := range connections { - if remoteName != peer.Router.Ourself.Name { - peer.Router.ConnectionMaker.EnsureConnection(conn.RemoteTCPAddr()) - } - } } func (peer *Peer) Forward(dstPeer *Peer, df bool, frame []byte, dec *EthernetDecoder) error { diff --git a/types.go b/types.go index e5effe91..c3b93394 100644 --- a/types.go +++ b/types.go @@ -155,7 +155,6 @@ type Topology struct { type ConnectionMakerInteraction struct { Interaction - isCmdLine bool address string } From 3644531c50455616f4e26465d8ac37250339c891 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Wed, 10 Sep 2014 12:35:28 +0100 Subject: [PATCH 25/38] change counts to boolean when deciding if everything is connnected --- connection_maker.go | 26 +++++++++++--------------- 1 file changed, 11 insertions(+), 15 deletions(-) diff --git a/connection_maker.go b/connection_maker.go index 01107664..05fe38f3 100644 --- a/connection_maker.go +++ b/connection_maker.go @@ -95,7 +95,7 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio func (cm *ConnectionMaker) checkStateAndAttemptConnections(now time.Time) { ourself := cm.router.Ourself - count_unconnected_commandline := 0 + has_unconnected_commandline := false // 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 { @@ -112,18 +112,18 @@ func (cm *ConnectionMaker) checkStateAndAttemptConnections(now time.Time) { if target.state != CSAttempting { target.state = CSUnconnected } - count_unconnected_commandline += 1 + has_unconnected_commandline = true } } // Look for peers that we don't have a connection to. - count_unconnected_peers := 0 + has_unconnected_peers := false 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 + has_unconnected_peers = true // 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 { @@ -136,17 +136,13 @@ func (cm *ConnectionMaker) checkStateAndAttemptConnections(now time.Time) { } }) - 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) + if has_unconnected_commandline || has_unconnected_peers { + 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) + } } } } From b3b3ba9a21b94eca624857670599b5c4c6f23883 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Wed, 10 Sep 2014 13:08:20 +0100 Subject: [PATCH 26/38] cleaned up reconnect algorithm to improve locking, replace count with bool --- connection_maker.go | 43 ++++++++++++++++++++++--------------------- 1 file changed, 22 insertions(+), 21 deletions(-) diff --git a/connection_maker.go b/connection_maker.go index 05fe38f3..15d27400 100644 --- a/connection_maker.go +++ b/connection_maker.go @@ -95,7 +95,6 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio func (cm *ConnectionMaker) checkStateAndAttemptConnections(now time.Time) { ourself := cm.router.Ourself - has_unconnected_commandline := false // 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 { @@ -112,31 +111,33 @@ func (cm *ConnectionMaker) checkStateAndAttemptConnections(now time.Time) { if target.state != CSAttempting { target.state = CSUnconnected } - has_unconnected_commandline = true } } - // Look for peers that we don't have a connection to. - has_unconnected_peers := false - 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) - has_unconnected_peers = true - // 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) - } - } - } + // build a map of peers we are connected to, so we can access it without locking + our_connected_peers := make(map[PeerName]bool) + ourself.ForEachConnection(func(peer PeerName, _ Connection) { + our_connected_peers[peer] = true }) - if has_unconnected_commandline || has_unconnected_peers { + // Now look for peers that someone else is connected to, but we don't have a connection to. + cm.router.Peers.ForEach(func(name PeerName, peer *Peer) { + peer.ForEachConnection(func(peer2 PeerName, conn Connection) { + if peer2 != ourself.Name { + if _, found := our_connected_peers[peer2]; !found { + address := conn.RemoteTCPAddr() + if host, port, err := ExtractHostPort(address); err == nil { + if port != Port { + cm.addToTargets(false, address) + } + cm.addToTargets(false, host) + } + } + } + }) + }) + + { for address, target := range cm.targets { if target.state == CSUnconnected && now.After(target.tryAfter) { target.attemptCount += 1 From cd92fad3695a961e2b8631d1f4f8241f76e12e9b Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Wed, 10 Sep 2014 13:15:56 +0100 Subject: [PATCH 27/38] Simplify lookup in our_connected_peers --- connection_maker.go | 14 ++++++-------- 1 file changed, 6 insertions(+), 8 deletions(-) diff --git a/connection_maker.go b/connection_maker.go index 15d27400..9cd6b945 100644 --- a/connection_maker.go +++ b/connection_maker.go @@ -123,15 +123,13 @@ func (cm *ConnectionMaker) checkStateAndAttemptConnections(now time.Time) { // Now look for peers that someone else is connected to, but we don't have a connection to. cm.router.Peers.ForEach(func(name PeerName, peer *Peer) { peer.ForEachConnection(func(peer2 PeerName, conn Connection) { - if peer2 != ourself.Name { - if _, found := our_connected_peers[peer2]; !found { - address := conn.RemoteTCPAddr() - if host, port, err := ExtractHostPort(address); err == nil { - if port != Port { - cm.addToTargets(false, address) - } - cm.addToTargets(false, host) + if peer2 != ourself.Name && !our_connected_peers[peer2] { + address := conn.RemoteTCPAddr() + if host, port, err := ExtractHostPort(address); err == nil { + if port != Port { + cm.addToTargets(false, address) } + cm.addToTargets(false, host) } } }) From 634d4e0fe3c359f2f2c602d73399774563684c41 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Wed, 10 Sep 2014 13:45:56 +0100 Subject: [PATCH 28/38] cosmetic --- connection_maker.go | 14 +++++++------- types.go | 2 +- 2 files changed, 8 insertions(+), 8 deletions(-) diff --git a/connection_maker.go b/connection_maker.go index 9cd6b945..62df7e5f 100644 --- a/connection_maker.go +++ b/connection_maker.go @@ -95,6 +95,7 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio func (cm *ConnectionMaker) checkStateAndAttemptConnections(now time.Time) { ourself := cm.router.Ourself + // 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 { @@ -108,6 +109,7 @@ func (cm *ConnectionMaker) checkStateAndAttemptConnections(now time.Time) { delete(cm.targets, address) } } else if target.isCmdLine { + // This is a target that we are not connected to - if we are not currently attempting to connect, make sure we record it as unconnected if target.state != CSAttempting { target.state = CSUnconnected } @@ -135,13 +137,11 @@ func (cm *ConnectionMaker) checkStateAndAttemptConnections(now time.Time) { }) }) - { - 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) - } + 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) } } } diff --git a/types.go b/types.go index c3b93394..3c0c730d 100644 --- a/types.go +++ b/types.go @@ -155,7 +155,7 @@ type Topology struct { type ConnectionMakerInteraction struct { Interaction - address string + address string } type ConnectionMaker struct { From 4335615688210ac7a2a584aa6f94586dc92b99f8 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Wed, 10 Sep 2014 14:12:37 +0100 Subject: [PATCH 29/38] Track command-line targets in a separate set, to simplify tracking of targets --- connection_maker.go | 65 ++++++++++++++++++++++++--------------------- router.go | 2 +- types.go | 2 +- 3 files changed, 37 insertions(+), 32 deletions(-) diff --git a/connection_maker.go b/connection_maker.go index 62df7e5f..3fb7eb7e 100644 --- a/connection_maker.go +++ b/connection_maker.go @@ -21,7 +21,6 @@ const ( const ( CSUnconnected ConnectionState = iota CSAttempting = iota - CSEstablished = iota ) func StartConnectionMaker(router *Router) *ConnectionMaker { @@ -29,6 +28,7 @@ func StartConnectionMaker(router *Router) *ConnectionMaker { state := &ConnectionMaker{ router: router, queryChan: queryChan, + cmdLineAddress: make(map[string]bool), targets: make(map[string]*Target)} go state.queryLoop(queryChan) return state @@ -69,7 +69,7 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio } switch { case query.code == CMInitiate: - cm.addToTargets(true, query.address) + cm.cmdLineAddress[query.address] = true; tickNow() case query.code == CMStatus: query.resultChan <- cm.status() @@ -96,62 +96,69 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio func (cm *ConnectionMaker) checkStateAndAttemptConnections(now time.Time) { ourself := cm.router.Ourself + validTarget := make(map[string]bool) + // Any targets that are now connected, we don't need to attempt any more - for address, target := range cm.targets { + for address, _ := 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 { - // This is a target that we are not connected to - if we are not currently attempting to connect, make sure we record it as unconnected - if target.state != CSAttempting { - target.state = CSUnconnected - } + //log.Println("Deleting target now connected:", address) + delete(cm.targets, address) } } - // build a map of peers we are connected to, so we can access it without locking + // Add command-line targets that are not connected + for address, _ := range cm.cmdLineAddress { + if _, found := ourself.ConnectionOn(address); !found { + //log.Println("Unconnected cmdline:", address) + cm.addToTargets(address) + validTarget[address] = true + } + } + + // copy the set of peers we are connected to, so we can access it without locking our_connected_peers := make(map[PeerName]bool) ourself.ForEachConnection(func(peer PeerName, _ Connection) { our_connected_peers[peer] = true }) - // Now look for peers that someone else is connected to, but we don't have a connection to. + // 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 { - cm.addToTargets(false, address) + cm.addToTargets(address) + validTarget[address] = true } - cm.addToTargets(false, host) + cm.addToTargets(host) + validTarget[host] = true } } }) }) 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) + if validTarget[address] { + if target.state == CSUnconnected && now.After(target.tryAfter) { + target.attemptCount += 1 + target.state = CSAttempting + go cm.attemptConnection(address, cm.cmdLineAddress[address]) + } + } else { + //log.Println("Deleting target no longer valid:", address) + delete(cm.targets, address) } } } -func (cm *ConnectionMaker) addToTargets(isCmdLine bool, address string) { +func (cm *ConnectionMaker) addToTargets(address string) { target := cm.targets[address] if target == nil { target = &Target{ state: CSUnconnected, - isCmdLine: isCmdLine, } target.tryAfter, target.tryInterval = tryImmediately() cm.targets[address] = target @@ -161,9 +168,7 @@ func (cm *ConnectionMaker) addToTargets(isCmdLine bool, address string) { func (cm *ConnectionMaker) status() string { var buf bytes.Buffer for address, target := range cm.targets { - if target.state == CSEstablished { - buf.WriteString(fmt.Sprintf("%s connected\n", address)) - } else if target.state == CSAttempting { + 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)) diff --git a/router.go b/router.go index bdaa9d3d..de84ec2e 100644 --- a/router.go +++ b/router.go @@ -63,7 +63,7 @@ func (router *Router) Status() string { buf.WriteString(fmt.Sprintf("MACs:\n%s", router.Macs)) buf.WriteString(fmt.Sprintf("Peers:\n%s", router.Peers)) buf.WriteString(fmt.Sprintf("Topology:\n%s", router.Topology)) - buf.WriteString(fmt.Sprintf("Connections:\n%s", router.ConnectionMaker)) + buf.WriteString(fmt.Sprintf("Reconnections:\n%s", router.ConnectionMaker)) return buf.String() } diff --git a/types.go b/types.go index 3c0c730d..1626b5aa 100644 --- a/types.go +++ b/types.go @@ -162,13 +162,13 @@ type ConnectionMaker struct { router *Router queryChan chan<- *ConnectionMakerInteraction targets map[string]*Target + cmdLineAddress map[string]bool } type ConnectionState int // Information about an address where we may find a peer type Target struct { - 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 From 9d0af9dbfea5b1ddf906281cc3cb7fb62b4c8dfb Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Wed, 10 Sep 2014 14:18:41 +0100 Subject: [PATCH 30/38] Keep targets around until their attempt fails --- connection_maker.go | 18 ++++++++++-------- 1 file changed, 10 insertions(+), 8 deletions(-) diff --git a/connection_maker.go b/connection_maker.go index 3fb7eb7e..f026918a 100644 --- a/connection_maker.go +++ b/connection_maker.go @@ -141,15 +141,17 @@ func (cm *ConnectionMaker) checkStateAndAttemptConnections(now time.Time) { }) for address, target := range cm.targets { - if validTarget[address] { - if target.state == CSUnconnected && now.After(target.tryAfter) { - target.attemptCount += 1 - target.state = CSAttempting - go cm.attemptConnection(address, cm.cmdLineAddress[address]) + if target.state == CSUnconnected { + if validTarget[address] { + if now.After(target.tryAfter) { + target.attemptCount += 1 + target.state = CSAttempting + go cm.attemptConnection(address, cm.cmdLineAddress[address]) + } + } else { + //log.Println("Deleting target no longer valid:", address) + delete(cm.targets, address) } - } else { - //log.Println("Deleting target no longer valid:", address) - delete(cm.targets, address) } } } From 2050569fb6dcd9516489aec3d0b37cec2c0566bd Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Wed, 10 Sep 2014 14:50:43 +0100 Subject: [PATCH 31/38] Break checkStateAndAttemptConnections into phases, and track the end of all connection attempts --- connection_maker.go | 71 ++++++++++++++++++++------------------------- peer.go | 12 -------- types.go | 6 ++-- 3 files changed, 34 insertions(+), 55 deletions(-) diff --git a/connection_maker.go b/connection_maker.go index f026918a..0c95965f 100644 --- a/connection_maker.go +++ b/connection_maker.go @@ -14,8 +14,7 @@ const ( MaxAttemptCount = 100 CMInitiate = iota CMStatus = iota - CMEstablished = iota - CMConnFailed = iota + CMAttemptDone = iota ) const ( @@ -26,10 +25,10 @@ const ( func StartConnectionMaker(router *Router) *ConnectionMaker { queryChan := make(chan *ConnectionMakerInteraction, ChannelSize) state := &ConnectionMaker{ - router: router, - queryChan: queryChan, + router: router, + queryChan: queryChan, cmdLineAddress: make(map[string]bool), - targets: make(map[string]*Target)} + targets: make(map[string]*Target)} go state.queryLoop(queryChan) return state } @@ -69,18 +68,18 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio } switch { case query.code == CMInitiate: - cm.cmdLineAddress[query.address] = true; + cm.cmdLineAddress[query.address] = true tickNow() case query.code == CMStatus: query.resultChan <- cm.status() - case query.code == CMConnFailed: + case query.code == CMAttemptDone: target := cm.targets[query.address] if target != nil { target.state = CSUnconnected target.tryAfter, target.tryInterval = tryAfter(target.tryInterval) maybeTick() } else { - log.Fatal("CMConnFailed unknown address", query.address) + log.Fatal("CMAttemptDone unknown address", query.address) } default: log.Fatal("Unexpected connection maker query:", query) @@ -95,32 +94,24 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio func (cm *ConnectionMaker) checkStateAndAttemptConnections(now time.Time) { ourself := cm.router.Ourself - validTarget := make(map[string]bool) - // Any targets that are now connected, we don't need to attempt any more - for address, _ := range cm.targets { - if _, found := ourself.ConnectionOn(address); found { - //log.Println("Deleting target now connected:", address) - delete(cm.targets, address) - } - } - - // Add command-line targets that are not connected - for address, _ := range cm.cmdLineAddress { - if _, found := ourself.ConnectionOn(address); !found { - //log.Println("Unconnected cmdline:", address) - cm.addToTargets(address) - validTarget[address] = true - } - } - // copy the set of peers we are connected to, so we can access it without locking our_connected_peers := make(map[PeerName]bool) - ourself.ForEachConnection(func(peer PeerName, _ Connection) { + 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 }) + // Add command-line targets that are not connected + for address, _ := range cm.cmdLineAddress { + if !our_connected_targets[address] { + //log.Println("Unconnected cmdline:", address) + validTarget[address] = true + } + } + // 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) { @@ -130,27 +121,27 @@ func (cm *ConnectionMaker) checkStateAndAttemptConnections(now time.Time) { // try both portnumber of connection and standart port if host, port, err := ExtractHostPort(address); err == nil { if port != Port { - cm.addToTargets(address) validTarget[address] = true } - cm.addToTargets(host) validTarget[host] = true } } }) }) + for address, _ := range validTarget { + cm.addToTargets(address) + } + for address, target := range cm.targets { if target.state == CSUnconnected { - if validTarget[address] { - if now.After(target.tryAfter) { - target.attemptCount += 1 - target.state = CSAttempting - go cm.attemptConnection(address, cm.cmdLineAddress[address]) - } - } else { + 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]) } } } @@ -160,7 +151,7 @@ func (cm *ConnectionMaker) addToTargets(address string) { target := cm.targets[address] if target == nil { target = &Target{ - state: CSUnconnected, + state: CSUnconnected, } target.tryAfter, target.tryInterval = tryImmediately() cm.targets[address] = target @@ -183,10 +174,10 @@ 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.queryChan <- &ConnectionMakerInteraction{ - Interaction: Interaction{code: CMConnFailed}, - address: address} } + cm.queryChan <- &ConnectionMakerInteraction{ + Interaction: Interaction{code: CMAttemptDone}, + address: address} } func tryImmediately() (time.Time, time.Duration) { diff --git a/peer.go b/peer.go index 4d3dd0c6..b62e8b27 100644 --- a/peer.go +++ b/peer.go @@ -57,18 +57,6 @@ func (peer *Peer) ConnectionTo(name PeerName) (Connection, bool) { return conn, found // yes, you really can't inline that. FFS. } -func (peer *Peer) ConnectionOn(peerAddr string) (Connection, bool) { - peerAddr = NormalisePeerAddr(peerAddr) - peer.RLock() - defer peer.RUnlock() - for _, conn := range peer.connections { - if conn.RemoteTCPAddr() == peerAddr { - return conn, true - } - } - return nil, false -} - func (peer *Peer) ForEachConnection(fun func(PeerName, Connection)) { peer.RLock() defer peer.RUnlock() diff --git a/types.go b/types.go index 1626b5aa..afff09d7 100644 --- a/types.go +++ b/types.go @@ -159,9 +159,9 @@ type ConnectionMakerInteraction struct { } type ConnectionMaker struct { - router *Router - queryChan chan<- *ConnectionMakerInteraction - targets map[string]*Target + router *Router + queryChan chan<- *ConnectionMakerInteraction + targets map[string]*Target cmdLineAddress map[string]bool } From e5bd3933d8ac86677a7f6dbea3d1cfcacd3cef51 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Wed, 10 Sep 2014 14:59:32 +0100 Subject: [PATCH 32/38] Normalise peer address before checking --- connection_maker.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/connection_maker.go b/connection_maker.go index 0c95965f..8eaf4dce 100644 --- a/connection_maker.go +++ b/connection_maker.go @@ -100,13 +100,14 @@ func (cm *ConnectionMaker) checkStateAndAttemptConnections(now time.Time) { 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[address] { + if !our_connected_targets[NormalisePeerAddr(address)] { //log.Println("Unconnected cmdline:", address) validTarget[address] = true } From 83175cdd0af27703746617d7f816be349545fa87 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Wed, 10 Sep 2014 15:25:10 +0100 Subject: [PATCH 33/38] Normalise IP addresses when adding to collections --- connection_maker.go | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/connection_maker.go b/connection_maker.go index 8eaf4dce..b65a21a9 100644 --- a/connection_maker.go +++ b/connection_maker.go @@ -68,7 +68,7 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio } switch { case query.code == CMInitiate: - cm.cmdLineAddress[query.address] = true + cm.cmdLineAddress[NormalisePeerAddr(query.address)] = true tickNow() case query.code == CMStatus: query.resultChan <- cm.status() @@ -122,9 +122,11 @@ func (cm *ConnectionMaker) checkStateAndAttemptConnections(now time.Time) { // 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 } - validTarget[host] = true + // Add the address with standard port + validTarget[NormalisePeerAddr(host)] = true } } }) @@ -149,6 +151,7 @@ func (cm *ConnectionMaker) checkStateAndAttemptConnections(now time.Time) { } func (cm *ConnectionMaker) addToTargets(address string) { + address = NormalisePeerAddr(address) target := cm.targets[address] if target == nil { target = &Target{ From c28ff33c845cef5a9240fad477801981afe95270 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Wed, 10 Sep 2014 15:26:13 +0100 Subject: [PATCH 34/38] Avoid waiting 5 seconds before starting --- connection_maker.go | 1 + 1 file changed, 1 insertion(+) diff --git a/connection_maker.go b/connection_maker.go index b65a21a9..58763f99 100644 --- a/connection_maker.go +++ b/connection_maker.go @@ -136,6 +136,7 @@ func (cm *ConnectionMaker) checkStateAndAttemptConnections(now time.Time) { 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] { From 651c0a0c27a2394c496d31e0189284bd873c482e Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Wed, 10 Sep 2014 16:00:31 +0100 Subject: [PATCH 35/38] Remove tickNow() function --- connection_maker.go | 13 ++++--------- 1 file changed, 4 insertions(+), 9 deletions(-) diff --git a/connection_maker.go b/connection_maker.go index 58763f99..09958200 100644 --- a/connection_maker.go +++ b/connection_maker.go @@ -55,11 +55,6 @@ 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: @@ -69,12 +64,12 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio switch { case query.code == CMInitiate: cm.cmdLineAddress[NormalisePeerAddr(query.address)] = true - tickNow() + cm.checkStateAndAttemptConnections(time.Now()) + maybeTick() case query.code == CMStatus: query.resultChan <- cm.status() case query.code == CMAttemptDone: - target := cm.targets[query.address] - if target != nil { + if target, found := cm.targets[query.address]; found { target.state = CSUnconnected target.tryAfter, target.tryInterval = tryAfter(target.tryInterval) maybeTick() @@ -96,7 +91,7 @@ func (cm *ConnectionMaker) checkStateAndAttemptConnections(now time.Time) { ourself := cm.router.Ourself validTarget := make(map[string]bool) - // copy the set of peers we are connected to, so we can access it without locking + // 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) { From ba72999b0411c918b238759861919f20f1103555 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Wed, 10 Sep 2014 16:02:17 +0100 Subject: [PATCH 36/38] cosmetic --- router.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/router.go b/router.go index de84ec2e..ec2d9de1 100644 --- a/router.go +++ b/router.go @@ -63,7 +63,7 @@ func (router *Router) Status() string { buf.WriteString(fmt.Sprintf("MACs:\n%s", router.Macs)) buf.WriteString(fmt.Sprintf("Peers:\n%s", router.Peers)) buf.WriteString(fmt.Sprintf("Topology:\n%s", router.Topology)) - buf.WriteString(fmt.Sprintf("Reconnections:\n%s", router.ConnectionMaker)) + buf.WriteString(fmt.Sprintf("Reconnects:\n%s", router.ConnectionMaker)) return buf.String() } From 0e23c29cae0dcae21387b3447f96e79ae6cbae05 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Wed, 10 Sep 2014 16:02:50 +0100 Subject: [PATCH 37/38] cosmetic --- connection_maker.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/connection_maker.go b/connection_maker.go index 09958200..8501216b 100644 --- a/connection_maker.go +++ b/connection_maker.go @@ -131,7 +131,7 @@ func (cm *ConnectionMaker) checkStateAndAttemptConnections(now time.Time) { cm.addToTargets(address) } - now = time.Now() // make sure we catch items just added + 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] { From c6d8623b6fd5beac71481fe385b43b8855073320 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Wed, 10 Sep 2014 16:17:56 +0100 Subject: [PATCH 38/38] Only signal connection failures; remove successful connections when we see them --- connection.go | 4 ++++ connection_maker.go | 19 ++++++++++++------- 2 files changed, 16 insertions(+), 7 deletions(-) diff --git a/connection.go b/connection.go index f0aed6f9..a88d7d42 100644 --- a/connection.go +++ b/connection.go @@ -277,6 +277,10 @@ func (conn *LocalConnection) handleShutdown() { // blank out the forwardChan so that the router processes don't // try to send any more conn.stopForwarders() + + if conn.remote != nil { + conn.Router.ConnectionMaker.ShutdownConnection(conn.remoteTCPAddr) + } } func (conn *LocalConnection) handshake(acceptNewPeer bool) error { diff --git a/connection_maker.go b/connection_maker.go index 8501216b..babcee3b 100644 --- a/connection_maker.go +++ b/connection_maker.go @@ -14,7 +14,7 @@ const ( MaxAttemptCount = 100 CMInitiate = iota CMStatus = iota - CMAttemptDone = iota + CMFailed = iota ) const ( @@ -39,6 +39,12 @@ func (cm *ConnectionMaker) InitiateConnection(address string) { address: address} } +func (cm *ConnectionMaker) ShutdownConnection(address string) { + cm.queryChan <- &ConnectionMakerInteraction{ + Interaction: Interaction{code: CMFailed}, + address: address} +} + func (cm *ConnectionMaker) String() string { resultChan := make(chan interface{}, 0) cm.queryChan <- &ConnectionMakerInteraction{ @@ -68,13 +74,11 @@ func (cm *ConnectionMaker) queryLoop(queryChan <-chan *ConnectionMakerInteractio maybeTick() case query.code == CMStatus: query.resultChan <- cm.status() - case query.code == CMAttemptDone: + case query.code == CMFailed: if target, found := cm.targets[query.address]; found { target.state = CSUnconnected target.tryAfter, target.tryInterval = tryAfter(target.tryInterval) maybeTick() - } else { - log.Fatal("CMAttemptDone unknown address", query.address) } default: log.Fatal("Unexpected connection maker query:", query) @@ -142,6 +146,9 @@ func (cm *ConnectionMaker) checkStateAndAttemptConnections(now time.Time) { 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) } } } @@ -174,10 +181,8 @@ 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) } - cm.queryChan <- &ConnectionMakerInteraction{ - Interaction: Interaction{code: CMAttemptDone}, - address: address} } func tryImmediately() (time.Time, time.Duration) {