From 9cf190bdf7c60bce1392d1be24a362016124e68c Mon Sep 17 00:00:00 2001 From: David Colburn Date: Wed, 12 Jul 2023 18:46:48 -0700 Subject: [PATCH 01/18] update protocol, psrpc (#1872) --- go.mod | 6 +++--- go.sum | 12 ++++++------ 2 files changed, 9 insertions(+), 9 deletions(-) diff --git a/go.mod b/go.mod index 57516561d..09dbd2d90 100644 --- a/go.mod +++ b/go.mod @@ -18,8 +18,8 @@ require ( github.com/jxskiss/base62 v1.1.0 github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1 github.com/livekit/mediatransportutil v0.0.0-20230612070454-d5299b956135 - github.com/livekit/protocol v1.5.9 - github.com/livekit/psrpc v0.3.1 + github.com/livekit/protocol v1.5.10-0.20230713013703-71d539707e4c + github.com/livekit/psrpc v0.3.2 github.com/mackerelio/go-osstat v0.2.4 github.com/magefile/mage v1.15.0 github.com/maxbrunsfeld/counterfeiter/v6 v6.6.2 @@ -101,6 +101,6 @@ require ( golang.org/x/text v0.10.0 // indirect golang.org/x/tools v0.9.3 // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20230530153820-e85fd2cbaebc // indirect - google.golang.org/grpc v1.56.1 // indirect + google.golang.org/grpc v1.56.2 // indirect gopkg.in/yaml.v2 v2.4.0 // indirect ) diff --git a/go.sum b/go.sum index e22cfb93e..496e76719 100644 --- a/go.sum +++ b/go.sum @@ -124,10 +124,10 @@ github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1 h1:jm09419p0lqTkD github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ= github.com/livekit/mediatransportutil v0.0.0-20230612070454-d5299b956135 h1:lWYbsondvqG69czxoACDwaJ/BoyD57BahCo70ZH+m4U= github.com/livekit/mediatransportutil v0.0.0-20230612070454-d5299b956135/go.mod h1:MRc0zSOSzXuFt0X218SgabzlaKevkvCckPgBEoHYc34= -github.com/livekit/protocol v1.5.9 h1:fqPOLgKkWmkmUMnpfj2KDZlidHgAazAPU2T3FewyLew= -github.com/livekit/protocol v1.5.9/go.mod h1:GMTlFbc0JypUGo+PMilDrL45AX+yUBiQ1Cl/FGZibDY= -github.com/livekit/psrpc v0.3.1 h1:KfylgJHvoLQcc22t/oflwMOeSnx0c14G7cWsS+9MYS4= -github.com/livekit/psrpc v0.3.1/go.mod h1:n6JntEg+zT6Ji8InoyTpV7wusPNwGqqtxmHlkNhDN0U= +github.com/livekit/protocol v1.5.10-0.20230713013703-71d539707e4c h1:o2xIq3un7M7un8wU7jYw+6mV1uSp7oWA2lCHRm5m3+Y= +github.com/livekit/protocol v1.5.10-0.20230713013703-71d539707e4c/go.mod h1:eRzojAYSPJuNgDHMlvLji/CPauj9hrgvb6rVPUj6MoU= +github.com/livekit/psrpc v0.3.2 h1:eAaJhASme33gtoBhCRLH9jsnWcdm1tHWf0WzaDk56ew= +github.com/livekit/psrpc v0.3.2/go.mod h1:n6JntEg+zT6Ji8InoyTpV7wusPNwGqqtxmHlkNhDN0U= github.com/mackerelio/go-osstat v0.2.4 h1:qxGbdPkFo65PXOb/F/nhDKpF2nGmGaCFDLXoZjJTtUs= github.com/mackerelio/go-osstat v0.2.4/go.mod h1:Zy+qzGdZs3A9cuIqmgbJvwbmLQH9dJvtio5ZjJTbdlQ= github.com/magefile/mage v1.15.0 h1:BvGheCMAsG3bWUDbZ8AyXXpCNwU9u5CB6sM+HNb9HYg= @@ -411,8 +411,8 @@ golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8T golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= google.golang.org/genproto/googleapis/rpc v0.0.0-20230530153820-e85fd2cbaebc h1:XSJ8Vk1SWuNr8S18z1NZSziL0CPIXLCCMDOEFtHBOFc= google.golang.org/genproto/googleapis/rpc v0.0.0-20230530153820-e85fd2cbaebc/go.mod h1:66JfowdXAEgad5O9NnYcsNPLCPZJD++2L9X0PCMODrA= -google.golang.org/grpc v1.56.1 h1:z0dNfjIl0VpaZ9iSVjA6daGatAYwPGstTjt5vkRMFkQ= -google.golang.org/grpc v1.56.1/go.mod h1:I9bI3vqKfayGqPUAwGdOSu7kt6oIJLixfffKrpXqQ9s= +google.golang.org/grpc v1.56.2 h1:fVRFRnXvU+x6C4IlHZewvJOVHoOv1TUuQyoRsYnB4bI= +google.golang.org/grpc v1.56.2/go.mod h1:I9bI3vqKfayGqPUAwGdOSu7kt6oIJLixfffKrpXqQ9s= google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8= google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0= google.golang.org/protobuf v0.0.0-20200228230310-ab0ca4ff8a60/go.mod h1:cfTl7dwQJ+fmap5saPgwCLgHXTUD7jkjRqWcaiX5VyM= From e746fe14e1267052b43386dd9949c9fa722ef675 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Thu, 13 Jul 2023 10:42:23 +0530 Subject: [PATCH 02/18] Mark active when switching to parked layer. (#1873) * Mark active when switching to parked layer. Parked layer lock is not a switch. It is just a restart at the same layer. * make explicit bool for switching --- pkg/sfu/videolayerselector/simulcast.go | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/pkg/sfu/videolayerselector/simulcast.go b/pkg/sfu/videolayerselector/simulcast.go index ebb80d113..4d0c9294d 100644 --- a/pkg/sfu/videolayerselector/simulcast.go +++ b/pkg/sfu/videolayerselector/simulcast.go @@ -26,8 +26,11 @@ func (s *Simulcast) IsOvershootOkay() bool { } func (s *Simulcast) Select(extPkt *buffer.ExtPacket, layer int32) (result VideoLayerSelectorResult) { - populateSwitches := func(isActive bool, reason string) { - result.IsSwitching = true + populateSwitches := func(isSwitching bool, isActive bool, reason string) { + if isSwitching { + result.IsSwitching = true + } + if !isActive { result.IsResuming = true } @@ -70,6 +73,7 @@ func (s *Simulcast) Select(extPkt *buffer.ExtPacket, layer int32) (result VideoL // 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 isActive := s.currentLayer.IsValid() found := false reason := "" @@ -77,6 +81,7 @@ func (s *Simulcast) Select(extPkt *buffer.ExtPacket, layer int32) (result VideoL if s.parkedLayer.Spatial == layer { reason = "resuming at parked layer" currentLayer = s.parkedLayer + isSwitching = false found = true } } else { @@ -110,7 +115,7 @@ func (s *Simulcast) Select(extPkt *buffer.ExtPacket, layer int32) (result VideoL s.targetLayer.Spatial = s.currentLayer.Spatial } - populateSwitches(isActive, reason) + populateSwitches(isSwitching, isActive, reason) } } @@ -124,7 +129,7 @@ func (s *Simulcast) Select(extPkt *buffer.ExtPacket, layer int32) (result VideoL s.targetLayer.Spatial = layer } - populateSwitches(true, "adjusting overshoot") + populateSwitches(true, true, "adjusting overshoot") } result.RTPMarker = extPkt.Packet.Marker From 557fe7c9d3e722fe58eaacfe7d1f9e7ea4fb5d11 Mon Sep 17 00:00:00 2001 From: David Zhao Date: Thu, 13 Jul 2023 16:33:04 -0700 Subject: [PATCH 03/18] Mark room as dirty after track published changes (#1878) Ensure that we are recomputing NumPublished when needed --- pkg/rtc/room.go | 2 ++ 1 file changed, 2 insertions(+) diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index dcd5a6b1c..a90b90676 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -828,6 +828,7 @@ func (r *Room) createJoinResponseLocked(participant types.LocalParticipant, iceS func (r *Room) onTrackPublished(participant types.LocalParticipant, track types.MediaTrack) { // publish participant update, since track state is changed r.broadcastParticipantState(participant, broadcastOptions{skipSource: true}) + r.protoProxy.MarkDirty(false) r.lock.RLock() // subscribe all existing participants to this MediaTrack @@ -887,6 +888,7 @@ func (r *Room) onTrackUpdated(p types.LocalParticipant, _ types.MediaTrack) { func (r *Room) onTrackUnpublished(p types.LocalParticipant, track types.MediaTrack) { r.trackManager.RemoveTrack(track) + r.protoProxy.MarkDirty(false) if !p.IsClosed() { r.broadcastParticipantState(p, broadcastOptions{skipSource: true}) } From 68e5fa8e1c85975d98b4c91b00c077a10eae3557 Mon Sep 17 00:00:00 2001 From: Benjamin Pracht Date: Fri, 14 Jul 2023 09:11:55 +0800 Subject: [PATCH 04/18] Allow listing ingress by id (#1874) --- go.mod | 2 +- go.sum | 4 ++-- pkg/service/ingress.go | 17 +++++++++++++---- 3 files changed, 16 insertions(+), 7 deletions(-) diff --git a/go.mod b/go.mod index 09dbd2d90..894e44d78 100644 --- a/go.mod +++ b/go.mod @@ -18,7 +18,7 @@ require ( github.com/jxskiss/base62 v1.1.0 github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1 github.com/livekit/mediatransportutil v0.0.0-20230612070454-d5299b956135 - github.com/livekit/protocol v1.5.10-0.20230713013703-71d539707e4c + github.com/livekit/protocol v1.5.10-0.20230714010226-3c53edc91962 github.com/livekit/psrpc v0.3.2 github.com/mackerelio/go-osstat v0.2.4 github.com/magefile/mage v1.15.0 diff --git a/go.sum b/go.sum index 496e76719..dba27c2ea 100644 --- a/go.sum +++ b/go.sum @@ -124,8 +124,8 @@ github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1 h1:jm09419p0lqTkD github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ= github.com/livekit/mediatransportutil v0.0.0-20230612070454-d5299b956135 h1:lWYbsondvqG69czxoACDwaJ/BoyD57BahCo70ZH+m4U= github.com/livekit/mediatransportutil v0.0.0-20230612070454-d5299b956135/go.mod h1:MRc0zSOSzXuFt0X218SgabzlaKevkvCckPgBEoHYc34= -github.com/livekit/protocol v1.5.10-0.20230713013703-71d539707e4c h1:o2xIq3un7M7un8wU7jYw+6mV1uSp7oWA2lCHRm5m3+Y= -github.com/livekit/protocol v1.5.10-0.20230713013703-71d539707e4c/go.mod h1:eRzojAYSPJuNgDHMlvLji/CPauj9hrgvb6rVPUj6MoU= +github.com/livekit/protocol v1.5.10-0.20230714010226-3c53edc91962 h1:y+rtYNMGmvpEgQlNG/wOUO16S497ygh83wQSUBKHpG4= +github.com/livekit/protocol v1.5.10-0.20230714010226-3c53edc91962/go.mod h1:eRzojAYSPJuNgDHMlvLji/CPauj9hrgvb6rVPUj6MoU= github.com/livekit/psrpc v0.3.2 h1:eAaJhASme33gtoBhCRLH9jsnWcdm1tHWf0WzaDk56ew= github.com/livekit/psrpc v0.3.2/go.mod h1:n6JntEg+zT6Ji8InoyTpV7wusPNwGqqtxmHlkNhDN0U= github.com/mackerelio/go-osstat v0.2.4 h1:qxGbdPkFo65PXOb/F/nhDKpF2nGmGaCFDLXoZjJTtUs= diff --git a/pkg/service/ingress.go b/pkg/service/ingress.go index 04d4a6bd8..5fa7990ee 100644 --- a/pkg/service/ingress.go +++ b/pkg/service/ingress.go @@ -216,10 +216,19 @@ func (s *IngressService) ListIngress(ctx context.Context, req *livekit.ListIngre return nil, ErrIngressNotConnected } - infos, err := s.store.ListIngress(ctx, livekit.RoomName(req.RoomName)) - if err != nil { - logger.Errorw("could not list ingress info", err) - return nil, err + var infos []*livekit.IngressInfo + if req.IngressId != "" { + info, err := s.store.LoadIngress(ctx, req.IngressId) + if err != nil { + return nil, err + } + infos = []*livekit.IngressInfo{info} + } else { + infos, err = s.store.ListIngress(ctx, livekit.RoomName(req.RoomName)) + if err != nil { + logger.Errorw("could not list ingress info", err) + return nil, err + } } return &livekit.ListIngressResponse{Items: infos}, nil From 4c02a6d71720062a0c61bdb88a6106cc62dbe392 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Fri, 14 Jul 2023 11:47:07 +0530 Subject: [PATCH 05/18] Time stamp adjustments v2 (I think) (#1875) * WIP commit * WIP commit * WIP commit * Some clean up - Removed a chatty debug log - some spelling, punctuation correction in comments - missed an `Abs` in check, add it. --- pkg/rtc/wrappedreceiver.go | 7 +- pkg/sfu/buffer/rtpstats.go | 93 ++++----- pkg/sfu/downtrack.go | 5 +- pkg/sfu/forwarder.go | 321 +++++++++++++++++--------------- pkg/sfu/receiver.go | 6 +- pkg/sfu/streamtrackermanager.go | 141 +++++++++----- 6 files changed, 320 insertions(+), 253 deletions(-) diff --git a/pkg/rtc/wrappedreceiver.go b/pkg/rtc/wrappedreceiver.go index 545ac1b78..7028084fb 100644 --- a/pkg/rtc/wrappedreceiver.go +++ b/pkg/rtc/wrappedreceiver.go @@ -12,7 +12,6 @@ import ( "github.com/livekit/protocol/logger" "github.com/livekit/livekit-server/pkg/sfu" - "github.com/livekit/livekit-server/pkg/sfu/buffer" ) // wrapper around WebRTC receiver, overriding its ID @@ -297,11 +296,11 @@ func (d *DummyReceiver) GetRedReceiver() sfu.TrackReceiver { return d } -func (d *DummyReceiver) GetRTCPSenderReportData(layer int32) (*buffer.RTCPSenderReportData, *buffer.RTCPSenderReportData) { +func (d *DummyReceiver) GetCalculatedClockRate(layer int32) uint32 { if r, ok := d.receiver.Load().(sfu.TrackReceiver); ok { - return r.GetRTCPSenderReportData(layer) + return r.GetCalculatedClockRate(layer) } - return nil, nil + return 0 } func (d *DummyReceiver) GetReferenceLayerRTPTimestamp(ts uint32, layer int32, referenceLayer int32) (uint32, error) { diff --git a/pkg/sfu/buffer/rtpstats.go b/pkg/sfu/buffer/rtpstats.go index c21938d36..8bf34b958 100644 --- a/pkg/sfu/buffer/rtpstats.go +++ b/pkg/sfu/buffer/rtpstats.go @@ -750,13 +750,13 @@ func (r *RTPStats) GetRtt() uint32 { } func (r *RTPStats) SetRtcpSenderReportData(srData *RTCPSenderReportData) { - if srData == nil { - return - } - r.lock.Lock() defer r.lock.Unlock() + if srData == nil || !r.initialized { + return + } + // prevent against extreme case of anachronous sender reports if r.srNewest != nil && r.srNewest.NTPTimestamp > srData.NTPTimestamp { r.logger.Infow( @@ -767,29 +767,6 @@ func (r *RTPStats) SetRtcpSenderReportData(srData *RTCPSenderReportData) { return } - // monitor and log RTP timestamp anomalies - var ntpDiffSinceLast time.Duration - var rtpDiffSinceLast uint32 - var arrivalDiffSinceLast time.Duration - var expectedTimeDiffSinceLast float64 - var reason string - if r.srNewest != nil { - ntpDiffSinceLast = srData.NTPTimestamp.Time().Sub(r.srNewest.NTPTimestamp.Time()) - rtpDiffSinceLast = srData.RTPTimestamp - r.srNewest.RTPTimestamp - arrivalDiffSinceLast = srData.At.Sub(r.srNewest.At) - - expectedTimeDiffSinceLast = float64(rtpDiffSinceLast) / float64(r.params.ClockRate) - - if (srData.RTPTimestamp - r.srNewest.RTPTimestamp) > (1 << 31) { - reason = "received sender report, out-of-order" // should not happen, just a sanity check - } else { - if math.Abs(expectedTimeDiffSinceLast-ntpDiffSinceLast.Seconds()) > 0.2 { - // more than 200 ms away from expected delta - reason = "received sender report, time warp" - } - } - } - cycles := uint64(0) if r.srNewest != nil { cycles = r.srNewest.RTPTimestampExt & 0xFF_FF_FF_FF_00_00_00_00 @@ -800,15 +777,47 @@ func (r *RTPStats) SetRtcpSenderReportData(srData *RTCPSenderReportData) { srDataCopy := *srData srDataCopy.RTPTimestampExt = uint64(srDataCopy.RTPTimestamp) + cycles + + // monitor and log RTP timestamp anomalies + var ntpDiffSinceLast time.Duration + var rtpDiffSinceLast uint32 + var arrivalDiffSinceLast time.Duration + var expectedTimeDiffSinceLast float64 + var isWarped bool + if r.srNewest != nil { + if srDataCopy.RTPTimestampExt < r.srNewest.RTPTimestampExt { + // This can happen when a track is replaced with a null and then restored - + // i. e. muting replacing with null and unmute restoring the original track. + // Under such a condition reset the sender reports to start from this point. + // Resetting will ensure sample rate calculations do not go haywire due to negative time. + r.logger.Infow( + "received sender report, out-of-order, resetting", + "prevTSExt", r.srNewest.RTPTimestampExt, + "currTSExt", srDataCopy.RTPTimestampExt, + ) + r.srFirst = &srDataCopy + r.srNewest = &srDataCopy + } + + ntpDiffSinceLast = srDataCopy.NTPTimestamp.Time().Sub(r.srNewest.NTPTimestamp.Time()) + rtpDiffSinceLast = srDataCopy.RTPTimestamp - r.srNewest.RTPTimestamp + arrivalDiffSinceLast = srDataCopy.At.Sub(r.srNewest.At) + expectedTimeDiffSinceLast = float64(rtpDiffSinceLast) / float64(r.params.ClockRate) + if math.Abs(expectedTimeDiffSinceLast-ntpDiffSinceLast.Seconds()) > 0.2 { + // more than 200 ms away from expected delta + isWarped = true + } + } + r.srNewest = &srDataCopy if r.srFirst == nil { r.srFirst = &srDataCopy } - if reason != "" { + if isWarped { packetDriftResult, reportDriftResult := r.getDrift() r.logger.Infow( - reason, + "received sender report, time warp", "ntp", srData.NTPTimestamp.Time().String(), "rtp", srData.RTPTimestamp, "arrival", srData.At.String(), @@ -840,26 +849,22 @@ func (r *RTPStats) GetRtcpSenderReportData() (srFirst *RTCPSenderReportData, srN return } -func (r *RTPStats) GetExpectedRTPTimestamp(at time.Time) (uint32, uint64, error) { +func (r *RTPStats) GetExpectedRTPTimestamp(at time.Time) (expectedTSExt uint64, err error) { r.lock.RLock() defer r.lock.RUnlock() if !r.initialized { - return 0, 0, errors.New("uninitilaized") + err = errors.New("uninitilaized") + return } timeDiff := at.Sub(r.firstTime) expectedRTPDiff := timeDiff.Nanoseconds() * int64(r.params.ClockRate) / 1e9 - expectedExtRTP := r.extStartTS + uint64(expectedRTPDiff) - - minTS := ^uint64(0) - if r.srNewest != nil { - minTS = r.srNewest.RTPTimestampExt - } - return uint32(expectedExtRTP), minTS, nil + expectedTSExt = r.extStartTS + uint64(expectedRTPDiff) + return } -func (r *RTPStats) GetRtcpSenderReport(ssrc uint32, srFirst *RTCPSenderReportData, srNewest *RTCPSenderReportData) *rtcp.SenderReport { +func (r *RTPStats) GetRtcpSenderReport(ssrc uint32, calculatedClockRate uint32) *rtcp.SenderReport { r.lock.Lock() defer r.lock.Unlock() @@ -877,18 +882,14 @@ func (r *RTPStats) GetRtcpSenderReport(ssrc uint32, srFirst *RTCPSenderReportDat // It is possible that publisher is pacing at a slower rate. // That would make `highestTS` to be lagging the RTP time stamp in the RTCP Sender Report from publisher. - // Check for that and use the later time stamp if applicable. + // Check for that using calculated clock rate and use the later time stamp if applicable. tsCycles := r.tsCycles if nowRTP < r.highestTS { tsCycles++ } nowRTPExt := getExtTS(nowRTP, tsCycles) - if srFirst != nil && srNewest != nil && srFirst.RTPTimestamp != srNewest.RTPTimestamp { - // use incoming rate as a guide - tsf := srNewest.NTPTimestamp.Time().Sub(srFirst.NTPTimestamp.Time()) - rdsf := srNewest.RTPTimestampExt - srFirst.RTPTimestampExt - sr := float64(rdsf) / tsf.Seconds() - nowRTPExtUsingRate := r.extStartTS + uint64(sr*timeSinceFirst.Seconds()) + if calculatedClockRate != 0 { + nowRTPExtUsingRate := r.extStartTS + uint64(float64(calculatedClockRate)*timeSinceFirst.Seconds()) if nowRTPExtUsingRate > nowRTPExt { nowRTPExt = nowRTPExtUsingRate nowRTP = uint32(nowRTPExtUsingRate) diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index f0463a1e0..c6dbb34fd 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -1102,8 +1102,7 @@ func (d *DownTrack) CreateSenderReport() *rtcp.SenderReport { return nil } - srFirst, srNewest := d.receiver.GetRTCPSenderReportData(d.forwarder.GetReferenceLayerSpatial()) - return d.rtpStats.GetRtcpSenderReport(d.ssrc, srFirst, srNewest) + return d.rtpStats.GetRtcpSenderReport(d.ssrc, d.receiver.GetCalculatedClockRate(d.forwarder.CurrentLayer().Spatial)) } func (d *DownTrack) writeBlankFrameRTP(duration float32, generation uint32) chan struct{} { @@ -1542,7 +1541,7 @@ func (d *DownTrack) DebugInfo() map[string]interface{} { } } -func (d *DownTrack) getExpectedRTPTimestamp(at time.Time) (uint32, uint64, error) { +func (d *DownTrack) getExpectedRTPTimestamp(at time.Time) (uint64, error) { return d.rtpStats.GetExpectedRTPTimestamp(at) } diff --git a/pkg/sfu/forwarder.go b/pkg/sfu/forwarder.go index 8bb21c387..1da413ee8 100644 --- a/pkg/sfu/forwarder.go +++ b/pkg/sfu/forwarder.go @@ -26,6 +26,10 @@ const ( FlagFilterRTX = true TransitionCostSpatial = 10 ParkedLayerWaitDuration = 2 * time.Second + + ResumeBehindThresholdSeconds = float64(0.1) // 100ms + LayerSwitchBehindThresholdSeconds = float64(0.05) // 50ms + SwitchAheadThresholdSeconds = float64(0.025) // 25ms ) // ------------------------------------------------------------------- @@ -171,7 +175,7 @@ type Forwarder struct { kind webrtc.RTPCodecType logger logger.Logger getReferenceLayerRTPTimestamp func(ts uint32, layer int32, referenceLayer int32) (uint32, error) - getExpectedRTPTimestamp func(at time.Time) (uint32, uint64, error) + getExpectedRTPTimestamp func(at time.Time) (uint64, error) muted bool pubMuted bool @@ -202,7 +206,7 @@ func NewForwarder( kind webrtc.RTPCodecType, logger logger.Logger, getReferenceLayerRTPTimestamp func(ts uint32, layer int32, referenceLayer int32) (uint32, error), - getExpectedRTPTimestamp func(at time.Time) (uint32, uint64, error), + getExpectedRTPTimestamp func(at time.Time) (uint64, error), ) *Forwarder { f := &Forwarder{ kind: kind, @@ -508,13 +512,6 @@ func (f *Forwarder) TargetLayer() buffer.VideoLayer { return f.vls.GetTarget() } -func (f *Forwarder) GetReferenceLayerSpatial() int32 { - f.lock.RLock() - defer f.lock.RUnlock() - - return f.referenceLayerSpatial -} - func (f *Forwarder) isDeficientLocked() bool { return f.lastAllocation.IsDeficient } @@ -1477,103 +1474,176 @@ func (f *Forwarder) GetTranslationParams(extPkt *buffer.ExtPacket, layer int32) return nil, ErrUnknownKind } +func (f *Forwarder) processSourceSwitch(extPkt *buffer.ExtPacket, layer int32) error { + if !f.started { + f.started = true + f.referenceLayerSpatial = layer + f.rtpMunger.SetLastSnTs(extPkt) + f.codecMunger.SetLast(extPkt) + f.logger.Infow( + "starting forwarding", + "sequenceNumber", extPkt.Packet.SequenceNumber, + "timestamp", extPkt.Packet.Timestamp, + "layer", layer, + "referenceLayerSpatial", f.referenceLayerSpatial, + ) + return nil + } + + if f.referenceLayerSpatial == buffer.InvalidLayerSpatial { + // on a resume, reference layer may not be set, so only set when it is invalid + f.referenceLayerSpatial = layer + } + + // Compute how much time passed between the previous forwarded packet + // and the current incoming (to be forwarded) packet and calculate + // timestamp offset on source change. + // + // There are three timestamps to consider here + // 1. lastTS -> timestamp of last sent packet + // 2. refTS -> timestamp of this packet (after munging) calculated using feed's RTCP sender report + // 3. expectedTS -> expected timestamp of this packet calculated based on elapsed time since first packet + // Ideally, refTS and expectedTS should be very close and lastTS should be before both of those. + // But, cases like muting/unmuting, clock vagaries, pacing, etc. make them not satisfy those conditions always. + lastTS := f.rtpMunger.GetLast().LastTS + refTS := lastTS + expectedTS := lastTS + switchingAt := time.Now() + if f.getReferenceLayerRTPTimestamp != nil { + ts, err := f.getReferenceLayerRTPTimestamp(extPkt.Packet.Timestamp, layer, f.referenceLayerSpatial) + if err == nil { + refTS = ts + } + // AVSYNC-TODO: can error out here if refTS is not available. It can happen when there is no sender report + // for the layer being switched to. Can especially happen at the start of the track when layer switches are + // potentially happening very quickly. Erroring out and waiting for a layer for which a sender report has been + // received will calculate a better offset, but may result in initial adaptation to take a bit longer depending + // on how often publisher/remote side sends RTCP sender report. + } + + if f.getExpectedRTPTimestamp != nil { + tsExt, err := f.getExpectedRTPTimestamp(switchingAt) + if err == nil { + expectedTS = uint32(tsExt) + } else { + rtpDiff := uint32(0) + if !f.preStartTime.IsZero() && f.refTSOffset == 0 { + timeSinceFirst := time.Since(f.preStartTime) + rtpDiff = uint32(timeSinceFirst.Nanoseconds() * int64(f.codec.ClockRate) / 1e9) + f.refTSOffset = f.firstTS + rtpDiff - refTS + f.logger.Infow( + "calculating refTSOffset", + "preStartTime", f.preStartTime.String(), + "firstTS", f.firstTS, + "timeSinceFirst", timeSinceFirst, + "rtpDiff", rtpDiff, + "refTS", refTS, + "refTSOffset", f.refTSOffset, + ) + } + expectedTS += rtpDiff + } + } + refTS += f.refTSOffset + + var nextTS uint32 + if f.lastSSRC == 0 { + // If resuming (e. g. on unmute), keep next timestamp close to expected timestamp. + // + // Rationale: + // Case 1: If mute is implemented via something like stopping a track and resuming it on unmute, + // the RTP timestamp may not have jumped across mute valley. In this case, old timestamp + // should not be used. + // + // Case 2: OTOH, something like pacing may be adding latency in the publisher path (even if + // the timestamps incremented correctly across the mute valley). In this case, reference + // timestamp should be used as things will catch up to real time when channel capacity + // increases and pacer starts sending at faster rate. + // + // But, the challenege is distinguishing between the two cases. As a compromise, the difference + // between expectedTS and refTS is thresholded. Difference below the threshold is treated as Case 2 + // and above as Case 1. + // + // In the event of refTS > expectedTS, another threshold is used to pick the next timestamp. + // Ideally, refTS should not be ahead of expectedTS, but expectedTS uses the first packet's + // wall clock time. So, if the first packet experienced abmormal latency, it is possible + // for refTS > expectedTS + diffSeconds := float64(expectedTS-refTS) / float64(f.codec.ClockRate) + if diffSeconds >= 0.0 { + if diffSeconds > ResumeBehindThresholdSeconds { + f.logger.Infow("resume, reference too far behind", "expectedTS", expectedTS, "refTS", refTS, "diffSeconds", diffSeconds) + nextTS = expectedTS + } else { + nextTS = refTS + } + } else { + if math.Abs(diffSeconds) > SwitchAheadThresholdSeconds { + f.logger.Infow("resume, reference too far ahead", "expectedTS", expectedTS, "refTS", refTS, "diffSeconds", math.Abs(diffSeconds)) + nextTS = expectedTS + } else { + nextTS = refTS + } + } + } else { + // switching between layers, check if refTS is too far behind the last sent + diffSeconds := float64(refTS-lastTS) / float64(f.codec.ClockRate) + if diffSeconds < 0.0 { + if math.Abs(diffSeconds) > LayerSwitchBehindThresholdSeconds { + // AVSYNC-TODO: This could be due to pacer trickling out this layer. Should potentially return error here and wait for a more opportune time + // or some forcing function (like "have waited for too long for layer switch, nothing available, switch to whatever is available" kind of condition) + // to do the switch. Just logging it for now. + f.logger.Infow("layer switch, reference too far behind", "expectedTS", expectedTS, "refTS", refTS, "lastTS", lastTS, "diffSeconds", math.Abs(diffSeconds)) + } + // use a nominal increase to ensure that timestamp is always moving forward + nextTS = lastTS + 1 + } else { + diffSeconds = float64(expectedTS-refTS) / float64(f.codec.ClockRate) + if diffSeconds < 0.0 && math.Abs(diffSeconds) > SwitchAheadThresholdSeconds { + f.logger.Infow("layer switch, reference too far ahead", "expectedTS", expectedTS, "refTS", refTS, "diffSeconds", math.Abs(diffSeconds)) + nextTS = expectedTS + } else { + nextTS = refTS + } + } + } + + if nextTS-lastTS == 0 || nextTS-lastTS > (1<<31) { + f.logger.Infow("next timestamp is before last, adjusting", "nextTS", nextTS, "lastTS", lastTS) + // nominal increase + nextTS = lastTS + 1 + } + f.logger.Infow( + "next timestamp on switch", + "switchingAt", switchingAt.String(), + "layer", layer, + "lastTS", lastTS, + "refTS", refTS, + "refTSOffset", f.refTSOffset, + "referenceLayerSpatial", f.referenceLayerSpatial, + "expectedTS", expectedTS, + "nextTS", nextTS, + "jump", nextTS-lastTS, + ) + + f.rtpMunger.UpdateSnTsOffsets(extPkt, 1, nextTS-lastTS) + f.codecMunger.UpdateOffsets(extPkt) + return nil +} + // should be called with lock held func (f *Forwarder) getTranslationParamsCommon(extPkt *buffer.ExtPacket, layer int32, tp *TranslationParams) (*TranslationParams, error) { + if tp == nil { + tp = &TranslationParams{} + } if f.lastSSRC != extPkt.Packet.SSRC { - if !f.started { - f.started = true - f.referenceLayerSpatial = layer - f.rtpMunger.SetLastSnTs(extPkt) - f.codecMunger.SetLast(extPkt) - f.logger.Infow( - "starting forwarding", - "sequenceNumber", extPkt.Packet.SequenceNumber, - "timestamp", extPkt.Packet.Timestamp, - "layer", layer, - "referenceLayerSpatial", f.referenceLayerSpatial, - ) - } else { - if f.referenceLayerSpatial == buffer.InvalidLayerSpatial { - // on a resume, reference layer may not be set, so only set when it is invalid - f.referenceLayerSpatial = layer - } - - // Compute how much time passed between the old RTP extPkt - // and the current packet, and fix timestamp on source change - // - // There are three time stamps to consider here - // 1. lastTS -> time stamp of last sent packet - // 2. refTS -> time stamp of this packet (after munging) calculated using feed's RTCP sender report - // 3. expectedTS -> time stamp of this packet (after munging) calculated using this stream's RTCP sender report - // Ideally, refTS and expectedTS should be very close and lastTS should be before both of those. - // But, cases like muting/unmuting, clock vagaries make them not satisfy those conditions always. - // - // There are 6 orderings to consider (considering only inequalities). Resolve them using following rules - // 1. Timestamp has to move forward - // 2. Keep next time stamp close to expected - lastTS := f.rtpMunger.GetLast().LastTS - refTS := lastTS - expectedTS := lastTS - minTS := ^uint64(0) - switchingAt := time.Now() - if f.getReferenceLayerRTPTimestamp != nil { - ts, err := f.getReferenceLayerRTPTimestamp(extPkt.Packet.Timestamp, layer, f.referenceLayerSpatial) - if err == nil { - refTS = ts - } - } - if f.getExpectedRTPTimestamp != nil { - ts, min, err := f.getExpectedRTPTimestamp(switchingAt) - if err == nil { - expectedTS = ts - minTS = min - } else { - rtpDiff := uint32(0) - if !f.preStartTime.IsZero() && f.refTSOffset == 0 { - timeSinceFirst := time.Since(f.preStartTime) - rtpDiff = uint32(timeSinceFirst.Nanoseconds() * int64(f.codec.ClockRate) / 1e9) - f.refTSOffset = f.firstTS + rtpDiff - refTS - f.logger.Infow( - "calculating refTSOffset", - "preStartTime", f.preStartTime.String(), - "firstTS", f.firstTS, - "timeSinceFirst", timeSinceFirst, - "rtpDiff", rtpDiff, - "refTS", refTS, - "refTSOffset", f.refTSOffset, - ) - } - expectedTS += rtpDiff - } - } - refTS += f.refTSOffset - nextTS, explain := getNextTimestamp(lastTS, refTS, expectedTS, minTS) - f.logger.Infow( - "next timestamp on switch", - "switchingAt", switchingAt.String(), - "layer", layer, - "lastTS", lastTS, - "refTS", refTS, - "refTSOffset", f.refTSOffset, - "referenceLayerSpatial", f.referenceLayerSpatial, - "expectedTS", expectedTS, - "minTS", minTS, - "nextTS", nextTS, - "jump", nextTS-lastTS, - "explanation", explain, - ) - - f.rtpMunger.UpdateSnTsOffsets(extPkt, 1, nextTS-lastTS) - f.codecMunger.UpdateOffsets(extPkt) + if err := f.processSourceSwitch(extPkt, layer); err != nil { + tp.shouldDrop = true + return tp, err } - f.logger.Debugw("switching feed", "from", f.lastSSRC, "to", extPkt.Packet.SSRC) f.lastSSRC = extPkt.Packet.SSRC } - if tp == nil { - tp = &TranslationParams{} - } tpRTP, err := f.rtpMunger.UpdateAndGetSnTs(extPkt) if err != nil { tp.shouldDrop = true @@ -1696,7 +1766,7 @@ func (f *Forwarder) maybeStart() { Packet: &rtp.Packet{ Header: rtp.Header{ SequenceNumber: uint16(rand.Intn(1<<14)) + uint16(1<<15), // a random number in third quartile of sequence number space - Timestamp: uint32(rand.Intn(1<<30)) + uint32(1<<31), // a random number in third quartile of time stamp space + Timestamp: uint32(rand.Intn(1<<30)) + uint32(1<<31), // a random number in third quartile of timestamp space }, }, } @@ -1741,16 +1811,16 @@ func (f *Forwarder) GetSnTsForBlankFrames(frameRate uint32, numPackets int) ([]S lastTS := f.rtpMunger.GetLast().LastTS expectedTS := lastTS - minTS := ^uint64(0) if f.getExpectedRTPTimestamp != nil { - ts, min, err := f.getExpectedRTPTimestamp(time.Now()) + tsExt, err := f.getExpectedRTPTimestamp(time.Now()) if err == nil { - expectedTS = ts - minTS = min + expectedTS = uint32(tsExt) } } - nextTS, _ := getNextTimestamp(lastTS, expectedTS, expectedTS, minTS) - snts, err := f.rtpMunger.UpdateAndGetPaddingSnTs(numPackets, f.codec.ClockRate, frameRate, frameEndNeeded, nextTS) + if expectedTS-lastTS == 0 || expectedTS-lastTS > (1<<31) { + expectedTS = lastTS + 1 + } + snts, err := f.rtpMunger.UpdateAndGetPaddingSnTs(numPackets, f.codec.ClockRate, frameRate, frameEndNeeded, expectedTS) return snts, frameEndNeeded, err } @@ -1888,44 +1958,3 @@ done: return float64(distance) / float64(maxSeenLayer.Temporal+1) } - -func getNextTimestamp(lastTS uint32, refTS uint32, expectedTS uint32, minTS uint64) (uint32, string) { - isInOrder := func(val1, val2 uint32) bool { - diff := val1 - val2 - return diff != 0 && diff < (1<<31) - } - - rl := isInOrder(refTS, lastTS) - el := isInOrder(expectedTS, lastTS) - er := isInOrder(expectedTS, refTS) - - nextTS := lastTS + 1 - explain := "l = r = e" - - switch { - case rl && el && er: // lastTS < refTS < expectedTS - nextTS = uint32(float64(refTS) + 0.05*float64(expectedTS-refTS)) - explain = fmt.Sprintf("l < r < e, %d, %d", refTS-lastTS, expectedTS-refTS) - case rl && el && !er: // lastTS < expectedTS < refTS - nextTS = uint32(float64(expectedTS) + 0.5*float64(refTS-expectedTS)) - explain = fmt.Sprintf("l < e < r, %d, %d", expectedTS-lastTS, refTS-expectedTS) - case !rl && el && er: // refTS < lastTS < expectedTS - nextTS = uint32(float64(lastTS) + 0.5*float64(expectedTS-lastTS)) - explain = fmt.Sprintf("r < l < e, %d, %d", lastTS-refTS, expectedTS-lastTS) - case !rl && !el && er: // refTS < expectedTS < lastTS - nextTS = lastTS + 1 - explain = fmt.Sprintf("r < e < l, %d, %d", expectedTS-refTS, lastTS-expectedTS) - case rl && !el && !er: // expectedTS < lastTS < refTS - nextTS = uint32(float64(lastTS) + 0.75*float64(refTS-lastTS)) - explain = fmt.Sprintf("e < l < r, %d, %d", lastTS-expectedTS, refTS-lastTS) - case !rl && !el && !er: // expectedTS < refTS < lastTS - nextTS = lastTS + 1 - explain = fmt.Sprintf("e < r < l, %d, %d", refTS-expectedTS, lastTS-refTS) - } - - if minTS != ^uint64(0) && !isInOrder(nextTS, uint32(minTS)) { - nextTS = uint32(minTS) + 1 - } - - return nextTS, explain -} diff --git a/pkg/sfu/receiver.go b/pkg/sfu/receiver.go index 837b534ab..410143844 100644 --- a/pkg/sfu/receiver.go +++ b/pkg/sfu/receiver.go @@ -66,7 +66,7 @@ type TrackReceiver interface { GetTemporalLayerFpsForSpatial(layer int32) []float32 - GetRTCPSenderReportData(layer int32) (*buffer.RTCPSenderReportData, *buffer.RTCPSenderReportData) + GetCalculatedClockRate(layer int32) uint32 GetReferenceLayerRTPTimestamp(ts uint32, layer int32, referenceLayer int32) (uint32, error) } @@ -752,8 +752,8 @@ func (w *WebRTCReceiver) GetTemporalLayerFpsForSpatial(layer int32) []float32 { return b.GetTemporalLayerFpsForSpatial(layer) } -func (w *WebRTCReceiver) GetRTCPSenderReportData(layer int32) (*buffer.RTCPSenderReportData, *buffer.RTCPSenderReportData) { - return w.streamTrackerManager.GetRTCPSenderReportData(layer) +func (w *WebRTCReceiver) GetCalculatedClockRate(layer int32) uint32 { + return w.streamTrackerManager.GetCalculatedClockRate(layer) } func (w *WebRTCReceiver) GetReferenceLayerRTPTimestamp(ts uint32, layer int32, referenceLayer int32) (uint32, error) { diff --git a/pkg/sfu/streamtrackermanager.go b/pkg/sfu/streamtrackermanager.go index ef92d6618..1a5d24ffe 100644 --- a/pkg/sfu/streamtrackermanager.go +++ b/pkg/sfu/streamtrackermanager.go @@ -2,6 +2,7 @@ package sfu import ( "fmt" + "math" "sort" "sync" "time" @@ -15,6 +16,14 @@ import ( "github.com/livekit/protocol/logger" ) +const ( + senderReportThresholdSeconds = float64(60.0) + + minDurationForClockRateCalculation = 15 * time.Second +) + +// --------------------------------------------------- + type StreamTrackerManagerListener interface { OnAvailableLayersChanged() OnBitrateAvailabilityChanged() @@ -24,9 +33,12 @@ type StreamTrackerManagerListener interface { OnBitrateReport(availableLayers []int32, bitrates Bitrates) } +// --------------------------------------------------- + type endsSenderReport struct { - first *buffer.RTCPSenderReportData - newest *buffer.RTCPSenderReportData + first *buffer.RTCPSenderReportData + newest *buffer.RTCPSenderReportData + lastUpdated time.Time } type StreamTrackerManager struct { @@ -50,6 +62,7 @@ type StreamTrackerManager struct { senderReportMu sync.RWMutex senderReports [buffer.DefaultMaxLayerSpatial + 1]endsSenderReport + layerOffsets [buffer.DefaultMaxLayerSpatial + 1][buffer.DefaultMaxLayerSpatial + 1]uint32 closed core.Fuse @@ -517,6 +530,40 @@ func (s *StreamTrackerManager) maxExpectedLayerFromTrackInfo() { } } +func (s *StreamTrackerManager) updateLayerOffsetLocked(ref, other int32) { + srRef := s.senderReports[ref].newest + srOther := s.senderReports[other].newest + if srRef == nil || srRef.NTPTimestamp == 0 || srOther == nil || srOther.NTPTimestamp == 0 { + return + } + + ntpDiff := srRef.NTPTimestamp.Time().Sub(srOther.NTPTimestamp.Time()) + if math.Abs(ntpDiff.Seconds()) > senderReportThresholdSeconds { + // offset is updated only if the layers' sender reports are close enough. + // + // Rationale: higher layers could be paused for extended periods of time + // due to adaptive stream/dynacast or publisher constraints like CPU/bandwidth. + // The check is to avoid using very old reports. + return + } + rtpDiff := ntpDiff.Nanoseconds() * int64(s.clockRate) / 1e9 + + // calculate other layer's time stamp at the same time as ref layer's NTP time + normalizedOtherTS := srOther.RTPTimestamp + uint32(rtpDiff) + + // now both layers' time stamp refer to the same NTP time and the diff is the offset between the layers + offset := srRef.RTPTimestamp - normalizedOtherTS + + // use minimal offset to indicate value availability in the extremely unlikely case of + // both layers using the same timestamp + if offset == 0 { + s.logger.Infow("using default offset", "ref", ref, "other", other) + offset = 1 + } + + s.layerOffsets[ref][other] = offset +} + func (s *StreamTrackerManager) SetRTCPSenderReportData(layer int32, srFirst *buffer.RTCPSenderReportData, srNewest *buffer.RTCPSenderReportData) { s.senderReportMu.Lock() defer s.senderReportMu.Unlock() @@ -527,74 +574,66 @@ func (s *StreamTrackerManager) SetRTCPSenderReportData(layer int32, srFirst *buf s.senderReports[layer].first = srFirst s.senderReports[layer].newest = srNewest + s.senderReports[layer].lastUpdated = time.Now() + + // (re)fill offsets as necessary for received layer. + for i := int32(0); i < buffer.DefaultMaxLayerSpatial+1; i++ { + if i == layer { + continue + } + + // treating layer for which report was received as reference layer + s.updateLayerOffsetLocked(layer, i) + + // and the other way + s.updateLayerOffsetLocked(i, layer) + } } -func (s *StreamTrackerManager) GetRTCPSenderReportData(layer int32) (*buffer.RTCPSenderReportData, *buffer.RTCPSenderReportData) { +func (s *StreamTrackerManager) GetCalculatedClockRate(layer int32) uint32 { s.senderReportMu.RLock() defer s.senderReportMu.RUnlock() if layer < 0 || int(layer) >= len(s.senderReports) { - return nil, nil + // invalid layer + return 0 } - return s.senderReports[layer].first, s.senderReports[layer].newest + srFirst := s.senderReports[layer].first + srNewest := s.senderReports[layer].newest + if srFirst == nil || srFirst.NTPTimestamp == 0 || srNewest == nil || srNewest.NTPTimestamp == 0 || srFirst.RTPTimestamp == srNewest.RTPTimestamp { + // sender reports invalid or same + return 0 + } + + if s.senderReports[layer].lastUpdated.IsZero() || time.Since(s.senderReports[layer].lastUpdated).Seconds() > senderReportThresholdSeconds { + // sender report updated too far back + return 0 + } + + tsf := srNewest.NTPTimestamp.Time().Sub(srFirst.NTPTimestamp.Time()) + if tsf < minDurationForClockRateCalculation { + // not enough time has elapsed to get a stable clock rate calculation + return 0 + } + + rdsf := srNewest.RTPTimestampExt - srFirst.RTPTimestampExt + return uint32(float64(rdsf) / tsf.Seconds()) } func (s *StreamTrackerManager) GetReferenceLayerRTPTimestamp(ts uint32, layer int32, referenceLayer int32) (uint32, error) { s.senderReportMu.RLock() defer s.senderReportMu.RUnlock() - if layer < 0 || referenceLayer < 0 { + if layer < 0 || int(layer) >= len(s.layerOffsets[0]) || referenceLayer < 0 || int(referenceLayer) >= len(s.layerOffsets) { return 0, fmt.Errorf("invalid layer, target: %d, reference: %d", layer, referenceLayer) } - if layer == referenceLayer { - return ts, nil + if layer != referenceLayer && s.layerOffsets[referenceLayer][layer] == 0 { + return 0, fmt.Errorf("offset unavailable, target: %d, reference: %d", layer, referenceLayer) } - var srLayer *buffer.RTCPSenderReportData - if int(layer) < len(s.senderReports) { - srLayer = s.senderReports[layer].newest - } - if srLayer == nil || srLayer.NTPTimestamp == 0 { - return 0, fmt.Errorf("layer rtcp sender report not available: %d", layer) - } - - var srRef *buffer.RTCPSenderReportData - if int(referenceLayer) < len(s.senderReports) { - srRef = s.senderReports[referenceLayer].newest - } - if srRef == nil || srRef.NTPTimestamp == 0 { - return 0, fmt.Errorf("reference layer rtcp sender report not available: %d", referenceLayer) - } - - // line up the RTP time stamps using NTP time of most recent sender report of layer and referenceLayer - // NOTE: It is possible that reference layer has stopped (due to dynacast/adaptive streaming OR publisher - // constraints). It should be okay even if the layer has stopped for a long time when using modulo arithmetic for - // RTP time stamp (uint32 arithmetic). - ntpDiff := srRef.NTPTimestamp.Time().Sub(srLayer.NTPTimestamp.Time()) - rtpDiff := ntpDiff.Nanoseconds() * int64(s.clockRate) / 1e9 - normalizedTS := srLayer.RTPTimestamp + uint32(rtpDiff) - s.logger.Infow( - "getting reference timestamp", - "layer", layer, - "referenceLayer", referenceLayer, - "incomingTS", ts, - "layerNTP", srLayer.NTPTimestamp.Time().String(), - "refNTP", srRef.NTPTimestamp.Time().String(), - "ntpDiff", ntpDiff.String(), - "layerRTP", srLayer.RTPTimestamp, - "refRTP", srRef.RTPTimestamp, - "rtpDiff", rtpDiff, - "normalizedTS", normalizedTS, - "mappedTS", ts+(srRef.RTPTimestamp-normalizedTS), - ) - - // now that both RTP timestamps correspond to roughly the same NTP time, - // the diff between them is the offset in RTP timestamp units between layer and referenceLayer. - // Add the offset to layer's ts to map it to corresponding RTP timestamp in - // the reference layer. - return ts + (srRef.RTPTimestamp - normalizedTS), nil + return ts + s.layerOffsets[referenceLayer][layer], nil } func (s *StreamTrackerManager) GetMaxTemporalLayerSeen() int32 { From 469f1cd073b2806c463a94f3831bba5ae4aa6748 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Sat, 15 Jul 2023 12:43:05 +0530 Subject: [PATCH 06/18] Minor changes to publisher bool. (#1880) * Minor changes to publisher bool. * address feedback --- pkg/rtc/participant.go | 61 ++++++++++++++++++++++-------------------- pkg/rtc/room.go | 3 +-- 2 files changed, 33 insertions(+), 31 deletions(-) diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 5de6904c1..e0d5214d4 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -370,13 +370,14 @@ func (p *ParticipantImpl) SetPermission(permission *livekit.ParticipantPermissio return false } - p.GetLogger().Infow("updating participant permission", "permission", permission) + p.params.Logger.Infow("updating participant permission", "permission", permission) video.UpdateFromPermission(permission) p.dirty.Store(true) canPublish := video.GetCanPublish() canSubscribe := video.GetCanSubscribe() + onParticipantUpdate := p.onParticipantUpdate onClaimsChanged := p.onClaimsChanged @@ -387,13 +388,7 @@ func (p *ParticipantImpl) SetPermission(permission *livekit.ParticipantPermissio // publish permission has been revoked then remove offending tracks for _, track := range p.GetPublishedTracks() { if !video.GetCanPublishSource(track.Source()) { - p.RemovePublishedTrack(track, false, false) - if p.ProtocolVersion().SupportsUnpublish() { - p.sendTrackUnpublished(track.ID()) - } else { - // for older clients that don't support unpublish, mute to avoid them sending data - p.sendTrackMuted(track.ID(), true) - } + p.removePublishedTrack(track) } } @@ -1190,22 +1185,24 @@ func (p *ParticipantImpl) updateState(state livekit.ParticipantInfo_State) { } func (p *ParticipantImpl) setIsPublisher(isPublisher bool) { - if p.isPublisher.Swap(isPublisher) != isPublisher { - p.lock.Lock() - p.requireBroadcast = true - p.lock.Unlock() + if p.isPublisher.Swap(isPublisher) == isPublisher { + return + } - p.dirty.Store(true) + p.lock.Lock() + p.requireBroadcast = true + p.lock.Unlock() - // trigger update as well if participant is already fully connected - if p.State() == livekit.ParticipantInfo_ACTIVE { - p.lock.RLock() - onParticipantUpdate := p.onParticipantUpdate - p.lock.RUnlock() + p.dirty.Store(true) - if onParticipantUpdate != nil { - onParticipantUpdate(p) - } + // trigger update as well if participant is already fully connected + if p.State() == livekit.ParticipantInfo_ACTIVE { + p.lock.RLock() + onParticipantUpdate := p.onParticipantUpdate + p.lock.RUnlock() + + if onParticipantUpdate != nil { + onParticipantUpdate(p) } } } @@ -1220,6 +1217,16 @@ func (p *ParticipantImpl) onSubscriberOffer(offer webrtc.SessionDescription) err }) } +func (p *ParticipantImpl) removePublishedTrack(track types.MediaTrack) { + p.RemovePublishedTrack(track, false, false) + if p.ProtocolVersion().SupportsUnpublish() { + p.sendTrackUnpublished(track.ID()) + } else { + // for older clients that don't support unpublish, mute to avoid them sending data + p.sendTrackMuted(track.ID(), true) + } +} + // when a new remoteTrack is created, creates a Track and adds it to room func (p *ParticipantImpl) onMediaTrack(track *webrtc.TrackRemote, rtpReceiver *webrtc.RTPReceiver) { if p.IsDisconnected() { @@ -1242,12 +1249,12 @@ func (p *ParticipantImpl) onMediaTrack(track *webrtc.TrackRemote, rtpReceiver *w p.params.Logger.Warnw("no permission to publish mediaTrack", nil, "source", publishedTrack.Source(), ) + p.removePublishedTrack(publishedTrack) return } - if !p.IsPublisher() { - p.setIsPublisher(true) - } + p.setIsPublisher(true) + p.dirty.Store(true) p.params.Logger.Infow("mediaTrack published", "kind", track.Kind().String(), @@ -1258,8 +1265,6 @@ func (p *ParticipantImpl) onMediaTrack(track *webrtc.TrackRemote, rtpReceiver *w "mime", track.Codec().MimeType, ) - p.dirty.Store(true) - if !isNewTrack && !publishedTrack.HasPendingCodec() && p.IsReady() { p.lock.RLock() onTrackUpdated := p.onTrackUpdated @@ -1300,9 +1305,7 @@ func (p *ParticipantImpl) onDataMessage(kind livekit.DataPacket_Kind, data []byt p.params.Logger.Warnw("received unsupported data packet", nil, "payload", payload) } - if !p.IsPublisher() { - p.setIsPublisher(true) - } + p.setIsPublisher(true) } func (p *ParticipantImpl) onICECandidate(c *webrtc.ICECandidate, target livekit.SignalTarget) error { diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index a90b90676..7602659ac 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -828,7 +828,6 @@ func (r *Room) createJoinResponseLocked(participant types.LocalParticipant, iceS func (r *Room) onTrackPublished(participant types.LocalParticipant, track types.MediaTrack) { // publish participant update, since track state is changed r.broadcastParticipantState(participant, broadcastOptions{skipSource: true}) - r.protoProxy.MarkDirty(false) r.lock.RLock() // subscribe all existing participants to this MediaTrack @@ -888,7 +887,6 @@ func (r *Room) onTrackUpdated(p types.LocalParticipant, _ types.MediaTrack) { func (r *Room) onTrackUnpublished(p types.LocalParticipant, track types.MediaTrack) { r.trackManager.RemoveTrack(track) - r.protoProxy.MarkDirty(false) if !p.IsClosed() { r.broadcastParticipantState(p, broadcastOptions{skipSource: true}) } @@ -898,6 +896,7 @@ func (r *Room) onTrackUnpublished(p types.LocalParticipant, track types.MediaTra } func (r *Room) onParticipantUpdate(p types.LocalParticipant) { + r.protoProxy.MarkDirty(false) // immediately notify when permissions or metadata changed r.broadcastParticipantState(p, broadcastOptions{immediate: true}) if r.onParticipantChanged != nil { From 06d8459234b04a20abe8475ab335a35ac21c0451 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Sat, 15 Jul 2023 14:09:22 +0530 Subject: [PATCH 07/18] Pick up proto proxy no update on no change (#1881) --- go.mod | 2 +- go.sum | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/go.mod b/go.mod index 894e44d78..fad272d8d 100644 --- a/go.mod +++ b/go.mod @@ -18,7 +18,7 @@ require ( github.com/jxskiss/base62 v1.1.0 github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1 github.com/livekit/mediatransportutil v0.0.0-20230612070454-d5299b956135 - github.com/livekit/protocol v1.5.10-0.20230714010226-3c53edc91962 + github.com/livekit/protocol v1.5.10-0.20230715082801-9d0d6c9e9427 github.com/livekit/psrpc v0.3.2 github.com/mackerelio/go-osstat v0.2.4 github.com/magefile/mage v1.15.0 diff --git a/go.sum b/go.sum index dba27c2ea..1c95ae879 100644 --- a/go.sum +++ b/go.sum @@ -124,8 +124,8 @@ github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1 h1:jm09419p0lqTkD github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ= github.com/livekit/mediatransportutil v0.0.0-20230612070454-d5299b956135 h1:lWYbsondvqG69czxoACDwaJ/BoyD57BahCo70ZH+m4U= github.com/livekit/mediatransportutil v0.0.0-20230612070454-d5299b956135/go.mod h1:MRc0zSOSzXuFt0X218SgabzlaKevkvCckPgBEoHYc34= -github.com/livekit/protocol v1.5.10-0.20230714010226-3c53edc91962 h1:y+rtYNMGmvpEgQlNG/wOUO16S497ygh83wQSUBKHpG4= -github.com/livekit/protocol v1.5.10-0.20230714010226-3c53edc91962/go.mod h1:eRzojAYSPJuNgDHMlvLji/CPauj9hrgvb6rVPUj6MoU= +github.com/livekit/protocol v1.5.10-0.20230715082801-9d0d6c9e9427 h1:NyNdCT8+glCnGGdkgbEO2rpX1iRscwp4HCf1u21Clzo= +github.com/livekit/protocol v1.5.10-0.20230715082801-9d0d6c9e9427/go.mod h1:eRzojAYSPJuNgDHMlvLji/CPauj9hrgvb6rVPUj6MoU= github.com/livekit/psrpc v0.3.2 h1:eAaJhASme33gtoBhCRLH9jsnWcdm1tHWf0WzaDk56ew= github.com/livekit/psrpc v0.3.2/go.mod h1:n6JntEg+zT6Ji8InoyTpV7wusPNwGqqtxmHlkNhDN0U= github.com/mackerelio/go-osstat v0.2.4 h1:qxGbdPkFo65PXOb/F/nhDKpF2nGmGaCFDLXoZjJTtUs= From 11e1eb00fa5d7fedc89ff9b2700738b259707e51 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Sun, 16 Jul 2023 23:28:20 +0530 Subject: [PATCH 08/18] Attempt to avoid out-of-order max subscribed layer notifications. (#1882) * Check for request layer lock only in the goroutine * check before sending PLI * max layer notifier worker * test cleanup * clean up * do notification in the callback --- pkg/sfu/downtrack.go | 192 +++++++++++------- pkg/sfu/forwarder.go | 69 ++++--- pkg/sfu/forwarder_test.go | 34 ++-- pkg/sfu/streamallocator/streamallocator.go | 4 + .../dependencydescriptor.go | 20 -- pkg/sfu/videolayerselector/simulcast.go | 13 -- .../videolayerselector/videolayerselector.go | 3 - pkg/sfu/videolayerselector/vp9.go | 20 -- 8 files changed, 183 insertions(+), 172 deletions(-) diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index c6dbb34fd..e3dd65a04 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -42,6 +42,8 @@ type TrackSender interface { HandleRTCPSenderReportData(payloadType webrtc.PayloadType, layer int32, srData *buffer.RTCPSenderReportData) error } +// ------------------------------------------------------------------- + const ( RTPPaddingMaxPayloadSize = 255 RTPPaddingEstimatedHeaderSize = 20 @@ -60,6 +62,8 @@ const ( maxPaddingOnMuteDuration = 5 * time.Second ) +// ------------------------------------------------------------------- + var ( ErrUnknownKind = errors.New("unknown kind of codec") ErrOutOfOrderSequenceNumberCacheMiss = errors.New("out-of-order sequence number not found in cache") @@ -197,14 +201,13 @@ type DownTrack struct { transceiver *webrtc.RTPTransceiver writeStream webrtc.TrackLocalWriter rtcpReader *buffer.RTCPReader - onCloseHandler func(willBeResumed bool) - onBinding func(error) listenerLock sync.RWMutex receiverReportListeners []ReceiverReportListener - bindLock sync.Mutex - bound atomic.Bool + bindLock sync.Mutex + bound atomic.Bool + onBinding func(error) isClosed atomic.Bool connected atomic.Bool @@ -235,14 +238,13 @@ type DownTrack struct { pacer pacer.Pacer - // update stats - onStatsUpdate func(dt *DownTrack, stat *livekit.AnalyticsStat) + maxLayerNotifierCh chan struct{} - // when max subscribed layer changes + cbMu sync.RWMutex + onStatsUpdate func(dt *DownTrack, stat *livekit.AnalyticsStat) onMaxSubscribedLayerChanged func(dt *DownTrack, layer int32) - - // update rtt - onRttUpdate func(dt *DownTrack, rtt uint32) + onRttUpdate func(dt *DownTrack, rtt uint32) + onCloseHandler func(willBeResumed bool) } // NewDownTrack returns a DownTrack. @@ -266,17 +268,18 @@ func NewDownTrack( } d := &DownTrack{ - logger: logger, - id: r.TrackID(), - subscriberID: subID, - maxTrack: mt, - streamID: r.StreamID(), - bufferFactory: bf, - receiver: r, - upstreamCodecs: codecs, - kind: kind, - codec: codecs[0].RTPCodecCapability, - pacer: pacer, + logger: logger, + id: r.TrackID(), + subscriberID: subID, + maxTrack: mt, + streamID: r.StreamID(), + bufferFactory: bf, + receiver: r, + upstreamCodecs: codecs, + kind: kind, + codec: codecs[0].RTPCodecCapability, + pacer: pacer, + maxLayerNotifierCh: make(chan struct{}, 20), } d.forwarder = NewForwarder( d.kind, @@ -307,11 +310,15 @@ func NewDownTrack( Logger: d.logger.WithValues("direction", "down"), }) d.connectionStats.OnStatsUpdate(func(_cs *connectionquality.ConnectionStats, stat *livekit.AnalyticsStat) { - if d.onStatsUpdate != nil { - d.onStatsUpdate(d, stat) + if onStatsUpdate := d.getOnStatsUpdate(); onStatsUpdate != nil { + onStatsUpdate(d, stat) } }) + if d.kind == webrtc.RTPCodecTypeVideo { + go d.maxLayerNotifierWorker() + } + return d, nil } @@ -541,6 +548,7 @@ func (d *DownTrack) keyFrameRequester(generation uint32, layer int32) { if d.IsClosed() || layer == buffer.InvalidLayerSpatial { return } + interval := 2 * d.rtpStats.GetRtt() if interval < keyFrameIntervalMin { interval = keyFrameIntervalMin @@ -550,7 +558,13 @@ func (d *DownTrack) keyFrameRequester(generation uint32, layer int32) { } ticker := time.NewTicker(time.Duration(interval) * time.Millisecond) defer ticker.Stop() + for { + locked, _ := d.forwarder.CheckSync() + if locked { + return + } + if d.connected.Load() { d.logger.Debugw("sending PLI for layer lock", "generation", generation, "layer", layer) d.receiver.SendPLI(layer, false) @@ -565,6 +579,34 @@ func (d *DownTrack) keyFrameRequester(generation uint32, layer int32) { } } +func (d *DownTrack) postMaxLayerNotifierEvent() { + if d.IsClosed() { + return + } + + select { + case d.maxLayerNotifierCh <- struct{}{}: + default: + d.logger.Warnw("max layer notifier event queue full", nil) + } +} + +func (d *DownTrack) maxLayerNotifierWorker() { + more := true + for more { + _, more = <-d.maxLayerNotifierCh + + maxLayerSpatial := buffer.InvalidLayerSpatial + if more { + maxLayerSpatial = d.forwarder.GetMaxSubscribedSpatial() + } + if onMaxSubscribedLayerChanged := d.getOnMaxLayerChanged(); onMaxSubscribedLayerChanged != nil { + d.logger.Infow("max subscribed layer changed", maxLayerSpatial) + onMaxSubscribedLayerChanged(d, maxLayerSpatial) + } + } +} + // WriteRTP writes an RTP Packet to the DownTrack func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error { if !d.bound.Load() || !d.connected.Load() { @@ -725,17 +767,17 @@ func (d *DownTrack) WritePaddingRTP(bytesToSend int, paddingOnMute bool, forceMa // Mute enables or disables media forwarding - subscriber triggered func (d *DownTrack) Mute(muted bool) { - changed, maxLayer := d.forwarder.Mute(muted) - d.handleMute(muted, false, changed, maxLayer) + changed := d.forwarder.Mute(muted) + d.handleMute(muted, changed) } // PubMute enables or disables media forwarding - publisher side func (d *DownTrack) PubMute(pubMuted bool) { - changed, maxLayer := d.forwarder.PubMute(pubMuted) - d.handleMute(pubMuted, true, changed, maxLayer) + changed := d.forwarder.PubMute(pubMuted) + d.handleMute(pubMuted, changed) } -func (d *DownTrack) handleMute(muted bool, isPub bool, changed bool, maxLayer buffer.VideoLayer) { +func (d *DownTrack) handleMute(muted bool, changed bool) { if !changed { return } @@ -762,18 +804,7 @@ func (d *DownTrack) handleMute(muted bool, isPub bool, changed bool, maxLayer bu // Note that while publisher mute is active, subscriber changes can also happen // and that could turn on/off layers on publisher side. // - if !isPub && d.onMaxSubscribedLayerChanged != nil && d.kind == webrtc.RTPCodecTypeVideo { - notifyLayer := buffer.InvalidLayerSpatial - if !muted { - // - // When unmuting, don't wait for layer lock as - // client might need to be notified to start layers - // before locking can happen in the forwarder. - // - notifyLayer = maxLayer.Spatial - } - d.onMaxSubscribedLayerChanged(d, notifyLayer) - } + d.postMaxLayerNotifierEvent() if sal := d.getStreamAllocatorListener(); sal != nil { sal.OnSubscriptionChanged(d) @@ -856,12 +887,10 @@ func (d *DownTrack) CloseWithFlush(flush bool) { d.rtpStats.Stop() d.logger.Infow("rtp stats", "direction", "downstream", "mime", d.mime, "ssrc", d.ssrc, "stats", d.rtpStats.ToString()) - if d.onMaxSubscribedLayerChanged != nil && d.kind == webrtc.RTPCodecTypeVideo { - d.onMaxSubscribedLayerChanged(d, buffer.InvalidLayerSpatial) - } + close(d.maxLayerNotifierCh) - if d.onCloseHandler != nil { - d.onCloseHandler(!flush) + if onCloseHandler := d.getOnCloseHandler(); onCloseHandler != nil { + onCloseHandler(!flush) } d.stopKeyFrameRequester() @@ -869,21 +898,12 @@ func (d *DownTrack) CloseWithFlush(flush bool) { } func (d *DownTrack) SetMaxSpatialLayer(spatialLayer int32) { - changed, maxLayer, currentLayer := d.forwarder.SetMaxSpatialLayer(spatialLayer) + changed, maxLayer := d.forwarder.SetMaxSpatialLayer(spatialLayer) if !changed { return } - if d.onMaxSubscribedLayerChanged != nil && d.kind == webrtc.RTPCodecTypeVideo && maxLayer.SpatialGreaterThanOrEqual(currentLayer) { - // - // Notify when new max is - // 1. Equal to current -> already locked to the new max - // 2. Greater than current -> two scenarios - // a. is higher than previous max -> client may need to start higher layer before forwarder can lock - // b. is lower than previous max -> client can stop higher layer(s) - // - d.onMaxSubscribedLayerChanged(d, maxLayer.Spatial) - } + d.postMaxLayerNotifierEvent() if sal := d.getStreamAllocatorListener(); sal != nil { sal.OnSubscribedLayerChanged(d, maxLayer) @@ -891,7 +911,7 @@ func (d *DownTrack) SetMaxSpatialLayer(spatialLayer int32) { } func (d *DownTrack) SetMaxTemporalLayer(temporalLayer int32) { - changed, maxLayer, _ := d.forwarder.SetMaxTemporalLayer(temporalLayer) + changed, maxLayer := d.forwarder.SetMaxTemporalLayer(temporalLayer) if !changed { return } @@ -973,10 +993,23 @@ func (d *DownTrack) UpTrackBitrateReport(availableLayers []int32, bitrates Bitra // OnCloseHandler method to be called on remote tracked removed func (d *DownTrack) OnCloseHandler(fn func(willBeResumed bool)) { + d.cbMu.Lock() + defer d.cbMu.Unlock() + d.onCloseHandler = fn } +func (d *DownTrack) getOnCloseHandler() func(willBeResumed bool) { + d.cbMu.RLock() + defer d.cbMu.RUnlock() + + return d.onCloseHandler +} + func (d *DownTrack) OnBinding(fn func(error)) { + d.bindLock.Lock() + defer d.bindLock.Unlock() + d.onBinding = fn } @@ -988,17 +1021,47 @@ func (d *DownTrack) AddReceiverReportListener(listener ReceiverReportListener) { } func (d *DownTrack) OnStatsUpdate(fn func(dt *DownTrack, stat *livekit.AnalyticsStat)) { + d.cbMu.Lock() + defer d.cbMu.Unlock() + d.onStatsUpdate = fn } +func (d *DownTrack) getOnStatsUpdate() func(dt *DownTrack, stat *livekit.AnalyticsStat) { + d.cbMu.RLock() + defer d.cbMu.RUnlock() + + return d.onStatsUpdate +} + func (d *DownTrack) OnRttUpdate(fn func(dt *DownTrack, rtt uint32)) { + d.cbMu.Lock() + defer d.cbMu.Unlock() + d.onRttUpdate = fn } +func (d *DownTrack) getOnRttUpdate() func(dt *DownTrack, rtt uint32) { + d.cbMu.RLock() + defer d.cbMu.RUnlock() + + return d.onRttUpdate +} + func (d *DownTrack) OnMaxLayerChanged(fn func(dt *DownTrack, layer int32)) { + d.cbMu.Lock() + defer d.cbMu.Unlock() + d.onMaxSubscribedLayerChanged = fn } +func (d *DownTrack) getOnMaxLayerChanged() func(dt *DownTrack, layer int32) { + d.cbMu.RLock() + defer d.cbMu.RUnlock() + + return d.onMaxSubscribedLayerChanged +} + func (d *DownTrack) IsDeficient() bool { return d.forwarder.IsDeficient() } @@ -1355,8 +1418,8 @@ func (d *DownTrack) handleRTCP(bytes []byte) { d.sequencer.setRTT(rttToReport) } - if d.onRttUpdate != nil { - d.onRttUpdate(d, rttToReport) + if onRttUpdate := d.getOnRttUpdate(); onRttUpdate != nil { + onRttUpdate(d, rttToReport) } } } @@ -1744,15 +1807,8 @@ func (d *DownTrack) packetSent(md interface{}, hdr *rtp.Header, payloadSize int, } if spmd.tp != nil { - if spmd.tp.isSwitchingToMaxSpatial && d.onMaxSubscribedLayerChanged != nil && d.kind == webrtc.RTPCodecTypeVideo { - d.onMaxSubscribedLayerChanged(d, spmd.tp.maxSpatialLayer) - } - - if spmd.tp.isSwitchingToRequestSpatial { - locked, _ := d.forwarder.CheckSync() - if locked { - d.stopKeyFrameRequester() - } + if spmd.tp.isSwitching { + d.postMaxLayerNotifierEvent() } if spmd.tp.isResuming { diff --git a/pkg/sfu/forwarder.go b/pkg/sfu/forwarder.go index 1da413ee8..23a8d694d 100644 --- a/pkg/sfu/forwarder.go +++ b/pkg/sfu/forwarder.go @@ -129,15 +129,13 @@ func (v VideoTransition) String() string { // ------------------------------------------------------------------- type TranslationParams struct { - shouldDrop bool - isResuming bool - isSwitchingToRequestSpatial bool - isSwitchingToMaxSpatial bool - maxSpatialLayer int32 - rtp *TranslationParamsRTP - codecBytes []byte - ddBytes []byte - marker bool + shouldDrop bool + isResuming bool + isSwitching bool + rtp *TranslationParamsRTP + codecBytes []byte + ddBytes []byte + marker bool } // ------------------------------------------------------------------- @@ -360,12 +358,12 @@ func (f *Forwarder) SeedState(state ForwarderState) { f.refTSOffset = state.RefTSOffset } -func (f *Forwarder) Mute(muted bool) (bool, buffer.VideoLayer) { +func (f *Forwarder) Mute(muted bool) bool { f.lock.Lock() defer f.lock.Unlock() if f.muted == muted { - return false, f.vls.GetMax() + return false } // Do not mute when paused due to bandwidth limitation. @@ -384,7 +382,7 @@ func (f *Forwarder) Mute(muted bool) (bool, buffer.VideoLayer) { // the case of intentional mute. if muted && f.isDeficientLocked() && f.lastAllocation.PauseReason == VideoPauseReasonBandwidth { f.logger.Infow("ignoring forwarder mute, paused due to congestion") - return false, f.vls.GetMax() + return false } f.logger.Debugw("setting forwarder mute", "muted", muted) @@ -395,7 +393,7 @@ func (f *Forwarder) Mute(muted bool) (bool, buffer.VideoLayer) { f.resyncLocked() } - return true, f.vls.GetMax() + return true } func (f *Forwarder) IsMuted() bool { @@ -405,12 +403,12 @@ func (f *Forwarder) IsMuted() bool { return f.muted } -func (f *Forwarder) PubMute(pubMuted bool) (bool, buffer.VideoLayer) { +func (f *Forwarder) PubMute(pubMuted bool) bool { f.lock.Lock() defer f.lock.Unlock() if f.pubMuted == pubMuted { - return false, f.vls.GetMax() + return false } f.logger.Debugw("setting forwarder pub mute", "pubMuted", pubMuted) @@ -432,7 +430,7 @@ func (f *Forwarder) PubMute(pubMuted bool) (bool, buffer.VideoLayer) { } } - return true, f.vls.GetMax() + return true } func (f *Forwarder) IsPubMuted() bool { @@ -449,17 +447,17 @@ func (f *Forwarder) IsAnyMuted() bool { return f.muted || f.pubMuted } -func (f *Forwarder) SetMaxSpatialLayer(spatialLayer int32) (bool, buffer.VideoLayer, buffer.VideoLayer) { +func (f *Forwarder) SetMaxSpatialLayer(spatialLayer int32) (bool, buffer.VideoLayer) { f.lock.Lock() defer f.lock.Unlock() if f.kind == webrtc.RTPCodecTypeAudio { - return false, buffer.InvalidLayer, buffer.InvalidLayer + return false, buffer.InvalidLayer } existingMax := f.vls.GetMax() if spatialLayer == existingMax.Spatial { - return false, existingMax, f.vls.GetCurrent() + return false, existingMax } f.logger.Debugw("setting max spatial layer", "layer", spatialLayer) @@ -467,20 +465,20 @@ func (f *Forwarder) SetMaxSpatialLayer(spatialLayer int32) (bool, buffer.VideoLa f.clearParkedLayer() - return true, f.vls.GetMax(), f.vls.GetCurrent() + return true, f.vls.GetMax() } -func (f *Forwarder) SetMaxTemporalLayer(temporalLayer int32) (bool, buffer.VideoLayer, buffer.VideoLayer) { +func (f *Forwarder) SetMaxTemporalLayer(temporalLayer int32) (bool, buffer.VideoLayer) { f.lock.Lock() defer f.lock.Unlock() if f.kind == webrtc.RTPCodecTypeAudio { - return false, buffer.InvalidLayer, buffer.InvalidLayer + return false, buffer.InvalidLayer } existingMax := f.vls.GetMax() if temporalLayer == existingMax.Temporal { - return false, existingMax, f.vls.GetCurrent() + return false, existingMax } f.logger.Debugw("setting max temporal layer", "layer", temporalLayer) @@ -488,7 +486,7 @@ func (f *Forwarder) SetMaxTemporalLayer(temporalLayer int32) (bool, buffer.Video f.clearParkedLayer() - return true, f.vls.GetMax(), f.vls.GetCurrent() + return true, f.vls.GetMax() } func (f *Forwarder) MaxLayer() buffer.VideoLayer { @@ -512,6 +510,25 @@ func (f *Forwarder) TargetLayer() buffer.VideoLayer { return f.vls.GetTarget() } +func (f *Forwarder) GetMaxSubscribedSpatial() int32 { + f.lock.RLock() + defer f.lock.RUnlock() + + layer := buffer.InvalidLayerSpatial // covers muted case + if !f.muted { + layer = f.vls.GetMax().Spatial + + // If current is higher, mark the current layer as max subscribed layer + // to prevent the current layer from stopping before forwarder switches + // to the new and lower max layer, + if layer < f.vls.GetCurrent().Spatial { + layer = f.vls.GetCurrent().Spatial + } + } + + return layer +} + func (f *Forwarder) isDeficientLocked() bool { return f.lastAllocation.IsDeficient } @@ -1690,9 +1707,7 @@ func (f *Forwarder) getTranslationParamsVideo(extPkt *buffer.ExtPacket, layer in return tp, nil } tp.isResuming = result.IsResuming - tp.isSwitchingToRequestSpatial = result.IsSwitchingToRequestSpatial - tp.isSwitchingToMaxSpatial = result.IsSwitchingToMaxSpatial - tp.maxSpatialLayer = result.MaxSpatialLayer + tp.isSwitching = result.IsSwitching tp.ddBytes = result.DependencyDescriptorExtension tp.marker = result.RTPMarker diff --git a/pkg/sfu/forwarder_test.go b/pkg/sfu/forwarder_test.go index a0e340800..2da1c524b 100644 --- a/pkg/sfu/forwarder_test.go +++ b/pkg/sfu/forwarder_test.go @@ -26,13 +26,13 @@ func newForwarder(codec webrtc.RTPCodecCapability, kind webrtc.RTPCodecType) *Fo func TestForwarderMute(t *testing.T) { f := newForwarder(testutils.TestOpusCodec, webrtc.RTPCodecTypeAudio) require.False(t, f.IsMuted()) - muted, _ := f.Mute(false) + muted := f.Mute(false) require.False(t, muted) // no change in mute state require.False(t, f.IsMuted()) - muted, _ = f.Mute(true) + muted = f.Mute(true) require.True(t, muted) require.True(t, f.IsMuted()) - muted, _ = f.Mute(false) + muted = f.Mute(false) require.True(t, muted) require.False(t, f.IsMuted()) } @@ -45,15 +45,13 @@ func TestForwarderLayersAudio(t *testing.T) { require.Equal(t, buffer.InvalidLayer, f.CurrentLayer()) require.Equal(t, buffer.InvalidLayer, f.TargetLayer()) - changed, maxLayer, currentLayer := f.SetMaxSpatialLayer(1) + changed, maxLayer := f.SetMaxSpatialLayer(1) require.False(t, changed) require.Equal(t, buffer.InvalidLayer, maxLayer) - require.Equal(t, buffer.InvalidLayer, currentLayer) - changed, maxLayer, currentLayer = f.SetMaxTemporalLayer(1) + changed, maxLayer = f.SetMaxTemporalLayer(1) require.False(t, changed) require.Equal(t, buffer.InvalidLayer, maxLayer) - require.Equal(t, buffer.InvalidLayer, currentLayer) require.Equal(t, buffer.InvalidLayer, f.MaxLayer()) } @@ -72,12 +70,11 @@ func TestForwarderLayersVideo(t *testing.T) { Spatial: buffer.DefaultMaxLayerSpatial, Temporal: buffer.DefaultMaxLayerTemporal, } - changed, maxLayer, currentLayer := f.SetMaxSpatialLayer(buffer.DefaultMaxLayerSpatial) + changed, maxLayer := f.SetMaxSpatialLayer(buffer.DefaultMaxLayerSpatial) require.True(t, changed) require.Equal(t, expectedLayers, maxLayer) - require.Equal(t, buffer.InvalidLayer, currentLayer) - changed, maxLayer, currentLayer = f.SetMaxSpatialLayer(buffer.DefaultMaxLayerSpatial - 1) + changed, maxLayer = f.SetMaxSpatialLayer(buffer.DefaultMaxLayerSpatial - 1) require.True(t, changed) expectedLayers = buffer.VideoLayer{ Spatial: buffer.DefaultMaxLayerSpatial - 1, @@ -85,21 +82,18 @@ func TestForwarderLayersVideo(t *testing.T) { } require.Equal(t, expectedLayers, maxLayer) require.Equal(t, expectedLayers, f.MaxLayer()) - require.Equal(t, buffer.InvalidLayer, currentLayer) f.vls.SetCurrent(buffer.VideoLayer{Spatial: 0, Temporal: 1}) - changed, maxLayer, currentLayer = f.SetMaxSpatialLayer(buffer.DefaultMaxLayerSpatial - 1) + changed, maxLayer = f.SetMaxSpatialLayer(buffer.DefaultMaxLayerSpatial - 1) require.False(t, changed) require.Equal(t, expectedLayers, maxLayer) require.Equal(t, expectedLayers, f.MaxLayer()) - require.Equal(t, buffer.VideoLayer{Spatial: 0, Temporal: 1}, currentLayer) - changed, maxLayer, currentLayer = f.SetMaxTemporalLayer(buffer.DefaultMaxLayerTemporal) + changed, maxLayer = f.SetMaxTemporalLayer(buffer.DefaultMaxLayerTemporal) require.False(t, changed) require.Equal(t, expectedLayers, maxLayer) - require.Equal(t, buffer.VideoLayer{Spatial: 0, Temporal: 1}, currentLayer) - changed, maxLayer, currentLayer = f.SetMaxTemporalLayer(buffer.DefaultMaxLayerTemporal - 1) + changed, maxLayer = f.SetMaxTemporalLayer(buffer.DefaultMaxLayerTemporal - 1) require.True(t, changed) expectedLayers = buffer.VideoLayer{ Spatial: buffer.DefaultMaxLayerSpatial - 1, @@ -107,7 +101,6 @@ func TestForwarderLayersVideo(t *testing.T) { } require.Equal(t, expectedLayers, maxLayer) require.Equal(t, expectedLayers, f.MaxLayer()) - require.Equal(t, buffer.VideoLayer{Spatial: 0, Temporal: 1}, currentLayer) } func TestForwarderAllocateOptimal(t *testing.T) { @@ -1404,8 +1397,8 @@ func TestForwarderGetTranslationParamsVideo(t *testing.T) { marshalledVP8, err := expectedVP8.Marshal() require.NoError(t, err) expectedTP = TranslationParams{ - isSwitchingToMaxSpatial: true, - isResuming: true, + isSwitching: true, + isResuming: true, rtp: &TranslationParamsRTP{ snOrdering: SequenceNumberOrderingContiguous, sequenceNumber: 23333, @@ -1716,8 +1709,7 @@ func TestForwarderGetTranslationParamsVideo(t *testing.T) { marshalledVP8, err = expectedVP8.Marshal() require.NoError(t, err) expectedTP = TranslationParams{ - isSwitchingToMaxSpatial: true, - maxSpatialLayer: 1, + isSwitching: true, rtp: &TranslationParamsRTP{ snOrdering: SequenceNumberOrderingContiguous, sequenceNumber: 23339, diff --git a/pkg/sfu/streamallocator/streamallocator.go b/pkg/sfu/streamallocator/streamallocator.go index 475b38e94..f0c17db9e 100644 --- a/pkg/sfu/streamallocator/streamallocator.go +++ b/pkg/sfu/streamallocator/streamallocator.go @@ -548,6 +548,10 @@ func (s *StreamAllocator) postEvent(event Event) { func (s *StreamAllocator) processEvents() { for event := range s.eventCh { + if s.isStopped.Load() { + break + } + s.handleEvent(&event) } diff --git a/pkg/sfu/videolayerselector/dependencydescriptor.go b/pkg/sfu/videolayerselector/dependencydescriptor.go index 7b8c95310..34f8b0b55 100644 --- a/pkg/sfu/videolayerselector/dependencydescriptor.go +++ b/pkg/sfu/videolayerselector/dependencydescriptor.go @@ -204,26 +204,6 @@ func (d *DependencyDescriptor) Select(extPkt *buffer.ExtPacket, _layer int32) (r d.previousActiveDecodeTargetsBitmask = d.activeDecodeTargetsBitmask d.activeDecodeTargetsBitmask = buffer.GetActiveDecodeTargetBitmask(d.currentLayer, ddwdt.DecodeTargets) - - if d.currentLayer.Spatial == d.requestSpatial { - result.IsSwitchingToRequestSpatial = true - } - if d.currentLayer.Spatial == d.maxLayer.Spatial { - result.IsSwitchingToMaxSpatial = true - result.MaxSpatialLayer = d.currentLayer.Spatial - d.logger.Infow( - "reached max layer", - "previous", d.previousLayer, - "current", d.currentLayer, - "previousTarget", d.previousTargetLayer, - "target", d.targetLayer, - "max", d.maxLayer, - "layer", fd.SpatialId, - "req", d.requestSpatial, - "maxSeen", d.maxSeenLayer, - "feed", extPkt.Packet.SSRC, - ) - } } ddExtension := &dede.DependencyDescriptorExtension{ diff --git a/pkg/sfu/videolayerselector/simulcast.go b/pkg/sfu/videolayerselector/simulcast.go index 4d0c9294d..06e0bad72 100644 --- a/pkg/sfu/videolayerselector/simulcast.go +++ b/pkg/sfu/videolayerselector/simulcast.go @@ -35,19 +35,6 @@ func (s *Simulcast) Select(extPkt *buffer.ExtPacket, layer int32) (result VideoL result.IsResuming = true } - if s.currentLayer.Spatial == s.requestSpatial { - result.IsSwitchingToRequestSpatial = true - } - - if s.currentLayer.Spatial >= s.maxLayer.Spatial { - result.IsSwitchingToMaxSpatial = true - result.MaxSpatialLayer = s.currentLayer.Spatial - if reason != "" { - reason += ", " - } - reason += "reached max layer" - } - if reason != "" { s.logger.Infow( reason, diff --git a/pkg/sfu/videolayerselector/videolayerselector.go b/pkg/sfu/videolayerselector/videolayerselector.go index ffbb9f42c..f17d745d1 100644 --- a/pkg/sfu/videolayerselector/videolayerselector.go +++ b/pkg/sfu/videolayerselector/videolayerselector.go @@ -10,9 +10,6 @@ type VideoLayerSelectorResult struct { IsRelevant bool IsSwitching bool IsResuming bool - IsSwitchingToRequestSpatial bool - IsSwitchingToMaxSpatial bool - MaxSpatialLayer int32 RTPMarker bool DependencyDescriptorExtension []byte } diff --git a/pkg/sfu/videolayerselector/vp9.go b/pkg/sfu/videolayerselector/vp9.go index ed1a165b6..508cdf289 100644 --- a/pkg/sfu/videolayerselector/vp9.go +++ b/pkg/sfu/videolayerselector/vp9.go @@ -80,26 +80,6 @@ func (v *VP9) Select(extPkt *buffer.ExtPacket, _layer int32) (result VideoLayerS result.IsResuming = true } - if v.currentLayer.Spatial != v.requestSpatial && updatedLayer.Spatial == v.requestSpatial { - result.IsSwitchingToRequestSpatial = true - } - - if v.currentLayer.Spatial != v.maxLayer.Spatial && updatedLayer.Spatial == v.maxLayer.Spatial { - result.IsSwitchingToMaxSpatial = true - result.MaxSpatialLayer = updatedLayer.Spatial - v.logger.Infow( - "reached max layer", - "current", v.currentLayer, - "updated", updatedLayer, - "target", v.targetLayer, - "max", v.maxLayer, - "layer", extPkt.VideoLayer.Spatial, - "req", v.requestSpatial, - "maxSeen", v.maxSeenLayer, - "feed", extPkt.Packet.SSRC, - ) - } - v.previousLayer = v.currentLayer v.currentLayer = updatedLayer } From e6a47a24a726caca1aab58b45fa4df20c39f821d Mon Sep 17 00:00:00 2001 From: "renovate[bot]" <29139614+renovate[bot]@users.noreply.github.com> Date: Sun, 16 Jul 2023 13:24:06 -0700 Subject: [PATCH 09/18] Update livekit deps (#1869) Generated by renovateBot Co-authored-by: renovate[bot] <29139614+renovate[bot]@users.noreply.github.com> --- go.mod | 4 ++-- go.sum | 8 ++++---- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/go.mod b/go.mod index fad272d8d..34d36beea 100644 --- a/go.mod +++ b/go.mod @@ -17,8 +17,8 @@ require ( github.com/hashicorp/golang-lru/v2 v2.0.4 github.com/jxskiss/base62 v1.1.0 github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1 - github.com/livekit/mediatransportutil v0.0.0-20230612070454-d5299b956135 - github.com/livekit/protocol v1.5.10-0.20230715082801-9d0d6c9e9427 + github.com/livekit/mediatransportutil v0.0.0-20230716190407-fc4944cbc33a + github.com/livekit/protocol v1.5.10 github.com/livekit/psrpc v0.3.2 github.com/mackerelio/go-osstat v0.2.4 github.com/magefile/mage v1.15.0 diff --git a/go.sum b/go.sum index 1c95ae879..417006f39 100644 --- a/go.sum +++ b/go.sum @@ -122,10 +122,10 @@ github.com/lithammer/shortuuid/v4 v4.0.0 h1:QRbbVkfgNippHOS8PXDkti4NaWeyYfcBTHtw github.com/lithammer/shortuuid/v4 v4.0.0/go.mod h1:Zs8puNcrvf2rV9rTH51ZLLcj7ZXqQI3lv67aw4KiB1Y= github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1 h1:jm09419p0lqTkDaKb5iXdynYrzB84ErPPO4LbRASk58= github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ= -github.com/livekit/mediatransportutil v0.0.0-20230612070454-d5299b956135 h1:lWYbsondvqG69czxoACDwaJ/BoyD57BahCo70ZH+m4U= -github.com/livekit/mediatransportutil v0.0.0-20230612070454-d5299b956135/go.mod h1:MRc0zSOSzXuFt0X218SgabzlaKevkvCckPgBEoHYc34= -github.com/livekit/protocol v1.5.10-0.20230715082801-9d0d6c9e9427 h1:NyNdCT8+glCnGGdkgbEO2rpX1iRscwp4HCf1u21Clzo= -github.com/livekit/protocol v1.5.10-0.20230715082801-9d0d6c9e9427/go.mod h1:eRzojAYSPJuNgDHMlvLji/CPauj9hrgvb6rVPUj6MoU= +github.com/livekit/mediatransportutil v0.0.0-20230716190407-fc4944cbc33a h1:JWpPHcMFuw0fP4swE89CfMgeUXiSN5IKvCJL/5HLI3A= +github.com/livekit/mediatransportutil v0.0.0-20230716190407-fc4944cbc33a/go.mod h1:xirUXW8xnLGmfCwUeAv/nj1VGo1OO1BmgxrYP7jK/14= +github.com/livekit/protocol v1.5.10 h1:lnaHMa27cbRkHybi/jvOVuRSaLsho2wCLRjKiC6ce2Y= +github.com/livekit/protocol v1.5.10/go.mod h1:eRzojAYSPJuNgDHMlvLji/CPauj9hrgvb6rVPUj6MoU= github.com/livekit/psrpc v0.3.2 h1:eAaJhASme33gtoBhCRLH9jsnWcdm1tHWf0WzaDk56ew= github.com/livekit/psrpc v0.3.2/go.mod h1:n6JntEg+zT6Ji8InoyTpV7wusPNwGqqtxmHlkNhDN0U= github.com/mackerelio/go-osstat v0.2.4 h1:qxGbdPkFo65PXOb/F/nhDKpF2nGmGaCFDLXoZjJTtUs= From 7dc60bb1bf59394dcd05d5f8ff9046aabd9c4834 Mon Sep 17 00:00:00 2001 From: Paul Wells Date: Sun, 16 Jul 2023 13:40:53 -0700 Subject: [PATCH 10/18] start reading signal messages before session handler finishes (#1883) * start reading signal messages before session handler finishes * fix err scope --- pkg/service/signal.go | 20 ++++++++++++++------ 1 file changed, 14 insertions(+), 6 deletions(-) diff --git a/pkg/service/signal.go b/pkg/service/signal.go index c6907dadb..aa6ab5578 100644 --- a/pkg/service/signal.go +++ b/pkg/service/signal.go @@ -139,9 +139,6 @@ func (r *signalService) RelaySignal(stream psrpc.ServerStream[*rpc.RelaySignalRe "connID", ss.ConnectionId, ) - reqChan := routing.NewDefaultMessageChannel(livekit.ConnectionID(ss.ConnectionId)) - defer reqChan.Close() - sink := routing.NewSignalMessageSink(routing.SignalSinkParams[*rpc.RelaySignalResponse, *rpc.RelaySignalRequest]{ Logger: l, Stream: stream, @@ -149,6 +146,19 @@ func (r *signalService) RelaySignal(stream psrpc.ServerStream[*rpc.RelaySignalRe Writer: signalResponseMessageWriter{}, ConnectionID: livekit.ConnectionID(ss.ConnectionId), }) + reqChan := routing.NewDefaultMessageChannel(livekit.ConnectionID(ss.ConnectionId)) + + go func() { + err := routing.CopySignalStreamToMessageChannel[*rpc.RelaySignalResponse, *rpc.RelaySignalRequest]( + stream, + reqChan, + signalRequestMessageReader{}, + r.config, + ) + l.Infow("signal stream closed", "error", err) + + reqChan.Close() + }() err = r.sessionHandler(ctx, livekit.RoomName(ss.RoomName), *pi, livekit.ConnectionID(ss.ConnectionId), reqChan, sink) if err != nil { @@ -156,9 +166,7 @@ func (r *signalService) RelaySignal(stream psrpc.ServerStream[*rpc.RelaySignalRe return } - err = routing.CopySignalStreamToMessageChannel[*rpc.RelaySignalResponse, *rpc.RelaySignalRequest](stream, reqChan, signalRequestMessageReader{}, r.config) - l.Infow("signal stream closed", "error", err) - + stream.Hijack() return } From 8784449fc6c476f45889189e85478e15139a998e Mon Sep 17 00:00:00 2001 From: Paul Wells Date: Sun, 16 Jul 2023 15:03:47 -0700 Subject: [PATCH 11/18] manually cancel signal relay context (#1884) --- pkg/service/signal.go | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/pkg/service/signal.go b/pkg/service/signal.go index aa6ab5578..2c28465d8 100644 --- a/pkg/service/signal.go +++ b/pkg/service/signal.go @@ -112,12 +112,6 @@ type signalService struct { } func (r *signalService) RelaySignal(stream psrpc.ServerStream[*rpc.RelaySignalResponse, *rpc.RelaySignalRequest]) (err error) { - // copy the context to prevent a race between the session handler closing - // and the delivery of any parting messages from the client. take care to - // copy the incoming rpc headers to avoid dropping any session vars. - ctx, cancel := context.WithCancel(metadata.NewContextWithIncomingHeader(context.Background(), metadata.IncomingHeader(stream.Context()))) - defer cancel() - req, ok := <-stream.Channel() if !ok { return nil @@ -139,6 +133,11 @@ func (r *signalService) RelaySignal(stream psrpc.ServerStream[*rpc.RelaySignalRe "connID", ss.ConnectionId, ) + // copy the context to prevent a race between the session handler closing + // and the delivery of any parting messages from the client. take care to + // copy the incoming rpc headers to avoid dropping any session vars. + ctx, cancel := context.WithCancel(metadata.NewContextWithIncomingHeader(context.Background(), metadata.IncomingHeader(stream.Context()))) + sink := routing.NewSignalMessageSink(routing.SignalSinkParams[*rpc.RelaySignalResponse, *rpc.RelaySignalRequest]{ Logger: l, Stream: stream, @@ -158,11 +157,13 @@ func (r *signalService) RelaySignal(stream psrpc.ServerStream[*rpc.RelaySignalRe l.Infow("signal stream closed", "error", err) reqChan.Close() + cancel() }() err = r.sessionHandler(ctx, livekit.RoomName(ss.RoomName), *pi, livekit.ConnectionID(ss.ConnectionId), reqChan, sink) if err != nil { l.Errorw("could not handle new participant", err) + cancel() return } From 5535916ff27d985430cf95acae1f4007c3676d1e Mon Sep 17 00:00:00 2001 From: Paul Wells Date: Sun, 16 Jul 2023 19:01:53 -0700 Subject: [PATCH 12/18] prevent signal context from closing before room setup finishes (#1885) --- pkg/service/signal.go | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/pkg/service/signal.go b/pkg/service/signal.go index 2c28465d8..299df055a 100644 --- a/pkg/service/signal.go +++ b/pkg/service/signal.go @@ -138,6 +138,12 @@ func (r *signalService) RelaySignal(stream psrpc.ServerStream[*rpc.RelaySignalRe // copy the incoming rpc headers to avoid dropping any session vars. ctx, cancel := context.WithCancel(metadata.NewContextWithIncomingHeader(context.Background(), metadata.IncomingHeader(stream.Context()))) + // wait until room setup finishes to cancel the session context even if the + // stream has closed. this is required to complete setup i/o after room api + // has discarded the temp client. + sessionHandlerDone := make(chan struct{}) + defer close(sessionHandlerDone) + sink := routing.NewSignalMessageSink(routing.SignalSinkParams[*rpc.RelaySignalResponse, *rpc.RelaySignalRequest]{ Logger: l, Stream: stream, @@ -157,13 +163,14 @@ func (r *signalService) RelaySignal(stream psrpc.ServerStream[*rpc.RelaySignalRe l.Infow("signal stream closed", "error", err) reqChan.Close() + + <-sessionHandlerDone cancel() }() err = r.sessionHandler(ctx, livekit.RoomName(ss.RoomName), *pi, livekit.ConnectionID(ss.ConnectionId), reqChan, sink) if err != nil { l.Errorw("could not handle new participant", err) - cancel() return } From 5d1d454a9861226dace41f2ee8b7ea86e361b9f2 Mon Sep 17 00:00:00 2001 From: David Zhao Date: Sun, 16 Jul 2023 20:05:41 -0700 Subject: [PATCH 13/18] Fix missed label arg in logger (#1886) --- pkg/sfu/downtrack.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index e3dd65a04..e21a001f9 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -601,7 +601,7 @@ func (d *DownTrack) maxLayerNotifierWorker() { maxLayerSpatial = d.forwarder.GetMaxSubscribedSpatial() } if onMaxSubscribedLayerChanged := d.getOnMaxLayerChanged(); onMaxSubscribedLayerChanged != nil { - d.logger.Infow("max subscribed layer changed", maxLayerSpatial) + d.logger.Infow("max subscribed layer changed", "maxLayerSpatial", maxLayerSpatial) onMaxSubscribedLayerChanged(d, maxLayerSpatial) } } From 9f3c975b1c415b4bd16310a77c12ed4c0c814c7b Mon Sep 17 00:00:00 2001 From: Paul Wells Date: Sun, 16 Jul 2023 20:24:11 -0700 Subject: [PATCH 14/18] leave signal context open after stream closes (#1887) --- pkg/service/signal.go | 19 +++++-------------- 1 file changed, 5 insertions(+), 14 deletions(-) diff --git a/pkg/service/signal.go b/pkg/service/signal.go index 299df055a..3de03c1ae 100644 --- a/pkg/service/signal.go +++ b/pkg/service/signal.go @@ -133,17 +133,6 @@ func (r *signalService) RelaySignal(stream psrpc.ServerStream[*rpc.RelaySignalRe "connID", ss.ConnectionId, ) - // copy the context to prevent a race between the session handler closing - // and the delivery of any parting messages from the client. take care to - // copy the incoming rpc headers to avoid dropping any session vars. - ctx, cancel := context.WithCancel(metadata.NewContextWithIncomingHeader(context.Background(), metadata.IncomingHeader(stream.Context()))) - - // wait until room setup finishes to cancel the session context even if the - // stream has closed. this is required to complete setup i/o after room api - // has discarded the temp client. - sessionHandlerDone := make(chan struct{}) - defer close(sessionHandlerDone) - sink := routing.NewSignalMessageSink(routing.SignalSinkParams[*rpc.RelaySignalResponse, *rpc.RelaySignalRequest]{ Logger: l, Stream: stream, @@ -163,11 +152,13 @@ func (r *signalService) RelaySignal(stream psrpc.ServerStream[*rpc.RelaySignalRe l.Infow("signal stream closed", "error", err) reqChan.Close() - - <-sessionHandlerDone - cancel() }() + // copy the context to prevent a race between the session handler closing + // and the delivery of any parting messages from the client. take care to + // copy the incoming rpc headers to avoid dropping any session vars. + ctx := metadata.NewContextWithIncomingHeader(context.Background(), metadata.IncomingHeader(stream.Context())) + err = r.sessionHandler(ctx, livekit.RoomName(ss.RoomName), *pi, livekit.ConnectionID(ss.ConnectionId), reqChan, sink) if err != nil { l.Errorw("could not handle new participant", err) From f41b93657e6e9c3663445ea9563adb5a806d4e29 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Tue, 18 Jul 2023 09:14:41 +0530 Subject: [PATCH 15/18] Log a bit more in sender report warp report. (#1888) --- pkg/sfu/buffer/rtpstats.go | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/pkg/sfu/buffer/rtpstats.go b/pkg/sfu/buffer/rtpstats.go index 8bf34b958..1774eb28c 100644 --- a/pkg/sfu/buffer/rtpstats.go +++ b/pkg/sfu/buffer/rtpstats.go @@ -888,8 +888,9 @@ func (r *RTPStats) GetRtcpSenderReport(ssrc uint32, calculatedClockRate uint32) tsCycles++ } nowRTPExt := getExtTS(nowRTP, tsCycles) + var nowRTPExtUsingRate uint64 if calculatedClockRate != 0 { - nowRTPExtUsingRate := r.extStartTS + uint64(float64(calculatedClockRate)*timeSinceFirst.Seconds()) + nowRTPExtUsingRate = r.extStartTS + uint64(float64(calculatedClockRate)*timeSinceFirst.Seconds()) if nowRTPExtUsingRate > nowRTPExt { nowRTPExt = nowRTPExtUsingRate nowRTP = uint32(nowRTPExtUsingRate) @@ -939,6 +940,9 @@ func (r *RTPStats) GetRtcpSenderReport(ssrc uint32, calculatedClockRate uint32) "reportDrift", reportDriftResult.String(), "highestTS", r.highestTS, "highestTime", r.highestTime.String(), + "calculatedClockRate", calculatedClockRate, + "nowRTPExt", nowRTPExt, + "nowRTPExtUsingRate", nowRTPExtUsingRate, ) } From 66de9ff4a0c2c9a3832d61bf3b9383b40bcc3713 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Tue, 18 Jul 2023 23:21:06 +0530 Subject: [PATCH 16/18] Add debug log for RTCP sender report. (#1890) * Add debug log for RTCP sender report. Temporary to collect more data. Hitting scenarios under congestion where the sender report gets off sync. Need some data to pore through and understand and implement changes. * Debugw --- pkg/sfu/buffer/rtpstats.go | 19 +++++++++++++++++++ pkg/sfu/downtrack.go | 6 +++++- pkg/sfu/forwarder.go | 19 +++++++++++++++++++ 3 files changed, 43 insertions(+), 1 deletion(-) diff --git a/pkg/sfu/buffer/rtpstats.go b/pkg/sfu/buffer/rtpstats.go index 1774eb28c..f5bb0a5d9 100644 --- a/pkg/sfu/buffer/rtpstats.go +++ b/pkg/sfu/buffer/rtpstats.go @@ -944,6 +944,25 @@ 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/downtrack.go b/pkg/sfu/downtrack.go index e21a001f9..d943b7c75 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -1165,7 +1165,11 @@ func (d *DownTrack) CreateSenderReport() *rtcp.SenderReport { return nil } - return d.rtpStats.GetRtcpSenderReport(d.ssrc, d.receiver.GetCalculatedClockRate(d.forwarder.CurrentLayer().Spatial)) + clockLayer := d.forwarder.CurrentLayer().Spatial + if clockLayer == buffer.InvalidLayerSpatial { + clockLayer = d.forwarder.GetReferenceLayerSpatial() + } + return d.rtpStats.GetRtcpSenderReport(d.ssrc, d.receiver.GetCalculatedClockRate(clockLayer)) } func (d *DownTrack) writeBlankFrameRTP(duration float32, generation uint32) chan struct{} { diff --git a/pkg/sfu/forwarder.go b/pkg/sfu/forwarder.go index 23a8d694d..79b2b48c1 100644 --- a/pkg/sfu/forwarder.go +++ b/pkg/sfu/forwarder.go @@ -529,6 +529,13 @@ func (f *Forwarder) GetMaxSubscribedSpatial() int32 { return layer } +func (f *Forwarder) GetReferenceLayerSpatial() int32 { + f.lock.RLock() + defer f.lock.RUnlock() + + return f.referenceLayerSpatial +} + func (f *Forwarder) isDeficientLocked() bool { return f.lastAllocation.IsDeficient } @@ -1970,6 +1977,18 @@ done: if !targetLayer.IsValid() { distance += (maxSeenLayer.Temporal + 1) } + // TODO-REMOVE-AFTER-DEBUG + logger.Debugw( + "distance to desired", + "maxSeenLauer", maxSeenLayer, + "availableLayers", availableLayers, + "brs", brs, + "targetLayer", targetLayer, + "maxLayer", maxLayer, + "adjustedMaxLayer", adjustedMaxLayer, + "maxAvailableSpatial", maxAvailableSpatial, + "maxAvailableTemporal", maxAvailableTemporal, + ) return float64(distance) / float64(maxSeenLayer.Temporal+1) } From cf8cf1a87f7ed8a1df34e00d41f2875308e271a8 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Wed, 19 Jul 2023 10:22:51 +0530 Subject: [PATCH 17/18] Forgot to log important bits :-( (#1891) --- pkg/sfu/forwarder.go | 2 ++ 1 file changed, 2 insertions(+) diff --git a/pkg/sfu/forwarder.go b/pkg/sfu/forwarder.go index 79b2b48c1..02f711d06 100644 --- a/pkg/sfu/forwarder.go +++ b/pkg/sfu/forwarder.go @@ -1988,6 +1988,8 @@ done: "adjustedMaxLayer", adjustedMaxLayer, "maxAvailableSpatial", maxAvailableSpatial, "maxAvailableTemporal", maxAvailableTemporal, + "distance", distance, + "distanceToDesired", float64(distance)/float64(maxSeenLayer.Temporal+1), ) return float64(distance) / float64(maxSeenLayer.Temporal+1) From dd995899bf25120aed3e418f173bd745eff71422 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Wed, 19 Jul 2023 12:50:03 +0530 Subject: [PATCH 18/18] Handle extreme case of sender report lagging. (#1892) --- pkg/sfu/buffer/rtpstats.go | 23 +++++++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/pkg/sfu/buffer/rtpstats.go b/pkg/sfu/buffer/rtpstats.go index f5bb0a5d9..075fe72fe 100644 --- a/pkg/sfu/buffer/rtpstats.go +++ b/pkg/sfu/buffer/rtpstats.go @@ -793,7 +793,9 @@ func (r *RTPStats) SetRtcpSenderReportData(srData *RTCPSenderReportData) { r.logger.Infow( "received sender report, out-of-order, resetting", "prevTSExt", r.srNewest.RTPTimestampExt, + "prevNTP", r.srNewest.NTPTimestamp.Time().String(), "currTSExt", srDataCopy.RTPTimestampExt, + "currNTP", srDataCopy.NTPTimestamp.Time().String(), ) r.srFirst = &srDataCopy r.srNewest = &srDataCopy @@ -897,6 +899,27 @@ 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. + // Current report could be behind due to the following + // - Publisher pacing + // - Due to above, report from publisher side is ahead of packet timestamps. + // Note that report will map wall clock to timestamp at capture time and happens before the pacer. + // - Pause/Mute followed by resume, some combination of events that could + // result in this module not having calculated clock rate of publisher side. + // - When the above happens, current will be generated using highestTS which could be behind. + // That could end up behind the last report's timestamp in extreme cases + r.logger.Infow( + "sending sender report, out-of-order, repairing", + "prevTSExt", r.srNewest.RTPTimestampExt, + "prevNTP", r.srNewest.NTPTimestamp.Time().String(), + "currTSExt", nowRTPExt, + "currNTP", nowNTP.Time().String(), + ) + ntpDiffSinceLast := nowNTP.Time().Sub(r.srNewest.NTPTimestamp.Time()) + nowRTPExt = r.srNewest.RTPTimestampExt + uint64(ntpDiffSinceLast.Seconds()*float64(r.params.ClockRate)) + } + // monitor and log RTP timestamp anomalies var ntpDiffSinceLast time.Duration var rtpDiffSinceLast uint32