From 039a2030db09d288c6a96dfa66d6b786c8e615bb Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Fri, 26 Jun 2015 04:15:05 +0100 Subject: [PATCH] refactor: make DF flag passing logic more obvious ...and provide a better abstraction for sending special frames. --- router/connection.go | 31 +++++++++---------------------- router/forwarder.go | 26 +++++++++++++++++--------- router/local_peer.go | 4 ++-- 3 files changed, 28 insertions(+), 33 deletions(-) diff --git a/router/connection.go b/router/connection.go index 3d15c269..eebbce1b 100644 --- a/router/connection.go +++ b/router/connection.go @@ -50,7 +50,7 @@ type LocalConnection struct { SessionKey *[32]byte heartbeatTCP *time.Ticker heartbeatTimeout *time.Timer - heartbeatFrame *ForwardedFrame + heartbeatFrame []byte heartbeat *time.Ticker fragTest *time.Ticker forwarder *Forwarder @@ -156,8 +156,7 @@ func (conn *LocalConnection) setEffectivePMTU(pmtu int) { } // 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) +// TCP receiver process. StackFrag is read in conn.forward func (conn *LocalConnection) setStackFrag(frag bool) { conn.Lock() defer conn.Unlock() @@ -251,15 +250,11 @@ func (conn *LocalConnection) SetEstablished() { } // 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.Send(true, PMTUDiscovery) conn.heartbeat = time.NewTicker(SlowHeartbeat) conn.fragTest = time.NewTicker(FragTestInterval) // avoid initial waits for timers to fire - conn.Forward(true, conn.heartbeatFrame, nil) + conn.Send(true, conn.heartbeatFrame) conn.performFragTest() return nil }) @@ -355,12 +350,8 @@ func (conn *LocalConnection) run(actionChan <-chan ConnectionAction, finished ch func (conn *LocalConnection) initHeartbeats() error { conn.heartbeatTCP = time.NewTicker(TCPHeartbeat) conn.heartbeatTimeout = time.NewTimer(HeartbeatTimeout) - heartbeatFrameBytes := make([]byte, EthernetOverhead+8) - binary.BigEndian.PutUint64(heartbeatFrameBytes[EthernetOverhead:], conn.uid) - conn.heartbeatFrame = &ForwardedFrame{ - srcPeer: conn.local, - dstPeer: conn.remote, - frame: heartbeatFrameBytes} + conn.heartbeatFrame = make([]byte, EthernetOverhead+8) + binary.BigEndian.PutUint64(conn.heartbeatFrame[EthernetOverhead:], conn.uid) if conn.remoteUDPAddr == nil { return nil } @@ -377,7 +368,7 @@ func (conn *LocalConnection) actorLoop(actionChan <-chan ConnectionAction) (err case <-conn.heartbeatTimeout.C: err = fmt.Errorf("timed out waiting for UDP heartbeat") case <-tickerChan(conn.heartbeat): - conn.Forward(true, conn.heartbeatFrame, nil) + conn.Send(true, conn.heartbeatFrame) case <-tickerChan(conn.fragTest): conn.performFragTest() } @@ -485,18 +476,14 @@ func (conn *LocalConnection) sendFastHeartbeats() error { err := conn.ensureForwarders() if err == nil { conn.heartbeat = time.NewTicker(FastHeartbeat) - conn.Forward(true, conn.heartbeatFrame, nil) // avoid initial wait + conn.Send(true, conn.heartbeatFrame) // avoid initial wait } return err } func (conn *LocalConnection) performFragTest() { conn.setStackFrag(false) - conn.Forward(false, &ForwardedFrame{ - srcPeer: conn.local, - dstPeer: conn.remote, - frame: FragTest}, - nil) + conn.Send(false, FragTest) } func tickerChan(ticker *time.Ticker) <-chan time.Time { diff --git a/router/forwarder.go b/router/forwarder.go index cab1c0ff..9143c030 100644 --- a/router/forwarder.go +++ b/router/forwarder.go @@ -59,11 +59,23 @@ func (conn *LocalConnection) stopForwarders() { conn.forwarderDF.Shutdown() } -// Called from peer.Relay[Broadcast] which is itself invoked from -// router (both UDP listener process and sniffer process). Also called -// from connection's heartbeat process, and from the connection's TCP -// receiver process. -func (conn *LocalConnection) Forward(df bool, frame *ForwardedFrame, dec *EthernetDecoder) error { +// Called from connection's actor process, and from the connection's +// TCP receiver process. +func (conn *LocalConnection) Send(df bool, frameBytes []byte) error { + frame := &ForwardedFrame{ + srcPeer: conn.local, + dstPeer: conn.remote, + frame: frameBytes} + return conn.forward(frame, nil, df) +} + +// Called from LocalPeer.Relay[Broadcast] which is itself invoked from +// router (both UDP listener process and sniffer process). +func (conn *LocalConnection) Forward(frame *ForwardedFrame, dec *EthernetDecoder) error { + return conn.forward(frame, dec, dec != nil && dec.DF()) +} + +func (conn *LocalConnection) forward(frame *ForwardedFrame, dec *EthernetDecoder, df bool) error { conn.RLock() var ( forwarder = conn.forwarder @@ -78,10 +90,6 @@ func (conn *LocalConnection) Forward(df bool, frame *ForwardedFrame, dec *Ethern return nil } - if dec != nil { - df = dec.DF() - } - // We could use non-blocking channel sends here, i.e. drop frames // on the floor when the forwarder is busy. This would allow our // caller - the capturing loop in the router - to read frames more diff --git a/router/local_peer.go b/router/local_peer.go index 930fbf1b..3463e6ba 100644 --- a/router/local_peer.go +++ b/router/local_peer.go @@ -50,7 +50,7 @@ func (peer *LocalPeer) Relay(srcPeer, dstPeer *Peer, frame []byte, dec *Ethernet log.Println("Unable to find connection to relay peer", relayPeerName) return nil } - return conn.(*LocalConnection).Forward(false, &ForwardedFrame{ + return conn.(*LocalConnection).Forward(&ForwardedFrame{ srcPeer: srcPeer, dstPeer: dstPeer, frame: frame}, @@ -63,7 +63,7 @@ func (peer *LocalPeer) RelayBroadcast(srcPeer *Peer, frame []byte, dec *Ethernet return } for _, conn := range peer.ConnectionsTo(nextHops) { - err := conn.(*LocalConnection).Forward(false, &ForwardedFrame{ + err := conn.(*LocalConnection).Forward(&ForwardedFrame{ srcPeer: srcPeer, dstPeer: conn.Remote(), frame: frame},