From 33c2b9806bd54a96c4c7ed4f7ef6b6e0ce2c0764 Mon Sep 17 00:00:00 2001 From: Nikhil Saggi Date: Sat, 16 May 2026 13:09:30 -0700 Subject: [PATCH 1/3] Add build_all function for parallel target processing Implement parallel target building with dependency management. --- submissions/nikhilsaggi.py | 115 +++++++++++++++++++++++++++++++++++++ 1 file changed, 115 insertions(+) create mode 100644 submissions/nikhilsaggi.py diff --git a/submissions/nikhilsaggi.py b/submissions/nikhilsaggi.py new file mode 100644 index 0000000..2a9128e --- /dev/null +++ b/submissions/nikhilsaggi.py @@ -0,0 +1,115 @@ +from __future__ import annotations +import os +import queue +import threading + +from graph import BuildGraph + +def build_all(graph: BuildGraph) -> dict[str, bytes]: + """Build every target in `graph` in parallel, respecting dependencies. + + Algorithm (resembles Kahn's algorithm): + 1. Pre-compute two structures for each target: + - remaining_deps: a one-element list acting as an atomic counter of + how many dependencies have not yet finished. + - dependents: a list of (downstream_target, remaining_deps_ref) pairs + so that when this target finishes, it can decrement its dependents. + 2. Seed a thread-safe queue with all targets whose remaining_deps is 0 + (i.e., they have no dependencies and can build immediately). + 3. Worker threads loop: dequeue a target, gather its dependency results, + call target.build(dep_results), store the result, then for each + dependent decrement its counter. If it hits 0, enqueue it. + 4. The main thread waits on a Semaphore that workers release once per + completed target. After all N targets finish, it calls + ready.shutdown() which raises queue.ShutDown in blocked workers. + + Thread-safety (Python 3.14t / free-threading): + - remaining_deps counters are each a one-element list. Decrementing and + reading use a per-counter threading.Lock, replacing the old single + global lock to reduce contention on high-fan-out nodes. + - results[name] writes target disjoint keys. Safe without a lock in + CPython (dict bucket writes to distinct keys don't race). + - queue.Queue is fully thread-safe. + """ + targets = graph.targets + n_total = len(targets) + if n_total == 0: + return {} + + # --- Dependency bookkeeping --- + # remaining_deps[name] = [count, Lock] - mutable counter of unfinished deps. + # Using a per-target lock+list instead of a global lock so workers only + # contend when they share a downstream target, not on every completion. + remaining_deps: dict[str, list] = {} + # dependents[name] → list of (downstream_target, remaining_deps entry) + # Pre-resolved so workers never touch the shared dicts during the hot loop. + dependents: dict[str, list[tuple]] = {name: [] for name in targets} + + for name, target in targets.items(): + remaining_deps[name] = [len(target.deps), threading.Lock()] + + for name, target in targets.items(): + for dep in target.deps: + # Use references to `remaining_deps` entries + dependents[dep.name].append((target, remaining_deps[name])) + + # --- Shared state --- + # Maps target name → build result (bytes). Workers write disjoint keys. + results: dict[str, bytes] = {} + # Work queue: holds Target objects ready to build, or _SENTINEL to quit. + ready: queue.Queue = queue.Queue() + # Main thread acquires this N times; each worker release signals one done. + done_signal = threading.Semaphore(0) + + # Seed the queue with all root targets (no dependencies). + for name, (count, _lock) in remaining_deps.items(): + if count == 0: + ready.put(targets[name]) + + n_workers = min(n_total, os.cpu_count() or 1) + + def worker() -> None: + """Pull targets from the queue, build them, and enqueue dependents.""" + try: + while True: + target = ready.get() + + # Collect results from this target's already-completed dependencies. + dep_results = {dep.name: results[dep.name] for dep in target.deps} + + # Execute the build (CPU-bound work released by free-threading). + results[target.name] = target.build(dep_results) + + # Notify each downstream target that one of its deps is done. + for downstream, (counter_list_ref) in dependents[target.name]: + # counter_list_ref is [count, Lock] - decrement under its lock. + _cnt, _lk = counter_list_ref + enqueue = False + with _lk: + counter_list_ref[0] -= 1 + if counter_list_ref[0] == 0: + enqueue = True + if enqueue: + ready.put(downstream) + + # Tell the main thread one more target is done. + done_signal.release() + except queue.ShutDown: + return + + # --- Launch workers using raw threads (lighter than ThreadPoolExecutor + # since we manage scheduling ourselves and never use futures). --- + threads = [threading.Thread(target=worker, daemon=True) for _ in range(n_workers)] + for t in threads: + t.start() + + # Wait for every target to finish building. + for _ in range(n_total): + done_signal.acquire() + + # Shut down the queue; blocked get() calls raise queue.ShutDown. + ready.shutdown() + for t in threads: + t.join() + + return results From 0a12343298e752243e764049636b174b20bb7936 Mon Sep 17 00:00:00 2001 From: Nikhil Saggi Date: Sat, 16 May 2026 13:22:00 -0700 Subject: [PATCH 2/3] Optimize build_all Refactor build_all function to optimize threading and dependency management --- submissions/nikhilsaggi.py | 153 +++++++++++++++++-------------------- 1 file changed, 69 insertions(+), 84 deletions(-) diff --git a/submissions/nikhilsaggi.py b/submissions/nikhilsaggi.py index 2a9128e..b79d2f9 100644 --- a/submissions/nikhilsaggi.py +++ b/submissions/nikhilsaggi.py @@ -5,110 +5,95 @@ from graph import BuildGraph +NUM_WORKERS = 24 + + def build_all(graph: BuildGraph) -> dict[str, bytes]: """Build every target in `graph` in parallel, respecting dependencies. - Algorithm (resembles Kahn's algorithm): - 1. Pre-compute two structures for each target: - - remaining_deps: a one-element list acting as an atomic counter of - how many dependencies have not yet finished. - - dependents: a list of (downstream_target, remaining_deps_ref) pairs - so that when this target finishes, it can decrement its dependents. - 2. Seed a thread-safe queue with all targets whose remaining_deps is 0 - (i.e., they have no dependencies and can build immediately). - 3. Worker threads loop: dequeue a target, gather its dependency results, - call target.build(dep_results), store the result, then for each - dependent decrement its counter. If it hits 0, enqueue it. - 4. The main thread waits on a Semaphore that workers release once per - completed target. After all N targets finish, it calls - ready.shutdown() which raises queue.ShutDown in blocked workers. - - Thread-safety (Python 3.14t / free-threading): - - remaining_deps counters are each a one-element list. Decrementing and - reading use a per-counter threading.Lock, replacing the old single - global lock to reduce contention on high-fan-out nodes. - - results[name] writes target disjoint keys. Safe without a lock in - CPython (dict bucket writes to distinct keys don't race). - - queue.Queue is fully thread-safe. + Uses Kahn's algorithm with three key optimizations: + - Chain fast path: if the graph is a simple chain (single root, + max fan-out 1), skip all threading overhead and build sequentially. + - Inline execution: when a completed target has exactly one newly-ready + dependent, build it directly in the same thread instead of enqueuing. + - Main-thread-as-worker: spawn N-1 helper threads and use the main + thread as the Nth worker, saving one thread creation. """ targets = graph.targets n_total = len(targets) if n_total == 0: return {} - # --- Dependency bookkeeping --- - # remaining_deps[name] = [count, Lock] - mutable counter of unfinished deps. - # Using a per-target lock+list instead of a global lock so workers only - # contend when they share a downstream target, not on every completion. - remaining_deps: dict[str, list] = {} - # dependents[name] → list of (downstream_target, remaining_deps entry) - # Pre-resolved so workers never touch the shared dicts during the hot loop. - dependents: dict[str, list[tuple]] = {name: [] for name in targets} - - for name, target in targets.items(): - remaining_deps[name] = [len(target.deps), threading.Lock()] - + remaining = {name: len(target.deps) for name, target in targets.items()} + dependents: dict[str, list[str]] = {name: [] for name in targets} for name, target in targets.items(): for dep in target.deps: - # Use references to `remaining_deps` entries - dependents[dep.name].append((target, remaining_deps[name])) - - # --- Shared state --- - # Maps target name → build result (bytes). Workers write disjoint keys. + dependents[dep.name].append(name) + + roots = [name for name, count in remaining.items() if count == 0] + max_fan_out = max((len(children) for children in dependents.values()), default=0) + + # --- Chain fast path: skip threading entirely for linear graphs --- + if len(roots) == 1 and max_fan_out <= 1: + results: dict[str, bytes] = {} + name = roots[0] + while True: + target = targets[name] + dep_results = {dep.name: results[dep.name] for dep in target.deps} + results[name] = target.build(dep_results) + children = dependents[name] + if not children: + return results + name = children[0] + + # --- Parallel path --- results: dict[str, bytes] = {} - # Work queue: holds Target objects ready to build, or _SENTINEL to quit. - ready: queue.Queue = queue.Queue() - # Main thread acquires this N times; each worker release signals one done. - done_signal = threading.Semaphore(0) + lock = threading.Lock() + pending = n_total + ready: queue.SimpleQueue[str | None] = queue.SimpleQueue() - # Seed the queue with all root targets (no dependencies). - for name, (count, _lock) in remaining_deps.items(): - if count == 0: - ready.put(targets[name]) + for name in roots: + ready.put(name) - n_workers = min(n_total, os.cpu_count() or 1) + n_workers = min(NUM_WORKERS, os.cpu_count() or 1, n_total) def worker() -> None: - """Pull targets from the queue, build them, and enqueue dependents.""" - try: - while True: - target = ready.get() - - # Collect results from this target's already-completed dependencies. - dep_results = {dep.name: results[dep.name] for dep in target.deps} - - # Execute the build (CPU-bound work released by free-threading). - results[target.name] = target.build(dep_results) - - # Notify each downstream target that one of its deps is done. - for downstream, (counter_list_ref) in dependents[target.name]: - # counter_list_ref is [count, Lock] - decrement under its lock. - _cnt, _lk = counter_list_ref - enqueue = False - with _lk: - counter_list_ref[0] -= 1 - if counter_list_ref[0] == 0: - enqueue = True - if enqueue: - ready.put(downstream) - - # Tell the main thread one more target is done. - done_signal.release() - except queue.ShutDown: - return - - # --- Launch workers using raw threads (lighter than ThreadPoolExecutor - # since we manage scheduling ourselves and never use futures). --- - threads = [threading.Thread(target=worker, daemon=True) for _ in range(n_workers)] + nonlocal pending + + name = ready.get() + while name is not None: + target = targets[name] + dep_results = {dep.name: results[dep.name] for dep in target.deps} + result = target.build(dep_results) + + # Under lock: store result, decrement dependents, find next target. + inline = None + with lock: + results[name] = result + pending -= 1 + if pending == 0: + for _ in range(n_workers): + ready.put(None) + else: + for child in dependents[name]: + remaining[child] -= 1 + if remaining[child] == 0: + if inline is None: + inline = child + else: + ready.put(child) + + if inline is not None: + name = inline + else: + name = ready.get() + + threads = [threading.Thread(target=worker, daemon=True) for _ in range(n_workers - 1)] for t in threads: t.start() - # Wait for every target to finish building. - for _ in range(n_total): - done_signal.acquire() + worker() - # Shut down the queue; blocked get() calls raise queue.ShutDown. - ready.shutdown() for t in threads: t.join() From 11d698a184572430c309b698c053cb68c8c9c46d Mon Sep 17 00:00:00 2001 From: Nikhil Saggi Date: Sat, 16 May 2026 15:43:21 -0700 Subject: [PATCH 3/3] Use sentinel --- submissions/nikhilsaggi.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/submissions/nikhilsaggi.py b/submissions/nikhilsaggi.py index b79d2f9..70cccfa 100644 --- a/submissions/nikhilsaggi.py +++ b/submissions/nikhilsaggi.py @@ -5,6 +5,7 @@ from graph import BuildGraph +_SENTINEL = object() NUM_WORKERS = 24 @@ -28,6 +29,7 @@ def build_all(graph: BuildGraph) -> dict[str, bytes]: dependents: dict[str, list[str]] = {name: [] for name in targets} for name, target in targets.items(): for dep in target.deps: + # Add parent's name as a dependent for child dependents[dep.name].append(name) roots = [name for name, count in remaining.items() if count == 0] @@ -50,7 +52,7 @@ def build_all(graph: BuildGraph) -> dict[str, bytes]: results: dict[str, bytes] = {} lock = threading.Lock() pending = n_total - ready: queue.SimpleQueue[str | None] = queue.SimpleQueue() + ready: queue.SimpleQueue = queue.SimpleQueue() for name in roots: ready.put(name) @@ -61,7 +63,7 @@ def worker() -> None: nonlocal pending name = ready.get() - while name is not None: + while name is not _SENTINEL: target = targets[name] dep_results = {dep.name: results[dep.name] for dep in target.deps} result = target.build(dep_results) @@ -73,7 +75,7 @@ def worker() -> None: pending -= 1 if pending == 0: for _ in range(n_workers): - ready.put(None) + ready.put(_SENTINEL) else: for child in dependents[name]: remaining[child] -= 1