Implement calculating of RTT for down tracks (#300)

* Implement calculating of RTT for down tracks
* Fix code style issues

pkg/telemetry/rtt.go

@@ -0,0 +1,106 @@
+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(time_us int64) *NtpTime {
+	timeNtpUs := time_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
+}

pkg/telemetry/telemetryserviceinternal.go

@@ -66,14 +66,17 @@ func (t *telemetryServiceInternal) HandleRTCP(streamType livekit.StreamType, par
 			stats.FirCount++
 		case *rtcp.ReceiverReport:
 			for _, rr := range pkt.Reports {
-				if delay := uint64(rr.Delay); delay > stats.Delay {
-					stats.Delay = delay
-				}
 				if jitter := float64(rr.Jitter); jitter > stats.Jitter {
 					stats.Jitter = jitter
 				}
 				stats.PacketLost += uint64(rr.TotalLost)
 			}
+			if streamType == livekit.StreamType_DOWNSTREAM {
+				rtt := GetRttMs(pkt)
+				if rtt >= 0 {
+					stats.Delay = uint64(rtt)
+				}
+			}
 		}
 	}
 

pkg/telemetry/test/telemetry_service_test.go

@@ -137,8 +137,8 @@ func Test_OnDownStreamRTCP(t *testing.T) {
 		&rtcp.FullIntraRequest{},
 		&rtcp.ReceiverReport{
 			Reports: []rtcp.ReceptionReport{
-				{Delay: 1, Jitter: 5, TotalLost: 3},
-				{Delay: 4, Jitter: 2, TotalLost: 4},
+				{Jitter: 5, TotalLost: 3},
+				{Jitter: 2, TotalLost: 4},
 			},
 		},
 	}
@@ -155,7 +155,7 @@ func Test_OnDownStreamRTCP(t *testing.T) {
 	require.Equal(t, 1, int(stats[0].NackCount))
 	require.Equal(t, 1, int(stats[0].PliCount))
 	require.Equal(t, 1, int(stats[0].FirCount))
-	require.Equal(t, 4, int(stats[0].Delay))      // max of delay, see list of rtcp.ReceptionReport above
+	require.Equal(t, 0, int(stats[0].Delay))      // TODO: test for RTT
 	require.Equal(t, 5, int(stats[0].Jitter))     // max of jitter, see list of rtcp.ReceptionReport above
 	require.Equal(t, 7, int(stats[0].PacketLost)) // sum of lost packets, see list of rtcp.ReceptionReport above
 	require.Equal(t, trackID, stats[0].TrackId)
@@ -222,8 +222,8 @@ func Test_OnUpstreamRTCP(t *testing.T) {
 		&rtcp.FullIntraRequest{},
 		&rtcp.ReceiverReport{
 			Reports: []rtcp.ReceptionReport{
-				{Delay: 1, Jitter: 5, TotalLost: 3},
-				{Delay: 4, Jitter: 2, TotalLost: 4},
+				{Jitter: 5, TotalLost: 3},
+				{Jitter: 2, TotalLost: 4},
 			},
 		},
 	}
@@ -246,7 +246,7 @@ func Test_OnUpstreamRTCP(t *testing.T) {
 	require.Equal(t, 1, int(stats[0].NackCount))
 	require.Equal(t, 1, int(stats[0].PliCount))
 	require.Equal(t, 1, int(stats[0].FirCount))
-	require.Equal(t, 4, int(stats[0].Delay))      // max of delay, see list of rtcp.ReceptionReport above
+	require.Equal(t, 0, int(stats[0].Delay))      // TODO: test for RTT
 	require.Equal(t, 5, int(stats[0].Jitter))     // max of jitter, see list of rtcp.ReceptionReport above
 	require.Equal(t, 7, int(stats[0].PacketLost)) // sum of lost packets, see list of rtcp.ReceptionReport above
 	require.Equal(t, trackID, stats[0].TrackId)