Build mini-ps: свой ps из /proc на Python

ps — одна из первых команд, которую узнают на Linux. ps aux, ps -ef, ps -eo pid,ppid,user,comm. Кажется чем-то магическим: откуда оно знает обо всех процессах? И почему иногда показывает чудовищные числа VSZ — 120 GB — на машине с 16 GB RAM?

В этом уроке мы напишем свой ps. Не «обёртку над psutil», а настоящий: открыть /proc, прочитать каждый numeric каталог (это и есть PID), извлечь из stat/status поля, отформатировать колонками. Получится 150 строк Python, которые показывают то же, что коммерческий ps. После урока эта команда станет полностью прозрачной.

Что должно получиться

Запускаем python mini_ps.py:

   PID    PPID USER         STAT  RSS(MB)  VSZ(MB) THR COMMAND
     1       0 root         S        12.3    164.1   1 systemd
   832       1 root         S        18.5    320.2   3 systemd-journald
  1024       1 root         S         8.1    115.4   1 systemd-logind
  4521    4520 alice        S        45.2    520.3   1 -bash
 12345    4521 alice        R         3.2     20.1   1 python mini_ps.py

Колонки: PID, PPID, owner (uid -> username), STAT (R/S/D/Z/T…), RSS в MB, VSZ в MB, число потоков, командная строка. Это то же, что показывает ps -eo pid,ppid,user,stat,rss,vsz,nlwp,args.

Без флагов, фильтров — простой dump. Расширения (фильтры, sort) — в Lab-01.

Архитектура

Цикл: список PID -> для каждого открываем 3 файла -> парсим -> кладём в dataclass -> печатаем строку.

Шаг 1: получаем список PID

PID — это каталог в /proc, имя которого — число. Всё остальное (cpuinfo, version, sys/) — не PID.

import os

def list_pids() -> list[int]:
    """Return all numeric directory names from /proc as ints (PIDs)."""
    pids = []
    for name in os.listdir('/proc'):
        if name.isdigit():
            pids.append(int(name))
    return sorted(pids)

isdigit() — простой способ отфильтровать число от строки. На моей машине это возвращает что-то вроде [1, 2, 3, ..., 12345] — список из 200-300 элементов на типичной системе. На сервере с контейнерами — может быть 1000+.

Шаг 2: парсим /proc/[pid]/stat

Это самый коварный файл в /proc. Выглядит так:

$ cat /proc/12345/stat
12345 (python mini_ps.py) R 4521 12345 4521 34816 12345 4194304 87 0 0 0 1 0 0 0 20 0 1 0 1234567 20582400 813 18446744073709551615 ...

Одна строка, 50+ полей через пробел. По таблице из man 5 proc:

• Поле 1: pid
• Поле 2: (comm) — имя в скобках
• Поле 3: state — одна буква R/S/D/Z/T/…
• Поле 4: ppid
• Поле 14: utime (CPU user mode, ticks)
• Поле 15: stime (CPU kernel mode, ticks)
• Поле 20: num_threads
• Поле 22: starttime (since boot, ticks)
• Поле 23: vsize (bytes)
• Поле 24: rss (страницы, не bytes)

Ловушка: comm в скобках может содержать пробелы и скобки. Имя python mini_ps.py будет в stat как (python mini_ps.py). Если просто разбить по пробелу — поля разъедутся.

Правильный подход: найти позицию последней ) в строке, всё до неё с первым символом — pid, в скобках — comm, после — остальные поля.

def parse_stat(stat_text: str) -> dict:
    """Parse /proc/[pid]/stat. Handle (comm) with parens/spaces correctly."""
    # find LAST ')' to safely bound comm
    paren_close = stat_text.rfind(')')
    pid_part = stat_text[: stat_text.find('(')]
    comm = stat_text[stat_text.find('(') + 1 : paren_close]
    rest = stat_text[paren_close + 2 :].split()

    return {
        'pid': int(pid_part.strip()),
        'comm': comm,
        'state': rest[0],          # field 3
        'ppid': int(rest[1]),      # field 4
        'num_threads': int(rest[17]),  # field 20 (zero-indexed in rest: 20 - 3 = 17)
        'vsize_bytes': int(rest[20]),  # field 23
        'rss_pages': int(rest[21]),    # field 24
    }

Индексы в rest: первое поле в rest — это поле 3 (state) исходной строки. Чтобы получить поле N -> rest[N - 3]. Это «дочка» того, что мы вырезали pid (поле 1) и comm (поле 2) отдельно.

Эта же ловушка в стандартной ps-имплементации (procps-ng) обработана через rfind. Если будете гуглить — увидите тот же код в C.

Шаг 3: парсим /proc/[pid]/status

