From a4353682785d9b8568d5f728d6fd8b353496fc12 Mon Sep 17 00:00:00 2001 From: cnderrauber Date: Wed, 28 Feb 2024 16:24:23 +0800 Subject: [PATCH] use dynamic bucket size (#2524) --- go.mod | 2 +- go.sum | 4 +-- pkg/sfu/buffer/buffer.go | 68 +++++++++++++++++++++++------------ pkg/sfu/buffer/buffer_test.go | 38 ++++++++------------ pkg/sfu/buffer/factory.go | 42 ++++++++-------------- 5 files changed, 79 insertions(+), 75 deletions(-) diff --git a/go.mod b/go.mod index 2e84a9357..1552338a6 100644 --- a/go.mod +++ b/go.mod @@ -18,7 +18,7 @@ require ( github.com/hashicorp/golang-lru/v2 v2.0.7 github.com/jxskiss/base62 v1.1.0 github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1 - github.com/livekit/mediatransportutil v0.0.0-20240206082112-9bf41dcbce76 + github.com/livekit/mediatransportutil v0.0.0-20240228075855-6fbf3be6f6ef github.com/livekit/protocol v1.10.0 github.com/livekit/psrpc v0.5.3-0.20240227154351-b7f99eaaf7b3 github.com/mackerelio/go-osstat v0.2.4 diff --git a/go.sum b/go.sum index 1d8c2f69c..1f87a241b 100644 --- a/go.sum +++ b/go.sum @@ -128,8 +128,8 @@ github.com/lithammer/shortuuid/v4 v4.0.0 h1:QRbbVkfgNippHOS8PXDkti4NaWeyYfcBTHtw github.com/lithammer/shortuuid/v4 v4.0.0/go.mod h1:Zs8puNcrvf2rV9rTH51ZLLcj7ZXqQI3lv67aw4KiB1Y= github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1 h1:jm09419p0lqTkDaKb5iXdynYrzB84ErPPO4LbRASk58= github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ= -github.com/livekit/mediatransportutil v0.0.0-20240206082112-9bf41dcbce76 h1:Zw88krOHni51OzDUlrduYb3m7VcsaKw06TnnDhsQpjg= -github.com/livekit/mediatransportutil v0.0.0-20240206082112-9bf41dcbce76/go.mod h1:GBzn9xL+mivI1pW+tyExcKgbc0VOc29I9yJsNcAVaAc= +github.com/livekit/mediatransportutil v0.0.0-20240228075855-6fbf3be6f6ef h1:Db/UItb+Cvm1trBRJiEZOdRSyss+LDY4e8gU9aE4GRc= +github.com/livekit/mediatransportutil v0.0.0-20240228075855-6fbf3be6f6ef/go.mod h1:GBzn9xL+mivI1pW+tyExcKgbc0VOc29I9yJsNcAVaAc= github.com/livekit/protocol v1.10.0 h1:HKBCitK7+Nuezktqv/h9h5AOllttsmNnZFpwIlAIRRw= github.com/livekit/protocol v1.10.0/go.mod h1:NnlGwusu/SvwBxFe9Fpi9P2IKCA/V+kIqObZ3USWq0g= github.com/livekit/psrpc v0.5.3-0.20240227154351-b7f99eaaf7b3 h1:bvjzDR+Rvdf3JgzQMtLiGVHBQ8KoOWM7x7sHj79jevQ= diff --git a/pkg/sfu/buffer/buffer.go b/pkg/sfu/buffer/buffer.go index 2f920dd02..06c27b6bf 100644 --- a/pkg/sfu/buffer/buffer.go +++ b/pkg/sfu/buffer/buffer.go @@ -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 + ppsSnapshortId 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.ppsSnapshortId = 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 { @@ -776,6 +776,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.deltaStatsSnapshotId) + 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) { diff --git a/pkg/sfu/buffer/buffer_test.go b/pkg/sfu/buffer/buffer_test.go index 7f685186c..2125ca907 100644 --- a/pkg/sfu/buffer/buffer_test.go +++ b/pkg/sfu/buffer/buffer_test.go @@ -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) + } + +} diff --git a/pkg/sfu/buffer/factory.go b/pkg/sfu/buffer/factory.go index 2520c6cce..2785f95ae 100644 --- a/pkg/sfu/buffer/factory.go +++ b/pkg/sfu/buffer/factory.go @@ -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(trackingPacketsVideo int, trackingPacketsAudio int) *FactoryOfBufferFactory { return &FactoryOfBufferFactory{ - videoPool: &sync.Pool{ - New: func() interface{} { - b := make([]byte, trackingPacketsVideo*bucket.MaxPktSize) - return &b - }, - }, - audioPool: &sync.Pool{ - New: func() interface{} { - b := make([]byte, trackingPacketsAudio*bucket.MaxPktSize) - 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 {