diff --git a/pkg/telemetry/rtt.go b/pkg/telemetry/rtt.go new file mode 100644 index 000000000..f385e13e2 --- /dev/null +++ b/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 +} diff --git a/pkg/telemetry/telemetryserviceinternal.go b/pkg/telemetry/telemetryserviceinternal.go index e1af45f90..a361fb188 100644 --- a/pkg/telemetry/telemetryserviceinternal.go +++ b/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) + } + } } } diff --git a/pkg/telemetry/test/telemetry_service_test.go b/pkg/telemetry/test/telemetry_service_test.go index 839c82151..36219a960 100644 --- a/pkg/telemetry/test/telemetry_service_test.go +++ b/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)