From e850127ed436762cbdfbfffd9c1b09edd41b7c1f Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Tue, 23 Dec 2014 22:01:03 +0000 Subject: [PATCH] git rid of heartbeater goroutine ...by moving the logic into the connection actor goroutine. The heartbeater goroutine was doing a lot more than just heartbeating; namely it also had tickers for triggering a ProtocolFetchAll and initiating a fragmentation test. Furhermore, ensureHeartbeats also spawned the forwarders. So this was quite a mix of functionality, and, furthermore, most of it related quite closely to the connection actor. So moving the functionality into that simplifies the code. In the process we also fix #294. --- docs/architecture.txt | 10 ++-- router/connection.go | 121 +++++++++++++++++++----------------------- router/types.go | 4 +- 3 files changed, 62 insertions(+), 73 deletions(-) diff --git a/docs/architecture.txt b/docs/architecture.txt index 5bc6445b..f2cfb3b3 100644 --- a/docs/architecture.txt +++ b/docs/architecture.txt @@ -66,10 +66,9 @@ Router: simply responds to traffic received from the remote peer via TCP 5. We register this connection with the local peer 6a. If we initiated this connection then we now start sending fast - heartbeats (a new thread is spawned to do the heartbeating) to the - remote peer so that the remote peer can determine what + heartbeats to the remote peer so that the remote peer can determine what address/port it should use to send UDP back to us. To do this, we - also spawn off two further threads which are the forwarder + spawn off two further threads which are the forwarder loops. These receive frames which are to be sent to the remote peer. One of them sends frames without the DF flag set. To do this it just sends the packets out of the UDP Listener socket. The @@ -84,9 +83,8 @@ Router: local peer. It will tell the local connection (communicating to the actor thread) about the UDP address of the remote peer. The local connection will then start its forwarder threads as - described in 6a, and start sending slow heartbeats (heartbeating - is done by another process). The local connection is marked as - established which means it is included in network updates + described in 6a, and start sending slow heartbeats. The local connection + is marked as established which means it is included in network updates broadcast to our peers. We also send to the remote peer via TCP a ConnectionEstablished message. The remote peer receives this (on the TCP receiver process), tells the connection actor process, diff --git a/router/connection.go b/router/connection.go index fe9073d3..a662f7d6 100644 --- a/router/connection.go +++ b/router/connection.go @@ -105,10 +105,9 @@ func (conn *LocalConnection) setEffectivePMTU(pmtu int) { } } -// Called by the connection's actor process, by the connection's -// heartbeat process and by the connection's TCP received -// process. StackFrag is read in conn.Forward (called by router udp -// listener and sniffer processes) +// Called by the connection's actor process, and by the connection's +// TCP received process. StackFrag is read in conn.Forward (called by +// router udp listener and sniffer processes) func (conn *LocalConnection) setStackFrag(frag bool) { conn.Lock() defer conn.Unlock() @@ -168,27 +167,50 @@ func (conn *LocalConnection) queryLoop(queryChan <-chan *ConnectionInteraction, } conn.local.AddConnection(conn) if conn.remoteUDPAddr != nil { - err = conn.ensureHeartbeat(true) + if err = conn.ensureForwarders(); err == nil { + conn.heartbeat = time.NewTicker(FastHeartbeat) + conn.forwardHeartbeatFrame() // avoid initial wait + } } + var heartbeat, fetchAll, fragTest <-chan time.Time terminate := false for !terminate { if err != nil { conn.log("error:", err) break } - query, ok := <-queryChan - if !ok { - break + if conn.heartbeat != nil { + heartbeat = conn.heartbeat.C } - switch query.code { - case CShutdown: - terminate = true - case CSetEstablished: - err = conn.handleSetEstablished() - case CSetRemoteUDPAddr: - err = conn.handleSetRemoteUDPAddr(query.payload.(*net.UDPAddr)) - case CSendTCP: - err = conn.handleSendTCP(query.payload.([]byte)) + if conn.fetchAll != nil { + fetchAll = conn.fetchAll.C + } + if conn.fragTest != nil { + fragTest = conn.fragTest.C + } + select { + case query, ok := <-queryChan: + if !ok { + break + } + switch query.code { + case CShutdown: + terminate = true + case CSetEstablished: + err = conn.handleSetEstablished() + case CSetRemoteUDPAddr: + err = conn.handleSetRemoteUDPAddr(query.payload.(*net.UDPAddr)) + case CSendTCP: + err = conn.handleSendTCP(query.payload.([]byte)) + } + case <-heartbeat: + conn.forwardHeartbeatFrame() + case <-fetchAll: + err = conn.handleSendTCP(ProtocolFetchAllByte) + case <-fragTest: + conn.setStackFrag(false) + err = conn.handleSendTCP(ProtocolStartFragmentationTestByte) + } } conn.handleShutdown() @@ -220,9 +242,14 @@ func (conn *LocalConnection) handleSetEstablished() error { conn.Unlock() if !old { conn.local.ConnectionEstablished(conn) - if err := conn.ensureHeartbeat(false); err != nil { + if err := conn.ensureForwarders(); err != nil { return err } + stopTicker(conn.heartbeat) + conn.heartbeat = time.NewTicker(SlowHeartbeat) + conn.fetchAll = time.NewTicker(FetchAllInterval) + conn.fragTest = time.NewTicker(FragTestInterval) + conn.forwardHeartbeatFrame() // avoid initial wait // Send a large frame down the DF channel in order to prompt // PMTU discovery to start. conn.Forward(true, &ForwardedFrame{ @@ -266,11 +293,9 @@ func (conn *LocalConnection) handleShutdown() { conn.local.DeleteConnection(conn) } - if conn.heartbeatStop != nil { - // heartbeatStop is 0 length, so this send will synchronise - // with the receive. - conn.heartbeatStop <- nil - } + stopTicker(conn.heartbeat) + stopTicker(conn.fetchAll) + stopTicker(conn.fragTest) // blank out the forwardChan so that the router processes don't // try to send any more @@ -502,56 +527,20 @@ func (conn *LocalConnection) receiveTCP(decoder *gob.Decoder, usingPassword bool } } -// Heartbeats - // Heartbeating serves two purposes: a) keeping NAT paths alive, and // b) updating a remote peer's knowledge of our address, in the event // it changes (e.g. because NAT paths expired). // Called only by connection actor process. -func (conn *LocalConnection) ensureHeartbeat(fast bool) error { - if err := conn.ensureForwarders(); err != nil { - return err - } - var heartbeat, fetchAll, fragTest <-chan time.Time - // explicitly 0 length chan - make send block until receive occurs - stop := make(chan interface{}, 0) - if fast { - // fast, nofetchall, no fragtest - // Lang Spec: "A nil channel is never ready for communication." - heartbeat = time.Tick(FastHeartbeat) - } else { - heartbeat = time.Tick(SlowHeartbeat) - fetchAll = time.Tick(FetchAllInterval) - fragTest = time.Tick(FragTestInterval) - } - // Don't need locks here as this is only read here and in - // handleShutdown, both of which are called by the connection - // actor process only. - if conn.heartbeatStop != nil { - conn.heartbeatStop <- nil - } - conn.heartbeatStop = stop - go conn.forwardHeartbeats(heartbeat, fetchAll, fragTest, stop) - return nil -} - -func (conn *LocalConnection) forwardHeartbeats(heartbeat, fetchAll, fragTest <-chan time.Time, stop <-chan interface{}) { +func (conn *LocalConnection) forwardHeartbeatFrame() { heartbeatFrame := &ForwardedFrame{ srcPeer: conn.local, dstPeer: conn.remote, frame: []byte{}} - conn.Forward(true, heartbeatFrame, nil) // avoid initial wait - for { - select { - case <-stop: - return - case <-heartbeat: - conn.Forward(true, heartbeatFrame, nil) - case <-fetchAll: - conn.SendTCP(ProtocolFetchAllByte) - case <-fragTest: - conn.setStackFrag(false) - conn.SendTCP(ProtocolStartFragmentationTestByte) - } + conn.Forward(true, heartbeatFrame, nil) +} + +func stopTicker(ticker *time.Ticker) { + if ticker != nil { + ticker.Stop() } } diff --git a/router/types.go b/router/types.go index 332669f2..d5cbe29a 100644 --- a/router/types.go +++ b/router/types.go @@ -67,7 +67,9 @@ type LocalConnection struct { stackFrag bool effectivePMTU int SessionKey *[32]byte - heartbeatStop chan<- interface{} + heartbeat *time.Ticker + fetchAll *time.Ticker + fragTest *time.Ticker forwardChan chan<- *ForwardedFrame forwardChanDF chan<- *ForwardedFrame stopForward chan<- interface{}