From 5f9b28cfc1279c2e384bdb7255b56f0bdedf0b5b Mon Sep 17 00:00:00 2001 From: Paul Lorenz Date: Mon, 18 Jan 2021 11:29:36 -0500 Subject: [PATCH] Shutdown ack handler on link close. Remove unnecessary timers --- router/handler_link/accepter.go | 6 +++-- router/handler_link/ack.go | 6 ----- router/handler_link/close.go | 24 +++++++++++++++---- router/handler_link/payload.go | 6 ----- router/handler_link/queue_ack.go | 40 +++++++++++++++++--------------- router/xgress/acker.go | 3 --- router/xgress/metrics.go | 6 ----- router/xgress/xgress.go | 4 ---- 8 files changed, 44 insertions(+), 51 deletions(-) diff --git a/router/handler_link/accepter.go b/router/handler_link/accepter.go index 90a4b5087..4ff5ec937 100644 --- a/router/handler_link/accepter.go +++ b/router/handler_link/accepter.go @@ -21,12 +21,14 @@ func NewChannelAccepter(c xgress.CtrlChannel, f *forwarder.Forwarder, fo *forwar } func (self *channelAccepter) AcceptChannel(xlink xlink.Xlink, ch channel2.Channel, trackLatency bool) error { + closeNotify := make(chan struct{}) + ch.SetLogicalName("l/" + xlink.Id().Token) ch.SetUserData(xlink.Id().Token) - ch.AddCloseHandler(newCloseHandler(xlink, self.ctrl, self.forwarder)) + ch.AddCloseHandler(newCloseHandler(xlink, self.ctrl, self.forwarder, closeNotify)) ch.AddErrorHandler(newErrorHandler(xlink, self.ctrl)) ch.AddReceiveHandler(newPayloadHandler(xlink, self.ctrl, self.forwarder)) - ch.AddReceiveHandler(newQueuingAckHandler(xlink, self.ctrl, self.forwarder)) + ch.AddReceiveHandler(newQueuingAckHandler(xlink, self.ctrl, self.forwarder, closeNotify)) ch.AddReceiveHandler(&channel2.LatencyHandler{}) ch.AddPeekHandler(metrics2.NewChannelPeekHandler(xlink.Id().Token, self.forwarder.MetricsRegistry())) ch.AddPeekHandler(trace.NewChannelPeekHandler(xlink.Id(), ch, self.forwarder.TraceController(), trace.NewChannelSink(self.ctrl.Channel()))) diff --git a/router/handler_link/ack.go b/router/handler_link/ack.go index d6b318899..18c62434c 100644 --- a/router/handler_link/ack.go +++ b/router/handler_link/ack.go @@ -22,15 +22,12 @@ import ( "github.com/openziti/fabric/router/xgress" "github.com/openziti/fabric/router/xlink" "github.com/openziti/foundation/channel2" - "github.com/openziti/foundation/metrics" - "time" ) type ackHandler struct { link xlink.Xlink ctrl xgress.CtrlChannel forwarder *forwarder.Forwarder - timer metrics.Timer } func newAckHandler(link xlink.Xlink, ctrl xgress.CtrlChannel, forwarder *forwarder.Forwarder) *ackHandler { @@ -38,7 +35,6 @@ func newAckHandler(link xlink.Xlink, ctrl xgress.CtrlChannel, forwarder *forward link: link, ctrl: ctrl, forwarder: forwarder, - timer: forwarder.MetricsRegistry().Timer("xgress.ack.handle_time"), } } @@ -47,7 +43,6 @@ func (self *ackHandler) ContentType() int32 { } func (self *ackHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) { - start := time.Now() log := pfxlog.ContextLogger(ch.Label()) if ack, err := xgress.UnmarshallAcknowledgement(msg); err == nil { @@ -57,5 +52,4 @@ func (self *ackHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel } else { log.Errorf("unexpected error (%v)", err) } - self.timer.UpdateSince(start) } diff --git a/router/handler_link/close.go b/router/handler_link/close.go index ac87ea79c..5c05656e4 100644 --- a/router/handler_link/close.go +++ b/router/handler_link/close.go @@ -24,19 +24,33 @@ import ( "github.com/openziti/fabric/router/xgress" "github.com/openziti/fabric/router/xlink" "github.com/openziti/foundation/channel2" + "github.com/openziti/foundation/util/concurrenz" ) type closeHandler struct { - link xlink.Xlink - ctrl xgress.CtrlChannel - forwarder *forwarder.Forwarder + link xlink.Xlink + ctrl xgress.CtrlChannel + forwarder *forwarder.Forwarder + closeNotify chan struct{} + closed concurrenz.AtomicBoolean } -func newCloseHandler(link xlink.Xlink, ctrl xgress.CtrlChannel, forwarder *forwarder.Forwarder) *closeHandler { - return &closeHandler{link: link, ctrl: ctrl, forwarder: forwarder} +func newCloseHandler(link xlink.Xlink, ctrl xgress.CtrlChannel, forwarder *forwarder.Forwarder, closeNotify chan struct{}) *closeHandler { + return &closeHandler{ + link: link, + ctrl: ctrl, + forwarder: forwarder, + closeNotify: closeNotify, + } } func (self *closeHandler) HandleClose(ch channel2.Channel) { + defer func() { + if self.closed.CompareAndSwap(false, true) { + close(self.closeNotify) + } + }() + log := pfxlog.ContextLogger(ch.Label()) log.Info("link closed") diff --git a/router/handler_link/payload.go b/router/handler_link/payload.go index 52f2f3169..d9ea65abf 100644 --- a/router/handler_link/payload.go +++ b/router/handler_link/payload.go @@ -22,15 +22,12 @@ import ( "github.com/openziti/fabric/router/xgress" "github.com/openziti/fabric/router/xlink" "github.com/openziti/foundation/channel2" - "github.com/openziti/foundation/metrics" - "time" ) type payloadHandler struct { link xlink.Xlink ctrl xgress.CtrlChannel forwarder *forwarder.Forwarder - timer metrics.Timer } func newPayloadHandler(link xlink.Xlink, ctrl xgress.CtrlChannel, forwarder *forwarder.Forwarder) *payloadHandler { @@ -38,7 +35,6 @@ func newPayloadHandler(link xlink.Xlink, ctrl xgress.CtrlChannel, forwarder *for link: link, ctrl: ctrl, forwarder: forwarder, - timer: forwarder.MetricsRegistry().Timer("xgress.payload.handle_time"), } } @@ -47,7 +43,6 @@ func (self *payloadHandler) ContentType() int32 { } func (self *payloadHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) { - start := time.Now() log := pfxlog.ContextLogger(ch.Label()) payload, err := xgress.UnmarshallPayload(msg) @@ -61,5 +56,4 @@ func (self *payloadHandler) HandleReceive(msg *channel2.Message, ch channel2.Cha } else { log.Errorf("unexpected error (%v)", err) } - self.timer.UpdateSince(start) } diff --git a/router/handler_link/queue_ack.go b/router/handler_link/queue_ack.go index c4080c80e..b7d720ad1 100644 --- a/router/handler_link/queue_ack.go +++ b/router/handler_link/queue_ack.go @@ -23,31 +23,29 @@ import ( "github.com/openziti/fabric/router/xgress" "github.com/openziti/fabric/router/xlink" "github.com/openziti/foundation/channel2" - "github.com/openziti/foundation/metrics" "sync/atomic" - "time" ) type queuingAckHandler struct { link xlink.Xlink ctrl xgress.CtrlChannel forwarder *forwarder.Forwarder - timer metrics.Timer acks *deque.Deque ackIngest chan *xgress.Acknowledgement ackForward chan *xgress.Acknowledgement acksQueueSize int64 + closeNotify <-chan struct{} } -func newQueuingAckHandler(link xlink.Xlink, ctrl xgress.CtrlChannel, forwarder *forwarder.Forwarder) *queuingAckHandler { +func newQueuingAckHandler(link xlink.Xlink, ctrl xgress.CtrlChannel, forwarder *forwarder.Forwarder, closeNotify <-chan struct{}) *queuingAckHandler { result := &queuingAckHandler{ - link: link, - ctrl: ctrl, - forwarder: forwarder, - timer: forwarder.MetricsRegistry().Timer("xgress.ack.handle_time"), - acks: deque.New(), - ackIngest: make(chan *xgress.Acknowledgement, 16), - ackForward: make(chan *xgress.Acknowledgement, 1), + link: link, + ctrl: ctrl, + forwarder: forwarder, + acks: deque.New(), + ackIngest: make(chan *xgress.Acknowledgement, 16), + ackForward: make(chan *xgress.Acknowledgement, 1), + closeNotify: closeNotify, } go result.ackIngester() @@ -65,7 +63,6 @@ func (self *queuingAckHandler) ContentType() int32 { } func (self *queuingAckHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) { - start := time.Now() log := pfxlog.ContextLogger(ch.Label()) if ack, err := xgress.UnmarshallAcknowledgement(msg); err == nil { @@ -73,7 +70,6 @@ func (self *queuingAckHandler) HandleReceive(msg *channel2.Message, ch channel2. } else { log.Errorf("unexpected error (%v)", err) } - self.timer.UpdateSince(start) } func (self *queuingAckHandler) ackIngester() { @@ -89,6 +85,8 @@ func (self *queuingAckHandler) ackIngester() { select { case ack := <-self.ackIngest: self.acks.PushBack(ack) + case <-self.closeNotify: + return } } else { select { @@ -96,6 +94,8 @@ func (self *queuingAckHandler) ackIngester() { self.acks.PushBack(ack) case self.ackForward <- next: next = nil + case <-self.closeNotify: + return } } atomic.StoreInt64(&self.acksQueueSize, int64(self.acks.Len())) @@ -104,12 +104,14 @@ func (self *queuingAckHandler) ackIngester() { func (self *queuingAckHandler) ackForwarder() { logger := pfxlog.Logger() - for ack := range self.ackForward { - now := time.Now() - if err := self.forwarder.ForwardAcknowledgement(xgress.Address(self.link.Id().Token), ack); err != nil { - logger.WithError(err).Debugf("unable to forward acknowledgement (%v)", err) - } else { - self.timer.UpdateSince(now) + for { + select { + case ack := <-self.ackForward: + if err := self.forwarder.ForwardAcknowledgement(xgress.Address(self.link.Id().Token), ack); err != nil { + logger.WithError(err).Debugf("unable to forward acknowledgement (%v)", err) + } + case <-self.closeNotify: + return } } } diff --git a/router/xgress/acker.go b/router/xgress/acker.go index 7c8abd7cc..0ae24e494 100644 --- a/router/xgress/acker.go +++ b/router/xgress/acker.go @@ -5,7 +5,6 @@ import ( "github.com/michaelquigley/pfxlog" "github.com/openziti/foundation/metrics" "sync/atomic" - "time" ) var acker *Acker @@ -89,12 +88,10 @@ func (acker *Acker) ackSender() { 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: diff --git a/router/xgress/metrics.go b/router/xgress/metrics.go index 1a2f04628..88ab78e6a 100644 --- a/router/xgress/metrics.go +++ b/router/xgress/metrics.go @@ -18,9 +18,6 @@ var localRecvBufferSizeBytesHistogram metrics.Histogram var localRecvBufferSizeMsgsHistogram metrics.Histogram var remoteRecvBufferSizeHistogram metrics.Histogram var payloadWriteTimer metrics.Timer -var ackWriteTimer metrics.Timer -var payloadBufferTimer metrics.Timer -var payloadRelayTimer metrics.Timer var duplicateAcksMeter metrics.Meter var txWindowSize metrics.Histogram @@ -42,9 +39,6 @@ func InitMetrics(registry metrics.UsageRegistry) { localRecvBufferSizeMsgsHistogram = registry.Histogram("xgress.local.rx_buffer_msgs_size") remoteRecvBufferSizeHistogram = registry.Histogram("xgress.remote.rx_buffer_size") payloadWriteTimer = registry.Timer("xgress.tx_write_time") - ackWriteTimer = registry.Timer("xgress.ack_write_time") - payloadBufferTimer = registry.Timer("xgress.payload_buffer_time") - payloadRelayTimer = registry.Timer("xgress.payload_relay_time") duplicateAcksMeter = registry.Meter("xgress.ack_duplicates") txWindowSize = registry.Histogram("xgress.tx_window_size") diff --git a/router/xgress/xgress.go b/router/xgress/xgress.go index 916865296..e563af196 100644 --- a/router/xgress/xgress.go +++ b/router/xgress/xgress.go @@ -256,16 +256,12 @@ func (self *Xgress) payloadIngester(payload *Payload) { go self.rx() } - start := time.Now() if !self.Options.RandomDrops || rand.Int31n(self.Options.Drop1InN) != 1 { self.PayloadReceived(payload) } else { pfxlog.ContextLogger(self.Label()).WithFields(payload.GetLoggerFields()).Error("drop!") } - next := time.Now() - payloadBufferTimer.Update(next.Sub(start)) self.queueSends() - payloadRelayTimer.UpdateSince(next) } func (self *Xgress) queueSends() {