Reviewing a Regression System: Isolation, Races, Resume, Accounting

FAILURES = [
    ("a parser returns the wrong status", "a wrong answer", "a row is wrong"),
    ("a report sorts badly", "a wrong order", "a diff is noisy"),
    ("a job's result is lost", "a SHORTER answer", "nothing looks wrong"),
    ("two jobs share a directory", "one answer twice", "two rows, one log"),
    ("a resume deletes a log", "no answer at all", "the evidence is gone"),
]

print(f"{'what goes wrong':<38} {'class':<18} what a reader sees")
for what, cls, sees in FAILURES:
    print(f"{what:<38} {cls:<18} {sees}")

print()
print("failures that produce a wrong value :", 2)
print("failures that produce LESS data     :", 3)
print()
print("a wrong value can be checked against:", "the input")
print("missing data can only be checked    ")
print("  against                            :", "a COUNT of what was asked for")
print()
print("which is why the review's core check is:", "an accounting assertion")

what goes wrong                        class              what a reader sees
a parser returns the wrong status      a wrong answer     a row is wrong
a report sorts badly                   a wrong order      a diff is noisy
a job's result is lost                 a SHORTER answer   nothing looks wrong
two jobs share a directory             one answer twice   two rows, one log
a resume deletes a log                 no answer at all   the evidence is gone

failures that produce a wrong value : 2
failures that produce LESS data     : 3

a wrong value can be checked against: the input
missing data can only be checked    
  against                            : a COUNT of what was asked for

which is why the review's core check is: an accounting assertion

assert len(results) == len(jobs)   # the check that catches silent job loss

PROPERTIES = [
    ("isolation", "one directory per job", "§6"),
    ("races", "no shared mutable state", "§6"),
    ("exceptions", "worker failures surfaced", "§4"),
    ("accounting", "every job in the results", "the annotation"),
    ("resume", "does not discard evidence", "§7"),
]

print(f"{'property':<14} {'what it requires':<30} where")
for what, requires, where in PROPERTIES:
    print(f"{what:<14} {requires:<30} {where}")

print()
print("properties                          :", len(PROPERTIES))
print("that the one assertion detects      :", 3)
print("  — a swallowed worker exception")
print("  — a results dict that collapses keys")
print("  — a resume that skips without a result")
print()
print("what it does NOT detect             :",
      ["a shared directory", "a wrong value"])
print("so it is necessary and              :", "not sufficient")

property       what it requires               where
isolation      one directory per job          §6
races          no shared mutable state        §6
exceptions     worker failures surfaced       §4
accounting     every job in the results       the annotation
resume         does not discard evidence      §7

properties                          : 5
that the one assertion detects      : 3
  — a swallowed worker exception
  — a results dict that collapses keys
  — a resume that skips without a result

what it does NOT detect             : ['a shared directory', 'a wrong value']
so it is necessary and              : not sufficient

from concurrent.futures import ThreadPoolExecutor

JOBS = ([("axi_burst", s) for s in (1, 2, 3)]
        + [("axi_lite", s) for s in (1, 2, 3)])


def work(job):
    """One job. Seed 2 raises, as a flaky tool would."""
    test, seed = job
    if seed == 2:
        raise RuntimeError(f"{test} seed {seed} blew up")
    return {"test": test, "seed": seed, "status": "PASS"}


print("jobs submitted:", len(JOBS))
print()

# --- specimen A: collect by filtering out the futures that raised
with ThreadPoolExecutor(max_workers=3) as pool:
    futures = [pool.submit(work, j) for j in JOBS]
    results_a = [f.result() for f in futures if not f.exception()]

print("specimen A — filter out futures with an exception")
print("  results collected :", len(results_a))
print("  jobs lost         :", len(JOBS) - len(results_a))
print("  exception reached ")
print("    the caller      :", False)
print()

# --- specimen B: collect with map
raised = None
try:
    with ThreadPoolExecutor(max_workers=3) as pool:
        results_b = list(pool.map(work, JOBS))
