Signed-off-by: Yaxiong Tian <[email protected]>
---
tools/testing/selftests/Makefile | 1 +
tools/testing/selftests/cpuidle/Makefile | 6 +
tools/testing/selftests/cpuidle/config | 3 +
.../cpuidle/cpuidle_latency_req_qos.py | 331 ++++++++++++++++++
tools/testing/selftests/cpuidle/settings | 2 +
5 files changed, 343 insertions(+)
create mode 100644 tools/testing/selftests/cpuidle/Makefile
create mode 100644 tools/testing/selftests/cpuidle/config
create mode 100755 tools/testing/selftests/cpuidle/cpuidle_latency_req_qos.py
create mode 100644 tools/testing/selftests/cpuidle/settings
diff --git a/tools/testing/selftests/Makefile b/tools/testing/selftests/Makefile
index 8189f333814c..a8fb620bebd1 100644
--- a/tools/testing/selftests/Makefile
+++ b/tools/testing/selftests/Makefile
@@ -12,6 +12,7 @@ TARGETS += clone3
TARGETS += connector
TARGETS += core
TARGETS += cpufreq
+TARGETS += cpuidle
TARGETS += cpu-hotplug
TARGETS += damon
TARGETS += devices/error_logs
diff --git a/tools/testing/selftests/cpuidle/Makefile
b/tools/testing/selftests/cpuidle/Makefile
new file mode 100644
index 000000000000..f960d72c3a65
--- /dev/null
+++ b/tools/testing/selftests/cpuidle/Makefile
@@ -0,0 +1,6 @@
+# SPDX-License-Identifier: GPL-2.0
+all:
+
+TEST_PROGS := cpuidle_latency_req_qos.py
+
+include ../lib.mk
diff --git a/tools/testing/selftests/cpuidle/config
b/tools/testing/selftests/cpuidle/config
new file mode 100644
index 000000000000..86e8f87d4e62
--- /dev/null
+++ b/tools/testing/selftests/cpuidle/config
@@ -0,0 +1,3 @@
+CONFIG_CPU_IDLE=y
+CONFIG_CPU_IDLE_GOV_MENU=y
+CONFIG_PM_QOS_CPU_SYSTEM_WAKEUP=y
diff --git a/tools/testing/selftests/cpuidle/cpuidle_latency_req_qos.py
b/tools/testing/selftests/cpuidle/cpuidle_latency_req_qos.py
new file mode 100755
index 000000000000..875b623da530
--- /dev/null
+++ b/tools/testing/selftests/cpuidle/cpuidle_latency_req_qos.py
@@ -0,0 +1,331 @@
+#!/usr/bin/env python3
+# SPDX-License-Identifier: GPL-2.0
+"""
+cpuidle: verify latency_req QoS ceilings restrict idle-state selection.
+
+Constrains exit latency via each of the three QoS inputs and checks that
+cpuidle states whose exit latency exceeds the ceiling do not gain usage:
+
+ 1) /dev/cpu_dma_latency
+ 2) /dev/cpu_wakeup_latency
+ 3) /sys/devices/system/cpu/cpuN/power/pm_qos_resume_latency_us
+"""
+
+from __future__ import annotations
+
+import glob
+import os
+import struct
+import sys
+import time
+from dataclasses import dataclass
+from typing import Dict, List, Optional, Tuple
+
+# Source tree: tools/testing/selftests/cpuidle/../kselftest
+# Install tree: kselftest_install/cpuidle/../kselftest
+sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)),
+ "..", "kselftest"))
+import ksft
+
+
+CPUIDLE_BASE = "/sys/devices/system/cpu"
+DMA_LAT_DEV = "/dev/cpu_dma_latency"
+WAKEUP_LAT_DEV = "/dev/cpu_wakeup_latency"
+
+# Keep windows short so the whole collection fits under settings timeout.
+IDLE_SEC = 2.0
+CPU = 0
+
+
+@dataclass
+class IdleState:
+ index: int
+ name: str
+ latency_us: int
+ residency_us: int
+ usage: int
+ time_us: int
+
+
+def read_states(cpu: int) -> Dict[int, IdleState]:
+ base = f"{CPUIDLE_BASE}/cpu{cpu}/cpuidle"
+ states: Dict[int, IdleState] = {}
+ paths = sorted(
+ glob.glob(f"{base}/state*"),
+ key=lambda p: int(os.path.basename(p).replace("state", "")),
+ )
+ for path in paths:
+ idx = int(os.path.basename(path).replace("state", ""))
+ states[idx] = IdleState(
+ index=idx,
+ name=open(f"{path}/name").read().strip(),
+ latency_us=int(open(f"{path}/latency").read()),
+ residency_us=int(open(f"{path}/residency").read()),
+ usage=int(open(f"{path}/usage").read()),
+ time_us=int(open(f"{path}/time").read()),
+ )
+ return states
+
+
+def usage_delta(before: Dict[int, IdleState],
+ after: Dict[int, IdleState]) -> Dict[int, int]:
+ return {i: after[i].usage - before[i].usage for i in before}
+
+
+def pick_ceilings(states: Dict[int, IdleState]) -> List[int]:
+ nonzero = sorted({s.latency_us for s in states.values() if s.latency_us
> 0})
+ out: List[int] = []
+ if nonzero:
+ out.append(nonzero[0])
+ if len(nonzero) >= 2:
+ mid = (nonzero[0] + nonzero[1]) // 2
+ out.append(mid if mid > nonzero[0] else max(1, nonzero[1] - 1))
+ if len(nonzero) >= 3:
+ out.append((nonzero[1] + nonzero[2]) // 2)
+ # Dedup, keep order
+ seen = set()
+ uniq = []
+ for c in out:
+ if c not in seen and c >= 1:
+ seen.add(c)
+ uniq.append(c)
+ return uniq or [1]
+
+
+def idle_on_cpu(seconds: float) -> None:
+ end = time.monotonic() + seconds
+ burn_end = time.monotonic() + min(0.1, seconds / 10)
+ while time.monotonic() < burn_end:
+ pass
+ while time.monotonic() < end:
+ time.sleep(0.05)
+
+
+def allowed_states(states: Dict[int, IdleState], ceiling: int) -> List[int]:
+ return [i for i, s in states.items() if s.latency_us <= ceiling]
+
+
+def forbidden_states(states: Dict[int, IdleState], ceiling: int) -> List[int]:
+ return [i for i, s in states.items() if s.latency_us > ceiling]
+
+
+def fmt_state_list(states: Dict[int, IdleState], idxs: List[int]) -> str:
+ if not idxs:
+ return "(none)"
+ return ", ".join(
+ f"state{i}:{states[i].name}(lat={states[i].latency_us})"
+ for i in idxs
+ )
+
+
+def forbidden_violations(states: Dict[int, IdleState], udelta: Dict[int, int],
+ ceiling: int) -> List[str]:
+ """States that must not be entered but still gained usage."""
+ bad = []
+ for i in forbidden_states(states, ceiling):
+ if udelta[i] > 0:
+ s = states[i]
+ bad.append(
+ f"state{i}({s.name},lat={s.latency_us})
usage+={udelta[i]}"
+ )
+ return bad
+
+
+def print_usage_table(states: Dict[int, IdleState], udelta: Dict[int, int],
+ ceiling: int) -> None:
+ ksft.print_msg(
+ f"{'idx':>3} {'name':<12} {'lat':>6} {'d_usage':>8}
{'expect':>8}"
+ )
+ for i in sorted(states):
+ s = states[i]
+ expect = "allow" if s.latency_us <= ceiling else "forbid"
+ ksft.print_msg(
+ f"{i:3d} {s.name:<12} {s.latency_us:6d} {udelta[i]:8d}
{expect:>8}"
+ )
+
+
+class DmaLatencyGuard:
+ def __init__(self, latency_us: int):
+ self.latency_us = latency_us
+ self.fd = -1
+
+ def __enter__(self):
+ self.fd = os.open(DMA_LAT_DEV, os.O_RDWR)
+ os.write(self.fd, struct.pack("i", int(self.latency_us)))
+ return self
+
+ def __exit__(self, *args):
+ if self.fd >= 0:
+ os.close(self.fd)
+ self.fd = -1
+
+
+class WakeupLatencyGuard:
+ def __init__(self, latency_us: int):
+ self.latency_us = latency_us
+ self.fd = -1
+
+ def __enter__(self):
+ self.fd = os.open(WAKEUP_LAT_DEV, os.O_RDWR)
+ os.write(self.fd, struct.pack("i", int(self.latency_us)))
+ return self
+
+ def __exit__(self, *args):
+ if self.fd >= 0:
+ os.close(self.fd)
+ self.fd = -1
+
+
+class ResumeLatencyGuard:
+ def __init__(self, cpu: int, latency_us: int):
+ self.path =
f"{CPUIDLE_BASE}/cpu{cpu}/power/pm_qos_resume_latency_us"
+ self.latency_us = latency_us
+ self.prev: Optional[str] = None
+
+ def __enter__(self):
+ self.prev = open(self.path).read().strip()
+ with open(self.path, "w") as f:
+ f.write(f"{int(self.latency_us)}\n")
+ return self
+
+ def __exit__(self, *args):
+ restore = "0" if self.prev in ("0", "n/a", None) else self.prev
