From 3b4396077928b83c55b408229b76bb2c5c4ce559 Mon Sep 17 00:00:00 2001
From: Matthias Radestock <matthias.radestock@gmail.com>
Date: Wed, 25 Nov 2015 10:51:18 +0000
Subject: [PATCH] prevent unbounded number of goroutines for connection
 shutdown

We need Connection.Shutdown to be non-blocked, and the way we achieved
this is by spawning a goroutine. However, there can be a potentially
unbounded number of Connection.Shutdown invocations. For example, if
the TCP connection has gone bad then any Connection.SendProtocolMsg
invocation will end up invoking Connection.Shutdown. There can be many
such invocations until the Connection has been removed from the
various data structures (e.g. LocalPeer.connections and
GossipSender.senders). Hence we can end up with with an unbounded
number of spawned goroutines.

To fix this, we introduce a one-element error channel on the
connection. Connection.Shutdown does a non-blocking send on that, and
the Connection.actorLoop reads from it and terminates.

This will throw away all but the first error encountered, but that is
fine and not much different from what was happening before (where we'd
throw away all but one of the errors encountered).
---
 mesh/connection.go | 19 +++++++++++++------
 1 file changed, 13 insertions(+), 6 deletions(-)

diff --git a/mesh/connection.go b/mesh/connection.go
index b7b5d270..fdcdcf3d 100644
--- a/mesh/connection.go
+++ b/mesh/connection.go
@@ -48,6 +48,7 @@ type LocalConnection struct {
 	Router       *Router
 	uid          uint64
 	actionChan   chan<- ConnectionAction
+	errorChan    chan<- error
 	finished     <-chan struct{} // closed to signal that actorLoop has finished
 	OverlayConn  OverlayConnection
 }
@@ -87,6 +88,7 @@ func StartLocalConnection(connRemote *RemoteConnection, tcpConn *net.TCPConn, ro
 		log.Fatal("Attempt to create local connection from a peer which is not ourself")
 	}
 	actionChan := make(chan ConnectionAction, ChannelSize)
+	errorChan := make(chan error, 1)
 	finished := make(chan struct{})
 	conn := &LocalConnection{
 		RemoteConnection: *connRemote, // NB, we're taking a copy of connRemote here.
@@ -94,8 +96,9 @@ func StartLocalConnection(connRemote *RemoteConnection, tcpConn *net.TCPConn, ro
 		TCPConn:          tcpConn,
 		uid:              randUint64(),
 		actionChan:       actionChan,
+		errorChan:        errorChan,
 		finished:         finished}
-	go conn.run(actionChan, finished, acceptNewPeer)
+	go conn.run(actionChan, errorChan, finished, acceptNewPeer)
 }
 
 func (conn *LocalConnection) BreakTie(dupConn Connection) ConnectionTieBreak {
@@ -135,8 +138,10 @@ func (conn *LocalConnection) Shutdown(err error) {
 		panic("nil error")
 	}
 
-	// Run on its own goroutine in case the channel is backed up
-	go func() { conn.sendAction(func() error { return err }) }()
+	select {
+	case conn.errorChan <- err:
+	default:
+	}
 }
 
 // Send an actor request to the actorLoop, but don't block if
@@ -151,7 +156,7 @@ func (conn *LocalConnection) sendAction(action ConnectionAction) {
 
 // ACTOR server
 
-func (conn *LocalConnection) run(actionChan <-chan ConnectionAction, finished chan<- struct{}, acceptNewPeer bool) {
+func (conn *LocalConnection) run(actionChan <-chan ConnectionAction, errorChan <-chan error, finished chan<- struct{}, acceptNewPeer bool) {
 	var err error // important to use this var and not create another one with 'err :='
 	defer func() { conn.shutdown(err) }()
 	defer close(finished)
@@ -237,7 +242,7 @@ func (conn *LocalConnection) run(actionChan <-chan ConnectionAction, finished ch
 	// running AddConnection in a separate goroutine, at least not
 	// without some synchronisation. Which in turn requires the
 	// launching of the receiveTCP goroutine to precede actorLoop.
-	err = conn.actorLoop(actionChan)
+	err = conn.actorLoop(actionChan, errorChan)
 }
 
 func (conn *LocalConnection) makeFeatures() map[string]string {
@@ -329,7 +334,7 @@ func (conn *LocalConnection) registerRemote(remote *Peer, acceptNewPeer bool) er
 	return nil
 }
 
-func (conn *LocalConnection) actorLoop(actionChan <-chan ConnectionAction) (err error) {
+func (conn *LocalConnection) actorLoop(actionChan <-chan ConnectionAction, errorChan <-chan error) (err error) {
 	fwdErrorChan := conn.OverlayConn.ErrorChannel()
 	fwdEstablishedChan := conn.OverlayConn.EstablishedChannel()
 
@@ -338,6 +343,8 @@ func (conn *LocalConnection) actorLoop(actionChan <-chan ConnectionAction) (err
 		case action := <-actionChan:
 			err = action()
 
+		case err = <-errorChan:
+
 		case <-conn.heartbeatTCP.C:
 			err = conn.sendSimpleProtocolMsg(ProtocolHeartbeat)