diff --git a/pkg/rtc/mediatrack.go b/pkg/rtc/mediatrack.go index b99ff2ca5..56a583d2f 100644 --- a/pkg/rtc/mediatrack.go +++ b/pkg/rtc/mediatrack.go @@ -236,3 +236,12 @@ func (t *MediaTrack) GetConnectionScore() float32 { return receiver.(*sfu.WebRTCReceiver).GetConnectionScore() } + +func (t *MediaTrack) SetRTT(rtt uint32) { + receiver := t.Receiver() + if receiver == nil { + return + } + + receiver.(*sfu.WebRTCReceiver).SetRTT(rtt) +} diff --git a/pkg/rtc/mediatracksubscriptions.go b/pkg/rtc/mediatracksubscriptions.go index ab01af7dd..0cd1771d5 100644 --- a/pkg/rtc/mediatracksubscriptions.go +++ b/pkg/rtc/mediatracksubscriptions.go @@ -199,6 +199,10 @@ func (t *MediaTrackSubscriptions) AddSubscriber(sub types.LocalParticipant, code t.notifySubscriberMaxQuality(subscriberID, QualityForSpatialLayer(layer)) }) + downTrack.OnRttUpdate(func(_ *sfu.DownTrack, rtt uint32) { + go sub.UpdateRTT(rtt) + }) + downTrack.OnCloseHandler(func() { t.subscribedTracksMu.Lock() delete(t.subscribedTracks, subscriberID) diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index cdbf5fd5e..6fe1f666d 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -33,6 +33,7 @@ const ( lossyDataChannel = "_lossy" reliableDataChannel = "_reliable" sdBatchSize = 20 + rttUpdateInterval = 5 * time.Second ) type pendingTrackInfo struct { @@ -102,6 +103,9 @@ type ParticipantImpl struct { // keep track of other publishers identities that we are subscribed to subscribedTo sync.Map // livekit.ParticipantID => struct{} + rttUpdatedAt time.Time + lastRTT uint32 + lock sync.RWMutex once sync.Once updateLock sync.Mutex @@ -131,6 +135,7 @@ func NewParticipant(params ParticipantParams, perms *livekit.ParticipantPermissi subscribedTracksSettings: make(map[livekit.TrackID]*livekit.UpdateTrackSettings), disallowedSubscriptions: make(map[livekit.TrackID]livekit.ParticipantID), connectedAt: time.Now(), + rttUpdatedAt: time.Now(), version: params.InitialVersion, } p.migrateState.Store(types.MigrateStateInit) @@ -932,6 +937,22 @@ func (p *ParticipantImpl) SubscriptionPermissionUpdate(publisherID livekit.Parti } } +func (p *ParticipantImpl) UpdateRTT(rtt uint32) { + now := time.Now() + p.lock.Lock() + if now.Sub(p.rttUpdatedAt) < rttUpdateInterval || p.lastRTT == rtt { + p.lock.Unlock() + return + } + p.rttUpdatedAt = now + p.lastRTT = rtt + p.lock.Unlock() + + for _, pt := range p.GetPublishedTracks() { + pt.(types.LocalMediaTrack).SetRTT(rtt) + } +} + func (p *ParticipantImpl) setupUpTrackManager() { p.UpTrackManager = NewUpTrackManager(UpTrackManagerParams{ SID: p.params.SID, diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index bb9cd87bd..46fbd7524 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -153,6 +153,8 @@ type LocalParticipant interface { MigrateState() MigrateState AddMigratedTrack(cid string, ti *livekit.TrackInfo) SetPreviousAnswer(previous *webrtc.SessionDescription) + + UpdateRTT(rtt uint32) } // Room is a container of participants, and can provide room-level actions @@ -215,6 +217,8 @@ type LocalMediaTrack interface { GetAudioLevel() (level uint8, active bool) GetConnectionScore() float32 + + SetRTT(rtt uint32) } // MediaTrack is the main interface representing a track published to the room diff --git a/pkg/rtc/types/typesfakes/fake_local_media_track.go b/pkg/rtc/types/typesfakes/fake_local_media_track.go index 085ca9eb9..f98ff878b 100644 --- a/pkg/rtc/types/typesfakes/fake_local_media_track.go +++ b/pkg/rtc/types/typesfakes/fake_local_media_track.go @@ -199,6 +199,11 @@ type FakeLocalMediaTrack struct { setMutedArgsForCall []struct { arg1 bool } + SetRTTStub func(uint32) + setRTTMutex sync.RWMutex + setRTTArgsForCall []struct { + arg1 uint32 + } SignalCidStub func() string signalCidMutex sync.RWMutex signalCidArgsForCall []struct { @@ -1261,6 +1266,38 @@ func (fake *FakeLocalMediaTrack) SetMutedArgsForCall(i int) bool { return argsForCall.arg1 } +func (fake *FakeLocalMediaTrack) SetRTT(arg1 uint32) { + fake.setRTTMutex.Lock() + fake.setRTTArgsForCall = append(fake.setRTTArgsForCall, struct { + arg1 uint32 + }{arg1}) + stub := fake.SetRTTStub + fake.recordInvocation("SetRTT", []interface{}{arg1}) + fake.setRTTMutex.Unlock() + if stub != nil { + fake.SetRTTStub(arg1) + } +} + +func (fake *FakeLocalMediaTrack) SetRTTCallCount() int { + fake.setRTTMutex.RLock() + defer fake.setRTTMutex.RUnlock() + return len(fake.setRTTArgsForCall) +} + +func (fake *FakeLocalMediaTrack) SetRTTCalls(stub func(uint32)) { + fake.setRTTMutex.Lock() + defer fake.setRTTMutex.Unlock() + fake.SetRTTStub = stub +} + +func (fake *FakeLocalMediaTrack) SetRTTArgsForCall(i int) uint32 { + fake.setRTTMutex.RLock() + defer fake.setRTTMutex.RUnlock() + argsForCall := fake.setRTTArgsForCall[i] + return argsForCall.arg1 +} + func (fake *FakeLocalMediaTrack) SignalCid() string { fake.signalCidMutex.Lock() ret, specificReturn := fake.signalCidReturnsOnCall[len(fake.signalCidArgsForCall)] @@ -1502,6 +1539,8 @@ func (fake *FakeLocalMediaTrack) Invocations() map[string][][]interface{} { defer fake.sdpCidMutex.RUnlock() fake.setMutedMutex.RLock() defer fake.setMutedMutex.RUnlock() + fake.setRTTMutex.RLock() + defer fake.setRTTMutex.RUnlock() fake.signalCidMutex.RLock() defer fake.signalCidMutex.RUnlock() fake.sourceMutex.RLock() diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index cb1fb530c..cb59677d2 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -550,6 +550,11 @@ type FakeLocalParticipant struct { updateMediaLossReturnsOnCall map[int]struct { result1 error } + UpdateRTTStub func(uint32) + updateRTTMutex sync.RWMutex + updateRTTArgsForCall []struct { + arg1 uint32 + } UpdateSubscribedQualityStub func(string, livekit.TrackID, livekit.VideoQuality) error updateSubscribedQualityMutex sync.RWMutex updateSubscribedQualityArgsForCall []struct { @@ -3532,6 +3537,38 @@ func (fake *FakeLocalParticipant) UpdateMediaLossReturnsOnCall(i int, result1 er }{result1} } +func (fake *FakeLocalParticipant) UpdateRTT(arg1 uint32) { + fake.updateRTTMutex.Lock() + fake.updateRTTArgsForCall = append(fake.updateRTTArgsForCall, struct { + arg1 uint32 + }{arg1}) + stub := fake.UpdateRTTStub + fake.recordInvocation("UpdateRTT", []interface{}{arg1}) + fake.updateRTTMutex.Unlock() + if stub != nil { + fake.UpdateRTTStub(arg1) + } +} + +func (fake *FakeLocalParticipant) UpdateRTTCallCount() int { + fake.updateRTTMutex.RLock() + defer fake.updateRTTMutex.RUnlock() + return len(fake.updateRTTArgsForCall) +} + +func (fake *FakeLocalParticipant) UpdateRTTCalls(stub func(uint32)) { + fake.updateRTTMutex.Lock() + defer fake.updateRTTMutex.Unlock() + fake.UpdateRTTStub = stub +} + +func (fake *FakeLocalParticipant) UpdateRTTArgsForCall(i int) uint32 { + fake.updateRTTMutex.RLock() + defer fake.updateRTTMutex.RUnlock() + argsForCall := fake.updateRTTArgsForCall[i] + return argsForCall.arg1 +} + func (fake *FakeLocalParticipant) UpdateSubscribedQuality(arg1 string, arg2 livekit.TrackID, arg3 livekit.VideoQuality) error { fake.updateSubscribedQualityMutex.Lock() ret, specificReturn := fake.updateSubscribedQualityReturnsOnCall[len(fake.updateSubscribedQualityArgsForCall)] @@ -3905,6 +3942,8 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} { defer fake.toProtoMutex.RUnlock() fake.updateMediaLossMutex.RLock() defer fake.updateMediaLossMutex.RUnlock() + fake.updateRTTMutex.RLock() + defer fake.updateRTTMutex.RUnlock() fake.updateSubscribedQualityMutex.RLock() defer fake.updateSubscribedQualityMutex.RUnlock() fake.updateSubscribedTrackSettingsMutex.RLock() diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index d1c0040a0..e41ad9d30 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -125,6 +125,9 @@ type DownTrack struct { // when max subscribed layer changes onMaxLayerChanged func(dt *DownTrack, layer int32) + + // update rtt + onRttUpdate func(dt *DownTrack, rtt uint32) } // NewDownTrack returns a DownTrack. @@ -588,6 +591,10 @@ func (d *DownTrack) OnStatsUpdate(fn func(dt *DownTrack, stat *livekit.Analytics d.onStatsUpdate = fn } +func (d *DownTrack) OnRttUpdate(fn func(dt *DownTrack, rtt uint32)) { + d.onRttUpdate = fn +} + func (d *DownTrack) OnMaxLayerChanged(fn func(dt *DownTrack, layer int32)) { d.onMaxLayerChanged = fn @@ -850,6 +857,8 @@ func (d *DownTrack) handleRTCP(bytes []byte) { } } + rttToReport := uint32(0) + var numNACKs uint32 var numPLIs uint32 var numFIRs uint32 @@ -882,10 +891,16 @@ func (d *DownTrack) handleRTCP(bytes []byte) { d.statsLock.Lock() d.stats.TotalPacketsLost = r.TotalLost - // RAJA-TODO - calculate RTT and update + + rtt := getRttMs(&r) + if rtt != d.stats.RTT { + rttToReport = rtt + } + d.stats.RTT = rtt + d.stats.Jitter = float64(r.Jitter) - d.connectionStats.UpdateWindow(r.SSRC, r.LastSequenceNumber, r.TotalLost, 0, r.Jitter) + d.connectionStats.UpdateWindow(r.SSRC, r.LastSequenceNumber, r.TotalLost, rtt, r.Jitter) d.statsLock.Unlock() } if len(rr.Reports) > 0 { @@ -916,6 +931,10 @@ func (d *DownTrack) handleRTCP(bytes []byte) { d.stats.TotalPLIs += numPLIs d.stats.TotalFIRs += numFIRs d.statsLock.Unlock() + + if rttToReport != 0 && d.onRttUpdate != nil { + d.onRttUpdate(d, rttToReport) + } } func (d *DownTrack) retransmitPackets(nackedPackets []packetMeta) { diff --git a/pkg/sfu/helpers.go b/pkg/sfu/helpers.go index 022eb4baf..fde31c33c 100644 --- a/pkg/sfu/helpers.go +++ b/pkg/sfu/helpers.go @@ -1,9 +1,11 @@ package sfu import ( + "math" "strings" "time" + "github.com/pion/rtcp" "github.com/pion/webrtc/v3" ) @@ -64,3 +66,16 @@ func toNtpTime(t time.Time) ntpTime { } return ntpTime(sec<<32 | frac) } + +func getRttMs(report *rtcp.ReceptionReport) uint32 { + if report.LastSenderReport == 0 { + return 0 + } + + // RTT calculation reference: https://datatracker.ietf.org/doc/html/rfc3550#section-6.4.1 + + // middle 32-bits of current NTP time + now := uint32(toNtpTime(time.Now()) >> 16) + ntpDiff := now - report.LastSenderReport - report.Delay + return uint32(math.Ceil(float64(ntpDiff) * 1000.0 / 65536.0)) +} diff --git a/pkg/sfu/receiver.go b/pkg/sfu/receiver.go index dce372c47..8a6dc5103 100644 --- a/pkg/sfu/receiver.go +++ b/pkg/sfu/receiver.go @@ -191,6 +191,10 @@ func (w *WebRTCReceiver) SetRTT(rtt uint32) { w.bufferMu.RLock() for _, buffer := range w.buffers { + if buffer == nil { + continue + } + buffer.SetRTT(rtt) } w.bufferMu.RUnlock() diff --git a/pkg/telemetry/rtt.go b/pkg/telemetry/rtt.go deleted file mode 100644 index 387f88be1..000000000 --- a/pkg/telemetry/rtt.go +++ /dev/null @@ -1,106 +0,0 @@ -package telemetry - -import ( - "math" - "time" - - "github.com/pion/rtcp" -) - -const ( - NanoSecondsInMicroSecond = 1000 - MicroSecondsInSecond int64 = 1000000 - NtpJan1970Sec int64 = 2208988800 - MaxCompactNtp uint32 = 0xFFFFFFFF - CompactNtpInSecond int64 = 0x10000 - NtpInSecond int64 = 1 << 32 -) - -// SaturatedUsToCompactNtp convert us to rtp compact ntptime field (from webrtc) -func SaturatedUsToCompactNtp(us int64) uint32 { - if us <= 0 { - return 0 - } - if us >= int64(MaxCompactNtp)*MicroSecondsInSecond/CompactNtpInSecond { - return MaxCompactNtp - } - // To convert to compact ntp need to divide by 1e6 to get seconds, - // then multiply by 0x10000 to get the final result. - // To avoid float operations, multiplication and division swapped. - half := (MicroSecondsInSecond - 1) / 2 - quot := us * CompactNtpInSecond / MicroSecondsInSecond - remain := us * CompactNtpInSecond % MicroSecondsInSecond - if remain > half { - return uint32(quot + 1) - } - return uint32(quot) -} - -// TimeMicrosToNtp convert us to NtpTime (from webrtc) -func TimeMicrosToNtp(us int64) *NtpTime { - timeNtpUs := us + NtpJan1970Sec*MicroSecondsInSecond - - // Convert seconds to uint32 through uint64 for well-defined cast. - // Wrap around (will happen in 2036) is expected for ntp time. - ntpSeconds := uint32(timeNtpUs / MicroSecondsInSecond) - - // Scale fractions of the second to ntp resolution. - usFrac := timeNtpUs % MicroSecondsInSecond - ntpFrac := usFrac * NtpInSecond / MicroSecondsInSecond - return &NtpTime{ntpSeconds, uint32(ntpFrac)} -} - -func NtpToTimeMicros(val NtpTime) int64 { - sec := int64(val.sec) - NtpJan1970Sec - micro := int64(val.frac) * MicroSecondsInSecond / NtpInSecond - return sec*MicroSecondsInSecond + micro -} - -func NtpCompactToTime(val uint32) NtpTime { - sec := (val & 0xFFFF0000) >> 16 - frac := (val & 0x0000FFFF) << 16 - return NtpTime{sec, frac} -} - -// NtpTime represents ntp time in rtp -type NtpTime struct { - sec, frac uint32 -} - -// Set set ntptime from rtp's ntptime field (uint64) -func (n *NtpTime) Set(val uint64) { - n.sec = uint32(val >> 32) - n.frac = uint32(val) -} - -// Compact convert ntptime to rtp compact ntptime field (middle 32 bit, uint32) -func (n *NtpTime) Compact() uint32 { - return (n.sec << 16) | (n.frac >> 16) -} - -// FromCompact set ntp fields from compact ntp filed(32bit) -func (n *NtpTime) FromCompact(val uint32) { - n.sec = (val & 0xFFFF0000) >> 16 - n.frac = (val & 0x0000FFFF) << 16 -} - -// Ntp return rtp ntptime field -func (n *NtpTime) Ntp() uint64 { - return (uint64(n.sec) << 32) | (uint64(n.frac)) -} - -func GetRttMs(packet *rtcp.ReceiverReport) int64 { - var rttMs int64 = -1 - nowus := time.Now().UnixNano() / NanoSecondsInMicroSecond - nowntp := TimeMicrosToNtp(nowus).Compact() - for _, v := range packet.Reports { - if v.LastSenderReport != 0 { - currUs := NtpToTimeMicros(NtpCompactToTime(nowntp)) - lastUs := NtpToTimeMicros(NtpCompactToTime(v.LastSenderReport)) - ms := int64(math.Round((float64(currUs-lastUs) / 1000) - (float64(v.Delay) * 1000 / 65536))) - rttMs = ms - break - } - } - return rttMs -}