dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-09-16 23:24:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/master' into raja_fr

Browse Source
This commit is contained in:
boks1971
2024-03-21 15:37:13 +05:30
parent 19cae0e6ad 0f597e7e46
commit aa85174993
72 changed files with 2042 additions and 1412 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/sfu/buffer/buffer.go
+53 -29
View File
@@ -44,6 +44,9 @@ import (
const (
ReportDelta = time.Second
InitPacketBufferSizeVideo = 300
InitPacketBufferSizeAudio = 70
)
type pendingPacket struct {
@@ -68,8 +71,8 @@ type Buffer struct {
sync.RWMutex
bucket *bucket.Bucket
nacker *nack.NackQueue
videoPool *sync.Pool
audioPool *sync.Pool
maxVideoPkts int
maxAudioPkts int
codecType webrtc.RTPCodecType
payloadType uint8
extPackets deque.Deque[*ExtPacket]
@@ -100,6 +103,7 @@ type Buffer struct {
rtpStats *RTPStatsReceiver
rrSnapshotId uint32
deltaStatsSnapshotId uint32
ppsSnapshotId uint32
lastFractionLostToReport uint8 // Last fraction lost from subscribers, should report to publisher; Audio only
@@ -126,18 +130,19 @@ type Buffer struct {
extPacketTooMuchCount atomic.Uint32
primaryBufferForRTX *Buffer
rtxPktBuf []byte
}
// NewBuffer constructs a new Buffer
func NewBuffer(ssrc uint32, vp, ap *sync.Pool) *Buffer {
func NewBuffer(ssrc uint32, maxVideoPkts, maxAudioPkts int) *Buffer {
l := logger.GetLogger() // will be reset with correct context via SetLogger
b := &Buffer{
mediaSSRC: ssrc,
videoPool: vp,
audioPool: ap,
snRangeMap: utils.NewRangeMap[uint64, uint64](100),
pliThrottle: int64(500 * time.Millisecond),
logger: l.WithComponent(sutils.ComponentPub).WithComponent(sutils.ComponentSFU),
mediaSSRC: ssrc,
maxVideoPkts: maxVideoPkts,
maxAudioPkts: maxAudioPkts,
snRangeMap: utils.NewRangeMap[uint64, uint64](100),
pliThrottle: int64(500 * time.Millisecond),
logger: l.WithComponent(sutils.ComponentPub).WithComponent(sutils.ComponentSFU),
}
b.extPackets.SetMinCapacity(7)
return b
@@ -188,6 +193,7 @@ func (b *Buffer) Bind(params webrtc.RTPParameters, codec webrtc.RTPCodecCapabili
})
b.rrSnapshotId = b.rtpStats.NewSnapshotId()
b.deltaStatsSnapshotId = b.rtpStats.NewSnapshotId()
b.ppsSnapshotId = b.rtpStats.NewSnapshotId()
b.clockRate = codec.ClockRate
b.lastReport = time.Now()
@@ -225,10 +231,10 @@ 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(b.audioPool.Get().(*[]byte))
b.bucket = bucket.NewBucket(InitPacketBufferSizeAudio)
case strings.HasPrefix(b.mime, "video/"):
b.codecType = webrtc.RTPCodecTypeVideo
b.bucket = bucket.NewBucket(b.videoPool.Get().(*[]byte))
b.bucket = bucket.NewBucket(InitPacketBufferSizeVideo)
if b.frameRateCalculator[0] == nil {
if strings.EqualFold(codec.MimeType, webrtc.MimeTypeVP8) {
b.frameRateCalculator[0] = NewFrameRateCalculatorVP8(b.clockRate, b.logger)
@@ -347,22 +353,23 @@ func (b *Buffer) writeRTX(rtxPkt *rtp.Packet) (n int, err error) {
return
}
videoPktPtr := b.videoPool.Get().(*[]byte)
defer b.videoPool.Put(videoPktPtr)
if b.rtxPktBuf == nil {
b.rtxPktBuf = make([]byte, bucket.MaxPktSize)
}
videoPkt := *rtxPkt
videoPkt.PayloadType = b.payloadType
videoPkt.SequenceNumber = binary.BigEndian.Uint16(rtxPkt.Payload[:2])
videoPkt.SSRC = b.mediaSSRC
videoPkt.Payload = rtxPkt.Payload[2:]
n, err = videoPkt.MarshalTo((*videoPktPtr))
n, err = videoPkt.MarshalTo(b.rtxPktBuf)
if err != nil {
b.logger.Errorw("could not marshal repaired packet", err, "ssrc", b.mediaSSRC, "sn", videoPkt.SequenceNumber)
return
}
b.calc((*videoPktPtr)[:n], &videoPkt, time.Now(), true)
b.calc(b.rtxPktBuf[:n], &videoPkt, time.Now(), true)
return
}
@@ -417,13 +424,6 @@ func (b *Buffer) Close() error {
defer b.Unlock()
b.closeOnce.Do(func() {
if b.bucket != nil && b.codecType == webrtc.RTPCodecTypeVideo {
b.videoPool.Put(b.bucket.Src())
}
if b.bucket != nil && b.codecType == webrtc.RTPCodecTypeAudio {
b.audioPool.Put(b.bucket.Src())
}
b.closed.Store(true)
if b.rtpStats != nil {
@@ -457,14 +457,14 @@ func (b *Buffer) SetPLIThrottle(duration int64) {
func (b *Buffer) SendPLI(force bool) {
b.RLock()
if (b.rtpStats == nil || b.rtpStats.TimeSinceLastPli() < b.pliThrottle) && !force {
b.RUnlock()
rtpStats := b.rtpStats
pliThrottle := b.pliThrottle
b.RUnlock()
if (rtpStats == nil && !force) || !rtpStats.CheckAndUpdatePli(pliThrottle, force) {
return
}
b.rtpStats.UpdatePliAndTime(1)
b.RUnlock()
b.logger.Debugw("send pli", "ssrc", b.mediaSSRC, "force", force)
pli := []rtcp.Packet{
&rtcp.PictureLossIndication{SenderSSRC: b.mediaSSRC, MediaSSRC: b.mediaSSRC},
@@ -785,6 +785,30 @@ func (b *Buffer) doReports(arrivalTime time.Time) {
if pkts != nil && b.onRtcpFeedback != nil {
b.onRtcpFeedback(pkts)
}
b.mayGrowBucket()
}
func (b *Buffer) mayGrowBucket() {
cap := b.bucket.Capacity()
maxPkts := b.maxVideoPkts
if b.codecType == webrtc.RTPCodecTypeAudio {
maxPkts = b.maxAudioPkts
}
if cap >= maxPkts {
return
}
oldCap := cap
deltaInfo := b.rtpStats.DeltaInfo(b.ppsSnapshotId)
if deltaInfo != nil && deltaInfo.Duration > 500*time.Millisecond {
pps := int(time.Duration(deltaInfo.Packets) * time.Second / deltaInfo.Duration)
for pps > cap && cap < maxPkts {
cap = b.bucket.Grow()
}
if cap > oldCap {
b.logger.Debugw("grow bucket", "from", oldCap, "to", cap, "pps", pps)
}
}
}
func (b *Buffer) buildNACKPacket() ([]rtcp.Packet, int) {
@@ -825,7 +849,7 @@ func (b *Buffer) SetSenderReportData(rtpTime uint32, ntpTime uint64) {
}
}
func (b *Buffer) GetSenderReportData() (*RTCPSenderReportData, *RTCPSenderReportData) {
func (b *Buffer) GetSenderReportData() *RTCPSenderReportData {
b.RLock()
defer b.RUnlock()
@@ -833,7 +857,7 @@ func (b *Buffer) GetSenderReportData() (*RTCPSenderReportData, *RTCPSenderReport
return b.rtpStats.GetRtcpSenderReportData()
}
return nil, nil
return nil
}
func (b *Buffer) SetLastFractionLostReport(lost uint8) {
pkg/sfu/buffer/buffer_test.go
+15 -23
View File
@@ -48,15 +48,8 @@ var opusCodec = webrtc.RTPCodecParameters{
}
func TestNack(t *testing.T) {
pool := &sync.Pool{
New: func() interface{} {
b := make([]byte, 1500)
return &b
},
}
t.Run("nack normal", func(t *testing.T) {
buff := NewBuffer(123, pool, pool)
buff := NewBuffer(123, 1, 1)
buff.codecType = webrtc.RTPCodecTypeVideo
require.NotNil(t, buff)
var wg sync.WaitGroup
@@ -101,7 +94,7 @@ func TestNack(t *testing.T) {
})
t.Run("nack with seq wrap", func(t *testing.T) {
buff := NewBuffer(123, pool, pool)
buff := NewBuffer(123, 1, 1)
buff.codecType = webrtc.RTPCodecTypeVideo
require.NotNil(t, buff)
var wg sync.WaitGroup
@@ -193,13 +186,7 @@ func TestNewBuffer(t *testing.T) {
},
},
}
pool := &sync.Pool{
New: func() interface{} {
b := make([]byte, 1500)
return &b
},
}
buff := NewBuffer(123, pool, pool)
buff := NewBuffer(123, 1, 1)
buff.codecType = webrtc.RTPCodecTypeVideo
require.NotNil(t, buff)
buff.OnRtcpFeedback(func(_ []rtcp.Packet) {})
@@ -219,13 +206,7 @@ func TestNewBuffer(t *testing.T) {
}
func TestFractionLostReport(t *testing.T) {
pool := &sync.Pool{
New: func() interface{} {
b := make([]byte, 1500)
return &b
},
}
buff := NewBuffer(123, pool, pool)
buff := NewBuffer(123, 1, 1)
require.NotNil(t, buff)
buff.codecType = webrtc.RTPCodecTypeVideo
var wg sync.WaitGroup
@@ -261,3 +242,14 @@ func TestFractionLostReport(t *testing.T) {
}
wg.Wait()
}
func BenchmarkMemcpu(b *testing.B) {
buf := make([]byte, 1500*1500*10)
buf2 := make([]byte, 1500*1500*20)
b.ResetTimer()
for i := 0; i < b.N; i++ {
copy(buf2, buf)
}
}
pkg/sfu/buffer/factory.go
+16 -28
View File
@@ -19,49 +19,37 @@ import (
"sync"
"github.com/pion/transport/v2/packetio"
"github.com/livekit/mediatransportutil/pkg/bucket"
)
type FactoryOfBufferFactory struct {
videoPool *sync.Pool
audioPool *sync.Pool
trackingPacketsVideo int
trackingPacketsAudio int
}
func NewFactoryOfBufferFactory(trackingPackets int) *FactoryOfBufferFactory {
func NewFactoryOfBufferFactory(trackingPacketsVideo int, trackingPacketsAudio int) *FactoryOfBufferFactory {
return &FactoryOfBufferFactory{
videoPool: &sync.Pool{
New: func() interface{} {
b := make([]byte, trackingPackets*bucket.MaxPktSize)
return &b
},
},
audioPool: &sync.Pool{
New: func() interface{} {
b := make([]byte, bucket.MaxPktSize*200)
return &b
},
},
trackingPacketsVideo: trackingPacketsVideo,
trackingPacketsAudio: trackingPacketsAudio,
}
}
func (f *FactoryOfBufferFactory) CreateBufferFactory() *Factory {
return &Factory{
videoPool: f.videoPool,
audioPool: f.audioPool,
rtpBuffers: make(map[uint32]*Buffer),
rtcpReaders: make(map[uint32]*RTCPReader),
rtxPair: make(map[uint32]uint32),
trackingPacketsVideo: f.trackingPacketsVideo,
trackingPacketsAudio: f.trackingPacketsAudio,
rtpBuffers: make(map[uint32]*Buffer),
rtcpReaders: make(map[uint32]*RTCPReader),
rtxPair: make(map[uint32]uint32),
}
}
type Factory struct {
sync.RWMutex
videoPool *sync.Pool
audioPool *sync.Pool
rtpBuffers map[uint32]*Buffer
rtcpReaders map[uint32]*RTCPReader
rtxPair map[uint32]uint32 // repair -> base
trackingPacketsVideo int
trackingPacketsAudio int
rtpBuffers map[uint32]*Buffer
rtcpReaders map[uint32]*RTCPReader
rtxPair map[uint32]uint32 // repair -> base
}
func (f *Factory) GetOrNew(packetType packetio.BufferPacketType, ssrc uint32) io.ReadWriteCloser {
@@ -84,7 +72,7 @@ func (f *Factory) GetOrNew(packetType packetio.BufferPacketType, ssrc uint32) io
if reader, ok := f.rtpBuffers[ssrc]; ok {
return reader
}
buffer := NewBuffer(ssrc, f.videoPool, f.audioPool)
buffer := NewBuffer(ssrc, f.trackingPacketsVideo, f.trackingPacketsAudio)
f.rtpBuffers[ssrc] = buffer
for repair, base := range f.rtxPair {
if repair == ssrc {
pkg/sfu/buffer/frameintegrity.go
+3 -3
View File
@@ -23,7 +23,7 @@ type FrameEntity struct {
endSeq *uint64
integrity bool
packetsConsective func(uint64, uint64) bool
pktHistory *PacketHistory
}
func (fe *FrameEntity) AddPacket(extSeq uint64, ddVal *dd.DependencyDescriptor) {
@@ -40,7 +40,7 @@ func (fe *FrameEntity) AddPacket(extSeq uint64, ddVal *dd.DependencyDescriptor)
}
if fe.startSeq != nil && fe.endSeq != nil {
if fe.packetsConsective(*fe.startSeq, *fe.endSeq) {
if fe.pktHistory.PacketsConsecutive(*fe.startSeq, *fe.endSeq) {
fe.integrity = true
}
}
@@ -179,7 +179,7 @@ func NewFrameIntegrityChecker(frameCount, packetCount int) *FrameIntegrityChecke
}
for i := range fc.frames {
fc.frames[i].packetsConsective = fc.pktHistory.PacketsConsecutive
fc.frames[i].pktHistory = fc.pktHistory
fc.frames[i].Reset()
}
return fc
pkg/sfu/buffer/helpers.go
+8 -5
View File
@@ -163,13 +163,16 @@ func (v *VP8) Unmarshal(payload []byte) error {
func (v *VP8) Marshal() ([]byte, error) {
buf := make([]byte, v.HeaderSize)
err := v.MarshalTo(buf)
return buf, err
n, err := v.MarshalTo(buf)
if err != nil {
return nil, err
}
return buf[:n], err
}
func (v *VP8) MarshalTo(buf []byte) error {
func (v *VP8) MarshalTo(buf []byte) (int, error) {
if len(buf) < v.HeaderSize {
return errShortPacket
return 0, errShortPacket
}
idx := 0
@@ -223,7 +226,7 @@ func (v *VP8) MarshalTo(buf []byte) error {
idx++
}
return nil
return idx, nil
}
// -------------------------------------
pkg/sfu/buffer/rtpstats_base.go
+65 -31
View File
@@ -19,6 +19,7 @@ import (
"sync"
"time"
"go.uber.org/zap/zapcore"
"google.golang.org/protobuf/types/known/timestamppb"
"github.com/livekit/mediatransportutil"
@@ -33,7 +34,7 @@ const (
cFirstSnapshotID = 1
cFirstPacketTimeAdjustWindow = 2 * time.Minute
cFirstPacketTimeAdjustThreshold = 5 * time.Minute
cFirstPacketTimeAdjustThreshold = 15 * time.Second
)
// -------------------------------------------------------
@@ -123,6 +124,18 @@ func (r *RTCPSenderReportData) ToString() string {
return fmt.Sprintf("ntp: %s, rtp: %d, extRtp: %d, at: %s", r.NTPTimestamp.Time().String(), r.RTPTimestamp, r.RTPTimestampExt, r.At.String())
}
func (r *RTCPSenderReportData) MarshalLogObject(e zapcore.ObjectEncoder) error {
if r == nil {
return nil
}
e.AddTime("NTPTimestamp", r.NTPTimestamp.Time())
e.AddUint32("RTPTimestamp", r.RTPTimestamp)
e.AddUint64("RTPTimestampExt", r.RTPTimestampExt)
e.AddTime("At", r.At)
return nil
}
// ------------------------------------------------------------------
type RTPStatsParams struct {
@@ -336,6 +349,18 @@ func (r *rtpStatsBase) UpdateNackProcessed(nackAckCount uint32, nackMissCount ui
r.nackRepeated += nackRepeatedCount
}
func (r *rtpStatsBase) CheckAndUpdatePli(throttle int64, force bool) bool {
r.lock.Lock()
defer r.lock.Unlock()
if !r.endTime.IsZero() || (!force && time.Now().UnixNano()-r.lastPli.UnixNano() < throttle) {
return false
}
r.updatePliLocked(1)
r.updatePliTimeLocked()
return true
}
func (r *rtpStatsBase) UpdatePliAndTime(pliCount uint32) {
r.lock.Lock()
defer r.lock.Unlock()
@@ -385,13 +410,6 @@ func (r *rtpStatsBase) LastPli() time.Time {
return r.lastPli
}
func (r *rtpStatsBase) TimeSinceLastPli() int64 {
r.lock.RLock()
defer r.lock.RUnlock()
return time.Now().UnixNano() - r.lastPli.UnixNano()
}
func (r *rtpStatsBase) UpdateLayerLockPliAndTime(pliCount uint32) {
r.lock.Lock()
defer r.lock.Unlock()
@@ -466,7 +484,7 @@ func (r *rtpStatsBase) GetRtt() uint32 {
return r.rtt
}
func (r *rtpStatsBase) maybeAdjustFirstPacketTime(ts uint32, startTS uint32) {
func (r *rtpStatsBase) maybeAdjustFirstPacketTime(srData *RTCPSenderReportData, tsOffset uint64, extStartTS uint64) {
if time.Since(r.startTime) > cFirstPacketTimeAdjustWindow {
return
}
@@ -477,7 +495,9 @@ func (r *rtpStatsBase) maybeAdjustFirstPacketTime(ts uint32, startTS uint32) {
// abnormal delay (maybe due to pacing or maybe due to queuing
// in some network element along the way), push back first time
// to an earlier instance.
samplesDiff := int32(ts - startTS)
timeSinceReceive := time.Since(srData.At)
extNowTS := srData.RTPTimestampExt - tsOffset + uint64(timeSinceReceive.Nanoseconds()*int64(r.params.ClockRate)/1e9)
samplesDiff := int64(extNowTS - extStartTS)
if samplesDiff < 0 {
// out-of-order, skip
return
@@ -487,28 +507,24 @@ func (r *rtpStatsBase) maybeAdjustFirstPacketTime(ts uint32, startTS uint32) {
timeSinceFirst := time.Since(r.firstTime)
now := r.firstTime.Add(timeSinceFirst)
firstTime := now.Add(-samplesDuration)
if firstTime.Before(r.firstTime) {
r.logger.Debugw(
"adjusting first packet time",
getFields := func() []interface{} {
return []interface{}{
"startTime", r.startTime.String(),
"nowTime", now.String(),
"before", r.firstTime.String(),
"after", firstTime.String(),
"adjustment", r.firstTime.Sub(firstTime).String(),
"nowTS", ts,
"startTS", startTS,
)
"extNowTS", extNowTS,
"extStartTS", extStartTS,
}
}
if firstTime.Before(r.firstTime) {
if r.firstTime.Sub(firstTime) > cFirstPacketTimeAdjustThreshold {
r.logger.Infow("first packet time adjustment too big, ignoring",
"startTime", r.startTime.String(),
"nowTime", now.String(),
"before", r.firstTime.String(),
"after", firstTime.String(),
"adjustment", r.firstTime.Sub(firstTime).String(),
"nowTS", ts,
"startTS", startTS,
)
r.logger.Infow("adjusting first packet time, too big, ignoring", getFields()...)
} else {
r.logger.Debugw("adjusting first packet time", getFields()...)
r.firstTime = firstTime
}
}
@@ -660,7 +676,7 @@ func (r *rtpStatsBase) toString(
str += ", rtt(ms):"
str += fmt.Sprintf("%d|%d", p.RttCurrent, p.RttMax)
str += fmt.Sprintf(", pd: %s, rd: %s", RTPDriftToString(p.PacketDrift), RTPDriftToString(p.ReportDrift))
str += fmt.Sprintf(", pd: %s, nrd: %s, rrd: %s", RTPDriftToString(p.PacketDrift), RTPDriftToString(p.ReportDrift), RTPDriftToString(p.RebasedReportDrift))
return str
}
@@ -701,7 +717,7 @@ func (r *rtpStatsBase) toProto(
jitterTime := jitter / float64(r.params.ClockRate) * 1e6
maxJitterTime := maxJitter / float64(r.params.ClockRate) * 1e6
packetDrift, reportDrift := r.getDrift(extStartTS, extHighestTS)
packetDrift, ntpReportDrift, rebasedReportDrift := r.getDrift(extStartTS, extHighestTS)
p := &livekit.RTPStats{
StartTime: timestamppb.New(r.startTime),
@@ -745,7 +761,8 @@ func (r *rtpStatsBase) toProto(
RttCurrent: r.rtt,
RttMax: r.maxRtt,
PacketDrift: packetDrift,
ReportDrift: reportDrift,
ReportDrift: ntpReportDrift,
RebasedReportDrift: rebasedReportDrift,
}
gapsPresent := false
@@ -827,7 +844,7 @@ func (r *rtpStatsBase) getAndResetSnapshot(snapshotID uint32, extStartSN uint64,
return &then, &now
}
func (r *rtpStatsBase) getDrift(extStartTS, extHighestTS uint64) (packetDrift *livekit.RTPDrift, reportDrift *livekit.RTPDrift) {
func (r *rtpStatsBase) getDrift(extStartTS, extHighestTS uint64) (packetDrift *livekit.RTPDrift, ntpReportDrift *livekit.RTPDrift, rebasedReportDrift *livekit.RTPDrift) {
if !r.firstTime.IsZero() {
elapsed := r.highestTime.Sub(r.firstTime)
rtpClockTicks := extHighestTS - extStartTS
@@ -848,11 +865,12 @@ func (r *rtpStatsBase) getDrift(extStartTS, extHighestTS uint64) (packetDrift *l
}
if r.srFirst != nil && r.srNewest != nil && r.srFirst.RTPTimestamp != r.srNewest.RTPTimestamp {
elapsed := r.srNewest.NTPTimestamp.Time().Sub(r.srFirst.NTPTimestamp.Time())
rtpClockTicks := r.srNewest.RTPTimestampExt - r.srFirst.RTPTimestampExt
elapsed := r.srNewest.NTPTimestamp.Time().Sub(r.srFirst.NTPTimestamp.Time())
driftSamples := int64(rtpClockTicks - uint64(elapsed.Nanoseconds()*int64(r.params.ClockRate)/1e9))
if elapsed.Seconds() > 0.0 {
reportDrift = &livekit.RTPDrift{
ntpReportDrift = &livekit.RTPDrift{
StartTime: timestamppb.New(r.srFirst.NTPTimestamp.Time()),
EndTime: timestamppb.New(r.srNewest.NTPTimestamp.Time()),
Duration: elapsed.Seconds(),
@@ -864,6 +882,22 @@ func (r *rtpStatsBase) getDrift(extStartTS, extHighestTS uint64) (packetDrift *l
ClockRate: float64(rtpClockTicks) / elapsed.Seconds(),
}
}
elapsed = r.srNewest.At.Sub(r.srFirst.At)
driftSamples = int64(rtpClockTicks - uint64(elapsed.Nanoseconds()*int64(r.params.ClockRate)/1e9))
if elapsed.Seconds() > 0.0 {
rebasedReportDrift = &livekit.RTPDrift{
StartTime: timestamppb.New(r.srFirst.At),
EndTime: timestamppb.New(r.srNewest.At),
Duration: elapsed.Seconds(),
StartTimestamp: r.srFirst.RTPTimestampExt,
EndTimestamp: r.srNewest.RTPTimestampExt,
RtpClockTicks: rtpClockTicks,
DriftSamples: driftSamples,
DriftMs: (float64(driftSamples) * 1000) / float64(r.params.ClockRate),
ClockRate: float64(rtpClockTicks) / elapsed.Seconds(),
}
}
}
return
}
pkg/sfu/buffer/rtpstats_receiver.go
+132 -27
View File
@@ -28,6 +28,32 @@ import (
const (
cHistorySize = 4096
// RTCP Sender Reports are re-based to SFU time base so that all subscriber side
// can have the same time base (i. e. SFU time base). To convert publisher side
// RTCP Sender Reports to SFU timebase, a propagation delay is maintained.
// propagation_delay = time_of_report_reception - ntp_timestamp_in_report
//
// Propagation delay is adapted continuously. If it falls, adapt quickly to the
// lower value as that could be the real propagation delay. If it rises, adapt slowly
// as it might be a temporary change or slow drift. See below for handling of high deltas
// which could be a result of a path change.
cPropagationDelayFallFactor = float64(0.95)
cPropagationDelayRiseFactor = float64(0.05)
// do not adapt to small OR large (outlier) changes
cPropagationDelayDeltaThresholdMin = 5 * time.Millisecond
cPropagationDelayDeltaThresholdMaxFactor = 2
// To account for path changes mid-stream, if the delta of the propagation delay is consistently higher, reset.
// Reset at whichever of the below happens later.
//
// A long term version of delta of propagation delay is maintained and delta propagation delay exceeding
// a factor of the long term version is considered a sharp increase. That will trigger the start of the
// path change condition and if it persists, propagation delay will be reset.
cPropagationDelayDeltaAdaptationFactor = float64(0.05)
cPropagationDelayDeltaHighResetNumReports = 3
cPropagationDelayDeltaHighResetWait = 10 * time.Second
)
type RTPFlowState struct {
@@ -53,6 +79,11 @@ type RTPStatsReceiver struct {
history *protoutils.Bitmap[uint64]
propagationDelay time.Duration
longTermDeltaPropagationDelay time.Duration
propagationDelayDeltaHighCount int
propagationDelayDeltaHighStartTime time.Time
clockSkewCount int
outOfOrderSsenderReportCount int
}
@@ -251,9 +282,10 @@ func (r *RTPStatsReceiver) SetRtcpSenderReportData(srData *RTCPSenderReportData)
// prevent against extreme case of anachronous sender reports
if r.srNewest != nil && r.srNewest.NTPTimestamp > srData.NTPTimestamp {
r.logger.Infow(
"received anachronous sender report",
"last", r.srNewest.ToString(),
"current", srData.ToString(),
"received sender report, anachronous, dropping",
"first", r.srFirst,
"last", r.srNewest,
"current", srData,
)
return
}
@@ -264,30 +296,33 @@ func (r *RTPStatsReceiver) SetRtcpSenderReportData(srData *RTCPSenderReportData)
if (srData.RTPTimestamp-r.srNewest.RTPTimestamp) < (1<<31) && srData.RTPTimestamp < r.srNewest.RTPTimestamp {
tsCycles += (1 << 32)
}
if tsCycles >= (1 << 32) {
if (srData.RTPTimestamp-r.srNewest.RTPTimestamp) >= (1<<31) && srData.RTPTimestamp > r.srNewest.RTPTimestamp {
tsCycles -= (1 << 32)
}
}
}
srDataCopy := *srData
srDataCopy.RTPTimestampExt = uint64(srDataCopy.RTPTimestamp) + tsCycles
r.maybeAdjustFirstPacketTime(srDataCopy.RTPTimestamp, r.timestamp.GetStart())
if r.srNewest != nil && srDataCopy.RTPTimestampExt < r.srNewest.RTPTimestampExt {
// This can happen when a track is replaced with a null and then restored -
// i. e. muting replacing with null and unmute restoring the original track.
// Under such a condition reset the sender reports to start from this point.
// Resetting will ensure sample rate calculations do not go haywire due to negative time.
// Or it could be due bad report generation.
// In any case, ignore out-of-order reports.
if r.outOfOrderSsenderReportCount%10 == 0 {
r.logger.Infow(
"received sender report, out-of-order, resetting",
"last", r.srNewest.ToString(),
"current", srDataCopy.ToString(),
"received sender report, out-of-order, skipping",
"first", r.srFirst,
"last", r.srNewest,
"current", &srDataCopy,
"count", r.outOfOrderSsenderReportCount,
)
}
r.outOfOrderSsenderReportCount++
r.srFirst = nil
r.srNewest = nil
return
}
if r.srNewest != nil {
@@ -303,11 +338,15 @@ func (r *RTPStatsReceiver) SetRtcpSenderReportData(srData *RTCPSenderReportData)
(timeSinceFirst > 0.2 && math.Abs(float64(r.params.ClockRate)-calculatedClockRateFromFirst) > 0.2*float64(r.params.ClockRate)) {
if r.clockSkewCount%100 == 0 {
r.logger.Infow(
"clock rate skew",
"first", r.srFirst.ToString(),
"last", r.srNewest.ToString(),
"current", srDataCopy.ToString(),
"received sender report, clock skew",
"first", r.srFirst,
"last", r.srNewest,
"current", &srDataCopy,
"timeSinceFirst", timeSinceFirst,
"rtpDiffSinceFirst", rtpDiffSinceFirst,
"calculatedFirst", calculatedClockRateFromFirst,
"timeSinceLast", timeSinceLast,
"rtpDiffSinceLast", rtpDiffSinceLast,
"calculatedLast", calculatedClockRateFromLast,
"count", r.clockSkewCount,
)
@@ -316,26 +355,92 @@ func (r *RTPStatsReceiver) SetRtcpSenderReportData(srData *RTCPSenderReportData)
}
}
r.srNewest = &srDataCopy
var propagationDelay time.Duration
var deltaPropagationDelay time.Duration
getPropagationFields := func() []interface{} {
return []interface{}{
"propagationDelay", r.propagationDelay.String(),
"receivedPropagationDelay", propagationDelay.String(),
"longTermDeltaPropagationDelay", r.longTermDeltaPropagationDelay.String(),
"receivedDeltaPropagationDelay", deltaPropagationDelay.String(),
"deltaHighCount", r.propagationDelayDeltaHighCount,
"sinceDeltaHighStart", time.Since(r.propagationDelayDeltaHighStartTime).String(),
"first", r.srFirst,
"last", r.srNewest,
"current", &srDataCopy,
}
}
initPropagationDelay := func(pd time.Duration) {
r.propagationDelay = pd
r.longTermDeltaPropagationDelay = 0
r.propagationDelayDeltaHighCount = 0
r.propagationDelayDeltaHighStartTime = time.Time{}
}
ntpTime := srDataCopy.NTPTimestamp.Time()
propagationDelay = srDataCopy.At.Sub(ntpTime)
if r.srFirst == nil {
r.srFirst = &srDataCopy
initPropagationDelay(propagationDelay)
r.logger.Debugw("initializing propagation delay", getPropagationFields()...)
} else {
deltaPropagationDelay = propagationDelay - r.propagationDelay
if deltaPropagationDelay.Abs() > cPropagationDelayDeltaThresholdMin { // ignore small changes
if r.longTermDeltaPropagationDelay != 0 && deltaPropagationDelay > 0 && deltaPropagationDelay > r.longTermDeltaPropagationDelay*time.Duration(cPropagationDelayDeltaThresholdMaxFactor) {
r.logger.Debugw("sharp increase in propagation delay, skipping", getPropagationFields()...) // TODO-REMOVE
r.propagationDelayDeltaHighCount++
if r.propagationDelayDeltaHighStartTime.IsZero() {
r.propagationDelayDeltaHighStartTime = time.Now()
}
if r.propagationDelayDeltaHighCount >= cPropagationDelayDeltaHighResetNumReports && time.Since(r.propagationDelayDeltaHighStartTime) >= cPropagationDelayDeltaHighResetWait {
r.logger.Debugw("re-initializing propagation delay", append(getPropagationFields(), "newPropagationDelay", propagationDelay.String())...)
initPropagationDelay(propagationDelay)
}
} else {
r.propagationDelayDeltaHighCount = 0
r.propagationDelayDeltaHighStartTime = time.Time{}
if deltaPropagationDelay.Abs() > cPropagationDelayDeltaThresholdMin {
factor := cPropagationDelayFallFactor
if propagationDelay > r.propagationDelay {
factor = cPropagationDelayRiseFactor
}
fields := append(
getPropagationFields(),
"adjustedPropagationDelay", r.propagationDelay+time.Duration(factor*float64(propagationDelay-r.propagationDelay)),
) // TODO-REMOVE
r.logger.Debugw("adapting propagation delay", fields...) // TODO-REMOVE
r.propagationDelay += time.Duration(factor * float64(propagationDelay-r.propagationDelay))
}
}
} else {
r.propagationDelayDeltaHighCount = 0
r.propagationDelayDeltaHighStartTime = time.Time{}
}
if r.longTermDeltaPropagationDelay == 0 {
r.longTermDeltaPropagationDelay = deltaPropagationDelay
} else {
r.longTermDeltaPropagationDelay += time.Duration(cPropagationDelayDeltaAdaptationFactor * float64(deltaPropagationDelay-r.longTermDeltaPropagationDelay))
}
}
// adjust receive time to estimated propagation delay
srDataCopy.At = ntpTime.Add(r.propagationDelay)
r.srNewest = &srDataCopy
r.maybeAdjustFirstPacketTime(r.srNewest, 0, r.timestamp.GetExtendedStart())
}
func (r *RTPStatsReceiver) GetRtcpSenderReportData() (srFirst *RTCPSenderReportData, srNewest *RTCPSenderReportData) {
func (r *RTPStatsReceiver) GetRtcpSenderReportData() *RTCPSenderReportData {
r.lock.RLock()
defer r.lock.RUnlock()
if r.srFirst != nil {
srFirstCopy := *r.srFirst
srFirst = &srFirstCopy
if r.srNewest == nil {
return nil
}
if r.srNewest != nil {
srNewestCopy := *r.srNewest
srNewest = &srNewestCopy
}
return
srNewestCopy := *r.srNewest
return &srNewestCopy
}
func (r *RTPStatsReceiver) GetRtcpReceptionReport(ssrc uint32, proxyFracLost uint8, snapshotID uint32) *rtcp.ReceptionReport {
pkg/sfu/buffer/rtpstats_sender.go
+69 -103
View File
@@ -29,6 +29,8 @@ import (
const (
cSnInfoSize = 4096
cSnInfoMask = cSnInfoSize - 1
cSenderReportInitialWait = time.Second
)
type snInfoFlag byte
@@ -157,12 +159,8 @@ type RTPStatsSender struct {
nextSenderSnapshotID uint32
senderSnapshots []senderSnapshot
clockSkewCount int
outOfOrderSenderReportCount int
metadataCacheOverflowCount int
srFeedFirst *RTCPSenderReportData
srFeedNewest *RTCPSenderReportData
clockSkewCount int
metadataCacheOverflowCount int
}
func NewRTPStatsSender(params RTPStatsParams) *RTPStatsSender {
@@ -201,15 +199,6 @@ func (r *RTPStatsSender) Seed(from *RTPStatsSender) {
r.nextSenderSnapshotID = from.nextSenderSnapshotID
r.senderSnapshots = make([]senderSnapshot, cap(from.senderSnapshots))
copy(r.senderSnapshots, from.senderSnapshots)
if from.srFeedFirst != nil {
srFeedFirst := *from.srFeedFirst
r.srFeedFirst = &srFeedFirst
}
if from.srFeedNewest != nil {
srFeedNewest := *from.srFeedNewest
r.srFeedNewest = &srFeedNewest
}
}
func (r *RTPStatsSender) NewSnapshotId() uint32 {
@@ -533,6 +522,33 @@ func (r *RTPStatsSender) UpdateFromReceiverReport(rr rtcp.ReceptionReport) (rtt
s.maxJitter = r.jitterFromRR
}
if int64(extReceivedRRSN-s.extLastRRSN) < 0 || (extReceivedRRSN-s.extLastRRSN) > (1<<15) {
timeSinceLastRR := time.Since(r.lastRRTime)
if r.lastRRTime.IsZero() {
timeSinceLastRR = time.Since(r.startTime)
}
r.logger.Infow(
"rr interval too big, skipping",
"lastRRTime", r.lastRRTime.String(),
"lastRR", r.lastRR,
"timeSinceLastRR", timeSinceLastRR.String(),
"receivedRR", rr,
"extStartSN", r.extStartSN,
"extHighestSN", r.extHighestSN,
"extStartTS", r.extStartTS,
"extHighestTS", r.extHighestTS,
"extLastRRSN", s.extLastRRSN,
"firstTime", r.firstTime.String(),
"startTime", r.startTime.String(),
"highestTime", r.highestTime.String(),
"extReceivedRRSN", extReceivedRRSN,
"packetsInInterval", extReceivedRRSN-s.extLastRRSN,
"extHighestSNFromRR", r.extHighestSNFromRR,
"packetsLostFromRR", r.packetsLostFromRR,
)
continue
}
// on every RR, calculate delta since last RR using packet metadata cache
is := r.getIntervalStats(s.extLastRRSN+1, extReceivedRRSN+1, r.extHighestSN)
eis := &s.intervalStats
@@ -583,17 +599,15 @@ func (r *RTPStatsSender) LastReceiverReportTime() time.Time {
return r.lastRRTime
}
func (r *RTPStatsSender) MaybeAdjustFirstPacketTime(srFirst *RTCPSenderReportData, srNewest *RTCPSenderReportData, ts uint32) {
func (r *RTPStatsSender) MaybeAdjustFirstPacketTime(publisherSRData *RTCPSenderReportData, tsOffset uint64) {
r.lock.Lock()
defer r.lock.Unlock()
srFirstCopy := *srFirst
r.srFeedFirst = &srFirstCopy
if !r.initialized || publisherSRData == nil {
return
}
srNewestCopy := *srNewest
r.srFeedNewest = &srNewestCopy
r.maybeAdjustFirstPacketTime(ts, uint32(r.extStartTS))
r.maybeAdjustFirstPacketTime(publisherSRData, tsOffset, r.extStartTS)
}
func (r *RTPStatsSender) GetExpectedRTPTimestamp(at time.Time) (expectedTSExt uint64, err error) {
@@ -611,116 +625,68 @@ func (r *RTPStatsSender) GetExpectedRTPTimestamp(at time.Time) (expectedTSExt ui
return
}
func (r *RTPStatsSender) GetRtcpSenderReport(ssrc uint32, calculatedClockRate uint32) *rtcp.SenderReport {
func (r *RTPStatsSender) GetRtcpSenderReport(ssrc uint32, publisherSRData *RTCPSenderReportData, tsOffset uint64) *rtcp.SenderReport {
r.lock.Lock()
defer r.lock.Unlock()
if !r.initialized {
if !r.initialized || publisherSRData == nil {
return nil
}
// construct current time based on monotonic clock
timeSinceFirst := time.Since(r.firstTime)
now := r.firstTime.Add(timeSinceFirst)
timeSincePublisherSR := time.Since(publisherSRData.At)
now := publisherSRData.At.Add(timeSincePublisherSR)
nowNTP := mediatransportutil.ToNtpTime(now)
timeSinceHighest := now.Sub(r.highestTime)
nowRTPExt := r.extHighestTS + uint64(timeSinceHighest.Nanoseconds()*int64(r.params.ClockRate)/1e9)
nowRTPExtUsingTime := nowRTPExt
nowRTP := uint32(nowRTPExt)
// It is possible that publisher is pacing at a slower rate.
// That would make `highestTS` to be lagging the RTP time stamp in the RTCP Sender Report from publisher.
// Check for that using calculated clock rate and use the later time stamp if applicable.
var nowRTPExtUsingRate uint64
if calculatedClockRate != 0 {
nowRTPExtUsingRate = r.extStartTS + uint64(float64(calculatedClockRate)*timeSinceFirst.Seconds())
if nowRTPExtUsingRate > nowRTPExt {
nowRTPExt = nowRTPExtUsingRate
nowRTP = uint32(nowRTPExt)
}
}
nowRTPExt := publisherSRData.RTPTimestampExt - tsOffset + uint64(timeSincePublisherSR.Nanoseconds()*int64(r.params.ClockRate)/1e9)
srData := &RTCPSenderReportData{
NTPTimestamp: nowNTP,
RTPTimestamp: nowRTP,
RTPTimestamp: uint32(nowRTPExt),
RTPTimestampExt: nowRTPExt,
At: now,
}
getFields := func() []interface{} {
return []interface{}{
"first", r.srFirst,
"last", r.srNewest,
"curr", srData,
"feed", publisherSRData,
"tsOffset", tsOffset,
"timeNow", time.Now().String(),
"extStartTS", r.extStartTS,
"extHighestTS", r.extHighestTS,
"highestTime", r.highestTime.String(),
"timeSinceHighest", now.Sub(r.highestTime).String(),
"firstTime", r.firstTime.String(),
"timeSinceFirst", now.Sub(r.firstTime).String(),
"timeSincePublisherSR", timeSincePublisherSR.String(),
"nowRTPExt", nowRTPExt,
}
}
if r.srNewest != nil && nowRTPExt >= r.srNewest.RTPTimestampExt {
timeSinceLastReport := nowNTP.Time().Sub(r.srNewest.NTPTimestamp.Time())
rtpDiffSinceLastReport := nowRTPExt - r.srNewest.RTPTimestampExt
windowClockRate := float64(rtpDiffSinceLastReport) / timeSinceLastReport.Seconds()
if timeSinceLastReport.Seconds() > 0.2 && math.Abs(float64(r.params.ClockRate)-windowClockRate) > 0.2*float64(r.params.ClockRate) {
if r.clockSkewCount%10 == 0 {
r.logger.Infow(
"sending sender report, clock skew",
"first", r.srFirst.ToString(),
"last", r.srNewest.ToString(),
"curr", srData.ToString(),
"firstFeed", r.srFeedFirst.ToString(),
"lastFeed", r.srFeedNewest.ToString(),
"timeNow", time.Now().String(),
"extStartTS", r.extStartTS,
"extHighestTS", r.extHighestTS,
"highestTime", r.highestTime.String(),
"timeSinceHighest", timeSinceHighest.String(),
"firstTime", r.firstTime.String(),
"timeSinceFirst", timeSinceFirst.String(),
"nowRTPExtUsingTime", nowRTPExtUsingTime,
"calculatedClockRate", calculatedClockRate,
"nowRTPExtUsingRate", nowRTPExtUsingRate,
fields := append(
getFields(),
"timeSinceLastReport", timeSinceLastReport.String(),
"rtpDiffSinceLastReport", rtpDiffSinceLastReport,
"windowClockRate", windowClockRate,
"count", r.clockSkewCount,
)
r.logger.Infow("sending sender report, clock skew", fields...)
}
r.clockSkewCount++
}
}
if r.srNewest != nil && nowRTPExt < r.srNewest.RTPTimestampExt {
// If report being generated is behind, use the time difference and
// clock rate of codec to produce next report.
//
// Current report could be behind due to the following
// - Publisher pacing
// - Due to above, report from publisher side is ahead of packet timestamps.
// Note that report will map wall clock to timestamp at capture time and happens before the pacer.
// - Pause/Mute followed by resume, some combination of events that could
// result in this module not having calculated clock rate of publisher side.
// - When the above happens, current will be generated using highestTS which could be behind.
// That could end up behind the last report's timestamp in extreme cases
if r.outOfOrderSenderReportCount%10 == 0 {
r.logger.Infow(
"sending sender report, out-of-order, repairing",
"first", r.srFirst.ToString(),
"last", r.srNewest.ToString(),
"curr", srData.ToString(),
"firstFeed", r.srFeedFirst.ToString(),
"lastFeed", r.srFeedNewest.ToString(),
"timeNow", time.Now().String(),
"extStartTS", r.extStartTS,
"extHighestTS", r.extHighestTS,
"highestTime", r.highestTime.String(),
"timeSinceHighest", timeSinceHighest.String(),
"firstTime", r.firstTime.String(),
"timeSinceFirst", timeSinceFirst.String(),
"nowRTPExtUsingTime", nowRTPExtUsingTime,
"calculatedClockRate", calculatedClockRate,
"nowRTPExtUsingRate", nowRTPExtUsingRate,
"count", r.outOfOrderSenderReportCount,
)
}
r.outOfOrderSenderReportCount++
ntpDiffSinceLast := nowNTP.Time().Sub(r.srNewest.NTPTimestamp.Time())
nowRTPExt = r.srNewest.RTPTimestampExt + uint64(ntpDiffSinceLast.Seconds()*float64(r.params.ClockRate))
nowRTP = uint32(nowRTPExt)
srData.RTPTimestamp = nowRTP
srData.RTPTimestampExt = nowRTPExt
// If report being generated is behind the last report, skip it.
// Should not happen.
r.logger.Infow("sending sender report, out-of-order, skipping", getFields()...)
return nil
}
r.srNewest = srData
@@ -731,7 +697,7 @@ func (r *RTPStatsSender) GetRtcpSenderReport(ssrc uint32, calculatedClockRate ui
return &rtcp.SenderReport{
SSRC: ssrc,
NTPTime: uint64(nowNTP),
RTPTime: nowRTP,
RTPTime: uint32(nowRTPExt),
PacketCount: uint32(r.getTotalPacketsPrimary(r.extStartSN, r.extHighestSN) + r.packetsDuplicate + r.packetsPadding),
OctetCount: uint32(r.bytes + r.bytesDuplicate + r.bytesPadding),
}