except RuntimeError as exc:
    raised = str(exc)

print("specimen B — ThreadPoolExecutor.map")
print("  RuntimeError raised:", raised is not None)
print("  message            :", (raised or "")[:34])
print("  so map() is        :", "NOT silent")
print()

# --- specimen C: collect into a dict keyed by test name only
results_c = {}
with ThreadPoolExecutor(max_workers=3) as pool:
    for job, fut in [(j, pool.submit(work, j)) for j in JOBS]:
        if fut.exception():
            continue
        r = fut.result()
        results_c[r["test"]] = r

print("specimen C — a results dict keyed by test name")
print("  jobs               :", len(JOBS))
print("  dict entries       :", len(results_c))
print("  silently collapsed :", len(JOBS) - len(results_c))
print()

# --- specimen D: the annotation
print("specimen D — assert len(results) == len(jobs)")
for label, got in (("A", len(results_a)), ("C", len(results_c))):
    try:
        assert got == len(JOBS), f"{got} results for {len(JOBS)} jobs"
        print(f"  on {label}: passed")
    except AssertionError as exc:
        print(f"  on {label}: AssertionError: {exc}")

jobs submitted: 6

specimen A — filter out futures with an exception
  results collected : 4
  jobs lost         : 2
  exception reached 
    the caller      : False

specimen B — ThreadPoolExecutor.map
  RuntimeError raised: True
  message            : axi_burst seed 2 blew up
  so map() is        : NOT silent

specimen C — a results dict keyed by test name
  jobs               : 6
  dict entries       : 2
  silently collapsed : 4

specimen D — assert len(results) == len(jobs)
  on A: AssertionError: 4 results for 6 jobs
  on C: AssertionError: 2 results for 6 jobs

IDIOMS = [
    ("pool.map(work, jobs)", "raises on iteration", "safe by default"),
    ("[f.result() for f in futures]", "raises on the first bad one",
     "safe by default"),
    ("[f.result() for f in futures if not f.exception()]",
     "silently drops", "the specimen A defect"),
    ("for f in as_completed(futures): try/except: pass",
     "silently drops", "the same defect, spelled out"),
    ("results[r['test']] = r", "silently collapses", "specimen C"),
    ("results.append(r) inside the worker",
     "races, and may drop", "shared mutable state"),
]

print(f"{'collection idiom':<52} {'on a worker error':<28} verdict")
for idiom, behaviour, verdict in IDIOMS:
    print(f"{idiom:<52} {behaviour:<28} {verdict}")

print()
safe = sum(1 for _, b, _ in IDIOMS if b.startswith("raises"))
print("idioms that surface the error:", safe)
print("idioms that hide it          :", len(IDIOMS) - safe)
print()
print("the two safe ones share      :", "they call .result() unconditionally")
print("and the review question is   :",
      "is there an `if` between the future and its result?")

collection idiom                                     on a worker error            verdict
pool.map(work, jobs)                                 raises on iteration          safe by default
[f.result() for f in futures]                        raises on the first bad one  safe by default
[f.result() for f in futures if not f.exception()]   silently drops               the specimen A defect
for f in as_completed(futures): try/except: pass     silently drops               the same defect, spelled out
results[r['test']] = r                               silently collapses           specimen C
results.append(r) inside the worker                  races, and may drop          shared mutable state

idioms that surface the error: 2
idioms that hide it          : 4

the two safe ones share      : they call .result() unconditionally
and the review question is   : is there an `if` between the future and its result?

from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass

JOBS = ([("axi_burst", s) for s in (1, 2, 3)]
        + [("axi_lite", s) for s in (1, 2, 3)])


@dataclass(frozen=True)
class Result:
    test: str
    seed: int
    status: str
    reason: str = ""

    @property
    def name(self):
        return f"{self.test}_{self.seed}"


def work(job):
    test, seed = job
    if seed == 2:
        raise RuntimeError(f"{test} seed {seed} blew up")
    return Result(test, seed, "PASS")


