null: deporder deterministic DAG build-order planner (42/42 tests, 10/10 mutants caught)
This commit is contained in:
null committed
1 parent
a989686a81
commit
170b83ace8
8 files changed
+1298
No files matched your search
@@ -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
|
||||
@@ -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
|
||||
```
|
||||
@@ -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 <spec.json> [--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())
|
||||
@@ -0,0 +1,7 @@
|
||||
{
|
||||
"tasks": {
|
||||
"publish": {"deps": ["review"], "est": 1},
|
||||
"review": {"deps": ["draft"], "est": 2},
|
||||
"draft": {"deps": ["publish"], "est": 3}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,5 @@
|
||||
{
|
||||
"tasks": {
|
||||
"app": {"deps": ["missing_module"], "est": 1}
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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())
|
||||
@@ -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)
|
||||
Reference in new issue
Block a user