#!/usr/bin/env python3 import collections import glob import os import re import subprocess import sys import threading import time from datetime import datetime from pathlib import Path def now_iso(): return datetime.now().astimezone().isoformat(timespec="seconds") def read_lines(path): if not os.path.exists(path): return [] with open(path) as f: return [l.rstrip("\n") for l in f if l.strip()] def write_lines(path, lines): with open(path, "w") as f: for l in lines: f.write(l + "\n") def begin_test_case_names(qpa_path): names = set() if not os.path.exists(qpa_path): return names txt = open(qpa_path, errors="replace").read() for line in txt.splitlines(): if line.startswith("#beginTestCaseResult "): names.add(line[len("#beginTestCaseResult "):].strip()) return names def last_begin_test_case_name(qpa_path): last = None if not os.path.exists(qpa_path): return last for line in open(qpa_path, errors="replace"): if line.startswith("#beginTestCaseResult "): last = line[len("#beginTestCaseResult "):].strip() return last def nproc(): return os.cpu_count() or 1 def main(): lab_root = Path(os.environ.get("LAB_ROOT", Path(__file__).resolve().parent.parent)) drivers_root = Path(os.environ.get("DRIVERS_ROOT", lab_root.parent)) mesa_src = Path(os.environ.get("MESA_SRC", drivers_root / "intel-hard-graphics")) mesa_build = Path(os.environ.get("MESA_BUILD", mesa_src / "build")) cts_dir = Path(os.environ.get("CTS_DIR", mesa_src / "VK-GL-CTS/build-cts/external/vulkancts/modules/vulkan")) cts_mustpass = Path(os.environ.get("CTS_MUSTPASS", mesa_src / "VK-GL-CTS/external/vulkancts/mustpass/main/vk-default")) icd = Path(os.environ.get("VK_DRIVER_FILES", mesa_build / "src/intel/vulkan_hasvk/intel_hasvk13_devenv_icd.x86_64.json")) exclude = lab_root / "cts/cts-exclude-gen7.txt" scope = sys.argv[1] if len(sys.argv) > 1 else "mustpass" shards = int(sys.argv[2]) if len(sys.argv) > 2 else nproc() cooldown = int(os.environ.get("COOLDOWN", "60")) outdir = Path(os.environ.get( "OUTDIR", Path.cwd() / f"cts-run-{datetime.now().strftime('%Y%m%d-%H%M%S')}")) deqp = cts_dir / "deqp-vk" for p in (deqp, icd): if not p.exists(): print(f"missing: {p}", file=sys.stderr) sys.exit(1) outdir.mkdir(parents=True, exist_ok=True) os.chdir(cts_dir) cases_txt = outdir / "cases.txt" if scope == "all": out = subprocess.run([str(deqp), "--deqp-runmode=stdout-caselist"], capture_output=True, text=True).stdout lines = [l[len("TEST: "):] for l in out.splitlines() if l.startswith("TEST: ")] src_txt = outdir / "all.txt" write_lines(src_txt, lines) else: lines = [] for f in sorted(glob.glob(str(cts_mustpass / "*.txt"))): lines.extend(l for l in read_lines(f) if l.startswith("dEQP-VK")) src_txt = outdir / "mustpass.txt" write_lines(src_txt, lines) src_lines = read_lines(src_txt) if exclude.exists() and exclude.stat().st_size > 0: excl = set(read_lines(exclude)) case_lines = [l for l in src_lines if l not in excl] else: case_lines = list(src_lines) write_lines(cases_txt, case_lines) print(f"{len(src_lines)} cases, {len(src_lines) - len(case_lines)} excluded, {len(case_lines)} to run") prev = sorted(glob.glob(str(outdir / "*.qpa"))) + sorted(glob.glob(str(outdir / "attempt-*/*.qpa"))) if prev: done = set() for f in prev: done |= begin_test_case_names(f) done_txt = outdir / "done.txt" write_lines(done_txt, sorted(done)) if done: case_lines = [l for l in read_lines(cases_txt) if l not in done] write_lines(cases_txt, case_lines) print(f"resume: {len(done)} already recorded, {len(case_lines)} left") n = 1 while (outdir / f"attempt-{n}").exists(): n += 1 attempt_dir = outdir / f"attempt-{n}" attempt_dir.mkdir(parents=True, exist_ok=True) for f in list(glob.glob(str(outdir / "shard*.qpa"))) + list(glob.glob(str(outdir / "shard*.log"))): os.rename(f, attempt_dir / os.path.basename(f)) print(f"previous attempt parked in attempt-{n}/") case_lines = read_lines(cases_txt) if not case_lines: print("nothing left to run") sys.exit(0) def run_shard(i): list_path = outdir / f"shard{i}.txt" shard_lines = [l for j, l in enumerate(case_lines) if (j + 1) % shards == i] write_lines(list_path, shard_lines) pass_n = 1 while list_path.exists() and list_path.stat().st_size > 0: qpa = outdir / f"shard{i}.pass{pass_n}.qpa" log = outdir / f"shard{i}.pass{pass_n}.log" env = dict(os.environ, VK_DRIVER_FILES=str(icd)) with open(log, "w") as lf: subprocess.run( [str(deqp), f"--deqp-caselist-file={list_path}", f"--deqp-log-filename={qpa}", "--deqp-log-images=disable", "--deqp-log-shader-sources=disable"], stdout=lf, stderr=subprocess.STDOUT, env=env) log_text = open(log, errors="replace").read() if log.exists() else "" if "ABORTED" not in log_text: break last = last_begin_test_case_name(qpa) if not last: break print(f"shard {i} aborted at {last} -- cooling down {cooldown}s", file=sys.stderr) with open(outdir / "aborted.txt", "a") as af: af.write(last + "\n") time.sleep(cooldown) shard_lines = read_lines(list_path) seen = False remaining = [] for l in shard_lines: if seen: remaining.append(l) if l == last: seen = True write_lines(list_path, remaining) pass_n += 1 threads = [threading.Thread(target=run_shard, args=(i,)) for i in range(shards)] for t in threads: t.start() for t in threads: t.join() res = {} for f in sorted(glob.glob(str(outdir / "shard*.qpa"))) + sorted(glob.glob(str(outdir / "attempt-*/shard*.qpa"))): txt = open(f, errors="replace").read() for block in re.split(r"#beginTestCaseResult ", txt)[1:]: name = block.split("\n", 1)[0].strip() m = re.search(r'', block) if m: res[name] = m.group(1) counts = collections.Counter(res.values()) print(f"\n{sum(counts.values())} results") for k, v in counts.most_common(): print(f" {k:16} {v}") with open(outdir / "failures.txt", "w") as fh: for name, status in sorted(res.items()): if status not in ("Pass", "NotSupported"): fh.write(f"{status}\t{name}\n") print(f"\nfailures -> {outdir}/failures.txt") if __name__ == "__main__": main()