diff --git a/pkg/sfu/forwardstats.go b/pkg/sfu/forwardstats.go index dd41d0bb0..34bb188a4 100644 --- a/pkg/sfu/forwardstats.go +++ b/pkg/sfu/forwardstats.go @@ -1,13 +1,12 @@ package sfu import ( - "sync" + "math" "time" "github.com/livekit/livekit-server/pkg/telemetry/prometheus" "github.com/livekit/protocol/logger" - "github.com/livekit/protocol/utils" - "github.com/livekit/protocol/utils/mono" + "go.uber.org/atomic" ) const ( @@ -15,97 +14,256 @@ const ( cSkewFactor = 10 ) -type ForwardStats struct { - lock sync.Mutex - latency *utils.LatencyAggregate - lowest int64 - highest int64 - lastUpdateAt int64 - closeCh chan struct{} +const ( + // A summary interval's worth of samples across all tracks must fit without + // dropping (ForwardStats is a singleton). Shard count spreads the per-packet + // atomic; shard capacity bounds memory (numShards*shardCap*16 bytes = 2MiB). + forwardSampleNumShards = 16 + forwardSampleShardCap = 8192 + forwardSampleShardMask = forwardSampleShardCap - 1 + forwardSampleShardSel = forwardSampleNumShards - 1 +) + +// forwardSampleShard is a ring of transit samples with multiple producers and a +// single consumer. A producer reserves a slot, stores the value, then publishes +// the slot's epoch (reserved index + 1). The consumer reads a slot only once its +// epoch marks the value committed for that index. +type forwardSampleShard struct { + writeIdx atomic.Uint64 // advanced by producers to reserve a slot + readIdx uint64 // consumer-only cursor + ring [forwardSampleShardCap]atomic.Int64 + seq [forwardSampleShardCap]atomic.Uint64 // per-slot publish epoch } -func NewForwardStats(latencyUpdateInterval, reportInterval, latencyWindowLength time.Duration) *ForwardStats { - s := &ForwardStats{ - latency: utils.NewLatencyAggregate(latencyUpdateInterval, latencyWindowLength), - lowest: time.Second.Nanoseconds(), - closeCh: make(chan struct{}), +// forwardSampleBuffer holds per-packet transit samples produced on the packet +// path and consumed by the background worker, which performs metric emission. +type forwardSampleBuffer struct { + shards [forwardSampleNumShards]forwardSampleShard + dropped atomic.Uint64 +} + +// push records a sample: reserve a slot, store the value, then publish the +// slot's epoch. The shard is selected from arrival time bits. +func (b *forwardSampleBuffer) push(arrival, transitNs int64) { + sh := &b.shards[(uint64(arrival)>>6)&forwardSampleShardSel] + i := sh.writeIdx.Add(1) - 1 + slot := i & forwardSampleShardMask + sh.ring[slot].Store(transitNs) + sh.seq[slot].Store(i + 1) +} + +// drain passes every committed sample to fn and advances the read cursor. Only +// the background worker calls this. +// +// A slot holds index r's value once its epoch equals r+1. If the slot at the +// cursor is still uncommitted (a producer reserved it but has not published), +// draining stops and resumes from there on the next call, so no sample is read +// stale or skipped. When producers get a shard's capacity ahead, or overwrite a +// slot before it is read, the affected samples are counted as dropped. +func (b *forwardSampleBuffer) drain(fn func(transitNs int64)) { + for si := range b.shards { + sh := &b.shards[si] + w := sh.writeIdx.Load() + r := sh.readIdx + if w-r > forwardSampleShardCap { + b.dropped.Add(w - r - forwardSampleShardCap) + r = w - forwardSampleShardCap + } + for r < w { + slot := r & forwardSampleShardMask + if sh.seq[slot].Load() < r+1 { + // reserved but not yet published; resume here next drain + break + } + v := sh.ring[slot].Load() + if sh.seq[slot].Load() != r+1 { + // overwritten by a newer sample during the read; original lost + b.dropped.Add(1) + r++ + continue + } + fn(v) + r++ + } + sh.readIdx = r + } +} + +func (b *forwardSampleBuffer) takeDropped() uint64 { + return b.dropped.Swap(0) +} + +// forwardSummary is a mergeable summary of forwarding transit over an interval. +// The sum of squares is kept in microseconds so it does not overflow int64. +type forwardSummary struct { + count int64 + sumUs int64 + sumSqUs int64 + minNs int64 + maxNs int64 +} + +func (s forwardSummary) addSample(transitNs int64) forwardSummary { + us := transitNs / 1000 + if s.count == 0 { + return forwardSummary{count: 1, sumUs: us, sumSqUs: us * us, minNs: transitNs, maxNs: transitNs} } - go s.report(reportInterval) + s.count++ + s.sumUs += us + s.sumSqUs += us * us + if transitNs < s.minNs { + s.minNs = transitNs + } + if transitNs > s.maxNs { + s.maxNs = transitNs + } return s } +func (s forwardSummary) merge(o forwardSummary) forwardSummary { + if o.count == 0 { + return s + } + if s.count == 0 { + return o + } + return forwardSummary{ + count: s.count + o.count, + sumUs: s.sumUs + o.sumUs, + sumSqUs: s.sumSqUs + o.sumSqUs, + minNs: min(s.minNs, o.minNs), + maxNs: max(s.maxNs, o.maxNs), + } +} + +func (s forwardSummary) meanStdDev() (mean, stdDev time.Duration) { + if s.count == 0 { + return 0, 0 + } + + meanUs := float64(s.sumUs) / float64(s.count) + mean = time.Duration(meanUs * float64(time.Microsecond)) + if s.count < 2 { + return mean, 0 + } + + // sample variance (divisor count-1) + m2 := float64(s.sumSqUs) - float64(s.sumUs)*meanUs + varUs2 := m2 / float64(s.count-1) + if varUs2 < 0 { + // floating point rounding can push a (near-zero) variance slightly negative + varUs2 = 0 + } + stdDev = time.Duration(math.Sqrt(varUs2) * float64(time.Microsecond)) + return mean, stdDev +} + +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. + ring []forwardSummary + ringHead int + ringLen int + + summaryInterval time.Duration + reportInterval time.Duration + + closeCh chan struct{} +} + +func NewForwardStats(summaryInterval, reportInterval, reportWindow time.Duration) *ForwardStats { + ringCap := int((reportWindow + summaryInterval - 1) / summaryInterval) + if ringCap < 1 { + ringCap = 1 + } + + s := &ForwardStats{ + ring: make([]forwardSummary, ringCap), + summaryInterval: summaryInterval, + reportInterval: reportInterval, + closeCh: make(chan struct{}), + } + + go s.run() + return s +} + +// Update records a forwarded packet's transit latency. It buffers the sample +// and returns the transit and whether it exceeds the high-latency threshold. +// The sample is aggregated and emitted by the background worker. func (s *ForwardStats) Update(arrival, left int64) (int64, bool) { transit := left - arrival - isHighForwardingLatency := time.Duration(transit) > cHighForwardingLatency - - s.lock.Lock() - s.latency.Update(time.Duration(arrival), float64(transit)) - s.lowest = min(transit, s.lowest) - s.highest = max(transit, s.highest) - s.lastUpdateAt = arrival - s.lock.Unlock() - - prometheus.RecordForwardLatencySample(transit) - return transit, isHighForwardingLatency -} - -func (s *ForwardStats) GetStats(shortDuration time.Duration) (time.Duration, time.Duration) { - s.lock.Lock() - // a dummy sample to flush the pipe to current time - now := mono.UnixNano() - if (now - s.lastUpdateAt) > shortDuration.Nanoseconds() { - s.latency.Update(time.Duration(now), 0) - } - - wLong := s.latency.Summarize() - - lowest := s.lowest - s.lowest = time.Second.Nanoseconds() - - highest := s.highest - s.highest = 0 - s.lock.Unlock() - - latencyLong, jitterLong := time.Duration(wLong.Mean()), time.Duration(wLong.StdDev()) - if jitterLong > latencyLong*cSkewFactor { - logger.Infow( - "high jitter in forwarding path", - "lowest", time.Duration(lowest), - "highest", time.Duration(highest), - "countLong", wLong.Count(), - "latencyLong", latencyLong, - "jitterLong", jitterLong, - ) - } - return latencyLong, jitterLong -} - -func (s *ForwardStats) GetShortStats(shortDuration time.Duration) (time.Duration, time.Duration) { - s.lock.Lock() - wShort := s.latency.SummarizeLast(shortDuration) - s.lock.Unlock() - - return time.Duration(wShort.Mean()), time.Duration(wShort.StdDev()) + s.samples.push(arrival, transit) + return transit, time.Duration(transit) > cHighForwardingLatency } func (s *ForwardStats) Stop() { close(s.closeCh) } -func (s *ForwardStats) report(reportInterval time.Duration) { - ticker := time.NewTicker(reportInterval) - defer ticker.Stop() +func (s *ForwardStats) run() { + summaryTicker := time.NewTicker(s.summaryInterval) + defer summaryTicker.Stop() + reportTicker := time.NewTicker(s.reportInterval) + defer reportTicker.Stop() for { select { case <-s.closeCh: return - case <-ticker.C: - latencyLong, jitterLong := s.GetStats(reportInterval) - prometheus.RecordForwardJitter(uint32(jitterLong.Nanoseconds())) - prometheus.RecordForwardLatency(uint32(latencyLong.Nanoseconds())) + case <-summaryTicker.C: + s.flush() + + case <-reportTicker.C: + // the summary ticker keeps the window ring current to within one + // summary interval; report over it without advancing the ring. + s.report() } } } + +// flush drains the buffered samples, observes each into the Prometheus +// histogram, and folds the interval summary into the window ring used for the +// latency/jitter gauges. +func (s *ForwardStats) flush() { + var summ forwardSummary + s.samples.drain(func(transitNs int64) { + prometheus.RecordForwardLatencySample(transitNs) + summ = summ.addSample(transitNs) + }) + + s.ring[s.ringHead] = summ + s.ringHead = (s.ringHead + 1) % len(s.ring) + if s.ringLen < len(s.ring) { + s.ringLen++ + } +} + +func (s *ForwardStats) report() { + var w forwardSummary + for i := 0; i < s.ringLen; i++ { + w = w.merge(s.ring[i]) + } + + latency, jitter := w.meanStdDev() + if dropped := s.samples.takeDropped(); dropped > 0 { + logger.Warnw("forward stats sample buffer overflow", nil, "dropped", dropped) + } + if w.count > 0 && jitter > latency*cSkewFactor { + logger.Infow( + "high jitter in forwarding path", + "lowest", time.Duration(w.minNs), + "highest", time.Duration(w.maxNs), + "count", w.count, + "latency", latency, + "jitter", jitter, + ) + } + + prometheus.RecordForwardJitter(uint32(jitter.Nanoseconds())) + prometheus.RecordForwardLatency(uint32(latency.Nanoseconds())) +} diff --git a/pkg/sfu/forwardstats_test.go b/pkg/sfu/forwardstats_test.go new file mode 100644 index 000000000..e51b612e1 --- /dev/null +++ b/pkg/sfu/forwardstats_test.go @@ -0,0 +1,334 @@ +package sfu + +import ( + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" + "go.uber.org/atomic" + + "github.com/livekit/livekit-server/pkg/telemetry/prometheus" + "github.com/livekit/protocol/livekit" +) + +// initPrometheus initializes the global forward-latency collectors so that the +// worker's metric emission has non-nil targets. Init returns early if already +// initialized, so it is safe to call from multiple tests. +func initPrometheus(t *testing.T) { + t.Helper() + require.NoError(t, prometheus.Init("test", livekit.NodeType_SERVER)) +} + +// --------------------------------------------------------------------------- +// forwardSummary +// --------------------------------------------------------------------------- + +func TestForwardSummary_AddSample(t *testing.T) { + var s forwardSummary + + // empty summary + require.Equal(t, int64(0), s.count) + + // microsecond-aligned transits so the /1000 truncation is exact + s = s.addSample(3000) // 3us + s = s.addSample(1000) // 1us + s = s.addSample(2000) // 2us + + require.Equal(t, int64(3), s.count) + require.Equal(t, int64(1+2+3), s.sumUs) + require.Equal(t, int64(1+4+9), s.sumSqUs) + require.Equal(t, int64(1000), s.minNs) + require.Equal(t, int64(3000), s.maxNs) +} + +func TestForwardSummary_Merge(t *testing.T) { + var empty forwardSummary + a := forwardSummary{}.addSample(1000).addSample(2000) + b := forwardSummary{}.addSample(5000).addSample(3000) + + // merging with empty is identity, in both directions + require.Equal(t, a, a.merge(empty)) + require.Equal(t, a, empty.merge(a)) + + m := a.merge(b) + require.Equal(t, int64(4), m.count) + require.Equal(t, a.sumUs+b.sumUs, m.sumUs) + require.Equal(t, a.sumSqUs+b.sumSqUs, m.sumSqUs) + require.Equal(t, int64(1000), m.minNs) + require.Equal(t, int64(5000), m.maxNs) +} + +func TestForwardSummary_MeanStdDev(t *testing.T) { + // empty -> zero + mean, stdDev := forwardSummary{}.meanStdDev() + require.Zero(t, mean) + require.Zero(t, stdDev) + + // single sample -> mean set, stddev zero (needs >= 2 for variance) + mean, stdDev = forwardSummary{}.addSample(4000).meanStdDev() + require.Equal(t, 4*time.Microsecond, mean) + require.Zero(t, stdDev) + + // identical samples -> zero variance + s := forwardSummary{}.addSample(2000).addSample(2000).addSample(2000) + mean, stdDev = s.meanStdDev() + require.Equal(t, 2*time.Microsecond, mean) + require.Zero(t, stdDev) + + // known dataset [1us, 2us, 3us]: mean 2us, sample variance 1us^2 -> stddev 1us + s = forwardSummary{}.addSample(1000).addSample(2000).addSample(3000) + mean, stdDev = s.meanStdDev() + require.Equal(t, 2*time.Microsecond, mean) + require.InDelta(t, float64(time.Microsecond), float64(stdDev), float64(50*time.Nanosecond)) +} + +// --------------------------------------------------------------------------- +// forwardSampleBuffer +// --------------------------------------------------------------------------- + +func shardOf(arrival int64) int { + return int((uint64(arrival) >> 6) & forwardSampleShardSel) +} + +// arrivalForShard returns the n-th arrival value that maps to a fixed shard. +// Incrementing arrival by (1<<10) advances (arrival>>6) by 16, leaving the low +// 4 selection bits unchanged. +func arrivalForShard(n int) int64 { + return int64(n) << 10 +} + +func TestForwardSampleBuffer_PushDrain(t *testing.T) { + var b forwardSampleBuffer + + const n = 1000 + for i := 0; i < n; i++ { + b.push(int64(i), int64((i+1)*1000)) + } + + got := map[int64]int{} + total := 0 + b.drain(func(v int64) { + got[v]++ + total++ + }) + require.Equal(t, n, total) + require.Equal(t, uint64(0), b.dropped.Load()) + for i := 0; i < n; i++ { + require.Equal(t, 1, got[int64((i+1)*1000)], "sample %d missing", i) + } + + // draining again yields nothing (read cursor advanced) + total = 0 + b.drain(func(v int64) { total++ }) + require.Equal(t, 0, total) +} + +func TestForwardSampleBuffer_Overflow(t *testing.T) { + var b forwardSampleBuffer + + const extra = 100 + const n = forwardSampleShardCap + extra + + // pin every push to a single shard so it overflows + for i := 0; i < n; i++ { + b.push(arrivalForShard(i), int64(i)*1000) + } + require.Equal(t, 0, shardOf(arrivalForShard(0))) + require.Equal(t, shardOf(arrivalForShard(0)), shardOf(arrivalForShard(n-1))) + + var drained []int64 + b.drain(func(v int64) { drained = append(drained, v) }) + + // exactly a shard's worth survives; the oldest `extra` are dropped and counted + require.Len(t, drained, forwardSampleShardCap) + require.Equal(t, uint64(extra), b.dropped.Load()) + + // survivors are the most recent cap samples, in order + for j, v := range drained { + require.Equal(t, int64(extra+j)*1000, v) + } +} + +func TestForwardSampleBuffer_DefersUncommitted(t *testing.T) { + var b forwardSampleBuffer + sh := &b.shards[0] + + // simulate a producer that reserved index 0 but has not published its value + sh.writeIdx.Store(1) + + got := 0 + b.drain(func(int64) { got++ }) + require.Equal(t, 0, got, "uncommitted slot must not be read") + require.Equal(t, uint64(0), sh.readIdx, "cursor must not advance past an uncommitted slot") + require.Equal(t, uint64(0), b.dropped.Load()) + + // producer publishes the value; next drain picks it up + sh.ring[0].Store(1234) + sh.seq[0].Store(1) + + var vals []int64 + b.drain(func(v int64) { vals = append(vals, v) }) + require.Equal(t, []int64{1234}, vals) + require.Equal(t, uint64(1), sh.readIdx) + require.Equal(t, uint64(0), b.dropped.Load()) +} + +func TestForwardSampleBuffer_Concurrent(t *testing.T) { + var b forwardSampleBuffer + var stop atomic.Bool + + var consumed int64 + done := make(chan struct{}) + go func() { + defer close(done) + for !stop.Load() { + b.drain(func(int64) { consumed++ }) + time.Sleep(time.Millisecond) + } + b.drain(func(int64) { consumed++ }) // final sweep + }() + + const producers = 8 + const perProducer = 100_000 + var wg sync.WaitGroup + for p := 0; p < producers; p++ { + wg.Add(1) + go func(seed int64) { + defer wg.Done() + for i := int64(0); i < perProducer; i++ { + b.push(seed*7+i, (i%50)*int64(time.Microsecond)) + } + }(int64(p)) + } + wg.Wait() + stop.Store(true) + <-done + + // with a consumer keeping pace no samples should be lost + require.Equal(t, int64(producers*perProducer), consumed+int64(b.dropped.Load())) +} + +// --------------------------------------------------------------------------- +// ForwardStats +// --------------------------------------------------------------------------- + +func TestForwardStats_Update(t *testing.T) { + s := &ForwardStats{ring: make([]forwardSummary, 1)} + + // below threshold + transit, isHigh := s.Update(1000, 1000+int64(5*time.Millisecond)) + require.Equal(t, int64(5*time.Millisecond), transit) + require.False(t, isHigh) + + // above threshold + transit, isHigh = s.Update(1000, 1000+int64(25*time.Millisecond)) + require.Equal(t, int64(25*time.Millisecond), transit) + require.True(t, isHigh) + + // exactly at threshold is not "high" (strictly greater) + _, isHigh = s.Update(0, int64(cHighForwardingLatency)) + require.False(t, isHigh) +} + +func TestForwardStats_Flush(t *testing.T) { + initPrometheus(t) + + s := &ForwardStats{ring: make([]forwardSummary, 4)} + + for i := 0; i < 10; i++ { + s.Update(0, int64((i+1)*1000)) // 1us..10us + } + + s.flush() + + require.Equal(t, 1, s.ringLen) + summ := s.ring[0] + require.Equal(t, int64(10), summ.count) + require.Equal(t, int64(1000), summ.minNs) + require.Equal(t, int64(10000), summ.maxNs) + require.Equal(t, uint64(0), s.samples.dropped.Load()) + + // a subsequent flush with no new samples appends an empty summary + s.flush() + require.Equal(t, 2, s.ringLen) + require.Equal(t, int64(0), s.ring[1].count) +} + +func TestForwardStats_ReportWindow(t *testing.T) { + initPrometheus(t) + + // window of 3 summary buckets + s := &ForwardStats{ring: make([]forwardSummary, 3)} + + s.Update(0, 1000) + s.flush() + s.Update(0, 3000) + s.flush() + + // report merges the whole window without panicking and reflects both samples + var w forwardSummary + for i := 0; i < s.ringLen; i++ { + w = w.merge(s.ring[i]) + } + require.Equal(t, int64(2), w.count) + require.Equal(t, int64(1000), w.minNs) + require.Equal(t, int64(3000), w.maxNs) + + require.NotPanics(t, s.report) +} + +func TestForwardStats_Lifecycle(t *testing.T) { + initPrometheus(t) + + s := NewForwardStats(5*time.Millisecond, 20*time.Millisecond, 100*time.Millisecond) + for i := 0; i < 1000; i++ { + s.Update(int64(i), int64(i)+int64(time.Millisecond)) + } + time.Sleep(60 * time.Millisecond) // let the worker flush/report a few times + require.NotPanics(t, s.Stop) +} + +func TestNewForwardStats_RingSizing(t *testing.T) { + // ringCap = ceil(window / summaryInterval) + s := NewForwardStats(100*time.Millisecond, time.Second, time.Second) + require.Equal(t, 10, len(s.ring)) + s.Stop() + + // rounds up a partial interval + s = NewForwardStats(100*time.Millisecond, time.Second, 250*time.Millisecond) + require.Equal(t, 3, len(s.ring)) + s.Stop() + + // never smaller than one bucket, even if window < summaryInterval + s = NewForwardStats(time.Second, time.Second, 100*time.Millisecond) + require.Equal(t, 1, len(s.ring)) + s.Stop() +} + +// --------------------------------------------------------------------------- +// benchmark: per-packet cost of Update (run with -cpu 1,8). +// --------------------------------------------------------------------------- + +// benchArrival advances the arrival timestamp by 64ns per packet so that +// consecutive packets from one goroutine map to successive shards +// ((arrival>>6)&mask increments each step). A distinct per-goroutine base +// spreads goroutines across shards. +func benchArrival(base, i int64) int64 { + return base + i*64 +} + +func BenchmarkForwardStatsUpdate(b *testing.B) { + s := &ForwardStats{ring: make([]forwardSummary, 1)} + + var gid atomic.Int64 + b.RunParallel(func(pb *testing.PB) { + base := gid.Add(1) * 1_000_003 + var i int64 + for pb.Next() { + i++ + arrival := benchArrival(base, i) + s.Update(arrival, arrival+int64(2*time.Millisecond)) + } + }) +}