RTT (#420)

* Consolidating PLI throttle

Use the throttler in `sfu.WebRTCReceiver`.

Does change shape of config object.

* Move PLIThrottleConfig to sfu.WebRTCReceiver

* fix test compile

* Cleaning up unused stuff

* improve readability

* RTT

- Calculate down track RTT using RTCP Receiver report
- Surface it back to the participant
- Participant updates all its published trackes
  (throttled to limit update to once in 5 seconds)
- That propagates to all the upstream sfu.Buffer and the nacker.
  So, we will have RTT throttled NACKs.

* rtt callback

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)
+}

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)

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,

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

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()

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()

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) {

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))
+}

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()

pkg/telemetry/rtt.go

@@ -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
-}