Merge pull request #428 from bboreham/415_direct_tcpSend

Remove deadlock in sending protocol messages

Closes #415.

router/connection.go

@@ -181,11 +181,21 @@ func (conn *LocalConnection) log(args ...interface{}) {
 	log.Println(append(append([]interface{}{}, fmt.Sprintf("->[%s]:", conn.remote.Name)), args...)...)
 }
 
+// Send directly, not via the Actor.  If it goes via the Actor we can
+// get a deadlock where LocalConnection is blocked talking to
+// LocalPeer and LocalPeer is blocked trying send a ProtocolMsg via
+// LocalConnection, and the channels are full in both directions so
+// nothing can proceed.
+func (conn *LocalConnection) SendProtocolMsg(m ProtocolMsg) {
+	if err := conn.sendProtocolMsg(m); err != nil {
+		conn.Shutdown(err)
+	}
+}
+
 // ACTOR client API
 
 const (
-	CSendProtocolMsg = iota
-	CSetEstablished
+	CSetEstablished = iota
 	CReceivedHeartbeat
 	CShutdown
 )
@@ -203,7 +213,8 @@ func (conn *LocalConnection) sendQuery(code int, payload interface{}) {
 
 // Async
 func (conn *LocalConnection) Shutdown(err error) {
-	conn.sendQuery(CShutdown, err)
+	// Run on its own goroutine in case the channel is backed up
+	go conn.sendQuery(CShutdown, err)
 }
 
 // Async
@@ -223,11 +234,6 @@ func (conn *LocalConnection) SetEstablished() {
 	conn.sendQuery(CSetEstablished, nil)
 }
 
-// Async
-func (conn *LocalConnection) SendProtocolMsg(m ProtocolMsg) {
-	conn.sendQuery(CSendProtocolMsg, m)
-}
-
 // ACTOR server
 
 func (conn *LocalConnection) run(queryChan <-chan *ConnectionInteraction, finished chan<- struct{}, acceptNewPeer bool) {
@@ -306,8 +312,6 @@ func (conn *LocalConnection) queryLoop(queryChan <-chan *ConnectionInteraction)
 				err = conn.handleReceivedHeartbeat(query.payload.(*net.UDPAddr))
 			case CSetEstablished:
 				err = conn.handleSetEstablished()
-			case CSendProtocolMsg:
-				err = conn.handleSendProtocolMsg(query.payload.(ProtocolMsg))
 			}
 		case <-conn.heartbeatTimeout.C:
 			err = fmt.Errorf("timed out waiting for UDP heartbeat")
@@ -315,7 +319,7 @@ func (conn *LocalConnection) queryLoop(queryChan <-chan *ConnectionInteraction)
 			conn.Forward(true, conn.heartbeatFrame, nil)
 		case <-tickerChan(conn.fragTest):
 			conn.setStackFrag(false)
-			err = conn.handleSendSimpleProtocolMsg(ProtocolStartFragmentationTest)
+			err = conn.sendSimpleProtocolMsg(ProtocolStartFragmentationTest)
 		}
 	}
 	return
@@ -337,7 +341,7 @@ func (conn *LocalConnection) handleReceivedHeartbeat(remoteUDPAddr *net.UDPAddr)
 	conn.Unlock()
 	conn.heartbeatTimeout.Reset(HeartbeatTimeout)
 	if !old {
-		if err := conn.handleSendSimpleProtocolMsg(ProtocolConnectionEstablished); err != nil {
+		if err := conn.sendSimpleProtocolMsg(ProtocolConnectionEstablished); err != nil {
 			return err
 		}
 	}
@@ -374,20 +378,12 @@ func (conn *LocalConnection) handleSetEstablished() error {
 	// avoid initial waits for timers to fire
 	conn.Forward(true, conn.heartbeatFrame, nil)
 	conn.setStackFrag(false)
-	if err := conn.handleSendSimpleProtocolMsg(ProtocolStartFragmentationTest); err != nil {
+	if err := conn.sendSimpleProtocolMsg(ProtocolStartFragmentationTest); err != nil {
 		return err
 	}
 	return nil
 }
 
-func (conn *LocalConnection) handleSendSimpleProtocolMsg(tag ProtocolTag) error {
-	return conn.handleSendProtocolMsg(ProtocolMsg{tag: tag})
-}
-
-func (conn *LocalConnection) handleSendProtocolMsg(m ProtocolMsg) error {
-	return conn.tcpSender.Send(Concat([]byte{byte(m.tag)}, m.msg))
-}
-
 func (conn *LocalConnection) handleShutdown() {
 	if conn.TCPConn != nil {
 		checkWarn(conn.TCPConn.Close())
@@ -414,6 +410,14 @@ func (conn *LocalConnection) handleShutdown() {
 
 // Helpers
 
+func (conn *LocalConnection) sendSimpleProtocolMsg(tag ProtocolTag) error {
+	return conn.sendProtocolMsg(ProtocolMsg{tag: tag})
+}
+
+func (conn *LocalConnection) sendProtocolMsg(m ProtocolMsg) error {
+	return conn.tcpSender.Send(Concat([]byte{byte(m.tag)}, m.msg))
+}
+
 func (conn *LocalConnection) receiveTCP(decoder *gob.Decoder) {
 	defer conn.Decryptor.Shutdown()
 	usingPassword := conn.SessionKey != nil