diff --git a/go.mod b/go.mod index 49d11d08b..24d5bf8fb 100644 --- a/go.mod +++ b/go.mod @@ -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 diff --git a/go.sum b/go.sum index 35686219f..31050de5f 100644 --- a/go.sum +++ b/go.sum @@ -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= diff --git a/pkg/sfu/buffer/buffer.go b/pkg/sfu/buffer/buffer.go index baae76b46..957aaa452 100644 --- a/pkg/sfu/buffer/buffer.go +++ b/pkg/sfu/buffer/buffer.go @@ -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) { diff --git a/pkg/sfu/buffer/rtpstats.go b/pkg/sfu/buffer/rtpstats.go index 24a0c34a7..91069d377 100644 --- a/pkg/sfu/buffer/rtpstats.go +++ b/pkg/sfu/buffer/rtpstats.go @@ -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) } diff --git a/pkg/sfu/buffer/rtpstats_test.go b/pkg/sfu/buffer/rtpstats_test.go index 389d330c3..74e4774f4 100644 --- a/pkg/sfu/buffer/rtpstats_test.go +++ b/pkg/sfu/buffer/rtpstats_test.go @@ -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 diff --git a/pkg/sfu/utils/rangemap.go b/pkg/sfu/utils/rangemap.go new file mode 100644 index 000000000..75ffd1646 --- /dev/null +++ b/pkg/sfu/utils/rangemap.go @@ -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:] + } +} diff --git a/pkg/sfu/utils/rangemap_test.go b/pkg/sfu/utils/rangemap_test.go new file mode 100644 index 000000000..dcc6a745d --- /dev/null +++ b/pkg/sfu/utils/rangemap_test.go @@ -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) +}