diff --git a/pkg/sfu/pacer/base.go b/pkg/sfu/pacer/base.go index fef7b413e..4b5da20c9 100644 --- a/pkg/sfu/pacer/base.go +++ b/pkg/sfu/pacer/base.go @@ -22,7 +22,13 @@ func NewBase(logger logger.Logger) *Base { } } -func (b *Base) SendPacket(p *Packet) error { +func (b *Base) SetInterval(_interval time.Duration) { +} + +func (b *Base) SetBitrate(_bitrate int) { +} + +func (b *Base) SendPacket(p *Packet) (int, error) { var sendingAt time.Time var err error defer func() { @@ -34,18 +40,19 @@ func (b *Base) SendPacket(p *Packet) error { sendingAt, err = b.writeRTPHeaderExtensions(p) if err != nil { b.logger.Errorw("writing rtp header extensions err", err) - return err + return 0, err } - _, err = p.WriteStream.WriteRTP(p.Header, p.Payload) + var written int + written, err = p.WriteStream.WriteRTP(p.Header, p.Payload) if err != nil { if !errors.Is(err, io.ErrClosedPipe) { b.logger.Errorw("write rtp packet failed", err) } - return err + return 0, err } - return nil + return written, nil } // writes RTP header extensions of track diff --git a/pkg/sfu/pacer/leaky_bucket.go b/pkg/sfu/pacer/leaky_bucket.go new file mode 100644 index 000000000..6e16f7f4e --- /dev/null +++ b/pkg/sfu/pacer/leaky_bucket.go @@ -0,0 +1,137 @@ +package pacer + +import ( + "sync" + "time" + + "github.com/gammazero/deque" + "github.com/livekit/protocol/logger" +) + +const ( + maxOvershootFactor = 2.0 +) + +type LeakyBucket struct { + *Base + + logger logger.Logger + + lock sync.RWMutex + packets deque.Deque[Packet] + interval time.Duration + bitrate int + isStopped bool +} + +func NewLeakyBucket(logger logger.Logger, interval time.Duration, bitrate int) *LeakyBucket { + l := &LeakyBucket{ + Base: NewBase(logger), + logger: logger, + interval: interval, + bitrate: bitrate, + } + l.packets.SetMinCapacity(9) + + go l.sendWorker() + return l +} + +func (l *LeakyBucket) SetInterval(interval time.Duration) { + l.lock.Lock() + defer l.lock.Unlock() + + l.interval = interval +} + +func (l *LeakyBucket) SetBitrate(bitrate int) { + l.lock.Lock() + defer l.lock.Unlock() + + l.bitrate = bitrate +} + +func (l *LeakyBucket) Stop() { + l.lock.Lock() + if l.isStopped { + l.lock.Unlock() + return + } + + l.isStopped = true + l.lock.Unlock() +} + +func (l *LeakyBucket) Enqueue(p Packet) { + l.lock.Lock() + defer l.lock.Unlock() + + if !l.isStopped { + l.packets.PushBack(p) + } +} + +func (l *LeakyBucket) sendWorker() { + l.lock.RLock() + interval := l.interval + bitrate := l.bitrate + l.lock.RUnlock() + + timer := time.NewTimer(interval) + overage := 0 + + for { + <-timer.C + + l.lock.RLock() + interval = l.interval + bitrate = l.bitrate + l.lock.RUnlock() + + // calculate number of bytes that can be sent in this interval + // adjusting for overage. + intervalBytes := int(interval.Seconds() * float64(bitrate) / 8.0) + maxOvershootBytes := int(float64(intervalBytes) * maxOvershootFactor) + toSendBytes := intervalBytes - overage + if toSendBytes < 0 { + // too much overage, wait for next interval + overage = -toSendBytes + timer.Reset(interval) + continue + } + + // do not allow too much overshoot in an interval + if toSendBytes > maxOvershootBytes { + toSendBytes = maxOvershootBytes + } + + for { + l.lock.Lock() + if l.isStopped { + l.lock.Unlock() + return + } + + if l.packets.Len() == 0 { + l.lock.Unlock() + // allow overshoot in next interval with shortage in this interval + overage = -toSendBytes + timer.Reset(interval) + break + } + p := l.packets.PopFront() + l.lock.Unlock() + + written, _ := l.Base.SendPacket(&p) + toSendBytes -= written + if toSendBytes < 0 { + // overage, wait for next interval + overage = -toSendBytes + timer.Reset(interval) + break + } + } + } +} + +// ------------------------------------------------ diff --git a/pkg/sfu/pacer/pacer.go b/pkg/sfu/pacer/pacer.go index 3be8a8a86..2288d8639 100644 --- a/pkg/sfu/pacer/pacer.go +++ b/pkg/sfu/pacer/pacer.go @@ -26,6 +26,9 @@ type Packet struct { type Pacer interface { Enqueue(p Packet) Stop() + + SetInterval(interval time.Duration) + SetBitrate(bitrate int) } // ------------------------------------------------