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

rename.

Browse Source
This commit is contained in:
cloudwebrtc
2026-06-29 10:53:44 +08:00
parent 52ac105d64
commit 84443e66e2
6 changed files with 46 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/rtc/mediatrack.go
+3 -5
View File
@@ -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(
pkg/rtc/participant.go
+2 -4
View File
@@ -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)
pkg/sfu/buffer/buffer_base.go
+30 -30
View File
@@ -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
}
pkg/sfu/buffer/videoframecache_test.go
+1 -1
View File
@@ -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)
}
pkg/sfu/receiver.go
-10
View File
@@ -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
}
pkg/sfu/receiver_base.go
+10 -12
View File
@@ -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)
})