def run_job(job):
    """A worker error becomes a RESULT, not a missing row."""
    try:
        return work(job)
    except Exception as exc:
        return Result(job[0], job[1], "ERROR",
                      f"{type(exc).__name__}: {exc}")


with ThreadPoolExecutor(max_workers=3) as pool:
    results = list(pool.map(run_job, JOBS))

assert len(results) == len(JOBS), f"{len(results)} results for {len(JOBS)} jobs"

print(f"{'job':<18} {'status':<8} reason")
for r in sorted(results, key=lambda r: (r.test, r.seed)):
    print(f"{r.name:<18} {r.status:<8} {r.reason}")

print()
print("jobs submitted  :", len(JOBS))
print("results          :", len(results))
print("the assertion    :", "passed")
print()
from collections import Counter
for status, n in sorted(Counter(r.status for r in results).items()):
    print(f"  {status:<8} {n}")
print()
print("jobs lost        :", 0)
print("errors visible   :", sum(1 for r in results if r.status == "ERROR"))
print("and the sweep     :", "did not abort")

job                status   reason
axi_burst_1        PASS     
axi_burst_2        ERROR    RuntimeError: axi_burst seed 2 blew up
axi_burst_3        PASS     
axi_lite_1         PASS     
axi_lite_2         ERROR    RuntimeError: axi_lite seed 2 blew up
axi_lite_3         PASS     

jobs submitted  : 6
results          : 6
the assertion    : passed

  ERROR    2
  PASS     4

jobs lost        : 0
errors visible   : 2
and the sweep     : did not abort

from pathlib import Path

JOBS = [("axi_burst", 4121), ("axi_burst", 7734)]


def run_shared(job, root):
    """Specimen: the run directory is named from the TEST only."""
    test, seed = job
    d = Path(root) / test
    d.mkdir(parents=True, exist_ok=True)
    (d / "sim.log").write_text(f"UVM_INFO seed={seed}\nTEST DONE\n",
                               encoding="utf-8")
    return {"test": test, "seed": seed, "log": str((d / "sim.log").name),
            "dir": d.name}


def run_isolated(job, root):
    """The fix: the directory name includes the seed — 35.4 §6."""
    test, seed = job
    d = Path(root) / f"{test}_{seed}"
    d.mkdir(parents=True, exist_ok=True)
    (d / "sim.log").write_text(f"UVM_INFO seed={seed}\nTEST DONE\n",
                               encoding="utf-8")
    return {"test": test, "seed": seed, "log": str((d / "sim.log").name),
            "dir": d.name}


for label, fn, root in (("shared directory", run_shared, "a"),
                        ("one per job", run_isolated, "b")):
    Path(root).mkdir(exist_ok=True)
    results = [fn(j, root) for j in JOBS]
    dirs = sorted({r["dir"] for r in results})
    logs = sorted(p.relative_to(root).as_posix()
                  for p in Path(root).rglob("sim.log"))
    contents = {p.relative_to(root).as_posix():
                p.read_text(encoding="utf-8").splitlines()[0]
                for p in Path(root).rglob("sim.log")}
    print(f"--- {label}")
    print("    jobs run        :", len(results))
    print("    results         :", len(results))
    print("    run directories :", len(dirs), dirs)
    print("    sim.log files   :", len(logs))
    for name, first in sorted(contents.items()):
        print(f"      {name:<22} {first}")

print()
print("the shared case produced results:", 2)
print("and logs                        :", 1)
print("so the accounting assertion     :", "PASSES")
print("and the evidence for job 1      :", "overwritten by job 2")
print()
print("which is why accounting is      :", "necessary and not sufficient")

--- shared directory
    jobs run        : 2
    results         : 2
    run directories : 1 ['axi_burst']
    sim.log files   : 1
      axi_burst/sim.log      UVM_INFO seed=7734
