mirror of
https://github.com/livekit/livekit.git
synced 2026-07-29 11:59:49 +00:00
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) <noreply@anthropic.com>
* Trim comments in data channel bitrate calculator
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
* Make BitrateMode.String a method on the enum
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---------
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Raja Subramanian
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
cc87a83e1f
commit
cce14a6fec
@@ -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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user