Detect unusable perf counters

Author: vrslev

README.md

@@ -162,9 +162,12 @@ instruction counts and RSS growth remain independently guarded.
 
 ## Measurement scope and limitations
 
-- Linux instruction counting uses `perf_event_open`. Rootless or restricted
-  runners commonly deny it; the artifact then says `process_time`, never
-  `instructions`.
+- Linux instruction counting uses `perf_event_open`. Each pytest process runs
+  one brief startup calibration and validates the counter's scheduled time on
+  every read. Access denial, a zero/nonfunctional counter, no scheduled time,
+  or less than 90% scheduled time causes an automatic `process_time` fallback;
+  the artifact records the exact unavailable reason. This avoids both false
+  zero-instruction baselines and noisy heavily multiplexed baselines.
 - CPU time is process CPU consumed during each pytest call. Hardware
   instructions count only the thread running that call. The artifact records
   these scopes explicitly. Summed job and session CPU remain guarded so

src/pytest_perfguard/metrics.py

@@ -29,8 +29,13 @@
 _PERF_EVENT_IOC_ENABLE = 0x2400
 _PERF_EVENT_IOC_DISABLE = 0x2401
 _PERF_EVENT_IOC_RESET = 0x2403
+_PERF_FORMAT_TOTAL_TIME_ENABLED = 1 << 0
+_PERF_FORMAT_TOTAL_TIME_RUNNING = 1 << 1
 _PERF_FLAG_FD_CLOEXEC = 1 << 3
-_COUNTER_VALUE_SIZE = struct.calcsize("=Q")
+_COUNTER_READ_FORMAT = "=QQQ"
+_COUNTER_VALUE_SIZE = struct.calcsize(_COUNTER_READ_FORMAT)
+_MIN_COUNTER_RUNNING_PERCENT = 90
+_CALIBRATION_ITERATIONS = 20_000
 
 _ATTR_DISABLED = 1 << 0
 _ATTR_EXCLUDE_KERNEL = 1 << 5
@@ -64,11 +69,50 @@ def _perf_event_syscall_number() -> int | None:
     }.get(platform.machine().lower())
 
 
+def _normalize_instruction_count(value: int, time_enabled: int, time_running: int) -> int:
+    """Validate and scale a perf counter reading."""
+    if time_running == 0:
+        raise OSError("perf hardware counter was not scheduled (time_running=0)")
+    if time_running > time_enabled:
+        raise OSError(
+            "perf hardware counter returned inconsistent timing "
+            f"(time_enabled={time_enabled}, time_running={time_running})"
+        )
+    if value == 0:
+        raise OSError("perf hardware instruction counter returned zero despite scheduled work")
+    if time_running * 100 < time_enabled * _MIN_COUNTER_RUNNING_PERCENT:
+        raise OSError(
+            "perf hardware counter was heavily multiplexed "
+            f"(time_enabled={time_enabled}, time_running={time_running}, "
+            f"minimum_running={_MIN_COUNTER_RUNNING_PERCENT}%)"
+        )
+    return value * time_enabled // time_running
+
+
+def _instruction_calibration_workload() -> int:
+    """Execute enough deterministic Python work to prove that a counter advances."""
+    checksum = 0
+    for value in range(_CALIBRATION_ITERATIONS):
+        checksum = (checksum + value) & 0xFFFFFFFF
+    return checksum
+
+
+def _errno_details(error_number: int) -> str:
+    error_name = errno.errorcode.get(error_number, "UNKNOWN")
+    try:
+        error_message = os.strerror(error_number)
+    except ValueError:
+        error_message = "unknown error"
+    return f"errno={error_number} {error_name}: {error_message}"
+
+
 class _InstructionCounter:
     """A Linux perf-event counter attached to the current test thread."""
 
     def __init__(self, file_descriptor: int) -> None:
         self._file_descriptor = file_descriptor
+        self._time_enabled = 0
+        self._time_running = 0
 
     @classmethod
     def open(cls) -> tuple[_InstructionCounter | None, str | None]:
