From 83724a08c95cfd6c1d0b06b6936c329303fbf0bb Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Fri, 27 Mar 2015 11:32:50 +0000 Subject: [PATCH] refactor: get rid of Forwarder.stop channel signal termination by sending nil to the main channel instead --- router/forwarder.go | 39 ++++++++++++++++++++------------------- 1 file changed, 20 insertions(+), 19 deletions(-) diff --git a/router/forwarder.go b/router/forwarder.go index b067368d..a2edb6c5 100644 --- a/router/forwarder.go +++ b/router/forwarder.go @@ -40,22 +40,18 @@ func (conn *LocalConnection) ensureForwarders() error { var ( forwardChan = make(chan *ForwardedFrame, ChannelSize) forwardChanDF = make(chan *ForwardedFrame, ChannelSize) - stopForward = make(chan interface{}, 0) - stopForwardDF = make(chan interface{}, 0) verifyPMTU = make(chan int, ChannelSize) ) //NB: only forwarderDF can ever encounter EMSGSIZE errors, and //thus perform PMTU verification - forwarder := NewForwarder(conn, forwardChan, stopForward, nil, encryptor, udpSender, DefaultPMTU) - forwarderDF := NewForwarder(conn, forwardChanDF, stopForwardDF, verifyPMTU, encryptorDF, udpSenderDF, DefaultPMTU) + forwarder := NewForwarder(conn, forwardChan, nil, encryptor, udpSender, DefaultPMTU) + forwarderDF := NewForwarder(conn, forwardChanDF, verifyPMTU, encryptorDF, udpSenderDF, DefaultPMTU) // Various fields in the conn struct are read by other processes, // so we have to use locks. conn.Lock() conn.forwardChan = forwardChan conn.forwardChanDF = forwardChanDF - conn.stopForward = stopForward - conn.stopForwardDF = stopForwardDF conn.verifyPMTU = verifyPMTU conn.effectivePMTU = forwarder.unverifiedPMTU conn.Unlock() @@ -68,16 +64,16 @@ func (conn *LocalConnection) ensureForwarders() error { func (conn *LocalConnection) stopForwarders() { conn.Lock() + forwardChan := conn.forwardChan + forwardChanDF := conn.forwardChanDF conn.forwardChan = nil conn.forwardChanDF = nil conn.Unlock() // Now signal the forwarder loops to exit. They will drain the // forwarder chans in order to unblock any router processes // blocked on sending. - if conn.stopForward != nil { - conn.stopForward <- nil - conn.stopForwardDF <- nil - } + forwardChan <- nil + forwardChanDF <- nil } // Called from peer.Relay[Broadcast] which is itself invoked from @@ -204,7 +200,6 @@ func fragment(eth layers.Ethernet, ip layers.IPv4, pmtu int, frame *ForwardedFra type Forwarder struct { conn *LocalConnection ch <-chan *ForwardedFrame - stop <-chan interface{} verifyPMTUTick <-chan time.Time verifyPMTU <-chan int pmtuVerifyCount uint @@ -217,11 +212,10 @@ type Forwarder struct { lowestBadPMTU int } -func NewForwarder(conn *LocalConnection, ch <-chan *ForwardedFrame, stop <-chan interface{}, verifyPMTU <-chan int, enc Encryptor, udpSender UDPSender, pmtu int) *Forwarder { +func NewForwarder(conn *LocalConnection, ch <-chan *ForwardedFrame, verifyPMTU <-chan int, enc Encryptor, udpSender UDPSender, pmtu int) *Forwarder { fwd := &Forwarder{ conn: conn, ch: ch, - stop: stop, verifyPMTU: verifyPMTU, enc: enc, udpSender: udpSender} @@ -238,9 +232,6 @@ func (fwd *Forwarder) run() { defer fwd.udpSender.Shutdown() for { select { - case <-fwd.stop: - fwd.drain() - return case <-fwd.verifyPMTUTick: // We only do this case here when we know the buffers are // all empty so that we don't risk appending verify-frames @@ -272,11 +263,17 @@ func (fwd *Forwarder) run() { fwd.conn.Log("Effective PMTU verified at", epmtu) } case frame := <-fwd.ch: + if frame == nil { + fwd.drain() + return + } if !fwd.appendFrame(frame) { fwd.logDrop(frame) continue } - fwd.accumulateAndSendFrames() + if !fwd.accumulateAndSendFrames() { + return + } } } } @@ -310,10 +307,14 @@ func (fwd *Forwarder) attemptVerifyEffectivePMTU() { // franes get sent to the forwarder, we can be going around this loop // forever. That is bad since there may be other stuff for us to do, // i.e. the other branches in of the run loop. -func (fwd *Forwarder) accumulateAndSendFrames() { +func (fwd *Forwarder) accumulateAndSendFrames() bool { for { select { case frame := <-fwd.ch: + if frame == nil { + fwd.drain() + return false + } if !fwd.appendFrame(frame) { fwd.flush() if !fwd.appendFrame(frame) { @@ -322,7 +323,7 @@ func (fwd *Forwarder) accumulateAndSendFrames() { } default: fwd.flush() - return + return true } } }