"""Цена ОДНОВРЕМЕННОГО ОЖИДАНИЯ: N ждущих сразу, потоками и корутинами.

ЗАЧЕМ ЭТОТ ЗАМЕР ПОСЛЕ ОСТАЛЬНЫХ. Предыдущие показали, что на работе с базой
модель исполнения не решает почти ничего: потолок ставит пул соединений. Отсюда
легко сделать вывод «async не нужен», и он был бы неверным. Этот замер про то,
где разница настоящая, — не в скорости одного ожидания, а в цене того, чтобы
ДЕРЖАТЬ много ожиданий сразу.

ЧТО ИЗМЕРЯЕТСЯ. N единиц исполнения, каждая ждёт одно и то же время. База здесь
ни при чём намеренно: ожидание берётся самое дешёвое, какое бывает
(`time.sleep` / `asyncio.sleep`), чтобы в числах не осталось ничего, кроме цены
самой единицы исполнения.

  * время: сколько занимает завести N ждущих и дождаться всех;
  * память: прирост RSS процесса и, отдельно, прирост по `tracemalloc`.

ПОЧЕМУ ДВА СЧЁТЧИКА ПАМЯТИ, А НЕ ОДИН. Они меряют разное, и в этом всё дело.
Стек потока выделяет операционная система, а не аллокатор Python: `tracemalloc`
его не видит вовсе, и по нему тысяча потоков «ничего не стоит». RSS видит.
Обратная сторона: RSS растёт страницами и не уменьшается сразу, поэтому мелкие
объекты — задачи — в нём могут не проявиться, если место в куче уже было. Одно
число тут соврало бы в любую сторону; два показывают, какая память чья.
"""

import asyncio
import gc
import os
import sys
import threading
import time
import tracemalloc

sys.path.insert(0, __file__.rsplit("/", 1)[0])

WAIT = 0.2
COUNTS = [100, 500, 1000, 2000, 4000]


def rss_kb() -> int:
    """Резидентная память процесса. /proc — потому что стеки потоков видны только там."""
    with open("/proc/self/status", encoding="ascii") as fh:
        for line in fh:
            if line.startswith("VmRSS:"):
                return int(line.split()[1])
    return 0


def threads(n: int) -> tuple[float, int, int]:
    gc.collect()
    tracemalloc.start()
    base = rss_kb()
    peak = base
    started_all = threading.Event()

    def hold() -> None:
        started_all.wait()
        time.sleep(WAIT)

    started = time.perf_counter()
    pool = [threading.Thread(target=hold) for _ in range(n)]
    for t in pool:
        t.start()
    # Все потоки уже созданы и стоят на барьере — здесь и снимается память:
    # именно в этот момент они существуют одновременно.
    peak = max(peak, rss_kb())
    traced = tracemalloc.get_traced_memory()[1]
    started_all.set()
    for t in pool:
        t.join()
    tracemalloc.stop()
    return time.perf_counter() - started, peak - base, traced


async def _tasks(n: int) -> tuple[float, int, int]:
    gc.collect()
    tracemalloc.start()
    base = rss_kb()
    gate = asyncio.Event()

    async def hold() -> None:
        await gate.wait()
        await asyncio.sleep(WAIT)

    started = time.perf_counter()
    pool = [asyncio.create_task(hold()) for _ in range(n)]
    await asyncio.sleep(0)  # дать задачам дойти до ожидания
    peak = rss_kb()
    traced = tracemalloc.get_traced_memory()[1]
    gate.set()
    await asyncio.gather(*pool)
    tracemalloc.stop()
    return time.perf_counter() - started, peak - base, traced


def main() -> None:
    print(f"python      {sys.version.split()[0]}")
    print(f"cpu         {os.cpu_count()}")
    print(f"каждый ждёт {WAIT * 1000:.0f} мс; память снята, пока все существуют сразу\n")
    print(
        f"{'сколько':>8} | {'потоки':>8} {'RSS':>8} {'traced':>8} | "
        f"{'корутины':>9} {'RSS':>8} {'traced':>8}"
    )
    print("-" * 70)

    for n in COUNTS:
        t_thr, rss_thr, tr_thr = threads(n)
        t_task, rss_task, tr_task = asyncio.run(_tasks(n))
        print(
            f"{n:>8} | {t_thr:>7.3f}с {rss_thr / 1024:>6.1f}МБ {tr_thr / 1048576:>6.1f}МБ | "
            f"{t_task:>8.3f}с {rss_task / 1024:>6.1f}МБ {tr_task / 1048576:>6.1f}МБ"
        )


if __name__ == "__main__":
    main()