From a595c09bd2f74420ba2897531fecaed74125f99d Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Tue, 1 Aug 2023 00:18:05 +0530 Subject: [PATCH 1/7] Log the 32-bit RTP timestamp (#1921) --- pkg/sfu/buffer/rtpstats.go | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/pkg/sfu/buffer/rtpstats.go b/pkg/sfu/buffer/rtpstats.go index cd9c96066..1035f8f06 100644 --- a/pkg/sfu/buffer/rtpstats.go +++ b/pkg/sfu/buffer/rtpstats.go @@ -807,8 +807,10 @@ func (r *RTPStats) SetRtcpSenderReportData(srData *RTCPSenderReportData) { r.logger.Infow( "received sender report, out-of-order, resetting", "prevTSExt", r.srNewest.RTPTimestampExt, + "prevRTP", r.srNewest.RTPTimestamp, "prevNTP", r.srNewest.NTPTimestamp.Time().String(), "currTSExt", srDataCopy.RTPTimestampExt, + "currRTP", srDataCopy.RTPTimestamp, "currNTP", srDataCopy.NTPTimestamp.Time().String(), ) r.srFirst = &srDataCopy @@ -914,7 +916,9 @@ func (r *RTPStats) GetRtcpSenderReport(ssrc uint32, calculatedClockRate uint32) } if r.srNewest != nil && nowRTPExt < r.srNewest.RTPTimestampExt { - // If report being generated is behind, use the time different and clock rate of codec to produce next report. + // If report being generated is behind, use the time difference and + // clock rate of codec to produce next report. + // // Current report could be behind due to the following // - Publisher pacing // - Due to above, report from publisher side is ahead of packet timestamps. @@ -926,8 +930,10 @@ func (r *RTPStats) GetRtcpSenderReport(ssrc uint32, calculatedClockRate uint32) r.logger.Infow( "sending sender report, out-of-order, repairing", "prevTSExt", r.srNewest.RTPTimestampExt, + "prevRTP", r.srNewest.RTPTimestamp, "prevNTP", r.srNewest.NTPTimestamp.Time().String(), "currTSExt", nowRTPExt, + "currRTP", nowRTP, "currNTP", nowNTP.Time().String(), ) ntpDiffSinceLast := nowNTP.Time().Sub(r.srNewest.NTPTimestamp.Time()) From db3fbb57aed138b617722a3ccd90dd0d889f741d Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Tue, 1 Aug 2023 19:04:54 +0530 Subject: [PATCH 2/7] Do not log as error on connection reset by peer (#1923) --- pkg/service/rtcservice.go | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/pkg/service/rtcservice.go b/pkg/service/rtcservice.go index 5259622bc..f19c70fce 100644 --- a/pkg/service/rtcservice.go +++ b/pkg/service/rtcservice.go @@ -361,8 +361,16 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) { req, count, err := sigConn.ReadRequest() if err != nil { // normal/expected closure - if err == io.EOF || strings.HasSuffix(err.Error(), "use of closed network connection") || - websocket.IsCloseError(err, websocket.CloseAbnormalClosure, websocket.CloseGoingAway, websocket.CloseNormalClosure, websocket.CloseNoStatusReceived) { + if err == io.EOF || + strings.HasSuffix(err.Error(), "use of closed network connection") || + strings.HasSuffix(err.Error(), "connection reset by peer") || + websocket.IsCloseError( + err, + websocket.CloseAbnormalClosure, + websocket.CloseGoingAway, + websocket.CloseNormalClosure, + websocket.CloseNoStatusReceived, + ) { pLogger.Infow("exit ws read loop for closed connection", "connID", cr.ConnectionID, "wsError", err) } else { pLogger.Errorw("error reading from websocket", err, "connID", cr.ConnectionID) From 0c34f12fa10b7c94e727f4f650cde1d15376c21d Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Wed, 2 Aug 2023 00:03:38 +0530 Subject: [PATCH 3/7] Demote some high frequency logs to Debugw (#1925) --- pkg/rtc/dynacastmanager.go | 2 +- pkg/rtc/subscribedtrack.go | 2 +- pkg/rtc/subscriptionmanager.go | 4 ++-- pkg/sfu/buffer/rtpstats.go | 19 ------------------- pkg/sfu/forwarder.go | 8 ++++---- pkg/sfu/videolayerselector/base.go | 2 +- pkg/sfu/videolayerselector/simulcast.go | 2 +- 7 files changed, 10 insertions(+), 29 deletions(-) diff --git a/pkg/rtc/dynacastmanager.go b/pkg/rtc/dynacastmanager.go index edaadb4f2..8943646fe 100644 --- a/pkg/rtc/dynacastmanager.go +++ b/pkg/rtc/dynacastmanager.go @@ -238,7 +238,7 @@ func (d *DynacastManager) update(force bool) { // clear debounce on send d.maxSubscribedQualityDebounce(func() {}) - d.params.Logger.Infow("committing quality change", + d.params.Logger.Debugw("committing quality change", "force", force, "committedMaxSubscribedQuality", d.committedMaxSubscribedQuality, "maxSubscribedQuality", d.maxSubscribedQuality, diff --git a/pkg/rtc/subscribedtrack.go b/pkg/rtc/subscribedtrack.go index 5a83e0a1f..c586ec3a7 100644 --- a/pkg/rtc/subscribedtrack.go +++ b/pkg/rtc/subscribedtrack.go @@ -193,7 +193,7 @@ func (t *SubscribedTrack) UpdateSubscriberSettings(settings *livekit.UpdateTrack t.settings.Store(settings) if prevDisabled != settings.Disabled { - t.logger.Infow("updated subscribed track enabled", "enabled", !settings.Disabled) + t.logger.Debugw("updated subscribed track enabled", "enabled", !settings.Disabled) } // avoid frequent changes to mute & video layers, unless it became visible diff --git a/pkg/rtc/subscriptionmanager.go b/pkg/rtc/subscriptionmanager.go index 1ed7d3f8c..8864c99b4 100644 --- a/pkg/rtc/subscriptionmanager.go +++ b/pkg/rtc/subscriptionmanager.go @@ -148,7 +148,7 @@ func (m *SubscriptionManager) SubscribeToTrack(trackID livekit.TrackID) { sub, desireChanged = m.setDesired(trackID, true) } if desireChanged { - sub.logger.Infow("subscribing to track") + sub.logger.Debugw("subscribing to track") } // always reconcile, since SubscribeToTrack could be called when the track is ready @@ -161,7 +161,7 @@ func (m *SubscriptionManager) UnsubscribeFromTrack(trackID livekit.TrackID) { return } - sub.logger.Infow("unsubscribing from track") + sub.logger.Debugw("unsubscribing from track") m.queueReconcile(trackID) } diff --git a/pkg/sfu/buffer/rtpstats.go b/pkg/sfu/buffer/rtpstats.go index 1035f8f06..d1069c00a 100644 --- a/pkg/sfu/buffer/rtpstats.go +++ b/pkg/sfu/buffer/rtpstats.go @@ -987,25 +987,6 @@ func (r *RTPStats) GetRtcpSenderReport(ssrc uint32, calculatedClockRate uint32) "nowRTPExt", nowRTPExt, "nowRTPExtUsingRate", nowRTPExtUsingRate, ) - } else { - packetDriftResult, reportDriftResult := r.getDrift() - r.logger.Debugw( - "sending sender report", - "ntp", nowNTP.Time().String(), - "rtp", nowRTP, - "departure", now.String(), - "ntpDiffSinceLast", ntpDiffSinceLast.Seconds(), - "rtpDiffSinceLast", int32(rtpDiffSinceLast), - "departureDiffSinceLast", departureDiffSinceLast.Seconds(), - "expectedTimeDiffSinceLast", expectedTimeDiffSinceLast, - "packetDrift", packetDriftResult.String(), - "reportDrift", reportDriftResult.String(), - "highestTS", r.highestTS, - "highestTime", r.highestTime.String(), - "calculatedClockRate", calculatedClockRate, - "nowRTPExt", nowRTPExt, - "nowRTPExtUsingRate", nowRTPExtUsingRate, - ) } return &rtcp.SenderReport{ diff --git a/pkg/sfu/forwarder.go b/pkg/sfu/forwarder.go index 9839fd76c..2da3ec9ef 100644 --- a/pkg/sfu/forwarder.go +++ b/pkg/sfu/forwarder.go @@ -395,7 +395,7 @@ func (f *Forwarder) Mute(muted bool) bool { // It could result in some bandwidth consumed for stream without visibility in // the case of intentional mute. if muted && f.isDeficientLocked() && f.lastAllocation.PauseReason == VideoPauseReasonBandwidth { - f.logger.Infow("ignoring forwarder mute, paused due to congestion") + f.logger.Debugw("ignoring forwarder mute, paused due to congestion") return false } @@ -1525,7 +1525,7 @@ func (f *Forwarder) processSourceSwitch(extPkt *buffer.ExtPacket, layer int32) e f.referenceLayerSpatial = layer f.rtpMunger.SetLastSnTs(extPkt) f.codecMunger.SetLast(extPkt) - f.logger.Infow( + f.logger.Debugw( "starting forwarding", "sequenceNumber", extPkt.Packet.SequenceNumber, "timestamp", extPkt.Packet.Timestamp, @@ -1657,7 +1657,7 @@ func (f *Forwarder) processSourceSwitch(extPkt *buffer.ExtPacket, layer int32) e // nominal increase nextTS = lastTS + 1 } - f.logger.Infow( + f.logger.Debugw( "next timestamp on switch", "switchingAt", switchingAt.String(), "layer", layer, @@ -1816,7 +1816,7 @@ func (f *Forwarder) maybeStart() { f.rtpMunger.SetLastSnTs(extPkt) f.firstTS = extPkt.Packet.Timestamp - f.logger.Infow( + f.logger.Debugw( "starting with dummy forwarding", "sequenceNumber", extPkt.Packet.SequenceNumber, "timestamp", extPkt.Packet.Timestamp, diff --git a/pkg/sfu/videolayerselector/base.go b/pkg/sfu/videolayerselector/base.go index 37b223948..6708e7030 100644 --- a/pkg/sfu/videolayerselector/base.go +++ b/pkg/sfu/videolayerselector/base.go @@ -166,7 +166,7 @@ func (b *Base) SelectTemporal(extPkt *buffer.ExtPacket) (int32, bool) { b.previousLayer = b.currentLayer b.currentLayer.Temporal = next - b.logger.Infow( + b.logger.Debugw( "updating temporal layer", "previous", b.previousLayer, "current", b.currentLayer, diff --git a/pkg/sfu/videolayerselector/simulcast.go b/pkg/sfu/videolayerselector/simulcast.go index 7d7e173a8..b4a5c527e 100644 --- a/pkg/sfu/videolayerselector/simulcast.go +++ b/pkg/sfu/videolayerselector/simulcast.go @@ -50,7 +50,7 @@ func (s *Simulcast) Select(extPkt *buffer.ExtPacket, layer int32) (result VideoL } if reason != "" { - s.logger.Infow( + s.logger.Debugw( reason, "previous", s.previousLayer, "current", s.currentLayer, From 0c5161e9abf559618245d4fbc3e0eb1fe125a7f4 Mon Sep 17 00:00:00 2001 From: "renovate[bot]" <29139614+renovate[bot]@users.noreply.github.com> Date: Tue, 1 Aug 2023 20:36:23 -0700 Subject: [PATCH 4/7] Update golang.org/x/exp digest to d63ba01 (#1924) Generated by renovateBot Co-authored-by: renovate[bot] <29139614+renovate[bot]@users.noreply.github.com> --- go.mod | 2 +- go.sum | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/go.mod b/go.mod index 1d7e21bf7..6bde0c1ee 100644 --- a/go.mod +++ b/go.mod @@ -45,7 +45,7 @@ require ( github.com/urfave/cli/v2 v2.25.7 github.com/urfave/negroni/v3 v3.0.0 go.uber.org/atomic v1.11.0 - golang.org/x/exp v0.0.0-20230728194245-b0cb94b80691 + golang.org/x/exp v0.0.0-20230801115018-d63ba01acd4b golang.org/x/sync v0.3.0 google.golang.org/protobuf v1.31.0 gopkg.in/yaml.v3 v3.0.1 diff --git a/go.sum b/go.sum index 9a2e52f3c..d038f6bd6 100644 --- a/go.sum +++ b/go.sum @@ -289,8 +289,8 @@ golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5y golang.org/x/crypto v0.8.0/go.mod h1:mRqEX+O9/h5TFCrQhkgjo2yKi0yYA+9ecGkdQoHrywE= golang.org/x/crypto v0.10.0 h1:LKqV2xt9+kDzSTfOhx4FrkEBcMrAgHSYgzywV9zcGmM= golang.org/x/crypto v0.10.0/go.mod h1:o4eNf7Ede1fv+hwOwZsTHl9EsPFO6q6ZvYR8vYfY45I= -golang.org/x/exp v0.0.0-20230728194245-b0cb94b80691 h1:/yRP+0AN7mf5DkD3BAI6TOFnd51gEoDEb8o35jIFtgw= -golang.org/x/exp v0.0.0-20230728194245-b0cb94b80691/go.mod h1:FXUEEKJgO7OQYeo8N01OfiKP8RXMtf6e8aTskBGqWdc= +golang.org/x/exp v0.0.0-20230801115018-d63ba01acd4b h1:r+vk0EmXNmekl0S0BascoeeoHk/L7wmaW2QF90K+kYI= +golang.org/x/exp v0.0.0-20230801115018-d63ba01acd4b/go.mod h1:FXUEEKJgO7OQYeo8N01OfiKP8RXMtf6e8aTskBGqWdc= golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4= golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs= From f7a1776f4c890c6b91e40b12cd4ba8c7a77e6a7e Mon Sep 17 00:00:00 2001 From: cnderrauber Date: Wed, 2 Aug 2023 16:12:23 +0800 Subject: [PATCH 5/7] Add control of playout delay (#1838) * Add control of playout delay Add config to enable playout delay. The delay will be limited by [min,max] in the config option and calculated by upstream & downstream RTT. * check protocol version to enable playout delay * Move config to room, limit playout-delay update interval, solve comments * Remove adaptive playout-delay * Remove unused config --- config-sample.yaml | 5 + pkg/config/config.go | 19 +- pkg/rtc/config.go | 4 +- pkg/rtc/mediatrackreceiver.go | 9 +- pkg/rtc/mediatracksubscriptions.go | 34 ++- pkg/rtc/participant.go | 14 ++ pkg/rtc/participant_sdp.go | 2 +- pkg/rtc/transport.go | 10 +- pkg/rtc/transportmanager.go | 2 + pkg/rtc/types/interfaces.go | 3 + pkg/rtc/types/protocol_version.go | 6 +- .../typesfakes/fake_local_media_track.go | 65 +++++ .../typesfakes/fake_local_participant.go | 130 ++++++++++ pkg/rtc/types/typesfakes/fake_media_track.go | 65 +++++ pkg/rtc/utils.go | 19 ++ pkg/service/roomallocator.go | 5 + pkg/service/roommanager.go | 1 + pkg/sfu/buffer/buffer.go | 2 +- .../dependencydescriptorextension.go | 2 +- pkg/sfu/downtrack.go | 231 ++++++++++-------- pkg/sfu/forwarder.go | 2 +- pkg/sfu/receiver.go | 14 +- pkg/sfu/rtpextension/playoutdelay.go | 56 +++++ pkg/sfu/rtpextension/playoutdelay_test.go | 35 +++ 24 files changed, 598 insertions(+), 137 deletions(-) create mode 100644 pkg/sfu/rtpextension/playoutdelay.go create mode 100644 pkg/sfu/rtpextension/playoutdelay_test.go diff --git a/config-sample.yaml b/config-sample.yaml index bdf63e6a0..1e44c6c2b 100644 --- a/config-sample.yaml +++ b/config-sample.yaml @@ -152,6 +152,11 @@ keys: # enable_remote_unmute: true # # limit size of room and participant's metadata, 0 for no limit # max_metadata_size: 0 +# # control playout delay in ms of video track (and associated audio track) +# playout_delay: +# enabled: true +# min: 100 +# max: 300 # Webhooks # when configured, LiveKit notifies your URL handler with room events diff --git a/pkg/config/config.go b/pkg/config/config.go index c280423e4..923bab6e7 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -197,6 +197,12 @@ type StreamTrackersConfig struct { Screenshare StreamTrackerConfig `yaml:"screenshare,omitempty"` } +type PlayoutDelayConfig struct { + Enabled bool `yaml:"enabled,omitempty"` + Min int `yaml:"min,omitempty"` + Max int `yaml:"max,omitempty"` +} + type VideoConfig struct { DynacastPauseDelay time.Duration `yaml:"dynacast_pause_delay,omitempty"` StreamTracker StreamTrackersConfig `yaml:"stream_tracker,omitempty"` @@ -204,12 +210,13 @@ type VideoConfig struct { type RoomConfig struct { // enable rooms to be automatically created - AutoCreate bool `yaml:"auto_create,omitempty"` - EnabledCodecs []CodecSpec `yaml:"enabled_codecs,omitempty"` - MaxParticipants uint32 `yaml:"max_participants,omitempty"` - EmptyTimeout uint32 `yaml:"empty_timeout,omitempty"` - EnableRemoteUnmute bool `yaml:"enable_remote_unmute,omitempty"` - MaxMetadataSize uint32 `yaml:"max_metadata_size,omitempty"` + AutoCreate bool `yaml:"auto_create,omitempty"` + EnabledCodecs []CodecSpec `yaml:"enabled_codecs,omitempty"` + MaxParticipants uint32 `yaml:"max_participants,omitempty"` + EmptyTimeout uint32 `yaml:"empty_timeout,omitempty"` + EnableRemoteUnmute bool `yaml:"enable_remote_unmute,omitempty"` + MaxMetadataSize uint32 `yaml:"max_metadata_size,omitempty"` + PlayoutDelay PlayoutDelayConfig `yaml:"playout_delay,omitempty"` } type CodecSpec struct { diff --git a/pkg/rtc/config.go b/pkg/rtc/config.go index efe2a6a1f..820d41857 100644 --- a/pkg/rtc/config.go +++ b/pkg/rtc/config.go @@ -86,7 +86,7 @@ func NewWebRTCConfig(conf *config.Config) (*WebRTCConfig, error) { sdp.SDESRTPStreamIDURI, sdp.TransportCCURI, frameMarking, - dd.ExtensionUrl, + dd.ExtensionURI, }, }, RTCPFeedback: RTCPFeedbackConfig{ @@ -106,7 +106,7 @@ func NewWebRTCConfig(conf *config.Config) (*WebRTCConfig, error) { subscriberConfig := DirectionConfig{ StrictACKs: conf.RTC.StrictACKs, RTPHeaderExtension: RTPHeaderExtensionConfig{ - Video: []string{dd.ExtensionUrl}, + Video: []string{dd.ExtensionURI}, }, RTCPFeedback: RTCPFeedbackConfig{ Video: []webrtc.RTCPFeedback{ diff --git a/pkg/rtc/mediatrackreceiver.go b/pkg/rtc/mediatrackreceiver.go index cff226ca0..a89bd2c8a 100644 --- a/pkg/rtc/mediatrackreceiver.go +++ b/pkg/rtc/mediatrackreceiver.go @@ -226,7 +226,7 @@ func (t *MediaTrackReceiver) SetPotentialCodecs(codecs []webrtc.RTPCodecParamete // that is munged in svc codec. headersWithoutDD := make([]webrtc.RTPHeaderExtensionParameter, 0, len(headers)) for _, h := range headers { - if h.URI != dependencydescriptor.ExtensionUrl { + if h.URI != dependencydescriptor.ExtensionURI { headersWithoutDD = append(headersWithoutDD, h) } } @@ -401,6 +401,13 @@ func (t *MediaTrackReceiver) Source() livekit.TrackSource { return t.trackInfo.Source } +func (t *MediaTrackReceiver) Stream() string { + t.lock.RLock() + defer t.lock.RUnlock() + + return t.trackInfo.Stream +} + func (t *MediaTrackReceiver) PublisherID() livekit.ParticipantID { return t.params.ParticipantID } diff --git a/pkg/rtc/mediatracksubscriptions.go b/pkg/rtc/mediatracksubscriptions.go index 51499a5ba..378c235b7 100644 --- a/pkg/rtc/mediatracksubscriptions.go +++ b/pkg/rtc/mediatracksubscriptions.go @@ -112,20 +112,29 @@ func (t *MediaTrackSubscriptions) AddSubscriber(sub types.LocalParticipant, wr * for _, c := range codecs { c.RTCPFeedback = rtcpFeedback } + + streamID := wr.StreamID() + if sub.SupportSyncStreamID() && t.params.MediaTrack.Stream() != "" { + streamID = PackSyncStreamID(t.params.MediaTrack.PublisherID(), t.params.MediaTrack.Stream()) + } + var trailer []byte if t.params.MediaTrack.IsEncrypted() { trailer = sub.GetTrailer() } - downTrack, err := sfu.NewDownTrack( - codecs, - wr, - sub.GetBufferFactory(), - subscriberID, - t.params.ReceiverConfig.PacketBufferSize, - sub.GetPacer(), - trailer, - LoggerWithTrack(sub.GetLogger(), trackID, t.params.IsRelayed), - ) + + downTrack, err := sfu.NewDownTrack(sfu.DowntrackParams{ + Codecs: codecs, + Receiver: wr, + BufferFactory: sub.GetBufferFactory(), + SubID: subscriberID, + StreamID: streamID, + MaxTrack: t.params.ReceiverConfig.PacketBufferSize, + PlayoutDelayLimit: sub.GetPlayoutDelayConfig(), + Pacer: sub.GetPacer(), + Trailer: trailer, + Logger: LoggerWithTrack(sub.GetLogger(), trackID, t.params.IsRelayed), + }) if err != nil { return nil, err } @@ -200,6 +209,8 @@ func (t *MediaTrackSubscriptions) AddSubscriber(sub types.LocalParticipant, wr * reusingTransceiver.Store(true) rtpSender := existingTransceiver.Sender() if rtpSender != nil { + // replaced track will bind immediately without negotiation, SetTransceiver first before bind + downTrack.SetTransceiver(existingTransceiver) err := rtpSender.ReplaceTrack(downTrack) if err == nil { sender = rtpSender @@ -260,9 +271,6 @@ func (t *MediaTrackSubscriptions) AddSubscriber(sub types.LocalParticipant, wr * subTrack.SetNeedsNegotiation(!replacedTrack) subTrack.SetRTPSender(sender) - sendParameters := sender.GetParameters() - downTrack.SetRTPHeaderExtensions(sendParameters.HeaderExtensions) - downTrack.SetTransceiver(transceiver) downTrack.OnCloseHandler(func(willBeResumed bool) { diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index cd0d61e9d..490a6e4a9 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -108,6 +108,7 @@ type ParticipantParams struct { SubscriberAllowPause bool SubscriptionLimitAudio int32 SubscriptionLimitVideo int32 + PlayoutDelay *livekit.PlayoutDelay } type ParticipantImpl struct { @@ -1104,6 +1105,7 @@ func (p *ParticipantImpl) setupTransportManager() error { TCPFallbackRTTThreshold: p.params.TCPFallbackRTTThreshold, AllowUDPUnstableFallback: p.params.AllowUDPUnstableFallback, TURNSEnabled: p.params.TURNSEnabled, + AllowPlayoutDelay: p.params.PlayoutDelay.GetEnabled() && p.SupportSyncStreamID(), Logger: p.params.Logger, }) if err != nil { @@ -1584,6 +1586,10 @@ func (p *ParticipantImpl) addPendingTrackLocked(req *livekit.AddTrackRequest) *l DisableRed: req.DisableRed, Stereo: req.Stereo, Encryption: req.Encryption, + Stream: req.Stream, + } + if ti.Stream == "" { + ti.Stream = StreamFromTrackSource(ti.Source) } p.setStableTrackID(req.Cid, ti) for _, codec := range req.SimulcastCodecs { @@ -2214,6 +2220,14 @@ func (p *ParticipantImpl) UpdateMediaLoss(nodeID livekit.NodeID, trackID livekit return nil } +func (p *ParticipantImpl) GetPlayoutDelayConfig() *livekit.PlayoutDelay { + return p.params.PlayoutDelay +} + +func (p *ParticipantImpl) SupportSyncStreamID() bool { + return p.ProtocolVersion().SupportSyncStreamID() && !p.params.ClientInfo.isFirefox() +} + func codecsFromMediaDescription(m *sdp.MediaDescription) (out []sdp.Codec, err error) { s := &sdp.SessionDescription{ MediaDescriptions: []*sdp.MediaDescription{m}, diff --git a/pkg/rtc/participant_sdp.go b/pkg/rtc/participant_sdp.go index 01a50b6b0..fc11324d6 100644 --- a/pkg/rtc/participant_sdp.go +++ b/pkg/rtc/participant_sdp.go @@ -155,7 +155,7 @@ func (p *ParticipantImpl) setCodecPreferencesVideoForPublisher(offer webrtc.Sess // remove dd extension if av1/vp9 not preferred if !strings.Contains(strings.ToLower(mime), "av1") && !strings.Contains(strings.ToLower(mime), "vp9") { for i, attr := range unmatchVideo.Attributes { - if strings.Contains(attr.Value, dd.ExtensionUrl) { + if strings.Contains(attr.Value, dd.ExtensionURI) { unmatchVideo.Attributes[i] = unmatchVideo.Attributes[len(unmatchVideo.Attributes)-1] unmatchVideo.Attributes = unmatchVideo.Attributes[:len(unmatchVideo.Attributes)-1] break diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index 2c565f194..65698500f 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -43,6 +43,7 @@ import ( "github.com/livekit/livekit-server/pkg/config" "github.com/livekit/livekit-server/pkg/rtc/types" "github.com/livekit/livekit-server/pkg/sfu/pacer" + "github.com/livekit/livekit-server/pkg/sfu/rtpextension" "github.com/livekit/livekit-server/pkg/sfu/streamallocator" "github.com/livekit/livekit-server/pkg/telemetry" "github.com/livekit/livekit-server/pkg/telemetry/prometheus" @@ -251,10 +252,17 @@ type TransportParams struct { ClientInfo ClientInfo IsOfferer bool IsSendSide bool + AllowPlayoutDelay bool } func newPeerConnection(params TransportParams, onBandwidthEstimator func(estimator cc.BandwidthEstimator)) (*webrtc.PeerConnection, *webrtc.MediaEngine, error) { - me, err := createMediaEngine(params.EnabledCodecs, params.DirectionConfig) + directionConfig := params.DirectionConfig + + if params.AllowPlayoutDelay { + directionConfig.RTPHeaderExtension.Video = append(directionConfig.RTPHeaderExtension.Video, rtpextension.PlayoutDelayURI) + } + + me, err := createMediaEngine(params.EnabledCodecs, directionConfig) if err != nil { return nil, nil, err } diff --git a/pkg/rtc/transportmanager.go b/pkg/rtc/transportmanager.go index 1b056b6e4..6e5ef78e3 100644 --- a/pkg/rtc/transportmanager.go +++ b/pkg/rtc/transportmanager.go @@ -65,6 +65,7 @@ type TransportManagerParams struct { TCPFallbackRTTThreshold int AllowUDPUnstableFallback bool TURNSEnabled bool + AllowPlayoutDelay bool Logger logger.Logger } @@ -184,6 +185,7 @@ func NewTransportManager(params TransportManagerParams) (*TransportManager, erro ClientInfo: params.ClientInfo, IsOfferer: true, IsSendSide: true, + AllowPlayoutDelay: params.AllowPlayoutDelay, }) if err != nil { return nil, err diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 6431dc661..baacff5dd 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -295,6 +295,7 @@ type LocalParticipant interface { GetLogger() logger.Logger GetAdaptiveStream() bool ProtocolVersion() ProtocolVersion + SupportSyncStreamID() bool ConnectedAt() time.Time IsClosed() bool IsReady() bool @@ -305,6 +306,7 @@ type LocalParticipant interface { GetClientConfiguration() *livekit.ClientConfiguration GetICEConnectionType() ICEConnectionType GetBufferFactory() *buffer.Factory + GetPlayoutDelayConfig() *livekit.PlayoutDelay SetResponseSink(sink routing.MessageSink) CloseSignalConnection(reason SignallingCloseReason) @@ -429,6 +431,7 @@ type MediaTrack interface { Kind() livekit.TrackType Name() string Source() livekit.TrackSource + Stream() string ToProto() *livekit.TrackInfo diff --git a/pkg/rtc/types/protocol_version.go b/pkg/rtc/types/protocol_version.go index 93449ae79..9c4e47cc3 100644 --- a/pkg/rtc/types/protocol_version.go +++ b/pkg/rtc/types/protocol_version.go @@ -16,7 +16,7 @@ package types type ProtocolVersion int -const CurrentProtocol = 9 +const CurrentProtocol = 10 func (v ProtocolVersion) SupportsPackedStreamId() bool { return v > 0 @@ -71,3 +71,7 @@ func (v ProtocolVersion) SupportFastStart() bool { func (v ProtocolVersion) SupportHandlesDisconnectedUpdate() bool { return v > 8 } + +func (v ProtocolVersion) SupportSyncStreamID() bool { + return v > 9 +} diff --git a/pkg/rtc/types/typesfakes/fake_local_media_track.go b/pkg/rtc/types/typesfakes/fake_local_media_track.go index 0b9a4517c..83fdb1580 100644 --- a/pkg/rtc/types/typesfakes/fake_local_media_track.go +++ b/pkg/rtc/types/typesfakes/fake_local_media_track.go @@ -302,6 +302,16 @@ type FakeLocalMediaTrack struct { sourceReturnsOnCall map[int]struct { result1 livekit.TrackSource } + StreamStub func() string + streamMutex sync.RWMutex + streamArgsForCall []struct { + } + streamReturns struct { + result1 string + } + streamReturnsOnCall map[int]struct { + result1 string + } ToProtoStub func() *livekit.TrackInfo toProtoMutex sync.RWMutex toProtoArgsForCall []struct { @@ -1893,6 +1903,59 @@ func (fake *FakeLocalMediaTrack) SourceReturnsOnCall(i int, result1 livekit.Trac }{result1} } +func (fake *FakeLocalMediaTrack) Stream() string { + fake.streamMutex.Lock() + ret, specificReturn := fake.streamReturnsOnCall[len(fake.streamArgsForCall)] + fake.streamArgsForCall = append(fake.streamArgsForCall, struct { + }{}) + stub := fake.StreamStub + fakeReturns := fake.streamReturns + fake.recordInvocation("Stream", []interface{}{}) + fake.streamMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeLocalMediaTrack) StreamCallCount() int { + fake.streamMutex.RLock() + defer fake.streamMutex.RUnlock() + return len(fake.streamArgsForCall) +} + +func (fake *FakeLocalMediaTrack) StreamCalls(stub func() string) { + fake.streamMutex.Lock() + defer fake.streamMutex.Unlock() + fake.StreamStub = stub +} + +func (fake *FakeLocalMediaTrack) StreamReturns(result1 string) { + fake.streamMutex.Lock() + defer fake.streamMutex.Unlock() + fake.StreamStub = nil + fake.streamReturns = struct { + result1 string + }{result1} +} + +func (fake *FakeLocalMediaTrack) StreamReturnsOnCall(i int, result1 string) { + fake.streamMutex.Lock() + defer fake.streamMutex.Unlock() + fake.StreamStub = nil + if fake.streamReturnsOnCall == nil { + fake.streamReturnsOnCall = make(map[int]struct { + result1 string + }) + } + fake.streamReturnsOnCall[i] = struct { + result1 string + }{result1} +} + func (fake *FakeLocalMediaTrack) ToProto() *livekit.TrackInfo { fake.toProtoMutex.Lock() ret, specificReturn := fake.toProtoReturnsOnCall[len(fake.toProtoArgsForCall)] @@ -2050,6 +2113,8 @@ func (fake *FakeLocalMediaTrack) Invocations() map[string][][]interface{} { defer fake.signalCidMutex.RUnlock() fake.sourceMutex.RLock() defer fake.sourceMutex.RUnlock() + fake.streamMutex.RLock() + defer fake.streamMutex.RUnlock() fake.toProtoMutex.RLock() defer fake.toProtoMutex.RUnlock() fake.updateVideoLayersMutex.RLock() diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index 6e3863a45..b5338c33a 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -263,6 +263,16 @@ type FakeLocalParticipant struct { getPacerReturnsOnCall map[int]struct { result1 pacer.Pacer } + GetPlayoutDelayConfigStub func() *livekit.PlayoutDelay + getPlayoutDelayConfigMutex sync.RWMutex + getPlayoutDelayConfigArgsForCall []struct { + } + getPlayoutDelayConfigReturns struct { + result1 *livekit.PlayoutDelay + } + getPlayoutDelayConfigReturnsOnCall map[int]struct { + result1 *livekit.PlayoutDelay + } GetPublishedTrackStub func(livekit.TrackID) types.MediaTrack getPublishedTrackMutex sync.RWMutex getPublishedTrackArgsForCall []struct { @@ -772,6 +782,16 @@ type FakeLocalParticipant struct { arg2 livekit.TrackID arg3 bool } + SupportSyncStreamIDStub func() bool + supportSyncStreamIDMutex sync.RWMutex + supportSyncStreamIDArgsForCall []struct { + } + supportSyncStreamIDReturns struct { + result1 bool + } + supportSyncStreamIDReturnsOnCall map[int]struct { + result1 bool + } ToProtoStub func() *livekit.ParticipantInfo toProtoMutex sync.RWMutex toProtoArgsForCall []struct { @@ -2149,6 +2169,59 @@ func (fake *FakeLocalParticipant) GetPacerReturnsOnCall(i int, result1 pacer.Pac }{result1} } +func (fake *FakeLocalParticipant) GetPlayoutDelayConfig() *livekit.PlayoutDelay { + fake.getPlayoutDelayConfigMutex.Lock() + ret, specificReturn := fake.getPlayoutDelayConfigReturnsOnCall[len(fake.getPlayoutDelayConfigArgsForCall)] + fake.getPlayoutDelayConfigArgsForCall = append(fake.getPlayoutDelayConfigArgsForCall, struct { + }{}) + stub := fake.GetPlayoutDelayConfigStub + fakeReturns := fake.getPlayoutDelayConfigReturns + fake.recordInvocation("GetPlayoutDelayConfig", []interface{}{}) + fake.getPlayoutDelayConfigMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeLocalParticipant) GetPlayoutDelayConfigCallCount() int { + fake.getPlayoutDelayConfigMutex.RLock() + defer fake.getPlayoutDelayConfigMutex.RUnlock() + return len(fake.getPlayoutDelayConfigArgsForCall) +} + +func (fake *FakeLocalParticipant) GetPlayoutDelayConfigCalls(stub func() *livekit.PlayoutDelay) { + fake.getPlayoutDelayConfigMutex.Lock() + defer fake.getPlayoutDelayConfigMutex.Unlock() + fake.GetPlayoutDelayConfigStub = stub +} + +func (fake *FakeLocalParticipant) GetPlayoutDelayConfigReturns(result1 *livekit.PlayoutDelay) { + fake.getPlayoutDelayConfigMutex.Lock() + defer fake.getPlayoutDelayConfigMutex.Unlock() + fake.GetPlayoutDelayConfigStub = nil + fake.getPlayoutDelayConfigReturns = struct { + result1 *livekit.PlayoutDelay + }{result1} +} + +func (fake *FakeLocalParticipant) GetPlayoutDelayConfigReturnsOnCall(i int, result1 *livekit.PlayoutDelay) { + fake.getPlayoutDelayConfigMutex.Lock() + defer fake.getPlayoutDelayConfigMutex.Unlock() + fake.GetPlayoutDelayConfigStub = nil + if fake.getPlayoutDelayConfigReturnsOnCall == nil { + fake.getPlayoutDelayConfigReturnsOnCall = make(map[int]struct { + result1 *livekit.PlayoutDelay + }) + } + fake.getPlayoutDelayConfigReturnsOnCall[i] = struct { + result1 *livekit.PlayoutDelay + }{result1} +} + func (fake *FakeLocalParticipant) GetPublishedTrack(arg1 livekit.TrackID) types.MediaTrack { fake.getPublishedTrackMutex.Lock() ret, specificReturn := fake.getPublishedTrackReturnsOnCall[len(fake.getPublishedTrackArgsForCall)] @@ -5008,6 +5081,59 @@ func (fake *FakeLocalParticipant) SubscriptionPermissionUpdateArgsForCall(i int) return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3 } +func (fake *FakeLocalParticipant) SupportSyncStreamID() bool { + fake.supportSyncStreamIDMutex.Lock() + ret, specificReturn := fake.supportSyncStreamIDReturnsOnCall[len(fake.supportSyncStreamIDArgsForCall)] + fake.supportSyncStreamIDArgsForCall = append(fake.supportSyncStreamIDArgsForCall, struct { + }{}) + stub := fake.SupportSyncStreamIDStub + fakeReturns := fake.supportSyncStreamIDReturns + fake.recordInvocation("SupportSyncStreamID", []interface{}{}) + fake.supportSyncStreamIDMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeLocalParticipant) SupportSyncStreamIDCallCount() int { + fake.supportSyncStreamIDMutex.RLock() + defer fake.supportSyncStreamIDMutex.RUnlock() + return len(fake.supportSyncStreamIDArgsForCall) +} + +func (fake *FakeLocalParticipant) SupportSyncStreamIDCalls(stub func() bool) { + fake.supportSyncStreamIDMutex.Lock() + defer fake.supportSyncStreamIDMutex.Unlock() + fake.SupportSyncStreamIDStub = stub +} + +func (fake *FakeLocalParticipant) SupportSyncStreamIDReturns(result1 bool) { + fake.supportSyncStreamIDMutex.Lock() + defer fake.supportSyncStreamIDMutex.Unlock() + fake.SupportSyncStreamIDStub = nil + fake.supportSyncStreamIDReturns = struct { + result1 bool + }{result1} +} + +func (fake *FakeLocalParticipant) SupportSyncStreamIDReturnsOnCall(i int, result1 bool) { + fake.supportSyncStreamIDMutex.Lock() + defer fake.supportSyncStreamIDMutex.Unlock() + fake.SupportSyncStreamIDStub = nil + if fake.supportSyncStreamIDReturnsOnCall == nil { + fake.supportSyncStreamIDReturnsOnCall = make(map[int]struct { + result1 bool + }) + } + fake.supportSyncStreamIDReturnsOnCall[i] = struct { + result1 bool + }{result1} +} + func (fake *FakeLocalParticipant) ToProto() *livekit.ParticipantInfo { fake.toProtoMutex.Lock() ret, specificReturn := fake.toProtoReturnsOnCall[len(fake.toProtoArgsForCall)] @@ -5703,6 +5829,8 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} { defer fake.getLoggerMutex.RUnlock() fake.getPacerMutex.RLock() defer fake.getPacerMutex.RUnlock() + fake.getPlayoutDelayConfigMutex.RLock() + defer fake.getPlayoutDelayConfigMutex.RUnlock() fake.getPublishedTrackMutex.RLock() defer fake.getPublishedTrackMutex.RUnlock() fake.getPublishedTracksMutex.RLock() @@ -5831,6 +5959,8 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} { defer fake.subscriptionPermissionMutex.RUnlock() fake.subscriptionPermissionUpdateMutex.RLock() defer fake.subscriptionPermissionUpdateMutex.RUnlock() + fake.supportSyncStreamIDMutex.RLock() + defer fake.supportSyncStreamIDMutex.RUnlock() fake.toProtoMutex.RLock() defer fake.toProtoMutex.RUnlock() fake.toProtoWithVersionMutex.RLock() diff --git a/pkg/rtc/types/typesfakes/fake_media_track.go b/pkg/rtc/types/typesfakes/fake_media_track.go index bb0c07e31..af0c8c1b0 100644 --- a/pkg/rtc/types/typesfakes/fake_media_track.go +++ b/pkg/rtc/types/typesfakes/fake_media_track.go @@ -236,6 +236,16 @@ type FakeMediaTrack struct { sourceReturnsOnCall map[int]struct { result1 livekit.TrackSource } + StreamStub func() string + streamMutex sync.RWMutex + streamArgsForCall []struct { + } + streamReturns struct { + result1 string + } + streamReturnsOnCall map[int]struct { + result1 string + } ToProtoStub func() *livekit.TrackInfo toProtoMutex sync.RWMutex toProtoArgsForCall []struct { @@ -1474,6 +1484,59 @@ func (fake *FakeMediaTrack) SourceReturnsOnCall(i int, result1 livekit.TrackSour }{result1} } +func (fake *FakeMediaTrack) Stream() string { + fake.streamMutex.Lock() + ret, specificReturn := fake.streamReturnsOnCall[len(fake.streamArgsForCall)] + fake.streamArgsForCall = append(fake.streamArgsForCall, struct { + }{}) + stub := fake.StreamStub + fakeReturns := fake.streamReturns + fake.recordInvocation("Stream", []interface{}{}) + fake.streamMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeMediaTrack) StreamCallCount() int { + fake.streamMutex.RLock() + defer fake.streamMutex.RUnlock() + return len(fake.streamArgsForCall) +} + +func (fake *FakeMediaTrack) StreamCalls(stub func() string) { + fake.streamMutex.Lock() + defer fake.streamMutex.Unlock() + fake.StreamStub = stub +} + +func (fake *FakeMediaTrack) StreamReturns(result1 string) { + fake.streamMutex.Lock() + defer fake.streamMutex.Unlock() + fake.StreamStub = nil + fake.streamReturns = struct { + result1 string + }{result1} +} + +func (fake *FakeMediaTrack) StreamReturnsOnCall(i int, result1 string) { + fake.streamMutex.Lock() + defer fake.streamMutex.Unlock() + fake.StreamStub = nil + if fake.streamReturnsOnCall == nil { + fake.streamReturnsOnCall = make(map[int]struct { + result1 string + }) + } + fake.streamReturnsOnCall[i] = struct { + result1 string + }{result1} +} + func (fake *FakeMediaTrack) ToProto() *livekit.TrackInfo { fake.toProtoMutex.Lock() ret, specificReturn := fake.toProtoReturnsOnCall[len(fake.toProtoArgsForCall)] @@ -1615,6 +1678,8 @@ func (fake *FakeMediaTrack) Invocations() map[string][][]interface{} { defer fake.setMutedMutex.RUnlock() fake.sourceMutex.RLock() defer fake.sourceMutex.RUnlock() + fake.streamMutex.RLock() + defer fake.streamMutex.RUnlock() fake.toProtoMutex.RLock() defer fake.toProtoMutex.RUnlock() fake.updateVideoLayersMutex.RLock() diff --git a/pkg/rtc/utils.go b/pkg/rtc/utils.go index b9b532d8b..661646204 100644 --- a/pkg/rtc/utils.go +++ b/pkg/rtc/utils.go @@ -42,6 +42,25 @@ func PackStreamID(participantID livekit.ParticipantID, trackID livekit.TrackID) return string(participantID) + trackIdSeparator + string(trackID) } +func PackSyncStreamID(participantID livekit.ParticipantID, stream string) string { + return string(participantID) + trackIdSeparator + stream +} + +func StreamFromTrackSource(source livekit.TrackSource) string { + // group camera/mic, screenshare/audio together + switch source { + case livekit.TrackSource_SCREEN_SHARE: + return "screen" + case livekit.TrackSource_SCREEN_SHARE_AUDIO: + return "screen" + case livekit.TrackSource_CAMERA: + return "camera" + case livekit.TrackSource_MICROPHONE: + return "camera" + } + return "unknown" +} + func PackDataTrackLabel(participantID livekit.ParticipantID, trackID livekit.TrackID, label string) string { return string(participantID) + trackIdSeparator + string(trackID) + trackIdSeparator + label } diff --git a/pkg/service/roomallocator.go b/pkg/service/roomallocator.go index 9b38b9768..f07545c52 100644 --- a/pkg/service/roomallocator.go +++ b/pkg/service/roomallocator.go @@ -151,4 +151,9 @@ func applyDefaultRoomConfig(room *livekit.Room, conf *config.RoomConfig) { FmtpLine: codec.FmtpLine, }) } + room.PlayoutDelay = &livekit.PlayoutDelay{ + Enabled: conf.PlayoutDelay.Enabled, + Min: uint32(conf.PlayoutDelay.Min), + Max: uint32(conf.PlayoutDelay.Max), + } } diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index fe4008cb3..5834a3805 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -390,6 +390,7 @@ func (r *RoomManager) StartSession( SubscriberAllowPause: subscriberAllowPause, SubscriptionLimitAudio: r.config.Limit.SubscriptionLimitAudio, SubscriptionLimitVideo: r.config.Limit.SubscriptionLimitVideo, + PlayoutDelay: protoRoom.PlayoutDelay, }) if err != nil { return err diff --git a/pkg/sfu/buffer/buffer.go b/pkg/sfu/buffer/buffer.go index 4ed4c0a8b..fa7bb8352 100644 --- a/pkg/sfu/buffer/buffer.go +++ b/pkg/sfu/buffer/buffer.go @@ -181,7 +181,7 @@ func (b *Buffer) Bind(params webrtc.RTPParameters, codec webrtc.RTPCodecCapabili for _, ext := range params.HeaderExtensions { switch ext.URI { - case dd.ExtensionUrl: + case dd.ExtensionURI: b.ddExt = uint8(ext.ID) frc := NewFrameRateCalculatorDD(b.clockRate, b.logger) for i := range b.frameRateCalculator { diff --git a/pkg/sfu/dependencydescriptor/dependencydescriptorextension.go b/pkg/sfu/dependencydescriptor/dependencydescriptorextension.go index b573d7d60..3e7ed2510 100644 --- a/pkg/sfu/dependencydescriptor/dependencydescriptorextension.go +++ b/pkg/sfu/dependencydescriptor/dependencydescriptorextension.go @@ -69,7 +69,7 @@ const ( AllChainsAreActive = uint32(0) - ExtensionUrl = "https://aomediacodec.github.io/av1-rtp-spec/#dependency-descriptor-rtp-header-extension" + ExtensionURI = "https://aomediacodec.github.io/av1-rtp-spec/#dependency-descriptor-rtp-header-extension" ) // ------------------------------------------------------------------------------ diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index 16c8f4456..d1d10b92c 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -37,6 +37,7 @@ import ( "github.com/livekit/livekit-server/pkg/sfu/connectionquality" dd "github.com/livekit/livekit-server/pkg/sfu/dependencydescriptor" "github.com/livekit/livekit-server/pkg/sfu/pacer" + "github.com/livekit/livekit-server/pkg/sfu/rtpextension" ) // TrackSender defines an interface send media to remote peer @@ -183,6 +184,19 @@ type DownTrackStreamAllocatorListener interface { type ReceiverReportListener func(dt *DownTrack, report *rtcp.ReceiverReport) +type DowntrackParams struct { + Codecs []webrtc.RTPCodecParameters + Receiver TrackReceiver + BufferFactory *buffer.Factory + SubID livekit.ParticipantID + StreamID string + MaxTrack int + PlayoutDelayLimit *livekit.PlayoutDelay + Pacer pacer.Pacer + Logger logger.Logger + Trailer []byte +} + // DownTrack implements TrackLocal, is the track used to write packets // to SFU Subscriber, the track handle the packets for simple, simulcast // and SVC Publisher. @@ -192,17 +206,13 @@ type ReceiverReportListener func(dt *DownTrack, report *rtcp.ReceiverReport) // - closed // once closed, a DownTrack cannot be re-used. type DownTrack struct { - logger logger.Logger - id livekit.TrackID - subscriberID livekit.ParticipantID - kind webrtc.RTPCodecType - mime string - ssrc uint32 - streamID string - maxTrack int - payloadType uint8 - sequencer *sequencer - bufferFactory *buffer.Factory + params DowntrackParams + id livekit.TrackID + kind webrtc.RTPCodecType + mime string + ssrc uint32 + payloadType uint8 + sequencer *sequencer forwarder *Forwarder @@ -211,8 +221,8 @@ type DownTrack struct { absSendTimeExtID int transportWideExtID int dependencyDescriptorExtID int - receiver TrackReceiver - transceiver *webrtc.RTPTransceiver + playoutDelayExtID int + transceiver atomic.Pointer[webrtc.RTPTransceiver] writeStream webrtc.TrackLocalWriter rtcpReader *buffer.RTCPReader @@ -250,12 +260,13 @@ type DownTrack struct { bytesSent atomic.Uint32 bytesRetransmitted atomic.Uint32 + playoutDelayBytes atomic.Value //bytes of marshalled playout delay + playoudDelayAcked atomic.Bool + pacer pacer.Pacer maxLayerNotifierCh chan struct{} - trailer []byte - cbMu sync.RWMutex onStatsUpdate func(dt *DownTrack, stat *livekit.AnalyticsStat) onMaxSubscribedLayerChanged func(dt *DownTrack, layer int32) @@ -264,16 +275,8 @@ type DownTrack struct { } // NewDownTrack returns a DownTrack. -func NewDownTrack( - codecs []webrtc.RTPCodecParameters, - r TrackReceiver, - bf *buffer.Factory, - subID livekit.ParticipantID, - mt int, - pacer pacer.Pacer, - trailer []byte, - logger logger.Logger, -) (*DownTrack, error) { +func NewDownTrack(params DowntrackParams) (*DownTrack, error) { + codecs := params.Codecs var kind webrtc.RTPCodecType switch { case strings.HasPrefix(codecs[0].MimeType, "audio/"): @@ -285,24 +288,18 @@ func NewDownTrack( } d := &DownTrack{ - logger: logger, - id: r.TrackID(), - subscriberID: subID, - maxTrack: mt, - streamID: r.StreamID(), - bufferFactory: bf, - receiver: r, + params: params, + id: params.Receiver.TrackID(), upstreamCodecs: codecs, kind: kind, codec: codecs[0].RTPCodecCapability, - pacer: pacer, - trailer: trailer, + pacer: params.Pacer, maxLayerNotifierCh: make(chan struct{}, 20), } d.forwarder = NewForwarder( d.kind, - d.logger, - d.receiver.GetReferenceLayerRTPTimestamp, + params.Logger, + d.params.Receiver.GetReferenceLayerRTPTimestamp, d.getExpectedRTPTimestamp, ) d.forwarder.OnParkedLayerExpired(func() { @@ -314,7 +311,7 @@ func NewDownTrack( d.rtpStats = buffer.NewRTPStats(buffer.RTPStatsParams{ ClockRate: d.codec.ClockRate, IsReceiverReportDriven: true, - Logger: d.logger, + Logger: params.Logger, }) d.deltaStatsSnapshotId = d.rtpStats.NewSnapshotId() d.deltaStatsOverriddenSnapshotId = d.rtpStats.NewSnapshotId() @@ -325,7 +322,7 @@ func NewDownTrack( GetDeltaStats: d.getDeltaStats, GetDeltaStatsOverridden: d.getDeltaStatsOverridden, GetLastReceiverReportTime: func() time.Time { return d.rtpStats.LastReceiverReport() }, - Logger: d.logger.WithValues("direction", "down"), + Logger: params.Logger.WithValues("direction", "down"), }) d.connectionStats.OnStatsUpdate(func(_cs *connectionquality.ConnectionStats, stat *livekit.AnalyticsStat) { if onStatsUpdate := d.getOnStatsUpdate(); onStatsUpdate != nil { @@ -333,6 +330,18 @@ func NewDownTrack( } }) + // set initial playout delay to minimum value + + if d.params.PlayoutDelayLimit.GetEnabled() && d.params.PlayoutDelayLimit.GetMin() > 0 { + delay := rtpextension.PlayoutDelayFromValue( + uint16(d.params.PlayoutDelayLimit.GetMin()), + uint16(d.params.PlayoutDelayLimit.GetMax()), + ) + b, err := delay.Marshal() + if err == nil { + d.playoutDelayBytes.Store(b) + } + } if d.kind == webrtc.RTPCodecTypeVideo { go d.maxLayerNotifierWorker() } @@ -362,7 +371,7 @@ func (d *DownTrack) Bind(t webrtc.TrackLocalContext) (webrtc.RTPCodecParameters, err := webrtc.ErrUnsupportedCodec onBinding := d.onBinding d.bindLock.Unlock() - d.logger.Infow("bind error for unsupported codec", "codecs", d.upstreamCodecs, "remoteParameters", t.CodecParameters()) + d.params.Logger.Infow("bind error for unsupported codec", "codecs", d.upstreamCodecs, "remoteParameters", t.CodecParameters()) if onBinding != nil { onBinding(err) } @@ -371,17 +380,17 @@ func (d *DownTrack) Bind(t webrtc.TrackLocalContext) (webrtc.RTPCodecParameters, // if a downtrack is closed before bind, it already unsubscribed from client, don't do subsequent operation and return here. if d.IsClosed() { - d.logger.Debugw("DownTrack closed before bind") + d.params.Logger.Debugw("DownTrack closed before bind") d.bindLock.Unlock() return codec, nil } - d.logger.Debugw("DownTrack.Bind", "codecs", d.upstreamCodecs, "matchCodec", codec, "ssrc", t.SSRC()) + d.params.Logger.Debugw("DownTrack.Bind", "codecs", d.upstreamCodecs, "matchCodec", codec, "ssrc", t.SSRC()) d.ssrc = uint32(t.SSRC()) d.payloadType = uint8(codec.PayloadType) d.writeStream = t.WriteStream() d.mime = strings.ToLower(codec.MimeType) - if rr := d.bufferFactory.GetOrNew(packetio.RTCPBufferPacket, uint32(t.SSRC())).(*buffer.RTCPReader); rr != nil { + if rr := d.params.BufferFactory.GetOrNew(packetio.RTCPBufferPacket, uint32(t.SSRC())).(*buffer.RTCPReader); rr != nil { rr.OnPacket(func(pkt []byte) { d.handleRTCP(pkt) }) @@ -389,9 +398,9 @@ func (d *DownTrack) Bind(t webrtc.TrackLocalContext) (webrtc.RTPCodecParameters, } if d.kind == webrtc.RTPCodecTypeAudio { - d.sequencer = newSequencer(d.maxTrack, 0, d.logger) + d.sequencer = newSequencer(d.params.MaxTrack, 0, d.params.Logger) } else { - d.sequencer = newSequencer(d.maxTrack, maxPadding, d.logger) + d.sequencer = newSequencer(d.params.MaxTrack, maxPadding, d.params.Logger) } d.codec = codec.RTPCodecCapability @@ -401,9 +410,18 @@ func (d *DownTrack) Bind(t webrtc.TrackLocalContext) (webrtc.RTPCodecParameters, d.bound.Store(true) d.bindLock.Unlock() - d.forwarder.DetermineCodec(d.codec, d.receiver.HeaderExtensions()) + // Bind is called under RTPSender.mu lock, call the RTPSender.GetParameters in goroutine to avoid deadlock + go func() { + if tr := d.transceiver.Load(); tr != nil { + extensions := tr.Sender().GetParameters().HeaderExtensions + d.params.Logger.Debugw("negotiated downtrack extensions", "extensions", extensions) + d.SetRTPHeaderExtensions(extensions) + } + }() - d.logger.Debugw("downtrack bound") + d.forwarder.DetermineCodec(d.codec, d.params.Receiver.HeaderExtensions()) + + d.params.Logger.Debugw("downtrack bound") d.onBindAndConnected() return codec, nil @@ -417,7 +435,7 @@ func (d *DownTrack) Unbind(_ webrtc.TrackLocalContext) error { } func (d *DownTrack) TrackInfoAvailable() { - ti := d.receiver.TrackInfo() + ti := d.params.Receiver.TrackInfo() if ti == nil { return } @@ -495,9 +513,9 @@ func (d *DownTrack) ID() string { return string(d.id) } func (d *DownTrack) Codec() webrtc.RTPCodecCapability { return d.codec } // StreamID is the group this track belongs too. This must be unique -func (d *DownTrack) StreamID() string { return d.streamID } +func (d *DownTrack) StreamID() string { return d.params.StreamID } -func (d *DownTrack) SubscriberID() livekit.ParticipantID { return d.subscriberID } +func (d *DownTrack) SubscriberID() livekit.ParticipantID { return d.params.SubID } // Sets RTP header extensions for this track func (d *DownTrack) SetRTPHeaderExtensions(rtpHeaderExtensions []webrtc.RTPHeaderExtensionParameter) { @@ -505,10 +523,12 @@ func (d *DownTrack) SetRTPHeaderExtensions(rtpHeaderExtensions []webrtc.RTPHeade switch ext.URI { case sdp.ABSSendTimeURI: d.absSendTimeExtID = ext.ID + case dd.ExtensionURI: + d.dependencyDescriptorExtID = ext.ID + case rtpextension.PlayoutDelayURI: + d.playoutDelayExtID = ext.ID case sdp.TransportCCURI: d.transportWideExtID = ext.ID - case dd.ExtensionUrl: - d.dependencyDescriptorExtID = ext.ID } } } @@ -528,18 +548,18 @@ func (d *DownTrack) SSRC() uint32 { } func (d *DownTrack) Stop() error { - if d.transceiver != nil { - return d.transceiver.Stop() + if tr := d.transceiver.Load(); tr != nil { + return tr.Stop() } return errors.New("downtrack transceiver does not exist") } func (d *DownTrack) SetTransceiver(transceiver *webrtc.RTPTransceiver) { - d.transceiver = transceiver + d.transceiver.Store(transceiver) } func (d *DownTrack) GetTransceiver() *webrtc.RTPTransceiver { - return d.transceiver + return d.transceiver.Load() } func (d *DownTrack) maybeStartKeyFrameRequester() { @@ -551,7 +571,6 @@ func (d *DownTrack) maybeStartKeyFrameRequester() { // d.stopKeyFrameRequester() - // SVC-TODO : don't need pli/lrr when layer comes down locked, layer := d.forwarder.CheckSync() if !locked { go d.keyFrameRequester(d.keyFrameRequestGeneration.Load(), layer) @@ -584,8 +603,8 @@ func (d *DownTrack) keyFrameRequester(generation uint32, layer int32) { } if d.connected.Load() { - d.logger.Debugw("sending PLI for layer lock", "generation", generation, "layer", layer) - d.receiver.SendPLI(layer, false) + d.params.Logger.Debugw("sending PLI for layer lock", "generation", generation, "layer", layer) + d.params.Receiver.SendPLI(layer, false) d.rtpStats.UpdateLayerLockPliAndTime(1) } @@ -605,7 +624,7 @@ func (d *DownTrack) postMaxLayerNotifierEvent() { select { case d.maxLayerNotifierCh <- struct{}{}: default: - d.logger.Warnw("max layer notifier event queue full", nil) + d.params.Logger.Warnw("max layer notifier event queue full", nil) } } @@ -619,7 +638,7 @@ func (d *DownTrack) maxLayerNotifierWorker() { maxLayerSpatial = d.forwarder.GetMaxSubscribedSpatial() } if onMaxSubscribedLayerChanged := d.getOnMaxLayerChanged(); onMaxSubscribedLayerChanged != nil { - d.logger.Infow("max subscribed layer changed", "maxLayerSpatial", maxLayerSpatial) + d.params.Logger.Infow("max subscribed layer changed", "maxLayerSpatial", maxLayerSpatial) onMaxSubscribedLayerChanged(d, maxLayerSpatial) } } @@ -634,7 +653,7 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error { tp, err := d.forwarder.GetTranslationParams(extPkt, layer) if tp.shouldDrop { if err != nil { - d.logger.Errorw("write rtp packet failed", err) + d.params.Logger.Errorw("write rtp packet failed", err) } return err } @@ -654,13 +673,19 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error { hdr, err := d.getTranslatedRTPHeader(extPkt, tp) if err != nil { - d.logger.Errorw("write rtp packet failed", err) + d.params.Logger.Errorw("write rtp packet failed", err) if pool != nil { PacketFactory.Put(pool) } return err } + extensions := []pacer.ExtensionData{{ID: uint8(d.dependencyDescriptorExtID), Payload: tp.ddBytes}} + if d.playoutDelayExtID != 0 && !d.playoudDelayAcked.Load() { + if val := d.playoutDelayBytes.Load(); val != nil { + extensions = append(extensions, pacer.ExtensionData{ID: uint8(d.playoutDelayExtID), Payload: val.([]byte)}) + } + } if d.sequencer != nil { d.sequencer.push( extPkt.Packet.SequenceNumber, @@ -675,7 +700,7 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error { d.pacer.Enqueue(pacer.Packet{ Header: hdr, - Extensions: []pacer.ExtensionData{{ID: uint8(d.dependencyDescriptorExtID), Payload: tp.ddBytes}}, + Extensions: extensions, Payload: payload, AbsSendTimeExtID: uint8(d.absSendTimeExtID), TransportWideExtID: uint8(d.transportWideExtID), @@ -869,7 +894,7 @@ func (d *DownTrack) CloseWithFlush(flush bool) { } d.bindLock.Lock() - d.logger.Debugw("close down track", "flushBlankFrame", flush) + d.params.Logger.Debugw("close down track", "flushBlankFrame", flush) if d.bound.Load() { if d.forwarder != nil { d.forwarder.Mute(true) @@ -893,12 +918,12 @@ func (d *DownTrack) CloseWithFlush(flush bool) { } d.bound.Store(false) - d.logger.Debugw("closing sender", "kind", d.kind) + d.params.Logger.Debugw("closing sender", "kind", d.kind) } - d.receiver.DeleteDownTrack(d.subscriberID) + d.params.Receiver.DeleteDownTrack(d.params.SubID) if d.rtcpReader != nil && flush { - d.logger.Debugw("downtrack close rtcp reader") + d.params.Logger.Debugw("downtrack close rtcp reader") d.rtcpReader.Close() d.rtcpReader.OnPacket(nil) } @@ -906,7 +931,7 @@ func (d *DownTrack) CloseWithFlush(flush bool) { d.bindLock.Unlock() d.connectionStats.Close() d.rtpStats.Stop() - d.logger.Infow("rtp stats", "direction", "downstream", "mime", d.mime, "ssrc", d.ssrc, "stats", d.rtpStats.ToString()) + d.params.Logger.Infow("rtp stats", "direction", "downstream", "mime", d.mime, "ssrc", d.ssrc, "stats", d.rtpStats.ToString()) close(d.maxLayerNotifierCh) @@ -1088,17 +1113,17 @@ func (d *DownTrack) IsDeficient() bool { } func (d *DownTrack) BandwidthRequested() int64 { - _, brs := d.receiver.GetLayeredBitrate() + _, brs := d.params.Receiver.GetLayeredBitrate() return d.forwarder.BandwidthRequested(brs) } func (d *DownTrack) DistanceToDesired() float64 { - al, brs := d.receiver.GetLayeredBitrate() + al, brs := d.params.Receiver.GetLayeredBitrate() return d.forwarder.DistanceToDesired(al, brs) } func (d *DownTrack) AllocateOptimal(allowOvershoot bool) VideoAllocation { - al, brs := d.receiver.GetLayeredBitrate() + al, brs := d.params.Receiver.GetLayeredBitrate() allocation := d.forwarder.AllocateOptimal(al, brs, allowOvershoot) d.maybeStartKeyFrameRequester() d.maybeAddTransition(allocation.BandwidthNeeded, allocation.DistanceToDesired, allocation.PauseReason) @@ -1106,7 +1131,7 @@ func (d *DownTrack) AllocateOptimal(allowOvershoot bool) VideoAllocation { } func (d *DownTrack) ProvisionalAllocatePrepare() { - al, brs := d.receiver.GetLayeredBitrate() + al, brs := d.params.Receiver.GetLayeredBitrate() d.forwarder.ProvisionalAllocatePrepare(al, brs) } @@ -1120,13 +1145,13 @@ func (d *DownTrack) ProvisionalAllocate(availableChannelCapacity int64, layers b func (d *DownTrack) ProvisionalAllocateGetCooperativeTransition(allowOvershoot bool) VideoTransition { transition := d.forwarder.ProvisionalAllocateGetCooperativeTransition(allowOvershoot) - d.logger.Debugw("stream: cooperative transition", "transition", transition) + d.params.Logger.Debugw("stream: cooperative transition", "transition", transition) return transition } func (d *DownTrack) ProvisionalAllocateGetBestWeightedTransition() VideoTransition { transition := d.forwarder.ProvisionalAllocateGetBestWeightedTransition() - d.logger.Debugw("stream: best weighted transition", "transition", transition) + d.params.Logger.Debugw("stream: best weighted transition", "transition", transition) return transition } @@ -1138,7 +1163,7 @@ func (d *DownTrack) ProvisionalAllocateCommit() VideoAllocation { } func (d *DownTrack) AllocateNextHigher(availableChannelCapacity int64, allowOvershoot bool) (VideoAllocation, bool) { - al, brs := d.receiver.GetLayeredBitrate() + al, brs := d.params.Receiver.GetLayeredBitrate() allocation, available := d.forwarder.AllocateNextHigher(availableChannelCapacity, al, brs, allowOvershoot) d.maybeStartKeyFrameRequester() d.maybeAddTransition(allocation.BandwidthNeeded, allocation.DistanceToDesired, allocation.PauseReason) @@ -1146,14 +1171,14 @@ func (d *DownTrack) AllocateNextHigher(availableChannelCapacity int64, allowOver } func (d *DownTrack) GetNextHigherTransition(allowOvershoot bool) (VideoTransition, bool) { - _, brs := d.receiver.GetLayeredBitrate() + _, brs := d.params.Receiver.GetLayeredBitrate() transition, available := d.forwarder.GetNextHigherTransition(brs, allowOvershoot) - d.logger.Debugw("stream: get next higher layer", "transition", transition, "available", available, "bitrates", brs) + d.params.Logger.Debugw("stream: get next higher layer", "transition", transition, "available", available, "bitrates", brs) return transition, available } func (d *DownTrack) Pause() VideoAllocation { - al, brs := d.receiver.GetLayeredBitrate() + al, brs := d.params.Receiver.GetLayeredBitrate() allocation := d.forwarder.Pause(al, brs) d.maybeStartKeyFrameRequester() d.maybeAddTransition(allocation.BandwidthNeeded, allocation.DistanceToDesired, allocation.PauseReason) @@ -1165,7 +1190,7 @@ func (d *DownTrack) Resync() { } func (d *DownTrack) CreateSourceDescriptionChunks() []rtcp.SourceDescriptionChunk { - if !d.bound.Load() { + if !d.bound.Load() || d.transceiver.Load() == nil { return nil } return []rtcp.SourceDescriptionChunk{ @@ -1173,13 +1198,13 @@ func (d *DownTrack) CreateSourceDescriptionChunks() []rtcp.SourceDescriptionChun Source: d.ssrc, Items: []rtcp.SourceDescriptionItem{{ Type: rtcp.SDESCNAME, - Text: d.streamID, + Text: d.params.StreamID, }}, }, { Source: d.ssrc, Items: []rtcp.SourceDescriptionItem{{ Type: rtcp.SDESType(15), - Text: d.transceiver.Mid(), + Text: d.transceiver.Load().Mid(), }}, }, } @@ -1194,7 +1219,7 @@ func (d *DownTrack) CreateSenderReport() *rtcp.SenderReport { if clockLayer == buffer.InvalidLayerSpatial { clockLayer = d.forwarder.GetReferenceLayerSpatial() } - return d.rtpStats.GetRtcpSenderReport(d.ssrc, d.receiver.GetCalculatedClockRate(clockLayer)) + return d.rtpStats.GetRtcpSenderReport(d.ssrc, d.params.Receiver.GetCalculatedClockRate(clockLayer)) } func (d *DownTrack) writeBlankFrameRTP(duration float32, generation uint32) chan struct{} { @@ -1243,7 +1268,7 @@ func (d *DownTrack) writeBlankFrameRTP(duration float32, generation uint32) chan snts, frameEndNeeded, err := d.forwarder.GetSnTsForBlankFrames(frameRate, 1) if err != nil { - d.logger.Warnw("could not get SN/TS for blank frame", err) + d.params.Logger.Warnw("could not get SN/TS for blank frame", err) close(done) return } @@ -1262,7 +1287,7 @@ func (d *DownTrack) writeBlankFrameRTP(duration float32, generation uint32) chan payload, err := getBlankFrame(frameEndNeeded) if err != nil { - d.logger.Warnw("could not get blank frame", err) + d.params.Logger.Warnw("could not get blank frame", err) close(done) return } @@ -1293,13 +1318,13 @@ func (d *DownTrack) writeBlankFrameRTP(duration float32, generation uint32) chan } func (d *DownTrack) maybeAddTrailer(buf []byte) int { - if len(buf) < len(d.trailer) { - d.logger.Warnw("trailer too big", nil, "bufLen", len(buf), "trailerLen", len(d.trailer)) + if len(buf) < len(d.params.Trailer) { + d.params.Logger.Warnw("trailer too big", nil, "bufLen", len(buf), "trailerLen", len(d.params.Trailer)) return 0 } - copy(buf, d.trailer) - return len(d.trailer) + copy(buf, d.params.Trailer) + return len(d.params.Trailer) } func (d *DownTrack) getOpusBlankFrame(_frameEndNeeded bool) ([]byte, error) { @@ -1366,7 +1391,7 @@ func (d *DownTrack) getH264BlankFrame(_frameEndNeeded bool) ([]byte, error) { func (d *DownTrack) handleRTCP(bytes []byte) { pkts, err := rtcp.Unmarshal(bytes) if err != nil { - d.logger.Errorw("unmarshal rtcp receiver packets err", err) + d.params.Logger.Errorw("unmarshal rtcp receiver packets err", err) return } @@ -1375,8 +1400,8 @@ func (d *DownTrack) handleRTCP(bytes []byte) { if pliOnce { _, layer := d.forwarder.CheckSync() if layer != buffer.InvalidLayerSpatial && !d.forwarder.IsAnyMuted() { - d.logger.Debugw("sending PLI RTCP", "layer", layer) - d.receiver.SendPLI(layer, false) + d.params.Logger.Debugw("sending PLI RTCP", "layer", layer) + d.params.Receiver.SendPLI(layer, false) d.isNACKThrottled.Store(true) d.rtpStats.UpdatePliTime() pliOnce = false @@ -1424,6 +1449,8 @@ func (d *DownTrack) handleRTCP(bytes []byte) { if sal := d.getStreamAllocatorListener(); sal != nil { sal.OnRTCPReceiverReport(d, r) } + + d.playoudDelayAcked.Store(true) } if len(rr.Reports) > 0 { d.listenerLock.RLock() @@ -1513,7 +1540,7 @@ func (d *DownTrack) retransmitPackets(nacks []uint16) { }) pktBuff := *src - n, err := d.receiver.ReadRTP(pktBuff, uint8(meta.layer), meta.sourceSeqNo) + n, err := d.params.Receiver.ReadRTP(pktBuff, uint8(meta.layer), meta.sourceSeqNo) if err != nil { if err == io.EOF { break @@ -1528,7 +1555,7 @@ func (d *DownTrack) retransmitPackets(nacks []uint16) { var pkt rtp.Packet if err = pkt.Unmarshal(pktBuff[:n]); err != nil { - d.logger.Errorw("unmarshalling rtp packet failed in retransmit", err) + d.params.Logger.Errorw("unmarshalling rtp packet failed in retransmit", err) continue } pkt.Header.Marker = meta.marker @@ -1542,7 +1569,7 @@ func (d *DownTrack) retransmitPackets(nacks []uint16) { if d.mime == "video/vp8" && len(pkt.Payload) > 0 && len(meta.codecBytes) != 0 { var incomingVP8 buffer.VP8 if err = incomingVP8.Unmarshal(pkt.Payload); err != nil { - d.logger.Errorw("unmarshalling VP8 packet err", err) + d.params.Logger.Errorw("unmarshalling VP8 packet err", err) PacketFactory.Put(pool) continue } @@ -1633,9 +1660,9 @@ func (d *DownTrack) DebugInfo() map[string]interface{} { } return map[string]interface{}{ - "SubscriberID": d.subscriberID, + "SubscriberID": d.params.SubID, "TrackID": d.id, - "StreamID": d.streamID, + "StreamID": d.params.StreamID, "SSRC": d.ssrc, "MimeType": d.codec.MimeType, "Bound": d.bound.Load(), @@ -1697,7 +1724,7 @@ func (d *DownTrack) onBindAndConnected() { if d.kind == webrtc.RTPCodecTypeVideo { _, layer := d.forwarder.CheckSync() if layer != buffer.InvalidLayerSpatial { - d.receiver.SendPLI(layer, true) + d.params.Receiver.SendPLI(layer, true) } } @@ -1711,7 +1738,7 @@ func (d *DownTrack) sendPaddingOnMute() { // let uptrack have chance to send packet before we send padding time.Sleep(waitBeforeSendPaddingOnMute) - d.logger.Debugw("sending padding on mute") + d.params.Logger.Debugw("sending padding on mute") if d.kind == webrtc.RTPCodecTypeVideo { d.sendPaddingOnMuteForVideo() } else if d.mime == "audio/opus" { @@ -1741,7 +1768,7 @@ func (d *DownTrack) sendSilentFrameOnMuteForOpus() { } snts, _, err := d.forwarder.GetSnTsForBlankFrames(frameRate, 1) if err != nil { - d.logger.Warnw("could not get SN/TS for blank frame", err) + d.params.Logger.Warnw("could not get SN/TS for blank frame", err) return } for i := 0; i < len(snts); i++ { @@ -1758,7 +1785,7 @@ func (d *DownTrack) sendSilentFrameOnMuteForOpus() { payload, err := d.getOpusBlankFrame(false) if err != nil { - d.logger.Warnw("could not get blank frame", err) + d.params.Logger.Warnw("could not get blank frame", err) return } @@ -1801,7 +1828,7 @@ type sendPacketMetadata struct { func (d *DownTrack) packetSent(md interface{}, hdr *rtp.Header, payloadSize int, sendTime time.Time, sendError error) { spmd, ok := md.(sendPacketMetadata) if !ok { - d.logger.Errorw("invalid send packet metadata", nil) + d.params.Logger.Errorw("invalid send packet metadata", nil) return } @@ -1840,7 +1867,7 @@ func (d *DownTrack) packetSent(md interface{}, hdr *rtp.Header, payloadSize int, if spmd.isKeyFrame { d.isNACKThrottled.Store(false) d.rtpStats.UpdateKeyFrame(1) - d.logger.Debugw( + d.params.Logger.Debugw( "forwarded key frame", "layer", spmd.layer, "rtpsn", hdr.SequenceNumber, diff --git a/pkg/sfu/forwarder.go b/pkg/sfu/forwarder.go index 2da3ec9ef..ec7db7753 100644 --- a/pkg/sfu/forwarder.go +++ b/pkg/sfu/forwarder.go @@ -309,7 +309,7 @@ func (f *Forwarder) DetermineCodec(codec webrtc.RTPCodecCapability, extensions [ searchDone: for _, ext := range extensions { switch ext.URI { - case dd.ExtensionUrl: + case dd.ExtensionURI: isDDAvailable = true break searchDone } diff --git a/pkg/sfu/receiver.go b/pkg/sfu/receiver.go index 46508acb5..1b36811d5 100644 --- a/pkg/sfu/receiver.go +++ b/pkg/sfu/receiver.go @@ -126,8 +126,8 @@ type WebRTCReceiver struct { onStatsUpdate func(w *WebRTCReceiver, stat *livekit.AnalyticsStat) onMaxLayerChange func(maxLayer int32) - primaryReceiver atomic.Value // *RedPrimaryReceiver - redReceiver atomic.Value // *RedReceiver + primaryReceiver atomic.Pointer[RedPrimaryReceiver] + redReceiver atomic.Pointer[RedReceiver] redPktWriter func(pkt *buffer.ExtPacket, spatialLayer int32) } @@ -232,7 +232,7 @@ func NewWebRTCReceiver( w.connectionStats.Start(w.trackInfo) for _, ext := range receiver.GetParameters().HeaderExtensions { - if ext.URI == dd.ExtensionUrl { + if ext.URI == dd.ExtensionURI { w.streamTrackerManager.AddDependencyDescriptorTrackers() break } @@ -624,10 +624,10 @@ func (w *WebRTCReceiver) forwardRTP(layer int32) { w.closed.Store(true) w.closeTracks() if pr := w.primaryReceiver.Load(); pr != nil { - pr.(*RedPrimaryReceiver).Close() + pr.Close() } if pr := w.redReceiver.Load(); pr != nil { - pr.(*RedReceiver).Close() + pr.Close() } }) @@ -731,7 +731,7 @@ func (w *WebRTCReceiver) GetPrimaryReceiverForRed() TrackReceiver { w.bufferMu.Unlock() } } - return w.primaryReceiver.Load().(*RedPrimaryReceiver) + return w.primaryReceiver.Load() } func (w *WebRTCReceiver) GetRedReceiver() TrackReceiver { @@ -750,7 +750,7 @@ func (w *WebRTCReceiver) GetRedReceiver() TrackReceiver { w.bufferMu.Unlock() } } - return w.redReceiver.Load().(*RedReceiver) + return w.redReceiver.Load() } func (w *WebRTCReceiver) GetTemporalLayerFpsForSpatial(layer int32) []float32 { diff --git a/pkg/sfu/rtpextension/playoutdelay.go b/pkg/sfu/rtpextension/playoutdelay.go new file mode 100644 index 000000000..2cb719269 --- /dev/null +++ b/pkg/sfu/rtpextension/playoutdelay.go @@ -0,0 +1,56 @@ +package rtpextension + +import ( + "encoding/binary" + "errors" +) + +const ( + PlayoutDelayURI = "http://www.webrtc.org/experiments/rtp-hdrext/playout-delay" + + playoutDelayExtensionSize = 3 +) + +var ( + errPlayoutDelayOverflow = errors.New("playout delay overflow") + errTooSmall = errors.New("buffer too small") +) + +// 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 +// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ +// | ID | len=2 | MIN delay | MAX delay | +// +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ +// The wired MIN/MAX delay is in 10ms unit + +type PlayOutDelay struct { + Min, Max uint16 // delay in ms +} + +func PlayoutDelayFromValue(min, max uint16) PlayOutDelay { + if min >= (1<<12)*10 { + min = (1<<12 - 1) * 10 + } + if max >= (1<<12)*10 { + max = (1<<12 - 1) * 10 + } + return PlayOutDelay{Min: min, Max: max} +} + +func (p PlayOutDelay) Marshal() ([]byte, error) { + min, max := p.Min/10, p.Max/10 + if min >= 1<<12 || max >= 1<<12 { + return nil, errPlayoutDelayOverflow + } + + return []byte{byte(min >> 4), byte(min<<4) | byte(max>>8), byte(max)}, nil +} + +func (p *PlayOutDelay) Unmarshal(rawData []byte) error { + if len(rawData) < playoutDelayExtensionSize { + return errTooSmall + } + + p.Min = (binary.BigEndian.Uint16(rawData) >> 4) * 10 + p.Max = (binary.BigEndian.Uint16(rawData[1:]) & 0x0FFF) * 10 + return nil +} diff --git a/pkg/sfu/rtpextension/playoutdelay_test.go b/pkg/sfu/rtpextension/playoutdelay_test.go new file mode 100644 index 000000000..a92f4cae5 --- /dev/null +++ b/pkg/sfu/rtpextension/playoutdelay_test.go @@ -0,0 +1,35 @@ +package rtpextension + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func TestPlayoutDelay(t *testing.T) { + p1 := PlayOutDelay{Min: 100, Max: 200} + b, err := p1.Marshal() + require.NoError(t, err) + require.Len(t, b, playoutDelayExtensionSize) + var p2 PlayOutDelay + err = p2.Unmarshal(b) + require.NoError(t, err) + require.Equal(t, p1, p2) + + // overflow + p3 := PlayOutDelay{Min: 100, Max: (1 << 12) * 10} + _, err = p3.Marshal() + require.ErrorIs(t, err, errPlayoutDelayOverflow) + + // too small + p4 := PlayOutDelay{} + err = p4.Unmarshal([]byte{0x00, 0x00}) + require.ErrorIs(t, err, errTooSmall) + + // from value + p5 := PlayoutDelayFromValue(1<<12*10, 1<<12*10+10) + _, err = p5.Marshal() + require.NoError(t, err) + require.Equal(t, uint16((1<<12)-1)*10, p5.Min) + require.Equal(t, uint16((1<<12)-1)*10, p5.Max) +} From f3a0e3e71c35fd677d9c1fa45513986a102ab5fd Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Wed, 2 Aug 2023 14:00:58 +0530 Subject: [PATCH 6/7] skip logging when stream closed (#1928) --- pkg/rtc/room.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index c0fce9d25..fa6a285ae 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -30,6 +30,7 @@ import ( "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" "github.com/livekit/protocol/utils" + "github.com/livekit/psrpc" "github.com/livekit/livekit-server/pkg/config" "github.com/livekit/livekit-server/pkg/routing" @@ -1310,7 +1311,7 @@ func BroadcastDataPacketForRoom(r types.Room, source types.LocalParticipant, dp utils.ParallelExec(destParticipants, dataForwardLoadBalanceThreshold, 1, func(op types.LocalParticipant) { err := op.SendDataPacket(dp, dpData) - if err != nil && !errors.Is(err, io.ErrClosedPipe) { + if err != nil && !errors.Is(err, io.ErrClosedPipe) && !errors.Is(err, psrpc.Canceled) { op.GetLogger().Infow("send data packet error", "error", err) } }) From 0dc92ef2739de0c263878af68750cf165ac6bffc Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Wed, 2 Aug 2023 14:02:29 +0530 Subject: [PATCH 7/7] Remove parked layer feature. (#1927) * Remove parked layer feature. Not worth the added complexity. Several reasons - Not seeing black frames on pub mute always. - If they are there, it can consume more than 30kbps if the parked layer is high res. That is wasted bandwidth downstream when pub is muted. - On resume, client some time sends PLI and that triggers a key frame request. But, leaving the separate `PubMuted` flag in forwarder in case we can use it for better handling. * need the request spatial --- pkg/rtc/subscribedtrack.go | 14 +- pkg/sfu/downtrack.go | 5 - pkg/sfu/forwarder.go | 121 ++---------------- pkg/sfu/forwarder_test.go | 23 +--- pkg/sfu/videolayerselector/base.go | 20 +-- .../dependencydescriptor.go | 4 - pkg/sfu/videolayerselector/simulcast.go | 52 +++----- .../videolayerselector/videolayerselector.go | 3 - 8 files changed, 36 insertions(+), 206 deletions(-) diff --git a/pkg/rtc/subscribedtrack.go b/pkg/rtc/subscribedtrack.go index c586ec3a7..10bf25fcb 100644 --- a/pkg/rtc/subscribedtrack.go +++ b/pkg/rtc/subscribedtrack.go @@ -215,14 +215,14 @@ func (t *SubscribedTrack) UpdateVideoLayer() { return } - t.logger.Debugw("updating video layer", - "settings", settings, - ) + t.logger.Debugw("updating video layer", "settings", settings) - spatial := t.spatialLayerFromSettings(settings) - t.DownTrack().SetMaxSpatialLayer(spatial) - if settings.Fps > 0 { - t.DownTrack().SetMaxTemporalLayer(t.MediaTrack().GetTemporalLayerForSpatialFps(spatial, settings.Fps, t.DownTrack().Codec().MimeType)) + if settings.Width > 0 || settings.Fps > 0 { + spatial := t.spatialLayerFromSettings(settings) + t.DownTrack().SetMaxSpatialLayer(spatial) + if settings.Fps > 0 { + t.DownTrack().SetMaxTemporalLayer(t.MediaTrack().GetTemporalLayerForSpatialFps(spatial, settings.Fps, t.DownTrack().Codec().MimeType)) + } } } diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index d1d10b92c..8f66820e2 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -302,11 +302,6 @@ func NewDownTrack(params DowntrackParams) (*DownTrack, error) { d.params.Receiver.GetReferenceLayerRTPTimestamp, d.getExpectedRTPTimestamp, ) - d.forwarder.OnParkedLayerExpired(func() { - if sal := d.getStreamAllocatorListener(); sal != nil { - sal.OnSubscriptionChanged(d) - } - }) d.rtpStats = buffer.NewRTPStats(buffer.RTPStatsParams{ ClockRate: d.codec.ClockRate, diff --git a/pkg/sfu/forwarder.go b/pkg/sfu/forwarder.go index ec7db7753..6ab522da1 100644 --- a/pkg/sfu/forwarder.go +++ b/pkg/sfu/forwarder.go @@ -36,10 +36,9 @@ import ( // Forwarder const ( - FlagPauseOnDowngrade = true - FlagFilterRTX = true - TransitionCostSpatial = 10 - ParkedLayerWaitDuration = 2 * time.Second + FlagPauseOnDowngrade = true + FlagFilterRTX = true + TransitionCostSpatial = 10 ResumeBehindThresholdSeconds = float64(0.1) // 100ms LayerSwitchBehindThresholdSeconds = float64(0.05) // 50ms @@ -124,7 +123,6 @@ type VideoAllocationProvisional struct { Bitrates Bitrates maxLayer buffer.VideoLayer currentLayer buffer.VideoLayer - parkedLayer buffer.VideoLayer allocatedLayer buffer.VideoLayer } @@ -199,8 +197,6 @@ type Forwarder struct { referenceLayerSpatial int32 refTSOffset uint32 - parkedLayerTimer *time.Timer - provisional *VideoAllocationProvisional lastAllocation VideoAllocation @@ -210,8 +206,6 @@ type Forwarder struct { vls videolayerselector.VideoLayerSelector codecMunger codecmunger.CodecMunger - - onParkedLayerExpired func() } func NewForwarder( @@ -266,20 +260,6 @@ func (f *Forwarder) SetMaxTemporalLayerSeen(maxTemporalLayerSeen int32) bool { return true } -func (f *Forwarder) OnParkedLayerExpired(fn func()) { - f.lock.Lock() - defer f.lock.Unlock() - - f.onParkedLayerExpired = fn -} - -func (f *Forwarder) getOnParkedLayerExpired() func() { - f.lock.RLock() - defer f.lock.RUnlock() - - return f.onParkedLayerExpired -} - func (f *Forwarder) DetermineCodec(codec webrtc.RTPCodecCapability, extensions []webrtc.RTPHeaderExtensionParameter) { f.lock.Lock() defer f.lock.Unlock() @@ -428,22 +408,10 @@ func (f *Forwarder) PubMute(pubMuted bool) bool { f.logger.Debugw("setting forwarder pub mute", "pubMuted", pubMuted) f.pubMuted = pubMuted - if f.kind == webrtc.RTPCodecTypeAudio { - // for audio resync when pub muted so that sequence numbers do not jump on unmute - // audio stops forwarding during pub mute too - if pubMuted { - f.resyncLocked() - } - } else { - // Do not resync on publisher mute as forwarding can continue on unmute using same layer. - // On unmute, park current layers as streaming can continue without a key frame when publisher starts the stream. - targetLayer := f.vls.GetTarget() - if !pubMuted && targetLayer.IsValid() && f.vls.GetCurrent().Spatial == targetLayer.Spatial { - f.setupParkedLayer(targetLayer) - f.vls.SetCurrent(buffer.InvalidLayer) - } + // resync when pub muted so that sequence numbers do not jump on unmute + if pubMuted { + f.resyncLocked() } - return true } @@ -476,9 +444,6 @@ func (f *Forwarder) SetMaxSpatialLayer(spatialLayer int32) (bool, buffer.VideoLa f.logger.Debugw("setting max spatial layer", "layer", spatialLayer) f.vls.SetMaxSpatial(spatialLayer) - - f.clearParkedLayer() - return true, f.vls.GetMax() } @@ -497,9 +462,6 @@ func (f *Forwarder) SetMaxTemporalLayer(temporalLayer int32) (bool, buffer.Video f.logger.Debugw("setting max temporal layer", "layer", temporalLayer) f.vls.SetMaxTemporal(temporalLayer) - - f.clearParkedLayer() - return true, f.vls.GetMax() } @@ -607,7 +569,6 @@ func (f *Forwarder) AllocateOptimal(availableLayers []int32, brs Bitrates, allow maxLayer := f.vls.GetMax() maxSeenLayer := f.vls.GetMaxSeen() - parkedLayer := f.vls.GetParked() currentLayer := f.vls.GetCurrent() requestSpatial := f.vls.GetRequestSpatial() alloc := VideoAllocation{ @@ -653,14 +614,6 @@ func (f *Forwarder) AllocateOptimal(availableLayers []int32, brs Bitrates, allow case f.pubMuted: alloc.PauseReason = VideoPauseReasonPubMuted - // leave it at current layers for opportunistic resume - alloc.TargetLayer = currentLayer - alloc.RequestLayerSpatial = alloc.TargetLayer.Spatial - - case parkedLayer.IsValid(): - // if parked on a layer, let it continue - alloc.TargetLayer = parkedLayer - alloc.RequestLayerSpatial = alloc.TargetLayer.Spatial default: // lots of different events could end up here @@ -754,7 +707,6 @@ func (f *Forwarder) ProvisionalAllocatePrepare(availableLayers []int32, Bitrates Bitrates: Bitrates, maxLayer: f.vls.GetMax(), currentLayer: f.vls.GetCurrent(), - parkedLayer: f.vls.GetParked(), } f.provisional.availableLayers = make([]int32, len(availableLayers)) @@ -837,19 +789,11 @@ func (f *Forwarder) ProvisionalAllocateGetCooperativeTransition(allowOvershoot b existingTargetLayer := f.vls.GetTarget() if f.provisional.muted || f.provisional.pubMuted { - bandwidthRequired := int64(0) f.provisional.allocatedLayer = buffer.InvalidLayer - if f.provisional.pubMuted { - // leave it at current for opportunistic forwarding, there is still bandwidth saving with publisher mute - f.provisional.allocatedLayer = f.provisional.currentLayer - if f.provisional.allocatedLayer.IsValid() { - bandwidthRequired = f.provisional.Bitrates[f.provisional.allocatedLayer.Spatial][f.provisional.allocatedLayer.Temporal] - } - } return VideoTransition{ From: f.vls.GetTarget(), To: f.provisional.allocatedLayer, - BandwidthDelta: bandwidthRequired - getBandwidthNeeded(f.provisional.Bitrates, existingTargetLayer, f.lastAllocation.BandwidthRequested), + BandwidthDelta: -getBandwidthNeeded(f.provisional.Bitrates, existingTargetLayer, f.lastAllocation.BandwidthRequested), } } @@ -941,12 +885,7 @@ func (f *Forwarder) ProvisionalAllocateGetCooperativeTransition(allowOvershoot b // if nothing available, just leave target at current to enable opportunistic forwarding in case current resumes if !targetLayer.IsValid() { - if f.provisional.parkedLayer.IsValid() { - targetLayer = f.provisional.parkedLayer - } else { - targetLayer = f.provisional.currentLayer - } - + targetLayer = f.provisional.currentLayer if targetLayer.IsValid() { bandwidthRequired = f.provisional.Bitrates[targetLayer.Spatial][targetLayer.Temporal] } @@ -981,7 +920,6 @@ func (f *Forwarder) ProvisionalAllocateGetBestWeightedTransition() VideoTransiti targetLayer := f.vls.GetTarget() if f.provisional.muted || f.provisional.pubMuted { - // if publisher muted, give up opportunistic resume and give back the bandwidth f.provisional.allocatedLayer = buffer.InvalidLayer return VideoTransition{ From: targetLayer, @@ -1007,11 +945,7 @@ func (f *Forwarder) ProvisionalAllocateGetBestWeightedTransition() VideoTransiti // feed has gone dry, just leave target at current to enable opportunistic forwarding in case current resumes. // Note that this is giving back bits and opportunistic forwarding resuming might trigger congestion again, // but that should be handled by stream allocator. - if f.provisional.parkedLayer.IsValid() { - f.provisional.allocatedLayer = f.provisional.parkedLayer - } else { - f.provisional.allocatedLayer = f.provisional.currentLayer - } + f.provisional.allocatedLayer = f.provisional.currentLayer return VideoTransition{ From: targetLayer, To: f.provisional.allocatedLayer, @@ -1138,7 +1072,6 @@ func (f *Forwarder) ProvisionalAllocateCommit() VideoAllocation { } } - f.clearParkedLayer() return f.updateAllocation(alloc, "cooperative") } @@ -1382,7 +1315,6 @@ func (f *Forwarder) Pause(availableLayers []int32, brs Bitrates) VideoAllocation alloc.PauseReason = VideoPauseReasonBandwidth } - f.clearParkedLayer() return f.updateAllocation(alloc, "pause") } @@ -1427,31 +1359,6 @@ func (f *Forwarder) Resync() { func (f *Forwarder) resyncLocked() { f.vls.SetCurrent(buffer.InvalidLayer) f.lastSSRC = 0 - f.clearParkedLayer() -} - -func (f *Forwarder) clearParkedLayer() { - f.vls.SetParked(buffer.InvalidLayer) - if f.parkedLayerTimer != nil { - f.parkedLayerTimer.Stop() - f.parkedLayerTimer = nil - } -} - -func (f *Forwarder) setupParkedLayer(parkedLayer buffer.VideoLayer) { - f.clearParkedLayer() - - f.vls.SetParked(parkedLayer) - f.parkedLayerTimer = time.AfterFunc(ParkedLayerWaitDuration, func() { - f.lock.Lock() - notify := f.vls.GetParked().IsValid() - f.clearParkedLayer() - f.lock.Unlock() - - if onParkedLayerExpired := f.getOnParkedLayerExpired(); onParkedLayerExpired != nil && notify { - onParkedLayerExpired() - } - }) } func (f *Forwarder) CheckSync() (locked bool, layer int32) { @@ -1495,8 +1402,7 @@ func (f *Forwarder) GetTranslationParams(extPkt *buffer.ExtPacket, layer int32) f.lock.Lock() defer f.lock.Unlock() - // Video: Do not drop on publisher mute to enable resume on publisher unmute without a key frame. - if f.muted { + if f.muted || f.pubMuted { return &TranslationParams{ shouldDrop: true, }, nil @@ -1504,13 +1410,6 @@ func (f *Forwarder) GetTranslationParams(extPkt *buffer.ExtPacket, layer int32) switch f.kind { case webrtc.RTPCodecTypeAudio: - // Audio: Blank frames are injected on publisher mute to ensure decoder does not get stuck at a noise frame. So, do not forward. - if f.pubMuted { - return &TranslationParams{ - shouldDrop: true, - }, nil - } - return f.getTranslationParamsAudio(extPkt, layer) case webrtc.RTPCodecTypeVideo: return f.getTranslationParamsVideo(extPkt, layer) diff --git a/pkg/sfu/forwarder_test.go b/pkg/sfu/forwarder_test.go index 935009792..01b224634 100644 --- a/pkg/sfu/forwarder_test.go +++ b/pkg/sfu/forwarder_test.go @@ -201,34 +201,13 @@ func TestForwarderAllocateOptimal(t *testing.T) { f.PubMute(false) - // when parked layers valid, should stay there - f.vls.SetParked(buffer.VideoLayer{ - Spatial: 0, - Temporal: 1, - }) - expectedResult = VideoAllocation{ - PauseReason: VideoPauseReasonFeedDry, - BandwidthRequested: 0, - BandwidthDelta: 0, - Bitrates: emptyBitrates, - TargetLayer: f.vls.GetParked(), - RequestLayerSpatial: f.vls.GetParked().Spatial, - MaxLayer: buffer.DefaultMaxLayer, - DistanceToDesired: 0, - } - result = f.AllocateOptimal(nil, emptyBitrates, true) - require.Equal(t, expectedResult, result) - require.Equal(t, expectedResult, f.lastAllocation) - require.Equal(t, f.vls.GetParked(), f.TargetLayer()) - f.vls.SetParked(buffer.InvalidLayer) - // when max layers changes, target is opportunistic, but requested spatial layer should be at max f.SetMaxTemporalLayerSeen(3) f.vls.SetMax(buffer.VideoLayer{Spatial: 1, Temporal: 3}) expectedResult = VideoAllocation{ PauseReason: VideoPauseReasonNone, BandwidthRequested: bitrates[1][3], - BandwidthDelta: bitrates[1][3] - bitrates[0][1], + BandwidthDelta: bitrates[1][3], BandwidthNeeded: bitrates[1][3], Bitrates: bitrates, TargetLayer: buffer.DefaultMaxLayer, diff --git a/pkg/sfu/videolayerselector/base.go b/pkg/sfu/videolayerselector/base.go index 6708e7030..346632cb9 100644 --- a/pkg/sfu/videolayerselector/base.go +++ b/pkg/sfu/videolayerselector/base.go @@ -33,9 +33,6 @@ type Base struct { requestSpatial int32 - parkedLayer buffer.VideoLayer - previousParkedLayer buffer.VideoLayer - currentLayer buffer.VideoLayer previousLayer buffer.VideoLayer } @@ -48,8 +45,6 @@ func NewBase(logger logger.Logger) *Base { targetLayer: buffer.InvalidLayer, // start off with nothing, let streamallocator/opportunistic forwarder set the target previousTargetLayer: buffer.InvalidLayer, requestSpatial: buffer.InvalidLayerSpatial, - parkedLayer: buffer.InvalidLayer, - previousParkedLayer: buffer.InvalidLayer, currentLayer: buffer.InvalidLayer, previousLayer: buffer.InvalidLayer, } @@ -98,7 +93,7 @@ func (b *Base) GetRequestSpatial() int32 { func (b *Base) CheckSync() (locked bool, layer int32) { layer = b.GetRequestSpatial() - locked = layer == b.GetCurrent().Spatial || b.GetParked().IsValid() + locked = layer == b.GetCurrent().Spatial return } @@ -118,14 +113,6 @@ func (b *Base) GetMaxSeen() buffer.VideoLayer { return b.maxSeenLayer } -func (b *Base) SetParked(parkedLayer buffer.VideoLayer) { - b.parkedLayer = parkedLayer -} - -func (b *Base) GetParked() buffer.VideoLayer { - return b.parkedLayer -} - func (b *Base) SetCurrent(currentLayer buffer.VideoLayer) { b.currentLayer = currentLayer } @@ -143,15 +130,12 @@ func (b *Base) Rollback() { "rolling back", "previous", b.previousLayer, "current", b.currentLayer, - "previousParked", b.previousParkedLayer, - "parked", b.parkedLayer, "previousTarget", b.previousTargetLayer, "target", b.targetLayer, "max", b.maxLayer, "req", b.requestSpatial, "maxSeen", b.maxSeenLayer, ) - b.parkedLayer = b.previousParkedLayer b.currentLayer = b.previousLayer b.targetLayer = b.previousTargetLayer } @@ -170,8 +154,6 @@ func (b *Base) SelectTemporal(extPkt *buffer.ExtPacket) (int32, bool) { "updating temporal layer", "previous", b.previousLayer, "current", b.currentLayer, - "previousParked", b.previousParkedLayer, - "parked", b.parkedLayer, "previousTarget", b.previousTargetLayer, "target", b.targetLayer, "max", b.maxLayer, diff --git a/pkg/sfu/videolayerselector/dependencydescriptor.go b/pkg/sfu/videolayerselector/dependencydescriptor.go index 555437c6d..9d29427be 100644 --- a/pkg/sfu/videolayerselector/dependencydescriptor.go +++ b/pkg/sfu/videolayerselector/dependencydescriptor.go @@ -286,10 +286,6 @@ func (d *DependencyDescriptor) updateActiveDecodeTargets(activeDecodeTargetsBitm func (d *DependencyDescriptor) CheckSync() (locked bool, layer int32) { layer = d.GetRequestSpatial() - if d.GetParked().IsValid() { - return true, layer - } - d.decodeTargetsLock.RLock() defer d.decodeTargetsLock.RUnlock() for _, dt := range d.decodeTargets { diff --git a/pkg/sfu/videolayerselector/simulcast.go b/pkg/sfu/videolayerselector/simulcast.go index b4a5c527e..6b20787b7 100644 --- a/pkg/sfu/videolayerselector/simulcast.go +++ b/pkg/sfu/videolayerselector/simulcast.go @@ -40,10 +40,8 @@ func (s *Simulcast) IsOvershootOkay() bool { } func (s *Simulcast) Select(extPkt *buffer.ExtPacket, layer int32) (result VideoLayerSelectorResult) { - populateSwitches := func(isSwitching bool, isActive bool, reason string) { - if isSwitching { - result.IsSwitching = true - } + populateSwitches := func(isActive bool, reason string) { + result.IsSwitching = true if !isActive { result.IsResuming = true @@ -54,8 +52,6 @@ func (s *Simulcast) Select(extPkt *buffer.ExtPacket, layer int32) (result VideoL reason, "previous", s.previousLayer, "current", s.currentLayer, - "previousParked", s.previousParkedLayer, - "parked", s.parkedLayer, "previousTarget", s.previousTargetLayer, "target", s.targetLayer, "max", s.maxLayer, @@ -70,44 +66,30 @@ func (s *Simulcast) Select(extPkt *buffer.ExtPacket, layer int32) (result VideoL if s.currentLayer.Spatial != s.targetLayer.Spatial { currentLayer := s.currentLayer - // Three things to check when not locked to target - // 1. Resumable layer - don't need a key frame - // 2. Opportunistic layer upgrade - needs a key frame - // 3. Need to downgrade - needs a key frame - isSwitching := true + // Two things to check when not locked to target + // 1. Opportunistic layer upgrade - needs a key frame + // 2. Need to downgrade - needs a key frame isActive := s.currentLayer.IsValid() found := false reason := "" - if s.parkedLayer.IsValid() { - if s.parkedLayer.Spatial == layer { - reason = "resuming at parked layer" - currentLayer = s.parkedLayer - isSwitching = false + if extPkt.KeyFrame { + if layer > s.currentLayer.Spatial && layer <= s.targetLayer.Spatial { + reason = "upgrading layer" found = true } - } else { - if extPkt.KeyFrame { - if layer > s.currentLayer.Spatial && layer <= s.targetLayer.Spatial { - reason = "upgrading layer" - found = true - } - if layer < s.currentLayer.Spatial && layer >= s.targetLayer.Spatial { - reason = "downgrading layer" - found = true - } + if layer < s.currentLayer.Spatial && layer >= s.targetLayer.Spatial { + reason = "downgrading layer" + found = true + } - if found { - currentLayer.Spatial = layer - currentLayer.Temporal = extPkt.VideoLayer.Temporal - } + if found { + currentLayer.Spatial = layer + currentLayer.Temporal = extPkt.VideoLayer.Temporal } } if found { - s.previousParkedLayer = s.parkedLayer - s.parkedLayer = buffer.InvalidLayer - s.previousLayer = s.currentLayer s.currentLayer = currentLayer @@ -116,7 +98,7 @@ func (s *Simulcast) Select(extPkt *buffer.ExtPacket, layer int32) (result VideoL s.targetLayer.Spatial = s.currentLayer.Spatial } - populateSwitches(isSwitching, isActive, reason) + populateSwitches(isActive, reason) } } @@ -130,7 +112,7 @@ func (s *Simulcast) Select(extPkt *buffer.ExtPacket, layer int32) (result VideoL s.targetLayer.Spatial = layer } - populateSwitches(true, true, "adjusting overshoot") + populateSwitches(true, "adjusting overshoot") } result.RTPMarker = extPkt.Packet.Marker diff --git a/pkg/sfu/videolayerselector/videolayerselector.go b/pkg/sfu/videolayerselector/videolayerselector.go index 545196eae..186c8f64d 100644 --- a/pkg/sfu/videolayerselector/videolayerselector.go +++ b/pkg/sfu/videolayerselector/videolayerselector.go @@ -51,9 +51,6 @@ type VideoLayerSelector interface { SetMaxSeenTemporal(layer int32) GetMaxSeen() buffer.VideoLayer - SetParked(parkedLayer buffer.VideoLayer) - GetParked() buffer.VideoLayer - SetCurrent(currentLayer buffer.VideoLayer) GetCurrent() buffer.VideoLayer