Budget placement pressure and localize the linearize witness

src/cpp/analysis/linearize.cpp

@@ -48,10 +48,42 @@ int64_t dominated_weight(const DedupedRows& ends,
   return count;
 }
 
-// Random 2+2 witnesses: ends e1, e2 and starts s1, s2 with e1 dominated by
-// s1 but not s2 and e2 dominated by s2 but not s1 prove two incomparable
+// A predecessor of `yes` that is not a predecessor of `no`, probed among
+// the `window` end rows just below the component-0 boundary: rows ascend on
+// component 0, and near-boundary ends are the likeliest to split two
+// predecessor sets apart.
+bool split_witness(const DedupedRows& ends, const int64_t* yes,
+                   const int64_t* no, size_t window) noexcept {
+  const size_t d = ends.dim;
+  size_t lo = 0;
+  size_t hi = ends.count();
+  while (lo < hi) {
+    const size_t mid = lo + (hi - lo) / 2;
+    if (ends.row(mid)[0] <= yes[0]) {
+      lo = mid + 1;
+    } else {
+      hi = mid;
+    }
+  }
+  const size_t begin = lo > window ? lo - window : 0;
+  for (size_t j = lo; j-- > begin;) {
+    if (dominates(ends.row(j), yes, d) && !dominates(ends.row(j), no, d)) {
+      return true;
+    }
+  }
+  return false;
+}
+
+// 2+2 witnesses: ends e1, e2 and starts s1, s2 with e1 dominated by s1 but
+// not s2 and e2 dominated by s2 but not s1 prove two incomparable
 // predecessor sets, so genuinely concurrent (non-linearizable) instances
-// bail here in microseconds. Deterministic seed keeps callers reproducible.
+// bail here instead of paying the full O(k * m * d) chain test. A random
+// global pass catches coarse concurrency in microseconds; concurrency in
+// realistic workloads is temporally local, so a deterministic
+// O(k * (log m + d)) pass then probes every lexicographically adjacent
+// start-row pair against the end windows just below them. Every hit is a
+// genuine 2+2; a miss just falls through to the exact chain test.
+// Deterministic seed keeps callers reproducible.
 bool find_incomparability_witness(const DedupedRows& starts,
                                   const DedupedRows& ends) {
   if (starts.count() < 2 || ends.count() < 2) {
@@ -72,6 +104,15 @@ bool find_incomparability_witness(const DedupedRows& starts,
       return true;
     }
   }
+  constexpr size_t kWindow = 16;
+  for (size_t p = 0; p + 1 < starts.count(); ++p) {
+    const int64_t* s1 = starts.row(p);
+    const int64_t* s2 = starts.row(p + 1);
+    if (split_witness(ends, s1, s2, kWindow) &&
+        split_witness(ends, s2, s1, kWindow)) {
+      return true;
+    }
+  }
   return false;
 }
 
