From 57d89b5adb7e31d869e70102a6f1254a7b67a41c Mon Sep 17 00:00:00 2001 From: Mathis Engelbart Date: Mon, 7 Sep 2026 17:30:41 +0200 Subject: [PATCH] Buffer packets when no RTCP writer is set --- pkg/rfc8888/interceptor.go | 25 ++++++++++++++-- pkg/rfc8888/interceptor_test.go | 38 +++++++++++++++++++++++++ pkg/twcc/sender_interceptor.go | 25 ++++++++++++++-- pkg/twcc/sender_interceptor_test.go | 44 +++++++++++++++++++++++++++++ 4 files changed, 126 insertions(+), 6 deletions(-) diff --git a/pkg/rfc8888/interceptor.go b/pkg/rfc8888/interceptor.go index 83149ac3..d5db74cd 100644 --- a/pkg/rfc8888/interceptor.go +++ b/pkg/rfc8888/interceptor.go @@ -15,6 +15,10 @@ import ( "github.com/pion/rtcp" ) +// packetChanBufferSize is the number of packets buffered for the report loop +// before new packets are dropped. +const packetChanBufferSize = 1000 + var errClosed = errors.New("interceptor is closed") // TickerFactory is a factory to create new tickers. @@ -34,7 +38,7 @@ func (s *SenderInterceptorFactory) NewInterceptor(_ string) (interceptor.Interce recorder: NewRecorder(), interval: 100 * time.Millisecond, maxReportSize: 1200, - packetChan: make(chan packet), + packetChan: make(chan packet, packetChanBufferSize), newTicker: func(d time.Duration) ticker { return &timeTicker{time.NewTicker(d)} }, @@ -128,10 +132,13 @@ func (s *SenderInterceptor) BindRemoteStream( sequenceNumber: header.SequenceNumber, ecn: 0, // ECN is not supported (yet). } - select { - case <-s.close: + if s.isClosed() { return 0, nil, errClosed + } + select { case s.packetChan <- p: + default: + s.log.Trace("packet channel is full, dropping packet") } return i, attr, nil @@ -161,6 +168,17 @@ func (s *SenderInterceptor) isClosed() bool { } } +func (s *SenderInterceptor) recordQueuedPackets() { + for range len(s.packetChan) { + select { + case pkt := <-s.packetChan: + s.recorder.AddPacket(pkt.arrival, pkt.ssrc, pkt.sequenceNumber, pkt.ecn) + default: + return + } + } +} + func (s *SenderInterceptor) loop(writer interceptor.RTCPWriter) { defer s.wg.Done() @@ -186,6 +204,7 @@ func (s *SenderInterceptor) loop(writer interceptor.RTCPWriter) { s.recorder.AddPacket(pkt.arrival, pkt.ssrc, pkt.sequenceNumber, pkt.ecn) case <-t.Ch(): + s.recordQueuedPackets() now := s.now() s.log.Tracef("report triggered at %v", now) if writer == nil { diff --git a/pkg/rfc8888/interceptor_test.go b/pkg/rfc8888/interceptor_test.go index 891061ba..a4646ad7 100644 --- a/pkg/rfc8888/interceptor_test.go +++ b/pkg/rfc8888/interceptor_test.go @@ -348,3 +348,41 @@ func TestConcurrentClose(t *testing.T) { } wg.Wait() } + +func TestReadWithoutRTCPWriter(t *testing.T) { + f, err := NewSenderInterceptor() + assert.NoError(t, err) + + intcp, err := f.NewInterceptor("") + assert.NoError(t, err) + defer func() { + assert.NoError(t, intcp.Close()) + }() + + raw, err := (&rtp.Packet{ + Header: rtp.Header{Version: 2, SequenceNumber: 1, SSRC: 123456}, + Payload: []byte{}, + }).Marshal() + assert.NoError(t, err) + + reader := intcp.BindRemoteStream(&interceptor.StreamInfo{SSRC: 123456}, interceptor.RTPReaderFunc( + func(b []byte, a interceptor.Attributes) (int, interceptor.Attributes, error) { + return copy(b, raw), a, nil + }, + )) + + done := make(chan struct{}) + go func() { + defer close(done) + for range 2 * packetChanBufferSize { + _, _, err := reader.Read(make([]byte, 1500), nil) + assert.NoError(t, err) + } + }() + + select { + case <-done: + case <-time.After(time.Second): + assert.Fail(t, "read blocked without a bound RTCP writer") + } +} diff --git a/pkg/twcc/sender_interceptor.go b/pkg/twcc/sender_interceptor.go index 782dd1b9..fd7b5b81 100644 --- a/pkg/twcc/sender_interceptor.go +++ b/pkg/twcc/sender_interceptor.go @@ -19,12 +19,16 @@ type SenderInterceptorFactory struct { opts []Option } +// packetChanBufferSize is the number of packets buffered for the report loop +// before new packets are dropped. +const packetChanBufferSize = 1000 + var errClosed = errors.New("interceptor is closed") // NewInterceptor constructs a new SenderInterceptor. func (s *SenderInterceptorFactory) NewInterceptor(_ string) (interceptor.Interceptor, error) { senderInterceptor := &SenderInterceptor{ - packetChan: make(chan packet), + packetChan: make(chan packet, packetChanBufferSize), close: make(chan struct{}), interval: 100 * time.Millisecond, startTime: time.Now(), @@ -166,10 +170,13 @@ func (s *SenderInterceptor) BindRemoteStream( arrivalTime: time.Since(s.startTime).Microseconds(), ssrc: info.SSRC, } - select { - case <-s.close: + if s.isClosed() { return 0, nil, errClosed + } + select { case s.packetChan <- p: + default: + s.log.Trace("packet channel is full, dropping packet") } } @@ -200,6 +207,17 @@ func (s *SenderInterceptor) isClosed() bool { } } +func (s *SenderInterceptor) recordQueuedPackets() { + for range len(s.packetChan) { + select { + case p := <-s.packetChan: + s.recorder.Record(p.ssrc, p.sequenceNumber, p.arrivalTime) + default: + return + } + } +} + func (s *SenderInterceptor) loop(writer interceptor.RTCPWriter) { defer s.wg.Done() @@ -221,6 +239,7 @@ func (s *SenderInterceptor) loop(writer interceptor.RTCPWriter) { s.recorder.Record(p.ssrc, p.sequenceNumber, p.arrivalTime) case <-ticker.C: + s.recordQueuedPackets() // build and send twcc pkts := s.recorder.BuildFeedbackPacket() if len(pkts) == 0 { diff --git a/pkg/twcc/sender_interceptor_test.go b/pkg/twcc/sender_interceptor_test.go index 3f951332..5e9b680f 100644 --- a/pkg/twcc/sender_interceptor_test.go +++ b/pkg/twcc/sender_interceptor_test.go @@ -310,3 +310,47 @@ func TestSenderInterceptor_Leak(t *testing.T) { stream.ReceiveRTP(&rtp.Packet{Header: hdr}) } } + +func TestSenderInterceptorReadWithoutRTCPWriter(t *testing.T) { + f, err := NewSenderInterceptor() + assert.NoError(t, err) + + testInterceptor, err := f.NewInterceptor("") + assert.NoError(t, err) + defer func() { + assert.NoError(t, testInterceptor.Close()) + }() + + hdr := rtp.Header{Version: 2} + tcc, err := (&rtp.TransportCCExtension{TransportSequence: 1}).Marshal() + assert.NoError(t, err) + assert.NoError(t, hdr.SetExtension(1, tcc)) + raw, err := (&rtp.Packet{Header: hdr, Payload: []byte{}}).Marshal() + assert.NoError(t, err) + + reader := testInterceptor.BindRemoteStream(&interceptor.StreamInfo{ + SSRC: 1, + RTPHeaderExtensions: []interceptor.RTPHeaderExtension{ + {URI: transportCCURI, ID: 1}, + }, + }, interceptor.RTPReaderFunc( + func(b []byte, a interceptor.Attributes) (int, interceptor.Attributes, error) { + return copy(b, raw), a, nil + }, + )) + + done := make(chan struct{}) + go func() { + defer close(done) + for range 2 * packetChanBufferSize { + _, _, err := reader.Read(make([]byte, 1500), nil) + assert.NoError(t, err) + } + }() + + select { + case <-done: + case <-time.After(time.Second): + assert.Fail(t, "read blocked without a bound RTCP writer") + } +}