From d68dd3033d45fd747926a0d55d4d9da8292c797b Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Tue, 30 Jul 2024 14:21:37 +0530 Subject: [PATCH] Use extended sequence number in bucket (#2895) --- go.mod | 2 +- go.sum | 4 ++-- pkg/rtc/wrappedreceiver.go | 4 ++-- pkg/sfu/buffer/buffer.go | 19 ++++++++++--------- pkg/sfu/buffer/rtpstats_sender.go | 7 +++++++ pkg/sfu/receiver.go | 6 +++--- pkg/sfu/redprimaryreceiver.go | 4 ++-- pkg/sfu/redreceiver.go | 2 +- pkg/sfu/sequencer.go | 8 ++++---- pkg/sfu/sequencer_test.go | 8 ++++---- 10 files changed, 36 insertions(+), 28 deletions(-) diff --git a/go.mod b/go.mod index add999cac..e9fb31d38 100644 --- a/go.mod +++ b/go.mod @@ -19,7 +19,7 @@ require ( github.com/jellydator/ttlcache/v3 v3.2.0 github.com/jxskiss/base62 v1.1.0 github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1 - github.com/livekit/mediatransportutil v0.0.0-20240625074155-301bb4a816b7 + github.com/livekit/mediatransportutil v0.0.0-20240730083616-559fa5ece598 github.com/livekit/protocol v1.19.2-0.20240719172332-0df8e893874b github.com/livekit/psrpc v0.5.3-0.20240616012458-ac39c8549a0a github.com/mackerelio/go-osstat v0.2.5 diff --git a/go.sum b/go.sum index 6b708f506..ad46d5297 100644 --- a/go.sum +++ b/go.sum @@ -165,8 +165,8 @@ github.com/lithammer/shortuuid/v4 v4.0.0 h1:QRbbVkfgNippHOS8PXDkti4NaWeyYfcBTHtw github.com/lithammer/shortuuid/v4 v4.0.0/go.mod h1:Zs8puNcrvf2rV9rTH51ZLLcj7ZXqQI3lv67aw4KiB1Y= github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1 h1:jm09419p0lqTkDaKb5iXdynYrzB84ErPPO4LbRASk58= github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ= -github.com/livekit/mediatransportutil v0.0.0-20240625074155-301bb4a816b7 h1:F1L8inJoynwIAYpZENNYS+1xHJMF5RFRorsnAlcxfSY= -github.com/livekit/mediatransportutil v0.0.0-20240625074155-301bb4a816b7/go.mod h1:jwKUCmObuiEDH0iiuJHaGMXwRs3RjrB4G6qqgkr/5oE= +github.com/livekit/mediatransportutil v0.0.0-20240730083616-559fa5ece598 h1:yLlkHk2feSLHstD9n4VKg7YEBR4rLODTI4WE8gNBEnQ= +github.com/livekit/mediatransportutil v0.0.0-20240730083616-559fa5ece598/go.mod h1:jwKUCmObuiEDH0iiuJHaGMXwRs3RjrB4G6qqgkr/5oE= github.com/livekit/protocol v1.19.2-0.20240719172332-0df8e893874b h1:Wn6D+B5YbMe1tH7WCazLJz+msBQzR69dK2wTdgJsF5k= github.com/livekit/protocol v1.19.2-0.20240719172332-0df8e893874b/go.mod h1:bNjJi+8frdvC84xG0CJ/7VfVvqerLg2MzjOks0ucyC4= github.com/livekit/psrpc v0.5.3-0.20240616012458-ac39c8549a0a h1:EQAHmcYEGlc6V517cQ3Iy0+jHgP6+tM/B4l2vGuLpQo= diff --git a/pkg/rtc/wrappedreceiver.go b/pkg/rtc/wrappedreceiver.go index 547c936e1..854d4f792 100644 --- a/pkg/rtc/wrappedreceiver.go +++ b/pkg/rtc/wrappedreceiver.go @@ -201,9 +201,9 @@ func (d *DummyReceiver) HeaderExtensions() []webrtc.RTPHeaderExtensionParameter return d.headerExtensions } -func (d *DummyReceiver) ReadRTP(buf []byte, layer uint8, sn uint16) (int, error) { +func (d *DummyReceiver) ReadRTP(buf []byte, layer uint8, esn uint64) (int, error) { if r, ok := d.receiver.Load().(sfu.TrackReceiver); ok { - return r.ReadRTP(buf, layer, sn) + return r.ReadRTP(buf, layer, esn) } return 0, errors.New("no receiver") } diff --git a/pkg/sfu/buffer/buffer.go b/pkg/sfu/buffer/buffer.go index 023cade00..1e38ef8f1 100644 --- a/pkg/sfu/buffer/buffer.go +++ b/pkg/sfu/buffer/buffer.go @@ -73,7 +73,7 @@ type ExtPacket struct { type Buffer struct { sync.RWMutex readCond *sync.Cond - bucket *bucket.Bucket + bucket *bucket.Bucket[uint64] nacker *nack.NackQueue maxVideoPkts int maxAudioPkts int @@ -252,10 +252,11 @@ func (b *Buffer) Bind(params webrtc.RTPParameters, codec webrtc.RTPCodecCapabili switch { case strings.HasPrefix(b.mime, "audio/"): b.codecType = webrtc.RTPCodecTypeAudio - b.bucket = bucket.NewBucket(InitPacketBufferSizeAudio) + b.bucket = bucket.NewBucket[uint64](InitPacketBufferSizeAudio) + case strings.HasPrefix(b.mime, "video/"): b.codecType = webrtc.RTPCodecTypeVideo - b.bucket = bucket.NewBucket(InitPacketBufferSizeVideo) + b.bucket = bucket.NewBucket[uint64](InitPacketBufferSizeVideo) if b.frameRateCalculator[0] == nil { if strings.EqualFold(codec.MimeType, webrtc.MimeTypeVP8) { b.frameRateCalculator[0] = NewFrameRateCalculatorVP8(b.clockRate, b.logger) @@ -625,7 +626,7 @@ func (b *Buffer) calc(rawPkt []byte, rtpPacket *rtp.Packet, arrivalTime int64, i } flowState.ExtSequenceNumber -= snAdjustment rtpPacket.Header.SequenceNumber = uint16(flowState.ExtSequenceNumber) - _, err = b.bucket.AddPacketWithSequenceNumber(rawPkt, rtpPacket.Header.SequenceNumber) + _, err = b.bucket.AddPacketWithSequenceNumber(rawPkt, flowState.ExtSequenceNumber) if err != nil { if !flowState.IsDuplicate { if errors.Is(err, bucket.ErrPacketTooOld) { @@ -671,7 +672,7 @@ func (b *Buffer) calc(rawPkt []byte, rtpPacket *rtp.Packet, arrivalTime int64, i } func (b *Buffer) patchExtPacket(ep *ExtPacket, buf []byte) *ExtPacket { - n, err := b.getPacket(buf, ep.Packet.SequenceNumber) + n, err := b.getPacket(buf, ep.ExtSequenceNumber) if err != nil { packetNotFoundCount := b.packetNotFoundCount.Inc() if (packetNotFoundCount-1)%20 == 0 { @@ -1025,18 +1026,18 @@ func (b *Buffer) getRTCP() []rtcp.Packet { return pkts } -func (b *Buffer) GetPacket(buff []byte, sn uint16) (int, error) { +func (b *Buffer) GetPacket(buff []byte, esn uint64) (int, error) { b.Lock() defer b.Unlock() - return b.getPacket(buff, sn) + return b.getPacket(buff, esn) } -func (b *Buffer) getPacket(buff []byte, sn uint16) (int, error) { +func (b *Buffer) getPacket(buff []byte, esn uint64) (int, error) { if b.closed.Load() { return 0, io.EOF } - return b.bucket.GetPacket(buff, sn) + return b.bucket.GetPacket(buff, esn) } func (b *Buffer) OnRtcpFeedback(fn func(fb []rtcp.Packet)) { diff --git a/pkg/sfu/buffer/rtpstats_sender.go b/pkg/sfu/buffer/rtpstats_sender.go index 6494013c6..43ac5eb4a 100644 --- a/pkg/sfu/buffer/rtpstats_sender.go +++ b/pkg/sfu/buffer/rtpstats_sender.go @@ -990,6 +990,13 @@ func (r *RTPStatsSender) getIntervalStats( return } +func (r *RTPStatsSender) ExtHighestSequenceNumber() uint64 { + r.lock.RLock() + defer r.lock.RUnlock() + + return r.extHighestSN +} + // ------------------------------------------------------------------- type lockedRTPStatsSenderLogEncoder struct { diff --git a/pkg/sfu/receiver.go b/pkg/sfu/receiver.go index 7841a8308..cb55b19c4 100644 --- a/pkg/sfu/receiver.go +++ b/pkg/sfu/receiver.go @@ -56,7 +56,7 @@ type TrackReceiver interface { HeaderExtensions() []webrtc.RTPHeaderExtensionParameter IsClosed() bool - ReadRTP(buf []byte, layer uint8, sn uint16) (int, error) + ReadRTP(buf []byte, layer uint8, esn uint64) (int, error) GetLayeredBitrate() ([]int32, Bitrates) GetAudioLevel() (float64, bool) @@ -575,13 +575,13 @@ func (w *WebRTCReceiver) getBufferLocked(layer int32) *buffer.Buffer { return w.buffers[layer] } -func (w *WebRTCReceiver) ReadRTP(buf []byte, layer uint8, sn uint16) (int, error) { +func (w *WebRTCReceiver) ReadRTP(buf []byte, layer uint8, esn uint64) (int, error) { b := w.getBuffer(int32(layer)) if b == nil { return 0, ErrBufferNotFound } - return b.GetPacket(buf, sn) + return b.GetPacket(buf, esn) } func (w *WebRTCReceiver) GetTrackStats() *livekit.RTPStats { diff --git a/pkg/sfu/redprimaryreceiver.go b/pkg/sfu/redprimaryreceiver.go index 956ee7251..e04d56f82 100644 --- a/pkg/sfu/redprimaryreceiver.go +++ b/pkg/sfu/redprimaryreceiver.go @@ -136,8 +136,8 @@ func (r *RedPrimaryReceiver) Close() { closeTrackSenders(r.downTrackSpreader.ResetAndGetDownTracks()) } -func (r *RedPrimaryReceiver) ReadRTP(buf []byte, layer uint8, sn uint16) (int, error) { - n, err := r.TrackReceiver.ReadRTP(buf, layer, sn) +func (r *RedPrimaryReceiver) ReadRTP(buf []byte, layer uint8, esn uint64) (int, error) { + n, err := r.TrackReceiver.ReadRTP(buf, layer, esn) if err != nil { return n, err } diff --git a/pkg/sfu/redreceiver.go b/pkg/sfu/redreceiver.go index 9d24b997b..4af4fb5f5 100644 --- a/pkg/sfu/redreceiver.go +++ b/pkg/sfu/redreceiver.go @@ -127,7 +127,7 @@ func (r *RedReceiver) Close() { closeTrackSenders(r.downTrackSpreader.ResetAndGetDownTracks()) } -func (r *RedReceiver) ReadRTP(buf []byte, layer uint8, sn uint16) (int, error) { +func (r *RedReceiver) ReadRTP(buf []byte, layer uint8, esn uint64) (int, error) { // red encoding doesn't support nack return 0, bucket.ErrPacketMismatch } diff --git a/pkg/sfu/sequencer.go b/pkg/sfu/sequencer.go index 22750a037..294029cc5 100644 --- a/pkg/sfu/sequencer.go +++ b/pkg/sfu/sequencer.go @@ -42,10 +42,10 @@ func itob(i int) bool { } type packetMeta struct { - // Original sequence number from stream. - // The original sequence number is used to find the original + // Original extended sequence number from stream. + // The original extended sequence number is used to find the original // packet from publisher - sourceSeqNo uint16 + sourceSeqNo uint64 // Modified sequence number after offset. // This sequence number is used for the associated // down track, is modified according the offsets, and @@ -199,7 +199,7 @@ func (s *sequencer) push( slot := extModifiedSNAdjusted % uint64(s.size) s.meta[slot] = packetMeta{ - sourceSeqNo: uint16(extIncomingSN), + sourceSeqNo: extIncomingSN, targetSeqNo: uint16(extModifiedSN), timestamp: uint32(extModifiedTS), marker: marker, diff --git a/pkg/sfu/sequencer_test.go b/pkg/sfu/sequencer_test.go index 773c54309..862ac531b 100644 --- a/pkg/sfu/sequencer_test.go +++ b/pkg/sfu/sequencer_test.go @@ -45,7 +45,7 @@ func Test_sequencer(t *testing.T) { require.Equal(t, len(req), len(res)) for i, val := range res { require.Equal(t, val.targetSeqNo, req[i]) - require.Equal(t, val.sourceSeqNo, req[i]-off) + require.Equal(t, val.sourceSeqNo, uint64(req[i]-off)) require.Equal(t, val.layer, int8(2)) require.Equal(t, val.extSequenceNumber, uint64(req[i])) require.Equal(t, val.extTimestamp, uint64(123)) @@ -57,7 +57,7 @@ func Test_sequencer(t *testing.T) { require.Equal(t, len(req), len(res)) for i, val := range res { require.Equal(t, val.targetSeqNo, req[i]) - require.Equal(t, val.sourceSeqNo, req[i]-off) + require.Equal(t, val.sourceSeqNo, uint64(req[i]-off)) require.Equal(t, val.layer, int8(2)) require.Equal(t, val.extSequenceNumber, uint64(req[i])) require.Equal(t, val.extTimestamp, uint64(123)) @@ -204,7 +204,7 @@ func Test_sequencer_getNACKSeqNo_exclusion(t *testing.T) { g := n.getExtPacketMetas(tt.args.seqNo) var got []uint16 for _, sn := range g { - got = append(got, sn.sourceSeqNo) + got = append(got, uint16(sn.sourceSeqNo)) if sn.sourceSeqNo%5 == 0 { require.Equal(t, tt.fields.markerOdd, sn.marker) require.Equal(t, tt.fields.codecBytesOversized, sn.codecBytesSlice) @@ -343,7 +343,7 @@ func Test_sequencer_getNACKSeqNo_no_exclusion(t *testing.T) { g := n.getExtPacketMetas(tt.args.seqNo) var got []uint16 for _, sn := range g { - got = append(got, sn.sourceSeqNo) + got = append(got, uint16(sn.sourceSeqNo)) if sn.sourceSeqNo%2 == 0 { require.Equal(t, tt.fields.markerEven, sn.marker) require.Equal(t, tt.fields.codecBytesEven, sn.codecBytes[:sn.numCodecBytesOut])