"""Диагностическая рамка: `tracemalloc` против RSS.

Скрипт показывает три состояния, по которым различают причину роста памяти:

  1. RSS вырос, `tracemalloc` — нет (память взята мимо аллокатора объектов
     либо арена удержана процессом после освобождения объектов);
  2. `tracemalloc` вырос, RSS — нет (объекты легли в уже удержанные арены);
  3. оба выросли (обычный живой рост: объекты созданы и живы).

Ничего не моделируется: RSS читается из `/proc/self/status` (VmRSS),
`tracemalloc` опрашивается через `get_traced_memory()`. Скрипт работает только
на Linux — на других системах VmRSS не читается и он честно об этом скажет.

    python3.13 bench/memory/tracemalloc_vs_rss.py

Числа в килобайтах/мегабайтах привязаны к машине и к прогону; воспроизводится
не число, а **знак** каждой из трёх строк.
"""
import gc
import mmap
import sys
import tracemalloc

MiB = 1024 * 1024


def rss_bytes():
    """Резидентный размер процесса по данным ядра (Linux)."""
    try:
        with open("/proc/self/status", "r", encoding="ascii") as fh:
            for line in fh:
                if line.startswith("VmRSS:"):
                    return int(line.split()[1]) * 1024
    except OSError:
        pass
    return None


class Probe:
    """Снимок пары (RSS, tracemalloc.current) и разница между снимками."""

    def __init__(self):
        self.rss = rss_bytes()
        self.traced = tracemalloc.get_traced_memory()[0]

    def delta(self, other):
        drss = None if (self.rss is None or other.rss is None) else other.rss - self.rss
        return drss, other.traced - self.traced


def mb(n):
    if n is None:
        return "    н/д"
    return f"{n / MiB:+8.2f}"


def line(title, drss, dtraced):
    print(f"  {title:<44} RSS {mb(drss)} МиБ   tracemalloc {mb(dtraced)} МиБ")


def case_untraced_mapping(n_mib=64):
    """1a. Память, которой `tracemalloc` не видит вовсе.

    `mmap.mmap` берёт страницы у ядра напрямую, минуя `PyObject_Malloc`.
    `tracemalloc` перехватывает только домены аллокатора CPython, поэтому
    страницы, которые мы тут же тронем, попадут в RSS и не попадут в трассу.
    Ровно так же выглядят буферы, взятые сторонней C-библиотекой.
    """
    before = Probe()
    buf = mmap.mmap(-1, n_mib * MiB)
    for off in range(0, n_mib * MiB, 4096):   # тронуть каждую страницу
        buf[off] = 1
    after = Probe()
    drss, dtraced = before.delta(after)
    buf.close()
    return drss, dtraced


def case_arena_held(n_objects=400_000):
    """1b. Объекты созданы и удалены: `tracemalloc` вернулся, RSS — нет.

    Арена отдаётся системе только когда пусты все её пулы. После удаления
    списка `tracemalloc` показывает возврат к исходному, а RSS остаётся
    поднятым — это и есть «аллокатор держит арену», а не утечка.
    """
    before = Probe()
    data = [object() for _ in range(n_objects)]
    peak = Probe()
    del data
    gc.collect()
    after = Probe()
    return before.delta(peak), before.delta(after)


def case_reuse(n_objects=400_000):
    """2. Обратный случай: `tracemalloc` вырос ровно так же, RSS — меньше.

    Тот же цикл «создать — удалить» повторяется несколько раз подряд. Арены
    уже у процесса, новые объекты ложатся в них, и у ядра каждый следующий раз
    просят всё меньше. `tracemalloc` при этом каждый раз показывает одно и то
    же число: он считает запрошенные байты, а не страницы процесса.
    """
    rounds = []
    for _ in range(3):
        before = Probe()
        data = [object() for _ in range(n_objects)]
        peak = Probe()
        rounds.append(before.delta(peak))
        del data
        gc.collect()
    return rounds


def top_sources(limit=3):
    """Что именно показывает `tracemalloc` — строку исходника и байты."""
    snap = tracemalloc.take_snapshot()
    return snap.statistics("lineno")[:limit]


def main():
    print("PY", sys.version.split()[0])
    if rss_bytes() is None:
        print("VmRSS недоступен (не Linux) — колонка RSS будет пустой")
    print("gc.isenabled():", gc.isenabled())
    print()

    tracemalloc.start(1)
    gc.collect()

    print("1a. mmap на 64 МиБ, все страницы тронуты "
          "(аллокатор объектов не участвует):")
    drss, dtraced = case_untraced_mapping()
    line("после mmap + touch", drss, dtraced)
    print("    ожидаемый знак: RSS вырос, tracemalloc ~0")
    print()

    print("1b. 400 000 объектов созданы, затем удалены:")
    (drss_peak, dtr_peak), (drss_end, dtr_end) = case_arena_held()
    line("на пике (объекты живы)", drss_peak, dtr_peak)
    line("после del + gc.collect()", drss_end, dtr_end)
    print("    ожидаемый знак: tracemalloc вернулся к нулю, RSS остался поднят")
    print()

    print("2. Тот же цикл ещё три раза подряд, пик каждого:")
    for i, (drss_i, dtr_i) in enumerate(case_reuse(), start=1):
        line(f"круг {i}", drss_i, dtr_i)
    print("    ожидаемый знак: tracemalloc одинаков на всех кругах,")
    print("    RSS от круга к кругу растёт всё меньше — арены уже у процесса")
    print()

    print("3. Что tracemalloc показывает про источник (top-3 строки):")
    for stat in top_sources():
        frame = stat.traceback[0]
        print(f"  {frame.filename}:{frame.lineno}  "
              f"{stat.size / 1024:9.1f} КиБ в {stat.count} блоках")
    print()

    cur, peak = tracemalloc.get_traced_memory()
    print(f"итог tracemalloc: current {cur / MiB:.2f} МиБ, peak {peak / MiB:.2f} МиБ")
    print(f"итог RSS        : {rss_bytes() / MiB:.2f} МиБ"
          if rss_bytes() is not None else "итог RSS        : н/д")
    print()
    print("Чего tracemalloc не покажет никогда: память вне доменов аллокатора")
    print("CPython (mmap, буферы C-библиотек, стеки потоков) и фрагментацию —")
    print("он считает запрошенные байты, а не занятые процессом страницы.")
    print("Обратное тоже верно и видно в блоке 1b: RSS на пике больше traced,")
    print("потому что в RSS входит и служебная таблица самого tracemalloc.")
    tracemalloc.stop()


if __name__ == "__main__":
    main()