From 4e24df4200a0d80f9d51694492490f5245e3a666 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Sat, 12 Mar 2022 11:21:05 +0530 Subject: [PATCH] Move RTPStats -> OSS (#506) --- pkg/rtc/datastats.go | 88 +++++ pkg/rtc/datastats_test.go | 36 ++ pkg/rtc/rtpstats.go | 727 ++++++++++++++++++++++++++++++++++++++ pkg/rtc/rtpstats_test.go | 62 ++++ 4 files changed, 913 insertions(+) create mode 100644 pkg/rtc/datastats.go create mode 100644 pkg/rtc/datastats_test.go create mode 100644 pkg/rtc/rtpstats.go create mode 100644 pkg/rtc/rtpstats_test.go diff --git a/pkg/rtc/datastats.go b/pkg/rtc/datastats.go new file mode 100644 index 000000000..7794a607d --- /dev/null +++ b/pkg/rtc/datastats.go @@ -0,0 +1,88 @@ +package rtc + +import ( + "sync" + "time" + + "github.com/livekit/protocol/livekit" + "google.golang.org/protobuf/types/known/timestamppb" +) + +type DataStatsParam struct { + WindowDuration time.Duration +} + +type DataStats struct { + params DataStatsParam + lock sync.RWMutex + totalBytes int64 + startTime time.Time + endTime time.Time + windowStart int64 + windowBytes int64 +} + +func NewDataStats(params DataStatsParam) *DataStats { + return &DataStats{ + params: params, + startTime: time.Now(), + windowStart: time.Now().UnixNano(), + } +} + +func (s *DataStats) Update(bytes int, time int64) { + s.lock.Lock() + defer s.lock.Unlock() + s.totalBytes += int64(bytes) + + if s.params.WindowDuration > 0 && time-s.windowStart > s.params.WindowDuration.Nanoseconds() { + s.windowBytes = 0 + s.windowStart = time + } + s.windowBytes += int64(bytes) +} + +func (s *DataStats) ToProtoActive() *livekit.RTPStats { + if s.params.WindowDuration == 0 { + return &livekit.RTPStats{} + } + s.lock.RLock() + defer s.lock.RUnlock() + now := time.Now().UnixNano() + duration := now - s.windowStart + if duration > s.params.WindowDuration.Nanoseconds() { + return &livekit.RTPStats{} + } + + return &livekit.RTPStats{ + StartTime: timestamppb.New(time.Unix(s.windowStart/1e9, s.windowStart%1e9)), + EndTime: timestamppb.New(time.Now()), + Duration: float64(duration / 1e9), + Bytes: uint64(s.windowBytes), + Bitrate: float64(s.windowBytes) * 8 / float64(duration) / 1e9, + } + +} + +func (s *DataStats) Stop() { + s.lock.Lock() + s.endTime = time.Now() + s.lock.Unlock() +} + +func (s *DataStats) ToProtoAggregateOnly() *livekit.RTPStats { + s.lock.RLock() + defer s.lock.RUnlock() + + end := s.endTime + if end.IsZero() { + end = time.Now() + } + return &livekit.RTPStats{ + StartTime: timestamppb.New(s.startTime), + EndTime: timestamppb.New(end), + Duration: end.Sub(s.startTime).Seconds(), + Bytes: uint64(s.windowBytes), + Bitrate: float64(s.windowBytes) * 8 / float64(end.Sub(s.startTime).Seconds()), + } +} diff --git a/pkg/rtc/datastats_test.go b/pkg/rtc/datastats_test.go new file mode 100644 index 000000000..ff0bd69d0 --- /dev/null +++ b/pkg/rtc/datastats_test.go @@ -0,0 +1,36 @@ +package rtc + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestDataStats(t *testing.T) { + stats := NewDataStats(DataStatsParam{WindowDuration: time.Second}) + + time.Sleep(time.Millisecond) + r := stats.ToProtoAggregateOnly() + require.Equal(t, r.StartTime.AsTime().UnixNano(), stats.startTime.UnixNano()) + require.NotZero(t, r.EndTime) + require.NotZero(t, r.Duration) + r.StartTime = nil + r.EndTime = nil + r.Duration = 0 + require.Zero(t, *r) + + stats.Update(100, time.Now().UnixNano()) + r = stats.ToProtoActive() + require.EqualValues(t, 100, r.Bytes) + require.NotZero(t, r.Bitrate) + + // wait for window duration + time.Sleep(time.Second) + r = stats.ToProtoActive() + require.Zero(t, *r) + stats.Stop() + r = stats.ToProtoAggregateOnly() + require.EqualValues(t, 100, r.Bytes) + require.NotZero(t, r.Bitrate) +} diff --git a/pkg/rtc/rtpstats.go b/pkg/rtc/rtpstats.go new file mode 100644 index 000000000..ea639941e --- /dev/null +++ b/pkg/rtc/rtpstats.go @@ -0,0 +1,727 @@ +package rtc + +import ( + "fmt" + "sync" + "time" + + "github.com/livekit/protocol/livekit" + "github.com/pion/rtp" + "google.golang.org/protobuf/types/known/timestamppb" +) + +const ( + GapHistogramNumBins = 101 +) + +func getPos(sn uint16) (uint16, uint16) { + return sn >> 6, sn & 0x3f +} + +type RTPStatsParams struct { + ClockRate uint32 + WindowDuration time.Duration +} + +type RTPStats struct { + params RTPStatsParams + + lock sync.RWMutex + + aggregateWindow *Window + + activeWindowStartTime time.Time + activeWindow *Window + windows []*Window +} + +func NewRTPStats(params RTPStatsParams) *RTPStats { + return &RTPStats{ + params: params, + } +} + +func (r *RTPStats) Stop() { + r.lock.Lock() + defer r.lock.Unlock() + + now := time.Now() + + // close active window + if r.activeWindow != nil { + r.activeWindow.Close(now) + r.windows = append(r.windows, r.activeWindow) + r.activeWindow = nil + } + + if r.aggregateWindow != nil { + r.aggregateWindow.Close(now) + } +} + +func (r *RTPStats) Update(rtp *rtp.Packet, packetTime int64) { + r.lock.Lock() + defer r.lock.Unlock() + + now := time.Now() + if aw := r.getAggregateWindow(now); aw != nil { + aw.Update(rtp, packetTime) + } + + if aw := r.getActiveWindow(now); aw != nil { + aw.Update(rtp, packetTime) + } +} + +func (r *RTPStats) UpdateNack(nackCount int, nackMissCount int) { + r.lock.Lock() + defer r.lock.Unlock() + + now := time.Now() + if aw := r.getAggregateWindow(now); aw != nil { + aw.UpdateNack(nackCount, nackMissCount) + } + + if aw := r.getActiveWindow(now); aw != nil { + aw.UpdateNack(nackCount, nackMissCount) + } +} + +func (r *RTPStats) UpdatePli() { + r.lock.Lock() + defer r.lock.Unlock() + + now := time.Now() + if aw := r.getAggregateWindow(now); aw != nil { + aw.UpdatePli(now) + } + + if aw := r.getActiveWindow(now); aw != nil { + aw.UpdatePli(now) + } +} + +func (r *RTPStats) TimeSinceLastPliAggregate() int64 { + r.lock.RLock() + defer r.lock.RUnlock() + + now := time.Now() + if aw := r.getAggregateWindow(now); aw != nil { + aw.TimeSinceLastPli(now) + } + + return 0 +} + +func (r *RTPStats) TimeSinceLastPliActive() int64 { + r.lock.RLock() + defer r.lock.RUnlock() + + now := time.Now() + if aw := r.getActiveWindow(now); aw != nil { + aw.TimeSinceLastPli(now) + } + + return 0 +} + +func (r *RTPStats) UpdateFir() { + r.lock.Lock() + defer r.lock.Unlock() + + now := time.Now() + if aw := r.getAggregateWindow(now); aw != nil { + aw.UpdateFir(now) + } + + if aw := r.getActiveWindow(now); aw != nil { + aw.UpdateFir(now) + } +} + +func (r *RTPStats) ToProtoActive() *livekit.RTPStats { + r.lock.RLock() + defer r.lock.RUnlock() + + if r.activeWindow == nil { + return &livekit.RTPStats{} + } + + return r.activeWindow.ToProto() +} + +func (r *RTPStats) ToString() string { + r.lock.RLock() + defer r.lock.RUnlock() + + var str string + if r.aggregateWindow != nil { + str += "a:[" + str += r.aggregateWindow.ToString() + str += "]" + } + + if len(r.windows) > 1 { + // report only if more than one, else the aggregate window captures it + str += ", w:[" + for idx, window := range r.windows { + if idx != 0 { + str += ", " + } + str += fmt.Sprintf("%d:[%s]", idx, window.ToString()) + } + str += "]" + } + + return str +} + +func (r *RTPStats) ToStringAggregateOnly() string { + r.lock.RLock() + defer r.lock.RUnlock() + + var str string + if r.aggregateWindow != nil { + str += r.aggregateWindow.ToString() + } + + return str +} + +func (r *RTPStats) ToProtoAggregateOnly() *livekit.RTPStats { + r.lock.RLock() + defer r.lock.RUnlock() + + if r.aggregateWindow == nil { + return nil + } + + return r.aggregateWindow.ToProto() +} + +func (r *RTPStats) getAggregateWindow(now time.Time) *Window { + if r.aggregateWindow == nil { + r.aggregateWindow = newWindow(now, r.params.ClockRate) + } + + return r.aggregateWindow +} + +func (r *RTPStats) getActiveWindow(now time.Time) *Window { + if r.params.WindowDuration != 0 && (r.activeWindowStartTime.IsZero() || time.Since(r.activeWindowStartTime) > r.params.WindowDuration) { + // close active window if any + if r.activeWindow != nil { + r.activeWindow.Close(now) + r.windows = append(r.windows, r.activeWindow) + r.activeWindow = nil + } + + // start a new window + if r.activeWindow == nil { + r.activeWindowStartTime = now + r.activeWindow = newWindow(r.activeWindowStartTime, r.params.ClockRate) + } + } + + return r.activeWindow +} + +// ------------------------------------------------------------ + +type Stats struct { + clockRate uint32 + + initialized bool + highestSN uint16 + extStartSN uint32 + cycles uint16 + + lastTransit uint32 + + bytes uint64 + bytesDuplicate uint64 + bytesPadding uint64 + packetsDuplicate uint32 + packetsPadding uint32 + packetsOutOfOrder uint32 + packetsLost uint32 + frames uint32 + + jitter float64 + maxJitter float64 + + missingSNs [65536 / 64]uint64 + gapHistogram [GapHistogramNumBins]uint32 + + nacks uint32 + nackMisses uint32 + + plis uint32 + lastPli time.Time + + firs uint32 + lastFir time.Time +} + +func newStats(clockRate uint32) *Stats { + return &Stats{ + clockRate: clockRate, + } +} + +func (s *Stats) Update(rtp *rtp.Packet, packetTime int64) { + if !s.initialized { + s.initialized = true + s.highestSN = rtp.SequenceNumber - 1 + + s.extStartSN = uint32(rtp.SequenceNumber) + s.cycles = 0 + } + + isDuplicate := false + + pktSize := uint64(rtp.MarshalSize()) + s.bytes += pktSize + + diff := rtp.SequenceNumber - s.highestSN + switch { + // duplicate + case diff == 0: + s.bytesDuplicate += pktSize + s.packetsDuplicate++ + isDuplicate = true + + // out-of-order + case diff > (1 << 15): + s.packetsOutOfOrder++ + if !s.isMissingSN(rtp.SequenceNumber) { + s.bytesDuplicate += pktSize + s.packetsDuplicate++ + isDuplicate = true + } + + // in-order + default: + // update gap histogram + s.updateGapHistogram(int(diff)) + + // update missing sequence numbers + for lost := s.highestSN + 1; lost != rtp.SequenceNumber; lost++ { + s.setMissingSN(lost) + } + + if rtp.SequenceNumber < s.highestSN { + s.cycles++ + } + s.highestSN = rtp.SequenceNumber + + if rtp.Marker { + s.frames++ + } + } + + // clear received sequence number from missing list + if s.isMissingSN(rtp.SequenceNumber) { + s.packetsLost++ + } + s.clearMissingSN(rtp.SequenceNumber) + + if !isDuplicate { + s.updateJitter(rtp, packetTime) + } +} + +func (s *Stats) UpdateNack(nackCount int, nackMissCount int) { + s.nacks += uint32(nackCount) + s.nackMisses += uint32(nackMissCount) +} + +func (s *Stats) UpdatePli(now time.Time) { + s.plis++ + s.lastPli = now +} + +func (s *Stats) TimeSinceLastPli(now time.Time) int64 { + return now.UnixNano() - s.lastPli.UnixNano() +} + +func (s *Stats) UpdateFir(now time.Time) { + s.firs++ + s.lastFir = now +} + +func (s *Stats) ToString(startTime time.Time, endTime time.Time) string { + p := s.ToProto(startTime, endTime) + if p == nil { + return "" + } + + str := fmt.Sprintf("t: %+v|%+v|%.2fs", p.StartTime.AsTime().Format(time.UnixDate), p.EndTime.AsTime().Format(time.UnixDate), p.Duration) + + str += fmt.Sprintf(", p: %d|%.2f/s", p.Packets, p.PacketRate) + str += fmt.Sprintf(", l: %d|%.1f/s|%.2f%%", p.PacketsLost, p.PacketLossRate, p.PacketLossPercentage) + str += fmt.Sprintf(", b: %d|%.1fbps", p.Bytes, p.Bitrate) + str += fmt.Sprintf(", f: %d|%.1f/s", p.Frames, p.FrameRate) + + str += fmt.Sprintf(", d: %d|%.2f/s", p.PacketsDuplicate, p.PacketDuplicateRate) + str += fmt.Sprintf(", bd: %d|%.1fbps", p.BytesDuplicate, p.BitrateDuplicate) + + str += fmt.Sprintf(", pp: %d|%.2f/s", p.PacketsPadding, p.PacketPaddingRate) + str += fmt.Sprintf(", bp: %d|%.1fbps", p.BytesPadding, p.BitratePadding) + + str += fmt.Sprintf(", o: %d", p.PacketsOutOfOrder) + + str += fmt.Sprintf(", c: %d, j: %d(%.1fus)|%d(%.1fus)", s.clockRate, uint32(s.jitter), p.JitterCurrent, uint32(s.maxJitter), p.JitterMax) + + if len(p.GapHistogram) != 0 { + first := true + str += ", gh:[" + for burst, count := range p.GapHistogram { + if !first { + str += ", " + } + first = false + str += fmt.Sprintf("%d:%d", burst, count) + } + str += "]" + } + + str += ", n:" + str += fmt.Sprintf("%d|%d", p.Nacks, p.NackMisses) + + str += ", pli:" + str += fmt.Sprintf("%d|%+v", p.Plis, p.LastPli.AsTime().Format(time.UnixDate)) + + str += ", fir:" + str += fmt.Sprintf("%d|%+v", p.Firs, p.LastFir.AsTime().Format(time.UnixDate)) + + return str +} + +func (s *Stats) ToProto(startTime time.Time, endTime time.Time) *livekit.RTPStats { + if startTime.IsZero() { + return nil + } + + if endTime.IsZero() { + endTime = time.Now() + } + elapsed := endTime.Sub(startTime).Seconds() + + extHighestSN := uint32(s.cycles)<<16 | uint32(s.highestSN) + packetsExpected := extHighestSN - s.extStartSN + + packetsLost := s.packetsLost + for idx := range s.missingSNs { + n := s.missingSNs[idx] + for n != 0 { + packetsLost++ + n &= (n - 1) + } + } + packetLostRate := float64(packetsLost) / elapsed + packetLostPercentage := float32(packetsLost) / float32(packetsExpected) * 100.0 + + packets := packetsExpected - packetsLost + packetRate := float64(packets) / elapsed + packetDuplicateRate := float64(s.packetsDuplicate) / elapsed + packetPaddingRate := float64(s.packetsPadding) / elapsed + + bitrate := float64(s.bytes) * 8.0 / elapsed + bitrateDuplicate := float64(s.bytesDuplicate) * 8.0 / elapsed + bitratePadding := float64(s.bytesPadding) * 8.0 / elapsed + + frameRate := float64(s.frames) / elapsed + + jitterTime := s.jitter / float64(s.clockRate) * 1e6 + maxJitterTime := s.maxJitter / float64(s.clockRate) * 1e6 + + p := &livekit.RTPStats{ + StartTime: timestamppb.New(startTime), + EndTime: timestamppb.New(endTime), + Duration: elapsed, + Packets: packets, + PacketRate: packetRate, + Bytes: s.bytes, + Bitrate: bitrate, + PacketsLost: packetsLost, + PacketLossRate: packetLostRate, + PacketLossPercentage: packetLostPercentage, + PacketsDuplicate: s.packetsDuplicate, + PacketDuplicateRate: packetDuplicateRate, + BytesDuplicate: s.bytesDuplicate, + BitrateDuplicate: bitrateDuplicate, + PacketsPadding: s.packetsPadding, + PacketPaddingRate: packetPaddingRate, + BytesPadding: s.bytesPadding, + BitratePadding: bitratePadding, + PacketsOutOfOrder: s.packetsOutOfOrder, + Frames: s.frames, + FrameRate: frameRate, + JitterCurrent: jitterTime, + JitterMax: maxJitterTime, + Nacks: s.nacks, + NackMisses: s.nackMisses, + Plis: s.plis, + LastPli: timestamppb.New(s.lastPli), + Firs: s.firs, + LastFir: timestamppb.New(s.lastFir), + } + + gapsPresent := false + for i := 0; i < len(s.gapHistogram); i++ { + if s.gapHistogram[i] == 0 { + continue + } + + gapsPresent = true + break + } + + if gapsPresent { + p.GapHistogram = make(map[int32]uint32, GapHistogramNumBins) + for i := 0; i < len(s.gapHistogram); i++ { + if s.gapHistogram[i] == 0 { + continue + } + + p.GapHistogram[int32(i+1)] = s.gapHistogram[i] + } + } + + return p +} + +func (s *Stats) setMissingSN(sn uint16) { + idx, rem := getPos(sn) + s.missingSNs[idx] |= (1 << rem) +} + +func (s *Stats) clearMissingSN(sn uint16) { + idx, rem := getPos(sn) + s.missingSNs[idx] &^= (1 << rem) +} + +func (s *Stats) isMissingSN(sn uint16) bool { + idx, rem := getPos(sn) + return (s.missingSNs[idx] & (1 << rem)) != 0 +} + +func (s *Stats) updateJitter(rtp *rtp.Packet, packetTime int64) { + packetTimeRTP := uint32(packetTime / 1e6 * int64(s.clockRate/1e3)) + transit := packetTimeRTP - rtp.Timestamp + + if s.lastTransit != 0 { + d := int32(transit - s.lastTransit) + if d < 0 { + d = -d + } + s.jitter += (float64(d) - s.jitter) / 16 + if s.jitter > s.maxJitter { + s.maxJitter = s.jitter + } + } + + s.lastTransit = transit +} + +func (s *Stats) updateGapHistogram(gap int) { + if gap < 2 { + return + } + + missing := gap - 1 + if missing > len(s.gapHistogram) { + s.gapHistogram[len(s.gapHistogram)-1]++ + } else { + s.gapHistogram[missing-1]++ + } +} + +// ---------------------------------- + +type Window struct { + startTime time.Time + endTime time.Time + + stats *Stats +} + +func newWindow(startTime time.Time, clockRate uint32) *Window { + return &Window{ + startTime: startTime, + stats: newStats(clockRate), + } +} + +func (w *Window) Close(endTime time.Time) { + w.endTime = endTime +} + +func (w *Window) Update(rtp *rtp.Packet, packetTime int64) { + if !w.endTime.IsZero() { + return + } + + w.stats.Update(rtp, packetTime) +} + +func (w *Window) UpdateNack(nackCount int, nackMissCount int) { + if !w.endTime.IsZero() { + return + } + + w.stats.UpdateNack(nackCount, nackMissCount) +} + +func (w *Window) UpdatePli(now time.Time) { + if !w.endTime.IsZero() { + return + } + + w.stats.UpdatePli(now) +} + +func (w *Window) TimeSinceLastPli(now time.Time) int64 { + return w.stats.TimeSinceLastPli(now) +} + +func (w *Window) UpdateFir(now time.Time) { + if !w.endTime.IsZero() { + return + } + + w.stats.UpdateFir(now) +} + +func (w *Window) ToString() string { + return w.stats.ToString(w.startTime, w.endTime) +} + +func (w *Window) ToProto() *livekit.RTPStats { + return w.stats.ToProto(w.startTime, w.endTime) +} + +// ---------------------------------- + +func AggregateRTPStats(statses []*livekit.RTPStats) *livekit.RTPStats { + startTime := time.Time{} + endTime := time.Time{} + + packets := uint32(0) + bytes := uint64(0) + packetsLost := uint32(0) + packetsDuplicate := uint32(0) + bytesDuplicate := uint64(0) + packetsPadding := uint32(0) + bytesPadding := uint64(0) + packetsOutOfOrder := uint32(0) + frames := uint32(0) + jitter := float64(0.0) + maxJitter := float64(0) + gapHistogram := make(map[int32]uint32, GapHistogramNumBins) + nacks := uint32(0) + nackMisses := uint32(0) + plis := uint32(0) + lastPli := time.Time{} + firs := uint32(0) + lastFir := time.Time{} + + for _, stats := range statses { + if startTime.IsZero() || startTime.After(stats.StartTime.AsTime()) { + startTime = stats.StartTime.AsTime() + } + + if endTime.IsZero() || endTime.Before(stats.EndTime.AsTime()) { + endTime = stats.EndTime.AsTime() + } + + packets += stats.Packets + bytes += stats.Bytes + + packetsLost += stats.PacketsLost + + packetsDuplicate += stats.PacketsDuplicate + bytesDuplicate += stats.BytesDuplicate + + packetsPadding += stats.PacketsPadding + bytesPadding += stats.BytesPadding + + packetsOutOfOrder += stats.PacketsOutOfOrder + + frames += stats.Frames + + jitter += stats.JitterCurrent + if stats.JitterMax > maxJitter { + maxJitter = stats.JitterMax + } + + for burst, count := range stats.GapHistogram { + gapHistogram[burst] += count + } + + nacks += stats.Nacks + nackMisses += stats.NackMisses + + plis += stats.Plis + if lastPli.IsZero() || lastPli.Before(stats.LastPli.AsTime()) { + lastPli = stats.LastPli.AsTime() + } + + firs += stats.Firs + if lastFir.IsZero() || lastPli.Before(stats.LastFir.AsTime()) { + lastFir = stats.LastFir.AsTime() + } + } + + if endTime.IsZero() { + endTime = time.Now() + } + elapsed := endTime.Sub(startTime).Seconds() + + packetLostRate := float64(packetsLost) / elapsed + packetLostPercentage := float32(packetsLost) / (float32(packets) + float32(packetsLost)) * 100.0 + + packetRate := float64(packets) / elapsed + packetDuplicateRate := float64(packetsDuplicate) / elapsed + packetPaddingRate := float64(packetsPadding) / elapsed + + bitrate := float64(bytes) * 8.0 / elapsed + bitrateDuplicate := float64(bytesDuplicate) * 8.0 / elapsed + bitratePadding := float64(bytesPadding) * 8.0 / elapsed + + frameRate := float64(frames) / elapsed + + return &livekit.RTPStats{ + StartTime: timestamppb.New(startTime), + EndTime: timestamppb.New(endTime), + Duration: elapsed, + Packets: packets, + PacketRate: packetRate, + Bytes: bytes, + Bitrate: bitrate, + PacketsLost: packetsLost, + PacketLossRate: packetLostRate, + PacketLossPercentage: packetLostPercentage, + PacketsDuplicate: packetsDuplicate, + PacketDuplicateRate: packetDuplicateRate, + BytesDuplicate: bytesDuplicate, + BitrateDuplicate: bitrateDuplicate, + PacketsPadding: packetsPadding, + PacketPaddingRate: packetPaddingRate, + BytesPadding: bytesPadding, + BitratePadding: bitratePadding, + PacketsOutOfOrder: packetsOutOfOrder, + Frames: frames, + FrameRate: frameRate, + JitterCurrent: jitter / float64(len(statses)), + JitterMax: maxJitter, + GapHistogram: gapHistogram, + Nacks: nacks, + NackMisses: nackMisses, + Plis: plis, + LastPli: timestamppb.New(lastPli), + Firs: firs, + LastFir: timestamppb.New(lastFir), + } +} diff --git a/pkg/rtc/rtpstats_test.go b/pkg/rtc/rtpstats_test.go new file mode 100644 index 000000000..5507554f1 --- /dev/null +++ b/pkg/rtc/rtpstats_test.go @@ -0,0 +1,62 @@ +package rtc + +import ( + "fmt" + "math/rand" + "testing" + "time" + + "github.com/pion/rtp" +) + +func getPacket(sn uint16, ts uint32, payloadSize int) *rtp.Packet { + return &rtp.Packet{ + Header: rtp.Header{ + SequenceNumber: sn, + Timestamp: ts, + }, + Payload: make([]byte, payloadSize), + } +} + +func TestRTPStats(t *testing.T) { + clockRate := uint32(90000) + windowDuration := 5 * time.Second + r := NewRTPStats(RTPStatsParams{ + ClockRate: clockRate, + WindowDuration: windowDuration, + }) + + totalDuration := 4 * windowDuration + bitrate := 1000000 + packetSize := 1000 + pps := (((bitrate + 7) / 8) + packetSize - 1) / packetSize + framerate := 30 + sleep := 1000 / framerate + packetsPerFrame := (pps + framerate - 1) / framerate + + sequenceNumber := uint16(rand.Float64() * float64(1<<16)) + timestamp := uint32(rand.Float64() * float64(1<<32)) + now := time.Now() + startTime := now + lastFrameTime := now + for now.Sub(startTime) < totalDuration { + timestamp += uint32(now.Sub(lastFrameTime).Seconds() * float64(clockRate)) + for i := 0; i < packetsPerFrame; i++ { + r.Update(getPacket(sequenceNumber, timestamp, packetSize), time.Now().UnixNano()) + if (sequenceNumber % 100) == 0 { + jump := uint16(rand.Float64() * 120.0) + sequenceNumber += jump + } else { + sequenceNumber++ + } + } + + lastFrameTime = now + time.Sleep(time.Duration(sleep) * time.Millisecond) + now = time.Now() + } + + r.Stop() + fmt.Printf("%s\n", r.ToString()) +}