diff --git a/pkg/sfu/forwardstats.go b/pkg/sfu/forwardstats.go index 34bb188a4..2ad8497a7 100644 --- a/pkg/sfu/forwardstats.go +++ b/pkg/sfu/forwardstats.go @@ -2,6 +2,7 @@ package sfu import ( "math" + "sync" "time" "github.com/livekit/livekit-server/pkg/telemetry/prometheus" @@ -163,7 +164,9 @@ type ForwardStats struct { samples forwardSampleBuffer // ring of per-summary-interval summaries covering the report window. - // only touched by the background worker, so it needs no locking. + // written by the background worker (flush) and read both by the worker + // (report) and by external callers (GetStats), so it is guarded by lock. + lock sync.Mutex ring []forwardSummary ringHead int ringLen int @@ -236,18 +239,52 @@ func (s *ForwardStats) flush() { summ = summ.addSample(transitNs) }) + s.lock.Lock() s.ring[s.ringHead] = summ s.ringHead = (s.ringHead + 1) % len(s.ring) if s.ringLen < len(s.ring) { s.ringLen++ } + s.lock.Unlock() +} + +// summarize merges the ring summaries covering the most recent window. A +// window <= 0 (or >= the report window) covers the entire ring. +func (s *ForwardStats) summarize(window time.Duration) forwardSummary { + s.lock.Lock() + defer s.lock.Unlock() + + n := s.ringLen + if window > 0 && s.summaryInterval > 0 { + want := int((window + s.summaryInterval - 1) / s.summaryInterval) + if want < 1 { + want = 1 + } + if want < n { + n = want + } + } + + // walk backwards from the most recent entry (ringHead-1) over n entries. + var w forwardSummary + for i := 0; i < n; i++ { + idx := (s.ringHead - 1 - i + len(s.ring)) % len(s.ring) + w = w.merge(s.ring[idx]) + } + return w +} + +// GetStats returns the mean latency and jitter (std dev) of the forwarding +// transit over the most recent duration. The duration is rounded up to a whole +// number of summary intervals (the smallest bucket span that covers it). A +// duration <= 0, or one that meets/exceeds the report window, covers the full +// window. +func (s *ForwardStats) GetStats(duration time.Duration) (time.Duration, time.Duration) { + return s.summarize(duration).meanStdDev() } func (s *ForwardStats) report() { - var w forwardSummary - for i := 0; i < s.ringLen; i++ { - w = w.merge(s.ring[i]) - } + w := s.summarize(0) latency, jitter := w.meanStdDev() if dropped := s.samples.takeDropped(); dropped > 0 { diff --git a/pkg/sfu/forwardstats_test.go b/pkg/sfu/forwardstats_test.go index e51b612e1..152404978 100644 --- a/pkg/sfu/forwardstats_test.go +++ b/pkg/sfu/forwardstats_test.go @@ -278,6 +278,37 @@ func TestForwardStats_ReportWindow(t *testing.T) { require.NotPanics(t, s.report) } +func TestForwardStats_GetStats(t *testing.T) { + initPrometheus(t) + + // 5 buckets, each covering one 100ms summary interval. + s := &ForwardStats{ring: make([]forwardSummary, 5), summaryInterval: 100 * time.Millisecond} + + // fold five 100ms buckets, one sample each: 1ms, 2ms, 3ms, 4ms, 5ms. + for i := 1; i <= 5; i++ { + s.Update(0, int64(i)*int64(time.Millisecond)) + s.flush() + } + require.Equal(t, 5, s.ringLen) + + // a duration <= 0 covers the whole window: mean of 1..5ms == 3ms. + latency, jitter := s.GetStats(0) + require.InDelta(t, float64(3*time.Millisecond), float64(latency), float64(50*time.Microsecond)) + require.Greater(t, jitter, time.Duration(0)) + + // a duration meeting/exceeding the window also covers it. + fullLatency, _ := s.GetStats(time.Second) + require.InDelta(t, float64(3*time.Millisecond), float64(fullLatency), float64(50*time.Microsecond)) + + // ~200ms rounds up to the two most recent buckets (4ms, 5ms): mean == 4.5ms. + shortLatency, _ := s.GetStats(200 * time.Millisecond) + require.InDelta(t, float64(4500*time.Microsecond), float64(shortLatency), float64(50*time.Microsecond)) + + // a sub-interval duration still yields at least the most recent bucket (5ms). + lastLatency, _ := s.GetStats(time.Nanosecond) + require.InDelta(t, float64(5*time.Millisecond), float64(lastLatency), float64(50*time.Microsecond)) +} + func TestForwardStats_Lifecycle(t *testing.T) { initPrometheus(t)