#!/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())