Resuming and Re-Running Only What Failed

TOTAL, DONE = 400, 320

print("jobs in the regression:", TOTAL)
print("jobs completed        :", DONE)
print("jobs remaining        :", TOTAL - DONE)
print()
print("restarting from zero re-runs:", DONE, "jobs that already finished")
print("wasted fraction             :", f"{100 * DONE / TOTAL:.0f}%")
print()
print("what is needed to avoid it  :", "which jobs finished, and what they produced")
print("how much data that is       :", f"{TOTAL} lines")

jobs in the regression: 400
jobs completed        : 320
jobs remaining        : 80

restarting from zero re-runs: 320 jobs that already finished
wasted fraction             : 80%

what is needed to avoid it  : which jobs finished, and what they produced
how much data that is       : 400 lines

todo = [j for j in jobs if state.get(j) != "PASS"]

JOBS = [("a", 1), ("b", 2), ("c", 3), ("d", 4), ("e", 5)]
STATE = {("a", 1): "PASS", ("b", 2): "FAIL", ("c", 3): "ERROR", ("d", 4): "NO_FINISH"}

todo = [j for j in JOBS if STATE.get(j) != "PASS"]

print("jobs          :", len(JOBS))
print("have a state  :", len(STATE))
print("known to pass :", sum(1 for v in STATE.values() if v == "PASS"))
print()
print("todo          :", todo)
print()
print("('e', 5) has no state :", STATE.get(("e", 5)))
print("and None != 'PASS'    :", STATE.get(("e", 5)) != "PASS")
print("so it is included     :", ("e", 5) in todo)

jobs          : 5
have a state  : 4
known to pass : 1

todo          : [('b', 2), ('c', 3), ('d', 4), ('e', 5)]

('e', 5) has no state : None
and None != 'PASS'    : True
so it is included     : True

JOBS = [("a", 1), ("b", 2), ("e", 5)]
STATE = {("a", 1): "PASS", ("b", 2): "FAIL"}

annotation = [j for j in JOBS if STATE.get(j) != "PASS"]
membership = [j for j in JOBS if j in STATE and STATE[j] != "PASS"]

print("annotation form:", annotation)
print("membership form:", membership)
print()
print("dropped by the membership form:", [j for j in annotation if j not in membership])
print("what those jobs are            :", "the ones the regression never reached")

annotation form: [('b', 2), ('e', 5)]
membership form: [('b', 2)]

dropped by the membership form: [('e', 5)]
what those jobs are            : the ones the regression never reached

JOBS = [("a", 1), ("b", 2), ("c", 3), ("d", 4), ("e", 5)]
STATE = {("a", 1): "PASS", ("b", 2): "FAIL", ("c", 3): "ERROR", ("d", 4): "NO_FINISH"}

MODES = {
    "not-pass": lambda s: s != "PASS",
    "never-ran-only": lambda s: s is None,
    "resume-plus-infra": lambda s: s not in ("PASS", "FAIL"),
}

for name, keep in MODES.items():
    todo = [j for j in JOBS if keep(STATE.get(j))]
    print(f"{name:<19} {len(todo)} job(s): {todo}")

print()
print("all three are defensible:", True)
print("they answer            :", "different questions")

not-pass            4 job(s): [('b', 2), ('c', 3), ('d', 4), ('e', 5)]
never-ran-only      1 job(s): [('e', 5)]
resume-plus-infra   3 job(s): [('c', 3), ('d', 4), ('e', 5)]

all three are defensible: True
they answer            : different questions

SITUATIONS = [
    ("job finished, record written", "readable", "state recoverable from disk"),
    ("job finished, record corrupt", "unreadable", "looks identical to never-ran"),
    ("job was running when the crash hit", "no record", "looks identical to never-ran"),
    ("job never started", "no record", "never ran"),
]

print(f"{'situation':<38} {'on disk':<12} inference")
for what, disk, inference in SITUATIONS:
    print(f"{what:<38} {disk:<12} {inference}")

print()
indistinguishable = [s for s, d, _ in SITUATIONS if d != "readable"]
print("situations that look the same on disk:", len(indistinguishable), "of", len(SITUATIONS))
print("why it matters:", "a job that was mid-run left a half-written run directory")

situation                              on disk      inference
job finished, record written           readable     state recoverable from disk
job finished, record corrupt           unreadable   looks identical to never-ran
job was running when the crash hit     no record    looks identical to never-ran
job never started                      no record    never ran

situations that look the same on disk: 3 of 4
why it matters: a job that was mid-run left a half-written run directory

import json
from pathlib import Path

Path("st").mkdir(exist_ok=True)
state_file = Path("st/state.json")

state_file.write_text('{"a_1": "PASS", "b_2": "PASS", "c_3": "FA', encoding="utf-8")

print("what is on disk:", repr(state_file.read_text(encoding="utf-8")))
print()
try:
    loaded = json.loads(state_file.read_text(encoding="utf-8"))
    print("loaded:", loaded)
