dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-08-29 07:39:09 +00:00
Code Issues Packages Projects Releases Wiki Activity

use dynamic bucket size (#2524)

Browse Source
This commit is contained in:
cnderrauber
2024-02-28 16:24:23 +08:00
committed by GitHub
parent 6519c7596a
commit a435368278
5 changed files with 79 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go.mod
+1 -1
View File
@@ -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
go.sum
+2 -2
View File
@@ -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=
pkg/sfu/buffer/buffer.go
+46 -22
View File
@@ -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) {
pkg/sfu/buffer/buffer_test.go
+15 -23
View File
@@ -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)
}
}
pkg/sfu/buffer/factory.go
+15 -27
View File
@@ -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 {