From 345bc5eeb0aeeb0d9d3fe57a6786bf34a0a7aa37 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Sun, 12 Jul 2026 12:46:46 +0530 Subject: [PATCH] Move ForwardStats aggregation off the packet forwarding path (#4660) * Move ForwardStats aggregation off the packet forwarding path ForwardStats is a process-wide singleton and its Update runs for every forwarded packet. It previously took a shared mutex, updated a windowed aggregate, and observed into a global Prometheus histogram on every call. Update now only buffers the transit sample into a sharded lock-free ring (one atomic add to reserve a slot, one atomic store to publish). A background worker drains the ring every summary interval, observes each sample into the histogram (per-packet fidelity retained) and folds the interval summary into a window ring for the latency/jitter gauges. Under sustained overload the oldest excess samples are dropped and counted, and the count is logged. Ring slots are atomic.Int64 so producer store / consumer load are synchronized (race-clean). Capacity is numShards*shardCap = 131072 samples; at the default 50ms summary interval that sustains ~2.6M samples/s before dropping. Benchmark: BenchmarkForwardStatsUpdate (0 allocs/op both), Update per call: cores before after speedup 1 ~8.6 ns ~1.5 ns ~6x 8 ~175 ns ~22 ns ~8x The before path (mutex + windowed Welford aggregate + per-packet histogram observe) degrades ~20x from 1 to 8 cores under contention; the after path does not block and stays an order of magnitude lower under load. Co-Authored-By: Claude Opus 4.8 (1M context) * Fix ring publish race and report-path double-flush Addresses two review findings on the forward-stats sample buffer. 1. Publish race (reserve-before-store): push reserved a ring slot by advancing writeIdx and stored the value afterwards, so drain could read a slot the producer had reserved but not yet written, getting a stale/zero value; the producer's later store then landed behind the read cursor and was lost. Each slot now carries a publish epoch. push stores the value and then publishes seq = index+1. drain reads a slot only when seq == r+1; a reserved-but-unpublished slot at the cursor stops the drain and is picked up on the next call, so no sample is read stale or lost. A slot overwritten during the read is detected on re-check and counted as dropped. The windowed latency/jitter stats did not permanently drift even before this change (report recomputes from a fixed-size ring that overwrites, not discounts, old buckets, so any error aged out within the report window), but these values feed the capacity manager, so the buffer is made exact. 2. Report-path double-flush: run() flushed on both the summary tick and the report tick, and every flush advances the window ring, so the ring cycled faster than intended and the effective window was shorter than configured (~5% at 50ms/1s/1m). The report tick no longer flushes; the summary ticker keeps the ring current to within one summary interval. Benchmark, Update per call (0 allocs/op), 1 and 8 cores: after fixes: ~5.5 ns / ~29 ns before fixes (this branch): ~1.5 ns / ~22 ns baseline (mutex + per-packet histogram): ~8.6 ns / ~175 ns The publish epoch adds one atomic store to push; the path stays non-blocking, zero-allocation, and well below the baseline under contention. Co-Authored-By: Claude Opus 4.8 (1M context) --------- Co-authored-by: Claude Opus 4.8 (1M context) --- pkg/sfu/forwardstats.go | 304 +++++++++++++++++++++++-------- pkg/sfu/forwardstats_test.go | 334 +++++++++++++++++++++++++++++++++++ 2 files changed, 565 insertions(+), 73 deletions(-) create mode 100644 pkg/sfu/forwardstats_test.go 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)) + } + }) +}