--- one per job
    jobs run        : 2
    results         : 2
    run directories : 2 ['axi_burst_4121', 'axi_burst_7734']
    sim.log files   : 2
      axi_burst_4121/sim.log UVM_INFO seed=4121
      axi_burst_7734/sim.log UVM_INFO seed=7734

the shared case produced results: 2
and logs                        : 1
so the accounting assertion     : PASSES
and the evidence for job 1      : overwritten by job 2

which is why accounting is      : necessary and not sufficient

import json
from pathlib import Path

JOBS = ["axi_burst_4121", "axi_lite_7734"]


def seed_a_failed_run(root):
    for name in JOBS:
        d = Path(root) / name
        d.mkdir(parents=True, exist_ok=True)
        (d / "sim.log").write_text("UVM_ERROR drv.sv(88) x\nTEST DONE\n",
                                   encoding="utf-8")
        (d / "cmd.sh").write_text("#!/bin/sh\nvvp sim.vvp\n", encoding="utf-8")
    # the first job completed; the second was interrupted
    (Path(root) / JOBS[0] / "result.json").write_text(
        json.dumps({"status": "FAIL"}), encoding="utf-8")


def resume_clean_first(root, job):
    """Specimen: wipe the directory, then re-run."""
    import shutil
    d = Path(root) / job
    if d.exists():
        shutil.rmtree(d)
    d.mkdir(parents=True)
    return sorted(p.name for p in d.iterdir())


def resume_in_place(root, job):
    """The fix: keep the evidence, write the new result alongside."""
    d = Path(root) / job
    d.mkdir(parents=True, exist_ok=True)
    return sorted(p.name for p in d.iterdir())


for label, fn, root in (("resume wipes the dir", resume_clean_first, "ra"),
                        ("resume in place", resume_in_place, "rb")):
    Path(root).mkdir(exist_ok=True)
    seed_a_failed_run(root)
    before = sorted(p.name for p in (Path(root) / JOBS[1]).iterdir())
    after = fn(root, JOBS[1])
    print(f"--- {label}")
    print("    before resume :", before)
    print("    after resume  :", after)
    print("    log kept      :", "sim.log" in after)
    print("    cmd.sh kept   :", "cmd.sh" in after)

print()
print("the job being resumed is the one that:", "was interrupted")
print("so its log is                        :",
      "the evidence for why it was interrupted")
print()
print("wiping it before re-running makes     ")
print("  resume                              :", "worse than starting over")
print("because starting over would at least  :", "leave the old directory")

--- resume wipes the dir
    before resume : ['cmd.sh', 'sim.log']
    after resume  : []
    log kept      : False
    cmd.sh kept   : False
--- resume in place
    before resume : ['cmd.sh', 'sim.log']
    after resume  : ['cmd.sh', 'sim.log']
    log kept      : True
    cmd.sh kept   : True

the job being resumed is the one that: was interrupted
so its log is                        : the evidence for why it was interrupted

wiping it before re-running makes     
  resume                              : worse than starting over
because starting over would at least  : leave the old directory

import ast
from dataclasses import dataclass


@dataclass(frozen=True)
class Finding:
    area: str
    severity: str
    detail: str


RUNNERS = {"submit", "map"}


