glchat/scripts/desktop-perf.py

#!/usr/bin/env python3
"""desktop-perf.py — замеры холодного старта и памяти desktop-обёртки.

Требование спецификации: idle RAM ≤ 350 MB, холодный старт ≤ 2 с
(docs/client-tauri.md §1, §7). Скрипт запускает собранный бандл, замеряет время
от старта процесса до загрузки страницы клиента (строка «загружена страница» в
логе обёртки) и память после того, как клиент успокоился.

Память считается по дереву процессов: на macOS страница живёт в отдельных
процессах WebKit (`WebContent`, `GPU`, `Networking`), поэтому одного RSS
главного процесса мало. Помощники WebKit запускаются через launchd (их
ppid = 1), поэтому «свои» определяются сравнением списка процессов до и после
запуска приложения.

Примеры:
    python3 scripts/desktop-perf.py                 # 3 холодных старта + простой
    python3 scripts/desktop-perf.py --runs 5 --json build/desktop-perf.json
"""

from __future__ import annotations

import argparse
import json
import os
import platform
import signal
import subprocess
import sys
import time
from dataclasses import dataclass
from pathlib import Path

REPO = Path(__file__).resolve().parent.parent
DEFAULT_APP = REPO / "desktop/src-tauri/target/release/bundle/macos/glchat.app"
APP_MARKER = "glchat.app/Contents/MacOS/glchat-desktop"
PAGE_LOADED = "загружена страница"


@dataclass
class ProcInfo:
    """Сведения о процессе, нужные для замера."""

    pid: int
    ppid: int
    path: str
    rss: int
    footprint: int


