diff --git a/utils/hwstats/cpu.go b/utils/hwstats/cpu.go index 1ec118a..97e9ab7 100644 --- a/utils/hwstats/cpu.go +++ b/utils/hwstats/cpu.go @@ -15,6 +15,9 @@ package hwstats import ( + "os" + "strconv" + "sync" "time" "github.com/frostbyte73/core" @@ -27,6 +30,33 @@ import ( // This object returns cgroup quota aware cpu stats. On other systems than Linux, // it falls back to full system stats +const CPURequestEnvVar = "LIVEKIT_CPU_REQUEST" + +var ( + effectiveCPURequestOnce sync.Once + effectiveCPURequest float64 +) + +// EffectiveCPURequest returns the value of LIVEKIT_CPU_REQUEST when set and valid. +// The result is parsed at most once; use for stats normalization and optional +// GOMAXPROCS tuning (see utils/hwstats/maxprocs). +func EffectiveCPURequest() float64 { + effectiveCPURequestOnce.Do(parseCPURequestEnv) + return effectiveCPURequest +} + +func parseCPURequestEnv() { + cpuReq := os.Getenv(CPURequestEnvVar) + if cpuReq == "" { + return + } + n, err := strconv.ParseFloat(cpuReq, 64) + if err != nil || n <= 0 { + return + } + effectiveCPURequest = n +} + type platformCPUMonitor interface { getCPUIdle() (float64, error) numCPU() float64 @@ -139,6 +169,9 @@ func (c *CPUStats) GetCPUIdle() float64 { } func (c *CPUStats) NumCPU() float64 { + if n := EffectiveCPURequest(); n > 0 { + return n + } return c.platform.numCPU() } @@ -186,7 +219,7 @@ func (c *CPUStats) monitorCPULoad() { } func (c *CPUStats) monitorProcesses() { - numCPU := c.platform.numCPU() + numCPU := c.NumCPU() pageSize := getPageSize() fs, err := procfs.NewFS(procfs.DefaultMountPoint) diff --git a/utils/hwstats/cpu_request_test.go b/utils/hwstats/cpu_request_test.go new file mode 100644 index 0000000..f70715f --- /dev/null +++ b/utils/hwstats/cpu_request_test.go @@ -0,0 +1,145 @@ +package hwstats + +import ( + "encoding/json" + "fmt" + "os" + "os/exec" + "runtime" + "strings" + "testing" + "time" + + "github.com/frostbyte73/core" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// --------------------------------------------------------------------------- +// Mock platform monitor +// --------------------------------------------------------------------------- + +type mockPlatform struct { + nCPU float64 +} + +func (m *mockPlatform) getCPUIdle() (float64, error) { return m.nCPU, nil } +func (m *mockPlatform) numCPU() float64 { return m.nCPU } + +func newTestCPUStats(plat platformCPUMonitor) *CPUStats { + return &CPUStats{ + platform: plat, + warningThrottle: core.NewThrottle(time.Minute), + closeChan: make(chan struct{}), + } +} + +// --------------------------------------------------------------------------- +// NumCPU without override (no env var set in this process) +// --------------------------------------------------------------------------- + +func TestNumCPU_NoOverride(t *testing.T) { + c := newTestCPUStats(&mockPlatform{nCPU: 64}) + assert.Equal(t, 64.0, c.NumCPU()) + + c2 := newTestCPUStats(&mockPlatform{nCPU: 8.5}) + assert.Equal(t, 8.5, c2.NumCPU()) +} + +// --------------------------------------------------------------------------- +// Subprocess tests for EffectiveCPURequest and NumCPU with override +// +// sync.Once means we can only test one env-var value per process. +// Each test case spawns a subprocess with the desired env. +// --------------------------------------------------------------------------- + +const subprocessEnvKey = "HWSTATS_TEST_SUBPROCESS" + +type subprocessResult struct { + EffectiveCPU float64 `json:"effective_cpu"` + NumCPU float64 `json:"num_cpu"` + PlatformCPU float64 `json:"platform_cpu"` +} + +func TestEffectiveCPURequest(t *testing.T) { + tests := []struct { + name string + envValue string + expectedCPU float64 + }{ + {"valid integer", "4", 4}, + {"valid float", "3.33", 3.33}, + {"valid large", "15", 15}, + {"valid fractional", "0.5", 0.5}, + {"empty", "", 0}, + {"negative", "-1", 0}, + {"zero", "0", 0}, + {"invalid string", "abc", 0}, + {"invalid mixed", "4abc", 0}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + result := runSubprocess(t, tt.envValue) + assert.InDelta(t, tt.expectedCPU, result.EffectiveCPU, 0.001) + }) + } +} + +func runSubprocess(t *testing.T, cpuRequestEnv string) subprocessResult { + t.Helper() + + cmd := exec.Command(os.Args[0], "-test.run=TestSubprocessWorker", "-test.v") + + // Build clean env + var env []string + for _, e := range os.Environ() { + if !strings.HasPrefix(e, "LIVEKIT_CPU_REQUEST=") && + !strings.HasPrefix(e, subprocessEnvKey+"=") { + env = append(env, e) + } + } + env = append(env, subprocessEnvKey+"=1") + if cpuRequestEnv != "" { + env = append(env, "LIVEKIT_CPU_REQUEST="+cpuRequestEnv) + } + cmd.Env = env + + out, err := cmd.Output() + if err != nil { + if exitErr, ok := err.(*exec.ExitError); ok { + t.Fatalf("subprocess failed: %s\nstderr: %s", err, exitErr.Stderr) + } + t.Fatalf("subprocess failed: %s", err) + } + + var result subprocessResult + for _, line := range strings.Split(string(out), "\n") { + line = strings.TrimSpace(line) + if strings.HasPrefix(line, "{") { + require.NoError(t, json.Unmarshal([]byte(line), &result)) + return result + } + } + t.Fatalf("no JSON result in subprocess output: %s", string(out)) + return result +} + +// TestSubprocessWorker runs inside spawned subprocesses only. +func TestSubprocessWorker(t *testing.T) { + if os.Getenv(subprocessEnvKey) != "1" { + t.Skip("only runs as subprocess") + } + + plat := &mockPlatform{nCPU: float64(runtime.NumCPU())} + c := newTestCPUStats(plat) + + result := subprocessResult{ + EffectiveCPU: EffectiveCPURequest(), + NumCPU: c.NumCPU(), + PlatformCPU: plat.numCPU(), + } + + b, _ := json.Marshal(result) + fmt.Println(string(b)) +} diff --git a/utils/hwstats/maxprocs/maxprocs.go b/utils/hwstats/maxprocs/maxprocs.go new file mode 100644 index 0000000..496d795 --- /dev/null +++ b/utils/hwstats/maxprocs/maxprocs.go @@ -0,0 +1,77 @@ +// Copyright 2026 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 maxprocs caps GOMAXPROCS to the Kubernetes CPU request so the Go +// scheduler doesn't create more OS threads than the pod is guaranteed. +// +// # Why not CPU limits / automaxprocs +// +// Real-time media workloads (egress, ingress) intentionally omit CPU limits +// to avoid CFS throttling — even brief throttle events cause audio glitches +// and video frame drops. Without a limit the cgroup quota is "max", so +// libraries like uber/automaxprocs fall back to runtime.NumCPU (the full +// node). This package uses the CPU request instead, which reflects what the +// Kubernetes scheduler actually guarantees to the pod. +// +// When a CPU limit IS set, the cgroup quota gives a tighter bound and Go +// (or automaxprocs) already derives GOMAXPROCS from it. This package still +// works in that case — it caps GOMAXPROCS down to the request but never +// increases it, so the stricter of (limit, request) wins. +// +// # Usage +// +// The CPU request is read from the LIVEKIT_CPU_REQUEST environment variable, +// which should be populated e.g via the Kubernetes Downward API: +// +// env: +// - name: LIVEKIT_CPU_REQUEST +// valueFrom: +// resourceFieldRef: +// containerName: +// resource: requests.cpu +// +// Without this variable the package is a no-op. A blank import is +// sufficient — init runs at process startup: +// +// import _ "github.com/livekit/protocol/utils/hwstats/maxprocs" +package maxprocs + +import ( + "math" + "runtime" + "sync" + + "github.com/livekit/protocol/utils/hwstats" +) + +var once sync.Once + +func init() { Install() } + +// Install caps GOMAXPROCS to ceil(LIVEKIT_CPU_REQUEST) when that environment +// variable is set and valid. It never increases GOMAXPROCS beyond its current +// value. It is a no-op when the variable is absent. Runs at most once per +// process. +func Install() { + once.Do(func() { + req := hwstats.EffectiveCPURequest() + if req <= 0 { + return + } + cap := int(math.Ceil(req)) + if runtime.GOMAXPROCS(0) > cap { + runtime.GOMAXPROCS(cap) + } + }) +} diff --git a/utils/hwstats/maxprocs/maxprocs_test.go b/utils/hwstats/maxprocs/maxprocs_test.go new file mode 100644 index 0000000..de75cc4 --- /dev/null +++ b/utils/hwstats/maxprocs/maxprocs_test.go @@ -0,0 +1,160 @@ +package maxprocs_test + +import ( + "encoding/json" + "fmt" + "os" + "os/exec" + "runtime" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const subprocessEnvKey = "MAXPROCS_TEST_SUBPROCESS" + +type subprocessResult struct { + GOMAXPROCS int `json:"gomaxprocs"` +} + +func TestInstall(t *testing.T) { + hostCPU := runtime.NumCPU() + + tests := []struct { + name string + cpuRequest string // LIVEKIT_CPU_REQUEST value ("" = unset) + gomaxprocsEnv string // GOMAXPROCS value ("" = unset) + expectedMaxProcs int + }{ + { + name: "no env vars set", + expectedMaxProcs: hostCPU, + }, + { + name: "request caps GOMAXPROCS down", + cpuRequest: "4", + expectedMaxProcs: 4, + }, + { + name: "fractional request rounds up", + cpuRequest: "3.33", + expectedMaxProcs: 4, + }, + { + name: "request of 1", + cpuRequest: "1", + expectedMaxProcs: 1, + }, + { + name: "request larger than host CPUs (no change)", + cpuRequest: "9999", + expectedMaxProcs: hostCPU, + }, + { + name: "explicit GOMAXPROCS lower than request (respected)", + cpuRequest: "8", + gomaxprocsEnv: "4", + expectedMaxProcs: 4, + }, + { + name: "explicit GOMAXPROCS higher than request (capped)", + cpuRequest: "4", + gomaxprocsEnv: "16", + expectedMaxProcs: 4, + }, + { + name: "explicit GOMAXPROCS equal to request", + cpuRequest: "8", + gomaxprocsEnv: "8", + expectedMaxProcs: 8, + }, + { + name: "explicit GOMAXPROCS without request (unchanged)", + gomaxprocsEnv: "4", + expectedMaxProcs: 4, + }, + { + name: "invalid request ignored", + cpuRequest: "abc", + expectedMaxProcs: hostCPU, + }, + { + name: "negative request ignored", + cpuRequest: "-2", + expectedMaxProcs: hostCPU, + }, + { + name: "zero request ignored", + cpuRequest: "0", + expectedMaxProcs: hostCPU, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + result := runSubprocess(t, tt.cpuRequest, tt.gomaxprocsEnv) + assert.Equal(t, tt.expectedMaxProcs, result.GOMAXPROCS) + }) + } +} + +func runSubprocess(t *testing.T, cpuRequest, gomaxprocs string) subprocessResult { + t.Helper() + + cmd := exec.Command(os.Args[0], "-test.run=TestSubprocessWorker", "-test.v") + + // Build clean env: start from current, filter out the vars we control + var env []string + for _, e := range os.Environ() { + if strings.HasPrefix(e, "LIVEKIT_CPU_REQUEST=") || + strings.HasPrefix(e, "GOMAXPROCS=") || + strings.HasPrefix(e, subprocessEnvKey+"=") { + continue + } + env = append(env, e) + } + env = append(env, subprocessEnvKey+"=1") + if cpuRequest != "" { + env = append(env, "LIVEKIT_CPU_REQUEST="+cpuRequest) + } + if gomaxprocs != "" { + env = append(env, "GOMAXPROCS="+gomaxprocs) + } + cmd.Env = env + + out, err := cmd.Output() + if err != nil { + if exitErr, ok := err.(*exec.ExitError); ok { + t.Fatalf("subprocess failed: %s\nstderr: %s", err, exitErr.Stderr) + } + t.Fatalf("subprocess failed: %s", err) + } + + var result subprocessResult + for _, line := range strings.Split(string(out), "\n") { + line = strings.TrimSpace(line) + if strings.HasPrefix(line, "{") { + require.NoError(t, json.Unmarshal([]byte(line), &result)) + return result + } + } + t.Fatalf("no JSON result in subprocess output: %s", string(out)) + return result +} + +// TestSubprocessWorker runs inside spawned subprocesses. +// The maxprocs init() has already fired by the time we get here. +func TestSubprocessWorker(t *testing.T) { + if os.Getenv(subprocessEnvKey) != "1" { + t.Skip("only runs as subprocess") + } + + // init() already called Install() — just report the result + result := subprocessResult{ + GOMAXPROCS: runtime.GOMAXPROCS(0), + } + b, _ := json.Marshal(result) + fmt.Println(string(b)) +}