From bc5250c8998d4ad1cfc4845dca60ac7e4c447614 Mon Sep 17 00:00:00 2001 From: David Wragg Date: Thu, 27 Aug 2015 17:26:53 +0100 Subject: [PATCH] Replace OverlayForwarderListener with channels It's more robust to signal the conditions asynchronously via channels rather than callbacks. --- router/connection.go | 53 +++++++++++++++++-------------------------- router/overlay.go | 40 ++++++++++++++++++++------------ router/sleeve.go | 54 ++++++++++++++++++++------------------------ 3 files changed, 72 insertions(+), 75 deletions(-) diff --git a/router/connection.go b/router/connection.go index 99577e57..1090aa89 100644 --- a/router/connection.go +++ b/router/connection.go @@ -115,8 +115,6 @@ func (conn *LocalConnection) BreakTie(dupConn Connection) ConnectionTieBreak { } func (conn *LocalConnection) Established() bool { - conn.RLock() - defer conn.RUnlock() return conn.established } @@ -214,17 +212,16 @@ func (conn *LocalConnection) run(actionChan <-chan ConnectionAction, finished ch } conn.Router.ConnectionMaker.ConnectionCreated(conn) - // SetListener has the side-effect of telling the forwarder - // that the connection is confirmed. This comes after - // AddConnection, because only after that completes do we know - // the connection is valid: in particular that it is not a - // duplicate connection to the same peer. Sending heartbeats - // on a duplicate connection can trip up crypto at the other - // end, since the associated UDP packets may get decoded by - // the other connection. It is also generally wasteful to - // engage in any interaction with the remote on a connection - // that turns out to be invalid. - conn.forwarder.SetListener(ConnectionAsForwarderListener{conn}) + // Forwarder confirmation comes after AddConnection, because + // only after that completes do we know the connection is + // valid: in particular that it is not a duplicate connection + // to the same peer. Sending heartbeats on a duplicate + // connection can trip up crypto at the other end, since the + // associated UDP packets may get decoded by the other + // connection. It is also generally wasteful to engage in any + // interaction with the remote on a connection that turns out + // to be invalid. + conn.forwarder.Confirm() // receiveTCP must follow also AddConnection. In the absence // of any indirect connectivity to the remote peer, the first @@ -335,12 +332,23 @@ func (conn *LocalConnection) registerRemote(remote *Peer, acceptNewPeer bool) er } func (conn *LocalConnection) actorLoop(actionChan <-chan ConnectionAction) (err error) { + fwdErrorChan := conn.forwarder.ErrorChannel() + fwdEstablishedChan := conn.forwarder.EstablishedChannel() + for err == nil { select { case action := <-actionChan: err = action() + case <-conn.heartbeatTCP.C: err = conn.sendSimpleProtocolMsg(ProtocolHeartbeat) + + case <-fwdEstablishedChan: + conn.established = true + fwdEstablishedChan = nil + conn.Router.Ourself.ConnectionEstablished(conn) + + case err = <-fwdErrorChan: } } return @@ -388,25 +396,6 @@ func (conn *LocalConnection) sendOverlayControlMessage(tag ProtocolTag, msg []by return conn.sendProtocolMsg(ProtocolMsg{tag, msg}) } -type ConnectionAsForwarderListener struct{ conn *LocalConnection } - -func (l ConnectionAsForwarderListener) Established() { - l.conn.sendAction(func() error { - old := l.conn.established - l.conn.Lock() - l.conn.established = true - l.conn.Unlock() - if !old { - l.conn.Router.Ourself.ConnectionEstablished(l.conn) - } - return nil - }) -} - -func (l ConnectionAsForwarderListener) Error(err error) { - l.conn.sendAction(func() error { return err }) -} - // Helpers func (conn *LocalConnection) sendSimpleProtocolMsg(tag ProtocolTag) error { diff --git a/router/overlay.go b/router/overlay.go index b552e179..9da023be 100644 --- a/router/overlay.go +++ b/router/overlay.go @@ -57,16 +57,25 @@ type OverlayCrypto struct { // All of the machinery to forward packets to a particular peer type OverlayForwarder interface { - // Register a callback for forwarder state changes. - // side-effect, calling this confirms that the connection is - // really wanted, and so the provider should activate it. - // However, Forward might be called before this is called - // (e.g. on another thread). - SetListener(OverlayForwarderListener) - - // Forward a packet across the connection. + // Forward a packet across the connection. May be called as + // soon as the forwarder is created, in particular before + // Confirm(). Forward(ForwardPacketKey) FlowOp + // Confirm that the connection is really wanted, and so the + // Overlay should begin heartbeats etc. to verify the + // operation of the forwarder. + Confirm() + + // A channel indicating that the forwarder is established, + // i.e. its operation has been confirmed. + EstablishedChannel() <-chan struct{} + + // A channel indicating an error from the forwarder. The + // forwarder is not expected to be operational after the first + // error, so the channel only needs to buffer a single error. + ErrorChannel() <-chan error + Stop() // Handle a message from the peer. 'tag' exists for @@ -75,11 +84,6 @@ type OverlayForwarder interface { ControlMessage(tag ProtocolTag, msg []byte) } -type OverlayForwarderListener interface { - Established() - Error(error) -} - type NullOverlay struct{} func (NullOverlay) StartConsumingPackets(*Peer, *Peers, OverlayConsumer) error { @@ -96,7 +100,15 @@ func (NullOverlay) InvalidateRoutes() { func (NullOverlay) InvalidateShortIDs() { } -func (NullOverlay) SetListener(OverlayForwarderListener) { +func (NullOverlay) Confirm() { +} + +func (NullOverlay) EstablishedChannel() <-chan struct{} { + return nil +} + +func (NullOverlay) ErrorChannel() <-chan error { + return nil } func (NullOverlay) Forward(ForwardPacketKey) FlowOp { diff --git a/router/sleeve.go b/router/sleeve.go index 957ab9dd..ed08d44e 100644 --- a/router/sleeve.go +++ b/router/sleeve.go @@ -292,9 +292,12 @@ type sleeveForwarder struct { confirmedChan chan<- struct{} finishedChan <-chan struct{} + // listener channels + establishedChan chan struct{} + errorChan chan error + // Explicitly locked state lock sync.RWMutex - listener OverlayForwarderListener remoteAddr *net.UDPAddr // These fields are accessed and updated independently, so no @@ -375,6 +378,8 @@ func (sleeve *SleeveOverlay) MakeForwarder(params ForwarderParams) (OverlayForwa controlMsgChan: controlMsgChan, confirmedChan: confirmedChan, finishedChan: finishedChan, + establishedChan: make(chan struct{}), + errorChan: make(chan error, 1), remoteAddr: params.RemoteAddr, mtu: DefaultMTU, crypto: crypto, @@ -400,23 +405,23 @@ func (fwd *sleeveForwarder) logPrefix() string { return fwd.logPrefixFor(remoteAddr) } -func (fwd *sleeveForwarder) SetListener(listener OverlayForwarderListener) { - log.Debug(fwd.logPrefix(), "SetListener ", listener) +func (fwd *sleeveForwarder) Confirm() { + log.Debug(fwd.logPrefix(), "Confirm") - fwd.lock.Lock() - fwd.listener = listener - fwd.lock.Unlock() - - // Setting the listener confirms that the forwarder is really - // wanted - if listener != nil { - select { - case fwd.confirmedChan <- struct{}{}: - case <-fwd.finishedChan: - } + select { + case fwd.confirmedChan <- struct{}{}: + case <-fwd.finishedChan: } } +func (fwd *sleeveForwarder) EstablishedChannel() <-chan struct{} { + return fwd.establishedChan +} + +func (fwd *sleeveForwarder) ErrorChannel() <-chan error { + return fwd.errorChan +} + type curriedForward struct { fwd *sleeveForwarder key ForwardPacketKey @@ -590,7 +595,6 @@ func (fwd *sleeveForwarder) ControlMessage(tag ProtocolTag, msg []byte) { func (fwd *sleeveForwarder) Stop() { fwd.sleeve.removeForwarder(fwd.remotePeer.Name, fwd) - fwd.SetListener(nil) // Tell the forwarder goroutine to finish. We don't need to // wait for it. @@ -664,9 +668,9 @@ loop: fwd.lock.RLock() defer fwd.lock.RUnlock() - if fwd.listener != nil { - fwd.listener.Error(err) - } + + // this is the only place we send an error to errorChan + fwd.errorChan <- err } func (fwd *sleeveForwarder) aggregateAndSend(frame aggregatorFrame, @@ -857,14 +861,14 @@ func (fwd *sleeveForwarder) setRemoteAddr(addr *net.UDPAddr) { func (fwd *sleeveForwarder) handleHeartbeatAck() error { log.Debug(fwd.logPrefix(), "handleHeartbeatAck") - // The connection is now regarded as established - fwd.notifyEstablished() - if fwd.heartbeatInterval != SlowHeartbeat { fwd.heartbeatInterval = SlowHeartbeat if fwd.heartbeatTimer != nil { fwd.heartbeatTimer.Reset(fwd.heartbeatInterval) } + + // The connection is now regarded as established + close(fwd.establishedChan) } fwd.fragTestTicker = time.NewTicker(FragTestInterval) @@ -879,14 +883,6 @@ func (fwd *sleeveForwarder) handleHeartbeatAck() error { make([]byte, PMTUDiscoverySize)) } -func (fwd *sleeveForwarder) notifyEstablished() { - fwd.lock.RLock() - defer fwd.lock.RUnlock() - if fwd.listener != nil { - fwd.listener.Established() - } -} - func (fwd *sleeveForwarder) sendFragTest() error { log.Debug(fwd.logPrefix(), "sendFragTest") fwd.stackFrag = false