Take multiple RSS samples and return the median (KiB).
(
pid: int,
samples: int = RSS_SAMPLES,
interval: float = RSS_SAMPLE_INTERVAL,
)
| 422 | |
| 423 | |
| 424 | def sample_rss( |
| 425 | pid: int, |
| 426 | samples: int = RSS_SAMPLES, |
| 427 | interval: float = RSS_SAMPLE_INTERVAL, |
| 428 | ) -> int: |
| 429 | """Take multiple RSS samples and return the median (KiB).""" |
| 430 | values: list[int] = [] |
| 431 | for _ in range(samples): |
| 432 | rss = get_rss_kib(pid) |
| 433 | if rss is not None: |
| 434 | values.append(rss) |
| 435 | time.sleep(interval) |
| 436 | if not values: |
| 437 | raise RuntimeError(f"Could not read RSS for PID {pid}") |
| 438 | values.sort() |
| 439 | return values[len(values) // 2] |
| 440 | |
| 441 | |
| 442 | # ── Scenario runners ───────────────────────────────────────────────────────── |
no test coverage detected