From cf0b8b10c5163966f08a4c267b1129a7587ba22a Mon Sep 17 00:00:00 2001 From: Fabian Peddinghaus Date: Mon, 3 Aug 2026 16:01:25 +0200 Subject: [PATCH] Keep every placement a broken pool would have lost, and drop the pin cliff A worker dying abruptly breaks the whole executor, so every future still in flight fails with it: `allocate_parallel` was catching that per variant and finding nothing left, turning one OOM kill into "Every allocator variant failed". The variants a broken pool stranded never ran at all, so they are retried one pool each -- never in the calling process, where whatever killed the worker would kill the caller, and never sharing a pool, where the same variant would strand them again. The thread ceiling lives in native process-global state, which a forked worker inherits and a spawned one does not, so the cap was silently absent inside every pool worker off Linux. Pools hand it down explicitly now, split by the worker count: the ceiling covers the product, not each level of it. `stack_around_pins` fell off its linear path the moment a single pin existed, rescanning every occupied range per item and inserting into it: 20k allocations went from 9ms to 14.9s, and the naive allocator is the linear baseline. Claims only ever shrink a free range from below and never split one, so the range count is fixed at what the pins leave and a max tree over the capacities answers each first fit in O(log n). The adjacency ceiling counts 4 bytes per directed edge, but the placers took their rows through `ConflictIndices`, widening every one to size_t behind a per-row vector. They walk the CSR rows in place now, so the accounting is the whole cost rather than a stage of it, and the type is gone. TwoPlusTwoSource truncated its last group below four allocations, handing back a linearizable instance from a source that promises the opposite; it refuses the count instead. Validating a Memory directly no longer names it twice, and `available_cores` reports the benchmark environment's core count rather than being unused beside `os.cpu_count`. --- src/cpp/allocators/best_fit.cpp | 2 +- src/cpp/allocators/first_fit.cpp | 22 ++--- src/cpp/allocators/first_fit.hpp | 15 ++- src/cpp/allocators/local_search.cpp | 11 ++- src/cpp/allocators/local_search.hpp | 4 +- src/cpp/allocators/simulated_annealing.cpp | 2 +- src/cpp/allocators/tabu_search.cpp | 2 +- src/cpp/allocators/telamalloc.cpp | 24 +++-- src/cpp/analysis/clock.hpp | 17 +++- src/cpp/analysis/conflicts.cpp | 16 --- src/cpp/analysis/conflicts.hpp | 7 -- .../omnimalloc/allocators/greedy_base.py | 97 +++++++++++++++---- .../omnimalloc/benchmark/results/utils.py | 4 +- .../benchmark/sources/adversarial.py | 11 ++- src/python/omnimalloc/common/intervals.py | 83 +++++++++++----- src/python/omnimalloc/common/parallel.py | 16 ++- src/python/omnimalloc/validate.py | 14 ++- tests/integration/test_torture.py | 51 ++++++++++ .../benchmark/sources/test_adversarial.py | 12 +++ tests/unit/common/test_intervals.py | 77 ++++++++++++--- tests/unit/common/test_parallel.py | 27 ++++++ 21 files changed, 382 insertions(+), 132 deletions(-) diff --git a/src/cpp/allocators/best_fit.cpp b/src/cpp/allocators/best_fit.cpp index 4472c99..cc65459 100644 --- a/src/cpp/allocators/best_fit.cpp +++ b/src/cpp/allocators/best_fit.cpp @@ -37,7 +37,7 @@ std::vector best_fit_place( const std::vector& allocations) { // Lambda rather than the function pointer so the placement loop inlines // the offset scan instead of an indirect call per allocation - return place_indexed(allocations, compute_conflict_indices(allocations), + return place_indexed(allocations, build_conflict_adjacency(allocations), [](int64_t size, const auto& spans) { return find_best_fit_offset(size, spans); }); diff --git a/src/cpp/allocators/first_fit.cpp b/src/cpp/allocators/first_fit.cpp index baeaade..00edcf4 100644 --- a/src/cpp/allocators/first_fit.cpp +++ b/src/cpp/allocators/first_fit.cpp @@ -10,6 +10,7 @@ #include #include #include +#include #include #include #include @@ -302,12 +303,13 @@ PortfolioPlacement place_portfolio(const std::vector& allocations, return best; } -void gather_spans(const std::vector& neighbors, +void gather_spans(std::span neighbors, const std::vector>& offsets, const std::vector& allocations, std::vector>& spans) { spans.clear(); - for (size_t j : neighbors) { + for (const int32_t neighbor : neighbors) { + const auto j = static_cast(neighbor); if (offsets[j].has_value()) { spans.emplace_back(*offsets[j], *offsets[j] + allocations[j].size()); } @@ -328,25 +330,23 @@ int64_t first_fit_offset( } std::vector first_fit_place_indexed( - const std::vector& allocations, - const ConflictIndices& indices) { + const std::vector& allocations, const CsrAdjacency& adj) { // Lambda rather than the function pointer so the placement loop inlines // the offset scan instead of an indirect call per allocation - return place_indexed(allocations, indices, - [](int64_t size, const auto& spans) { - return first_fit_offset(size, spans); - }); + return place_indexed(allocations, adj, [](int64_t size, const auto& spans) { + return first_fit_offset(size, spans); + }); } std::vector first_fit_place( const std::vector& allocations) { return first_fit_place_indexed(allocations, - compute_conflict_indices(allocations)); + build_conflict_adjacency(allocations)); } FirstFitPlacer::FirstFitPlacer(std::vector allocations) : allocations_(std::move(allocations)), - indices_(compute_conflict_indices(allocations_)) { + adj_(build_conflict_adjacency(allocations_)) { check_total_size(allocations_); } @@ -379,7 +379,7 @@ std::vector> FirstFitPlacer::place_offsets( if (alloc.offset().has_value()) { continue; } - gather_spans(indices_[idx], offsets, allocations_, spans); + gather_spans(adj_.row(idx), offsets, allocations_, spans); offsets[idx] = first_fit_offset(alloc.size(), spans); } return offsets; diff --git a/src/cpp/allocators/first_fit.hpp b/src/cpp/allocators/first_fit.hpp index e861375..fee1918 100644 --- a/src/cpp/allocators/first_fit.hpp +++ b/src/cpp/allocators/first_fit.hpp @@ -6,6 +6,7 @@ #include #include +#include #include #include @@ -16,7 +17,7 @@ namespace omnimalloc { // Occupied (offset, end) spans of the already-placed neighbors of one // allocation, sorted by offset so the gap scans can go left-to-right -void gather_spans(const std::vector& neighbors, +void gather_spans(std::span neighbors, const std::vector>& offsets, const std::vector& allocations, std::vector>& spans); @@ -47,14 +48,14 @@ struct PortfolioPlacement { // Greedily place allocations in order using first-fit over an index-based // adjacency (the fast path: each step only visits the allocation's neighbors) [[nodiscard]] std::vector first_fit_place_indexed( - const std::vector& allocations, const ConflictIndices& indices); + const std::vector& allocations, const CsrAdjacency& adj); // Shared placement skeleton of the first-fit and best-fit placers: place in // index order, choosing each offset with `choose_offset` over the sorted spans // of already-placed neighbors. Seeding pins makes them obstacles throughout. template [[nodiscard]] std::vector place_indexed( - const std::vector& allocations, const ConflictIndices& indices, + const std::vector& allocations, const CsrAdjacency& adj, OffsetFn choose_offset) { check_total_size(allocations); std::vector> offsets(allocations.size()); @@ -66,7 +67,7 @@ template placed.reserve(allocations.size()); for (size_t i = 0; i < allocations.size(); ++i) { if (!allocations[i].offset().has_value()) { - gather_spans(indices[i], offsets, allocations, spans); + gather_spans(adj.row(i), offsets, allocations, spans); offsets[i] = choose_offset(allocations[i].size(), spans); } placed.push_back(allocations[i].with_offset(*offsets[i])); @@ -91,9 +92,7 @@ class FirstFitPlacer { const std::vector& order) const; // The resident index adjacency, for the local searches' inner loops. - [[nodiscard]] const ConflictIndices& indices() const noexcept { - return indices_; - } + [[nodiscard]] const CsrAdjacency& adjacency() const noexcept { return adj_; } private: // Throw std::invalid_argument unless every index in `order` is in range @@ -106,7 +105,7 @@ class FirstFitPlacer { const std::vector& order) const; std::vector allocations_; - ConflictIndices indices_; + CsrAdjacency adj_; }; } // namespace omnimalloc diff --git a/src/cpp/allocators/local_search.cpp b/src/cpp/allocators/local_search.cpp index e28cd87..844f033 100644 --- a/src/cpp/allocators/local_search.cpp +++ b/src/cpp/allocators/local_search.cpp @@ -6,6 +6,7 @@ #include #include +#include #include namespace omnimalloc { @@ -43,13 +44,13 @@ std::vector initial_order(const std::vector& allocations) { std::vector earlier_neighbors(const std::vector& order, size_t target_pos, - const ConflictIndices& indices) { + const CsrAdjacency& adj) { // Mark the target's conflicts, then keep the earlier positions holding // one; the marks cost a pass over the order, the alternative a hash // lookup per earlier position std::vector conflicting(order.size(), 0); - for (size_t other : indices[order[target_pos]]) { - conflicting[other] = 1; + for (const int32_t other : adj.row(order[target_pos])) { + conflicting[static_cast(other)] = 1; } std::vector neighbors; for (size_t pos = 0; pos < target_pos; ++pos) { @@ -66,12 +67,12 @@ std::vector earlier_neighbors(const std::vector& order, std::optional> propose_peak_swap( const std::vector& peaks, const std::vector& order, - const ConflictIndices& indices, std::mt19937_64& rng) { + const CsrAdjacency& adj, std::mt19937_64& rng) { assert(!peaks.empty()); // full placements always attain their peak std::uniform_int_distribution pick_peak(0, peaks.size() - 1); const size_t target_pos = peaks[pick_peak(rng)]; const std::vector neighbors = - earlier_neighbors(order, target_pos, indices); + earlier_neighbors(order, target_pos, adj); if (neighbors.empty()) { return std::nullopt; } diff --git a/src/cpp/allocators/local_search.hpp b/src/cpp/allocators/local_search.hpp index bfc0b1c..c0ba25c 100644 --- a/src/cpp/allocators/local_search.hpp +++ b/src/cpp/allocators/local_search.hpp @@ -33,13 +33,13 @@ namespace omnimalloc { // adjacency: this runs per sampled move, where hashing ids would dominate. [[nodiscard]] std::vector earlier_neighbors( const std::vector& order, size_t target_pos, - const ConflictIndices& indices); + const CsrAdjacency& adj); // One random peak-lowering move for the local searches: a random position // among `peaks` paired with a random earlier temporal neighbor, or nullopt // when the chosen target has no earlier position to swap with. [[nodiscard]] std::optional> propose_peak_swap( const std::vector& peaks, const std::vector& order, - const ConflictIndices& indices, std::mt19937_64& rng); + const CsrAdjacency& adj, std::mt19937_64& rng); } // namespace omnimalloc diff --git a/src/cpp/allocators/simulated_annealing.cpp b/src/cpp/allocators/simulated_annealing.cpp index 9a8384f..e652eec 100644 --- a/src/cpp/allocators/simulated_annealing.cpp +++ b/src/cpp/allocators/simulated_annealing.cpp @@ -43,7 +43,7 @@ std::vector simulated_annealing_place( peak_positions(current_placed, current_peak); const auto proposal = - propose_peak_swap(peaks, order, placer.indices(), rng); + propose_peak_swap(peaks, order, placer.adjacency(), rng); if (!proposal) { temperature *= config.cooling_rate; continue; diff --git a/src/cpp/allocators/tabu_search.cpp b/src/cpp/allocators/tabu_search.cpp index 2211fea..86a84ca 100644 --- a/src/cpp/allocators/tabu_search.cpp +++ b/src/cpp/allocators/tabu_search.cpp @@ -71,7 +71,7 @@ std::vector tabu_search_place( break; } const auto proposal = - propose_peak_swap(peaks, order, placer.indices(), rng); + propose_peak_swap(peaks, order, placer.adjacency(), rng); if (!proposal) { continue; } diff --git a/src/cpp/allocators/telamalloc.cpp b/src/cpp/allocators/telamalloc.cpp index 246e5d8..3b32c87 100644 --- a/src/cpp/allocators/telamalloc.cpp +++ b/src/cpp/allocators/telamalloc.cpp @@ -30,7 +30,7 @@ constexpr int64_t kUnbounded = std::numeric_limits::max() / 4; // Connected components of the overlap graph: the paper's "phases". Buffers // in different components never interact, so each packs independently. -std::vector> build_phases(const ConflictIndices& neighbors) { +std::vector> build_phases(const CsrAdjacency& neighbors) { const int n = static_cast(neighbors.size()); std::vector> phases; std::vector visited(n, 0); @@ -45,7 +45,7 @@ std::vector> build_phases(const ConflictIndices& neighbors) { int idx = stack.back(); stack.pop_back(); phase.push_back(idx); - for (size_t other : neighbors[idx]) { + for (const int32_t other : neighbors.row(static_cast(idx))) { if (!visited[other]) { visited[other] = 1; stack.push_back(static_cast(other)); @@ -88,10 +88,9 @@ QueueKey queue_key(const Allocation& alloc, int idx, int evictions, // fitting gap, evicting the cheapest blocking set on conflict and requeueing it // at raised priority; a spent eviction budget wipes and re-packs the phase. std::optional> pack_phase( - const std::vector& allocations, - const ConflictIndices& neighbors, const std::vector& phase, - int64_t capacity, int max_backtracks, const Deadline& deadline, - bool size_major, uint64_t seed) { + const std::vector& allocations, const CsrAdjacency& neighbors, + const std::vector& phase, int64_t capacity, int max_backtracks, + const Deadline& deadline, bool size_major, uint64_t seed) { std::vector offsets(allocations.size(), -1); std::vector evictions(allocations.size(), 0); std::mt19937_64 rng(seed); @@ -130,9 +129,9 @@ std::optional> pack_phase( // spans inherit occupied's offset order (ties differ only in end // order, which cannot change a gap scan's result), so one sort does. occupied.clear(); - for (size_t other : neighbors[idx]) { + for (const int32_t other : neighbors.row(static_cast(idx))) { if (offsets[other] >= 0) { - occupied.emplace_back(offsets[other], static_cast(other)); + occupied.emplace_back(offsets[other], other); } } std::sort(occupied.begin(), occupied.end()); @@ -232,10 +231,9 @@ int64_t phase_peak(const std::vector& allocations, // search on capacity down toward the load lower bound. Budget-exhausted // attempts count as infeasible, keeping the search anytime rather than exact. void solve_phase(const std::vector& allocations, - const ConflictIndices& neighbors, - const std::vector& phase, int64_t lower_bound, - const TelamallocConfig& config, const Deadline& deadline, - std::vector& result) { + const CsrAdjacency& neighbors, const std::vector& phase, + int64_t lower_bound, const TelamallocConfig& config, + const Deadline& deadline, std::vector& result) { // Unbounded capacity never conflicts, so these incumbents are plain // first-fit in each tiered order and cannot fail. The winner's order also // steers the capacity search below. @@ -284,7 +282,7 @@ std::vector telamalloc_place( // Bound by kUnbounded so the unbounded-capacity pack_phase incumbents in // solve_phase can never fail and the cursor arithmetic cannot overflow. check_total_size(allocations, kUnbounded); - const ConflictIndices neighbors = compute_conflict_indices(allocations); + const CsrAdjacency neighbors = build_conflict_adjacency(allocations); if (allocations.size() < 2) { return first_fit_place_indexed(allocations, neighbors); } diff --git a/src/cpp/analysis/clock.hpp b/src/cpp/analysis/clock.hpp index f4e558b..5ec9e27 100644 --- a/src/cpp/analysis/clock.hpp +++ b/src/cpp/analysis/clock.hpp @@ -339,11 +339,22 @@ inline std::vector interval_peaks( struct CsrAdjacency { std::vector offsets; std::vector neighbors; + + [[nodiscard]] size_t size() const noexcept { return offsets.size() - 1; } + + // One row as a view into the shared storage, so walking the relation costs + // no copy and the 4-bytes-per-edge accounting holds all the way through. + [[nodiscard]] std::span row(size_t index) const noexcept { + return {neighbors.data() + offsets[index], + static_cast(offsets[index + 1] - offsets[index])}; + } }; -// Ceiling on the neighbor entries one adjacency may materialize, 4 bytes each. -// No work budget bounds the CSR: budgets count the sweep, and the sweep is what -// fills the rows. A guard against taking the host down, not a tuning knob. +// Ceiling on the neighbor entries one adjacency may materialize, 4 bytes each +// and 4 all the way through: every consumer walks the CSR rows in place, so the +// accounting is the whole cost, not a stage of it. No work budget bounds the +// CSR: budgets count the sweep, and the sweep is what fills the rows. A guard +// against taking the host down, not a tuning knob. inline constexpr uint64_t kMaxAdjacencyEntries = uint64_t{1} << 31; // Pairwise happens-before conflict sweep, O(n^2 * T) worst case; vector clocks diff --git a/src/cpp/analysis/conflicts.cpp b/src/cpp/analysis/conflicts.cpp index fcac1ea..3975333 100644 --- a/src/cpp/analysis/conflicts.cpp +++ b/src/cpp/analysis/conflicts.cpp @@ -37,16 +37,6 @@ void check_sweep_budget(const ConflictSweep& sweep, } } -// Pairwise happens-before adjacency unpacked from CSR form -ConflictIndices indices_from_adjacency(size_t n, const CsrAdjacency& adj) { - ConflictIndices indices(n); - for (size_t i = 0; i < n; ++i) { - indices[i].assign(adj.neighbors.begin() + adj.offsets[i], - adj.neighbors.begin() + adj.offsets[i + 1]); - } - return indices; -} - // Total id-keyed conflict map straight from CSR: the index adjacency in // between would cost a widening copy of every directed edge. ConflictMap conflict_map_from_adjacency( @@ -74,12 +64,6 @@ CsrAdjacency build_conflict_adjacency( return sweep.adjacency(parallel_threads(sweep.count())); } -ConflictIndices compute_conflict_indices( - const std::vector& allocations) { - return indices_from_adjacency(allocations.size(), - build_conflict_adjacency(allocations)); -} - ConflictMap conflicts(const std::vector& allocations, std::optional work_budget) { // The sweep work equals the conflict-pair count on scalar input, so one diff --git a/src/cpp/analysis/conflicts.hpp b/src/cpp/analysis/conflicts.hpp index fca4d51..15d6eca 100644 --- a/src/cpp/analysis/conflicts.hpp +++ b/src/cpp/analysis/conflicts.hpp @@ -22,19 +22,12 @@ using ConflictMap = std::unordered_map, IdTypeHash>; -// Index-based conflict adjacency: position i -> positions conflicting with i -using ConflictIndices = std::vector>; - // Map each allocation id to the ids of conflicting allocations (the // happens-before conflict relation every placement packs against). A set // `work_budget` bounds the quadratic sweep, throwing rather than stalling. [[nodiscard]] ConflictMap conflicts(const std::vector& allocations, std::optional work_budget); -// Map each allocation index to the indices of conflicting allocations -[[nodiscard]] ConflictIndices compute_conflict_indices( - const std::vector& allocations); - // Per-allocation count of conflicting allocations, aligned with `allocations` // and counted with multiplicity. Scalar timelines count in O(N log N) without // enumerating pairs; on vector clocks `work_budget` bounds the sweep. diff --git a/src/python/omnimalloc/allocators/greedy_base.py b/src/python/omnimalloc/allocators/greedy_base.py index 99a95f3..4d49c25 100644 --- a/src/python/omnimalloc/allocators/greedy_base.py +++ b/src/python/omnimalloc/allocators/greedy_base.py @@ -3,12 +3,18 @@ # import logging +from collections.abc import Sequence from concurrent.futures import ProcessPoolExecutor +from concurrent.futures.process import BrokenProcessPool from omnimalloc.analysis import conflict_degrees, placement_pressure from omnimalloc.analysis.clock import uniform_dim from omnimalloc.common.constants import DEFAULT_WORK_BUDGET -from omnimalloc.common.parallel import resolve_num_threads +from omnimalloc.common.parallel import ( + adopt_max_threads, + max_threads, + resolve_num_threads, +) from omnimalloc.primitives import Allocation from .base import BaseAllocator @@ -77,6 +83,73 @@ def _allocate( return allocator.allocate(allocations) +def _run_here( + variant: BaseAllocator, allocations: tuple[Allocation, ...] +) -> tuple[Allocation, ...] | None: + """One variant's placement in this process, or None once it has failed.""" + try: + return variant.allocate(allocations) + except Exception: # noqa: BLE001 + logger.warning("Variant %s failed; skipping it", variant, exc_info=True) + return None + + +def _worker_ceiling(workers: int) -> int: + """Thread ceiling one pool worker gets, so the pool respects the whole one. + + Workers run the native kernels too, and the ceiling is native process-global + state a spawned worker does not inherit, so a pool hands down its share. + """ + return max(1, max_threads() // workers) + + +def _run_in_pool( + allocations: tuple[Allocation, ...], + variants: Sequence[BaseAllocator], + workers: int, +) -> tuple[list[tuple[Allocation, ...]], list[BaseAllocator]]: + """Placements from the pool, plus the variants a broken pool took down. + + A worker dying abruptly (OOM kill, segfault) breaks the executor, so every + future still in flight fails alongside it. Those variants never produced an + answer at all, and come back for a retry rather than counting as failures. + """ + results = [] + stranded = [] + with ProcessPoolExecutor( + max_workers=workers, + initializer=adopt_max_threads, + initargs=(_worker_ceiling(workers),), + ) as pool: + futures = [pool.submit(_allocate, v, allocations) for v in variants] + for variant, future in zip(variants, futures, strict=True): + try: + results.append(future.result()) + except BrokenProcessPool: + stranded.append(variant) + except Exception: # noqa: BLE001 + logger.warning("Variant %s failed; skipping it", variant, exc_info=True) + return results, stranded + + +def _rerun_stranded( + allocations: tuple[Allocation, ...], stranded: list[BaseAllocator] +) -> list[tuple[Allocation, ...]]: + """Retry the variants a broken pool took down, one pool each. + + One at a time, and never in this process: whatever killed the worker would + kill the caller too, and a shared retry pool would just strand them again. + """ + logger.warning("Worker pool broke; retrying %d variant(s)", len(stranded)) + results = [] + for variant in stranded: + placed, failed = _run_in_pool(allocations, [variant], workers=1) + results += placed + if failed: + logger.warning("Variant %s took its worker down; skipping it", variant) + return results + + def allocate_parallel( allocations: tuple[Allocation, ...], variants: tuple[BaseAllocator, ...], @@ -95,25 +168,13 @@ def allocate_parallel( if num_threads is None: workers = min(workers, len(variants)) - # One variant dying (raised, or its worker OOM-killed) must not discard - # the placements the others already produced, on either path - results = [] if workers <= 1: - for variant in variants: - try: - results.append(variant.allocate(allocations)) - except Exception: # noqa: BLE001 - logger.warning("Variant %s failed; skipping it", variant, exc_info=True) + placements = (_run_here(variant, allocations) for variant in variants) + results = [placed for placed in placements if placed is not None] else: - with ProcessPoolExecutor(max_workers=workers) as pool: - futures = [pool.submit(_allocate, v, allocations) for v in variants] - for variant, future in zip(variants, futures, strict=True): - try: - results.append(future.result()) - except Exception: # noqa: BLE001 - logger.warning( - "Variant %s failed; skipping it", variant, exc_info=True - ) + results, stranded = _run_in_pool(allocations, variants, workers) + if stranded: + results += _rerun_stranded(allocations, stranded) if not results: raise RuntimeError("Every allocator variant failed") diff --git a/src/python/omnimalloc/benchmark/results/utils.py b/src/python/omnimalloc/benchmark/results/utils.py index 7fa1916..cba1bcc 100644 --- a/src/python/omnimalloc/benchmark/results/utils.py +++ b/src/python/omnimalloc/benchmark/results/utils.py @@ -2,7 +2,6 @@ # SPDX-License-Identifier: Apache-2.0 # -import os import platform import subprocess from datetime import datetime @@ -11,6 +10,7 @@ from omnimalloc import __version__ from omnimalloc.benchmark.sources import BaseSource from omnimalloc.common.directories import PROJECT_DIR +from omnimalloc.common.parallel import available_cores def source_label(source: BaseSource | type[BaseSource] | str) -> str: @@ -45,5 +45,5 @@ def get_environment_metadata() -> dict[str, Any]: "omnimalloc_git_hash": _get_git_hash(), "os_info": f"{platform.system()} {platform.release()}", "cpu_info": platform.processor(), - "num_cores": os.cpu_count() or 1, + "num_cores": available_cores(), } diff --git a/src/python/omnimalloc/benchmark/sources/adversarial.py b/src/python/omnimalloc/benchmark/sources/adversarial.py index c6d7f4d..f9a1c19 100644 --- a/src/python/omnimalloc/benchmark/sources/adversarial.py +++ b/src/python/omnimalloc/benchmark/sources/adversarial.py @@ -70,7 +70,8 @@ class TwoPlusTwoSource(BaseSource): """Vector-clock instances that provably are not interval orders. Every group of four allocations is a 2+2, inducing a chordless 4-cycle no set - of intervals can realize, so `try_linearize` always returns None here. + of intervals can realize, so `try_linearize` always returns None here. Below + four allocations there is no room for the obstruction, and the source says so. """ _GROUP = 4 @@ -99,6 +100,12 @@ def get_allocations( self, num_allocations: int | None = None, skip: int = 0 ) -> tuple[Allocation, ...]: num = num_allocations if num_allocations is not None else self.num_allocations + # A truncated group is no obstruction, so the source cannot express a + # count below one: refuse rather than hand back a linearizable instance + if num < self._GROUP: + raise ValueError( + f"a 2+2 obstruction needs {self._GROUP} allocations, got {num}" + ) rng = random.Random(None if self.seed is None else self.seed + skip) obstructions = max(1, round(num * (1.0 - self.noise)) // self._GROUP) @@ -119,7 +126,7 @@ def get_allocations( while len(allocations) < num: lane = rng.randrange(2) - step = rng.randrange(self._GROUP * obstructions or 1) + step = rng.randrange(self._GROUP * obstructions) start = (step, 0) if lane == 0 else (0, step) end = (step + 1, 0) if lane == 0 else (0, step + 1) allocations.append( diff --git a/src/python/omnimalloc/common/intervals.py b/src/python/omnimalloc/common/intervals.py index 7533f25..22594b3 100644 --- a/src/python/omnimalloc/common/intervals.py +++ b/src/python/omnimalloc/common/intervals.py @@ -2,40 +2,77 @@ # SPDX-License-Identifier: Apache-2.0 # -from bisect import insort from collections.abc import Sequence from itertools import accumulate +from math import inf -def lowest_gap(occupied: list[tuple[int, int]], size: int) -> int: - """Lowest offset where `size` fits between the ascending occupied ranges.""" - cursor = 0 - for start, end in occupied: - if start - cursor >= size: - break - cursor = max(cursor, end) - return cursor +class FreeGaps: + """Maximal free ranges above a set of occupied ones, claimed first-fit. + + A claim takes space at a range's low end, so it only ever shrinks that range + and never splits one: the range count is fixed at what the occupied ranges + leave. That lets a max tree over the capacities answer each first fit in + O(log n), where rescanning the occupied ranges costs the whole set per claim. + """ + + def __init__(self, occupied: Sequence[tuple[int, int]]) -> None: + self._offsets = [] + capacities: list[float] = [] + cursor = 0 + for start, end in sorted(occupied): + if start > cursor: + self._offsets.append(cursor) + capacities.append(start - cursor) + cursor = max(cursor, end) + # The range above every occupied one is unbounded, so a claim that fits + # nowhere below still lands, and the tree root always admits a descent + self._offsets.append(cursor) + capacities.append(inf) + self._build(capacities) + + def _build(self, capacities: list[float]) -> None: + self._leaves = 1 << (len(capacities) - 1).bit_length() + self._tree: list[float] = [-inf] * (2 * self._leaves) + self._tree[self._leaves : self._leaves + len(capacities)] = capacities + for node in range(self._leaves - 1, 0, -1): + self._tree[node] = max(self._tree[2 * node], self._tree[2 * node + 1]) + + def claim(self, size: int) -> int: + """Lowest offset where `size` fits, marking that space occupied.""" + if size == 0: + return 0 # reserves nothing, and an empty range collides with nothing + node = 1 + while node < self._leaves: + node *= 2 + if self._tree[node] < size: + node += 1 + index = node - self._leaves + offset = self._offsets[index] + self._offsets[index] = offset + size + self._tree[node] -= size + node //= 2 + while node >= 1: + self._tree[node] = max(self._tree[2 * node], self._tree[2 * node + 1]) + node //= 2 + return offset def stack_around_pins(sizes: Sequence[int], offsets: Sequence[int | None]) -> list[int]: """Offset per item in input order: pinned ones keep theirs, the rest stack.""" - occupied = sorted( + occupied = [ (offset, offset + size) for size, offset in zip(sizes, offsets, strict=True) if offset is not None - ) + ] if not occupied: - # Sizes are positive, so no gap ever opens below the top and the scan - # returns the running total every time. Take it directly: the scan is - # quadratic and this is the hot path for the baseline allocators. + # Sizes are non-negative, so no gap ever opens below the top and every + # claim returns the running total. Take it directly: the baseline + # allocators run this path over millions of allocations. return list(accumulate(sizes, initial=0))[:-1] - resolved = [] - for size, offset in zip(sizes, offsets, strict=True): - if offset is not None: - resolved.append(offset) - continue - gap = lowest_gap(occupied, size) - resolved.append(gap) - insort(occupied, (gap, gap + size)) - return resolved + gaps = FreeGaps(occupied) + return [ + offset if offset is not None else gaps.claim(size) + for size, offset in zip(sizes, offsets, strict=True) + ] diff --git a/src/python/omnimalloc/common/parallel.py b/src/python/omnimalloc/common/parallel.py index 79c3309..2ba7185 100644 --- a/src/python/omnimalloc/common/parallel.py +++ b/src/python/omnimalloc/common/parallel.py @@ -2,10 +2,9 @@ # SPDX-License-Identifier: Apache-2.0 # -import os - from omnimalloc._cpp import max_threads as _max_threads from omnimalloc._cpp import set_max_threads as _set_max_threads +from omnimalloc._cpp import usable_cores as _usable_cores def set_max_threads(value: int | None) -> None: @@ -24,6 +23,15 @@ def max_threads() -> int: return _max_threads() +def adopt_max_threads(value: int) -> None: + """Worker-process entry point: take `value` as this process's ceiling. + + The ceiling lives in native process-global state, which a forked worker + inherits and a spawned one does not, so a pool passes it down explicitly. + """ + set_max_threads(value) + + def ensure_valid_num_threads(num_threads: int | None) -> None: """Raise ValueError if num_threads is not positive or None (disabled).""" if num_threads is not None and num_threads < 1: @@ -39,9 +47,7 @@ def available_cores() -> int: Under an affinity mask or a CPU-limited container `os.cpu_count()` still reports the whole machine, oversubscribing every pool by the ratio. """ - if hasattr(os, "sched_getaffinity"): - return len(os.sched_getaffinity(0)) or 1 - return os.cpu_count() or 1 + return _usable_cores() def resolve_num_threads(num_threads: int | None) -> int: diff --git a/src/python/omnimalloc/validate.py b/src/python/omnimalloc/validate.py index fd3633a..c055dec 100644 --- a/src/python/omnimalloc/validate.py +++ b/src/python/omnimalloc/validate.py @@ -98,14 +98,20 @@ def _check_size(memory: Memory, require_capacity: bool) -> None: raise ValueError(f"used size {memory.extent} exceeds memory size {memory.size}") +def _validate_memory( + memory: Memory, require_capacity: bool, alignment: int | None +) -> None: + _validate_pools(memory.pools, alignment) + _check_size(memory, require_capacity) + + def _validate_memories( memories: tuple[Memory, ...], require_capacity: bool, alignment: int | None ) -> None: _check_unique_ids(memories) for memory in memories: try: - _validate_pools(memory.pools, alignment) - _check_size(memory, require_capacity) + _validate_memory(memory, require_capacity, alignment) except ValueError as e: raise ValueError(f"in memory {memory.id!r}, {e}") from e @@ -131,7 +137,9 @@ def validate_allocation( if isinstance(entity, System): _validate_memories(entity.memories, require_capacity, alignment) elif isinstance(entity, Memory): - _validate_memories((entity,), require_capacity, alignment) + # Directly, not through `_validate_memories`: `described` already + # names this memory, and the wrapper would name it a second time + _validate_memory(entity, require_capacity, alignment) elif isinstance(entity, Pool): _validate_allocations(entity.allocations, alignment) else: diff --git a/tests/integration/test_torture.py b/tests/integration/test_torture.py index 6e4664b..a32a6c0 100644 --- a/tests/integration/test_torture.py +++ b/tests/integration/test_torture.py @@ -3,6 +3,7 @@ # import json +import os import random import shutil import subprocess @@ -109,6 +110,7 @@ CORE_COUNT_PROBE = """ import json +import os import random import sys @@ -163,11 +165,34 @@ """ +SPAWN_CEILING_PROBE = """ +import multiprocessing + +from omnimalloc.common.parallel import adopt_max_threads, max_threads, set_max_threads + + +def report(_arg): + return max_threads() + + +if __name__ == "__main__": + set_max_threads(2) + context = multiprocessing.get_context("spawn") + with context.Pool(1, initializer=adopt_max_threads, initargs=(2,)) as pool: + print(pool.map(report, [0])[0]) +""" + + class FailingVariant: def allocate(self, _allocations: tuple[Allocation, ...]) -> tuple[Allocation, ...]: raise RuntimeError("Variant failure") +class SuicidalVariant: + def allocate(self, _allocations: tuple[Allocation, ...]) -> tuple[Allocation, ...]: + os._exit(1) + + def _dense_instance(num_allocations: int, seed: int) -> tuple[Allocation, ...]: rng = random.Random(seed) horizon = max(1, num_allocations // 4) @@ -443,6 +468,32 @@ def test_one_failing_variant_does_not_sink_the_parallel_call() -> None: ) +def test_a_spawned_worker_inherits_the_thread_ceiling(tmp_path: Path) -> None: + # Spawn re-imports rather than copying the parent, so the native ceiling + # only reaches the worker if the pool hands it down. A script, not -c: the + # spawned child has to import __main__ to unpickle the task. + probe = tmp_path / "spawn_ceiling_probe.py" + probe.write_text(SPAWN_CEILING_PROBE) + completed = subprocess.run( + [sys.executable, str(probe)], + capture_output=True, + text=True, + timeout=300, + check=True, + ) + assert completed.stdout.strip() == "2" + + +def test_a_dying_worker_does_not_sink_the_parallel_call() -> None: + allocations = _small_instance(80, seed=3) + variants = (SuicidalVariant(), GreedyAllocator(), GreedyBySizeAllocator()) + placed = allocate_parallel(allocations, variants, num_threads=4) + assert placement_pressure(placed) == min( + placement_pressure(GreedyAllocator().allocate(allocations)), + placement_pressure(GreedyBySizeAllocator().allocate(allocations)), + ) + + def test_every_variant_failing_raises_runtime_error() -> None: allocations = _small_instance(80, seed=3) with pytest.raises(RuntimeError, match="Every allocator variant failed"): diff --git a/tests/unit/benchmark/sources/test_adversarial.py b/tests/unit/benchmark/sources/test_adversarial.py index b8fa253..3a776ed 100644 --- a/tests/unit/benchmark/sources/test_adversarial.py +++ b/tests/unit/benchmark/sources/test_adversarial.py @@ -85,6 +85,18 @@ def test_two_plus_two_generates_vector_clocks() -> None: assert all(a.dim == 2 for a in allocations) +def test_two_plus_two_refuses_a_count_below_one_obstruction() -> None: + for count in (1, 2, 3): + with pytest.raises(ValueError, match="a 2\\+2 obstruction needs"): + TwoPlusTwoSource(num_allocations=count).get_allocations() + + +def test_two_plus_two_never_linearizes_at_its_smallest_count() -> None: + allocations = TwoPlusTwoSource(num_allocations=4).get_allocations() + assert len(allocations) == 4 + assert try_linearize(allocations, work_budget=None) is None + + def test_sample_sizes_is_empty_for_a_non_positive_count() -> None: assert sample_sizes(random.Random(0), 0, "uniform", 1, 2) == [] diff --git a/tests/unit/common/test_intervals.py b/tests/unit/common/test_intervals.py index 6f76f5d..b9bede3 100644 --- a/tests/unit/common/test_intervals.py +++ b/tests/unit/common/test_intervals.py @@ -2,27 +2,53 @@ # SPDX-License-Identifier: Apache-2.0 # -from omnimalloc.common.intervals import lowest_gap, stack_around_pins +import random +from omnimalloc.common.intervals import FreeGaps, stack_around_pins -def test_lowest_gap_of_nothing_occupied_is_zero() -> None: - assert lowest_gap([], 16) == 0 +def test_free_gaps_of_nothing_occupied_claims_from_zero() -> None: + assert FreeGaps([]).claim(16) == 0 -def test_lowest_gap_takes_an_exactly_fitting_hole() -> None: - assert lowest_gap([(8, 16)], 8) == 0 +def test_free_gaps_takes_an_exactly_fitting_hole() -> None: + assert FreeGaps([(8, 16)]).claim(8) == 0 -def test_lowest_gap_skips_a_hole_one_short() -> None: - assert lowest_gap([(7, 16)], 8) == 16 +def test_free_gaps_skips_a_hole_one_short() -> None: + assert FreeGaps([(7, 16)]).claim(8) == 16 -def test_lowest_gap_lands_above_the_last_range_when_nothing_fits() -> None: - assert lowest_gap([(0, 8), (8, 24)], 16) == 24 +def test_free_gaps_lands_above_the_last_range_when_nothing_fits() -> None: + assert FreeGaps([(0, 8), (8, 24)]).claim(16) == 24 -def test_lowest_gap_tolerates_overlapping_occupied_ranges() -> None: - assert lowest_gap([(0, 32), (8, 16)], 8) == 32 + +def test_free_gaps_tolerates_overlapping_occupied_ranges() -> None: + assert FreeGaps([(0, 32), (8, 16)]).claim(8) == 32 + + +def test_free_gaps_tolerates_unsorted_occupied_ranges() -> None: + assert FreeGaps([(24, 32), (0, 8)]).claim(16) == 8 + + +def test_free_gaps_claiming_nothing_reserves_nothing() -> None: + gaps = FreeGaps([(8, 16)]) + assert gaps.claim(0) == 0 + assert gaps.claim(8) == 0 + + +def test_free_gaps_reuses_a_hole_a_later_claim_still_fits() -> None: + gaps = FreeGaps([(16, 24)]) + assert gaps.claim(12) == 0 + assert gaps.claim(4) == 12 + assert gaps.claim(4) == 24 + + +def test_free_gaps_keeps_claiming_from_the_hole_it_shrank() -> None: + gaps = FreeGaps([(0, 4), (20, 24)]) + assert gaps.claim(8) == 4 + assert gaps.claim(8) == 12 + assert gaps.claim(8) == 24 def test_stack_around_pins_stacks_when_nothing_is_pinned() -> None: @@ -48,3 +74,32 @@ def test_stack_around_pins_steps_over_a_pin_that_blocks_the_gap() -> None: def test_stack_around_pins_packs_several_free_items_around_two_pins() -> None: offsets = stack_around_pins([4, 4, 2, 2, 8], [0, 12, None, None, None]) assert offsets == [0, 12, 4, 6, 16] + + +def test_stack_around_pins_never_overlaps_a_pin_on_random_instances() -> None: + rng = random.Random(7) + for _ in range(200): + count = rng.randint(1, 24) + sizes = [rng.randint(0, 12) for _ in range(count)] + offsets = [rng.choice([None, None, 0, 5, 11, 24]) for _ in range(count)] + placed = stack_around_pins(sizes, offsets) + assert [p for p, o in zip(placed, offsets, strict=True) if o is not None] == [ + o for o in offsets if o is not None + ] + ranges = [ + (p, p + s, o is not None) + for p, s, o in zip(placed, sizes, offsets, strict=True) + if s > 0 + ] + for i, (lo, hi, pinned) in enumerate(ranges): + for other_lo, other_hi, other_pinned in ranges[i + 1 :]: + if pinned and other_pinned: + continue # colliding pins are the caller's to reject + assert hi <= other_lo or other_hi <= lo + + +def test_stack_around_pins_places_a_hundred_thousand_items_around_one_pin() -> None: + count = 100_000 + offsets = stack_around_pins([8] * count, [10**9, *([None] * (count - 1))]) + assert offsets[0] == 10**9 + assert offsets[1:] == [8 * i for i in range(count - 1)] diff --git a/tests/unit/common/test_parallel.py b/tests/unit/common/test_parallel.py index 72009d9..4ffcdd1 100644 --- a/tests/unit/common/test_parallel.py +++ b/tests/unit/common/test_parallel.py @@ -7,8 +7,10 @@ import pytest from omnimalloc import Allocation, allocate +from omnimalloc.allocators.greedy_base import _run_in_pool, _worker_ceiling from omnimalloc.analysis import placement_pressure from omnimalloc.common.parallel import ( + adopt_max_threads, available_cores, ensure_valid_num_threads, max_threads, @@ -17,6 +19,11 @@ ) +class _CeilingProbe: + def allocate(self, allocations: tuple[Allocation, ...]) -> tuple[Allocation, ...]: + return (allocations[0].with_offset(max_threads()),) + + @pytest.fixture(autouse=True) def restore_max_threads() -> Iterator[None]: original = max_threads() @@ -73,6 +80,26 @@ def test_ceiling_never_exceeds_the_usable_cores() -> None: assert max_threads() == available_cores() +def test_adopt_max_threads_sets_the_ceiling_of_a_worker_process() -> None: + adopt_max_threads(3) + assert max_threads() == min(3, available_cores()) + + +def test_worker_ceiling_splits_the_pool_share() -> None: + set_max_threads(8) + assert _worker_ceiling(4) == 2 + assert _worker_ceiling(8) == 1 + assert _worker_ceiling(64) == 1 + + +def test_pool_workers_run_under_the_split_ceiling() -> None: + set_max_threads(8) + allocations = (Allocation(id=0, size=1, start=0, end=1),) + results, stranded = _run_in_pool(allocations, [_CeilingProbe()] * 2, workers=2) + assert not stranded + assert [placed[0].offset for placed in results] == [min(4, available_cores())] * 2 + + def test_non_positive_ceiling_rejected() -> None: with pytest.raises(ValueError, match="max threads must be positive"): set_max_threads(0)