From 043e0fd52607a6157dbd6cb35efc854f29193ac1 Mon Sep 17 00:00:00 2001 From: ddt Date: Thu, 16 Apr 2026 13:48:56 -0600 Subject: [PATCH] feat: add packet_buffer_init_video config parameter MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add a new YAML config parameter `packet_buffer_init_video` that sets the initial size of the SFU's circular packet buffer for video, independently of the existing `packet_buffer_size_video` (max capacity). Under degraded networks (WiFi/5G congestion), H.264 packets arrive in bursts of 1,000-2,500 that exceed the default initial bucket size (300, grown to 600). The existing growth logic uses average PPS which doesn't reflect burst peaks, so the bucket never grows large enough. This new parameter allows deployments to start the bucket at the right size. If not set or zero, falls back to InitPacketBufferSizeVideo (300) — preserving existing behavior for all current users. Co-Authored-By: Claude Opus 4.6 (1M context) --- pkg/config/config.go | 9 ++++++++- pkg/rtc/config.go | 2 ++ pkg/rtc/room.go | 2 +- pkg/sfu/buffer/buffer.go | 17 ++++++++++++----- pkg/sfu/buffer/factory.go | 8 ++++++-- 5 files changed, 29 insertions(+), 9 deletions(-) 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 {