94 lines
2.3 KiB
Go
94 lines
2.3 KiB
Go
// Copyright 2023 LiveKit, Inc.
|
|
//
|
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
|
// you may not use this file except in compliance with the License.
|
|
// You may obtain a copy of the License at
|
|
//
|
|
// http://www.apache.org/licenses/LICENSE-2.0
|
|
//
|
|
// Unless required by applicable law or agreed to in writing, software
|
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
// See the License for the specific language governing permissions and
|
|
// limitations under the License.
|
|
|
|
package utils
|
|
|
|
import (
|
|
"time"
|
|
|
|
"go.uber.org/zap/zapcore"
|
|
)
|
|
|
|
// a ring buffer of welford mean/var summaries used to aggregate jitter and rtt.
|
|
type LatencyAggregate struct {
|
|
summary []Welford
|
|
ivl time.Duration
|
|
cap uint64
|
|
head uint64
|
|
}
|
|
|
|
func NewLatencyAggregate(interval, windowLength time.Duration) *LatencyAggregate {
|
|
c := uint64((windowLength + interval - 1) / interval)
|
|
return &LatencyAggregate{
|
|
summary: make([]Welford, c),
|
|
ivl: interval,
|
|
cap: uint64(c),
|
|
}
|
|
}
|
|
|
|
// extend the ring to contain ts then merge the value into the interval summary.
|
|
func (a *LatencyAggregate) Update(ts time.Duration, v float64) {
|
|
i := uint64(ts / a.ivl)
|
|
if i+a.cap < a.head {
|
|
return
|
|
}
|
|
|
|
if i > a.head {
|
|
k := a.head + 1
|
|
if k+a.cap < i {
|
|
k = i - a.cap
|
|
}
|
|
for ; k <= i; k++ {
|
|
a.summary[k%a.cap].Reset()
|
|
}
|
|
a.head = i
|
|
}
|
|
|
|
a.summary[i%a.cap].Update(v)
|
|
}
|
|
|
|
func (a *LatencyAggregate) Get(ts time.Duration) (Welford, bool) {
|
|
i := uint64(ts / a.ivl)
|
|
if i+a.cap < a.head || i > a.head {
|
|
return Welford{}, false
|
|
}
|
|
return a.summary[i%a.cap], true
|
|
}
|
|
|
|
// aggregate the interval summaries
|
|
func (a *LatencyAggregate) Summarize() Welford {
|
|
return WelfordMerge(a.summary...)
|
|
}
|
|
|
|
func (a *LatencyAggregate) SummarizeLast(d time.Duration) Welford {
|
|
n := min(a.head, a.cap, uint64((d+a.ivl-1)/a.ivl))
|
|
l := (a.head - n) % a.cap
|
|
r := (a.head % a.cap)
|
|
if l < r {
|
|
return WelfordMerge(a.summary[l:r]...)
|
|
}
|
|
return WelfordMerge(
|
|
WelfordMerge(a.summary[:r]...),
|
|
WelfordMerge(a.summary[l:]...),
|
|
)
|
|
}
|
|
|
|
func (a *LatencyAggregate) MarshalLogObject(e zapcore.ObjectEncoder) error {
|
|
summary := a.Summarize()
|
|
e.AddFloat64("count", summary.Count())
|
|
e.AddFloat64("mean", summary.Mean())
|
|
e.AddFloat64("stddev", summary.StdDev())
|
|
return nil
|
|
}
|