reweave/weave
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

git rid of heartbeater goroutine

Browse Source 
...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.
This commit is contained in:
Matthias Radestock committed 2014-12-23 22:01:03 +00:00
1 parent ad67dc5d07
commit e850127ed4
3 files changed
+62 -73
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

docs/architecture.txt
+4 -6
View File
@@ -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,
router/connection.go
+55 -66
View File
@@ -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()
}
}
router/types.go
+3 -1
View File
@@ -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{}