dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-08-27 22:34:25 +00:00
Code Issues Packages Projects Releases Wiki Activity

Use extended sequence number in bucket (#2895)

Browse Source
This commit is contained in:
Raja Subramanian
2024-07-30 14:21:37 +05:30
committed by GitHub
parent 4e29e18129
commit d68dd3033d
10 changed files with 36 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go.mod
+1 -1
View File
@@ -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
go.sum
+2 -2
View File
@@ -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=
pkg/rtc/wrappedreceiver.go
+2 -2
View File
@@ -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")
}
pkg/sfu/buffer/buffer.go
+10 -9
View File
@@ -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)) {
pkg/sfu/buffer/rtpstats_sender.go
+7
View File
@@ -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 {
pkg/sfu/receiver.go
+3 -3
View File
@@ -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 {
pkg/sfu/redprimaryreceiver.go
+2 -2
View File
@@ -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
}
pkg/sfu/redreceiver.go
+1 -1
View File
@@ -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
}
pkg/sfu/sequencer.go
+4 -4
View File
@@ -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,
pkg/sfu/sequencer_test.go
+4 -4
View File
@@ -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])