$ cat /proc/12345/status | head -10
Name:	python
Umask:	0022
State:	S (sleeping)
Tgid:	12345
Ngid:	0
Pid:	12345
PPid:	4521
TracerPid:	0
Uid:	1000	1000	1000	1000
Gid:	1000	1000	1000	1000

Формат — key:\t...\tvalues. Парсится тривиально:

def parse_status(status_text: str) -> dict:
    """Parse /proc/[pid]/status. Format: 'Key:\\tvalues'."""
    out = {}
    for line in status_text.splitlines():
        if ':' not in line:
            continue
        key, _, value = line.partition(':')
        out[key.strip()] = value.strip()
    return out

Из status нам нужен Uid — там 4 числа (real, effective, saved, fs), первое — real UID. И Threads (дублирует num_threads из stat).

uid_str = status_dict['Uid'].split()[0]
uid = int(uid_str)

Шаг 4: cmdline

def read_cmdline(pid: int) -> str:
    """Return cmdline with \\0 -> space. Empty for kernel threads."""
    try:
        with open(f'/proc/{pid}/cmdline', 'rb') as f:
            data = f.read()
    except (FileNotFoundError, ProcessLookupError):
        return ''
    if not data:
        return ''
    # ends with trailing \0, strip it; separators inside are \0
    return data.rstrip(b'\x00').replace(b'\x00', b' ').decode('utf-8', errors='replace')

Тонкости:

1. Бинарное чтение ('rb') — потому что аргументы могут содержать non-UTF-8 байты, и 'r' может упасть на UnicodeDecodeError. Декодим в конце с errors='replace'.
2. Пустая cmdline — это kernel thread. Обычно их имена показывают в [brackets] в реальном ps. Здесь упростим: показываем comm из stat.
3. Trailing \0 — после последнего аргумента стоит null-byte. Если не отрезать — будет лишний пробел в выводе.

Шаг 5: uid -> username

Uid: 1000 — это число. Чтобы получить alice (или root для UID=0), используем модуль pwd:

import pwd

def uid_to_name(uid: int) -> str:
    try:
        return pwd.getpwuid(uid).pw_name
    except KeyError:
        return str(uid)

pwd.getpwuid() идёт в /etc/passwd (или NSS — если LDAP/SSSD настроены). Если запись не найдена — kid вариант: показываем сам UID числом. Это бывает для процессов в namespaces с другой user map.

Шаг 6: собираем всё вместе

Структура процесса как dataclass:

from dataclasses import dataclass

@dataclass
class ProcessInfo:
    pid: int
    ppid: int
    user: str
    state: str
    rss_mb: float
    vsize_mb: float
    nthreads: int
    command: str

Функция, которая по PID собирает ProcessInfo:

PAGE_SIZE = os.sysconf('SC_PAGESIZE')  # обычно 4096

def collect_process(pid: int) -> ProcessInfo | None:
    """Read /proc/[pid]/{stat,status,cmdline} -> ProcessInfo. None if pid disappeared."""
    try:
        with open(f'/proc/{pid}/stat') as f:
            stat_text = f.read()
        with open(f'/proc/{pid}/status') as f:
            status_text = f.read()
    except (FileNotFoundError, ProcessLookupError):
        return None  # process died between listdir and open

    stat = parse_stat(stat_text)
    status = parse_status(status_text)

    uid = int(status['Uid'].split()[0])
    user = uid_to_name(uid)

    command = read_cmdline(pid) or f'[{stat["comm"]}]'  # kthread

    return ProcessInfo(
        pid=stat['pid'],
        ppid=stat['ppid'],
        user=user,
        state=stat['state'],
        rss_mb=stat['rss_pages'] * PAGE_SIZE / (1024 * 1024),
        vsize_mb=stat['vsize_bytes'] / (1024 * 1024),
        nthreads=stat['num_threads'],
        command=command,
    )

И вывод колонками:

def print_table(processes: list[ProcessInfo]) -> None:
    print(f'{"PID":>6} {"PPID":>6} {"USER":<12} {"STAT":>4} '
          f'{"RSS(MB)":>8} {"VSZ(MB)":>8} {"THR":>3} COMMAND')
    for p in processes:
        cmd = p.command[:80]  # truncate long
        print(f'{p.pid:>6} {p.ppid:>6} {p.user:<12} {p.state:>4} '
              f'{p.rss_mb:>8.1f} {p.vsize_mb:>8.1f} {p.nthreads:>3} {cmd}')

Main:

def main():
    pids = list_pids()
    processes = []
    for pid in pids:
        info = collect_process(pid)
        if info is not None:
            processes.append(info)
    print_table(processes)

if __name__ == '__main__':
    main()

Запускаем:

