382 lines
17 KiB
Python
382 lines
17 KiB
Python
"""Keeps the match and emulator results that the Logbook's exact and verified tiers come from current, so nobody has to
|
|
remember to rerun them:
|
|
|
|
match the whole-program check of main ([reverse] ltcg: `ltcg.py check --build --decomp <main> --json`),
|
|
written to [reverse] match_results; when main's HEAD has moved, at most every match_every_minutes.
|
|
behaviour the emulator over every readable file ([reverse] diffrun_command: `diffrun.py --readable --jobs 4
|
|
--json`), written to [reverse] diffrun_results, its printout to readable.txt beside it; when main's
|
|
HEAD has moved, at most every behaviour_every_hours and only inside behaviour_window (local hours).
|
|
|
|
Configured per project in the [logbook] table (main is [logbook] decomp):
|
|
results_refresh = true off unless set
|
|
match_every_minutes = 60
|
|
behaviour_every_hours = 12
|
|
behaviour_window = "01-06" from 01:00 to 06:00 local time; "" for any hour
|
|
|
|
No run starts while the lander runs its gates ([logbook] lander_lock, harbor.lander_gate).
|
|
One run at a time (the server runs one thread for every project), at the lowest process priority, into a .partial file
|
|
that replaces the results only once it is complete. The match run holds the same per-checkout lock as re_match_program,
|
|
so it waits its turn behind any other whole-program build of main.
|
|
|
|
State (per job: the commit, start and finish of the last run, its counts and error; a run in progress) is kept in the
|
|
logbook folder as results.json, so a restarted server neither repeats a run nor starts a second one beside a run its
|
|
predecessor left going (that run's output is adopted when it finishes). `harbor logbook results --job match|behaviour`
|
|
asks the server for a run now, whatever the schedule; it is a file, refresh-<job>.request, in the same folder."""
|
|
import datetime
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import traceback
|
|
from pathlib import Path
|
|
|
|
from . import lander_gate, logbook
|
|
|
|
JOBS = ("match", "behaviour")
|
|
RESULTS_KEY = {"match": "match_results", "behaviour": "diffrun_results"}
|
|
DEFAULTS = {"results_refresh": False, "match_every_minutes": 60, "behaviour_every_hours": 12,
|
|
"behaviour_window": "01-06"}
|
|
TIMEOUT = {"match": 2 * 3600, "behaviour": 8 * 3600} # generous: the runs are at idle priority on a busy machine
|
|
TICK = 120 # seconds between looks
|
|
START_DELAY = 60 # let a restarted server settle first
|
|
STATE = "results.json"
|
|
|
|
|
|
# ---------------------------------------------------------------------------------------------- configuration
|
|
def settings(project):
|
|
lb = getattr(project, "logbook", None) or {}
|
|
return {k: lb.get(k, v) for k, v in DEFAULTS.items()}
|
|
|
|
|
|
def window(text):
|
|
""""01-06" -> (1, 6): from 01:00 up to 06:00. "22-04" wraps past midnight. ""/None -> None (any hour)."""
|
|
text = str(text or "").strip()
|
|
if not text or text.lower() in ("any", "always"):
|
|
return None
|
|
start, end = (int(part) for part in text.split("-", 1))
|
|
if not (0 <= start <= 24 and 0 <= end <= 24):
|
|
raise ValueError(f"behaviour_window hours must be 0-24: {text!r}")
|
|
return start, end
|
|
|
|
|
|
def in_window(span, hour):
|
|
if span is None:
|
|
return True
|
|
start, end = span
|
|
if start == end:
|
|
return True
|
|
return start <= hour < end if start < end else hour >= start or hour < end
|
|
|
|
|
|
def configured(project, job):
|
|
"""What a job needs from the project, or None when it cannot run here."""
|
|
rev = project.reverse or {}
|
|
lb = getattr(project, "logbook", None) or {}
|
|
if not lb.get("decomp") or not rev.get(RESULTS_KEY[job]):
|
|
return None
|
|
if job == "match" and rev.get("ltcg"):
|
|
return {"script": rev["ltcg"], "results": rev["match_results"]}
|
|
if job == "behaviour" and rev.get("diffrun_command"):
|
|
command = rev["diffrun_command"]
|
|
return {"command": command if isinstance(command, list) else [command], "results": rev["diffrun_results"]}
|
|
return None
|
|
|
|
|
|
# ---------------------------------------------------------------------------------------------- state
|
|
def _folder(project):
|
|
return logbook.folder(project)
|
|
|
|
|
|
def load_state(project):
|
|
path = _folder(project) / STATE
|
|
try:
|
|
state = json.loads(path.read_text(encoding="utf-8")) if path.exists() else {}
|
|
except (OSError, ValueError):
|
|
state = {}
|
|
state.setdefault("jobs", {})
|
|
return state
|
|
|
|
|
|
def save_state(project, state):
|
|
path = _folder(project) / STATE
|
|
partial = path.with_name(path.name + ".partial")
|
|
partial.write_text(json.dumps(state, indent=1), encoding="utf-8")
|
|
os.replace(partial, path)
|
|
|
|
|
|
def request(project, job):
|
|
"""Ask the server's refresher for a run of `job` at its next look, whatever the schedule."""
|
|
if job not in JOBS:
|
|
raise ValueError(f"job is one of {', '.join(JOBS)}")
|
|
path = _folder(project) / f"refresh-{job}.request"
|
|
path.write_text(_iso(time.time()), encoding="utf-8")
|
|
return {"requested": job, "file": str(path)}
|
|
|
|
|
|
def _requested(project):
|
|
return {job for job in JOBS if (_folder(project) / f"refresh-{job}.request").exists()}
|
|
|
|
|
|
def _clear_request(project, job):
|
|
try:
|
|
(_folder(project) / f"refresh-{job}.request").unlink()
|
|
except FileNotFoundError:
|
|
pass
|
|
|
|
|
|
def _iso(ts):
|
|
return datetime.datetime.fromtimestamp(ts, datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
|
|
|
|
# ---------------------------------------------------------------------------------------------- the decision
|
|
def due(cfg, state, job, head, now, hour, requested=()):
|
|
"""(True, why) when `job` should run now; (False, why not) otherwise. Pure: the tests drive it directly."""
|
|
if job in requested:
|
|
return True, "requested"
|
|
if not cfg.get("results_refresh"):
|
|
return False, "results_refresh is off"
|
|
last = state.get("jobs", {}).get(job, {})
|
|
if last.get("head") == head:
|
|
return False, "already at main's HEAD"
|
|
if job == "match":
|
|
every = float(cfg.get("match_every_minutes", 60)) * 60
|
|
else:
|
|
every = float(cfg.get("behaviour_every_hours", 12)) * 3600
|
|
if last.get("started_ts") and now - last["started_ts"] < every:
|
|
return False, f"last run started {int((now - last['started_ts']) / 60)} min ago"
|
|
if job == "behaviour" and not in_window(window(cfg.get("behaviour_window")), hour):
|
|
return False, f"outside behaviour_window {cfg.get('behaviour_window')}"
|
|
return True, "main moved"
|
|
|
|
|
|
def choose(cfg, state, head, now, hour, requested=(), available=JOBS):
|
|
"""The one job to run now (the match first: it is quicker and more often due), or (None, reasons)."""
|
|
reasons = {}
|
|
for job in JOBS:
|
|
if job not in available:
|
|
reasons[job] = "not configured"
|
|
continue
|
|
ok, why = due(cfg, state, job, head, now, hour, requested)
|
|
if ok:
|
|
return job, why
|
|
reasons[job] = why
|
|
return None, reasons
|
|
|
|
|
|
# ---------------------------------------------------------------------------------------------- running
|
|
def _head(project):
|
|
from .logbook_auto import _git
|
|
return _git(project.logbook["decomp"], "rev-parse", "HEAD").strip()
|
|
|
|
|
|
def _spawn(argv, cwd, stdout, stderr):
|
|
flags = 0
|
|
if os.name == "nt":
|
|
flags = subprocess.IDLE_PRIORITY_CLASS | subprocess.CREATE_NO_WINDOW # children (cl.exe) inherit idle
|
|
env = dict(os.environ, PYTHONIOENCODING="utf-8")
|
|
return subprocess.Popen(argv, cwd=str(cwd), stdin=subprocess.DEVNULL, stdout=stdout, stderr=stderr, env=env,
|
|
creationflags=flags)
|
|
|
|
|
|
def _kill_tree(pid):
|
|
lander_gate.kill_tree(pid)
|
|
|
|
|
|
def _alive(pid):
|
|
from .reverse import pid_alive
|
|
return bool(pid) and pid_alive(pid)
|
|
|
|
|
|
def _paths(project, job):
|
|
target = Path(configured(project, job)["results"])
|
|
paths = {"target": target, "partial": target.with_name(target.name + ".partial"),
|
|
"log": _folder(project) / f"{job}.log"}
|
|
if job == "behaviour":
|
|
paths["text"] = target.with_suffix(".txt")
|
|
paths["text_partial"] = target.with_suffix(".txt.partial")
|
|
return paths
|
|
|
|
|
|
def _argv(project, job, out, decomp=None):
|
|
need = configured(project, job)
|
|
if job == "match":
|
|
script = Path(need["script"])
|
|
return [sys.executable, "-B", str(script), "check", "--build", "--decomp",
|
|
str(Path(decomp or project.logbook["decomp"])), "--json", str(out)], script.parent
|
|
command = [str(c) for c in need["command"]]
|
|
return command + ["--readable", "--jobs", "4", "--json", str(out)], Path(command[-1]).parent
|
|
|
|
|
|
def _match_lock(project):
|
|
"""The lock re_match_program holds around a whole-program build of main (harbor.reverse.match_program)."""
|
|
from .reverse import _exclusive
|
|
rev = project.reverse or {}
|
|
return _exclusive(Path(rev.get("debugger_temp") or tempfile.gettempdir()) / "ltcg-main.lock", wait_seconds=0)
|
|
|
|
|
|
def _counts(rows, field):
|
|
counts = {}
|
|
for r in rows:
|
|
counts[str(r.get(field))] = counts.get(str(r.get(field)), 0) + 1
|
|
return counts
|
|
|
|
|
|
def _complete(project, state, run, now, returncode=None):
|
|
"""Move a finished run's output into place, or record why it failed. Clears state['running']."""
|
|
job, paths = run["job"], _paths(project, run["job"])
|
|
last = dict(state["jobs"].get(job, {}))
|
|
last.update(started_ts=run["started_ts"], started=_iso(run["started_ts"]), finished=_iso(now))
|
|
try:
|
|
partial = paths["partial"]
|
|
text = partial.read_text(encoding="utf-8") if partial.exists() else ""
|
|
if not text.strip():
|
|
ended = "the run was interrupted" if returncode is None else \
|
|
returncode if isinstance(returncode, str) else f"exit code {returncode}"
|
|
raise RuntimeError(f"no results were written ({ended}); see {paths['log']}")
|
|
rows = json.loads(text)
|
|
if not isinstance(rows, list) or not rows:
|
|
raise RuntimeError("the results file is empty")
|
|
os.replace(partial, paths["target"])
|
|
if job == "behaviour" and paths["text_partial"].exists():
|
|
os.replace(paths["text_partial"], paths["text"])
|
|
last.update(head=run["head"], ok=True, error=None, functions=len(rows),
|
|
counts=_counts(rows, "status" if job == "match" else "verdict"),
|
|
results_mtime_ns=paths["target"].stat().st_mtime_ns)
|
|
except (OSError, ValueError, RuntimeError) as error:
|
|
last.update(ok=False, error=str(error)[:500])
|
|
state["jobs"][job] = last
|
|
state.pop("running", None)
|
|
save_state(project, state)
|
|
return {"job": job, "ok": last["ok"], "head": run["head"][:8], **({"error": last["error"]} if not last["ok"] else {})}
|
|
|
|
|
|
def run_job(project, state, job, head, now, reason=""):
|
|
paths = _paths(project, job)
|
|
decomp = None
|
|
if job == "match" and project.logbook.get("results_snapshot") and head:
|
|
# a clean checkout of exactly `head`: the live checkout is the lander's and changes under a running build
|
|
from .logbook_auto import snapshot_at
|
|
decomp = snapshot_at({"snapshot": project.logbook["results_snapshot"]}, project.logbook["decomp"], head)
|
|
argv, cwd = _argv(project, job, paths["partial"], decomp)
|
|
for stale in [paths["partial"]] + ([paths["text_partial"]] if job == "behaviour" else []):
|
|
if stale.exists():
|
|
stale.write_bytes(b"") # a leftover of an interrupted run is never taken for this one
|
|
run = {"job": job, "pid": None, "head": head, "started_ts": now, "reason": reason}
|
|
clock = time.monotonic()
|
|
returncode = None
|
|
with open(paths["log"], "wb") as log:
|
|
stdout = open(paths["text_partial"], "wb") if job == "behaviour" else log
|
|
try:
|
|
log.write(f"{_iso(now)} {job} at {head[:8]} ({reason}): {' '.join(argv)}\n".encode("utf-8"))
|
|
log.flush()
|
|
proc = _spawn(argv, cwd, stdout, log if job == "behaviour" else subprocess.STDOUT)
|
|
run["pid"] = proc.pid
|
|
state["running"] = {**run, "started": _iso(now)}
|
|
save_state(project, state)
|
|
try:
|
|
returncode = proc.wait(timeout=TIMEOUT[job])
|
|
except subprocess.TimeoutExpired:
|
|
_kill_tree(proc.pid)
|
|
proc.wait()
|
|
returncode = f"timed out after {TIMEOUT[job] // 60} min"
|
|
except OSError as error:
|
|
returncode = f"could not start: {error}"
|
|
finally:
|
|
if stdout is not log:
|
|
stdout.close()
|
|
return _complete(project, state, run, now + (time.monotonic() - clock), returncode)
|
|
|
|
|
|
def tick(project, now=None, hour=None):
|
|
"""One look: finish off a run an earlier server left, then start the job that is due (and wait for it)."""
|
|
now = time.time() if now is None else now
|
|
hour = datetime.datetime.now().hour if hour is None else hour
|
|
available = [job for job in JOBS if configured(project, job)]
|
|
if not available:
|
|
return {"idle": "not configured"}
|
|
state = load_state(project)
|
|
running = state.get("running")
|
|
if running:
|
|
if _alive(running.get("pid")) and now - running.get("started_ts", 0) < TIMEOUT.get(running.get("job"), 0):
|
|
return {"busy": running["job"], "pid": running["pid"]}
|
|
return {"adopted": _complete(project, state, running, now)}
|
|
requested = _requested(project)
|
|
if not requested and not settings(project).get("results_refresh"):
|
|
return {"idle": "results_refresh is off"}
|
|
head = _head(project)
|
|
job, why = choose(settings(project), state, head, now, hour, requested, available)
|
|
if not job:
|
|
return {"idle": why, "head": head[:8]}
|
|
busy = lander_gate.lander_busy(project, now)
|
|
if busy: # a request is kept for the next look
|
|
return {"waiting": busy, "job": job}
|
|
if job == "match":
|
|
from .util import HarborError
|
|
try:
|
|
with _match_lock(project):
|
|
_clear_request(project, job)
|
|
return run_job(project, state, job, head, now, why)
|
|
except HarborError:
|
|
return {"waiting": "another whole-program build of main holds the lock"}
|
|
_clear_request(project, job)
|
|
return run_job(project, state, job, head, now, why)
|
|
|
|
|
|
# ---------------------------------------------------------------------------------------------- reading
|
|
def freshness(project):
|
|
"""For the progress view: which commit the exact and verified tiers are from, and how old they are."""
|
|
state = load_state(project)
|
|
out = {}
|
|
for job in JOBS:
|
|
need = configured(project, job)
|
|
target = Path(need["results"]) if need else Path((project.reverse or {}).get(RESULTS_KEY[job]) or "")
|
|
last = state["jobs"].get(job, {})
|
|
try:
|
|
mtime_ns = target.stat().st_mtime_ns if str(target) not in ("", ".") else None
|
|
except OSError:
|
|
mtime_ns = None
|
|
if mtime_ns is None:
|
|
out[job] = None
|
|
continue
|
|
ours = last.get("ok") and last.get("results_mtime_ns") == mtime_ns
|
|
out[job] = {"sha": last["head"][:7] if ours else None,
|
|
"at": last.get("finished") if ours else _iso(mtime_ns / 1e9),
|
|
"error": last.get("error") if last.get("ok") is False else None}
|
|
running = state.get("running")
|
|
out["running"] = {"job": running["job"], "started": running.get("started")} if running else None
|
|
out["enabled"] = bool(settings(project).get("results_refresh"))
|
|
return out
|
|
|
|
|
|
def status(project, now=None, hour=None):
|
|
"""For `harbor logbook results`: the state plus what the next look would do."""
|
|
now = time.time() if now is None else now
|
|
hour = datetime.datetime.now().hour if hour is None else hour
|
|
state = load_state(project)
|
|
available = [job for job in JOBS if configured(project, job)]
|
|
try:
|
|
head = _head(project) if available else ""
|
|
except RuntimeError:
|
|
head = ""
|
|
nxt = {job: due(settings(project), state, job, head, now, hour, _requested(project))
|
|
for job in available}
|
|
return {"settings": settings(project), "head": head[:8], "jobs": state["jobs"], "running": state.get("running"),
|
|
"requested": sorted(_requested(project)),
|
|
"next": {job: {"due": ok, "why": why} for job, (ok, why) in nxt.items()},
|
|
"state_file": str(_folder(project) / STATE)}
|
|
|
|
|
|
def loop(registry):
|
|
"""Server thread: one look every two minutes at every project (results_refresh on, or a run requested)."""
|
|
time.sleep(START_DELAY)
|
|
while True:
|
|
for name in registry.names():
|
|
try:
|
|
h, _ = registry.get(name)
|
|
tick(h.project)
|
|
except Exception:
|
|
traceback.print_exc()
|
|
time.sleep(TICK)
|