Add dharamendrak submission

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