diff --git a/go.mod b/go.mod index 5b93017e4..58481d40a 100644 --- a/go.mod +++ b/go.mod @@ -21,7 +21,7 @@ require ( github.com/jxskiss/base62 v1.1.0 github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 github.com/livekit/mediatransportutil v0.0.0-20261014192331-dd8e31a0adc0 - github.com/livekit/protocol v1.53.1-0.20261023175527-46332a036e25 + github.com/livekit/protocol v1.53.1-0.20261026092902-78ad907b61e0 github.com/livekit/psrpc v0.8.2 github.com/mackerelio/go-osstat v0.2.8 github.com/magefile/mage v1.17.2 diff --git a/go.sum b/go.sum index 2cd875cd8..031ba11c3 100644 --- a/go.sum +++ b/go.sum @@ -166,8 +166,8 @@ github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 h1:9x+U2HGLrSw5AT github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ= github.com/livekit/mediatransportutil v0.0.0-20261014192331-dd8e31a0adc0 h1:BSLyHr3Lsl71bHGHsfiYNmbohugR2yoaf9HjAZeFJnc= github.com/livekit/mediatransportutil v0.0.0-20261014192331-dd8e31a0adc0/go.mod h1:o8CFmAdrVwzJNOCsQCLUzXRjokkufNshnQHOe4fRaqU= -github.com/livekit/protocol v1.53.1-0.20261023175527-46332a036e25 h1:v9UMGn8kGnrs0gja5795d8RBCQqKCVlXVeH14UDiuwc= -github.com/livekit/protocol v1.53.1-0.20261023175527-46332a036e25/go.mod h1:h9P4KCc3K715NCgCrkkFsb8/RRY1kKqOS1EIHUNRlCk= +github.com/livekit/protocol v1.53.1-0.20261026092902-78ad907b61e0 h1:YUmqxA8BFQMnJ8YLVuB0t9eAYDitz3icWkfk3Ua+4Wc= +github.com/livekit/protocol v1.53.1-0.20261026092902-78ad907b61e0/go.mod h1:h9P4KCc3K715NCgCrkkFsb8/RRY1kKqOS1EIHUNRlCk= github.com/livekit/psrpc v0.8.2 h1:7tbkhBJoYlK5Csfkdv1XYbZjyAQeaEZgupvIXNCv7Dc= github.com/livekit/psrpc v0.8.2/go.mod h1:6KnU8vULoizX0Er9NNfeKJDV8ug87j3fuo19M0bhj6Q= github.com/livekit/webrtc-pion/v4 v4.2.18-warp.7 h1:I1UkbeJI5V6slqLmef0JoaGF8wPnFsKzzXGUzz4KDvs= diff --git a/pkg/sfu/buffer/buffer_base.go b/pkg/sfu/buffer/buffer_base.go index 03cf77eec..51dc860f6 100644 --- a/pkg/sfu/buffer/buffer_base.go +++ b/pkg/sfu/buffer/buffer_base.go @@ -88,6 +88,8 @@ type ExtPacket struct { AbsCaptureTimeExt *act.AbsCaptureTime IsOutOfOrder bool IsBuffered bool + // the reader was waiting on an empty buffer when this packet arrived + ReaderWasWaiting bool // backing storage for Packet when the buffer creates the ext packet, // extension and CSRC slices point into the write input until ReadExtended re-points them into RawPacket @@ -170,7 +172,8 @@ type BufferBase struct { params BufferBaseParams - readCond *sync.Cond + readCond *sync.Cond + readerWaiting bool bucket *bucket.Bucket[uint64, uint16] lastBucketCapCheckAt int64 @@ -674,7 +677,9 @@ func (b *BufferBase) ReadExtended(buf []byte) (*ExtPacket, error) { return patched, nil } + b.readerWaiting = true b.readCond.Wait() + b.readerWaiting = false } } @@ -947,6 +952,7 @@ func (b *BufferBase) HandleIncomingPacketLocked( if ep == nil { return 0, errors.New("could not get ext packet") } + ep.ReaderWasWaiting = b.readerWaiting && b.extPackets.Len() == 0 b.extPackets.PushBack(ep) b.readCond.Broadcast() diff --git a/pkg/sfu/buffer/buffer_test.go b/pkg/sfu/buffer/buffer_test.go index b965c0565..799267e8e 100644 --- a/pkg/sfu/buffer/buffer_test.go +++ b/pkg/sfu/buffer/buffer_test.go @@ -779,3 +779,54 @@ func TestGetPacketConcurrentWithWrite(t *testing.T) { wg.Wait() }) } + +func TestReaderWasWaiting(t *testing.T) { + buff := NewBuffer(123, 1, 1) + buff.OnRtcpFeedback(func(_ []rtcp.Packet) {}) + require.NoError(t, buff.Bind(webrtc.RTPParameters{Codecs: []webrtc.RTPCodecParameters{opusCodec}}, opusCodec.RTPCodecCapability, 0)) + write := func(sn uint16) { + pkt := &rtp.Packet{ + Header: rtp.Header{Version: 2, PayloadType: 111, SequenceNumber: sn, Timestamp: uint32(sn) * 960, SSRC: 123}, + Payload: make([]byte, 100), + } + b, err := pkt.Marshal() + require.NoError(t, err) + _, err = buff.Write(b) + require.NoError(t, err) + } + read := func() *ExtPacket { + extPkt, err := buff.ReadExtended(make([]byte, 1500)) + require.NoError(t, err) + require.NotNil(t, extPkt) + return extPkt + } + + // packets that arrive while nobody reads queue up, none found the reader waiting + write(1) + write(2) + require.False(t, read().ReaderWasWaiting) + require.False(t, read().ReaderWasWaiting) + + // a reader parked on the empty buffer is found waiting by the next packet + got := make(chan bool, 1) + go func() { + extPkt, err := buff.ReadExtended(make([]byte, 1500)) + got <- err == nil && extPkt != nil && extPkt.ReaderWasWaiting + }() + require.Eventually(t, func() bool { + buff.RLock() + defer buff.RUnlock() + return buff.readerWaiting + }, time.Second, time.Millisecond) + write(3) + require.True(t, <-got) + + // two packets that arrive before the parked reader wakes: only the first found it waiting + buff.Lock() + buff.readerWaiting = true + buff.Unlock() + write(4) + write(5) + require.True(t, read().ReaderWasWaiting) + require.False(t, read().ReaderWasWaiting) +} diff --git a/pkg/sfu/forwardstats.go b/pkg/sfu/forwardstats.go index a7dfbb0dc..983dd917c 100644 --- a/pkg/sfu/forwardstats.go +++ b/pkg/sfu/forwardstats.go @@ -1,6 +1,7 @@ package sfu import ( + "math" "math/bits" "sync" "time" @@ -31,7 +32,7 @@ const ( forwardSampleShardSel = forwardSampleNumShards - 1 ) -// forwardSampleShard is a ring of transit samples with multiple producers and a +// forwardSampleShard is a ring of packed samples with multiple producers and a // single consumer. A producer reserves a slot, stores the value, then publishes // the slot's epoch (reserved index + 1). The consumer reads a slot only once its // epoch marks the value committed for that index. @@ -42,7 +43,7 @@ type forwardSampleShard struct { seq [forwardSampleShardCap]atomic.Uint64 // per-slot publish epoch } -// forwardSampleBuffer holds per-packet transit samples produced on the packet +// forwardSampleBuffer holds per-packet samples produced on the packet // path and consumed by the background worker, which performs metric emission. type forwardSampleBuffer struct { shards [forwardSampleNumShards]forwardSampleShard @@ -51,11 +52,11 @@ type forwardSampleBuffer struct { // push records a sample: reserve a slot, store the value, then publish the // slot's epoch. The shard is selected from arrival time bits. -func (b *forwardSampleBuffer) push(arrival, transitNs int64) { +func (b *forwardSampleBuffer) push(arrival, sample int64) { sh := &b.shards[(uint64(arrival)>>6)&forwardSampleShardSel] i := sh.writeIdx.Add(1) - 1 slot := i & forwardSampleShardMask - sh.ring[slot].Store(transitNs) + sh.ring[slot].Store(sample) sh.seq[slot].Store(i + 1) } @@ -67,7 +68,7 @@ func (b *forwardSampleBuffer) push(arrival, transitNs int64) { // draining stops and resumes from there on the next call, so no sample is read // stale or skipped. When producers get a shard's capacity ahead, or overwrite a // slot before it is read, the affected samples are counted as dropped. -func (b *forwardSampleBuffer) drain(fn func(transitNs int64)) { +func (b *forwardSampleBuffer) drain(fn func(sample int64)) { for si := range b.shards { sh := &b.shards[si] w := sh.writeIdx.Load() @@ -100,6 +101,34 @@ func (b *forwardSampleBuffer) takeDropped() uint64 { return b.dropped.Swap(0) } +// A sample packs transit (high 40 bits, in ns, up to ~18 minutes) and scheduling delay +// (low 24 bits, in 256 ns units, up to ~4.29s) into one slot, so the second value costs +// no extra atomic. +const ( + forwardSchedulingDelayBits = 24 + forwardSchedulingDelayShift = 8 // 256 ns units + forwardNoSchedulingDelay = 1<= 0 { + delay = min(uint64(schedulingDelayNs)>>forwardSchedulingDelayShift, forwardNoSchedulingDelay-1) + } + return int64(uint64(min(max(transitNs, 0), forwardMaxTransitNs))<> forwardSchedulingDelayBits) + if delay := uint64(sample) & forwardNoSchedulingDelay; delay != forwardNoSchedulingDelay { + return transitNs, int64(delay << forwardSchedulingDelayShift) + } + return transitNs, -1 +} + // forwardSummary is a mergeable histogram of forwarding transit over an // interval. Each power-of-two octave [2^e, 2^(e+1)) us is split into // forwardHistSub linear sub-buckets, and each bucket keeps the observed @@ -221,6 +250,19 @@ func (s forwardSummary) percentile(p float64) time.Duration { return time.Duration(s.maxNs) } +// forwardIntervalSummary holds one summary interval's transit and scheduling delay. +type forwardIntervalSummary struct { + transit forwardSummary + scheduling forwardSummary +} + +func (s forwardIntervalSummary) merge(o forwardIntervalSummary) forwardIntervalSummary { + return forwardIntervalSummary{ + transit: s.transit.merge(o.transit), + scheduling: s.scheduling.merge(o.scheduling), + } +} + type ForwardStats struct { samples forwardSampleBuffer @@ -228,7 +270,7 @@ type ForwardStats struct { // written by the background worker (flush) and read both by the worker // (report) and by external callers (GetStats), so it is guarded by lock. lock sync.Mutex - ring []forwardSummary + ring []forwardIntervalSummary ringHead int ringLen int @@ -245,7 +287,7 @@ func NewForwardStats(summaryInterval, reportInterval, reportWindow time.Duration } s := &ForwardStats{ - ring: make([]forwardSummary, ringCap), + ring: make([]forwardIntervalSummary, ringCap), summaryInterval: summaryInterval, reportInterval: reportInterval, closeCh: make(chan struct{}), @@ -259,8 +301,14 @@ func NewForwardStats(summaryInterval, reportInterval, reportWindow time.Duration // and returns the transit and whether it exceeds the high-latency threshold. // The sample is aggregated and emitted by the background worker. func (s *ForwardStats) Update(arrival, left int64) (int64, bool) { + return s.UpdateWithSchedulingDelay(arrival, left, -1) +} + +// UpdateWithSchedulingDelay is Update that also records a scheduling delay when it is +// not negative. +func (s *ForwardStats) UpdateWithSchedulingDelay(arrival, left, schedulingDelay int64) (int64, bool) { transit := left - arrival - s.samples.push(arrival, transit) + s.samples.push(arrival, packForwardSample(transit, schedulingDelay)) return transit, time.Duration(transit) > cHighForwardingLatency } @@ -294,10 +342,14 @@ func (s *ForwardStats) run() { // histogram, and folds the interval summary into the window ring used for the // latency gauge. func (s *ForwardStats) flush() { - var summ forwardSummary - s.samples.drain(func(transitNs int64) { + var summ forwardIntervalSummary + s.samples.drain(func(sample int64) { + transitNs, schedulingDelayNs := unpackForwardSample(sample) prometheus.RecordForwardLatencySample(transitNs) - summ.addSample(transitNs) + summ.transit.addSample(transitNs) + if schedulingDelayNs >= 0 { + summ.scheduling.addSample(schedulingDelayNs) + } }) s.lock.Lock() @@ -311,7 +363,7 @@ func (s *ForwardStats) flush() { // summarize merges the ring summaries covering the most recent window. A // window <= 0 (or >= the report window) covers the entire ring. -func (s *ForwardStats) summarize(window time.Duration) forwardSummary { +func (s *ForwardStats) summarize(window time.Duration) forwardIntervalSummary { s.lock.Lock() defer s.lock.Unlock() @@ -327,7 +379,7 @@ func (s *ForwardStats) summarize(window time.Duration) forwardSummary { } // walk backwards from the most recent entry (ringHead-1) over n entries. - var w forwardSummary + var w forwardIntervalSummary for i := 0; i < n; i++ { idx := (s.ringHead - 1 - i + len(s.ring)) % len(s.ring) w = w.merge(s.ring[idx]) @@ -341,11 +393,12 @@ func (s *ForwardStats) summarize(window time.Duration) forwardSummary { // that covers it). A duration <= 0, or one that meets/exceeds the report window, // covers the full window. func (s *ForwardStats) GetStats(duration time.Duration) time.Duration { - return s.summarize(duration).percentile(forwardLatencyPercentile) + return s.summarize(duration).transit.percentile(forwardLatencyPercentile) } func (s *ForwardStats) report() { - w := s.summarize(0) + ws := s.summarize(0) + w := ws.transit p90 := w.percentile(forwardLatencyPercentile) if dropped := s.samples.takeDropped(); dropped > 0 { @@ -363,5 +416,11 @@ func (s *ForwardStats) report() { ) } - prometheus.RecordForwardLatency(uint32(p90.Nanoseconds())) + prometheus.RecordForwardLatency(saturatingNs(p90)) + prometheus.RecordForwardSchedulingDelay(saturatingNs(ws.scheduling.percentile(forwardLatencyPercentile))) +} + +// saturatingNs converts to uint32 nanoseconds, saturating at ~4.29s instead of wrapping. +func saturatingNs(d time.Duration) uint32 { + return uint32(min(max(d.Nanoseconds(), 0), math.MaxUint32)) } diff --git a/pkg/sfu/forwardstats_test.go b/pkg/sfu/forwardstats_test.go index fe145659a..73f885c52 100644 --- a/pkg/sfu/forwardstats_test.go +++ b/pkg/sfu/forwardstats_test.go @@ -1,6 +1,7 @@ package sfu import ( + "math" "sync" "testing" "time" @@ -263,7 +264,7 @@ func TestForwardSampleBuffer_Concurrent(t *testing.T) { // --------------------------------------------------------------------------- func TestForwardStats_Update(t *testing.T) { - s := &ForwardStats{ring: make([]forwardSummary, 1)} + s := &ForwardStats{ring: make([]forwardIntervalSummary, 1)} // below threshold transit, isHigh := s.Update(1000, 1000+int64(5*time.Millisecond)) @@ -283,7 +284,7 @@ func TestForwardStats_Update(t *testing.T) { func TestForwardStats_Flush(t *testing.T) { initPrometheus(t) - s := &ForwardStats{ring: make([]forwardSummary, 4)} + s := &ForwardStats{ring: make([]forwardIntervalSummary, 4)} for i := 0; i < 10; i++ { s.Update(0, int64((i+1)*1000)) // 1us..10us @@ -292,7 +293,7 @@ func TestForwardStats_Flush(t *testing.T) { s.flush() require.Equal(t, 1, s.ringLen) - summ := s.ring[0] + summ := s.ring[0].transit require.Equal(t, int64(10), summ.count) require.Equal(t, int64(1000), summ.minNs) require.Equal(t, int64(10000), summ.maxNs) @@ -301,14 +302,69 @@ func TestForwardStats_Flush(t *testing.T) { // a subsequent flush with no new samples appends an empty summary s.flush() require.Equal(t, 2, s.ringLen) - require.Equal(t, int64(0), s.ring[1].count) + require.Equal(t, int64(0), s.ring[1].transit.count) +} + +func TestForwardSample_Pack(t *testing.T) { + transit, delay := unpackForwardSample(packForwardSample(250_000, 7_168)) + require.Equal(t, int64(250_000), transit) + require.Equal(t, int64(7_168), delay) + + // scheduling delay keeps 256 ns resolution + _, delay = unpackForwardSample(packForwardSample(250_000, 7_300)) + require.Equal(t, int64(7_168), delay) + + // no scheduling delay, and a zero delay is a sample, not "none" + _, delay = unpackForwardSample(packForwardSample(1_000, -1)) + require.Equal(t, int64(-1), delay) + _, delay = unpackForwardSample(packForwardSample(1_000, 0)) + require.Equal(t, int64(0), delay) + + // a long stall keeps its full transit; the delay saturates and never reads as "none" + transit, delay = unpackForwardSample(packForwardSample(10*int64(time.Second), 10*int64(time.Second))) + require.Equal(t, 10*int64(time.Second), transit) + require.Equal(t, int64((forwardNoSchedulingDelay-1)< 0 && r.forwardStats != nil && !extPkt.IsBuffered && !extPkt.IsOutOfOrder { - if latency, isHigh := r.forwardStats.Update(extPkt.Arrival, mono.UnixNano()); isHigh { + // a packet that found the forwarder waiting measures only its wake-up; a queued + // one also waited behind earlier packets, so it carries no scheduling delay + schedulingDelay := int64(-1) + if extPkt.ReaderWasWaiting { + schedulingDelay = dequeuedAt - extPkt.Arrival + } + if latency, isHigh := r.forwardStats.UpdateWithSchedulingDelay(extPkt.Arrival, mono.UnixNano(), schedulingDelay); isHigh { r.params.Logger.Debugw( "high forwarding latency", "latency", time.Duration(latency), diff --git a/pkg/sfu/receiver_test.go b/pkg/sfu/receiver_test.go index 27e53ae78..80c9eade3 100644 --- a/pkg/sfu/receiver_test.go +++ b/pkg/sfu/receiver_test.go @@ -17,19 +17,23 @@ package sfu import ( "fmt" "hash/fnv" + "io" "math/rand" "runtime" "sync" "testing" "github.com/gammazero/workerpool" + "github.com/pion/rtp" "github.com/pion/webrtc/v4" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "go.uber.org/atomic" + "github.com/livekit/livekit-server/pkg/sfu/buffer" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" + "github.com/livekit/protocol/utils/mono" ) func TestWebRTCReceiver_OnCloseHandler(t *testing.T) { @@ -128,6 +132,79 @@ func TestReceiverBaseAddDownTrackReturnsClosedWhenCloseWinsRace(t *testing.T) { require.Empty(t, receiver.GetDownTracks()) } +// burstBuffer serves one packet per flag from ReadExtended, then io.EOF. A flag says +// whether the packet found the reader waiting. +type burstBuffer struct { + buffer.BufferProvider + readerWasWaiting []bool +} + +func (b *burstBuffer) ReadExtended([]byte) (*buffer.ExtPacket, error) { + if len(b.readerWasWaiting) == 0 { + return nil, io.EOF + } + readerWasWaiting := b.readerWasWaiting[0] + b.readerWasWaiting = b.readerWasWaiting[1:] + return &buffer.ExtPacket{ + Arrival: mono.UnixNano(), + Packet: &rtp.Packet{Header: rtp.Header{PayloadType: 111}}, + ReaderWasWaiting: readerWasWaiting, + }, nil +} + +type writeTestSender struct { + receiverCloseTestSender +} + +func (s *writeTestSender) WriteRTP(*buffer.ExtPacket, int32) int32 { + return 1 +} + +func (s *writeTestSender) Close() {} + +func TestReceiverBaseSchedulingDelayOnlyWhenIdle(t *testing.T) { + receiver := NewReceiverBase( + ReceiverBaseParams{ + TrackID: "track", + StreamID: "stream", + Kind: webrtc.RTPCodecTypeAudio, + Codec: webrtc.RTPCodecParameters{ + RTPCodecCapability: webrtc.RTPCodecCapability{ + MimeType: webrtc.MimeTypeOpus, + ClockRate: 48000, + Channels: 2, + }, + PayloadType: 111, + }, + Logger: logger.GetLogger(), + }, + &livekit.TrackInfo{ + Sid: "track", + Type: livekit.TrackType_AUDIO, + Source: livekit.TrackSource_MICROPHONE, + }, + ReceiverCodecStateNormal, + ) + defer receiver.Close("test", false) + + fs := &ForwardStats{ring: make([]forwardIntervalSummary, 1)} + receiver.SetForwardStats(fs) + require.NoError(t, receiver.AddDownTrack(&writeTestSender{receiverCloseTestSender{subscriberID: "subscriber"}})) + + // a burst of five where only the first found the forwarder waiting, then one more that did + buff := &burstBuffer{readerWasWaiting: []bool{true, false, false, false, false, true}} + + var wg sync.WaitGroup + wg.Add(1) + receiver.forwardRTP(0, buff, receiver.forwardersGeneration.Load(), &wg) + + fs.flush() + summ := fs.ring[0] + require.Equal(t, int64(6), summ.transit.count) + // the burst's later packets queued behind earlier ones, so they carry no scheduling delay + require.Equal(t, int64(2), summ.scheduling.count) +} + func BenchmarkWriteRTP(b *testing.B) { cases := []int{1, 2, 5, 10, 100, 250, 500} workers := runtime.NumCPU() diff --git a/pkg/telemetry/prometheus/node.go b/pkg/telemetry/prometheus/node.go index 915b0322a..0bde6fdae 100644 --- a/pkg/telemetry/prometheus/node.go +++ b/pkg/telemetry/prometheus/node.go @@ -174,6 +174,7 @@ func GetNodeStats(nodeStartedAt int64, prevStats []*livekit.NodeStats, rateInter ParticipantRtcCanceled: participantRTCCanceled.Load(), ParticipantRtcActive: participantRTCActive.Load(), ForwardLatency: forwardLatency.Load(), + ForwardSchedulingDelay: forwardSchedulingDelay.Load(), NumCpus: uint32(cpuStats.NumCPU()), // this will round down to the nearest integer CpuLoad: float32(cpuStats.GetCPULoad()), MemoryTotal: memTotal, diff --git a/pkg/telemetry/prometheus/packets.go b/pkg/telemetry/prometheus/packets.go index a741ef9c2..5c35ecdef 100644 --- a/pkg/telemetry/prometheus/packets.go +++ b/pkg/telemetry/prometheus/packets.go @@ -51,30 +51,32 @@ var ( participantRTCCanceled atomic.Uint64 participantRTCActive atomic.Uint64 forwardLatency atomic.Uint32 + forwardSchedulingDelay atomic.Uint32 - promPacketLabels = []string{"direction", "transmission", "country"} - promPacketTotal *prometheus.CounterVec - promPacketBytes *prometheus.CounterVec - promFECLabels = []string{"direction", "type"} - promFECPacketTotal *prometheus.CounterVec - promFECPacketBytes *prometheus.CounterVec - promRTCPLabels = []string{"direction", "country"} - promStreamLabels = []string{"direction", "source", "type", "country"} - promNackTotal *prometheus.CounterVec - promRTXCappedTotal *prometheus.CounterVec - promKeyFrameTotal *prometheus.CounterVec - promPliTotal *prometheus.CounterVec - promFirTotal *prometheus.CounterVec - promPacketLossTotal *prometheus.CounterVec - promPacketLoss *prometheus.HistogramVec - promPacketOutOfOrderTotal *prometheus.CounterVec - promPacketOutOfOrder *prometheus.HistogramVec - promJitter *prometheus.HistogramVec - promRTT *prometheus.HistogramVec - promParticipantJoin *prometheus.CounterVec - promConnections *prometheus.GaugeVec - promForwardLatency prometheus.Gauge - promForwardLatencyHist prometheus.Histogram + promPacketLabels = []string{"direction", "transmission", "country"} + promPacketTotal *prometheus.CounterVec + promPacketBytes *prometheus.CounterVec + promFECLabels = []string{"direction", "type"} + promFECPacketTotal *prometheus.CounterVec + promFECPacketBytes *prometheus.CounterVec + promRTCPLabels = []string{"direction", "country"} + promStreamLabels = []string{"direction", "source", "type", "country"} + promNackTotal *prometheus.CounterVec + promRTXCappedTotal *prometheus.CounterVec + promKeyFrameTotal *prometheus.CounterVec + promPliTotal *prometheus.CounterVec + promFirTotal *prometheus.CounterVec + promPacketLossTotal *prometheus.CounterVec + promPacketLoss *prometheus.HistogramVec + promPacketOutOfOrderTotal *prometheus.CounterVec + promPacketOutOfOrder *prometheus.HistogramVec + promJitter *prometheus.HistogramVec + promRTT *prometheus.HistogramVec + promParticipantJoin *prometheus.CounterVec + promConnections *prometheus.GaugeVec + promForwardLatency prometheus.Gauge + promForwardSchedulingDelay prometheus.Gauge + promForwardLatencyHist prometheus.Histogram ) func initPacketStats(nodeID string, nodeType livekit.NodeType) { @@ -193,6 +195,13 @@ func initPacketStats(nodeID string, nodeType livekit.NodeType) { Name: "latency", ConstLabels: prometheus.Labels{"node_id": nodeID, "node_type": nodeType.String()}, }) + promForwardSchedulingDelay = prometheus.NewGauge(prometheus.GaugeOpts{ + Namespace: livekitNamespace, + Subsystem: "forward", + Name: "scheduling_delay", + Help: "p90 time from arrival to dequeue, in ns, for packets that found their forwarder idle.", + ConstLabels: prometheus.Labels{"node_id": nodeID, "node_type": nodeType.String()}, + }) promForwardLatencyHist = prometheus.NewHistogram(prometheus.HistogramOpts{ Namespace: livekitNamespace, Subsystem: "forward_latency", @@ -231,6 +240,7 @@ func initPacketStats(nodeID string, nodeType livekit.NodeType) { prometheus.MustRegister(promParticipantJoin) prometheus.MustRegister(promConnections) prometheus.MustRegister(promForwardLatency) + prometheus.MustRegister(promForwardSchedulingDelay) prometheus.MustRegister(promForwardLatencyHist) } @@ -442,3 +452,8 @@ func RecordForwardLatency(p90 uint32) { forwardLatency.Store(p90) promForwardLatency.Set(float64(p90)) } + +func RecordForwardSchedulingDelay(p90 uint32) { + forwardSchedulingDelay.Store(p90) + promForwardSchedulingDelay.Set(float64(p90)) +}