From 341dac1be90b4b334438297d4893ae069d450555 Mon Sep 17 00:00:00 2001 From: Etienne Perot Date: Wed, 15 May 2024 16:56:26 -0700 Subject: [PATCH] Profiling metrics: Make profiling metric collection more reliable. MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This includes multiple improvements which, when combined, get me a clean run on a long-running benchmark like `bazel_test` with a 100μs sample rate. This includes: - Not using `time.Sleep` for short durations (this made the most impact), instead it either spins or yields. - Adaptive initially-faster backoff time that increases exponentially if it finds that it's not not sleeping long enough, rather than 100x profiling rate backoff right off the bat. - Larger buffers. - More precise computation for duration by using a mix of `nanotime` and actual `time.Now` readjusted once every N collections. - Using an atomic boolean for checking if metric collection should be stopped (should be cheaper than checking if a channel has data). - Putting the ringbuffer item slice being processed on the collector/writer stacks rather than dereferencing it for each cell. PiperOrigin-RevId: 634123127 --- pkg/metric/profiling_metric.go | 136 ++++++++++++++++++++++----------- 1 file changed, 93 insertions(+), 43 deletions(-) diff --git a/pkg/metric/profiling_metric.go b/pkg/metric/profiling_metric.go index 6c2684fb7..aa321077e 100644 --- a/pkg/metric/profiling_metric.go +++ b/pkg/metric/profiling_metric.go @@ -22,6 +22,7 @@ import ( "hash/adler32" "io" "os" + "runtime" "strings" "time" @@ -32,8 +33,17 @@ import ( ) const ( - snapshotBufferSize = 1000 - snapshotRingbufferSize = 16 + // snapshotBufferSize is the number of snapshots within one item of the + // ringbuffer. Increasing this number means less context-switching + // overhead between collector and writer goroutines, but worse time + // precision, as the precise time is refreshed every this many snapshots. + snapshotBufferSize = 1024 + // snapshotRingbufferSize is the number of items in the ringbuffer. + // Increasing this number means the writer has more slack to catch up + // if it falls behind, but it also means that the collector may need + // to wait for longer intervals when the writer does fall behind, + // adding more variance to the time gaps between collections. + snapshotRingbufferSize = 128 // MetricsPrefix is prepended before every metrics line. MetricsPrefix = "GVISOR_METRICS\t" // MetricsHashIndicator is prepended before the hash of the metrics @@ -55,7 +65,7 @@ var ( profilingMetricsStarted atomicbitops.Bool // stopProfilingMetrics is used to signal to the profiling metrics // goroutine to stop recording and writing metrics. - stopProfilingMetrics chan bool + stopProfilingMetrics atomicbitops.Bool // doneProfilingMetrics is used to signal that the profiling metrics // goroutines are finished. doneProfilingMetrics chan bool @@ -68,8 +78,7 @@ var ( // before it's written to the writer. type snapshots struct { numMetrics int - // startTime is the time at which collection started, as reported by - // CheapNowNano() which does *not* start counting from the epoch time. + // startTime is the time at which collection started in nanoseconds. startTime int64 // ringbuffer is used to store metric data. ringbuffer [][]uint64 @@ -189,11 +198,12 @@ func StartProfilingMetrics[T ProfilingMetricsWriter](opts ProfilingMetricsOption s.ringbuffer[i] = make([]uint64, snapshotBufferSize*(numMetrics+1)) } - stopProfilingMetrics = make(chan bool, 1) + stopProfilingMetrics = atomicbitops.FromBool(false) doneProfilingMetrics = make(chan bool, 1) writeCh := make(chan writeReq, snapshotRingbufferSize) - s.startTime = CheapNowNano() - go collectProfilingMetrics(&s, values, opts.Rate, writeCh) + s.startTime = time.Now().UnixNano() + cheapStartTime := CheapNowNano() + go collectProfilingMetrics(&s, values, cheapStartTime, opts.Rate, writeCh) if opts.Lossy { lossySink := newLossyBufferedWriter(opts.Sink) go writeProfilingMetrics[*lossyBufferedWriter[T]](lossySink, &s, headers, writeCh) @@ -208,57 +218,97 @@ func StartProfilingMetrics[T ProfilingMetricsWriter](opts ProfilingMetricsOption // collectProfilingMetrics will send metrics to the writeCh until it receives a // signal via the stopProfilingMetrics channel. -func collectProfilingMetrics(s *snapshots, values []func(fieldValues ...*FieldValue) uint64, profilingRate time.Duration, writeCh chan<- writeReq) { +func collectProfilingMetrics(s *snapshots, values []func(fieldValues ...*FieldValue) uint64, cheapStartTime int64, profilingRate time.Duration, writeCh chan<- writeReq) { defer close(writeCh) numEntries := s.numMetrics + 1 // to account for the timestamp ringbufferIdx := 0 curSnapshot := 0 - // getNewRingbufferIdx will block until the writer indicates that some part - // of the ringbuffer is available for writing. - getNewRingbufferIdx := func() { - for { - nextIdx := (ringbufferIdx + 1) % snapshotRingbufferSize - if nextIdx != int(s.curWriterIndex.Load()) { - ringbufferIdx = nextIdx - break - } - // Going too fast, stop collecting for a bit. - log.Warningf("Profiling metrics collector exhausted the entire ringbuffer... backing off to let writer catch up.") - time.Sleep(profilingRate * 100) - } - } + + // If we write faster than the writer can keep up, we back off. + // The backoff factor starts small but increases exponentially + // each time we find that we are still faster than the writer. + const ( + // How much slower than the profiling rate we sleep for, as a + // multiplier for the profiling rate. + initialBackoffFactor = 1.0 + + // The exponential factor by which the backoff factor increases. + backoffFactorGrowth = 1.125 + + // The maximum backoff factor, i.e. the maximum multiplier of + // the profiling rate for which we sleep. + backoffFactorMax = 256.0 + ) + backoffFactor := initialBackoffFactor + + // To keep track of time cheaply, we use `CheapNowNano`. + // However, this can drift as it has poor precision. + // To get something more precise, we periodically call `time.Now` + // and `CheapNowNano` and use these two variables to track both. + // This way, we can compute a more precise time by using + // `CheapNowNano() - cheapTime + preciseTime`. + preciseTime := s.startTime + cheapTime := cheapStartTime stopCollecting := false - for nextCollection := CheapNowNano() + profilingRate.Nanoseconds(); !stopCollecting; nextCollection += profilingRate.Nanoseconds() { - now := CheapNowNano() - if now < nextCollection { - time.Sleep(time.Duration(nextCollection-now) * time.Nanosecond) - } else { - // Skip collection since we just did one anyway. - continue + for nextCollection := s.startTime; !stopCollecting; nextCollection += profilingRate.Nanoseconds() { + + // For small durations, just spin. Otherwise sleep. + for { + const ( + wakeUpNanos = 10 + spinMaxNanos = 250 + yieldMaxNanos = 1_000 + ) + now := CheapNowNano() - cheapTime + preciseTime + nanosToNextCollection := nextCollection - now + if nanosToNextCollection <= 0 { + // Collect now. + break + } + if nanosToNextCollection < spinMaxNanos { + continue // Spin. + } + if nanosToNextCollection < yieldMaxNanos { + // Yield then spin. + runtime.Gosched() + continue + } + // Sleep. + time.Sleep(time.Duration(nanosToNextCollection-wakeUpNanos) * time.Nanosecond) } - select { - case <-stopProfilingMetrics: + if stopProfilingMetrics.Load() { stopCollecting = true // Collect one last time before stopping. - default: } - collectStart := CheapNowNano() + collectStart := CheapNowNano() - cheapTime + preciseTime timestamp := time.Duration(collectStart - s.startTime) base := curSnapshot * numEntries - s.ringbuffer[ringbufferIdx][base] = uint64(timestamp) + ringBuf := s.ringbuffer[ringbufferIdx] + ringBuf[base] = uint64(timestamp) for i := 1; i < numEntries; i++ { - s.ringbuffer[ringbufferIdx][base+i] = values[i-1]() + ringBuf[base+i] = values[i-1]() } curSnapshot++ if curSnapshot == snapshotBufferSize { writeCh <- writeReq{ringbufferIdx: ringbufferIdx, numLines: curSnapshot} curSnapshot = 0 - getNewRingbufferIdx() + // Block until the writer indicates that this part of the ringbuffer + // is available for writing. + for ringbufferIdx = (ringbufferIdx + 1) % snapshotRingbufferSize; ringbufferIdx == int(s.curWriterIndex.Load()); { + // Going too fast, stop collecting for a bit. + backoffSleep := profilingRate * time.Duration(backoffFactor) + log.Warningf("Profiling metrics collector exhausted the entire ringbuffer... backing off for %v to let writer catch up.", backoffSleep) + time.Sleep(backoffSleep) + backoffFactor = min(backoffFactor*backoffFactorGrowth, backoffFactorMax) + } + // Refresh precise time. + preciseTime = time.Now().UnixNano() + cheapTime = CheapNowNano() } } if curSnapshot != 0 { @@ -455,14 +505,15 @@ func writeProfilingMetrics[T bufferedMetricsWriter](sink T, s *snapshots, header } for req := range writeReqs { s.curWriterIndex.Store(int32(req.ringbufferIdx)) + ringBuf := s.ringbuffer[req.ringbufferIdx] for i := 0; i < req.numLines; i++ { base := i * numEntries // Write the time - prometheus.WriteInteger(sink, int64(s.ringbuffer[req.ringbufferIdx][base])) + prometheus.WriteInteger(sink, int64(ringBuf[base])) // Then everything else for j := 1; j < numEntries; j++ { sink.WriteString("\t") - prometheus.WriteInteger(sink, int64(s.ringbuffer[req.ringbufferIdx][base+j])) + prometheus.WriteInteger(sink, int64(ringBuf[base+j])) } sink.NewLine() } @@ -481,10 +532,9 @@ func StopProfilingMetrics() { if !profilingMetricsStarted.Load() { return } - - select { - case stopProfilingMetrics <- true: + if stopProfilingMetrics.CompareAndSwap(false, true) { <-doneProfilingMetrics - default: // Stop signal was already sent } + // If the CAS fails, this means the signal was already sent, + // so don't wait on doneProfilingMetrics. }