Skip to content

feat(PCMLocalTrack): add option to grab track and hw stats #879

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
anunaym14 merged 2 commits into from
Apr 8, 2026
Merged

feat(PCMLocalTrack): add option to grab track and hw stats #879

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
9141f75
feat(PCMLocalTrack): add option to grab track and hw stats
anunaym14 Apr 8, 2026
d62548b
cleanup
anunaym14 Apr 8, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,8 @@ require (
google.golang.org/protobuf v1.36.11
)

require github.com/mackerelio/go-osstat v0.2.7 // indirect

require (
buf.build/gen/go/bufbuild/protovalidate/protocolbuffers/go v1.36.11-20260209202127-80ab13bee0bf.1 // indirect
buf.build/go/protovalidate v1.1.2 // indirect
Expand Down Expand Up @@ -136,7 +138,7 @@ require (
golang.org/x/exp v0.0.0-20260212183809-81e46e3db34a // indirect
golang.org/x/net v0.50.0 // indirect
golang.org/x/sync v0.19.0
golang.org/x/sys v0.41.0 // indirect
golang.org/x/sys v0.42.0 // indirect
golang.org/x/text v0.34.0 // indirect
golang.org/x/time v0.14.0 // indirect
golang.org/x/tools v0.42.0 // indirect
Expand Down
6 changes: 4 additions & 2 deletions go.sum
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -163,6 +163,8 @@ github.com/livekit/protocol v1.45.2-0.20260403151849-8a360e8d0221 h1:loe7h+z1kOu
github.com/livekit/protocol v1.45.2-0.20260403151849-8a360e8d0221/go.mod h1:e6QdWDkfot+M2nRh0eitJUS0ZLuwvKCsfiz2pWWSG3s=
github.com/livekit/psrpc v0.7.1 h1:ms37az0QTD3UXIWuUC5D/SkmKOlRMVRsI261eBWu/Vw=
github.com/livekit/psrpc v0.7.1/go.mod h1:bZ4iHFQptTkbPnB0LasvRNu/OBYXEu1NA6O5BMFo9kk=
github.com/mackerelio/go-osstat v0.2.7 h1:TCavZi10wF49bT6iQZ9eT2keGZQpC69MTDfdJej5e94=
github.com/mackerelio/go-osstat v0.2.7/go.mod h1:dwpYh5pIPmvk+IEwBKNIWRFMB92mrC08CmXOhDC7nQk=
github.com/magefile/mage v1.15.0 h1:BvGheCMAsG3bWUDbZ8AyXXpCNwU9u5CB6sM+HNb9HYg=
github.com/magefile/mage v1.15.0/go.mod h1:z5UZb/iS3GoOSn0JgWuiw7dxlurVYTu+/jHXqQg881A=
github.com/maxbrunsfeld/counterfeiter/v6 v6.12.1 h1:D4O2wLxB384TS3ohBJMfolnxb4qGmoZ1PnWNtit8LYo=
Expand Down Expand Up @@ -390,8 +392,8 @@ golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.17.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/sys v0.41.0 h1:Ivj+2Cp/ylzLiEU89QhWblYnOE9zerudt9Ftecq2C6k=
golang.org/x/sys v0.41.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo=
golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k=
Expand Down
170 changes: 167 additions & 3 deletions pkg/media/pcmlocaltrack.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,14 +10,17 @@ import (
"github.com/livekit/media-sdk"
"github.com/livekit/media-sdk/opus"
protoLogger "github.com/livekit/protocol/logger"
"github.com/livekit/protocol/utils/hwstats"
"github.com/pion/webrtc/v4"
"go.uber.org/atomic"

lksdk "github.com/livekit/server-sdk-go/v2"
)

type PCMLocalTrackParams struct {
Encryptor Encryptor
Encryptor Encryptor
EnableStats bool
EnableHWStats bool
}

type PCMLocalTrackOption func(*PCMLocalTrackParams)
Expand All @@ -28,6 +31,18 @@ func WithEncryptor(encryptor Encryptor) PCMLocalTrackOption {
}
}

func WithTrackStats() PCMLocalTrackOption {
return func(p *PCMLocalTrackParams) {
p.EnableStats = true
}
}

func WithHWStats() PCMLocalTrackOption {
return func(p *PCMLocalTrackParams) {
p.EnableHWStats = true
}
}

type PCMLocalTrack struct {
*webrtc.TrackLocalStaticSample

Expand All @@ -53,6 +68,21 @@ type PCMLocalTrack struct {
closed atomic.Bool
muted atomic.Bool
bound atomic.Bool

logger protoLogger.Logger
enableStats bool
loggingEnabled bool
logState pcmLocalTrackLogState
cpuStats *hwstats.CPUStats
memStats *hwstats.MemoryStats
}

type pcmLocalTrackLogState struct {
at time.Time
totalWritten uint64
totalProcessed uint64
prevWritten uint64
prevProcessed uint64
}

// NewPCMLocalTrack creates a wrapper around a webrtc.TrackLocalStaticSample that accepts PCM16 samples via the WriteSample method,
Expand Down Expand Up @@ -103,6 +133,13 @@ func NewPCMLocalTrack(
resampledPCMWriter = media.ResampleWriter(pcmWriter, sourceSampleRate)
}

var cpuStats *hwstats.CPUStats
var memStats *hwstats.MemoryStats
if params.EnableHWStats {
cpuStats, _ = hwstats.NewCPUStats(nil)
memStats, _ = hwstats.NewMemoryStats()
}

// the final chain of writers:
// WriteSample -> resamplesPCMWriter (resamples source to target sample rate as necessary)
// -> PCMWriter (encodes PCM -> Opus)
Expand All @@ -117,6 +154,14 @@ func NewPCMLocalTrack(
sourceChannels: sourceChannels,
chunkBuffer: new(deque.Deque[media.PCM16Sample]),
samplesPerFrame: (sourceSampleRate * sourceChannels * int(defaultPCMFrameDuration/time.Nanosecond)) / 1e9,
logger: logger,
enableStats: params.EnableStats,
loggingEnabled: params.EnableStats || params.EnableHWStats,
cpuStats: cpuStats,
memStats: memStats,
logState: pcmLocalTrackLogState{
Copy link
Copy Markdown
Contributor

@boks1971 boks1971 Apr 8, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

looks like logState is on irrespective of the enables, can that be turned off if stats are not enabled or just too much of hassle?

Copy link
Copy Markdown
Member Author

@anunaym14 anunaym14 Apr 8, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

fixed

at: time.Now(),
},
}

t.cond = sync.NewCond(&t.mu)
Expand Down Expand Up @@ -184,7 +229,15 @@ func (t *PCMLocalTrack) WriteSample(chunk media.PCM16Sample) error {
t.mu.Lock()
t.chunkBuffer.PushBack(chunkCopy)
t.cond.Broadcast()
var snapshot *pcmLocalTrackLogSnapshot
if t.loggingEnabled {
t.logState.totalWritten += uint64(len(chunk))
snapshot = t.collectLogSnapshotLocked(time.Now())
}
t.mu.Unlock()
if snapshot != nil {
t.emitLogSnapshot(snapshot)
}
return nil
}

Expand All @@ -197,12 +250,23 @@ func (t *PCMLocalTrack) processSamples() {
break
}

var frame media.PCM16Sample
var snapshot *pcmLocalTrackLogSnapshot

t.mu.Lock()
frame := t.getFrameFromChunkBuffer()
frame = t.getFrameFromChunkBuffer()
if frame != nil && t.loggingEnabled {
t.logState.totalProcessed += uint64(len(frame))
snapshot = t.collectLogSnapshotLocked(time.Now())
}
t.mu.Unlock()

if frame != nil {
t.resampledPCMWriter.WriteSample(frame)
if snapshot != nil {
t.emitLogSnapshot(snapshot)
}
}
t.mu.Unlock()

<-ticker.C
}
Expand Down Expand Up @@ -256,6 +320,9 @@ func (t *PCMLocalTrack) Close() error {
t.mu.Lock()
t.cond.Broadcast()
t.mu.Unlock()
if t.cpuStats != nil {
t.cpuStats.Stop()
}
}
return nil
}
Expand All @@ -267,3 +334,100 @@ func (t *PCMLocalTrack) SampleRate() int {
func (t *PCMLocalTrack) String() string {
return "PCMLocalTrack"
}

type pcmLocalTrackLogSnapshot struct {
interval time.Duration
queueSamples int
totalWritten uint64
totalProcessed uint64
deltaWritten uint64
deltaProcessed uint64
}

func (t *PCMLocalTrack) collectLogSnapshotLocked(now time.Time) *pcmLocalTrackLogSnapshot {
if t.logState.at.IsZero() {
t.logState.at = now
return nil
}

const logInterval = 5 * time.Second
if now.Sub(t.logState.at) < logInterval {
return nil
}

interval := now.Sub(t.logState.at)
snapshot := &pcmLocalTrackLogSnapshot{
interval: interval,
queueSamples: t.getNumSamplesInChunkBuffer(),
totalWritten: t.logState.totalWritten,
totalProcessed: t.logState.totalProcessed,
deltaWritten: t.logState.totalWritten - t.logState.prevWritten,
deltaProcessed: t.logState.totalProcessed - t.logState.prevProcessed,
}

t.logState.at = now
t.logState.prevWritten = t.logState.totalWritten
t.logState.prevProcessed = t.logState.totalProcessed

return snapshot
}

func (t *PCMLocalTrack) emitLogSnapshot(snapshot *pcmLocalTrackLogSnapshot) {
if snapshot == nil || snapshot.interval <= 0 {
return
}

elapsed := snapshot.interval.Seconds()
if elapsed == 0 {
return
}

fields := make([]interface{}, 0, 20)

if t.enableStats {
chanCount := float64(t.sourceChannels)
if chanCount == 0 {
chanCount = 1
}

ingressHz := float64(snapshot.deltaWritten) / chanCount / elapsed

processedSamples := float64(snapshot.deltaProcessed)
if t.sourceSampleRate != 0 && t.sourceSampleRate != DefaultOpusSampleRate {
processedSamples *= float64(DefaultOpusSampleRate) / float64(t.sourceSampleRate)
}
egressHz := processedSamples / chanCount / elapsed

queueSeconds := 0.0
if t.sourceSampleRate != 0 {
queueSeconds = float64(snapshot.queueSamples) / (chanCount * float64(t.sourceSampleRate))
}

fields = append(fields,
"interval_s", elapsed,
"ingress_hz", ingressHz,
"egress_hz", egressHz,
"queue_samples", snapshot.queueSamples,
"queue_s", queueSeconds,
"total_written", snapshot.totalWritten,
"total_processed", snapshot.totalProcessed,
"source_sample_rate", t.sourceSampleRate,
)
}

if t.cpuStats != nil {
fields = append(fields,
"cpu_load", t.cpuStats.GetCPULoad(),
"num_cpu", t.cpuStats.NumCPU(),
)
}

if t.memStats != nil {
used, total, err := t.memStats.GetMemory()
if err == nil {
fields = append(fields, "mem_used", used, "mem_total", total)
}
}

t.logger.Infow("pcm local track stats", fields...)
}
Loading