gam_runtime/resource.rs
1use crate::cgroup_memory::detect_cgroup_memory;
2pub use crate::cgroup_memory::{
3 CgroupMemoryAvailability, CgroupMemoryLimit, CgroupMemoryObservation, CgroupMemoryProbeFailure,
4 CgroupMemoryProbeFailureKind,
5};
6
7/// The library-default streamed row-chunk target (8 MiB), shared as a `const`
8/// so compile-time consumers (e.g. device tile geometry) stay in lockstep with
9/// [`ResourcePolicy::default_library`] without a runtime policy query.
10///
11/// **The 8 MiB value is unmeasured.** The paragraph above justifies there being
12/// ONE copy; it does not justify the number. No cache level, measured
13/// throughput, or consequence-for-being-wrong is recorded anywhere for it.
14/// Twelve transcribed copies across six crates were collapsed onto this
15/// definition (#2704) purely to remove duplication — that consolidation moved
16/// no value and measured nothing, and a single canonical copy of an arbitrary
17/// number must not be read as a decided one. Anyone retuning it is retuning an
18/// unsupported literal, not overriding a derived budget.
19///
20/// Consumers should size chunks with [`rows_for_target_bytes`], which performs
21/// the `target / (cols * 8)` division and floors the result at one row.
22///
23/// Re-declaring this value — as a module-local `const`, an inline literal, or
24/// any other spelling of 8 MiB — is caught by the workspace guard
25/// `tests/row_chunk_target_bytes_single_source_2704.rs`, which fails on any
26/// transcription of this value outside the exemption table it carries for the
27/// three sites that are a genuinely DIFFERENT quantity coinciding at 8 MiB.
28/// Import this constant instead, or derive from it under a name.
29pub const LIBRARY_ROW_CHUNK_TARGET_BYTES: usize = 8 * 1024 * 1024;
30
31/// Row-block size that keeps each streamed `rows × cols` `f64` chunk near the
32/// library row-chunk target, with a 512-row floor so a wide design still makes
33/// useful BLAS-3 progress per block, capped at the total row count and never
34/// zero (a zero-row chunk is not a chunk; every consumer strides by it). The
35/// ONE row-chunk rule for streamed dense kernels — CPU (`gam-solve`) and GPU
36/// (`gam-gpu`) alike — so the two cannot drift apart again (#2469, #2704).
37pub fn byte_balanced_row_chunk(cols: usize, n_rows: usize) -> usize {
38 const MIN_CHUNK_ROWS: usize = 512;
39 let bytes_per_row = cols.max(1) * std::mem::size_of::<f64>();
40 (LIBRARY_ROW_CHUNK_TARGET_BYTES / bytes_per_row)
41 .max(MIN_CHUNK_ROWS)
42 .min(n_rows.max(1))
43}
44
45#[derive(Clone, Debug)]
46pub struct ResourcePolicy {
47 pub max_single_materialization_bytes: usize,
48 pub max_operator_cache_bytes: usize,
49 pub max_spatial_distance_cache_bytes: usize,
50 pub max_owned_data_cache_bytes: usize,
51 pub row_chunk_target_bytes: usize,
52 pub derivative_storage_mode: DerivativeStorageMode,
53}
54
55pub const OWNED_DATA_CACHE_MAX_ENTRIES: usize = 2;
56
57// ─────────────────────────────────────────────────────────────────────────────
58// Process-wide memory governor
59// ─────────────────────────────────────────────────────────────────────────────
60
61/// Fraction of this process's memory **capacity** the governor is allowed to
62/// hand out as reservations: 3/4.
63///
64/// The ledger only accounts for the large, planned allocations that route
65/// through [`MemoryGovernor::try_reserve`] (dense design materializations,
66/// covariance blocks, sampler design assemblies). Everything else — allocator
67/// slack, thread stacks, code, and small per-iteration temporaries — lives in
68/// the remaining quarter.
69///
70/// The base quantity is *capacity* (`min(host total, binding cgroup hard
71/// limit)`), not free space, and that is the same separation #2684 established
72/// for the materialization ceiling — see
73/// `governor_budget_from_availability` for why the ledger cannot be an
74/// exception to it (#2702).
75const GOVERNOR_BUDGET_NUMERATOR: u128 = 3;
76const GOVERNOR_BUDGET_DENOMINATOR: u128 = 4;
77
78/// Which observation is the binding ceiling on memory available to this
79/// process.
80#[derive(Clone, Copy, Debug, PartialEq, Eq)]
81pub enum MemoryAvailabilitySource {
82 Host,
83 Cgroup,
84 HostAndCgroup,
85 CgroupProbeFailure,
86}
87
88impl std::fmt::Display for MemoryAvailabilitySource {
89 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
90 match self {
91 Self::Host => formatter.write_str("host"),
92 Self::Cgroup => formatter.write_str("cgroup"),
93 Self::HostAndCgroup => formatter.write_str("host and cgroup equally"),
94 Self::CgroupProbeFailure => formatter.write_str("cgroup probe failure"),
95 }
96 }
97}
98
99/// Provenance-preserving memory availability for the current process.
100///
101/// Two quantities live here and they answer different questions.
102///
103/// * **Available** (`available_bytes`) is *free space right now*: a finite
104/// cgroup-v1 or cgroup-v2 ceiling admits exactly `min(host available,
105/// reclaim-aware cgroup available)`. A v2 literal `memory.max = max` remains
106/// typed as unbounded and therefore defers to the host; v1's numeric
107/// unlimited sentinel participates exactly and likewise loses to a tighter
108/// host value. This quantity *moves*, and under pressure it goes to zero.
109/// * **Capacity** (`capacity_bytes`) is *how much memory this process could
110/// ever address*: `min(host total, the binding cgroup's hard limit)`. It is
111/// a property of the box and the cgroup configuration, not of the schedule,
112/// so it does not move when a sibling allocates.
113///
114/// The distinction is load-bearing (#2684). "Does this dense footprint fit
115/// here at all?" is a *capacity* question and its answer must be stationary —
116/// see [`process_memory_availability`] on why a route derived from a moving
117/// quantity is a defect. "Does this allocation fit *right now*?" is an
118/// availability question, and the joint ledger
119/// ([`MemoryGovernor::try_reserve`]) is the instrument for it. Reading the
120/// second where the first was meant is what let a cgroup a few pages from its
121/// limit refuse a 28,800-byte design on a host with 448 GB free.
122///
123/// If an active controller cannot be parsed exactly, admission fails closed
124/// with zero bytes — both available *and* capacity — while retaining the typed
125/// probe failure; it never silently inherits host capacity.
126#[derive(Clone, Debug, PartialEq, Eq)]
127pub struct MemoryAvailability {
128 host_available_bytes: u64,
129 host_total_bytes: u64,
130 cgroup: CgroupMemoryObservation,
131 available_bytes: u64,
132 capacity_bytes: u64,
133 limiting_source: MemoryAvailabilitySource,
134}
135
136impl MemoryAvailability {
137 pub(crate) fn from_observation(
138 host_available_bytes: u64,
139 host_total_bytes: u64,
140 cgroup: CgroupMemoryObservation,
141 ) -> Self {
142 use std::cmp::Ordering;
143
144 // Capacity never depends on what is currently resident: a finite
145 // cgroup ceiling clamps the host's total, an unbounded or absent
146 // controller defers to it, and a probe that failed closed admits
147 // nothing at all. A host whose reported available exceeds its reported
148 // total (a torn /proc/meminfo read) must not shrink capacity below
149 // what is demonstrably reachable, hence the max.
150 let capacity_bytes = match &cgroup {
151 CgroupMemoryObservation::NotPresent | CgroupMemoryObservation::V2Unbounded { .. } => {
152 host_total_bytes.max(host_available_bytes)
153 }
154 CgroupMemoryObservation::V2Limited(observation)
155 | CgroupMemoryObservation::V1Limited(observation) => host_total_bytes
156 .max(host_available_bytes)
157 .min(observation.limit_bytes()),
158 CgroupMemoryObservation::ProbeFailed(_) => 0,
159 };
160
161 let (available_bytes, limiting_source) = match &cgroup {
162 CgroupMemoryObservation::NotPresent | CgroupMemoryObservation::V2Unbounded { .. } => {
163 (host_available_bytes, MemoryAvailabilitySource::Host)
164 }
165 CgroupMemoryObservation::V2Limited(observation)
166 | CgroupMemoryObservation::V1Limited(observation) => {
167 match observation.available_bytes().cmp(&host_available_bytes) {
168 Ordering::Less => (
169 observation.available_bytes(),
170 MemoryAvailabilitySource::Cgroup,
171 ),
172 Ordering::Equal => (
173 host_available_bytes,
174 MemoryAvailabilitySource::HostAndCgroup,
175 ),
176 Ordering::Greater => (host_available_bytes, MemoryAvailabilitySource::Host),
177 }
178 }
179 CgroupMemoryObservation::ProbeFailed(_) => {
180 (0, MemoryAvailabilitySource::CgroupProbeFailure)
181 }
182 };
183 Self {
184 host_available_bytes,
185 host_total_bytes,
186 cgroup,
187 available_bytes,
188 capacity_bytes,
189 limiting_source,
190 }
191 }
192
193 /// The stationary ceiling on memory this process could ever address:
194 /// `min(host total, binding cgroup hard limit)`, or zero when the cgroup
195 /// probe failed closed. Unlike [`Self::available_bytes`] this does not move
196 /// when other work allocates, which is what makes it usable as a *routing*
197 /// threshold (#2684).
198 pub const fn capacity_bytes(&self) -> u64 {
199 self.capacity_bytes
200 }
201
202 pub const fn available_bytes(&self) -> u64 {
203 self.available_bytes
204 }
205
206 pub fn available_bytes_usize(&self) -> usize {
207 usize::try_from(self.available_bytes).unwrap_or(usize::MAX)
208 }
209
210}
211
212impl std::fmt::Display for MemoryAvailability {
213 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
214 match &self.cgroup {
215 CgroupMemoryObservation::ProbeFailed(failure) => write!(
216 formatter,
217 "0 bytes admitted because the active cgroup probe failed closed (host_available={}, failure={})",
218 self.host_available_bytes, failure,
219 ),
220 observation => write!(
221 formatter,
222 "{} bytes limited by {} (capacity={}, host_available={}, host_total={}, {})",
223 self.available_bytes,
224 self.limiting_source,
225 self.capacity_bytes,
226 self.host_available_bytes,
227 self.host_total_bytes,
228 observation,
229 ),
230 }
231 }
232}
233
234static MEMORY_AVAILABILITY_PROBES: std::sync::atomic::AtomicU64 =
235 std::sync::atomic::AtomicU64::new(0);
236
237/// Take a **fresh** reading of the OS and cgroup memory observations.
238///
239/// This is a syscall-bearing probe, not an accessor: it refreshes `sysinfo`
240/// (`/proc/meminfo`) and re-reads the cgroup limit/usage files on every call,
241/// and the value it returns *moves* — available memory is a live, shared
242/// quantity owned by the whole machine, not by this process.
243///
244/// Call this only where a live reading is the point (an OOM guard immediately
245/// before committing an allocation). Anything that *routes* — picks a
246/// strategy, sizes a plan, chooses an operator — must use
247/// [`process_memory_availability`] instead, because a route derived from a
248/// moving quantity makes the fitted answer depend on what else the box was
249/// doing (SPEC-20: a fit object must only ever come from a converged
250/// optimization, and reproducibly so).
251pub fn resample_memory_availability() -> MemoryAvailability {
252 static SYSTEM: OnceLock<Mutex<sysinfo::System>> = OnceLock::new();
253 let system = SYSTEM.get_or_init(|| Mutex::new(sysinfo::System::new()));
254 let mut system = system.lock().expect("sysinfo system mutex poisoned");
255 system.refresh_memory();
256 let cgroup = detect_cgroup_memory();
257 MEMORY_AVAILABILITY_PROBES.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
258 MemoryAvailability::from_observation(
259 system.available_memory(),
260 system.total_memory(),
261 cgroup,
262 )
263}
264
265/// The process's memory availability: **one** observation, taken once, shared
266/// by every planner in the process.
267///
268/// This is the same reading the [`MemoryGovernor`] sized its budget from, so a
269/// planner that admits a plan against this figure and the governor that then
270/// accounts the allocation cannot disagree about how much memory the process
271/// believes it has. Two consequences follow, and both are the point:
272///
273/// * **Free.** It is an atomic-free borrow of an already-initialized static —
274/// no syscalls, no parsing, no allocation — so it can sit on the hottest
275/// inner loop without a caching layer of its own.
276/// * **Constant.** Two runs of the same fit in one process see byte-identical
277/// budgets, so any route chosen from it is reproducible. Planners used to
278/// defend against the movement themselves (a monotone high-water floor in
279/// BMS, a per-term captured field in the SAE); a stationary primitive is what
280/// makes those compensations unnecessary rather than load-bearing.
281pub fn process_memory_availability() -> &'static MemoryAvailability {
282 &MemoryGovernor::global().ledger.availability
283}
284
285/// The process's available-memory figure in bytes, saturating to `usize`.
286/// Convenience over [`process_memory_availability`] for planners that only
287/// need the scalar.
288pub fn process_available_memory_bytes() -> usize {
289 process_memory_availability().available_bytes_usize()
290}
291
292/// Convert one provenance-preserving availability observation to the process
293/// ledger budget: 3/4 of the observation's stationary **capacity**. A probe
294/// that failed closed reports zero capacity and therefore admits nothing, which
295/// is the authoritative exhausted-memory signal.
296///
297/// gam#2702. This used to read `available_bytes` — free space — and that made
298/// every reservation verdict in the process a function of what the rest of the
299/// box happened to be doing, in exactly the way #2684 removed one layer up:
300///
301/// * the observation behind it is taken **once**, in the [`MemoryGovernor`]'s
302/// `OnceLock`, so it never tracked exhaustion in the first place. It froze
303/// whatever free memory existed at the instant of first governed allocation
304/// and used that number for the rest of the process's life. A frozen sample
305/// of a moving quantity is neither a live guard nor a reproducible
306/// threshold;
307/// * two processes launched from one job cgroup therefore derived different
308/// budgets from the same configuration, because page cache and sibling
309/// processes charge that cgroup continuously. Measured consequence: three
310/// `gam` inference tests passed in a 71-test run and failed in a 6-test
311/// subset of the same binary at the same commit, with
312/// `resource policy refused exact coefficient-SE columns 0..1` — a refusal of
313/// a few kilobytes.
314///
315/// The ledger's job is to bound **this process's own** jointly-live governed
316/// allocations against the ceiling this process could ever address. Memory
317/// already committed elsewhere on the box is not memory this ledger can account
318/// for, release, or route around; reading it only makes a fit's verdict a
319/// function of its neighbours, which SPEC-20 forbids (see
320/// [`process_memory_availability`]). Under a job scheduler the cgroup hard
321/// limit *is* the per-job ceiling, so capacity is the bound that actually binds.
322fn governor_budget_from_availability(availability: &MemoryAvailability) -> usize {
323 stationary_headroom_of_capacity(availability)
324}
325
326/// Convert the same observation to the process's **stationary** materialization
327/// ceiling: the answer to "could a dense footprint this large ever live here?",
328/// which must not move when a sibling allocates (#2684). A probe that failed
329/// closed reports zero capacity and therefore admits nothing.
330fn governor_materialization_cap_from_availability(availability: &MemoryAvailability) -> usize {
331 stationary_headroom_of_capacity(availability)
332}
333
334/// The one arithmetic both process-wide ceilings are derived from: the headroom
335/// fraction applied to stationary capacity.
336///
337/// The ledger budget and the materialization cap are therefore equal by
338/// construction rather than by coincidence, and that equality is the point:
339/// they are two questions about the same ceiling ("does this fit alongside what
340/// this process already has live" and "could this ever fit here at all"), so a
341/// routing decision taken from the cap can never be contradicted by the ledger
342/// for a reason that is not another live reservation. Keeping the arithmetic in
343/// one place is what stops the two from drifting back apart (#2684, #2702).
344fn stationary_headroom_of_capacity(availability: &MemoryAvailability) -> usize {
345 let scaled = u128::from(availability.capacity_bytes()) * GOVERNOR_BUDGET_NUMERATOR
346 / GOVERNOR_BUDGET_DENOMINATOR;
347 usize::try_from(scaled).unwrap_or(usize::MAX)
348}
349
350/// Typed refusal from [`MemoryGovernor::try_reserve`].
351///
352/// Carries the full ledger evidence so callers can route to a chunked or
353/// matrix-free strategy (and so error messages explain *why* dense was
354/// refused). This is a routing signal, never an abort: the process still has
355/// its unreserved headroom, the requested allocation just does not fit the
356/// joint budget.
357#[derive(Debug, Clone, thiserror::Error, PartialEq, Eq)]
358pub enum MemoryReservationError {
359 #[error(
360 "{context}: cannot reserve {requested_bytes} bytes; {reserved_bytes} of {budget_bytes} bytes already reserved process-wide; detected availability: {availability}"
361 )]
362 BudgetExceeded {
363 context: Box<str>,
364 requested_bytes: usize,
365 reserved_bytes: usize,
366 budget_bytes: usize,
367 availability: MemoryAvailability,
368 },
369
370 #[error(
371 "{context}: dense allocation size overflow for {copies} copies of a {nrows}x{ncols} f64 matrix"
372 )]
373 SizeOverflow {
374 context: Box<str>,
375 nrows: usize,
376 ncols: usize,
377 copies: usize,
378 },
379}
380
381#[derive(Debug)]
382struct GovernorLedger {
383 budget_bytes: usize,
384 materialization_cap_bytes: usize,
385 availability: MemoryAvailability,
386 reserved_bytes: std::sync::atomic::AtomicUsize,
387}
388
389/// Process-wide byte-accounting governor for large allocations.
390///
391/// Every large planned allocation (dense design materialization, covariance
392/// block, sampler design assembly) reserves its byte footprint against one
393/// shared ledger via [`try_reserve`](Self::try_reserve) and holds the returned
394/// RAII [`MemoryReservation`] for as long as the allocation is live. Because
395/// the ledger is shared, allocations that are each individually acceptable can
396/// no longer *jointly* exceed memory: whichever request would tip the ledger
397/// past the budget gets a typed [`MemoryReservationError`] and routes to a
398/// chunked or matrix-free strategy instead. Strategy selection is thereby a
399/// continuous function of predicted live bytes vs remaining budget, not of
400/// row/column thresholds.
401///
402/// The global budget is sized once from this process's stationary capacity
403/// (host total memory, clamped by the binding cgroup's hard limit inside
404/// containers or under a job scheduler) — see `GOVERNOR_BUDGET_NUMERATOR` for
405/// the headroom rationale and `governor_budget_from_availability` for why it
406/// is not denominated in free memory (#2702).
407#[derive(Debug, Clone)]
408pub struct MemoryGovernor {
409 ledger: Arc<GovernorLedger>,
410}
411
412impl MemoryGovernor {
413 /// The process-wide governor. Budget detection runs once, on first use.
414 ///
415 /// This `OnceLock` is where the process's single availability observation
416 /// is taken; [`process_memory_availability`] hands the same observation to
417 /// every other planner rather than probing again.
418 pub fn global() -> &'static MemoryGovernor {
419 static GLOBAL: OnceLock<MemoryGovernor> = OnceLock::new();
420 GLOBAL.get_or_init(|| {
421 let availability = resample_memory_availability();
422 MemoryGovernor::with_detected_availability(availability)
423 })
424 }
425
426 fn with_detected_availability(availability: MemoryAvailability) -> Self {
427 let budget_bytes = governor_budget_from_availability(&availability);
428 let materialization_cap_bytes = governor_materialization_cap_from_availability(&availability);
429 Self {
430 ledger: Arc::new(GovernorLedger {
431 budget_bytes,
432 materialization_cap_bytes,
433 availability,
434 reserved_bytes: std::sync::atomic::AtomicUsize::new(0),
435 }),
436 }
437 }
438
439 pub fn reserved_bytes(&self) -> usize {
440 self.ledger
441 .reserved_bytes
442 .load(std::sync::atomic::Ordering::Acquire)
443 }
444
445 /// Bytes still admissible: the budget less what is reserved *by this
446 /// process* right now. The only quantity here that moves is this process's
447 /// own live governed footprint — which is the one thing a caller can
448 /// actually route around.
449 pub fn remaining_bytes(&self) -> usize {
450 self.ledger
451 .budget_bytes
452 .saturating_sub(self.reserved_bytes())
453 }
454
455 /// Absolute ceiling for one governed operation: 3/4 of this process's
456 /// memory **capacity** (`min(host total, binding cgroup limit)`).
457 ///
458 /// This is a *routing* threshold — the question it answers is "could a
459 /// dense footprint this large ever live in this process?" — so it is
460 /// deliberately stationary. It is **not** the live budget: whether an
461 /// allocation fits *right now* is decided by [`Self::try_reserve`] against
462 /// the joint ledger, which returns a typed, routable refusal instead of an
463 /// abort. Since #2702 both are denominated in the same capacity, so the two
464 /// can only disagree because of a live reservation.
465 ///
466 /// Returning the live budget here was gam#2684: the budget was 3/4 of
467 /// *available* memory, so a cgroup sitting at its limit drove this ceiling
468 /// continuously to zero (measured at 53,248 bytes available with 448 GB
469 /// free on the host), and every caller comparing a request against it —
470 /// including `DenseDesignMatrix::to_dense`, which panics on refusal —
471 /// refused allocations as small as a 300x12 design. Threshold-shaped
472 /// routing decisions taken from a moving quantity also make the chosen
473 /// route depend on what else the box was doing, which SPEC-20 forbids
474 /// (see [`process_memory_availability`]); one consumer even hashes this
475 /// cap into a basis cache key.
476 pub fn single_materialization_cap_bytes(&self) -> usize {
477 self.ledger.materialization_cap_bytes
478 }
479
480 /// Reserve `bytes` against the joint ledger.
481 ///
482 /// On success the returned [`MemoryReservation`] must be held for as long
483 /// as the allocation it accounts for is live; dropping it releases the
484 /// bytes. On failure the caller receives the ledger evidence and is
485 /// expected to fall back to a chunked or matrix-free strategy.
486 pub fn try_reserve(
487 &self,
488 bytes: usize,
489 context: &str,
490 ) -> Result<MemoryReservation, MemoryReservationError> {
491 use std::sync::atomic::Ordering;
492 let mut current = self.ledger.reserved_bytes.load(Ordering::Relaxed);
493 loop {
494 let next = match current.checked_add(bytes) {
495 Some(next) if next <= self.ledger.budget_bytes => next,
496 _ => {
497 return Err(MemoryReservationError::BudgetExceeded {
498 context: context.into(),
499 requested_bytes: bytes,
500 reserved_bytes: current,
501 budget_bytes: self.ledger.budget_bytes,
502 availability: self.ledger.availability.clone(),
503 });
504 }
505 };
506 match self.ledger.reserved_bytes.compare_exchange_weak(
507 current,
508 next,
509 Ordering::AcqRel,
510 Ordering::Relaxed,
511 ) {
512 Ok(_) => {
513 return Ok(MemoryReservation {
514 ledger: Arc::clone(&self.ledger),
515 bytes,
516 });
517 }
518 Err(observed) => current = observed,
519 }
520 }
521 }
522
523 /// Reserve the footprint of a dense `nrows × ncols` `f64` matrix.
524 /// Dimension-product overflow is reported as a budget refusal (an
525 /// allocation whose size cannot even be computed certainly does not fit).
526 pub fn try_reserve_dense_f64(
527 &self,
528 nrows: usize,
529 ncols: usize,
530 context: &str,
531 ) -> Result<MemoryReservation, MemoryReservationError> {
532 self.try_reserve_dense_f64_copies(nrows, ncols, 1, context)
533 }
534
535 /// Reserve the predicted live footprint of `copies` simultaneous dense
536 /// matrices with one atomic ledger charge.
537 pub fn try_reserve_dense_f64_copies(
538 &self,
539 nrows: usize,
540 ncols: usize,
541 copies: usize,
542 context: &str,
543 ) -> Result<MemoryReservation, MemoryReservationError> {
544 let bytes = dense_f64_bytes(nrows, ncols)
545 .and_then(|one| one.checked_mul(copies))
546 .ok_or_else(|| MemoryReservationError::SizeOverflow {
547 context: context.into(),
548 nrows,
549 ncols,
550 copies,
551 })?;
552 self.try_reserve(bytes, context)
553 }
554}
555
556/// Checked byte footprint of a dense `nrows × ncols` `f64` matrix.
557pub const fn dense_f64_bytes(nrows: usize, ncols: usize) -> Option<usize> {
558 match nrows.checked_mul(ncols) {
559 Some(cells) => cells.checked_mul(std::mem::size_of::<f64>()),
560 None => None,
561 }
562}
563
564/// RAII guard for bytes reserved on a [`MemoryGovernor`] ledger; dropping it
565/// releases the reservation. Hold it exactly as long as the accounted
566/// allocation is live.
567#[derive(Debug)]
568#[must_use = "dropping a memory reservation immediately releases its ledger charge"]
569pub struct MemoryReservation {
570 ledger: Arc<GovernorLedger>,
571 bytes: usize,
572}
573
574impl MemoryReservation {
575 pub fn bytes(&self) -> usize {
576 self.bytes
577 }
578
579 /// Couple this reservation to the value whose memory it accounts for.
580 pub fn bind<T>(self, value: T) -> Governed<T> {
581 Governed {
582 value,
583 }
584 }
585}
586
587/// A value whose live memory is coupled to a process-wide reservation.
588///
589/// Large fallible materializations return this owner so the allocation cannot
590/// outlive its ledger charge. It dereferences to the wrapped value for normal
591/// ndarray and collection operations.
592#[derive(Debug)]
593#[must_use = "the governed value owns a live process-wide memory reservation"]
594pub struct Governed<T> {
595 value: T,
596}
597
598impl<T> std::ops::Deref for Governed<T> {
599 type Target = T;
600
601 fn deref(&self) -> &Self::Target {
602 &self.value
603 }
604}
605
606impl<T> std::ops::DerefMut for Governed<T> {
607 fn deref_mut(&mut self) -> &mut Self::Target {
608 &mut self.value
609 }
610}
611
612impl<T> AsRef<T> for Governed<T> {
613 fn as_ref(&self) -> &T {
614 &self.value
615 }
616}
617
618impl<T> AsMut<T> for Governed<T> {
619 fn as_mut(&mut self) -> &mut T {
620 &mut self.value
621 }
622}
623
624impl Drop for MemoryReservation {
625 fn drop(&mut self) {
626 self.ledger
627 .reserved_bytes
628 .fetch_sub(self.bytes, std::sync::atomic::Ordering::AcqRel);
629 }
630}
631
632/// Hints that flip strict mode on regardless of n/p — used when a code path
633/// is structurally operator-only and any dense fallback would be a bug.
634#[derive(Clone, Copy, Debug, Default)]
635pub struct ProblemHints {
636 pub marginal_slope_large_scale_active: bool,
637}
638
639#[derive(Clone, Copy, Debug, PartialEq, Eq)]
640pub enum DerivativeStorageMode {
641 /// Production exact-math: operator-backed, no dense fallback.
642 AnalyticOperatorRequired,
643 /// Allow dense materialization if under the single-materialization budget.
644 MaterializeIfSmall,
645 /// Dense materialization only permitted for diagnostic code paths.
646 DiagnosticsOnly,
647}
648
649#[derive(Clone, Debug)]
650pub struct MaterializationPolicy {
651 pub max_single_dense_bytes: usize,
652 pub max_cached_dense_bytes: usize,
653 pub row_chunk_target_bytes: usize,
654 pub allow_operator_materialization: bool,
655 pub allow_diagnostic_materialization: bool,
656}
657
658#[derive(Debug, thiserror::Error)]
659pub enum MatrixMaterializationError {
660 #[error(
661 "{context}: dense materialization of {nrows}x{ncols} requires {bytes} bytes (limit {limit_bytes})"
662 )]
663 TooLarge {
664 context: &'static str,
665 nrows: usize,
666 ncols: usize,
667 bytes: usize,
668 limit_bytes: usize,
669 },
670
671 #[error("{context}: operator does not implement chunked row access")]
672 MissingRowChunk { context: &'static str },
673
674 #[error("{context}: row materialization failed: {reason}")]
675 RowMaterializationFailed {
676 context: &'static str,
677 reason: String,
678 },
679
680 #[error("{context}: materialization forbidden by policy (mode={mode:?})")]
681 Forbidden {
682 context: &'static str,
683 mode: DerivativeStorageMode,
684 },
685
686 /// The process-wide [`MemoryGovernor`] could not reserve the requested live
687 /// footprint (or its checked byte size overflowed). Callers route to a
688 /// chunked or matrix-free strategy.
689 #[error(transparent)]
690 Reservation(#[from] MemoryReservationError),
691}
692
693pub trait ResidentBytes {
694 fn resident_bytes(&self) -> usize;
695}
696
697impl ResourcePolicy {
698 /// Conservative default suitable for general-purpose use.
699 ///
700 /// Uses `MaterializeIfSmall`: dense materialization is allowed only when
701 /// the matrix fits under `max_single_materialization_bytes`. This lets
702 /// small-data families that lack an implicit operator work out of the box,
703 /// while problems whose dense footprint does not fit real memory get a
704 /// typed refusal that forces the analytic-operator path. Set
705 /// `derivative_storage_mode = AnalyticOperatorRequired` explicitly to
706 /// reject all dense fallback.
707 ///
708 /// Scalar caps expose the governor's stationary capacity ceiling; they are
709 /// routing thresholds, not independent budgets, and they do not move with
710 /// load. Actual materializations and caches must reserve their complete
711 /// live footprint against the shared governor, so any combination of
712 /// categories is bounded by one ledger — that reservation, not these caps,
713 /// is what enforces "fits right now" (#2684).
714 pub fn default_library() -> Self {
715 Self::for_observed_memory(process_memory_availability())
716 }
717
718 /// The library-default policy a process would derive from `availability`.
719 ///
720 /// [`Self::default_library`] is exactly this function applied to the
721 /// process's own observation, so the routing thresholds a fit sees are a
722 /// pure function of the *observed environment* — and, because every scalar
723 /// below is taken from
724 /// [`MemoryAvailability::capacity_bytes`] rather than
725 /// [`MemoryAvailability::available_bytes`], a pure function of that
726 /// environment's stationary CAPACITY. Two observations of the same box that
727 /// differ only in how much memory happened to be free at the instant of the
728 /// probe therefore produce byte-identical policies, and any route selected
729 /// from one of these caps is reproducible across processes and across load
730 /// (#2684).
731 ///
732 /// Exposed rather than kept private so a test can hold capacity fixed and
733 /// vary free memory without racing the real machine; the live path uses the
734 /// same code, so the property under test is the shipped one.
735 pub fn for_observed_memory(availability: &MemoryAvailability) -> Self {
736 let single_cap = governor_materialization_cap_from_availability(availability);
737 Self {
738 max_single_materialization_bytes: single_cap,
739 max_operator_cache_bytes: single_cap,
740 max_spatial_distance_cache_bytes: single_cap,
741 max_owned_data_cache_bytes: single_cap,
742 row_chunk_target_bytes: LIBRARY_ROW_CHUNK_TARGET_BYTES,
743 derivative_storage_mode: DerivativeStorageMode::MaterializeIfSmall,
744 }
745 }
746
747 /// Strict mode that rejects every dense fallback. Use when you intend to
748 /// run only on operator-backed bases (large-scale Duchon/TPS, exact
749 /// GAMLSS marginal slope, CTN, etc.). The byte caps only govern the
750 /// residual diagnostic surfaces (materialization itself is forbidden by
751 /// the mode).
752 pub fn analytic_operator_required() -> Self {
753 let base = Self::default_library();
754 Self {
755 derivative_storage_mode: DerivativeStorageMode::AnalyticOperatorRequired,
756 ..base
757 }
758 }
759
760 /// Auto-derive the resource policy from the shape of the problem rather
761 /// than from an explicit CLI flag.
762 ///
763 /// Shape alone never flips a policy mode: doing so merely moves the old
764 /// row/column cliff to a byte threshold. Every non-structural path starts
765 /// permissive and makes its strategy decision from the operation's checked
766 /// predicted live bytes versus the governor's current remaining budget.
767 ///
768 /// `hints.marginal_slope_large_scale_active` forces strict mode regardless
769 /// of shape: that path is structurally operator-only and any dense
770 /// fallback would be a bug, not a memory question.
771 pub fn for_problem(hints: ProblemHints) -> Self {
772 if hints.marginal_slope_large_scale_active {
773 return Self::analytic_operator_required();
774 }
775 Self::default_library()
776 }
777
778 pub const fn material_policy(&self) -> MaterializationPolicy {
779 MaterializationPolicy {
780 max_single_dense_bytes: self.max_single_materialization_bytes,
781 max_cached_dense_bytes: self.max_operator_cache_bytes,
782 row_chunk_target_bytes: self.row_chunk_target_bytes,
783 allow_operator_materialization: matches!(
784 self.derivative_storage_mode,
785 DerivativeStorageMode::MaterializeIfSmall
786 ),
787 allow_diagnostic_materialization: !matches!(
788 self.derivative_storage_mode,
789 DerivativeStorageMode::AnalyticOperatorRequired
790 ),
791 }
792 }
793}
794
795/// Returns how many rows to stream per chunk so that each chunk uses approximately
796/// `target_bytes` given a row width of `cols` f64 entries.
797pub const fn rows_for_target_bytes(target_bytes: usize, cols: usize) -> usize {
798 let raw_bytes_per_row = cols.saturating_mul(std::mem::size_of::<f64>());
799 let bytes_per_row = if raw_bytes_per_row == 0 {
800 1
801 } else {
802 raw_bytes_per_row
803 };
804 let rows = target_bytes / bytes_per_row;
805 if rows == 0 { 1 } else { rows }
806}
807
808/// Select the row count for prediction-time covariance work.
809///
810/// A prediction row can keep roughly four `parameter_dim × local_dim`
811/// `f64` workspaces live while gradients and covariance solves are assembled.
812/// This central policy prevents predictors from drifting to different memory
813/// budgets and chunk bounds for the same operation.
814pub fn prediction_chunk_rows(parameter_dim: usize, local_dim: usize, total_rows: usize) -> usize {
815 const MIN_ROWS: usize = 16;
816 const MAX_ROWS: usize = 4096;
817
818 if total_rows == 0 {
819 return 1;
820 }
821 let live_f64_values_per_row = parameter_dim
822 .max(1)
823 .saturating_mul(local_dim.max(1))
824 .saturating_mul(4);
825 rows_for_target_bytes(
826 ResourcePolicy::default_library().row_chunk_target_bytes,
827 live_f64_values_per_row,
828 )
829 .clamp(MIN_ROWS, MAX_ROWS)
830 .min(total_rows)
831}
832
833use std::collections::{HashMap, VecDeque};
834use std::hash::{Hash, Hasher};
835use std::sync::{Arc, Mutex, OnceLock};
836
837/// Byte-limited LRU cache with an optional entry cap.
838///
839/// Unlike an entry-count-limited LRU, this cache tracks the resident byte cost
840/// of each value (via [`ResidentBytes`]) and evicts the least-recently-used
841/// entries until the total resident bytes fit under `max_bytes`. This is the
842/// correct policy for large-scale payloads where a single cache entry (e.g.
843/// an n*K distance matrix) can itself be multiple gigabytes and an entry-count
844/// cap would silently blow the memory budget. Small entry caps are still useful
845/// for payloads with known shape, such as owned PC data matrices shared across
846/// model blocks.
847pub struct ByteLruCache<K: Eq + Hash + Clone, V> {
848 /// One independent LRU partition per shard. A single shard (the default)
849 /// is byte-for-byte equivalent to the original single-`Mutex` cache; with
850 /// `shard_count > 1` the key hash selects the shard, so concurrent traffic
851 /// on distinct keys contends `1/shard_count` as often and each shard's
852 /// recency `VecDeque` is `1/shard_count` as long (the hit-path rescan is a
853 /// linear `position` lookup, so shrinking the per-shard order also cuts
854 /// per-access cost). Sharding is opt-in (`new_sharded`) precisely because
855 /// the byte budget is split across shards — that is correct for caches of
856 /// many small entries (e.g. cell-moment memos) but wrong for caches of a
857 /// few multi-GiB entries (distance matrices), which keep `shard_count == 1`.
858 shards: Box<[Mutex<ByteLruInner<K, V>>]>,
859 /// Per-shard byte budget. `shard_bytes * shards.len() >= max_bytes`.
860 shard_bytes: usize,
861 /// Per-shard entry budget, if any (`0` disables caching, as before).
862 shard_entries: Option<usize>,
863 max_bytes: usize,
864 governor: MemoryGovernor,
865}
866
867struct ByteLruInner<K, V> {
868 // The reservation is stored beside the value, so eviction and clear drop
869 // the process-wide charge at exactly the same time as cache ownership.
870 map: HashMap<K, (V, usize, MemoryReservation)>,
871 order: VecDeque<K>,
872 resident_bytes: usize,
873}
874
875impl<K: Eq + Hash + Clone, V: Clone + ResidentBytes> ByteLruCache<K, V> {
876 pub fn new(max_bytes: usize) -> Self {
877 Self::build(max_bytes, None, 1)
878 }
879
880 pub fn with_max_entries(max_bytes: usize, max_entries: usize) -> Self {
881 Self::build(max_bytes, Some(max_entries), 1)
882 }
883
884 /// Like [`new`](Self::new) but partitions the cache across `shard_count`
885 /// independently-locked LRU shards to cut lock contention under heavy
886 /// concurrent access. The byte budget is divided evenly across shards, so
887 /// this is only appropriate for caches holding many small entries.
888 pub fn new_sharded(max_bytes: usize, shard_count: usize) -> Self {
889 Self::build(max_bytes, None, shard_count)
890 }
891
892 /// Like [`with_max_entries`](Self::with_max_entries) but sharded; see
893 /// [`new_sharded`](Self::new_sharded).
894 pub fn with_max_entries_sharded(
895 max_bytes: usize,
896 max_entries: usize,
897 shard_count: usize,
898 ) -> Self {
899 Self::build(max_bytes, Some(max_entries), shard_count)
900 }
901
902 fn build(max_bytes: usize, max_entries: Option<usize>, shard_count: usize) -> Self {
903 Self::build_with_governor(
904 max_bytes,
905 max_entries,
906 shard_count,
907 MemoryGovernor::global().clone(),
908 )
909 }
910
911 fn build_with_governor(
912 max_bytes: usize,
913 max_entries: Option<usize>,
914 shard_count: usize,
915 governor: MemoryGovernor,
916 ) -> Self {
917 let shard_count = shard_count.max(1);
918 // Split the global budgets across shards, rounding up so the aggregate
919 // capacity never falls below the requested budget. With a single shard
920 // these equal the global budgets exactly (legacy behavior). A `0`
921 // entry budget still disables caching and must not be rounded up to 1.
922 let shard_bytes = max_bytes.div_ceil(shard_count);
923 let shard_entries = max_entries.map(|m| {
924 if m == 0 {
925 0
926 } else {
927 m.div_ceil(shard_count).max(1)
928 }
929 });
930 let shards = (0..shard_count)
931 .map(|_| {
932 Mutex::new(ByteLruInner {
933 map: HashMap::new(),
934 order: VecDeque::new(),
935 resident_bytes: 0,
936 })
937 })
938 .collect::<Vec<_>>()
939 .into_boxed_slice();
940 Self {
941 shards,
942 shard_bytes,
943 shard_entries,
944 max_bytes,
945 governor,
946 }
947 }
948
949 #[inline]
950 fn shard(&self, key: &K) -> &Mutex<ByteLruInner<K, V>> {
951 if self.shards.len() == 1 {
952 return &self.shards[0];
953 }
954 let mut hasher = std::collections::hash_map::DefaultHasher::new();
955 key.hash(&mut hasher);
956 &self.shards[(hasher.finish() as usize) % self.shards.len()]
957 }
958
959 pub fn get(&self, key: &K) -> Option<V> {
960 // recover from poison
961 let mut g = self.shard(key).lock().unwrap_or_else(|p| p.into_inner());
962 let v = g.map.get(key)?.0.clone();
963 // move to back (most-recently-used)
964 if let Some(pos) = g.order.iter().position(|k| k == key) {
965 let k = g
966 .order
967 .remove(pos)
968 .expect("position() returned an in-bounds index into this same deque");
969 g.order.push_back(k);
970 }
971 Some(v)
972 }
973
974 pub fn insert(&self, key: K, value: V) {
975 let charge = value.resident_bytes();
976 let mut g = self.shard(&key).lock().unwrap_or_else(|p| p.into_inner());
977
978 // If already present, remove the old entry first so resident bytes stay
979 // accurate and the LRU ordering reflects this insertion.
980 if let Some((_old, old_charge, _reservation)) = g.map.remove(&key) {
981 g.resident_bytes = g.resident_bytes.saturating_sub(old_charge);
982 if let Some(pos) = g.order.iter().position(|k| k == &key) {
983 g.order.remove(pos);
984 }
985 }
986
987 if charge > self.shard_bytes {
988 // Too large to cache; skip insertion.
989 return;
990 }
991
992 if let Some(max_entries) = self.shard_entries {
993 if max_entries == 0 {
994 return;
995 }
996 while g.map.len() >= max_entries {
997 if let Some(evict_key) = g.order.pop_front() {
998 if let Some((_v, c, _reservation)) = g.map.remove(&evict_key) {
999 g.resident_bytes = g.resident_bytes.saturating_sub(c);
1000 }
1001 } else {
1002 break;
1003 }
1004 }
1005 }
1006
1007 while g.resident_bytes + charge > self.shard_bytes {
1008 if let Some(evict_key) = g.order.pop_front() {
1009 if let Some((_v, c, _reservation)) = g.map.remove(&evict_key) {
1010 g.resident_bytes = g.resident_bytes.saturating_sub(c);
1011 }
1012 } else {
1013 break;
1014 }
1015 }
1016
1017 let reservation = match self
1018 .governor
1019 .try_reserve(charge, "ByteLruCache resident entry")
1020 {
1021 Ok(reservation) => reservation,
1022 Err(_) => return,
1023 };
1024 g.map.insert(key.clone(), (value, charge, reservation));
1025 g.order.push_back(key);
1026 g.resident_bytes = g.resident_bytes.saturating_add(charge);
1027 }
1028
1029 pub fn resident_bytes(&self) -> usize {
1030 self.shards
1031 .iter()
1032 .map(|shard| {
1033 shard
1034 .lock()
1035 .unwrap_or_else(|p| p.into_inner())
1036 .resident_bytes
1037 })
1038 .sum()
1039 }
1040
1041 pub const fn max_bytes(&self) -> usize {
1042 self.max_bytes
1043 }
1044
1045 pub fn len(&self) -> usize {
1046 self.shards
1047 .iter()
1048 .map(|shard| shard.lock().unwrap_or_else(|p| p.into_inner()).map.len())
1049 .sum()
1050 }
1051
1052 pub fn is_empty(&self) -> bool {
1053 self.len() == 0
1054 }
1055
1056 pub fn clear(&self) {
1057 for shard in self.shards.iter() {
1058 let mut g = shard.lock().unwrap_or_else(|p| p.into_inner());
1059 g.map.clear();
1060 g.order.clear();
1061 g.resident_bytes = 0;
1062 }
1063 }
1064}
1065
1066impl<K: Eq + Hash + Clone, V: Clone + ResidentBytes> std::fmt::Debug for ByteLruCache<K, V> {
1067 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1068 f.debug_struct("ByteLruCache")
1069 .field("resident_bytes", &self.resident_bytes())
1070 .field("max_bytes", &self.max_bytes)
1071 .field("shard_count", &self.shards.len())
1072 .field("shard_bytes", &self.shard_bytes)
1073 .field("shard_entries", &self.shard_entries)
1074 .finish()
1075 }
1076}
1077
1078/// Byte-accounting for `Arc<Array2<f64>>`.
1079///
1080/// Reports the full dense footprint of the owned array. Multiple `Arc`s
1081/// pointing to the same allocation will each report the full size; this is
1082/// the conservative accounting the caches want because a single residency in
1083/// the cache is what we are budgeting for.
1084impl ResidentBytes for Arc<ndarray::Array2<f64>> {
1085 fn resident_bytes(&self) -> usize {
1086 std::mem::size_of::<f64>()
1087 .saturating_mul(self.nrows())
1088 .saturating_mul(self.ncols())
1089 }
1090}
1091
1092/// Lazy-init cache safe to call from inside rayon par_iter.
1093///
1094/// `std::sync::OnceLock::get_or_init` parks racing threads on an OS
1095/// condition variable until the leader's init closure finishes. If the
1096/// leader's init closure itself dispatches a nested `into_par_iter`, the
1097/// parked threads are now unavailable as rayon workers, and the leader
1098/// blocks waiting for chunks that no one can service. Classic deadlock.
1099///
1100/// `RayonSafeOnce` removes the trap by computing the value *outside* any
1101/// lock. Concurrent racers may produce duplicate values; the first to
1102/// publish wins, the rest drop their result. No thread ever parks waiting
1103/// for another thread's init to finish, so nested rayon par_iter inside
1104/// the init closure is safe.
1105///
1106/// Use this in place of `OnceLock` whenever the init closure transitively
1107/// runs rayon work *and* the cache may be entered concurrently from
1108/// inside another rayon par_iter. The redundant-work cost on first race
1109/// is the price for never deadlocking; in practice the loser threads
1110/// throw away one round of work and steady-state is identical to
1111/// `OnceLock`.
1112pub struct RayonSafeOnce<T> {
1113 slot: std::sync::OnceLock<T>,
1114}
1115
1116impl<T> RayonSafeOnce<T> {
1117 pub const fn new() -> Self {
1118 Self {
1119 slot: std::sync::OnceLock::new(),
1120 }
1121 }
1122
1123 /// Returns the cached value if already populated.
1124 #[inline]
1125 pub fn get(&self) -> Option<&T> {
1126 self.slot.get()
1127 }
1128
1129 /// Returns the cached value, computing it if absent.
1130 ///
1131 /// The init closure runs WITHOUT holding any lock — calls from
1132 /// concurrent rayon workers may all run it, and all but the first
1133 /// to call `set` discard their result. This is the contract that
1134 /// keeps nested `into_par_iter` inside `init` from deadlocking on
1135 /// other workers parked on a `OnceLock`.
1136 ///
1137 /// Named `get_or_compute` (not `get_or_init`) so the codebase-level
1138 /// lint that bans `OnceLock::get_or_init` near rayon `par_iter` does
1139 /// not flag this safe-by-construction path.
1140 pub fn get_or_compute<F>(&self, init: F) -> &T
1141 where
1142 F: FnOnce() -> T,
1143 {
1144 if let Some(v) = self.slot.get() {
1145 return v;
1146 }
1147 let candidate = init();
1148 if self.slot.set(candidate).is_err() {
1149 log::trace!(
1150 "RayonSafeOnce: a concurrent initializer won the race; \
1151 keeping its value and discarding this candidate"
1152 );
1153 }
1154 self.slot
1155 .get()
1156 .expect("RayonSafeOnce slot populated by set() above")
1157 }
1158}
1159
1160impl<T> Default for RayonSafeOnce<T> {
1161 fn default() -> Self {
1162 Self::new()
1163 }
1164}
1165
1166impl<T: Clone> Clone for RayonSafeOnce<T> {
1167 fn clone(&self) -> Self {
1168 let cloned = Self::new();
1169 if let Some(value) = self.slot.get() {
1170 // `get_or_init` rather than `set(..).expect(..)`: the slot of a
1171 // freshly constructed `Self` is empty, so the initializer always
1172 // runs — and this spelling has no `Result` to discard, which keeps
1173 // the impl's bound at `T: Clone` instead of forcing `T: Debug` on
1174 // every caller just to name a panic that cannot happen.
1175 cloned.slot.get_or_init(|| value.clone());
1176 }
1177 cloned
1178 }
1179}
1180
1181impl<T: std::fmt::Debug> std::fmt::Debug for RayonSafeOnce<T> {
1182 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1183 f.debug_struct("RayonSafeOnce")
1184 .field("slot", &self.slot.get())
1185 .finish()
1186 }
1187}
1188
1189#[cfg(test)]
1190mod byte_lru_tests {
1191 use super::*;
1192
1193 fn cache_test_governor(budget_bytes: usize) -> MemoryGovernor {
1194 let available_bytes = (budget_bytes as u128 * GOVERNOR_BUDGET_DENOMINATOR)
1195 .div_ceil(GOVERNOR_BUDGET_NUMERATOR);
1196 let available_bytes =
1197 u64::try_from(available_bytes).expect("test cache budget must fit in u64");
1198 MemoryGovernor::with_detected_availability(MemoryAvailability::from_observation(
1199 available_bytes,
1200 available_bytes,
1201 CgroupMemoryObservation::NotPresent,
1202 ))
1203 }
1204
1205 /// Fixed-charge value so byte-budget arithmetic in the tests is exact.
1206 #[derive(Clone, PartialEq, Debug)]
1207 struct Payload(u64);
1208 impl ResidentBytes for Payload {
1209 fn resident_bytes(&self) -> usize {
1210 8
1211 }
1212 }
1213
1214 #[test]
1215 fn single_shard_round_trips_and_evicts_by_bytes() {
1216 // 3 entries' worth of budget; a single shard preserves strict global LRU.
1217 let cache: ByteLruCache<u64, Payload> =
1218 ByteLruCache::build_with_governor(24, None, 1, cache_test_governor(24));
1219 for k in 0..3 {
1220 cache.insert(k, Payload(k));
1221 }
1222 assert_eq!(cache.len(), 3);
1223 assert_eq!(cache.resident_bytes(), 24);
1224 // Touch key 0 so it is most-recently-used, then overflow by one.
1225 assert_eq!(cache.get(&0), Some(Payload(0)));
1226 cache.insert(3, Payload(3));
1227 // Key 1 (now least-recently-used) is evicted; 0 survives the touch.
1228 assert_eq!(cache.len(), 3);
1229 assert_eq!(cache.get(&1), None);
1230 assert_eq!(cache.get(&0), Some(Payload(0)));
1231 assert_eq!(cache.get(&3), Some(Payload(3)));
1232 }
1233
1234 #[test]
1235 fn zero_entry_budget_disables_caching_in_every_shard() {
1236 let single: ByteLruCache<u64, Payload> = ByteLruCache::with_max_entries(1 << 20, 0);
1237 single.insert(7, Payload(7));
1238 assert_eq!(single.get(&7), None);
1239 let sharded: ByteLruCache<u64, Payload> =
1240 ByteLruCache::with_max_entries_sharded(1 << 20, 0, 16);
1241 sharded.insert(7, Payload(7));
1242 assert_eq!(sharded.get(&7), None);
1243 }
1244
1245 #[test]
1246 fn sharded_cache_retrieves_all_keys_and_respects_aggregate_budget() {
1247 // Generous budget split across 8 shards; every inserted key must be
1248 // retrievable and the aggregate residency must never exceed the global
1249 // budget (shard_bytes * shard_count, rounded up).
1250 let shard_count = 8usize;
1251 let max_bytes = 8 * 64; // 64 entries' worth, 8 per shard on average.
1252 let cache: ByteLruCache<u64, Payload> = ByteLruCache::build_with_governor(
1253 max_bytes,
1254 None,
1255 shard_count,
1256 cache_test_governor(max_bytes),
1257 );
1258 for k in 0..64u64 {
1259 cache.insert(k, Payload(k));
1260 }
1261 // Per-shard budgets sum to >= the requested global budget.
1262 assert!(cache.resident_bytes() <= max_bytes.div_ceil(shard_count) * shard_count);
1263 // Re-inserting then reading back a key returns the stored payload.
1264 cache.insert(123, Payload(123));
1265 assert_eq!(cache.get(&123), Some(Payload(123)));
1266 assert!(!cache.is_empty());
1267 cache.clear();
1268 assert_eq!(cache.len(), 0);
1269 assert_eq!(cache.resident_bytes(), 0);
1270 }
1271}
1272
1273#[cfg(test)]
1274mod resource_policy_tests {
1275 use super::*;
1276
1277 // ── rows_for_target_bytes ─────────────────────────────────────────────────
1278
1279 #[test]
1280 fn rows_for_target_bytes_exact_fit() {
1281 // 1 col × 8 bytes/f64; target 8 bytes → 1 row
1282 assert_eq!(rows_for_target_bytes(8, 1), 1);
1283 }
1284
1285 #[test]
1286 fn rows_for_target_bytes_multiple_rows() {
1287 // 1 col × 8 bytes/f64; target 80 bytes → 10 rows
1288 assert_eq!(rows_for_target_bytes(80, 1), 10);
1289 }
1290
1291 #[test]
1292 fn rows_for_target_bytes_multiple_cols() {
1293 // 4 cols × 8 = 32 bytes/row; target 128 → 4 rows
1294 assert_eq!(rows_for_target_bytes(128, 4), 4);
1295 }
1296
1297 #[test]
1298 fn rows_for_target_bytes_zero_target_returns_one() {
1299 // Zero target cannot give 0 rows — floor to 1
1300 assert_eq!(rows_for_target_bytes(0, 1), 1);
1301 }
1302
1303 #[test]
1304 fn rows_for_target_bytes_zero_cols_returns_non_zero() {
1305 // Zero cols → bytes_per_row falls back to 1 → rows = target
1306 assert_eq!(rows_for_target_bytes(100, 0), 100);
1307 }
1308
1309 #[test]
1310 fn rows_for_target_bytes_large_target() {
1311 // Canonical target, 1024 cols → 8192 bytes/row → rows = target/8192.
1312 // Taken by name rather than transcribed, so the case stays pinned to the
1313 // library target if that target ever moves (#2704).
1314 let target = LIBRARY_ROW_CHUNK_TARGET_BYTES;
1315 let cols = 1024_usize;
1316 let expected = target / (cols * std::mem::size_of::<f64>());
1317 assert_eq!(rows_for_target_bytes(target, cols), expected);
1318 }
1319
1320 #[test]
1321 fn prediction_chunks_share_the_runtime_byte_budget() {
1322 assert_eq!(prediction_chunk_rows(1024, 1, 100_000), 256);
1323 assert_eq!(prediction_chunk_rows(32, 2, 100_000), 4096);
1324 }
1325
1326 #[test]
1327 fn prediction_chunks_respect_dataset_bounds() {
1328 assert_eq!(prediction_chunk_rows(1, 1, 7), 7);
1329 assert_eq!(prediction_chunk_rows(1, 1, 0), 1);
1330 }
1331
1332 // ── ResourcePolicy::for_problem ──────────────────────────────────────────
1333
1334 #[test]
1335 fn for_problem_small_data_uses_materialize_if_small() {
1336 let p = ResourcePolicy::for_problem(ProblemHints::default());
1337 assert_eq!(
1338 p.derivative_storage_mode,
1339 DerivativeStorageMode::MaterializeIfSmall
1340 );
1341 }
1342
1343 #[test]
1344 fn for_problem_has_no_row_or_column_cliff() {
1345 let narrow = ResourcePolicy::for_problem(ProblemHints::default());
1346 let wide = ResourcePolicy::for_problem(ProblemHints::default());
1347 assert_eq!(
1348 narrow.derivative_storage_mode,
1349 DerivativeStorageMode::MaterializeIfSmall
1350 );
1351 assert_eq!(
1352 wide.derivative_storage_mode,
1353 DerivativeStorageMode::MaterializeIfSmall
1354 );
1355 }
1356
1357 #[test]
1358 fn for_problem_dimension_overflow_defers_to_typed_reservation() {
1359 let policy = ResourcePolicy::for_problem(ProblemHints::default());
1360 assert_eq!(
1361 policy.derivative_storage_mode,
1362 DerivativeStorageMode::MaterializeIfSmall
1363 );
1364 }
1365
1366 #[test]
1367 fn for_problem_marginal_slope_hint_is_strict() {
1368 let p = ResourcePolicy::for_problem(ProblemHints {
1369 marginal_slope_large_scale_active: true,
1370 });
1371 assert_eq!(
1372 p.derivative_storage_mode,
1373 DerivativeStorageMode::AnalyticOperatorRequired
1374 );
1375 }
1376
1377 // ── ResourcePolicy::material_policy ─────────────────────────────────────
1378
1379 #[test]
1380 fn material_policy_default_library_allows_operator_and_diagnostics() {
1381 let mp = ResourcePolicy::default_library().material_policy();
1382 assert!(mp.allow_operator_materialization);
1383 assert!(mp.allow_diagnostic_materialization);
1384 }
1385
1386 #[test]
1387 fn material_policy_analytic_operator_required_blocks_both() {
1388 let mp = ResourcePolicy::analytic_operator_required().material_policy();
1389 assert!(!mp.allow_operator_materialization);
1390 assert!(!mp.allow_diagnostic_materialization);
1391 }
1392
1393 #[test]
1394 fn material_policy_propagates_byte_limits() {
1395 let policy = ResourcePolicy::default_library();
1396 let mp = policy.material_policy();
1397 assert_eq!(
1398 mp.max_single_dense_bytes,
1399 policy.max_single_materialization_bytes
1400 );
1401 assert_eq!(mp.max_cached_dense_bytes, policy.max_operator_cache_bytes);
1402 assert_eq!(mp.row_chunk_target_bytes, policy.row_chunk_target_bytes);
1403 }
1404
1405 // ── MemoryGovernor ledger ────────────────────────────────────────────────
1406
1407 #[test]
1408 fn compressed_macos_observation_keeps_xnu_available_memory_positive() {
1409 // #2316's healthy 8 GiB macOS host had more compressed than
1410 // free+inactive pages. sysinfo 0.33 subtracted compressor pages and
1411 // saturated to zero; 0.38 follows XNU and reports
1412 // (active + inactive + free) * page_size.
1413 let xnu_available = (75_514_u64 + 69_056 + 3_802) * 16_384;
1414 assert_eq!(xnu_available, 2_430_926_848);
1415 let availability = MemoryAvailability::from_observation(
1416 xnu_available,
1417 8 * 1024 * 1024 * 1024,
1418 CgroupMemoryObservation::NotPresent,
1419 );
1420 assert_eq!(availability.available_bytes(), xnu_available);
1421 // Capacity is the host's 8 GiB, so the budget is 6 GiB (#2702); the
1422 // 2.43 GB free figure decides nothing but the availability provenance.
1423 assert_eq!(availability.capacity_bytes(), 8 * 1024 * 1024 * 1024);
1424 assert_eq!(
1425 governor_budget_from_availability(&availability),
1426 6 * 1024 * 1024 * 1024
1427 );
1428 }
1429
1430 #[test]
1431 fn literal_unlimited_cgroup_defers_to_host_available_memory_2317() {
1432 let availability = MemoryAvailability::from_observation(
1433 2_430_926_848,
1434 4_000_000_000,
1435 CgroupMemoryObservation::V2Unbounded {
1436 cgroup_path: "/fixture/leaf".into(),
1437 inspected_levels: 3,
1438 },
1439 );
1440 assert_eq!(availability.available_bytes(), 2_430_926_848);
1441 assert_eq!(availability.capacity_bytes(), 4_000_000_000);
1442 assert_eq!(availability.limiting_source, MemoryAvailabilitySource::Host);
1443 assert!(format!("{availability}").contains("unbounded cgroup-v2"));
1444 let governor = MemoryGovernor::with_detected_availability(availability);
1445 assert_eq!(governor.remaining_bytes(), 3_000_000_000);
1446 assert_eq!(governor.single_materialization_cap_bytes(), 3_000_000_000);
1447 // Exercise the actual ledger. A literal unlimited controller delegates
1448 // to the finite host; it does not authorize unbounded reservations.
1449 let reservation = governor
1450 .try_reserve(3_000_000_000, "host budget")
1451 .unwrap();
1452 assert_eq!(governor.remaining_bytes(), 0);
1453 assert!(matches!(
1454 governor.try_reserve(1, "one byte beyond the host budget"),
1455 Err(MemoryReservationError::BudgetExceeded { .. })
1456 ));
1457 drop(reservation);
1458 assert_eq!(governor.remaining_bytes(), 3_000_000_000);
1459 }
1460}
1461
1462/// gam#2702: a reservation verdict is a function of the request and this
1463/// process's own live footprint — never of what else was on the box, and never
1464/// of what this process happened to do earlier.
1465///
1466/// The filed incident: three `gam` inference tests passed in a 71-test run of
1467/// the `inference` binary and failed in a 6-test subset of that same binary at
1468/// that same commit, all three with
1469/// `resource policy refused exact coefficient-SE columns 0..1`. The refused
1470/// allocation was two dense f64 copies of one column — kilobytes. The ledger
1471/// budget was `3/4 x FREE memory` sampled once, at whichever moment this
1472/// process first touched the governor, so processes launched from one job
1473/// cgroup derived different budgets according to how much page cache and how
1474/// many sibling test processes were charged to that cgroup at their own instant
1475/// of first touch.
1476///
1477/// #2684 had already removed exactly this mechanism from the materialization
1478/// ceiling. What follows asserts the ledger is not an exception, on the shipped
1479/// derivation, with the pre-fix arithmetic spelled out as the falsification
1480/// control.
1481#[cfg(test)]
1482mod governor_budget_is_capacity_determined_2702_tests {
1483 use super::*;
1484
1485 /// The MSI compute node the incident was measured on: a `--mem=8g` job on a
1486 /// box with hundreds of GB free.
1487 const HOST_AVAILABLE_BYTES: u64 = 448_648_040_448;
1488 const HOST_TOTAL_BYTES: u64 = 527_799_400 * 1024;
1489 const JOB_LIMIT_BYTES: u64 = 8 * 1024 * 1024 * 1024;
1490
1491 /// The refused allocation, restated from the failing path
1492 /// (`gam-solve` `optimizer.rs`): the exact factorized coefficient-SE solve
1493 /// reserves two dense `p_t x chunk` f64 workspaces at once. `p_t = 512` and
1494 /// `chunk = 1` is the smallest request that path can ever make on a model of
1495 /// this width — one column at a time.
1496 const SE_TRANSFORMED_ROWS: usize = 512;
1497 /// One column, two copies, eight bytes per f64.
1498 const SE_CHUNK_BYTES: usize = SE_TRANSFORMED_ROWS * 8 * 2;
1499
1500 /// One cgroup read at load `charged`, everything else held fixed.
1501 fn one_job_cgroup_at_load(charged: u64) -> MemoryAvailability {
1502 crate::test_support::simulated_cgroup_memory_environment(
1503 HOST_AVAILABLE_BYTES,
1504 HOST_TOTAL_BYTES,
1505 JOB_LIMIT_BYTES,
1506 charged,
1507 )
1508 }
1509
1510 #[test]
1511 fn a_cgroup_at_its_limit_moves_neither_the_budget_nor_the_materialization_cap_2684_2702() {
1512 const LIMIT: u64 = 6 * 1024 * 1024 * 1024;
1513 const DESIGN_BYTES: usize = 300 * 12 * 8;
1514 let budget = LIMIT as usize / 4 * 3;
1515 // The incident reading had 53,248 bytes available in a 6 GiB job.
1516 // Also cover a fully charged job and an idle one: only availability
1517 // changes. These reserve ledger entries, never actual GiB allocations.
1518 for charged in [0, LIMIT - 53_248, LIMIT] {
1519 let availability = crate::test_support::simulated_cgroup_memory_environment(
1520 HOST_AVAILABLE_BYTES,
1521 HOST_TOTAL_BYTES,
1522 LIMIT,
1523 charged,
1524 );
1525 assert_eq!(availability.available_bytes(), LIMIT - charged);
1526 assert_eq!(availability.capacity_bytes(), LIMIT);
1527 let governor = MemoryGovernor::with_detected_availability(availability);
1528 assert_eq!(governor.remaining_bytes(), budget);
1529 assert_eq!(governor.single_materialization_cap_bytes(), budget);
1530 assert!(governor.single_materialization_cap_bytes() > DESIGN_BYTES);
1531 assert!(matches!(
1532 governor.try_reserve(8 << 30, "larger than the job"),
1533 Err(MemoryReservationError::BudgetExceeded { .. })
1534 ));
1535 let chunk = governor
1536 .try_reserve_dense_f64_copies(64, 1, 2, "coefficient-SE solve chunk")
1537 .expect("the incident's 1,024-byte chunk fits at every ambient load");
1538 assert_eq!(chunk.bytes(), 1_024);
1539 assert_eq!(governor.remaining_bytes(), budget - 1_024);
1540 assert_eq!(governor.single_materialization_cap_bytes(), budget);
1541 drop(chunk);
1542 assert_eq!(governor.remaining_bytes(), budget);
1543 }
1544 // A governor that always grants requests must fail this control.
1545 let tiny = crate::test_support::simulated_cgroup_memory_environment(
1546 HOST_AVAILABLE_BYTES,
1547 HOST_TOTAL_BYTES,
1548 1_024,
1549 0,
1550 );
1551 let governor = MemoryGovernor::with_detected_availability(tiny);
1552 assert_eq!(governor.single_materialization_cap_bytes(), 768);
1553 assert!(matches!(
1554 governor.try_reserve(DESIGN_BYTES, "300 by 12 design"),
1555 Err(MemoryReservationError::BudgetExceeded { .. })
1556 ));
1557 }
1558
1559 #[test]
1560 fn a_request_larger_than_the_job_is_still_refused_at_every_load() {
1561 // The ceiling must keep saying no, or the test above is satisfied by a
1562 // governor that admits everything.
1563 for charged in [0, JOB_LIMIT_BYTES / 2, JOB_LIMIT_BYTES - 4_096] {
1564 let governor = MemoryGovernor::with_detected_availability(one_job_cgroup_at_load(charged));
1565 let refusal = governor
1566 .try_reserve(16 * 1024 * 1024 * 1024, "twice the job's ceiling")
1567 .expect_err("16 GiB cannot be admitted in an 8 GiB job");
1568 match refusal {
1569 MemoryReservationError::BudgetExceeded { budget_bytes, .. } => {
1570 assert_eq!(budget_bytes, (JOB_LIMIT_BYTES as usize) / 4 * 3);
1571 }
1572 other => panic!("expected a budget refusal naming the ceiling, got {other:?}"),
1573 }
1574 }
1575 }
1576
1577 #[test]
1578 fn the_verdict_does_not_depend_on_what_this_process_did_earlier() {
1579 // The property the issue asks for, stated as history-independence: a
1580 // request's verdict must be the same whether or not the process has
1581 // already built and dropped something large. Reservations are the only
1582 // process state the ledger has, and a released one must leave no trace.
1583 let governor = MemoryGovernor::with_detected_availability(one_job_cgroup_at_load(0));
1584 let request = || {
1585 governor
1586 .try_reserve_dense_f64_copies(
1587 SE_TRANSFORMED_ROWS,
1588 1,
1589 2,
1590 "factorized coefficient-SE solve chunk",
1591 )
1592 .map(|reservation| reservation.bytes())
1593 };
1594
1595 let before = request().expect("admissible on a fresh ledger");
1596 {
1597 // Something big enough that a free-denominated ledger would have
1598 // been left visibly poorer by it: all but one SE chunk of the budget.
1599 let bulk = governor
1600 .try_reserve(
1601 governor.remaining_bytes() - SE_CHUNK_BYTES,
1602 "prior work in this process",
1603 )
1604 .expect("the bulk reservation is exactly the remaining budget");
1605 assert_eq!(governor.remaining_bytes(), SE_CHUNK_BYTES);
1606 // While it is live the ledger honestly reports the pressure: this is
1607 // the one thing that MAY change a verdict, and the next request still
1608 // fits because the bulk left exactly one chunk.
1609 assert!(request().is_ok());
1610 drop(bulk);
1611 }
1612 assert_eq!(governor.reserved_bytes(), 0);
1613 assert_eq!(request().expect("admissible again"), before);
1614 }
1615}