mirror of
https://github.com/hanzoai/livekit.git
synced 2026-10-30 18:29:34 +00:00
zeekay
e9292da479
This fork lived at github.com/hanzoai/livekit but still declared
`module github.com/livekit/livekit-server`, so all 505 self-references pointed
at upstream. A fork that does not own its module path is the exact thing the
owned-module-paths rule exists to prevent: every import line in every consumer
carries someone else's brand, and `go get` resolves to their code, not ours.
module + 505 self-imports github.com/livekit/livekit-server -> github.com/hanzoai/livekit
KV client github.com/redis/go-redis/v9 -> github.com/hanzokv/go/v9
message bus github.com/livekit/psrpc -> github.com/hanzoai/psrpc
psrpc had to be forked (hanzoai/psrpc, Apache-2.0) because its bus is typed
against the KV client — the same shape redsync had for hanzoai/git.
DELIBERATELY NOT FORKED: github.com/livekit/{protocol,mediatransportutil,
mageutil} keep their addresses. Only `protocol` still reaches go-redis, through
a single file (redis/redis.go) out of 305. Forking an 11.5MB core wire-protocol
library that this server must track in lockstep — and that upstream pushed
yesterday — to change one file trades a tidy dependency graph for permanent
divergence on the fastest-moving dependency here. So one indirect
github.com/redis/go-redis entry remains, via livekit/protocol/redis, and that
is the correct place for it to live.
112 lines
2.7 KiB
Go
112 lines
2.7 KiB
Go
package sfu
|
|
|
|
import (
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/hanzoai/livekit/pkg/telemetry/prometheus"
|
|
"github.com/livekit/protocol/logger"
|
|
"github.com/livekit/protocol/utils"
|
|
"github.com/livekit/protocol/utils/mono"
|
|
)
|
|
|
|
const (
|
|
cHighForwardingLatency = 20 * time.Millisecond
|
|
cSkewFactor = 10
|
|
)
|
|
|
|
type ForwardStats struct {
|
|
lock sync.Mutex
|
|
latency *utils.LatencyAggregate
|
|
lowest int64
|
|
highest int64
|
|
lastUpdateAt int64
|
|
closeCh chan struct{}
|
|
}
|
|
|
|
func NewForwardStats(latencyUpdateInterval, reportInterval, latencyWindowLength time.Duration) *ForwardStats {
|
|
s := &ForwardStats{
|
|
latency: utils.NewLatencyAggregate(latencyUpdateInterval, latencyWindowLength),
|
|
lowest: time.Second.Nanoseconds(),
|
|
closeCh: make(chan struct{}),
|
|
}
|
|
|
|
go s.report(reportInterval)
|
|
return s
|
|
}
|
|
|
|
func (s *ForwardStats) Update(arrival, left int64) (int64, bool) {
|
|
transit := left - arrival
|
|
isHighForwardingLatency := time.Duration(transit) > cHighForwardingLatency
|
|
|
|
s.lock.Lock()
|
|
s.latency.Update(time.Duration(arrival), float64(transit))
|
|
s.lowest = min(transit, s.lowest)
|
|
s.highest = max(transit, s.highest)
|
|
s.lastUpdateAt = arrival
|
|
s.lock.Unlock()
|
|
|
|
prometheus.RecordForwardLatencySample(transit)
|
|
return transit, isHighForwardingLatency
|
|
}
|
|
|
|
func (s *ForwardStats) GetStats(shortDuration time.Duration) (time.Duration, time.Duration) {
|
|
s.lock.Lock()
|
|
// a dummy sample to flush the pipe to current time
|
|
now := mono.UnixNano()
|
|
if (now - s.lastUpdateAt) > shortDuration.Nanoseconds() {
|
|
s.latency.Update(time.Duration(now), 0)
|
|
}
|
|
|
|
wLong := s.latency.Summarize()
|
|
|
|
lowest := s.lowest
|
|
s.lowest = time.Second.Nanoseconds()
|
|
|
|
highest := s.highest
|
|
s.highest = 0
|
|
s.lock.Unlock()
|
|
|
|
latencyLong, jitterLong := time.Duration(wLong.Mean()), time.Duration(wLong.StdDev())
|
|
if jitterLong > latencyLong*cSkewFactor {
|
|
logger.Infow(
|
|
"high jitter in forwarding path",
|
|
"lowest", time.Duration(lowest),
|
|
"highest", time.Duration(highest),
|
|
"countLong", wLong.Count(),
|
|
"latencyLong", latencyLong,
|
|
"jitterLong", jitterLong,
|
|
)
|
|
}
|
|
return latencyLong, jitterLong
|
|
}
|
|
|
|
func (s *ForwardStats) GetShortStats(shortDuration time.Duration) (time.Duration, time.Duration) {
|
|
s.lock.Lock()
|
|
wShort := s.latency.SummarizeLast(shortDuration)
|
|
s.lock.Unlock()
|
|
|
|
return time.Duration(wShort.Mean()), time.Duration(wShort.StdDev())
|
|
}
|
|
|
|
func (s *ForwardStats) Stop() {
|
|
close(s.closeCh)
|
|
}
|
|
|
|
func (s *ForwardStats) report(reportInterval time.Duration) {
|
|
ticker := time.NewTicker(reportInterval)
|
|
defer ticker.Stop()
|
|
|
|
for {
|
|
select {
|
|
case <-s.closeCh:
|
|
return
|
|
|
|
case <-ticker.C:
|
|
latencyLong, jitterLong := s.GetStats(reportInterval)
|
|
prometheus.RecordForwardJitter(uint32(jitterLong.Nanoseconds()))
|
|
prometheus.RecordForwardLatency(uint32(latencyLong.Nanoseconds()))
|
|
}
|
|
}
|
|
}
|