From d70232bb9dfc89338fcb9ebc1c5ab23209deb3ce Mon Sep 17 00:00:00 2001 From: Will Miao Date: Fri, 13 Nov 2026 20:07:31 +0800 Subject: [PATCH] perf(benchmark): make the example-image A/B cache-neutral Civitai's image CDN is a Cloudflare edge cache: the rendition redirect is stored for 48 h and the blob for 1 h, so the same URL set is about 2.3x faster on a second visit (measured 2.82 vs 6.6 req/s at concurrency 6). The harness reused one URL set across modes and repeats, so the first mode through it paid every cold miss while later modes looked faster. - Resolve each tree's exact URLs with that tree's own builder (in a subprocess, so the old and new modules cannot alias), warm them before timing, then probe a sample and record the edge hit ratio per run. - Print a per-run table and report cache_clean_runs; warn instead of silently mixing cold runs into the medians. - Pin the rendition width in the sandbox settings so the warm-up and the server agree even when the user's own settings differ. - Skip models outside folder_paths roots: their sidecars resolve alongside the model file, so the old tree cannot read them and silently processed them as failed. --any-root opts back in. - Add reverify: populate the sandbox examples tree and time a library-wide Re-process All (--force) or Download Missing (--no-force) pass in ms/model, with --global to mirror the UI's global entry. - Default the old worktree under --workdir. --- scripts/benchmark_example_images.py | 944 +++++++++++++++++++++++++--- 1 file changed, 869 insertions(+), 75 deletions(-) diff --git a/scripts/benchmark_example_images.py b/scripts/benchmark_example_images.py index efcd940e..15717c8c 100644 --- a/scripts/benchmark_example_images.py +++ b/scripts/benchmark_example_images.py @@ -7,18 +7,54 @@ under ``--workdir``. It selects one model subset from the user's cache/sidecars, then runs the same subset against: * ``serial``: the pre-pipeline tree (or any tree), model concurrency forced to 1 +* ``new-serial``: the current tree with the model pipeline disabled * ``pipeline``: the current tree, model concurrency 2, AIMD ceiling = base 6 * ``aimd``: the current tree, model concurrency 2, AIMD ceiling 8 Each run starts from an empty examples directory and a fresh cache/sidecar copy. The script never writes to the real settings directory or model library. + +The A/B subset skips models outside ``folder_paths`` roots: those resolve their +sidecar alongside the model file, so the sandbox sidecar copy is not used and +the old tree cannot read their media at all. Pass ``--any-root`` to include +them (the ``--full`` selection always includes every model). + +CDN cache control +----------------- +Civitai's image CDN is a Cloudflare edge cache: the 301 that maps a rendition +to its blob is stored for 48 h and the blob itself for 1 h. The same URL set +therefore gets much faster after its first visit (measured: 2.8 req/s cold vs +6.6 req/s warm at concurrency 6). Because every mode of this harness requests +the *same* URLs, an un-warmed baseline is slower than every later mode and the +A/B reports a fake gain. + +``--warm-cache`` (default for ``compare``) fetches every URL the mode will +request *before* timing starts, then probes a sample and records the edge hit +ratio in the result. A run whose probe hit ratio is below 1.0 is reported as a +cache warning instead of being silently mixed into the median. A single ``run`` +(e.g. the full-library validation) skips warming unless the flag is passed, so +it does not pay for a second full download. Use ``--no-warm-cache`` only to +measure the edge state as-is; read the per-run table rather than the medians in +that mode. + +``reverify`` +------------ +``compare`` only measures the download phase. A large library whose images are +already on disk spends its time in per-model bookkeeping instead, which the +download benchmark cannot see. The ``reverify`` command populates the sandbox +examples tree from the selected sidecars and times a library-wide +``force=True`` pass (the UI's *Re-process All*) or a ``force=False`` skip pass, +reporting milliseconds per model. """ from __future__ import annotations import argparse +import asyncio import json import os +import random +import re import shutil import sqlite3 import subprocess @@ -42,6 +78,51 @@ MODES = { "aimd": {"model_concurrency": 2, "probe_ceiling": 8}, } +# Rendition width used both in the sandbox settings and in the cache warm-up. +# Pinned so the warm-up requests exactly the URLs the server will request. +DEFAULT_IMAGE_WIDTH = 800 +# Number of URLs re-fetched after the warm-up pass to verify the edge state. +CACHE_PROBE_SAMPLE = 24 +# Names of the URLs each tree would request, resolved inside that tree so the +# warm-up cannot drift from the code under test. +_URL_DUMP_SCRIPT = r""" +import json, re, sys + +payload = json.load(sys.stdin) +width = payload["width"] +images = payload["images"] + +try: + from py.utils.civitai_utils import build_civitai_media_url as build + builder = "civitai_utils.build_civitai_media_url" +except Exception: + build = None + builder = "legacy.get_civitai_optimized_url" + + +def legacy(url): + match = re.match(r"(https://image\.civitai\.com/[^/]+/[^/]+)", url) + return f"{match.group(1)}/optimized=true" if match else url + + +urls = [] +for item in images: + url = (item.get("url") or "").strip() + if not url: + continue + if build is not None: + urls.append(build(url, item.get("type"), width=width, optimize=True) or url) + else: + urls.append(legacy(url)) + +json.dump({"builder": builder, "urls": urls}, sys.stdout) +""" + +_IMAGE_INDEX_CACHE: dict[str, dict[str, list[dict[str, Any]]]] = {} +# The pre-`build_civitai_media_url` helper matched the first two path segments +# and replaced everything after them with `optimized=true`. +_LEGACY_URL_PATTERN = re.compile(r"(https://image\.civitai\.com/[^/]+/[^/]+)") + def _log(message: str) -> None: print(message, flush=True) @@ -87,6 +168,229 @@ def _wait_for_server(base_url: str, proc: subprocess.Popen, timeout: float = 120 raise TimeoutError(f"server did not become ready: {last_error}") +def _configured_model_roots() -> list[Path]: + """Return the ``folder_paths`` roots from the real settings file.""" + + try: + settings = _read_json(REAL_CONFIG / "settings.json") + except (OSError, ValueError): + return [] + roots: list[Path] = [] + for values in (settings.get("folder_paths") or {}).values(): + if isinstance(values, str): + values = [values] + if not isinstance(values, list): + continue + for value in values: + if isinstance(value, str) and value.strip(): + roots.append(Path(os.path.normcase(os.path.abspath(value)))) + return roots + + +def _is_under_roots(path: Path, roots: list[Path]) -> bool: + """Return whether *path* lives under one of *roots* (empty = allow all).""" + + if not roots: + return True + candidate = os.path.normcase(os.path.abspath(str(path))) + for root in roots: + root_text = str(root) + if candidate == root_text or candidate.startswith(root_text + os.sep): + return True + return False + + +def _sidecar_image_index(root: Path) -> dict[str, list[dict[str, Any]]]: + """Return ``{sha256: [{"url", "type"}]}`` for every sidecar under *root*. + + Only the media list is kept, so a full-library index costs far less memory + than parsing the sidecars into dicts and retaining them. The index is + memoized per root: ``compare`` builds it once for four modes and repeats. + """ + + key = str(root) + cached = _IMAGE_INDEX_CACHE.get(key) + if cached is not None: + return cached + + index: dict[str, list[dict[str, Any]]] = {} + for path in root.rglob("*.metadata.json"): + try: + payload = _read_json(path) + except (OSError, ValueError): + continue + sha = str(payload.get("sha256") or "").lower() + if len(sha) != 64: + continue + civitai = payload.get("civitai") or {} + media = [] + for image in civitai.get("images") or []: + if not isinstance(image, dict): + continue + url = (image.get("url") or "").strip() + if url: + media.append({"url": url, "type": image.get("type")}) + index.setdefault(sha, media) + + _IMAGE_INDEX_CACHE[key] = index + _log(f"indexed media for {len(index)} sidecars under {root}") + return index + + +def _selected_images(workdir: Path, selection: dict[str, Any]) -> list[dict[str, Any]]: + """Return the media entries the selected models will ask the server for.""" + + index = _sidecar_image_index(workdir / "sidecars") + images: list[dict[str, Any]] = [] + for model in selection["models"]: + images.extend(index.get(str(model["sha256"]).lower(), [])) + return images + + +def _legacy_media_url(url: str) -> str: + """Reproduce the pre-``build_civitai_media_url`` rendition URL.""" + + match = _LEGACY_URL_PATTERN.match(url) + return f"{match.group(1)}/optimized=true" if match else url + + +def _tree_media_urls( + tree: Path, images: list[dict[str, Any]], width: int +) -> tuple[list[str], str]: + """Resolve the URLs *tree* would request, using that tree's own builder. + + The builder runs in a subprocess with ``cwd=tree``: the old and new trees + define the same module names, so importing both in one process would alias + whichever was imported first. If the tree cannot be loaded, the harness + falls back to the frozen legacy rewrite and says so. + """ + + if not images: + return [], "none" + payload = json.dumps({"width": width, "images": images}) + try: + proc = subprocess.run( + [sys.executable, "-c", _URL_DUMP_SCRIPT], + cwd=str(tree), + input=payload, + capture_output=True, + text=True, + timeout=180, + ) + if proc.returncode == 0: + data = json.loads(proc.stdout) + return list(data["urls"]), str(data.get("builder") or "unknown") + _log(f"warning: URL builder in {tree} failed: {proc.stderr.strip()[-300:]}") + except (OSError, subprocess.SubprocessError, ValueError, KeyError) as exc: + _log(f"warning: URL builder in {tree} failed: {exc}") + return [_legacy_media_url(str(item.get("url") or "")) for item in images], "legacy-fallback" + + +async def _http_pass( + urls: list[str], concurrency: int = 6 +) -> dict[str, Any]: + """Fetch *urls* concurrently and summarize the edge cache state.""" + + import aiohttp + + stats: dict[str, Any] = { + "requests": len(urls), + "hit": 0, + "miss": 0, + "other": 0, + "errors": 0, + "bytes": 0, + } + latencies: list[float] = [] + if not urls: + stats["elapsed_seconds"] = 0.0 + stats["latency_p50_ms"] = 0.0 + return stats + + semaphore = asyncio.Semaphore(max(1, concurrency)) + timeout = aiohttp.ClientTimeout(total=120) + connector = aiohttp.TCPConnector(limit=max(1, concurrency)) + + async with aiohttp.ClientSession(timeout=timeout, connector=connector) as session: + + async def fetch(url: str) -> None: + async with semaphore: + started = time.perf_counter() + try: + async with session.get(url, allow_redirects=True) as response: + body = await response.read() + except Exception: # noqa: BLE001 - a failed probe must not abort the run + stats["errors"] += 1 + return + state = (response.headers.get("cf-cache-status") or "").upper() + if state == "HIT": + stats["hit"] += 1 + elif state == "MISS": + stats["miss"] += 1 + else: + stats["other"] += 1 + stats["bytes"] += len(body) + latencies.append((time.perf_counter() - started) * 1000.0) + + started = time.perf_counter() + await asyncio.gather(*(fetch(url) for url in urls)) + stats["elapsed_seconds"] = round(time.perf_counter() - started, 3) + + if latencies: + latencies.sort() + stats["latency_p50_ms"] = round(latencies[len(latencies) // 2], 1) + else: + stats["latency_p50_ms"] = 0.0 + return stats + + +def _warm_cdn_cache( + urls: list[str], + *, + concurrency: int = 6, + probe_sample: int = CACHE_PROBE_SAMPLE, +) -> dict[str, Any]: + """Warm every URL a run will request, then verify the edge is hot. + + Civitai's CDN caches both the rendition redirect (48 h) and the blob (1 h), + so an un-warmed first pass is ~2.3x slower than every later pass. Warming + outside the timed run and probing afterwards makes every mode comparable + and records whether that assumption actually held. + """ + + if not urls: + return {"enabled": False, "hit_ratio": None, "urls": 0} + + _log(f"warming {len(urls)} CDN URLs before timing ...") + warm = asyncio.run(_http_pass(urls, concurrency)) + _log( + f"warm-up: {warm['elapsed_seconds']:.1f}s " + f"{warm['bytes'] / 1e6:.1f} MB " + f"(hit={warm['hit']} miss={warm['miss']} other={warm['other']} " + f"errors={warm['errors']})" + ) + + sample = ( + list(urls) + if len(urls) <= probe_sample + else random.Random(0).sample(urls, probe_sample) + ) + probe = asyncio.run(_http_pass(sample, concurrency)) + hit_ratio = probe["hit"] / max(probe["requests"], 1) + if hit_ratio < 1.0: + _log( + f"WARNING: only {probe['hit']}/{probe['requests']} probed URLs are edge-cached; " + "this run is not cache-neutral" + ) + return { + "enabled": True, + "urls": len(urls), + "warmup": warm, + "probe": probe, + "hit_ratio": round(hit_ratio, 3), + } + + def _index_sidecars() -> tuple[Path, dict[str, Path]]: """Index real centralized sidecars by sha256 without touching settings. @@ -182,8 +486,11 @@ def prepare(args: argparse.Namespace) -> None: ).fetchall() connection.close() + model_roots = [] if args.any_root else _configured_model_roots() + selected: list[dict[str, Any]] = [] total_images = 0 + outside_roots = 0 for row in rows: if len(selected) >= args.count or total_images >= args.images: break @@ -197,6 +504,13 @@ def prepare(args: argparse.Namespace) -> None: file_path = Path(row["file_path"] or "") if not file_path.is_file(): continue + if not _is_under_roots(file_path, model_roots): + # ``extra_folder_paths`` models resolve their sidecar alongside the + # model file, so the sandbox sidecar copy is not used and the old + # tree cannot read their media at all. Exclude them from the A/B + # unless --any-root asks for them. + outside_roots += 1 + continue metadata_path = sidecar_by_hash.get(sha) if metadata_path is None or not metadata_path.is_file(): continue @@ -256,13 +570,29 @@ def prepare(args: argparse.Namespace) -> None: f"selected {len(selected)} models / {total_images} images " f"-> {workdir / 'selection.json'}" ) + if outside_roots: + _log( + f"skipped {outside_roots} models outside folder_paths roots " + "(their sidecars resolve alongside the model file, which the old " + "tree cannot read; pass --any-root to include them)" + ) def _build_settings( run_dir: Path, - mode: str, examples_path: Path, + *, + model_concurrency: int, + probe_ceiling: int, + image_width: int = DEFAULT_IMAGE_WIDTH, ) -> dict[str, Any]: + """Write deterministic sandbox settings for one run. + + The rendition width is pinned here (and used by the cache warm-up) so a + user's own ``settings.json`` cannot silently change the URL set between + runs or between the warm-up and the timed run. + """ + source = REAL_CONFIG / "settings.json" settings = _read_json(source) settings.update( @@ -274,8 +604,10 @@ def _build_settings( "auto_download_example_images": False, "optimize_example_images": True, "example_images_download_concurrency": 6, - "example_images_download_max_concurrency": MODES[mode]["probe_ceiling"], - "example_images_model_concurrency": MODES[mode]["model_concurrency"], + "example_images_download_max_concurrency": probe_ceiling, + "example_images_model_concurrency": model_concurrency, + "example_images_max_width": image_width, + "example_images_video_max_width": 450, } ) return settings @@ -288,7 +620,11 @@ def _media_stats(root: Path) -> tuple[int, int]: return files, size for current, _dirs, names in os.walk(root): for name in names: - if name.startswith(".lm-examples-complete") or name == ".download_progress.json": + if ( + name.startswith(".lm-examples-complete") + or name == ".download_progress.json" + or name == ".lm-examples-backfill" + ): continue path = Path(current) / name try: @@ -320,28 +656,21 @@ def _terminate(proc: subprocess.Popen) -> None: proc.wait(timeout=5) -def run_once( +def _run_download_pass( tree: Path, - workdir: Path, - selection: dict[str, Any], - mode: str, + run_dir: Path, port: int, -) -> dict[str, Any]: - stamp = time.strftime("%Y%m%d-%H%M%S") - run_dir = workdir / "runs" / f"{mode}-{stamp}-{os.getpid()}" - run_dir.mkdir(parents=True, exist_ok=True) - _log(f"[{mode}] preparing {run_dir}") + payload: dict[str, Any], + *, + label: str, + timeout: float, +) -> tuple[float, dict[str, Any], float, Path]: + """Start *tree* and time one ``download-example-images`` pass. - shutil.copytree(workdir / "cache", run_dir / "cache", dirs_exist_ok=True) - shutil.copytree(workdir / "sidecars", run_dir / "sidecars", dirs_exist_ok=True) - - examples_path = workdir / "examples" - if examples_path.exists(): - shutil.rmtree(examples_path) - examples_path.mkdir(parents=True, exist_ok=True) - - settings = _build_settings(run_dir, mode, examples_path) - _write_json(run_dir / "settings.json", settings) + Returns ``(elapsed_seconds, status, startup_seconds, log_path)``. The timer + starts after the POST is accepted and stops when the manager reports a + terminal state, so server startup and setting up the sandbox are excluded. + """ log_path = run_dir / "server.log" log_handle = log_path.open("w", encoding="utf-8") @@ -356,7 +685,7 @@ def run_once( "--settings-path", str(run_dir), ] - _log(f"[{mode}] starting {tree} on :{port}") + _log(f"[{label}] starting {tree} on :{port}") proc = subprocess.Popen( command, cwd=str(tree), @@ -369,27 +698,18 @@ def run_once( try: _wait_for_server(base_url, proc) ready = time.monotonic() - model_hashes = [model["sha256"] for model in selection["models"]] - response = _http_json( - base_url + "/api/lm/download-example-images", - { - "model_types": ["lora", "checkpoint", "embedding"], - "model_hashes": model_hashes, - "delay": 0, - "optimize": True, - }, - ) + response = _http_json(base_url + "/api/lm/download-example-images", payload) if not response.get("success"): - raise RuntimeError(f"download rejected: {response}") + raise RuntimeError(f"{label} rejected: {response}") - download_start = time.monotonic() - deadline = download_start + 2700 + pass_start = time.monotonic() + deadline = pass_start + timeout status: dict[str, Any] = {} while time.monotonic() < deadline: - payload = _http_json(base_url + "/api/lm/example-images-status") - status = payload.get("status") or {} + polled = _http_json(base_url + "/api/lm/example-images-status") + status = polled.get("status") or {} state = status.get("status") - if not payload.get("is_downloading") and state in { + if not polled.get("is_downloading") and state in { "completed", "stopped", "error", @@ -397,39 +717,104 @@ def run_once( break time.sleep(0.5) else: - raise TimeoutError("download did not finish within 2700s") - elapsed = time.monotonic() - download_start - files, size = _media_stats(examples_path) - logs = _count_log_events(log_path) - return { - "mode": mode, - "tree": str(tree), - "commit": subprocess.check_output( - ["git", "-C", str(tree), "rev-parse", "--short", "HEAD"], - text=True, - ).strip(), - "dirty": subprocess.run( - ["git", "-C", str(tree), "diff", "--quiet"], check=False - ).returncode - != 0, - "model_count": selection["model_count"], - "total_images": selection["total_images"], - "elapsed_seconds": round(elapsed, 3), - "startup_seconds": round(ready - started, 3), - "status": status.get("status"), - "total": status.get("total"), - "completed": status.get("completed"), - "failed_count": len(status.get("failed_models") or []), - "files": files, - "bytes": size, - **logs, - "run_dir": str(run_dir), - } + raise TimeoutError(f"{label} did not finish within {timeout}s") + return time.monotonic() - pass_start, status, ready - started, log_path finally: _terminate(proc) log_handle.close() +def run_once( + tree: Path, + workdir: Path, + selection: dict[str, Any], + mode: str, + port: int, + *, + image_width: int = DEFAULT_IMAGE_WIDTH, + warm_cache: bool = True, + download_timeout: float = 2700.0, +) -> dict[str, Any]: + stamp = time.strftime("%Y%m%d-%H%M%S") + run_dir = workdir / "runs" / f"{mode}-{stamp}-{os.getpid()}" + run_dir.mkdir(parents=True, exist_ok=True) + _log(f"[{mode}] preparing {run_dir}") + + shutil.copytree(workdir / "cache", run_dir / "cache", dirs_exist_ok=True) + shutil.copytree(workdir / "sidecars", run_dir / "sidecars", dirs_exist_ok=True) + + examples_path = workdir / "examples" + if examples_path.exists(): + shutil.rmtree(examples_path) + examples_path.mkdir(parents=True, exist_ok=True) + + settings = _build_settings( + run_dir, + examples_path, + model_concurrency=MODES[mode]["model_concurrency"], + probe_ceiling=MODES[mode]["probe_ceiling"], + image_width=image_width, + ) + _write_json(run_dir / "settings.json", settings) + + # Warm (or explicitly skip warming) the exact URLs this tree will request. + # Civitai's edge caches the rendition redirect and the blob, so without this + # the first mode through a URL set is 2-3x slower than every later mode. + cache: dict[str, Any] = {"enabled": False, "hit_ratio": None} + if warm_cache: + images = _selected_images(workdir, selection) + urls, builder = _tree_media_urls(tree, images, image_width) + cache = _warm_cdn_cache(urls) + cache["builder"] = builder + else: + _log( + f"[{mode}] cache warm-up disabled; the CDN state is unverified and " + "mode order biases the comparison" + ) + + elapsed, status, startup_seconds, log_path = _run_download_pass( + tree, + run_dir, + port, + { + "model_types": ["lora", "checkpoint", "embedding"], + "model_hashes": [model["sha256"] for model in selection["models"]], + "delay": 0, + "optimize": True, + }, + label=mode, + timeout=download_timeout, + ) + + files, size = _media_stats(examples_path) + logs = _count_log_events(log_path) + return { + "mode": mode, + "tree": str(tree), + "commit": subprocess.check_output( + ["git", "-C", str(tree), "rev-parse", "--short", "HEAD"], + text=True, + ).strip(), + "dirty": subprocess.run( + ["git", "-C", str(tree), "diff", "--quiet"], check=False + ).returncode + != 0, + "model_count": selection["model_count"], + "total_images": selection["total_images"], + "elapsed_seconds": round(elapsed, 3), + "startup_seconds": round(startup_seconds, 3), + "status": status.get("status"), + "total": status.get("total"), + "completed": status.get("completed"), + "failed_count": len(status.get("failed_models") or []), + "files": files, + "bytes": size, + "cache": cache, + **logs, + "run_dir": str(run_dir), + } + + def _median(values: list[float]) -> float: ordered = sorted(values) middle = len(ordered) // 2 @@ -438,6 +823,16 @@ def _median(values: list[float]) -> float: return (ordered[middle - 1] + ordered[middle]) / 2 +def _cache_hit_ratio(result: dict[str, Any]) -> float | None: + """Return the verified edge hit ratio, or None when it was not measured.""" + + cache = result.get("cache") or {} + if not cache.get("enabled"): + return None + ratio = cache.get("hit_ratio") + return None if ratio is None else float(ratio) + + def compare(args: argparse.Namespace) -> None: workdir = Path(args.workdir).resolve() selection_path = workdir / "selection.json" @@ -479,14 +874,26 @@ def compare(args: argparse.Namespace) -> None: order = plans if repeat % 2 == 0 else list(reversed(plans)) for mode, tree in order: _log(f"=== repeat {repeat + 1}/{args.repeats} mode={mode} ===") - result = run_once(tree, workdir, selection, mode, port) + result = run_once( + tree, + workdir, + selection, + mode, + port, + image_width=args.image_width, + warm_cache=args.warm_cache, + download_timeout=args.download_timeout, + ) + result["repeat"] = repeat + 1 results.append(result) port += 1 + cache = result.get("cache") or {} _log( f"[{mode}] {result['elapsed_seconds']:.1f}s " f"{result['files']} files {result['bytes'] / 1_000_000:.1f} MB " f"completed={result['completed']}/{result['total']} " - f"429={result['rate_limited']}" + f"429={result['rate_limited']} " + f"cache={cache.get('hit_ratio', 1.0):.0%}" ) summary: dict[str, Any] = {"selection": selection, "results": results, "modes": {}} @@ -494,11 +901,20 @@ def compare(args: argparse.Namespace) -> None: rows = [row for row in results if row["mode"] == mode] if not rows: continue + clean = 0 + for row in rows: + ratio = _cache_hit_ratio(row) + if ratio is not None and ratio >= 0.999: + clean += 1 elapsed = _median([row["elapsed_seconds"] for row in rows]) files = _median([float(row["files"]) for row in rows]) size = _median([float(row["bytes"]) for row in rows]) summary["modes"][mode] = { "runs": len(rows), + "cache_verified_runs": sum( + 1 for row in rows if _cache_hit_ratio(row) is not None + ), + "cache_clean_runs": clean if args.warm_cache else None, "median_elapsed_seconds": round(elapsed, 3), "median_files": int(files), "median_bytes": int(size), @@ -521,20 +937,287 @@ def compare(args: argparse.Namespace) -> None: output = workdir / "results.json" _write_json(output, summary) + + # Per-run table first: with --no-warm-cache the first run through a URL set + # is cold and the medians are not comparable, so the raw runs must be + # readable without opening results.json. _log("") - _log("mode median_s speedup files MB 429 404 probe") + _log("repeat mode wall_s files MB 429 404 probe cache") + for row in results: + ratio = _cache_hit_ratio(row) + cache_text = " cold" if ratio is None else f"{ratio:>6.0%}" + _log( + f"{row.get('repeat', 0):<6} {row['mode']:<10} " + f"{row['elapsed_seconds']:>7.1f} " + f"{row['files']:>6} " + f"{row['bytes'] / 1_000_000:>6.1f} " + f"{row['rate_limited']:>5} {row['not_found']:>4} " + f"{row.get('probe_steps', 0):>6} " + f"{cache_text}" + ) + + _log("") + _log("mode median_s speedup files MB 429 404 probe clean") for mode in MODES: data = summary["modes"].get(mode) if not data: continue + clean = data["cache_clean_runs"] + clean_text = "-" if clean is None else f"{clean}/{data['runs']}" _log( f"{mode:<10} {data['median_elapsed_seconds']:>9.1f} " f"{data.get('speedup_vs_new_serial', 0):>7.3f} " f"{data['median_files']:>6} " f"{data['median_bytes'] / 1_000_000:>7.1f} " f"{data['rate_limited']:>5} {data['not_found']:>4} " - f"{data.get('probe_steps', 0):>6}" + f"{data.get('probe_steps', 0):>6} " + f"{clean_text:>5}" ) + if args.warm_cache and any( + data["cache_clean_runs"] is not None + and data["cache_clean_runs"] < data["runs"] + for data in summary["modes"].values() + ): + _log("") + _log( + "WARNING: some runs were measured with a cold CDN edge cache. The " + "medians are not cache-neutral; re-run with --warm-cache (default) " + "or compare only the 'clean' runs." + ) + if not args.warm_cache: + _log("") + _log( + "NOTE: --no-warm-cache does not force cold URLs; it measures whatever " + "state the edge happens to be in, so mode order biases the result " + "and the medians are not comparable." + ) + _log(f"results -> {output}") + + +def _cached_model_hashes(workdir: Path) -> list[str]: + """Return every model hash in the sandbox cache copy.""" + + db_path = workdir / "cache" / "model" / "comfyui.sqlite" + if not db_path.is_file(): + return [] + connection = sqlite3.connect(str(db_path)) + try: + rows = connection.execute( + "select sha256 from models where model_type in ('lora','checkpoint','embedding')" + ).fetchall() + finally: + connection.close() + return [str(row[0]).lower() for row in rows if row[0]] + + +def _populate_examples( + workdir: Path, + selection: dict[str, Any], + *, + with_markers: bool, + all_cached: bool = False, +) -> dict[str, int]: + """Create an already-populated examples tree. + + Each media entry gets a placeholder file named the way the download would + have named it, so the re-verify pass performs its real checks (folder scan, + per-image existence stat, sidecar read) without making a single request. + ``with_markers`` writes the completion marker an upgraded library has after + its first pass; without it the run models a legacy library. + + ``all_cached`` populates every model in the sandbox cache instead of only + the selected ones. The skip guard only looks at "folder has files" plus the + marker, so this is what a fully verified library looks like to a + ``force=False`` pass even when the sandbox has sidecars for a few models + only. + """ + + index = _sidecar_image_index(workdir / "sidecars") + examples = workdir / "examples" + if examples.exists(): + shutil.rmtree(examples) + examples.mkdir(parents=True) + + if all_cached: + targets = [ + {"sha256": sha, "media": index.get(sha) or [{"type": None}]} + for sha in _cached_model_hashes(workdir) + ] + else: + targets = [ + { + "sha256": str(model["sha256"]).lower(), + "media": index.get(str(model["sha256"]).lower()) or [], + } + for model in selection["models"] + ] + + models = 0 + files = 0 + markers = 0 + for target in targets: + sha = target["sha256"] + media = target["media"] + if not all_cached and not media: + continue + folder = examples / sha + folder.mkdir(parents=True, exist_ok=True) + for position, item in enumerate(media): + extension = ".mp4" if (item.get("type") or "").lower() == "video" else ".jpeg" + (folder / f"image_{position}{extension}").write_bytes(b"x") + files += 1 + if with_markers: + (folder / ".lm-examples-complete").write_text( + json.dumps({"media_count": len(media)}) + "\n", encoding="utf-8" + ) + markers += 1 + models += 1 + + _log( + f"populated {models} model folders, {files} placeholder media files, " + f"{markers} completion markers" + ) + return {"models": models, "files": files, "markers": markers} + + +def run_reverify( + tree: Path, + workdir: Path, + *, + force: bool, + model_concurrency: int, + port: int, + model_hashes: list[str] | None = None, + image_width: int = DEFAULT_IMAGE_WIDTH, + download_timeout: float = 2700.0, +) -> dict[str, Any]: + """Time a re-verify or skip pass over a populated tree. + + This is the workload of a large library whose images are already on disk: + the download benchmark cannot see it because no media request is made. + + ``force=True`` without ``model_hashes`` mirrors the UI's global + *Re-process All* entry: the app's force path skips a verified folder unless + the model is on the failed list. Passing ``model_hashes`` instead models a + per-model / bulk *Re-process*, where explicit targets always bypass the + guard. ``reverify --global`` omits the hashes; use it with ``--full`` so + every cached model is in scope. + + ``force=False`` mirrors *Download Missing* and must not pass hashes (which + would mark the models as explicit targets and bypass the skip guard). The + tree is populated for every cached model instead, which is the state the + skip guard sees in a verified library. + """ + + label = "reverify-force" if force else "reverify-skip" + stamp = time.strftime("%Y%m%d-%H%M%S") + run_dir = workdir / "runs" / f"{label}-{stamp}-{os.getpid()}" + run_dir.mkdir(parents=True, exist_ok=True) + _log(f"[{label}] preparing {run_dir}") + + shutil.copytree(workdir / "cache", run_dir / "cache", dirs_exist_ok=True) + shutil.copytree(workdir / "sidecars", run_dir / "sidecars", dirs_exist_ok=True) + + examples_path = workdir / "examples" + examples_path.mkdir(parents=True, exist_ok=True) + + settings = _build_settings( + run_dir, + examples_path, + model_concurrency=model_concurrency, + probe_ceiling=8, + image_width=image_width, + ) + _write_json(run_dir / "settings.json", settings) + + request: dict[str, Any] = { + "force": force, + "optimize": True, + "model_types": ["lora", "checkpoint", "embedding"], + } + if model_hashes: + request["model_hashes"] = model_hashes + + elapsed, status, startup_seconds, _log_path = _run_download_pass( + tree, + run_dir, + port, + request, + label=label, + timeout=download_timeout, + ) + + completed = int(status.get("completed") or 0) + return { + "mode": label, + "force": force, + "explicit_targets": bool(model_hashes), + "model_concurrency": model_concurrency, + "tree": str(tree), + "commit": subprocess.check_output( + ["git", "-C", str(tree), "rev-parse", "--short", "HEAD"], text=True + ).strip(), + "elapsed_seconds": round(elapsed, 3), + "startup_seconds": round(startup_seconds, 3), + "status": status.get("status"), + "total": status.get("total"), + "completed": completed, + "failed_count": len(status.get("failed_models") or []), + "ms_per_model": round(1000.0 * elapsed / completed, 2) if completed else None, + "models_per_second": round(completed / elapsed, 2) if completed else None, + "run_dir": str(run_dir), + } + + +def reverify(args: argparse.Namespace) -> None: + """Run the populated-tree re-verify benchmark one or more times.""" + + workdir = Path(args.workdir).resolve() + workdir.mkdir(parents=True, exist_ok=True) + if not (workdir / "selection.json").is_file(): + prepare(args) + selection = _read_json(workdir / "selection.json") + + tree = Path(args.tree).resolve() + if not (tree / "standalone.py").is_file(): + raise FileNotFoundError(f"tree missing standalone.py: {tree}") + + if not (workdir / "examples").is_dir() or not args.reuse_examples: + _populate_examples( + workdir, + selection, + with_markers=args.markers, + all_cached=not args.force, + ) + + hashes = [str(model["sha256"]).lower() for model in selection["models"]] + results: list[dict[str, Any]] = [] + port = args.port + for repeat in range(args.repeat): + _log(f"=== reverify {repeat + 1}/{args.repeat} force={args.force} ===") + result = run_reverify( + tree, + workdir, + force=args.force, + model_concurrency=args.model_concurrency, + port=port, + model_hashes=hashes if (args.force and not args.global_run) else None, + image_width=args.image_width, + download_timeout=args.download_timeout, + ) + result["repeat"] = repeat + 1 + results.append(result) + port += 1 + _log( + f"[{result['mode']}] {result['elapsed_seconds']:.1f}s " + f"{result['completed']}/{result['total']} models " + f"{result['ms_per_model']} ms/model " + f"failed={result['failed_count']}" + ) + + summary = {"selection": selection, "results": results} + output = workdir / "reverify.json" + _write_json(output, summary) _log(f"results -> {output}") @@ -548,28 +1231,139 @@ def main() -> None: parser.add_argument("--full", action="store_true", help="select all example-capable models") parser.add_argument("--repeats", type=int, default=2) parser.add_argument("--port", type=int, default=18200) - parser.add_argument("--old-tree", default=str(DEFAULT_WORKDIR / "old-tree")) + parser.add_argument( + "--old-tree", + default=None, + help="old worktree path (default: /old-tree)", + ) parser.add_argument("--old-ref", default="9ed75e8f") parser.add_argument("--new-tree", default=str(REPO_ROOT)) parser.add_argument("--tree", default=str(REPO_ROOT)) parser.add_argument("--mode", choices=sorted(MODES), default="serial") + parser.add_argument( + "--any-root", + action="store_true", + help=( + "include models outside folder_paths roots; their sidecars are not " + "sandboxed and the old tree cannot read them" + ), + ) + parser.add_argument( + "--image-width", + type=int, + default=DEFAULT_IMAGE_WIDTH, + help="example image rendition width pinned for every run", + ) + parser.add_argument( + "--warm-cache", + dest="warm_cache", + action="store_true", + default=None, + help="warm the CDN edge for the run's URLs before timing (default: compare only)", + ) + parser.add_argument( + "--no-warm-cache", + dest="warm_cache", + action="store_false", + help=( + "do not warm or verify the CDN; measures whatever edge state the " + "URLs happen to be in, so mode order biases the result" + ), + ) + parser.add_argument( + "--download-timeout", + type=float, + default=2700.0, + help="seconds to wait for one download pass", + ) + parser.add_argument( + "--force", + dest="force", + action="store_true", + default=True, + help="reverify only: mirror the UI's Re-process All pass", + ) + parser.add_argument( + "--no-force", + dest="force", + action="store_false", + help="reverify only: mirror the UI's Download Missing skip pass", + ) + parser.add_argument( + "--markers", + dest="markers", + action="store_true", + default=True, + help="reverify only: populate completion markers (post-upgrade library)", + ) + parser.add_argument( + "--no-markers", + dest="markers", + action="store_false", + help="reverify only: leave folders unmarked (legacy library)", + ) + parser.add_argument( + "--model-concurrency", + type=int, + default=2, + help="reverify only: model pipeline width", + ) + parser.add_argument( + "--global", + dest="global_run", + action="store_true", + help=( + "reverify only: omit model_hashes to mirror the UI's global entry; " + "the pass then covers every model in the sandbox cache (use --full)" + ), + ) + parser.add_argument( + "--reuse-examples", + action="store_true", + help=( + "reverify only: keep the current examples tree instead of rebuilding " + "it; the tree must match the selected mode (marked/unmarked)" + ), + ) + parser.add_argument( + "--repeat", + type=int, + default=1, + help="reverify only: number of passes over the populated tree", + ) parser.add_argument( "command", - choices=["prepare", "run", "compare"], + choices=["prepare", "run", "compare", "reverify"], nargs="?", default="compare", ) args = parser.parse_args() + if args.old_tree is None: + args.old_tree = str(Path(args.workdir).resolve() / "old-tree") + if args.warm_cache is None: + # Comparison needs cache-neutral runs; a single `run` (e.g. the + # full-library validation) should not pay for a second full download + # unless the caller asks for it. + args.warm_cache = args.command == "compare" if args.command == "prepare": prepare(args) elif args.command == "run": workdir = Path(args.workdir).resolve() selection = _read_json(workdir / "selection.json") result = run_once( - Path(args.tree).resolve(), workdir, selection, args.mode, args.port + Path(args.tree).resolve(), + workdir, + selection, + args.mode, + args.port, + image_width=args.image_width, + warm_cache=args.warm_cache, + download_timeout=args.download_timeout, ) _write_json(workdir / f"run-{args.mode}.json", result) _log(json.dumps(result, indent=2)) + elif args.command == "reverify": + reverify(args) else: compare(args)