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), } }