From a3312c822c40b197a5b8f5de3c225de4a92935b8 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Mon, 8 Sep 2014 16:05:54 +0100 Subject: [PATCH 1/5] 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 2/5] 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 3/5] 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 4/5] 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 5/5] 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{