diff --git a/tools/headless_batch/README.md b/tools/headless_batch/README.md new file mode 100644 index 00000000000..f2874742ab0 --- /dev/null +++ b/tools/headless_batch/README.md @@ -0,0 +1,185 @@ +# Headless batch runner + +Run many FreeSpace Open missions as fast as the CPU allows, with no window, no rendering and no audio, then +collect one result per run. Each run is a separate engine process with its own workspace, seed and UDP port, +so runs share no in-process state and the batch scales with the number of cores. + +This directory has the runner (`run_batch.py`), its tests (`test_run_batch.py`) and a starting plan +(`example.plan.json`). The engine side of the feature is the set of `-headless` command-line options described +below. + +## The engine options + +| Option | Effect | +|---|---| +| `-headless` | No window, renderer, audio, speech, joystick, HUD/radar presentation or movies. Graphics runs on the stub backend. Frames run as fast as possible, and mission time comes from a virtual clock that advances only by simulated time, which removes the main dependence of results on machine speed (see Reproducibility for what remains). Wall-clock timers (profiling, UI deadlines) keep real time. Implies `-nomovies` and `-noninteractive` (warnings are only logged; an error or a Lua error ends the process instead of opening a dialog). | +| `-fixed-dt ` | Length of one simulation step. Default 1/30; must be in [1/65536, 0.25]. It sets the step, it does not enlarge the engine's integration step. | +| `-max-sim-time ` | End the run once this much mission time has been simulated. Must be above 0 and at most 32767 (the range of the engine's fixed-point mission clock). | +| `-sim-result ` | When the engine shuts down, write a JSON file describing the run (below). | +| `-seed ` | Existing option. With this change it seeds the generators the simulation uses: the main generator, the generator that `RandomRange` instances use, and the ship firepoint, asteroid-split and random-message-sender shuffles. | + +`-sim-result` file: + +```json +{ "schema": 1, "mission": "my-mission", "seed": 1000, "fixed_dt": 0.0333333, + "max_sim_time": 120.0, "simulated_seconds": 120.03, "wall_seconds": 9.4, + "time_limit_reached": true, "termination": "time_limit", + "goals": [ { "name": "Destroy the convoy", "type": "primary", "status": "complete" } ] } +``` + +`termination` is `time_limit` when `-max-sim-time` ended the run, `start_failed` when the default loadout could not +be committed so the mission never started, `player_died` when the player's ship blew up, and `mission_exit` when the +mission or the game ended it first. `goals` lists the mission's goals (invalidated ones left out) with `type` +`primary`, `secondary` or `bonus` and `status` `complete`, `failed` or `incomplete`, as they stood when the run +ended. + +## What runs and what does not + +The authoritative frame is unchanged: `game_do_frame()` calls `game_frame()` and `game_simulation_frame()`, so +object movement, ship and weapon processing, collisions, mission parsing and goals, and scripting all run. + +Script frame callbacks (`On Frame`) normally fire from `gr_flip()`, which headless mode never reaches. The +engine therefore runs them once per simulated frame through `gr_process_frame_callbacks()`, the same function the +rendered path uses. A mod that runs its game logic from `On Frame` behaves as usual. + +Keyboard, mouse and joystick input is not read, but the player's flight controls still run, so a script can fly +the player ship (scripted flight control) and the player warp-out sequence works. + +Screens that wait for a person are passed automatically. `-start_mission` normally stops at the fiction, command +briefing, red alert and briefing screens and at the debriefing. In a headless single-player run the engine skips +the fiction and command briefing. At the briefing it takes the default loadout through the same `commit_pressed()` +call it uses for missions flagged no-briefing; at the red alert screen it continues into the mission as that +screen's button does. When the mission ends it leaves the debriefing by quitting, so the result file is written. +If the default loadout cannot be committed (for example a required weapon is not on any ship) the run ends with a +warning in the engine log and `start_failed` in the result. The death popup cannot be answered either, so when the +player's ship blows up the run ends with `player_died`. Any other modal popup is logged and answered as if +cancelled, unless a script handles `On Dialog Frame`. + +Skipped: renderer setup and frame rendering, sound and speech, input devices, HUD and radar, camera and photo +mode interpolation, the intro movie, and real-time frame pacing. Particle, trail, shockwave, asteroid and nebula +updates still run, because nothing has shown them to be irrelevant to gameplay. Model and collision data still +load. + +## Quick start + +1. Make a **base workspace**: a portable FreeSpace Open folder holding the root `.vp` files, any mod folders the + mission needs, and a stored pilot (`data/players/.json`, or `.plr`, in the root `data` folder or in a mod). + The runner never writes to it. +2. Copy `example.plan.json`, set `engine`, `base_workspace`, `output_dir`, `mission` and `pilot`. +3. Check the plan, then run it: + + ``` + python tools/headless_batch/run_batch.py my.plan.json --dry-run # validate and print the commands + python tools/headless_batch/run_batch.py my.plan.json + ``` + +Before a long batch, run one short simulation (`"repeats": 1, "max_sim_time": 20`) and confirm the result file +reports `simulated_seconds` close to 20. A run that never writes a result ends as a timeout and is reported as a +failure with the reason. + +## Plan reference + +Unknown keys are rejected, so a misspelt option fails instead of being ignored. + +| Key | Required | Default | Meaning | +|---|---|---|---| +| `engine` | yes | | Path to the engine executable. | +| `base_workspace` | yes | | Portable workspace described above. | +| `output_dir` | yes | | Where runs and results go. Keep the path short on Windows, see Limits. | +| `mission` | yes | | Passed to `-start_mission`. A simple name (letters, digits, `_ . -`). | +| `pilot` | yes | | Stored pilot, passed to `-pilot`. | +| `max_sim_time` | yes | | Simulated seconds to run, passed to `-max-sim-time`. | +| `mods` | no | `[]` | Mod folder names under the base workspace, passed to `-mod` in this order. | +| `seeds` | no | | Explicit list of seeds, integers in 1..2147483647. One run per seed. | +| `seed_start`, `repeats` | no | 1000, 1 | Used when `seeds` is absent: seeds `seed_start` .. `seed_start + repeats - 1`. | +| `workers` | no | 1 | Engine processes running at once. | +| `fixed_dt` | no | 1/30 | Passed to `-fixed-dt`. | +| `timeout_seconds` | no | 600 | Wall-clock limit per run; a run over it is killed and counted as failed. | +| `base_port` | no | 39000 | Run *i* binds UDP port `base_port + i`. Every engine process binds one at startup, so ports must be unique. | +| `extra_args` | no | `[]` | Extra engine arguments appended to the command line. | + +## What a run does + +For run *i* with seed *s* the runner: + +1. creates `output_dir/runs/r000i/w`, hardlinks the root `.vp` files and every `.vp` inside the mod folders, + and copies the rest, so a run does not write through to the base workspace (the engine only reads `.vp` files; + a run that rewrote one in place would change the base too); +2. starts `engine -portable_mode -parse_cmdline_only -port -start_mission -pilot + -headless -fixed-dt
-max-sim-time -sim-result -noninteractive -nomovies -seed `, then + `-mod` if any, then `extra_args`, with the workspace as the working directory, stdout and stderr going to the + run folder, and `FSO_KEEP_STDOUT=1` in the environment (without it the Windows engine reattaches its output to + the parent's console, if there is one, and the files stay empty); +3. waits up to `timeout_seconds`, killing the whole process tree if it is exceeded; +4. reads the result file and records the run. + +A run is `ok` only if the engine exited with code 0 and wrote a readable result file whose `termination` is +`time_limit`, `player_died` or `mission_exit` (a player death is an outcome of the mission, not a failure of the +run). A nonzero exit, a missing or unreadable result, a mission that never started, or a +timeout is a failure and is recorded with the reason; a run whose result could not be read is not +reported as a success. With `-noninteractive` the engine prints an error, or a Lua error and its traceback, to +stderr before it exits, so a failed run's problems quote that message (`engine said: ...`) and `stderr.txt` has it +in full. Release builds write no `fs2_open.log`, so stderr is often the only record of why a run stopped. On Windows the runner turns off crash and missing-file dialogs for its children so a +crashing engine exits instead of blocking an unattended batch (this changes no registry or machine setting). + +## Output + +``` +output_dir/ + results.jsonl one JSON record per finished run (index, seed, port, status, exit code, wall seconds, + the engine's result, and a list of problems) + summary.json runs, ok, failed, timed out, exit codes, terminations, simulated seconds, batch wall + seconds and simulated seconds per wall second + runs/r0000/ result.json (same record), stdout.txt, stderr.txt, engine-result.json, w/ (workspace) +``` + +`--resume` keeps finished (`ok`) runs whose seed still matches the plan and reruns the others (failed, +interrupted, or run with another seed) from a clean workspace; without it a non-empty +`runs/` folder is refused so earlier results are never overwritten. The exit code is 0 when every run is `ok`, +1 when any failed, and 2 for a plan or usage error. + +## Running the engine directly + +``` +fs2_open_x64.exe -portable_mode -parse_cmdline_only -mod mymod -start_mission my-mission -pilot Tester + -headless -fixed-dt 0.0333333333 -max-sim-time 120 -sim-result result.json -noninteractive -seed 1000 +``` + +On Windows, set `FSO_KEEP_STDOUT=1` if you redirect the engine's stdout or stderr to a file; otherwise, started +from a console, the engine writes them to that console instead. + +## Reproducibility + +`-seed` removes the known engine-side sources of run-to-run variation, but it does not make runs exact replays. +A script that draws its own random numbers (for example Lua `math.random()`, which uses a separate generator) +must seed itself from the run seed. In our tests two short same-seed runs of one scenario were identical, while +two same-seed runs of a long scenario with many AI-driven ships diverged (24 versus 25 losses), and the cause was +not found (a real-clock afterburner cooldown, since moved to the mission clock, is a possible but unconfirmed +cause). Treat same-seed runs as correlated, not identical, and use several seeds for any comparison. The +generators behind `RandomRange` are seeded once, when the tables load, so a mission restarted inside one process +does not repeat; that is why the runner starts one process per run. The +engine has no fixed-step mode outside this feature, so rendered and headless runs of one mission are not +expected to match frame for frame either. + +## Limits and known issues + +- **Windows paths.** The engine cannot open files whose full path is 260 characters or longer. The runner checks + this before it starts: it computes the longest path it will stage under `output_dir` and refuses the plan, + naming the longest `output_dir` that fits. +- **Hardlinks.** `.vp` files are hardlinked, which needs `base_workspace` and `output_dir` on the same volume. + The runner checks this before it starts and refuses the plan otherwise. +- **Process-level parallelism only.** The engine has a lot of global state, so each simulation gets its own + process. More workers than physical cores did not help in our measurements. +- **Not a rendering test.** Anything that depends on the renderer, audio or input does not run. +- **Single player, through `-start_mission`.** The screens are only passed automatically on the way into and out of + a single-player mission. Without `-start_mission` the engine waits at the main hall, unless a script drives it. +- **Negative option values.** The engine's command-line parser reads a value that starts with `-` (for example + `-seed -1`) as a new option and shows an "Unrecognized command line parameter" message box, which blocks a + headless run. The runner cannot produce this (seeds must be above zero and `fixed_dt` is range-checked); keep to + positive values when you start the engine by hand. + +## Tests + +`python -m unittest tools/headless_batch/test_run_batch.py`. They cover plan validation, command construction, +workspace staging, the same-volume and path-length checks, and a stub engine for success, crash, missing result and +timeout. They need no game files. diff --git a/tools/headless_batch/example.plan.json b/tools/headless_batch/example.plan.json new file mode 100644 index 00000000000..103e30b1ee0 --- /dev/null +++ b/tools/headless_batch/example.plan.json @@ -0,0 +1,16 @@ +{ + "engine": "C:/path/to/fs2_open_x64.exe", + "base_workspace": "C:/path/to/portable-workspace", + "output_dir": "C:/path/to/batch-output", + "mods": ["mymod"], + "mission": "my-mission", + "pilot": "Tester", + "max_sim_time": 120, + "seed_start": 1000, + "repeats": 4, + "workers": 2, + "fixed_dt": 0.03333333333333333, + "timeout_seconds": 600, + "base_port": 39000, + "extra_args": [] +} diff --git a/tools/headless_batch/run_batch.py b/tools/headless_batch/run_batch.py new file mode 100644 index 00000000000..a6a1d6b7cd2 --- /dev/null +++ b/tools/headless_batch/run_batch.py @@ -0,0 +1,402 @@ +#!/usr/bin/env python3 +"""Run many headless FreeSpace Open simulations in parallel OS processes. + +Each run is one engine process in its own portable workspace, with its own seed +and UDP port, started with -headless, a fixed step and a simulated-time limit. +The engine's -sim-result sidecar is collected into results.jsonl and summary.json. +See README.md for the plan format and the engine options this relies on. +""" + +from __future__ import annotations + +import argparse +import concurrent.futures +import ctypes +import json +import os +import re +import shutil +import subprocess +import sys +import time +from pathlib import Path +from typing import Any + +NAME = re.compile(r"[A-Za-z0-9_.-]+") +RESERVED_NAMES = {".", ".."} +MAX_SEED = 2**31 - 1 +MAX_SIM_TIME = 32767.0 # the engine's mission clock is 16.16 fixed point +FIXED_DT_MIN = 1.0 / 65536 # a shorter step rounds to zero in the engine +FINISHED = {"time_limit", "player_died", "mission_exit"} # engine terminations that mean the mission ran + +# Plan keys, with None meaning "required". Unknown keys are rejected so typos fail loudly. +PLAN_DEFAULTS: dict[str, Any] = { + "engine": None, + "base_workspace": None, + "output_dir": None, + "mission": None, + "pilot": None, + "max_sim_time": None, + "mods": [], + "seeds": None, + "seed_start": 1000, + "repeats": 1, + "workers": 1, + "fixed_dt": 1.0 / 30.0, + "timeout_seconds": 600.0, + "base_port": 39000, + "extra_args": [], +} + + +class PlanError(ValueError): + """The plan file or the workspace it names is not usable.""" + + +def _name(value: Any, what: str) -> str: + if not isinstance(value, str) or not NAME.fullmatch(value) or value in RESERVED_NAMES: + raise PlanError(f"{what} must be a simple name (letters, digits, _ . -), got {value!r}") + return value + + +def _path(value: Any, what: str) -> Path: + if not isinstance(value, str) or not value or "\0" in value: + raise PlanError(f"{what} must be a path string, got {value!r}") + return Path(value).resolve() + + +def _number(plan: dict[str, Any], key: str, low: float, high: float, integer: bool = False, + low_inclusive: bool = True) -> float | int: + """A finite number in the given range (a duration or step must be strictly above zero: low_inclusive=False).""" + value = plan[key] + if isinstance(value, bool) or not isinstance(value, (int, float)) or value != value or abs(value) == float("inf"): + raise PlanError(f"{key} must be a finite number, got {value!r}") + if integer and int(value) != value: + raise PlanError(f"{key} must be a whole number, got {value!r}") + above_low = value >= low if low_inclusive else value > low + if not (above_low and value <= high): + raise PlanError(f"{key} must be {'at least' if low_inclusive else 'above'} {low} and at most {high}, got {value!r}") + return int(value) if integer else float(value) + + +def load_plan(path: Path) -> dict[str, Any]: + """Read, validate and normalise a plan. Raises PlanError with a message that names the problem.""" + try: + raw = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise PlanError(f"cannot read plan {path}: {exc}") from exc + if not isinstance(raw, dict): + raise PlanError("the plan must be a JSON object") + unknown = sorted(set(raw) - set(PLAN_DEFAULTS)) + if unknown: + raise PlanError(f"unknown plan keys: {', '.join(unknown)}") + plan = {**PLAN_DEFAULTS, **raw} + missing = [key for key in ("engine", "base_workspace", "output_dir", "mission", "pilot", "max_sim_time") + if plan[key] is None] + if missing: + raise PlanError(f"missing required plan keys: {', '.join(missing)}") + + plan["mission"] = _name(plan["mission"], "mission") + plan["pilot"] = _name(plan["pilot"], "pilot") + if not isinstance(plan["mods"], list): + raise PlanError("mods must be a list of directory names") + plan["mods"] = [_name(mod, "mod") for mod in plan["mods"]] + if not isinstance(plan["extra_args"], list) or not all(isinstance(a, str) for a in plan["extra_args"]): + raise PlanError("extra_args must be a list of strings") + + plan["fixed_dt"] = _number(plan, "fixed_dt", FIXED_DT_MIN, 0.25) + plan["max_sim_time"] = _number(plan, "max_sim_time", 0.0, MAX_SIM_TIME, low_inclusive=False) + plan["timeout_seconds"] = _number(plan, "timeout_seconds", 0.0, 1e7, low_inclusive=False) + plan["workers"] = _number(plan, "workers", 1, 256, integer=True) + plan["base_port"] = _number(plan, "base_port", 1024, 65000, integer=True) + + if plan["seeds"] is None: + start = int(_number(plan, "seed_start", 1, MAX_SEED, integer=True)) + repeats = int(_number(plan, "repeats", 1, 100000, integer=True)) + plan["seeds"] = [start + i for i in range(repeats)] + else: + seeds = plan["seeds"] + if (not isinstance(seeds, list) or not seeds + or not all(isinstance(s, int) and not isinstance(s, bool) and 1 <= s <= MAX_SEED for s in seeds)): + raise PlanError(f"seeds must be a non-empty list of integers in 1..{MAX_SEED}") + if plan["base_port"] + len(plan["seeds"]) > 65535: + raise PlanError("base_port plus the number of runs exceeds the port range") + + # Absolute, because the engine runs in its run folder: a relative -sim-result path would land there. + plan["engine"] = _path(plan["engine"], "engine") + plan["base_workspace"] = _path(plan["base_workspace"], "base_workspace") + plan["output_dir"] = _path(plan["output_dir"], "output_dir") + if not plan["engine"].is_file(): + raise PlanError(f"engine not found: {plan['engine']}") + check_workspace(plan["base_workspace"], plan["mods"], plan["pilot"]) + check_same_volume(plan["base_workspace"], plan["output_dir"]) + check_path_length(plan["base_workspace"], plan["mods"], plan["output_dir"]) + return plan + + +# The engine cannot open a file whose full path reaches this many characters (Windows MAX_PATH, terminator included). +WINDOWS_PATH_LIMIT = 260 +RUN_FOLDER_SUFFIX = os.sep + os.path.join("runs", "r0000", "w") # the longest run folder name the runner makes + + +def _volume_of(path: Path) -> int: + """Device id of the nearest existing ancestor (the folder may not exist yet).""" + probe = path.resolve() + while not probe.exists(): + probe = probe.parent + return probe.stat().st_dev + + +def check_same_volume(base: Path, output_dir: Path) -> None: + """.vp packs are hardlinked into each run, which only works inside one volume. Fail now, not on run 1.""" + if _volume_of(base) != _volume_of(output_dir): + raise PlanError(f"base_workspace and output_dir must be on the same volume so the .vp packs can be hardlinked " + f"({base} and {output_dir} are not)") + + +def longest_relative_path(base: Path, mods: list[str]) -> int: + """Length of the longest path, relative to a workspace root, among the files a run will have.""" + longest = 0 + for top in [base / "data", *(base / mod for mod in mods)]: + for folder, _dirs, files in os.walk(top): + relative = os.path.relpath(folder, base) + for name in files: + longest = max(longest, len(os.path.join(relative, name))) + return longest + + +def check_path_length(base: Path, mods: list[str], output_dir: Path, limit: int = WINDOWS_PATH_LIMIT) -> None: + """Windows only: refuse an output_dir so deep that the engine could not open some staged file.""" + if os.name != "nt": + return + worst = len(str(output_dir.resolve())) + len(RUN_FOLDER_SUFFIX) + 1 + longest_relative_path(base, mods) + if worst >= limit: + allowed = len(str(output_dir.resolve())) - (worst - limit + 1) + raise PlanError(f"output_dir is too long: the longest staged file path would be {worst} characters but the engine " + f"cannot open paths of {limit} or more. Use an output_dir of at most {allowed} characters " + f"(it is {len(str(output_dir.resolve()))} now)") + + +def check_workspace(base: Path, mods: list[str], pilot: str) -> None: + if not base.is_dir(): + raise PlanError(f"base_workspace not found: {base}") + if not list(base.glob("*.vp")): + raise PlanError(f"base_workspace has no root .vp files: {base}") + for mod in mods: + if not (base / mod).is_dir(): + raise PlanError(f"mod directory not found under base_workspace: {base / mod}") + # A stored pilot is mandatory: without one the engine stops at the pilot chooser. + places = [base / "data" / "players"] + [base / mod / "data" / "players" for mod in mods] + if not any((place / (pilot + ext)).is_file() for place in places for ext in (".json", ".plr")): + raise PlanError(f"no stored pilot {pilot!r} (data/players/{pilot}.json or .plr) in the base workspace or its mods") + + +def link_or_copy(source: str, destination: str) -> None: + """Hardlink .vp packs (immutable and large); copy everything else so a run can never write through to the base.""" + if source.lower().endswith(".vp"): + try: + os.link(source, destination) + return + except OSError as exc: + raise RuntimeError( + f"cannot hardlink {source}; put base_workspace and output_dir on the same volume") from exc + shutil.copy2(source, destination) + + +def _ignore_logs(_directory: str, names: list[str]) -> set[str]: + return {name for name in names if name.lower().endswith(".log")} + + +def prepare_workspace(base: Path, destination: Path, mods: list[str]) -> None: + destination.mkdir(parents=True, exist_ok=False) + for item in base.iterdir(): + if item.is_file() and item.suffix.lower() == ".vp": + link_or_copy(str(item), str(destination / item.name)) + if (base / "data").is_dir(): + shutil.copytree(base / "data", destination / "data", ignore=_ignore_logs, copy_function=link_or_copy) + else: + (destination / "data").mkdir() + for mod in mods: + shutil.copytree(base / mod, destination / mod, ignore=_ignore_logs, copy_function=link_or_copy) + + +def build_command(plan: dict[str, Any], seed: int, port: int, result_path: Path, launcher: tuple[str, ...] = ()) -> list[str]: + """The engine command line for one run. `launcher` is test plumbing (an interpreter argument before the flags).""" + command = [str(plan["engine"]), *launcher, + "-portable_mode", "-parse_cmdline_only", "-port", str(port), + "-start_mission", plan["mission"], "-pilot", plan["pilot"], + "-headless", "-fixed-dt", repr(plan["fixed_dt"]), "-max-sim-time", repr(plan["max_sim_time"]), + "-sim-result", str(result_path), "-noninteractive", "-nomovies", "-seed", str(seed)] + if plan["mods"]: + command += ["-mod", ",".join(plan["mods"])] + return command + plan["extra_args"] + + +def suppress_crash_dialogs() -> None: + """Windows only: stop a crashing engine from blocking an unattended batch on a dialog. + + Process-wide for this runner and inherited by its children; it changes no registry or machine setting. + """ + if os.name == "nt": + # SEM_FAILCRITICALERRORS | SEM_NOGPFAULTERRORBOX | SEM_NOOPENFILEERRORBOX + ctypes.windll.kernel32.SetErrorMode(0x0001 | 0x0002 | 0x8000) + + +def kill_tree(process: subprocess.Popen[bytes]) -> None: + if os.name == "nt": + subprocess.run(["taskkill", "/PID", str(process.pid), "/T", "/F"], capture_output=True, check=False) + else: + try: + os.killpg(process.pid, 9) + except ProcessLookupError: + pass + process.wait() + + +def engine_reason(stderr_text: str) -> str | None: + """Why the engine stopped, from its stderr: with -noninteractive, Error() prints "Error: ", and a Lua + error prints its message followed by a dashed separator and the traceback. Warnings are skipped.""" + lines = [line.strip() for line in stderr_text.splitlines()] + errors = [line for line in lines if line.startswith("Error: ")] + if errors: + return errors[-1] + separator = next((i for i, line in enumerate(lines) if len(line) > 10 and set(line) == {"-"}), len(lines)) + candidates = [line for line in lines[:separator] if line and not line.startswith("WARNING:")] + return candidates[-1] if candidates else None + + +def run_one(plan: dict[str, Any], index: int, seed: int, launcher: tuple[str, ...] = ()) -> dict[str, Any]: + """Stage a workspace, run one engine process to completion or timeout, and return its result record.""" + run_dir = plan["output_dir"] / "runs" / f"r{index:04d}" + workspace = run_dir / "w" + result_path = run_dir / "engine-result.json" + port = plan["base_port"] + index + record: dict[str, Any] = {"index": index, "seed": seed, "port": port, "status": "failed", + "exit_code": None, "timed_out": False, "wall_seconds": None, + "sim_result": None, "problems": []} + try: + prepare_workspace(plan["base_workspace"], workspace, plan["mods"]) + command = build_command(plan, seed, port, result_path, launcher) + started = time.monotonic() + with open(run_dir / "stdout.txt", "wb") as out, open(run_dir / "stderr.txt", "wb") as err: + # On Windows the engine reattaches stdout/stderr to the parent's console unless FSO_KEEP_STDOUT is 1, + # and the files above would stay empty + process = subprocess.Popen(command, cwd=workspace, stdout=out, stderr=err, + env={**os.environ, "FSO_KEEP_STDOUT": "1"}, + start_new_session=(os.name != "nt")) + try: + process.wait(timeout=plan["timeout_seconds"]) + except subprocess.TimeoutExpired: + record["timed_out"] = True + kill_tree(process) + record["wall_seconds"] = round(time.monotonic() - started, 3) + record["exit_code"] = process.returncode + if result_path.is_file(): + try: + record["sim_result"] = json.loads(result_path.read_text(encoding="utf-8")) + except json.JSONDecodeError as exc: + record["problems"].append(f"unreadable engine result: {exc}") + else: + termination = record["sim_result"].get("termination") + if termination not in FINISHED: + record["problems"].append(f"mission did not run to an end (termination {termination!r})") + else: + record["problems"].append("engine wrote no -sim-result file") + if record["timed_out"]: + record["problems"].append(f"killed after {plan['timeout_seconds']} s") + if record["exit_code"] not in (0, None) and not record["timed_out"]: + record["problems"].append(f"engine exit code {record['exit_code']}") + reason = engine_reason((run_dir / "stderr.txt").read_text(encoding="utf-8", errors="replace")) + if record["problems"] and reason: + record["problems"].append(f"engine said: {reason[:300]}") + if not record["problems"]: + record["status"] = "ok" + except (OSError, RuntimeError) as exc: + record["problems"].append(str(exc)) + (run_dir / "result.json").write_text(json.dumps(record, indent=2, sort_keys=True), encoding="utf-8") + return record + + +def read_record(path: Path) -> dict[str, Any] | None: + """A run's result.json, or None if it is missing or unreadable (an interrupted write).""" + try: + return json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return None + + +def summarize(records: list[dict[str, Any]], batch_wall: float) -> dict[str, Any]: + ok = [r for r in records if r["status"] == "ok"] + sim = sum(r["sim_result"].get("simulated_seconds", 0.0) for r in ok) + exits: dict[str, int] = {} + ends: dict[str, int] = {} + for r in records: + exits[str(r["exit_code"])] = exits.get(str(r["exit_code"]), 0) + 1 + if r["sim_result"]: + term = str(r["sim_result"].get("termination")) + ends[term] = ends.get(term, 0) + 1 + return {"runs": len(records), "ok": len(ok), "failed": len(records) - len(ok), + "timed_out": sum(1 for r in records if r["timed_out"]), + "exit_codes": exits, "terminations": ends, + "simulated_seconds_total": round(sim, 3), "batch_wall_seconds": round(batch_wall, 3), + "simulated_seconds_per_wall_second": round(sim / batch_wall, 3) if batch_wall > 0 else None} + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=(__doc__ or "").splitlines()[0]) + parser.add_argument("plan", type=Path, help="plan JSON file (see README.md)") + parser.add_argument("--resume", action="store_true", help="keep finished runs in output_dir and run only the rest") + parser.add_argument("--dry-run", action="store_true", help="validate the plan and print the commands without running") + args = parser.parse_args(argv) + + try: + plan = load_plan(args.plan) + except PlanError as exc: + print(f"plan error: {exc}", file=sys.stderr) + return 2 + runs_dir = plan["output_dir"] / "runs" + + if args.dry_run: + for index, seed in enumerate(plan["seeds"]): + print(subprocess.list2cmdline(build_command(plan, seed, plan["base_port"] + index, runs_dir / f"r{index:04d}" / "engine-result.json"))) + return 0 + + done: dict[int, dict[str, Any]] = {} + if runs_dir.exists() and any(runs_dir.iterdir()): + if not args.resume: + print(f"plan error: {runs_dir} is not empty; use --resume or a new output_dir", file=sys.stderr) + return 2 + for index, seed in enumerate(plan["seeds"]): + run_dir = runs_dir / f"r{index:04d}" + if not run_dir.exists(): + continue + previous = read_record(run_dir / "result.json") + if previous and previous.get("status") == "ok" and previous.get("seed") == seed: + done[index] = previous + else: + # failed, interrupted before it wrote a record, or run with another seed: rerun from a clean workspace + shutil.rmtree(run_dir) + runs_dir.mkdir(parents=True, exist_ok=True) + + suppress_crash_dialogs() + pending = [(i, s) for i, s in enumerate(plan["seeds"]) if i not in done] + records = dict(done) + started = time.monotonic() + with concurrent.futures.ThreadPoolExecutor(max_workers=plan["workers"]) as pool: + futures = [pool.submit(run_one, plan, i, s) for i, s in pending] + for future in concurrent.futures.as_completed(futures): + record = future.result() + records[record["index"]] = record + with open(plan["output_dir"] / "results.jsonl", "a", encoding="utf-8") as sink: + sink.write(json.dumps(record, sort_keys=True) + "\n") + print(f"run {record['index']:>4} seed {record['seed']}: {record['status']}" + + (f" ({'; '.join(record['problems'])})" if record["problems"] else ""), flush=True) + ordered = [records[i] for i in sorted(records)] + summary = summarize(ordered, time.monotonic() - started) + (plan["output_dir"] / "summary.json").write_text(json.dumps(summary, indent=2, sort_keys=True), encoding="utf-8") + print(json.dumps(summary, indent=2, sort_keys=True)) + return 0 if summary["failed"] == 0 else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/headless_batch/test_run_batch.py b/tools/headless_batch/test_run_batch.py new file mode 100644 index 00000000000..45993fa88c7 --- /dev/null +++ b/tools/headless_batch/test_run_batch.py @@ -0,0 +1,245 @@ +"""Tests for run_batch.py. They need no FreeSpace Open install: a stub "engine" (a Python script run by the +current interpreter) stands in for the executable and writes the -sim-result sidecar the real engine writes. + + python -m unittest tools/headless_batch/test_run_batch.py +""" + +import functools +import json +import os +import sys +import tempfile +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) +import run_batch # noqa: E402 + +STUB_OK = r""" +import json, os, sys +args = sys.argv[1:] +path = args[args.index("-sim-result") + 1] +seed = int(args[args.index("-seed") + 1]) +json.dump({"schema": 1, "seed": seed, "simulated_seconds": 12.5, "termination": "time_limit", + "keep_stdout": os.environ.get("FSO_KEEP_STDOUT")}, open(path, "w")) +""" +STUB_START_FAILED = STUB_OK.replace('"time_limit"', '"start_failed"') +STUB_PLAYER_DIED = STUB_OK.replace('"time_limit"', '"player_died"') +STUB_SLEEP = "import time; time.sleep(60)" +STUB_CRASH = "import sys; sys.exit(3)" +# What the engine prints to stderr under -noninteractive for a warning, then a Lua error (message, separator, traceback) +LUA_ERROR_STDERR = ('WARNING: "a warning" at ship.cpp:10\n\n' + 'LUA ERROR: [string "hbtest-sct.tbm - On Frame"]:22: \'end\' expected near \'\'\n\n' + + "-" * 66 + "\nstack traceback:\n\t[C]: ?\n" + "-" * 66 + "\n") +ERROR_STDERR = 'WARNING: "a warning" at ship.cpp:10\n\nError: Mission sm9-99 not found\nFile: missionparse.cpp\nLine: 42\n\n' +STUB_LUA_ERROR = f"import sys; sys.stderr.write({LUA_ERROR_STDERR!r}); sys.exit(1)" + + +class Fixture(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.root = Path(self._tmp.name) + self.base = self.root / "base" + (self.base / "data" / "players").mkdir(parents=True) + (self.base / "root_fs2.vp").write_bytes(b"vp") + (self.base / "data" / "players" / "Tester.json").write_text("{}") + (self.base / "data" / "debug.log").write_text("noise") + (self.base / "mymod" / "data").mkdir(parents=True) + (self.base / "mymod" / "data" / "mod.vp").write_bytes(b"vp") + (self.base / "mymod" / "data" / "table.tbm").write_text("x") + self.engine = Path(sys.executable) + + def tearDown(self): + self._tmp.cleanup() + + def plan_file(self, **overrides): + plan = {"engine": str(self.engine), "base_workspace": str(self.base), "output_dir": str(self.root / "out"), + "mission": "test-mission", "pilot": "Tester", "max_sim_time": 20, "mods": ["mymod"], + "seed_start": 500, "repeats": 3, "workers": 2, "base_port": 40000} + plan.update(overrides) + path = self.root / "plan.json" + path.write_text(json.dumps(plan)) + return path + + def stub(self, source): + path = self.root / "stub.py" + path.write_text(source) + return (str(path),) + + +class PlanTests(Fixture): + def test_defaults_and_seed_expansion(self): + plan = run_batch.load_plan(self.plan_file()) + self.assertEqual(plan["seeds"], [500, 501, 502]) + self.assertAlmostEqual(plan["fixed_dt"], 1 / 30) + self.assertEqual(plan["workers"], 2) + + def test_explicit_seeds_win_over_seed_start(self): + self.assertEqual(run_batch.load_plan(self.plan_file(seeds=[7, 9]))["seeds"], [7, 9]) + + def test_rejects_bad_plans_with_a_message_that_names_the_problem(self): + cases = { + "unknown plan keys": {"fixd_dt": 0.1}, + "fixed_dt must be at least": {"fixed_dt": 0}, + "fixed_dt must be at least 1.52587890625e-05 and at most 0.25": {"fixed_dt": 0.5}, + "max_sim_time": {"max_sim_time": -1}, + "max_sim_time must be above 0.0 and at most 32767": {"max_sim_time": 40000}, + "seeds must be": {"seeds": [0]}, + "simple name": {"mission": "../escape"}, + "mod directory not found": {"mods": ["nope"]}, + "no stored pilot": {"pilot": "Ghost"}, + "port range": {"base_port": 65000, "repeats": 1000}, + "engine must be a path string": {"engine": 5}, + "base_workspace must be a path string": {"base_workspace": ["x"]}, + "output_dir must be a path string": {"output_dir": "a\0b"}, + } + for fragment, overrides in cases.items(): + with self.subTest(fragment): + with self.assertRaises(run_batch.PlanError) as ctx: + run_batch.load_plan(self.plan_file(**overrides)) + self.assertIn(fragment, str(ctx.exception)) + + def test_missing_required_keys_are_listed(self): + path = self.root / "plan.json" + path.write_text(json.dumps({"engine": str(self.engine)})) + with self.assertRaises(run_batch.PlanError) as ctx: + run_batch.load_plan(path) + self.assertIn("base_workspace", str(ctx.exception)) + self.assertIn("pilot", str(ctx.exception)) + + def test_base_workspace_without_vp_files_is_rejected(self): + (self.base / "root_fs2.vp").unlink() + with self.assertRaises(run_batch.PlanError): + run_batch.load_plan(self.plan_file()) + + def test_output_dir_on_another_volume_is_rejected(self): + original = run_batch._volume_of + try: + run_batch._volume_of = lambda path: 1 if path == self.base.resolve() else 2 + with self.assertRaises(run_batch.PlanError) as ctx: + run_batch.load_plan(self.plan_file()) + self.assertIn("same volume", str(ctx.exception)) + finally: + run_batch._volume_of = original + + @unittest.skipUnless(sys.platform == "win32", "the path limit is a Windows limit") + def test_a_too_long_output_dir_is_rejected_with_the_allowed_length(self): + with self.assertRaises(run_batch.PlanError) as ctx: + run_batch.check_path_length(self.base, ["mymod"], self.root / "out", limit=60) + self.assertIn("too long", str(ctx.exception)) + self.assertIn("at most", str(ctx.exception)) + run_batch.check_path_length(self.base, ["mymod"], self.root / "out", limit=10000) # roomy limit: no error + + +class RunTests(Fixture): + def test_command_has_the_headless_flags_and_the_per_run_values(self): + plan = run_batch.load_plan(self.plan_file(extra_args=["-nosound"])) + command = run_batch.build_command(plan, 501, 40001, Path("r.json")) + self.assertEqual(command[0], str(self.engine)) + for flag in ("-headless", "-noninteractive", "-portable_mode", "-nomovies", "-nosound"): + self.assertIn(flag, command) + self.assertEqual(command[command.index("-seed") + 1], "501") + self.assertEqual(command[command.index("-port") + 1], "40001") + self.assertEqual(command[command.index("-mod") + 1], "mymod") + self.assertEqual(command[command.index("-max-sim-time") + 1], "20.0") + + def test_workspace_hardlinks_packs_copies_the_rest_and_skips_logs(self): + destination = self.root / "w" + run_batch.prepare_workspace(self.base, destination, ["mymod"]) + self.assertEqual((destination / "root_fs2.vp").stat().st_ino, (self.base / "root_fs2.vp").stat().st_ino) + self.assertEqual((destination / "mymod" / "data" / "mod.vp").stat().st_ino, + (self.base / "mymod" / "data" / "mod.vp").stat().st_ino) + (destination / "mymod" / "data" / "table.tbm").write_text("changed") + self.assertEqual((self.base / "mymod" / "data" / "table.tbm").read_text(), "x") # a run cannot write through + self.assertFalse((destination / "data" / "debug.log").exists()) + self.assertTrue((destination / "data" / "players" / "Tester.json").is_file()) + + def test_successful_run_collects_the_sidecar(self): + plan = run_batch.load_plan(self.plan_file()) + record = run_batch.run_one(plan, 1, 501, launcher=self.stub(STUB_OK)) + self.assertEqual(record["status"], "ok", record["problems"]) + self.assertEqual(record["sim_result"]["seed"], 501) + # without it the Windows engine sends its output to the runner's console instead of stdout.txt/stderr.txt + self.assertEqual(record["sim_result"]["keep_stdout"], "1") + self.assertEqual(record["port"], 40001) + self.assertTrue((plan["output_dir"] / "runs" / "r0001" / "result.json").is_file()) + + def test_the_engines_own_error_text_is_reported(self): + self.assertEqual(run_batch.engine_reason(ERROR_STDERR), "Error: Mission sm9-99 not found") + self.assertTrue((run_batch.engine_reason(LUA_ERROR_STDERR) or "").startswith("LUA ERROR: [string")) + self.assertIsNone(run_batch.engine_reason('WARNING: "only a warning" at a.cpp:1\n')) + plan = run_batch.load_plan(self.plan_file()) + record = run_batch.run_one(plan, 0, 500, launcher=self.stub(STUB_LUA_ERROR)) + self.assertEqual(record["status"], "failed") + self.assertTrue(any(p.startswith("engine said: LUA ERROR") for p in record["problems"]), record["problems"]) + + def test_crash_and_missing_sidecar_are_failures_not_silent(self): + plan = run_batch.load_plan(self.plan_file()) + record = run_batch.run_one(plan, 0, 500, launcher=self.stub(STUB_CRASH)) + self.assertEqual(record["status"], "failed") + self.assertEqual(record["exit_code"], 3) + self.assertTrue(any("no -sim-result" in p for p in record["problems"])) + + def test_relative_output_dir_still_finds_the_engine_result(self): + # The engine runs in its run folder, so a relative -sim-result path would be written there instead. + old_cwd = os.getcwd() + os.chdir(self.root) + try: + plan = run_batch.load_plan(self.plan_file(output_dir="out")) + record = run_batch.run_one(plan, 0, 500, launcher=self.stub(STUB_OK)) + finally: + os.chdir(old_cwd) + self.assertEqual(record["status"], "ok", record["problems"]) + + def test_a_mission_that_never_started_is_a_failure(self): + plan = run_batch.load_plan(self.plan_file()) + record = run_batch.run_one(plan, 0, 500, launcher=self.stub(STUB_START_FAILED)) + self.assertEqual(record["status"], "failed") + self.assertTrue(any("start_failed" in p for p in record["problems"])) + + def test_a_player_death_is_an_outcome_not_a_failure(self): + plan = run_batch.load_plan(self.plan_file()) + record = run_batch.run_one(plan, 0, 500, launcher=self.stub(STUB_PLAYER_DIED)) + self.assertEqual(record["status"], "ok", record["problems"]) + + def test_resume_reruns_interrupted_unreadable_and_reseeded_runs_and_keeps_good_ones(self): + path = self.plan_file(seeds=[7, 8, 9, 10], repeats=None, seed_start=None) + runs = self.root / "out" / "runs" + (runs / "r0000" / "w").mkdir(parents=True) # interrupted before it wrote result.json + (runs / "r0001").mkdir() + (runs / "r0001" / "result.json").write_text('{"status": "ok", "se') # interrupted mid-write + (runs / "r0002").mkdir() + (runs / "r0002" / "result.json").write_text(json.dumps({"index": 2, "seed": 99, "status": "ok"})) # old plan + (runs / "r0003").mkdir() + kept = {"index": 3, "seed": 10, "status": "ok", "exit_code": 0, "timed_out": False, + "sim_result": {"simulated_seconds": 1.0, "termination": "time_limit"}, "problems": []} + (runs / "r0003" / "result.json").write_text(json.dumps(kept)) + original = run_batch.run_one + run_batch.run_one = functools.partial(original, launcher=self.stub(STUB_OK)) + try: + self.assertEqual(run_batch.main([str(path), "--resume"]), 0) + finally: + run_batch.run_one = original + for index, seed in enumerate([7, 8, 9]): + record = json.loads((runs / f"r{index:04d}" / "result.json").read_text()) + self.assertEqual((record["status"], record["seed"]), ("ok", seed)) + self.assertEqual(json.loads((runs / "r0003" / "result.json").read_text()), kept) # not rerun + + def test_timeout_kills_the_process(self): + plan = run_batch.load_plan(self.plan_file(timeout_seconds=1)) + record = run_batch.run_one(plan, 0, 500, launcher=self.stub(STUB_SLEEP)) + self.assertTrue(record["timed_out"]) + self.assertEqual(record["status"], "failed") + self.assertLess(record["wall_seconds"], 30) + + def test_summary_counts_and_rate(self): + ok = {"status": "ok", "exit_code": 0, "timed_out": False, "sim_result": {"simulated_seconds": 10.0, "termination": "time_limit"}} + bad = {"status": "failed", "exit_code": 3, "timed_out": False, "sim_result": None} + summary = run_batch.summarize([ok, ok, bad], batch_wall=5.0) + self.assertEqual((summary["runs"], summary["ok"], summary["failed"]), (3, 2, 1)) + self.assertEqual(summary["simulated_seconds_per_wall_second"], 4.0) + self.assertEqual(summary["terminations"], {"time_limit": 2}) + + +if __name__ == "__main__": + unittest.main()