From 9f6a8e8dc6b3e0062789a1fe525528aa029dde93 Mon Sep 17 00:00:00 2001 From: Paul Lorenz Date: Wed, 10 Feb 2021 14:40:24 -0500 Subject: [PATCH 1/2] Remove histograms from data flows paths to reduce lock contention --- router/xgress/link_receive_buffer.go | 3 --- router/xgress/link_send_buffer.go | 5 ----- router/xgress/metrics.go | 12 ------------ router/xgress/xgress.go | 1 - 4 files changed, 21 deletions(-) diff --git a/router/xgress/link_receive_buffer.go b/router/xgress/link_receive_buffer.go index 34e6e1ea7..cc31bee83 100644 --- a/router/xgress/link_receive_buffer.go +++ b/router/xgress/link_receive_buffer.go @@ -58,12 +58,10 @@ func (buffer *LinkReceiveBuffer) ReceiveUnordered(payload *Payload, maxSize uint payloadSize := len(payload.Data) size := atomic.AddUint32(&buffer.size, uint32(payloadSize)) pfxlog.Logger().Tracef("Payload %v of size %v added to transmit buffer. New size: %v", payload.Sequence, payloadSize, size) - localRecvBufferSizeBytesHistogram.Update(int64(size)) if payload.Sequence > buffer.maxSequence { buffer.maxSequence = payload.Sequence } } - localRecvBufferSizeMsgsHistogram.Update(int64(buffer.tree.Size())) return true } @@ -80,7 +78,6 @@ func (buffer *LinkReceiveBuffer) PeekHead() *Payload { func (buffer *LinkReceiveBuffer) Remove(payload *Payload) { buffer.tree.Remove(payload.Sequence) buffer.sequence = payload.Sequence - localRecvBufferSizeMsgsHistogram.Update(int64(buffer.tree.Size())) } func (buffer *LinkReceiveBuffer) getLastBufferSizeSent() uint32 { diff --git a/router/xgress/link_send_buffer.go b/router/xgress/link_send_buffer.go index a69df1234..4bec6fd12 100644 --- a/router/xgress/link_send_buffer.go +++ b/router/xgress/link_send_buffer.go @@ -183,9 +183,6 @@ func (buffer *LinkSendBuffer) run() { defer retransmitTicker.Stop() for { - txBufferSizeHistogram.Update(int64(buffer.linkSendBufferSize)) - txWindowSize.Update(int64(buffer.windowsSize)) - // don't block when we're closing, since the only thing that should still be coming in is end-of-session if buffer.isBlocked() && !buffer.closeWhenEmpty.Get() { buffered = nil @@ -286,11 +283,9 @@ func (buffer *LinkSendBuffer) receiveAcknowledgement(ack *Acknowledgement) { } buffer.linkRecvBufferSize = ack.RecvBufferSize - remoteRecvBufferSizeHistogram.Update(int64(buffer.linkRecvBufferSize)) if ack.RTT > 0 { rtt := uint16(info.NowInMilliseconds()) - ack.RTT buffer.retxThreshold = uint32(float64(rtt)*buffer.retxScale) + buffer.x.Options.RetxAddMs - rttHistogram.Update(int64(rtt)) } } diff --git a/router/xgress/metrics.go b/router/xgress/metrics.go index 88ab78e6a..762d94fce 100644 --- a/router/xgress/metrics.go +++ b/router/xgress/metrics.go @@ -12,14 +12,8 @@ var retransmissions metrics.Meter var retransmissionFailures metrics.Meter var ackFailures metrics.Meter -var rttHistogram metrics.Histogram -var txBufferSizeHistogram metrics.Histogram -var localRecvBufferSizeBytesHistogram metrics.Histogram -var localRecvBufferSizeMsgsHistogram metrics.Histogram -var remoteRecvBufferSizeHistogram metrics.Histogram var payloadWriteTimer metrics.Timer var duplicateAcksMeter metrics.Meter -var txWindowSize metrics.Histogram var buffersBlockedByLocalWindow int64 var buffersBlockedByRemoteWindow int64 @@ -33,14 +27,8 @@ func InitMetrics(registry metrics.UsageRegistry) { ackRxMeter = registry.Meter("xgress.rx.acks") ackTxMeter = registry.Meter("xgress.tx.acks") ackFailures = registry.Meter("xgress.ack_failures") - rttHistogram = registry.Histogram("xgress.rtt") - txBufferSizeHistogram = registry.Histogram("xgress.tx_buffer_size") - localRecvBufferSizeBytesHistogram = registry.Histogram("xgress.local.rx_buffer_bytes_size") - localRecvBufferSizeMsgsHistogram = registry.Histogram("xgress.local.rx_buffer_msgs_size") - remoteRecvBufferSizeHistogram = registry.Histogram("xgress.remote.rx_buffer_size") payloadWriteTimer = registry.Timer("xgress.tx_write_time") duplicateAcksMeter = registry.Meter("xgress.ack_duplicates") - txWindowSize = registry.Histogram("xgress.tx_window_size") registry.FuncGauge("xgress.blocked_by_local_window", func() int64 { return atomic.LoadInt64(&buffersBlockedByLocalWindow) diff --git a/router/xgress/xgress.go b/router/xgress/xgress.go index 8d65083d0..32980da85 100644 --- a/router/xgress/xgress.go +++ b/router/xgress/xgress.go @@ -409,7 +409,6 @@ func (self *Xgress) tx() { payloadSize := len(payload.Data) size := atomic.AddUint32(&self.linkRxBuffer.size, ^uint32(payloadSize-1)) // subtraction for uint32 payloadLogger.Debugf("Payload %v of size %v removed from rx buffer. New size: %v", payload.Sequence, payloadSize, size) - localRecvBufferSizeBytesHistogram.Update(int64(size)) lastBufferSizeSent := self.linkRxBuffer.getLastBufferSizeSent() if lastBufferSizeSent > 10000 && (lastBufferSizeSent>>1) > size { From a1e1e6fa80eb91640881e76ed257e5b30be810d1 Mon Sep 17 00:00:00 2001 From: Paul Lorenz Date: Wed, 10 Feb 2021 15:27:24 -0500 Subject: [PATCH 2/2] Remove unnecessary check. Make debugs -> warns --- router/forwarder/forwarder.go | 4 ++-- router/xgress/xgress.go | 4 +--- 2 files changed, 3 insertions(+), 5 deletions(-) diff --git a/router/forwarder/forwarder.go b/router/forwarder/forwarder.go index 82e5c045f..54da37475 100644 --- a/router/forwarder/forwarder.go +++ b/router/forwarder/forwarder.go @@ -83,12 +83,12 @@ func (forwarder *Forwarder) UnregisterDestinations(sessionId string) { forwarder.destinations.removeDestination(address) go destination.(XgressDestination).Close() // create close queue? } else { - pfxlog.Logger().Debugf("no destinations found for [@/%v] for [s/%v]", address, sessionId) + pfxlog.Logger().Warnf("no destinations found for [@/%v] for [s/%v]", address, sessionId) } } forwarder.destinations.unlinkSession(sessionId) } else { - pfxlog.Logger().Debugf("found no addresses to unregister for [s/%v]", sessionId) + pfxlog.Logger().Warnf("found no addresses to unregister for [s/%v]", sessionId) } } diff --git a/router/xgress/xgress.go b/router/xgress/xgress.go index 32980da85..b837192ca 100644 --- a/router/xgress/xgress.go +++ b/router/xgress/xgress.go @@ -525,9 +525,7 @@ func (self *Xgress) nextReceiveSequence() int32 { func (self *Xgress) closeTimeoutHandler(duration time.Duration) { self.payloadBuffer.CloseWhenEmpty() // If we clear the send buffer, close sooner time.Sleep(duration) - if !self.closed.Get() { - self.Close() - } + self.Close() } func (self *Xgress) PayloadReceived(payload *Payload) {