From d191056ff2c118b8d0170e6550b4d2f5e88628fe Mon Sep 17 00:00:00 2001 From: Kevin Newton Date: Fri, 15 May 2026 13:38:24 -0700 Subject: [PATCH 01/15] kddnewton --- submissions/kddnewton.py | 73 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 73 insertions(+) create mode 100644 submissions/kddnewton.py diff --git a/submissions/kddnewton.py b/submissions/kddnewton.py new file mode 100644 index 0000000..0e2652f --- /dev/null +++ b/submissions/kddnewton.py @@ -0,0 +1,73 @@ +import _thread +import queue +import threading + +from graph import BuildGraph + +NUM_WORKERS = 24 + + +def build_all(graph: BuildGraph): + targets = graph.targets + n = len(targets) + in_degree = [0] * n + dependents = [None] * n + ready = queue.SimpleQueue() + dep_info = [None] * n + has_parallelism = False + + for i, target in enumerate(target_list := list(targets.values())): + target._id = i + if target.deps: + in_degree[i] = len(target.deps) + dep_info[i] = [(dep.name, dep._id) for dep in target.deps] + for dep in target.deps: + dep_id = dep._id + if dependents[dep_id] is None: + dependents[dep_id] = [i] + else: + if not has_parallelism: + has_parallelism = True + dependents[dep_id].append(i) + else: + dep_info[i] = () + if not has_parallelism and not ready.empty(): + has_parallelism = True + ready.put(i) + + if has_parallelism: + results = [None] * n + lock = threading.Lock() + done = threading.Event() + + def worker(): + nonlocal n + while (tid := ready.get()) != -1: + results[tid] = target_list[tid].build( + {name: results[idx] for name, idx in dep_info[tid]} + ) + + with lock: + n -= 1 + if n == 0: + done.set() + for _ in range(NUM_WORKERS - 1): + ready.put(-1) + return + succs = dependents[tid] + if succs: + for succ_id in succs: + in_degree[succ_id] -= 1 + if in_degree[succ_id] == 0: + ready.put(succ_id) + + for _ in range(NUM_WORKERS): + _thread.start_joinable_thread(worker, daemon=True) + done.wait() + else: + results = [None] * n + result = target_list[tid := ready.get()].build({}) + for _ in range(n - 1): + results[tid] = result + name, idx = dep_info[tid := dependents[tid][0]][0] + result = target_list[tid].build({name: results[idx]}) From b7954fbf26f96dda1e4088a7efb73bbdd15e8edb Mon Sep 17 00:00:00 2001 From: Kevin Newton Date: Fri, 15 May 2026 17:16:18 -0700 Subject: [PATCH 02/15] Again --- submissions/kddnewton.py | 87 +++++++++++++++++++--------------------- 1 file changed, 41 insertions(+), 46 deletions(-) diff --git a/submissions/kddnewton.py b/submissions/kddnewton.py index 0e2652f..6be04bd 100644 --- a/submissions/kddnewton.py +++ b/submissions/kddnewton.py @@ -9,65 +9,60 @@ def build_all(graph: BuildGraph): targets = graph.targets - n = len(targets) - in_degree = [0] * n - dependents = [None] * n - ready = queue.SimpleQueue() - dep_info = [None] * n + remaining = len(targets) has_parallelism = False - for i, target in enumerate(target_list := list(targets.values())): - target._id = i - if target.deps: - in_degree[i] = len(target.deps) - dep_info[i] = [(dep.name, dep._id) for dep in target.deps] - for dep in target.deps: - dep_id = dep._id - if dependents[dep_id] is None: - dependents[dep_id] = [i] - else: - if not has_parallelism: - has_parallelism = True - dependents[dep_id].append(i) + ready = queue.SimpleQueue() + in_degree = {name: 0 for name in targets} + dependents = {name: [] for name in targets} + empty_deps = {} + + for name, target in targets.items(): + if (deps := target.deps): + in_degree[name] = len(deps) + for dep in deps: + if not has_parallelism and dependents[dep.name]: + has_parallelism = True + dependents[dep.name].append(name) else: - dep_info[i] = () if not has_parallelism and not ready.empty(): has_parallelism = True - ready.put(i) + ready.put(name) + + results = {} if has_parallelism: - results = [None] * n lock = threading.Lock() - done = threading.Event() + sentinel = None def worker(): - nonlocal n - while (tid := ready.get()) != -1: - results[tid] = target_list[tid].build( - {name: results[idx] for name, idx in dep_info[tid]} - ) + nonlocal remaining + while True: + if (name := ready.get()) is sentinel: + return + + target = targets[name] + results[name] = target.build({dep.name: results[dep.name] for dep in target.deps} if target.deps else empty_deps) with lock: - n -= 1 - if n == 0: - done.set() + remaining -= 1 + if remaining == 0: for _ in range(NUM_WORKERS - 1): - ready.put(-1) + ready.put(sentinel) return - succs = dependents[tid] - if succs: - for succ_id in succs: - in_degree[succ_id] -= 1 - if in_degree[succ_id] == 0: - ready.put(succ_id) + for dependent in dependents[name]: + in_degree[dependent] -= 1 + if in_degree[dependent] == 0: + ready.put(dependent) - for _ in range(NUM_WORKERS): - _thread.start_joinable_thread(worker, daemon=True) - done.wait() + handles = [_thread.start_joinable_thread(worker, daemon=True) for _ in range(NUM_WORKERS)] + for handle in handles: + handle.join() else: - results = [None] * n - result = target_list[tid := ready.get()].build({}) - for _ in range(n - 1): - results[tid] = result - name, idx = dep_info[tid := dependents[tid][0]][0] - result = target_list[tid].build({name: results[idx]}) + name = ready.get() + results[name] = targets[name].build({}) + + for _ in range(remaining - 1): + dependent = dependents[name][0] + results[dependent] = targets[dependent].build({name: results[name]}) + name = dependent From b98782b81c2afd4548efbb56799aaec13c517fab Mon Sep 17 00:00:00 2001 From: Kevin Newton Date: Fri, 15 May 2026 17:27:21 -0700 Subject: [PATCH 03/15] Again --- submissions/kddnewton.py | 16 ++++------------ 1 file changed, 4 insertions(+), 12 deletions(-) diff --git a/submissions/kddnewton.py b/submissions/kddnewton.py index 6be04bd..7afc1ab 100644 --- a/submissions/kddnewton.py +++ b/submissions/kddnewton.py @@ -10,35 +10,27 @@ def build_all(graph: BuildGraph): targets = graph.targets remaining = len(targets) - has_parallelism = False - ready = queue.SimpleQueue() - in_degree = {name: 0 for name in targets} + in_degree = {name: len(target.deps) for name, target in targets.items()} dependents = {name: [] for name in targets} empty_deps = {} for name, target in targets.items(): if (deps := target.deps): - in_degree[name] = len(deps) for dep in deps: - if not has_parallelism and dependents[dep.name]: - has_parallelism = True dependents[dep.name].append(name) else: - if not has_parallelism and not ready.empty(): - has_parallelism = True ready.put(name) results = {} - if has_parallelism: + if ready.qsize() > 1 or any(len(value) > 1 for value in dependents.values()): lock = threading.Lock() - sentinel = None def worker(): nonlocal remaining while True: - if (name := ready.get()) is sentinel: + if (name := ready.get()) is None: return target = targets[name] @@ -48,7 +40,7 @@ def worker(): remaining -= 1 if remaining == 0: for _ in range(NUM_WORKERS - 1): - ready.put(sentinel) + ready.put(None) return for dependent in dependents[name]: in_degree[dependent] -= 1 From d3dacc3ef36a09f5633bce3d0cecd43c7edab1f3 Mon Sep 17 00:00:00 2001 From: Kevin Newton Date: Fri, 15 May 2026 17:35:17 -0700 Subject: [PATCH 04/15] Again --- submissions/kddnewton.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/submissions/kddnewton.py b/submissions/kddnewton.py index 7afc1ab..cea3b0a 100644 --- a/submissions/kddnewton.py +++ b/submissions/kddnewton.py @@ -1,4 +1,5 @@ import _thread +import gc import queue import threading @@ -8,6 +9,7 @@ def build_all(graph: BuildGraph): + gc.disable() targets = graph.targets remaining = len(targets) ready = queue.SimpleQueue() From eaffde8243d6e4aea212659be9934c2243963f72 Mon Sep 17 00:00:00 2001 From: Kevin Newton Date: Fri, 15 May 2026 17:47:20 -0700 Subject: [PATCH 05/15] Again --- submissions/kddnewton.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/submissions/kddnewton.py b/submissions/kddnewton.py index cea3b0a..391e671 100644 --- a/submissions/kddnewton.py +++ b/submissions/kddnewton.py @@ -1,5 +1,3 @@ -import _thread -import gc import queue import threading @@ -9,7 +7,6 @@ def build_all(graph: BuildGraph): - gc.disable() targets = graph.targets remaining = len(targets) ready = queue.SimpleQueue() @@ -49,9 +46,12 @@ def worker(): if in_degree[dependent] == 0: ready.put(dependent) - handles = [_thread.start_joinable_thread(worker, daemon=True) for _ in range(NUM_WORKERS)] - for handle in handles: - handle.join() + threads = [threading.Thread(target=worker) for _ in range(NUM_WORKERS - 1)] + for t in threads: + t.start() + worker() + for t in threads: + t.join() else: name = ready.get() results[name] = targets[name].build({}) From 505ab60191a85623b52c8bfb38e22232e8a26a60 Mon Sep 17 00:00:00 2001 From: Kevin Newton Date: Sat, 16 May 2026 11:12:00 -0700 Subject: [PATCH 06/15] Again --- submissions/kddnewton.py | 27 +++++++++++++++++++-------- 1 file changed, 19 insertions(+), 8 deletions(-) diff --git a/submissions/kddnewton.py b/submissions/kddnewton.py index 391e671..b0edc0f 100644 --- a/submissions/kddnewton.py +++ b/submissions/kddnewton.py @@ -1,5 +1,6 @@ import queue import threading +import heapq from graph import BuildGraph @@ -9,7 +10,10 @@ def build_all(graph: BuildGraph): targets = graph.targets remaining = len(targets) + + heap = [] ready = queue.SimpleQueue() + in_degree = {name: len(target.deps) for name, target in targets.items()} dependents = {name: [] for name in targets} empty_deps = {} @@ -19,13 +23,17 @@ def build_all(graph: BuildGraph): for dep in deps: dependents[dep.name].append(name) else: - ready.put(name) + heapq.heappush(heap, (-target.work, name)) results = {} - if ready.qsize() > 1 or any(len(value) > 1 for value in dependents.values()): + if len(heap) > 1 or any(len(v) > 1 for v in dependents.values()): lock = threading.Lock() + while heap: + _, name = heapq.heappop(heap) + ready.put(name) + def worker(): nonlocal remaining while True: @@ -44,16 +52,19 @@ def worker(): for dependent in dependents[name]: in_degree[dependent] -= 1 if in_degree[dependent] == 0: - ready.put(dependent) + heapq.heappush(heap, (-targets[dependent].work, dependent)) + while heap: + _, ready_name = heapq.heappop(heap) + ready.put(ready_name) threads = [threading.Thread(target=worker) for _ in range(NUM_WORKERS - 1)] - for t in threads: - t.start() + for thread in threads: + thread.start() worker() - for t in threads: - t.join() + for thread in threads: + thread.join() else: - name = ready.get() + _, name = heap[0] results[name] = targets[name].build({}) for _ in range(remaining - 1): From 2130a74cdc925bfcce0751b3dfe5bb0283962a14 Mon Sep 17 00:00:00 2001 From: Kevin Newton Date: Sat, 16 May 2026 11:23:27 -0700 Subject: [PATCH 07/15] Again --- submissions/kddnewton.py | 64 +++++++++++++++++++++++++++------------- 1 file changed, 44 insertions(+), 20 deletions(-) diff --git a/submissions/kddnewton.py b/submissions/kddnewton.py index b0edc0f..2cb5b65 100644 --- a/submissions/kddnewton.py +++ b/submissions/kddnewton.py @@ -27,35 +27,59 @@ def build_all(graph: BuildGraph): results = {} - if len(heap) > 1 or any(len(v) > 1 for v in dependents.values()): + max_fan_in = max(in_degree.values()) + max_fan_out = max((len(v) for v in dependents.values()), default=0) + + if len(heap) > 1 or max_fan_out > 1: lock = threading.Lock() while heap: _, name = heapq.heappop(heap) ready.put(name) - def worker(): - nonlocal remaining - while True: - if (name := ready.get()) is None: - return + if max_fan_in > NUM_WORKERS and max_fan_out > NUM_WORKERS: + def worker(): + nonlocal remaining + while True: + if (name := ready.get()) is None: + return - target = targets[name] - results[name] = target.build({dep.name: results[dep.name] for dep in target.deps} if target.deps else empty_deps) + target = targets[name] + results[name] = target.build({dep.name: results[dep.name] for dep in target.deps} if target.deps else empty_deps) - with lock: - remaining -= 1 - if remaining == 0: - for _ in range(NUM_WORKERS - 1): - ready.put(None) + with lock: + remaining -= 1 + if remaining == 0: + for _ in range(NUM_WORKERS - 1): + ready.put(None) + return + for dependent in dependents[name]: + in_degree[dependent] -= 1 + if in_degree[dependent] == 0: + heapq.heappush(heap, (-targets[dependent].work, dependent)) + while heap: + _, ready_name = heapq.heappop(heap) + ready.put(ready_name) + else: + def worker(): + nonlocal remaining + while True: + if (name := ready.get()) is None: return - for dependent in dependents[name]: - in_degree[dependent] -= 1 - if in_degree[dependent] == 0: - heapq.heappush(heap, (-targets[dependent].work, dependent)) - while heap: - _, ready_name = heapq.heappop(heap) - ready.put(ready_name) + + target = targets[name] + results[name] = target.build({dep.name: results[dep.name] for dep in target.deps} if target.deps else empty_deps) + + with lock: + remaining -= 1 + if remaining == 0: + for _ in range(NUM_WORKERS - 1): + ready.put(None) + return + for dependent in dependents[name]: + in_degree[dependent] -= 1 + if in_degree[dependent] == 0: + ready.put(dependent) threads = [threading.Thread(target=worker) for _ in range(NUM_WORKERS - 1)] for thread in threads: From f12df6ffedbe2a07f935b4fb0462a4e110bd109f Mon Sep 17 00:00:00 2001 From: Kevin Newton Date: Sat, 16 May 2026 11:57:06 -0700 Subject: [PATCH 08/15] Again --- submissions/kddnewton.py | 32 +++++++++++++------------------- 1 file changed, 13 insertions(+), 19 deletions(-) diff --git a/submissions/kddnewton.py b/submissions/kddnewton.py index 2cb5b65..0771a59 100644 --- a/submissions/kddnewton.py +++ b/submissions/kddnewton.py @@ -1,6 +1,6 @@ +import _thread import queue import threading -import heapq from graph import BuildGraph @@ -11,33 +11,29 @@ def build_all(graph: BuildGraph): targets = graph.targets remaining = len(targets) - heap = [] ready = queue.SimpleQueue() in_degree = {name: len(target.deps) for name, target in targets.items()} dependents = {name: [] for name in targets} - empty_deps = {} - for name, target in targets.items(): if (deps := target.deps): for dep in deps: dependents[dep.name].append(name) else: - heapq.heappush(heap, (-target.work, name)) + ready.put(name) results = {} max_fan_in = max(in_degree.values()) - max_fan_out = max((len(v) for v in dependents.values()), default=0) + max_fan_out = max((len(value) for value in dependents.values()), default=0) - if len(heap) > 1 or max_fan_out > 1: + if ready.qsize() > 1 or max_fan_out > 1: lock = threading.Lock() - - while heap: - _, name = heapq.heappop(heap) - ready.put(name) + heap = [] if max_fan_in > NUM_WORKERS and max_fan_out > NUM_WORKERS: + import heapq + def worker(): nonlocal remaining while True: @@ -45,7 +41,7 @@ def worker(): return target = targets[name] - results[name] = target.build({dep.name: results[dep.name] for dep in target.deps} if target.deps else empty_deps) + results[name] = target.build({dep.name: results[dep.name] for dep in target.deps}) with lock: remaining -= 1 @@ -68,7 +64,7 @@ def worker(): return target = targets[name] - results[name] = target.build({dep.name: results[dep.name] for dep in target.deps} if target.deps else empty_deps) + results[name] = target.build({dep.name: results[dep.name] for dep in target.deps}) with lock: remaining -= 1 @@ -81,14 +77,12 @@ def worker(): if in_degree[dependent] == 0: ready.put(dependent) - threads = [threading.Thread(target=worker) for _ in range(NUM_WORKERS - 1)] - for thread in threads: - thread.start() + handles = [_thread.start_joinable_thread(worker, daemon=True) for _ in range(NUM_WORKERS - 1)] worker() - for thread in threads: - thread.join() + for handle in handles: + handle.join() else: - _, name = heap[0] + name = ready.get() results[name] = targets[name].build({}) for _ in range(remaining - 1): From f7532f667ebf830bd028cbe0e0e70c0ddc7e9171 Mon Sep 17 00:00:00 2001 From: Kevin Newton Date: Sat, 16 May 2026 12:24:21 -0700 Subject: [PATCH 09/15] Again --- submissions/kddnewton.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/submissions/kddnewton.py b/submissions/kddnewton.py index 0771a59..1a5404f 100644 --- a/submissions/kddnewton.py +++ b/submissions/kddnewton.py @@ -77,10 +77,9 @@ def worker(): if in_degree[dependent] == 0: ready.put(dependent) - handles = [_thread.start_joinable_thread(worker, daemon=True) for _ in range(NUM_WORKERS - 1)] + for _ in range(NUM_WORKERS - 1): + _thread.start_joinable_thread(worker, daemon=True) worker() - for handle in handles: - handle.join() else: name = ready.get() results[name] = targets[name].build({}) From bc183c8c2b486c7aff838fbbdf934a9a077a5859 Mon Sep 17 00:00:00 2001 From: Kevin Newton Date: Sat, 16 May 2026 13:00:33 -0700 Subject: [PATCH 10/15] Again --- submissions/kddnewton.py | 38 ++++++++++++++++++++++---------------- 1 file changed, 22 insertions(+), 16 deletions(-) diff --git a/submissions/kddnewton.py b/submissions/kddnewton.py index 1a5404f..72dac1e 100644 --- a/submissions/kddnewton.py +++ b/submissions/kddnewton.py @@ -40,22 +40,28 @@ def worker(): if (name := ready.get()) is None: return - target = targets[name] - results[name] = target.build({dep.name: results[dep.name] for dep in target.deps}) - - with lock: - remaining -= 1 - if remaining == 0: - for _ in range(NUM_WORKERS - 1): - ready.put(None) - return - for dependent in dependents[name]: - in_degree[dependent] -= 1 - if in_degree[dependent] == 0: - heapq.heappush(heap, (-targets[dependent].work, dependent)) - while heap: - _, ready_name = heapq.heappop(heap) - ready.put(ready_name) + while name is not None: + target = targets[name] + results[name] = target.build({dep.name: results[dep.name] for dep in target.deps}) + + next_name = None + with lock: + remaining -= 1 + if remaining == 0: + for _ in range(NUM_WORKERS - 1): + ready.put(None) + return + for dependent in dependents[name]: + in_degree[dependent] -= 1 + if in_degree[dependent] == 0: + heapq.heappush(heap, (-targets[dependent].work, dependent)) + if len(heap) == 1: + _, next_name = heapq.heappop(heap) + else: + while heap: + _, ready_name = heapq.heappop(heap) + ready.put(ready_name) + name = next_name else: def worker(): nonlocal remaining From 432aaa1983b83f9034afe1ca8edd4591d49fe8e6 Mon Sep 17 00:00:00 2001 From: Kevin Newton Date: Sat, 16 May 2026 13:46:03 -0700 Subject: [PATCH 11/15] Again --- submissions/kddnewton.py | 56 +++++++++++++++++++++------------------- 1 file changed, 29 insertions(+), 27 deletions(-) diff --git a/submissions/kddnewton.py b/submissions/kddnewton.py index 72dac1e..29749c1 100644 --- a/submissions/kddnewton.py +++ b/submissions/kddnewton.py @@ -1,55 +1,53 @@ import _thread import queue import threading +import heapq from graph import BuildGraph NUM_WORKERS = 24 +_ready = queue.SimpleQueue() +_lock = threading.Lock() +_results = [None] * 20000 + def build_all(graph: BuildGraph): targets = graph.targets remaining = len(targets) - ready = queue.SimpleQueue() - in_degree = {name: len(target.deps) for name, target in targets.items()} dependents = {name: [] for name in targets} - for name, target in targets.items(): + + for id, (name, target) in enumerate(targets.items()): + target._id = id if (deps := target.deps): + deps.sort(key=lambda d: d.name) for dep in deps: dependents[dep.name].append(name) else: - ready.put(name) - - results = {} + _ready.put(name) - max_fan_in = max(in_degree.values()) - max_fan_out = max((len(value) for value in dependents.values()), default=0) - - if ready.qsize() > 1 or max_fan_out > 1: - lock = threading.Lock() + if _ready.qsize() > 1 or (max_fan_out := max((len(value) for value in dependents.values()), default=0)) > 1: heap = [] - if max_fan_in > NUM_WORKERS and max_fan_out > NUM_WORKERS: - import heapq - + if max(in_degree.values()) > NUM_WORKERS and max_fan_out > NUM_WORKERS: def worker(): nonlocal remaining while True: - if (name := ready.get()) is None: + if (name := _ready.get()) is None: return while name is not None: target = targets[name] - results[name] = target.build({dep.name: results[dep.name] for dep in target.deps}) + _results[target._id] = target.build({dep.name: _results[dep._id] for dep in target.deps}) next_name = None - with lock: + with _lock: remaining -= 1 if remaining == 0: for _ in range(NUM_WORKERS - 1): - ready.put(None) + _ready.put(None) return for dependent in dependents[name]: in_degree[dependent] -= 1 @@ -60,37 +58,41 @@ def worker(): else: while heap: _, ready_name = heapq.heappop(heap) - ready.put(ready_name) + _ready.put(ready_name) name = next_name else: def worker(): nonlocal remaining while True: - if (name := ready.get()) is None: + if (name := _ready.get()) is None: return target = targets[name] - results[name] = target.build({dep.name: results[dep.name] for dep in target.deps}) + _results[target._id] = target.build({dep.name: _results[dep._id] for dep in target.deps}) - with lock: + with _lock: remaining -= 1 if remaining == 0: for _ in range(NUM_WORKERS - 1): - ready.put(None) + _ready.put(None) return for dependent in dependents[name]: in_degree[dependent] -= 1 if in_degree[dependent] == 0: - ready.put(dependent) + _ready.put(dependent) for _ in range(NUM_WORKERS - 1): _thread.start_joinable_thread(worker, daemon=True) worker() else: - name = ready.get() - results[name] = targets[name].build({}) + name = _ready.get() + target = targets[name] + + results = _results + results[target._id] = target.build({}) for _ in range(remaining - 1): dependent = dependents[name][0] - results[dependent] = targets[dependent].build({name: results[name]}) + target = targets[dependent] + results[target._id] = target.build({name: results[targets[name]._id]}) name = dependent From f14276c196ac4559402e453c39d8709deb826560 Mon Sep 17 00:00:00 2001 From: Kevin Newton Date: Sat, 16 May 2026 15:09:01 -0700 Subject: [PATCH 12/15] Again --- submissions/kddnewton.py | 144 ++++++++++++++++++++++----------------- 1 file changed, 81 insertions(+), 63 deletions(-) diff --git a/submissions/kddnewton.py b/submissions/kddnewton.py index 29749c1..76516ea 100644 --- a/submissions/kddnewton.py +++ b/submissions/kddnewton.py @@ -1,98 +1,116 @@ -import _thread +import ctypes +import heapq import queue import threading -import heapq from graph import BuildGraph -NUM_WORKERS = 24 + +_NUM_WORKERS = 24 +_NUM_WORKERS_M1 = _NUM_WORKERS - 1 + + +def immortalize(obj): + ctypes.c_uint32.from_address(id(obj) + 12).value = 0xFFFFFFFF + _ready = queue.SimpleQueue() _lock = threading.Lock() _results = [None] * 20000 +_SENTINEL = object() + +immortalize(_ready) +immortalize(_lock) +immortalize(_results) +immortalize(_SENTINEL) + + +class _Remaining: + __slots__ = ("value",) + def __init__(self, value): + self.value = value def build_all(graph: BuildGraph): targets = graph.targets - remaining = len(targets) - - in_degree = {name: len(target.deps) for name, target in targets.items()} - dependents = {name: [] for name in targets} + remaining = _Remaining(len(targets)) for id, (name, target) in enumerate(targets.items()): target._id = id + target._dependents = [] + + for target in targets.values(): if (deps := target.deps): - deps.sort(key=lambda d: d.name) + deps.sort(key=lambda dep: dep.name) + target._in_degree = len(deps) for dep in deps: - dependents[dep.name].append(name) + dep._dependents.append(target) else: - _ready.put(name) + target._in_degree = 0 + _ready.put(target) - if _ready.qsize() > 1 or (max_fan_out := max((len(value) for value in dependents.values()), default=0)) > 1: + if _ready.qsize() > 1 or (max_fan_out := max((len(target._dependents) for target in targets.values()), default=0)) > 1: heap = [] - if max(in_degree.values()) > NUM_WORKERS and max_fan_out > NUM_WORKERS: - def worker(): - nonlocal remaining - while True: - if (name := _ready.get()) is None: - return - - while name is not None: - target = targets[name] - _results[target._id] = target.build({dep.name: _results[dep._id] for dep in target.deps}) - - next_name = None - with _lock: - remaining -= 1 - if remaining == 0: - for _ in range(NUM_WORKERS - 1): - _ready.put(None) - return - for dependent in dependents[name]: - in_degree[dependent] -= 1 - if in_degree[dependent] == 0: - heapq.heappush(heap, (-targets[dependent].work, dependent)) - if len(heap) == 1: - _, next_name = heapq.heappop(heap) + if max(target._in_degree for target in targets.values()) > _NUM_WORKERS and max_fan_out > _NUM_WORKERS: + def worker(remaining): + target = _ready.get() + + while target is not _SENTINEL: + _results[target._id] = target.build({dep.name: _results[dep._id] for dep in target.deps}) + + next_target = None + with _lock: + if remaining.value == 1: + for _ in range(_NUM_WORKERS_M1): + _ready.put(_SENTINEL) + return + + remaining.value -= 1 + for dependent in target._dependents: + if dependent._in_degree == 1: + heapq.heappush(heap, (-dependent.work, dependent._id, dependent)) else: - while heap: - _, ready_name = heapq.heappop(heap) - _ready.put(ready_name) - name = next_name + dependent._in_degree -= 1 + + if len(heap) == 1: + next_target = heapq.heappop(heap)[2] + else: + while heap: + _ready.put(heapq.heappop(heap)[2]) + + if next_target is not None: + target = next_target + else: + target = _ready.get() else: - def worker(): - nonlocal remaining - while True: - if (name := _ready.get()) is None: - return - - target = targets[name] + def worker(remaining): + while (target := _ready.get()) is not _SENTINEL: _results[target._id] = target.build({dep.name: _results[dep._id] for dep in target.deps}) with _lock: - remaining -= 1 - if remaining == 0: - for _ in range(NUM_WORKERS - 1): - _ready.put(None) + if remaining.value == 1: + for _ in range(_NUM_WORKERS_M1): + _ready.put(_SENTINEL) return - for dependent in dependents[name]: - in_degree[dependent] -= 1 - if in_degree[dependent] == 0: + + remaining.value -= 1 + for dependent in target._dependents: + if dependent._in_degree == 1: _ready.put(dependent) + else: + dependent._in_degree -= 1 - for _ in range(NUM_WORKERS - 1): - _thread.start_joinable_thread(worker, daemon=True) - worker() + for _ in range(_NUM_WORKERS_M1): + threading.Thread(target=worker, args=(remaining,), daemon=True).start() + worker(remaining) else: - name = _ready.get() - target = targets[name] + target = _ready.get() results = _results results[target._id] = target.build({}) - for _ in range(remaining - 1): - dependent = dependents[name][0] - target = targets[dependent] - results[target._id] = target.build({name: results[targets[name]._id]}) - name = dependent + for _ in range(remaining.value - 1): + dependent = target._dependents[0] + results[dependent._id] = dependent.build({target.name: results[target._id]}) + target = dependent From 7755284d88975ac2878d00389d8c913d470972ec Mon Sep 17 00:00:00 2001 From: Kevin Newton Date: Sun, 17 May 2026 12:29:58 -0700 Subject: [PATCH 13/15] Again --- submissions/kddnewton.py | 186 ++++++++++++++++++++++----------------- 1 file changed, 107 insertions(+), 79 deletions(-) diff --git a/submissions/kddnewton.py b/submissions/kddnewton.py index 76516ea..ad85664 100644 --- a/submissions/kddnewton.py +++ b/submissions/kddnewton.py @@ -6,111 +6,139 @@ from graph import BuildGraph -_NUM_WORKERS = 24 -_NUM_WORKERS_M1 = _NUM_WORKERS - 1 +class _Remaining: + __slots__ = ("value",) + def __init__(self): + self.value = 0 -def immortalize(obj): - ctypes.c_uint32.from_address(id(obj) + 12).value = 0xFFFFFFFF +_NUM_WORKERS = 24 -_ready = queue.SimpleQueue() +_parallel_ready = queue.SimpleQueue() +_diamond_ready = queue.SimpleQueue() _lock = threading.Lock() +_done = threading.Event() _results = [None] * 20000 -_SENTINEL = object() +_remaining = _Remaining() -immortalize(_ready) -immortalize(_lock) -immortalize(_results) -immortalize(_SENTINEL) +def _immortalize(obj): + ctypes.c_uint32.from_address(id(obj) + 12).value = 0xFFFFFFFF -class _Remaining: - __slots__ = ("value",) - def __init__(self, value): - self.value = value + +def _parallel_worker(): + remaining = _remaining + + while True: + target = _parallel_ready.get() + + _results[target._id] = target.build( + {dep.name: _results[dep._id] for dep in target.deps} + ) + + with _lock: + if remaining.value == 1: + _done.set() + continue + + remaining.value -= 1 + for dependent in target._dependents: + if dependent._in_degree == 1: + _parallel_ready.put(dependent) + else: + dependent._in_degree -= 1 + + +def _diamond_worker(): + heap = [] + remaining = _remaining + + while True: + target = _diamond_ready.get() + + while target is not None: + _results[target._id] = target.build( + {dep.name: _results[dep._id] for dep in target.deps} + ) + + next_target = None + with _lock: + if remaining.value == 1: + _done.set() + break + + remaining.value -= 1 + for dependent in target._dependents: + if dependent._in_degree == 1: + heapq.heappush( + heap, (-dependent.work, dependent._id, dependent) + ) + else: + dependent._in_degree -= 1 + + if len(heap) == 1: + next_target = heapq.heappop(heap)[2] + else: + while heap: + _diamond_ready.put(heapq.heappop(heap)[2]) + target = next_target + + +_immortalize(_parallel_ready) +_immortalize(_diamond_ready) +_immortalize(_lock) +_immortalize(_results) +_immortalize(_done) +_immortalize(_remaining) + +for _ in range(_NUM_WORKERS): + threading.Thread(target=_parallel_worker, daemon=True).start() + threading.Thread(target=_diamond_worker, daemon=True).start() def build_all(graph: BuildGraph): targets = graph.targets - remaining = _Remaining(len(targets)) - for id, (name, target) in enumerate(targets.items()): - target._id = id + for target in targets.values(): target._dependents = [] - for target in targets.values(): - if (deps := target.deps): + roots = [] + for id, target in enumerate(targets.values()): + target._id = id + if deps := target.deps: deps.sort(key=lambda dep: dep.name) target._in_degree = len(deps) for dep in deps: dep._dependents.append(target) else: target._in_degree = 0 - _ready.put(target) - - if _ready.qsize() > 1 or (max_fan_out := max((len(target._dependents) for target in targets.values()), default=0)) > 1: - heap = [] - - if max(target._in_degree for target in targets.values()) > _NUM_WORKERS and max_fan_out > _NUM_WORKERS: - def worker(remaining): - target = _ready.get() - - while target is not _SENTINEL: - _results[target._id] = target.build({dep.name: _results[dep._id] for dep in target.deps}) - - next_target = None - with _lock: - if remaining.value == 1: - for _ in range(_NUM_WORKERS_M1): - _ready.put(_SENTINEL) - return - - remaining.value -= 1 - for dependent in target._dependents: - if dependent._in_degree == 1: - heapq.heappush(heap, (-dependent.work, dependent._id, dependent)) - else: - dependent._in_degree -= 1 - - if len(heap) == 1: - next_target = heapq.heappop(heap)[2] - else: - while heap: - _ready.put(heapq.heappop(heap)[2]) - - if next_target is not None: - target = next_target - else: - target = _ready.get() - else: - def worker(remaining): - while (target := _ready.get()) is not _SENTINEL: - _results[target._id] = target.build({dep.name: _results[dep._id] for dep in target.deps}) - - with _lock: - if remaining.value == 1: - for _ in range(_NUM_WORKERS_M1): - _ready.put(_SENTINEL) - return - - remaining.value -= 1 - for dependent in target._dependents: - if dependent._in_degree == 1: - _ready.put(dependent) - else: - dependent._in_degree -= 1 - - for _ in range(_NUM_WORKERS_M1): - threading.Thread(target=worker, args=(remaining,), daemon=True).start() - worker(remaining) - else: - target = _ready.get() + roots.append(target) + if len(roots) == 1 and all( + len(target._dependents) <= 1 for target in targets.values() + ): + target = roots[0] results = _results results[target._id] = target.build({}) - for _ in range(remaining.value - 1): + for _ in range(len(targets) - 1): dependent = target._dependents[0] results[dependent._id] = dependent.build({target.name: results[target._id]}) target = dependent + else: + ready = ( + _diamond_ready + if ( + max(target._in_degree for target in targets.values()) > _NUM_WORKERS + and max(len(target._dependents) for target in targets.values()) + > _NUM_WORKERS + ) + else _parallel_ready + ) + + _remaining.value = len(targets) + _done.clear() + for target in roots: + ready.put(target) + _done.wait() From a220413678a7097a029793cbc40eeaf9fa32e3c8 Mon Sep 17 00:00:00 2001 From: Kevin Newton Date: Sun, 17 May 2026 16:18:36 -0700 Subject: [PATCH 14/15] Again --- submissions/kddnewton.py | 169 ++++++++++++++++++--------------------- 1 file changed, 76 insertions(+), 93 deletions(-) diff --git a/submissions/kddnewton.py b/submissions/kddnewton.py index ad85664..2206527 100644 --- a/submissions/kddnewton.py +++ b/submissions/kddnewton.py @@ -1,144 +1,127 @@ -import ctypes -import heapq -import queue -import threading +from queue import SimpleQueue +from threading import Condition, Event, Lock, Thread from graph import BuildGraph -class _Remaining: - __slots__ = ("value",) +class _BuildGraphState: + __slots__ = ("remaining", "done", "queue") - def __init__(self): - self.value = 0 + def __init__(self, remaining, done, queue): + self.remaining = remaining + self.done = done + self.queue = queue -_NUM_WORKERS = 24 - -_parallel_ready = queue.SimpleQueue() -_diamond_ready = queue.SimpleQueue() -_lock = threading.Lock() -_done = threading.Event() +_NWORKERS = 24 +_lock = Lock() +_condition = Condition() _results = [None] * 20000 -_remaining = _Remaining() - -def _immortalize(obj): - ctypes.c_uint32.from_address(id(obj) + 12).value = 0xFFFFFFFF +_current_state = None +_current_state_id = 0 -def _parallel_worker(): - remaining = _remaining +def _worker(condition, lock, results): + state_id = 0 while True: - target = _parallel_ready.get() - - _results[target._id] = target.build( - {dep.name: _results[dep._id] for dep in target.deps} - ) - - with _lock: - if remaining.value == 1: - _done.set() - continue - - remaining.value -= 1 - for dependent in target._dependents: - if dependent._in_degree == 1: - _parallel_ready.put(dependent) - else: - dependent._in_degree -= 1 - + with condition: + while state_id == _current_state_id: + condition.wait() + state_id = _current_state_id + state = _current_state -def _diamond_worker(): - heap = [] - remaining = _remaining - - while True: - target = _diamond_ready.get() + queue = state.queue + target = queue.get() + completed = False while target is not None: - _results[target._id] = target.build( - {dep.name: _results[dep._id] for dep in target.deps} + results[target._id] = target.build( + {dep.name: results[dep._id] for dep in target.deps} ) next_target = None - with _lock: - if remaining.value == 1: - _done.set() + ready = [] + + with lock: + if state.remaining == 1: + state.done.set() + completed = True break - remaining.value -= 1 + state.remaining -= 1 for dependent in target._dependents: if dependent._in_degree == 1: - heapq.heappush( - heap, (-dependent.work, dependent._id, dependent) - ) + if next_target is None: + next_target = dependent + else: + ready.append(dependent) else: dependent._in_degree -= 1 - if len(heap) == 1: - next_target = heapq.heappop(heap)[2] - else: - while heap: - _diamond_ready.put(heapq.heappop(heap)[2]) - target = next_target + for dependent in ready: + queue.put(dependent) + if next_target is not None: + target = next_target + else: + target = queue.get() -_immortalize(_parallel_ready) -_immortalize(_diamond_ready) -_immortalize(_lock) -_immortalize(_results) -_immortalize(_done) -_immortalize(_remaining) + if completed: + for _ in range(_NWORKERS): + queue.put(None) -for _ in range(_NUM_WORKERS): - threading.Thread(target=_parallel_worker, daemon=True).start() - threading.Thread(target=_diamond_worker, daemon=True).start() + +for _ in range(_NWORKERS): + Thread(target=_worker, args=(_condition, _lock, _results), daemon=True).start() def build_all(graph: BuildGraph): - targets = graph.targets + global _current_state_id, _current_state + + state = _BuildGraphState( + len(targets := graph.targets), + done := Event(), + queue := SimpleQueue(), + ) for target in targets.values(): target._dependents = [] - roots = [] + has_parallelism = False for id, target in enumerate(targets.values()): target._id = id if deps := target.deps: - deps.sort(key=lambda dep: dep.name) target._in_degree = len(deps) for dep in deps: + if not has_parallelism and dep._dependents: + has_parallelism = True dep._dependents.append(target) else: target._in_degree = 0 - roots.append(target) + if not has_parallelism and not queue.empty(): + has_parallelism = True + queue.put(target) - if len(roots) == 1 and all( - len(target._dependents) <= 1 for target in targets.values() - ): - target = roots[0] - results = _results - results[target._id] = target.build({}) + if not has_parallelism: + target = queue.get() + (results := _results)[target._id] = target.build({}) for _ in range(len(targets) - 1): dependent = target._dependents[0] results[dependent._id] = dependent.build({target.name: results[target._id]}) target = dependent else: - ready = ( - _diamond_ready - if ( - max(target._in_degree for target in targets.values()) > _NUM_WORKERS - and max(len(target._dependents) for target in targets.values()) - > _NUM_WORKERS - ) - else _parallel_ready - ) - - _remaining.value = len(targets) - _done.clear() - for target in roots: - ready.put(target) - _done.wait() + for target in targets.values(): + if len(target._dependents) > 1: + target._dependents.sort( + key=lambda dependent: dependent.work, reverse=True + ) + + with _condition: + _current_state = state + _current_state_id += 1 + _condition.notify_all() + + done.wait() From 130a41f9fe1e05bc8dd4d8b2622743c1080577a4 Mon Sep 17 00:00:00 2001 From: Kevin Newton Date: Sun, 17 May 2026 16:40:27 -0700 Subject: [PATCH 15/15] Again --- submissions/kddnewton.py | 23 +++++++---------------- 1 file changed, 7 insertions(+), 16 deletions(-) diff --git a/submissions/kddnewton.py b/submissions/kddnewton.py index 2206527..e98d27f 100644 --- a/submissions/kddnewton.py +++ b/submissions/kddnewton.py @@ -41,7 +41,6 @@ def _worker(condition, lock, results): {dep.name: results[dep._id] for dep in target.deps} ) - next_target = None ready = [] with lock: @@ -53,18 +52,16 @@ def _worker(condition, lock, results): state.remaining -= 1 for dependent in target._dependents: if dependent._in_degree == 1: - if next_target is None: - next_target = dependent - else: - ready.append(dependent) + ready.append(dependent) else: dependent._in_degree -= 1 - for dependent in ready: - queue.put(dependent) - - if next_target is not None: - target = next_target + if ready: + if len(ready) >= _NWORKERS: + ready.sort(key=lambda d: d.work) + target = ready.pop() + for dependent in reversed(ready): + queue.put(dependent) else: target = queue.get() @@ -113,12 +110,6 @@ def build_all(graph: BuildGraph): results[dependent._id] = dependent.build({target.name: results[target._id]}) target = dependent else: - for target in targets.values(): - if len(target._dependents) > 1: - target._dependents.sort( - key=lambda dependent: dependent.work, reverse=True - ) - with _condition: _current_state = state _current_state_id += 1