From 657125fd54e6ed6759a529b1852d4e048c82d7cb Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Fri, 27 Feb 2015 14:12:32 +0000 Subject: [PATCH 1/3] Send to tcpSender on the calling thread, so the Actor channel doesn't get clogged up --- router/connection.go | 22 ++++++++++++---------- 1 file changed, 12 insertions(+), 10 deletions(-) diff --git a/router/connection.go b/router/connection.go index 111830b3..bcdfe120 100644 --- a/router/connection.go +++ b/router/connection.go @@ -181,11 +181,19 @@ func (conn *LocalConnection) log(args ...interface{}) { log.Println(append(append([]interface{}{}, fmt.Sprintf("->[%s]:", conn.remote.Name)), args...)...) } +// Send directly - not via the Actor queryLoop. +// note that tcpSender is immutable and guaranteed to be set by the +// time anybody can get hold of a ref to the Connection +func (conn *LocalConnection) SendProtocolMsg(m ProtocolMsg) { + if err := conn.handleSendProtocolMsg(m); err != nil { + conn.Shutdown(err) + } +} + // ACTOR client API const ( - CSendProtocolMsg = iota - CSetEstablished + CSetEstablished = iota CReceivedHeartbeat CShutdown ) @@ -203,7 +211,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 +232,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 +310,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") From 139758256b6601ae792eb4f4300a25f340e227fa Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Fri, 27 Feb 2015 14:52:47 +0000 Subject: [PATCH 2/3] Make clear that tcp-send methods are not handlers and may be called outside of queryLoop --- router/connection.go | 24 ++++++++++++------------ 1 file changed, 12 insertions(+), 12 deletions(-) diff --git a/router/connection.go b/router/connection.go index bcdfe120..096f969b 100644 --- a/router/connection.go +++ b/router/connection.go @@ -185,7 +185,7 @@ func (conn *LocalConnection) log(args ...interface{}) { // note that tcpSender is immutable and guaranteed to be set by the // time anybody can get hold of a ref to the Connection func (conn *LocalConnection) SendProtocolMsg(m ProtocolMsg) { - if err := conn.handleSendProtocolMsg(m); err != nil { + if err := conn.sendProtocolMsg(m); err != nil { conn.Shutdown(err) } } @@ -317,7 +317,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 @@ -339,7 +339,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 } } @@ -376,20 +376,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()) @@ -416,6 +408,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 From 6437caddf2afbe4b1864943618d5da4993eec93b Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Fri, 27 Feb 2015 15:02:30 +0000 Subject: [PATCH 3/3] Reworded comment --- router/connection.go | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/router/connection.go b/router/connection.go index 096f969b..1e4166e4 100644 --- a/router/connection.go +++ b/router/connection.go @@ -181,9 +181,11 @@ func (conn *LocalConnection) log(args ...interface{}) { log.Println(append(append([]interface{}{}, fmt.Sprintf("->[%s]:", conn.remote.Name)), args...)...) } -// Send directly - not via the Actor queryLoop. -// note that tcpSender is immutable and guaranteed to be set by the -// time anybody can get hold of a ref to the Connection +// 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)