diff --git a/pkg/config/config.go b/pkg/config/config.go index 77592ea5d1b..9e5f9f97c20 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -89,8 +89,15 @@ type RTCConfig struct { // Deprecated: use PacketBufferSizeVideo and PacketBufferSizeAudio PacketBufferSize int `yaml:"packet_buffer_size,omitempty"` - // Number of packets to buffer for NACK - video + // Number of packets to buffer for NACK - video (max capacity) PacketBufferSizeVideo int `yaml:"packet_buffer_size_video,omitempty"` + // Initial bucket size for video packets. If set, the bucket starts at + // this size instead of InitPacketBufferSizeVideo (300). Useful for + // high-PPS H.264 streams under network delay where burst patterns + // exceed the default initial size but the PPS-based growth logic + // doesn't detect the need because it uses average PPS, not peak burst. + // If not set or zero, falls back to InitPacketBufferSizeVideo. + PacketBufferInitVideo int `yaml:"packet_buffer_init_video,omitempty"` // Number of packets to buffer for NACK - audio PacketBufferSizeAudio int `yaml:"packet_buffer_size_audio,omitempty"` diff --git a/pkg/rtc/config.go b/pkg/rtc/config.go index 94f982d9884..c4a7247e827 100644 --- a/pkg/rtc/config.go +++ b/pkg/rtc/config.go @@ -40,6 +40,7 @@ type WebRTCConfig struct { type ReceiverConfig struct { PacketBufferSizeVideo int + PacketBufferInitVideo int // Initial bucket size (0 = use InitPacketBufferSizeVideo) PacketBufferSizeAudio int } @@ -117,6 +118,7 @@ func NewWebRTCConfig(conf *config.Config) (*WebRTCConfig, error) { WebRTCConfig: *webRTCConfig, Receiver: ReceiverConfig{ PacketBufferSizeVideo: rtcConf.PacketBufferSizeVideo, + PacketBufferInitVideo: rtcConf.PacketBufferInitVideo, PacketBufferSizeAudio: rtcConf.PacketBufferSizeAudio, }, Publisher: publisherConfig, diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index 412f787b158..bcce1fc55a1 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -264,7 +264,7 @@ func NewRoom( participantRequestSources: make(map[livekit.ParticipantIdentity]routing.MessageSource), hasPublished: make(map[livekit.ParticipantIdentity]bool), agentParticpants: make(map[livekit.ParticipantIdentity]*agentJob), - bufferFactory: buffer.NewFactoryOfBufferFactory(config.Receiver.PacketBufferSizeVideo, config.Receiver.PacketBufferSizeAudio), + bufferFactory: buffer.NewFactoryOfBufferFactory(config.Receiver.PacketBufferSizeVideo, config.Receiver.PacketBufferInitVideo, config.Receiver.PacketBufferSizeAudio), batchedUpdates: make(map[livekit.ParticipantIdentity]*participantUpdate), closed: make(chan struct{}), trailer: []byte(utils.RandomSecret()), diff --git a/pkg/sfu/buffer/buffer.go b/pkg/sfu/buffer/buffer.go index 8e79e0d316b..6108a6d9a0f 100644 --- a/pkg/sfu/buffer/buffer.go +++ b/pkg/sfu/buffer/buffer.go @@ -77,6 +77,7 @@ type Buffer struct { bucket *bucket.Bucket[uint64] nacker *nack.NackQueue maxVideoPkts int + initVideoPkts int maxAudioPkts int codecType webrtc.RTPCodecType payloadType uint8 @@ -142,12 +143,13 @@ type Buffer struct { } // NewBuffer constructs a new Buffer -func NewBuffer(ssrc uint32, maxVideoPkts, maxAudioPkts int) *Buffer { +func NewBuffer(ssrc uint32, maxVideoPkts, initVideoPkts, maxAudioPkts int) *Buffer { l := logger.GetLogger() // will be reset with correct context via SetLogger b := &Buffer{ - mediaSSRC: ssrc, - maxVideoPkts: maxVideoPkts, - maxAudioPkts: maxAudioPkts, + mediaSSRC: ssrc, + maxVideoPkts: maxVideoPkts, + initVideoPkts: initVideoPkts, + maxAudioPkts: maxAudioPkts, snRangeMap: utils.NewRangeMap[uint64, uint64](100), pliThrottle: int64(500 * time.Millisecond), logger: l.WithComponent(sutils.ComponentPub).WithComponent(sutils.ComponentSFU), @@ -260,7 +262,12 @@ func (b *Buffer) Bind(params webrtc.RTPParameters, codec webrtc.RTPCodecCapabili case strings.HasPrefix(b.mime, "video/"): b.codecType = webrtc.RTPCodecTypeVideo - b.bucket = bucket.NewBucket[uint64](InitPacketBufferSizeVideo) + // Use initVideoPkts from config if set, otherwise fall back to default. + initSize := InitPacketBufferSizeVideo + if b.initVideoPkts > 0 { + initSize = b.initVideoPkts + } + b.bucket = bucket.NewBucket[uint64](initSize) if b.frameRateCalculator[0] == nil { if strings.EqualFold(codec.MimeType, webrtc.MimeTypeVP8) { b.frameRateCalculator[0] = NewFrameRateCalculatorVP8(b.clockRate, b.logger) diff --git a/pkg/sfu/buffer/factory.go b/pkg/sfu/buffer/factory.go index ae0bb073305..1b154c52451 100644 --- a/pkg/sfu/buffer/factory.go +++ b/pkg/sfu/buffer/factory.go @@ -23,12 +23,14 @@ import ( type FactoryOfBufferFactory struct { trackingPacketsVideo int + initPacketsVideo int trackingPacketsAudio int } -func NewFactoryOfBufferFactory(trackingPacketsVideo int, trackingPacketsAudio int) *FactoryOfBufferFactory { +func NewFactoryOfBufferFactory(trackingPacketsVideo int, initPacketsVideo int, trackingPacketsAudio int) *FactoryOfBufferFactory { return &FactoryOfBufferFactory{ trackingPacketsVideo: trackingPacketsVideo, + initPacketsVideo: initPacketsVideo, trackingPacketsAudio: trackingPacketsAudio, } } @@ -36,6 +38,7 @@ func NewFactoryOfBufferFactory(trackingPacketsVideo int, trackingPacketsAudio in func (f *FactoryOfBufferFactory) CreateBufferFactory() *Factory { return &Factory{ trackingPacketsVideo: f.trackingPacketsVideo, + initPacketsVideo: f.initPacketsVideo, trackingPacketsAudio: f.trackingPacketsAudio, rtpBuffers: make(map[uint32]*Buffer), rtcpReaders: make(map[uint32]*RTCPReader), @@ -46,6 +49,7 @@ func (f *FactoryOfBufferFactory) CreateBufferFactory() *Factory { type Factory struct { sync.RWMutex trackingPacketsVideo int + initPacketsVideo int trackingPacketsAudio int rtpBuffers map[uint32]*Buffer rtcpReaders map[uint32]*RTCPReader @@ -72,7 +76,7 @@ func (f *Factory) GetOrNew(packetType packetio.BufferPacketType, ssrc uint32) io if reader, ok := f.rtpBuffers[ssrc]; ok { return reader } - buffer := NewBuffer(ssrc, f.trackingPacketsVideo, f.trackingPacketsAudio) + buffer := NewBuffer(ssrc, f.trackingPacketsVideo, f.initPacketsVideo, f.trackingPacketsAudio) f.rtpBuffers[ssrc] = buffer for repair, base := range f.rtxPair { if repair == ssrc {