def review_regression(src):
    findings = []
    tree = ast.parse(src)
    text = src

    # --- accounting: is there an assertion comparing results with jobs?
    has_assert = any(
        isinstance(n, ast.Assert) and "len" in ast.dump(n.test)
        for n in ast.walk(tree))
    if not has_assert:
        findings.append(Finding("accounting", "blocker",
                                "no assertion compares len(results) with "
                                "len(jobs)"))

    # --- collection: an `if` or `try` between a future and its result
    if ".exception()" in text:
        findings.append(Finding("collection", "blocker",
                                "futures are filtered on .exception(): a "
                                "failed job is dropped, not recorded"))
    for n in ast.walk(tree):
        if isinstance(n, ast.Try):
            for h in n.handlers:
                if len(h.body) == 1 and isinstance(h.body[0], ast.Pass):
                    findings.append(Finding(
                        "collection", "blocker",
                        f"line {h.lineno}: except: pass around a worker "
                        f"result"))

    # --- isolation: does the run-dir name include the seed?
    for n in ast.walk(tree):
        if isinstance(n, ast.JoinedStr):
            rendered = "".join(
                v.value if isinstance(v, ast.Constant) else
                f"{{{getattr(v.value, 'id', '?')}}}"
                for v in n.values)
            if "run" in rendered or "/" in rendered:
                if "{seed}" not in rendered and "seed" not in rendered:
                    findings.append(Finding(
                        "isolation", "blocker",
                        f"run-dir name {rendered!r} does not include the seed"))

    # --- resume: rmtree before re-running
    if "rmtree" in text and "resume" in text:
        findings.append(Finding("resume", "blocker",
                                "rmtree in the resume path discards the "
                                "evidence of the interrupted run"))

    # --- shared mutable state in the worker
    for n in ast.walk(tree):
        if isinstance(n, ast.Call) and isinstance(n.func, ast.Attribute) \
           and n.func.attr == "append" \
           and isinstance(n.func.value, ast.Name) \
           and n.func.value.id in {"results", "rows", "out"}:
            findings.append(Finding("races", "major",
                                    f"line {n.lineno}: appending to a shared "
                                    f"list from a worker"))
    return findings


SPECIMEN = '''import shutil
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

results = []


def run_one(job, resume=False):
    test, seed = job
    d = Path(f"runs/{test}")
    if resume and d.exists():
        shutil.rmtree(d)
    d.mkdir(parents=True, exist_ok=True)
    results.append({"test": test, "seed": seed})
    return {"test": test, "seed": seed}


def run_suite(jobs):
    with ThreadPoolExecutor(max_workers=8) as pool:
        futures = [pool.submit(run_one, j) for j in jobs]
        out = [f.result() for f in futures if not f.exception()]
    return out
'''

findings = review_regression(SPECIMEN)
ORDER = {"blocker": 0, "major": 1, "minor": 2}

print("regression-system review")
for f in sorted(findings, key=lambda f: (ORDER[f.severity], f.area, f.detail)):
    print(f"  [{f.severity:<7}] {f.area:<12} {f.detail}")

print()
print("findings :", len(findings))
for sev in ("blocker", "major"):
    print(f"  {sev:<8} {sum(1 for f in findings if f.severity == sev)}")
print()
print("areas    :", sorted({f.area for f in findings}))
print("lines    :", len(SPECIMEN.strip().splitlines()))
print()
print("every one of the five properties")
print("  from §2 is represented        :",
      {"accounting", "collection", "isolation", "resume", "races"}
      <= {f.area for f in findings})

regression-system review
  [blocker] accounting   no assertion compares len(results) with len(jobs)
  [blocker] collection   futures are filtered on .exception(): a failed job is dropped, not recorded
  [blocker] isolation    run-dir name 'runs/{test}' does not include the seed
  [blocker] resume       rmtree in the resume path discards the evidence of the interrupted run
  [major  ] races        line 14: appending to a shared list from a worker

findings : 5
  blocker  4
  major    1

areas    : ['accounting', 'collection', 'isolation', 'races', 'resume']
lines    : 22

every one of the five properties
  from §2 is represented        : True

import json
import shutil
from pathlib import Path

JOBS = [f"t{i}" for i in range(4)]
ROOT = Path("acc")


def seed_two_already_done(root):
    """Two of the four jobs completed in an earlier pass."""
    shutil.rmtree(root, ignore_errors=True)
    root.mkdir(parents=True)
    for job in JOBS[:2]:
        (root / job).mkdir()
        (root / job / "result.json").write_text(
            json.dumps({"job": job, "status": "PASS"}), encoding="utf-8")


def _run(root, job):
    (root / job).mkdir(parents=True, exist_ok=True)
    (root / job / "result.json").write_text(
        json.dumps({"job": job, "status": "PASS"}), encoding="utf-8")
    return {"job": job, "status": "PASS"}


