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).
This commit is contained in:
1 parent
c420c21219
commit
3b43960779
1 file changed
+13
-6
+13
-6
@@ -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)
|
||||
|
||||
|
||||
Reference in new issue
Block a user