sysmon: fix first-sample handling for interval and cpuUsage

Author: Guy Suday

pymobiledevice3/cli/developer/dvt/sysmon/process.py

@@ -160,12 +160,16 @@ def _duration_elapsed(start_time: float, duration_ms: Optional[int]) -> bool:
     return ((asyncio.get_running_loop().time() - start_time) * 1000) >= duration_ms
 
 
-async def iter_initialized_processes(sysmon: Sysmontap):
-    sample_index = 0
+def _should_skip_first_snapshot(keys: Optional[list[str]]) -> bool:
+    # The first sample does not contain initialized cpuUsage values
+    return (keys is None) or ("cpuUsage" in keys)
+
+
+async def iter_processes(sysmon: Sysmontap, skip_first_snapshot: bool = False):
+    should_skip_snapshot = skip_first_snapshot
     async for process_snapshot in sysmon.iter_processes():
-        sample_index += 1
-        # The first sample does not contain initialized cpuUsage values.
-        if sample_index < 2:
+        if should_skip_snapshot:
+            should_skip_snapshot = False
             continue
         yield process_snapshot
 
@@ -240,7 +244,9 @@ async def _select_process_from_sysmon(
     dvt, parsed_filters: dict[str, list[str]], keys: Optional[list[str]], selection_mode: ProcessSelectionMode
 ) -> dict:
     async with await Sysmontap.create(dvt) as selection_sysmon:
-        async for process_snapshot in iter_initialized_processes(selection_sysmon):
+        async for process_snapshot in iter_processes(
+            selection_sysmon, skip_first_snapshot=_should_skip_first_snapshot(keys)
+        ):
             # All process entries in a sysmon sample share the same schema, so validating one entry is sufficient.
             _validate_process_keys(process_snapshot[0], keys or [])
             return _select_process_from_snapshot(process_snapshot, parsed_filters, selection_mode)
@@ -304,7 +310,7 @@ async def sysmon_process_single_task(
         DeviceInfo(dvt) as device_info,
         await Sysmontap.create(dvt) as sysmon,
     ):
-        async for process_snapshot in iter_initialized_processes(sysmon):
+        async for process_snapshot in iter_processes(sysmon, skip_first_snapshot=_should_skip_first_snapshot(keys)):
             if process_snapshot and parsed_filters:
                 # All process entries in a sysmon sample share the same schema, so validating one entry is sufficient.
                 _validate_process_keys(process_snapshot[0], list(parsed_filters))
@@ -398,7 +404,7 @@ async def sysmon_process_monitor_process(
             "-i",
             help="Minimum interval in milliseconds between outputs (optional)",
         ),
