dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-08-28 02:54:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

Consolidating PLI throttle (#419)

Browse Source 
* Consolidating PLI throttle

Use the throttler in `sfu.WebRTCReceiver`.

Does change shape of config object.

* Move PLIThrottleConfig to sfu.WebRTCReceiver

* fix test compile

* Cleaning up unused stuff

* readability improvement
This commit is contained in:
Raja Subramanian
2022-02-08 22:50:43 +05:30
committed by GitHub
parent fee3009853
commit 3117547d60
7 changed files with 62 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/rtc/mediatrack.go
+9 -8
View File
@@ -48,13 +48,14 @@ type MediaTrackParams struct {
ParticipantID livekit.ParticipantID
ParticipantIdentity livekit.ParticipantIdentity
// channel to send RTCP packets to the source
RTCPChan chan []rtcp.Packet
BufferFactory *buffer.Factory
ReceiverConfig ReceiverConfig
SubscriberConfig DirectionConfig
AudioConfig config.AudioConfig
Telemetry telemetry.TelemetryService
Logger logger.Logger
RTCPChan chan []rtcp.Packet
BufferFactory *buffer.Factory
ReceiverConfig ReceiverConfig
SubscriberConfig DirectionConfig
PLIThrottleConfig config.PLIThrottleConfig
AudioConfig config.AudioConfig
Telemetry telemetry.TelemetryService
Logger logger.Logger
}
func NewMediaTrack(track *webrtc.TrackRemote, params MediaTrackParams) *MediaTrack {
@@ -169,7 +170,7 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.Tra
track,
t.PublisherID(),
t.params.Logger,
sfu.WithPliThrottle(0),
sfu.WithPliThrottle(t.params.PLIThrottleConfig),
sfu.WithLoadBalanceThreshold(20),
sfu.WithStreamTrackers(),
)
pkg/rtc/participant.go
+5 -29
View File
@@ -49,7 +49,7 @@ type ParticipantParams struct {
AudioConfig config.AudioConfig
ProtocolVersion types.ProtocolVersion
Telemetry telemetry.TelemetryService
ThrottleConfig config.PLIThrottleConfig
PLIThrottleConfig config.PLIThrottleConfig
CongestionControlConfig config.CongestionControlConfig
EnabledCodecs []*livekit.Codec
Hidden bool
@@ -82,8 +82,7 @@ type ParticipantImpl struct {
// JSON encoded metadata to pass to clients
metadata string
rtcpCh chan []rtcp.Packet
pliThrottle *pliThrottle
rtcpCh chan []rtcp.Packet
// hold reference for MediaTrack
twcc *twcc.Responder
@@ -127,7 +126,6 @@ func NewParticipant(params ParticipantParams, perms *livekit.ParticipantPermissi
p := &ParticipantImpl{
params: params,
rtcpCh: make(chan []rtcp.Packet, 50),
pliThrottle: newPLIThrottle(params.ThrottleConfig),
pendingTracks: make(map[string]*pendingTrackInfo),
subscribedTracks: make(map[livekit.TrackID]types.SubscribedTrack),
subscribedTracksSettings: make(map[livekit.TrackID]*livekit.UpdateTrackSettings),
@@ -1426,6 +1424,7 @@ func (p *ParticipantImpl) mediaTrackReceived(track *webrtc.TrackRemote, rtpRecei
Telemetry: p.params.Telemetry,
Logger: LoggerWithTrack(p.params.Logger, livekit.TrackID(ti.Sid)),
SubscriberConfig: p.params.Config.Subscriber,
PLIThrottleConfig: p.params.PLIThrottleConfig,
})
for ssrc, info := range p.params.SimTracks {
@@ -1444,7 +1443,6 @@ func (p *ParticipantImpl) mediaTrackReceived(track *webrtc.TrackRemote, rtpRecei
}
ssrc := uint32(track.SSRC())
p.pliThrottle.addTrack(ssrc, track.RID())
if p.twcc == nil {
p.twcc = twcc.NewTransportWideCCResponder(ssrc)
p.twcc.OnFeedback(func(pkt rtcp.RawPacket) {
@@ -1548,30 +1546,8 @@ func (p *ParticipantImpl) rtcpSendWorker() {
return
}
fwdPkts := make([]rtcp.Packet, 0, len(pkts))
for _, pkt := range pkts {
switch packet := pkt.(type) {
case *rtcp.PictureLossIndication:
mediaSSRC := packet.MediaSSRC
if p.pliThrottle.canSend(mediaSSRC) {
p.params.Logger.Debugw("send pli", "ssrc", mediaSSRC)
fwdPkts = append(fwdPkts, pkt)
}
case *rtcp.FullIntraRequest:
mediaSSRC := packet.MediaSSRC
if p.pliThrottle.canSend(mediaSSRC) {
p.params.Logger.Debugw("send fir", "ssrc", mediaSSRC)
fwdPkts = append(fwdPkts, pkt)
}
default:
fwdPkts = append(fwdPkts, pkt)
}
}
if len(fwdPkts) > 0 {
if err := p.publisher.pc.WriteRTCP(fwdPkts); err != nil {
p.params.Logger.Errorw("could not write RTCP to participant", err)
}
if err := p.publisher.pc.WriteRTCP(pkts); err != nil {
p.params.Logger.Errorw("could not write RTCP to participant", err)
}
}
}
pkg/rtc/participant_internal_test.go
+5 -5
View File
@@ -368,11 +368,11 @@ func newParticipantForTestWithOpts(identity livekit.ParticipantIdentity, opts *p
panic(err)
}
p, _ := NewParticipant(ParticipantParams{
Identity: identity,
Config: rtcConf,
Sink: &routingfakes.FakeMessageSink{},
ProtocolVersion: opts.protocolVersion,
ThrottleConfig: conf.RTC.PLIThrottle,
Identity: identity,
Config: rtcConf,
Sink: &routingfakes.FakeMessageSink{},
ProtocolVersion: opts.protocolVersion,
PLIThrottleConfig: conf.RTC.PLIThrottle,
}, opts.permissions)
return p
}
pkg/rtc/plithrottle.go
-58
View File
@@ -1,58 +0,0 @@
package rtc
import (
"sync"
"time"
"github.com/livekit/livekit-server/pkg/config"
"github.com/livekit/livekit-server/pkg/sfu"
)
type pliThrottle struct {
config config.PLIThrottleConfig
mu sync.RWMutex
periods map[uint32]int64
lastSent map[uint32]int64
}
func newPLIThrottle(conf config.PLIThrottleConfig) *pliThrottle {
return &pliThrottle{
config: conf,
periods: make(map[uint32]int64),
lastSent: make(map[uint32]int64),
}
}
func (t *pliThrottle) addTrack(ssrc uint32, rid string) {
t.mu.Lock()
defer t.mu.Unlock()
var duration time.Duration
switch rid {
case sfu.FullResolution:
duration = t.config.HighQuality
case sfu.HalfResolution:
duration = t.config.MidQuality
case sfu.QuarterResolution:
duration = t.config.LowQuality
default:
duration = t.config.MidQuality
}
t.periods[ssrc] = duration.Nanoseconds()
}
func (t *pliThrottle) canSend(ssrc uint32) bool {
t.mu.Lock()
defer t.mu.Unlock()
if period, ok := t.periods[ssrc]; ok {
if n := time.Now().UnixNano(); n-t.lastSent[ssrc] > period {
t.lastSent[ssrc] = n
return true
} else {
return false
}
}
return true
}
pkg/service/roommanager.go
+1 -1
View File
@@ -242,7 +242,7 @@ func (r *RoomManager) StartSession(ctx context.Context, roomName livekit.RoomNam
AudioConfig: r.config.Audio,
ProtocolVersion: pv,
Telemetry: r.telemetry,
ThrottleConfig: r.config.RTC.PLIThrottle,
PLIThrottleConfig: r.config.RTC.PLIThrottle,
CongestionControlConfig: r.config.RTC.CongestionControl,
EnabledCodecs: room.Room.EnabledCodecs,
Grants: pi.Grants,
pkg/sfu/buffer/buffer.go
+21 -1
View File
@@ -72,6 +72,9 @@ type Buffer struct {
lastSRRecv int64 // Represents wall clock of the most recent sender report arrival
lastTransit uint32
pliThrottle int64
lastPli int64
started bool
stats StreamStats
rrSnapshot *receiverReportSnapshot
@@ -109,6 +112,7 @@ func NewBuffer(ssrc uint32, vp, ap *sync.Pool) *Buffer {
mediaSSRC: ssrc,
videoPool: vp,
audioPool: ap,
pliThrottle: int64(500 * time.Millisecond),
logger: logger.Logger(logger.GetLogger()), // will be reset with correct context via SetLogger
callbackOps: make(chan func(), 50),
}
@@ -278,11 +282,27 @@ func (b *Buffer) OnClose(fn func()) {
b.onClose = fn
}
func (b *Buffer) SendPLI() {
func (b *Buffer) SetPLIThrottle(duration int64) {
b.Lock()
defer b.Unlock()
b.pliThrottle = duration
}
func (b *Buffer) SendPLI() {
now := time.Now().UnixNano()
b.Lock()
throttled := now-b.lastPli < b.pliThrottle
if throttled {
b.Unlock()
return
}
b.lastPli = now
b.stats.TotalPLIs++
b.Unlock()
b.logger.Debugw("send pli", "ssrc", b.mediaSSRC)
pli := []rtcp.Packet{
&rtcp.PictureLossIndication{SenderSSRC: rand.Uint32(), MediaSSRC: b.mediaSSRC},
}
pkg/sfu/receiver.go
+21 -15
View File
@@ -13,6 +13,7 @@ import (
"github.com/pion/webrtc/v3"
"github.com/rs/zerolog/log"
"github.com/livekit/livekit-server/pkg/config"
"github.com/livekit/livekit-server/pkg/sfu/buffer"
"github.com/livekit/livekit-server/pkg/sfu/connectionquality"
)
@@ -45,6 +46,8 @@ type TrackReceiver interface {
type WebRTCReceiver struct {
logger logger.Logger
pliThrottleConfig config.PLIThrottleConfig
peerID livekit.ParticipantID
trackID livekit.TrackID
streamID string
@@ -57,10 +60,7 @@ type WebRTCReceiver struct {
closed atomicBool
useTrackers bool
rtcpMu sync.Mutex
rtcpCh chan []rtcp.Packet
lastPli atomicInt64
pliThrottle int64
rtcpCh chan []rtcp.Packet
bufferMu sync.RWMutex
buffers [DefaultMaxLayerSpatial + 1]*buffer.Buffer
@@ -98,9 +98,9 @@ func RidToLayer(rid string) int32 {
type ReceiverOpts func(w *WebRTCReceiver) *WebRTCReceiver
// WithPliThrottle indicates minimum time(ms) between sending PLIs
func WithPliThrottle(period int64) ReceiverOpts {
func WithPliThrottle(pliThrottleConfig config.PLIThrottleConfig) ReceiverOpts {
return func(w *WebRTCReceiver) *WebRTCReceiver {
w.pliThrottle = period * 1e6
w.pliThrottleConfig = pliThrottleConfig
return w
}
}
@@ -143,7 +143,6 @@ func NewWebRTCReceiver(
kind: track.Kind(),
// LK-TODO: this should be based on VideoLayers protocol message rather than RID based
isSimulcast: len(track.RID()) > 0,
pliThrottle: 500e6,
downTracks: make([]TrackSender, 0),
index: make(map[livekit.ParticipantID]int),
free: make(map[int]struct{}),
@@ -236,8 +235,22 @@ func (w *WebRTCReceiver) AddUpTrack(track *webrtc.TrackRemote, buff *buffer.Buff
buff.SetLogger(w.logger)
buff.OnFeedback(w.sendRTCP)
layer := RidToLayer(track.RID())
var duration time.Duration
switch track.RID() {
case FullResolution:
duration = w.pliThrottleConfig.HighQuality
case HalfResolution:
duration = w.pliThrottleConfig.MidQuality
case QuarterResolution:
duration = w.pliThrottleConfig.LowQuality
default:
duration = w.pliThrottleConfig.MidQuality
}
if duration != 0 {
buff.SetPLIThrottle(duration.Nanoseconds())
}
layer := RidToLayer(track.RID())
w.upTrackMu.Lock()
w.upTracks[layer] = track
w.upTrackMu.Unlock()
@@ -362,12 +375,6 @@ func (w *WebRTCReceiver) SendPLI(layer int32) {
return
}
throttled := time.Now().UnixNano()-w.lastPli.get() < w.pliThrottle
if throttled {
return
}
w.lastPli.set(time.Now().UnixNano())
buff.SendPLI()
}
@@ -530,7 +537,6 @@ func (w *WebRTCReceiver) storeDownTrack(track TrackSender) {
func (w *WebRTCReceiver) DebugInfo() map[string]interface{} {
info := map[string]interface{}{
"Simulcast": w.isSimulcast,
"LastPli": w.lastPli,
}
w.upTrackMu.RLock()