@@ -100,14 +141,15 @@ std::optional<std::vector<std::pair<int64_t, int64_t>>> linearize_times(
   const size_t k = starts.count();
   const size_t m = ends.count();
 
-  if (find_incomparability_witness(starts, ends)) {
-    return std::nullopt;
-  }
   // Dominance counting costs O(k * m * d) before pruning; under a set
-  // budget, give up undecided instead of stalling the caller.
+  // budget, give up undecided instead of stalling the caller (and before
+  // spending the witness scan on an instance already refused).
   if (work_budget && static_cast<uint64_t>(k) * m * d > *work_budget) {
     return std::nullopt;
   }
+  if (find_incomparability_witness(starts, ends)) {
+    return std::nullopt;
+  }
 
   // Predecessor-set size per distinct start, with multiplicity
   std::vector<int64_t> counts(k);

src/cpp/analysis/placement.cpp

@@ -6,7 +6,9 @@
 
 #include <algorithm>
 #include <atomic>
+#include <cstdint>
 #include <functional>
+#include <optional>
 #include <set>
 #include <span>
 #include <stdexcept>
@@ -22,7 +24,9 @@
 // occupies disjoint address ranges below the neighborhood's top, so the
 // top certifies an upper bound without any search. Interval orders take
 // one single-timeline sweep with range-max queries; genuinely partial
-// orders take the pruned pairwise conflict sweep.
+// orders take the pruned pairwise conflict sweep. A set work budget bounds
+// the linearize attempt (which falls through silently) and the fallback
+// sweep (which throws), so huge vector-clock instances fail fast.
 
 namespace omnimalloc {
 
@@ -73,10 +77,9 @@ std::vector<int64_t> scalar_peaks(
   return peaks;
 }
 
-std::vector<int64_t> vector_peaks(const ClockSpans& spans,
+std::vector<int64_t> vector_peaks(const ConflictSweep& sweep,
                                   const std::vector<int64_t>& heights) {
-  const size_t n = spans.starts.size();
-  const ConflictSweep sweep(spans.starts, spans.ends, spans.dim);
+  const size_t n = heights.size();
   std::vector<std::atomic<int64_t>> top(n);
   for (size_t i = 0; i < n; ++i) {
     top[i].store(heights[i], std::memory_order_relaxed);
@@ -95,7 +98,8 @@ std::vector<int64_t> vector_peaks(const ClockSpans& spans,
 }  // namespace
 
 std::vector<int64_t> placement_pressure_per_allocation(
-    const std::vector<Allocation>& allocations) {
+    const std::vector<Allocation>& allocations,
+    std::optional<uint64_t> work_budget) {
   const size_t n = allocations.size();
   if (n == 0) {
     return {};
@@ -111,10 +115,16 @@ std::vector<int64_t> placement_pressure_per_allocation(
   }
   // Linearization preserves the conflict relation exactly, so neighborhood
   // tops transfer verbatim to the surrogate timeline.
-  if (const auto times = linearize_times(allocations, std::nullopt)) {
+  if (const auto times = linearize_times(allocations, work_budget)) {
     return scalar_peaks(*times, heights);
   }
-  return vector_peaks(spans, heights);
+  const ConflictSweep sweep(spans.starts, spans.ends, spans.dim);
+  if (work_budget && sweep.sweep_work() > *work_budget) {
+    throw std::runtime_error(
+        "Conflict sweep work exceeds work_budget; pass None to always "
+        "compute the placement pressure");
+  }
+  return vector_peaks(sweep, heights);
 }
 
 }  // namespace omnimalloc

src/cpp/analysis/placement.hpp

@@ -5,6 +5,7 @@
 #pragma once
 
 #include <cstdint>
+#include <optional>
 #include <vector>
 
 #include "primitives/allocation.hpp"
@@ -15,8 +16,11 @@ namespace omnimalloc {
 // the highest occupied address among each allocation and its conflict
 // neighbors, an upper bound on the pressure while it is live whose maximum
 // entry equals the placement's peak. Throws unless every allocation is
-// placed.
+// placed. A set `work_budget` (nullopt means unbounded) bounds both the
+// linearize attempt and the fallback conflict sweep; past it, throw instead
+// of stalling.
 [[nodiscard]] std::vector<int64_t> placement_pressure_per_allocation(
-    const std::vector<Allocation>& allocations);
+    const std::vector<Allocation>& allocations,
+    std::optional<uint64_t> work_budget);
 
 }  // namespace omnimalloc

src/cpp/bindings.cpp

@@ -144,8 +144,8 @@ NB_MODULE(_cpp, m) {
         "allocations"_a, "closure_cap"_a.none(),
         nb::call_guard<nb::gil_scoped_release>(), nb::rv_policy::move);
   m.def("placement_pressure_per_allocation", &placement_pressure_per_allocation,
-        "allocations"_a, nb::call_guard<nb::gil_scoped_release>(),
-        nb::rv_policy::move);
+        "allocations"_a, "work_budget"_a.none(),
+        nb::call_guard<nb::gil_scoped_release>(), nb::rv_policy::move);
   m.def("first_fit_place", &first_fit_place, "allocations"_a,
         nb::call_guard<nb::gil_scoped_release>(), nb::rv_policy::move);
 

src/python/omnimalloc/_cpp.pyi

@@ -90,7 +90,7 @@ def antichain_pressure_per_allocation(allocations: Sequence[Allocation], work_bu
 
 def closure_pressure_per_allocation(allocations: Sequence[Allocation], closure_cap: int | None) -> list[int]: ...
 
-def placement_pressure_per_allocation(allocations: Sequence[Allocation]) -> list[int]: ...
+def placement_pressure_per_allocation(allocations: Sequence[Allocation], work_budget: int | None) -> list[int]: ...
 
 def first_fit_place(allocations: Sequence[Allocation]) -> list[Allocation]: ...
 

src/python/omnimalloc/analysis/_pressure.py

@@ -111,17 +111,20 @@ def closure_pressure_per_allocation(
 
 
 def placement_pressure_per_allocation(
-    allocations: Sequence[Allocation],
+    allocations: Sequence[Allocation], work_budget: int | None = DEFAULT_WORK_BUDGET
 ) -> dict[IdType, int]:
     """Placement-certified peak over each allocation's lifetime, keyed by id.
 
     Read off assigned offsets: the highest occupied address among each
     allocation and its conflict neighbors, an upper bound on every exact
     per-allocation pressure whose max entry equals `placement_pressure`.
-    Raises `ValueError` on unplaced input.
+    Raises `ValueError` on unplaced input and `RuntimeError` once the
+    vector-clock conflict sweep exceeds `work_budget` (which also bounds
+    the internal linearize attempt); pass `None` to always compute.
     """
+    ensure_valid_budget(work_budget)
     ensure_unique_ids(allocations)
-    peaks = _placement_pressure_per_allocation(allocations)
+    peaks = _placement_pressure_per_allocation(allocations, work_budget)
     return _keyed_by_id(allocations, peaks)
 
 

tests/unit/analysis/test_pressure.py

@@ -329,6 +329,29 @@ def test_per_allocation_placement_pressure_max_equals_peak() -> None:
     assert max(peaks.values()) == 55
 
 
+def test_per_allocation_placement_pressure_budget_raises() -> None:
+    placed = (
+        Allocation(id="a", size=8, start=(0, 0), end=(1, 0), offset=96),
+        Allocation(id="b", size=16, start=(1, 0), end=(2, 0), offset=96),
+        Allocation(id="c", size=32, start=(0, 0), end=(0, 1), offset=0),
+        Allocation(id="d", size=64, start=(0, 1), end=(0, 2), offset=32),
+    )
+    with pytest.raises(RuntimeError, match="work_budget"):
+        placement_pressure_per_allocation(placed, work_budget=1)
+
+
+def test_per_allocation_placement_pressure_unbounded_budget_computes() -> None:
+    placed = (
+        Allocation(id="a", size=8, start=(0, 0), end=(1, 0), offset=96),
+        Allocation(id="b", size=16, start=(1, 0), end=(2, 0), offset=96),
+        Allocation(id="c", size=32, start=(0, 0), end=(0, 1), offset=0),
+        Allocation(id="d", size=64, start=(0, 1), end=(0, 2), offset=32),
+    )
+    expected = {"a": 104, "b": 112, "c": 112, "d": 112}
+    assert placement_pressure_per_allocation(placed, work_budget=None) == expected
+    assert placement_pressure_per_allocation(placed) == expected
+
+
 def _brute_antichain(allocations: tuple[Allocation, ...]) -> int:
     best = 0
     for count in range(1, len(allocations) + 1):