$ python mini_ps.py | head -10
   PID   PPID USER         STAT  RSS(MB)  VSZ(MB) THR COMMAND
     1      0 root            S     12.3    164.1   1 /sbin/init
     2      0 root            S      0.0      0.0   1 [kthreadd]
     3      2 root            I      0.0      0.0   1 [rcu_gp]
   ...

150 строк. Работает. Это и есть упрощённый ps.

Edge cases

Race conditions

Между listdir('/proc') и open('/proc/[pid]/stat') процесс может умереть. Файл исчезнет, open бросит FileNotFoundError. Обработали уже выше: return None и продолжаем.

Сложнее: процесс может умереть посередине чтения stat. Это редко — kernel держит /proc/[pid]/* «снэпшотом» на момент open. Но статус полей (rss, threads) меняется в реальном времени; если процесс fork-ает быстро, num_threads может «прыгнуть» с 1 на 5. Мы это не видим в нашем выводе — мы делаем один snapshot.

Permissions

Без root некоторые поля недоступны. Конкретно:

• /proc/[pid]/cmdline — обычно читаем всеми (если процесс не ставил prctl(PR_SET_DUMPABLE, 0)).
• /proc/[pid]/status — Uid/Gid читаются всеми, но Cwd, Environ — только owner или root.
• /proc/[pid]/maps — owner или root.

В нашем mini-ps мы читаем только то, что доступно без root, поэтому работает без sudo. Но для процессов другого пользователя могут отсутствовать некоторые детали — заметили бы при попытке читать /proc/[pid]/fd, который требует privileged access.

Kernel threads

[kthreadd], [ksoftirqd/0], [rcu_gp] — это kernel threads. Они существуют как PIDs, но не имеют ни cmdline, ни ничего связанного с userspace. Их RSS=0, VSZ=0. В нашем коде это обрабатывается через read_cmdline() -> '', тогда подставляем [comm] (имя из stat в скобках), как делает реальный ps.

Что значит state буква

В нашем mini-ps мы просто выводим букву. Какая разница между R и S — это уже знание из модуля 3-processes.

Производительность

На системе с 500 процессами наш mini-ps делает:

• 500 × 3 = 1500 open() + read() + close() syscalls.
• Каждый /proc/[pid]/stat — ~1-2 KB. /proc/[pid]/status — ~1 KB. cmdline — 100 bytes.
• Итого: ~1 MB чтений из ядра, ~50ms на типичной машине.

ps из procps делает примерно то же. На 10K процессов (busy сервер) можно прийти к 1+ секунды на полный listing. Если хочется быстрее — getdents() для списка PID + openat() + одна общая buffer reuse. Но для capstone это не нужно.

Попробуй сам

Прямо после прочтения:

1. Скопируйте код выше в файл mini_ps.py. Это ~80 строк.

2. Запустите: python mini_ps.py | head -20. Сравните с ps -eo pid,ppid,user,stat,rss,vsz,nlwp,args | head -20. Поля должны совпасть (с точностью до format/scale).

3. Найдите процесс с самым большим RSS:

python mini_ps.py | sort -k5 -rn | head -5

Что это? Обычно — браузер (Firefox/Chrome) или JVM. Запомните это RSS — пригодится в уроке про monitor.

4. Посмотрите процессы в состоянии D:

python mini_ps.py | awk '$4 == "D"'

Если есть — ваш диск что-то тормозит, или DB-процесс ждёт fsync. Пустой вывод — норма.

5. Расширьте: добавьте колонку TIME — это CPU-время процесса. В stat это поля 14 (utime) + 15 (stime), в ticks. Чтобы получить секунды: (utime + stime) / os.sysconf('SC_CLK_TCK'). SC_CLK_TCK обычно 100, значит ticks = центисекунды.

clock_ticks = os.sysconf('SC_CLK_TCK')
# в parse_stat добавьте:
'utime': int(rest[11]),  # field 14
'stime': int(rest[12]),  # field 15
# в collect_process:
cpu_seconds = (stat['utime'] + stat['stime']) / clock_ticks

Это сразу даёт всё, что вытаскивает ps -eo time.

KnowledgeCheck

Итог

mini-ps готов. Меньше 150 строк Python, без зависимостей. Делает то же, что коммерческий ps (упрощённый формат). Главное, что вы вынесли:

• /proc — это унифицированный API. listdir + open + read + close — весь инструментарий.
• Race conditions неизбежны: процесс может умереть между listdir и open. FileNotFoundError -> skip.
• /proc/[pid]/stat — коварный формат с (comm) в скобках. rfind(')') — стандартный паттерн.
• RSS в stat — в страницах, не в bytes. * PAGE_SIZE = bytes.
• uid -> username через pwd.getpwuid().

В уроке 3 — mini-shell. Перейдём от наблюдения процессов к их созданию. fork, exec, pipe.