diff --git a/submissions/dharamendrak.py b/submissions/dharamendrak.py new file mode 100644 index 0000000..dbe8065 --- /dev/null +++ b/submissions/dharamendrak.py @@ -0,0 +1,208 @@ +"""Build graph scheduler — parallel implementation for Python 3.14t free-threading. + +Design (in order of impact): + +- Custom workers on `queue.SimpleQueue` — leaner than ThreadPoolExecutor. +- Chain fast-path: pure linear DAG skips threading entirely. +- `results[name] = build(...)` runs OUTSIDE the lock. Dict writes are + internally synchronized in 3.14t; the subsequent lock acquire publishes + the write to other workers via the memory barrier. +- Inline-execution: when a target finishes and one+ children become ready, + the worker continues with the first directly instead of round-tripping + through the queue (saves ~5us per chain link). +- Heap-based work priority (LPT) kicks in only when both max_fan_in and + max_fan_out exceed worker count — e.g., diamond — so we don't pay heap + overhead on graphs where FIFO already keeps every core busy. +- Single lock; main thread also acts as a worker. +""" + +from __future__ import annotations + +import heapq +import queue +import threading + +from graph import BuildGraph + + +def build_all(graph: BuildGraph) -> dict[str, bytes]: + targets = graph.targets + total = len(targets) + if total == 0: + return {} + + results: dict[str, bytes] = {} + empty_deps: dict[str, bytes] = {} + + # Fast chain detection: single pass, early exit. Avoid building the full + # dependents dict for chain graphs (saves ~10ms on the 5k-target chain). + chain_successor: dict[str, str] = {} + chain_root: str | None = None + is_chain = True + for name, target in targets.items(): + deps = target.deps + if len(deps) == 0: + if chain_root is not None: + is_chain = False + break + chain_root = name + elif len(deps) == 1: + dep_name = deps[0].name + if dep_name in chain_successor: + is_chain = False + break + chain_successor[dep_name] = name + else: + is_chain = False + break + + if is_chain and chain_root is not None: + name = chain_root + while True: + target = targets[name] + dep_results = ( + {d.name: results[d.name] for d in target.deps} + if target.deps else empty_deps + ) + results[name] = target.build(dep_results) + next_name = chain_successor.get(name) + if next_name is None: + return results + name = next_name + + # Not a chain: build full structures for parallel scheduling. + remaining = {name: len(t.deps) for name, t in targets.items()} + dependents: dict[str, list[str]] = {name: [] for name in targets} + for name, target in targets.items(): + for dep in target.deps: + dependents[dep.name].append(name) + roots = [name for name, count in remaining.items() if count == 0] + + # Match the eval server's core count. Slight oversubscription on smaller + # machines is fine — CPU-bound tasks queue up cleanly under free-threading. + num_workers = min(24, total) + ready: queue.SimpleQueue = queue.SimpleQueue() + lock = threading.Lock() + pending = total + + max_fan_in = max(remaining.values(), default=0) + max_fan_out = max((len(v) for v in dependents.values()), default=0) + # Use priority only on graphs with extremely wide fan-in/out (e.g. diamond). + # Threshold is fixed at 24 (eval server core count) rather than scaling with + # local cpu_count — priority overhead isn't worth it for moderate fan-out. + use_priority = max_fan_in > 24 and max_fan_out > 24 + + if use_priority: + # LPT scheduling: prefer heavier targets so the longest jobs start + # while plenty of workers are still free. + heap: list[tuple[int, str]] = [(-targets[n].work, n) for n in roots] + heapq.heapify(heap) + while heap: + _, n = heapq.heappop(heap) + ready.put(n) + + def worker() -> None: + nonlocal pending + while True: + name = ready.get() + if name is None: + return + + target = targets[name] + dep_results = ( + {d.name: results[d.name] for d in target.deps} + if target.deps else empty_deps + ) + results[name] = target.build(dep_results) + + to_queue: list[str] = [] + done_flag = False + with lock: + pending -= 1 + if pending == 0: + done_flag = True + else: + for child in dependents[name]: + remaining[child] -= 1 + if remaining[child] == 0: + heapq.heappush(heap, (-targets[child].work, child)) + while heap: + _, n = heapq.heappop(heap) + to_queue.append(n) + + if done_flag: + for _ in range(num_workers - 1): + ready.put(None) + return + for n in to_queue: + ready.put(n) + else: + for name in roots: + ready.put(name) + + def worker() -> None: + nonlocal pending + # Bind hot names as locals (LOAD_FAST vs LOAD_DEREF for closures). + _targets = targets + _results = results + _dependents = dependents + _remaining = remaining + _empty = empty_deps + _ready_get = ready.get + _ready_put = ready.put + _lock = lock + _nworkers = num_workers + + while True: + name = _ready_get() + if name is None: + return + + # Inline-execution loop: continue down chains without + # round-tripping through the queue. + while True: + target = _targets[name] + deps = target.deps + dep_results = ( + {d.name: _results[d.name] for d in deps} + if deps else _empty + ) + # Publish result outside the lock. Dict writes are + # internally synchronized in 3.14t; the lock that follows + # provides the publish barrier for other workers. + _results[name] = target.build(dep_results) + + # Collect newly-ready children under the lock; enqueue + # them afterwards so the queue's lock isn't taken while + # we still hold the main lock. + new_ready: list[str] = [] + done_flag = False + with _lock: + pending -= 1 + if pending == 0: + done_flag = True + else: + for child in _dependents[name]: + _remaining[child] -= 1 + if _remaining[child] == 0: + new_ready.append(child) + + if done_flag: + for _ in range(_nworkers - 1): + _ready_put(None) + return + if not new_ready: + break + # Inline the first; queue the rest for other workers. + for c in new_ready[1:]: + _ready_put(c) + name = new_ready[0] + + threads = [threading.Thread(target=worker) for _ in range(num_workers - 1)] + for t in threads: + t.start() + worker() + for t in threads: + t.join() + + return results