From b2a54729f50976227091e7b7ebacb24a6ab3e5f9 Mon Sep 17 00:00:00 2001 From: cnderrauber Date: Tue, 18 Feb 2025 14:41:41 +0800 Subject: [PATCH] Don't drop message if calculate duration is too small (#3442) * Don't drop message if calculate duration is too small * fix test --- go.mod | 2 +- go.sum | 4 ++-- pkg/sfu/datachannel/bitrate.go | 10 +++++++--- pkg/sfu/datachannel/bitrate_test.go | 11 +++++++++-- pkg/sfu/datachannel/datachannel_writer.go | 2 +- test/client/datachannel_reader.go | 5 +++-- 6 files changed, 23 insertions(+), 11 deletions(-) diff --git a/go.mod b/go.mod index 21fb911e9..2e44b8c1a 100644 --- a/go.mod +++ b/go.mod @@ -36,7 +36,7 @@ require ( github.com/pion/interceptor v0.1.37 github.com/pion/rtcp v1.2.15 github.com/pion/rtp v1.8.11 - github.com/pion/sctp v1.8.35 + github.com/pion/sctp v1.8.36 github.com/pion/sdp/v3 v3.0.10 github.com/pion/transport/v3 v3.0.7 github.com/pion/turn/v4 v4.0.0 diff --git a/go.sum b/go.sum index e0df8b6bd..416ca99fb 100644 --- a/go.sum +++ b/go.sum @@ -251,8 +251,8 @@ github.com/pion/rtcp v1.2.15 h1:LZQi2JbdipLOj4eBjK4wlVoQWfrZbh3Q6eHtWtJBZBo= github.com/pion/rtcp v1.2.15/go.mod h1:jlGuAjHMEXwMUHK78RgX0UmEJFV4zUKOFHR7OP+D3D0= github.com/pion/rtp v1.8.11 h1:17xjnY5WO5hgO6SD3/NTIUPvSFw/PbLsIJyz1r1yNIk= github.com/pion/rtp v1.8.11/go.mod h1:8uMBJj32Pa1wwx8Fuv/AsFhn8jsgw+3rUC2PfoBZ8p4= -github.com/pion/sctp v1.8.35 h1:qwtKvNK1Wc5tHMIYgTDJhfZk7vATGVHhXbUDfHbYwzA= -github.com/pion/sctp v1.8.35/go.mod h1:EcXP8zCYVTRy3W9xtOF7wJm1L1aXfKRQzaM33SjQlzg= +github.com/pion/sctp v1.8.36 h1:owNudmnz1xmhfYje5L/FCav3V9wpPRePHle3Zi+P+M0= +github.com/pion/sctp v1.8.36/go.mod h1:cNiLdchXra8fHQwmIoqw0MbLLMs+f7uQ+dGMG2gWebE= github.com/pion/sdp/v3 v3.0.10 h1:6MChLE/1xYB+CjumMw+gZ9ufp2DPApuVSnDT8t5MIgA= github.com/pion/sdp/v3 v3.0.10/go.mod h1:88GMahN5xnScv1hIMTqLdu/cOcUkj6a9ytbncwMCq2E= github.com/pion/srtp/v3 v3.0.4 h1:2Z6vDVxzrX3UHEgrUyIGM4rRouoC7v+NiF1IHtp9B5M= diff --git a/pkg/sfu/datachannel/bitrate.go b/pkg/sfu/datachannel/bitrate.go index 74e30d3b9..ac538033d 100644 --- a/pkg/sfu/datachannel/bitrate.go +++ b/pkg/sfu/datachannel/bitrate.go @@ -50,6 +50,10 @@ func (c *BitrateCalculator) AddBytes(bytes int, bufferedAmout int, ts time.Time) defer c.lock.Unlock() 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 ts.Sub(c.active.start) >= c.windowDuration { c.windows.PushBack(c.active) @@ -76,15 +80,15 @@ func (c *BitrateCalculator) AddBytes(bytes int, bufferedAmout int, ts time.Time) } -func (c *BitrateCalculator) Bitrate(ts time.Time) int { +func (c *BitrateCalculator) Bitrate(ts time.Time) (int, bool) { c.lock.Lock() defer c.lock.Unlock() duration := ts.Sub(c.start) if duration < c.windowDuration { - duration = c.windowDuration + return 0, false } - return c.bytes * 8 * 1000 / int(duration.Milliseconds()) + return c.bytes * 8 * 1000 / int(duration.Milliseconds()), true } type bitrateWindow struct { diff --git a/pkg/sfu/datachannel/bitrate_test.go b/pkg/sfu/datachannel/bitrate_test.go index 4b4e0a1a6..c1414c7f0 100644 --- a/pkg/sfu/datachannel/bitrate_test.go +++ b/pkg/sfu/datachannel/bitrate_test.go @@ -15,15 +15,22 @@ func TestBitrateCalculator(t *testing.T) { 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)) // 250 bytes sent in 1 second - require.Equal(t, 2000, c.Bitrate(t0.Add(time.Second))) + bitrate, ok = c.Bitrate(t0.Add(time.Second)) + require.Equal(t, 2000, bitrate) + require.True(t, ok) // silence for long time t1 := t0.Add(2 * BitrateDuration) // 150 bytes sent (50 bytes buffer flushed) c.AddBytes(100, 0, t1) - require.Equal(t, 1200, c.Bitrate(t1.Add(time.Second))) + bitrate, ok = c.Bitrate(t1.Add(time.Second)) + require.Equal(t, 1200, bitrate) + require.True(t, ok) } diff --git a/pkg/sfu/datachannel/datachannel_writer.go b/pkg/sfu/datachannel/datachannel_writer.go index 9e9ee6767..707c92035 100644 --- a/pkg/sfu/datachannel/datachannel_writer.go +++ b/pkg/sfu/datachannel/datachannel_writer.go @@ -64,7 +64,7 @@ func (w *DataChannelWriter[T]) Write(p []byte) (n int, err error) { w.rate.AddBytes(n, int(w.bufferGetter.BufferedAmount()), now) // retry if the write timed out on a non-slow receiver if errors.Is(err, context.DeadlineExceeded) { - if bitrate := w.rate.Bitrate(now); bitrate >= w.slowThreshold { + if bitrate, ok := w.rate.Bitrate(now); !ok || bitrate >= w.slowThreshold { continue } else { err = fmt.Errorf("%w: bitrate %d, threshold %d", ErrDataDroppedBySlowReader, bitrate, w.slowThreshold) diff --git a/test/client/datachannel_reader.go b/test/client/datachannel_reader.go index 2b8c72102..f13d6234f 100644 --- a/test/client/datachannel_reader.go +++ b/test/client/datachannel_reader.go @@ -20,7 +20,7 @@ func NewDataChannelReader(bitrate int) *DataChannelReader { func (d *DataChannelReader) Read(p []byte, sid string) { for { - if bitrate := d.bitrate.Bitrate(time.Now()); bitrate > 0 && bitrate > d.target { + if bitrate, ok := d.bitrate.Bitrate(time.Now()); ok && bitrate > 0 && bitrate > d.target { time.Sleep(10 * time.Millisecond) d.bitrate.AddBytes(0, 0, time.Now()) continue @@ -31,5 +31,6 @@ func (d *DataChannelReader) Read(p []byte, sid string) { } func (d *DataChannelReader) Bitrate() int { - return d.bitrate.Bitrate(time.Now()) + b, _ := d.bitrate.Bitrate(time.Now()) + return b }