Skip to main content

mz_ore/
pool.rs

1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License in the LICENSE file at the
6// root of this repository, or online at
7//
8//     http://www.apache.org/licenses/LICENSE-2.0
9//
10// Unless required by applicable law or agreed to in writing, software
11// distributed under the License is distributed on an "AS IS" BASIS,
12// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13// See the License for the specific language governing permissions and
14// limitations under the License.
15
16//! Prototype buffer pool for dataflow state. See
17//! `doc/developer/design/20260610_buffer_managed_state.md`.
18//!
19//! The pool is the cache: size-class anonymous virtual-memory regions whose
20//! slots hold resident chunks. Slots are scoped to residency — eviction
21//! returns a chunk's slot to the free list along with its physical pages —
22//! so slot demand tracks the resident set (bounded by the budget), not the
23//! potentially unbounded live backlog. Reads are copy-out
24//! ([`ChunkHandle::read_into`]): a resident slot is copied and an evicted
25//! extent decompressed straight into the caller's buffer, all under the
26//! chunk's state lock, so no reference into pool memory escapes the pool and
27//! a read leaves residency untouched. The backing is the swap-backed extent
28//! store of the design's Layer 1: a slot in a pool-owned anonymous-memory
29//! extent arena holding the chunk's lz4-compressed bytes.
30//!
31//! Memory descends a ladder of tiers, each with its own ceiling and each
32//! cheaper to vacate than the one above:
33//!
34//! * **Slots** (uncompressed, free reads) — bounded by the budget; crossing
35//!   it compresses the oldest chunks into extents and releases their slots.
36//! * **Warm free slots** (pages kept for fault-free reuse) — bounded by the
37//!   warm cap.
38//! * **Compressed-resident extents** (reads decompress, no device) — bounded
39//!   by the headroom the RSS target leaves above the first two; crossing it
40//!   pushes the oldest extents to the swap device with `MADV_PAGEOUT`.
41//! * **The swap device** — overflow; reads fault and decompress.
42//!
43//! The identity `total pool RSS <= budget + warm cap + compressed cap`,
44//! where `compressed cap = max(0, rss_target - budget - warm cap)`, makes
45//! every resident byte's ceiling nameable. A zero RSS target (or one below
46//! the budget plus warm cap) collapses the compressed tier and extents page
47//! out as soon as they are written. Heap-backed chunks (oversize payloads
48//! and class-exhaustion fallbacks) sit outside the identity: they can never
49//! be evicted, so the budget is enforced against evictable bytes only.
50//! Slots never reach the swap device: only compressed extents are offered
51//! to it. Pageout is observed rather than assumed: an extent counts against
52//! the compressed tier until the page table shows its whole range unmapped,
53//! so reclaim the kernel declines surfaces as `extent_pageout_incomplete`
54//! (and a tier settled above its capacity) instead of as RSS the ledger
55//! cannot see.
56//!
57//! Residency is a state, not a type. It descends through eviction and
58//! ascends through exactly one transition: an admitting read
59//! ([`ChunkHandle::read_into_admit`]) lifts an evicted chunk back to
60//! `BackedResident` when a slot is free within the budget or stealable from
61//! a clean backed victim of the same class, never by evicting or
62//! compressing anything. Plain reads ([`ChunkHandle::read_into`]) leave
63//! residency untouched. Eviction I/O runs on spill threads when enabled —
64//! `WriteInFlight` marks a chunk whose compression a spill thread owns — and
65//! inline on the evicting caller otherwise. Chunks are immutable after
66//! [`Pool::insert_with`], which is what makes a `BackedResident` slot always
67//! identical to its extent and its eviction free of I/O.
68//!
69//! Freeing an `UnbackedResident` chunk is a pure memory operation — the
70//! design's "never write dead data" win, surfaced as `writes_elided` in
71//! [`PoolStats`]. Budget pressure evicts cold chunks via second-chance
72//! FIFOs banded by the caller-supplied generational depth ([`ChunkHints`]):
73//! eviction and eager backing both visit deeper (colder) bands first, so
74//! the youngest data keeps its die-before-write chance longest. Within a
75//! band the FIFO is the design's backstop policy, and unannotated chunks
76//! (depth 0) get exactly that.
77
78mod extent;
79mod region;
80
81use std::collections::VecDeque;
82use std::ops::Range;
83use std::sync::atomic::{AtomicU64, Ordering};
84use std::sync::{Arc, Mutex, MutexGuard, Weak};
85
86use crate::cast::CastFrom;
87use crate::pool::extent::{ExtentArena, Scratch, SwapExtent};
88use crate::pool::region::{Region, SIZE_CLASSES};
89
90/// Virtual reservation per size class. Purely virtual: physical memory
91/// materializes only for slots in use, and slots are scoped to residency,
92/// so this must exceed the largest plausible *resident* set per class, the
93/// budget plus in-flight slack, not the backlog. It is deliberately enormous
94/// (address space costs nothing, and touched pages are bounded by peak
95/// residency) so that no realistic budget, on any machine size, reaches the
96/// heap-fallback path.
97///
98/// NOTE: Seen OoMs with Miri since it actually allocates the capacity.
99const CLASS_CAPACITY_BYTES: usize = if cfg!(miri) { 16 << 20 } else { 1 << 40 };
100
101/// A chunk-provided transform between a chunk's body bytes and the stored
102/// bytes its extent holds. The pool owns scheduling: spill threads, the
103/// residency state machine, cancellation, and the ledger. It invokes the
104/// codec on opaque bytes at the extent boundary, `encode` when backing a
105/// chunk (on a spill thread, or inline under overload) and `decode` when
106/// reading an evicted one, under the chunk's state lock. The pool itself
107/// has no opinion on the stored form: framing, compression, and validation
108/// all belong to the codec.
109///
110/// Implementations must be pure transforms: no locking, no calls back into
111/// the pool (the state lock is held at `decode` sites), and no panic on
112/// bytes their own `encode` produced. `decode` must exactly invert
113/// `encode`, and `encode`'s output must never exceed
114/// [`max_stored_len`]`(body.len())`, the bound the extent store's size
115/// classes are provisioned to.
116pub trait ExtentCodec: std::fmt::Debug + Send + Sync {
117    /// Transforms `body` into its stored form, replacing `out`'s contents.
118    /// `out`'s capacity is reused across calls; implementations size it
119    /// themselves.
120    fn encode(&self, body: &[u8], out: &mut Vec<u8>);
121
122    /// Inverts [`ExtentCodec::encode`]: reconstructs into `body` exactly
123    /// the bytes whose encoding produced `stored`. `body` is exactly the
124    /// original body's length, and implementations must panic on a length
125    /// mismatch rather than truncate or pad.
126    fn decode(&self, stored: &[u8], body: &mut [u8]);
127}
128
129/// The largest stored form [`ExtentCodec::encode`] may produce for a
130/// `body_len`-byte body: an incompressible-input expansion matching lz4's
131/// worst case plus a four-byte length prefix. The extent store's size-class
132/// ladder is provisioned to this bound, so a codec that exceeds it can
133/// strand payloads with no class to hold them (they degrade to unpageable
134/// heap fallbacks).
135pub fn max_stored_len(body_len: usize) -> usize {
136    4 + body_len + body_len / 255 + 16
137}
138
139/// Advisory placement hints for a chunk, supplied at insert and immutable
140/// thereafter (merges mint new chunks, so a chunk's generation never
141/// changes). Hints steer policy — eviction order and write-behind
142/// candidacy — never correctness: a mislabeled chunk performs worse, while
143/// the budget and residency invariants hold regardless.
144#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
145pub struct ChunkHints {
146    /// Generational depth of the chunk in its producer's merge structure,
147    /// 0 for the youngest generation (and the unannotated default). Deeper
148    /// chunks are treated as colder: preferred write-behind candidates and
149    /// preferred eviction victims, cheap to evict once backed.
150    pub depth: u8,
151}
152
153/// Number of depth bands the eviction queues are split into; depths at or
154/// beyond the last band share it.
155const DEPTH_BANDS: usize = 4;
156
157/// The eviction-queue band for a chunk of `depth`.
158fn band(depth: u8) -> usize {
159    usize::from(depth).min(DEPTH_BANDS - 1)
160}
161
162/// Residency state of a chunk.
163#[derive(Debug, Clone, Copy, PartialEq, Eq)]
164enum Residency {
165    /// Lives only in the pool; no extent copy exists. Freeing it never
166    /// touches the backing store.
167    UnbackedResident,
168    /// Resident, and an identical extent copy exists; eviction releases
169    /// physical pages without I/O.
170    BackedResident,
171    /// Resident and readable, with compression into an extent scheduled on a
172    /// spill thread. Completion moves an evicting chunk to
173    /// [`Residency::Evicted`] and an eagerly backed one to
174    /// [`Residency::BackedResident`]; a free observed at dequeue cancels the
175    /// write instead.
176    WriteInFlight,
177    /// Extent copy only; the chunk holds no slot. The extent itself may
178    /// still be RAM-resident (the compressed tier) or paged out to the swap
179    /// device. Reads decompress the extent straight into the caller's
180    /// buffer and leave the chunk evicted, except that an admitting read
181    /// may lift it back to [`Residency::BackedResident`].
182    Evicted,
183    /// Larger than the largest size class; held as a plain heap allocation,
184    /// always resident. A prototype limitation, not a design state.
185    Oversize,
186}
187
188/// Snapshot of pool counters.
189#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
190pub struct PoolStats {
191    /// Chunks inserted.
192    pub inserts: u64,
193    /// Chunks freed (handle dropped).
194    pub frees: u64,
195    /// Backing writes elided: chunks dead before their compression
196    /// completed, so no extent write happened. Covers chunks freed while
197    /// `UnbackedResident` and chunks freed while queued for a spill thread
198    /// that had not yet compressed them.
199    pub writes_elided: u64,
200    /// Evictions that compressed the chunk into a new extent.
201    pub evictions_compress: u64,
202    /// Evictions of `BackedResident` chunks: pure page release, no I/O.
203    pub evictions_cheap: u64,
204    /// Compressed bytes written into extents.
205    pub extent_bytes_written: u64,
206    /// Evictions handed to spill threads.
207    pub spill_scheduled: u64,
208    /// Compressions cancelled because the chunk was freed while queued or
209    /// in flight, whatever scheduled them. Eager-backing work counts here
210    /// but never in `spill_scheduled`, so this can exceed that counter.
211    pub spill_cancelled: u64,
212    /// Entries currently queued for or being processed by spill threads.
213    pub spill_in_flight: u64,
214    /// Inserts that fell back to the heap because their size class had no
215    /// free slot (the live set outgrew the class reservation). Heap-backed
216    /// chunks behave like oversize ones: always resident, never paged.
217    pub slot_exhausted_fallbacks: u64,
218    /// Inserts whose payload exceeded the largest size class and therefore
219    /// went straight to a heap-backed oversize chunk.
220    pub oversize_payloads: u64,
221    /// Live size-classed chunks across all classes, whatever their residency.
222    /// For backlog-shaped consumers this tracks the un-drained backlog in
223    /// chunks. (Slots are scoped to residency, so the quantity that exhausts
224    /// a class reservation is the resident subset, bounded by the budget.)
225    pub live_chunks: u64,
226    /// Uncompressed bytes of currently resident chunks (including oversize).
227    pub resident_bytes: u64,
228    /// Uncompressed bytes of live oversize chunks.
229    pub oversize_bytes: u64,
230    /// Class bytes of free slots currently kept warm (pages resident for
231    /// fault-free reuse). Bounded by a fraction of the budget; RSS exceeds
232    /// `resident_bytes` by up to this amount.
233    pub warm_bytes: u64,
234    /// Slot allocations served from the warm list: reuses that faulted no
235    /// pages and skipped the kernel's page zeroing.
236    pub warm_reuses: u64,
237    /// Chunks eagerly compressed to `BackedResident` by idle spill threads
238    /// (write-behind): still readable in their slots, with eviction
239    /// pre-paid.
240    pub eager_backs: u64,
241    /// Evicted chunks re-admitted to `BackedResident` by an admitting read
242    /// out of free budget headroom.
243    pub admissions_budget: u64,
244    /// Evicted chunks re-admitted to `BackedResident` by an admitting read
245    /// stealing the slot of a clean backed victim of the same size class.
246    /// The victim becomes `Evicted` with zero I/O and its extent intact.
247    pub admissions_steal: u64,
248    /// Admitting reads of evicted chunks served as a plain decompress
249    /// instead: no budget headroom (or an exhausted size class), and no
250    /// clean victim whose growth the budget could absorb.
251    pub admissions_denied: u64,
252    /// Allocation bytes of compressed extents currently resident — the
253    /// compressed-but-resident middle tier. Bounded by the RSS target;
254    /// exceeding it pages the oldest extents out to the swap device.
255    pub extent_resident_bytes: u64,
256    /// Allocation bytes of resident extents the RSS target cannot push out:
257    /// retry-capped arena extents (the kernel declined the reclaim advice
258    /// until the retry budget ran out) and heap-fallback extents. The
259    /// compressed tier settles above its capacity by this amount. Climbing
260    /// steadily means pages cannot reach the swap device (no swap, or a
261    /// cgroup that cannot reclaim).
262    pub extent_unreclaimable_bytes: u64,
263    /// Extents pushed to the swap device by RSS-target enforcement, with
264    /// the whole range observed nonresident afterwards.
265    pub extent_pageouts: u64,
266    /// Pageout passes whose observation found some of the extent's pages
267    /// still mapped: `MADV_PAGEOUT` may decline pages and still succeed, so
268    /// the page table decides. The extent keeps its full resident
269    /// accounting and is retried until its per-extent retry cap. Climbing
270    /// steadily on a loaded pool means pages cannot actually be unmapped to
271    /// the swap device (no swap, or a cgroup that cannot reclaim).
272    pub extent_pageout_incomplete: u64,
273    /// Extent writes that fell back to the heap because their extent-arena
274    /// class had no free slot. Heap-backed extents stay readable but are
275    /// never paged out, so their compressed bytes hold RAM until freed.
276    pub extent_arena_fallbacks: u64,
277}
278
279#[derive(Debug, Default)]
280struct Counters {
281    inserts: AtomicU64,
282    spill_scheduled: AtomicU64,
283    spill_cancelled: AtomicU64,
284    slot_exhausted_fallbacks: AtomicU64,
285    oversize_payloads: AtomicU64,
286    frees: AtomicU64,
287    writes_elided: AtomicU64,
288    evictions_compress: AtomicU64,
289    evictions_cheap: AtomicU64,
290    extent_bytes_written: AtomicU64,
291    resident_bytes: AtomicU64,
292    oversize_bytes: AtomicU64,
293    warm_bytes: AtomicU64,
294    warm_reuses: AtomicU64,
295    eager_backs: AtomicU64,
296    admissions_budget: AtomicU64,
297    admissions_steal: AtomicU64,
298    admissions_denied: AtomicU64,
299    extent_resident_bytes: AtomicU64,
300    extent_unreclaimable_bytes: AtomicU64,
301    extent_pageouts: AtomicU64,
302    extent_pageout_incomplete: AtomicU64,
303}
304
305/// A buffer pool over swap-backed extents. Cheap to clone; all clones share
306/// one budget and one backing store.
307#[derive(Debug, Clone)]
308pub struct Pool(Arc<PoolInner>);
309
310/// The shared state behind every [`Pool`] handle: the budget and RSS
311/// ledgers, the size-class slot regions and the extent arena, the eviction
312/// and backing queues, and the spill-thread hand-off. One per process in
313/// practice; [`Pool`] clones and chunk handles share it through an `Arc`,
314/// so it lives until the last handle and spill thread release it.
315///
316/// Lock order: a chunk's `state` mutex may be held while taking any of the
317/// leaf locks — the eviction `queue`, the `extent_queue`, the spill queue,
318/// and the region slot allocators — but never the reverse. The enforcement
319/// and backing scans additionally drop the queue guard before trying a
320/// chunk's state lock (and only ever `try_lock` it), so no path holds a
321/// queue lock while waiting on chunk state. The admitting read's victim
322/// steal is the one place a chunk's state lock is held while probing
323/// another chunk's, and the victim is only ever `try_lock`ed, so two
324/// admitters stealing toward each other skip instead of deadlocking. Reads
325/// copy out under the chunk's state lock — the same lock eviction takes —
326/// so there is no reader-side count and no reader the evictor must account
327/// for.
328#[derive(Debug)]
329struct PoolInner {
330    /// Resident-bytes target, enforced against evictable bytes (resident
331    /// minus heap-backed, which no eviction can reclaim). Atomic so a
332    /// running pool can be retuned in place (operator-driven budget
333    /// changes) without orphaning live handles, which share this value
334    /// through their `Arc<PoolInner>`.
335    budget_bytes: AtomicU64,
336    /// Ceiling on the pool's *total* RSS: slots (the budget) plus warm free
337    /// slots plus compressed-resident extents. The compressed tier's
338    /// capacity derives as `max(0, rss_target - budget - warm cap)`; zero
339    /// (the default) collapses the tier, paging every extent out as soon as
340    /// it is written.
341    rss_target_bytes: AtomicU64,
342    /// One region per entry of [`SIZE_CLASSES`], same order.
343    regions: Vec<Region>,
344    /// The arena backing extents. Shared with every live [`SwapExtent`],
345    /// whose drop returns its slot.
346    extent_arena: Arc<ExtentArena>,
347    /// Second-chance FIFOs of eviction candidates, one per depth band; a
348    /// chunk joins the band of its [`ChunkHints`] depth at insert and again
349    /// on re-admission. Entries for freed chunks go stale in place and are
350    /// dropped by [`PoolInner::prune_queues`].
351    ///
352    /// Two scanners walk them with different obligations, both visiting the
353    /// deepest band first. Budget enforcement is the one that ages chunks:
354    /// it spends the touched bit (second chance) and drops entries it
355    /// evicts. Eager backing ([`PoolInner::back_one`]) rotates visited
356    /// entries to the back but never spends a touched bit, so a backing
357    /// pass shuffles FIFO order without aging any chunk toward eviction.
358    queues: [Mutex<VecDeque<Weak<ChunkMeta>>>; DEPTH_BANDS],
359    /// FIFO of chunks whose extents are resident, oldest first — the
360    /// RSS-target enforcement's victim queue. Entries go stale when an
361    /// extent pages out, is dropped, or its chunk dies; visits drop them,
362    /// and [`PoolInner::prune_extent_queue`] compacts dead-chunk entries
363    /// that under-cap operation never visits.
364    extent_queue: Mutex<VecDeque<Weak<ChunkMeta>>>,
365    /// Number of live size-classed chunks (whatever their residency), which
366    /// is the number of non-stale queue entries across all bands;
367    /// [`PoolInner::prune_queues`] compacts the queues against it.
368    live_chunks: AtomicU64,
369    /// Number of live chunks whose extent is currently resident, including
370    /// unreclaimable extents that deliberately hold no `extent_queue` entry
371    /// (heap-backed and retry-capped ones). It therefore upper-bounds the
372    /// queue's non-stale entries, and [`PoolInner::prune_extent_queue`]'s
373    /// compaction threshold is conservative by the unreclaimable count.
374    extent_residents: AtomicU64,
375    /// Single-flight claim for budget enforcement.
376    enforcing: Mutex<()>,
377    /// Set by an insert turned away from `enforcing`. The holder re-runs its
378    /// pass while it is set, so a caller turned away after the holder's final
379    /// counter read still has its bytes enforced rather than dropped.
380    enforce_pending: std::sync::atomic::AtomicBool,
381    counters: Counters,
382    spill: Spill,
383}
384
385/// Hand-off point between budget enforcement and spill threads. Eviction I/O
386/// (compression and the synchronous-reclaim `pageout`) runs on spill threads
387/// when enabled, keeping multi-millisecond work off the threads that trip the
388/// budget; with no spill threads, eviction runs inline on the caller.
389#[derive(Debug, Default)]
390struct Spill {
391    /// Chunks in `WriteInFlight`, awaiting a spill thread.
392    queue: Mutex<VecDeque<Arc<ChunkMeta>>>,
393    /// Parks idle spill threads. Notified when work lands in `queue`, when
394    /// eager backing turns on, and at shutdown; threads additionally wake
395    /// on a timeout so eager backing scans for write-behind work without a
396    /// dedicated wakeup per candidate.
397    cv: std::sync::Condvar,
398    /// Whether evictions are handed to spill threads. Set when threads are
399    /// first spawned; cleared to fall back to inline eviction.
400    enabled: std::sync::atomic::AtomicBool,
401    /// Whether idle spill threads eagerly compress unbacked chunks to
402    /// `BackedResident` (write-behind): the chunk stays readable in its slot
403    /// while a compressed extent accumulates on the swap device, so a later
404    /// budget-driven eviction is a pure page release instead of a
405    /// compression. Costs CPU on chunks that die before eviction would have
406    /// reached them; pays at every pressure event.
407    eager: std::sync::atomic::AtomicBool,
408    /// Number of spill threads spawned (spawn-once; later config changes
409    /// only toggle `enabled`).
410    threads: AtomicU64,
411    /// Queued plus currently-processing entries; `quiesce` waits on zero.
412    in_flight: AtomicU64,
413    /// Test-only lifecycle: production spill threads are immortal (the pool
414    /// is a process singleton), but Miri rejects a test binary exiting with
415    /// live threads, so tests stop and join them.
416    #[cfg(test)]
417    stop: std::sync::atomic::AtomicBool,
418    #[cfg(test)]
419    handles: Mutex<Vec<std::thread::JoinHandle<()>>>,
420}
421
422/// Beyond this many queued or in-flight spill entries, eviction degrades to
423/// inline on the caller: bounded memory overshoot under burst beats an
424/// unbounded queue of still-resident chunks.
425const SPILL_IN_FLIGHT_MAX: usize = 64;
426
427/// What a spill thread does with a chunk once compressed.
428#[derive(Clone, Copy, PartialEq, Eq)]
429enum SpillKind {
430    /// Budget-driven: release the slot, leaving the chunk `Evicted`.
431    Evict,
432    /// Eager write-behind: keep the slot, leaving the chunk
433    /// `BackedResident`.
434    Back,
435}
436
437#[derive(Debug)]
438struct ChunkMeta {
439    pool: Arc<PoolInner>,
440    /// Length in `u64` words; immutable.
441    len: usize,
442    /// Size class for slot allocations; `None` for empty chunks and payloads
443    /// beyond the largest class. Immutable: the chunk's *slot* comes and goes
444    /// with residency, but it is always drawn from this class.
445    class: Option<usize>,
446    /// The insert-time [`ChunkHints`] depth; immutable. Names the eviction
447    /// band the chunk's queue entries belong to.
448    depth: u8,
449    /// The insert-time [`ExtentCodec`]; immutable. Encodes the chunk when it
450    /// is backed and decodes its extent on reads, so it must outlive any
451    /// extent it produced, hence `'static`.
452    codec: &'static dyn ExtentCodec,
453    state: Mutex<ChunkState>,
454}
455
456#[derive(Debug)]
457struct ChunkState {
458    residency: Residency,
459    /// Second-chance bit, set on read and cleared (in lieu of eviction) when
460    /// the budget enforcer first visits the chunk.
461    touched: bool,
462    /// Set when the owning handle is dropped, so a queue entry upgraded
463    /// concurrently with the free cannot touch a recycled slot.
464    freed: bool,
465    /// The chunk's slot index within its class's region, held exactly while
466    /// the chunk occupies pool memory (the resident states and
467    /// `WriteInFlight`). Eviction returns the slot to the region free list.
468    /// Reads copy the slot out under the state lock; no pointer into the
469    /// slot outlives the lock under which it was formed.
470    slot: Option<u32>,
471    /// The backing copy; present exactly in the `BackedResident` and
472    /// `Evicted` states.
473    extent: Option<SwapExtent>,
474    /// The payload of an `Oversize` chunk.
475    oversize: Option<Vec<u64>>,
476}
477
478impl ChunkMeta {
479    /// A fresh chunk in its insert-time state.
480    fn new(
481        pool: &Arc<PoolInner>,
482        len: usize,
483        class: Option<usize>,
484        depth: u8,
485        codec: &'static dyn ExtentCodec,
486        residency: Residency,
487        slot: Option<u32>,
488        oversize: Option<Vec<u64>>,
489    ) -> ChunkMeta {
490        ChunkMeta {
491            pool: Arc::clone(pool),
492            len,
493            class,
494            depth,
495            codec,
496            state: Mutex::new(ChunkState {
497                residency,
498                touched: false,
499                freed: false,
500                slot,
501                extent: None,
502                oversize,
503            }),
504        }
505    }
506
507    fn len_bytes(&self) -> usize {
508        self.len * std::mem::size_of::<u64>()
509    }
510
511    /// Locks the chunk's state.
512    fn state(&self) -> MutexGuard<'_, ChunkState> {
513        self.state.lock().expect("chunk state poisoned")
514    }
515}
516
517/// Handle to one immutable chunk in a [`Pool`]. Dropping the handle frees the
518/// chunk: the slot (if resident) returns to the region free list with its
519/// physical pages released, and the extent (if any) is deallocated,
520/// discarding any swapped copy for free. Releasing the pages keeps RSS
521/// aligned with the `resident_bytes` gauge the budget enforcer trusts;
522/// without it, freed slots would hold warm pages the enforcer cannot see.
523#[derive(Debug)]
524pub struct ChunkHandle {
525    meta: Arc<ChunkMeta>,
526}
527
528// Test hook fired inside `PoolInner::enforce_budget`, between a pass's final
529// counter read and the release of the `enforcing` guard. A test arms it on the
530// thread whose pass it wants to freeze, to interleave a concurrent over-budget
531// insert. One-shot: the hook is taken before it runs, so a re-enforcing pass
532// does not re-arm.
533#[cfg(test)]
534thread_local! {
535    static ENFORCE_BUDGET_HOOK: std::cell::RefCell<Option<Box<dyn FnOnce()>>> =
536        const { std::cell::RefCell::new(None) };
537}
538
539#[cfg(test)]
540fn run_enforce_budget_hook() {
541    let hook = ENFORCE_BUDGET_HOOK.with(|cell| cell.borrow_mut().take());
542    if let Some(hook) = hook {
543        hook();
544    }
545}
546
547impl Pool {
548    /// Creates a pool, reserving one virtual region per size class. The
549    /// pool starts with an unlimited budget — nothing is evicted until
550    /// [`Pool::set_budget`] tunes it.
551    pub fn new() -> std::io::Result<Pool> {
552        Pool::with_class_capacity(CLASS_CAPACITY_BYTES)
553    }
554
555    /// As [`Pool::new`], with a caller-chosen virtual reservation per size
556    /// class. Small reservations let tests exercise slot exhaustion.
557    fn with_class_capacity(class_capacity_bytes: usize) -> std::io::Result<Pool> {
558        let regions = SIZE_CLASSES
559            .iter()
560            .map(|&class_size| Region::new(class_size, class_capacity_bytes))
561            .collect::<std::io::Result<Vec<_>>>()?;
562        let extent_arena = Arc::new(ExtentArena::new(class_capacity_bytes)?);
563        Ok(Pool(Arc::new(PoolInner {
564            budget_bytes: AtomicU64::new(u64::MAX),
565            rss_target_bytes: AtomicU64::new(0),
566            regions,
567            extent_arena,
568            queues: std::array::from_fn(|_| Mutex::new(VecDeque::new())),
569            extent_queue: Mutex::new(VecDeque::new()),
570            live_chunks: AtomicU64::new(0),
571            extent_residents: AtomicU64::new(0),
572            enforcing: Mutex::new(()),
573            enforce_pending: std::sync::atomic::AtomicBool::new(false),
574            counters: Counters::default(),
575            spill: Spill::default(),
576        })))
577    }
578
579    /// Allocates a chunk of `len` words and fills it in place: `fill`
580    /// receives the chunk's slot memory directly and must overwrite all of
581    /// it (the slot's prior contents are unspecified). The returned handle
582    /// starts `UnbackedResident`. A zero `len` returns a length-0 handle
583    /// holding no slot; payloads beyond the largest size class fall back to
584    /// a plain heap allocation, always resident, a prototype limitation.
585    /// `hints` steer eviction and write-behind policy; callers without
586    /// placement knowledge pass the default.
587    ///
588    /// This is the zero-staging insert: serialization can write its single
589    /// copy straight into pool memory, paying one page population instead of
590    /// staging through caller-side buffers that fault their own pages and
591    /// die immediately after.
592    ///
593    /// `codec` is the chunk's [`ExtentCodec`], fixed for its lifetime: the
594    /// pool invokes it whenever the chunk moves across the extent boundary,
595    /// and takes no interest in the stored form it produces.
596    ///
597    /// Relies on abort-on-panic: a panic in `fill` that was caught would
598    /// leak the slot and its resident-bytes accounting. All hosting
599    /// binaries abort via `mz_ore::panic::install_enhanced_handler`, and
600    /// pool consumers are dataflow operators, never code hosted under a
601    /// `catch_unwind` boundary the way the optimizer is.
602    pub fn insert_with(
603        &self,
604        len: usize,
605        hints: ChunkHints,
606        codec: &'static dyn ExtentCodec,
607        fill: impl FnOnce(&mut [u64]),
608    ) -> ChunkHandle {
609        let inner = &self.0;
610        inner.counters.inserts.fetch_add(1, Ordering::Relaxed);
611        let len_bytes = len * std::mem::size_of::<u64>();
612        if len == 0 {
613            fill(&mut []);
614            let meta = ChunkMeta::new(
615                inner,
616                0,
617                None,
618                hints.depth,
619                codec,
620                Residency::UnbackedResident,
621                None,
622                None,
623            );
624            return ChunkHandle {
625                meta: Arc::new(meta),
626            };
627        }
628        let class = region::size_class_for(len_bytes);
629        if class.is_none() {
630            // The payload exceeds the largest size class, so it goes straight
631            // to a heap-backed oversize chunk.
632            inner
633                .counters
634                .oversize_payloads
635                .fetch_add(1, Ordering::Relaxed);
636        }
637        // A class with no free slot degrades to the heap path below: the
638        // resident set outgrew the class reservation, and an unpageable chunk
639        // beats a dead replica. Warn once; the fallback counter tracks scale.
640        let slot = class.and_then(|class| inner.alloc_slot(class, len_bytes));
641        // Whichever home the payload found, it is resident.
642        inner
643            .counters
644            .resident_bytes
645            .fetch_add(u64::cast_from(len_bytes), Ordering::Relaxed);
646        let meta = match (class, slot) {
647            (Some(class), Some(slot)) => {
648                let region = &inner.regions[class];
649                // SAFETY: the freshly allocated slot is at least `len_bytes`
650                // long (the class fits the payload) and is exclusively owned
651                // by this not-yet-shared chunk, so the mutable borrow is
652                // unique; region memory is mapped and writable, and `u64` has
653                // no validity requirements beyond size, so exposing the
654                // unspecified prior contents through `&mut [u64]` is sound.
655                let dst = unsafe {
656                    std::slice::from_raw_parts_mut(region.slot_ptr(slot).cast::<u64>(), len)
657                };
658                // The fill contract (overwrite all `len` words) is
659                // discipline-only. Poison in debug builds so an
660                // under-writing fill reads back as deterministic garbage
661                // instead of a previous occupant's bytes, which the heap
662                // path's zero fill would otherwise mask in tests.
663                #[cfg(debug_assertions)]
664                dst.fill(u64::from_ne_bytes([0xDE; 8]));
665                fill(dst);
666                ChunkMeta::new(
667                    inner,
668                    len,
669                    Some(class),
670                    hints.depth,
671                    codec,
672                    Residency::UnbackedResident,
673                    Some(slot),
674                    None,
675                )
676            }
677            _ => {
678                let mut payload = vec![0u64; len];
679                fill(&mut payload);
680                inner
681                    .counters
682                    .oversize_bytes
683                    .fetch_add(u64::cast_from(len_bytes), Ordering::Relaxed);
684                ChunkMeta::new(
685                    inner,
686                    len,
687                    None,
688                    hints.depth,
689                    codec,
690                    Residency::Oversize,
691                    None,
692                    Some(payload),
693                )
694            }
695        };
696        let meta = Arc::new(meta);
697        if meta.class.is_some() {
698            inner.live_chunks.fetch_add(1, Ordering::Relaxed);
699            inner
700                .queue(band(meta.depth))
701                .push_back(Arc::downgrade(&meta));
702        }
703        inner.enforce_budget();
704        ChunkHandle { meta }
705    }
706
707    /// Snapshot of the pool's counters.
708    pub fn stats(&self) -> PoolStats {
709        let c = &self.0.counters;
710        PoolStats {
711            inserts: c.inserts.load(Ordering::Relaxed),
712            frees: c.frees.load(Ordering::Relaxed),
713            writes_elided: c.writes_elided.load(Ordering::Relaxed),
714            evictions_compress: c.evictions_compress.load(Ordering::Relaxed),
715            evictions_cheap: c.evictions_cheap.load(Ordering::Relaxed),
716            extent_bytes_written: c.extent_bytes_written.load(Ordering::Relaxed),
717            resident_bytes: c.resident_bytes.load(Ordering::Relaxed),
718            oversize_bytes: c.oversize_bytes.load(Ordering::Relaxed),
719            warm_bytes: c.warm_bytes.load(Ordering::Relaxed),
720            warm_reuses: c.warm_reuses.load(Ordering::Relaxed),
721            eager_backs: c.eager_backs.load(Ordering::Relaxed),
722            admissions_budget: c.admissions_budget.load(Ordering::Relaxed),
723            admissions_steal: c.admissions_steal.load(Ordering::Relaxed),
724            admissions_denied: c.admissions_denied.load(Ordering::Relaxed),
725            extent_resident_bytes: c.extent_resident_bytes.load(Ordering::Relaxed),
726            extent_unreclaimable_bytes: c.extent_unreclaimable_bytes.load(Ordering::Relaxed),
727            extent_pageouts: c.extent_pageouts.load(Ordering::Relaxed),
728            extent_pageout_incomplete: c.extent_pageout_incomplete.load(Ordering::Relaxed),
729            extent_arena_fallbacks: self.0.extent_arena.fallbacks(),
730            spill_scheduled: c.spill_scheduled.load(Ordering::Relaxed),
731            spill_cancelled: c.spill_cancelled.load(Ordering::Relaxed),
732            spill_in_flight: self.0.spill.in_flight.load(Ordering::Relaxed),
733            slot_exhausted_fallbacks: c.slot_exhausted_fallbacks.load(Ordering::Relaxed),
734            oversize_payloads: c.oversize_payloads.load(Ordering::Relaxed),
735            live_chunks: self.0.live_chunks.load(Ordering::Relaxed),
736        }
737    }
738
739    /// Enables or disables off-worker eviction I/O. The first call with
740    /// `threads > 0` spawns that many spill threads (spawn-once: later calls
741    /// only toggle participation); `threads == 0` falls back to inline
742    /// eviction on the caller for subsequent victims, letting any queued
743    /// work drain.
744    pub fn set_spill_threads(&self, threads: usize) {
745        if threads == 0 {
746            self.0.spill.enabled.store(false, Ordering::Relaxed);
747            return;
748        }
749        let spawned = self.0.spill.threads.load(Ordering::Relaxed);
750        if spawned == 0 {
751            let to_spawn = u64::cast_from(threads);
752            if self
753                .0
754                .spill
755                .threads
756                .compare_exchange(0, to_spawn, Ordering::Relaxed, Ordering::Relaxed)
757                .is_ok()
758            {
759                for i in 0..threads {
760                    let inner = Arc::clone(&self.0);
761                    let handle = std::thread::Builder::new()
762                        .name(format!("pool-spill-{i}"))
763                        .spawn(move || inner.spill_worker())
764                        .expect("spawn pool spill thread");
765                    #[cfg(test)]
766                    self.0
767                        .spill
768                        .handles
769                        .lock()
770                        .expect("spill handles poisoned")
771                        .push(handle);
772                    #[cfg(not(test))]
773                    drop(handle);
774                }
775            }
776        }
777        self.0.spill.enabled.store(true, Ordering::Relaxed);
778    }
779
780    /// Enables or disables eager backing: when on, idle spill threads
781    /// compress unbacked chunks to `BackedResident` ahead of pressure, so
782    /// budget-driven eviction becomes a pure page release. Only meaningful
783    /// with spill threads spawned.
784    pub fn set_eager_backing(&self, eager: bool) {
785        self.0.spill.eager.store(eager, Ordering::Relaxed);
786        if eager {
787            self.0.spill.cv.notify_all();
788        }
789    }
790
791    /// Test hook: performs one eager-backing step on the calling thread.
792    /// Returns whether progress was made.
793    #[cfg(test)]
794    fn back_step(&self) -> bool {
795        self.0.back_one()
796    }
797
798    /// Test hook: waits until the spill queue is empty and no entry is being
799    /// processed, so tests observe deterministic post-eviction states.
800    #[cfg(test)]
801    fn quiesce_spill(&self) {
802        while self.0.spill.in_flight.load(Ordering::Relaxed) > 0 {
803            std::thread::yield_now();
804        }
805    }
806
807    /// Test hook: stops and joins the spill threads, so a test binary exits
808    /// with none alive (which Miri requires). Stopped threads process no
809    /// further queued work; call [`Pool::quiesce_spill`] first when the test
810    /// depends on the queue draining.
811    #[cfg(test)]
812    fn join_spill_threads(&self) {
813        self.0.spill.stop.store(true, Ordering::Relaxed);
814        self.0.spill.cv.notify_all();
815        let handles =
816            std::mem::take(&mut *self.0.spill.handles.lock().expect("spill handles poisoned"));
817        for handle in handles {
818            handle.join().expect("spill thread panicked");
819        }
820    }
821
822    /// Test hook: enables spill scheduling without spawning threads, so tests
823    /// drive the queue deterministically via [`Pool::spill_step`].
824    #[cfg(test)]
825    fn enable_spill_without_threads(&self) {
826        self.0.spill.enabled.store(true, Ordering::Relaxed);
827    }
828
829    /// Test hook: processes one queued spill entry on the calling thread.
830    /// Returns whether an entry was processed.
831    #[cfg(test)]
832    fn spill_step(&self) -> bool {
833        let popped = self.0.spill_queue().pop_front();
834        let Some(meta) = popped else {
835            return false;
836        };
837        self.0.spill_process(&meta, SpillKind::Evict);
838        self.0.spill.in_flight.fetch_sub(1, Ordering::Relaxed);
839        true
840    }
841
842    /// Test hook: runs one compressed-cap enforcement pass on the calling
843    /// thread.
844    #[cfg(test)]
845    fn enforce_compressed(&self) {
846        self.0.enforce_compressed_cap();
847    }
848
849    /// Test hook: evicts cold chunks until resident bytes fall to the budget
850    /// or every queued chunk has been visited once. Enforcement runs
851    /// automatically on every insert and budget shrink.
852    #[cfg(test)]
853    fn enforce_budget(&self) {
854        self.0.enforce_budget();
855    }
856
857    /// Test hook: runs one compressed-cap enforcement pass inline on the
858    /// calling thread, where the fake residency observation applies.
859    #[cfg(test)]
860    fn enforce_rss_target(&self) {
861        self.0.enforce_compressed_cap();
862    }
863
864    /// Retunes the resident-bytes budget in place and enforces it. Live
865    /// handles share the new value immediately through their `Arc<PoolInner>`;
866    /// a shrink takes effect by evicting on this call, a grow simply leaves
867    /// more headroom for future inserts.
868    pub fn set_budget(&self, budget_bytes: usize) {
869        let new = u64::cast_from(budget_bytes);
870        let prev = self.0.budget_bytes.swap(new, Ordering::Relaxed);
871        // Config application calls this per worker per tick; only a change
872        // warrants an enforcement pass (a grow needs none, and inserts
873        // enforce continuously anyway).
874        if new < prev {
875            self.0.trim_warm_pool();
876            self.0.enforce_budget();
877        }
878    }
879
880    /// Retunes the ceiling on the pool's total RSS — slots plus warm slots
881    /// plus compressed-resident extents. The compressed tier's capacity is
882    /// the gap above the budget and warm cap; zero (the default) collapses
883    /// the tier, paging extents out as soon as they are written. A shrink
884    /// takes effect by paging out the oldest extents on this call.
885    pub fn set_rss_target(&self, target_bytes: usize) {
886        let new = u64::cast_from(target_bytes);
887        let prev = self.0.rss_target_bytes.swap(new, Ordering::Relaxed);
888        if new < prev {
889            self.0.enforce_compressed_cap();
890        }
891    }
892
893    /// Test-only: the number of entries across the second-chance queues,
894    /// live and stale.
895    #[cfg(test)]
896    fn queue_len(&self) -> usize {
897        (0..DEPTH_BANDS).map(|band| self.0.queue(band).len()).sum()
898    }
899
900    /// Test-only: the number of resident-extent queue entries, live and
901    /// stale.
902    #[cfg(test)]
903    fn extent_queue_len(&self) -> usize {
904        self.0.extent_queue().len()
905    }
906
907    /// Test hook: explicitly evicts one chunk. No-op if the chunk is already
908    /// evicted, in flight, empty, or oversize. With spill threads enabled the
909    /// compression is handed off and completes asynchronously (observable via
910    /// [`Residency::WriteInFlight`]); without them it runs inline.
911    #[cfg(test)]
912    fn evict(&self, handle: &ChunkHandle) {
913        let meta = &handle.meta;
914        let mut state = meta.state();
915        if !meta.pool.spill_handoff(meta, &mut state) {
916            meta.pool.evict_locked(meta, &mut state);
917        }
918        drop(state);
919        meta.pool.enforce_or_defer_compressed_cap();
920    }
921
922    /// Test hook: overwrites every free slot's bytes with `0xDE`. The free
923    /// list keeps a freed slot's old bytes on platforms where
924    /// `MADV_DONTNEED` retains contents (macOS); poisoning lets tests prove
925    /// that reads of evicted chunks decompress from the extent rather than
926    /// passing stale slot memory through.
927    #[cfg(test)]
928    fn poison_free_slots(&self) {
929        for region in &self.0.regions {
930            region.poison_free_slots();
931        }
932    }
933}
934
935impl PoolInner {
936    /// Locks the eviction queue of one depth band.
937    fn queue(&self, band: usize) -> MutexGuard<'_, VecDeque<Weak<ChunkMeta>>> {
938        self.queues[band].lock().expect("pool queue poisoned")
939    }
940
941    /// Locks the resident-extent queue.
942    fn extent_queue(&self) -> MutexGuard<'_, VecDeque<Weak<ChunkMeta>>> {
943        self.extent_queue.lock().expect("extent queue poisoned")
944    }
945
946    /// Locks the spill hand-off queue.
947    fn spill_queue(&self) -> MutexGuard<'_, VecDeque<Arc<ChunkMeta>>> {
948        self.spill.queue.lock().expect("spill queue poisoned")
949    }
950
951    /// The region behind a slotted chunk's size class.
952    fn region_of(&self, meta: &ChunkMeta) -> &Region {
953        &self.regions[meta.class.expect("slotted chunk has a class")]
954    }
955
956    /// Borrows the payload of a slotted chunk.
957    ///
958    /// # Safety
959    ///
960    /// `slot` must be `meta`'s slot, its contents must be initialized (they
961    /// are from insert onward), and nothing may write the slot while the
962    /// borrow lives.
963    unsafe fn slot_data(&self, meta: &ChunkMeta, slot: u32) -> &[u64] {
964        let ptr = self
965            .region_of(meta)
966            .slot_ptr(slot)
967            .cast_const()
968            .cast::<u64>();
969        // SAFETY: per the function contract; `meta.len` words fit the class
970        // by construction.
971        unsafe { std::slice::from_raw_parts(ptr, meta.len) }
972    }
973
974    /// Records a freshly written extent under the chunk's state lock: the
975    /// compressed-bytes counter, the compressed-tier accounting, and the
976    /// state's extent field.
977    fn commit_extent(&self, meta: &Arc<ChunkMeta>, state: &mut ChunkState, extent: SwapExtent) {
978        self.counters
979            .extent_bytes_written
980            .fetch_add(u64::cast_from(extent.comp_len()), Ordering::Relaxed);
981        // A heap-fallback extent is born permanently capped and counts as
982        // unreclaimable from the start.
983        self.note_extent_resident(meta, extent.alloc_size(), !extent.pageout_capped());
984        state.extent = Some(extent);
985    }
986
987    /// Drops queue entries whose chunk has been freed, detected by their
988    /// `Weak` no longer holding a live chunk. Each band compacts only when
989    /// its stale entries outnumber all live chunks (plus a small floor), so
990    /// the cost amortizes to a constant per insert and the total queue
991    /// length stays proportional to the number of live slotted chunks even
992    /// when the pool never comes under budget pressure.
993    fn prune_queues(&self) {
994        let live = usize::cast_from(self.live_chunks.load(Ordering::Relaxed));
995        for band in 0..DEPTH_BANDS {
996            let mut queue = self.queue(band);
997            if queue.len() > 2 * live + 16 {
998                queue.retain(|weak| weak.strong_count() > 0);
999            }
1000        }
1001    }
1002
1003    fn enforce_budget(&self) {
1004        // Single-flight: enforcement runs synchronously on whichever thread
1005        // trips it (every insert), and concurrent passes would
1006        // convoy on the queue mutex doing redundant scans of the same
1007        // candidates. One pass at a time reaches the budget just as well;
1008        // skipped callers hand their bytes to the in-progress pass through
1009        // `enforce_pending`. A poisoned claim means a prior pass panicked.
1010        // Recover and keep enforcing rather than silently disabling the
1011        // budget for the process's lifetime.
1012        let guard = match self.enforcing.try_lock() {
1013            Ok(guard) => guard,
1014            Err(std::sync::TryLockError::WouldBlock) => {
1015                // Release pairs with the holder's Acquire: the `resident_bytes`
1016                // bump this caller just made must be visible to the re-read.
1017                self.enforce_pending.store(true, Ordering::Release);
1018                return;
1019            }
1020            Err(std::sync::TryLockError::Poisoned(poisoned)) => poisoned.into_inner(),
1021        };
1022        loop {
1023            self.enforce_budget_inner();
1024            #[cfg(test)]
1025            run_enforce_budget_hook();
1026            // A caller turned away since this pass's counter reads may have
1027            // left bytes unenforced. Re-run rather than drop them. The Acquire
1028            // pairs with the turned-away Release so the re-read sees the bump.
1029            //
1030            // This swap is the only place the flag is cleared, so no set can be
1031            // lost. The cost is that a flag left over from an earlier call's
1032            // residual window buys one extra pass here, which is harmless.
1033            //
1034            // NOTE: a caller turned away between this swap and `drop(guard)`
1035            // sets the flag but finds no re-reader. That residual window is a
1036            // few instructions wide, versus the whole pass before.
1037            if !self.enforce_pending.swap(false, Ordering::Acquire) {
1038                break;
1039            }
1040        }
1041        drop(guard);
1042        // Inline evictions above may have grown the compressed tier.
1043        self.enforce_or_defer_compressed_cap();
1044    }
1045
1046    /// Bytes budget enforcement can actually reclaim: resident bytes minus
1047    /// heap-backed (oversize and class-exhaustion) chunks, which hold no
1048    /// slot and can never be evicted. Enforcing against raw resident bytes
1049    /// would, once unevictable bytes alone exceed the budget, compress
1050    /// every slotted chunk on arrival forever.
1051    fn evictable_bytes(&self) -> u64 {
1052        self.counters
1053            .resident_bytes
1054            .load(Ordering::Relaxed)
1055            .saturating_sub(self.counters.oversize_bytes.load(Ordering::Relaxed))
1056    }
1057
1058    fn enforce_budget_inner(&self) {
1059        self.prune_queues();
1060        // Deepest band first: deep chunks are the coldest, and once eager
1061        // backing has visited them (same order) their eviction is a pure
1062        // page release. The youngest band is reached only when the deeper
1063        // bands cannot satisfy the budget, keeping young data's
1064        // die-before-write chance longest.
1065        for band in (0..DEPTH_BANDS).rev() {
1066            if self.evictable_bytes() <= self.budget_bytes.load(Ordering::Relaxed) {
1067                return;
1068            }
1069            self.enforce_budget_band(band);
1070        }
1071    }
1072
1073    fn enforce_budget_band(&self, band: usize) {
1074        // The queue holds resident chunks only (entries for evicted chunks
1075        // are dropped on visit and never re-added), so a full pass is
1076        // proportional to the resident set. Visit each queued chunk at most
1077        // twice per call: a first visit may only clear the second-chance
1078        // bit, so a second is needed before an over-budget call is
1079        // guaranteed to evict every chunk it saw. The bound keeps contended
1080        // and in-flight entries from spinning this loop forever.
1081        let mut remaining = self.queue(band).len().saturating_mul(2);
1082        while remaining > 0 && self.evictable_bytes() > self.budget_bytes.load(Ordering::Relaxed) {
1083            remaining -= 1;
1084            let popped = self.queue(band).pop_front();
1085            let Some(weak) = popped else {
1086                break;
1087            };
1088            let Some(meta) = weak.upgrade() else {
1089                continue;
1090            };
1091            let requeue = {
1092                // `try_lock`: a chunk mid-eviction or mid-read holds its
1093                // lock for milliseconds; skipping it beats convoying every
1094                // budget enforcer in the process behind one chunk's I/O.
1095                let Ok(mut state) = meta.state.try_lock() else {
1096                    self.queue(band).push_back(weak);
1097                    continue;
1098                };
1099                if state.freed {
1100                    false
1101                } else if matches!(state.residency, Residency::Evicted | Residency::Oversize) {
1102                    // Nothing to evict: drop the entry. A chunk re-enters
1103                    // the queue only when it becomes resident again (insert
1104                    // or re-admission), so the queue stays proportional to
1105                    // the resident set rather than accumulating every chunk
1106                    // ever evicted.
1107                    false
1108                } else if state.touched {
1109                    state.touched = false;
1110                    true
1111                } else if self.spill_handoff(&meta, &mut state) {
1112                    // Stays queued while in flight; once the spill commits to
1113                    // `Evicted`, the next visit drops the entry.
1114                    true
1115                } else {
1116                    self.evict_locked(&meta, &mut state);
1117                    state.residency != Residency::Evicted
1118                }
1119            };
1120            if requeue {
1121                self.queue(band).push_back(weak);
1122            }
1123        }
1124    }
1125
1126    fn evict_locked(&self, meta: &Arc<ChunkMeta>, state: &mut ChunkState) {
1127        let Some(slot) = state.slot else {
1128            return;
1129        };
1130        if state.freed {
1131            return;
1132        }
1133        match state.residency {
1134            Residency::UnbackedResident => {
1135                // SAFETY: the slot belongs to this live chunk and the state
1136                // lock is held, so nothing else touches the slot while this
1137                // borrow is live (reads copy out under the same lock).
1138                let data = unsafe { self.slot_data(meta, slot) };
1139                // Inline eviction runs on whichever thread tripped the
1140                // budget, so the compression scratch must not stay parked
1141                // on it.
1142                let extent =
1143                    SwapExtent::write(&self.extent_arena, data, meta.codec, Scratch::Shrink);
1144                self.counters
1145                    .evictions_compress
1146                    .fetch_add(1, Ordering::Relaxed);
1147                self.commit_extent(meta, state, extent);
1148            }
1149            Residency::BackedResident => {
1150                self.counters
1151                    .evictions_cheap
1152                    .fetch_add(1, Ordering::Relaxed);
1153            }
1154            Residency::WriteInFlight | Residency::Evicted | Residency::Oversize => return,
1155        }
1156        // `release_slot`'s precondition holds: the state lock is held and
1157        // `!freed` was checked above under it.
1158        self.release_slot(meta, state);
1159        state.residency = Residency::Evicted;
1160    }
1161
1162    /// Whether the next eviction should be handed to spill threads: enabled,
1163    /// and the queue is below the backpressure bound (beyond it, callers
1164    /// evict inline rather than growing an unbounded queue of still-resident
1165    /// chunks).
1166    fn spill_eligible(&self) -> bool {
1167        self.spill.enabled.load(Ordering::Relaxed)
1168            && usize::cast_from(self.spill.in_flight.load(Ordering::Relaxed)) < SPILL_IN_FLIGHT_MAX
1169    }
1170
1171    /// Hands a `WriteInFlight` chunk to the spill threads.
1172    fn spill_schedule(&self, meta: Arc<ChunkMeta>) {
1173        self.counters
1174            .spill_scheduled
1175            .fetch_add(1, Ordering::Relaxed);
1176        self.spill.in_flight.fetch_add(1, Ordering::Relaxed);
1177        self.spill_queue().push_back(meta);
1178        self.spill.cv.notify_one();
1179    }
1180
1181    /// Spill-thread main loop. The thread owns an `Arc<PoolInner>`, so the
1182    /// pool (a process-wide singleton in production) lives as long as its
1183    /// threads. Queued (budget-driven) evictions take priority; with eager
1184    /// backing enabled, idle threads compress unbacked chunks to
1185    /// `BackedResident` instead of parking, and park with a timeout once
1186    /// everything reachable is backed.
1187    fn spill_worker(self: Arc<Self>) {
1188        loop {
1189            #[cfg(test)]
1190            if self.spill.stop.load(Ordering::Relaxed) {
1191                return;
1192            }
1193            // Tier-2 pageouts ride the spill threads: every pass through the
1194            // loop (job completion, condvar wakeup, park timeout) trims the
1195            // compressed tier if needed. A single atomic load when under cap.
1196            self.enforce_compressed_cap();
1197            let popped = self.spill_queue().pop_front();
1198            if let Some(meta) = popped {
1199                self.spill_process(&meta, SpillKind::Evict);
1200                self.spill.in_flight.fetch_sub(1, Ordering::Relaxed);
1201                continue;
1202            }
1203            if self.spill.eager.load(Ordering::Relaxed) && self.back_one() {
1204                continue;
1205            }
1206            // Nothing to evict or back: park. Re-checking emptiness under
1207            // the queue lock closes the lost-wakeup window (hand-offs push
1208            // under this lock before notifying); the timeout backstops
1209            // everything else (fresh inserts, tier growth, lost notifies).
1210            let queue = self.spill_queue();
1211            if queue.is_empty() {
1212                let _ = self
1213                    .spill
1214                    .cv
1215                    .wait_timeout(queue, std::time::Duration::from_millis(100))
1216                    .expect("spill queue poisoned");
1217            }
1218        }
1219    }
1220
1221    /// Eagerly compresses one unbacked chunk from the eviction queues into
1222    /// `BackedResident`, returning whether a chunk was backed — `false`
1223    /// means nothing was actionable (queues empty, or the bounded scans
1224    /// found only already-backed, in-flight, contended, or stale entries)
1225    /// and the caller should park rather than rescan. Bands are visited
1226    /// deepest first, mirroring eviction order so the chunks evicted first
1227    /// are the ones whose backing is already pre-paid.
1228    fn back_one(&self) -> bool {
1229        for band in (0..DEPTH_BANDS).rev() {
1230            if self.back_one_from(band) {
1231                return true;
1232            }
1233        }
1234        false
1235    }
1236
1237    /// One bounded backing scan over a single band's queue. Non-actionable
1238    /// entries are requeued or dropped per the same rules budget
1239    /// enforcement uses, except that the second-chance `touched` bit is
1240    /// left alone — backing is not an eviction and must not consume a
1241    /// chunk's reprieve.
1242    fn back_one_from(&self, band: usize) -> bool {
1243        for _ in 0..16 {
1244            let popped = self.queue(band).pop_front();
1245            let Some(weak) = popped else {
1246                return false;
1247            };
1248            let Some(meta) = weak.upgrade() else {
1249                continue;
1250            };
1251            {
1252                let Ok(mut state) = meta.state.try_lock() else {
1253                    self.queue(band).push_back(weak);
1254                    continue;
1255                };
1256                if state.freed {
1257                    continue;
1258                }
1259                match state.residency {
1260                    Residency::Evicted | Residency::Oversize => {
1261                        continue;
1262                    }
1263                    Residency::UnbackedResident => {
1264                        state.residency = Residency::WriteInFlight;
1265                    }
1266                    Residency::BackedResident | Residency::WriteInFlight => {
1267                        self.queue(band).push_back(weak);
1268                        continue;
1269                    }
1270                }
1271            }
1272            self.spill.in_flight.fetch_add(1, Ordering::Relaxed);
1273            self.spill_process(&meta, SpillKind::Back);
1274            self.spill.in_flight.fetch_sub(1, Ordering::Relaxed);
1275            // The chunk remains an eviction candidate (now a cheap one).
1276            self.queue(band).push_back(weak);
1277            return true;
1278        }
1279        false
1280    }
1281
1282    /// Performs (or cancels) one scheduled compression. Lock discipline: the
1283    /// chunk lock is held only to validate and to commit — never across the
1284    /// compression or the `pageout` reclaim, which are the multi-millisecond
1285    /// costs this path exists to keep off budget-enforcing threads.
1286    fn spill_process(&self, meta: &Arc<ChunkMeta>, kind: SpillKind) {
1287        // Validate under the lock, then release it for the I/O. The slot is
1288        // captured under the lock and remains owned by this chunk for the
1289        // unlocked compression: in `WriteInFlight`, eviction skips the chunk
1290        // and `ChunkHandle::drop` defers slot release to this thread.
1291        let slot;
1292        {
1293            let mut state = meta.state();
1294            if state.freed {
1295                // Freed while queued: the deferred cleanup is ours, and the
1296                // chunk dies without ever compressing — the write-behind
1297                // cancellation window. `ChunkHandle::drop` already counted
1298                // the free and the live-chunks decrement.
1299                self.counters
1300                    .spill_cancelled
1301                    .fetch_add(1, Ordering::Relaxed);
1302                self.counters.writes_elided.fetch_add(1, Ordering::Relaxed);
1303                self.release_slot(meta, &mut state);
1304                return;
1305            }
1306            if state.residency != Residency::WriteInFlight {
1307                return;
1308            }
1309            slot = state.slot.expect("write-in-flight chunk has a slot");
1310        }
1311        // SAFETY: the chunk is live (the queue holds an `Arc`) and in
1312        // `WriteInFlight`, so the slot is not recycled (`ChunkHandle::drop`
1313        // defers slot release to this thread in that state) and its contents
1314        // are immutable; concurrent copy-out reads take the state lock and
1315        // read the slot, but nothing writes it.
1316        let data = unsafe { self.slot_data(meta, slot) };
1317        // Spill threads see a steady job stream, so they keep the grown
1318        // compression scratch for the next job.
1319        let extent = SwapExtent::write(&self.extent_arena, data, meta.codec, Scratch::Retain);
1320        // Commit under the lock.
1321        let mut state = meta.state();
1322        if state.freed {
1323            // Freed during compression: the extent is garbage; cleanup is
1324            // ours as above. Compression ran, so this is not an elided free.
1325            self.counters
1326                .spill_cancelled
1327                .fetch_add(1, Ordering::Relaxed);
1328            self.release_slot(meta, &mut state);
1329            return;
1330        }
1331        self.commit_extent(meta, &mut state, extent);
1332        match kind {
1333            SpillKind::Back => {
1334                // The slot stays for write-behind: the chunk remains
1335                // readable, and the extent makes a later budget eviction a
1336                // pure page release.
1337                self.counters.eager_backs.fetch_add(1, Ordering::Relaxed);
1338                state.residency = Residency::BackedResident;
1339            }
1340            SpillKind::Evict => {
1341                // `release_slot`'s precondition holds: the state lock is
1342                // held and `!freed` was observed under it.
1343                self.counters
1344                    .evictions_compress
1345                    .fetch_add(1, Ordering::Relaxed);
1346                self.release_slot(meta, &mut state);
1347                state.residency = Residency::Evicted;
1348            }
1349        };
1350        drop(state);
1351        // Counted a fresh resident extent: the tier may need trimming. Kept
1352        // here (rather than relying on the spill loop alone) so the
1353        // threadless test hooks observe deterministic post-commit states.
1354        self.enforce_compressed_cap();
1355    }
1356
1357    /// Releases `state`'s slot — slot returned to the region free list,
1358    /// physical pages discarded unless the slot joins the bounded warm pool —
1359    /// and decrements resident bytes. Releasing pages beyond the warm pool is
1360    /// what keeps RSS aligned with the `resident_bytes` gauge the budget
1361    /// enforcer trusts; the warm pool relaxes that alignment by an explicit,
1362    /// bounded amount (`warm_bytes`, capped at a fraction of the budget) so
1363    /// slot reuse faults no pages and skips the kernel's page zeroing.
1364    ///
1365    /// Precondition: the caller holds the chunk's state lock, and no
1366    /// reference into the slot exists — copy-out reads borrow the slot only
1367    /// under that same lock, and a `WriteInFlight` chunk's unlocked
1368    /// compression read belongs to the spill thread, which is the only
1369    /// caller that releases the slot in that state. This is what makes the
1370    /// `dontneed` below sound, and what makes keeping a warm slot's stale
1371    /// contents safe: the slot's next occupant fully overwrites every byte
1372    /// it reads, satisfying the contents-undefined contract either way.
1373    fn release_slot(&self, meta: &ChunkMeta, state: &mut ChunkState) {
1374        let slot = state.slot.take().expect("slotted chunk");
1375        let region = self.region_of(meta);
1376        let warm = self.try_keep_warm(region.class_size());
1377        if !warm {
1378            // SAFETY: no reference into the slot exists (the function-level
1379            // precondition, established under the held state lock).
1380            unsafe {
1381                region::dontneed(region.slot_ptr(slot), region.class_size());
1382            }
1383        }
1384        region.free(slot, warm);
1385        self.counters
1386            .resident_bytes
1387            .fetch_sub(u64::cast_from(meta.len_bytes()), Ordering::Relaxed);
1388    }
1389
1390    /// The warm pool's byte ceiling: an eighth of the budget, clamped at an
1391    /// absolute maximum. The fraction sizes fault amortization at small
1392    /// budgets; the clamp keeps large budgets from parking gigabytes of idle
1393    /// warm slots no fault rate could justify.
1394    fn warm_cap(&self) -> u64 {
1395        (self.budget_bytes.load(Ordering::Relaxed) / 8).min(1 << 30)
1396    }
1397
1398    /// Cools warm free slots until `warm_bytes` falls to the warm cap. A
1399    /// budget shrink lowers the cap, and warm capacity is checked only when
1400    /// a slot is freed, so without this pass slots parked under the old cap
1401    /// would hold their pages until same-class reuse happened to drain them,
1402    /// exactly when the shrink wanted the memory back.
1403    fn trim_warm_pool(&self) {
1404        let mut over = self
1405            .counters
1406            .warm_bytes
1407            .load(Ordering::Relaxed)
1408            .saturating_sub(self.warm_cap());
1409        for region in &self.regions {
1410            if over == 0 {
1411                return;
1412            }
1413            let cooled = u64::cast_from(region.cool_warm_slots(usize::cast_from(over)));
1414            self.counters
1415                .warm_bytes
1416                .fetch_sub(cooled, Ordering::Relaxed);
1417            over = over.saturating_sub(cooled);
1418        }
1419    }
1420
1421    /// Claims warm-pool capacity for a slot of `class_size` bytes, returning
1422    /// whether the slot may keep its pages. The RSS overshoot the warm pool
1423    /// introduces is bounded by [`PoolInner::warm_cap`] and visible as the
1424    /// `warm_bytes` stat.
1425    fn try_keep_warm(&self, class_size: usize) -> bool {
1426        let cap = self.warm_cap();
1427        let class_bytes = u64::cast_from(class_size);
1428        self.counters
1429            .warm_bytes
1430            .try_update(Ordering::Relaxed, Ordering::Relaxed, |cur| {
1431                (cur + class_bytes <= cap).then_some(cur + class_bytes)
1432            })
1433            .is_ok()
1434    }
1435
1436    /// Allocates a slot in `class` for a payload of `len_bytes` with
1437    /// warm-pool accounting (a warm allocation is counted as a reuse and
1438    /// trimmed to the payload), or `None` when the class has no free slot.
1439    fn try_alloc_slot(&self, class: usize, len_bytes: usize) -> Option<u32> {
1440        let (index, warm) = self.regions[class].alloc()?;
1441        if warm {
1442            // The allocation faulted no pages; its bytes leave the warm
1443            // pool.
1444            let class_bytes = u64::cast_from(self.regions[class].class_size());
1445            self.counters
1446                .warm_bytes
1447                .fetch_sub(class_bytes, Ordering::Relaxed);
1448            self.counters.warm_reuses.fetch_add(1, Ordering::Relaxed);
1449            // A warm slot keeps the prior occupant's resident pages, which
1450            // may extend past the new payload while the ledger credits only
1451            // `len_bytes`.
1452            self.trim_slot_tail(class, index, len_bytes);
1453        }
1454        Some(index)
1455    }
1456
1457    /// Releases a slot's pages beyond the first `len_bytes` (rounded up to
1458    /// a page), so a slot reused for a smaller payload does not keep its
1459    /// prior occupant's tail pages resident with no bytes in the ledger to
1460    /// answer for them.
1461    ///
1462    /// Precondition: the caller exclusively owns the slot (freshly
1463    /// allocated, or taken from a victim under the victim's state lock)
1464    /// with no reference into it.
1465    fn trim_slot_tail(&self, class: usize, slot: u32, len_bytes: usize) {
1466        let region = &self.regions[class];
1467        // Hugepage-class slots trim at huge-page granularity: a base-page
1468        // trim would split the slot's `MADV_HUGEPAGE` folios, and khugepaged
1469        // may later re-collapse a partially trimmed range, re-instantiating
1470        // pages the ledger counts as released. Whole-folio trims leave no
1471        // partial folio to split or resurrect.
1472        let granule = if region.class_size() >= region::HUGE_PAGE {
1473            region::HUGE_PAGE
1474        } else {
1475            region::page_size()
1476        };
1477        let keep = len_bytes.next_multiple_of(granule).min(region.class_size());
1478        let tail = region.class_size() - keep;
1479        if tail == 0 {
1480            return;
1481        }
1482        // SAFETY: the caller exclusively owns the slot per the
1483        // precondition, and `keep + tail` is exactly the class size, so the
1484        // range stays within the slot.
1485        unsafe {
1486            region::dontneed(region.slot_ptr(slot).add(keep), tail);
1487        }
1488    }
1489
1490    /// Allocates a slot in `class` for an insert: as
1491    /// [`PoolInner::try_alloc_slot`], with an exhausted class counted as a
1492    /// heap fallback for a `len_bytes` payload (warned about once). `None`
1493    /// means the caller must degrade to the heap.
1494    fn alloc_slot(&self, class: usize, len_bytes: usize) -> Option<u32> {
1495        match self.try_alloc_slot(class, len_bytes) {
1496            Some(index) => Some(index),
1497            None => {
1498                self.counters
1499                    .slot_exhausted_fallbacks
1500                    .fetch_add(1, Ordering::Relaxed);
1501                static EXHAUSTED_ONCE: std::sync::Once = std::sync::Once::new();
1502                EXHAUSTED_ONCE.call_once(|| {
1503                    tracing::warn!(
1504                        len_bytes,
1505                        "buffer pool size class exhausted; falling back to heap chunks \
1506                         (raise the pool's per-class virtual reservation)",
1507                    );
1508                });
1509                None
1510            }
1511        }
1512    }
1513
1514    /// Acquires a slot for re-admitting an evicted chunk, from free budget
1515    /// headroom or by stealing a clean backed victim's slot, never by
1516    /// evicting or compressing anything. `None` counts a denied admission.
1517    /// On success the admitted chunk's resident-bytes accounting and the
1518    /// admission counter are settled, and the caller (who holds the chunk's
1519    /// state lock) owns the slot: its contents are unspecified (fresh,
1520    /// warm, or the victim's stale bytes) and must be fully overwritten.
1521    fn admit_slot(&self, meta: &ChunkMeta) -> Option<u32> {
1522        let class = meta.class.expect("evicted chunk has a class");
1523        let len_bytes = u64::cast_from(meta.len_bytes());
1524        // Free budget first: reserve the bytes, then a slot. The
1525        // reservation never pushes resident bytes past the budget, and a
1526        // class with no free slot hands the reservation back rather than
1527        // evicting anything to make room. The headroom test uses evictable
1528        // bytes, matching budget enforcement: unevictable heap-backed bytes
1529        // must not permanently veto budget-path admissions the enforcer
1530        // would never need to undo.
1531        let reserved = self
1532            .counters
1533            .resident_bytes
1534            .try_update(Ordering::Relaxed, Ordering::Relaxed, |cur| {
1535                // Loaded inside the closure so a CAS retry sees oversize
1536                // frees that landed since the last attempt.
1537                let oversize = self.counters.oversize_bytes.load(Ordering::Relaxed);
1538                let next = cur.checked_add(len_bytes)?;
1539                (next.saturating_sub(oversize) <= self.budget_bytes.load(Ordering::Relaxed))
1540                    .then_some(next)
1541            })
1542            .is_ok();
1543        if reserved {
1544            if let Some(slot) = self.try_alloc_slot(class, meta.len_bytes()) {
1545                self.counters
1546                    .admissions_budget
1547                    .fetch_add(1, Ordering::Relaxed);
1548                return Some(slot);
1549            }
1550            self.counters
1551                .resident_bytes
1552                .fetch_sub(len_bytes, Ordering::Relaxed);
1553        }
1554        if let Some(slot) = self.steal_clean_victim(class, meta.len_bytes()) {
1555            // The ledger was settled inside the steal: the victim's bytes
1556            // out, the admitted chunk's in, with any growth reserved
1557            // against the budget. The slot's physical pages transfer
1558            // deliberately, but only up to the admitted payload: the
1559            // victim's pages past it would stay resident with no ledger
1560            // bytes to answer for them.
1561            self.trim_slot_tail(class, slot, meta.len_bytes());
1562            self.counters
1563                .admissions_steal
1564                .fetch_add(1, Ordering::Relaxed);
1565            return Some(slot);
1566        }
1567        self.counters
1568            .admissions_denied
1569            .fetch_add(1, Ordering::Relaxed);
1570        None
1571    }
1572
1573    /// Takes the slot of a clean victim in `class` for an admitted payload
1574    /// of `admitted_len_bytes`: a `BackedResident` chunk with a clear
1575    /// touched bit, whose extent already duplicates its slot, so the victim
1576    /// transitions to `Evicted` with zero I/O, its extent intact, and its
1577    /// queue entry dropped. The returned slot keeps its physical pages (no
1578    /// `dontneed`, no free-list round trip). They hold the victim's stale
1579    /// bytes. The ledger is settled inside the steal: the victim's bytes
1580    /// leave and the admitted payload's enter in one step, and a steal that
1581    /// grows resident bytes must fit the budget like any other admission
1582    /// (a shrinking steal always may proceed). `None` when the bounded scan
1583    /// finds no such victim, or none whose growth the budget can absorb.
1584    ///
1585    /// The caller holds its own chunk's state lock. The scan follows the
1586    /// enforcement discipline (deepest band first, queue guard dropped
1587    /// before any chunk lock, victims only ever `try_lock`ed), which is
1588    /// what keeps the chunk-lock-while-probing-chunk-lock window
1589    /// deadlock-free: two admitters stealing toward each other both fail
1590    /// the `try_lock` and skip. Unlike enforcement, the scan rotates
1591    /// unsuitable entries (touched, wrong class, unbacked) to the back
1592    /// without spending touched bits, shuffling FIFO order the way the
1593    /// backing scan does.
1594    fn steal_clean_victim(&self, class: usize, admitted_len_bytes: usize) -> Option<u32> {
1595        // Bound on entries examined per band before moving to the next.
1596        // The budget is per band, not shared across the scan: a deep band
1597        // densely populated with touched resident chunks (a probe-heavy
1598        // arrangement whose reads refresh every bit) would otherwise spend
1599        // the entire scan on hopeless candidates and starve the shallower
1600        // bands where the clean-victim stock actually sits (eager backing
1601        // stocks young backed chunks in band zero). Eight visits per band
1602        // absorb a handful of lock-busy or freshly touched entries without
1603        // degrading a hopeless scan into a full queue walk, and cap the
1604        // whole scan at eight times the band count.
1605        const VISITS_PER_BAND: usize = 8;
1606        for band in (0..DEPTH_BANDS).rev() {
1607            let mut visits = VISITS_PER_BAND;
1608            while visits > 0 {
1609                let popped = self.queue(band).pop_front();
1610                let Some(weak) = popped else {
1611                    // Band exhausted; the next band has its own budget.
1612                    break;
1613                };
1614                let Some(meta) = weak.upgrade() else {
1615                    // Stale entries drop for free and do not spend a visit.
1616                    continue;
1617                };
1618                visits -= 1;
1619                let Ok(mut state) = meta.state.try_lock() else {
1620                    self.queue(band).push_back(weak);
1621                    continue;
1622                };
1623                if state.freed {
1624                    continue;
1625                }
1626                match state.residency {
1627                    // Entries for non-resident chunks drop, as in
1628                    // enforcement.
1629                    Residency::Evicted | Residency::Oversize => continue,
1630                    Residency::UnbackedResident | Residency::WriteInFlight => {
1631                        self.queue(band).push_back(weak);
1632                        continue;
1633                    }
1634                    Residency::BackedResident => {}
1635                }
1636                if state.touched || meta.class != Some(class) {
1637                    self.queue(band).push_back(weak);
1638                    continue;
1639                }
1640                // Settle the ledger in one step: the victim's bytes out, the
1641                // admitted payload's in. A steal that grows resident bytes
1642                // is an admission and must fit the budget (against evictable
1643                // bytes, as everywhere); a shrinking steal always may
1644                // proceed. On failure the victim is requeued untouched.
1645                let victim_len = u64::cast_from(meta.len_bytes());
1646                let admitted_len = u64::cast_from(admitted_len_bytes);
1647                let settled = self
1648                    .counters
1649                    .resident_bytes
1650                    .try_update(Ordering::Relaxed, Ordering::Relaxed, |cur| {
1651                        let next = cur.checked_add(admitted_len)?.saturating_sub(victim_len);
1652                        let oversize = self.counters.oversize_bytes.load(Ordering::Relaxed);
1653                        (next <= cur
1654                            || next.saturating_sub(oversize)
1655                                <= self.budget_bytes.load(Ordering::Relaxed))
1656                        .then_some(next)
1657                    })
1658                    .is_ok();
1659                if !settled {
1660                    self.queue(band).push_back(weak);
1661                    continue;
1662                }
1663                let slot = state.slot.take().expect("backed chunk has a slot");
1664                state.residency = Residency::Evicted;
1665                return Some(slot);
1666            }
1667        }
1668        None
1669    }
1670
1671    /// Capacity of the compressed-resident tier: the RSS target's headroom
1672    /// above the slot budget and the warm cap. With no target set the tier
1673    /// has zero capacity, so extents page out as soon as they are written.
1674    fn compressed_cap(&self) -> u64 {
1675        let target = self.rss_target_bytes.load(Ordering::Relaxed);
1676        let floor = self
1677            .budget_bytes
1678            .load(Ordering::Relaxed)
1679            .saturating_add(self.warm_cap());
1680        target.saturating_sub(floor)
1681    }
1682
1683    /// Counts a newly resident extent (written, or revived by a read)
1684    /// against the compressed tier. A reclaimable extent additionally
1685    /// enqueues its chunk for RSS-target enforcement; an unreclaimable one
1686    /// (a heap-fallback extent, which is never advised out) counts against
1687    /// the unreclaimable gauge instead and stays out of the queue, so
1688    /// enforcement never walks entries it cannot act on. Callers hold the
1689    /// chunk's state lock with the extent present and resident, and follow
1690    /// up with [`PoolInner::enforce_compressed_cap`] once the lock is
1691    /// released.
1692    ///
1693    /// Invariant: `extent_resident_bytes` equals the sum of `alloc_size`
1694    /// over live chunks' extents whose `is_resident()` is true, and
1695    /// `extent_residents` counts those extents; `extent_unreclaimable_bytes`
1696    /// is the subset whose `pageout_capped()` is true. This method,
1697    /// [`PoolInner::note_extent_reclaimable`],
1698    /// [`PoolInner::note_extent_released`], and the pageout arms in
1699    /// [`PoolInner::enforce_compressed_cap`] are the only adjusters; every
1700    /// flag flip pairs with one of them under the chunk's state lock.
1701    fn note_extent_resident(&self, meta: &Arc<ChunkMeta>, extent_alloc: usize, reclaimable: bool) {
1702        self.counters
1703            .extent_resident_bytes
1704            .fetch_add(u64::cast_from(extent_alloc), Ordering::Relaxed);
1705        self.extent_residents.fetch_add(1, Ordering::Relaxed);
1706        if reclaimable {
1707            self.prune_extent_queue();
1708            self.extent_queue().push_back(Arc::downgrade(meta));
1709        } else {
1710            self.counters
1711                .extent_unreclaimable_bytes
1712                .fetch_add(u64::cast_from(extent_alloc), Ordering::Relaxed);
1713        }
1714    }
1715
1716    /// Returns a retry-capped resident extent to the reclaimable set after a
1717    /// read restored its pageout budget: uncounts it from the unreclaimable
1718    /// gauge and re-enqueues its chunk for RSS-target enforcement. The
1719    /// caller holds the chunk's state lock with the extent present, resident,
1720    /// and no longer `pageout_capped()`.
1721    fn note_extent_reclaimable(&self, meta: &Arc<ChunkMeta>, extent_alloc: usize) {
1722        self.counters
1723            .extent_unreclaimable_bytes
1724            .fetch_sub(u64::cast_from(extent_alloc), Ordering::Relaxed);
1725        self.prune_extent_queue();
1726        self.extent_queue().push_back(Arc::downgrade(meta));
1727    }
1728
1729    /// Uncounts a resident extent that is being dropped (chunk freed or
1730    /// degraded). Its queue entry goes stale and is dropped on visit or by
1731    /// [`PoolInner::prune_extent_queue`].
1732    fn note_extent_released(&self, extent: &SwapExtent) {
1733        if extent.is_resident() {
1734            self.counters
1735                .extent_resident_bytes
1736                .fetch_sub(u64::cast_from(extent.alloc_size()), Ordering::Relaxed);
1737            self.extent_residents.fetch_sub(1, Ordering::Relaxed);
1738            if extent.pageout_capped() {
1739                self.counters
1740                    .extent_unreclaimable_bytes
1741                    .fetch_sub(u64::cast_from(extent.alloc_size()), Ordering::Relaxed);
1742            }
1743        }
1744    }
1745
1746    /// Drops extent-queue entries whose chunk has been freed, mirroring
1747    /// [`PoolInner::prune_queues`]: compact only when the queue outgrows
1748    /// all live resident extents (plus a small floor), so the cost
1749    /// amortizes to a constant per push. Enforcement drops stale entries
1750    /// too, but only while the tier is over capacity. A pool that stays
1751    /// under its compressed cap would otherwise accumulate an entry (and a
1752    /// pin on the dead chunk's allocation) per freed extent forever.
1753    fn prune_extent_queue(&self) {
1754        let live = usize::cast_from(self.extent_residents.load(Ordering::Relaxed));
1755        let mut queue = self.extent_queue();
1756        if queue.len() > 2 * live + 16 {
1757            queue.retain(|weak| weak.strong_count() > 0);
1758        }
1759    }
1760
1761    /// Routes compressed-cap enforcement off latency-sensitive threads: with
1762    /// spill threads running, wakes one to perform the pageouts
1763    /// (`MADV_PAGEOUT` is synchronous reclaim — page-table walks, TLB
1764    /// shootdowns, writeback submission — bounded per compressed extent but
1765    /// not free at chunk rates); without them, enforces inline as the only
1766    /// option.
1767    ///
1768    /// The routing rule across the two ceilings: budget pressure goes
1769    /// through [`PoolInner::enforce_budget`], single-flighted because
1770    /// concurrent passes would convoy on redundant compression scans; tier
1771    /// pressure goes through this router, and the enforcement itself is
1772    /// deliberately *not* single-flighted, since concurrent passes pop
1773    /// disjoint victims and each visit is microseconds. Spill threads call
1774    /// [`PoolInner::enforce_compressed_cap`] directly (they *are* the
1775    /// deferral target), and [`Pool::set_rss_target`] enforces a shrink
1776    /// inline so config changes land synchronously, mirroring `set_budget`.
1777    ///
1778    /// Deferral makes the target eventually-enforced with bounded lag (a
1779    /// notify with every spill thread mid-job is absorbed; the next loop
1780    /// pass catches up). The backstop below turns that into a bound by
1781    /// construction: a caller finding the tier at double its capacity
1782    /// enforces inline regardless, so sustained creation can never outrun
1783    /// trimming by more than one capacity's worth.
1784    ///
1785    /// Deferral tests for thread existence alone, not `spill.enabled`:
1786    /// spawned threads trim the tier in their loop for as long as they live,
1787    /// even with eviction hand-off disabled, so they remain the better home
1788    /// for the pageouts.
1789    fn enforce_or_defer_compressed_cap(&self) {
1790        if self.spill.threads.load(Ordering::Relaxed) > 0 {
1791            // The inline backstop keys on the bytes enforcement can actually
1792            // reclaim. Unreclaimable extents (retry-capped, heap-backed)
1793            // would otherwise hold the backstop permanently over threshold
1794            // and put a full enforcement pass on every caller.
1795            let resident = self.counters.extent_resident_bytes.load(Ordering::Relaxed);
1796            let unreclaimable = self
1797                .counters
1798                .extent_unreclaimable_bytes
1799                .load(Ordering::Relaxed);
1800            if resident.saturating_sub(unreclaimable) > self.compressed_cap().saturating_mul(2) {
1801                self.enforce_compressed_cap();
1802            } else {
1803                self.spill.cv.notify_one();
1804            }
1805        } else {
1806            self.enforce_compressed_cap();
1807        }
1808    }
1809
1810    /// Pages out the oldest resident extents until the compressed tier falls
1811    /// to its capacity. The compression is already paid and the device write
1812    /// is the kernel's async writeback, so each pageout is one bounded
1813    /// madvise plus a page-table observation; spill threads run this between
1814    /// jobs, and other threads only when no spill threads exist (see
1815    /// [`PoolInner::enforce_or_defer_compressed_cap`]). Not single-flighted:
1816    /// concurrent passes pop disjoint victims. Visits are bounded by the
1817    /// queue's length at entry; stale entries (extent paged out, dropped, or
1818    /// chunk dead) are dropped. Incomplete extents are requeued with their
1819    /// accounting intact until their retry budget runs out, at which point
1820    /// they leave the queue with their bytes on the unreclaimable gauge, so
1821    /// the tier may settle above its capacity by the bytes the kernel
1822    /// declined to reclaim without enforcement re-walking them.
1823    fn enforce_compressed_cap(&self) {
1824        let cap = self.compressed_cap();
1825        let resident = |c: &Counters| c.extent_resident_bytes.load(Ordering::Relaxed);
1826        // Under-cap is the common case: answer it with one atomic load and
1827        // no queue lock, so frequent callers (the spill loop) stay cheap.
1828        if resident(&self.counters) <= cap {
1829            return;
1830        }
1831        let mut remaining = self.extent_queue().len();
1832        while remaining > 0 && resident(&self.counters) > cap {
1833            remaining -= 1;
1834            let popped = self.extent_queue().pop_front();
1835            let Some(weak) = popped else {
1836                break;
1837            };
1838            let Some(meta) = weak.upgrade() else {
1839                continue;
1840            };
1841            // `try_lock`: a chunk mid-read or mid-compression holds its lock
1842            // for milliseconds; requeue rather than convoy behind it.
1843            let Ok(mut state) = meta.state.try_lock() else {
1844                self.extent_queue().push_back(weak);
1845                continue;
1846            };
1847            match &mut state.extent {
1848                Some(extent) if extent.is_resident() => {
1849                    if extent.pageout_capped() {
1850                        // A leftover entry for an already-capped extent (its
1851                        // capping transition below accounted it and dropped
1852                        // its entry): drop this one too. The read that
1853                        // restores the retry budget re-enqueues the chunk.
1854                    } else if extent.pageout() {
1855                        self.counters
1856                            .extent_resident_bytes
1857                            .fetch_sub(u64::cast_from(extent.alloc_size()), Ordering::Relaxed);
1858                        self.extent_residents.fetch_sub(1, Ordering::Relaxed);
1859                        self.counters
1860                            .extent_pageouts
1861                            .fetch_add(1, Ordering::Relaxed);
1862                    } else {
1863                        // The advice left pages resident. The extent keeps
1864                        // its full accounting (the ledger may over-count
1865                        // RSS, the safe direction).
1866                        self.counters
1867                            .extent_pageout_incomplete
1868                            .fetch_add(1, Ordering::Relaxed);
1869                        if extent.pageout_capped() {
1870                            // The retry budget just ran out: the extent
1871                            // leaves the queue and its bytes move to the
1872                            // unreclaimable gauge, so enforcement and the
1873                            // inline backstop stop chasing memory the kernel
1874                            // will not give back. A read that restores the
1875                            // budget re-counts and re-enqueues it.
1876                            self.counters
1877                                .extent_unreclaimable_bytes
1878                                .fetch_add(u64::cast_from(extent.alloc_size()), Ordering::Relaxed);
1879                        } else {
1880                            // Budget remains: keep the queue slot so later
1881                            // passes retry it up to the cap.
1882                            self.extent_queue().push_back(weak);
1883                        }
1884                    }
1885                }
1886                // Paged out already or dropped: the entry is stale. A later
1887                // resident event re-enqueues.
1888                _ => {}
1889            }
1890        }
1891    }
1892
1893    /// If the chunk is a live `UnbackedResident` holding a slot and the
1894    /// spill threads have capacity, transitions it to `WriteInFlight` and
1895    /// hands it to them, returning `true`. The hand-off happens under the
1896    /// held state lock; the spill thread blocks on that lock only after this
1897    /// call returns and the caller releases it.
1898    fn spill_handoff(&self, meta: &Arc<ChunkMeta>, state: &mut ChunkState) -> bool {
1899        // The slot check excludes empty chunks, which are `UnbackedResident`
1900        // without a slot: handing one off would panic the spill thread on
1901        // the missing slot.
1902        if state.residency != Residency::UnbackedResident
1903            || state.freed
1904            || state.slot.is_none()
1905            || !self.spill_eligible()
1906        {
1907            return false;
1908        }
1909        state.residency = Residency::WriteInFlight;
1910        self.spill_schedule(Arc::clone(meta));
1911        true
1912    }
1913}
1914
1915impl ChunkHandle {
1916    /// Test hook: the chunk's current residency state.
1917    #[cfg(test)]
1918    fn residency(&self) -> Residency {
1919        self.meta.state().residency
1920    }
1921
1922    /// Copies the whole contents into `dst` (cleared first), leaving the
1923    /// chunk's residency untouched: a resident slot is copied out directly,
1924    /// and an evicted extent decompresses straight into `dst` without
1925    /// allocating a slot. A read therefore never raises resident bytes,
1926    /// never converts the chunk's state, and hands out no reference into
1927    /// pool memory.
1928    ///
1929    /// The copy runs under the chunk's state lock, which is what makes the
1930    /// no-reference contract cheap: eviction takes the same lock, so there
1931    /// is no reader it could race. The admitting variant is
1932    /// [`ChunkHandle::read_into_admit`].
1933    pub fn read_into(&self, dst: &mut Vec<u64>) {
1934        self.read_impl(0..self.meta.len, dst, false);
1935    }
1936
1937    /// As [`ChunkHandle::read_into`], restricted to the word range `range`
1938    /// of the chunk's contents, which must lie within them. `dst` receives
1939    /// exactly the range.
1940    ///
1941    /// The range narrows only the copy into `dst`: the swap backend's
1942    /// stored form is a whole compressed block, so a cold read still
1943    /// faults and decompresses the entire extent, and accounting is that
1944    /// of a whole-chunk read. A backend with a rangeable stored form can
1945    /// serve the same call with range-proportional I/O.
1946    pub fn read_range_into(&self, range: Range<usize>, dst: &mut Vec<u64>) {
1947        self.read_impl(range, dst, false);
1948    }
1949
1950    /// As [`ChunkHandle::read_into`], except that an evicted chunk is
1951    /// re-admitted to `BackedResident` (its extent kept, its touched bit
1952    /// set) when a slot is available from free budget headroom or by
1953    /// stealing from a clean backed victim of the same size class, never by
1954    /// evicting or compressing anything. When neither source yields a slot
1955    /// the read is served as a plain decompress and the chunk stays
1956    /// evicted.
1957    ///
1958    /// For demand reads on probe paths, where the same chunk is likely to
1959    /// be read again. Merge, drain, and other consume-once paths should use
1960    /// [`ChunkHandle::read_into`] or [`ChunkHandle::take`]: admitting there
1961    /// churns the clean-victim stock that eager backing exists to build,
1962    /// evicting probe targets to house data about to die.
1963    pub fn read_into_admit(&self, dst: &mut Vec<u64>) {
1964        self.read_impl(0..self.meta.len, dst, true);
1965    }
1966
1967    /// As [`ChunkHandle::read_into_admit`], restricted to the word range
1968    /// `range` per [`ChunkHandle::read_range_into`]. Admission is
1969    /// whole-chunk regardless of the range: the acquired slot holds the
1970    /// entire body.
1971    pub fn read_range_into_admit(&self, range: Range<usize>, dst: &mut Vec<u64>) {
1972        self.read_impl(range, dst, true);
1973    }
1974
1975    /// Shared body of the copy-out reads: fills `dst` with the word range
1976    /// `range` of the chunk's contents under the chunk's state lock,
1977    /// re-admitting an evicted chunk when `admit` is set and a slot is
1978    /// available. An empty range returns without locking or touching the
1979    /// chunk, like the whole-chunk read of an empty chunk always has.
1980    fn read_impl(&self, range: Range<usize>, dst: &mut Vec<u64>, admit: bool) {
1981        dst.clear();
1982        let meta = &*self.meta;
1983        assert!(
1984            range.start <= range.end && range.end <= meta.len,
1985            "range {range:?} exceeds the chunk's {} words",
1986            meta.len,
1987        );
1988        if range.is_empty() {
1989            return;
1990        }
1991        let mut state = meta.state();
1992        state.touched = true;
1993        let mut extent_revived = false;
1994        match state.residency {
1995            Residency::Oversize => {
1996                let payload = state.oversize.as_ref().expect("oversize chunk has payload");
1997                dst.extend_from_slice(&payload[range]);
1998            }
1999            Residency::Evicted => {
2000                let slot = if admit {
2001                    meta.pool.admit_slot(meta)
2002                } else {
2003                    None
2004                };
2005                let extent = state.extent.as_mut().expect("evicted chunk has an extent");
2006                // Reading faults the extent's pages back in either way, so
2007                // it is re-counted against the compressed tier below.
2008                let was_resident = extent.is_resident();
2009                let was_capped = extent.pageout_capped();
2010                let extent_alloc = extent.alloc_size();
2011                match slot {
2012                    Some(slot) => {
2013                        // Admission: the extent decompresses straight into
2014                        // the acquired slot, fully overwriting its
2015                        // unspecified prior contents, and the caller's
2016                        // buffer is filled from the slot.
2017                        let region = meta.pool.region_of(meta);
2018                        // SAFETY: the slot was acquired for this chunk
2019                        // under its held state lock (freshly allocated, or
2020                        // transferred from the victim under the victim's
2021                        // lock), so it is exclusively owned with no other
2022                        // reference into it, and `len_bytes` fits the
2023                        // class.
2024                        let slot_bytes = unsafe {
2025                            std::slice::from_raw_parts_mut(region.slot_ptr(slot), meta.len_bytes())
2026                        };
2027                        extent.read_into(meta.codec, slot_bytes);
2028                        state.slot = Some(slot);
2029                        state.residency = Residency::BackedResident;
2030                        // SAFETY: the slot belongs to this chunk while the
2031                        // state lock is held (eviction and free both take
2032                        // it).
2033                        let src = unsafe { meta.pool.slot_data(meta, slot) };
2034                        dst.extend_from_slice(&src[range.start..range.end]);
2035                        // Resident again: rejoin the eviction candidates.
2036                        // A leftover entry from before the chunk's eviction
2037                        // stays sound (each entry is validated against the
2038                        // chunk's state on visit) but costs policy: two live
2039                        // entries give the enforcer two chances to spend
2040                        // this chunk's single touched bit, halving its
2041                        // second chance until one entry drains.
2042                        meta.pool
2043                            .queue(band(meta.depth))
2044                            .push_back(Arc::downgrade(&self.meta));
2045                    }
2046                    None => {
2047                        // The zero-fill ahead of the decompress is deliberate
2048                        // waste (~a tenth of the decompress cost): the extent
2049                        // read takes an initialized `&mut [u8]`, so skipping
2050                        // the fill would mean exposing uninitialized memory
2051                        // through a safe reference. Callers that read
2052                        // repeatedly amortize it by reusing `dst` within a
2053                        // pass, shrinking it back after oversized reads (a
2054                        // reused buffer otherwise ratchets to the largest
2055                        // chunk it ever carried, invisible to every pool
2056                        // gauge; the design doc's reader discipline).
2057                        dst.resize(range.end - range.start, 0);
2058                        let bytes: &mut [u8] = bytemuck::cast_slice_mut(dst.as_mut_slice());
2059                        extent.read_range_into(
2060                            meta.codec,
2061                            meta.len_bytes(),
2062                            range.start * 8,
2063                            bytes,
2064                        );
2065                    }
2066                }
2067                // TODO: a sub-range read of a rangeable stored form (file
2068                // extents, a sub-block-framed codec) revives only part of
2069                // the extent; the whole-extent accounting below would then
2070                // overcount and needs a partial-revival variant.
2071                if !was_resident {
2072                    // Revived from the device: the decompress reset any
2073                    // retry budget, so the extent re-enters reclaimable.
2074                    meta.pool
2075                        .note_extent_resident(&self.meta, extent_alloc, true);
2076                    extent_revived = true;
2077                } else if was_capped {
2078                    // The decompress faulted every page and reset the
2079                    // pageout retry budget, so a retry-capped extent is
2080                    // reclaimable again. Heap-backed extents stay
2081                    // structurally capped and stay out of the queue.
2082                    let capped = state
2083                        .extent
2084                        .as_ref()
2085                        .expect("evicted chunk has an extent")
2086                        .pageout_capped();
2087                    if !capped {
2088                        meta.pool.note_extent_reclaimable(&self.meta, extent_alloc);
2089                        extent_revived = true;
2090                    }
2091                }
2092            }
2093            Residency::UnbackedResident | Residency::BackedResident | Residency::WriteInFlight => {
2094                let slot = state.slot.expect("resident non-empty chunk has a slot");
2095                // SAFETY: the slot belongs to this chunk while the state lock
2096                // is held (eviction and free both take it).
2097                let src = unsafe { meta.pool.slot_data(meta, slot) };
2098                dst.extend_from_slice(&src[range.start..range.end]);
2099            }
2100        }
2101        drop(state);
2102        // The read revived the extent's compressed pages; the tier may need
2103        // trimming. Enforcement locks chunk states itself, so it must run
2104        // after the unlock.
2105        if extent_revived {
2106            meta.pool.enforce_or_defer_compressed_cap();
2107        }
2108    }
2109
2110    /// Copies the whole contents into `dst` (per [`ChunkHandle::read_into`],
2111    /// never admitting) and frees the chunk, cancelling any in-flight
2112    /// backing write.
2113    pub fn take(self, dst: &mut Vec<u64>) {
2114        self.read_into(dst);
2115    }
2116
2117    /// Advisory a consumer may issue before a bulk read: hints the kernel to
2118    /// swap an evicted chunk's extent back in, and is a no-op in every other
2119    /// state. Never blocks on I/O (`MADV_WILLNEED` is asynchronous).
2120    pub fn prefetch(&self) {
2121        let state = self.meta.state();
2122        if state.residency == Residency::Evicted {
2123            let extent = state.extent.as_ref().expect("evicted chunk has an extent");
2124            extent.prefetch();
2125        }
2126    }
2127
2128    /// As [`ChunkHandle::prefetch`], scoped to the word range `range` of the
2129    /// chunk's contents. The range is advisory: a backend hints at whatever
2130    /// granularity its stored form permits, and the swap backend's stored
2131    /// form is a whole compressed block, so it hints the entire extent.
2132    pub fn prefetch_range(&self, range: Range<usize>) {
2133        let _ = range;
2134        self.prefetch();
2135    }
2136
2137    /// Test hook: the byte size of the chunk's size class, or `None` for
2138    /// empty and oversize chunks.
2139    #[cfg(test)]
2140    fn size_class_bytes(&self) -> Option<usize> {
2141        self.meta.class.map(|class| SIZE_CLASSES[class])
2142    }
2143}
2144
2145impl Drop for ChunkHandle {
2146    fn drop(&mut self) {
2147        let pool = &self.meta.pool;
2148        let mut state = self.meta.state();
2149        pool.counters.frees.fetch_add(1, Ordering::Relaxed);
2150        state.freed = true;
2151        if self.meta.class.is_some() {
2152            pool.live_chunks.fetch_sub(1, Ordering::Relaxed);
2153        }
2154        let len_bytes = u64::cast_from(self.meta.len_bytes());
2155        // `release_slot`'s precondition holds in every arm below: the handle
2156        // is being dropped, so no copy-out read (which borrows the handle)
2157        // is in progress, and `freed` was set under the state lock held
2158        // here, so concurrent queue visitors skip the chunk.
2159        match state.residency {
2160            Residency::UnbackedResident => {
2161                if state.slot.is_some() {
2162                    pool.counters.writes_elided.fetch_add(1, Ordering::Relaxed);
2163                    pool.release_slot(&self.meta, &mut state);
2164                }
2165            }
2166            Residency::BackedResident => {
2167                pool.release_slot(&self.meta, &mut state);
2168                if let Some(extent) = &state.extent {
2169                    pool.note_extent_released(extent);
2170                }
2171                state.extent = None;
2172            }
2173            Residency::Evicted => {
2174                // Eviction already released the slot.
2175                crate::soft_assert_no_log!(state.slot.is_none(), "evicted chunk holds no slot");
2176                if let Some(extent) = &state.extent {
2177                    pool.note_extent_released(extent);
2178                }
2179                state.extent = None;
2180            }
2181            Residency::WriteInFlight => {
2182                // A spill thread may be reading the slot to compress it.
2183                // `freed` (set above) tells it the chunk died; it owns the
2184                // slot release, the `resident_bytes` decrement, and the
2185                // cancellation accounting from here.
2186            }
2187            Residency::Oversize => {
2188                pool.counters
2189                    .resident_bytes
2190                    .fetch_sub(len_bytes, Ordering::Relaxed);
2191                pool.counters
2192                    .oversize_bytes
2193                    .fetch_sub(len_bytes, Ordering::Relaxed);
2194                state.oversize = None;
2195            }
2196        }
2197    }
2198}
2199
2200#[cfg(test)]
2201mod tests {
2202    use super::*;
2203    use crate::pool::extent::TEST_CODEC;
2204
2205    /// Keep test pools small: 64 MiB of virtual reservation per class.
2206    /// Under Miri the backing is real interpreter heap rather than lazy
2207    /// virtual memory, so shrink further. Classes above the capacity yield
2208    /// empty regions whose inserts degrade to the heap fallback, which is
2209    /// fine: slotted-chunk tests exercise only the smallest classes.
2210    fn test_pool(budget_bytes: usize) -> Pool {
2211        let capacity = if cfg!(miri) { 1 << 20 } else { 64 << 20 };
2212        let pool = Pool::with_class_capacity(capacity).expect("pool creation");
2213        pool.set_budget(budget_bytes);
2214        pool
2215    }
2216
2217    /// Scales an iteration count down under Miri, where one interpreted
2218    /// compression costs what thousands do natively.
2219    fn rounds(native: u64, miri: u64) -> u64 {
2220        if cfg!(miri) { miri } else { native }
2221    }
2222
2223    fn payload(words: usize, seed: u64) -> Vec<u64> {
2224        (0..u64::cast_from(words))
2225            .map(|i| seed.wrapping_mul(0x9E3779B97F4A7C15).wrapping_add(i))
2226            .collect()
2227    }
2228
2229    /// Copies `data` into the pool and clears it.
2230    fn insert(pool: &Pool, data: &mut Vec<u64>) -> ChunkHandle {
2231        insert_at_depth(pool, 0, data)
2232    }
2233
2234    /// Copies `data` into the pool at a hinted depth and clears it.
2235    fn insert_at_depth(pool: &Pool, depth: u8, data: &mut Vec<u64>) -> ChunkHandle {
2236        let hints = ChunkHints { depth };
2237        let handle = pool.insert_with(data.len(), hints, &TEST_CODEC, |dst| {
2238            dst.copy_from_slice(data.as_slice())
2239        });
2240        data.clear();
2241        handle
2242    }
2243
2244    /// Copies a chunk's contents out into a fresh buffer.
2245    fn read(handle: &ChunkHandle) -> Vec<u64> {
2246        let mut out = Vec::new();
2247        handle.read_into(&mut out);
2248        out
2249    }
2250
2251    /// Copies a chunk's contents out into a fresh buffer via the admitting
2252    /// read.
2253    fn read_admit(handle: &ChunkHandle) -> Vec<u64> {
2254        let mut out = Vec::new();
2255        handle.read_into_admit(&mut out);
2256        out
2257    }
2258
2259    /// Words that fill a 64 KiB class exactly.
2260    const SMALL: usize = (64 << 10) / 8;
2261
2262    #[allow(dead_code)]
2263    fn assert_handle_send_sync() {
2264        fn check<T: Send + Sync>() {}
2265        check::<Pool>();
2266        check::<ChunkHandle>();
2267    }
2268
2269    /// With an RSS target set, evicted chunks keep their extents resident
2270    /// (the compressed tier); shrinking the target pages the oldest extents
2271    /// out; reads revive them and re-count them.
2272    #[mz_ore::test]
2273    fn compressed_tier_round_trip() {
2274        let pool = test_pool(256 << 20);
2275        pool.set_rss_target(1 << 30);
2276        let orig = payload(SMALL, 21);
2277        let handle = insert(&pool, &mut orig.clone());
2278        pool.evict(&handle);
2279        assert_eq!(handle.residency(), Residency::Evicted);
2280        let stats = pool.stats();
2281        assert!(
2282            stats.extent_resident_bytes > 0,
2283            "under the target, the extent stays resident",
2284        );
2285        assert_eq!(stats.extent_pageouts, 0);
2286
2287        // Shrinking the target to zero pages the extent out.
2288        pool.set_rss_target(0);
2289        let stats = pool.stats();
2290        assert_eq!(stats.extent_resident_bytes, 0, "tier collapsed");
2291        assert_eq!(stats.extent_pageouts, 1);
2292
2293        // Reading revives the extent: contents round-trip, the chunk stays
2294        // evicted, and with the target restored the revived extent is
2295        // counted again.
2296        pool.set_rss_target(1 << 30);
2297        assert_eq!(read(&handle), orig);
2298        assert_eq!(handle.residency(), Residency::Evicted);
2299        assert!(
2300            pool.stats().extent_resident_bytes > 0,
2301            "revived and counted"
2302        );
2303
2304        // Dropping the handle uncounts the resident extent.
2305        drop(handle);
2306        assert_eq!(pool.stats().extent_resident_bytes, 0);
2307    }
2308
2309    /// A ranged read returns exactly the corresponding slice of a
2310    /// whole-chunk read in every residency state, and changes residency
2311    /// exactly as the equivalent whole-chunk read would.
2312    #[mz_ore::test]
2313    fn ranged_reads_match_full_read_slice() {
2314        let pool = test_pool(256 << 20);
2315        pool.set_rss_target(1 << 30);
2316        let orig = payload(SMALL, 33);
2317        let handle = insert(&pool, &mut orig.clone());
2318        let ranges = [
2319            (0usize, 7usize),
2320            (13, 100),
2321            (SMALL - 9, 9),
2322            (0, SMALL),
2323            (5, 0),
2324        ];
2325        let check = |label: &str| {
2326            for (start, len) in ranges {
2327                let mut out = Vec::new();
2328                handle.read_range_into(start..start + len, &mut out);
2329                assert_eq!(
2330                    out,
2331                    &orig[start..start + len],
2332                    "{label} range ({start}, {len})"
2333                );
2334            }
2335        };
2336        assert_eq!(handle.residency(), Residency::UnbackedResident);
2337        check("resident");
2338        pool.evict(&handle);
2339        assert_eq!(handle.residency(), Residency::Evicted);
2340        check("evicted");
2341        assert_eq!(
2342            handle.residency(),
2343            Residency::Evicted,
2344            "plain ranged reads do not admit"
2345        );
2346        // An admitting ranged read returns the range and admits the whole
2347        // chunk.
2348        let mut out = Vec::new();
2349        handle.read_range_into_admit(3..19, &mut out);
2350        assert_eq!(out, &orig[3..19]);
2351        assert_eq!(handle.residency(), Residency::BackedResident);
2352        check("backed");
2353    }
2354
2355    #[mz_ore::test]
2356    #[should_panic(expected = "exceeds the chunk's")]
2357    fn ranged_read_out_of_bounds_panics() {
2358        let pool = test_pool(256 << 20);
2359        let handle = insert(&pool, &mut payload(SMALL, 34));
2360        let mut out = Vec::new();
2361        handle.read_range_into(SMALL - 1..SMALL + 1, &mut out);
2362    }
2363
2364    /// With no RSS target (the default), the compressed tier has zero
2365    /// capacity and extents page out as soon as they are written.
2366    #[mz_ore::test]
2367    fn default_target_pages_extents_immediately() {
2368        let pool = test_pool(256 << 20);
2369        let handle = insert(&pool, &mut payload(SMALL, 22));
2370        pool.evict(&handle);
2371        let stats = pool.stats();
2372        assert_eq!(stats.extent_resident_bytes, 0);
2373        assert_eq!(stats.extent_pageouts, 1);
2374    }
2375
2376    /// A pageout pass whose residency observation finds the whole range
2377    /// gone uncounts exactly the extent's allocation bytes.
2378    #[mz_ore::test]
2379    fn full_pageout_uncounts_exactly_the_extent() {
2380        let pool = test_pool(256 << 20);
2381        pool.set_rss_target(1 << 30);
2382        let handle = insert(&pool, &mut payload(SMALL, 50));
2383        pool.evict(&handle);
2384        let counted = pool.stats().extent_resident_bytes;
2385        assert!(counted > 0, "under the target, the extent stays counted");
2386        pool.set_rss_target(0);
2387        let stats = pool.stats();
2388        assert_eq!(stats.extent_resident_bytes, 0, "exactly `counted` left");
2389        assert_eq!(stats.extent_pageouts, 1);
2390        assert_eq!(stats.extent_pageout_incomplete, 0);
2391    }
2392
2393    /// A pageout pass the kernel declines leaves the extent fully counted
2394    /// and queued, and increments the incomplete counter. The next pass
2395    /// (advice now accepted) pages it out.
2396    #[mz_ore::test]
2397    fn incomplete_pageout_keeps_accounting_and_queue_position() {
2398        let pool = test_pool(256 << 20);
2399        pool.set_rss_target(1 << 30);
2400        let handle = insert(&pool, &mut payload(SMALL, 51));
2401        pool.evict(&handle);
2402        let counted = pool.stats().extent_resident_bytes;
2403        assert!(counted > 0);
2404        region::fake_residency::decline_next(1);
2405        pool.set_rss_target(0);
2406        let stats = pool.stats();
2407        assert_eq!(
2408            stats.extent_resident_bytes, counted,
2409            "full accounting stays"
2410        );
2411        assert_eq!(stats.extent_pageouts, 0);
2412        assert_eq!(stats.extent_pageout_incomplete, 1);
2413        assert_eq!(handle.residency(), Residency::Evicted);
2414        // The requeued entry is retried by the next enforcement pass.
2415        pool.enforce_rss_target();
2416        let stats = pool.stats();
2417        assert_eq!(stats.extent_resident_bytes, 0);
2418        assert_eq!(stats.extent_pageouts, 1);
2419        assert_eq!(stats.extent_pageout_incomplete, 1);
2420    }
2421
2422    /// A never-reclaimable extent stops being advised after the retry cap:
2423    /// the incomplete counter stops climbing, the bytes stay counted
2424    /// resident, and the tier keeps paging other extents out around it.
2425    #[mz_ore::test]
2426    fn pageout_retry_cap_stops_advising() {
2427        let pool = test_pool(256 << 20);
2428        let handle = insert(&pool, &mut payload(SMALL, 52));
2429        region::fake_residency::decline_next(u64::MAX);
2430        // RSS target zero: the eviction's enforcement pass advises at once.
2431        pool.evict(&handle);
2432        for _ in 0..5 {
2433            pool.enforce_rss_target();
2434        }
2435        let stats = pool.stats();
2436        assert_eq!(
2437            stats.extent_pageout_incomplete,
2438            u64::from(extent::PAGEOUT_RETRY_CAP),
2439            "advised exactly retry-cap times, then left alone",
2440        );
2441        assert_eq!(stats.extent_pageouts, 0);
2442        let counted = stats.extent_resident_bytes;
2443        assert!(counted > 0, "capped extent stays counted resident");
2444        // The tier functions around the capped extent: a fresh extent still
2445        // pages out.
2446        region::fake_residency::decline_next(0);
2447        let other = insert(&pool, &mut payload(SMALL, 53));
2448        pool.evict(&other);
2449        let stats = pool.stats();
2450        assert_eq!(stats.extent_pageouts, 1);
2451        assert_eq!(
2452            stats.extent_resident_bytes, counted,
2453            "only the capped extent remains counted",
2454        );
2455        assert_eq!(read(&handle).len(), SMALL, "capped extent stays readable");
2456    }
2457
2458    /// Reading a retry-capped extent faults its pages back in and resets
2459    /// the retry budget, so a later pass can page it out.
2460    #[mz_ore::test]
2461    fn read_resets_pageout_retry_budget() {
2462        let pool = test_pool(256 << 20);
2463        let orig = payload(SMALL, 54);
2464        let handle = insert(&pool, &mut orig.clone());
2465        region::fake_residency::decline_next(u64::MAX);
2466        pool.evict(&handle);
2467        for _ in 0..4 {
2468            pool.enforce_rss_target();
2469        }
2470        assert_eq!(
2471            pool.stats().extent_pageout_incomplete,
2472            u64::from(extent::PAGEOUT_RETRY_CAP),
2473            "capped",
2474        );
2475        assert!(pool.stats().extent_resident_bytes > 0);
2476        region::fake_residency::decline_next(0);
2477        assert_eq!(read(&handle), orig);
2478        pool.enforce_rss_target();
2479        let stats = pool.stats();
2480        assert_eq!(stats.extent_pageouts, 1, "the budget reset re-advised it");
2481        assert_eq!(stats.extent_resident_bytes, 0);
2482        // The paged-out extent still round-trips.
2483        assert_eq!(read(&handle), orig);
2484    }
2485
2486    /// Eager backing compresses a chunk to `BackedResident` while it stays
2487    /// readable in its slot; the later budget-driven eviction is a pure page
2488    /// release, and the contents round-trip through the extent.
2489    #[mz_ore::test]
2490    fn eager_backing_round_trip() {
2491        let pool = test_pool(256 << 20);
2492        let orig = payload(SMALL, 11);
2493        let handle = insert(&pool, &mut orig.clone());
2494        assert_eq!(handle.residency(), Residency::UnbackedResident);
2495
2496        assert!(pool.back_step(), "one chunk is backable");
2497        assert_eq!(handle.residency(), Residency::BackedResident);
2498        let stats = pool.stats();
2499        assert_eq!(stats.eager_backs, 1);
2500        assert_eq!(
2501            stats.evictions_compress, 0,
2502            "backing is not an eviction and compresses off the eviction counter",
2503        );
2504        assert!(stats.extent_bytes_written > 0);
2505
2506        // Still readable straight from the slot: the chunk is resident.
2507        assert_eq!(read(&handle), orig);
2508
2509        // The pre-paid eviction is cheap, and the extent round-trips.
2510        pool.evict(&handle);
2511        assert_eq!(handle.residency(), Residency::Evicted);
2512        assert_eq!(pool.stats().evictions_cheap, 1);
2513        pool.poison_free_slots();
2514        assert_eq!(read(&handle), orig);
2515    }
2516
2517    /// Once everything reachable is backed, the backing scan reports no
2518    /// progress so spill threads park instead of rescanning a fully-backed
2519    /// queue forever.
2520    #[mz_ore::test]
2521    fn backing_reports_no_progress_when_all_backed() {
2522        let pool = test_pool(256 << 20);
2523        let _handle = insert(&pool, &mut payload(SMALL, 31));
2524        assert!(pool.back_step(), "one unbacked chunk is actionable");
2525        assert!(
2526            !pool.back_step(),
2527            "a fully-backed queue is not progress; callers must park",
2528        );
2529        assert_eq!(pool.stats().eager_backs, 1);
2530    }
2531
2532    /// Freeing under the warm cap parks the slot warm; the next insert of the
2533    /// same class reuses it fault-free and the accounting balances.
2534    #[mz_ore::test]
2535    fn warm_slot_reuse() {
2536        // Budget 8 MiB: warm cap = 1 MiB, so a 64 KiB slot fits warm.
2537        let pool = test_pool(8 << 20);
2538        let orig = payload(SMALL, 7);
2539        let handle = insert(&pool, &mut orig.clone());
2540        drop(handle);
2541        let after_free = pool.stats();
2542        assert_eq!(after_free.warm_bytes, 64 << 10, "freed slot parks warm");
2543        assert_eq!(after_free.warm_reuses, 0);
2544
2545        let handle = insert(&pool, &mut orig.clone());
2546        let after_reuse = pool.stats();
2547        assert_eq!(after_reuse.warm_reuses, 1, "second insert reuses warm slot");
2548        assert_eq!(after_reuse.warm_bytes, 0, "reuse drains the warm pool");
2549        // Contents are correct despite the skipped page release.
2550        assert_eq!(read(&handle), orig);
2551    }
2552
2553    /// The warm pool is capped at an eighth of the budget; frees beyond the
2554    /// cap release their pages and park cold.
2555    #[mz_ore::test]
2556    fn warm_pool_respects_cap() {
2557        // Budget 1 MiB: warm cap = 128 KiB = two 64 KiB slots.
2558        let pool = test_pool(1 << 20);
2559        let handles: Vec<_> = (0..4)
2560            .map(|seed| insert(&pool, &mut payload(SMALL, seed)))
2561            .collect();
2562        drop(handles);
2563        let stats = pool.stats();
2564        assert_eq!(
2565            stats.warm_bytes,
2566            128 << 10,
2567            "warm pool stops at the budget/8 cap",
2568        );
2569    }
2570
2571    /// A kernel that keeps declining the reclaim advice caps the extent's
2572    /// retry budget: the extent leaves the enforcement queue and moves to
2573    /// the unreclaimable gauge, so enforcement stops walking it, and a read
2574    /// that restores the budget makes it reclaimable and pageable again.
2575    #[mz_ore::test]
2576    fn capped_extents_leave_the_enforcement_queue() {
2577        let pool = test_pool(256 << 20);
2578        let orig = payload(SMALL, 960);
2579        let handle = insert(&pool, &mut orig.clone());
2580        // Decline every observation: eviction's own enforcement pass plus
2581        // the passes below spend the whole retry budget.
2582        region::fake_residency::decline_next(u64::from(extent::PAGEOUT_RETRY_CAP));
2583        pool.evict(&handle);
2584        for _ in 0..extent::PAGEOUT_RETRY_CAP {
2585            pool.enforce_compressed();
2586        }
2587        let stats = pool.stats();
2588        assert_eq!(
2589            stats.extent_pageout_incomplete,
2590            u64::from(extent::PAGEOUT_RETRY_CAP),
2591        );
2592        assert!(stats.extent_unreclaimable_bytes > 0, "capped bytes counted");
2593        assert_eq!(pool.extent_queue_len(), 0, "capped extents leave the queue",);
2594        // Further enforcement is a no-op: nothing queued, no advice spent.
2595        pool.enforce_compressed();
2596        assert_eq!(
2597            pool.stats().extent_pageout_incomplete,
2598            u64::from(extent::PAGEOUT_RETRY_CAP),
2599        );
2600
2601        // A read faults everything back in and restores the retry budget:
2602        // the extent re-enters the reclaimable set and, with the kernel now
2603        // cooperating, the read's own enforcement pass pages it out.
2604        assert_eq!(read(&handle), orig);
2605        let stats = pool.stats();
2606        assert_eq!(stats.extent_unreclaimable_bytes, 0, "budget restored");
2607        assert_eq!(stats.extent_pageouts, 1, "re-enqueued extent pages out");
2608        assert_eq!(stats.extent_resident_bytes, 0);
2609        assert_eq!(pool.extent_queue_len(), 0);
2610    }
2611
2612    /// Shrinking the budget cools warm slots parked under the old, larger
2613    /// cap: their pages are released and `warm_bytes` falls to the new cap
2614    /// on the shrink itself, not on eventual same-class reuse.
2615    #[mz_ore::test]
2616    fn budget_shrink_trims_warm_pool() {
2617        // Budget 8 MiB: warm cap 1 MiB, so four 64 KiB frees all park warm.
2618        let pool = test_pool(8 << 20);
2619        let handles: Vec<_> = (0..4)
2620            .map(|seed| insert(&pool, &mut payload(SMALL, 950 + seed)))
2621            .collect();
2622        drop(handles);
2623        assert_eq!(pool.stats().warm_bytes, 4 * (64 << 10));
2624
2625        // Budget 1 MiB: warm cap 128 KiB, so two of the four slots cool.
2626        pool.set_budget(1 << 20);
2627        assert_eq!(pool.stats().warm_bytes, 128 << 10);
2628    }
2629
2630    #[mz_ore::test]
2631    fn round_trip_resident() {
2632        let pool = test_pool(256 << 20);
2633        let orig = payload(1000, 1);
2634        let mut data = orig.clone();
2635        let capacity = data.capacity();
2636        let handle = insert(&pool, &mut data);
2637        assert!(data.is_empty());
2638        assert_eq!(data.capacity(), capacity, "insert preserves capacity");
2639        assert_eq!(handle.residency(), Residency::UnbackedResident);
2640        assert_eq!(read(&handle), orig);
2641        drop(handle);
2642        let stats = pool.stats();
2643        assert_eq!(stats.inserts, 1);
2644        assert_eq!(stats.frees, 1);
2645        assert_eq!(stats.resident_bytes, 0);
2646    }
2647
2648    /// `take` is the terminal read: the contents come back and the consumed
2649    /// handle frees the chunk, eliding the backing write of a chunk that was
2650    /// still unbacked.
2651    #[mz_ore::test]
2652    fn take_reads_and_frees() {
2653        let pool = test_pool(256 << 20);
2654        let orig = payload(SMALL, 40);
2655        let handle = insert(&pool, &mut orig.clone());
2656        let mut out = Vec::new();
2657        handle.take(&mut out);
2658        assert_eq!(out, orig);
2659        let stats = pool.stats();
2660        assert_eq!(stats.frees, 1);
2661        assert_eq!(stats.writes_elided, 1, "a resident take never writes");
2662        assert_eq!(stats.resident_bytes, 0);
2663        assert_eq!(stats.live_chunks, 0);
2664    }
2665
2666    /// `prefetch` is safe wherever it lands: on a resident chunk (a no-op),
2667    /// on an evicted chunk (whose read then round-trips), and issued with no
2668    /// read following it. It never changes residency or resident bytes.
2669    #[mz_ore::test]
2670    fn prefetch_is_safe_in_every_state() {
2671        let pool = test_pool(256 << 20);
2672        let orig = payload(SMALL, 41);
2673        let handle = insert(&pool, &mut orig.clone());
2674        handle.prefetch();
2675        assert_eq!(handle.residency(), Residency::UnbackedResident);
2676        assert_eq!(read(&handle), orig);
2677        pool.evict(&handle);
2678        handle.prefetch();
2679        assert_eq!(handle.residency(), Residency::Evicted);
2680        assert_eq!(read(&handle), orig);
2681        // An advisory with no read behind it leaves nothing to clean up.
2682        let idle = insert(&pool, &mut payload(SMALL, 42));
2683        idle.prefetch();
2684        drop(idle);
2685        drop(handle);
2686        assert_eq!(pool.stats().resident_bytes, 0);
2687    }
2688
2689    /// Reading an evicted chunk decompresses its extent straight into the
2690    /// caller's buffer and leaves the chunk evicted. Free slots are poisoned
2691    /// first, so a read passing stale slot memory through (the macOS
2692    /// `MADV_DONTNEED` hazard) would fail the content check.
2693    #[mz_ore::test]
2694    fn evict_then_read_preserves_contents() {
2695        let pool = test_pool(256 << 20);
2696        let orig = payload(SMALL, 2);
2697        let handle = insert(&pool, &mut orig.clone());
2698        pool.evict(&handle);
2699        assert_eq!(handle.residency(), Residency::Evicted);
2700        let stats = pool.stats();
2701        assert_eq!(stats.evictions_compress, 1);
2702        assert_eq!(stats.resident_bytes, 0);
2703        assert!(stats.extent_bytes_written > 0);
2704        pool.poison_free_slots();
2705        assert_eq!(read(&handle), orig);
2706        assert_eq!(handle.residency(), Residency::Evicted);
2707        assert_eq!(pool.stats().resident_bytes, 0, "reads copy out");
2708    }
2709
2710    /// An admitting read of an evicted chunk with budget headroom re-admits
2711    /// it: contents round-trip, the chunk lands `BackedResident` with its
2712    /// extent kept, and later reads serve from the slot without touching
2713    /// the extent.
2714    #[mz_ore::test]
2715    fn admit_from_free_budget_backs_the_chunk() {
2716        let pool = test_pool(256 << 20);
2717        let orig = payload(SMALL, 70);
2718        let handle = insert(&pool, &mut orig.clone());
2719        pool.evict(&handle);
2720        assert_eq!(handle.residency(), Residency::Evicted);
2721        assert_eq!(pool.stats().resident_bytes, 0);
2722
2723        pool.poison_free_slots();
2724        assert_eq!(read_admit(&handle), orig);
2725        assert_eq!(handle.residency(), Residency::BackedResident);
2726        let stats = pool.stats();
2727        assert_eq!(stats.admissions_budget, 1);
2728        assert_eq!(stats.admissions_steal, 0);
2729        assert_eq!(stats.admissions_denied, 0);
2730        assert_eq!(stats.resident_bytes, 64 << 10);
2731
2732        // Later reads serve from the slot and never touch the extent: a
2733        // decompress would revive its pages and move the revival and
2734        // pageout counters.
2735        let pageouts = stats.extent_pageouts;
2736        let extent_resident = stats.extent_resident_bytes;
2737        assert_eq!(read(&handle), orig);
2738        assert_eq!(handle.residency(), Residency::BackedResident);
2739        let stats = pool.stats();
2740        assert_eq!(stats.extent_pageouts, pageouts);
2741        assert_eq!(stats.extent_resident_bytes, extent_resident);
2742
2743        // The kept extent pre-pays the next eviction, and round-trips.
2744        pool.evict(&handle);
2745        assert_eq!(handle.residency(), Residency::Evicted);
2746        let stats = pool.stats();
2747        assert_eq!(stats.evictions_cheap, 1);
2748        assert_eq!(stats.evictions_compress, 1, "admission wrote no extent");
2749        pool.poison_free_slots();
2750        assert_eq!(read(&handle), orig);
2751        drop(handle);
2752        assert_eq!(pool.stats().resident_bytes, 0);
2753    }
2754
2755    /// With the budget pinned full and a clean backed victim of the same
2756    /// class, an admitting read steals the victim's slot: the victim is
2757    /// evicted with zero I/O and its extent intact, the admitted chunk
2758    /// lands `BackedResident`, and resident bytes, warm bytes, and the
2759    /// compression and pageout counters are all unchanged.
2760    #[mz_ore::test]
2761    fn admit_steals_clean_victim_slot() {
2762        let pool = test_pool(256 << 20);
2763        pool.set_rss_target(1 << 30);
2764        let victim_orig = payload(SMALL, 71);
2765        let target_orig = payload(SMALL, 72);
2766        let victim = insert(&pool, &mut victim_orig.clone());
2767        let target = insert(&pool, &mut target_orig.clone());
2768        pool.evict(&target);
2769        assert!(pool.back_step(), "victim is backable");
2770        assert_eq!(victim.residency(), Residency::BackedResident);
2771        // The budget now holds exactly the victim: no admission headroom.
2772        pool.set_budget(64 << 10);
2773        assert_eq!(victim.residency(), Residency::BackedResident);
2774        let before = pool.stats();
2775
2776        assert_eq!(read_admit(&target), target_orig);
2777        assert_eq!(target.residency(), Residency::BackedResident);
2778        assert_eq!(victim.residency(), Residency::Evicted);
2779        let after = pool.stats();
2780        assert_eq!(after.admissions_steal, 1);
2781        assert_eq!(after.admissions_budget, 0);
2782        assert_eq!(after.admissions_denied, 0);
2783        assert_eq!(
2784            after.resident_bytes, before.resident_bytes,
2785            "same class, same bytes",
2786        );
2787        assert_eq!(
2788            after.evictions_compress, before.evictions_compress,
2789            "no compression",
2790        );
2791        assert_eq!(
2792            after.evictions_cheap, before.evictions_cheap,
2793            "a steal is not an enforcement eviction",
2794        );
2795        assert_eq!(after.extent_bytes_written, before.extent_bytes_written);
2796        assert_eq!(after.extent_pageouts, 0, "no pageout");
2797        assert_eq!(
2798            after.warm_bytes, before.warm_bytes,
2799            "the stolen slot skipped the free list",
2800        );
2801        assert_eq!(after.warm_reuses, before.warm_reuses);
2802
2803        // The victim's extent is intact: its old slot now holds the
2804        // admitted chunk's bytes, so a correct read must come from the
2805        // extent.
2806        assert_eq!(read(&victim), victim_orig);
2807        assert_eq!(victim.residency(), Residency::Evicted);
2808
2809        drop(victim);
2810        drop(target);
2811        let stats = pool.stats();
2812        assert_eq!(stats.resident_bytes, 0);
2813        assert_eq!(stats.extent_resident_bytes, 0);
2814    }
2815
2816    /// With the budget full and every candidate touched, the admitting read
2817    /// still returns correct data, the chunk stays evicted, and the denial
2818    /// counter increments.
2819    #[mz_ore::test]
2820    fn admit_denied_when_victims_touched() {
2821        let pool = test_pool(256 << 20);
2822        let victim_orig = payload(SMALL, 73);
2823        let target_orig = payload(SMALL, 74);
2824        let victim = insert(&pool, &mut victim_orig.clone());
2825        let target = insert(&pool, &mut target_orig.clone());
2826        pool.evict(&target);
2827        assert!(pool.back_step());
2828        // Reading the victim sets its second-chance bit, disqualifying it.
2829        assert_eq!(read(&victim), victim_orig);
2830        pool.set_budget(64 << 10);
2831        let resident = pool.stats().resident_bytes;
2832
2833        assert_eq!(read_admit(&target), target_orig);
2834        assert_eq!(target.residency(), Residency::Evicted);
2835        assert_eq!(victim.residency(), Residency::BackedResident);
2836        let stats = pool.stats();
2837        assert_eq!(stats.admissions_denied, 1);
2838        assert_eq!(stats.admissions_budget, 0);
2839        assert_eq!(stats.admissions_steal, 0);
2840        assert_eq!(stats.resident_bytes, resident);
2841    }
2842
2843    /// An unbacked resident candidate is never stolen from: evicting it
2844    /// would require the compression that admission forbids.
2845    #[mz_ore::test]
2846    fn admit_denied_when_victims_unbacked() {
2847        let pool = test_pool(256 << 20);
2848        let victim = insert(&pool, &mut payload(SMALL, 75));
2849        let target_orig = payload(SMALL, 76);
2850        let target = insert(&pool, &mut target_orig.clone());
2851        pool.evict(&target);
2852        pool.set_budget(64 << 10);
2853        assert_eq!(read_admit(&target), target_orig);
2854        assert_eq!(target.residency(), Residency::Evicted);
2855        assert_eq!(victim.residency(), Residency::UnbackedResident);
2856        assert_eq!(pool.stats().admissions_denied, 1);
2857    }
2858
2859    /// A clean backed victim of a different size class is never stolen
2860    /// from: slot reuse in place requires the classes to match.
2861    #[mz_ore::test]
2862    fn admit_denied_when_victims_wrong_class() {
2863        let pool = test_pool(256 << 20);
2864        // The victim fills the 128 KiB class; the target lives in the
2865        // 64 KiB one.
2866        let victim = insert(&pool, &mut payload(2 * SMALL, 77));
2867        let target_orig = payload(SMALL, 78);
2868        let target = insert(&pool, &mut target_orig.clone());
2869        pool.evict(&target);
2870        assert!(pool.back_step());
2871        assert_eq!(victim.residency(), Residency::BackedResident);
2872        // The budget holds exactly the victim: no headroom for the target.
2873        pool.set_budget(128 << 10);
2874        assert_eq!(read_admit(&target), target_orig);
2875        assert_eq!(target.residency(), Residency::Evicted);
2876        assert_eq!(victim.residency(), Residency::BackedResident);
2877        assert_eq!(pool.stats().admissions_denied, 1);
2878    }
2879
2880    /// Plain reads and `take` never admit: an evicted chunk with plenty of
2881    /// budget headroom stays evicted through both, and neither counts a
2882    /// denial.
2883    #[mz_ore::test]
2884    fn plain_read_and_take_never_admit() {
2885        let pool = test_pool(256 << 20);
2886        let orig = payload(SMALL, 79);
2887        let handle = insert(&pool, &mut orig.clone());
2888        pool.evict(&handle);
2889        assert_eq!(read(&handle), orig);
2890        assert_eq!(handle.residency(), Residency::Evicted);
2891        assert_eq!(pool.stats().resident_bytes, 0);
2892        let mut out = Vec::new();
2893        handle.take(&mut out);
2894        assert_eq!(out, orig);
2895        let stats = pool.stats();
2896        assert_eq!(stats.admissions_budget, 0);
2897        assert_eq!(stats.admissions_steal, 0);
2898        assert_eq!(stats.admissions_denied, 0);
2899        assert_eq!(stats.resident_bytes, 0);
2900        assert_eq!(stats.frees, 1);
2901    }
2902
2903    /// A steal settles the ledger with the payload difference: a shrinking
2904    /// steal always proceeds, while a steal that would grow resident bytes
2905    /// past the budget is denied and leaves the victim untouched.
2906    #[mz_ore::test]
2907    fn steal_admission_charges_the_budget() {
2908        // Shrinking steal: the victim is larger than the admitted payload,
2909        // so the steal lowers resident bytes and always may proceed.
2910        let pool = test_pool(256 << 20);
2911        let victim_orig = payload(SMALL, 84);
2912        let victim = insert(&pool, &mut victim_orig.clone());
2913        assert!(pool.back_step(), "victim backs");
2914        let small_orig = payload(SMALL / 2, 85);
2915        let handle = insert(&pool, &mut small_orig.clone());
2916        pool.evict(&handle);
2917        pool.set_budget(64 << 10);
2918        assert_eq!(read_admit(&handle), small_orig);
2919        let stats = pool.stats();
2920        assert_eq!(stats.admissions_steal, 1, "no headroom, so the read steals");
2921        assert_eq!(stats.resident_bytes, u64::cast_from(SMALL / 2 * 8));
2922        assert_eq!(victim.residency(), Residency::Evicted);
2923        assert_eq!(read(&victim), victim_orig, "victim serves from its extent");
2924
2925        // Growing steal: the admitted payload is larger than the only
2926        // victim, and the growth does not fit the budget, so the admission
2927        // is denied and the victim is left untouched.
2928        let pool = test_pool(256 << 20);
2929        let big_orig = payload(SMALL, 86);
2930        let big = insert(&pool, &mut big_orig.clone());
2931        pool.evict(&big);
2932        let small_victim = insert(&pool, &mut payload(SMALL / 2, 87));
2933        assert!(pool.back_step(), "victim backs");
2934        pool.set_budget(32 << 10);
2935        assert_eq!(read_admit(&big), big_orig);
2936        let stats = pool.stats();
2937        assert_eq!(stats.admissions_denied, 1, "growth exceeds the budget");
2938        assert_eq!(stats.admissions_steal, 0);
2939        assert_eq!(big.residency(), Residency::Evicted);
2940        assert_eq!(small_victim.residency(), Residency::BackedResident);
2941    }
2942
2943    /// Admission of a chunk whose extent was pushed to the device: the read
2944    /// revives the extent into the acquired slot and re-counts it.
2945    #[mz_ore::test]
2946    fn admission_revives_paged_out_extent() {
2947        let pool = test_pool(256 << 20);
2948        let orig = payload(SMALL, 88);
2949        let handle = insert(&pool, &mut orig.clone());
2950        pool.evict(&handle);
2951        assert_eq!(
2952            pool.stats().extent_resident_bytes,
2953            0,
2954            "zero RSS target pages the extent out on eviction",
2955        );
2956        // Raise the target so the read's own tier enforcement does not
2957        // page the revived extent straight back out.
2958        pool.set_rss_target(1 << 30);
2959        assert_eq!(read_admit(&handle), orig);
2960        assert_eq!(handle.residency(), Residency::BackedResident);
2961        let stats = pool.stats();
2962        assert_eq!(stats.admissions_budget, 1);
2963        assert!(stats.extent_resident_bytes > 0, "revived and re-counted");
2964    }
2965
2966    /// An admitting read of a chunk that is not evicted is a plain read:
2967    /// no admission counter moves and no state changes.
2968    #[mz_ore::test]
2969    fn admit_is_plain_read_on_non_evicted_chunks() {
2970        let pool = test_pool(256 << 20);
2971        let orig = payload(SMALL, 89);
2972        let resident = insert(&pool, &mut orig.clone());
2973        assert_eq!(read_admit(&resident), orig);
2974        assert_eq!(resident.residency(), Residency::UnbackedResident);
2975        let words = SIZE_CLASSES[SIZE_CLASSES.len() - 1] / 8 + 1;
2976        let oversize_orig = payload(words, 90);
2977        let oversize = insert(&pool, &mut oversize_orig.clone());
2978        assert_eq!(read_admit(&oversize), oversize_orig);
2979        let empty = insert(&pool, &mut Vec::new());
2980        assert!(read_admit(&empty).is_empty());
2981        let stats = pool.stats();
2982        assert_eq!(stats.admissions_budget, 0);
2983        assert_eq!(stats.admissions_steal, 0);
2984        assert_eq!(stats.admissions_denied, 0);
2985    }
2986
2987    /// Concurrent admitting reads with no budget headroom: every admission
2988    /// must go through the steal path, racing steals against each other on
2989    /// the same victims (the pool's only two-chunk lock edge). Contents are
2990    /// asserted on every read.
2991    #[mz_ore::test]
2992    #[cfg_attr(miri, ignore)] // too slow
2993    fn concurrent_admits_exercise_the_steal_path() {
2994        const CHUNKS: u64 = 8;
2995        let pool = test_pool(usize::MAX);
2996        let origs: Vec<_> = (0..CHUNKS).map(|seed| payload(SMALL, 900 + seed)).collect();
2997        let evicted: Arc<Vec<(Vec<u64>, ChunkHandle)>> = Arc::new(
2998            origs
2999                .iter()
3000                .map(|orig| {
3001                    let handle = insert(&pool, &mut orig.clone());
3002                    pool.evict(&handle);
3003                    (orig.clone(), handle)
3004                })
3005                .collect(),
3006        );
3007        let mut victims = Vec::new();
3008        for seed in 0..CHUNKS {
3009            victims.push(insert(&pool, &mut payload(SMALL, 950 + seed)));
3010            assert!(pool.back_step(), "victim backs");
3011        }
3012        // Exactly the victims' bytes: no free headroom, so every admission
3013        // steals or is denied.
3014        pool.set_budget(usize::cast_from(CHUNKS) * (64 << 10));
3015        let threads: Vec<_> = (0..2u64)
3016            .map(|t| {
3017                let evicted = Arc::clone(&evicted);
3018                std::thread::spawn(move || {
3019                    for round in 0..CHUNKS {
3020                        let (orig, handle) = &evicted[usize::cast_from((t + round) % CHUNKS)];
3021                        let mut out = Vec::new();
3022                        handle.read_into_admit(&mut out);
3023                        assert_eq!(&out, orig);
3024                    }
3025                })
3026            })
3027            .collect();
3028        for thread in threads {
3029            thread.join().expect("admitting thread panicked");
3030        }
3031        let stats = pool.stats();
3032        assert!(stats.admissions_steal > 0, "no headroom forces steals");
3033        assert!(
3034            stats.resident_bytes <= u64::cast_from(usize::cast_from(CHUNKS) * (64 << 10)),
3035            "steals never grow resident bytes past the budget",
3036        );
3037        // The victims were held live as steal targets; a steal leaves its
3038        // victim evicted, so at least one is evicted here.
3039        let stolen = victims
3040            .iter()
3041            .filter(|v| v.residency() == Residency::Evicted)
3042            .count();
3043        assert!(stolen > 0, "a steal evicts its victim");
3044    }
3045
3046    /// A re-admitted chunk keeps its insert-time depth: under budget
3047    /// pressure it is evicted from its own deeper band before a younger
3048    /// band-0 chunk, which a re-admission into band 0 would have inverted.
3049    #[mz_ore::test]
3050    fn admitted_chunk_keeps_its_depth() {
3051        let pool = test_pool(256 << 20);
3052        let deep_orig = payload(SMALL, 80);
3053        let deep = insert_at_depth(&pool, 2, &mut deep_orig.clone());
3054        let young = insert(&pool, &mut payload(SMALL, 81));
3055        pool.evict(&deep);
3056        assert_eq!(read_admit(&deep), deep_orig);
3057        assert_eq!(deep.residency(), Residency::BackedResident);
3058        assert_eq!(pool.stats().admissions_budget, 1);
3059
3060        // Budget of one chunk: enforcement visits the deep band first.
3061        pool.set_budget(64 << 10);
3062        assert_eq!(deep.residency(), Residency::Evicted);
3063        assert_eq!(young.residency(), Residency::UnbackedResident);
3064        assert_eq!(
3065            pool.stats().evictions_cheap,
3066            1,
3067            "the extent kept through admission pre-paid the eviction",
3068        );
3069    }
3070
3071    /// Admitting reads racing enforcement, opposing steals, and frees:
3072    /// contents stay correct, contended steals degrade to skips, and the
3073    /// accounting identity settles to zero.
3074    #[mz_ore::test]
3075    #[cfg_attr(miri, ignore)] // too slow
3076    fn concurrent_admits_race_cleanly() {
3077        let pool = test_pool(64 << 10);
3078        let per_thread = rounds(50, 3);
3079        let threads: Vec<_> = (0..4u64)
3080            .map(|t| {
3081                let pool = pool.clone();
3082                std::thread::spawn(move || {
3083                    let mut out = Vec::new();
3084                    for round in 0..per_thread {
3085                        let orig = payload(SMALL, t * 1000 + round);
3086                        let handle = insert(&pool, &mut orig.clone());
3087                        pool.evict(&handle);
3088                        handle.read_into_admit(&mut out);
3089                        assert_eq!(out, orig);
3090                        handle.read_into_admit(&mut out);
3091                        assert_eq!(out, orig);
3092                        assert_eq!(read(&handle), orig);
3093                    }
3094                })
3095            })
3096            .collect();
3097        for thread in threads {
3098            thread.join().expect("worker thread panicked");
3099        }
3100        let stats = pool.stats();
3101        assert_eq!(stats.inserts, 4 * per_thread);
3102        assert_eq!(stats.frees, 4 * per_thread);
3103        assert_eq!(stats.resident_bytes, 0);
3104        assert_eq!(stats.extent_resident_bytes, 0);
3105    }
3106
3107    /// Slots are scoped to residency: eviction releases the slot, so a
3108    /// capacity holding exactly one chunk can serve any number of chunks one
3109    /// at a time, and reads of evicted chunks need no slot at all.
3110    #[mz_ore::test]
3111    fn eviction_releases_the_slot() {
3112        // One 64 KiB slot per class.
3113        let pool = Pool::with_class_capacity(64 << 10).expect("pool creation");
3114        let a = insert(&pool, &mut payload(SMALL, 6));
3115        pool.evict(&a);
3116        // The class's only slot is free again: a second chunk fits without
3117        // falling back to the heap.
3118        let b = insert(&pool, &mut payload(SMALL, 7));
3119        assert_eq!(b.residency(), Residency::UnbackedResident);
3120        assert_eq!(pool.stats().slot_exhausted_fallbacks, 0);
3121        // Reading `a` decompresses straight from its extent while `b` holds
3122        // the class's only slot: copy-out allocates nothing.
3123        assert_eq!(read(&a), payload(SMALL, 6));
3124        assert_eq!(a.residency(), Residency::Evicted);
3125        assert_eq!(read(&b), payload(SMALL, 7));
3126    }
3127
3128    /// The eviction queue holds resident chunks only: an enforcement pass
3129    /// drops entries for evicted chunks, and reads never re-add them, so the
3130    /// scan each insert pays stays proportional to the resident set rather
3131    /// than every chunk ever evicted.
3132    #[mz_ore::test]
3133    fn queue_holds_resident_chunks_only() {
3134        let pool = test_pool(128 << 10);
3135        let mut handles = Vec::new();
3136        for seed in 0..8 {
3137            handles.push(insert(&pool, &mut payload(SMALL, 800 + seed)));
3138        }
3139        // Budget pressure evicted ~6 of 8; one more pass visits the evicted
3140        // entries and drops them (their first visit performed the eviction
3141        // and dropped them already, but second-chance survivors may linger).
3142        pool.enforce_budget();
3143        let resident = handles
3144            .iter()
3145            .filter(|h| h.residency() != Residency::Evicted)
3146            .count();
3147        assert!(
3148            pool.queue_len() <= resident + 1,
3149            "queue ({}) tracks the resident set ({resident}), not all 8 live chunks",
3150            pool.queue_len(),
3151        );
3152        // Reading an evicted chunk copies out of its extent and does not
3153        // re-enqueue it: the queue keeps tracking the resident set.
3154        let evicted = handles
3155            .iter()
3156            .find(|h| h.residency() == Residency::Evicted)
3157            .expect("something was evicted");
3158        let before = pool.queue_len();
3159        assert_eq!(read(evicted).len(), SMALL);
3160        assert_eq!(evicted.residency(), Residency::Evicted);
3161        assert_eq!(pool.queue_len(), before, "reads leave the queue alone");
3162    }
3163
3164    #[mz_ore::test]
3165    fn dead_data_is_never_written() {
3166        let pool = test_pool(256 << 20);
3167        let handle = insert(&pool, &mut payload(SMALL, 7));
3168        drop(handle);
3169        let stats = pool.stats();
3170        assert_eq!(stats.frees, 1);
3171        assert_eq!(stats.writes_elided, 1);
3172        assert_eq!(stats.extent_bytes_written, 0);
3173        assert_eq!(stats.resident_bytes, 0);
3174    }
3175
3176    #[mz_ore::test]
3177    fn budget_is_enforced_on_insert() {
3178        let budget = 128 << 10;
3179        let pool = test_pool(budget);
3180        let mut handles = Vec::new();
3181        for seed in 0..8 {
3182            handles.push(insert(&pool, &mut payload(SMALL, 100 + seed)));
3183        }
3184        let stats = pool.stats();
3185        assert!(
3186            stats.resident_bytes <= u64::cast_from(budget),
3187            "resident {} exceeds budget {}",
3188            stats.resident_bytes,
3189            budget,
3190        );
3191        assert!(stats.evictions_compress >= 6);
3192        let resident = handles
3193            .iter()
3194            .filter(|h| {
3195                matches!(
3196                    h.residency(),
3197                    Residency::UnbackedResident | Residency::BackedResident
3198                )
3199            })
3200            .count();
3201        assert_eq!(resident, 2, "budget holds exactly two small chunks");
3202    }
3203
3204    /// Budget enforcement is single-flight: an insert that trips it while a
3205    /// pass holds the `enforcing` guard bails on `WouldBlock`, trusting that
3206    /// pass. If the holder is already past its final `resident_bytes` read, the
3207    /// bailed insert's bytes are neither read by the holder nor enforced by the
3208    /// bailer, and no later insert re-trips enforcement, so the pool stays over
3209    /// budget. The fix re-runs the pass while any caller was turned away.
3210    ///
3211    /// The test hook freezes the holder's pass in that window to make the race
3212    /// deterministic: the holder parks having found the budget satisfied, the
3213    /// main thread inserts over budget and is turned away, then the holder
3214    /// resumes. The `gate` is used for both rendezvous.
3215    #[mz_ore::test]
3216    fn racing_insert_is_not_dropped_by_budget_single_flight() {
3217        // Budget for exactly one small chunk.
3218        let budget = 64 << 10;
3219        let pool = test_pool(budget);
3220        let gate = std::sync::Arc::new(std::sync::Barrier::new(2));
3221
3222        let holder = {
3223            let pool = pool.clone();
3224            let gate = std::sync::Arc::clone(&gate);
3225            std::thread::spawn(move || -> ChunkHandle {
3226                ENFORCE_BUDGET_HOOK.with(|cell| {
3227                    *cell.borrow_mut() = Some(Box::new(move || {
3228                        gate.wait(); // parked, holding the guard
3229                        gate.wait(); // resume once the race is done
3230                    }));
3231                });
3232                // At budget: the pass finds it satisfied and parks at the hook.
3233                insert(&pool, &mut payload(SMALL, 1))
3234            })
3235        };
3236
3237        gate.wait(); // holder is parked in enforcement, holding the guard
3238        // Push over budget; this insert is turned away by the held guard.
3239        let _over = insert(&pool, &mut payload(SMALL, 2));
3240        gate.wait(); // let the holder resume and release the guard
3241        // Kept alive past the assert: freeing it would drop its bytes and mask
3242        // the overshoot.
3243        let _held = holder.join().expect("holder panicked");
3244
3245        // Nothing re-trips enforcement, so the pool must not be left over budget.
3246        let resident = pool.stats().resident_bytes;
3247        assert!(
3248            resident <= u64::cast_from(budget),
3249            "resident {resident} exceeds budget {budget}: racing insert escaped enforcement",
3250        );
3251    }
3252
3253    #[mz_ore::test]
3254    fn set_budget_retunes_in_place() {
3255        let pool = test_pool(usize::MAX);
3256        let mut handles = Vec::new();
3257        for seed in 0..8 {
3258            handles.push(insert(&pool, &mut payload(SMALL, 200 + seed)));
3259        }
3260        assert_eq!(pool.stats().evictions_compress, 0);
3261
3262        // Shrinking the budget evicts immediately.
3263        pool.set_budget(128 << 10);
3264        let stats = pool.stats();
3265        assert!(stats.resident_bytes <= 128 << 10);
3266        assert!(stats.evictions_compress >= 6);
3267
3268        // Growing it leaves headroom: a fresh insert stays resident.
3269        pool.set_budget(usize::MAX);
3270        let h = insert(&pool, &mut payload(SMALL, 300));
3271        assert_eq!(h.residency(), Residency::UnbackedResident);
3272        for h in &handles {
3273            assert_eq!(read(h).len(), SMALL);
3274        }
3275    }
3276
3277    #[mz_ore::test]
3278    fn second_chance_prefers_untouched_victims() {
3279        // Budget holds one and a half small chunks.
3280        let pool = test_pool((64 << 10) + (32 << 10));
3281        let orig_a = payload(SMALL, 8);
3282        let handle_a = insert(&pool, &mut orig_a.clone());
3283        assert_eq!(read(&handle_a), orig_a);
3284        // Inserting B overflows the budget; A is older but touched, so the
3285        // enforcer gives it a second chance and evicts untouched B instead.
3286        let handle_b = insert(&pool, &mut payload(SMALL, 9));
3287        assert_eq!(handle_a.residency(), Residency::UnbackedResident);
3288        assert_eq!(handle_b.residency(), Residency::Evicted);
3289    }
3290
3291    /// Depth-hinted chunks are evicted before younger ones: the deep chunk
3292    /// loses even though the young chunk is older and both are untouched
3293    /// (plain FIFO would have evicted the older, young one). Also exercises
3294    /// band clamping: depths beyond the last band share it.
3295    #[mz_ore::test]
3296    fn eviction_prefers_deeper_chunks() {
3297        // Budget of one small chunk.
3298        let pool = test_pool(64 << 10);
3299        let young = insert(&pool, &mut payload(SMALL, 900));
3300        let deep = insert_at_depth(&pool, 255, &mut payload(SMALL, 901));
3301        assert_eq!(young.residency(), Residency::UnbackedResident);
3302        assert_eq!(deep.residency(), Residency::Evicted);
3303    }
3304
3305    /// Eager backing visits deeper chunks first, mirroring eviction order,
3306    /// so the chunks evicted first are the ones already backed.
3307    #[mz_ore::test]
3308    fn backing_prefers_deeper_chunks() {
3309        let pool = test_pool(256 << 20);
3310        let young = insert(&pool, &mut payload(SMALL, 902));
3311        let deep = insert_at_depth(&pool, 2, &mut payload(SMALL, 903));
3312        assert!(pool.back_step());
3313        assert_eq!(deep.residency(), Residency::BackedResident);
3314        assert_eq!(young.residency(), Residency::UnbackedResident);
3315        assert!(pool.back_step());
3316        assert_eq!(young.residency(), Residency::BackedResident);
3317    }
3318
3319    #[mz_ore::test]
3320    fn empty_insert_consumes_no_slot() {
3321        let pool = test_pool(256 << 20);
3322        let mut data = Vec::new();
3323        let handle = insert(&pool, &mut data);
3324        assert_eq!(handle.size_class_bytes(), None);
3325        assert!(read(&handle).is_empty());
3326        // Reads clear the destination even for empty chunks.
3327        let mut out = vec![1u64, 2, 3];
3328        handle.read_into(&mut out);
3329        assert!(out.is_empty());
3330        drop(handle);
3331        let stats = pool.stats();
3332        assert_eq!(stats.resident_bytes, 0);
3333        assert_eq!(stats.writes_elided, 0);
3334    }
3335
3336    #[mz_ore::test]
3337    fn oversize_round_trips() {
3338        let pool = test_pool(256 << 20);
3339        let words = SIZE_CLASSES[SIZE_CLASSES.len() - 1] / 8 + 1;
3340        let orig = payload(words, 10);
3341        let handle = insert(&pool, &mut orig.clone());
3342        assert_eq!(handle.residency(), Residency::Oversize);
3343        assert_eq!(handle.size_class_bytes(), None);
3344        let stats = pool.stats();
3345        assert_eq!(stats.oversize_bytes, u64::cast_from(words * 8));
3346        // The payload outgrew the largest class, and no class was exhausted.
3347        assert_eq!(stats.oversize_payloads, 1);
3348        assert_eq!(stats.slot_exhausted_fallbacks, 0);
3349        // Explicit eviction and budget enforcement leave oversize chunks
3350        // resident.
3351        pool.evict(&handle);
3352        pool.enforce_budget();
3353        assert_eq!(handle.residency(), Residency::Oversize);
3354        assert_eq!(read(&handle), orig);
3355        drop(handle);
3356        let stats = pool.stats();
3357        assert_eq!(stats.oversize_bytes, 0);
3358        assert_eq!(stats.resident_bytes, 0);
3359    }
3360
3361    #[mz_ore::test]
3362    fn payload_lands_in_smallest_fitting_class() {
3363        let pool = test_pool(256 << 20);
3364        let handle = insert(&pool, &mut payload((100 << 10) / 8, 11));
3365        assert_eq!(handle.size_class_bytes(), Some(128 << 10));
3366        let exact = insert(&pool, &mut payload(SMALL, 12));
3367        assert_eq!(exact.size_class_bytes(), Some(64 << 10));
3368    }
3369
3370    #[mz_ore::test]
3371    #[cfg_attr(miri, ignore)] // too slow
3372    fn multithreaded_smoke() {
3373        // Budget of one small chunk: four inserting threads keep the pool
3374        // over budget, so every insert's enforcement pass selects victims
3375        // owned by other threads, racing cross-thread eviction against
3376        // copy-out reads and frees.
3377        let pool = test_pool(64 << 10);
3378        let per_thread = rounds(50, 3);
3379        let threads: Vec<_> = (0..4u64)
3380            .map(|t| {
3381                let pool = pool.clone();
3382                std::thread::spawn(move || {
3383                    for round in 0..per_thread {
3384                        let seed = t * 1000 + round;
3385                        let orig = payload(SMALL, seed);
3386                        let handle = insert(&pool, &mut orig.clone());
3387                        pool.evict(&handle);
3388                        assert_eq!(read(&handle), orig);
3389                        // Enforcement racing reads must never corrupt them.
3390                        pool.enforce_budget();
3391                        assert_eq!(read(&handle), orig);
3392                        drop(handle);
3393                    }
3394                })
3395            })
3396            .collect();
3397        for thread in threads {
3398            thread.join().expect("worker thread panicked");
3399        }
3400        let stats = pool.stats();
3401        assert_eq!(stats.inserts, 4 * per_thread);
3402        assert_eq!(stats.frees, 4 * per_thread);
3403        assert_eq!(stats.resident_bytes, 0);
3404    }
3405
3406    #[mz_ore::test]
3407    #[cfg_attr(miri, ignore)] // too slow
3408    fn concurrent_read_enforce_churn() {
3409        // Races the three actors that can touch one chunk's slot: readers
3410        // copying shared chunks out and verifying them, an enforcer evicting
3411        // them (the zero budget makes every chunk a victim), and a churner
3412        // whose insert/free traffic turns the queue over. Contents are
3413        // asserted on every read, so an eviction or slot recycle racing a
3414        // copy-out shows up as corruption.
3415        let pool = test_pool(0);
3416        let shared: Arc<Vec<(Vec<u64>, ChunkHandle)>> = Arc::new(
3417            (0..4u64)
3418                .map(|seed| {
3419                    let orig = payload(SMALL, 600 + seed);
3420                    let handle = insert(&pool, &mut orig.clone());
3421                    (orig, handle)
3422                })
3423                .collect(),
3424        );
3425        let churn = rounds(300, 6);
3426        let mut threads = Vec::new();
3427        for t in 0..2u64 {
3428            let shared = Arc::clone(&shared);
3429            threads.push(std::thread::spawn(move || {
3430                for round in 0..churn {
3431                    let (orig, handle) = &shared[usize::cast_from((t + round) % 4)];
3432                    assert_eq!(&read(handle), orig);
3433                }
3434            }));
3435        }
3436        {
3437            let pool = pool.clone();
3438            threads.push(std::thread::spawn(move || {
3439                for _ in 0..2 * churn {
3440                    pool.enforce_budget();
3441                }
3442            }));
3443        }
3444        {
3445            let pool = pool.clone();
3446            threads.push(std::thread::spawn(move || {
3447                for round in 0..churn {
3448                    let orig = payload(SMALL, 700 + round);
3449                    let handle = insert(&pool, &mut orig.clone());
3450                    assert_eq!(read(&handle), orig);
3451                }
3452            }));
3453        }
3454        for thread in threads {
3455            thread.join().expect("worker thread panicked");
3456        }
3457        drop(shared);
3458        assert_eq!(pool.stats().resident_bytes, 0);
3459    }
3460
3461    /// Read-only traffic never raises resident bytes: every chunk starts
3462    /// evicted and is then read once, with no inserts in between. Reads copy
3463    /// out of the extents and leave every chunk evicted, so a seek-heavy
3464    /// phase costs no pool memory at all.
3465    #[mz_ore::test]
3466    fn reads_never_raise_resident_bytes() {
3467        let pool = test_pool(128 << 10);
3468        let origs: Vec<_> = (0..8u64).map(|seed| payload(SMALL, 300 + seed)).collect();
3469        let handles: Vec<_> = origs
3470            .iter()
3471            .map(|o| insert(&pool, &mut o.clone()))
3472            .collect();
3473        for handle in &handles {
3474            pool.evict(handle);
3475        }
3476        assert_eq!(pool.stats().resident_bytes, 0);
3477        for (index, handle) in handles.iter().enumerate() {
3478            assert_eq!(read(handle), origs[index]);
3479            assert_eq!(handle.residency(), Residency::Evicted);
3480            assert_eq!(pool.stats().resident_bytes, 0);
3481        }
3482    }
3483
3484    /// Evict-then-free churn under a generous RSS target: the compressed
3485    /// tier never crosses its cap, so enforcement never visits (and never
3486    /// drops) extent-queue entries, and pruning alone must keep the queue
3487    /// proportional to the live resident extents.
3488    #[mz_ore::test]
3489    #[cfg_attr(miri, ignore)] // too slow
3490    fn extent_queue_stays_bounded_under_cap() {
3491        let pool = test_pool(256 << 20);
3492        pool.set_rss_target(1 << 40);
3493        for seed in 0..rounds(1000, 48) {
3494            let handle = insert(&pool, &mut payload(SMALL, seed));
3495            pool.evict(&handle);
3496            drop(handle);
3497        }
3498        assert_eq!(pool.stats().extent_resident_bytes, 0);
3499        let len = pool.extent_queue_len();
3500        assert!(
3501            len <= 32,
3502            "extent queue holds {len} entries for zero resident extents",
3503        );
3504    }
3505
3506    /// A warm slot reused for a smaller payload round-trips: the tail
3507    /// release past the new payload must not disturb the payload itself,
3508    /// and the ledger credits exactly the payload.
3509    #[mz_ore::test]
3510    fn warm_reuse_with_smaller_payload_round_trips() {
3511        // Budget 8 MiB: warm cap = 1 MiB, so a 64 KiB slot parks warm.
3512        let pool = test_pool(8 << 20);
3513        let full = insert(&pool, &mut payload(SMALL, 60));
3514        drop(full);
3515        assert_eq!(pool.stats().warm_bytes, 64 << 10, "freed slot parks warm");
3516        // A payload of just over a page reuses the warm slot; the slot's
3517        // pages past it are released.
3518        let words = 4096 / 8 + 1;
3519        let orig = payload(words, 61);
3520        let handle = insert(&pool, &mut orig.clone());
3521        let stats = pool.stats();
3522        assert_eq!(stats.warm_reuses, 1, "reused the warm slot");
3523        assert_eq!(stats.resident_bytes, u64::cast_from(words * 8));
3524        assert_eq!(read(&handle), orig);
3525        // Round-trips through the extent as well.
3526        pool.evict(&handle);
3527        pool.poison_free_slots();
3528        assert_eq!(read(&handle), orig);
3529        drop(handle);
3530        assert_eq!(pool.stats().resident_bytes, 0);
3531    }
3532
3533    /// Heap-backed chunks count as resident but can never be evicted, so
3534    /// the budget must not force slotted chunks out on their account: with
3535    /// unevictable bytes alone exceeding the budget, a slotted chunk that
3536    /// fits the budget stays resident.
3537    #[mz_ore::test]
3538    fn unevictable_bytes_do_not_force_eviction() {
3539        // One 64 KiB slot per class: the second and third inserts fall
3540        // back to the heap.
3541        let pool = Pool::with_class_capacity(64 << 10).expect("pool creation");
3542        pool.set_budget(64 << 10);
3543        let slotted = insert(&pool, &mut payload(SMALL, 91));
3544        let heap_a = insert(&pool, &mut payload(SMALL, 92));
3545        let heap_b = insert(&pool, &mut payload(SMALL, 93));
3546        assert_eq!(heap_a.residency(), Residency::Oversize);
3547        assert_eq!(heap_b.residency(), Residency::Oversize);
3548        let stats = pool.stats();
3549        assert!(stats.oversize_bytes > 64 << 10, "unevictable exceed budget");
3550        assert_eq!(slotted.residency(), Residency::UnbackedResident);
3551        assert_eq!(stats.evictions_compress, 0);
3552        assert_eq!(read(&slotted), payload(SMALL, 91));
3553        assert_eq!(read(&heap_a), payload(SMALL, 92));
3554    }
3555
3556    /// A slotless empty chunk survives an explicit evict with spill
3557    /// scheduling enabled: nothing is handed to the spill threads and the
3558    /// chunk stays readable.
3559    #[mz_ore::test]
3560    fn evict_of_empty_chunk_is_a_no_op() {
3561        let pool = test_pool(usize::MAX);
3562        pool.enable_spill_without_threads();
3563        let empty = insert(&pool, &mut Vec::new());
3564        pool.evict(&empty);
3565        assert_eq!(empty.residency(), Residency::UnbackedResident);
3566        assert!(!pool.spill_step(), "nothing was scheduled");
3567        assert_eq!(pool.stats().spill_scheduled, 0);
3568        assert!(read(&empty).is_empty());
3569    }
3570
3571    #[mz_ore::test]
3572    fn queue_stays_bounded_under_budget() {
3573        // Chunk churn that never exceeds the budget: the enforcer's eviction
3574        // loop never runs, so stale queue entries must be reclaimed by
3575        // pruning alone.
3576        let pool = test_pool(256 << 20);
3577        for seed in 0..rounds(1000, 48) {
3578            let handle = insert(&pool, &mut payload(SMALL, seed));
3579            drop(handle);
3580        }
3581        let len = pool.queue_len();
3582        assert!(len <= 32, "queue holds {len} entries for zero live chunks");
3583    }
3584
3585    #[mz_ore::test]
3586    fn spill_async_evict_round_trip() {
3587        let pool = test_pool(usize::MAX);
3588        pool.enable_spill_without_threads();
3589        let h = insert(&pool, &mut payload(SMALL, 400));
3590        pool.evict(&h);
3591        assert_eq!(h.residency(), Residency::WriteInFlight);
3592        // Readable while in flight: the slot is still populated, and the
3593        // copy-out coexists with the spill thread's compression read.
3594        assert_eq!(read(&h), payload(SMALL, 400));
3595        // Reads leave no trace, so the eviction commits.
3596        assert!(pool.spill_step());
3597        assert_eq!(h.residency(), Residency::Evicted);
3598        let stats = pool.stats();
3599        assert_eq!(stats.spill_scheduled, 1);
3600        assert_eq!(stats.evictions_compress, 1);
3601        pool.poison_free_slots();
3602        assert_eq!(read(&h), payload(SMALL, 400));
3603    }
3604
3605    #[mz_ore::test]
3606    fn spill_freed_while_queued_is_elided() {
3607        let pool = test_pool(usize::MAX);
3608        pool.enable_spill_without_threads();
3609        let h = insert(&pool, &mut payload(SMALL, 401));
3610        pool.evict(&h);
3611        assert_eq!(h.residency(), Residency::WriteInFlight);
3612        drop(h);
3613        assert!(pool.spill_step());
3614        let stats = pool.stats();
3615        assert_eq!(stats.spill_cancelled, 1);
3616        assert_eq!(stats.writes_elided, 1, "freed before compression: elided");
3617        assert_eq!(stats.extent_bytes_written, 0, "no extent was written");
3618        assert_eq!(stats.resident_bytes, 0, "slot accounting settled");
3619    }
3620
3621    /// `take` on a chunk whose backing write is still in flight copies the
3622    /// contents out of the slot and cancels the write: the spill thread finds
3623    /// the chunk freed, elides the extent, and settles the slot accounting.
3624    #[mz_ore::test]
3625    fn spill_take_in_flight_cancels_write() {
3626        let pool = test_pool(usize::MAX);
3627        pool.enable_spill_without_threads();
3628        let orig = payload(SMALL, 402);
3629        let h = insert(&pool, &mut orig.clone());
3630        pool.evict(&h);
3631        assert_eq!(h.residency(), Residency::WriteInFlight);
3632        let mut out = Vec::new();
3633        h.take(&mut out);
3634        assert_eq!(out, orig);
3635        assert!(pool.spill_step());
3636        let stats = pool.stats();
3637        assert_eq!(stats.frees, 1);
3638        assert_eq!(stats.spill_cancelled, 1);
3639        assert_eq!(stats.writes_elided, 1, "taken before compression: elided");
3640        assert_eq!(stats.extent_bytes_written, 0, "no extent was written");
3641        assert_eq!(stats.resident_bytes, 0, "slot accounting settled");
3642        assert_eq!(stats.live_chunks, 0);
3643    }
3644
3645    #[mz_ore::test]
3646    #[cfg_attr(miri, ignore)] // too slow
3647    fn spill_threads_end_to_end() {
3648        let pool = test_pool(128 << 10);
3649        pool.set_spill_threads(2);
3650        let mut handles = Vec::new();
3651        for seed in 0..rounds(16, 6) {
3652            handles.push(insert(&pool, &mut payload(SMALL, 500 + seed)));
3653        }
3654        pool.quiesce_spill();
3655        let stats = pool.stats();
3656        assert!(
3657            stats.spill_scheduled > 0,
3658            "budget pressure should have scheduled spills",
3659        );
3660        for (i, h) in handles.iter().enumerate() {
3661            assert_eq!(read(h), payload(SMALL, 500 + u64::cast_from(i)));
3662        }
3663        pool.join_spill_threads();
3664    }
3665
3666    /// Races the `WriteInFlight` protocol in its true concurrent form:
3667    /// spill threads compress slots without the state lock while owner
3668    /// threads copy the same chunks out under it and drop chunks mid-flight
3669    /// (both cancellation windows). Contents are asserted on every read, so
3670    /// a compression or slot release racing a copy-out shows up as
3671    /// corruption; under Miri the aliasing itself is checked.
3672    #[mz_ore::test]
3673    fn spill_threads_race_reads_and_drops() {
3674        let pool = test_pool(usize::MAX);
3675        pool.set_spill_threads(2);
3676        let iters = rounds(50, 6);
3677        let mut threads = Vec::new();
3678        for t in 0..2u64 {
3679            let pool = pool.clone();
3680            threads.push(std::thread::spawn(move || {
3681                for round in 0..iters {
3682                    let orig = payload(SMALL, t * 10_000 + round);
3683                    let handle = insert(&pool, &mut orig.clone());
3684                    // Hands the chunk to the spill threads (`WriteInFlight`).
3685                    pool.evict(&handle);
3686                    // Copy-out read racing the unlocked compression read.
3687                    assert_eq!(read(&handle), orig);
3688                    if round % 2 == 0 {
3689                        // Free while queued or mid-compression: the
3690                        // cancellation windows own the deferred cleanup.
3691                        drop(handle);
3692                    } else {
3693                        assert_eq!(read(&handle), orig);
3694                    }
3695                }
3696            }));
3697        }
3698        for thread in threads {
3699            thread.join().expect("worker thread panicked");
3700        }
3701        pool.quiesce_spill();
3702        pool.join_spill_threads();
3703        assert_eq!(pool.stats().resident_bytes, 0);
3704    }
3705
3706    #[mz_ore::test]
3707    fn insert_with_fills_in_place() {
3708        let pool = test_pool(usize::MAX);
3709        let want = payload(SMALL, 600);
3710        let h = pool.insert_with(SMALL, ChunkHints::default(), &TEST_CODEC, |dst| {
3711            assert_eq!(dst.len(), SMALL, "fill sees exactly the chunk length");
3712            dst.copy_from_slice(&want);
3713        });
3714        assert_eq!(h.residency(), Residency::UnbackedResident);
3715        assert_eq!(read(&h), want);
3716        pool.evict(&h);
3717        assert_eq!(read(&h), want, "round-trips through the extent");
3718
3719        // Empty and oversize take their fallback paths.
3720        let empty = pool.insert_with(0, ChunkHints::default(), &TEST_CODEC, |dst| {
3721            assert!(dst.is_empty())
3722        });
3723        assert!(read(&empty).is_empty());
3724        let big_len = (SIZE_CLASSES[SIZE_CLASSES.len() - 1] / 8) + 1;
3725        let big = pool.insert_with(big_len, ChunkHints::default(), &TEST_CODEC, |dst| {
3726            dst.fill(7)
3727        });
3728        assert_eq!(big.residency(), Residency::Oversize);
3729        assert_eq!(read(&big).len(), big_len);
3730    }
3731
3732    #[mz_ore::test]
3733    fn slot_exhaustion_degrades_to_heap() {
3734        // Two 64 KiB slots per class at this capacity; the third insert finds
3735        // no slot and must fall back to the heap rather than panic.
3736        let pool = Pool::with_class_capacity(128 << 10).expect("pool creation");
3737        let a = insert(&pool, &mut payload(SMALL, 700));
3738        let b = insert(&pool, &mut payload(SMALL, 701));
3739        let c = insert(&pool, &mut payload(SMALL, 702));
3740        assert_eq!(a.residency(), Residency::UnbackedResident);
3741        assert_eq!(b.residency(), Residency::UnbackedResident);
3742        assert_eq!(
3743            c.residency(),
3744            Residency::Oversize,
3745            "fallback is heap-backed"
3746        );
3747        assert_eq!(pool.stats().slot_exhausted_fallbacks, 1);
3748        assert_eq!(read(&c), payload(SMALL, 702));
3749        // Freeing a slotted chunk lets the next insert use the region again.
3750        drop(a);
3751        let d = insert(&pool, &mut payload(SMALL, 703));
3752        assert_eq!(d.residency(), Residency::UnbackedResident);
3753        assert_eq!(read(&d), payload(SMALL, 703));
3754    }
3755}