From 1596f74d32763c7e8ad4e893556e35d494c1d43e Mon Sep 17 00:00:00 2001 From: Matt Page Date: Sat, 16 May 2026 10:50:40 -0700 Subject: [PATCH 1/7] Initial solution --- submissions/mpage.py | 95 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 95 insertions(+) create mode 100644 submissions/mpage.py diff --git a/submissions/mpage.py b/submissions/mpage.py new file mode 100644 index 0000000..a15890b --- /dev/null +++ b/submissions/mpage.py @@ -0,0 +1,95 @@ +"""Build graph simulator challenge — parallel scheduler using threads.""" + +import os +import threading +from collections import deque +from concurrent.futures import ThreadPoolExecutor + +from graph import BuildGraph, Target + + +NUM_WORKERS = os.cpu_count() or 24 + +# Strategy: +# +# - SimpleQueue to feed workers +# - Attempt to guess graph shape -> use different strategies? +# - Do setup in parallel? + +def build_all(graph: BuildGraph) -> dict[str, bytes]: + results: dict[str, bytes] = {} + + # Assign each target an id + idx_ctr = 0 + for target in graph.targets.values(): + target.index = idx_ctr + idx_ctr += 1 + + # Precompute reverse deps and in-degree + num_targets = len(graph.targets) + dependents: list[list[Target]] = [[] for _ in range(num_targets)] + for name, target in graph.targets.items(): + target.in_degree = len(target.deps) + for dep in target.deps: + dependents[dep.index].append(target) + + # Track remaining count for completion signaling + remaining = len(graph.targets) + lock = threading.Lock() + done = threading.Event() + queue: deque[Target] = deque() + queue_not_empty = threading.Condition(lock) + + # Seed with zero-dependency targets + for target in graph.targets.values(): + if target.in_degree == 0: + queue.append(target) + + def worker(): + nonlocal remaining + while True: + with queue_not_empty: + while not queue and not done.is_set(): + queue_not_empty.wait() + if done.is_set() and not queue: + return + target = queue.popleft() + + target_index = target.index + dep_results = {d.name: results[d.index] for d in target.deps} + result = target.build(dep_results) + results[target_index] = result + + newly_ready = [] + with lock: + for dep in dependents[target_index]: + dep.in_degree -= 1 + if dep.in_degree == 0: + newly_ready.append(dep) + remaining -= 1 + if remaining == 0: + done.set() + + if newly_ready: + with queue_not_empty: + queue.extend(newly_ready) + if len(newly_ready) > 1: + queue_not_empty.notify_all() + else: + queue_not_empty.notify() + + if done.is_set(): + with queue_not_empty: + queue_not_empty.notify_all() + return + + threads = [] + for _ in range(NUM_WORKERS): + t = threading.Thread(target=worker) + t.start() + threads.append(t) + + for t in threads: + t.join() + + return results From dcacfd7461de8325d5a26a8f1883670a3f796074 Mon Sep 17 00:00:00 2001 From: Matt Page Date: Sat, 16 May 2026 10:56:45 -0700 Subject: [PATCH 2/7] Micro optimizations - Daemon threads - Make the main thread a worker --- submissions/mpage.py | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/submissions/mpage.py b/submissions/mpage.py index a15890b..617a046 100644 --- a/submissions/mpage.py +++ b/submissions/mpage.py @@ -83,13 +83,10 @@ def worker(): queue_not_empty.notify_all() return - threads = [] for _ in range(NUM_WORKERS): - t = threading.Thread(target=worker) + t = threading.Thread(target=worker, daemon=True) t.start() - threads.append(t) - for t in threads: - t.join() + worker() return results From b2e3f9ec6cc6539635f8d29e9c920eecfdf1bc4b Mon Sep 17 00:00:00 2001 From: Matt Page Date: Sat, 16 May 2026 13:06:57 -0700 Subject: [PATCH 3/7] More small optimizations --- submissions/mpage.py | 137 +++++++++++++++++++++++++------------------ 1 file changed, 80 insertions(+), 57 deletions(-) diff --git a/submissions/mpage.py b/submissions/mpage.py index 617a046..a314fbd 100644 --- a/submissions/mpage.py +++ b/submissions/mpage.py @@ -2,91 +2,114 @@ import os import threading -from collections import deque -from concurrent.futures import ThreadPoolExecutor +from queue import SimpleQueue from graph import BuildGraph, Target NUM_WORKERS = os.cpu_count() or 24 +_SENTINEL = None -# Strategy: -# -# - SimpleQueue to feed workers -# - Attempt to guess graph shape -> use different strategies? -# - Do setup in parallel? def build_all(graph: BuildGraph) -> dict[str, bytes]: - results: dict[str, bytes] = {} + targets = graph.targets + num_targets = len(targets) - # Assign each target an id - idx_ctr = 0 - for target in graph.targets.values(): - target.index = idx_ctr - idx_ctr += 1 + idx = 0 + for target in targets.values(): + target.index = idx + idx += 1 + + results: list[bytes | None] = [None] * num_targets - # Precompute reverse deps and in-degree - num_targets = len(graph.targets) dependents: list[list[Target]] = [[] for _ in range(num_targets)] - for name, target in graph.targets.items(): + num_sources = 0 + for target in targets.values(): target.in_degree = len(target.deps) + if target.in_degree == 0: + num_sources += 1 for dep in target.deps: dependents[dep.index].append(target) - # Track remaining count for completion signaling - remaining = len(graph.targets) + max_fan_out = max((len(d) for d in dependents), default=0) + + # Sort dependents heaviest-first for better load balancing + for dep_list in dependents: + if len(dep_list) > 1: + dep_list.sort(key=lambda t: t.work, reverse=True) + + # Sequential fast path for chain-like graphs + if max_fan_out <= 1 and num_sources <= 1: + target = None + for t in targets.values(): + if t.in_degree == 0: + target = t + break + + while target is not None: + tidx = target.index + dep_results = {d.name: results[d.index] for d in target.deps} + results[tidx] = target.build(dep_results) + next_target = None + for dep in dependents[tidx]: + dep.in_degree -= 1 + if dep.in_degree == 0: + next_target = dep + target = next_target + + return results + + # Parallel path + remaining = num_targets lock = threading.Lock() - done = threading.Event() - queue: deque[Target] = deque() - queue_not_empty = threading.Condition(lock) + queue: SimpleQueue[Target | None] = SimpleQueue() - # Seed with zero-dependency targets - for target in graph.targets.values(): + for target in targets.values(): if target.in_degree == 0: - queue.append(target) + queue.put(target) def worker(): nonlocal remaining - while True: - with queue_not_empty: - while not queue and not done.is_set(): - queue_not_empty.wait() - if done.is_set() and not queue: - return - target = queue.popleft() - - target_index = target.index - dep_results = {d.name: results[d.index] for d in target.deps} - result = target.build(dep_results) - results[target_index] = result - - newly_ready = [] - with lock: - for dep in dependents[target_index]: + _results = results + _dependents = dependents + _lock = lock + _queue = queue + _NW = NUM_WORKERS + + target = _queue.get() + while target is not _SENTINEL: + tidx = target.index + dep_results = {d.name: _results[d.index] for d in target.deps} + _results[tidx] = target.build(dep_results) + + next_target = None + with _lock: + for dep in _dependents[tidx]: dep.in_degree -= 1 if dep.in_degree == 0: - newly_ready.append(dep) + if next_target is None: + next_target = dep + else: + _queue.put(dep) remaining -= 1 if remaining == 0: - done.set() - - if newly_ready: - with queue_not_empty: - queue.extend(newly_ready) - if len(newly_ready) > 1: - queue_not_empty.notify_all() - else: - queue_not_empty.notify() - - if done.is_set(): - with queue_not_empty: - queue_not_empty.notify_all() - return - - for _ in range(NUM_WORKERS): + for _ in range(_NW): + _queue.put(_SENTINEL) + + if next_target is not None: + target = next_target + else: + target = _queue.get() + + threads = [] + for _ in range(NUM_WORKERS - 1): t = threading.Thread(target=worker, daemon=True) t.start() + threads.append(t) worker() + for t in threads: + t.join() + return results From d0727157f578b133f60b1938cdc1086f68b46996 Mon Sep 17 00:00:00 2001 From: Matt Page Date: Sat, 16 May 2026 13:39:06 -0700 Subject: [PATCH 4/7] Don't wait for workers --- submissions/mpage.py | 5 ----- 1 file changed, 5 deletions(-) diff --git a/submissions/mpage.py b/submissions/mpage.py index a314fbd..968ad01 100644 --- a/submissions/mpage.py +++ b/submissions/mpage.py @@ -101,15 +101,10 @@ def worker(): else: target = _queue.get() - threads = [] for _ in range(NUM_WORKERS - 1): t = threading.Thread(target=worker, daemon=True) t.start() - threads.append(t) worker() - for t in threads: - t.join() - return results From 7cdcaa6e4ed69627252c17c08026b2951c1ad228 Mon Sep 17 00:00:00 2001 From: Matt Page Date: Sat, 16 May 2026 15:41:54 -0700 Subject: [PATCH 5/7] Atomics --- submissions/mpage.py | 139 ++++++++++++++++++++++++++++++++++++------- 1 file changed, 116 insertions(+), 23 deletions(-) diff --git a/submissions/mpage.py b/submissions/mpage.py index 968ad01..8c7f3b7 100644 --- a/submissions/mpage.py +++ b/submissions/mpage.py @@ -1,5 +1,7 @@ """Build graph simulator challenge — parallel scheduler using threads.""" +import ctypes +import mmap import os import threading from queue import SimpleQueue @@ -11,6 +13,97 @@ _SENTINEL = None +# --------------------------------------------------------------------------- +# AtomicInt: a class with a single __slots__ field storing a raw int64. +# atomic_fetch_sub: a Python function whose vectorcall is patched with +# hand-assembled x86-64 machine code implementing lock xadd. +# --------------------------------------------------------------------------- + +class AtomicInt: + __slots__ = ('value',) + + +def _setup_atomics(): + """Patch AtomicInt's slot to store raw int64 and build atomic_fetch_sub.""" + + # -- Step 1: Find PyMemberDef for 'value' slot and patch to T_LONGLONG -- + desc = AtomicInt.__dict__['value'] + desc_addr = id(desc) + + # CPython 3.14t: PyObject_HEAD(32) + d_type(8) + d_name(8) + d_qualname(8) = desc+56 + d_member = ctypes.c_void_p.from_address(desc_addr + 56).value + + slot_offset = ctypes.c_ssize_t.from_address(d_member + 16).value + + # Patch type from T_OBJECT_EX (16) to T_LONGLONG (17) + ctypes.c_int.from_address(d_member + 8).value = 17 + + # Now AtomicInt().__init__ can do self.value = n to write raw int64 + + # -- Step 2: Get C function addresses -- + pylong_aslong = ctypes.cast(ctypes.pythonapi.PyLong_AsLong, ctypes.c_void_p).value + pylong_fromlong = ctypes.cast(ctypes.pythonapi.PyLong_FromLong, ctypes.c_void_p).value + + # CPython 3.14t: vectorcall is at offset 152 in PyFunctionObject + vc_offset = 152 + + # -- Step 4: Assemble x86-64 machine code for atomic_fetch_sub -- + # + # vectorcall(callable=rdi, args=rsi, nargsf=rdx, kwnames=rcx) + # args[0] = AtomicInt, args[1] = Python int amount + # + # push rbx ; save callee-saved, align stack to 16 + # mov rbx, [rsi] ; rbx = args[0] (AtomicInt object) + # mov rdi, [rsi + 8] ; rdi = args[1] (Python int amount) + # movabs rax, + # call rax ; rax = C long value of amount + # neg rax ; negate (xadd adds, we want sub) + # lock xadd [rbx + offset], rax ; atomic fetch-sub; old value -> rax + # mov rdi, rax ; arg for PyLong_FromLong + # movabs rax, + # call rax ; rax = new Python int (old value) + # pop rbx + # ret + + code = bytearray() + code += b'\x53' # push rbx + code += b'\x48\x8B\x1E' # mov rbx, [rsi] + code += b'\x48\x8B\x7E\x08' # mov rdi, [rsi+8] + code += b'\x48\xB8' + pylong_aslong.to_bytes(8, 'little') # movabs rax, addr + code += b'\xFF\xD0' # call rax + code += b'\x48\xF7\xD8' # neg rax + if slot_offset < 128: + code += b'\xF0\x48\x0F\xC1\x43' + bytes([slot_offset]) # lock xadd [rbx+disp8], rax + else: + code += b'\xF0\x48\x0F\xC1\x83' + slot_offset.to_bytes(4, 'little') # lock xadd [rbx+disp32], rax + code += b'\x48\x89\xC7' # mov rdi, rax + code += b'\x48\xB8' + pylong_fromlong.to_bytes(8, 'little') # movabs rax, addr + code += b'\xFF\xD0' # call rax + code += b'\x5B' # pop rbx + code += b'\xC3' # ret + + # -- Step 5: Allocate executable memory and write code -- + exec_page = mmap.mmap(-1, mmap.PAGESIZE, + prot=mmap.PROT_READ | mmap.PROT_WRITE | mmap.PROT_EXEC) + exec_page.write(bytes(code)) + exec_addr = ctypes.addressof(ctypes.c_char.from_buffer(exec_page)) + + # -- Step 6: Create Python function and patch its vectorcall -- + def atomic_fetch_sub(atom, amount): + raise RuntimeError("vectorcall not patched") + + ctypes.c_void_p.from_address(id(atomic_fetch_sub) + vc_offset).value = exec_addr + + return atomic_fetch_sub, exec_page + + +atomic_fetch_sub, _exec_page = _setup_atomics() + + +# --------------------------------------------------------------------------- +# Scheduler +# --------------------------------------------------------------------------- + def build_all(graph: BuildGraph) -> dict[str, bytes]: targets = graph.targets num_targets = len(targets) @@ -25,8 +118,9 @@ def build_all(graph: BuildGraph) -> dict[str, bytes]: dependents: list[list[Target]] = [[] for _ in range(num_targets)] num_sources = 0 for target in targets.values(): - target.in_degree = len(target.deps) - if target.in_degree == 0: + target.in_degree = AtomicInt() + target.in_degree.value = len(target.deps) + if target.in_degree.value == 0: num_sources += 1 for dep in target.deps: dependents[dep.index].append(target) @@ -42,7 +136,7 @@ def build_all(graph: BuildGraph) -> dict[str, bytes]: if max_fan_out <= 1 and num_sources <= 1: target = None for t in targets.values(): - if t.in_degree == 0: + if t.in_degree.value == 0: target = t break @@ -52,29 +146,30 @@ def build_all(graph: BuildGraph) -> dict[str, bytes]: results[tidx] = target.build(dep_results) next_target = None for dep in dependents[tidx]: - dep.in_degree -= 1 - if dep.in_degree == 0: + if atomic_fetch_sub(dep.in_degree, 1) == 1: next_target = dep target = next_target return results - # Parallel path - remaining = num_targets - lock = threading.Lock() + # Parallel path — lock-free using atomic_fetch_sub + remaining = AtomicInt() + remaining.value = num_targets queue: SimpleQueue[Target | None] = SimpleQueue() for target in targets.values(): - if target.in_degree == 0: + if target.in_degree.value == 0: queue.put(target) + _fetch_sub = atomic_fetch_sub + def worker(): - nonlocal remaining _results = results _dependents = dependents - _lock = lock _queue = queue _NW = NUM_WORKERS + _fsub = _fetch_sub + _remaining = remaining target = _queue.get() while target is not _SENTINEL: @@ -83,18 +178,16 @@ def worker(): _results[tidx] = target.build(dep_results) next_target = None - with _lock: - for dep in _dependents[tidx]: - dep.in_degree -= 1 - if dep.in_degree == 0: - if next_target is None: - next_target = dep - else: - _queue.put(dep) - remaining -= 1 - if remaining == 0: - for _ in range(_NW): - _queue.put(_SENTINEL) + for dep in _dependents[tidx]: + if _fsub(dep.in_degree, 1) == 1: + if next_target is None: + next_target = dep + else: + _queue.put(dep) + + if _fsub(_remaining, 1) == 1: + for _ in range(_NW): + _queue.put(_SENTINEL) if next_target is not None: target = next_target From 6f3588a196cbedba7bb3aa35e79f60699e2ba28d Mon Sep 17 00:00:00 2001 From: Matt Page Date: Sat, 16 May 2026 15:55:22 -0700 Subject: [PATCH 6/7] Try atomics --- submissions/mpage.py | 117 +++++++++++++++++-------------------------- 1 file changed, 47 insertions(+), 70 deletions(-) diff --git a/submissions/mpage.py b/submissions/mpage.py index 8c7f3b7..f279a3d 100644 --- a/submissions/mpage.py +++ b/submissions/mpage.py @@ -14,90 +14,72 @@ # --------------------------------------------------------------------------- -# AtomicInt: a class with a single __slots__ field storing a raw int64. -# atomic_fetch_sub: a Python function whose vectorcall is patched with -# hand-assembled x86-64 machine code implementing lock xadd. +# AtomicInt with a raw uint64 slot and a fetch_sub method whose vectorcall +# is patched with x86-64 assembly: lock xadd, cmp, cmovz → True/False. # --------------------------------------------------------------------------- class AtomicInt: __slots__ = ('value',) + def __init__(self, val=0): + self.value = val -def _setup_atomics(): - """Patch AtomicInt's slot to store raw int64 and build atomic_fetch_sub.""" - - # -- Step 1: Find PyMemberDef for 'value' slot and patch to T_LONGLONG -- - desc = AtomicInt.__dict__['value'] - desc_addr = id(desc) + def fetch_sub(self): + raise RuntimeError("vectorcall not patched") - # CPython 3.14t: PyObject_HEAD(32) + d_type(8) + d_name(8) + d_qualname(8) = desc+56 - d_member = ctypes.c_void_p.from_address(desc_addr + 56).value +def _setup_atomics(): + # Patch 'value' slot from T_OBJECT_EX to T_LONGLONG so it stores raw int64. + # CPython 3.14t: d_member pointer is at descriptor + 56. + desc = AtomicInt.__dict__['value'] + d_member = ctypes.c_void_p.from_address(id(desc) + 56).value slot_offset = ctypes.c_ssize_t.from_address(d_member + 16).value + ctypes.c_int.from_address(d_member + 8).value = 17 # T_LONGLONG - # Patch type from T_OBJECT_EX (16) to T_LONGLONG (17) - ctypes.c_int.from_address(d_member + 8).value = 17 - - # Now AtomicInt().__init__ can do self.value = n to write raw int64 + # Get addresses of Py_True and Py_False (immortal, no refcounting needed). + py_false = id(False) + py_true = id(True) - # -- Step 2: Get C function addresses -- - pylong_aslong = ctypes.cast(ctypes.pythonapi.PyLong_AsLong, ctypes.c_void_p).value - pylong_fromlong = ctypes.cast(ctypes.pythonapi.PyLong_FromLong, ctypes.c_void_p).value - - # CPython 3.14t: vectorcall is at offset 152 in PyFunctionObject - vc_offset = 152 - - # -- Step 4: Assemble x86-64 machine code for atomic_fetch_sub -- + # Assemble fetch_sub: decrements [self + slot_offset] by 1, + # returns True if old value was 1 (counter hit zero), False otherwise. # # vectorcall(callable=rdi, args=rsi, nargsf=rdx, kwnames=rcx) - # args[0] = AtomicInt, args[1] = Python int amount + # args[0] = self (AtomicInt) # - # push rbx ; save callee-saved, align stack to 16 - # mov rbx, [rsi] ; rbx = args[0] (AtomicInt object) - # mov rdi, [rsi + 8] ; rdi = args[1] (Python int amount) - # movabs rax, - # call rax ; rax = C long value of amount - # neg rax ; negate (xadd adds, we want sub) - # lock xadd [rbx + offset], rax ; atomic fetch-sub; old value -> rax - # mov rdi, rax ; arg for PyLong_FromLong - # movabs rax, - # call rax ; rax = new Python int (old value) - # pop rbx - # ret - + # mov rax, [rsi] ; rax = self + # mov rcx, -1 ; add -1 = subtract 1 + # lock xadd [rax + offset], rcx ; rcx = old value, [slot] -= 1 + # cmp rcx, 1 ; was old value 1? + # movabs rax, + # movabs rcx, + # cmovz rax, rcx ; if old == 1: return True + # ret code = bytearray() - code += b'\x53' # push rbx - code += b'\x48\x8B\x1E' # mov rbx, [rsi] - code += b'\x48\x8B\x7E\x08' # mov rdi, [rsi+8] - code += b'\x48\xB8' + pylong_aslong.to_bytes(8, 'little') # movabs rax, addr - code += b'\xFF\xD0' # call rax - code += b'\x48\xF7\xD8' # neg rax + code += b'\x48\x8B\x06' # mov rax, [rsi] + code += b'\x48\xC7\xC1\xFF\xFF\xFF\xFF' # mov rcx, -1 if slot_offset < 128: - code += b'\xF0\x48\x0F\xC1\x43' + bytes([slot_offset]) # lock xadd [rbx+disp8], rax + code += b'\xF0\x48\x0F\xC1\x48' + bytes([slot_offset]) # lock xadd [rax+disp8], rcx else: - code += b'\xF0\x48\x0F\xC1\x83' + slot_offset.to_bytes(4, 'little') # lock xadd [rbx+disp32], rax - code += b'\x48\x89\xC7' # mov rdi, rax - code += b'\x48\xB8' + pylong_fromlong.to_bytes(8, 'little') # movabs rax, addr - code += b'\xFF\xD0' # call rax - code += b'\x5B' # pop rbx - code += b'\xC3' # ret - - # -- Step 5: Allocate executable memory and write code -- + code += b'\xF0\x48\x0F\xC1\x88' + slot_offset.to_bytes(4, 'little') + code += b'\x48\x83\xF9\x01' # cmp rcx, 1 + code += b'\x48\xB8' + py_false.to_bytes(8, 'little') # movabs rax, Py_False + code += b'\x48\xB9' + py_true.to_bytes(8, 'little') # movabs rcx, Py_True + code += b'\x48\x0F\x44\xC1' # cmovz rax, rcx + code += b'\xC3' # ret + exec_page = mmap.mmap(-1, mmap.PAGESIZE, prot=mmap.PROT_READ | mmap.PROT_WRITE | mmap.PROT_EXEC) exec_page.write(bytes(code)) exec_addr = ctypes.addressof(ctypes.c_char.from_buffer(exec_page)) - # -- Step 6: Create Python function and patch its vectorcall -- - def atomic_fetch_sub(atom, amount): - raise RuntimeError("vectorcall not patched") - - ctypes.c_void_p.from_address(id(atomic_fetch_sub) + vc_offset).value = exec_addr + # Patch fetch_sub's vectorcall (offset 152 in PyFunctionObject on 3.14t). + func = AtomicInt.__dict__['fetch_sub'] + ctypes.c_void_p.from_address(id(func) + 152).value = exec_addr - return atomic_fetch_sub, exec_page + return exec_page -atomic_fetch_sub, _exec_page = _setup_atomics() +_exec_page = _setup_atomics() # --------------------------------------------------------------------------- @@ -118,8 +100,7 @@ def build_all(graph: BuildGraph) -> dict[str, bytes]: dependents: list[list[Target]] = [[] for _ in range(num_targets)] num_sources = 0 for target in targets.values(): - target.in_degree = AtomicInt() - target.in_degree.value = len(target.deps) + target.in_degree = AtomicInt(len(target.deps)) if target.in_degree.value == 0: num_sources += 1 for dep in target.deps: @@ -146,29 +127,25 @@ def build_all(graph: BuildGraph) -> dict[str, bytes]: results[tidx] = target.build(dep_results) next_target = None for dep in dependents[tidx]: - if atomic_fetch_sub(dep.in_degree, 1) == 1: + if dep.in_degree.fetch_sub(): next_target = dep target = next_target return results - # Parallel path — lock-free using atomic_fetch_sub - remaining = AtomicInt() - remaining.value = num_targets + # Parallel path — lock-free using atomic fetch_sub + remaining = AtomicInt(num_targets) queue: SimpleQueue[Target | None] = SimpleQueue() for target in targets.values(): if target.in_degree.value == 0: queue.put(target) - _fetch_sub = atomic_fetch_sub - def worker(): _results = results _dependents = dependents _queue = queue _NW = NUM_WORKERS - _fsub = _fetch_sub _remaining = remaining target = _queue.get() @@ -179,13 +156,13 @@ def worker(): next_target = None for dep in _dependents[tidx]: - if _fsub(dep.in_degree, 1) == 1: + if dep.in_degree.fetch_sub(): if next_target is None: next_target = dep else: _queue.put(dep) - if _fsub(_remaining, 1) == 1: + if _remaining.fetch_sub(): for _ in range(_NW): _queue.put(_SENTINEL) From 9753b8417d02d6de5365ad1d4bb9823a2801746f Mon Sep 17 00:00:00 2001 From: Matt Page Date: Sat, 16 May 2026 19:31:08 -0700 Subject: [PATCH 7/7] Use a persistent worker pool --- submissions/mpage.py | 23 +++++++++++++++++++++-- 1 file changed, 21 insertions(+), 2 deletions(-) diff --git a/submissions/mpage.py b/submissions/mpage.py index f279a3d..23a7e7f 100644 --- a/submissions/mpage.py +++ b/submissions/mpage.py @@ -82,6 +82,23 @@ def _setup_atomics(): _exec_page = _setup_atomics() +# --------------------------------------------------------------------------- +# Persistent thread pool +# --------------------------------------------------------------------------- + +_task_queue: SimpleQueue = SimpleQueue() + +def _pool_worker(): + _queue = _task_queue + while True: + fn = _queue.get() + fn() + +for _ in range(NUM_WORKERS - 1): + _t = threading.Thread(target=_pool_worker, daemon=True) + _t.start() + + # --------------------------------------------------------------------------- # Scheduler # --------------------------------------------------------------------------- @@ -136,6 +153,7 @@ def build_all(graph: BuildGraph) -> dict[str, bytes]: # Parallel path — lock-free using atomic fetch_sub remaining = AtomicInt(num_targets) queue: SimpleQueue[Target | None] = SimpleQueue() + done = threading.Event() for target in targets.values(): if target.in_degree.value == 0: @@ -165,6 +183,7 @@ def worker(): if _remaining.fetch_sub(): for _ in range(_NW): _queue.put(_SENTINEL) + done.set() if next_target is not None: target = next_target @@ -172,9 +191,9 @@ def worker(): target = _queue.get() for _ in range(NUM_WORKERS - 1): - t = threading.Thread(target=worker, daemon=True) - t.start() + _task_queue.put(worker) worker() + done.wait() return results