(pid: int)
| 110 | |
| 111 | |
| 112 | def sample_proc(pid: int) -> ProcSample | None: |
| 113 | status = _read(f"/proc/{pid}/status") |
| 114 | stat = _read(f"/proc/{pid}/stat") |
| 115 | if status is None or stat is None: |
| 116 | return None |
| 117 | s = ProcSample() |
| 118 | s.rss_bytes = _status_kib(status, "VmRSS:") |
| 119 | s.rss_anon_bytes = _status_kib(status, "RssAnon:") |
| 120 | s.rss_file_bytes = _status_kib(status, "RssFile:") |
| 121 | s.peak_rss_bytes = _status_kib(status, "VmHWM:") |
| 122 | s.swap_bytes = _status_kib(status, "VmSwap:") |
| 123 | tm = re.search(r"^Threads:\s+(\d+)", status, re.MULTILINE) |
| 124 | s.threads = int(tm.group(1)) if tm else None |
| 125 | |
| 126 | rollup = _read(f"/proc/{pid}/smaps_rollup") |
| 127 | if rollup: |
| 128 | pm = re.search(r"^Pss:\s+(\d+)\s*kB", rollup, re.MULTILINE) |
| 129 | s.pss_bytes = int(pm.group(1)) * 1024 if pm else None |
| 130 | |
| 131 | # /proc/<pid>/stat fields after "comm) ": index 0 == field 3 (state), |
| 132 | # so field N maps to fields[N-3]. minflt=10 majflt=12 utime=14 stime=15. |
| 133 | rparen = stat.rfind(")") |
| 134 | if rparen != -1: |
| 135 | fields = stat[rparen + 2 :].split() |
| 136 | try: |
| 137 | s.minflt = int(fields[10 - 3]) # field 10 |
| 138 | s.majflt = int(fields[12 - 3]) # field 12 |
| 139 | s.utime_s = int(fields[14 - 3]) / CLK_TCK # field 14 |
| 140 | s.stime_s = int(fields[15 - 3]) / CLK_TCK # field 15 |
| 141 | except (IndexError, ValueError): |
| 142 | pass |
| 143 | |
| 144 | sched = status |
| 145 | vm = re.search(r"^voluntary_ctxt_switches:\s+(\d+)", sched, re.MULTILINE) |
| 146 | nm = re.search(r"^nonvoluntary_ctxt_switches:\s+(\d+)", sched, re.MULTILINE) |
| 147 | s.vol_ctxt = int(vm.group(1)) if vm else None |
| 148 | s.nonvol_ctxt = int(nm.group(1)) if nm else None |
| 149 | |
| 150 | io = _read(f"/proc/{pid}/io") |
| 151 | if io: |
| 152 | rm = re.search(r"^read_bytes:\s+(\d+)", io, re.MULTILINE) |
| 153 | wm = re.search(r"^write_bytes:\s+(\d+)", io, re.MULTILINE) |
| 154 | s.read_bytes = int(rm.group(1)) if rm else None |
| 155 | s.write_bytes = int(wm.group(1)) if wm else None |
| 156 | |
| 157 | try: |
| 158 | s.fds = len(os.listdir(f"/proc/{pid}/fd")) |
| 159 | except OSError: |
| 160 | s.fds = None |
| 161 | return s |
| 162 | |
| 163 | |
| 164 | @dataclass |
no test coverage detected