diff --git a/wizard-work/null/deporder/EVIDENCE.txt b/wizard-work/null/deporder/EVIDENCE.txt new file mode 100644 index 0000000..05850b4 --- /dev/null +++ b/wizard-work/null/deporder/EVIDENCE.txt @@ -0,0 +1,179 @@ +=== deporder EVIDENCE === +Wed Oct 21 10:30:31 UTC 2026 +Python 3.14.7 + +=== 1. full suite: python3 test_deporder.py === +test_bad_spec_exit_4 (__main__.CliTests.test_bad_spec_exit_4) ... ok +test_bad_subcommand_exit_4 (__main__.CliTests.test_bad_subcommand_exit_4) ... ok +test_cycle_exit_2 (__main__.CliTests.test_cycle_exit_2) ... ok +test_missing_file_exit_4 (__main__.CliTests.test_missing_file_exit_4) ... ok +test_ok_exit_0 (__main__.CliTests.test_ok_exit_0) ... ok +test_unknown_dep_exit_3 (__main__.CliTests.test_unknown_dep_exit_3) ... ok +test_workers_flag_overrides_spec (__main__.CliTests.test_workers_flag_overrides_spec) ... ok +test_known_length (__main__.CriticalPathTests.test_known_length) ... ok +test_matches_reference (__main__.CriticalPathTests.test_matches_reference) ... ok +test_path_is_precedence_chained (__main__.CriticalPathTests.test_path_is_precedence_chained) ... ok +test_single_node (__main__.CriticalPathTests.test_single_node) ... ok +test_tie_break_is_lexicographic (__main__.CriticalPathTests.test_tie_break_is_lexicographic) ... ok +test_acyclic_has_no_cycle (__main__.GraphTests.test_acyclic_has_no_cycle) ... ok +test_cycle_error_message (__main__.GraphTests.test_cycle_error_message) ... ok +test_cycle_in_one_component_only (__main__.GraphTests.test_cycle_in_one_component_only) ... ok +test_self_loop (__main__.GraphTests.test_self_loop) ... ok +test_three_cycle_path_is_a_real_cycle (__main__.GraphTests.test_three_cycle_path_is_a_real_cycle) ... ok +test_two_cycle (__main__.GraphTests.test_two_cycle) ... ok +test_unknown_dep_reports_sorted_pairs (__main__.GraphTests.test_unknown_dep_reports_sorted_pairs) ... ok +test_chain (__main__.OrderTests.test_chain) ... ok +test_diamond_is_lexicographic (__main__.OrderTests.test_diamond_is_lexicographic) ... ok +test_disconnected_components_interleave (__main__.OrderTests.test_disconnected_components_interleave) ... ok +test_order_prefix_property_on_random_dags (__main__.OrderTests.test_order_prefix_property_on_random_dags) ... ok +test_20k_node_graph_under_budget (__main__.PerfTests.test_20k_node_graph_under_budget) ... +[perf] 20k nodes / 39987 edges: 0.92s (critical path 20000) +ok +test_key_insertion_order_does_not_change_output (__main__.PlanTests.test_key_insertion_order_does_not_change_output) ... ok +test_plan_fields (__main__.PlanTests.test_plan_fields) ... ok +test_plan_is_repeatable (__main__.PlanTests.test_plan_is_repeatable) ... ok +test_empty (__main__.ScheduleTests.test_empty) ... ok +test_schedule_invariants_on_random_dags (__main__.ScheduleTests.test_schedule_invariants_on_random_dags) ... ok +test_single_worker_equals_serial (__main__.ScheduleTests.test_single_worker_equals_serial) ... ok +test_two_workers_never_exceed_capacity (__main__.ScheduleTests.test_two_workers_never_exceed_capacity) ... ok +test_unbounded_workers_equals_critical_path (__main__.ScheduleTests.test_unbounded_workers_equals_critical_path) ... ok +test_defaults (__main__.SpecTests.test_defaults) ... ok +test_duplicate_deps_collapse (__main__.SpecTests.test_duplicate_deps_collapse) ... ok +test_est_coerced_to_float (__main__.SpecTests.test_est_coerced_to_float) ... ok +test_rejects_malformed (__main__.SpecTests.test_rejects_malformed) ... ok +test_workers_override (__main__.SpecTests.test_workers_override) ... ok +test_chain (__main__.WaveTests.test_chain) ... ok +test_diamond (__main__.WaveTests.test_diamond) ... ok +test_parallel_roots_and_skew (__main__.WaveTests.test_parallel_roots_and_skew) ... ok +test_single_node (__main__.WaveTests.test_single_node) ... ok +test_waves_match_reference_and_cover_every_task (__main__.WaveTests.test_waves_match_reference_and_cover_every_task) ... ok + +---------------------------------------------------------------------- +Ran 42 tests in 1.902s + +OK +SUITE_EXIT=0 + +=== 2. CLI on examples/pipeline.json === +{ + "critical_path": { + "length": 14.0, + "tasks": [ + "fetch_sources", + "compile_core", + "unit_tests", + "package" + ] + }, + "makespan": 14.0, + "order": [ + "fetch_sources", + "compile_core", + "compile_plugins", + "lint", + "unit_tests", + "package" + ], + "schedule": [ + { + "end": 2.0, + "start": 0.0, + "task": "fetch_sources", + "worker": 0 + }, + { + "end": 8.0, + "start": 2.0, + "task": "compile_core", + "worker": 1 + }, + { + "end": 5.0, + "start": 2.0, + "task": "compile_plugins", + "worker": 2 + }, + { + "end": 9.0, + "start": 8.0, + "task": "lint", + "worker": 0 + }, + { + "end": 12.0, + "start": 8.0, + "task": "unit_tests", + "worker": 2 + }, + { + "end": 14.0, + "start": 12.0, + "task": "package", + "worker": 1 + } + ], + "serial_time": 18.0, + "tasks": [ + "compile_core", + "compile_plugins", + "fetch_sources", + "lint", + "package", + "unit_tests" + ], + "waves": [ + [ + "fetch_sources" + ], + [ + "compile_core", + "compile_plugins" + ], + [ + "lint", + "unit_tests" + ], + [ + "package" + ] + ], + "workers": 3 +} +CLI_EXIT=0 + +=== 3. CLI on examples/cycle.json === +dependency cycle: draft -> review -> publish -> draft +CYCLE_EXIT=2 + +=== 4. CLI on examples/ghost.json === +unknown dependencies: app -> missing_module +GHOST_EXIT=3 + +=== 5. mutation harness: python3 mutate_teeth.py === +deporder mutation harness: 11 mutations +---------------------------------------------------------------------------------------------------- +waves: drop the +1 level bump CAUGHT exit 1 | FAILED (failures=5) + why: every task collapses into wave 0 +order: pop the heap tail instead of the min CAUGHT exit 1 | FAILED (failures=5) + why: lexicographic tie-break lost in Kahn +cycles: disable the residual-node check CAUGHT exit 1 | FAILED (failures=5) + why: cycle silently yields a partial order +cycle path: report a single node CAUGHT exit 1 | FAILED (failures=3) + why: reported 'cycle' is not a cycle +unknown deps: stop validating CAUGHT exit 1 | FAILED (failures=1, errors=1) + why: ghost dependency accepted +spec: allow est == 0 CAUGHT exit 1 | FAILED (failures=1) + why: zero-duration task accepted +critical path: break the predecessor tie-break CAUGHT exit 1 | FAILED (failures=1) + why: path no longer lexicographically canonical +schedule: release completions strictly before now CAUGHT exit 1 | FAILED (failures=1, errors=8) + why: worker never freed / clock stalls +schedule: double every task duration CAUGHT exit 1 | FAILED (failures=4) + why: makespan and critical-path bound violated +schedule: pin the clock to itself (kill the stall guard) CAUGHT exit 1 | FAILED (failures=1, errors=8) + why: guard fires: stalled scheduler must not spin +graph: drop the redundant set() in build_graph SURVIVED suite PASSED on mutant (not caught) + why: parse_spec already deduplicates; guard is redundant +---------------------------------------------------------------------------------------------------- +caught 10/11 mutants that must die; 0 problem(s) +MUT_EXIT=0 diff --git a/wizard-work/null/deporder/README.md b/wizard-work/null/deporder/README.md new file mode 100644 index 0000000..bf6ea5e --- /dev/null +++ b/wizard-work/null/deporder/README.md @@ -0,0 +1,89 @@ +# deporder — deterministic build-order planner for task DAGs + +Lane: Codeforge (hard coding / agentic reasoning). Author: null. +Location of record: `/root/.hermes/wizard-outputs/null/deporder/`. + +## What it is + +A stdlib-only Python module + CLI that turns a task/dependency spec (JSON) into +a build plan: + +* `topological_order` — serial order, lexicographically smallest ready task first +* `waves` — maximal parallelism batches (wave k = longest dependency chain has k edges) +* `critical_path` — weighted longest path (the makespan floor), with a unique, + lexicographically canonical path +* `schedule` — greedy list scheduling on N identical workers, with precedence and + capacity guarantees +* `find_cycle` — reports one cycle as a real path `x -> y -> ... -> x` + +Every ordering decision is deterministic: ties break on lexicographically smallest +name, so two runs over the same spec — and runs over specs whose keys were inserted +in a different order — produce byte-identical output. + +## CLI + +``` +python3 deporder.py plan examples/pipeline.json [--workers N] +``` + +Exit codes: `0` ok · `2` dependency cycle · `3` unknown dependency · `4` malformed spec. +Success prints the plan as JSON on stdout; failures print one diagnostic line on +stderr and nothing on stdout. + +## Run the tests + +``` +python3 test_deporder.py # 42 tests, stdlib unittest, no pytest needed +python3 mutate_teeth.py # mutation harness: do the tests bite? +``` + +## Evidence (measured on this box, Python 3.14.7) + +| Claim | How measured | Result | +|---|---|---| +| 42 tests pass | `python3 test_deporder.py` | `Ran 42 tests in 1.902s — OK` (exit 0) | +| Differential vs naive reference | 200 random DAGs × (order, waves, critical path) compared against O(V·E) fixpoint / O(V²) reference implementations inside the suite | all equal | +| Schedule invariants | 150 random DAGs × 1–4 workers: each task placed once, precedence `end[d] ≤ start[n]`, ≤ N tasks in flight (event sweep), per-worker non-overlap, `makespan ≥ critical path` | all hold | +| 1 worker == serial | `makespan == Σest` | exact | +| ≥N workers == critical path | `makespan == critical_path` | exact | +| Scale | 20 000 nodes / 39 987 edges: order + waves + critical path + 8-worker schedule | **0.92 s** (budget in-suite: 10 s) | +| Determinism | shuffled spec key insertion order → identical `json.dumps(sort_keys=True)` | identical | +| Tests have teeth | `python3 mutate_teeth.py`: 11 literal source mutations | **10/10 required mutants caught**, 1 documented equivalent, harness exit 0 | +| CLI contract | exit 0 on `examples/pipeline.json`; exit 2 + `dependency cycle: draft -> review -> publish -> draft`; exit 3 + `unknown dependencies: app -> missing_module` | as specified | + +### Bugs this suite actually caught while building it + +All three were found by running things, not by reading the code: + +1. **Scheduler stall (infinite loop).** The clock-advance step took + `min(free_time, completion_time)` even when a worker was already free at `now` + with nothing ready — `now` stopped moving and `schedule()` never returned. + Caught by `PlanTests` hanging. Fixed by advancing only to strictly-future + events and raising `RuntimeError` on stall (unreachable for a DAG). +2. **O(V²) critical path.** `min(n for n in dist if dist[n] == max(dist.values()))` + recomputed the max inside the generator: a 20k-node plan took **15.98 s**. + Hoisting `max()` gave **0.85 s** (≈19× faster) with identical results. +3. **Dead branch found by the mutation harness.** The first version of the fix + above also consulted future worker-release times when choosing the next event. + The mutation that deleted that branch *survived* — it was unreachable, since + workers are only ever released by completions already drained at `now`. The + branch was removed rather than left as decoration. + +## Mutation harness (`mutate_teeth.py`) + +Each mutation is a literal source edit applied to a copy of `deporder.py` in a +temp dir; the mutation is *caught* if the suite exits non-zero on the mutant. +Mutations tagged `equivalent` are expected to survive — they document a +genuinely redundant guard rather than a hole in the tests. + +See the harness output in `EVIDENCE.txt` (verbatim run log). + +## Files + +``` +deporder.py implementation (module + CLI) +test_deporder.py 42-test stdlib suite (unit + differential + CLI + perf) +mutate_teeth.py mutation harness +examples/pipeline.json worked example (makespan 14.0 vs 18.0 serial, 3 workers) +EVIDENCE.txt verbatim transcripts of the runs quoted above +``` diff --git a/wizard-work/null/deporder/deporder.py b/wizard-work/null/deporder/deporder.py new file mode 100644 index 0000000..40aa80d --- /dev/null +++ b/wizard-work/null/deporder/deporder.py @@ -0,0 +1,410 @@ +#!/usr/bin/env python3 +"""deporder -- deterministic build-order planner for task dependency DAGs. + +Stdlib only. Spec (JSON): + + { + "tasks": { + "compile": {"deps": [], "est": 3}, + "link": {"deps": ["compile"], "est": 1} + }, + "workers": 2 + } + +Semantics +--------- +* Edge set E = {(d, n) : d in tasks[n].deps} -- "d runs before n". +* `est` is a positive finite duration estimate; default 1.0. +* Every ordering decision is deterministic: ties are broken by + lexicographically smallest task name (heap order), so two runs over the + same spec -- and runs over specs whose keys are inserted in a different + order -- produce byte-identical output. + +API +--- +parse_spec(obj) -> (tasks, workers) +build_graph(tasks) -> (deps, rdeps) raises UnknownDepError +find_cycle(deps) -> cycle path or None +topological_order(deps) -> list[str] raises CycleError +waves(deps) -> list[list[str]] raises CycleError +critical_path(deps, est) -> (length, path) raises CycleError +schedule(deps, est, workers) -> list[dict] raises CycleError +plan(obj) -> dict full plan +main(argv) -> int CLI entry point + +CLI +--- + python3 deporder.py plan spec.json [--workers N] + +Exit codes: 0 ok, 2 dependency cycle, 3 unknown dependency, 4 malformed spec. +""" + +from __future__ import annotations + +import heapq +import json +import sys + +INF = float("inf") +__all__ = [ + "SpecError", + "UnknownDepError", + "CycleError", + "parse_spec", + "build_graph", + "find_cycle", + "topological_order", + "waves", + "critical_path", + "schedule", + "plan", + "main", +] + + +class SpecError(ValueError): + """Malformed spec (exit code 4).""" + + +class UnknownDepError(SpecError): + """A dependency names a task that does not exist (exit code 3).""" + + def __init__(self, pairs): + self.pairs = sorted(tuple(p) for p in pairs) + detail = ", ".join("%s -> %s" % (t, d) for t, d in self.pairs) + super().__init__("unknown dependencies: " + detail) + + +class CycleError(SpecError): + """The dependency graph is not a DAG (exit code 2).""" + + def __init__(self, cycle): + self.cycle = list(cycle) + super().__init__("dependency cycle: " + " -> ".join(self.cycle + [self.cycle[0]])) + + +# ---------------------------------------------------------------- spec parsing + + +def parse_spec(obj): + """Validate a decoded spec. Returns (tasks, workers).""" + if not isinstance(obj, dict): + raise SpecError("spec must be a JSON object") + raw_tasks = obj.get("tasks") + if not isinstance(raw_tasks, dict) or not raw_tasks: + raise SpecError("spec.tasks must be a non-empty object") + + tasks = {} + for name, body in raw_tasks.items(): + if not isinstance(name, str) or not name: + raise SpecError("task names must be non-empty strings") + if body is None: + body = {} + if not isinstance(body, dict): + raise SpecError("task %r: body must be an object" % (name,)) + + raw_deps = body.get("deps", []) + if not isinstance(raw_deps, list): + raise SpecError("task %r: deps must be a list" % (name,)) + deps = [] + for d in raw_deps: + if not isinstance(d, str) or not d: + raise SpecError("task %r: dependency names must be non-empty strings" % (name,)) + if d not in deps: + deps.append(d) + + est = body.get("est", 1.0) + if isinstance(est, bool) or not isinstance(est, (int, float)): + raise SpecError("task %r: est must be a number" % (name,)) + est = float(est) + if not (est > 0.0) or est == INF: + raise SpecError("task %r: est must be positive and finite" % (name,)) + tasks[name] = {"deps": deps, "est": est} + + workers = obj.get("workers", 1) + if isinstance(workers, bool) or not isinstance(workers, int) or workers < 1: + raise SpecError("workers must be a positive integer") + return tasks, workers + + +def build_graph(tasks): + """Return (deps, rdeps) with sorted, deduplicated adjacency.""" + names = sorted(tasks) + deps = {n: tuple(sorted(set(tasks[n]["deps"]))) for n in names} + unknown = [(n, d) for n in names for d in deps[n] if d not in tasks] + if unknown: + raise UnknownDepError(unknown) + rdeps = {n: [] for n in names} + for n in names: + for d in deps[n]: + rdeps[d].append(n) + for n in names: + rdeps[n].sort() + return deps, rdeps + + +# ------------------------------------------------------------------- cycle find + + +def find_cycle(deps): + """Return one cycle as a list of names in execution order, or None. + + Traverses the E = {(d, n)} direction, so the returned list [x, y, ...] + reads "x before y before ... before x". + """ + rdeps = {n: [m for m in deps if n in deps[m]] for n in deps} + for n in rdeps: + rdeps[n].sort() + + WHITE, GREY, BLACK = 0, 1, 2 + color = {n: WHITE for n in deps} + for root in sorted(deps): + if color[root] != WHITE: + continue + color[root] = GREY + path = [root] + stack = [(root, iter(rdeps[root]))] + while stack: + node, it = stack[-1] + descended = False + for nxt in it: + if color[nxt] == GREY: + return path[path.index(nxt):] + if color[nxt] == WHITE: + color[nxt] = GREY + path.append(nxt) + stack.append((nxt, iter(rdeps[nxt]))) + descended = True + break + if not descended: + color[node] = BLACK + stack.pop() + path.pop() + return None + + +# ------------------------------------------------------------------- orderings + + +def _kahn(deps, rdeps): + """Kahn's algorithm with a min-heap of ready tasks. Raises CycleError.""" + indeg = {n: len(deps[n]) for n in deps} + ready = [n for n in deps if indeg[n] == 0] + heapq.heapify(ready) + order = [] + while ready: + n = heapq.heappop(ready) + order.append(n) + for m in rdeps[n]: + indeg[m] -= 1 + if indeg[m] == 0: + heapq.heappush(ready, m) + if len(order) != len(deps): + raise CycleError(find_cycle(deps) or []) + return order + + +def topological_order(deps): + """A serial build order; lexicographically smallest ready task first.""" + _, rdeps = build_graph({n: {"deps": list(deps[n]), "est": 1.0} for n in deps}) + return _kahn(deps, rdeps) + + +def waves(deps): + """Parallel build waves. wave k holds every task whose longest + dependency chain has k edges. Each wave is sorted.""" + _, rdeps = build_graph({n: {"deps": list(deps[n]), "est": 1.0} for n in deps}) + order = _kahn(deps, rdeps) + level = {n: 0 for n in deps} + for n in order: + if deps[n]: + level[n] = 1 + max(level[d] for d in deps[n]) + depth = max(level.values()) + 1 + out = [[] for _ in range(depth)] + for n in sorted(level): + out[level[n]].append(n) + return out + + +def critical_path(deps, est): + """Longest weighted path. Returns (length, path). + + Ties (equal-length predecessors / equal-length ends) resolve to the + lexicographically smallest name, so the answer is unique. + """ + _, rdeps = build_graph({n: {"deps": list(deps[n]), "est": float(est[n])} for n in deps}) + order = _kahn(deps, rdeps) + dist = {} + pred = {} + for n in order: + best, best_d = 0.0, None + for d in deps[n]: + if best_d is None or dist[d] > best or (dist[d] == best and d < best_d): + best, best_d = dist[d], d + dist[n] = best + est[n] + pred[n] = best_d + if not dist: + return 0.0, [] + longest = max(dist.values()) + end = min(n for n in dist if dist[n] == longest) + path = [] + cur = end + while cur is not None: + path.append(cur) + cur = pred[cur] + path.reverse() + return dist[end], path + + +# ------------------------------------------------------------------- scheduling + + +def schedule(deps, est, workers): + """Greedy list scheduling on `workers` identical machines. + + Priority: lexicographically smallest ready task. A worker takes the + first ready task at or after the moment it becomes free. Returns a list + of dicts sorted by (start, name). + """ + _, rdeps = build_graph({n: {"deps": list(deps[n]), "est": float(est[n])} for n in deps}) + order = _kahn(deps, rdeps) + if not order: + return [] + + indeg = {n: len(deps[n]) for n in deps} + ready = [n for n in deps if indeg[n] == 0] + heapq.heapify(ready) + free = [(0.0, w) for w in range(workers)] + heapq.heapify(free) + running = [] # (end, name, worker) + out = [] + now = 0.0 + placed = 0 + + while placed < len(order): + while running and running[0][0] <= now: + end, name, w = heapq.heappop(running) + heapq.heappush(free, (end, w)) + for m in rdeps[name]: + indeg[m] -= 1 + if indeg[m] == 0: + heapq.heappush(ready, m) + while ready and free and free[0][0] <= now: + _, w = heapq.heappop(free) + name = heapq.heappop(ready) + end = now + est[name] + out.append({"task": name, "start": now, "end": end, "worker": w}) + heapq.heappush(running, (end, name, w)) + placed += 1 + if placed == len(order): + break + # Every completion at or before `now` has been drained, and workers are + # only released by those completions, so the next event is strictly in + # the future. A DAG always leaves a runnable task here; the guard turns + # any hypothetical scheduler bug into a loud failure instead of a spin. + nxt = running[0][0] if running else None + if nxt is None or nxt <= now: + raise RuntimeError("scheduler stalled with %d task(s) unplaced" % (len(order) - placed)) + now = nxt + + out.sort(key=lambda r: (r["start"], r["task"])) + return out + + +# ------------------------------------------------------------------------ plan + + +def plan(obj): + """Full plan dict for a decoded spec.""" + tasks, workers = parse_spec(obj) + deps, _ = build_graph(tasks) + est = {n: tasks[n]["est"] for n in deps} + order = topological_order(deps) + wv = waves(deps) + cp_len, cp_path = critical_path(deps, est) + sched = schedule(deps, est, workers) + return { + "tasks": sorted(deps), + "order": order, + "waves": wv, + "critical_path": {"tasks": cp_path, "length": cp_len}, + "workers": workers, + "makespan": max((r["end"] for r in sched), default=0.0), + "serial_time": sum(est.values()), + "schedule": sched, + } + + +# ------------------------------------------------------------------------- CLI + + +def main(argv=None): + argv = list(sys.argv[1:] if argv is None else argv) + if not argv or argv[0] in ("-h", "--help"): + sys.stderr.write("usage: deporder.py plan [--workers N]\n") + return 0 if argv else 4 + if argv[0] != "plan": + sys.stderr.write("unknown command %r\n" % (argv[0],)) + return 4 + + rest = argv[1:] + workers = None + path = None + i = 0 + while i < len(rest): + if rest[i] == "--workers": + if i + 1 >= len(rest): + sys.stderr.write("--workers needs a value\n") + return 4 + try: + workers = int(rest[i + 1]) + except ValueError: + sys.stderr.write("--workers must be an integer\n") + return 4 + i += 2 + elif path is None: + path = rest[i] + i += 1 + else: + sys.stderr.write("unexpected argument %r\n" % (rest[i],)) + return 4 + + if path is None: + sys.stderr.write("missing spec path\n") + return 4 + try: + with open(path, "r", encoding="utf-8") as fh: + obj = json.load(fh) + except OSError as exc: + sys.stderr.write("cannot read %s: %s\n" % (path, exc)) + return 4 + except ValueError as exc: + sys.stderr.write("invalid JSON in %s: %s\n" % (path, exc)) + return 4 + + if workers is not None: + if not isinstance(obj, dict): + sys.stderr.write("spec must be a JSON object\n") + return 4 + obj = dict(obj) + obj["workers"] = workers + + try: + result = plan(obj) + except UnknownDepError as exc: + sys.stderr.write(str(exc) + "\n") + return 3 + except CycleError as exc: + sys.stderr.write(str(exc) + "\n") + return 2 + except SpecError as exc: + sys.stderr.write(str(exc) + "\n") + return 4 + + json.dump(result, sys.stdout, indent=2, sort_keys=True) + sys.stdout.write("\n") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/wizard-work/null/deporder/examples/cycle.json b/wizard-work/null/deporder/examples/cycle.json new file mode 100644 index 0000000..404ff3d --- /dev/null +++ b/wizard-work/null/deporder/examples/cycle.json @@ -0,0 +1,7 @@ +{ + "tasks": { + "publish": {"deps": ["review"], "est": 1}, + "review": {"deps": ["draft"], "est": 2}, + "draft": {"deps": ["publish"], "est": 3} + } +} diff --git a/wizard-work/null/deporder/examples/ghost.json b/wizard-work/null/deporder/examples/ghost.json new file mode 100644 index 0000000..0312407 --- /dev/null +++ b/wizard-work/null/deporder/examples/ghost.json @@ -0,0 +1,5 @@ +{ + "tasks": { + "app": {"deps": ["missing_module"], "est": 1} + } +} diff --git a/wizard-work/null/deporder/examples/pipeline.json b/wizard-work/null/deporder/examples/pipeline.json new file mode 100644 index 0000000..c91f42e --- /dev/null +++ b/wizard-work/null/deporder/examples/pipeline.json @@ -0,0 +1,11 @@ +{ + "tasks": { + "fetch_sources": {"deps": [], "est": 2}, + "compile_core": {"deps": ["fetch_sources"], "est": 6}, + "compile_plugins": {"deps": ["fetch_sources"], "est": 3}, + "unit_tests": {"deps": ["compile_core"], "est": 4}, + "lint": {"deps": ["compile_core", "compile_plugins"], "est": 1}, + "package": {"deps": ["unit_tests", "lint"], "est": 2} + }, + "workers": 3 +} diff --git a/wizard-work/null/deporder/mutate_teeth.py b/wizard-work/null/deporder/mutate_teeth.py new file mode 100644 index 0000000..dd950d6 --- /dev/null +++ b/wizard-work/null/deporder/mutate_teeth.py @@ -0,0 +1,126 @@ +#!/usr/bin/env python3 +"""Mutation harness for deporder: does the test suite actually have teeth? + +Each mutation is a literal source edit to deporder.py, applied in an isolated +temp directory holding a copy of deporder.py + test_deporder.py. A mutation is +"caught" when the suite exits non-zero on the mutant. + +Exit code is 0 only when every mutation labelled expect="caught" is caught and +every mutation labelled expect="equivalent" survives (a documented redundant +guard, not a hole in the tests). + + python3 mutate_teeth.py +""" + +from __future__ import annotations + +import os +import shutil +import subprocess +import sys +import tempfile + +HERE = os.path.dirname(os.path.abspath(__file__)) +SRC = os.path.join(HERE, "deporder.py") +TESTS = os.path.join(HERE, "test_deporder.py") + +# (label, old, new, expectation, why) +MUTATIONS = [ + ("waves: drop the +1 level bump", + "level[n] = 1 + max(level[d] for d in deps[n])", + "level[n] = max(level[d] for d in deps[n])", + "caught", "every task collapses into wave 0"), + ("order: pop the heap tail instead of the min", + "n = heapq.heappop(ready)", + "n = ready.pop()", + "caught", "lexicographic tie-break lost in Kahn"), + ("cycles: disable the residual-node check", + "if len(order) != len(deps):", + "if False:", + "caught", "cycle silently yields a partial order"), + ("cycle path: report a single node", + "return path[path.index(nxt):]", + "return [nxt]", + "caught", "reported 'cycle' is not a cycle"), + ("unknown deps: stop validating", + "if unknown:", + "if False:", + "caught", "ghost dependency accepted"), + ("spec: allow est == 0", + "if not (est > 0.0) or est == INF:", + "if not (est >= 0.0) or est == INF:", + "caught", "zero-duration task accepted"), + ("critical path: break the predecessor tie-break", + "if best_d is None or dist[d] > best or (dist[d] == best and d < best_d):", + "if best_d is None or dist[d] >= best:", + "caught", "path no longer lexicographically canonical"), + ("schedule: release completions strictly before now", + "while running and running[0][0] <= now:", + "while running and running[0][0] < now:", + "caught", "worker never freed / clock stalls"), + ("schedule: double every task duration", + "end = now + est[name]", + "end = now + est[name] * 2", + "caught", "makespan and critical-path bound violated"), + ("schedule: pin the clock to itself (kill the stall guard)", + "nxt = running[0][0] if running else None", + "nxt = now", + "caught", "guard fires: stalled scheduler must not spin"), + ("graph: drop the redundant set() in build_graph", + 'deps = {n: tuple(sorted(set(tasks[n]["deps"]))) for n in names}', + 'deps = {n: tuple(sorted(tasks[n]["deps"])) for n in names}', + "equivalent", "parse_spec already deduplicates; guard is redundant"), +] + + +def run_mutant(label, old, new, timeout=300): + tmp = tempfile.mkdtemp(prefix="deporder_mut_") + try: + src = open(SRC, "r", encoding="utf-8").read() + hits = src.count(old) + if hits == 0: + return None, "ANCHOR MISSING" + mutated = src.replace(old, new) + with open(os.path.join(tmp, "deporder.py"), "w", encoding="utf-8") as fh: + fh.write(mutated) + shutil.copy(TESTS, os.path.join(tmp, "test_deporder.py")) + try: + proc = subprocess.run([sys.executable, "test_deporder.py"], cwd=tmp, + capture_output=True, text=True, timeout=timeout) + except subprocess.TimeoutExpired: + return True, "timeout/hang -> caught" + if proc.returncode == 0: + return False, "suite PASSED on mutant (not caught)" + tail = [ln for ln in proc.stderr.strip().splitlines() if ln.strip()] + detail = tail[-1] if tail else "nonzero exit" + return True, "exit %d | %s" % (proc.returncode, detail[:90]) + finally: + shutil.rmtree(tmp, ignore_errors=True) + + +def main(): + print("deporder mutation harness: %d mutations" % len(MUTATIONS)) + print("-" * 100) + bad = 0 + caught = 0 + for label, old, new, expect, why in MUTATIONS: + got, detail = run_mutant(label, old, new) + if got is None: + bad += 1 + status = "BROKEN (%s)" % detail + else: + ok = (got and expect == "caught") or (not got and expect == "equivalent") + bad += 0 if ok else 1 + caught += 1 if got else 0 + status = ("CAUGHT " if got else "SURVIVED ") + detail + if not ok: + status += " <-- UNEXPECTED (expect=%s)" % expect + print("%-46s %s" % (label, status)) + print("%-46s why: %s" % ("", why)) + print("-" * 100) + print("caught %d/%d mutants that must die; %d problem(s)" % (caught, len(MUTATIONS), bad)) + return 1 if bad else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/wizard-work/null/deporder/test_deporder.py b/wizard-work/null/deporder/test_deporder.py new file mode 100644 index 0000000..a5f333b --- /dev/null +++ b/wizard-work/null/deporder/test_deporder.py @@ -0,0 +1,471 @@ +#!/usr/bin/env python3 +"""Test suite for deporder. Stdlib only: python3 test_deporder.py + +Covers: spec validation, unknown-dependency reporting, cycle detection (with +cycle path), deterministic topological order, parallel waves, weighted +critical path, greedy multi-worker schedule invariants, differential testing +against naive O(V*E) reference implementations on 200 random DAGs, the CLI +contract (exit codes 0/2/3/4), and a 20k-node performance budget. +""" + +from __future__ import annotations + +import json +import os +import random +import subprocess +import sys +import tempfile +import time +import unittest + +HERE = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, HERE) + +import deporder as D # noqa: E402 + +DEPORDER = os.path.join(HERE, "deporder.py") +EPS = 1e-9 + + +# --------------------------------------------------------------- helpers + + +def tasks_from(deps, est=None): + est = est or {} + return {n: {"deps": list(deps[n]), "est": est.get(n, 1.0)} for n in deps} + + +def ref_waves(deps): + """O(V*E) fixpoint reference for wave assignment.""" + level = {n: 0 for n in deps} + changed = True + while changed: + changed = False + for n in sorted(deps): + want = 0 if not deps[n] else 1 + max(level[d] for d in deps[n]) + if want != level[n]: + level[n] = want + changed = True + buckets = {} + for n, lv in level.items(): + buckets.setdefault(lv, []).append(n) + return [sorted(buckets[k]) for k in sorted(buckets)] + + +def ref_order(deps): + """O(V^2) reference: repeatedly emit the smallest fully-ready task.""" + done = [] + left = set(deps) + while left: + cand = sorted(n for n in left if all(d in done for d in deps[n])) + if not cand: + raise AssertionError("reference hit a cycle") + done.append(cand[0]) + left.discard(cand[0]) + return done + + +def ref_dist(deps, est): + """Bellman-Ford style fixpoint for longest weighted path from any root.""" + dist = {n: float(est[n]) for n in deps} + changed = True + while changed: + changed = False + for n in sorted(deps): + want = float(est[n]) + max((dist[d] for d in deps[n]), default=0.0) + if want > dist[n]: + dist[n] = want + changed = True + return dist + + +def rand_dag(rng, n, max_deps=3): + names = ["n%03d" % i for i in range(n)] + deps = {} + for i, nm in enumerate(names): + pool = names[:i] + k = rng.randint(0, min(max_deps, len(pool))) + deps[nm] = sorted(rng.sample(pool, k)) + return deps + + +def check_schedule(tc, deps, est, workers, sched, makespan): + """Assert every scheduling invariant that must hold for any valid plan.""" + names = sorted(deps) + tc.assertEqual(sorted(r["task"] for r in sched), names, "each task scheduled exactly once") + by = {r["task"]: r for r in sched} + for r in sched: + tc.assertGreaterEqual(r["end"] - r["start"], float(est[r["task"]]) - EPS) + for n in names: + for d in deps[n]: + tc.assertLessEqual(by[d]["end"], by[n]["start"] + EPS, "precedence %s<%s" % (d, n)) + per_worker = {} + for r in sched: + per_worker.setdefault(r["worker"], []).append(r) + for w, rs in per_worker.items(): + rs.sort(key=lambda r: r["start"]) + for a, b in zip(rs, rs[1:]): + tc.assertLessEqual(a["end"], b["start"] + EPS, "worker %d overlap" % w) + events = [] + for r in sched: + events.append((r["start"], 1)) + events.append((r["end"], -1)) + events.sort() + live = 0 + for _, delta in events: + live += delta + tc.assertLessEqual(live, workers, "more than %d tasks in flight" % workers) + tc.assertAlmostEqual(makespan, max(r["end"] for r in sched), places=9) + cp_len, _ = D.critical_path(deps, {n: float(est[n]) for n in deps}) + tc.assertGreaterEqual(makespan, cp_len - EPS, "makespan below critical path") + + +def run_cli(args, spec=None): + path = None + tmpdir = None + if spec is not None: + tmpdir = tempfile.mkdtemp() + path = os.path.join(tmpdir, "spec.json") + with open(path, "w", encoding="utf-8") as fh: + fh.write(spec if isinstance(spec, str) else json.dumps(spec)) + cmd = [sys.executable, DEPORDER] + list(args) + ([path] if path else []) + proc = subprocess.run(cmd, capture_output=True, text=True) + if tmpdir: + try: + os.remove(path) + os.rmdir(tmpdir) + except OSError: + pass + return proc + + +# ------------------------------------------------------------------ tests + + +class SpecTests(unittest.TestCase): + def test_defaults(self): + tasks, workers = D.parse_spec({"tasks": {"a": {}, "b": {"deps": ["a"]}}}) + self.assertEqual(workers, 1) + self.assertEqual(tasks["a"], {"deps": [], "est": 1.0}) + self.assertEqual(tasks["b"], {"deps": ["a"], "est": 1.0}) + + def test_workers_override(self): + _, workers = D.parse_spec({"tasks": {"a": {}}, "workers": 3}) + self.assertEqual(workers, 3) + + def test_duplicate_deps_collapse(self): + tasks, _ = D.parse_spec({"tasks": {"a": {}, "b": {"deps": ["a", "a", "a"]}}}) + self.assertEqual(tasks["b"]["deps"], ["a"]) + + def test_est_coerced_to_float(self): + tasks, _ = D.parse_spec({"tasks": {"a": {"est": 2}}}) + self.assertIsInstance(tasks["a"]["est"], float) + + def test_rejects_malformed(self): + bad = [ + None, [], "x", {}, + {"tasks": []}, + {"tasks": {}}, + {"tasks": {"a": []}}, + {"tasks": {"a": {"deps": "b"}}}, + {"tasks": {"a": {"deps": [1]}}}, + {"tasks": {"a": {"deps": [""]}}}, + {"tasks": {"a": {"est": 0}}}, + {"tasks": {"a": {"est": -1}}}, + {"tasks": {"a": {"est": True}}}, + {"tasks": {"a": {"est": "1"}}}, + {"tasks": {"a": {"est": float("inf")}}}, + {"tasks": {"a": {}}, "workers": 0}, + {"tasks": {"a": {}}, "workers": -2}, + {"tasks": {"a": {}}, "workers": 2.5}, + {"tasks": {"a": {}}, "workers": True}, + ] + for spec in bad: + with self.assertRaises(D.SpecError, msg="should reject %r" % (spec,)): + D.parse_spec(spec) + + +class GraphTests(unittest.TestCase): + def test_unknown_dep_reports_sorted_pairs(self): + spec = {"tasks": {"b": {"deps": ["zz"]}, "a": {"deps": ["yy"]}}} + tasks, _ = D.parse_spec(spec) + with self.assertRaises(D.UnknownDepError) as ctx: + D.build_graph(tasks) + self.assertEqual(ctx.exception.pairs, [("a", "yy"), ("b", "zz")]) + self.assertIn("a -> yy", str(ctx.exception)) + + def test_acyclic_has_no_cycle(self): + deps = {"a": [], "b": ["a"], "c": ["a", "b"]} + self.assertIsNone(D.find_cycle(deps)) + + def test_self_loop(self): + deps = {"a": ["a"]} + self.assertEqual(D.find_cycle(deps), ["a"]) + with self.assertRaises(D.CycleError) as ctx: + D.topological_order(deps) + self.assertEqual(ctx.exception.cycle, ["a"]) + + def test_two_cycle(self): + deps = {"a": ["b"], "b": ["a"]} + cycle = D.find_cycle(deps) + self.assertEqual(len(cycle), 2) + self.assertEqual(set(cycle), {"a", "b"}) + self.assertIn(cycle[0], deps[cycle[1]]) + + def test_three_cycle_path_is_a_real_cycle(self): + deps = {"a": ["c"], "b": ["a"], "c": ["b"]} + cycle = D.find_cycle(deps) + self.assertEqual(len(cycle), 3) + for i, n in enumerate(cycle): + self.assertIn(cycle[i - 1], deps[n], "edge %s -> %s missing" % (cycle[i - 1], n)) + with self.assertRaises(D.CycleError): + D.waves(deps) + + def test_cycle_in_one_component_only(self): + deps = {"a": [], "b": ["a"], "x": ["y"], "y": ["x"]} + self.assertEqual(set(D.find_cycle(deps)), {"x", "y"}) + with self.assertRaises(D.CycleError): + D.topological_order(deps) + + def test_cycle_error_message(self): + deps = {"a": ["b"], "b": ["a"]} + with self.assertRaises(D.CycleError) as ctx: + D.critical_path(deps, {"a": 1.0, "b": 1.0}) + msg = str(ctx.exception) + self.assertTrue(msg.startswith("dependency cycle: "), msg) + head, tail = msg.split(": ", 1) + parts = tail.split(" -> ") + self.assertEqual(parts[0], parts[-1], "cycle path must return to its start") + + +class OrderTests(unittest.TestCase): + def test_diamond_is_lexicographic(self): + deps = {"d": ["b", "c"], "c": ["a"], "b": ["a"], "a": []} + self.assertEqual(D.topological_order(deps), ["a", "b", "c", "d"]) + + def test_chain(self): + deps = {"c": ["b"], "b": ["a"], "a": []} + self.assertEqual(D.topological_order(deps), ["a", "b", "c"]) + + def test_disconnected_components_interleave(self): + deps = {"a1": [], "b1": [], "a2": ["a1"], "b2": ["b1"]} + self.assertEqual(D.topological_order(deps), ["a1", "a2", "b1", "b2"]) + self.assertEqual(D.topological_order(deps), ref_order(deps)) + + def test_order_prefix_property_on_random_dags(self): + rng = random.Random(20261021) + for _ in range(200): + deps = rand_dag(rng, rng.randint(1, 12)) + order = D.topological_order(deps) + seen = set() + for n in order: + self.assertTrue(set(deps[n]) <= seen, "dep of %s emitted late" % n) + seen.add(n) + self.assertEqual(order, ref_order(deps)) + + +class WaveTests(unittest.TestCase): + def test_diamond(self): + deps = {"d": ["b", "c"], "c": ["a"], "b": ["a"], "a": []} + self.assertEqual(D.waves(deps), [["a"], ["b", "c"], ["d"]]) + + def test_chain(self): + deps = {"c": ["b"], "b": ["a"], "a": []} + self.assertEqual(D.waves(deps), [["a"], ["b"], ["c"]]) + + def test_parallel_roots_and_skew(self): + deps = {"a": [], "b": [], "c": ["a"], "d": ["c"], "e": ["a", "b"]} + self.assertEqual(D.waves(deps), [["a", "b"], ["c", "e"], ["d"]]) + + def test_single_node(self): + self.assertEqual(D.waves({"solo": []}), [["solo"]]) + + def test_waves_match_reference_and_cover_every_task(self): + rng = random.Random(7) + for _ in range(200): + deps = rand_dag(rng, rng.randint(1, 14)) + wv = D.waves(deps) + self.assertEqual(wv, ref_waves(deps)) + self.assertEqual(sorted(x for w in wv for x in w), sorted(deps)) + for w in wv: + self.assertEqual(w, sorted(w)) + + +class CriticalPathTests(unittest.TestCase): + def test_known_length(self): + deps = {"a": [], "b": ["a"], "c": ["a"], "d": ["b", "c"]} + est = {"a": 3.0, "b": 1.0, "c": 4.0, "d": 2.0} + length, path = D.critical_path(deps, est) + self.assertAlmostEqual(length, 3.0 + 4.0 + 2.0, places=9) + self.assertEqual(path, ["a", "c", "d"]) + self.assertAlmostEqual(sum(est[n] for n in path), length, places=9) + + def test_path_is_precedence_chained(self): + deps = {"a": [], "b": ["a"], "c": ["a", "b"], "d": ["b"]} + est = {"a": 1.0, "b": 1.0, "c": 5.0, "d": 1.0} + _, path = D.critical_path(deps, est) + for prev, nxt in zip(path, path[1:]): + self.assertIn(prev, deps[nxt]) + + def test_tie_break_is_lexicographic(self): + deps = {"a": [], "b": [], "z": ["a", "b"]} + est = {"a": 2.0, "b": 2.0, "z": 1.0} + length, path = D.critical_path(deps, est) + self.assertAlmostEqual(length, 3.0, places=9) + self.assertEqual(path, ["a", "z"]) + + def test_single_node(self): + self.assertEqual(D.critical_path({"a": []}, {"a": 2.5}), (2.5, ["a"])) + + def test_matches_reference(self): + rng = random.Random(99) + for _ in range(200): + deps = rand_dag(rng, rng.randint(1, 14)) + est = {n: float(rng.choice([0.5, 1, 2, 3, 4])) for n in deps} + length, path = D.critical_path(deps, est) + self.assertAlmostEqual(length, max(ref_dist(deps, est).values()), places=9) + self.assertAlmostEqual(sum(est[n] for n in path), length, places=9) + + +class ScheduleTests(unittest.TestCase): + DEPS = {"a": [], "b": ["a"], "c": ["a"], "d": ["b", "c"], "e": ["d"]} + EST = {"a": 3.0, "b": 1.0, "c": 4.0, "d": 2.0, "e": 1.0} + + def test_single_worker_equals_serial(self): + sched = D.schedule(self.DEPS, self.EST, 1) + check_schedule(self, self.DEPS, self.EST, 1, sched, max(r["end"] for r in sched)) + self.assertAlmostEqual(max(r["end"] for r in sched), sum(self.EST.values()), places=9) + + def test_unbounded_workers_equals_critical_path(self): + w = len(self.DEPS) + sched = D.schedule(self.DEPS, self.EST, w) + check_schedule(self, self.DEPS, self.EST, w, sched, max(r["end"] for r in sched)) + cp, _ = D.critical_path(self.DEPS, self.EST) + self.assertAlmostEqual(max(r["end"] for r in sched), cp, places=9) + + def test_two_workers_never_exceed_capacity(self): + sched = D.schedule(self.DEPS, self.EST, 2) + check_schedule(self, self.DEPS, self.EST, 2, sched, max(r["end"] for r in sched)) + self.assertAlmostEqual(max(r["end"] for r in sched), 10.0, places=9) + + def test_empty(self): + self.assertEqual(D.schedule({}, {}, 3), []) + + def test_schedule_invariants_on_random_dags(self): + rng = random.Random(4242) + for _ in range(150): + deps = rand_dag(rng, rng.randint(1, 12)) + est = {n: float(rng.choice([1, 2, 3, 5])) for n in deps} + workers = rng.randint(1, 4) + sched = D.schedule(deps, est, workers) + check_schedule(self, deps, est, workers, sched, max(r["end"] for r in sched)) + + +class PlanTests(unittest.TestCase): + SPEC = { + "tasks": {"a": {"est": 3}, "b": {"deps": ["a"], "est": 1}, + "c": {"deps": ["a"], "est": 4}, "d": {"deps": ["b", "c"], "est": 2}}, + "workers": 2, + } + + def test_plan_fields(self): + p = D.plan(self.SPEC) + self.assertEqual(p["order"], ["a", "b", "c", "d"]) + self.assertEqual(p["waves"], [["a"], ["b", "c"], ["d"]]) + self.assertEqual(p["critical_path"]["tasks"], ["a", "c", "d"]) + self.assertAlmostEqual(p["critical_path"]["length"], 9.0, places=9) + self.assertAlmostEqual(p["serial_time"], 10.0, places=9) + self.assertAlmostEqual(p["makespan"], 9.0, places=9) + + def test_key_insertion_order_does_not_change_output(self): + base = D.plan(self.SPEC) + shuffled = dict(reversed(list(self.SPEC["tasks"].items()))) + alt = D.plan({"tasks": shuffled, "workers": 2}) + self.assertEqual(json.dumps(base, sort_keys=True), json.dumps(alt, sort_keys=True)) + + def test_plan_is_repeatable(self): + a = json.dumps(D.plan(self.SPEC), sort_keys=True) + b = json.dumps(D.plan(self.SPEC), sort_keys=True) + self.assertEqual(a, b) + + +class CliTests(unittest.TestCase): + def test_ok_exit_0(self): + proc = run_cli(["plan"], PlanTests.SPEC) + self.assertEqual(proc.returncode, 0, proc.stderr) + out = json.loads(proc.stdout) + self.assertEqual(out["order"], ["a", "b", "c", "d"]) + self.assertEqual(out["workers"], 2) + + def test_workers_flag_overrides_spec(self): + spec = {"tasks": {"a": {}, "b": {}}} + proc = run_cli(["plan", "--workers", "2"], spec) + self.assertEqual(proc.returncode, 0, proc.stderr) + self.assertEqual(json.loads(proc.stdout)["workers"], 2) + + def test_cycle_exit_2(self): + spec = {"tasks": {"a": {"deps": ["b"]}, "b": {"deps": ["a"]}}} + proc = run_cli(["plan"], spec) + self.assertEqual(proc.returncode, 2, proc.stdout) + self.assertIn("dependency cycle", proc.stderr) + self.assertEqual(proc.stdout.strip(), "") + + def test_unknown_dep_exit_3(self): + proc = run_cli(["plan"], {"tasks": {"a": {"deps": ["ghost"]}}}) + self.assertEqual(proc.returncode, 3) + self.assertIn("unknown dependencies: a -> ghost", proc.stderr) + + def test_bad_spec_exit_4(self): + proc = run_cli(["plan"], {"tasks": {"a": {"est": -1}}}) + self.assertEqual(proc.returncode, 4) + proc = run_cli(["plan"], "{not json") + self.assertEqual(proc.returncode, 4) + self.assertIn("invalid JSON", proc.stderr) + + def test_missing_file_exit_4(self): + proc = subprocess.run([sys.executable, DEPORDER, "plan", "/nonexistent/spec.json"], + capture_output=True, text=True) + self.assertEqual(proc.returncode, 4) + + def test_bad_subcommand_exit_4(self): + proc = run_cli(["frobnicate"], {"tasks": {"a": {}}}) + self.assertEqual(proc.returncode, 4) + + +class PerfTests(unittest.TestCase): + def test_20k_node_graph_under_budget(self): + n = 20000 + deps = {} + est = {} + for i in range(n): + nm = "t%05d" % i + deps[nm] = ["t%05d" % (i - 1)] if i else [] + est[nm] = 1.0 + rng = random.Random(1) + names = sorted(deps) + for _ in range(n): + j = rng.randrange(1, n) + i = rng.randrange(0, j) + deps[names[j]].append(names[i]) + for nm in deps: + deps[nm] = sorted(set(deps[nm])) + + t0 = time.perf_counter() + order = D.topological_order(deps) + wv = D.waves(deps) + cp_len, cp_path = D.critical_path(deps, est) + sched = D.schedule(deps, est, 8) + elapsed = time.perf_counter() - t0 + + self.assertEqual(len(order), n) + self.assertEqual(sorted(x for w in wv for x in w), names) + self.assertAlmostEqual(cp_len, float(len(cp_path)), places=9) + self.assertEqual(len(sched), n) + self.assertLess(elapsed, 10.0, "20k-node plan took %.2fs" % elapsed) + sys.stderr.write("\n[perf] 20k nodes / %d edges: %.2fs (critical path %.0f)\n" + % (sum(len(v) for v in deps.values()), elapsed, cp_len)) + + +if __name__ == "__main__": + unittest.main(verbosity=2)