Replace OverlayForwarderListener with channels

It's more robust to signal the conditions asynchronously via channels
rather than callbacks.

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 {

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 {

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