-    ] = Sysmontap.DEFAULT_INTERVAL,
+    ] = Sysmontap.DEFAULT_INTERVAL_MS,
     duration: Annotated[
         Optional[int],
         typer.Option(
@@ -456,7 +462,7 @@ async def sysmon_process_monitor_threshold_task(
     start_time = None
 
     async with DvtProvider(service_provider) as dvt, await Sysmontap.create(dvt) as sysmon:
-        async for process_snapshot in iter_initialized_processes(sysmon):
+        async for process_snapshot in iter_processes(sysmon, skip_first_snapshot=True):
             if start_time is None:
                 start_time = asyncio.get_running_loop().time()
 
@@ -481,7 +487,7 @@ async def sysmon_process_monitor_threshold_task(
 async def sysmon_process_monitor_process_task(
     service_provider: ServiceProviderDep,
     filter_expressions: Optional[list[str]] = None,
-    interval: int = Sysmontap.DEFAULT_INTERVAL,
+    interval: int = Sysmontap.DEFAULT_INTERVAL_MS,
     duration: Optional[int] = None,
     choose: ProcessSelectionMode = ProcessSelectionMode.PROMPT,
     keys: Optional[list[str]] = None,
@@ -501,7 +507,9 @@ async def sysmon_process_monitor_process_task(
 
         monitoring_start_time = None
         async with await Sysmontap.create(dvt, interval=interval) as monitor_sysmon:
-            async for process_snapshot in iter_initialized_processes(monitor_sysmon):
+            async for process_snapshot in iter_processes(
+                monitor_sysmon, skip_first_snapshot=_should_skip_first_snapshot(keys)
+            ):
                 if monitoring_start_time is None:
                     monitoring_start_time = asyncio.get_running_loop().time()
 

pymobiledevice3/services/dvt/instruments/sysmontap.py

@@ -6,32 +6,37 @@
 
 class Sysmontap(Tap):
     IDENTIFIER = "com.apple.instruments.server.services.sysmontap"
-    DEFAULT_INTERVAL = 500
+    DEFAULT_INTERVAL_MS = 500
+    MINIMUM_INTERVAL_MS = 1
 
     def __init__(
-        self, dvt, process_attributes: list[str], system_attributes: list[str], interval: int = DEFAULT_INTERVAL
+        self,
+        dvt,
+        process_attributes: list[str],
+        system_attributes: list[str],
+        interval_ms: int = DEFAULT_INTERVAL_MS,
     ) -> None:
         self.process_attributes_cls = dataclasses.make_dataclass("SysmonProcessAttributes", process_attributes)
         self.system_attributes_cls = dataclasses.make_dataclass("SysmonSystemAttributes", system_attributes)
 
         config = {
-            "ur": interval,  # Output frequency ms
+            "ur": Sysmontap.MINIMUM_INTERVAL_MS,  # Output frequency ms
             "bm": 0,
             "procAttrs": process_attributes,
             "sysAttrs": system_attributes,
             "cpuUsage": True,
             "physFootprint": True,  # memory value
-            "sampleInterval": interval * 1000000,
+            "sampleInterval": interval_ms * 1_000_000,
         }
 
         super().__init__(dvt, self.IDENTIFIER, config)
 
     @classmethod
-    async def create(cls, dvt, interval: int = DEFAULT_INTERVAL) -> "Sysmontap":
+    async def create(cls, dvt, interval: int = DEFAULT_INTERVAL_MS) -> "Sysmontap":
         async with DeviceInfo(dvt) as device_info:
             process_attributes = list(await device_info.sysmon_process_attributes())
             system_attributes = list(await device_info.sysmon_system_attributes())
-        return cls(dvt, process_attributes, system_attributes, interval=interval)
+        return cls(dvt, process_attributes, system_attributes, interval_ms=interval)
 
     async def iter_processes(self):
         async for row in self:

tests/cli/developer/dvt/sysmon/test_process.py

@@ -8,7 +8,7 @@
     ProcessSelectionMode,
     _process_sort_key,
     _select_process_from_snapshot,
-    iter_initialized_processes,
+    iter_processes,
     sysmon_process_monitor_threshold_task,
     sysmon_process_single_task,
 )
@@ -18,14 +18,14 @@
 @pytest_asyncio.fixture
 async def process_snapshot(dvt) -> list[dict]:
     async with await Sysmontap.create(dvt) as sysmon:
-        async for process_snapshot in iter_initialized_processes(sysmon):
+        async for process_snapshot in iter_processes(sysmon):
             return process_snapshot
 
     pytest.fail("failed to collect an initialized process snapshot")
 
 
 @pytest.mark.asyncio
-async def test_iter_initialized_processes_yields_process_dicts(process_snapshot) -> None:
+async def test_iter_processes_yields_process_dicts(process_snapshot) -> None:
     assert len(process_snapshot) > 0
     assert isinstance(process_snapshot[0], dict)
     assert process_snapshot[0].get("pid")

tests/cli/developer/dvt/sysmon/test_process_unit.py

@@ -22,13 +22,41 @@
     _select_process_from_sysmon,
     _select_process_output_keys,
     _serialize_process,
+    _should_skip_first_snapshot,
     _validate_process_keys,
     _write_json,
     _write_process,
-    iter_initialized_processes,
+    iter_processes,
+    sysmon_process_monitor_threshold_task,
 )
 
 
+class _FakeSysmontap:
+    def __init__(self, snapshots):
+        self._snapshots = snapshots
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, exc, tb):
+        return None
+
+    async def iter_processes(self):
+        for snapshot in self._snapshots:
+            yield snapshot
+
+
+class _FakeDvtProvider:
+    def __init__(self, service_provider):
+        self.service_provider = service_provider
+
+    async def __aenter__(self):
+        return object()
+
+    async def __aexit__(self, exc_type, exc, tb):
+        return None
+
+
 def test_parse_process_filters_groups_values_by_key():
     assert _parse_process_filters(["name=abc", "name=def", "pid=7"]) == {
         "name": ["abc", "def"],
@@ -180,46 +208,75 @@ def test_duration_elapsed(monkeypatch, start_time, duration_ms, current_time, ex
     assert _duration_elapsed(start_time, duration_ms) is expected
 
 
-class _FakeSysmontap:
-    def __init__(self, snapshots):
-        self._snapshots = snapshots
-
-    async def __aenter__(self):
-        return self
-
-    async def __aexit__(self, exc_type, exc, tb):
-        return None
-
-    async def iter_processes(self):
-        for snapshot in self._snapshots:
-            yield snapshot
+@pytest.mark.parametrize(
+    ("keys", "expected"),
+    [
+        (None, True),
+        (["pid"], False),
+        (["pid", "cpuUsage"], True),
+    ],
+)
+def test_should_skip_first_snapshot(keys, expected):
+    assert _should_skip_first_snapshot(keys) is expected
 
 
 @pytest.mark.asyncio
-async def test_iter_initialized_processes_skips_first_snapshot():
-    sysmon = _FakeSysmontap([
+async def test_iter_processes_skips_first_snapshot_when_requested():
+    snapshots = [
         [{"pid": 10, "ppid": 1, "name": "first-snapshot"}],
         [{"pid": 20, "ppid": 2, "name": "second-snapshot"}],
         [{"pid": 30, "ppid": 3, "name": "third-snapshot"}],
-    ])
+    ]
+    sysmon = _FakeSysmontap(snapshots)
 
-    snapshots = [snapshot async for snapshot in iter_initialized_processes(sysmon)]
+    iterated_snapshots = [snapshot async for snapshot in iter_processes(sysmon, skip_first_snapshot=True)]
 
-    assert snapshots == [
-        [{"pid": 20, "ppid": 2, "name": "second-snapshot"}],
-        [{"pid": 30, "ppid": 3, "name": "third-snapshot"}],
-    ]
+    assert iterated_snapshots == snapshots[1:]
 
 
 @pytest.mark.asyncio
-async def test_iter_initialized_processes_empty_when_only_warmup_snapshot_exists():
+async def test_iter_processes_empty_when_only_warmup_snapshot_exists():
     sysmon = _FakeSysmontap([[{"pid": 10, "ppid": 1, "name": "first-snapshot"}]])
 
-    snapshots = [snapshot async for snapshot in iter_initialized_processes(sysmon)]
+    snapshots = [snapshot async for snapshot in iter_processes(sysmon, skip_first_snapshot=True)]
 
     assert snapshots == []
 
 
+@pytest.mark.asyncio
+async def test_iter_processes_keeps_first_snapshot_when_cpu_usage_not_required():
+    snapshots = [
+        [{"pid": 10, "ppid": 1, "name": "first-snapshot"}],
+        [{"pid": 20, "ppid": 2, "name": "second-snapshot"}],
+    ]
+
+    sysmon = _FakeSysmontap(snapshots)
+
+    iterated_snapshots = [snapshot async for snapshot in iter_processes(sysmon, skip_first_snapshot=False)]
+
+    assert iterated_snapshots == snapshots
+
+
+@pytest.mark.asyncio
+async def test_sysmon_process_monitor_threshold_task_skips_first_snapshot(monkeypatch):
+    async def fake_create(_dvt):
+        return _FakeSysmontap([
+            [{"pid": 10, "cpuUsage": 99.0}],
+            [{"pid": 20, "cpuUsage": 1.5}],
+        ])
+
+    monkeypatch.setattr(process_module, "DvtProvider", _FakeDvtProvider)
+    monkeypatch.setattr(process_module.Sysmontap, "create", fake_create)
+
+    out = StringIO()
+
+    await sysmon_process_monitor_threshold_task(object(), threshold=0.0, keys=["pid"], out=out, duration=1)
+
+    lines = [json.loads(line) for line in out.getvalue().splitlines() if line.strip()]
+    assert len(lines) == 1
+    assert lines[0]["pid"] == 20
+
+
 def test_process_sort_key_normalizes_missing_values():
     assert _process_sort_key({"name": "abc"}) == (-1, -1, "abc")
 
@@ -337,6 +394,23 @@ async def fake_create(_dvt):
     assert selected == {"pid": 20, "ppid": 2, "name": "second-snapshot"}
 
 
+@pytest.mark.asyncio
+async def test_select_process_from_sysmon_doesnt_skip_first_snapshot_when_cpu_usage_not_requested(monkeypatch):
+    async def fake_create(_dvt):
+        return _FakeSysmontap([
+            [{"pid": 10, "ppid": 1, "name": "first-snapshot"}],
+            [{"pid": 20, "ppid": 2, "name": "second-snapshot"}],
+        ])
+
+    monkeypatch.setattr(process_module.Sysmontap, "create", fake_create)
+
+    selected = await _select_process_from_sysmon(
+        object(), {"name": ["first-snapshot"]}, ["pid", "name"], ProcessSelectionMode.FIRST
+    )
+
+    assert selected == {"pid": 10, "ppid": 1, "name": "first-snapshot"}
+
+
 @pytest.mark.asyncio
 async def test_select_process_from_sysmon_raises_when_no_usable_snapshot(monkeypatch):
     async def fake_create(_dvt):