class MacProcessTable:
    """Чтение процессов через libproc (ps и top в песочнице недоступны)."""

    PROC_ALL_PIDS = 1
    PROC_PIDTBSDINFO = 3
    RUSAGE_INFO_V2 = 2

    def __init__(self) -> None:
        import ctypes
        import ctypes.util

        self.ctypes = ctypes
        self.libproc = ctypes.CDLL(ctypes.util.find_library("proc"), use_errno=True)

        class BsdInfo(ctypes.Structure):
            """struct proc_bsdinfo: нужны pbi_ppid и pbi_pid."""

            _fields_ = [
                ("pbi_flags", ctypes.c_uint32), ("pbi_status", ctypes.c_uint32),
                ("pbi_xstatus", ctypes.c_uint32), ("pbi_pid", ctypes.c_uint32),
                ("pbi_ppid", ctypes.c_uint32), ("pbi_uid", ctypes.c_uint32),
                ("pbi_gid", ctypes.c_uint32), ("pbi_ruid", ctypes.c_uint32),
                ("pbi_rgid", ctypes.c_uint32), ("pbi_svuid", ctypes.c_uint32),
                ("pbi_svgid", ctypes.c_uint32), ("rfu_1", ctypes.c_uint32),
                ("pbi_comm", ctypes.c_char * 16), ("pbi_name", ctypes.c_char * 32),
                ("pbi_nfiles", ctypes.c_uint32), ("pbi_pgid", ctypes.c_uint32),
                ("pbi_pjobc", ctypes.c_uint32), ("e_tdev", ctypes.c_uint32),
                ("e_tpgid", ctypes.c_uint32), ("pbi_nice", ctypes.c_int32),
                ("pbi_start_tvsec", ctypes.c_uint64), ("pbi_start_tvusec", ctypes.c_uint64),
            ]

        self.bsd_info = BsdInfo

    def _ppid(self, pid: int) -> int | None:
        ctypes = self.ctypes
        info = self.bsd_info()
        size = self.libproc.proc_pidinfo(
            ctypes.c_int(pid), ctypes.c_int(self.PROC_PIDTBSDINFO), ctypes.c_uint64(0),
            ctypes.byref(info), ctypes.c_int(ctypes.sizeof(info)),
        )
        return int(info.pbi_ppid) if size == ctypes.sizeof(info) else None

    def _rusage(self, pid: int) -> tuple[int, int]:
        """RSS и физический след процесса (rusage_info_v2)."""
        ctypes = self.ctypes
        buf = (ctypes.c_uint64 * 32)()
        if self.libproc.proc_pid_rusage(ctypes.c_int(pid), ctypes.c_int(self.RUSAGE_INFO_V2), ctypes.byref(buf)) != 0:
            return 0, 0
        # ri_uuid занимает два 64-битных слова, дальше: resident_size, phys_footprint.
        return int(buf[8]), int(buf[9])

    def _path(self, pid: int) -> str:
        ctypes = self.ctypes
        buf = ctypes.create_string_buffer(4096)
        size = self.libproc.proc_pidpath(ctypes.c_int(pid), buf, ctypes.c_uint32(4096))
        return buf.value.decode("utf-8", "replace") if size > 0 else ""

    def snapshot(self) -> dict[int, ProcInfo]:
        ctypes = self.ctypes
        count = 8192
        buf = (ctypes.c_int * count)()
        size = self.libproc.proc_listpids(
            ctypes.c_int(self.PROC_ALL_PIDS), ctypes.c_uint32(0), ctypes.byref(buf),
            ctypes.c_int(ctypes.sizeof(buf)),
        )
        result: dict[int, ProcInfo] = {}
        for index in range(max(size, 0) // ctypes.sizeof(ctypes.c_int)):
            pid = buf[index]
            if pid <= 0:
                continue
            ppid = self._ppid(pid)
            if ppid is None:
                continue
            resident, footprint = self._rusage(pid)
            result[pid] = ProcInfo(
                pid=pid, ppid=ppid, path=self._path(pid), rss=resident, footprint=footprint,
            )
        return result


class LinuxProcessTable:
    """Чтение процессов через /proc (для сборок под Linux)."""

    def snapshot(self) -> dict[int, ProcInfo]:
        result: dict[int, ProcInfo] = {}
        for entry in Path("/proc").iterdir():
            if not entry.name.isdigit():
                continue
            pid = int(entry.name)
            try:
                fields = (entry / "stat").read_text().rsplit(") ", 1)[1].split()
                ppid = int(fields[1])
                rss = int(fields[21]) * os.sysconf("SC_PAGE_SIZE")
                path = os.readlink(entry / "exe")
            except (OSError, IndexError, ValueError):
                continue
            result[pid] = ProcInfo(pid=pid, ppid=ppid, path=path, rss=rss, footprint=rss)
        return result


def process_table():
    if platform.system() == "Darwin":
        return MacProcessTable()
    return LinuxProcessTable()


def app_pids(table, marker: str = APP_MARKER) -> list[int]:
    return [info.pid for info in table.snapshot().values() if marker in info.path]


def log_candidates() -> list[Path]:
    home = Path.home()
    return [
        home / "Library/Logs/su.mhspx.glchat/glchat.log",
        home / ".local/share/su.mhspx.glchat/logs/glchat.log",
        home / ".config/su.mhspx.glchat/logs/glchat.log",
    ]


def log_file() -> Path | None:
    existing = [path for path in log_candidates() if path.exists()]
    if not existing:
        return None
    return max(existing, key=lambda path: path.stat().st_mtime)


def quit_app(table, timeout: float = 15.0) -> None:
    """Завершает запущенные экземпляры обёртки и ждёт их исчезновения."""
    for pid in app_pids(table):
        try:
            os.kill(pid, signal.SIGTERM)
        except OSError:
            pass
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        if not app_pids(table):
            time.sleep(0.6)  # окну и WebKit нужно время освободить ресурсы
            return
        time.sleep(0.1)
    raise SystemExit("не удалось завершить запущенный экземпляр glchat")


def measure_cold_start(table, app: Path, extra_args: list[str], timeout: float = 30.0) -> dict:
    """Запускает приложение и измеряет время до загрузки страницы клиента."""
    before = table.snapshot()
    log = log_file()
    offset = log.stat().st_size if log is not None else 0

    started = time.monotonic()
    subprocess.run(["open", "-a", str(app), "--args", *extra_args], check=True)

    pid: int | None = None
    pid_time: float | None = None
    page_time: float | None = None
    deadline = started + timeout
    while time.monotonic() < deadline:
        if pid is None:
            found = app_pids(table)
            if found:
                pid = found[0]
                pid_time = time.monotonic()
        if log is not None:
            with log.open("rb") as handle:
                handle.seek(offset)
                chunk = handle.read().decode("utf-8", "replace")
            if PAGE_LOADED in chunk:
                page_time = time.monotonic()
                break
        time.sleep(0.02)

    if pid is None or page_time is None:
        raise SystemExit("приложение не запустилось или страница не загрузилась: смотрите лог обёртки")

    return {
        "pid": pid,
        "open_ms": round((pid_time - started) * 1000),
        "page_ms": round((page_time - pid_time) * 1000),
        "total_ms": round((page_time - started) * 1000),
        "before_pids": set(before),
    }


def measure_memory(table, pid: int, before_pids: set[int], settle: float) -> dict:
    """Суммирует память обёртки и её процессов WebKit после паузы."""
    time.sleep(settle)
    snapshot = table.snapshot()
    own = snapshot.get(pid)
    helpers = [
        info for info in snapshot.values()
        if info.pid != pid and info.pid not in before_pids and "WebKit" in info.path
    ]
    rss = (own.rss if own else 0) + sum(info.rss for info in helpers)
    footprint = (own.footprint if own else 0) + sum(info.footprint for info in helpers)
    return {
        "rss_mb": round(rss / 1024 / 1024, 1),
        "footprint_mb": round(footprint / 1024 / 1024, 1),
        "app_rss_mb": round((own.rss if own else 0) / 1024 / 1024, 1),
        "webkit_rss_mb": round(sum(info.rss for info in helpers) / 1024 / 1024, 1),
        "webkit_processes": len(helpers),
    }


def main() -> int:
    parser = argparse.ArgumentParser(description="Замеры холодного старта и памяти glchat.app")
    parser.add_argument("--app", type=Path, default=DEFAULT_APP, help="путь к собранному .app")
    parser.add_argument("--runs", type=int, default=3, help="сколько холодных стартов измерить")
    parser.add_argument("--settle", type=float, default=20.0, help="пауза перед замером памяти, с")
    parser.add_argument("--json", type=Path, default=None, help="сохранить результаты в JSON")
    args = parser.parse_args()

    if not args.app.exists():
        print(f"нет собранного приложения: {args.app} (сначала make desktop-build)", file=sys.stderr)
        return 1

    table = process_table()
    results: dict[str, object] = {"app": str(args.app), "runs": []}

    # Настройки обёртки замер не меняет: приложение запускается как обычно,
    # автозапуск не включается и не выключается.
    for _ in range(args.runs):
        quit_app(table)
        run = measure_cold_start(table, args.app, [])
        run.update(measure_memory(table, run["pid"], run.pop("before_pids"), args.settle))
        results["runs"].append(run)
        print(
            f"старт: open→процесс {run['open_ms']} мс, процесс→страница {run['page_ms']} мс, "
            f"итого {run['total_ms']} мс; память RSS {run['rss_mb']} МБ "
            f"(обёртка {run['app_rss_mb']} + WebKit {run['webkit_rss_mb']}), "
            f"footprint {run['footprint_mb']} МБ"
        )

    # Простой в трее: окно скрыто, клиент загружен (closeToTray).
    quit_app(table)
    tray = measure_cold_start(table, args.app, ["--minimized"])
    tray.update(measure_memory(table, tray["pid"], tray.pop("before_pids"), args.settle))
    results["tray"] = tray
    print(
        f"трей:    старт {tray['total_ms']} мс; память RSS {tray['rss_mb']} МБ "
        f"(обёртка {tray['app_rss_mb']} + WebKit {tray['webkit_rss_mb']}), "
        f"footprint {tray['footprint_mb']} МБ"
    )

    cold = [run["total_ms"] for run in results["runs"]]
    results["summary"] = {
        "cold_ms_min": min(cold),
        "cold_ms_max": max(cold),
        "cold_ms_avg": round(sum(cold) / len(cold)),
        "ram_open_mb": max(run["rss_mb"] for run in results["runs"]),
        "ram_tray_mb": tray["rss_mb"],
        "budget_ram_mb": 350,
        "budget_cold_ms": 2000,
    }
    print(
        f"итог: холодный старт {results['summary']['cold_ms_min']}–"
        f"{results['summary']['cold_ms_max']} мс (бюджет 2000), "
        f"RAM открытого клиента {results['summary']['ram_open_mb']} МБ, "
        f"в трее {results['summary']['ram_tray_mb']} МБ (бюджет 350)"
    )

    if args.json is not None:
        args.json.parent.mkdir(parents=True, exist_ok=True)
        args.json.write_text(json.dumps(results, ensure_ascii=False, indent=2) + "\n")
        print(f"результаты: {args.json}")

    quit_app(table)
    return 0


if __name__ == "__main__":
    sys.exit(main())