From 48c73598c46123e6f154275222dd279fbffe5daa Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Tue, 24 Mar 2015 18:53:51 +0000 Subject: [PATCH] convert the connection actor to use closures The only slight trickiness is Shutdown. Generally connections shut down when an error is encountered, but we also need to be able to shut down connections w/o an error. Rather than making all the actor methods return a (terminated bool, error) tuple, we signal the non-error shutdown by sending nil as the action (instead of a closure). --- router/connection.go | 193 +++++++++++++++++++------------------------ 1 file changed, 86 insertions(+), 107 deletions(-) diff --git a/router/connection.go b/router/connection.go index 4d72a286..f3a88c2a 100644 --- a/router/connection.go +++ b/router/connection.go @@ -59,14 +59,11 @@ type LocalConnection struct { Decryptor Decryptor Router *Router uid uint64 - queryChan chan<- *ConnectionInteraction - finished <-chan struct{} // closed to signal that queryLoop has finished + actionChan chan<- ConnectionAction + finished <-chan struct{} // closed to signal that actorLoop has finished } -type ConnectionInteraction struct { - Interaction - payload interface{} -} +type ConnectionAction func() error func NewRemoteConnection(from, to *Peer, tcpAddr string, outbound bool, established bool) *RemoteConnection { return &RemoteConnection{ @@ -118,11 +115,11 @@ func NewLocalConnection(connRemote *RemoteConnection, tcpConn *net.TCPConn, udpA // Async. Does not return anything. If the connection is successful, // it will end up in the local peer's connections map. func (conn *LocalConnection) Start(acceptNewPeer bool) { - queryChan := make(chan *ConnectionInteraction, ChannelSize) - conn.queryChan = queryChan + actionChan := make(chan ConnectionAction, ChannelSize) + conn.actionChan = actionChan finished := make(chan struct{}) conn.finished = finished - go conn.run(queryChan, finished, acceptNewPeer) + go conn.run(actionChan, finished, acceptNewPeer) } func (conn *LocalConnection) BreakTie(dupConn Connection) ConnectionTieBreak { @@ -182,29 +179,21 @@ func (conn *LocalConnection) SendProtocolMsg(m ProtocolMsg) { } } -// ACTOR client API +// ACTOR methods -const ( - CSetEstablished = iota - CReceivedHeartbeat - CShutdown -) - -// Send an actor request to the queryLoop, but don't block if queryLoop has exited -// - see http://blog.golang.org/pipelines for pattern -func (conn *LocalConnection) sendQuery(code int, payload interface{}) { - select { - case conn.queryChan <- &ConnectionInteraction{ - Interaction: Interaction{code: code}, - payload: payload}: - case <-conn.finished: - } -} +// NB: The conn.* fields are only written by the connection actor +// process, which is the caller of the ConnectionAction funs. Hence we +// do not need locks for reading, and only need write locks for fields +// read by other processes. // Async func (conn *LocalConnection) Shutdown(err error) { - // Run on its own goroutine in case the channel is backed up - go conn.sendQuery(CShutdown, err) + if err == nil { + conn.actionChan <- nil + } else { + // Run on its own goroutine in case the channel is backed up + go func() { conn.sendAction(func() error { return err }) }() + } } // Async @@ -216,18 +205,76 @@ func (conn *LocalConnection) ReceivedHeartbeat(remoteUDPAddr *net.UDPAddr, connU if remoteUDPAddr == nil || connUID != conn.uid { return } - conn.sendQuery(CReceivedHeartbeat, remoteUDPAddr) + conn.sendAction(func() error { + oldRemoteUDPAddr := conn.remoteUDPAddr + old := conn.receivedHeartbeat + conn.Lock() + conn.remoteUDPAddr = remoteUDPAddr + conn.receivedHeartbeat = true + conn.Unlock() + conn.heartbeatTimeout.Reset(HeartbeatTimeout) + if !old { + if err := conn.sendSimpleProtocolMsg(ProtocolConnectionEstablished); err != nil { + return err + } + } + if oldRemoteUDPAddr == nil { + return conn.sendFastHeartbeats() + } else if oldRemoteUDPAddr.String() != remoteUDPAddr.String() { + log.Println("Peer", conn.remote.Name, "moved from", old, "to", remoteUDPAddr) + } + return nil + }) } // Async func (conn *LocalConnection) SetEstablished() { - conn.sendQuery(CSetEstablished, nil) + conn.sendAction(func() error { + stopTicker(conn.heartbeat) + old := conn.established + conn.Lock() + conn.established = true + conn.Unlock() + if old { + return nil + } + conn.Router.Ourself.ConnectionEstablished(conn) + if err := conn.ensureForwarders(); err != nil { + return err + } + // Send a large frame down the DF channel in order to prompt + // PMTU discovery to start. + conn.Forward(true, &ForwardedFrame{ + srcPeer: conn.local, + dstPeer: conn.remote, + frame: PMTUDiscovery}, + nil) + conn.heartbeat = time.NewTicker(SlowHeartbeat) + conn.fragTest = time.NewTicker(FragTestInterval) + // avoid initial waits for timers to fire + conn.Forward(true, conn.heartbeatFrame, nil) + conn.setStackFrag(false) + if err := conn.sendSimpleProtocolMsg(ProtocolStartFragmentationTest); err != nil { + return err + } + return nil + }) +} + +// Send an actor request to the actorLoop, but don't block if +// actorLoop has exited - see http://blog.golang.org/pipelines for +// pattern +func (conn *LocalConnection) sendAction(action ConnectionAction) { + select { + case conn.actionChan <- action: + case <-conn.finished: + } } // ACTOR server -func (conn *LocalConnection) run(queryChan <-chan *ConnectionInteraction, finished chan<- struct{}, acceptNewPeer bool) { - defer conn.handleShutdown() +func (conn *LocalConnection) run(actionChan <-chan ConnectionAction, finished chan<- struct{}, acceptNewPeer bool) { + defer conn.shutdown() defer close(finished) tcpConn := conn.TCPConn @@ -268,7 +315,7 @@ func (conn *LocalConnection) run(queryChan <-chan *ConnectionInteraction, finish conn.heartbeatTimeout = time.NewTimer(HeartbeatTimeout) - if err := conn.queryLoop(queryChan); err != nil { + if err := conn.actorLoop(actionChan); err != nil { conn.Log("connection shutting down due to error:", err) } else { conn.Log("connection shutting down") @@ -286,23 +333,14 @@ func (conn *LocalConnection) run(queryChan <-chan *ConnectionInteraction, finish // AddConnection, which is *synchronous*, before entering the tcp // receive loop. -func (conn *LocalConnection) queryLoop(queryChan <-chan *ConnectionInteraction) (err error) { - terminate := false - for !terminate && err == nil { +func (conn *LocalConnection) actorLoop(actionChan <-chan ConnectionAction) (err error) { + for err == nil { select { - case query, ok := <-queryChan: - if !ok { + case action := <-actionChan: + if action == nil { break } - switch query.code { - case CShutdown: - err = query.payload.(error) - terminate = true - case CReceivedHeartbeat: - err = conn.handleReceivedHeartbeat(query.payload.(*net.UDPAddr)) - case CSetEstablished: - err = conn.handleSetEstablished() - } + err = action() case <-conn.heartbeatTimeout.C: err = fmt.Errorf("timed out waiting for UDP heartbeat") case <-tickerChan(conn.heartbeat): @@ -315,66 +353,7 @@ func (conn *LocalConnection) queryLoop(queryChan <-chan *ConnectionInteraction) return } -// Handlers -// -// NB: The conn.* fields are only written by the connection actor -// process, which is the caller of the handlers. Hence we do not need -// locks for reading, and only need write locks for fields read by -// other processes. - -func (conn *LocalConnection) handleReceivedHeartbeat(remoteUDPAddr *net.UDPAddr) error { - oldRemoteUDPAddr := conn.remoteUDPAddr - old := conn.receivedHeartbeat - conn.Lock() - conn.remoteUDPAddr = remoteUDPAddr - conn.receivedHeartbeat = true - conn.Unlock() - conn.heartbeatTimeout.Reset(HeartbeatTimeout) - if !old { - if err := conn.sendSimpleProtocolMsg(ProtocolConnectionEstablished); err != nil { - return err - } - } - if oldRemoteUDPAddr == nil { - return conn.sendFastHeartbeats() - } else if oldRemoteUDPAddr.String() != remoteUDPAddr.String() { - log.Println("Peer", conn.remote.Name, "moved from", old, "to", remoteUDPAddr) - } - return nil -} - -func (conn *LocalConnection) handleSetEstablished() error { - stopTicker(conn.heartbeat) - old := conn.established - conn.Lock() - conn.established = true - conn.Unlock() - if old { - return nil - } - conn.Router.Ourself.ConnectionEstablished(conn) - if err := conn.ensureForwarders(); err != nil { - return err - } - // Send a large frame down the DF channel in order to prompt - // PMTU discovery to start. - conn.Forward(true, &ForwardedFrame{ - srcPeer: conn.local, - dstPeer: conn.remote, - frame: PMTUDiscovery}, - nil) - conn.heartbeat = time.NewTicker(SlowHeartbeat) - conn.fragTest = time.NewTicker(FragTestInterval) - // avoid initial waits for timers to fire - conn.Forward(true, conn.heartbeatFrame, nil) - conn.setStackFrag(false) - if err := conn.sendSimpleProtocolMsg(ProtocolStartFragmentationTest); err != nil { - return err - } - return nil -} - -func (conn *LocalConnection) handleShutdown() { +func (conn *LocalConnection) shutdown() { if conn.TCPConn != nil { checkWarn(conn.TCPConn.Close()) }