From 5890bb0a1319084b629d8f127de919c759c0208c Mon Sep 17 00:00:00 2001 From: Adam Wildavsky Date: Mon, 27 Jul 2026 05:07:09 +0200 Subject: [PATCH 1/9] Cap parallel workers under the wasm32 2 GiB heap. Uncapped auto thread counts allocate one Large TT per worker and OOM large WASM batches; clamp to a fixed memory budget instead. Co-authored-by: Cursor --- library/src/system/parallel_boards.cpp | 27 ++++++++++++++++- library/src/system/parallel_boards.hpp | 17 +++++++++++ library/tests/system/worker_count_test.cpp | 35 ++++++++++++++++++++++ 3 files changed, 78 insertions(+), 1 deletion(-) diff --git a/library/src/system/parallel_boards.cpp b/library/src/system/parallel_boards.cpp index c330a14c..014882aa 100644 --- a/library/src/system/parallel_boards.cpp +++ b/library/src/system/parallel_boards.cpp @@ -14,9 +14,22 @@ #include #include +#include #include +auto clamp_workers_to_memory_budget( + const int workers, + const int budget_mb, + const int per_worker_mb) -> int +{ + const int safe_workers = std::max(1, workers); + if (budget_mb <= 0 || per_worker_mb <= 0) + return safe_workers; + return std::max(1, std::min(safe_workers, budget_mb / per_worker_mb)); +} + + auto resolve_worker_count( const int max_threads, const int count) -> int @@ -27,7 +40,19 @@ auto resolve_worker_count( const unsigned hw = std::thread::hardware_concurrency(); workers = hw > 0 ? static_cast(hw) : 1; } - return std::max(1, std::min(workers, count)); + workers = std::max(1, std::min(workers, count)); + + // Parallel workers each keep a Large TT (via SolverContext). On wasm32 / + // other ILP32 targets the heap cannot grow past ~2 GiB; uncapped auto + // (= HW concurrency) OOMs large multi-board batches. Leave headroom under + // that ceiling for the main module, board vectors, and growth slack. +#if defined(__EMSCRIPTEN__) || UINTPTR_MAX == 0xffffffffu + constexpr int kHeapBudgetMB = 1400; + constexpr int kPerWorkerMB = THREADMEM_LARGE_DEF_MB + 24; + workers = clamp_workers_to_memory_budget(workers, kHeapBudgetMB, kPerWorkerMB); +#endif + + return workers; } diff --git a/library/src/system/parallel_boards.hpp b/library/src/system/parallel_boards.hpp index 29ae953b..00bfd339 100644 --- a/library/src/system/parallel_boards.hpp +++ b/library/src/system/parallel_boards.hpp @@ -22,6 +22,23 @@ */ auto resolve_worker_count(int max_threads, int count) -> int; +/** + * @brief Clamp a worker count to a memory budget (MB). + * + * Each parallel worker owns a SolverContext with a Large transposition table + * plus stack; on wasm32 / ILP32 the process heap tops out near 2 GiB. Tests and + * resolve_worker_count share this helper so the budget math stays in one place. + * + * @param workers Requested workers (values < 1 become 1) + * @param budget_mb Total MB available for worker TT/stack footprints + * @param per_worker_mb Assumed MB cost of one worker + * @return workers clamped to max(1, budget_mb / per_worker_mb) + */ +auto clamp_workers_to_memory_budget( + int workers, + int budget_mb, + int per_worker_mb) -> int; + /** * @brief Process boards [0, count) with work-stealing parallelism. * diff --git a/library/tests/system/worker_count_test.cpp b/library/tests/system/worker_count_test.cpp index dfe6aecd..755d08ae 100644 --- a/library/tests/system/worker_count_test.cpp +++ b/library/tests/system/worker_count_test.cpp @@ -6,8 +6,10 @@ #include #include +#include #include +#include #include namespace @@ -47,3 +49,36 @@ TEST(ResolveWorkerCount, SingleItemAlwaysOneWorker) EXPECT_EQ(resolve_worker_count(16, 1), 1); EXPECT_EQ(resolve_worker_count(1, 1), 1); } + +TEST(ClampWorkersToMemoryBudget, CapsByBudgetPerWorker) +{ + EXPECT_EQ(clamp_workers_to_memory_budget(18, 1400, 95 + 24), 11); + EXPECT_EQ(clamp_workers_to_memory_budget(4, 1400, 95 + 24), 4); + EXPECT_EQ(clamp_workers_to_memory_budget(0, 1400, 119), 1); + EXPECT_EQ(clamp_workers_to_memory_budget(8, 100, 200), 1); +} + +TEST(ClampWorkersToMemoryBudget, PlatformCapMatchesWasmBudgetConstants) +{ + // Document the wasm32 / ILP32 budget used by resolve_worker_count so a + // quiet change to THREADMEM_* cannot silently re-OOM large WASM batches. + constexpr int kHeapBudgetMB = 1400; + constexpr int kPerWorkerMB = THREADMEM_LARGE_DEF_MB + 24; + constexpr int kExpectedCap = kHeapBudgetMB / kPerWorkerMB; + static_assert(kExpectedCap >= 1); + EXPECT_EQ(kExpectedCap, 11); + EXPECT_EQ( + clamp_workers_to_memory_budget(64, kHeapBudgetMB, kPerWorkerMB), + kExpectedCap); +} + +#if defined(__EMSCRIPTEN__) || UINTPTR_MAX == 0xffffffffu +TEST(ResolveWorkerCount, AddressSpaceCapLimitsAutoWorkers) +{ + constexpr int kHeapBudgetMB = 1400; + constexpr int kPerWorkerMB = THREADMEM_LARGE_DEF_MB + 24; + const int mem_cap = kHeapBudgetMB / kPerWorkerMB; + EXPECT_EQ(resolve_worker_count(0, 5000), std::min(auto_workers(5000), mem_cap)); + EXPECT_EQ(resolve_worker_count(64, 5000), mem_cap); +} +#endif From c7aa8a2036e0382c1a4131cb8f5fff0231d36608 Mon Sep 17 00:00:00 2001 From: Adam Wildavsky Date: Mon, 27 Jul 2026 05:14:30 +0200 Subject: [PATCH 2/9] Limit the worker memory cap to Emscripten builds only. Keep native thread counts uncapped so the WASM heap guard does not change host performance. Co-authored-by: Cursor --- library/src/system/parallel_boards.cpp | 11 ++++++----- library/src/system/parallel_boards.hpp | 2 +- library/tests/system/worker_count_test.cpp | 17 ++++++++++++----- 3 files changed, 19 insertions(+), 11 deletions(-) diff --git a/library/src/system/parallel_boards.cpp b/library/src/system/parallel_boards.cpp index 014882aa..93a84cdb 100644 --- a/library/src/system/parallel_boards.cpp +++ b/library/src/system/parallel_boards.cpp @@ -42,11 +42,12 @@ auto resolve_worker_count( } workers = std::max(1, std::min(workers, count)); - // Parallel workers each keep a Large TT (via SolverContext). On wasm32 / - // other ILP32 targets the heap cannot grow past ~2 GiB; uncapped auto - // (= HW concurrency) OOMs large multi-board batches. Leave headroom under - // that ceiling for the main module, board vectors, and growth slack. -#if defined(__EMSCRIPTEN__) || UINTPTR_MAX == 0xffffffffu + // Parallel workers each keep a Large TT (via SolverContext). Under + // Emscripten the wasm32 heap cannot grow past ~2 GiB; uncapped auto + // (= HW concurrency) OOMs large multi-board batches. Native builds keep + // the uncapped count. Leave headroom under the WASM ceiling for the main + // module, board vectors, and growth slack. +#if defined(__EMSCRIPTEN__) constexpr int kHeapBudgetMB = 1400; constexpr int kPerWorkerMB = THREADMEM_LARGE_DEF_MB + 24; workers = clamp_workers_to_memory_budget(workers, kHeapBudgetMB, kPerWorkerMB); diff --git a/library/src/system/parallel_boards.hpp b/library/src/system/parallel_boards.hpp index 00bfd339..e39bed89 100644 --- a/library/src/system/parallel_boards.hpp +++ b/library/src/system/parallel_boards.hpp @@ -26,7 +26,7 @@ auto resolve_worker_count(int max_threads, int count) -> int; * @brief Clamp a worker count to a memory budget (MB). * * Each parallel worker owns a SolverContext with a Large transposition table - * plus stack; on wasm32 / ILP32 the process heap tops out near 2 GiB. Tests and + * plus stack; under Emscripten the wasm32 heap tops out near 2 GiB. Tests and * resolve_worker_count share this helper so the budget math stays in one place. * * @param workers Requested workers (values < 1 become 1) diff --git a/library/tests/system/worker_count_test.cpp b/library/tests/system/worker_count_test.cpp index 755d08ae..7929934c 100644 --- a/library/tests/system/worker_count_test.cpp +++ b/library/tests/system/worker_count_test.cpp @@ -6,7 +6,6 @@ #include #include -#include #include #include @@ -60,8 +59,8 @@ TEST(ClampWorkersToMemoryBudget, CapsByBudgetPerWorker) TEST(ClampWorkersToMemoryBudget, PlatformCapMatchesWasmBudgetConstants) { - // Document the wasm32 / ILP32 budget used by resolve_worker_count so a - // quiet change to THREADMEM_* cannot silently re-OOM large WASM batches. + // Document the Emscripten budget used by resolve_worker_count so a quiet + // change to THREADMEM_* cannot silently re-OOM large WASM batches. constexpr int kHeapBudgetMB = 1400; constexpr int kPerWorkerMB = THREADMEM_LARGE_DEF_MB + 24; constexpr int kExpectedCap = kHeapBudgetMB / kPerWorkerMB; @@ -72,8 +71,8 @@ TEST(ClampWorkersToMemoryBudget, PlatformCapMatchesWasmBudgetConstants) kExpectedCap); } -#if defined(__EMSCRIPTEN__) || UINTPTR_MAX == 0xffffffffu -TEST(ResolveWorkerCount, AddressSpaceCapLimitsAutoWorkers) +#if defined(__EMSCRIPTEN__) +TEST(ResolveWorkerCount, WasmMemoryCapLimitsAutoWorkers) { constexpr int kHeapBudgetMB = 1400; constexpr int kPerWorkerMB = THREADMEM_LARGE_DEF_MB + 24; @@ -81,4 +80,12 @@ TEST(ResolveWorkerCount, AddressSpaceCapLimitsAutoWorkers) EXPECT_EQ(resolve_worker_count(0, 5000), std::min(auto_workers(5000), mem_cap)); EXPECT_EQ(resolve_worker_count(64, 5000), mem_cap); } +#else +TEST(ResolveWorkerCount, NativeBuildsDoNotApplyWasmMemoryCap) +{ + // Explicit high caps must remain uncapped on native hosts; the WASM heap + // budget is Emscripten-only. + EXPECT_EQ(resolve_worker_count(64, 5000), 64); + EXPECT_EQ(resolve_worker_count(0, 5000), auto_workers(5000)); +} #endif From d56cbbe12f81a3e0e292557f54af6c416cb904d1 Mon Sep 17 00:00:00 2001 From: Adam Wildavsky Date: Mon, 27 Jul 2026 05:19:20 +0200 Subject: [PATCH 3/9] Keep native resolve_worker_count identical to develop. Avoid pulling api/dds.h into the native parallel_boards TU so host codegen of the hot path stays unchanged. Co-authored-by: Cursor --- library/src/system/parallel_boards.cpp | 23 +++++++++++++---------- 1 file changed, 13 insertions(+), 10 deletions(-) diff --git a/library/src/system/parallel_boards.cpp b/library/src/system/parallel_boards.cpp index 93a84cdb..4a7ca5c8 100644 --- a/library/src/system/parallel_boards.cpp +++ b/library/src/system/parallel_boards.cpp @@ -14,9 +14,12 @@ #include #include -#include #include +#if defined(__EMSCRIPTEN__) +#include +#endif + auto clamp_workers_to_memory_budget( const int workers, @@ -40,20 +43,20 @@ auto resolve_worker_count( const unsigned hw = std::thread::hardware_concurrency(); workers = hw > 0 ? static_cast(hw) : 1; } - workers = std::max(1, std::min(workers, count)); - +#if defined(__EMSCRIPTEN__) // Parallel workers each keep a Large TT (via SolverContext). Under // Emscripten the wasm32 heap cannot grow past ~2 GiB; uncapped auto - // (= HW concurrency) OOMs large multi-board batches. Native builds keep - // the uncapped count. Leave headroom under the WASM ceiling for the main - // module, board vectors, and growth slack. -#if defined(__EMSCRIPTEN__) + // (= HW concurrency) OOMs large multi-board batches. Leave headroom under + // that ceiling for the main module, board vectors, and growth slack. + // Native builds keep the develop resolve_worker_count body unchanged so + // host codegen of this TU is unaffected. + workers = std::max(1, std::min(workers, count)); constexpr int kHeapBudgetMB = 1400; constexpr int kPerWorkerMB = THREADMEM_LARGE_DEF_MB + 24; - workers = clamp_workers_to_memory_budget(workers, kHeapBudgetMB, kPerWorkerMB); + return clamp_workers_to_memory_budget(workers, kHeapBudgetMB, kPerWorkerMB); +#else + return std::max(1, std::min(workers, count)); #endif - - return workers; } From 47992335cb697c245fa718555fd13a22b382a42a Mon Sep 17 00:00:00 2001 From: Adam Wildavsky Date: Mon, 27 Jul 2026 05:28:29 +0200 Subject: [PATCH 4/9] Isolate the WASM worker cap in its own translation unit. Keep native parallel_boards identical to develop so the heap guard cannot change host codegen of the hot path. Co-authored-by: Cursor --- library/src/system/BUILD.bazel | 22 ++- library/src/system/parallel_boards.cpp | 29 ---- library/src/system/parallel_boards.hpp | 17 --- library/src/system/parallel_boards_wasm.cpp | 145 ++++++++++++++++++++ library/src/system/worker_memory_budget.cpp | 24 ++++ library/src/system/worker_memory_budget.hpp | 27 ++++ library/tests/system/worker_count_test.cpp | 1 + 7 files changed, 216 insertions(+), 49 deletions(-) create mode 100644 library/src/system/parallel_boards_wasm.cpp create mode 100644 library/src/system/worker_memory_budget.cpp create mode 100644 library/src/system/worker_memory_budget.hpp diff --git a/library/src/system/BUILD.bazel b/library/src/system/BUILD.bazel index a647463c..97850b41 100644 --- a/library/src/system/BUILD.bazel +++ b/library/src/system/BUILD.bazel @@ -1,11 +1,27 @@ load("//:CPPVARIABLES.bzl", "DDS_CPPOPTS", "DDS_LINKOPTS", "DDS_LOCAL_DEFINES", "DDS_SCHEDULER_DEFINE") load("@rules_cc//cc:defs.bzl", "cc_library") +# Native keeps develop's parallel_boards.cpp unchanged. Emscripten builds swap +# in parallel_boards_wasm.cpp so the wasm32 heap worker cap cannot affect host +# codegen of the hot parallel_all_boards_n translation unit. +_PARALLEL_BOARDS_SRC = select({ + "//:build_wasm": ["parallel_boards_wasm.cpp"], + "//conditions:default": ["parallel_boards.cpp"], +}) + +_SYSTEM_SRCS = glob( + ["*.cpp"], + exclude = [ + "parallel_boards.cpp", + "parallel_boards_wasm.cpp", + ], +) + _PARALLEL_BOARDS_SRC + # system, system_util_log, and system_util_stats compile the same sources with # different feature defines. Link at most one variant into a binary. cc_library( name = "system", - srcs = glob(["*.cpp"]), + srcs = _SYSTEM_SRCS, hdrs = glob(["*.hpp", "util/*.hpp"]), visibility = ["//visibility:public"], includes = ["."], @@ -27,7 +43,7 @@ cc_library( # Variant with Utilities logging enabled for tests that assert log behavior cc_library( name = "system_util_log", - srcs = glob(["*.cpp"]), + srcs = _SYSTEM_SRCS, hdrs = glob(["*.hpp", "util/*.hpp"]), visibility = ["//visibility:public"], includes = ["."], @@ -47,7 +63,7 @@ cc_library( cc_library( name = "system_util_stats", - srcs = glob(["*.cpp"]), + srcs = _SYSTEM_SRCS, hdrs = glob(["*.hpp", "util/*.hpp"]), visibility = ["//visibility:public"], includes = ["."], diff --git a/library/src/system/parallel_boards.cpp b/library/src/system/parallel_boards.cpp index 4a7ca5c8..c330a14c 100644 --- a/library/src/system/parallel_boards.cpp +++ b/library/src/system/parallel_boards.cpp @@ -16,22 +16,6 @@ #include -#if defined(__EMSCRIPTEN__) -#include -#endif - - -auto clamp_workers_to_memory_budget( - const int workers, - const int budget_mb, - const int per_worker_mb) -> int -{ - const int safe_workers = std::max(1, workers); - if (budget_mb <= 0 || per_worker_mb <= 0) - return safe_workers; - return std::max(1, std::min(safe_workers, budget_mb / per_worker_mb)); -} - auto resolve_worker_count( const int max_threads, @@ -43,20 +27,7 @@ auto resolve_worker_count( const unsigned hw = std::thread::hardware_concurrency(); workers = hw > 0 ? static_cast(hw) : 1; } -#if defined(__EMSCRIPTEN__) - // Parallel workers each keep a Large TT (via SolverContext). Under - // Emscripten the wasm32 heap cannot grow past ~2 GiB; uncapped auto - // (= HW concurrency) OOMs large multi-board batches. Leave headroom under - // that ceiling for the main module, board vectors, and growth slack. - // Native builds keep the develop resolve_worker_count body unchanged so - // host codegen of this TU is unaffected. - workers = std::max(1, std::min(workers, count)); - constexpr int kHeapBudgetMB = 1400; - constexpr int kPerWorkerMB = THREADMEM_LARGE_DEF_MB + 24; - return clamp_workers_to_memory_budget(workers, kHeapBudgetMB, kPerWorkerMB); -#else return std::max(1, std::min(workers, count)); -#endif } diff --git a/library/src/system/parallel_boards.hpp b/library/src/system/parallel_boards.hpp index e39bed89..29ae953b 100644 --- a/library/src/system/parallel_boards.hpp +++ b/library/src/system/parallel_boards.hpp @@ -22,23 +22,6 @@ */ auto resolve_worker_count(int max_threads, int count) -> int; -/** - * @brief Clamp a worker count to a memory budget (MB). - * - * Each parallel worker owns a SolverContext with a Large transposition table - * plus stack; under Emscripten the wasm32 heap tops out near 2 GiB. Tests and - * resolve_worker_count share this helper so the budget math stays in one place. - * - * @param workers Requested workers (values < 1 become 1) - * @param budget_mb Total MB available for worker TT/stack footprints - * @param per_worker_mb Assumed MB cost of one worker - * @return workers clamped to max(1, budget_mb / per_worker_mb) - */ -auto clamp_workers_to_memory_budget( - int workers, - int budget_mb, - int per_worker_mb) -> int; - /** * @brief Process boards [0, count) with work-stealing parallelism. * diff --git a/library/src/system/parallel_boards_wasm.cpp b/library/src/system/parallel_boards_wasm.cpp new file mode 100644 index 00000000..09d3be06 --- /dev/null +++ b/library/src/system/parallel_boards_wasm.cpp @@ -0,0 +1,145 @@ +/* + DDS, a bridge double dummy solver. + + Copyright (C) 2006-2014 by Bo Haglund / + 2014-2018 by Bo Haglund & Soren Hein. + + See LICENSE and README. +*/ + +#include "parallel_boards.hpp" + +#include +#include +#include +#include + +#include +#include +#include "worker_memory_budget.hpp" + + +auto resolve_worker_count( + const int max_threads, + const int count) -> int +{ + int workers = max_threads; + if (workers <= 0) + { + const unsigned hw = std::thread::hardware_concurrency(); + workers = hw > 0 ? static_cast(hw) : 1; + } + workers = std::max(1, std::min(workers, count)); + // Cap workers so Large TT footprints fit under the wasm32 ~2 GiB heap. + constexpr int kHeapBudgetMB = 1400; + constexpr int kPerWorkerMB = THREADMEM_LARGE_DEF_MB + 24; + return clamp_workers_to_memory_budget(workers, kHeapBudgetMB, kPerWorkerMB); +} + + +static auto is_permutation_of_range( + const std::vector& order, + const int count) -> bool +{ + std::vector seen(static_cast(count), 0); + for (const int v : order) + { + if (v < 0 || v >= count || seen[static_cast(v)]) + return false; + seen[static_cast(v)] = 1; + } + return true; +} + + +auto parallel_all_boards_n( + const int count, + const int worker_cap, + const std::function& process_board, + const std::vector* order) -> int +{ + if (count <= 0) + { + return RETURN_NO_FAULT; + } + + // Map a dispatch slot to the board number to process. With an order, hand out + // boards in that sequence (e.g. hardest first); otherwise in index order. The + // order is only honored when it is a valid permutation of [0, count); a + // malformed order falls back to index order to avoid invalid board indices. + const bool use_order = + (order != nullptr && + order->size() == static_cast(count) && + is_permutation_of_range(*order, count)); + auto board_of = [&](const int slot) -> int { + return use_order ? (*order)[static_cast(slot)] : slot; + }; + + const int workers = resolve_worker_count(worker_cap, count); + + if (workers == 1) + { + for (int slot = 0; slot < count; ++slot) + { + const int rc = process_board(0, board_of(slot)); + if (rc != RETURN_NO_FAULT) + { + return rc; + } + } + return RETURN_NO_FAULT; + } + + std::atomic next{0}; + std::atomic first_error{RETURN_NO_FAULT}; + + auto worker = [&](const int worker_id) { + for (;;) + { + const int slot = next.fetch_add(1, std::memory_order_relaxed); + if (slot >= count || first_error.load(std::memory_order_relaxed) != RETURN_NO_FAULT) + { + break; + } + const int bno = board_of(slot); + + const int rc = process_board(worker_id, bno); + if (rc != RETURN_NO_FAULT) + { + int expected = RETURN_NO_FAULT; + first_error.compare_exchange_strong( + expected, rc, std::memory_order_relaxed); + break; + } + } + }; + + std::vector threads; + threads.reserve(static_cast(workers)); + try + { + for (int t = 0; t < workers; ++t) + { + threads.emplace_back(worker, t); + } + } + catch (...) + { + for (auto & th : threads) + { + if (th.joinable()) + { + th.join(); + } + } + throw; + } + + for (auto & th : threads) + { + th.join(); + } + + const int err = first_error.load(std::memory_order_relaxed); + return err != RETURN_NO_FAULT ? err : RETURN_NO_FAULT; +} diff --git a/library/src/system/worker_memory_budget.cpp b/library/src/system/worker_memory_budget.cpp new file mode 100644 index 00000000..721455a1 --- /dev/null +++ b/library/src/system/worker_memory_budget.cpp @@ -0,0 +1,24 @@ +/* + DDS, a bridge double dummy solver. + + Copyright (C) 2006-2014 by Bo Haglund / + 2014-2018 by Bo Haglund & Soren Hein. + + See LICENSE and README. +*/ + +#include "worker_memory_budget.hpp" + +#include + + +auto clamp_workers_to_memory_budget( + const int workers, + const int budget_mb, + const int per_worker_mb) -> int +{ + const int safe_workers = std::max(1, workers); + if (budget_mb <= 0 || per_worker_mb <= 0) + return safe_workers; + return std::max(1, std::min(safe_workers, budget_mb / per_worker_mb)); +} diff --git a/library/src/system/worker_memory_budget.hpp b/library/src/system/worker_memory_budget.hpp new file mode 100644 index 00000000..ff4ac325 --- /dev/null +++ b/library/src/system/worker_memory_budget.hpp @@ -0,0 +1,27 @@ +/* + DDS, a bridge double dummy solver. + + Copyright (C) 2006-2014 by Bo Haglund / + 2014-2018 by Bo Haglund & Soren Hein. + + See LICENSE and README. +*/ + +#pragma once + +/** + * @brief Clamp a worker count to a memory budget (MB). + * + * Each parallel worker owns a SolverContext with a Large transposition table + * plus stack; under Emscripten the wasm32 heap tops out near 2 GiB. Kept in + * its own TU so native parallel_boards codegen stays identical to develop. + * + * @param workers Requested workers (values < 1 become 1) + * @param budget_mb Total MB available for worker TT/stack footprints + * @param per_worker_mb Assumed MB cost of one worker + * @return workers clamped to max(1, budget_mb / per_worker_mb) + */ +auto clamp_workers_to_memory_budget( + int workers, + int budget_mb, + int per_worker_mb) -> int; diff --git a/library/tests/system/worker_count_test.cpp b/library/tests/system/worker_count_test.cpp index 7929934c..295097d0 100644 --- a/library/tests/system/worker_count_test.cpp +++ b/library/tests/system/worker_count_test.cpp @@ -10,6 +10,7 @@ #include #include +#include namespace { From c496603f186a02a09a1da8a4df46d1e025d79636 Mon Sep 17 00:00:00 2001 From: Adam Wildavsky Date: Mon, 27 Jul 2026 17:09:11 +0200 Subject: [PATCH 5/9] Keep the WASM parallel_boards TU API-compatible with develop. Rebuild it from the current board-pool implementation so shutdown_parallel_boards_pool links under Emscripten after merging dtest clean exit. Co-authored-by: Cursor --- library/src/system/parallel_boards_wasm.cpp | 305 ++++++++++++++++---- 1 file changed, 249 insertions(+), 56 deletions(-) diff --git a/library/src/system/parallel_boards_wasm.cpp b/library/src/system/parallel_boards_wasm.cpp index 09d3be06..f97c455d 100644 --- a/library/src/system/parallel_boards_wasm.cpp +++ b/library/src/system/parallel_boards_wasm.cpp @@ -11,6 +11,10 @@ #include #include +#include +#include +#include +#include #include #include @@ -19,6 +23,222 @@ #include "worker_memory_budget.hpp" +namespace +{ + +std::atomic g_threads_created{0}; + + +class BoardWorkerPool +{ +public: + BoardWorkerPool() = default; + + ~BoardWorkerPool() + { + { + std::lock_guard lock(mu_); + stopping_ = true; + ++generation_; + job_ = nullptr; + workers_for_job_ = 0; + } + cv_start_.notify_all(); + for (auto& th : threads_) + { + if (th.joinable()) + th.join(); + } + } + + BoardWorkerPool(const BoardWorkerPool&) = delete; + auto operator=(const BoardWorkerPool&) -> BoardWorkerPool& = delete; + + auto run( + const int workers, + const int count, + const std::function& process_board, + const bool use_order, + const std::vector* order) -> int + { + // The pool tracks exactly one outstanding job (job_, workers_for_job_, + // generation_). Serialize whole runs so a second concurrent caller queues + // up instead of overwriting the first caller's job, which would leave the + // first run waiting forever for workers that never saw its job. + std::lock_guard run_lock(run_mu_); + + ensure_workers(workers); + + std::atomic next{0}; + std::atomic first_error{RETURN_NO_FAULT}; + // finished is protected by mu_: every worker increments it under that lock + // so the unlock→lock handoff through cv_done_ happens-before the caller's + // continued use of process_board side effects (e.g. result buffers). + int finished{0}; + + Job job{ + &process_board, + order, + &next, + &first_error, + &finished, + count, + workers, + use_order}; + + { + std::lock_guard lock(mu_); + job_ = &job; + workers_for_job_ = workers; + ++generation_; + } + cv_start_.notify_all(); + + { + std::unique_lock lock(mu_); + cv_done_.wait(lock, [&] { return finished >= workers; }); + job_ = nullptr; + } + + const int err = first_error.load(std::memory_order_relaxed); + return err != RETURN_NO_FAULT ? err : RETURN_NO_FAULT; + } + +private: + struct Job + { + const std::function* process_board = nullptr; + const std::vector* order = nullptr; + std::atomic* next = nullptr; + std::atomic* first_error = nullptr; + int* finished = nullptr; + int count = 0; + int workers = 0; + bool use_order = false; + }; + + void ensure_workers(const int workers) + { + std::lock_guard lock(mu_); + while (static_cast(threads_.size()) < workers) + { + const int worker_id = static_cast(threads_.size()); + threads_.emplace_back([this, worker_id] { worker_main(worker_id); }); + g_threads_created.fetch_add(1, std::memory_order_relaxed); + } + } + + void worker_main(const int worker_id) + { + std::uint64_t seen_generation = 0; + for (;;) + { + Job local{}; + { + std::unique_lock lock(mu_); + cv_start_.wait(lock, [&] { + return stopping_ || + (job_ != nullptr && + generation_ != seen_generation && + worker_id < workers_for_job_); + }); + if (stopping_) + return; + seen_generation = generation_; + local = *job_; + } + + // Workers with id >= local.workers for this job still wake on notify_all; + // only the selected prefix participates and signals completion. + if (worker_id < local.workers) + { + try + { + for (;;) + { + const int slot = local.next->fetch_add(1, std::memory_order_relaxed); + if (slot >= local.count || + local.first_error->load(std::memory_order_relaxed) != RETURN_NO_FAULT) + { + break; + } + const int bno = + local.use_order + ? (*local.order)[static_cast(slot)] + : slot; + const int rc = (*local.process_board)(worker_id, bno); + if (rc != RETURN_NO_FAULT) + { + int expected = RETURN_NO_FAULT; + local.first_error->compare_exchange_strong( + expected, rc, std::memory_order_relaxed); + break; + } + } + } + catch (...) + { + // process_board must not leave the pool hanging or abort the process + // via std::terminate. Any throw maps the run to RETURN_UNKNOWN_FAULT + // (even if another worker already recorded a non-success return code), + // then fall through to finished accounting so cv_done_ is signaled. + local.first_error->store( + RETURN_UNKNOWN_FAULT, std::memory_order_relaxed); + } + // Account completion under mu_ so (1) the predicate change cannot race + // with cv_done_.wait's check-and-sleep (lost wakeup) and (2) unlocking + // mu_ after process_board writes publishes those writes to the caller + // when wait reacquires the mutex. + bool notify = false; + { + std::lock_guard lock(mu_); + notify = ++(*local.finished) >= local.workers; + } + if (notify) + cv_done_.notify_one(); + } + } + } + + // Held for the duration of run(); see the comment there. + std::mutex run_mu_; + std::mutex mu_; + std::condition_variable cv_start_; + std::condition_variable cv_done_; + std::vector threads_; + bool stopping_ = false; + std::uint64_t generation_ = 0; + Job* job_ = nullptr; + int workers_for_job_ = 0; +}; + + +struct PoolHolder +{ + std::mutex mu; + std::shared_ptr pool; +}; + + +auto pool_holder() -> PoolHolder& +{ + static PoolHolder holder; + return holder; +} + + +auto default_pool() -> std::shared_ptr +{ + auto& h = pool_holder(); + std::lock_guard lock(h.mu); + if (!h.pool) + h.pool = std::make_shared(); + return h.pool; +} + +} // namespace + + auto resolve_worker_count( const int max_threads, const int count) -> int @@ -52,6 +272,31 @@ static auto is_permutation_of_range( } +namespace dds::internal +{ + +auto parallel_boards_worker_threads_created() -> std::uint64_t +{ + return g_threads_created.load(std::memory_order_relaxed); +} + + +void shutdown_parallel_boards_pool() +{ + std::shared_ptr dying; + { + auto& h = pool_holder(); + std::lock_guard lock(h.mu); + dying = std::move(h.pool); + } + // Drop the last shared_ptr outside the holder lock so joins cannot deadlock + // against a concurrent parallel_all_boards_n that needs the mutex to recreate + // the pool. In-flight runs keep their own shared_ptr alive until run returns. +} + +} // namespace dds::internal + + auto parallel_all_boards_n( const int count, const int worker_cap, @@ -71,9 +316,6 @@ auto parallel_all_boards_n( (order != nullptr && order->size() == static_cast(count) && is_permutation_of_range(*order, count)); - auto board_of = [&](const int slot) -> int { - return use_order ? (*order)[static_cast(slot)] : slot; - }; const int workers = resolve_worker_count(worker_cap, count); @@ -81,7 +323,9 @@ auto parallel_all_boards_n( { for (int slot = 0; slot < count; ++slot) { - const int rc = process_board(0, board_of(slot)); + const int bno = + use_order ? (*order)[static_cast(slot)] : slot; + const int rc = process_board(0, bno); if (rc != RETURN_NO_FAULT) { return rc; @@ -90,56 +334,5 @@ auto parallel_all_boards_n( return RETURN_NO_FAULT; } - std::atomic next{0}; - std::atomic first_error{RETURN_NO_FAULT}; - - auto worker = [&](const int worker_id) { - for (;;) - { - const int slot = next.fetch_add(1, std::memory_order_relaxed); - if (slot >= count || first_error.load(std::memory_order_relaxed) != RETURN_NO_FAULT) - { - break; - } - const int bno = board_of(slot); - - const int rc = process_board(worker_id, bno); - if (rc != RETURN_NO_FAULT) - { - int expected = RETURN_NO_FAULT; - first_error.compare_exchange_strong( - expected, rc, std::memory_order_relaxed); - break; - } - } - }; - - std::vector threads; - threads.reserve(static_cast(workers)); - try - { - for (int t = 0; t < workers; ++t) - { - threads.emplace_back(worker, t); - } - } - catch (...) - { - for (auto & th : threads) - { - if (th.joinable()) - { - th.join(); - } - } - throw; - } - - for (auto & th : threads) - { - th.join(); - } - - const int err = first_error.load(std::memory_order_relaxed); - return err != RETURN_NO_FAULT ? err : RETURN_NO_FAULT; + return default_pool()->run(workers, count, process_board, use_order, order); } From 281a4966c9c177f1085b465b65c713bffbb6dd14 Mon Sep 17 00:00:00 2001 From: Adam Wildavsky Date: Mon, 27 Jul 2026 17:30:18 +0200 Subject: [PATCH 6/9] Clarify that the parallel_boards select is native vs WASM. Avoid referring to the develop branch in the BUILD comment so the split reads as a host/Emscripten source choice. Co-authored-by: Cursor --- library/src/system/BUILD.bazel | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/library/src/system/BUILD.bazel b/library/src/system/BUILD.bazel index 97850b41..6abb1666 100644 --- a/library/src/system/BUILD.bazel +++ b/library/src/system/BUILD.bazel @@ -1,8 +1,8 @@ load("//:CPPVARIABLES.bzl", "DDS_CPPOPTS", "DDS_LINKOPTS", "DDS_LOCAL_DEFINES", "DDS_SCHEDULER_DEFINE") load("@rules_cc//cc:defs.bzl", "cc_library") -# Native keeps develop's parallel_boards.cpp unchanged. Emscripten builds swap -# in parallel_boards_wasm.cpp so the wasm32 heap worker cap cannot affect host +# Native builds keep parallel_boards.cpp unchanged. Emscripten builds swap in +# parallel_boards_wasm.cpp so the wasm32 heap worker cap cannot affect host # codegen of the hot parallel_all_boards_n translation unit. _PARALLEL_BOARDS_SRC = select({ "//:build_wasm": ["parallel_boards_wasm.cpp"], From 11dd0150e1f695c44af7fa8effda7fac1f3d09ae Mon Sep 17 00:00:00 2001 From: Adam Wildavsky Date: Mon, 27 Jul 2026 17:30:54 +0200 Subject: [PATCH 7/9] Derive worker budget test expectations from THREADMEM_LARGE_DEF_MB. Avoid hard-coding the Large TT default so the clamp tests stay aligned if that constant changes. Co-authored-by: Cursor --- library/tests/system/worker_count_test.cpp | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/library/tests/system/worker_count_test.cpp b/library/tests/system/worker_count_test.cpp index 295097d0..0c267167 100644 --- a/library/tests/system/worker_count_test.cpp +++ b/library/tests/system/worker_count_test.cpp @@ -52,9 +52,10 @@ TEST(ResolveWorkerCount, SingleItemAlwaysOneWorker) TEST(ClampWorkersToMemoryBudget, CapsByBudgetPerWorker) { - EXPECT_EQ(clamp_workers_to_memory_budget(18, 1400, 95 + 24), 11); - EXPECT_EQ(clamp_workers_to_memory_budget(4, 1400, 95 + 24), 4); - EXPECT_EQ(clamp_workers_to_memory_budget(0, 1400, 119), 1); + constexpr int kPerWorkerMB = THREADMEM_LARGE_DEF_MB + 24; + EXPECT_EQ(clamp_workers_to_memory_budget(18, 1400, kPerWorkerMB), 11); + EXPECT_EQ(clamp_workers_to_memory_budget(4, 1400, kPerWorkerMB), 4); + EXPECT_EQ(clamp_workers_to_memory_budget(0, 1400, kPerWorkerMB), 1); EXPECT_EQ(clamp_workers_to_memory_budget(8, 100, 200), 1); } From a060648116f3fdb76b0b60ae0017b4ee1b476c56 Mon Sep 17 00:00:00 2001 From: Adam Wildavsky Date: Mon, 27 Jul 2026 17:31:45 +0200 Subject: [PATCH 8/9] Clarify worker_memory_budget docs without branch-specific wording. Note that the clamp lives in its own .cpp rather than referring to develop. Co-authored-by: Cursor --- library/src/system/worker_memory_budget.hpp | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/library/src/system/worker_memory_budget.hpp b/library/src/system/worker_memory_budget.hpp index ff4ac325..7d752f07 100644 --- a/library/src/system/worker_memory_budget.hpp +++ b/library/src/system/worker_memory_budget.hpp @@ -13,8 +13,9 @@ * @brief Clamp a worker count to a memory budget (MB). * * Each parallel worker owns a SolverContext with a Large transposition table - * plus stack; under Emscripten the wasm32 heap tops out near 2 GiB. Kept in - * its own TU so native parallel_boards codegen stays identical to develop. + * plus stack; under Emscripten the wasm32 heap tops out near 2 GiB. The + * implementation is in worker_memory_budget.cpp so this helper stays out of the + * native parallel_boards translation unit. * * @param workers Requested workers (values < 1 become 1) * @param budget_mb Total MB available for worker TT/stack footprints From 4deba90cb3040423b62e381b7593dac9c80acfa0 Mon Sep 17 00:00:00 2001 From: Adam Wildavsky Date: Mon, 27 Jul 2026 18:17:44 +0200 Subject: [PATCH 9/9] Document that parallel_boards native and WASM copies must stay in sync. Call out that only resolve_worker_count should differ, reducing drift risk between the duplicated worker-pool implementations. Co-authored-by: Cursor --- library/src/system/parallel_boards.cpp | 5 +++++ library/src/system/parallel_boards_wasm.cpp | 5 +++++ 2 files changed, 10 insertions(+) diff --git a/library/src/system/parallel_boards.cpp b/library/src/system/parallel_boards.cpp index ce69cb36..a69b191b 100644 --- a/library/src/system/parallel_boards.cpp +++ b/library/src/system/parallel_boards.cpp @@ -9,6 +9,11 @@ #include "parallel_boards.hpp" +// NOTE: parallel_boards_wasm.cpp is a wasm-only copy of this file so the +// Emscripten worker-count memory cap does not affect native codegen. Keep the +// worker pool and parallel_all_boards_n logic in sync with that file; only +// resolve_worker_count should differ (WASM heap cap). + #include #include #include diff --git a/library/src/system/parallel_boards_wasm.cpp b/library/src/system/parallel_boards_wasm.cpp index f97c455d..a10695e7 100644 --- a/library/src/system/parallel_boards_wasm.cpp +++ b/library/src/system/parallel_boards_wasm.cpp @@ -9,6 +9,11 @@ #include "parallel_boards.hpp" +// NOTE: This is a wasm-only copy of parallel_boards.cpp so the Emscripten +// worker-count memory cap does not affect native codegen. Keep the worker pool +// and parallel_all_boards_n logic in sync with parallel_boards.cpp; only +// resolve_worker_count should differ (WASM heap cap via worker_memory_budget). + #include #include #include