except json.JSONDecodeError as exc:
    print("reload :", type(exc).__name__, "-", exc.msg)
    print("entries recovered:", 0)
    print("entries actually written before the crash:", 2)

what is on disk: '{"a_1": "PASS", "b_2": "PASS", "c_3": "FA'

reload : JSONDecodeError - Unterminated string starting at
entries recovered: 0
entries actually written before the crash: 2

import json
import os
from pathlib import Path


def save_atomic(path, obj):
    """Write to a temp file in the same directory, then rename over the target."""
    path = Path(path)
    tmp = path.with_suffix(path.suffix + ".tmp")
    tmp.write_text(json.dumps(obj), encoding="utf-8")
    os.replace(tmp, path)


Path("st2").mkdir(exist_ok=True)
save_atomic("st2/state.json", {"a_1": "PASS", "b_2": "FAIL"})

print("reloads cleanly :", json.loads(Path("st2/state.json").read_text(encoding="utf-8")))
print("temp file gone  :", not Path("st2/state.json.tmp").exists())
print()
print("why it is safe  :", "os.replace is atomic — the target is the old file or the new one")
print("the cost        :", "the whole file is rewritten once per job")

reloads cleanly : {'a_1': 'PASS', 'b_2': 'FAIL'}
temp file gone  : True

why it is safe  : os.replace is atomic — the target is the old file or the new one
the cost        : the whole file is rewritten once per job

import json
from pathlib import Path

Path("st3").mkdir(exist_ok=True)
jsonl = Path("st3/state.jsonl")

jsonl.write_text('{"job": "a_1", "status": "PASS"}\n'
                 '{"job": "b_2", "status": "FAIL"}\n'
                 '{"job": "c_3", "sta', encoding="utf-8")

recovered, discarded = [], []
for number, line in enumerate(jsonl.read_text(encoding="utf-8").splitlines(), start=1):
    try:
        recovered.append(json.loads(line))
    except json.JSONDecodeError:
        discarded.append(number)

print("lines on disk   :", len(jsonl.read_text(encoding='utf-8').splitlines()))
print("recovered       :", len(recovered))
print("discarded lines :", discarded)
print()
for entry in recovered:
    print("  ", entry)
print()
print("entries lost    :", len(discarded), "— only the one being written")

lines on disk   : 3
recovered       : 2
discarded lines : [3]

   {'job': 'a_1', 'status': 'PASS'}
   {'job': 'b_2', 'status': 'FAIL'}

entries lost    : 1 — only the one being written

JOBS = 400

print(f"{'format':<22} {'writes':>8} {'bytes rewritten':>18} {'worst-case loss'}")
print(f"{'whole-file JSON':<22} {JOBS:>8} {'O(n) each time':>18} {'all entries'}")
print(f"{'whole-file + atomic':<22} {JOBS:>8} {'O(n) each time':>18} {'nothing'}")
print(f"{'append-only JSONL':<22} {JOBS:>8} {'one line each':>18} {'one entry'}")
print()
print("total bytes written, whole-file, for 400 jobs:", "proportional to 400 x 400")
print("total bytes written, append-only            :", "proportional to 400")
print("ratio                                       :", JOBS)

format                   writes    bytes rewritten worst-case loss
whole-file JSON             400     O(n) each time all entries
whole-file + atomic         400     O(n) each time nothing
append-only JSONL           400      one line each one entry

total bytes written, whole-file, for 400 jobs: proportional to 400 x 400
total bytes written, append-only            : proportional to 400
ratio                                       : 400

import json

try:
    json.dumps({("axi_burst", 1839274): "PASS"})
except TypeError as exc:
    print("TypeError:", exc)

print()
print("encoded:", json.dumps({"axi_burst/1839274": "PASS"}))
print()


def key_of(test, seed):
    return f"{test}/{seed}"


def parse_key(key):
    test, _, seed = key.rpartition("/")
    return test, int(seed)


for test, seed in (("axi_burst", 1839274), ("grp/axi_smoke", 7)):
    k = key_of(test, seed)
    print(f"{str((test, seed)):<28} -> {k:<24} -> {parse_key(k)}")

print()
print("round-trips with a slash in the test name:",
      parse_key(key_of("grp/axi_smoke", 7)) == ("grp/axi_smoke", 7))

TypeError: keys must be str, int, float, bool or None, not tuple

encoded: {"axi_burst/1839274": "PASS"}

('axi_burst', 1839274)       -> axi_burst/1839274        -> ('axi_burst', 1839274)
('grp/axi_smoke', 7)         -> grp/axi_smoke/7          -> ('grp/axi_smoke', 7)

round-trips with a slash in the test name: True

import json
from dataclasses import dataclass
from pathlib import Path


@dataclass(frozen=True)
class Job:
    test: str
    seed: int


MODES = {
    "never-ran-only": lambda s: s is None,
    "resume-plus-infra": lambda s: s not in ("PASS", "FAIL"),
    "not-pass": lambda s: s != "PASS",
}


