Skip to main content

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}