dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-07-29 18:29:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

Drop padding only packets on publisher side. (#1990)

Browse Source 
* Drop padding only packet on publisher side.

* add UT

* update deps

* remove debug

* add fast path short cut

* correct comment

* fix test

* fix for Linux
This commit is contained in:
Raja Subramanian
2023-08-24 09:09:42 +05:30
committed by GitHub
parent 3733c4af6c
commit 36dadbacb2
7 changed files with 315 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go.mod
+1 -1
View File
@@ -17,7 +17,7 @@ require (
github.com/hashicorp/golang-lru/v2 v2.0.5
github.com/jxskiss/base62 v1.1.0
github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1
github.com/livekit/mediatransportutil v0.0.0-20230815100155-96164dbcfd8c
github.com/livekit/mediatransportutil v0.0.0-20230823131232-12f579dc9af0
github.com/livekit/protocol v1.6.1
github.com/livekit/psrpc v0.3.3
github.com/mackerelio/go-osstat v0.2.4
go.sum
+2 -2
View File
@@ -122,8 +122,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-20230815100155-96164dbcfd8c h1:4udPqCusH93MK/7q8ZfDqcLJHGoQeKKsMi5b+/BpQvk=
github.com/livekit/mediatransportutil v0.0.0-20230815100155-96164dbcfd8c/go.mod h1:xirUXW8xnLGmfCwUeAv/nj1VGo1OO1BmgxrYP7jK/14=
github.com/livekit/mediatransportutil v0.0.0-20230823131232-12f579dc9af0 h1:cHNvPzn6VHFcsHx8ZC9LwU/4jj22mW3LILrNg/y5A6I=
github.com/livekit/mediatransportutil v0.0.0-20230823131232-12f579dc9af0/go.mod h1:xirUXW8xnLGmfCwUeAv/nj1VGo1OO1BmgxrYP7jK/14=
github.com/livekit/protocol v1.6.1 h1:MjRg/UBmynE636In1GD9PbrF2u/C10WwaVIkObsZYtk=
github.com/livekit/protocol v1.6.1/go.mod h1:/JuO+G/btZ5gNwX2+901L6za3UvVO6DHRXHsv8kkLsU=
github.com/livekit/psrpc v0.3.3 h1:+lltbuN39IdaynXhLLxRShgYqYsRMWeeXKzv60oqyWo=
pkg/sfu/buffer/buffer.go
+32 -27
View File
@@ -30,6 +30,7 @@ import (
"go.uber.org/atomic"
"github.com/livekit/livekit-server/pkg/sfu/audio"
"github.com/livekit/livekit-server/pkg/sfu/utils"
sutils "github.com/livekit/livekit-server/pkg/utils"
"github.com/livekit/mediatransportutil"
"github.com/livekit/mediatransportutil/pkg/bucket"
@@ -80,6 +81,8 @@ type Buffer struct {
closed atomic.Bool
mime string
snRangeMap *utils.RangeMap[uint32, uint32]
// supported feedbacks
latestTSForAudioLevelInitialized bool
latestTSForAudioLevel uint32
@@ -124,6 +127,7 @@ func NewBuffer(ssrc uint32, vp, ap *sync.Pool) *Buffer {
mediaSSRC: ssrc,
videoPool: vp,
audioPool: ap,
snRangeMap: utils.NewRangeMap[uint32, uint32](100),
pliThrottle: int64(500 * time.Millisecond),
logger: l.WithComponent(sutils.ComponentPub).WithComponent(sutils.ComponentSFU),
}
@@ -404,42 +408,40 @@ func (b *Buffer) SetRTT(rtt uint32) {
}
func (b *Buffer) calc(pkt []byte, arrivalTime time.Time) {
pktBuf, err := b.bucket.AddPacket(pkt)
if err != nil {
//
// Even when erroring, do
// 1. state update
// 2. TWCC just in case remote side is retransmitting an old packet for probing
//
// But, do not forward those packets
//
var rtpPacket rtp.Packet
if uerr := rtpPacket.Unmarshal(pkt); uerr == nil {
b.updateStreamState(&rtpPacket, arrivalTime)
b.processHeaderExtensions(&rtpPacket, arrivalTime)
}
var rtpPacket rtp.Packet
if err := rtpPacket.Unmarshal(pkt); err != nil {
b.logger.Errorw("could not unmarshal RTP packet", err)
return
}
extSeqNumber, isOutOfOrder := b.updateStreamState(&rtpPacket, arrivalTime)
b.processHeaderExtensions(&rtpPacket, arrivalTime)
if !isOutOfOrder && len(rtpPacket.Payload) == 0 {
// drop padding only in-order packet
b.snRangeMap.IncValue(1)
return
}
// add to RTX buffer using sequence number after accounting for dropped padding only packets
snAdjustment, err := b.snRangeMap.GetValue(extSeqNumber)
if err != nil {
b.logger.Errorw("could not get sequence number adjustment", err)
return
}
rtpPacket.Header.SequenceNumber = uint16(extSeqNumber - snAdjustment)
_, err = b.bucket.AddPacketWithSequenceNumber(pkt, rtpPacket.Header.SequenceNumber)
if err != nil {
if err != bucket.ErrRTXPacket {
b.logger.Warnw("could not add RTP packet to bucket", err)
}
return
}
var p rtp.Packet
err = p.Unmarshal(pktBuf)
if err != nil {
b.logger.Warnw("error unmarshaling RTP packet", err)
return
}
b.updateStreamState(&p, arrivalTime)
b.processHeaderExtensions(&p, arrivalTime)
b.doNACKs()
b.doReports(arrivalTime)
ep := b.getExtPacket(&p, arrivalTime)
ep := b.getExtPacket(&rtpPacket, arrivalTime)
if ep == nil {
return
}
@@ -497,18 +499,21 @@ func (b *Buffer) doFpsCalc(ep *ExtPacket) {
}
}
func (b *Buffer) updateStreamState(p *rtp.Packet, arrivalTime time.Time) {
func (b *Buffer) updateStreamState(p *rtp.Packet, arrivalTime time.Time) (uint32, bool) {
flowState := b.rtpStats.Update(&p.Header, len(p.Payload), int(p.PaddingSize), arrivalTime)
if b.nacker != nil {
b.nacker.Remove(p.SequenceNumber)
if flowState.HasLoss {
b.snRangeMap.AddRange(flowState.LossStartInclusive, flowState.LossEndExclusive)
for lost := flowState.LossStartInclusive; lost != flowState.LossEndExclusive; lost++ {
b.nacker.Push(lost)
b.nacker.Push(uint16(lost))
}
}
}
return flowState.ExtSeqNumber, flowState.IsOutOfOrder
}
func (b *Buffer) processHeaderExtensions(p *rtp.Packet, arrivalTime time.Time) {
pkg/sfu/buffer/rtpstats.go
+33 -8
View File
@@ -65,8 +65,12 @@ func (d driftResult) String() string {
type RTPFlowState struct {
HasLoss bool
LossStartInclusive uint16
LossEndExclusive uint16
LossStartInclusive uint32
LossEndExclusive uint32
IsOutOfOrder bool
ExtSeqNumber uint32
}
type IntervalStats struct {
@@ -449,14 +453,16 @@ func (r *RTPStats) Update(rtph *rtp.Header, payloadSize int, paddingSize int, pa
}
}
flowState.IsOutOfOrder = true
cycles := r.cycles
if rtph.SequenceNumber > r.highestSN {
cycles--
}
flowState.ExtSeqNumber = getExtSN(rtph.SequenceNumber, cycles)
// in-order
default:
if diff > 1 {
flowState.HasLoss = true
flowState.LossStartInclusive = r.highestSN + 1
flowState.LossEndExclusive = rtph.SequenceNumber
}
// update gap histogram
r.updateGapHistogram(int(diff))
@@ -466,6 +472,16 @@ func (r *RTPStats) Update(rtph *rtp.Header, payloadSize int, paddingSize int, pa
r.setSnInfo(rtph.SequenceNumber, uint16(pktSize), uint16(hdrSize), uint16(payloadSize), rtph.Marker, false)
if diff > 1 {
flowState.HasLoss = true
cycles := r.cycles
if r.highestSN+1 < r.highestSN {
cycles++
}
flowState.LossStartInclusive = getExtSN(r.highestSN+1, cycles)
}
if rtph.SequenceNumber < r.highestSN && !first {
r.cycles++
}
@@ -481,6 +497,11 @@ func (r *RTPStats) Update(rtph *rtp.Header, payloadSize int, paddingSize int, pa
// NOTE: this may not be the first packet with this time stamp if there is packet loss.
r.highestTime = packetTime
}
if flowState.HasLoss {
flowState.LossEndExclusive = getExtSN(rtph.SequenceNumber, r.cycles)
}
flowState.ExtSeqNumber = getExtSN(rtph.SequenceNumber, r.cycles)
}
if !isDuplicate {
@@ -1733,6 +1754,10 @@ func (r *RTPStats) getAndResetSnapshot(snapshotId uint32, override bool) (*Snaps
// ----------------------------------
func getExtSN(sn uint16, cycles uint16) uint32 {
return (uint32(cycles) << 16) | uint32(sn)
}
func getExtTS(ts uint32, cycles uint32) uint64 {
return (uint64(cycles) << 32) | uint64(ts)
}
pkg/sfu/buffer/rtpstats_test.go
+2 -2
View File
@@ -126,8 +126,8 @@ func TestRTPStats_Update(t *testing.T) {
packet = getPacket(sequenceNumber, timestamp, 1000)
flowState = r.Update(&packet.Header, len(packet.Payload), 0, time.Now())
require.True(t, flowState.HasLoss)
require.Equal(t, sequenceNumber-9, flowState.LossStartInclusive)
require.Equal(t, sequenceNumber, flowState.LossEndExclusive)
require.Equal(t, uint32(sequenceNumber-9), flowState.LossStartInclusive)
require.Equal(t, uint32(sequenceNumber), flowState.LossEndExclusive)
require.Equal(t, uint32(17), r.packetsLost)
// out-of-order should decrement number of lost packets
pkg/sfu/utils/rangemap.go
+123
View File
@@ -0,0 +1,123 @@
// Copyright 2023 LiveKit, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package utils
import (
"errors"
"math"
"unsafe"
)
const (
minRanges = 1
)
var (
errReversedOrder = errors.New("end is before start")
errKeyNotFound = errors.New("key not found")
)
type rangeType interface {
uint32
}
type valueType interface {
uint32
}
type rangeVal[RT rangeType, VT valueType] struct {
start RT
end RT
value VT
}
type RangeMap[RT rangeType, VT valueType] struct {
halfRange RT
size int
ranges []rangeVal[RT, VT]
runningValue VT
}
func NewRangeMap[RT rangeType, VT valueType](size int) *RangeMap[RT, VT] {
var t RT
return &RangeMap[RT, VT]{
halfRange: 1 << ((unsafe.Sizeof(t) * 8) - 1),
size: int(math.Max(float64(size), float64(minRanges))),
}
}
func (r *RangeMap[RT, VT]) IncValue(inc VT) {
r.runningValue += inc
}
func (r *RangeMap[RT, VT]) AddRange(startInclusive RT, endExclusive RT) error {
if endExclusive-startInclusive > r.halfRange {
return errReversedOrder
}
isNewRange := true
// check if last range can be extended
if len(r.ranges) != 0 {
lr := &r.ranges[len(r.ranges)-1]
if startInclusive <= lr.end {
return errReversedOrder
}
if lr.value == r.runningValue {
lr.end = endExclusive - 1
isNewRange = false
} else {
// end last range before start and start a new range
lr.end = startInclusive - 1
}
}
if isNewRange {
r.ranges = append(r.ranges, rangeVal[RT, VT]{
start: startInclusive,
end: endExclusive - 1,
value: r.runningValue,
})
}
r.prune()
return nil
}
func (r *RangeMap[RT, VT]) GetValue(key RT) (VT, error) {
numRanges := len(r.ranges)
if numRanges != 0 {
if key > r.ranges[numRanges-1].end {
return r.runningValue, nil
}
if key < r.ranges[0].start {
return 0, errKeyNotFound
}
}
for _, rv := range r.ranges {
if key-rv.start < r.halfRange && rv.end-key < r.halfRange {
return rv.value, nil
}
}
return r.runningValue, nil
}
func (r *RangeMap[RT, VT]) prune() {
if len(r.ranges) > r.size {
r.ranges = r.ranges[len(r.ranges)-r.size:]
}
}
pkg/sfu/utils/rangemap_test.go
+122
View File
@@ -0,0 +1,122 @@
// Copyright 2023 LiveKit, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package utils
import (
"testing"
"github.com/stretchr/testify/require"
)
func TestRangeMapUint32(t *testing.T) {
r := NewRangeMap[uint32, uint32](2)
// getting value for any key should be 0 default
value, err := r.GetValue(33333)
require.NoError(t, err)
require.Equal(t, uint32(0), value)
value, err = r.GetValue(0xffffffff)
require.NoError(t, err)
require.Equal(t, uint32(0), value)
// getting value for any key should be incremented value
r.IncValue(2)
value, err = r.GetValue(66666666)
require.NoError(t, err)
require.Equal(t, uint32(2), value)
value, err = r.GetValue(0)
require.NoError(t, err)
require.Equal(t, uint32(2), value)
// add a couple of ranges, as the value is same should just extend
err = r.AddRange(10, 20)
require.NoError(t, err)
err = r.AddRange(30, 40)
require.NoError(t, err)
require.Equal(t, 1, len(r.ranges))
require.Equal(t, uint32(10), r.ranges[0].start)
require.Equal(t, uint32(39), r.ranges[0].end)
require.Equal(t, uint32(2), r.ranges[0].value)
// bump value
r.IncValue(1)
// getting value in previously added range should return 2
value, err = r.GetValue(22)
require.NoError(t, err)
require.Equal(t, uint32(2), value)
// outside range should return 3
value, err = r.GetValue(662)
require.NoError(t, err)
require.Equal(t, uint32(3), value)
// adding out-of-order range should return error
err = r.AddRange(60, 50)
require.Error(t, err, errReversedOrder)
// adding overlapping should return error
err = r.AddRange(30, 50)
require.Error(t, err, errReversedOrder)
// adding a non-overlapping range should extend previous range and add new one
err = r.AddRange(50, 60)
require.NoError(t, err)
require.Equal(t, 2, len(r.ranges))
require.Equal(t, uint32(10), r.ranges[0].start)
require.Equal(t, uint32(49), r.ranges[0].end)
require.Equal(t, uint32(2), r.ranges[0].value)
require.Equal(t, uint32(50), r.ranges[1].start)
require.Equal(t, uint32(59), r.ranges[1].end)
require.Equal(t, uint32(3), r.ranges[1].value)
// getting an old value should not succeed, but start of first range should return no error
value, err = r.GetValue(9)
require.Error(t, err, errKeyNotFound)
value, err = r.GetValue(10)
require.NoError(t, err)
require.Equal(t, uint32(2), value)
// adding another range should prune the first one as size if set to 2
r.IncValue(10)
err = r.AddRange(1000, 1233)
require.NoError(t, err)
require.Equal(t, 2, len(r.ranges))
require.Equal(t, uint32(50), r.ranges[0].start)
require.Equal(t, uint32(999), r.ranges[0].end)
require.Equal(t, uint32(3), r.ranges[0].value)
require.Equal(t, uint32(1000), r.ranges[1].start)
require.Equal(t, uint32(1232), r.ranges[1].end)
require.Equal(t, uint32(13), r.ranges[1].value)
// previously valid range should return key not found after pruning
value, err = r.GetValue(10)
require.Error(t, err, errKeyNotFound)
value, err = r.GetValue(999)
require.NoError(t, err)
require.Equal(t, uint32(3), value)
value, err = r.GetValue(1200)
require.NoError(t, err)
require.Equal(t, uint32(13), value)
// something newer than what is in ranges should return running value
value, err = r.GetValue(3000)
require.NoError(t, err)
require.Equal(t, uint32(13), value)
}