From cce14a6fecaa728ebbadd609d9040e3e32c18ad6 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Thu, 16 Jul 2026 00:29:23 +0530 Subject: [PATCH] fix: don't undercount data channel bitrate over idle gaps (#4676) * fix: don't undercount data channel bitrate over idle gaps BitrateCalculator divided drained bytes by wall-clock time, including idle gaps between sparse writes. A fast channel fed sparsely (e.g. 1 KB every second, acked in a few ms) therefore reported its low offered load (~8 kbps) instead of its drain capacity (~800 kbps). In the unreliable writer that shrinks targetLatencyLimit and drops bursts on a channel that is actually keeping up. Measure the rate over backlogged ("busy") time only, so idle time no longer dilutes the estimate. Backlog is detected from the buffer occupancy just before a write (bufferedAmount - bytesWritten): since no bytes are added between writes the buffer only drains, so a non-zero pre-write level means it never emptied and the interval was genuinely busy. Gating on the post-write buffered amount would be wrong -- it always includes the just-enqueued (not yet SACKed) bytes and is ~never zero, collapsing the estimate back to wall-clock. When no backlog is ever observed the calculator reports no estimate (ok=false) instead of a confidently-low number. Introduce BitrateMode: - BusyOnly (writer default): busy denominator, all drained bytes counted. - ExcludeIdleDrain: also drops bytes drained across non-backlogged intervals for a clean drain-capacity estimate free of idle contamination. - WallClock: elapsed-time denominator, all bytes; for the test-client reader which has no send buffer to observe. Call sites keep the conservative behaviour (writers BusyOnly, reader WallClock). Tests exercise both writer modes and the mixed idle-then- backlog window where they diverge. Co-Authored-By: Claude Opus 4.8 (1M context) * Trim comments in data channel bitrate calculator Co-Authored-By: Claude Opus 4.8 (1M context) * Make BitrateMode.String a method on the enum Co-Authored-By: Claude Opus 4.8 (1M context) --------- Co-authored-by: Claude Opus 4.8 (1M context) --- pkg/sfu/datachannel/bitrate.go | 78 +++++++++- pkg/sfu/datachannel/bitrate_test.go | 138 +++++++++++++++--- pkg/sfu/datachannel/datachannel_writer.go | 4 +- .../datachannel/datachannel_writer_test.go | 7 +- test/client/datachannel_reader.go | 2 +- 5 files changed, 197 insertions(+), 32 deletions(-) diff --git a/pkg/sfu/datachannel/bitrate.go b/pkg/sfu/datachannel/bitrate.go index b95198aa8..83278557b 100644 --- a/pkg/sfu/datachannel/bitrate.go +++ b/pkg/sfu/datachannel/bitrate.go @@ -14,6 +14,36 @@ const ( BitrateWindow = 100 * time.Millisecond ) +// BitrateMode selects how the sliding-window rate is computed. +type BitrateMode int + +const ( + // BitrateModeBusyOnly divides drained bytes by backlogged ("busy") time + // only; idle gaps are excluded from the denominator. All drained bytes count. + BitrateModeBusyOnly BitrateMode = iota + + // BitrateModeExcludeIdleDrain is like BitrateModeBusyOnly but also excludes + // bytes drained across a non-backlogged interval from the numerator. + BitrateModeExcludeIdleDrain + + // BitrateModeWallClock divides all bytes by elapsed wall-clock time. For + // consumers with no send buffer, e.g. a receiver measuring its incoming rate. + BitrateModeWallClock +) + +func (m BitrateMode) String() string { + switch m { + case BitrateModeBusyOnly: + return "busyOnly" + case BitrateModeExcludeIdleDrain: + return "excludeIdleDrain" + case BitrateModeWallClock: + return "wallClock" + default: + return "unknown" + } +} + // BitrateCalculator calculates bitrate over sliding window type BitrateCalculator struct { lock sync.Mutex @@ -24,11 +54,16 @@ type BitrateCalculator struct { active bitrateWindow bytes int + busy time.Duration lastBufferedAmount int + last time.Time start time.Time + mode BitrateMode } -func NewBitrateCalculator(duration time.Duration, window time.Duration) *BitrateCalculator { +// NewBitrateCalculator creates a calculator. See BitrateMode for how mode +// governs the numerator and denominator. +func NewBitrateCalculator(duration time.Duration, window time.Duration, mode BitrateMode) *BitrateCalculator { windowCnt := int((duration + (window - 1)) / window) if windowCnt == 0 { windowCnt = 1 @@ -39,6 +74,7 @@ func NewBitrateCalculator(duration time.Duration, window time.Duration) *Bitrate windowDuration: window, start: now, active: bitrateWindow{start: now}, + mode: mode, } c.windows.SetBaseCap(windowCnt + 1) @@ -49,21 +85,44 @@ func (c *BitrateCalculator) AddBytes(bytes int, bufferedAmout int, ts time.Time) c.lock.Lock() defer c.lock.Unlock() + // bufferedAmout is sampled right after the write, so it includes the bytes + // just enqueued; subtracting them gives the buffer occupancy before the + // write. Between writes the buffer only drains, so a non-zero pre-write + // level means it stayed backlogged for the whole interval. + backlogged := bufferedAmout-bytes > 0 + var busy time.Duration + if !c.last.IsZero() && backlogged { + if busy = ts.Sub(c.last); busy < 0 { + busy = 0 + } + } + c.last = ts + + // bytes that actually drained since the previous sample bytes -= bufferedAmout - c.lastBufferedAmount if bytes < 0 { // it is possible that internal buffering (non-data like DCEP packet from webrtc) caused bytes to be negative bytes = 0 } c.lastBufferedAmount = bufferedAmout + + if c.mode == BitrateModeExcludeIdleDrain && !backlogged { + // drain across a non-backlogged interval finished at an unknown point, + // so it is not a reliable rate sample + bytes = 0 + } + if ts.Sub(c.active.start) >= c.windowDuration { c.windows.PushBack(c.active) c.active.start = ts c.active.bytes = 0 + c.active.busy = 0 for c.windows.Len() > 0 { // pop expired windows if w := c.windows.Front(); ts.Sub(w.start) > (c.duration + c.windowDuration) { c.bytes -= w.bytes + c.busy -= w.busy c.windows.PopFront() } else { c.start = w.start @@ -73,11 +132,13 @@ func (c *BitrateCalculator) AddBytes(bytes int, bufferedAmout int, ts time.Time) if c.windows.Len() == 0 { c.start = ts c.bytes = 0 + c.busy = 0 } } c.bytes += bytes c.active.bytes += bytes - + c.busy += busy + c.active.busy += busy } func (c *BitrateCalculator) Bitrate(ts time.Time) (int, bool) { @@ -91,19 +152,24 @@ func (c *BitrateCalculator) ForceBitrate(ts time.Time) (int, bool) { func (c *BitrateCalculator) bitrate(ts time.Time, force bool) (int, bool) { c.lock.Lock() defer c.lock.Unlock() - duration := ts.Sub(c.start) - if duration < c.windowDuration { + // busy modes divide by backlogged time; wall-clock mode by elapsed time + denom := c.busy + if c.mode == BitrateModeWallClock { + denom = ts.Sub(c.start) + } + if denom < c.windowDuration { if force { - duration = c.windowDuration + denom = c.windowDuration } else { return 0, false } } - return c.bytes * 8 * 1000 / int(duration.Milliseconds()), true + return c.bytes * 8 * 1000 / int(denom.Milliseconds()), true } type bitrateWindow struct { start time.Time bytes int + busy time.Duration } diff --git a/pkg/sfu/datachannel/bitrate_test.go b/pkg/sfu/datachannel/bitrate_test.go index c1414c7f0..1288dc1dc 100644 --- a/pkg/sfu/datachannel/bitrate_test.go +++ b/pkg/sfu/datachannel/bitrate_test.go @@ -7,30 +7,124 @@ import ( "github.com/stretchr/testify/require" ) -func TestBitrateCalculator(t *testing.T) { - c := NewBitrateCalculator(BitrateDuration, BitrateWindow) - require.NotNil(t, c) +var busyModes = []BitrateMode{BitrateModeBusyOnly, BitrateModeExcludeIdleDrain} + +// 1 KB written every second, each fully drained before the next write: never +// backlogged, so there is no estimate. +func TestBitrateCalculatorSparseFastAck(t *testing.T) { + for _, mode := range busyModes { + t.Run(mode.String(), func(t *testing.T) { + c := NewBitrateCalculator(BitrateDuration, BitrateWindow, mode) + require.NotNil(t, c) + + t0 := time.Now() + for i := 0; i < 5; i++ { + ts := t0.Add(time.Duration(i) * time.Second) + c.AddBytes(1000, 1000, ts) + _, ok := c.Bitrate(ts) + require.False(t, ok) + } + }) + } +} + +// Standing backlog of 1000 bytes; every 100ms write 800 and 800 drain, sampled +// buffer 1000+800=1800. 800 B / 100ms = 64000 bps. +func TestBitrateCalculatorSustainedBacklog(t *testing.T) { + for _, mode := range busyModes { + t.Run(mode.String(), func(t *testing.T) { + c := NewBitrateCalculator(BitrateDuration, BitrateWindow, mode) + + t0 := time.Now() + c.AddBytes(800, 1800, t0) + for i := 1; i <= 30; i++ { + c.AddBytes(800, 1800, t0.Add(time.Duration(i)*100*time.Millisecond)) + } + + bitrate, ok := c.Bitrate(t0.Add(3 * time.Second)) + require.True(t, ok) + require.InDelta(t, 64000, bitrate, 2000) + }) + } +} + +// Buffer only climbs and nothing drains (stalled connection): the estimate +// converges to ~0. +func TestBitrateCalculatorStalledConnection(t *testing.T) { + for _, mode := range busyModes { + t.Run(mode.String(), func(t *testing.T) { + c := NewBitrateCalculator(BitrateDuration, BitrateWindow, mode) + + t0 := time.Now() + buffered := 0 + for i := 0; i < 20; i++ { + buffered += 1000 // wrote 1000, nothing drained + c.AddBytes(1000, buffered, t0.Add(time.Duration(i)*50*time.Millisecond)) + } + + bitrate, ok := c.Bitrate(t0.Add(time.Second)) + require.True(t, ok) + require.Less(t, bitrate, 1000) + }) + } +} + +// A single write that drains to empty is not backlogged at the next sample, so +// no estimate is produced. +func TestBitrateCalculatorFlushNotCounted(t *testing.T) { + for _, mode := range busyModes { + t.Run(mode.String(), func(t *testing.T) { + c := NewBitrateCalculator(BitrateDuration, BitrateWindow, mode) + + t0 := time.Now() + c.AddBytes(100, 100, t0) + ts := t0.Add(100 * time.Millisecond) + c.AddBytes(0, 0, ts) + _, ok := c.Bitrate(ts) + require.False(t, ok) + }) + } +} + +// An idle drain followed by a backlog: BusyOnly counts the idle-drained bytes +// over busy time only and reports a higher rate than ExcludeIdleDrain. +func TestBitrateCalculatorMixedIdleThenBacklog(t *testing.T) { + off := NewBitrateCalculator(BitrateDuration, BitrateWindow, BitrateModeBusyOnly) + on := NewBitrateCalculator(BitrateDuration, BitrateWindow, BitrateModeExcludeIdleDrain) t0 := time.Now() - c.AddBytes(100, 0, t0) - // bytes buffered - c.AddBytes(100, 100, t0.Add(50*time.Millisecond)) - bitrate, ok := c.Bitrate(t0.Add(50 * time.Millisecond)) - require.Equal(t, 0, bitrate) - require.False(t, ok) - // 50 bytes sent (50 bytes buffer flushed) - c.AddBytes(100, 50, t0.Add(time.Second)) + feed := func(written, buffered int, ts time.Time) { + off.AddBytes(written, buffered, ts) + on.AddBytes(written, buffered, ts) + } - // 250 bytes sent in 1 second - bitrate, ok = c.Bitrate(t0.Add(time.Second)) - require.Equal(t, 2000, bitrate) - require.True(t, ok) + // 5000 drains during an idle gap (not backlogged), then a sustained backlog + // of 800 every 100ms over a standing 5000. + feed(5000, 5000, t0) + feed(5000, 5000, t0.Add(100*time.Millisecond)) + for i := 1; i <= 5; i++ { + feed(800, 5800, t0.Add(time.Duration(100+i*100)*time.Millisecond)) + } + queryTs := t0.Add(700 * time.Millisecond) - // silence for long time - t1 := t0.Add(2 * BitrateDuration) - // 150 bytes sent (50 bytes buffer flushed) - c.AddBytes(100, 0, t1) - bitrate, ok = c.Bitrate(t1.Add(time.Second)) - require.Equal(t, 1200, bitrate) - require.True(t, ok) + rOff, okOff := off.Bitrate(queryTs) + rOn, okOn := on.Bitrate(queryTs) + require.True(t, okOff) + require.True(t, okOn) + require.Greater(t, rOff, rOn) +} + +// No observable buffer: throughput over elapsed time. 1000 B every 100ms = +// 80000 bps. +func TestBitrateCalculatorWallClock(t *testing.T) { + c := NewBitrateCalculator(BitrateDuration, BitrateWindow, BitrateModeWallClock) + + t0 := time.Now() + for i := 0; i < 20; i++ { + c.AddBytes(1000, 0, t0.Add(time.Duration(i)*100*time.Millisecond)) + } + + bitrate, ok := c.Bitrate(t0.Add(2 * time.Second)) + require.True(t, ok) + require.InDelta(t, 80000, bitrate, 8000) } diff --git a/pkg/sfu/datachannel/datachannel_writer.go b/pkg/sfu/datachannel/datachannel_writer.go index da17cf340..2e45bb146 100644 --- a/pkg/sfu/datachannel/datachannel_writer.go +++ b/pkg/sfu/datachannel/datachannel_writer.go @@ -41,7 +41,7 @@ type DataChannelWriter[T BufferedAmountGetter] struct { func NewDataChannelWriterReliable[T BufferedAmountGetter](bufferGetter T, rawDC datachannel.ReadWriteCloserDeadliner, slowThreshold int) *DataChannelWriter[T] { var rate *BitrateCalculator if slowThreshold > 0 { - rate = NewBitrateCalculator(BitrateDuration, BitrateWindow) + rate = NewBitrateCalculator(BitrateDuration, BitrateWindow, BitrateModeBusyOnly) } return &DataChannelWriter[T]{ bufferGetter: bufferGetter, @@ -59,7 +59,7 @@ func NewDataChannelWriterReliable[T BufferedAmountGetter](bufferGetter T, rawDC func NewDataChannelWriterUnreliable[T BufferedAmountGetter](bufferGetter T, rawDC datachannel.ReadWriteCloserDeadliner, targetLatency time.Duration, minBufferedAmount uint64) *DataChannelWriter[T] { var rate *BitrateCalculator if targetLatency > 0 { - rate = NewBitrateCalculator(BitrateDuration, BitrateWindow) + rate = NewBitrateCalculator(BitrateDuration, BitrateWindow, BitrateModeBusyOnly) } return &DataChannelWriter[T]{ bufferGetter: bufferGetter, diff --git a/pkg/sfu/datachannel/datachannel_writer_test.go b/pkg/sfu/datachannel/datachannel_writer_test.go index 3bdcefbf4..2ef20f6f3 100644 --- a/pkg/sfu/datachannel/datachannel_writer_test.go +++ b/pkg/sfu/datachannel/datachannel_writer_test.go @@ -79,6 +79,7 @@ type mockDataChannelWriter struct { datachannel.ReadWriteCloserDeadliner nextWriteCompleteAt time.Time deadline *deadline.Deadline + buffered atomic.Int64 } func newMockDataChannelWriter() *mockDataChannelWriter { @@ -88,18 +89,22 @@ func newMockDataChannelWriter() *mockDataChannelWriter { } func (m *mockDataChannelWriter) BufferedAmount() uint64 { - return 0 + return uint64(m.buffered.Load()) } func (m *mockDataChannelWriter) Write(b []byte) (int, error) { + // buffered while the write is in flight, cleared once it completes + m.buffered.Store(int64(len(b))) wait := time.Until(m.nextWriteCompleteAt) if wait <= 0 { + m.buffered.Store(0) return len(b), nil } select { case <-m.deadline.Done(): return 0, m.deadline.Err() case <-time.After(wait): + m.buffered.Store(0) return len(b), nil } } diff --git a/test/client/datachannel_reader.go b/test/client/datachannel_reader.go index eb3b64200..ff2b1e866 100644 --- a/test/client/datachannel_reader.go +++ b/test/client/datachannel_reader.go @@ -14,7 +14,7 @@ type DataChannelReader struct { func NewDataChannelReader(bitrate int) *DataChannelReader { return &DataChannelReader{ target: bitrate, - bitrate: datachannel.NewBitrateCalculator(datachannel.BitrateDuration*5, datachannel.BitrateWindow), + bitrate: datachannel.NewBitrateCalculator(datachannel.BitrateDuration*5, datachannel.BitrateWindow, datachannel.BitrateModeWallClock), } }