From 3117547d60ff0de6a1ed48e242fe3afd3082c093 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Tue, 8 Feb 2022 22:50:43 +0530 Subject: [PATCH] Consolidating PLI throttle (#419) * 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 --- pkg/rtc/mediatrack.go | 17 ++++---- pkg/rtc/participant.go | 34 +++------------- pkg/rtc/participant_internal_test.go | 10 ++--- pkg/rtc/plithrottle.go | 58 ---------------------------- pkg/service/roommanager.go | 2 +- pkg/sfu/buffer/buffer.go | 22 ++++++++++- pkg/sfu/receiver.go | 36 ++++++++++------- 7 files changed, 62 insertions(+), 117 deletions(-) delete mode 100644 pkg/rtc/plithrottle.go diff --git a/pkg/rtc/mediatrack.go b/pkg/rtc/mediatrack.go index 2bde9016a..b99ff2ca5 100644 --- a/pkg/rtc/mediatrack.go +++ b/pkg/rtc/mediatrack.go @@ -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(), ) diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index e82d87858..cdbf5fd5e 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -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) } } } diff --git a/pkg/rtc/participant_internal_test.go b/pkg/rtc/participant_internal_test.go index 61c4e5261..7c19a0470 100644 --- a/pkg/rtc/participant_internal_test.go +++ b/pkg/rtc/participant_internal_test.go @@ -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 } diff --git a/pkg/rtc/plithrottle.go b/pkg/rtc/plithrottle.go deleted file mode 100644 index 11838ba7e..000000000 --- a/pkg/rtc/plithrottle.go +++ /dev/null @@ -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 -} diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index f47334f5a..1dddd23e9 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -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, diff --git a/pkg/sfu/buffer/buffer.go b/pkg/sfu/buffer/buffer.go index 95b1b4957..43520d669 100644 --- a/pkg/sfu/buffer/buffer.go +++ b/pkg/sfu/buffer/buffer.go @@ -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}, } diff --git a/pkg/sfu/receiver.go b/pkg/sfu/receiver.go index fe4d61dcb..dce372c47 100644 --- a/pkg/sfu/receiver.go +++ b/pkg/sfu/receiver.go @@ -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()