alexpires.me/a13labs.infra
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

auto_suspend: binary GPU idle counter replaces EMA for GPU

Browse Source
This commit is contained in:
Alexandre Pires
2026-06-09 22:13:45 +02:00
parent 4825ad1495
commit df5a98853b
2 changed files with 186 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ansible/roles/auto_suspend/files/auto_suspend.py
+47 -40
View File
@@ -2,9 +2,10 @@
"""
Auto-suspend daemon — checks system idleness and suspends when idle.
Uses an exponential moving average (EMA) of CPU idle % and GPU utilization
to smooth out transient spikes. The system suspends only after
idle_timeout_cycles consecutive idle checks.
Uses an exponential moving average (EMA) of CPU idle % to smooth out
transient spikes. GPU utilization is tracked as a binary busy/idle value
per cycle: if any GPU exceeds the threshold, the idle counter resets.
The system suspends only after idle_timeout_cycles consecutive idle checks.
State is persisted to /var/lib/auto_suspend/state.json and survives
across timer invocations. State is fully reset on suspend resume.
@@ -78,10 +79,8 @@ def fresh_state():
return {
"cpu_idle_n": None,
"cpu_idle_n1": None,
"gpu_util_n": None,
"gpu_util_n1": None,
"cpu_avg": None,
"gpu_avg": None,
"gpu_idle_counter": 0,
"idle_counter": 0,
"sample_count": 0,
}
@@ -101,8 +100,9 @@ def load_state(state_dir, logger):
for key in fresh_state():
if key not in state:
state[key] = fresh_state()[key]
logger.debug("Loaded state: sample_count=%d, idle_counter=%d",
state["sample_count"], state["idle_counter"])
logger.debug("Loaded state: sample_count=%d, idle_counter=%d, gpu_idle=%d",
state["sample_count"], state["idle_counter"],
state.get("gpu_idle_counter", 0))
return state
except (json.JSONDecodeError, IOError, OSError) as exc:
logger.warning("Failed to load state (%s) — starting fresh", exc)
@@ -122,8 +122,9 @@ def save_state(state_dir, state, logger):
json.dump(state, f, indent=2)
f.write("\n")
os.replace(tmp, str(state_path))
logger.debug("State saved: sample_count=%d, idle_counter=%d",
state["sample_count"], state["idle_counter"])
logger.debug("State saved: sample_count=%d, idle_counter=%d, gpu_idle=%d",
state["sample_count"], state["idle_counter"],
state.get("gpu_idle_counter", 0))
except BaseException:
try:
os.unlink(tmp)
@@ -288,31 +289,33 @@ def get_active_users(logger):
# EMA update
# ---------------------------------------------------------------------------
def update_averages(state, cpu_n, gpu_n, alpha, logger):
def update_averages(state, cpu_n, gpu_n, gpu_threshold, alpha, logger):
"""
Update EMA averages using the new measurement and shift n → n-1.
Update CPU EMA averages and GPU idle counter.
GPU is binary per-cycle: if utilization > threshold, the GPU idle
counter resets. If idle (below threshold), the counter increments.
Args:
state: current state dict (modified in place for averages)
cpu_n: current CPU idle %
gpu_n: current GPU utilization %
gpu_n: current GPU utilization % (None = no GPU, treated as idle)
gpu_threshold: utilization threshold above which GPU is "busy"
alpha: EMA weight for current sample
logger: logger instance
Returns:
dict with updated state (new keys: cpu_avg, gpu_avg, cpu_idle_n,
gpu_util_n, cpu_idle_n1, gpu_util_n1, sample_count)
dict with updated state (new keys: cpu_avg, cpu_idle_n, cpu_idle_n1,
gpu_idle_counter, sample_count)
"""
new_state = dict(state)
new_state["sample_count"] = state["sample_count"] + 1
# Shift current → previous
new_state["cpu_idle_n1"] = state["cpu_idle_n"]
new_state["gpu_util_n1"] = state["gpu_util_n"]
# Update current measurements
new_state["cpu_idle_n"] = cpu_n
new_state["gpu_util_n"] = gpu_n
# EMA for CPU idle average
if state["cpu_avg"] is not None and cpu_n is not None:
@@ -321,16 +324,20 @@ def update_averages(state, cpu_n, gpu_n, alpha, logger):
# First valid sample seeds the average
new_state["cpu_avg"] = cpu_n
# EMA for GPU utilization average
if state["gpu_avg"] is not None and gpu_n is not None:
new_state["gpu_avg"] = round(alpha * gpu_n + (1 - alpha) * state["gpu_avg"], 2)
elif gpu_n is not None:
new_state["gpu_avg"] = gpu_n
# GPU idle counter — binary per-cycle (no EMA)
gpu_busy = gpu_n is not None and gpu_n > gpu_threshold
if gpu_busy:
new_state["gpu_idle_counter"] = 0
logger.debug("GPU busy (%.1f%% >= %.0f%%) — reset gpu_idle_counter",
gpu_n, gpu_threshold)
else:
# gpu_n is None (no GPU) or below threshold — counts as idle
new_state["gpu_idle_counter"] = state.get("gpu_idle_counter", 0) + 1
logger.debug(
"Updated averages: cpu_avg=%.1f%%, gpu_avg=%.1f%%, sample_count=%d",
"Updated averages: cpu_avg=%.1f%%, gpu_idle_counter=%d, sample_count=%d",
new_state["cpu_avg"] if new_state["cpu_avg"] is not None else 0,
new_state["gpu_avg"] if new_state["gpu_avg"] is not None else 0,
new_state["gpu_idle_counter"],
new_state["sample_count"],
)
return new_state
@@ -340,7 +347,8 @@ def update_averages(state, cpu_n, gpu_n, alpha, logger):
# ---------------------------------------------------------------------------
def get_why_not_idle(state, has_users, cpu_idle_threshold=DEFAULT_CPU_IDLE_THRESHOLD,
gpu_util_threshold=DEFAULT_GPU_UTIL_THRESHOLD, min_samples=DEFAULT_MIN_SAMPLES, logger=None):
gpu_util_threshold=DEFAULT_GPU_UTIL_THRESHOLD, min_samples=DEFAULT_MIN_SAMPLES,
gpu_idle_counter=None, logger=None):
"""Return a human-readable reason why the system is not idle."""
reasons = []
if has_users:
@@ -350,11 +358,8 @@ def get_why_not_idle(state, has_users, cpu_idle_threshold=DEFAULT_CPU_IDLE_THRES
reasons.append(
f"CPU idle avg {state['cpu_avg']:.1f}% <= threshold {cpu_idle_threshold:.0f}%"
)
if state["gpu_avg"] is not None:
if state["gpu_avg"] >= gpu_util_threshold:
reasons.append(
f"GPU util avg {state['gpu_avg']:.1f}% >= threshold {gpu_util_threshold:.0f}%"
)
if gpu_idle_counter is not None and gpu_idle_counter == 0 and state["sample_count"] >= min_samples:
reasons.append("GPU busy (idle_counter=0)")
if not reasons:
if state["sample_count"] < min_samples:
reasons.append(f"warming up ({state['sample_count']}/{min_samples} samples)")
@@ -363,11 +368,12 @@ def get_why_not_idle(state, has_users, cpu_idle_threshold=DEFAULT_CPU_IDLE_THRES
return "; ".join(reasons)
def evaluate(state, has_users, idle_threshold, gpu_threshold,
def evaluate(state, has_users, gpu_idle_counter, idle_threshold, gpu_threshold,
min_samples, logger):
"""
Evaluate whether the system should be considered idle.
GPU is binary per-cycle: busy if any GPU > threshold, otherwise idle.
Returns True if all conditions are met for one idle cycle.
"""
# Need enough samples before trusting the average
@@ -391,11 +397,10 @@ def evaluate(state, has_users, idle_threshold, gpu_threshold,
)
return False
# Check GPU utilization average (None means no GPUs = ok)
if state["gpu_avg"] is not None and state["gpu_avg"] >= gpu_threshold:
# GPU is binary per-cycle: busy if counter was reset this cycle
if gpu_idle_counter == 0 and state["sample_count"] >= min_samples:
logger.debug(
"Not idle: GPU avg %.1f%% >= threshold %.0f%%",
state["gpu_avg"], gpu_threshold,
"Not idle: GPU busy this cycle (idle_counter=0)"
)
return False
@@ -457,14 +462,14 @@ def main():
# --- Update averages ---
state = update_averages(
state, cpu_idle, gpu_util, args.alpha, logger,
state, cpu_idle, gpu_util, args.gpu_util_threshold, args.alpha, logger,
)
# --- Evaluate ---
has_users = get_active_users(logger)
is_idle = evaluate(
state, has_users,
state, has_users, state["gpu_idle_counter"],
idle_threshold=args.cpu_idle_threshold,
gpu_threshold=args.gpu_util_threshold,
min_samples=args.min_samples,
@@ -474,10 +479,10 @@ def main():
if is_idle:
state["idle_counter"] = state.get("idle_counter", 0) + 1
logger.info(
"System idle (cpu_avg=%.1f%%, gpu_avg=%.1f%%, "
"System idle (cpu_avg=%.1f%%, gpu_idle=%d, "
"idle_counter=%d/%d)",
state["cpu_avg"] or 0,
state["gpu_avg"] or 0,
state["gpu_idle_counter"],
state["idle_counter"],
args.idle_timeout_cycles,
)
@@ -512,7 +517,9 @@ def main():
else:
if not args.dry_run:
state["idle_counter"] = 0
reason = get_why_not_idle(state, has_users, args.cpu_idle_threshold, args.gpu_util_threshold, args.min_samples, logger)
reason = get_why_not_idle(
state, has_users, args.cpu_idle_threshold, args.gpu_util_threshold,
args.min_samples, state.get("gpu_idle_counter"), logger)
logger.info("Not idle (%s) — reset counter", reason)
# --- Save and exit ---
ansible/roles/auto_suspend/files/test_auto_suspend.py
+139 -101
View File
@@ -72,13 +72,11 @@ class TestFreshState(unittest.TestCase):
from auto_suspend import fresh_state
state = fresh_state()
self.assertEqual(state["idle_counter"], 0)
self.assertEqual(state["gpu_idle_counter"], 0)
self.assertEqual(state["sample_count"], 0)
self.assertIsNone(state["cpu_idle_n"])
self.assertIsNone(state["cpu_idle_n1"])
self.assertIsNone(state["gpu_util_n"])
self.assertIsNone(state["gpu_util_n1"])
self.assertIsNone(state["cpu_avg"])
self.assertIsNone(state["gpu_avg"])
class TestLoadState(unittest.TestCase):
@@ -347,77 +345,116 @@ class TestUpdateAverages(unittest.TestCase):
from auto_suspend import fresh_state, update_averages
logger = MagicMock()
state = fresh_state()
new_state = update_averages(state, cpu_n=95.0, gpu_n=2.0, alpha=0.3, logger=logger)
new_state = update_averages(state, cpu_n=95.0, gpu_n=2.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
self.assertEqual(new_state["sample_count"], 1)
self.assertAlmostEqual(new_state["cpu_avg"], 95.0)
self.assertAlmostEqual(new_state["gpu_avg"], 2.0)
self.assertEqual(new_state["gpu_idle_counter"], 1) # idle → +1
self.assertEqual(new_state["cpu_idle_n1"], None)
self.assertEqual(new_state["gpu_util_n1"], None)
def test_ema_update(self):
def test_ema_update_cpu_only(self):
"""CPU EMA updates, GPU idle counter increments."""
from auto_suspend import fresh_state, update_averages
logger = MagicMock()
state = fresh_state()
state = update_averages(state, cpu_n=90.0, gpu_n=10.0, alpha=0.3, logger=logger)
state = update_averages(state, cpu_n=95.0, gpu_n=5.0, alpha=0.3, logger=logger)
state = update_averages(state, cpu_n=90.0, gpu_n=1.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
state = update_averages(state, cpu_n=95.0, gpu_n=2.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
# EMA = 0.3 * 95 + 0.7 * 90 = 91.5
self.assertAlmostEqual(state["cpu_avg"], 91.5)
# EMA = 0.3 * 5 + 0.7 * 10 = 8.5
self.assertAlmostEqual(state["gpu_avg"], 8.5)
# GPU idle both cycles → counter = 2
self.assertEqual(state["gpu_idle_counter"], 2)
self.assertEqual(state["sample_count"], 2)
def test_gpu_busy_resets_counter(self):
"""GPU above threshold resets idle counter to 0."""
from auto_suspend import fresh_state, update_averages
logger = MagicMock()
state = fresh_state()
# Cycle 1: idle GPU → counter = 1
state = update_averages(state, cpu_n=90.0, gpu_n=1.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
self.assertEqual(state["gpu_idle_counter"], 1)
# Cycle 2: busy GPU → counter resets to 0
state = update_averages(state, cpu_n=90.0, gpu_n=50.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
self.assertEqual(state["gpu_idle_counter"], 0)
# Cycle 3: idle GPU → counter = 1
state = update_averages(state, cpu_n=90.0, gpu_n=1.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
self.assertEqual(state["gpu_idle_counter"], 1)
def test_gpu_at_threshold_is_idle(self):
"""GPU exactly at threshold counts as idle (strictly greater = busy)."""
from auto_suspend import fresh_state, update_averages
logger = MagicMock()
state = fresh_state()
# GPU at exactly 5.0 with threshold 5.0 → idle (not > threshold)
state = update_averages(state, cpu_n=90.0, gpu_n=5.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
self.assertEqual(state["gpu_idle_counter"], 1)
# GPU at 5.1 → busy
state = update_averages(state, cpu_n=90.0, gpu_n=5.1, gpu_threshold=5.0, alpha=0.3, logger=logger)
self.assertEqual(state["gpu_idle_counter"], 0)
def test_gpu_none_is_idle(self):
"""None GPU (no GPU) counts as idle."""
from auto_suspend import fresh_state, update_averages
logger = MagicMock()
state = fresh_state()
state = update_averages(state, cpu_n=90.0, gpu_n=None, gpu_threshold=5.0, alpha=0.3, logger=logger)
self.assertEqual(state["gpu_idle_counter"], 1)
def test_shift_n_to_n1(self):
from auto_suspend import fresh_state, update_averages
logger = MagicMock()
state = fresh_state()
state = update_averages(state, cpu_n=80.0, gpu_n=50.0, alpha=0.3, logger=logger)
state = update_averages(state, cpu_n=85.0, gpu_n=30.0, alpha=0.3, logger=logger)
state = update_averages(state, cpu_n=80.0, gpu_n=50.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
state = update_averages(state, cpu_n=85.0, gpu_n=30.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
self.assertEqual(state["cpu_idle_n"], 85.0)
self.assertEqual(state["cpu_idle_n1"], 80.0)
self.assertEqual(state["gpu_util_n"], 30.0)
self.assertEqual(state["gpu_util_n1"], 50.0)
def test_none_measurements_preserve_avg(self):
def test_none_cpu_preserves_avg(self):
from auto_suspend import fresh_state, update_averages
logger = MagicMock()
state = fresh_state()
state = update_averages(state, cpu_n=90.0, gpu_n=5.0, alpha=0.3, logger=logger)
state = update_averages(state, cpu_n=None, gpu_n=None, alpha=0.3, logger=logger)
state = update_averages(state, cpu_n=90.0, gpu_n=5.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
state = update_averages(state, cpu_n=None, gpu_n=None, gpu_threshold=5.0, alpha=0.3, logger=logger)
self.assertAlmostEqual(state["cpu_avg"], 90.0)
self.assertAlmostEqual(state["gpu_avg"], 5.0)
self.assertEqual(state["sample_count"], 2)
def test_ema_decay_idle_gpu(self):
"""
Simulate GPU going from 100% to 0% idle and verify EMA decays.
def test_gpu_idle_counter_survives_cpu_none(self):
"""GPU counter increments even when CPU measurement is None."""
from auto_suspend import fresh_state, update_averages
logger = MagicMock()
state = fresh_state()
state = update_averages(state, cpu_n=90.0, gpu_n=1.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
state = update_averages(state, cpu_n=None, gpu_n=0.5, gpu_threshold=5.0, alpha=0.3, logger=logger)
self.assertEqual(state["gpu_idle_counter"], 2)
self.assertAlmostEqual(state["cpu_avg"], 90.0)
This validates that the EMA calculation correctly handles the scenario
where a busy GPU becomes idle — the average should converge toward
idle values after enough consecutive idle samples.
def test_consecutive_idle_then_spike(self):
"""
Simulate the real data pattern: idle GPU with periodic 100% spikes.
Each spike resets the counter. This is the key scenario that the
old EMA approach handled poorly (spikes lingered in the average).
"""
from auto_suspend import fresh_state, update_averages
logger = MagicMock()
state = fresh_state()
# Phase 1: GPU was busy (simulate 10 consecutive 100% samples)
for _ in range(10):
state = update_averages(state, cpu_n=95.0, gpu_n=100.0, alpha=0.3, logger=logger)
# 3 idle cycles → counter = 3
for _ in range(3):
state = update_averages(state, cpu_n=99.0, gpu_n=1.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
self.assertEqual(state["gpu_idle_counter"], 3)
self.assertAlmostEqual(state["gpu_avg"], 100.0, places=1)
self.assertEqual(state["sample_count"], 10)
# 1 spike → counter resets to 0
state = update_averages(state, cpu_n=99.0, gpu_n=100.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
self.assertEqual(state["gpu_idle_counter"], 0)
# Phase 2: GPU goes idle (0%) for 20 consecutive samples
for _ in range(20):
state = update_averages(state, cpu_n=99.0, gpu_n=0.0, alpha=0.3, logger=logger)
# 3 idle cycles after spike → counter = 3
for _ in range(3):
state = update_averages(state, cpu_n=99.0, gpu_n=1.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
self.assertEqual(state["gpu_idle_counter"], 3)
# After 20 idle samples from 100%, EMA should be very close to 0
# EMA = 0.7^20 * 100 ≈ 0.08%
self.assertAlmostEqual(state["gpu_avg"], 0.08, places=1)
self.assertEqual(state["sample_count"], 30)
def test_ema_convergence_multiple_values(self):
def test_ema_cpu_convergence(self):
"""
Verify EMA converges to the steady-state value regardless of start.
Verify CPU EMA converges to the steady-state value.
With alpha=0.3 and 100 identical samples, the average should equal
the sample value (since EMA_new = 0.3*x + 0.7*x = x).
@@ -426,37 +463,11 @@ class TestUpdateAverages(unittest.TestCase):
logger = MagicMock()
state = fresh_state()
gpu_idle = 2.0
cpu_idle = 97.0
for _ in range(50):
state = update_averages(state, cpu_n=95.0, gpu_n=gpu_idle, alpha=0.3, logger=logger)
state = update_averages(state, cpu_n=cpu_idle, gpu_n=1.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
self.assertAlmostEqual(state["gpu_avg"], gpu_idle, places=2)
self.assertAlmostEqual(state["cpu_avg"], 95.0, places=2)
def test_ema_with_missing_gpu_sample(self):
"""
Verify None GPU sample preserves the average.
When measure_gpu_amd returns None (e.g., tool unavailable), the
average should NOT be updated — it should keep the last value.
"""
from auto_suspend import fresh_state, update_averages
logger = MagicMock()
state = fresh_state()
# Seed with a GPU average
state = update_averages(state, cpu_n=95.0, gpu_n=10.0, alpha=0.3, logger=logger)
self.assertAlmostEqual(state["gpu_avg"], 10.0)
# Now GPU measurement fails (None) — average should be preserved
state = update_averages(state, cpu_n=95.0, gpu_n=None, alpha=0.3, logger=logger)
self.assertAlmostEqual(state["gpu_avg"], 10.0)
self.assertEqual(state["sample_count"], 2)
# After GPU returns, EMA should continue from the preserved value
state = update_averages(state, cpu_n=95.0, gpu_n=0.0, alpha=0.3, logger=logger)
# EMA = 0.3 * 0 + 0.7 * 10 = 7.0
self.assertAlmostEqual(state["gpu_avg"], 7.0)
self.assertAlmostEqual(state["cpu_avg"], cpu_idle, places=2)
class TestEvaluate(unittest.TestCase):
@@ -466,106 +477,133 @@ class TestEvaluate(unittest.TestCase):
logger = MagicMock()
state = fresh_state()
for _ in range(3):
state = update_averages(state, cpu_n=95.0, gpu_n=1.0, alpha=0.3, logger=logger)
self.assertTrue(evaluate(state, has_users=False, idle_threshold=90.0, gpu_threshold=5.0, min_samples=3, logger=logger))
state = update_averages(state, cpu_n=95.0, gpu_n=1.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
self.assertTrue(evaluate(state, has_users=False, gpu_idle_counter=3, idle_threshold=90.0, gpu_threshold=5.0, min_samples=3, logger=logger))
def test_warmup_skips(self):
from auto_suspend import fresh_state, evaluate
logger = MagicMock()
state = fresh_state()
self.assertFalse(evaluate(state, has_users=False, idle_threshold=90.0, gpu_threshold=5.0, min_samples=3, logger=logger))
self.assertFalse(evaluate(state, has_users=False, gpu_idle_counter=0, idle_threshold=90.0, gpu_threshold=5.0, min_samples=3, logger=logger))
def test_users_logged_in(self):
from auto_suspend import fresh_state, evaluate, update_averages
logger = MagicMock()
state = fresh_state()
for _ in range(3):
state = update_averages(state, cpu_n=95.0, gpu_n=1.0, alpha=0.3, logger=logger)
self.assertFalse(evaluate(state, has_users=True, idle_threshold=90.0, gpu_threshold=5.0, min_samples=3, logger=logger))
state = update_averages(state, cpu_n=95.0, gpu_n=1.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
self.assertFalse(evaluate(state, has_users=True, gpu_idle_counter=3, idle_threshold=90.0, gpu_threshold=5.0, min_samples=3, logger=logger))
def test_cpu_not_idle(self):
from auto_suspend import fresh_state, evaluate, update_averages
logger = MagicMock()
state = fresh_state()
for _ in range(3):
state = update_averages(state, cpu_n=40.0, gpu_n=1.0, alpha=0.3, logger=logger)
self.assertFalse(evaluate(state, has_users=False, idle_threshold=90.0, gpu_threshold=5.0, min_samples=3, logger=logger))
state = update_averages(state, cpu_n=40.0, gpu_n=1.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
self.assertFalse(evaluate(state, has_users=False, gpu_idle_counter=3, idle_threshold=90.0, gpu_threshold=5.0, min_samples=3, logger=logger))
def test_gpu_busy(self):
from auto_suspend import fresh_state, evaluate, update_averages
def test_gpu_busy_resets_evaluation(self):
"""GPU busy this cycle (counter=0) blocks evaluation even if CPU is idle."""
from auto_suspend import fresh_state, update_averages, evaluate
logger = MagicMock()
state = fresh_state()
for _ in range(3):
state = update_averages(state, cpu_n=95.0, gpu_n=50.0, alpha=0.3, logger=logger)
self.assertFalse(evaluate(state, has_users=False, idle_threshold=90.0, gpu_threshold=5.0, min_samples=3, logger=logger))
state = update_averages(state, cpu_n=95.0, gpu_n=1.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
# Force GPU busy by resetting counter
self.assertFalse(evaluate(state, has_users=False, gpu_idle_counter=0, idle_threshold=90.0, gpu_threshold=5.0, min_samples=3, logger=logger))
def test_gpu_none_is_idle(self):
from auto_suspend import fresh_state, evaluate, update_averages
logger = MagicMock()
state = fresh_state()
for _ in range(3):
state = update_averages(state, cpu_n=95.0, gpu_n=None, alpha=0.3, logger=logger)
self.assertTrue(evaluate(state, has_users=False, idle_threshold=90.0, gpu_threshold=5.0, min_samples=3, logger=logger))
state = update_averages(state, cpu_n=95.0, gpu_n=None, gpu_threshold=5.0, alpha=0.3, logger=logger)
self.assertTrue(evaluate(state, has_users=False, gpu_idle_counter=3, idle_threshold=90.0, gpu_threshold=5.0, min_samples=3, logger=logger))
def test_at_threshold_cpu_below(self):
from auto_suspend import fresh_state, update_averages, evaluate
logger = MagicMock()
state = fresh_state()
for _ in range(3):
state = update_averages(state, cpu_n=90.0, gpu_n=1.0, alpha=0.3, logger=logger)
self.assertFalse(evaluate(state, has_users=False, idle_threshold=90.0, gpu_threshold=5.0, min_samples=3, logger=logger))
state = update_averages(state, cpu_n=90.0, gpu_n=1.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
self.assertFalse(evaluate(state, has_users=False, gpu_idle_counter=3, idle_threshold=90.0, gpu_threshold=5.0, min_samples=3, logger=logger))
def test_at_threshold_gpu_above(self):
def test_gpu_at_threshold_is_idle(self):
"""GPU at exactly threshold should NOT block evaluation."""
from auto_suspend import fresh_state, update_averages, evaluate
logger = MagicMock()
state = fresh_state()
for _ in range(3):
state = update_averages(state, cpu_n=95.0, gpu_n=5.0, alpha=0.3, logger=logger)
self.assertFalse(evaluate(state, has_users=False, idle_threshold=90.0, gpu_threshold=5.0, min_samples=3, logger=logger))
state = update_averages(state, cpu_n=95.0, gpu_n=5.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
self.assertTrue(evaluate(state, has_users=False, gpu_idle_counter=3, idle_threshold=90.0, gpu_threshold=5.0, min_samples=3, logger=logger))
def test_gpu_above_threshold_blocks(self):
"""GPU above threshold should block evaluation."""
from auto_suspend import fresh_state, update_averages, evaluate
logger = MagicMock()
state = fresh_state()
state = update_averages(state, cpu_n=95.0, gpu_n=50.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
self.assertFalse(evaluate(state, has_users=False, gpu_idle_counter=0, idle_threshold=90.0, gpu_threshold=5.0, min_samples=1, logger=logger))
def test_custom_thresholds(self):
from auto_suspend import fresh_state, update_averages, evaluate
logger = MagicMock()
state = fresh_state()
for _ in range(3):
state = update_averages(state, cpu_n=95.0, gpu_n=3.0, alpha=0.3, logger=logger)
self.assertTrue(evaluate(state, has_users=False, idle_threshold=90.0, gpu_threshold=5.0, min_samples=3, logger=logger))
self.assertFalse(evaluate(state, has_users=False, idle_threshold=95.0, gpu_threshold=5.0, min_samples=3, logger=logger))
state = update_averages(state, cpu_n=95.0, gpu_n=3.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
self.assertTrue(evaluate(state, has_users=False, gpu_idle_counter=3, idle_threshold=90.0, gpu_threshold=5.0, min_samples=3, logger=logger))
self.assertFalse(evaluate(state, has_users=False, gpu_idle_counter=3, idle_threshold=95.0, gpu_threshold=5.0, min_samples=3, logger=logger))
def test_gpu_busy_during_warmup(self):
"""GPU busy during warmup: counter=0 but still in warmup."""
from auto_suspend import fresh_state, evaluate, update_averages
logger = MagicMock()
state = fresh_state()
# Only 1 sample, counter=0 because GPU was busy
state = update_averages(state, cpu_n=95.0, gpu_n=100.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
self.assertFalse(evaluate(state, has_users=False, gpu_idle_counter=0, idle_threshold=90.0, gpu_threshold=5.0, min_samples=3, logger=logger))
class TestGetWhyNotIdle(unittest.TestCase):
def test_users(self):
from auto_suspend import get_why_not_idle
state = {"sample_count": 5, "cpu_avg": 95.0, "gpu_avg": 1.0}
state = {"sample_count": 5, "cpu_avg": 95.0, "gpu_idle_counter": 3}
reason = get_why_not_idle(state, has_users=True, logger=MagicMock())
self.assertIn("users logged in", reason)
def test_cpu_low(self):
from auto_suspend import get_why_not_idle
state = {"sample_count": 5, "cpu_avg": 40.0, "gpu_avg": 1.0}
state = {"sample_count": 5, "cpu_avg": 40.0, "gpu_idle_counter": 3}
reason = get_why_not_idle(state, has_users=False, logger=MagicMock())
self.assertIn("CPU idle avg", reason)
def test_gpu_high(self):
def test_gpu_busy(self):
from auto_suspend import get_why_not_idle
state = {"sample_count": 5, "cpu_avg": 95.0, "gpu_avg": 50.0}
reason = get_why_not_idle(state, has_users=False, logger=MagicMock())
self.assertIn("GPU util avg", reason)
state = {"sample_count": 5, "cpu_avg": 95.0, "gpu_idle_counter": 0}
reason = get_why_not_idle(state, has_users=False, gpu_idle_counter=0, logger=MagicMock())
self.assertIn("GPU busy", reason)
def test_gpu_idle_no_reason(self):
"""GPU idle counter > 0 should not add a reason."""
from auto_suspend import get_why_not_idle
state = {"sample_count": 5, "cpu_avg": 95.0, "gpu_idle_counter": 3}
reason = get_why_not_idle(state, has_users=False, gpu_idle_counter=3, logger=MagicMock())
self.assertNotIn("GPU busy", reason)
def test_warming_up(self):
from auto_suspend import get_why_not_idle
state = {"sample_count": 2, "cpu_avg": None, "gpu_avg": None}
reason = get_why_not_idle(state, has_users=False, logger=MagicMock())
state = {"sample_count": 2, "cpu_avg": None, "gpu_idle_counter": 0}
reason = get_why_not_idle(state, has_users=False, gpu_idle_counter=0, logger=MagicMock())
self.assertIn("warming up", reason)
def test_multiple_reasons(self):
from auto_suspend import get_why_not_idle
state = {"sample_count": 5, "cpu_avg": 40.0, "gpu_avg": 50.0}
reason = get_why_not_idle(state, has_users=True, logger=MagicMock())
state = {"sample_count": 5, "cpu_avg": 40.0, "gpu_idle_counter": 0}
reason = get_why_not_idle(state, has_users=True, gpu_idle_counter=0, logger=MagicMock())
self.assertIn("users logged in", reason)
self.assertIn("CPU idle avg", reason)
self.assertIn("GPU util avg", reason)
self.assertIn("GPU busy", reason)
class TestSetupLogging(unittest.TestCase):