@@ -80,6 +124,7 @@ def open(cls) -> tuple[_InstructionCounter | None, str | None]:
         attributes.type = _PERF_TYPE_HARDWARE
         attributes.size = ctypes.sizeof(_PerfEventAttr)
         attributes.config = _PERF_COUNT_HW_INSTRUCTIONS
+        attributes.read_format = _PERF_FORMAT_TOTAL_TIME_ENABLED | _PERF_FORMAT_TOTAL_TIME_RUNNING
         attributes.flags = _ATTR_DISABLED | _ATTR_EXCLUDE_KERNEL | _ATTR_EXCLUDE_HYPERVISOR
 
         libc = ctypes.CDLL(None, use_errno=True)
@@ -94,9 +139,10 @@ def open(cls) -> tuple[_InstructionCounter | None, str | None]:
         )
         if file_descriptor < 0:
             error_number = ctypes.get_errno()
+            error_details = _errno_details(error_number)
             if error_number in {errno.EACCES, errno.EPERM}:
-                return None, "perf_event_open is not permitted by this runner"
-            return None, f"perf_event_open failed with errno {error_number}"
+                return None, f"perf_event_open is not permitted by this runner ({error_details})"
+            return None, f"perf_event_open failed ({error_details})"
         return cls(int(file_descriptor)), None
 
     def start(self) -> None:
@@ -112,7 +158,14 @@ def stop(self) -> int:
         raw_value = os.read(self._file_descriptor, _COUNTER_VALUE_SIZE)
         if len(raw_value) != _COUNTER_VALUE_SIZE:
             raise OSError("perf event returned an incomplete counter value")
-        return int(struct.unpack("=Q", raw_value)[0])
+        value, time_enabled, time_running = struct.unpack(_COUNTER_READ_FORMAT, raw_value)
+        if time_enabled < self._time_enabled or time_running < self._time_running:
+            raise OSError("perf hardware counter cumulative timing decreased")
+        enabled_delta = time_enabled - self._time_enabled
+        running_delta = time_running - self._time_running
+        self._time_enabled = time_enabled
+        self._time_running = time_running
+        return _normalize_instruction_count(value, enabled_delta, running_delta)
 
     def close(self) -> None:
         os.close(self._file_descriptor)
@@ -135,6 +188,13 @@ class CpuCounter:
 
     def __init__(self) -> None:
         self._instructions, self.unavailable_reason = _InstructionCounter.open()
+        if self._instructions is not None:
+            try:
+                self._instructions.start()
+                _instruction_calibration_workload()
+                self._instructions.stop()
+            except OSError as error:
+                self._disable_instructions(f"perf hardware counter calibration failed: {error}")
 
     @property
     def backend(self) -> str:

src/pytest_perfguard/plugin.py

