From 1640669e13bab5e8a64eaf0f855dc59e0b8634bf Mon Sep 17 00:00:00 2001 From: Paul Lorenz Date: Tue, 15 Dec 2020 13:02:30 -0500 Subject: [PATCH] Shutdown payload ingester, acker and retransmitter go-routines on router shutdown --- router/forwarder/forwarder.go | 8 +++--- router/router.go | 7 ++++-- router/xgress/acker.go | 41 ++++++++++++++++++++----------- router/xgress/ordering_test.go | 5 ++-- router/xgress/payload_ingester.go | 10 +++++--- router/xgress/retransmitter.go | 35 +++++++++++++++++--------- 6 files changed, 68 insertions(+), 38 deletions(-) diff --git a/router/forwarder/forwarder.go b/router/forwarder/forwarder.go index 1a46a5391..046413865 100644 --- a/router/forwarder/forwarder.go +++ b/router/forwarder/forwarder.go @@ -49,7 +49,7 @@ type XgressDestination interface { Label() string } -func NewForwarder(metricsRegistry metrics.UsageRegistry, options *Options) *Forwarder { +func NewForwarder(metricsRegistry metrics.UsageRegistry, options *Options, closeNotify <-chan struct{}) *Forwarder { forwarder := &Forwarder{ sessions: newSessionTable(), destinations: newDestinationTable(), @@ -58,9 +58,9 @@ func NewForwarder(metricsRegistry metrics.UsageRegistry, options *Options) *Forw Options: options, } - xgress.InitPayloadIngester() - xgress.InitAcker(forwarder, metricsRegistry) - xgress.InitRetransmitter(forwarder, metricsRegistry) + xgress.InitPayloadIngester(closeNotify) + xgress.InitAcker(forwarder, metricsRegistry, closeNotify) + xgress.InitRetransmitter(forwarder, metricsRegistry, closeNotify) return forwarder } diff --git a/router/router.go b/router/router.go index 08dcc77e2..c4a8b469a 100644 --- a/router/router.go +++ b/router/router.go @@ -81,11 +81,14 @@ func Create(config *Config, versionProvider common.VersionProvider) *Router { metricsRegistry := metrics.NewUsageRegistryFromConfig(metricsConfig) xgress.InitMetrics(metricsRegistry) + closeNotify := make(chan struct{}) + fwd := forwarder.NewForwarder(metricsRegistry, config.Forwarder, closeNotify) + return &Router{ config: config, - forwarder: forwarder.NewForwarder(metricsRegistry, config.Forwarder), + forwarder: fwd, metricsRegistry: metricsRegistry, - shutdownC: make(chan struct{}), + shutdownC: closeNotify, eventDispatcher: eventDispatcher, versionProvider: versionProvider, } diff --git a/router/xgress/acker.go b/router/xgress/acker.go index 06c4692ff..7c8abd7cc 100644 --- a/router/xgress/acker.go +++ b/router/xgress/acker.go @@ -10,8 +10,8 @@ import ( var acker *Acker -func InitAcker(forwarder PayloadBufferForwarder, metrics metrics.Registry) { - acker = NewAcker(forwarder, metrics) +func InitAcker(forwarder PayloadBufferForwarder, metrics metrics.Registry, closeNotify <-chan struct{}) { + acker = NewAcker(forwarder, metrics, closeNotify) } type ackEntry struct { @@ -25,14 +25,16 @@ type Acker struct { ackIngest chan *ackEntry ackSend chan *ackEntry acksQueueSize int64 + closeNotify <-chan struct{} } -func NewAcker(forwarder PayloadBufferForwarder, metrics metrics.Registry) *Acker { +func NewAcker(forwarder PayloadBufferForwarder, metrics metrics.Registry, closeNotify <-chan struct{}) *Acker { result := &Acker{ - forwarder: forwarder, - acks: deque.New(), - ackIngest: make(chan *ackEntry, 16), - ackSend: make(chan *ackEntry, 1), + forwarder: forwarder, + acks: deque.New(), + ackIngest: make(chan *ackEntry, 16), + ackSend: make(chan *ackEntry, 1), + closeNotify: closeNotify, } go result.ackIngester() @@ -65,6 +67,8 @@ func (acker *Acker) ackIngester() { select { case ack := <-acker.ackIngest: acker.acks.PushBack(ack) + case <-acker.closeNotify: + return } } else { select { @@ -72,6 +76,8 @@ func (acker *Acker) ackIngester() { acker.acks.PushBack(ack) case acker.ackSend <- next: next = nil + case <-acker.closeNotify: + return } } atomic.StoreInt64(&acker.acksQueueSize, int64(acker.acks.Len())) @@ -80,14 +86,19 @@ func (acker *Acker) ackIngester() { func (acker *Acker) ackSender() { logger := pfxlog.Logger() - for nextAck := range acker.ackSend { - now := time.Now() - if err := acker.forwarder.ForwardAcknowledgement(nextAck.Address, nextAck.Acknowledgement); err != nil { - logger.WithError(err).Debugf("unexpected error while sending ack from %v", nextAck.Address) - ackFailures.Mark(1) - } else { - ackWriteTimer.UpdateSince(now) - ackTxMeter.Mark(1) + for { + select { + case nextAck := <-acker.ackSend: + now := time.Now() + if err := acker.forwarder.ForwardAcknowledgement(nextAck.Address, nextAck.Acknowledgement); err != nil { + logger.WithError(err).Debugf("unexpected error while sending ack from %v", nextAck.Address) + ackFailures.Mark(1) + } else { + ackWriteTimer.UpdateSince(now) + ackTxMeter.Mark(1) + } + case <-acker.closeNotify: + return } } } diff --git a/router/xgress/ordering_test.go b/router/xgress/ordering_test.go index e0ee408c9..a75c188cd 100644 --- a/router/xgress/ordering_test.go +++ b/router/xgress/ordering_test.go @@ -52,9 +52,10 @@ func (n noopForwarder) ForwardAcknowledgement(srcAddr Address, acknowledgement * func Test_Ordering(t *testing.T) { metricsRegistry := metrics.NewUsageRegistry("test", map[string]string{}, time.Minute, noopMetricsHandler{}) - InitPayloadIngester() + closeNotify := make(chan struct{}) + InitPayloadIngester(closeNotify) InitMetrics(metricsRegistry) - InitAcker(&noopForwarder{}, metricsRegistry) + InitAcker(&noopForwarder{}, metricsRegistry, closeNotify) conn := &testConn{ ch: make(chan uint64, 1), diff --git a/router/xgress/payload_ingester.go b/router/xgress/payload_ingester.go index c4db5e351..b96ca2048 100644 --- a/router/xgress/payload_ingester.go +++ b/router/xgress/payload_ingester.go @@ -2,8 +2,8 @@ package xgress var payloadIngester *PayloadIngester -func InitPayloadIngester() { - payloadIngester = NewPayloadIngester() +func InitPayloadIngester(closeNotify <-chan struct{}) { + payloadIngester = NewPayloadIngester(closeNotify) } type payloadEntry struct { @@ -14,12 +14,14 @@ type payloadEntry struct { type PayloadIngester struct { payloadIngest chan *payloadEntry payloadSendReq chan *Xgress + closeNotify <-chan struct{} } -func NewPayloadIngester() *PayloadIngester { +func NewPayloadIngester(closeNotify <-chan struct{}) *PayloadIngester { pi := &PayloadIngester{ payloadIngest: make(chan *payloadEntry, 16), payloadSendReq: make(chan *Xgress, 16), + closeNotify: closeNotify, } go pi.run() @@ -41,6 +43,8 @@ func (payloadIngester *PayloadIngester) run() { payloadEntry.x.payloadIngester(payloadEntry.payload) case x := <-payloadIngester.payloadSendReq: x.queueSends() + case <-payloadIngester.closeNotify: + return } } } diff --git a/router/xgress/retransmitter.go b/router/xgress/retransmitter.go index 8e5385b32..797fc473c 100644 --- a/router/xgress/retransmitter.go +++ b/router/xgress/retransmitter.go @@ -8,8 +8,8 @@ import ( var retransmitter *Retransmitter -func InitRetransmitter(forwarder PayloadBufferForwarder, metrics metrics.Registry) { - retransmitter = NewRetransmitter(forwarder, metrics) +func InitRetransmitter(forwarder PayloadBufferForwarder, metrics metrics.Registry, closeNotify <-chan struct{}) { + retransmitter = NewRetransmitter(forwarder, metrics, closeNotify) } type Retransmitter struct { @@ -19,13 +19,15 @@ type Retransmitter struct { retransmitIngest chan *txPayload retransmitSend chan *txPayload retransmitsQueueSize int64 + closeNotify <-chan struct{} } -func NewRetransmitter(forwarder PayloadBufferForwarder, metrics metrics.Registry) *Retransmitter { +func NewRetransmitter(forwarder PayloadBufferForwarder, metrics metrics.Registry, closeNotify <-chan struct{}) *Retransmitter { ctrl := &Retransmitter{ forwarder: forwarder, retransmitIngest: make(chan *txPayload, 16), retransmitSend: make(chan *txPayload, 1), + closeNotify: closeNotify, } go ctrl.retransmitIngester() @@ -107,6 +109,8 @@ func (retransmitter *Retransmitter) retransmitIngester() { select { case retransmit := <-retransmitter.retransmitIngest: retransmitter.acceptRetransmit(retransmit) + case <-retransmitter.closeNotify: + return } } else { select { @@ -114,6 +118,8 @@ func (retransmitter *Retransmitter) retransmitIngester() { retransmitter.acceptRetransmit(retransmit) case retransmitter.retransmitSend <- next: next = nil + case <-retransmitter.closeNotify: + return } } } @@ -129,16 +135,21 @@ func (retransmitter *Retransmitter) acceptRetransmit(txp *txPayload) { func (retransmitter *Retransmitter) retransmitSender() { logger := pfxlog.Logger() - for retransmit := range retransmitter.retransmitSend { - if !retransmit.isAcked() { - if err := retransmitter.forwarder.ForwardPayload(retransmit.x.address, retransmit.payload); err != nil { - logger.WithError(err).Errorf("unexpected error while retransmitting payload from %v", retransmit.x.address) - retransmissionFailures.Mark(1) - } else { - retransmit.markSent() - retransmissions.Mark(1) + for { + select { + case retransmit := <-retransmitter.retransmitSend: + if !retransmit.isAcked() { + if err := retransmitter.forwarder.ForwardPayload(retransmit.x.address, retransmit.payload); err != nil { + logger.WithError(err).Errorf("unexpected error while retransmitting payload from %v", retransmit.x.address) + retransmissionFailures.Mark(1) + } else { + retransmit.markSent() + retransmissions.Mark(1) + } + retransmit.dequeued() } - retransmit.dequeued() + case <-retransmitter.closeNotify: + return } } }