Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

jitterbuffer: Use jitterbuffer in SampleBuilder #2959

Draft
wants to merge 1 commit into
base: master
Choose a base branch
from
+42 −21
Draft

jitterbuffer: Use jitterbuffer in SampleBuilder #2959

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
52 changes: 34 additions & 18 deletions pkg/media/samplebuilder/samplebuilder.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
"math"
"time"

"github.com/pion/interceptor/pkg/jitterbuffer"
"github.com/pion/rtp"
"github.com/pion/webrtc/v4/pkg/media"
)
Expand All @@ -16,7 +17,7 @@
type SampleBuilder struct {
maxLate uint16 // how many packets to wait until we get a valid Sample
maxLateTimestamp uint32 // max timestamp between old and new timestamps before dropping packets
buffer [math.MaxUint16 + 1]*rtp.Packet
buffer *jitterbuffer.JitterBuffer
preparedSamples [math.MaxUint16 + 1]*media.Sample

// Interface that allows us to take RTP packets to samples
Expand Down Expand Up @@ -60,7 +61,7 @@
// The depacketizer extracts media samples from RTP packets.
// Several depacketizers are available in package github.com/pion/rtp/codecs.
func New(maxLate uint16, depacketizer rtp.Depacketizer, sampleRate uint32, opts ...Option) *SampleBuilder {
s := &SampleBuilder{maxLate: maxLate, depacketizer: depacketizer, sampleRate: sampleRate}
s := &SampleBuilder{maxLate: maxLate, depacketizer: depacketizer, sampleRate: sampleRate, buffer: jitterbuffer.New(jitterbuffer.WithMinimumPacketCount(1))}
for _, o := range opts {
o(s)
}
Expand All @@ -76,7 +77,7 @@
var foundTail *rtp.Packet

for i := location.head; i != location.tail; i++ {
if packet := s.buffer[i]; packet != nil {
if packet, _ := s.buffer.PeekAtSequence(i); packet != nil {
foundHead = packet
break
}
Expand All @@ -87,7 +88,7 @@
}

for i := location.tail - 1; i != location.head; i-- {
if packet := s.buffer[i]; packet != nil {
if packet, _ := s.buffer.PeekAtSequence(i); packet != nil {
foundTail = packet
break
}
Expand All @@ -105,7 +106,7 @@
if location.empty() {
return 0, false
}
packet := s.buffer[location.head]
packet, _ := s.buffer.PeekAtSequence(location.head)
if packet == nil {
return 0, false
}
Expand All @@ -114,7 +115,7 @@

func (s *SampleBuilder) releasePacket(i uint16) {
var p *rtp.Packet
p, s.buffer[i] = s.buffer[i], nil
p, _ = s.buffer.PopAtSequence(i)
if p != nil && s.packetReleaseHandler != nil {
s.packetReleaseHandler(p)
}
Expand Down Expand Up @@ -178,7 +179,7 @@
// Push does not copy the input. If you wish to reuse
// this memory make sure to copy before calling Push
func (s *SampleBuilder) Push(p *rtp.Packet) {
s.buffer[p.SequenceNumber] = p
s.buffer.Push(p)

switch s.filled.compare(p.SequenceNumber) {
case slCompareVoid:
Expand Down Expand Up @@ -220,14 +221,18 @@

var consume sampleSequenceLocation

for i := s.active.head; s.buffer[i] != nil && s.active.compare(i) != slCompareAfter; i++ {
if s.depacketizer.IsPartitionTail(s.buffer[i].Marker, s.buffer[i].Payload) {
for i := s.active.head; s.active.compare(i) != slCompareAfter; i++ {
pkt, err := s.buffer.PeekAtSequence(i)
if pkt == nil || err != nil {
break
}
if s.depacketizer.IsPartitionTail(pkt.Marker, pkt.Payload) {
consume.head = s.active.head
consume.tail = i + 1
break
}
headTimestamp, hasData := s.fetchTimestamp(s.active)
if hasData && s.buffer[i].Timestamp != headTimestamp {
if hasData && pkt.Timestamp != headTimestamp {
consume.head = s.active.head
consume.tail = i
break
Expand All @@ -237,8 +242,8 @@
if consume.empty() {
return nil
}

if !purgingBuffers && s.buffer[consume.tail] == nil {
pkt, _ := s.buffer.PeekAtSequence(consume.tail)
if !purgingBuffers && pkt == nil {
// wait for the next packet after this set of packets to arrive
// to ensure at least one post sample timestamp is known
// (unless we have to release right now)
Expand All @@ -250,8 +255,9 @@

// scan for any packet after the current and use that time stamp as the diff point
for i := consume.tail; i < s.active.tail; i++ {
if s.buffer[i] != nil {
afterTimestamp = s.buffer[i].Timestamp
pkt, _ := s.buffer.PeekAtSequence(i)

Check failure on line 258 in pkg/media/samplebuilder/samplebuilder.go

View workflow job for this annotation

GitHub Actions / lint / Go 

shadow: declaration of "pkt" shadows declaration at line 245 (govet)
if pkt != nil {
afterTimestamp = pkt.Timestamp
break
}
}
Expand All @@ -261,10 +267,12 @@

// prior to decoding all the packets, check if this packet
// would end being disposed anyway
if !s.depacketizer.IsPartitionHead(s.buffer[consume.head].Payload) {
pkt, err := s.buffer.PeekAtSequence(consume.head)
if pkt != nil && err == nil && !s.depacketizer.IsPartitionHead(pkt.Payload) {
isPadding := false
for i := consume.head; i != consume.tail; i++ {
if s.lastSampleTimestamp != nil && *s.lastSampleTimestamp == s.buffer[i].Timestamp && len(s.buffer[i].Payload) == 0 {
pkt, _ := s.buffer.PeekAtSequence(i)
if s.lastSampleTimestamp != nil && *s.lastSampleTimestamp == pkt.Timestamp && len(pkt.Payload) == 0 {
isPadding = true
}
}
Expand All @@ -282,15 +290,23 @@
var metadata interface{}
var rtpHeaders []*rtp.Header
for i := consume.head; i != consume.tail; i++ {
p, err := s.depacketizer.Unmarshal(s.buffer[i].Payload)
pkt, _ := s.buffer.PeekAtSequence(i)
if pkt == nil {
return nil
}

Check warning on line 296 in pkg/media/samplebuilder/samplebuilder.go

View check run for this annotation

Codecov / codecov/patch

pkg/media/samplebuilder/samplebuilder.go#L295-L296 

Added lines #L295 - L296 were not covered by tests
p, err := s.depacketizer.Unmarshal(pkt.Payload)
if err != nil {
return nil
}
if i == consume.head && s.packetHeadHandler != nil {
metadata = s.packetHeadHandler(s.depacketizer)
}
if s.returnRTPHeaders {
h := s.buffer[i].Header.Clone()
pkt, err := s.buffer.PeekAtSequence(i)
if err != nil {
return nil
}

Check warning on line 308 in pkg/media/samplebuilder/samplebuilder.go

View check run for this annotation

Codecov / codecov/patch

pkg/media/samplebuilder/samplebuilder.go#L307-L308 

Added lines #L307 - L308 were not covered by tests
h := pkt.Header.Clone()
rtpHeaders = append(rtpHeaders, &h)
}

Expand Down
11 changes: 8 additions & 3 deletions pkg/media/samplebuilder/samplebuilder_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -396,14 +396,19 @@ func TestSampleBuilderCleanReference(t *testing.T) {
s.Push(pkt5)

for i := 0; i < 3; i++ {
if s.buffer[(i+int(seqStart))%0x10000] != nil {
pkt, err := s.buffer.PeekAtSequence(uint16((i + int(seqStart)) % 0x10000))

if pkt != nil || err == nil {
t.Errorf("Old packet (%d) is not unreferenced (maxLate: 10, pushed: 12)", i)
}
}
if s.buffer[(14+int(seqStart))%0x10000] != pkt4 {
pkt, _ := s.buffer.PeekAtSequence(uint16((14 + int(seqStart)) % 0x10000))
if pkt != pkt4 {
t.Error("New packet must be referenced after jump")
}
if s.buffer[(12+int(seqStart))%0x10000] != pkt5 {
pkt, _ = s.buffer.PeekAtSequence(uint16((12 + int(seqStart)) % 0x10000))

if pkt != pkt5 {
t.Error("New packet must be referenced after jump")
}
})
Expand Down
Loading