diff --git a/go.mod b/go.mod index 57516561d..34d36beea 100644 --- a/go.mod +++ b/go.mod @@ -17,9 +17,9 @@ 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.9 - github.com/livekit/psrpc v0.3.1 + 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 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..417006f39 100644 --- a/go.sum +++ b/go.sum @@ -122,12 +122,12 @@ 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.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/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= 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= 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 dcd5a6b1c..7602659ac 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -896,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 { 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/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 diff --git a/pkg/service/signal.go b/pkg/service/signal.go index c6907dadb..3de03c1ae 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,9 +133,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 +140,24 @@ 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() + }() + + // 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 { @@ -156,9 +165,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 } diff --git a/pkg/sfu/buffer/rtpstats.go b/pkg/sfu/buffer/rtpstats.go index c21938d36..075fe72fe 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,49 @@ 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, + "prevNTP", r.srNewest.NTPTimestamp.Time().String(), + "currTSExt", srDataCopy.RTPTimestampExt, + "currNTP", srDataCopy.NTPTimestamp.Time().String(), + ) + 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 +851,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,24 +884,42 @@ 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()) + var nowRTPExtUsingRate uint64 + if calculatedClockRate != 0 { + nowRTPExtUsingRate = r.extStartTS + uint64(float64(calculatedClockRate)*timeSinceFirst.Seconds()) if nowRTPExtUsingRate > nowRTPExt { nowRTPExt = nowRTPExtUsingRate nowRTP = uint32(nowRTPExtUsingRate) } } + 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 @@ -938,6 +963,28 @@ func (r *RTPStats) GetRtcpSenderReport(ssrc uint32, srFirst *RTCPSenderReportDat "reportDrift", reportDriftResult.String(), "highestTS", r.highestTS, "highestTime", r.highestTime.String(), + "calculatedClockRate", calculatedClockRate, + "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, ) } diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index f0463a1e0..d943b7c75 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", 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() } @@ -1102,8 +1165,11 @@ func (d *DownTrack) CreateSenderReport() *rtcp.SenderReport { return nil } - srFirst, srNewest := d.receiver.GetRTCPSenderReportData(d.forwarder.GetReferenceLayerSpatial()) - return d.rtpStats.GetRtcpSenderReport(d.ssrc, srFirst, srNewest) + 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{} { @@ -1356,8 +1422,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) } } } @@ -1542,7 +1608,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) } @@ -1745,15 +1811,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 8bb21c387..02f711d06 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 ) // ------------------------------------------------------------------- @@ -125,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 } // ------------------------------------------------------------------- @@ -171,7 +173,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 +204,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, @@ -356,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. @@ -380,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) @@ -391,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 { @@ -401,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) @@ -428,7 +430,7 @@ func (f *Forwarder) PubMute(pubMuted bool) (bool, buffer.VideoLayer) { } } - return true, f.vls.GetMax() + return true } func (f *Forwarder) IsPubMuted() bool { @@ -445,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) @@ -463,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) @@ -484,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 { @@ -508,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) GetReferenceLayerSpatial() int32 { f.lock.RLock() defer f.lock.RUnlock() @@ -1477,103 +1498,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 @@ -1620,9 +1714,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 @@ -1696,7 +1788,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 +1833,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 } @@ -1885,47 +1977,20 @@ 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, + "distance", distance, + "distanceToDesired", float64(distance)/float64(maxSeenLayer.Temporal+1), + ) 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/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/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/streamallocator/streamallocator.go b/pkg/sfu/streamallocator/streamallocator.go index 207118f76..b98f20d66 100644 --- a/pkg/sfu/streamallocator/streamallocator.go +++ b/pkg/sfu/streamallocator/streamallocator.go @@ -550,6 +550,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/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 { 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 ebb80d113..06e0bad72 100644 --- a/pkg/sfu/videolayerselector/simulcast.go +++ b/pkg/sfu/videolayerselector/simulcast.go @@ -26,25 +26,15 @@ 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 } - 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, @@ -70,6 +60,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 +68,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 +102,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 +116,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 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 }