def collect_in_pass(jobs, root):
    """Return what THIS pass ran -- the natural implementation."""
    pending = [j for j in jobs if not (root / j / "result.json").is_file()]
    return [_run(root, j) for j in pending]


def load_from_disk(jobs, root):
    """Return EVERY job's result, whether this pass ran it or not."""
    pending = [j for j in jobs if not (root / j / "result.json").is_file()]
    for job in pending:
        _run(root, job)
    return [json.loads((root / j / "result.json").read_text(encoding="utf-8"))
            for j in jobs]


for label, fn in (("collect in-pass", collect_in_pass),
                  ("load from disk", load_from_disk)):
    seed_two_already_done(ROOT)
    results = fn(JOBS, ROOT)
    try:
        assert len(results) == len(JOBS)
        verdict = "passes"
    except AssertionError:
        verdict = "FAILS"
    print(f"{label:<16} jobs={len(JOBS)} results={len(results)}  "
          f"assertion {verdict}")

print()
print("jobs already complete  :", 2)
print("jobs this pass ran     :", 2)
print("anything actually wrong:", False)

collect in-pass  jobs=4 results=2  assertion FAILS
load from disk   jobs=4 results=4  assertion passes

jobs already complete  : 2
jobs this pass ran     : 2
anything actually wrong: False

VARIANTS = [
    ("len(results) == len(jobs)", "pool", "FALSE POSITIVE on a resume"),
    ("len(results) == len(ran)", "pool", "cannot ever fail"),
    ("len(results) == len(jobs)", "disk", "correct"),
    ("set(r['job'] for r in results) == set(jobs)", "disk",
     "correct, and names the missing job"),
]

print(f"{'the assertion':<44} {'results from':<13} verdict")
for assertion, source, verdict in VARIANTS:
    print(f"{assertion:<44} {source:<13} {verdict}")

print()
print("variants              :", len(VARIANTS))
print("that are correct      :",
      sum(1 for *_, v in VARIANTS if v.startswith("correct")))
print("that can never fail   :",
      sum(1 for *_, v in VARIANTS if "never" in v or "cannot ever" in v))
print()
print("`ran` is derived from :", "the collection being checked")
print("so variant 2 is       :", "a tautology in an assertion")
print()
print("variant 4 reports     :", "axi_wrap_9001 is missing")
print("variant 3 reports     :", "396 != 400")

the assertion                                results from  verdict
len(results) == len(jobs)                    pool          FALSE POSITIVE on a resume
len(results) == len(ran)                     pool          cannot ever fail
len(results) == len(jobs)                    disk          correct
set(r['job'] for r in results) == set(jobs)  disk          correct, and names the missing job

variants              : 4
that are correct      : 2
that can never fail   : 1

`ran` is derived from : the collection being checked
so variant 2 is       : a tautology in an assertion

variant 4 reports     : axi_wrap_9001 is missing
variant 3 reports     : 396 != 400

from concurrent.futures import ThreadPoolExecutor

JOBS = [1, 2, 3, 4]


def work(n):
    if n == 2:
        raise ValueError("two")
    return n * 10


# A
with ThreadPoolExecutor(max_workers=2) as pool:
    futures = [pool.submit(work, n) for n in JOBS]
    kept = [f.result() for f in futures if not f.exception()]
print("A:", len(kept), kept)

# B
raised = None
try:
    with ThreadPoolExecutor(max_workers=2) as pool:
        got = list(pool.map(work, JOBS))
except ValueError as exc:
    raised = str(exc)
print("B:", raised)

# C
d = {}
for n in JOBS:
    d[n % 2] = n
print("C:", len(JOBS), len(d))

# D
try:
    assert len(kept) == len(JOBS)
    print("D:", "passed")
except AssertionError:
    print("D:", f"{len(kept)} != {len(JOBS)}")

A: 3 [10, 30, 40]
B: two
C: 4 2
D: 3 != 4