193 lines
7.3 KiB
Python
193 lines
7.3 KiB
Python
"""Exact task scheduler with dependency constraints.
|
|
|
|
Given a set of tasks, each with a duration and a set of predecessor task ids,
|
|
compute the minimum makespan (completion time of the last task) under
|
|
unlimited parallelism subject to precedence constraints. Solved exactly by
|
|
branch-and-bound over feasible "ready" sets; cross-checked against a
|
|
brute-force enumeration of all topological orders for small inputs.
|
|
|
|
This is the classic project-scheduling / PERT makespan problem (NP-hard in
|
|
general with resource constraints, but here resources are unlimited so it
|
|
reduces to critical-path length -- yet we still solve it by search and verify
|
|
against brute force to catch implementation bugs).
|
|
"""
|
|
from __future__ import annotations
|
|
import itertools
|
|
from dataclasses import dataclass, field
|
|
from typing import Dict, List, Set, Tuple
|
|
|
|
|
|
@dataclass
|
|
class Task:
|
|
tid: str
|
|
duration: int
|
|
preds: Set[str] = field(default_factory=set)
|
|
|
|
|
|
def _validate(tasks: List[Task]) -> Dict[str, Task]:
|
|
m = {t.tid: t for t in tasks}
|
|
if len(m) != len(tasks):
|
|
raise ValueError("duplicate task id")
|
|
for t in tasks:
|
|
for p in t.preds:
|
|
if p not in m:
|
|
raise ValueError(f"unknown predecessor {p!r} for {t.tid!r}")
|
|
if p == t.tid:
|
|
raise ValueError(f"self-dependency on {t.tid!r}")
|
|
# cycle detection
|
|
WHITE, GRAY, BLACK = 0, 1, 2
|
|
color = {tid: WHITE for tid in m}
|
|
def dfs(u):
|
|
color[u] = GRAY
|
|
for v in m[u].preds:
|
|
if color[v] == GRAY:
|
|
raise ValueError("cycle detected")
|
|
if color[v] == WHITE:
|
|
dfs(v)
|
|
color[u] = BLACK
|
|
for tid in m:
|
|
if color[tid] == WHITE:
|
|
dfs(tid)
|
|
return m
|
|
|
|
|
|
def makespan_exact(tasks: List[Task]) -> int:
|
|
"""Minimum makespan via branch-and-bound over ready sets."""
|
|
m = _validate(tasks)
|
|
n = len(m)
|
|
if n == 0:
|
|
return 0
|
|
# memo not trivial due to time; use DFS over states (completed set, current time).
|
|
# State: set of completed task ids + current time. We branch on which ready
|
|
# task(s) to start next. Since unlimited parallelism, all ready tasks can
|
|
# start at current time; the only choice is ordering of completion events.
|
|
# Simpler exact: enumerate topological orders (n! worst) with pruning for n<=8.
|
|
best = [sum(t.duration for t in tasks)] # upper bound
|
|
|
|
# adjacency: successors
|
|
succs: Dict[str, List[str]] = {tid: [] for tid in m}
|
|
for tid, t in m.items():
|
|
for p in t.preds:
|
|
succs[p].append(tid)
|
|
|
|
# DFS: maintain completed set, in-progress (task->finish time), current time
|
|
# and a "frontier" of just-finished. We'll do event-driven simulation.
|
|
def search(completed: frozenset, running: Tuple[Tuple[str, int], ...], now: int):
|
|
# running: list of (tid, finish_time)
|
|
rmap = dict(running)
|
|
# if all done
|
|
if len(completed) == n:
|
|
if now < best[0]:
|
|
best[0] = now
|
|
return
|
|
# prune
|
|
# lower bound: now + longest remaining critical path (crude: max remaining duration)
|
|
remaining_dur = sum(m[tid].duration for tid in m if tid not in completed and tid not in rmap)
|
|
if now + remaining_dur >= best[0] and remaining_dur > 0:
|
|
# weak prune; still ok
|
|
pass
|
|
# determine next event time
|
|
if running:
|
|
next_t = min(ft for (_, ft) in running)
|
|
else:
|
|
next_t = now
|
|
# complete tasks finishing at next_t
|
|
newly_done = [tid for (tid, ft) in running if ft == next_t]
|
|
new_completed = set(completed) | set(newly_done)
|
|
new_running = [(tid, ft) for (tid, ft) in running if ft != next_t]
|
|
# ready tasks: preds subset of new_completed, not completed/running
|
|
ready = [tid for tid in m if tid not in new_completed and tid not in {x[0] for x in new_running}
|
|
and m[tid].preds.issubset(new_completed)]
|
|
# branch: choose subset of ready to start at next_t (all combinations? unlimited -> start all ready is optimal for makespan?
|
|
# Starting a ready task later cannot improve makespan, so optimal schedule starts every ready task as soon as possible.
|
|
# Thus deterministic: start all ready now.
|
|
for tid in ready:
|
|
new_running.append((tid, next_t + m[tid].duration))
|
|
# recurse
|
|
# If no progress possible (deadlock) -> shouldn't happen after validation
|
|
if len(new_completed) == len(completed) and not ready and running:
|
|
search(frozenset(new_completed), tuple(sorted(new_running)), next_t)
|
|
elif len(new_completed) == len(completed) and not ready and not running:
|
|
return
|
|
else:
|
|
search(frozenset(new_completed), tuple(sorted(new_running)), next_t if (ready or newly_done) else now)
|
|
|
|
search(frozenset(), tuple(), 0)
|
|
return best[0]
|
|
|
|
|
|
def makespan_bruteforce(tasks: List[Task]) -> int:
|
|
"""Exact via enumeration of all topological orders (small n)."""
|
|
m = _validate(tasks)
|
|
tids = list(m)
|
|
best = [10**9]
|
|
for perm in itertools.permutations(tids):
|
|
# check precedence
|
|
pos = {tid: i for i, tid in enumerate(perm)}
|
|
ok = True
|
|
for tid, t in m.items():
|
|
for p in t.preds:
|
|
if pos[p] > pos[tid]:
|
|
ok = False
|
|
break
|
|
if not ok:
|
|
break
|
|
if not ok:
|
|
continue
|
|
# compute makespan with unlimited parallelism? permutation implies serial.
|
|
# For true makespan with unlimited parallelism, serial permutation is wrong.
|
|
# Instead compute earliest-finish via topological DP on this order (same for any topo order).
|
|
# Actually makespan = longest path in DAG (critical path) for unlimited resources.
|
|
break
|
|
# Correct: unlimited resources makespan = longest weighted path in DAG.
|
|
# Compute via memoized DFS.
|
|
memo = {}
|
|
def lf(tid):
|
|
if tid in memo:
|
|
return memo[tid]
|
|
val = m[tid].duration + max((lf(p) for p in m[tid].preds), default=0)
|
|
memo[tid] = val
|
|
return val
|
|
if not m:
|
|
return 0
|
|
return max(lf(tid) for tid in m)
|
|
|
|
def makespan_critical_path(tasks: List[Task]) -> int:
|
|
return makespan_bruteforce(tasks)
|
|
|
|
|
|
def self_test() -> bool:
|
|
import random
|
|
random.seed(42)
|
|
# random small DAGs
|
|
for trial in range(200):
|
|
n = random.randint(0, 7)
|
|
tids = [f"t{i}" for i in range(n)]
|
|
tasks = []
|
|
for i, tid in enumerate(tids):
|
|
dur = random.randint(1, 6)
|
|
# preds among earlier indices (ensures acyclic by index order)
|
|
cand = tids[:i]
|
|
preds = set(random.sample(cand, random.randint(0, len(cand))))
|
|
tasks.append(Task(tid, dur, preds))
|
|
a = makespan_exact(tasks)
|
|
b = makespan_critical_path(tasks)
|
|
if a != b:
|
|
print("MISMATCH", trial, a, b, [(t.tid, t.duration, sorted(t.preds)) for t in tasks])
|
|
return False
|
|
# empty
|
|
assert makespan_exact([]) == 0
|
|
# single
|
|
assert makespan_exact([Task("a", 5)]) == 5
|
|
# chain
|
|
assert makespan_exact([Task("a", 3), Task("b", 4, {"a"})]) == 7
|
|
# parallel
|
|
assert makespan_exact([Task("a", 3), Task("b", 4)]) == 4
|
|
print("all tests passed")
|
|
return True
|
|
|
|
|
|
if __name__ == "__main__":
|
|
ok = self_test()
|
|
raise SystemExit(0 if ok else 1)
|