@@ -127,14 +127,7 @@ def pytest_sessionfinish(self, session: pytest.Session, exitstatus: int | pytest
             worker_environment = worker_fragment.get("environment")
             if isinstance(worker_environment, dict):
                 environments.append(worker_environment)
-        environment = dict(environments[0]) if environments else self._environment()
-        backends = {str(worker_environment.get("counter_backend")) for worker_environment in environments}
-        if len(backends) > 1:
-            environment = {
-                **environment,
-                "counter_backend": "mixed",
-                "cpu_instructions_scope": "mixed",
-            }
+        environment = _merge_environments(environments or [self._environment()])
 
         payload = {
             "schema_version": SCHEMA_VERSION,
@@ -199,6 +192,7 @@ def _environment(self) -> dict[str, object]:
         return {
             "architecture": platform.machine() or "unknown",
             "counter_backend": self._cpu_counter.backend,
+            "cpu_instructions_unavailable_reason": self._cpu_counter.unavailable_reason,
             "cpu_instructions_scope": instruction_scope,
             "cpu_model": cpu_model(),
             "cpu_time_scope": "pytest_call_process",
@@ -209,6 +203,25 @@ def _environment(self) -> dict[str, object]:
         }
 
 
+def _merge_environments(environments: list[dict[str, object]]) -> dict[str, object]:
+    environment = dict(environments[0])
+    backends = {str(worker_environment.get("counter_backend")) for worker_environment in environments}
+    scopes = {str(worker_environment.get("cpu_instructions_scope")) for worker_environment in environments}
+    reasons = sorted(
+        {
+            reason
+            for worker_environment in environments
+            if isinstance((reason := worker_environment.get("cpu_instructions_unavailable_reason")), str) and reason
+        }
+    )
+    environment["cpu_instructions_unavailable_reason"] = " | ".join(reasons) if reasons else None
+    if len(backends) > 1:
+        environment["counter_backend"] = "mixed"
+    if len(scopes) > 1:
+        environment["cpu_instructions_scope"] = "mixed"
+    return environment
+
+
 def _aggregate_job_metrics(
     fragments: list[dict[str, Any]],
     tests: list[dict[str, Any]],

tests/test_compare.py

@@ -207,6 +207,14 @@ def test_incompatible_runner_uses_a_separate_bucket() -> None:
     assert environment_fingerprint(current["environment"]) not in baseline["jobs"]["test"]["buckets"]
 
 
+def test_instruction_unavailable_reason_does_not_split_environment_bucket() -> None:
+    current = _current()
+    fingerprint = environment_fingerprint(current["environment"])
+    current["environment"]["cpu_instructions_unavailable_reason"] = "errno=1 EPERM"
+
+    assert environment_fingerprint(current["environment"]) == fingerprint
+
+
 def test_record_mode_keeps_rolling_history_but_not_a_moving_anchor() -> None:
     baseline = None
     for index in range(25):

tests/test_plugin.py

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
 import ctypes
+import errno
 import json
 import os
+import struct
 import subprocess
 import sys
 from pathlib import Path
@@ -34,6 +36,105 @@ def test_process_metrics_are_available() -> None:
     assert memory_backend() in {"peak_rss_fallback", "proc_rss", "windows_working_set"}
 
 
+@pytest.mark.parametrize(
+    ("value", "time_enabled", "time_running", "expected"),
+    [
+        (123, 1_000, 1_000, 123),
+        (900, 1_000, 900, 1_000),
+    ],
+)
+def test_instruction_count_normalizes_scheduled_time(
+    value: int,
+    time_enabled: int,
+    time_running: int,
+    expected: int,
+) -> None:
+    assert metrics_module._normalize_instruction_count(value, time_enabled, time_running) == expected  # noqa: SLF001
+
+
+@pytest.mark.parametrize(
+    ("value", "time_enabled", "time_running", "message"),
+    [
+        (100, 1_000, 899, "heavily multiplexed"),
+        (100, 1_000, 0, "not scheduled"),
+        (0, 1_000, 1_000, "returned zero"),
+        (100, 999, 1_000, "inconsistent timing"),
+    ],
+)
+def test_instruction_count_rejects_unreliable_readings(
+    value: int,
+    time_enabled: int,
+    time_running: int,
+    message: str,
+) -> None:
+    with pytest.raises(OSError, match=message):
+        metrics_module._normalize_instruction_count(value, time_enabled, time_running)  # noqa: SLF001
+
+
+def test_instruction_counter_uses_per_measurement_scheduling_time(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(metrics_module, "fcntl", SimpleNamespace(ioctl=mock.Mock()))
+    monkeypatch.setattr(
+        os,
+        "read",
+        mock.Mock(
+            side_effect=[
+                struct.pack("=QQQ", 100, 1_000, 1_000),
+                struct.pack("=QQQ", 90, 2_000, 1_900),
+            ]
+        ),
+    )
+    counter = metrics_module._InstructionCounter(42)  # noqa: SLF001
+
+    assert counter.stop() == 100
+    assert counter.stop() == 100
+
+
+def test_instruction_calibration_failure_falls_back_to_process_time() -> None:
+    instruction_counter = mock.Mock(spec=metrics_module._InstructionCounter)  # noqa: SLF001
+    instruction_counter.stop.side_effect = OSError("perf hardware instruction counter returned zero")
+
+    with mock.patch.object(
+        metrics_module._InstructionCounter,  # noqa: SLF001
+        "open",
+        return_value=(instruction_counter, None),
+    ):
+        counter = CpuCounter()
+
+    assert counter.backend == "process_time"
+    assert counter.unavailable_reason == (
+        "perf hardware counter calibration failed: perf hardware instruction counter returned zero"
+    )
+    instruction_counter.start.assert_called_once_with()
+    instruction_counter.stop.assert_called_once_with()
+    instruction_counter.close.assert_called_once_with()
+
+
+@pytest.mark.parametrize(
+    ("error_number", "error_name"),
+    [
+        (errno.EPERM, "EPERM"),
+        (errno.EACCES, "EACCES"),
+    ],
+)
+def test_perf_event_open_reason_preserves_errno(
+    monkeypatch: pytest.MonkeyPatch,
+    error_number: int,
+    error_name: str,
+) -> None:
+    syscall = mock.Mock(return_value=-1)
+    libc = SimpleNamespace(syscall=syscall)
+    monkeypatch.setattr(metrics_module, "_perf_event_syscall_number", lambda: 298)
+    monkeypatch.setattr(metrics_module, "fcntl", SimpleNamespace())
+    monkeypatch.setattr(ctypes, "CDLL", lambda *_args, **_kwargs: libc)
+    monkeypatch.setattr(ctypes, "get_errno", lambda: error_number)
+
+    counter, reason = metrics_module._InstructionCounter.open()  # noqa: SLF001
+
+    assert counter is None
+    assert reason is not None
+    assert f"errno={error_number} {error_name}" in reason
+
+
 def test_runner_key_prefers_override_then_gitlab_identity(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setenv("CI_RUNNER_ID", "runner-123")
     monkeypatch.setenv("CI_RUNNER_DESCRIPTION", "shared-medium-host")
@@ -102,6 +203,12 @@ def test_plugin_writes_one_current_artifact(tmp_path: Path) -> None:
     assert record["outcome"] == "passed"
     assert record["metrics"]["cpu_time_ns"] > 0
     assert payload["environment"]["counter_backend"] in {"perf_event_open", "process_time"}
+    unavailable_reason = payload["environment"]["cpu_instructions_unavailable_reason"]
+    if payload["environment"]["counter_backend"] == "process_time":
+        assert isinstance(unavailable_reason, str)
+        assert unavailable_reason
+    else:
+        assert unavailable_reason is None
     assert payload["environment"]["cpu_time_scope"] == "pytest_call_process"
     assert payload["environment"]["memory_scope"] == "process_rss_boundaries"
     assert payload["environment"]["runner_key"] == "runner-123"
@@ -133,6 +240,32 @@ def test_xdist_workers_are_merged_into_one_artifact(tmp_path: Path) -> None:
     assert payload["job_metrics"]["cpu_time_ns"] == sum(test["metrics"]["cpu_time_ns"] for test in payload["tests"])
 
 
+def test_xdist_environment_merges_counter_backends_and_reasons() -> None:
+    environment = plugin_module._merge_environments(  # noqa: SLF001
+        [
+            {
+                "counter_backend": "perf_event_open",
+                "cpu_instructions_scope": "current_thread",
+                "cpu_instructions_unavailable_reason": None,
+            },
+            {
+                "counter_backend": "process_time",
+                "cpu_instructions_scope": "unavailable",
+                "cpu_instructions_unavailable_reason": "errno=1 EPERM",
+            },
+            {
+                "counter_backend": "process_time",
+                "cpu_instructions_scope": "unavailable",
+                "cpu_instructions_unavailable_reason": "counter returned zero",
+            },
+        ]
+    )
+
+    assert environment["counter_backend"] == "mixed"
+    assert environment["cpu_instructions_scope"] == "mixed"
+    assert environment["cpu_instructions_unavailable_reason"] == "counter returned zero | errno=1 EPERM"
+
+
 def test_source_fingerprint_is_scoped_to_the_test_function(tmp_path: Path) -> None:
     first_artifact = tmp_path / "first.json"
     second_artifact = tmp_path / "second.json"