diff --git a/pkg/rtc/mediatrack.go b/pkg/rtc/mediatrack.go index 431f6aaa1..600bfe1f1 100644 --- a/pkg/rtc/mediatrack.go +++ b/pkg/rtc/mediatrack.go @@ -99,8 +99,7 @@ type MediaTrackParams struct { EnableRTPStreamRestartDetection bool UpdateTrackInfoByVideoSizeChange bool ForceBackupCodecPolicySimulcast bool - EnableVideoCaching bool - VideoCachingMaxDuration time.Duration + VideoFrameCachingDuration time.Duration } func NewMediaTrack(params MediaTrackParams, ti *livekit.TrackInfo) *MediaTrack { @@ -416,9 +415,8 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track sfu.TrackRe sfu.WithForwardStats(t.params.ForwardStats), sfu.WithEnableRTPStreamRestartDetection(t.params.EnableRTPStreamRestartDetection), } - if t.params.EnableVideoCaching { - // VideoCachingMaxDuration <= 0 falls back to sfu.DefaultVideoFrameCacheMaxDuration in WithVideoFrameCache - receiverOpts = append(receiverOpts, sfu.WithVideoFrameCache(t.params.VideoCachingMaxDuration)) + if t.params.VideoFrameCachingDuration > 0 { + receiverOpts = append(receiverOpts, sfu.WithVideoFrameCache(t.params.VideoFrameCachingDuration)) } newWR := sfu.NewWebRTCReceiver( diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index db8de5823..c85901bac 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -226,8 +226,7 @@ type ParticipantParams struct { ForceBackupCodecPolicySimulcast bool DisableTransceiverReuseForE2EE bool EnableStartAtDesiredQuality bool - EnableVideoCaching bool - VideoCachingMaxDuration time.Duration + VideoFrameCachingDuration time.Duration } type ParticipantImpl struct { @@ -3384,8 +3383,7 @@ func (p *ParticipantImpl) addMediaTrack(signalCid string, ti *livekit.TrackInfo) EnableRTPStreamRestartDetection: p.params.EnableRTPStreamRestartDetection, UpdateTrackInfoByVideoSizeChange: p.params.UseOneShotSignallingMode, ForceBackupCodecPolicySimulcast: p.params.ForceBackupCodecPolicySimulcast, - EnableVideoCaching: p.params.EnableVideoCaching, - VideoCachingMaxDuration: p.params.VideoCachingMaxDuration, + VideoFrameCachingDuration: p.params.VideoFrameCachingDuration, }, ti) mt.OnSubscribedMaxQualityChange(p.onSubscribedMaxQualityChange) diff --git a/pkg/sfu/buffer/buffer_base.go b/pkg/sfu/buffer/buffer_base.go index 6e5ae9fec..49316fd07 100644 --- a/pkg/sfu/buffer/buffer_base.go +++ b/pkg/sfu/buffer/buffer_base.go @@ -204,11 +204,7 @@ type BufferBase struct { ddExtID uint8 ddParser *DependencyDescriptorParser - // video frame cache: mark the most recent key frame so the current group-of-pictures can be read - // back from the retransmit bucket (see EnableVideoFrameCache / GetVideoFrameCache). No packets are copied here - only - // the key-frame boundary is tracked. - videoFrameCacheEnabled bool - videoFrameCacheMaxDuration time.Duration // optional bound on key-frame interval (0 = bucket-bounded only) + videoFrameCacheMaxDuration time.Duration // optional bound on key-frame interval (0 = disabled) videoFrameCacheHasKeyFrame bool videoFrameCacheKeyFrameESN uint64 // ext sequence number of the current video frame cache group's first key-frame packet videoFrameCacheKeyFrameETS uint64 // ext timestamp of the current video frame cache group's key frame @@ -239,11 +235,12 @@ func NewBufferBase(params BufferBaseParams) *BufferBase { l = l.WithValues("ssrc", params.SSRC) b := &BufferBase{ - params: params, - lastBucketCapCheckAt: mono.UnixNano(), - snRangeMap: utils.NewRangeMap[uint64, uint64](100), - pliThrottle: int64(500 * time.Millisecond), - logger: l, + params: params, + lastBucketCapCheckAt: mono.UnixNano(), + snRangeMap: utils.NewRangeMap[uint64, uint64](100), + pliThrottle: int64(500 * time.Millisecond), + logger: l, + videoFrameCacheMaxDuration: 0, } b.readCond = sync.NewCond(&b.RWMutex) b.extPackets.SetBaseCap(128) @@ -659,19 +656,20 @@ func (b *BufferBase) ReadExtended(buf []byte) (*ExtPacket, error) { } } -// EnableVideoFrameCache turns on video frame cache tracking for this (video) buffer: the most recent key frame is marked -// so the current group-of-pictures can be read back from the retransmit bucket via GetVideoFrameCache. No -// packets are copied - only the key-frame boundary is tracked. maxDuration bounds the served -// key-frame interval AND drives the retransmit bucket to retain that much history (see -// maybeGrowBucket), so the key frame is not evicted before it can be read; <= 0 keeps the default -// ~1s retransmit window (and bounds the video frame cache group to it). No-op for audio buffers. +// EnableVideoFrameCache turns on video frame cache tracking for this (video) buffer: the most recent +// key frame is marked so the current group-of-pictures can be read back from the retransmit bucket +// via GetVideoFrameCache. No packets are copied - only the key-frame boundary is tracked. maxDuration +// bounds the served key-frame interval AND drives the retransmit bucket to retain that much history +// (see maybeGrowBucket), so the key frame is not evicted before it can be read; maxDuration <= 0 +// disables the cache. func (b *BufferBase) EnableVideoFrameCache(maxDuration time.Duration) { b.Lock() defer b.Unlock() - b.videoFrameCacheEnabled = true b.videoFrameCacheMaxDuration = maxDuration - b.logger.Debugw("video frame cache enabled on buffer", "maxDuration", maxDuration) + if maxDuration > 0 { + b.logger.Debugw("video frame cache enabled on buffer", "maxDuration", maxDuration) + } } // markVideoFrameCacheLocked records the key-frame boundary of the current video frame cache group and tracks its span. Caller holds @@ -714,24 +712,26 @@ func (b *BufferBase) GetVideoFrameCache() ([]*ExtPacket, bool) { b.Lock() defer b.Unlock() - if !b.videoFrameCacheEnabled || !b.videoFrameCacheHasKeyFrame || b.bucket == nil { + videoFrameCacheEnabled := b.videoFrameCacheMaxDuration > 0 + + if !videoFrameCacheEnabled || !b.videoFrameCacheHasKeyFrame || b.bucket == nil { b.logger.Debugw( "video frame cache miss: not ready", - "videoFrameCacheEnabled", b.videoFrameCacheEnabled, + "videoFrameCacheMaxDuration", b.videoFrameCacheMaxDuration, "videoFrameCacheHasKeyFrame", b.videoFrameCacheHasKeyFrame, "hasBucket", b.bucket != nil, ) return nil, false } - if b.videoFrameCacheMaxDuration > 0 && b.clockRate > 0 { + if videoFrameCacheEnabled && b.clockRate > 0 { maxTicks := uint64(b.videoFrameCacheMaxDuration.Seconds() * float64(b.clockRate)) if b.videoFrameCacheLatestTS > b.videoFrameCacheKeyFrameETS+maxTicks { // key-frame interval longer than the bound - too old to serve a complete replay b.logger.Debugw( "video frame cache miss: key-frame interval exceeds bound", - "keyFrameTS", b.videoFrameCacheKeyFrameETS, - "latestTS", b.videoFrameCacheLatestTS, + "keyFrameETS", b.videoFrameCacheKeyFrameETS, + "latestETS", b.videoFrameCacheLatestTS, "spanTicks", b.videoFrameCacheLatestTS-b.videoFrameCacheKeyFrameETS, "maxTicks", maxTicks, ) @@ -741,16 +741,16 @@ func (b *BufferBase) GetVideoFrameCache() ([]*ExtPacket, bool) { headESN := uint64(b.bucket.HeadSequenceNumber()) if headESN < b.videoFrameCacheKeyFrameESN { - b.logger.Debugw("video frame cache miss: head behind key frame", "headESN", headESN, "keyFrameSN", b.videoFrameCacheKeyFrameESN) + b.logger.Debugw("video frame cache miss: head behind key frame", "headESN", headESN, "keyCacheFrameESN", b.videoFrameCacheKeyFrameESN) return nil, false } pkts := b.reconstructPacketsLocked(b.videoFrameCacheKeyFrameESN, headESN) // the key frame itself must be present (its first packet), otherwise the video frame cache group cannot be served if len(pkts) == 0 || pkts[0].ExtSequenceNumber != b.videoFrameCacheKeyFrameESN { - var firstSN uint64 + var firstESN uint64 if len(pkts) > 0 { - firstSN = pkts[0].ExtSequenceNumber + firstESN = pkts[0].ExtSequenceNumber } b.logger.Debugw( "video frame cache miss: key frame evicted from bucket", @@ -758,7 +758,7 @@ func (b *BufferBase) GetVideoFrameCache() ([]*ExtPacket, bool) { "headESN", headESN, "bucketCapacity", b.bucket.Capacity(), "reconstructed", len(pkts), - "firstSN", firstSN, + "firstESN", firstESN, ) return nil, false } @@ -773,7 +773,7 @@ func (b *BufferBase) GetPacketsAfter(afterESN uint64) ([]*ExtPacket, bool) { b.Lock() defer b.Unlock() - if !b.videoFrameCacheEnabled || !b.videoFrameCacheHasKeyFrame || b.bucket == nil { + if b.videoFrameCacheMaxDuration <= 0 || !b.videoFrameCacheHasKeyFrame || b.bucket == nil { return nil, false } headESN := uint64(b.bucket.HeadSequenceNumber()) @@ -1089,7 +1089,7 @@ func (b *BufferBase) HandleIncomingPacketLocked( return 0, errors.New("could not get ext packet") } b.extPackets.PushBack(ep) - if b.videoFrameCacheEnabled && b.codecType == webrtc.RTPCodecTypeVideo { + if b.videoFrameCacheMaxDuration > 0 && b.codecType == webrtc.RTPCodecTypeVideo { b.markVideoFrameCacheLocked(ep) } b.readCond.Broadcast() @@ -1448,7 +1448,7 @@ func (b *BufferBase) maybeGrowBucket(now int64) { // duration (the key frame must survive until GetVideoFrameCache reads it), which is more than the normal // ~1s retransmit window. In that case the target / cap are computed below from pps; otherwise // keep the original fast path. - videoFrameCacheSizing := b.codecType == webrtc.RTPCodecTypeVideo && b.videoFrameCacheEnabled && b.videoFrameCacheMaxDuration > 0 + videoFrameCacheSizing := b.codecType == webrtc.RTPCodecTypeVideo && b.videoFrameCacheMaxDuration > 0 if cap >= maxPkts && !videoFrameCacheSizing { return } diff --git a/pkg/sfu/buffer/videoframecache_test.go b/pkg/sfu/buffer/videoframecache_test.go index 0c3271085..754a17523 100644 --- a/pkg/sfu/buffer/videoframecache_test.go +++ b/pkg/sfu/buffer/videoframecache_test.go @@ -229,5 +229,5 @@ func TestVideoFrameCacheDisabled(t *testing.T) { // audio buffers never enable the cache a := &BufferBase{codecType: webrtc.RTPCodecTypeAudio, clockRate: 48000, logger: logger.GetLogger()} a.EnableVideoFrameCache(0) - require.False(t, a.videoFrameCacheEnabled) + require.False(t, a.videoFrameCacheMaxDuration > 0) } diff --git a/pkg/sfu/receiver.go b/pkg/sfu/receiver.go index 72e11ae57..a3aa2722d 100644 --- a/pkg/sfu/receiver.go +++ b/pkg/sfu/receiver.go @@ -93,18 +93,8 @@ func WithForwardStats(forwardStats *ForwardStats) ReceiverOpts { } } -// DefaultVideoFrameCacheMaxDuration bounds the cached video frame cache group (and sizes the retransmit bucket) when the video -// frame cache is enabled via WithVideoFrameCache. -const DefaultVideoFrameCacheMaxDuration = 1 * time.Second - -// WithVideoFrameCache enables the video frame cache on the receiver so a newly added down track is -// bootstrapped by replaying the cached video frame cache group instead of triggering a PLI. maxDuration bounds the -// cached video frame (<= 0 uses DefaultVideoFrameCacheMaxDuration). No-op for audio receivers. func WithVideoFrameCache(maxDuration time.Duration) ReceiverOpts { return func(w *WebRTCReceiver) *WebRTCReceiver { - if maxDuration <= 0 { - maxDuration = DefaultVideoFrameCacheMaxDuration - } w.ReceiverBase.EnableVideoFrameCache(maxDuration) return w } diff --git a/pkg/sfu/receiver_base.go b/pkg/sfu/receiver_base.go index 2b4a6f958..12e25c635 100644 --- a/pkg/sfu/receiver_base.go +++ b/pkg/sfu/receiver_base.go @@ -215,9 +215,6 @@ type ReceiverBase struct { videoSizes [buffer.DefaultMaxLayerSpatial + 1]codec.VideoSize onVideoSizeChanged func() - // video frame cache: when enabled, buffers retain the current frame cache and a newly added down track is - // bootstrapped by replaying it (see EnableVideoFrameCache / replayVideoFrameCache) - videoFrameCacheEnabled bool videoFrameCacheMaxDuration time.Duration // counts of subscribers bootstrapped from the video frame cache (hit) vs. those that fell back to a PLI @@ -244,11 +241,12 @@ type ReceiverBase struct { func NewReceiverBase(params ReceiverBaseParams, trackInfo *livekit.TrackInfo, codecState ReceiverCodecState) *ReceiverBase { r := &ReceiverBase{ - params: params, - codecState: codecState, - isRED: mime.IsMimeTypeStringRED(params.Codec.MimeType), - trackInfo: utils.CloneProto(trackInfo), - videoLayerMode: buffer.GetVideoLayerModeForMimeType(mime.NormalizeMimeType(params.Codec.MimeType), trackInfo), + params: params, + codecState: codecState, + isRED: mime.IsMimeTypeStringRED(params.Codec.MimeType), + trackInfo: utils.CloneProto(trackInfo), + videoLayerMode: buffer.GetVideoLayerModeForMimeType(mime.NormalizeMimeType(params.Codec.MimeType), trackInfo), + videoFrameCacheMaxDuration: 0, } r.downTrackSpreader = sfuutils.NewDownTrackSpreader[TrackSender](sfuutils.DownTrackSpreaderParams{ @@ -570,7 +568,7 @@ func (r *ReceiverBase) AddDownTrack(track TrackSender) error { track.UpTrackMaxPublishedLayerChange(r.streamTrackerManager.GetMaxPublishedLayer()) track.UpTrackMaxTemporalLayerSeenChange(r.streamTrackerManager.GetMaxTemporalLayerSeen()) - if r.videoFrameCacheEnabled && r.Kind() == webrtc.RTPCodecTypeVideo { + if r.videoFrameCacheMaxDuration > 0 && r.Kind() == webrtc.RTPCodecTypeVideo { // Bootstrap the new track from the cached video frame before joining the live broadcast: replay the // cached video frame (paced) and catch up to the live forwarding point, then Store so the live feed takes // over seamlessly. Held out of the broadcast during replay to avoid interleaving replayed and @@ -606,7 +604,6 @@ func (r *ReceiverBase) EnableVideoFrameCache(maxDuration time.Duration) { } r.bufferMu.Lock() - r.videoFrameCacheEnabled = true r.videoFrameCacheMaxDuration = maxDuration buffers := r.buffers r.bufferMu.Unlock() @@ -655,7 +652,7 @@ func (r *ReceiverBase) replayVideoFrameCache(track TrackSender) { r.params.Logger.Debugw( "subscriber bootstrap: video frame cache miss, falling back to PLI", "subscriberID", track.SubscriberID(), - "videoFrameCacheEnabled", r.videoFrameCacheEnabled, + "videoFrameCacheMaxDuration", r.videoFrameCacheMaxDuration, "videoFrameCacheHitCount", r.videoFrameCacheHitCount.Load(), "videoFrameCacheMissCount", missCount, ) @@ -954,9 +951,10 @@ func (r *ReceiverBase) setupBuffer(buff buffer.BufferProvider, layer int32, rtt if r.Kind() == webrtc.RTPCodecTypeVideo && layer == 0 { buff.OnCodecChange(r.handleCodecChange) } - if r.videoFrameCacheEnabled && r.Kind() == webrtc.RTPCodecTypeVideo { + if r.videoFrameCacheMaxDuration > 0 && r.Kind() == webrtc.RTPCodecTypeVideo { buff.EnableVideoFrameCache(r.videoFrameCacheMaxDuration) } + buff.OnStreamRestart(func(reason string) { r.restartInternal(reason, true) })