#!/usr/bin/env python3 # SPDX-License-Identifier: GPL-2.0-only """Display r/w activity for all processes.""" from __future__ import annotations import argparse from collections import defaultdict import sys from typing import Optional, Dict, List, Tuple, Any import perf class RwByPid: """Tracks and displays read/write activity by PID.""" def __init__(self) -> None: self.reads: Dict[int, Dict[str, Any]] = defaultdict( lambda: { "bytes_requested": 0, "bytes_read": 0, "total_reads": 0, "comm": "", "errors": defaultdict(int), } ) self.writes: Dict[int, Dict[str, Any]] = defaultdict( lambda: { "bytes_requested": 0, "bytes_written": 0, "total_writes": 0, "comm": "", "errors": defaultdict(int), } ) self.unhandled: Dict[str, int] = defaultdict(int) self.session: Optional[perf.session] = None def _handle_sys_enter_read(self, pid: int, comm: str, sample: perf.sample_event) -> None: try: count = sample.count self.reads[pid]["bytes_requested"] += count self.reads[pid]["total_reads"] += 1 self.reads[pid]["comm"] = comm except AttributeError: self.unhandled["sys_enter_read_attr_err"] += 1 def _handle_sys_exit_read(self, pid: int, sample: perf.sample_event, comm: str = "") -> None: try: ret = sample.ret # Convert unsigned 64-bit or 32-bit tracepoint return values to signed integers. # Because kernel read/write transfers are bounded by MAX_RW_COUNT (0x7ffff000 < 2 GiB), # 0xfffff000..0xffffffff is always a 32-bit negative errno even on 64-bit hosts. if ret >= 0x8000000000000000: ret -= 0x10000000000000000 elif 0xfffff000 <= ret <= 0xffffffff: ret -= 0x100000000 if not self.reads[pid]["comm"] and comm: self.reads[pid]["comm"] = comm if ret >= 0: self.reads[pid]["bytes_read"] += ret else: self.reads[pid]["errors"][ret] += 1 except AttributeError: self.unhandled["sys_exit_read_attr_err"] += 1 def _handle_sys_enter_write(self, pid: int, comm: str, sample: perf.sample_event) -> None: try: count = sample.count self.writes[pid]["bytes_requested"] += count self.writes[pid]["total_writes"] += 1 self.writes[pid]["comm"] = comm except AttributeError: self.unhandled["sys_enter_write_attr_err"] += 1 def _handle_sys_exit_write(self, pid: int, sample: perf.sample_event, comm: str = "") -> None: try: ret = sample.ret # Convert unsigned 64-bit or 32-bit tracepoint return values to signed integers. # Because kernel read/write transfers are bounded by MAX_RW_COUNT (0x7ffff000 < 2 GiB), # 0xfffff000..0xffffffff is always a 32-bit negative errno even on 64-bit hosts. if ret >= 0x8000000000000000: ret -= 0x10000000000000000 elif 0xfffff000 <= ret <= 0xffffffff: ret -= 0x100000000 if not self.writes[pid]["comm"] and comm: self.writes[pid]["comm"] = comm if ret >= 0: self.writes[pid]["bytes_written"] += ret else: self.writes[pid]["errors"][ret] += 1 except AttributeError: self.unhandled["sys_exit_write_attr_err"] += 1 def process_event(self, sample: perf.sample_event) -> None: """Process events.""" event_name = str(sample.evsel) if event_name.startswith("evsel(") and event_name.endswith(")"): event_name = event_name[6:-1] event_name = "".join(c if c.isprintable() else "?" for c in event_name) pid = sample.sample_pid assert self.session is not None try: thread = self.session.find_thread(pid, sample.sample_tid) comm = (thread.comm() if thread else None) or "unknown" except (TypeError, AttributeError): comm = "unknown" comm = "".join(c if c.isprintable() else "?" for c in comm) if event_name == "syscalls:sys_enter_read": self._handle_sys_enter_read(pid, comm, sample) elif event_name == "syscalls:sys_exit_read": self._handle_sys_exit_read(pid, sample, comm) elif event_name == "syscalls:sys_enter_write": self._handle_sys_enter_write(pid, comm, sample) elif event_name == "syscalls:sys_exit_write": self._handle_sys_exit_write(pid, sample, comm) else: self.unhandled[event_name] += 1 def print_totals(self) -> None: """Print summary tables.""" print("read counts by pid:\n") print( f"{'pid':>6s} {'comm':<20s} {'# reads':>10s} " f"{'bytes_requested':>15s} {'bytes_read':>10s}" ) print(f"{'-'*6} {'-'*20} {'-'*10} {'-'*15} {'-'*10}") for pid, data in sorted(self.reads.items(), key=lambda kv: kv[1]["bytes_read"], reverse=True): print( f"{pid:6d} {data['comm']:<20s} {data['total_reads']:10d} " f"{data['bytes_requested']:15d} {data['bytes_read']:10d}" ) print("\nfailed reads by pid:\n") print(f"{'pid':>6s} {'comm':<20s} {'error #':>7s} {'# errors':>10s}") print(f"{'-'*6} {'-'*20} {'-'*7} {'-'*10}") errcounts: List[Tuple[int, str, int, int]] = [] for pid, data in self.reads.items(): for error, count in data["errors"].items(): errcounts.append((pid, data["comm"], error, count)) for pid, comm, error, count in sorted(errcounts, key=lambda x: x[3], reverse=True): print(f"{pid:6d} {comm:<20s} {error:7d} {count:10d}") print("\nwrite counts by pid:\n") print( f"{'pid':>6s} {'comm':<20s} {'# writes':>10s} " f"{'bytes_requested':>15s} {'bytes_written':>15s}" ) print(f"{'-'*6} {'-'*20} {'-'*10} {'-'*15} {'-'*15}") for pid, data in sorted(self.writes.items(), key=lambda kv: kv[1]["bytes_written"], reverse=True): print( f"{pid:6d} {data['comm']:<20s} {data['total_writes']:10d} " f"{data['bytes_requested']:15d} {data['bytes_written']:15d}" ) print("\nfailed writes by pid:\n") print(f"{'pid':>6s} {'comm':<20s} {'error #':>7s} {'# errors':>10s}") print(f"{'-'*6} {'-'*20} {'-'*7} {'-'*10}") errcounts = [] for pid, data in self.writes.items(): for error, count in data["errors"].items(): errcounts.append((pid, data["comm"], error, count)) for pid, comm, error, count in sorted(errcounts, key=lambda x: x[3], reverse=True): print(f"{pid:6d} {comm:<20s} {error:7d} {count:10d}") if self.unhandled: print("\nunhandled events:\n") print(f"{'event':<40s} {'count':>10s}") print(f"{'-'*40} {'-'*10}") for event_name, count in self.unhandled.items(): print(f"{event_name:<40s} {count:10d}") def run(self, input_file: str) -> None: """Run the session.""" self.session = perf.session(perf.data(input_file), sample=self.process_event) try: self.session.process_events() finally: # Break the reference cycle between perf.session and self.process_event # because perf.session lacks cyclic GC support (tp_traverse). self.session = None self.print_totals() def main() -> None: """Main function.""" parser = argparse.ArgumentParser(description="Trace r/w activity by PID") parser.add_argument("-i", "--input", default="perf.data", help="Input file") args = parser.parse_args() analyzer = RwByPid() try: analyzer.run(args.input) except IOError as e: print(e, file=sys.stderr) sys.exit(1) if __name__ == "__main__": main()