#!/usr/bin/env python3 """Generate, validate, benchmark and remove the temporary ~1 GiB PDF workload. Use --xvfb for a dedicated virtual X11 display. Chromium/worker sandboxes remain enabled. The 3-open/100-operation run is a stress observation, not a 30-open performance acceptance run or a representative scanned-book compatibility test. """ import argparse import hashlib import json import math import os from pathlib import Path import re import select import signal import statistics import subprocess import tempfile import time from generate_stress_pdf import ROOT, generate, validate_source RUNS, OPERATIONS, CLOSE_CYCLES = 3, 100, 2 def file_hash(path): with Path(path).open("rb") as source: return hashlib.file_digest(source, "sha256").hexdigest() def executable_hashes(binary): return {name: file_hash(binary if name == "docview" else binary.parent / name) for name in ("docview", "docview-pdf-worker", "docview-archive-worker")} def clear_previous_evidence(output): # Only this runner's known outputs are replaced. Historical runs elsewhere # and unrelated files remain untouched, including on startup failure. for name in ("report.json", "benchmark.json", "source-manifest.json", "pdfinfo.txt", "benchmark.log", "last-page.log", "last-page/state.txt", "last-page/runtime.json", "last-page/screen.png"): (output / name).unlink(missing_ok=True) def summarize_benchmark(result, source_hash): actions, opens = result.get("operations", []), result.get("openFrameMs", []) cycles = result.get("openCloseCycles", []) uniform = [v for v in cycles if v.get("workload") == "open-first-visible-only"] final = [v for v in cycles if v.get("workload") == "navigation-and-scroll"] page_actions = [v for v in actions if v.get("command") == "nav.page"] noops = sum(v.get("positionChanged") is not True for v in page_actions) finite = lambda values: all(type(v) in (int, float) and math.isfinite(v) and v >= 0 for v in values) checks = { "schema": result.get("schemaVersion") == 2, "harness_success": result.get("success") is True and result.get("format") == "pdf", "requested_workload": result.get("requestedOpenTimings") == RUNS and result.get("requestedCloseCycles") == CLOSE_CYCLES, "actual_three_opens": result.get("openCount") == RUNS and len(opens) == RUNS and len(result.get("openRecords", [])) == RUNS, "actual_two_uniform_closes_and_final_close": len(cycles) == RUNS and len(uniform) == CLOSE_CYCLES and len(final) == 1, "close_releases_document": bool(cycles) and all(v.get("stateEmpty") is True and v.get("cacheBytesAfterClose") == 0 for v in cycles), "hundred_page_scroll_operations": len(actions) == OPERATIONS and all(v.get("series") == "page-scroll" for v in actions), "page_navigation_moved": len(page_actions) == 80 and noops == 0, "operation_sequence": [v.get("index") for v in actions] == list(range(OPERATIONS)) and all(v.get("command") == ("zoom.in" if i % 10 == 8 else "zoom.fit_width" if i % 10 == 9 else "nav.page") for i, v in enumerate(actions)), "finite_timings": finite(opens) and all(finite([v.get("firstResponseFrameMs"), v.get("finalQualityFrameMs")]) and v["finalQualityFrameMs"] >= v["firstResponseFrameMs"] for v in actions), "full_resolution_conditions": type(result.get("maximumMemoryPressureLevel")) is int and result["maximumMemoryPressureLevel"] == 0, "source_hash": result.get("documentSha256") == source_hash, "continuous_scroll_disabled": result.get("continuousScroll", {}).get("requestedInputs") == 0 and result.get("continuousScroll", {}).get("dispatchedInputs") == 0, "tile_cache_within_configured_budget": type(result.get("tileCachePeakChargedBytes")) is int and type(result.get("tileCacheBudgetBytes")) is int and 0 <= result["tileCachePeakChargedBytes"] <= result["tileCacheBudgetBytes"], } summary = {"open": stats(opens) if finite(opens) else {"n": 0}, "excludedNoopPageNavigation": noops} if checks["finite_timings"]: summary.update({"firstResponse": stats([v["firstResponseFrameMs"] for v in actions]), "finalQuality": stats([v["finalQualityFrameMs"] for v in actions])}) for cached in (True, False): summary["cachedNavigation" if cached else "uncachedNavigation"] = stats( [v["finalQualityFrameMs"] for v in page_actions if v.get("cacheHit") is cached and v.get("positionChanged") is True]) return checks, summary def rss(pid): try: match = re.search(r"VmRSS:\s+(\d+)", Path(f"/proc/{pid}/status").read_text()) return int(match[1]) * 1024 if match else 0 except (OSError, ProcessLookupError): return 0 def group_rss(pid, seen=None): seen = set() if seen is None else seen if pid in seen: return 0 seen.add(pid) value = rss(pid) try: children = set() for task in Path(f"/proc/{pid}/task").iterdir(): try: children.update((task / "children").read_text().split()) except OSError: pass except OSError: children = [] return value + sum(group_rss(int(child), seen) for child in children) def run_observed(command, env, output, name, timeout=180): started = time.monotonic() app_peak = group_peak = samples = 0 with (output / f"{name}.log").open("wb") as log: process = subprocess.Popen(command, env=env, stdout=log, stderr=subprocess.STDOUT, start_new_session=True) try: while process.poll() is None: app_peak = max(app_peak, rss(process.pid)) group_peak = max(group_peak, group_rss(process.pid)) samples += 1 if time.monotonic() - started > timeout: raise TimeoutError(f"{name} exceeded {timeout} seconds") time.sleep(.05) finally: if process.poll() is None: os.killpg(process.pid, signal.SIGTERM) try: process.wait(timeout=3) except subprocess.TimeoutExpired: os.killpg(process.pid, signal.SIGKILL) process.wait(timeout=3) return {"command": command, "exitCode": process.returncode, "elapsedSeconds": time.monotonic() - started, "sampleIntervalSeconds": .05, "samples": samples, "sampledApplicationRssPeakBytes": app_peak, "sampledProcessGroupRssPeakBytes": group_peak, "rssMethod": "Sum of /proc VmRSS for app and child tree; includes shared pages per process, excludes Xvfb"} def stats(values): if not values: return {"n": 0} ordered = sorted(values) return {"n": len(values), "p50Ms": statistics.median(values), "p95Ms": ordered[max(0, math.ceil(.95 * len(values)) - 1)], "maxMs": ordered[-1]} def main(argv=None): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--binary", type=Path, default=ROOT / "build/docview") parser.add_argument("--output", type=Path, default=ROOT / "tests/results/stress-final") parser.add_argument("--work-directory", type=Path, default=ROOT / "build/stress") parser.add_argument("--xvfb", action="store_true") args = parser.parse_args(argv) output = args.output.resolve() output.mkdir(parents=True, exist_ok=True) clear_previous_evidence(output) binary = args.binary.resolve() report = {"schemaVersion": 2, "success": False, "runsRequested": RUNS, "operationsRequested": OPERATIONS, "closeCyclesRequested": CLOSE_CYCLES, "continuousScrollInputsRequested": 0, "sourceRemoved": False, "method": "OS file cache retained after generation; synthetic raster content"} source = None display = None started = time.monotonic() try: args.work_directory.mkdir(parents=True, exist_ok=True) report["executableSha256"] = executable_hashes(binary) report["binarySha256"] = report["executableSha256"]["docview"] report["runnerSha256"] = file_hash(__file__) report["generatorSha256"] = file_hash(Path(__file__).with_name("generate_stress_pdf.py")) if os.environ.get("QTWEBENGINE_DISABLE_SANDBOX", "0") != "0" or any( flag in os.environ.get("QTWEBENGINE_CHROMIUM_FLAGS", "") for flag in ("--no-sandbox", "--disable-seccomp-filter-sandbox", "--disable-setuid-sandbox")): raise RuntimeError("Sandbox-disabling environment is not permitted") with tempfile.TemporaryDirectory(prefix="scan-workload-", dir=args.work_directory) as workload: source = Path(workload) / "scan-5000.pdf" print("Generating 5000 pages / 1280 uncompressed RGB images after free-disk check", flush=True) manifest = generate(source) report["source"] = {key: value for key, value in manifest.items() if key != "images"} report["sourceValidation"] = validate_source(source, manifest) (output / "source-manifest.json").write_text(json.dumps(manifest, indent=2) + "\n") print(f"Generated {manifest['sizeBytes']} bytes; SHA-256 {manifest['sha256']}", flush=True) info = subprocess.run(["/usr/bin/pdfinfo", str(source)], check=True, capture_output=True, text=True, timeout=30) (output / "pdfinfo.txt").write_text(info.stdout) if not re.search(r"Pages:\s+5000\b", info.stdout): raise AssertionError("Independent PDF page count is not 5000") report["independentMarkerChecks"] = [] for physical_page in (1, 1280, 5000): text = subprocess.run(["/usr/bin/pdftotext", "-f", str(physical_page), "-l", str(physical_page), str(source), "-"], check=True, capture_output=True, text=True, timeout=30).stdout expected = f"DOCVIEW STRESS PAGE {physical_page:05d} OF 05000 IMAGE {(physical_page - 1) % 1280:04d}" if expected not in text: raise AssertionError(f"Source marker absent on page {physical_page}") report["independentMarkerChecks"].append({"physicalPage": physical_page, "marker": expected, "found": True}) with tempfile.TemporaryDirectory(prefix="docview-stress-state-") as temporary: directory = Path(temporary) config = directory / "config.toml" config.write_text("schema_version=1\n[privacy]\nremember_position=false\nrecent_documents=0\n") env = os.environ.copy() for name, leaf in (("XDG_CONFIG_HOME", "config"), ("XDG_STATE_HOME", "state"), ("XDG_CACHE_HOME", "cache")): env[name] = str(directory / leaf) env["QT_QUICK_BACKEND"] = "software" env["QT_QPA_PLATFORM"] = "xcb" env["LIBGL_ALWAYS_SOFTWARE"] = "1" if args.xvfb: display = subprocess.Popen(["/usr/bin/Xvfb", "-displayfd", "1", "-screen", "0", "1400x900x24", "-nolisten", "tcp"], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True) if not select.select([display.stdout], [], [], 5)[0]: raise RuntimeError("Xvfb did not allocate a display") number = display.stdout.readline().strip() if not number.isdigit(): raise RuntimeError("Xvfb returned an invalid display") env["DISPLAY"] = ":" + number benchmark = output / "benchmark.json" print("Running actual frame benchmark: 3 opens and 100 keyboard operations", flush=True) report["benchmarkProcess"] = run_observed( [str(binary), "--config", str(config), "--benchmark-output", str(benchmark), "--benchmark-runs", str(RUNS), "--benchmark-operations", str(OPERATIONS), "--benchmark-close-cycles", str(CLOSE_CYCLES), "--benchmark-scroll-steps", "0", str(source)], env, output, "benchmark") if report["benchmarkProcess"]["exitCode"] != 0: raise RuntimeError("DocView stress benchmark failed") result = json.loads(benchmark.read_text()) report["benchmarkEvidence"] = result report["measurementChecks"], report["latency"] = summarize_benchmark(result, manifest["sha256"]) if not all(report["measurementChecks"].values()): raise AssertionError("Benchmark validation failed: " + ", ".join(k for k, v in report["measurementChecks"].items() if not v)) smoke = output / "last-page" print("Checking direct physical-page 5000 open and capturing the actual window", flush=True) report["lastPageProcess"] = run_observed( [str(binary), "--config", str(config), "--page", "5000", "--smoke-output", str(smoke), str(source)], env, output, "last-page", timeout=30) state = (smoke / "state.txt").read_text().splitlines() if report["lastPageProcess"]["exitCode"] != 0 or state[:2] != ["Ready", "pdf"] or not (smoke / "screen.png").is_file(): raise AssertionError("Last-page smoke did not reach Ready PDF and save its image") report["lastPageSmoke"] = {"state": state, "screenshot": str(smoke / "screen.png"), "runtime": json.loads((smoke / "runtime.json").read_text()), "requestedPhysicalPage": 5000, "visualPageMarkerReview": "pending"} report["originalSha256After"] = file_hash(source) report["originalUnchanged"] = report["originalSha256After"] == manifest["sha256"] == result["documentSha256"] report["measurementChecks"]["source_unchanged_after_smoke"] = report["originalUnchanged"] if not report["originalUnchanged"]: raise AssertionError("Source changed during benchmark or smoke") report["success"] = True except Exception as error: report["error"] = str(error) finally: if display is not None: display.terminate() try: display.wait(timeout=3) except subprocess.TimeoutExpired: display.kill() display.wait(timeout=3) report["sourceRemoved"] = source is not None and not source.exists() if "executableSha256" in report: try: report["executableSha256After"] = executable_hashes(binary) report["executablesUnchanged"] = report["executableSha256"] == report["executableSha256After"] except OSError: report["executablesUnchanged"] = False report.setdefault("measurementChecks", {})["executables_unchanged"] = report["executablesUnchanged"] report["success"] = report["success"] and report["executablesUnchanged"] report["success"] = report["success"] and report["sourceRemoved"] report["elapsedSeconds"] = time.monotonic() - started (output / "report.json").write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n") print(json.dumps({key: value for key, value in report.items() if key not in ("benchmarkEvidence", "source")}, ensure_ascii=False, indent=2), flush=True) return 0 if report["success"] and report["sourceRemoved"] else 1 if __name__ == "__main__": raise SystemExit(main())