def record_state(path, job, status):
    """Append one line. Never rewrites, so a crash costs at most this entry."""
    with open(path, "a", encoding="utf-8") as handle:
        handle.write(json.dumps({"test": job.test, "seed": job.seed,
                                 "status": status}) + "\n")
        handle.flush()


def load_state(path):
    """Last entry wins. A truncated final line is discarded."""
    state, dropped = {}, 0
    if not Path(path).exists():
        return state, dropped
    for line in Path(path).read_text(encoding="utf-8").splitlines():
        try:
            e = json.loads(line)
            state[Job(e["test"], e["seed"])] = e["status"]
        except (json.JSONDecodeError, KeyError):
            dropped += 1
    return state, dropped


def plan_resume(jobs, state, *, mode="resume-plus-infra"):
    if mode not in MODES:
        raise ValueError(f"unknown resume mode {mode!r} — expected one of {sorted(MODES)}")
    keep = MODES[mode]
    return [j for j in jobs if keep(state.get(j))]


JOBS = [Job("axi_smoke", 1), Job("axi_reset", 2), Job("axi_burst", 3),
        Job("axi_lic", 4), Job("axi_new", 5)]

path = Path("state.jsonl")
for job, status in zip(JOBS, ["PASS", "PASS", "FAIL", "ERROR"]):
    record_state(path, job, status)
# the crash: a truncated final line
with open(path, "a", encoding="utf-8") as h:
    h.write('{"test": "axi_new", "se')

state, dropped = load_state(path)
print("entries recovered :", len(state))
print("lines discarded   :", dropped)
print()
for mode in MODES:
    todo = plan_resume(JOBS, state, mode=mode)
    print(f"{mode:<19} {len(todo)} job(s): {[(j.test, j.seed) for j in todo]}")

print()
try:
    plan_resume(JOBS, state, mode="clever")
except ValueError as exc:
    print("guard:", exc)

print()
print("total is still the saved job list:", len(JOBS))
print("so the invariant holds on resume :", True)

entries recovered : 4
lines discarded   : 1

never-ran-only      1 job(s): [('axi_new', 5)]
resume-plus-infra   2 job(s): [('axi_lic', 4), ('axi_new', 5)]
not-pass            3 job(s): [('axi_burst', 3), ('axi_lic', 4), ('axi_new', 5)]

guard: unknown resume mode 'clever' — expected one of ['never-ran-only', 'not-pass', 'resume-plus-infra']

total is still the saved job list: 5
so the invariant holds on resume : True

from dataclasses import dataclass
from typing import Optional


@dataclass(frozen=True)
class Result:
    test: str
    seed: int
    sim: str
    status: str
    reason: Optional[str] = None

    @property
    def rerun(self):
        return f"flow run --test {self.test} --seed {self.seed} --sim {self.sim} --waves"


FAILURES = [Result("axi_burst", 1839274, "vcs", "FAIL", "UVM_ERROR"),
            Result("axi_hang", 771002, "vcs", "NO_FINISH", "TIMEOUT")]

print("re-run one failure:")
for f in FAILURES:
    print(f"   {f.test:<11} {f.rerun}")

print()
print("re-run everything that failed:")
print("   flow regress --resume --mode not-pass")
print()
print("the difference:", "one command debugs; the other re-confirms")

re-run one failure:
   axi_burst   flow run --test axi_burst --seed 1839274 --sim vcs --waves
   axi_hang    flow run --test axi_hang --seed 771002 --sim vcs --waves

re-run everything that failed:
   flow regress --resume --mode not-pass

the difference: one command debugs; the other re-confirms

import json
from pathlib import Path

JOBS = [("a", 1), ("b", 2), ("c", 3)]
STATE = {("a", 1): "PASS", ("b", 2): "FAIL"}

# A
print("A:", STATE.get(("c", 3)))

# B
print("B:", [j for j in JOBS if STATE.get(j) != "PASS"])

# C
print("C:", [j for j in JOBS if j in STATE and STATE[j] != "PASS"])

# D
print("D:", [j for j in JOBS if STATE.get(j) is None])

# E
try:
    json.dumps({("a", 1): "PASS"})
except TypeError as exc:
    print("E:", str(exc)[:42])

# F
print("F:", "grp/t/7".rpartition("/")[0], "grp/t/7".split("/")[0])

# G
Path("pro.jsonl").write_text('{"x": 1}\n{"y": 2}\n{"z"', encoding="utf-8")
ok = 0
for line in Path("pro.jsonl").read_text(encoding="utf-8").splitlines():
    try:
        json.loads(line)
        ok += 1
    except json.JSONDecodeError:
        pass
print("G:", ok)

# H
Path("pro.json").write_text('{"x": 1, "y"', encoding="utf-8")
try:
    json.loads(Path("pro.json").read_text(encoding="utf-8"))
    print("H: loaded")
except json.JSONDecodeError:
    print("H: 0 recovered")

A: None
B: [('b', 2), ('c', 3)]
C: [('b', 2)]
D: [('c', 3)]
E: keys must be str, int, float, bool or None
F: grp/t grp
G: 2
H: 0 recovered